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>
40 static struct kmem_cache
*rds_tcp_incoming_slab
;
42 static void rds_tcp_inc_purge(struct rds_incoming
*inc
)
44 struct rds_tcp_incoming
*tinc
;
45 tinc
= container_of(inc
, struct rds_tcp_incoming
, ti_inc
);
46 rdsdebug("purging tinc %p inc %p\n", tinc
, inc
);
47 skb_queue_purge(&tinc
->ti_skb_list
);
50 void rds_tcp_inc_free(struct rds_incoming
*inc
)
52 struct rds_tcp_incoming
*tinc
;
53 tinc
= container_of(inc
, struct rds_tcp_incoming
, ti_inc
);
54 rds_tcp_inc_purge(inc
);
55 rdsdebug("freeing tinc %p inc %p\n", tinc
, inc
);
56 kmem_cache_free(rds_tcp_incoming_slab
, tinc
);
60 * this is pretty lame, but, whatever.
62 int rds_tcp_inc_copy_to_user(struct rds_incoming
*inc
, struct iov_iter
*to
)
64 struct rds_tcp_incoming
*tinc
;
68 if (!iov_iter_count(to
))
71 tinc
= container_of(inc
, struct rds_tcp_incoming
, ti_inc
);
73 skb_queue_walk(&tinc
->ti_skb_list
, skb
) {
74 unsigned long to_copy
, skb_off
;
75 for (skb_off
= 0; skb_off
< skb
->len
; skb_off
+= to_copy
) {
76 to_copy
= iov_iter_count(to
);
77 to_copy
= min(to_copy
, skb
->len
- skb_off
);
79 if (skb_copy_datagram_iter(skb
, skb_off
, to
, to_copy
))
82 rds_stats_add(s_copy_to_user
, to_copy
);
85 if (!iov_iter_count(to
))
94 * We have a series of skbs that have fragmented pieces of the congestion
95 * bitmap. They must add up to the exact size of the congestion bitmap. We
96 * use the skb helpers to copy those into the pages that make up the in-memory
97 * congestion bitmap for the remote address of this connection. We then tell
98 * the congestion core that the bitmap has been changed so that it can wake up
101 * This is racing with sending paths which are using test_bit to see if the
102 * bitmap indicates that their recipient is congested.
105 static void rds_tcp_cong_recv(struct rds_connection
*conn
,
106 struct rds_tcp_incoming
*tinc
)
109 unsigned int to_copy
, skb_off
;
110 unsigned int map_off
;
111 unsigned int map_page
;
112 struct rds_cong_map
*map
;
115 /* catch completely corrupt packets */
116 if (be32_to_cpu(tinc
->ti_inc
.i_hdr
.h_len
) != RDS_CONG_MAP_BYTES
)
123 skb_queue_walk(&tinc
->ti_skb_list
, skb
) {
125 while (skb_off
< skb
->len
) {
126 to_copy
= min_t(unsigned int, PAGE_SIZE
- map_off
,
129 BUG_ON(map_page
>= RDS_CONG_MAP_PAGES
);
131 /* only returns 0 or -error */
132 ret
= skb_copy_bits(skb
, skb_off
,
133 (void *)map
->m_page_addrs
[map_page
] + map_off
,
139 if (map_off
== PAGE_SIZE
) {
146 rds_cong_map_updated(map
, ~(u64
) 0);
149 struct rds_tcp_desc_arg
{
150 struct rds_connection
*conn
;
154 static int rds_tcp_data_recv(read_descriptor_t
*desc
, struct sk_buff
*skb
,
155 unsigned int offset
, size_t len
)
157 struct rds_tcp_desc_arg
*arg
= desc
->arg
.data
;
158 struct rds_connection
*conn
= arg
->conn
;
159 struct rds_tcp_connection
*tc
= conn
->c_transport_data
;
160 struct rds_tcp_incoming
*tinc
= tc
->t_tinc
;
161 struct sk_buff
*clone
;
162 size_t left
= len
, to_copy
;
164 rdsdebug("tcp data tc %p skb %p offset %u len %zu\n", tc
, skb
, offset
,
168 * tcp_read_sock() interprets partial progress as an indication to stop
173 tinc
= kmem_cache_alloc(rds_tcp_incoming_slab
,
176 desc
->error
= -ENOMEM
;
180 rdsdebug("alloced tinc %p\n", tinc
);
181 rds_inc_init(&tinc
->ti_inc
, conn
, conn
->c_faddr
);
183 * XXX * we might be able to use the __ variants when
184 * we've already serialized at a higher level.
186 skb_queue_head_init(&tinc
->ti_skb_list
);
189 if (left
&& tc
->t_tinc_hdr_rem
) {
190 to_copy
= min(tc
->t_tinc_hdr_rem
, left
);
191 rdsdebug("copying %zu header from skb %p\n", to_copy
,
193 skb_copy_bits(skb
, offset
,
194 (char *)&tinc
->ti_inc
.i_hdr
+
195 sizeof(struct rds_header
) -
198 tc
->t_tinc_hdr_rem
-= to_copy
;
202 if (tc
->t_tinc_hdr_rem
== 0) {
203 /* could be 0 for a 0 len message */
204 tc
->t_tinc_data_rem
=
205 be32_to_cpu(tinc
->ti_inc
.i_hdr
.h_len
);
209 if (left
&& tc
->t_tinc_data_rem
) {
210 to_copy
= min(tc
->t_tinc_data_rem
, left
);
212 clone
= pskb_extract(skb
, offset
, to_copy
, arg
->gfp
);
214 desc
->error
= -ENOMEM
;
218 skb_queue_tail(&tinc
->ti_skb_list
, clone
);
220 rdsdebug("skb %p data %p len %d off %u to_copy %zu -> "
221 "clone %p data %p len %d\n",
222 skb
, skb
->data
, skb
->len
, offset
, to_copy
,
223 clone
, clone
->data
, clone
->len
);
225 tc
->t_tinc_data_rem
-= to_copy
;
230 if (tc
->t_tinc_hdr_rem
== 0 && tc
->t_tinc_data_rem
== 0) {
231 if (tinc
->ti_inc
.i_hdr
.h_flags
== RDS_FLAG_CONG_BITMAP
)
232 rds_tcp_cong_recv(conn
, tinc
);
234 rds_recv_incoming(conn
, conn
->c_faddr
,
235 conn
->c_laddr
, &tinc
->ti_inc
,
238 tc
->t_tinc_hdr_rem
= sizeof(struct rds_header
);
239 tc
->t_tinc_data_rem
= 0;
241 rds_inc_put(&tinc
->ti_inc
);
246 rdsdebug("returning len %zu left %zu skb len %d rx queue depth %d\n",
248 skb_queue_len(&tc
->t_sock
->sk
->sk_receive_queue
));
252 /* the caller has to hold the sock lock */
253 static int rds_tcp_read_sock(struct rds_connection
*conn
, gfp_t gfp
)
255 struct rds_tcp_connection
*tc
= conn
->c_transport_data
;
256 struct socket
*sock
= tc
->t_sock
;
257 read_descriptor_t desc
;
258 struct rds_tcp_desc_arg arg
;
260 /* It's like glib in the kernel! */
263 desc
.arg
.data
= &arg
;
265 desc
.count
= 1; /* give more than one skb per call */
267 tcp_read_sock(sock
->sk
, &desc
, rds_tcp_data_recv
);
268 rdsdebug("tcp_read_sock for tc %p gfp 0x%x returned %d\n", tc
, gfp
,
275 * We hold the sock lock to serialize our rds_tcp_recv->tcp_read_sock from
278 * if we fail to allocate we're in trouble.. blindly wait some time before
279 * trying again to see if the VM can free up something for us.
281 int rds_tcp_recv(struct rds_connection
*conn
)
283 struct rds_tcp_connection
*tc
= conn
->c_transport_data
;
284 struct socket
*sock
= tc
->t_sock
;
287 rdsdebug("recv worker conn %p tc %p sock %p\n", conn
, tc
, sock
);
290 ret
= rds_tcp_read_sock(conn
, GFP_KERNEL
);
291 release_sock(sock
->sk
);
296 void rds_tcp_data_ready(struct sock
*sk
)
298 void (*ready
)(struct sock
*sk
);
299 struct rds_connection
*conn
;
300 struct rds_tcp_connection
*tc
;
302 rdsdebug("data ready sk %p\n", sk
);
304 read_lock_bh(&sk
->sk_callback_lock
);
305 conn
= sk
->sk_user_data
;
306 if (!conn
) { /* check for teardown race */
307 ready
= sk
->sk_data_ready
;
311 tc
= conn
->c_transport_data
;
312 ready
= tc
->t_orig_data_ready
;
313 rds_tcp_stats_inc(s_tcp_data_ready_calls
);
315 if (rds_tcp_read_sock(conn
, GFP_ATOMIC
) == -ENOMEM
)
316 queue_delayed_work(rds_wq
, &conn
->c_recv_w
, 0);
318 read_unlock_bh(&sk
->sk_callback_lock
);
322 int rds_tcp_recv_init(void)
324 rds_tcp_incoming_slab
= kmem_cache_create("rds_tcp_incoming",
325 sizeof(struct rds_tcp_incoming
),
327 if (!rds_tcp_incoming_slab
)
332 void rds_tcp_recv_exit(void)
334 kmem_cache_destroy(rds_tcp_incoming_slab
);