1	// SPDX-License-Identifier: GPL-2.0-only
2	/*
3	* linux/net/sunrpc/clnt.c
4	*
5	* This file contains the high-level RPC interface.
6	* It is modeled as a finite state machine to support both synchronous
7	* and asynchronous requests.
8	*
9	* - RPC header generation and argument serialization.
10	* - Credential refresh.
11	* - TCP connect handling.
12	* - Retry of operation when it is suspected the operation failed because
13	* of uid squashing on the server, or when the credentials were stale
14	* and need to be refreshed, or when a packet was damaged in transit.
15	* This may be have to be moved to the VFS layer.
16	*
17	* Copyright (C) 1992,1993 Rick Sladkey <jrs@world.std.com>
18	* Copyright (C) 1995,1996 Olaf Kirch <okir@monad.swb.de>
19	*/
20
21
22	#include <linux/module.h>
23	#include <linux/types.h>
24	#include <linux/kallsyms.h>
25	#include <linux/mm.h>
26	#include <linux/namei.h>
27	#include <linux/mount.h>
28	#include <linux/slab.h>
29	#include <linux/rcupdate.h>
30	#include <linux/utsname.h>
31	#include <linux/workqueue.h>
32	#include <linux/in.h>
33	#include <linux/in6.h>
34	#include <linux/un.h>
35
36	#include <linux/sunrpc/clnt.h>
37	#include <linux/sunrpc/addr.h>
38	#include <linux/sunrpc/rpc_pipe_fs.h>
39	#include <linux/sunrpc/metrics.h>
40	#include <linux/sunrpc/bc_xprt.h>
41	#include <trace/events/sunrpc.h>
42
43	#include "sunrpc.h"
44	#include "sysfs.h"
45	#include "netns.h"
46
47	#if IS_ENABLED(CONFIG_SUNRPC_DEBUG)
48	# define RPCDBG_FACILITY RPCDBG_CALL
49	#endif
50
51	/*
52	* All RPC clients are linked into this list
53	*/
54
55	static DECLARE_WAIT_QUEUE_HEAD(destroy_wait);
56
57
58	static void call_start(struct rpc_task *task);
59	static void call_reserve(struct rpc_task *task);
60	static void call_reserveresult(struct rpc_task *task);
61	static void call_allocate(struct rpc_task *task);
62	static void call_encode(struct rpc_task *task);
63	static void call_decode(struct rpc_task *task);
64	static void call_bind(struct rpc_task *task);
65	static void call_bind_status(struct rpc_task *task);
66	static void call_transmit(struct rpc_task *task);
67	static void call_status(struct rpc_task *task);
68	static void call_transmit_status(struct rpc_task *task);
69	static void call_refresh(struct rpc_task *task);
70	static void call_refreshresult(struct rpc_task *task);
71	static void call_connect(struct rpc_task *task);
72	static void call_connect_status(struct rpc_task *task);
73
74	static int rpc_encode_header(struct rpc_task *task,
75	struct xdr_stream *xdr);
76	static int rpc_decode_header(struct rpc_task *task,
77	struct xdr_stream *xdr);
78	static int rpc_ping(struct rpc_clnt *clnt);
79	static int rpc_ping_noreply(struct rpc_clnt *clnt);
80	static void rpc_check_timeout(struct rpc_task *task);
81
82	static void rpc_register_client(struct rpc_clnt *clnt)
83	{
84	struct net *net = rpc_net_ns(clnt);
85	struct sunrpc_net *sn = net_generic(net, id: sunrpc_net_id);
86
87	spin_lock(lock: &sn->rpc_client_lock);
88	list_add(new: &clnt->cl_clients, head: &sn->all_clients);
89	spin_unlock(lock: &sn->rpc_client_lock);
90	}
91
92	static void rpc_unregister_client(struct rpc_clnt *clnt)
93	{
94	struct net *net = rpc_net_ns(clnt);
95	struct sunrpc_net *sn = net_generic(net, id: sunrpc_net_id);
96
97	spin_lock(lock: &sn->rpc_client_lock);
98	list_del(entry: &clnt->cl_clients);
99	spin_unlock(lock: &sn->rpc_client_lock);
100	}
101
102	static void __rpc_clnt_remove_pipedir(struct rpc_clnt *clnt)
103	{
104	rpc_remove_client_dir(clnt);
105	}
106
107	static void rpc_clnt_remove_pipedir(struct rpc_clnt *clnt)
108	{
109	struct net *net = rpc_net_ns(clnt);
110	struct super_block *pipefs_sb;
111
112	pipefs_sb = rpc_get_sb_net(net);
113	if (pipefs_sb) {
114	if (pipefs_sb == clnt->pipefs_sb)
115	__rpc_clnt_remove_pipedir(clnt);
116	rpc_put_sb_net(net);
117	}
118	}
119
120	static struct dentry *rpc_setup_pipedir_sb(struct super_block *sb,
121	struct rpc_clnt *clnt)
122	{
123	static uint32_t clntid;
124	const char *dir_name = clnt->cl_program->pipe_dir_name;
125	char name[15];
126	struct dentry *dir, *dentry;
127
128	dir = rpc_d_lookup_sb(sb, dir_name);
129	if (dir == NULL) {
130	pr_info("RPC: pipefs directory doesn't exist: %s\n", dir_name);
131	return dir;
132	}
133	for (;;) {
134	snprintf(buf: name, size: sizeof(name), fmt: "clnt%x", (unsigned int)clntid++);
135	name[sizeof(name) - 1] = '\0';
136	dentry = rpc_create_client_dir(dir, name, clnt);
137	if (!IS_ERR(ptr: dentry))
138	break;
139	if (dentry == ERR_PTR(error: -EEXIST))
140	continue;
141	printk(KERN_INFO "RPC: Couldn't create pipefs entry"
142	" %s/%s, error %ld\n",
143	dir_name, name, PTR_ERR(dentry));
144	break;
145	}
146	dput(dir);
147	return dentry;
148	}
149
150	static int
151	rpc_setup_pipedir(struct super_block *pipefs_sb, struct rpc_clnt *clnt)
152	{
153	struct dentry *dentry;
154
155	clnt->pipefs_sb = pipefs_sb;
156
157	if (clnt->cl_program->pipe_dir_name != NULL) {
158	dentry = rpc_setup_pipedir_sb(sb: pipefs_sb, clnt);
159	if (IS_ERR(ptr: dentry))
160	return PTR_ERR(ptr: dentry);
161	}
162	return 0;
163	}
164
165	static int rpc_clnt_skip_event(struct rpc_clnt *clnt, unsigned long event)
166	{
167	if (clnt->cl_program->pipe_dir_name == NULL)
168	return 1;
169
170	switch (event) {
171	case RPC_PIPEFS_MOUNT:
172	if (clnt->cl_pipedir_objects.pdh_dentry != NULL)
173	return 1;
174	if (refcount_read(r: &clnt->cl_count) == 0)
175	return 1;
176	break;
177	case RPC_PIPEFS_UMOUNT:
178	if (clnt->cl_pipedir_objects.pdh_dentry == NULL)
179	return 1;
180	break;
181	}
182	return 0;
183	}
184
185	static int __rpc_clnt_handle_event(struct rpc_clnt *clnt, unsigned long event,
186	struct super_block *sb)
187	{
188	struct dentry *dentry;
189
190	switch (event) {
191	case RPC_PIPEFS_MOUNT:
192	dentry = rpc_setup_pipedir_sb(sb, clnt);
193	if (!dentry)
194	return -ENOENT;
195	if (IS_ERR(ptr: dentry))
196	return PTR_ERR(ptr: dentry);
197	break;
198	case RPC_PIPEFS_UMOUNT:
199	__rpc_clnt_remove_pipedir(clnt);
200	break;
201	default:
202	printk(KERN_ERR "%s: unknown event: %ld\n", __func__, event);
203	return -ENOTSUPP;
204	}
205	return 0;
206	}
207
208	static int __rpc_pipefs_event(struct rpc_clnt *clnt, unsigned long event,
209	struct super_block *sb)
210	{
211	int error = 0;
212
213	for (;; clnt = clnt->cl_parent) {
214	if (!rpc_clnt_skip_event(clnt, event))
215	error = __rpc_clnt_handle_event(clnt, event, sb);
216	if (error || clnt == clnt->cl_parent)
217	break;
218	}
219	return error;
220	}
221
222	static struct rpc_clnt *rpc_get_client_for_event(struct net *net, int event)
223	{
224	struct sunrpc_net *sn = net_generic(net, id: sunrpc_net_id);
225	struct rpc_clnt *clnt;
226
227	spin_lock(lock: &sn->rpc_client_lock);
228	list_for_each_entry(clnt, &sn->all_clients, cl_clients) {
229	if (rpc_clnt_skip_event(clnt, event))
230	continue;
231	spin_unlock(lock: &sn->rpc_client_lock);
232	return clnt;
233	}
234	spin_unlock(lock: &sn->rpc_client_lock);
235	return NULL;
236	}
237
238	static int rpc_pipefs_event(struct notifier_block *nb, unsigned long event,
239	void *ptr)
240	{
241	struct super_block *sb = ptr;
242	struct rpc_clnt *clnt;
243	int error = 0;
244
245	while ((clnt = rpc_get_client_for_event(net: sb->s_fs_info, event))) {
246	error = __rpc_pipefs_event(clnt, event, sb);
247	if (error)
248	break;
249	}
250	return error;
251	}
252
253	static struct notifier_block rpc_clients_block = {
254	.notifier_call = rpc_pipefs_event,
255	.priority = SUNRPC_PIPEFS_RPC_PRIO,
256	};
257
258	int rpc_clients_notifier_register(void)
259	{
260	return rpc_pipefs_notifier_register(&rpc_clients_block);
261	}
262
263	void rpc_clients_notifier_unregister(void)
264	{
265	return rpc_pipefs_notifier_unregister(&rpc_clients_block);
266	}
267
268	static struct rpc_xprt *rpc_clnt_set_transport(struct rpc_clnt *clnt,
269	struct rpc_xprt *xprt,
270	const struct rpc_timeout *timeout)
271	{
272	struct rpc_xprt *old;
273
274	spin_lock(lock: &clnt->cl_lock);
275	old = rcu_dereference_protected(clnt->cl_xprt,
276	lockdep_is_held(&clnt->cl_lock));
277
278	if (!xprt_bound(xprt))
279	clnt->cl_autobind = 1;
280
281	clnt->cl_timeout = timeout;
282	rcu_assign_pointer(clnt->cl_xprt, xprt);
283	spin_unlock(lock: &clnt->cl_lock);
284
285	return old;
286	}
287
288	static void rpc_clnt_set_nodename(struct rpc_clnt *clnt, const char *nodename)
289	{
290	clnt->cl_nodelen = strlcpy(p: clnt->cl_nodename,
291	q: nodename, size: sizeof(clnt->cl_nodename));
292	}
293
294	static int rpc_client_register(struct rpc_clnt *clnt,
295	rpc_authflavor_t pseudoflavor,
296	const char *client_name)
297	{
298	struct rpc_auth_create_args auth_args = {
299	.pseudoflavor = pseudoflavor,
300	.target_name = client_name,
301	};
302	struct rpc_auth *auth;
303	struct net *net = rpc_net_ns(clnt);
304	struct super_block *pipefs_sb;
305	int err;
306
307	rpc_clnt_debugfs_register(clnt);
308
309	pipefs_sb = rpc_get_sb_net(net);
310	if (pipefs_sb) {
311	err = rpc_setup_pipedir(pipefs_sb, clnt);
312	if (err)
313	goto out;
314	}
315
316	rpc_register_client(clnt);
317	if (pipefs_sb)
318	rpc_put_sb_net(net);
319
320	auth = rpcauth_create(&auth_args, clnt);
321	if (IS_ERR(ptr: auth)) {
322	dprintk("RPC: Couldn't create auth handle (flavor %u)\n",
323	pseudoflavor);
324	err = PTR_ERR(ptr: auth);
325	goto err_auth;
326	}
327	return 0;
328	err_auth:
329	pipefs_sb = rpc_get_sb_net(net);
330	rpc_unregister_client(clnt);
331	__rpc_clnt_remove_pipedir(clnt);
332	out:
333	if (pipefs_sb)
334	rpc_put_sb_net(net);
335	rpc_sysfs_client_destroy(clnt);
336	rpc_clnt_debugfs_unregister(clnt);
337	return err;
338	}
339
340	static DEFINE_IDA(rpc_clids);
341
342	void rpc_cleanup_clids(void)
343	{
344	ida_destroy(ida: &rpc_clids);
345	}
346
347	static int rpc_alloc_clid(struct rpc_clnt *clnt)
348	{
349	int clid;
350
351	clid = ida_alloc(ida: &rpc_clids, GFP_KERNEL);
352	if (clid < 0)
353	return clid;
354	clnt->cl_clid = clid;
355	return 0;
356	}
357
358	static void rpc_free_clid(struct rpc_clnt *clnt)
359	{
360	ida_free(&rpc_clids, id: clnt->cl_clid);
361	}
362
363	static struct rpc_clnt * rpc_new_client(const struct rpc_create_args *args,
364	struct rpc_xprt_switch *xps,
365	struct rpc_xprt *xprt,
366	struct rpc_clnt *parent)
367	{
368	const struct rpc_program *program = args->program;
369	const struct rpc_version *version;
370	struct rpc_clnt *clnt = NULL;
371	const struct rpc_timeout *timeout;
372	const char *nodename = args->nodename;
373	int err;
374
375	err = rpciod_up();
376	if (err)
377	goto out_no_rpciod;
378
379	err = -EINVAL;
380	if (args->version >= program->nrvers)
381	goto out_err;
382	version = program->version[args->version];
383	if (version == NULL)
384	goto out_err;
385
386	err = -ENOMEM;
387	clnt = kzalloc(size: sizeof(*clnt), GFP_KERNEL);
388	if (!clnt)
389	goto out_err;
390	clnt->cl_parent = parent ? : clnt;
391	clnt->cl_xprtsec = args->xprtsec;
392
393	err = rpc_alloc_clid(clnt);
394	if (err)
395	goto out_no_clid;
396
397	clnt->cl_cred = get_cred(cred: args->cred);
398	clnt->cl_procinfo = version->procs;
399	clnt->cl_maxproc = version->nrprocs;
400	clnt->cl_prog = args->prognumber ? : program->number;
401	clnt->cl_vers = version->number;
402	clnt->cl_stats = program->stats;
403	clnt->cl_metrics = rpc_alloc_iostats(clnt);
404	rpc_init_pipe_dir_head(pdh: &clnt->cl_pipedir_objects);
405	err = -ENOMEM;
406	if (clnt->cl_metrics == NULL)
407	goto out_no_stats;
408	clnt->cl_program = program;
409	INIT_LIST_HEAD(list: &clnt->cl_tasks);
410	spin_lock_init(&clnt->cl_lock);
411
412	timeout = xprt->timeout;
413	if (args->timeout != NULL) {
414	memcpy(&clnt->cl_timeout_default, args->timeout,
415	sizeof(clnt->cl_timeout_default));
416	timeout = &clnt->cl_timeout_default;
417	}
418
419	rpc_clnt_set_transport(clnt, xprt, timeout);
420	xprt->main = true;
421	xprt_iter_init(xpi: &clnt->cl_xpi, xps);
422	xprt_switch_put(xps);
423
424	clnt->cl_rtt = &clnt->cl_rtt_default;
425	rpc_init_rtt(rt: &clnt->cl_rtt_default, timeo: clnt->cl_timeout->to_initval);
426
427	refcount_set(r: &clnt->cl_count, n: 1);
428
429	if (nodename == NULL)
430	nodename = utsname()->nodename;
431	/* save the nodename */
432	rpc_clnt_set_nodename(clnt, nodename);
433
434	rpc_sysfs_client_setup(clnt, xprt_switch: xps, net: rpc_net_ns(clnt));
435	err = rpc_client_register(clnt, pseudoflavor: args->authflavor, client_name: args->client_name);
436	if (err)
437	goto out_no_path;
438	if (parent)
439	refcount_inc(r: &parent->cl_count);
440
441	trace_rpc_clnt_new(clnt, xprt, args);
442	return clnt;
443
444	out_no_path:
445	rpc_free_iostats(clnt->cl_metrics);
446	out_no_stats:
447	put_cred(cred: clnt->cl_cred);
448	rpc_free_clid(clnt);
449	out_no_clid:
450	kfree(objp: clnt);
451	out_err:
452	rpciod_down();
453	out_no_rpciod:
454	xprt_switch_put(xps);
455	xprt_put(xprt);
456	trace_rpc_clnt_new_err(program: program->name, server: args->servername, error: err);
457	return ERR_PTR(error: err);
458	}
459
460	static struct rpc_clnt *rpc_create_xprt(struct rpc_create_args *args,
461	struct rpc_xprt *xprt)
462	{
463	struct rpc_clnt *clnt = NULL;
464	struct rpc_xprt_switch *xps;
465
466	if (args->bc_xprt && args->bc_xprt->xpt_bc_xps) {
467	WARN_ON_ONCE(!(args->protocol & XPRT_TRANSPORT_BC));
468	xps = args->bc_xprt->xpt_bc_xps;
469	xprt_switch_get(xps);
470	} else {
471	xps = xprt_switch_alloc(xprt, GFP_KERNEL);
472	if (xps == NULL) {
473	xprt_put(xprt);
474	return ERR_PTR(error: -ENOMEM);
475	}
476	if (xprt->bc_xprt) {
477	xprt_switch_get(xps);
478	xprt->bc_xprt->xpt_bc_xps = xps;
479	}
480	}
481	clnt = rpc_new_client(args, xps, xprt, NULL);
482	if (IS_ERR(ptr: clnt))
483	return clnt;
484
485	if (!(args->flags & RPC_CLNT_CREATE_NOPING)) {
486	int err = rpc_ping(clnt);
487	if (err != 0) {
488	rpc_shutdown_client(clnt);
489	return ERR_PTR(error: err);
490	}
491	} else if (args->flags & RPC_CLNT_CREATE_CONNECTED) {
492	int err = rpc_ping_noreply(clnt);
493	if (err != 0) {
494	rpc_shutdown_client(clnt);
495	return ERR_PTR(error: err);
496	}
497	}
498
499	clnt->cl_softrtry = 1;
500	if (args->flags & (RPC_CLNT_CREATE_HARDRTRY|RPC_CLNT_CREATE_SOFTERR)) {
501	clnt->cl_softrtry = 0;
502	if (args->flags & RPC_CLNT_CREATE_SOFTERR)
503	clnt->cl_softerr = 1;
504	}
505
506	if (args->flags & RPC_CLNT_CREATE_AUTOBIND)
507	clnt->cl_autobind = 1;
508	if (args->flags & RPC_CLNT_CREATE_NO_RETRANS_TIMEOUT)
509	clnt->cl_noretranstimeo = 1;
510	if (args->flags & RPC_CLNT_CREATE_DISCRTRY)
511	clnt->cl_discrtry = 1;
512	if (!(args->flags & RPC_CLNT_CREATE_QUIET))
513	clnt->cl_chatty = 1;
514
515	return clnt;
516	}
517
518	/**
519	* rpc_create - create an RPC client and transport with one call
520	* @args: rpc_clnt create argument structure
521	*
522	* Creates and initializes an RPC transport and an RPC client.
523	*
524	* It can ping the server in order to determine if it is up, and to see if
525	* it supports this program and version. RPC_CLNT_CREATE_NOPING disables
526	* this behavior so asynchronous tasks can also use rpc_create.
527	*/
528	struct rpc_clnt *rpc_create(struct rpc_create_args *args)
529	{
530	struct rpc_xprt *xprt;
531	struct xprt_create xprtargs = {
532	.net = args->net,
533	.ident = args->protocol,
534	.srcaddr = args->saddress,
535	.dstaddr = args->address,
536	.addrlen = args->addrsize,
537	.servername = args->servername,
538	.bc_xprt = args->bc_xprt,
539	.xprtsec = args->xprtsec,
540	.connect_timeout = args->connect_timeout,
541	.reconnect_timeout = args->reconnect_timeout,
542	};
543	char servername[48];
544	struct rpc_clnt *clnt;
545	int i;
546
547	if (args->bc_xprt) {
548	WARN_ON_ONCE(!(args->protocol & XPRT_TRANSPORT_BC));
549	xprt = args->bc_xprt->xpt_bc_xprt;
550	if (xprt) {
551	xprt_get(xprt);
552	return rpc_create_xprt(args, xprt);
553	}
554	}
555
556	if (args->flags & RPC_CLNT_CREATE_INFINITE_SLOTS)
557	xprtargs.flags |= XPRT_CREATE_INFINITE_SLOTS;
558	if (args->flags & RPC_CLNT_CREATE_NO_IDLE_TIMEOUT)
559	xprtargs.flags |= XPRT_CREATE_NO_IDLE_TIMEOUT;
560	/*
561	* If the caller chooses not to specify a hostname, whip
562	* up a string representation of the passed-in address.
563	*/
564	if (xprtargs.servername == NULL) {
565	struct sockaddr_un *sun =
566	(struct sockaddr_un *)args->address;
567	struct sockaddr_in *sin =
568	(struct sockaddr_in *)args->address;
569	struct sockaddr_in6 *sin6 =
570	(struct sockaddr_in6 *)args->address;
571
572	servername[0] = '\0';
573	switch (args->address->sa_family) {
574	case AF_LOCAL:
575	if (sun->sun_path[0])
576	snprintf(buf: servername, size: sizeof(servername), fmt: "%s",
577	sun->sun_path);
578	else
579	snprintf(buf: servername, size: sizeof(servername), fmt: "@%s",
580	sun->sun_path+1);
581	break;
582	case AF_INET:
583	snprintf(buf: servername, size: sizeof(servername), fmt: "%pI4",
584	&sin->sin_addr.s_addr);
585	break;
586	case AF_INET6:
587	snprintf(buf: servername, size: sizeof(servername), fmt: "%pI6",
588	&sin6->sin6_addr);
589	break;
590	default:
591	/* caller wants default server name, but
592	* address family isn't recognized. */
593	return ERR_PTR(error: -EINVAL);
594	}
595	xprtargs.servername = servername;
596	}
597
598	xprt = xprt_create_transport(args: &xprtargs);
599	if (IS_ERR(ptr: xprt))
600	return (struct rpc_clnt *)xprt;
601
602	/*
603	* By default, kernel RPC client connects from a reserved port.
604	* CAP_NET_BIND_SERVICE will not be set for unprivileged requesters,
605	* but it is always enabled for rpciod, which handles the connect
606	* operation.
607	*/
608	xprt->resvport = 1;
609	if (args->flags & RPC_CLNT_CREATE_NONPRIVPORT)
610	xprt->resvport = 0;
611	xprt->reuseport = 0;
612	if (args->flags & RPC_CLNT_CREATE_REUSEPORT)
613	xprt->reuseport = 1;
614
615	clnt = rpc_create_xprt(args, xprt);
616	if (IS_ERR(ptr: clnt) || args->nconnect <= 1)
617	return clnt;
618
619	for (i = 0; i < args->nconnect - 1; i++) {
620	if (rpc_clnt_add_xprt(clnt, &xprtargs, NULL, NULL) < 0)
621	break;
622	}
623	return clnt;
624	}
625	EXPORT_SYMBOL_GPL(rpc_create);
626
627	/*
628	* This function clones the RPC client structure. It allows us to share the
629	* same transport while varying parameters such as the authentication
630	* flavour.
631	*/
632	static struct rpc_clnt *__rpc_clone_client(struct rpc_create_args *args,
633	struct rpc_clnt *clnt)
634	{
635	struct rpc_xprt_switch *xps;
636	struct rpc_xprt *xprt;
637	struct rpc_clnt *new;
638	int err;
639
640	err = -ENOMEM;
641	rcu_read_lock();
642	xprt = xprt_get(rcu_dereference(clnt->cl_xprt));
643	xps = xprt_switch_get(rcu_dereference(clnt->cl_xpi.xpi_xpswitch));
644	rcu_read_unlock();
645	if (xprt == NULL || xps == NULL) {
646	xprt_put(xprt);
647	xprt_switch_put(xps);
648	goto out_err;
649	}
650	args->servername = xprt->servername;
651	args->nodename = clnt->cl_nodename;
652
653	new = rpc_new_client(args, xps, xprt, parent: clnt);
654	if (IS_ERR(ptr: new))
655	return new;
656
657	/* Turn off autobind on clones */
658	new->cl_autobind = 0;
659	new->cl_softrtry = clnt->cl_softrtry;
660	new->cl_softerr = clnt->cl_softerr;
661	new->cl_noretranstimeo = clnt->cl_noretranstimeo;
662	new->cl_discrtry = clnt->cl_discrtry;
663	new->cl_chatty = clnt->cl_chatty;
664	new->cl_principal = clnt->cl_principal;
665	new->cl_max_connect = clnt->cl_max_connect;
666	return new;
667
668	out_err:
669	trace_rpc_clnt_clone_err(clnt, error: err);
670	return ERR_PTR(error: err);
671	}
672
673	/**
674	* rpc_clone_client - Clone an RPC client structure
675	*
676	* @clnt: RPC client whose parameters are copied
677	*
678	* Returns a fresh RPC client or an ERR_PTR.
679	*/
680	struct rpc_clnt *rpc_clone_client(struct rpc_clnt *clnt)
681	{
682	struct rpc_create_args args = {
683	.program = clnt->cl_program,
684	.prognumber = clnt->cl_prog,
685	.version = clnt->cl_vers,
686	.authflavor = clnt->cl_auth->au_flavor,
687	.cred = clnt->cl_cred,
688	};
689	return __rpc_clone_client(args: &args, clnt);
690	}
691	EXPORT_SYMBOL_GPL(rpc_clone_client);
692
693	/**
694	* rpc_clone_client_set_auth - Clone an RPC client structure and set its auth
695	*
696	* @clnt: RPC client whose parameters are copied
697	* @flavor: security flavor for new client
698	*
699	* Returns a fresh RPC client or an ERR_PTR.
700	*/
701	struct rpc_clnt *
702	rpc_clone_client_set_auth(struct rpc_clnt *clnt, rpc_authflavor_t flavor)
703	{
704	struct rpc_create_args args = {
705	.program = clnt->cl_program,
706	.prognumber = clnt->cl_prog,
707	.version = clnt->cl_vers,
708	.authflavor = flavor,
709	.cred = clnt->cl_cred,
710	};
711	return __rpc_clone_client(args: &args, clnt);
712	}
713	EXPORT_SYMBOL_GPL(rpc_clone_client_set_auth);
714
715	/**
716	* rpc_switch_client_transport: switch the RPC transport on the fly
717	* @clnt: pointer to a struct rpc_clnt
718	* @args: pointer to the new transport arguments
719	* @timeout: pointer to the new timeout parameters
720	*
721	* This function allows the caller to switch the RPC transport for the
722	* rpc_clnt structure 'clnt' to allow it to connect to a mirrored NFS
723	* server, for instance. It assumes that the caller has ensured that
724	* there are no active RPC tasks by using some form of locking.
725	*
726	* Returns zero if "clnt" is now using the new xprt. Otherwise a
727	* negative errno is returned, and "clnt" continues to use the old
728	* xprt.
729	*/
730	int rpc_switch_client_transport(struct rpc_clnt *clnt,
731	struct xprt_create *args,
732	const struct rpc_timeout *timeout)
733	{
734	const struct rpc_timeout *old_timeo;
735	rpc_authflavor_t pseudoflavor;
736	struct rpc_xprt_switch *xps, *oldxps;
737	struct rpc_xprt *xprt, *old;
738	struct rpc_clnt *parent;
739	int err;
740
741	args->xprtsec = clnt->cl_xprtsec;
742	xprt = xprt_create_transport(args);
743	if (IS_ERR(ptr: xprt))
744	return PTR_ERR(ptr: xprt);
745
746	xps = xprt_switch_alloc(xprt, GFP_KERNEL);
747	if (xps == NULL) {
748	xprt_put(xprt);
749	return -ENOMEM;
750	}
751
752	pseudoflavor = clnt->cl_auth->au_flavor;
753
754	old_timeo = clnt->cl_timeout;
755	old = rpc_clnt_set_transport(clnt, xprt, timeout);
756	oldxps = xprt_iter_xchg_switch(xpi: &clnt->cl_xpi, newswitch: xps);
757
758	rpc_unregister_client(clnt);
759	__rpc_clnt_remove_pipedir(clnt);
760	rpc_sysfs_client_destroy(clnt);
761	rpc_clnt_debugfs_unregister(clnt);
762
763	/*
764	* A new transport was created. "clnt" therefore
765	* becomes the root of a new cl_parent tree. clnt's
766	* children, if it has any, still point to the old xprt.
767	*/
768	parent = clnt->cl_parent;
769	clnt->cl_parent = clnt;
770
771	/*
772	* The old rpc_auth cache cannot be re-used. GSS
773	* contexts in particular are between a single
774	* client and server.
775	*/
776	err = rpc_client_register(clnt, pseudoflavor, NULL);
777	if (err)
778	goto out_revert;
779
780	synchronize_rcu();
781	if (parent != clnt)
782	rpc_release_client(parent);
783	xprt_switch_put(xps: oldxps);
784	xprt_put(xprt: old);
785	trace_rpc_clnt_replace_xprt(clnt);
786	return 0;
787
788	out_revert:
789	xps = xprt_iter_xchg_switch(xpi: &clnt->cl_xpi, newswitch: oldxps);
790	rpc_clnt_set_transport(clnt, xprt: old, timeout: old_timeo);
791	clnt->cl_parent = parent;
792	rpc_client_register(clnt, pseudoflavor, NULL);
793	xprt_switch_put(xps);
794	xprt_put(xprt);
795	trace_rpc_clnt_replace_xprt_err(clnt);
796	return err;
797	}
798	EXPORT_SYMBOL_GPL(rpc_switch_client_transport);
799
800	static
801	int _rpc_clnt_xprt_iter_init(struct rpc_clnt *clnt, struct rpc_xprt_iter *xpi,
802	void func(struct rpc_xprt_iter *xpi, struct rpc_xprt_switch *xps))
803	{
804	struct rpc_xprt_switch *xps;
805
806	rcu_read_lock();
807	xps = xprt_switch_get(rcu_dereference(clnt->cl_xpi.xpi_xpswitch));
808	rcu_read_unlock();
809	if (xps == NULL)
810	return -EAGAIN;
811	func(xpi, xps);
812	xprt_switch_put(xps);
813	return 0;
814	}
815
816	static
817	int rpc_clnt_xprt_iter_init(struct rpc_clnt *clnt, struct rpc_xprt_iter *xpi)
818	{
819	return _rpc_clnt_xprt_iter_init(clnt, xpi, func: xprt_iter_init_listall);
820	}
821
822	static
823	int rpc_clnt_xprt_iter_offline_init(struct rpc_clnt *clnt,
824	struct rpc_xprt_iter *xpi)
825	{
826	return _rpc_clnt_xprt_iter_init(clnt, xpi, func: xprt_iter_init_listoffline);
827	}
828
829	/**
830	* rpc_clnt_iterate_for_each_xprt - Apply a function to all transports
831	* @clnt: pointer to client
832	* @fn: function to apply
833	* @data: void pointer to function data
834	*
835	* Iterates through the list of RPC transports currently attached to the
836	* client and applies the function fn(clnt, xprt, data).
837	*
838	* On error, the iteration stops, and the function returns the error value.
839	*/
840	int rpc_clnt_iterate_for_each_xprt(struct rpc_clnt *clnt,
841	int (*fn)(struct rpc_clnt *, struct rpc_xprt *, void *),
842	void *data)
843	{
844	struct rpc_xprt_iter xpi;
845	int ret;
846
847	ret = rpc_clnt_xprt_iter_init(clnt, xpi: &xpi);
848	if (ret)
849	return ret;
850	for (;;) {
851	struct rpc_xprt *xprt = xprt_iter_get_next(xpi: &xpi);
852
853	if (!xprt)
854	break;
855	ret = fn(clnt, xprt, data);
856	xprt_put(xprt);
857	if (ret < 0)
858	break;
859	}
860	xprt_iter_destroy(xpi: &xpi);
861	return ret;
862	}
863	EXPORT_SYMBOL_GPL(rpc_clnt_iterate_for_each_xprt);
864
865	/*
866	* Kill all tasks for the given client.
867	* XXX: kill their descendants as well?
868	*/
869	void rpc_killall_tasks(struct rpc_clnt *clnt)
870	{
871	struct rpc_task *rovr;
872
873
874	if (list_empty(head: &clnt->cl_tasks))
875	return;
876
877	/*
878	* Spin lock all_tasks to prevent changes...
879	*/
880	trace_rpc_clnt_killall(clnt);
881	spin_lock(lock: &clnt->cl_lock);
882	list_for_each_entry(rovr, &clnt->cl_tasks, tk_task)
883	rpc_signal_task(rovr);
884	spin_unlock(lock: &clnt->cl_lock);
885	}
886	EXPORT_SYMBOL_GPL(rpc_killall_tasks);
887
888	/**
889	* rpc_cancel_tasks - try to cancel a set of RPC tasks
890	* @clnt: Pointer to RPC client
891	* @error: RPC task error value to set
892	* @fnmatch: Pointer to selector function
893	* @data: User data
894	*
895	* Uses @fnmatch to define a set of RPC tasks that are to be cancelled.
896	* The argument @error must be a negative error value.
897	*/
898	unsigned long rpc_cancel_tasks(struct rpc_clnt *clnt, int error,
899	bool (*fnmatch)(const struct rpc_task *,
900	const void *),
901	const void *data)
902	{
903	struct rpc_task *task;
904	unsigned long count = 0;
905
906	if (list_empty(head: &clnt->cl_tasks))
907	return 0;
908	/*
909	* Spin lock all_tasks to prevent changes...
910	*/
911	spin_lock(lock: &clnt->cl_lock);
912	list_for_each_entry(task, &clnt->cl_tasks, tk_task) {
913	if (!RPC_IS_ACTIVATED(task))
914	continue;
915	if (!fnmatch(task, data))
916	continue;
917	rpc_task_try_cancel(task, error);
918	count++;
919	}
920	spin_unlock(lock: &clnt->cl_lock);
921	return count;
922	}
923	EXPORT_SYMBOL_GPL(rpc_cancel_tasks);
924
925	static int rpc_clnt_disconnect_xprt(struct rpc_clnt *clnt,
926	struct rpc_xprt *xprt, void *dummy)
927	{
928	if (xprt_connected(xprt))
929	xprt_force_disconnect(xprt);
930	return 0;
931	}
932
933	void rpc_clnt_disconnect(struct rpc_clnt *clnt)
934	{
935	rpc_clnt_iterate_for_each_xprt(clnt, rpc_clnt_disconnect_xprt, NULL);
936	}
937	EXPORT_SYMBOL_GPL(rpc_clnt_disconnect);
938
939	/*
940	* Properly shut down an RPC client, terminating all outstanding
941	* requests.
942	*/
943	void rpc_shutdown_client(struct rpc_clnt *clnt)
944	{
945	might_sleep();
946
947	trace_rpc_clnt_shutdown(clnt);
948
949	while (!list_empty(head: &clnt->cl_tasks)) {
950	rpc_killall_tasks(clnt);
951	wait_event_timeout(destroy_wait,
952	list_empty(&clnt->cl_tasks), 1*HZ);
953	}
954
955	rpc_release_client(clnt);
956	}
957	EXPORT_SYMBOL_GPL(rpc_shutdown_client);
958
959	/*
960	* Free an RPC client
961	*/
962	static void rpc_free_client_work(struct work_struct *work)
963	{
964	struct rpc_clnt *clnt = container_of(work, struct rpc_clnt, cl_work);
965
966	trace_rpc_clnt_free(clnt);
967
968	/* These might block on processes that might allocate memory,
969	* so they cannot be called in rpciod, so they are handled separately
970	* here.
971	*/
972	rpc_sysfs_client_destroy(clnt);
973	rpc_clnt_debugfs_unregister(clnt);
974	rpc_free_clid(clnt);
975	rpc_clnt_remove_pipedir(clnt);
976	xprt_put(rcu_dereference_raw(clnt->cl_xprt));
977
978	kfree(objp: clnt);
979	rpciod_down();
980	}
981	static struct rpc_clnt *
982	rpc_free_client(struct rpc_clnt *clnt)
983	{
984	struct rpc_clnt *parent = NULL;
985
986	trace_rpc_clnt_release(clnt);
987	if (clnt->cl_parent != clnt)
988	parent = clnt->cl_parent;
989	rpc_unregister_client(clnt);
990	rpc_free_iostats(clnt->cl_metrics);
991	clnt->cl_metrics = NULL;
992	xprt_iter_destroy(xpi: &clnt->cl_xpi);
993	put_cred(cred: clnt->cl_cred);
994
995	INIT_WORK(&clnt->cl_work, rpc_free_client_work);
996	schedule_work(work: &clnt->cl_work);
997	return parent;
998	}
999
1000	/*
1001	* Free an RPC client
1002	*/
1003	static struct rpc_clnt *
1004	rpc_free_auth(struct rpc_clnt *clnt)
1005	{
1006	/*
1007	* Note: RPCSEC_GSS may need to send NULL RPC calls in order to
1008	* release remaining GSS contexts. This mechanism ensures
1009	* that it can do so safely.
1010	*/
1011	if (clnt->cl_auth != NULL) {
1012	rpcauth_release(clnt->cl_auth);
1013	clnt->cl_auth = NULL;
1014	}
1015	if (refcount_dec_and_test(r: &clnt->cl_count))
1016	return rpc_free_client(clnt);
1017	return NULL;
1018	}
1019
1020	/*
1021	* Release reference to the RPC client
1022	*/
1023	void
1024	rpc_release_client(struct rpc_clnt *clnt)
1025	{
1026	do {
1027	if (list_empty(head: &clnt->cl_tasks))
1028	wake_up(&destroy_wait);
1029	if (refcount_dec_not_one(r: &clnt->cl_count))
1030	break;
1031	clnt = rpc_free_auth(clnt);
1032	} while (clnt != NULL);
1033	}
1034	EXPORT_SYMBOL_GPL(rpc_release_client);
1035
1036	/**
1037	* rpc_bind_new_program - bind a new RPC program to an existing client
1038	* @old: old rpc_client
1039	* @program: rpc program to set
1040	* @vers: rpc program version
1041	*
1042	* Clones the rpc client and sets up a new RPC program. This is mainly
1043	* of use for enabling different RPC programs to share the same transport.
1044	* The Sun NFSv2/v3 ACL protocol can do this.
1045	*/
1046	struct rpc_clnt *rpc_bind_new_program(struct rpc_clnt *old,
1047	const struct rpc_program *program,
1048	u32 vers)
1049	{
1050	struct rpc_create_args args = {
1051	.program = program,
1052	.prognumber = program->number,
1053	.version = vers,
1054	.authflavor = old->cl_auth->au_flavor,
1055	.cred = old->cl_cred,
1056	};
1057	struct rpc_clnt *clnt;
1058	int err;
1059
1060	clnt = __rpc_clone_client(args: &args, clnt: old);
1061	if (IS_ERR(ptr: clnt))
1062	goto out;
1063	err = rpc_ping(clnt);
1064	if (err != 0) {
1065	rpc_shutdown_client(clnt);
1066	clnt = ERR_PTR(error: err);
1067	}
1068	out:
1069	return clnt;
1070	}
1071	EXPORT_SYMBOL_GPL(rpc_bind_new_program);
1072
1073	struct rpc_xprt *
1074	rpc_task_get_xprt(struct rpc_clnt *clnt, struct rpc_xprt *xprt)
1075	{
1076	struct rpc_xprt_switch *xps;
1077
1078	if (!xprt)
1079	return NULL;
1080	rcu_read_lock();
1081	xps = rcu_dereference(clnt->cl_xpi.xpi_xpswitch);
1082	atomic_long_inc(v: &xps->xps_queuelen);
1083	rcu_read_unlock();
1084	atomic_long_inc(v: &xprt->queuelen);
1085
1086	return xprt;
1087	}
1088
1089	static void
1090	rpc_task_release_xprt(struct rpc_clnt *clnt, struct rpc_xprt *xprt)
1091	{
1092	struct rpc_xprt_switch *xps;
1093
1094	atomic_long_dec(v: &xprt->queuelen);
1095	rcu_read_lock();
1096	xps = rcu_dereference(clnt->cl_xpi.xpi_xpswitch);
1097	atomic_long_dec(v: &xps->xps_queuelen);
1098	rcu_read_unlock();
1099
1100	xprt_put(xprt);
1101	}
1102
1103	void rpc_task_release_transport(struct rpc_task *task)
1104	{
1105	struct rpc_xprt *xprt = task->tk_xprt;
1106
1107	if (xprt) {
1108	task->tk_xprt = NULL;
1109	if (task->tk_client)
1110	rpc_task_release_xprt(clnt: task->tk_client, xprt);
1111	else
1112	xprt_put(xprt);
1113	}
1114	}
1115	EXPORT_SYMBOL_GPL(rpc_task_release_transport);
1116
1117	void rpc_task_release_client(struct rpc_task *task)
1118	{
1119	struct rpc_clnt *clnt = task->tk_client;
1120
1121	rpc_task_release_transport(task);
1122	if (clnt != NULL) {
1123	/* Remove from client task list */
1124	spin_lock(lock: &clnt->cl_lock);
1125	list_del(entry: &task->tk_task);
1126	spin_unlock(lock: &clnt->cl_lock);
1127	task->tk_client = NULL;
1128
1129	rpc_release_client(clnt);
1130	}
1131	}
1132
1133	static struct rpc_xprt *
1134	rpc_task_get_first_xprt(struct rpc_clnt *clnt)
1135	{
1136	struct rpc_xprt *xprt;
1137
1138	rcu_read_lock();
1139	xprt = xprt_get(rcu_dereference(clnt->cl_xprt));
1140	rcu_read_unlock();
1141	return rpc_task_get_xprt(clnt, xprt);
1142	}
1143
1144	static struct rpc_xprt *
1145	rpc_task_get_next_xprt(struct rpc_clnt *clnt)
1146	{
1147	return rpc_task_get_xprt(clnt, xprt: xprt_iter_get_next(xpi: &clnt->cl_xpi));
1148	}
1149
1150	static
1151	void rpc_task_set_transport(struct rpc_task *task, struct rpc_clnt *clnt)
1152	{
1153	if (task->tk_xprt) {
1154	if (!(test_bit(XPRT_OFFLINE, &task->tk_xprt->state) &&
1155	(task->tk_flags & RPC_TASK_MOVEABLE)))
1156	return;
1157	xprt_release(task);
1158	xprt_put(xprt: task->tk_xprt);
1159	}
1160	if (task->tk_flags & RPC_TASK_NO_ROUND_ROBIN)
1161	task->tk_xprt = rpc_task_get_first_xprt(clnt);
1162	else
1163	task->tk_xprt = rpc_task_get_next_xprt(clnt);
1164	}
1165
1166	static
1167	void rpc_task_set_client(struct rpc_task *task, struct rpc_clnt *clnt)
1168	{
1169	rpc_task_set_transport(task, clnt);
1170	task->tk_client = clnt;
1171	refcount_inc(r: &clnt->cl_count);
1172	if (clnt->cl_softrtry)
1173	task->tk_flags |= RPC_TASK_SOFT;
1174	if (clnt->cl_softerr)
1175	task->tk_flags |= RPC_TASK_TIMEOUT;
1176	if (clnt->cl_noretranstimeo)
1177	task->tk_flags |= RPC_TASK_NO_RETRANS_TIMEOUT;
1178	/* Add to the client's list of all tasks */
1179	spin_lock(lock: &clnt->cl_lock);
1180	list_add_tail(new: &task->tk_task, head: &clnt->cl_tasks);
1181	spin_unlock(lock: &clnt->cl_lock);
1182	}
1183
1184	static void
1185	rpc_task_set_rpc_message(struct rpc_task *task, const struct rpc_message *msg)
1186	{
1187	if (msg != NULL) {
1188	task->tk_msg.rpc_proc = msg->rpc_proc;
1189	task->tk_msg.rpc_argp = msg->rpc_argp;
1190	task->tk_msg.rpc_resp = msg->rpc_resp;
1191	task->tk_msg.rpc_cred = msg->rpc_cred;
1192	if (!(task->tk_flags & RPC_TASK_CRED_NOREF))
1193	get_cred(cred: task->tk_msg.rpc_cred);
1194	}
1195	}
1196
1197	/*
1198	* Default callback for async RPC calls
1199	*/
1200	static void
1201	rpc_default_callback(struct rpc_task *task, void *data)
1202	{
1203	}
1204
1205	static const struct rpc_call_ops rpc_default_ops = {
1206	.rpc_call_done = rpc_default_callback,
1207	};
1208
1209	/**
1210	* rpc_run_task - Allocate a new RPC task, then run rpc_execute against it
1211	* @task_setup_data: pointer to task initialisation data
1212	*/
1213	struct rpc_task *rpc_run_task(const struct rpc_task_setup *task_setup_data)
1214	{
1215	struct rpc_task *task;
1216
1217	task = rpc_new_task(task_setup_data);
1218	if (IS_ERR(ptr: task))
1219	return task;
1220
1221	if (!RPC_IS_ASYNC(task))
1222	task->tk_flags |= RPC_TASK_CRED_NOREF;
1223
1224	rpc_task_set_client(task, clnt: task_setup_data->rpc_client);
1225	rpc_task_set_rpc_message(task, msg: task_setup_data->rpc_message);
1226
1227	if (task->tk_action == NULL)
1228	rpc_call_start(task);
1229
1230	atomic_inc(v: &task->tk_count);
1231	rpc_execute(task);
1232	return task;
1233	}
1234	EXPORT_SYMBOL_GPL(rpc_run_task);
1235
1236	/**
1237	* rpc_call_sync - Perform a synchronous RPC call
1238	* @clnt: pointer to RPC client
1239	* @msg: RPC call parameters
1240	* @flags: RPC call flags
1241	*/
1242	int rpc_call_sync(struct rpc_clnt *clnt, const struct rpc_message *msg, int flags)
1243	{
1244	struct rpc_task *task;
1245	struct rpc_task_setup task_setup_data = {
1246	.rpc_client = clnt,
1247	.rpc_message = msg,
1248	.callback_ops = &rpc_default_ops,
1249	.flags = flags,
1250	};
1251	int status;
1252
1253	WARN_ON_ONCE(flags & RPC_TASK_ASYNC);
1254	if (flags & RPC_TASK_ASYNC) {
1255	rpc_release_calldata(task_setup_data.callback_ops,
1256	task_setup_data.callback_data);
1257	return -EINVAL;
1258	}
1259
1260	task = rpc_run_task(&task_setup_data);
1261	if (IS_ERR(ptr: task))
1262	return PTR_ERR(ptr: task);
1263	status = task->tk_status;
1264	rpc_put_task(task);
1265	return status;
1266	}
1267	EXPORT_SYMBOL_GPL(rpc_call_sync);
1268
1269	/**
1270	* rpc_call_async - Perform an asynchronous RPC call
1271	* @clnt: pointer to RPC client
1272	* @msg: RPC call parameters
1273	* @flags: RPC call flags
1274	* @tk_ops: RPC call ops
1275	* @data: user call data
1276	*/
1277	int
1278	rpc_call_async(struct rpc_clnt *clnt, const struct rpc_message *msg, int flags,
1279	const struct rpc_call_ops *tk_ops, void *data)
1280	{
1281	struct rpc_task *task;
1282	struct rpc_task_setup task_setup_data = {
1283	.rpc_client = clnt,
1284	.rpc_message = msg,
1285	.callback_ops = tk_ops,
1286	.callback_data = data,
1287	.flags = flags|RPC_TASK_ASYNC,
1288	};
1289
1290	task = rpc_run_task(&task_setup_data);
1291	if (IS_ERR(ptr: task))
1292	return PTR_ERR(ptr: task);
1293	rpc_put_task(task);
1294	return 0;
1295	}
1296	EXPORT_SYMBOL_GPL(rpc_call_async);
1297
1298	#if defined(CONFIG_SUNRPC_BACKCHANNEL)
1299	static void call_bc_encode(struct rpc_task *task);
1300
1301	/**
1302	* rpc_run_bc_task - Allocate a new RPC task for backchannel use, then run
1303	* rpc_execute against it
1304	* @req: RPC request
1305	*/
1306	struct rpc_task *rpc_run_bc_task(struct rpc_rqst *req)
1307	{
1308	struct rpc_task *task;
1309	struct rpc_task_setup task_setup_data = {
1310	.callback_ops = &rpc_default_ops,
1311	.flags = RPC_TASK_SOFTCONN |
1312	RPC_TASK_NO_RETRANS_TIMEOUT,
1313	};
1314
1315	dprintk("RPC: rpc_run_bc_task req= %p\n", req);
1316	/*
1317	* Create an rpc_task to send the data
1318	*/
1319	task = rpc_new_task(&task_setup_data);
1320	if (IS_ERR(ptr: task)) {
1321	xprt_free_bc_request(req);
1322	return task;
1323	}
1324
1325	xprt_init_bc_request(req, task);
1326
1327	task->tk_action = call_bc_encode;
1328	atomic_inc(v: &task->tk_count);
1329	WARN_ON_ONCE(atomic_read(&task->tk_count) != 2);
1330	rpc_execute(task);
1331
1332	dprintk("RPC: rpc_run_bc_task: task= %p\n", task);
1333	return task;
1334	}
1335	#endif /* CONFIG_SUNRPC_BACKCHANNEL */
1336
1337	/**
1338	* rpc_prepare_reply_pages - Prepare to receive a reply data payload into pages
1339	* @req: RPC request to prepare
1340	* @pages: vector of struct page pointers
1341	* @base: offset in first page where receive should start, in bytes
1342	* @len: expected size of the upper layer data payload, in bytes
1343	* @hdrsize: expected size of upper layer reply header, in XDR words
1344	*
1345	*/
1346	void rpc_prepare_reply_pages(struct rpc_rqst *req, struct page **pages,
1347	unsigned int base, unsigned int len,
1348	unsigned int hdrsize)
1349	{
1350	hdrsize += RPC_REPHDRSIZE + req->rq_cred->cr_auth->au_ralign;
1351
1352	xdr_inline_pages(&req->rq_rcv_buf, hdrsize << 2, pages, base, len);
1353	trace_rpc_xdr_reply_pages(task: req->rq_task, xdr: &req->rq_rcv_buf);
1354	}
1355	EXPORT_SYMBOL_GPL(rpc_prepare_reply_pages);
1356
1357	void
1358	rpc_call_start(struct rpc_task *task)
1359	{
1360	task->tk_action = call_start;
1361	}
1362	EXPORT_SYMBOL_GPL(rpc_call_start);
1363
1364	/**
1365	* rpc_peeraddr - extract remote peer address from clnt's xprt
1366	* @clnt: RPC client structure
1367	* @buf: target buffer
1368	* @bufsize: length of target buffer
1369	*
1370	* Returns the number of bytes that are actually in the stored address.
1371	*/
1372	size_t rpc_peeraddr(struct rpc_clnt *clnt, struct sockaddr *buf, size_t bufsize)
1373	{
1374	size_t bytes;
1375	struct rpc_xprt *xprt;
1376
1377	rcu_read_lock();
1378	xprt = rcu_dereference(clnt->cl_xprt);
1379
1380	bytes = xprt->addrlen;
1381	if (bytes > bufsize)
1382	bytes = bufsize;
1383	memcpy(buf, &xprt->addr, bytes);
1384	rcu_read_unlock();
1385
1386	return bytes;
1387	}
1388	EXPORT_SYMBOL_GPL(rpc_peeraddr);
1389
1390	/**
1391	* rpc_peeraddr2str - return remote peer address in printable format
1392	* @clnt: RPC client structure
1393	* @format: address format
1394	*
1395	* NB: the lifetime of the memory referenced by the returned pointer is
1396	* the same as the rpc_xprt itself. As long as the caller uses this
1397	* pointer, it must hold the RCU read lock.
1398	*/
1399	const char *rpc_peeraddr2str(struct rpc_clnt *clnt,
1400	enum rpc_display_format_t format)
1401	{
1402	struct rpc_xprt *xprt;
1403
1404	xprt = rcu_dereference(clnt->cl_xprt);
1405
1406	if (xprt->address_strings[format] != NULL)
1407	return xprt->address_strings[format];
1408	else
1409	return "unprintable";
1410	}
1411	EXPORT_SYMBOL_GPL(rpc_peeraddr2str);
1412
1413	static const struct sockaddr_in rpc_inaddr_loopback = {
1414	.sin_family = AF_INET,
1415	.sin_addr.s_addr = htonl(INADDR_ANY),
1416	};
1417
1418	static const struct sockaddr_in6 rpc_in6addr_loopback = {
1419	.sin6_family = AF_INET6,
1420	.sin6_addr = IN6ADDR_ANY_INIT,
1421	};
1422
1423	/*
1424	* Try a getsockname() on a connected datagram socket. Using a
1425	* connected datagram socket prevents leaving a socket in TIME_WAIT.
1426	* This conserves the ephemeral port number space.
1427	*
1428	* Returns zero and fills in "buf" if successful; otherwise, a
1429	* negative errno is returned.
1430	*/
1431	static int rpc_sockname(struct net *net, struct sockaddr *sap, size_t salen,
1432	struct sockaddr *buf)
1433	{
1434	struct socket *sock;
1435	int err;
1436
1437	err = __sock_create(net, family: sap->sa_family,
1438	type: SOCK_DGRAM, IPPROTO_UDP, res: &sock, kern: 1);
1439	if (err < 0) {
1440	dprintk("RPC: can't create UDP socket (%d)\n", err);
1441	goto out;
1442	}
1443
1444	switch (sap->sa_family) {
1445	case AF_INET:
1446	err = kernel_bind(sock,
1447	addr: (struct sockaddr *)&rpc_inaddr_loopback,
1448	addrlen: sizeof(rpc_inaddr_loopback));
1449	break;
1450	case AF_INET6:
1451	err = kernel_bind(sock,
1452	addr: (struct sockaddr *)&rpc_in6addr_loopback,
1453	addrlen: sizeof(rpc_in6addr_loopback));
1454	break;
1455	default:
1456	err = -EAFNOSUPPORT;
1457	goto out_release;
1458	}
1459	if (err < 0) {
1460	dprintk("RPC: can't bind UDP socket (%d)\n", err);
1461	goto out_release;
1462	}
1463
1464	err = kernel_connect(sock, addr: sap, addrlen: salen, flags: 0);
1465	if (err < 0) {
1466	dprintk("RPC: can't connect UDP socket (%d)\n", err);
1467	goto out_release;
1468	}
1469
1470	err = kernel_getsockname(sock, addr: buf);
1471	if (err < 0) {
1472	dprintk("RPC: getsockname failed (%d)\n", err);
1473	goto out_release;
1474	}
1475
1476	err = 0;
1477	if (buf->sa_family == AF_INET6) {
1478	struct sockaddr_in6 *sin6 = (struct sockaddr_in6 *)buf;
1479	sin6->sin6_scope_id = 0;
1480	}
1481	dprintk("RPC: %s succeeded\n", __func__);
1482
1483	out_release:
1484	sock_release(sock);
1485	out:
1486	return err;
1487	}
1488
1489	/*
1490	* Scraping a connected socket failed, so we don't have a useable
1491	* local address. Fallback: generate an address that will prevent
1492	* the server from calling us back.
1493	*
1494	* Returns zero and fills in "buf" if successful; otherwise, a
1495	* negative errno is returned.
1496	*/
1497	static int rpc_anyaddr(int family, struct sockaddr *buf, size_t buflen)
1498	{
1499	switch (family) {
1500	case AF_INET:
1501	if (buflen < sizeof(rpc_inaddr_loopback))
1502	return -EINVAL;
1503	memcpy(buf, &rpc_inaddr_loopback,
1504	sizeof(rpc_inaddr_loopback));
1505	break;
1506	case AF_INET6:
1507	if (buflen < sizeof(rpc_in6addr_loopback))
1508	return -EINVAL;
1509	memcpy(buf, &rpc_in6addr_loopback,
1510	sizeof(rpc_in6addr_loopback));
1511	break;
1512	default:
1513	dprintk("RPC: %s: address family not supported\n",
1514	__func__);
1515	return -EAFNOSUPPORT;
1516	}
1517	dprintk("RPC: %s: succeeded\n", __func__);
1518	return 0;
1519	}
1520
1521	/**
1522	* rpc_localaddr - discover local endpoint address for an RPC client
1523	* @clnt: RPC client structure
1524	* @buf: target buffer
1525	* @buflen: size of target buffer, in bytes
1526	*
1527	* Returns zero and fills in "buf" and "buflen" if successful;
1528	* otherwise, a negative errno is returned.
1529	*
1530	* This works even if the underlying transport is not currently connected,
1531	* or if the upper layer never previously provided a source address.
1532	*
1533	* The result of this function call is transient: multiple calls in
1534	* succession may give different results, depending on how local
1535	* networking configuration changes over time.
1536	*/
1537	int rpc_localaddr(struct rpc_clnt *clnt, struct sockaddr *buf, size_t buflen)
1538	{
1539	struct sockaddr_storage address;
1540	struct sockaddr *sap = (struct sockaddr *)&address;
1541	struct rpc_xprt *xprt;
1542	struct net *net;
1543	size_t salen;
1544	int err;
1545
1546	rcu_read_lock();
1547	xprt = rcu_dereference(clnt->cl_xprt);
1548	salen = xprt->addrlen;
1549	memcpy(sap, &xprt->addr, salen);
1550	net = get_net(net: xprt->xprt_net);
1551	rcu_read_unlock();
1552
1553	rpc_set_port(sap, port: 0);
1554	err = rpc_sockname(net, sap, salen, buf);
1555	put_net(net);
1556	if (err != 0)
1557	/* Couldn't discover local address, return ANYADDR */
1558	return rpc_anyaddr(family: sap->sa_family, buf, buflen);
1559	return 0;
1560	}
1561	EXPORT_SYMBOL_GPL(rpc_localaddr);
1562
1563	void
1564	rpc_setbufsize(struct rpc_clnt *clnt, unsigned int sndsize, unsigned int rcvsize)
1565	{
1566	struct rpc_xprt *xprt;
1567
1568	rcu_read_lock();
1569	xprt = rcu_dereference(clnt->cl_xprt);
1570	if (xprt->ops->set_buffer_size)
1571	xprt->ops->set_buffer_size(xprt, sndsize, rcvsize);
1572	rcu_read_unlock();
1573	}
1574	EXPORT_SYMBOL_GPL(rpc_setbufsize);
1575
1576	/**
1577	* rpc_net_ns - Get the network namespace for this RPC client
1578	* @clnt: RPC client to query
1579	*
1580	*/
1581	struct net *rpc_net_ns(struct rpc_clnt *clnt)
1582	{
1583	struct net *ret;
1584
1585	rcu_read_lock();
1586	ret = rcu_dereference(clnt->cl_xprt)->xprt_net;
1587	rcu_read_unlock();
1588	return ret;
1589	}
1590	EXPORT_SYMBOL_GPL(rpc_net_ns);
1591
1592	/**
1593	* rpc_max_payload - Get maximum payload size for a transport, in bytes
1594	* @clnt: RPC client to query
1595	*
1596	* For stream transports, this is one RPC record fragment (see RFC
1597	* 1831), as we don't support multi-record requests yet. For datagram
1598	* transports, this is the size of an IP packet minus the IP, UDP, and
1599	* RPC header sizes.
1600	*/
1601	size_t rpc_max_payload(struct rpc_clnt *clnt)
1602	{
1603	size_t ret;
1604
1605	rcu_read_lock();
1606	ret = rcu_dereference(clnt->cl_xprt)->max_payload;
1607	rcu_read_unlock();
1608	return ret;
1609	}
1610	EXPORT_SYMBOL_GPL(rpc_max_payload);
1611
1612	/**
1613	* rpc_max_bc_payload - Get maximum backchannel payload size, in bytes
1614	* @clnt: RPC client to query
1615	*/
1616	size_t rpc_max_bc_payload(struct rpc_clnt *clnt)
1617	{
1618	struct rpc_xprt *xprt;
1619	size_t ret;
1620
1621	rcu_read_lock();
1622	xprt = rcu_dereference(clnt->cl_xprt);
1623	ret = xprt->ops->bc_maxpayload(xprt);
1624	rcu_read_unlock();
1625	return ret;
1626	}
1627	EXPORT_SYMBOL_GPL(rpc_max_bc_payload);
1628
1629	unsigned int rpc_num_bc_slots(struct rpc_clnt *clnt)
1630	{
1631	struct rpc_xprt *xprt;
1632	unsigned int ret;
1633
1634	rcu_read_lock();
1635	xprt = rcu_dereference(clnt->cl_xprt);
1636	ret = xprt->ops->bc_num_slots(xprt);
1637	rcu_read_unlock();
1638	return ret;
1639	}
1640	EXPORT_SYMBOL_GPL(rpc_num_bc_slots);
1641
1642	/**
1643	* rpc_force_rebind - force transport to check that remote port is unchanged
1644	* @clnt: client to rebind
1645	*
1646	*/
1647	void rpc_force_rebind(struct rpc_clnt *clnt)
1648	{
1649	if (clnt->cl_autobind) {
1650	rcu_read_lock();
1651	xprt_clear_bound(rcu_dereference(clnt->cl_xprt));
1652	rcu_read_unlock();
1653	}
1654	}
1655	EXPORT_SYMBOL_GPL(rpc_force_rebind);
1656
1657	static int
1658	__rpc_restart_call(struct rpc_task *task, void (*action)(struct rpc_task *))
1659	{
1660	task->tk_status = 0;
1661	task->tk_rpc_status = 0;
1662	task->tk_action = action;
1663	return 1;
1664	}
1665
1666	/*
1667	* Restart an (async) RPC call. Usually called from within the
1668	* exit handler.
1669	*/
1670	int
1671	rpc_restart_call(struct rpc_task *task)
1672	{
1673	return __rpc_restart_call(task, action: call_start);
1674	}
1675	EXPORT_SYMBOL_GPL(rpc_restart_call);
1676
1677	/*
1678	* Restart an (async) RPC call from the call_prepare state.
1679	* Usually called from within the exit handler.
1680	*/
1681	int
1682	rpc_restart_call_prepare(struct rpc_task *task)
1683	{
1684	if (task->tk_ops->rpc_call_prepare != NULL)
1685	return __rpc_restart_call(task, action: rpc_prepare_task);
1686	return rpc_restart_call(task);
1687	}
1688	EXPORT_SYMBOL_GPL(rpc_restart_call_prepare);
1689
1690	const char
1691	*rpc_proc_name(const struct rpc_task *task)
1692	{
1693	const struct rpc_procinfo *proc = task->tk_msg.rpc_proc;
1694
1695	if (proc) {
1696	if (proc->p_name)
1697	return proc->p_name;
1698	else
1699	return "NULL";
1700	} else
1701	return "no proc";
1702	}
1703
1704	static void
1705	__rpc_call_rpcerror(struct rpc_task *task, int tk_status, int rpc_status)
1706	{
1707	trace_rpc_call_rpcerror(task, tk_status, rpc_status);
1708	rpc_task_set_rpc_status(task, rpc_status);
1709	rpc_exit(task, tk_status);
1710	}
1711
1712	static void
1713	rpc_call_rpcerror(struct rpc_task *task, int status)
1714	{
1715	__rpc_call_rpcerror(task, tk_status: status, rpc_status: status);
1716	}
1717
1718	/*
1719	* 0. Initial state
1720	*
1721	* Other FSM states can be visited zero or more times, but
1722	* this state is visited exactly once for each RPC.
1723	*/
1724	static void
1725	call_start(struct rpc_task *task)
1726	{
1727	struct rpc_clnt *clnt = task->tk_client;
1728	int idx = task->tk_msg.rpc_proc->p_statidx;
1729
1730	trace_rpc_request(task);
1731
1732	if (task->tk_client->cl_shutdown) {
1733	rpc_call_rpcerror(task, status: -EIO);
1734	return;
1735	}
1736
1737	/* Increment call count (version might not be valid for ping) */
1738	if (clnt->cl_program->version[clnt->cl_vers])
1739	clnt->cl_program->version[clnt->cl_vers]->counts[idx]++;
1740	clnt->cl_stats->rpccnt++;
1741	task->tk_action = call_reserve;
1742	rpc_task_set_transport(task, clnt);
1743	}
1744
1745	/*
1746	* 1. Reserve an RPC call slot
1747	*/
1748	static void
1749	call_reserve(struct rpc_task *task)
1750	{
1751	task->tk_status = 0;
1752	task->tk_action = call_reserveresult;
1753	xprt_reserve(task);
1754	}
1755
1756	static void call_retry_reserve(struct rpc_task *task);
1757
1758	/*
1759	* 1b. Grok the result of xprt_reserve()
1760	*/
1761	static void
1762	call_reserveresult(struct rpc_task *task)
1763	{
1764	int status = task->tk_status;
1765
1766	/*
1767	* After a call to xprt_reserve(), we must have either
1768	* a request slot or else an error status.
1769	*/
1770	task->tk_status = 0;
1771	if (status >= 0) {
1772	if (task->tk_rqstp) {
1773	task->tk_action = call_refresh;
1774	return;
1775	}
1776
1777	rpc_call_rpcerror(task, status: -EIO);
1778	return;
1779	}
1780
1781	switch (status) {
1782	case -ENOMEM:
1783	rpc_delay(task, HZ >> 2);
1784	fallthrough;
1785	case -EAGAIN: /* woken up; retry */
1786	task->tk_action = call_retry_reserve;
1787	return;
1788	default:
1789	rpc_call_rpcerror(task, status);
1790	}
1791	}
1792
1793	/*
1794	* 1c. Retry reserving an RPC call slot
1795	*/
1796	static void
1797	call_retry_reserve(struct rpc_task *task)
1798	{
1799	task->tk_status = 0;
1800	task->tk_action = call_reserveresult;
1801	xprt_retry_reserve(task);
1802	}
1803
1804	/*
1805	* 2. Bind and/or refresh the credentials
1806	*/
1807	static void
1808	call_refresh(struct rpc_task *task)
1809	{
1810	task->tk_action = call_refreshresult;
1811	task->tk_status = 0;
1812	task->tk_client->cl_stats->rpcauthrefresh++;
1813	rpcauth_refreshcred(task);
1814	}
1815
1816	/*
1817	* 2a. Process the results of a credential refresh
1818	*/
1819	static void
1820	call_refreshresult(struct rpc_task *task)
1821	{
1822	int status = task->tk_status;
1823
1824	task->tk_status = 0;
1825	task->tk_action = call_refresh;
1826	switch (status) {
1827	case 0:
1828	if (rpcauth_uptodatecred(task)) {
1829	task->tk_action = call_allocate;
1830	return;
1831	}
1832	/* Use rate-limiting and a max number of retries if refresh
1833	* had status 0 but failed to update the cred.
1834	*/
1835	fallthrough;
1836	case -ETIMEDOUT:
1837	rpc_delay(task, 3*HZ);
1838	fallthrough;
1839	case -EAGAIN:
1840	status = -EACCES;
1841	fallthrough;
1842	case -EKEYEXPIRED:
1843	if (!task->tk_cred_retry)
1844	break;
1845	task->tk_cred_retry--;
1846	trace_rpc_retry_refresh_status(task);
1847	return;
1848	case -ENOMEM:
1849	rpc_delay(task, HZ >> 4);
1850	return;
1851	}
1852	trace_rpc_refresh_status(task);
1853	rpc_call_rpcerror(task, status);
1854	}
1855
1856	/*
1857	* 2b. Allocate the buffer. For details, see sched.c:rpc_malloc.
1858	* (Note: buffer memory is freed in xprt_release).
1859	*/
1860	static void
1861	call_allocate(struct rpc_task *task)
1862	{
1863	const struct rpc_auth *auth = task->tk_rqstp->rq_cred->cr_auth;
1864	struct rpc_rqst *req = task->tk_rqstp;
1865	struct rpc_xprt *xprt = req->rq_xprt;
1866	const struct rpc_procinfo *proc = task->tk_msg.rpc_proc;
1867	int status;
1868
1869	task->tk_status = 0;
1870	task->tk_action = call_encode;
1871
1872	if (req->rq_buffer)
1873	return;
1874
1875	if (proc->p_proc != 0) {
1876	BUG_ON(proc->p_arglen == 0);
1877	if (proc->p_decode != NULL)
1878	BUG_ON(proc->p_replen == 0);
1879	}
1880
1881	/*
1882	* Calculate the size (in quads) of the RPC call
1883	* and reply headers, and convert both values
1884	* to byte sizes.
1885	*/
1886	req->rq_callsize = RPC_CALLHDRSIZE + (auth->au_cslack << 1) +
1887	proc->p_arglen;
1888	req->rq_callsize <<= 2;
1889	/*
1890	* Note: the reply buffer must at minimum allocate enough space
1891	* for the 'struct accepted_reply' from RFC5531.
1892	*/
1893	req->rq_rcvsize = RPC_REPHDRSIZE + auth->au_rslack + \
1894	max_t(size_t, proc->p_replen, 2);
1895	req->rq_rcvsize <<= 2;
1896
1897	status = xprt->ops->buf_alloc(task);
1898	trace_rpc_buf_alloc(task, status);
1899	if (status == 0)
1900	return;
1901	if (status != -ENOMEM) {
1902	rpc_call_rpcerror(task, status);
1903	return;
1904	}
1905
1906	if (RPC_IS_ASYNC(task) || !fatal_signal_pending(current)) {
1907	task->tk_action = call_allocate;
1908	rpc_delay(task, HZ>>4);
1909	return;
1910	}
1911
1912	rpc_call_rpcerror(task, status: -ERESTARTSYS);
1913	}
1914
1915	static int
1916	rpc_task_need_encode(struct rpc_task *task)
1917	{
1918	return test_bit(RPC_TASK_NEED_XMIT, &task->tk_runstate) == 0 &&
1919	(!(task->tk_flags & RPC_TASK_SENT) ||
1920	!(task->tk_flags & RPC_TASK_NO_RETRANS_TIMEOUT) ||
1921	xprt_request_need_retransmit(task));
1922	}
1923
1924	static void
1925	rpc_xdr_encode(struct rpc_task *task)
1926	{
1927	struct rpc_rqst *req = task->tk_rqstp;
1928	struct xdr_stream xdr;
1929
1930	xdr_buf_init(buf: &req->rq_snd_buf,
1931	start: req->rq_buffer,
1932	len: req->rq_callsize);
1933	xdr_buf_init(buf: &req->rq_rcv_buf,
1934	start: req->rq_rbuffer,
1935	len: req->rq_rcvsize);
1936
1937	req->rq_reply_bytes_recvd = 0;
1938	req->rq_snd_buf.head[0].iov_len = 0;
1939	xdr_init_encode(xdr: &xdr, buf: &req->rq_snd_buf,
1940	p: req->rq_snd_buf.head[0].iov_base, rqst: req);
1941	if (rpc_encode_header(task, xdr: &xdr))
1942	return;
1943
1944	task->tk_status = rpcauth_wrap_req(task, xdr: &xdr);
1945	}
1946
1947	/*
1948	* 3. Encode arguments of an RPC call
1949	*/
1950	static void
1951	call_encode(struct rpc_task *task)
1952	{
1953	if (!rpc_task_need_encode(task))
1954	goto out;
1955
1956	/* Dequeue task from the receive queue while we're encoding */
1957	xprt_request_dequeue_xprt(task);
1958	/* Encode here so that rpcsec_gss can use correct sequence number. */
1959	rpc_xdr_encode(task);
1960	/* Add task to reply queue before transmission to avoid races */
1961	if (task->tk_status == 0 && rpc_reply_expected(task))
1962	task->tk_status = xprt_request_enqueue_receive(task);
1963	/* Did the encode result in an error condition? */
1964	if (task->tk_status != 0) {
1965	/* Was the error nonfatal? */
1966	switch (task->tk_status) {
1967	case -EAGAIN:
1968	case -ENOMEM:
1969	rpc_delay(task, HZ >> 4);
1970	break;
1971	case -EKEYEXPIRED:
1972	if (!task->tk_cred_retry) {
1973	rpc_call_rpcerror(task, status: task->tk_status);
1974	} else {
1975	task->tk_action = call_refresh;
1976	task->tk_cred_retry--;
1977	trace_rpc_retry_refresh_status(task);
1978	}
1979	break;
1980	default:
1981	rpc_call_rpcerror(task, status: task->tk_status);
1982	}
1983	return;
1984	}
1985
1986	xprt_request_enqueue_transmit(task);
1987	out:
1988	task->tk_action = call_transmit;
1989	/* Check that the connection is OK */
1990	if (!xprt_bound(xprt: task->tk_xprt))
1991	task->tk_action = call_bind;
1992	else if (!xprt_connected(xprt: task->tk_xprt))
1993	task->tk_action = call_connect;
1994	}
1995
1996	/*
1997	* Helpers to check if the task was already transmitted, and
1998	* to take action when that is the case.
1999	*/
2000	static bool
2001	rpc_task_transmitted(struct rpc_task *task)
2002	{
2003	return !test_bit(RPC_TASK_NEED_XMIT, &task->tk_runstate);
2004	}
2005
2006	static void
2007	rpc_task_handle_transmitted(struct rpc_task *task)
2008	{
2009	xprt_end_transmit(task);
2010	task->tk_action = call_transmit_status;
2011	}
2012
2013	/*
2014	* 4. Get the server port number if not yet set
2015	*/
2016	static void
2017	call_bind(struct rpc_task *task)
2018	{
2019	struct rpc_xprt *xprt = task->tk_rqstp->rq_xprt;
2020
2021	if (rpc_task_transmitted(task)) {
2022	rpc_task_handle_transmitted(task);
2023	return;
2024	}
2025
2026	if (xprt_bound(xprt)) {
2027	task->tk_action = call_connect;
2028	return;
2029	}
2030
2031	task->tk_action = call_bind_status;
2032	if (!xprt_prepare_transmit(task))
2033	return;
2034
2035	xprt->ops->rpcbind(task);
2036	}
2037
2038	/*
2039	* 4a. Sort out bind result
2040	*/
2041	static void
2042	call_bind_status(struct rpc_task *task)
2043	{
2044	struct rpc_xprt *xprt = task->tk_rqstp->rq_xprt;
2045	int status = -EIO;
2046
2047	if (rpc_task_transmitted(task)) {
2048	rpc_task_handle_transmitted(task);
2049	return;
2050	}
2051
2052	if (task->tk_status >= 0)
2053	goto out_next;
2054	if (xprt_bound(xprt)) {
2055	task->tk_status = 0;
2056	goto out_next;
2057	}
2058
2059	switch (task->tk_status) {
2060	case -ENOMEM:
2061	rpc_delay(task, HZ >> 2);
2062	goto retry_timeout;
2063	case -EACCES:
2064	trace_rpcb_prog_unavail_err(task);
2065	/* fail immediately if this is an RPC ping */
2066	if (task->tk_msg.rpc_proc->p_proc == 0) {
2067	status = -EOPNOTSUPP;
2068	break;
2069	}
2070	rpc_delay(task, 3*HZ);
2071	goto retry_timeout;
2072	case -ENOBUFS:
2073	rpc_delay(task, HZ >> 2);
2074	goto retry_timeout;
2075	case -EAGAIN:
2076	goto retry_timeout;
2077	case -ETIMEDOUT:
2078	trace_rpcb_timeout_err(task);
2079	goto retry_timeout;
2080	case -EPFNOSUPPORT:
2081	/* server doesn't support any rpcbind version we know of */
2082	trace_rpcb_bind_version_err(task);
2083	break;
2084	case -EPROTONOSUPPORT:
2085	trace_rpcb_bind_version_err(task);
2086	goto retry_timeout;
2087	case -ECONNREFUSED: /* connection problems */
2088	case -ECONNRESET:
2089	case -ECONNABORTED:
2090	case -ENOTCONN:
2091	case -EHOSTDOWN:
2092	case -ENETDOWN:
2093	case -EHOSTUNREACH:
2094	case -ENETUNREACH:
2095	case -EPIPE:
2096	trace_rpcb_unreachable_err(task);
2097	if (!RPC_IS_SOFTCONN(task)) {
2098	rpc_delay(task, 5*HZ);
2099	goto retry_timeout;
2100	}
2101	status = task->tk_status;
2102	break;
2103	default:
2104	trace_rpcb_unrecognized_err(task);
2105	}
2106
2107	rpc_call_rpcerror(task, status);
2108	return;
2109	out_next:
2110	task->tk_action = call_connect;
2111	return;
2112	retry_timeout:
2113	task->tk_status = 0;
2114	task->tk_action = call_bind;
2115	rpc_check_timeout(task);
2116	}
2117
2118	/*
2119	* 4b. Connect to the RPC server
2120	*/
2121	static void
2122	call_connect(struct rpc_task *task)
2123	{
2124	struct rpc_xprt *xprt = task->tk_rqstp->rq_xprt;
2125
2126	if (rpc_task_transmitted(task)) {
2127	rpc_task_handle_transmitted(task);
2128	return;
2129	}
2130
2131	if (xprt_connected(xprt)) {
2132	task->tk_action = call_transmit;
2133	return;
2134	}
2135
2136	task->tk_action = call_connect_status;
2137	if (task->tk_status < 0)
2138	return;
2139	if (task->tk_flags & RPC_TASK_NOCONNECT) {
2140	rpc_call_rpcerror(task, status: -ENOTCONN);
2141	return;
2142	}
2143	if (!xprt_prepare_transmit(task))
2144	return;
2145	xprt_connect(task);
2146	}
2147
2148	/*
2149	* 4c. Sort out connect result
2150	*/
2151	static void
2152	call_connect_status(struct rpc_task *task)
2153	{
2154	struct rpc_xprt *xprt = task->tk_rqstp->rq_xprt;
2155	struct rpc_clnt *clnt = task->tk_client;
2156	int status = task->tk_status;
2157
2158	if (rpc_task_transmitted(task)) {
2159	rpc_task_handle_transmitted(task);
2160	return;
2161	}
2162
2163	trace_rpc_connect_status(task);
2164
2165	if (task->tk_status == 0) {
2166	clnt->cl_stats->netreconn++;
2167	goto out_next;
2168	}
2169	if (xprt_connected(xprt)) {
2170	task->tk_status = 0;
2171	goto out_next;
2172	}
2173
2174	task->tk_status = 0;
2175	switch (status) {
2176	case -ECONNREFUSED:
2177	case -ECONNRESET:
2178	/* A positive refusal suggests a rebind is needed. */
2179	if (RPC_IS_SOFTCONN(task))
2180	break;
2181	if (clnt->cl_autobind) {
2182	rpc_force_rebind(clnt);
2183	goto out_retry;
2184	}
2185	fallthrough;
2186	case -ECONNABORTED:
2187	case -ENETDOWN:
2188	case -ENETUNREACH:
2189	case -EHOSTUNREACH:
2190	case -EPIPE:
2191	case -EPROTO:
2192	xprt_conditional_disconnect(xprt: task->tk_rqstp->rq_xprt,
2193	cookie: task->tk_rqstp->rq_connect_cookie);
2194	if (RPC_IS_SOFTCONN(task))
2195	break;
2196	/* retry with existing socket, after a delay */
2197	rpc_delay(task, 3*HZ);
2198	fallthrough;
2199	case -EADDRINUSE:
2200	case -ENOTCONN:
2201	case -EAGAIN:
2202	case -ETIMEDOUT:
2203	if (!(task->tk_flags & RPC_TASK_NO_ROUND_ROBIN) &&
2204	(task->tk_flags & RPC_TASK_MOVEABLE) &&
2205	test_bit(XPRT_REMOVE, &xprt->state)) {
2206	struct rpc_xprt *saved = task->tk_xprt;
2207	struct rpc_xprt_switch *xps;
2208
2209	rcu_read_lock();
2210	xps = xprt_switch_get(rcu_dereference(clnt->cl_xpi.xpi_xpswitch));
2211	rcu_read_unlock();
2212	if (xps->xps_nxprts > 1) {
2213	long value;
2214
2215	xprt_release(task);
2216	value = atomic_long_dec_return(v: &xprt->queuelen);
2217	if (value == 0)
2218	rpc_xprt_switch_remove_xprt(xps, xprt: saved,
2219	offline: true);
2220	xprt_put(xprt: saved);
2221	task->tk_xprt = NULL;
2222	task->tk_action = call_start;
2223	}
2224	xprt_switch_put(xps);
2225	if (!task->tk_xprt)
2226	goto out;
2227	}
2228	goto out_retry;
2229	case -ENOBUFS:
2230	rpc_delay(task, HZ >> 2);
2231	goto out_retry;
2232	}
2233	rpc_call_rpcerror(task, status);
2234	return;
2235	out_next:
2236	task->tk_action = call_transmit;
2237	return;
2238	out_retry:
2239	/* Check for timeouts before looping back to call_bind */
2240	task->tk_action = call_bind;
2241	out:
2242	rpc_check_timeout(task);
2243	}
2244
2245	/*
2246	* 5. Transmit the RPC request, and wait for reply
2247	*/
2248	static void
2249	call_transmit(struct rpc_task *task)
2250	{
2251	if (rpc_task_transmitted(task)) {
2252	rpc_task_handle_transmitted(task);
2253	return;
2254	}
2255
2256	task->tk_action = call_transmit_status;
2257	if (!xprt_prepare_transmit(task))
2258	return;
2259	task->tk_status = 0;
2260	if (test_bit(RPC_TASK_NEED_XMIT, &task->tk_runstate)) {
2261	if (!xprt_connected(xprt: task->tk_xprt)) {
2262	task->tk_status = -ENOTCONN;
2263	return;
2264	}
2265	xprt_transmit(task);
2266	}
2267	xprt_end_transmit(task);
2268	}
2269
2270	/*
2271	* 5a. Handle cleanup after a transmission
2272	*/
2273	static void
2274	call_transmit_status(struct rpc_task *task)
2275	{
2276	task->tk_action = call_status;
2277
2278	/*
2279	* Common case: success. Force the compiler to put this
2280	* test first.
2281	*/
2282	if (rpc_task_transmitted(task)) {
2283	task->tk_status = 0;
2284	xprt_request_wait_receive(task);
2285	return;
2286	}
2287
2288	switch (task->tk_status) {
2289	default:
2290	break;
2291	case -EBADMSG:
2292	task->tk_status = 0;
2293	task->tk_action = call_encode;
2294	break;
2295	/*
2296	* Special cases: if we've been waiting on the
2297	* socket's write_space() callback, or if the
2298	* socket just returned a connection error,
2299	* then hold onto the transport lock.
2300	*/
2301	case -ENOMEM:
2302	case -ENOBUFS:
2303	rpc_delay(task, HZ>>2);
2304	fallthrough;
2305	case -EBADSLT:
2306	case -EAGAIN:
2307	task->tk_action = call_transmit;
2308	task->tk_status = 0;
2309	break;
2310	case -ECONNREFUSED:
2311	case -EHOSTDOWN:
2312	case -ENETDOWN:
2313	case -EHOSTUNREACH:
2314	case -ENETUNREACH:
2315	case -EPERM:
2316	if (RPC_IS_SOFTCONN(task)) {
2317	if (!task->tk_msg.rpc_proc->p_proc)
2318	trace_xprt_ping(xprt: task->tk_xprt,
2319	status: task->tk_status);
2320	rpc_call_rpcerror(task, status: task->tk_status);
2321	return;
2322	}
2323	fallthrough;
2324	case -ECONNRESET:
2325	case -ECONNABORTED:
2326	case -EADDRINUSE:
2327	case -ENOTCONN:
2328	case -EPIPE:
2329	task->tk_action = call_bind;
2330	task->tk_status = 0;
2331	break;
2332	}
2333	rpc_check_timeout(task);
2334	}
2335
2336	#if defined(CONFIG_SUNRPC_BACKCHANNEL)
2337	static void call_bc_transmit(struct rpc_task *task);
2338	static void call_bc_transmit_status(struct rpc_task *task);
2339
2340	static void
2341	call_bc_encode(struct rpc_task *task)
2342	{
2343	xprt_request_enqueue_transmit(task);
2344	task->tk_action = call_bc_transmit;
2345	}
2346
2347	/*
2348	* 5b. Send the backchannel RPC reply. On error, drop the reply. In
2349	* addition, disconnect on connectivity errors.
2350	*/
2351	static void
2352	call_bc_transmit(struct rpc_task *task)
2353	{
2354	task->tk_action = call_bc_transmit_status;
2355	if (test_bit(RPC_TASK_NEED_XMIT, &task->tk_runstate)) {
2356	if (!xprt_prepare_transmit(task))
2357	return;
2358	task->tk_status = 0;
2359	xprt_transmit(task);
2360	}
2361	xprt_end_transmit(task);
2362	}
2363
2364	static void
2365	call_bc_transmit_status(struct rpc_task *task)
2366	{
2367	struct rpc_rqst *req = task->tk_rqstp;
2368
2369	if (rpc_task_transmitted(task))
2370	task->tk_status = 0;
2371
2372	switch (task->tk_status) {
2373	case 0:
2374	/* Success */
2375	case -ENETDOWN:
2376	case -EHOSTDOWN:
2377	case -EHOSTUNREACH:
2378	case -ENETUNREACH:
2379	case -ECONNRESET:
2380	case -ECONNREFUSED:
2381	case -EADDRINUSE:
2382	case -ENOTCONN:
2383	case -EPIPE:
2384	break;
2385	case -ENOMEM:
2386	case -ENOBUFS:
2387	rpc_delay(task, HZ>>2);
2388	fallthrough;
2389	case -EBADSLT:
2390	case -EAGAIN:
2391	task->tk_status = 0;
2392	task->tk_action = call_bc_transmit;
2393	return;
2394	case -ETIMEDOUT:
2395	/*
2396	* Problem reaching the server. Disconnect and let the
2397	* forechannel reestablish the connection. The server will
2398	* have to retransmit the backchannel request and we'll
2399	* reprocess it. Since these ops are idempotent, there's no
2400	* need to cache our reply at this time.
2401	*/
2402	printk(KERN_NOTICE "RPC: Could not send backchannel reply "
2403	"error: %d\n", task->tk_status);
2404	xprt_conditional_disconnect(xprt: req->rq_xprt,
2405	cookie: req->rq_connect_cookie);
2406	break;
2407	default:
2408	/*
2409	* We were unable to reply and will have to drop the
2410	* request. The server should reconnect and retransmit.
2411	*/
2412	printk(KERN_NOTICE "RPC: Could not send backchannel reply "
2413	"error: %d\n", task->tk_status);
2414	break;
2415	}
2416	task->tk_action = rpc_exit_task;
2417	}
2418	#endif /* CONFIG_SUNRPC_BACKCHANNEL */
2419
2420	/*
2421	* 6. Sort out the RPC call status
2422	*/
2423	static void
2424	call_status(struct rpc_task *task)
2425	{
2426	struct rpc_clnt *clnt = task->tk_client;
2427	int status;
2428
2429	if (!task->tk_msg.rpc_proc->p_proc)
2430	trace_xprt_ping(xprt: task->tk_xprt, status: task->tk_status);
2431
2432	status = task->tk_status;
2433	if (status >= 0) {
2434	task->tk_action = call_decode;
2435	return;
2436	}
2437
2438	trace_rpc_call_status(task);
2439	task->tk_status = 0;
2440	switch(status) {
2441	case -EHOSTDOWN:
2442	case -ENETDOWN:
2443	case -EHOSTUNREACH:
2444	case -ENETUNREACH:
2445	case -EPERM:
2446	if (RPC_IS_SOFTCONN(task))
2447	goto out_exit;
2448	/*
2449	* Delay any retries for 3 seconds, then handle as if it
2450	* were a timeout.
2451	*/
2452	rpc_delay(task, 3*HZ);
2453	fallthrough;
2454	case -ETIMEDOUT:
2455	break;
2456	case -ECONNREFUSED:
2457	case -ECONNRESET:
2458	case -ECONNABORTED:
2459	case -ENOTCONN:
2460	rpc_force_rebind(clnt);
2461	break;
2462	case -EADDRINUSE:
2463	rpc_delay(task, 3*HZ);
2464	fallthrough;
2465	case -EPIPE:
2466	case -EAGAIN:
2467	break;
2468	case -ENFILE:
2469	case -ENOBUFS:
2470	case -ENOMEM:
2471	rpc_delay(task, HZ>>2);
2472	break;
2473	case -EIO:
2474	/* shutdown or soft timeout */
2475	goto out_exit;
2476	default:
2477	if (clnt->cl_chatty)
2478	printk("%s: RPC call returned error %d\n",
2479	clnt->cl_program->name, -status);
2480	goto out_exit;
2481	}
2482	task->tk_action = call_encode;
2483	rpc_check_timeout(task);
2484	return;
2485	out_exit:
2486	rpc_call_rpcerror(task, status);
2487	}
2488
2489	static bool
2490	rpc_check_connected(const struct rpc_rqst *req)
2491	{
2492	/* No allocated request or transport? return true */
2493	if (!req || !req->rq_xprt)
2494	return true;
2495	return xprt_connected(xprt: req->rq_xprt);
2496	}
2497
2498	static void
2499	rpc_check_timeout(struct rpc_task *task)
2500	{
2501	struct rpc_clnt *clnt = task->tk_client;
2502
2503	if (RPC_SIGNALLED(task))
2504	return;
2505
2506	if (xprt_adjust_timeout(req: task->tk_rqstp) == 0)
2507	return;
2508
2509	trace_rpc_timeout_status(task);
2510	task->tk_timeouts++;
2511
2512	if (RPC_IS_SOFTCONN(task) && !rpc_check_connected(req: task->tk_rqstp)) {
2513	rpc_call_rpcerror(task, status: -ETIMEDOUT);
2514	return;
2515	}
2516
2517	if (RPC_IS_SOFT(task)) {
2518	/*
2519	* Once a "no retrans timeout" soft tasks (a.k.a NFSv4) has
2520	* been sent, it should time out only if the transport
2521	* connection gets terminally broken.
2522	*/
2523	if ((task->tk_flags & RPC_TASK_NO_RETRANS_TIMEOUT) &&
2524	rpc_check_connected(req: task->tk_rqstp))
2525	return;
2526
2527	if (clnt->cl_chatty) {
2528	pr_notice_ratelimited(
2529	"%s: server %s not responding, timed out\n",
2530	clnt->cl_program->name,
2531	task->tk_xprt->servername);
2532	}
2533	if (task->tk_flags & RPC_TASK_TIMEOUT)
2534	rpc_call_rpcerror(task, status: -ETIMEDOUT);
2535	else
2536	__rpc_call_rpcerror(task, tk_status: -EIO, rpc_status: -ETIMEDOUT);
2537	return;
2538	}
2539
2540	if (!(task->tk_flags & RPC_CALL_MAJORSEEN)) {
2541	task->tk_flags |= RPC_CALL_MAJORSEEN;
2542	if (clnt->cl_chatty) {
2543	pr_notice_ratelimited(
2544	"%s: server %s not responding, still trying\n",
2545	clnt->cl_program->name,
2546	task->tk_xprt->servername);
2547	}
2548	}
2549	rpc_force_rebind(clnt);
2550	/*
2551	* Did our request time out due to an RPCSEC_GSS out-of-sequence
2552	* event? RFC2203 requires the server to drop all such requests.
2553	*/
2554	rpcauth_invalcred(task);
2555	}
2556
2557	/*
2558	* 7. Decode the RPC reply
2559	*/
2560	static void
2561	call_decode(struct rpc_task *task)
2562	{
2563	struct rpc_clnt *clnt = task->tk_client;
2564	struct rpc_rqst *req = task->tk_rqstp;
2565	struct xdr_stream xdr;
2566	int err;
2567
2568	if (!task->tk_msg.rpc_proc->p_decode) {
2569	task->tk_action = rpc_exit_task;
2570	return;
2571	}
2572
2573	if (task->tk_flags & RPC_CALL_MAJORSEEN) {
2574	if (clnt->cl_chatty) {
2575	pr_notice_ratelimited("%s: server %s OK\n",
2576	clnt->cl_program->name,
2577	task->tk_xprt->servername);
2578	}
2579	task->tk_flags &= ~RPC_CALL_MAJORSEEN;
2580	}
2581
2582	/*
2583	* Did we ever call xprt_complete_rqst()? If not, we should assume
2584	* the message is incomplete.
2585	*/
2586	err = -EAGAIN;
2587	if (!req->rq_reply_bytes_recvd)
2588	goto out;
2589
2590	/* Ensure that we see all writes made by xprt_complete_rqst()
2591	* before it changed req->rq_reply_bytes_recvd.
2592	*/
2593	smp_rmb();
2594
2595	req->rq_rcv_buf.len = req->rq_private_buf.len;
2596	trace_rpc_xdr_recvfrom(task, xdr: &req->rq_rcv_buf);
2597
2598	/* Check that the softirq receive buffer is valid */
2599	WARN_ON(memcmp(&req->rq_rcv_buf, &req->rq_private_buf,
2600	sizeof(req->rq_rcv_buf)) != 0);
2601
2602	xdr_init_decode(xdr: &xdr, buf: &req->rq_rcv_buf,
2603	p: req->rq_rcv_buf.head[0].iov_base, rqst: req);
2604	err = rpc_decode_header(task, xdr: &xdr);
2605	out:
2606	switch (err) {
2607	case 0:
2608	task->tk_action = rpc_exit_task;
2609	task->tk_status = rpcauth_unwrap_resp(task, xdr: &xdr);
2610	xdr_finish_decode(xdr: &xdr);
2611	return;
2612	case -EAGAIN:
2613	task->tk_status = 0;
2614	if (task->tk_client->cl_discrtry)
2615	xprt_conditional_disconnect(xprt: req->rq_xprt,
2616	cookie: req->rq_connect_cookie);
2617	task->tk_action = call_encode;
2618	rpc_check_timeout(task);
2619	break;
2620	case -EKEYREJECTED:
2621	task->tk_action = call_reserve;
2622	rpc_check_timeout(task);
2623	rpcauth_invalcred(task);
2624	/* Ensure we obtain a new XID if we retry! */
2625	xprt_release(task);
2626	}
2627	}
2628
2629	static int
2630	rpc_encode_header(struct rpc_task *task, struct xdr_stream *xdr)
2631	{
2632	struct rpc_clnt *clnt = task->tk_client;
2633	struct rpc_rqst *req = task->tk_rqstp;
2634	__be32 *p;
2635	int error;
2636
2637	error = -EMSGSIZE;
2638	p = xdr_reserve_space(xdr, RPC_CALLHDRSIZE << 2);
2639	if (!p)
2640	goto out_fail;
2641	*p++ = req->rq_xid;
2642	*p++ = rpc_call;
2643	*p++ = cpu_to_be32(RPC_VERSION);
2644	*p++ = cpu_to_be32(clnt->cl_prog);
2645	*p++ = cpu_to_be32(clnt->cl_vers);
2646	*p = cpu_to_be32(task->tk_msg.rpc_proc->p_proc);
2647
2648	error = rpcauth_marshcred(task, xdr);
2649	if (error < 0)
2650	goto out_fail;
2651	return 0;
2652	out_fail:
2653	trace_rpc_bad_callhdr(task);
2654	rpc_call_rpcerror(task, status: error);
2655	return error;
2656	}
2657
2658	static noinline int
2659	rpc_decode_header(struct rpc_task *task, struct xdr_stream *xdr)
2660	{
2661	struct rpc_clnt *clnt = task->tk_client;
2662	int error;
2663	__be32 *p;
2664
2665	/* RFC-1014 says that the representation of XDR data must be a
2666	* multiple of four bytes
2667	* - if it isn't pointer subtraction in the NFS client may give
2668	* undefined results
2669	*/
2670	if (task->tk_rqstp->rq_rcv_buf.len & 3)
2671	goto out_unparsable;
2672
2673	p = xdr_inline_decode(xdr, nbytes: 3 * sizeof(*p));
2674	if (!p)
2675	goto out_unparsable;
2676	p++; /* skip XID */
2677	if (*p++ != rpc_reply)
2678	goto out_unparsable;
2679	if (*p++ != rpc_msg_accepted)
2680	goto out_msg_denied;
2681
2682	error = rpcauth_checkverf(task, xdr);
2683	if (error)
2684	goto out_verifier;
2685
2686	p = xdr_inline_decode(xdr, nbytes: sizeof(*p));
2687	if (!p)
2688	goto out_unparsable;
2689	switch (*p) {
2690	case rpc_success:
2691	return 0;
2692	case rpc_prog_unavail:
2693	trace_rpc__prog_unavail(task);
2694	error = -EPFNOSUPPORT;
2695	goto out_err;
2696	case rpc_prog_mismatch:
2697	trace_rpc__prog_mismatch(task);
2698	error = -EPROTONOSUPPORT;
2699	goto out_err;
2700	case rpc_proc_unavail:
2701	trace_rpc__proc_unavail(task);
2702	error = -EOPNOTSUPP;
2703	goto out_err;
2704	case rpc_garbage_args:
2705	case rpc_system_err:
2706	trace_rpc__garbage_args(task);
2707	error = -EIO;
2708	break;
2709	default:
2710	goto out_unparsable;
2711	}
2712
2713	out_garbage:
2714	clnt->cl_stats->rpcgarbage++;
2715	if (task->tk_garb_retry) {
2716	task->tk_garb_retry--;
2717	task->tk_action = call_encode;
2718	return -EAGAIN;
2719	}
2720	out_err:
2721	rpc_call_rpcerror(task, status: error);
2722	return error;
2723
2724	out_unparsable:
2725	trace_rpc__unparsable(task);
2726	error = -EIO;
2727	goto out_garbage;
2728
2729	out_verifier:
2730	trace_rpc_bad_verifier(task);
2731	switch (error) {
2732	case -EPROTONOSUPPORT:
2733	goto out_err;
2734	case -EACCES:
2735	/* Re-encode with a fresh cred */
2736	fallthrough;
2737	default:
2738	goto out_garbage;
2739	}
2740
2741	out_msg_denied:
2742	error = -EACCES;
2743	p = xdr_inline_decode(xdr, nbytes: sizeof(*p));
2744	if (!p)
2745	goto out_unparsable;
2746	switch (*p++) {
2747	case rpc_auth_error:
2748	break;
2749	case rpc_mismatch:
2750	trace_rpc__mismatch(task);
2751	error = -EPROTONOSUPPORT;
2752	goto out_err;
2753	default:
2754	goto out_unparsable;
2755	}
2756
2757	p = xdr_inline_decode(xdr, nbytes: sizeof(*p));
2758	if (!p)
2759	goto out_unparsable;
2760	switch (*p++) {
2761	case rpc_autherr_rejectedcred:
2762	case rpc_autherr_rejectedverf:
2763	case rpcsec_gsserr_credproblem:
2764	case rpcsec_gsserr_ctxproblem:
2765	rpcauth_invalcred(task);
2766	if (!task->tk_cred_retry)
2767	break;
2768	task->tk_cred_retry--;
2769	trace_rpc__stale_creds(task);
2770	return -EKEYREJECTED;
2771	case rpc_autherr_badcred:
2772	case rpc_autherr_badverf:
2773	/* possibly garbled cred/verf? */
2774	if (!task->tk_garb_retry)
2775	break;
2776	task->tk_garb_retry--;
2777	trace_rpc__bad_creds(task);
2778	task->tk_action = call_encode;
2779	return -EAGAIN;
2780	case rpc_autherr_tooweak:
2781	trace_rpc__auth_tooweak(task);
2782	pr_warn("RPC: server %s requires stronger authentication.\n",
2783	task->tk_xprt->servername);
2784	break;
2785	default:
2786	goto out_unparsable;
2787	}
2788	goto out_err;
2789	}
2790
2791	static void rpcproc_encode_null(struct rpc_rqst *rqstp, struct xdr_stream *xdr,
2792	const void *obj)
2793	{
2794	}
2795
2796	static int rpcproc_decode_null(struct rpc_rqst *rqstp, struct xdr_stream *xdr,
2797	void *obj)
2798	{
2799	return 0;
2800	}
2801
2802	static const struct rpc_procinfo rpcproc_null = {
2803	.p_encode = rpcproc_encode_null,
2804	.p_decode = rpcproc_decode_null,
2805	};
2806
2807	static const struct rpc_procinfo rpcproc_null_noreply = {
2808	.p_encode = rpcproc_encode_null,
2809	};
2810
2811	static void
2812	rpc_null_call_prepare(struct rpc_task *task, void *data)
2813	{
2814	task->tk_flags &= ~RPC_TASK_NO_RETRANS_TIMEOUT;
2815	rpc_call_start(task);
2816	}
2817
2818	static const struct rpc_call_ops rpc_null_ops = {
2819	.rpc_call_prepare = rpc_null_call_prepare,
2820	.rpc_call_done = rpc_default_callback,
2821	};
2822
2823	static
2824	struct rpc_task *rpc_call_null_helper(struct rpc_clnt *clnt,
2825	struct rpc_xprt *xprt, struct rpc_cred *cred, int flags,
2826	const struct rpc_call_ops *ops, void *data)
2827	{
2828	struct rpc_message msg = {
2829	.rpc_proc = &rpcproc_null,
2830	};
2831	struct rpc_task_setup task_setup_data = {
2832	.rpc_client = clnt,
2833	.rpc_xprt = xprt,
2834	.rpc_message = &msg,
2835	.rpc_op_cred = cred,
2836	.callback_ops = ops ?: &rpc_null_ops,
2837	.callback_data = data,
2838	.flags = flags | RPC_TASK_SOFT | RPC_TASK_SOFTCONN |
2839	RPC_TASK_NULLCREDS,
2840	};
2841
2842	return rpc_run_task(&task_setup_data);
2843	}
2844
2845	struct rpc_task *rpc_call_null(struct rpc_clnt *clnt, struct rpc_cred *cred, int flags)
2846	{
2847	return rpc_call_null_helper(clnt, NULL, cred, flags, NULL, NULL);
2848	}
2849	EXPORT_SYMBOL_GPL(rpc_call_null);
2850
2851	static int rpc_ping(struct rpc_clnt *clnt)
2852	{
2853	struct rpc_task *task;
2854	int status;
2855
2856	if (clnt->cl_auth->au_ops->ping)
2857	return clnt->cl_auth->au_ops->ping(clnt);
2858
2859	task = rpc_call_null_helper(clnt, NULL, NULL, flags: 0, NULL, NULL);
2860	if (IS_ERR(ptr: task))
2861	return PTR_ERR(ptr: task);
2862	status = task->tk_status;
2863	rpc_put_task(task);
2864	return status;
2865	}
2866
2867	static int rpc_ping_noreply(struct rpc_clnt *clnt)
2868	{
2869	struct rpc_message msg = {
2870	.rpc_proc = &rpcproc_null_noreply,
2871	};
2872	struct rpc_task_setup task_setup_data = {
2873	.rpc_client = clnt,
2874	.rpc_message = &msg,
2875	.callback_ops = &rpc_null_ops,
2876	.flags = RPC_TASK_SOFT | RPC_TASK_SOFTCONN | RPC_TASK_NULLCREDS,
2877	};
2878	struct rpc_task *task;
2879	int status;
2880
2881	task = rpc_run_task(&task_setup_data);
2882	if (IS_ERR(ptr: task))
2883	return PTR_ERR(ptr: task);
2884	status = task->tk_status;
2885	rpc_put_task(task);
2886	return status;
2887	}
2888
2889	struct rpc_cb_add_xprt_calldata {
2890	struct rpc_xprt_switch *xps;
2891	struct rpc_xprt *xprt;
2892	};
2893
2894	static void rpc_cb_add_xprt_done(struct rpc_task *task, void *calldata)
2895	{
2896	struct rpc_cb_add_xprt_calldata *data = calldata;
2897
2898	if (task->tk_status == 0)
2899	rpc_xprt_switch_add_xprt(xps: data->xps, xprt: data->xprt);
2900	}
2901
2902	static void rpc_cb_add_xprt_release(void *calldata)
2903	{
2904	struct rpc_cb_add_xprt_calldata *data = calldata;
2905
2906	xprt_put(xprt: data->xprt);
2907	xprt_switch_put(xps: data->xps);
2908	kfree(objp: data);
2909	}
2910
2911	static const struct rpc_call_ops rpc_cb_add_xprt_call_ops = {
2912	.rpc_call_prepare = rpc_null_call_prepare,
2913	.rpc_call_done = rpc_cb_add_xprt_done,
2914	.rpc_release = rpc_cb_add_xprt_release,
2915	};
2916
2917	/**
2918	* rpc_clnt_test_and_add_xprt - Test and add a new transport to a rpc_clnt
2919	* @clnt: pointer to struct rpc_clnt
2920	* @xps: pointer to struct rpc_xprt_switch,
2921	* @xprt: pointer struct rpc_xprt
2922	* @in_max_connect: pointer to the max_connect value for the passed in xprt transport
2923	*/
2924	int rpc_clnt_test_and_add_xprt(struct rpc_clnt *clnt,
2925	struct rpc_xprt_switch *xps, struct rpc_xprt *xprt,
2926	void *in_max_connect)
2927	{
2928	struct rpc_cb_add_xprt_calldata *data;
2929	struct rpc_task *task;
2930	int max_connect = clnt->cl_max_connect;
2931
2932	if (in_max_connect)
2933	max_connect = *(int *)in_max_connect;
2934	if (xps->xps_nunique_destaddr_xprts + 1 > max_connect) {
2935	rcu_read_lock();
2936	pr_warn("SUNRPC: reached max allowed number (%d) did not add "
2937	"transport to server: %s\n", max_connect,
2938	rpc_peeraddr2str(clnt, RPC_DISPLAY_ADDR));
2939	rcu_read_unlock();
2940	return -EINVAL;
2941	}
2942
2943	data = kmalloc(size: sizeof(*data), GFP_KERNEL);
2944	if (!data)
2945	return -ENOMEM;
2946	data->xps = xprt_switch_get(xps);
2947	data->xprt = xprt_get(xprt);
2948	if (rpc_xprt_switch_has_addr(xps: data->xps, sap: (struct sockaddr *)&xprt->addr)) {
2949	rpc_cb_add_xprt_release(calldata: data);
2950	goto success;
2951	}
2952
2953	task = rpc_call_null_helper(clnt, xprt, NULL, RPC_TASK_ASYNC,
2954	ops: &rpc_cb_add_xprt_call_ops, data);
2955	if (IS_ERR(ptr: task))
2956	return PTR_ERR(ptr: task);
2957
2958	data->xps->xps_nunique_destaddr_xprts++;
2959	rpc_put_task(task);
2960	success:
2961	return 1;
2962	}
2963	EXPORT_SYMBOL_GPL(rpc_clnt_test_and_add_xprt);
2964
2965	static int rpc_clnt_add_xprt_helper(struct rpc_clnt *clnt,
2966	struct rpc_xprt *xprt,
2967	struct rpc_add_xprt_test *data)
2968	{
2969	struct rpc_task *task;
2970	int status = -EADDRINUSE;
2971
2972	/* Test the connection */
2973	task = rpc_call_null_helper(clnt, xprt, NULL, flags: 0, NULL, NULL);
2974	if (IS_ERR(ptr: task))
2975	return PTR_ERR(ptr: task);
2976
2977	status = task->tk_status;
2978	rpc_put_task(task);
2979
2980	if (status < 0)
2981	return status;
2982
2983	/* rpc_xprt_switch and rpc_xprt are deferrenced by add_xprt_test() */
2984	data->add_xprt_test(clnt, xprt, data->data);
2985
2986	return 0;
2987	}
2988
2989	/**
2990	* rpc_clnt_setup_test_and_add_xprt()
2991	*
2992	* This is an rpc_clnt_add_xprt setup() function which returns 1 so:
2993	* 1) caller of the test function must dereference the rpc_xprt_switch
2994	* and the rpc_xprt.
2995	* 2) test function must call rpc_xprt_switch_add_xprt, usually in
2996	* the rpc_call_done routine.
2997	*
2998	* Upon success (return of 1), the test function adds the new
2999	* transport to the rpc_clnt xprt switch
3000	*
3001	* @clnt: struct rpc_clnt to get the new transport
3002	* @xps: the rpc_xprt_switch to hold the new transport
3003	* @xprt: the rpc_xprt to test
3004	* @data: a struct rpc_add_xprt_test pointer that holds the test function
3005	* and test function call data
3006	*/
3007	int rpc_clnt_setup_test_and_add_xprt(struct rpc_clnt *clnt,
3008	struct rpc_xprt_switch *xps,
3009	struct rpc_xprt *xprt,
3010	void *data)
3011	{
3012	int status = -EADDRINUSE;
3013
3014	xprt = xprt_get(xprt);
3015	xprt_switch_get(xps);
3016
3017	if (rpc_xprt_switch_has_addr(xps, sap: (struct sockaddr *)&xprt->addr))
3018	goto out_err;
3019
3020	status = rpc_clnt_add_xprt_helper(clnt, xprt, data);
3021	if (status < 0)
3022	goto out_err;
3023
3024	status = 1;
3025	out_err:
3026	xprt_put(xprt);
3027	xprt_switch_put(xps);
3028	if (status < 0)
3029	pr_info("RPC: rpc_clnt_test_xprt failed: %d addr %s not "
3030	"added\n", status,
3031	xprt->address_strings[RPC_DISPLAY_ADDR]);
3032	/* so that rpc_clnt_add_xprt does not call rpc_xprt_switch_add_xprt */
3033	return status;
3034	}
3035	EXPORT_SYMBOL_GPL(rpc_clnt_setup_test_and_add_xprt);
3036
3037	/**
3038	* rpc_clnt_add_xprt - Add a new transport to a rpc_clnt
3039	* @clnt: pointer to struct rpc_clnt
3040	* @xprtargs: pointer to struct xprt_create
3041	* @setup: callback to test and/or set up the connection
3042	* @data: pointer to setup function data
3043	*
3044	* Creates a new transport using the parameters set in args and
3045	* adds it to clnt.
3046	* If ping is set, then test that connectivity succeeds before
3047	* adding the new transport.
3048	*
3049	*/
3050	int rpc_clnt_add_xprt(struct rpc_clnt *clnt,
3051	struct xprt_create *xprtargs,
3052	int (*setup)(struct rpc_clnt *,
3053	struct rpc_xprt_switch *,
3054	struct rpc_xprt *,
3055	void *),
3056	void *data)
3057	{
3058	struct rpc_xprt_switch *xps;
3059	struct rpc_xprt *xprt;
3060	unsigned long connect_timeout;
3061	unsigned long reconnect_timeout;
3062	unsigned char resvport, reuseport;
3063	int ret = 0, ident;
3064
3065	rcu_read_lock();
3066	xps = xprt_switch_get(rcu_dereference(clnt->cl_xpi.xpi_xpswitch));
3067	xprt = xprt_iter_xprt(xpi: &clnt->cl_xpi);
3068	if (xps == NULL || xprt == NULL) {
3069	rcu_read_unlock();
3070	xprt_switch_put(xps);
3071	return -EAGAIN;
3072	}
3073	resvport = xprt->resvport;
3074	reuseport = xprt->reuseport;
3075	connect_timeout = xprt->connect_timeout;
3076	reconnect_timeout = xprt->max_reconnect_timeout;
3077	ident = xprt->xprt_class->ident;
3078	rcu_read_unlock();
3079
3080	if (!xprtargs->ident)
3081	xprtargs->ident = ident;
3082	xprtargs->xprtsec = clnt->cl_xprtsec;
3083	xprt = xprt_create_transport(args: xprtargs);
3084	if (IS_ERR(ptr: xprt)) {
3085	ret = PTR_ERR(ptr: xprt);
3086	goto out_put_switch;
3087	}
3088	xprt->resvport = resvport;
3089	xprt->reuseport = reuseport;
3090
3091	if (xprtargs->connect_timeout)
3092	connect_timeout = xprtargs->connect_timeout;
3093	if (xprtargs->reconnect_timeout)
3094	reconnect_timeout = xprtargs->reconnect_timeout;
3095	if (xprt->ops->set_connect_timeout != NULL)
3096	xprt->ops->set_connect_timeout(xprt,
3097	connect_timeout,
3098	reconnect_timeout);
3099
3100	rpc_xprt_switch_set_roundrobin(xps);
3101	if (setup) {
3102	ret = setup(clnt, xps, xprt, data);
3103	if (ret != 0)
3104	goto out_put_xprt;
3105	}
3106	rpc_xprt_switch_add_xprt(xps, xprt);
3107	out_put_xprt:
3108	xprt_put(xprt);
3109	out_put_switch:
3110	xprt_switch_put(xps);
3111	return ret;
3112	}
3113	EXPORT_SYMBOL_GPL(rpc_clnt_add_xprt);
3114
3115	static int rpc_xprt_probe_trunked(struct rpc_clnt *clnt,
3116	struct rpc_xprt *xprt,
3117	struct rpc_add_xprt_test *data)
3118	{
3119	struct rpc_xprt_switch *xps;
3120	struct rpc_xprt *main_xprt;
3121	int status = 0;
3122
3123	xprt_get(xprt);
3124
3125	rcu_read_lock();
3126	main_xprt = xprt_get(rcu_dereference(clnt->cl_xprt));
3127	xps = xprt_switch_get(rcu_dereference(clnt->cl_xpi.xpi_xpswitch));
3128	status = rpc_cmp_addr_port(sap1: (struct sockaddr *)&xprt->addr,
3129	sap2: (struct sockaddr *)&main_xprt->addr);
3130	rcu_read_unlock();
3131	xprt_put(xprt: main_xprt);
3132	if (status || !test_bit(XPRT_OFFLINE, &xprt->state))
3133	goto out;
3134
3135	status = rpc_clnt_add_xprt_helper(clnt, xprt, data);
3136	out:
3137	xprt_put(xprt);
3138	xprt_switch_put(xps);
3139	return status;
3140	}
3141
3142	/* rpc_clnt_probe_trunked_xprt -- probe offlined transport for session trunking
3143	* @clnt rpc_clnt structure
3144	*
3145	* For each offlined transport found in the rpc_clnt structure call
3146	* the function rpc_xprt_probe_trunked() which will determine if this
3147	* transport still belongs to the trunking group.
3148	*/
3149	void rpc_clnt_probe_trunked_xprts(struct rpc_clnt *clnt,
3150	struct rpc_add_xprt_test *data)
3151	{
3152	struct rpc_xprt_iter xpi;
3153	int ret;
3154
3155	ret = rpc_clnt_xprt_iter_offline_init(clnt, xpi: &xpi);
3156	if (ret)
3157	return;
3158	for (;;) {
3159	struct rpc_xprt *xprt = xprt_iter_get_next(xpi: &xpi);
3160
3161	if (!xprt)
3162	break;
3163	ret = rpc_xprt_probe_trunked(clnt, xprt, data);
3164	xprt_put(xprt);
3165	if (ret < 0)
3166	break;
3167	xprt_iter_rewind(xpi: &xpi);
3168	}
3169	xprt_iter_destroy(xpi: &xpi);
3170	}
3171	EXPORT_SYMBOL_GPL(rpc_clnt_probe_trunked_xprts);
3172
3173	static int rpc_xprt_offline(struct rpc_clnt *clnt,
3174	struct rpc_xprt *xprt,
3175	void *data)
3176	{
3177	struct rpc_xprt *main_xprt;
3178	struct rpc_xprt_switch *xps;
3179	int err = 0;
3180
3181	xprt_get(xprt);
3182
3183	rcu_read_lock();
3184	main_xprt = xprt_get(rcu_dereference(clnt->cl_xprt));
3185	xps = xprt_switch_get(rcu_dereference(clnt->cl_xpi.xpi_xpswitch));
3186	err = rpc_cmp_addr_port(sap1: (struct sockaddr *)&xprt->addr,
3187	sap2: (struct sockaddr *)&main_xprt->addr);
3188	rcu_read_unlock();
3189	xprt_put(xprt: main_xprt);
3190	if (err)
3191	goto out;
3192
3193	if (wait_on_bit_lock(word: &xprt->state, XPRT_LOCKED, TASK_KILLABLE)) {
3194	err = -EINTR;
3195	goto out;
3196	}
3197	xprt_set_offline_locked(xprt, xps);
3198
3199	xprt_release_write(xprt, NULL);
3200	out:
3201	xprt_put(xprt);
3202	xprt_switch_put(xps);
3203	return err;
3204	}
3205
3206	/* rpc_clnt_manage_trunked_xprts -- offline trunked transports
3207	* @clnt rpc_clnt structure
3208	*
3209	* For each active transport found in the rpc_clnt structure call
3210	* the function rpc_xprt_offline() which will identify trunked transports
3211	* and will mark them offline.
3212	*/
3213	void rpc_clnt_manage_trunked_xprts(struct rpc_clnt *clnt)
3214	{
3215	rpc_clnt_iterate_for_each_xprt(clnt, rpc_xprt_offline, NULL);
3216	}
3217	EXPORT_SYMBOL_GPL(rpc_clnt_manage_trunked_xprts);
3218
3219	struct connect_timeout_data {
3220	unsigned long connect_timeout;
3221	unsigned long reconnect_timeout;
3222	};
3223
3224	static int
3225	rpc_xprt_set_connect_timeout(struct rpc_clnt *clnt,
3226	struct rpc_xprt *xprt,
3227	void *data)
3228	{
3229	struct connect_timeout_data *timeo = data;
3230
3231	if (xprt->ops->set_connect_timeout)
3232	xprt->ops->set_connect_timeout(xprt,
3233	timeo->connect_timeout,
3234	timeo->reconnect_timeout);
3235	return 0;
3236	}
3237
3238	void
3239	rpc_set_connect_timeout(struct rpc_clnt *clnt,
3240	unsigned long connect_timeout,
3241	unsigned long reconnect_timeout)
3242	{
3243	struct connect_timeout_data timeout = {
3244	.connect_timeout = connect_timeout,
3245	.reconnect_timeout = reconnect_timeout,
3246	};
3247	rpc_clnt_iterate_for_each_xprt(clnt,
3248	rpc_xprt_set_connect_timeout,
3249	&timeout);
3250	}
3251	EXPORT_SYMBOL_GPL(rpc_set_connect_timeout);
3252
3253	void rpc_clnt_xprt_switch_put(struct rpc_clnt *clnt)
3254	{
3255	rcu_read_lock();
3256	xprt_switch_put(rcu_dereference(clnt->cl_xpi.xpi_xpswitch));
3257	rcu_read_unlock();
3258	}
3259	EXPORT_SYMBOL_GPL(rpc_clnt_xprt_switch_put);
3260
3261	void rpc_clnt_xprt_set_online(struct rpc_clnt *clnt, struct rpc_xprt *xprt)
3262	{
3263	struct rpc_xprt_switch *xps;
3264
3265	rcu_read_lock();
3266	xps = rcu_dereference(clnt->cl_xpi.xpi_xpswitch);
3267	rcu_read_unlock();
3268	xprt_set_online_locked(xprt, xps);
3269	}
3270
3271	void rpc_clnt_xprt_switch_add_xprt(struct rpc_clnt *clnt, struct rpc_xprt *xprt)
3272	{
3273	if (rpc_clnt_xprt_switch_has_addr(clnt,
3274	sap: (const struct sockaddr *)&xprt->addr)) {
3275	return rpc_clnt_xprt_set_online(clnt, xprt);
3276	}
3277	rcu_read_lock();
3278	rpc_xprt_switch_add_xprt(rcu_dereference(clnt->cl_xpi.xpi_xpswitch),
3279	xprt);
3280	rcu_read_unlock();
3281	}
3282	EXPORT_SYMBOL_GPL(rpc_clnt_xprt_switch_add_xprt);
3283
3284	void rpc_clnt_xprt_switch_remove_xprt(struct rpc_clnt *clnt, struct rpc_xprt *xprt)
3285	{
3286	struct rpc_xprt_switch *xps;
3287
3288	rcu_read_lock();
3289	xps = rcu_dereference(clnt->cl_xpi.xpi_xpswitch);
3290	rpc_xprt_switch_remove_xprt(rcu_dereference(clnt->cl_xpi.xpi_xpswitch),
3291	xprt, offline: 0);
3292	xps->xps_nunique_destaddr_xprts--;
3293	rcu_read_unlock();
3294	}
3295	EXPORT_SYMBOL_GPL(rpc_clnt_xprt_switch_remove_xprt);
3296
3297	bool rpc_clnt_xprt_switch_has_addr(struct rpc_clnt *clnt,
3298	const struct sockaddr *sap)
3299	{
3300	struct rpc_xprt_switch *xps;
3301	bool ret;
3302
3303	rcu_read_lock();
3304	xps = rcu_dereference(clnt->cl_xpi.xpi_xpswitch);
3305	ret = rpc_xprt_switch_has_addr(xps, sap);
3306	rcu_read_unlock();
3307	return ret;
3308	}
3309	EXPORT_SYMBOL_GPL(rpc_clnt_xprt_switch_has_addr);
3310
3311	#if IS_ENABLED(CONFIG_SUNRPC_DEBUG)
3312	static void rpc_show_header(void)
3313	{
3314	printk(KERN_INFO "-pid- flgs status -client- --rqstp- "
3315	"-timeout ---ops--\n");
3316	}
3317
3318	static void rpc_show_task(const struct rpc_clnt *clnt,
3319	const struct rpc_task *task)
3320	{
3321	const char *rpc_waitq = "none";
3322
3323	if (RPC_IS_QUEUED(task))
3324	rpc_waitq = rpc_qname(q: task->tk_waitqueue);
3325
3326	printk(KERN_INFO "%5u %04x %6d %8p %8p %8ld %8p %sv%u %s a:%ps q:%s\n",
3327	task->tk_pid, task->tk_flags, task->tk_status,
3328	clnt, task->tk_rqstp, rpc_task_timeout(task), task->tk_ops,
3329	clnt->cl_program->name, clnt->cl_vers, rpc_proc_name(task),
3330	task->tk_action, rpc_waitq);
3331	}
3332
3333	void rpc_show_tasks(struct net *net)
3334	{
3335	struct rpc_clnt *clnt;
3336	struct rpc_task *task;
3337	int header = 0;
3338	struct sunrpc_net *sn = net_generic(net, id: sunrpc_net_id);
3339
3340	spin_lock(lock: &sn->rpc_client_lock);
3341	list_for_each_entry(clnt, &sn->all_clients, cl_clients) {
3342	spin_lock(lock: &clnt->cl_lock);
3343	list_for_each_entry(task, &clnt->cl_tasks, tk_task) {
3344	if (!header) {
3345	rpc_show_header();
3346	header++;
3347	}
3348	rpc_show_task(clnt, task);
3349	}
3350	spin_unlock(lock: &clnt->cl_lock);
3351	}
3352	spin_unlock(lock: &sn->rpc_client_lock);
3353	}
3354	#endif
3355
3356	#if IS_ENABLED(CONFIG_SUNRPC_SWAP)
3357	static int
3358	rpc_clnt_swap_activate_callback(struct rpc_clnt *clnt,
3359	struct rpc_xprt *xprt,
3360	void *dummy)
3361	{
3362	return xprt_enable_swap(xprt);
3363	}
3364
3365	int
3366	rpc_clnt_swap_activate(struct rpc_clnt *clnt)
3367	{
3368	while (clnt != clnt->cl_parent)
3369	clnt = clnt->cl_parent;
3370	if (atomic_inc_return(v: &clnt->cl_swapper) == 1)
3371	return rpc_clnt_iterate_for_each_xprt(clnt,
3372	rpc_clnt_swap_activate_callback, NULL);
3373	return 0;
3374	}
3375	EXPORT_SYMBOL_GPL(rpc_clnt_swap_activate);
3376
3377	static int
3378	rpc_clnt_swap_deactivate_callback(struct rpc_clnt *clnt,
3379	struct rpc_xprt *xprt,
3380	void *dummy)
3381	{
3382	xprt_disable_swap(xprt);
3383	return 0;
3384	}
3385
3386	void
3387	rpc_clnt_swap_deactivate(struct rpc_clnt *clnt)
3388	{
3389	while (clnt != clnt->cl_parent)
3390	clnt = clnt->cl_parent;
3391	if (atomic_dec_if_positive(v: &clnt->cl_swapper) == 0)
3392	rpc_clnt_iterate_for_each_xprt(clnt,
3393	rpc_clnt_swap_deactivate_callback, NULL);
3394	}
3395	EXPORT_SYMBOL_GPL(rpc_clnt_swap_deactivate);
3396	#endif /* CONFIG_SUNRPC_SWAP */
3397