net/sunrpc/clnt.c

   1 // SPDX-License-Identifier: GPL-2.0-only
   2 /*
   3  *  linux/net/sunrpc/clnt.c
   4  *
   5  *  This file contains the high-level RPC interface.
   6  *  It is modeled as a finite state machine to support both synchronous
   7  *  and asynchronous requests.
   8  *
   9  *  -   RPC header generation and argument serialization.
  10  *  -   Credential refresh.
  11  *  -   TCP connect handling.
  12  *  -   Retry of operation when it is suspected the operation failed because
  13  *      of uid squashing on the server, or when the credentials were stale
  14  *      and need to be refreshed, or when a packet was damaged in transit.
  15  *      This may be have to be moved to the VFS layer.
  16  *
  17  *  Copyright (C) 1992,1993 Rick Sladkey <jrs@world.std.com>
  18  *  Copyright (C) 1995,1996 Olaf Kirch <okir@monad.swb.de>
  19  */
  20
  21
  22 #include <linux/module.h>
  23 #include <linux/types.h>
  24 #include <linux/kallsyms.h>
  25 #include <linux/mm.h>
  26 #include <linux/namei.h>
  27 #include <linux/mount.h>
  28 #include <linux/slab.h>
  29 #include <linux/rcupdate.h>
  30 #include <linux/utsname.h>
  31 #include <linux/workqueue.h>
  32 #include <linux/in.h>
  33 #include <linux/in6.h>
  34 #include <linux/un.h>
  35
  36 #include <linux/sunrpc/clnt.h>
  37 #include <linux/sunrpc/addr.h>
  38 #include <linux/sunrpc/rpc_pipe_fs.h>
  39 #include <linux/sunrpc/metrics.h>
  40 #include <linux/sunrpc/bc_xprt.h>
  41 #include <trace/events/sunrpc.h>
  42
  43 #include "sunrpc.h"
  44 #include "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                                 return;
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         rpc_check_timeout(task);
2242 }
2243
2244 /*
2245  * 5.   Transmit the RPC request, and wait for reply
2246  */
2247 static void
2248 call_transmit(struct rpc_task *task)
2249 {
2250         if (rpc_task_transmitted(task)) {
2251                 rpc_task_handle_transmitted(task);
2252                 return;
2253         }
2254
2255         task->tk_action = call_transmit_status;
2256         if (!xprt_prepare_transmit(task))
2257                 return;
2258         task->tk_status = 0;
2259         if (test_bit(RPC_TASK_NEED_XMIT, &task->tk_runstate)) {
2260                 if (!xprt_connected(task->tk_xprt)) {
2261                         task->tk_status = -ENOTCONN;
2262                         return;
2263                 }
2264                 xprt_transmit(task);
2265         }
2266         xprt_end_transmit(task);
2267 }
2268
2269 /*
2270  * 5a.  Handle cleanup after a transmission
2271  */
2272 static void
2273 call_transmit_status(struct rpc_task *task)
2274 {
2275         task->tk_action = call_status;
2276
2277         /*
2278          * Common case: success.  Force the compiler to put this
2279          * test first.
2280          */
2281         if (rpc_task_transmitted(task)) {
2282                 task->tk_status = 0;
2283                 xprt_request_wait_receive(task);
2284                 return;
2285         }
2286
2287         switch (task->tk_status) {
2288         default:
2289                 break;
2290         case -EBADMSG:
2291                 task->tk_status = 0;
2292                 task->tk_action = call_encode;
2293                 break;
2294                 /*
2295                  * Special cases: if we've been waiting on the
2296                  * socket's write_space() callback, or if the
2297                  * socket just returned a connection error,
2298                  * then hold onto the transport lock.
2299                  */
2300         case -ENOMEM:
2301         case -ENOBUFS:
2302                 rpc_delay(task, HZ>>2);
2303                 fallthrough;
2304         case -EBADSLT:
2305         case -EAGAIN:
2306                 task->tk_action = call_transmit;
2307                 task->tk_status = 0;
2308                 break;
2309         case -ECONNREFUSED:
2310         case -EHOSTDOWN:
2311         case -ENETDOWN:
2312         case -EHOSTUNREACH:
2313         case -ENETUNREACH:
2314         case -EPERM:
2315                 if (RPC_IS_SOFTCONN(task)) {
2316                         if (!task->tk_msg.rpc_proc->p_proc)
2317                                 trace_xprt_ping(task->tk_xprt,
2318                                                 task->tk_status);
2319                         rpc_call_rpcerror(task, task->tk_status);
2320                         return;
2321                 }
2322                 fallthrough;
2323         case -ECONNRESET:
2324         case -ECONNABORTED:
2325         case -EADDRINUSE:
2326         case -ENOTCONN:
2327         case -EPIPE:
2328                 task->tk_action = call_bind;
2329                 task->tk_status = 0;
2330                 break;
2331         }
2332         rpc_check_timeout(task);
2333 }
2334
2335 #if defined(CONFIG_SUNRPC_BACKCHANNEL)
2336 static void call_bc_transmit(struct rpc_task *task);
2337 static void call_bc_transmit_status(struct rpc_task *task);
2338
2339 static void
2340 call_bc_encode(struct rpc_task *task)
2341 {
2342         xprt_request_enqueue_transmit(task);
2343         task->tk_action = call_bc_transmit;
2344 }
2345
2346 /*
2347  * 5b.  Send the backchannel RPC reply.  On error, drop the reply.  In
2348  * addition, disconnect on connectivity errors.
2349  */
2350 static void
2351 call_bc_transmit(struct rpc_task *task)
2352 {
2353         task->tk_action = call_bc_transmit_status;
2354         if (test_bit(RPC_TASK_NEED_XMIT, &task->tk_runstate)) {
2355                 if (!xprt_prepare_transmit(task))
2356                         return;
2357                 task->tk_status = 0;
2358                 xprt_transmit(task);
2359         }
2360         xprt_end_transmit(task);
2361 }
2362
2363 static void
2364 call_bc_transmit_status(struct rpc_task *task)
2365 {
2366         struct rpc_rqst *req = task->tk_rqstp;
2367
2368         if (rpc_task_transmitted(task))
2369                 task->tk_status = 0;
2370
2371         switch (task->tk_status) {
2372         case 0:
2373                 /* Success */
2374         case -ENETDOWN:
2375         case -EHOSTDOWN:
2376         case -EHOSTUNREACH:
2377         case -ENETUNREACH:
2378         case -ECONNRESET:
2379         case -ECONNREFUSED:
2380         case -EADDRINUSE:
2381         case -ENOTCONN:
2382         case -EPIPE:
2383                 break;
2384         case -ENOMEM:
2385         case -ENOBUFS:
2386                 rpc_delay(task, HZ>>2);
2387                 fallthrough;
2388         case -EBADSLT:
2389         case -EAGAIN:
2390                 task->tk_status = 0;
2391                 task->tk_action = call_bc_transmit;
2392                 return;
2393         case -ETIMEDOUT:
2394                 /*
2395                  * Problem reaching the server.  Disconnect and let the
2396                  * forechannel reestablish the connection.  The server will
2397                  * have to retransmit the backchannel request and we'll
2398                  * reprocess it.  Since these ops are idempotent, there's no
2399                  * need to cache our reply at this time.
2400                  */
2401                 printk(KERN_NOTICE "RPC: Could not send backchannel reply "
2402                         "error: %d\n", task->tk_status);
2403                 xprt_conditional_disconnect(req->rq_xprt,
2404                         req->rq_connect_cookie);
2405                 break;
2406         default:
2407                 /*
2408                  * We were unable to reply and will have to drop the
2409                  * request.  The server should reconnect and retransmit.
2410                  */
2411                 printk(KERN_NOTICE "RPC: Could not send backchannel reply "
2412                         "error: %d\n", task->tk_status);
2413                 break;
2414         }
2415         task->tk_action = rpc_exit_task;
2416 }
2417 #endif /* CONFIG_SUNRPC_BACKCHANNEL */
2418
2419 /*
2420  * 6.   Sort out the RPC call status
2421  */
2422 static void
2423 call_status(struct rpc_task *task)
2424 {
2425         struct rpc_clnt *clnt = task->tk_client;
2426         int             status;
2427
2428         if (!task->tk_msg.rpc_proc->p_proc)
2429                 trace_xprt_ping(task->tk_xprt, task->tk_status);
2430
2431         status = task->tk_status;
2432         if (status >= 0) {
2433                 task->tk_action = call_decode;
2434                 return;
2435         }
2436
2437         trace_rpc_call_status(task);
2438         task->tk_status = 0;
2439         switch(status) {
2440         case -EHOSTDOWN:
2441         case -ENETDOWN:
2442         case -EHOSTUNREACH:
2443         case -ENETUNREACH:
2444         case -EPERM:
2445                 if (RPC_IS_SOFTCONN(task))
2446                         goto out_exit;
2447                 /*
2448                  * Delay any retries for 3 seconds, then handle as if it
2449                  * were a timeout.
2450                  */
2451                 rpc_delay(task, 3*HZ);
2452                 fallthrough;
2453         case -ETIMEDOUT:
2454                 break;
2455         case -ECONNREFUSED:
2456         case -ECONNRESET:
2457         case -ECONNABORTED:
2458         case -ENOTCONN:
2459                 rpc_force_rebind(clnt);
2460                 break;
2461         case -EADDRINUSE:
2462                 rpc_delay(task, 3*HZ);
2463                 fallthrough;
2464         case -EPIPE:
2465         case -EAGAIN:
2466                 break;
2467         case -ENFILE:
2468         case -ENOBUFS:
2469         case -ENOMEM:
2470                 rpc_delay(task, HZ>>2);
2471                 break;
2472         case -EIO:
2473                 /* shutdown or soft timeout */
2474                 goto out_exit;
2475         default:
2476                 if (clnt->cl_chatty)
2477                         printk("%s: RPC call returned error %d\n",
2478                                clnt->cl_program->name, -status);
2479                 goto out_exit;
2480         }
2481         task->tk_action = call_encode;
2482         rpc_check_timeout(task);
2483         return;
2484 out_exit:
2485         rpc_call_rpcerror(task, status);
2486 }
2487
2488 static bool
2489 rpc_check_connected(const struct rpc_rqst *req)
2490 {
2491         /* No allocated request or transport? return true */
2492         if (!req || !req->rq_xprt)
2493                 return true;
2494         return xprt_connected(req->rq_xprt);
2495 }
2496
2497 static void
2498 rpc_check_timeout(struct rpc_task *task)
2499 {
2500         struct rpc_clnt *clnt = task->tk_client;
2501
2502         if (RPC_SIGNALLED(task))
2503                 return;
2504
2505         if (xprt_adjust_timeout(task->tk_rqstp) == 0)
2506                 return;
2507
2508         trace_rpc_timeout_status(task);
2509         task->tk_timeouts++;
2510
2511         if (RPC_IS_SOFTCONN(task) && !rpc_check_connected(task->tk_rqstp)) {
2512                 rpc_call_rpcerror(task, -ETIMEDOUT);
2513                 return;
2514         }
2515
2516         if (RPC_IS_SOFT(task)) {
2517                 /*
2518                  * Once a "no retrans timeout" soft tasks (a.k.a NFSv4) has
2519                  * been sent, it should time out only if the transport
2520                  * connection gets terminally broken.
2521                  */
2522                 if ((task->tk_flags & RPC_TASK_NO_RETRANS_TIMEOUT) &&
2523                     rpc_check_connected(task->tk_rqstp))
2524                         return;
2525
2526                 if (clnt->cl_chatty) {
2527                         pr_notice_ratelimited(
2528                                 "%s: server %s not responding, timed out\n",
2529                                 clnt->cl_program->name,
2530                                 task->tk_xprt->servername);
2531                 }
2532                 if (task->tk_flags & RPC_TASK_TIMEOUT)
2533                         rpc_call_rpcerror(task, -ETIMEDOUT);
2534                 else
2535                         __rpc_call_rpcerror(task, -EIO, -ETIMEDOUT);
2536                 return;
2537         }
2538
2539         if (!(task->tk_flags & RPC_CALL_MAJORSEEN)) {
2540                 task->tk_flags |= RPC_CALL_MAJORSEEN;
2541                 if (clnt->cl_chatty) {
2542                         pr_notice_ratelimited(
2543                                 "%s: server %s not responding, still trying\n",
2544                                 clnt->cl_program->name,
2545                                 task->tk_xprt->servername);
2546                 }
2547         }
2548         rpc_force_rebind(clnt);
2549         /*
2550          * Did our request time out due to an RPCSEC_GSS out-of-sequence
2551          * event? RFC2203 requires the server to drop all such requests.
2552          */
2553         rpcauth_invalcred(task);
2554 }
2555
2556 /*
2557  * 7.   Decode the RPC reply
2558  */
2559 static void
2560 call_decode(struct rpc_task *task)
2561 {
2562         struct rpc_clnt *clnt = task->tk_client;
2563         struct rpc_rqst *req = task->tk_rqstp;
2564         struct xdr_stream xdr;
2565         int err;
2566
2567         if (!task->tk_msg.rpc_proc->p_decode) {
2568                 task->tk_action = rpc_exit_task;
2569                 return;
2570         }
2571
2572         if (task->tk_flags & RPC_CALL_MAJORSEEN) {
2573                 if (clnt->cl_chatty) {
2574                         pr_notice_ratelimited("%s: server %s OK\n",
2575                                 clnt->cl_program->name,
2576                                 task->tk_xprt->servername);
2577                 }
2578                 task->tk_flags &= ~RPC_CALL_MAJORSEEN;
2579         }
2580
2581         /*
2582          * Did we ever call xprt_complete_rqst()? If not, we should assume
2583          * the message is incomplete.
2584          */
2585         err = -EAGAIN;
2586         if (!req->rq_reply_bytes_recvd)
2587                 goto out;
2588
2589         /* Ensure that we see all writes made by xprt_complete_rqst()
2590          * before it changed req->rq_reply_bytes_recvd.
2591          */
2592         smp_rmb();
2593
2594         req->rq_rcv_buf.len = req->rq_private_buf.len;
2595         trace_rpc_xdr_recvfrom(task, &req->rq_rcv_buf);
2596
2597         /* Check that the softirq receive buffer is valid */
2598         WARN_ON(memcmp(&req->rq_rcv_buf, &req->rq_private_buf,
2599                                 sizeof(req->rq_rcv_buf)) != 0);
2600
2601         xdr_init_decode(&xdr, &req->rq_rcv_buf,
2602                         req->rq_rcv_buf.head[0].iov_base, req);
2603         err = rpc_decode_header(task, &xdr);
2604 out:
2605         switch (err) {
2606         case 0:
2607                 task->tk_action = rpc_exit_task;
2608                 task->tk_status = rpcauth_unwrap_resp(task, &xdr);
2609                 xdr_finish_decode(&xdr);
2610                 return;
2611         case -EAGAIN:
2612                 task->tk_status = 0;
2613                 if (task->tk_client->cl_discrtry)
2614                         xprt_conditional_disconnect(req->rq_xprt,
2615                                                     req->rq_connect_cookie);
2616                 task->tk_action = call_encode;
2617                 rpc_check_timeout(task);
2618                 break;
2619         case -EKEYREJECTED:
2620                 task->tk_action = call_reserve;
2621                 rpc_check_timeout(task);
2622                 rpcauth_invalcred(task);
2623                 /* Ensure we obtain a new XID if we retry! */
2624                 xprt_release(task);
2625         }
2626 }
2627
2628 static int
2629 rpc_encode_header(struct rpc_task *task, struct xdr_stream *xdr)
2630 {
2631         struct rpc_clnt *clnt = task->tk_client;
2632         struct rpc_rqst *req = task->tk_rqstp;
2633         __be32 *p;
2634         int error;
2635
2636         error = -EMSGSIZE;
2637         p = xdr_reserve_space(xdr, RPC_CALLHDRSIZE << 2);
2638         if (!p)
2639                 goto out_fail;
2640         *p++ = req->rq_xid;
2641         *p++ = rpc_call;
2642         *p++ = cpu_to_be32(RPC_VERSION);
2643         *p++ = cpu_to_be32(clnt->cl_prog);
2644         *p++ = cpu_to_be32(clnt->cl_vers);
2645         *p   = cpu_to_be32(task->tk_msg.rpc_proc->p_proc);
2646
2647         error = rpcauth_marshcred(task, xdr);
2648         if (error < 0)
2649                 goto out_fail;
2650         return 0;
2651 out_fail:
2652         trace_rpc_bad_callhdr(task);
2653         rpc_call_rpcerror(task, error);
2654         return error;
2655 }
2656
2657 static noinline int
2658 rpc_decode_header(struct rpc_task *task, struct xdr_stream *xdr)
2659 {
2660         struct rpc_clnt *clnt = task->tk_client;
2661         int error;
2662         __be32 *p;
2663
2664         /* RFC-1014 says that the representation of XDR data must be a
2665          * multiple of four bytes
2666          * - if it isn't pointer subtraction in the NFS client may give
2667          *   undefined results
2668          */
2669         if (task->tk_rqstp->rq_rcv_buf.len & 3)
2670                 goto out_unparsable;
2671
2672         p = xdr_inline_decode(xdr, 3 * sizeof(*p));
2673         if (!p)
2674                 goto out_unparsable;
2675         p++;    /* skip XID */
2676         if (*p++ != rpc_reply)
2677                 goto out_unparsable;
2678         if (*p++ != rpc_msg_accepted)
2679                 goto out_msg_denied;
2680
2681         error = rpcauth_checkverf(task, xdr);
2682         if (error)
2683                 goto out_verifier;
2684
2685         p = xdr_inline_decode(xdr, sizeof(*p));
2686         if (!p)
2687                 goto out_unparsable;
2688         switch (*p) {
2689         case rpc_success:
2690                 return 0;
2691         case rpc_prog_unavail:
2692                 trace_rpc__prog_unavail(task);
2693                 error = -EPFNOSUPPORT;
2694                 goto out_err;
2695         case rpc_prog_mismatch:
2696                 trace_rpc__prog_mismatch(task);
2697                 error = -EPROTONOSUPPORT;
2698                 goto out_err;
2699         case rpc_proc_unavail:
2700                 trace_rpc__proc_unavail(task);
2701                 error = -EOPNOTSUPP;
2702                 goto out_err;
2703         case rpc_garbage_args:
2704         case rpc_system_err:
2705                 trace_rpc__garbage_args(task);
2706                 error = -EIO;
2707                 break;
2708         default:
2709                 goto out_unparsable;
2710         }
2711
2712 out_garbage:
2713         clnt->cl_stats->rpcgarbage++;
2714         if (task->tk_garb_retry) {
2715                 task->tk_garb_retry--;
2716                 task->tk_action = call_encode;
2717                 return -EAGAIN;
2718         }
2719 out_err:
2720         rpc_call_rpcerror(task, error);
2721         return error;
2722
2723 out_unparsable:
2724         trace_rpc__unparsable(task);
2725         error = -EIO;
2726         goto out_garbage;
2727
2728 out_verifier:
2729         trace_rpc_bad_verifier(task);
2730         switch (error) {
2731         case -EPROTONOSUPPORT:
2732                 goto out_err;
2733         case -EACCES:
2734                 /* Re-encode with a fresh cred */
2735                 fallthrough;
2736         default:
2737                 goto out_garbage;
2738         }
2739
2740 out_msg_denied:
2741         error = -EACCES;
2742         p = xdr_inline_decode(xdr, sizeof(*p));
2743         if (!p)
2744                 goto out_unparsable;
2745         switch (*p++) {
2746         case rpc_auth_error:
2747                 break;
2748         case rpc_mismatch:
2749                 trace_rpc__mismatch(task);
2750                 error = -EPROTONOSUPPORT;
2751                 goto out_err;
2752         default:
2753                 goto out_unparsable;
2754         }
2755
2756         p = xdr_inline_decode(xdr, sizeof(*p));
2757         if (!p)
2758                 goto out_unparsable;
2759         switch (*p++) {
2760         case rpc_autherr_rejectedcred:
2761         case rpc_autherr_rejectedverf:
2762         case rpcsec_gsserr_credproblem:
2763         case rpcsec_gsserr_ctxproblem:
2764                 rpcauth_invalcred(task);
2765                 if (!task->tk_cred_retry)
2766                         break;
2767                 task->tk_cred_retry--;
2768                 trace_rpc__stale_creds(task);
2769                 return -EKEYREJECTED;
2770         case rpc_autherr_badcred:
2771         case rpc_autherr_badverf:
2772                 /* possibly garbled cred/verf? */
2773                 if (!task->tk_garb_retry)
2774                         break;
2775                 task->tk_garb_retry--;
2776                 trace_rpc__bad_creds(task);
2777                 task->tk_action = call_encode;
2778                 return -EAGAIN;
2779         case rpc_autherr_tooweak:
2780                 trace_rpc__auth_tooweak(task);
2781                 pr_warn("RPC: server %s requires stronger authentication.\n",
2782                         task->tk_xprt->servername);
2783                 break;
2784         default:
2785                 goto out_unparsable;
2786         }
2787         goto out_err;
2788 }
2789
2790 static void rpcproc_encode_null(struct rpc_rqst *rqstp, struct xdr_stream *xdr,
2791                 const void *obj)
2792 {
2793 }
2794
2795 static int rpcproc_decode_null(struct rpc_rqst *rqstp, struct xdr_stream *xdr,
2796                 void *obj)
2797 {
2798         return 0;
2799 }
2800
2801 static const struct rpc_procinfo rpcproc_null = {
2802         .p_encode = rpcproc_encode_null,
2803         .p_decode = rpcproc_decode_null,
2804 };
2805
2806 static const struct rpc_procinfo rpcproc_null_noreply = {
2807         .p_encode = rpcproc_encode_null,
2808 };
2809
2810 static void
2811 rpc_null_call_prepare(struct rpc_task *task, void *data)
2812 {
2813         task->tk_flags &= ~RPC_TASK_NO_RETRANS_TIMEOUT;
2814         rpc_call_start(task);
2815 }
2816
2817 static const struct rpc_call_ops rpc_null_ops = {
2818         .rpc_call_prepare = rpc_null_call_prepare,
2819         .rpc_call_done = rpc_default_callback,
2820 };
2821
2822 static
2823 struct rpc_task *rpc_call_null_helper(struct rpc_clnt *clnt,
2824                 struct rpc_xprt *xprt, struct rpc_cred *cred, int flags,
2825                 const struct rpc_call_ops *ops, void *data)
2826 {
2827         struct rpc_message msg = {
2828                 .rpc_proc = &rpcproc_null,
2829         };
2830         struct rpc_task_setup task_setup_data = {
2831                 .rpc_client = clnt,
2832                 .rpc_xprt = xprt,
2833                 .rpc_message = &msg,
2834                 .rpc_op_cred = cred,
2835                 .callback_ops = ops ?: &rpc_null_ops,
2836                 .callback_data = data,
2837                 .flags = flags | RPC_TASK_SOFT | RPC_TASK_SOFTCONN |
2838                          RPC_TASK_NULLCREDS,
2839         };
2840
2841         return rpc_run_task(&task_setup_data);
2842 }
2843
2844 struct rpc_task *rpc_call_null(struct rpc_clnt *clnt, struct rpc_cred *cred, int flags)
2845 {
2846         return rpc_call_null_helper(clnt, NULL, cred, flags, NULL, NULL);
2847 }
2848 EXPORT_SYMBOL_GPL(rpc_call_null);
2849
2850 static int rpc_ping(struct rpc_clnt *clnt)
2851 {
2852         struct rpc_task *task;
2853         int status;
2854
2855         if (clnt->cl_auth->au_ops->ping)
2856                 return clnt->cl_auth->au_ops->ping(clnt);
2857
2858         task = rpc_call_null_helper(clnt, NULL, NULL, 0, NULL, NULL);
2859         if (IS_ERR(task))
2860                 return PTR_ERR(task);
2861         status = task->tk_status;
2862         rpc_put_task(task);
2863         return status;
2864 }
2865
2866 static int rpc_ping_noreply(struct rpc_clnt *clnt)
2867 {
2868         struct rpc_message msg = {
2869                 .rpc_proc = &rpcproc_null_noreply,
2870         };
2871         struct rpc_task_setup task_setup_data = {
2872                 .rpc_client = clnt,
2873                 .rpc_message = &msg,
2874                 .callback_ops = &rpc_null_ops,
2875                 .flags = RPC_TASK_SOFT | RPC_TASK_SOFTCONN | RPC_TASK_NULLCREDS,
2876         };
2877         struct rpc_task *task;
2878         int status;
2879
2880         task = rpc_run_task(&task_setup_data);
2881         if (IS_ERR(task))
2882                 return PTR_ERR(task);
2883         status = task->tk_status;
2884         rpc_put_task(task);
2885         return status;
2886 }
2887
2888 struct rpc_cb_add_xprt_calldata {
2889         struct rpc_xprt_switch *xps;
2890         struct rpc_xprt *xprt;
2891 };
2892
2893 static void rpc_cb_add_xprt_done(struct rpc_task *task, void *calldata)
2894 {
2895         struct rpc_cb_add_xprt_calldata *data = calldata;
2896
2897         if (task->tk_status == 0)
2898                 rpc_xprt_switch_add_xprt(data->xps, data->xprt);
2899 }
2900
2901 static void rpc_cb_add_xprt_release(void *calldata)
2902 {
2903         struct rpc_cb_add_xprt_calldata *data = calldata;
2904
2905         xprt_put(data->xprt);
2906         xprt_switch_put(data->xps);
2907         kfree(data);
2908 }
2909
2910 static const struct rpc_call_ops rpc_cb_add_xprt_call_ops = {
2911         .rpc_call_prepare = rpc_null_call_prepare,
2912         .rpc_call_done = rpc_cb_add_xprt_done,
2913         .rpc_release = rpc_cb_add_xprt_release,
2914 };
2915
2916 /**
2917  * rpc_clnt_test_and_add_xprt - Test and add a new transport to a rpc_clnt
2918  * @clnt: pointer to struct rpc_clnt
2919  * @xps: pointer to struct rpc_xprt_switch,
2920  * @xprt: pointer struct rpc_xprt
2921  * @in_max_connect: pointer to the max_connect value for the passed in xprt transport
2922  */
2923 int rpc_clnt_test_and_add_xprt(struct rpc_clnt *clnt,
2924                 struct rpc_xprt_switch *xps, struct rpc_xprt *xprt,
2925                 void *in_max_connect)
2926 {
2927         struct rpc_cb_add_xprt_calldata *data;
2928         struct rpc_task *task;
2929         int max_connect = clnt->cl_max_connect;
2930
2931         if (in_max_connect)
2932                 max_connect = *(int *)in_max_connect;
2933         if (xps->xps_nunique_destaddr_xprts + 1 > max_connect) {
2934                 rcu_read_lock();
2935                 pr_warn("SUNRPC: reached max allowed number (%d) did not add "
2936                         "transport to server: %s\n", max_connect,
2937                         rpc_peeraddr2str(clnt, RPC_DISPLAY_ADDR));
2938                 rcu_read_unlock();
2939                 return -EINVAL;
2940         }
2941
2942         data = kmalloc(sizeof(*data), GFP_KERNEL);
2943         if (!data)
2944                 return -ENOMEM;
2945         data->xps = xprt_switch_get(xps);
2946         data->xprt = xprt_get(xprt);
2947         if (rpc_xprt_switch_has_addr(data->xps, (struct sockaddr *)&xprt->addr)) {
2948                 rpc_cb_add_xprt_release(data);
2949                 goto success;
2950         }
2951
2952         task = rpc_call_null_helper(clnt, xprt, NULL, RPC_TASK_ASYNC,
2953                         &rpc_cb_add_xprt_call_ops, data);
2954         if (IS_ERR(task))
2955                 return PTR_ERR(task);
2956
2957         data->xps->xps_nunique_destaddr_xprts++;
2958         rpc_put_task(task);
2959 success:
2960         return 1;
2961 }
2962 EXPORT_SYMBOL_GPL(rpc_clnt_test_and_add_xprt);
2963
2964 static int rpc_clnt_add_xprt_helper(struct rpc_clnt *clnt,
2965                                     struct rpc_xprt *xprt,
2966                                     struct rpc_add_xprt_test *data)
2967 {
2968         struct rpc_task *task;
2969         int status = -EADDRINUSE;
2970
2971         /* Test the connection */
2972         task = rpc_call_null_helper(clnt, xprt, NULL, 0, NULL, NULL);
2973         if (IS_ERR(task))
2974                 return PTR_ERR(task);
2975
2976         status = task->tk_status;
2977         rpc_put_task(task);
2978
2979         if (status < 0)
2980                 return status;
2981
2982         /* rpc_xprt_switch and rpc_xprt are deferrenced by add_xprt_test() */
2983         data->add_xprt_test(clnt, xprt, data->data);
2984
2985         return 0;
2986 }
2987
2988 /**
2989  * rpc_clnt_setup_test_and_add_xprt()
2990  *
2991  * This is an rpc_clnt_add_xprt setup() function which returns 1 so:
2992  *   1) caller of the test function must dereference the rpc_xprt_switch
2993  *   and the rpc_xprt.
2994  *   2) test function must call rpc_xprt_switch_add_xprt, usually in
2995  *   the rpc_call_done routine.
2996  *
2997  * Upon success (return of 1), the test function adds the new
2998  * transport to the rpc_clnt xprt switch
2999  *
3000  * @clnt: struct rpc_clnt to get the new transport
3001  * @xps:  the rpc_xprt_switch to hold the new transport
3002  * @xprt: the rpc_xprt to test
3003  * @data: a struct rpc_add_xprt_test pointer that holds the test function
3004  *        and test function call data
3005  */
3006 int rpc_clnt_setup_test_and_add_xprt(struct rpc_clnt *clnt,
3007                                      struct rpc_xprt_switch *xps,
3008                                      struct rpc_xprt *xprt,
3009                                      void *data)
3010 {
3011         int status = -EADDRINUSE;
3012
3013         xprt = xprt_get(xprt);
3014         xprt_switch_get(xps);
3015
3016         if (rpc_xprt_switch_has_addr(xps, (struct sockaddr *)&xprt->addr))
3017                 goto out_err;
3018
3019         status = rpc_clnt_add_xprt_helper(clnt, xprt, data);
3020         if (status < 0)
3021                 goto out_err;
3022
3023         status = 1;
3024 out_err:
3025         xprt_put(xprt);
3026         xprt_switch_put(xps);
3027         if (status < 0)
3028                 pr_info("RPC:   rpc_clnt_test_xprt failed: %d addr %s not "
3029                         "added\n", status,
3030                         xprt->address_strings[RPC_DISPLAY_ADDR]);
3031         /* so that rpc_clnt_add_xprt does not call rpc_xprt_switch_add_xprt */
3032         return status;
3033 }
3034 EXPORT_SYMBOL_GPL(rpc_clnt_setup_test_and_add_xprt);
3035
3036 /**
3037  * rpc_clnt_add_xprt - Add a new transport to a rpc_clnt
3038  * @clnt: pointer to struct rpc_clnt
3039  * @xprtargs: pointer to struct xprt_create
3040  * @setup: callback to test and/or set up the connection
3041  * @data: pointer to setup function data
3042  *
3043  * Creates a new transport using the parameters set in args and
3044  * adds it to clnt.
3045  * If ping is set, then test that connectivity succeeds before
3046  * adding the new transport.
3047  *
3048  */
3049 int rpc_clnt_add_xprt(struct rpc_clnt *clnt,
3050                 struct xprt_create *xprtargs,
3051                 int (*setup)(struct rpc_clnt *,
3052                         struct rpc_xprt_switch *,
3053                         struct rpc_xprt *,
3054                         void *),
3055                 void *data)
3056 {
3057         struct rpc_xprt_switch *xps;
3058         struct rpc_xprt *xprt;
3059         unsigned long connect_timeout;
3060         unsigned long reconnect_timeout;
3061         unsigned char resvport, reuseport;
3062         int ret = 0, ident;
3063
3064         rcu_read_lock();
3065         xps = xprt_switch_get(rcu_dereference(clnt->cl_xpi.xpi_xpswitch));
3066         xprt = xprt_iter_xprt(&clnt->cl_xpi);
3067         if (xps == NULL || xprt == NULL) {
3068                 rcu_read_unlock();
3069                 xprt_switch_put(xps);
3070                 return -EAGAIN;
3071         }
3072         resvport = xprt->resvport;
3073         reuseport = xprt->reuseport;
3074         connect_timeout = xprt->connect_timeout;
3075         reconnect_timeout = xprt->max_reconnect_timeout;
3076         ident = xprt->xprt_class->ident;
3077         rcu_read_unlock();
3078
3079         if (!xprtargs->ident)
3080                 xprtargs->ident = ident;
3081         xprtargs->xprtsec = clnt->cl_xprtsec;
3082         xprt = xprt_create_transport(xprtargs);
3083         if (IS_ERR(xprt)) {
3084                 ret = PTR_ERR(xprt);
3085                 goto out_put_switch;
3086         }
3087         xprt->resvport = resvport;
3088         xprt->reuseport = reuseport;
3089
3090         if (xprtargs->connect_timeout)
3091                 connect_timeout = xprtargs->connect_timeout;
3092         if (xprtargs->reconnect_timeout)
3093                 reconnect_timeout = xprtargs->reconnect_timeout;
3094         if (xprt->ops->set_connect_timeout != NULL)
3095                 xprt->ops->set_connect_timeout(xprt,
3096                                 connect_timeout,
3097                                 reconnect_timeout);
3098
3099         rpc_xprt_switch_set_roundrobin(xps);
3100         if (setup) {
3101                 ret = setup(clnt, xps, xprt, data);
3102                 if (ret != 0)
3103                         goto out_put_xprt;
3104         }
3105         rpc_xprt_switch_add_xprt(xps, xprt);
3106 out_put_xprt:
3107         xprt_put(xprt);
3108 out_put_switch:
3109         xprt_switch_put(xps);
3110         return ret;
3111 }
3112 EXPORT_SYMBOL_GPL(rpc_clnt_add_xprt);
3113
3114 static int rpc_xprt_probe_trunked(struct rpc_clnt *clnt,
3115                                   struct rpc_xprt *xprt,
3116                                   struct rpc_add_xprt_test *data)
3117 {
3118         struct rpc_xprt_switch *xps;
3119         struct rpc_xprt *main_xprt;
3120         int status = 0;
3121
3122         xprt_get(xprt);
3123
3124         rcu_read_lock();
3125         main_xprt = xprt_get(rcu_dereference(clnt->cl_xprt));
3126         xps = xprt_switch_get(rcu_dereference(clnt->cl_xpi.xpi_xpswitch));
3127         status = rpc_cmp_addr_port((struct sockaddr *)&xprt->addr,
3128                                    (struct sockaddr *)&main_xprt->addr);
3129         rcu_read_unlock();
3130         xprt_put(main_xprt);
3131         if (status || !test_bit(XPRT_OFFLINE, &xprt->state))
3132                 goto out;
3133
3134         status = rpc_clnt_add_xprt_helper(clnt, xprt, data);
3135 out:
3136         xprt_put(xprt);
3137         xprt_switch_put(xps);
3138         return status;
3139 }
3140
3141 /* rpc_clnt_probe_trunked_xprt -- probe offlined transport for session trunking
3142  * @clnt rpc_clnt structure
3143  *
3144  * For each offlined transport found in the rpc_clnt structure call
3145  * the function rpc_xprt_probe_trunked() which will determine if this
3146  * transport still belongs to the trunking group.
3147  */
3148 void rpc_clnt_probe_trunked_xprts(struct rpc_clnt *clnt,
3149                                   struct rpc_add_xprt_test *data)
3150 {
3151         struct rpc_xprt_iter xpi;
3152         int ret;
3153
3154         ret = rpc_clnt_xprt_iter_offline_init(clnt, &xpi);
3155         if (ret)
3156                 return;
3157         for (;;) {
3158                 struct rpc_xprt *xprt = xprt_iter_get_next(&xpi);
3159
3160                 if (!xprt)
3161                         break;
3162                 ret = rpc_xprt_probe_trunked(clnt, xprt, data);
3163                 xprt_put(xprt);
3164                 if (ret < 0)
3165                         break;
3166                 xprt_iter_rewind(&xpi);
3167         }
3168         xprt_iter_destroy(&xpi);
3169 }
3170 EXPORT_SYMBOL_GPL(rpc_clnt_probe_trunked_xprts);
3171
3172 static int rpc_xprt_offline(struct rpc_clnt *clnt,
3173                             struct rpc_xprt *xprt,
3174                             void *data)
3175 {
3176         struct rpc_xprt *main_xprt;
3177         struct rpc_xprt_switch *xps;
3178         int err = 0;
3179
3180         xprt_get(xprt);
3181
3182         rcu_read_lock();
3183         main_xprt = xprt_get(rcu_dereference(clnt->cl_xprt));
3184         xps = xprt_switch_get(rcu_dereference(clnt->cl_xpi.xpi_xpswitch));
3185         err = rpc_cmp_addr_port((struct sockaddr *)&xprt->addr,
3186                                 (struct sockaddr *)&main_xprt->addr);
3187         rcu_read_unlock();
3188         xprt_put(main_xprt);
3189         if (err)
3190                 goto out;
3191
3192         if (wait_on_bit_lock(&xprt->state, XPRT_LOCKED, TASK_KILLABLE)) {
3193                 err = -EINTR;
3194                 goto out;
3195         }
3196         xprt_set_offline_locked(xprt, xps);
3197
3198         xprt_release_write(xprt, NULL);
3199 out:
3200         xprt_put(xprt);
3201         xprt_switch_put(xps);
3202         return err;
3203 }
3204
3205 /* rpc_clnt_manage_trunked_xprts -- offline trunked transports
3206  * @clnt rpc_clnt structure
3207  *
3208  * For each active transport found in the rpc_clnt structure call
3209  * the function rpc_xprt_offline() which will identify trunked transports
3210  * and will mark them offline.
3211  */
3212 void rpc_clnt_manage_trunked_xprts(struct rpc_clnt *clnt)
3213 {
3214         rpc_clnt_iterate_for_each_xprt(clnt, rpc_xprt_offline, NULL);
3215 }
3216 EXPORT_SYMBOL_GPL(rpc_clnt_manage_trunked_xprts);
3217
3218 struct connect_timeout_data {
3219         unsigned long connect_timeout;
3220         unsigned long reconnect_timeout;
3221 };
3222
3223 static int
3224 rpc_xprt_set_connect_timeout(struct rpc_clnt *clnt,
3225                 struct rpc_xprt *xprt,
3226                 void *data)
3227 {
3228         struct connect_timeout_data *timeo = data;
3229
3230         if (xprt->ops->set_connect_timeout)
3231                 xprt->ops->set_connect_timeout(xprt,
3232                                 timeo->connect_timeout,
3233                                 timeo->reconnect_timeout);
3234         return 0;
3235 }
3236
3237 void
3238 rpc_set_connect_timeout(struct rpc_clnt *clnt,
3239                 unsigned long connect_timeout,
3240                 unsigned long reconnect_timeout)
3241 {
3242         struct connect_timeout_data timeout = {
3243                 .connect_timeout = connect_timeout,
3244                 .reconnect_timeout = reconnect_timeout,
3245         };
3246         rpc_clnt_iterate_for_each_xprt(clnt,
3247                         rpc_xprt_set_connect_timeout,
3248                         &timeout);
3249 }
3250 EXPORT_SYMBOL_GPL(rpc_set_connect_timeout);
3251
3252 void rpc_clnt_xprt_switch_put(struct rpc_clnt *clnt)
3253 {
3254         rcu_read_lock();
3255         xprt_switch_put(rcu_dereference(clnt->cl_xpi.xpi_xpswitch));
3256         rcu_read_unlock();
3257 }
3258 EXPORT_SYMBOL_GPL(rpc_clnt_xprt_switch_put);
3259
3260 void rpc_clnt_xprt_set_online(struct rpc_clnt *clnt, struct rpc_xprt *xprt)
3261 {
3262         struct rpc_xprt_switch *xps;
3263
3264         rcu_read_lock();
3265         xps = rcu_dereference(clnt->cl_xpi.xpi_xpswitch);
3266         rcu_read_unlock();
3267         xprt_set_online_locked(xprt, xps);
3268 }
3269
3270 void rpc_clnt_xprt_switch_add_xprt(struct rpc_clnt *clnt, struct rpc_xprt *xprt)
3271 {
3272         if (rpc_clnt_xprt_switch_has_addr(clnt,
3273                 (const struct sockaddr *)&xprt->addr)) {
3274                 return rpc_clnt_xprt_set_online(clnt, xprt);
3275         }
3276         rcu_read_lock();
3277         rpc_xprt_switch_add_xprt(rcu_dereference(clnt->cl_xpi.xpi_xpswitch),
3278                                  xprt);
3279         rcu_read_unlock();
3280 }
3281 EXPORT_SYMBOL_GPL(rpc_clnt_xprt_switch_add_xprt);
3282
3283 void rpc_clnt_xprt_switch_remove_xprt(struct rpc_clnt *clnt, struct rpc_xprt *xprt)
3284 {
3285         struct rpc_xprt_switch *xps;
3286
3287         rcu_read_lock();
3288         xps = rcu_dereference(clnt->cl_xpi.xpi_xpswitch);
3289         rpc_xprt_switch_remove_xprt(rcu_dereference(clnt->cl_xpi.xpi_xpswitch),
3290                                     xprt, 0);
3291         xps->xps_nunique_destaddr_xprts--;
3292         rcu_read_unlock();
3293 }
3294 EXPORT_SYMBOL_GPL(rpc_clnt_xprt_switch_remove_xprt);
3295
3296 bool rpc_clnt_xprt_switch_has_addr(struct rpc_clnt *clnt,
3297                                    const struct sockaddr *sap)
3298 {
3299         struct rpc_xprt_switch *xps;
3300         bool ret;
3301
3302         rcu_read_lock();
3303         xps = rcu_dereference(clnt->cl_xpi.xpi_xpswitch);
3304         ret = rpc_xprt_switch_has_addr(xps, sap);
3305         rcu_read_unlock();
3306         return ret;
3307 }
3308 EXPORT_SYMBOL_GPL(rpc_clnt_xprt_switch_has_addr);
3309
3310 #if IS_ENABLED(CONFIG_SUNRPC_DEBUG)
3311 static void rpc_show_header(void)
3312 {
3313         printk(KERN_INFO "-pid- flgs status -client- --rqstp- "
3314                 "-timeout ---ops--\n");
3315 }
3316
3317 static void rpc_show_task(const struct rpc_clnt *clnt,
3318                           const struct rpc_task *task)
3319 {
3320         const char *rpc_waitq = "none";
3321
3322         if (RPC_IS_QUEUED(task))
3323                 rpc_waitq = rpc_qname(task->tk_waitqueue);
3324
3325         printk(KERN_INFO "%5u %04x %6d %8p %8p %8ld %8p %sv%u %s a:%ps q:%s\n",
3326                 task->tk_pid, task->tk_flags, task->tk_status,
3327                 clnt, task->tk_rqstp, rpc_task_timeout(task), task->tk_ops,
3328                 clnt->cl_program->name, clnt->cl_vers, rpc_proc_name(task),
3329                 task->tk_action, rpc_waitq);
3330 }
3331
3332 void rpc_show_tasks(struct net *net)
3333 {
3334         struct rpc_clnt *clnt;
3335         struct rpc_task *task;
3336         int header = 0;
3337         struct sunrpc_net *sn = net_generic(net, sunrpc_net_id);
3338
3339         spin_lock(&sn->rpc_client_lock);
3340         list_for_each_entry(clnt, &sn->all_clients, cl_clients) {
3341                 spin_lock(&clnt->cl_lock);
3342                 list_for_each_entry(task, &clnt->cl_tasks, tk_task) {
3343                         if (!header) {
3344                                 rpc_show_header();
3345                                 header++;
3346                         }
3347                         rpc_show_task(clnt, task);
3348                 }
3349                 spin_unlock(&clnt->cl_lock);
3350         }
3351         spin_unlock(&sn->rpc_client_lock);
3352 }
3353 #endif
3354
3355 #if IS_ENABLED(CONFIG_SUNRPC_SWAP)
3356 static int
3357 rpc_clnt_swap_activate_callback(struct rpc_clnt *clnt,
3358                 struct rpc_xprt *xprt,
3359                 void *dummy)
3360 {
3361         return xprt_enable_swap(xprt);
3362 }
3363
3364 int
3365 rpc_clnt_swap_activate(struct rpc_clnt *clnt)
3366 {
3367         while (clnt != clnt->cl_parent)
3368                 clnt = clnt->cl_parent;
3369         if (atomic_inc_return(&clnt->cl_swapper) == 1)
3370                 return rpc_clnt_iterate_for_each_xprt(clnt,
3371                                 rpc_clnt_swap_activate_callback, NULL);
3372         return 0;
3373 }
3374 EXPORT_SYMBOL_GPL(rpc_clnt_swap_activate);
3375
3376 static int
3377 rpc_clnt_swap_deactivate_callback(struct rpc_clnt *clnt,
3378                 struct rpc_xprt *xprt,
3379                 void *dummy)
3380 {
3381         xprt_disable_swap(xprt);
3382         return 0;
3383 }
3384
3385 void
3386 rpc_clnt_swap_deactivate(struct rpc_clnt *clnt)
3387 {
3388         while (clnt != clnt->cl_parent)
3389                 clnt = clnt->cl_parent;
3390         if (atomic_dec_if_positive(&clnt->cl_swapper) == 0)
3391                 rpc_clnt_iterate_for_each_xprt(clnt,
3392                                 rpc_clnt_swap_deactivate_callback, NULL);
3393 }
3394 EXPORT_SYMBOL_GPL(rpc_clnt_swap_deactivate);
3395 #endif /* CONFIG_SUNRPC_SWAP */