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_stats_inc(s_atomic_cswp
);
137 rds_stats_inc(s_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
);
271 if (&send
->s_op
== &rm
->m_final_op
) {
272 /* If anyone waited for this message to get flushed out, wake
274 rds_message_unmapped(rm
);
280 oldest
= (oldest
+ 1) % ic
->i_send_ring
.w_nr
;
283 rds_ib_ring_free(&ic
->i_send_ring
, completed
);
285 if (test_and_clear_bit(RDS_LL_SEND_FULL
, &conn
->c_flags
) ||
286 test_bit(0, &conn
->c_map_queued
))
287 queue_delayed_work(rds_wq
, &conn
->c_send_w
, 0);
289 /* We expect errors as the qp is drained during shutdown */
290 if (wc
.status
!= IB_WC_SUCCESS
&& rds_conn_up(conn
)) {
291 rds_ib_conn_error(conn
,
292 "send completion on %pI4 "
293 "had status %u, disconnecting and reconnecting\n",
294 &conn
->c_faddr
, wc
.status
);
300 * This is the main function for allocating credits when sending
303 * Conceptually, we have two counters:
304 * - send credits: this tells us how many WRs we're allowed
305 * to submit without overruning the reciever's queue. For
306 * each SEND WR we post, we decrement this by one.
308 * - posted credits: this tells us how many WRs we recently
309 * posted to the receive queue. This value is transferred
310 * to the peer as a "credit update" in a RDS header field.
311 * Every time we transmit credits to the peer, we subtract
312 * the amount of transferred credits from this counter.
314 * It is essential that we avoid situations where both sides have
315 * exhausted their send credits, and are unable to send new credits
316 * to the peer. We achieve this by requiring that we send at least
317 * one credit update to the peer before exhausting our credits.
318 * When new credits arrive, we subtract one credit that is withheld
319 * until we've posted new buffers and are ready to transmit these
320 * credits (see rds_ib_send_add_credits below).
322 * The RDS send code is essentially single-threaded; rds_send_xmit
323 * grabs c_send_lock to ensure exclusive access to the send ring.
324 * However, the ACK sending code is independent and can race with
327 * In the send path, we need to update the counters for send credits
328 * and the counter of posted buffers atomically - when we use the
329 * last available credit, we cannot allow another thread to race us
330 * and grab the posted credits counter. Hence, we have to use a
331 * spinlock to protect the credit counter, or use atomics.
333 * Spinlocks shared between the send and the receive path are bad,
334 * because they create unnecessary delays. An early implementation
335 * using a spinlock showed a 5% degradation in throughput at some
338 * This implementation avoids spinlocks completely, putting both
339 * counters into a single atomic, and updating that atomic using
340 * atomic_add (in the receive path, when receiving fresh credits),
341 * and using atomic_cmpxchg when updating the two counters.
343 int rds_ib_send_grab_credits(struct rds_ib_connection
*ic
,
344 u32 wanted
, u32
*adv_credits
, int need_posted
, int max_posted
)
346 unsigned int avail
, posted
, got
= 0, advertise
;
355 oldval
= newval
= atomic_read(&ic
->i_credits
);
356 posted
= IB_GET_POST_CREDITS(oldval
);
357 avail
= IB_GET_SEND_CREDITS(oldval
);
359 rdsdebug("rds_ib_send_grab_credits(%u): credits=%u posted=%u\n",
360 wanted
, avail
, posted
);
362 /* The last credit must be used to send a credit update. */
363 if (avail
&& !posted
)
366 if (avail
< wanted
) {
367 struct rds_connection
*conn
= ic
->i_cm_id
->context
;
369 /* Oops, there aren't that many credits left! */
370 set_bit(RDS_LL_SEND_FULL
, &conn
->c_flags
);
373 /* Sometimes you get what you want, lalala. */
376 newval
-= IB_SET_SEND_CREDITS(got
);
379 * If need_posted is non-zero, then the caller wants
380 * the posted regardless of whether any send credits are
383 if (posted
&& (got
|| need_posted
)) {
384 advertise
= min_t(unsigned int, posted
, max_posted
);
385 newval
-= IB_SET_POST_CREDITS(advertise
);
388 /* Finally bill everything */
389 if (atomic_cmpxchg(&ic
->i_credits
, oldval
, newval
) != oldval
)
392 *adv_credits
= advertise
;
396 void rds_ib_send_add_credits(struct rds_connection
*conn
, unsigned int credits
)
398 struct rds_ib_connection
*ic
= conn
->c_transport_data
;
403 rdsdebug("rds_ib_send_add_credits(%u): current=%u%s\n",
405 IB_GET_SEND_CREDITS(atomic_read(&ic
->i_credits
)),
406 test_bit(RDS_LL_SEND_FULL
, &conn
->c_flags
) ? ", ll_send_full" : "");
408 atomic_add(IB_SET_SEND_CREDITS(credits
), &ic
->i_credits
);
409 if (test_and_clear_bit(RDS_LL_SEND_FULL
, &conn
->c_flags
))
410 queue_delayed_work(rds_wq
, &conn
->c_send_w
, 0);
412 WARN_ON(IB_GET_SEND_CREDITS(credits
) >= 16384);
414 rds_ib_stats_inc(s_ib_rx_credit_updates
);
417 void rds_ib_advertise_credits(struct rds_connection
*conn
, unsigned int posted
)
419 struct rds_ib_connection
*ic
= conn
->c_transport_data
;
424 atomic_add(IB_SET_POST_CREDITS(posted
), &ic
->i_credits
);
426 /* Decide whether to send an update to the peer now.
427 * If we would send a credit update for every single buffer we
428 * post, we would end up with an ACK storm (ACK arrives,
429 * consumes buffer, we refill the ring, send ACK to remote
430 * advertising the newly posted buffer... ad inf)
432 * Performance pretty much depends on how often we send
433 * credit updates - too frequent updates mean lots of ACKs.
434 * Too infrequent updates, and the peer will run out of
435 * credits and has to throttle.
436 * For the time being, 16 seems to be a good compromise.
438 if (IB_GET_POST_CREDITS(atomic_read(&ic
->i_credits
)) >= 16)
439 set_bit(IB_ACK_REQUESTED
, &ic
->i_ack_flags
);
442 static inline void rds_ib_set_wr_signal_state(struct rds_ib_connection
*ic
,
443 struct rds_ib_send_work
*send
,
447 * We want to delay signaling completions just enough to get
448 * the batching benefits but not so much that we create dead time
451 if (ic
->i_unsignaled_wrs
-- == 0 || notify
) {
452 ic
->i_unsignaled_wrs
= rds_ib_sysctl_max_unsig_wrs
;
453 send
->s_wr
.send_flags
|= IB_SEND_SIGNALED
;
458 * This can be called multiple times for a given message. The first time
459 * we see a message we map its scatterlist into the IB device so that
460 * we can provide that mapped address to the IB scatter gather entries
461 * in the IB work requests. We translate the scatterlist into a series
462 * of work requests that fragment the message. These work requests complete
463 * in order so we pass ownership of the message to the completion handler
464 * once we send the final fragment.
466 * The RDS core uses the c_send_lock to only enter this function once
467 * per connection. This makes sure that the tx ring alloc/unalloc pairs
468 * don't get out of sync and confuse the ring.
470 int rds_ib_xmit(struct rds_connection
*conn
, struct rds_message
*rm
,
471 unsigned int hdr_off
, unsigned int sg
, unsigned int off
)
473 struct rds_ib_connection
*ic
= conn
->c_transport_data
;
474 struct ib_device
*dev
= ic
->i_cm_id
->device
;
475 struct rds_ib_send_work
*send
= NULL
;
476 struct rds_ib_send_work
*first
;
477 struct rds_ib_send_work
*prev
;
478 struct ib_send_wr
*failed_wr
;
479 struct scatterlist
*scat
;
483 u32 credit_alloc
= 0;
489 int flow_controlled
= 0;
491 BUG_ON(off
% RDS_FRAG_SIZE
);
492 BUG_ON(hdr_off
!= 0 && hdr_off
!= sizeof(struct rds_header
));
494 /* Do not send cong updates to IB loopback */
496 && rm
->m_inc
.i_hdr
.h_flags
& RDS_FLAG_CONG_BITMAP
) {
497 rds_cong_map_updated(conn
->c_fcong
, ~(u64
) 0);
498 return sizeof(struct rds_header
) + RDS_CONG_MAP_BYTES
;
501 /* FIXME we may overallocate here */
502 if (be32_to_cpu(rm
->m_inc
.i_hdr
.h_len
) == 0)
505 i
= ceil(be32_to_cpu(rm
->m_inc
.i_hdr
.h_len
), RDS_FRAG_SIZE
);
507 work_alloc
= rds_ib_ring_alloc(&ic
->i_send_ring
, i
, &pos
);
508 if (work_alloc
== 0) {
509 set_bit(RDS_LL_SEND_FULL
, &conn
->c_flags
);
510 rds_ib_stats_inc(s_ib_tx_ring_full
);
516 credit_alloc
= rds_ib_send_grab_credits(ic
, work_alloc
, &posted
, 0, RDS_MAX_ADV_CREDIT
);
517 adv_credits
+= posted
;
518 if (credit_alloc
< work_alloc
) {
519 rds_ib_ring_unalloc(&ic
->i_send_ring
, work_alloc
- credit_alloc
);
520 work_alloc
= credit_alloc
;
523 if (work_alloc
== 0) {
524 set_bit(RDS_LL_SEND_FULL
, &conn
->c_flags
);
525 rds_ib_stats_inc(s_ib_tx_throttle
);
531 /* map the message the first time we see it */
532 if (!ic
->i_data_op
) {
533 if (rm
->data
.op_nents
) {
534 rm
->data
.op_count
= ib_dma_map_sg(dev
,
538 rdsdebug("ic %p mapping rm %p: %d\n", ic
, rm
, rm
->data
.op_count
);
539 if (rm
->data
.op_count
== 0) {
540 rds_ib_stats_inc(s_ib_tx_sg_mapping_failure
);
541 rds_ib_ring_unalloc(&ic
->i_send_ring
, work_alloc
);
542 ret
= -ENOMEM
; /* XXX ? */
546 rm
->data
.op_count
= 0;
549 rds_message_addref(rm
);
550 ic
->i_data_op
= &rm
->data
;
552 /* Finalize the header */
553 if (test_bit(RDS_MSG_ACK_REQUIRED
, &rm
->m_flags
))
554 rm
->m_inc
.i_hdr
.h_flags
|= RDS_FLAG_ACK_REQUIRED
;
555 if (test_bit(RDS_MSG_RETRANSMITTED
, &rm
->m_flags
))
556 rm
->m_inc
.i_hdr
.h_flags
|= RDS_FLAG_RETRANSMITTED
;
558 /* If it has a RDMA op, tell the peer we did it. This is
559 * used by the peer to release use-once RDMA MRs. */
560 if (rm
->rdma
.op_active
) {
561 struct rds_ext_header_rdma ext_hdr
;
563 ext_hdr
.h_rdma_rkey
= cpu_to_be32(rm
->rdma
.op_rkey
);
564 rds_message_add_extension(&rm
->m_inc
.i_hdr
,
565 RDS_EXTHDR_RDMA
, &ext_hdr
, sizeof(ext_hdr
));
567 if (rm
->m_rdma_cookie
) {
568 rds_message_add_rdma_dest_extension(&rm
->m_inc
.i_hdr
,
569 rds_rdma_cookie_key(rm
->m_rdma_cookie
),
570 rds_rdma_cookie_offset(rm
->m_rdma_cookie
));
573 /* Note - rds_ib_piggyb_ack clears the ACK_REQUIRED bit, so
574 * we should not do this unless we have a chance of at least
575 * sticking the header into the send ring. Which is why we
576 * should call rds_ib_ring_alloc first. */
577 rm
->m_inc
.i_hdr
.h_ack
= cpu_to_be64(rds_ib_piggyb_ack(ic
));
578 rds_message_make_checksum(&rm
->m_inc
.i_hdr
);
581 * Update adv_credits since we reset the ACK_REQUIRED bit.
584 rds_ib_send_grab_credits(ic
, 0, &posted
, 1, RDS_MAX_ADV_CREDIT
- adv_credits
);
585 adv_credits
+= posted
;
586 BUG_ON(adv_credits
> 255);
590 /* Sometimes you want to put a fence between an RDMA
591 * READ and the following SEND.
592 * We could either do this all the time
593 * or when requested by the user. Right now, we let
594 * the application choose.
596 if (rm
->rdma
.op_active
&& rm
->rdma
.op_fence
)
597 send_flags
= IB_SEND_FENCE
;
599 /* Each frag gets a header. Msgs may be 0 bytes */
600 send
= &ic
->i_sends
[pos
];
603 scat
= &ic
->i_data_op
->op_sg
[sg
];
606 unsigned int len
= 0;
608 /* Set up the header */
609 send
->s_wr
.send_flags
= send_flags
;
610 send
->s_wr
.opcode
= IB_WR_SEND
;
611 send
->s_wr
.num_sge
= 1;
612 send
->s_wr
.next
= NULL
;
613 send
->s_queued
= jiffies
;
616 send
->s_sge
[0].addr
= ic
->i_send_hdrs_dma
617 + (pos
* sizeof(struct rds_header
));
618 send
->s_sge
[0].length
= sizeof(struct rds_header
);
620 memcpy(&ic
->i_send_hdrs
[pos
], &rm
->m_inc
.i_hdr
, sizeof(struct rds_header
));
622 /* Set up the data, if present */
624 && scat
!= &rm
->data
.op_sg
[rm
->data
.op_count
]) {
625 len
= min(RDS_FRAG_SIZE
, ib_sg_dma_len(dev
, scat
) - off
);
626 send
->s_wr
.num_sge
= 2;
628 send
->s_sge
[1].addr
= ib_sg_dma_address(dev
, scat
) + off
;
629 send
->s_sge
[1].length
= len
;
633 if (off
== ib_sg_dma_len(dev
, scat
)) {
639 rds_ib_set_wr_signal_state(ic
, send
, 0);
642 * Always signal the last one if we're stopping due to flow control.
644 if (ic
->i_flowctl
&& flow_controlled
&& i
== (work_alloc
-1))
645 send
->s_wr
.send_flags
|= IB_SEND_SIGNALED
| IB_SEND_SOLICITED
;
647 rdsdebug("send %p wr %p num_sge %u next %p\n", send
,
648 &send
->s_wr
, send
->s_wr
.num_sge
, send
->s_wr
.next
);
650 if (ic
->i_flowctl
&& adv_credits
) {
651 struct rds_header
*hdr
= &ic
->i_send_hdrs
[pos
];
653 /* add credit and redo the header checksum */
654 hdr
->h_credit
= adv_credits
;
655 rds_message_make_checksum(hdr
);
657 rds_ib_stats_inc(s_ib_tx_credit_updates
);
661 prev
->s_wr
.next
= &send
->s_wr
;
664 pos
= (pos
+ 1) % ic
->i_send_ring
.w_nr
;
665 send
= &ic
->i_sends
[pos
];
668 } while (i
< work_alloc
669 && scat
!= &rm
->data
.op_sg
[rm
->data
.op_count
]);
671 /* Account the RDS header in the number of bytes we sent, but just once.
672 * The caller has no concept of fragmentation. */
674 bytes_sent
+= sizeof(struct rds_header
);
676 /* if we finished the message then send completion owns it */
677 if (scat
== &rm
->data
.op_sg
[rm
->data
.op_count
]) {
678 prev
->s_op
= ic
->i_data_op
;
679 prev
->s_wr
.send_flags
|= IB_SEND_SOLICITED
;
680 ic
->i_data_op
= NULL
;
683 /* Put back wrs & credits we didn't use */
684 if (i
< work_alloc
) {
685 rds_ib_ring_unalloc(&ic
->i_send_ring
, work_alloc
- i
);
688 if (ic
->i_flowctl
&& i
< credit_alloc
)
689 rds_ib_send_add_credits(conn
, credit_alloc
- i
);
691 /* XXX need to worry about failed_wr and partial sends. */
692 failed_wr
= &first
->s_wr
;
693 ret
= ib_post_send(ic
->i_cm_id
->qp
, &first
->s_wr
, &failed_wr
);
694 rdsdebug("ic %p first %p (wr %p) ret %d wr %p\n", ic
,
695 first
, &first
->s_wr
, ret
, failed_wr
);
696 BUG_ON(failed_wr
!= &first
->s_wr
);
698 printk(KERN_WARNING
"RDS/IB: ib_post_send to %pI4 "
699 "returned %d\n", &conn
->c_faddr
, ret
);
700 rds_ib_ring_unalloc(&ic
->i_send_ring
, work_alloc
);
702 ic
->i_data_op
= prev
->s_op
;
706 rds_ib_conn_error(ic
->conn
, "ib_post_send failed\n");
717 * Issue atomic operation.
718 * A simplified version of the rdma case, we always map 1 SG, and
719 * only 8 bytes, for the return value from the atomic operation.
721 int rds_ib_xmit_atomic(struct rds_connection
*conn
, struct rm_atomic_op
*op
)
723 struct rds_ib_connection
*ic
= conn
->c_transport_data
;
724 struct rds_ib_send_work
*send
= NULL
;
725 struct ib_send_wr
*failed_wr
;
726 struct rds_ib_device
*rds_ibdev
;
731 rds_ibdev
= ib_get_client_data(ic
->i_cm_id
->device
, &rds_ib_client
);
733 work_alloc
= rds_ib_ring_alloc(&ic
->i_send_ring
, 1, &pos
);
734 if (work_alloc
!= 1) {
735 rds_ib_ring_unalloc(&ic
->i_send_ring
, work_alloc
);
736 rds_ib_stats_inc(s_ib_tx_ring_full
);
741 /* address of send request in ring */
742 send
= &ic
->i_sends
[pos
];
743 send
->s_queued
= jiffies
;
745 if (op
->op_type
== RDS_ATOMIC_TYPE_CSWP
) {
746 send
->s_wr
.opcode
= IB_WR_ATOMIC_CMP_AND_SWP
;
747 send
->s_wr
.wr
.atomic
.compare_add
= op
->op_compare
;
748 send
->s_wr
.wr
.atomic
.swap
= op
->op_swap_add
;
750 send
->s_wr
.opcode
= IB_WR_ATOMIC_FETCH_AND_ADD
;
751 send
->s_wr
.wr
.atomic
.compare_add
= op
->op_swap_add
;
752 send
->s_wr
.wr
.atomic
.swap
= 0;
754 rds_ib_set_wr_signal_state(ic
, send
, op
->op_notify
);
755 send
->s_wr
.num_sge
= 1;
756 send
->s_wr
.next
= NULL
;
757 send
->s_wr
.wr
.atomic
.remote_addr
= op
->op_remote_addr
;
758 send
->s_wr
.wr
.atomic
.rkey
= op
->op_rkey
;
760 /* map 8 byte retval buffer to the device */
761 ret
= ib_dma_map_sg(ic
->i_cm_id
->device
, op
->op_sg
, 1, DMA_FROM_DEVICE
);
762 rdsdebug("ic %p mapping atomic op %p. mapped %d pg\n", ic
, op
, ret
);
764 rds_ib_ring_unalloc(&ic
->i_send_ring
, work_alloc
);
765 rds_ib_stats_inc(s_ib_tx_sg_mapping_failure
);
766 ret
= -ENOMEM
; /* XXX ? */
770 /* Convert our struct scatterlist to struct ib_sge */
771 send
->s_sge
[0].addr
= ib_sg_dma_address(ic
->i_cm_id
->device
, op
->op_sg
);
772 send
->s_sge
[0].length
= ib_sg_dma_len(ic
->i_cm_id
->device
, op
->op_sg
);
773 send
->s_sge
[0].lkey
= ic
->i_mr
->lkey
;
775 rdsdebug("rva %Lx rpa %Lx len %u\n", op
->op_remote_addr
,
776 send
->s_sge
[0].addr
, send
->s_sge
[0].length
);
778 failed_wr
= &send
->s_wr
;
779 ret
= ib_post_send(ic
->i_cm_id
->qp
, &send
->s_wr
, &failed_wr
);
780 rdsdebug("ic %p send %p (wr %p) ret %d wr %p\n", ic
,
781 send
, &send
->s_wr
, ret
, failed_wr
);
782 BUG_ON(failed_wr
!= &send
->s_wr
);
784 printk(KERN_WARNING
"RDS/IB: atomic ib_post_send to %pI4 "
785 "returned %d\n", &conn
->c_faddr
, ret
);
786 rds_ib_ring_unalloc(&ic
->i_send_ring
, work_alloc
);
790 if (unlikely(failed_wr
!= &send
->s_wr
)) {
791 printk(KERN_WARNING
"RDS/IB: atomic ib_post_send() rc=%d, but failed_wqe updated!\n", ret
);
792 BUG_ON(failed_wr
!= &send
->s_wr
);
799 int rds_ib_xmit_rdma(struct rds_connection
*conn
, struct rm_rdma_op
*op
)
801 struct rds_ib_connection
*ic
= conn
->c_transport_data
;
802 struct rds_ib_send_work
*send
= NULL
;
803 struct rds_ib_send_work
*first
;
804 struct rds_ib_send_work
*prev
;
805 struct ib_send_wr
*failed_wr
;
806 struct rds_ib_device
*rds_ibdev
;
807 struct scatterlist
*scat
;
809 u64 remote_addr
= op
->op_remote_addr
;
818 rds_ibdev
= ib_get_client_data(ic
->i_cm_id
->device
, &rds_ib_client
);
820 /* map the op the first time we see it */
821 if (!op
->op_mapped
) {
822 op
->op_count
= ib_dma_map_sg(ic
->i_cm_id
->device
,
823 op
->op_sg
, op
->op_nents
, (op
->op_write
) ?
824 DMA_TO_DEVICE
: DMA_FROM_DEVICE
);
825 rdsdebug("ic %p mapping op %p: %d\n", ic
, op
, op
->op_count
);
826 if (op
->op_count
== 0) {
827 rds_ib_stats_inc(s_ib_tx_sg_mapping_failure
);
828 ret
= -ENOMEM
; /* XXX ? */
836 * Instead of knowing how to return a partial rdma read/write we insist that there
837 * be enough work requests to send the entire message.
839 i
= ceil(op
->op_count
, rds_ibdev
->max_sge
);
841 work_alloc
= rds_ib_ring_alloc(&ic
->i_send_ring
, i
, &pos
);
842 if (work_alloc
!= i
) {
843 rds_ib_ring_unalloc(&ic
->i_send_ring
, work_alloc
);
844 rds_ib_stats_inc(s_ib_tx_ring_full
);
849 send
= &ic
->i_sends
[pos
];
852 scat
= &op
->op_sg
[0];
854 num_sge
= op
->op_count
;
856 for (i
= 0; i
< work_alloc
&& scat
!= &op
->op_sg
[op
->op_count
]; i
++) {
857 send
->s_wr
.send_flags
= 0;
858 send
->s_queued
= jiffies
;
860 rds_ib_set_wr_signal_state(ic
, send
, op
->op_notify
);
862 send
->s_wr
.opcode
= op
->op_write
? IB_WR_RDMA_WRITE
: IB_WR_RDMA_READ
;
863 send
->s_wr
.wr
.rdma
.remote_addr
= remote_addr
;
864 send
->s_wr
.wr
.rdma
.rkey
= op
->op_rkey
;
867 if (num_sge
> rds_ibdev
->max_sge
) {
868 send
->s_wr
.num_sge
= rds_ibdev
->max_sge
;
869 num_sge
-= rds_ibdev
->max_sge
;
871 send
->s_wr
.num_sge
= num_sge
;
874 send
->s_wr
.next
= NULL
;
877 prev
->s_wr
.next
= &send
->s_wr
;
879 for (j
= 0; j
< send
->s_wr
.num_sge
&& scat
!= &op
->op_sg
[op
->op_count
]; j
++) {
880 len
= ib_sg_dma_len(ic
->i_cm_id
->device
, scat
);
881 send
->s_sge
[j
].addr
=
882 ib_sg_dma_address(ic
->i_cm_id
->device
, scat
);
883 send
->s_sge
[j
].length
= len
;
884 send
->s_sge
[j
].lkey
= ic
->i_mr
->lkey
;
887 rdsdebug("ic %p sent %d remote_addr %llu\n", ic
, sent
, remote_addr
);
893 rdsdebug("send %p wr %p num_sge %u next %p\n", send
,
894 &send
->s_wr
, send
->s_wr
.num_sge
, send
->s_wr
.next
);
897 if (++send
== &ic
->i_sends
[ic
->i_send_ring
.w_nr
])
901 if (i
< work_alloc
) {
902 rds_ib_ring_unalloc(&ic
->i_send_ring
, work_alloc
- i
);
906 failed_wr
= &first
->s_wr
;
907 ret
= ib_post_send(ic
->i_cm_id
->qp
, &first
->s_wr
, &failed_wr
);
908 rdsdebug("ic %p first %p (wr %p) ret %d wr %p\n", ic
,
909 first
, &first
->s_wr
, ret
, failed_wr
);
910 BUG_ON(failed_wr
!= &first
->s_wr
);
912 printk(KERN_WARNING
"RDS/IB: rdma ib_post_send to %pI4 "
913 "returned %d\n", &conn
->c_faddr
, ret
);
914 rds_ib_ring_unalloc(&ic
->i_send_ring
, work_alloc
);
918 if (unlikely(failed_wr
!= &first
->s_wr
)) {
919 printk(KERN_WARNING
"RDS/IB: ib_post_send() rc=%d, but failed_wqe updated!\n", ret
);
920 BUG_ON(failed_wr
!= &first
->s_wr
);
928 void rds_ib_xmit_complete(struct rds_connection
*conn
)
930 struct rds_ib_connection
*ic
= conn
->c_transport_data
;
932 /* We may have a pending ACK or window update we were unable
933 * to send previously (due to flow control). Try again. */
934 rds_ib_attempt_ack(ic
);