net/sunrpc/xprtmultipath.c

   1 // SPDX-License-Identifier: GPL-2.0
   2 /*
   3  * Multipath support for RPC
   4  *
   5  * Copyright (c) 2015, 2016, Primary Data, Inc. All rights reserved.
   6  *
   7  * Trond Myklebust <trond.myklebust@primarydata.com>
   8  *
   9  */
  10 #include <linux/types.h>
  11 #include <linux/kref.h>
  12 #include <linux/list.h>
  13 #include <linux/rcupdate.h>
  14 #include <linux/rculist.h>
  15 #include <linux/slab.h>
  16 #include <asm/cmpxchg.h>
  17 #include <linux/spinlock.h>
  18 #include <linux/sunrpc/xprt.h>
  19 #include <linux/sunrpc/addr.h>
  20 #include <linux/sunrpc/xprtmultipath.h>
  21
  22 #include "sysfs.h"
  23
  24 typedef struct rpc_xprt *(*xprt_switch_find_xprt_t)(struct rpc_xprt_switch *xps,
  25                 const struct rpc_xprt *cur);
  26
  27 static const struct rpc_xprt_iter_ops rpc_xprt_iter_singular;
  28 static const struct rpc_xprt_iter_ops rpc_xprt_iter_roundrobin;
  29 static const struct rpc_xprt_iter_ops rpc_xprt_iter_listall;
  30
  31 static void xprt_switch_add_xprt_locked(struct rpc_xprt_switch *xps,
  32                 struct rpc_xprt *xprt)
  33 {
  34         if (unlikely(xprt_get(xprt) == NULL))
  35                 return;
  36         list_add_tail_rcu(&xprt->xprt_switch, &xps->xps_xprt_list);
  37         smp_wmb();
  38         if (xps->xps_nxprts == 0)
  39                 xps->xps_net = xprt->xprt_net;
  40         xps->xps_nxprts++;
  41         xps->xps_nactive++;
  42 }
  43
  44 /**
  45  * rpc_xprt_switch_add_xprt - Add a new rpc_xprt to an rpc_xprt_switch
  46  * @xps: pointer to struct rpc_xprt_switch
  47  * @xprt: pointer to struct rpc_xprt
  48  *
  49  * Adds xprt to the end of the list of struct rpc_xprt in xps.
  50  */
  51 void rpc_xprt_switch_add_xprt(struct rpc_xprt_switch *xps,
  52                 struct rpc_xprt *xprt)
  53 {
  54         if (xprt == NULL)
  55                 return;
  56         spin_lock(&xps->xps_lock);
  57         if (xps->xps_net == xprt->xprt_net || xps->xps_net == NULL)
  58                 xprt_switch_add_xprt_locked(xps, xprt);
  59         spin_unlock(&xps->xps_lock);
  60         rpc_sysfs_xprt_setup(xps, xprt, GFP_KERNEL);
  61 }
  62
  63 static void xprt_switch_remove_xprt_locked(struct rpc_xprt_switch *xps,
  64                 struct rpc_xprt *xprt)
  65 {
  66         if (unlikely(xprt == NULL))
  67                 return;
  68         xps->xps_nactive--;
  69         xps->xps_nxprts--;
  70         if (xps->xps_nxprts == 0)
  71                 xps->xps_net = NULL;
  72         smp_wmb();
  73         list_del_rcu(&xprt->xprt_switch);
  74 }
  75
  76 /**
  77  * rpc_xprt_switch_remove_xprt - Removes an rpc_xprt from a rpc_xprt_switch
  78  * @xps: pointer to struct rpc_xprt_switch
  79  * @xprt: pointer to struct rpc_xprt
  80  *
  81  * Removes xprt from the list of struct rpc_xprt in xps.
  82  */
  83 void rpc_xprt_switch_remove_xprt(struct rpc_xprt_switch *xps,
  84                 struct rpc_xprt *xprt)
  85 {
  86         spin_lock(&xps->xps_lock);
  87         xprt_switch_remove_xprt_locked(xps, xprt);
  88         spin_unlock(&xps->xps_lock);
  89         rpc_sysfs_xprt_destroy(xprt);
  90         xprt_put(xprt);
  91 }
  92
  93 static DEFINE_IDA(rpc_xprtswitch_ids);
  94
  95 void xprt_multipath_cleanup_ids(void)
  96 {
  97         ida_destroy(&rpc_xprtswitch_ids);
  98 }
  99
 100 static int xprt_switch_alloc_id(struct rpc_xprt_switch *xps, gfp_t gfp_flags)
 101 {
 102         int id;
 103
 104         id = ida_simple_get(&rpc_xprtswitch_ids, 0, 0, gfp_flags);
 105         if (id < 0)
 106                 return id;
 107
 108         xps->xps_id = id;
 109         return 0;
 110 }
 111
 112 static void xprt_switch_free_id(struct rpc_xprt_switch *xps)
 113 {
 114         ida_simple_remove(&rpc_xprtswitch_ids, xps->xps_id);
 115 }
 116
 117 /**
 118  * xprt_switch_alloc - Allocate a new struct rpc_xprt_switch
 119  * @xprt: pointer to struct rpc_xprt
 120  * @gfp_flags: allocation flags
 121  *
 122  * On success, returns an initialised struct rpc_xprt_switch, containing
 123  * the entry xprt. Returns NULL on failure.
 124  */
 125 struct rpc_xprt_switch *xprt_switch_alloc(struct rpc_xprt *xprt,
 126                 gfp_t gfp_flags)
 127 {
 128         struct rpc_xprt_switch *xps;
 129
 130         xps = kmalloc(sizeof(*xps), gfp_flags);
 131         if (xps != NULL) {
 132                 spin_lock_init(&xps->xps_lock);
 133                 kref_init(&xps->xps_kref);
 134                 xprt_switch_alloc_id(xps, gfp_flags);
 135                 xps->xps_nxprts = xps->xps_nactive = 0;
 136                 atomic_long_set(&xps->xps_queuelen, 0);
 137                 xps->xps_net = NULL;
 138                 INIT_LIST_HEAD(&xps->xps_xprt_list);
 139                 xps->xps_iter_ops = &rpc_xprt_iter_singular;
 140                 rpc_sysfs_xprt_switch_setup(xps, xprt, gfp_flags);
 141                 xprt_switch_add_xprt_locked(xps, xprt);
 142                 rpc_sysfs_xprt_setup(xps, xprt, gfp_flags);
 143         }
 144
 145         return xps;
 146 }
 147
 148 static void xprt_switch_free_entries(struct rpc_xprt_switch *xps)
 149 {
 150         spin_lock(&xps->xps_lock);
 151         while (!list_empty(&xps->xps_xprt_list)) {
 152                 struct rpc_xprt *xprt;
 153
 154                 xprt = list_first_entry(&xps->xps_xprt_list,
 155                                 struct rpc_xprt, xprt_switch);
 156                 xprt_switch_remove_xprt_locked(xps, xprt);
 157                 spin_unlock(&xps->xps_lock);
 158                 rpc_sysfs_xprt_destroy(xprt);
 159                 xprt_put(xprt);
 160                 spin_lock(&xps->xps_lock);
 161         }
 162         spin_unlock(&xps->xps_lock);
 163 }
 164
 165 static void xprt_switch_free(struct kref *kref)
 166 {
 167         struct rpc_xprt_switch *xps = container_of(kref,
 168                         struct rpc_xprt_switch, xps_kref);
 169
 170         xprt_switch_free_entries(xps);
 171         rpc_sysfs_xprt_switch_destroy(xps);
 172         xprt_switch_free_id(xps);
 173         kfree_rcu(xps, xps_rcu);
 174 }
 175
 176 /**
 177  * xprt_switch_get - Return a reference to a rpc_xprt_switch
 178  * @xps: pointer to struct rpc_xprt_switch
 179  *
 180  * Returns a reference to xps unless the refcount is already zero.
 181  */
 182 struct rpc_xprt_switch *xprt_switch_get(struct rpc_xprt_switch *xps)
 183 {
 184         if (xps != NULL && kref_get_unless_zero(&xps->xps_kref))
 185                 return xps;
 186         return NULL;
 187 }
 188
 189 /**
 190  * xprt_switch_put - Release a reference to a rpc_xprt_switch
 191  * @xps: pointer to struct rpc_xprt_switch
 192  *
 193  * Release the reference to xps, and free it once the refcount is zero.
 194  */
 195 void xprt_switch_put(struct rpc_xprt_switch *xps)
 196 {
 197         if (xps != NULL)
 198                 kref_put(&xps->xps_kref, xprt_switch_free);
 199 }
 200
 201 /**
 202  * rpc_xprt_switch_set_roundrobin - Set a round-robin policy on rpc_xprt_switch
 203  * @xps: pointer to struct rpc_xprt_switch
 204  *
 205  * Sets a round-robin default policy for iterators acting on xps.
 206  */
 207 void rpc_xprt_switch_set_roundrobin(struct rpc_xprt_switch *xps)
 208 {
 209         if (READ_ONCE(xps->xps_iter_ops) != &rpc_xprt_iter_roundrobin)
 210                 WRITE_ONCE(xps->xps_iter_ops, &rpc_xprt_iter_roundrobin);
 211 }
 212
 213 static
 214 const struct rpc_xprt_iter_ops *xprt_iter_ops(const struct rpc_xprt_iter *xpi)
 215 {
 216         if (xpi->xpi_ops != NULL)
 217                 return xpi->xpi_ops;
 218         return rcu_dereference(xpi->xpi_xpswitch)->xps_iter_ops;
 219 }
 220
 221 static
 222 void xprt_iter_no_rewind(struct rpc_xprt_iter *xpi)
 223 {
 224 }
 225
 226 static
 227 void xprt_iter_default_rewind(struct rpc_xprt_iter *xpi)
 228 {
 229         WRITE_ONCE(xpi->xpi_cursor, NULL);
 230 }
 231
 232 static
 233 bool xprt_is_active(const struct rpc_xprt *xprt)
 234 {
 235         return kref_read(&xprt->kref) != 0;
 236 }
 237
 238 static
 239 struct rpc_xprt *xprt_switch_find_first_entry(struct list_head *head)
 240 {
 241         struct rpc_xprt *pos;
 242
 243         list_for_each_entry_rcu(pos, head, xprt_switch) {
 244                 if (xprt_is_active(pos))
 245                         return pos;
 246         }
 247         return NULL;
 248 }
 249
 250 static
 251 struct rpc_xprt *xprt_iter_first_entry(struct rpc_xprt_iter *xpi)
 252 {
 253         struct rpc_xprt_switch *xps = rcu_dereference(xpi->xpi_xpswitch);
 254
 255         if (xps == NULL)
 256                 return NULL;
 257         return xprt_switch_find_first_entry(&xps->xps_xprt_list);
 258 }
 259
 260 static
 261 struct rpc_xprt *xprt_switch_find_current_entry(struct list_head *head,
 262                 const struct rpc_xprt *cur)
 263 {
 264         struct rpc_xprt *pos;
 265         bool found = false;
 266
 267         list_for_each_entry_rcu(pos, head, xprt_switch) {
 268                 if (cur == pos)
 269                         found = true;
 270                 if (found && xprt_is_active(pos))
 271                         return pos;
 272         }
 273         return NULL;
 274 }
 275
 276 static
 277 struct rpc_xprt *xprt_iter_current_entry(struct rpc_xprt_iter *xpi)
 278 {
 279         struct rpc_xprt_switch *xps = rcu_dereference(xpi->xpi_xpswitch);
 280         struct list_head *head;
 281
 282         if (xps == NULL)
 283                 return NULL;
 284         head = &xps->xps_xprt_list;
 285         if (xpi->xpi_cursor == NULL || xps->xps_nxprts < 2)
 286                 return xprt_switch_find_first_entry(head);
 287         return xprt_switch_find_current_entry(head, xpi->xpi_cursor);
 288 }
 289
 290 bool rpc_xprt_switch_has_addr(struct rpc_xprt_switch *xps,
 291                               const struct sockaddr *sap)
 292 {
 293         struct list_head *head;
 294         struct rpc_xprt *pos;
 295
 296         if (xps == NULL || sap == NULL)
 297                 return false;
 298
 299         head = &xps->xps_xprt_list;
 300         list_for_each_entry_rcu(pos, head, xprt_switch) {
 301                 if (rpc_cmp_addr_port(sap, (struct sockaddr *)&pos->addr)) {
 302                         pr_info("RPC:   addr %s already in xprt switch\n",
 303                                 pos->address_strings[RPC_DISPLAY_ADDR]);
 304                         return true;
 305                 }
 306         }
 307         return false;
 308 }
 309
 310 static
 311 struct rpc_xprt *xprt_switch_find_next_entry(struct list_head *head,
 312                 const struct rpc_xprt *cur)
 313 {
 314         struct rpc_xprt *pos, *prev = NULL;
 315         bool found = false;
 316
 317         list_for_each_entry_rcu(pos, head, xprt_switch) {
 318                 if (cur == prev)
 319                         found = true;
 320                 if (found && xprt_is_active(pos))
 321                         return pos;
 322                 prev = pos;
 323         }
 324         return NULL;
 325 }
 326
 327 static
 328 struct rpc_xprt *xprt_switch_set_next_cursor(struct rpc_xprt_switch *xps,
 329                 struct rpc_xprt **cursor,
 330                 xprt_switch_find_xprt_t find_next)
 331 {
 332         struct rpc_xprt *pos, *old;
 333
 334         old = smp_load_acquire(cursor);
 335         pos = find_next(xps, old);
 336         smp_store_release(cursor, pos);
 337         return pos;
 338 }
 339
 340 static
 341 struct rpc_xprt *xprt_iter_next_entry_multiple(struct rpc_xprt_iter *xpi,
 342                 xprt_switch_find_xprt_t find_next)
 343 {
 344         struct rpc_xprt_switch *xps = rcu_dereference(xpi->xpi_xpswitch);
 345
 346         if (xps == NULL)
 347                 return NULL;
 348         return xprt_switch_set_next_cursor(xps, &xpi->xpi_cursor, find_next);
 349 }
 350
 351 static
 352 struct rpc_xprt *__xprt_switch_find_next_entry_roundrobin(struct list_head *head,
 353                 const struct rpc_xprt *cur)
 354 {
 355         struct rpc_xprt *ret;
 356
 357         ret = xprt_switch_find_next_entry(head, cur);
 358         if (ret != NULL)
 359                 return ret;
 360         return xprt_switch_find_first_entry(head);
 361 }
 362
 363 static
 364 struct rpc_xprt *xprt_switch_find_next_entry_roundrobin(struct rpc_xprt_switch *xps,
 365                 const struct rpc_xprt *cur)
 366 {
 367         struct list_head *head = &xps->xps_xprt_list;
 368         struct rpc_xprt *xprt;
 369         unsigned int nactive;
 370
 371         for (;;) {
 372                 unsigned long xprt_queuelen, xps_queuelen;
 373
 374                 xprt = __xprt_switch_find_next_entry_roundrobin(head, cur);
 375                 if (!xprt)
 376                         break;
 377                 xprt_queuelen = atomic_long_read(&xprt->queuelen);
 378                 xps_queuelen = atomic_long_read(&xps->xps_queuelen);
 379                 nactive = READ_ONCE(xps->xps_nactive);
 380                 /* Exit loop if xprt_queuelen <= average queue length */
 381                 if (xprt_queuelen * nactive <= xps_queuelen)
 382                         break;
 383                 cur = xprt;
 384         }
 385         return xprt;
 386 }
 387
 388 static
 389 struct rpc_xprt *xprt_iter_next_entry_roundrobin(struct rpc_xprt_iter *xpi)
 390 {
 391         return xprt_iter_next_entry_multiple(xpi,
 392                         xprt_switch_find_next_entry_roundrobin);
 393 }
 394
 395 static
 396 struct rpc_xprt *xprt_switch_find_next_entry_all(struct rpc_xprt_switch *xps,
 397                 const struct rpc_xprt *cur)
 398 {
 399         return xprt_switch_find_next_entry(&xps->xps_xprt_list, cur);
 400 }
 401
 402 static
 403 struct rpc_xprt *xprt_iter_next_entry_all(struct rpc_xprt_iter *xpi)
 404 {
 405         return xprt_iter_next_entry_multiple(xpi,
 406                         xprt_switch_find_next_entry_all);
 407 }
 408
 409 /*
 410  * xprt_iter_rewind - Resets the xprt iterator
 411  * @xpi: pointer to rpc_xprt_iter
 412  *
 413  * Resets xpi to ensure that it points to the first entry in the list
 414  * of transports.
 415  */
 416 static
 417 void xprt_iter_rewind(struct rpc_xprt_iter *xpi)
 418 {
 419         rcu_read_lock();
 420         xprt_iter_ops(xpi)->xpi_rewind(xpi);
 421         rcu_read_unlock();
 422 }
 423
 424 static void __xprt_iter_init(struct rpc_xprt_iter *xpi,
 425                 struct rpc_xprt_switch *xps,
 426                 const struct rpc_xprt_iter_ops *ops)
 427 {
 428         rcu_assign_pointer(xpi->xpi_xpswitch, xprt_switch_get(xps));
 429         xpi->xpi_cursor = NULL;
 430         xpi->xpi_ops = ops;
 431 }
 432
 433 /**
 434  * xprt_iter_init - Initialise an xprt iterator
 435  * @xpi: pointer to rpc_xprt_iter
 436  * @xps: pointer to rpc_xprt_switch
 437  *
 438  * Initialises the iterator to use the default iterator ops
 439  * as set in xps. This function is mainly intended for internal
 440  * use in the rpc_client.
 441  */
 442 void xprt_iter_init(struct rpc_xprt_iter *xpi,
 443                 struct rpc_xprt_switch *xps)
 444 {
 445         __xprt_iter_init(xpi, xps, NULL);
 446 }
 447
 448 /**
 449  * xprt_iter_init_listall - Initialise an xprt iterator
 450  * @xpi: pointer to rpc_xprt_iter
 451  * @xps: pointer to rpc_xprt_switch
 452  *
 453  * Initialises the iterator to iterate once through the entire list
 454  * of entries in xps.
 455  */
 456 void xprt_iter_init_listall(struct rpc_xprt_iter *xpi,
 457                 struct rpc_xprt_switch *xps)
 458 {
 459         __xprt_iter_init(xpi, xps, &rpc_xprt_iter_listall);
 460 }
 461
 462 /**
 463  * xprt_iter_xchg_switch - Atomically swap out the rpc_xprt_switch
 464  * @xpi: pointer to rpc_xprt_iter
 465  * @newswitch: pointer to a new rpc_xprt_switch or NULL
 466  *
 467  * Swaps out the existing xpi->xpi_xpswitch with a new value.
 468  */
 469 struct rpc_xprt_switch *xprt_iter_xchg_switch(struct rpc_xprt_iter *xpi,
 470                 struct rpc_xprt_switch *newswitch)
 471 {
 472         struct rpc_xprt_switch __rcu *oldswitch;
 473
 474         /* Atomically swap out the old xpswitch */
 475         oldswitch = xchg(&xpi->xpi_xpswitch, RCU_INITIALIZER(newswitch));
 476         if (newswitch != NULL)
 477                 xprt_iter_rewind(xpi);
 478         return rcu_dereference_protected(oldswitch, true);
 479 }
 480
 481 /**
 482  * xprt_iter_destroy - Destroys the xprt iterator
 483  * @xpi: pointer to rpc_xprt_iter
 484  */
 485 void xprt_iter_destroy(struct rpc_xprt_iter *xpi)
 486 {
 487         xprt_switch_put(xprt_iter_xchg_switch(xpi, NULL));
 488 }
 489
 490 /**
 491  * xprt_iter_xprt - Returns the rpc_xprt pointed to by the cursor
 492  * @xpi: pointer to rpc_xprt_iter
 493  *
 494  * Returns a pointer to the struct rpc_xprt that is currently
 495  * pointed to by the cursor.
 496  * Caller must be holding rcu_read_lock().
 497  */
 498 struct rpc_xprt *xprt_iter_xprt(struct rpc_xprt_iter *xpi)
 499 {
 500         WARN_ON_ONCE(!rcu_read_lock_held());
 501         return xprt_iter_ops(xpi)->xpi_xprt(xpi);
 502 }
 503
 504 static
 505 struct rpc_xprt *xprt_iter_get_helper(struct rpc_xprt_iter *xpi,
 506                 struct rpc_xprt *(*fn)(struct rpc_xprt_iter *))
 507 {
 508         struct rpc_xprt *ret;
 509
 510         do {
 511                 ret = fn(xpi);
 512                 if (ret == NULL)
 513                         break;
 514                 ret = xprt_get(ret);
 515         } while (ret == NULL);
 516         return ret;
 517 }
 518
 519 /**
 520  * xprt_iter_get_xprt - Returns the rpc_xprt pointed to by the cursor
 521  * @xpi: pointer to rpc_xprt_iter
 522  *
 523  * Returns a reference to the struct rpc_xprt that is currently
 524  * pointed to by the cursor.
 525  */
 526 struct rpc_xprt *xprt_iter_get_xprt(struct rpc_xprt_iter *xpi)
 527 {
 528         struct rpc_xprt *xprt;
 529
 530         rcu_read_lock();
 531         xprt = xprt_iter_get_helper(xpi, xprt_iter_ops(xpi)->xpi_xprt);
 532         rcu_read_unlock();
 533         return xprt;
 534 }
 535
 536 /**
 537  * xprt_iter_get_next - Returns the next rpc_xprt following the cursor
 538  * @xpi: pointer to rpc_xprt_iter
 539  *
 540  * Returns a reference to the struct rpc_xprt that immediately follows the
 541  * entry pointed to by the cursor.
 542  */
 543 struct rpc_xprt *xprt_iter_get_next(struct rpc_xprt_iter *xpi)
 544 {
 545         struct rpc_xprt *xprt;
 546
 547         rcu_read_lock();
 548         xprt = xprt_iter_get_helper(xpi, xprt_iter_ops(xpi)->xpi_next);
 549         rcu_read_unlock();
 550         return xprt;
 551 }
 552
 553 /* Policy for always returning the first entry in the rpc_xprt_switch */
 554 static
 555 const struct rpc_xprt_iter_ops rpc_xprt_iter_singular = {
 556         .xpi_rewind = xprt_iter_no_rewind,
 557         .xpi_xprt = xprt_iter_first_entry,
 558         .xpi_next = xprt_iter_first_entry,
 559 };
 560
 561 /* Policy for round-robin iteration of entries in the rpc_xprt_switch */
 562 static
 563 const struct rpc_xprt_iter_ops rpc_xprt_iter_roundrobin = {
 564         .xpi_rewind = xprt_iter_default_rewind,
 565         .xpi_xprt = xprt_iter_current_entry,
 566         .xpi_next = xprt_iter_next_entry_roundrobin,
 567 };
 568
 569 /* Policy for once-through iteration of entries in the rpc_xprt_switch */
 570 static
 571 const struct rpc_xprt_iter_ops rpc_xprt_iter_listall = {
 572         .xpi_rewind = xprt_iter_default_rewind,
 573         .xpi_xprt = xprt_iter_current_entry,
 574         .xpi_next = xprt_iter_next_entry_all,
 575 };