xprt.c revision 87e3c0553fcbea79bf9f17fc5694484ecf3ae5e8
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; 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 xprt->stat.sends++; 926 xprt->stat.req_u += xprt->stat.sends - xprt->stat.recvs; 927 xprt->stat.bklog_u += xprt->backlog.qlen; 928 929 /* Don't race with disconnect */ 930 if (!xprt_connected(xprt)) 931 task->tk_status = -ENOTCONN; 932 else if (!req->rq_reply_bytes_recvd && rpc_reply_expected(task)) { 933 /* 934 * Sleep on the pending queue since 935 * we're expecting a reply. 936 */ 937 rpc_sleep_on(&xprt->pending, task, xprt_timer); 938 } 939 spin_unlock_bh(&xprt->transport_lock); 940} 941 942static struct rpc_rqst *xprt_dynamic_alloc_slot(struct rpc_xprt *xprt, gfp_t gfp_flags) 943{ 944 struct rpc_rqst *req = ERR_PTR(-EAGAIN); 945 946 if (!atomic_add_unless(&xprt->num_reqs, 1, xprt->max_reqs)) 947 goto out; 948 req = kzalloc(sizeof(struct rpc_rqst), gfp_flags); 949 if (req != NULL) 950 goto out; 951 atomic_dec(&xprt->num_reqs); 952 req = ERR_PTR(-ENOMEM); 953out: 954 return req; 955} 956 957static bool xprt_dynamic_free_slot(struct rpc_xprt *xprt, struct rpc_rqst *req) 958{ 959 if (atomic_add_unless(&xprt->num_reqs, -1, xprt->min_reqs)) { 960 kfree(req); 961 return true; 962 } 963 return false; 964} 965 966static void xprt_alloc_slot(struct rpc_task *task) 967{ 968 struct rpc_xprt *xprt = task->tk_xprt; 969 struct rpc_rqst *req; 970 971 if (!list_empty(&xprt->free)) { 972 req = list_entry(xprt->free.next, struct rpc_rqst, rq_list); 973 list_del(&req->rq_list); 974 goto out_init_req; 975 } 976 req = xprt_dynamic_alloc_slot(xprt, GFP_NOWAIT); 977 if (!IS_ERR(req)) 978 goto out_init_req; 979 switch (PTR_ERR(req)) { 980 case -ENOMEM: 981 rpc_delay(task, HZ >> 2); 982 dprintk("RPC: dynamic allocation of request slot " 983 "failed! Retrying\n"); 984 break; 985 case -EAGAIN: 986 rpc_sleep_on(&xprt->backlog, task, NULL); 987 dprintk("RPC: waiting for request slot\n"); 988 } 989 task->tk_status = -EAGAIN; 990 return; 991out_init_req: 992 task->tk_status = 0; 993 task->tk_rqstp = req; 994 xprt_request_init(task, xprt); 995} 996 997static void xprt_free_slot(struct rpc_xprt *xprt, struct rpc_rqst *req) 998{ 999 spin_lock(&xprt->reserve_lock); 1000 if (!xprt_dynamic_free_slot(xprt, req)) { 1001 memset(req, 0, sizeof(*req)); /* mark unused */ 1002 list_add(&req->rq_list, &xprt->free); 1003 } 1004 rpc_wake_up_next(&xprt->backlog); 1005 spin_unlock(&xprt->reserve_lock); 1006} 1007 1008static void xprt_free_all_slots(struct rpc_xprt *xprt) 1009{ 1010 struct rpc_rqst *req; 1011 while (!list_empty(&xprt->free)) { 1012 req = list_first_entry(&xprt->free, struct rpc_rqst, rq_list); 1013 list_del(&req->rq_list); 1014 kfree(req); 1015 } 1016} 1017 1018struct rpc_xprt *xprt_alloc(struct net *net, size_t size, 1019 unsigned int num_prealloc, 1020 unsigned int max_alloc) 1021{ 1022 struct rpc_xprt *xprt; 1023 struct rpc_rqst *req; 1024 int i; 1025 1026 xprt = kzalloc(size, GFP_KERNEL); 1027 if (xprt == NULL) 1028 goto out; 1029 1030 xprt_init(xprt, net); 1031 1032 for (i = 0; i < num_prealloc; i++) { 1033 req = kzalloc(sizeof(struct rpc_rqst), GFP_KERNEL); 1034 if (!req) 1035 break; 1036 list_add(&req->rq_list, &xprt->free); 1037 } 1038 if (i < num_prealloc) 1039 goto out_free; 1040 if (max_alloc > num_prealloc) 1041 xprt->max_reqs = max_alloc; 1042 else 1043 xprt->max_reqs = num_prealloc; 1044 xprt->min_reqs = num_prealloc; 1045 atomic_set(&xprt->num_reqs, num_prealloc); 1046 1047 return xprt; 1048 1049out_free: 1050 xprt_free(xprt); 1051out: 1052 return NULL; 1053} 1054EXPORT_SYMBOL_GPL(xprt_alloc); 1055 1056void xprt_free(struct rpc_xprt *xprt) 1057{ 1058 put_net(xprt->xprt_net); 1059 xprt_free_all_slots(xprt); 1060 kfree(xprt); 1061} 1062EXPORT_SYMBOL_GPL(xprt_free); 1063 1064/** 1065 * xprt_reserve - allocate an RPC request slot 1066 * @task: RPC task requesting a slot allocation 1067 * 1068 * If no more slots are available, place the task on the transport's 1069 * backlog queue. 1070 */ 1071void xprt_reserve(struct rpc_task *task) 1072{ 1073 struct rpc_xprt *xprt = task->tk_xprt; 1074 1075 task->tk_status = 0; 1076 if (task->tk_rqstp != NULL) 1077 return; 1078 1079 /* Note: grabbing the xprt_lock_write() here is not strictly needed, 1080 * but ensures that we throttle new slot allocation if the transport 1081 * is congested (e.g. if reconnecting or if we're out of socket 1082 * write buffer space). 1083 */ 1084 task->tk_timeout = 0; 1085 task->tk_status = -EAGAIN; 1086 if (!xprt_lock_write(xprt, task)) 1087 return; 1088 1089 spin_lock(&xprt->reserve_lock); 1090 xprt_alloc_slot(task); 1091 spin_unlock(&xprt->reserve_lock); 1092 xprt_release_write(xprt, task); 1093} 1094 1095static inline __be32 xprt_alloc_xid(struct rpc_xprt *xprt) 1096{ 1097 return (__force __be32)xprt->xid++; 1098} 1099 1100static inline void xprt_init_xid(struct rpc_xprt *xprt) 1101{ 1102 xprt->xid = net_random(); 1103} 1104 1105static void xprt_request_init(struct rpc_task *task, struct rpc_xprt *xprt) 1106{ 1107 struct rpc_rqst *req = task->tk_rqstp; 1108 1109 INIT_LIST_HEAD(&req->rq_list); 1110 req->rq_timeout = task->tk_client->cl_timeout->to_initval; 1111 req->rq_task = task; 1112 req->rq_xprt = xprt; 1113 req->rq_buffer = NULL; 1114 req->rq_xid = xprt_alloc_xid(xprt); 1115 req->rq_release_snd_buf = NULL; 1116 xprt_reset_majortimeo(req); 1117 dprintk("RPC: %5u reserved req %p xid %08x\n", task->tk_pid, 1118 req, ntohl(req->rq_xid)); 1119} 1120 1121/** 1122 * xprt_release - release an RPC request slot 1123 * @task: task which is finished with the slot 1124 * 1125 */ 1126void xprt_release(struct rpc_task *task) 1127{ 1128 struct rpc_xprt *xprt; 1129 struct rpc_rqst *req; 1130 1131 if (!(req = task->tk_rqstp)) 1132 return; 1133 1134 xprt = req->rq_xprt; 1135 rpc_count_iostats(task); 1136 spin_lock_bh(&xprt->transport_lock); 1137 xprt->ops->release_xprt(xprt, task); 1138 if (xprt->ops->release_request) 1139 xprt->ops->release_request(task); 1140 if (!list_empty(&req->rq_list)) 1141 list_del(&req->rq_list); 1142 xprt->last_used = jiffies; 1143 if (list_empty(&xprt->recv) && xprt_has_timer(xprt)) 1144 mod_timer(&xprt->timer, 1145 xprt->last_used + xprt->idle_timeout); 1146 spin_unlock_bh(&xprt->transport_lock); 1147 if (req->rq_buffer) 1148 xprt->ops->buf_free(req->rq_buffer); 1149 if (req->rq_cred != NULL) 1150 put_rpccred(req->rq_cred); 1151 task->tk_rqstp = NULL; 1152 if (req->rq_release_snd_buf) 1153 req->rq_release_snd_buf(req); 1154 1155 dprintk("RPC: %5u release request %p\n", task->tk_pid, req); 1156 if (likely(!bc_prealloc(req))) 1157 xprt_free_slot(xprt, req); 1158 else 1159 xprt_free_bc_request(req); 1160} 1161 1162static void xprt_init(struct rpc_xprt *xprt, struct net *net) 1163{ 1164 atomic_set(&xprt->count, 1); 1165 1166 spin_lock_init(&xprt->transport_lock); 1167 spin_lock_init(&xprt->reserve_lock); 1168 1169 INIT_LIST_HEAD(&xprt->free); 1170 INIT_LIST_HEAD(&xprt->recv); 1171#if defined(CONFIG_SUNRPC_BACKCHANNEL) 1172 spin_lock_init(&xprt->bc_pa_lock); 1173 INIT_LIST_HEAD(&xprt->bc_pa_list); 1174#endif /* CONFIG_SUNRPC_BACKCHANNEL */ 1175 1176 xprt->last_used = jiffies; 1177 xprt->cwnd = RPC_INITCWND; 1178 xprt->bind_index = 0; 1179 1180 rpc_init_wait_queue(&xprt->binding, "xprt_binding"); 1181 rpc_init_wait_queue(&xprt->pending, "xprt_pending"); 1182 rpc_init_priority_wait_queue(&xprt->sending, "xprt_sending"); 1183 rpc_init_priority_wait_queue(&xprt->backlog, "xprt_backlog"); 1184 1185 xprt_init_xid(xprt); 1186 1187 xprt->xprt_net = get_net(net); 1188} 1189 1190/** 1191 * xprt_create_transport - create an RPC transport 1192 * @args: rpc transport creation arguments 1193 * 1194 */ 1195struct rpc_xprt *xprt_create_transport(struct xprt_create *args) 1196{ 1197 struct rpc_xprt *xprt; 1198 struct xprt_class *t; 1199 1200 spin_lock(&xprt_list_lock); 1201 list_for_each_entry(t, &xprt_list, list) { 1202 if (t->ident == args->ident) { 1203 spin_unlock(&xprt_list_lock); 1204 goto found; 1205 } 1206 } 1207 spin_unlock(&xprt_list_lock); 1208 printk(KERN_ERR "RPC: transport (%d) not supported\n", args->ident); 1209 return ERR_PTR(-EIO); 1210 1211found: 1212 xprt = t->setup(args); 1213 if (IS_ERR(xprt)) { 1214 dprintk("RPC: xprt_create_transport: failed, %ld\n", 1215 -PTR_ERR(xprt)); 1216 goto out; 1217 } 1218 INIT_WORK(&xprt->task_cleanup, xprt_autoclose); 1219 if (xprt_has_timer(xprt)) 1220 setup_timer(&xprt->timer, xprt_init_autodisconnect, 1221 (unsigned long)xprt); 1222 else 1223 init_timer(&xprt->timer); 1224 dprintk("RPC: created transport %p with %u slots\n", xprt, 1225 xprt->max_reqs); 1226out: 1227 return xprt; 1228} 1229 1230/** 1231 * xprt_destroy - destroy an RPC transport, killing off all requests. 1232 * @xprt: transport to destroy 1233 * 1234 */ 1235static void xprt_destroy(struct rpc_xprt *xprt) 1236{ 1237 dprintk("RPC: destroying transport %p\n", xprt); 1238 xprt->shutdown = 1; 1239 del_timer_sync(&xprt->timer); 1240 1241 rpc_destroy_wait_queue(&xprt->binding); 1242 rpc_destroy_wait_queue(&xprt->pending); 1243 rpc_destroy_wait_queue(&xprt->sending); 1244 rpc_destroy_wait_queue(&xprt->backlog); 1245 cancel_work_sync(&xprt->task_cleanup); 1246 /* 1247 * Tear down transport state and free the rpc_xprt 1248 */ 1249 xprt->ops->destroy(xprt); 1250} 1251 1252/** 1253 * xprt_put - release a reference to an RPC transport. 1254 * @xprt: pointer to the transport 1255 * 1256 */ 1257void xprt_put(struct rpc_xprt *xprt) 1258{ 1259 if (atomic_dec_and_test(&xprt->count)) 1260 xprt_destroy(xprt); 1261} 1262 1263/** 1264 * xprt_get - return a reference to an RPC transport. 1265 * @xprt: pointer to the transport 1266 * 1267 */ 1268struct rpc_xprt *xprt_get(struct rpc_xprt *xprt) 1269{ 1270 if (atomic_inc_not_zero(&xprt->count)) 1271 return xprt; 1272 return NULL; 1273} 1274