net/sunrpc/clnt.c

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