rdma.c 14.1 KB
Newer Older
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
/*
 * Copyright 2021 Max Planck Institute for Software Systems, and
 * National University of Singapore
 *
 * Permission is hereby granted, free of charge, to any person obtaining
 * a copy of this software and associated documentation files (the
 * "Software"), to deal in the Software without restriction, including
 * without limitation the rights to use, copy, modify, merge, publish,
 * distribute, sublicense, and/or sell copies of the Software, and to
 * permit persons to whom the Software is furnished to do so, subject to
 * the following conditions:
 *
 * The above copyright notice and this permission notice shall be
 * included in all copies or substantial portions of the Software.
 *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
 * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
 * MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
 * IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY
 * CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT,
 * TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE
 * SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 */

25
26
#include "dist/rdma/rdma.h"
#include "dist/rdma/net_rdma.h"
27
28

#include <fcntl.h>
29
#include <infiniband/verbs.h>
30
#include <pthread.h>
31
32
33
34
35
#include <stdio.h>
#include <stdlib.h>
#include <sys/epoll.h>
#include <unistd.h>

36
#define SENDQ_LEN (8 * 1024)
37
38
#define MSG_RXBUFS 512
#define MSG_TXBUFS 512
39
#define MAX_PEERS 32
40
#define SIG_THRESHOLD 32
41

Antoine Kaufmann's avatar
Antoine Kaufmann committed
42
43
44
45
46
struct NetRdmaIntroMsg {
  uint32_t payload_len;
  uint8_t data[1024];
} __attribute__((packed));

47
48
49
50
struct NetRdmaReportMsg {
  uint32_t written_pos[MAX_PEERS];
  uint32_t clean_pos[MAX_PEERS];
  bool valid[MAX_PEERS];
Antoine Kaufmann's avatar
Antoine Kaufmann committed
51
} __attribute__((packed));
52
53
54

struct NetRdmaMsg {
  union {
Antoine Kaufmann's avatar
Antoine Kaufmann committed
55
    struct NetRdmaIntroMsg intro;
56
    struct NetRdmaReportMsg report;
57
58
59
60
61
62
63
    struct NetRdmaMsg *next_free;
  };
  uint64_t id;
  uint64_t base_addr;
  uint64_t queue_off;
  uint64_t rkey;
  enum {
Antoine Kaufmann's avatar
Antoine Kaufmann committed
64
    kMsgIntro,
65
    kMsgReport,
66
  } msg_type;
Antoine Kaufmann's avatar
Antoine Kaufmann committed
67
} __attribute__((packed));
68

69
70
static struct ibv_context *ib_ctx;
static struct ibv_qp *qp;
71
72
73
74
75
76
77
78
static struct ibv_pd *pd;
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 NetRdmaMsg msgs[MSG_RXBUFS + MSG_TXBUFS];
79
pthread_spinlock_t freelist_spin;
80
static struct NetRdmaMsg *msgs_free = NULL;
81
static uint32_t last_signaled = 0;
82
83

static struct NetRdmaMsg *RdmaMsgAlloc() {
84
  pthread_spin_lock(&freelist_spin);
85
86
87
88
  struct NetRdmaMsg *msg = msgs_free;
  if (msg != NULL) {
    msgs_free = msg->next_free;
  }
89
  pthread_spin_unlock(&freelist_spin);
90
91
92
93
  return msg;
}

static void RdmaMsgFree(struct NetRdmaMsg *msg) {
94
  pthread_spin_lock(&freelist_spin);
95
96
  msg->next_free = msgs_free;
  msgs_free = msg;
97
  pthread_spin_unlock(&freelist_spin);
98
99
100
101
102
103
104
105
106
107
108
109
110
}

static int RdmMsgRxEnqueue(struct NetRdmaMsg *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 = { };
  recv_wr.wr_id = msg - msgs;
  recv_wr.sg_list = &sge;
  recv_wr.num_sge = 1;
  struct ibv_recv_wr *bad_recv_wr;
111
  if (ibv_post_recv(qp, &recv_wr, &bad_recv_wr)) {
112
113
114
115
116
117
118
    perror("RdmMsgRxEnqueue: ibv_post_recv failed");
    return 1;
  }

  return 0;
}

