dist/net_rdma: passing entries works, cleanup still missing

dist/net_rdma.c

@@ -197,11 +197,8 @@ static int PeersInitDevs() {
 int PeerDevSendIntro(struct Peer *peer) {
   fprintf(stderr, "PeerDevSendIntro(%s)\n", peer->sock_path);
 
-  peer->shm_fd = shm_fd;
-  peer->shm_base = shm_base;
-
   struct SimbricksProtoNetDevIntro *di = &peer->dev_intro;
-  peer->local_base = (void *) ((uintptr_t) shm_base + di->d2n_offset);
+  peer->local_base = (void *) ((uintptr_t) peer->shm_base + di->d2n_offset);
   peer->local_elen = di->d2n_elen;
   peer->local_enum = di->d2n_nentries;
 
@@ -297,6 +294,32 @@ static int PeerEvent(struct Peer *peer, uint32_t events) {
 }
 
 static void *PollThread(void *data) {
+  while (true) {
+    for (size_t i = 0; i < peer_num; i++) {
+      struct Peer *peer = &peers[i];
+      if (!peer->ready)
+        continue;
+
+      void *entry = (peer->local_base + peer->local_pos * peer->local_elen);
+      bool ready;
+      if (peer->is_dev) {
+        struct SimbricksProtoNetD2NDummy *d2n = entry;
+        ready = (d2n->own_type & SIMBRICKS_PROTO_NET_D2N_OWN_MASK) ==
+            SIMBRICKS_PROTO_NET_D2N_OWN_NET;
+      } else {
+        struct SimbricksProtoNetN2DDummy *n2d = entry;
+        ready = (n2d->own_type & SIMBRICKS_PROTO_NET_N2D_OWN_MASK) ==
+            SIMBRICKS_PROTO_NET_N2D_OWN_DEV;
+      }
+
+      if (ready) {
+        RdmaPassEntry(peer);
+        peer->local_pos += 1;
+        if (peer->local_pos >= peer->local_enum)
+          peer->local_pos -= peer->local_enum;
+      }
+    }
+  }
   return NULL;
 }
 

dist/net_rdma.h

@@ -38,7 +38,7 @@ struct Peer {
 
   /* base address of the local queue we're polling.
      (d2n or n2d depending on is_dev). */
-  void *local_base;
+  uint8_t *local_base;
   uint64_t local_elen;
   uint64_t local_enum;
   uint64_t local_pos;
@@ -82,6 +82,7 @@ int PeerNetSetupQueues(struct Peer *peer);
 int RdmaListen(struct sockaddr_in *addr);
 int RdmaConnect(struct sockaddr_in *addr);
 int RdmaPassIntro(struct Peer *peer);
+int RdmaPassEntry(struct Peer *peer);
 int RdmaEvent();
 
 #endif  // DIST_NET_RDMA_NET_RDMA_H_
\ No newline at end of file

dist/rdma.c

@@ -164,14 +164,13 @@ static int RdmaCommonInit() {
     return 1;
   }
 
-  qp_attr.cap.max_send_wr = MSG_TXBUFS;
+  qp_attr.cap.max_send_wr = 1024;
   qp_attr.cap.max_send_sge = 1;
   qp_attr.cap.max_recv_wr = MSG_RXBUFS;
   qp_attr.cap.max_recv_sge = 1;
   qp_attr.send_cq = cq;
   qp_attr.recv_cq = cq;
   qp_attr.qp_type = IBV_QPT_RC;
-  qp_attr.sq_sig_all = 1;
 
   if (rdma_create_qp(cm_id, pd, &qp_attr)) {
     perror("RdmaCommonInit: rdma_create_qp failed");
@@ -471,3 +470,29 @@ int RdmaPassIntro(struct Peer *peer) {
 
   return 0;
 }
+
+int RdmaPassEntry(struct Peer *peer) {
+  fprintf(stderr, "RdmaPassEntry(%s,%lu)\n", peer->sock_path, peer->local_pos);
+  fprintf(stderr, "  remote_base=%lx local_base=%lx\n", peer->remote_base,
+          peer->local_base);
+  uint64_t pos = peer->local_pos * peer->local_elen;
+  struct ibv_sge sge;
+  sge.addr = (uintptr_t) (peer->local_base + pos);
+  sge.length = peer->local_elen;
+  sge.lkey = peer->shm_mr->lkey;
+
+  struct ibv_send_wr send_wr = { };
+  send_wr.wr_id = -1ULL;
+  send_wr.opcode = IBV_WR_RDMA_WRITE;
+  send_wr.wr.rdma.remote_addr = peer->remote_base + pos;
+  send_wr.wr.rdma.rkey = peer->remote_rkey;
+  send_wr.sg_list = &sge;
+  send_wr.num_sge = 1;
+
+  struct ibv_send_wr *bad_send_wr;
+  if (ibv_post_send(cm_id->qp, &send_wr, &bad_send_wr)) {
+    perror("RdmaPassEntry: ibv_post_send failed");
+    return 1;
+  }
+  return 0;
+}
\ No newline at end of file