net/rxrpc/conn_client.c

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