1// SPDX-License-Identifier: GPL-2.0-only
2/*
3 * linux/net/sunrpc/svc.c
4 *
5 * High-level RPC service routines
6 *
7 * Copyright (C) 1995, 1996 Olaf Kirch <okir@monad.swb.de>
8 *
9 * Multiple threads pools and NUMAisation
10 * Copyright (c) 2006 Silicon Graphics, Inc.
11 * by Greg Banks <gnb@melbourne.sgi.com>
12 */
13
14#include <linux/linkage.h>
15#include <linux/sched/signal.h>
16#include <linux/errno.h>
17#include <linux/net.h>
18#include <linux/in.h>
19#include <linux/mm.h>
20#include <linux/interrupt.h>
21#include <linux/module.h>
22#include <linux/kthread.h>
23#include <linux/slab.h>
24
25#include <linux/sunrpc/types.h>
26#include <linux/sunrpc/xdr.h>
27#include <linux/sunrpc/stats.h>
28#include <linux/sunrpc/svcsock.h>
29#include <linux/sunrpc/clnt.h>
30#include <linux/sunrpc/bc_xprt.h>
31
32#include <trace/events/sunrpc.h>
33
34#include "fail.h"
35#include "sunrpc.h"
36
37#define RPCDBG_FACILITY RPCDBG_SVCDSP
38
39static void svc_unregister(const struct svc_serv *serv, struct net *net);
40
41#define SVC_POOL_DEFAULT SVC_POOL_GLOBAL
42
43/*
44 * Mode for mapping cpus to pools.
45 */
46enum {
47 SVC_POOL_AUTO = -1, /* choose one of the others */
48 SVC_POOL_GLOBAL, /* no mapping, just a single global pool
49 * (legacy & UP mode) */
50 SVC_POOL_PERCPU, /* one pool per cpu */
51 SVC_POOL_PERNODE /* one pool per numa node */
52};
53
54/*
55 * Structure for mapping cpus to pools and vice versa.
56 * Setup once during sunrpc initialisation.
57 */
58
59struct svc_pool_map {
60 int count; /* How many svc_servs use us */
61 int mode; /* Note: int not enum to avoid
62 * warnings about "enumeration value
63 * not handled in switch" */
64 unsigned int npools;
65 unsigned int *pool_to; /* maps pool id to cpu or node */
66 unsigned int *to_pool; /* maps cpu or node to pool id */
67};
68
69static struct svc_pool_map svc_pool_map = {
70 .mode = SVC_POOL_DEFAULT
71};
72
73static DEFINE_MUTEX(svc_pool_map_mutex);/* protects svc_pool_map.count only */
74
75static int
76__param_set_pool_mode(const char *val, struct svc_pool_map *m)
77{
78 int err, mode;
79
80 mutex_lock(lock: &svc_pool_map_mutex);
81
82 err = 0;
83 if (!strncmp(val, "auto", 4))
84 mode = SVC_POOL_AUTO;
85 else if (!strncmp(val, "global", 6))
86 mode = SVC_POOL_GLOBAL;
87 else if (!strncmp(val, "percpu", 6))
88 mode = SVC_POOL_PERCPU;
89 else if (!strncmp(val, "pernode", 7))
90 mode = SVC_POOL_PERNODE;
91 else
92 err = -EINVAL;
93
94 if (err)
95 goto out;
96
97 if (m->count == 0)
98 m->mode = mode;
99 else if (mode != m->mode)
100 err = -EBUSY;
101out:
102 mutex_unlock(lock: &svc_pool_map_mutex);
103 return err;
104}
105
106static int
107param_set_pool_mode(const char *val, const struct kernel_param *kp)
108{
109 struct svc_pool_map *m = kp->arg;
110
111 return __param_set_pool_mode(val, m);
112}
113
114int sunrpc_set_pool_mode(const char *val)
115{
116 return __param_set_pool_mode(val, m: &svc_pool_map);
117}
118EXPORT_SYMBOL(sunrpc_set_pool_mode);
119
120/**
121 * sunrpc_get_pool_mode - get the current pool_mode for the host
122 * @buf: where to write the current pool_mode
123 * @size: size of @buf
124 *
125 * Grab the current pool_mode from the svc_pool_map and write
126 * the resulting string to @buf. Returns the number of characters
127 * written to @buf (a'la snprintf()).
128 */
129int
130sunrpc_get_pool_mode(char *buf, size_t size)
131{
132 struct svc_pool_map *m = &svc_pool_map;
133
134 switch (m->mode)
135 {
136 case SVC_POOL_AUTO:
137 return snprintf(buf, size, fmt: "auto");
138 case SVC_POOL_GLOBAL:
139 return snprintf(buf, size, fmt: "global");
140 case SVC_POOL_PERCPU:
141 return snprintf(buf, size, fmt: "percpu");
142 case SVC_POOL_PERNODE:
143 return snprintf(buf, size, fmt: "pernode");
144 default:
145 return snprintf(buf, size, fmt: "%d", m->mode);
146 }
147}
148EXPORT_SYMBOL(sunrpc_get_pool_mode);
149
150static int
151param_get_pool_mode(char *buf, const struct kernel_param *kp)
152{
153 char str[16];
154 int len;
155
156 len = sunrpc_get_pool_mode(str, ARRAY_SIZE(str));
157
158 /* Ensure we have room for newline and NUL */
159 len = min_t(int, len, ARRAY_SIZE(str) - 2);
160
161 /* tack on the newline */
162 str[len] = '\n';
163 str[len + 1] = '\0';
164
165 return sysfs_emit(buf, fmt: "%s", str);
166}
167
168module_param_call(pool_mode, param_set_pool_mode, param_get_pool_mode,
169 &svc_pool_map, 0644);
170
171/*
172 * Detect best pool mapping mode heuristically,
173 * according to the machine's topology.
174 */
175static int
176svc_pool_map_choose_mode(void)
177{
178 unsigned int node;
179
180 if (nr_online_nodes > 1) {
181 /*
182 * Actually have multiple NUMA nodes,
183 * so split pools on NUMA node boundaries
184 */
185 return SVC_POOL_PERNODE;
186 }
187
188 node = first_online_node;
189 if (nr_cpus_node(node) > 2) {
190 /*
191 * Non-trivial SMP, or CONFIG_NUMA on
192 * non-NUMA hardware, e.g. with a generic
193 * x86_64 kernel on Xeons. In this case we
194 * want to divide the pools on cpu boundaries.
195 */
196 return SVC_POOL_PERCPU;
197 }
198
199 /* default: one global pool */
200 return SVC_POOL_GLOBAL;
201}
202
203/*
204 * Allocate the to_pool[] and pool_to[] arrays.
205 * Returns 0 on success or an errno.
206 */
207static int
208svc_pool_map_alloc_arrays(struct svc_pool_map *m, unsigned int maxpools)
209{
210 m->to_pool = kcalloc(maxpools, sizeof(unsigned int), GFP_KERNEL);
211 if (!m->to_pool)
212 goto fail;
213 m->pool_to = kcalloc(maxpools, sizeof(unsigned int), GFP_KERNEL);
214 if (!m->pool_to)
215 goto fail_free;
216
217 return 0;
218
219fail_free:
220 kfree(objp: m->to_pool);
221 m->to_pool = NULL;
222fail:
223 return -ENOMEM;
224}
225
226/*
227 * Initialise the pool map for SVC_POOL_PERCPU mode.
228 * Returns number of pools or <0 on error.
229 */
230static int
231svc_pool_map_init_percpu(struct svc_pool_map *m)
232{
233 unsigned int maxpools = nr_cpu_ids;
234 unsigned int pidx = 0;
235 unsigned int cpu;
236 int err;
237
238 err = svc_pool_map_alloc_arrays(m, maxpools);
239 if (err)
240 return err;
241
242 for_each_online_cpu(cpu) {
243 BUG_ON(pidx >= maxpools);
244 m->to_pool[cpu] = pidx;
245 m->pool_to[pidx] = cpu;
246 pidx++;
247 }
248 /* cpus brought online later all get mapped to pool0, sorry */
249
250 return pidx;
251};
252
253
254/*
255 * Initialise the pool map for SVC_POOL_PERNODE mode.
256 * Returns number of pools or <0 on error.
257 */
258static int
259svc_pool_map_init_pernode(struct svc_pool_map *m)
260{
261 unsigned int maxpools = nr_node_ids;
262 unsigned int pidx = 0;
263 unsigned int node;
264 int err;
265
266 err = svc_pool_map_alloc_arrays(m, maxpools);
267 if (err)
268 return err;
269
270 for_each_node_with_cpus(node) {
271 /* some architectures (e.g. SN2) have cpuless nodes */
272 BUG_ON(pidx > maxpools);
273 m->to_pool[node] = pidx;
274 m->pool_to[pidx] = node;
275 pidx++;
276 }
277 /* nodes brought online later all get mapped to pool0, sorry */
278
279 return pidx;
280}
281
282
283/*
284 * Add a reference to the global map of cpus to pools (and
285 * vice versa) if pools are in use.
286 * Initialise the map if we're the first user.
287 * Returns the number of pools. If this is '1', no reference
288 * was taken.
289 */
290static unsigned int
291svc_pool_map_get(void)
292{
293 struct svc_pool_map *m = &svc_pool_map;
294 int npools = -1;
295
296 mutex_lock(lock: &svc_pool_map_mutex);
297 if (m->count++) {
298 mutex_unlock(lock: &svc_pool_map_mutex);
299 return m->npools;
300 }
301
302 if (m->mode == SVC_POOL_AUTO)
303 m->mode = svc_pool_map_choose_mode();
304
305 switch (m->mode) {
306 case SVC_POOL_PERCPU:
307 npools = svc_pool_map_init_percpu(m);
308 break;
309 case SVC_POOL_PERNODE:
310 npools = svc_pool_map_init_pernode(m);
311 break;
312 }
313
314 if (npools <= 0) {
315 /* default, or memory allocation failure */
316 npools = 1;
317 m->mode = SVC_POOL_GLOBAL;
318 }
319 m->npools = npools;
320 mutex_unlock(lock: &svc_pool_map_mutex);
321 return npools;
322}
323
324/*
325 * Drop a reference to the global map of cpus to pools.
326 * When the last reference is dropped, the map data is
327 * freed; this allows the sysadmin to change the pool.
328 */
329static void
330svc_pool_map_put(void)
331{
332 struct svc_pool_map *m = &svc_pool_map;
333
334 mutex_lock(lock: &svc_pool_map_mutex);
335 if (!--m->count) {
336 kfree(objp: m->to_pool);
337 m->to_pool = NULL;
338 kfree(objp: m->pool_to);
339 m->pool_to = NULL;
340 m->npools = 0;
341 }
342 mutex_unlock(lock: &svc_pool_map_mutex);
343}
344
345static int svc_pool_map_get_node(unsigned int pidx)
346{
347 const struct svc_pool_map *m = &svc_pool_map;
348
349 if (m->count) {
350 if (m->mode == SVC_POOL_PERCPU)
351 return cpu_to_node(cpu: m->pool_to[pidx]);
352 if (m->mode == SVC_POOL_PERNODE)
353 return m->pool_to[pidx];
354 }
355 return numa_mem_id();
356}
357/*
358 * Set the given thread's cpus_allowed mask so that it
359 * will only run on cpus in the given pool.
360 */
361static inline void
362svc_pool_map_set_cpumask(struct task_struct *task, unsigned int pidx)
363{
364 struct svc_pool_map *m = &svc_pool_map;
365 unsigned int node = m->pool_to[pidx];
366
367 /*
368 * The caller checks for sv_nrpools > 1, which
369 * implies that we've been initialized.
370 */
371 WARN_ON_ONCE(m->count == 0);
372 if (m->count == 0)
373 return;
374
375 switch (m->mode) {
376 case SVC_POOL_PERCPU:
377 {
378 set_cpus_allowed_ptr(p: task, cpumask_of(node));
379 break;
380 }
381 case SVC_POOL_PERNODE:
382 {
383 set_cpus_allowed_ptr(p: task, new_mask: cpumask_of_node(node));
384 break;
385 }
386 }
387}
388
389/**
390 * svc_pool_for_cpu - Select pool to run a thread on this cpu
391 * @serv: An RPC service
392 *
393 * Use the active CPU and the svc_pool_map's mode setting to
394 * select the svc thread pool to use. Once initialized, the
395 * svc_pool_map does not change.
396 *
397 * Return value:
398 * A pointer to an svc_pool
399 */
400struct svc_pool *svc_pool_for_cpu(struct svc_serv *serv)
401{
402 struct svc_pool_map *m = &svc_pool_map;
403 int cpu = raw_smp_processor_id();
404 unsigned int pidx = 0;
405
406 if (serv->sv_nrpools <= 1)
407 return serv->sv_pools;
408
409 switch (m->mode) {
410 case SVC_POOL_PERCPU:
411 pidx = m->to_pool[cpu];
412 break;
413 case SVC_POOL_PERNODE:
414 pidx = m->to_pool[cpu_to_node(cpu)];
415 break;
416 }
417
418 return &serv->sv_pools[pidx % serv->sv_nrpools];
419}
420
421static int svc_rpcb_setup(struct svc_serv *serv, struct net *net)
422{
423 int err;
424
425 err = rpcb_create_local(net);
426 if (err)
427 return err;
428
429 /* Remove any stale portmap registrations */
430 svc_unregister(serv, net);
431 return 0;
432}
433
434void svc_rpcb_cleanup(struct svc_serv *serv, struct net *net)
435{
436 svc_unregister(serv, net);
437 rpcb_put_local(net);
438}
439
440static int svc_uses_rpcbind(struct svc_serv *serv)
441{
442 unsigned int p, i;
443
444 for (p = 0; p < serv->sv_nprogs; p++) {
445 struct svc_program *progp = &serv->sv_programs[p];
446
447 for (i = 0; i < progp->pg_nvers; i++) {
448 if (progp->pg_vers[i] == NULL)
449 continue;
450 if (!progp->pg_vers[i]->vs_hidden)
451 return 1;
452 }
453 }
454
455 return 0;
456}
457
458int svc_bind(struct svc_serv *serv, struct net *net)
459{
460 if (!svc_uses_rpcbind(serv))
461 return 0;
462 return svc_rpcb_setup(serv, net);
463}
464EXPORT_SYMBOL_GPL(svc_bind);
465
466#if defined(CONFIG_SUNRPC_BACKCHANNEL)
467static void
468__svc_init_bc(struct svc_serv *serv)
469{
470 lwq_init(&serv->sv_cb_list);
471}
472#else
473static void
474__svc_init_bc(struct svc_serv *serv)
475{
476}
477#endif
478
479/*
480 * Create an RPC service
481 */
482static struct svc_serv *
483__svc_create(struct svc_program *prog, int nprogs, struct svc_stat *stats,
484 unsigned int bufsize, int npools, int (*threadfn)(void *data))
485{
486 struct svc_serv *serv;
487 unsigned int vers;
488 unsigned int xdrsize;
489 unsigned int i;
490
491 if (!(serv = kzalloc(sizeof(*serv), GFP_KERNEL)))
492 return NULL;
493 serv->sv_name = prog->pg_name;
494 serv->sv_programs = prog;
495 serv->sv_nprogs = nprogs;
496 serv->sv_stats = stats;
497 if (bufsize > RPCSVC_MAXPAYLOAD)
498 bufsize = RPCSVC_MAXPAYLOAD;
499 serv->sv_max_payload = bufsize? bufsize : 4096;
500 serv->sv_max_mesg = roundup(serv->sv_max_payload + PAGE_SIZE, PAGE_SIZE);
501 serv->sv_threadfn = threadfn;
502 xdrsize = 0;
503 for (i = 0; i < nprogs; i++) {
504 struct svc_program *progp = &prog[i];
505
506 progp->pg_lovers = progp->pg_nvers-1;
507 for (vers = 0; vers < progp->pg_nvers ; vers++)
508 if (progp->pg_vers[vers]) {
509 progp->pg_hivers = vers;
510 if (progp->pg_lovers > vers)
511 progp->pg_lovers = vers;
512 if (progp->pg_vers[vers]->vs_xdrsize > xdrsize)
513 xdrsize = progp->pg_vers[vers]->vs_xdrsize;
514 }
515 }
516 serv->sv_xdrsize = xdrsize;
517 INIT_LIST_HEAD(list: &serv->sv_tempsocks);
518 INIT_LIST_HEAD(list: &serv->sv_permsocks);
519 timer_setup(&serv->sv_temptimer, NULL, 0);
520 spin_lock_init(&serv->sv_lock);
521
522 __svc_init_bc(serv);
523
524 serv->sv_nrpools = npools;
525 serv->sv_pools =
526 kcalloc(serv->sv_nrpools, sizeof(struct svc_pool),
527 GFP_KERNEL);
528 if (!serv->sv_pools) {
529 kfree(objp: serv);
530 return NULL;
531 }
532
533 for (i = 0; i < serv->sv_nrpools; i++) {
534 struct svc_pool *pool = &serv->sv_pools[i];
535
536 dprintk("svc: initialising pool %u for %s\n",
537 i, serv->sv_name);
538
539 pool->sp_id = i;
540 lwq_init(q: &pool->sp_xprts);
541 INIT_LIST_HEAD(list: &pool->sp_all_threads);
542 init_llist_head(list: &pool->sp_idle_threads);
543
544 percpu_counter_init(&pool->sp_messages_arrived, 0, GFP_KERNEL);
545 percpu_counter_init(&pool->sp_sockets_queued, 0, GFP_KERNEL);
546 percpu_counter_init(&pool->sp_threads_woken, 0, GFP_KERNEL);
547 }
548
549 return serv;
550}
551
552/**
553 * svc_create - Create an RPC service
554 * @prog: the RPC program the new service will handle
555 * @bufsize: maximum message size for @prog
556 * @threadfn: a function to service RPC requests for @prog
557 *
558 * Returns an instantiated struct svc_serv object or NULL.
559 */
560struct svc_serv *svc_create(struct svc_program *prog, unsigned int bufsize,
561 int (*threadfn)(void *data))
562{
563 return __svc_create(prog, nprogs: 1, NULL, bufsize, npools: 1, threadfn);
564}
565EXPORT_SYMBOL_GPL(svc_create);
566
567/**
568 * svc_create_pooled - Create an RPC service with pooled threads
569 * @prog: Array of RPC programs the new service will handle
570 * @nprogs: Number of programs in the array
571 * @stats: the stats struct if desired
572 * @bufsize: maximum message size for @prog
573 * @threadfn: a function to service RPC requests for @prog
574 *
575 * Returns an instantiated struct svc_serv object or NULL.
576 */
577struct svc_serv *svc_create_pooled(struct svc_program *prog,
578 unsigned int nprogs,
579 struct svc_stat *stats,
580 unsigned int bufsize,
581 int (*threadfn)(void *data))
582{
583 struct svc_serv *serv;
584 unsigned int npools = svc_pool_map_get();
585
586 serv = __svc_create(prog, nprogs, stats, bufsize, npools, threadfn);
587 if (!serv)
588 goto out_err;
589 serv->sv_is_pooled = true;
590 return serv;
591out_err:
592 svc_pool_map_put();
593 return NULL;
594}
595EXPORT_SYMBOL_GPL(svc_create_pooled);
596
597/*
598 * Destroy an RPC service. Should be called with appropriate locking to
599 * protect sv_permsocks and sv_tempsocks.
600 */
601void
602svc_destroy(struct svc_serv **servp)
603{
604 struct svc_serv *serv = *servp;
605 unsigned int i;
606
607 *servp = NULL;
608
609 dprintk("svc: svc_destroy(%s)\n", serv->sv_programs->pg_name);
610 timer_shutdown_sync(timer: &serv->sv_temptimer);
611
612 /*
613 * Remaining transports at this point are not expected.
614 */
615 WARN_ONCE(!list_empty(&serv->sv_permsocks),
616 "SVC: permsocks remain for %s\n", serv->sv_programs->pg_name);
617 WARN_ONCE(!list_empty(&serv->sv_tempsocks),
618 "SVC: tempsocks remain for %s\n", serv->sv_programs->pg_name);
619
620 cache_clean_deferred(owner: serv);
621
622 if (serv->sv_is_pooled)
623 svc_pool_map_put();
624
625 for (i = 0; i < serv->sv_nrpools; i++) {
626 struct svc_pool *pool = &serv->sv_pools[i];
627
628 percpu_counter_destroy(fbc: &pool->sp_messages_arrived);
629 percpu_counter_destroy(fbc: &pool->sp_sockets_queued);
630 percpu_counter_destroy(fbc: &pool->sp_threads_woken);
631 }
632 kfree(objp: serv->sv_pools);
633 kfree(objp: serv);
634}
635EXPORT_SYMBOL_GPL(svc_destroy);
636
637static bool
638svc_init_buffer(struct svc_rqst *rqstp, const struct svc_serv *serv, int node)
639{
640 rqstp->rq_maxpages = svc_serv_maxpages(serv);
641
642 /* rq_pages' last entry is NULL for historical reasons. */
643 rqstp->rq_pages = kcalloc_node(rqstp->rq_maxpages + 1,
644 sizeof(struct page *),
645 GFP_KERNEL, node);
646 if (!rqstp->rq_pages)
647 return false;
648
649 return true;
650}
651
652/*
653 * Release an RPC server buffer
654 */
655static void
656svc_release_buffer(struct svc_rqst *rqstp)
657{
658 unsigned long i;
659
660 for (i = 0; i < rqstp->rq_maxpages; i++)
661 if (rqstp->rq_pages[i])
662 put_page(page: rqstp->rq_pages[i]);
663 kfree(objp: rqstp->rq_pages);
664}
665
666static void
667svc_rqst_free(struct svc_rqst *rqstp)
668{
669 folio_batch_release(fbatch: &rqstp->rq_fbatch);
670 kfree(objp: rqstp->rq_bvec);
671 svc_release_buffer(rqstp);
672 if (rqstp->rq_scratch_folio)
673 folio_put(folio: rqstp->rq_scratch_folio);
674 kfree(objp: rqstp->rq_resp);
675 kfree(objp: rqstp->rq_argp);
676 kfree(objp: rqstp->rq_auth_data);
677 kfree_rcu(rqstp, rq_rcu_head);
678}
679
680static struct svc_rqst *
681svc_prepare_thread(struct svc_serv *serv, struct svc_pool *pool, int node)
682{
683 struct svc_rqst *rqstp;
684
685 rqstp = kzalloc_node(sizeof(*rqstp), GFP_KERNEL, node);
686 if (!rqstp)
687 return rqstp;
688
689 folio_batch_init(fbatch: &rqstp->rq_fbatch);
690
691 rqstp->rq_server = serv;
692 rqstp->rq_pool = pool;
693
694 rqstp->rq_scratch_folio = __folio_alloc_node(GFP_KERNEL, 0, node);
695 if (!rqstp->rq_scratch_folio)
696 goto out_enomem;
697
698 rqstp->rq_argp = kmalloc_node(serv->sv_xdrsize, GFP_KERNEL, node);
699 if (!rqstp->rq_argp)
700 goto out_enomem;
701
702 rqstp->rq_resp = kmalloc_node(serv->sv_xdrsize, GFP_KERNEL, node);
703 if (!rqstp->rq_resp)
704 goto out_enomem;
705
706 if (!svc_init_buffer(rqstp, serv, node))
707 goto out_enomem;
708
709 rqstp->rq_bvec = kcalloc_node(rqstp->rq_maxpages,
710 sizeof(struct bio_vec),
711 GFP_KERNEL, node);
712 if (!rqstp->rq_bvec)
713 goto out_enomem;
714
715 rqstp->rq_err = -EAGAIN; /* No error yet */
716
717 serv->sv_nrthreads += 1;
718 pool->sp_nrthreads += 1;
719
720 /* Protected by whatever lock the service uses when calling
721 * svc_set_num_threads()
722 */
723 list_add_rcu(new: &rqstp->rq_all, head: &pool->sp_all_threads);
724
725 return rqstp;
726
727out_enomem:
728 svc_rqst_free(rqstp);
729 return NULL;
730}
731
732/**
733 * svc_pool_wake_idle_thread - Awaken an idle thread in @pool
734 * @pool: service thread pool
735 *
736 * Can be called from soft IRQ or process context. Finding an idle
737 * service thread and marking it BUSY is atomic with respect to
738 * other calls to svc_pool_wake_idle_thread().
739 *
740 */
741void svc_pool_wake_idle_thread(struct svc_pool *pool)
742{
743 struct svc_rqst *rqstp;
744 struct llist_node *ln;
745
746 rcu_read_lock();
747 ln = READ_ONCE(pool->sp_idle_threads.first);
748 if (ln) {
749 rqstp = llist_entry(ln, struct svc_rqst, rq_idle);
750 WRITE_ONCE(rqstp->rq_qtime, ktime_get());
751 if (!task_is_running(rqstp->rq_task)) {
752 wake_up_process(tsk: rqstp->rq_task);
753 trace_svc_pool_thread_wake(pool, pid: rqstp->rq_task->pid);
754 percpu_counter_inc(fbc: &pool->sp_threads_woken);
755 } else {
756 trace_svc_pool_thread_running(pool, pid: rqstp->rq_task->pid);
757 }
758 rcu_read_unlock();
759 return;
760 }
761 rcu_read_unlock();
762 trace_svc_pool_thread_noidle(pool, pid: 0);
763}
764EXPORT_SYMBOL_GPL(svc_pool_wake_idle_thread);
765
766static struct svc_pool *
767svc_pool_next(struct svc_serv *serv, struct svc_pool *pool, unsigned int *state)
768{
769 return pool ? pool : &serv->sv_pools[(*state)++ % serv->sv_nrpools];
770}
771
772static struct svc_pool *
773svc_pool_victim(struct svc_serv *serv, struct svc_pool *target_pool,
774 unsigned int *state)
775{
776 struct svc_pool *pool;
777 unsigned int i;
778
779 pool = target_pool;
780
781 if (!pool) {
782 for (i = 0; i < serv->sv_nrpools; i++) {
783 pool = &serv->sv_pools[--(*state) % serv->sv_nrpools];
784 if (pool->sp_nrthreads)
785 break;
786 }
787 }
788
789 if (pool && pool->sp_nrthreads) {
790 set_bit(nr: SP_VICTIM_REMAINS, addr: &pool->sp_flags);
791 set_bit(nr: SP_NEED_VICTIM, addr: &pool->sp_flags);
792 return pool;
793 }
794 return NULL;
795}
796
797static int
798svc_start_kthreads(struct svc_serv *serv, struct svc_pool *pool, int nrservs)
799{
800 struct svc_rqst *rqstp;
801 struct task_struct *task;
802 struct svc_pool *chosen_pool;
803 unsigned int state = serv->sv_nrthreads-1;
804 int node;
805 int err;
806
807 do {
808 nrservs--;
809 chosen_pool = svc_pool_next(serv, pool, state: &state);
810 node = svc_pool_map_get_node(pidx: chosen_pool->sp_id);
811
812 rqstp = svc_prepare_thread(serv, pool: chosen_pool, node);
813 if (!rqstp)
814 return -ENOMEM;
815 task = kthread_create_on_node(threadfn: serv->sv_threadfn, data: rqstp,
816 node, namefmt: "%s", serv->sv_name);
817 if (IS_ERR(ptr: task)) {
818 svc_exit_thread(rqstp);
819 return PTR_ERR(ptr: task);
820 }
821
822 rqstp->rq_task = task;
823 if (serv->sv_nrpools > 1)
824 svc_pool_map_set_cpumask(task, pidx: chosen_pool->sp_id);
825
826 svc_sock_update_bufs(serv);
827 wake_up_process(tsk: task);
828
829 wait_var_event(&rqstp->rq_err, rqstp->rq_err != -EAGAIN);
830 err = rqstp->rq_err;
831 if (err) {
832 svc_exit_thread(rqstp);
833 return err;
834 }
835 } while (nrservs > 0);
836
837 return 0;
838}
839
840static int
841svc_stop_kthreads(struct svc_serv *serv, struct svc_pool *pool, int nrservs)
842{
843 unsigned int state = serv->sv_nrthreads-1;
844 struct svc_pool *victim;
845
846 do {
847 victim = svc_pool_victim(serv, target_pool: pool, state: &state);
848 if (!victim)
849 break;
850 svc_pool_wake_idle_thread(victim);
851 wait_on_bit(word: &victim->sp_flags, bit: SP_VICTIM_REMAINS,
852 TASK_IDLE);
853 nrservs++;
854 } while (nrservs < 0);
855 return 0;
856}
857
858/**
859 * svc_set_num_threads - adjust number of threads per RPC service
860 * @serv: RPC service to adjust
861 * @pool: Specific pool from which to choose threads, or NULL
862 * @nrservs: New number of threads for @serv (0 or less means kill all threads)
863 *
864 * Create or destroy threads to make the number of threads for @serv the
865 * given number. If @pool is non-NULL, change only threads in that pool;
866 * otherwise, round-robin between all pools for @serv. @serv's
867 * sv_nrthreads is adjusted for each thread created or destroyed.
868 *
869 * Caller must ensure mutual exclusion between this and server startup or
870 * shutdown.
871 *
872 * Returns zero on success or a negative errno if an error occurred while
873 * starting a thread.
874 */
875int
876svc_set_num_threads(struct svc_serv *serv, struct svc_pool *pool, int nrservs)
877{
878 if (!pool)
879 nrservs -= serv->sv_nrthreads;
880 else
881 nrservs -= pool->sp_nrthreads;
882
883 if (nrservs > 0)
884 return svc_start_kthreads(serv, pool, nrservs);
885 if (nrservs < 0)
886 return svc_stop_kthreads(serv, pool, nrservs);
887 return 0;
888}
889EXPORT_SYMBOL_GPL(svc_set_num_threads);
890
891/**
892 * svc_rqst_replace_page - Replace one page in rq_pages[]
893 * @rqstp: svc_rqst with pages to replace
894 * @page: replacement page
895 *
896 * When replacing a page in rq_pages, batch the release of the
897 * replaced pages to avoid hammering the page allocator.
898 *
899 * Return values:
900 * %true: page replaced
901 * %false: array bounds checking failed
902 */
903bool svc_rqst_replace_page(struct svc_rqst *rqstp, struct page *page)
904{
905 struct page **begin = rqstp->rq_pages;
906 struct page **end = &rqstp->rq_pages[rqstp->rq_maxpages];
907
908 if (unlikely(rqstp->rq_next_page < begin || rqstp->rq_next_page > end)) {
909 trace_svc_replace_page_err(rqst: rqstp);
910 return false;
911 }
912
913 if (*rqstp->rq_next_page) {
914 if (!folio_batch_add(fbatch: &rqstp->rq_fbatch,
915 page_folio(*rqstp->rq_next_page)))
916 __folio_batch_release(pvec: &rqstp->rq_fbatch);
917 }
918
919 get_page(page);
920 *(rqstp->rq_next_page++) = page;
921 return true;
922}
923EXPORT_SYMBOL_GPL(svc_rqst_replace_page);
924
925/**
926 * svc_rqst_release_pages - Release Reply buffer pages
927 * @rqstp: RPC transaction context
928 *
929 * Release response pages that might still be in flight after
930 * svc_send, and any spliced filesystem-owned pages.
931 */
932void svc_rqst_release_pages(struct svc_rqst *rqstp)
933{
934 int i, count = rqstp->rq_next_page - rqstp->rq_respages;
935
936 if (count) {
937 release_pages(rqstp->rq_respages, nr: count);
938 for (i = 0; i < count; i++)
939 rqstp->rq_respages[i] = NULL;
940 }
941}
942
943/**
944 * svc_exit_thread - finalise the termination of a sunrpc server thread
945 * @rqstp: the svc_rqst which represents the thread.
946 *
947 * When a thread started with svc_new_thread() exits it must call
948 * svc_exit_thread() as its last act. This must be done with the
949 * service mutex held. Normally this is held by a DIFFERENT thread, the
950 * one that is calling svc_set_num_threads() and which will wait for
951 * SP_VICTIM_REMAINS to be cleared before dropping the mutex. If the
952 * thread exits for any reason other than svc_thread_should_stop()
953 * returning %true (which indicated that svc_set_num_threads() is
954 * waiting for it to exit), then it must take the service mutex itself,
955 * which can only safely be done using mutex_try_lock().
956 */
957void
958svc_exit_thread(struct svc_rqst *rqstp)
959{
960 struct svc_serv *serv = rqstp->rq_server;
961 struct svc_pool *pool = rqstp->rq_pool;
962
963 list_del_rcu(entry: &rqstp->rq_all);
964
965 pool->sp_nrthreads -= 1;
966 serv->sv_nrthreads -= 1;
967 svc_sock_update_bufs(serv);
968
969 svc_rqst_free(rqstp);
970
971 clear_and_wake_up_bit(bit: SP_VICTIM_REMAINS, word: &pool->sp_flags);
972}
973EXPORT_SYMBOL_GPL(svc_exit_thread);
974
975/*
976 * Register an "inet" protocol family netid with the local
977 * rpcbind daemon via an rpcbind v4 SET request.
978 *
979 * No netconfig infrastructure is available in the kernel, so
980 * we map IP_ protocol numbers to netids by hand.
981 *
982 * Returns zero on success; a negative errno value is returned
983 * if any error occurs.
984 */
985static int __svc_rpcb_register4(struct net *net, const u32 program,
986 const u32 version,
987 const unsigned short protocol,
988 const unsigned short port)
989{
990 const struct sockaddr_in sin = {
991 .sin_family = AF_INET,
992 .sin_addr.s_addr = htonl(INADDR_ANY),
993 .sin_port = htons(port),
994 };
995 const char *netid;
996 int error;
997
998 switch (protocol) {
999 case IPPROTO_UDP:
1000 netid = RPCBIND_NETID_UDP;
1001 break;
1002 case IPPROTO_TCP:
1003 netid = RPCBIND_NETID_TCP;
1004 break;
1005 default:
1006 return -ENOPROTOOPT;
1007 }
1008
1009 error = rpcb_v4_register(net, program, version,
1010 address: (const struct sockaddr *)&sin, netid);
1011
1012 /*
1013 * User space didn't support rpcbind v4, so retry this
1014 * registration request with the legacy rpcbind v2 protocol.
1015 */
1016 if (error == -EPROTONOSUPPORT)
1017 error = rpcb_register(net, program, version, protocol, port);
1018
1019 return error;
1020}
1021
1022#if IS_ENABLED(CONFIG_IPV6)
1023/*
1024 * Register an "inet6" protocol family netid with the local
1025 * rpcbind daemon via an rpcbind v4 SET request.
1026 *
1027 * No netconfig infrastructure is available in the kernel, so
1028 * we map IP_ protocol numbers to netids by hand.
1029 *
1030 * Returns zero on success; a negative errno value is returned
1031 * if any error occurs.
1032 */
1033static int __svc_rpcb_register6(struct net *net, const u32 program,
1034 const u32 version,
1035 const unsigned short protocol,
1036 const unsigned short port)
1037{
1038 const struct sockaddr_in6 sin6 = {
1039 .sin6_family = AF_INET6,
1040 .sin6_addr = IN6ADDR_ANY_INIT,
1041 .sin6_port = htons(port),
1042 };
1043 const char *netid;
1044 int error;
1045
1046 switch (protocol) {
1047 case IPPROTO_UDP:
1048 netid = RPCBIND_NETID_UDP6;
1049 break;
1050 case IPPROTO_TCP:
1051 netid = RPCBIND_NETID_TCP6;
1052 break;
1053 default:
1054 return -ENOPROTOOPT;
1055 }
1056
1057 error = rpcb_v4_register(net, program, version,
1058 address: (const struct sockaddr *)&sin6, netid);
1059
1060 /*
1061 * User space didn't support rpcbind version 4, so we won't
1062 * use a PF_INET6 listener.
1063 */
1064 if (error == -EPROTONOSUPPORT)
1065 error = -EAFNOSUPPORT;
1066
1067 return error;
1068}
1069#endif /* IS_ENABLED(CONFIG_IPV6) */
1070
1071/*
1072 * Register a kernel RPC service via rpcbind version 4.
1073 *
1074 * Returns zero on success; a negative errno value is returned
1075 * if any error occurs.
1076 */
1077static int __svc_register(struct net *net, const char *progname,
1078 const u32 program, const u32 version,
1079 const int family,
1080 const unsigned short protocol,
1081 const unsigned short port)
1082{
1083 int error = -EAFNOSUPPORT;
1084
1085 switch (family) {
1086 case PF_INET:
1087 error = __svc_rpcb_register4(net, program, version,
1088 protocol, port);
1089 break;
1090#if IS_ENABLED(CONFIG_IPV6)
1091 case PF_INET6:
1092 error = __svc_rpcb_register6(net, program, version,
1093 protocol, port);
1094#endif
1095 }
1096
1097 trace_svc_register(program: progname, version, family, protocol, port, error);
1098 return error;
1099}
1100
1101static
1102int svc_rpcbind_set_version(struct net *net,
1103 const struct svc_program *progp,
1104 u32 version, int family,
1105 unsigned short proto,
1106 unsigned short port)
1107{
1108 return __svc_register(net, progname: progp->pg_name, program: progp->pg_prog,
1109 version, family, protocol: proto, port);
1110
1111}
1112
1113int svc_generic_rpcbind_set(struct net *net,
1114 const struct svc_program *progp,
1115 u32 version, int family,
1116 unsigned short proto,
1117 unsigned short port)
1118{
1119 const struct svc_version *vers = progp->pg_vers[version];
1120 int error;
1121
1122 if (vers == NULL)
1123 return 0;
1124
1125 if (vers->vs_hidden) {
1126 trace_svc_noregister(program: progp->pg_name, version, family: proto,
1127 protocol: port, port: family, error: 0);
1128 return 0;
1129 }
1130
1131 /*
1132 * Don't register a UDP port if we need congestion
1133 * control.
1134 */
1135 if (vers->vs_need_cong_ctrl && proto == IPPROTO_UDP)
1136 return 0;
1137
1138 error = svc_rpcbind_set_version(net, progp, version,
1139 family, proto, port);
1140
1141 return (vers->vs_rpcb_optnl) ? 0 : error;
1142}
1143EXPORT_SYMBOL_GPL(svc_generic_rpcbind_set);
1144
1145/**
1146 * svc_register - register an RPC service with the local portmapper
1147 * @serv: svc_serv struct for the service to register
1148 * @net: net namespace for the service to register
1149 * @family: protocol family of service's listener socket
1150 * @proto: transport protocol number to advertise
1151 * @port: port to advertise
1152 *
1153 * Service is registered for any address in the passed-in protocol family
1154 */
1155int svc_register(const struct svc_serv *serv, struct net *net,
1156 const int family, const unsigned short proto,
1157 const unsigned short port)
1158{
1159 unsigned int p, i;
1160 int error = 0;
1161
1162 WARN_ON_ONCE(proto == 0 && port == 0);
1163 if (proto == 0 && port == 0)
1164 return -EINVAL;
1165
1166 for (p = 0; p < serv->sv_nprogs; p++) {
1167 struct svc_program *progp = &serv->sv_programs[p];
1168
1169 for (i = 0; i < progp->pg_nvers; i++) {
1170
1171 error = progp->pg_rpcbind_set(net, progp, i,
1172 family, proto, port);
1173 if (error < 0) {
1174 printk(KERN_WARNING "svc: failed to register "
1175 "%sv%u RPC service (errno %d).\n",
1176 progp->pg_name, i, -error);
1177 break;
1178 }
1179 }
1180 }
1181
1182 return error;
1183}
1184
1185/*
1186 * If user space is running rpcbind, it should take the v4 UNSET
1187 * and clear everything for this [program, version]. If user space
1188 * is running portmap, it will reject the v4 UNSET, but won't have
1189 * any "inet6" entries anyway. So a PMAP_UNSET should be sufficient
1190 * in this case to clear all existing entries for [program, version].
1191 */
1192static void __svc_unregister(struct net *net, const u32 program, const u32 version,
1193 const char *progname)
1194{
1195 int error;
1196
1197 error = rpcb_v4_register(net, program, version, NULL, netid: "");
1198
1199 /*
1200 * User space didn't support rpcbind v4, so retry this
1201 * request with the legacy rpcbind v2 protocol.
1202 */
1203 if (error == -EPROTONOSUPPORT)
1204 error = rpcb_register(net, program, version, 0, 0);
1205
1206 trace_svc_unregister(program: progname, version, error);
1207}
1208
1209/*
1210 * All netids, bind addresses and ports registered for [program, version]
1211 * are removed from the local rpcbind database (if the service is not
1212 * hidden) to make way for a new instance of the service.
1213 *
1214 * The result of unregistration is reported via dprintk for those who want
1215 * verification of the result, but is otherwise not important.
1216 */
1217static void svc_unregister(const struct svc_serv *serv, struct net *net)
1218{
1219 struct sighand_struct *sighand;
1220 unsigned long flags;
1221 unsigned int p, i;
1222
1223 clear_thread_flag(TIF_SIGPENDING);
1224
1225 for (p = 0; p < serv->sv_nprogs; p++) {
1226 struct svc_program *progp = &serv->sv_programs[p];
1227
1228 for (i = 0; i < progp->pg_nvers; i++) {
1229 if (progp->pg_vers[i] == NULL)
1230 continue;
1231 if (progp->pg_vers[i]->vs_hidden)
1232 continue;
1233 __svc_unregister(net, program: progp->pg_prog, version: i, progname: progp->pg_name);
1234 }
1235 }
1236
1237 rcu_read_lock();
1238 sighand = rcu_dereference(current->sighand);
1239 spin_lock_irqsave(&sighand->siglock, flags);
1240 recalc_sigpending();
1241 spin_unlock_irqrestore(lock: &sighand->siglock, flags);
1242 rcu_read_unlock();
1243}
1244
1245/*
1246 * dprintk the given error with the address of the client that caused it.
1247 */
1248#if IS_ENABLED(CONFIG_SUNRPC_DEBUG)
1249static __printf(2, 3)
1250void svc_printk(struct svc_rqst *rqstp, const char *fmt, ...)
1251{
1252 struct va_format vaf;
1253 va_list args;
1254 char buf[RPC_MAX_ADDRBUFLEN];
1255
1256 va_start(args, fmt);
1257
1258 vaf.fmt = fmt;
1259 vaf.va = &args;
1260
1261 dprintk("svc: %s: %pV", svc_print_addr(rqstp, buf, sizeof(buf)), &vaf);
1262
1263 va_end(args);
1264}
1265#else
1266static __printf(2,3) void svc_printk(struct svc_rqst *rqstp, const char *fmt, ...) {}
1267#endif
1268
1269__be32
1270svc_generic_init_request(struct svc_rqst *rqstp,
1271 const struct svc_program *progp,
1272 struct svc_process_info *ret)
1273{
1274 const struct svc_version *versp = NULL; /* compiler food */
1275 const struct svc_procedure *procp = NULL;
1276
1277 if (rqstp->rq_vers >= progp->pg_nvers )
1278 goto err_bad_vers;
1279 versp = progp->pg_vers[rqstp->rq_vers];
1280 if (!versp)
1281 goto err_bad_vers;
1282
1283 /*
1284 * Some protocol versions (namely NFSv4) require some form of
1285 * congestion control. (See RFC 7530 section 3.1 paragraph 2)
1286 * In other words, UDP is not allowed. We mark those when setting
1287 * up the svc_xprt, and verify that here.
1288 *
1289 * The spec is not very clear about what error should be returned
1290 * when someone tries to access a server that is listening on UDP
1291 * for lower versions. RPC_PROG_MISMATCH seems to be the closest
1292 * fit.
1293 */
1294 if (versp->vs_need_cong_ctrl && rqstp->rq_xprt &&
1295 !test_bit(XPT_CONG_CTRL, &rqstp->rq_xprt->xpt_flags))
1296 goto err_bad_vers;
1297
1298 if (rqstp->rq_proc >= versp->vs_nproc)
1299 goto err_bad_proc;
1300 rqstp->rq_procinfo = procp = &versp->vs_proc[rqstp->rq_proc];
1301
1302 /* Initialize storage for argp and resp */
1303 memset(s: rqstp->rq_argp, c: 0, n: procp->pc_argzero);
1304 memset(s: rqstp->rq_resp, c: 0, n: procp->pc_ressize);
1305
1306 /* Bump per-procedure stats counter */
1307 this_cpu_inc(versp->vs_count[rqstp->rq_proc]);
1308
1309 ret->dispatch = versp->vs_dispatch;
1310 return rpc_success;
1311err_bad_vers:
1312 ret->mismatch.lovers = progp->pg_lovers;
1313 ret->mismatch.hivers = progp->pg_hivers;
1314 return rpc_prog_mismatch;
1315err_bad_proc:
1316 return rpc_proc_unavail;
1317}
1318EXPORT_SYMBOL_GPL(svc_generic_init_request);
1319
1320/*
1321 * Common routine for processing the RPC request.
1322 */
1323static int
1324svc_process_common(struct svc_rqst *rqstp)
1325{
1326 struct xdr_stream *xdr = &rqstp->rq_res_stream;
1327 struct svc_program *progp = NULL;
1328 const struct svc_procedure *procp = NULL;
1329 struct svc_serv *serv = rqstp->rq_server;
1330 struct svc_process_info process;
1331 enum svc_auth_status auth_res;
1332 unsigned int aoffset;
1333 int pr, rc;
1334 __be32 *p;
1335
1336 /* Reset the accept_stat for the RPC */
1337 rqstp->rq_accept_statp = NULL;
1338
1339 /* Will be turned off only when NFSv4 Sessions are used */
1340 set_bit(nr: RQ_USEDEFERRAL, addr: &rqstp->rq_flags);
1341 clear_bit(nr: RQ_DROPME, addr: &rqstp->rq_flags);
1342
1343 /* Construct the first words of the reply: */
1344 svcxdr_init_encode(rqstp);
1345 xdr_stream_encode_be32(xdr, n: rqstp->rq_xid);
1346 xdr_stream_encode_be32(xdr, rpc_reply);
1347
1348 p = xdr_inline_decode(xdr: &rqstp->rq_arg_stream, XDR_UNIT * 4);
1349 if (unlikely(!p))
1350 goto err_short_len;
1351 if (*p++ != cpu_to_be32(RPC_VERSION))
1352 goto err_bad_rpc;
1353
1354 xdr_stream_encode_be32(xdr, rpc_msg_accepted);
1355
1356 rqstp->rq_prog = be32_to_cpup(p: p++);
1357 rqstp->rq_vers = be32_to_cpup(p: p++);
1358 rqstp->rq_proc = be32_to_cpup(p);
1359
1360 for (pr = 0; pr < serv->sv_nprogs; pr++)
1361 if (rqstp->rq_prog == serv->sv_programs[pr].pg_prog)
1362 progp = &serv->sv_programs[pr];
1363
1364 /*
1365 * Decode auth data, and add verifier to reply buffer.
1366 * We do this before anything else in order to get a decent
1367 * auth verifier.
1368 */
1369 auth_res = svc_authenticate(rqstp);
1370 /* Also give the program a chance to reject this call: */
1371 if (auth_res == SVC_OK && progp)
1372 auth_res = progp->pg_authenticate(rqstp);
1373 trace_svc_authenticate(rqst: rqstp, auth_res);
1374 switch (auth_res) {
1375 case SVC_OK:
1376 break;
1377 case SVC_GARBAGE:
1378 rqstp->rq_auth_stat = rpc_autherr_badcred;
1379 goto err_bad_auth;
1380 case SVC_DENIED:
1381 goto err_bad_auth;
1382 case SVC_CLOSE:
1383 goto close;
1384 case SVC_DROP:
1385 goto dropit;
1386 case SVC_COMPLETE:
1387 goto sendit;
1388 default:
1389 pr_warn_once("Unexpected svc_auth_status (%d)\n", auth_res);
1390 rqstp->rq_auth_stat = rpc_autherr_failed;
1391 goto err_bad_auth;
1392 }
1393
1394 if (progp == NULL)
1395 goto err_bad_prog;
1396
1397 switch (progp->pg_init_request(rqstp, progp, &process)) {
1398 case rpc_success:
1399 break;
1400 case rpc_prog_unavail:
1401 goto err_bad_prog;
1402 case rpc_prog_mismatch:
1403 goto err_bad_vers;
1404 case rpc_proc_unavail:
1405 goto err_bad_proc;
1406 }
1407
1408 procp = rqstp->rq_procinfo;
1409 /* Should this check go into the dispatcher? */
1410 if (!procp || !procp->pc_func)
1411 goto err_bad_proc;
1412
1413 /* Syntactic check complete */
1414 if (serv->sv_stats)
1415 serv->sv_stats->rpccnt++;
1416 trace_svc_process(rqst: rqstp, name: progp->pg_name);
1417
1418 aoffset = xdr_stream_pos(xdr);
1419
1420 /* un-reserve some of the out-queue now that we have a
1421 * better idea of reply size
1422 */
1423 if (procp->pc_xdrressize)
1424 svc_reserve_auth(rqstp, space: procp->pc_xdrressize<<2);
1425
1426 /* Call the function that processes the request. */
1427 rc = process.dispatch(rqstp);
1428 xdr_finish_decode(xdr);
1429
1430 if (!rc)
1431 goto dropit;
1432 if (rqstp->rq_auth_stat != rpc_auth_ok)
1433 goto err_bad_auth;
1434
1435 if (*rqstp->rq_accept_statp != rpc_success)
1436 xdr_truncate_encode(xdr, len: aoffset);
1437
1438 if (procp->pc_encode == NULL)
1439 goto dropit;
1440
1441 sendit:
1442 if (svc_authorise(rqstp))
1443 goto close_xprt;
1444 return 1; /* Caller can now send it */
1445
1446 dropit:
1447 svc_authorise(rqstp); /* doesn't hurt to call this twice */
1448 dprintk("svc: svc_process dropit\n");
1449 return 0;
1450
1451 close:
1452 svc_authorise(rqstp);
1453close_xprt:
1454 if (rqstp->rq_xprt && test_bit(XPT_TEMP, &rqstp->rq_xprt->xpt_flags))
1455 svc_xprt_close(xprt: rqstp->rq_xprt);
1456 dprintk("svc: svc_process close\n");
1457 return 0;
1458
1459err_short_len:
1460 svc_printk(rqstp, fmt: "short len %u, dropping request\n",
1461 rqstp->rq_arg.len);
1462 goto close_xprt;
1463
1464err_bad_rpc:
1465 if (serv->sv_stats)
1466 serv->sv_stats->rpcbadfmt++;
1467 xdr_stream_encode_u32(xdr, n: RPC_MSG_DENIED);
1468 xdr_stream_encode_u32(xdr, n: RPC_MISMATCH);
1469 /* Only RPCv2 supported */
1470 xdr_stream_encode_u32(xdr, RPC_VERSION);
1471 xdr_stream_encode_u32(xdr, RPC_VERSION);
1472 return 1; /* don't wrap */
1473
1474err_bad_auth:
1475 dprintk("svc: authentication failed (%d)\n",
1476 be32_to_cpu(rqstp->rq_auth_stat));
1477 if (serv->sv_stats)
1478 serv->sv_stats->rpcbadauth++;
1479 /* Restore write pointer to location of reply status: */
1480 xdr_truncate_encode(xdr, XDR_UNIT * 2);
1481 xdr_stream_encode_u32(xdr, n: RPC_MSG_DENIED);
1482 xdr_stream_encode_u32(xdr, n: RPC_AUTH_ERROR);
1483 xdr_stream_encode_be32(xdr, n: rqstp->rq_auth_stat);
1484 goto sendit;
1485
1486err_bad_prog:
1487 dprintk("svc: unknown program %d\n", rqstp->rq_prog);
1488 if (serv->sv_stats)
1489 serv->sv_stats->rpcbadfmt++;
1490 *rqstp->rq_accept_statp = rpc_prog_unavail;
1491 goto sendit;
1492
1493err_bad_vers:
1494 svc_printk(rqstp, fmt: "unknown version (%d for prog %d, %s)\n",
1495 rqstp->rq_vers, rqstp->rq_prog, progp->pg_name);
1496
1497 if (serv->sv_stats)
1498 serv->sv_stats->rpcbadfmt++;
1499 *rqstp->rq_accept_statp = rpc_prog_mismatch;
1500
1501 /*
1502 * svc_authenticate() has already added the verifier and
1503 * advanced the stream just past rq_accept_statp.
1504 */
1505 xdr_stream_encode_u32(xdr, n: process.mismatch.lovers);
1506 xdr_stream_encode_u32(xdr, n: process.mismatch.hivers);
1507 goto sendit;
1508
1509err_bad_proc:
1510 svc_printk(rqstp, fmt: "unknown procedure (%d)\n", rqstp->rq_proc);
1511
1512 if (serv->sv_stats)
1513 serv->sv_stats->rpcbadfmt++;
1514 *rqstp->rq_accept_statp = rpc_proc_unavail;
1515 goto sendit;
1516}
1517
1518/*
1519 * Drop request
1520 */
1521static void svc_drop(struct svc_rqst *rqstp)
1522{
1523 trace_svc_drop(rqst: rqstp);
1524}
1525
1526static void svc_release_rqst(struct svc_rqst *rqstp)
1527{
1528 const struct svc_procedure *procp = rqstp->rq_procinfo;
1529
1530 if (procp && procp->pc_release)
1531 procp->pc_release(rqstp);
1532}
1533
1534/**
1535 * svc_process - Execute one RPC transaction
1536 * @rqstp: RPC transaction context
1537 *
1538 */
1539void svc_process(struct svc_rqst *rqstp)
1540{
1541 struct kvec *resv = &rqstp->rq_res.head[0];
1542 __be32 *p;
1543
1544#if IS_ENABLED(CONFIG_FAIL_SUNRPC)
1545 if (!fail_sunrpc.ignore_server_disconnect &&
1546 should_fail(&fail_sunrpc.attr, 1))
1547 svc_xprt_deferred_close(rqstp->rq_xprt);
1548#endif
1549
1550 /*
1551 * Setup response xdr_buf.
1552 * Initially it has just one page
1553 */
1554 rqstp->rq_next_page = &rqstp->rq_respages[1];
1555 resv->iov_base = page_address(rqstp->rq_respages[0]);
1556 resv->iov_len = 0;
1557 rqstp->rq_res.pages = rqstp->rq_next_page;
1558 rqstp->rq_res.len = 0;
1559 rqstp->rq_res.page_base = 0;
1560 rqstp->rq_res.page_len = 0;
1561 rqstp->rq_res.buflen = PAGE_SIZE;
1562 rqstp->rq_res.tail[0].iov_base = NULL;
1563 rqstp->rq_res.tail[0].iov_len = 0;
1564
1565 svcxdr_init_decode(rqstp);
1566 p = xdr_inline_decode(xdr: &rqstp->rq_arg_stream, XDR_UNIT * 2);
1567 if (unlikely(!p))
1568 goto out_drop;
1569 rqstp->rq_xid = *p++;
1570 if (unlikely(*p != rpc_call))
1571 goto out_baddir;
1572
1573 if (!svc_process_common(rqstp)) {
1574 svc_release_rqst(rqstp);
1575 goto out_drop;
1576 }
1577 svc_send(rqstp);
1578 svc_release_rqst(rqstp);
1579 return;
1580
1581out_baddir:
1582 svc_printk(rqstp, fmt: "bad direction 0x%08x, dropping request\n",
1583 be32_to_cpu(*p));
1584 if (rqstp->rq_server->sv_stats)
1585 rqstp->rq_server->sv_stats->rpcbadfmt++;
1586out_drop:
1587 svc_drop(rqstp);
1588}
1589
1590#if defined(CONFIG_SUNRPC_BACKCHANNEL)
1591/**
1592 * svc_process_bc - process a reverse-direction RPC request
1593 * @req: RPC request to be used for client-side processing
1594 * @rqstp: server-side execution context
1595 *
1596 */
1597void svc_process_bc(struct rpc_rqst *req, struct svc_rqst *rqstp)
1598{
1599 struct rpc_timeout timeout = {
1600 .to_increment = 0,
1601 };
1602 struct rpc_task *task;
1603 int proc_error;
1604
1605 /* Build the svc_rqst used by the common processing routine */
1606 rqstp->rq_xid = req->rq_xid;
1607 rqstp->rq_prot = req->rq_xprt->prot;
1608 rqstp->rq_bc_net = req->rq_xprt->xprt_net;
1609
1610 rqstp->rq_addrlen = sizeof(req->rq_xprt->addr);
1611 memcpy(&rqstp->rq_addr, &req->rq_xprt->addr, rqstp->rq_addrlen);
1612 memcpy(&rqstp->rq_arg, &req->rq_rcv_buf, sizeof(rqstp->rq_arg));
1613 memcpy(&rqstp->rq_res, &req->rq_snd_buf, sizeof(rqstp->rq_res));
1614
1615 /* Adjust the argument buffer length */
1616 rqstp->rq_arg.len = req->rq_private_buf.len;
1617 if (rqstp->rq_arg.len <= rqstp->rq_arg.head[0].iov_len) {
1618 rqstp->rq_arg.head[0].iov_len = rqstp->rq_arg.len;
1619 rqstp->rq_arg.page_len = 0;
1620 } else if (rqstp->rq_arg.len <= rqstp->rq_arg.head[0].iov_len +
1621 rqstp->rq_arg.page_len)
1622 rqstp->rq_arg.page_len = rqstp->rq_arg.len -
1623 rqstp->rq_arg.head[0].iov_len;
1624 else
1625 rqstp->rq_arg.len = rqstp->rq_arg.head[0].iov_len +
1626 rqstp->rq_arg.page_len;
1627
1628 /* Reset the response buffer */
1629 rqstp->rq_res.head[0].iov_len = 0;
1630
1631 /*
1632 * Skip the XID and calldir fields because they've already
1633 * been processed by the caller.
1634 */
1635 svcxdr_init_decode(rqstp);
1636 if (!xdr_inline_decode(&rqstp->rq_arg_stream, XDR_UNIT * 2))
1637 return;
1638
1639 /* Parse and execute the bc call */
1640 proc_error = svc_process_common(rqstp);
1641
1642 atomic_dec(&req->rq_xprt->bc_slot_count);
1643 if (!proc_error) {
1644 /* Processing error: drop the request */
1645 xprt_free_bc_request(req);
1646 svc_release_rqst(rqstp);
1647 return;
1648 }
1649 /* Finally, send the reply synchronously */
1650 if (rqstp->bc_to_initval > 0) {
1651 timeout.to_initval = rqstp->bc_to_initval;
1652 timeout.to_retries = rqstp->bc_to_retries;
1653 } else {
1654 timeout.to_initval = req->rq_xprt->timeout->to_initval;
1655 timeout.to_retries = req->rq_xprt->timeout->to_retries;
1656 }
1657 timeout.to_maxval = timeout.to_initval;
1658 memcpy(&req->rq_snd_buf, &rqstp->rq_res, sizeof(req->rq_snd_buf));
1659 task = rpc_run_bc_task(req, &timeout);
1660 svc_release_rqst(rqstp);
1661
1662 if (IS_ERR(task))
1663 return;
1664
1665 WARN_ON_ONCE(atomic_read(&task->tk_count) != 1);
1666 rpc_put_task(task);
1667}
1668#endif /* CONFIG_SUNRPC_BACKCHANNEL */
1669
1670/**
1671 * svc_max_payload - Return transport-specific limit on the RPC payload
1672 * @rqstp: RPC transaction context
1673 *
1674 * Returns the maximum number of payload bytes the current transport
1675 * allows.
1676 */
1677u32 svc_max_payload(const struct svc_rqst *rqstp)
1678{
1679 u32 max = rqstp->rq_xprt->xpt_class->xcl_max_payload;
1680
1681 if (rqstp->rq_server->sv_max_payload < max)
1682 max = rqstp->rq_server->sv_max_payload;
1683 return max;
1684}
1685EXPORT_SYMBOL_GPL(svc_max_payload);
1686
1687/**
1688 * svc_proc_name - Return RPC procedure name in string form
1689 * @rqstp: svc_rqst to operate on
1690 *
1691 * Return value:
1692 * Pointer to a NUL-terminated string
1693 */
1694const char *svc_proc_name(const struct svc_rqst *rqstp)
1695{
1696 if (rqstp && rqstp->rq_procinfo)
1697 return rqstp->rq_procinfo->pc_name;
1698 return "unknown";
1699}
1700
1701
1702/**
1703 * svc_encode_result_payload - mark a range of bytes as a result payload
1704 * @rqstp: svc_rqst to operate on
1705 * @offset: payload's byte offset in rqstp->rq_res
1706 * @length: size of payload, in bytes
1707 *
1708 * Returns zero on success, or a negative errno if a permanent
1709 * error occurred.
1710 */
1711int svc_encode_result_payload(struct svc_rqst *rqstp, unsigned int offset,
1712 unsigned int length)
1713{
1714 return rqstp->rq_xprt->xpt_ops->xpo_result_payload(rqstp, offset,
1715 length);
1716}
1717EXPORT_SYMBOL_GPL(svc_encode_result_payload);
1718
1719/**
1720 * svc_fill_symlink_pathname - Construct pathname argument for VFS symlink call
1721 * @rqstp: svc_rqst to operate on
1722 * @first: buffer containing first section of pathname
1723 * @p: buffer containing remaining section of pathname
1724 * @total: total length of the pathname argument
1725 *
1726 * The VFS symlink API demands a NUL-terminated pathname in mapped memory.
1727 * Returns pointer to a NUL-terminated string, or an ERR_PTR. Caller must free
1728 * the returned string.
1729 */
1730char *svc_fill_symlink_pathname(struct svc_rqst *rqstp, struct kvec *first,
1731 void *p, size_t total)
1732{
1733 size_t len, remaining;
1734 char *result, *dst;
1735
1736 result = kmalloc(total + 1, GFP_KERNEL);
1737 if (!result)
1738 return ERR_PTR(error: -ESERVERFAULT);
1739
1740 dst = result;
1741 remaining = total;
1742
1743 len = min_t(size_t, total, first->iov_len);
1744 if (len) {
1745 memcpy(to: dst, from: first->iov_base, len);
1746 dst += len;
1747 remaining -= len;
1748 }
1749
1750 if (remaining) {
1751 len = min_t(size_t, remaining, PAGE_SIZE);
1752 memcpy(to: dst, from: p, len);
1753 dst += len;
1754 }
1755
1756 *dst = '\0';
1757
1758 /* Sanity check: Linux doesn't allow the pathname argument to
1759 * contain a NUL byte.
1760 */
1761 if (strlen(result) != total) {
1762 kfree(objp: result);
1763 return ERR_PTR(error: -EINVAL);
1764 }
1765 return result;
1766}
1767EXPORT_SYMBOL_GPL(svc_fill_symlink_pathname);
1768