08de328ce43304b0890253fe343d78e726612ad4
[linux-2.6.git] / net / sunrpc / svcsock.c
1 /*
2  * linux/net/sunrpc/svcsock.c
3  *
4  * These are the RPC server socket internals.
5  *
6  * The server scheduling algorithm does not always distribute the load
7  * evenly when servicing a single client. May need to modify the
8  * svc_sock_enqueue procedure...
9  *
10  * TCP support is largely untested and may be a little slow. The problem
11  * is that we currently do two separate recvfrom's, one for the 4-byte
12  * record length, and the second for the actual record. This could possibly
13  * be improved by always reading a minimum size of around 100 bytes and
14  * tucking any superfluous bytes away in a temporary store. Still, that
15  * leaves write requests out in the rain. An alternative may be to peek at
16  * the first skb in the queue, and if it matches the next TCP sequence
17  * number, to extract the record marker. Yuck.
18  *
19  * Copyright (C) 1995, 1996 Olaf Kirch <okir@monad.swb.de>
20  */
21
22 #include <linux/sched.h>
23 #include <linux/errno.h>
24 #include <linux/fcntl.h>
25 #include <linux/net.h>
26 #include <linux/in.h>
27 #include <linux/inet.h>
28 #include <linux/udp.h>
29 #include <linux/tcp.h>
30 #include <linux/unistd.h>
31 #include <linux/slab.h>
32 #include <linux/netdevice.h>
33 #include <linux/skbuff.h>
34 #include <linux/file.h>
35 #include <linux/freezer.h>
36 #include <net/sock.h>
37 #include <net/checksum.h>
38 #include <net/ip.h>
39 #include <net/tcp_states.h>
40 #include <asm/uaccess.h>
41 #include <asm/ioctls.h>
42
43 #include <linux/sunrpc/types.h>
44 #include <linux/sunrpc/clnt.h>
45 #include <linux/sunrpc/xdr.h>
46 #include <linux/sunrpc/svcsock.h>
47 #include <linux/sunrpc/stats.h>
48
49 /* SMP locking strategy:
50  *
51  *      svc_pool->sp_lock protects most of the fields of that pool.
52  *      svc_serv->sv_lock protects sv_tempsocks, sv_permsocks, sv_tmpcnt.
53  *      when both need to be taken (rare), svc_serv->sv_lock is first.
54  *      BKL protects svc_serv->sv_nrthread.
55  *      svc_sock->sk_defer_lock protects the svc_sock->sk_deferred list
56  *      svc_sock->sk_flags.SK_BUSY prevents a svc_sock being enqueued multiply.
57  *
58  *      Some flags can be set to certain values at any time
59  *      providing that certain rules are followed:
60  *
61  *      SK_CONN, SK_DATA, can be set or cleared at any time.
62  *              after a set, svc_sock_enqueue must be called.
63  *              after a clear, the socket must be read/accepted
64  *               if this succeeds, it must be set again.
65  *      SK_CLOSE can set at any time. It is never cleared.
66  *      sk_inuse contains a bias of '1' until SK_DEAD is set.
67  *             so when sk_inuse hits zero, we know the socket is dead
68  *             and no-one is using it.
69  *      SK_DEAD can only be set while SK_BUSY is held which ensures
70  *             no other thread will be using the socket or will try to
71  *             set SK_DEAD.
72  *
73  */
74
75 #define RPCDBG_FACILITY RPCDBG_SVCSOCK
76
77
78 static struct svc_sock *svc_setup_socket(struct svc_serv *, struct socket *,
79                                          int *errp, int flags);
80 static void             svc_delete_socket(struct svc_sock *svsk);
81 static void             svc_udp_data_ready(struct sock *, int);
82 static int              svc_udp_recvfrom(struct svc_rqst *);
83 static int              svc_udp_sendto(struct svc_rqst *);
84
85 static struct svc_deferred_req *svc_deferred_dequeue(struct svc_sock *svsk);
86 static int svc_deferred_recv(struct svc_rqst *rqstp);
87 static struct cache_deferred_req *svc_defer(struct cache_req *req);
88
89 /* apparently the "standard" is that clients close
90  * idle connections after 5 minutes, servers after
91  * 6 minutes
92  *   http://www.connectathon.org/talks96/nfstcp.pdf
93  */
94 static int svc_conn_age_period = 6*60;
95
96 #ifdef CONFIG_DEBUG_LOCK_ALLOC
97 static struct lock_class_key svc_key[2];
98 static struct lock_class_key svc_slock_key[2];
99
100 static inline void svc_reclassify_socket(struct socket *sock)
101 {
102         struct sock *sk = sock->sk;
103         BUG_ON(sk->sk_lock.owner != NULL);
104         switch (sk->sk_family) {
105         case AF_INET:
106                 sock_lock_init_class_and_name(sk, "slock-AF_INET-NFSD",
107                     &svc_slock_key[0], "sk_lock-AF_INET-NFSD", &svc_key[0]);
108                 break;
109
110         case AF_INET6:
111                 sock_lock_init_class_and_name(sk, "slock-AF_INET6-NFSD",
112                     &svc_slock_key[1], "sk_lock-AF_INET6-NFSD", &svc_key[1]);
113                 break;
114
115         default:
116                 BUG();
117         }
118 }
119 #else
120 static inline void svc_reclassify_socket(struct socket *sock)
121 {
122 }
123 #endif
124
125 static char *__svc_print_addr(struct sockaddr *addr, char *buf, size_t len)
126 {
127         switch (addr->sa_family) {
128         case AF_INET:
129                 snprintf(buf, len, "%u.%u.%u.%u, port=%u",
130                         NIPQUAD(((struct sockaddr_in *) addr)->sin_addr),
131                         htons(((struct sockaddr_in *) addr)->sin_port));
132                 break;
133 #if defined(CONFIG_IPV6) || defined(CONFIG_IPV6_MODULE)
134         case AF_INET6:
135                 snprintf(buf, len, "%x:%x:%x:%x:%x:%x:%x:%x, port=%u",
136                         NIP6(((struct sockaddr_in6 *) addr)->sin6_addr),
137                         htons(((struct sockaddr_in6 *) addr)->sin6_port));
138                 break;
139 #endif
140         default:
141                 snprintf(buf, len, "unknown address type: %d", addr->sa_family);
142                 break;
143         }
144         return buf;
145 }
146
147 /**
148  * svc_print_addr - Format rq_addr field for printing
149  * @rqstp: svc_rqst struct containing address to print
150  * @buf: target buffer for formatted address
151  * @len: length of target buffer
152  *
153  */
154 char *svc_print_addr(struct svc_rqst *rqstp, char *buf, size_t len)
155 {
156         return __svc_print_addr((struct sockaddr *) &rqstp->rq_addr, buf, len);
157 }
158 EXPORT_SYMBOL_GPL(svc_print_addr);
159
160 /*
161  * Queue up an idle server thread.  Must have pool->sp_lock held.
162  * Note: this is really a stack rather than a queue, so that we only
163  * use as many different threads as we need, and the rest don't pollute
164  * the cache.
165  */
166 static inline void
167 svc_thread_enqueue(struct svc_pool *pool, struct svc_rqst *rqstp)
168 {
169         list_add(&rqstp->rq_list, &pool->sp_threads);
170 }
171
172 /*
173  * Dequeue an nfsd thread.  Must have pool->sp_lock held.
174  */
175 static inline void
176 svc_thread_dequeue(struct svc_pool *pool, struct svc_rqst *rqstp)
177 {
178         list_del(&rqstp->rq_list);
179 }
180
181 /*
182  * Release an skbuff after use
183  */
184 static inline void
185 svc_release_skb(struct svc_rqst *rqstp)
186 {
187         struct sk_buff *skb = rqstp->rq_skbuff;
188         struct svc_deferred_req *dr = rqstp->rq_deferred;
189
190         if (skb) {
191                 rqstp->rq_skbuff = NULL;
192
193                 dprintk("svc: service %p, releasing skb %p\n", rqstp, skb);
194                 skb_free_datagram(rqstp->rq_sock->sk_sk, skb);
195         }
196         if (dr) {
197                 rqstp->rq_deferred = NULL;
198                 kfree(dr);
199         }
200 }
201
202 /*
203  * Any space to write?
204  */
205 static inline unsigned long
206 svc_sock_wspace(struct svc_sock *svsk)
207 {
208         int wspace;
209
210         if (svsk->sk_sock->type == SOCK_STREAM)
211                 wspace = sk_stream_wspace(svsk->sk_sk);
212         else
213                 wspace = sock_wspace(svsk->sk_sk);
214
215         return wspace;
216 }
217
218 /*
219  * Queue up a socket with data pending. If there are idle nfsd
220  * processes, wake 'em up.
221  *
222  */
223 static void
224 svc_sock_enqueue(struct svc_sock *svsk)
225 {
226         struct svc_serv *serv = svsk->sk_server;
227         struct svc_pool *pool;
228         struct svc_rqst *rqstp;
229         int cpu;
230
231         if (!(svsk->sk_flags &
232               ( (1<<SK_CONN)|(1<<SK_DATA)|(1<<SK_CLOSE)|(1<<SK_DEFERRED)) ))
233                 return;
234         if (test_bit(SK_DEAD, &svsk->sk_flags))
235                 return;
236
237         cpu = get_cpu();
238         pool = svc_pool_for_cpu(svsk->sk_server, cpu);
239         put_cpu();
240
241         spin_lock_bh(&pool->sp_lock);
242
243         if (!list_empty(&pool->sp_threads) &&
244             !list_empty(&pool->sp_sockets))
245                 printk(KERN_ERR
246                         "svc_sock_enqueue: threads and sockets both waiting??\n");
247
248         if (test_bit(SK_DEAD, &svsk->sk_flags)) {
249                 /* Don't enqueue dead sockets */
250                 dprintk("svc: socket %p is dead, not enqueued\n", svsk->sk_sk);
251                 goto out_unlock;
252         }
253
254         /* Mark socket as busy. It will remain in this state until the
255          * server has processed all pending data and put the socket back
256          * on the idle list.  We update SK_BUSY atomically because
257          * it also guards against trying to enqueue the svc_sock twice.
258          */
259         if (test_and_set_bit(SK_BUSY, &svsk->sk_flags)) {
260                 /* Don't enqueue socket while already enqueued */
261                 dprintk("svc: socket %p busy, not enqueued\n", svsk->sk_sk);
262                 goto out_unlock;
263         }
264         BUG_ON(svsk->sk_pool != NULL);
265         svsk->sk_pool = pool;
266
267         set_bit(SOCK_NOSPACE, &svsk->sk_sock->flags);
268         if (((atomic_read(&svsk->sk_reserved) + serv->sv_max_mesg)*2
269              > svc_sock_wspace(svsk))
270             && !test_bit(SK_CLOSE, &svsk->sk_flags)
271             && !test_bit(SK_CONN, &svsk->sk_flags)) {
272                 /* Don't enqueue while not enough space for reply */
273                 dprintk("svc: socket %p  no space, %d*2 > %ld, not enqueued\n",
274                         svsk->sk_sk, atomic_read(&svsk->sk_reserved)+serv->sv_max_mesg,
275                         svc_sock_wspace(svsk));
276                 svsk->sk_pool = NULL;
277                 clear_bit(SK_BUSY, &svsk->sk_flags);
278                 goto out_unlock;
279         }
280         clear_bit(SOCK_NOSPACE, &svsk->sk_sock->flags);
281
282
283         if (!list_empty(&pool->sp_threads)) {
284                 rqstp = list_entry(pool->sp_threads.next,
285                                    struct svc_rqst,
286                                    rq_list);
287                 dprintk("svc: socket %p served by daemon %p\n",
288                         svsk->sk_sk, rqstp);
289                 svc_thread_dequeue(pool, rqstp);
290                 if (rqstp->rq_sock)
291                         printk(KERN_ERR
292                                 "svc_sock_enqueue: server %p, rq_sock=%p!\n",
293                                 rqstp, rqstp->rq_sock);
294                 rqstp->rq_sock = svsk;
295                 atomic_inc(&svsk->sk_inuse);
296                 rqstp->rq_reserved = serv->sv_max_mesg;
297                 atomic_add(rqstp->rq_reserved, &svsk->sk_reserved);
298                 BUG_ON(svsk->sk_pool != pool);
299                 wake_up(&rqstp->rq_wait);
300         } else {
301                 dprintk("svc: socket %p put into queue\n", svsk->sk_sk);
302                 list_add_tail(&svsk->sk_ready, &pool->sp_sockets);
303                 BUG_ON(svsk->sk_pool != pool);
304         }
305
306 out_unlock:
307         spin_unlock_bh(&pool->sp_lock);
308 }
309
310 /*
311  * Dequeue the first socket.  Must be called with the pool->sp_lock held.
312  */
313 static inline struct svc_sock *
314 svc_sock_dequeue(struct svc_pool *pool)
315 {
316         struct svc_sock *svsk;
317
318         if (list_empty(&pool->sp_sockets))
319                 return NULL;
320
321         svsk = list_entry(pool->sp_sockets.next,
322                           struct svc_sock, sk_ready);
323         list_del_init(&svsk->sk_ready);
324
325         dprintk("svc: socket %p dequeued, inuse=%d\n",
326                 svsk->sk_sk, atomic_read(&svsk->sk_inuse));
327
328         return svsk;
329 }
330
331 /*
332  * Having read something from a socket, check whether it
333  * needs to be re-enqueued.
334  * Note: SK_DATA only gets cleared when a read-attempt finds
335  * no (or insufficient) data.
336  */
337 static inline void
338 svc_sock_received(struct svc_sock *svsk)
339 {
340         svsk->sk_pool = NULL;
341         clear_bit(SK_BUSY, &svsk->sk_flags);
342         svc_sock_enqueue(svsk);
343 }
344
345
346 /**
347  * svc_reserve - change the space reserved for the reply to a request.
348  * @rqstp:  The request in question
349  * @space: new max space to reserve
350  *
351  * Each request reserves some space on the output queue of the socket
352  * to make sure the reply fits.  This function reduces that reserved
353  * space to be the amount of space used already, plus @space.
354  *
355  */
356 void svc_reserve(struct svc_rqst *rqstp, int space)
357 {
358         space += rqstp->rq_res.head[0].iov_len;
359
360         if (space < rqstp->rq_reserved) {
361                 struct svc_sock *svsk = rqstp->rq_sock;
362                 atomic_sub((rqstp->rq_reserved - space), &svsk->sk_reserved);
363                 rqstp->rq_reserved = space;
364
365                 svc_sock_enqueue(svsk);
366         }
367 }
368
369 /*
370  * Release a socket after use.
371  */
372 static inline void
373 svc_sock_put(struct svc_sock *svsk)
374 {
375         if (atomic_dec_and_test(&svsk->sk_inuse)) {
376                 BUG_ON(! test_bit(SK_DEAD, &svsk->sk_flags));
377
378                 dprintk("svc: releasing dead socket\n");
379                 if (svsk->sk_sock->file)
380                         sockfd_put(svsk->sk_sock);
381                 else
382                         sock_release(svsk->sk_sock);
383                 if (svsk->sk_info_authunix != NULL)
384                         svcauth_unix_info_release(svsk->sk_info_authunix);
385                 kfree(svsk);
386         }
387 }
388
389 static void
390 svc_sock_release(struct svc_rqst *rqstp)
391 {
392         struct svc_sock *svsk = rqstp->rq_sock;
393
394         svc_release_skb(rqstp);
395
396         svc_free_res_pages(rqstp);
397         rqstp->rq_res.page_len = 0;
398         rqstp->rq_res.page_base = 0;
399
400
401         /* Reset response buffer and release
402          * the reservation.
403          * But first, check that enough space was reserved
404          * for the reply, otherwise we have a bug!
405          */
406         if ((rqstp->rq_res.len) >  rqstp->rq_reserved)
407                 printk(KERN_ERR "RPC request reserved %d but used %d\n",
408                        rqstp->rq_reserved,
409                        rqstp->rq_res.len);
410
411         rqstp->rq_res.head[0].iov_len = 0;
412         svc_reserve(rqstp, 0);
413         rqstp->rq_sock = NULL;
414
415         svc_sock_put(svsk);
416 }
417
418 /*
419  * External function to wake up a server waiting for data
420  * This really only makes sense for services like lockd
421  * which have exactly one thread anyway.
422  */
423 void
424 svc_wake_up(struct svc_serv *serv)
425 {
426         struct svc_rqst *rqstp;
427         unsigned int i;
428         struct svc_pool *pool;
429
430         for (i = 0; i < serv->sv_nrpools; i++) {
431                 pool = &serv->sv_pools[i];
432
433                 spin_lock_bh(&pool->sp_lock);
434                 if (!list_empty(&pool->sp_threads)) {
435                         rqstp = list_entry(pool->sp_threads.next,
436                                            struct svc_rqst,
437                                            rq_list);
438                         dprintk("svc: daemon %p woken up.\n", rqstp);
439                         /*
440                         svc_thread_dequeue(pool, rqstp);
441                         rqstp->rq_sock = NULL;
442                          */
443                         wake_up(&rqstp->rq_wait);
444                 }
445                 spin_unlock_bh(&pool->sp_lock);
446         }
447 }
448
449 /*
450  * Generic sendto routine
451  */
452 static int
453 svc_sendto(struct svc_rqst *rqstp, struct xdr_buf *xdr)
454 {
455         struct svc_sock *svsk = rqstp->rq_sock;
456         struct socket   *sock = svsk->sk_sock;
457         int             slen;
458         char            buffer[CMSG_SPACE(sizeof(struct in_pktinfo))];
459         struct cmsghdr *cmh = (struct cmsghdr *)buffer;
460         struct in_pktinfo *pki = (struct in_pktinfo *)CMSG_DATA(cmh);
461         int             len = 0;
462         int             result;
463         int             size;
464         struct page     **ppage = xdr->pages;
465         size_t          base = xdr->page_base;
466         unsigned int    pglen = xdr->page_len;
467         unsigned int    flags = MSG_MORE;
468         char            buf[RPC_MAX_ADDRBUFLEN];
469
470         slen = xdr->len;
471
472         if (rqstp->rq_prot == IPPROTO_UDP) {
473                 /* set the source and destination */
474                 struct msghdr   msg;
475                 msg.msg_name    = &rqstp->rq_addr;
476                 msg.msg_namelen = sizeof(rqstp->rq_addr);
477                 msg.msg_iov     = NULL;
478                 msg.msg_iovlen  = 0;
479                 msg.msg_flags   = MSG_MORE;
480
481                 msg.msg_control = cmh;
482                 msg.msg_controllen = sizeof(buffer);
483                 cmh->cmsg_len = CMSG_LEN(sizeof(*pki));
484                 cmh->cmsg_level = SOL_IP;
485                 cmh->cmsg_type = IP_PKTINFO;
486                 pki->ipi_ifindex = 0;
487                 pki->ipi_spec_dst.s_addr = rqstp->rq_daddr;
488
489                 if (sock_sendmsg(sock, &msg, 0) < 0)
490                         goto out;
491         }
492
493         /* send head */
494         if (slen == xdr->head[0].iov_len)
495                 flags = 0;
496         len = kernel_sendpage(sock, rqstp->rq_respages[0], 0,
497                                   xdr->head[0].iov_len, flags);
498         if (len != xdr->head[0].iov_len)
499                 goto out;
500         slen -= xdr->head[0].iov_len;
501         if (slen == 0)
502                 goto out;
503
504         /* send page data */
505         size = PAGE_SIZE - base < pglen ? PAGE_SIZE - base : pglen;
506         while (pglen > 0) {
507                 if (slen == size)
508                         flags = 0;
509                 result = kernel_sendpage(sock, *ppage, base, size, flags);
510                 if (result > 0)
511                         len += result;
512                 if (result != size)
513                         goto out;
514                 slen -= size;
515                 pglen -= size;
516                 size = PAGE_SIZE < pglen ? PAGE_SIZE : pglen;
517                 base = 0;
518                 ppage++;
519         }
520         /* send tail */
521         if (xdr->tail[0].iov_len) {
522                 result = kernel_sendpage(sock, rqstp->rq_respages[0],
523                                              ((unsigned long)xdr->tail[0].iov_base)
524                                                 & (PAGE_SIZE-1),
525                                              xdr->tail[0].iov_len, 0);
526
527                 if (result > 0)
528                         len += result;
529         }
530 out:
531         dprintk("svc: socket %p sendto([%p %Zu... ], %d) = %d (addr %s)\n",
532                 rqstp->rq_sock, xdr->head[0].iov_base, xdr->head[0].iov_len,
533                 xdr->len, len, svc_print_addr(rqstp, buf, sizeof(buf)));
534
535         return len;
536 }
537
538 /*
539  * Report socket names for nfsdfs
540  */
541 static int one_sock_name(char *buf, struct svc_sock *svsk)
542 {
543         int len;
544
545         switch(svsk->sk_sk->sk_family) {
546         case AF_INET:
547                 len = sprintf(buf, "ipv4 %s %u.%u.%u.%u %d\n",
548                               svsk->sk_sk->sk_protocol==IPPROTO_UDP?
549                               "udp" : "tcp",
550                               NIPQUAD(inet_sk(svsk->sk_sk)->rcv_saddr),
551                               inet_sk(svsk->sk_sk)->num);
552                 break;
553         default:
554                 len = sprintf(buf, "*unknown-%d*\n",
555                                svsk->sk_sk->sk_family);
556         }
557         return len;
558 }
559
560 int
561 svc_sock_names(char *buf, struct svc_serv *serv, char *toclose)
562 {
563         struct svc_sock *svsk, *closesk = NULL;
564         int len = 0;
565
566         if (!serv)
567                 return 0;
568         spin_lock_bh(&serv->sv_lock);
569         list_for_each_entry(svsk, &serv->sv_permsocks, sk_list) {
570                 int onelen = one_sock_name(buf+len, svsk);
571                 if (toclose && strcmp(toclose, buf+len) == 0)
572                         closesk = svsk;
573                 else
574                         len += onelen;
575         }
576         spin_unlock_bh(&serv->sv_lock);
577         if (closesk)
578                 /* Should unregister with portmap, but you cannot
579                  * unregister just one protocol...
580                  */
581                 svc_close_socket(closesk);
582         else if (toclose)
583                 return -ENOENT;
584         return len;
585 }
586 EXPORT_SYMBOL(svc_sock_names);
587
588 /*
589  * Check input queue length
590  */
591 static int
592 svc_recv_available(struct svc_sock *svsk)
593 {
594         struct socket   *sock = svsk->sk_sock;
595         int             avail, err;
596
597         err = kernel_sock_ioctl(sock, TIOCINQ, (unsigned long) &avail);
598
599         return (err >= 0)? avail : err;
600 }
601
602 /*
603  * Generic recvfrom routine.
604  */
605 static int
606 svc_recvfrom(struct svc_rqst *rqstp, struct kvec *iov, int nr, int buflen)
607 {
608         struct svc_sock *svsk = rqstp->rq_sock;
609         struct msghdr msg = {
610                 .msg_flags      = MSG_DONTWAIT,
611         };
612         int len;
613
614         len = kernel_recvmsg(svsk->sk_sock, &msg, iov, nr, buflen,
615                                 msg.msg_flags);
616
617         /* sock_recvmsg doesn't fill in the name/namelen, so we must..
618          */
619         memcpy(&rqstp->rq_addr, &svsk->sk_remote, svsk->sk_remotelen);
620         rqstp->rq_addrlen = svsk->sk_remotelen;
621
622         dprintk("svc: socket %p recvfrom(%p, %Zu) = %d\n",
623                 svsk, iov[0].iov_base, iov[0].iov_len, len);
624
625         return len;
626 }
627
628 /*
629  * Set socket snd and rcv buffer lengths
630  */
631 static inline void
632 svc_sock_setbufsize(struct socket *sock, unsigned int snd, unsigned int rcv)
633 {
634 #if 0
635         mm_segment_t    oldfs;
636         oldfs = get_fs(); set_fs(KERNEL_DS);
637         sock_setsockopt(sock, SOL_SOCKET, SO_SNDBUF,
638                         (char*)&snd, sizeof(snd));
639         sock_setsockopt(sock, SOL_SOCKET, SO_RCVBUF,
640                         (char*)&rcv, sizeof(rcv));
641 #else
642         /* sock_setsockopt limits use to sysctl_?mem_max,
643          * which isn't acceptable.  Until that is made conditional
644          * on not having CAP_SYS_RESOURCE or similar, we go direct...
645          * DaveM said I could!
646          */
647         lock_sock(sock->sk);
648         sock->sk->sk_sndbuf = snd * 2;
649         sock->sk->sk_rcvbuf = rcv * 2;
650         sock->sk->sk_userlocks |= SOCK_SNDBUF_LOCK|SOCK_RCVBUF_LOCK;
651         release_sock(sock->sk);
652 #endif
653 }
654 /*
655  * INET callback when data has been received on the socket.
656  */
657 static void
658 svc_udp_data_ready(struct sock *sk, int count)
659 {
660         struct svc_sock *svsk = (struct svc_sock *)sk->sk_user_data;
661
662         if (svsk) {
663                 dprintk("svc: socket %p(inet %p), count=%d, busy=%d\n",
664                         svsk, sk, count, test_bit(SK_BUSY, &svsk->sk_flags));
665                 set_bit(SK_DATA, &svsk->sk_flags);
666                 svc_sock_enqueue(svsk);
667         }
668         if (sk->sk_sleep && waitqueue_active(sk->sk_sleep))
669                 wake_up_interruptible(sk->sk_sleep);
670 }
671
672 /*
673  * INET callback when space is newly available on the socket.
674  */
675 static void
676 svc_write_space(struct sock *sk)
677 {
678         struct svc_sock *svsk = (struct svc_sock *)(sk->sk_user_data);
679
680         if (svsk) {
681                 dprintk("svc: socket %p(inet %p), write_space busy=%d\n",
682                         svsk, sk, test_bit(SK_BUSY, &svsk->sk_flags));
683                 svc_sock_enqueue(svsk);
684         }
685
686         if (sk->sk_sleep && waitqueue_active(sk->sk_sleep)) {
687                 dprintk("RPC svc_write_space: someone sleeping on %p\n",
688                        svsk);
689                 wake_up_interruptible(sk->sk_sleep);
690         }
691 }
692
693 /*
694  * Receive a datagram from a UDP socket.
695  */
696 static int
697 svc_udp_recvfrom(struct svc_rqst *rqstp)
698 {
699         struct svc_sock *svsk = rqstp->rq_sock;
700         struct svc_serv *serv = svsk->sk_server;
701         struct sk_buff  *skb;
702         int             err, len;
703
704         if (test_and_clear_bit(SK_CHNGBUF, &svsk->sk_flags))
705             /* udp sockets need large rcvbuf as all pending
706              * requests are still in that buffer.  sndbuf must
707              * also be large enough that there is enough space
708              * for one reply per thread.  We count all threads
709              * rather than threads in a particular pool, which
710              * provides an upper bound on the number of threads
711              * which will access the socket.
712              */
713             svc_sock_setbufsize(svsk->sk_sock,
714                                 (serv->sv_nrthreads+3) * serv->sv_max_mesg,
715                                 (serv->sv_nrthreads+3) * serv->sv_max_mesg);
716
717         if ((rqstp->rq_deferred = svc_deferred_dequeue(svsk))) {
718                 svc_sock_received(svsk);
719                 return svc_deferred_recv(rqstp);
720         }
721
722         if (test_bit(SK_CLOSE, &svsk->sk_flags)) {
723                 svc_delete_socket(svsk);
724                 return 0;
725         }
726
727         clear_bit(SK_DATA, &svsk->sk_flags);
728         while ((skb = skb_recv_datagram(svsk->sk_sk, 0, 1, &err)) == NULL) {
729                 if (err == -EAGAIN) {
730                         svc_sock_received(svsk);
731                         return err;
732                 }
733                 /* possibly an icmp error */
734                 dprintk("svc: recvfrom returned error %d\n", -err);
735         }
736         if (skb->tstamp.off_sec == 0) {
737                 struct timeval tv;
738
739                 tv.tv_sec = xtime.tv_sec;
740                 tv.tv_usec = xtime.tv_nsec / NSEC_PER_USEC;
741                 skb_set_timestamp(skb, &tv);
742                 /* Don't enable netstamp, sunrpc doesn't
743                    need that much accuracy */
744         }
745         skb_get_timestamp(skb, &svsk->sk_sk->sk_stamp);
746         set_bit(SK_DATA, &svsk->sk_flags); /* there may be more data... */
747
748         /*
749          * Maybe more packets - kick another thread ASAP.
750          */
751         svc_sock_received(svsk);
752
753         len  = skb->len - sizeof(struct udphdr);
754         rqstp->rq_arg.len = len;
755
756         rqstp->rq_prot        = IPPROTO_UDP;
757
758         /* Get sender address */
759         rqstp->rq_addr.sin_family = AF_INET;
760         rqstp->rq_addr.sin_port = skb->h.uh->source;
761         rqstp->rq_addr.sin_addr.s_addr = skb->nh.iph->saddr;
762         rqstp->rq_daddr = skb->nh.iph->daddr;
763
764         if (skb_is_nonlinear(skb)) {
765                 /* we have to copy */
766                 local_bh_disable();
767                 if (csum_partial_copy_to_xdr(&rqstp->rq_arg, skb)) {
768                         local_bh_enable();
769                         /* checksum error */
770                         skb_free_datagram(svsk->sk_sk, skb);
771                         return 0;
772                 }
773                 local_bh_enable();
774                 skb_free_datagram(svsk->sk_sk, skb);
775         } else {
776                 /* we can use it in-place */
777                 rqstp->rq_arg.head[0].iov_base = skb->data + sizeof(struct udphdr);
778                 rqstp->rq_arg.head[0].iov_len = len;
779                 if (skb_checksum_complete(skb)) {
780                         skb_free_datagram(svsk->sk_sk, skb);
781                         return 0;
782                 }
783                 rqstp->rq_skbuff = skb;
784         }
785
786         rqstp->rq_arg.page_base = 0;
787         if (len <= rqstp->rq_arg.head[0].iov_len) {
788                 rqstp->rq_arg.head[0].iov_len = len;
789                 rqstp->rq_arg.page_len = 0;
790                 rqstp->rq_respages = rqstp->rq_pages+1;
791         } else {
792                 rqstp->rq_arg.page_len = len - rqstp->rq_arg.head[0].iov_len;
793                 rqstp->rq_respages = rqstp->rq_pages + 1 +
794                         (rqstp->rq_arg.page_len + PAGE_SIZE - 1)/ PAGE_SIZE;
795         }
796
797         if (serv->sv_stats)
798                 serv->sv_stats->netudpcnt++;
799
800         return len;
801 }
802
803 static int
804 svc_udp_sendto(struct svc_rqst *rqstp)
805 {
806         int             error;
807
808         error = svc_sendto(rqstp, &rqstp->rq_res);
809         if (error == -ECONNREFUSED)
810                 /* ICMP error on earlier request. */
811                 error = svc_sendto(rqstp, &rqstp->rq_res);
812
813         return error;
814 }
815
816 static void
817 svc_udp_init(struct svc_sock *svsk)
818 {
819         svsk->sk_sk->sk_data_ready = svc_udp_data_ready;
820         svsk->sk_sk->sk_write_space = svc_write_space;
821         svsk->sk_recvfrom = svc_udp_recvfrom;
822         svsk->sk_sendto = svc_udp_sendto;
823
824         /* initialise setting must have enough space to
825          * receive and respond to one request.
826          * svc_udp_recvfrom will re-adjust if necessary
827          */
828         svc_sock_setbufsize(svsk->sk_sock,
829                             3 * svsk->sk_server->sv_max_mesg,
830                             3 * svsk->sk_server->sv_max_mesg);
831
832         set_bit(SK_DATA, &svsk->sk_flags); /* might have come in before data_ready set up */
833         set_bit(SK_CHNGBUF, &svsk->sk_flags);
834 }
835
836 /*
837  * A data_ready event on a listening socket means there's a connection
838  * pending. Do not use state_change as a substitute for it.
839  */
840 static void
841 svc_tcp_listen_data_ready(struct sock *sk, int count_unused)
842 {
843         struct svc_sock *svsk = (struct svc_sock *)sk->sk_user_data;
844
845         dprintk("svc: socket %p TCP (listen) state change %d\n",
846                 sk, sk->sk_state);
847
848         /*
849          * This callback may called twice when a new connection
850          * is established as a child socket inherits everything
851          * from a parent LISTEN socket.
852          * 1) data_ready method of the parent socket will be called
853          *    when one of child sockets become ESTABLISHED.
854          * 2) data_ready method of the child socket may be called
855          *    when it receives data before the socket is accepted.
856          * In case of 2, we should ignore it silently.
857          */
858         if (sk->sk_state == TCP_LISTEN) {
859                 if (svsk) {
860                         set_bit(SK_CONN, &svsk->sk_flags);
861                         svc_sock_enqueue(svsk);
862                 } else
863                         printk("svc: socket %p: no user data\n", sk);
864         }
865
866         if (sk->sk_sleep && waitqueue_active(sk->sk_sleep))
867                 wake_up_interruptible_all(sk->sk_sleep);
868 }
869
870 /*
871  * A state change on a connected socket means it's dying or dead.
872  */
873 static void
874 svc_tcp_state_change(struct sock *sk)
875 {
876         struct svc_sock *svsk = (struct svc_sock *)sk->sk_user_data;
877
878         dprintk("svc: socket %p TCP (connected) state change %d (svsk %p)\n",
879                 sk, sk->sk_state, sk->sk_user_data);
880
881         if (!svsk)
882                 printk("svc: socket %p: no user data\n", sk);
883         else {
884                 set_bit(SK_CLOSE, &svsk->sk_flags);
885                 svc_sock_enqueue(svsk);
886         }
887         if (sk->sk_sleep && waitqueue_active(sk->sk_sleep))
888                 wake_up_interruptible_all(sk->sk_sleep);
889 }
890
891 static void
892 svc_tcp_data_ready(struct sock *sk, int count)
893 {
894         struct svc_sock *svsk = (struct svc_sock *)sk->sk_user_data;
895
896         dprintk("svc: socket %p TCP data ready (svsk %p)\n",
897                 sk, sk->sk_user_data);
898         if (svsk) {
899                 set_bit(SK_DATA, &svsk->sk_flags);
900                 svc_sock_enqueue(svsk);
901         }
902         if (sk->sk_sleep && waitqueue_active(sk->sk_sleep))
903                 wake_up_interruptible(sk->sk_sleep);
904 }
905
906 /*
907  * Accept a TCP connection
908  */
909 static void
910 svc_tcp_accept(struct svc_sock *svsk)
911 {
912         struct sockaddr_in sin;
913         struct svc_serv *serv = svsk->sk_server;
914         struct socket   *sock = svsk->sk_sock;
915         struct socket   *newsock;
916         struct svc_sock *newsvsk;
917         int             err, slen;
918         char            buf[RPC_MAX_ADDRBUFLEN];
919
920         dprintk("svc: tcp_accept %p sock %p\n", svsk, sock);
921         if (!sock)
922                 return;
923
924         clear_bit(SK_CONN, &svsk->sk_flags);
925         err = kernel_accept(sock, &newsock, O_NONBLOCK);
926         if (err < 0) {
927                 if (err == -ENOMEM)
928                         printk(KERN_WARNING "%s: no more sockets!\n",
929                                serv->sv_name);
930                 else if (err != -EAGAIN && net_ratelimit())
931                         printk(KERN_WARNING "%s: accept failed (err %d)!\n",
932                                    serv->sv_name, -err);
933                 return;
934         }
935
936         set_bit(SK_CONN, &svsk->sk_flags);
937         svc_sock_enqueue(svsk);
938
939         slen = sizeof(sin);
940         err = kernel_getpeername(newsock, (struct sockaddr *) &sin, &slen);
941         if (err < 0) {
942                 if (net_ratelimit())
943                         printk(KERN_WARNING "%s: peername failed (err %d)!\n",
944                                    serv->sv_name, -err);
945                 goto failed;            /* aborted connection or whatever */
946         }
947
948         /* Ideally, we would want to reject connections from unauthorized
949          * hosts here, but when we get encryption, the IP of the host won't
950          * tell us anything.  For now just warn about unpriv connections.
951          */
952         if (ntohs(sin.sin_port) >= 1024) {
953                 dprintk(KERN_WARNING
954                         "%s: connect from unprivileged port: %s\n",
955                         serv->sv_name,
956                         __svc_print_addr((struct sockaddr *) &sin, buf,
957                                                                 sizeof(buf)));
958         }
959         dprintk("%s: connect from %s\n", serv->sv_name,
960                 __svc_print_addr((struct sockaddr *) &sin, buf,
961                                  sizeof(buf)));
962
963         /* make sure that a write doesn't block forever when
964          * low on memory
965          */
966         newsock->sk->sk_sndtimeo = HZ*30;
967
968         if (!(newsvsk = svc_setup_socket(serv, newsock, &err,
969                                  (SVC_SOCK_ANONYMOUS | SVC_SOCK_TEMPORARY))))
970                 goto failed;
971         memcpy(&newsvsk->sk_remote, &sin, slen);
972         newsvsk->sk_remotelen = slen;
973
974         svc_sock_received(newsvsk);
975
976         /* make sure that we don't have too many active connections.
977          * If we have, something must be dropped.
978          *
979          * There's no point in trying to do random drop here for
980          * DoS prevention. The NFS clients does 1 reconnect in 15
981          * seconds. An attacker can easily beat that.
982          *
983          * The only somewhat efficient mechanism would be if drop
984          * old connections from the same IP first. But right now
985          * we don't even record the client IP in svc_sock.
986          */
987         if (serv->sv_tmpcnt > (serv->sv_nrthreads+3)*20) {
988                 struct svc_sock *svsk = NULL;
989                 spin_lock_bh(&serv->sv_lock);
990                 if (!list_empty(&serv->sv_tempsocks)) {
991                         if (net_ratelimit()) {
992                                 /* Try to help the admin */
993                                 printk(KERN_NOTICE "%s: too many open TCP "
994                                         "sockets, consider increasing the "
995                                         "number of nfsd threads\n",
996                                                    serv->sv_name);
997                                 printk(KERN_NOTICE
998                                        "%s: last TCP connect from %s\n",
999                                        serv->sv_name, buf);
1000                         }
1001                         /*
1002                          * Always select the oldest socket. It's not fair,
1003                          * but so is life
1004                          */
1005                         svsk = list_entry(serv->sv_tempsocks.prev,
1006                                           struct svc_sock,
1007                                           sk_list);
1008                         set_bit(SK_CLOSE, &svsk->sk_flags);
1009                         atomic_inc(&svsk->sk_inuse);
1010                 }
1011                 spin_unlock_bh(&serv->sv_lock);
1012
1013                 if (svsk) {
1014                         svc_sock_enqueue(svsk);
1015                         svc_sock_put(svsk);
1016                 }
1017
1018         }
1019
1020         if (serv->sv_stats)
1021                 serv->sv_stats->nettcpconn++;
1022
1023         return;
1024
1025 failed:
1026         sock_release(newsock);
1027         return;
1028 }
1029
1030 /*
1031  * Receive data from a TCP socket.
1032  */
1033 static int
1034 svc_tcp_recvfrom(struct svc_rqst *rqstp)
1035 {
1036         struct svc_sock *svsk = rqstp->rq_sock;
1037         struct svc_serv *serv = svsk->sk_server;
1038         int             len;
1039         struct kvec *vec;
1040         int pnum, vlen;
1041
1042         dprintk("svc: tcp_recv %p data %d conn %d close %d\n",
1043                 svsk, test_bit(SK_DATA, &svsk->sk_flags),
1044                 test_bit(SK_CONN, &svsk->sk_flags),
1045                 test_bit(SK_CLOSE, &svsk->sk_flags));
1046
1047         if ((rqstp->rq_deferred = svc_deferred_dequeue(svsk))) {
1048                 svc_sock_received(svsk);
1049                 return svc_deferred_recv(rqstp);
1050         }
1051
1052         if (test_bit(SK_CLOSE, &svsk->sk_flags)) {
1053                 svc_delete_socket(svsk);
1054                 return 0;
1055         }
1056
1057         if (svsk->sk_sk->sk_state == TCP_LISTEN) {
1058                 svc_tcp_accept(svsk);
1059                 svc_sock_received(svsk);
1060                 return 0;
1061         }
1062
1063         if (test_and_clear_bit(SK_CHNGBUF, &svsk->sk_flags))
1064                 /* sndbuf needs to have room for one request
1065                  * per thread, otherwise we can stall even when the
1066                  * network isn't a bottleneck.
1067                  *
1068                  * We count all threads rather than threads in a
1069                  * particular pool, which provides an upper bound
1070                  * on the number of threads which will access the socket.
1071                  *
1072                  * rcvbuf just needs to be able to hold a few requests.
1073                  * Normally they will be removed from the queue
1074                  * as soon a a complete request arrives.
1075                  */
1076                 svc_sock_setbufsize(svsk->sk_sock,
1077                                     (serv->sv_nrthreads+3) * serv->sv_max_mesg,
1078                                     3 * serv->sv_max_mesg);
1079
1080         clear_bit(SK_DATA, &svsk->sk_flags);
1081
1082         /* Receive data. If we haven't got the record length yet, get
1083          * the next four bytes. Otherwise try to gobble up as much as
1084          * possible up to the complete record length.
1085          */
1086         if (svsk->sk_tcplen < 4) {
1087                 unsigned long   want = 4 - svsk->sk_tcplen;
1088                 struct kvec     iov;
1089
1090                 iov.iov_base = ((char *) &svsk->sk_reclen) + svsk->sk_tcplen;
1091                 iov.iov_len  = want;
1092                 if ((len = svc_recvfrom(rqstp, &iov, 1, want)) < 0)
1093                         goto error;
1094                 svsk->sk_tcplen += len;
1095
1096                 if (len < want) {
1097                         dprintk("svc: short recvfrom while reading record length (%d of %lu)\n",
1098                                 len, want);
1099                         svc_sock_received(svsk);
1100                         return -EAGAIN; /* record header not complete */
1101                 }
1102
1103                 svsk->sk_reclen = ntohl(svsk->sk_reclen);
1104                 if (!(svsk->sk_reclen & 0x80000000)) {
1105                         /* FIXME: technically, a record can be fragmented,
1106                          *  and non-terminal fragments will not have the top
1107                          *  bit set in the fragment length header.
1108                          *  But apparently no known nfs clients send fragmented
1109                          *  records. */
1110                         if (net_ratelimit())
1111                                 printk(KERN_NOTICE "RPC: bad TCP reclen 0x%08lx"
1112                                        " (non-terminal)\n",
1113                                        (unsigned long) svsk->sk_reclen);
1114                         goto err_delete;
1115                 }
1116                 svsk->sk_reclen &= 0x7fffffff;
1117                 dprintk("svc: TCP record, %d bytes\n", svsk->sk_reclen);
1118                 if (svsk->sk_reclen > serv->sv_max_mesg) {
1119                         if (net_ratelimit())
1120                                 printk(KERN_NOTICE "RPC: bad TCP reclen 0x%08lx"
1121                                        " (large)\n",
1122                                        (unsigned long) svsk->sk_reclen);
1123                         goto err_delete;
1124                 }
1125         }
1126
1127         /* Check whether enough data is available */
1128         len = svc_recv_available(svsk);
1129         if (len < 0)
1130                 goto error;
1131
1132         if (len < svsk->sk_reclen) {
1133                 dprintk("svc: incomplete TCP record (%d of %d)\n",
1134                         len, svsk->sk_reclen);
1135                 svc_sock_received(svsk);
1136                 return -EAGAIN; /* record not complete */
1137         }
1138         len = svsk->sk_reclen;
1139         set_bit(SK_DATA, &svsk->sk_flags);
1140
1141         vec = rqstp->rq_vec;
1142         vec[0] = rqstp->rq_arg.head[0];
1143         vlen = PAGE_SIZE;
1144         pnum = 1;
1145         while (vlen < len) {
1146                 vec[pnum].iov_base = page_address(rqstp->rq_pages[pnum]);
1147                 vec[pnum].iov_len = PAGE_SIZE;
1148                 pnum++;
1149                 vlen += PAGE_SIZE;
1150         }
1151         rqstp->rq_respages = &rqstp->rq_pages[pnum];
1152
1153         /* Now receive data */
1154         len = svc_recvfrom(rqstp, vec, pnum, len);
1155         if (len < 0)
1156                 goto error;
1157
1158         dprintk("svc: TCP complete record (%d bytes)\n", len);
1159         rqstp->rq_arg.len = len;
1160         rqstp->rq_arg.page_base = 0;
1161         if (len <= rqstp->rq_arg.head[0].iov_len) {
1162                 rqstp->rq_arg.head[0].iov_len = len;
1163                 rqstp->rq_arg.page_len = 0;
1164         } else {
1165                 rqstp->rq_arg.page_len = len - rqstp->rq_arg.head[0].iov_len;
1166         }
1167
1168         rqstp->rq_skbuff      = NULL;
1169         rqstp->rq_prot        = IPPROTO_TCP;
1170
1171         /* Reset TCP read info */
1172         svsk->sk_reclen = 0;
1173         svsk->sk_tcplen = 0;
1174
1175         svc_sock_received(svsk);
1176         if (serv->sv_stats)
1177                 serv->sv_stats->nettcpcnt++;
1178
1179         return len;
1180
1181  err_delete:
1182         svc_delete_socket(svsk);
1183         return -EAGAIN;
1184
1185  error:
1186         if (len == -EAGAIN) {
1187                 dprintk("RPC: TCP recvfrom got EAGAIN\n");
1188                 svc_sock_received(svsk);
1189         } else {
1190                 printk(KERN_NOTICE "%s: recvfrom returned errno %d\n",
1191                                         svsk->sk_server->sv_name, -len);
1192                 goto err_delete;
1193         }
1194
1195         return len;
1196 }
1197
1198 /*
1199  * Send out data on TCP socket.
1200  */
1201 static int
1202 svc_tcp_sendto(struct svc_rqst *rqstp)
1203 {
1204         struct xdr_buf  *xbufp = &rqstp->rq_res;
1205         int sent;
1206         __be32 reclen;
1207
1208         /* Set up the first element of the reply kvec.
1209          * Any other kvecs that may be in use have been taken
1210          * care of by the server implementation itself.
1211          */
1212         reclen = htonl(0x80000000|((xbufp->len ) - 4));
1213         memcpy(xbufp->head[0].iov_base, &reclen, 4);
1214
1215         if (test_bit(SK_DEAD, &rqstp->rq_sock->sk_flags))
1216                 return -ENOTCONN;
1217
1218         sent = svc_sendto(rqstp, &rqstp->rq_res);
1219         if (sent != xbufp->len) {
1220                 printk(KERN_NOTICE "rpc-srv/tcp: %s: %s %d when sending %d bytes - shutting down socket\n",
1221                        rqstp->rq_sock->sk_server->sv_name,
1222                        (sent<0)?"got error":"sent only",
1223                        sent, xbufp->len);
1224                 set_bit(SK_CLOSE, &rqstp->rq_sock->sk_flags);
1225                 svc_sock_enqueue(rqstp->rq_sock);
1226                 sent = -EAGAIN;
1227         }
1228         return sent;
1229 }
1230
1231 static void
1232 svc_tcp_init(struct svc_sock *svsk)
1233 {
1234         struct sock     *sk = svsk->sk_sk;
1235         struct tcp_sock *tp = tcp_sk(sk);
1236
1237         svsk->sk_recvfrom = svc_tcp_recvfrom;
1238         svsk->sk_sendto = svc_tcp_sendto;
1239
1240         if (sk->sk_state == TCP_LISTEN) {
1241                 dprintk("setting up TCP socket for listening\n");
1242                 sk->sk_data_ready = svc_tcp_listen_data_ready;
1243                 set_bit(SK_CONN, &svsk->sk_flags);
1244         } else {
1245                 dprintk("setting up TCP socket for reading\n");
1246                 sk->sk_state_change = svc_tcp_state_change;
1247                 sk->sk_data_ready = svc_tcp_data_ready;
1248                 sk->sk_write_space = svc_write_space;
1249
1250                 svsk->sk_reclen = 0;
1251                 svsk->sk_tcplen = 0;
1252
1253                 tp->nonagle = 1;        /* disable Nagle's algorithm */
1254
1255                 /* initialise setting must have enough space to
1256                  * receive and respond to one request.
1257                  * svc_tcp_recvfrom will re-adjust if necessary
1258                  */
1259                 svc_sock_setbufsize(svsk->sk_sock,
1260                                     3 * svsk->sk_server->sv_max_mesg,
1261                                     3 * svsk->sk_server->sv_max_mesg);
1262
1263                 set_bit(SK_CHNGBUF, &svsk->sk_flags);
1264                 set_bit(SK_DATA, &svsk->sk_flags);
1265                 if (sk->sk_state != TCP_ESTABLISHED)
1266                         set_bit(SK_CLOSE, &svsk->sk_flags);
1267         }
1268 }
1269
1270 void
1271 svc_sock_update_bufs(struct svc_serv *serv)
1272 {
1273         /*
1274          * The number of server threads has changed. Update
1275          * rcvbuf and sndbuf accordingly on all sockets
1276          */
1277         struct list_head *le;
1278
1279         spin_lock_bh(&serv->sv_lock);
1280         list_for_each(le, &serv->sv_permsocks) {
1281                 struct svc_sock *svsk =
1282                         list_entry(le, struct svc_sock, sk_list);
1283                 set_bit(SK_CHNGBUF, &svsk->sk_flags);
1284         }
1285         list_for_each(le, &serv->sv_tempsocks) {
1286                 struct svc_sock *svsk =
1287                         list_entry(le, struct svc_sock, sk_list);
1288                 set_bit(SK_CHNGBUF, &svsk->sk_flags);
1289         }
1290         spin_unlock_bh(&serv->sv_lock);
1291 }
1292
1293 /*
1294  * Receive the next request on any socket.  This code is carefully
1295  * organised not to touch any cachelines in the shared svc_serv
1296  * structure, only cachelines in the local svc_pool.
1297  */
1298 int
1299 svc_recv(struct svc_rqst *rqstp, long timeout)
1300 {
1301         struct svc_sock         *svsk =NULL;
1302         struct svc_serv         *serv = rqstp->rq_server;
1303         struct svc_pool         *pool = rqstp->rq_pool;
1304         int                     len, i;
1305         int                     pages;
1306         struct xdr_buf          *arg;
1307         DECLARE_WAITQUEUE(wait, current);
1308
1309         dprintk("svc: server %p waiting for data (to = %ld)\n",
1310                 rqstp, timeout);
1311
1312         if (rqstp->rq_sock)
1313                 printk(KERN_ERR
1314                         "svc_recv: service %p, socket not NULL!\n",
1315                          rqstp);
1316         if (waitqueue_active(&rqstp->rq_wait))
1317                 printk(KERN_ERR
1318                         "svc_recv: service %p, wait queue active!\n",
1319                          rqstp);
1320
1321
1322         /* now allocate needed pages.  If we get a failure, sleep briefly */
1323         pages = (serv->sv_max_mesg + PAGE_SIZE) / PAGE_SIZE;
1324         for (i=0; i < pages ; i++)
1325                 while (rqstp->rq_pages[i] == NULL) {
1326                         struct page *p = alloc_page(GFP_KERNEL);
1327                         if (!p)
1328                                 schedule_timeout_uninterruptible(msecs_to_jiffies(500));
1329                         rqstp->rq_pages[i] = p;
1330                 }
1331         rqstp->rq_pages[i++] = NULL; /* this might be seen in nfs_read_actor */
1332         BUG_ON(pages >= RPCSVC_MAXPAGES);
1333
1334         /* Make arg->head point to first page and arg->pages point to rest */
1335         arg = &rqstp->rq_arg;
1336         arg->head[0].iov_base = page_address(rqstp->rq_pages[0]);
1337         arg->head[0].iov_len = PAGE_SIZE;
1338         arg->pages = rqstp->rq_pages + 1;
1339         arg->page_base = 0;
1340         /* save at least one page for response */
1341         arg->page_len = (pages-2)*PAGE_SIZE;
1342         arg->len = (pages-1)*PAGE_SIZE;
1343         arg->tail[0].iov_len = 0;
1344
1345         try_to_freeze();
1346         cond_resched();
1347         if (signalled())
1348                 return -EINTR;
1349
1350         spin_lock_bh(&pool->sp_lock);
1351         if ((svsk = svc_sock_dequeue(pool)) != NULL) {
1352                 rqstp->rq_sock = svsk;
1353                 atomic_inc(&svsk->sk_inuse);
1354                 rqstp->rq_reserved = serv->sv_max_mesg;
1355                 atomic_add(rqstp->rq_reserved, &svsk->sk_reserved);
1356         } else {
1357                 /* No data pending. Go to sleep */
1358                 svc_thread_enqueue(pool, rqstp);
1359
1360                 /*
1361                  * We have to be able to interrupt this wait
1362                  * to bring down the daemons ...
1363                  */
1364                 set_current_state(TASK_INTERRUPTIBLE);
1365                 add_wait_queue(&rqstp->rq_wait, &wait);
1366                 spin_unlock_bh(&pool->sp_lock);
1367
1368                 schedule_timeout(timeout);
1369
1370                 try_to_freeze();
1371
1372                 spin_lock_bh(&pool->sp_lock);
1373                 remove_wait_queue(&rqstp->rq_wait, &wait);
1374
1375                 if (!(svsk = rqstp->rq_sock)) {
1376                         svc_thread_dequeue(pool, rqstp);
1377                         spin_unlock_bh(&pool->sp_lock);
1378                         dprintk("svc: server %p, no data yet\n", rqstp);
1379                         return signalled()? -EINTR : -EAGAIN;
1380                 }
1381         }
1382         spin_unlock_bh(&pool->sp_lock);
1383
1384         dprintk("svc: server %p, pool %u, socket %p, inuse=%d\n",
1385                  rqstp, pool->sp_id, svsk, atomic_read(&svsk->sk_inuse));
1386         len = svsk->sk_recvfrom(rqstp);
1387         dprintk("svc: got len=%d\n", len);
1388
1389         /* No data, incomplete (TCP) read, or accept() */
1390         if (len == 0 || len == -EAGAIN) {
1391                 rqstp->rq_res.len = 0;
1392                 svc_sock_release(rqstp);
1393                 return -EAGAIN;
1394         }
1395         svsk->sk_lastrecv = get_seconds();
1396         clear_bit(SK_OLD, &svsk->sk_flags);
1397
1398         rqstp->rq_secure  = ntohs(rqstp->rq_addr.sin_port) < 1024;
1399         rqstp->rq_chandle.defer = svc_defer;
1400
1401         if (serv->sv_stats)
1402                 serv->sv_stats->netcnt++;
1403         return len;
1404 }
1405
1406 /*
1407  * Drop request
1408  */
1409 void
1410 svc_drop(struct svc_rqst *rqstp)
1411 {
1412         dprintk("svc: socket %p dropped request\n", rqstp->rq_sock);
1413         svc_sock_release(rqstp);
1414 }
1415
1416 /*
1417  * Return reply to client.
1418  */
1419 int
1420 svc_send(struct svc_rqst *rqstp)
1421 {
1422         struct svc_sock *svsk;
1423         int             len;
1424         struct xdr_buf  *xb;
1425
1426         if ((svsk = rqstp->rq_sock) == NULL) {
1427                 printk(KERN_WARNING "NULL socket pointer in %s:%d\n",
1428                                 __FILE__, __LINE__);
1429                 return -EFAULT;
1430         }
1431
1432         /* release the receive skb before sending the reply */
1433         svc_release_skb(rqstp);
1434
1435         /* calculate over-all length */
1436         xb = & rqstp->rq_res;
1437         xb->len = xb->head[0].iov_len +
1438                 xb->page_len +
1439                 xb->tail[0].iov_len;
1440
1441         /* Grab svsk->sk_mutex to serialize outgoing data. */
1442         mutex_lock(&svsk->sk_mutex);
1443         if (test_bit(SK_DEAD, &svsk->sk_flags))
1444                 len = -ENOTCONN;
1445         else
1446                 len = svsk->sk_sendto(rqstp);
1447         mutex_unlock(&svsk->sk_mutex);
1448         svc_sock_release(rqstp);
1449
1450         if (len == -ECONNREFUSED || len == -ENOTCONN || len == -EAGAIN)
1451                 return 0;
1452         return len;
1453 }
1454
1455 /*
1456  * Timer function to close old temporary sockets, using
1457  * a mark-and-sweep algorithm.
1458  */
1459 static void
1460 svc_age_temp_sockets(unsigned long closure)
1461 {
1462         struct svc_serv *serv = (struct svc_serv *)closure;
1463         struct svc_sock *svsk;
1464         struct list_head *le, *next;
1465         LIST_HEAD(to_be_aged);
1466
1467         dprintk("svc_age_temp_sockets\n");
1468
1469         if (!spin_trylock_bh(&serv->sv_lock)) {
1470                 /* busy, try again 1 sec later */
1471                 dprintk("svc_age_temp_sockets: busy\n");
1472                 mod_timer(&serv->sv_temptimer, jiffies + HZ);
1473                 return;
1474         }
1475
1476         list_for_each_safe(le, next, &serv->sv_tempsocks) {
1477                 svsk = list_entry(le, struct svc_sock, sk_list);
1478
1479                 if (!test_and_set_bit(SK_OLD, &svsk->sk_flags))
1480                         continue;
1481                 if (atomic_read(&svsk->sk_inuse) || test_bit(SK_BUSY, &svsk->sk_flags))
1482                         continue;
1483                 atomic_inc(&svsk->sk_inuse);
1484                 list_move(le, &to_be_aged);
1485                 set_bit(SK_CLOSE, &svsk->sk_flags);
1486                 set_bit(SK_DETACHED, &svsk->sk_flags);
1487         }
1488         spin_unlock_bh(&serv->sv_lock);
1489
1490         while (!list_empty(&to_be_aged)) {
1491                 le = to_be_aged.next;
1492                 /* fiddling the sk_list node is safe 'cos we're SK_DETACHED */
1493                 list_del_init(le);
1494                 svsk = list_entry(le, struct svc_sock, sk_list);
1495
1496                 dprintk("queuing svsk %p for closing, %lu seconds old\n",
1497                         svsk, get_seconds() - svsk->sk_lastrecv);
1498
1499                 /* a thread will dequeue and close it soon */
1500                 svc_sock_enqueue(svsk);
1501                 svc_sock_put(svsk);
1502         }
1503
1504         mod_timer(&serv->sv_temptimer, jiffies + svc_conn_age_period * HZ);
1505 }
1506
1507 /*
1508  * Initialize socket for RPC use and create svc_sock struct
1509  * XXX: May want to setsockopt SO_SNDBUF and SO_RCVBUF.
1510  */
1511 static struct svc_sock *svc_setup_socket(struct svc_serv *serv,
1512                                                 struct socket *sock,
1513                                                 int *errp, int flags)
1514 {
1515         struct svc_sock *svsk;
1516         struct sock     *inet;
1517         int             pmap_register = !(flags & SVC_SOCK_ANONYMOUS);
1518         int             is_temporary = flags & SVC_SOCK_TEMPORARY;
1519
1520         dprintk("svc: svc_setup_socket %p\n", sock);
1521         if (!(svsk = kzalloc(sizeof(*svsk), GFP_KERNEL))) {
1522                 *errp = -ENOMEM;
1523                 return NULL;
1524         }
1525
1526         inet = sock->sk;
1527
1528         /* Register socket with portmapper */
1529         if (*errp >= 0 && pmap_register)
1530                 *errp = svc_register(serv, inet->sk_protocol,
1531                                      ntohs(inet_sk(inet)->sport));
1532
1533         if (*errp < 0) {
1534                 kfree(svsk);
1535                 return NULL;
1536         }
1537
1538         set_bit(SK_BUSY, &svsk->sk_flags);
1539         inet->sk_user_data = svsk;
1540         svsk->sk_sock = sock;
1541         svsk->sk_sk = inet;
1542         svsk->sk_ostate = inet->sk_state_change;
1543         svsk->sk_odata = inet->sk_data_ready;
1544         svsk->sk_owspace = inet->sk_write_space;
1545         svsk->sk_server = serv;
1546         atomic_set(&svsk->sk_inuse, 1);
1547         svsk->sk_lastrecv = get_seconds();
1548         spin_lock_init(&svsk->sk_defer_lock);
1549         INIT_LIST_HEAD(&svsk->sk_deferred);
1550         INIT_LIST_HEAD(&svsk->sk_ready);
1551         mutex_init(&svsk->sk_mutex);
1552
1553         /* Initialize the socket */
1554         if (sock->type == SOCK_DGRAM)
1555                 svc_udp_init(svsk);
1556         else
1557                 svc_tcp_init(svsk);
1558
1559         spin_lock_bh(&serv->sv_lock);
1560         if (is_temporary) {
1561                 set_bit(SK_TEMP, &svsk->sk_flags);
1562                 list_add(&svsk->sk_list, &serv->sv_tempsocks);
1563                 serv->sv_tmpcnt++;
1564                 if (serv->sv_temptimer.function == NULL) {
1565                         /* setup timer to age temp sockets */
1566                         setup_timer(&serv->sv_temptimer, svc_age_temp_sockets,
1567                                         (unsigned long)serv);
1568                         mod_timer(&serv->sv_temptimer,
1569                                         jiffies + svc_conn_age_period * HZ);
1570                 }
1571         } else {
1572                 clear_bit(SK_TEMP, &svsk->sk_flags);
1573                 list_add(&svsk->sk_list, &serv->sv_permsocks);
1574         }
1575         spin_unlock_bh(&serv->sv_lock);
1576
1577         dprintk("svc: svc_setup_socket created %p (inet %p)\n",
1578                                 svsk, svsk->sk_sk);
1579
1580         return svsk;
1581 }
1582
1583 int svc_addsock(struct svc_serv *serv,
1584                 int fd,
1585                 char *name_return,
1586                 int *proto)
1587 {
1588         int err = 0;
1589         struct socket *so = sockfd_lookup(fd, &err);
1590         struct svc_sock *svsk = NULL;
1591
1592         if (!so)
1593                 return err;
1594         if (so->sk->sk_family != AF_INET)
1595                 err =  -EAFNOSUPPORT;
1596         else if (so->sk->sk_protocol != IPPROTO_TCP &&
1597             so->sk->sk_protocol != IPPROTO_UDP)
1598                 err =  -EPROTONOSUPPORT;
1599         else if (so->state > SS_UNCONNECTED)
1600                 err = -EISCONN;
1601         else {
1602                 svsk = svc_setup_socket(serv, so, &err, SVC_SOCK_DEFAULTS);
1603                 if (svsk) {
1604                         svc_sock_received(svsk);
1605                         err = 0;
1606                 }
1607         }
1608         if (err) {
1609                 sockfd_put(so);
1610                 return err;
1611         }
1612         if (proto) *proto = so->sk->sk_protocol;
1613         return one_sock_name(name_return, svsk);
1614 }
1615 EXPORT_SYMBOL_GPL(svc_addsock);
1616
1617 /*
1618  * Create socket for RPC service.
1619  */
1620 static int svc_create_socket(struct svc_serv *serv, int protocol,
1621                                 struct sockaddr_in *sin, int flags)
1622 {
1623         struct svc_sock *svsk;
1624         struct socket   *sock;
1625         int             error;
1626         int             type;
1627         char            buf[RPC_MAX_ADDRBUFLEN];
1628
1629         dprintk("svc: svc_create_socket(%s, %d, %s)\n",
1630                         serv->sv_program->pg_name, protocol,
1631                         __svc_print_addr((struct sockaddr *) sin, buf,
1632                                                                 sizeof(buf)));
1633
1634         if (protocol != IPPROTO_UDP && protocol != IPPROTO_TCP) {
1635                 printk(KERN_WARNING "svc: only UDP and TCP "
1636                                 "sockets supported\n");
1637                 return -EINVAL;
1638         }
1639         type = (protocol == IPPROTO_UDP)? SOCK_DGRAM : SOCK_STREAM;
1640
1641         if ((error = sock_create_kern(PF_INET, type, protocol, &sock)) < 0)
1642                 return error;
1643
1644         svc_reclassify_socket(sock);
1645
1646         if (type == SOCK_STREAM)
1647                 sock->sk->sk_reuse = 1; /* allow address reuse */
1648         error = kernel_bind(sock, (struct sockaddr *) sin,
1649                                         sizeof(*sin));
1650         if (error < 0)
1651                 goto bummer;
1652
1653         if (protocol == IPPROTO_TCP) {
1654                 if ((error = kernel_listen(sock, 64)) < 0)
1655                         goto bummer;
1656         }
1657
1658         if ((svsk = svc_setup_socket(serv, sock, &error, flags)) != NULL) {
1659                 svc_sock_received(svsk);
1660                 return ntohs(inet_sk(svsk->sk_sk)->sport);
1661         }
1662
1663 bummer:
1664         dprintk("svc: svc_create_socket error = %d\n", -error);
1665         sock_release(sock);
1666         return error;
1667 }
1668
1669 /*
1670  * Remove a dead socket
1671  */
1672 static void
1673 svc_delete_socket(struct svc_sock *svsk)
1674 {
1675         struct svc_serv *serv;
1676         struct sock     *sk;
1677
1678         dprintk("svc: svc_delete_socket(%p)\n", svsk);
1679
1680         serv = svsk->sk_server;
1681         sk = svsk->sk_sk;
1682
1683         sk->sk_state_change = svsk->sk_ostate;
1684         sk->sk_data_ready = svsk->sk_odata;
1685         sk->sk_write_space = svsk->sk_owspace;
1686
1687         spin_lock_bh(&serv->sv_lock);
1688
1689         if (!test_and_set_bit(SK_DETACHED, &svsk->sk_flags))
1690                 list_del_init(&svsk->sk_list);
1691         /*
1692          * We used to delete the svc_sock from whichever list
1693          * it's sk_ready node was on, but we don't actually
1694          * need to.  This is because the only time we're called
1695          * while still attached to a queue, the queue itself
1696          * is about to be destroyed (in svc_destroy).
1697          */
1698         if (!test_and_set_bit(SK_DEAD, &svsk->sk_flags)) {
1699                 BUG_ON(atomic_read(&svsk->sk_inuse)<2);
1700                 atomic_dec(&svsk->sk_inuse);
1701                 if (test_bit(SK_TEMP, &svsk->sk_flags))
1702                         serv->sv_tmpcnt--;
1703         }
1704
1705         spin_unlock_bh(&serv->sv_lock);
1706 }
1707
1708 void svc_close_socket(struct svc_sock *svsk)
1709 {
1710         set_bit(SK_CLOSE, &svsk->sk_flags);
1711         if (test_and_set_bit(SK_BUSY, &svsk->sk_flags))
1712                 /* someone else will have to effect the close */
1713                 return;
1714
1715         atomic_inc(&svsk->sk_inuse);
1716         svc_delete_socket(svsk);
1717         clear_bit(SK_BUSY, &svsk->sk_flags);
1718         svc_sock_put(svsk);
1719 }
1720
1721 /**
1722  * svc_makesock - Make a socket for nfsd and lockd
1723  * @serv: RPC server structure
1724  * @protocol: transport protocol to use
1725  * @port: port to use
1726  * @flags: requested socket characteristics
1727  *
1728  */
1729 int svc_makesock(struct svc_serv *serv, int protocol, unsigned short port,
1730                         int flags)
1731 {
1732         struct sockaddr_in sin = {
1733                 .sin_family             = AF_INET,
1734                 .sin_addr.s_addr        = INADDR_ANY,
1735                 .sin_port               = htons(port),
1736         };
1737
1738         dprintk("svc: creating socket proto = %d\n", protocol);
1739         return svc_create_socket(serv, protocol, &sin, flags);
1740 }
1741
1742 /*
1743  * Handle defer and revisit of requests
1744  */
1745
1746 static void svc_revisit(struct cache_deferred_req *dreq, int too_many)
1747 {
1748         struct svc_deferred_req *dr = container_of(dreq, struct svc_deferred_req, handle);
1749         struct svc_sock *svsk;
1750
1751         if (too_many) {
1752                 svc_sock_put(dr->svsk);
1753                 kfree(dr);
1754                 return;
1755         }
1756         dprintk("revisit queued\n");
1757         svsk = dr->svsk;
1758         dr->svsk = NULL;
1759         spin_lock_bh(&svsk->sk_defer_lock);
1760         list_add(&dr->handle.recent, &svsk->sk_deferred);
1761         spin_unlock_bh(&svsk->sk_defer_lock);
1762         set_bit(SK_DEFERRED, &svsk->sk_flags);
1763         svc_sock_enqueue(svsk);
1764         svc_sock_put(svsk);
1765 }
1766
1767 static struct cache_deferred_req *
1768 svc_defer(struct cache_req *req)
1769 {
1770         struct svc_rqst *rqstp = container_of(req, struct svc_rqst, rq_chandle);
1771         int size = sizeof(struct svc_deferred_req) + (rqstp->rq_arg.len);
1772         struct svc_deferred_req *dr;
1773
1774         if (rqstp->rq_arg.page_len)
1775                 return NULL; /* if more than a page, give up FIXME */
1776         if (rqstp->rq_deferred) {
1777                 dr = rqstp->rq_deferred;
1778                 rqstp->rq_deferred = NULL;
1779         } else {
1780                 int skip  = rqstp->rq_arg.len - rqstp->rq_arg.head[0].iov_len;
1781                 /* FIXME maybe discard if size too large */
1782                 dr = kmalloc(size, GFP_KERNEL);
1783                 if (dr == NULL)
1784                         return NULL;
1785
1786                 dr->handle.owner = rqstp->rq_server;
1787                 dr->prot = rqstp->rq_prot;
1788                 dr->addr = rqstp->rq_addr;
1789                 dr->daddr = rqstp->rq_daddr;
1790                 dr->argslen = rqstp->rq_arg.len >> 2;
1791                 memcpy(dr->args, rqstp->rq_arg.head[0].iov_base-skip, dr->argslen<<2);
1792         }
1793         atomic_inc(&rqstp->rq_sock->sk_inuse);
1794         dr->svsk = rqstp->rq_sock;
1795
1796         dr->handle.revisit = svc_revisit;
1797         return &dr->handle;
1798 }
1799
1800 /*
1801  * recv data from a deferred request into an active one
1802  */
1803 static int svc_deferred_recv(struct svc_rqst *rqstp)
1804 {
1805         struct svc_deferred_req *dr = rqstp->rq_deferred;
1806
1807         rqstp->rq_arg.head[0].iov_base = dr->args;
1808         rqstp->rq_arg.head[0].iov_len = dr->argslen<<2;
1809         rqstp->rq_arg.page_len = 0;
1810         rqstp->rq_arg.len = dr->argslen<<2;
1811         rqstp->rq_prot        = dr->prot;
1812         rqstp->rq_addr        = dr->addr;
1813         rqstp->rq_daddr       = dr->daddr;
1814         rqstp->rq_respages    = rqstp->rq_pages;
1815         return dr->argslen<<2;
1816 }
1817
1818
1819 static struct svc_deferred_req *svc_deferred_dequeue(struct svc_sock *svsk)
1820 {
1821         struct svc_deferred_req *dr = NULL;
1822
1823         if (!test_bit(SK_DEFERRED, &svsk->sk_flags))
1824                 return NULL;
1825         spin_lock_bh(&svsk->sk_defer_lock);
1826         clear_bit(SK_DEFERRED, &svsk->sk_flags);
1827         if (!list_empty(&svsk->sk_deferred)) {
1828                 dr = list_entry(svsk->sk_deferred.next,
1829                                 struct svc_deferred_req,
1830                                 handle.recent);
1831                 list_del_init(&dr->handle.recent);
1832                 set_bit(SK_DEFERRED, &svsk->sk_flags);
1833         }
1834         spin_unlock_bh(&svsk->sk_defer_lock);
1835         return dr;
1836 }