Merge tag 'split-asm_system_h-for-linus-20120328' of git://git.kernel.org/pub/scm...
[linux-2.6.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/utsname.h>
29 #include <linux/workqueue.h>
30 #include <linux/in.h>
31 #include <linux/in6.h>
32 #include <linux/un.h>
33 #include <linux/rcupdate.h>
34
35 #include <linux/sunrpc/clnt.h>
36 #include <linux/sunrpc/rpc_pipe_fs.h>
37 #include <linux/sunrpc/metrics.h>
38 #include <linux/sunrpc/bc_xprt.h>
39 #include <trace/events/sunrpc.h>
40
41 #include "sunrpc.h"
42 #include "netns.h"
43
44 #ifdef RPC_DEBUG
45 # define RPCDBG_FACILITY        RPCDBG_CALL
46 #endif
47
48 #define dprint_status(t)                                        \
49         dprintk("RPC: %5u %s (status %d)\n", t->tk_pid,         \
50                         __func__, t->tk_status)
51
52 /*
53  * All RPC clients are linked into this list
54  */
55
56 static DECLARE_WAIT_QUEUE_HEAD(destroy_wait);
57
58
59 static void     call_start(struct rpc_task *task);
60 static void     call_reserve(struct rpc_task *task);
61 static void     call_reserveresult(struct rpc_task *task);
62 static void     call_allocate(struct rpc_task *task);
63 static void     call_decode(struct rpc_task *task);
64 static void     call_bind(struct rpc_task *task);
65 static void     call_bind_status(struct rpc_task *task);
66 static void     call_transmit(struct rpc_task *task);
67 #if defined(CONFIG_SUNRPC_BACKCHANNEL)
68 static void     call_bc_transmit(struct rpc_task *task);
69 #endif /* CONFIG_SUNRPC_BACKCHANNEL */
70 static void     call_status(struct rpc_task *task);
71 static void     call_transmit_status(struct rpc_task *task);
72 static void     call_refresh(struct rpc_task *task);
73 static void     call_refreshresult(struct rpc_task *task);
74 static void     call_timeout(struct rpc_task *task);
75 static void     call_connect(struct rpc_task *task);
76 static void     call_connect_status(struct rpc_task *task);
77
78 static __be32   *rpc_encode_header(struct rpc_task *task);
79 static __be32   *rpc_verify_header(struct rpc_task *task);
80 static int      rpc_ping(struct rpc_clnt *clnt);
81
82 static void rpc_register_client(struct rpc_clnt *clnt)
83 {
84         struct net *net = rpc_net_ns(clnt);
85         struct sunrpc_net *sn = net_generic(net, sunrpc_net_id);
86
87         spin_lock(&sn->rpc_client_lock);
88         list_add(&clnt->cl_clients, &sn->all_clients);
89         spin_unlock(&sn->rpc_client_lock);
90 }
91
92 static void rpc_unregister_client(struct rpc_clnt *clnt)
93 {
94         struct net *net = rpc_net_ns(clnt);
95         struct sunrpc_net *sn = net_generic(net, sunrpc_net_id);
96
97         spin_lock(&sn->rpc_client_lock);
98         list_del(&clnt->cl_clients);
99         spin_unlock(&sn->rpc_client_lock);
100 }
101
102 static void __rpc_clnt_remove_pipedir(struct rpc_clnt *clnt)
103 {
104         if (clnt->cl_dentry) {
105                 if (clnt->cl_auth && clnt->cl_auth->au_ops->pipes_destroy)
106                         clnt->cl_auth->au_ops->pipes_destroy(clnt->cl_auth);
107                 rpc_remove_client_dir(clnt->cl_dentry);
108         }
109         clnt->cl_dentry = NULL;
110 }
111
112 static void rpc_clnt_remove_pipedir(struct rpc_clnt *clnt)
113 {
114         struct net *net = rpc_net_ns(clnt);
115         struct super_block *pipefs_sb;
116
117         pipefs_sb = rpc_get_sb_net(net);
118         if (pipefs_sb) {
119                 __rpc_clnt_remove_pipedir(clnt);
120                 rpc_put_sb_net(net);
121         }
122 }
123
124 static struct dentry *rpc_setup_pipedir_sb(struct super_block *sb,
125                                     struct rpc_clnt *clnt,
126                                     const char *dir_name)
127 {
128         static uint32_t clntid;
129         char name[15];
130         struct qstr q = {
131                 .name = name,
132         };
133         struct dentry *dir, *dentry;
134         int error;
135
136         dir = rpc_d_lookup_sb(sb, dir_name);
137         if (dir == NULL)
138                 return dir;
139         for (;;) {
140                 q.len = snprintf(name, sizeof(name), "clnt%x", (unsigned int)clntid++);
141                 name[sizeof(name) - 1] = '\0';
142                 q.hash = full_name_hash(q.name, q.len);
143                 dentry = rpc_create_client_dir(dir, &q, clnt);
144                 if (!IS_ERR(dentry))
145                         break;
146                 error = PTR_ERR(dentry);
147                 if (error != -EEXIST) {
148                         printk(KERN_INFO "RPC: Couldn't create pipefs entry"
149                                         " %s/%s, error %d\n",
150                                         dir_name, name, error);
151                         break;
152                 }
153         }
154         dput(dir);
155         return dentry;
156 }
157
158 static int
159 rpc_setup_pipedir(struct rpc_clnt *clnt, const char *dir_name)
160 {
161         struct net *net = rpc_net_ns(clnt);
162         struct super_block *pipefs_sb;
163         struct dentry *dentry;
164
165         clnt->cl_dentry = NULL;
166         if (dir_name == NULL)
167                 return 0;
168         pipefs_sb = rpc_get_sb_net(net);
169         if (!pipefs_sb)
170                 return 0;
171         dentry = rpc_setup_pipedir_sb(pipefs_sb, clnt, dir_name);
172         rpc_put_sb_net(net);
173         if (IS_ERR(dentry))
174                 return PTR_ERR(dentry);
175         clnt->cl_dentry = dentry;
176         return 0;
177 }
178
179 static int __rpc_pipefs_event(struct rpc_clnt *clnt, unsigned long event,
180                                 struct super_block *sb)
181 {
182         struct dentry *dentry;
183         int err = 0;
184
185         switch (event) {
186         case RPC_PIPEFS_MOUNT:
187                 if (clnt->cl_program->pipe_dir_name == NULL)
188                         break;
189                 dentry = rpc_setup_pipedir_sb(sb, clnt,
190                                               clnt->cl_program->pipe_dir_name);
191                 BUG_ON(dentry == NULL);
192                 if (IS_ERR(dentry))
193                         return PTR_ERR(dentry);
194                 clnt->cl_dentry = dentry;
195                 if (clnt->cl_auth->au_ops->pipes_create) {
196                         err = clnt->cl_auth->au_ops->pipes_create(clnt->cl_auth);
197                         if (err)
198                                 __rpc_clnt_remove_pipedir(clnt);
199                 }
200                 break;
201         case RPC_PIPEFS_UMOUNT:
202                 __rpc_clnt_remove_pipedir(clnt);
203                 break;
204         default:
205                 printk(KERN_ERR "%s: unknown event: %ld\n", __func__, event);
206                 return -ENOTSUPP;
207         }
208         return err;
209 }
210
211 static struct rpc_clnt *rpc_get_client_for_event(struct net *net, int event)
212 {
213         struct sunrpc_net *sn = net_generic(net, sunrpc_net_id);
214         struct rpc_clnt *clnt;
215
216         spin_lock(&sn->rpc_client_lock);
217         list_for_each_entry(clnt, &sn->all_clients, cl_clients) {
218                 if (((event == RPC_PIPEFS_MOUNT) && clnt->cl_dentry) ||
219                     ((event == RPC_PIPEFS_UMOUNT) && !clnt->cl_dentry))
220                         continue;
221                 atomic_inc(&clnt->cl_count);
222                 spin_unlock(&sn->rpc_client_lock);
223                 return clnt;
224         }
225         spin_unlock(&sn->rpc_client_lock);
226         return NULL;
227 }
228
229 static int rpc_pipefs_event(struct notifier_block *nb, unsigned long event,
230                             void *ptr)
231 {
232         struct super_block *sb = ptr;
233         struct rpc_clnt *clnt;
234         int error = 0;
235
236         while ((clnt = rpc_get_client_for_event(sb->s_fs_info, event))) {
237                 error = __rpc_pipefs_event(clnt, event, sb);
238                 rpc_release_client(clnt);
239                 if (error)
240                         break;
241         }
242         return error;
243 }
244
245 static struct notifier_block rpc_clients_block = {
246         .notifier_call  = rpc_pipefs_event,
247         .priority       = SUNRPC_PIPEFS_RPC_PRIO,
248 };
249
250 int rpc_clients_notifier_register(void)
251 {
252         return rpc_pipefs_notifier_register(&rpc_clients_block);
253 }
254
255 void rpc_clients_notifier_unregister(void)
256 {
257         return rpc_pipefs_notifier_unregister(&rpc_clients_block);
258 }
259
260 static struct rpc_clnt * rpc_new_client(const struct rpc_create_args *args, struct rpc_xprt *xprt)
261 {
262         const struct rpc_program *program = args->program;
263         const struct rpc_version *version;
264         struct rpc_clnt         *clnt = NULL;
265         struct rpc_auth         *auth;
266         int err;
267
268         /* sanity check the name before trying to print it */
269         dprintk("RPC:       creating %s client for %s (xprt %p)\n",
270                         program->name, args->servername, xprt);
271
272         err = rpciod_up();
273         if (err)
274                 goto out_no_rpciod;
275         err = -EINVAL;
276         if (!xprt)
277                 goto out_no_xprt;
278
279         if (args->version >= program->nrvers)
280                 goto out_err;
281         version = program->version[args->version];
282         if (version == NULL)
283                 goto out_err;
284
285         err = -ENOMEM;
286         clnt = kzalloc(sizeof(*clnt), GFP_KERNEL);
287         if (!clnt)
288                 goto out_err;
289         clnt->cl_parent = clnt;
290
291         rcu_assign_pointer(clnt->cl_xprt, xprt);
292         clnt->cl_procinfo = version->procs;
293         clnt->cl_maxproc  = version->nrprocs;
294         clnt->cl_protname = program->name;
295         clnt->cl_prog     = args->prognumber ? : program->number;
296         clnt->cl_vers     = version->number;
297         clnt->cl_stats    = program->stats;
298         clnt->cl_metrics  = rpc_alloc_iostats(clnt);
299         err = -ENOMEM;
300         if (clnt->cl_metrics == NULL)
301                 goto out_no_stats;
302         clnt->cl_program  = program;
303         INIT_LIST_HEAD(&clnt->cl_tasks);
304         spin_lock_init(&clnt->cl_lock);
305
306         if (!xprt_bound(xprt))
307                 clnt->cl_autobind = 1;
308
309         clnt->cl_timeout = xprt->timeout;
310         if (args->timeout != NULL) {
311                 memcpy(&clnt->cl_timeout_default, args->timeout,
312                                 sizeof(clnt->cl_timeout_default));
313                 clnt->cl_timeout = &clnt->cl_timeout_default;
314         }
315
316         clnt->cl_rtt = &clnt->cl_rtt_default;
317         rpc_init_rtt(&clnt->cl_rtt_default, clnt->cl_timeout->to_initval);
318         clnt->cl_principal = NULL;
319         if (args->client_name) {
320                 clnt->cl_principal = kstrdup(args->client_name, GFP_KERNEL);
321                 if (!clnt->cl_principal)
322                         goto out_no_principal;
323         }
324
325         atomic_set(&clnt->cl_count, 1);
326
327         err = rpc_setup_pipedir(clnt, program->pipe_dir_name);
328         if (err < 0)
329                 goto out_no_path;
330
331         auth = rpcauth_create(args->authflavor, clnt);
332         if (IS_ERR(auth)) {
333                 printk(KERN_INFO "RPC: Couldn't create auth handle (flavor %u)\n",
334                                 args->authflavor);
335                 err = PTR_ERR(auth);
336                 goto out_no_auth;
337         }
338
339         /* save the nodename */
340         clnt->cl_nodelen = strlen(init_utsname()->nodename);
341         if (clnt->cl_nodelen > UNX_MAXNODENAME)
342                 clnt->cl_nodelen = UNX_MAXNODENAME;
343         memcpy(clnt->cl_nodename, init_utsname()->nodename, clnt->cl_nodelen);
344         rpc_register_client(clnt);
345         return clnt;
346
347 out_no_auth:
348         rpc_clnt_remove_pipedir(clnt);
349 out_no_path:
350         kfree(clnt->cl_principal);
351 out_no_principal:
352         rpc_free_iostats(clnt->cl_metrics);
353 out_no_stats:
354         kfree(clnt);
355 out_err:
356         xprt_put(xprt);
357 out_no_xprt:
358         rpciod_down();
359 out_no_rpciod:
360         return ERR_PTR(err);
361 }
362
363 /*
364  * rpc_create - create an RPC client and transport with one call
365  * @args: rpc_clnt create argument structure
366  *
367  * Creates and initializes an RPC transport and an RPC client.
368  *
369  * It can ping the server in order to determine if it is up, and to see if
370  * it supports this program and version.  RPC_CLNT_CREATE_NOPING disables
371  * this behavior so asynchronous tasks can also use rpc_create.
372  */
373 struct rpc_clnt *rpc_create(struct rpc_create_args *args)
374 {
375         struct rpc_xprt *xprt;
376         struct rpc_clnt *clnt;
377         struct xprt_create xprtargs = {
378                 .net = args->net,
379                 .ident = args->protocol,
380                 .srcaddr = args->saddress,
381                 .dstaddr = args->address,
382                 .addrlen = args->addrsize,
383                 .servername = args->servername,
384                 .bc_xprt = args->bc_xprt,
385         };
386         char servername[48];
387
388         /*
389          * If the caller chooses not to specify a hostname, whip
390          * up a string representation of the passed-in address.
391          */
392         if (xprtargs.servername == NULL) {
393                 struct sockaddr_un *sun =
394                                 (struct sockaddr_un *)args->address;
395                 struct sockaddr_in *sin =
396                                 (struct sockaddr_in *)args->address;
397                 struct sockaddr_in6 *sin6 =
398                                 (struct sockaddr_in6 *)args->address;
399
400                 servername[0] = '\0';
401                 switch (args->address->sa_family) {
402                 case AF_LOCAL:
403                         snprintf(servername, sizeof(servername), "%s",
404                                  sun->sun_path);
405                         break;
406                 case AF_INET:
407                         snprintf(servername, sizeof(servername), "%pI4",
408                                  &sin->sin_addr.s_addr);
409                         break;
410                 case AF_INET6:
411                         snprintf(servername, sizeof(servername), "%pI6",
412                                  &sin6->sin6_addr);
413                         break;
414                 default:
415                         /* caller wants default server name, but
416                          * address family isn't recognized. */
417                         return ERR_PTR(-EINVAL);
418                 }
419                 xprtargs.servername = servername;
420         }
421
422         xprt = xprt_create_transport(&xprtargs);
423         if (IS_ERR(xprt))
424                 return (struct rpc_clnt *)xprt;
425
426         /*
427          * By default, kernel RPC client connects from a reserved port.
428          * CAP_NET_BIND_SERVICE will not be set for unprivileged requesters,
429          * but it is always enabled for rpciod, which handles the connect
430          * operation.
431          */
432         xprt->resvport = 1;
433         if (args->flags & RPC_CLNT_CREATE_NONPRIVPORT)
434                 xprt->resvport = 0;
435
436         clnt = rpc_new_client(args, xprt);
437         if (IS_ERR(clnt))
438                 return clnt;
439
440         if (!(args->flags & RPC_CLNT_CREATE_NOPING)) {
441                 int err = rpc_ping(clnt);
442                 if (err != 0) {
443                         rpc_shutdown_client(clnt);
444                         return ERR_PTR(err);
445                 }
446         }
447
448         clnt->cl_softrtry = 1;
449         if (args->flags & RPC_CLNT_CREATE_HARDRTRY)
450                 clnt->cl_softrtry = 0;
451
452         if (args->flags & RPC_CLNT_CREATE_AUTOBIND)
453                 clnt->cl_autobind = 1;
454         if (args->flags & RPC_CLNT_CREATE_DISCRTRY)
455                 clnt->cl_discrtry = 1;
456         if (!(args->flags & RPC_CLNT_CREATE_QUIET))
457                 clnt->cl_chatty = 1;
458
459         return clnt;
460 }
461 EXPORT_SYMBOL_GPL(rpc_create);
462
463 /*
464  * This function clones the RPC client structure. It allows us to share the
465  * same transport while varying parameters such as the authentication
466  * flavour.
467  */
468 struct rpc_clnt *
469 rpc_clone_client(struct rpc_clnt *clnt)
470 {
471         struct rpc_clnt *new;
472         struct rpc_xprt *xprt;
473         int err = -ENOMEM;
474
475         new = kmemdup(clnt, sizeof(*new), GFP_KERNEL);
476         if (!new)
477                 goto out_no_clnt;
478         new->cl_parent = clnt;
479         /* Turn off autobind on clones */
480         new->cl_autobind = 0;
481         INIT_LIST_HEAD(&new->cl_tasks);
482         spin_lock_init(&new->cl_lock);
483         rpc_init_rtt(&new->cl_rtt_default, clnt->cl_timeout->to_initval);
484         new->cl_metrics = rpc_alloc_iostats(clnt);
485         if (new->cl_metrics == NULL)
486                 goto out_no_stats;
487         if (clnt->cl_principal) {
488                 new->cl_principal = kstrdup(clnt->cl_principal, GFP_KERNEL);
489                 if (new->cl_principal == NULL)
490                         goto out_no_principal;
491         }
492         rcu_read_lock();
493         xprt = xprt_get(rcu_dereference(clnt->cl_xprt));
494         rcu_read_unlock();
495         if (xprt == NULL)
496                 goto out_no_transport;
497         rcu_assign_pointer(new->cl_xprt, xprt);
498         atomic_set(&new->cl_count, 1);
499         err = rpc_setup_pipedir(new, clnt->cl_program->pipe_dir_name);
500         if (err != 0)
501                 goto out_no_path;
502         if (new->cl_auth)
503                 atomic_inc(&new->cl_auth->au_count);
504         atomic_inc(&clnt->cl_count);
505         rpc_register_client(new);
506         rpciod_up();
507         return new;
508 out_no_path:
509         xprt_put(xprt);
510 out_no_transport:
511         kfree(new->cl_principal);
512 out_no_principal:
513         rpc_free_iostats(new->cl_metrics);
514 out_no_stats:
515         kfree(new);
516 out_no_clnt:
517         dprintk("RPC:       %s: returned error %d\n", __func__, err);
518         return ERR_PTR(err);
519 }
520 EXPORT_SYMBOL_GPL(rpc_clone_client);
521
522 /*
523  * Kill all tasks for the given client.
524  * XXX: kill their descendants as well?
525  */
526 void rpc_killall_tasks(struct rpc_clnt *clnt)
527 {
528         struct rpc_task *rovr;
529
530
531         if (list_empty(&clnt->cl_tasks))
532                 return;
533         dprintk("RPC:       killing all tasks for client %p\n", clnt);
534         /*
535          * Spin lock all_tasks to prevent changes...
536          */
537         spin_lock(&clnt->cl_lock);
538         list_for_each_entry(rovr, &clnt->cl_tasks, tk_task) {
539                 if (!RPC_IS_ACTIVATED(rovr))
540                         continue;
541                 if (!(rovr->tk_flags & RPC_TASK_KILLED)) {
542                         rovr->tk_flags |= RPC_TASK_KILLED;
543                         rpc_exit(rovr, -EIO);
544                         if (RPC_IS_QUEUED(rovr))
545                                 rpc_wake_up_queued_task(rovr->tk_waitqueue,
546                                                         rovr);
547                 }
548         }
549         spin_unlock(&clnt->cl_lock);
550 }
551 EXPORT_SYMBOL_GPL(rpc_killall_tasks);
552
553 /*
554  * Properly shut down an RPC client, terminating all outstanding
555  * requests.
556  */
557 void rpc_shutdown_client(struct rpc_clnt *clnt)
558 {
559         dprintk_rcu("RPC:       shutting down %s client for %s\n",
560                         clnt->cl_protname,
561                         rcu_dereference(clnt->cl_xprt)->servername);
562
563         while (!list_empty(&clnt->cl_tasks)) {
564                 rpc_killall_tasks(clnt);
565                 wait_event_timeout(destroy_wait,
566                         list_empty(&clnt->cl_tasks), 1*HZ);
567         }
568
569         rpc_release_client(clnt);
570 }
571 EXPORT_SYMBOL_GPL(rpc_shutdown_client);
572
573 /*
574  * Free an RPC client
575  */
576 static void
577 rpc_free_client(struct rpc_clnt *clnt)
578 {
579         dprintk_rcu("RPC:       destroying %s client for %s\n",
580                         clnt->cl_protname,
581                         rcu_dereference(clnt->cl_xprt)->servername);
582         if (clnt->cl_parent != clnt)
583                 rpc_release_client(clnt->cl_parent);
584         rpc_unregister_client(clnt);
585         rpc_clnt_remove_pipedir(clnt);
586         rpc_free_iostats(clnt->cl_metrics);
587         kfree(clnt->cl_principal);
588         clnt->cl_metrics = NULL;
589         xprt_put(rcu_dereference_raw(clnt->cl_xprt));
590         rpciod_down();
591         kfree(clnt);
592 }
593
594 /*
595  * Free an RPC client
596  */
597 static void
598 rpc_free_auth(struct rpc_clnt *clnt)
599 {
600         if (clnt->cl_auth == NULL) {
601                 rpc_free_client(clnt);
602                 return;
603         }
604
605         /*
606          * Note: RPCSEC_GSS may need to send NULL RPC calls in order to
607          *       release remaining GSS contexts. This mechanism ensures
608          *       that it can do so safely.
609          */
610         atomic_inc(&clnt->cl_count);
611         rpcauth_release(clnt->cl_auth);
612         clnt->cl_auth = NULL;
613         if (atomic_dec_and_test(&clnt->cl_count))
614                 rpc_free_client(clnt);
615 }
616
617 /*
618  * Release reference to the RPC client
619  */
620 void
621 rpc_release_client(struct rpc_clnt *clnt)
622 {
623         dprintk("RPC:       rpc_release_client(%p)\n", clnt);
624
625         if (list_empty(&clnt->cl_tasks))
626                 wake_up(&destroy_wait);
627         if (atomic_dec_and_test(&clnt->cl_count))
628                 rpc_free_auth(clnt);
629 }
630
631 /**
632  * rpc_bind_new_program - bind a new RPC program to an existing client
633  * @old: old rpc_client
634  * @program: rpc program to set
635  * @vers: rpc program version
636  *
637  * Clones the rpc client and sets up a new RPC program. This is mainly
638  * of use for enabling different RPC programs to share the same transport.
639  * The Sun NFSv2/v3 ACL protocol can do this.
640  */
641 struct rpc_clnt *rpc_bind_new_program(struct rpc_clnt *old,
642                                       const struct rpc_program *program,
643                                       u32 vers)
644 {
645         struct rpc_clnt *clnt;
646         const struct rpc_version *version;
647         int err;
648
649         BUG_ON(vers >= program->nrvers || !program->version[vers]);
650         version = program->version[vers];
651         clnt = rpc_clone_client(old);
652         if (IS_ERR(clnt))
653                 goto out;
654         clnt->cl_procinfo = version->procs;
655         clnt->cl_maxproc  = version->nrprocs;
656         clnt->cl_protname = program->name;
657         clnt->cl_prog     = program->number;
658         clnt->cl_vers     = version->number;
659         clnt->cl_stats    = program->stats;
660         err = rpc_ping(clnt);
661         if (err != 0) {
662                 rpc_shutdown_client(clnt);
663                 clnt = ERR_PTR(err);
664         }
665 out:
666         return clnt;
667 }
668 EXPORT_SYMBOL_GPL(rpc_bind_new_program);
669
670 void rpc_task_release_client(struct rpc_task *task)
671 {
672         struct rpc_clnt *clnt = task->tk_client;
673
674         if (clnt != NULL) {
675                 /* Remove from client task list */
676                 spin_lock(&clnt->cl_lock);
677                 list_del(&task->tk_task);
678                 spin_unlock(&clnt->cl_lock);
679                 task->tk_client = NULL;
680
681                 rpc_release_client(clnt);
682         }
683 }
684
685 static
686 void rpc_task_set_client(struct rpc_task *task, struct rpc_clnt *clnt)
687 {
688         if (clnt != NULL) {
689                 rpc_task_release_client(task);
690                 task->tk_client = clnt;
691                 atomic_inc(&clnt->cl_count);
692                 if (clnt->cl_softrtry)
693                         task->tk_flags |= RPC_TASK_SOFT;
694                 /* Add to the client's list of all tasks */
695                 spin_lock(&clnt->cl_lock);
696                 list_add_tail(&task->tk_task, &clnt->cl_tasks);
697                 spin_unlock(&clnt->cl_lock);
698         }
699 }
700
701 void rpc_task_reset_client(struct rpc_task *task, struct rpc_clnt *clnt)
702 {
703         rpc_task_release_client(task);
704         rpc_task_set_client(task, clnt);
705 }
706 EXPORT_SYMBOL_GPL(rpc_task_reset_client);
707
708
709 static void
710 rpc_task_set_rpc_message(struct rpc_task *task, const struct rpc_message *msg)
711 {
712         if (msg != NULL) {
713                 task->tk_msg.rpc_proc = msg->rpc_proc;
714                 task->tk_msg.rpc_argp = msg->rpc_argp;
715                 task->tk_msg.rpc_resp = msg->rpc_resp;
716                 if (msg->rpc_cred != NULL)
717                         task->tk_msg.rpc_cred = get_rpccred(msg->rpc_cred);
718         }
719 }
720
721 /*
722  * Default callback for async RPC calls
723  */
724 static void
725 rpc_default_callback(struct rpc_task *task, void *data)
726 {
727 }
728
729 static const struct rpc_call_ops rpc_default_ops = {
730         .rpc_call_done = rpc_default_callback,
731 };
732
733 /**
734  * rpc_run_task - Allocate a new RPC task, then run rpc_execute against it
735  * @task_setup_data: pointer to task initialisation data
736  */
737 struct rpc_task *rpc_run_task(const struct rpc_task_setup *task_setup_data)
738 {
739         struct rpc_task *task;
740
741         task = rpc_new_task(task_setup_data);
742         if (IS_ERR(task))
743                 goto out;
744
745         rpc_task_set_client(task, task_setup_data->rpc_client);
746         rpc_task_set_rpc_message(task, task_setup_data->rpc_message);
747
748         if (task->tk_action == NULL)
749                 rpc_call_start(task);
750
751         atomic_inc(&task->tk_count);
752         rpc_execute(task);
753 out:
754         return task;
755 }
756 EXPORT_SYMBOL_GPL(rpc_run_task);
757
758 /**
759  * rpc_call_sync - Perform a synchronous RPC call
760  * @clnt: pointer to RPC client
761  * @msg: RPC call parameters
762  * @flags: RPC call flags
763  */
764 int rpc_call_sync(struct rpc_clnt *clnt, const struct rpc_message *msg, int flags)
765 {
766         struct rpc_task *task;
767         struct rpc_task_setup task_setup_data = {
768                 .rpc_client = clnt,
769                 .rpc_message = msg,
770                 .callback_ops = &rpc_default_ops,
771                 .flags = flags,
772         };
773         int status;
774
775         BUG_ON(flags & RPC_TASK_ASYNC);
776
777         task = rpc_run_task(&task_setup_data);
778         if (IS_ERR(task))
779                 return PTR_ERR(task);
780         status = task->tk_status;
781         rpc_put_task(task);
782         return status;
783 }
784 EXPORT_SYMBOL_GPL(rpc_call_sync);
785
786 /**
787  * rpc_call_async - Perform an asynchronous RPC call
788  * @clnt: pointer to RPC client
789  * @msg: RPC call parameters
790  * @flags: RPC call flags
791  * @tk_ops: RPC call ops
792  * @data: user call data
793  */
794 int
795 rpc_call_async(struct rpc_clnt *clnt, const struct rpc_message *msg, int flags,
796                const struct rpc_call_ops *tk_ops, void *data)
797 {
798         struct rpc_task *task;
799         struct rpc_task_setup task_setup_data = {
800                 .rpc_client = clnt,
801                 .rpc_message = msg,
802                 .callback_ops = tk_ops,
803                 .callback_data = data,
804                 .flags = flags|RPC_TASK_ASYNC,
805         };
806
807         task = rpc_run_task(&task_setup_data);
808         if (IS_ERR(task))
809                 return PTR_ERR(task);
810         rpc_put_task(task);
811         return 0;
812 }
813 EXPORT_SYMBOL_GPL(rpc_call_async);
814
815 #if defined(CONFIG_SUNRPC_BACKCHANNEL)
816 /**
817  * rpc_run_bc_task - Allocate a new RPC task for backchannel use, then run
818  * rpc_execute against it
819  * @req: RPC request
820  * @tk_ops: RPC call ops
821  */
822 struct rpc_task *rpc_run_bc_task(struct rpc_rqst *req,
823                                 const struct rpc_call_ops *tk_ops)
824 {
825         struct rpc_task *task;
826         struct xdr_buf *xbufp = &req->rq_snd_buf;
827         struct rpc_task_setup task_setup_data = {
828                 .callback_ops = tk_ops,
829         };
830
831         dprintk("RPC: rpc_run_bc_task req= %p\n", req);
832         /*
833          * Create an rpc_task to send the data
834          */
835         task = rpc_new_task(&task_setup_data);
836         if (IS_ERR(task)) {
837                 xprt_free_bc_request(req);
838                 goto out;
839         }
840         task->tk_rqstp = req;
841
842         /*
843          * Set up the xdr_buf length.
844          * This also indicates that the buffer is XDR encoded already.
845          */
846         xbufp->len = xbufp->head[0].iov_len + xbufp->page_len +
847                         xbufp->tail[0].iov_len;
848
849         task->tk_action = call_bc_transmit;
850         atomic_inc(&task->tk_count);
851         BUG_ON(atomic_read(&task->tk_count) != 2);
852         rpc_execute(task);
853
854 out:
855         dprintk("RPC: rpc_run_bc_task: task= %p\n", task);
856         return task;
857 }
858 #endif /* CONFIG_SUNRPC_BACKCHANNEL */
859
860 void
861 rpc_call_start(struct rpc_task *task)
862 {
863         task->tk_action = call_start;
864 }
865 EXPORT_SYMBOL_GPL(rpc_call_start);
866
867 /**
868  * rpc_peeraddr - extract remote peer address from clnt's xprt
869  * @clnt: RPC client structure
870  * @buf: target buffer
871  * @bufsize: length of target buffer
872  *
873  * Returns the number of bytes that are actually in the stored address.
874  */
875 size_t rpc_peeraddr(struct rpc_clnt *clnt, struct sockaddr *buf, size_t bufsize)
876 {
877         size_t bytes;
878         struct rpc_xprt *xprt;
879
880         rcu_read_lock();
881         xprt = rcu_dereference(clnt->cl_xprt);
882
883         bytes = xprt->addrlen;
884         if (bytes > bufsize)
885                 bytes = bufsize;
886         memcpy(buf, &xprt->addr, bytes);
887         rcu_read_unlock();
888
889         return bytes;
890 }
891 EXPORT_SYMBOL_GPL(rpc_peeraddr);
892
893 /**
894  * rpc_peeraddr2str - return remote peer address in printable format
895  * @clnt: RPC client structure
896  * @format: address format
897  *
898  * NB: the lifetime of the memory referenced by the returned pointer is
899  * the same as the rpc_xprt itself.  As long as the caller uses this
900  * pointer, it must hold the RCU read lock.
901  */
902 const char *rpc_peeraddr2str(struct rpc_clnt *clnt,
903                              enum rpc_display_format_t format)
904 {
905         struct rpc_xprt *xprt;
906
907         xprt = rcu_dereference(clnt->cl_xprt);
908
909         if (xprt->address_strings[format] != NULL)
910                 return xprt->address_strings[format];
911         else
912                 return "unprintable";
913 }
914 EXPORT_SYMBOL_GPL(rpc_peeraddr2str);
915
916 static const struct sockaddr_in rpc_inaddr_loopback = {
917         .sin_family             = AF_INET,
918         .sin_addr.s_addr        = htonl(INADDR_ANY),
919 };
920
921 static const struct sockaddr_in6 rpc_in6addr_loopback = {
922         .sin6_family            = AF_INET6,
923         .sin6_addr              = IN6ADDR_ANY_INIT,
924 };
925
926 /*
927  * Try a getsockname() on a connected datagram socket.  Using a
928  * connected datagram socket prevents leaving a socket in TIME_WAIT.
929  * This conserves the ephemeral port number space.
930  *
931  * Returns zero and fills in "buf" if successful; otherwise, a
932  * negative errno is returned.
933  */
934 static int rpc_sockname(struct net *net, struct sockaddr *sap, size_t salen,
935                         struct sockaddr *buf, int buflen)
936 {
937         struct socket *sock;
938         int err;
939
940         err = __sock_create(net, sap->sa_family,
941                                 SOCK_DGRAM, IPPROTO_UDP, &sock, 1);
942         if (err < 0) {
943                 dprintk("RPC:       can't create UDP socket (%d)\n", err);
944                 goto out;
945         }
946
947         switch (sap->sa_family) {
948         case AF_INET:
949                 err = kernel_bind(sock,
950                                 (struct sockaddr *)&rpc_inaddr_loopback,
951                                 sizeof(rpc_inaddr_loopback));
952                 break;
953         case AF_INET6:
954                 err = kernel_bind(sock,
955                                 (struct sockaddr *)&rpc_in6addr_loopback,
956                                 sizeof(rpc_in6addr_loopback));
957                 break;
958         default:
959                 err = -EAFNOSUPPORT;
960                 goto out;
961         }
962         if (err < 0) {
963                 dprintk("RPC:       can't bind UDP socket (%d)\n", err);
964                 goto out_release;
965         }
966
967         err = kernel_connect(sock, sap, salen, 0);
968         if (err < 0) {
969                 dprintk("RPC:       can't connect UDP socket (%d)\n", err);
970                 goto out_release;
971         }
972
973         err = kernel_getsockname(sock, buf, &buflen);
974         if (err < 0) {
975                 dprintk("RPC:       getsockname failed (%d)\n", err);
976                 goto out_release;
977         }
978
979         err = 0;
980         if (buf->sa_family == AF_INET6) {
981                 struct sockaddr_in6 *sin6 = (struct sockaddr_in6 *)buf;
982                 sin6->sin6_scope_id = 0;
983         }
984         dprintk("RPC:       %s succeeded\n", __func__);
985
986 out_release:
987         sock_release(sock);
988 out:
989         return err;
990 }
991
992 /*
993  * Scraping a connected socket failed, so we don't have a useable
994  * local address.  Fallback: generate an address that will prevent
995  * the server from calling us back.
996  *
997  * Returns zero and fills in "buf" if successful; otherwise, a
998  * negative errno is returned.
999  */
1000 static int rpc_anyaddr(int family, struct sockaddr *buf, size_t buflen)
1001 {
1002         switch (family) {
1003         case AF_INET:
1004                 if (buflen < sizeof(rpc_inaddr_loopback))
1005                         return -EINVAL;
1006                 memcpy(buf, &rpc_inaddr_loopback,
1007                                 sizeof(rpc_inaddr_loopback));
1008                 break;
1009         case AF_INET6:
1010                 if (buflen < sizeof(rpc_in6addr_loopback))
1011                         return -EINVAL;
1012                 memcpy(buf, &rpc_in6addr_loopback,
1013                                 sizeof(rpc_in6addr_loopback));
1014         default:
1015                 dprintk("RPC:       %s: address family not supported\n",
1016                         __func__);
1017                 return -EAFNOSUPPORT;
1018         }
1019         dprintk("RPC:       %s: succeeded\n", __func__);
1020         return 0;
1021 }
1022
1023 /**
1024  * rpc_localaddr - discover local endpoint address for an RPC client
1025  * @clnt: RPC client structure
1026  * @buf: target buffer
1027  * @buflen: size of target buffer, in bytes
1028  *
1029  * Returns zero and fills in "buf" and "buflen" if successful;
1030  * otherwise, a negative errno is returned.
1031  *
1032  * This works even if the underlying transport is not currently connected,
1033  * or if the upper layer never previously provided a source address.
1034  *
1035  * The result of this function call is transient: multiple calls in
1036  * succession may give different results, depending on how local
1037  * networking configuration changes over time.
1038  */
1039 int rpc_localaddr(struct rpc_clnt *clnt, struct sockaddr *buf, size_t buflen)
1040 {
1041         struct sockaddr_storage address;
1042         struct sockaddr *sap = (struct sockaddr *)&address;
1043         struct rpc_xprt *xprt;
1044         struct net *net;
1045         size_t salen;
1046         int err;
1047
1048         rcu_read_lock();
1049         xprt = rcu_dereference(clnt->cl_xprt);
1050         salen = xprt->addrlen;
1051         memcpy(sap, &xprt->addr, salen);
1052         net = get_net(xprt->xprt_net);
1053         rcu_read_unlock();
1054
1055         rpc_set_port(sap, 0);
1056         err = rpc_sockname(net, sap, salen, buf, buflen);
1057         put_net(net);
1058         if (err != 0)
1059                 /* Couldn't discover local address, return ANYADDR */
1060                 return rpc_anyaddr(sap->sa_family, buf, buflen);
1061         return 0;
1062 }
1063 EXPORT_SYMBOL_GPL(rpc_localaddr);
1064
1065 void
1066 rpc_setbufsize(struct rpc_clnt *clnt, unsigned int sndsize, unsigned int rcvsize)
1067 {
1068         struct rpc_xprt *xprt;
1069
1070         rcu_read_lock();
1071         xprt = rcu_dereference(clnt->cl_xprt);
1072         if (xprt->ops->set_buffer_size)
1073                 xprt->ops->set_buffer_size(xprt, sndsize, rcvsize);
1074         rcu_read_unlock();
1075 }
1076 EXPORT_SYMBOL_GPL(rpc_setbufsize);
1077
1078 /**
1079  * rpc_protocol - Get transport protocol number for an RPC client
1080  * @clnt: RPC client to query
1081  *
1082  */
1083 int rpc_protocol(struct rpc_clnt *clnt)
1084 {
1085         int protocol;
1086
1087         rcu_read_lock();
1088         protocol = rcu_dereference(clnt->cl_xprt)->prot;
1089         rcu_read_unlock();
1090         return protocol;
1091 }
1092 EXPORT_SYMBOL_GPL(rpc_protocol);
1093
1094 /**
1095  * rpc_net_ns - Get the network namespace for this RPC client
1096  * @clnt: RPC client to query
1097  *
1098  */
1099 struct net *rpc_net_ns(struct rpc_clnt *clnt)
1100 {
1101         struct net *ret;
1102
1103         rcu_read_lock();
1104         ret = rcu_dereference(clnt->cl_xprt)->xprt_net;
1105         rcu_read_unlock();
1106         return ret;
1107 }
1108 EXPORT_SYMBOL_GPL(rpc_net_ns);
1109
1110 /**
1111  * rpc_max_payload - Get maximum payload size for a transport, in bytes
1112  * @clnt: RPC client to query
1113  *
1114  * For stream transports, this is one RPC record fragment (see RFC
1115  * 1831), as we don't support multi-record requests yet.  For datagram
1116  * transports, this is the size of an IP packet minus the IP, UDP, and
1117  * RPC header sizes.
1118  */
1119 size_t rpc_max_payload(struct rpc_clnt *clnt)
1120 {
1121         size_t ret;
1122
1123         rcu_read_lock();
1124         ret = rcu_dereference(clnt->cl_xprt)->max_payload;
1125         rcu_read_unlock();
1126         return ret;
1127 }
1128 EXPORT_SYMBOL_GPL(rpc_max_payload);
1129
1130 /**
1131  * rpc_force_rebind - force transport to check that remote port is unchanged
1132  * @clnt: client to rebind
1133  *
1134  */
1135 void rpc_force_rebind(struct rpc_clnt *clnt)
1136 {
1137         if (clnt->cl_autobind) {
1138                 rcu_read_lock();
1139                 xprt_clear_bound(rcu_dereference(clnt->cl_xprt));
1140                 rcu_read_unlock();
1141         }
1142 }
1143 EXPORT_SYMBOL_GPL(rpc_force_rebind);
1144
1145 /*
1146  * Restart an (async) RPC call from the call_prepare state.
1147  * Usually called from within the exit handler.
1148  */
1149 int
1150 rpc_restart_call_prepare(struct rpc_task *task)
1151 {
1152         if (RPC_ASSASSINATED(task))
1153                 return 0;
1154         task->tk_action = call_start;
1155         if (task->tk_ops->rpc_call_prepare != NULL)
1156                 task->tk_action = rpc_prepare_task;
1157         return 1;
1158 }
1159 EXPORT_SYMBOL_GPL(rpc_restart_call_prepare);
1160
1161 /*
1162  * Restart an (async) RPC call. Usually called from within the
1163  * exit handler.
1164  */
1165 int
1166 rpc_restart_call(struct rpc_task *task)
1167 {
1168         if (RPC_ASSASSINATED(task))
1169                 return 0;
1170         task->tk_action = call_start;
1171         return 1;
1172 }
1173 EXPORT_SYMBOL_GPL(rpc_restart_call);
1174
1175 #ifdef RPC_DEBUG
1176 static const char *rpc_proc_name(const struct rpc_task *task)
1177 {
1178         const struct rpc_procinfo *proc = task->tk_msg.rpc_proc;
1179
1180         if (proc) {
1181                 if (proc->p_name)
1182                         return proc->p_name;
1183                 else
1184                         return "NULL";
1185         } else
1186                 return "no proc";
1187 }
1188 #endif
1189
1190 /*
1191  * 0.  Initial state
1192  *
1193  *     Other FSM states can be visited zero or more times, but
1194  *     this state is visited exactly once for each RPC.
1195  */
1196 static void
1197 call_start(struct rpc_task *task)
1198 {
1199         struct rpc_clnt *clnt = task->tk_client;
1200
1201         dprintk("RPC: %5u call_start %s%d proc %s (%s)\n", task->tk_pid,
1202                         clnt->cl_protname, clnt->cl_vers,
1203                         rpc_proc_name(task),
1204                         (RPC_IS_ASYNC(task) ? "async" : "sync"));
1205
1206         /* Increment call count */
1207         task->tk_msg.rpc_proc->p_count++;
1208         clnt->cl_stats->rpccnt++;
1209         task->tk_action = call_reserve;
1210 }
1211
1212 /*
1213  * 1.   Reserve an RPC call slot
1214  */
1215 static void
1216 call_reserve(struct rpc_task *task)
1217 {
1218         dprint_status(task);
1219
1220         task->tk_status  = 0;
1221         task->tk_action  = call_reserveresult;
1222         xprt_reserve(task);
1223 }
1224
1225 /*
1226  * 1b.  Grok the result of xprt_reserve()
1227  */
1228 static void
1229 call_reserveresult(struct rpc_task *task)
1230 {
1231         int status = task->tk_status;
1232
1233         dprint_status(task);
1234
1235         /*
1236          * After a call to xprt_reserve(), we must have either
1237          * a request slot or else an error status.
1238          */
1239         task->tk_status = 0;
1240         if (status >= 0) {
1241                 if (task->tk_rqstp) {
1242                         task->tk_action = call_refresh;
1243                         return;
1244                 }
1245
1246                 printk(KERN_ERR "%s: status=%d, but no request slot, exiting\n",
1247                                 __func__, status);
1248                 rpc_exit(task, -EIO);
1249                 return;
1250         }
1251
1252         /*
1253          * Even though there was an error, we may have acquired
1254          * a request slot somehow.  Make sure not to leak it.
1255          */
1256         if (task->tk_rqstp) {
1257                 printk(KERN_ERR "%s: status=%d, request allocated anyway\n",
1258                                 __func__, status);
1259                 xprt_release(task);
1260         }
1261
1262         switch (status) {
1263         case -EAGAIN:   /* woken up; retry */
1264                 task->tk_action = call_reserve;
1265                 return;
1266         case -EIO:      /* probably a shutdown */
1267                 break;
1268         default:
1269                 printk(KERN_ERR "%s: unrecognized error %d, exiting\n",
1270                                 __func__, status);
1271                 break;
1272         }
1273         rpc_exit(task, status);
1274 }
1275
1276 /*
1277  * 2.   Bind and/or refresh the credentials
1278  */
1279 static void
1280 call_refresh(struct rpc_task *task)
1281 {
1282         dprint_status(task);
1283
1284         task->tk_action = call_refreshresult;
1285         task->tk_status = 0;
1286         task->tk_client->cl_stats->rpcauthrefresh++;
1287         rpcauth_refreshcred(task);
1288 }
1289
1290 /*
1291  * 2a.  Process the results of a credential refresh
1292  */
1293 static void
1294 call_refreshresult(struct rpc_task *task)
1295 {
1296         int status = task->tk_status;
1297
1298         dprint_status(task);
1299
1300         task->tk_status = 0;
1301         task->tk_action = call_refresh;
1302         switch (status) {
1303         case 0:
1304                 if (rpcauth_uptodatecred(task))
1305                         task->tk_action = call_allocate;
1306                 return;
1307         case -ETIMEDOUT:
1308                 rpc_delay(task, 3*HZ);
1309         case -EAGAIN:
1310                 status = -EACCES;
1311                 if (!task->tk_cred_retry)
1312                         break;
1313                 task->tk_cred_retry--;
1314                 dprintk("RPC: %5u %s: retry refresh creds\n",
1315                                 task->tk_pid, __func__);
1316                 return;
1317         }
1318         dprintk("RPC: %5u %s: refresh creds failed with error %d\n",
1319                                 task->tk_pid, __func__, status);
1320         rpc_exit(task, status);
1321 }
1322
1323 /*
1324  * 2b.  Allocate the buffer. For details, see sched.c:rpc_malloc.
1325  *      (Note: buffer memory is freed in xprt_release).
1326  */
1327 static void
1328 call_allocate(struct rpc_task *task)
1329 {
1330         unsigned int slack = task->tk_rqstp->rq_cred->cr_auth->au_cslack;
1331         struct rpc_rqst *req = task->tk_rqstp;
1332         struct rpc_xprt *xprt = task->tk_xprt;
1333         struct rpc_procinfo *proc = task->tk_msg.rpc_proc;
1334
1335         dprint_status(task);
1336
1337         task->tk_status = 0;
1338         task->tk_action = call_bind;
1339
1340         if (req->rq_buffer)
1341                 return;
1342
1343         if (proc->p_proc != 0) {
1344                 BUG_ON(proc->p_arglen == 0);
1345                 if (proc->p_decode != NULL)
1346                         BUG_ON(proc->p_replen == 0);
1347         }
1348
1349         /*
1350          * Calculate the size (in quads) of the RPC call
1351          * and reply headers, and convert both values
1352          * to byte sizes.
1353          */
1354         req->rq_callsize = RPC_CALLHDRSIZE + (slack << 1) + proc->p_arglen;
1355         req->rq_callsize <<= 2;
1356         req->rq_rcvsize = RPC_REPHDRSIZE + slack + proc->p_replen;
1357         req->rq_rcvsize <<= 2;
1358
1359         req->rq_buffer = xprt->ops->buf_alloc(task,
1360                                         req->rq_callsize + req->rq_rcvsize);
1361         if (req->rq_buffer != NULL)
1362                 return;
1363
1364         dprintk("RPC: %5u rpc_buffer allocation failed\n", task->tk_pid);
1365
1366         if (RPC_IS_ASYNC(task) || !fatal_signal_pending(current)) {
1367                 task->tk_action = call_allocate;
1368                 rpc_delay(task, HZ>>4);
1369                 return;
1370         }
1371
1372         rpc_exit(task, -ERESTARTSYS);
1373 }
1374
1375 static inline int
1376 rpc_task_need_encode(struct rpc_task *task)
1377 {
1378         return task->tk_rqstp->rq_snd_buf.len == 0;
1379 }
1380
1381 static inline void
1382 rpc_task_force_reencode(struct rpc_task *task)
1383 {
1384         task->tk_rqstp->rq_snd_buf.len = 0;
1385         task->tk_rqstp->rq_bytes_sent = 0;
1386 }
1387
1388 static inline void
1389 rpc_xdr_buf_init(struct xdr_buf *buf, void *start, size_t len)
1390 {
1391         buf->head[0].iov_base = start;
1392         buf->head[0].iov_len = len;
1393         buf->tail[0].iov_len = 0;
1394         buf->page_len = 0;
1395         buf->flags = 0;
1396         buf->len = 0;
1397         buf->buflen = len;
1398 }
1399
1400 /*
1401  * 3.   Encode arguments of an RPC call
1402  */
1403 static void
1404 rpc_xdr_encode(struct rpc_task *task)
1405 {
1406         struct rpc_rqst *req = task->tk_rqstp;
1407         kxdreproc_t     encode;
1408         __be32          *p;
1409
1410         dprint_status(task);
1411
1412         rpc_xdr_buf_init(&req->rq_snd_buf,
1413                          req->rq_buffer,
1414                          req->rq_callsize);
1415         rpc_xdr_buf_init(&req->rq_rcv_buf,
1416                          (char *)req->rq_buffer + req->rq_callsize,
1417                          req->rq_rcvsize);
1418
1419         p = rpc_encode_header(task);
1420         if (p == NULL) {
1421                 printk(KERN_INFO "RPC: couldn't encode RPC header, exit EIO\n");
1422                 rpc_exit(task, -EIO);
1423                 return;
1424         }
1425
1426         encode = task->tk_msg.rpc_proc->p_encode;
1427         if (encode == NULL)
1428                 return;
1429
1430         task->tk_status = rpcauth_wrap_req(task, encode, req, p,
1431                         task->tk_msg.rpc_argp);
1432 }
1433
1434 /*
1435  * 4.   Get the server port number if not yet set
1436  */
1437 static void
1438 call_bind(struct rpc_task *task)
1439 {
1440         struct rpc_xprt *xprt = task->tk_xprt;
1441
1442         dprint_status(task);
1443
1444         task->tk_action = call_connect;
1445         if (!xprt_bound(xprt)) {
1446                 task->tk_action = call_bind_status;
1447                 task->tk_timeout = xprt->bind_timeout;
1448                 xprt->ops->rpcbind(task);
1449         }
1450 }
1451
1452 /*
1453  * 4a.  Sort out bind result
1454  */
1455 static void
1456 call_bind_status(struct rpc_task *task)
1457 {
1458         int status = -EIO;
1459
1460         if (task->tk_status >= 0) {
1461                 dprint_status(task);
1462                 task->tk_status = 0;
1463                 task->tk_action = call_connect;
1464                 return;
1465         }
1466
1467         trace_rpc_bind_status(task);
1468         switch (task->tk_status) {
1469         case -ENOMEM:
1470                 dprintk("RPC: %5u rpcbind out of memory\n", task->tk_pid);
1471                 rpc_delay(task, HZ >> 2);
1472                 goto retry_timeout;
1473         case -EACCES:
1474                 dprintk("RPC: %5u remote rpcbind: RPC program/version "
1475                                 "unavailable\n", task->tk_pid);
1476                 /* fail immediately if this is an RPC ping */
1477                 if (task->tk_msg.rpc_proc->p_proc == 0) {
1478                         status = -EOPNOTSUPP;
1479                         break;
1480                 }
1481                 if (task->tk_rebind_retry == 0)
1482                         break;
1483                 task->tk_rebind_retry--;
1484                 rpc_delay(task, 3*HZ);
1485                 goto retry_timeout;
1486         case -ETIMEDOUT:
1487                 dprintk("RPC: %5u rpcbind request timed out\n",
1488                                 task->tk_pid);
1489                 goto retry_timeout;
1490         case -EPFNOSUPPORT:
1491                 /* server doesn't support any rpcbind version we know of */
1492                 dprintk("RPC: %5u unrecognized remote rpcbind service\n",
1493                                 task->tk_pid);
1494                 break;
1495         case -EPROTONOSUPPORT:
1496                 dprintk("RPC: %5u remote rpcbind version unavailable, retrying\n",
1497                                 task->tk_pid);
1498                 task->tk_status = 0;
1499                 task->tk_action = call_bind;
1500                 return;
1501         case -ECONNREFUSED:             /* connection problems */
1502         case -ECONNRESET:
1503         case -ENOTCONN:
1504         case -EHOSTDOWN:
1505         case -EHOSTUNREACH:
1506         case -ENETUNREACH:
1507         case -EPIPE:
1508                 dprintk("RPC: %5u remote rpcbind unreachable: %d\n",
1509                                 task->tk_pid, task->tk_status);
1510                 if (!RPC_IS_SOFTCONN(task)) {
1511                         rpc_delay(task, 5*HZ);
1512                         goto retry_timeout;
1513                 }
1514                 status = task->tk_status;
1515                 break;
1516         default:
1517                 dprintk("RPC: %5u unrecognized rpcbind error (%d)\n",
1518                                 task->tk_pid, -task->tk_status);
1519         }
1520
1521         rpc_exit(task, status);
1522         return;
1523
1524 retry_timeout:
1525         task->tk_action = call_timeout;
1526 }
1527
1528 /*
1529  * 4b.  Connect to the RPC server
1530  */
1531 static void
1532 call_connect(struct rpc_task *task)
1533 {
1534         struct rpc_xprt *xprt = task->tk_xprt;
1535
1536         dprintk("RPC: %5u call_connect xprt %p %s connected\n",
1537                         task->tk_pid, xprt,
1538                         (xprt_connected(xprt) ? "is" : "is not"));
1539
1540         task->tk_action = call_transmit;
1541         if (!xprt_connected(xprt)) {
1542                 task->tk_action = call_connect_status;
1543                 if (task->tk_status < 0)
1544                         return;
1545                 xprt_connect(task);
1546         }
1547 }
1548
1549 /*
1550  * 4c.  Sort out connect result
1551  */
1552 static void
1553 call_connect_status(struct rpc_task *task)
1554 {
1555         struct rpc_clnt *clnt = task->tk_client;
1556         int status = task->tk_status;
1557
1558         dprint_status(task);
1559
1560         task->tk_status = 0;
1561         if (status >= 0 || status == -EAGAIN) {
1562                 clnt->cl_stats->netreconn++;
1563                 task->tk_action = call_transmit;
1564                 return;
1565         }
1566
1567         trace_rpc_connect_status(task, status);
1568         switch (status) {
1569                 /* if soft mounted, test if we've timed out */
1570         case -ETIMEDOUT:
1571                 task->tk_action = call_timeout;
1572                 break;
1573         default:
1574                 rpc_exit(task, -EIO);
1575         }
1576 }
1577
1578 /*
1579  * 5.   Transmit the RPC request, and wait for reply
1580  */
1581 static void
1582 call_transmit(struct rpc_task *task)
1583 {
1584         dprint_status(task);
1585
1586         task->tk_action = call_status;
1587         if (task->tk_status < 0)
1588                 return;
1589         task->tk_status = xprt_prepare_transmit(task);
1590         if (task->tk_status != 0)
1591                 return;
1592         task->tk_action = call_transmit_status;
1593         /* Encode here so that rpcsec_gss can use correct sequence number. */
1594         if (rpc_task_need_encode(task)) {
1595                 BUG_ON(task->tk_rqstp->rq_bytes_sent != 0);
1596                 rpc_xdr_encode(task);
1597                 /* Did the encode result in an error condition? */
1598                 if (task->tk_status != 0) {
1599                         /* Was the error nonfatal? */
1600                         if (task->tk_status == -EAGAIN)
1601                                 rpc_delay(task, HZ >> 4);
1602                         else
1603                                 rpc_exit(task, task->tk_status);
1604                         return;
1605                 }
1606         }
1607         xprt_transmit(task);
1608         if (task->tk_status < 0)
1609                 return;
1610         /*
1611          * On success, ensure that we call xprt_end_transmit() before sleeping
1612          * in order to allow access to the socket to other RPC requests.
1613          */
1614         call_transmit_status(task);
1615         if (rpc_reply_expected(task))
1616                 return;
1617         task->tk_action = rpc_exit_task;
1618         rpc_wake_up_queued_task(&task->tk_xprt->pending, task);
1619 }
1620
1621 /*
1622  * 5a.  Handle cleanup after a transmission
1623  */
1624 static void
1625 call_transmit_status(struct rpc_task *task)
1626 {
1627         task->tk_action = call_status;
1628
1629         /*
1630          * Common case: success.  Force the compiler to put this
1631          * test first.
1632          */
1633         if (task->tk_status == 0) {
1634                 xprt_end_transmit(task);
1635                 rpc_task_force_reencode(task);
1636                 return;
1637         }
1638
1639         switch (task->tk_status) {
1640         case -EAGAIN:
1641                 break;
1642         default:
1643                 dprint_status(task);
1644                 xprt_end_transmit(task);
1645                 rpc_task_force_reencode(task);
1646                 break;
1647                 /*
1648                  * Special cases: if we've been waiting on the
1649                  * socket's write_space() callback, or if the
1650                  * socket just returned a connection error,
1651                  * then hold onto the transport lock.
1652                  */
1653         case -ECONNREFUSED:
1654         case -EHOSTDOWN:
1655         case -EHOSTUNREACH:
1656         case -ENETUNREACH:
1657                 if (RPC_IS_SOFTCONN(task)) {
1658                         xprt_end_transmit(task);
1659                         rpc_exit(task, task->tk_status);
1660                         break;
1661                 }
1662         case -ECONNRESET:
1663         case -ENOTCONN:
1664         case -EPIPE:
1665                 rpc_task_force_reencode(task);
1666         }
1667 }
1668
1669 #if defined(CONFIG_SUNRPC_BACKCHANNEL)
1670 /*
1671  * 5b.  Send the backchannel RPC reply.  On error, drop the reply.  In
1672  * addition, disconnect on connectivity errors.
1673  */
1674 static void
1675 call_bc_transmit(struct rpc_task *task)
1676 {
1677         struct rpc_rqst *req = task->tk_rqstp;
1678
1679         BUG_ON(task->tk_status != 0);
1680         task->tk_status = xprt_prepare_transmit(task);
1681         if (task->tk_status == -EAGAIN) {
1682                 /*
1683                  * Could not reserve the transport. Try again after the
1684                  * transport is released.
1685                  */
1686                 task->tk_status = 0;
1687                 task->tk_action = call_bc_transmit;
1688                 return;
1689         }
1690
1691         task->tk_action = rpc_exit_task;
1692         if (task->tk_status < 0) {
1693                 printk(KERN_NOTICE "RPC: Could not send backchannel reply "
1694                         "error: %d\n", task->tk_status);
1695                 return;
1696         }
1697
1698         xprt_transmit(task);
1699         xprt_end_transmit(task);
1700         dprint_status(task);
1701         switch (task->tk_status) {
1702         case 0:
1703                 /* Success */
1704                 break;
1705         case -EHOSTDOWN:
1706         case -EHOSTUNREACH:
1707         case -ENETUNREACH:
1708         case -ETIMEDOUT:
1709                 /*
1710                  * Problem reaching the server.  Disconnect and let the
1711                  * forechannel reestablish the connection.  The server will
1712                  * have to retransmit the backchannel request and we'll
1713                  * reprocess it.  Since these ops are idempotent, there's no
1714                  * need to cache our reply at this time.
1715                  */
1716                 printk(KERN_NOTICE "RPC: Could not send backchannel reply "
1717                         "error: %d\n", task->tk_status);
1718                 xprt_conditional_disconnect(task->tk_xprt,
1719                         req->rq_connect_cookie);
1720                 break;
1721         default:
1722                 /*
1723                  * We were unable to reply and will have to drop the
1724                  * request.  The server should reconnect and retransmit.
1725                  */
1726                 BUG_ON(task->tk_status == -EAGAIN);
1727                 printk(KERN_NOTICE "RPC: Could not send backchannel reply "
1728                         "error: %d\n", task->tk_status);
1729                 break;
1730         }
1731         rpc_wake_up_queued_task(&req->rq_xprt->pending, task);
1732 }
1733 #endif /* CONFIG_SUNRPC_BACKCHANNEL */
1734
1735 /*
1736  * 6.   Sort out the RPC call status
1737  */
1738 static void
1739 call_status(struct rpc_task *task)
1740 {
1741         struct rpc_clnt *clnt = task->tk_client;
1742         struct rpc_rqst *req = task->tk_rqstp;
1743         int             status;
1744
1745         if (req->rq_reply_bytes_recvd > 0 && !req->rq_bytes_sent)
1746                 task->tk_status = req->rq_reply_bytes_recvd;
1747
1748         dprint_status(task);
1749
1750         status = task->tk_status;
1751         if (status >= 0) {
1752                 task->tk_action = call_decode;
1753                 return;
1754         }
1755
1756         trace_rpc_call_status(task);
1757         task->tk_status = 0;
1758         switch(status) {
1759         case -EHOSTDOWN:
1760         case -EHOSTUNREACH:
1761         case -ENETUNREACH:
1762                 /*
1763                  * Delay any retries for 3 seconds, then handle as if it
1764                  * were a timeout.
1765                  */
1766                 rpc_delay(task, 3*HZ);
1767         case -ETIMEDOUT:
1768                 task->tk_action = call_timeout;
1769                 if (task->tk_client->cl_discrtry)
1770                         xprt_conditional_disconnect(task->tk_xprt,
1771                                         req->rq_connect_cookie);
1772                 break;
1773         case -ECONNRESET:
1774         case -ECONNREFUSED:
1775                 rpc_force_rebind(clnt);
1776                 rpc_delay(task, 3*HZ);
1777         case -EPIPE:
1778         case -ENOTCONN:
1779                 task->tk_action = call_bind;
1780                 break;
1781         case -EAGAIN:
1782                 task->tk_action = call_transmit;
1783                 break;
1784         case -EIO:
1785                 /* shutdown or soft timeout */
1786                 rpc_exit(task, status);
1787                 break;
1788         default:
1789                 if (clnt->cl_chatty)
1790                         printk("%s: RPC call returned error %d\n",
1791                                clnt->cl_protname, -status);
1792                 rpc_exit(task, status);
1793         }
1794 }
1795
1796 /*
1797  * 6a.  Handle RPC timeout
1798  *      We do not release the request slot, so we keep using the
1799  *      same XID for all retransmits.
1800  */
1801 static void
1802 call_timeout(struct rpc_task *task)
1803 {
1804         struct rpc_clnt *clnt = task->tk_client;
1805
1806         if (xprt_adjust_timeout(task->tk_rqstp) == 0) {
1807                 dprintk("RPC: %5u call_timeout (minor)\n", task->tk_pid);
1808                 goto retry;
1809         }
1810
1811         dprintk("RPC: %5u call_timeout (major)\n", task->tk_pid);
1812         task->tk_timeouts++;
1813
1814         if (RPC_IS_SOFTCONN(task)) {
1815                 rpc_exit(task, -ETIMEDOUT);
1816                 return;
1817         }
1818         if (RPC_IS_SOFT(task)) {
1819                 if (clnt->cl_chatty)
1820                         rcu_read_lock();
1821                         printk(KERN_NOTICE "%s: server %s not responding, timed out\n",
1822                                 clnt->cl_protname,
1823                                 rcu_dereference(clnt->cl_xprt)->servername);
1824                         rcu_read_unlock();
1825                 if (task->tk_flags & RPC_TASK_TIMEOUT)
1826                         rpc_exit(task, -ETIMEDOUT);
1827                 else
1828                         rpc_exit(task, -EIO);
1829                 return;
1830         }
1831
1832         if (!(task->tk_flags & RPC_CALL_MAJORSEEN)) {
1833                 task->tk_flags |= RPC_CALL_MAJORSEEN;
1834                 if (clnt->cl_chatty) {
1835                         rcu_read_lock();
1836                         printk(KERN_NOTICE "%s: server %s not responding, still trying\n",
1837                         clnt->cl_protname,
1838                         rcu_dereference(clnt->cl_xprt)->servername);
1839                         rcu_read_unlock();
1840                 }
1841         }
1842         rpc_force_rebind(clnt);
1843         /*
1844          * Did our request time out due to an RPCSEC_GSS out-of-sequence
1845          * event? RFC2203 requires the server to drop all such requests.
1846          */
1847         rpcauth_invalcred(task);
1848
1849 retry:
1850         clnt->cl_stats->rpcretrans++;
1851         task->tk_action = call_bind;
1852         task->tk_status = 0;
1853 }
1854
1855 /*
1856  * 7.   Decode the RPC reply
1857  */
1858 static void
1859 call_decode(struct rpc_task *task)
1860 {
1861         struct rpc_clnt *clnt = task->tk_client;
1862         struct rpc_rqst *req = task->tk_rqstp;
1863         kxdrdproc_t     decode = task->tk_msg.rpc_proc->p_decode;
1864         __be32          *p;
1865
1866         dprint_status(task);
1867
1868         if (task->tk_flags & RPC_CALL_MAJORSEEN) {
1869                 if (clnt->cl_chatty) {
1870                         rcu_read_lock();
1871                         printk(KERN_NOTICE "%s: server %s OK\n",
1872                                 clnt->cl_protname,
1873                                 rcu_dereference(clnt->cl_xprt)->servername);
1874                         rcu_read_unlock();
1875                 }
1876                 task->tk_flags &= ~RPC_CALL_MAJORSEEN;
1877         }
1878
1879         /*
1880          * Ensure that we see all writes made by xprt_complete_rqst()
1881          * before it changed req->rq_reply_bytes_recvd.
1882          */
1883         smp_rmb();
1884         req->rq_rcv_buf.len = req->rq_private_buf.len;
1885
1886         /* Check that the softirq receive buffer is valid */
1887         WARN_ON(memcmp(&req->rq_rcv_buf, &req->rq_private_buf,
1888                                 sizeof(req->rq_rcv_buf)) != 0);
1889
1890         if (req->rq_rcv_buf.len < 12) {
1891                 if (!RPC_IS_SOFT(task)) {
1892                         task->tk_action = call_bind;
1893                         clnt->cl_stats->rpcretrans++;
1894                         goto out_retry;
1895                 }
1896                 dprintk("RPC:       %s: too small RPC reply size (%d bytes)\n",
1897                                 clnt->cl_protname, task->tk_status);
1898                 task->tk_action = call_timeout;
1899                 goto out_retry;
1900         }
1901
1902         p = rpc_verify_header(task);
1903         if (IS_ERR(p)) {
1904                 if (p == ERR_PTR(-EAGAIN))
1905                         goto out_retry;
1906                 return;
1907         }
1908
1909         task->tk_action = rpc_exit_task;
1910
1911         if (decode) {
1912                 task->tk_status = rpcauth_unwrap_resp(task, decode, req, p,
1913                                                       task->tk_msg.rpc_resp);
1914         }
1915         dprintk("RPC: %5u call_decode result %d\n", task->tk_pid,
1916                         task->tk_status);
1917         return;
1918 out_retry:
1919         task->tk_status = 0;
1920         /* Note: rpc_verify_header() may have freed the RPC slot */
1921         if (task->tk_rqstp == req) {
1922                 req->rq_reply_bytes_recvd = req->rq_rcv_buf.len = 0;
1923                 if (task->tk_client->cl_discrtry)
1924                         xprt_conditional_disconnect(task->tk_xprt,
1925                                         req->rq_connect_cookie);
1926         }
1927 }
1928
1929 static __be32 *
1930 rpc_encode_header(struct rpc_task *task)
1931 {
1932         struct rpc_clnt *clnt = task->tk_client;
1933         struct rpc_rqst *req = task->tk_rqstp;
1934         __be32          *p = req->rq_svec[0].iov_base;
1935
1936         /* FIXME: check buffer size? */
1937
1938         p = xprt_skip_transport_header(task->tk_xprt, p);
1939         *p++ = req->rq_xid;             /* XID */
1940         *p++ = htonl(RPC_CALL);         /* CALL */
1941         *p++ = htonl(RPC_VERSION);      /* RPC version */
1942         *p++ = htonl(clnt->cl_prog);    /* program number */
1943         *p++ = htonl(clnt->cl_vers);    /* program version */
1944         *p++ = htonl(task->tk_msg.rpc_proc->p_proc);    /* procedure */
1945         p = rpcauth_marshcred(task, p);
1946         req->rq_slen = xdr_adjust_iovec(&req->rq_svec[0], p);
1947         return p;
1948 }
1949
1950 static __be32 *
1951 rpc_verify_header(struct rpc_task *task)
1952 {
1953         struct rpc_clnt *clnt = task->tk_client;
1954         struct kvec *iov = &task->tk_rqstp->rq_rcv_buf.head[0];
1955         int len = task->tk_rqstp->rq_rcv_buf.len >> 2;
1956         __be32  *p = iov->iov_base;
1957         u32 n;
1958         int error = -EACCES;
1959
1960         if ((task->tk_rqstp->rq_rcv_buf.len & 3) != 0) {
1961                 /* RFC-1014 says that the representation of XDR data must be a
1962                  * multiple of four bytes
1963                  * - if it isn't pointer subtraction in the NFS client may give
1964                  *   undefined results
1965                  */
1966                 dprintk("RPC: %5u %s: XDR representation not a multiple of"
1967                        " 4 bytes: 0x%x\n", task->tk_pid, __func__,
1968                        task->tk_rqstp->rq_rcv_buf.len);
1969                 goto out_eio;
1970         }
1971         if ((len -= 3) < 0)
1972                 goto out_overflow;
1973
1974         p += 1; /* skip XID */
1975         if ((n = ntohl(*p++)) != RPC_REPLY) {
1976                 dprintk("RPC: %5u %s: not an RPC reply: %x\n",
1977                         task->tk_pid, __func__, n);
1978                 goto out_garbage;
1979         }
1980
1981         if ((n = ntohl(*p++)) != RPC_MSG_ACCEPTED) {
1982                 if (--len < 0)
1983                         goto out_overflow;
1984                 switch ((n = ntohl(*p++))) {
1985                 case RPC_AUTH_ERROR:
1986                         break;
1987                 case RPC_MISMATCH:
1988                         dprintk("RPC: %5u %s: RPC call version mismatch!\n",
1989                                 task->tk_pid, __func__);
1990                         error = -EPROTONOSUPPORT;
1991                         goto out_err;
1992                 default:
1993                         dprintk("RPC: %5u %s: RPC call rejected, "
1994                                 "unknown error: %x\n",
1995                                 task->tk_pid, __func__, n);
1996                         goto out_eio;
1997                 }
1998                 if (--len < 0)
1999                         goto out_overflow;
2000                 switch ((n = ntohl(*p++))) {
2001                 case RPC_AUTH_REJECTEDCRED:
2002                 case RPC_AUTH_REJECTEDVERF:
2003                 case RPCSEC_GSS_CREDPROBLEM:
2004                 case RPCSEC_GSS_CTXPROBLEM:
2005                         if (!task->tk_cred_retry)
2006                                 break;
2007                         task->tk_cred_retry--;
2008                         dprintk("RPC: %5u %s: retry stale creds\n",
2009                                         task->tk_pid, __func__);
2010                         rpcauth_invalcred(task);
2011                         /* Ensure we obtain a new XID! */
2012                         xprt_release(task);
2013                         task->tk_action = call_reserve;
2014                         goto out_retry;
2015                 case RPC_AUTH_BADCRED:
2016                 case RPC_AUTH_BADVERF:
2017                         /* possibly garbled cred/verf? */
2018                         if (!task->tk_garb_retry)
2019                                 break;
2020                         task->tk_garb_retry--;
2021                         dprintk("RPC: %5u %s: retry garbled creds\n",
2022                                         task->tk_pid, __func__);
2023                         task->tk_action = call_bind;
2024                         goto out_retry;
2025                 case RPC_AUTH_TOOWEAK:
2026                         rcu_read_lock();
2027                         printk(KERN_NOTICE "RPC: server %s requires stronger "
2028                                "authentication.\n",
2029                                rcu_dereference(clnt->cl_xprt)->servername);
2030                         rcu_read_unlock();
2031                         break;
2032                 default:
2033                         dprintk("RPC: %5u %s: unknown auth error: %x\n",
2034                                         task->tk_pid, __func__, n);
2035                         error = -EIO;
2036                 }
2037                 dprintk("RPC: %5u %s: call rejected %d\n",
2038                                 task->tk_pid, __func__, n);
2039                 goto out_err;
2040         }
2041         if (!(p = rpcauth_checkverf(task, p))) {
2042                 dprintk("RPC: %5u %s: auth check failed\n",
2043                                 task->tk_pid, __func__);
2044                 goto out_garbage;               /* bad verifier, retry */
2045         }
2046         len = p - (__be32 *)iov->iov_base - 1;
2047         if (len < 0)
2048                 goto out_overflow;
2049         switch ((n = ntohl(*p++))) {
2050         case RPC_SUCCESS:
2051                 return p;
2052         case RPC_PROG_UNAVAIL:
2053                 dprintk_rcu("RPC: %5u %s: program %u is unsupported "
2054                                 "by server %s\n", task->tk_pid, __func__,
2055                                 (unsigned int)clnt->cl_prog,
2056                                 rcu_dereference(clnt->cl_xprt)->servername);
2057                 error = -EPFNOSUPPORT;
2058                 goto out_err;
2059         case RPC_PROG_MISMATCH:
2060                 dprintk_rcu("RPC: %5u %s: program %u, version %u unsupported "
2061                                 "by server %s\n", task->tk_pid, __func__,
2062                                 (unsigned int)clnt->cl_prog,
2063                                 (unsigned int)clnt->cl_vers,
2064                                 rcu_dereference(clnt->cl_xprt)->servername);
2065                 error = -EPROTONOSUPPORT;
2066                 goto out_err;
2067         case RPC_PROC_UNAVAIL:
2068                 dprintk_rcu("RPC: %5u %s: proc %s unsupported by program %u, "
2069                                 "version %u on server %s\n",
2070                                 task->tk_pid, __func__,
2071                                 rpc_proc_name(task),
2072                                 clnt->cl_prog, clnt->cl_vers,
2073                                 rcu_dereference(clnt->cl_xprt)->servername);
2074                 error = -EOPNOTSUPP;
2075                 goto out_err;
2076         case RPC_GARBAGE_ARGS:
2077                 dprintk("RPC: %5u %s: server saw garbage\n",
2078                                 task->tk_pid, __func__);
2079                 break;                  /* retry */
2080         default:
2081                 dprintk("RPC: %5u %s: server accept status: %x\n",
2082                                 task->tk_pid, __func__, n);
2083                 /* Also retry */
2084         }
2085
2086 out_garbage:
2087         clnt->cl_stats->rpcgarbage++;
2088         if (task->tk_garb_retry) {
2089                 task->tk_garb_retry--;
2090                 dprintk("RPC: %5u %s: retrying\n",
2091                                 task->tk_pid, __func__);
2092                 task->tk_action = call_bind;
2093 out_retry:
2094                 return ERR_PTR(-EAGAIN);
2095         }
2096 out_eio:
2097         error = -EIO;
2098 out_err:
2099         rpc_exit(task, error);
2100         dprintk("RPC: %5u %s: call failed with error %d\n", task->tk_pid,
2101                         __func__, error);
2102         return ERR_PTR(error);
2103 out_overflow:
2104         dprintk("RPC: %5u %s: server reply was truncated.\n", task->tk_pid,
2105                         __func__);
2106         goto out_garbage;
2107 }
2108
2109 static void rpcproc_encode_null(void *rqstp, struct xdr_stream *xdr, void *obj)
2110 {
2111 }
2112
2113 static int rpcproc_decode_null(void *rqstp, struct xdr_stream *xdr, void *obj)
2114 {
2115         return 0;
2116 }
2117
2118 static struct rpc_procinfo rpcproc_null = {
2119         .p_encode = rpcproc_encode_null,
2120         .p_decode = rpcproc_decode_null,
2121 };
2122
2123 static int rpc_ping(struct rpc_clnt *clnt)
2124 {
2125         struct rpc_message msg = {
2126                 .rpc_proc = &rpcproc_null,
2127         };
2128         int err;
2129         msg.rpc_cred = authnull_ops.lookup_cred(NULL, NULL, 0);
2130         err = rpc_call_sync(clnt, &msg, RPC_TASK_SOFT | RPC_TASK_SOFTCONN);
2131         put_rpccred(msg.rpc_cred);
2132         return err;
2133 }
2134
2135 struct rpc_task *rpc_call_null(struct rpc_clnt *clnt, struct rpc_cred *cred, int flags)
2136 {
2137         struct rpc_message msg = {
2138                 .rpc_proc = &rpcproc_null,
2139                 .rpc_cred = cred,
2140         };
2141         struct rpc_task_setup task_setup_data = {
2142                 .rpc_client = clnt,
2143                 .rpc_message = &msg,
2144                 .callback_ops = &rpc_default_ops,
2145                 .flags = flags,
2146         };
2147         return rpc_run_task(&task_setup_data);
2148 }
2149 EXPORT_SYMBOL_GPL(rpc_call_null);
2150
2151 #ifdef RPC_DEBUG
2152 static void rpc_show_header(void)
2153 {
2154         printk(KERN_INFO "-pid- flgs status -client- --rqstp- "
2155                 "-timeout ---ops--\n");
2156 }
2157
2158 static void rpc_show_task(const struct rpc_clnt *clnt,
2159                           const struct rpc_task *task)
2160 {
2161         const char *rpc_waitq = "none";
2162
2163         if (RPC_IS_QUEUED(task))
2164                 rpc_waitq = rpc_qname(task->tk_waitqueue);
2165
2166         printk(KERN_INFO "%5u %04x %6d %8p %8p %8ld %8p %sv%u %s a:%ps q:%s\n",
2167                 task->tk_pid, task->tk_flags, task->tk_status,
2168                 clnt, task->tk_rqstp, task->tk_timeout, task->tk_ops,
2169                 clnt->cl_protname, clnt->cl_vers, rpc_proc_name(task),
2170                 task->tk_action, rpc_waitq);
2171 }
2172
2173 void rpc_show_tasks(struct net *net)
2174 {
2175         struct rpc_clnt *clnt;
2176         struct rpc_task *task;
2177         int header = 0;
2178         struct sunrpc_net *sn = net_generic(net, sunrpc_net_id);
2179
2180         spin_lock(&sn->rpc_client_lock);
2181         list_for_each_entry(clnt, &sn->all_clients, cl_clients) {
2182                 spin_lock(&clnt->cl_lock);
2183                 list_for_each_entry(task, &clnt->cl_tasks, tk_task) {
2184                         if (!header) {
2185                                 rpc_show_header();
2186                                 header++;
2187                         }
2188                         rpc_show_task(clnt, task);
2189                 }
2190                 spin_unlock(&clnt->cl_lock);
2191         }
2192         spin_unlock(&sn->rpc_client_lock);
2193 }
2194 #endif