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_conn_path
*conn_path
;
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_conn_path
*cp
= arg
->conn_path
;
159 struct rds_tcp_connection
*tc
= cp
->cp_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_path_init(&tinc
->ti_inc
, cp
,
182 cp
->cp_conn
->c_faddr
);
184 * XXX * we might be able to use the __ variants when
185 * we've already serialized at a higher level.
187 skb_queue_head_init(&tinc
->ti_skb_list
);
190 if (left
&& tc
->t_tinc_hdr_rem
) {
191 to_copy
= min(tc
->t_tinc_hdr_rem
, left
);
192 rdsdebug("copying %zu header from skb %p\n", to_copy
,
194 skb_copy_bits(skb
, offset
,
195 (char *)&tinc
->ti_inc
.i_hdr
+
196 sizeof(struct rds_header
) -
199 tc
->t_tinc_hdr_rem
-= to_copy
;
203 if (tc
->t_tinc_hdr_rem
== 0) {
204 /* could be 0 for a 0 len message */
205 tc
->t_tinc_data_rem
=
206 be32_to_cpu(tinc
->ti_inc
.i_hdr
.h_len
);
210 if (left
&& tc
->t_tinc_data_rem
) {
211 to_copy
= min(tc
->t_tinc_data_rem
, left
);
213 clone
= pskb_extract(skb
, offset
, to_copy
, arg
->gfp
);
215 desc
->error
= -ENOMEM
;
219 skb_queue_tail(&tinc
->ti_skb_list
, clone
);
221 rdsdebug("skb %p data %p len %d off %u to_copy %zu -> "
222 "clone %p data %p len %d\n",
223 skb
, skb
->data
, skb
->len
, offset
, to_copy
,
224 clone
, clone
->data
, clone
->len
);
226 tc
->t_tinc_data_rem
-= to_copy
;
231 if (tc
->t_tinc_hdr_rem
== 0 && tc
->t_tinc_data_rem
== 0) {
232 struct rds_connection
*conn
= cp
->cp_conn
;
234 if (tinc
->ti_inc
.i_hdr
.h_flags
== RDS_FLAG_CONG_BITMAP
)
235 rds_tcp_cong_recv(conn
, tinc
);
237 rds_recv_incoming(conn
, conn
->c_faddr
,
238 conn
->c_laddr
, &tinc
->ti_inc
,
241 tc
->t_tinc_hdr_rem
= sizeof(struct rds_header
);
242 tc
->t_tinc_data_rem
= 0;
244 rds_inc_put(&tinc
->ti_inc
);
249 rdsdebug("returning len %zu left %zu skb len %d rx queue depth %d\n",
251 skb_queue_len(&tc
->t_sock
->sk
->sk_receive_queue
));
255 /* the caller has to hold the sock lock */
256 static int rds_tcp_read_sock(struct rds_conn_path
*cp
, gfp_t gfp
)
258 struct rds_tcp_connection
*tc
= cp
->cp_transport_data
;
259 struct socket
*sock
= tc
->t_sock
;
260 read_descriptor_t desc
;
261 struct rds_tcp_desc_arg arg
;
263 /* It's like glib in the kernel! */
266 desc
.arg
.data
= &arg
;
268 desc
.count
= 1; /* give more than one skb per call */
270 tcp_read_sock(sock
->sk
, &desc
, rds_tcp_data_recv
);
271 rdsdebug("tcp_read_sock for tc %p gfp 0x%x returned %d\n", tc
, gfp
,
278 * We hold the sock lock to serialize our rds_tcp_recv->tcp_read_sock from
281 * if we fail to allocate we're in trouble.. blindly wait some time before
282 * trying again to see if the VM can free up something for us.
284 int rds_tcp_recv_path(struct rds_conn_path
*cp
)
286 struct rds_tcp_connection
*tc
= cp
->cp_transport_data
;
287 struct socket
*sock
= tc
->t_sock
;
290 rdsdebug("recv worker path [%d] tc %p sock %p\n",
291 cp
->cp_index
, tc
, sock
);
294 ret
= rds_tcp_read_sock(cp
, GFP_KERNEL
);
295 release_sock(sock
->sk
);
300 void rds_tcp_data_ready(struct sock
*sk
)
302 void (*ready
)(struct sock
*sk
);
303 struct rds_conn_path
*cp
;
304 struct rds_tcp_connection
*tc
;
306 rdsdebug("data ready sk %p\n", sk
);
308 read_lock_bh(&sk
->sk_callback_lock
);
309 cp
= sk
->sk_user_data
;
310 if (!cp
) { /* check for teardown race */
311 ready
= sk
->sk_data_ready
;
315 tc
= cp
->cp_transport_data
;
316 ready
= tc
->t_orig_data_ready
;
317 rds_tcp_stats_inc(s_tcp_data_ready_calls
);
319 if (rds_tcp_read_sock(cp
, GFP_ATOMIC
) == -ENOMEM
)
320 queue_delayed_work(rds_wq
, &cp
->cp_recv_w
, 0);
322 read_unlock_bh(&sk
->sk_callback_lock
);
326 int rds_tcp_recv_init(void)
328 rds_tcp_incoming_slab
= kmem_cache_create("rds_tcp_incoming",
329 sizeof(struct rds_tcp_incoming
),
331 if (!rds_tcp_incoming_slab
)
336 void rds_tcp_recv_exit(void)
338 kmem_cache_destroy(rds_tcp_incoming_slab
);
This page took 0.03919 seconds and 6 git commands to generate.