net/sunrpc/clnt.c

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