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 /**
1168  * rpc_prepare_reply_pages - Prepare to receive a reply data payload into pages
1169  * @req: RPC request to prepare
1170  * @pages: vector of struct page pointers
1171  * @base: offset in first page where receive should start, in bytes
1172  * @len: expected size of the upper layer data payload, in bytes
1173  * @hdrsize: expected size of upper layer reply header, in XDR words
1174  *
1175  */
1176 void rpc_prepare_reply_pages(struct rpc_rqst *req, struct page **pages,
1177                              unsigned int base, unsigned int len,
1178                              unsigned int hdrsize)
1179 {
1180         /* Subtract one to force an extra word of buffer space for the
1181          * payload's XDR pad to fall into the rcv_buf's tail iovec.
1182          */
1183         hdrsize += RPC_REPHDRSIZE + req->rq_cred->cr_auth->au_ralign - 1;
1184
1185         xdr_inline_pages(&req->rq_rcv_buf, hdrsize << 2, pages, base, len);
1186         trace_rpc_reply_pages(req);
1187 }
1188 EXPORT_SYMBOL_GPL(rpc_prepare_reply_pages);
1189
1190 void
1191 rpc_call_start(struct rpc_task *task)
1192 {
1193         task->tk_action = call_start;
1194 }
1195 EXPORT_SYMBOL_GPL(rpc_call_start);
1196
1197 /**
1198  * rpc_peeraddr - extract remote peer address from clnt's xprt
1199  * @clnt: RPC client structure
1200  * @buf: target buffer
1201  * @bufsize: length of target buffer
1202  *
1203  * Returns the number of bytes that are actually in the stored address.
1204  */
1205 size_t rpc_peeraddr(struct rpc_clnt *clnt, struct sockaddr *buf, size_t bufsize)
1206 {
1207         size_t bytes;
1208         struct rpc_xprt *xprt;
1209
1210         rcu_read_lock();
1211         xprt = rcu_dereference(clnt->cl_xprt);
1212
1213         bytes = xprt->addrlen;
1214         if (bytes > bufsize)
1215                 bytes = bufsize;
1216         memcpy(buf, &xprt->addr, bytes);
1217         rcu_read_unlock();
1218
1219         return bytes;
1220 }
1221 EXPORT_SYMBOL_GPL(rpc_peeraddr);
1222
1223 /**
1224  * rpc_peeraddr2str - return remote peer address in printable format
1225  * @clnt: RPC client structure
1226  * @format: address format
1227  *
1228  * NB: the lifetime of the memory referenced by the returned pointer is
1229  * the same as the rpc_xprt itself.  As long as the caller uses this
1230  * pointer, it must hold the RCU read lock.
1231  */
1232 const char *rpc_peeraddr2str(struct rpc_clnt *clnt,
1233                              enum rpc_display_format_t format)
1234 {
1235         struct rpc_xprt *xprt;
1236
1237         xprt = rcu_dereference(clnt->cl_xprt);
1238
1239         if (xprt->address_strings[format] != NULL)
1240                 return xprt->address_strings[format];
1241         else
1242                 return "unprintable";
1243 }
1244 EXPORT_SYMBOL_GPL(rpc_peeraddr2str);
1245
1246 static const struct sockaddr_in rpc_inaddr_loopback = {
1247         .sin_family             = AF_INET,
1248         .sin_addr.s_addr        = htonl(INADDR_ANY),
1249 };
1250
1251 static const struct sockaddr_in6 rpc_in6addr_loopback = {
1252         .sin6_family            = AF_INET6,
1253         .sin6_addr              = IN6ADDR_ANY_INIT,
1254 };
1255
1256 /*
1257  * Try a getsockname() on a connected datagram socket.  Using a
1258  * connected datagram socket prevents leaving a socket in TIME_WAIT.
1259  * This conserves the ephemeral port number space.
1260  *
1261  * Returns zero and fills in "buf" if successful; otherwise, a
1262  * negative errno is returned.
1263  */
1264 static int rpc_sockname(struct net *net, struct sockaddr *sap, size_t salen,
1265                         struct sockaddr *buf)
1266 {
1267         struct socket *sock;
1268         int err;
1269
1270         err = __sock_create(net, sap->sa_family,
1271                                 SOCK_DGRAM, IPPROTO_UDP, &sock, 1);
1272         if (err < 0) {
1273                 dprintk("RPC:       can't create UDP socket (%d)\n", err);
1274                 goto out;
1275         }
1276
1277         switch (sap->sa_family) {
1278         case AF_INET:
1279                 err = kernel_bind(sock,
1280                                 (struct sockaddr *)&rpc_inaddr_loopback,
1281                                 sizeof(rpc_inaddr_loopback));
1282                 break;
1283         case AF_INET6:
1284                 err = kernel_bind(sock,
1285                                 (struct sockaddr *)&rpc_in6addr_loopback,
1286                                 sizeof(rpc_in6addr_loopback));
1287                 break;
1288         default:
1289                 err = -EAFNOSUPPORT;
1290                 goto out;
1291         }
1292         if (err < 0) {
1293                 dprintk("RPC:       can't bind UDP socket (%d)\n", err);
1294                 goto out_release;
1295         }
1296
1297         err = kernel_connect(sock, sap, salen, 0);
1298         if (err < 0) {
1299                 dprintk("RPC:       can't connect UDP socket (%d)\n", err);
1300                 goto out_release;
1301         }
1302
1303         err = kernel_getsockname(sock, buf);
1304         if (err < 0) {
1305                 dprintk("RPC:       getsockname failed (%d)\n", err);
1306                 goto out_release;
1307         }
1308
1309         err = 0;
1310         if (buf->sa_family == AF_INET6) {
1311                 struct sockaddr_in6 *sin6 = (struct sockaddr_in6 *)buf;
1312                 sin6->sin6_scope_id = 0;
1313         }
1314         dprintk("RPC:       %s succeeded\n", __func__);
1315
1316 out_release:
1317         sock_release(sock);
1318 out:
1319         return err;
1320 }
1321
1322 /*
1323  * Scraping a connected socket failed, so we don't have a useable
1324  * local address.  Fallback: generate an address that will prevent
1325  * the server from calling us back.
1326  *
1327  * Returns zero and fills in "buf" if successful; otherwise, a
1328  * negative errno is returned.
1329  */
1330 static int rpc_anyaddr(int family, struct sockaddr *buf, size_t buflen)
1331 {
1332         switch (family) {
1333         case AF_INET:
1334                 if (buflen < sizeof(rpc_inaddr_loopback))
1335                         return -EINVAL;
1336                 memcpy(buf, &rpc_inaddr_loopback,
1337                                 sizeof(rpc_inaddr_loopback));
1338                 break;
1339         case AF_INET6:
1340                 if (buflen < sizeof(rpc_in6addr_loopback))
1341                         return -EINVAL;
1342                 memcpy(buf, &rpc_in6addr_loopback,
1343                                 sizeof(rpc_in6addr_loopback));
1344                 break;
1345         default:
1346                 dprintk("RPC:       %s: address family not supported\n",
1347                         __func__);
1348                 return -EAFNOSUPPORT;
1349         }
1350         dprintk("RPC:       %s: succeeded\n", __func__);
1351         return 0;
1352 }
1353
1354 /**
1355  * rpc_localaddr - discover local endpoint address for an RPC client
1356  * @clnt: RPC client structure
1357  * @buf: target buffer
1358  * @buflen: size of target buffer, in bytes
1359  *
1360  * Returns zero and fills in "buf" and "buflen" if successful;
1361  * otherwise, a negative errno is returned.
1362  *
1363  * This works even if the underlying transport is not currently connected,
1364  * or if the upper layer never previously provided a source address.
1365  *
1366  * The result of this function call is transient: multiple calls in
1367  * succession may give different results, depending on how local
1368  * networking configuration changes over time.
1369  */
1370 int rpc_localaddr(struct rpc_clnt *clnt, struct sockaddr *buf, size_t buflen)
1371 {
1372         struct sockaddr_storage address;
1373         struct sockaddr *sap = (struct sockaddr *)&address;
1374         struct rpc_xprt *xprt;
1375         struct net *net;
1376         size_t salen;
1377         int err;
1378
1379         rcu_read_lock();
1380         xprt = rcu_dereference(clnt->cl_xprt);
1381         salen = xprt->addrlen;
1382         memcpy(sap, &xprt->addr, salen);
1383         net = get_net(xprt->xprt_net);
1384         rcu_read_unlock();
1385
1386         rpc_set_port(sap, 0);
1387         err = rpc_sockname(net, sap, salen, buf);
1388         put_net(net);
1389         if (err != 0)
1390                 /* Couldn't discover local address, return ANYADDR */
1391                 return rpc_anyaddr(sap->sa_family, buf, buflen);
1392         return 0;
1393 }
1394 EXPORT_SYMBOL_GPL(rpc_localaddr);
1395
1396 void
1397 rpc_setbufsize(struct rpc_clnt *clnt, unsigned int sndsize, unsigned int rcvsize)
1398 {
1399         struct rpc_xprt *xprt;
1400
1401         rcu_read_lock();
1402         xprt = rcu_dereference(clnt->cl_xprt);
1403         if (xprt->ops->set_buffer_size)
1404                 xprt->ops->set_buffer_size(xprt, sndsize, rcvsize);
1405         rcu_read_unlock();
1406 }
1407 EXPORT_SYMBOL_GPL(rpc_setbufsize);
1408
1409 /**
1410  * rpc_net_ns - Get the network namespace for this RPC client
1411  * @clnt: RPC client to query
1412  *
1413  */
1414 struct net *rpc_net_ns(struct rpc_clnt *clnt)
1415 {
1416         struct net *ret;
1417
1418         rcu_read_lock();
1419         ret = rcu_dereference(clnt->cl_xprt)->xprt_net;
1420         rcu_read_unlock();
1421         return ret;
1422 }
1423 EXPORT_SYMBOL_GPL(rpc_net_ns);
1424
1425 /**
1426  * rpc_max_payload - Get maximum payload size for a transport, in bytes
1427  * @clnt: RPC client to query
1428  *
1429  * For stream transports, this is one RPC record fragment (see RFC
1430  * 1831), as we don't support multi-record requests yet.  For datagram
1431  * transports, this is the size of an IP packet minus the IP, UDP, and
1432  * RPC header sizes.
1433  */
1434 size_t rpc_max_payload(struct rpc_clnt *clnt)
1435 {
1436         size_t ret;
1437
1438         rcu_read_lock();
1439         ret = rcu_dereference(clnt->cl_xprt)->max_payload;
1440         rcu_read_unlock();
1441         return ret;
1442 }
1443 EXPORT_SYMBOL_GPL(rpc_max_payload);
1444
1445 /**
1446  * rpc_max_bc_payload - Get maximum backchannel payload size, in bytes
1447  * @clnt: RPC client to query
1448  */
1449 size_t rpc_max_bc_payload(struct rpc_clnt *clnt)
1450 {
1451         struct rpc_xprt *xprt;
1452         size_t ret;
1453
1454         rcu_read_lock();
1455         xprt = rcu_dereference(clnt->cl_xprt);
1456         ret = xprt->ops->bc_maxpayload(xprt);
1457         rcu_read_unlock();
1458         return ret;
1459 }
1460 EXPORT_SYMBOL_GPL(rpc_max_bc_payload);
1461
1462 /**
1463  * rpc_force_rebind - force transport to check that remote port is unchanged
1464  * @clnt: client to rebind
1465  *
1466  */
1467 void rpc_force_rebind(struct rpc_clnt *clnt)
1468 {
1469         if (clnt->cl_autobind) {
1470                 rcu_read_lock();
1471                 xprt_clear_bound(rcu_dereference(clnt->cl_xprt));
1472                 rcu_read_unlock();
1473         }
1474 }
1475 EXPORT_SYMBOL_GPL(rpc_force_rebind);
1476
1477 /*
1478  * Restart an (async) RPC call from the call_prepare state.
1479  * Usually called from within the exit handler.
1480  */
1481 int
1482 rpc_restart_call_prepare(struct rpc_task *task)
1483 {
1484         if (RPC_ASSASSINATED(task))
1485                 return 0;
1486         task->tk_action = call_start;
1487         task->tk_status = 0;
1488         if (task->tk_ops->rpc_call_prepare != NULL)
1489                 task->tk_action = rpc_prepare_task;
1490         return 1;
1491 }
1492 EXPORT_SYMBOL_GPL(rpc_restart_call_prepare);
1493
1494 /*
1495  * Restart an (async) RPC call. Usually called from within the
1496  * exit handler.
1497  */
1498 int
1499 rpc_restart_call(struct rpc_task *task)
1500 {
1501         if (RPC_ASSASSINATED(task))
1502                 return 0;
1503         task->tk_action = call_start;
1504         task->tk_status = 0;
1505         return 1;
1506 }
1507 EXPORT_SYMBOL_GPL(rpc_restart_call);
1508
1509 const char
1510 *rpc_proc_name(const struct rpc_task *task)
1511 {
1512         const struct rpc_procinfo *proc = task->tk_msg.rpc_proc;
1513
1514         if (proc) {
1515                 if (proc->p_name)
1516                         return proc->p_name;
1517                 else
1518                         return "NULL";
1519         } else
1520                 return "no proc";
1521 }
1522
1523 /*
1524  * 0.  Initial state
1525  *
1526  *     Other FSM states can be visited zero or more times, but
1527  *     this state is visited exactly once for each RPC.
1528  */
1529 static void
1530 call_start(struct rpc_task *task)
1531 {
1532         struct rpc_clnt *clnt = task->tk_client;
1533         int idx = task->tk_msg.rpc_proc->p_statidx;
1534
1535         trace_rpc_request(task);
1536         dprintk("RPC: %5u call_start %s%d proc %s (%s)\n", task->tk_pid,
1537                         clnt->cl_program->name, clnt->cl_vers,
1538                         rpc_proc_name(task),
1539                         (RPC_IS_ASYNC(task) ? "async" : "sync"));
1540
1541         /* Increment call count (version might not be valid for ping) */
1542         if (clnt->cl_program->version[clnt->cl_vers])
1543                 clnt->cl_program->version[clnt->cl_vers]->counts[idx]++;
1544         clnt->cl_stats->rpccnt++;
1545         task->tk_action = call_reserve;
1546         rpc_task_set_transport(task, clnt);
1547 }
1548
1549 /*
1550  * 1.   Reserve an RPC call slot
1551  */
1552 static void
1553 call_reserve(struct rpc_task *task)
1554 {
1555         dprint_status(task);
1556
1557         task->tk_status  = 0;
1558         task->tk_action  = call_reserveresult;
1559         xprt_reserve(task);
1560 }
1561
1562 static void call_retry_reserve(struct rpc_task *task);
1563
1564 /*
1565  * 1b.  Grok the result of xprt_reserve()
1566  */
1567 static void
1568 call_reserveresult(struct rpc_task *task)
1569 {
1570         int status = task->tk_status;
1571
1572         dprint_status(task);
1573
1574         /*
1575          * After a call to xprt_reserve(), we must have either
1576          * a request slot or else an error status.
1577          */
1578         task->tk_status = 0;
1579         if (status >= 0) {
1580                 if (task->tk_rqstp) {
1581                         task->tk_action = call_refresh;
1582                         return;
1583                 }
1584
1585                 printk(KERN_ERR "%s: status=%d, but no request slot, exiting\n",
1586                                 __func__, status);
1587                 rpc_exit(task, -EIO);
1588                 return;
1589         }
1590
1591         /*
1592          * Even though there was an error, we may have acquired
1593          * a request slot somehow.  Make sure not to leak it.
1594          */
1595         if (task->tk_rqstp) {
1596                 printk(KERN_ERR "%s: status=%d, request allocated anyway\n",
1597                                 __func__, status);
1598                 xprt_release(task);
1599         }
1600
1601         switch (status) {
1602         case -ENOMEM:
1603                 rpc_delay(task, HZ >> 2);
1604                 /* fall through */
1605         case -EAGAIN:   /* woken up; retry */
1606                 task->tk_action = call_retry_reserve;
1607                 return;
1608         case -EIO:      /* probably a shutdown */
1609                 break;
1610         default:
1611                 printk(KERN_ERR "%s: unrecognized error %d, exiting\n",
1612                                 __func__, status);
1613                 break;
1614         }
1615         rpc_exit(task, status);
1616 }
1617
1618 /*
1619  * 1c.  Retry reserving an RPC call slot
1620  */
1621 static void
1622 call_retry_reserve(struct rpc_task *task)
1623 {
1624         dprint_status(task);
1625
1626         task->tk_status  = 0;
1627         task->tk_action  = call_reserveresult;
1628         xprt_retry_reserve(task);
1629 }
1630
1631 /*
1632  * 2.   Bind and/or refresh the credentials
1633  */
1634 static void
1635 call_refresh(struct rpc_task *task)
1636 {
1637         dprint_status(task);
1638
1639         task->tk_action = call_refreshresult;
1640         task->tk_status = 0;
1641         task->tk_client->cl_stats->rpcauthrefresh++;
1642         rpcauth_refreshcred(task);
1643 }
1644
1645 /*
1646  * 2a.  Process the results of a credential refresh
1647  */
1648 static void
1649 call_refreshresult(struct rpc_task *task)
1650 {
1651         int status = task->tk_status;
1652
1653         dprint_status(task);
1654
1655         task->tk_status = 0;
1656         task->tk_action = call_refresh;
1657         switch (status) {
1658         case 0:
1659                 if (rpcauth_uptodatecred(task)) {
1660                         task->tk_action = call_allocate;
1661                         return;
1662                 }
1663                 /* Use rate-limiting and a max number of retries if refresh
1664                  * had status 0 but failed to update the cred.
1665                  */
1666                 /* fall through */
1667         case -ETIMEDOUT:
1668                 rpc_delay(task, 3*HZ);
1669                 /* fall through */
1670         case -EAGAIN:
1671                 status = -EACCES;
1672                 /* fall through */
1673         case -EKEYEXPIRED:
1674                 if (!task->tk_cred_retry)
1675                         break;
1676                 task->tk_cred_retry--;
1677                 dprintk("RPC: %5u %s: retry refresh creds\n",
1678                                 task->tk_pid, __func__);
1679                 return;
1680         }
1681         dprintk("RPC: %5u %s: refresh creds failed with error %d\n",
1682                                 task->tk_pid, __func__, status);
1683         rpc_exit(task, status);
1684 }
1685
1686 /*
1687  * 2b.  Allocate the buffer. For details, see sched.c:rpc_malloc.
1688  *      (Note: buffer memory is freed in xprt_release).
1689  */
1690 static void
1691 call_allocate(struct rpc_task *task)
1692 {
1693         const struct rpc_auth *auth = task->tk_rqstp->rq_cred->cr_auth;
1694         struct rpc_rqst *req = task->tk_rqstp;
1695         struct rpc_xprt *xprt = req->rq_xprt;
1696         const struct rpc_procinfo *proc = task->tk_msg.rpc_proc;
1697         int status;
1698
1699         dprint_status(task);
1700
1701         task->tk_status = 0;
1702         task->tk_action = call_encode;
1703
1704         if (req->rq_buffer)
1705                 return;
1706
1707         if (proc->p_proc != 0) {
1708                 BUG_ON(proc->p_arglen == 0);
1709                 if (proc->p_decode != NULL)
1710                         BUG_ON(proc->p_replen == 0);
1711         }
1712
1713         /*
1714          * Calculate the size (in quads) of the RPC call
1715          * and reply headers, and convert both values
1716          * to byte sizes.
1717          */
1718         req->rq_callsize = RPC_CALLHDRSIZE + (auth->au_cslack << 1) +
1719                            proc->p_arglen;
1720         req->rq_callsize <<= 2;
1721         req->rq_rcvsize = RPC_REPHDRSIZE + auth->au_rslack + proc->p_replen;
1722         req->rq_rcvsize <<= 2;
1723
1724         status = xprt->ops->buf_alloc(task);
1725         xprt_inject_disconnect(xprt);
1726         if (status == 0)
1727                 return;
1728         if (status != -ENOMEM) {
1729                 rpc_exit(task, status);
1730                 return;
1731         }
1732
1733         dprintk("RPC: %5u rpc_buffer allocation failed\n", task->tk_pid);
1734
1735         if (RPC_IS_ASYNC(task) || !fatal_signal_pending(current)) {
1736                 task->tk_action = call_allocate;
1737                 rpc_delay(task, HZ>>4);
1738                 return;
1739         }
1740
1741         rpc_exit(task, -ERESTARTSYS);
1742 }
1743
1744 static int
1745 rpc_task_need_encode(struct rpc_task *task)
1746 {
1747         return test_bit(RPC_TASK_NEED_XMIT, &task->tk_runstate) == 0 &&
1748                 (!(task->tk_flags & RPC_TASK_SENT) ||
1749                  !(task->tk_flags & RPC_TASK_NO_RETRANS_TIMEOUT) ||
1750                  xprt_request_need_retransmit(task));
1751 }
1752
1753 static void
1754 rpc_xdr_encode(struct rpc_task *task)
1755 {
1756         struct rpc_rqst *req = task->tk_rqstp;
1757         struct xdr_stream xdr;
1758
1759         xdr_buf_init(&req->rq_snd_buf,
1760                      req->rq_buffer,
1761                      req->rq_callsize);
1762         xdr_buf_init(&req->rq_rcv_buf,
1763                      req->rq_rbuffer,
1764                      req->rq_rcvsize);
1765
1766         req->rq_snd_buf.head[0].iov_len = 0;
1767         xdr_init_encode(&xdr, &req->rq_snd_buf,
1768                         req->rq_snd_buf.head[0].iov_base, req);
1769         if (rpc_encode_header(task, &xdr))
1770                 return;
1771
1772         task->tk_status = rpcauth_wrap_req(task, &xdr);
1773 }
1774
1775 /*
1776  * 3.   Encode arguments of an RPC call
1777  */
1778 static void
1779 call_encode(struct rpc_task *task)
1780 {
1781         if (!rpc_task_need_encode(task))
1782                 goto out;
1783         dprint_status(task);
1784         /* Encode here so that rpcsec_gss can use correct sequence number. */
1785         rpc_xdr_encode(task);
1786         /* Did the encode result in an error condition? */
1787         if (task->tk_status != 0) {
1788                 /* Was the error nonfatal? */
1789                 switch (task->tk_status) {
1790                 case -EAGAIN:
1791                 case -ENOMEM:
1792                         rpc_delay(task, HZ >> 4);
1793                         break;
1794                 case -EKEYEXPIRED:
1795                         task->tk_action = call_refresh;
1796                         break;
1797                 default:
1798                         rpc_exit(task, task->tk_status);
1799                 }
1800                 return;
1801         } else {
1802                 xprt_request_prepare(task->tk_rqstp);
1803         }
1804
1805         /* Add task to reply queue before transmission to avoid races */
1806         if (rpc_reply_expected(task))
1807                 xprt_request_enqueue_receive(task);
1808         xprt_request_enqueue_transmit(task);
1809 out:
1810         task->tk_action = call_bind;
1811 }
1812
1813 /*
1814  * 4.   Get the server port number if not yet set
1815  */
1816 static void
1817 call_bind(struct rpc_task *task)
1818 {
1819         struct rpc_xprt *xprt = task->tk_rqstp->rq_xprt;
1820
1821         dprint_status(task);
1822
1823         task->tk_action = call_connect;
1824         if (!xprt_bound(xprt)) {
1825                 task->tk_action = call_bind_status;
1826                 task->tk_timeout = xprt->bind_timeout;
1827                 xprt->ops->rpcbind(task);
1828         }
1829 }
1830
1831 /*
1832  * 4a.  Sort out bind result
1833  */
1834 static void
1835 call_bind_status(struct rpc_task *task)
1836 {
1837         int status = -EIO;
1838
1839         if (task->tk_status >= 0) {
1840                 dprint_status(task);
1841                 task->tk_status = 0;
1842                 task->tk_action = call_connect;
1843                 return;
1844         }
1845
1846         trace_rpc_bind_status(task);
1847         switch (task->tk_status) {
1848         case -ENOMEM:
1849                 dprintk("RPC: %5u rpcbind out of memory\n", task->tk_pid);
1850                 rpc_delay(task, HZ >> 2);
1851                 goto retry_timeout;
1852         case -EACCES:
1853                 dprintk("RPC: %5u remote rpcbind: RPC program/version "
1854                                 "unavailable\n", task->tk_pid);
1855                 /* fail immediately if this is an RPC ping */
1856                 if (task->tk_msg.rpc_proc->p_proc == 0) {
1857                         status = -EOPNOTSUPP;
1858                         break;
1859                 }
1860                 if (task->tk_rebind_retry == 0)
1861                         break;
1862                 task->tk_rebind_retry--;
1863                 rpc_delay(task, 3*HZ);
1864                 goto retry_timeout;
1865         case -ETIMEDOUT:
1866                 dprintk("RPC: %5u rpcbind request timed out\n",
1867                                 task->tk_pid);
1868                 goto retry_timeout;
1869         case -EPFNOSUPPORT:
1870                 /* server doesn't support any rpcbind version we know of */
1871                 dprintk("RPC: %5u unrecognized remote rpcbind service\n",
1872                                 task->tk_pid);
1873                 break;
1874         case -EPROTONOSUPPORT:
1875                 dprintk("RPC: %5u remote rpcbind version unavailable, retrying\n",
1876                                 task->tk_pid);
1877                 goto retry_timeout;
1878         case -ECONNREFUSED:             /* connection problems */
1879         case -ECONNRESET:
1880         case -ECONNABORTED:
1881         case -ENOTCONN:
1882         case -EHOSTDOWN:
1883         case -ENETDOWN:
1884         case -EHOSTUNREACH:
1885         case -ENETUNREACH:
1886         case -ENOBUFS:
1887         case -EPIPE:
1888                 dprintk("RPC: %5u remote rpcbind unreachable: %d\n",
1889                                 task->tk_pid, task->tk_status);
1890                 if (!RPC_IS_SOFTCONN(task)) {
1891                         rpc_delay(task, 5*HZ);
1892                         goto retry_timeout;
1893                 }
1894                 status = task->tk_status;
1895                 break;
1896         default:
1897                 dprintk("RPC: %5u unrecognized rpcbind error (%d)\n",
1898                                 task->tk_pid, -task->tk_status);
1899         }
1900
1901         rpc_exit(task, status);
1902         return;
1903
1904 retry_timeout:
1905         task->tk_status = 0;
1906         task->tk_action = call_timeout;
1907 }
1908
1909 /*
1910  * 4b.  Connect to the RPC server
1911  */
1912 static void
1913 call_connect(struct rpc_task *task)
1914 {
1915         struct rpc_xprt *xprt = task->tk_rqstp->rq_xprt;
1916
1917         dprintk("RPC: %5u call_connect xprt %p %s connected\n",
1918                         task->tk_pid, xprt,
1919                         (xprt_connected(xprt) ? "is" : "is not"));
1920
1921         task->tk_action = call_transmit;
1922         if (!xprt_connected(xprt)) {
1923                 task->tk_action = call_connect_status;
1924                 if (task->tk_status < 0)
1925                         return;
1926                 if (task->tk_flags & RPC_TASK_NOCONNECT) {
1927                         rpc_exit(task, -ENOTCONN);
1928                         return;
1929                 }
1930                 xprt_connect(task);
1931         }
1932 }
1933
1934 /*
1935  * 4c.  Sort out connect result
1936  */
1937 static void
1938 call_connect_status(struct rpc_task *task)
1939 {
1940         struct rpc_clnt *clnt = task->tk_client;
1941         int status = task->tk_status;
1942
1943         /* Check if the task was already transmitted */
1944         if (!test_bit(RPC_TASK_NEED_XMIT, &task->tk_runstate)) {
1945                 xprt_end_transmit(task);
1946                 task->tk_action = call_transmit_status;
1947                 return;
1948         }
1949
1950         dprint_status(task);
1951
1952         trace_rpc_connect_status(task);
1953         task->tk_status = 0;
1954         switch (status) {
1955         case -ECONNREFUSED:
1956                 /* A positive refusal suggests a rebind is needed. */
1957                 if (RPC_IS_SOFTCONN(task))
1958                         break;
1959                 if (clnt->cl_autobind) {
1960                         rpc_force_rebind(clnt);
1961                         task->tk_action = call_bind;
1962                         return;
1963                 }
1964                 /* fall through */
1965         case -ECONNRESET:
1966         case -ECONNABORTED:
1967         case -ENETDOWN:
1968         case -ENETUNREACH:
1969         case -EHOSTUNREACH:
1970         case -EADDRINUSE:
1971         case -ENOBUFS:
1972         case -EPIPE:
1973                 xprt_conditional_disconnect(task->tk_rqstp->rq_xprt,
1974                                             task->tk_rqstp->rq_connect_cookie);
1975                 if (RPC_IS_SOFTCONN(task))
1976                         break;
1977                 /* retry with existing socket, after a delay */
1978                 rpc_delay(task, 3*HZ);
1979                 /* fall through */
1980         case -ENOTCONN:
1981         case -EAGAIN:
1982                 /* Check for timeouts before looping back to call_bind */
1983         case -ETIMEDOUT:
1984                 task->tk_action = call_timeout;
1985                 return;
1986         case 0:
1987                 clnt->cl_stats->netreconn++;
1988                 task->tk_action = call_transmit;
1989                 return;
1990         }
1991         rpc_exit(task, status);
1992 }
1993
1994 /*
1995  * 5.   Transmit the RPC request, and wait for reply
1996  */
1997 static void
1998 call_transmit(struct rpc_task *task)
1999 {
2000         dprint_status(task);
2001
2002         task->tk_status = 0;
2003         if (test_bit(RPC_TASK_NEED_XMIT, &task->tk_runstate)) {
2004                 if (!xprt_prepare_transmit(task))
2005                         return;
2006                 xprt_transmit(task);
2007         }
2008         task->tk_action = call_transmit_status;
2009         xprt_end_transmit(task);
2010 }
2011
2012 /*
2013  * 5a.  Handle cleanup after a transmission
2014  */
2015 static void
2016 call_transmit_status(struct rpc_task *task)
2017 {
2018         task->tk_action = call_status;
2019
2020         /*
2021          * Common case: success.  Force the compiler to put this
2022          * test first.
2023          */
2024         if (task->tk_status == 0) {
2025                 xprt_request_wait_receive(task);
2026                 return;
2027         }
2028
2029         switch (task->tk_status) {
2030         default:
2031                 dprint_status(task);
2032                 break;
2033         case -EBADMSG:
2034                 task->tk_status = 0;
2035                 task->tk_action = call_encode;
2036                 break;
2037                 /*
2038                  * Special cases: if we've been waiting on the
2039                  * socket's write_space() callback, or if the
2040                  * socket just returned a connection error,
2041                  * then hold onto the transport lock.
2042                  */
2043         case -ENOBUFS:
2044                 rpc_delay(task, HZ>>2);
2045                 /* fall through */
2046         case -EBADSLT:
2047         case -EAGAIN:
2048                 task->tk_action = call_transmit;
2049                 task->tk_status = 0;
2050                 break;
2051         case -ECONNREFUSED:
2052         case -EHOSTDOWN:
2053         case -ENETDOWN:
2054         case -EHOSTUNREACH:
2055         case -ENETUNREACH:
2056         case -EPERM:
2057                 if (RPC_IS_SOFTCONN(task)) {
2058                         if (!task->tk_msg.rpc_proc->p_proc)
2059                                 trace_xprt_ping(task->tk_xprt,
2060                                                 task->tk_status);
2061                         rpc_exit(task, task->tk_status);
2062                         break;
2063                 }
2064                 /* fall through */
2065         case -ECONNRESET:
2066         case -ECONNABORTED:
2067         case -EADDRINUSE:
2068         case -ENOTCONN:
2069         case -EPIPE:
2070                 break;
2071         }
2072 }
2073
2074 #if defined(CONFIG_SUNRPC_BACKCHANNEL)
2075 /*
2076  * 5b.  Send the backchannel RPC reply.  On error, drop the reply.  In
2077  * addition, disconnect on connectivity errors.
2078  */
2079 static void
2080 call_bc_transmit(struct rpc_task *task)
2081 {
2082         struct rpc_rqst *req = task->tk_rqstp;
2083
2084         if (rpc_task_need_encode(task))
2085                 xprt_request_enqueue_transmit(task);
2086         if (!test_bit(RPC_TASK_NEED_XMIT, &task->tk_runstate))
2087                 goto out_wakeup;
2088
2089         if (!xprt_prepare_transmit(task))
2090                 goto out_retry;
2091
2092         if (task->tk_status < 0) {
2093                 printk(KERN_NOTICE "RPC: Could not send backchannel reply "
2094                         "error: %d\n", task->tk_status);
2095                 goto out_done;
2096         }
2097
2098         xprt_transmit(task);
2099
2100         xprt_end_transmit(task);
2101         dprint_status(task);
2102         switch (task->tk_status) {
2103         case 0:
2104                 /* Success */
2105         case -ENETDOWN:
2106         case -EHOSTDOWN:
2107         case -EHOSTUNREACH:
2108         case -ENETUNREACH:
2109         case -ECONNRESET:
2110         case -ECONNREFUSED:
2111         case -EADDRINUSE:
2112         case -ENOTCONN:
2113         case -EPIPE:
2114                 break;
2115         case -EAGAIN:
2116                 goto out_retry;
2117         case -ETIMEDOUT:
2118                 /*
2119                  * Problem reaching the server.  Disconnect and let the
2120                  * forechannel reestablish the connection.  The server will
2121                  * have to retransmit the backchannel request and we'll
2122                  * reprocess it.  Since these ops are idempotent, there's no
2123                  * need to cache our reply at this time.
2124                  */
2125                 printk(KERN_NOTICE "RPC: Could not send backchannel reply "
2126                         "error: %d\n", task->tk_status);
2127                 xprt_conditional_disconnect(req->rq_xprt,
2128                         req->rq_connect_cookie);
2129                 break;
2130         default:
2131                 /*
2132                  * We were unable to reply and will have to drop the
2133                  * request.  The server should reconnect and retransmit.
2134                  */
2135                 WARN_ON_ONCE(task->tk_status == -EAGAIN);
2136                 printk(KERN_NOTICE "RPC: Could not send backchannel reply "
2137                         "error: %d\n", task->tk_status);
2138                 break;
2139         }
2140 out_wakeup:
2141         rpc_wake_up_queued_task(&req->rq_xprt->pending, task);
2142 out_done:
2143         task->tk_action = rpc_exit_task;
2144         return;
2145 out_retry:
2146         task->tk_status = 0;
2147 }
2148 #endif /* CONFIG_SUNRPC_BACKCHANNEL */
2149
2150 /*
2151  * 6.   Sort out the RPC call status
2152  */
2153 static void
2154 call_status(struct rpc_task *task)
2155 {
2156         struct rpc_clnt *clnt = task->tk_client;
2157         int             status;
2158
2159         if (!task->tk_msg.rpc_proc->p_proc)
2160                 trace_xprt_ping(task->tk_xprt, task->tk_status);
2161
2162         dprint_status(task);
2163
2164         status = task->tk_status;
2165         if (status >= 0) {
2166                 task->tk_action = call_decode;
2167                 return;
2168         }
2169
2170         trace_rpc_call_status(task);
2171         task->tk_status = 0;
2172         switch(status) {
2173         case -EHOSTDOWN:
2174         case -ENETDOWN:
2175         case -EHOSTUNREACH:
2176         case -ENETUNREACH:
2177         case -EPERM:
2178                 if (RPC_IS_SOFTCONN(task)) {
2179                         rpc_exit(task, status);
2180                         break;
2181                 }
2182                 /*
2183                  * Delay any retries for 3 seconds, then handle as if it
2184                  * were a timeout.
2185                  */
2186                 rpc_delay(task, 3*HZ);
2187                 /* fall through */
2188         case -ETIMEDOUT:
2189                 task->tk_action = call_timeout;
2190                 break;
2191         case -ECONNREFUSED:
2192         case -ECONNRESET:
2193         case -ECONNABORTED:
2194                 rpc_force_rebind(clnt);
2195                 /* fall through */
2196         case -EADDRINUSE:
2197                 rpc_delay(task, 3*HZ);
2198                 /* fall through */
2199         case -EPIPE:
2200         case -ENOTCONN:
2201         case -EAGAIN:
2202                 task->tk_action = call_encode;
2203                 break;
2204         case -EIO:
2205                 /* shutdown or soft timeout */
2206                 rpc_exit(task, status);
2207                 break;
2208         default:
2209                 if (clnt->cl_chatty)
2210                         printk("%s: RPC call returned error %d\n",
2211                                clnt->cl_program->name, -status);
2212                 rpc_exit(task, status);
2213         }
2214 }
2215
2216 /*
2217  * 6a.  Handle RPC timeout
2218  *      We do not release the request slot, so we keep using the
2219  *      same XID for all retransmits.
2220  */
2221 static void
2222 call_timeout(struct rpc_task *task)
2223 {
2224         struct rpc_clnt *clnt = task->tk_client;
2225
2226         if (xprt_adjust_timeout(task->tk_rqstp) == 0) {
2227                 dprintk("RPC: %5u call_timeout (minor)\n", task->tk_pid);
2228                 goto retry;
2229         }
2230
2231         dprintk("RPC: %5u call_timeout (major)\n", task->tk_pid);
2232         task->tk_timeouts++;
2233
2234         if (RPC_IS_SOFTCONN(task)) {
2235                 rpc_exit(task, -ETIMEDOUT);
2236                 return;
2237         }
2238         if (RPC_IS_SOFT(task)) {
2239                 if (clnt->cl_chatty) {
2240                         printk(KERN_NOTICE "%s: server %s not responding, timed out\n",
2241                                 clnt->cl_program->name,
2242                                 task->tk_xprt->servername);
2243                 }
2244                 if (task->tk_flags & RPC_TASK_TIMEOUT)
2245                         rpc_exit(task, -ETIMEDOUT);
2246                 else
2247                         rpc_exit(task, -EIO);
2248                 return;
2249         }
2250
2251         if (!(task->tk_flags & RPC_CALL_MAJORSEEN)) {
2252                 task->tk_flags |= RPC_CALL_MAJORSEEN;
2253                 if (clnt->cl_chatty) {
2254                         printk(KERN_NOTICE "%s: server %s not responding, still trying\n",
2255                         clnt->cl_program->name,
2256                         task->tk_xprt->servername);
2257                 }
2258         }
2259         rpc_force_rebind(clnt);
2260         /*
2261          * Did our request time out due to an RPCSEC_GSS out-of-sequence
2262          * event? RFC2203 requires the server to drop all such requests.
2263          */
2264         rpcauth_invalcred(task);
2265
2266 retry:
2267         task->tk_action = call_encode;
2268         task->tk_status = 0;
2269 }
2270
2271 /*
2272  * 7.   Decode the RPC reply
2273  */
2274 static void
2275 call_decode(struct rpc_task *task)
2276 {
2277         struct rpc_clnt *clnt = task->tk_client;
2278         struct rpc_rqst *req = task->tk_rqstp;
2279         struct xdr_stream xdr;
2280
2281         dprint_status(task);
2282
2283         if (!task->tk_msg.rpc_proc->p_decode) {
2284                 task->tk_action = rpc_exit_task;
2285                 return;
2286         }
2287
2288         if (task->tk_flags & RPC_CALL_MAJORSEEN) {
2289                 if (clnt->cl_chatty) {
2290                         printk(KERN_NOTICE "%s: server %s OK\n",
2291                                 clnt->cl_program->name,
2292                                 task->tk_xprt->servername);
2293                 }
2294                 task->tk_flags &= ~RPC_CALL_MAJORSEEN;
2295         }
2296
2297         /*
2298          * Ensure that we see all writes made by xprt_complete_rqst()
2299          * before it changed req->rq_reply_bytes_recvd.
2300          */
2301         smp_rmb();
2302         req->rq_rcv_buf.len = req->rq_private_buf.len;
2303
2304         /* Check that the softirq receive buffer is valid */
2305         WARN_ON(memcmp(&req->rq_rcv_buf, &req->rq_private_buf,
2306                                 sizeof(req->rq_rcv_buf)) != 0);
2307
2308         if (req->rq_rcv_buf.len < 12) {
2309                 if (!RPC_IS_SOFT(task)) {
2310                         task->tk_action = call_encode;
2311                         goto out_retry;
2312                 }
2313                 dprintk("RPC:       %s: too small RPC reply size (%d bytes)\n",
2314                                 clnt->cl_program->name, task->tk_status);
2315                 task->tk_action = call_timeout;
2316                 goto out_retry;
2317         }
2318
2319         xdr_init_decode(&xdr, &req->rq_rcv_buf,
2320                         req->rq_rcv_buf.head[0].iov_base, req);
2321         switch (rpc_decode_header(task, &xdr)) {
2322         case 0:
2323                 task->tk_action = rpc_exit_task;
2324                 task->tk_status = rpcauth_unwrap_resp(task, &xdr);
2325                 dprintk("RPC: %5u %s result %d\n",
2326                         task->tk_pid, __func__, task->tk_status);
2327                 return;
2328         case -EAGAIN:
2329 out_retry:
2330                 task->tk_status = 0;
2331                 /* Note: rpc_decode_header() may have freed the RPC slot */
2332                 if (task->tk_rqstp == req) {
2333                         xdr_free_bvec(&req->rq_rcv_buf);
2334                         req->rq_reply_bytes_recvd = 0;
2335                         req->rq_rcv_buf.len = 0;
2336                         if (task->tk_client->cl_discrtry)
2337                                 xprt_conditional_disconnect(req->rq_xprt,
2338                                                             req->rq_connect_cookie);
2339                 }
2340         }
2341 }
2342
2343 static int
2344 rpc_encode_header(struct rpc_task *task, struct xdr_stream *xdr)
2345 {
2346         struct rpc_clnt *clnt = task->tk_client;
2347         struct rpc_rqst *req = task->tk_rqstp;
2348         __be32 *p;
2349         int error;
2350
2351         error = -EMSGSIZE;
2352         p = xdr_reserve_space(xdr, RPC_CALLHDRSIZE << 2);
2353         if (!p)
2354                 goto out_fail;
2355         *p++ = req->rq_xid;
2356         *p++ = rpc_call;
2357         *p++ = cpu_to_be32(RPC_VERSION);
2358         *p++ = cpu_to_be32(clnt->cl_prog);
2359         *p++ = cpu_to_be32(clnt->cl_vers);
2360         *p   = cpu_to_be32(task->tk_msg.rpc_proc->p_proc);
2361
2362         error = rpcauth_marshcred(task, xdr);
2363         if (error < 0)
2364                 goto out_fail;
2365         return 0;
2366 out_fail:
2367         trace_rpc_bad_callhdr(task);
2368         rpc_exit(task, error);
2369         return error;
2370 }
2371
2372 static noinline int
2373 rpc_decode_header(struct rpc_task *task, struct xdr_stream *xdr)
2374 {
2375         struct rpc_clnt *clnt = task->tk_client;
2376         int error = -EACCES;
2377         __be32 *p;
2378
2379         /* RFC-1014 says that the representation of XDR data must be a
2380          * multiple of four bytes
2381          * - if it isn't pointer subtraction in the NFS client may give
2382          *   undefined results
2383          */
2384         if (task->tk_rqstp->rq_rcv_buf.len & 3)
2385                 goto out_badlen;
2386
2387         p = xdr_inline_decode(xdr, 3 * sizeof(*p));
2388         if (!p)
2389                 goto out_unparsable;
2390         p++;    /* skip XID */
2391         if (*p++ != rpc_reply)
2392                 goto out_unparsable;
2393         if (*p++ != rpc_msg_accepted)
2394                 goto out_msg_denied;
2395
2396         error = rpcauth_checkverf(task, xdr);
2397         if (error)
2398                 goto out_verifier;
2399
2400         p = xdr_inline_decode(xdr, sizeof(*p));
2401         if (!p)
2402                 goto out_unparsable;
2403         switch (*p) {
2404         case rpc_success:
2405                 return 0;
2406         case rpc_prog_unavail:
2407                 trace_rpc__prog_unavail(task);
2408                 error = -EPFNOSUPPORT;
2409                 goto out_err;
2410         case rpc_prog_mismatch:
2411                 trace_rpc__prog_mismatch(task);
2412                 error = -EPROTONOSUPPORT;
2413                 goto out_err;
2414         case rpc_proc_unavail:
2415                 trace_rpc__proc_unavail(task);
2416                 error = -EOPNOTSUPP;
2417                 goto out_err;
2418         case rpc_garbage_args:
2419                 trace_rpc__garbage_args(task);
2420                 break;
2421         default:
2422                 trace_rpc__unparsable(task);
2423         }
2424
2425 out_garbage:
2426         clnt->cl_stats->rpcgarbage++;
2427         if (task->tk_garb_retry) {
2428                 task->tk_garb_retry--;
2429                 task->tk_action = call_encode;
2430                 return -EAGAIN;
2431         }
2432 out_err:
2433         rpc_exit(task, error);
2434         return error;
2435
2436 out_badlen:
2437         trace_rpc__unparsable(task);
2438         error = -EIO;
2439         goto out_err;
2440
2441 out_unparsable:
2442         trace_rpc__unparsable(task);
2443         error = -EIO;
2444         goto out_garbage;
2445
2446 out_verifier:
2447         trace_rpc_bad_verifier(task);
2448         goto out_garbage;
2449
2450 out_msg_denied:
2451         p = xdr_inline_decode(xdr, sizeof(*p));
2452         if (!p)
2453                 goto out_unparsable;
2454         switch (*p++) {
2455         case rpc_auth_error:
2456                 break;
2457         case rpc_mismatch:
2458                 trace_rpc__mismatch(task);
2459                 error = -EPROTONOSUPPORT;
2460                 goto out_err;
2461         default:
2462                 trace_rpc__unparsable(task);
2463                 error = -EIO;
2464                 goto out_err;
2465         }
2466
2467         p = xdr_inline_decode(xdr, sizeof(*p));
2468         if (!p)
2469                 goto out_unparsable;
2470         switch (*p++) {
2471         case rpc_autherr_rejectedcred:
2472         case rpc_autherr_rejectedverf:
2473         case rpcsec_gsserr_credproblem:
2474         case rpcsec_gsserr_ctxproblem:
2475                 if (!task->tk_cred_retry)
2476                         break;
2477                 task->tk_cred_retry--;
2478                 trace_rpc__stale_creds(task);
2479                 rpcauth_invalcred(task);
2480                 /* Ensure we obtain a new XID! */
2481                 xprt_release(task);
2482                 task->tk_action = call_reserve;
2483                 return -EAGAIN;
2484         case rpc_autherr_badcred:
2485         case rpc_autherr_badverf:
2486                 /* possibly garbled cred/verf? */
2487                 if (!task->tk_garb_retry)
2488                         break;
2489                 task->tk_garb_retry--;
2490                 trace_rpc__bad_creds(task);
2491                 task->tk_action = call_encode;
2492                 return -EAGAIN;
2493         case rpc_autherr_tooweak:
2494                 trace_rpc__auth_tooweak(task);
2495                 pr_warn("RPC: server %s requires stronger authentication.\n",
2496                         task->tk_xprt->servername);
2497                 break;
2498         default:
2499                 trace_rpc__unparsable(task);
2500                 error = -EIO;
2501         }
2502         goto out_err;
2503 }
2504
2505 static void rpcproc_encode_null(struct rpc_rqst *rqstp, struct xdr_stream *xdr,
2506                 const void *obj)
2507 {
2508 }
2509
2510 static int rpcproc_decode_null(struct rpc_rqst *rqstp, struct xdr_stream *xdr,
2511                 void *obj)
2512 {
2513         return 0;
2514 }
2515
2516 static const struct rpc_procinfo rpcproc_null = {
2517         .p_encode = rpcproc_encode_null,
2518         .p_decode = rpcproc_decode_null,
2519 };
2520
2521 static int rpc_ping(struct rpc_clnt *clnt)
2522 {
2523         struct rpc_message msg = {
2524                 .rpc_proc = &rpcproc_null,
2525         };
2526         int err;
2527         err = rpc_call_sync(clnt, &msg, RPC_TASK_SOFT | RPC_TASK_SOFTCONN |
2528                             RPC_TASK_NULLCREDS);
2529         return err;
2530 }
2531
2532 static
2533 struct rpc_task *rpc_call_null_helper(struct rpc_clnt *clnt,
2534                 struct rpc_xprt *xprt, struct rpc_cred *cred, int flags,
2535                 const struct rpc_call_ops *ops, void *data)
2536 {
2537         struct rpc_message msg = {
2538                 .rpc_proc = &rpcproc_null,
2539         };
2540         struct rpc_task_setup task_setup_data = {
2541                 .rpc_client = clnt,
2542                 .rpc_xprt = xprt,
2543                 .rpc_message = &msg,
2544                 .rpc_op_cred = cred,
2545                 .callback_ops = (ops != NULL) ? ops : &rpc_default_ops,
2546                 .callback_data = data,
2547                 .flags = flags | RPC_TASK_NULLCREDS,
2548         };
2549
2550         return rpc_run_task(&task_setup_data);
2551 }
2552
2553 struct rpc_task *rpc_call_null(struct rpc_clnt *clnt, struct rpc_cred *cred, int flags)
2554 {
2555         return rpc_call_null_helper(clnt, NULL, cred, flags, NULL, NULL);
2556 }
2557 EXPORT_SYMBOL_GPL(rpc_call_null);
2558
2559 struct rpc_cb_add_xprt_calldata {
2560         struct rpc_xprt_switch *xps;
2561         struct rpc_xprt *xprt;
2562 };
2563
2564 static void rpc_cb_add_xprt_done(struct rpc_task *task, void *calldata)
2565 {
2566         struct rpc_cb_add_xprt_calldata *data = calldata;
2567
2568         if (task->tk_status == 0)
2569                 rpc_xprt_switch_add_xprt(data->xps, data->xprt);
2570 }
2571
2572 static void rpc_cb_add_xprt_release(void *calldata)
2573 {
2574         struct rpc_cb_add_xprt_calldata *data = calldata;
2575
2576         xprt_put(data->xprt);
2577         xprt_switch_put(data->xps);
2578         kfree(data);
2579 }
2580
2581 static const struct rpc_call_ops rpc_cb_add_xprt_call_ops = {
2582         .rpc_call_done = rpc_cb_add_xprt_done,
2583         .rpc_release = rpc_cb_add_xprt_release,
2584 };
2585
2586 /**
2587  * rpc_clnt_test_and_add_xprt - Test and add a new transport to a rpc_clnt
2588  * @clnt: pointer to struct rpc_clnt
2589  * @xps: pointer to struct rpc_xprt_switch,
2590  * @xprt: pointer struct rpc_xprt
2591  * @dummy: unused
2592  */
2593 int rpc_clnt_test_and_add_xprt(struct rpc_clnt *clnt,
2594                 struct rpc_xprt_switch *xps, struct rpc_xprt *xprt,
2595                 void *dummy)
2596 {
2597         struct rpc_cb_add_xprt_calldata *data;
2598         struct rpc_task *task;
2599
2600         data = kmalloc(sizeof(*data), GFP_NOFS);
2601         if (!data)
2602                 return -ENOMEM;
2603         data->xps = xprt_switch_get(xps);
2604         data->xprt = xprt_get(xprt);
2605
2606         task = rpc_call_null_helper(clnt, xprt, NULL,
2607                         RPC_TASK_SOFT|RPC_TASK_SOFTCONN|RPC_TASK_ASYNC|RPC_TASK_NULLCREDS,
2608                         &rpc_cb_add_xprt_call_ops, data);
2609         if (IS_ERR(task))
2610                 return PTR_ERR(task);
2611         rpc_put_task(task);
2612         return 1;
2613 }
2614 EXPORT_SYMBOL_GPL(rpc_clnt_test_and_add_xprt);
2615
2616 /**
2617  * rpc_clnt_setup_test_and_add_xprt()
2618  *
2619  * This is an rpc_clnt_add_xprt setup() function which returns 1 so:
2620  *   1) caller of the test function must dereference the rpc_xprt_switch
2621  *   and the rpc_xprt.
2622  *   2) test function must call rpc_xprt_switch_add_xprt, usually in
2623  *   the rpc_call_done routine.
2624  *
2625  * Upon success (return of 1), the test function adds the new
2626  * transport to the rpc_clnt xprt switch
2627  *
2628  * @clnt: struct rpc_clnt to get the new transport
2629  * @xps:  the rpc_xprt_switch to hold the new transport
2630  * @xprt: the rpc_xprt to test
2631  * @data: a struct rpc_add_xprt_test pointer that holds the test function
2632  *        and test function call data
2633  */
2634 int rpc_clnt_setup_test_and_add_xprt(struct rpc_clnt *clnt,
2635                                      struct rpc_xprt_switch *xps,
2636                                      struct rpc_xprt *xprt,
2637                                      void *data)
2638 {
2639         struct rpc_task *task;
2640         struct rpc_add_xprt_test *xtest = (struct rpc_add_xprt_test *)data;
2641         int status = -EADDRINUSE;
2642
2643         xprt = xprt_get(xprt);
2644         xprt_switch_get(xps);
2645
2646         if (rpc_xprt_switch_has_addr(xps, (struct sockaddr *)&xprt->addr))
2647                 goto out_err;
2648
2649         /* Test the connection */
2650         task = rpc_call_null_helper(clnt, xprt, NULL,
2651                                     RPC_TASK_SOFT | RPC_TASK_SOFTCONN | RPC_TASK_NULLCREDS,
2652                                     NULL, NULL);
2653         if (IS_ERR(task)) {
2654                 status = PTR_ERR(task);
2655                 goto out_err;
2656         }
2657         status = task->tk_status;
2658         rpc_put_task(task);
2659
2660         if (status < 0)
2661                 goto out_err;
2662
2663         /* rpc_xprt_switch and rpc_xprt are deferrenced by add_xprt_test() */
2664         xtest->add_xprt_test(clnt, xprt, xtest->data);
2665
2666         xprt_put(xprt);
2667         xprt_switch_put(xps);
2668
2669         /* so that rpc_clnt_add_xprt does not call rpc_xprt_switch_add_xprt */
2670         return 1;
2671 out_err:
2672         xprt_put(xprt);
2673         xprt_switch_put(xps);
2674         pr_info("RPC:   rpc_clnt_test_xprt failed: %d addr %s not added\n",
2675                 status, xprt->address_strings[RPC_DISPLAY_ADDR]);
2676         return status;
2677 }
2678 EXPORT_SYMBOL_GPL(rpc_clnt_setup_test_and_add_xprt);
2679
2680 /**
2681  * rpc_clnt_add_xprt - Add a new transport to a rpc_clnt
2682  * @clnt: pointer to struct rpc_clnt
2683  * @xprtargs: pointer to struct xprt_create
2684  * @setup: callback to test and/or set up the connection
2685  * @data: pointer to setup function data
2686  *
2687  * Creates a new transport using the parameters set in args and
2688  * adds it to clnt.
2689  * If ping is set, then test that connectivity succeeds before
2690  * adding the new transport.
2691  *
2692  */
2693 int rpc_clnt_add_xprt(struct rpc_clnt *clnt,
2694                 struct xprt_create *xprtargs,
2695                 int (*setup)(struct rpc_clnt *,
2696                         struct rpc_xprt_switch *,
2697                         struct rpc_xprt *,
2698                         void *),
2699                 void *data)
2700 {
2701         struct rpc_xprt_switch *xps;
2702         struct rpc_xprt *xprt;
2703         unsigned long connect_timeout;
2704         unsigned long reconnect_timeout;
2705         unsigned char resvport;
2706         int ret = 0;
2707
2708         rcu_read_lock();
2709         xps = xprt_switch_get(rcu_dereference(clnt->cl_xpi.xpi_xpswitch));
2710         xprt = xprt_iter_xprt(&clnt->cl_xpi);
2711         if (xps == NULL || xprt == NULL) {
2712                 rcu_read_unlock();
2713                 return -EAGAIN;
2714         }
2715         resvport = xprt->resvport;
2716         connect_timeout = xprt->connect_timeout;
2717         reconnect_timeout = xprt->max_reconnect_timeout;
2718         rcu_read_unlock();
2719
2720         xprt = xprt_create_transport(xprtargs);
2721         if (IS_ERR(xprt)) {
2722                 ret = PTR_ERR(xprt);
2723                 goto out_put_switch;
2724         }
2725         xprt->resvport = resvport;
2726         if (xprt->ops->set_connect_timeout != NULL)
2727                 xprt->ops->set_connect_timeout(xprt,
2728                                 connect_timeout,
2729                                 reconnect_timeout);
2730
2731         rpc_xprt_switch_set_roundrobin(xps);
2732         if (setup) {
2733                 ret = setup(clnt, xps, xprt, data);
2734                 if (ret != 0)
2735                         goto out_put_xprt;
2736         }
2737         rpc_xprt_switch_add_xprt(xps, xprt);
2738 out_put_xprt:
2739         xprt_put(xprt);
2740 out_put_switch:
2741         xprt_switch_put(xps);
2742         return ret;
2743 }
2744 EXPORT_SYMBOL_GPL(rpc_clnt_add_xprt);
2745
2746 struct connect_timeout_data {
2747         unsigned long connect_timeout;
2748         unsigned long reconnect_timeout;
2749 };
2750
2751 static int
2752 rpc_xprt_set_connect_timeout(struct rpc_clnt *clnt,
2753                 struct rpc_xprt *xprt,
2754                 void *data)
2755 {
2756         struct connect_timeout_data *timeo = data;
2757
2758         if (xprt->ops->set_connect_timeout)
2759                 xprt->ops->set_connect_timeout(xprt,
2760                                 timeo->connect_timeout,
2761                                 timeo->reconnect_timeout);
2762         return 0;
2763 }
2764
2765 void
2766 rpc_set_connect_timeout(struct rpc_clnt *clnt,
2767                 unsigned long connect_timeout,
2768                 unsigned long reconnect_timeout)
2769 {
2770         struct connect_timeout_data timeout = {
2771                 .connect_timeout = connect_timeout,
2772                 .reconnect_timeout = reconnect_timeout,
2773         };
2774         rpc_clnt_iterate_for_each_xprt(clnt,
2775                         rpc_xprt_set_connect_timeout,
2776                         &timeout);
2777 }
2778 EXPORT_SYMBOL_GPL(rpc_set_connect_timeout);
2779
2780 void rpc_clnt_xprt_switch_put(struct rpc_clnt *clnt)
2781 {
2782         rcu_read_lock();
2783         xprt_switch_put(rcu_dereference(clnt->cl_xpi.xpi_xpswitch));
2784         rcu_read_unlock();
2785 }
2786 EXPORT_SYMBOL_GPL(rpc_clnt_xprt_switch_put);
2787
2788 void rpc_clnt_xprt_switch_add_xprt(struct rpc_clnt *clnt, struct rpc_xprt *xprt)
2789 {
2790         rcu_read_lock();
2791         rpc_xprt_switch_add_xprt(rcu_dereference(clnt->cl_xpi.xpi_xpswitch),
2792                                  xprt);
2793         rcu_read_unlock();
2794 }
2795 EXPORT_SYMBOL_GPL(rpc_clnt_xprt_switch_add_xprt);
2796
2797 bool rpc_clnt_xprt_switch_has_addr(struct rpc_clnt *clnt,
2798                                    const struct sockaddr *sap)
2799 {
2800         struct rpc_xprt_switch *xps;
2801         bool ret;
2802
2803         rcu_read_lock();
2804         xps = rcu_dereference(clnt->cl_xpi.xpi_xpswitch);
2805         ret = rpc_xprt_switch_has_addr(xps, sap);
2806         rcu_read_unlock();
2807         return ret;
2808 }
2809 EXPORT_SYMBOL_GPL(rpc_clnt_xprt_switch_has_addr);
2810
2811 #if IS_ENABLED(CONFIG_SUNRPC_DEBUG)
2812 static void rpc_show_header(void)
2813 {
2814         printk(KERN_INFO "-pid- flgs status -client- --rqstp- "
2815                 "-timeout ---ops--\n");
2816 }
2817
2818 static void rpc_show_task(const struct rpc_clnt *clnt,
2819                           const struct rpc_task *task)
2820 {
2821         const char *rpc_waitq = "none";
2822
2823         if (RPC_IS_QUEUED(task))
2824                 rpc_waitq = rpc_qname(task->tk_waitqueue);
2825
2826         printk(KERN_INFO "%5u %04x %6d %8p %8p %8ld %8p %sv%u %s a:%ps q:%s\n",
2827                 task->tk_pid, task->tk_flags, task->tk_status,
2828                 clnt, task->tk_rqstp, task->tk_timeout, task->tk_ops,
2829                 clnt->cl_program->name, clnt->cl_vers, rpc_proc_name(task),
2830                 task->tk_action, rpc_waitq);
2831 }
2832
2833 void rpc_show_tasks(struct net *net)
2834 {
2835         struct rpc_clnt *clnt;
2836         struct rpc_task *task;
2837         int header = 0;
2838         struct sunrpc_net *sn = net_generic(net, sunrpc_net_id);
2839
2840         spin_lock(&sn->rpc_client_lock);
2841         list_for_each_entry(clnt, &sn->all_clients, cl_clients) {
2842                 spin_lock(&clnt->cl_lock);
2843                 list_for_each_entry(task, &clnt->cl_tasks, tk_task) {
2844                         if (!header) {
2845                                 rpc_show_header();
2846                                 header++;
2847                         }
2848                         rpc_show_task(clnt, task);
2849                 }
2850                 spin_unlock(&clnt->cl_lock);
2851         }
2852         spin_unlock(&sn->rpc_client_lock);
2853 }
2854 #endif
2855
2856 #if IS_ENABLED(CONFIG_SUNRPC_SWAP)
2857 static int
2858 rpc_clnt_swap_activate_callback(struct rpc_clnt *clnt,
2859                 struct rpc_xprt *xprt,
2860                 void *dummy)
2861 {
2862         return xprt_enable_swap(xprt);
2863 }
2864
2865 int
2866 rpc_clnt_swap_activate(struct rpc_clnt *clnt)
2867 {
2868         if (atomic_inc_return(&clnt->cl_swapper) == 1)
2869                 return rpc_clnt_iterate_for_each_xprt(clnt,
2870                                 rpc_clnt_swap_activate_callback, NULL);
2871         return 0;
2872 }
2873 EXPORT_SYMBOL_GPL(rpc_clnt_swap_activate);
2874
2875 static int
2876 rpc_clnt_swap_deactivate_callback(struct rpc_clnt *clnt,
2877                 struct rpc_xprt *xprt,
2878                 void *dummy)
2879 {
2880         xprt_disable_swap(xprt);
2881         return 0;
2882 }
2883
2884 void
2885 rpc_clnt_swap_deactivate(struct rpc_clnt *clnt)
2886 {
2887         if (atomic_dec_if_positive(&clnt->cl_swapper) == 0)
2888                 rpc_clnt_iterate_for_each_xprt(clnt,
2889                                 rpc_clnt_swap_deactivate_callback, NULL);
2890 }
2891 EXPORT_SYMBOL_GPL(rpc_clnt_swap_deactivate);
2892 #endif /* CONFIG_SUNRPC_SWAP */