2 * Copyright (c) 2006 Oracle. All rights reserved.
4 * This software is available to you under a choice of one of two
5 * licenses. You may choose to be licensed under the terms of the GNU
6 * General Public License (GPL) Version 2, available from the file
7 * COPYING in the main directory of this source tree, or the
8 * OpenIB.org BSD license below:
10 * Redistribution and use in source and binary forms, with or
11 * without modification, are permitted provided that the following
14 * - Redistributions of source code must retain the above
15 * copyright notice, this list of conditions and the following
18 * - Redistributions in binary form must reproduce the above
19 * copyright notice, this list of conditions and the following
20 * disclaimer in the documentation and/or other materials
21 * provided with the distribution.
23 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
24 * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
25 * MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
26 * NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS
27 * BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN
28 * ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
29 * CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
33 #include <linux/kernel.h>
35 #include <linux/device.h>
36 #include <linux/dmapool.h>
42 * Convert IB-specific error message to RDS error message and call core
45 static void rds_ib_send_complete(struct rds_message
*rm
,
47 void (*complete
)(struct rds_message
*rm
, int status
))
52 case IB_WC_WR_FLUSH_ERR
:
56 notify_status
= RDS_RDMA_SUCCESS
;
59 case IB_WC_REM_ACCESS_ERR
:
60 notify_status
= RDS_RDMA_REMOTE_ERROR
;
64 notify_status
= RDS_RDMA_OTHER_ERROR
;
67 complete(rm
, notify_status
);
70 static void rds_ib_send_unmap_data(struct rds_ib_connection
*ic
,
71 struct rm_data_op
*op
,
75 ib_dma_unmap_sg(ic
->i_cm_id
->device
,
76 op
->op_sg
, op
->op_nents
,
80 static void rds_ib_send_unmap_rdma(struct rds_ib_connection
*ic
,
81 struct rm_rdma_op
*op
,
85 ib_dma_unmap_sg(ic
->i_cm_id
->device
,
86 op
->op_sg
, op
->op_nents
,
87 op
->op_write
? DMA_TO_DEVICE
: DMA_FROM_DEVICE
);
91 /* If the user asked for a completion notification on this
92 * message, we can implement three different semantics:
93 * 1. Notify when we received the ACK on the RDS message
94 * that was queued with the RDMA. This provides reliable
95 * notification of RDMA status at the expense of a one-way
97 * 2. Notify when the IB stack gives us the completion event for
99 * 3. Notify when the IB stack gives us the completion event for
100 * the accompanying RDS messages.
101 * Here, we implement approach #3. To implement approach #2,
102 * we would need to take an event for the rdma WR. To implement #1,
103 * don't call rds_rdma_send_complete at all, and fall back to the notify
104 * handling in the ACK processing code.
106 * Note: There's no need to explicitly sync any RDMA buffers using
107 * ib_dma_sync_sg_for_cpu - the completion for the RDMA
108 * operation itself unmapped the RDMA buffers, which takes care
111 rds_ib_send_complete(container_of(op
, struct rds_message
, rdma
),
112 wc_status
, rds_rdma_send_complete
);
115 rds_stats_add(s_send_rdma_bytes
, op
->op_bytes
);
117 rds_stats_add(s_recv_rdma_bytes
, op
->op_bytes
);
120 static void rds_ib_send_unmap_atomic(struct rds_ib_connection
*ic
,
121 struct rm_atomic_op
*op
,
124 /* unmap atomic recvbuf */
126 ib_dma_unmap_sg(ic
->i_cm_id
->device
, op
->op_sg
, 1,
131 rds_ib_send_complete(container_of(op
, struct rds_message
, atomic
),
132 wc_status
, rds_atomic_send_complete
);
134 if (op
->op_type
== RDS_ATOMIC_TYPE_CSWP
)
135 rds_ib_stats_inc(s_ib_atomic_cswp
);
137 rds_ib_stats_inc(s_ib_atomic_fadd
);
141 * Unmap the resources associated with a struct send_work.
143 * Returns the rm for no good reason other than it is unobtainable
144 * other than by switching on wr.opcode, currently, and the caller,
145 * the event handler, needs it.
147 static struct rds_message
*rds_ib_send_unmap_op(struct rds_ib_connection
*ic
,
148 struct rds_ib_send_work
*send
,
151 struct rds_message
*rm
= NULL
;
153 /* In the error case, wc.opcode sometimes contains garbage */
154 switch (send
->s_wr
.opcode
) {
157 rm
= container_of(send
->s_op
, struct rds_message
, data
);
158 rds_ib_send_unmap_data(ic
, send
->s_op
, wc_status
);
161 case IB_WR_RDMA_WRITE
:
162 case IB_WR_RDMA_READ
:
164 rm
= container_of(send
->s_op
, struct rds_message
, rdma
);
165 rds_ib_send_unmap_rdma(ic
, send
->s_op
, wc_status
);
168 case IB_WR_ATOMIC_FETCH_AND_ADD
:
169 case IB_WR_ATOMIC_CMP_AND_SWP
:
171 rm
= container_of(send
->s_op
, struct rds_message
, atomic
);
172 rds_ib_send_unmap_atomic(ic
, send
->s_op
, wc_status
);
176 if (printk_ratelimit())
178 "RDS/IB: %s: unexpected opcode 0x%x in WR!\n",
179 __func__
, send
->s_wr
.opcode
);
183 send
->s_wr
.opcode
= 0xdead;
188 void rds_ib_send_init_ring(struct rds_ib_connection
*ic
)
190 struct rds_ib_send_work
*send
;
193 for (i
= 0, send
= ic
->i_sends
; i
< ic
->i_send_ring
.w_nr
; i
++, send
++) {
198 send
->s_wr
.wr_id
= i
;
199 send
->s_wr
.sg_list
= send
->s_sge
;
200 send
->s_wr
.ex
.imm_data
= 0;
202 sge
= &send
->s_sge
[0];
203 sge
->addr
= ic
->i_send_hdrs_dma
+ (i
* sizeof(struct rds_header
));
204 sge
->length
= sizeof(struct rds_header
);
205 sge
->lkey
= ic
->i_mr
->lkey
;
207 send
->s_sge
[1].lkey
= ic
->i_mr
->lkey
;
211 void rds_ib_send_clear_ring(struct rds_ib_connection
*ic
)
213 struct rds_ib_send_work
*send
;
216 for (i
= 0, send
= ic
->i_sends
; i
< ic
->i_send_ring
.w_nr
; i
++, send
++) {
217 if (send
->s_op
&& send
->s_wr
.opcode
!= 0xdead)
218 rds_ib_send_unmap_op(ic
, send
, IB_WC_WR_FLUSH_ERR
);
223 * The _oldest/_free ring operations here race cleanly with the alloc/unalloc
224 * operations performed in the send path. As the sender allocs and potentially
225 * unallocs the next free entry in the ring it doesn't alter which is
226 * the next to be freed, which is what this is concerned with.
228 void rds_ib_send_cq_comp_handler(struct ib_cq
*cq
, void *context
)
230 struct rds_connection
*conn
= context
;
231 struct rds_ib_connection
*ic
= conn
->c_transport_data
;
232 struct rds_message
*rm
= NULL
;
234 struct rds_ib_send_work
*send
;
240 rdsdebug("cq %p conn %p\n", cq
, conn
);
241 rds_ib_stats_inc(s_ib_tx_cq_call
);
242 ret
= ib_req_notify_cq(cq
, IB_CQ_NEXT_COMP
);
244 rdsdebug("ib_req_notify_cq send failed: %d\n", ret
);
246 while (ib_poll_cq(cq
, 1, &wc
) > 0) {
247 rdsdebug("wc wr_id 0x%llx status %u byte_len %u imm_data %u\n",
248 (unsigned long long)wc
.wr_id
, wc
.status
, wc
.byte_len
,
249 be32_to_cpu(wc
.ex
.imm_data
));
250 rds_ib_stats_inc(s_ib_tx_cq_event
);
252 if (wc
.wr_id
== RDS_IB_ACK_WR_ID
) {
253 if (ic
->i_ack_queued
+ HZ
/2 < jiffies
)
254 rds_ib_stats_inc(s_ib_tx_stalled
);
255 rds_ib_ack_send_complete(ic
);
259 oldest
= rds_ib_ring_oldest(&ic
->i_send_ring
);
261 completed
= rds_ib_ring_completed(&ic
->i_send_ring
, wc
.wr_id
, oldest
);
263 for (i
= 0; i
< completed
; i
++) {
264 send
= &ic
->i_sends
[oldest
];
266 rm
= rds_ib_send_unmap_op(ic
, send
, wc
.status
);
268 if (send
->s_queued
+ HZ
/2 < jiffies
)
269 rds_ib_stats_inc(s_ib_tx_stalled
);
272 if (send
->s_op
== rm
->m_final_op
) {
273 /* If anyone waited for this message to get flushed out, wake
275 rds_message_unmapped(rm
);
281 oldest
= (oldest
+ 1) % ic
->i_send_ring
.w_nr
;
284 rds_ib_ring_free(&ic
->i_send_ring
, completed
);
286 if (test_and_clear_bit(RDS_LL_SEND_FULL
, &conn
->c_flags
) ||
287 test_bit(0, &conn
->c_map_queued
))
288 queue_delayed_work(rds_wq
, &conn
->c_send_w
, 0);
290 /* We expect errors as the qp is drained during shutdown */
291 if (wc
.status
!= IB_WC_SUCCESS
&& rds_conn_up(conn
)) {
292 rds_ib_conn_error(conn
,
293 "send completion on %pI4 "
294 "had status %u, disconnecting and reconnecting\n",
295 &conn
->c_faddr
, wc
.status
);
301 * This is the main function for allocating credits when sending
304 * Conceptually, we have two counters:
305 * - send credits: this tells us how many WRs we're allowed
306 * to submit without overruning the reciever's queue. For
307 * each SEND WR we post, we decrement this by one.
309 * - posted credits: this tells us how many WRs we recently
310 * posted to the receive queue. This value is transferred
311 * to the peer as a "credit update" in a RDS header field.
312 * Every time we transmit credits to the peer, we subtract
313 * the amount of transferred credits from this counter.
315 * It is essential that we avoid situations where both sides have
316 * exhausted their send credits, and are unable to send new credits
317 * to the peer. We achieve this by requiring that we send at least
318 * one credit update to the peer before exhausting our credits.
319 * When new credits arrive, we subtract one credit that is withheld
320 * until we've posted new buffers and are ready to transmit these
321 * credits (see rds_ib_send_add_credits below).
323 * The RDS send code is essentially single-threaded; rds_send_xmit
324 * grabs c_send_lock to ensure exclusive access to the send ring.
325 * However, the ACK sending code is independent and can race with
328 * In the send path, we need to update the counters for send credits
329 * and the counter of posted buffers atomically - when we use the
330 * last available credit, we cannot allow another thread to race us
331 * and grab the posted credits counter. Hence, we have to use a
332 * spinlock to protect the credit counter, or use atomics.
334 * Spinlocks shared between the send and the receive path are bad,
335 * because they create unnecessary delays. An early implementation
336 * using a spinlock showed a 5% degradation in throughput at some
339 * This implementation avoids spinlocks completely, putting both
340 * counters into a single atomic, and updating that atomic using
341 * atomic_add (in the receive path, when receiving fresh credits),
342 * and using atomic_cmpxchg when updating the two counters.
344 int rds_ib_send_grab_credits(struct rds_ib_connection
*ic
,
345 u32 wanted
, u32
*adv_credits
, int need_posted
, int max_posted
)
347 unsigned int avail
, posted
, got
= 0, advertise
;
356 oldval
= newval
= atomic_read(&ic
->i_credits
);
357 posted
= IB_GET_POST_CREDITS(oldval
);
358 avail
= IB_GET_SEND_CREDITS(oldval
);
360 rdsdebug("rds_ib_send_grab_credits(%u): credits=%u posted=%u\n",
361 wanted
, avail
, posted
);
363 /* The last credit must be used to send a credit update. */
364 if (avail
&& !posted
)
367 if (avail
< wanted
) {
368 struct rds_connection
*conn
= ic
->i_cm_id
->context
;
370 /* Oops, there aren't that many credits left! */
371 set_bit(RDS_LL_SEND_FULL
, &conn
->c_flags
);
374 /* Sometimes you get what you want, lalala. */
377 newval
-= IB_SET_SEND_CREDITS(got
);
380 * If need_posted is non-zero, then the caller wants
381 * the posted regardless of whether any send credits are
384 if (posted
&& (got
|| need_posted
)) {
385 advertise
= min_t(unsigned int, posted
, max_posted
);
386 newval
-= IB_SET_POST_CREDITS(advertise
);
389 /* Finally bill everything */
390 if (atomic_cmpxchg(&ic
->i_credits
, oldval
, newval
) != oldval
)
393 *adv_credits
= advertise
;
397 void rds_ib_send_add_credits(struct rds_connection
*conn
, unsigned int credits
)
399 struct rds_ib_connection
*ic
= conn
->c_transport_data
;
404 rdsdebug("rds_ib_send_add_credits(%u): current=%u%s\n",
406 IB_GET_SEND_CREDITS(atomic_read(&ic
->i_credits
)),
407 test_bit(RDS_LL_SEND_FULL
, &conn
->c_flags
) ? ", ll_send_full" : "");
409 atomic_add(IB_SET_SEND_CREDITS(credits
), &ic
->i_credits
);
410 if (test_and_clear_bit(RDS_LL_SEND_FULL
, &conn
->c_flags
))
411 queue_delayed_work(rds_wq
, &conn
->c_send_w
, 0);
413 WARN_ON(IB_GET_SEND_CREDITS(credits
) >= 16384);
415 rds_ib_stats_inc(s_ib_rx_credit_updates
);
418 void rds_ib_advertise_credits(struct rds_connection
*conn
, unsigned int posted
)
420 struct rds_ib_connection
*ic
= conn
->c_transport_data
;
425 atomic_add(IB_SET_POST_CREDITS(posted
), &ic
->i_credits
);
427 /* Decide whether to send an update to the peer now.
428 * If we would send a credit update for every single buffer we
429 * post, we would end up with an ACK storm (ACK arrives,
430 * consumes buffer, we refill the ring, send ACK to remote
431 * advertising the newly posted buffer... ad inf)
433 * Performance pretty much depends on how often we send
434 * credit updates - too frequent updates mean lots of ACKs.
435 * Too infrequent updates, and the peer will run out of
436 * credits and has to throttle.
437 * For the time being, 16 seems to be a good compromise.
439 if (IB_GET_POST_CREDITS(atomic_read(&ic
->i_credits
)) >= 16)
440 set_bit(IB_ACK_REQUESTED
, &ic
->i_ack_flags
);
443 static inline void rds_ib_set_wr_signal_state(struct rds_ib_connection
*ic
,
444 struct rds_ib_send_work
*send
,
448 * We want to delay signaling completions just enough to get
449 * the batching benefits but not so much that we create dead time
452 if (ic
->i_unsignaled_wrs
-- == 0 || notify
) {
453 ic
->i_unsignaled_wrs
= rds_ib_sysctl_max_unsig_wrs
;
454 send
->s_wr
.send_flags
|= IB_SEND_SIGNALED
;
459 * This can be called multiple times for a given message. The first time
460 * we see a message we map its scatterlist into the IB device so that
461 * we can provide that mapped address to the IB scatter gather entries
462 * in the IB work requests. We translate the scatterlist into a series
463 * of work requests that fragment the message. These work requests complete
464 * in order so we pass ownership of the message to the completion handler
465 * once we send the final fragment.
467 * The RDS core uses the c_send_lock to only enter this function once
468 * per connection. This makes sure that the tx ring alloc/unalloc pairs
469 * don't get out of sync and confuse the ring.
471 int rds_ib_xmit(struct rds_connection
*conn
, struct rds_message
*rm
,
472 unsigned int hdr_off
, unsigned int sg
, unsigned int off
)
474 struct rds_ib_connection
*ic
= conn
->c_transport_data
;
475 struct ib_device
*dev
= ic
->i_cm_id
->device
;
476 struct rds_ib_send_work
*send
= NULL
;
477 struct rds_ib_send_work
*first
;
478 struct rds_ib_send_work
*prev
;
479 struct ib_send_wr
*failed_wr
;
480 struct scatterlist
*scat
;
484 u32 credit_alloc
= 0;
490 int flow_controlled
= 0;
492 BUG_ON(off
% RDS_FRAG_SIZE
);
493 BUG_ON(hdr_off
!= 0 && hdr_off
!= sizeof(struct rds_header
));
495 /* Do not send cong updates to IB loopback */
497 && rm
->m_inc
.i_hdr
.h_flags
& RDS_FLAG_CONG_BITMAP
) {
498 rds_cong_map_updated(conn
->c_fcong
, ~(u64
) 0);
499 return sizeof(struct rds_header
) + RDS_CONG_MAP_BYTES
;
502 /* FIXME we may overallocate here */
503 if (be32_to_cpu(rm
->m_inc
.i_hdr
.h_len
) == 0)
506 i
= ceil(be32_to_cpu(rm
->m_inc
.i_hdr
.h_len
), RDS_FRAG_SIZE
);
508 work_alloc
= rds_ib_ring_alloc(&ic
->i_send_ring
, i
, &pos
);
509 if (work_alloc
== 0) {
510 set_bit(RDS_LL_SEND_FULL
, &conn
->c_flags
);
511 rds_ib_stats_inc(s_ib_tx_ring_full
);
517 credit_alloc
= rds_ib_send_grab_credits(ic
, work_alloc
, &posted
, 0, RDS_MAX_ADV_CREDIT
);
518 adv_credits
+= posted
;
519 if (credit_alloc
< work_alloc
) {
520 rds_ib_ring_unalloc(&ic
->i_send_ring
, work_alloc
- credit_alloc
);
521 work_alloc
= credit_alloc
;
524 if (work_alloc
== 0) {
525 set_bit(RDS_LL_SEND_FULL
, &conn
->c_flags
);
526 rds_ib_stats_inc(s_ib_tx_throttle
);
532 /* map the message the first time we see it */
533 if (!ic
->i_data_op
) {
534 if (rm
->data
.op_nents
) {
535 rm
->data
.op_count
= ib_dma_map_sg(dev
,
539 rdsdebug("ic %p mapping rm %p: %d\n", ic
, rm
, rm
->data
.op_count
);
540 if (rm
->data
.op_count
== 0) {
541 rds_ib_stats_inc(s_ib_tx_sg_mapping_failure
);
542 rds_ib_ring_unalloc(&ic
->i_send_ring
, work_alloc
);
543 ret
= -ENOMEM
; /* XXX ? */
547 rm
->data
.op_count
= 0;
550 rds_message_addref(rm
);
551 ic
->i_data_op
= &rm
->data
;
553 /* Finalize the header */
554 if (test_bit(RDS_MSG_ACK_REQUIRED
, &rm
->m_flags
))
555 rm
->m_inc
.i_hdr
.h_flags
|= RDS_FLAG_ACK_REQUIRED
;
556 if (test_bit(RDS_MSG_RETRANSMITTED
, &rm
->m_flags
))
557 rm
->m_inc
.i_hdr
.h_flags
|= RDS_FLAG_RETRANSMITTED
;
559 /* If it has a RDMA op, tell the peer we did it. This is
560 * used by the peer to release use-once RDMA MRs. */
561 if (rm
->rdma
.op_active
) {
562 struct rds_ext_header_rdma ext_hdr
;
564 ext_hdr
.h_rdma_rkey
= cpu_to_be32(rm
->rdma
.op_rkey
);
565 rds_message_add_extension(&rm
->m_inc
.i_hdr
,
566 RDS_EXTHDR_RDMA
, &ext_hdr
, sizeof(ext_hdr
));
568 if (rm
->m_rdma_cookie
) {
569 rds_message_add_rdma_dest_extension(&rm
->m_inc
.i_hdr
,
570 rds_rdma_cookie_key(rm
->m_rdma_cookie
),
571 rds_rdma_cookie_offset(rm
->m_rdma_cookie
));
574 /* Note - rds_ib_piggyb_ack clears the ACK_REQUIRED bit, so
575 * we should not do this unless we have a chance of at least
576 * sticking the header into the send ring. Which is why we
577 * should call rds_ib_ring_alloc first. */
578 rm
->m_inc
.i_hdr
.h_ack
= cpu_to_be64(rds_ib_piggyb_ack(ic
));
579 rds_message_make_checksum(&rm
->m_inc
.i_hdr
);
582 * Update adv_credits since we reset the ACK_REQUIRED bit.
585 rds_ib_send_grab_credits(ic
, 0, &posted
, 1, RDS_MAX_ADV_CREDIT
- adv_credits
);
586 adv_credits
+= posted
;
587 BUG_ON(adv_credits
> 255);
591 /* Sometimes you want to put a fence between an RDMA
592 * READ and the following SEND.
593 * We could either do this all the time
594 * or when requested by the user. Right now, we let
595 * the application choose.
597 if (rm
->rdma
.op_active
&& rm
->rdma
.op_fence
)
598 send_flags
= IB_SEND_FENCE
;
600 /* Each frag gets a header. Msgs may be 0 bytes */
601 send
= &ic
->i_sends
[pos
];
604 scat
= &ic
->i_data_op
->op_sg
[sg
];
607 unsigned int len
= 0;
609 /* Set up the header */
610 send
->s_wr
.send_flags
= send_flags
;
611 send
->s_wr
.opcode
= IB_WR_SEND
;
612 send
->s_wr
.num_sge
= 1;
613 send
->s_wr
.next
= NULL
;
614 send
->s_queued
= jiffies
;
617 send
->s_sge
[0].addr
= ic
->i_send_hdrs_dma
618 + (pos
* sizeof(struct rds_header
));
619 send
->s_sge
[0].length
= sizeof(struct rds_header
);
621 memcpy(&ic
->i_send_hdrs
[pos
], &rm
->m_inc
.i_hdr
, sizeof(struct rds_header
));
623 /* Set up the data, if present */
625 && scat
!= &rm
->data
.op_sg
[rm
->data
.op_count
]) {
626 len
= min(RDS_FRAG_SIZE
, ib_sg_dma_len(dev
, scat
) - off
);
627 send
->s_wr
.num_sge
= 2;
629 send
->s_sge
[1].addr
= ib_sg_dma_address(dev
, scat
) + off
;
630 send
->s_sge
[1].length
= len
;
634 if (off
== ib_sg_dma_len(dev
, scat
)) {
640 rds_ib_set_wr_signal_state(ic
, send
, 0);
643 * Always signal the last one if we're stopping due to flow control.
645 if (ic
->i_flowctl
&& flow_controlled
&& i
== (work_alloc
-1))
646 send
->s_wr
.send_flags
|= IB_SEND_SIGNALED
| IB_SEND_SOLICITED
;
648 rdsdebug("send %p wr %p num_sge %u next %p\n", send
,
649 &send
->s_wr
, send
->s_wr
.num_sge
, send
->s_wr
.next
);
651 if (ic
->i_flowctl
&& adv_credits
) {
652 struct rds_header
*hdr
= &ic
->i_send_hdrs
[pos
];
654 /* add credit and redo the header checksum */
655 hdr
->h_credit
= adv_credits
;
656 rds_message_make_checksum(hdr
);
658 rds_ib_stats_inc(s_ib_tx_credit_updates
);
662 prev
->s_wr
.next
= &send
->s_wr
;
665 pos
= (pos
+ 1) % ic
->i_send_ring
.w_nr
;
666 send
= &ic
->i_sends
[pos
];
669 } while (i
< work_alloc
670 && scat
!= &rm
->data
.op_sg
[rm
->data
.op_count
]);
672 /* Account the RDS header in the number of bytes we sent, but just once.
673 * The caller has no concept of fragmentation. */
675 bytes_sent
+= sizeof(struct rds_header
);
677 /* if we finished the message then send completion owns it */
678 if (scat
== &rm
->data
.op_sg
[rm
->data
.op_count
]) {
679 prev
->s_op
= ic
->i_data_op
;
680 prev
->s_wr
.send_flags
|= IB_SEND_SOLICITED
;
681 ic
->i_data_op
= NULL
;
684 /* Put back wrs & credits we didn't use */
685 if (i
< work_alloc
) {
686 rds_ib_ring_unalloc(&ic
->i_send_ring
, work_alloc
- i
);
689 if (ic
->i_flowctl
&& i
< credit_alloc
)
690 rds_ib_send_add_credits(conn
, credit_alloc
- i
);
692 /* XXX need to worry about failed_wr and partial sends. */
693 failed_wr
= &first
->s_wr
;
694 ret
= ib_post_send(ic
->i_cm_id
->qp
, &first
->s_wr
, &failed_wr
);
695 rdsdebug("ic %p first %p (wr %p) ret %d wr %p\n", ic
,
696 first
, &first
->s_wr
, ret
, failed_wr
);
697 BUG_ON(failed_wr
!= &first
->s_wr
);
699 printk(KERN_WARNING
"RDS/IB: ib_post_send to %pI4 "
700 "returned %d\n", &conn
->c_faddr
, ret
);
701 rds_ib_ring_unalloc(&ic
->i_send_ring
, work_alloc
);
703 ic
->i_data_op
= prev
->s_op
;
707 rds_ib_conn_error(ic
->conn
, "ib_post_send failed\n");
718 * Issue atomic operation.
719 * A simplified version of the rdma case, we always map 1 SG, and
720 * only 8 bytes, for the return value from the atomic operation.
722 int rds_ib_xmit_atomic(struct rds_connection
*conn
, struct rm_atomic_op
*op
)
724 struct rds_ib_connection
*ic
= conn
->c_transport_data
;
725 struct rds_ib_send_work
*send
= NULL
;
726 struct ib_send_wr
*failed_wr
;
727 struct rds_ib_device
*rds_ibdev
;
732 rds_ibdev
= ib_get_client_data(ic
->i_cm_id
->device
, &rds_ib_client
);
734 work_alloc
= rds_ib_ring_alloc(&ic
->i_send_ring
, 1, &pos
);
735 if (work_alloc
!= 1) {
736 rds_ib_ring_unalloc(&ic
->i_send_ring
, work_alloc
);
737 rds_ib_stats_inc(s_ib_tx_ring_full
);
742 /* address of send request in ring */
743 send
= &ic
->i_sends
[pos
];
744 send
->s_queued
= jiffies
;
746 if (op
->op_type
== RDS_ATOMIC_TYPE_CSWP
) {
747 send
->s_wr
.opcode
= IB_WR_ATOMIC_CMP_AND_SWP
;
748 send
->s_wr
.wr
.atomic
.compare_add
= op
->op_compare
;
749 send
->s_wr
.wr
.atomic
.swap
= op
->op_swap_add
;
751 send
->s_wr
.opcode
= IB_WR_ATOMIC_FETCH_AND_ADD
;
752 send
->s_wr
.wr
.atomic
.compare_add
= op
->op_swap_add
;
753 send
->s_wr
.wr
.atomic
.swap
= 0;
755 rds_ib_set_wr_signal_state(ic
, send
, op
->op_notify
);
756 send
->s_wr
.num_sge
= 1;
757 send
->s_wr
.next
= NULL
;
758 send
->s_wr
.wr
.atomic
.remote_addr
= op
->op_remote_addr
;
759 send
->s_wr
.wr
.atomic
.rkey
= op
->op_rkey
;
761 /* map 8 byte retval buffer to the device */
762 ret
= ib_dma_map_sg(ic
->i_cm_id
->device
, op
->op_sg
, 1, DMA_FROM_DEVICE
);
763 rdsdebug("ic %p mapping atomic op %p. mapped %d pg\n", ic
, op
, ret
);
765 rds_ib_ring_unalloc(&ic
->i_send_ring
, work_alloc
);
766 rds_ib_stats_inc(s_ib_tx_sg_mapping_failure
);
767 ret
= -ENOMEM
; /* XXX ? */
771 /* Convert our struct scatterlist to struct ib_sge */
772 send
->s_sge
[0].addr
= ib_sg_dma_address(ic
->i_cm_id
->device
, op
->op_sg
);
773 send
->s_sge
[0].length
= ib_sg_dma_len(ic
->i_cm_id
->device
, op
->op_sg
);
774 send
->s_sge
[0].lkey
= ic
->i_mr
->lkey
;
776 rdsdebug("rva %Lx rpa %Lx len %u\n", op
->op_remote_addr
,
777 send
->s_sge
[0].addr
, send
->s_sge
[0].length
);
779 failed_wr
= &send
->s_wr
;
780 ret
= ib_post_send(ic
->i_cm_id
->qp
, &send
->s_wr
, &failed_wr
);
781 rdsdebug("ic %p send %p (wr %p) ret %d wr %p\n", ic
,
782 send
, &send
->s_wr
, ret
, failed_wr
);
783 BUG_ON(failed_wr
!= &send
->s_wr
);
785 printk(KERN_WARNING
"RDS/IB: atomic ib_post_send to %pI4 "
786 "returned %d\n", &conn
->c_faddr
, ret
);
787 rds_ib_ring_unalloc(&ic
->i_send_ring
, work_alloc
);
791 if (unlikely(failed_wr
!= &send
->s_wr
)) {
792 printk(KERN_WARNING
"RDS/IB: atomic ib_post_send() rc=%d, but failed_wqe updated!\n", ret
);
793 BUG_ON(failed_wr
!= &send
->s_wr
);
800 int rds_ib_xmit_rdma(struct rds_connection
*conn
, struct rm_rdma_op
*op
)
802 struct rds_ib_connection
*ic
= conn
->c_transport_data
;
803 struct rds_ib_send_work
*send
= NULL
;
804 struct rds_ib_send_work
*first
;
805 struct rds_ib_send_work
*prev
;
806 struct ib_send_wr
*failed_wr
;
807 struct rds_ib_device
*rds_ibdev
;
808 struct scatterlist
*scat
;
810 u64 remote_addr
= op
->op_remote_addr
;
819 rds_ibdev
= ib_get_client_data(ic
->i_cm_id
->device
, &rds_ib_client
);
821 /* map the op the first time we see it */
822 if (!op
->op_mapped
) {
823 op
->op_count
= ib_dma_map_sg(ic
->i_cm_id
->device
,
824 op
->op_sg
, op
->op_nents
, (op
->op_write
) ?
825 DMA_TO_DEVICE
: DMA_FROM_DEVICE
);
826 rdsdebug("ic %p mapping op %p: %d\n", ic
, op
, op
->op_count
);
827 if (op
->op_count
== 0) {
828 rds_ib_stats_inc(s_ib_tx_sg_mapping_failure
);
829 ret
= -ENOMEM
; /* XXX ? */
837 * Instead of knowing how to return a partial rdma read/write we insist that there
838 * be enough work requests to send the entire message.
840 i
= ceil(op
->op_count
, rds_ibdev
->max_sge
);
842 work_alloc
= rds_ib_ring_alloc(&ic
->i_send_ring
, i
, &pos
);
843 if (work_alloc
!= i
) {
844 rds_ib_ring_unalloc(&ic
->i_send_ring
, work_alloc
);
845 rds_ib_stats_inc(s_ib_tx_ring_full
);
850 send
= &ic
->i_sends
[pos
];
853 scat
= &op
->op_sg
[0];
855 num_sge
= op
->op_count
;
857 for (i
= 0; i
< work_alloc
&& scat
!= &op
->op_sg
[op
->op_count
]; i
++) {
858 send
->s_wr
.send_flags
= 0;
859 send
->s_queued
= jiffies
;
861 rds_ib_set_wr_signal_state(ic
, send
, op
->op_notify
);
863 send
->s_wr
.opcode
= op
->op_write
? IB_WR_RDMA_WRITE
: IB_WR_RDMA_READ
;
864 send
->s_wr
.wr
.rdma
.remote_addr
= remote_addr
;
865 send
->s_wr
.wr
.rdma
.rkey
= op
->op_rkey
;
868 if (num_sge
> rds_ibdev
->max_sge
) {
869 send
->s_wr
.num_sge
= rds_ibdev
->max_sge
;
870 num_sge
-= rds_ibdev
->max_sge
;
872 send
->s_wr
.num_sge
= num_sge
;
875 send
->s_wr
.next
= NULL
;
878 prev
->s_wr
.next
= &send
->s_wr
;
880 for (j
= 0; j
< send
->s_wr
.num_sge
&& scat
!= &op
->op_sg
[op
->op_count
]; j
++) {
881 len
= ib_sg_dma_len(ic
->i_cm_id
->device
, scat
);
882 send
->s_sge
[j
].addr
=
883 ib_sg_dma_address(ic
->i_cm_id
->device
, scat
);
884 send
->s_sge
[j
].length
= len
;
885 send
->s_sge
[j
].lkey
= ic
->i_mr
->lkey
;
888 rdsdebug("ic %p sent %d remote_addr %llu\n", ic
, sent
, remote_addr
);
894 rdsdebug("send %p wr %p num_sge %u next %p\n", send
,
895 &send
->s_wr
, send
->s_wr
.num_sge
, send
->s_wr
.next
);
898 if (++send
== &ic
->i_sends
[ic
->i_send_ring
.w_nr
])
902 if (i
< work_alloc
) {
903 rds_ib_ring_unalloc(&ic
->i_send_ring
, work_alloc
- i
);
907 failed_wr
= &first
->s_wr
;
908 ret
= ib_post_send(ic
->i_cm_id
->qp
, &first
->s_wr
, &failed_wr
);
909 rdsdebug("ic %p first %p (wr %p) ret %d wr %p\n", ic
,
910 first
, &first
->s_wr
, ret
, failed_wr
);
911 BUG_ON(failed_wr
!= &first
->s_wr
);
913 printk(KERN_WARNING
"RDS/IB: rdma ib_post_send to %pI4 "
914 "returned %d\n", &conn
->c_faddr
, ret
);
915 rds_ib_ring_unalloc(&ic
->i_send_ring
, work_alloc
);
919 if (unlikely(failed_wr
!= &first
->s_wr
)) {
920 printk(KERN_WARNING
"RDS/IB: ib_post_send() rc=%d, but failed_wqe updated!\n", ret
);
921 BUG_ON(failed_wr
!= &first
->s_wr
);
929 void rds_ib_xmit_complete(struct rds_connection
*conn
)
931 struct rds_ib_connection
*ic
= conn
->c_transport_data
;
933 /* We may have a pending ACK or window update we were unable
934 * to send previously (due to flow control). Try again. */
935 rds_ib_attempt_ack(ic
);