2 * linux/net/sunrpc/xprt.c
4 * This is a generic RPC call interface supporting congestion avoidance,
5 * and asynchronous calls.
7 * The interface works like this:
9 * - When a process places a call, it allocates a request slot if
10 * one is available. Otherwise, it sleeps on the backlog queue
12 * - Next, the caller puts together the RPC message, stuffs it into
13 * the request struct, and calls xprt_transmit().
14 * - xprt_transmit sends the message and installs the caller on the
15 * transport's wait list. At the same time, if a reply is expected,
16 * it installs a timer that is run after the packet's timeout has
18 * - When a packet arrives, the data_ready handler walks the list of
19 * pending requests for that transport. If a matching XID is found, the
20 * caller is woken up, and the timer removed.
21 * - When no reply arrives within the timeout interval, the timer is
22 * fired by the kernel and runs xprt_timer(). It either adjusts the
23 * timeout values (minor timeout) or wakes up the caller with a status
25 * - When the caller receives a notification from RPC that a reply arrived,
26 * it should release the RPC slot, and process the reply.
27 * If the call timed out, it may choose to retry the operation by
28 * adjusting the initial timeout value, and simply calling rpc_call
31 * Support for async RPC is done through a set of RPC-specific scheduling
32 * primitives that `transparently' work for processes as well as async
33 * tasks that rely on callbacks.
35 * Copyright (C) 1995-1997, Olaf Kirch <okir@monad.swb.de>
37 * Transport switch API copyright (C) 2005, Chuck Lever <cel@netapp.com>
40 #include <linux/module.h>
42 #include <linux/types.h>
43 #include <linux/interrupt.h>
44 #include <linux/workqueue.h>
45 #include <linux/net.h>
46 #include <linux/ktime.h>
48 #include <linux/sunrpc/clnt.h>
49 #include <linux/sunrpc/metrics.h>
50 #include <linux/sunrpc/bc_xprt.h>
51 #include <linux/rcupdate.h>
53 #include <trace/events/sunrpc.h>
61 #if IS_ENABLED(CONFIG_SUNRPC_DEBUG)
62 # define RPCDBG_FACILITY RPCDBG_XPRT
68 static void xprt_init(struct rpc_xprt *xprt, struct net *net);
69 static __be32 xprt_alloc_xid(struct rpc_xprt *xprt);
70 static void xprt_connect_status(struct rpc_task *task);
71 static int __xprt_get_cong(struct rpc_xprt *, struct rpc_task *);
72 static void __xprt_put_cong(struct rpc_xprt *, struct rpc_rqst *);
73 static void xprt_destroy(struct rpc_xprt *xprt);
75 static DEFINE_SPINLOCK(xprt_list_lock);
76 static LIST_HEAD(xprt_list);
79 * xprt_register_transport - register a transport implementation
80 * @transport: transport to register
82 * If a transport implementation is loaded as a kernel module, it can
83 * call this interface to make itself known to the RPC client.
86 * 0: transport successfully registered
87 * -EEXIST: transport already registered
88 * -EINVAL: transport module being unloaded
90 int xprt_register_transport(struct xprt_class *transport)
96 spin_lock(&xprt_list_lock);
97 list_for_each_entry(t, &xprt_list, list) {
98 /* don't register the same transport class twice */
99 if (t->ident == transport->ident)
103 list_add_tail(&transport->list, &xprt_list);
104 printk(KERN_INFO "RPC: Registered %s transport module.\n",
109 spin_unlock(&xprt_list_lock);
112 EXPORT_SYMBOL_GPL(xprt_register_transport);
115 * xprt_unregister_transport - unregister a transport implementation
116 * @transport: transport to unregister
119 * 0: transport successfully unregistered
120 * -ENOENT: transport never registered
122 int xprt_unregister_transport(struct xprt_class *transport)
124 struct xprt_class *t;
128 spin_lock(&xprt_list_lock);
129 list_for_each_entry(t, &xprt_list, list) {
130 if (t == transport) {
132 "RPC: Unregistered %s transport module.\n",
134 list_del_init(&transport->list);
141 spin_unlock(&xprt_list_lock);
144 EXPORT_SYMBOL_GPL(xprt_unregister_transport);
147 * xprt_load_transport - load a transport implementation
148 * @transport_name: transport to load
151 * 0: transport successfully loaded
152 * -ENOENT: transport module not available
154 int xprt_load_transport(const char *transport_name)
156 struct xprt_class *t;
160 spin_lock(&xprt_list_lock);
161 list_for_each_entry(t, &xprt_list, list) {
162 if (strcmp(t->name, transport_name) == 0) {
163 spin_unlock(&xprt_list_lock);
167 spin_unlock(&xprt_list_lock);
168 result = request_module("xprt%s", transport_name);
172 EXPORT_SYMBOL_GPL(xprt_load_transport);
175 * xprt_reserve_xprt - serialize write access to transports
176 * @task: task that is requesting access to the transport
177 * @xprt: pointer to the target transport
179 * This prevents mixing the payload of separate requests, and prevents
180 * transport connects from colliding with writes. No congestion control
183 int xprt_reserve_xprt(struct rpc_xprt *xprt, struct rpc_task *task)
185 struct rpc_rqst *req = task->tk_rqstp;
188 if (test_and_set_bit(XPRT_LOCKED, &xprt->state)) {
189 if (task == xprt->snd_task)
193 xprt->snd_task = task;
200 dprintk("RPC: %5u failed to lock transport %p\n",
202 task->tk_timeout = 0;
203 task->tk_status = -EAGAIN;
205 priority = RPC_PRIORITY_LOW;
206 else if (!req->rq_ntrans)
207 priority = RPC_PRIORITY_NORMAL;
209 priority = RPC_PRIORITY_HIGH;
210 rpc_sleep_on_priority(&xprt->sending, task, NULL, priority);
213 EXPORT_SYMBOL_GPL(xprt_reserve_xprt);
215 static void xprt_clear_locked(struct rpc_xprt *xprt)
217 xprt->snd_task = NULL;
218 if (!test_bit(XPRT_CLOSE_WAIT, &xprt->state)) {
219 smp_mb__before_atomic();
220 clear_bit(XPRT_LOCKED, &xprt->state);
221 smp_mb__after_atomic();
223 queue_work(xprtiod_workqueue, &xprt->task_cleanup);
227 * xprt_reserve_xprt_cong - serialize write access to transports
228 * @task: task that is requesting access to the transport
230 * Same as xprt_reserve_xprt, but Van Jacobson congestion control is
231 * integrated into the decision of whether a request is allowed to be
232 * woken up and given access to the transport.
234 int xprt_reserve_xprt_cong(struct rpc_xprt *xprt, struct rpc_task *task)
236 struct rpc_rqst *req = task->tk_rqstp;
239 if (test_and_set_bit(XPRT_LOCKED, &xprt->state)) {
240 if (task == xprt->snd_task)
245 xprt->snd_task = task;
248 if (__xprt_get_cong(xprt, task)) {
249 xprt->snd_task = task;
253 xprt_clear_locked(xprt);
256 __xprt_put_cong(xprt, req);
257 dprintk("RPC: %5u failed to lock transport %p\n", task->tk_pid, xprt);
258 task->tk_timeout = 0;
259 task->tk_status = -EAGAIN;
261 priority = RPC_PRIORITY_LOW;
262 else if (!req->rq_ntrans)
263 priority = RPC_PRIORITY_NORMAL;
265 priority = RPC_PRIORITY_HIGH;
266 rpc_sleep_on_priority(&xprt->sending, task, NULL, priority);
269 EXPORT_SYMBOL_GPL(xprt_reserve_xprt_cong);
271 static inline int xprt_lock_write(struct rpc_xprt *xprt, struct rpc_task *task)
275 spin_lock_bh(&xprt->transport_lock);
276 retval = xprt->ops->reserve_xprt(xprt, task);
277 spin_unlock_bh(&xprt->transport_lock);
281 static bool __xprt_lock_write_func(struct rpc_task *task, void *data)
283 struct rpc_xprt *xprt = data;
284 struct rpc_rqst *req;
286 req = task->tk_rqstp;
287 xprt->snd_task = task;
293 static void __xprt_lock_write_next(struct rpc_xprt *xprt)
295 if (test_and_set_bit(XPRT_LOCKED, &xprt->state))
298 if (rpc_wake_up_first_on_wq(xprtiod_workqueue, &xprt->sending,
299 __xprt_lock_write_func, xprt))
301 xprt_clear_locked(xprt);
304 static bool __xprt_lock_write_cong_func(struct rpc_task *task, void *data)
306 struct rpc_xprt *xprt = data;
307 struct rpc_rqst *req;
309 req = task->tk_rqstp;
311 xprt->snd_task = task;
314 if (__xprt_get_cong(xprt, task)) {
315 xprt->snd_task = task;
322 static void __xprt_lock_write_next_cong(struct rpc_xprt *xprt)
324 if (test_and_set_bit(XPRT_LOCKED, &xprt->state))
326 if (RPCXPRT_CONGESTED(xprt))
328 if (rpc_wake_up_first_on_wq(xprtiod_workqueue, &xprt->sending,
329 __xprt_lock_write_cong_func, xprt))
332 xprt_clear_locked(xprt);
335 static void xprt_task_clear_bytes_sent(struct rpc_task *task)
338 struct rpc_rqst *req = task->tk_rqstp;
340 req->rq_bytes_sent = 0;
345 * xprt_release_xprt - allow other requests to use a transport
346 * @xprt: transport with other tasks potentially waiting
347 * @task: task that is releasing access to the transport
349 * Note that "task" can be NULL. No congestion control is provided.
351 void xprt_release_xprt(struct rpc_xprt *xprt, struct rpc_task *task)
353 if (xprt->snd_task == task) {
354 xprt_task_clear_bytes_sent(task);
355 xprt_clear_locked(xprt);
356 __xprt_lock_write_next(xprt);
359 EXPORT_SYMBOL_GPL(xprt_release_xprt);
362 * xprt_release_xprt_cong - allow other requests to use a transport
363 * @xprt: transport with other tasks potentially waiting
364 * @task: task that is releasing access to the transport
366 * Note that "task" can be NULL. Another task is awoken to use the
367 * transport if the transport's congestion window allows it.
369 void xprt_release_xprt_cong(struct rpc_xprt *xprt, struct rpc_task *task)
371 if (xprt->snd_task == task) {
372 xprt_task_clear_bytes_sent(task);
373 xprt_clear_locked(xprt);
374 __xprt_lock_write_next_cong(xprt);
377 EXPORT_SYMBOL_GPL(xprt_release_xprt_cong);
379 static inline void xprt_release_write(struct rpc_xprt *xprt, struct rpc_task *task)
381 spin_lock_bh(&xprt->transport_lock);
382 xprt->ops->release_xprt(xprt, task);
383 spin_unlock_bh(&xprt->transport_lock);
387 * Van Jacobson congestion avoidance. Check if the congestion window
388 * overflowed. Put the task to sleep if this is the case.
391 __xprt_get_cong(struct rpc_xprt *xprt, struct rpc_task *task)
393 struct rpc_rqst *req = task->tk_rqstp;
397 dprintk("RPC: %5u xprt_cwnd_limited cong = %lu cwnd = %lu\n",
398 task->tk_pid, xprt->cong, xprt->cwnd);
399 if (RPCXPRT_CONGESTED(xprt))
402 xprt->cong += RPC_CWNDSCALE;
407 * Adjust the congestion window, and wake up the next task
408 * that has been sleeping due to congestion
411 __xprt_put_cong(struct rpc_xprt *xprt, struct rpc_rqst *req)
416 xprt->cong -= RPC_CWNDSCALE;
417 __xprt_lock_write_next_cong(xprt);
421 * xprt_release_rqst_cong - housekeeping when request is complete
422 * @task: RPC request that recently completed
424 * Useful for transports that require congestion control.
426 void xprt_release_rqst_cong(struct rpc_task *task)
428 struct rpc_rqst *req = task->tk_rqstp;
430 __xprt_put_cong(req->rq_xprt, req);
432 EXPORT_SYMBOL_GPL(xprt_release_rqst_cong);
435 * xprt_adjust_cwnd - adjust transport congestion window
436 * @xprt: pointer to xprt
437 * @task: recently completed RPC request used to adjust window
438 * @result: result code of completed RPC request
440 * The transport code maintains an estimate on the maximum number of out-
441 * standing RPC requests, using a smoothed version of the congestion
442 * avoidance implemented in 44BSD. This is basically the Van Jacobson
443 * congestion algorithm: If a retransmit occurs, the congestion window is
444 * halved; otherwise, it is incremented by 1/cwnd when
446 * - a reply is received and
447 * - a full number of requests are outstanding and
448 * - the congestion window hasn't been updated recently.
450 void xprt_adjust_cwnd(struct rpc_xprt *xprt, struct rpc_task *task, int result)
452 struct rpc_rqst *req = task->tk_rqstp;
453 unsigned long cwnd = xprt->cwnd;
455 if (result >= 0 && cwnd <= xprt->cong) {
456 /* The (cwnd >> 1) term makes sure
457 * the result gets rounded properly. */
458 cwnd += (RPC_CWNDSCALE * RPC_CWNDSCALE + (cwnd >> 1)) / cwnd;
459 if (cwnd > RPC_MAXCWND(xprt))
460 cwnd = RPC_MAXCWND(xprt);
461 __xprt_lock_write_next_cong(xprt);
462 } else if (result == -ETIMEDOUT) {
464 if (cwnd < RPC_CWNDSCALE)
465 cwnd = RPC_CWNDSCALE;
467 dprintk("RPC: cong %ld, cwnd was %ld, now %ld\n",
468 xprt->cong, xprt->cwnd, cwnd);
470 __xprt_put_cong(xprt, req);
472 EXPORT_SYMBOL_GPL(xprt_adjust_cwnd);
475 * xprt_wake_pending_tasks - wake all tasks on a transport's pending queue
476 * @xprt: transport with waiting tasks
477 * @status: result code to plant in each task before waking it
480 void xprt_wake_pending_tasks(struct rpc_xprt *xprt, int status)
483 rpc_wake_up_status(&xprt->pending, status);
485 rpc_wake_up(&xprt->pending);
487 EXPORT_SYMBOL_GPL(xprt_wake_pending_tasks);
490 * xprt_wait_for_buffer_space - wait for transport output buffer to clear
491 * @task: task to be put to sleep
492 * @action: function pointer to be executed after wait
494 * Note that we only set the timer for the case of RPC_IS_SOFT(), since
495 * we don't in general want to force a socket disconnection due to
496 * an incomplete RPC call transmission.
498 void xprt_wait_for_buffer_space(struct rpc_task *task, rpc_action action)
500 struct rpc_rqst *req = task->tk_rqstp;
501 struct rpc_xprt *xprt = req->rq_xprt;
503 task->tk_timeout = RPC_IS_SOFT(task) ? req->rq_timeout : 0;
504 rpc_sleep_on(&xprt->pending, task, action);
506 EXPORT_SYMBOL_GPL(xprt_wait_for_buffer_space);
509 * xprt_write_space - wake the task waiting for transport output buffer space
510 * @xprt: transport with waiting tasks
512 * Can be called in a soft IRQ context, so xprt_write_space never sleeps.
514 void xprt_write_space(struct rpc_xprt *xprt)
516 spin_lock_bh(&xprt->transport_lock);
517 if (xprt->snd_task) {
518 dprintk("RPC: write space: waking waiting task on "
520 rpc_wake_up_queued_task_on_wq(xprtiod_workqueue,
521 &xprt->pending, xprt->snd_task);
523 spin_unlock_bh(&xprt->transport_lock);
525 EXPORT_SYMBOL_GPL(xprt_write_space);
528 * xprt_set_retrans_timeout_def - set a request's retransmit timeout
529 * @task: task whose timeout is to be set
531 * Set a request's retransmit timeout based on the transport's
532 * default timeout parameters. Used by transports that don't adjust
533 * the retransmit timeout based on round-trip time estimation.
535 void xprt_set_retrans_timeout_def(struct rpc_task *task)
537 task->tk_timeout = task->tk_rqstp->rq_timeout;
539 EXPORT_SYMBOL_GPL(xprt_set_retrans_timeout_def);
542 * xprt_set_retrans_timeout_rtt - set a request's retransmit timeout
543 * @task: task whose timeout is to be set
545 * Set a request's retransmit timeout using the RTT estimator.
547 void xprt_set_retrans_timeout_rtt(struct rpc_task *task)
549 int timer = task->tk_msg.rpc_proc->p_timer;
550 struct rpc_clnt *clnt = task->tk_client;
551 struct rpc_rtt *rtt = clnt->cl_rtt;
552 struct rpc_rqst *req = task->tk_rqstp;
553 unsigned long max_timeout = clnt->cl_timeout->to_maxval;
555 task->tk_timeout = rpc_calc_rto(rtt, timer);
556 task->tk_timeout <<= rpc_ntimeo(rtt, timer) + req->rq_retries;
557 if (task->tk_timeout > max_timeout || task->tk_timeout == 0)
558 task->tk_timeout = max_timeout;
560 EXPORT_SYMBOL_GPL(xprt_set_retrans_timeout_rtt);
562 static void xprt_reset_majortimeo(struct rpc_rqst *req)
564 const struct rpc_timeout *to = req->rq_task->tk_client->cl_timeout;
566 req->rq_majortimeo = req->rq_timeout;
567 if (to->to_exponential)
568 req->rq_majortimeo <<= to->to_retries;
570 req->rq_majortimeo += to->to_increment * to->to_retries;
571 if (req->rq_majortimeo > to->to_maxval || req->rq_majortimeo == 0)
572 req->rq_majortimeo = to->to_maxval;
573 req->rq_majortimeo += jiffies;
577 * xprt_adjust_timeout - adjust timeout values for next retransmit
578 * @req: RPC request containing parameters to use for the adjustment
581 int xprt_adjust_timeout(struct rpc_rqst *req)
583 struct rpc_xprt *xprt = req->rq_xprt;
584 const struct rpc_timeout *to = req->rq_task->tk_client->cl_timeout;
587 if (time_before(jiffies, req->rq_majortimeo)) {
588 if (to->to_exponential)
589 req->rq_timeout <<= 1;
591 req->rq_timeout += to->to_increment;
592 if (to->to_maxval && req->rq_timeout >= to->to_maxval)
593 req->rq_timeout = to->to_maxval;
596 req->rq_timeout = to->to_initval;
598 xprt_reset_majortimeo(req);
599 /* Reset the RTT counters == "slow start" */
600 spin_lock_bh(&xprt->transport_lock);
601 rpc_init_rtt(req->rq_task->tk_client->cl_rtt, to->to_initval);
602 spin_unlock_bh(&xprt->transport_lock);
606 if (req->rq_timeout == 0) {
607 printk(KERN_WARNING "xprt_adjust_timeout: rq_timeout = 0!\n");
608 req->rq_timeout = 5 * HZ;
613 static void xprt_autoclose(struct work_struct *work)
615 struct rpc_xprt *xprt =
616 container_of(work, struct rpc_xprt, task_cleanup);
618 clear_bit(XPRT_CLOSE_WAIT, &xprt->state);
619 xprt->ops->close(xprt);
620 xprt_release_write(xprt, NULL);
621 wake_up_bit(&xprt->state, XPRT_LOCKED);
625 * xprt_disconnect_done - mark a transport as disconnected
626 * @xprt: transport to flag for disconnect
629 void xprt_disconnect_done(struct rpc_xprt *xprt)
631 dprintk("RPC: disconnected transport %p\n", xprt);
632 spin_lock_bh(&xprt->transport_lock);
633 xprt_clear_connected(xprt);
634 xprt_wake_pending_tasks(xprt, -EAGAIN);
635 spin_unlock_bh(&xprt->transport_lock);
637 EXPORT_SYMBOL_GPL(xprt_disconnect_done);
640 * xprt_force_disconnect - force a transport to disconnect
641 * @xprt: transport to disconnect
644 void xprt_force_disconnect(struct rpc_xprt *xprt)
646 /* Don't race with the test_bit() in xprt_clear_locked() */
647 spin_lock_bh(&xprt->transport_lock);
648 set_bit(XPRT_CLOSE_WAIT, &xprt->state);
649 /* Try to schedule an autoclose RPC call */
650 if (test_and_set_bit(XPRT_LOCKED, &xprt->state) == 0)
651 queue_work(xprtiod_workqueue, &xprt->task_cleanup);
652 xprt_wake_pending_tasks(xprt, -EAGAIN);
653 spin_unlock_bh(&xprt->transport_lock);
655 EXPORT_SYMBOL_GPL(xprt_force_disconnect);
658 xprt_connect_cookie(struct rpc_xprt *xprt)
660 return READ_ONCE(xprt->connect_cookie);
664 xprt_request_retransmit_after_disconnect(struct rpc_task *task)
666 struct rpc_rqst *req = task->tk_rqstp;
667 struct rpc_xprt *xprt = req->rq_xprt;
669 return req->rq_connect_cookie != xprt_connect_cookie(xprt) ||
670 !xprt_connected(xprt);
674 * xprt_conditional_disconnect - force a transport to disconnect
675 * @xprt: transport to disconnect
676 * @cookie: 'connection cookie'
678 * This attempts to break the connection if and only if 'cookie' matches
679 * the current transport 'connection cookie'. It ensures that we don't
680 * try to break the connection more than once when we need to retransmit
681 * a batch of RPC requests.
684 void xprt_conditional_disconnect(struct rpc_xprt *xprt, unsigned int cookie)
686 /* Don't race with the test_bit() in xprt_clear_locked() */
687 spin_lock_bh(&xprt->transport_lock);
688 if (cookie != xprt->connect_cookie)
690 if (test_bit(XPRT_CLOSING, &xprt->state))
692 set_bit(XPRT_CLOSE_WAIT, &xprt->state);
693 /* Try to schedule an autoclose RPC call */
694 if (test_and_set_bit(XPRT_LOCKED, &xprt->state) == 0)
695 queue_work(xprtiod_workqueue, &xprt->task_cleanup);
696 xprt_wake_pending_tasks(xprt, -EAGAIN);
698 spin_unlock_bh(&xprt->transport_lock);
702 xprt_has_timer(const struct rpc_xprt *xprt)
704 return xprt->idle_timeout != 0;
708 xprt_schedule_autodisconnect(struct rpc_xprt *xprt)
709 __must_hold(&xprt->transport_lock)
711 if (list_empty(&xprt->recv_queue) && xprt_has_timer(xprt))
712 mod_timer(&xprt->timer, xprt->last_used + xprt->idle_timeout);
716 xprt_init_autodisconnect(struct timer_list *t)
718 struct rpc_xprt *xprt = from_timer(xprt, t, timer);
720 spin_lock(&xprt->transport_lock);
721 if (!list_empty(&xprt->recv_queue))
723 /* Reset xprt->last_used to avoid connect/autodisconnect cycling */
724 xprt->last_used = jiffies;
725 if (test_and_set_bit(XPRT_LOCKED, &xprt->state))
727 spin_unlock(&xprt->transport_lock);
728 queue_work(xprtiod_workqueue, &xprt->task_cleanup);
731 spin_unlock(&xprt->transport_lock);
734 bool xprt_lock_connect(struct rpc_xprt *xprt,
735 struct rpc_task *task,
740 spin_lock_bh(&xprt->transport_lock);
741 if (!test_bit(XPRT_LOCKED, &xprt->state))
743 if (xprt->snd_task != task)
745 xprt_task_clear_bytes_sent(task);
746 xprt->snd_task = cookie;
749 spin_unlock_bh(&xprt->transport_lock);
753 void xprt_unlock_connect(struct rpc_xprt *xprt, void *cookie)
755 spin_lock_bh(&xprt->transport_lock);
756 if (xprt->snd_task != cookie)
758 if (!test_bit(XPRT_LOCKED, &xprt->state))
760 xprt->snd_task =NULL;
761 xprt->ops->release_xprt(xprt, NULL);
762 xprt_schedule_autodisconnect(xprt);
764 spin_unlock_bh(&xprt->transport_lock);
765 wake_up_bit(&xprt->state, XPRT_LOCKED);
769 * xprt_connect - schedule a transport connect operation
770 * @task: RPC task that is requesting the connect
773 void xprt_connect(struct rpc_task *task)
775 struct rpc_xprt *xprt = task->tk_rqstp->rq_xprt;
777 dprintk("RPC: %5u xprt_connect xprt %p %s connected\n", task->tk_pid,
778 xprt, (xprt_connected(xprt) ? "is" : "is not"));
780 if (!xprt_bound(xprt)) {
781 task->tk_status = -EAGAIN;
784 if (!xprt_lock_write(xprt, task))
787 if (test_and_clear_bit(XPRT_CLOSE_WAIT, &xprt->state))
788 xprt->ops->close(xprt);
790 if (!xprt_connected(xprt)) {
791 task->tk_rqstp->rq_bytes_sent = 0;
792 task->tk_timeout = task->tk_rqstp->rq_timeout;
793 task->tk_rqstp->rq_connect_cookie = xprt->connect_cookie;
794 rpc_sleep_on(&xprt->pending, task, xprt_connect_status);
796 if (test_bit(XPRT_CLOSING, &xprt->state))
798 if (xprt_test_and_set_connecting(xprt))
800 xprt->stat.connect_start = jiffies;
801 xprt->ops->connect(xprt, task);
803 xprt_release_write(xprt, task);
806 static void xprt_connect_status(struct rpc_task *task)
808 struct rpc_xprt *xprt = task->tk_rqstp->rq_xprt;
810 if (task->tk_status == 0) {
811 xprt->stat.connect_count++;
812 xprt->stat.connect_time += (long)jiffies - xprt->stat.connect_start;
813 dprintk("RPC: %5u xprt_connect_status: connection established\n",
818 switch (task->tk_status) {
826 dprintk("RPC: %5u xprt_connect_status: retrying\n", task->tk_pid);
829 dprintk("RPC: %5u xprt_connect_status: connect attempt timed "
830 "out\n", task->tk_pid);
833 dprintk("RPC: %5u xprt_connect_status: error %d connecting to "
834 "server %s\n", task->tk_pid, -task->tk_status,
836 task->tk_status = -EIO;
841 * xprt_lookup_rqst - find an RPC request corresponding to an XID
842 * @xprt: transport on which the original request was transmitted
843 * @xid: RPC XID of incoming reply
845 * Caller holds xprt->queue_lock.
847 struct rpc_rqst *xprt_lookup_rqst(struct rpc_xprt *xprt, __be32 xid)
849 struct rpc_rqst *entry;
851 list_for_each_entry(entry, &xprt->recv_queue, rq_recv)
852 if (entry->rq_xid == xid) {
853 trace_xprt_lookup_rqst(xprt, xid, 0);
854 entry->rq_rtt = ktime_sub(ktime_get(), entry->rq_xtime);
858 dprintk("RPC: xprt_lookup_rqst did not find xid %08x\n",
860 trace_xprt_lookup_rqst(xprt, xid, -ENOENT);
861 xprt->stat.bad_xids++;
864 EXPORT_SYMBOL_GPL(xprt_lookup_rqst);
867 xprt_is_pinned_rqst(struct rpc_rqst *req)
869 return atomic_read(&req->rq_pin) != 0;
873 * xprt_pin_rqst - Pin a request on the transport receive list
874 * @req: Request to pin
876 * Caller must ensure this is atomic with the call to xprt_lookup_rqst()
877 * so should be holding the xprt receive lock.
879 void xprt_pin_rqst(struct rpc_rqst *req)
881 atomic_inc(&req->rq_pin);
883 EXPORT_SYMBOL_GPL(xprt_pin_rqst);
886 * xprt_unpin_rqst - Unpin a request on the transport receive list
887 * @req: Request to pin
889 * Caller should be holding the xprt receive lock.
891 void xprt_unpin_rqst(struct rpc_rqst *req)
893 if (!test_bit(RPC_TASK_MSG_PIN_WAIT, &req->rq_task->tk_runstate)) {
894 atomic_dec(&req->rq_pin);
897 if (atomic_dec_and_test(&req->rq_pin))
898 wake_up_var(&req->rq_pin);
900 EXPORT_SYMBOL_GPL(xprt_unpin_rqst);
902 static void xprt_wait_on_pinned_rqst(struct rpc_rqst *req)
904 wait_var_event(&req->rq_pin, !xprt_is_pinned_rqst(req));
908 xprt_request_data_received(struct rpc_task *task)
910 return !test_bit(RPC_TASK_NEED_RECV, &task->tk_runstate) &&
911 READ_ONCE(task->tk_rqstp->rq_reply_bytes_recvd) != 0;
915 xprt_request_need_enqueue_receive(struct rpc_task *task, struct rpc_rqst *req)
917 return !test_bit(RPC_TASK_NEED_RECV, &task->tk_runstate) &&
918 READ_ONCE(task->tk_rqstp->rq_reply_bytes_recvd) == 0;
922 * xprt_request_enqueue_receive - Add an request to the receive queue
927 xprt_request_enqueue_receive(struct rpc_task *task)
929 struct rpc_rqst *req = task->tk_rqstp;
930 struct rpc_xprt *xprt = req->rq_xprt;
932 if (!xprt_request_need_enqueue_receive(task, req))
934 spin_lock(&xprt->queue_lock);
936 /* Update the softirq receive buffer */
937 memcpy(&req->rq_private_buf, &req->rq_rcv_buf,
938 sizeof(req->rq_private_buf));
940 /* Add request to the receive list */
941 list_add_tail(&req->rq_recv, &xprt->recv_queue);
942 set_bit(RPC_TASK_NEED_RECV, &task->tk_runstate);
943 spin_unlock(&xprt->queue_lock);
945 xprt_reset_majortimeo(req);
946 /* Turn off autodisconnect */
947 del_singleshot_timer_sync(&xprt->timer);
951 * xprt_request_dequeue_receive_locked - Remove a request from the receive queue
954 * Caller must hold xprt->queue_lock.
957 xprt_request_dequeue_receive_locked(struct rpc_task *task)
959 if (test_and_clear_bit(RPC_TASK_NEED_RECV, &task->tk_runstate))
960 list_del(&task->tk_rqstp->rq_recv);
964 * xprt_update_rtt - Update RPC RTT statistics
965 * @task: RPC request that recently completed
967 * Caller holds xprt->queue_lock.
969 void xprt_update_rtt(struct rpc_task *task)
971 struct rpc_rqst *req = task->tk_rqstp;
972 struct rpc_rtt *rtt = task->tk_client->cl_rtt;
973 unsigned int timer = task->tk_msg.rpc_proc->p_timer;
974 long m = usecs_to_jiffies(ktime_to_us(req->rq_rtt));
977 if (req->rq_ntrans == 1)
978 rpc_update_rtt(rtt, timer, m);
979 rpc_set_timeo(rtt, timer, req->rq_ntrans - 1);
982 EXPORT_SYMBOL_GPL(xprt_update_rtt);
985 * xprt_complete_rqst - called when reply processing is complete
986 * @task: RPC request that recently completed
987 * @copied: actual number of bytes received from the transport
989 * Caller holds xprt->queue_lock.
991 void xprt_complete_rqst(struct rpc_task *task, int copied)
993 struct rpc_rqst *req = task->tk_rqstp;
994 struct rpc_xprt *xprt = req->rq_xprt;
996 dprintk("RPC: %5u xid %08x complete (%d bytes received)\n",
997 task->tk_pid, ntohl(req->rq_xid), copied);
998 trace_xprt_complete_rqst(xprt, req->rq_xid, copied);
1002 req->rq_private_buf.len = copied;
1003 /* Ensure all writes are done before we update */
1004 /* req->rq_reply_bytes_recvd */
1006 req->rq_reply_bytes_recvd = copied;
1007 xprt_request_dequeue_receive_locked(task);
1008 rpc_wake_up_queued_task(&xprt->pending, task);
1010 EXPORT_SYMBOL_GPL(xprt_complete_rqst);
1012 static void xprt_timer(struct rpc_task *task)
1014 struct rpc_rqst *req = task->tk_rqstp;
1015 struct rpc_xprt *xprt = req->rq_xprt;
1017 if (task->tk_status != -ETIMEDOUT)
1020 trace_xprt_timer(xprt, req->rq_xid, task->tk_status);
1021 if (!req->rq_reply_bytes_recvd) {
1022 if (xprt->ops->timer)
1023 xprt->ops->timer(xprt, task);
1025 task->tk_status = 0;
1029 * xprt_request_wait_receive - wait for the reply to an RPC request
1030 * @task: RPC task about to send a request
1033 void xprt_request_wait_receive(struct rpc_task *task)
1035 struct rpc_rqst *req = task->tk_rqstp;
1036 struct rpc_xprt *xprt = req->rq_xprt;
1038 if (!test_bit(RPC_TASK_NEED_RECV, &task->tk_runstate))
1041 * Sleep on the pending queue if we're expecting a reply.
1042 * The spinlock ensures atomicity between the test of
1043 * req->rq_reply_bytes_recvd, and the call to rpc_sleep_on().
1045 spin_lock(&xprt->queue_lock);
1046 if (test_bit(RPC_TASK_NEED_RECV, &task->tk_runstate)) {
1047 xprt->ops->set_retrans_timeout(task);
1048 rpc_sleep_on(&xprt->pending, task, xprt_timer);
1050 * Send an extra queue wakeup call if the
1051 * connection was dropped in case the call to
1052 * rpc_sleep_on() raced.
1054 if (xprt_request_retransmit_after_disconnect(task))
1055 rpc_wake_up_queued_task_set_status(&xprt->pending,
1058 spin_unlock(&xprt->queue_lock);
1062 * xprt_prepare_transmit - reserve the transport before sending a request
1063 * @task: RPC task about to send a request
1066 bool xprt_prepare_transmit(struct rpc_task *task)
1068 struct rpc_rqst *req = task->tk_rqstp;
1069 struct rpc_xprt *xprt = req->rq_xprt;
1072 dprintk("RPC: %5u xprt_prepare_transmit\n", task->tk_pid);
1074 spin_lock_bh(&xprt->transport_lock);
1075 if (!req->rq_bytes_sent) {
1076 if (req->rq_reply_bytes_recvd) {
1077 task->tk_status = req->rq_reply_bytes_recvd;
1080 if ((task->tk_flags & RPC_TASK_NO_RETRANS_TIMEOUT) &&
1081 !xprt_request_retransmit_after_disconnect(task)) {
1082 xprt->ops->set_retrans_timeout(task);
1083 rpc_sleep_on(&xprt->pending, task, xprt_timer);
1087 if (!xprt->ops->reserve_xprt(xprt, task)) {
1088 task->tk_status = -EAGAIN;
1093 spin_unlock_bh(&xprt->transport_lock);
1097 void xprt_end_transmit(struct rpc_task *task)
1099 xprt_release_write(task->tk_rqstp->rq_xprt, task);
1103 * xprt_transmit - send an RPC request on a transport
1104 * @task: controlling RPC task
1106 * We have to copy the iovec because sendmsg fiddles with its contents.
1108 void xprt_transmit(struct rpc_task *task)
1110 struct rpc_rqst *req = task->tk_rqstp;
1111 struct rpc_xprt *xprt = req->rq_xprt;
1112 unsigned int connect_cookie;
1115 dprintk("RPC: %5u xprt_transmit(%u)\n", task->tk_pid, req->rq_slen);
1117 if (!req->rq_bytes_sent) {
1118 if (xprt_request_data_received(task))
1120 /* Verify that our message lies in the RPCSEC_GSS window */
1121 if (rpcauth_xmit_need_reencode(task)) {
1122 task->tk_status = -EBADMSG;
1127 connect_cookie = xprt->connect_cookie;
1128 status = xprt->ops->send_request(task);
1129 trace_xprt_transmit(xprt, req->rq_xid, status);
1131 task->tk_status = status;
1135 xprt_inject_disconnect(xprt);
1137 dprintk("RPC: %5u xmit complete\n", task->tk_pid);
1138 clear_bit(RPC_TASK_NEED_XMIT, &task->tk_runstate);
1139 task->tk_flags |= RPC_TASK_SENT;
1140 spin_lock_bh(&xprt->transport_lock);
1143 xprt->stat.req_u += xprt->stat.sends - xprt->stat.recvs;
1144 xprt->stat.bklog_u += xprt->backlog.qlen;
1145 xprt->stat.sending_u += xprt->sending.qlen;
1146 xprt->stat.pending_u += xprt->pending.qlen;
1147 spin_unlock_bh(&xprt->transport_lock);
1149 req->rq_connect_cookie = connect_cookie;
1152 static void xprt_add_backlog(struct rpc_xprt *xprt, struct rpc_task *task)
1154 set_bit(XPRT_CONGESTED, &xprt->state);
1155 rpc_sleep_on(&xprt->backlog, task, NULL);
1158 static void xprt_wake_up_backlog(struct rpc_xprt *xprt)
1160 if (rpc_wake_up_next(&xprt->backlog) == NULL)
1161 clear_bit(XPRT_CONGESTED, &xprt->state);
1164 static bool xprt_throttle_congested(struct rpc_xprt *xprt, struct rpc_task *task)
1168 if (!test_bit(XPRT_CONGESTED, &xprt->state))
1170 spin_lock(&xprt->reserve_lock);
1171 if (test_bit(XPRT_CONGESTED, &xprt->state)) {
1172 rpc_sleep_on(&xprt->backlog, task, NULL);
1175 spin_unlock(&xprt->reserve_lock);
1180 static struct rpc_rqst *xprt_dynamic_alloc_slot(struct rpc_xprt *xprt)
1182 struct rpc_rqst *req = ERR_PTR(-EAGAIN);
1184 if (xprt->num_reqs >= xprt->max_reqs)
1187 spin_unlock(&xprt->reserve_lock);
1188 req = kzalloc(sizeof(struct rpc_rqst), GFP_NOFS);
1189 spin_lock(&xprt->reserve_lock);
1193 req = ERR_PTR(-ENOMEM);
1198 static bool xprt_dynamic_free_slot(struct rpc_xprt *xprt, struct rpc_rqst *req)
1200 if (xprt->num_reqs > xprt->min_reqs) {
1208 void xprt_alloc_slot(struct rpc_xprt *xprt, struct rpc_task *task)
1210 struct rpc_rqst *req;
1212 spin_lock(&xprt->reserve_lock);
1213 if (!list_empty(&xprt->free)) {
1214 req = list_entry(xprt->free.next, struct rpc_rqst, rq_list);
1215 list_del(&req->rq_list);
1218 req = xprt_dynamic_alloc_slot(xprt);
1221 switch (PTR_ERR(req)) {
1223 dprintk("RPC: dynamic allocation of request slot "
1224 "failed! Retrying\n");
1225 task->tk_status = -ENOMEM;
1228 xprt_add_backlog(xprt, task);
1229 dprintk("RPC: waiting for request slot\n");
1232 task->tk_status = -EAGAIN;
1234 spin_unlock(&xprt->reserve_lock);
1237 xprt->stat.max_slots = max_t(unsigned int, xprt->stat.max_slots,
1239 spin_unlock(&xprt->reserve_lock);
1241 task->tk_status = 0;
1242 task->tk_rqstp = req;
1244 EXPORT_SYMBOL_GPL(xprt_alloc_slot);
1246 void xprt_lock_and_alloc_slot(struct rpc_xprt *xprt, struct rpc_task *task)
1248 /* Note: grabbing the xprt_lock_write() ensures that we throttle
1249 * new slot allocation if the transport is congested (i.e. when
1250 * reconnecting a stream transport or when out of socket write
1253 if (xprt_lock_write(xprt, task)) {
1254 xprt_alloc_slot(xprt, task);
1255 xprt_release_write(xprt, task);
1258 EXPORT_SYMBOL_GPL(xprt_lock_and_alloc_slot);
1260 void xprt_free_slot(struct rpc_xprt *xprt, struct rpc_rqst *req)
1262 spin_lock(&xprt->reserve_lock);
1263 if (!xprt_dynamic_free_slot(xprt, req)) {
1264 memset(req, 0, sizeof(*req)); /* mark unused */
1265 list_add(&req->rq_list, &xprt->free);
1267 xprt_wake_up_backlog(xprt);
1268 spin_unlock(&xprt->reserve_lock);
1270 EXPORT_SYMBOL_GPL(xprt_free_slot);
1272 static void xprt_free_all_slots(struct rpc_xprt *xprt)
1274 struct rpc_rqst *req;
1275 while (!list_empty(&xprt->free)) {
1276 req = list_first_entry(&xprt->free, struct rpc_rqst, rq_list);
1277 list_del(&req->rq_list);
1282 struct rpc_xprt *xprt_alloc(struct net *net, size_t size,
1283 unsigned int num_prealloc,
1284 unsigned int max_alloc)
1286 struct rpc_xprt *xprt;
1287 struct rpc_rqst *req;
1290 xprt = kzalloc(size, GFP_KERNEL);
1294 xprt_init(xprt, net);
1296 for (i = 0; i < num_prealloc; i++) {
1297 req = kzalloc(sizeof(struct rpc_rqst), GFP_KERNEL);
1300 list_add(&req->rq_list, &xprt->free);
1302 if (max_alloc > num_prealloc)
1303 xprt->max_reqs = max_alloc;
1305 xprt->max_reqs = num_prealloc;
1306 xprt->min_reqs = num_prealloc;
1307 xprt->num_reqs = num_prealloc;
1316 EXPORT_SYMBOL_GPL(xprt_alloc);
1318 void xprt_free(struct rpc_xprt *xprt)
1320 put_net(xprt->xprt_net);
1321 xprt_free_all_slots(xprt);
1322 kfree_rcu(xprt, rcu);
1324 EXPORT_SYMBOL_GPL(xprt_free);
1327 xprt_alloc_xid(struct rpc_xprt *xprt)
1331 spin_lock(&xprt->reserve_lock);
1332 xid = (__force __be32)xprt->xid++;
1333 spin_unlock(&xprt->reserve_lock);
1338 xprt_init_xid(struct rpc_xprt *xprt)
1340 xprt->xid = prandom_u32();
1344 xprt_request_init(struct rpc_task *task)
1346 struct rpc_xprt *xprt = task->tk_xprt;
1347 struct rpc_rqst *req = task->tk_rqstp;
1349 req->rq_timeout = task->tk_client->cl_timeout->to_initval;
1350 req->rq_task = task;
1351 req->rq_xprt = xprt;
1352 req->rq_buffer = NULL;
1353 req->rq_xid = xprt_alloc_xid(xprt);
1354 req->rq_connect_cookie = xprt_connect_cookie(xprt) - 1;
1355 req->rq_bytes_sent = 0;
1356 req->rq_snd_buf.len = 0;
1357 req->rq_snd_buf.buflen = 0;
1358 req->rq_rcv_buf.len = 0;
1359 req->rq_rcv_buf.buflen = 0;
1360 req->rq_release_snd_buf = NULL;
1361 xprt_reset_majortimeo(req);
1362 dprintk("RPC: %5u reserved req %p xid %08x\n", task->tk_pid,
1363 req, ntohl(req->rq_xid));
1367 xprt_do_reserve(struct rpc_xprt *xprt, struct rpc_task *task)
1369 xprt->ops->alloc_slot(xprt, task);
1370 if (task->tk_rqstp != NULL)
1371 xprt_request_init(task);
1375 * xprt_reserve - allocate an RPC request slot
1376 * @task: RPC task requesting a slot allocation
1378 * If the transport is marked as being congested, or if no more
1379 * slots are available, place the task on the transport's
1382 void xprt_reserve(struct rpc_task *task)
1384 struct rpc_xprt *xprt = task->tk_xprt;
1386 task->tk_status = 0;
1387 if (task->tk_rqstp != NULL)
1390 task->tk_timeout = 0;
1391 task->tk_status = -EAGAIN;
1392 if (!xprt_throttle_congested(xprt, task))
1393 xprt_do_reserve(xprt, task);
1397 * xprt_retry_reserve - allocate an RPC request slot
1398 * @task: RPC task requesting a slot allocation
1400 * If no more slots are available, place the task on the transport's
1402 * Note that the only difference with xprt_reserve is that we now
1403 * ignore the value of the XPRT_CONGESTED flag.
1405 void xprt_retry_reserve(struct rpc_task *task)
1407 struct rpc_xprt *xprt = task->tk_xprt;
1409 task->tk_status = 0;
1410 if (task->tk_rqstp != NULL)
1413 task->tk_timeout = 0;
1414 task->tk_status = -EAGAIN;
1415 xprt_do_reserve(xprt, task);
1419 xprt_request_dequeue_all(struct rpc_task *task, struct rpc_rqst *req)
1421 struct rpc_xprt *xprt = req->rq_xprt;
1423 if (test_bit(RPC_TASK_NEED_RECV, &task->tk_runstate) ||
1424 xprt_is_pinned_rqst(req)) {
1425 spin_lock(&xprt->queue_lock);
1426 xprt_request_dequeue_receive_locked(task);
1427 while (xprt_is_pinned_rqst(req)) {
1428 set_bit(RPC_TASK_MSG_PIN_WAIT, &task->tk_runstate);
1429 spin_unlock(&xprt->queue_lock);
1430 xprt_wait_on_pinned_rqst(req);
1431 spin_lock(&xprt->queue_lock);
1432 clear_bit(RPC_TASK_MSG_PIN_WAIT, &task->tk_runstate);
1434 spin_unlock(&xprt->queue_lock);
1439 * xprt_release - release an RPC request slot
1440 * @task: task which is finished with the slot
1443 void xprt_release(struct rpc_task *task)
1445 struct rpc_xprt *xprt;
1446 struct rpc_rqst *req = task->tk_rqstp;
1449 if (task->tk_client) {
1450 xprt = task->tk_xprt;
1451 if (xprt->snd_task == task)
1452 xprt_release_write(xprt, task);
1457 xprt = req->rq_xprt;
1458 if (task->tk_ops->rpc_count_stats != NULL)
1459 task->tk_ops->rpc_count_stats(task, task->tk_calldata);
1460 else if (task->tk_client)
1461 rpc_count_iostats(task, task->tk_client->cl_metrics);
1462 xprt_request_dequeue_all(task, req);
1463 spin_lock_bh(&xprt->transport_lock);
1464 xprt->ops->release_xprt(xprt, task);
1465 if (xprt->ops->release_request)
1466 xprt->ops->release_request(task);
1467 xprt->last_used = jiffies;
1468 xprt_schedule_autodisconnect(xprt);
1469 spin_unlock_bh(&xprt->transport_lock);
1471 xprt->ops->buf_free(task);
1472 xprt_inject_disconnect(xprt);
1473 if (req->rq_cred != NULL)
1474 put_rpccred(req->rq_cred);
1475 task->tk_rqstp = NULL;
1476 if (req->rq_release_snd_buf)
1477 req->rq_release_snd_buf(req);
1479 dprintk("RPC: %5u release request %p\n", task->tk_pid, req);
1480 if (likely(!bc_prealloc(req)))
1481 xprt->ops->free_slot(xprt, req);
1483 xprt_free_bc_request(req);
1486 static void xprt_init(struct rpc_xprt *xprt, struct net *net)
1488 kref_init(&xprt->kref);
1490 spin_lock_init(&xprt->transport_lock);
1491 spin_lock_init(&xprt->reserve_lock);
1492 spin_lock_init(&xprt->queue_lock);
1494 INIT_LIST_HEAD(&xprt->free);
1495 INIT_LIST_HEAD(&xprt->recv_queue);
1496 #if defined(CONFIG_SUNRPC_BACKCHANNEL)
1497 spin_lock_init(&xprt->bc_pa_lock);
1498 INIT_LIST_HEAD(&xprt->bc_pa_list);
1499 #endif /* CONFIG_SUNRPC_BACKCHANNEL */
1500 INIT_LIST_HEAD(&xprt->xprt_switch);
1502 xprt->last_used = jiffies;
1503 xprt->cwnd = RPC_INITCWND;
1504 xprt->bind_index = 0;
1506 rpc_init_wait_queue(&xprt->binding, "xprt_binding");
1507 rpc_init_wait_queue(&xprt->pending, "xprt_pending");
1508 rpc_init_priority_wait_queue(&xprt->sending, "xprt_sending");
1509 rpc_init_priority_wait_queue(&xprt->backlog, "xprt_backlog");
1511 xprt_init_xid(xprt);
1513 xprt->xprt_net = get_net(net);
1517 * xprt_create_transport - create an RPC transport
1518 * @args: rpc transport creation arguments
1521 struct rpc_xprt *xprt_create_transport(struct xprt_create *args)
1523 struct rpc_xprt *xprt;
1524 struct xprt_class *t;
1526 spin_lock(&xprt_list_lock);
1527 list_for_each_entry(t, &xprt_list, list) {
1528 if (t->ident == args->ident) {
1529 spin_unlock(&xprt_list_lock);
1533 spin_unlock(&xprt_list_lock);
1534 dprintk("RPC: transport (%d) not supported\n", args->ident);
1535 return ERR_PTR(-EIO);
1538 xprt = t->setup(args);
1540 dprintk("RPC: xprt_create_transport: failed, %ld\n",
1544 if (args->flags & XPRT_CREATE_NO_IDLE_TIMEOUT)
1545 xprt->idle_timeout = 0;
1546 INIT_WORK(&xprt->task_cleanup, xprt_autoclose);
1547 if (xprt_has_timer(xprt))
1548 timer_setup(&xprt->timer, xprt_init_autodisconnect, 0);
1550 timer_setup(&xprt->timer, NULL, 0);
1552 if (strlen(args->servername) > RPC_MAXNETNAMELEN) {
1554 return ERR_PTR(-EINVAL);
1556 xprt->servername = kstrdup(args->servername, GFP_KERNEL);
1557 if (xprt->servername == NULL) {
1559 return ERR_PTR(-ENOMEM);
1562 rpc_xprt_debugfs_register(xprt);
1564 dprintk("RPC: created transport %p with %u slots\n", xprt,
1570 static void xprt_destroy_cb(struct work_struct *work)
1572 struct rpc_xprt *xprt =
1573 container_of(work, struct rpc_xprt, task_cleanup);
1575 rpc_xprt_debugfs_unregister(xprt);
1576 rpc_destroy_wait_queue(&xprt->binding);
1577 rpc_destroy_wait_queue(&xprt->pending);
1578 rpc_destroy_wait_queue(&xprt->sending);
1579 rpc_destroy_wait_queue(&xprt->backlog);
1580 kfree(xprt->servername);
1582 * Tear down transport state and free the rpc_xprt
1584 xprt->ops->destroy(xprt);
1588 * xprt_destroy - destroy an RPC transport, killing off all requests.
1589 * @xprt: transport to destroy
1592 static void xprt_destroy(struct rpc_xprt *xprt)
1594 dprintk("RPC: destroying transport %p\n", xprt);
1597 * Exclude transport connect/disconnect handlers and autoclose
1599 wait_on_bit_lock(&xprt->state, XPRT_LOCKED, TASK_UNINTERRUPTIBLE);
1601 del_timer_sync(&xprt->timer);
1604 * Destroy sockets etc from the system workqueue so they can
1605 * safely flush receive work running on rpciod.
1607 INIT_WORK(&xprt->task_cleanup, xprt_destroy_cb);
1608 schedule_work(&xprt->task_cleanup);
1611 static void xprt_destroy_kref(struct kref *kref)
1613 xprt_destroy(container_of(kref, struct rpc_xprt, kref));
1617 * xprt_get - return a reference to an RPC transport.
1618 * @xprt: pointer to the transport
1621 struct rpc_xprt *xprt_get(struct rpc_xprt *xprt)
1623 if (xprt != NULL && kref_get_unless_zero(&xprt->kref))
1627 EXPORT_SYMBOL_GPL(xprt_get);
1630 * xprt_put - release a reference to an RPC transport.
1631 * @xprt: pointer to the transport
1634 void xprt_put(struct rpc_xprt *xprt)
1637 kref_put(&xprt->kref, xprt_destroy_kref);
1639 EXPORT_SYMBOL_GPL(xprt_put);