xprt.c revision 15a4520621824a3c2eb2de2d1f3984bc1663d3c8
1/* 2 * linux/net/sunrpc/xprt.c 3 * 4 * This is a generic RPC call interface supporting congestion avoidance, 5 * and asynchronous calls. 6 * 7 * The interface works like this: 8 * 9 * - When a process places a call, it allocates a request slot if 10 * one is available. Otherwise, it sleeps on the backlog queue 11 * (xprt_reserve). 12 * - Next, the caller puts together the RPC message, stuffs it into 13 * the request struct, and calls xprt_transmit(). 14 * - xprt_transmit sends the message and installs the caller on the 15 * transport's wait list. At the same time, if a reply is expected, 16 * it installs a timer that is run after the packet's timeout has 17 * expired. 18 * - When a packet arrives, the data_ready handler walks the list of 19 * pending requests for that transport. If a matching XID is found, the 20 * caller is woken up, and the timer removed. 21 * - When no reply arrives within the timeout interval, the timer is 22 * fired by the kernel and runs xprt_timer(). It either adjusts the 23 * timeout values (minor timeout) or wakes up the caller with a status 24 * of -ETIMEDOUT. 25 * - When the caller receives a notification from RPC that a reply arrived, 26 * it should release the RPC slot, and process the reply. 27 * If the call timed out, it may choose to retry the operation by 28 * adjusting the initial timeout value, and simply calling rpc_call 29 * again. 30 * 31 * Support for async RPC is done through a set of RPC-specific scheduling 32 * primitives that `transparently' work for processes as well as async 33 * tasks that rely on callbacks. 34 * 35 * Copyright (C) 1995-1997, Olaf Kirch <okir@monad.swb.de> 36 * 37 * Transport switch API copyright (C) 2005, Chuck Lever <cel@netapp.com> 38 */ 39 40#include <linux/module.h> 41 42#include <linux/types.h> 43#include <linux/interrupt.h> 44#include <linux/workqueue.h> 45#include <linux/net.h> 46#include <linux/ktime.h> 47 48#include <linux/sunrpc/clnt.h> 49#include <linux/sunrpc/metrics.h> 50#include <linux/sunrpc/bc_xprt.h> 51 52#include "sunrpc.h" 53 54/* 55 * Local variables 56 */ 57 58#ifdef RPC_DEBUG 59# define RPCDBG_FACILITY RPCDBG_XPRT 60#endif 61 62/* 63 * Local functions 64 */ 65static void xprt_init(struct rpc_xprt *xprt, struct net *net); 66static void xprt_request_init(struct rpc_task *, struct rpc_xprt *); 67static void xprt_connect_status(struct rpc_task *task); 68static int __xprt_get_cong(struct rpc_xprt *, struct rpc_task *); 69 70static DEFINE_SPINLOCK(xprt_list_lock); 71static LIST_HEAD(xprt_list); 72 73/* 74 * The transport code maintains an estimate on the maximum number of out- 75 * standing RPC requests, using a smoothed version of the congestion 76 * avoidance implemented in 44BSD. This is basically the Van Jacobson 77 * congestion algorithm: If a retransmit occurs, the congestion window is 78 * halved; otherwise, it is incremented by 1/cwnd when 79 * 80 * - a reply is received and 81 * - a full number of requests are outstanding and 82 * - the congestion window hasn't been updated recently. 83 */ 84#define RPC_CWNDSHIFT (8U) 85#define RPC_CWNDSCALE (1U << RPC_CWNDSHIFT) 86#define RPC_INITCWND RPC_CWNDSCALE 87#define RPC_MAXCWND(xprt) ((xprt)->max_reqs << RPC_CWNDSHIFT) 88 89#define RPCXPRT_CONGESTED(xprt) ((xprt)->cong >= (xprt)->cwnd) 90 91/** 92 * xprt_register_transport - register a transport implementation 93 * @transport: transport to register 94 * 95 * If a transport implementation is loaded as a kernel module, it can 96 * call this interface to make itself known to the RPC client. 97 * 98 * Returns: 99 * 0: transport successfully registered 100 * -EEXIST: transport already registered 101 * -EINVAL: transport module being unloaded 102 */ 103int xprt_register_transport(struct xprt_class *transport) 104{ 105 struct xprt_class *t; 106 int result; 107 108 result = -EEXIST; 109 spin_lock(&xprt_list_lock); 110 list_for_each_entry(t, &xprt_list, list) { 111 /* don't register the same transport class twice */ 112 if (t->ident == transport->ident) 113 goto out; 114 } 115 116 list_add_tail(&transport->list, &xprt_list); 117 printk(KERN_INFO "RPC: Registered %s transport module.\n", 118 transport->name); 119 result = 0; 120 121out: 122 spin_unlock(&xprt_list_lock); 123 return result; 124} 125EXPORT_SYMBOL_GPL(xprt_register_transport); 126 127/** 128 * xprt_unregister_transport - unregister a transport implementation 129 * @transport: transport to unregister 130 * 131 * Returns: 132 * 0: transport successfully unregistered 133 * -ENOENT: transport never registered 134 */ 135int xprt_unregister_transport(struct xprt_class *transport) 136{ 137 struct xprt_class *t; 138 int result; 139 140 result = 0; 141 spin_lock(&xprt_list_lock); 142 list_for_each_entry(t, &xprt_list, list) { 143 if (t == transport) { 144 printk(KERN_INFO 145 "RPC: Unregistered %s transport module.\n", 146 transport->name); 147 list_del_init(&transport->list); 148 goto out; 149 } 150 } 151 result = -ENOENT; 152 153out: 154 spin_unlock(&xprt_list_lock); 155 return result; 156} 157EXPORT_SYMBOL_GPL(xprt_unregister_transport); 158 159/** 160 * xprt_load_transport - load a transport implementation 161 * @transport_name: transport to load 162 * 163 * Returns: 164 * 0: transport successfully loaded 165 * -ENOENT: transport module not available 166 */ 167int xprt_load_transport(const char *transport_name) 168{ 169 struct xprt_class *t; 170 int result; 171 172 result = 0; 173 spin_lock(&xprt_list_lock); 174 list_for_each_entry(t, &xprt_list, list) { 175 if (strcmp(t->name, transport_name) == 0) { 176 spin_unlock(&xprt_list_lock); 177 goto out; 178 } 179 } 180 spin_unlock(&xprt_list_lock); 181 result = request_module("xprt%s", transport_name); 182out: 183 return result; 184} 185EXPORT_SYMBOL_GPL(xprt_load_transport); 186 187/** 188 * xprt_reserve_xprt - serialize write access to transports 189 * @task: task that is requesting access to the transport 190 * @xprt: pointer to the target transport 191 * 192 * This prevents mixing the payload of separate requests, and prevents 193 * transport connects from colliding with writes. No congestion control 194 * is provided. 195 */ 196int xprt_reserve_xprt(struct rpc_xprt *xprt, struct rpc_task *task) 197{ 198 struct rpc_rqst *req = task->tk_rqstp; 199 int priority; 200 201 if (test_and_set_bit(XPRT_LOCKED, &xprt->state)) { 202 if (task == xprt->snd_task) 203 return 1; 204 goto out_sleep; 205 } 206 xprt->snd_task = task; 207 if (req != NULL) { 208 req->rq_bytes_sent = 0; 209 req->rq_ntrans++; 210 } 211 212 return 1; 213 214out_sleep: 215 dprintk("RPC: %5u failed to lock transport %p\n", 216 task->tk_pid, xprt); 217 task->tk_timeout = 0; 218 task->tk_status = -EAGAIN; 219 if (req == NULL) 220 priority = RPC_PRIORITY_LOW; 221 else if (!req->rq_ntrans) 222 priority = RPC_PRIORITY_NORMAL; 223 else 224 priority = RPC_PRIORITY_HIGH; 225 rpc_sleep_on_priority(&xprt->sending, task, NULL, priority); 226 return 0; 227} 228EXPORT_SYMBOL_GPL(xprt_reserve_xprt); 229 230static void xprt_clear_locked(struct rpc_xprt *xprt) 231{ 232 xprt->snd_task = NULL; 233 if (!test_bit(XPRT_CLOSE_WAIT, &xprt->state) || xprt->shutdown) { 234 smp_mb__before_clear_bit(); 235 clear_bit(XPRT_LOCKED, &xprt->state); 236 smp_mb__after_clear_bit(); 237 } else 238 queue_work(rpciod_workqueue, &xprt->task_cleanup); 239} 240 241/* 242 * xprt_reserve_xprt_cong - serialize write access to transports 243 * @task: task that is requesting access to the transport 244 * 245 * Same as xprt_reserve_xprt, but Van Jacobson congestion control is 246 * integrated into the decision of whether a request is allowed to be 247 * woken up and given access to the transport. 248 */ 249int xprt_reserve_xprt_cong(struct rpc_xprt *xprt, struct rpc_task *task) 250{ 251 struct rpc_rqst *req = task->tk_rqstp; 252 int priority; 253 254 if (test_and_set_bit(XPRT_LOCKED, &xprt->state)) { 255 if (task == xprt->snd_task) 256 return 1; 257 goto out_sleep; 258 } 259 if (req == NULL) { 260 xprt->snd_task = task; 261 return 1; 262 } 263 if (__xprt_get_cong(xprt, task)) { 264 xprt->snd_task = task; 265 req->rq_bytes_sent = 0; 266 req->rq_ntrans++; 267 return 1; 268 } 269 xprt_clear_locked(xprt); 270out_sleep: 271 dprintk("RPC: %5u failed to lock transport %p\n", task->tk_pid, xprt); 272 task->tk_timeout = 0; 273 task->tk_status = -EAGAIN; 274 if (req == NULL) 275 priority = RPC_PRIORITY_LOW; 276 else if (!req->rq_ntrans) 277 priority = RPC_PRIORITY_NORMAL; 278 else 279 priority = RPC_PRIORITY_HIGH; 280 rpc_sleep_on_priority(&xprt->sending, task, NULL, priority); 281 return 0; 282} 283EXPORT_SYMBOL_GPL(xprt_reserve_xprt_cong); 284 285static inline int xprt_lock_write(struct rpc_xprt *xprt, struct rpc_task *task) 286{ 287 int retval; 288 289 spin_lock_bh(&xprt->transport_lock); 290 retval = xprt->ops->reserve_xprt(xprt, task); 291 spin_unlock_bh(&xprt->transport_lock); 292 return retval; 293} 294 295static bool __xprt_lock_write_func(struct rpc_task *task, void *data) 296{ 297 struct rpc_xprt *xprt = data; 298 struct rpc_rqst *req; 299 300 req = task->tk_rqstp; 301 xprt->snd_task = task; 302 if (req) { 303 req->rq_bytes_sent = 0; 304 req->rq_ntrans++; 305 } 306 return true; 307} 308 309static void __xprt_lock_write_next(struct rpc_xprt *xprt) 310{ 311 if (test_and_set_bit(XPRT_LOCKED, &xprt->state)) 312 return; 313 314 if (rpc_wake_up_first(&xprt->sending, __xprt_lock_write_func, xprt)) 315 return; 316 xprt_clear_locked(xprt); 317} 318 319static bool __xprt_lock_write_cong_func(struct rpc_task *task, void *data) 320{ 321 struct rpc_xprt *xprt = data; 322 struct rpc_rqst *req; 323 324 req = task->tk_rqstp; 325 if (req == NULL) { 326 xprt->snd_task = task; 327 return true; 328 } 329 if (__xprt_get_cong(xprt, task)) { 330 xprt->snd_task = task; 331 req->rq_bytes_sent = 0; 332 req->rq_ntrans++; 333 return true; 334 } 335 return false; 336} 337 338static void __xprt_lock_write_next_cong(struct rpc_xprt *xprt) 339{ 340 if (test_and_set_bit(XPRT_LOCKED, &xprt->state)) 341 return; 342 if (RPCXPRT_CONGESTED(xprt)) 343 goto out_unlock; 344 if (rpc_wake_up_first(&xprt->sending, __xprt_lock_write_cong_func, xprt)) 345 return; 346out_unlock: 347 xprt_clear_locked(xprt); 348} 349 350/** 351 * xprt_release_xprt - allow other requests to use a transport 352 * @xprt: transport with other tasks potentially waiting 353 * @task: task that is releasing access to the transport 354 * 355 * Note that "task" can be NULL. No congestion control is provided. 356 */ 357void xprt_release_xprt(struct rpc_xprt *xprt, struct rpc_task *task) 358{ 359 if (xprt->snd_task == task) { 360 xprt_clear_locked(xprt); 361 __xprt_lock_write_next(xprt); 362 } 363} 364EXPORT_SYMBOL_GPL(xprt_release_xprt); 365 366/** 367 * xprt_release_xprt_cong - allow other requests to use a transport 368 * @xprt: transport with other tasks potentially waiting 369 * @task: task that is releasing access to the transport 370 * 371 * Note that "task" can be NULL. Another task is awoken to use the 372 * transport if the transport's congestion window allows it. 373 */ 374void xprt_release_xprt_cong(struct rpc_xprt *xprt, struct rpc_task *task) 375{ 376 if (xprt->snd_task == task) { 377 xprt_clear_locked(xprt); 378 __xprt_lock_write_next_cong(xprt); 379 } 380} 381EXPORT_SYMBOL_GPL(xprt_release_xprt_cong); 382 383static inline void xprt_release_write(struct rpc_xprt *xprt, struct rpc_task *task) 384{ 385 spin_lock_bh(&xprt->transport_lock); 386 xprt->ops->release_xprt(xprt, task); 387 spin_unlock_bh(&xprt->transport_lock); 388} 389 390/* 391 * Van Jacobson congestion avoidance. Check if the congestion window 392 * overflowed. Put the task to sleep if this is the case. 393 */ 394static int 395__xprt_get_cong(struct rpc_xprt *xprt, struct rpc_task *task) 396{ 397 struct rpc_rqst *req = task->tk_rqstp; 398 399 if (req->rq_cong) 400 return 1; 401 dprintk("RPC: %5u xprt_cwnd_limited cong = %lu cwnd = %lu\n", 402 task->tk_pid, xprt->cong, xprt->cwnd); 403 if (RPCXPRT_CONGESTED(xprt)) 404 return 0; 405 req->rq_cong = 1; 406 xprt->cong += RPC_CWNDSCALE; 407 return 1; 408} 409 410/* 411 * Adjust the congestion window, and wake up the next task 412 * that has been sleeping due to congestion 413 */ 414static void 415__xprt_put_cong(struct rpc_xprt *xprt, struct rpc_rqst *req) 416{ 417 if (!req->rq_cong) 418 return; 419 req->rq_cong = 0; 420 xprt->cong -= RPC_CWNDSCALE; 421 __xprt_lock_write_next_cong(xprt); 422} 423 424/** 425 * xprt_release_rqst_cong - housekeeping when request is complete 426 * @task: RPC request that recently completed 427 * 428 * Useful for transports that require congestion control. 429 */ 430void xprt_release_rqst_cong(struct rpc_task *task) 431{ 432 __xprt_put_cong(task->tk_xprt, task->tk_rqstp); 433} 434EXPORT_SYMBOL_GPL(xprt_release_rqst_cong); 435 436/** 437 * xprt_adjust_cwnd - adjust transport congestion window 438 * @task: recently completed RPC request used to adjust window 439 * @result: result code of completed RPC request 440 * 441 * We use a time-smoothed congestion estimator to avoid heavy oscillation. 442 */ 443void xprt_adjust_cwnd(struct rpc_task *task, int result) 444{ 445 struct rpc_rqst *req = task->tk_rqstp; 446 struct rpc_xprt *xprt = task->tk_xprt; 447 unsigned long cwnd = xprt->cwnd; 448 449 if (result >= 0 && cwnd <= xprt->cong) { 450 /* The (cwnd >> 1) term makes sure 451 * the result gets rounded properly. */ 452 cwnd += (RPC_CWNDSCALE * RPC_CWNDSCALE + (cwnd >> 1)) / cwnd; 453 if (cwnd > RPC_MAXCWND(xprt)) 454 cwnd = RPC_MAXCWND(xprt); 455 __xprt_lock_write_next_cong(xprt); 456 } else if (result == -ETIMEDOUT) { 457 cwnd >>= 1; 458 if (cwnd < RPC_CWNDSCALE) 459 cwnd = RPC_CWNDSCALE; 460 } 461 dprintk("RPC: cong %ld, cwnd was %ld, now %ld\n", 462 xprt->cong, xprt->cwnd, cwnd); 463 xprt->cwnd = cwnd; 464 __xprt_put_cong(xprt, req); 465} 466EXPORT_SYMBOL_GPL(xprt_adjust_cwnd); 467 468/** 469 * xprt_wake_pending_tasks - wake all tasks on a transport's pending queue 470 * @xprt: transport with waiting tasks 471 * @status: result code to plant in each task before waking it 472 * 473 */ 474void xprt_wake_pending_tasks(struct rpc_xprt *xprt, int status) 475{ 476 if (status < 0) 477 rpc_wake_up_status(&xprt->pending, status); 478 else 479 rpc_wake_up(&xprt->pending); 480} 481EXPORT_SYMBOL_GPL(xprt_wake_pending_tasks); 482 483/** 484 * xprt_wait_for_buffer_space - wait for transport output buffer to clear 485 * @task: task to be put to sleep 486 * @action: function pointer to be executed after wait 487 */ 488void xprt_wait_for_buffer_space(struct rpc_task *task, rpc_action action) 489{ 490 struct rpc_rqst *req = task->tk_rqstp; 491 struct rpc_xprt *xprt = req->rq_xprt; 492 493 task->tk_timeout = req->rq_timeout; 494 rpc_sleep_on(&xprt->pending, task, action); 495} 496EXPORT_SYMBOL_GPL(xprt_wait_for_buffer_space); 497 498/** 499 * xprt_write_space - wake the task waiting for transport output buffer space 500 * @xprt: transport with waiting tasks 501 * 502 * Can be called in a soft IRQ context, so xprt_write_space never sleeps. 503 */ 504void xprt_write_space(struct rpc_xprt *xprt) 505{ 506 if (unlikely(xprt->shutdown)) 507 return; 508 509 spin_lock_bh(&xprt->transport_lock); 510 if (xprt->snd_task) { 511 dprintk("RPC: write space: waking waiting task on " 512 "xprt %p\n", xprt); 513 rpc_wake_up_queued_task(&xprt->pending, xprt->snd_task); 514 } 515 spin_unlock_bh(&xprt->transport_lock); 516} 517EXPORT_SYMBOL_GPL(xprt_write_space); 518 519/** 520 * xprt_set_retrans_timeout_def - set a request's retransmit timeout 521 * @task: task whose timeout is to be set 522 * 523 * Set a request's retransmit timeout based on the transport's 524 * default timeout parameters. Used by transports that don't adjust 525 * the retransmit timeout based on round-trip time estimation. 526 */ 527void xprt_set_retrans_timeout_def(struct rpc_task *task) 528{ 529 task->tk_timeout = task->tk_rqstp->rq_timeout; 530} 531EXPORT_SYMBOL_GPL(xprt_set_retrans_timeout_def); 532 533/* 534 * xprt_set_retrans_timeout_rtt - set a request's retransmit timeout 535 * @task: task whose timeout is to be set 536 * 537 * Set a request's retransmit timeout using the RTT estimator. 538 */ 539void xprt_set_retrans_timeout_rtt(struct rpc_task *task) 540{ 541 int timer = task->tk_msg.rpc_proc->p_timer; 542 struct rpc_clnt *clnt = task->tk_client; 543 struct rpc_rtt *rtt = clnt->cl_rtt; 544 struct rpc_rqst *req = task->tk_rqstp; 545 unsigned long max_timeout = clnt->cl_timeout->to_maxval; 546 547 task->tk_timeout = rpc_calc_rto(rtt, timer); 548 task->tk_timeout <<= rpc_ntimeo(rtt, timer) + req->rq_retries; 549 if (task->tk_timeout > max_timeout || task->tk_timeout == 0) 550 task->tk_timeout = max_timeout; 551} 552EXPORT_SYMBOL_GPL(xprt_set_retrans_timeout_rtt); 553 554static void xprt_reset_majortimeo(struct rpc_rqst *req) 555{ 556 const struct rpc_timeout *to = req->rq_task->tk_client->cl_timeout; 557 558 req->rq_majortimeo = req->rq_timeout; 559 if (to->to_exponential) 560 req->rq_majortimeo <<= to->to_retries; 561 else 562 req->rq_majortimeo += to->to_increment * to->to_retries; 563 if (req->rq_majortimeo > to->to_maxval || req->rq_majortimeo == 0) 564 req->rq_majortimeo = to->to_maxval; 565 req->rq_majortimeo += jiffies; 566} 567 568/** 569 * xprt_adjust_timeout - adjust timeout values for next retransmit 570 * @req: RPC request containing parameters to use for the adjustment 571 * 572 */ 573int xprt_adjust_timeout(struct rpc_rqst *req) 574{ 575 struct rpc_xprt *xprt = req->rq_xprt; 576 const struct rpc_timeout *to = req->rq_task->tk_client->cl_timeout; 577 int status = 0; 578 579 if (time_before(jiffies, req->rq_majortimeo)) { 580 if (to->to_exponential) 581 req->rq_timeout <<= 1; 582 else 583 req->rq_timeout += to->to_increment; 584 if (to->to_maxval && req->rq_timeout >= to->to_maxval) 585 req->rq_timeout = to->to_maxval; 586 req->rq_retries++; 587 } else { 588 req->rq_timeout = to->to_initval; 589 req->rq_retries = 0; 590 xprt_reset_majortimeo(req); 591 /* Reset the RTT counters == "slow start" */ 592 spin_lock_bh(&xprt->transport_lock); 593 rpc_init_rtt(req->rq_task->tk_client->cl_rtt, to->to_initval); 594 spin_unlock_bh(&xprt->transport_lock); 595 status = -ETIMEDOUT; 596 } 597 598 if (req->rq_timeout == 0) { 599 printk(KERN_WARNING "xprt_adjust_timeout: rq_timeout = 0!\n"); 600 req->rq_timeout = 5 * HZ; 601 } 602 return status; 603} 604 605static void xprt_autoclose(struct work_struct *work) 606{ 607 struct rpc_xprt *xprt = 608 container_of(work, struct rpc_xprt, task_cleanup); 609 610 xprt->ops->close(xprt); 611 clear_bit(XPRT_CLOSE_WAIT, &xprt->state); 612 xprt_release_write(xprt, NULL); 613} 614 615/** 616 * xprt_disconnect_done - mark a transport as disconnected 617 * @xprt: transport to flag for disconnect 618 * 619 */ 620void xprt_disconnect_done(struct rpc_xprt *xprt) 621{ 622 dprintk("RPC: disconnected transport %p\n", xprt); 623 spin_lock_bh(&xprt->transport_lock); 624 xprt_clear_connected(xprt); 625 xprt_wake_pending_tasks(xprt, -EAGAIN); 626 spin_unlock_bh(&xprt->transport_lock); 627} 628EXPORT_SYMBOL_GPL(xprt_disconnect_done); 629 630/** 631 * xprt_force_disconnect - force a transport to disconnect 632 * @xprt: transport to disconnect 633 * 634 */ 635void xprt_force_disconnect(struct rpc_xprt *xprt) 636{ 637 /* Don't race with the test_bit() in xprt_clear_locked() */ 638 spin_lock_bh(&xprt->transport_lock); 639 set_bit(XPRT_CLOSE_WAIT, &xprt->state); 640 /* Try to schedule an autoclose RPC call */ 641 if (test_and_set_bit(XPRT_LOCKED, &xprt->state) == 0) 642 queue_work(rpciod_workqueue, &xprt->task_cleanup); 643 xprt_wake_pending_tasks(xprt, -EAGAIN); 644 spin_unlock_bh(&xprt->transport_lock); 645} 646 647/** 648 * xprt_conditional_disconnect - force a transport to disconnect 649 * @xprt: transport to disconnect 650 * @cookie: 'connection cookie' 651 * 652 * This attempts to break the connection if and only if 'cookie' matches 653 * the current transport 'connection cookie'. It ensures that we don't 654 * try to break the connection more than once when we need to retransmit 655 * a batch of RPC requests. 656 * 657 */ 658void xprt_conditional_disconnect(struct rpc_xprt *xprt, unsigned int cookie) 659{ 660 /* Don't race with the test_bit() in xprt_clear_locked() */ 661 spin_lock_bh(&xprt->transport_lock); 662 if (cookie != xprt->connect_cookie) 663 goto out; 664 if (test_bit(XPRT_CLOSING, &xprt->state) || !xprt_connected(xprt)) 665 goto out; 666 set_bit(XPRT_CLOSE_WAIT, &xprt->state); 667 /* Try to schedule an autoclose RPC call */ 668 if (test_and_set_bit(XPRT_LOCKED, &xprt->state) == 0) 669 queue_work(rpciod_workqueue, &xprt->task_cleanup); 670 xprt_wake_pending_tasks(xprt, -EAGAIN); 671out: 672 spin_unlock_bh(&xprt->transport_lock); 673} 674 675static void 676xprt_init_autodisconnect(unsigned long data) 677{ 678 struct rpc_xprt *xprt = (struct rpc_xprt *)data; 679 680 spin_lock(&xprt->transport_lock); 681 if (!list_empty(&xprt->recv) || xprt->shutdown) 682 goto out_abort; 683 if (test_and_set_bit(XPRT_LOCKED, &xprt->state)) 684 goto out_abort; 685 spin_unlock(&xprt->transport_lock); 686 set_bit(XPRT_CONNECTION_CLOSE, &xprt->state); 687 queue_work(rpciod_workqueue, &xprt->task_cleanup); 688 return; 689out_abort: 690 spin_unlock(&xprt->transport_lock); 691} 692 693/** 694 * xprt_connect - schedule a transport connect operation 695 * @task: RPC task that is requesting the connect 696 * 697 */ 698void xprt_connect(struct rpc_task *task) 699{ 700 struct rpc_xprt *xprt = task->tk_xprt; 701 702 dprintk("RPC: %5u xprt_connect xprt %p %s connected\n", task->tk_pid, 703 xprt, (xprt_connected(xprt) ? "is" : "is not")); 704 705 if (!xprt_bound(xprt)) { 706 task->tk_status = -EAGAIN; 707 return; 708 } 709 if (!xprt_lock_write(xprt, task)) 710 return; 711 712 if (test_and_clear_bit(XPRT_CLOSE_WAIT, &xprt->state)) 713 xprt->ops->close(xprt); 714 715 if (xprt_connected(xprt)) 716 xprt_release_write(xprt, task); 717 else { 718 task->tk_rqstp->rq_bytes_sent = 0; 719 task->tk_timeout = task->tk_rqstp->rq_timeout; 720 rpc_sleep_on(&xprt->pending, task, xprt_connect_status); 721 722 if (test_bit(XPRT_CLOSING, &xprt->state)) 723 return; 724 if (xprt_test_and_set_connecting(xprt)) 725 return; 726 xprt->stat.connect_start = jiffies; 727 xprt->ops->connect(task); 728 } 729} 730 731static void xprt_connect_status(struct rpc_task *task) 732{ 733 struct rpc_xprt *xprt = task->tk_xprt; 734 735 if (task->tk_status == 0) { 736 xprt->stat.connect_count++; 737 xprt->stat.connect_time += (long)jiffies - xprt->stat.connect_start; 738 dprintk("RPC: %5u xprt_connect_status: connection established\n", 739 task->tk_pid); 740 return; 741 } 742 743 switch (task->tk_status) { 744 case -EAGAIN: 745 dprintk("RPC: %5u xprt_connect_status: retrying\n", task->tk_pid); 746 break; 747 case -ETIMEDOUT: 748 dprintk("RPC: %5u xprt_connect_status: connect attempt timed " 749 "out\n", task->tk_pid); 750 break; 751 default: 752 dprintk("RPC: %5u xprt_connect_status: error %d connecting to " 753 "server %s\n", task->tk_pid, -task->tk_status, 754 task->tk_client->cl_server); 755 xprt_release_write(xprt, task); 756 task->tk_status = -EIO; 757 } 758} 759 760/** 761 * xprt_lookup_rqst - find an RPC request corresponding to an XID 762 * @xprt: transport on which the original request was transmitted 763 * @xid: RPC XID of incoming reply 764 * 765 */ 766struct rpc_rqst *xprt_lookup_rqst(struct rpc_xprt *xprt, __be32 xid) 767{ 768 struct rpc_rqst *entry; 769 770 list_for_each_entry(entry, &xprt->recv, rq_list) 771 if (entry->rq_xid == xid) 772 return entry; 773 774 dprintk("RPC: xprt_lookup_rqst did not find xid %08x\n", 775 ntohl(xid)); 776 xprt->stat.bad_xids++; 777 return NULL; 778} 779EXPORT_SYMBOL_GPL(xprt_lookup_rqst); 780 781static void xprt_update_rtt(struct rpc_task *task) 782{ 783 struct rpc_rqst *req = task->tk_rqstp; 784 struct rpc_rtt *rtt = task->tk_client->cl_rtt; 785 unsigned timer = task->tk_msg.rpc_proc->p_timer; 786 long m = usecs_to_jiffies(ktime_to_us(req->rq_rtt)); 787 788 if (timer) { 789 if (req->rq_ntrans == 1) 790 rpc_update_rtt(rtt, timer, m); 791 rpc_set_timeo(rtt, timer, req->rq_ntrans - 1); 792 } 793} 794 795/** 796 * xprt_complete_rqst - called when reply processing is complete 797 * @task: RPC request that recently completed 798 * @copied: actual number of bytes received from the transport 799 * 800 * Caller holds transport lock. 801 */ 802void xprt_complete_rqst(struct rpc_task *task, int copied) 803{ 804 struct rpc_rqst *req = task->tk_rqstp; 805 struct rpc_xprt *xprt = req->rq_xprt; 806 807 dprintk("RPC: %5u xid %08x complete (%d bytes received)\n", 808 task->tk_pid, ntohl(req->rq_xid), copied); 809 810 xprt->stat.recvs++; 811 req->rq_rtt = ktime_sub(ktime_get(), req->rq_xtime); 812 if (xprt->ops->timer != NULL) 813 xprt_update_rtt(task); 814 815 list_del_init(&req->rq_list); 816 req->rq_private_buf.len = copied; 817 /* Ensure all writes are done before we update */ 818 /* req->rq_reply_bytes_recvd */ 819 smp_wmb(); 820 req->rq_reply_bytes_recvd = copied; 821 rpc_wake_up_queued_task(&xprt->pending, task); 822} 823EXPORT_SYMBOL_GPL(xprt_complete_rqst); 824 825static void xprt_timer(struct rpc_task *task) 826{ 827 struct rpc_rqst *req = task->tk_rqstp; 828 struct rpc_xprt *xprt = req->rq_xprt; 829 830 if (task->tk_status != -ETIMEDOUT) 831 return; 832 dprintk("RPC: %5u xprt_timer\n", task->tk_pid); 833 834 spin_lock_bh(&xprt->transport_lock); 835 if (!req->rq_reply_bytes_recvd) { 836 if (xprt->ops->timer) 837 xprt->ops->timer(task); 838 } else 839 task->tk_status = 0; 840 spin_unlock_bh(&xprt->transport_lock); 841} 842 843static inline int xprt_has_timer(struct rpc_xprt *xprt) 844{ 845 return xprt->idle_timeout != 0; 846} 847 848/** 849 * xprt_prepare_transmit - reserve the transport before sending a request 850 * @task: RPC task about to send a request 851 * 852 */ 853int xprt_prepare_transmit(struct rpc_task *task) 854{ 855 struct rpc_rqst *req = task->tk_rqstp; 856 struct rpc_xprt *xprt = req->rq_xprt; 857 int err = 0; 858 859 dprintk("RPC: %5u xprt_prepare_transmit\n", task->tk_pid); 860 861 spin_lock_bh(&xprt->transport_lock); 862 if (req->rq_reply_bytes_recvd && !req->rq_bytes_sent) { 863 err = req->rq_reply_bytes_recvd; 864 goto out_unlock; 865 } 866 if (!xprt->ops->reserve_xprt(xprt, task)) 867 err = -EAGAIN; 868out_unlock: 869 spin_unlock_bh(&xprt->transport_lock); 870 return err; 871} 872 873void xprt_end_transmit(struct rpc_task *task) 874{ 875 xprt_release_write(task->tk_rqstp->rq_xprt, task); 876} 877 878/** 879 * xprt_transmit - send an RPC request on a transport 880 * @task: controlling RPC task 881 * 882 * We have to copy the iovec because sendmsg fiddles with its contents. 883 */ 884void xprt_transmit(struct rpc_task *task) 885{ 886 struct rpc_rqst *req = task->tk_rqstp; 887 struct rpc_xprt *xprt = req->rq_xprt; 888 int status, numreqs; 889 890 dprintk("RPC: %5u xprt_transmit(%u)\n", task->tk_pid, req->rq_slen); 891 892 if (!req->rq_reply_bytes_recvd) { 893 if (list_empty(&req->rq_list) && rpc_reply_expected(task)) { 894 /* 895 * Add to the list only if we're expecting a reply 896 */ 897 spin_lock_bh(&xprt->transport_lock); 898 /* Update the softirq receive buffer */ 899 memcpy(&req->rq_private_buf, &req->rq_rcv_buf, 900 sizeof(req->rq_private_buf)); 901 /* Add request to the receive list */ 902 list_add_tail(&req->rq_list, &xprt->recv); 903 spin_unlock_bh(&xprt->transport_lock); 904 xprt_reset_majortimeo(req); 905 /* Turn off autodisconnect */ 906 del_singleshot_timer_sync(&xprt->timer); 907 } 908 } else if (!req->rq_bytes_sent) 909 return; 910 911 req->rq_connect_cookie = xprt->connect_cookie; 912 req->rq_xtime = ktime_get(); 913 status = xprt->ops->send_request(task); 914 if (status != 0) { 915 task->tk_status = status; 916 return; 917 } 918 919 dprintk("RPC: %5u xmit complete\n", task->tk_pid); 920 task->tk_flags |= RPC_TASK_SENT; 921 spin_lock_bh(&xprt->transport_lock); 922 923 xprt->ops->set_retrans_timeout(task); 924 925 numreqs = atomic_read(&xprt->num_reqs); 926 if (numreqs > xprt->stat.max_slots) 927 xprt->stat.max_slots = numreqs; 928 xprt->stat.sends++; 929 xprt->stat.req_u += xprt->stat.sends - xprt->stat.recvs; 930 xprt->stat.bklog_u += xprt->backlog.qlen; 931 xprt->stat.sending_u += xprt->sending.qlen; 932 xprt->stat.pending_u += xprt->pending.qlen; 933 934 /* Don't race with disconnect */ 935 if (!xprt_connected(xprt)) 936 task->tk_status = -ENOTCONN; 937 else if (!req->rq_reply_bytes_recvd && rpc_reply_expected(task)) { 938 /* 939 * Sleep on the pending queue since 940 * we're expecting a reply. 941 */ 942 rpc_sleep_on(&xprt->pending, task, xprt_timer); 943 } 944 spin_unlock_bh(&xprt->transport_lock); 945} 946 947static struct rpc_rqst *xprt_dynamic_alloc_slot(struct rpc_xprt *xprt, gfp_t gfp_flags) 948{ 949 struct rpc_rqst *req = ERR_PTR(-EAGAIN); 950 951 if (!atomic_add_unless(&xprt->num_reqs, 1, xprt->max_reqs)) 952 goto out; 953 req = kzalloc(sizeof(struct rpc_rqst), gfp_flags); 954 if (req != NULL) 955 goto out; 956 atomic_dec(&xprt->num_reqs); 957 req = ERR_PTR(-ENOMEM); 958out: 959 return req; 960} 961 962static bool xprt_dynamic_free_slot(struct rpc_xprt *xprt, struct rpc_rqst *req) 963{ 964 if (atomic_add_unless(&xprt->num_reqs, -1, xprt->min_reqs)) { 965 kfree(req); 966 return true; 967 } 968 return false; 969} 970 971static void xprt_alloc_slot(struct rpc_task *task) 972{ 973 struct rpc_xprt *xprt = task->tk_xprt; 974 struct rpc_rqst *req; 975 976 if (!list_empty(&xprt->free)) { 977 req = list_entry(xprt->free.next, struct rpc_rqst, rq_list); 978 list_del(&req->rq_list); 979 goto out_init_req; 980 } 981 req = xprt_dynamic_alloc_slot(xprt, GFP_NOWAIT); 982 if (!IS_ERR(req)) 983 goto out_init_req; 984 switch (PTR_ERR(req)) { 985 case -ENOMEM: 986 rpc_delay(task, HZ >> 2); 987 dprintk("RPC: dynamic allocation of request slot " 988 "failed! Retrying\n"); 989 break; 990 case -EAGAIN: 991 rpc_sleep_on(&xprt->backlog, task, NULL); 992 dprintk("RPC: waiting for request slot\n"); 993 } 994 task->tk_status = -EAGAIN; 995 return; 996out_init_req: 997 task->tk_status = 0; 998 task->tk_rqstp = req; 999 xprt_request_init(task, xprt); 1000} 1001 1002static void xprt_free_slot(struct rpc_xprt *xprt, struct rpc_rqst *req) 1003{ 1004 spin_lock(&xprt->reserve_lock); 1005 if (!xprt_dynamic_free_slot(xprt, req)) { 1006 memset(req, 0, sizeof(*req)); /* mark unused */ 1007 list_add(&req->rq_list, &xprt->free); 1008 } 1009 rpc_wake_up_next(&xprt->backlog); 1010 spin_unlock(&xprt->reserve_lock); 1011} 1012 1013static void xprt_free_all_slots(struct rpc_xprt *xprt) 1014{ 1015 struct rpc_rqst *req; 1016 while (!list_empty(&xprt->free)) { 1017 req = list_first_entry(&xprt->free, struct rpc_rqst, rq_list); 1018 list_del(&req->rq_list); 1019 kfree(req); 1020 } 1021} 1022 1023struct rpc_xprt *xprt_alloc(struct net *net, size_t size, 1024 unsigned int num_prealloc, 1025 unsigned int max_alloc) 1026{ 1027 struct rpc_xprt *xprt; 1028 struct rpc_rqst *req; 1029 int i; 1030 1031 xprt = kzalloc(size, GFP_KERNEL); 1032 if (xprt == NULL) 1033 goto out; 1034 1035 xprt_init(xprt, net); 1036 1037 for (i = 0; i < num_prealloc; i++) { 1038 req = kzalloc(sizeof(struct rpc_rqst), GFP_KERNEL); 1039 if (!req) 1040 break; 1041 list_add(&req->rq_list, &xprt->free); 1042 } 1043 if (i < num_prealloc) 1044 goto out_free; 1045 if (max_alloc > num_prealloc) 1046 xprt->max_reqs = max_alloc; 1047 else 1048 xprt->max_reqs = num_prealloc; 1049 xprt->min_reqs = num_prealloc; 1050 atomic_set(&xprt->num_reqs, num_prealloc); 1051 1052 return xprt; 1053 1054out_free: 1055 xprt_free(xprt); 1056out: 1057 return NULL; 1058} 1059EXPORT_SYMBOL_GPL(xprt_alloc); 1060 1061void xprt_free(struct rpc_xprt *xprt) 1062{ 1063 put_net(xprt->xprt_net); 1064 xprt_free_all_slots(xprt); 1065 kfree(xprt); 1066} 1067EXPORT_SYMBOL_GPL(xprt_free); 1068 1069/** 1070 * xprt_reserve - allocate an RPC request slot 1071 * @task: RPC task requesting a slot allocation 1072 * 1073 * If no more slots are available, place the task on the transport's 1074 * backlog queue. 1075 */ 1076void xprt_reserve(struct rpc_task *task) 1077{ 1078 struct rpc_xprt *xprt = task->tk_xprt; 1079 1080 task->tk_status = 0; 1081 if (task->tk_rqstp != NULL) 1082 return; 1083 1084 /* Note: grabbing the xprt_lock_write() here is not strictly needed, 1085 * but ensures that we throttle new slot allocation if the transport 1086 * is congested (e.g. if reconnecting or if we're out of socket 1087 * write buffer space). 1088 */ 1089 task->tk_timeout = 0; 1090 task->tk_status = -EAGAIN; 1091 if (!xprt_lock_write(xprt, task)) 1092 return; 1093 1094 spin_lock(&xprt->reserve_lock); 1095 xprt_alloc_slot(task); 1096 spin_unlock(&xprt->reserve_lock); 1097 xprt_release_write(xprt, task); 1098} 1099 1100static inline __be32 xprt_alloc_xid(struct rpc_xprt *xprt) 1101{ 1102 return (__force __be32)xprt->xid++; 1103} 1104 1105static inline void xprt_init_xid(struct rpc_xprt *xprt) 1106{ 1107 xprt->xid = net_random(); 1108} 1109 1110static void xprt_request_init(struct rpc_task *task, struct rpc_xprt *xprt) 1111{ 1112 struct rpc_rqst *req = task->tk_rqstp; 1113 1114 INIT_LIST_HEAD(&req->rq_list); 1115 req->rq_timeout = task->tk_client->cl_timeout->to_initval; 1116 req->rq_task = task; 1117 req->rq_xprt = xprt; 1118 req->rq_buffer = NULL; 1119 req->rq_xid = xprt_alloc_xid(xprt); 1120 req->rq_release_snd_buf = NULL; 1121 xprt_reset_majortimeo(req); 1122 dprintk("RPC: %5u reserved req %p xid %08x\n", task->tk_pid, 1123 req, ntohl(req->rq_xid)); 1124} 1125 1126/** 1127 * xprt_release - release an RPC request slot 1128 * @task: task which is finished with the slot 1129 * 1130 */ 1131void xprt_release(struct rpc_task *task) 1132{ 1133 struct rpc_xprt *xprt; 1134 struct rpc_rqst *req; 1135 1136 if (!(req = task->tk_rqstp)) 1137 return; 1138 1139 xprt = req->rq_xprt; 1140 rpc_count_iostats(task); 1141 spin_lock_bh(&xprt->transport_lock); 1142 xprt->ops->release_xprt(xprt, task); 1143 if (xprt->ops->release_request) 1144 xprt->ops->release_request(task); 1145 if (!list_empty(&req->rq_list)) 1146 list_del(&req->rq_list); 1147 xprt->last_used = jiffies; 1148 if (list_empty(&xprt->recv) && xprt_has_timer(xprt)) 1149 mod_timer(&xprt->timer, 1150 xprt->last_used + xprt->idle_timeout); 1151 spin_unlock_bh(&xprt->transport_lock); 1152 if (req->rq_buffer) 1153 xprt->ops->buf_free(req->rq_buffer); 1154 if (req->rq_cred != NULL) 1155 put_rpccred(req->rq_cred); 1156 task->tk_rqstp = NULL; 1157 if (req->rq_release_snd_buf) 1158 req->rq_release_snd_buf(req); 1159 1160 dprintk("RPC: %5u release request %p\n", task->tk_pid, req); 1161 if (likely(!bc_prealloc(req))) 1162 xprt_free_slot(xprt, req); 1163 else 1164 xprt_free_bc_request(req); 1165} 1166 1167static void xprt_init(struct rpc_xprt *xprt, struct net *net) 1168{ 1169 atomic_set(&xprt->count, 1); 1170 1171 spin_lock_init(&xprt->transport_lock); 1172 spin_lock_init(&xprt->reserve_lock); 1173 1174 INIT_LIST_HEAD(&xprt->free); 1175 INIT_LIST_HEAD(&xprt->recv); 1176#if defined(CONFIG_SUNRPC_BACKCHANNEL) 1177 spin_lock_init(&xprt->bc_pa_lock); 1178 INIT_LIST_HEAD(&xprt->bc_pa_list); 1179#endif /* CONFIG_SUNRPC_BACKCHANNEL */ 1180 1181 xprt->last_used = jiffies; 1182 xprt->cwnd = RPC_INITCWND; 1183 xprt->bind_index = 0; 1184 1185 rpc_init_wait_queue(&xprt->binding, "xprt_binding"); 1186 rpc_init_wait_queue(&xprt->pending, "xprt_pending"); 1187 rpc_init_priority_wait_queue(&xprt->sending, "xprt_sending"); 1188 rpc_init_priority_wait_queue(&xprt->backlog, "xprt_backlog"); 1189 1190 xprt_init_xid(xprt); 1191 1192 xprt->xprt_net = get_net(net); 1193} 1194 1195/** 1196 * xprt_create_transport - create an RPC transport 1197 * @args: rpc transport creation arguments 1198 * 1199 */ 1200struct rpc_xprt *xprt_create_transport(struct xprt_create *args) 1201{ 1202 struct rpc_xprt *xprt; 1203 struct xprt_class *t; 1204 1205 spin_lock(&xprt_list_lock); 1206 list_for_each_entry(t, &xprt_list, list) { 1207 if (t->ident == args->ident) { 1208 spin_unlock(&xprt_list_lock); 1209 goto found; 1210 } 1211 } 1212 spin_unlock(&xprt_list_lock); 1213 printk(KERN_ERR "RPC: transport (%d) not supported\n", args->ident); 1214 return ERR_PTR(-EIO); 1215 1216found: 1217 xprt = t->setup(args); 1218 if (IS_ERR(xprt)) { 1219 dprintk("RPC: xprt_create_transport: failed, %ld\n", 1220 -PTR_ERR(xprt)); 1221 goto out; 1222 } 1223 INIT_WORK(&xprt->task_cleanup, xprt_autoclose); 1224 if (xprt_has_timer(xprt)) 1225 setup_timer(&xprt->timer, xprt_init_autodisconnect, 1226 (unsigned long)xprt); 1227 else 1228 init_timer(&xprt->timer); 1229 dprintk("RPC: created transport %p with %u slots\n", xprt, 1230 xprt->max_reqs); 1231out: 1232 return xprt; 1233} 1234 1235/** 1236 * xprt_destroy - destroy an RPC transport, killing off all requests. 1237 * @xprt: transport to destroy 1238 * 1239 */ 1240static void xprt_destroy(struct rpc_xprt *xprt) 1241{ 1242 dprintk("RPC: destroying transport %p\n", xprt); 1243 xprt->shutdown = 1; 1244 del_timer_sync(&xprt->timer); 1245 1246 rpc_destroy_wait_queue(&xprt->binding); 1247 rpc_destroy_wait_queue(&xprt->pending); 1248 rpc_destroy_wait_queue(&xprt->sending); 1249 rpc_destroy_wait_queue(&xprt->backlog); 1250 cancel_work_sync(&xprt->task_cleanup); 1251 /* 1252 * Tear down transport state and free the rpc_xprt 1253 */ 1254 xprt->ops->destroy(xprt); 1255} 1256 1257/** 1258 * xprt_put - release a reference to an RPC transport. 1259 * @xprt: pointer to the transport 1260 * 1261 */ 1262void xprt_put(struct rpc_xprt *xprt) 1263{ 1264 if (atomic_dec_and_test(&xprt->count)) 1265 xprt_destroy(xprt); 1266} 1267 1268/** 1269 * xprt_get - return a reference to an RPC transport. 1270 * @xprt: pointer to the transport 1271 * 1272 */ 1273struct rpc_xprt *xprt_get(struct rpc_xprt *xprt) 1274{ 1275 if (atomic_inc_not_zero(&xprt->count)) 1276 return xprt; 1277 return NULL; 1278} 1279