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