1 // SPDX-License-Identifier: GPL-2.0-only 2 /* 3 * linux/net/sunrpc/xprt.c 4 * 5 * This is a generic RPC call interface supporting congestion avoidance, 6 * and asynchronous calls. 7 * 8 * The interface works like this: 9 * 10 * - When a process places a call, it allocates a request slot if 11 * one is available. Otherwise, it sleeps on the backlog queue 12 * (xprt_reserve). 13 * - Next, the caller puts together the RPC message, stuffs it into 14 * the request struct, and calls xprt_transmit(). 15 * - xprt_transmit sends the message and installs the caller on the 16 * transport's wait list. At the same time, if a reply is expected, 17 * it installs a timer that is run after the packet's timeout has 18 * expired. 19 * - When a packet arrives, the data_ready handler walks the list of 20 * pending requests for that transport. If a matching XID is found, the 21 * caller is woken up, and the timer removed. 22 * - When no reply arrives within the timeout interval, the timer is 23 * fired by the kernel and runs xprt_timer(). It either adjusts the 24 * timeout values (minor timeout) or wakes up the caller with a status 25 * of -ETIMEDOUT. 26 * - When the caller receives a notification from RPC that a reply arrived, 27 * it should release the RPC slot, and process the reply. 28 * If the call timed out, it may choose to retry the operation by 29 * adjusting the initial timeout value, and simply calling rpc_call 30 * again. 31 * 32 * Support for async RPC is done through a set of RPC-specific scheduling 33 * primitives that `transparently' work for processes as well as async 34 * tasks that rely on callbacks. 35 * 36 * Copyright (C) 1995-1997, Olaf Kirch <okir@monad.swb.de> 37 * 38 * Transport switch API copyright (C) 2005, Chuck Lever <cel@netapp.com> 39 */ 40 41 #include <linux/module.h> 42 43 #include <linux/types.h> 44 #include <linux/interrupt.h> 45 #include <linux/workqueue.h> 46 #include <linux/net.h> 47 #include <linux/ktime.h> 48 49 #include <linux/sunrpc/clnt.h> 50 #include <linux/sunrpc/metrics.h> 51 #include <linux/sunrpc/bc_xprt.h> 52 #include <linux/rcupdate.h> 53 #include <linux/sched/mm.h> 54 55 #include <trace/events/sunrpc.h> 56 57 #include "sunrpc.h" 58 #include "sysfs.h" 59 #include "fail.h" 60 61 /* 62 * Local variables 63 */ 64 65 #if IS_ENABLED(CONFIG_SUNRPC_DEBUG) 66 # define RPCDBG_FACILITY RPCDBG_XPRT 67 #endif 68 69 /* 70 * Local functions 71 */ 72 static void xprt_init(struct rpc_xprt *xprt, struct net *net); 73 static __be32 xprt_alloc_xid(struct rpc_xprt *xprt); 74 static void xprt_destroy(struct rpc_xprt *xprt); 75 static void xprt_request_init(struct rpc_task *task); 76 77 static DEFINE_SPINLOCK(xprt_list_lock); 78 static LIST_HEAD(xprt_list); 79 80 static unsigned long xprt_request_timeout(const struct rpc_rqst *req) 81 { 82 unsigned long timeout = jiffies + req->rq_timeout; 83 84 if (time_before(timeout, req->rq_majortimeo)) 85 return timeout; 86 return req->rq_majortimeo; 87 } 88 89 /** 90 * xprt_register_transport - register a transport implementation 91 * @transport: transport to register 92 * 93 * If a transport implementation is loaded as a kernel module, it can 94 * call this interface to make itself known to the RPC client. 95 * 96 * Returns: 97 * 0: transport successfully registered 98 * -EEXIST: transport already registered 99 * -EINVAL: transport module being unloaded 100 */ 101 int xprt_register_transport(struct xprt_class *transport) 102 { 103 struct xprt_class *t; 104 int result; 105 106 result = -EEXIST; 107 spin_lock(&xprt_list_lock); 108 list_for_each_entry(t, &xprt_list, list) { 109 /* don't register the same transport class twice */ 110 if (t->ident == transport->ident) 111 goto out; 112 } 113 114 list_add_tail(&transport->list, &xprt_list); 115 printk(KERN_INFO "RPC: Registered %s transport module.\n", 116 transport->name); 117 result = 0; 118 119 out: 120 spin_unlock(&xprt_list_lock); 121 return result; 122 } 123 EXPORT_SYMBOL_GPL(xprt_register_transport); 124 125 /** 126 * xprt_unregister_transport - unregister a transport implementation 127 * @transport: transport to unregister 128 * 129 * Returns: 130 * 0: transport successfully unregistered 131 * -ENOENT: transport never registered 132 */ 133 int xprt_unregister_transport(struct xprt_class *transport) 134 { 135 struct xprt_class *t; 136 int result; 137 138 result = 0; 139 spin_lock(&xprt_list_lock); 140 list_for_each_entry(t, &xprt_list, list) { 141 if (t == transport) { 142 printk(KERN_INFO 143 "RPC: Unregistered %s transport module.\n", 144 transport->name); 145 list_del_init(&transport->list); 146 goto out; 147 } 148 } 149 result = -ENOENT; 150 151 out: 152 spin_unlock(&xprt_list_lock); 153 return result; 154 } 155 EXPORT_SYMBOL_GPL(xprt_unregister_transport); 156 157 static void 158 xprt_class_release(const struct xprt_class *t) 159 { 160 module_put(t->owner); 161 } 162 163 static const struct xprt_class * 164 xprt_class_find_by_ident_locked(int ident) 165 { 166 const struct xprt_class *t; 167 168 list_for_each_entry(t, &xprt_list, list) { 169 if (t->ident != ident) 170 continue; 171 if (!try_module_get(t->owner)) 172 continue; 173 return t; 174 } 175 return NULL; 176 } 177 178 static const struct xprt_class * 179 xprt_class_find_by_ident(int ident) 180 { 181 const struct xprt_class *t; 182 183 spin_lock(&xprt_list_lock); 184 t = xprt_class_find_by_ident_locked(ident); 185 spin_unlock(&xprt_list_lock); 186 return t; 187 } 188 189 static const struct xprt_class * 190 xprt_class_find_by_netid_locked(const char *netid) 191 { 192 const struct xprt_class *t; 193 unsigned int i; 194 195 list_for_each_entry(t, &xprt_list, list) { 196 for (i = 0; t->netid[i][0] != '\0'; i++) { 197 if (strcmp(t->netid[i], netid) != 0) 198 continue; 199 if (!try_module_get(t->owner)) 200 continue; 201 return t; 202 } 203 } 204 return NULL; 205 } 206 207 static const struct xprt_class * 208 xprt_class_find_by_netid(const char *netid) 209 { 210 const struct xprt_class *t; 211 212 spin_lock(&xprt_list_lock); 213 t = xprt_class_find_by_netid_locked(netid); 214 if (!t) { 215 spin_unlock(&xprt_list_lock); 216 request_module("rpc%s", netid); 217 spin_lock(&xprt_list_lock); 218 t = xprt_class_find_by_netid_locked(netid); 219 } 220 spin_unlock(&xprt_list_lock); 221 return t; 222 } 223 224 /** 225 * xprt_find_transport_ident - convert a netid into a transport identifier 226 * @netid: transport to load 227 * 228 * Returns: 229 * > 0: transport identifier 230 * -ENOENT: transport module not available 231 */ 232 int xprt_find_transport_ident(const char *netid) 233 { 234 const struct xprt_class *t; 235 int ret; 236 237 t = xprt_class_find_by_netid(netid); 238 if (!t) 239 return -ENOENT; 240 ret = t->ident; 241 xprt_class_release(t); 242 return ret; 243 } 244 EXPORT_SYMBOL_GPL(xprt_find_transport_ident); 245 246 static void xprt_clear_locked(struct rpc_xprt *xprt) 247 { 248 xprt->snd_task = NULL; 249 if (!test_bit(XPRT_CLOSE_WAIT, &xprt->state)) 250 clear_bit_unlock(XPRT_LOCKED, &xprt->state); 251 else 252 queue_work(xprtiod_workqueue, &xprt->task_cleanup); 253 } 254 255 /** 256 * xprt_reserve_xprt - serialize write access to transports 257 * @task: task that is requesting access to the transport 258 * @xprt: pointer to the target transport 259 * 260 * This prevents mixing the payload of separate requests, and prevents 261 * transport connects from colliding with writes. No congestion control 262 * is provided. 263 */ 264 int xprt_reserve_xprt(struct rpc_xprt *xprt, struct rpc_task *task) 265 { 266 struct rpc_rqst *req = task->tk_rqstp; 267 268 if (test_and_set_bit(XPRT_LOCKED, &xprt->state)) { 269 if (task == xprt->snd_task) 270 goto out_locked; 271 goto out_sleep; 272 } 273 if (test_bit(XPRT_WRITE_SPACE, &xprt->state)) 274 goto out_unlock; 275 xprt->snd_task = task; 276 277 out_locked: 278 trace_xprt_reserve_xprt(xprt, task); 279 return 1; 280 281 out_unlock: 282 xprt_clear_locked(xprt); 283 out_sleep: 284 task->tk_status = -EAGAIN; 285 if (RPC_IS_SOFT(task)) 286 rpc_sleep_on_timeout(&xprt->sending, task, NULL, 287 xprt_request_timeout(req)); 288 else 289 rpc_sleep_on(&xprt->sending, task, NULL); 290 return 0; 291 } 292 EXPORT_SYMBOL_GPL(xprt_reserve_xprt); 293 294 static bool 295 xprt_need_congestion_window_wait(struct rpc_xprt *xprt) 296 { 297 return test_bit(XPRT_CWND_WAIT, &xprt->state); 298 } 299 300 static void 301 xprt_set_congestion_window_wait(struct rpc_xprt *xprt) 302 { 303 if (!list_empty(&xprt->xmit_queue)) { 304 /* Peek at head of queue to see if it can make progress */ 305 if (list_first_entry(&xprt->xmit_queue, struct rpc_rqst, 306 rq_xmit)->rq_cong) 307 return; 308 } 309 set_bit(XPRT_CWND_WAIT, &xprt->state); 310 } 311 312 static void 313 xprt_test_and_clear_congestion_window_wait(struct rpc_xprt *xprt) 314 { 315 if (!RPCXPRT_CONGESTED(xprt)) 316 clear_bit(XPRT_CWND_WAIT, &xprt->state); 317 } 318 319 /* 320 * xprt_reserve_xprt_cong - serialize write access to transports 321 * @task: task that is requesting access to the transport 322 * 323 * Same as xprt_reserve_xprt, but Van Jacobson congestion control is 324 * integrated into the decision of whether a request is allowed to be 325 * woken up and given access to the transport. 326 * Note that the lock is only granted if we know there are free slots. 327 */ 328 int xprt_reserve_xprt_cong(struct rpc_xprt *xprt, struct rpc_task *task) 329 { 330 struct rpc_rqst *req = task->tk_rqstp; 331 332 if (test_and_set_bit(XPRT_LOCKED, &xprt->state)) { 333 if (task == xprt->snd_task) 334 goto out_locked; 335 goto out_sleep; 336 } 337 if (req == NULL) { 338 xprt->snd_task = task; 339 goto out_locked; 340 } 341 if (test_bit(XPRT_WRITE_SPACE, &xprt->state)) 342 goto out_unlock; 343 if (!xprt_need_congestion_window_wait(xprt)) { 344 xprt->snd_task = task; 345 goto out_locked; 346 } 347 out_unlock: 348 xprt_clear_locked(xprt); 349 out_sleep: 350 task->tk_status = -EAGAIN; 351 if (RPC_IS_SOFT(task)) 352 rpc_sleep_on_timeout(&xprt->sending, task, NULL, 353 xprt_request_timeout(req)); 354 else 355 rpc_sleep_on(&xprt->sending, task, NULL); 356 return 0; 357 out_locked: 358 trace_xprt_reserve_cong(xprt, task); 359 return 1; 360 } 361 EXPORT_SYMBOL_GPL(xprt_reserve_xprt_cong); 362 363 static inline int xprt_lock_write(struct rpc_xprt *xprt, struct rpc_task *task) 364 { 365 int retval; 366 367 if (test_bit(XPRT_LOCKED, &xprt->state) && xprt->snd_task == task) 368 return 1; 369 spin_lock(&xprt->transport_lock); 370 retval = xprt->ops->reserve_xprt(xprt, task); 371 spin_unlock(&xprt->transport_lock); 372 return retval; 373 } 374 375 static bool __xprt_lock_write_func(struct rpc_task *task, void *data) 376 { 377 struct rpc_xprt *xprt = data; 378 379 xprt->snd_task = task; 380 return true; 381 } 382 383 static void __xprt_lock_write_next(struct rpc_xprt *xprt) 384 { 385 if (test_and_set_bit(XPRT_LOCKED, &xprt->state)) 386 return; 387 if (test_bit(XPRT_WRITE_SPACE, &xprt->state)) 388 goto out_unlock; 389 if (rpc_wake_up_first_on_wq(xprtiod_workqueue, &xprt->sending, 390 __xprt_lock_write_func, xprt)) 391 return; 392 out_unlock: 393 xprt_clear_locked(xprt); 394 } 395 396 static void __xprt_lock_write_next_cong(struct rpc_xprt *xprt) 397 { 398 if (test_and_set_bit(XPRT_LOCKED, &xprt->state)) 399 return; 400 if (test_bit(XPRT_WRITE_SPACE, &xprt->state)) 401 goto out_unlock; 402 if (xprt_need_congestion_window_wait(xprt)) 403 goto out_unlock; 404 if (rpc_wake_up_first_on_wq(xprtiod_workqueue, &xprt->sending, 405 __xprt_lock_write_func, xprt)) 406 return; 407 out_unlock: 408 xprt_clear_locked(xprt); 409 } 410 411 /** 412 * xprt_release_xprt - allow other requests to use a transport 413 * @xprt: transport with other tasks potentially waiting 414 * @task: task that is releasing access to the transport 415 * 416 * Note that "task" can be NULL. No congestion control is provided. 417 */ 418 void xprt_release_xprt(struct rpc_xprt *xprt, struct rpc_task *task) 419 { 420 if (xprt->snd_task == task) { 421 xprt_clear_locked(xprt); 422 __xprt_lock_write_next(xprt); 423 } 424 trace_xprt_release_xprt(xprt, task); 425 } 426 EXPORT_SYMBOL_GPL(xprt_release_xprt); 427 428 /** 429 * xprt_release_xprt_cong - allow other requests to use a transport 430 * @xprt: transport with other tasks potentially waiting 431 * @task: task that is releasing access to the transport 432 * 433 * Note that "task" can be NULL. Another task is awoken to use the 434 * transport if the transport's congestion window allows it. 435 */ 436 void xprt_release_xprt_cong(struct rpc_xprt *xprt, struct rpc_task *task) 437 { 438 if (xprt->snd_task == task) { 439 xprt_clear_locked(xprt); 440 __xprt_lock_write_next_cong(xprt); 441 } 442 trace_xprt_release_cong(xprt, task); 443 } 444 EXPORT_SYMBOL_GPL(xprt_release_xprt_cong); 445 446 void xprt_release_write(struct rpc_xprt *xprt, struct rpc_task *task) 447 { 448 if (xprt->snd_task != task) 449 return; 450 spin_lock(&xprt->transport_lock); 451 xprt->ops->release_xprt(xprt, task); 452 spin_unlock(&xprt->transport_lock); 453 } 454 455 /* 456 * Van Jacobson congestion avoidance. Check if the congestion window 457 * overflowed. Put the task to sleep if this is the case. 458 */ 459 static int 460 __xprt_get_cong(struct rpc_xprt *xprt, struct rpc_rqst *req) 461 { 462 if (req->rq_cong) 463 return 1; 464 trace_xprt_get_cong(xprt, req->rq_task); 465 if (RPCXPRT_CONGESTED(xprt)) { 466 xprt_set_congestion_window_wait(xprt); 467 return 0; 468 } 469 req->rq_cong = 1; 470 xprt->cong += RPC_CWNDSCALE; 471 return 1; 472 } 473 474 /* 475 * Adjust the congestion window, and wake up the next task 476 * that has been sleeping due to congestion 477 */ 478 static void 479 __xprt_put_cong(struct rpc_xprt *xprt, struct rpc_rqst *req) 480 { 481 if (!req->rq_cong) 482 return; 483 req->rq_cong = 0; 484 xprt->cong -= RPC_CWNDSCALE; 485 xprt_test_and_clear_congestion_window_wait(xprt); 486 trace_xprt_put_cong(xprt, req->rq_task); 487 __xprt_lock_write_next_cong(xprt); 488 } 489 490 /** 491 * xprt_request_get_cong - Request congestion control credits 492 * @xprt: pointer to transport 493 * @req: pointer to RPC request 494 * 495 * Useful for transports that require congestion control. 496 */ 497 bool 498 xprt_request_get_cong(struct rpc_xprt *xprt, struct rpc_rqst *req) 499 { 500 bool ret = false; 501 502 if (req->rq_cong) 503 return true; 504 spin_lock(&xprt->transport_lock); 505 ret = __xprt_get_cong(xprt, req) != 0; 506 spin_unlock(&xprt->transport_lock); 507 return ret; 508 } 509 EXPORT_SYMBOL_GPL(xprt_request_get_cong); 510 511 /** 512 * xprt_release_rqst_cong - housekeeping when request is complete 513 * @task: RPC request that recently completed 514 * 515 * Useful for transports that require congestion control. 516 */ 517 void xprt_release_rqst_cong(struct rpc_task *task) 518 { 519 struct rpc_rqst *req = task->tk_rqstp; 520 521 __xprt_put_cong(req->rq_xprt, req); 522 } 523 EXPORT_SYMBOL_GPL(xprt_release_rqst_cong); 524 525 static void xprt_clear_congestion_window_wait_locked(struct rpc_xprt *xprt) 526 { 527 if (test_and_clear_bit(XPRT_CWND_WAIT, &xprt->state)) 528 __xprt_lock_write_next_cong(xprt); 529 } 530 531 /* 532 * Clear the congestion window wait flag and wake up the next 533 * entry on xprt->sending 534 */ 535 static void 536 xprt_clear_congestion_window_wait(struct rpc_xprt *xprt) 537 { 538 if (test_and_clear_bit(XPRT_CWND_WAIT, &xprt->state)) { 539 spin_lock(&xprt->transport_lock); 540 __xprt_lock_write_next_cong(xprt); 541 spin_unlock(&xprt->transport_lock); 542 } 543 } 544 545 /** 546 * xprt_adjust_cwnd - adjust transport congestion window 547 * @xprt: pointer to xprt 548 * @task: recently completed RPC request used to adjust window 549 * @result: result code of completed RPC request 550 * 551 * The transport code maintains an estimate on the maximum number of out- 552 * standing RPC requests, using a smoothed version of the congestion 553 * avoidance implemented in 44BSD. This is basically the Van Jacobson 554 * congestion algorithm: If a retransmit occurs, the congestion window is 555 * halved; otherwise, it is incremented by 1/cwnd when 556 * 557 * - a reply is received and 558 * - a full number of requests are outstanding and 559 * - the congestion window hasn't been updated recently. 560 */ 561 void xprt_adjust_cwnd(struct rpc_xprt *xprt, struct rpc_task *task, int result) 562 { 563 struct rpc_rqst *req = task->tk_rqstp; 564 unsigned long cwnd = xprt->cwnd; 565 566 if (result >= 0 && cwnd <= xprt->cong) { 567 /* The (cwnd >> 1) term makes sure 568 * the result gets rounded properly. */ 569 cwnd += (RPC_CWNDSCALE * RPC_CWNDSCALE + (cwnd >> 1)) / cwnd; 570 if (cwnd > RPC_MAXCWND(xprt)) 571 cwnd = RPC_MAXCWND(xprt); 572 __xprt_lock_write_next_cong(xprt); 573 } else if (result == -ETIMEDOUT) { 574 cwnd >>= 1; 575 if (cwnd < RPC_CWNDSCALE) 576 cwnd = RPC_CWNDSCALE; 577 } 578 dprintk("RPC: cong %ld, cwnd was %ld, now %ld\n", 579 xprt->cong, xprt->cwnd, cwnd); 580 xprt->cwnd = cwnd; 581 __xprt_put_cong(xprt, req); 582 } 583 EXPORT_SYMBOL_GPL(xprt_adjust_cwnd); 584 585 /** 586 * xprt_wake_pending_tasks - wake all tasks on a transport's pending queue 587 * @xprt: transport with waiting tasks 588 * @status: result code to plant in each task before waking it 589 * 590 */ 591 void xprt_wake_pending_tasks(struct rpc_xprt *xprt, int status) 592 { 593 if (status < 0) 594 rpc_wake_up_status(&xprt->pending, status); 595 else 596 rpc_wake_up(&xprt->pending); 597 } 598 EXPORT_SYMBOL_GPL(xprt_wake_pending_tasks); 599 600 /** 601 * xprt_wait_for_buffer_space - wait for transport output buffer to clear 602 * @xprt: transport 603 * 604 * Note that we only set the timer for the case of RPC_IS_SOFT(), since 605 * we don't in general want to force a socket disconnection due to 606 * an incomplete RPC call transmission. 607 */ 608 void xprt_wait_for_buffer_space(struct rpc_xprt *xprt) 609 { 610 set_bit(XPRT_WRITE_SPACE, &xprt->state); 611 } 612 EXPORT_SYMBOL_GPL(xprt_wait_for_buffer_space); 613 614 static bool 615 xprt_clear_write_space_locked(struct rpc_xprt *xprt) 616 { 617 if (test_and_clear_bit(XPRT_WRITE_SPACE, &xprt->state)) { 618 __xprt_lock_write_next(xprt); 619 dprintk("RPC: write space: waking waiting task on " 620 "xprt %p\n", xprt); 621 return true; 622 } 623 return false; 624 } 625 626 /** 627 * xprt_write_space - wake the task waiting for transport output buffer space 628 * @xprt: transport with waiting tasks 629 * 630 * Can be called in a soft IRQ context, so xprt_write_space never sleeps. 631 */ 632 bool xprt_write_space(struct rpc_xprt *xprt) 633 { 634 bool ret; 635 636 if (!test_bit(XPRT_WRITE_SPACE, &xprt->state)) 637 return false; 638 spin_lock(&xprt->transport_lock); 639 ret = xprt_clear_write_space_locked(xprt); 640 spin_unlock(&xprt->transport_lock); 641 return ret; 642 } 643 EXPORT_SYMBOL_GPL(xprt_write_space); 644 645 static unsigned long xprt_abs_ktime_to_jiffies(ktime_t abstime) 646 { 647 s64 delta = ktime_to_ns(ktime_get() - abstime); 648 return likely(delta >= 0) ? 649 jiffies - nsecs_to_jiffies(delta) : 650 jiffies + nsecs_to_jiffies(-delta); 651 } 652 653 static unsigned long xprt_calc_majortimeo(struct rpc_rqst *req) 654 { 655 const struct rpc_timeout *to = req->rq_task->tk_client->cl_timeout; 656 unsigned long majortimeo = req->rq_timeout; 657 658 if (to->to_exponential) 659 majortimeo <<= to->to_retries; 660 else 661 majortimeo += to->to_increment * to->to_retries; 662 if (majortimeo > to->to_maxval || majortimeo == 0) 663 majortimeo = to->to_maxval; 664 return majortimeo; 665 } 666 667 static void xprt_reset_majortimeo(struct rpc_rqst *req) 668 { 669 req->rq_majortimeo += xprt_calc_majortimeo(req); 670 } 671 672 static void xprt_reset_minortimeo(struct rpc_rqst *req) 673 { 674 req->rq_minortimeo += req->rq_timeout; 675 } 676 677 static void xprt_init_majortimeo(struct rpc_task *task, struct rpc_rqst *req) 678 { 679 unsigned long time_init; 680 struct rpc_xprt *xprt = req->rq_xprt; 681 682 if (likely(xprt && xprt_connected(xprt))) 683 time_init = jiffies; 684 else 685 time_init = xprt_abs_ktime_to_jiffies(task->tk_start); 686 req->rq_timeout = task->tk_client->cl_timeout->to_initval; 687 req->rq_majortimeo = time_init + xprt_calc_majortimeo(req); 688 req->rq_minortimeo = time_init + req->rq_timeout; 689 } 690 691 /** 692 * xprt_adjust_timeout - adjust timeout values for next retransmit 693 * @req: RPC request containing parameters to use for the adjustment 694 * 695 */ 696 int xprt_adjust_timeout(struct rpc_rqst *req) 697 { 698 struct rpc_xprt *xprt = req->rq_xprt; 699 const struct rpc_timeout *to = req->rq_task->tk_client->cl_timeout; 700 int status = 0; 701 702 if (time_before(jiffies, req->rq_majortimeo)) { 703 if (time_before(jiffies, req->rq_minortimeo)) 704 return status; 705 if (to->to_exponential) 706 req->rq_timeout <<= 1; 707 else 708 req->rq_timeout += to->to_increment; 709 if (to->to_maxval && req->rq_timeout >= to->to_maxval) 710 req->rq_timeout = to->to_maxval; 711 req->rq_retries++; 712 } else { 713 req->rq_timeout = to->to_initval; 714 req->rq_retries = 0; 715 xprt_reset_majortimeo(req); 716 /* Reset the RTT counters == "slow start" */ 717 spin_lock(&xprt->transport_lock); 718 rpc_init_rtt(req->rq_task->tk_client->cl_rtt, to->to_initval); 719 spin_unlock(&xprt->transport_lock); 720 status = -ETIMEDOUT; 721 } 722 xprt_reset_minortimeo(req); 723 724 if (req->rq_timeout == 0) { 725 printk(KERN_WARNING "xprt_adjust_timeout: rq_timeout = 0!\n"); 726 req->rq_timeout = 5 * HZ; 727 } 728 return status; 729 } 730 731 static void xprt_autoclose(struct work_struct *work) 732 { 733 struct rpc_xprt *xprt = 734 container_of(work, struct rpc_xprt, task_cleanup); 735 unsigned int pflags = memalloc_nofs_save(); 736 737 trace_xprt_disconnect_auto(xprt); 738 xprt->connect_cookie++; 739 smp_mb__before_atomic(); 740 clear_bit(XPRT_CLOSE_WAIT, &xprt->state); 741 xprt->ops->close(xprt); 742 xprt_release_write(xprt, NULL); 743 wake_up_bit(&xprt->state, XPRT_LOCKED); 744 memalloc_nofs_restore(pflags); 745 } 746 747 /** 748 * xprt_disconnect_done - mark a transport as disconnected 749 * @xprt: transport to flag for disconnect 750 * 751 */ 752 void xprt_disconnect_done(struct rpc_xprt *xprt) 753 { 754 trace_xprt_disconnect_done(xprt); 755 spin_lock(&xprt->transport_lock); 756 xprt_clear_connected(xprt); 757 xprt_clear_write_space_locked(xprt); 758 xprt_clear_congestion_window_wait_locked(xprt); 759 xprt_wake_pending_tasks(xprt, -ENOTCONN); 760 spin_unlock(&xprt->transport_lock); 761 } 762 EXPORT_SYMBOL_GPL(xprt_disconnect_done); 763 764 /** 765 * xprt_schedule_autoclose_locked - Try to schedule an autoclose RPC call 766 * @xprt: transport to disconnect 767 */ 768 static void xprt_schedule_autoclose_locked(struct rpc_xprt *xprt) 769 { 770 if (test_and_set_bit(XPRT_CLOSE_WAIT, &xprt->state)) 771 return; 772 if (test_and_set_bit(XPRT_LOCKED, &xprt->state) == 0) 773 queue_work(xprtiod_workqueue, &xprt->task_cleanup); 774 else if (xprt->snd_task && !test_bit(XPRT_SND_IS_COOKIE, &xprt->state)) 775 rpc_wake_up_queued_task_set_status(&xprt->pending, 776 xprt->snd_task, -ENOTCONN); 777 } 778 779 /** 780 * xprt_force_disconnect - force a transport to disconnect 781 * @xprt: transport to disconnect 782 * 783 */ 784 void xprt_force_disconnect(struct rpc_xprt *xprt) 785 { 786 trace_xprt_disconnect_force(xprt); 787 788 /* Don't race with the test_bit() in xprt_clear_locked() */ 789 spin_lock(&xprt->transport_lock); 790 xprt_schedule_autoclose_locked(xprt); 791 spin_unlock(&xprt->transport_lock); 792 } 793 EXPORT_SYMBOL_GPL(xprt_force_disconnect); 794 795 static unsigned int 796 xprt_connect_cookie(struct rpc_xprt *xprt) 797 { 798 return READ_ONCE(xprt->connect_cookie); 799 } 800 801 static bool 802 xprt_request_retransmit_after_disconnect(struct rpc_task *task) 803 { 804 struct rpc_rqst *req = task->tk_rqstp; 805 struct rpc_xprt *xprt = req->rq_xprt; 806 807 return req->rq_connect_cookie != xprt_connect_cookie(xprt) || 808 !xprt_connected(xprt); 809 } 810 811 /** 812 * xprt_conditional_disconnect - force a transport to disconnect 813 * @xprt: transport to disconnect 814 * @cookie: 'connection cookie' 815 * 816 * This attempts to break the connection if and only if 'cookie' matches 817 * the current transport 'connection cookie'. It ensures that we don't 818 * try to break the connection more than once when we need to retransmit 819 * a batch of RPC requests. 820 * 821 */ 822 void xprt_conditional_disconnect(struct rpc_xprt *xprt, unsigned int cookie) 823 { 824 /* Don't race with the test_bit() in xprt_clear_locked() */ 825 spin_lock(&xprt->transport_lock); 826 if (cookie != xprt->connect_cookie) 827 goto out; 828 if (test_bit(XPRT_CLOSING, &xprt->state)) 829 goto out; 830 xprt_schedule_autoclose_locked(xprt); 831 out: 832 spin_unlock(&xprt->transport_lock); 833 } 834 835 static bool 836 xprt_has_timer(const struct rpc_xprt *xprt) 837 { 838 return xprt->idle_timeout != 0; 839 } 840 841 static void 842 xprt_schedule_autodisconnect(struct rpc_xprt *xprt) 843 __must_hold(&xprt->transport_lock) 844 { 845 xprt->last_used = jiffies; 846 if (RB_EMPTY_ROOT(&xprt->recv_queue) && xprt_has_timer(xprt)) 847 mod_timer(&xprt->timer, xprt->last_used + xprt->idle_timeout); 848 } 849 850 static void 851 xprt_init_autodisconnect(struct timer_list *t) 852 { 853 struct rpc_xprt *xprt = from_timer(xprt, t, timer); 854 855 if (!RB_EMPTY_ROOT(&xprt->recv_queue)) 856 return; 857 /* Reset xprt->last_used to avoid connect/autodisconnect cycling */ 858 xprt->last_used = jiffies; 859 if (test_and_set_bit(XPRT_LOCKED, &xprt->state)) 860 return; 861 queue_work(xprtiod_workqueue, &xprt->task_cleanup); 862 } 863 864 #if IS_ENABLED(CONFIG_FAIL_SUNRPC) 865 static void xprt_inject_disconnect(struct rpc_xprt *xprt) 866 { 867 if (!fail_sunrpc.ignore_client_disconnect && 868 should_fail(&fail_sunrpc.attr, 1)) 869 xprt->ops->inject_disconnect(xprt); 870 } 871 #else 872 static inline void xprt_inject_disconnect(struct rpc_xprt *xprt) 873 { 874 } 875 #endif 876 877 bool xprt_lock_connect(struct rpc_xprt *xprt, 878 struct rpc_task *task, 879 void *cookie) 880 { 881 bool ret = false; 882 883 spin_lock(&xprt->transport_lock); 884 if (!test_bit(XPRT_LOCKED, &xprt->state)) 885 goto out; 886 if (xprt->snd_task != task) 887 goto out; 888 set_bit(XPRT_SND_IS_COOKIE, &xprt->state); 889 xprt->snd_task = cookie; 890 ret = true; 891 out: 892 spin_unlock(&xprt->transport_lock); 893 return ret; 894 } 895 EXPORT_SYMBOL_GPL(xprt_lock_connect); 896 897 void xprt_unlock_connect(struct rpc_xprt *xprt, void *cookie) 898 { 899 spin_lock(&xprt->transport_lock); 900 if (xprt->snd_task != cookie) 901 goto out; 902 if (!test_bit(XPRT_LOCKED, &xprt->state)) 903 goto out; 904 xprt->snd_task =NULL; 905 clear_bit(XPRT_SND_IS_COOKIE, &xprt->state); 906 xprt->ops->release_xprt(xprt, NULL); 907 xprt_schedule_autodisconnect(xprt); 908 out: 909 spin_unlock(&xprt->transport_lock); 910 wake_up_bit(&xprt->state, XPRT_LOCKED); 911 } 912 EXPORT_SYMBOL_GPL(xprt_unlock_connect); 913 914 /** 915 * xprt_connect - schedule a transport connect operation 916 * @task: RPC task that is requesting the connect 917 * 918 */ 919 void xprt_connect(struct rpc_task *task) 920 { 921 struct rpc_xprt *xprt = task->tk_rqstp->rq_xprt; 922 923 trace_xprt_connect(xprt); 924 925 if (!xprt_bound(xprt)) { 926 task->tk_status = -EAGAIN; 927 return; 928 } 929 if (!xprt_lock_write(xprt, task)) 930 return; 931 932 if (test_and_clear_bit(XPRT_CLOSE_WAIT, &xprt->state)) { 933 trace_xprt_disconnect_cleanup(xprt); 934 xprt->ops->close(xprt); 935 } 936 937 if (!xprt_connected(xprt)) { 938 task->tk_rqstp->rq_connect_cookie = xprt->connect_cookie; 939 rpc_sleep_on_timeout(&xprt->pending, task, NULL, 940 xprt_request_timeout(task->tk_rqstp)); 941 942 if (test_bit(XPRT_CLOSING, &xprt->state)) 943 return; 944 if (xprt_test_and_set_connecting(xprt)) 945 return; 946 /* Race breaker */ 947 if (!xprt_connected(xprt)) { 948 xprt->stat.connect_start = jiffies; 949 xprt->ops->connect(xprt, task); 950 } else { 951 xprt_clear_connecting(xprt); 952 task->tk_status = 0; 953 rpc_wake_up_queued_task(&xprt->pending, task); 954 } 955 } 956 xprt_release_write(xprt, task); 957 } 958 959 /** 960 * xprt_reconnect_delay - compute the wait before scheduling a connect 961 * @xprt: transport instance 962 * 963 */ 964 unsigned long xprt_reconnect_delay(const struct rpc_xprt *xprt) 965 { 966 unsigned long start, now = jiffies; 967 968 start = xprt->stat.connect_start + xprt->reestablish_timeout; 969 if (time_after(start, now)) 970 return start - now; 971 return 0; 972 } 973 EXPORT_SYMBOL_GPL(xprt_reconnect_delay); 974 975 /** 976 * xprt_reconnect_backoff - compute the new re-establish timeout 977 * @xprt: transport instance 978 * @init_to: initial reestablish timeout 979 * 980 */ 981 void xprt_reconnect_backoff(struct rpc_xprt *xprt, unsigned long init_to) 982 { 983 xprt->reestablish_timeout <<= 1; 984 if (xprt->reestablish_timeout > xprt->max_reconnect_timeout) 985 xprt->reestablish_timeout = xprt->max_reconnect_timeout; 986 if (xprt->reestablish_timeout < init_to) 987 xprt->reestablish_timeout = init_to; 988 } 989 EXPORT_SYMBOL_GPL(xprt_reconnect_backoff); 990 991 enum xprt_xid_rb_cmp { 992 XID_RB_EQUAL, 993 XID_RB_LEFT, 994 XID_RB_RIGHT, 995 }; 996 static enum xprt_xid_rb_cmp 997 xprt_xid_cmp(__be32 xid1, __be32 xid2) 998 { 999 if (xid1 == xid2) 1000 return XID_RB_EQUAL; 1001 if ((__force u32)xid1 < (__force u32)xid2) 1002 return XID_RB_LEFT; 1003 return XID_RB_RIGHT; 1004 } 1005 1006 static struct rpc_rqst * 1007 xprt_request_rb_find(struct rpc_xprt *xprt, __be32 xid) 1008 { 1009 struct rb_node *n = xprt->recv_queue.rb_node; 1010 struct rpc_rqst *req; 1011 1012 while (n != NULL) { 1013 req = rb_entry(n, struct rpc_rqst, rq_recv); 1014 switch (xprt_xid_cmp(xid, req->rq_xid)) { 1015 case XID_RB_LEFT: 1016 n = n->rb_left; 1017 break; 1018 case XID_RB_RIGHT: 1019 n = n->rb_right; 1020 break; 1021 case XID_RB_EQUAL: 1022 return req; 1023 } 1024 } 1025 return NULL; 1026 } 1027 1028 static void 1029 xprt_request_rb_insert(struct rpc_xprt *xprt, struct rpc_rqst *new) 1030 { 1031 struct rb_node **p = &xprt->recv_queue.rb_node; 1032 struct rb_node *n = NULL; 1033 struct rpc_rqst *req; 1034 1035 while (*p != NULL) { 1036 n = *p; 1037 req = rb_entry(n, struct rpc_rqst, rq_recv); 1038 switch(xprt_xid_cmp(new->rq_xid, req->rq_xid)) { 1039 case XID_RB_LEFT: 1040 p = &n->rb_left; 1041 break; 1042 case XID_RB_RIGHT: 1043 p = &n->rb_right; 1044 break; 1045 case XID_RB_EQUAL: 1046 WARN_ON_ONCE(new != req); 1047 return; 1048 } 1049 } 1050 rb_link_node(&new->rq_recv, n, p); 1051 rb_insert_color(&new->rq_recv, &xprt->recv_queue); 1052 } 1053 1054 static void 1055 xprt_request_rb_remove(struct rpc_xprt *xprt, struct rpc_rqst *req) 1056 { 1057 rb_erase(&req->rq_recv, &xprt->recv_queue); 1058 } 1059 1060 /** 1061 * xprt_lookup_rqst - find an RPC request corresponding to an XID 1062 * @xprt: transport on which the original request was transmitted 1063 * @xid: RPC XID of incoming reply 1064 * 1065 * Caller holds xprt->queue_lock. 1066 */ 1067 struct rpc_rqst *xprt_lookup_rqst(struct rpc_xprt *xprt, __be32 xid) 1068 { 1069 struct rpc_rqst *entry; 1070 1071 entry = xprt_request_rb_find(xprt, xid); 1072 if (entry != NULL) { 1073 trace_xprt_lookup_rqst(xprt, xid, 0); 1074 entry->rq_rtt = ktime_sub(ktime_get(), entry->rq_xtime); 1075 return entry; 1076 } 1077 1078 dprintk("RPC: xprt_lookup_rqst did not find xid %08x\n", 1079 ntohl(xid)); 1080 trace_xprt_lookup_rqst(xprt, xid, -ENOENT); 1081 xprt->stat.bad_xids++; 1082 return NULL; 1083 } 1084 EXPORT_SYMBOL_GPL(xprt_lookup_rqst); 1085 1086 static bool 1087 xprt_is_pinned_rqst(struct rpc_rqst *req) 1088 { 1089 return atomic_read(&req->rq_pin) != 0; 1090 } 1091 1092 /** 1093 * xprt_pin_rqst - Pin a request on the transport receive list 1094 * @req: Request to pin 1095 * 1096 * Caller must ensure this is atomic with the call to xprt_lookup_rqst() 1097 * so should be holding xprt->queue_lock. 1098 */ 1099 void xprt_pin_rqst(struct rpc_rqst *req) 1100 { 1101 atomic_inc(&req->rq_pin); 1102 } 1103 EXPORT_SYMBOL_GPL(xprt_pin_rqst); 1104 1105 /** 1106 * xprt_unpin_rqst - Unpin a request on the transport receive list 1107 * @req: Request to pin 1108 * 1109 * Caller should be holding xprt->queue_lock. 1110 */ 1111 void xprt_unpin_rqst(struct rpc_rqst *req) 1112 { 1113 if (!test_bit(RPC_TASK_MSG_PIN_WAIT, &req->rq_task->tk_runstate)) { 1114 atomic_dec(&req->rq_pin); 1115 return; 1116 } 1117 if (atomic_dec_and_test(&req->rq_pin)) 1118 wake_up_var(&req->rq_pin); 1119 } 1120 EXPORT_SYMBOL_GPL(xprt_unpin_rqst); 1121 1122 static void xprt_wait_on_pinned_rqst(struct rpc_rqst *req) 1123 { 1124 wait_var_event(&req->rq_pin, !xprt_is_pinned_rqst(req)); 1125 } 1126 1127 static bool 1128 xprt_request_data_received(struct rpc_task *task) 1129 { 1130 return !test_bit(RPC_TASK_NEED_RECV, &task->tk_runstate) && 1131 READ_ONCE(task->tk_rqstp->rq_reply_bytes_recvd) != 0; 1132 } 1133 1134 static bool 1135 xprt_request_need_enqueue_receive(struct rpc_task *task, struct rpc_rqst *req) 1136 { 1137 return !test_bit(RPC_TASK_NEED_RECV, &task->tk_runstate) && 1138 READ_ONCE(task->tk_rqstp->rq_reply_bytes_recvd) == 0; 1139 } 1140 1141 /** 1142 * xprt_request_enqueue_receive - Add an request to the receive queue 1143 * @task: RPC task 1144 * 1145 */ 1146 void 1147 xprt_request_enqueue_receive(struct rpc_task *task) 1148 { 1149 struct rpc_rqst *req = task->tk_rqstp; 1150 struct rpc_xprt *xprt = req->rq_xprt; 1151 1152 if (!xprt_request_need_enqueue_receive(task, req)) 1153 return; 1154 1155 xprt_request_prepare(task->tk_rqstp); 1156 spin_lock(&xprt->queue_lock); 1157 1158 /* Update the softirq receive buffer */ 1159 memcpy(&req->rq_private_buf, &req->rq_rcv_buf, 1160 sizeof(req->rq_private_buf)); 1161 1162 /* Add request to the receive list */ 1163 xprt_request_rb_insert(xprt, req); 1164 set_bit(RPC_TASK_NEED_RECV, &task->tk_runstate); 1165 spin_unlock(&xprt->queue_lock); 1166 1167 /* Turn off autodisconnect */ 1168 del_singleshot_timer_sync(&xprt->timer); 1169 } 1170 1171 /** 1172 * xprt_request_dequeue_receive_locked - Remove a request from the receive queue 1173 * @task: RPC task 1174 * 1175 * Caller must hold xprt->queue_lock. 1176 */ 1177 static void 1178 xprt_request_dequeue_receive_locked(struct rpc_task *task) 1179 { 1180 struct rpc_rqst *req = task->tk_rqstp; 1181 1182 if (test_and_clear_bit(RPC_TASK_NEED_RECV, &task->tk_runstate)) 1183 xprt_request_rb_remove(req->rq_xprt, req); 1184 } 1185 1186 /** 1187 * xprt_update_rtt - Update RPC RTT statistics 1188 * @task: RPC request that recently completed 1189 * 1190 * Caller holds xprt->queue_lock. 1191 */ 1192 void xprt_update_rtt(struct rpc_task *task) 1193 { 1194 struct rpc_rqst *req = task->tk_rqstp; 1195 struct rpc_rtt *rtt = task->tk_client->cl_rtt; 1196 unsigned int timer = task->tk_msg.rpc_proc->p_timer; 1197 long m = usecs_to_jiffies(ktime_to_us(req->rq_rtt)); 1198 1199 if (timer) { 1200 if (req->rq_ntrans == 1) 1201 rpc_update_rtt(rtt, timer, m); 1202 rpc_set_timeo(rtt, timer, req->rq_ntrans - 1); 1203 } 1204 } 1205 EXPORT_SYMBOL_GPL(xprt_update_rtt); 1206 1207 /** 1208 * xprt_complete_rqst - called when reply processing is complete 1209 * @task: RPC request that recently completed 1210 * @copied: actual number of bytes received from the transport 1211 * 1212 * Caller holds xprt->queue_lock. 1213 */ 1214 void xprt_complete_rqst(struct rpc_task *task, int copied) 1215 { 1216 struct rpc_rqst *req = task->tk_rqstp; 1217 struct rpc_xprt *xprt = req->rq_xprt; 1218 1219 xprt->stat.recvs++; 1220 1221 req->rq_private_buf.len = copied; 1222 /* Ensure all writes are done before we update */ 1223 /* req->rq_reply_bytes_recvd */ 1224 smp_wmb(); 1225 req->rq_reply_bytes_recvd = copied; 1226 xprt_request_dequeue_receive_locked(task); 1227 rpc_wake_up_queued_task(&xprt->pending, task); 1228 } 1229 EXPORT_SYMBOL_GPL(xprt_complete_rqst); 1230 1231 static void xprt_timer(struct rpc_task *task) 1232 { 1233 struct rpc_rqst *req = task->tk_rqstp; 1234 struct rpc_xprt *xprt = req->rq_xprt; 1235 1236 if (task->tk_status != -ETIMEDOUT) 1237 return; 1238 1239 trace_xprt_timer(xprt, req->rq_xid, task->tk_status); 1240 if (!req->rq_reply_bytes_recvd) { 1241 if (xprt->ops->timer) 1242 xprt->ops->timer(xprt, task); 1243 } else 1244 task->tk_status = 0; 1245 } 1246 1247 /** 1248 * xprt_wait_for_reply_request_def - wait for reply 1249 * @task: pointer to rpc_task 1250 * 1251 * Set a request's retransmit timeout based on the transport's 1252 * default timeout parameters. Used by transports that don't adjust 1253 * the retransmit timeout based on round-trip time estimation, 1254 * and put the task to sleep on the pending queue. 1255 */ 1256 void xprt_wait_for_reply_request_def(struct rpc_task *task) 1257 { 1258 struct rpc_rqst *req = task->tk_rqstp; 1259 1260 rpc_sleep_on_timeout(&req->rq_xprt->pending, task, xprt_timer, 1261 xprt_request_timeout(req)); 1262 } 1263 EXPORT_SYMBOL_GPL(xprt_wait_for_reply_request_def); 1264 1265 /** 1266 * xprt_wait_for_reply_request_rtt - wait for reply using RTT estimator 1267 * @task: pointer to rpc_task 1268 * 1269 * Set a request's retransmit timeout using the RTT estimator, 1270 * and put the task to sleep on the pending queue. 1271 */ 1272 void xprt_wait_for_reply_request_rtt(struct rpc_task *task) 1273 { 1274 int timer = task->tk_msg.rpc_proc->p_timer; 1275 struct rpc_clnt *clnt = task->tk_client; 1276 struct rpc_rtt *rtt = clnt->cl_rtt; 1277 struct rpc_rqst *req = task->tk_rqstp; 1278 unsigned long max_timeout = clnt->cl_timeout->to_maxval; 1279 unsigned long timeout; 1280 1281 timeout = rpc_calc_rto(rtt, timer); 1282 timeout <<= rpc_ntimeo(rtt, timer) + req->rq_retries; 1283 if (timeout > max_timeout || timeout == 0) 1284 timeout = max_timeout; 1285 rpc_sleep_on_timeout(&req->rq_xprt->pending, task, xprt_timer, 1286 jiffies + timeout); 1287 } 1288 EXPORT_SYMBOL_GPL(xprt_wait_for_reply_request_rtt); 1289 1290 /** 1291 * xprt_request_wait_receive - wait for the reply to an RPC request 1292 * @task: RPC task about to send a request 1293 * 1294 */ 1295 void xprt_request_wait_receive(struct rpc_task *task) 1296 { 1297 struct rpc_rqst *req = task->tk_rqstp; 1298 struct rpc_xprt *xprt = req->rq_xprt; 1299 1300 if (!test_bit(RPC_TASK_NEED_RECV, &task->tk_runstate)) 1301 return; 1302 /* 1303 * Sleep on the pending queue if we're expecting a reply. 1304 * The spinlock ensures atomicity between the test of 1305 * req->rq_reply_bytes_recvd, and the call to rpc_sleep_on(). 1306 */ 1307 spin_lock(&xprt->queue_lock); 1308 if (test_bit(RPC_TASK_NEED_RECV, &task->tk_runstate)) { 1309 xprt->ops->wait_for_reply_request(task); 1310 /* 1311 * Send an extra queue wakeup call if the 1312 * connection was dropped in case the call to 1313 * rpc_sleep_on() raced. 1314 */ 1315 if (xprt_request_retransmit_after_disconnect(task)) 1316 rpc_wake_up_queued_task_set_status(&xprt->pending, 1317 task, -ENOTCONN); 1318 } 1319 spin_unlock(&xprt->queue_lock); 1320 } 1321 1322 static bool 1323 xprt_request_need_enqueue_transmit(struct rpc_task *task, struct rpc_rqst *req) 1324 { 1325 return !test_bit(RPC_TASK_NEED_XMIT, &task->tk_runstate); 1326 } 1327 1328 /** 1329 * xprt_request_enqueue_transmit - queue a task for transmission 1330 * @task: pointer to rpc_task 1331 * 1332 * Add a task to the transmission queue. 1333 */ 1334 void 1335 xprt_request_enqueue_transmit(struct rpc_task *task) 1336 { 1337 struct rpc_rqst *pos, *req = task->tk_rqstp; 1338 struct rpc_xprt *xprt = req->rq_xprt; 1339 1340 if (xprt_request_need_enqueue_transmit(task, req)) { 1341 req->rq_bytes_sent = 0; 1342 spin_lock(&xprt->queue_lock); 1343 /* 1344 * Requests that carry congestion control credits are added 1345 * to the head of the list to avoid starvation issues. 1346 */ 1347 if (req->rq_cong) { 1348 xprt_clear_congestion_window_wait(xprt); 1349 list_for_each_entry(pos, &xprt->xmit_queue, rq_xmit) { 1350 if (pos->rq_cong) 1351 continue; 1352 /* Note: req is added _before_ pos */ 1353 list_add_tail(&req->rq_xmit, &pos->rq_xmit); 1354 INIT_LIST_HEAD(&req->rq_xmit2); 1355 goto out; 1356 } 1357 } else if (!req->rq_seqno) { 1358 list_for_each_entry(pos, &xprt->xmit_queue, rq_xmit) { 1359 if (pos->rq_task->tk_owner != task->tk_owner) 1360 continue; 1361 list_add_tail(&req->rq_xmit2, &pos->rq_xmit2); 1362 INIT_LIST_HEAD(&req->rq_xmit); 1363 goto out; 1364 } 1365 } 1366 list_add_tail(&req->rq_xmit, &xprt->xmit_queue); 1367 INIT_LIST_HEAD(&req->rq_xmit2); 1368 out: 1369 atomic_long_inc(&xprt->xmit_queuelen); 1370 set_bit(RPC_TASK_NEED_XMIT, &task->tk_runstate); 1371 spin_unlock(&xprt->queue_lock); 1372 } 1373 } 1374 1375 /** 1376 * xprt_request_dequeue_transmit_locked - remove a task from the transmission queue 1377 * @task: pointer to rpc_task 1378 * 1379 * Remove a task from the transmission queue 1380 * Caller must hold xprt->queue_lock 1381 */ 1382 static void 1383 xprt_request_dequeue_transmit_locked(struct rpc_task *task) 1384 { 1385 struct rpc_rqst *req = task->tk_rqstp; 1386 1387 if (!test_and_clear_bit(RPC_TASK_NEED_XMIT, &task->tk_runstate)) 1388 return; 1389 if (!list_empty(&req->rq_xmit)) { 1390 list_del(&req->rq_xmit); 1391 if (!list_empty(&req->rq_xmit2)) { 1392 struct rpc_rqst *next = list_first_entry(&req->rq_xmit2, 1393 struct rpc_rqst, rq_xmit2); 1394 list_del(&req->rq_xmit2); 1395 list_add_tail(&next->rq_xmit, &next->rq_xprt->xmit_queue); 1396 } 1397 } else 1398 list_del(&req->rq_xmit2); 1399 atomic_long_dec(&req->rq_xprt->xmit_queuelen); 1400 } 1401 1402 /** 1403 * xprt_request_dequeue_transmit - remove a task from the transmission queue 1404 * @task: pointer to rpc_task 1405 * 1406 * Remove a task from the transmission queue 1407 */ 1408 static void 1409 xprt_request_dequeue_transmit(struct rpc_task *task) 1410 { 1411 struct rpc_rqst *req = task->tk_rqstp; 1412 struct rpc_xprt *xprt = req->rq_xprt; 1413 1414 spin_lock(&xprt->queue_lock); 1415 xprt_request_dequeue_transmit_locked(task); 1416 spin_unlock(&xprt->queue_lock); 1417 } 1418 1419 /** 1420 * xprt_request_dequeue_xprt - remove a task from the transmit+receive queue 1421 * @task: pointer to rpc_task 1422 * 1423 * Remove a task from the transmit and receive queues, and ensure that 1424 * it is not pinned by the receive work item. 1425 */ 1426 void 1427 xprt_request_dequeue_xprt(struct rpc_task *task) 1428 { 1429 struct rpc_rqst *req = task->tk_rqstp; 1430 struct rpc_xprt *xprt = req->rq_xprt; 1431 1432 if (test_bit(RPC_TASK_NEED_XMIT, &task->tk_runstate) || 1433 test_bit(RPC_TASK_NEED_RECV, &task->tk_runstate) || 1434 xprt_is_pinned_rqst(req)) { 1435 spin_lock(&xprt->queue_lock); 1436 xprt_request_dequeue_transmit_locked(task); 1437 xprt_request_dequeue_receive_locked(task); 1438 while (xprt_is_pinned_rqst(req)) { 1439 set_bit(RPC_TASK_MSG_PIN_WAIT, &task->tk_runstate); 1440 spin_unlock(&xprt->queue_lock); 1441 xprt_wait_on_pinned_rqst(req); 1442 spin_lock(&xprt->queue_lock); 1443 clear_bit(RPC_TASK_MSG_PIN_WAIT, &task->tk_runstate); 1444 } 1445 spin_unlock(&xprt->queue_lock); 1446 } 1447 } 1448 1449 /** 1450 * xprt_request_prepare - prepare an encoded request for transport 1451 * @req: pointer to rpc_rqst 1452 * 1453 * Calls into the transport layer to do whatever is needed to prepare 1454 * the request for transmission or receive. 1455 */ 1456 void 1457 xprt_request_prepare(struct rpc_rqst *req) 1458 { 1459 struct rpc_xprt *xprt = req->rq_xprt; 1460 1461 if (xprt->ops->prepare_request) 1462 xprt->ops->prepare_request(req); 1463 } 1464 1465 /** 1466 * xprt_request_need_retransmit - Test if a task needs retransmission 1467 * @task: pointer to rpc_task 1468 * 1469 * Test for whether a connection breakage requires the task to retransmit 1470 */ 1471 bool 1472 xprt_request_need_retransmit(struct rpc_task *task) 1473 { 1474 return xprt_request_retransmit_after_disconnect(task); 1475 } 1476 1477 /** 1478 * xprt_prepare_transmit - reserve the transport before sending a request 1479 * @task: RPC task about to send a request 1480 * 1481 */ 1482 bool xprt_prepare_transmit(struct rpc_task *task) 1483 { 1484 struct rpc_rqst *req = task->tk_rqstp; 1485 struct rpc_xprt *xprt = req->rq_xprt; 1486 1487 if (!xprt_lock_write(xprt, task)) { 1488 /* Race breaker: someone may have transmitted us */ 1489 if (!test_bit(RPC_TASK_NEED_XMIT, &task->tk_runstate)) 1490 rpc_wake_up_queued_task_set_status(&xprt->sending, 1491 task, 0); 1492 return false; 1493 1494 } 1495 if (atomic_read(&xprt->swapper)) 1496 /* This will be clear in __rpc_execute */ 1497 current->flags |= PF_MEMALLOC; 1498 return true; 1499 } 1500 1501 void xprt_end_transmit(struct rpc_task *task) 1502 { 1503 struct rpc_xprt *xprt = task->tk_rqstp->rq_xprt; 1504 1505 xprt_inject_disconnect(xprt); 1506 xprt_release_write(xprt, task); 1507 } 1508 1509 /** 1510 * xprt_request_transmit - send an RPC request on a transport 1511 * @req: pointer to request to transmit 1512 * @snd_task: RPC task that owns the transport lock 1513 * 1514 * This performs the transmission of a single request. 1515 * Note that if the request is not the same as snd_task, then it 1516 * does need to be pinned. 1517 * Returns '0' on success. 1518 */ 1519 static int 1520 xprt_request_transmit(struct rpc_rqst *req, struct rpc_task *snd_task) 1521 { 1522 struct rpc_xprt *xprt = req->rq_xprt; 1523 struct rpc_task *task = req->rq_task; 1524 unsigned int connect_cookie; 1525 int is_retrans = RPC_WAS_SENT(task); 1526 int status; 1527 1528 if (!req->rq_bytes_sent) { 1529 if (xprt_request_data_received(task)) { 1530 status = 0; 1531 goto out_dequeue; 1532 } 1533 /* Verify that our message lies in the RPCSEC_GSS window */ 1534 if (rpcauth_xmit_need_reencode(task)) { 1535 status = -EBADMSG; 1536 goto out_dequeue; 1537 } 1538 if (RPC_SIGNALLED(task)) { 1539 status = -ERESTARTSYS; 1540 goto out_dequeue; 1541 } 1542 } 1543 1544 /* 1545 * Update req->rq_ntrans before transmitting to avoid races with 1546 * xprt_update_rtt(), which needs to know that it is recording a 1547 * reply to the first transmission. 1548 */ 1549 req->rq_ntrans++; 1550 1551 trace_rpc_xdr_sendto(task, &req->rq_snd_buf); 1552 connect_cookie = xprt->connect_cookie; 1553 status = xprt->ops->send_request(req); 1554 if (status != 0) { 1555 req->rq_ntrans--; 1556 trace_xprt_transmit(req, status); 1557 return status; 1558 } 1559 1560 if (is_retrans) { 1561 task->tk_client->cl_stats->rpcretrans++; 1562 trace_xprt_retransmit(req); 1563 } 1564 1565 xprt_inject_disconnect(xprt); 1566 1567 task->tk_flags |= RPC_TASK_SENT; 1568 spin_lock(&xprt->transport_lock); 1569 1570 xprt->stat.sends++; 1571 xprt->stat.req_u += xprt->stat.sends - xprt->stat.recvs; 1572 xprt->stat.bklog_u += xprt->backlog.qlen; 1573 xprt->stat.sending_u += xprt->sending.qlen; 1574 xprt->stat.pending_u += xprt->pending.qlen; 1575 spin_unlock(&xprt->transport_lock); 1576 1577 req->rq_connect_cookie = connect_cookie; 1578 out_dequeue: 1579 trace_xprt_transmit(req, status); 1580 xprt_request_dequeue_transmit(task); 1581 rpc_wake_up_queued_task_set_status(&xprt->sending, task, status); 1582 return status; 1583 } 1584 1585 /** 1586 * xprt_transmit - send an RPC request on a transport 1587 * @task: controlling RPC task 1588 * 1589 * Attempts to drain the transmit queue. On exit, either the transport 1590 * signalled an error that needs to be handled before transmission can 1591 * resume, or @task finished transmitting, and detected that it already 1592 * received a reply. 1593 */ 1594 void 1595 xprt_transmit(struct rpc_task *task) 1596 { 1597 struct rpc_rqst *next, *req = task->tk_rqstp; 1598 struct rpc_xprt *xprt = req->rq_xprt; 1599 int status; 1600 1601 spin_lock(&xprt->queue_lock); 1602 for (;;) { 1603 next = list_first_entry_or_null(&xprt->xmit_queue, 1604 struct rpc_rqst, rq_xmit); 1605 if (!next) 1606 break; 1607 xprt_pin_rqst(next); 1608 spin_unlock(&xprt->queue_lock); 1609 status = xprt_request_transmit(next, task); 1610 if (status == -EBADMSG && next != req) 1611 status = 0; 1612 spin_lock(&xprt->queue_lock); 1613 xprt_unpin_rqst(next); 1614 if (status < 0) { 1615 if (test_bit(RPC_TASK_NEED_XMIT, &task->tk_runstate)) 1616 task->tk_status = status; 1617 break; 1618 } 1619 /* Was @task transmitted, and has it received a reply? */ 1620 if (xprt_request_data_received(task) && 1621 !test_bit(RPC_TASK_NEED_XMIT, &task->tk_runstate)) 1622 break; 1623 cond_resched_lock(&xprt->queue_lock); 1624 } 1625 spin_unlock(&xprt->queue_lock); 1626 } 1627 1628 static void xprt_complete_request_init(struct rpc_task *task) 1629 { 1630 if (task->tk_rqstp) 1631 xprt_request_init(task); 1632 } 1633 1634 void xprt_add_backlog(struct rpc_xprt *xprt, struct rpc_task *task) 1635 { 1636 set_bit(XPRT_CONGESTED, &xprt->state); 1637 rpc_sleep_on(&xprt->backlog, task, xprt_complete_request_init); 1638 } 1639 EXPORT_SYMBOL_GPL(xprt_add_backlog); 1640 1641 static bool __xprt_set_rq(struct rpc_task *task, void *data) 1642 { 1643 struct rpc_rqst *req = data; 1644 1645 if (task->tk_rqstp == NULL) { 1646 memset(req, 0, sizeof(*req)); /* mark unused */ 1647 task->tk_rqstp = req; 1648 return true; 1649 } 1650 return false; 1651 } 1652 1653 bool xprt_wake_up_backlog(struct rpc_xprt *xprt, struct rpc_rqst *req) 1654 { 1655 if (rpc_wake_up_first(&xprt->backlog, __xprt_set_rq, req) == NULL) { 1656 clear_bit(XPRT_CONGESTED, &xprt->state); 1657 return false; 1658 } 1659 return true; 1660 } 1661 EXPORT_SYMBOL_GPL(xprt_wake_up_backlog); 1662 1663 static bool xprt_throttle_congested(struct rpc_xprt *xprt, struct rpc_task *task) 1664 { 1665 bool ret = false; 1666 1667 if (!test_bit(XPRT_CONGESTED, &xprt->state)) 1668 goto out; 1669 spin_lock(&xprt->reserve_lock); 1670 if (test_bit(XPRT_CONGESTED, &xprt->state)) { 1671 xprt_add_backlog(xprt, task); 1672 ret = true; 1673 } 1674 spin_unlock(&xprt->reserve_lock); 1675 out: 1676 return ret; 1677 } 1678 1679 static struct rpc_rqst *xprt_dynamic_alloc_slot(struct rpc_xprt *xprt) 1680 { 1681 struct rpc_rqst *req = ERR_PTR(-EAGAIN); 1682 1683 if (xprt->num_reqs >= xprt->max_reqs) 1684 goto out; 1685 ++xprt->num_reqs; 1686 spin_unlock(&xprt->reserve_lock); 1687 req = kzalloc(sizeof(*req), rpc_task_gfp_mask()); 1688 spin_lock(&xprt->reserve_lock); 1689 if (req != NULL) 1690 goto out; 1691 --xprt->num_reqs; 1692 req = ERR_PTR(-ENOMEM); 1693 out: 1694 return req; 1695 } 1696 1697 static bool xprt_dynamic_free_slot(struct rpc_xprt *xprt, struct rpc_rqst *req) 1698 { 1699 if (xprt->num_reqs > xprt->min_reqs) { 1700 --xprt->num_reqs; 1701 kfree(req); 1702 return true; 1703 } 1704 return false; 1705 } 1706 1707 void xprt_alloc_slot(struct rpc_xprt *xprt, struct rpc_task *task) 1708 { 1709 struct rpc_rqst *req; 1710 1711 spin_lock(&xprt->reserve_lock); 1712 if (!list_empty(&xprt->free)) { 1713 req = list_entry(xprt->free.next, struct rpc_rqst, rq_list); 1714 list_del(&req->rq_list); 1715 goto out_init_req; 1716 } 1717 req = xprt_dynamic_alloc_slot(xprt); 1718 if (!IS_ERR(req)) 1719 goto out_init_req; 1720 switch (PTR_ERR(req)) { 1721 case -ENOMEM: 1722 dprintk("RPC: dynamic allocation of request slot " 1723 "failed! Retrying\n"); 1724 task->tk_status = -ENOMEM; 1725 break; 1726 case -EAGAIN: 1727 xprt_add_backlog(xprt, task); 1728 dprintk("RPC: waiting for request slot\n"); 1729 fallthrough; 1730 default: 1731 task->tk_status = -EAGAIN; 1732 } 1733 spin_unlock(&xprt->reserve_lock); 1734 return; 1735 out_init_req: 1736 xprt->stat.max_slots = max_t(unsigned int, xprt->stat.max_slots, 1737 xprt->num_reqs); 1738 spin_unlock(&xprt->reserve_lock); 1739 1740 task->tk_status = 0; 1741 task->tk_rqstp = req; 1742 } 1743 EXPORT_SYMBOL_GPL(xprt_alloc_slot); 1744 1745 void xprt_free_slot(struct rpc_xprt *xprt, struct rpc_rqst *req) 1746 { 1747 spin_lock(&xprt->reserve_lock); 1748 if (!xprt_wake_up_backlog(xprt, req) && 1749 !xprt_dynamic_free_slot(xprt, req)) { 1750 memset(req, 0, sizeof(*req)); /* mark unused */ 1751 list_add(&req->rq_list, &xprt->free); 1752 } 1753 spin_unlock(&xprt->reserve_lock); 1754 } 1755 EXPORT_SYMBOL_GPL(xprt_free_slot); 1756 1757 static void xprt_free_all_slots(struct rpc_xprt *xprt) 1758 { 1759 struct rpc_rqst *req; 1760 while (!list_empty(&xprt->free)) { 1761 req = list_first_entry(&xprt->free, struct rpc_rqst, rq_list); 1762 list_del(&req->rq_list); 1763 kfree(req); 1764 } 1765 } 1766 1767 static DEFINE_IDA(rpc_xprt_ids); 1768 1769 void xprt_cleanup_ids(void) 1770 { 1771 ida_destroy(&rpc_xprt_ids); 1772 } 1773 1774 static int xprt_alloc_id(struct rpc_xprt *xprt) 1775 { 1776 int id; 1777 1778 id = ida_simple_get(&rpc_xprt_ids, 0, 0, GFP_KERNEL); 1779 if (id < 0) 1780 return id; 1781 1782 xprt->id = id; 1783 return 0; 1784 } 1785 1786 static void xprt_free_id(struct rpc_xprt *xprt) 1787 { 1788 ida_simple_remove(&rpc_xprt_ids, xprt->id); 1789 } 1790 1791 struct rpc_xprt *xprt_alloc(struct net *net, size_t size, 1792 unsigned int num_prealloc, 1793 unsigned int max_alloc) 1794 { 1795 struct rpc_xprt *xprt; 1796 struct rpc_rqst *req; 1797 int i; 1798 1799 xprt = kzalloc(size, GFP_KERNEL); 1800 if (xprt == NULL) 1801 goto out; 1802 1803 xprt_alloc_id(xprt); 1804 xprt_init(xprt, net); 1805 1806 for (i = 0; i < num_prealloc; i++) { 1807 req = kzalloc(sizeof(struct rpc_rqst), GFP_KERNEL); 1808 if (!req) 1809 goto out_free; 1810 list_add(&req->rq_list, &xprt->free); 1811 } 1812 if (max_alloc > num_prealloc) 1813 xprt->max_reqs = max_alloc; 1814 else 1815 xprt->max_reqs = num_prealloc; 1816 xprt->min_reqs = num_prealloc; 1817 xprt->num_reqs = num_prealloc; 1818 1819 return xprt; 1820 1821 out_free: 1822 xprt_free(xprt); 1823 out: 1824 return NULL; 1825 } 1826 EXPORT_SYMBOL_GPL(xprt_alloc); 1827 1828 void xprt_free(struct rpc_xprt *xprt) 1829 { 1830 put_net_track(xprt->xprt_net, &xprt->ns_tracker); 1831 xprt_free_all_slots(xprt); 1832 xprt_free_id(xprt); 1833 rpc_sysfs_xprt_destroy(xprt); 1834 kfree_rcu(xprt, rcu); 1835 } 1836 EXPORT_SYMBOL_GPL(xprt_free); 1837 1838 static void 1839 xprt_init_connect_cookie(struct rpc_rqst *req, struct rpc_xprt *xprt) 1840 { 1841 req->rq_connect_cookie = xprt_connect_cookie(xprt) - 1; 1842 } 1843 1844 static __be32 1845 xprt_alloc_xid(struct rpc_xprt *xprt) 1846 { 1847 __be32 xid; 1848 1849 spin_lock(&xprt->reserve_lock); 1850 xid = (__force __be32)xprt->xid++; 1851 spin_unlock(&xprt->reserve_lock); 1852 return xid; 1853 } 1854 1855 static void 1856 xprt_init_xid(struct rpc_xprt *xprt) 1857 { 1858 xprt->xid = prandom_u32(); 1859 } 1860 1861 static void 1862 xprt_request_init(struct rpc_task *task) 1863 { 1864 struct rpc_xprt *xprt = task->tk_xprt; 1865 struct rpc_rqst *req = task->tk_rqstp; 1866 1867 req->rq_task = task; 1868 req->rq_xprt = xprt; 1869 req->rq_buffer = NULL; 1870 req->rq_xid = xprt_alloc_xid(xprt); 1871 xprt_init_connect_cookie(req, xprt); 1872 req->rq_snd_buf.len = 0; 1873 req->rq_snd_buf.buflen = 0; 1874 req->rq_rcv_buf.len = 0; 1875 req->rq_rcv_buf.buflen = 0; 1876 req->rq_snd_buf.bvec = NULL; 1877 req->rq_rcv_buf.bvec = NULL; 1878 req->rq_release_snd_buf = NULL; 1879 xprt_init_majortimeo(task, req); 1880 1881 trace_xprt_reserve(req); 1882 } 1883 1884 static void 1885 xprt_do_reserve(struct rpc_xprt *xprt, struct rpc_task *task) 1886 { 1887 xprt->ops->alloc_slot(xprt, task); 1888 if (task->tk_rqstp != NULL) 1889 xprt_request_init(task); 1890 } 1891 1892 /** 1893 * xprt_reserve - allocate an RPC request slot 1894 * @task: RPC task requesting a slot allocation 1895 * 1896 * If the transport is marked as being congested, or if no more 1897 * slots are available, place the task on the transport's 1898 * backlog queue. 1899 */ 1900 void xprt_reserve(struct rpc_task *task) 1901 { 1902 struct rpc_xprt *xprt = task->tk_xprt; 1903 1904 task->tk_status = 0; 1905 if (task->tk_rqstp != NULL) 1906 return; 1907 1908 task->tk_status = -EAGAIN; 1909 if (!xprt_throttle_congested(xprt, task)) 1910 xprt_do_reserve(xprt, task); 1911 } 1912 1913 /** 1914 * xprt_retry_reserve - allocate an RPC request slot 1915 * @task: RPC task requesting a slot allocation 1916 * 1917 * If no more slots are available, place the task on the transport's 1918 * backlog queue. 1919 * Note that the only difference with xprt_reserve is that we now 1920 * ignore the value of the XPRT_CONGESTED flag. 1921 */ 1922 void xprt_retry_reserve(struct rpc_task *task) 1923 { 1924 struct rpc_xprt *xprt = task->tk_xprt; 1925 1926 task->tk_status = 0; 1927 if (task->tk_rqstp != NULL) 1928 return; 1929 1930 task->tk_status = -EAGAIN; 1931 xprt_do_reserve(xprt, task); 1932 } 1933 1934 /** 1935 * xprt_release - release an RPC request slot 1936 * @task: task which is finished with the slot 1937 * 1938 */ 1939 void xprt_release(struct rpc_task *task) 1940 { 1941 struct rpc_xprt *xprt; 1942 struct rpc_rqst *req = task->tk_rqstp; 1943 1944 if (req == NULL) { 1945 if (task->tk_client) { 1946 xprt = task->tk_xprt; 1947 xprt_release_write(xprt, task); 1948 } 1949 return; 1950 } 1951 1952 xprt = req->rq_xprt; 1953 xprt_request_dequeue_xprt(task); 1954 spin_lock(&xprt->transport_lock); 1955 xprt->ops->release_xprt(xprt, task); 1956 if (xprt->ops->release_request) 1957 xprt->ops->release_request(task); 1958 xprt_schedule_autodisconnect(xprt); 1959 spin_unlock(&xprt->transport_lock); 1960 if (req->rq_buffer) 1961 xprt->ops->buf_free(task); 1962 xdr_free_bvec(&req->rq_rcv_buf); 1963 xdr_free_bvec(&req->rq_snd_buf); 1964 if (req->rq_cred != NULL) 1965 put_rpccred(req->rq_cred); 1966 if (req->rq_release_snd_buf) 1967 req->rq_release_snd_buf(req); 1968 1969 task->tk_rqstp = NULL; 1970 if (likely(!bc_prealloc(req))) 1971 xprt->ops->free_slot(xprt, req); 1972 else 1973 xprt_free_bc_request(req); 1974 } 1975 1976 #ifdef CONFIG_SUNRPC_BACKCHANNEL 1977 void 1978 xprt_init_bc_request(struct rpc_rqst *req, struct rpc_task *task) 1979 { 1980 struct xdr_buf *xbufp = &req->rq_snd_buf; 1981 1982 task->tk_rqstp = req; 1983 req->rq_task = task; 1984 xprt_init_connect_cookie(req, req->rq_xprt); 1985 /* 1986 * Set up the xdr_buf length. 1987 * This also indicates that the buffer is XDR encoded already. 1988 */ 1989 xbufp->len = xbufp->head[0].iov_len + xbufp->page_len + 1990 xbufp->tail[0].iov_len; 1991 } 1992 #endif 1993 1994 static void xprt_init(struct rpc_xprt *xprt, struct net *net) 1995 { 1996 kref_init(&xprt->kref); 1997 1998 spin_lock_init(&xprt->transport_lock); 1999 spin_lock_init(&xprt->reserve_lock); 2000 spin_lock_init(&xprt->queue_lock); 2001 2002 INIT_LIST_HEAD(&xprt->free); 2003 xprt->recv_queue = RB_ROOT; 2004 INIT_LIST_HEAD(&xprt->xmit_queue); 2005 #if defined(CONFIG_SUNRPC_BACKCHANNEL) 2006 spin_lock_init(&xprt->bc_pa_lock); 2007 INIT_LIST_HEAD(&xprt->bc_pa_list); 2008 #endif /* CONFIG_SUNRPC_BACKCHANNEL */ 2009 INIT_LIST_HEAD(&xprt->xprt_switch); 2010 2011 xprt->last_used = jiffies; 2012 xprt->cwnd = RPC_INITCWND; 2013 xprt->bind_index = 0; 2014 2015 rpc_init_wait_queue(&xprt->binding, "xprt_binding"); 2016 rpc_init_wait_queue(&xprt->pending, "xprt_pending"); 2017 rpc_init_wait_queue(&xprt->sending, "xprt_sending"); 2018 rpc_init_priority_wait_queue(&xprt->backlog, "xprt_backlog"); 2019 2020 xprt_init_xid(xprt); 2021 2022 xprt->xprt_net = get_net_track(net, &xprt->ns_tracker, GFP_KERNEL); 2023 } 2024 2025 /** 2026 * xprt_create_transport - create an RPC transport 2027 * @args: rpc transport creation arguments 2028 * 2029 */ 2030 struct rpc_xprt *xprt_create_transport(struct xprt_create *args) 2031 { 2032 struct rpc_xprt *xprt; 2033 const struct xprt_class *t; 2034 2035 t = xprt_class_find_by_ident(args->ident); 2036 if (!t) { 2037 dprintk("RPC: transport (%d) not supported\n", args->ident); 2038 return ERR_PTR(-EIO); 2039 } 2040 2041 xprt = t->setup(args); 2042 xprt_class_release(t); 2043 2044 if (IS_ERR(xprt)) 2045 goto out; 2046 if (args->flags & XPRT_CREATE_NO_IDLE_TIMEOUT) 2047 xprt->idle_timeout = 0; 2048 INIT_WORK(&xprt->task_cleanup, xprt_autoclose); 2049 if (xprt_has_timer(xprt)) 2050 timer_setup(&xprt->timer, xprt_init_autodisconnect, 0); 2051 else 2052 timer_setup(&xprt->timer, NULL, 0); 2053 2054 if (strlen(args->servername) > RPC_MAXNETNAMELEN) { 2055 xprt_destroy(xprt); 2056 return ERR_PTR(-EINVAL); 2057 } 2058 xprt->servername = kstrdup(args->servername, GFP_KERNEL); 2059 if (xprt->servername == NULL) { 2060 xprt_destroy(xprt); 2061 return ERR_PTR(-ENOMEM); 2062 } 2063 2064 rpc_xprt_debugfs_register(xprt); 2065 2066 trace_xprt_create(xprt); 2067 out: 2068 return xprt; 2069 } 2070 2071 static void xprt_destroy_cb(struct work_struct *work) 2072 { 2073 struct rpc_xprt *xprt = 2074 container_of(work, struct rpc_xprt, task_cleanup); 2075 2076 trace_xprt_destroy(xprt); 2077 2078 rpc_xprt_debugfs_unregister(xprt); 2079 rpc_destroy_wait_queue(&xprt->binding); 2080 rpc_destroy_wait_queue(&xprt->pending); 2081 rpc_destroy_wait_queue(&xprt->sending); 2082 rpc_destroy_wait_queue(&xprt->backlog); 2083 kfree(xprt->servername); 2084 /* 2085 * Destroy any existing back channel 2086 */ 2087 xprt_destroy_backchannel(xprt, UINT_MAX); 2088 2089 /* 2090 * Tear down transport state and free the rpc_xprt 2091 */ 2092 xprt->ops->destroy(xprt); 2093 } 2094 2095 /** 2096 * xprt_destroy - destroy an RPC transport, killing off all requests. 2097 * @xprt: transport to destroy 2098 * 2099 */ 2100 static void xprt_destroy(struct rpc_xprt *xprt) 2101 { 2102 /* 2103 * Exclude transport connect/disconnect handlers and autoclose 2104 */ 2105 wait_on_bit_lock(&xprt->state, XPRT_LOCKED, TASK_UNINTERRUPTIBLE); 2106 2107 /* 2108 * xprt_schedule_autodisconnect() can run after XPRT_LOCKED 2109 * is cleared. We use ->transport_lock to ensure the mod_timer() 2110 * can only run *before* del_time_sync(), never after. 2111 */ 2112 spin_lock(&xprt->transport_lock); 2113 del_timer_sync(&xprt->timer); 2114 spin_unlock(&xprt->transport_lock); 2115 2116 /* 2117 * Destroy sockets etc from the system workqueue so they can 2118 * safely flush receive work running on rpciod. 2119 */ 2120 INIT_WORK(&xprt->task_cleanup, xprt_destroy_cb); 2121 schedule_work(&xprt->task_cleanup); 2122 } 2123 2124 static void xprt_destroy_kref(struct kref *kref) 2125 { 2126 xprt_destroy(container_of(kref, struct rpc_xprt, kref)); 2127 } 2128 2129 /** 2130 * xprt_get - return a reference to an RPC transport. 2131 * @xprt: pointer to the transport 2132 * 2133 */ 2134 struct rpc_xprt *xprt_get(struct rpc_xprt *xprt) 2135 { 2136 if (xprt != NULL && kref_get_unless_zero(&xprt->kref)) 2137 return xprt; 2138 return NULL; 2139 } 2140 EXPORT_SYMBOL_GPL(xprt_get); 2141 2142 /** 2143 * xprt_put - release a reference to an RPC transport. 2144 * @xprt: pointer to the transport 2145 * 2146 */ 2147 void xprt_put(struct rpc_xprt *xprt) 2148 { 2149 if (xprt != NULL) 2150 kref_put(&xprt->kref, xprt_destroy_kref); 2151 } 2152 EXPORT_SYMBOL_GPL(xprt_put); 2153