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