1 // SPDX-License-Identifier: GPL-2.0-only
3 * linux/net/sunrpc/clnt.c
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.
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.
17 * Copyright (C) 1992,1993 Rick Sladkey <jrs@world.std.com>
18 * Copyright (C) 1995,1996 Olaf Kirch <okir@monad.swb.de>
22 #include <linux/module.h>
23 #include <linux/types.h>
24 #include <linux/kallsyms.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>
33 #include <linux/in6.h>
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>
47 #if IS_ENABLED(CONFIG_SUNRPC_DEBUG)
48 # define RPCDBG_FACILITY RPCDBG_CALL
52 * All RPC clients are linked into this list
55 static DECLARE_WAIT_QUEUE_HEAD(destroy_wait);
58 static void call_start(struct rpc_task *task);
59 static void call_reserve(struct rpc_task *task);
60 static void call_reserveresult(struct rpc_task *task);
61 static void call_allocate(struct rpc_task *task);
62 static void call_encode(struct rpc_task *task);
63 static void call_decode(struct rpc_task *task);
64 static void call_bind(struct rpc_task *task);
65 static void call_bind_status(struct rpc_task *task);
66 static void call_transmit(struct rpc_task *task);
67 static void call_status(struct rpc_task *task);
68 static void call_transmit_status(struct rpc_task *task);
69 static void call_refresh(struct rpc_task *task);
70 static void call_refreshresult(struct rpc_task *task);
71 static void call_connect(struct rpc_task *task);
72 static void call_connect_status(struct rpc_task *task);
74 static int rpc_encode_header(struct rpc_task *task,
75 struct xdr_stream *xdr);
76 static int rpc_decode_header(struct rpc_task *task,
77 struct xdr_stream *xdr);
78 static int rpc_ping(struct rpc_clnt *clnt);
79 static void rpc_check_timeout(struct rpc_task *task);
81 static void rpc_register_client(struct rpc_clnt *clnt)
83 struct net *net = rpc_net_ns(clnt);
84 struct sunrpc_net *sn = net_generic(net, sunrpc_net_id);
86 spin_lock(&sn->rpc_client_lock);
87 list_add(&clnt->cl_clients, &sn->all_clients);
88 spin_unlock(&sn->rpc_client_lock);
91 static void rpc_unregister_client(struct rpc_clnt *clnt)
93 struct net *net = rpc_net_ns(clnt);
94 struct sunrpc_net *sn = net_generic(net, sunrpc_net_id);
96 spin_lock(&sn->rpc_client_lock);
97 list_del(&clnt->cl_clients);
98 spin_unlock(&sn->rpc_client_lock);
101 static void __rpc_clnt_remove_pipedir(struct rpc_clnt *clnt)
103 rpc_remove_client_dir(clnt);
106 static void rpc_clnt_remove_pipedir(struct rpc_clnt *clnt)
108 struct net *net = rpc_net_ns(clnt);
109 struct super_block *pipefs_sb;
111 pipefs_sb = rpc_get_sb_net(net);
113 __rpc_clnt_remove_pipedir(clnt);
118 static struct dentry *rpc_setup_pipedir_sb(struct super_block *sb,
119 struct rpc_clnt *clnt)
121 static uint32_t clntid;
122 const char *dir_name = clnt->cl_program->pipe_dir_name;
124 struct dentry *dir, *dentry;
126 dir = rpc_d_lookup_sb(sb, dir_name);
128 pr_info("RPC: pipefs directory doesn't exist: %s\n", dir_name);
132 snprintf(name, sizeof(name), "clnt%x", (unsigned int)clntid++);
133 name[sizeof(name) - 1] = '\0';
134 dentry = rpc_create_client_dir(dir, name, clnt);
137 if (dentry == ERR_PTR(-EEXIST))
139 printk(KERN_INFO "RPC: Couldn't create pipefs entry"
140 " %s/%s, error %ld\n",
141 dir_name, name, PTR_ERR(dentry));
149 rpc_setup_pipedir(struct super_block *pipefs_sb, struct rpc_clnt *clnt)
151 struct dentry *dentry;
153 if (clnt->cl_program->pipe_dir_name != NULL) {
154 dentry = rpc_setup_pipedir_sb(pipefs_sb, clnt);
156 return PTR_ERR(dentry);
161 static int rpc_clnt_skip_event(struct rpc_clnt *clnt, unsigned long event)
163 if (clnt->cl_program->pipe_dir_name == NULL)
167 case RPC_PIPEFS_MOUNT:
168 if (clnt->cl_pipedir_objects.pdh_dentry != NULL)
170 if (atomic_read(&clnt->cl_count) == 0)
173 case RPC_PIPEFS_UMOUNT:
174 if (clnt->cl_pipedir_objects.pdh_dentry == NULL)
181 static int __rpc_clnt_handle_event(struct rpc_clnt *clnt, unsigned long event,
182 struct super_block *sb)
184 struct dentry *dentry;
187 case RPC_PIPEFS_MOUNT:
188 dentry = rpc_setup_pipedir_sb(sb, clnt);
192 return PTR_ERR(dentry);
194 case RPC_PIPEFS_UMOUNT:
195 __rpc_clnt_remove_pipedir(clnt);
198 printk(KERN_ERR "%s: unknown event: %ld\n", __func__, event);
204 static int __rpc_pipefs_event(struct rpc_clnt *clnt, unsigned long event,
205 struct super_block *sb)
209 for (;; clnt = clnt->cl_parent) {
210 if (!rpc_clnt_skip_event(clnt, event))
211 error = __rpc_clnt_handle_event(clnt, event, sb);
212 if (error || clnt == clnt->cl_parent)
218 static struct rpc_clnt *rpc_get_client_for_event(struct net *net, int event)
220 struct sunrpc_net *sn = net_generic(net, sunrpc_net_id);
221 struct rpc_clnt *clnt;
223 spin_lock(&sn->rpc_client_lock);
224 list_for_each_entry(clnt, &sn->all_clients, cl_clients) {
225 if (rpc_clnt_skip_event(clnt, event))
227 spin_unlock(&sn->rpc_client_lock);
230 spin_unlock(&sn->rpc_client_lock);
234 static int rpc_pipefs_event(struct notifier_block *nb, unsigned long event,
237 struct super_block *sb = ptr;
238 struct rpc_clnt *clnt;
241 while ((clnt = rpc_get_client_for_event(sb->s_fs_info, event))) {
242 error = __rpc_pipefs_event(clnt, event, sb);
249 static struct notifier_block rpc_clients_block = {
250 .notifier_call = rpc_pipefs_event,
251 .priority = SUNRPC_PIPEFS_RPC_PRIO,
254 int rpc_clients_notifier_register(void)
256 return rpc_pipefs_notifier_register(&rpc_clients_block);
259 void rpc_clients_notifier_unregister(void)
261 return rpc_pipefs_notifier_unregister(&rpc_clients_block);
264 static struct rpc_xprt *rpc_clnt_set_transport(struct rpc_clnt *clnt,
265 struct rpc_xprt *xprt,
266 const struct rpc_timeout *timeout)
268 struct rpc_xprt *old;
270 spin_lock(&clnt->cl_lock);
271 old = rcu_dereference_protected(clnt->cl_xprt,
272 lockdep_is_held(&clnt->cl_lock));
274 if (!xprt_bound(xprt))
275 clnt->cl_autobind = 1;
277 clnt->cl_timeout = timeout;
278 rcu_assign_pointer(clnt->cl_xprt, xprt);
279 spin_unlock(&clnt->cl_lock);
284 static void rpc_clnt_set_nodename(struct rpc_clnt *clnt, const char *nodename)
286 clnt->cl_nodelen = strlcpy(clnt->cl_nodename,
287 nodename, sizeof(clnt->cl_nodename));
290 static int rpc_client_register(struct rpc_clnt *clnt,
291 rpc_authflavor_t pseudoflavor,
292 const char *client_name)
294 struct rpc_auth_create_args auth_args = {
295 .pseudoflavor = pseudoflavor,
296 .target_name = client_name,
298 struct rpc_auth *auth;
299 struct net *net = rpc_net_ns(clnt);
300 struct super_block *pipefs_sb;
303 rpc_clnt_debugfs_register(clnt);
304 rpc_sysfs_client_setup(clnt, net);
306 pipefs_sb = rpc_get_sb_net(net);
308 err = rpc_setup_pipedir(pipefs_sb, clnt);
313 rpc_register_client(clnt);
317 auth = rpcauth_create(&auth_args, clnt);
319 dprintk("RPC: Couldn't create auth handle (flavor %u)\n",
326 pipefs_sb = rpc_get_sb_net(net);
327 rpc_unregister_client(clnt);
328 __rpc_clnt_remove_pipedir(clnt);
332 rpc_sysfs_client_destroy(clnt);
333 rpc_clnt_debugfs_unregister(clnt);
337 static DEFINE_IDA(rpc_clids);
339 void rpc_cleanup_clids(void)
341 ida_destroy(&rpc_clids);
344 static int rpc_alloc_clid(struct rpc_clnt *clnt)
348 clid = ida_simple_get(&rpc_clids, 0, 0, GFP_KERNEL);
351 clnt->cl_clid = clid;
355 static void rpc_free_clid(struct rpc_clnt *clnt)
357 ida_simple_remove(&rpc_clids, clnt->cl_clid);
360 static struct rpc_clnt * rpc_new_client(const struct rpc_create_args *args,
361 struct rpc_xprt_switch *xps,
362 struct rpc_xprt *xprt,
363 struct rpc_clnt *parent)
365 const struct rpc_program *program = args->program;
366 const struct rpc_version *version;
367 struct rpc_clnt *clnt = NULL;
368 const struct rpc_timeout *timeout;
369 const char *nodename = args->nodename;
377 if (args->version >= program->nrvers)
379 version = program->version[args->version];
384 clnt = kzalloc(sizeof(*clnt), GFP_KERNEL);
387 clnt->cl_parent = parent ? : clnt;
389 err = rpc_alloc_clid(clnt);
393 clnt->cl_cred = get_cred(args->cred);
394 clnt->cl_procinfo = version->procs;
395 clnt->cl_maxproc = version->nrprocs;
396 clnt->cl_prog = args->prognumber ? : program->number;
397 clnt->cl_vers = version->number;
398 clnt->cl_stats = program->stats;
399 clnt->cl_metrics = rpc_alloc_iostats(clnt);
400 rpc_init_pipe_dir_head(&clnt->cl_pipedir_objects);
402 if (clnt->cl_metrics == NULL)
404 clnt->cl_program = program;
405 INIT_LIST_HEAD(&clnt->cl_tasks);
406 spin_lock_init(&clnt->cl_lock);
408 timeout = xprt->timeout;
409 if (args->timeout != NULL) {
410 memcpy(&clnt->cl_timeout_default, args->timeout,
411 sizeof(clnt->cl_timeout_default));
412 timeout = &clnt->cl_timeout_default;
415 rpc_clnt_set_transport(clnt, xprt, timeout);
416 xprt_iter_init(&clnt->cl_xpi, xps);
417 xprt_switch_put(xps);
419 clnt->cl_rtt = &clnt->cl_rtt_default;
420 rpc_init_rtt(&clnt->cl_rtt_default, clnt->cl_timeout->to_initval);
422 atomic_set(&clnt->cl_count, 1);
424 if (nodename == NULL)
425 nodename = utsname()->nodename;
426 /* save the nodename */
427 rpc_clnt_set_nodename(clnt, nodename);
429 err = rpc_client_register(clnt, args->authflavor, args->client_name);
433 atomic_inc(&parent->cl_count);
435 trace_rpc_clnt_new(clnt, xprt, program->name, args->servername);
439 rpc_free_iostats(clnt->cl_metrics);
441 put_cred(clnt->cl_cred);
448 xprt_switch_put(xps);
450 trace_rpc_clnt_new_err(program->name, args->servername, err);
454 static struct rpc_clnt *rpc_create_xprt(struct rpc_create_args *args,
455 struct rpc_xprt *xprt)
457 struct rpc_clnt *clnt = NULL;
458 struct rpc_xprt_switch *xps;
460 if (args->bc_xprt && args->bc_xprt->xpt_bc_xps) {
461 WARN_ON_ONCE(!(args->protocol & XPRT_TRANSPORT_BC));
462 xps = args->bc_xprt->xpt_bc_xps;
463 xprt_switch_get(xps);
465 xps = xprt_switch_alloc(xprt, GFP_KERNEL);
468 return ERR_PTR(-ENOMEM);
471 xprt_switch_get(xps);
472 xprt->bc_xprt->xpt_bc_xps = xps;
475 clnt = rpc_new_client(args, xps, xprt, NULL);
479 if (!(args->flags & RPC_CLNT_CREATE_NOPING)) {
480 int err = rpc_ping(clnt);
482 rpc_shutdown_client(clnt);
487 clnt->cl_softrtry = 1;
488 if (args->flags & (RPC_CLNT_CREATE_HARDRTRY|RPC_CLNT_CREATE_SOFTERR)) {
489 clnt->cl_softrtry = 0;
490 if (args->flags & RPC_CLNT_CREATE_SOFTERR)
491 clnt->cl_softerr = 1;
494 if (args->flags & RPC_CLNT_CREATE_AUTOBIND)
495 clnt->cl_autobind = 1;
496 if (args->flags & RPC_CLNT_CREATE_NO_RETRANS_TIMEOUT)
497 clnt->cl_noretranstimeo = 1;
498 if (args->flags & RPC_CLNT_CREATE_DISCRTRY)
499 clnt->cl_discrtry = 1;
500 if (!(args->flags & RPC_CLNT_CREATE_QUIET))
507 * rpc_create - create an RPC client and transport with one call
508 * @args: rpc_clnt create argument structure
510 * Creates and initializes an RPC transport and an RPC client.
512 * It can ping the server in order to determine if it is up, and to see if
513 * it supports this program and version. RPC_CLNT_CREATE_NOPING disables
514 * this behavior so asynchronous tasks can also use rpc_create.
516 struct rpc_clnt *rpc_create(struct rpc_create_args *args)
518 struct rpc_xprt *xprt;
519 struct xprt_create xprtargs = {
521 .ident = args->protocol,
522 .srcaddr = args->saddress,
523 .dstaddr = args->address,
524 .addrlen = args->addrsize,
525 .servername = args->servername,
526 .bc_xprt = args->bc_xprt,
529 struct rpc_clnt *clnt;
533 WARN_ON_ONCE(!(args->protocol & XPRT_TRANSPORT_BC));
534 xprt = args->bc_xprt->xpt_bc_xprt;
537 return rpc_create_xprt(args, xprt);
541 if (args->flags & RPC_CLNT_CREATE_INFINITE_SLOTS)
542 xprtargs.flags |= XPRT_CREATE_INFINITE_SLOTS;
543 if (args->flags & RPC_CLNT_CREATE_NO_IDLE_TIMEOUT)
544 xprtargs.flags |= XPRT_CREATE_NO_IDLE_TIMEOUT;
546 * If the caller chooses not to specify a hostname, whip
547 * up a string representation of the passed-in address.
549 if (xprtargs.servername == NULL) {
550 struct sockaddr_un *sun =
551 (struct sockaddr_un *)args->address;
552 struct sockaddr_in *sin =
553 (struct sockaddr_in *)args->address;
554 struct sockaddr_in6 *sin6 =
555 (struct sockaddr_in6 *)args->address;
557 servername[0] = '\0';
558 switch (args->address->sa_family) {
560 snprintf(servername, sizeof(servername), "%s",
564 snprintf(servername, sizeof(servername), "%pI4",
565 &sin->sin_addr.s_addr);
568 snprintf(servername, sizeof(servername), "%pI6",
572 /* caller wants default server name, but
573 * address family isn't recognized. */
574 return ERR_PTR(-EINVAL);
576 xprtargs.servername = servername;
579 xprt = xprt_create_transport(&xprtargs);
581 return (struct rpc_clnt *)xprt;
584 * By default, kernel RPC client connects from a reserved port.
585 * CAP_NET_BIND_SERVICE will not be set for unprivileged requesters,
586 * but it is always enabled for rpciod, which handles the connect
590 if (args->flags & RPC_CLNT_CREATE_NONPRIVPORT)
593 if (args->flags & RPC_CLNT_CREATE_REUSEPORT)
596 clnt = rpc_create_xprt(args, xprt);
597 if (IS_ERR(clnt) || args->nconnect <= 1)
600 for (i = 0; i < args->nconnect - 1; i++) {
601 if (rpc_clnt_add_xprt(clnt, &xprtargs, NULL, NULL) < 0)
606 EXPORT_SYMBOL_GPL(rpc_create);
609 * This function clones the RPC client structure. It allows us to share the
610 * same transport while varying parameters such as the authentication
613 static struct rpc_clnt *__rpc_clone_client(struct rpc_create_args *args,
614 struct rpc_clnt *clnt)
616 struct rpc_xprt_switch *xps;
617 struct rpc_xprt *xprt;
618 struct rpc_clnt *new;
623 xprt = xprt_get(rcu_dereference(clnt->cl_xprt));
624 xps = xprt_switch_get(rcu_dereference(clnt->cl_xpi.xpi_xpswitch));
626 if (xprt == NULL || xps == NULL) {
628 xprt_switch_put(xps);
631 args->servername = xprt->servername;
632 args->nodename = clnt->cl_nodename;
634 new = rpc_new_client(args, xps, xprt, clnt);
638 /* Turn off autobind on clones */
639 new->cl_autobind = 0;
640 new->cl_softrtry = clnt->cl_softrtry;
641 new->cl_softerr = clnt->cl_softerr;
642 new->cl_noretranstimeo = clnt->cl_noretranstimeo;
643 new->cl_discrtry = clnt->cl_discrtry;
644 new->cl_chatty = clnt->cl_chatty;
645 new->cl_principal = clnt->cl_principal;
649 trace_rpc_clnt_clone_err(clnt, err);
654 * rpc_clone_client - Clone an RPC client structure
656 * @clnt: RPC client whose parameters are copied
658 * Returns a fresh RPC client or an ERR_PTR.
660 struct rpc_clnt *rpc_clone_client(struct rpc_clnt *clnt)
662 struct rpc_create_args args = {
663 .program = clnt->cl_program,
664 .prognumber = clnt->cl_prog,
665 .version = clnt->cl_vers,
666 .authflavor = clnt->cl_auth->au_flavor,
667 .cred = clnt->cl_cred,
669 return __rpc_clone_client(&args, clnt);
671 EXPORT_SYMBOL_GPL(rpc_clone_client);
674 * rpc_clone_client_set_auth - Clone an RPC client structure and set its auth
676 * @clnt: RPC client whose parameters are copied
677 * @flavor: security flavor for new client
679 * Returns a fresh RPC client or an ERR_PTR.
682 rpc_clone_client_set_auth(struct rpc_clnt *clnt, rpc_authflavor_t flavor)
684 struct rpc_create_args args = {
685 .program = clnt->cl_program,
686 .prognumber = clnt->cl_prog,
687 .version = clnt->cl_vers,
688 .authflavor = flavor,
689 .cred = clnt->cl_cred,
691 return __rpc_clone_client(&args, clnt);
693 EXPORT_SYMBOL_GPL(rpc_clone_client_set_auth);
696 * rpc_switch_client_transport: switch the RPC transport on the fly
697 * @clnt: pointer to a struct rpc_clnt
698 * @args: pointer to the new transport arguments
699 * @timeout: pointer to the new timeout parameters
701 * This function allows the caller to switch the RPC transport for the
702 * rpc_clnt structure 'clnt' to allow it to connect to a mirrored NFS
703 * server, for instance. It assumes that the caller has ensured that
704 * there are no active RPC tasks by using some form of locking.
706 * Returns zero if "clnt" is now using the new xprt. Otherwise a
707 * negative errno is returned, and "clnt" continues to use the old
710 int rpc_switch_client_transport(struct rpc_clnt *clnt,
711 struct xprt_create *args,
712 const struct rpc_timeout *timeout)
714 const struct rpc_timeout *old_timeo;
715 rpc_authflavor_t pseudoflavor;
716 struct rpc_xprt_switch *xps, *oldxps;
717 struct rpc_xprt *xprt, *old;
718 struct rpc_clnt *parent;
721 xprt = xprt_create_transport(args);
723 return PTR_ERR(xprt);
725 xps = xprt_switch_alloc(xprt, GFP_KERNEL);
731 pseudoflavor = clnt->cl_auth->au_flavor;
733 old_timeo = clnt->cl_timeout;
734 old = rpc_clnt_set_transport(clnt, xprt, timeout);
735 oldxps = xprt_iter_xchg_switch(&clnt->cl_xpi, xps);
737 rpc_unregister_client(clnt);
738 __rpc_clnt_remove_pipedir(clnt);
739 rpc_sysfs_client_destroy(clnt);
740 rpc_clnt_debugfs_unregister(clnt);
743 * A new transport was created. "clnt" therefore
744 * becomes the root of a new cl_parent tree. clnt's
745 * children, if it has any, still point to the old xprt.
747 parent = clnt->cl_parent;
748 clnt->cl_parent = clnt;
751 * The old rpc_auth cache cannot be re-used. GSS
752 * contexts in particular are between a single
755 err = rpc_client_register(clnt, pseudoflavor, NULL);
761 rpc_release_client(parent);
762 xprt_switch_put(oldxps);
764 trace_rpc_clnt_replace_xprt(clnt);
768 xps = xprt_iter_xchg_switch(&clnt->cl_xpi, oldxps);
769 rpc_clnt_set_transport(clnt, old, old_timeo);
770 clnt->cl_parent = parent;
771 rpc_client_register(clnt, pseudoflavor, NULL);
772 xprt_switch_put(xps);
774 trace_rpc_clnt_replace_xprt_err(clnt);
777 EXPORT_SYMBOL_GPL(rpc_switch_client_transport);
780 int rpc_clnt_xprt_iter_init(struct rpc_clnt *clnt, struct rpc_xprt_iter *xpi)
782 struct rpc_xprt_switch *xps;
785 xps = xprt_switch_get(rcu_dereference(clnt->cl_xpi.xpi_xpswitch));
789 xprt_iter_init_listall(xpi, xps);
790 xprt_switch_put(xps);
795 * rpc_clnt_iterate_for_each_xprt - Apply a function to all transports
796 * @clnt: pointer to client
797 * @fn: function to apply
798 * @data: void pointer to function data
800 * Iterates through the list of RPC transports currently attached to the
801 * client and applies the function fn(clnt, xprt, data).
803 * On error, the iteration stops, and the function returns the error value.
805 int rpc_clnt_iterate_for_each_xprt(struct rpc_clnt *clnt,
806 int (*fn)(struct rpc_clnt *, struct rpc_xprt *, void *),
809 struct rpc_xprt_iter xpi;
812 ret = rpc_clnt_xprt_iter_init(clnt, &xpi);
816 struct rpc_xprt *xprt = xprt_iter_get_next(&xpi);
820 ret = fn(clnt, xprt, data);
825 xprt_iter_destroy(&xpi);
828 EXPORT_SYMBOL_GPL(rpc_clnt_iterate_for_each_xprt);
831 * Kill all tasks for the given client.
832 * XXX: kill their descendants as well?
834 void rpc_killall_tasks(struct rpc_clnt *clnt)
836 struct rpc_task *rovr;
839 if (list_empty(&clnt->cl_tasks))
843 * Spin lock all_tasks to prevent changes...
845 trace_rpc_clnt_killall(clnt);
846 spin_lock(&clnt->cl_lock);
847 list_for_each_entry(rovr, &clnt->cl_tasks, tk_task)
848 rpc_signal_task(rovr);
849 spin_unlock(&clnt->cl_lock);
851 EXPORT_SYMBOL_GPL(rpc_killall_tasks);
854 * Properly shut down an RPC client, terminating all outstanding
857 void rpc_shutdown_client(struct rpc_clnt *clnt)
861 trace_rpc_clnt_shutdown(clnt);
863 while (!list_empty(&clnt->cl_tasks)) {
864 rpc_killall_tasks(clnt);
865 wait_event_timeout(destroy_wait,
866 list_empty(&clnt->cl_tasks), 1*HZ);
869 rpc_release_client(clnt);
871 EXPORT_SYMBOL_GPL(rpc_shutdown_client);
876 static void rpc_free_client_work(struct work_struct *work)
878 struct rpc_clnt *clnt = container_of(work, struct rpc_clnt, cl_work);
880 trace_rpc_clnt_free(clnt);
882 /* These might block on processes that might allocate memory,
883 * so they cannot be called in rpciod, so they are handled separately
886 rpc_sysfs_client_destroy(clnt);
887 rpc_clnt_debugfs_unregister(clnt);
889 rpc_clnt_remove_pipedir(clnt);
890 xprt_put(rcu_dereference_raw(clnt->cl_xprt));
895 static struct rpc_clnt *
896 rpc_free_client(struct rpc_clnt *clnt)
898 struct rpc_clnt *parent = NULL;
900 trace_rpc_clnt_release(clnt);
901 if (clnt->cl_parent != clnt)
902 parent = clnt->cl_parent;
903 rpc_unregister_client(clnt);
904 rpc_free_iostats(clnt->cl_metrics);
905 clnt->cl_metrics = NULL;
906 xprt_iter_destroy(&clnt->cl_xpi);
907 put_cred(clnt->cl_cred);
909 INIT_WORK(&clnt->cl_work, rpc_free_client_work);
910 schedule_work(&clnt->cl_work);
917 static struct rpc_clnt *
918 rpc_free_auth(struct rpc_clnt *clnt)
920 if (clnt->cl_auth == NULL)
921 return rpc_free_client(clnt);
924 * Note: RPCSEC_GSS may need to send NULL RPC calls in order to
925 * release remaining GSS contexts. This mechanism ensures
926 * that it can do so safely.
928 atomic_inc(&clnt->cl_count);
929 rpcauth_release(clnt->cl_auth);
930 clnt->cl_auth = NULL;
931 if (atomic_dec_and_test(&clnt->cl_count))
932 return rpc_free_client(clnt);
937 * Release reference to the RPC client
940 rpc_release_client(struct rpc_clnt *clnt)
943 if (list_empty(&clnt->cl_tasks))
944 wake_up(&destroy_wait);
945 if (!atomic_dec_and_test(&clnt->cl_count))
947 clnt = rpc_free_auth(clnt);
948 } while (clnt != NULL);
950 EXPORT_SYMBOL_GPL(rpc_release_client);
953 * rpc_bind_new_program - bind a new RPC program to an existing client
954 * @old: old rpc_client
955 * @program: rpc program to set
956 * @vers: rpc program version
958 * Clones the rpc client and sets up a new RPC program. This is mainly
959 * of use for enabling different RPC programs to share the same transport.
960 * The Sun NFSv2/v3 ACL protocol can do this.
962 struct rpc_clnt *rpc_bind_new_program(struct rpc_clnt *old,
963 const struct rpc_program *program,
966 struct rpc_create_args args = {
968 .prognumber = program->number,
970 .authflavor = old->cl_auth->au_flavor,
971 .cred = old->cl_cred,
973 struct rpc_clnt *clnt;
976 clnt = __rpc_clone_client(&args, old);
979 err = rpc_ping(clnt);
981 rpc_shutdown_client(clnt);
987 EXPORT_SYMBOL_GPL(rpc_bind_new_program);
990 rpc_task_get_xprt(struct rpc_clnt *clnt, struct rpc_xprt *xprt)
992 struct rpc_xprt_switch *xps;
997 xps = rcu_dereference(clnt->cl_xpi.xpi_xpswitch);
998 atomic_long_inc(&xps->xps_queuelen);
1000 atomic_long_inc(&xprt->queuelen);
1006 rpc_task_release_xprt(struct rpc_clnt *clnt, struct rpc_xprt *xprt)
1008 struct rpc_xprt_switch *xps;
1010 atomic_long_dec(&xprt->queuelen);
1012 xps = rcu_dereference(clnt->cl_xpi.xpi_xpswitch);
1013 atomic_long_dec(&xps->xps_queuelen);
1019 void rpc_task_release_transport(struct rpc_task *task)
1021 struct rpc_xprt *xprt = task->tk_xprt;
1024 task->tk_xprt = NULL;
1025 if (task->tk_client)
1026 rpc_task_release_xprt(task->tk_client, xprt);
1031 EXPORT_SYMBOL_GPL(rpc_task_release_transport);
1033 void rpc_task_release_client(struct rpc_task *task)
1035 struct rpc_clnt *clnt = task->tk_client;
1037 rpc_task_release_transport(task);
1039 /* Remove from client task list */
1040 spin_lock(&clnt->cl_lock);
1041 list_del(&task->tk_task);
1042 spin_unlock(&clnt->cl_lock);
1043 task->tk_client = NULL;
1045 rpc_release_client(clnt);
1049 static struct rpc_xprt *
1050 rpc_task_get_first_xprt(struct rpc_clnt *clnt)
1052 struct rpc_xprt *xprt;
1055 xprt = xprt_get(rcu_dereference(clnt->cl_xprt));
1057 return rpc_task_get_xprt(clnt, xprt);
1060 static struct rpc_xprt *
1061 rpc_task_get_next_xprt(struct rpc_clnt *clnt)
1063 return rpc_task_get_xprt(clnt, xprt_iter_get_next(&clnt->cl_xpi));
1067 void rpc_task_set_transport(struct rpc_task *task, struct rpc_clnt *clnt)
1071 if (task->tk_flags & RPC_TASK_NO_ROUND_ROBIN)
1072 task->tk_xprt = rpc_task_get_first_xprt(clnt);
1074 task->tk_xprt = rpc_task_get_next_xprt(clnt);
1078 void rpc_task_set_client(struct rpc_task *task, struct rpc_clnt *clnt)
1082 rpc_task_set_transport(task, clnt);
1083 task->tk_client = clnt;
1084 atomic_inc(&clnt->cl_count);
1085 if (clnt->cl_softrtry)
1086 task->tk_flags |= RPC_TASK_SOFT;
1087 if (clnt->cl_softerr)
1088 task->tk_flags |= RPC_TASK_TIMEOUT;
1089 if (clnt->cl_noretranstimeo)
1090 task->tk_flags |= RPC_TASK_NO_RETRANS_TIMEOUT;
1091 if (atomic_read(&clnt->cl_swapper))
1092 task->tk_flags |= RPC_TASK_SWAPPER;
1093 /* Add to the client's list of all tasks */
1094 spin_lock(&clnt->cl_lock);
1095 list_add_tail(&task->tk_task, &clnt->cl_tasks);
1096 spin_unlock(&clnt->cl_lock);
1101 rpc_task_set_rpc_message(struct rpc_task *task, const struct rpc_message *msg)
1104 task->tk_msg.rpc_proc = msg->rpc_proc;
1105 task->tk_msg.rpc_argp = msg->rpc_argp;
1106 task->tk_msg.rpc_resp = msg->rpc_resp;
1107 task->tk_msg.rpc_cred = msg->rpc_cred;
1108 if (!(task->tk_flags & RPC_TASK_CRED_NOREF))
1109 get_cred(task->tk_msg.rpc_cred);
1114 * Default callback for async RPC calls
1117 rpc_default_callback(struct rpc_task *task, void *data)
1121 static const struct rpc_call_ops rpc_default_ops = {
1122 .rpc_call_done = rpc_default_callback,
1126 * rpc_run_task - Allocate a new RPC task, then run rpc_execute against it
1127 * @task_setup_data: pointer to task initialisation data
1129 struct rpc_task *rpc_run_task(const struct rpc_task_setup *task_setup_data)
1131 struct rpc_task *task;
1133 task = rpc_new_task(task_setup_data);
1135 if (!RPC_IS_ASYNC(task))
1136 task->tk_flags |= RPC_TASK_CRED_NOREF;
1138 rpc_task_set_client(task, task_setup_data->rpc_client);
1139 rpc_task_set_rpc_message(task, task_setup_data->rpc_message);
1141 if (task->tk_action == NULL)
1142 rpc_call_start(task);
1144 atomic_inc(&task->tk_count);
1148 EXPORT_SYMBOL_GPL(rpc_run_task);
1151 * rpc_call_sync - Perform a synchronous RPC call
1152 * @clnt: pointer to RPC client
1153 * @msg: RPC call parameters
1154 * @flags: RPC call flags
1156 int rpc_call_sync(struct rpc_clnt *clnt, const struct rpc_message *msg, int flags)
1158 struct rpc_task *task;
1159 struct rpc_task_setup task_setup_data = {
1162 .callback_ops = &rpc_default_ops,
1167 WARN_ON_ONCE(flags & RPC_TASK_ASYNC);
1168 if (flags & RPC_TASK_ASYNC) {
1169 rpc_release_calldata(task_setup_data.callback_ops,
1170 task_setup_data.callback_data);
1174 task = rpc_run_task(&task_setup_data);
1176 return PTR_ERR(task);
1177 status = task->tk_status;
1181 EXPORT_SYMBOL_GPL(rpc_call_sync);
1184 * rpc_call_async - Perform an asynchronous RPC call
1185 * @clnt: pointer to RPC client
1186 * @msg: RPC call parameters
1187 * @flags: RPC call flags
1188 * @tk_ops: RPC call ops
1189 * @data: user call data
1192 rpc_call_async(struct rpc_clnt *clnt, const struct rpc_message *msg, int flags,
1193 const struct rpc_call_ops *tk_ops, void *data)
1195 struct rpc_task *task;
1196 struct rpc_task_setup task_setup_data = {
1199 .callback_ops = tk_ops,
1200 .callback_data = data,
1201 .flags = flags|RPC_TASK_ASYNC,
1204 task = rpc_run_task(&task_setup_data);
1206 return PTR_ERR(task);
1210 EXPORT_SYMBOL_GPL(rpc_call_async);
1212 #if defined(CONFIG_SUNRPC_BACKCHANNEL)
1213 static void call_bc_encode(struct rpc_task *task);
1216 * rpc_run_bc_task - Allocate a new RPC task for backchannel use, then run
1217 * rpc_execute against it
1220 struct rpc_task *rpc_run_bc_task(struct rpc_rqst *req)
1222 struct rpc_task *task;
1223 struct rpc_task_setup task_setup_data = {
1224 .callback_ops = &rpc_default_ops,
1225 .flags = RPC_TASK_SOFTCONN |
1226 RPC_TASK_NO_RETRANS_TIMEOUT,
1229 dprintk("RPC: rpc_run_bc_task req= %p\n", req);
1231 * Create an rpc_task to send the data
1233 task = rpc_new_task(&task_setup_data);
1234 xprt_init_bc_request(req, task);
1236 task->tk_action = call_bc_encode;
1237 atomic_inc(&task->tk_count);
1238 WARN_ON_ONCE(atomic_read(&task->tk_count) != 2);
1241 dprintk("RPC: rpc_run_bc_task: task= %p\n", task);
1244 #endif /* CONFIG_SUNRPC_BACKCHANNEL */
1247 * rpc_prepare_reply_pages - Prepare to receive a reply data payload into pages
1248 * @req: RPC request to prepare
1249 * @pages: vector of struct page pointers
1250 * @base: offset in first page where receive should start, in bytes
1251 * @len: expected size of the upper layer data payload, in bytes
1252 * @hdrsize: expected size of upper layer reply header, in XDR words
1255 void rpc_prepare_reply_pages(struct rpc_rqst *req, struct page **pages,
1256 unsigned int base, unsigned int len,
1257 unsigned int hdrsize)
1259 hdrsize += RPC_REPHDRSIZE + req->rq_cred->cr_auth->au_ralign;
1261 xdr_inline_pages(&req->rq_rcv_buf, hdrsize << 2, pages, base, len);
1262 trace_rpc_xdr_reply_pages(req->rq_task, &req->rq_rcv_buf);
1264 EXPORT_SYMBOL_GPL(rpc_prepare_reply_pages);
1267 rpc_call_start(struct rpc_task *task)
1269 task->tk_action = call_start;
1271 EXPORT_SYMBOL_GPL(rpc_call_start);
1274 * rpc_peeraddr - extract remote peer address from clnt's xprt
1275 * @clnt: RPC client structure
1276 * @buf: target buffer
1277 * @bufsize: length of target buffer
1279 * Returns the number of bytes that are actually in the stored address.
1281 size_t rpc_peeraddr(struct rpc_clnt *clnt, struct sockaddr *buf, size_t bufsize)
1284 struct rpc_xprt *xprt;
1287 xprt = rcu_dereference(clnt->cl_xprt);
1289 bytes = xprt->addrlen;
1290 if (bytes > bufsize)
1292 memcpy(buf, &xprt->addr, bytes);
1297 EXPORT_SYMBOL_GPL(rpc_peeraddr);
1300 * rpc_peeraddr2str - return remote peer address in printable format
1301 * @clnt: RPC client structure
1302 * @format: address format
1304 * NB: the lifetime of the memory referenced by the returned pointer is
1305 * the same as the rpc_xprt itself. As long as the caller uses this
1306 * pointer, it must hold the RCU read lock.
1308 const char *rpc_peeraddr2str(struct rpc_clnt *clnt,
1309 enum rpc_display_format_t format)
1311 struct rpc_xprt *xprt;
1313 xprt = rcu_dereference(clnt->cl_xprt);
1315 if (xprt->address_strings[format] != NULL)
1316 return xprt->address_strings[format];
1318 return "unprintable";
1320 EXPORT_SYMBOL_GPL(rpc_peeraddr2str);
1322 static const struct sockaddr_in rpc_inaddr_loopback = {
1323 .sin_family = AF_INET,
1324 .sin_addr.s_addr = htonl(INADDR_ANY),
1327 static const struct sockaddr_in6 rpc_in6addr_loopback = {
1328 .sin6_family = AF_INET6,
1329 .sin6_addr = IN6ADDR_ANY_INIT,
1333 * Try a getsockname() on a connected datagram socket. Using a
1334 * connected datagram socket prevents leaving a socket in TIME_WAIT.
1335 * This conserves the ephemeral port number space.
1337 * Returns zero and fills in "buf" if successful; otherwise, a
1338 * negative errno is returned.
1340 static int rpc_sockname(struct net *net, struct sockaddr *sap, size_t salen,
1341 struct sockaddr *buf)
1343 struct socket *sock;
1346 err = __sock_create(net, sap->sa_family,
1347 SOCK_DGRAM, IPPROTO_UDP, &sock, 1);
1349 dprintk("RPC: can't create UDP socket (%d)\n", err);
1353 switch (sap->sa_family) {
1355 err = kernel_bind(sock,
1356 (struct sockaddr *)&rpc_inaddr_loopback,
1357 sizeof(rpc_inaddr_loopback));
1360 err = kernel_bind(sock,
1361 (struct sockaddr *)&rpc_in6addr_loopback,
1362 sizeof(rpc_in6addr_loopback));
1365 err = -EAFNOSUPPORT;
1369 dprintk("RPC: can't bind UDP socket (%d)\n", err);
1373 err = kernel_connect(sock, sap, salen, 0);
1375 dprintk("RPC: can't connect UDP socket (%d)\n", err);
1379 err = kernel_getsockname(sock, buf);
1381 dprintk("RPC: getsockname failed (%d)\n", err);
1386 if (buf->sa_family == AF_INET6) {
1387 struct sockaddr_in6 *sin6 = (struct sockaddr_in6 *)buf;
1388 sin6->sin6_scope_id = 0;
1390 dprintk("RPC: %s succeeded\n", __func__);
1399 * Scraping a connected socket failed, so we don't have a useable
1400 * local address. Fallback: generate an address that will prevent
1401 * the server from calling us back.
1403 * Returns zero and fills in "buf" if successful; otherwise, a
1404 * negative errno is returned.
1406 static int rpc_anyaddr(int family, struct sockaddr *buf, size_t buflen)
1410 if (buflen < sizeof(rpc_inaddr_loopback))
1412 memcpy(buf, &rpc_inaddr_loopback,
1413 sizeof(rpc_inaddr_loopback));
1416 if (buflen < sizeof(rpc_in6addr_loopback))
1418 memcpy(buf, &rpc_in6addr_loopback,
1419 sizeof(rpc_in6addr_loopback));
1422 dprintk("RPC: %s: address family not supported\n",
1424 return -EAFNOSUPPORT;
1426 dprintk("RPC: %s: succeeded\n", __func__);
1431 * rpc_localaddr - discover local endpoint address for an RPC client
1432 * @clnt: RPC client structure
1433 * @buf: target buffer
1434 * @buflen: size of target buffer, in bytes
1436 * Returns zero and fills in "buf" and "buflen" if successful;
1437 * otherwise, a negative errno is returned.
1439 * This works even if the underlying transport is not currently connected,
1440 * or if the upper layer never previously provided a source address.
1442 * The result of this function call is transient: multiple calls in
1443 * succession may give different results, depending on how local
1444 * networking configuration changes over time.
1446 int rpc_localaddr(struct rpc_clnt *clnt, struct sockaddr *buf, size_t buflen)
1448 struct sockaddr_storage address;
1449 struct sockaddr *sap = (struct sockaddr *)&address;
1450 struct rpc_xprt *xprt;
1456 xprt = rcu_dereference(clnt->cl_xprt);
1457 salen = xprt->addrlen;
1458 memcpy(sap, &xprt->addr, salen);
1459 net = get_net(xprt->xprt_net);
1462 rpc_set_port(sap, 0);
1463 err = rpc_sockname(net, sap, salen, buf);
1466 /* Couldn't discover local address, return ANYADDR */
1467 return rpc_anyaddr(sap->sa_family, buf, buflen);
1470 EXPORT_SYMBOL_GPL(rpc_localaddr);
1473 rpc_setbufsize(struct rpc_clnt *clnt, unsigned int sndsize, unsigned int rcvsize)
1475 struct rpc_xprt *xprt;
1478 xprt = rcu_dereference(clnt->cl_xprt);
1479 if (xprt->ops->set_buffer_size)
1480 xprt->ops->set_buffer_size(xprt, sndsize, rcvsize);
1483 EXPORT_SYMBOL_GPL(rpc_setbufsize);
1486 * rpc_net_ns - Get the network namespace for this RPC client
1487 * @clnt: RPC client to query
1490 struct net *rpc_net_ns(struct rpc_clnt *clnt)
1495 ret = rcu_dereference(clnt->cl_xprt)->xprt_net;
1499 EXPORT_SYMBOL_GPL(rpc_net_ns);
1502 * rpc_max_payload - Get maximum payload size for a transport, in bytes
1503 * @clnt: RPC client to query
1505 * For stream transports, this is one RPC record fragment (see RFC
1506 * 1831), as we don't support multi-record requests yet. For datagram
1507 * transports, this is the size of an IP packet minus the IP, UDP, and
1510 size_t rpc_max_payload(struct rpc_clnt *clnt)
1515 ret = rcu_dereference(clnt->cl_xprt)->max_payload;
1519 EXPORT_SYMBOL_GPL(rpc_max_payload);
1522 * rpc_max_bc_payload - Get maximum backchannel payload size, in bytes
1523 * @clnt: RPC client to query
1525 size_t rpc_max_bc_payload(struct rpc_clnt *clnt)
1527 struct rpc_xprt *xprt;
1531 xprt = rcu_dereference(clnt->cl_xprt);
1532 ret = xprt->ops->bc_maxpayload(xprt);
1536 EXPORT_SYMBOL_GPL(rpc_max_bc_payload);
1538 unsigned int rpc_num_bc_slots(struct rpc_clnt *clnt)
1540 struct rpc_xprt *xprt;
1544 xprt = rcu_dereference(clnt->cl_xprt);
1545 ret = xprt->ops->bc_num_slots(xprt);
1549 EXPORT_SYMBOL_GPL(rpc_num_bc_slots);
1552 * rpc_force_rebind - force transport to check that remote port is unchanged
1553 * @clnt: client to rebind
1556 void rpc_force_rebind(struct rpc_clnt *clnt)
1558 if (clnt->cl_autobind) {
1560 xprt_clear_bound(rcu_dereference(clnt->cl_xprt));
1564 EXPORT_SYMBOL_GPL(rpc_force_rebind);
1567 __rpc_restart_call(struct rpc_task *task, void (*action)(struct rpc_task *))
1569 task->tk_status = 0;
1570 task->tk_rpc_status = 0;
1571 task->tk_action = action;
1576 * Restart an (async) RPC call. Usually called from within the
1580 rpc_restart_call(struct rpc_task *task)
1582 return __rpc_restart_call(task, call_start);
1584 EXPORT_SYMBOL_GPL(rpc_restart_call);
1587 * Restart an (async) RPC call from the call_prepare state.
1588 * Usually called from within the exit handler.
1591 rpc_restart_call_prepare(struct rpc_task *task)
1593 if (task->tk_ops->rpc_call_prepare != NULL)
1594 return __rpc_restart_call(task, rpc_prepare_task);
1595 return rpc_restart_call(task);
1597 EXPORT_SYMBOL_GPL(rpc_restart_call_prepare);
1600 *rpc_proc_name(const struct rpc_task *task)
1602 const struct rpc_procinfo *proc = task->tk_msg.rpc_proc;
1606 return proc->p_name;
1614 __rpc_call_rpcerror(struct rpc_task *task, int tk_status, int rpc_status)
1616 trace_rpc_call_rpcerror(task, tk_status, rpc_status);
1617 task->tk_rpc_status = rpc_status;
1618 rpc_exit(task, tk_status);
1622 rpc_call_rpcerror(struct rpc_task *task, int status)
1624 __rpc_call_rpcerror(task, status, status);
1630 * Other FSM states can be visited zero or more times, but
1631 * this state is visited exactly once for each RPC.
1634 call_start(struct rpc_task *task)
1636 struct rpc_clnt *clnt = task->tk_client;
1637 int idx = task->tk_msg.rpc_proc->p_statidx;
1639 trace_rpc_request(task);
1641 /* Increment call count (version might not be valid for ping) */
1642 if (clnt->cl_program->version[clnt->cl_vers])
1643 clnt->cl_program->version[clnt->cl_vers]->counts[idx]++;
1644 clnt->cl_stats->rpccnt++;
1645 task->tk_action = call_reserve;
1646 rpc_task_set_transport(task, clnt);
1650 * 1. Reserve an RPC call slot
1653 call_reserve(struct rpc_task *task)
1655 task->tk_status = 0;
1656 task->tk_action = call_reserveresult;
1660 static void call_retry_reserve(struct rpc_task *task);
1663 * 1b. Grok the result of xprt_reserve()
1666 call_reserveresult(struct rpc_task *task)
1668 int status = task->tk_status;
1671 * After a call to xprt_reserve(), we must have either
1672 * a request slot or else an error status.
1674 task->tk_status = 0;
1676 if (task->tk_rqstp) {
1677 task->tk_action = call_refresh;
1681 rpc_call_rpcerror(task, -EIO);
1687 rpc_delay(task, HZ >> 2);
1689 case -EAGAIN: /* woken up; retry */
1690 task->tk_action = call_retry_reserve;
1693 rpc_call_rpcerror(task, status);
1698 * 1c. Retry reserving an RPC call slot
1701 call_retry_reserve(struct rpc_task *task)
1703 task->tk_status = 0;
1704 task->tk_action = call_reserveresult;
1705 xprt_retry_reserve(task);
1709 * 2. Bind and/or refresh the credentials
1712 call_refresh(struct rpc_task *task)
1714 task->tk_action = call_refreshresult;
1715 task->tk_status = 0;
1716 task->tk_client->cl_stats->rpcauthrefresh++;
1717 rpcauth_refreshcred(task);
1721 * 2a. Process the results of a credential refresh
1724 call_refreshresult(struct rpc_task *task)
1726 int status = task->tk_status;
1728 task->tk_status = 0;
1729 task->tk_action = call_refresh;
1732 if (rpcauth_uptodatecred(task)) {
1733 task->tk_action = call_allocate;
1736 /* Use rate-limiting and a max number of retries if refresh
1737 * had status 0 but failed to update the cred.
1741 rpc_delay(task, 3*HZ);
1747 if (!task->tk_cred_retry)
1749 task->tk_cred_retry--;
1750 trace_rpc_retry_refresh_status(task);
1753 trace_rpc_refresh_status(task);
1754 rpc_call_rpcerror(task, status);
1758 * 2b. Allocate the buffer. For details, see sched.c:rpc_malloc.
1759 * (Note: buffer memory is freed in xprt_release).
1762 call_allocate(struct rpc_task *task)
1764 const struct rpc_auth *auth = task->tk_rqstp->rq_cred->cr_auth;
1765 struct rpc_rqst *req = task->tk_rqstp;
1766 struct rpc_xprt *xprt = req->rq_xprt;
1767 const struct rpc_procinfo *proc = task->tk_msg.rpc_proc;
1770 task->tk_status = 0;
1771 task->tk_action = call_encode;
1776 if (proc->p_proc != 0) {
1777 BUG_ON(proc->p_arglen == 0);
1778 if (proc->p_decode != NULL)
1779 BUG_ON(proc->p_replen == 0);
1783 * Calculate the size (in quads) of the RPC call
1784 * and reply headers, and convert both values
1787 req->rq_callsize = RPC_CALLHDRSIZE + (auth->au_cslack << 1) +
1789 req->rq_callsize <<= 2;
1791 * Note: the reply buffer must at minimum allocate enough space
1792 * for the 'struct accepted_reply' from RFC5531.
1794 req->rq_rcvsize = RPC_REPHDRSIZE + auth->au_rslack + \
1795 max_t(size_t, proc->p_replen, 2);
1796 req->rq_rcvsize <<= 2;
1798 status = xprt->ops->buf_alloc(task);
1799 trace_rpc_buf_alloc(task, status);
1802 if (status != -ENOMEM) {
1803 rpc_call_rpcerror(task, status);
1807 if (RPC_IS_ASYNC(task) || !fatal_signal_pending(current)) {
1808 task->tk_action = call_allocate;
1809 rpc_delay(task, HZ>>4);
1813 rpc_call_rpcerror(task, -ERESTARTSYS);
1817 rpc_task_need_encode(struct rpc_task *task)
1819 return test_bit(RPC_TASK_NEED_XMIT, &task->tk_runstate) == 0 &&
1820 (!(task->tk_flags & RPC_TASK_SENT) ||
1821 !(task->tk_flags & RPC_TASK_NO_RETRANS_TIMEOUT) ||
1822 xprt_request_need_retransmit(task));
1826 rpc_xdr_encode(struct rpc_task *task)
1828 struct rpc_rqst *req = task->tk_rqstp;
1829 struct xdr_stream xdr;
1831 xdr_buf_init(&req->rq_snd_buf,
1834 xdr_buf_init(&req->rq_rcv_buf,
1838 req->rq_reply_bytes_recvd = 0;
1839 req->rq_snd_buf.head[0].iov_len = 0;
1840 xdr_init_encode(&xdr, &req->rq_snd_buf,
1841 req->rq_snd_buf.head[0].iov_base, req);
1842 xdr_free_bvec(&req->rq_snd_buf);
1843 if (rpc_encode_header(task, &xdr))
1846 task->tk_status = rpcauth_wrap_req(task, &xdr);
1850 * 3. Encode arguments of an RPC call
1853 call_encode(struct rpc_task *task)
1855 if (!rpc_task_need_encode(task))
1858 /* Dequeue task from the receive queue while we're encoding */
1859 xprt_request_dequeue_xprt(task);
1860 /* Encode here so that rpcsec_gss can use correct sequence number. */
1861 rpc_xdr_encode(task);
1862 /* Did the encode result in an error condition? */
1863 if (task->tk_status != 0) {
1864 /* Was the error nonfatal? */
1865 switch (task->tk_status) {
1868 rpc_delay(task, HZ >> 4);
1871 if (!task->tk_cred_retry) {
1872 rpc_exit(task, task->tk_status);
1874 task->tk_action = call_refresh;
1875 task->tk_cred_retry--;
1876 trace_rpc_retry_refresh_status(task);
1880 rpc_call_rpcerror(task, task->tk_status);
1885 /* Add task to reply queue before transmission to avoid races */
1886 if (rpc_reply_expected(task))
1887 xprt_request_enqueue_receive(task);
1888 xprt_request_enqueue_transmit(task);
1890 task->tk_action = call_transmit;
1891 /* Check that the connection is OK */
1892 if (!xprt_bound(task->tk_xprt))
1893 task->tk_action = call_bind;
1894 else if (!xprt_connected(task->tk_xprt))
1895 task->tk_action = call_connect;
1899 * Helpers to check if the task was already transmitted, and
1900 * to take action when that is the case.
1903 rpc_task_transmitted(struct rpc_task *task)
1905 return !test_bit(RPC_TASK_NEED_XMIT, &task->tk_runstate);
1909 rpc_task_handle_transmitted(struct rpc_task *task)
1911 xprt_end_transmit(task);
1912 task->tk_action = call_transmit_status;
1916 * 4. Get the server port number if not yet set
1919 call_bind(struct rpc_task *task)
1921 struct rpc_xprt *xprt = task->tk_rqstp->rq_xprt;
1923 if (rpc_task_transmitted(task)) {
1924 rpc_task_handle_transmitted(task);
1928 if (xprt_bound(xprt)) {
1929 task->tk_action = call_connect;
1933 task->tk_action = call_bind_status;
1934 if (!xprt_prepare_transmit(task))
1937 xprt->ops->rpcbind(task);
1941 * 4a. Sort out bind result
1944 call_bind_status(struct rpc_task *task)
1946 struct rpc_xprt *xprt = task->tk_rqstp->rq_xprt;
1949 if (rpc_task_transmitted(task)) {
1950 rpc_task_handle_transmitted(task);
1954 if (task->tk_status >= 0)
1956 if (xprt_bound(xprt)) {
1957 task->tk_status = 0;
1961 switch (task->tk_status) {
1963 rpc_delay(task, HZ >> 2);
1966 trace_rpcb_prog_unavail_err(task);
1967 /* fail immediately if this is an RPC ping */
1968 if (task->tk_msg.rpc_proc->p_proc == 0) {
1969 status = -EOPNOTSUPP;
1972 if (task->tk_rebind_retry == 0)
1974 task->tk_rebind_retry--;
1975 rpc_delay(task, 3*HZ);
1978 rpc_delay(task, HZ >> 2);
1983 trace_rpcb_timeout_err(task);
1986 /* server doesn't support any rpcbind version we know of */
1987 trace_rpcb_bind_version_err(task);
1989 case -EPROTONOSUPPORT:
1990 trace_rpcb_bind_version_err(task);
1992 case -ECONNREFUSED: /* connection problems */
2001 trace_rpcb_unreachable_err(task);
2002 if (!RPC_IS_SOFTCONN(task)) {
2003 rpc_delay(task, 5*HZ);
2006 status = task->tk_status;
2009 trace_rpcb_unrecognized_err(task);
2012 rpc_call_rpcerror(task, status);
2015 task->tk_action = call_connect;
2018 task->tk_status = 0;
2019 task->tk_action = call_bind;
2020 rpc_check_timeout(task);
2024 * 4b. Connect to the RPC server
2027 call_connect(struct rpc_task *task)
2029 struct rpc_xprt *xprt = task->tk_rqstp->rq_xprt;
2031 if (rpc_task_transmitted(task)) {
2032 rpc_task_handle_transmitted(task);
2036 if (xprt_connected(xprt)) {
2037 task->tk_action = call_transmit;
2041 task->tk_action = call_connect_status;
2042 if (task->tk_status < 0)
2044 if (task->tk_flags & RPC_TASK_NOCONNECT) {
2045 rpc_call_rpcerror(task, -ENOTCONN);
2048 if (!xprt_prepare_transmit(task))
2054 * 4c. Sort out connect result
2057 call_connect_status(struct rpc_task *task)
2059 struct rpc_xprt *xprt = task->tk_rqstp->rq_xprt;
2060 struct rpc_clnt *clnt = task->tk_client;
2061 int status = task->tk_status;
2063 if (rpc_task_transmitted(task)) {
2064 rpc_task_handle_transmitted(task);
2068 trace_rpc_connect_status(task);
2070 if (task->tk_status == 0) {
2071 clnt->cl_stats->netreconn++;
2074 if (xprt_connected(xprt)) {
2075 task->tk_status = 0;
2079 task->tk_status = 0;
2082 /* A positive refusal suggests a rebind is needed. */
2083 if (RPC_IS_SOFTCONN(task))
2085 if (clnt->cl_autobind) {
2086 rpc_force_rebind(clnt);
2097 xprt_conditional_disconnect(task->tk_rqstp->rq_xprt,
2098 task->tk_rqstp->rq_connect_cookie);
2099 if (RPC_IS_SOFTCONN(task))
2101 /* retry with existing socket, after a delay */
2102 rpc_delay(task, 3*HZ);
2110 rpc_delay(task, HZ >> 2);
2113 rpc_call_rpcerror(task, status);
2116 task->tk_action = call_transmit;
2119 /* Check for timeouts before looping back to call_bind */
2120 task->tk_action = call_bind;
2121 rpc_check_timeout(task);
2125 * 5. Transmit the RPC request, and wait for reply
2128 call_transmit(struct rpc_task *task)
2130 if (rpc_task_transmitted(task)) {
2131 rpc_task_handle_transmitted(task);
2135 task->tk_action = call_transmit_status;
2136 if (!xprt_prepare_transmit(task))
2138 task->tk_status = 0;
2139 if (test_bit(RPC_TASK_NEED_XMIT, &task->tk_runstate)) {
2140 if (!xprt_connected(task->tk_xprt)) {
2141 task->tk_status = -ENOTCONN;
2144 xprt_transmit(task);
2146 xprt_end_transmit(task);
2150 * 5a. Handle cleanup after a transmission
2153 call_transmit_status(struct rpc_task *task)
2155 task->tk_action = call_status;
2158 * Common case: success. Force the compiler to put this
2161 if (rpc_task_transmitted(task)) {
2162 task->tk_status = 0;
2163 xprt_request_wait_receive(task);
2167 switch (task->tk_status) {
2171 task->tk_status = 0;
2172 task->tk_action = call_encode;
2175 * Special cases: if we've been waiting on the
2176 * socket's write_space() callback, or if the
2177 * socket just returned a connection error,
2178 * then hold onto the transport lock.
2181 rpc_delay(task, HZ>>2);
2185 task->tk_action = call_transmit;
2186 task->tk_status = 0;
2194 if (RPC_IS_SOFTCONN(task)) {
2195 if (!task->tk_msg.rpc_proc->p_proc)
2196 trace_xprt_ping(task->tk_xprt,
2198 rpc_call_rpcerror(task, task->tk_status);
2207 task->tk_action = call_bind;
2208 task->tk_status = 0;
2211 rpc_check_timeout(task);
2214 #if defined(CONFIG_SUNRPC_BACKCHANNEL)
2215 static void call_bc_transmit(struct rpc_task *task);
2216 static void call_bc_transmit_status(struct rpc_task *task);
2219 call_bc_encode(struct rpc_task *task)
2221 xprt_request_enqueue_transmit(task);
2222 task->tk_action = call_bc_transmit;
2226 * 5b. Send the backchannel RPC reply. On error, drop the reply. In
2227 * addition, disconnect on connectivity errors.
2230 call_bc_transmit(struct rpc_task *task)
2232 task->tk_action = call_bc_transmit_status;
2233 if (test_bit(RPC_TASK_NEED_XMIT, &task->tk_runstate)) {
2234 if (!xprt_prepare_transmit(task))
2236 task->tk_status = 0;
2237 xprt_transmit(task);
2239 xprt_end_transmit(task);
2243 call_bc_transmit_status(struct rpc_task *task)
2245 struct rpc_rqst *req = task->tk_rqstp;
2247 if (rpc_task_transmitted(task))
2248 task->tk_status = 0;
2250 switch (task->tk_status) {
2264 rpc_delay(task, HZ>>2);
2268 task->tk_status = 0;
2269 task->tk_action = call_bc_transmit;
2273 * Problem reaching the server. Disconnect and let the
2274 * forechannel reestablish the connection. The server will
2275 * have to retransmit the backchannel request and we'll
2276 * reprocess it. Since these ops are idempotent, there's no
2277 * need to cache our reply at this time.
2279 printk(KERN_NOTICE "RPC: Could not send backchannel reply "
2280 "error: %d\n", task->tk_status);
2281 xprt_conditional_disconnect(req->rq_xprt,
2282 req->rq_connect_cookie);
2286 * We were unable to reply and will have to drop the
2287 * request. The server should reconnect and retransmit.
2289 printk(KERN_NOTICE "RPC: Could not send backchannel reply "
2290 "error: %d\n", task->tk_status);
2293 task->tk_action = rpc_exit_task;
2295 #endif /* CONFIG_SUNRPC_BACKCHANNEL */
2298 * 6. Sort out the RPC call status
2301 call_status(struct rpc_task *task)
2303 struct rpc_clnt *clnt = task->tk_client;
2306 if (!task->tk_msg.rpc_proc->p_proc)
2307 trace_xprt_ping(task->tk_xprt, task->tk_status);
2309 status = task->tk_status;
2311 task->tk_action = call_decode;
2315 trace_rpc_call_status(task);
2316 task->tk_status = 0;
2323 if (RPC_IS_SOFTCONN(task))
2326 * Delay any retries for 3 seconds, then handle as if it
2329 rpc_delay(task, 3*HZ);
2337 rpc_force_rebind(clnt);
2340 rpc_delay(task, 3*HZ);
2346 /* shutdown or soft timeout */
2349 if (clnt->cl_chatty)
2350 printk("%s: RPC call returned error %d\n",
2351 clnt->cl_program->name, -status);
2354 task->tk_action = call_encode;
2355 if (status != -ECONNRESET && status != -ECONNABORTED)
2356 rpc_check_timeout(task);
2359 rpc_call_rpcerror(task, status);
2363 rpc_check_connected(const struct rpc_rqst *req)
2365 /* No allocated request or transport? return true */
2366 if (!req || !req->rq_xprt)
2368 return xprt_connected(req->rq_xprt);
2372 rpc_check_timeout(struct rpc_task *task)
2374 struct rpc_clnt *clnt = task->tk_client;
2376 if (RPC_SIGNALLED(task)) {
2377 rpc_call_rpcerror(task, -ERESTARTSYS);
2381 if (xprt_adjust_timeout(task->tk_rqstp) == 0)
2384 trace_rpc_timeout_status(task);
2385 task->tk_timeouts++;
2387 if (RPC_IS_SOFTCONN(task) && !rpc_check_connected(task->tk_rqstp)) {
2388 rpc_call_rpcerror(task, -ETIMEDOUT);
2392 if (RPC_IS_SOFT(task)) {
2394 * Once a "no retrans timeout" soft tasks (a.k.a NFSv4) has
2395 * been sent, it should time out only if the transport
2396 * connection gets terminally broken.
2398 if ((task->tk_flags & RPC_TASK_NO_RETRANS_TIMEOUT) &&
2399 rpc_check_connected(task->tk_rqstp))
2402 if (clnt->cl_chatty) {
2403 pr_notice_ratelimited(
2404 "%s: server %s not responding, timed out\n",
2405 clnt->cl_program->name,
2406 task->tk_xprt->servername);
2408 if (task->tk_flags & RPC_TASK_TIMEOUT)
2409 rpc_call_rpcerror(task, -ETIMEDOUT);
2411 __rpc_call_rpcerror(task, -EIO, -ETIMEDOUT);
2415 if (!(task->tk_flags & RPC_CALL_MAJORSEEN)) {
2416 task->tk_flags |= RPC_CALL_MAJORSEEN;
2417 if (clnt->cl_chatty) {
2418 pr_notice_ratelimited(
2419 "%s: server %s not responding, still trying\n",
2420 clnt->cl_program->name,
2421 task->tk_xprt->servername);
2424 rpc_force_rebind(clnt);
2426 * Did our request time out due to an RPCSEC_GSS out-of-sequence
2427 * event? RFC2203 requires the server to drop all such requests.
2429 rpcauth_invalcred(task);
2433 * 7. Decode the RPC reply
2436 call_decode(struct rpc_task *task)
2438 struct rpc_clnt *clnt = task->tk_client;
2439 struct rpc_rqst *req = task->tk_rqstp;
2440 struct xdr_stream xdr;
2443 if (!task->tk_msg.rpc_proc->p_decode) {
2444 task->tk_action = rpc_exit_task;
2448 if (task->tk_flags & RPC_CALL_MAJORSEEN) {
2449 if (clnt->cl_chatty) {
2450 pr_notice_ratelimited("%s: server %s OK\n",
2451 clnt->cl_program->name,
2452 task->tk_xprt->servername);
2454 task->tk_flags &= ~RPC_CALL_MAJORSEEN;
2458 * Did we ever call xprt_complete_rqst()? If not, we should assume
2459 * the message is incomplete.
2462 if (!req->rq_reply_bytes_recvd)
2465 /* Ensure that we see all writes made by xprt_complete_rqst()
2466 * before it changed req->rq_reply_bytes_recvd.
2470 req->rq_rcv_buf.len = req->rq_private_buf.len;
2471 trace_rpc_xdr_recvfrom(task, &req->rq_rcv_buf);
2473 /* Check that the softirq receive buffer is valid */
2474 WARN_ON(memcmp(&req->rq_rcv_buf, &req->rq_private_buf,
2475 sizeof(req->rq_rcv_buf)) != 0);
2477 xdr_init_decode(&xdr, &req->rq_rcv_buf,
2478 req->rq_rcv_buf.head[0].iov_base, req);
2479 err = rpc_decode_header(task, &xdr);
2483 task->tk_action = rpc_exit_task;
2484 task->tk_status = rpcauth_unwrap_resp(task, &xdr);
2487 task->tk_status = 0;
2488 if (task->tk_client->cl_discrtry)
2489 xprt_conditional_disconnect(req->rq_xprt,
2490 req->rq_connect_cookie);
2491 task->tk_action = call_encode;
2492 rpc_check_timeout(task);
2495 task->tk_action = call_reserve;
2496 rpc_check_timeout(task);
2497 rpcauth_invalcred(task);
2498 /* Ensure we obtain a new XID if we retry! */
2504 rpc_encode_header(struct rpc_task *task, struct xdr_stream *xdr)
2506 struct rpc_clnt *clnt = task->tk_client;
2507 struct rpc_rqst *req = task->tk_rqstp;
2512 p = xdr_reserve_space(xdr, RPC_CALLHDRSIZE << 2);
2517 *p++ = cpu_to_be32(RPC_VERSION);
2518 *p++ = cpu_to_be32(clnt->cl_prog);
2519 *p++ = cpu_to_be32(clnt->cl_vers);
2520 *p = cpu_to_be32(task->tk_msg.rpc_proc->p_proc);
2522 error = rpcauth_marshcred(task, xdr);
2527 trace_rpc_bad_callhdr(task);
2528 rpc_call_rpcerror(task, error);
2533 rpc_decode_header(struct rpc_task *task, struct xdr_stream *xdr)
2535 struct rpc_clnt *clnt = task->tk_client;
2539 /* RFC-1014 says that the representation of XDR data must be a
2540 * multiple of four bytes
2541 * - if it isn't pointer subtraction in the NFS client may give
2544 if (task->tk_rqstp->rq_rcv_buf.len & 3)
2545 goto out_unparsable;
2547 p = xdr_inline_decode(xdr, 3 * sizeof(*p));
2549 goto out_unparsable;
2551 if (*p++ != rpc_reply)
2552 goto out_unparsable;
2553 if (*p++ != rpc_msg_accepted)
2554 goto out_msg_denied;
2556 error = rpcauth_checkverf(task, xdr);
2560 p = xdr_inline_decode(xdr, sizeof(*p));
2562 goto out_unparsable;
2566 case rpc_prog_unavail:
2567 trace_rpc__prog_unavail(task);
2568 error = -EPFNOSUPPORT;
2570 case rpc_prog_mismatch:
2571 trace_rpc__prog_mismatch(task);
2572 error = -EPROTONOSUPPORT;
2574 case rpc_proc_unavail:
2575 trace_rpc__proc_unavail(task);
2576 error = -EOPNOTSUPP;
2578 case rpc_garbage_args:
2579 case rpc_system_err:
2580 trace_rpc__garbage_args(task);
2584 goto out_unparsable;
2588 clnt->cl_stats->rpcgarbage++;
2589 if (task->tk_garb_retry) {
2590 task->tk_garb_retry--;
2591 task->tk_action = call_encode;
2595 rpc_call_rpcerror(task, error);
2599 trace_rpc__unparsable(task);
2604 trace_rpc_bad_verifier(task);
2609 p = xdr_inline_decode(xdr, sizeof(*p));
2611 goto out_unparsable;
2613 case rpc_auth_error:
2616 trace_rpc__mismatch(task);
2617 error = -EPROTONOSUPPORT;
2620 goto out_unparsable;
2623 p = xdr_inline_decode(xdr, sizeof(*p));
2625 goto out_unparsable;
2627 case rpc_autherr_rejectedcred:
2628 case rpc_autherr_rejectedverf:
2629 case rpcsec_gsserr_credproblem:
2630 case rpcsec_gsserr_ctxproblem:
2631 if (!task->tk_cred_retry)
2633 task->tk_cred_retry--;
2634 trace_rpc__stale_creds(task);
2635 return -EKEYREJECTED;
2636 case rpc_autherr_badcred:
2637 case rpc_autherr_badverf:
2638 /* possibly garbled cred/verf? */
2639 if (!task->tk_garb_retry)
2641 task->tk_garb_retry--;
2642 trace_rpc__bad_creds(task);
2643 task->tk_action = call_encode;
2645 case rpc_autherr_tooweak:
2646 trace_rpc__auth_tooweak(task);
2647 pr_warn("RPC: server %s requires stronger authentication.\n",
2648 task->tk_xprt->servername);
2651 goto out_unparsable;
2656 static void rpcproc_encode_null(struct rpc_rqst *rqstp, struct xdr_stream *xdr,
2661 static int rpcproc_decode_null(struct rpc_rqst *rqstp, struct xdr_stream *xdr,
2667 static const struct rpc_procinfo rpcproc_null = {
2668 .p_encode = rpcproc_encode_null,
2669 .p_decode = rpcproc_decode_null,
2672 static int rpc_ping(struct rpc_clnt *clnt)
2674 struct rpc_message msg = {
2675 .rpc_proc = &rpcproc_null,
2678 err = rpc_call_sync(clnt, &msg, RPC_TASK_SOFT | RPC_TASK_SOFTCONN |
2679 RPC_TASK_NULLCREDS);
2684 struct rpc_task *rpc_call_null_helper(struct rpc_clnt *clnt,
2685 struct rpc_xprt *xprt, struct rpc_cred *cred, int flags,
2686 const struct rpc_call_ops *ops, void *data)
2688 struct rpc_message msg = {
2689 .rpc_proc = &rpcproc_null,
2691 struct rpc_task_setup task_setup_data = {
2694 .rpc_message = &msg,
2695 .rpc_op_cred = cred,
2696 .callback_ops = (ops != NULL) ? ops : &rpc_default_ops,
2697 .callback_data = data,
2698 .flags = flags | RPC_TASK_SOFT | RPC_TASK_SOFTCONN |
2702 return rpc_run_task(&task_setup_data);
2705 struct rpc_task *rpc_call_null(struct rpc_clnt *clnt, struct rpc_cred *cred, int flags)
2707 return rpc_call_null_helper(clnt, NULL, cred, flags, NULL, NULL);
2709 EXPORT_SYMBOL_GPL(rpc_call_null);
2711 struct rpc_cb_add_xprt_calldata {
2712 struct rpc_xprt_switch *xps;
2713 struct rpc_xprt *xprt;
2716 static void rpc_cb_add_xprt_done(struct rpc_task *task, void *calldata)
2718 struct rpc_cb_add_xprt_calldata *data = calldata;
2720 if (task->tk_status == 0)
2721 rpc_xprt_switch_add_xprt(data->xps, data->xprt);
2724 static void rpc_cb_add_xprt_release(void *calldata)
2726 struct rpc_cb_add_xprt_calldata *data = calldata;
2728 xprt_put(data->xprt);
2729 xprt_switch_put(data->xps);
2733 static const struct rpc_call_ops rpc_cb_add_xprt_call_ops = {
2734 .rpc_call_done = rpc_cb_add_xprt_done,
2735 .rpc_release = rpc_cb_add_xprt_release,
2739 * rpc_clnt_test_and_add_xprt - Test and add a new transport to a rpc_clnt
2740 * @clnt: pointer to struct rpc_clnt
2741 * @xps: pointer to struct rpc_xprt_switch,
2742 * @xprt: pointer struct rpc_xprt
2745 int rpc_clnt_test_and_add_xprt(struct rpc_clnt *clnt,
2746 struct rpc_xprt_switch *xps, struct rpc_xprt *xprt,
2749 struct rpc_cb_add_xprt_calldata *data;
2750 struct rpc_task *task;
2752 data = kmalloc(sizeof(*data), GFP_NOFS);
2755 data->xps = xprt_switch_get(xps);
2756 data->xprt = xprt_get(xprt);
2757 if (rpc_xprt_switch_has_addr(data->xps, (struct sockaddr *)&xprt->addr)) {
2758 rpc_cb_add_xprt_release(data);
2762 task = rpc_call_null_helper(clnt, xprt, NULL, RPC_TASK_ASYNC,
2763 &rpc_cb_add_xprt_call_ops, data);
2769 EXPORT_SYMBOL_GPL(rpc_clnt_test_and_add_xprt);
2772 * rpc_clnt_setup_test_and_add_xprt()
2774 * This is an rpc_clnt_add_xprt setup() function which returns 1 so:
2775 * 1) caller of the test function must dereference the rpc_xprt_switch
2777 * 2) test function must call rpc_xprt_switch_add_xprt, usually in
2778 * the rpc_call_done routine.
2780 * Upon success (return of 1), the test function adds the new
2781 * transport to the rpc_clnt xprt switch
2783 * @clnt: struct rpc_clnt to get the new transport
2784 * @xps: the rpc_xprt_switch to hold the new transport
2785 * @xprt: the rpc_xprt to test
2786 * @data: a struct rpc_add_xprt_test pointer that holds the test function
2787 * and test function call data
2789 int rpc_clnt_setup_test_and_add_xprt(struct rpc_clnt *clnt,
2790 struct rpc_xprt_switch *xps,
2791 struct rpc_xprt *xprt,
2794 struct rpc_task *task;
2795 struct rpc_add_xprt_test *xtest = (struct rpc_add_xprt_test *)data;
2796 int status = -EADDRINUSE;
2798 xprt = xprt_get(xprt);
2799 xprt_switch_get(xps);
2801 if (rpc_xprt_switch_has_addr(xps, (struct sockaddr *)&xprt->addr))
2804 /* Test the connection */
2805 task = rpc_call_null_helper(clnt, xprt, NULL, 0, NULL, NULL);
2807 status = PTR_ERR(task);
2810 status = task->tk_status;
2816 /* rpc_xprt_switch and rpc_xprt are deferrenced by add_xprt_test() */
2817 xtest->add_xprt_test(clnt, xprt, xtest->data);
2820 xprt_switch_put(xps);
2822 /* so that rpc_clnt_add_xprt does not call rpc_xprt_switch_add_xprt */
2826 xprt_switch_put(xps);
2827 pr_info("RPC: rpc_clnt_test_xprt failed: %d addr %s not added\n",
2828 status, xprt->address_strings[RPC_DISPLAY_ADDR]);
2831 EXPORT_SYMBOL_GPL(rpc_clnt_setup_test_and_add_xprt);
2834 * rpc_clnt_add_xprt - Add a new transport to a rpc_clnt
2835 * @clnt: pointer to struct rpc_clnt
2836 * @xprtargs: pointer to struct xprt_create
2837 * @setup: callback to test and/or set up the connection
2838 * @data: pointer to setup function data
2840 * Creates a new transport using the parameters set in args and
2842 * If ping is set, then test that connectivity succeeds before
2843 * adding the new transport.
2846 int rpc_clnt_add_xprt(struct rpc_clnt *clnt,
2847 struct xprt_create *xprtargs,
2848 int (*setup)(struct rpc_clnt *,
2849 struct rpc_xprt_switch *,
2854 struct rpc_xprt_switch *xps;
2855 struct rpc_xprt *xprt;
2856 unsigned long connect_timeout;
2857 unsigned long reconnect_timeout;
2858 unsigned char resvport, reuseport;
2862 xps = xprt_switch_get(rcu_dereference(clnt->cl_xpi.xpi_xpswitch));
2863 xprt = xprt_iter_xprt(&clnt->cl_xpi);
2864 if (xps == NULL || xprt == NULL) {
2866 xprt_switch_put(xps);
2869 resvport = xprt->resvport;
2870 reuseport = xprt->reuseport;
2871 connect_timeout = xprt->connect_timeout;
2872 reconnect_timeout = xprt->max_reconnect_timeout;
2875 xprt = xprt_create_transport(xprtargs);
2877 ret = PTR_ERR(xprt);
2878 goto out_put_switch;
2880 xprt->resvport = resvport;
2881 xprt->reuseport = reuseport;
2882 if (xprt->ops->set_connect_timeout != NULL)
2883 xprt->ops->set_connect_timeout(xprt,
2887 rpc_xprt_switch_set_roundrobin(xps);
2889 ret = setup(clnt, xps, xprt, data);
2893 rpc_xprt_switch_add_xprt(xps, xprt);
2897 xprt_switch_put(xps);
2900 EXPORT_SYMBOL_GPL(rpc_clnt_add_xprt);
2902 struct connect_timeout_data {
2903 unsigned long connect_timeout;
2904 unsigned long reconnect_timeout;
2908 rpc_xprt_set_connect_timeout(struct rpc_clnt *clnt,
2909 struct rpc_xprt *xprt,
2912 struct connect_timeout_data *timeo = data;
2914 if (xprt->ops->set_connect_timeout)
2915 xprt->ops->set_connect_timeout(xprt,
2916 timeo->connect_timeout,
2917 timeo->reconnect_timeout);
2922 rpc_set_connect_timeout(struct rpc_clnt *clnt,
2923 unsigned long connect_timeout,
2924 unsigned long reconnect_timeout)
2926 struct connect_timeout_data timeout = {
2927 .connect_timeout = connect_timeout,
2928 .reconnect_timeout = reconnect_timeout,
2930 rpc_clnt_iterate_for_each_xprt(clnt,
2931 rpc_xprt_set_connect_timeout,
2934 EXPORT_SYMBOL_GPL(rpc_set_connect_timeout);
2936 void rpc_clnt_xprt_switch_put(struct rpc_clnt *clnt)
2939 xprt_switch_put(rcu_dereference(clnt->cl_xpi.xpi_xpswitch));
2942 EXPORT_SYMBOL_GPL(rpc_clnt_xprt_switch_put);
2944 void rpc_clnt_xprt_switch_add_xprt(struct rpc_clnt *clnt, struct rpc_xprt *xprt)
2947 rpc_xprt_switch_add_xprt(rcu_dereference(clnt->cl_xpi.xpi_xpswitch),
2951 EXPORT_SYMBOL_GPL(rpc_clnt_xprt_switch_add_xprt);
2953 bool rpc_clnt_xprt_switch_has_addr(struct rpc_clnt *clnt,
2954 const struct sockaddr *sap)
2956 struct rpc_xprt_switch *xps;
2960 xps = rcu_dereference(clnt->cl_xpi.xpi_xpswitch);
2961 ret = rpc_xprt_switch_has_addr(xps, sap);
2965 EXPORT_SYMBOL_GPL(rpc_clnt_xprt_switch_has_addr);
2967 #if IS_ENABLED(CONFIG_SUNRPC_DEBUG)
2968 static void rpc_show_header(void)
2970 printk(KERN_INFO "-pid- flgs status -client- --rqstp- "
2971 "-timeout ---ops--\n");
2974 static void rpc_show_task(const struct rpc_clnt *clnt,
2975 const struct rpc_task *task)
2977 const char *rpc_waitq = "none";
2979 if (RPC_IS_QUEUED(task))
2980 rpc_waitq = rpc_qname(task->tk_waitqueue);
2982 printk(KERN_INFO "%5u %04x %6d %8p %8p %8ld %8p %sv%u %s a:%ps q:%s\n",
2983 task->tk_pid, task->tk_flags, task->tk_status,
2984 clnt, task->tk_rqstp, rpc_task_timeout(task), task->tk_ops,
2985 clnt->cl_program->name, clnt->cl_vers, rpc_proc_name(task),
2986 task->tk_action, rpc_waitq);
2989 void rpc_show_tasks(struct net *net)
2991 struct rpc_clnt *clnt;
2992 struct rpc_task *task;
2994 struct sunrpc_net *sn = net_generic(net, sunrpc_net_id);
2996 spin_lock(&sn->rpc_client_lock);
2997 list_for_each_entry(clnt, &sn->all_clients, cl_clients) {
2998 spin_lock(&clnt->cl_lock);
2999 list_for_each_entry(task, &clnt->cl_tasks, tk_task) {
3004 rpc_show_task(clnt, task);
3006 spin_unlock(&clnt->cl_lock);
3008 spin_unlock(&sn->rpc_client_lock);
3012 #if IS_ENABLED(CONFIG_SUNRPC_SWAP)
3014 rpc_clnt_swap_activate_callback(struct rpc_clnt *clnt,
3015 struct rpc_xprt *xprt,
3018 return xprt_enable_swap(xprt);
3022 rpc_clnt_swap_activate(struct rpc_clnt *clnt)
3024 if (atomic_inc_return(&clnt->cl_swapper) == 1)
3025 return rpc_clnt_iterate_for_each_xprt(clnt,
3026 rpc_clnt_swap_activate_callback, NULL);
3029 EXPORT_SYMBOL_GPL(rpc_clnt_swap_activate);
3032 rpc_clnt_swap_deactivate_callback(struct rpc_clnt *clnt,
3033 struct rpc_xprt *xprt,
3036 xprt_disable_swap(xprt);
3041 rpc_clnt_swap_deactivate(struct rpc_clnt *clnt)
3043 if (atomic_dec_if_positive(&clnt->cl_swapper) == 0)
3044 rpc_clnt_iterate_for_each_xprt(clnt,
3045 rpc_clnt_swap_deactivate_callback, NULL);
3047 EXPORT_SYMBOL_GPL(rpc_clnt_swap_deactivate);
3048 #endif /* CONFIG_SUNRPC_SWAP */