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