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