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