svcrpc: assume svc_delete_xprt() called only once
[linux-2.6.git] / net / sunrpc / svc_xprt.c
1 /*
2  * linux/net/sunrpc/svc_xprt.c
3  *
4  * Author: Tom Tucker <tom@opengridcomputing.com>
5  */
6
7 #include <linux/sched.h>
8 #include <linux/smp_lock.h>
9 #include <linux/errno.h>
10 #include <linux/freezer.h>
11 #include <linux/kthread.h>
12 #include <linux/slab.h>
13 #include <net/sock.h>
14 #include <linux/sunrpc/stats.h>
15 #include <linux/sunrpc/svc_xprt.h>
16 #include <linux/sunrpc/svcsock.h>
17
18 #define RPCDBG_FACILITY RPCDBG_SVCXPRT
19
20 static struct svc_deferred_req *svc_deferred_dequeue(struct svc_xprt *xprt);
21 static int svc_deferred_recv(struct svc_rqst *rqstp);
22 static struct cache_deferred_req *svc_defer(struct cache_req *req);
23 static void svc_age_temp_xprts(unsigned long closure);
24
25 /* apparently the "standard" is that clients close
26  * idle connections after 5 minutes, servers after
27  * 6 minutes
28  *   http://www.connectathon.org/talks96/nfstcp.pdf
29  */
30 static int svc_conn_age_period = 6*60;
31
32 /* List of registered transport classes */
33 static DEFINE_SPINLOCK(svc_xprt_class_lock);
34 static LIST_HEAD(svc_xprt_class_list);
35
36 /* SMP locking strategy:
37  *
38  *      svc_pool->sp_lock protects most of the fields of that pool.
39  *      svc_serv->sv_lock protects sv_tempsocks, sv_permsocks, sv_tmpcnt.
40  *      when both need to be taken (rare), svc_serv->sv_lock is first.
41  *      BKL protects svc_serv->sv_nrthread.
42  *      svc_sock->sk_lock protects the svc_sock->sk_deferred list
43  *             and the ->sk_info_authunix cache.
44  *
45  *      The XPT_BUSY bit in xprt->xpt_flags prevents a transport being
46  *      enqueued multiply. During normal transport processing this bit
47  *      is set by svc_xprt_enqueue and cleared by svc_xprt_received.
48  *      Providers should not manipulate this bit directly.
49  *
50  *      Some flags can be set to certain values at any time
51  *      providing that certain rules are followed:
52  *
53  *      XPT_CONN, XPT_DATA:
54  *              - Can be set or cleared at any time.
55  *              - After a set, svc_xprt_enqueue must be called to enqueue
56  *                the transport for processing.
57  *              - After a clear, the transport must be read/accepted.
58  *                If this succeeds, it must be set again.
59  *      XPT_CLOSE:
60  *              - Can set at any time. It is never cleared.
61  *      XPT_DEAD:
62  *              - Can only be set while XPT_BUSY is held which ensures
63  *                that no other thread will be using the transport or will
64  *                try to set XPT_DEAD.
65  */
66
67 int svc_reg_xprt_class(struct svc_xprt_class *xcl)
68 {
69         struct svc_xprt_class *cl;
70         int res = -EEXIST;
71
72         dprintk("svc: Adding svc transport class '%s'\n", xcl->xcl_name);
73
74         INIT_LIST_HEAD(&xcl->xcl_list);
75         spin_lock(&svc_xprt_class_lock);
76         /* Make sure there isn't already a class with the same name */
77         list_for_each_entry(cl, &svc_xprt_class_list, xcl_list) {
78                 if (strcmp(xcl->xcl_name, cl->xcl_name) == 0)
79                         goto out;
80         }
81         list_add_tail(&xcl->xcl_list, &svc_xprt_class_list);
82         res = 0;
83 out:
84         spin_unlock(&svc_xprt_class_lock);
85         return res;
86 }
87 EXPORT_SYMBOL_GPL(svc_reg_xprt_class);
88
89 void svc_unreg_xprt_class(struct svc_xprt_class *xcl)
90 {
91         dprintk("svc: Removing svc transport class '%s'\n", xcl->xcl_name);
92         spin_lock(&svc_xprt_class_lock);
93         list_del_init(&xcl->xcl_list);
94         spin_unlock(&svc_xprt_class_lock);
95 }
96 EXPORT_SYMBOL_GPL(svc_unreg_xprt_class);
97
98 /*
99  * Format the transport list for printing
100  */
101 int svc_print_xprts(char *buf, int maxlen)
102 {
103         struct svc_xprt_class *xcl;
104         char tmpstr[80];
105         int len = 0;
106         buf[0] = '\0';
107
108         spin_lock(&svc_xprt_class_lock);
109         list_for_each_entry(xcl, &svc_xprt_class_list, xcl_list) {
110                 int slen;
111
112                 sprintf(tmpstr, "%s %d\n", xcl->xcl_name, xcl->xcl_max_payload);
113                 slen = strlen(tmpstr);
114                 if (len + slen > maxlen)
115                         break;
116                 len += slen;
117                 strcat(buf, tmpstr);
118         }
119         spin_unlock(&svc_xprt_class_lock);
120
121         return len;
122 }
123
124 static void svc_xprt_free(struct kref *kref)
125 {
126         struct svc_xprt *xprt =
127                 container_of(kref, struct svc_xprt, xpt_ref);
128         struct module *owner = xprt->xpt_class->xcl_owner;
129         if (test_bit(XPT_CACHE_AUTH, &xprt->xpt_flags))
130                 svcauth_unix_info_release(xprt);
131         put_net(xprt->xpt_net);
132         xprt->xpt_ops->xpo_free(xprt);
133         module_put(owner);
134 }
135
136 void svc_xprt_put(struct svc_xprt *xprt)
137 {
138         kref_put(&xprt->xpt_ref, svc_xprt_free);
139 }
140 EXPORT_SYMBOL_GPL(svc_xprt_put);
141
142 /*
143  * Called by transport drivers to initialize the transport independent
144  * portion of the transport instance.
145  */
146 void svc_xprt_init(struct svc_xprt_class *xcl, struct svc_xprt *xprt,
147                    struct svc_serv *serv)
148 {
149         memset(xprt, 0, sizeof(*xprt));
150         xprt->xpt_class = xcl;
151         xprt->xpt_ops = xcl->xcl_ops;
152         kref_init(&xprt->xpt_ref);
153         xprt->xpt_server = serv;
154         INIT_LIST_HEAD(&xprt->xpt_list);
155         INIT_LIST_HEAD(&xprt->xpt_ready);
156         INIT_LIST_HEAD(&xprt->xpt_deferred);
157         INIT_LIST_HEAD(&xprt->xpt_users);
158         mutex_init(&xprt->xpt_mutex);
159         spin_lock_init(&xprt->xpt_lock);
160         set_bit(XPT_BUSY, &xprt->xpt_flags);
161         rpc_init_wait_queue(&xprt->xpt_bc_pending, "xpt_bc_pending");
162         xprt->xpt_net = get_net(&init_net);
163 }
164 EXPORT_SYMBOL_GPL(svc_xprt_init);
165
166 static struct svc_xprt *__svc_xpo_create(struct svc_xprt_class *xcl,
167                                          struct svc_serv *serv,
168                                          struct net *net,
169                                          const int family,
170                                          const unsigned short port,
171                                          int flags)
172 {
173         struct sockaddr_in sin = {
174                 .sin_family             = AF_INET,
175                 .sin_addr.s_addr        = htonl(INADDR_ANY),
176                 .sin_port               = htons(port),
177         };
178 #if defined(CONFIG_IPV6) || defined(CONFIG_IPV6_MODULE)
179         struct sockaddr_in6 sin6 = {
180                 .sin6_family            = AF_INET6,
181                 .sin6_addr              = IN6ADDR_ANY_INIT,
182                 .sin6_port              = htons(port),
183         };
184 #endif  /* defined(CONFIG_IPV6) || defined(CONFIG_IPV6_MODULE) */
185         struct sockaddr *sap;
186         size_t len;
187
188         switch (family) {
189         case PF_INET:
190                 sap = (struct sockaddr *)&sin;
191                 len = sizeof(sin);
192                 break;
193 #if defined(CONFIG_IPV6) || defined(CONFIG_IPV6_MODULE)
194         case PF_INET6:
195                 sap = (struct sockaddr *)&sin6;
196                 len = sizeof(sin6);
197                 break;
198 #endif  /* defined(CONFIG_IPV6) || defined(CONFIG_IPV6_MODULE) */
199         default:
200                 return ERR_PTR(-EAFNOSUPPORT);
201         }
202
203         return xcl->xcl_ops->xpo_create(serv, net, sap, len, flags);
204 }
205
206 int svc_create_xprt(struct svc_serv *serv, const char *xprt_name,
207                     struct net *net, const int family,
208                     const unsigned short port, int flags)
209 {
210         struct svc_xprt_class *xcl;
211
212         dprintk("svc: creating transport %s[%d]\n", xprt_name, port);
213         spin_lock(&svc_xprt_class_lock);
214         list_for_each_entry(xcl, &svc_xprt_class_list, xcl_list) {
215                 struct svc_xprt *newxprt;
216
217                 if (strcmp(xprt_name, xcl->xcl_name))
218                         continue;
219
220                 if (!try_module_get(xcl->xcl_owner))
221                         goto err;
222
223                 spin_unlock(&svc_xprt_class_lock);
224                 newxprt = __svc_xpo_create(xcl, serv, net, family, port, flags);
225                 if (IS_ERR(newxprt)) {
226                         module_put(xcl->xcl_owner);
227                         return PTR_ERR(newxprt);
228                 }
229
230                 clear_bit(XPT_TEMP, &newxprt->xpt_flags);
231                 spin_lock_bh(&serv->sv_lock);
232                 list_add(&newxprt->xpt_list, &serv->sv_permsocks);
233                 spin_unlock_bh(&serv->sv_lock);
234                 clear_bit(XPT_BUSY, &newxprt->xpt_flags);
235                 return svc_xprt_local_port(newxprt);
236         }
237  err:
238         spin_unlock(&svc_xprt_class_lock);
239         dprintk("svc: transport %s not found\n", xprt_name);
240
241         /* This errno is exposed to user space.  Provide a reasonable
242          * perror msg for a bad transport. */
243         return -EPROTONOSUPPORT;
244 }
245 EXPORT_SYMBOL_GPL(svc_create_xprt);
246
247 /*
248  * Copy the local and remote xprt addresses to the rqstp structure
249  */
250 void svc_xprt_copy_addrs(struct svc_rqst *rqstp, struct svc_xprt *xprt)
251 {
252         struct sockaddr *sin;
253
254         memcpy(&rqstp->rq_addr, &xprt->xpt_remote, xprt->xpt_remotelen);
255         rqstp->rq_addrlen = xprt->xpt_remotelen;
256
257         /*
258          * Destination address in request is needed for binding the
259          * source address in RPC replies/callbacks later.
260          */
261         sin = (struct sockaddr *)&xprt->xpt_local;
262         switch (sin->sa_family) {
263         case AF_INET:
264                 rqstp->rq_daddr.addr = ((struct sockaddr_in *)sin)->sin_addr;
265                 break;
266         case AF_INET6:
267                 rqstp->rq_daddr.addr6 = ((struct sockaddr_in6 *)sin)->sin6_addr;
268                 break;
269         }
270 }
271 EXPORT_SYMBOL_GPL(svc_xprt_copy_addrs);
272
273 /**
274  * svc_print_addr - Format rq_addr field for printing
275  * @rqstp: svc_rqst struct containing address to print
276  * @buf: target buffer for formatted address
277  * @len: length of target buffer
278  *
279  */
280 char *svc_print_addr(struct svc_rqst *rqstp, char *buf, size_t len)
281 {
282         return __svc_print_addr(svc_addr(rqstp), buf, len);
283 }
284 EXPORT_SYMBOL_GPL(svc_print_addr);
285
286 /*
287  * Queue up an idle server thread.  Must have pool->sp_lock held.
288  * Note: this is really a stack rather than a queue, so that we only
289  * use as many different threads as we need, and the rest don't pollute
290  * the cache.
291  */
292 static void svc_thread_enqueue(struct svc_pool *pool, struct svc_rqst *rqstp)
293 {
294         list_add(&rqstp->rq_list, &pool->sp_threads);
295 }
296
297 /*
298  * Dequeue an nfsd thread.  Must have pool->sp_lock held.
299  */
300 static void svc_thread_dequeue(struct svc_pool *pool, struct svc_rqst *rqstp)
301 {
302         list_del(&rqstp->rq_list);
303 }
304
305 /*
306  * Queue up a transport with data pending. If there are idle nfsd
307  * processes, wake 'em up.
308  *
309  */
310 void svc_xprt_enqueue(struct svc_xprt *xprt)
311 {
312         struct svc_serv *serv = xprt->xpt_server;
313         struct svc_pool *pool;
314         struct svc_rqst *rqstp;
315         int cpu;
316
317         if (!(xprt->xpt_flags &
318               ((1<<XPT_CONN)|(1<<XPT_DATA)|(1<<XPT_CLOSE)|(1<<XPT_DEFERRED))))
319                 return;
320
321         cpu = get_cpu();
322         pool = svc_pool_for_cpu(xprt->xpt_server, cpu);
323         put_cpu();
324
325         spin_lock_bh(&pool->sp_lock);
326
327         if (!list_empty(&pool->sp_threads) &&
328             !list_empty(&pool->sp_sockets))
329                 printk(KERN_ERR
330                        "svc_xprt_enqueue: "
331                        "threads and transports both waiting??\n");
332
333         if (test_bit(XPT_DEAD, &xprt->xpt_flags)) {
334                 /* Don't enqueue dead transports */
335                 dprintk("svc: transport %p is dead, not enqueued\n", xprt);
336                 goto out_unlock;
337         }
338
339         pool->sp_stats.packets++;
340
341         /* Mark transport as busy. It will remain in this state until
342          * the provider calls svc_xprt_received. We update XPT_BUSY
343          * atomically because it also guards against trying to enqueue
344          * the transport twice.
345          */
346         if (test_and_set_bit(XPT_BUSY, &xprt->xpt_flags)) {
347                 /* Don't enqueue transport while already enqueued */
348                 dprintk("svc: transport %p busy, not enqueued\n", xprt);
349                 goto out_unlock;
350         }
351         BUG_ON(xprt->xpt_pool != NULL);
352         xprt->xpt_pool = pool;
353
354         /* Handle pending connection */
355         if (test_bit(XPT_CONN, &xprt->xpt_flags))
356                 goto process;
357
358         /* Handle close in-progress */
359         if (test_bit(XPT_CLOSE, &xprt->xpt_flags))
360                 goto process;
361
362         /* Check if we have space to reply to a request */
363         if (!xprt->xpt_ops->xpo_has_wspace(xprt)) {
364                 /* Don't enqueue while not enough space for reply */
365                 dprintk("svc: no write space, transport %p  not enqueued\n",
366                         xprt);
367                 xprt->xpt_pool = NULL;
368                 clear_bit(XPT_BUSY, &xprt->xpt_flags);
369                 goto out_unlock;
370         }
371
372  process:
373         if (!list_empty(&pool->sp_threads)) {
374                 rqstp = list_entry(pool->sp_threads.next,
375                                    struct svc_rqst,
376                                    rq_list);
377                 dprintk("svc: transport %p served by daemon %p\n",
378                         xprt, rqstp);
379                 svc_thread_dequeue(pool, rqstp);
380                 if (rqstp->rq_xprt)
381                         printk(KERN_ERR
382                                 "svc_xprt_enqueue: server %p, rq_xprt=%p!\n",
383                                 rqstp, rqstp->rq_xprt);
384                 rqstp->rq_xprt = xprt;
385                 svc_xprt_get(xprt);
386                 rqstp->rq_reserved = serv->sv_max_mesg;
387                 atomic_add(rqstp->rq_reserved, &xprt->xpt_reserved);
388                 pool->sp_stats.threads_woken++;
389                 BUG_ON(xprt->xpt_pool != pool);
390                 wake_up(&rqstp->rq_wait);
391         } else {
392                 dprintk("svc: transport %p put into queue\n", xprt);
393                 list_add_tail(&xprt->xpt_ready, &pool->sp_sockets);
394                 pool->sp_stats.sockets_queued++;
395                 BUG_ON(xprt->xpt_pool != pool);
396         }
397
398 out_unlock:
399         spin_unlock_bh(&pool->sp_lock);
400 }
401 EXPORT_SYMBOL_GPL(svc_xprt_enqueue);
402
403 /*
404  * Dequeue the first transport.  Must be called with the pool->sp_lock held.
405  */
406 static struct svc_xprt *svc_xprt_dequeue(struct svc_pool *pool)
407 {
408         struct svc_xprt *xprt;
409
410         if (list_empty(&pool->sp_sockets))
411                 return NULL;
412
413         xprt = list_entry(pool->sp_sockets.next,
414                           struct svc_xprt, xpt_ready);
415         list_del_init(&xprt->xpt_ready);
416
417         dprintk("svc: transport %p dequeued, inuse=%d\n",
418                 xprt, atomic_read(&xprt->xpt_ref.refcount));
419
420         return xprt;
421 }
422
423 /*
424  * svc_xprt_received conditionally queues the transport for processing
425  * by another thread. The caller must hold the XPT_BUSY bit and must
426  * not thereafter touch transport data.
427  *
428  * Note: XPT_DATA only gets cleared when a read-attempt finds no (or
429  * insufficient) data.
430  */
431 void svc_xprt_received(struct svc_xprt *xprt)
432 {
433         BUG_ON(!test_bit(XPT_BUSY, &xprt->xpt_flags));
434         xprt->xpt_pool = NULL;
435         clear_bit(XPT_BUSY, &xprt->xpt_flags);
436         svc_xprt_enqueue(xprt);
437 }
438 EXPORT_SYMBOL_GPL(svc_xprt_received);
439
440 /**
441  * svc_reserve - change the space reserved for the reply to a request.
442  * @rqstp:  The request in question
443  * @space: new max space to reserve
444  *
445  * Each request reserves some space on the output queue of the transport
446  * to make sure the reply fits.  This function reduces that reserved
447  * space to be the amount of space used already, plus @space.
448  *
449  */
450 void svc_reserve(struct svc_rqst *rqstp, int space)
451 {
452         space += rqstp->rq_res.head[0].iov_len;
453
454         if (space < rqstp->rq_reserved) {
455                 struct svc_xprt *xprt = rqstp->rq_xprt;
456                 atomic_sub((rqstp->rq_reserved - space), &xprt->xpt_reserved);
457                 rqstp->rq_reserved = space;
458
459                 svc_xprt_enqueue(xprt);
460         }
461 }
462 EXPORT_SYMBOL_GPL(svc_reserve);
463
464 static void svc_xprt_release(struct svc_rqst *rqstp)
465 {
466         struct svc_xprt *xprt = rqstp->rq_xprt;
467
468         rqstp->rq_xprt->xpt_ops->xpo_release_rqst(rqstp);
469
470         kfree(rqstp->rq_deferred);
471         rqstp->rq_deferred = NULL;
472
473         svc_free_res_pages(rqstp);
474         rqstp->rq_res.page_len = 0;
475         rqstp->rq_res.page_base = 0;
476
477         /* Reset response buffer and release
478          * the reservation.
479          * But first, check that enough space was reserved
480          * for the reply, otherwise we have a bug!
481          */
482         if ((rqstp->rq_res.len) >  rqstp->rq_reserved)
483                 printk(KERN_ERR "RPC request reserved %d but used %d\n",
484                        rqstp->rq_reserved,
485                        rqstp->rq_res.len);
486
487         rqstp->rq_res.head[0].iov_len = 0;
488         svc_reserve(rqstp, 0);
489         rqstp->rq_xprt = NULL;
490
491         svc_xprt_put(xprt);
492 }
493
494 /*
495  * External function to wake up a server waiting for data
496  * This really only makes sense for services like lockd
497  * which have exactly one thread anyway.
498  */
499 void svc_wake_up(struct svc_serv *serv)
500 {
501         struct svc_rqst *rqstp;
502         unsigned int i;
503         struct svc_pool *pool;
504
505         for (i = 0; i < serv->sv_nrpools; i++) {
506                 pool = &serv->sv_pools[i];
507
508                 spin_lock_bh(&pool->sp_lock);
509                 if (!list_empty(&pool->sp_threads)) {
510                         rqstp = list_entry(pool->sp_threads.next,
511                                            struct svc_rqst,
512                                            rq_list);
513                         dprintk("svc: daemon %p woken up.\n", rqstp);
514                         /*
515                         svc_thread_dequeue(pool, rqstp);
516                         rqstp->rq_xprt = NULL;
517                          */
518                         wake_up(&rqstp->rq_wait);
519                 }
520                 spin_unlock_bh(&pool->sp_lock);
521         }
522 }
523 EXPORT_SYMBOL_GPL(svc_wake_up);
524
525 int svc_port_is_privileged(struct sockaddr *sin)
526 {
527         switch (sin->sa_family) {
528         case AF_INET:
529                 return ntohs(((struct sockaddr_in *)sin)->sin_port)
530                         < PROT_SOCK;
531         case AF_INET6:
532                 return ntohs(((struct sockaddr_in6 *)sin)->sin6_port)
533                         < PROT_SOCK;
534         default:
535                 return 0;
536         }
537 }
538
539 /*
540  * Make sure that we don't have too many active connections. If we have,
541  * something must be dropped. It's not clear what will happen if we allow
542  * "too many" connections, but when dealing with network-facing software,
543  * we have to code defensively. Here we do that by imposing hard limits.
544  *
545  * There's no point in trying to do random drop here for DoS
546  * prevention. The NFS clients does 1 reconnect in 15 seconds. An
547  * attacker can easily beat that.
548  *
549  * The only somewhat efficient mechanism would be if drop old
550  * connections from the same IP first. But right now we don't even
551  * record the client IP in svc_sock.
552  *
553  * single-threaded services that expect a lot of clients will probably
554  * need to set sv_maxconn to override the default value which is based
555  * on the number of threads
556  */
557 static void svc_check_conn_limits(struct svc_serv *serv)
558 {
559         unsigned int limit = serv->sv_maxconn ? serv->sv_maxconn :
560                                 (serv->sv_nrthreads+3) * 20;
561
562         if (serv->sv_tmpcnt > limit) {
563                 struct svc_xprt *xprt = NULL;
564                 spin_lock_bh(&serv->sv_lock);
565                 if (!list_empty(&serv->sv_tempsocks)) {
566                         if (net_ratelimit()) {
567                                 /* Try to help the admin */
568                                 printk(KERN_NOTICE "%s: too many open  "
569                                        "connections, consider increasing %s\n",
570                                        serv->sv_name, serv->sv_maxconn ?
571                                        "the max number of connections." :
572                                        "the number of threads.");
573                         }
574                         /*
575                          * Always select the oldest connection. It's not fair,
576                          * but so is life
577                          */
578                         xprt = list_entry(serv->sv_tempsocks.prev,
579                                           struct svc_xprt,
580                                           xpt_list);
581                         set_bit(XPT_CLOSE, &xprt->xpt_flags);
582                         svc_xprt_get(xprt);
583                 }
584                 spin_unlock_bh(&serv->sv_lock);
585
586                 if (xprt) {
587                         svc_xprt_enqueue(xprt);
588                         svc_xprt_put(xprt);
589                 }
590         }
591 }
592
593 /*
594  * Receive the next request on any transport.  This code is carefully
595  * organised not to touch any cachelines in the shared svc_serv
596  * structure, only cachelines in the local svc_pool.
597  */
598 int svc_recv(struct svc_rqst *rqstp, long timeout)
599 {
600         struct svc_xprt         *xprt = NULL;
601         struct svc_serv         *serv = rqstp->rq_server;
602         struct svc_pool         *pool = rqstp->rq_pool;
603         int                     len, i;
604         int                     pages;
605         struct xdr_buf          *arg;
606         DECLARE_WAITQUEUE(wait, current);
607         long                    time_left;
608
609         dprintk("svc: server %p waiting for data (to = %ld)\n",
610                 rqstp, timeout);
611
612         if (rqstp->rq_xprt)
613                 printk(KERN_ERR
614                         "svc_recv: service %p, transport not NULL!\n",
615                          rqstp);
616         if (waitqueue_active(&rqstp->rq_wait))
617                 printk(KERN_ERR
618                         "svc_recv: service %p, wait queue active!\n",
619                          rqstp);
620
621         /* now allocate needed pages.  If we get a failure, sleep briefly */
622         pages = (serv->sv_max_mesg + PAGE_SIZE) / PAGE_SIZE;
623         for (i = 0; i < pages ; i++)
624                 while (rqstp->rq_pages[i] == NULL) {
625                         struct page *p = alloc_page(GFP_KERNEL);
626                         if (!p) {
627                                 set_current_state(TASK_INTERRUPTIBLE);
628                                 if (signalled() || kthread_should_stop()) {
629                                         set_current_state(TASK_RUNNING);
630                                         return -EINTR;
631                                 }
632                                 schedule_timeout(msecs_to_jiffies(500));
633                         }
634                         rqstp->rq_pages[i] = p;
635                 }
636         rqstp->rq_pages[i++] = NULL; /* this might be seen in nfs_read_actor */
637         BUG_ON(pages >= RPCSVC_MAXPAGES);
638
639         /* Make arg->head point to first page and arg->pages point to rest */
640         arg = &rqstp->rq_arg;
641         arg->head[0].iov_base = page_address(rqstp->rq_pages[0]);
642         arg->head[0].iov_len = PAGE_SIZE;
643         arg->pages = rqstp->rq_pages + 1;
644         arg->page_base = 0;
645         /* save at least one page for response */
646         arg->page_len = (pages-2)*PAGE_SIZE;
647         arg->len = (pages-1)*PAGE_SIZE;
648         arg->tail[0].iov_len = 0;
649
650         try_to_freeze();
651         cond_resched();
652         if (signalled() || kthread_should_stop())
653                 return -EINTR;
654
655         /* Normally we will wait up to 5 seconds for any required
656          * cache information to be provided.
657          */
658         rqstp->rq_chandle.thread_wait = 5*HZ;
659
660         spin_lock_bh(&pool->sp_lock);
661         xprt = svc_xprt_dequeue(pool);
662         if (xprt) {
663                 rqstp->rq_xprt = xprt;
664                 svc_xprt_get(xprt);
665                 rqstp->rq_reserved = serv->sv_max_mesg;
666                 atomic_add(rqstp->rq_reserved, &xprt->xpt_reserved);
667
668                 /* As there is a shortage of threads and this request
669                  * had to be queued, don't allow the thread to wait so
670                  * long for cache updates.
671                  */
672                 rqstp->rq_chandle.thread_wait = 1*HZ;
673         } else {
674                 /* No data pending. Go to sleep */
675                 svc_thread_enqueue(pool, rqstp);
676
677                 /*
678                  * We have to be able to interrupt this wait
679                  * to bring down the daemons ...
680                  */
681                 set_current_state(TASK_INTERRUPTIBLE);
682
683                 /*
684                  * checking kthread_should_stop() here allows us to avoid
685                  * locking and signalling when stopping kthreads that call
686                  * svc_recv. If the thread has already been woken up, then
687                  * we can exit here without sleeping. If not, then it
688                  * it'll be woken up quickly during the schedule_timeout
689                  */
690                 if (kthread_should_stop()) {
691                         set_current_state(TASK_RUNNING);
692                         spin_unlock_bh(&pool->sp_lock);
693                         return -EINTR;
694                 }
695
696                 add_wait_queue(&rqstp->rq_wait, &wait);
697                 spin_unlock_bh(&pool->sp_lock);
698
699                 time_left = schedule_timeout(timeout);
700
701                 try_to_freeze();
702
703                 spin_lock_bh(&pool->sp_lock);
704                 remove_wait_queue(&rqstp->rq_wait, &wait);
705                 if (!time_left)
706                         pool->sp_stats.threads_timedout++;
707
708                 xprt = rqstp->rq_xprt;
709                 if (!xprt) {
710                         svc_thread_dequeue(pool, rqstp);
711                         spin_unlock_bh(&pool->sp_lock);
712                         dprintk("svc: server %p, no data yet\n", rqstp);
713                         if (signalled() || kthread_should_stop())
714                                 return -EINTR;
715                         else
716                                 return -EAGAIN;
717                 }
718         }
719         spin_unlock_bh(&pool->sp_lock);
720
721         len = 0;
722         if (test_bit(XPT_CLOSE, &xprt->xpt_flags)) {
723                 dprintk("svc_recv: found XPT_CLOSE\n");
724                 svc_delete_xprt(xprt);
725         } else if (test_bit(XPT_LISTENER, &xprt->xpt_flags)) {
726                 struct svc_xprt *newxpt;
727                 newxpt = xprt->xpt_ops->xpo_accept(xprt);
728                 if (newxpt) {
729                         /*
730                          * We know this module_get will succeed because the
731                          * listener holds a reference too
732                          */
733                         __module_get(newxpt->xpt_class->xcl_owner);
734                         svc_check_conn_limits(xprt->xpt_server);
735                         spin_lock_bh(&serv->sv_lock);
736                         set_bit(XPT_TEMP, &newxpt->xpt_flags);
737                         list_add(&newxpt->xpt_list, &serv->sv_tempsocks);
738                         serv->sv_tmpcnt++;
739                         if (serv->sv_temptimer.function == NULL) {
740                                 /* setup timer to age temp transports */
741                                 setup_timer(&serv->sv_temptimer,
742                                             svc_age_temp_xprts,
743                                             (unsigned long)serv);
744                                 mod_timer(&serv->sv_temptimer,
745                                           jiffies + svc_conn_age_period * HZ);
746                         }
747                         spin_unlock_bh(&serv->sv_lock);
748                         svc_xprt_received(newxpt);
749                 }
750                 svc_xprt_received(xprt);
751         } else {
752                 dprintk("svc: server %p, pool %u, transport %p, inuse=%d\n",
753                         rqstp, pool->sp_id, xprt,
754                         atomic_read(&xprt->xpt_ref.refcount));
755                 rqstp->rq_deferred = svc_deferred_dequeue(xprt);
756                 if (rqstp->rq_deferred) {
757                         svc_xprt_received(xprt);
758                         len = svc_deferred_recv(rqstp);
759                 } else {
760                         len = xprt->xpt_ops->xpo_recvfrom(rqstp);
761                         svc_xprt_received(xprt);
762                 }
763                 dprintk("svc: got len=%d\n", len);
764         }
765
766         /* No data, incomplete (TCP) read, or accept() */
767         if (len == 0 || len == -EAGAIN) {
768                 rqstp->rq_res.len = 0;
769                 svc_xprt_release(rqstp);
770                 return -EAGAIN;
771         }
772         clear_bit(XPT_OLD, &xprt->xpt_flags);
773
774         rqstp->rq_secure = svc_port_is_privileged(svc_addr(rqstp));
775         rqstp->rq_chandle.defer = svc_defer;
776
777         if (serv->sv_stats)
778                 serv->sv_stats->netcnt++;
779         return len;
780 }
781 EXPORT_SYMBOL_GPL(svc_recv);
782
783 /*
784  * Drop request
785  */
786 void svc_drop(struct svc_rqst *rqstp)
787 {
788         dprintk("svc: xprt %p dropped request\n", rqstp->rq_xprt);
789         svc_xprt_release(rqstp);
790 }
791 EXPORT_SYMBOL_GPL(svc_drop);
792
793 /*
794  * Return reply to client.
795  */
796 int svc_send(struct svc_rqst *rqstp)
797 {
798         struct svc_xprt *xprt;
799         int             len;
800         struct xdr_buf  *xb;
801
802         xprt = rqstp->rq_xprt;
803         if (!xprt)
804                 return -EFAULT;
805
806         /* release the receive skb before sending the reply */
807         rqstp->rq_xprt->xpt_ops->xpo_release_rqst(rqstp);
808
809         /* calculate over-all length */
810         xb = &rqstp->rq_res;
811         xb->len = xb->head[0].iov_len +
812                 xb->page_len +
813                 xb->tail[0].iov_len;
814
815         /* Grab mutex to serialize outgoing data. */
816         mutex_lock(&xprt->xpt_mutex);
817         if (test_bit(XPT_DEAD, &xprt->xpt_flags))
818                 len = -ENOTCONN;
819         else
820                 len = xprt->xpt_ops->xpo_sendto(rqstp);
821         mutex_unlock(&xprt->xpt_mutex);
822         rpc_wake_up(&xprt->xpt_bc_pending);
823         svc_xprt_release(rqstp);
824
825         if (len == -ECONNREFUSED || len == -ENOTCONN || len == -EAGAIN)
826                 return 0;
827         return len;
828 }
829
830 /*
831  * Timer function to close old temporary transports, using
832  * a mark-and-sweep algorithm.
833  */
834 static void svc_age_temp_xprts(unsigned long closure)
835 {
836         struct svc_serv *serv = (struct svc_serv *)closure;
837         struct svc_xprt *xprt;
838         struct list_head *le, *next;
839         LIST_HEAD(to_be_aged);
840
841         dprintk("svc_age_temp_xprts\n");
842
843         if (!spin_trylock_bh(&serv->sv_lock)) {
844                 /* busy, try again 1 sec later */
845                 dprintk("svc_age_temp_xprts: busy\n");
846                 mod_timer(&serv->sv_temptimer, jiffies + HZ);
847                 return;
848         }
849
850         list_for_each_safe(le, next, &serv->sv_tempsocks) {
851                 xprt = list_entry(le, struct svc_xprt, xpt_list);
852
853                 /* First time through, just mark it OLD. Second time
854                  * through, close it. */
855                 if (!test_and_set_bit(XPT_OLD, &xprt->xpt_flags))
856                         continue;
857                 if (atomic_read(&xprt->xpt_ref.refcount) > 1 ||
858                     test_bit(XPT_BUSY, &xprt->xpt_flags))
859                         continue;
860                 svc_xprt_get(xprt);
861                 list_move(le, &to_be_aged);
862                 set_bit(XPT_CLOSE, &xprt->xpt_flags);
863                 set_bit(XPT_DETACHED, &xprt->xpt_flags);
864         }
865         spin_unlock_bh(&serv->sv_lock);
866
867         while (!list_empty(&to_be_aged)) {
868                 le = to_be_aged.next;
869                 /* fiddling the xpt_list node is safe 'cos we're XPT_DETACHED */
870                 list_del_init(le);
871                 xprt = list_entry(le, struct svc_xprt, xpt_list);
872
873                 dprintk("queuing xprt %p for closing\n", xprt);
874
875                 /* a thread will dequeue and close it soon */
876                 svc_xprt_enqueue(xprt);
877                 svc_xprt_put(xprt);
878         }
879
880         mod_timer(&serv->sv_temptimer, jiffies + svc_conn_age_period * HZ);
881 }
882
883 static void call_xpt_users(struct svc_xprt *xprt)
884 {
885         struct svc_xpt_user *u;
886
887         spin_lock(&xprt->xpt_lock);
888         while (!list_empty(&xprt->xpt_users)) {
889                 u = list_first_entry(&xprt->xpt_users, struct svc_xpt_user, list);
890                 list_del(&u->list);
891                 u->callback(u);
892         }
893         spin_unlock(&xprt->xpt_lock);
894 }
895
896 /*
897  * Remove a dead transport
898  */
899 void svc_delete_xprt(struct svc_xprt *xprt)
900 {
901         struct svc_serv *serv = xprt->xpt_server;
902         struct svc_deferred_req *dr;
903
904         /* Only do this once */
905         if (test_and_set_bit(XPT_DEAD, &xprt->xpt_flags))
906                 BUG();
907
908         dprintk("svc: svc_delete_xprt(%p)\n", xprt);
909         xprt->xpt_ops->xpo_detach(xprt);
910
911         spin_lock_bh(&serv->sv_lock);
912         if (!test_and_set_bit(XPT_DETACHED, &xprt->xpt_flags))
913                 list_del_init(&xprt->xpt_list);
914         /*
915          * We used to delete the transport from whichever list
916          * it's sk_xprt.xpt_ready node was on, but we don't actually
917          * need to.  This is because the only time we're called
918          * while still attached to a queue, the queue itself
919          * is about to be destroyed (in svc_destroy).
920          */
921         if (test_bit(XPT_TEMP, &xprt->xpt_flags))
922                 serv->sv_tmpcnt--;
923         spin_unlock_bh(&serv->sv_lock);
924
925         while ((dr = svc_deferred_dequeue(xprt)) != NULL)
926                 kfree(dr);
927
928         call_xpt_users(xprt);
929         svc_xprt_put(xprt);
930 }
931
932 void svc_close_xprt(struct svc_xprt *xprt)
933 {
934         set_bit(XPT_CLOSE, &xprt->xpt_flags);
935         if (test_and_set_bit(XPT_BUSY, &xprt->xpt_flags))
936                 /* someone else will have to effect the close */
937                 return;
938
939         svc_delete_xprt(xprt);
940 }
941 EXPORT_SYMBOL_GPL(svc_close_xprt);
942
943 void svc_close_all(struct list_head *xprt_list)
944 {
945         struct svc_xprt *xprt;
946         struct svc_xprt *tmp;
947
948         list_for_each_entry_safe(xprt, tmp, xprt_list, xpt_list) {
949                 set_bit(XPT_CLOSE, &xprt->xpt_flags);
950                 if (test_bit(XPT_BUSY, &xprt->xpt_flags)) {
951                         /* Waiting to be processed, but no threads left,
952                          * So just remove it from the waiting list
953                          */
954                         list_del_init(&xprt->xpt_ready);
955                         clear_bit(XPT_BUSY, &xprt->xpt_flags);
956                 }
957                 svc_close_xprt(xprt);
958         }
959 }
960
961 /*
962  * Handle defer and revisit of requests
963  */
964
965 static void svc_revisit(struct cache_deferred_req *dreq, int too_many)
966 {
967         struct svc_deferred_req *dr =
968                 container_of(dreq, struct svc_deferred_req, handle);
969         struct svc_xprt *xprt = dr->xprt;
970
971         spin_lock(&xprt->xpt_lock);
972         set_bit(XPT_DEFERRED, &xprt->xpt_flags);
973         if (too_many || test_bit(XPT_DEAD, &xprt->xpt_flags)) {
974                 spin_unlock(&xprt->xpt_lock);
975                 dprintk("revisit canceled\n");
976                 svc_xprt_put(xprt);
977                 kfree(dr);
978                 return;
979         }
980         dprintk("revisit queued\n");
981         dr->xprt = NULL;
982         list_add(&dr->handle.recent, &xprt->xpt_deferred);
983         spin_unlock(&xprt->xpt_lock);
984         svc_xprt_enqueue(xprt);
985         svc_xprt_put(xprt);
986 }
987
988 /*
989  * Save the request off for later processing. The request buffer looks
990  * like this:
991  *
992  * <xprt-header><rpc-header><rpc-pagelist><rpc-tail>
993  *
994  * This code can only handle requests that consist of an xprt-header
995  * and rpc-header.
996  */
997 static struct cache_deferred_req *svc_defer(struct cache_req *req)
998 {
999         struct svc_rqst *rqstp = container_of(req, struct svc_rqst, rq_chandle);
1000         struct svc_deferred_req *dr;
1001
1002         if (rqstp->rq_arg.page_len || !rqstp->rq_usedeferral)
1003                 return NULL; /* if more than a page, give up FIXME */
1004         if (rqstp->rq_deferred) {
1005                 dr = rqstp->rq_deferred;
1006                 rqstp->rq_deferred = NULL;
1007         } else {
1008                 size_t skip;
1009                 size_t size;
1010                 /* FIXME maybe discard if size too large */
1011                 size = sizeof(struct svc_deferred_req) + rqstp->rq_arg.len;
1012                 dr = kmalloc(size, GFP_KERNEL);
1013                 if (dr == NULL)
1014                         return NULL;
1015
1016                 dr->handle.owner = rqstp->rq_server;
1017                 dr->prot = rqstp->rq_prot;
1018                 memcpy(&dr->addr, &rqstp->rq_addr, rqstp->rq_addrlen);
1019                 dr->addrlen = rqstp->rq_addrlen;
1020                 dr->daddr = rqstp->rq_daddr;
1021                 dr->argslen = rqstp->rq_arg.len >> 2;
1022                 dr->xprt_hlen = rqstp->rq_xprt_hlen;
1023
1024                 /* back up head to the start of the buffer and copy */
1025                 skip = rqstp->rq_arg.len - rqstp->rq_arg.head[0].iov_len;
1026                 memcpy(dr->args, rqstp->rq_arg.head[0].iov_base - skip,
1027                        dr->argslen << 2);
1028         }
1029         svc_xprt_get(rqstp->rq_xprt);
1030         dr->xprt = rqstp->rq_xprt;
1031
1032         dr->handle.revisit = svc_revisit;
1033         return &dr->handle;
1034 }
1035
1036 /*
1037  * recv data from a deferred request into an active one
1038  */
1039 static int svc_deferred_recv(struct svc_rqst *rqstp)
1040 {
1041         struct svc_deferred_req *dr = rqstp->rq_deferred;
1042
1043         /* setup iov_base past transport header */
1044         rqstp->rq_arg.head[0].iov_base = dr->args + (dr->xprt_hlen>>2);
1045         /* The iov_len does not include the transport header bytes */
1046         rqstp->rq_arg.head[0].iov_len = (dr->argslen<<2) - dr->xprt_hlen;
1047         rqstp->rq_arg.page_len = 0;
1048         /* The rq_arg.len includes the transport header bytes */
1049         rqstp->rq_arg.len     = dr->argslen<<2;
1050         rqstp->rq_prot        = dr->prot;
1051         memcpy(&rqstp->rq_addr, &dr->addr, dr->addrlen);
1052         rqstp->rq_addrlen     = dr->addrlen;
1053         /* Save off transport header len in case we get deferred again */
1054         rqstp->rq_xprt_hlen   = dr->xprt_hlen;
1055         rqstp->rq_daddr       = dr->daddr;
1056         rqstp->rq_respages    = rqstp->rq_pages;
1057         return (dr->argslen<<2) - dr->xprt_hlen;
1058 }
1059
1060
1061 static struct svc_deferred_req *svc_deferred_dequeue(struct svc_xprt *xprt)
1062 {
1063         struct svc_deferred_req *dr = NULL;
1064
1065         if (!test_bit(XPT_DEFERRED, &xprt->xpt_flags))
1066                 return NULL;
1067         spin_lock(&xprt->xpt_lock);
1068         clear_bit(XPT_DEFERRED, &xprt->xpt_flags);
1069         if (!list_empty(&xprt->xpt_deferred)) {
1070                 dr = list_entry(xprt->xpt_deferred.next,
1071                                 struct svc_deferred_req,
1072                                 handle.recent);
1073                 list_del_init(&dr->handle.recent);
1074                 set_bit(XPT_DEFERRED, &xprt->xpt_flags);
1075         }
1076         spin_unlock(&xprt->xpt_lock);
1077         return dr;
1078 }
1079
1080 /**
1081  * svc_find_xprt - find an RPC transport instance
1082  * @serv: pointer to svc_serv to search
1083  * @xcl_name: C string containing transport's class name
1084  * @af: Address family of transport's local address
1085  * @port: transport's IP port number
1086  *
1087  * Return the transport instance pointer for the endpoint accepting
1088  * connections/peer traffic from the specified transport class,
1089  * address family and port.
1090  *
1091  * Specifying 0 for the address family or port is effectively a
1092  * wild-card, and will result in matching the first transport in the
1093  * service's list that has a matching class name.
1094  */
1095 struct svc_xprt *svc_find_xprt(struct svc_serv *serv, const char *xcl_name,
1096                                const sa_family_t af, const unsigned short port)
1097 {
1098         struct svc_xprt *xprt;
1099         struct svc_xprt *found = NULL;
1100
1101         /* Sanity check the args */
1102         if (serv == NULL || xcl_name == NULL)
1103                 return found;
1104
1105         spin_lock_bh(&serv->sv_lock);
1106         list_for_each_entry(xprt, &serv->sv_permsocks, xpt_list) {
1107                 if (strcmp(xprt->xpt_class->xcl_name, xcl_name))
1108                         continue;
1109                 if (af != AF_UNSPEC && af != xprt->xpt_local.ss_family)
1110                         continue;
1111                 if (port != 0 && port != svc_xprt_local_port(xprt))
1112                         continue;
1113                 found = xprt;
1114                 svc_xprt_get(xprt);
1115                 break;
1116         }
1117         spin_unlock_bh(&serv->sv_lock);
1118         return found;
1119 }
1120 EXPORT_SYMBOL_GPL(svc_find_xprt);
1121
1122 static int svc_one_xprt_name(const struct svc_xprt *xprt,
1123                              char *pos, int remaining)
1124 {
1125         int len;
1126
1127         len = snprintf(pos, remaining, "%s %u\n",
1128                         xprt->xpt_class->xcl_name,
1129                         svc_xprt_local_port(xprt));
1130         if (len >= remaining)
1131                 return -ENAMETOOLONG;
1132         return len;
1133 }
1134
1135 /**
1136  * svc_xprt_names - format a buffer with a list of transport names
1137  * @serv: pointer to an RPC service
1138  * @buf: pointer to a buffer to be filled in
1139  * @buflen: length of buffer to be filled in
1140  *
1141  * Fills in @buf with a string containing a list of transport names,
1142  * each name terminated with '\n'.
1143  *
1144  * Returns positive length of the filled-in string on success; otherwise
1145  * a negative errno value is returned if an error occurs.
1146  */
1147 int svc_xprt_names(struct svc_serv *serv, char *buf, const int buflen)
1148 {
1149         struct svc_xprt *xprt;
1150         int len, totlen;
1151         char *pos;
1152
1153         /* Sanity check args */
1154         if (!serv)
1155                 return 0;
1156
1157         spin_lock_bh(&serv->sv_lock);
1158
1159         pos = buf;
1160         totlen = 0;
1161         list_for_each_entry(xprt, &serv->sv_permsocks, xpt_list) {
1162                 len = svc_one_xprt_name(xprt, pos, buflen - totlen);
1163                 if (len < 0) {
1164                         *buf = '\0';
1165                         totlen = len;
1166                 }
1167                 if (len <= 0)
1168                         break;
1169
1170                 pos += len;
1171                 totlen += len;
1172         }
1173
1174         spin_unlock_bh(&serv->sv_lock);
1175         return totlen;
1176 }
1177 EXPORT_SYMBOL_GPL(svc_xprt_names);
1178
1179
1180 /*----------------------------------------------------------------------------*/
1181
1182 static void *svc_pool_stats_start(struct seq_file *m, loff_t *pos)
1183 {
1184         unsigned int pidx = (unsigned int)*pos;
1185         struct svc_serv *serv = m->private;
1186
1187         dprintk("svc_pool_stats_start, *pidx=%u\n", pidx);
1188
1189         if (!pidx)
1190                 return SEQ_START_TOKEN;
1191         return (pidx > serv->sv_nrpools ? NULL : &serv->sv_pools[pidx-1]);
1192 }
1193
1194 static void *svc_pool_stats_next(struct seq_file *m, void *p, loff_t *pos)
1195 {
1196         struct svc_pool *pool = p;
1197         struct svc_serv *serv = m->private;
1198
1199         dprintk("svc_pool_stats_next, *pos=%llu\n", *pos);
1200
1201         if (p == SEQ_START_TOKEN) {
1202                 pool = &serv->sv_pools[0];
1203         } else {
1204                 unsigned int pidx = (pool - &serv->sv_pools[0]);
1205                 if (pidx < serv->sv_nrpools-1)
1206                         pool = &serv->sv_pools[pidx+1];
1207                 else
1208                         pool = NULL;
1209         }
1210         ++*pos;
1211         return pool;
1212 }
1213
1214 static void svc_pool_stats_stop(struct seq_file *m, void *p)
1215 {
1216 }
1217
1218 static int svc_pool_stats_show(struct seq_file *m, void *p)
1219 {
1220         struct svc_pool *pool = p;
1221
1222         if (p == SEQ_START_TOKEN) {
1223                 seq_puts(m, "# pool packets-arrived sockets-enqueued threads-woken threads-timedout\n");
1224                 return 0;
1225         }
1226
1227         seq_printf(m, "%u %lu %lu %lu %lu\n",
1228                 pool->sp_id,
1229                 pool->sp_stats.packets,
1230                 pool->sp_stats.sockets_queued,
1231                 pool->sp_stats.threads_woken,
1232                 pool->sp_stats.threads_timedout);
1233
1234         return 0;
1235 }
1236
1237 static const struct seq_operations svc_pool_stats_seq_ops = {
1238         .start  = svc_pool_stats_start,
1239         .next   = svc_pool_stats_next,
1240         .stop   = svc_pool_stats_stop,
1241         .show   = svc_pool_stats_show,
1242 };
1243
1244 int svc_pool_stats_open(struct svc_serv *serv, struct file *file)
1245 {
1246         int err;
1247
1248         err = seq_open(file, &svc_pool_stats_seq_ops);
1249         if (!err)
1250                 ((struct seq_file *) file->private_data)->private = serv;
1251         return err;
1252 }
1253 EXPORT_SYMBOL(svc_pool_stats_open);
1254
1255 /*----------------------------------------------------------------------------*/