1 // SPDX-License-Identifier: GPL-2.0-or-later
2 /* RxRPC individual remote procedure call handling
3 *
4 * Copyright (C) 2007 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/slab.h>
11 #include <linux/module.h>
12 #include <linux/circ_buf.h>
13 #include <linux/spinlock_types.h>
14 #include <net/sock.h>
15 #include <net/af_rxrpc.h>
16 #include "ar-internal.h"
17
18 const char *const rxrpc_call_states[NR__RXRPC_CALL_STATES] = {
19 [RXRPC_CALL_UNINITIALISED] = "Uninit ",
20 [RXRPC_CALL_CLIENT_AWAIT_CONN] = "ClWtConn",
21 [RXRPC_CALL_CLIENT_SEND_REQUEST] = "ClSndReq",
22 [RXRPC_CALL_CLIENT_AWAIT_REPLY] = "ClAwtRpl",
23 [RXRPC_CALL_CLIENT_RECV_REPLY] = "ClRcvRpl",
24 [RXRPC_CALL_SERVER_PREALLOC] = "SvPrealc",
25 [RXRPC_CALL_SERVER_RECV_REQUEST] = "SvRcvReq",
26 [RXRPC_CALL_SERVER_ACK_REQUEST] = "SvAckReq",
27 [RXRPC_CALL_SERVER_SEND_REPLY] = "SvSndRpl",
28 [RXRPC_CALL_SERVER_AWAIT_ACK] = "SvAwtACK",
29 [RXRPC_CALL_COMPLETE] = "Complete",
30 };
31
32 const char *const rxrpc_call_completions[NR__RXRPC_CALL_COMPLETIONS] = {
33 [RXRPC_CALL_SUCCEEDED] = "Complete",
34 [RXRPC_CALL_REMOTELY_ABORTED] = "RmtAbort",
35 [RXRPC_CALL_LOCALLY_ABORTED] = "LocAbort",
36 [RXRPC_CALL_LOCAL_ERROR] = "LocError",
37 [RXRPC_CALL_NETWORK_ERROR] = "NetError",
38 };
39
40 struct kmem_cache *rxrpc_call_jar;
41
42 static DEFINE_SEMAPHORE(rxrpc_call_limiter, 1000);
43 static DEFINE_SEMAPHORE(rxrpc_kernel_call_limiter, 1000);
44
rxrpc_poke_call(struct rxrpc_call * call,enum rxrpc_call_poke_trace what)45 void rxrpc_poke_call(struct rxrpc_call *call, enum rxrpc_call_poke_trace what)
46 {
47 struct rxrpc_local *local = call->local;
48 bool busy;
49
50 if (!test_bit(RXRPC_CALL_DISCONNECTED, &call->flags)) {
51 spin_lock_bh(&local->lock);
52 busy = !list_empty(&call->attend_link);
53 trace_rxrpc_poke_call(call, busy, what);
54 if (!busy && !rxrpc_try_get_call(call, rxrpc_call_get_poke))
55 busy = true;
56 if (!busy) {
57 list_add_tail(&call->attend_link, &local->call_attend_q);
58 }
59 spin_unlock_bh(&local->lock);
60 if (!busy)
61 rxrpc_wake_up_io_thread(local);
62 }
63 }
64
rxrpc_call_timer_expired(struct timer_list * t)65 static void rxrpc_call_timer_expired(struct timer_list *t)
66 {
67 struct rxrpc_call *call = from_timer(call, t, timer);
68
69 _enter("%d", call->debug_id);
70
71 if (!__rxrpc_call_is_complete(call)) {
72 trace_rxrpc_timer_expired(call, jiffies);
73 rxrpc_poke_call(call, rxrpc_call_poke_timer);
74 }
75 }
76
rxrpc_reduce_call_timer(struct rxrpc_call * call,unsigned long expire_at,unsigned long now,enum rxrpc_timer_trace why)77 void rxrpc_reduce_call_timer(struct rxrpc_call *call,
78 unsigned long expire_at,
79 unsigned long now,
80 enum rxrpc_timer_trace why)
81 {
82 trace_rxrpc_timer(call, why, now);
83 timer_reduce(&call->timer, expire_at);
84 }
85
86 static struct lock_class_key rxrpc_call_user_mutex_lock_class_key;
87
88 static void rxrpc_destroy_call(struct work_struct *);
89
90 /*
91 * find an extant server call
92 * - called in process context with IRQs enabled
93 */
rxrpc_find_call_by_user_ID(struct rxrpc_sock * rx,unsigned long user_call_ID)94 struct rxrpc_call *rxrpc_find_call_by_user_ID(struct rxrpc_sock *rx,
95 unsigned long user_call_ID)
96 {
97 struct rxrpc_call *call;
98 struct rb_node *p;
99
100 _enter("%p,%lx", rx, user_call_ID);
101
102 read_lock(&rx->call_lock);
103
104 p = rx->calls.rb_node;
105 while (p) {
106 call = rb_entry(p, struct rxrpc_call, sock_node);
107
108 if (user_call_ID < call->user_call_ID)
109 p = p->rb_left;
110 else if (user_call_ID > call->user_call_ID)
111 p = p->rb_right;
112 else
113 goto found_extant_call;
114 }
115
116 read_unlock(&rx->call_lock);
117 _leave(" = NULL");
118 return NULL;
119
120 found_extant_call:
121 rxrpc_get_call(call, rxrpc_call_get_sendmsg);
122 read_unlock(&rx->call_lock);
123 _leave(" = %p [%d]", call, refcount_read(&call->ref));
124 return call;
125 }
126
127 /*
128 * allocate a new call
129 */
rxrpc_alloc_call(struct rxrpc_sock * rx,gfp_t gfp,unsigned int debug_id)130 struct rxrpc_call *rxrpc_alloc_call(struct rxrpc_sock *rx, gfp_t gfp,
131 unsigned int debug_id)
132 {
133 struct rxrpc_call *call;
134 struct rxrpc_net *rxnet = rxrpc_net(sock_net(&rx->sk));
135
136 call = kmem_cache_zalloc(rxrpc_call_jar, gfp);
137 if (!call)
138 return NULL;
139
140 mutex_init(&call->user_mutex);
141
142 /* Prevent lockdep reporting a deadlock false positive between the afs
143 * filesystem and sys_sendmsg() via the mmap sem.
144 */
145 if (rx->sk.sk_kern_sock)
146 lockdep_set_class(&call->user_mutex,
147 &rxrpc_call_user_mutex_lock_class_key);
148
149 timer_setup(&call->timer, rxrpc_call_timer_expired, 0);
150 INIT_WORK(&call->destroyer, rxrpc_destroy_call);
151 INIT_LIST_HEAD(&call->link);
152 INIT_LIST_HEAD(&call->wait_link);
153 INIT_LIST_HEAD(&call->accept_link);
154 INIT_LIST_HEAD(&call->recvmsg_link);
155 INIT_LIST_HEAD(&call->sock_link);
156 INIT_LIST_HEAD(&call->attend_link);
157 INIT_LIST_HEAD(&call->tx_sendmsg);
158 INIT_LIST_HEAD(&call->tx_buffer);
159 skb_queue_head_init(&call->recvmsg_queue);
160 skb_queue_head_init(&call->rx_oos_queue);
161 init_waitqueue_head(&call->waitq);
162 spin_lock_init(&call->notify_lock);
163 spin_lock_init(&call->tx_lock);
164 refcount_set(&call->ref, 1);
165 call->debug_id = debug_id;
166 call->tx_total_len = -1;
167 call->next_rx_timo = 20 * HZ;
168 call->next_req_timo = 1 * HZ;
169 call->ackr_window = 1;
170 call->ackr_wtop = 1;
171
172 memset(&call->sock_node, 0xed, sizeof(call->sock_node));
173
174 call->rx_winsize = rxrpc_rx_window_size;
175 call->tx_winsize = 16;
176
177 call->cong_cwnd = RXRPC_MIN_CWND;
178 call->cong_ssthresh = RXRPC_TX_MAX_WINDOW;
179
180 call->rxnet = rxnet;
181 call->rtt_avail = RXRPC_CALL_RTT_AVAIL_MASK;
182 atomic_inc(&rxnet->nr_calls);
183 return call;
184 }
185
186 /*
187 * Allocate a new client call.
188 */
rxrpc_alloc_client_call(struct rxrpc_sock * rx,struct sockaddr_rxrpc * srx,struct rxrpc_conn_parameters * cp,struct rxrpc_call_params * p,gfp_t gfp,unsigned int debug_id)189 static struct rxrpc_call *rxrpc_alloc_client_call(struct rxrpc_sock *rx,
190 struct sockaddr_rxrpc *srx,
191 struct rxrpc_conn_parameters *cp,
192 struct rxrpc_call_params *p,
193 gfp_t gfp,
194 unsigned int debug_id)
195 {
196 struct rxrpc_call *call;
197 ktime_t now;
198 int ret;
199
200 _enter("");
201
202 call = rxrpc_alloc_call(rx, gfp, debug_id);
203 if (!call)
204 return ERR_PTR(-ENOMEM);
205 now = ktime_get_real();
206 call->acks_latest_ts = now;
207 call->cong_tstamp = now;
208 call->dest_srx = *srx;
209 call->interruptibility = p->interruptibility;
210 call->tx_total_len = p->tx_total_len;
211 call->key = key_get(cp->key);
212 call->local = rxrpc_get_local(cp->local, rxrpc_local_get_call);
213 call->security_level = cp->security_level;
214 if (p->kernel)
215 __set_bit(RXRPC_CALL_KERNEL, &call->flags);
216 if (cp->upgrade)
217 __set_bit(RXRPC_CALL_UPGRADE, &call->flags);
218 if (cp->exclusive)
219 __set_bit(RXRPC_CALL_EXCLUSIVE, &call->flags);
220
221 if (p->timeouts.normal)
222 call->next_rx_timo = min(msecs_to_jiffies(p->timeouts.normal), 1UL);
223 if (p->timeouts.idle)
224 call->next_req_timo = min(msecs_to_jiffies(p->timeouts.idle), 1UL);
225 if (p->timeouts.hard)
226 call->hard_timo = p->timeouts.hard * HZ;
227
228 ret = rxrpc_init_client_call_security(call);
229 if (ret < 0) {
230 rxrpc_prefail_call(call, RXRPC_CALL_LOCAL_ERROR, ret);
231 rxrpc_put_call(call, rxrpc_call_put_discard_error);
232 return ERR_PTR(ret);
233 }
234
235 rxrpc_set_call_state(call, RXRPC_CALL_CLIENT_AWAIT_CONN);
236
237 trace_rxrpc_call(call->debug_id, refcount_read(&call->ref),
238 p->user_call_ID, rxrpc_call_new_client);
239
240 _leave(" = %p", call);
241 return call;
242 }
243
244 /*
245 * Initiate the call ack/resend/expiry timer.
246 */
rxrpc_start_call_timer(struct rxrpc_call * call)247 void rxrpc_start_call_timer(struct rxrpc_call *call)
248 {
249 unsigned long now = jiffies;
250 unsigned long j = now + MAX_JIFFY_OFFSET;
251
252 call->delay_ack_at = j;
253 call->ack_lost_at = j;
254 call->resend_at = j;
255 call->ping_at = j;
256 call->keepalive_at = j;
257 call->expect_rx_by = j;
258 call->expect_req_by = j;
259 call->expect_term_by = j + call->hard_timo;
260 call->timer.expires = now;
261 }
262
263 /*
264 * Wait for a call slot to become available.
265 */
rxrpc_get_call_slot(struct rxrpc_call_params * p,gfp_t gfp)266 static struct semaphore *rxrpc_get_call_slot(struct rxrpc_call_params *p, gfp_t gfp)
267 {
268 struct semaphore *limiter = &rxrpc_call_limiter;
269
270 if (p->kernel)
271 limiter = &rxrpc_kernel_call_limiter;
272 if (p->interruptibility == RXRPC_UNINTERRUPTIBLE) {
273 down(limiter);
274 return limiter;
275 }
276 return down_interruptible(limiter) < 0 ? NULL : limiter;
277 }
278
279 /*
280 * Release a call slot.
281 */
rxrpc_put_call_slot(struct rxrpc_call * call)282 static void rxrpc_put_call_slot(struct rxrpc_call *call)
283 {
284 struct semaphore *limiter = &rxrpc_call_limiter;
285
286 if (test_bit(RXRPC_CALL_KERNEL, &call->flags))
287 limiter = &rxrpc_kernel_call_limiter;
288 up(limiter);
289 }
290
291 /*
292 * Start the process of connecting a call. We obtain a peer and a connection
293 * bundle, but the actual association of a call with a connection is offloaded
294 * to the I/O thread to simplify locking.
295 */
rxrpc_connect_call(struct rxrpc_call * call,gfp_t gfp)296 static int rxrpc_connect_call(struct rxrpc_call *call, gfp_t gfp)
297 {
298 struct rxrpc_local *local = call->local;
299 int ret = -ENOMEM;
300
301 _enter("{%d,%lx},", call->debug_id, call->user_call_ID);
302
303 call->peer = rxrpc_lookup_peer(local, &call->dest_srx, gfp);
304 if (!call->peer)
305 goto error;
306
307 ret = rxrpc_look_up_bundle(call, gfp);
308 if (ret < 0)
309 goto error;
310
311 trace_rxrpc_client(NULL, -1, rxrpc_client_queue_new_call);
312 rxrpc_get_call(call, rxrpc_call_get_io_thread);
313 spin_lock(&local->client_call_lock);
314 list_add_tail(&call->wait_link, &local->new_client_calls);
315 spin_unlock(&local->client_call_lock);
316 rxrpc_wake_up_io_thread(local);
317 return 0;
318
319 error:
320 __set_bit(RXRPC_CALL_DISCONNECTED, &call->flags);
321 return ret;
322 }
323
324 /*
325 * Set up a call for the given parameters.
326 * - Called with the socket lock held, which it must release.
327 * - If it returns a call, the call's lock will need releasing by the caller.
328 */
rxrpc_new_client_call(struct rxrpc_sock * rx,struct rxrpc_conn_parameters * cp,struct sockaddr_rxrpc * srx,struct rxrpc_call_params * p,gfp_t gfp,unsigned int debug_id)329 struct rxrpc_call *rxrpc_new_client_call(struct rxrpc_sock *rx,
330 struct rxrpc_conn_parameters *cp,
331 struct sockaddr_rxrpc *srx,
332 struct rxrpc_call_params *p,
333 gfp_t gfp,
334 unsigned int debug_id)
335 __releases(&rx->sk.sk_lock.slock)
336 __acquires(&call->user_mutex)
337 {
338 struct rxrpc_call *call, *xcall;
339 struct rxrpc_net *rxnet;
340 struct semaphore *limiter;
341 struct rb_node *parent, **pp;
342 int ret;
343
344 _enter("%p,%lx", rx, p->user_call_ID);
345
346 limiter = rxrpc_get_call_slot(p, gfp);
347 if (!limiter) {
348 release_sock(&rx->sk);
349 return ERR_PTR(-ERESTARTSYS);
350 }
351
352 call = rxrpc_alloc_client_call(rx, srx, cp, p, gfp, debug_id);
353 if (IS_ERR(call)) {
354 release_sock(&rx->sk);
355 up(limiter);
356 _leave(" = %ld", PTR_ERR(call));
357 return call;
358 }
359
360 /* We need to protect a partially set up call against the user as we
361 * will be acting outside the socket lock.
362 */
363 mutex_lock(&call->user_mutex);
364
365 /* Publish the call, even though it is incompletely set up as yet */
366 write_lock(&rx->call_lock);
367
368 pp = &rx->calls.rb_node;
369 parent = NULL;
370 while (*pp) {
371 parent = *pp;
372 xcall = rb_entry(parent, struct rxrpc_call, sock_node);
373
374 if (p->user_call_ID < xcall->user_call_ID)
375 pp = &(*pp)->rb_left;
376 else if (p->user_call_ID > xcall->user_call_ID)
377 pp = &(*pp)->rb_right;
378 else
379 goto error_dup_user_ID;
380 }
381
382 rcu_assign_pointer(call->socket, rx);
383 call->user_call_ID = p->user_call_ID;
384 __set_bit(RXRPC_CALL_HAS_USERID, &call->flags);
385 rxrpc_get_call(call, rxrpc_call_get_userid);
386 rb_link_node(&call->sock_node, parent, pp);
387 rb_insert_color(&call->sock_node, &rx->calls);
388 list_add(&call->sock_link, &rx->sock_calls);
389
390 write_unlock(&rx->call_lock);
391
392 rxnet = call->rxnet;
393 spin_lock(&rxnet->call_lock);
394 list_add_tail_rcu(&call->link, &rxnet->calls);
395 spin_unlock(&rxnet->call_lock);
396
397 /* From this point on, the call is protected by its own lock. */
398 release_sock(&rx->sk);
399
400 /* Set up or get a connection record and set the protocol parameters,
401 * including channel number and call ID.
402 */
403 ret = rxrpc_connect_call(call, gfp);
404 if (ret < 0)
405 goto error_attached_to_socket;
406
407 _leave(" = %p [new]", call);
408 return call;
409
410 /* We unexpectedly found the user ID in the list after taking
411 * the call_lock. This shouldn't happen unless the user races
412 * with itself and tries to add the same user ID twice at the
413 * same time in different threads.
414 */
415 error_dup_user_ID:
416 write_unlock(&rx->call_lock);
417 release_sock(&rx->sk);
418 rxrpc_prefail_call(call, RXRPC_CALL_LOCAL_ERROR, -EEXIST);
419 trace_rxrpc_call(call->debug_id, refcount_read(&call->ref), 0,
420 rxrpc_call_see_userid_exists);
421 mutex_unlock(&call->user_mutex);
422 rxrpc_put_call(call, rxrpc_call_put_userid_exists);
423 _leave(" = -EEXIST");
424 return ERR_PTR(-EEXIST);
425
426 /* We got an error, but the call is attached to the socket and is in
427 * need of release. However, we might now race with recvmsg() when it
428 * completion notifies the socket. Return 0 from sys_sendmsg() and
429 * leave the error to recvmsg() to deal with.
430 */
431 error_attached_to_socket:
432 trace_rxrpc_call(call->debug_id, refcount_read(&call->ref), ret,
433 rxrpc_call_see_connect_failed);
434 rxrpc_set_call_completion(call, RXRPC_CALL_LOCAL_ERROR, 0, ret);
435 _leave(" = c=%08x [err]", call->debug_id);
436 return call;
437 }
438
439 /*
440 * Set up an incoming call. call->conn points to the connection.
441 * This is called in BH context and isn't allowed to fail.
442 */
rxrpc_incoming_call(struct rxrpc_sock * rx,struct rxrpc_call * call,struct sk_buff * skb)443 void rxrpc_incoming_call(struct rxrpc_sock *rx,
444 struct rxrpc_call *call,
445 struct sk_buff *skb)
446 {
447 struct rxrpc_connection *conn = call->conn;
448 struct rxrpc_skb_priv *sp = rxrpc_skb(skb);
449 u32 chan;
450
451 _enter(",%d", call->conn->debug_id);
452
453 rcu_assign_pointer(call->socket, rx);
454 call->call_id = sp->hdr.callNumber;
455 call->dest_srx.srx_service = sp->hdr.serviceId;
456 call->cid = sp->hdr.cid;
457 call->cong_tstamp = skb->tstamp;
458
459 __set_bit(RXRPC_CALL_EXPOSED, &call->flags);
460 rxrpc_set_call_state(call, RXRPC_CALL_SERVER_RECV_REQUEST);
461
462 spin_lock(&conn->state_lock);
463
464 switch (conn->state) {
465 case RXRPC_CONN_SERVICE_UNSECURED:
466 case RXRPC_CONN_SERVICE_CHALLENGING:
467 __set_bit(RXRPC_CALL_CONN_CHALLENGING, &call->flags);
468 break;
469 case RXRPC_CONN_SERVICE:
470 break;
471
472 case RXRPC_CONN_ABORTED:
473 rxrpc_set_call_completion(call, conn->completion,
474 conn->abort_code, conn->error);
475 break;
476 default:
477 BUG();
478 }
479
480 rxrpc_get_call(call, rxrpc_call_get_io_thread);
481
482 /* Set the channel for this call. We don't get channel_lock as we're
483 * only defending against the data_ready handler (which we're called
484 * from) and the RESPONSE packet parser (which is only really
485 * interested in call_counter and can cope with a disagreement with the
486 * call pointer).
487 */
488 chan = sp->hdr.cid & RXRPC_CHANNELMASK;
489 conn->channels[chan].call_counter = call->call_id;
490 conn->channels[chan].call_id = call->call_id;
491 conn->channels[chan].call = call;
492 spin_unlock(&conn->state_lock);
493
494 spin_lock(&conn->peer->lock);
495 hlist_add_head(&call->error_link, &conn->peer->error_targets);
496 spin_unlock(&conn->peer->lock);
497
498 rxrpc_start_call_timer(call);
499 _leave("");
500 }
501
502 /*
503 * Note the re-emergence of a call.
504 */
rxrpc_see_call(struct rxrpc_call * call,enum rxrpc_call_trace why)505 void rxrpc_see_call(struct rxrpc_call *call, enum rxrpc_call_trace why)
506 {
507 if (call) {
508 int r = refcount_read(&call->ref);
509
510 trace_rxrpc_call(call->debug_id, r, 0, why);
511 }
512 }
513
rxrpc_try_get_call(struct rxrpc_call * call,enum rxrpc_call_trace why)514 struct rxrpc_call *rxrpc_try_get_call(struct rxrpc_call *call,
515 enum rxrpc_call_trace why)
516 {
517 int r;
518
519 if (!call || !__refcount_inc_not_zero(&call->ref, &r))
520 return NULL;
521 trace_rxrpc_call(call->debug_id, r + 1, 0, why);
522 return call;
523 }
524
525 /*
526 * Note the addition of a ref on a call.
527 */
rxrpc_get_call(struct rxrpc_call * call,enum rxrpc_call_trace why)528 void rxrpc_get_call(struct rxrpc_call *call, enum rxrpc_call_trace why)
529 {
530 int r;
531
532 __refcount_inc(&call->ref, &r);
533 trace_rxrpc_call(call->debug_id, r + 1, 0, why);
534 }
535
536 /*
537 * Clean up the Rx skb ring.
538 */
rxrpc_cleanup_ring(struct rxrpc_call * call)539 static void rxrpc_cleanup_ring(struct rxrpc_call *call)
540 {
541 rxrpc_purge_queue(&call->recvmsg_queue);
542 rxrpc_purge_queue(&call->rx_oos_queue);
543 }
544
545 /*
546 * Detach a call from its owning socket.
547 */
rxrpc_release_call(struct rxrpc_sock * rx,struct rxrpc_call * call)548 void rxrpc_release_call(struct rxrpc_sock *rx, struct rxrpc_call *call)
549 {
550 struct rxrpc_connection *conn = call->conn;
551 bool put = false, putu = false;
552
553 _enter("{%d,%d}", call->debug_id, refcount_read(&call->ref));
554
555 trace_rxrpc_call(call->debug_id, refcount_read(&call->ref),
556 call->flags, rxrpc_call_see_release);
557
558 if (test_and_set_bit(RXRPC_CALL_RELEASED, &call->flags))
559 BUG();
560
561 rxrpc_put_call_slot(call);
562
563 /* Make sure we don't get any more notifications */
564 spin_lock(&rx->recvmsg_lock);
565
566 if (!list_empty(&call->recvmsg_link)) {
567 _debug("unlinking once-pending call %p { e=%lx f=%lx }",
568 call, call->events, call->flags);
569 list_del(&call->recvmsg_link);
570 put = true;
571 }
572
573 /* list_empty() must return false in rxrpc_notify_socket() */
574 call->recvmsg_link.next = NULL;
575 call->recvmsg_link.prev = NULL;
576
577 spin_unlock(&rx->recvmsg_lock);
578 if (put)
579 rxrpc_put_call(call, rxrpc_call_put_unnotify);
580
581 write_lock(&rx->call_lock);
582
583 if (test_and_clear_bit(RXRPC_CALL_HAS_USERID, &call->flags)) {
584 rb_erase(&call->sock_node, &rx->calls);
585 memset(&call->sock_node, 0xdd, sizeof(call->sock_node));
586 putu = true;
587 }
588
589 list_del(&call->sock_link);
590 write_unlock(&rx->call_lock);
591
592 _debug("RELEASE CALL %p (%d CONN %p)", call, call->debug_id, conn);
593
594 if (putu)
595 rxrpc_put_call(call, rxrpc_call_put_userid);
596
597 _leave("");
598 }
599
600 /*
601 * release all the calls associated with a socket
602 */
rxrpc_release_calls_on_socket(struct rxrpc_sock * rx)603 void rxrpc_release_calls_on_socket(struct rxrpc_sock *rx)
604 {
605 struct rxrpc_call *call;
606
607 _enter("%p", rx);
608
609 while (!list_empty(&rx->to_be_accepted)) {
610 call = list_entry(rx->to_be_accepted.next,
611 struct rxrpc_call, accept_link);
612 list_del(&call->accept_link);
613 rxrpc_propose_abort(call, RX_CALL_DEAD, -ECONNRESET,
614 rxrpc_abort_call_sock_release_tba);
615 rxrpc_put_call(call, rxrpc_call_put_release_sock_tba);
616 }
617
618 while (!list_empty(&rx->sock_calls)) {
619 call = list_entry(rx->sock_calls.next,
620 struct rxrpc_call, sock_link);
621 rxrpc_get_call(call, rxrpc_call_get_release_sock);
622 rxrpc_propose_abort(call, RX_CALL_DEAD, -ECONNRESET,
623 rxrpc_abort_call_sock_release);
624 rxrpc_release_call(rx, call);
625 rxrpc_put_call(call, rxrpc_call_put_release_sock);
626 }
627
628 _leave("");
629 }
630
631 /*
632 * release a call
633 */
rxrpc_put_call(struct rxrpc_call * call,enum rxrpc_call_trace why)634 void rxrpc_put_call(struct rxrpc_call *call, enum rxrpc_call_trace why)
635 {
636 struct rxrpc_net *rxnet = call->rxnet;
637 unsigned int debug_id = call->debug_id;
638 bool dead;
639 int r;
640
641 ASSERT(call != NULL);
642
643 dead = __refcount_dec_and_test(&call->ref, &r);
644 trace_rxrpc_call(debug_id, r - 1, 0, why);
645 if (dead) {
646 ASSERTCMP(__rxrpc_call_state(call), ==, RXRPC_CALL_COMPLETE);
647
648 if (!list_empty(&call->link)) {
649 spin_lock(&rxnet->call_lock);
650 list_del_init(&call->link);
651 spin_unlock(&rxnet->call_lock);
652 }
653
654 rxrpc_cleanup_call(call);
655 }
656 }
657
658 /*
659 * Free up the call under RCU.
660 */
rxrpc_rcu_free_call(struct rcu_head * rcu)661 static void rxrpc_rcu_free_call(struct rcu_head *rcu)
662 {
663 struct rxrpc_call *call = container_of(rcu, struct rxrpc_call, rcu);
664 struct rxrpc_net *rxnet = READ_ONCE(call->rxnet);
665
666 kmem_cache_free(rxrpc_call_jar, call);
667 if (atomic_dec_and_test(&rxnet->nr_calls))
668 wake_up_var(&rxnet->nr_calls);
669 }
670
671 /*
672 * Final call destruction - but must be done in process context.
673 */
rxrpc_destroy_call(struct work_struct * work)674 static void rxrpc_destroy_call(struct work_struct *work)
675 {
676 struct rxrpc_call *call = container_of(work, struct rxrpc_call, destroyer);
677 struct rxrpc_txbuf *txb;
678
679 del_timer_sync(&call->timer);
680
681 rxrpc_free_skb(call->cong_last_nack, rxrpc_skb_put_last_nack);
682 rxrpc_cleanup_ring(call);
683 while ((txb = list_first_entry_or_null(&call->tx_sendmsg,
684 struct rxrpc_txbuf, call_link))) {
685 list_del(&txb->call_link);
686 rxrpc_put_txbuf(txb, rxrpc_txbuf_put_cleaned);
687 }
688 while ((txb = list_first_entry_or_null(&call->tx_buffer,
689 struct rxrpc_txbuf, call_link))) {
690 list_del(&txb->call_link);
691 rxrpc_put_txbuf(txb, rxrpc_txbuf_put_cleaned);
692 }
693
694 rxrpc_put_txbuf(call->tx_pending, rxrpc_txbuf_put_cleaned);
695 rxrpc_put_connection(call->conn, rxrpc_conn_put_call);
696 rxrpc_deactivate_bundle(call->bundle);
697 rxrpc_put_bundle(call->bundle, rxrpc_bundle_put_call);
698 rxrpc_put_peer(call->peer, rxrpc_peer_put_call);
699 rxrpc_put_local(call->local, rxrpc_local_put_call);
700 call_rcu(&call->rcu, rxrpc_rcu_free_call);
701 }
702
703 /*
704 * clean up a call
705 */
rxrpc_cleanup_call(struct rxrpc_call * call)706 void rxrpc_cleanup_call(struct rxrpc_call *call)
707 {
708 memset(&call->sock_node, 0xcd, sizeof(call->sock_node));
709
710 ASSERTCMP(__rxrpc_call_state(call), ==, RXRPC_CALL_COMPLETE);
711 ASSERT(test_bit(RXRPC_CALL_RELEASED, &call->flags));
712
713 del_timer(&call->timer);
714
715 if (rcu_read_lock_held())
716 /* Can't use the rxrpc workqueue as we need to cancel/flush
717 * something that may be running/waiting there.
718 */
719 schedule_work(&call->destroyer);
720 else
721 rxrpc_destroy_call(&call->destroyer);
722 }
723
724 /*
725 * Make sure that all calls are gone from a network namespace. To reach this
726 * point, any open UDP sockets in that namespace must have been closed, so any
727 * outstanding calls cannot be doing I/O.
728 */
rxrpc_destroy_all_calls(struct rxrpc_net * rxnet)729 void rxrpc_destroy_all_calls(struct rxrpc_net *rxnet)
730 {
731 struct rxrpc_call *call;
732
733 _enter("");
734
735 if (!list_empty(&rxnet->calls)) {
736 spin_lock(&rxnet->call_lock);
737
738 while (!list_empty(&rxnet->calls)) {
739 call = list_entry(rxnet->calls.next,
740 struct rxrpc_call, link);
741 _debug("Zapping call %p", call);
742
743 rxrpc_see_call(call, rxrpc_call_see_zap);
744 list_del_init(&call->link);
745
746 pr_err("Call %p still in use (%d,%s,%lx,%lx)!\n",
747 call, refcount_read(&call->ref),
748 rxrpc_call_states[__rxrpc_call_state(call)],
749 call->flags, call->events);
750
751 spin_unlock(&rxnet->call_lock);
752 cond_resched();
753 spin_lock(&rxnet->call_lock);
754 }
755
756 spin_unlock(&rxnet->call_lock);
757 }
758
759 atomic_dec(&rxnet->nr_calls);
760 wait_var_event(&rxnet->nr_calls, !atomic_read(&rxnet->nr_calls));
761 }
762