4ab137403e1ac3b0ff075b97dde73308bc843c8c
[linux-2.6.git] / net / sunrpc / svc.c
1 /*
2  * linux/net/sunrpc/svc.c
3  *
4  * High-level RPC service routines
5  *
6  * Copyright (C) 1995, 1996 Olaf Kirch <okir@monad.swb.de>
7  *
8  * Multiple threads pools and NUMAisation
9  * Copyright (c) 2006 Silicon Graphics, Inc.
10  * by Greg Banks <gnb@melbourne.sgi.com>
11  */
12
13 #include <linux/linkage.h>
14 #include <linux/sched.h>
15 #include <linux/errno.h>
16 #include <linux/net.h>
17 #include <linux/in.h>
18 #include <linux/mm.h>
19 #include <linux/interrupt.h>
20 #include <linux/module.h>
21
22 #include <linux/sunrpc/types.h>
23 #include <linux/sunrpc/xdr.h>
24 #include <linux/sunrpc/stats.h>
25 #include <linux/sunrpc/svcsock.h>
26 #include <linux/sunrpc/clnt.h>
27
28 #define RPCDBG_FACILITY RPCDBG_SVCDSP
29
30 /*
31  * Mode for mapping cpus to pools.
32  */
33 enum {
34         SVC_POOL_NONE = -1,     /* uninitialised, choose one of the others */
35         SVC_POOL_GLOBAL,        /* no mapping, just a single global pool
36                                  * (legacy & UP mode) */
37         SVC_POOL_PERCPU,        /* one pool per cpu */
38         SVC_POOL_PERNODE        /* one pool per numa node */
39 };
40
41 /*
42  * Structure for mapping cpus to pools and vice versa.
43  * Setup once during sunrpc initialisation.
44  */
45 static struct svc_pool_map {
46         int mode;                       /* Note: int not enum to avoid
47                                          * warnings about "enumeration value
48                                          * not handled in switch" */
49         unsigned int npools;
50         unsigned int *pool_to;          /* maps pool id to cpu or node */
51         unsigned int *to_pool;          /* maps cpu or node to pool id */
52 } svc_pool_map = {
53         .mode = SVC_POOL_NONE
54 };
55
56
57 /*
58  * Detect best pool mapping mode heuristically,
59  * according to the machine's topology.
60  */
61 static int
62 svc_pool_map_choose_mode(void)
63 {
64         unsigned int node;
65
66         if (num_online_nodes() > 1) {
67                 /*
68                  * Actually have multiple NUMA nodes,
69                  * so split pools on NUMA node boundaries
70                  */
71                 return SVC_POOL_PERNODE;
72         }
73
74         node = any_online_node(node_online_map);
75         if (nr_cpus_node(node) > 2) {
76                 /*
77                  * Non-trivial SMP, or CONFIG_NUMA on
78                  * non-NUMA hardware, e.g. with a generic
79                  * x86_64 kernel on Xeons.  In this case we
80                  * want to divide the pools on cpu boundaries.
81                  */
82                 return SVC_POOL_PERCPU;
83         }
84
85         /* default: one global pool */
86         return SVC_POOL_GLOBAL;
87 }
88
89 /*
90  * Allocate the to_pool[] and pool_to[] arrays.
91  * Returns 0 on success or an errno.
92  */
93 static int
94 svc_pool_map_alloc_arrays(struct svc_pool_map *m, unsigned int maxpools)
95 {
96         m->to_pool = kcalloc(maxpools, sizeof(unsigned int), GFP_KERNEL);
97         if (!m->to_pool)
98                 goto fail;
99         m->pool_to = kcalloc(maxpools, sizeof(unsigned int), GFP_KERNEL);
100         if (!m->pool_to)
101                 goto fail_free;
102
103         return 0;
104
105 fail_free:
106         kfree(m->to_pool);
107 fail:
108         return -ENOMEM;
109 }
110
111 /*
112  * Initialise the pool map for SVC_POOL_PERCPU mode.
113  * Returns number of pools or <0 on error.
114  */
115 static int
116 svc_pool_map_init_percpu(struct svc_pool_map *m)
117 {
118         unsigned int maxpools = highest_possible_processor_id()+1;
119         unsigned int pidx = 0;
120         unsigned int cpu;
121         int err;
122
123         err = svc_pool_map_alloc_arrays(m, maxpools);
124         if (err)
125                 return err;
126
127         for_each_online_cpu(cpu) {
128                 BUG_ON(pidx > maxpools);
129                 m->to_pool[cpu] = pidx;
130                 m->pool_to[pidx] = cpu;
131                 pidx++;
132         }
133         /* cpus brought online later all get mapped to pool0, sorry */
134
135         return pidx;
136 };
137
138
139 /*
140  * Initialise the pool map for SVC_POOL_PERNODE mode.
141  * Returns number of pools or <0 on error.
142  */
143 static int
144 svc_pool_map_init_pernode(struct svc_pool_map *m)
145 {
146         unsigned int maxpools = highest_possible_node_id()+1;
147         unsigned int pidx = 0;
148         unsigned int node;
149         int err;
150
151         err = svc_pool_map_alloc_arrays(m, maxpools);
152         if (err)
153                 return err;
154
155         for_each_node_with_cpus(node) {
156                 /* some architectures (e.g. SN2) have cpuless nodes */
157                 BUG_ON(pidx > maxpools);
158                 m->to_pool[node] = pidx;
159                 m->pool_to[pidx] = node;
160                 pidx++;
161         }
162         /* nodes brought online later all get mapped to pool0, sorry */
163
164         return pidx;
165 }
166
167
168 /*
169  * Build the global map of cpus to pools and vice versa.
170  */
171 static unsigned int
172 svc_pool_map_init(void)
173 {
174         struct svc_pool_map *m = &svc_pool_map;
175         int npools = -1;
176
177         if (m->mode != SVC_POOL_NONE)
178                 return m->npools;
179
180         m->mode = svc_pool_map_choose_mode();
181
182         switch (m->mode) {
183         case SVC_POOL_PERCPU:
184                 npools = svc_pool_map_init_percpu(m);
185                 break;
186         case SVC_POOL_PERNODE:
187                 npools = svc_pool_map_init_pernode(m);
188                 break;
189         }
190
191         if (npools < 0) {
192                 /* default, or memory allocation failure */
193                 npools = 1;
194                 m->mode = SVC_POOL_GLOBAL;
195         }
196         m->npools = npools;
197
198         return m->npools;
199 }
200
201 /*
202  * Set the current thread's cpus_allowed mask so that it
203  * will only run on cpus in the given pool.
204  *
205  * Returns 1 and fills in oldmask iff a cpumask was applied.
206  */
207 static inline int
208 svc_pool_map_set_cpumask(unsigned int pidx, cpumask_t *oldmask)
209 {
210         struct svc_pool_map *m = &svc_pool_map;
211         unsigned int node; /* or cpu */
212
213         /*
214          * The caller checks for sv_nrpools > 1, which
215          * implies that we've been initialized and the
216          * map mode is not NONE.
217          */
218         BUG_ON(m->mode == SVC_POOL_NONE);
219
220         switch (m->mode)
221         {
222         default:
223                 return 0;
224         case SVC_POOL_PERCPU:
225                 node = m->pool_to[pidx];
226                 *oldmask = current->cpus_allowed;
227                 set_cpus_allowed(current, cpumask_of_cpu(node));
228                 return 1;
229         case SVC_POOL_PERNODE:
230                 node = m->pool_to[pidx];
231                 *oldmask = current->cpus_allowed;
232                 set_cpus_allowed(current, node_to_cpumask(node));
233                 return 1;
234         }
235 }
236
237 /*
238  * Use the mapping mode to choose a pool for a given CPU.
239  * Used when enqueueing an incoming RPC.  Always returns
240  * a non-NULL pool pointer.
241  */
242 struct svc_pool *
243 svc_pool_for_cpu(struct svc_serv *serv, int cpu)
244 {
245         struct svc_pool_map *m = &svc_pool_map;
246         unsigned int pidx = 0;
247
248         /*
249          * SVC_POOL_NONE happens in a pure client when
250          * lockd is brought up, so silently treat it the
251          * same as SVC_POOL_GLOBAL.
252          */
253
254         switch (m->mode) {
255         case SVC_POOL_PERCPU:
256                 pidx = m->to_pool[cpu];
257                 break;
258         case SVC_POOL_PERNODE:
259                 pidx = m->to_pool[cpu_to_node(cpu)];
260                 break;
261         }
262         return &serv->sv_pools[pidx % serv->sv_nrpools];
263 }
264
265
266 /*
267  * Create an RPC service
268  */
269 static struct svc_serv *
270 __svc_create(struct svc_program *prog, unsigned int bufsize, int npools,
271            void (*shutdown)(struct svc_serv *serv))
272 {
273         struct svc_serv *serv;
274         int vers;
275         unsigned int xdrsize;
276         unsigned int i;
277
278         if (!(serv = kzalloc(sizeof(*serv), GFP_KERNEL)))
279                 return NULL;
280         serv->sv_name      = prog->pg_name;
281         serv->sv_program   = prog;
282         serv->sv_nrthreads = 1;
283         serv->sv_stats     = prog->pg_stats;
284         if (bufsize > RPCSVC_MAXPAYLOAD)
285                 bufsize = RPCSVC_MAXPAYLOAD;
286         serv->sv_max_payload = bufsize? bufsize : 4096;
287         serv->sv_max_mesg  = roundup(serv->sv_max_payload + PAGE_SIZE, PAGE_SIZE);
288         serv->sv_shutdown  = shutdown;
289         xdrsize = 0;
290         while (prog) {
291                 prog->pg_lovers = prog->pg_nvers-1;
292                 for (vers=0; vers<prog->pg_nvers ; vers++)
293                         if (prog->pg_vers[vers]) {
294                                 prog->pg_hivers = vers;
295                                 if (prog->pg_lovers > vers)
296                                         prog->pg_lovers = vers;
297                                 if (prog->pg_vers[vers]->vs_xdrsize > xdrsize)
298                                         xdrsize = prog->pg_vers[vers]->vs_xdrsize;
299                         }
300                 prog = prog->pg_next;
301         }
302         serv->sv_xdrsize   = xdrsize;
303         INIT_LIST_HEAD(&serv->sv_tempsocks);
304         INIT_LIST_HEAD(&serv->sv_permsocks);
305         init_timer(&serv->sv_temptimer);
306         spin_lock_init(&serv->sv_lock);
307
308         serv->sv_nrpools = npools;
309         serv->sv_pools =
310                 kcalloc(serv->sv_nrpools, sizeof(struct svc_pool),
311                         GFP_KERNEL);
312         if (!serv->sv_pools) {
313                 kfree(serv);
314                 return NULL;
315         }
316
317         for (i = 0; i < serv->sv_nrpools; i++) {
318                 struct svc_pool *pool = &serv->sv_pools[i];
319
320                 dprintk("svc: initialising pool %u for %s\n",
321                                 i, serv->sv_name);
322
323                 pool->sp_id = i;
324                 INIT_LIST_HEAD(&pool->sp_threads);
325                 INIT_LIST_HEAD(&pool->sp_sockets);
326                 INIT_LIST_HEAD(&pool->sp_all_threads);
327                 spin_lock_init(&pool->sp_lock);
328         }
329
330
331         /* Remove any stale portmap registrations */
332         svc_register(serv, 0, 0);
333
334         return serv;
335 }
336
337 struct svc_serv *
338 svc_create(struct svc_program *prog, unsigned int bufsize,
339                 void (*shutdown)(struct svc_serv *serv))
340 {
341         return __svc_create(prog, bufsize, /*npools*/1, shutdown);
342 }
343
344 struct svc_serv *
345 svc_create_pooled(struct svc_program *prog, unsigned int bufsize,
346                 void (*shutdown)(struct svc_serv *serv),
347                   svc_thread_fn func, int sig, struct module *mod)
348 {
349         struct svc_serv *serv;
350         unsigned int npools = svc_pool_map_init();
351
352         serv = __svc_create(prog, bufsize, npools, shutdown);
353
354         if (serv != NULL) {
355                 serv->sv_function = func;
356                 serv->sv_kill_signal = sig;
357                 serv->sv_module = mod;
358         }
359
360         return serv;
361 }
362
363 /*
364  * Destroy an RPC service.  Should be called with the BKL held
365  */
366 void
367 svc_destroy(struct svc_serv *serv)
368 {
369         struct svc_sock *svsk;
370
371         dprintk("svc: svc_destroy(%s, %d)\n",
372                                 serv->sv_program->pg_name,
373                                 serv->sv_nrthreads);
374
375         if (serv->sv_nrthreads) {
376                 if (--(serv->sv_nrthreads) != 0) {
377                         svc_sock_update_bufs(serv);
378                         return;
379                 }
380         } else
381                 printk("svc_destroy: no threads for serv=%p!\n", serv);
382
383         del_timer_sync(&serv->sv_temptimer);
384
385         while (!list_empty(&serv->sv_tempsocks)) {
386                 svsk = list_entry(serv->sv_tempsocks.next,
387                                   struct svc_sock,
388                                   sk_list);
389                 svc_close_socket(svsk);
390         }
391         if (serv->sv_shutdown)
392                 serv->sv_shutdown(serv);
393
394         while (!list_empty(&serv->sv_permsocks)) {
395                 svsk = list_entry(serv->sv_permsocks.next,
396                                   struct svc_sock,
397                                   sk_list);
398                 svc_close_socket(svsk);
399         }
400
401         cache_clean_deferred(serv);
402
403         /* Unregister service with the portmapper */
404         svc_register(serv, 0, 0);
405         kfree(serv->sv_pools);
406         kfree(serv);
407 }
408
409 /*
410  * Allocate an RPC server's buffer space.
411  * We allocate pages and place them in rq_argpages.
412  */
413 static int
414 svc_init_buffer(struct svc_rqst *rqstp, unsigned int size)
415 {
416         int pages;
417         int arghi;
418
419         pages = size / PAGE_SIZE + 1; /* extra page as we hold both request and reply.
420                                        * We assume one is at most one page
421                                        */
422         arghi = 0;
423         BUG_ON(pages > RPCSVC_MAXPAGES);
424         while (pages) {
425                 struct page *p = alloc_page(GFP_KERNEL);
426                 if (!p)
427                         break;
428                 rqstp->rq_pages[arghi++] = p;
429                 pages--;
430         }
431         return ! pages;
432 }
433
434 /*
435  * Release an RPC server buffer
436  */
437 static void
438 svc_release_buffer(struct svc_rqst *rqstp)
439 {
440         int i;
441         for (i=0; i<ARRAY_SIZE(rqstp->rq_pages); i++)
442                 if (rqstp->rq_pages[i])
443                         put_page(rqstp->rq_pages[i]);
444 }
445
446 /*
447  * Create a thread in the given pool.  Caller must hold BKL.
448  * On a NUMA or SMP machine, with a multi-pool serv, the thread
449  * will be restricted to run on the cpus belonging to the pool.
450  */
451 static int
452 __svc_create_thread(svc_thread_fn func, struct svc_serv *serv,
453                     struct svc_pool *pool)
454 {
455         struct svc_rqst *rqstp;
456         int             error = -ENOMEM;
457         int             have_oldmask = 0;
458         cpumask_t       oldmask;
459
460         rqstp = kzalloc(sizeof(*rqstp), GFP_KERNEL);
461         if (!rqstp)
462                 goto out;
463
464         init_waitqueue_head(&rqstp->rq_wait);
465
466         if (!(rqstp->rq_argp = kmalloc(serv->sv_xdrsize, GFP_KERNEL))
467          || !(rqstp->rq_resp = kmalloc(serv->sv_xdrsize, GFP_KERNEL))
468          || !svc_init_buffer(rqstp, serv->sv_max_mesg))
469                 goto out_thread;
470
471         serv->sv_nrthreads++;
472         spin_lock_bh(&pool->sp_lock);
473         pool->sp_nrthreads++;
474         list_add(&rqstp->rq_all, &pool->sp_all_threads);
475         spin_unlock_bh(&pool->sp_lock);
476         rqstp->rq_server = serv;
477         rqstp->rq_pool = pool;
478
479         if (serv->sv_nrpools > 1)
480                 have_oldmask = svc_pool_map_set_cpumask(pool->sp_id, &oldmask);
481
482         error = kernel_thread((int (*)(void *)) func, rqstp, 0);
483
484         if (have_oldmask)
485                 set_cpus_allowed(current, oldmask);
486
487         if (error < 0)
488                 goto out_thread;
489         svc_sock_update_bufs(serv);
490         error = 0;
491 out:
492         return error;
493
494 out_thread:
495         svc_exit_thread(rqstp);
496         goto out;
497 }
498
499 /*
500  * Create a thread in the default pool.  Caller must hold BKL.
501  */
502 int
503 svc_create_thread(svc_thread_fn func, struct svc_serv *serv)
504 {
505         return __svc_create_thread(func, serv, &serv->sv_pools[0]);
506 }
507
508 /*
509  * Choose a pool in which to create a new thread, for svc_set_num_threads
510  */
511 static inline struct svc_pool *
512 choose_pool(struct svc_serv *serv, struct svc_pool *pool, unsigned int *state)
513 {
514         if (pool != NULL)
515                 return pool;
516
517         return &serv->sv_pools[(*state)++ % serv->sv_nrpools];
518 }
519
520 /*
521  * Choose a thread to kill, for svc_set_num_threads
522  */
523 static inline struct task_struct *
524 choose_victim(struct svc_serv *serv, struct svc_pool *pool, unsigned int *state)
525 {
526         unsigned int i;
527         struct task_struct *task = NULL;
528
529         if (pool != NULL) {
530                 spin_lock_bh(&pool->sp_lock);
531         } else {
532                 /* choose a pool in round-robin fashion */
533                 for (i = 0; i < serv->sv_nrpools; i++) {
534                         pool = &serv->sv_pools[--(*state) % serv->sv_nrpools];
535                         spin_lock_bh(&pool->sp_lock);
536                         if (!list_empty(&pool->sp_all_threads))
537                                 goto found_pool;
538                         spin_unlock_bh(&pool->sp_lock);
539                 }
540                 return NULL;
541         }
542
543 found_pool:
544         if (!list_empty(&pool->sp_all_threads)) {
545                 struct svc_rqst *rqstp;
546
547                 /*
548                  * Remove from the pool->sp_all_threads list
549                  * so we don't try to kill it again.
550                  */
551                 rqstp = list_entry(pool->sp_all_threads.next, struct svc_rqst, rq_all);
552                 list_del_init(&rqstp->rq_all);
553                 task = rqstp->rq_task;
554         }
555         spin_unlock_bh(&pool->sp_lock);
556
557         return task;
558 }
559
560 /*
561  * Create or destroy enough new threads to make the number
562  * of threads the given number.  If `pool' is non-NULL, applies
563  * only to threads in that pool, otherwise round-robins between
564  * all pools.  Must be called with a svc_get() reference and
565  * the BKL held.
566  *
567  * Destroying threads relies on the service threads filling in
568  * rqstp->rq_task, which only the nfs ones do.  Assumes the serv
569  * has been created using svc_create_pooled().
570  *
571  * Based on code that used to be in nfsd_svc() but tweaked
572  * to be pool-aware.
573  */
574 int
575 svc_set_num_threads(struct svc_serv *serv, struct svc_pool *pool, int nrservs)
576 {
577         struct task_struct *victim;
578         int error = 0;
579         unsigned int state = serv->sv_nrthreads-1;
580
581         if (pool == NULL) {
582                 /* The -1 assumes caller has done a svc_get() */
583                 nrservs -= (serv->sv_nrthreads-1);
584         } else {
585                 spin_lock_bh(&pool->sp_lock);
586                 nrservs -= pool->sp_nrthreads;
587                 spin_unlock_bh(&pool->sp_lock);
588         }
589
590         /* create new threads */
591         while (nrservs > 0) {
592                 nrservs--;
593                 __module_get(serv->sv_module);
594                 error = __svc_create_thread(serv->sv_function, serv,
595                                             choose_pool(serv, pool, &state));
596                 if (error < 0) {
597                         module_put(serv->sv_module);
598                         break;
599                 }
600         }
601         /* destroy old threads */
602         while (nrservs < 0 &&
603                (victim = choose_victim(serv, pool, &state)) != NULL) {
604                 send_sig(serv->sv_kill_signal, victim, 1);
605                 nrservs++;
606         }
607
608         return error;
609 }
610
611 /*
612  * Called from a server thread as it's exiting.  Caller must hold BKL.
613  */
614 void
615 svc_exit_thread(struct svc_rqst *rqstp)
616 {
617         struct svc_serv *serv = rqstp->rq_server;
618         struct svc_pool *pool = rqstp->rq_pool;
619
620         svc_release_buffer(rqstp);
621         kfree(rqstp->rq_resp);
622         kfree(rqstp->rq_argp);
623         kfree(rqstp->rq_auth_data);
624
625         spin_lock_bh(&pool->sp_lock);
626         pool->sp_nrthreads--;
627         list_del(&rqstp->rq_all);
628         spin_unlock_bh(&pool->sp_lock);
629
630         kfree(rqstp);
631
632         /* Release the server */
633         if (serv)
634                 svc_destroy(serv);
635 }
636
637 /*
638  * Register an RPC service with the local portmapper.
639  * To unregister a service, call this routine with
640  * proto and port == 0.
641  */
642 int
643 svc_register(struct svc_serv *serv, int proto, unsigned short port)
644 {
645         struct svc_program      *progp;
646         unsigned long           flags;
647         int                     i, error = 0, dummy;
648
649         if (!port)
650                 clear_thread_flag(TIF_SIGPENDING);
651
652         for (progp = serv->sv_program; progp; progp = progp->pg_next) {
653                 for (i = 0; i < progp->pg_nvers; i++) {
654                         if (progp->pg_vers[i] == NULL)
655                                 continue;
656
657                         dprintk("svc: svc_register(%s, %s, %d, %d)%s\n",
658                                         progp->pg_name,
659                                         proto == IPPROTO_UDP?  "udp" : "tcp",
660                                         port,
661                                         i,
662                                         progp->pg_vers[i]->vs_hidden?
663                                                 " (but not telling portmap)" : "");
664
665                         if (progp->pg_vers[i]->vs_hidden)
666                                 continue;
667
668                         error = rpc_register(progp->pg_prog, i, proto, port, &dummy);
669                         if (error < 0)
670                                 break;
671                         if (port && !dummy) {
672                                 error = -EACCES;
673                                 break;
674                         }
675                 }
676         }
677
678         if (!port) {
679                 spin_lock_irqsave(&current->sighand->siglock, flags);
680                 recalc_sigpending();
681                 spin_unlock_irqrestore(&current->sighand->siglock, flags);
682         }
683
684         return error;
685 }
686
687 /*
688  * Process the RPC request.
689  */
690 int
691 svc_process(struct svc_rqst *rqstp)
692 {
693         struct svc_program      *progp;
694         struct svc_version      *versp = NULL;  /* compiler food */
695         struct svc_procedure    *procp = NULL;
696         struct kvec *           argv = &rqstp->rq_arg.head[0];
697         struct kvec *           resv = &rqstp->rq_res.head[0];
698         struct svc_serv         *serv = rqstp->rq_server;
699         kxdrproc_t              xdr;
700         __be32                  *statp;
701         u32                     dir, prog, vers, proc;
702         __be32                  auth_stat, rpc_stat;
703         int                     auth_res;
704         __be32                  *reply_statp;
705
706         rpc_stat = rpc_success;
707
708         if (argv->iov_len < 6*4)
709                 goto err_short_len;
710
711         /* setup response xdr_buf.
712          * Initially it has just one page
713          */
714         rqstp->rq_resused = 1;
715         resv->iov_base = page_address(rqstp->rq_respages[0]);
716         resv->iov_len = 0;
717         rqstp->rq_res.pages = rqstp->rq_respages + 1;
718         rqstp->rq_res.len = 0;
719         rqstp->rq_res.page_base = 0;
720         rqstp->rq_res.page_len = 0;
721         rqstp->rq_res.buflen = PAGE_SIZE;
722         rqstp->rq_res.tail[0].iov_base = NULL;
723         rqstp->rq_res.tail[0].iov_len = 0;
724         /* Will be turned off only in gss privacy case: */
725         rqstp->rq_sendfile_ok = 1;
726         /* tcp needs a space for the record length... */
727         if (rqstp->rq_prot == IPPROTO_TCP)
728                 svc_putnl(resv, 0);
729
730         rqstp->rq_xid = svc_getu32(argv);
731         svc_putu32(resv, rqstp->rq_xid);
732
733         dir  = svc_getnl(argv);
734         vers = svc_getnl(argv);
735
736         /* First words of reply: */
737         svc_putnl(resv, 1);             /* REPLY */
738
739         if (dir != 0)           /* direction != CALL */
740                 goto err_bad_dir;
741         if (vers != 2)          /* RPC version number */
742                 goto err_bad_rpc;
743
744         /* Save position in case we later decide to reject: */
745         reply_statp = resv->iov_base + resv->iov_len;
746
747         svc_putnl(resv, 0);             /* ACCEPT */
748
749         rqstp->rq_prog = prog = svc_getnl(argv);        /* program number */
750         rqstp->rq_vers = vers = svc_getnl(argv);        /* version number */
751         rqstp->rq_proc = proc = svc_getnl(argv);        /* procedure number */
752
753         progp = serv->sv_program;
754
755         for (progp = serv->sv_program; progp; progp = progp->pg_next)
756                 if (prog == progp->pg_prog)
757                         break;
758
759         /*
760          * Decode auth data, and add verifier to reply buffer.
761          * We do this before anything else in order to get a decent
762          * auth verifier.
763          */
764         auth_res = svc_authenticate(rqstp, &auth_stat);
765         /* Also give the program a chance to reject this call: */
766         if (auth_res == SVC_OK && progp) {
767                 auth_stat = rpc_autherr_badcred;
768                 auth_res = progp->pg_authenticate(rqstp);
769         }
770         switch (auth_res) {
771         case SVC_OK:
772                 break;
773         case SVC_GARBAGE:
774                 rpc_stat = rpc_garbage_args;
775                 goto err_bad;
776         case SVC_SYSERR:
777                 rpc_stat = rpc_system_err;
778                 goto err_bad;
779         case SVC_DENIED:
780                 goto err_bad_auth;
781         case SVC_DROP:
782                 goto dropit;
783         case SVC_COMPLETE:
784                 goto sendit;
785         }
786
787         if (progp == NULL)
788                 goto err_bad_prog;
789
790         if (vers >= progp->pg_nvers ||
791           !(versp = progp->pg_vers[vers]))
792                 goto err_bad_vers;
793
794         procp = versp->vs_proc + proc;
795         if (proc >= versp->vs_nproc || !procp->pc_func)
796                 goto err_bad_proc;
797         rqstp->rq_server   = serv;
798         rqstp->rq_procinfo = procp;
799
800         /* Syntactic check complete */
801         serv->sv_stats->rpccnt++;
802
803         /* Build the reply header. */
804         statp = resv->iov_base +resv->iov_len;
805         svc_putnl(resv, RPC_SUCCESS);
806
807         /* Bump per-procedure stats counter */
808         procp->pc_count++;
809
810         /* Initialize storage for argp and resp */
811         memset(rqstp->rq_argp, 0, procp->pc_argsize);
812         memset(rqstp->rq_resp, 0, procp->pc_ressize);
813
814         /* un-reserve some of the out-queue now that we have a
815          * better idea of reply size
816          */
817         if (procp->pc_xdrressize)
818                 svc_reserve(rqstp, procp->pc_xdrressize<<2);
819
820         /* Call the function that processes the request. */
821         if (!versp->vs_dispatch) {
822                 /* Decode arguments */
823                 xdr = procp->pc_decode;
824                 if (xdr && !xdr(rqstp, argv->iov_base, rqstp->rq_argp))
825                         goto err_garbage;
826
827                 *statp = procp->pc_func(rqstp, rqstp->rq_argp, rqstp->rq_resp);
828
829                 /* Encode reply */
830                 if (*statp == rpc_drop_reply) {
831                         if (procp->pc_release)
832                                 procp->pc_release(rqstp, NULL, rqstp->rq_resp);
833                         goto dropit;
834                 }
835                 if (*statp == rpc_success && (xdr = procp->pc_encode)
836                  && !xdr(rqstp, resv->iov_base+resv->iov_len, rqstp->rq_resp)) {
837                         dprintk("svc: failed to encode reply\n");
838                         /* serv->sv_stats->rpcsystemerr++; */
839                         *statp = rpc_system_err;
840                 }
841         } else {
842                 dprintk("svc: calling dispatcher\n");
843                 if (!versp->vs_dispatch(rqstp, statp)) {
844                         /* Release reply info */
845                         if (procp->pc_release)
846                                 procp->pc_release(rqstp, NULL, rqstp->rq_resp);
847                         goto dropit;
848                 }
849         }
850
851         /* Check RPC status result */
852         if (*statp != rpc_success)
853                 resv->iov_len = ((void*)statp)  - resv->iov_base + 4;
854
855         /* Release reply info */
856         if (procp->pc_release)
857                 procp->pc_release(rqstp, NULL, rqstp->rq_resp);
858
859         if (procp->pc_encode == NULL)
860                 goto dropit;
861
862  sendit:
863         if (svc_authorise(rqstp))
864                 goto dropit;
865         return svc_send(rqstp);
866
867  dropit:
868         svc_authorise(rqstp);   /* doesn't hurt to call this twice */
869         dprintk("svc: svc_process dropit\n");
870         svc_drop(rqstp);
871         return 0;
872
873 err_short_len:
874         if (net_ratelimit())
875                 printk("svc: short len %Zd, dropping request\n", argv->iov_len);
876
877         goto dropit;                    /* drop request */
878
879 err_bad_dir:
880         if (net_ratelimit())
881                 printk("svc: bad direction %d, dropping request\n", dir);
882
883         serv->sv_stats->rpcbadfmt++;
884         goto dropit;                    /* drop request */
885
886 err_bad_rpc:
887         serv->sv_stats->rpcbadfmt++;
888         svc_putnl(resv, 1);     /* REJECT */
889         svc_putnl(resv, 0);     /* RPC_MISMATCH */
890         svc_putnl(resv, 2);     /* Only RPCv2 supported */
891         svc_putnl(resv, 2);
892         goto sendit;
893
894 err_bad_auth:
895         dprintk("svc: authentication failed (%d)\n", ntohl(auth_stat));
896         serv->sv_stats->rpcbadauth++;
897         /* Restore write pointer to location of accept status: */
898         xdr_ressize_check(rqstp, reply_statp);
899         svc_putnl(resv, 1);     /* REJECT */
900         svc_putnl(resv, 1);     /* AUTH_ERROR */
901         svc_putnl(resv, ntohl(auth_stat));      /* status */
902         goto sendit;
903
904 err_bad_prog:
905         dprintk("svc: unknown program %d\n", prog);
906         serv->sv_stats->rpcbadfmt++;
907         svc_putnl(resv, RPC_PROG_UNAVAIL);
908         goto sendit;
909
910 err_bad_vers:
911         if (net_ratelimit())
912                 printk("svc: unknown version (%d for prog %d, %s)\n",
913                        vers, prog, progp->pg_name);
914
915         serv->sv_stats->rpcbadfmt++;
916         svc_putnl(resv, RPC_PROG_MISMATCH);
917         svc_putnl(resv, progp->pg_lovers);
918         svc_putnl(resv, progp->pg_hivers);
919         goto sendit;
920
921 err_bad_proc:
922         if (net_ratelimit())
923                 printk("svc: unknown procedure (%d)\n", proc);
924
925         serv->sv_stats->rpcbadfmt++;
926         svc_putnl(resv, RPC_PROC_UNAVAIL);
927         goto sendit;
928
929 err_garbage:
930         if (net_ratelimit())
931                 printk("svc: failed to decode args\n");
932
933         rpc_stat = rpc_garbage_args;
934 err_bad:
935         serv->sv_stats->rpcbadfmt++;
936         svc_putnl(resv, ntohl(rpc_stat));
937         goto sendit;
938 }
939
940 /*
941  * Return (transport-specific) limit on the rpc payload.
942  */
943 u32 svc_max_payload(const struct svc_rqst *rqstp)
944 {
945         int max = RPCSVC_MAXPAYLOAD_TCP;
946
947         if (rqstp->rq_sock->sk_sock->type == SOCK_DGRAM)
948                 max = RPCSVC_MAXPAYLOAD_UDP;
949         if (rqstp->rq_server->sv_max_payload < max)
950                 max = rqstp->rq_server->sv_max_payload;
951         return max;
952 }
953 EXPORT_SYMBOL_GPL(svc_max_payload);