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>
41 void rds_inc_init(struct rds_incoming
*inc
, struct rds_connection
*conn
,
44 atomic_set(&inc
->i_refcount
, 1);
45 INIT_LIST_HEAD(&inc
->i_item
);
48 inc
->i_rdma_cookie
= 0;
50 EXPORT_SYMBOL_GPL(rds_inc_init
);
52 void rds_inc_addref(struct rds_incoming
*inc
)
54 rdsdebug("addref inc %p ref %d\n", inc
, atomic_read(&inc
->i_refcount
));
55 atomic_inc(&inc
->i_refcount
);
57 EXPORT_SYMBOL_GPL(rds_inc_addref
);
59 void rds_inc_put(struct rds_incoming
*inc
)
61 rdsdebug("put inc %p ref %d\n", inc
, atomic_read(&inc
->i_refcount
));
62 if (atomic_dec_and_test(&inc
->i_refcount
)) {
63 BUG_ON(!list_empty(&inc
->i_item
));
65 inc
->i_conn
->c_trans
->inc_free(inc
);
68 EXPORT_SYMBOL_GPL(rds_inc_put
);
70 static void rds_recv_rcvbuf_delta(struct rds_sock
*rs
, struct sock
*sk
,
71 struct rds_cong_map
*map
,
72 int delta
, __be16 port
)
79 rs
->rs_rcv_bytes
+= delta
;
80 now_congested
= rs
->rs_rcv_bytes
> rds_sk_rcvbuf(rs
);
82 rdsdebug("rs %p (%pI4:%u) recv bytes %d buf %d "
83 "now_cong %d delta %d\n",
84 rs
, &rs
->rs_bound_addr
,
85 ntohs(rs
->rs_bound_port
), rs
->rs_rcv_bytes
,
86 rds_sk_rcvbuf(rs
), now_congested
, delta
);
88 /* wasn't -> am congested */
89 if (!rs
->rs_congested
&& now_congested
) {
91 rds_cong_set_bit(map
, port
);
92 rds_cong_queue_updates(map
);
94 /* was -> aren't congested */
95 /* Require more free space before reporting uncongested to prevent
96 bouncing cong/uncong state too often */
97 else if (rs
->rs_congested
&& (rs
->rs_rcv_bytes
< (rds_sk_rcvbuf(rs
)/2))) {
99 rds_cong_clear_bit(map
, port
);
100 rds_cong_queue_updates(map
);
103 /* do nothing if no change in cong state */
107 * Process all extension headers that come with this message.
109 static void rds_recv_incoming_exthdrs(struct rds_incoming
*inc
, struct rds_sock
*rs
)
111 struct rds_header
*hdr
= &inc
->i_hdr
;
112 unsigned int pos
= 0, type
, len
;
114 struct rds_ext_header_version version
;
115 struct rds_ext_header_rdma rdma
;
116 struct rds_ext_header_rdma_dest rdma_dest
;
120 len
= sizeof(buffer
);
121 type
= rds_message_next_extension(hdr
, &pos
, &buffer
, &len
);
122 if (type
== RDS_EXTHDR_NONE
)
124 /* Process extension header here */
126 case RDS_EXTHDR_RDMA
:
127 rds_rdma_unuse(rs
, be32_to_cpu(buffer
.rdma
.h_rdma_rkey
), 0);
130 case RDS_EXTHDR_RDMA_DEST
:
131 /* We ignore the size for now. We could stash it
132 * somewhere and use it for error checking. */
133 inc
->i_rdma_cookie
= rds_rdma_make_cookie(
134 be32_to_cpu(buffer
.rdma_dest
.h_rdma_rkey
),
135 be32_to_cpu(buffer
.rdma_dest
.h_rdma_offset
));
143 * The transport must make sure that this is serialized against other
144 * rx and conn reset on this specific conn.
146 * We currently assert that only one fragmented message will be sent
147 * down a connection at a time. This lets us reassemble in the conn
148 * instead of per-flow which means that we don't have to go digging through
149 * flows to tear down partial reassembly progress on conn failure and
150 * we save flow lookup and locking for each frag arrival. It does mean
151 * that small messages will wait behind large ones. Fragmenting at all
152 * is only to reduce the memory consumption of pre-posted buffers.
154 * The caller passes in saddr and daddr instead of us getting it from the
155 * conn. This lets loopback, who only has one conn for both directions,
156 * tell us which roles the addrs in the conn are playing for this message.
158 void rds_recv_incoming(struct rds_connection
*conn
, __be32 saddr
, __be32 daddr
,
159 struct rds_incoming
*inc
, gfp_t gfp
, enum km_type km
)
161 struct rds_sock
*rs
= NULL
;
166 inc
->i_rx_jiffies
= jiffies
;
168 rdsdebug("conn %p next %llu inc %p seq %llu len %u sport %u dport %u "
169 "flags 0x%x rx_jiffies %lu\n", conn
,
170 (unsigned long long)conn
->c_next_rx_seq
,
172 (unsigned long long)be64_to_cpu(inc
->i_hdr
.h_sequence
),
173 be32_to_cpu(inc
->i_hdr
.h_len
),
174 be16_to_cpu(inc
->i_hdr
.h_sport
),
175 be16_to_cpu(inc
->i_hdr
.h_dport
),
179 if (be64_to_cpu(inc
->i_hdr
.h_sequence
) < conn
->c_next_rx_seq
&&
180 (inc
->i_hdr
.h_flags
& RDS_FLAG_RETRANSMITTED
)) {
181 rds_stats_inc(s_recv_drop_old_seq
);
184 conn
->c_next_rx_seq
= be64_to_cpu(inc
->i_hdr
.h_sequence
) + 1;
186 if (rds_sysctl_ping_enable
&& inc
->i_hdr
.h_dport
== 0) {
187 rds_stats_inc(s_recv_ping
);
188 rds_send_pong(conn
, inc
->i_hdr
.h_sport
);
192 rs
= rds_find_bound(daddr
, inc
->i_hdr
.h_dport
);
194 rds_stats_inc(s_recv_drop_no_sock
);
198 /* Process extension headers */
199 rds_recv_incoming_exthdrs(inc
, rs
);
201 /* We can be racing with rds_release() which marks the socket dead. */
202 sk
= rds_rs_to_sk(rs
);
204 /* serialize with rds_release -> sock_orphan */
205 write_lock_irqsave(&rs
->rs_recv_lock
, flags
);
206 if (!sock_flag(sk
, SOCK_DEAD
)) {
207 rdsdebug("adding inc %p to rs %p's recv queue\n", inc
, rs
);
208 rds_stats_inc(s_recv_queued
);
209 rds_recv_rcvbuf_delta(rs
, sk
, inc
->i_conn
->c_lcong
,
210 be32_to_cpu(inc
->i_hdr
.h_len
),
213 list_add_tail(&inc
->i_item
, &rs
->rs_recv_queue
);
214 __rds_wake_sk_sleep(sk
);
216 rds_stats_inc(s_recv_drop_dead_sock
);
218 write_unlock_irqrestore(&rs
->rs_recv_lock
, flags
);
224 EXPORT_SYMBOL_GPL(rds_recv_incoming
);
227 * be very careful here. This is being called as the condition in
228 * wait_event_*() needs to cope with being called many times.
230 static int rds_next_incoming(struct rds_sock
*rs
, struct rds_incoming
**inc
)
235 read_lock_irqsave(&rs
->rs_recv_lock
, flags
);
236 if (!list_empty(&rs
->rs_recv_queue
)) {
237 *inc
= list_entry(rs
->rs_recv_queue
.next
,
240 rds_inc_addref(*inc
);
242 read_unlock_irqrestore(&rs
->rs_recv_lock
, flags
);
248 static int rds_still_queued(struct rds_sock
*rs
, struct rds_incoming
*inc
,
251 struct sock
*sk
= rds_rs_to_sk(rs
);
255 write_lock_irqsave(&rs
->rs_recv_lock
, flags
);
256 if (!list_empty(&inc
->i_item
)) {
259 rds_recv_rcvbuf_delta(rs
, sk
, inc
->i_conn
->c_lcong
,
260 -be32_to_cpu(inc
->i_hdr
.h_len
),
262 list_del_init(&inc
->i_item
);
266 write_unlock_irqrestore(&rs
->rs_recv_lock
, flags
);
268 rdsdebug("inc %p rs %p still %d dropped %d\n", inc
, rs
, ret
, drop
);
273 * Pull errors off the error queue.
274 * If msghdr is NULL, we will just purge the error queue.
276 int rds_notify_queue_get(struct rds_sock
*rs
, struct msghdr
*msghdr
)
278 struct rds_notifier
*notifier
;
279 struct rds_rdma_notify cmsg
= { 0 }; /* fill holes with zero */
280 unsigned int count
= 0, max_messages
= ~0U;
286 /* put_cmsg copies to user space and thus may sleep. We can't do this
287 * with rs_lock held, so first grab as many notifications as we can stuff
288 * in the user provided cmsg buffer. We don't try to copy more, to avoid
289 * losing notifications - except when the buffer is so small that it wouldn't
290 * even hold a single notification. Then we give him as much of this single
291 * msg as we can squeeze in, and set MSG_CTRUNC.
294 max_messages
= msghdr
->msg_controllen
/ CMSG_SPACE(sizeof(cmsg
));
299 spin_lock_irqsave(&rs
->rs_lock
, flags
);
300 while (!list_empty(&rs
->rs_notify_queue
) && count
< max_messages
) {
301 notifier
= list_entry(rs
->rs_notify_queue
.next
,
302 struct rds_notifier
, n_list
);
303 list_move(¬ifier
->n_list
, ©
);
306 spin_unlock_irqrestore(&rs
->rs_lock
, flags
);
311 while (!list_empty(©
)) {
312 notifier
= list_entry(copy
.next
, struct rds_notifier
, n_list
);
315 cmsg
.user_token
= notifier
->n_user_token
;
316 cmsg
.status
= notifier
->n_status
;
318 err
= put_cmsg(msghdr
, SOL_RDS
, RDS_CMSG_RDMA_STATUS
,
319 sizeof(cmsg
), &cmsg
);
324 list_del_init(¬ifier
->n_list
);
328 /* If we bailed out because of an error in put_cmsg,
329 * we may be left with one or more notifications that we
330 * didn't process. Return them to the head of the list. */
331 if (!list_empty(©
)) {
332 spin_lock_irqsave(&rs
->rs_lock
, flags
);
333 list_splice(©
, &rs
->rs_notify_queue
);
334 spin_unlock_irqrestore(&rs
->rs_lock
, flags
);
341 * Queue a congestion notification
343 static int rds_notify_cong(struct rds_sock
*rs
, struct msghdr
*msghdr
)
345 uint64_t notify
= rs
->rs_cong_notify
;
349 err
= put_cmsg(msghdr
, SOL_RDS
, RDS_CMSG_CONG_UPDATE
,
350 sizeof(notify
), ¬ify
);
354 spin_lock_irqsave(&rs
->rs_lock
, flags
);
355 rs
->rs_cong_notify
&= ~notify
;
356 spin_unlock_irqrestore(&rs
->rs_lock
, flags
);
362 * Receive any control messages.
364 static int rds_cmsg_recv(struct rds_incoming
*inc
, struct msghdr
*msg
)
368 if (inc
->i_rdma_cookie
) {
369 ret
= put_cmsg(msg
, SOL_RDS
, RDS_CMSG_RDMA_DEST
,
370 sizeof(inc
->i_rdma_cookie
), &inc
->i_rdma_cookie
);
378 int rds_recvmsg(struct kiocb
*iocb
, struct socket
*sock
, struct msghdr
*msg
,
379 size_t size
, int msg_flags
)
381 struct sock
*sk
= sock
->sk
;
382 struct rds_sock
*rs
= rds_sk_to_rs(sk
);
384 int ret
= 0, nonblock
= msg_flags
& MSG_DONTWAIT
;
385 struct sockaddr_in
*sin
;
386 struct rds_incoming
*inc
= NULL
;
388 /* udp_recvmsg()->sock_recvtimeo() gets away without locking too.. */
389 timeo
= sock_rcvtimeo(sk
, nonblock
);
391 rdsdebug("size %zu flags 0x%x timeo %ld\n", size
, msg_flags
, timeo
);
393 if (msg_flags
& MSG_OOB
)
397 /* If there are pending notifications, do those - and nothing else */
398 if (!list_empty(&rs
->rs_notify_queue
)) {
399 ret
= rds_notify_queue_get(rs
, msg
);
403 if (rs
->rs_cong_notify
) {
404 ret
= rds_notify_cong(rs
, msg
);
408 if (!rds_next_incoming(rs
, &inc
)) {
414 timeo
= wait_event_interruptible_timeout(*sk_sleep(sk
),
415 (!list_empty(&rs
->rs_notify_queue
) ||
416 rs
->rs_cong_notify
||
417 rds_next_incoming(rs
, &inc
)), timeo
);
418 rdsdebug("recvmsg woke inc %p timeo %ld\n", inc
,
420 if (timeo
> 0 || timeo
== MAX_SCHEDULE_TIMEOUT
)
429 rdsdebug("copying inc %p from %pI4:%u to user\n", inc
,
430 &inc
->i_conn
->c_faddr
,
431 ntohs(inc
->i_hdr
.h_sport
));
432 ret
= inc
->i_conn
->c_trans
->inc_copy_to_user(inc
, msg
->msg_iov
,
438 * if the message we just copied isn't at the head of the
439 * recv queue then someone else raced us to return it, try
440 * to get the next message.
442 if (!rds_still_queued(rs
, inc
, !(msg_flags
& MSG_PEEK
))) {
445 rds_stats_inc(s_recv_deliver_raced
);
449 if (ret
< be32_to_cpu(inc
->i_hdr
.h_len
)) {
450 if (msg_flags
& MSG_TRUNC
)
451 ret
= be32_to_cpu(inc
->i_hdr
.h_len
);
452 msg
->msg_flags
|= MSG_TRUNC
;
455 if (rds_cmsg_recv(inc
, msg
)) {
460 rds_stats_inc(s_recv_delivered
);
462 sin
= (struct sockaddr_in
*)msg
->msg_name
;
464 sin
->sin_family
= AF_INET
;
465 sin
->sin_port
= inc
->i_hdr
.h_sport
;
466 sin
->sin_addr
.s_addr
= inc
->i_saddr
;
467 memset(sin
->sin_zero
, 0, sizeof(sin
->sin_zero
));
480 * The socket is being shut down and we're asked to drop messages that were
481 * queued for recvmsg. The caller has unbound the socket so the receive path
482 * won't queue any more incoming fragments or messages on the socket.
484 void rds_clear_recv_queue(struct rds_sock
*rs
)
486 struct sock
*sk
= rds_rs_to_sk(rs
);
487 struct rds_incoming
*inc
, *tmp
;
490 write_lock_irqsave(&rs
->rs_recv_lock
, flags
);
491 list_for_each_entry_safe(inc
, tmp
, &rs
->rs_recv_queue
, i_item
) {
492 rds_recv_rcvbuf_delta(rs
, sk
, inc
->i_conn
->c_lcong
,
493 -be32_to_cpu(inc
->i_hdr
.h_len
),
495 list_del_init(&inc
->i_item
);
498 write_unlock_irqrestore(&rs
->rs_recv_lock
, flags
);
502 * inc->i_saddr isn't used here because it is only set in the receive
505 void rds_inc_info_copy(struct rds_incoming
*inc
,
506 struct rds_info_iterator
*iter
,
507 __be32 saddr
, __be32 daddr
, int flip
)
509 struct rds_info_message minfo
;
511 minfo
.seq
= be64_to_cpu(inc
->i_hdr
.h_sequence
);
512 minfo
.len
= be32_to_cpu(inc
->i_hdr
.h_len
);
517 minfo
.lport
= inc
->i_hdr
.h_dport
;
518 minfo
.fport
= inc
->i_hdr
.h_sport
;
522 minfo
.lport
= inc
->i_hdr
.h_sport
;
523 minfo
.fport
= inc
->i_hdr
.h_dport
;
526 rds_info_copy(iter
, &minfo
, sizeof(minfo
));