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