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