Skip to content

GitLab

Commit f9a3a2e6 authored by Antoine Kaufmann's avatar Antoine Kaufmann Committed by Antoine Kaufmann
Browse files

reformat with clang-format

parent e51835d1
Hide whitespace changes
Inline Side-by-side
Showing with 383 additions and 419 deletions
dist/common/base.c
View file @ f9a3a2e6
......@@ -101,7 +101,6 @@ bool BasePeerAdd(const char *path, bool listener) {
return true;
}
int BaseListen() {
#ifdef DEBUG
fprintf(stderr, "Creating listening sockets\n");
......@@ -170,12 +169,12 @@ int BasePeerSetupQueues(struct Peer *peer) {
}
struct SimbricksProtoListenerIntro *li =
(struct SimbricksProtoListenerIntro *) peer->intro_remote;
(struct SimbricksProtoListenerIntro *)peer->intro_remote;
#ifdef DEBUG
fprintf(stderr, "PeerNetSetupQueues(%s)\n", peer->sock_path);
fprintf(stderr, " l2c_el=%lu l2c_n=%lu c2l_el=%lu c2l_n=%lu\n", li->l2c_elen,
li->l2c_nentries, li->c2l_elen, li->c2l_nentries);
li->l2c_nentries, li->c2l_elen, li->c2l_nentries);
#endif
if (ShmAlloc(li->l2c_elen * li->l2c_nentries, &li->l2c_offset)) {
......@@ -189,12 +188,12 @@ int BasePeerSetupQueues(struct Peer *peer) {
peer->shm_fd = shm_fd;
peer->shm_base = shm_base;
peer->local_base = (void *) ((uintptr_t) shm_base + li->c2l_offset);
peer->local_base = (void *)((uintptr_t)shm_base + li->c2l_offset);
peer->local_offset = li->c2l_offset;
peer->local_elen = li->c2l_elen;
peer->local_enum = li->c2l_nentries;
peer->cleanup_base = (void *) ((uintptr_t) shm_base + li->l2c_offset);
peer->cleanup_base = (void *)((uintptr_t)shm_base + li->l2c_offset);
peer->cleanup_offset = li->l2c_offset;
peer->cleanup_elen = li->l2c_elen;
peer->cleanup_enum = li->l2c_nentries;
......@@ -207,30 +206,31 @@ int BasePeerSendIntro(struct Peer *peer) {
fprintf(stderr, "PeerDevSendIntro(%s)\n", peer->sock_path);
#endif
if (peer->sock_fd == -1) {
if (peer->sock_fd == -1) {
/* We can receive the welcome message from our peer before our local
connection to the simulator is established. In this case we hold the
message till the connection is established and send it then. */
#ifdef DEBUG
fprintf(stderr, "PeerNetSetupQueues: socket not ready yet, delaying "
"send\n");
fprintf(stderr,
"PeerNetSetupQueues: socket not ready yet, delaying "
"send\n");
#endif
return 0;
}
int shm_fd = (peer->is_listener ? peer->shm_fd : -1);
if (UxsocketSendFd(peer->sock_fd, peer->intro_remote, peer->intro_remote_len,
shm_fd)) {
shm_fd)) {
perror("BasePeerSendIntro: send failed");
return 1;
}
return 0;
}
int BasePeerReport(struct Peer *peer, uint32_t written_pos, uint32_t clean_pos) {
int BasePeerReport(struct Peer *peer, uint32_t written_pos,
uint32_t clean_pos) {
uint32_t pos = peer->local_pos_cleaned;
if (written_pos == peer->cleanup_pos_last &&
clean_pos == pos)
if (written_pos == peer->cleanup_pos_last && clean_pos == pos)
return 0;
#ifdef DEBUG
......@@ -243,9 +243,10 @@ int BasePeerReport(struct Peer *peer, uint32_t written_pos, uint32_t clean_pos)
while (pos != clean_pos) {
void *entry = (peer->local_base + pos * peer->local_elen);
volatile union SimbricksProtoBaseMsg *msg =
(volatile union SimbricksProtoBaseMsg *) entry;
msg->header.own_type = (msg->header.own_type &
(~SIMBRICKS_PROTO_MSG_OWN_MASK)) | SIMBRICKS_PROTO_MSG_OWN_PRO;
(volatile union SimbricksProtoBaseMsg *)entry;
msg->header.own_type =
(msg->header.own_type & (~SIMBRICKS_PROTO_MSG_OWN_MASK)) |
SIMBRICKS_PROTO_MSG_OWN_PRO;
pos += 1;
if (pos >= peer->local_enum)
......@@ -286,11 +287,11 @@ static int PeerAcceptEvent(struct Peer *peer) {
if (peer->intro_valid_remote) {
#ifdef DEBUG
fprintf(stderr, "PeerAcceptEvent(%s): sending welcome message\n",
peer->sock_path);
peer->sock_path);
#endif
if (BasePeerSendIntro(peer)) {
fprintf(stderr, "PeerAcceptEvent(%s): sending intro failed\n",
peer->sock_path);
peer->sock_path);
return 1;
}
}
......@@ -327,7 +328,7 @@ int BasePeerEvent(struct Peer *peer, uint32_t events) {
if (!peer->is_listener) {
/* not a listener, so we're expecting an fd for the shm region */
ret = UxsocketRecvFd(peer->sock_fd, peer->intro_local,
sizeof(peer->intro_local), &peer->shm_fd);
sizeof(peer->intro_local), &peer->shm_fd);
if (ret <= 0)
return 1;
......@@ -335,20 +336,19 @@ int BasePeerEvent(struct Peer *peer, uint32_t events) {
return 1;
struct SimbricksProtoListenerIntro *li =
(struct SimbricksProtoListenerIntro *) peer->intro_local;
peer->local_base = (void *) ((uintptr_t) peer->shm_base + li->l2c_offset);
(struct SimbricksProtoListenerIntro *)peer->intro_local;
peer->local_base = (void *)((uintptr_t)peer->shm_base + li->l2c_offset);
peer->local_offset = li->l2c_offset;
peer->local_elen = li->l2c_elen;
peer->local_enum = li->l2c_nentries;
peer->cleanup_base = (void *) ((uintptr_t) peer->shm_base + li->c2l_offset);
peer->cleanup_base = (void *)((uintptr_t)peer->shm_base + li->c2l_offset);
peer->cleanup_offset = li->c2l_offset;
peer->cleanup_elen = li->c2l_elen;
peer->cleanup_enum = li->c2l_nentries;
} else {
/* as a listener, we use our local shm region, so no fd is sent to us */
ret = recv(peer->sock_fd, peer->intro_local,
sizeof(peer->intro_local), 0);
ret = recv(peer->sock_fd, peer->intro_local, sizeof(peer->intro_local), 0);
if (ret <= 0) {
perror("PeerEvent: recv failed");
return 1;
......@@ -378,15 +378,15 @@ static inline void PollPeerTransfer(struct Peer *peer, bool *report) {
if ((peer->local_pos + n + 1) % peer->local_enum ==
peer->local_pos_cleaned) {
#ifdef DEBUG
fprintf(stderr, "PollPeerTransfer: waiting for cleanup (%u %u)\n",
n, peer->local_pos_cleaned);
fprintf(stderr, "PollPeerTransfer: waiting for cleanup (%u %u)\n", n,
peer->local_pos_cleaned);
#endif
break;
}
void *entry = (peer->local_base + (peer->local_pos + n) * peer->local_elen);
volatile union SimbricksProtoBaseMsg *msg =
(volatile union SimbricksProtoBaseMsg *) entry;
(volatile union SimbricksProtoBaseMsg *)entry;
if ((msg->header.own_type & SIMBRICKS_PROTO_MSG_OWN_MASK) !=
SIMBRICKS_PROTO_MSG_OWN_CON)
break;
......@@ -394,18 +394,17 @@ static inline void PollPeerTransfer(struct Peer *peer, bool *report) {
if (n > 0) {
#ifdef DEBUG
fprintf(stderr, "PollPeerTransfer: transferring [%u,%u] (lpc=%u lpr=%u)\n",
peer->local_pos, peer->local_pos + n, peer->local_pos_cleaned,
peer->local_pos_reported);
fprintf(stderr, "PollPeerTransfer: transferring [%u,%u] (lpc=%u lpr=%u)\n",
peer->local_pos, peer->local_pos + n, peer->local_pos_cleaned,
peer->local_pos_reported);
#endif
BaseOpPassEntries(peer, peer->local_pos, n);
uint32_t newpos = peer->local_pos + n;
peer->local_pos = (newpos < peer->local_enum ?
newpos :
newpos - peer->local_enum);
peer->local_pos =
(newpos < peer->local_enum ? newpos : newpos - peer->local_enum);
uint64_t unreported = (peer->local_pos - peer->local_pos_reported) %
peer->local_enum;
uint64_t unreported =
(peer->local_pos - peer->local_pos_reported) % peer->local_enum;
if (unreported >= kPollReportThreshold)
*report = true;
}
......@@ -420,13 +419,13 @@ static inline void PollPeerCleanup(struct Peer *peer, bool *report) {
void *entry =
(peer->cleanup_base + peer->cleanup_pos_next * peer->cleanup_elen);
volatile union SimbricksProtoBaseMsg *msg =
(volatile union SimbricksProtoBaseMsg *) entry;
(volatile union SimbricksProtoBaseMsg *)entry;
if ((msg->header.own_type & SIMBRICKS_PROTO_MSG_OWN_MASK) !=
SIMBRICKS_PROTO_MSG_OWN_PRO)
break;
#ifdef DEBUG
#ifdef DEBUG
fprintf(stderr, "PollPeerCleanup: peer %s has clean entry at %u\n",
peer->sock_path, peer->cleanup_pos_next);
#endif
......@@ -437,8 +436,9 @@ static inline void PollPeerCleanup(struct Peer *peer, bool *report) {
peer->cleanup_pos_next != peer->cleanup_pos_last);
if (cnt > 0) {
uint64_t unreported = (peer->cleanup_pos_next - peer->cleanup_pos_reported)
% peer->cleanup_enum;
uint64_t unreported =
(peer->cleanup_pos_next - peer->cleanup_pos_reported) %
peer->cleanup_enum;
if (unreported >= kCleanReportThreshold)
*report = true;
}
......@@ -459,27 +459,26 @@ void BasePoll() {
BaseOpPassReport();
}
void BaseEntryReceived(struct Peer *peer, uint32_t pos, void *data)
{
void BaseEntryReceived(struct Peer *peer, uint32_t pos, void *data) {
#ifdef DEBUG
fprintf(stderr, "BaseEntryReceived: pos=%u (cpr=%u cpl=%u)\n",
pos, peer->cleanup_pos_reported, peer->cleanup_pos_last);
fprintf(stderr, "BaseEntryReceived: pos=%u (cpr=%u cpl=%u)\n", pos,
peer->cleanup_pos_reported, peer->cleanup_pos_last);
#endif
uint64_t off = (uint64_t) pos * peer->cleanup_elen;
uint64_t off = (uint64_t)pos * peer->cleanup_elen;
void *entry = peer->cleanup_base + off;
volatile union SimbricksProtoBaseMsg *msg =
(volatile union SimbricksProtoBaseMsg *) entry;
(volatile union SimbricksProtoBaseMsg *)entry;
// first copy data after header
memcpy((void *) (msg + 1), (uint8_t *) data + sizeof(*msg),
peer->cleanup_elen - sizeof(*msg));
memcpy((void *)(msg + 1), (uint8_t *)data + sizeof(*msg),
peer->cleanup_elen - sizeof(*msg));
// then copy header except for last byte
memcpy((void *) msg, data, sizeof(*msg) - 1);
memcpy((void *)msg, data, sizeof(*msg) - 1);
// WMB()
// now copy last byte
volatile union SimbricksProtoBaseMsg *src_msg =
(volatile union SimbricksProtoBaseMsg *) data;
(volatile union SimbricksProtoBaseMsg *)data;
asm volatile("sfence" ::: "memory");
msg->header.own_type = src_msg->header.own_type;
}
\ No newline at end of file
dist/common/base.h
View file @ f9a3a2e6
......@@ -29,7 +29,6 @@
#include <stdbool.h>
#include <stddef.h>
struct Peer {
/* base address of the local queue we're polling. */
uint8_t *local_base;
......@@ -80,7 +79,6 @@ struct Peer {
// is our local peer a listener?
bool is_listener;
// set true when the queue is ready for polling
volatile bool ready;
......
dist/common/utils.c
View file @ f9a3a2e6
......@@ -215,4 +215,3 @@ void *ShmMap(int shm_fd, size_t *psize) {
*psize = statbuf.st_size;
return p;
}
dist/rdma/net_rdma.c
View file @ f9a3a2e6
......@@ -53,7 +53,6 @@ bool ib_connect = false;
uint8_t ib_port = 1;
int ib_sgid_idx = -1;
static void PrintUsage() {
fprintf(stderr,
"Usage: net_rdma [OPTIONS] IP PORT\n"
......@@ -117,7 +116,7 @@ static int ParseArgs(int argc, char *argv[]) {
}
}
if (optind + 2 != argc) {
if (optind + 2 != argc) {
PrintUsage();
return 1;
}
......
dist/rdma/net_rdma.h
View file @ f9a3a2e6
......@@ -25,12 +25,11 @@
#ifndef DIST_NET_RDMA_H_
#define DIST_NET_RDMA_H_
#include "dist/common/base.h"
#include <arpa/inet.h>
#include <stdbool.h>
#include <stddef.h>
#include "dist/common/base.h"
// configuration variables
extern size_t shm_size;
......
dist/rdma/rdma.c
View file @ f9a3a2e6
......@@ -23,7 +23,6 @@
*/
#include "dist/rdma/rdma.h"
#include "dist/rdma/net_rdma.h"
#include <fcntl.h>
#include <infiniband/verbs.h>
......@@ -33,6 +32,8 @@
#include <sys/epoll.h>
#include <unistd.h>
#include "dist/rdma/net_rdma.h"
#define SENDQ_LEN (8 * 1024)
#define MSG_RXBUFS 512
#define MSG_TXBUFS 512
......@@ -73,7 +74,7 @@ static struct ibv_cq *cq;
static struct ibv_comp_channel *comp_chan;
static struct ibv_mr *mr_shm;
static struct ibv_mr *mr_msgs;
static struct ibv_qp_init_attr qp_attr = { };
static struct ibv_qp_init_attr qp_attr = {};
static struct NetRdmaMsg msgs[MSG_RXBUFS + MSG_TXBUFS];
pthread_spinlock_t freelist_spin;
......@@ -98,12 +99,12 @@ static void RdmaMsgFree(struct NetRdmaMsg *msg) {
}
static int RdmMsgRxEnqueue(struct NetRdmaMsg *msg) {
struct ibv_sge sge = { };
sge.addr = (uintptr_t) msg;
struct ibv_sge sge = {};
sge.addr = (uintptr_t)msg;
sge.length = sizeof(*msg);
sge.lkey = mr_msgs->lkey;
struct ibv_recv_wr recv_wr = { };
struct ibv_recv_wr recv_wr = {};
recv_wr.wr_id = msg - msgs;
recv_wr.sg_list = &sge;
recv_wr.num_sge = 1;
......@@ -119,7 +120,7 @@ static int RdmMsgRxEnqueue(struct NetRdmaMsg *msg) {
static int RdmaMsgRxIntro(struct NetRdmaMsg *msg) {
if (msg->id >= peer_num) {
fprintf(stderr, "RdmaMsgRxIntro: invalid peer id in message (%lu)\n",
msg->id);
msg->id);
abort();
}
......@@ -128,7 +129,7 @@ static int RdmaMsgRxIntro(struct NetRdmaMsg *msg) {
if (peer->intro_valid_remote) {
fprintf(stderr, "RdmaMsgRxIntro: received multiple messages (%lu)\n",
msg->id);
msg->id);
abort();
}
......@@ -141,7 +142,7 @@ static int RdmaMsgRxIntro(struct NetRdmaMsg *msg) {
if (BasePeerSetupQueues(peer)) {
fprintf(stderr, "RdmaMsgRxIntro(%s): queue setup failed\n",
peer->sock_path);
peer->sock_path);
abort();
}
if (BasePeerSendIntro(peer))
......@@ -151,11 +152,11 @@ static int RdmaMsgRxIntro(struct NetRdmaMsg *msg) {
// now we can send our intro for a listener
if (peer->is_listener && BaseOpPassIntro(peer)) {
fprintf(stderr, "RdmaMsgRxIntro(%s): sending l intro failed\n",
peer->sock_path);
peer->sock_path);
return 1;
}
fprintf(stderr, "RdmaMsgRxIntro(%s): marking peer as ready\n",
peer->sock_path);
peer->sock_path);
peer->ready = true;
}
return 0;
......@@ -209,14 +210,13 @@ int RdmaCommonInit(struct ibv_context *ctx) {
return 1;
}
if (!(mr_shm = ibv_reg_mr(pd, shm_base, shm_size,
IBV_ACCESS_LOCAL_WRITE |
IBV_ACCESS_REMOTE_WRITE))) {
if (!(mr_shm =
ibv_reg_mr(pd, shm_base, shm_size,
IBV_ACCESS_LOCAL_WRITE | IBV_ACCESS_REMOTE_WRITE))) {
perror("RdmaCommonInit: ibv_reg_mr shm failed");
return 1;
}
if (!(mr_msgs = ibv_reg_mr(pd, msgs, sizeof(msgs),
IBV_ACCESS_LOCAL_WRITE))) {
if (!(mr_msgs = ibv_reg_mr(pd, msgs, sizeof(msgs), IBV_ACCESS_LOCAL_WRITE))) {
perror("RdmaCommonInit: ibv_reg_mr msgs failed");
return 1;
}
......@@ -384,9 +384,9 @@ int BaseOpPassIntro(struct Peer *peer) {
// connecting peers have sent us an SHM region, need to register this an as MR
if (!peer->is_listener) {
if (!(peer->shm_opaque = ibv_reg_mr(pd, peer->shm_base, peer->shm_size,
IBV_ACCESS_LOCAL_WRITE |
IBV_ACCESS_REMOTE_WRITE))) {
if (!(peer->shm_opaque =
ibv_reg_mr(pd, peer->shm_base, peer->shm_size,
IBV_ACCESS_LOCAL_WRITE | IBV_ACCESS_REMOTE_WRITE))) {
perror("BaseOpPassIntro: ibv_reg_mr shm failed");
return 1;
}
......@@ -407,7 +407,7 @@ int BaseOpPassIntro(struct Peer *peer) {
return 1;
msg->id = peer - peers;
msg->base_addr = (uintptr_t) peer->shm_base;
msg->base_addr = (uintptr_t)peer->shm_base;
struct ibv_mr *mr = peer->shm_opaque;
msg->rkey = mr->rkey;
msg->msg_type = kMsgIntro;
......@@ -420,11 +420,11 @@ int BaseOpPassIntro(struct Peer *peer) {
memcpy(msg->intro.data, peer->intro_local, peer->intro_local_len);
struct ibv_sge sge;
sge.addr = (uintptr_t) msg;
sge.addr = (uintptr_t)msg;
sge.length = sizeof(*msg);
sge.lkey = mr_msgs->lkey;
struct ibv_send_wr send_wr = { };
struct ibv_send_wr send_wr = {};
send_wr.wr_id = msg - msgs;
send_wr.opcode = IBV_WR_SEND;
send_wr.send_flags = IBV_SEND_SIGNALED;
......@@ -445,8 +445,7 @@ int BaseOpPassIntro(struct Peer *peer) {
int BaseOpPassEntries(struct Peer *peer, uint32_t pos, uint32_t n) {
#ifdef RDMA_DEBUG
fprintf(stderr, "BaseOpPassEntries(%s,%u)\n", peer->sock_path,
pos);
fprintf(stderr, "BaseOpPassEntries(%s,%u)\n", peer->sock_path, pos);
fprintf(stderr, " remote_base=%lx local_base=%p\n", peer->remote_base,
peer->local_base);
#endif
......@@ -458,12 +457,12 @@ int BaseOpPassEntries(struct Peer *peer, uint32_t pos, uint32_t n) {
while (1) {
uint64_t abs_pos = pos * peer->local_elen;
struct ibv_sge sge;
sge.addr = (uintptr_t) (peer->local_base + abs_pos);
sge.addr = (uintptr_t)(peer->local_base + abs_pos);
sge.length = peer->local_elen * n;
struct ibv_mr *mr = peer->shm_opaque;
sge.lkey = mr->lkey;
struct ibv_send_wr send_wr = { };
struct ibv_send_wr send_wr = {};
send_wr.wr_id = -1ULL;
send_wr.opcode = IBV_WR_RDMA_WRITE;
if (triggerSig)
......@@ -519,11 +518,11 @@ int BaseOpPassReport() {
while (1) {
struct ibv_sge sge;
sge.addr = (uintptr_t) msg;
sge.addr = (uintptr_t)msg;
sge.length = sizeof(*msg);
sge.lkey = mr_msgs->lkey;
struct ibv_send_wr send_wr = { };
struct ibv_send_wr send_wr = {};
send_wr.wr_id = msg - msgs;
send_wr.opcode = IBV_WR_SEND;
send_wr.send_flags = IBV_SEND_SIGNALED;
......
dist/rdma/rdma.h
View file @ f9a3a2e6
......@@ -25,21 +25,18 @@
#ifndef DIST_RDMA_H_
#define DIST_RDMA_H_
#include "dist/rdma/net_rdma.h"
#include <infiniband/verbs.h>
#include "dist/rdma/net_rdma.h"
int RdmaCommonInit(struct ibv_context *ctx);
int RdmaCMListen(struct sockaddr_in *addr);
int RdmaCMConnect(struct sockaddr_in *addr);
struct ibv_qp *RdmaCMCreateQP(struct ibv_pd *pd,
struct ibv_qp_init_attr *attr);
struct ibv_qp *RdmaCMCreateQP(struct ibv_pd *pd, struct ibv_qp_init_attr *attr);
int RdmaIBListen(struct sockaddr_in *addr);
int RdmaIBConnect(struct sockaddr_in *addr);
struct ibv_qp *RdmaIBCreateQP(struct ibv_pd *pd,
struct ibv_qp_init_attr *attr);
struct ibv_qp *RdmaIBCreateQP(struct ibv_pd *pd, struct ibv_qp_init_attr *attr);
#endif // DIST_RDMA_H_
dist/rdma/rdma_cm.c
View file @ f9a3a2e6
......@@ -22,15 +22,15 @@
* SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
*/
#include "dist/rdma/rdma.h"
#include "dist/rdma/net_rdma.h"
#include <rdma/rdma_cma.h>
#include <stdio.h>
#include <stdlib.h>
#include "dist/rdma/net_rdma.h"
#include "dist/rdma/rdma.h"
static struct rdma_event_channel *cm_channel;
static struct rdma_conn_param conn_param = { };
static struct rdma_conn_param conn_param = {};
static struct rdma_cm_id *cm_id;
int RdmaCMListen(struct sockaddr_in *addr) {
......@@ -45,7 +45,7 @@ int RdmaCMListen(struct sockaddr_in *addr) {
return 1;
}
if (rdma_bind_addr(listen_id, (struct sockaddr *) addr)) {
if (rdma_bind_addr(listen_id, (struct sockaddr *)addr)) {
perror("RdmaListen: rdma_bind_addr failed");
return 1;
}
......@@ -115,7 +115,7 @@ int RdmaCMConnect(struct sockaddr_in *addr) {
return 1;
}
if (rdma_resolve_addr(cm_id, NULL, (struct sockaddr *) addr, 5000)) {
if (rdma_resolve_addr(cm_id, NULL, (struct sockaddr *)addr, 5000)) {
perror("RdmaConnect: rdma_resolve_addr failed");
return 1;
}
......@@ -158,7 +158,7 @@ int RdmaCMConnect(struct sockaddr_in *addr) {
return 1;
conn_param.initiator_depth = 1;
conn_param.retry_count = 7;
conn_param.retry_count = 7;
if (rdma_connect(cm_id, &conn_param)) {
perror("RdmaConnect: rdma_connect failed");
return 1;
......
dist/rdma/rdma_ib.c
View file @ f9a3a2e6
......@@ -22,14 +22,13 @@
* SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
*/
#include "dist/rdma/rdma.h"
#include "dist/rdma/net_rdma.h"
#include <rdma/rdma_cma.h>
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include "dist/rdma/net_rdma.h"
#include "dist/rdma/rdma.h"
struct RdmaIBInitMsg {
union ibv_gid gid;
......@@ -57,7 +56,7 @@ static int SockListen(struct sockaddr_in *addr) {
return 1;
}
if (bind(lfd, (struct sockaddr *) addr, sizeof(*addr))) {
if (bind(lfd, (struct sockaddr *)addr, sizeof(*addr))) {
perror("RdmaIBListen: bind failed");
return 1;
}
......@@ -81,7 +80,7 @@ static int SockConnect(struct sockaddr_in *addr) {
return 1;
}
if (connect(sock_fd, (struct sockaddr *) addr, sizeof(*addr))) {
if (connect(sock_fd, (struct sockaddr *)addr, sizeof(*addr))) {
perror("RdmaIBConnect: connect failed");
}
return 0;
......@@ -152,7 +151,6 @@ static int CommonInit() {
out_msg.qpn = ib_qp->qp_num;
out_msg.psn = psn_local;
if (write(sock_fd, &out_msg, sizeof(out_msg)) != sizeof(out_msg)) {
perror("CommonInit: write failed");
}
......@@ -165,8 +163,8 @@ static int CommonInit() {
#ifdef RDMA_DEBUG
fprintf(stderr, "out: lid=%x qpn=%x psn=%x iid=%lx\n", out_msg.lid,
out_msg.qpn, out_msg.psn, out_msg.gid.global.interface_id);
fprintf(stderr, "in: lid=%x qpn=%x psn=%x iid=%lx\n", in_msg.lid,
in_msg.qpn, in_msg.psn, in_msg.gid.global.interface_id);
fprintf(stderr, "in: lid=%x qpn=%x psn=%x iid=%lx\n", in_msg.lid, in_msg.qpn,
in_msg.psn, in_msg.gid.global.interface_id);
#endif
// change queue pair to "ready to receive"
......@@ -191,13 +189,9 @@ static int CommonInit() {
attr.ah_attr.grh.sgid_index = ib_sgid_idx;
}
if (ibv_modify_qp(ib_qp, &attr,
IBV_QP_STATE |
IBV_QP_AV |
IBV_QP_PATH_MTU |
IBV_QP_DEST_QPN |
IBV_QP_RQ_PSN |
IBV_QP_MAX_DEST_RD_ATOMIC |
IBV_QP_MIN_RNR_TIMER)) {
IBV_QP_STATE | IBV_QP_AV | IBV_QP_PATH_MTU |
IBV_QP_DEST_QPN | IBV_QP_RQ_PSN |
IBV_QP_MAX_DEST_RD_ATOMIC | IBV_QP_MIN_RNR_TIMER)) {
perror("CommonInit: Failed to modify QP to RTR");
return 1;
}
......@@ -210,12 +204,9 @@ static int CommonInit() {
attr.sq_psn = psn_local;
attr.max_rd_atomic = 1;
if (ibv_modify_qp(ib_qp, &attr,
IBV_QP_STATE |
IBV_QP_TIMEOUT |
IBV_QP_RETRY_CNT |
IBV_QP_RNR_RETRY |
IBV_QP_SQ_PSN |
IBV_QP_MAX_QP_RD_ATOMIC)) {
IBV_QP_STATE | IBV_QP_TIMEOUT | IBV_QP_RETRY_CNT |
IBV_QP_RNR_RETRY | IBV_QP_SQ_PSN |
IBV_QP_MAX_QP_RD_ATOMIC)) {
perror("CommonInit: Failed to modify QP to RTS");
return 1;
}
......@@ -246,17 +237,13 @@ struct ibv_qp *RdmaIBCreateQP(struct ibv_pd *pd,
}
// transition queue pair from reset to init state
struct ibv_qp_attr attr_init = {
.qp_state = IBV_QPS_INIT,
.pkey_index = 0,
.port_num = ib_port,
.qp_access_flags = 0
};
struct ibv_qp_attr attr_init = {.qp_state = IBV_QPS_INIT,
.pkey_index = 0,
.port_num = ib_port,
.qp_access_flags = 0};
if (ibv_modify_qp(ib_qp, &attr_init,
IBV_QP_STATE |
IBV_QP_PKEY_INDEX |
IBV_QP_PORT |
IBV_QP_ACCESS_FLAGS)) {
IBV_QP_STATE | IBV_QP_PKEY_INDEX | IBV_QP_PORT |
IBV_QP_ACCESS_FLAGS)) {
perror("RdmaIBCreateQP: ibv_modify_qp failed (reset -> init)");
ibv_destroy_qp(ib_qp);
return NULL;
......
dist/sockets/net_sockets.c
View file @ f9a3a2e6
......@@ -84,7 +84,6 @@ struct SockMsg {
};
} __attribute__((packed));
const char *shm_path = NULL;
size_t shm_size = 256 * 1024 * 1024ULL; // 256MB
......@@ -148,7 +147,7 @@ static int ParseArgs(int argc, char *argv[]) {
}
}
if (optind + 2 != argc) {
if (optind + 2 != argc) {
PrintUsage();
return 1;
}
......@@ -258,7 +257,7 @@ static int SockListen(struct sockaddr_in *addr) {
return 1;
}
if (bind(lfd, (struct sockaddr *) addr, sizeof(*addr))) {
if (bind(lfd, (struct sockaddr *)addr, sizeof(*addr))) {
perror("RdmaIBListen: bind failed");
return 1;
}
......@@ -283,7 +282,7 @@ static int SockConnect(struct sockaddr_in *addr) {
return 1;
}
if (connect(sockfd, (struct sockaddr *) addr, sizeof(*addr))) {
if (connect(sockfd, (struct sockaddr *)addr, sizeof(*addr))) {
perror("RdmaIBConnect: connect failed");
}
......@@ -298,7 +297,7 @@ static int SockMsgRxIntro(struct SockMsg *msg) {
abort();
}
if (msg->msg_len <
offsetof(struct SockMsg, intro.data) + intro_msg->payload_len) {
offsetof(struct SockMsg, intro.data) + intro_msg->payload_len) {
fprintf(stderr, "SockMsgRxIntro: message too short for payload len\n");
abort();
}
......@@ -312,7 +311,7 @@ static int SockMsgRxIntro(struct SockMsg *msg) {
msg->id);
abort();
}
if (intro_msg->payload_len > (uint32_t) sizeof(peer->intro_remote)) {
if (intro_msg->payload_len > (uint32_t)sizeof(peer->intro_remote)) {
fprintf(stderr, "SockMsgRxIntro: Intro longer than buffer\n");
abort();
}
......@@ -323,7 +322,7 @@ static int SockMsgRxIntro(struct SockMsg *msg) {
if (BasePeerSetupQueues(peer)) {
fprintf(stderr, "SockMsgRxIntro(%s): queue setup failed\n",
peer->sock_path);
peer->sock_path);
abort();
}
if (BasePeerSendIntro(peer))
......@@ -423,7 +422,7 @@ static int SockEvent(uint32_t events) {
rx_buf_pos += ret;
struct SockMsg *msg = (struct SockMsg *) rx_buffer;
struct SockMsg *msg = (struct SockMsg *)rx_buffer;
while (rx_buf_pos >= sizeof(*msg) && rx_buf_pos >= msg->msg_len) {
if (SockMsgRx(msg))
return 1;
......@@ -451,7 +450,7 @@ static int SockSend(struct SockMsg *msg) {
msg->msg_id = __sync_fetch_and_add(&msg_id, 1);
size_t len = msg->msg_len;
size_t pos = 0;
uint8_t *buf = (uint8_t *) msg;
uint8_t *buf = (uint8_t *)msg;
do {
ssize_t ret = write(sockfd, buf + pos, len - pos);
if (ret > 0) {
......@@ -641,7 +640,7 @@ int main(int argc, char *argv[]) {
return EXIT_FAILURE;
if (BaseListen())
return EXIT_FAILURE;
return EXIT_FAILURE;
if (mode_listen) {
if (SockListen(&addr))
......
lib/simbricks/base/cxxatomicfix.h
View file @ f9a3a2e6
......@@ -32,7 +32,7 @@
*
* until then, this needs to be included before the generic header.
*/
#include <atomic>
#define _Atomic(T) std::atomic<T>
using std::atomic_load_explicit;
......
lib/simbricks/base/generic.h
View file @ f9a3a2e6
......@@ -50,58 +50,57 @@
* @param if_struct Interfacing struct, (member base must be
* `struct SimBricksBaseIf`).
*/
#define SIMBRICKS_BASEIF_GENERIC(prefix, msg_union, if_struct) \
\
static inline volatile union msg_union *prefix##InPeek( \
struct if_struct *base_if, uint64_t ts) { \
return (volatile union msg_union *) SimbricksBaseIfInPeek( \
&base_if->base, ts); \
} \
\
static inline volatile union msg_union *prefix##InPoll( \
struct if_struct *base_if, uint64_t ts) { \
return (volatile union msg_union *) SimbricksBaseIfInPoll( \
&base_if->base, ts); \
} \
\
static inline uint8_t prefix##InType( \
struct if_struct *base_if, volatile union msg_union *msg) { \
return SimbricksBaseIfInType(&base_if->base, &msg->base); \
} \
\
static inline void prefix##InDone( \
struct if_struct *base_if, volatile union msg_union *msg) { \
SimbricksBaseIfInDone(&base_if->base, &msg->base); \
} \
\
static inline uint64_t prefix##InTimestamp( struct if_struct *base_if) { \
return SimbricksBaseIfInTimestamp(&base_if->base); \
} \
\
static inline volatile union msg_union *prefix##OutAlloc( \
struct if_struct *base_if, \
uint64_t timestamp) { \
return (volatile union msg_union *) SimbricksBaseIfOutAlloc( \
&base_if->base, timestamp); \
} \
\
static inline void prefix##OutSend(struct if_struct *base_if, \
volatile union msg_union *msg, \
uint8_t msg_type) { \
SimbricksBaseIfOutSend(&base_if->base, &msg->base, msg_type); \
} \
\
static inline int prefix##OutSync(struct if_struct *base_if, \
uint64_t timestamp) { \
return SimbricksBaseIfOutSync(&base_if->base, timestamp); \
} \
\
static inline uint64_t prefix##OutNextSync(struct if_struct *base_if) { \
return SimbricksBaseIfOutNextSync(&base_if->base); \
} \
\
static inline size_t prefix##OutMsgLen(struct if_struct *base_if) { \
return SimbricksBaseIfOutMsgLen(&base_if->base); \
}
#define SIMBRICKS_BASEIF_GENERIC(prefix, msg_union, if_struct) \
\
static inline volatile union msg_union *prefix##InPeek( \
struct if_struct *base_if, uint64_t ts) { \
return (volatile union msg_union *)SimbricksBaseIfInPeek(&base_if->base, \
ts); \
} \
\
static inline volatile union msg_union *prefix##InPoll( \
struct if_struct *base_if, uint64_t ts) { \
return (volatile union msg_union *)SimbricksBaseIfInPoll(&base_if->base, \
ts); \
} \
\
static inline uint8_t prefix##InType(struct if_struct *base_if, \
volatile union msg_union *msg) { \
return SimbricksBaseIfInType(&base_if->base, &msg->base); \
} \
\
static inline void prefix##InDone(struct if_struct *base_if, \
volatile union msg_union *msg) { \
SimbricksBaseIfInDone(&base_if->base, &msg->base); \
} \
\
static inline uint64_t prefix##InTimestamp(struct if_struct *base_if) { \
return SimbricksBaseIfInTimestamp(&base_if->base); \
} \
\
static inline volatile union msg_union *prefix##OutAlloc( \
struct if_struct *base_if, uint64_t timestamp) { \
return (volatile union msg_union *)SimbricksBaseIfOutAlloc(&base_if->base, \
timestamp); \
} \
\
static inline void prefix##OutSend(struct if_struct *base_if, \
volatile union msg_union *msg, \
uint8_t msg_type) { \
SimbricksBaseIfOutSend(&base_if->base, &msg->base, msg_type); \
} \
\
static inline int prefix##OutSync(struct if_struct *base_if, \
uint64_t timestamp) { \
return SimbricksBaseIfOutSync(&base_if->base, timestamp); \
} \
\
static inline uint64_t prefix##OutNextSync(struct if_struct *base_if) { \
return SimbricksBaseIfOutNextSync(&base_if->base); \
} \
\
static inline size_t prefix##OutMsgLen(struct if_struct *base_if) { \
return SimbricksBaseIfOutMsgLen(&base_if->base); \
}
#endif // SIMBRICKS_BASEIF_BASEIF_H_
lib/simbricks/base/if.c
View file @ f9a3a2e6
......@@ -33,7 +33,6 @@
#include <stdlib.h>
#include <string.h>
#include <sys/mman.h>
#include <sys/socket.h>
#include <sys/stat.h>
#include <sys/un.h>
......@@ -51,10 +50,8 @@ enum ConnState {
kConnOpen,
};
int SimbricksBaseIfSHMPoolCreate(struct SimbricksBaseIfSHMPool *pool,
const char *path, size_t pool_size)
{
const char *path, size_t pool_size) {
pool->path = path;
pool->size = pool_size;
pool->pos = 0;
......@@ -81,8 +78,7 @@ int SimbricksBaseIfSHMPoolCreate(struct SimbricksBaseIfSHMPool *pool,
return 0;
}
int SimbricksBaseIfSHMPoolMapFd(struct SimbricksBaseIfSHMPool *pool, int fd)
{
int SimbricksBaseIfSHMPoolMapFd(struct SimbricksBaseIfSHMPool *pool, int fd) {
struct stat statbuf;
if (fstat(fd, &statbuf) != 0) {
......@@ -91,8 +87,8 @@ int SimbricksBaseIfSHMPoolMapFd(struct SimbricksBaseIfSHMPool *pool, int fd)
return -1;
}
pool->base = mmap(NULL, statbuf.st_size, PROT_READ | PROT_WRITE, MAP_SHARED,
fd, 0);
pool->base =
mmap(NULL, statbuf.st_size, PROT_READ | PROT_WRITE, MAP_SHARED, fd, 0);
if (pool->base == MAP_FAILED) {
perror("SimbricksBaseIfSHMPoolMap: mmap failed");
return -1;
......@@ -106,8 +102,7 @@ int SimbricksBaseIfSHMPoolMapFd(struct SimbricksBaseIfSHMPool *pool, int fd)
}
int SimbricksBaseIfSHMPoolMap(struct SimbricksBaseIfSHMPool *pool,
const char *path)
{
const char *path) {
int fd;
if ((fd = open(path, O_RDWR, 0666) == -1)) {
......@@ -122,8 +117,7 @@ int SimbricksBaseIfSHMPoolMap(struct SimbricksBaseIfSHMPool *pool,
return 0;
}
int SimbricksBaseIfSHMPoolUnmap(struct SimbricksBaseIfSHMPool *pool)
{
int SimbricksBaseIfSHMPoolUnmap(struct SimbricksBaseIfSHMPool *pool) {
if (munmap(pool->base, pool->size)) {
perror("SimbricksBaseIfSHMPoolUnmap: unmap failed");
return -1;
......@@ -136,13 +130,11 @@ int SimbricksBaseIfSHMPoolUnmap(struct SimbricksBaseIfSHMPool *pool)
return 0;
}
int SimbricksBaseIfSHMPoolUnlink(struct SimbricksBaseIfSHMPool *pool)
{
int SimbricksBaseIfSHMPoolUnlink(struct SimbricksBaseIfSHMPool *pool) {
return unlink(pool->path);
}
void SimbricksBaseIfDefaultParams(struct SimbricksBaseIfParams *params)
{
void SimbricksBaseIfDefaultParams(struct SimbricksBaseIfParams *params) {
params->link_latency = 500 * 1000;
params->sync_interval = params->link_latency;
params->sock_path = NULL;
......@@ -153,22 +145,19 @@ void SimbricksBaseIfDefaultParams(struct SimbricksBaseIfParams *params)
params->upper_layer_proto = SIMBRICKS_PROTO_ID_BASE;
}
size_t SimbricksBaseIfSHMSize(struct SimbricksBaseIfParams *params)
{
size_t SimbricksBaseIfSHMSize(struct SimbricksBaseIfParams *params) {
return params->in_num_entries * params->in_entries_size +
params->out_num_entries * params->out_entries_size;
params->out_num_entries * params->out_entries_size;
}
int SimbricksBaseIfInit(struct SimbricksBaseIf *base_if,
struct SimbricksBaseIfParams *params)
{
struct SimbricksBaseIfParams *params) {
memset(base_if, 0, sizeof(*base_if));
base_if->params = *params;
return 0;
}
static int AcceptOnBaseIf(struct SimbricksBaseIf *base_if)
{
static int AcceptOnBaseIf(struct SimbricksBaseIf *base_if) {
int flags = (!base_if->params.blocking_conn ? SOCK_NONBLOCK : 0);
base_if->conn_fd = accept4(base_if->listen_fd, NULL, NULL, flags);
if (base_if->conn_fd >= 0) {
......@@ -188,8 +177,7 @@ static int AcceptOnBaseIf(struct SimbricksBaseIf *base_if)
}
int SimbricksBaseIfListen(struct SimbricksBaseIf *base_if,
struct SimbricksBaseIfSHMPool *pool)
{
struct SimbricksBaseIfSHMPool *pool) {
struct sockaddr_un saun;
int flags;
struct SimbricksBaseIfParams *params = &base_if->params;
......@@ -199,8 +187,9 @@ int SimbricksBaseIfListen(struct SimbricksBaseIf *base_if,
size_t in_len = params->in_num_entries * params->in_entries_size;
size_t out_len = params->out_num_entries * params->out_entries_size;
if (pool->pos + in_len + out_len > pool->size) {
fprintf(stderr, "SimbricksBaseIfListen: not enough memory available in "
"pool");
fprintf(stderr,
"SimbricksBaseIfListen: not enough memory available in "
"pool");
return -1;
}
......@@ -256,8 +245,7 @@ out_error:
return -1;
}
int SimbricksBaseIfConnect(struct SimbricksBaseIf *base_if)
{
int SimbricksBaseIfConnect(struct SimbricksBaseIf *base_if) {
struct sockaddr_un saun;
int flags;
struct SimbricksBaseIfParams *params = &base_if->params;
......@@ -301,8 +289,7 @@ out_error:
return -1;
}
int SimbricksBaseIfConnected(struct SimbricksBaseIf *base_if)
{
int SimbricksBaseIfConnected(struct SimbricksBaseIf *base_if) {
switch (base_if->conn_state) {
case kConnClosed:
return -1;
......@@ -329,8 +316,8 @@ int SimbricksBaseIfConnected(struct SimbricksBaseIf *base_if)
int status = 0;
socklen_t slen = sizeof(status);
if (getsockopt(base_if->conn_fd, SOL_SOCKET, SO_ERROR, &status, &slen)
!= 0) {
if (getsockopt(base_if->conn_fd, SOL_SOCKET, SO_ERROR, &status, &slen) !=
0) {
perror("SimbricksBaseIfConnected: getsockopt failed");
close(base_if->conn_fd);
base_if->conn_fd = -1;
......@@ -364,8 +351,7 @@ int SimbricksBaseIfConnected(struct SimbricksBaseIf *base_if)
}
}
int SimbricksBaseIfConnFd(struct SimbricksBaseIf *base_if)
{
int SimbricksBaseIfConnFd(struct SimbricksBaseIf *base_if) {
if (base_if->conn_state == kConnListening) {
return base_if->listen_fd;
} else if (base_if->conn_state == kConnConnecting) {
......@@ -375,8 +361,7 @@ int SimbricksBaseIfConnFd(struct SimbricksBaseIf *base_if)
}
}
int SimbricksBaseIfConnsWait(struct SimbricksBaseIf **base_ifs, unsigned n)
{
int SimbricksBaseIfConnsWait(struct SimbricksBaseIf **base_ifs, unsigned n) {
unsigned i, n_wait;
struct pollfd pfds[n];
unsigned ids[n];
......@@ -447,11 +432,9 @@ int SimbricksBaseIfConnsWait(struct SimbricksBaseIf **base_ifs, unsigned n)
/** Send intro. */
int SimbricksBaseIfIntroSend(struct SimbricksBaseIf *base_if,
const void *payload, size_t payload_len)
{
const void *payload, size_t payload_len) {
if (base_if->conn_state != kConnAwaitHandshakeRxTx &&
base_if->conn_state != kConnAwaitHandshakeTx)
{
base_if->conn_state != kConnAwaitHandshakeTx) {
return -1;
}
......@@ -471,17 +454,20 @@ int SimbricksBaseIfIntroSend(struct SimbricksBaseIf *base_if,
};
/* fill in payload iov entry */
iov[1].iov_base = (void *) payload;
iov[1].iov_base = (void *)payload;
iov[1].iov_len = payload_len;
struct SimbricksProtoListenerIntro l_intro;
struct SimbricksProtoConnecterIntro c_intro;
if (base_if->listener) {
l_intro.version = SIMBRICKS_PROTO_VERSION;
l_intro.flags = (base_if->params.sync_mode == kSimbricksBaseIfSyncDisabled ?
0 : (SIMBRICKS_PROTO_FLAGS_LI_SYNC |
(base_if->params.sync_mode == kSimbricksBaseIfSyncRequired ?
SIMBRICKS_PROTO_FLAGS_LI_SYNC_FORCE : 0)));
l_intro.flags =
(base_if->params.sync_mode == kSimbricksBaseIfSyncDisabled
? 0
: (SIMBRICKS_PROTO_FLAGS_LI_SYNC |
(base_if->params.sync_mode == kSimbricksBaseIfSyncRequired
? SIMBRICKS_PROTO_FLAGS_LI_SYNC_FORCE
: 0)));
l_intro.l2c_offset = base_if->out_queue - base_if->shm->base;
l_intro.l2c_elen = base_if->out_elen;
......@@ -505,13 +491,16 @@ int SimbricksBaseIfIntroSend(struct SimbricksBaseIf *base_if,
cmsg->cmsg_level = SOL_SOCKET;
cmsg->cmsg_type = SCM_RIGHTS;
cmsg->cmsg_len = CMSG_LEN(sizeof(int));
*(int *) CMSG_DATA(cmsg) = base_if->shm->fd;
*(int *)CMSG_DATA(cmsg) = base_if->shm->fd;
} else {
c_intro.version = SIMBRICKS_PROTO_VERSION;
c_intro.flags = (base_if->params.sync_mode == kSimbricksBaseIfSyncDisabled ?
0 : (SIMBRICKS_PROTO_FLAGS_CO_SYNC |
(base_if->params.sync_mode == kSimbricksBaseIfSyncRequired ?
SIMBRICKS_PROTO_FLAGS_CO_SYNC_FORCE : 0)));
c_intro.flags =
(base_if->params.sync_mode == kSimbricksBaseIfSyncDisabled
? 0
: (SIMBRICKS_PROTO_FLAGS_CO_SYNC |
(base_if->params.sync_mode == kSimbricksBaseIfSyncRequired
? SIMBRICKS_PROTO_FLAGS_CO_SYNC_FORCE
: 0)));
c_intro.upper_layer_proto = base_if->params.upper_layer_proto;
c_intro.upper_layer_intro_off = sizeof(c_intro);
......@@ -523,10 +512,11 @@ int SimbricksBaseIfIntroSend(struct SimbricksBaseIf *base_if,
if (ret < 0) {
perror("SimbricksBaseIfIntroSend: sendmsg failed");
return -1;
} else if (ret != (ssize_t) (iov[0].iov_len + iov[1].iov_len)) {
fprintf(stderr, "SimbricksBaseIfIntroSend: sendmsg was short, "
"currently unsupported\n");
return -1;
} else if (ret != (ssize_t)(iov[0].iov_len + iov[1].iov_len)) {
fprintf(stderr,
"SimbricksBaseIfIntroSend: sendmsg was short, "
"currently unsupported\n");
return -1;
}
if (base_if->conn_state == kConnAwaitHandshakeTx) {
......@@ -534,8 +524,9 @@ int SimbricksBaseIfIntroSend(struct SimbricksBaseIf *base_if,
} else if (base_if->conn_state == kConnAwaitHandshakeRxTx) {
base_if->conn_state = kConnAwaitHandshakeRx;
} else {
fprintf(stderr, "SimbricksBaseIfIntroSend: connection in unexpected "
"state at the end.\n");
fprintf(stderr,
"SimbricksBaseIfIntroSend: connection in unexpected "
"state at the end.\n");
abort();
}
......@@ -543,12 +534,10 @@ int SimbricksBaseIfIntroSend(struct SimbricksBaseIf *base_if,
}
/** Receive intro. */
int SimbricksBaseIfIntroRecv(struct SimbricksBaseIf *base_if,
void *payload, size_t *payload_len)
{
int SimbricksBaseIfIntroRecv(struct SimbricksBaseIf *base_if, void *payload,
size_t *payload_len) {
if (base_if->conn_state != kConnAwaitHandshakeRxTx &&
base_if->conn_state != kConnAwaitHandshakeRx)
{
base_if->conn_state != kConnAwaitHandshakeRx) {
return -1;
}
......@@ -557,7 +546,7 @@ int SimbricksBaseIfIntroRecv(struct SimbricksBaseIf *base_if,
struct iovec iov;
iov.iov_base = intro_buf;
iov.iov_len = sizeof(intro_buf);
struct cmsghdr *cmsg;
union {
char buf[CMSG_SPACE(sizeof(int))];
......@@ -594,7 +583,7 @@ int SimbricksBaseIfIntroRecv(struct SimbricksBaseIf *base_if,
if (base_if->listener) {
struct SimbricksProtoConnecterIntro *c_intro =
(struct SimbricksProtoConnecterIntro *) intro_buf;
(struct SimbricksProtoConnecterIntro *)intro_buf;
sync = c_intro->flags & SIMBRICKS_PROTO_FLAGS_CO_SYNC;
sync_force = c_intro->flags & SIMBRICKS_PROTO_FLAGS_CO_SYNC_FORCE;
version = c_intro->version;
......@@ -602,7 +591,7 @@ int SimbricksBaseIfIntroRecv(struct SimbricksBaseIf *base_if,
upper_off = c_intro->upper_layer_intro_off;
} else {
struct SimbricksProtoListenerIntro *l_intro =
(struct SimbricksProtoListenerIntro *) intro_buf;
(struct SimbricksProtoListenerIntro *)intro_buf;
sync = l_intro->flags & SIMBRICKS_PROTO_FLAGS_LI_SYNC;
sync_force = l_intro->flags & SIMBRICKS_PROTO_FLAGS_LI_SYNC_FORCE;
......@@ -618,20 +607,23 @@ int SimbricksBaseIfIntroRecv(struct SimbricksBaseIf *base_if,
}
if (upper_proto != base_if->params.upper_layer_proto) {
fprintf(stderr, "SimbricksBaseIfIntroRecv: peer's upper layer proto (%lx) "
"does not match ours (%lx)\n",
fprintf(stderr,
"SimbricksBaseIfIntroRecv: peer's upper layer proto (%lx) "
"does not match ours (%lx)\n",
upper_proto, base_if->params.upper_layer_proto);
return -1;
}
if (sync_force && base_if->params.sync_mode == kSimbricksBaseIfSyncDisabled) {
fprintf(stderr, "SimbricksBaseIfIntroRecv: peer forced sync but we haved "
"it disabled.\n");
fprintf(stderr,
"SimbricksBaseIfIntroRecv: peer forced sync but we haved "
"it disabled.\n");
return -1;
} else if (!sync && !sync_force &&
base_if->params.sync_mode == kSimbricksBaseIfSyncRequired) {
fprintf(stderr, "SimbricksBaseIfIntroRecv: sync required locally, put peer "
"offers no sync.\n");
fprintf(stderr,
"SimbricksBaseIfIntroRecv: sync required locally, put peer "
"offers no sync.\n");
return -1;
} else if (base_if->params.sync_mode == kSimbricksBaseIfSyncDisabled) {
base_if->sync = false;
......@@ -639,32 +631,31 @@ int SimbricksBaseIfIntroRecv(struct SimbricksBaseIf *base_if,
base_if->sync = sync || sync_force;
}
size_t upper_layer_len = (size_t) ret - upper_off;
size_t upper_layer_len = (size_t)ret - upper_off;
if (*payload_len < upper_layer_len) {
fprintf(stderr, "SimbricksBaseIfIntroRecv: upper layer intro does not "
"fit in provided buffer\n");
fprintf(stderr,
"SimbricksBaseIfIntroRecv: upper layer intro does not "
"fit in provided buffer\n");
return -1;
}
memcpy(payload, intro_buf + upper_off, upper_layer_len);
*payload_len = upper_layer_len;
if (!base_if->listener) {
// handle shm setup
struct SimbricksProtoListenerIntro *l_intro =
(struct SimbricksProtoListenerIntro *) intro_buf;
(struct SimbricksProtoListenerIntro *)intro_buf;
cmsg = CMSG_FIRSTHDR(&msg);
if (msg.msg_controllen <= 0 ||
cmsg->cmsg_len != CMSG_LEN(sizeof(int)))
{
if (msg.msg_controllen <= 0 || cmsg->cmsg_len != CMSG_LEN(sizeof(int))) {
/* TODO fix error handling (leaking fds) */
fprintf(stderr, "SimbricksBaseIfIntroRecv: getting shm fd failed (%zu) "
"(%p != %zu)\n", msg.msg_controllen, cmsg,
CMSG_LEN(sizeof(int)));
fprintf(stderr,
"SimbricksBaseIfIntroRecv: getting shm fd failed (%zu) "
"(%p != %zu)\n",
msg.msg_controllen, cmsg, CMSG_LEN(sizeof(int)));
return -1;
}
int shmfd = *(int *) CMSG_DATA(cmsg);
int shmfd = *(int *)CMSG_DATA(cmsg);
if ((base_if->shm = calloc(1, sizeof(*base_if->shm))) == NULL) {
fprintf(stderr, "SimbricksBaseIfIntroRecv: getting shm fd failed\n");
return -1;
......@@ -691,8 +682,9 @@ int SimbricksBaseIfIntroRecv(struct SimbricksBaseIf *base_if,
} else if (base_if->conn_state == kConnAwaitHandshakeRxTx) {
base_if->conn_state = kConnAwaitHandshakeTx;
} else {
fprintf(stderr, "SimbricksBaseIfIntroRecv: connection in unexpected "
"state at the end.\n");
fprintf(stderr,
"SimbricksBaseIfIntroRecv: connection in unexpected "
"state at the end.\n");
abort();
}
......@@ -700,8 +692,7 @@ int SimbricksBaseIfIntroRecv(struct SimbricksBaseIf *base_if,
}
/** FD to wait on for intro events. */
int SimbricksBaseIfIntroFd(struct SimbricksBaseIf *base_if)
{
int SimbricksBaseIfIntroFd(struct SimbricksBaseIf *base_if) {
switch (base_if->conn_state) {
case kConnAwaitHandshakeRxTx: /* FALLTRHOUGH */
case kConnAwaitHandshakeRx: /* FALLTRHOUGH */
......@@ -714,8 +705,7 @@ int SimbricksBaseIfIntroFd(struct SimbricksBaseIf *base_if)
}
int SimBricksBaseIfEstablish(struct SimBricksBaseIfEstablishData *ifs,
size_t n)
{
size_t n) {
struct pollfd pfds[n];
unsigned n_pfd;
size_t established = 0;
......@@ -730,8 +720,10 @@ int SimBricksBaseIfEstablish(struct SimBricksBaseIfEstablishData *ifs,
// woops something went wrong on this connection
if (bif->conn_state == kConnClosed) {
fprintf(stderr, "SimBricksBaseIfEstablish: connection %zu is "
"closed\n", i);
fprintf(stderr,
"SimBricksBaseIfEstablish: connection %zu is "
"closed\n",
i);
return -1;
}
......@@ -742,8 +734,8 @@ int SimBricksBaseIfEstablish(struct SimBricksBaseIfEstablishData *ifs,
return -1;
} else if (ret > 0) {
pfds[n_pfd].fd = SimbricksBaseIfConnFd(bif);
pfds[n_pfd].events = (bif->conn_state == kConnListening ? POLLIN :
POLLOUT);
pfds[n_pfd].events =
(bif->conn_state == kConnListening ? POLLIN : POLLOUT);
pfds[n_pfd].revents = 0;
n_pfd++;
assert(n_pfd <= n);
......@@ -751,11 +743,13 @@ int SimBricksBaseIfEstablish(struct SimBricksBaseIfEstablishData *ifs,
// next check if we are now ready to send the handshake
if ((bif->conn_state == kConnAwaitHandshakeTx ||
bif->conn_state == kConnAwaitHandshakeRxTx) &&
SimbricksBaseIfIntroSend(bif, ifs[i].tx_intro, ifs[i].tx_intro_len)
!= 0) {
fprintf(stderr, "SimBricksBaseIfEstablish: Sending intro on %zu "
"failed\n", i);
bif->conn_state == kConnAwaitHandshakeRxTx) &&
SimbricksBaseIfIntroSend(bif, ifs[i].tx_intro, ifs[i].tx_intro_len) !=
0) {
fprintf(stderr,
"SimBricksBaseIfEstablish: Sending intro on %zu "
"failed\n",
i);
return -1;
}
......@@ -764,8 +758,10 @@ int SimBricksBaseIfEstablish(struct SimBricksBaseIfEstablishData *ifs,
ret = SimbricksBaseIfIntroRecv(bif, ifs[i].rx_intro,
&ifs[i].rx_intro_len);
if (ret < 0) {
fprintf(stderr, "SimBricksBaseIfEstablish: Receiving intro on %zu "
"failed\n", i);
fprintf(stderr,
"SimBricksBaseIfEstablish: Receiving intro on %zu "
"failed\n",
i);
return -1;
} else if (ret > 0) {
pfds[n_pfd].fd = SimbricksBaseIfIntroFd(bif);
......@@ -782,8 +778,9 @@ int SimBricksBaseIfEstablish(struct SimBricksBaseIfEstablishData *ifs,
}
if (n_pfd == 0 && established != n) {
fprintf(stderr, "SimBricksBaseIfEstablish: no poll events to wait for "
"but not all established (BUG)\n");
fprintf(stderr,
"SimBricksBaseIfEstablish: no poll events to wait for "
"but not all established (BUG)\n");
abort();
} else if (n_pfd > 0) {
ret = poll(pfds, n_pfd, -1);
......@@ -797,8 +794,7 @@ int SimBricksBaseIfEstablish(struct SimBricksBaseIfEstablishData *ifs,
return 0;
}
void SimbricksBaseIfClose(struct SimbricksBaseIf *base_if)
{
void SimbricksBaseIfClose(struct SimbricksBaseIf *base_if) {
if (base_if->conn_state == kConnListening) {
close(base_if->listen_fd);
base_if->listen_fd = -1;
......@@ -811,7 +807,8 @@ void SimbricksBaseIfClose(struct SimbricksBaseIf *base_if)
if (base_if->conn_state == kConnOpen) {
// send out termination message
volatile union SimbricksProtoBaseMsg *msg;
while ((msg = SimbricksBaseIfOutAlloc(base_if, UINT64_MAX)) == NULL);
while ((msg = SimbricksBaseIfOutAlloc(base_if, UINT64_MAX)) == NULL)
;
SimbricksBaseIfOutSend(base_if, msg, SIMBRICKS_PROTO_MSG_TYPE_TERMINATE);
}
......@@ -822,7 +819,6 @@ void SimbricksBaseIfClose(struct SimbricksBaseIf *base_if)
// TODO: if connecting end might need to unmap and free shm
}
void SimbricksBaseIfUnlink(struct SimbricksBaseIf *base_if)
{
void SimbricksBaseIfUnlink(struct SimbricksBaseIf *base_if) {
// TODO
}
\ No newline at end of file
lib/simbricks/base/if.h
View file @ f9a3a2e6
......@@ -32,9 +32,9 @@
#include <stdatomic.h>
#endif
#include <stdbool.h>
#include <stddef.h>
#include <stdint.h>
#include <stdbool.h>
#include <simbricks/base/proto.h>
......@@ -117,13 +117,11 @@ struct SimBricksBaseIfEstablishData {
size_t rx_intro_len;
};
/** Create and map a new shared memory pool with the specified path and size. */
int SimbricksBaseIfSHMPoolCreate(struct SimbricksBaseIfSHMPool *pool,
const char *path, size_t pool_size);
/** Map existing shared memory pool by file descriptor. */
int SimbricksBaseIfSHMPoolMapFd(struct SimbricksBaseIfSHMPool *pool,
int fd);
int SimbricksBaseIfSHMPoolMapFd(struct SimbricksBaseIfSHMPool *pool, int fd);
/** Map existing shared memory pool by path. */
int SimbricksBaseIfSHMPoolMap(struct SimbricksBaseIfSHMPool *pool,
const char *path);
......@@ -132,7 +130,6 @@ int SimbricksBaseIfSHMPoolUnmap(struct SimbricksBaseIfSHMPool *pool);
/** Delete but don't unmap shared memory pool. */
int SimbricksBaseIfSHMPoolUnlink(struct SimbricksBaseIfSHMPool *pool);
/** Initialize params struct with default values */
void SimbricksBaseIfDefaultParams(struct SimbricksBaseIfParams *params);
......@@ -158,8 +155,8 @@ int SimbricksBaseIfConnsWait(struct SimbricksBaseIf **base_ifs, unsigned n);
int SimbricksBaseIfIntroSend(struct SimbricksBaseIf *base_if,
const void *payload, size_t payload_len);
/** Receive intro. */
int SimbricksBaseIfIntroRecv(struct SimbricksBaseIf *base_if,
void *payload, size_t *payload_len);
int SimbricksBaseIfIntroRecv(struct SimbricksBaseIf *base_if, void *payload,
size_t *payload_len);
/** FD to wait on for intro events. */
int SimbricksBaseIfIntroFd(struct SimbricksBaseIf *base_if);
......@@ -190,7 +187,6 @@ static inline uint8_t SimbricksBaseIfInType(
struct SimbricksBaseIf *base_if,
volatile union SimbricksProtoBaseMsg *msg) {
return (msg->header.own_type & ~SIMBRICKS_PROTO_MSG_OWN_MASK);
}
/**
......@@ -202,18 +198,17 @@ static inline uint8_t SimbricksBaseIfInType(
* @return Pointer to the message struct if successful, NULL otherwise.
*/
static inline volatile union SimbricksProtoBaseMsg *SimbricksBaseIfInPeek(
struct SimbricksBaseIf *base_if,
uint64_t timestamp) {
struct SimbricksBaseIf *base_if, uint64_t timestamp) {
volatile union SimbricksProtoBaseMsg *msg =
(volatile union SimbricksProtoBaseMsg *) (void *) (
(uint8_t *) base_if->in_queue + base_if->in_pos * base_if->in_elen);
(volatile union SimbricksProtoBaseMsg *)(void *)((uint8_t *)
base_if->in_queue +
base_if->in_pos *
base_if->in_elen);
uint8_t own_type = atomic_load_explicit(
(volatile _Atomic(uint8_t) *) &msg->header.own_type,
memory_order_acquire);
(volatile _Atomic(uint8_t) *)&msg->header.own_type, memory_order_acquire);
/* message not ready */
if ((own_type & SIMBRICKS_PROTO_MSG_OWN_MASK) !=
SIMBRICKS_PROTO_MSG_OWN_CON)
if ((own_type & SIMBRICKS_PROTO_MSG_OWN_MASK) != SIMBRICKS_PROTO_MSG_OWN_CON)
return NULL;
/* if in sync mode, wait till message is ready */
......@@ -233,8 +228,7 @@ static inline volatile union SimbricksProtoBaseMsg *SimbricksBaseIfInPeek(
* @return Pointer to the message struct if successful, NULL otherwise.
*/
static inline volatile union SimbricksProtoBaseMsg *SimbricksBaseIfInPoll(
struct SimbricksBaseIf *base_if,
uint64_t timestamp) {
struct SimbricksBaseIf *base_if, uint64_t timestamp) {
volatile union SimbricksProtoBaseMsg *msg =
SimbricksBaseIfInPeek(base_if, timestamp);
......@@ -263,9 +257,9 @@ static inline void SimbricksBaseIfInDone(
struct SimbricksBaseIf *base_if,
volatile union SimbricksProtoBaseMsg *msg) {
atomic_store_explicit(
(volatile _Atomic(uint8_t) *) &msg->header.own_type,
(uint8_t) ((msg->header.own_type & ~SIMBRICKS_PROTO_MSG_OWN_MASK) |
SIMBRICKS_PROTO_MSG_OWN_PRO),
(volatile _Atomic(uint8_t) *)&msg->header.own_type,
(uint8_t)((msg->header.own_type & ~SIMBRICKS_PROTO_MSG_OWN_MASK) |
SIMBRICKS_PROTO_MSG_OWN_PRO),
memory_order_release);
}
......@@ -286,8 +280,7 @@ static inline uint64_t SimbricksBaseIfInTimestamp(
*
* @param base_if Base interface handle (connected).
*/
static inline int SimbricksBaseIfInTerminated(
struct SimbricksBaseIf *base_if) {
static inline int SimbricksBaseIfInTerminated(struct SimbricksBaseIf *base_if) {
return base_if->in_terminated;
}
......@@ -300,15 +293,15 @@ static inline int SimbricksBaseIfInTerminated(
* @return Pointer to the message struct if successful, NULL otherwise.
*/
static inline volatile union SimbricksProtoBaseMsg *SimbricksBaseIfOutAlloc(
struct SimbricksBaseIf *base_if,
uint64_t timestamp) {
struct SimbricksBaseIf *base_if, uint64_t timestamp) {
volatile union SimbricksProtoBaseMsg *msg =
(volatile union SimbricksProtoBaseMsg *) (void *) (
(uint8_t *) base_if->out_queue + base_if->out_pos * base_if->out_elen);
(volatile union SimbricksProtoBaseMsg *)(void *)((uint8_t *)
base_if->out_queue +
base_if->out_pos *
base_if->out_elen);
uint8_t own_type = atomic_load_explicit(
(volatile _Atomic(uint8_t) *) &msg->header.own_type,
memory_order_acquire);
(volatile _Atomic(uint8_t) *)&msg->header.own_type, memory_order_acquire);
if ((own_type & SIMBRICKS_PROTO_MSG_OWN_MASK) !=
SIMBRICKS_PROTO_MSG_OWN_PRO) {
return NULL;
......@@ -318,7 +311,7 @@ static inline volatile union SimbricksProtoBaseMsg *SimbricksBaseIfOutAlloc(
base_if->out_timestamp = timestamp;
base_if->out_pos = (base_if->out_pos + 1) % base_if->out_enum;
return msg;
return msg;
}
/**
......@@ -334,8 +327,8 @@ static inline volatile union SimbricksProtoBaseMsg *SimbricksBaseIfOutAlloc(
static inline void SimbricksBaseIfOutSend(
struct SimbricksBaseIf *base_if, volatile union SimbricksProtoBaseMsg *msg,
uint8_t msg_type) {
atomic_store_explicit((volatile _Atomic(uint8_t) *) &msg->header.own_type,
(uint8_t) (msg_type | SIMBRICKS_PROTO_MSG_OWN_CON),
atomic_store_explicit((volatile _Atomic(uint8_t) *)&msg->header.own_type,
(uint8_t)(msg_type | SIMBRICKS_PROTO_MSG_OWN_CON),
memory_order_release);
}
......@@ -349,9 +342,9 @@ static inline void SimbricksBaseIfOutSend(
*/
static inline int SimbricksBaseIfOutSync(struct SimbricksBaseIf *base_if,
uint64_t timestamp) {
if (!base_if->sync || (base_if->out_timestamp > 0 &&
timestamp - base_if->out_timestamp <
base_if->params.sync_interval))
if (!base_if->sync ||
(base_if->out_timestamp > 0 &&
timestamp - base_if->out_timestamp < base_if->params.sync_interval))
return 0;
volatile union SimbricksProtoBaseMsg *msg =
......@@ -360,7 +353,7 @@ static inline int SimbricksBaseIfOutSync(struct SimbricksBaseIf *base_if,
return -1;
SimbricksBaseIfOutSend(base_if, msg, SIMBRICKS_PROTO_MSG_TYPE_SYNC);
return 0;
return 0;
}
/**
......@@ -370,8 +363,7 @@ static inline int SimbricksBaseIfOutSync(struct SimbricksBaseIf *base_if,
* @return Timestamp. Undefined if synchronization is disabled.
*/
static inline uint64_t SimbricksBaseIfOutNextSync(
struct SimbricksBaseIf *base_if)
{
struct SimbricksBaseIf *base_if) {
if (base_if->out_timestamp == UINT64_MAX)
return UINT64_MAX;
return base_if->out_timestamp + base_if->params.sync_interval;
......@@ -379,7 +371,7 @@ static inline uint64_t SimbricksBaseIfOutNextSync(
/**
* Retrieve maximal total message length for outgoing messages.
*
*
* @param base_if Base interface handle (connected).
* @return Maximal message length in bytes.
*/
......
lib/simbricks/base/proto.h
View file @ f9a3a2e6
......@@ -28,14 +28,14 @@
#include <assert.h>
#include <stdint.h>
#define SIMBRICKS_PROTO_MSG_SZCHECK(s) static_assert(sizeof(s) == 64, \
"SimBrick message size check failed")
#define SIMBRICKS_PROTO_MSG_SZCHECK(s) \
static_assert(sizeof(s) == 64, "SimBrick message size check failed")
#define SIMBRICKS_PROTO_VERSION 1
#define SIMBRICKS_PROTO_ID_BASE 0x00
#define SIMBRICKS_PROTO_ID_NET 0x01
#define SIMBRICKS_PROTO_ID_PCIE 0x02
#define SIMBRICKS_PROTO_ID_BASE 0x00
#define SIMBRICKS_PROTO_ID_NET 0x01
#define SIMBRICKS_PROTO_ID_PCIE 0x02
/** Listener requests synchronization */
#define SIMBRICKS_PROTO_FLAGS_LI_SYNC (1 << 0)
......@@ -76,7 +76,6 @@ struct SimbricksProtoListenerIntro {
uint64_t upper_layer_intro_off;
} __attribute__((packed));
/** Connecter has synchronization enabled */
#define SIMBRICKS_PROTO_FLAGS_CO_SYNC (1 << 0)
/** Connecter forces synchronization */
......@@ -95,7 +94,6 @@ struct SimbricksProtoConnecterIntro {
uint64_t upper_layer_intro_off;
} __attribute__((packed));
/** Mask for ownership bit in own_type field */
#define SIMBRICKS_PROTO_MSG_OWN_MASK 0x80
/** Message is owned by producer */
......
lib/simbricks/network/if.c
View file @ f9a3a2e6
......@@ -28,8 +28,7 @@
#include <stdio.h>
#include <string.h>
void SimbricksNetIfDefaultParams(struct SimbricksBaseIfParams *params)
{
void SimbricksNetIfDefaultParams(struct SimbricksBaseIfParams *params) {
SimbricksBaseIfDefaultParams(params);
params->in_entries_size = params->out_entries_size = 1536 + 64;
params->upper_layer_proto = SIMBRICKS_PROTO_ID_NET;
......@@ -37,9 +36,7 @@ void SimbricksNetIfDefaultParams(struct SimbricksBaseIfParams *params)
int SimbricksNetIfInit(struct SimbricksNetIf *nsif,
struct SimbricksBaseIfParams *params,
const char *eth_socket_path,
int *sync_eth)
{
const char *eth_socket_path, int *sync_eth) {
// some threaded code using this interface
struct SimbricksBaseIfParams params_ = *params;
struct SimbricksBaseIf *bif = &nsif->base;
......
lib/simbricks/network/if.h
View file @ f9a3a2e6
......@@ -28,19 +28,17 @@
#include <stddef.h>
#include <stdint.h>
#include <simbricks/network/proto.h>
#include <simbricks/base/generic.h>
#include <simbricks/network/proto.h>
struct SimbricksNetIf {
struct SimbricksBaseIf base;
struct SimbricksBaseIf base;
};
void SimbricksNetIfDefaultParams(struct SimbricksBaseIfParams *params);
int SimbricksNetIfInit(struct SimbricksNetIf *nsif,
struct SimbricksBaseIfParams *params,
const char *eth_socket_path,
int *sync_eth);
const char *eth_socket_path, int *sync_eth);
/** Generate queue access functions */
SIMBRICKS_BASEIF_GENERIC(SimbricksNetIf, SimbricksProtoNetMsg, SimbricksNetIf);
......
lib/simbricks/nicbm/multinic.cc
View file @ f9a3a2e6
......@@ -25,11 +25,11 @@
#include "lib/simbricks/nicbm/multinic.h"
#include <string.h>
#include <thread>
#include <vector>
#include <boost/bind.hpp>
#include <boost/fiber/all.hpp>
#include <thread>
#include <vector>
namespace nicbm {
......@@ -42,10 +42,10 @@ int MultiNicRunner::CompRunner::NicIfInit() {
volatile int result = 0;
// NicIfInit will block, so run it in a separate thread and then wait for it
std::thread t([this, &ready, &result](){
result = Runner::NicIfInit();
ready = true;
});
std::thread t([this, &ready, &result]() {
result = Runner::NicIfInit();
ready = true;
});
while (!ready)
YieldPoll();
......@@ -57,7 +57,6 @@ MultiNicRunner::CompRunner::CompRunner(Device &dev) : Runner(dev) {
}
MultiNicRunner::MultiNicRunner(DeviceFactory &factory) : factory_(factory) {
}
int MultiNicRunner::RunMain(int argc, char *argv[]) {
......@@ -66,7 +65,8 @@ int MultiNicRunner::RunMain(int argc, char *argv[]) {
std::vector<boost::fibers::fiber *> fibers;
do {
int end;
for (end = start + 1; end < argc && strcmp(argv[end], "--"); end++);
for (end = start + 1; end < argc && strcmp(argv[end], "--"); end++)
;
argv[start] = argv[0];
CompRunner *r = new CompRunner(factory_.create());
......@@ -80,9 +80,9 @@ int MultiNicRunner::RunMain(int argc, char *argv[]) {
start = end;
} while (start < argc);
for (auto f: fibers) {
for (auto f : fibers) {
f->join();
delete(f);
delete (f);
}
return 0;
}
......
lib/simbricks/nicbm/nicbm.cc
View file @ f9a3a2e6
......@@ -51,11 +51,11 @@ static volatile int exiting = 0;
static std::vector<Runner *> runners;
#ifdef STAT_NICBM
#ifdef STAT_NICBM
static uint64_t h2d_poll_total = 0;
static uint64_t h2d_poll_suc = 0;
static uint64_t h2d_poll_sync = 0;
//count from signal USR2
// count from signal USR2
static uint64_t s_h2d_poll_total = 0;
static uint64_t s_h2d_poll_suc = 0;
static uint64_t s_h2d_poll_sync = 0;
......@@ -63,7 +63,7 @@ static uint64_t s_h2d_poll_sync = 0;
static uint64_t n2d_poll_total = 0;
static uint64_t n2d_poll_suc = 0;
static uint64_t n2d_poll_sync = 0;
//count from signal USR2
// count from signal USR2
static uint64_t s_n2d_poll_total = 0;
static uint64_t s_n2d_poll_suc = 0;
static uint64_t s_n2d_poll_sync = 0;
......@@ -75,7 +75,7 @@ static void sigint_handler(int dummy) {
}
static void sigusr1_handler(int dummy) {
for (Runner *r: runners)
for (Runner *r : runners)
fprintf(stderr, "[%p] main_time = %lu\n", r, r->TimePs());
}
......@@ -130,14 +130,18 @@ void Runner::IssueDma(DMAOp &op) {
if (dma_pending_ < DMA_MAX_PENDING) {
// can directly issue
#ifdef DEBUG_NICBM
printf("main_time = %lu: nicbm: issuing dma op %p addr %lx len %zu pending %zu\n",main_time_, &op,
op.dma_addr_, op.len_, dma_pending_);
printf(
"main_time = %lu: nicbm: issuing dma op %p addr %lx len %zu pending "
"%zu\n",
main_time_, &op, op.dma_addr_, op.len_, dma_pending_);
#endif
DmaDo(op);
} else {
#ifdef DEBUG_NICBM
printf("main_time = %lu: nicbm: enqueuing dma op %p addr %lx len %zu pending %zu\n", main_time_, &op,
op.dma_addr_, op.len_, dma_pending_);
printf(
"main_time = %lu: nicbm: enqueuing dma op %p addr %lx len %zu pending "
"%zu\n",
main_time_, &op, op.dma_addr_, op.len_, dma_pending_);
#endif
dma_queue_.push_back(&op);
}
......@@ -160,8 +164,10 @@ void Runner::DmaDo(DMAOp &op) {
volatile union SimbricksProtoPcieD2H *msg = D2HAlloc();
dma_pending_++;
#ifdef DEBUG_NICBM
printf("main_time = %lu: nicbm: executing dma op %p addr %lx len %zu pending %zu\n",main_time_, &op,
op.dma_addr_, op.len_, dma_pending_);
printf(
"main_time = %lu: nicbm: executing dma op %p addr %lx len %zu pending "
"%zu\n",
main_time_, &op, op.dma_addr_, op.len_, dma_pending_);
#endif
size_t maxlen = SimbricksBaseIfOutMsgLen(&nicif_.pcie.base);
......@@ -179,27 +185,25 @@ void Runner::DmaDo(DMAOp &op) {
write->offset = op.dma_addr_;
write->len = op.len_;
memcpy((void *)write->data, (void *)op.data_, op.len_);
#ifdef DEBUG_NICBM
uint8_t *tmp = (uint8_t*)op.data_;
uint8_t *tmp = (uint8_t *)op.data_;
int d;
printf("main_time = %lu: nicbm: dma write data: \n", main_time_);
for (d = 0; d < op.len_; d++){
for (d = 0; d < op.len_; d++) {
printf("%02X ", *tmp);
tmp++;
}
#endif
SimbricksPcieIfD2HOutSend(&nicif_.pcie, msg,
SIMBRICKS_PROTO_PCIE_D2H_MSG_WRITE);
SIMBRICKS_PROTO_PCIE_D2H_MSG_WRITE);
} else {
volatile struct SimbricksProtoPcieD2HRead *read = &msg->read;
if (maxlen < sizeof(struct SimbricksProtoPcieH2DReadcomp) + op.len_) {
fprintf(stderr,
"issue_dma: write too big (%zu), can only fit up "
"to (%zu)\n",
op.len_,
maxlen - sizeof(struct SimbricksProtoPcieH2DReadcomp));
op.len_, maxlen - sizeof(struct SimbricksProtoPcieH2DReadcomp));
abort();
}
......@@ -207,7 +211,7 @@ void Runner::DmaDo(DMAOp &op) {
read->offset = op.dma_addr_;
read->len = op.len_;
SimbricksPcieIfD2HOutSend(&nicif_.pcie, msg,
SIMBRICKS_PROTO_PCIE_D2H_MSG_READ);
SIMBRICKS_PROTO_PCIE_D2H_MSG_READ);
}
}
......@@ -217,14 +221,15 @@ void Runner::MsiIssue(uint8_t vec) {
volatile union SimbricksProtoPcieD2H *msg = D2HAlloc();
#ifdef DEBUG_NICBM
printf("main_time = %lu: nicbm: issue MSI interrupt vec %u\n", main_time_, vec);
printf("main_time = %lu: nicbm: issue MSI interrupt vec %u\n", main_time_,
vec);
#endif
volatile struct SimbricksProtoPcieD2HInterrupt *intr = &msg->interrupt;
intr->vector = vec;
intr->inttype = SIMBRICKS_PROTO_PCIE_INT_MSI;
SimbricksPcieIfD2HOutSend(&nicif_.pcie, msg,
SIMBRICKS_PROTO_PCIE_D2H_MSG_INTERRUPT);
SIMBRICKS_PROTO_PCIE_D2H_MSG_INTERRUPT);
}
void Runner::MsiXIssue(uint8_t vec) {
......@@ -233,14 +238,15 @@ void Runner::MsiXIssue(uint8_t vec) {
volatile union SimbricksProtoPcieD2H *msg = D2HAlloc();
#ifdef DEBUG_NICBM
printf("main_time = %lu: nicbm: issue MSI-X interrupt vec %u\n", main_time_, vec);
printf("main_time = %lu: nicbm: issue MSI-X interrupt vec %u\n", main_time_,
vec);
#endif
volatile struct SimbricksProtoPcieD2HInterrupt *intr = &msg->interrupt;
intr->vector = vec;
intr->inttype = SIMBRICKS_PROTO_PCIE_INT_MSIX;
SimbricksPcieIfD2HOutSend(&nicif_.pcie, msg,
SIMBRICKS_PROTO_PCIE_D2H_MSG_INTERRUPT);
SIMBRICKS_PROTO_PCIE_D2H_MSG_INTERRUPT);
}
void Runner::IntXIssue(bool level) {
......@@ -257,7 +263,7 @@ void Runner::IntXIssue(bool level) {
: SIMBRICKS_PROTO_PCIE_INT_LEGACY_LO);
SimbricksPcieIfD2HOutSend(&nicif_.pcie, msg,
SIMBRICKS_PROTO_PCIE_D2H_MSG_INTERRUPT);
SIMBRICKS_PROTO_PCIE_D2H_MSG_INTERRUPT);
}
void Runner::EventSchedule(TimedEvent &evt) {
......@@ -281,12 +287,12 @@ void Runner::H2DRead(volatile struct SimbricksProtoPcieH2DRead *read) {
#ifdef DEBUG_NICBM
uint64_t dbg_val = 0;
memcpy(&dbg_val, (const void *)rc->data, read->len <= 8 ? read->len : 8);
printf("main_time = %lu: nicbm: read(off=0x%lx, len=%u, val=0x%lx)\n", main_time_, read->offset, read->len,
dbg_val);
printf("main_time = %lu: nicbm: read(off=0x%lx, len=%u, val=0x%lx)\n",
main_time_, read->offset, read->len, dbg_val);
#endif
SimbricksPcieIfD2HOutSend(&nicif_.pcie, msg,
SIMBRICKS_PROTO_PCIE_D2H_MSG_READCOMP);
SIMBRICKS_PROTO_PCIE_D2H_MSG_READCOMP);
}
void Runner::H2DWrite(volatile struct SimbricksProtoPcieH2DWrite *write) {
......@@ -299,22 +305,22 @@ void Runner::H2DWrite(volatile struct SimbricksProtoPcieH2DWrite *write) {
#ifdef DEBUG_NICBM
uint64_t dbg_val = 0;
memcpy(&dbg_val, (const void *)write->data, write->len <= 8 ? write->len : 8);
printf("main_time = %lu: nicbm: write(off=0x%lx, len=%u, val=0x%lx)\n", main_time_, write->offset,
write->len, dbg_val);
printf("main_time = %lu: nicbm: write(off=0x%lx, len=%u, val=0x%lx)\n",
main_time_, write->offset, write->len, dbg_val);
#endif
dev_.RegWrite(write->bar, write->offset, (void *)write->data, write->len);
wc->req_id = write->req_id;
SimbricksPcieIfD2HOutSend(&nicif_.pcie, msg,
SIMBRICKS_PROTO_PCIE_D2H_MSG_WRITECOMP);
SIMBRICKS_PROTO_PCIE_D2H_MSG_WRITECOMP);
}
void Runner::H2DReadcomp(volatile struct SimbricksProtoPcieH2DReadcomp *rc) {
DMAOp *op = (DMAOp *)(uintptr_t)rc->req_id;
#ifdef DEBUG_NICBM
printf("main_time = %lu: nicbm: completed dma read op %p addr %lx len %zu\n", main_time_, op, op->dma_addr_,
op->len_);
printf("main_time = %lu: nicbm: completed dma read op %p addr %lx len %zu\n",
main_time_, op, op->dma_addr_, op->len_);
#endif
memcpy(op->data_, (void *)rc->data, op->len_);
......@@ -328,8 +334,8 @@ void Runner::H2DWritecomp(volatile struct SimbricksProtoPcieH2DWritecomp *wc) {
DMAOp *op = (DMAOp *)(uintptr_t)wc->req_id;
#ifdef DEBUG_NICBM
printf("main_time = %lu: nicbm: completed dma write op %p addr %lx len %zu\n", main_time_, op,
op->dma_addr_, op->len_);
printf("main_time = %lu: nicbm: completed dma write op %p addr %lx len %zu\n",
main_time_, op, op->dma_addr_, op->len_);
#endif
dev_.DmaComplete(*op);
......@@ -344,7 +350,8 @@ void Runner::H2DDevctrl(volatile struct SimbricksProtoPcieH2DDevctrl *dc) {
void Runner::EthRecv(volatile struct SimbricksProtoNetMsgPacket *packet) {
#ifdef DEBUG_NICBM
printf("main_time = %lu: nicbm: eth rx: port %u len %u\n", main_time_, packet->port, packet->len);
printf("main_time = %lu: nicbm: eth rx: port %u len %u\n", main_time_,
packet->port, packet->len);
#endif
dev_.EthRx(packet->port, (void *)packet->data, packet->len);
......@@ -360,8 +367,7 @@ void Runner::EthSend(const void *data, size_t len) {
packet->port = 0; // single port
packet->len = len;
memcpy((void *)packet->data, data, len);
SimbricksNetIfOutSend(&nicif_.net, msg,
SIMBRICKS_PROTO_NET_MSG_PACKET);
SimbricksNetIfOutSend(&nicif_.net, msg, SIMBRICKS_PROTO_NET_MSG_PACKET);
}
void Runner::PollH2D() {
......@@ -371,7 +377,7 @@ void Runner::PollH2D() {
#ifdef STAT_NICBM
h2d_poll_total += 1;
if (stat_flag){
if (stat_flag) {
s_h2d_poll_total += 1;
}
#endif
......@@ -381,7 +387,7 @@ void Runner::PollH2D() {
#ifdef STAT_NICBM
h2d_poll_suc += 1;
if (stat_flag){
if (stat_flag) {
s_h2d_poll_suc += 1;
}
#endif
......@@ -411,7 +417,7 @@ void Runner::PollH2D() {
case SIMBRICKS_PROTO_MSG_TYPE_SYNC:
#ifdef STAT_NICBM
h2d_poll_sync += 1;
if (stat_flag){
if (stat_flag) {
s_h2d_poll_sync += 1;
}
#endif
......@@ -435,7 +441,7 @@ void Runner::PollN2D() {
#ifdef STAT_NICBM
n2d_poll_total += 1;
if (stat_flag){
if (stat_flag) {
s_n2d_poll_total += 1;
}
#endif
......@@ -445,7 +451,7 @@ void Runner::PollN2D() {
#ifdef STAT_NICBM
n2d_poll_suc += 1;
if (stat_flag){
if (stat_flag) {
s_n2d_poll_suc += 1;
}
#endif
......@@ -459,10 +465,10 @@ void Runner::PollN2D() {
case SIMBRICKS_PROTO_MSG_TYPE_SYNC:
#ifdef STAT_NICBM
n2d_poll_sync += 1;
if (stat_flag){
if (stat_flag) {
s_n2d_poll_sync += 1;
}
#endif
#endif
break;
default:
......@@ -544,7 +550,7 @@ int Runner::ParseArgs(int argc, char *argv[]) {
main_time_ = strtoull(argv[5], NULL, 0);
if (argc >= 7)
netParams_.sync_interval = pcieParams_.sync_interval =
strtoull(argv[6], NULL, 0) * 1000ULL;
strtoull(argv[6], NULL, 0) * 1000ULL;
if (argc >= 8)
pcieParams_.link_latency = strtoull(argv[7], NULL, 0) * 1000ULL;
if (argc >= 9)
......@@ -637,23 +643,24 @@ int Runner::RunMain() {
(double)(h2d_poll_sync + n2d_poll_sync) / (h2d_poll_suc + n2d_poll_suc));
fprintf(stderr, "%20s: %22lu %20s: %22lu poll_suc_rate: %f\n",
"s_h2d_poll_total", s_h2d_poll_total, "s_h2d_poll_suc", s_h2d_poll_suc,
(double)s_h2d_poll_suc / s_h2d_poll_total);
"s_h2d_poll_total", s_h2d_poll_total, "s_h2d_poll_suc",
s_h2d_poll_suc, (double)s_h2d_poll_suc / s_h2d_poll_total);
fprintf(stderr, "%65s: %22lu sync_rate: %f\n", "s_h2d_poll_sync",
s_h2d_poll_sync, (double)s_h2d_poll_sync / s_h2d_poll_suc);
fprintf(stderr, "%20s: %22lu %20s: %22lu poll_suc_rate: %f\n",
"s_n2d_poll_total", s_n2d_poll_total, "s_n2d_poll_suc", s_n2d_poll_suc,
(double)s_n2d_poll_suc / s_n2d_poll_total);
"s_n2d_poll_total", s_n2d_poll_total, "s_n2d_poll_suc",
s_n2d_poll_suc, (double)s_n2d_poll_suc / s_n2d_poll_total);
fprintf(stderr, "%65s: %22lu sync_rate: %f\n", "s_n2d_poll_sync",
s_n2d_poll_sync, (double)s_n2d_poll_sync / s_n2d_poll_suc);
fprintf(
stderr, "%20s: %22lu %20s: %22lu sync_rate: %f\n", "s_recv_total",
s_h2d_poll_suc + s_n2d_poll_suc, "s_recv_sync", s_h2d_poll_sync + s_n2d_poll_sync,
(double)(s_h2d_poll_sync + s_n2d_poll_sync) / (s_h2d_poll_suc + s_n2d_poll_suc));
fprintf(stderr, "%20s: %22lu %20s: %22lu sync_rate: %f\n", "s_recv_total",
s_h2d_poll_suc + s_n2d_poll_suc, "s_recv_sync",
s_h2d_poll_sync + s_n2d_poll_sync,
(double)(s_h2d_poll_sync + s_n2d_poll_sync) /
(s_h2d_poll_suc + s_n2d_poll_suc));
#endif
SimbricksNicIfCleanup(&nicif_);
......
lib/simbricks/nicbm/nicbm.h
View file @ f9a3a2e6
......@@ -50,7 +50,8 @@ class DMAOp {
class TimedEvent {
public:
TimedEvent() : time_(0), priority_(0) {}
TimedEvent() : time_(0), priority_(0) {
}
virtual ~TimedEvent() = default;
uint64_t time_;
int priority_;
......@@ -120,7 +121,7 @@ class Runner {
struct EventCmp {
bool operator()(TimedEvent *a, TimedEvent *b) const {
return a->time_ < b->time_ ||
(a->time_ == b->time_ && a->priority_ < b->priority_);
(a->time_ == b->time_ && a->priority_ < b->priority_);
}
};
......
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment