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>
34 #include <linux/slab.h>
35 #include <linux/pci.h>
36 #include <linux/dma-mapping.h>
37 #include <rdma/rdma_cm.h>
42 static struct kmem_cache
*rds_ib_incoming_slab
;
43 static struct kmem_cache
*rds_ib_frag_slab
;
44 static atomic_t rds_ib_allocation
= ATOMIC_INIT(0);
46 static void rds_ib_frag_drop_page(struct rds_page_frag
*frag
)
48 rdsdebug("frag %p page %p\n", frag
, frag
->f_page
);
49 __free_page(frag
->f_page
);
53 static void rds_ib_frag_free(struct rds_page_frag
*frag
)
55 rdsdebug("frag %p page %p\n", frag
, frag
->f_page
);
56 BUG_ON(frag
->f_page
!= NULL
);
57 kmem_cache_free(rds_ib_frag_slab
, frag
);
61 * We map a page at a time. Its fragments are posted in order. This
62 * is called in fragment order as the fragments get send completion events.
63 * Only the last frag in the page performs the unmapping.
65 * It's OK for ring cleanup to call this in whatever order it likes because
66 * DMA is not in flight and so we can unmap while other ring entries still
67 * hold page references in their frags.
69 static void rds_ib_recv_unmap_page(struct rds_ib_connection
*ic
,
70 struct rds_ib_recv_work
*recv
)
72 struct rds_page_frag
*frag
= recv
->r_frag
;
74 rdsdebug("recv %p frag %p page %p\n", recv
, frag
, frag
->f_page
);
76 ib_dma_unmap_page(ic
->i_cm_id
->device
,
78 RDS_FRAG_SIZE
, DMA_FROM_DEVICE
);
82 void rds_ib_recv_init_ring(struct rds_ib_connection
*ic
)
84 struct rds_ib_recv_work
*recv
;
87 for (i
= 0, recv
= ic
->i_recvs
; i
< ic
->i_recv_ring
.w_nr
; i
++, recv
++) {
93 recv
->r_wr
.next
= NULL
;
95 recv
->r_wr
.sg_list
= recv
->r_sge
;
96 recv
->r_wr
.num_sge
= RDS_IB_RECV_SGE
;
98 sge
= rds_ib_data_sge(ic
, recv
->r_sge
);
100 sge
->length
= RDS_FRAG_SIZE
;
101 sge
->lkey
= ic
->i_mr
->lkey
;
103 sge
= rds_ib_header_sge(ic
, recv
->r_sge
);
104 sge
->addr
= ic
->i_recv_hdrs_dma
+ (i
* sizeof(struct rds_header
));
105 sge
->length
= sizeof(struct rds_header
);
106 sge
->lkey
= ic
->i_mr
->lkey
;
110 static void rds_ib_recv_clear_one(struct rds_ib_connection
*ic
,
111 struct rds_ib_recv_work
*recv
)
114 rds_inc_put(&recv
->r_ibinc
->ii_inc
);
115 recv
->r_ibinc
= NULL
;
118 rds_ib_recv_unmap_page(ic
, recv
);
119 if (recv
->r_frag
->f_page
)
120 rds_ib_frag_drop_page(recv
->r_frag
);
121 rds_ib_frag_free(recv
->r_frag
);
126 void rds_ib_recv_clear_ring(struct rds_ib_connection
*ic
)
130 for (i
= 0; i
< ic
->i_recv_ring
.w_nr
; i
++)
131 rds_ib_recv_clear_one(ic
, &ic
->i_recvs
[i
]);
133 if (ic
->i_frag
.f_page
)
134 rds_ib_frag_drop_page(&ic
->i_frag
);
137 static int rds_ib_recv_refill_one(struct rds_connection
*conn
,
138 struct rds_ib_recv_work
*recv
,
139 gfp_t kptr_gfp
, gfp_t page_gfp
)
141 struct rds_ib_connection
*ic
= conn
->c_transport_data
;
146 if (recv
->r_ibinc
== NULL
) {
147 if (!atomic_add_unless(&rds_ib_allocation
, 1, rds_ib_sysctl_max_recv_allocation
)) {
148 rds_ib_stats_inc(s_ib_rx_alloc_limit
);
151 recv
->r_ibinc
= kmem_cache_alloc(rds_ib_incoming_slab
,
153 if (recv
->r_ibinc
== NULL
) {
154 atomic_dec(&rds_ib_allocation
);
157 INIT_LIST_HEAD(&recv
->r_ibinc
->ii_frags
);
158 rds_inc_init(&recv
->r_ibinc
->ii_inc
, conn
, conn
->c_faddr
);
161 if (recv
->r_frag
== NULL
) {
162 recv
->r_frag
= kmem_cache_alloc(rds_ib_frag_slab
, kptr_gfp
);
163 if (recv
->r_frag
== NULL
)
165 INIT_LIST_HEAD(&recv
->r_frag
->f_item
);
166 recv
->r_frag
->f_page
= NULL
;
169 if (ic
->i_frag
.f_page
== NULL
) {
170 ic
->i_frag
.f_page
= alloc_page(page_gfp
);
171 if (ic
->i_frag
.f_page
== NULL
)
173 ic
->i_frag
.f_offset
= 0;
176 dma_addr
= ib_dma_map_page(ic
->i_cm_id
->device
,
181 if (ib_dma_mapping_error(ic
->i_cm_id
->device
, dma_addr
))
185 * Once we get the RDS_PAGE_LAST_OFF frag then rds_ib_frag_unmap()
186 * must be called on this recv. This happens as completions hit
187 * in order or on connection shutdown.
189 recv
->r_frag
->f_page
= ic
->i_frag
.f_page
;
190 recv
->r_frag
->f_offset
= ic
->i_frag
.f_offset
;
191 recv
->r_frag
->f_mapped
= dma_addr
;
193 sge
= rds_ib_data_sge(ic
, recv
->r_sge
);
194 sge
->addr
= dma_addr
;
195 sge
->length
= RDS_FRAG_SIZE
;
197 sge
= rds_ib_header_sge(ic
, recv
->r_sge
);
198 sge
->addr
= ic
->i_recv_hdrs_dma
+ (recv
- ic
->i_recvs
) * sizeof(struct rds_header
);
199 sge
->length
= sizeof(struct rds_header
);
201 get_page(recv
->r_frag
->f_page
);
203 if (ic
->i_frag
.f_offset
< RDS_PAGE_LAST_OFF
) {
204 ic
->i_frag
.f_offset
+= RDS_FRAG_SIZE
;
206 put_page(ic
->i_frag
.f_page
);
207 ic
->i_frag
.f_page
= NULL
;
208 ic
->i_frag
.f_offset
= 0;
217 * This tries to allocate and post unused work requests after making sure that
218 * they have all the allocations they need to queue received fragments into
219 * sockets. The i_recv_mutex is held here so that ring_alloc and _unalloc
220 * pairs don't go unmatched.
222 * -1 is returned if posting fails due to temporary resource exhaustion.
224 int rds_ib_recv_refill(struct rds_connection
*conn
, gfp_t kptr_gfp
,
225 gfp_t page_gfp
, int prefill
)
227 struct rds_ib_connection
*ic
= conn
->c_transport_data
;
228 struct rds_ib_recv_work
*recv
;
229 struct ib_recv_wr
*failed_wr
;
230 unsigned int posted
= 0;
234 while ((prefill
|| rds_conn_up(conn
)) &&
235 rds_ib_ring_alloc(&ic
->i_recv_ring
, 1, &pos
)) {
236 if (pos
>= ic
->i_recv_ring
.w_nr
) {
237 printk(KERN_NOTICE
"Argh - ring alloc returned pos=%u\n",
243 recv
= &ic
->i_recvs
[pos
];
244 ret
= rds_ib_recv_refill_one(conn
, recv
, kptr_gfp
, page_gfp
);
250 ret
= ib_post_recv(ic
->i_cm_id
->qp
, &recv
->r_wr
, &failed_wr
);
251 rdsdebug("recv %p ibinc %p page %p addr %lu ret %d\n", recv
,
252 recv
->r_ibinc
, recv
->r_frag
->f_page
,
253 (long) recv
->r_frag
->f_mapped
, ret
);
255 rds_ib_conn_error(conn
, "recv post on "
256 "%pI4 returned %d, disconnecting and "
257 "reconnecting\n", &conn
->c_faddr
,
266 /* We're doing flow control - update the window. */
267 if (ic
->i_flowctl
&& posted
)
268 rds_ib_advertise_credits(conn
, posted
);
271 rds_ib_ring_unalloc(&ic
->i_recv_ring
, 1);
275 void rds_ib_inc_purge(struct rds_incoming
*inc
)
277 struct rds_ib_incoming
*ibinc
;
278 struct rds_page_frag
*frag
;
279 struct rds_page_frag
*pos
;
281 ibinc
= container_of(inc
, struct rds_ib_incoming
, ii_inc
);
282 rdsdebug("purging ibinc %p inc %p\n", ibinc
, inc
);
284 list_for_each_entry_safe(frag
, pos
, &ibinc
->ii_frags
, f_item
) {
285 list_del_init(&frag
->f_item
);
286 rds_ib_frag_drop_page(frag
);
287 rds_ib_frag_free(frag
);
291 void rds_ib_inc_free(struct rds_incoming
*inc
)
293 struct rds_ib_incoming
*ibinc
;
295 ibinc
= container_of(inc
, struct rds_ib_incoming
, ii_inc
);
297 rds_ib_inc_purge(inc
);
298 rdsdebug("freeing ibinc %p inc %p\n", ibinc
, inc
);
299 BUG_ON(!list_empty(&ibinc
->ii_frags
));
300 kmem_cache_free(rds_ib_incoming_slab
, ibinc
);
301 atomic_dec(&rds_ib_allocation
);
302 BUG_ON(atomic_read(&rds_ib_allocation
) < 0);
305 int rds_ib_inc_copy_to_user(struct rds_incoming
*inc
, struct iovec
*first_iov
,
308 struct rds_ib_incoming
*ibinc
;
309 struct rds_page_frag
*frag
;
310 struct iovec
*iov
= first_iov
;
311 unsigned long to_copy
;
312 unsigned long frag_off
= 0;
313 unsigned long iov_off
= 0;
318 ibinc
= container_of(inc
, struct rds_ib_incoming
, ii_inc
);
319 frag
= list_entry(ibinc
->ii_frags
.next
, struct rds_page_frag
, f_item
);
320 len
= be32_to_cpu(inc
->i_hdr
.h_len
);
322 while (copied
< size
&& copied
< len
) {
323 if (frag_off
== RDS_FRAG_SIZE
) {
324 frag
= list_entry(frag
->f_item
.next
,
325 struct rds_page_frag
, f_item
);
328 while (iov_off
== iov
->iov_len
) {
333 to_copy
= min(iov
->iov_len
- iov_off
, RDS_FRAG_SIZE
- frag_off
);
334 to_copy
= min_t(size_t, to_copy
, size
- copied
);
335 to_copy
= min_t(unsigned long, to_copy
, len
- copied
);
337 rdsdebug("%lu bytes to user [%p, %zu] + %lu from frag "
339 to_copy
, iov
->iov_base
, iov
->iov_len
, iov_off
,
340 frag
->f_page
, frag
->f_offset
, frag_off
);
342 ret
= rds_page_copy_to_user(frag
->f_page
,
343 frag
->f_offset
+ frag_off
,
344 iov
->iov_base
+ iov_off
,
359 /* ic starts out kzalloc()ed */
360 void rds_ib_recv_init_ack(struct rds_ib_connection
*ic
)
362 struct ib_send_wr
*wr
= &ic
->i_ack_wr
;
363 struct ib_sge
*sge
= &ic
->i_ack_sge
;
365 sge
->addr
= ic
->i_ack_dma
;
366 sge
->length
= sizeof(struct rds_header
);
367 sge
->lkey
= ic
->i_mr
->lkey
;
371 wr
->opcode
= IB_WR_SEND
;
372 wr
->wr_id
= RDS_IB_ACK_WR_ID
;
373 wr
->send_flags
= IB_SEND_SIGNALED
| IB_SEND_SOLICITED
;
377 * You'd think that with reliable IB connections you wouldn't need to ack
378 * messages that have been received. The problem is that IB hardware generates
379 * an ack message before it has DMAed the message into memory. This creates a
380 * potential message loss if the HCA is disabled for any reason between when it
381 * sends the ack and before the message is DMAed and processed. This is only a
382 * potential issue if another HCA is available for fail-over.
384 * When the remote host receives our ack they'll free the sent message from
385 * their send queue. To decrease the latency of this we always send an ack
386 * immediately after we've received messages.
388 * For simplicity, we only have one ack in flight at a time. This puts
389 * pressure on senders to have deep enough send queues to absorb the latency of
390 * a single ack frame being in flight. This might not be good enough.
392 * This is implemented by have a long-lived send_wr and sge which point to a
393 * statically allocated ack frame. This ack wr does not fall under the ring
394 * accounting that the tx and rx wrs do. The QP attribute specifically makes
395 * room for it beyond the ring size. Send completion notices its special
396 * wr_id and avoids working with the ring in that case.
398 #ifndef KERNEL_HAS_ATOMIC64
399 static void rds_ib_set_ack(struct rds_ib_connection
*ic
, u64 seq
,
404 spin_lock_irqsave(&ic
->i_ack_lock
, flags
);
405 ic
->i_ack_next
= seq
;
407 set_bit(IB_ACK_REQUESTED
, &ic
->i_ack_flags
);
408 spin_unlock_irqrestore(&ic
->i_ack_lock
, flags
);
411 static u64
rds_ib_get_ack(struct rds_ib_connection
*ic
)
416 clear_bit(IB_ACK_REQUESTED
, &ic
->i_ack_flags
);
418 spin_lock_irqsave(&ic
->i_ack_lock
, flags
);
419 seq
= ic
->i_ack_next
;
420 spin_unlock_irqrestore(&ic
->i_ack_lock
, flags
);
425 static void rds_ib_set_ack(struct rds_ib_connection
*ic
, u64 seq
,
428 atomic64_set(&ic
->i_ack_next
, seq
);
430 smp_mb__before_clear_bit();
431 set_bit(IB_ACK_REQUESTED
, &ic
->i_ack_flags
);
435 static u64
rds_ib_get_ack(struct rds_ib_connection
*ic
)
437 clear_bit(IB_ACK_REQUESTED
, &ic
->i_ack_flags
);
438 smp_mb__after_clear_bit();
440 return atomic64_read(&ic
->i_ack_next
);
445 static void rds_ib_send_ack(struct rds_ib_connection
*ic
, unsigned int adv_credits
)
447 struct rds_header
*hdr
= ic
->i_ack
;
448 struct ib_send_wr
*failed_wr
;
452 seq
= rds_ib_get_ack(ic
);
454 rdsdebug("send_ack: ic %p ack %llu\n", ic
, (unsigned long long) seq
);
455 rds_message_populate_header(hdr
, 0, 0, 0);
456 hdr
->h_ack
= cpu_to_be64(seq
);
457 hdr
->h_credit
= adv_credits
;
458 rds_message_make_checksum(hdr
);
459 ic
->i_ack_queued
= jiffies
;
461 ret
= ib_post_send(ic
->i_cm_id
->qp
, &ic
->i_ack_wr
, &failed_wr
);
463 /* Failed to send. Release the WR, and
466 clear_bit(IB_ACK_IN_FLIGHT
, &ic
->i_ack_flags
);
467 set_bit(IB_ACK_REQUESTED
, &ic
->i_ack_flags
);
469 rds_ib_stats_inc(s_ib_ack_send_failure
);
471 rds_ib_conn_error(ic
->conn
, "sending ack failed\n");
473 rds_ib_stats_inc(s_ib_ack_sent
);
477 * There are 3 ways of getting acknowledgements to the peer:
478 * 1. We call rds_ib_attempt_ack from the recv completion handler
479 * to send an ACK-only frame.
480 * However, there can be only one such frame in the send queue
481 * at any time, so we may have to postpone it.
482 * 2. When another (data) packet is transmitted while there's
483 * an ACK in the queue, we piggyback the ACK sequence number
484 * on the data packet.
485 * 3. If the ACK WR is done sending, we get called from the
486 * send queue completion handler, and check whether there's
487 * another ACK pending (postponed because the WR was on the
488 * queue). If so, we transmit it.
490 * We maintain 2 variables:
491 * - i_ack_flags, which keeps track of whether the ACK WR
492 * is currently in the send queue or not (IB_ACK_IN_FLIGHT)
493 * - i_ack_next, which is the last sequence number we received
495 * Potentially, send queue and receive queue handlers can run concurrently.
496 * It would be nice to not have to use a spinlock to synchronize things,
497 * but the one problem that rules this out is that 64bit updates are
498 * not atomic on all platforms. Things would be a lot simpler if
499 * we had atomic64 or maybe cmpxchg64 everywhere.
501 * Reconnecting complicates this picture just slightly. When we
502 * reconnect, we may be seeing duplicate packets. The peer
503 * is retransmitting them, because it hasn't seen an ACK for
504 * them. It is important that we ACK these.
506 * ACK mitigation adds a header flag "ACK_REQUIRED"; any packet with
507 * this flag set *MUST* be acknowledged immediately.
511 * When we get here, we're called from the recv queue handler.
512 * Check whether we ought to transmit an ACK.
514 void rds_ib_attempt_ack(struct rds_ib_connection
*ic
)
516 unsigned int adv_credits
;
518 if (!test_bit(IB_ACK_REQUESTED
, &ic
->i_ack_flags
))
521 if (test_and_set_bit(IB_ACK_IN_FLIGHT
, &ic
->i_ack_flags
)) {
522 rds_ib_stats_inc(s_ib_ack_send_delayed
);
526 /* Can we get a send credit? */
527 if (!rds_ib_send_grab_credits(ic
, 1, &adv_credits
, 0, RDS_MAX_ADV_CREDIT
)) {
528 rds_ib_stats_inc(s_ib_tx_throttle
);
529 clear_bit(IB_ACK_IN_FLIGHT
, &ic
->i_ack_flags
);
533 clear_bit(IB_ACK_REQUESTED
, &ic
->i_ack_flags
);
534 rds_ib_send_ack(ic
, adv_credits
);
538 * We get here from the send completion handler, when the
539 * adapter tells us the ACK frame was sent.
541 void rds_ib_ack_send_complete(struct rds_ib_connection
*ic
)
543 clear_bit(IB_ACK_IN_FLIGHT
, &ic
->i_ack_flags
);
544 rds_ib_attempt_ack(ic
);
548 * This is called by the regular xmit code when it wants to piggyback
549 * an ACK on an outgoing frame.
551 u64
rds_ib_piggyb_ack(struct rds_ib_connection
*ic
)
553 if (test_and_clear_bit(IB_ACK_REQUESTED
, &ic
->i_ack_flags
))
554 rds_ib_stats_inc(s_ib_ack_send_piggybacked
);
555 return rds_ib_get_ack(ic
);
558 static struct rds_header
*rds_ib_get_header(struct rds_connection
*conn
,
559 struct rds_ib_recv_work
*recv
,
562 struct rds_ib_connection
*ic
= conn
->c_transport_data
;
563 void *hdr_buff
= &ic
->i_recv_hdrs
[recv
- ic
->i_recvs
];
565 u32 misplaced_hdr_bytes
;
568 * Support header at the front (RDS 3.1+) as well as header-at-end.
571 * 1) header all in header buff (great!)
572 * 2) header all in data page (copy all to header buff)
573 * 3) header split across hdr buf + data page
574 * (move bit in hdr buff to end before copying other bit from data page)
576 if (conn
->c_version
> RDS_PROTOCOL_3_0
|| data_len
== RDS_FRAG_SIZE
)
579 if (data_len
<= (RDS_FRAG_SIZE
- sizeof(struct rds_header
))) {
580 addr
= kmap_atomic(recv
->r_frag
->f_page
, KM_SOFTIRQ0
);
582 addr
+ recv
->r_frag
->f_offset
+ data_len
,
583 sizeof(struct rds_header
));
584 kunmap_atomic(addr
, KM_SOFTIRQ0
);
588 misplaced_hdr_bytes
= (sizeof(struct rds_header
) - (RDS_FRAG_SIZE
- data_len
));
590 memmove(hdr_buff
+ misplaced_hdr_bytes
, hdr_buff
, misplaced_hdr_bytes
);
592 addr
= kmap_atomic(recv
->r_frag
->f_page
, KM_SOFTIRQ0
);
593 memcpy(hdr_buff
, addr
+ recv
->r_frag
->f_offset
+ data_len
,
594 sizeof(struct rds_header
) - misplaced_hdr_bytes
);
595 kunmap_atomic(addr
, KM_SOFTIRQ0
);
600 * It's kind of lame that we're copying from the posted receive pages into
601 * long-lived bitmaps. We could have posted the bitmaps and rdma written into
602 * them. But receiving new congestion bitmaps should be a *rare* event, so
603 * hopefully we won't need to invest that complexity in making it more
604 * efficient. By copying we can share a simpler core with TCP which has to
607 static void rds_ib_cong_recv(struct rds_connection
*conn
,
608 struct rds_ib_incoming
*ibinc
)
610 struct rds_cong_map
*map
;
611 unsigned int map_off
;
612 unsigned int map_page
;
613 struct rds_page_frag
*frag
;
614 unsigned long frag_off
;
615 unsigned long to_copy
;
616 unsigned long copied
;
617 uint64_t uncongested
= 0;
620 /* catch completely corrupt packets */
621 if (be32_to_cpu(ibinc
->ii_inc
.i_hdr
.h_len
) != RDS_CONG_MAP_BYTES
)
628 frag
= list_entry(ibinc
->ii_frags
.next
, struct rds_page_frag
, f_item
);
633 while (copied
< RDS_CONG_MAP_BYTES
) {
637 to_copy
= min(RDS_FRAG_SIZE
- frag_off
, PAGE_SIZE
- map_off
);
638 BUG_ON(to_copy
& 7); /* Must be 64bit aligned. */
640 addr
= kmap_atomic(frag
->f_page
, KM_SOFTIRQ0
);
642 src
= addr
+ frag_off
;
643 dst
= (void *)map
->m_page_addrs
[map_page
] + map_off
;
644 for (k
= 0; k
< to_copy
; k
+= 8) {
645 /* Record ports that became uncongested, ie
646 * bits that changed from 0 to 1. */
647 uncongested
|= ~(*src
) & *dst
;
650 kunmap_atomic(addr
, KM_SOFTIRQ0
);
655 if (map_off
== PAGE_SIZE
) {
661 if (frag_off
== RDS_FRAG_SIZE
) {
662 frag
= list_entry(frag
->f_item
.next
,
663 struct rds_page_frag
, f_item
);
668 /* the congestion map is in little endian order */
669 uncongested
= le64_to_cpu(uncongested
);
671 rds_cong_map_updated(map
, uncongested
);
675 * Rings are posted with all the allocations they'll need to queue the
676 * incoming message to the receiving socket so this can't fail.
677 * All fragments start with a header, so we can make sure we're not receiving
678 * garbage, and we can tell a small 8 byte fragment from an ACK frame.
680 struct rds_ib_ack_state
{
683 unsigned int ack_required
:1;
684 unsigned int ack_next_valid
:1;
685 unsigned int ack_recv_valid
:1;
688 static void rds_ib_process_recv(struct rds_connection
*conn
,
689 struct rds_ib_recv_work
*recv
, u32 data_len
,
690 struct rds_ib_ack_state
*state
)
692 struct rds_ib_connection
*ic
= conn
->c_transport_data
;
693 struct rds_ib_incoming
*ibinc
= ic
->i_ibinc
;
694 struct rds_header
*ihdr
, *hdr
;
697 rdsdebug("ic %p ibinc %p recv %p byte len %u\n", ic
, ibinc
, recv
,
700 if (data_len
< sizeof(struct rds_header
)) {
701 rds_ib_conn_error(conn
, "incoming message "
702 "from %pI4 didn't inclue a "
703 "header, disconnecting and "
708 data_len
-= sizeof(struct rds_header
);
710 ihdr
= rds_ib_get_header(conn
, recv
, data_len
);
712 /* Validate the checksum. */
713 if (!rds_message_verify_checksum(ihdr
)) {
714 rds_ib_conn_error(conn
, "incoming message "
715 "from %pI4 has corrupted header - "
716 "forcing a reconnect\n",
718 rds_stats_inc(s_recv_drop_bad_checksum
);
722 /* Process the ACK sequence which comes with every packet */
723 state
->ack_recv
= be64_to_cpu(ihdr
->h_ack
);
724 state
->ack_recv_valid
= 1;
726 /* Process the credits update if there was one */
728 rds_ib_send_add_credits(conn
, ihdr
->h_credit
);
730 if (ihdr
->h_sport
== 0 && ihdr
->h_dport
== 0 && data_len
== 0) {
731 /* This is an ACK-only packet. The fact that it gets
732 * special treatment here is that historically, ACKs
733 * were rather special beasts.
735 rds_ib_stats_inc(s_ib_ack_received
);
737 rds_ib_frag_drop_page(recv
->r_frag
);
742 * If we don't already have an inc on the connection then this
743 * fragment has a header and starts a message.. copy its header
744 * into the inc and save the inc so we can hang upcoming fragments
748 ibinc
= recv
->r_ibinc
;
749 recv
->r_ibinc
= NULL
;
752 hdr
= &ibinc
->ii_inc
.i_hdr
;
753 memcpy(hdr
, ihdr
, sizeof(*hdr
));
754 ic
->i_recv_data_rem
= be32_to_cpu(hdr
->h_len
);
756 rdsdebug("ic %p ibinc %p rem %u flag 0x%x\n", ic
, ibinc
,
757 ic
->i_recv_data_rem
, hdr
->h_flags
);
759 hdr
= &ibinc
->ii_inc
.i_hdr
;
760 /* We can't just use memcmp here; fragments of a
761 * single message may carry different ACKs */
762 if (hdr
->h_sequence
!= ihdr
->h_sequence
||
763 hdr
->h_len
!= ihdr
->h_len
||
764 hdr
->h_sport
!= ihdr
->h_sport
||
765 hdr
->h_dport
!= ihdr
->h_dport
) {
766 rds_ib_conn_error(conn
,
767 "fragment header mismatch; forcing reconnect\n");
772 list_add_tail(&recv
->r_frag
->f_item
, &ibinc
->ii_frags
);
775 if (ic
->i_recv_data_rem
> RDS_FRAG_SIZE
)
776 ic
->i_recv_data_rem
-= RDS_FRAG_SIZE
;
778 ic
->i_recv_data_rem
= 0;
781 if (ibinc
->ii_inc
.i_hdr
.h_flags
== RDS_FLAG_CONG_BITMAP
)
782 rds_ib_cong_recv(conn
, ibinc
);
784 rds_recv_incoming(conn
, conn
->c_faddr
, conn
->c_laddr
,
785 &ibinc
->ii_inc
, GFP_ATOMIC
,
787 state
->ack_next
= be64_to_cpu(hdr
->h_sequence
);
788 state
->ack_next_valid
= 1;
791 /* Evaluate the ACK_REQUIRED flag *after* we received
792 * the complete frame, and after bumping the next_rx
794 if (hdr
->h_flags
& RDS_FLAG_ACK_REQUIRED
) {
795 rds_stats_inc(s_recv_ack_required
);
796 state
->ack_required
= 1;
799 rds_inc_put(&ibinc
->ii_inc
);
804 * Plucking the oldest entry from the ring can be done concurrently with
805 * the thread refilling the ring. Each ring operation is protected by
806 * spinlocks and the transient state of refilling doesn't change the
807 * recording of which entry is oldest.
809 * This relies on IB only calling one cq comp_handler for each cq so that
810 * there will only be one caller of rds_recv_incoming() per RDS connection.
812 void rds_ib_recv_cq_comp_handler(struct ib_cq
*cq
, void *context
)
814 struct rds_connection
*conn
= context
;
815 struct rds_ib_connection
*ic
= conn
->c_transport_data
;
817 rdsdebug("conn %p cq %p\n", conn
, cq
);
819 rds_ib_stats_inc(s_ib_rx_cq_call
);
821 tasklet_schedule(&ic
->i_recv_tasklet
);
824 static inline void rds_poll_cq(struct rds_ib_connection
*ic
,
825 struct rds_ib_ack_state
*state
)
827 struct rds_connection
*conn
= ic
->conn
;
829 struct rds_ib_recv_work
*recv
;
831 while (ib_poll_cq(ic
->i_recv_cq
, 1, &wc
) > 0) {
832 rdsdebug("wc wr_id 0x%llx status %u byte_len %u imm_data %u\n",
833 (unsigned long long)wc
.wr_id
, wc
.status
, wc
.byte_len
,
834 be32_to_cpu(wc
.ex
.imm_data
));
835 rds_ib_stats_inc(s_ib_rx_cq_event
);
837 recv
= &ic
->i_recvs
[rds_ib_ring_oldest(&ic
->i_recv_ring
)];
839 rds_ib_recv_unmap_page(ic
, recv
);
842 * Also process recvs in connecting state because it is possible
843 * to get a recv completion _before_ the rdmacm ESTABLISHED
844 * event is processed.
846 if (rds_conn_up(conn
) || rds_conn_connecting(conn
)) {
847 /* We expect errors as the qp is drained during shutdown */
848 if (wc
.status
== IB_WC_SUCCESS
) {
849 rds_ib_process_recv(conn
, recv
, wc
.byte_len
, state
);
851 rds_ib_conn_error(conn
, "recv completion on "
852 "%pI4 had status %u, disconnecting and "
853 "reconnecting\n", &conn
->c_faddr
,
858 rds_ib_ring_free(&ic
->i_recv_ring
, 1);
862 void rds_ib_recv_tasklet_fn(unsigned long data
)
864 struct rds_ib_connection
*ic
= (struct rds_ib_connection
*) data
;
865 struct rds_connection
*conn
= ic
->conn
;
866 struct rds_ib_ack_state state
= { 0, };
868 rds_poll_cq(ic
, &state
);
869 ib_req_notify_cq(ic
->i_recv_cq
, IB_CQ_SOLICITED
);
870 rds_poll_cq(ic
, &state
);
872 if (state
.ack_next_valid
)
873 rds_ib_set_ack(ic
, state
.ack_next
, state
.ack_required
);
874 if (state
.ack_recv_valid
&& state
.ack_recv
> ic
->i_ack_recv
) {
875 rds_send_drop_acked(conn
, state
.ack_recv
, NULL
);
876 ic
->i_ack_recv
= state
.ack_recv
;
878 if (rds_conn_up(conn
))
879 rds_ib_attempt_ack(ic
);
881 /* If we ever end up with a really empty receive ring, we're
882 * in deep trouble, as the sender will definitely see RNR
884 if (rds_ib_ring_empty(&ic
->i_recv_ring
))
885 rds_ib_stats_inc(s_ib_rx_ring_empty
);
888 * If the ring is running low, then schedule the thread to refill.
890 if (rds_ib_ring_low(&ic
->i_recv_ring
))
891 queue_delayed_work(rds_wq
, &conn
->c_recv_w
, 0);
894 int rds_ib_recv(struct rds_connection
*conn
)
896 struct rds_ib_connection
*ic
= conn
->c_transport_data
;
899 rdsdebug("conn %p\n", conn
);
902 * If we get a temporary posting failure in this context then
903 * we're really low and we want the caller to back off for a bit.
905 mutex_lock(&ic
->i_recv_mutex
);
906 if (rds_ib_recv_refill(conn
, GFP_KERNEL
, GFP_HIGHUSER
, 0))
909 rds_ib_stats_inc(s_ib_rx_refill_from_thread
);
910 mutex_unlock(&ic
->i_recv_mutex
);
912 if (rds_conn_up(conn
))
913 rds_ib_attempt_ack(ic
);
918 int __init
rds_ib_recv_init(void)
923 /* Default to 30% of all available RAM for recv memory */
925 rds_ib_sysctl_max_recv_allocation
= si
.totalram
/ 3 * PAGE_SIZE
/ RDS_FRAG_SIZE
;
927 rds_ib_incoming_slab
= kmem_cache_create("rds_ib_incoming",
928 sizeof(struct rds_ib_incoming
),
930 if (rds_ib_incoming_slab
== NULL
)
933 rds_ib_frag_slab
= kmem_cache_create("rds_ib_frag",
934 sizeof(struct rds_page_frag
),
936 if (rds_ib_frag_slab
== NULL
)
937 kmem_cache_destroy(rds_ib_incoming_slab
);
944 void rds_ib_recv_exit(void)
946 kmem_cache_destroy(rds_ib_incoming_slab
);
947 kmem_cache_destroy(rds_ib_frag_slab
);