net/rxrpc/conn_client.c

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