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