119
static int RdmaMsgRxIntro(struct NetRdmaMsg *msg) {
120
  if (msg->id >= peer_num) {
Antoine Kaufmann's avatar
Antoine Kaufmann committed
121
122
    fprintf(stderr, "RdmaMsgRxIntro: invalid peer id in message (%lu)\n",
        msg->id);
123
124
125
126
127
128
129
    abort();
  }

  struct Peer *peer = peers + msg->id;
  printf("RdmMsgRx -> peer %s\n", peer->sock_path);

  if (peer->intro_valid_remote) {
Antoine Kaufmann's avatar
Antoine Kaufmann committed
130
131
    fprintf(stderr, "RdmaMsgRxIntro: received multiple messages (%lu)\n",
        msg->id);
132
133
134
135
136
    abort();
  }

  peer->remote_rkey = msg->rkey;
  peer->remote_base = msg->base_addr + msg->queue_off;
Antoine Kaufmann's avatar
Antoine Kaufmann committed
137

138
  peer->intro_valid_remote = true;
Antoine Kaufmann's avatar
Antoine Kaufmann committed
139
140
141
142
143
144
145
  peer->intro_remote_len = msg->intro.payload_len;
  memcpy(peer->intro_remote, msg->intro.data, msg->intro.payload_len);

  if (BasePeerSetupQueues(peer)) {
    fprintf(stderr, "RdmaMsgRxIntro(%s): queue setup failed\n",
        peer->sock_path);
    abort();
146
  }
Antoine Kaufmann's avatar
Antoine Kaufmann committed
147
148
  if (BasePeerSendIntro(peer))
    return 1;
149
150

  if (peer->intro_valid_local) {
Antoine Kaufmann's avatar
Antoine Kaufmann committed
151
152
153
154
155
156
157
158
    // 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);
      return 1;
    }
    fprintf(stderr, "RdmaMsgRxIntro(%s): marking peer as ready\n",
        peer->sock_path);
159
160
161
162
163
    peer->ready = true;
  }
  return 0;
}

164
165
166
167
168
169
170
171
172
static int RdmaMsgRxReport(struct NetRdmaMsg *msg) {
  for (size_t i = 0; i < MAX_PEERS && i < peer_num; i++) {
    if (!msg->report.valid[i])
      continue;

    if (i >= peer_num) {
      fprintf(stderr, "RdmaMsgRxReport: invalid ready peer number %zu\n", i);
      abort();
    }
Antoine Kaufmann's avatar
Antoine Kaufmann committed
173
174
    BasePeerReport(&peers[i], msg->report.written_pos[i],
                   msg->report.clean_pos[i]);
175
176
177
178
179
  }
  return 0;
}

static int RdmaMsgRx(struct NetRdmaMsg *msg) {
Antoine Kaufmann's avatar
Antoine Kaufmann committed
180
  if (msg->msg_type == kMsgIntro)
181
182
183
184
185
186
187
188
    return RdmaMsgRxIntro(msg);
  else if (msg->msg_type == kMsgReport)
    return RdmaMsgRxReport(msg);

  fprintf(stderr, "RdmaMsgRx: unexpected message type = %u\n", msg->msg_type);
  abort();
}

189
190
191
int RdmaCommonInit(struct ibv_context *ctx) {
  ib_ctx = ctx;

192
193
194
195
196
  if (pthread_spin_init(&freelist_spin, PTHREAD_PROCESS_PRIVATE)) {
    perror("RdmaCommonInit: pthread_spin_init failed");
    return 1;
  }

197
  if (!(pd = ibv_alloc_pd(ib_ctx))) {
198
199
200
201
    perror("RdmaCommonInit: ibv_alloc_pd failed");
    return 1;
  }

202
  if (!(comp_chan = ibv_create_comp_channel(ib_ctx))) {
203
204
205
206
    perror("RdmaCommonInit: ibv_create_comp_channel failed");
    return 1;
  }

207
  if (!(cq = ibv_create_cq(ib_ctx, 1024, NULL, comp_chan, 0))) {
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
    perror("RdmaCommonInit: ibv_create_cq failed");
    return 1;
  }

  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))) {
    perror("RdmaCommonInit: ibv_reg_mr msgs failed");
    return 1;
  }

224
  qp_attr.cap.max_send_wr = SENDQ_LEN;
