Loading...
Note: File does not exist in v3.1.
1/* Client connection-specific management code.
2 *
3 * Copyright (C) 2016 Red Hat, Inc. All Rights Reserved.
4 * Written by David Howells (dhowells@redhat.com)
5 *
6 * This program is free software; you can redistribute it and/or
7 * modify it under the terms of the GNU General Public Licence
8 * as published by the Free Software Foundation; either version
9 * 2 of the Licence, or (at your option) any later version.
10 *
11 *
12 * Client connections need to be cached for a little while after they've made a
13 * call so as to handle retransmitted DATA packets in case the server didn't
14 * receive the final ACK or terminating ABORT we sent it.
15 *
16 * Client connections can be in one of a number of cache states:
17 *
18 * (1) INACTIVE - The connection is not held in any list and may not have been
19 * exposed to the world. If it has been previously exposed, it was
20 * discarded from the idle list after expiring.
21 *
22 * (2) WAITING - The connection is waiting for the number of client conns to
23 * drop below the maximum capacity. Calls may be in progress upon it from
24 * when it was active and got culled.
25 *
26 * The connection is on the rxrpc_waiting_client_conns list which is kept
27 * in to-be-granted order. Culled conns with waiters go to the back of
28 * the queue just like new conns.
29 *
30 * (3) ACTIVE - The connection has at least one call in progress upon it, it
31 * may freely grant available channels to new calls and calls may be
32 * waiting on it for channels to become available.
33 *
34 * The connection is on the rxrpc_active_client_conns list which is kept
35 * in activation order for culling purposes.
36 *
37 * rxrpc_nr_active_client_conns is held incremented also.
38 *
39 * (4) CULLED - The connection got summarily culled to try and free up
40 * capacity. Calls currently in progress on the connection are allowed to
41 * continue, but new calls will have to wait. There can be no waiters in
42 * this state - the conn would have to go to the WAITING state instead.
43 *
44 * (5) IDLE - The connection has no calls in progress upon it and must have
45 * been exposed to the world (ie. the EXPOSED flag must be set). When it
46 * expires, the EXPOSED flag is cleared and the connection transitions to
47 * the INACTIVE state.
48 *
49 * The connection is on the rxrpc_idle_client_conns list which is kept in
50 * order of how soon they'll expire.
51 *
52 * There are flags of relevance to the cache:
53 *
54 * (1) EXPOSED - The connection ID got exposed to the world. If this flag is
55 * set, an extra ref is added to the connection preventing it from being
56 * reaped when it has no calls outstanding. This flag is cleared and the
57 * ref dropped when a conn is discarded from the idle list.
58 *
59 * This allows us to move terminal call state retransmission to the
60 * connection and to discard the call immediately we think it is done
61 * with. It also give us a chance to reuse the connection.
62 *
63 * (2) DONT_REUSE - The connection should be discarded as soon as possible and
64 * should not be reused. This is set when an exclusive connection is used
65 * or a call ID counter overflows.
66 *
67 * The caching state may only be changed if the cache lock is held.
68 *
69 * There are two idle client connection expiry durations. If the total number
70 * of connections is below the reap threshold, we use the normal duration; if
71 * it's above, we use the fast duration.
72 */
73
74#define pr_fmt(fmt) KBUILD_MODNAME ": " fmt
75
76#include <linux/slab.h>
77#include <linux/idr.h>
78#include <linux/timer.h>
79#include "ar-internal.h"
80
81__read_mostly unsigned int rxrpc_max_client_connections = 1000;
82__read_mostly unsigned int rxrpc_reap_client_connections = 900;
83__read_mostly unsigned int rxrpc_conn_idle_client_expiry = 2 * 60 * HZ;
84__read_mostly unsigned int rxrpc_conn_idle_client_fast_expiry = 2 * HZ;
85
86static unsigned int rxrpc_nr_client_conns;
87static unsigned int rxrpc_nr_active_client_conns;
88static __read_mostly bool rxrpc_kill_all_client_conns;
89
90static DEFINE_SPINLOCK(rxrpc_client_conn_cache_lock);
91static DEFINE_SPINLOCK(rxrpc_client_conn_discard_mutex);
92static LIST_HEAD(rxrpc_waiting_client_conns);
93static LIST_HEAD(rxrpc_active_client_conns);
94static LIST_HEAD(rxrpc_idle_client_conns);
95
96/*
97 * We use machine-unique IDs for our client connections.
98 */
99DEFINE_IDR(rxrpc_client_conn_ids);
100static DEFINE_SPINLOCK(rxrpc_conn_id_lock);
101
102static void rxrpc_cull_active_client_conns(void);
103static void rxrpc_discard_expired_client_conns(struct work_struct *);
104
105static DECLARE_DELAYED_WORK(rxrpc_client_conn_reap,
106 rxrpc_discard_expired_client_conns);
107
108const char rxrpc_conn_cache_states[RXRPC_CONN__NR_CACHE_STATES][5] = {
109 [RXRPC_CONN_CLIENT_INACTIVE] = "Inac",
110 [RXRPC_CONN_CLIENT_WAITING] = "Wait",
111 [RXRPC_CONN_CLIENT_ACTIVE] = "Actv",
112 [RXRPC_CONN_CLIENT_CULLED] = "Cull",
113 [RXRPC_CONN_CLIENT_IDLE] = "Idle",
114};
115
116/*
117 * Get a connection ID and epoch for a client connection from the global pool.
118 * The connection struct pointer is then recorded in the idr radix tree. The
119 * epoch doesn't change until the client is rebooted (or, at least, unless the
120 * module is unloaded).
121 */
122static int rxrpc_get_client_connection_id(struct rxrpc_connection *conn,
123 gfp_t gfp)
124{
125 int id;
126
127 _enter("");
128
129 idr_preload(gfp);
130 spin_lock(&rxrpc_conn_id_lock);
131
132 id = idr_alloc_cyclic(&rxrpc_client_conn_ids, conn,
133 1, 0x40000000, GFP_NOWAIT);
134 if (id < 0)
135 goto error;
136
137 spin_unlock(&rxrpc_conn_id_lock);
138 idr_preload_end();
139
140 conn->proto.epoch = rxrpc_epoch;
141 conn->proto.cid = id << RXRPC_CIDSHIFT;
142 set_bit(RXRPC_CONN_HAS_IDR, &conn->flags);
143 _leave(" [CID %x]", conn->proto.cid);
144 return 0;
145
146error:
147 spin_unlock(&rxrpc_conn_id_lock);
148 idr_preload_end();
149 _leave(" = %d", id);
150 return id;
151}
152
153/*
154 * Release a connection ID for a client connection from the global pool.
155 */
156static void rxrpc_put_client_connection_id(struct rxrpc_connection *conn)
157{
158 if (test_bit(RXRPC_CONN_HAS_IDR, &conn->flags)) {
159 spin_lock(&rxrpc_conn_id_lock);
160 idr_remove(&rxrpc_client_conn_ids,
161 conn->proto.cid >> RXRPC_CIDSHIFT);
162 spin_unlock(&rxrpc_conn_id_lock);
163 }
164}
165
166/*
167 * Destroy the client connection ID tree.
168 */
169void rxrpc_destroy_client_conn_ids(void)
170{
171 struct rxrpc_connection *conn;
172 int id;
173
174 if (!idr_is_empty(&rxrpc_client_conn_ids)) {
175 idr_for_each_entry(&rxrpc_client_conn_ids, conn, id) {
176 pr_err("AF_RXRPC: Leaked client conn %p {%d}\n",
177 conn, atomic_read(&conn->usage));
178 }
179 BUG();
180 }
181
182 idr_destroy(&rxrpc_client_conn_ids);
183}
184
185/*
186 * Allocate a client connection.
187 */
188static struct rxrpc_connection *
189rxrpc_alloc_client_connection(struct rxrpc_conn_parameters *cp, gfp_t gfp)
190{
191 struct rxrpc_connection *conn;
192 int ret;
193
194 _enter("");
195
196 conn = rxrpc_alloc_connection(gfp);
197 if (!conn) {
198 _leave(" = -ENOMEM");
199 return ERR_PTR(-ENOMEM);
200 }
201
202 atomic_set(&conn->usage, 1);
203 if (cp->exclusive)
204 __set_bit(RXRPC_CONN_DONT_REUSE, &conn->flags);
205
206 conn->params = *cp;
207 conn->out_clientflag = RXRPC_CLIENT_INITIATED;
208 conn->state = RXRPC_CONN_CLIENT;
209
210 ret = rxrpc_get_client_connection_id(conn, gfp);
211 if (ret < 0)
212 goto error_0;
213
214 ret = rxrpc_init_client_conn_security(conn);
215 if (ret < 0)
216 goto error_1;
217
218 ret = conn->security->prime_packet_security(conn);
219 if (ret < 0)
220 goto error_2;
221
222 write_lock(&rxrpc_connection_lock);
223 list_add_tail(&conn->proc_link, &rxrpc_connection_proc_list);
224 write_unlock(&rxrpc_connection_lock);
225
226 /* We steal the caller's peer ref. */
227 cp->peer = NULL;
228 rxrpc_get_local(conn->params.local);
229 key_get(conn->params.key);
230
231 trace_rxrpc_conn(conn, rxrpc_conn_new_client, atomic_read(&conn->usage),
232 __builtin_return_address(0));
233 trace_rxrpc_client(conn, -1, rxrpc_client_alloc);
234 _leave(" = %p", conn);
235 return conn;
236
237error_2:
238 conn->security->clear(conn);
239error_1:
240 rxrpc_put_client_connection_id(conn);
241error_0:
242 kfree(conn);
243 _leave(" = %d", ret);
244 return ERR_PTR(ret);
245}
246
247/*
248 * Determine if a connection may be reused.
249 */
250static bool rxrpc_may_reuse_conn(struct rxrpc_connection *conn)
251{
252 int id_cursor, id, distance, limit;
253
254 if (test_bit(RXRPC_CONN_DONT_REUSE, &conn->flags))
255 goto dont_reuse;
256
257 if (conn->proto.epoch != rxrpc_epoch)
258 goto mark_dont_reuse;
259
260 /* The IDR tree gets very expensive on memory if the connection IDs are
261 * widely scattered throughout the number space, so we shall want to
262 * kill off connections that, say, have an ID more than about four
263 * times the maximum number of client conns away from the current
264 * allocation point to try and keep the IDs concentrated.
265 */
266 id_cursor = idr_get_cursor(&rxrpc_client_conn_ids);
267 id = conn->proto.cid >> RXRPC_CIDSHIFT;
268 distance = id - id_cursor;
269 if (distance < 0)
270 distance = -distance;
271 limit = max(rxrpc_max_client_connections * 4, 1024U);
272 if (distance > limit)
273 goto mark_dont_reuse;
274
275 return true;
276
277mark_dont_reuse:
278 set_bit(RXRPC_CONN_DONT_REUSE, &conn->flags);
279dont_reuse:
280 return false;
281}
282
283/*
284 * Create or find a client connection to use for a call.
285 *
286 * If we return with a connection, the call will be on its waiting list. It's
287 * left to the caller to assign a channel and wake up the call.
288 */
289static int rxrpc_get_client_conn(struct rxrpc_call *call,
290 struct rxrpc_conn_parameters *cp,
291 struct sockaddr_rxrpc *srx,
292 gfp_t gfp)
293{
294 struct rxrpc_connection *conn, *candidate = NULL;
295 struct rxrpc_local *local = cp->local;
296 struct rb_node *p, **pp, *parent;
297 long diff;
298 int ret = -ENOMEM;
299
300 _enter("{%d,%lx},", call->debug_id, call->user_call_ID);
301
302 cp->peer = rxrpc_lookup_peer(cp->local, srx, gfp);
303 if (!cp->peer)
304 goto error;
305
306 /* If the connection is not meant to be exclusive, search the available
307 * connections to see if the connection we want to use already exists.
308 */
309 if (!cp->exclusive) {
310 _debug("search 1");
311 spin_lock(&local->client_conns_lock);
312 p = local->client_conns.rb_node;
313 while (p) {
314 conn = rb_entry(p, struct rxrpc_connection, client_node);
315
316#define cmp(X) ((long)conn->params.X - (long)cp->X)
317 diff = (cmp(peer) ?:
318 cmp(key) ?:
319 cmp(security_level));
320#undef cmp
321 if (diff < 0) {
322 p = p->rb_left;
323 } else if (diff > 0) {
324 p = p->rb_right;
325 } else {
326 if (rxrpc_may_reuse_conn(conn) &&
327 rxrpc_get_connection_maybe(conn))
328 goto found_extant_conn;
329 /* The connection needs replacing. It's better
330 * to effect that when we have something to
331 * replace it with so that we don't have to
332 * rebalance the tree twice.
333 */
334 break;
335 }
336 }
337 spin_unlock(&local->client_conns_lock);
338 }
339
340 /* There wasn't a connection yet or we need an exclusive connection.
341 * We need to create a candidate and then potentially redo the search
342 * in case we're racing with another thread also trying to connect on a
343 * shareable connection.
344 */
345 _debug("new conn");
346 candidate = rxrpc_alloc_client_connection(cp, gfp);
347 if (IS_ERR(candidate)) {
348 ret = PTR_ERR(candidate);
349 goto error_peer;
350 }
351
352 /* Add the call to the new connection's waiting list in case we're
353 * going to have to wait for the connection to come live. It's our
354 * connection, so we want first dibs on the channel slots. We would
355 * normally have to take channel_lock but we do this before anyone else
356 * can see the connection.
357 */
358 list_add_tail(&call->chan_wait_link, &candidate->waiting_calls);
359
360 if (cp->exclusive) {
361 call->conn = candidate;
362 call->security_ix = candidate->security_ix;
363 _leave(" = 0 [exclusive %d]", candidate->debug_id);
364 return 0;
365 }
366
367 /* Publish the new connection for userspace to find. We need to redo
368 * the search before doing this lest we race with someone else adding a
369 * conflicting instance.
370 */
371 _debug("search 2");
372 spin_lock(&local->client_conns_lock);
373
374 pp = &local->client_conns.rb_node;
375 parent = NULL;
376 while (*pp) {
377 parent = *pp;
378 conn = rb_entry(parent, struct rxrpc_connection, client_node);
379
380#define cmp(X) ((long)conn->params.X - (long)candidate->params.X)
381 diff = (cmp(peer) ?:
382 cmp(key) ?:
383 cmp(security_level));
384#undef cmp
385 if (diff < 0) {
386 pp = &(*pp)->rb_left;
387 } else if (diff > 0) {
388 pp = &(*pp)->rb_right;
389 } else {
390 if (rxrpc_may_reuse_conn(conn) &&
391 rxrpc_get_connection_maybe(conn))
392 goto found_extant_conn;
393 /* The old connection is from an outdated epoch. */
394 _debug("replace conn");
395 clear_bit(RXRPC_CONN_IN_CLIENT_CONNS, &conn->flags);
396 rb_replace_node(&conn->client_node,
397 &candidate->client_node,
398 &local->client_conns);
399 trace_rxrpc_client(conn, -1, rxrpc_client_replace);
400 goto candidate_published;
401 }
402 }
403
404 _debug("new conn");
405 rb_link_node(&candidate->client_node, parent, pp);
406 rb_insert_color(&candidate->client_node, &local->client_conns);
407
408candidate_published:
409 set_bit(RXRPC_CONN_IN_CLIENT_CONNS, &candidate->flags);
410 call->conn = candidate;
411 call->security_ix = candidate->security_ix;
412 spin_unlock(&local->client_conns_lock);
413 _leave(" = 0 [new %d]", candidate->debug_id);
414 return 0;
415
416 /* We come here if we found a suitable connection already in existence.
417 * Discard any candidate we may have allocated, and try to get a
418 * channel on this one.
419 */
420found_extant_conn:
421 _debug("found conn");
422 spin_unlock(&local->client_conns_lock);
423
424 if (candidate) {
425 trace_rxrpc_client(candidate, -1, rxrpc_client_duplicate);
426 rxrpc_put_connection(candidate);
427 candidate = NULL;
428 }
429
430 spin_lock(&conn->channel_lock);
431 call->conn = conn;
432 call->security_ix = conn->security_ix;
433 list_add(&call->chan_wait_link, &conn->waiting_calls);
434 spin_unlock(&conn->channel_lock);
435 _leave(" = 0 [extant %d]", conn->debug_id);
436 return 0;
437
438error_peer:
439 rxrpc_put_peer(cp->peer);
440 cp->peer = NULL;
441error:
442 _leave(" = %d", ret);
443 return ret;
444}
445
446/*
447 * Activate a connection.
448 */
449static void rxrpc_activate_conn(struct rxrpc_connection *conn)
450{
451 trace_rxrpc_client(conn, -1, rxrpc_client_to_active);
452 conn->cache_state = RXRPC_CONN_CLIENT_ACTIVE;
453 rxrpc_nr_active_client_conns++;
454 list_move_tail(&conn->cache_link, &rxrpc_active_client_conns);
455}
456
457/*
458 * Attempt to animate a connection for a new call.
459 *
460 * If it's not exclusive, the connection is in the endpoint tree, and we're in
461 * the conn's list of those waiting to grab a channel. There is, however, a
462 * limit on the number of live connections allowed at any one time, so we may
463 * have to wait for capacity to become available.
464 *
465 * Note that a connection on the waiting queue might *also* have active
466 * channels if it has been culled to make space and then re-requested by a new
467 * call.
468 */
469static void rxrpc_animate_client_conn(struct rxrpc_connection *conn)
470{
471 unsigned int nr_conns;
472
473 _enter("%d,%d", conn->debug_id, conn->cache_state);
474
475 if (conn->cache_state == RXRPC_CONN_CLIENT_ACTIVE)
476 goto out;
477
478 spin_lock(&rxrpc_client_conn_cache_lock);
479
480 nr_conns = rxrpc_nr_client_conns;
481 if (!test_and_set_bit(RXRPC_CONN_COUNTED, &conn->flags)) {
482 trace_rxrpc_client(conn, -1, rxrpc_client_count);
483 rxrpc_nr_client_conns = nr_conns + 1;
484 }
485
486 switch (conn->cache_state) {
487 case RXRPC_CONN_CLIENT_ACTIVE:
488 case RXRPC_CONN_CLIENT_WAITING:
489 break;
490
491 case RXRPC_CONN_CLIENT_INACTIVE:
492 case RXRPC_CONN_CLIENT_CULLED:
493 case RXRPC_CONN_CLIENT_IDLE:
494 if (nr_conns >= rxrpc_max_client_connections)
495 goto wait_for_capacity;
496 goto activate_conn;
497
498 default:
499 BUG();
500 }
501
502out_unlock:
503 spin_unlock(&rxrpc_client_conn_cache_lock);
504out:
505 _leave(" [%d]", conn->cache_state);
506 return;
507
508activate_conn:
509 _debug("activate");
510 rxrpc_activate_conn(conn);
511 goto out_unlock;
512
513wait_for_capacity:
514 _debug("wait");
515 trace_rxrpc_client(conn, -1, rxrpc_client_to_waiting);
516 conn->cache_state = RXRPC_CONN_CLIENT_WAITING;
517 list_move_tail(&conn->cache_link, &rxrpc_waiting_client_conns);
518 goto out_unlock;
519}
520
521/*
522 * Deactivate a channel.
523 */
524static void rxrpc_deactivate_one_channel(struct rxrpc_connection *conn,
525 unsigned int channel)
526{
527 struct rxrpc_channel *chan = &conn->channels[channel];
528
529 rcu_assign_pointer(chan->call, NULL);
530 conn->active_chans &= ~(1 << channel);
531}
532
533/*
534 * Assign a channel to the call at the front of the queue and wake the call up.
535 * We don't increment the callNumber counter until this number has been exposed
536 * to the world.
537 */
538static void rxrpc_activate_one_channel(struct rxrpc_connection *conn,
539 unsigned int channel)
540{
541 struct rxrpc_channel *chan = &conn->channels[channel];
542 struct rxrpc_call *call = list_entry(conn->waiting_calls.next,
543 struct rxrpc_call, chan_wait_link);
544 u32 call_id = chan->call_counter + 1;
545
546 trace_rxrpc_client(conn, channel, rxrpc_client_chan_activate);
547
548 write_lock_bh(&call->state_lock);
549 call->state = RXRPC_CALL_CLIENT_SEND_REQUEST;
550 write_unlock_bh(&call->state_lock);
551
552 rxrpc_see_call(call);
553 list_del_init(&call->chan_wait_link);
554 conn->active_chans |= 1 << channel;
555 call->peer = rxrpc_get_peer(conn->params.peer);
556 call->cid = conn->proto.cid | channel;
557 call->call_id = call_id;
558
559 _net("CONNECT call %08x:%08x as call %d on conn %d",
560 call->cid, call->call_id, call->debug_id, conn->debug_id);
561
562 /* Paired with the read barrier in rxrpc_wait_for_channel(). This
563 * orders cid and epoch in the connection wrt to call_id without the
564 * need to take the channel_lock.
565 *
566 * We provisionally assign a callNumber at this point, but we don't
567 * confirm it until the call is about to be exposed.
568 *
569 * TODO: Pair with a barrier in the data_ready handler when that looks
570 * at the call ID through a connection channel.
571 */
572 smp_wmb();
573 chan->call_id = call_id;
574 rcu_assign_pointer(chan->call, call);
575 wake_up(&call->waitq);
576}
577
578/*
579 * Assign channels and callNumbers to waiting calls with channel_lock
580 * held by caller.
581 */
582static void rxrpc_activate_channels_locked(struct rxrpc_connection *conn)
583{
584 u8 avail, mask;
585
586 switch (conn->cache_state) {
587 case RXRPC_CONN_CLIENT_ACTIVE:
588 mask = RXRPC_ACTIVE_CHANS_MASK;
589 break;
590 default:
591 return;
592 }
593
594 while (!list_empty(&conn->waiting_calls) &&
595 (avail = ~conn->active_chans,
596 avail &= mask,
597 avail != 0))
598 rxrpc_activate_one_channel(conn, __ffs(avail));
599}
600
601/*
602 * Assign channels and callNumbers to waiting calls.
603 */
604static void rxrpc_activate_channels(struct rxrpc_connection *conn)
605{
606 _enter("%d", conn->debug_id);
607
608 trace_rxrpc_client(conn, -1, rxrpc_client_activate_chans);
609
610 if (conn->active_chans == RXRPC_ACTIVE_CHANS_MASK)
611 return;
612
613 spin_lock(&conn->channel_lock);
614 rxrpc_activate_channels_locked(conn);
615 spin_unlock(&conn->channel_lock);
616 _leave("");
617}
618
619/*
620 * Wait for a callNumber and a channel to be granted to a call.
621 */
622static int rxrpc_wait_for_channel(struct rxrpc_call *call, gfp_t gfp)
623{
624 int ret = 0;
625
626 _enter("%d", call->debug_id);
627
628 if (!call->call_id) {
629 DECLARE_WAITQUEUE(myself, current);
630
631 if (!gfpflags_allow_blocking(gfp)) {
632 ret = -EAGAIN;
633 goto out;
634 }
635
636 add_wait_queue_exclusive(&call->waitq, &myself);
637 for (;;) {
638 set_current_state(TASK_INTERRUPTIBLE);
639 if (call->call_id)
640 break;
641 if (signal_pending(current)) {
642 ret = -ERESTARTSYS;
643 break;
644 }
645 schedule();
646 }
647 remove_wait_queue(&call->waitq, &myself);
648 __set_current_state(TASK_RUNNING);
649 }
650
651 /* Paired with the write barrier in rxrpc_activate_one_channel(). */
652 smp_rmb();
653
654out:
655 _leave(" = %d", ret);
656 return ret;
657}
658
659/*
660 * find a connection for a call
661 * - called in process context with IRQs enabled
662 */
663int rxrpc_connect_call(struct rxrpc_call *call,
664 struct rxrpc_conn_parameters *cp,
665 struct sockaddr_rxrpc *srx,
666 gfp_t gfp)
667{
668 int ret;
669
670 _enter("{%d,%lx},", call->debug_id, call->user_call_ID);
671
672 rxrpc_discard_expired_client_conns(NULL);
673 rxrpc_cull_active_client_conns();
674
675 ret = rxrpc_get_client_conn(call, cp, srx, gfp);
676 if (ret < 0)
677 return ret;
678
679 rxrpc_animate_client_conn(call->conn);
680 rxrpc_activate_channels(call->conn);
681
682 ret = rxrpc_wait_for_channel(call, gfp);
683 if (ret < 0)
684 rxrpc_disconnect_client_call(call);
685
686 _leave(" = %d", ret);
687 return ret;
688}
689
690/*
691 * Note that a connection is about to be exposed to the world. Once it is
692 * exposed, we maintain an extra ref on it that stops it from being summarily
693 * discarded before it's (a) had a chance to deal with retransmission and (b)
694 * had a chance at re-use (the per-connection security negotiation is
695 * expensive).
696 */
697static void rxrpc_expose_client_conn(struct rxrpc_connection *conn,
698 unsigned int channel)
699{
700 if (!test_and_set_bit(RXRPC_CONN_EXPOSED, &conn->flags)) {
701 trace_rxrpc_client(conn, channel, rxrpc_client_exposed);
702 rxrpc_get_connection(conn);
703 }
704}
705
706/*
707 * Note that a call, and thus a connection, is about to be exposed to the
708 * world.
709 */
710void rxrpc_expose_client_call(struct rxrpc_call *call)
711{
712 unsigned int channel = call->cid & RXRPC_CHANNELMASK;
713 struct rxrpc_connection *conn = call->conn;
714 struct rxrpc_channel *chan = &conn->channels[channel];
715
716 if (!test_and_set_bit(RXRPC_CALL_EXPOSED, &call->flags)) {
717 /* Mark the call ID as being used. If the callNumber counter
718 * exceeds ~2 billion, we kill the connection after its
719 * outstanding calls have finished so that the counter doesn't
720 * wrap.
721 */
722 chan->call_counter++;
723 if (chan->call_counter >= INT_MAX)
724 set_bit(RXRPC_CONN_DONT_REUSE, &conn->flags);
725 rxrpc_expose_client_conn(conn, channel);
726 }
727}
728
729/*
730 * Disconnect a client call.
731 */
732void rxrpc_disconnect_client_call(struct rxrpc_call *call)
733{
734 unsigned int channel = call->cid & RXRPC_CHANNELMASK;
735 struct rxrpc_connection *conn = call->conn;
736 struct rxrpc_channel *chan = &conn->channels[channel];
737
738 trace_rxrpc_client(conn, channel, rxrpc_client_chan_disconnect);
739 call->conn = NULL;
740
741 spin_lock(&conn->channel_lock);
742
743 /* Calls that have never actually been assigned a channel can simply be
744 * discarded. If the conn didn't get used either, it will follow
745 * immediately unless someone else grabs it in the meantime.
746 */
747 if (!list_empty(&call->chan_wait_link)) {
748 _debug("call is waiting");
749 ASSERTCMP(call->call_id, ==, 0);
750 ASSERT(!test_bit(RXRPC_CALL_EXPOSED, &call->flags));
751 list_del_init(&call->chan_wait_link);
752
753 trace_rxrpc_client(conn, channel, rxrpc_client_chan_unstarted);
754
755 /* We must deactivate or idle the connection if it's now
756 * waiting for nothing.
757 */
758 spin_lock(&rxrpc_client_conn_cache_lock);
759 if (conn->cache_state == RXRPC_CONN_CLIENT_WAITING &&
760 list_empty(&conn->waiting_calls) &&
761 !conn->active_chans)
762 goto idle_connection;
763 goto out;
764 }
765
766 ASSERTCMP(rcu_access_pointer(chan->call), ==, call);
767
768 /* If a client call was exposed to the world, we save the result for
769 * retransmission.
770 *
771 * We use a barrier here so that the call number and abort code can be
772 * read without needing to take a lock.
773 *
774 * TODO: Make the incoming packet handler check this and handle
775 * terminal retransmission without requiring access to the call.
776 */
777 if (test_bit(RXRPC_CALL_EXPOSED, &call->flags)) {
778 _debug("exposed %u,%u", call->call_id, call->abort_code);
779 __rxrpc_disconnect_call(conn, call);
780 }
781
782 /* See if we can pass the channel directly to another call. */
783 if (conn->cache_state == RXRPC_CONN_CLIENT_ACTIVE &&
784 !list_empty(&conn->waiting_calls)) {
785 trace_rxrpc_client(conn, channel, rxrpc_client_chan_pass);
786 rxrpc_activate_one_channel(conn, channel);
787 goto out_2;
788 }
789
790 /* Things are more complex and we need the cache lock. We might be
791 * able to simply idle the conn or it might now be lurking on the wait
792 * list. It might even get moved back to the active list whilst we're
793 * waiting for the lock.
794 */
795 spin_lock(&rxrpc_client_conn_cache_lock);
796
797 switch (conn->cache_state) {
798 case RXRPC_CONN_CLIENT_ACTIVE:
799 if (list_empty(&conn->waiting_calls)) {
800 rxrpc_deactivate_one_channel(conn, channel);
801 if (!conn->active_chans) {
802 rxrpc_nr_active_client_conns--;
803 goto idle_connection;
804 }
805 goto out;
806 }
807
808 trace_rxrpc_client(conn, channel, rxrpc_client_chan_pass);
809 rxrpc_activate_one_channel(conn, channel);
810 goto out;
811
812 case RXRPC_CONN_CLIENT_CULLED:
813 rxrpc_deactivate_one_channel(conn, channel);
814 ASSERT(list_empty(&conn->waiting_calls));
815 if (!conn->active_chans)
816 goto idle_connection;
817 goto out;
818
819 case RXRPC_CONN_CLIENT_WAITING:
820 rxrpc_deactivate_one_channel(conn, channel);
821 goto out;
822
823 default:
824 BUG();
825 }
826
827out:
828 spin_unlock(&rxrpc_client_conn_cache_lock);
829out_2:
830 spin_unlock(&conn->channel_lock);
831 rxrpc_put_connection(conn);
832 _leave("");
833 return;
834
835idle_connection:
836 /* As no channels remain active, the connection gets deactivated
837 * immediately or moved to the idle list for a short while.
838 */
839 if (test_bit(RXRPC_CONN_EXPOSED, &conn->flags)) {
840 trace_rxrpc_client(conn, channel, rxrpc_client_to_idle);
841 conn->idle_timestamp = jiffies;
842 conn->cache_state = RXRPC_CONN_CLIENT_IDLE;
843 list_move_tail(&conn->cache_link, &rxrpc_idle_client_conns);
844 if (rxrpc_idle_client_conns.next == &conn->cache_link &&
845 !rxrpc_kill_all_client_conns)
846 queue_delayed_work(rxrpc_workqueue,
847 &rxrpc_client_conn_reap,
848 rxrpc_conn_idle_client_expiry);
849 } else {
850 trace_rxrpc_client(conn, channel, rxrpc_client_to_inactive);
851 conn->cache_state = RXRPC_CONN_CLIENT_INACTIVE;
852 list_del_init(&conn->cache_link);
853 }
854 goto out;
855}
856
857/*
858 * Clean up a dead client connection.
859 */
860static struct rxrpc_connection *
861rxrpc_put_one_client_conn(struct rxrpc_connection *conn)
862{
863 struct rxrpc_connection *next = NULL;
864 struct rxrpc_local *local = conn->params.local;
865 unsigned int nr_conns;
866
867 trace_rxrpc_client(conn, -1, rxrpc_client_cleanup);
868
869 if (test_bit(RXRPC_CONN_IN_CLIENT_CONNS, &conn->flags)) {
870 spin_lock(&local->client_conns_lock);
871 if (test_and_clear_bit(RXRPC_CONN_IN_CLIENT_CONNS,
872 &conn->flags))
873 rb_erase(&conn->client_node, &local->client_conns);
874 spin_unlock(&local->client_conns_lock);
875 }
876
877 rxrpc_put_client_connection_id(conn);
878
879 ASSERTCMP(conn->cache_state, ==, RXRPC_CONN_CLIENT_INACTIVE);
880
881 if (test_bit(RXRPC_CONN_COUNTED, &conn->flags)) {
882 trace_rxrpc_client(conn, -1, rxrpc_client_uncount);
883 spin_lock(&rxrpc_client_conn_cache_lock);
884 nr_conns = --rxrpc_nr_client_conns;
885
886 if (nr_conns < rxrpc_max_client_connections &&
887 !list_empty(&rxrpc_waiting_client_conns)) {
888 next = list_entry(rxrpc_waiting_client_conns.next,
889 struct rxrpc_connection, cache_link);
890 rxrpc_get_connection(next);
891 rxrpc_activate_conn(next);
892 }
893
894 spin_unlock(&rxrpc_client_conn_cache_lock);
895 }
896
897 rxrpc_kill_connection(conn);
898 if (next)
899 rxrpc_activate_channels(next);
900
901 /* We need to get rid of the temporary ref we took upon next, but we
902 * can't call rxrpc_put_connection() recursively.
903 */
904 return next;
905}
906
907/*
908 * Clean up a dead client connections.
909 */
910void rxrpc_put_client_conn(struct rxrpc_connection *conn)
911{
912 const void *here = __builtin_return_address(0);
913 int n;
914
915 do {
916 n = atomic_dec_return(&conn->usage);
917 trace_rxrpc_conn(conn, rxrpc_conn_put_client, n, here);
918 if (n > 0)
919 return;
920 ASSERTCMP(n, >=, 0);
921
922 conn = rxrpc_put_one_client_conn(conn);
923 } while (conn);
924}
925
926/*
927 * Kill the longest-active client connections to make room for new ones.
928 */
929static void rxrpc_cull_active_client_conns(void)
930{
931 struct rxrpc_connection *conn;
932 unsigned int nr_conns = rxrpc_nr_client_conns;
933 unsigned int nr_active, limit;
934
935 _enter("");
936
937 ASSERTCMP(nr_conns, >=, 0);
938 if (nr_conns < rxrpc_max_client_connections) {
939 _leave(" [ok]");
940 return;
941 }
942 limit = rxrpc_reap_client_connections;
943
944 spin_lock(&rxrpc_client_conn_cache_lock);
945 nr_active = rxrpc_nr_active_client_conns;
946
947 while (nr_active > limit) {
948 ASSERT(!list_empty(&rxrpc_active_client_conns));
949 conn = list_entry(rxrpc_active_client_conns.next,
950 struct rxrpc_connection, cache_link);
951 ASSERTCMP(conn->cache_state, ==, RXRPC_CONN_CLIENT_ACTIVE);
952
953 if (list_empty(&conn->waiting_calls)) {
954 trace_rxrpc_client(conn, -1, rxrpc_client_to_culled);
955 conn->cache_state = RXRPC_CONN_CLIENT_CULLED;
956 list_del_init(&conn->cache_link);
957 } else {
958 trace_rxrpc_client(conn, -1, rxrpc_client_to_waiting);
959 conn->cache_state = RXRPC_CONN_CLIENT_WAITING;
960 list_move_tail(&conn->cache_link,
961 &rxrpc_waiting_client_conns);
962 }
963
964 nr_active--;
965 }
966
967 rxrpc_nr_active_client_conns = nr_active;
968 spin_unlock(&rxrpc_client_conn_cache_lock);
969 ASSERTCMP(nr_active, >=, 0);
970 _leave(" [culled]");
971}
972
973/*
974 * Discard expired client connections from the idle list. Each conn in the
975 * idle list has been exposed and holds an extra ref because of that.
976 *
977 * This may be called from conn setup or from a work item so cannot be
978 * considered non-reentrant.
979 */
980static void rxrpc_discard_expired_client_conns(struct work_struct *work)
981{
982 struct rxrpc_connection *conn;
983 unsigned long expiry, conn_expires_at, now;
984 unsigned int nr_conns;
985 bool did_discard = false;
986
987 _enter("%c", work ? 'w' : 'n');
988
989 if (list_empty(&rxrpc_idle_client_conns)) {
990 _leave(" [empty]");
991 return;
992 }
993
994 /* Don't double up on the discarding */
995 if (!spin_trylock(&rxrpc_client_conn_discard_mutex)) {
996 _leave(" [already]");
997 return;
998 }
999
1000 /* We keep an estimate of what the number of conns ought to be after
1001 * we've discarded some so that we don't overdo the discarding.
1002 */
1003 nr_conns = rxrpc_nr_client_conns;
1004
1005next:
1006 spin_lock(&rxrpc_client_conn_cache_lock);
1007
1008 if (list_empty(&rxrpc_idle_client_conns))
1009 goto out;
1010
1011 conn = list_entry(rxrpc_idle_client_conns.next,
1012 struct rxrpc_connection, cache_link);
1013 ASSERT(test_bit(RXRPC_CONN_EXPOSED, &conn->flags));
1014
1015 if (!rxrpc_kill_all_client_conns) {
1016 /* If the number of connections is over the reap limit, we
1017 * expedite discard by reducing the expiry timeout. We must,
1018 * however, have at least a short grace period to be able to do
1019 * final-ACK or ABORT retransmission.
1020 */
1021 expiry = rxrpc_conn_idle_client_expiry;
1022 if (nr_conns > rxrpc_reap_client_connections)
1023 expiry = rxrpc_conn_idle_client_fast_expiry;
1024
1025 conn_expires_at = conn->idle_timestamp + expiry;
1026
1027 now = READ_ONCE(jiffies);
1028 if (time_after(conn_expires_at, now))
1029 goto not_yet_expired;
1030 }
1031
1032 trace_rxrpc_client(conn, -1, rxrpc_client_discard);
1033 if (!test_and_clear_bit(RXRPC_CONN_EXPOSED, &conn->flags))
1034 BUG();
1035 conn->cache_state = RXRPC_CONN_CLIENT_INACTIVE;
1036 list_del_init(&conn->cache_link);
1037
1038 spin_unlock(&rxrpc_client_conn_cache_lock);
1039
1040 /* When we cleared the EXPOSED flag, we took on responsibility for the
1041 * reference that that had on the usage count. We deal with that here.
1042 * If someone re-sets the flag and re-gets the ref, that's fine.
1043 */
1044 rxrpc_put_connection(conn);
1045 did_discard = true;
1046 nr_conns--;
1047 goto next;
1048
1049not_yet_expired:
1050 /* The connection at the front of the queue hasn't yet expired, so
1051 * schedule the work item for that point if we discarded something.
1052 *
1053 * We don't worry if the work item is already scheduled - it can look
1054 * after rescheduling itself at a later time. We could cancel it, but
1055 * then things get messier.
1056 */
1057 _debug("not yet");
1058 if (!rxrpc_kill_all_client_conns)
1059 queue_delayed_work(rxrpc_workqueue,
1060 &rxrpc_client_conn_reap,
1061 conn_expires_at - now);
1062
1063out:
1064 spin_unlock(&rxrpc_client_conn_cache_lock);
1065 spin_unlock(&rxrpc_client_conn_discard_mutex);
1066 _leave("");
1067}
1068
1069/*
1070 * Preemptively destroy all the client connection records rather than waiting
1071 * for them to time out
1072 */
1073void __exit rxrpc_destroy_all_client_connections(void)
1074{
1075 _enter("");
1076
1077 spin_lock(&rxrpc_client_conn_cache_lock);
1078 rxrpc_kill_all_client_conns = true;
1079 spin_unlock(&rxrpc_client_conn_cache_lock);
1080
1081 cancel_delayed_work(&rxrpc_client_conn_reap);
1082
1083 if (!queue_delayed_work(rxrpc_workqueue, &rxrpc_client_conn_reap, 0))
1084 _debug("destroy: queue failed");
1085
1086 _leave("");
1087}