Merge /pub/scm/linux/kernel/git/torvalds/linux-2.6
[linux-2.6] / 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  *  NB: BSD uses a more intelligent approach to guessing when a request
17  *  or reply has been lost by keeping the RTO estimate for each procedure.
18  *  We currently make do with a constant timeout value.
19  *
20  *  Copyright (C) 1992,1993 Rick Sladkey <jrs@world.std.com>
21  *  Copyright (C) 1995,1996 Olaf Kirch <okir@monad.swb.de>
22  */
23
24 #include <asm/system.h>
25
26 #include <linux/module.h>
27 #include <linux/types.h>
28 #include <linux/mm.h>
29 #include <linux/slab.h>
30 #include <linux/smp_lock.h>
31 #include <linux/utsname.h>
32 #include <linux/workqueue.h>
33
34 #include <linux/sunrpc/clnt.h>
35 #include <linux/sunrpc/rpc_pipe_fs.h>
36 #include <linux/sunrpc/metrics.h>
37
38
39 #define RPC_SLACK_SPACE         (1024)  /* total overkill */
40
41 #ifdef RPC_DEBUG
42 # define RPCDBG_FACILITY        RPCDBG_CALL
43 #endif
44
45 #define dprint_status(t)                                        \
46         dprintk("RPC: %5u %s (status %d)\n", t->tk_pid,         \
47                         __FUNCTION__, t->tk_status)
48
49 static DECLARE_WAIT_QUEUE_HEAD(destroy_wait);
50
51
52 static void     call_start(struct rpc_task *task);
53 static void     call_reserve(struct rpc_task *task);
54 static void     call_reserveresult(struct rpc_task *task);
55 static void     call_allocate(struct rpc_task *task);
56 static void     call_encode(struct rpc_task *task);
57 static void     call_decode(struct rpc_task *task);
58 static void     call_bind(struct rpc_task *task);
59 static void     call_bind_status(struct rpc_task *task);
60 static void     call_transmit(struct rpc_task *task);
61 static void     call_status(struct rpc_task *task);
62 static void     call_transmit_status(struct rpc_task *task);
63 static void     call_refresh(struct rpc_task *task);
64 static void     call_refreshresult(struct rpc_task *task);
65 static void     call_timeout(struct rpc_task *task);
66 static void     call_connect(struct rpc_task *task);
67 static void     call_connect_status(struct rpc_task *task);
68 static __be32 * call_header(struct rpc_task *task);
69 static __be32 * call_verify(struct rpc_task *task);
70
71
72 static int
73 rpc_setup_pipedir(struct rpc_clnt *clnt, char *dir_name)
74 {
75         static uint32_t clntid;
76         int error;
77
78         clnt->cl_vfsmnt = ERR_PTR(-ENOENT);
79         clnt->cl_dentry = ERR_PTR(-ENOENT);
80         if (dir_name == NULL)
81                 return 0;
82
83         clnt->cl_vfsmnt = rpc_get_mount();
84         if (IS_ERR(clnt->cl_vfsmnt))
85                 return PTR_ERR(clnt->cl_vfsmnt);
86
87         for (;;) {
88                 snprintf(clnt->cl_pathname, sizeof(clnt->cl_pathname),
89                                 "%s/clnt%x", dir_name,
90                                 (unsigned int)clntid++);
91                 clnt->cl_pathname[sizeof(clnt->cl_pathname) - 1] = '\0';
92                 clnt->cl_dentry = rpc_mkdir(clnt->cl_pathname, clnt);
93                 if (!IS_ERR(clnt->cl_dentry))
94                         return 0;
95                 error = PTR_ERR(clnt->cl_dentry);
96                 if (error != -EEXIST) {
97                         printk(KERN_INFO "RPC: Couldn't create pipefs entry %s, error %d\n",
98                                         clnt->cl_pathname, error);
99                         rpc_put_mount();
100                         return error;
101                 }
102         }
103 }
104
105 static struct rpc_clnt * rpc_new_client(struct rpc_xprt *xprt, char *servname, struct rpc_program *program, u32 vers, rpc_authflavor_t flavor)
106 {
107         struct rpc_version      *version;
108         struct rpc_clnt         *clnt = NULL;
109         struct rpc_auth         *auth;
110         int err;
111         int len;
112
113         dprintk("RPC:       creating %s client for %s (xprt %p)\n",
114                         program->name, servname, xprt);
115
116         err = -EINVAL;
117         if (!xprt)
118                 goto out_no_xprt;
119         if (vers >= program->nrvers || !(version = program->version[vers]))
120                 goto out_err;
121
122         err = -ENOMEM;
123         clnt = kzalloc(sizeof(*clnt), GFP_KERNEL);
124         if (!clnt)
125                 goto out_err;
126         atomic_set(&clnt->cl_users, 0);
127         atomic_set(&clnt->cl_count, 1);
128         clnt->cl_parent = clnt;
129
130         clnt->cl_server = clnt->cl_inline_name;
131         len = strlen(servname) + 1;
132         if (len > sizeof(clnt->cl_inline_name)) {
133                 char *buf = kmalloc(len, GFP_KERNEL);
134                 if (buf != 0)
135                         clnt->cl_server = buf;
136                 else
137                         len = sizeof(clnt->cl_inline_name);
138         }
139         strlcpy(clnt->cl_server, servname, len);
140
141         clnt->cl_xprt     = xprt;
142         clnt->cl_procinfo = version->procs;
143         clnt->cl_maxproc  = version->nrprocs;
144         clnt->cl_protname = program->name;
145         clnt->cl_prog     = program->number;
146         clnt->cl_vers     = version->number;
147         clnt->cl_stats    = program->stats;
148         clnt->cl_metrics  = rpc_alloc_iostats(clnt);
149         err = -ENOMEM;
150         if (clnt->cl_metrics == NULL)
151                 goto out_no_stats;
152         clnt->cl_program  = program;
153
154         if (!xprt_bound(clnt->cl_xprt))
155                 clnt->cl_autobind = 1;
156
157         clnt->cl_rtt = &clnt->cl_rtt_default;
158         rpc_init_rtt(&clnt->cl_rtt_default, xprt->timeout.to_initval);
159
160         err = rpc_setup_pipedir(clnt, program->pipe_dir_name);
161         if (err < 0)
162                 goto out_no_path;
163
164         auth = rpcauth_create(flavor, clnt);
165         if (IS_ERR(auth)) {
166                 printk(KERN_INFO "RPC: Couldn't create auth handle (flavor %u)\n",
167                                 flavor);
168                 err = PTR_ERR(auth);
169                 goto out_no_auth;
170         }
171
172         /* save the nodename */
173         clnt->cl_nodelen = strlen(utsname()->nodename);
174         if (clnt->cl_nodelen > UNX_MAXNODENAME)
175                 clnt->cl_nodelen = UNX_MAXNODENAME;
176         memcpy(clnt->cl_nodename, utsname()->nodename, clnt->cl_nodelen);
177         return clnt;
178
179 out_no_auth:
180         if (!IS_ERR(clnt->cl_dentry)) {
181                 rpc_rmdir(clnt->cl_dentry);
182                 rpc_put_mount();
183         }
184 out_no_path:
185         rpc_free_iostats(clnt->cl_metrics);
186 out_no_stats:
187         if (clnt->cl_server != clnt->cl_inline_name)
188                 kfree(clnt->cl_server);
189         kfree(clnt);
190 out_err:
191         xprt_put(xprt);
192 out_no_xprt:
193         return ERR_PTR(err);
194 }
195
196 /*
197  * rpc_create - create an RPC client and transport with one call
198  * @args: rpc_clnt create argument structure
199  *
200  * Creates and initializes an RPC transport and an RPC client.
201  *
202  * It can ping the server in order to determine if it is up, and to see if
203  * it supports this program and version.  RPC_CLNT_CREATE_NOPING disables
204  * this behavior so asynchronous tasks can also use rpc_create.
205  */
206 struct rpc_clnt *rpc_create(struct rpc_create_args *args)
207 {
208         struct rpc_xprt *xprt;
209         struct rpc_clnt *clnt;
210
211         xprt = xprt_create_transport(args->protocol, args->address,
212                                         args->addrsize, args->timeout);
213         if (IS_ERR(xprt))
214                 return (struct rpc_clnt *)xprt;
215
216         /*
217          * By default, kernel RPC client connects from a reserved port.
218          * CAP_NET_BIND_SERVICE will not be set for unprivileged requesters,
219          * but it is always enabled for rpciod, which handles the connect
220          * operation.
221          */
222         xprt->resvport = 1;
223         if (args->flags & RPC_CLNT_CREATE_NONPRIVPORT)
224                 xprt->resvport = 0;
225
226         dprintk("RPC:       creating %s client for %s (xprt %p)\n",
227                         args->program->name, args->servername, xprt);
228
229         clnt = rpc_new_client(xprt, args->servername, args->program,
230                                 args->version, args->authflavor);
231         if (IS_ERR(clnt))
232                 return clnt;
233
234         if (!(args->flags & RPC_CLNT_CREATE_NOPING)) {
235                 int err = rpc_ping(clnt, RPC_TASK_SOFT|RPC_TASK_NOINTR);
236                 if (err != 0) {
237                         rpc_shutdown_client(clnt);
238                         return ERR_PTR(err);
239                 }
240         }
241
242         clnt->cl_softrtry = 1;
243         if (args->flags & RPC_CLNT_CREATE_HARDRTRY)
244                 clnt->cl_softrtry = 0;
245
246         if (args->flags & RPC_CLNT_CREATE_INTR)
247                 clnt->cl_intr = 1;
248         if (args->flags & RPC_CLNT_CREATE_AUTOBIND)
249                 clnt->cl_autobind = 1;
250         if (args->flags & RPC_CLNT_CREATE_ONESHOT)
251                 clnt->cl_oneshot = 1;
252         if (args->flags & RPC_CLNT_CREATE_DISCRTRY)
253                 clnt->cl_discrtry = 1;
254
255         return clnt;
256 }
257 EXPORT_SYMBOL_GPL(rpc_create);
258
259 /*
260  * This function clones the RPC client structure. It allows us to share the
261  * same transport while varying parameters such as the authentication
262  * flavour.
263  */
264 struct rpc_clnt *
265 rpc_clone_client(struct rpc_clnt *clnt)
266 {
267         struct rpc_clnt *new;
268         int err = -ENOMEM;
269
270         new = kmemdup(clnt, sizeof(*new), GFP_KERNEL);
271         if (!new)
272                 goto out_no_clnt;
273         atomic_set(&new->cl_count, 1);
274         atomic_set(&new->cl_users, 0);
275         new->cl_metrics = rpc_alloc_iostats(clnt);
276         if (new->cl_metrics == NULL)
277                 goto out_no_stats;
278         err = rpc_setup_pipedir(new, clnt->cl_program->pipe_dir_name);
279         if (err != 0)
280                 goto out_no_path;
281         new->cl_parent = clnt;
282         atomic_inc(&clnt->cl_count);
283         new->cl_xprt = xprt_get(clnt->cl_xprt);
284         /* Turn off autobind on clones */
285         new->cl_autobind = 0;
286         new->cl_oneshot = 0;
287         new->cl_dead = 0;
288         rpc_init_rtt(&new->cl_rtt_default, clnt->cl_xprt->timeout.to_initval);
289         if (new->cl_auth)
290                 atomic_inc(&new->cl_auth->au_count);
291         return new;
292 out_no_path:
293         rpc_free_iostats(new->cl_metrics);
294 out_no_stats:
295         kfree(new);
296 out_no_clnt:
297         dprintk("RPC:       %s: returned error %d\n", __FUNCTION__, err);
298         return ERR_PTR(err);
299 }
300
301 /*
302  * Properly shut down an RPC client, terminating all outstanding
303  * requests. Note that we must be certain that cl_oneshot and
304  * cl_dead are cleared, or else the client would be destroyed
305  * when the last task releases it.
306  */
307 int
308 rpc_shutdown_client(struct rpc_clnt *clnt)
309 {
310         dprintk("RPC:       shutting down %s client for %s, tasks=%d\n",
311                         clnt->cl_protname, clnt->cl_server,
312                         atomic_read(&clnt->cl_users));
313
314         while (atomic_read(&clnt->cl_users) > 0) {
315                 /* Don't let rpc_release_client destroy us */
316                 clnt->cl_oneshot = 0;
317                 clnt->cl_dead = 0;
318                 rpc_killall_tasks(clnt);
319                 wait_event_timeout(destroy_wait,
320                         !atomic_read(&clnt->cl_users), 1*HZ);
321         }
322
323         if (atomic_read(&clnt->cl_users) < 0) {
324                 printk(KERN_ERR "RPC: rpc_shutdown_client clnt %p tasks=%d\n",
325                                 clnt, atomic_read(&clnt->cl_users));
326 #ifdef RPC_DEBUG
327                 rpc_show_tasks();
328 #endif
329                 BUG();
330         }
331
332         return rpc_destroy_client(clnt);
333 }
334
335 /*
336  * Delete an RPC client
337  */
338 int
339 rpc_destroy_client(struct rpc_clnt *clnt)
340 {
341         if (!atomic_dec_and_test(&clnt->cl_count))
342                 return 1;
343         BUG_ON(atomic_read(&clnt->cl_users) != 0);
344
345         dprintk("RPC:       destroying %s client for %s\n",
346                         clnt->cl_protname, clnt->cl_server);
347         if (clnt->cl_auth) {
348                 rpcauth_destroy(clnt->cl_auth);
349                 clnt->cl_auth = NULL;
350         }
351         if (!IS_ERR(clnt->cl_dentry)) {
352                 rpc_rmdir(clnt->cl_dentry);
353                 rpc_put_mount();
354         }
355         if (clnt->cl_parent != clnt) {
356                 rpc_destroy_client(clnt->cl_parent);
357                 goto out_free;
358         }
359         if (clnt->cl_server != clnt->cl_inline_name)
360                 kfree(clnt->cl_server);
361 out_free:
362         rpc_free_iostats(clnt->cl_metrics);
363         clnt->cl_metrics = NULL;
364         xprt_put(clnt->cl_xprt);
365         kfree(clnt);
366         return 0;
367 }
368
369 /*
370  * Release an RPC client
371  */
372 void
373 rpc_release_client(struct rpc_clnt *clnt)
374 {
375         dprintk("RPC:       rpc_release_client(%p, %d)\n",
376                         clnt, atomic_read(&clnt->cl_users));
377
378         if (!atomic_dec_and_test(&clnt->cl_users))
379                 return;
380         wake_up(&destroy_wait);
381         if (clnt->cl_oneshot || clnt->cl_dead)
382                 rpc_destroy_client(clnt);
383 }
384
385 /**
386  * rpc_bind_new_program - bind a new RPC program to an existing client
387  * @old - old rpc_client
388  * @program - rpc program to set
389  * @vers - rpc program version
390  *
391  * Clones the rpc client and sets up a new RPC program. This is mainly
392  * of use for enabling different RPC programs to share the same transport.
393  * The Sun NFSv2/v3 ACL protocol can do this.
394  */
395 struct rpc_clnt *rpc_bind_new_program(struct rpc_clnt *old,
396                                       struct rpc_program *program,
397                                       int vers)
398 {
399         struct rpc_clnt *clnt;
400         struct rpc_version *version;
401         int err;
402
403         BUG_ON(vers >= program->nrvers || !program->version[vers]);
404         version = program->version[vers];
405         clnt = rpc_clone_client(old);
406         if (IS_ERR(clnt))
407                 goto out;
408         clnt->cl_procinfo = version->procs;
409         clnt->cl_maxproc  = version->nrprocs;
410         clnt->cl_protname = program->name;
411         clnt->cl_prog     = program->number;
412         clnt->cl_vers     = version->number;
413         clnt->cl_stats    = program->stats;
414         err = rpc_ping(clnt, RPC_TASK_SOFT|RPC_TASK_NOINTR);
415         if (err != 0) {
416                 rpc_shutdown_client(clnt);
417                 clnt = ERR_PTR(err);
418         }
419 out:
420         return clnt;
421 }
422
423 /*
424  * Default callback for async RPC calls
425  */
426 static void
427 rpc_default_callback(struct rpc_task *task, void *data)
428 {
429 }
430
431 static const struct rpc_call_ops rpc_default_ops = {
432         .rpc_call_done = rpc_default_callback,
433 };
434
435 /*
436  *      Export the signal mask handling for synchronous code that
437  *      sleeps on RPC calls
438  */
439 #define RPC_INTR_SIGNALS (sigmask(SIGHUP) | sigmask(SIGINT) | sigmask(SIGQUIT) | sigmask(SIGTERM))
440
441 static void rpc_save_sigmask(sigset_t *oldset, int intr)
442 {
443         unsigned long   sigallow = sigmask(SIGKILL);
444         sigset_t sigmask;
445
446         /* Block all signals except those listed in sigallow */
447         if (intr)
448                 sigallow |= RPC_INTR_SIGNALS;
449         siginitsetinv(&sigmask, sigallow);
450         sigprocmask(SIG_BLOCK, &sigmask, oldset);
451 }
452
453 static inline void rpc_task_sigmask(struct rpc_task *task, sigset_t *oldset)
454 {
455         rpc_save_sigmask(oldset, !RPC_TASK_UNINTERRUPTIBLE(task));
456 }
457
458 static inline void rpc_restore_sigmask(sigset_t *oldset)
459 {
460         sigprocmask(SIG_SETMASK, oldset, NULL);
461 }
462
463 void rpc_clnt_sigmask(struct rpc_clnt *clnt, sigset_t *oldset)
464 {
465         rpc_save_sigmask(oldset, clnt->cl_intr);
466 }
467
468 void rpc_clnt_sigunmask(struct rpc_clnt *clnt, sigset_t *oldset)
469 {
470         rpc_restore_sigmask(oldset);
471 }
472
473 /*
474  * New rpc_call implementation
475  */
476 int rpc_call_sync(struct rpc_clnt *clnt, struct rpc_message *msg, int flags)
477 {
478         struct rpc_task *task;
479         sigset_t        oldset;
480         int             status;
481
482         /* If this client is slain all further I/O fails */
483         if (clnt->cl_dead)
484                 return -EIO;
485
486         BUG_ON(flags & RPC_TASK_ASYNC);
487
488         task = rpc_new_task(clnt, flags, &rpc_default_ops, NULL);
489         if (task == NULL)
490                 return -ENOMEM;
491
492         /* Mask signals on RPC calls _and_ GSS_AUTH upcalls */
493         rpc_task_sigmask(task, &oldset);
494
495         /* Set up the call info struct and execute the task */
496         rpc_call_setup(task, msg, 0);
497         if (task->tk_status == 0) {
498                 atomic_inc(&task->tk_count);
499                 rpc_execute(task);
500         }
501         status = task->tk_status;
502         rpc_put_task(task);
503         rpc_restore_sigmask(&oldset);
504         return status;
505 }
506
507 /*
508  * New rpc_call implementation
509  */
510 int
511 rpc_call_async(struct rpc_clnt *clnt, struct rpc_message *msg, int flags,
512                const struct rpc_call_ops *tk_ops, void *data)
513 {
514         struct rpc_task *task;
515         sigset_t        oldset;
516         int             status;
517
518         /* If this client is slain all further I/O fails */
519         status = -EIO;
520         if (clnt->cl_dead)
521                 goto out_release;
522
523         flags |= RPC_TASK_ASYNC;
524
525         /* Create/initialize a new RPC task */
526         status = -ENOMEM;
527         if (!(task = rpc_new_task(clnt, flags, tk_ops, data)))
528                 goto out_release;
529
530         /* Mask signals on GSS_AUTH upcalls */
531         rpc_task_sigmask(task, &oldset);
532
533         rpc_call_setup(task, msg, 0);
534
535         /* Set up the call info struct and execute the task */
536         status = task->tk_status;
537         if (status == 0)
538                 rpc_execute(task);
539         else
540                 rpc_put_task(task);
541
542         rpc_restore_sigmask(&oldset);
543         return status;
544 out_release:
545         rpc_release_calldata(tk_ops, data);
546         return status;
547 }
548
549
550 void
551 rpc_call_setup(struct rpc_task *task, struct rpc_message *msg, int flags)
552 {
553         task->tk_msg   = *msg;
554         task->tk_flags |= flags;
555         /* Bind the user cred */
556         if (task->tk_msg.rpc_cred != NULL)
557                 rpcauth_holdcred(task);
558         else
559                 rpcauth_bindcred(task);
560
561         if (task->tk_status == 0)
562                 task->tk_action = call_start;
563         else
564                 task->tk_action = rpc_exit_task;
565 }
566
567 /**
568  * rpc_peeraddr - extract remote peer address from clnt's xprt
569  * @clnt: RPC client structure
570  * @buf: target buffer
571  * @size: length of target buffer
572  *
573  * Returns the number of bytes that are actually in the stored address.
574  */
575 size_t rpc_peeraddr(struct rpc_clnt *clnt, struct sockaddr *buf, size_t bufsize)
576 {
577         size_t bytes;
578         struct rpc_xprt *xprt = clnt->cl_xprt;
579
580         bytes = sizeof(xprt->addr);
581         if (bytes > bufsize)
582                 bytes = bufsize;
583         memcpy(buf, &clnt->cl_xprt->addr, bytes);
584         return xprt->addrlen;
585 }
586 EXPORT_SYMBOL_GPL(rpc_peeraddr);
587
588 /**
589  * rpc_peeraddr2str - return remote peer address in printable format
590  * @clnt: RPC client structure
591  * @format: address format
592  *
593  */
594 char *rpc_peeraddr2str(struct rpc_clnt *clnt, enum rpc_display_format_t format)
595 {
596         struct rpc_xprt *xprt = clnt->cl_xprt;
597
598         if (xprt->address_strings[format] != NULL)
599                 return xprt->address_strings[format];
600         else
601                 return "unprintable";
602 }
603 EXPORT_SYMBOL_GPL(rpc_peeraddr2str);
604
605 void
606 rpc_setbufsize(struct rpc_clnt *clnt, unsigned int sndsize, unsigned int rcvsize)
607 {
608         struct rpc_xprt *xprt = clnt->cl_xprt;
609         if (xprt->ops->set_buffer_size)
610                 xprt->ops->set_buffer_size(xprt, sndsize, rcvsize);
611 }
612
613 /*
614  * Return size of largest payload RPC client can support, in bytes
615  *
616  * For stream transports, this is one RPC record fragment (see RFC
617  * 1831), as we don't support multi-record requests yet.  For datagram
618  * transports, this is the size of an IP packet minus the IP, UDP, and
619  * RPC header sizes.
620  */
621 size_t rpc_max_payload(struct rpc_clnt *clnt)
622 {
623         return clnt->cl_xprt->max_payload;
624 }
625 EXPORT_SYMBOL_GPL(rpc_max_payload);
626
627 /**
628  * rpc_force_rebind - force transport to check that remote port is unchanged
629  * @clnt: client to rebind
630  *
631  */
632 void rpc_force_rebind(struct rpc_clnt *clnt)
633 {
634         if (clnt->cl_autobind)
635                 xprt_clear_bound(clnt->cl_xprt);
636 }
637 EXPORT_SYMBOL_GPL(rpc_force_rebind);
638
639 /*
640  * Restart an (async) RPC call. Usually called from within the
641  * exit handler.
642  */
643 void
644 rpc_restart_call(struct rpc_task *task)
645 {
646         if (RPC_ASSASSINATED(task))
647                 return;
648
649         task->tk_action = call_start;
650 }
651
652 /*
653  * 0.  Initial state
654  *
655  *     Other FSM states can be visited zero or more times, but
656  *     this state is visited exactly once for each RPC.
657  */
658 static void
659 call_start(struct rpc_task *task)
660 {
661         struct rpc_clnt *clnt = task->tk_client;
662
663         dprintk("RPC: %5u call_start %s%d proc %d (%s)\n", task->tk_pid,
664                         clnt->cl_protname, clnt->cl_vers,
665                         task->tk_msg.rpc_proc->p_proc,
666                         (RPC_IS_ASYNC(task) ? "async" : "sync"));
667
668         /* Increment call count */
669         task->tk_msg.rpc_proc->p_count++;
670         clnt->cl_stats->rpccnt++;
671         task->tk_action = call_reserve;
672 }
673
674 /*
675  * 1.   Reserve an RPC call slot
676  */
677 static void
678 call_reserve(struct rpc_task *task)
679 {
680         dprint_status(task);
681
682         if (!rpcauth_uptodatecred(task)) {
683                 task->tk_action = call_refresh;
684                 return;
685         }
686
687         task->tk_status  = 0;
688         task->tk_action  = call_reserveresult;
689         xprt_reserve(task);
690 }
691
692 /*
693  * 1b.  Grok the result of xprt_reserve()
694  */
695 static void
696 call_reserveresult(struct rpc_task *task)
697 {
698         int status = task->tk_status;
699
700         dprint_status(task);
701
702         /*
703          * After a call to xprt_reserve(), we must have either
704          * a request slot or else an error status.
705          */
706         task->tk_status = 0;
707         if (status >= 0) {
708                 if (task->tk_rqstp) {
709                         task->tk_action = call_allocate;
710                         return;
711                 }
712
713                 printk(KERN_ERR "%s: status=%d, but no request slot, exiting\n",
714                                 __FUNCTION__, status);
715                 rpc_exit(task, -EIO);
716                 return;
717         }
718
719         /*
720          * Even though there was an error, we may have acquired
721          * a request slot somehow.  Make sure not to leak it.
722          */
723         if (task->tk_rqstp) {
724                 printk(KERN_ERR "%s: status=%d, request allocated anyway\n",
725                                 __FUNCTION__, status);
726                 xprt_release(task);
727         }
728
729         switch (status) {
730         case -EAGAIN:   /* woken up; retry */
731                 task->tk_action = call_reserve;
732                 return;
733         case -EIO:      /* probably a shutdown */
734                 break;
735         default:
736                 printk(KERN_ERR "%s: unrecognized error %d, exiting\n",
737                                 __FUNCTION__, status);
738                 break;
739         }
740         rpc_exit(task, status);
741 }
742
743 /*
744  * 2.   Allocate the buffer. For details, see sched.c:rpc_malloc.
745  *      (Note: buffer memory is freed in xprt_release).
746  */
747 static void
748 call_allocate(struct rpc_task *task)
749 {
750         struct rpc_rqst *req = task->tk_rqstp;
751         struct rpc_xprt *xprt = task->tk_xprt;
752         unsigned int    bufsiz;
753
754         dprint_status(task);
755
756         task->tk_action = call_bind;
757         if (req->rq_buffer)
758                 return;
759
760         /* FIXME: compute buffer requirements more exactly using
761          * auth->au_wslack */
762         bufsiz = task->tk_msg.rpc_proc->p_bufsiz + RPC_SLACK_SPACE;
763
764         if (xprt->ops->buf_alloc(task, bufsiz << 1) != NULL)
765                 return;
766
767         dprintk("RPC: %5u rpc_buffer allocation failed\n", task->tk_pid);
768
769         if (RPC_IS_ASYNC(task) || !signalled()) {
770                 xprt_release(task);
771                 task->tk_action = call_reserve;
772                 rpc_delay(task, HZ>>4);
773                 return;
774         }
775
776         rpc_exit(task, -ERESTARTSYS);
777 }
778
779 static inline int
780 rpc_task_need_encode(struct rpc_task *task)
781 {
782         return task->tk_rqstp->rq_snd_buf.len == 0;
783 }
784
785 static inline void
786 rpc_task_force_reencode(struct rpc_task *task)
787 {
788         task->tk_rqstp->rq_snd_buf.len = 0;
789 }
790
791 /*
792  * 3.   Encode arguments of an RPC call
793  */
794 static void
795 call_encode(struct rpc_task *task)
796 {
797         struct rpc_rqst *req = task->tk_rqstp;
798         struct xdr_buf *sndbuf = &req->rq_snd_buf;
799         struct xdr_buf *rcvbuf = &req->rq_rcv_buf;
800         unsigned int    bufsiz;
801         kxdrproc_t      encode;
802         __be32          *p;
803
804         dprint_status(task);
805
806         /* Default buffer setup */
807         bufsiz = req->rq_bufsize >> 1;
808         sndbuf->head[0].iov_base = (void *)req->rq_buffer;
809         sndbuf->head[0].iov_len  = bufsiz;
810         sndbuf->tail[0].iov_len  = 0;
811         sndbuf->page_len         = 0;
812         sndbuf->len              = 0;
813         sndbuf->buflen           = bufsiz;
814         rcvbuf->head[0].iov_base = (void *)((char *)req->rq_buffer + bufsiz);
815         rcvbuf->head[0].iov_len  = bufsiz;
816         rcvbuf->tail[0].iov_len  = 0;
817         rcvbuf->page_len         = 0;
818         rcvbuf->len              = 0;
819         rcvbuf->buflen           = bufsiz;
820
821         /* Encode header and provided arguments */
822         encode = task->tk_msg.rpc_proc->p_encode;
823         if (!(p = call_header(task))) {
824                 printk(KERN_INFO "RPC: call_header failed, exit EIO\n");
825                 rpc_exit(task, -EIO);
826                 return;
827         }
828         if (encode == NULL)
829                 return;
830
831         lock_kernel();
832         task->tk_status = rpcauth_wrap_req(task, encode, req, p,
833                         task->tk_msg.rpc_argp);
834         unlock_kernel();
835         if (task->tk_status == -ENOMEM) {
836                 /* XXX: Is this sane? */
837                 rpc_delay(task, 3*HZ);
838                 task->tk_status = -EAGAIN;
839         }
840 }
841
842 /*
843  * 4.   Get the server port number if not yet set
844  */
845 static void
846 call_bind(struct rpc_task *task)
847 {
848         struct rpc_xprt *xprt = task->tk_xprt;
849
850         dprint_status(task);
851
852         task->tk_action = call_connect;
853         if (!xprt_bound(xprt)) {
854                 task->tk_action = call_bind_status;
855                 task->tk_timeout = xprt->bind_timeout;
856                 xprt->ops->rpcbind(task);
857         }
858 }
859
860 /*
861  * 4a.  Sort out bind result
862  */
863 static void
864 call_bind_status(struct rpc_task *task)
865 {
866         int status = -EACCES;
867
868         if (task->tk_status >= 0) {
869                 dprint_status(task);
870                 task->tk_status = 0;
871                 task->tk_action = call_connect;
872                 return;
873         }
874
875         switch (task->tk_status) {
876         case -EACCES:
877                 dprintk("RPC: %5u remote rpcbind: RPC program/version "
878                                 "unavailable\n", task->tk_pid);
879                 rpc_delay(task, 3*HZ);
880                 goto retry_timeout;
881         case -ETIMEDOUT:
882                 dprintk("RPC: %5u rpcbind request timed out\n",
883                                 task->tk_pid);
884                 goto retry_timeout;
885         case -EPFNOSUPPORT:
886                 dprintk("RPC: %5u remote rpcbind service unavailable\n",
887                                 task->tk_pid);
888                 break;
889         case -EPROTONOSUPPORT:
890                 dprintk("RPC: %5u remote rpcbind version 2 unavailable\n",
891                                 task->tk_pid);
892                 break;
893         default:
894                 dprintk("RPC: %5u unrecognized rpcbind error (%d)\n",
895                                 task->tk_pid, -task->tk_status);
896                 status = -EIO;
897         }
898
899         rpc_exit(task, status);
900         return;
901
902 retry_timeout:
903         task->tk_action = call_timeout;
904 }
905
906 /*
907  * 4b.  Connect to the RPC server
908  */
909 static void
910 call_connect(struct rpc_task *task)
911 {
912         struct rpc_xprt *xprt = task->tk_xprt;
913
914         dprintk("RPC: %5u call_connect xprt %p %s connected\n",
915                         task->tk_pid, xprt,
916                         (xprt_connected(xprt) ? "is" : "is not"));
917
918         task->tk_action = call_transmit;
919         if (!xprt_connected(xprt)) {
920                 task->tk_action = call_connect_status;
921                 if (task->tk_status < 0)
922                         return;
923                 xprt_connect(task);
924         }
925 }
926
927 /*
928  * 4c.  Sort out connect result
929  */
930 static void
931 call_connect_status(struct rpc_task *task)
932 {
933         struct rpc_clnt *clnt = task->tk_client;
934         int status = task->tk_status;
935
936         dprint_status(task);
937
938         task->tk_status = 0;
939         if (status >= 0) {
940                 clnt->cl_stats->netreconn++;
941                 task->tk_action = call_transmit;
942                 return;
943         }
944
945         /* Something failed: remote service port may have changed */
946         rpc_force_rebind(clnt);
947
948         switch (status) {
949         case -ENOTCONN:
950         case -EAGAIN:
951                 task->tk_action = call_bind;
952                 if (!RPC_IS_SOFT(task))
953                         return;
954                 /* if soft mounted, test if we've timed out */
955         case -ETIMEDOUT:
956                 task->tk_action = call_timeout;
957                 return;
958         }
959         rpc_exit(task, -EIO);
960 }
961
962 /*
963  * 5.   Transmit the RPC request, and wait for reply
964  */
965 static void
966 call_transmit(struct rpc_task *task)
967 {
968         dprint_status(task);
969
970         task->tk_action = call_status;
971         if (task->tk_status < 0)
972                 return;
973         task->tk_status = xprt_prepare_transmit(task);
974         if (task->tk_status != 0)
975                 return;
976         task->tk_action = call_transmit_status;
977         /* Encode here so that rpcsec_gss can use correct sequence number. */
978         if (rpc_task_need_encode(task)) {
979                 BUG_ON(task->tk_rqstp->rq_bytes_sent != 0);
980                 call_encode(task);
981                 /* Did the encode result in an error condition? */
982                 if (task->tk_status != 0)
983                         return;
984         }
985         xprt_transmit(task);
986         if (task->tk_status < 0)
987                 return;
988         /*
989          * On success, ensure that we call xprt_end_transmit() before sleeping
990          * in order to allow access to the socket to other RPC requests.
991          */
992         call_transmit_status(task);
993         if (task->tk_msg.rpc_proc->p_decode != NULL)
994                 return;
995         task->tk_action = rpc_exit_task;
996         rpc_wake_up_task(task);
997 }
998
999 /*
1000  * 5a.  Handle cleanup after a transmission
1001  */
1002 static void
1003 call_transmit_status(struct rpc_task *task)
1004 {
1005         task->tk_action = call_status;
1006         /*
1007          * Special case: if we've been waiting on the socket's write_space()
1008          * callback, then don't call xprt_end_transmit().
1009          */
1010         if (task->tk_status == -EAGAIN)
1011                 return;
1012         xprt_end_transmit(task);
1013         rpc_task_force_reencode(task);
1014 }
1015
1016 /*
1017  * 6.   Sort out the RPC call status
1018  */
1019 static void
1020 call_status(struct rpc_task *task)
1021 {
1022         struct rpc_clnt *clnt = task->tk_client;
1023         struct rpc_rqst *req = task->tk_rqstp;
1024         int             status;
1025
1026         if (req->rq_received > 0 && !req->rq_bytes_sent)
1027                 task->tk_status = req->rq_received;
1028
1029         dprint_status(task);
1030
1031         status = task->tk_status;
1032         if (status >= 0) {
1033                 task->tk_action = call_decode;
1034                 return;
1035         }
1036
1037         task->tk_status = 0;
1038         switch(status) {
1039         case -EHOSTDOWN:
1040         case -EHOSTUNREACH:
1041         case -ENETUNREACH:
1042                 /*
1043                  * Delay any retries for 3 seconds, then handle as if it
1044                  * were a timeout.
1045                  */
1046                 rpc_delay(task, 3*HZ);
1047         case -ETIMEDOUT:
1048                 task->tk_action = call_timeout;
1049                 if (task->tk_client->cl_discrtry)
1050                         xprt_disconnect(task->tk_xprt);
1051                 break;
1052         case -ECONNREFUSED:
1053         case -ENOTCONN:
1054                 rpc_force_rebind(clnt);
1055                 task->tk_action = call_bind;
1056                 break;
1057         case -EAGAIN:
1058                 task->tk_action = call_transmit;
1059                 break;
1060         case -EIO:
1061                 /* shutdown or soft timeout */
1062                 rpc_exit(task, status);
1063                 break;
1064         default:
1065                 printk("%s: RPC call returned error %d\n",
1066                                clnt->cl_protname, -status);
1067                 rpc_exit(task, status);
1068         }
1069 }
1070
1071 /*
1072  * 6a.  Handle RPC timeout
1073  *      We do not release the request slot, so we keep using the
1074  *      same XID for all retransmits.
1075  */
1076 static void
1077 call_timeout(struct rpc_task *task)
1078 {
1079         struct rpc_clnt *clnt = task->tk_client;
1080
1081         if (xprt_adjust_timeout(task->tk_rqstp) == 0) {
1082                 dprintk("RPC: %5u call_timeout (minor)\n", task->tk_pid);
1083                 goto retry;
1084         }
1085
1086         dprintk("RPC: %5u call_timeout (major)\n", task->tk_pid);
1087         task->tk_timeouts++;
1088
1089         if (RPC_IS_SOFT(task)) {
1090                 printk(KERN_NOTICE "%s: server %s not responding, timed out\n",
1091                                 clnt->cl_protname, clnt->cl_server);
1092                 rpc_exit(task, -EIO);
1093                 return;
1094         }
1095
1096         if (!(task->tk_flags & RPC_CALL_MAJORSEEN)) {
1097                 task->tk_flags |= RPC_CALL_MAJORSEEN;
1098                 printk(KERN_NOTICE "%s: server %s not responding, still trying\n",
1099                         clnt->cl_protname, clnt->cl_server);
1100         }
1101         rpc_force_rebind(clnt);
1102
1103 retry:
1104         clnt->cl_stats->rpcretrans++;
1105         task->tk_action = call_bind;
1106         task->tk_status = 0;
1107 }
1108
1109 /*
1110  * 7.   Decode the RPC reply
1111  */
1112 static void
1113 call_decode(struct rpc_task *task)
1114 {
1115         struct rpc_clnt *clnt = task->tk_client;
1116         struct rpc_rqst *req = task->tk_rqstp;
1117         kxdrproc_t      decode = task->tk_msg.rpc_proc->p_decode;
1118         __be32          *p;
1119
1120         dprintk("RPC: %5u call_decode (status %d)\n",
1121                         task->tk_pid, task->tk_status);
1122
1123         if (task->tk_flags & RPC_CALL_MAJORSEEN) {
1124                 printk(KERN_NOTICE "%s: server %s OK\n",
1125                         clnt->cl_protname, clnt->cl_server);
1126                 task->tk_flags &= ~RPC_CALL_MAJORSEEN;
1127         }
1128
1129         if (task->tk_status < 12) {
1130                 if (!RPC_IS_SOFT(task)) {
1131                         task->tk_action = call_bind;
1132                         clnt->cl_stats->rpcretrans++;
1133                         goto out_retry;
1134                 }
1135                 dprintk("RPC:       %s: too small RPC reply size (%d bytes)\n",
1136                                 clnt->cl_protname, task->tk_status);
1137                 task->tk_action = call_timeout;
1138                 goto out_retry;
1139         }
1140
1141         /*
1142          * Ensure that we see all writes made by xprt_complete_rqst()
1143          * before it changed req->rq_received.
1144          */
1145         smp_rmb();
1146         req->rq_rcv_buf.len = req->rq_private_buf.len;
1147
1148         /* Check that the softirq receive buffer is valid */
1149         WARN_ON(memcmp(&req->rq_rcv_buf, &req->rq_private_buf,
1150                                 sizeof(req->rq_rcv_buf)) != 0);
1151
1152         /* Verify the RPC header */
1153         p = call_verify(task);
1154         if (IS_ERR(p)) {
1155                 if (p == ERR_PTR(-EAGAIN))
1156                         goto out_retry;
1157                 return;
1158         }
1159
1160         task->tk_action = rpc_exit_task;
1161
1162         if (decode) {
1163                 lock_kernel();
1164                 task->tk_status = rpcauth_unwrap_resp(task, decode, req, p,
1165                                                       task->tk_msg.rpc_resp);
1166                 unlock_kernel();
1167         }
1168         dprintk("RPC: %5u call_decode result %d\n", task->tk_pid,
1169                         task->tk_status);
1170         return;
1171 out_retry:
1172         req->rq_received = req->rq_private_buf.len = 0;
1173         task->tk_status = 0;
1174         if (task->tk_client->cl_discrtry)
1175                 xprt_disconnect(task->tk_xprt);
1176 }
1177
1178 /*
1179  * 8.   Refresh the credentials if rejected by the server
1180  */
1181 static void
1182 call_refresh(struct rpc_task *task)
1183 {
1184         dprint_status(task);
1185
1186         xprt_release(task);     /* Must do to obtain new XID */
1187         task->tk_action = call_refreshresult;
1188         task->tk_status = 0;
1189         task->tk_client->cl_stats->rpcauthrefresh++;
1190         rpcauth_refreshcred(task);
1191 }
1192
1193 /*
1194  * 8a.  Process the results of a credential refresh
1195  */
1196 static void
1197 call_refreshresult(struct rpc_task *task)
1198 {
1199         int status = task->tk_status;
1200
1201         dprint_status(task);
1202
1203         task->tk_status = 0;
1204         task->tk_action = call_reserve;
1205         if (status >= 0 && rpcauth_uptodatecred(task))
1206                 return;
1207         if (status == -EACCES) {
1208                 rpc_exit(task, -EACCES);
1209                 return;
1210         }
1211         task->tk_action = call_refresh;
1212         if (status != -ETIMEDOUT)
1213                 rpc_delay(task, 3*HZ);
1214         return;
1215 }
1216
1217 /*
1218  * Call header serialization
1219  */
1220 static __be32 *
1221 call_header(struct rpc_task *task)
1222 {
1223         struct rpc_clnt *clnt = task->tk_client;
1224         struct rpc_rqst *req = task->tk_rqstp;
1225         __be32          *p = req->rq_svec[0].iov_base;
1226
1227         /* FIXME: check buffer size? */
1228
1229         p = xprt_skip_transport_header(task->tk_xprt, p);
1230         *p++ = req->rq_xid;             /* XID */
1231         *p++ = htonl(RPC_CALL);         /* CALL */
1232         *p++ = htonl(RPC_VERSION);      /* RPC version */
1233         *p++ = htonl(clnt->cl_prog);    /* program number */
1234         *p++ = htonl(clnt->cl_vers);    /* program version */
1235         *p++ = htonl(task->tk_msg.rpc_proc->p_proc);    /* procedure */
1236         p = rpcauth_marshcred(task, p);
1237         req->rq_slen = xdr_adjust_iovec(&req->rq_svec[0], p);
1238         return p;
1239 }
1240
1241 /*
1242  * Reply header verification
1243  */
1244 static __be32 *
1245 call_verify(struct rpc_task *task)
1246 {
1247         struct kvec *iov = &task->tk_rqstp->rq_rcv_buf.head[0];
1248         int len = task->tk_rqstp->rq_rcv_buf.len >> 2;
1249         __be32  *p = iov->iov_base;
1250         u32 n;
1251         int error = -EACCES;
1252
1253         if ((task->tk_rqstp->rq_rcv_buf.len & 3) != 0) {
1254                 /* RFC-1014 says that the representation of XDR data must be a
1255                  * multiple of four bytes
1256                  * - if it isn't pointer subtraction in the NFS client may give
1257                  *   undefined results
1258                  */
1259                 printk(KERN_WARNING
1260                        "call_verify: XDR representation not a multiple of"
1261                        " 4 bytes: 0x%x\n", task->tk_rqstp->rq_rcv_buf.len);
1262                 goto out_eio;
1263         }
1264         if ((len -= 3) < 0)
1265                 goto out_overflow;
1266         p += 1; /* skip XID */
1267
1268         if ((n = ntohl(*p++)) != RPC_REPLY) {
1269                 printk(KERN_WARNING "call_verify: not an RPC reply: %x\n", n);
1270                 goto out_garbage;
1271         }
1272         if ((n = ntohl(*p++)) != RPC_MSG_ACCEPTED) {
1273                 if (--len < 0)
1274                         goto out_overflow;
1275                 switch ((n = ntohl(*p++))) {
1276                         case RPC_AUTH_ERROR:
1277                                 break;
1278                         case RPC_MISMATCH:
1279                                 dprintk("RPC: %5u %s: RPC call version "
1280                                                 "mismatch!\n",
1281                                                 task->tk_pid, __FUNCTION__);
1282                                 error = -EPROTONOSUPPORT;
1283                                 goto out_err;
1284                         default:
1285                                 dprintk("RPC: %5u %s: RPC call rejected, "
1286                                                 "unknown error: %x\n",
1287                                                 task->tk_pid, __FUNCTION__, n);
1288                                 goto out_eio;
1289                 }
1290                 if (--len < 0)
1291                         goto out_overflow;
1292                 switch ((n = ntohl(*p++))) {
1293                 case RPC_AUTH_REJECTEDCRED:
1294                 case RPC_AUTH_REJECTEDVERF:
1295                 case RPCSEC_GSS_CREDPROBLEM:
1296                 case RPCSEC_GSS_CTXPROBLEM:
1297                         if (!task->tk_cred_retry)
1298                                 break;
1299                         task->tk_cred_retry--;
1300                         dprintk("RPC: %5u %s: retry stale creds\n",
1301                                         task->tk_pid, __FUNCTION__);
1302                         rpcauth_invalcred(task);
1303                         task->tk_action = call_refresh;
1304                         goto out_retry;
1305                 case RPC_AUTH_BADCRED:
1306                 case RPC_AUTH_BADVERF:
1307                         /* possibly garbled cred/verf? */
1308                         if (!task->tk_garb_retry)
1309                                 break;
1310                         task->tk_garb_retry--;
1311                         dprintk("RPC: %5u %s: retry garbled creds\n",
1312                                         task->tk_pid, __FUNCTION__);
1313                         task->tk_action = call_bind;
1314                         goto out_retry;
1315                 case RPC_AUTH_TOOWEAK:
1316                         printk(KERN_NOTICE "call_verify: server %s requires stronger "
1317                                "authentication.\n", task->tk_client->cl_server);
1318                         break;
1319                 default:
1320                         printk(KERN_WARNING "call_verify: unknown auth error: %x\n", n);
1321                         error = -EIO;
1322                 }
1323                 dprintk("RPC: %5u %s: call rejected %d\n",
1324                                 task->tk_pid, __FUNCTION__, n);
1325                 goto out_err;
1326         }
1327         if (!(p = rpcauth_checkverf(task, p))) {
1328                 printk(KERN_WARNING "call_verify: auth check failed\n");
1329                 goto out_garbage;               /* bad verifier, retry */
1330         }
1331         len = p - (__be32 *)iov->iov_base - 1;
1332         if (len < 0)
1333                 goto out_overflow;
1334         switch ((n = ntohl(*p++))) {
1335         case RPC_SUCCESS:
1336                 return p;
1337         case RPC_PROG_UNAVAIL:
1338                 dprintk("RPC: %5u %s: program %u is unsupported by server %s\n",
1339                                 task->tk_pid, __FUNCTION__,
1340                                 (unsigned int)task->tk_client->cl_prog,
1341                                 task->tk_client->cl_server);
1342                 error = -EPFNOSUPPORT;
1343                 goto out_err;
1344         case RPC_PROG_MISMATCH:
1345                 dprintk("RPC: %5u %s: program %u, version %u unsupported by "
1346                                 "server %s\n", task->tk_pid, __FUNCTION__,
1347                                 (unsigned int)task->tk_client->cl_prog,
1348                                 (unsigned int)task->tk_client->cl_vers,
1349                                 task->tk_client->cl_server);
1350                 error = -EPROTONOSUPPORT;
1351                 goto out_err;
1352         case RPC_PROC_UNAVAIL:
1353                 dprintk("RPC: %5u %s: proc %p unsupported by program %u, "
1354                                 "version %u on server %s\n",
1355                                 task->tk_pid, __FUNCTION__,
1356                                 task->tk_msg.rpc_proc,
1357                                 task->tk_client->cl_prog,
1358                                 task->tk_client->cl_vers,
1359                                 task->tk_client->cl_server);
1360                 error = -EOPNOTSUPP;
1361                 goto out_err;
1362         case RPC_GARBAGE_ARGS:
1363                 dprintk("RPC: %5u %s: server saw garbage\n",
1364                                 task->tk_pid, __FUNCTION__);
1365                 break;                  /* retry */
1366         default:
1367                 printk(KERN_WARNING "call_verify: server accept status: %x\n", n);
1368                 /* Also retry */
1369         }
1370
1371 out_garbage:
1372         task->tk_client->cl_stats->rpcgarbage++;
1373         if (task->tk_garb_retry) {
1374                 task->tk_garb_retry--;
1375                 dprintk("RPC: %5u %s: retrying\n",
1376                                 task->tk_pid, __FUNCTION__);
1377                 task->tk_action = call_bind;
1378 out_retry:
1379                 return ERR_PTR(-EAGAIN);
1380         }
1381         printk(KERN_WARNING "RPC %s: retry failed, exit EIO\n", __FUNCTION__);
1382 out_eio:
1383         error = -EIO;
1384 out_err:
1385         rpc_exit(task, error);
1386         return ERR_PTR(error);
1387 out_overflow:
1388         printk(KERN_WARNING "RPC %s: server reply was truncated.\n", __FUNCTION__);
1389         goto out_garbage;
1390 }
1391
1392 static int rpcproc_encode_null(void *rqstp, __be32 *data, void *obj)
1393 {
1394         return 0;
1395 }
1396
1397 static int rpcproc_decode_null(void *rqstp, __be32 *data, void *obj)
1398 {
1399         return 0;
1400 }
1401
1402 static struct rpc_procinfo rpcproc_null = {
1403         .p_encode = rpcproc_encode_null,
1404         .p_decode = rpcproc_decode_null,
1405 };
1406
1407 int rpc_ping(struct rpc_clnt *clnt, int flags)
1408 {
1409         struct rpc_message msg = {
1410                 .rpc_proc = &rpcproc_null,
1411         };
1412         int err;
1413         msg.rpc_cred = authnull_ops.lookup_cred(NULL, NULL, 0);
1414         err = rpc_call_sync(clnt, &msg, flags);
1415         put_rpccred(msg.rpc_cred);
1416         return err;
1417 }