225
226
227
228
229
230
  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;
231
232
233
234
235
236
237
238

  if (!ib_connect)
    qp = RdmaCMCreateQP(pd, &qp_attr);
  else
    qp = RdmaIBCreateQP(pd, &qp_attr);

  if (!qp) {
    fprintf(stderr, "RdmaCommonInit: RdmaCreateQP failed\n");
239
240
241
242
    return 1;
  }

  if (ibv_req_notify_cq(cq, 0)) {
243
    perror("RdmaCommonInit: ibv_req_notify_cq failed");
244
245
    return 1;
  }
246
#ifdef RDMA_DEBUG
247
  fprintf(stderr, "Enqueue rx buffers\n");
248
#endif
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
  // post receive operations for all rx buffers
  for (int i = 0; i < MSG_RXBUFS; i++)
    if (RdmMsgRxEnqueue(&msgs[i]))
      return 1;

  // add tx buffers to freelist
  for (int i = 0; i < MSG_TXBUFS; i++)
    RdmaMsgFree(&msgs[MSG_RXBUFS + i]);

  return 0;
}

static int RdmaCommonSetNonblock() {
  int flags = fcntl(comp_chan->fd, F_GETFL);
  if (fcntl(comp_chan->fd, F_SETFL, flags | O_NONBLOCK)) {
    perror("RdmaCommonSetNonblock: fcntl set nonblock failed");
    return 1;
  }

  struct epoll_event epev;
  epev.events = EPOLLIN;
  epev.data.ptr = NULL;
  if (epoll_ctl(epfd, EPOLL_CTL_ADD, comp_chan->fd, &epev)) {
    perror("RdmaCommonSetNonblock: epoll_ctl failed");
    return 1;
  }

  return 0;
}

int RdmaListen(struct sockaddr_in *addr) {
280
281
282
283
284
285
286
287
  int ret;

  if (!ib_connect)
    ret = RdmaCMListen(addr);
  else
    ret = RdmaIBListen(addr);

  if (ret)
288
289
290
291
292
293
294
295
296
    return 1;

  if (RdmaCommonSetNonblock())
    return 1;

  return 0;
}

int RdmaConnect(struct sockaddr_in *addr) {
297
298
299
300
301
302
303
304
  int ret;

  if (!ib_connect)
    ret = RdmaCMConnect(addr);
  else
    ret = RdmaIBConnect(addr);

  if (ret)
305
306
307
308
309
310
311
312
313
    return 1;

  if (RdmaCommonSetNonblock())
    return 1;

  return 0;
}

int RdmaEvent() {
314
#ifdef RDMA_DEBUG
315
  fprintf(stderr, "RdmaEvent [pid=%d]\n", getpid());
316
#endif
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338

  struct ibv_cq *ecq;
  void *ectx;
  if (ibv_get_cq_event(comp_chan, &ecq, &ectx)) {
    perror("RdmaEvent: ibv_get_cq_event failed");
    return 1;
  }
  ibv_ack_cq_events(ecq, 1);

  if (ibv_req_notify_cq(cq, 0)) {
    perror("RdmaEvent: ibv_req_notify_cq failed");
    return 1;
  }

  int n;
  do {
    const size_t kNumWC = 8;
    struct ibv_wc wcs[kNumWC];
    if ((n = ibv_poll_cq(cq, kNumWC, wcs)) < 0) {
      perror("RdmaEvent: ibv_poll_cq failed");
      return 1;
    }
339
340

#ifdef RDMA_DEBUG
341
    fprintf(stderr, "  n=%d\n", n);
342
#endif
343
344
    for (int i = 0; i < n; i++) {
      if (wcs[i].opcode == IBV_WC_SEND) {
345
#ifdef RDMA_DEBUG
Antoine Kaufmann's avatar
Antoine Kaufmann committed
346
        fprintf(stderr, "Send done\n");
347
#endif
348
349
350
351
352
353
354
355
        if (wcs[i].status != IBV_WC_SUCCESS) {
          fprintf(stderr, "RdmaEvent: unsuccessful send (%u)\n", wcs[i].status);
          abort();
        }

        // need to free the send buffer again
        RdmaMsgFree(msgs + wcs[i].wr_id);
      } else if ((wcs[i].opcode & IBV_WC_RECV)) {
356
#ifdef RDMA_DEBUG
Antoine Kaufmann's avatar
Antoine Kaufmann committed
357
        fprintf(stderr, "Recv done\n");
358
#endif
359
360
361
362
363
364
365
366

        if (wcs[i].status != IBV_WC_SUCCESS) {
          fprintf(stderr, "RdmaEvent: unsuccessful recv (%u)\n", wcs[i].status);
          abort();
        }
        struct NetRdmaMsg *msg = msgs + wcs[i].wr_id;
        if (RdmaMsgRx(msg) || RdmMsgRxEnqueue(msg))
          return 1;
367
368
      } else if ((wcs[i].opcode & IBV_WC_RDMA_WRITE)) {
        /* just a signalled write every once in a while to clear queue*/
369
370
371
372
373
374
375
376
377
378
379
      } else {
        fprintf(stderr, "RdmaEvent: unexpected opcode %u\n", wcs[i].opcode);
        abort();
      }
    }
  } while (n > 0);

  fflush(stdout);
  return 0;
}

