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