2 * Unix SMB/CIFS implementation.
3 * Wrap Infiniband calls.
5 * Copyright (C) Sven Oehme <oehmes@de.ibm.com> 2006
7 * Major code contributions by Peter Somogyi <psomogyi@gamax.hu>
9 * This program is free software; you can redistribute it and/or modify
10 * it under the terms of the GNU General Public License as published by
11 * the Free Software Foundation; either version 3 of the License, or
12 * (at your option) any later version.
14 * This program is distributed in the hope that it will be useful,
15 * but WITHOUT ANY WARRANTY; without even the implied warranty of
16 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
17 * GNU General Public License for more details.
19 * You should have received a copy of the GNU General Public License
20 * along with this program; if not, see <http://www.gnu.org/licenses/>.
24 #include "system/network.h"
30 #include "lib/util/dlinklist.h"
31 #include "lib/util/debug.h"
33 #include "common/logging.h"
35 #include <infiniband/kern-abi.h>
36 #include <rdma/rdma_cma_abi.h>
37 #include <rdma/rdma_cma.h>
39 #include "ibwrapper.h"
40 #include "ibwrapper_internal.h"
42 #define IBW_LASTERR_BUFSIZE 512
43 static char ibw_lasterr
[IBW_LASTERR_BUFSIZE
];
45 #define IBW_MAX_SEND_WR 256
46 #define IBW_MAX_RECV_WR 1024
47 #define IBW_RECV_BUFSIZE 256
48 #define IBW_RECV_THRESHOLD (1 * 1024 * 1024)
50 static void ibw_event_handler_verbs(struct tevent_context
*ev
,
51 struct tevent_fd
*fde
, uint16_t flags
, void *private_data
);
52 static int ibw_fill_cq(struct ibw_conn
*conn
);
53 static int ibw_wc_recv(struct ibw_conn
*conn
, struct ibv_wc
*wc
);
54 static int ibw_wc_send(struct ibw_conn
*conn
, struct ibv_wc
*wc
);
55 static int ibw_send_packet(struct ibw_conn
*conn
, void *buf
, struct ibw_wr
*p
, uint32_t len
);
57 static void *ibw_alloc_mr(struct ibw_ctx_priv
*pctx
, struct ibw_conn_priv
*pconn
,
58 uint32_t n
, struct ibv_mr
**ppmr
)
62 DEBUG(DEBUG_DEBUG
, ("ibw_alloc_mr(cmid=%p, n=%u)\n", pconn
->cm_id
, n
));
63 buf
= memalign(pctx
->pagesize
, n
);
65 sprintf(ibw_lasterr
, "couldn't allocate memory\n");
69 *ppmr
= ibv_reg_mr(pconn
->pd
, buf
, n
, IBV_ACCESS_LOCAL_WRITE
);
71 sprintf(ibw_lasterr
, "couldn't allocate mr\n");
79 static void ibw_free_mr(char **ppbuf
, struct ibv_mr
**ppmr
)
81 DEBUG(DEBUG_DEBUG
, ("ibw_free_mr(%p %p)\n", *ppbuf
, *ppmr
));
92 static int ibw_init_memory(struct ibw_conn
*conn
)
94 struct ibw_ctx_priv
*pctx
= talloc_get_type(conn
->ctx
->internal
, struct ibw_ctx_priv
);
95 struct ibw_conn_priv
*pconn
= talloc_get_type(conn
->internal
, struct ibw_conn_priv
);
96 struct ibw_opts
*opts
= &pctx
->opts
;
100 DEBUG(DEBUG_DEBUG
, ("ibw_init_memory(cmid: %p)\n", pconn
->cm_id
));
101 pconn
->buf_send
= ibw_alloc_mr(pctx
, pconn
,
102 opts
->max_send_wr
* opts
->recv_bufsize
, &pconn
->mr_send
);
103 if (!pconn
->buf_send
) {
104 sprintf(ibw_lasterr
, "couldn't allocate work send buf\n");
108 pconn
->buf_recv
= ibw_alloc_mr(pctx
, pconn
,
109 opts
->max_recv_wr
* opts
->recv_bufsize
, &pconn
->mr_recv
);
110 if (!pconn
->buf_recv
) {
111 sprintf(ibw_lasterr
, "couldn't allocate work recv buf\n");
115 pconn
->wr_index
= talloc_size(pconn
, opts
->max_send_wr
* sizeof(struct ibw_wr
*));
116 assert(pconn
->wr_index
!=NULL
);
118 for(i
=0; i
<opts
->max_send_wr
; i
++) {
119 p
= pconn
->wr_index
[i
] = talloc_zero(pconn
, struct ibw_wr
);
120 p
->buf
= pconn
->buf_send
+ (i
* opts
->recv_bufsize
);
123 DLIST_ADD(pconn
->wr_list_avail
, p
);
129 static int ibw_ctx_priv_destruct(struct ibw_ctx_priv
*pctx
)
131 DEBUG(DEBUG_DEBUG
, ("ibw_ctx_priv_destruct(%p)\n", pctx
));
134 * tevent_fd must be removed before the fd is closed
136 TALLOC_FREE(pctx
->cm_channel_event
);
139 if (pctx
->cm_channel
) {
140 rdma_destroy_event_channel(pctx
->cm_channel
);
141 pctx
->cm_channel
= NULL
;
144 rdma_destroy_id(pctx
->cm_id
);
151 static int ibw_ctx_destruct(struct ibw_ctx
*ctx
)
153 DEBUG(DEBUG_DEBUG
, ("ibw_ctx_destruct(%p)\n", ctx
));
157 static int ibw_conn_priv_destruct(struct ibw_conn_priv
*pconn
)
159 DEBUG(DEBUG_DEBUG
, ("ibw_conn_priv_destruct(%p, cmid: %p)\n",
160 pconn
, pconn
->cm_id
));
162 /* pconn->wr_index is freed by talloc */
163 /* pconn->wr_index[i] are freed by talloc */
166 * tevent_fd must be removed before the fd is closed
168 TALLOC_FREE(pconn
->verbs_channel_event
);
171 if (pconn
->cm_id
!=NULL
&& pconn
->cm_id
->qp
!=NULL
) {
172 rdma_destroy_qp(pconn
->cm_id
);
173 pconn
->cm_id
->qp
= NULL
;
176 if (pconn
->cq
!=NULL
) {
177 ibv_destroy_cq(pconn
->cq
);
181 if (pconn
->verbs_channel
!=NULL
) {
182 ibv_destroy_comp_channel(pconn
->verbs_channel
);
183 pconn
->verbs_channel
= NULL
;
186 /* free memory regions */
187 ibw_free_mr(&pconn
->buf_send
, &pconn
->mr_send
);
188 ibw_free_mr(&pconn
->buf_recv
, &pconn
->mr_recv
);
191 ibv_dealloc_pd(pconn
->pd
);
193 DEBUG(DEBUG_DEBUG
, ("pconn=%p pd deallocated\n", pconn
));
197 rdma_destroy_id(pconn
->cm_id
);
199 DEBUG(DEBUG_DEBUG
, ("pconn=%p cm_id destroyed\n", pconn
));
205 static int ibw_wr_destruct(struct ibw_wr
*wr
)
207 if (wr
->buf_large
!=NULL
)
208 ibw_free_mr(&wr
->buf_large
, &wr
->mr_large
);
212 static int ibw_conn_destruct(struct ibw_conn
*conn
)
214 DEBUG(DEBUG_DEBUG
, ("ibw_conn_destruct(%p)\n", conn
));
216 /* important here: ctx is a talloc _parent_ */
217 DLIST_REMOVE(conn
->ctx
->conn_list
, conn
);
221 struct ibw_conn
*ibw_conn_new(struct ibw_ctx
*ctx
, TALLOC_CTX
*mem_ctx
)
223 struct ibw_conn
*conn
;
224 struct ibw_conn_priv
*pconn
;
228 conn
= talloc_zero(mem_ctx
, struct ibw_conn
);
230 talloc_set_destructor(conn
, ibw_conn_destruct
);
232 pconn
= talloc_zero(conn
, struct ibw_conn_priv
);
234 talloc_set_destructor(pconn
, ibw_conn_priv_destruct
);
237 conn
->internal
= (void *)pconn
;
239 DLIST_ADD(ctx
->conn_list
, conn
);
244 static int ibw_setup_cq_qp(struct ibw_conn
*conn
)
246 struct ibw_ctx_priv
*pctx
= talloc_get_type(conn
->ctx
->internal
, struct ibw_ctx_priv
);
247 struct ibw_conn_priv
*pconn
= talloc_get_type(conn
->internal
, struct ibw_conn_priv
);
248 struct ibv_qp_init_attr init_attr
;
249 struct ibv_qp_attr attr
;
252 DEBUG(DEBUG_DEBUG
, ("ibw_setup_cq_qp(cmid: %p)\n", pconn
->cm_id
));
255 pconn
->verbs_channel
= ibv_create_comp_channel(pconn
->cm_id
->verbs
);
256 if (!pconn
->verbs_channel
) {
257 sprintf(ibw_lasterr
, "ibv_create_comp_channel failed %d\n", errno
);
260 DEBUG(DEBUG_DEBUG
, ("created channel %p\n", pconn
->verbs_channel
));
262 pconn
->verbs_channel_event
= tevent_add_fd(pctx
->ectx
, NULL
, /* not pconn or conn */
263 pconn
->verbs_channel
->fd
, TEVENT_FD_READ
, ibw_event_handler_verbs
, conn
);
265 pconn
->pd
= ibv_alloc_pd(pconn
->cm_id
->verbs
);
267 sprintf(ibw_lasterr
, "ibv_alloc_pd failed %d\n", errno
);
270 DEBUG(DEBUG_DEBUG
, ("created pd %p\n", pconn
->pd
));
273 if (ibw_init_memory(conn
))
277 pconn
->cq
= ibv_create_cq(pconn
->cm_id
->verbs
,
278 pctx
->opts
.max_recv_wr
+ pctx
->opts
.max_send_wr
,
279 conn
, pconn
->verbs_channel
, 0);
280 if (pconn
->cq
==NULL
) {
281 sprintf(ibw_lasterr
, "ibv_create_cq failed\n");
285 rc
= ibv_req_notify_cq(pconn
->cq
, 0);
287 sprintf(ibw_lasterr
, "ibv_req_notify_cq failed with %d\n", rc
);
292 memset(&init_attr
, 0, sizeof(init_attr
));
293 init_attr
.cap
.max_send_wr
= pctx
->opts
.max_send_wr
;
294 init_attr
.cap
.max_recv_wr
= pctx
->opts
.max_recv_wr
;
295 init_attr
.cap
.max_recv_sge
= 1;
296 init_attr
.cap
.max_send_sge
= 1;
297 init_attr
.qp_type
= IBV_QPT_RC
;
298 init_attr
.send_cq
= pconn
->cq
;
299 init_attr
.recv_cq
= pconn
->cq
;
301 rc
= rdma_create_qp(pconn
->cm_id
, pconn
->pd
, &init_attr
);
303 sprintf(ibw_lasterr
, "rdma_create_qp failed with %d\n", rc
);
306 /* elase result is in pconn->cm_id->qp */
308 rc
= ibv_query_qp(pconn
->cm_id
->qp
, &attr
, IBV_QP_PATH_MTU
, &init_attr
);
310 sprintf(ibw_lasterr
, "ibv_query_qp failed with %d\n", rc
);
314 return ibw_fill_cq(conn
);
317 static int ibw_refill_cq_recv(struct ibw_conn
*conn
)
319 struct ibw_ctx_priv
*pctx
= talloc_get_type(conn
->ctx
->internal
, struct ibw_ctx_priv
);
320 struct ibw_conn_priv
*pconn
= talloc_get_type(conn
->internal
, struct ibw_conn_priv
);
322 struct ibv_sge list
= {
323 .addr
= (uintptr_t) NULL
, /* filled below */
324 .length
= pctx
->opts
.recv_bufsize
,
325 .lkey
= pconn
->mr_recv
->lkey
/* always the same */
327 struct ibv_recv_wr wr
= {
328 .wr_id
= 0, /* filled below */
332 struct ibv_recv_wr
*bad_wr
;
334 DEBUG(DEBUG_DEBUG
, ("ibw_refill_cq_recv(cmid: %p)\n", pconn
->cm_id
));
336 list
.addr
= (uintptr_t) pconn
->buf_recv
+ pctx
->opts
.recv_bufsize
* pconn
->recv_index
;
337 wr
.wr_id
= pconn
->recv_index
;
338 pconn
->recv_index
= (pconn
->recv_index
+ 1) % pctx
->opts
.max_recv_wr
;
340 rc
= ibv_post_recv(pconn
->cm_id
->qp
, &wr
, &bad_wr
);
342 sprintf(ibw_lasterr
, "refill/ibv_post_recv failed with %d\n", rc
);
343 DEBUG(DEBUG_ERR
, ("%s", ibw_lasterr
));
350 static int ibw_fill_cq(struct ibw_conn
*conn
)
352 struct ibw_ctx_priv
*pctx
= talloc_get_type(conn
->ctx
->internal
, struct ibw_ctx_priv
);
353 struct ibw_conn_priv
*pconn
= talloc_get_type(conn
->internal
, struct ibw_conn_priv
);
355 struct ibv_sge list
= {
356 .addr
= (uintptr_t) NULL
, /* filled below */
357 .length
= pctx
->opts
.recv_bufsize
,
358 .lkey
= pconn
->mr_recv
->lkey
/* always the same */
360 struct ibv_recv_wr wr
= {
361 .wr_id
= 0, /* filled below */
365 struct ibv_recv_wr
*bad_wr
;
367 DEBUG(DEBUG_DEBUG
, ("ibw_fill_cq(cmid: %p)\n", pconn
->cm_id
));
369 for(i
= pctx
->opts
.max_recv_wr
; i
!=0; i
--) {
370 list
.addr
= (uintptr_t) pconn
->buf_recv
+ pctx
->opts
.recv_bufsize
* pconn
->recv_index
;
371 wr
.wr_id
= pconn
->recv_index
;
372 pconn
->recv_index
= (pconn
->recv_index
+ 1) % pctx
->opts
.max_recv_wr
;
374 rc
= ibv_post_recv(pconn
->cm_id
->qp
, &wr
, &bad_wr
);
376 sprintf(ibw_lasterr
, "fill/ibv_post_recv failed with %d\n", rc
);
377 DEBUG(DEBUG_ERR
, ("%s", ibw_lasterr
));
385 static int ibw_manage_connect(struct ibw_conn
*conn
)
387 struct rdma_conn_param conn_param
;
388 struct ibw_conn_priv
*pconn
= talloc_get_type(conn
->internal
, struct ibw_conn_priv
);
391 DEBUG(DEBUG_DEBUG
, ("ibw_manage_connect(cmid: %p)\n", pconn
->cm_id
));
393 if (ibw_setup_cq_qp(conn
))
397 memset(&conn_param
, 0, sizeof conn_param
);
398 conn_param
.responder_resources
= 1;
399 conn_param
.initiator_depth
= 1;
400 conn_param
.retry_count
= 10;
402 rc
= rdma_connect(pconn
->cm_id
, &conn_param
);
404 sprintf(ibw_lasterr
, "rdma_connect error %d\n", rc
);
409 static void ibw_event_handler_cm(struct tevent_context
*ev
,
410 struct tevent_fd
*fde
, uint16_t flags
, void *private_data
)
413 struct ibw_ctx
*ctx
= talloc_get_type(private_data
, struct ibw_ctx
);
414 struct ibw_ctx_priv
*pctx
= talloc_get_type(ctx
->internal
, struct ibw_ctx_priv
);
415 struct ibw_conn
*conn
= NULL
;
416 struct ibw_conn_priv
*pconn
= NULL
;
417 struct rdma_cm_id
*cma_id
= NULL
;
418 struct rdma_cm_event
*event
= NULL
;
422 rc
= rdma_get_cm_event(pctx
->cm_channel
, &event
);
424 ctx
->state
= IBWS_ERROR
;
426 sprintf(ibw_lasterr
, "rdma_get_cm_event error %d\n", rc
);
431 DEBUG(DEBUG_DEBUG
, ("cma_event type %d cma_id %p (%s)\n", event
->event
, cma_id
,
432 (cma_id
== pctx
->cm_id
) ? "parent" : "child"));
434 switch (event
->event
) {
435 case RDMA_CM_EVENT_ADDR_RESOLVED
:
436 DEBUG(DEBUG_DEBUG
, ("RDMA_CM_EVENT_ADDR_RESOLVED\n"));
437 /* continuing from ibw_connect ... */
438 rc
= rdma_resolve_route(cma_id
, 2000);
440 sprintf(ibw_lasterr
, "rdma_resolve_route error %d\n", rc
);
443 /* continued at RDMA_CM_EVENT_ROUTE_RESOLVED */
446 case RDMA_CM_EVENT_ROUTE_RESOLVED
:
447 DEBUG(DEBUG_DEBUG
, ("RDMA_CM_EVENT_ROUTE_RESOLVED\n"));
448 /* after RDMA_CM_EVENT_ADDR_RESOLVED: */
449 assert(cma_id
->context
!=NULL
);
450 conn
= talloc_get_type(cma_id
->context
, struct ibw_conn
);
452 rc
= ibw_manage_connect(conn
);
458 case RDMA_CM_EVENT_CONNECT_REQUEST
:
459 DEBUG(DEBUG_DEBUG
, ("RDMA_CM_EVENT_CONNECT_REQUEST\n"));
460 ctx
->state
= IBWS_CONNECT_REQUEST
;
461 conn
= ibw_conn_new(ctx
, ctx
);
462 pconn
= talloc_get_type(conn
->internal
, struct ibw_conn_priv
);
463 pconn
->cm_id
= cma_id
; /* !!! event will be freed but id not */
464 cma_id
->context
= (void *)conn
;
465 DEBUG(DEBUG_DEBUG
, ("pconn->cm_id %p\n", pconn
->cm_id
));
467 if (ibw_setup_cq_qp(conn
))
470 conn
->state
= IBWC_INIT
;
471 pctx
->connstate_func(ctx
, conn
);
473 /* continued at ibw_accept when invoked by the func above */
474 if (!pconn
->is_accepted
) {
475 rc
= rdma_reject(cma_id
, NULL
, 0);
477 DEBUG(DEBUG_ERR
, ("rdma_reject failed with rc=%d\n", rc
));
479 DEBUG(DEBUG_DEBUG
, ("pconn->cm_id %p wasn't accepted\n", pconn
->cm_id
));
482 /* TODO: clarify whether if it's needed by upper layer: */
483 ctx
->state
= IBWS_READY
;
484 pctx
->connstate_func(ctx
, NULL
);
486 /* NOTE: more requests can arrive until RDMA_CM_EVENT_ESTABLISHED ! */
489 case RDMA_CM_EVENT_ESTABLISHED
:
490 /* expected after ibw_accept and ibw_connect[not directly] */
491 DEBUG(DEBUG_INFO
, ("ESTABLISHED (conn: %p)\n", cma_id
->context
));
492 conn
= talloc_get_type(cma_id
->context
, struct ibw_conn
);
493 assert(conn
!=NULL
); /* important assumption */
495 DEBUG(DEBUG_DEBUG
, ("ibw_setup_cq_qp succeeded (cmid=%p)\n", cma_id
));
497 /* client conn is up */
498 conn
->state
= IBWC_CONNECTED
;
500 /* both ctx and conn have changed */
501 pctx
->connstate_func(ctx
, conn
);
504 case RDMA_CM_EVENT_ADDR_ERROR
:
505 sprintf(ibw_lasterr
, "RDMA_CM_EVENT_ADDR_ERROR, error %d\n", event
->status
);
507 case RDMA_CM_EVENT_ROUTE_ERROR
:
508 sprintf(ibw_lasterr
, "RDMA_CM_EVENT_ROUTE_ERROR, error %d\n", event
->status
);
510 case RDMA_CM_EVENT_CONNECT_ERROR
:
511 sprintf(ibw_lasterr
, "RDMA_CM_EVENT_CONNECT_ERROR, error %d\n", event
->status
);
513 case RDMA_CM_EVENT_UNREACHABLE
:
514 sprintf(ibw_lasterr
, "RDMA_CM_EVENT_UNREACHABLE, error %d\n", event
->status
);
516 case RDMA_CM_EVENT_REJECTED
:
517 sprintf(ibw_lasterr
, "RDMA_CM_EVENT_REJECTED, error %d\n", event
->status
);
518 DEBUG(DEBUG_INFO
, ("cm event handler: %s", ibw_lasterr
));
519 conn
= talloc_get_type(cma_id
->context
, struct ibw_conn
);
521 /* must be done BEFORE connstate */
522 if ((rc
=rdma_ack_cm_event(event
)))
523 DEBUG(DEBUG_ERR
, ("reject/rdma_ack_cm_event failed with %d\n", rc
));
524 event
= NULL
; /* not to touch cma_id or conn */
525 conn
->state
= IBWC_ERROR
;
526 /* it should free the conn */
527 pctx
->connstate_func(NULL
, conn
);
529 break; /* this is not strictly an error */
531 case RDMA_CM_EVENT_DISCONNECTED
:
532 DEBUG(DEBUG_DEBUG
, ("RDMA_CM_EVENT_DISCONNECTED\n"));
533 if ((rc
=rdma_ack_cm_event(event
)))
534 DEBUG(DEBUG_ERR
, ("disc/rdma_ack_cm_event failed with %d\n", rc
));
535 event
= NULL
; /* don't ack more */
537 if (cma_id
!=pctx
->cm_id
) {
538 DEBUG(DEBUG_ERR
, ("client DISCONNECT event cm_id=%p\n", cma_id
));
539 conn
= talloc_get_type(cma_id
->context
, struct ibw_conn
);
540 conn
->state
= IBWC_DISCONNECTED
;
541 pctx
->connstate_func(NULL
, conn
);
545 case RDMA_CM_EVENT_DEVICE_REMOVAL
:
546 sprintf(ibw_lasterr
, "cma detected device removal!\n");
550 sprintf(ibw_lasterr
, "unknown event %d\n", event
->event
);
554 if (event
!=NULL
&& (rc
=rdma_ack_cm_event(event
))) {
555 sprintf(ibw_lasterr
, "rdma_ack_cm_event failed with %d\n", rc
);
561 DEBUG(DEBUG_ERR
, ("cm event handler: %s", ibw_lasterr
));
564 if (cma_id
!=NULL
&& cma_id
!=pctx
->cm_id
) {
565 conn
= talloc_get_type(cma_id
->context
, struct ibw_conn
);
567 conn
->state
= IBWC_ERROR
;
568 pctx
->connstate_func(NULL
, conn
);
571 ctx
->state
= IBWS_ERROR
;
572 pctx
->connstate_func(ctx
, NULL
);
575 if ((rc
=rdma_ack_cm_event(event
))!=0) {
576 DEBUG(DEBUG_ERR
, ("rdma_ack_cm_event failed with %d\n", rc
));
583 static void ibw_event_handler_verbs(struct tevent_context
*ev
,
584 struct tevent_fd
*fde
, uint16_t flags
, void *private_data
)
586 struct ibw_conn
*conn
= talloc_get_type(private_data
, struct ibw_conn
);
587 struct ibw_conn_priv
*pconn
= talloc_get_type(conn
->internal
, struct ibw_conn_priv
);
588 struct ibw_ctx_priv
*pctx
= talloc_get_type(conn
->ctx
->internal
, struct ibw_ctx_priv
);
592 struct ibv_cq
*ev_cq
;
595 DEBUG(DEBUG_DEBUG
, ("ibw_event_handler_verbs(%u)\n", (uint32_t)flags
));
597 /* TODO: check whether if it's good to have more channels here... */
598 rc
= ibv_get_cq_event(pconn
->verbs_channel
, &ev_cq
, &ev_ctx
);
600 sprintf(ibw_lasterr
, "Failed to get cq_event with %d\n", rc
);
603 if (ev_cq
!= pconn
->cq
) {
604 sprintf(ibw_lasterr
, "ev_cq(%p) != pconn->cq(%p)\n", ev_cq
, pconn
->cq
);
607 rc
= ibv_req_notify_cq(pconn
->cq
, 0);
609 sprintf(ibw_lasterr
, "Couldn't request CQ notification (%d)\n", rc
);
613 while((rc
=ibv_poll_cq(pconn
->cq
, 1, &wc
))==1) {
615 sprintf(ibw_lasterr
, "cq completion failed status=%d, opcode=%d, rc=%d\n",
616 wc
.status
, wc
.opcode
, rc
);
622 DEBUG(DEBUG_DEBUG
, ("send completion\n"));
623 if (ibw_wc_send(conn
, &wc
))
627 case IBV_WC_RDMA_WRITE
:
628 DEBUG(DEBUG_DEBUG
, ("rdma write completion\n"));
631 case IBV_WC_RDMA_READ
:
632 DEBUG(DEBUG_DEBUG
, ("rdma read completion\n"));
636 DEBUG(DEBUG_DEBUG
, ("recv completion\n"));
637 if (ibw_wc_recv(conn
, &wc
))
642 sprintf(ibw_lasterr
, "unknown completion %d\n", wc
.opcode
);
647 sprintf(ibw_lasterr
, "ibv_poll_cq error %d\n", rc
);
651 ibv_ack_cq_events(pconn
->cq
, 1);
655 ibv_ack_cq_events(pconn
->cq
, 1);
657 DEBUG(DEBUG_ERR
, ("%s", ibw_lasterr
));
659 if (conn
->state
!=IBWC_ERROR
) {
660 conn
->state
= IBWC_ERROR
;
661 pctx
->connstate_func(NULL
, conn
);
665 static int ibw_process_queue(struct ibw_conn
*conn
)
667 struct ibw_conn_priv
*pconn
= talloc_get_type(conn
->internal
, struct ibw_conn_priv
);
668 struct ibw_ctx_priv
*pctx
;
673 if (pconn
->queue
==NULL
)
678 /* we must have at least 1 fragment to send */
679 assert(p
->queued_ref_cnt
>0);
682 pctx
= talloc_get_type(conn
->ctx
->internal
, struct ibw_ctx_priv
);
683 msg_size
= (p
->queued_ref_cnt
) ? pctx
->opts
.recv_bufsize
: p
->queued_rlen
;
685 assert(p
->queued_msg
!=NULL
);
688 DEBUG(DEBUG_DEBUG
, ("ibw_process_queue refcnt=%d msgsize=%u\n",
689 p
->queued_ref_cnt
, msg_size
));
691 rc
= ibw_send_packet(conn
, p
->queued_msg
, p
, msg_size
);
693 /* was this the last fragment? */
694 if (p
->queued_ref_cnt
) {
695 p
->queued_msg
+= pctx
->opts
.recv_bufsize
;
697 DLIST_REMOVE2(pconn
->queue
, p
, qprev
, qnext
);
698 p
->queued_msg
= NULL
;
704 static int ibw_wc_send(struct ibw_conn
*conn
, struct ibv_wc
*wc
)
706 struct ibw_ctx_priv
*pctx
= talloc_get_type(conn
->ctx
->internal
, struct ibw_ctx_priv
);
707 struct ibw_conn_priv
*pconn
= talloc_get_type(conn
->internal
, struct ibw_conn_priv
);
711 DEBUG(DEBUG_DEBUG
, ("ibw_wc_send(cmid: %p, wr_id: %u, bl: %u)\n",
712 pconn
->cm_id
, (uint32_t)wc
->wr_id
, (uint32_t)wc
->byte_len
));
714 assert(pconn
->cm_id
->qp
->qp_num
==wc
->qp_num
);
715 assert(wc
->wr_id
>= pctx
->opts
.max_recv_wr
);
716 send_index
= wc
->wr_id
- pctx
->opts
.max_recv_wr
;
719 if (send_index
< pctx
->opts
.max_send_wr
) {
720 DEBUG(DEBUG_DEBUG
, ("ibw_wc_send#1 %u\n", (int)wc
->wr_id
));
721 p
= pconn
->wr_index
[send_index
];
722 if (p
->buf_large
!=NULL
) {
724 /* awaiting more of it... */
727 ibw_free_mr(&p
->buf_large
, &p
->mr_large
);
728 DLIST_REMOVE(pconn
->wr_list_used
, p
);
729 DLIST_ADD(pconn
->wr_list_avail
, p
);
731 } else { /* nasty - but necessary */
732 DLIST_REMOVE(pconn
->wr_list_used
, p
);
733 DLIST_ADD(pconn
->wr_list_avail
, p
);
735 } else { /* "extra" request - not optimized */
736 DEBUG(DEBUG_DEBUG
, ("ibw_wc_send#2 %u\n", (int)wc
->wr_id
));
737 for(p
=pconn
->extra_sent
; p
!=NULL
; p
=p
->next
)
738 if ((p
->wr_id
+ pctx
->opts
.max_recv_wr
)==(int)wc
->wr_id
)
741 sprintf(ibw_lasterr
, "failed to find wr_id %d\n", (int)wc
->wr_id
);
747 ibw_free_mr(&p
->buf_large
, &p
->mr_large
);
748 DLIST_REMOVE(pconn
->extra_sent
, p
);
749 DLIST_ADD(pconn
->extra_avail
, p
);
753 return ibw_process_queue(conn
);
756 static int ibw_append_to_part(struct ibw_conn_priv
*pconn
,
757 struct ibw_part
*part
, char **pp
, uint32_t add_len
, int info
)
759 DEBUG(DEBUG_DEBUG
, ("ibw_append_to_part: cmid=%p, (bs=%u, len=%u, tr=%u), al=%u, i=%u\n",
760 pconn
->cm_id
, part
->bufsize
, part
->len
, part
->to_read
, add_len
, info
));
762 /* allocate more if necessary - it's an "evergrowing" buffer... */
763 if (part
->len
+ add_len
> part
->bufsize
) {
764 if (part
->buf
==NULL
) {
765 assert(part
->len
==0);
766 part
->buf
= talloc_size(pconn
, add_len
);
767 if (part
->buf
==NULL
) {
768 sprintf(ibw_lasterr
, "recv talloc_size error (%u) #%d\n",
772 part
->bufsize
= add_len
;
774 part
->buf
= talloc_realloc_size(pconn
,
775 part
->buf
, part
->len
+ add_len
);
776 if (part
->buf
==NULL
) {
777 sprintf(ibw_lasterr
, "recv realloc error (%u + %u) #%d\n",
778 part
->len
, add_len
, info
);
782 part
->bufsize
= part
->len
+ add_len
;
786 memcpy(part
->buf
+ part
->len
, *pp
, add_len
);
788 part
->len
+= add_len
;
789 part
->to_read
-= add_len
;
794 static int ibw_wc_mem_threshold(struct ibw_conn_priv
*pconn
,
795 struct ibw_part
*part
, uint32_t threshold
)
797 DEBUG(DEBUG_DEBUG
, ("ibw_wc_mem_threshold: cmid=%p, (bs=%u, len=%u, tr=%u), thr=%u\n",
798 pconn
->cm_id
, part
->bufsize
, part
->len
, part
->to_read
, threshold
));
800 if (part
->bufsize
> threshold
) {
801 DEBUG(DEBUG_DEBUG
, ("ibw_wc_mem_threshold: cmid=%p, %u > %u\n",
802 pconn
->cm_id
, part
->bufsize
, threshold
));
803 talloc_free(part
->buf
);
804 part
->buf
= talloc_size(pconn
, threshold
);
805 if (part
->buf
==NULL
) {
806 sprintf(ibw_lasterr
, "talloc_size failed\n");
809 part
->bufsize
= threshold
;
814 static int ibw_wc_recv(struct ibw_conn
*conn
, struct ibv_wc
*wc
)
816 struct ibw_ctx_priv
*pctx
= talloc_get_type(conn
->ctx
->internal
, struct ibw_ctx_priv
);
817 struct ibw_conn_priv
*pconn
= talloc_get_type(conn
->internal
, struct ibw_conn_priv
);
818 struct ibw_part
*part
= &pconn
->part
;
820 uint32_t remain
= wc
->byte_len
;
822 DEBUG(DEBUG_DEBUG
, ("ibw_wc_recv: cmid=%p, wr_id: %u, bl: %u\n",
823 pconn
->cm_id
, (uint32_t)wc
->wr_id
, remain
));
825 assert(pconn
->cm_id
->qp
->qp_num
==wc
->qp_num
);
826 assert((int)wc
->wr_id
< pctx
->opts
.max_recv_wr
);
827 assert(wc
->byte_len
<= pctx
->opts
.recv_bufsize
);
829 p
= pconn
->buf_recv
+ ((int)wc
->wr_id
* pctx
->opts
.recv_bufsize
);
832 /* here always true: (part->len!=0 && part->to_read!=0) ||
833 (part->len==0 && part->to_read==0) */
834 if (part
->len
) { /* is there a partial msg to be continued? */
835 int read_len
= (part
->to_read
<=remain
) ? part
->to_read
: remain
;
836 if (ibw_append_to_part(pconn
, part
, &p
, read_len
, 421))
840 if (part
->len
<=sizeof(uint32_t) && part
->to_read
==0) {
841 assert(part
->len
==sizeof(uint32_t));
842 /* set it again now... */
843 part
->to_read
= *((uint32_t *)(part
->buf
)); /* TODO: ntohl */
844 if (part
->to_read
<sizeof(uint32_t)) {
845 sprintf(ibw_lasterr
, "got msglen=%u #2\n", part
->to_read
);
848 part
->to_read
-= sizeof(uint32_t); /* it's already read */
851 if (part
->to_read
==0) {
852 if (pctx
->receive_func(conn
, part
->buf
, part
->len
) != 0) {
855 part
->len
= 0; /* tells not having partial data (any more) */
856 if (ibw_wc_mem_threshold(pconn
, part
, pctx
->opts
.recv_threshold
))
860 if (remain
>=sizeof(uint32_t)) {
861 uint32_t msglen
= *(uint32_t *)p
; /* TODO: ntohl */
862 if (msglen
<sizeof(uint32_t)) {
863 sprintf(ibw_lasterr
, "got msglen=%u\n", msglen
);
867 /* mostly awaited case: */
868 if (msglen
<=remain
) {
869 if (pctx
->receive_func(conn
, p
, msglen
) != 0) {
875 part
->to_read
= msglen
;
876 /* part->len is already 0 */
877 if (ibw_append_to_part(pconn
, part
, &p
, remain
, 422))
879 remain
= 0; /* to be continued ... */
880 /* part->to_read > 0 here */
882 } else { /* edge case: */
883 part
->to_read
= sizeof(uint32_t);
884 /* part->len is already 0 */
885 if (ibw_append_to_part(pconn
, part
, &p
, remain
, 423))
888 /* part->to_read > 0 here */
891 } /* <remain> is always decreased at least by 1 */
893 if (ibw_refill_cq_recv(conn
))
899 DEBUG(DEBUG_ERR
, ("ibw_wc_recv error: %s", ibw_lasterr
));
903 static int ibw_process_init_attrs(struct ibw_initattr
*attr
, int nattr
, struct ibw_opts
*opts
)
906 const char *name
, *value
;
908 DEBUG(DEBUG_DEBUG
, ("ibw_process_init_attrs: nattr: %d\n", nattr
));
910 opts
->max_send_wr
= IBW_MAX_SEND_WR
;
911 opts
->max_recv_wr
= IBW_MAX_RECV_WR
;
912 opts
->recv_bufsize
= IBW_RECV_BUFSIZE
;
913 opts
->recv_threshold
= IBW_RECV_THRESHOLD
;
915 for(i
=0; i
<nattr
; i
++) {
917 value
= attr
[i
].value
;
919 assert(name
!=NULL
&& value
!=NULL
);
920 if (strcmp(name
, "max_send_wr")==0)
921 opts
->max_send_wr
= atoi(value
);
922 else if (strcmp(name
, "max_recv_wr")==0)
923 opts
->max_recv_wr
= atoi(value
);
924 else if (strcmp(name
, "recv_bufsize")==0)
925 opts
->recv_bufsize
= atoi(value
);
926 else if (strcmp(name
, "recv_threshold")==0)
927 opts
->recv_threshold
= atoi(value
);
929 sprintf(ibw_lasterr
, "ibw_init: unknown name %s\n", name
);
936 struct ibw_ctx
*ibw_init(struct ibw_initattr
*attr
, int nattr
,
938 ibw_connstate_fn_t ibw_connstate
,
939 ibw_receive_fn_t ibw_receive
,
940 struct tevent_context
*ectx
)
942 struct ibw_ctx
*ctx
= talloc_zero(NULL
, struct ibw_ctx
);
943 struct ibw_ctx_priv
*pctx
;
946 DEBUG(DEBUG_DEBUG
, ("ibw_init(ctx_userdata: %p, ectx: %p)\n", ctx_userdata
, ectx
));
948 /* initialize basic data structures */
949 memset(ibw_lasterr
, 0, IBW_LASTERR_BUFSIZE
);
952 ibw_lasterr
[0] = '\0';
953 talloc_set_destructor(ctx
, ibw_ctx_destruct
);
954 ctx
->ctx_userdata
= ctx_userdata
;
956 pctx
= talloc_zero(ctx
, struct ibw_ctx_priv
);
957 talloc_set_destructor(pctx
, ibw_ctx_priv_destruct
);
958 ctx
->internal
= (void *)pctx
;
961 pctx
->connstate_func
= ibw_connstate
;
962 pctx
->receive_func
= ibw_receive
;
966 /* process attributes */
967 if (ibw_process_init_attrs(attr
, nattr
, &pctx
->opts
))
971 pctx
->cm_channel
= rdma_create_event_channel();
972 if (!pctx
->cm_channel
) {
973 sprintf(ibw_lasterr
, "rdma_create_event_channel error %d\n", errno
);
977 pctx
->cm_channel_event
= tevent_add_fd(pctx
->ectx
, pctx
,
978 pctx
->cm_channel
->fd
, TEVENT_FD_READ
, ibw_event_handler_cm
, ctx
);
980 #if RDMA_USER_CM_MAX_ABI_VERSION >= 2
981 rc
= rdma_create_id(pctx
->cm_channel
, &pctx
->cm_id
, ctx
, RDMA_PS_TCP
);
983 rc
= rdma_create_id(pctx
->cm_channel
, &pctx
->cm_id
, ctx
);
987 sprintf(ibw_lasterr
, "rdma_create_id error %d\n", rc
);
990 DEBUG(DEBUG_DEBUG
, ("created cm_id %p\n", pctx
->cm_id
));
992 pctx
->pagesize
= sysconf(_SC_PAGESIZE
);
995 /* don't put code here */
997 DEBUG(DEBUG_ERR
, ("%s", ibw_lasterr
));
1005 int ibw_stop(struct ibw_ctx
*ctx
)
1007 struct ibw_ctx_priv
*pctx
= (struct ibw_ctx_priv
*)ctx
->internal
;
1010 DEBUG(DEBUG_DEBUG
, ("ibw_stop\n"));
1012 for(p
=ctx
->conn_list
; p
!=NULL
; p
=p
->next
) {
1013 if (p
->state
==IBWC_ERROR
|| p
->state
==IBWC_CONNECTED
) {
1014 if (ibw_disconnect(p
))
1019 ctx
->state
= IBWS_STOPPED
;
1020 pctx
->connstate_func(ctx
, NULL
);
1025 int ibw_bind(struct ibw_ctx
*ctx
, struct sockaddr_in
*my_addr
)
1027 struct ibw_ctx_priv
*pctx
= (struct ibw_ctx_priv
*)ctx
->internal
;
1030 DEBUG(DEBUG_DEBUG
, ("ibw_bind: addr=%s, port=%u\n",
1031 inet_ntoa(my_addr
->sin_addr
), ntohs(my_addr
->sin_port
)));
1032 rc
= rdma_bind_addr(pctx
->cm_id
, (struct sockaddr
*) my_addr
);
1034 sprintf(ibw_lasterr
, "rdma_bind_addr error %d\n", rc
);
1035 DEBUG(DEBUG_ERR
, ("%s", ibw_lasterr
));
1038 DEBUG(DEBUG_DEBUG
, ("rdma_bind_addr successful\n"));
1043 int ibw_listen(struct ibw_ctx
*ctx
, int backlog
)
1045 struct ibw_ctx_priv
*pctx
= talloc_get_type(ctx
->internal
, struct ibw_ctx_priv
);
1048 DEBUG(DEBUG_DEBUG
, ("ibw_listen\n"));
1049 rc
= rdma_listen(pctx
->cm_id
, backlog
);
1051 sprintf(ibw_lasterr
, "rdma_listen failed: %d\n", rc
);
1052 DEBUG(DEBUG_ERR
, ("%s", ibw_lasterr
));
1059 int ibw_accept(struct ibw_ctx
*ctx
, struct ibw_conn
*conn
, void *conn_userdata
)
1061 struct ibw_conn_priv
*pconn
= talloc_get_type(conn
->internal
, struct ibw_conn_priv
);
1062 struct rdma_conn_param conn_param
;
1065 DEBUG(DEBUG_DEBUG
, ("ibw_accept: cmid=%p\n", pconn
->cm_id
));
1066 conn
->conn_userdata
= conn_userdata
;
1068 memset(&conn_param
, 0, sizeof(struct rdma_conn_param
));
1069 conn_param
.responder_resources
= 1;
1070 conn_param
.initiator_depth
= 1;
1071 rc
= rdma_accept(pconn
->cm_id
, &conn_param
);
1073 sprintf(ibw_lasterr
, "rdma_accept failed %d\n", rc
);
1074 DEBUG(DEBUG_ERR
, ("%s", ibw_lasterr
));
1078 pconn
->is_accepted
= 1;
1080 /* continued at RDMA_CM_EVENT_ESTABLISHED */
1085 int ibw_connect(struct ibw_conn
*conn
, struct sockaddr_in
*serv_addr
, void *conn_userdata
)
1087 struct ibw_ctx_priv
*pctx
= talloc_get_type(conn
->ctx
->internal
, struct ibw_ctx_priv
);
1088 struct ibw_conn_priv
*pconn
= NULL
;
1093 conn
->conn_userdata
= conn_userdata
;
1094 pconn
= talloc_get_type(conn
->internal
, struct ibw_conn_priv
);
1095 DEBUG(DEBUG_DEBUG
, ("ibw_connect: addr=%s, port=%u\n", inet_ntoa(serv_addr
->sin_addr
),
1096 ntohs(serv_addr
->sin_port
)));
1098 /* clean previous - probably half - initialization */
1099 if (ibw_conn_priv_destruct(pconn
)) {
1100 DEBUG(DEBUG_ERR
, ("ibw_connect/ibw_pconn_destruct failed for cm_id=%p\n", pconn
->cm_id
));
1105 #if RDMA_USER_CM_MAX_ABI_VERSION >= 2
1106 rc
= rdma_create_id(pctx
->cm_channel
, &pconn
->cm_id
, conn
, RDMA_PS_TCP
);
1108 rc
= rdma_create_id(pctx
->cm_channel
, &pconn
->cm_id
, conn
);
1112 sprintf(ibw_lasterr
, "ibw_connect/rdma_create_id error %d\n", rc
);
1116 DEBUG(DEBUG_DEBUG
, ("ibw_connect: rdma_create_id succeeded, cm_id=%p\n", pconn
->cm_id
));
1118 rc
= rdma_resolve_addr(pconn
->cm_id
, NULL
, (struct sockaddr
*) serv_addr
, 2000);
1120 sprintf(ibw_lasterr
, "rdma_resolve_addr error %d\n", rc
);
1121 DEBUG(DEBUG_ERR
, ("%s", ibw_lasterr
));
1126 /* continued at RDMA_CM_EVENT_ADDR_RESOLVED */
1131 int ibw_disconnect(struct ibw_conn
*conn
)
1134 struct ibw_conn_priv
*pconn
= talloc_get_type(conn
->internal
, struct ibw_conn_priv
);
1136 DEBUG(DEBUG_DEBUG
, ("ibw_disconnect: cmid=%p\n", pconn
->cm_id
));
1138 assert(pconn
!=NULL
);
1140 switch(conn
->state
) {
1142 ibw_conn_priv_destruct(pconn
); /* do this here right now */
1144 case IBWC_CONNECTED
:
1145 rc
= rdma_disconnect(pconn
->cm_id
);
1147 sprintf(ibw_lasterr
, "ibw_disconnect failed with %d\n", rc
);
1148 DEBUG(DEBUG_ERR
, ("%s", ibw_lasterr
));
1153 DEBUG(DEBUG_DEBUG
, ("invalid state for disconnect: %d\n", conn
->state
));
1160 int ibw_alloc_send_buf(struct ibw_conn
*conn
, void **buf
, void **key
, uint32_t len
)
1162 struct ibw_ctx_priv
*pctx
= talloc_get_type(conn
->ctx
->internal
, struct ibw_ctx_priv
);
1163 struct ibw_conn_priv
*pconn
= talloc_get_type(conn
->internal
, struct ibw_conn_priv
);
1164 struct ibw_wr
*p
= pconn
->wr_list_avail
;
1167 DEBUG(DEBUG_DEBUG
, ("ibw_alloc_send_buf#1: cmid=%p, len=%d\n", pconn
->cm_id
, len
));
1169 DLIST_REMOVE(pconn
->wr_list_avail
, p
);
1170 DLIST_ADD(pconn
->wr_list_used
, p
);
1172 if (len
<= pctx
->opts
.recv_bufsize
) {
1173 *buf
= (void *)p
->buf
;
1175 p
->buf_large
= ibw_alloc_mr(pctx
, pconn
, len
, &p
->mr_large
);
1176 if (p
->buf_large
==NULL
) {
1177 sprintf(ibw_lasterr
, "ibw_alloc_mr#1 failed\n");
1180 *buf
= (void *)p
->buf_large
;
1182 /* p->wr_id is already filled in ibw_init_memory */
1184 DEBUG(DEBUG_DEBUG
, ("ibw_alloc_send_buf#2: cmid=%p, len=%d\n", pconn
->cm_id
, len
));
1186 p
= pconn
->extra_avail
;
1188 p
= pconn
->extra_avail
= talloc_zero(pconn
, struct ibw_wr
);
1189 talloc_set_destructor(p
, ibw_wr_destruct
);
1191 sprintf(ibw_lasterr
, "talloc_zero failed (emax: %u)\n", pconn
->extra_max
);
1194 p
->wr_id
= pctx
->opts
.max_send_wr
+ pconn
->extra_max
;
1196 switch(pconn
->extra_max
) {
1197 case 1: DEBUG(DEBUG_INFO
, ("warning: queue performed\n")); break;
1198 case 10: DEBUG(DEBUG_INFO
, ("warning: queue reached 10\n")); break;
1199 case 100: DEBUG(DEBUG_INFO
, ("warning: queue reached 100\n")); break;
1200 case 1000: DEBUG(DEBUG_INFO
, ("warning: queue reached 1000\n")); break;
1205 p
->buf_large
= ibw_alloc_mr(pctx
, pconn
, len
, &p
->mr_large
);
1206 if (p
->buf_large
==NULL
) {
1207 sprintf(ibw_lasterr
, "ibw_alloc_mr#2 failed\n");
1210 *buf
= (void *)p
->buf_large
;
1212 DLIST_REMOVE(pconn
->extra_avail
, p
);
1213 /* we don't have prepared index for this, so that
1214 * we will have to find this by wr_id later on */
1215 DLIST_ADD(pconn
->extra_sent
, p
);
1222 DEBUG(DEBUG_ERR
, ("ibw_alloc_send_buf error: %s", ibw_lasterr
));
1227 static int ibw_send_packet(struct ibw_conn
*conn
, void *buf
, struct ibw_wr
*p
, uint32_t len
)
1229 struct ibw_ctx_priv
*pctx
= talloc_get_type(conn
->ctx
->internal
, struct ibw_ctx_priv
);
1230 struct ibw_conn_priv
*pconn
= talloc_get_type(conn
->internal
, struct ibw_conn_priv
);
1233 /* can we send it right now? */
1234 if (pconn
->wr_sent
<pctx
->opts
.max_send_wr
) {
1235 struct ibv_send_wr
*bad_wr
;
1236 struct ibv_sge list
= {
1237 .addr
= (uintptr_t)buf
,
1239 .lkey
= pconn
->mr_send
->lkey
1241 struct ibv_send_wr wr
= {
1242 .wr_id
= p
->wr_id
+ pctx
->opts
.max_recv_wr
,
1245 .opcode
= IBV_WR_SEND
,
1246 .send_flags
= IBV_SEND_SIGNALED
,
1249 if (p
->buf_large
==NULL
) {
1250 DEBUG(DEBUG_DEBUG
, ("ibw_send#normal(cmid: %p, wrid: %u, n: %d)\n",
1251 pconn
->cm_id
, (uint32_t)wr
.wr_id
, len
));
1253 DEBUG(DEBUG_DEBUG
, ("ibw_send#large(cmid: %p, wrid: %u, n: %d)\n",
1254 pconn
->cm_id
, (uint32_t)wr
.wr_id
, len
));
1255 list
.lkey
= p
->mr_large
->lkey
;
1258 rc
= ibv_post_send(pconn
->cm_id
->qp
, &wr
, &bad_wr
);
1260 sprintf(ibw_lasterr
, "ibv_post_send error %d (%d)\n",
1261 rc
, pconn
->wr_sent
);
1268 } /* else put the request into our own queue: */
1270 DEBUG(DEBUG_DEBUG
, ("ibw_send#queued(cmid: %p, len: %u)\n", pconn
->cm_id
, len
));
1272 /* TODO: clarify how to continue when state==IBWC_STOPPED */
1274 /* to be sent by ibw_wc_send */
1275 /* regardless "normal" or [a part of] "large" packet */
1276 if (!p
->queued_ref_cnt
) {
1277 DLIST_ADD_END2(pconn
->queue
, p
, struct ibw_wr
*,
1278 qprev
, qnext
); /* TODO: optimize */
1279 p
->queued_msg
= buf
;
1281 p
->queued_ref_cnt
++;
1282 p
->queued_rlen
= len
; /* last wins; see ibw_wc_send */
1286 DEBUG(DEBUG_ERR
, ("%s", ibw_lasterr
));
1290 int ibw_send(struct ibw_conn
*conn
, void *buf
, void *key
, uint32_t len
)
1292 struct ibw_ctx_priv
*pctx
= talloc_get_type(conn
->ctx
->internal
, struct ibw_ctx_priv
);
1293 struct ibw_wr
*p
= talloc_get_type(key
, struct ibw_wr
);
1296 assert(len
>=sizeof(uint32_t));
1297 assert((*((uint32_t *)buf
)==len
)); /* TODO: htonl */
1299 if (len
> pctx
->opts
.recv_bufsize
) {
1300 struct ibw_conn_priv
*pconn
= talloc_get_type(conn
->internal
, struct ibw_conn_priv
);
1302 char *packet
= (char *)buf
;
1303 uint32_t recv_bufsize
= pctx
->opts
.recv_bufsize
;
1305 DEBUG(DEBUG_DEBUG
, ("ibw_send#frag(cmid: %p, buf: %p, len: %u)\n",
1306 pconn
->cm_id
, buf
, len
));
1308 /* single threaded => no race here: */
1309 assert(p
->ref_cnt
==0);
1310 while(rlen
> recv_bufsize
) {
1311 rc
= ibw_send_packet(conn
, packet
, p
, recv_bufsize
);
1314 packet
+= recv_bufsize
;
1315 rlen
-= recv_bufsize
;
1316 p
->ref_cnt
++; /* not good to have it in ibw_send_packet */
1319 rc
= ibw_send_packet(conn
, packet
, p
, rlen
);
1320 p
->ref_cnt
++; /* not good to have it in ibw_send_packet */
1322 p
->ref_cnt
--; /* for the same handling */
1324 assert(p
->ref_cnt
==0);
1325 assert(p
->queued_ref_cnt
==0);
1327 rc
= ibw_send_packet(conn
, buf
, p
, len
);
1332 int ibw_cancel_send_buf(struct ibw_conn
*conn
, void *buf
, void *key
)
1334 struct ibw_ctx_priv
*pctx
= talloc_get_type(conn
->ctx
->internal
, struct ibw_ctx_priv
);
1335 struct ibw_conn_priv
*pconn
= talloc_get_type(conn
->internal
, struct ibw_conn_priv
);
1336 struct ibw_wr
*p
= talloc_get_type(key
, struct ibw_wr
);
1342 if (p
->buf_large
!=NULL
)
1343 ibw_free_mr(&p
->buf_large
, &p
->mr_large
);
1346 if (p
->wr_id
< pctx
->opts
.max_send_wr
) {
1347 DEBUG(DEBUG_DEBUG
, ("ibw_cancel_send_buf#1 %u", (int)p
->wr_id
));
1348 DLIST_REMOVE(pconn
->wr_list_used
, p
);
1349 DLIST_ADD(pconn
->wr_list_avail
, p
);
1350 } else { /* "extra" packet */
1351 DEBUG(DEBUG_DEBUG
, ("ibw_cancel_send_buf#2 %u", (int)p
->wr_id
));
1352 DLIST_REMOVE(pconn
->extra_sent
, p
);
1353 DLIST_ADD(pconn
->extra_avail
, p
);
1359 const char *ibw_getLastError(void)