Antoine Kaufmann's avatar
Antoine Kaufmann committed
380
int BaseOpPassIntro(struct Peer *peer) {
381
#ifdef RDMA_DEBUG
Antoine Kaufmann's avatar
Antoine Kaufmann committed
382
  fprintf(stderr, "BaseOpPassIntro(%s)\n", peer->sock_path);
383
#endif
384

Antoine Kaufmann's avatar
Antoine Kaufmann committed
385
386
  // connecting peers have sent us an SHM region, need to register this an as MR
  if (!peer->is_listener) {
387
388
389
    if (!(peer->shm_opaque = ibv_reg_mr(pd, peer->shm_base, peer->shm_size,
                                        IBV_ACCESS_LOCAL_WRITE |
                                        IBV_ACCESS_REMOTE_WRITE))) {
Antoine Kaufmann's avatar
Antoine Kaufmann committed
390
      perror("BaseOpPassIntro: ibv_reg_mr shm failed");
391
392
393
394
395
396
397
      return 1;
    }
  } else {
    /* on the network side we need to make sure we have received the device
       intro from our RDMA peer, so we can include the queue position. */
    if (!peer->intro_valid_remote) {
      fprintf(stderr,
Antoine Kaufmann's avatar
Antoine Kaufmann committed
398
              "BaseOpPassIntro: skipping because remote intro not received\n");
399
400
401
      return 0;
    }

402
    peer->shm_opaque = mr_shm;
403
404
405
406
407
408
409
410
  }

  struct NetRdmaMsg *msg = RdmaMsgAlloc();
  if (!msg)
    return 1;

  msg->id = peer - peers;
  msg->base_addr = (uintptr_t) peer->shm_base;
411
412
  struct ibv_mr *mr = peer->shm_opaque;
  msg->rkey = mr->rkey;
Antoine Kaufmann's avatar
Antoine Kaufmann committed
413
414
415
416
417
418
  msg->msg_type = kMsgIntro;
  msg->queue_off = peer->cleanup_offset;
  msg->intro.payload_len = peer->intro_local_len;
  if (peer->intro_local_len > sizeof(msg->intro.data)) {
    fprintf(stderr, "BaseOpPassIntro: intro longer than buffer\n");
    abort();
419
  }
Antoine Kaufmann's avatar
Antoine Kaufmann committed
420
  memcpy(msg->intro.data, peer->intro_local, peer->intro_local_len);
421
422
423
424
425
426

  struct ibv_sge sge;
  sge.addr = (uintptr_t) msg;
  sge.length = sizeof(*msg);
  sge.lkey = mr_msgs->lkey;

427
  struct ibv_send_wr send_wr = { };
428
429
430
431
432
  send_wr.wr_id = msg - msgs;
  send_wr.opcode = IBV_WR_SEND;
  send_wr.send_flags = IBV_SEND_SIGNALED;
  send_wr.sg_list = &sge;
  send_wr.num_sge = 1;
433

434
  struct ibv_send_wr *bad_send_wr;
435
  if (ibv_post_send(qp, &send_wr, &bad_send_wr)) {
Antoine Kaufmann's avatar
Antoine Kaufmann committed
436
    perror("BaseOpPassIntro: ibv_post_send failed");
437
438
439
    return 1;
  }

440
#ifdef RDMA_DEBUG
Antoine Kaufmann's avatar
Antoine Kaufmann committed
441
  fprintf(stderr, "BaseOpPassIntro: ibv_post_send done\n");
442
#endif
443
444
445
  return 0;
}

