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