xref: /linux/net/rxrpc/conn_client.c (revision aec499c75cf8e0b599be4d559e6922b613085f8f)
1 // SPDX-License-Identifier: GPL-2.0-or-later
2 /* Client connection-specific management code.
3  *
4  * Copyright (C) 2016, 2020 Red Hat, Inc. All Rights Reserved.
5  * Written by David Howells (dhowells@redhat.com)
6  *
7  * Client connections need to be cached for a little while after they've made a
8  * call so as to handle retransmitted DATA packets in case the server didn't
9  * receive the final ACK or terminating ABORT we sent it.
10  *
11  * There are flags of relevance to the cache:
12  *
13  *  (2) DONT_REUSE - The connection should be discarded as soon as possible and
14  *      should not be reused.  This is set when an exclusive connection is used
15  *      or a call ID counter overflows.
16  *
17  * The caching state may only be changed if the cache lock is held.
18  *
19  * There are two idle client connection expiry durations.  If the total number
20  * of connections is below the reap threshold, we use the normal duration; if
21  * it's above, we use the fast duration.
22  */
23 
24 #define pr_fmt(fmt) KBUILD_MODNAME ": " fmt
25 
26 #include <linux/slab.h>
27 #include <linux/idr.h>
28 #include <linux/timer.h>
29 #include <linux/sched/signal.h>
30 
31 #include "ar-internal.h"
32 
33 __read_mostly unsigned int rxrpc_reap_client_connections = 900;
34 __read_mostly unsigned long rxrpc_conn_idle_client_expiry = 2 * 60 * HZ;
35 __read_mostly unsigned long rxrpc_conn_idle_client_fast_expiry = 2 * HZ;
36 
37 /*
38  * We use machine-unique IDs for our client connections.
39  */
40 DEFINE_IDR(rxrpc_client_conn_ids);
41 static DEFINE_SPINLOCK(rxrpc_conn_id_lock);
42 
43 /*
44  * Get a connection ID and epoch for a client connection from the global pool.
45  * The connection struct pointer is then recorded in the idr radix tree.  The
46  * epoch doesn't change until the client is rebooted (or, at least, unless the
47  * module is unloaded).
48  */
49 static int rxrpc_get_client_connection_id(struct rxrpc_connection *conn,
50 					  gfp_t gfp)
51 {
52 	struct rxrpc_net *rxnet = conn->params.local->rxnet;
53 	int id;
54 
55 	_enter("");
56 
57 	idr_preload(gfp);
58 	spin_lock(&rxrpc_conn_id_lock);
59 
60 	id = idr_alloc_cyclic(&rxrpc_client_conn_ids, conn,
61 			      1, 0x40000000, GFP_NOWAIT);
62 	if (id < 0)
63 		goto error;
64 
65 	spin_unlock(&rxrpc_conn_id_lock);
66 	idr_preload_end();
67 
68 	conn->proto.epoch = rxnet->epoch;
69 	conn->proto.cid = id << RXRPC_CIDSHIFT;
70 	set_bit(RXRPC_CONN_HAS_IDR, &conn->flags);
71 	_leave(" [CID %x]", conn->proto.cid);
72 	return 0;
73 
74 error:
75 	spin_unlock(&rxrpc_conn_id_lock);
76 	idr_preload_end();
77 	_leave(" = %d", id);
78 	return id;
79 }
80 
81 /*
82  * Release a connection ID for a client connection from the global pool.
83  */
84 static void rxrpc_put_client_connection_id(struct rxrpc_connection *conn)
85 {
86 	if (test_bit(RXRPC_CONN_HAS_IDR, &conn->flags)) {
87 		spin_lock(&rxrpc_conn_id_lock);
88 		idr_remove(&rxrpc_client_conn_ids,
89 			   conn->proto.cid >> RXRPC_CIDSHIFT);
90 		spin_unlock(&rxrpc_conn_id_lock);
91 	}
92 }
93 
94 /*
95  * Destroy the client connection ID tree.
96  */
97 void rxrpc_destroy_client_conn_ids(void)
98 {
99 	struct rxrpc_connection *conn;
100 	int id;
101 
102 	if (!idr_is_empty(&rxrpc_client_conn_ids)) {
103 		idr_for_each_entry(&rxrpc_client_conn_ids, conn, id) {
104 			pr_err("AF_RXRPC: Leaked client conn %p {%d}\n",
105 			       conn, atomic_read(&conn->usage));
106 		}
107 		BUG();
108 	}
109 
110 	idr_destroy(&rxrpc_client_conn_ids);
111 }
112 
113 /*
114  * Allocate a connection bundle.
115  */
116 static struct rxrpc_bundle *rxrpc_alloc_bundle(struct rxrpc_conn_parameters *cp,
117 					       gfp_t gfp)
118 {
119 	struct rxrpc_bundle *bundle;
120 
121 	bundle = kzalloc(sizeof(*bundle), gfp);
122 	if (bundle) {
123 		bundle->params = *cp;
124 		rxrpc_get_peer(bundle->params.peer);
125 		atomic_set(&bundle->usage, 1);
126 		spin_lock_init(&bundle->channel_lock);
127 		INIT_LIST_HEAD(&bundle->waiting_calls);
128 	}
129 	return bundle;
130 }
131 
132 struct rxrpc_bundle *rxrpc_get_bundle(struct rxrpc_bundle *bundle)
133 {
134 	atomic_inc(&bundle->usage);
135 	return bundle;
136 }
137 
138 static void rxrpc_free_bundle(struct rxrpc_bundle *bundle)
139 {
140 	rxrpc_put_peer(bundle->params.peer);
141 	kfree(bundle);
142 }
143 
144 void rxrpc_put_bundle(struct rxrpc_bundle *bundle)
145 {
146 	unsigned int d = bundle->debug_id;
147 	unsigned int u = atomic_dec_return(&bundle->usage);
148 
149 	_debug("PUT B=%x %u", d, u);
150 	if (u == 0)
151 		rxrpc_free_bundle(bundle);
152 }
153 
154 /*
155  * Allocate a client connection.
156  */
157 static struct rxrpc_connection *
158 rxrpc_alloc_client_connection(struct rxrpc_bundle *bundle, gfp_t gfp)
159 {
160 	struct rxrpc_connection *conn;
161 	struct rxrpc_net *rxnet = bundle->params.local->rxnet;
162 	int ret;
163 
164 	_enter("");
165 
166 	conn = rxrpc_alloc_connection(gfp);
167 	if (!conn) {
168 		_leave(" = -ENOMEM");
169 		return ERR_PTR(-ENOMEM);
170 	}
171 
172 	atomic_set(&conn->usage, 1);
173 	conn->bundle		= bundle;
174 	conn->params		= bundle->params;
175 	conn->out_clientflag	= RXRPC_CLIENT_INITIATED;
176 	conn->state		= RXRPC_CONN_CLIENT;
177 	conn->service_id	= conn->params.service_id;
178 
179 	ret = rxrpc_get_client_connection_id(conn, gfp);
180 	if (ret < 0)
181 		goto error_0;
182 
183 	ret = rxrpc_init_client_conn_security(conn);
184 	if (ret < 0)
185 		goto error_1;
186 
187 	atomic_inc(&rxnet->nr_conns);
188 	write_lock(&rxnet->conn_lock);
189 	list_add_tail(&conn->proc_link, &rxnet->conn_proc_list);
190 	write_unlock(&rxnet->conn_lock);
191 
192 	rxrpc_get_bundle(bundle);
193 	rxrpc_get_peer(conn->params.peer);
194 	rxrpc_get_local(conn->params.local);
195 	key_get(conn->params.key);
196 
197 	trace_rxrpc_conn(conn->debug_id, rxrpc_conn_new_client,
198 			 atomic_read(&conn->usage),
199 			 __builtin_return_address(0));
200 
201 	atomic_inc(&rxnet->nr_client_conns);
202 	trace_rxrpc_client(conn, -1, rxrpc_client_alloc);
203 	_leave(" = %p", conn);
204 	return conn;
205 
206 error_1:
207 	rxrpc_put_client_connection_id(conn);
208 error_0:
209 	kfree(conn);
210 	_leave(" = %d", ret);
211 	return ERR_PTR(ret);
212 }
213 
214 /*
215  * Determine if a connection may be reused.
216  */
217 static bool rxrpc_may_reuse_conn(struct rxrpc_connection *conn)
218 {
219 	struct rxrpc_net *rxnet;
220 	int id_cursor, id, distance, limit;
221 
222 	if (!conn)
223 		goto dont_reuse;
224 
225 	rxnet = conn->params.local->rxnet;
226 	if (test_bit(RXRPC_CONN_DONT_REUSE, &conn->flags))
227 		goto dont_reuse;
228 
229 	if (conn->state != RXRPC_CONN_CLIENT ||
230 	    conn->proto.epoch != rxnet->epoch)
231 		goto mark_dont_reuse;
232 
233 	/* The IDR tree gets very expensive on memory if the connection IDs are
234 	 * widely scattered throughout the number space, so we shall want to
235 	 * kill off connections that, say, have an ID more than about four
236 	 * times the maximum number of client conns away from the current
237 	 * allocation point to try and keep the IDs concentrated.
238 	 */
239 	id_cursor = idr_get_cursor(&rxrpc_client_conn_ids);
240 	id = conn->proto.cid >> RXRPC_CIDSHIFT;
241 	distance = id - id_cursor;
242 	if (distance < 0)
243 		distance = -distance;
244 	limit = max_t(unsigned long, atomic_read(&rxnet->nr_conns) * 4, 1024);
245 	if (distance > limit)
246 		goto mark_dont_reuse;
247 
248 	return true;
249 
250 mark_dont_reuse:
251 	set_bit(RXRPC_CONN_DONT_REUSE, &conn->flags);
252 dont_reuse:
253 	return false;
254 }
255 
256 /*
257  * Look up the conn bundle that matches the connection parameters, adding it if
258  * it doesn't yet exist.
259  */
260 static struct rxrpc_bundle *rxrpc_look_up_bundle(struct rxrpc_conn_parameters *cp,
261 						 gfp_t gfp)
262 {
263 	static atomic_t rxrpc_bundle_id;
264 	struct rxrpc_bundle *bundle, *candidate;
265 	struct rxrpc_local *local = cp->local;
266 	struct rb_node *p, **pp, *parent;
267 	long diff;
268 
269 	_enter("{%px,%x,%u,%u}",
270 	       cp->peer, key_serial(cp->key), cp->security_level, cp->upgrade);
271 
272 	if (cp->exclusive)
273 		return rxrpc_alloc_bundle(cp, gfp);
274 
275 	/* First, see if the bundle is already there. */
276 	_debug("search 1");
277 	spin_lock(&local->client_bundles_lock);
278 	p = local->client_bundles.rb_node;
279 	while (p) {
280 		bundle = rb_entry(p, struct rxrpc_bundle, local_node);
281 
282 #define cmp(X) ((long)bundle->params.X - (long)cp->X)
283 		diff = (cmp(peer) ?:
284 			cmp(key) ?:
285 			cmp(security_level) ?:
286 			cmp(upgrade));
287 #undef cmp
288 		if (diff < 0)
289 			p = p->rb_left;
290 		else if (diff > 0)
291 			p = p->rb_right;
292 		else
293 			goto found_bundle;
294 	}
295 	spin_unlock(&local->client_bundles_lock);
296 	_debug("not found");
297 
298 	/* It wasn't.  We need to add one. */
299 	candidate = rxrpc_alloc_bundle(cp, gfp);
300 	if (!candidate)
301 		return NULL;
302 
303 	_debug("search 2");
304 	spin_lock(&local->client_bundles_lock);
305 	pp = &local->client_bundles.rb_node;
306 	parent = NULL;
307 	while (*pp) {
308 		parent = *pp;
309 		bundle = rb_entry(parent, struct rxrpc_bundle, local_node);
310 
311 #define cmp(X) ((long)bundle->params.X - (long)cp->X)
312 		diff = (cmp(peer) ?:
313 			cmp(key) ?:
314 			cmp(security_level) ?:
315 			cmp(upgrade));
316 #undef cmp
317 		if (diff < 0)
318 			pp = &(*pp)->rb_left;
319 		else if (diff > 0)
320 			pp = &(*pp)->rb_right;
321 		else
322 			goto found_bundle_free;
323 	}
324 
325 	_debug("new bundle");
326 	candidate->debug_id = atomic_inc_return(&rxrpc_bundle_id);
327 	rb_link_node(&candidate->local_node, parent, pp);
328 	rb_insert_color(&candidate->local_node, &local->client_bundles);
329 	rxrpc_get_bundle(candidate);
330 	spin_unlock(&local->client_bundles_lock);
331 	_leave(" = %u [new]", candidate->debug_id);
332 	return candidate;
333 
334 found_bundle_free:
335 	rxrpc_free_bundle(candidate);
336 found_bundle:
337 	rxrpc_get_bundle(bundle);
338 	spin_unlock(&local->client_bundles_lock);
339 	_leave(" = %u [found]", bundle->debug_id);
340 	return bundle;
341 }
342 
343 /*
344  * Create or find a client bundle to use for a call.
345  *
346  * If we return with a connection, the call will be on its waiting list.  It's
347  * left to the caller to assign a channel and wake up the call.
348  */
349 static struct rxrpc_bundle *rxrpc_prep_call(struct rxrpc_sock *rx,
350 					    struct rxrpc_call *call,
351 					    struct rxrpc_conn_parameters *cp,
352 					    struct sockaddr_rxrpc *srx,
353 					    gfp_t gfp)
354 {
355 	struct rxrpc_bundle *bundle;
356 
357 	_enter("{%d,%lx},", call->debug_id, call->user_call_ID);
358 
359 	cp->peer = rxrpc_lookup_peer(rx, cp->local, srx, gfp);
360 	if (!cp->peer)
361 		goto error;
362 
363 	call->cong_cwnd = cp->peer->cong_cwnd;
364 	if (call->cong_cwnd >= call->cong_ssthresh)
365 		call->cong_mode = RXRPC_CALL_CONGEST_AVOIDANCE;
366 	else
367 		call->cong_mode = RXRPC_CALL_SLOW_START;
368 	if (cp->upgrade)
369 		__set_bit(RXRPC_CALL_UPGRADE, &call->flags);
370 
371 	/* Find the client connection bundle. */
372 	bundle = rxrpc_look_up_bundle(cp, gfp);
373 	if (!bundle)
374 		goto error;
375 
376 	/* Get this call queued.  Someone else may activate it whilst we're
377 	 * lining up a new connection, but that's fine.
378 	 */
379 	spin_lock(&bundle->channel_lock);
380 	list_add_tail(&call->chan_wait_link, &bundle->waiting_calls);
381 	spin_unlock(&bundle->channel_lock);
382 
383 	_leave(" = [B=%x]", bundle->debug_id);
384 	return bundle;
385 
386 error:
387 	_leave(" = -ENOMEM");
388 	return ERR_PTR(-ENOMEM);
389 }
390 
391 /*
392  * Allocate a new connection and add it into a bundle.
393  */
394 static void rxrpc_add_conn_to_bundle(struct rxrpc_bundle *bundle, gfp_t gfp)
395 	__releases(bundle->channel_lock)
396 {
397 	struct rxrpc_connection *candidate = NULL, *old = NULL;
398 	bool conflict;
399 	int i;
400 
401 	_enter("");
402 
403 	conflict = bundle->alloc_conn;
404 	if (!conflict)
405 		bundle->alloc_conn = true;
406 	spin_unlock(&bundle->channel_lock);
407 	if (conflict) {
408 		_leave(" [conf]");
409 		return;
410 	}
411 
412 	candidate = rxrpc_alloc_client_connection(bundle, gfp);
413 
414 	spin_lock(&bundle->channel_lock);
415 	bundle->alloc_conn = false;
416 
417 	if (IS_ERR(candidate)) {
418 		bundle->alloc_error = PTR_ERR(candidate);
419 		spin_unlock(&bundle->channel_lock);
420 		_leave(" [err %ld]", PTR_ERR(candidate));
421 		return;
422 	}
423 
424 	bundle->alloc_error = 0;
425 
426 	for (i = 0; i < ARRAY_SIZE(bundle->conns); i++) {
427 		unsigned int shift = i * RXRPC_MAXCALLS;
428 		int j;
429 
430 		old = bundle->conns[i];
431 		if (!rxrpc_may_reuse_conn(old)) {
432 			if (old)
433 				trace_rxrpc_client(old, -1, rxrpc_client_replace);
434 			candidate->bundle_shift = shift;
435 			bundle->conns[i] = candidate;
436 			for (j = 0; j < RXRPC_MAXCALLS; j++)
437 				set_bit(shift + j, &bundle->avail_chans);
438 			candidate = NULL;
439 			break;
440 		}
441 
442 		old = NULL;
443 	}
444 
445 	spin_unlock(&bundle->channel_lock);
446 
447 	if (candidate) {
448 		_debug("discard C=%x", candidate->debug_id);
449 		trace_rxrpc_client(candidate, -1, rxrpc_client_duplicate);
450 		rxrpc_put_connection(candidate);
451 	}
452 
453 	rxrpc_put_connection(old);
454 	_leave("");
455 }
456 
457 /*
458  * Add a connection to a bundle if there are no usable connections or we have
459  * connections waiting for extra capacity.
460  */
461 static void rxrpc_maybe_add_conn(struct rxrpc_bundle *bundle, gfp_t gfp)
462 {
463 	struct rxrpc_call *call;
464 	int i, usable;
465 
466 	_enter("");
467 
468 	spin_lock(&bundle->channel_lock);
469 
470 	/* See if there are any usable connections. */
471 	usable = 0;
472 	for (i = 0; i < ARRAY_SIZE(bundle->conns); i++)
473 		if (rxrpc_may_reuse_conn(bundle->conns[i]))
474 			usable++;
475 
476 	if (!usable && !list_empty(&bundle->waiting_calls)) {
477 		call = list_first_entry(&bundle->waiting_calls,
478 					struct rxrpc_call, chan_wait_link);
479 		if (test_bit(RXRPC_CALL_UPGRADE, &call->flags))
480 			bundle->try_upgrade = true;
481 	}
482 
483 	if (!usable)
484 		goto alloc_conn;
485 
486 	if (!bundle->avail_chans &&
487 	    !bundle->try_upgrade &&
488 	    !list_empty(&bundle->waiting_calls) &&
489 	    usable < ARRAY_SIZE(bundle->conns))
490 		goto alloc_conn;
491 
492 	spin_unlock(&bundle->channel_lock);
493 	_leave("");
494 	return;
495 
496 alloc_conn:
497 	return rxrpc_add_conn_to_bundle(bundle, gfp);
498 }
499 
500 /*
501  * Assign a channel to the call at the front of the queue and wake the call up.
502  * We don't increment the callNumber counter until this number has been exposed
503  * to the world.
504  */
505 static void rxrpc_activate_one_channel(struct rxrpc_connection *conn,
506 				       unsigned int channel)
507 {
508 	struct rxrpc_channel *chan = &conn->channels[channel];
509 	struct rxrpc_bundle *bundle = conn->bundle;
510 	struct rxrpc_call *call = list_entry(bundle->waiting_calls.next,
511 					     struct rxrpc_call, chan_wait_link);
512 	u32 call_id = chan->call_counter + 1;
513 
514 	_enter("C=%x,%u", conn->debug_id, channel);
515 
516 	trace_rxrpc_client(conn, channel, rxrpc_client_chan_activate);
517 
518 	/* Cancel the final ACK on the previous call if it hasn't been sent yet
519 	 * as the DATA packet will implicitly ACK it.
520 	 */
521 	clear_bit(RXRPC_CONN_FINAL_ACK_0 + channel, &conn->flags);
522 	clear_bit(conn->bundle_shift + channel, &bundle->avail_chans);
523 
524 	rxrpc_see_call(call);
525 	list_del_init(&call->chan_wait_link);
526 	call->peer	= rxrpc_get_peer(conn->params.peer);
527 	call->conn	= rxrpc_get_connection(conn);
528 	call->cid	= conn->proto.cid | channel;
529 	call->call_id	= call_id;
530 	call->security	= conn->security;
531 	call->security_ix = conn->security_ix;
532 	call->service_id = conn->service_id;
533 
534 	trace_rxrpc_connect_call(call);
535 	_net("CONNECT call %08x:%08x as call %d on conn %d",
536 	     call->cid, call->call_id, call->debug_id, conn->debug_id);
537 
538 	write_lock_bh(&call->state_lock);
539 	call->state = RXRPC_CALL_CLIENT_SEND_REQUEST;
540 	write_unlock_bh(&call->state_lock);
541 
542 	/* Paired with the read barrier in rxrpc_connect_call().  This orders
543 	 * cid and epoch in the connection wrt to call_id without the need to
544 	 * take the channel_lock.
545 	 *
546 	 * We provisionally assign a callNumber at this point, but we don't
547 	 * confirm it until the call is about to be exposed.
548 	 *
549 	 * TODO: Pair with a barrier in the data_ready handler when that looks
550 	 * at the call ID through a connection channel.
551 	 */
552 	smp_wmb();
553 
554 	chan->call_id		= call_id;
555 	chan->call_debug_id	= call->debug_id;
556 	rcu_assign_pointer(chan->call, call);
557 	wake_up(&call->waitq);
558 }
559 
560 /*
561  * Remove a connection from the idle list if it's on it.
562  */
563 static void rxrpc_unidle_conn(struct rxrpc_bundle *bundle, struct rxrpc_connection *conn)
564 {
565 	struct rxrpc_net *rxnet = bundle->params.local->rxnet;
566 	bool drop_ref;
567 
568 	if (!list_empty(&conn->cache_link)) {
569 		drop_ref = false;
570 		spin_lock(&rxnet->client_conn_cache_lock);
571 		if (!list_empty(&conn->cache_link)) {
572 			list_del_init(&conn->cache_link);
573 			drop_ref = true;
574 		}
575 		spin_unlock(&rxnet->client_conn_cache_lock);
576 		if (drop_ref)
577 			rxrpc_put_connection(conn);
578 	}
579 }
580 
581 /*
582  * Assign channels and callNumbers to waiting calls with channel_lock
583  * held by caller.
584  */
585 static void rxrpc_activate_channels_locked(struct rxrpc_bundle *bundle)
586 {
587 	struct rxrpc_connection *conn;
588 	unsigned long avail, mask;
589 	unsigned int channel, slot;
590 
591 	if (bundle->try_upgrade)
592 		mask = 1;
593 	else
594 		mask = ULONG_MAX;
595 
596 	while (!list_empty(&bundle->waiting_calls)) {
597 		avail = bundle->avail_chans & mask;
598 		if (!avail)
599 			break;
600 		channel = __ffs(avail);
601 		clear_bit(channel, &bundle->avail_chans);
602 
603 		slot = channel / RXRPC_MAXCALLS;
604 		conn = bundle->conns[slot];
605 		if (!conn)
606 			break;
607 
608 		if (bundle->try_upgrade)
609 			set_bit(RXRPC_CONN_PROBING_FOR_UPGRADE, &conn->flags);
610 		rxrpc_unidle_conn(bundle, conn);
611 
612 		channel &= (RXRPC_MAXCALLS - 1);
613 		conn->act_chans	|= 1 << channel;
614 		rxrpc_activate_one_channel(conn, channel);
615 	}
616 }
617 
618 /*
619  * Assign channels and callNumbers to waiting calls.
620  */
621 static void rxrpc_activate_channels(struct rxrpc_bundle *bundle)
622 {
623 	_enter("B=%x", bundle->debug_id);
624 
625 	trace_rxrpc_client(NULL, -1, rxrpc_client_activate_chans);
626 
627 	if (!bundle->avail_chans)
628 		return;
629 
630 	spin_lock(&bundle->channel_lock);
631 	rxrpc_activate_channels_locked(bundle);
632 	spin_unlock(&bundle->channel_lock);
633 	_leave("");
634 }
635 
636 /*
637  * Wait for a callNumber and a channel to be granted to a call.
638  */
639 static int rxrpc_wait_for_channel(struct rxrpc_bundle *bundle,
640 				  struct rxrpc_call *call, gfp_t gfp)
641 {
642 	DECLARE_WAITQUEUE(myself, current);
643 	int ret = 0;
644 
645 	_enter("%d", call->debug_id);
646 
647 	if (!gfpflags_allow_blocking(gfp)) {
648 		rxrpc_maybe_add_conn(bundle, gfp);
649 		rxrpc_activate_channels(bundle);
650 		ret = bundle->alloc_error ?: -EAGAIN;
651 		goto out;
652 	}
653 
654 	add_wait_queue_exclusive(&call->waitq, &myself);
655 	for (;;) {
656 		rxrpc_maybe_add_conn(bundle, gfp);
657 		rxrpc_activate_channels(bundle);
658 		ret = bundle->alloc_error;
659 		if (ret < 0)
660 			break;
661 
662 		switch (call->interruptibility) {
663 		case RXRPC_INTERRUPTIBLE:
664 		case RXRPC_PREINTERRUPTIBLE:
665 			set_current_state(TASK_INTERRUPTIBLE);
666 			break;
667 		case RXRPC_UNINTERRUPTIBLE:
668 		default:
669 			set_current_state(TASK_UNINTERRUPTIBLE);
670 			break;
671 		}
672 		if (READ_ONCE(call->state) != RXRPC_CALL_CLIENT_AWAIT_CONN)
673 			break;
674 		if ((call->interruptibility == RXRPC_INTERRUPTIBLE ||
675 		     call->interruptibility == RXRPC_PREINTERRUPTIBLE) &&
676 		    signal_pending(current)) {
677 			ret = -ERESTARTSYS;
678 			break;
679 		}
680 		schedule();
681 	}
682 	remove_wait_queue(&call->waitq, &myself);
683 	__set_current_state(TASK_RUNNING);
684 
685 out:
686 	_leave(" = %d", ret);
687 	return ret;
688 }
689 
690 /*
691  * find a connection for a call
692  * - called in process context with IRQs enabled
693  */
694 int rxrpc_connect_call(struct rxrpc_sock *rx,
695 		       struct rxrpc_call *call,
696 		       struct rxrpc_conn_parameters *cp,
697 		       struct sockaddr_rxrpc *srx,
698 		       gfp_t gfp)
699 {
700 	struct rxrpc_bundle *bundle;
701 	struct rxrpc_net *rxnet = cp->local->rxnet;
702 	int ret = 0;
703 
704 	_enter("{%d,%lx},", call->debug_id, call->user_call_ID);
705 
706 	rxrpc_discard_expired_client_conns(&rxnet->client_conn_reaper);
707 
708 	bundle = rxrpc_prep_call(rx, call, cp, srx, gfp);
709 	if (IS_ERR(bundle)) {
710 		ret = PTR_ERR(bundle);
711 		goto out;
712 	}
713 
714 	if (call->state == RXRPC_CALL_CLIENT_AWAIT_CONN) {
715 		ret = rxrpc_wait_for_channel(bundle, call, gfp);
716 		if (ret < 0)
717 			goto wait_failed;
718 	}
719 
720 granted_channel:
721 	/* Paired with the write barrier in rxrpc_activate_one_channel(). */
722 	smp_rmb();
723 
724 out_put_bundle:
725 	rxrpc_put_bundle(bundle);
726 out:
727 	_leave(" = %d", ret);
728 	return ret;
729 
730 wait_failed:
731 	spin_lock(&bundle->channel_lock);
732 	list_del_init(&call->chan_wait_link);
733 	spin_unlock(&bundle->channel_lock);
734 
735 	if (call->state != RXRPC_CALL_CLIENT_AWAIT_CONN) {
736 		ret = 0;
737 		goto granted_channel;
738 	}
739 
740 	trace_rxrpc_client(call->conn, ret, rxrpc_client_chan_wait_failed);
741 	rxrpc_set_call_completion(call, RXRPC_CALL_LOCAL_ERROR, 0, ret);
742 	rxrpc_disconnect_client_call(bundle, call);
743 	goto out_put_bundle;
744 }
745 
746 /*
747  * Note that a call, and thus a connection, is about to be exposed to the
748  * world.
749  */
750 void rxrpc_expose_client_call(struct rxrpc_call *call)
751 {
752 	unsigned int channel = call->cid & RXRPC_CHANNELMASK;
753 	struct rxrpc_connection *conn = call->conn;
754 	struct rxrpc_channel *chan = &conn->channels[channel];
755 
756 	if (!test_and_set_bit(RXRPC_CALL_EXPOSED, &call->flags)) {
757 		/* Mark the call ID as being used.  If the callNumber counter
758 		 * exceeds ~2 billion, we kill the connection after its
759 		 * outstanding calls have finished so that the counter doesn't
760 		 * wrap.
761 		 */
762 		chan->call_counter++;
763 		if (chan->call_counter >= INT_MAX)
764 			set_bit(RXRPC_CONN_DONT_REUSE, &conn->flags);
765 		trace_rxrpc_client(conn, channel, rxrpc_client_exposed);
766 	}
767 }
768 
769 /*
770  * Set the reap timer.
771  */
772 static void rxrpc_set_client_reap_timer(struct rxrpc_net *rxnet)
773 {
774 	if (!rxnet->kill_all_client_conns) {
775 		unsigned long now = jiffies;
776 		unsigned long reap_at = now + rxrpc_conn_idle_client_expiry;
777 
778 		if (rxnet->live)
779 			timer_reduce(&rxnet->client_conn_reap_timer, reap_at);
780 	}
781 }
782 
783 /*
784  * Disconnect a client call.
785  */
786 void rxrpc_disconnect_client_call(struct rxrpc_bundle *bundle, struct rxrpc_call *call)
787 {
788 	struct rxrpc_connection *conn;
789 	struct rxrpc_channel *chan = NULL;
790 	struct rxrpc_net *rxnet = bundle->params.local->rxnet;
791 	unsigned int channel;
792 	bool may_reuse;
793 	u32 cid;
794 
795 	_enter("c=%x", call->debug_id);
796 
797 	spin_lock(&bundle->channel_lock);
798 	set_bit(RXRPC_CALL_DISCONNECTED, &call->flags);
799 
800 	/* Calls that have never actually been assigned a channel can simply be
801 	 * discarded.
802 	 */
803 	conn = call->conn;
804 	if (!conn) {
805 		_debug("call is waiting");
806 		ASSERTCMP(call->call_id, ==, 0);
807 		ASSERT(!test_bit(RXRPC_CALL_EXPOSED, &call->flags));
808 		list_del_init(&call->chan_wait_link);
809 		goto out;
810 	}
811 
812 	cid = call->cid;
813 	channel = cid & RXRPC_CHANNELMASK;
814 	chan = &conn->channels[channel];
815 	trace_rxrpc_client(conn, channel, rxrpc_client_chan_disconnect);
816 
817 	if (rcu_access_pointer(chan->call) != call) {
818 		spin_unlock(&bundle->channel_lock);
819 		BUG();
820 	}
821 
822 	may_reuse = rxrpc_may_reuse_conn(conn);
823 
824 	/* If a client call was exposed to the world, we save the result for
825 	 * retransmission.
826 	 *
827 	 * We use a barrier here so that the call number and abort code can be
828 	 * read without needing to take a lock.
829 	 *
830 	 * TODO: Make the incoming packet handler check this and handle
831 	 * terminal retransmission without requiring access to the call.
832 	 */
833 	if (test_bit(RXRPC_CALL_EXPOSED, &call->flags)) {
834 		_debug("exposed %u,%u", call->call_id, call->abort_code);
835 		__rxrpc_disconnect_call(conn, call);
836 
837 		if (test_and_clear_bit(RXRPC_CONN_PROBING_FOR_UPGRADE, &conn->flags)) {
838 			trace_rxrpc_client(conn, channel, rxrpc_client_to_active);
839 			bundle->try_upgrade = false;
840 			if (may_reuse)
841 				rxrpc_activate_channels_locked(bundle);
842 		}
843 
844 	}
845 
846 	/* See if we can pass the channel directly to another call. */
847 	if (may_reuse && !list_empty(&bundle->waiting_calls)) {
848 		trace_rxrpc_client(conn, channel, rxrpc_client_chan_pass);
849 		rxrpc_activate_one_channel(conn, channel);
850 		goto out;
851 	}
852 
853 	/* Schedule the final ACK to be transmitted in a short while so that it
854 	 * can be skipped if we find a follow-on call.  The first DATA packet
855 	 * of the follow on call will implicitly ACK this call.
856 	 */
857 	if (call->completion == RXRPC_CALL_SUCCEEDED &&
858 	    test_bit(RXRPC_CALL_EXPOSED, &call->flags)) {
859 		unsigned long final_ack_at = jiffies + 2;
860 
861 		WRITE_ONCE(chan->final_ack_at, final_ack_at);
862 		smp_wmb(); /* vs rxrpc_process_delayed_final_acks() */
863 		set_bit(RXRPC_CONN_FINAL_ACK_0 + channel, &conn->flags);
864 		rxrpc_reduce_conn_timer(conn, final_ack_at);
865 	}
866 
867 	/* Deactivate the channel. */
868 	rcu_assign_pointer(chan->call, NULL);
869 	set_bit(conn->bundle_shift + channel, &conn->bundle->avail_chans);
870 	conn->act_chans	&= ~(1 << channel);
871 
872 	/* If no channels remain active, then put the connection on the idle
873 	 * list for a short while.  Give it a ref to stop it going away if it
874 	 * becomes unbundled.
875 	 */
876 	if (!conn->act_chans) {
877 		trace_rxrpc_client(conn, channel, rxrpc_client_to_idle);
878 		conn->idle_timestamp = jiffies;
879 
880 		rxrpc_get_connection(conn);
881 		spin_lock(&rxnet->client_conn_cache_lock);
882 		list_move_tail(&conn->cache_link, &rxnet->idle_client_conns);
883 		spin_unlock(&rxnet->client_conn_cache_lock);
884 
885 		rxrpc_set_client_reap_timer(rxnet);
886 	}
887 
888 out:
889 	spin_unlock(&bundle->channel_lock);
890 	_leave("");
891 	return;
892 }
893 
894 /*
895  * Remove a connection from a bundle.
896  */
897 static void rxrpc_unbundle_conn(struct rxrpc_connection *conn)
898 {
899 	struct rxrpc_bundle *bundle = conn->bundle;
900 	struct rxrpc_local *local = bundle->params.local;
901 	unsigned int bindex;
902 	bool need_drop = false, need_put = false;
903 	int i;
904 
905 	_enter("C=%x", conn->debug_id);
906 
907 	if (conn->flags & RXRPC_CONN_FINAL_ACK_MASK)
908 		rxrpc_process_delayed_final_acks(conn, true);
909 
910 	spin_lock(&bundle->channel_lock);
911 	bindex = conn->bundle_shift / RXRPC_MAXCALLS;
912 	if (bundle->conns[bindex] == conn) {
913 		_debug("clear slot %u", bindex);
914 		bundle->conns[bindex] = NULL;
915 		for (i = 0; i < RXRPC_MAXCALLS; i++)
916 			clear_bit(conn->bundle_shift + i, &bundle->avail_chans);
917 		need_drop = true;
918 	}
919 	spin_unlock(&bundle->channel_lock);
920 
921 	/* If there are no more connections, remove the bundle */
922 	if (!bundle->avail_chans) {
923 		_debug("maybe unbundle");
924 		spin_lock(&local->client_bundles_lock);
925 
926 		for (i = 0; i < ARRAY_SIZE(bundle->conns); i++)
927 			if (bundle->conns[i])
928 				break;
929 		if (i == ARRAY_SIZE(bundle->conns) && !bundle->params.exclusive) {
930 			_debug("erase bundle");
931 			rb_erase(&bundle->local_node, &local->client_bundles);
932 			need_put = true;
933 		}
934 
935 		spin_unlock(&local->client_bundles_lock);
936 		if (need_put)
937 			rxrpc_put_bundle(bundle);
938 	}
939 
940 	if (need_drop)
941 		rxrpc_put_connection(conn);
942 	_leave("");
943 }
944 
945 /*
946  * Clean up a dead client connection.
947  */
948 static void rxrpc_kill_client_conn(struct rxrpc_connection *conn)
949 {
950 	struct rxrpc_local *local = conn->params.local;
951 	struct rxrpc_net *rxnet = local->rxnet;
952 
953 	_enter("C=%x", conn->debug_id);
954 
955 	trace_rxrpc_client(conn, -1, rxrpc_client_cleanup);
956 	atomic_dec(&rxnet->nr_client_conns);
957 
958 	rxrpc_put_client_connection_id(conn);
959 	rxrpc_kill_connection(conn);
960 }
961 
962 /*
963  * Clean up a dead client connections.
964  */
965 void rxrpc_put_client_conn(struct rxrpc_connection *conn)
966 {
967 	const void *here = __builtin_return_address(0);
968 	unsigned int debug_id = conn->debug_id;
969 	int n;
970 
971 	n = atomic_dec_return(&conn->usage);
972 	trace_rxrpc_conn(debug_id, rxrpc_conn_put_client, n, here);
973 	if (n <= 0) {
974 		ASSERTCMP(n, >=, 0);
975 		rxrpc_kill_client_conn(conn);
976 	}
977 }
978 
979 /*
980  * Discard expired client connections from the idle list.  Each conn in the
981  * idle list has been exposed and holds an extra ref because of that.
982  *
983  * This may be called from conn setup or from a work item so cannot be
984  * considered non-reentrant.
985  */
986 void rxrpc_discard_expired_client_conns(struct work_struct *work)
987 {
988 	struct rxrpc_connection *conn;
989 	struct rxrpc_net *rxnet =
990 		container_of(work, struct rxrpc_net, client_conn_reaper);
991 	unsigned long expiry, conn_expires_at, now;
992 	unsigned int nr_conns;
993 
994 	_enter("");
995 
996 	if (list_empty(&rxnet->idle_client_conns)) {
997 		_leave(" [empty]");
998 		return;
999 	}
1000 
1001 	/* Don't double up on the discarding */
1002 	if (!spin_trylock(&rxnet->client_conn_discard_lock)) {
1003 		_leave(" [already]");
1004 		return;
1005 	}
1006 
1007 	/* We keep an estimate of what the number of conns ought to be after
1008 	 * we've discarded some so that we don't overdo the discarding.
1009 	 */
1010 	nr_conns = atomic_read(&rxnet->nr_client_conns);
1011 
1012 next:
1013 	spin_lock(&rxnet->client_conn_cache_lock);
1014 
1015 	if (list_empty(&rxnet->idle_client_conns))
1016 		goto out;
1017 
1018 	conn = list_entry(rxnet->idle_client_conns.next,
1019 			  struct rxrpc_connection, cache_link);
1020 
1021 	if (!rxnet->kill_all_client_conns) {
1022 		/* If the number of connections is over the reap limit, we
1023 		 * expedite discard by reducing the expiry timeout.  We must,
1024 		 * however, have at least a short grace period to be able to do
1025 		 * final-ACK or ABORT retransmission.
1026 		 */
1027 		expiry = rxrpc_conn_idle_client_expiry;
1028 		if (nr_conns > rxrpc_reap_client_connections)
1029 			expiry = rxrpc_conn_idle_client_fast_expiry;
1030 		if (conn->params.local->service_closed)
1031 			expiry = rxrpc_closed_conn_expiry * HZ;
1032 
1033 		conn_expires_at = conn->idle_timestamp + expiry;
1034 
1035 		now = READ_ONCE(jiffies);
1036 		if (time_after(conn_expires_at, now))
1037 			goto not_yet_expired;
1038 	}
1039 
1040 	trace_rxrpc_client(conn, -1, rxrpc_client_discard);
1041 	list_del_init(&conn->cache_link);
1042 
1043 	spin_unlock(&rxnet->client_conn_cache_lock);
1044 
1045 	rxrpc_unbundle_conn(conn);
1046 	rxrpc_put_connection(conn); /* Drop the ->cache_link ref */
1047 
1048 	nr_conns--;
1049 	goto next;
1050 
1051 not_yet_expired:
1052 	/* The connection at the front of the queue hasn't yet expired, so
1053 	 * schedule the work item for that point if we discarded something.
1054 	 *
1055 	 * We don't worry if the work item is already scheduled - it can look
1056 	 * after rescheduling itself at a later time.  We could cancel it, but
1057 	 * then things get messier.
1058 	 */
1059 	_debug("not yet");
1060 	if (!rxnet->kill_all_client_conns)
1061 		timer_reduce(&rxnet->client_conn_reap_timer, conn_expires_at);
1062 
1063 out:
1064 	spin_unlock(&rxnet->client_conn_cache_lock);
1065 	spin_unlock(&rxnet->client_conn_discard_lock);
1066 	_leave("");
1067 }
1068 
1069 /*
1070  * Preemptively destroy all the client connection records rather than waiting
1071  * for them to time out
1072  */
1073 void rxrpc_destroy_all_client_connections(struct rxrpc_net *rxnet)
1074 {
1075 	_enter("");
1076 
1077 	spin_lock(&rxnet->client_conn_cache_lock);
1078 	rxnet->kill_all_client_conns = true;
1079 	spin_unlock(&rxnet->client_conn_cache_lock);
1080 
1081 	del_timer_sync(&rxnet->client_conn_reap_timer);
1082 
1083 	if (!rxrpc_queue_work(&rxnet->client_conn_reaper))
1084 		_debug("destroy: queue failed");
1085 
1086 	_leave("");
1087 }
1088 
1089 /*
1090  * Clean up the client connections on a local endpoint.
1091  */
1092 void rxrpc_clean_up_local_conns(struct rxrpc_local *local)
1093 {
1094 	struct rxrpc_connection *conn, *tmp;
1095 	struct rxrpc_net *rxnet = local->rxnet;
1096 	LIST_HEAD(graveyard);
1097 
1098 	_enter("");
1099 
1100 	spin_lock(&rxnet->client_conn_cache_lock);
1101 
1102 	list_for_each_entry_safe(conn, tmp, &rxnet->idle_client_conns,
1103 				 cache_link) {
1104 		if (conn->params.local == local) {
1105 			trace_rxrpc_client(conn, -1, rxrpc_client_discard);
1106 			list_move(&conn->cache_link, &graveyard);
1107 		}
1108 	}
1109 
1110 	spin_unlock(&rxnet->client_conn_cache_lock);
1111 
1112 	while (!list_empty(&graveyard)) {
1113 		conn = list_entry(graveyard.next,
1114 				  struct rxrpc_connection, cache_link);
1115 		list_del_init(&conn->cache_link);
1116 		rxrpc_unbundle_conn(conn);
1117 		rxrpc_put_connection(conn);
1118 	}
1119 
1120 	_leave(" [culled]");
1121 }
1122