1 // SPDX-License-Identifier: GPL-2.0-or-later
2 /* Client connection-specific management code.
4 * Copyright (C) 2016, 2020 Red Hat, Inc. All Rights Reserved.
5 * Written by David Howells (dhowells@redhat.com)
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.
11 * There are flags of relevance to the cache:
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.
17 * The caching state may only be changed if the cache lock is held.
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.
24 #define pr_fmt(fmt) KBUILD_MODNAME ": " fmt
26 #include <linux/slab.h>
27 #include <linux/idr.h>
28 #include <linux/timer.h>
29 #include <linux/sched/signal.h>
31 #include "ar-internal.h"
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;
38 * We use machine-unique IDs for our client connections.
40 DEFINE_IDR(rxrpc_client_conn_ids);
41 static DEFINE_SPINLOCK(rxrpc_conn_id_lock);
43 static void rxrpc_deactivate_bundle(struct rxrpc_bundle *bundle);
46 * Get a connection ID and epoch for a client connection from the global pool.
47 * The connection struct pointer is then recorded in the idr radix tree. The
48 * epoch doesn't change until the client is rebooted (or, at least, unless the
49 * module is unloaded).
51 static int rxrpc_get_client_connection_id(struct rxrpc_connection *conn,
54 struct rxrpc_net *rxnet = conn->params.local->rxnet;
60 spin_lock(&rxrpc_conn_id_lock);
62 id = idr_alloc_cyclic(&rxrpc_client_conn_ids, conn,
63 1, 0x40000000, GFP_NOWAIT);
67 spin_unlock(&rxrpc_conn_id_lock);
70 conn->proto.epoch = rxnet->epoch;
71 conn->proto.cid = id << RXRPC_CIDSHIFT;
72 set_bit(RXRPC_CONN_HAS_IDR, &conn->flags);
73 _leave(" [CID %x]", conn->proto.cid);
77 spin_unlock(&rxrpc_conn_id_lock);
84 * Release a connection ID for a client connection from the global pool.
86 static void rxrpc_put_client_connection_id(struct rxrpc_connection *conn)
88 if (test_bit(RXRPC_CONN_HAS_IDR, &conn->flags)) {
89 spin_lock(&rxrpc_conn_id_lock);
90 idr_remove(&rxrpc_client_conn_ids,
91 conn->proto.cid >> RXRPC_CIDSHIFT);
92 spin_unlock(&rxrpc_conn_id_lock);
97 * Destroy the client connection ID tree.
99 void rxrpc_destroy_client_conn_ids(void)
101 struct rxrpc_connection *conn;
104 if (!idr_is_empty(&rxrpc_client_conn_ids)) {
105 idr_for_each_entry(&rxrpc_client_conn_ids, conn, id) {
106 pr_err("AF_RXRPC: Leaked client conn %p {%d}\n",
107 conn, refcount_read(&conn->ref));
112 idr_destroy(&rxrpc_client_conn_ids);
116 * Allocate a connection bundle.
118 static struct rxrpc_bundle *rxrpc_alloc_bundle(struct rxrpc_conn_parameters *cp,
121 struct rxrpc_bundle *bundle;
123 bundle = kzalloc(sizeof(*bundle), gfp);
125 bundle->params = *cp;
126 rxrpc_get_peer(bundle->params.peer);
127 refcount_set(&bundle->ref, 1);
128 atomic_set(&bundle->active, 1);
129 spin_lock_init(&bundle->channel_lock);
130 INIT_LIST_HEAD(&bundle->waiting_calls);
135 struct rxrpc_bundle *rxrpc_get_bundle(struct rxrpc_bundle *bundle)
137 refcount_inc(&bundle->ref);
141 static void rxrpc_free_bundle(struct rxrpc_bundle *bundle)
143 rxrpc_put_peer(bundle->params.peer);
147 void rxrpc_put_bundle(struct rxrpc_bundle *bundle)
149 unsigned int d = bundle->debug_id;
153 dead = __refcount_dec_and_test(&bundle->ref, &r);
155 _debug("PUT B=%x %d", d, r - 1);
157 rxrpc_free_bundle(bundle);
161 * Allocate a client connection.
163 static struct rxrpc_connection *
164 rxrpc_alloc_client_connection(struct rxrpc_bundle *bundle, gfp_t gfp)
166 struct rxrpc_connection *conn;
167 struct rxrpc_net *rxnet = bundle->params.local->rxnet;
172 conn = rxrpc_alloc_connection(gfp);
174 _leave(" = -ENOMEM");
175 return ERR_PTR(-ENOMEM);
178 refcount_set(&conn->ref, 1);
179 conn->bundle = bundle;
180 conn->params = bundle->params;
181 conn->out_clientflag = RXRPC_CLIENT_INITIATED;
182 conn->state = RXRPC_CONN_CLIENT;
183 conn->service_id = conn->params.service_id;
185 ret = rxrpc_get_client_connection_id(conn, gfp);
189 ret = rxrpc_init_client_conn_security(conn);
193 atomic_inc(&rxnet->nr_conns);
194 write_lock(&rxnet->conn_lock);
195 list_add_tail(&conn->proc_link, &rxnet->conn_proc_list);
196 write_unlock(&rxnet->conn_lock);
198 rxrpc_get_bundle(bundle);
199 rxrpc_get_peer(conn->params.peer);
200 rxrpc_get_local(conn->params.local);
201 key_get(conn->params.key);
203 trace_rxrpc_conn(conn->debug_id, rxrpc_conn_new_client,
204 refcount_read(&conn->ref),
205 __builtin_return_address(0));
207 atomic_inc(&rxnet->nr_client_conns);
208 trace_rxrpc_client(conn, -1, rxrpc_client_alloc);
209 _leave(" = %p", conn);
213 rxrpc_put_client_connection_id(conn);
216 _leave(" = %d", ret);
221 * Determine if a connection may be reused.
223 static bool rxrpc_may_reuse_conn(struct rxrpc_connection *conn)
225 struct rxrpc_net *rxnet;
226 int id_cursor, id, distance, limit;
231 rxnet = conn->params.local->rxnet;
232 if (test_bit(RXRPC_CONN_DONT_REUSE, &conn->flags))
235 if (conn->state != RXRPC_CONN_CLIENT ||
236 conn->proto.epoch != rxnet->epoch)
237 goto mark_dont_reuse;
239 /* The IDR tree gets very expensive on memory if the connection IDs are
240 * widely scattered throughout the number space, so we shall want to
241 * kill off connections that, say, have an ID more than about four
242 * times the maximum number of client conns away from the current
243 * allocation point to try and keep the IDs concentrated.
245 id_cursor = idr_get_cursor(&rxrpc_client_conn_ids);
246 id = conn->proto.cid >> RXRPC_CIDSHIFT;
247 distance = id - id_cursor;
249 distance = -distance;
250 limit = max_t(unsigned long, atomic_read(&rxnet->nr_conns) * 4, 1024);
251 if (distance > limit)
252 goto mark_dont_reuse;
257 set_bit(RXRPC_CONN_DONT_REUSE, &conn->flags);
263 * Look up the conn bundle that matches the connection parameters, adding it if
264 * it doesn't yet exist.
266 static struct rxrpc_bundle *rxrpc_look_up_bundle(struct rxrpc_conn_parameters *cp,
269 static atomic_t rxrpc_bundle_id;
270 struct rxrpc_bundle *bundle, *candidate;
271 struct rxrpc_local *local = cp->local;
272 struct rb_node *p, **pp, *parent;
275 _enter("{%px,%x,%u,%u}",
276 cp->peer, key_serial(cp->key), cp->security_level, cp->upgrade);
279 return rxrpc_alloc_bundle(cp, gfp);
281 /* First, see if the bundle is already there. */
283 spin_lock(&local->client_bundles_lock);
284 p = local->client_bundles.rb_node;
286 bundle = rb_entry(p, struct rxrpc_bundle, local_node);
288 #define cmp(X) ((long)bundle->params.X - (long)cp->X)
291 cmp(security_level) ?:
301 spin_unlock(&local->client_bundles_lock);
304 /* It wasn't. We need to add one. */
305 candidate = rxrpc_alloc_bundle(cp, gfp);
310 spin_lock(&local->client_bundles_lock);
311 pp = &local->client_bundles.rb_node;
315 bundle = rb_entry(parent, struct rxrpc_bundle, local_node);
317 #define cmp(X) ((long)bundle->params.X - (long)cp->X)
320 cmp(security_level) ?:
324 pp = &(*pp)->rb_left;
326 pp = &(*pp)->rb_right;
328 goto found_bundle_free;
331 _debug("new bundle");
332 candidate->debug_id = atomic_inc_return(&rxrpc_bundle_id);
333 rb_link_node(&candidate->local_node, parent, pp);
334 rb_insert_color(&candidate->local_node, &local->client_bundles);
335 rxrpc_get_bundle(candidate);
336 spin_unlock(&local->client_bundles_lock);
337 _leave(" = %u [new]", candidate->debug_id);
341 rxrpc_free_bundle(candidate);
343 rxrpc_get_bundle(bundle);
344 atomic_inc(&bundle->active);
345 spin_unlock(&local->client_bundles_lock);
346 _leave(" = %u [found]", bundle->debug_id);
351 * Create or find a client bundle to use for a call.
353 * If we return with a connection, the call will be on its waiting list. It's
354 * left to the caller to assign a channel and wake up the call.
356 static struct rxrpc_bundle *rxrpc_prep_call(struct rxrpc_sock *rx,
357 struct rxrpc_call *call,
358 struct rxrpc_conn_parameters *cp,
359 struct sockaddr_rxrpc *srx,
362 struct rxrpc_bundle *bundle;
364 _enter("{%d,%lx},", call->debug_id, call->user_call_ID);
366 cp->peer = rxrpc_lookup_peer(rx, cp->local, srx, gfp);
370 call->tx_last_sent = ktime_get_real();
371 call->cong_ssthresh = cp->peer->cong_ssthresh;
372 if (call->cong_cwnd >= call->cong_ssthresh)
373 call->cong_mode = RXRPC_CALL_CONGEST_AVOIDANCE;
375 call->cong_mode = RXRPC_CALL_SLOW_START;
377 __set_bit(RXRPC_CALL_UPGRADE, &call->flags);
379 /* Find the client connection bundle. */
380 bundle = rxrpc_look_up_bundle(cp, gfp);
384 /* Get this call queued. Someone else may activate it whilst we're
385 * lining up a new connection, but that's fine.
387 spin_lock(&bundle->channel_lock);
388 list_add_tail(&call->chan_wait_link, &bundle->waiting_calls);
389 spin_unlock(&bundle->channel_lock);
391 _leave(" = [B=%x]", bundle->debug_id);
395 _leave(" = -ENOMEM");
396 return ERR_PTR(-ENOMEM);
400 * Allocate a new connection and add it into a bundle.
402 static void rxrpc_add_conn_to_bundle(struct rxrpc_bundle *bundle, gfp_t gfp)
403 __releases(bundle->channel_lock)
405 struct rxrpc_connection *candidate = NULL, *old = NULL;
411 conflict = bundle->alloc_conn;
413 bundle->alloc_conn = true;
414 spin_unlock(&bundle->channel_lock);
420 candidate = rxrpc_alloc_client_connection(bundle, gfp);
422 spin_lock(&bundle->channel_lock);
423 bundle->alloc_conn = false;
425 if (IS_ERR(candidate)) {
426 bundle->alloc_error = PTR_ERR(candidate);
427 spin_unlock(&bundle->channel_lock);
428 _leave(" [err %ld]", PTR_ERR(candidate));
432 bundle->alloc_error = 0;
434 for (i = 0; i < ARRAY_SIZE(bundle->conns); i++) {
435 unsigned int shift = i * RXRPC_MAXCALLS;
438 old = bundle->conns[i];
439 if (!rxrpc_may_reuse_conn(old)) {
441 trace_rxrpc_client(old, -1, rxrpc_client_replace);
442 candidate->bundle_shift = shift;
443 atomic_inc(&bundle->active);
444 bundle->conns[i] = candidate;
445 for (j = 0; j < RXRPC_MAXCALLS; j++)
446 set_bit(shift + j, &bundle->avail_chans);
454 spin_unlock(&bundle->channel_lock);
457 _debug("discard C=%x", candidate->debug_id);
458 trace_rxrpc_client(candidate, -1, rxrpc_client_duplicate);
459 rxrpc_put_connection(candidate);
462 rxrpc_put_connection(old);
467 * Add a connection to a bundle if there are no usable connections or we have
468 * connections waiting for extra capacity.
470 static void rxrpc_maybe_add_conn(struct rxrpc_bundle *bundle, gfp_t gfp)
472 struct rxrpc_call *call;
477 spin_lock(&bundle->channel_lock);
479 /* See if there are any usable connections. */
481 for (i = 0; i < ARRAY_SIZE(bundle->conns); i++)
482 if (rxrpc_may_reuse_conn(bundle->conns[i]))
485 if (!usable && !list_empty(&bundle->waiting_calls)) {
486 call = list_first_entry(&bundle->waiting_calls,
487 struct rxrpc_call, chan_wait_link);
488 if (test_bit(RXRPC_CALL_UPGRADE, &call->flags))
489 bundle->try_upgrade = true;
495 if (!bundle->avail_chans &&
496 !bundle->try_upgrade &&
497 !list_empty(&bundle->waiting_calls) &&
498 usable < ARRAY_SIZE(bundle->conns))
501 spin_unlock(&bundle->channel_lock);
506 return rxrpc_add_conn_to_bundle(bundle, gfp);
510 * Assign a channel to the call at the front of the queue and wake the call up.
511 * We don't increment the callNumber counter until this number has been exposed
514 static void rxrpc_activate_one_channel(struct rxrpc_connection *conn,
515 unsigned int channel)
517 struct rxrpc_channel *chan = &conn->channels[channel];
518 struct rxrpc_bundle *bundle = conn->bundle;
519 struct rxrpc_call *call = list_entry(bundle->waiting_calls.next,
520 struct rxrpc_call, chan_wait_link);
521 u32 call_id = chan->call_counter + 1;
523 _enter("C=%x,%u", conn->debug_id, channel);
525 trace_rxrpc_client(conn, channel, rxrpc_client_chan_activate);
527 /* Cancel the final ACK on the previous call if it hasn't been sent yet
528 * as the DATA packet will implicitly ACK it.
530 clear_bit(RXRPC_CONN_FINAL_ACK_0 + channel, &conn->flags);
531 clear_bit(conn->bundle_shift + channel, &bundle->avail_chans);
533 rxrpc_see_call(call);
534 list_del_init(&call->chan_wait_link);
535 call->peer = rxrpc_get_peer(conn->params.peer);
536 call->conn = rxrpc_get_connection(conn);
537 call->cid = conn->proto.cid | channel;
538 call->call_id = call_id;
539 call->security = conn->security;
540 call->security_ix = conn->security_ix;
541 call->service_id = conn->service_id;
543 trace_rxrpc_connect_call(call);
545 write_lock_bh(&call->state_lock);
546 call->state = RXRPC_CALL_CLIENT_SEND_REQUEST;
547 write_unlock_bh(&call->state_lock);
549 /* Paired with the read barrier in rxrpc_connect_call(). This orders
550 * cid and epoch in the connection wrt to call_id without the need to
551 * take the channel_lock.
553 * We provisionally assign a callNumber at this point, but we don't
554 * confirm it until the call is about to be exposed.
556 * TODO: Pair with a barrier in the data_ready handler when that looks
557 * at the call ID through a connection channel.
561 chan->call_id = call_id;
562 chan->call_debug_id = call->debug_id;
563 rcu_assign_pointer(chan->call, call);
564 wake_up(&call->waitq);
568 * Remove a connection from the idle list if it's on it.
570 static void rxrpc_unidle_conn(struct rxrpc_bundle *bundle, struct rxrpc_connection *conn)
572 struct rxrpc_net *rxnet = bundle->params.local->rxnet;
575 if (!list_empty(&conn->cache_link)) {
577 spin_lock(&rxnet->client_conn_cache_lock);
578 if (!list_empty(&conn->cache_link)) {
579 list_del_init(&conn->cache_link);
582 spin_unlock(&rxnet->client_conn_cache_lock);
584 rxrpc_put_connection(conn);
589 * Assign channels and callNumbers to waiting calls with channel_lock
592 static void rxrpc_activate_channels_locked(struct rxrpc_bundle *bundle)
594 struct rxrpc_connection *conn;
595 unsigned long avail, mask;
596 unsigned int channel, slot;
598 if (bundle->try_upgrade)
603 while (!list_empty(&bundle->waiting_calls)) {
604 avail = bundle->avail_chans & mask;
607 channel = __ffs(avail);
608 clear_bit(channel, &bundle->avail_chans);
610 slot = channel / RXRPC_MAXCALLS;
611 conn = bundle->conns[slot];
615 if (bundle->try_upgrade)
616 set_bit(RXRPC_CONN_PROBING_FOR_UPGRADE, &conn->flags);
617 rxrpc_unidle_conn(bundle, conn);
619 channel &= (RXRPC_MAXCALLS - 1);
620 conn->act_chans |= 1 << channel;
621 rxrpc_activate_one_channel(conn, channel);
626 * Assign channels and callNumbers to waiting calls.
628 static void rxrpc_activate_channels(struct rxrpc_bundle *bundle)
630 _enter("B=%x", bundle->debug_id);
632 trace_rxrpc_client(NULL, -1, rxrpc_client_activate_chans);
634 if (!bundle->avail_chans)
637 spin_lock(&bundle->channel_lock);
638 rxrpc_activate_channels_locked(bundle);
639 spin_unlock(&bundle->channel_lock);
644 * Wait for a callNumber and a channel to be granted to a call.
646 static int rxrpc_wait_for_channel(struct rxrpc_bundle *bundle,
647 struct rxrpc_call *call, gfp_t gfp)
649 DECLARE_WAITQUEUE(myself, current);
652 _enter("%d", call->debug_id);
654 if (!gfpflags_allow_blocking(gfp)) {
655 rxrpc_maybe_add_conn(bundle, gfp);
656 rxrpc_activate_channels(bundle);
657 ret = bundle->alloc_error ?: -EAGAIN;
661 add_wait_queue_exclusive(&call->waitq, &myself);
663 rxrpc_maybe_add_conn(bundle, gfp);
664 rxrpc_activate_channels(bundle);
665 ret = bundle->alloc_error;
669 switch (call->interruptibility) {
670 case RXRPC_INTERRUPTIBLE:
671 case RXRPC_PREINTERRUPTIBLE:
672 set_current_state(TASK_INTERRUPTIBLE);
674 case RXRPC_UNINTERRUPTIBLE:
676 set_current_state(TASK_UNINTERRUPTIBLE);
679 if (READ_ONCE(call->state) != RXRPC_CALL_CLIENT_AWAIT_CONN)
681 if ((call->interruptibility == RXRPC_INTERRUPTIBLE ||
682 call->interruptibility == RXRPC_PREINTERRUPTIBLE) &&
683 signal_pending(current)) {
689 remove_wait_queue(&call->waitq, &myself);
690 __set_current_state(TASK_RUNNING);
693 _leave(" = %d", ret);
698 * find a connection for a call
699 * - called in process context with IRQs enabled
701 int rxrpc_connect_call(struct rxrpc_sock *rx,
702 struct rxrpc_call *call,
703 struct rxrpc_conn_parameters *cp,
704 struct sockaddr_rxrpc *srx,
707 struct rxrpc_bundle *bundle;
708 struct rxrpc_net *rxnet = cp->local->rxnet;
711 _enter("{%d,%lx},", call->debug_id, call->user_call_ID);
713 rxrpc_discard_expired_client_conns(&rxnet->client_conn_reaper);
715 bundle = rxrpc_prep_call(rx, call, cp, srx, gfp);
716 if (IS_ERR(bundle)) {
717 ret = PTR_ERR(bundle);
721 if (call->state == RXRPC_CALL_CLIENT_AWAIT_CONN) {
722 ret = rxrpc_wait_for_channel(bundle, call, gfp);
728 /* Paired with the write barrier in rxrpc_activate_one_channel(). */
732 rxrpc_deactivate_bundle(bundle);
733 rxrpc_put_bundle(bundle);
735 _leave(" = %d", ret);
739 spin_lock(&bundle->channel_lock);
740 list_del_init(&call->chan_wait_link);
741 spin_unlock(&bundle->channel_lock);
743 if (call->state != RXRPC_CALL_CLIENT_AWAIT_CONN) {
745 goto granted_channel;
748 trace_rxrpc_client(call->conn, ret, rxrpc_client_chan_wait_failed);
749 rxrpc_set_call_completion(call, RXRPC_CALL_LOCAL_ERROR, 0, ret);
750 rxrpc_disconnect_client_call(bundle, call);
755 * Note that a call, and thus a connection, is about to be exposed to the
758 void rxrpc_expose_client_call(struct rxrpc_call *call)
760 unsigned int channel = call->cid & RXRPC_CHANNELMASK;
761 struct rxrpc_connection *conn = call->conn;
762 struct rxrpc_channel *chan = &conn->channels[channel];
764 if (!test_and_set_bit(RXRPC_CALL_EXPOSED, &call->flags)) {
765 /* Mark the call ID as being used. If the callNumber counter
766 * exceeds ~2 billion, we kill the connection after its
767 * outstanding calls have finished so that the counter doesn't
770 chan->call_counter++;
771 if (chan->call_counter >= INT_MAX)
772 set_bit(RXRPC_CONN_DONT_REUSE, &conn->flags);
773 trace_rxrpc_client(conn, channel, rxrpc_client_exposed);
778 * Set the reap timer.
780 static void rxrpc_set_client_reap_timer(struct rxrpc_net *rxnet)
782 if (!rxnet->kill_all_client_conns) {
783 unsigned long now = jiffies;
784 unsigned long reap_at = now + rxrpc_conn_idle_client_expiry;
787 timer_reduce(&rxnet->client_conn_reap_timer, reap_at);
792 * Disconnect a client call.
794 void rxrpc_disconnect_client_call(struct rxrpc_bundle *bundle, struct rxrpc_call *call)
796 struct rxrpc_connection *conn;
797 struct rxrpc_channel *chan = NULL;
798 struct rxrpc_net *rxnet = bundle->params.local->rxnet;
799 unsigned int channel;
803 _enter("c=%x", call->debug_id);
805 spin_lock(&bundle->channel_lock);
806 set_bit(RXRPC_CALL_DISCONNECTED, &call->flags);
808 /* Calls that have never actually been assigned a channel can simply be
813 _debug("call is waiting");
814 ASSERTCMP(call->call_id, ==, 0);
815 ASSERT(!test_bit(RXRPC_CALL_EXPOSED, &call->flags));
816 list_del_init(&call->chan_wait_link);
821 channel = cid & RXRPC_CHANNELMASK;
822 chan = &conn->channels[channel];
823 trace_rxrpc_client(conn, channel, rxrpc_client_chan_disconnect);
825 if (rcu_access_pointer(chan->call) != call) {
826 spin_unlock(&bundle->channel_lock);
830 may_reuse = rxrpc_may_reuse_conn(conn);
832 /* If a client call was exposed to the world, we save the result for
835 * We use a barrier here so that the call number and abort code can be
836 * read without needing to take a lock.
838 * TODO: Make the incoming packet handler check this and handle
839 * terminal retransmission without requiring access to the call.
841 if (test_bit(RXRPC_CALL_EXPOSED, &call->flags)) {
842 _debug("exposed %u,%u", call->call_id, call->abort_code);
843 __rxrpc_disconnect_call(conn, call);
845 if (test_and_clear_bit(RXRPC_CONN_PROBING_FOR_UPGRADE, &conn->flags)) {
846 trace_rxrpc_client(conn, channel, rxrpc_client_to_active);
847 bundle->try_upgrade = false;
849 rxrpc_activate_channels_locked(bundle);
854 /* See if we can pass the channel directly to another call. */
855 if (may_reuse && !list_empty(&bundle->waiting_calls)) {
856 trace_rxrpc_client(conn, channel, rxrpc_client_chan_pass);
857 rxrpc_activate_one_channel(conn, channel);
861 /* Schedule the final ACK to be transmitted in a short while so that it
862 * can be skipped if we find a follow-on call. The first DATA packet
863 * of the follow on call will implicitly ACK this call.
865 if (call->completion == RXRPC_CALL_SUCCEEDED &&
866 test_bit(RXRPC_CALL_EXPOSED, &call->flags)) {
867 unsigned long final_ack_at = jiffies + 2;
869 WRITE_ONCE(chan->final_ack_at, final_ack_at);
870 smp_wmb(); /* vs rxrpc_process_delayed_final_acks() */
871 set_bit(RXRPC_CONN_FINAL_ACK_0 + channel, &conn->flags);
872 rxrpc_reduce_conn_timer(conn, final_ack_at);
875 /* Deactivate the channel. */
876 rcu_assign_pointer(chan->call, NULL);
877 set_bit(conn->bundle_shift + channel, &conn->bundle->avail_chans);
878 conn->act_chans &= ~(1 << channel);
880 /* If no channels remain active, then put the connection on the idle
881 * list for a short while. Give it a ref to stop it going away if it
884 if (!conn->act_chans) {
885 trace_rxrpc_client(conn, channel, rxrpc_client_to_idle);
886 conn->idle_timestamp = jiffies;
888 rxrpc_get_connection(conn);
889 spin_lock(&rxnet->client_conn_cache_lock);
890 list_move_tail(&conn->cache_link, &rxnet->idle_client_conns);
891 spin_unlock(&rxnet->client_conn_cache_lock);
893 rxrpc_set_client_reap_timer(rxnet);
897 spin_unlock(&bundle->channel_lock);
903 * Remove a connection from a bundle.
905 static void rxrpc_unbundle_conn(struct rxrpc_connection *conn)
907 struct rxrpc_bundle *bundle = conn->bundle;
909 bool need_drop = false;
912 _enter("C=%x", conn->debug_id);
914 if (conn->flags & RXRPC_CONN_FINAL_ACK_MASK)
915 rxrpc_process_delayed_final_acks(conn, true);
917 spin_lock(&bundle->channel_lock);
918 bindex = conn->bundle_shift / RXRPC_MAXCALLS;
919 if (bundle->conns[bindex] == conn) {
920 _debug("clear slot %u", bindex);
921 bundle->conns[bindex] = NULL;
922 for (i = 0; i < RXRPC_MAXCALLS; i++)
923 clear_bit(conn->bundle_shift + i, &bundle->avail_chans);
926 spin_unlock(&bundle->channel_lock);
929 rxrpc_deactivate_bundle(bundle);
930 rxrpc_put_connection(conn);
935 * Drop the active count on a bundle.
937 static void rxrpc_deactivate_bundle(struct rxrpc_bundle *bundle)
939 struct rxrpc_local *local = bundle->params.local;
940 bool need_put = false;
942 if (atomic_dec_and_lock(&bundle->active, &local->client_bundles_lock)) {
943 if (!bundle->params.exclusive) {
944 _debug("erase bundle");
945 rb_erase(&bundle->local_node, &local->client_bundles);
949 spin_unlock(&local->client_bundles_lock);
951 rxrpc_put_bundle(bundle);
956 * Clean up a dead client connection.
958 static void rxrpc_kill_client_conn(struct rxrpc_connection *conn)
960 struct rxrpc_local *local = conn->params.local;
961 struct rxrpc_net *rxnet = local->rxnet;
963 _enter("C=%x", conn->debug_id);
965 trace_rxrpc_client(conn, -1, rxrpc_client_cleanup);
966 atomic_dec(&rxnet->nr_client_conns);
968 rxrpc_put_client_connection_id(conn);
969 rxrpc_kill_connection(conn);
973 * Clean up a dead client connections.
975 void rxrpc_put_client_conn(struct rxrpc_connection *conn)
977 const void *here = __builtin_return_address(0);
978 unsigned int debug_id = conn->debug_id;
982 dead = __refcount_dec_and_test(&conn->ref, &r);
983 trace_rxrpc_conn(debug_id, rxrpc_conn_put_client, r - 1, here);
985 rxrpc_kill_client_conn(conn);
989 * Discard expired client connections from the idle list. Each conn in the
990 * idle list has been exposed and holds an extra ref because of that.
992 * This may be called from conn setup or from a work item so cannot be
993 * considered non-reentrant.
995 void rxrpc_discard_expired_client_conns(struct work_struct *work)
997 struct rxrpc_connection *conn;
998 struct rxrpc_net *rxnet =
999 container_of(work, struct rxrpc_net, client_conn_reaper);
1000 unsigned long expiry, conn_expires_at, now;
1001 unsigned int nr_conns;
1005 if (list_empty(&rxnet->idle_client_conns)) {
1010 /* Don't double up on the discarding */
1011 if (!spin_trylock(&rxnet->client_conn_discard_lock)) {
1012 _leave(" [already]");
1016 /* We keep an estimate of what the number of conns ought to be after
1017 * we've discarded some so that we don't overdo the discarding.
1019 nr_conns = atomic_read(&rxnet->nr_client_conns);
1022 spin_lock(&rxnet->client_conn_cache_lock);
1024 if (list_empty(&rxnet->idle_client_conns))
1027 conn = list_entry(rxnet->idle_client_conns.next,
1028 struct rxrpc_connection, cache_link);
1030 if (!rxnet->kill_all_client_conns) {
1031 /* If the number of connections is over the reap limit, we
1032 * expedite discard by reducing the expiry timeout. We must,
1033 * however, have at least a short grace period to be able to do
1034 * final-ACK or ABORT retransmission.
1036 expiry = rxrpc_conn_idle_client_expiry;
1037 if (nr_conns > rxrpc_reap_client_connections)
1038 expiry = rxrpc_conn_idle_client_fast_expiry;
1039 if (conn->params.local->service_closed)
1040 expiry = rxrpc_closed_conn_expiry * HZ;
1042 conn_expires_at = conn->idle_timestamp + expiry;
1044 now = READ_ONCE(jiffies);
1045 if (time_after(conn_expires_at, now))
1046 goto not_yet_expired;
1049 trace_rxrpc_client(conn, -1, rxrpc_client_discard);
1050 list_del_init(&conn->cache_link);
1052 spin_unlock(&rxnet->client_conn_cache_lock);
1054 rxrpc_unbundle_conn(conn);
1055 rxrpc_put_connection(conn); /* Drop the ->cache_link ref */
1061 /* The connection at the front of the queue hasn't yet expired, so
1062 * schedule the work item for that point if we discarded something.
1064 * We don't worry if the work item is already scheduled - it can look
1065 * after rescheduling itself at a later time. We could cancel it, but
1066 * then things get messier.
1069 if (!rxnet->kill_all_client_conns)
1070 timer_reduce(&rxnet->client_conn_reap_timer, conn_expires_at);
1073 spin_unlock(&rxnet->client_conn_cache_lock);
1074 spin_unlock(&rxnet->client_conn_discard_lock);
1079 * Preemptively destroy all the client connection records rather than waiting
1080 * for them to time out
1082 void rxrpc_destroy_all_client_connections(struct rxrpc_net *rxnet)
1086 spin_lock(&rxnet->client_conn_cache_lock);
1087 rxnet->kill_all_client_conns = true;
1088 spin_unlock(&rxnet->client_conn_cache_lock);
1090 del_timer_sync(&rxnet->client_conn_reap_timer);
1092 if (!rxrpc_queue_work(&rxnet->client_conn_reaper))
1093 _debug("destroy: queue failed");
1099 * Clean up the client connections on a local endpoint.
1101 void rxrpc_clean_up_local_conns(struct rxrpc_local *local)
1103 struct rxrpc_connection *conn, *tmp;
1104 struct rxrpc_net *rxnet = local->rxnet;
1105 LIST_HEAD(graveyard);
1109 spin_lock(&rxnet->client_conn_cache_lock);
1111 list_for_each_entry_safe(conn, tmp, &rxnet->idle_client_conns,
1113 if (conn->params.local == local) {
1114 trace_rxrpc_client(conn, -1, rxrpc_client_discard);
1115 list_move(&conn->cache_link, &graveyard);
1119 spin_unlock(&rxnet->client_conn_cache_lock);
1121 while (!list_empty(&graveyard)) {
1122 conn = list_entry(graveyard.next,
1123 struct rxrpc_connection, cache_link);
1124 list_del_init(&conn->cache_link);
1125 rxrpc_unbundle_conn(conn);
1126 rxrpc_put_connection(conn);
1129 _leave(" [culled]");