2 * librpma_fio: librpma_apm and librpma_gpspm engines' common part.
4 * Copyright 2021, Intel Corporation
6 * This program is free software; you can redistribute it and/or
7 * modify it under the terms of the GNU General Public License,
8 * version 2 as published by the Free Software Foundation..
10 * This program is distributed in the hope that it will be useful,
11 * but WITHOUT ANY WARRANTY; without even the implied warranty of
12 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
13 * GNU General Public License for more details.
16 #include "librpma_fio.h"
20 struct fio_option librpma_fio_options[] = {
23 .lname = "rpma_server_ip",
24 .type = FIO_OPT_STR_STORE,
25 .off1 = offsetof(struct librpma_fio_options_values, server_ip),
26 .help = "IP address the server is listening on",
28 .category = FIO_OPT_C_ENGINE,
29 .group = FIO_OPT_G_LIBRPMA,
33 .lname = "rpma_server port",
34 .type = FIO_OPT_STR_STORE,
35 .off1 = offsetof(struct librpma_fio_options_values, port),
36 .help = "port the server is listening on",
38 .category = FIO_OPT_C_ENGINE,
39 .group = FIO_OPT_G_LIBRPMA,
42 .name = "direct_write_to_pmem",
43 .lname = "Direct Write to PMem (via RDMA) from the remote host is possible",
45 .off1 = offsetof(struct librpma_fio_options_values,
46 direct_write_to_pmem),
47 .help = "Set to true ONLY when Direct Write to PMem from the remote host is possible (https://pmem.io/rpma/documentation/basic-direct-write-to-pmem.html)",
49 .category = FIO_OPT_C_ENGINE,
50 .group = FIO_OPT_G_LIBRPMA,
53 .name = "busy_wait_polling",
54 .lname = "Set to 0 to wait for completion instead of busy-wait polling completion.",
56 .off1 = offsetof(struct librpma_fio_options_values,
58 .help = "Set to false if you want to reduce CPU usage",
60 .category = FIO_OPT_C_ENGINE,
61 .group = FIO_OPT_G_LIBRPMA,
68 int librpma_fio_td_port(const char *port_base_str, struct thread_data *td,
71 unsigned long int port_ul = strtoul(port_base_str, NULL, 10);
72 unsigned int port_new;
76 if (port_ul == ULONG_MAX) {
77 td_verror(td, errno, "strtoul");
80 port_ul += td->thread_number - 1;
81 if (port_ul >= UINT_MAX) {
82 log_err("[%u] port number (%lu) bigger than UINT_MAX\n",
83 td->thread_number, port_ul);
88 snprintf(port_out, LIBRPMA_FIO_PORT_STR_LEN_MAX - 1, "%u", port_new);
93 char *librpma_fio_allocate_dram(struct thread_data *td, size_t size,
94 struct librpma_fio_mem *mem)
99 if ((ret = posix_memalign((void **)&mem_ptr, page_size, size))) {
100 log_err("fio: posix_memalign() failed\n");
101 td_verror(td, ret, "posix_memalign");
105 mem->mem_ptr = mem_ptr;
111 char *librpma_fio_allocate_pmem(struct thread_data *td, struct fio_file *f,
112 size_t size, struct librpma_fio_mem *mem)
114 size_t size_mmap = 0;
115 char *mem_ptr = NULL;
119 if (size % page_size) {
120 log_err("fio: size (%zu) is not aligned to page size (%zu)\n",
125 if (f->filetype == FIO_TYPE_CHAR) {
126 /* Each thread uses a separate offset within DeviceDAX. */
127 ws_offset = (td->thread_number - 1) * size;
129 /* Each thread uses a separate FileSystemDAX file. No offset is needed. */
134 log_err("fio: filename is not set\n");
139 mem_ptr = pmem_map_file(f->file_name, 0 /* len */, 0 /* flags */,
140 0 /* mode */, &size_mmap, &is_pmem);
141 if (mem_ptr == NULL) {
142 log_err("fio: pmem_map_file(%s) failed\n", f->file_name);
143 /* pmem_map_file() sets errno on failure */
144 td_verror(td, errno, "pmem_map_file");
148 /* pmem is expected */
150 log_err("fio: %s is not located in persistent memory\n",
155 /* check size of allocated persistent memory */
156 if (size_mmap < ws_offset + size) {
158 "fio: %s is too small to handle so many threads (%zu < %zu)\n",
159 f->file_name, size_mmap, ws_offset + size);
163 log_info("fio: size of memory mapped from the file %s: %zu\n",
164 f->file_name, size_mmap);
166 mem->mem_ptr = mem_ptr;
167 mem->size_mmap = size_mmap;
169 return mem_ptr + ws_offset;
172 (void) pmem_unmap(mem_ptr, size_mmap);
176 void librpma_fio_free(struct librpma_fio_mem *mem)
179 (void) pmem_unmap(mem->mem_ptr, mem->size_mmap);
184 #define LIBRPMA_FIO_RETRY_MAX_NO 10
185 #define LIBRPMA_FIO_RETRY_DELAY_S 5
187 int librpma_fio_client_init(struct thread_data *td,
188 struct rpma_conn_cfg *cfg)
190 struct librpma_fio_client_data *ccd;
191 struct librpma_fio_options_values *o = td->eo;
192 struct ibv_context *dev = NULL;
193 char port_td[LIBRPMA_FIO_PORT_STR_LEN_MAX];
194 struct rpma_conn_req *req = NULL;
195 enum rpma_conn_event event;
196 struct rpma_conn_private_data pdata;
197 enum rpma_log_level log_level_aux = RPMA_LOG_LEVEL_WARNING;
198 int remote_flush_type;
202 /* --debug=net sets RPMA_LOG_THRESHOLD_AUX to RPMA_LOG_LEVEL_INFO */
204 if ((1UL << FD_NET) & fio_debug)
205 log_level_aux = RPMA_LOG_LEVEL_INFO;
208 /* configure logging thresholds to see more details */
209 rpma_log_set_threshold(RPMA_LOG_THRESHOLD, RPMA_LOG_LEVEL_INFO);
210 rpma_log_set_threshold(RPMA_LOG_THRESHOLD_AUX, log_level_aux);
212 /* obtain an IBV context for a remote IP address */
213 if ((ret = rpma_utils_get_ibv_context(o->server_ip,
214 RPMA_UTIL_IBV_CONTEXT_REMOTE, &dev))) {
215 librpma_td_verror(td, ret, "rpma_utils_get_ibv_context");
219 /* allocate client's data */
220 ccd = calloc(1, sizeof(*ccd));
222 td_verror(td, errno, "calloc");
226 /* allocate all in-memory queues */
227 ccd->io_us_queued = calloc(td->o.iodepth, sizeof(*ccd->io_us_queued));
228 if (ccd->io_us_queued == NULL) {
229 td_verror(td, errno, "calloc");
233 ccd->io_us_flight = calloc(td->o.iodepth, sizeof(*ccd->io_us_flight));
234 if (ccd->io_us_flight == NULL) {
235 td_verror(td, errno, "calloc");
236 goto err_free_io_u_queues;
239 ccd->io_us_completed = calloc(td->o.iodepth,
240 sizeof(*ccd->io_us_completed));
241 if (ccd->io_us_completed == NULL) {
242 td_verror(td, errno, "calloc");
243 goto err_free_io_u_queues;
246 /* create a new peer object */
247 if ((ret = rpma_peer_new(dev, &ccd->peer))) {
248 librpma_td_verror(td, ret, "rpma_peer_new");
249 goto err_free_io_u_queues;
252 /* create a connection request */
253 if (librpma_fio_td_port(o->port, td, port_td))
254 goto err_peer_delete;
256 for (retry = 0; retry < LIBRPMA_FIO_RETRY_MAX_NO; retry++) {
257 if ((ret = rpma_conn_req_new(ccd->peer, o->server_ip, port_td,
259 librpma_td_verror(td, ret, "rpma_conn_req_new");
260 goto err_peer_delete;
264 * Connect the connection request
265 * and obtain the connection object.
267 if ((ret = rpma_conn_req_connect(&req, NULL, &ccd->conn))) {
268 librpma_td_verror(td, ret, "rpma_conn_req_connect");
272 /* wait for the connection to establish */
273 if ((ret = rpma_conn_next_event(ccd->conn, &event))) {
274 librpma_td_verror(td, ret, "rpma_conn_next_event");
275 goto err_conn_delete;
276 } else if (event == RPMA_CONN_ESTABLISHED) {
278 } else if (event == RPMA_CONN_REJECTED) {
279 (void) rpma_conn_disconnect(ccd->conn);
280 (void) rpma_conn_delete(&ccd->conn);
281 if (retry < LIBRPMA_FIO_RETRY_MAX_NO - 1) {
282 log_err("Thread [%d]: Retrying (#%i) ...\n",
283 td->thread_number, retry + 1);
284 sleep(LIBRPMA_FIO_RETRY_DELAY_S);
287 "Thread [%d]: The maximum number of retries exceeded. Closing.\n",
292 "rpma_conn_next_event returned an unexptected event: (%s != RPMA_CONN_ESTABLISHED)\n",
293 rpma_utils_conn_event_2str(event));
294 goto err_conn_delete;
299 log_err("Thread [%d]: Connected after retry #%i\n",
300 td->thread_number, retry);
302 if (ccd->conn == NULL)
303 goto err_peer_delete;
305 /* get the connection's main CQ */
306 if ((ret = rpma_conn_get_cq(ccd->conn, &ccd->cq))) {
307 librpma_td_verror(td, ret, "rpma_conn_get_cq");
308 goto err_conn_delete;
311 /* get the connection's private data sent from the server */
312 if ((ret = rpma_conn_get_private_data(ccd->conn, &pdata))) {
313 librpma_td_verror(td, ret, "rpma_conn_get_private_data");
314 goto err_conn_delete;
317 /* get the server's workspace representation */
320 /* create the server's memory representation */
321 if ((ret = rpma_mr_remote_from_descriptor(&ccd->ws->descriptor[0],
322 ccd->ws->mr_desc_size, &ccd->server_mr))) {
323 librpma_td_verror(td, ret, "rpma_mr_remote_from_descriptor");
324 goto err_conn_delete;
327 /* get the total size of the shared server memory */
328 if ((ret = rpma_mr_remote_get_size(ccd->server_mr, &ccd->ws_size))) {
329 librpma_td_verror(td, ret, "rpma_mr_remote_get_size");
330 goto err_conn_delete;
333 /* get flush type of the remote node */
334 if ((ret = rpma_mr_remote_get_flush_type(ccd->server_mr,
335 &remote_flush_type))) {
336 librpma_td_verror(td, ret, "rpma_mr_remote_get_flush_type");
337 goto err_conn_delete;
340 ccd->server_mr_flush_type =
341 (remote_flush_type & RPMA_MR_USAGE_FLUSH_TYPE_PERSISTENT) ?
342 RPMA_FLUSH_TYPE_PERSISTENT : RPMA_FLUSH_TYPE_VISIBILITY;
345 * Assure an io_us buffer allocation is page-size-aligned which is required
346 * to register for RDMA. User-provided value is intentionally ignored.
348 td->o.mem_align = page_size;
350 td->io_ops_data = ccd;
355 (void) rpma_conn_disconnect(ccd->conn);
356 (void) rpma_conn_delete(&ccd->conn);
359 (void) rpma_conn_req_delete(&req);
362 (void) rpma_peer_delete(&ccd->peer);
364 err_free_io_u_queues:
365 free(ccd->io_us_queued);
366 free(ccd->io_us_flight);
367 free(ccd->io_us_completed);
375 void librpma_fio_client_cleanup(struct thread_data *td)
377 struct librpma_fio_client_data *ccd = td->io_ops_data;
378 enum rpma_conn_event ev;
384 /* delete the iou's memory registration */
385 if ((ret = rpma_mr_dereg(&ccd->orig_mr)))
386 librpma_td_verror(td, ret, "rpma_mr_dereg");
387 /* delete the iou's memory registration */
388 if ((ret = rpma_mr_remote_delete(&ccd->server_mr)))
389 librpma_td_verror(td, ret, "rpma_mr_remote_delete");
390 /* initiate disconnection */
391 if ((ret = rpma_conn_disconnect(ccd->conn)))
392 librpma_td_verror(td, ret, "rpma_conn_disconnect");
393 /* wait for disconnection to end up */
394 if ((ret = rpma_conn_next_event(ccd->conn, &ev))) {
395 librpma_td_verror(td, ret, "rpma_conn_next_event");
396 } else if (ev != RPMA_CONN_CLOSED) {
398 "client_cleanup received an unexpected event (%s != RPMA_CONN_CLOSED)\n",
399 rpma_utils_conn_event_2str(ev));
401 /* delete the connection */
402 if ((ret = rpma_conn_delete(&ccd->conn)))
403 librpma_td_verror(td, ret, "rpma_conn_delete");
404 /* delete the peer */
405 if ((ret = rpma_peer_delete(&ccd->peer)))
406 librpma_td_verror(td, ret, "rpma_peer_delete");
407 /* free the software queues */
408 free(ccd->io_us_queued);
409 free(ccd->io_us_flight);
410 free(ccd->io_us_completed);
412 td->io_ops_data = NULL; /* zero ccd */
415 int librpma_fio_file_nop(struct thread_data *td, struct fio_file *f)
421 int librpma_fio_client_post_init(struct thread_data *td)
423 struct librpma_fio_client_data *ccd = td->io_ops_data;
428 * td->orig_buffer is not aligned. The engine requires aligned io_us
429 * so FIO aligns up the address using the formula below.
431 ccd->orig_buffer_aligned = PTR_ALIGN(td->orig_buffer, page_mask) +
435 * td->orig_buffer_size beside the space really consumed by io_us
436 * has paddings which can be omitted for the memory registration.
438 io_us_size = (unsigned long long)td_max_bs(td) *
439 (unsigned long long)td->o.iodepth;
441 if ((ret = rpma_mr_reg(ccd->peer, ccd->orig_buffer_aligned, io_us_size,
442 RPMA_MR_USAGE_READ_DST | RPMA_MR_USAGE_READ_SRC |
443 RPMA_MR_USAGE_WRITE_DST | RPMA_MR_USAGE_WRITE_SRC |
444 RPMA_MR_USAGE_FLUSH_TYPE_PERSISTENT, &ccd->orig_mr)))
445 librpma_td_verror(td, ret, "rpma_mr_reg");
449 int librpma_fio_client_get_file_size(struct thread_data *td,
452 struct librpma_fio_client_data *ccd = td->io_ops_data;
454 f->real_file_size = ccd->ws_size;
455 fio_file_set_size_known(f);
460 static enum fio_q_status client_queue_sync(struct thread_data *td,
463 struct librpma_fio_client_data *ccd = td->io_ops_data;
469 if (io_u->ddir == DDIR_READ) {
470 /* post an RDMA read operation */
471 if (librpma_fio_client_io_read(td, io_u,
472 RPMA_F_COMPLETION_ALWAYS))
474 } else if (io_u->ddir == DDIR_WRITE) {
475 /* post an RDMA write operation */
476 if (librpma_fio_client_io_write(td, io_u))
478 if (ccd->flush(td, io_u, io_u, io_u->xfer_buflen))
481 log_err("unsupported IO mode: %s\n", io_ddir_name(io_u->ddir));
486 /* get a completion */
487 ret = rpma_cq_get_wc(ccd->cq, 1, &wc, NULL);
488 if (ret == RPMA_E_NO_COMPLETION) {
489 /* lack of completion is not an error */
491 } else if (ret != 0) {
492 /* an error occurred */
493 librpma_td_verror(td, ret, "rpma_cq_get_wc");
497 /* if io_us has completed with an error */
498 if (wc.status != IBV_WC_SUCCESS)
501 if (wc.opcode == IBV_WC_SEND)
502 ++ccd->op_send_completed;
504 if (wc.opcode == IBV_WC_RECV)
505 ++ccd->op_recv_completed;
511 if (ccd->get_io_u_index(&wc, &io_u_index) != 1)
514 if (io_u->index != io_u_index) {
516 "no matching io_u for received completion found (io_u_index=%u)\n",
521 /* make sure all SENDs are completed before exit - clean up SQ */
522 if (librpma_fio_client_io_complete_all_sends(td))
525 return FIO_Q_COMPLETED;
529 return FIO_Q_COMPLETED;
532 enum fio_q_status librpma_fio_client_queue(struct thread_data *td,
535 struct librpma_fio_client_data *ccd = td->io_ops_data;
537 if (ccd->io_u_queued_nr == (int)td->o.iodepth)
541 return client_queue_sync(td, io_u);
543 /* io_u -> queued[] */
544 ccd->io_us_queued[ccd->io_u_queued_nr] = io_u;
545 ccd->io_u_queued_nr++;
550 int librpma_fio_client_commit(struct thread_data *td)
552 struct librpma_fio_client_data *ccd = td->io_ops_data;
553 int flags = RPMA_F_COMPLETION_ON_ERROR;
557 struct io_u *flush_first_io_u = NULL;
558 unsigned long long int flush_len = 0;
560 if (!ccd->io_us_queued)
563 /* execute all io_us from queued[] */
564 for (i = 0; i < ccd->io_u_queued_nr; i++) {
565 struct io_u *io_u = ccd->io_us_queued[i];
567 if (io_u->ddir == DDIR_READ) {
568 if (i + 1 == ccd->io_u_queued_nr ||
569 ccd->io_us_queued[i + 1]->ddir == DDIR_WRITE)
570 flags = RPMA_F_COMPLETION_ALWAYS;
571 /* post an RDMA read operation */
572 if (librpma_fio_client_io_read(td, io_u, flags))
574 } else if (io_u->ddir == DDIR_WRITE) {
575 /* post an RDMA write operation */
576 if (librpma_fio_client_io_write(td, io_u))
579 /* cache the first io_u in the sequence */
580 if (flush_first_io_u == NULL)
581 flush_first_io_u = io_u;
584 * the flush length is the sum of all io_u's creating
587 flush_len += io_u->xfer_buflen;
590 * if io_u's are random the rpma_flush is required
591 * after each one of them
593 if (!td_random(td)) {
595 * When the io_u's are sequential and
596 * the current io_u is not the last one and
597 * the next one is also a write operation
598 * the flush can be postponed by one io_u and
599 * cover all of them which build a continuous
602 if ((i + 1 < ccd->io_u_queued_nr) &&
603 (ccd->io_us_queued[i + 1]->ddir == DDIR_WRITE))
607 /* flush all writes which build a continuous sequence */
608 if (ccd->flush(td, flush_first_io_u, io_u, flush_len))
612 * reset the flush parameters in preparation for
615 flush_first_io_u = NULL;
618 log_err("unsupported IO mode: %s\n",
619 io_ddir_name(io_u->ddir));
624 if ((fill_time = fio_fill_issue_time(td))) {
625 fio_gettime(&now, NULL);
628 * only used for iolog
630 if (td->o.read_iolog_file)
631 memcpy(&td->last_issue, &now, sizeof(now));
634 /* move executed io_us from queued[] to flight[] */
635 for (i = 0; i < ccd->io_u_queued_nr; i++) {
636 struct io_u *io_u = ccd->io_us_queued[i];
638 /* FIO does not do this if the engine is asynchronous */
640 memcpy(&io_u->issue_time, &now, sizeof(now));
642 /* move executed io_us from queued[] to flight[] */
643 ccd->io_us_flight[ccd->io_u_flight_nr] = io_u;
644 ccd->io_u_flight_nr++;
648 * If an engine has the commit hook
649 * it has to call io_u_queued() itself.
651 io_u_queued(td, io_u);
654 /* FIO does not do this if an engine has the commit hook. */
655 io_u_mark_submit(td, ccd->io_u_queued_nr);
656 ccd->io_u_queued_nr = 0;
663 * - > 0 - a number of completed io_us
664 * - 0 - when no complicitions received
665 * - (-1) - when an error occurred
667 static int client_getevent_process(struct thread_data *td)
669 struct librpma_fio_client_data *ccd = td->io_ops_data;
671 /* io_u->index of completed io_u (wc.wr_id) */
672 unsigned int io_u_index;
673 /* # of completed io_us */
680 /* get a completion */
681 if ((ret = rpma_cq_get_wc(ccd->cq, 1, &wc, NULL))) {
682 /* lack of completion is not an error */
683 if (ret == RPMA_E_NO_COMPLETION) {
684 /* lack of completion is not an error */
688 /* an error occurred */
689 librpma_td_verror(td, ret, "rpma_cq_get_wc");
693 /* if io_us has completed with an error */
694 if (wc.status != IBV_WC_SUCCESS) {
695 td->error = wc.status;
699 if (wc.opcode == IBV_WC_SEND)
700 ++ccd->op_send_completed;
701 else if (wc.opcode == IBV_WC_RECV)
702 ++ccd->op_recv_completed;
704 if ((ret = ccd->get_io_u_index(&wc, &io_u_index)) != 1)
707 /* look for an io_u being completed */
708 for (i = 0; i < ccd->io_u_flight_nr; ++i) {
709 if (ccd->io_us_flight[i]->index == io_u_index) {
715 /* if no matching io_u has been found */
718 "no matching io_u for received completion found (io_u_index=%u)\n",
723 /* move completed io_us to the completed in-memory queue */
724 for (i = 0; i < cmpl_num; ++i) {
725 /* get and prepare io_u */
726 io_u = ccd->io_us_flight[i];
728 /* append to the queue */
729 ccd->io_us_completed[ccd->io_u_completed_nr] = io_u;
730 ccd->io_u_completed_nr++;
733 /* remove completed io_us from the flight queue */
734 for (i = cmpl_num; i < ccd->io_u_flight_nr; ++i)
735 ccd->io_us_flight[i - cmpl_num] = ccd->io_us_flight[i];
736 ccd->io_u_flight_nr -= cmpl_num;
741 int librpma_fio_client_getevents(struct thread_data *td, unsigned int min,
742 unsigned int max, const struct timespec *t)
744 struct librpma_fio_client_data *ccd = td->io_ops_data;
745 /* total # of completed io_us */
746 int cmpl_num_total = 0;
747 /* # of completed io_us from a single event */
751 cmpl_num = client_getevent_process(td);
753 /* new completions collected */
754 cmpl_num_total += cmpl_num;
755 } else if (cmpl_num == 0) {
757 * It is required to make sure that CQEs for SENDs
758 * will flow at least at the same pace as CQEs for RECVs.
760 if (cmpl_num_total >= min &&
761 ccd->op_send_completed >= ccd->op_recv_completed)
765 * To reduce CPU consumption one can use
766 * the rpma_cq_wait() function.
767 * Note this greatly increase the latency
768 * and make the results less stable.
769 * The bandwidth stays more or less the same.
772 /* an error occurred */
777 * The expected max can be exceeded if CQEs for RECVs will come up
778 * faster than CQEs for SENDs. But it is required to make sure CQEs for
779 * SENDs will flow at least at the same pace as CQEs for RECVs.
781 } while (cmpl_num_total < max ||
782 ccd->op_send_completed < ccd->op_recv_completed);
785 * All posted SENDs are completed and RECVs for them (responses) are
786 * completed. This is the initial situation so the counters are reset.
788 if (ccd->op_send_posted == ccd->op_send_completed &&
789 ccd->op_send_completed == ccd->op_recv_completed) {
790 ccd->op_send_posted = 0;
791 ccd->op_send_completed = 0;
792 ccd->op_recv_completed = 0;
795 return cmpl_num_total;
798 struct io_u *librpma_fio_client_event(struct thread_data *td, int event)
800 struct librpma_fio_client_data *ccd = td->io_ops_data;
804 /* get the first io_u from the queue */
805 io_u = ccd->io_us_completed[0];
807 /* remove the first io_u from the queue */
808 for (i = 1; i < ccd->io_u_completed_nr; ++i)
809 ccd->io_us_completed[i - 1] = ccd->io_us_completed[i];
810 ccd->io_u_completed_nr--;
812 dprint_io_u(io_u, "client_event");
817 char *librpma_fio_client_errdetails(struct io_u *io_u)
819 /* get the string representation of an error */
820 enum ibv_wc_status status = io_u->error;
821 const char *status_str = ibv_wc_status_str(status);
823 char *details = strdup(status_str);
824 if (details == NULL) {
825 fprintf(stderr, "Error: %s\n", status_str);
826 fprintf(stderr, "Fatal error: out of memory. Aborting.\n");
830 /* FIO frees the returned string when it becomes obsolete */
834 int librpma_fio_server_init(struct thread_data *td)
836 struct librpma_fio_options_values *o = td->eo;
837 struct librpma_fio_server_data *csd;
838 struct ibv_context *dev = NULL;
839 enum rpma_log_level log_level_aux = RPMA_LOG_LEVEL_WARNING;
842 /* --debug=net sets RPMA_LOG_THRESHOLD_AUX to RPMA_LOG_LEVEL_INFO */
844 if ((1UL << FD_NET) & fio_debug)
845 log_level_aux = RPMA_LOG_LEVEL_INFO;
848 /* configure logging thresholds to see more details */
849 rpma_log_set_threshold(RPMA_LOG_THRESHOLD, RPMA_LOG_LEVEL_INFO);
850 rpma_log_set_threshold(RPMA_LOG_THRESHOLD_AUX, log_level_aux);
853 /* obtain an IBV context for a remote IP address */
854 if ((ret = rpma_utils_get_ibv_context(o->server_ip,
855 RPMA_UTIL_IBV_CONTEXT_LOCAL, &dev))) {
856 librpma_td_verror(td, ret, "rpma_utils_get_ibv_context");
860 /* allocate server's data */
861 csd = calloc(1, sizeof(*csd));
863 td_verror(td, errno, "calloc");
867 /* create a new peer object */
868 if ((ret = rpma_peer_new(dev, &csd->peer))) {
869 librpma_td_verror(td, ret, "rpma_peer_new");
873 td->io_ops_data = csd;
883 void librpma_fio_server_cleanup(struct thread_data *td)
885 struct librpma_fio_server_data *csd = td->io_ops_data;
892 if ((ret = rpma_peer_delete(&csd->peer)))
893 librpma_td_verror(td, ret, "rpma_peer_delete");
898 int librpma_fio_server_open_file(struct thread_data *td, struct fio_file *f,
899 struct rpma_conn_cfg *cfg)
901 struct librpma_fio_server_data *csd = td->io_ops_data;
902 struct librpma_fio_options_values *o = td->eo;
903 enum rpma_conn_event conn_event = RPMA_CONN_UNDEFINED;
904 struct librpma_fio_workspace ws = {0};
905 struct rpma_conn_private_data pdata;
906 uint32_t max_msg_num;
907 struct rpma_conn_req *conn_req;
908 struct rpma_conn *conn;
909 struct rpma_mr_local *mr;
910 char port_td[LIBRPMA_FIO_PORT_STR_LEN_MAX];
912 size_t mem_size = td->o.size;
920 log_err("fio: filename is not set\n");
924 /* start a listening endpoint at addr:port */
925 if (librpma_fio_td_port(o->port, td, port_td))
928 if ((ret = rpma_ep_listen(csd->peer, o->server_ip, port_td, &ep))) {
929 librpma_td_verror(td, ret, "rpma_ep_listen");
933 is_dram = !strcmp(f->file_name, "malloc");
935 /* allocation from DRAM using posix_memalign() */
936 ws_ptr = librpma_fio_allocate_dram(td, mem_size, &csd->mem);
937 usage_mem_type = RPMA_MR_USAGE_FLUSH_TYPE_VISIBILITY;
939 /* allocation from PMEM using pmem_map_file() */
940 ws_ptr = librpma_fio_allocate_pmem(td, f, mem_size, &csd->mem);
941 usage_mem_type = RPMA_MR_USAGE_FLUSH_TYPE_PERSISTENT;
945 goto err_ep_shutdown;
947 f->real_file_size = mem_size;
949 if ((ret = rpma_mr_reg(csd->peer, ws_ptr, mem_size,
950 RPMA_MR_USAGE_READ_DST | RPMA_MR_USAGE_READ_SRC |
951 RPMA_MR_USAGE_WRITE_DST | RPMA_MR_USAGE_WRITE_SRC |
952 usage_mem_type, &mr))) {
953 librpma_td_verror(td, ret, "rpma_mr_reg");
957 if (!is_dram && f->filetype == FIO_TYPE_FILE) {
958 ret = rpma_mr_advise(mr, 0, mem_size,
959 IBV_ADVISE_MR_ADVICE_PREFETCH_WRITE,
960 IBV_ADVISE_MR_FLAG_FLUSH);
962 librpma_td_verror(td, ret, "rpma_mr_advise");
963 /* an invalid argument is an error */
964 if (ret == RPMA_E_INVAL)
967 /* log_err used instead of log_info to avoid corruption of the JSON output */
968 log_err("Note: having rpma_mr_advise(3) failed because of RPMA_E_NOSUPP or RPMA_E_PROVIDER may come with a performance penalty, but it is not a blocker for running the benchmark.\n");
972 /* get size of the memory region's descriptor */
973 if ((ret = rpma_mr_get_descriptor_size(mr, &mr_desc_size))) {
974 librpma_td_verror(td, ret, "rpma_mr_get_descriptor_size");
978 /* verify size of the memory region's descriptor */
979 if (mr_desc_size > LIBRPMA_FIO_DESCRIPTOR_MAX_SIZE) {
981 "size of the memory region's descriptor is too big (max=%i)\n",
982 LIBRPMA_FIO_DESCRIPTOR_MAX_SIZE);
986 /* get the memory region's descriptor */
987 if ((ret = rpma_mr_get_descriptor(mr, &ws.descriptor[0]))) {
988 librpma_td_verror(td, ret, "rpma_mr_get_descriptor");
993 if ((ret = rpma_conn_cfg_get_rq_size(cfg, &max_msg_num))) {
994 librpma_td_verror(td, ret, "rpma_conn_cfg_get_rq_size");
998 /* verify whether iodepth fits into uint16_t */
999 if (max_msg_num > UINT16_MAX) {
1000 log_err("fio: iodepth too big (%u > %u)\n",
1001 max_msg_num, UINT16_MAX);
1005 ws.max_msg_num = max_msg_num;
1008 /* prepare a workspace description */
1009 ws.direct_write_to_pmem = o->direct_write_to_pmem;
1010 ws.mr_desc_size = mr_desc_size;
1012 pdata.len = sizeof(ws);
1014 /* receive an incoming connection request */
1015 if ((ret = rpma_ep_next_conn_req(ep, cfg, &conn_req))) {
1016 librpma_td_verror(td, ret, "rpma_ep_next_conn_req");
1020 if (csd->prepare_connection && csd->prepare_connection(td, conn_req))
1021 goto err_req_delete;
1023 /* accept the connection request and obtain the connection object */
1024 if ((ret = rpma_conn_req_connect(&conn_req, &pdata, &conn))) {
1025 librpma_td_verror(td, ret, "rpma_conn_req_connect");
1026 goto err_req_delete;
1029 /* wait for the connection to be established */
1030 if ((ret = rpma_conn_next_event(conn, &conn_event))) {
1031 librpma_td_verror(td, ret, "rpma_conn_next_event");
1032 goto err_conn_delete;
1033 } else if (conn_event != RPMA_CONN_ESTABLISHED) {
1034 log_err("rpma_conn_next_event returned an unexptected event\n");
1035 goto err_conn_delete;
1038 /* end-point is no longer needed */
1039 (void) rpma_ep_shutdown(&ep);
1042 csd->ws_ptr = ws_ptr;
1045 /* get the connection's main CQ */
1046 if ((ret = rpma_conn_get_cq(csd->conn, &csd->cq))) {
1047 librpma_td_verror(td, ret, "rpma_conn_get_cq");
1048 goto err_conn_delete;
1054 (void) rpma_conn_delete(&conn);
1057 (void) rpma_conn_req_delete(&conn_req);
1060 (void) rpma_mr_dereg(&mr);
1063 librpma_fio_free(&csd->mem);
1066 (void) rpma_ep_shutdown(&ep);
1071 int librpma_fio_server_close_file(struct thread_data *td, struct fio_file *f)
1073 struct librpma_fio_server_data *csd = td->io_ops_data;
1074 enum rpma_conn_event conn_event = RPMA_CONN_UNDEFINED;
1078 /* wait for the connection to be closed */
1079 ret = rpma_conn_next_event(csd->conn, &conn_event);
1080 if (!ret && conn_event != RPMA_CONN_CLOSED) {
1081 log_err("rpma_conn_next_event returned an unexptected event\n");
1085 if ((ret = rpma_conn_disconnect(csd->conn))) {
1086 librpma_td_verror(td, ret, "rpma_conn_disconnect");
1090 if ((ret = rpma_conn_delete(&csd->conn))) {
1091 librpma_td_verror(td, ret, "rpma_conn_delete");
1095 if ((ret = rpma_mr_dereg(&csd->ws_mr))) {
1096 librpma_td_verror(td, ret, "rpma_mr_dereg");
1100 librpma_fio_free(&csd->mem);