net/sunrpc/clnt.c

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