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