net/sunrpc/xprtmultipath.c

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