rxrpc: Move packet reception processing into I/O thread
[linux-block.git] / net / rxrpc / local_object.c
1 // SPDX-License-Identifier: GPL-2.0-or-later
2 /* Local endpoint object management
3  *
4  * Copyright (C) 2016 Red Hat, Inc. All Rights Reserved.
5  * Written by David Howells (dhowells@redhat.com)
6  */
7
8 #define pr_fmt(fmt) KBUILD_MODNAME ": " fmt
9
10 #include <linux/module.h>
11 #include <linux/net.h>
12 #include <linux/skbuff.h>
13 #include <linux/slab.h>
14 #include <linux/udp.h>
15 #include <linux/ip.h>
16 #include <linux/hashtable.h>
17 #include <net/sock.h>
18 #include <net/udp.h>
19 #include <net/udp_tunnel.h>
20 #include <net/af_rxrpc.h>
21 #include "ar-internal.h"
22
23 static void rxrpc_local_processor(struct work_struct *);
24 static void rxrpc_local_rcu(struct rcu_head *);
25
26 /*
27  * Handle an ICMP/ICMP6 error turning up at the tunnel.  Push it through the
28  * usual mechanism so that it gets parsed and presented through the UDP
29  * socket's error_report().
30  */
31 static void rxrpc_encap_err_rcv(struct sock *sk, struct sk_buff *skb, int err,
32                                 __be16 port, u32 info, u8 *payload)
33 {
34         if (ip_hdr(skb)->version == IPVERSION)
35                 return ip_icmp_error(sk, skb, err, port, info, payload);
36         if (IS_ENABLED(CONFIG_AF_RXRPC_IPV6))
37                 return ipv6_icmp_error(sk, skb, err, port, info, payload);
38 }
39
40 /*
41  * Compare a local to an address.  Return -ve, 0 or +ve to indicate less than,
42  * same or greater than.
43  *
44  * We explicitly don't compare the RxRPC service ID as we want to reject
45  * conflicting uses by differing services.  Further, we don't want to share
46  * addresses with different options (IPv6), so we don't compare those bits
47  * either.
48  */
49 static long rxrpc_local_cmp_key(const struct rxrpc_local *local,
50                                 const struct sockaddr_rxrpc *srx)
51 {
52         long diff;
53
54         diff = ((local->srx.transport_type - srx->transport_type) ?:
55                 (local->srx.transport_len - srx->transport_len) ?:
56                 (local->srx.transport.family - srx->transport.family));
57         if (diff != 0)
58                 return diff;
59
60         switch (srx->transport.family) {
61         case AF_INET:
62                 /* If the choice of UDP port is left up to the transport, then
63                  * the endpoint record doesn't match.
64                  */
65                 return ((u16 __force)local->srx.transport.sin.sin_port -
66                         (u16 __force)srx->transport.sin.sin_port) ?:
67                         memcmp(&local->srx.transport.sin.sin_addr,
68                                &srx->transport.sin.sin_addr,
69                                sizeof(struct in_addr));
70 #ifdef CONFIG_AF_RXRPC_IPV6
71         case AF_INET6:
72                 /* If the choice of UDP6 port is left up to the transport, then
73                  * the endpoint record doesn't match.
74                  */
75                 return ((u16 __force)local->srx.transport.sin6.sin6_port -
76                         (u16 __force)srx->transport.sin6.sin6_port) ?:
77                         memcmp(&local->srx.transport.sin6.sin6_addr,
78                                &srx->transport.sin6.sin6_addr,
79                                sizeof(struct in6_addr));
80 #endif
81         default:
82                 BUG();
83         }
84 }
85
86 /*
87  * Allocate a new local endpoint.
88  */
89 static struct rxrpc_local *rxrpc_alloc_local(struct rxrpc_net *rxnet,
90                                              const struct sockaddr_rxrpc *srx)
91 {
92         struct rxrpc_local *local;
93
94         local = kzalloc(sizeof(struct rxrpc_local), GFP_KERNEL);
95         if (local) {
96                 refcount_set(&local->ref, 1);
97                 atomic_set(&local->active_users, 1);
98                 local->rxnet = rxnet;
99                 INIT_HLIST_NODE(&local->link);
100                 INIT_WORK(&local->processor, rxrpc_local_processor);
101                 INIT_LIST_HEAD(&local->ack_tx_queue);
102                 spin_lock_init(&local->ack_tx_lock);
103                 init_rwsem(&local->defrag_sem);
104                 skb_queue_head_init(&local->reject_queue);
105                 skb_queue_head_init(&local->event_queue);
106                 skb_queue_head_init(&local->rx_queue);
107                 local->client_bundles = RB_ROOT;
108                 spin_lock_init(&local->client_bundles_lock);
109                 spin_lock_init(&local->lock);
110                 rwlock_init(&local->services_lock);
111                 local->debug_id = atomic_inc_return(&rxrpc_debug_id);
112                 memcpy(&local->srx, srx, sizeof(*srx));
113                 local->srx.srx_service = 0;
114                 trace_rxrpc_local(local->debug_id, rxrpc_local_new, 1, 1);
115         }
116
117         _leave(" = %p", local);
118         return local;
119 }
120
121 /*
122  * create the local socket
123  * - must be called with rxrpc_local_mutex locked
124  */
125 static int rxrpc_open_socket(struct rxrpc_local *local, struct net *net)
126 {
127         struct udp_tunnel_sock_cfg tuncfg = {NULL};
128         struct sockaddr_rxrpc *srx = &local->srx;
129         struct udp_port_cfg udp_conf = {0};
130         struct task_struct *io_thread;
131         struct sock *usk;
132         int ret;
133
134         _enter("%p{%d,%d}",
135                local, srx->transport_type, srx->transport.family);
136
137         udp_conf.family = srx->transport.family;
138         udp_conf.use_udp_checksums = true;
139         if (udp_conf.family == AF_INET) {
140                 udp_conf.local_ip = srx->transport.sin.sin_addr;
141                 udp_conf.local_udp_port = srx->transport.sin.sin_port;
142 #if IS_ENABLED(CONFIG_AF_RXRPC_IPV6)
143         } else {
144                 udp_conf.local_ip6 = srx->transport.sin6.sin6_addr;
145                 udp_conf.local_udp_port = srx->transport.sin6.sin6_port;
146                 udp_conf.use_udp6_tx_checksums = true;
147                 udp_conf.use_udp6_rx_checksums = true;
148 #endif
149         }
150         ret = udp_sock_create(net, &udp_conf, &local->socket);
151         if (ret < 0) {
152                 _leave(" = %d [socket]", ret);
153                 return ret;
154         }
155
156         tuncfg.encap_type = UDP_ENCAP_RXRPC;
157         tuncfg.encap_rcv = rxrpc_encap_rcv;
158         tuncfg.encap_err_rcv = rxrpc_encap_err_rcv;
159         tuncfg.sk_user_data = local;
160         setup_udp_tunnel_sock(net, local->socket, &tuncfg);
161
162         /* set the socket up */
163         usk = local->socket->sk;
164         usk->sk_error_report = rxrpc_error_report;
165
166         switch (srx->transport.family) {
167         case AF_INET6:
168                 /* we want to receive ICMPv6 errors */
169                 ip6_sock_set_recverr(usk);
170
171                 /* Fall through and set IPv4 options too otherwise we don't get
172                  * errors from IPv4 packets sent through the IPv6 socket.
173                  */
174                 fallthrough;
175         case AF_INET:
176                 /* we want to receive ICMP errors */
177                 ip_sock_set_recverr(usk);
178
179                 /* we want to set the don't fragment bit */
180                 ip_sock_set_mtu_discover(usk, IP_PMTUDISC_DO);
181
182                 /* We want receive timestamps. */
183                 sock_enable_timestamps(usk);
184                 break;
185
186         default:
187                 BUG();
188         }
189
190         io_thread = kthread_run(rxrpc_io_thread, local,
191                                 "krxrpcio/%u", ntohs(udp_conf.local_udp_port));
192         if (IS_ERR(io_thread)) {
193                 ret = PTR_ERR(io_thread);
194                 goto error_sock;
195         }
196
197         local->io_thread = io_thread;
198         _leave(" = 0");
199         return 0;
200
201 error_sock:
202         kernel_sock_shutdown(local->socket, SHUT_RDWR);
203         local->socket->sk->sk_user_data = NULL;
204         sock_release(local->socket);
205         local->socket = NULL;
206         return ret;
207 }
208
209 /*
210  * Look up or create a new local endpoint using the specified local address.
211  */
212 struct rxrpc_local *rxrpc_lookup_local(struct net *net,
213                                        const struct sockaddr_rxrpc *srx)
214 {
215         struct rxrpc_local *local;
216         struct rxrpc_net *rxnet = rxrpc_net(net);
217         struct hlist_node *cursor;
218         long diff;
219         int ret;
220
221         _enter("{%d,%d,%pISp}",
222                srx->transport_type, srx->transport.family, &srx->transport);
223
224         mutex_lock(&rxnet->local_mutex);
225
226         hlist_for_each(cursor, &rxnet->local_endpoints) {
227                 local = hlist_entry(cursor, struct rxrpc_local, link);
228
229                 diff = rxrpc_local_cmp_key(local, srx);
230                 if (diff != 0)
231                         continue;
232
233                 /* Services aren't allowed to share transport sockets, so
234                  * reject that here.  It is possible that the object is dying -
235                  * but it may also still have the local transport address that
236                  * we want bound.
237                  */
238                 if (srx->srx_service) {
239                         local = NULL;
240                         goto addr_in_use;
241                 }
242
243                 /* Found a match.  We want to replace a dying object.
244                  * Attempting to bind the transport socket may still fail if
245                  * we're attempting to use a local address that the dying
246                  * object is still using.
247                  */
248                 if (!rxrpc_use_local(local, rxrpc_local_use_lookup))
249                         break;
250
251                 goto found;
252         }
253
254         local = rxrpc_alloc_local(rxnet, srx);
255         if (!local)
256                 goto nomem;
257
258         ret = rxrpc_open_socket(local, net);
259         if (ret < 0)
260                 goto sock_error;
261
262         if (cursor) {
263                 hlist_replace_rcu(cursor, &local->link);
264                 cursor->pprev = NULL;
265         } else {
266                 hlist_add_head_rcu(&local->link, &rxnet->local_endpoints);
267         }
268
269 found:
270         mutex_unlock(&rxnet->local_mutex);
271         _leave(" = %p", local);
272         return local;
273
274 nomem:
275         ret = -ENOMEM;
276 sock_error:
277         mutex_unlock(&rxnet->local_mutex);
278         if (local)
279                 call_rcu(&local->rcu, rxrpc_local_rcu);
280         _leave(" = %d", ret);
281         return ERR_PTR(ret);
282
283 addr_in_use:
284         mutex_unlock(&rxnet->local_mutex);
285         _leave(" = -EADDRINUSE");
286         return ERR_PTR(-EADDRINUSE);
287 }
288
289 /*
290  * Get a ref on a local endpoint.
291  */
292 struct rxrpc_local *rxrpc_get_local(struct rxrpc_local *local,
293                                     enum rxrpc_local_trace why)
294 {
295         int r, u;
296
297         u = atomic_read(&local->active_users);
298         __refcount_inc(&local->ref, &r);
299         trace_rxrpc_local(local->debug_id, why, r + 1, u);
300         return local;
301 }
302
303 /*
304  * Get a ref on a local endpoint unless its usage has already reached 0.
305  */
306 struct rxrpc_local *rxrpc_get_local_maybe(struct rxrpc_local *local,
307                                           enum rxrpc_local_trace why)
308 {
309         int r, u;
310
311         if (local && __refcount_inc_not_zero(&local->ref, &r)) {
312                 u = atomic_read(&local->active_users);
313                 trace_rxrpc_local(local->debug_id, why, r + 1, u);
314                 return local;
315         }
316
317         return NULL;
318 }
319
320 /*
321  * Queue a local endpoint and pass the caller's reference to the work item.
322  */
323 void rxrpc_queue_local(struct rxrpc_local *local)
324 {
325         unsigned int debug_id = local->debug_id;
326         int r = refcount_read(&local->ref);
327         int u = atomic_read(&local->active_users);
328
329         if (rxrpc_queue_work(&local->processor))
330                 trace_rxrpc_local(debug_id, rxrpc_local_queued, r, u);
331         else
332                 rxrpc_put_local(local, rxrpc_local_put_already_queued);
333 }
334
335 /*
336  * Drop a ref on a local endpoint.
337  */
338 void rxrpc_put_local(struct rxrpc_local *local, enum rxrpc_local_trace why)
339 {
340         unsigned int debug_id;
341         bool dead;
342         int r, u;
343
344         if (local) {
345                 debug_id = local->debug_id;
346
347                 u = atomic_read(&local->active_users);
348                 dead = __refcount_dec_and_test(&local->ref, &r);
349                 trace_rxrpc_local(debug_id, why, r, u);
350
351                 if (dead)
352                         call_rcu(&local->rcu, rxrpc_local_rcu);
353         }
354 }
355
356 /*
357  * Start using a local endpoint.
358  */
359 struct rxrpc_local *rxrpc_use_local(struct rxrpc_local *local,
360                                     enum rxrpc_local_trace why)
361 {
362         local = rxrpc_get_local_maybe(local, rxrpc_local_get_for_use);
363         if (!local)
364                 return NULL;
365
366         if (!__rxrpc_use_local(local, why)) {
367                 rxrpc_put_local(local, rxrpc_local_put_for_use);
368                 return NULL;
369         }
370
371         return local;
372 }
373
374 /*
375  * Cease using a local endpoint.  Once the number of active users reaches 0, we
376  * start the closure of the transport in the work processor.
377  */
378 void rxrpc_unuse_local(struct rxrpc_local *local, enum rxrpc_local_trace why)
379 {
380         if (local && __rxrpc_unuse_local(local, why))
381                 kthread_stop(local->io_thread);
382 }
383
384 /*
385  * Destroy a local endpoint's socket and then hand the record to RCU to dispose
386  * of.
387  *
388  * Closing the socket cannot be done from bottom half context or RCU callback
389  * context because it might sleep.
390  */
391 void rxrpc_destroy_local(struct rxrpc_local *local)
392 {
393         struct socket *socket = local->socket;
394         struct rxrpc_net *rxnet = local->rxnet;
395
396         _enter("%d", local->debug_id);
397
398         local->dead = true;
399
400         mutex_lock(&rxnet->local_mutex);
401         hlist_del_init_rcu(&local->link);
402         mutex_unlock(&rxnet->local_mutex);
403
404         rxrpc_clean_up_local_conns(local);
405         rxrpc_service_connection_reaper(&rxnet->service_conn_reaper);
406         ASSERT(!local->service);
407
408         if (socket) {
409                 local->socket = NULL;
410                 kernel_sock_shutdown(socket, SHUT_RDWR);
411                 socket->sk->sk_user_data = NULL;
412                 sock_release(socket);
413         }
414
415         /* At this point, there should be no more packets coming in to the
416          * local endpoint.
417          */
418         rxrpc_purge_queue(&local->reject_queue);
419         rxrpc_purge_queue(&local->event_queue);
420         rxrpc_purge_queue(&local->rx_queue);
421 }
422
423 /*
424  * Process events on an endpoint.  The work item carries a ref which
425  * we must release.
426  */
427 static void rxrpc_local_processor(struct work_struct *work)
428 {
429         struct rxrpc_local *local =
430                 container_of(work, struct rxrpc_local, processor);
431         bool again;
432
433         if (local->dead)
434                 return;
435
436         rxrpc_see_local(local, rxrpc_local_processing);
437
438         do {
439                 again = false;
440                 if (!__rxrpc_use_local(local, rxrpc_local_use_work))
441                         break;
442
443                 if (!list_empty(&local->ack_tx_queue)) {
444                         rxrpc_transmit_ack_packets(local);
445                         again = true;
446                 }
447
448                 if (!skb_queue_empty(&local->reject_queue)) {
449                         rxrpc_reject_packets(local);
450                         again = true;
451                 }
452
453                 if (!skb_queue_empty(&local->event_queue)) {
454                         rxrpc_process_local_events(local);
455                         again = true;
456                 }
457
458                 __rxrpc_unuse_local(local, rxrpc_local_unuse_work);
459         } while (again);
460
461         rxrpc_put_local(local, rxrpc_local_put_queue);
462 }
463
464 /*
465  * Destroy a local endpoint after the RCU grace period expires.
466  */
467 static void rxrpc_local_rcu(struct rcu_head *rcu)
468 {
469         struct rxrpc_local *local = container_of(rcu, struct rxrpc_local, rcu);
470
471         _enter("%d", local->debug_id);
472
473         ASSERT(!work_pending(&local->processor));
474
475         rxrpc_see_local(local, rxrpc_local_free);
476         kfree(local);
477         _leave("");
478 }
479
480 /*
481  * Verify the local endpoint list is empty by this point.
482  */
483 void rxrpc_destroy_all_locals(struct rxrpc_net *rxnet)
484 {
485         struct rxrpc_local *local;
486
487         _enter("");
488
489         flush_workqueue(rxrpc_workqueue);
490
491         if (!hlist_empty(&rxnet->local_endpoints)) {
492                 mutex_lock(&rxnet->local_mutex);
493                 hlist_for_each_entry(local, &rxnet->local_endpoints, link) {
494                         pr_err("AF_RXRPC: Leaked local %p {%d}\n",
495                                local, refcount_read(&local->ref));
496                 }
497                 mutex_unlock(&rxnet->local_mutex);
498                 BUG();
499         }
500 }