c11/ibperfser.c at main · wuxb45/c11 · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
/*
 * Copyright (c) 2016--2021  Wu, Xingbo <wuxb45@gmail.com>
 *
 * All rights reserved. No warranty, explicit or implicit, provided.
 */
#define _GNU_SOURCE

#include "lib.h"
#include "ib.h"

struct worker_info {
  struct ib_port * ibport;
  u64 depth;
  u64 reqsz;
};

  static void *
perf_worker(void * const ptr)
{
  struct server_wi * const si = (typeof(si))ptr;
  struct worker_info * const wi = (typeof(wi))server_worker_priv(si);
  const u64 depth = wi->depth;
  const u64 reqsz = wi->reqsz;

  struct ib_socket * const s = ib_socket_create(wi->ibport, wi->depth);
  if (!s)
    server_worker_exit(si);
  const bool rc = ib_socket_connect_fd(s, server_worker_fd(si));
  if (!rc) {
    ib_socket_destroy(s);
    server_worker_exit(si);
  }

  u64 memsize = 0;
  u8 * const mem = pages_alloc_best(depth * reqsz, true, &memsize);
  if (!mem) {
    ib_socket_destroy(s);
    server_worker_exit(si);
  }
  struct ibv_mr * const mr = ib_mr_helper(wi->ibport, mem, memsize);
  if (!mr) {
    pages_unmap(mem, memsize);
    ib_socket_destroy(s);
    server_worker_exit(si);
  }
  ib_mr_send_rptr_fd(server_worker_fd(si), mr);

  struct ibv_sge sg;
  struct ibv_send_wr swr;
  bool wait_more = true;

  for (u64 i = 0; i < depth; i++) {
    ib_sge_fill_off(&sg, mr, i*reqsz, reqsz);
    if (!ib_socket_post_recv_sge(s, i, &sg))
      debug_die();
  }

  do {
    int n = ib_socket_poll(s);
    if (n < 0) {
      perror("ib_socket_poll");
      break;
    } else if (n == 0) {
      cpu_pause();
    }
    for (int i = 0; i < n; i++) {
      struct ibv_wc * const wc = &(s->wcs[i]);
      if (wc->status != IBV_WC_SUCCESS) {
        wait_more = false;
        continue;
      }
      switch (wc->opcode) {
      case IBV_WC_RECV:
        if (wc->imm_data == UINT32_MAX) {
          wait_more = false;
        } else {
          const u64 id = wc->wr_id;
          ib_sge_fill_off(&sg, mr, id*reqsz, wc->byte_len);
          ib_wr_fill_send(&swr, id, &sg);
          if (!ib_socket_post_send(s, &swr, true))
            debug_die();
          ib_sge_fill_off(&sg, mr, id*reqsz, reqsz);
          if (!ib_socket_post_recv_sge(s, id, &sg)) // use the same buffer to receive
            debug_die();
        }
        break;
      default:
        break;
      }
    }
  } while (wait_more);

  ibv_dereg_mr(mr);
  pages_unmap(mem, memsize);
  ib_socket_destroy(s);
  server_worker_exit(si);
  return NULL;
}

  int
main(int argc, char ** argv)
{
  if (argc < 6) {
    fprintf(stderr, "Usage: <ib-port> <host> <port> <depth> <reqsz>\n");
    exit(0);
  }

  struct worker_info wi = {};
  wi.ibport = ib_port_create(atoi(argv[1]));
  if (wi.ibport == NULL) {
    fprintf(stderr, "ib_port_create() failed\n");
    exit(0);
  }

  const char * const host = argv[2];
  const char * const port = argv[3];
  wi.depth = bits_p2_up_u64(a2u64(argv[4]));
  wi.reqsz = bits_p2_up_u64(a2u64(argv[5]));

  struct server * const ser = server_create(host, port, perf_worker, &wi);
  if (ser == NULL) {
    fprintf(stderr, "server_create() failed\n");
    exit(0);
  }

  server_wait_destroy(ser);
  return 0;
}