Antoine Kaufmann's avatar
Antoine Kaufmann committed
446
int BaseOpPassEntries(struct Peer *peer, uint32_t pos, uint32_t n) {
447
#ifdef RDMA_DEBUG
Antoine Kaufmann's avatar
Antoine Kaufmann committed
448
  fprintf(stderr, "BaseOpPassEntries(%s,%u)\n", peer->sock_path,
449
          pos);
450
  fprintf(stderr, "  remote_base=%lx local_base=%p\n", peer->remote_base,
451
          peer->local_base);
452
453
#endif

454
455
456
457
458
  bool triggerSig = ++last_signaled > SIG_THRESHOLD;
  if (triggerSig)
    last_signaled = 0;

  while (1) {
459
    uint64_t abs_pos = pos * peer->local_elen;
460
    struct ibv_sge sge;
461
    sge.addr = (uintptr_t) (peer->local_base + abs_pos);
462
    sge.length = peer->local_elen * n;
463
464
    struct ibv_mr *mr = peer->shm_opaque;
    sge.lkey = mr->lkey;
465
466
467
468
469
470

    struct ibv_send_wr send_wr = { };
    send_wr.wr_id = -1ULL;
    send_wr.opcode = IBV_WR_RDMA_WRITE;
    if (triggerSig)
      send_wr.send_flags = IBV_SEND_SIGNALED;
471
    send_wr.wr.rdma.remote_addr = peer->remote_base + abs_pos;
472
473
474
475
476
    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;
477
    int ret = ibv_post_send(qp, &send_wr, &bad_send_wr);
478
479
480
    if (ret == 0) {
      break;
    } else if (ret != ENOMEM) {
Antoine Kaufmann's avatar
Antoine Kaufmann committed
481
      fprintf(stderr, "BaseOpPassEntries: ibv_post_send failed %d (%s)\n", ret,
482
483
484
              strerror(ret));
      return 1;
    }
485
  }
486
  return 0;
487
488
}

Antoine Kaufmann's avatar
Antoine Kaufmann committed
489
int BaseOpPassReport() {
490
  if (peer_num > MAX_PEERS) {
Antoine Kaufmann's avatar
Antoine Kaufmann committed
491
    fprintf(stderr, "BaseOpPassReport: peer_num (%zu) larger than max (%u)\n",
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
            peer_num, MAX_PEERS);
    abort();
  }

  struct NetRdmaMsg *msg = RdmaMsgAlloc();
  if (!msg)
    return 1;

  msg->msg_type = kMsgReport;
  for (size_t i = 0; i < MAX_PEERS; i++) {
    if (i >= peer_num) {
      msg->report.valid[i] = false;
      continue;
    }

    struct Peer *peer = &peers[i];
    msg->report.valid[i] = peer->ready;
    if (!peer->ready)
      continue;

    peer->cleanup_pos_reported = peer->cleanup_pos_next;
    msg->report.clean_pos[i] = peer->cleanup_pos_reported;
    peer->local_pos_reported = peer->local_pos;
    msg->report.written_pos[i] = peer->local_pos_reported;
  }

518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
  last_signaled = 0;

  while (1) {
    struct ibv_sge sge;
    sge.addr = (uintptr_t) msg;
    sge.length = sizeof(*msg);
    sge.lkey = mr_msgs->lkey;

    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;
    send_wr.sg_list = &sge;
    send_wr.num_sge = 1;

    struct ibv_send_wr *bad_send_wr;
534
    int ret = ibv_post_send(qp, &send_wr, &bad_send_wr);
535
536
537
    if (ret == 0) {
      break;
    } else if (ret != ENOMEM) {
538
      fprintf(stderr, "NetOpPassReport: ibv_post_send failed %u (%s)", ret,
539
540
541
              strerror(ret));
      return 1;
    }
542
543
544
  }

  return 0;
Antoine Kaufmann's avatar
Antoine Kaufmann committed
545
}