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