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 #define SVC_POOL_DEFAULT SVC_POOL_GLOBAL 42 43 /* 44 * Mode for mapping cpus to pools. 45 */ 46 enum { 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 59 struct 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 69 static struct svc_pool_map svc_pool_map = { 70 .mode = SVC_POOL_DEFAULT 71 }; 72 73 static DEFINE_MUTEX(svc_pool_map_mutex);/* protects svc_pool_map.count only */ 74 75 static int 76 __param_set_pool_mode(const char *val, struct svc_pool_map *m) 77 { 78 int err, mode; 79 80 mutex_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; 101 out: 102 mutex_unlock(&svc_pool_map_mutex); 103 return err; 104 } 105 106 static int 107 param_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 114 int sunrpc_set_pool_mode(const char *val) 115 { 116 return __param_set_pool_mode(val, &svc_pool_map); 117 } 118 EXPORT_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 */ 129 int 130 sunrpc_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, "auto"); 138 case SVC_POOL_GLOBAL: 139 return snprintf(buf, size, "global"); 140 case SVC_POOL_PERCPU: 141 return snprintf(buf, size, "percpu"); 142 case SVC_POOL_PERNODE: 143 return snprintf(buf, size, "pernode"); 144 default: 145 return snprintf(buf, size, "%d", m->mode); 146 } 147 } 148 EXPORT_SYMBOL(sunrpc_get_pool_mode); 149 150 static int 151 param_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, "%s", str); 166 } 167 168 module_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 */ 175 static int 176 svc_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 */ 207 static int 208 svc_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 219 fail_free: 220 kfree(m->to_pool); 221 m->to_pool = NULL; 222 fail: 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 */ 230 static int 231 svc_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 */ 258 static int 259 svc_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 */ 290 static unsigned int 291 svc_pool_map_get(void) 292 { 293 struct svc_pool_map *m = &svc_pool_map; 294 int npools = -1; 295 296 mutex_lock(&svc_pool_map_mutex); 297 if (m->count++) { 298 mutex_unlock(&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(&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 */ 329 static void 330 svc_pool_map_put(void) 331 { 332 struct svc_pool_map *m = &svc_pool_map; 333 334 mutex_lock(&svc_pool_map_mutex); 335 if (!--m->count) { 336 kfree(m->to_pool); 337 m->to_pool = NULL; 338 kfree(m->pool_to); 339 m->pool_to = NULL; 340 m->npools = 0; 341 } 342 mutex_unlock(&svc_pool_map_mutex); 343 } 344 345 static 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(m->pool_to[pidx]); 352 if (m->mode == SVC_POOL_PERNODE) 353 return m->pool_to[pidx]; 354 } 355 return NUMA_NO_NODE; 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 */ 361 static inline void 362 svc_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(task, cpumask_of(node)); 379 break; 380 } 381 case SVC_POOL_PERNODE: 382 { 383 set_cpus_allowed_ptr(task, 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 */ 400 struct 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 421 static 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 434 void svc_rpcb_cleanup(struct svc_serv *serv, struct net *net) 435 { 436 svc_unregister(serv, net); 437 rpcb_put_local(net); 438 } 439 EXPORT_SYMBOL_GPL(svc_rpcb_cleanup); 440 441 static int svc_uses_rpcbind(struct svc_serv *serv) 442 { 443 unsigned int p, i; 444 445 for (p = 0; p < serv->sv_nprogs; p++) { 446 struct svc_program *progp = &serv->sv_programs[p]; 447 448 for (i = 0; i < progp->pg_nvers; i++) { 449 if (progp->pg_vers[i] == NULL) 450 continue; 451 if (!progp->pg_vers[i]->vs_hidden) 452 return 1; 453 } 454 } 455 456 return 0; 457 } 458 459 int svc_bind(struct svc_serv *serv, struct net *net) 460 { 461 if (!svc_uses_rpcbind(serv)) 462 return 0; 463 return svc_rpcb_setup(serv, net); 464 } 465 EXPORT_SYMBOL_GPL(svc_bind); 466 467 #if defined(CONFIG_SUNRPC_BACKCHANNEL) 468 static void 469 __svc_init_bc(struct svc_serv *serv) 470 { 471 lwq_init(&serv->sv_cb_list); 472 } 473 #else 474 static void 475 __svc_init_bc(struct svc_serv *serv) 476 { 477 } 478 #endif 479 480 /* 481 * Create an RPC service 482 */ 483 static struct svc_serv * 484 __svc_create(struct svc_program *prog, int nprogs, struct svc_stat *stats, 485 unsigned int bufsize, int npools, int (*threadfn)(void *data)) 486 { 487 struct svc_serv *serv; 488 unsigned int vers; 489 unsigned int xdrsize; 490 unsigned int i; 491 492 if (!(serv = kzalloc(sizeof(*serv), GFP_KERNEL))) 493 return NULL; 494 serv->sv_name = prog->pg_name; 495 serv->sv_programs = prog; 496 serv->sv_nprogs = nprogs; 497 serv->sv_stats = stats; 498 if (bufsize > RPCSVC_MAXPAYLOAD) 499 bufsize = RPCSVC_MAXPAYLOAD; 500 serv->sv_max_payload = bufsize? bufsize : 4096; 501 serv->sv_max_mesg = roundup(serv->sv_max_payload + PAGE_SIZE, PAGE_SIZE); 502 serv->sv_threadfn = threadfn; 503 xdrsize = 0; 504 for (i = 0; i < nprogs; i++) { 505 struct svc_program *progp = &prog[i]; 506 507 progp->pg_lovers = progp->pg_nvers-1; 508 for (vers = 0; vers < progp->pg_nvers ; vers++) 509 if (progp->pg_vers[vers]) { 510 progp->pg_hivers = vers; 511 if (progp->pg_lovers > vers) 512 progp->pg_lovers = vers; 513 if (progp->pg_vers[vers]->vs_xdrsize > xdrsize) 514 xdrsize = progp->pg_vers[vers]->vs_xdrsize; 515 } 516 } 517 serv->sv_xdrsize = xdrsize; 518 INIT_LIST_HEAD(&serv->sv_tempsocks); 519 INIT_LIST_HEAD(&serv->sv_permsocks); 520 timer_setup(&serv->sv_temptimer, NULL, 0); 521 spin_lock_init(&serv->sv_lock); 522 523 __svc_init_bc(serv); 524 525 serv->sv_nrpools = npools; 526 serv->sv_pools = 527 kcalloc(serv->sv_nrpools, sizeof(struct svc_pool), 528 GFP_KERNEL); 529 if (!serv->sv_pools) { 530 kfree(serv); 531 return NULL; 532 } 533 534 for (i = 0; i < serv->sv_nrpools; i++) { 535 struct svc_pool *pool = &serv->sv_pools[i]; 536 537 dprintk("svc: initialising pool %u for %s\n", 538 i, serv->sv_name); 539 540 pool->sp_id = i; 541 lwq_init(&pool->sp_xprts); 542 INIT_LIST_HEAD(&pool->sp_all_threads); 543 init_llist_head(&pool->sp_idle_threads); 544 545 percpu_counter_init(&pool->sp_messages_arrived, 0, GFP_KERNEL); 546 percpu_counter_init(&pool->sp_sockets_queued, 0, GFP_KERNEL); 547 percpu_counter_init(&pool->sp_threads_woken, 0, GFP_KERNEL); 548 } 549 550 return serv; 551 } 552 553 /** 554 * svc_create - Create an RPC service 555 * @prog: the RPC program the new service will handle 556 * @bufsize: maximum message size for @prog 557 * @threadfn: a function to service RPC requests for @prog 558 * 559 * Returns an instantiated struct svc_serv object or NULL. 560 */ 561 struct svc_serv *svc_create(struct svc_program *prog, unsigned int bufsize, 562 int (*threadfn)(void *data)) 563 { 564 return __svc_create(prog, 1, NULL, bufsize, 1, threadfn); 565 } 566 EXPORT_SYMBOL_GPL(svc_create); 567 568 /** 569 * svc_create_pooled - Create an RPC service with pooled threads 570 * @prog: Array of RPC programs the new service will handle 571 * @nprogs: Number of programs in the array 572 * @stats: the stats struct if desired 573 * @bufsize: maximum message size for @prog 574 * @threadfn: a function to service RPC requests for @prog 575 * 576 * Returns an instantiated struct svc_serv object or NULL. 577 */ 578 struct svc_serv *svc_create_pooled(struct svc_program *prog, 579 unsigned int nprogs, 580 struct svc_stat *stats, 581 unsigned int bufsize, 582 int (*threadfn)(void *data)) 583 { 584 struct svc_serv *serv; 585 unsigned int npools = svc_pool_map_get(); 586 587 serv = __svc_create(prog, nprogs, stats, bufsize, npools, threadfn); 588 if (!serv) 589 goto out_err; 590 serv->sv_is_pooled = true; 591 return serv; 592 out_err: 593 svc_pool_map_put(); 594 return NULL; 595 } 596 EXPORT_SYMBOL_GPL(svc_create_pooled); 597 598 /* 599 * Destroy an RPC service. Should be called with appropriate locking to 600 * protect sv_permsocks and sv_tempsocks. 601 */ 602 void 603 svc_destroy(struct svc_serv **servp) 604 { 605 struct svc_serv *serv = *servp; 606 unsigned int i; 607 608 *servp = NULL; 609 610 dprintk("svc: svc_destroy(%s)\n", serv->sv_programs->pg_name); 611 timer_shutdown_sync(&serv->sv_temptimer); 612 613 /* 614 * Remaining transports at this point are not expected. 615 */ 616 WARN_ONCE(!list_empty(&serv->sv_permsocks), 617 "SVC: permsocks remain for %s\n", serv->sv_programs->pg_name); 618 WARN_ONCE(!list_empty(&serv->sv_tempsocks), 619 "SVC: tempsocks remain for %s\n", serv->sv_programs->pg_name); 620 621 cache_clean_deferred(serv); 622 623 if (serv->sv_is_pooled) 624 svc_pool_map_put(); 625 626 for (i = 0; i < serv->sv_nrpools; i++) { 627 struct svc_pool *pool = &serv->sv_pools[i]; 628 629 percpu_counter_destroy(&pool->sp_messages_arrived); 630 percpu_counter_destroy(&pool->sp_sockets_queued); 631 percpu_counter_destroy(&pool->sp_threads_woken); 632 } 633 kfree(serv->sv_pools); 634 kfree(serv); 635 } 636 EXPORT_SYMBOL_GPL(svc_destroy); 637 638 static bool 639 svc_init_buffer(struct svc_rqst *rqstp, unsigned int size, int node) 640 { 641 unsigned long pages, ret; 642 643 /* bc_xprt uses fore channel allocated buffers */ 644 if (svc_is_backchannel(rqstp)) 645 return true; 646 647 pages = size / PAGE_SIZE + 1; /* extra page as we hold both request and reply. 648 * We assume one is at most one page 649 */ 650 WARN_ON_ONCE(pages > RPCSVC_MAXPAGES); 651 if (pages > RPCSVC_MAXPAGES) 652 pages = RPCSVC_MAXPAGES; 653 654 ret = alloc_pages_bulk_array_node(GFP_KERNEL, node, pages, 655 rqstp->rq_pages); 656 return ret == pages; 657 } 658 659 /* 660 * Release an RPC server buffer 661 */ 662 static void 663 svc_release_buffer(struct svc_rqst *rqstp) 664 { 665 unsigned int i; 666 667 for (i = 0; i < ARRAY_SIZE(rqstp->rq_pages); i++) 668 if (rqstp->rq_pages[i]) 669 put_page(rqstp->rq_pages[i]); 670 } 671 672 static void 673 svc_rqst_free(struct svc_rqst *rqstp) 674 { 675 folio_batch_release(&rqstp->rq_fbatch); 676 svc_release_buffer(rqstp); 677 if (rqstp->rq_scratch_page) 678 put_page(rqstp->rq_scratch_page); 679 kfree(rqstp->rq_resp); 680 kfree(rqstp->rq_argp); 681 kfree(rqstp->rq_auth_data); 682 kfree_rcu(rqstp, rq_rcu_head); 683 } 684 685 static struct svc_rqst * 686 svc_prepare_thread(struct svc_serv *serv, struct svc_pool *pool, int node) 687 { 688 struct svc_rqst *rqstp; 689 690 rqstp = kzalloc_node(sizeof(*rqstp), GFP_KERNEL, node); 691 if (!rqstp) 692 return rqstp; 693 694 folio_batch_init(&rqstp->rq_fbatch); 695 696 rqstp->rq_server = serv; 697 rqstp->rq_pool = pool; 698 699 rqstp->rq_scratch_page = alloc_pages_node(node, GFP_KERNEL, 0); 700 if (!rqstp->rq_scratch_page) 701 goto out_enomem; 702 703 rqstp->rq_argp = kmalloc_node(serv->sv_xdrsize, GFP_KERNEL, node); 704 if (!rqstp->rq_argp) 705 goto out_enomem; 706 707 rqstp->rq_resp = kmalloc_node(serv->sv_xdrsize, GFP_KERNEL, node); 708 if (!rqstp->rq_resp) 709 goto out_enomem; 710 711 if (!svc_init_buffer(rqstp, serv->sv_max_mesg, node)) 712 goto out_enomem; 713 714 rqstp->rq_err = -EAGAIN; /* No error yet */ 715 716 serv->sv_nrthreads += 1; 717 pool->sp_nrthreads += 1; 718 719 /* Protected by whatever lock the service uses when calling 720 * svc_set_num_threads() 721 */ 722 list_add_rcu(&rqstp->rq_all, &pool->sp_all_threads); 723 724 return rqstp; 725 726 out_enomem: 727 svc_rqst_free(rqstp); 728 return NULL; 729 } 730 731 /** 732 * svc_pool_wake_idle_thread - Awaken an idle thread in @pool 733 * @pool: service thread pool 734 * 735 * Can be called from soft IRQ or process context. Finding an idle 736 * service thread and marking it BUSY is atomic with respect to 737 * other calls to svc_pool_wake_idle_thread(). 738 * 739 */ 740 void svc_pool_wake_idle_thread(struct svc_pool *pool) 741 { 742 struct svc_rqst *rqstp; 743 struct llist_node *ln; 744 745 rcu_read_lock(); 746 ln = READ_ONCE(pool->sp_idle_threads.first); 747 if (ln) { 748 rqstp = llist_entry(ln, struct svc_rqst, rq_idle); 749 WRITE_ONCE(rqstp->rq_qtime, ktime_get()); 750 if (!task_is_running(rqstp->rq_task)) { 751 wake_up_process(rqstp->rq_task); 752 trace_svc_wake_up(rqstp->rq_task->pid); 753 percpu_counter_inc(&pool->sp_threads_woken); 754 } 755 rcu_read_unlock(); 756 return; 757 } 758 rcu_read_unlock(); 759 760 } 761 EXPORT_SYMBOL_GPL(svc_pool_wake_idle_thread); 762 763 static struct svc_pool * 764 svc_pool_next(struct svc_serv *serv, struct svc_pool *pool, unsigned int *state) 765 { 766 return pool ? pool : &serv->sv_pools[(*state)++ % serv->sv_nrpools]; 767 } 768 769 static struct svc_pool * 770 svc_pool_victim(struct svc_serv *serv, struct svc_pool *target_pool, 771 unsigned int *state) 772 { 773 struct svc_pool *pool; 774 unsigned int i; 775 776 pool = target_pool; 777 778 if (!pool) { 779 for (i = 0; i < serv->sv_nrpools; i++) { 780 pool = &serv->sv_pools[--(*state) % serv->sv_nrpools]; 781 if (pool->sp_nrthreads) 782 break; 783 } 784 } 785 786 if (pool && pool->sp_nrthreads) { 787 set_bit(SP_VICTIM_REMAINS, &pool->sp_flags); 788 set_bit(SP_NEED_VICTIM, &pool->sp_flags); 789 return pool; 790 } 791 return NULL; 792 } 793 794 static int 795 svc_start_kthreads(struct svc_serv *serv, struct svc_pool *pool, int nrservs) 796 { 797 struct svc_rqst *rqstp; 798 struct task_struct *task; 799 struct svc_pool *chosen_pool; 800 unsigned int state = serv->sv_nrthreads-1; 801 int node; 802 int err; 803 804 do { 805 nrservs--; 806 chosen_pool = svc_pool_next(serv, pool, &state); 807 node = svc_pool_map_get_node(chosen_pool->sp_id); 808 809 rqstp = svc_prepare_thread(serv, chosen_pool, node); 810 if (!rqstp) 811 return -ENOMEM; 812 task = kthread_create_on_node(serv->sv_threadfn, rqstp, 813 node, "%s", serv->sv_name); 814 if (IS_ERR(task)) { 815 svc_exit_thread(rqstp); 816 return PTR_ERR(task); 817 } 818 819 rqstp->rq_task = task; 820 if (serv->sv_nrpools > 1) 821 svc_pool_map_set_cpumask(task, chosen_pool->sp_id); 822 823 svc_sock_update_bufs(serv); 824 wake_up_process(task); 825 826 wait_var_event(&rqstp->rq_err, rqstp->rq_err != -EAGAIN); 827 err = rqstp->rq_err; 828 if (err) { 829 svc_exit_thread(rqstp); 830 return err; 831 } 832 } while (nrservs > 0); 833 834 return 0; 835 } 836 837 static int 838 svc_stop_kthreads(struct svc_serv *serv, struct svc_pool *pool, int nrservs) 839 { 840 unsigned int state = serv->sv_nrthreads-1; 841 struct svc_pool *victim; 842 843 do { 844 victim = svc_pool_victim(serv, pool, &state); 845 if (!victim) 846 break; 847 svc_pool_wake_idle_thread(victim); 848 wait_on_bit(&victim->sp_flags, SP_VICTIM_REMAINS, 849 TASK_IDLE); 850 nrservs++; 851 } while (nrservs < 0); 852 return 0; 853 } 854 855 /** 856 * svc_set_num_threads - adjust number of threads per RPC service 857 * @serv: RPC service to adjust 858 * @pool: Specific pool from which to choose threads, or NULL 859 * @nrservs: New number of threads for @serv (0 or less means kill all threads) 860 * 861 * Create or destroy threads to make the number of threads for @serv the 862 * given number. If @pool is non-NULL, change only threads in that pool; 863 * otherwise, round-robin between all pools for @serv. @serv's 864 * sv_nrthreads is adjusted for each thread created or destroyed. 865 * 866 * Caller must ensure mutual exclusion between this and server startup or 867 * shutdown. 868 * 869 * Returns zero on success or a negative errno if an error occurred while 870 * starting a thread. 871 */ 872 int 873 svc_set_num_threads(struct svc_serv *serv, struct svc_pool *pool, int nrservs) 874 { 875 if (!pool) 876 nrservs -= serv->sv_nrthreads; 877 else 878 nrservs -= pool->sp_nrthreads; 879 880 if (nrservs > 0) 881 return svc_start_kthreads(serv, pool, nrservs); 882 if (nrservs < 0) 883 return svc_stop_kthreads(serv, pool, nrservs); 884 return 0; 885 } 886 EXPORT_SYMBOL_GPL(svc_set_num_threads); 887 888 /** 889 * svc_rqst_replace_page - Replace one page in rq_pages[] 890 * @rqstp: svc_rqst with pages to replace 891 * @page: replacement page 892 * 893 * When replacing a page in rq_pages, batch the release of the 894 * replaced pages to avoid hammering the page allocator. 895 * 896 * Return values: 897 * %true: page replaced 898 * %false: array bounds checking failed 899 */ 900 bool svc_rqst_replace_page(struct svc_rqst *rqstp, struct page *page) 901 { 902 struct page **begin = rqstp->rq_pages; 903 struct page **end = &rqstp->rq_pages[RPCSVC_MAXPAGES]; 904 905 if (unlikely(rqstp->rq_next_page < begin || rqstp->rq_next_page > end)) { 906 trace_svc_replace_page_err(rqstp); 907 return false; 908 } 909 910 if (*rqstp->rq_next_page) { 911 if (!folio_batch_add(&rqstp->rq_fbatch, 912 page_folio(*rqstp->rq_next_page))) 913 __folio_batch_release(&rqstp->rq_fbatch); 914 } 915 916 get_page(page); 917 *(rqstp->rq_next_page++) = page; 918 return true; 919 } 920 EXPORT_SYMBOL_GPL(svc_rqst_replace_page); 921 922 /** 923 * svc_rqst_release_pages - Release Reply buffer pages 924 * @rqstp: RPC transaction context 925 * 926 * Release response pages that might still be in flight after 927 * svc_send, and any spliced filesystem-owned pages. 928 */ 929 void svc_rqst_release_pages(struct svc_rqst *rqstp) 930 { 931 int i, count = rqstp->rq_next_page - rqstp->rq_respages; 932 933 if (count) { 934 release_pages(rqstp->rq_respages, count); 935 for (i = 0; i < count; i++) 936 rqstp->rq_respages[i] = NULL; 937 } 938 } 939 940 /** 941 * svc_exit_thread - finalise the termination of a sunrpc server thread 942 * @rqstp: the svc_rqst which represents the thread. 943 * 944 * When a thread started with svc_new_thread() exits it must call 945 * svc_exit_thread() as its last act. This must be done with the 946 * service mutex held. Normally this is held by a DIFFERENT thread, the 947 * one that is calling svc_set_num_threads() and which will wait for 948 * SP_VICTIM_REMAINS to be cleared before dropping the mutex. If the 949 * thread exits for any reason other than svc_thread_should_stop() 950 * returning %true (which indicated that svc_set_num_threads() is 951 * waiting for it to exit), then it must take the service mutex itself, 952 * which can only safely be done using mutex_try_lock(). 953 */ 954 void 955 svc_exit_thread(struct svc_rqst *rqstp) 956 { 957 struct svc_serv *serv = rqstp->rq_server; 958 struct svc_pool *pool = rqstp->rq_pool; 959 960 list_del_rcu(&rqstp->rq_all); 961 962 pool->sp_nrthreads -= 1; 963 serv->sv_nrthreads -= 1; 964 svc_sock_update_bufs(serv); 965 966 svc_rqst_free(rqstp); 967 968 clear_and_wake_up_bit(SP_VICTIM_REMAINS, &pool->sp_flags); 969 } 970 EXPORT_SYMBOL_GPL(svc_exit_thread); 971 972 /* 973 * Register an "inet" protocol family netid with the local 974 * rpcbind daemon via an rpcbind v4 SET request. 975 * 976 * No netconfig infrastructure is available in the kernel, so 977 * we map IP_ protocol numbers to netids by hand. 978 * 979 * Returns zero on success; a negative errno value is returned 980 * if any error occurs. 981 */ 982 static int __svc_rpcb_register4(struct net *net, const u32 program, 983 const u32 version, 984 const unsigned short protocol, 985 const unsigned short port) 986 { 987 const struct sockaddr_in sin = { 988 .sin_family = AF_INET, 989 .sin_addr.s_addr = htonl(INADDR_ANY), 990 .sin_port = htons(port), 991 }; 992 const char *netid; 993 int error; 994 995 switch (protocol) { 996 case IPPROTO_UDP: 997 netid = RPCBIND_NETID_UDP; 998 break; 999 case IPPROTO_TCP: 1000 netid = RPCBIND_NETID_TCP; 1001 break; 1002 default: 1003 return -ENOPROTOOPT; 1004 } 1005 1006 error = rpcb_v4_register(net, program, version, 1007 (const struct sockaddr *)&sin, netid); 1008 1009 /* 1010 * User space didn't support rpcbind v4, so retry this 1011 * registration request with the legacy rpcbind v2 protocol. 1012 */ 1013 if (error == -EPROTONOSUPPORT) 1014 error = rpcb_register(net, program, version, protocol, port); 1015 1016 return error; 1017 } 1018 1019 #if IS_ENABLED(CONFIG_IPV6) 1020 /* 1021 * Register an "inet6" protocol family netid with the local 1022 * rpcbind daemon via an rpcbind v4 SET request. 1023 * 1024 * No netconfig infrastructure is available in the kernel, so 1025 * we map IP_ protocol numbers to netids by hand. 1026 * 1027 * Returns zero on success; a negative errno value is returned 1028 * if any error occurs. 1029 */ 1030 static int __svc_rpcb_register6(struct net *net, const u32 program, 1031 const u32 version, 1032 const unsigned short protocol, 1033 const unsigned short port) 1034 { 1035 const struct sockaddr_in6 sin6 = { 1036 .sin6_family = AF_INET6, 1037 .sin6_addr = IN6ADDR_ANY_INIT, 1038 .sin6_port = htons(port), 1039 }; 1040 const char *netid; 1041 int error; 1042 1043 switch (protocol) { 1044 case IPPROTO_UDP: 1045 netid = RPCBIND_NETID_UDP6; 1046 break; 1047 case IPPROTO_TCP: 1048 netid = RPCBIND_NETID_TCP6; 1049 break; 1050 default: 1051 return -ENOPROTOOPT; 1052 } 1053 1054 error = rpcb_v4_register(net, program, version, 1055 (const struct sockaddr *)&sin6, netid); 1056 1057 /* 1058 * User space didn't support rpcbind version 4, so we won't 1059 * use a PF_INET6 listener. 1060 */ 1061 if (error == -EPROTONOSUPPORT) 1062 error = -EAFNOSUPPORT; 1063 1064 return error; 1065 } 1066 #endif /* IS_ENABLED(CONFIG_IPV6) */ 1067 1068 /* 1069 * Register a kernel RPC service via rpcbind version 4. 1070 * 1071 * Returns zero on success; a negative errno value is returned 1072 * if any error occurs. 1073 */ 1074 static int __svc_register(struct net *net, const char *progname, 1075 const u32 program, const u32 version, 1076 const int family, 1077 const unsigned short protocol, 1078 const unsigned short port) 1079 { 1080 int error = -EAFNOSUPPORT; 1081 1082 switch (family) { 1083 case PF_INET: 1084 error = __svc_rpcb_register4(net, program, version, 1085 protocol, port); 1086 break; 1087 #if IS_ENABLED(CONFIG_IPV6) 1088 case PF_INET6: 1089 error = __svc_rpcb_register6(net, program, version, 1090 protocol, port); 1091 #endif 1092 } 1093 1094 trace_svc_register(progname, version, family, protocol, port, error); 1095 return error; 1096 } 1097 1098 static 1099 int svc_rpcbind_set_version(struct net *net, 1100 const struct svc_program *progp, 1101 u32 version, int family, 1102 unsigned short proto, 1103 unsigned short port) 1104 { 1105 return __svc_register(net, progp->pg_name, progp->pg_prog, 1106 version, family, proto, port); 1107 1108 } 1109 1110 int svc_generic_rpcbind_set(struct net *net, 1111 const struct svc_program *progp, 1112 u32 version, int family, 1113 unsigned short proto, 1114 unsigned short port) 1115 { 1116 const struct svc_version *vers = progp->pg_vers[version]; 1117 int error; 1118 1119 if (vers == NULL) 1120 return 0; 1121 1122 if (vers->vs_hidden) { 1123 trace_svc_noregister(progp->pg_name, version, proto, 1124 port, family, 0); 1125 return 0; 1126 } 1127 1128 /* 1129 * Don't register a UDP port if we need congestion 1130 * control. 1131 */ 1132 if (vers->vs_need_cong_ctrl && proto == IPPROTO_UDP) 1133 return 0; 1134 1135 error = svc_rpcbind_set_version(net, progp, version, 1136 family, proto, port); 1137 1138 return (vers->vs_rpcb_optnl) ? 0 : error; 1139 } 1140 EXPORT_SYMBOL_GPL(svc_generic_rpcbind_set); 1141 1142 /** 1143 * svc_register - register an RPC service with the local portmapper 1144 * @serv: svc_serv struct for the service to register 1145 * @net: net namespace for the service to register 1146 * @family: protocol family of service's listener socket 1147 * @proto: transport protocol number to advertise 1148 * @port: port to advertise 1149 * 1150 * Service is registered for any address in the passed-in protocol family 1151 */ 1152 int svc_register(const struct svc_serv *serv, struct net *net, 1153 const int family, const unsigned short proto, 1154 const unsigned short port) 1155 { 1156 unsigned int p, i; 1157 int error = 0; 1158 1159 WARN_ON_ONCE(proto == 0 && port == 0); 1160 if (proto == 0 && port == 0) 1161 return -EINVAL; 1162 1163 for (p = 0; p < serv->sv_nprogs; p++) { 1164 struct svc_program *progp = &serv->sv_programs[p]; 1165 1166 for (i = 0; i < progp->pg_nvers; i++) { 1167 1168 error = progp->pg_rpcbind_set(net, progp, i, 1169 family, proto, port); 1170 if (error < 0) { 1171 printk(KERN_WARNING "svc: failed to register " 1172 "%sv%u RPC service (errno %d).\n", 1173 progp->pg_name, i, -error); 1174 break; 1175 } 1176 } 1177 } 1178 1179 return error; 1180 } 1181 1182 /* 1183 * If user space is running rpcbind, it should take the v4 UNSET 1184 * and clear everything for this [program, version]. If user space 1185 * is running portmap, it will reject the v4 UNSET, but won't have 1186 * any "inet6" entries anyway. So a PMAP_UNSET should be sufficient 1187 * in this case to clear all existing entries for [program, version]. 1188 */ 1189 static void __svc_unregister(struct net *net, const u32 program, const u32 version, 1190 const char *progname) 1191 { 1192 int error; 1193 1194 error = rpcb_v4_register(net, program, version, NULL, ""); 1195 1196 /* 1197 * User space didn't support rpcbind v4, so retry this 1198 * request with the legacy rpcbind v2 protocol. 1199 */ 1200 if (error == -EPROTONOSUPPORT) 1201 error = rpcb_register(net, program, version, 0, 0); 1202 1203 trace_svc_unregister(progname, version, error); 1204 } 1205 1206 /* 1207 * All netids, bind addresses and ports registered for [program, version] 1208 * are removed from the local rpcbind database (if the service is not 1209 * hidden) to make way for a new instance of the service. 1210 * 1211 * The result of unregistration is reported via dprintk for those who want 1212 * verification of the result, but is otherwise not important. 1213 */ 1214 static void svc_unregister(const struct svc_serv *serv, struct net *net) 1215 { 1216 struct sighand_struct *sighand; 1217 unsigned long flags; 1218 unsigned int p, i; 1219 1220 clear_thread_flag(TIF_SIGPENDING); 1221 1222 for (p = 0; p < serv->sv_nprogs; p++) { 1223 struct svc_program *progp = &serv->sv_programs[p]; 1224 1225 for (i = 0; i < progp->pg_nvers; i++) { 1226 if (progp->pg_vers[i] == NULL) 1227 continue; 1228 if (progp->pg_vers[i]->vs_hidden) 1229 continue; 1230 __svc_unregister(net, progp->pg_prog, i, progp->pg_name); 1231 } 1232 } 1233 1234 rcu_read_lock(); 1235 sighand = rcu_dereference(current->sighand); 1236 spin_lock_irqsave(&sighand->siglock, flags); 1237 recalc_sigpending(); 1238 spin_unlock_irqrestore(&sighand->siglock, flags); 1239 rcu_read_unlock(); 1240 } 1241 1242 /* 1243 * dprintk the given error with the address of the client that caused it. 1244 */ 1245 #if IS_ENABLED(CONFIG_SUNRPC_DEBUG) 1246 static __printf(2, 3) 1247 void svc_printk(struct svc_rqst *rqstp, const char *fmt, ...) 1248 { 1249 struct va_format vaf; 1250 va_list args; 1251 char buf[RPC_MAX_ADDRBUFLEN]; 1252 1253 va_start(args, fmt); 1254 1255 vaf.fmt = fmt; 1256 vaf.va = &args; 1257 1258 dprintk("svc: %s: %pV", svc_print_addr(rqstp, buf, sizeof(buf)), &vaf); 1259 1260 va_end(args); 1261 } 1262 #else 1263 static __printf(2,3) void svc_printk(struct svc_rqst *rqstp, const char *fmt, ...) {} 1264 #endif 1265 1266 __be32 1267 svc_generic_init_request(struct svc_rqst *rqstp, 1268 const struct svc_program *progp, 1269 struct svc_process_info *ret) 1270 { 1271 const struct svc_version *versp = NULL; /* compiler food */ 1272 const struct svc_procedure *procp = NULL; 1273 1274 if (rqstp->rq_vers >= progp->pg_nvers ) 1275 goto err_bad_vers; 1276 versp = progp->pg_vers[rqstp->rq_vers]; 1277 if (!versp) 1278 goto err_bad_vers; 1279 1280 /* 1281 * Some protocol versions (namely NFSv4) require some form of 1282 * congestion control. (See RFC 7530 section 3.1 paragraph 2) 1283 * In other words, UDP is not allowed. We mark those when setting 1284 * up the svc_xprt, and verify that here. 1285 * 1286 * The spec is not very clear about what error should be returned 1287 * when someone tries to access a server that is listening on UDP 1288 * for lower versions. RPC_PROG_MISMATCH seems to be the closest 1289 * fit. 1290 */ 1291 if (versp->vs_need_cong_ctrl && rqstp->rq_xprt && 1292 !test_bit(XPT_CONG_CTRL, &rqstp->rq_xprt->xpt_flags)) 1293 goto err_bad_vers; 1294 1295 if (rqstp->rq_proc >= versp->vs_nproc) 1296 goto err_bad_proc; 1297 rqstp->rq_procinfo = procp = &versp->vs_proc[rqstp->rq_proc]; 1298 1299 /* Initialize storage for argp and resp */ 1300 memset(rqstp->rq_argp, 0, procp->pc_argzero); 1301 memset(rqstp->rq_resp, 0, procp->pc_ressize); 1302 1303 /* Bump per-procedure stats counter */ 1304 this_cpu_inc(versp->vs_count[rqstp->rq_proc]); 1305 1306 ret->dispatch = versp->vs_dispatch; 1307 return rpc_success; 1308 err_bad_vers: 1309 ret->mismatch.lovers = progp->pg_lovers; 1310 ret->mismatch.hivers = progp->pg_hivers; 1311 return rpc_prog_mismatch; 1312 err_bad_proc: 1313 return rpc_proc_unavail; 1314 } 1315 EXPORT_SYMBOL_GPL(svc_generic_init_request); 1316 1317 /* 1318 * Common routine for processing the RPC request. 1319 */ 1320 static int 1321 svc_process_common(struct svc_rqst *rqstp) 1322 { 1323 struct xdr_stream *xdr = &rqstp->rq_res_stream; 1324 struct svc_program *progp; 1325 const struct svc_procedure *procp = NULL; 1326 struct svc_serv *serv = rqstp->rq_server; 1327 struct svc_process_info process; 1328 enum svc_auth_status auth_res; 1329 unsigned int aoffset; 1330 int pr, rc; 1331 __be32 *p; 1332 1333 /* Will be turned off only when NFSv4 Sessions are used */ 1334 set_bit(RQ_USEDEFERRAL, &rqstp->rq_flags); 1335 clear_bit(RQ_DROPME, &rqstp->rq_flags); 1336 1337 /* Construct the first words of the reply: */ 1338 svcxdr_init_encode(rqstp); 1339 xdr_stream_encode_be32(xdr, rqstp->rq_xid); 1340 xdr_stream_encode_be32(xdr, rpc_reply); 1341 1342 p = xdr_inline_decode(&rqstp->rq_arg_stream, XDR_UNIT * 4); 1343 if (unlikely(!p)) 1344 goto err_short_len; 1345 if (*p++ != cpu_to_be32(RPC_VERSION)) 1346 goto err_bad_rpc; 1347 1348 xdr_stream_encode_be32(xdr, rpc_msg_accepted); 1349 1350 rqstp->rq_prog = be32_to_cpup(p++); 1351 rqstp->rq_vers = be32_to_cpup(p++); 1352 rqstp->rq_proc = be32_to_cpup(p); 1353 1354 for (pr = 0; pr < serv->sv_nprogs; pr++) { 1355 progp = &serv->sv_programs[pr]; 1356 1357 if (rqstp->rq_prog == progp->pg_prog) 1358 break; 1359 } 1360 1361 /* 1362 * Decode auth data, and add verifier to reply buffer. 1363 * We do this before anything else in order to get a decent 1364 * auth verifier. 1365 */ 1366 auth_res = svc_authenticate(rqstp); 1367 /* Also give the program a chance to reject this call: */ 1368 if (auth_res == SVC_OK && progp) 1369 auth_res = progp->pg_authenticate(rqstp); 1370 trace_svc_authenticate(rqstp, auth_res); 1371 switch (auth_res) { 1372 case SVC_OK: 1373 break; 1374 case SVC_GARBAGE: 1375 goto err_garbage_args; 1376 case SVC_SYSERR: 1377 goto err_system_err; 1378 case SVC_DENIED: 1379 goto err_bad_auth; 1380 case SVC_CLOSE: 1381 goto close; 1382 case SVC_DROP: 1383 goto dropit; 1384 case SVC_COMPLETE: 1385 goto sendit; 1386 default: 1387 pr_warn_once("Unexpected svc_auth_status (%d)\n", auth_res); 1388 goto err_system_err; 1389 } 1390 1391 if (progp == NULL) 1392 goto err_bad_prog; 1393 1394 switch (progp->pg_init_request(rqstp, progp, &process)) { 1395 case rpc_success: 1396 break; 1397 case rpc_prog_unavail: 1398 goto err_bad_prog; 1399 case rpc_prog_mismatch: 1400 goto err_bad_vers; 1401 case rpc_proc_unavail: 1402 goto err_bad_proc; 1403 } 1404 1405 procp = rqstp->rq_procinfo; 1406 /* Should this check go into the dispatcher? */ 1407 if (!procp || !procp->pc_func) 1408 goto err_bad_proc; 1409 1410 /* Syntactic check complete */ 1411 if (serv->sv_stats) 1412 serv->sv_stats->rpccnt++; 1413 trace_svc_process(rqstp, progp->pg_name); 1414 1415 aoffset = xdr_stream_pos(xdr); 1416 1417 /* un-reserve some of the out-queue now that we have a 1418 * better idea of reply size 1419 */ 1420 if (procp->pc_xdrressize) 1421 svc_reserve_auth(rqstp, procp->pc_xdrressize<<2); 1422 1423 /* Call the function that processes the request. */ 1424 rc = process.dispatch(rqstp); 1425 if (procp->pc_release) 1426 procp->pc_release(rqstp); 1427 xdr_finish_decode(xdr); 1428 1429 if (!rc) 1430 goto dropit; 1431 if (rqstp->rq_auth_stat != rpc_auth_ok) 1432 goto err_bad_auth; 1433 1434 if (*rqstp->rq_accept_statp != rpc_success) 1435 xdr_truncate_encode(xdr, aoffset); 1436 1437 if (procp->pc_encode == NULL) 1438 goto dropit; 1439 1440 sendit: 1441 if (svc_authorise(rqstp)) 1442 goto close_xprt; 1443 return 1; /* Caller can now send it */ 1444 1445 dropit: 1446 svc_authorise(rqstp); /* doesn't hurt to call this twice */ 1447 dprintk("svc: svc_process dropit\n"); 1448 return 0; 1449 1450 close: 1451 svc_authorise(rqstp); 1452 close_xprt: 1453 if (rqstp->rq_xprt && test_bit(XPT_TEMP, &rqstp->rq_xprt->xpt_flags)) 1454 svc_xprt_close(rqstp->rq_xprt); 1455 dprintk("svc: svc_process close\n"); 1456 return 0; 1457 1458 err_short_len: 1459 svc_printk(rqstp, "short len %u, dropping request\n", 1460 rqstp->rq_arg.len); 1461 goto close_xprt; 1462 1463 err_bad_rpc: 1464 if (serv->sv_stats) 1465 serv->sv_stats->rpcbadfmt++; 1466 xdr_stream_encode_u32(xdr, RPC_MSG_DENIED); 1467 xdr_stream_encode_u32(xdr, RPC_MISMATCH); 1468 /* Only RPCv2 supported */ 1469 xdr_stream_encode_u32(xdr, RPC_VERSION); 1470 xdr_stream_encode_u32(xdr, RPC_VERSION); 1471 return 1; /* don't wrap */ 1472 1473 err_bad_auth: 1474 dprintk("svc: authentication failed (%d)\n", 1475 be32_to_cpu(rqstp->rq_auth_stat)); 1476 if (serv->sv_stats) 1477 serv->sv_stats->rpcbadauth++; 1478 /* Restore write pointer to location of reply status: */ 1479 xdr_truncate_encode(xdr, XDR_UNIT * 2); 1480 xdr_stream_encode_u32(xdr, RPC_MSG_DENIED); 1481 xdr_stream_encode_u32(xdr, RPC_AUTH_ERROR); 1482 xdr_stream_encode_be32(xdr, rqstp->rq_auth_stat); 1483 goto sendit; 1484 1485 err_bad_prog: 1486 dprintk("svc: unknown program %d\n", rqstp->rq_prog); 1487 if (serv->sv_stats) 1488 serv->sv_stats->rpcbadfmt++; 1489 *rqstp->rq_accept_statp = rpc_prog_unavail; 1490 goto sendit; 1491 1492 err_bad_vers: 1493 svc_printk(rqstp, "unknown version (%d for prog %d, %s)\n", 1494 rqstp->rq_vers, rqstp->rq_prog, progp->pg_name); 1495 1496 if (serv->sv_stats) 1497 serv->sv_stats->rpcbadfmt++; 1498 *rqstp->rq_accept_statp = rpc_prog_mismatch; 1499 1500 /* 1501 * svc_authenticate() has already added the verifier and 1502 * advanced the stream just past rq_accept_statp. 1503 */ 1504 xdr_stream_encode_u32(xdr, process.mismatch.lovers); 1505 xdr_stream_encode_u32(xdr, process.mismatch.hivers); 1506 goto sendit; 1507 1508 err_bad_proc: 1509 svc_printk(rqstp, "unknown procedure (%d)\n", rqstp->rq_proc); 1510 1511 if (serv->sv_stats) 1512 serv->sv_stats->rpcbadfmt++; 1513 *rqstp->rq_accept_statp = rpc_proc_unavail; 1514 goto sendit; 1515 1516 err_garbage_args: 1517 svc_printk(rqstp, "failed to decode RPC header\n"); 1518 1519 if (serv->sv_stats) 1520 serv->sv_stats->rpcbadfmt++; 1521 *rqstp->rq_accept_statp = rpc_garbage_args; 1522 goto sendit; 1523 1524 err_system_err: 1525 if (serv->sv_stats) 1526 serv->sv_stats->rpcbadfmt++; 1527 *rqstp->rq_accept_statp = rpc_system_err; 1528 goto sendit; 1529 } 1530 1531 /* 1532 * Drop request 1533 */ 1534 static void svc_drop(struct svc_rqst *rqstp) 1535 { 1536 trace_svc_drop(rqstp); 1537 } 1538 1539 /** 1540 * svc_process - Execute one RPC transaction 1541 * @rqstp: RPC transaction context 1542 * 1543 */ 1544 void svc_process(struct svc_rqst *rqstp) 1545 { 1546 struct kvec *resv = &rqstp->rq_res.head[0]; 1547 __be32 *p; 1548 1549 #if IS_ENABLED(CONFIG_FAIL_SUNRPC) 1550 if (!fail_sunrpc.ignore_server_disconnect && 1551 should_fail(&fail_sunrpc.attr, 1)) 1552 svc_xprt_deferred_close(rqstp->rq_xprt); 1553 #endif 1554 1555 /* 1556 * Setup response xdr_buf. 1557 * Initially it has just one page 1558 */ 1559 rqstp->rq_next_page = &rqstp->rq_respages[1]; 1560 resv->iov_base = page_address(rqstp->rq_respages[0]); 1561 resv->iov_len = 0; 1562 rqstp->rq_res.pages = rqstp->rq_next_page; 1563 rqstp->rq_res.len = 0; 1564 rqstp->rq_res.page_base = 0; 1565 rqstp->rq_res.page_len = 0; 1566 rqstp->rq_res.buflen = PAGE_SIZE; 1567 rqstp->rq_res.tail[0].iov_base = NULL; 1568 rqstp->rq_res.tail[0].iov_len = 0; 1569 1570 svcxdr_init_decode(rqstp); 1571 p = xdr_inline_decode(&rqstp->rq_arg_stream, XDR_UNIT * 2); 1572 if (unlikely(!p)) 1573 goto out_drop; 1574 rqstp->rq_xid = *p++; 1575 if (unlikely(*p != rpc_call)) 1576 goto out_baddir; 1577 1578 if (!svc_process_common(rqstp)) 1579 goto out_drop; 1580 svc_send(rqstp); 1581 return; 1582 1583 out_baddir: 1584 svc_printk(rqstp, "bad direction 0x%08x, dropping request\n", 1585 be32_to_cpu(*p)); 1586 if (rqstp->rq_server->sv_stats) 1587 rqstp->rq_server->sv_stats->rpcbadfmt++; 1588 out_drop: 1589 svc_drop(rqstp); 1590 } 1591 1592 #if defined(CONFIG_SUNRPC_BACKCHANNEL) 1593 /** 1594 * svc_process_bc - process a reverse-direction RPC request 1595 * @req: RPC request to be used for client-side processing 1596 * @rqstp: server-side execution context 1597 * 1598 */ 1599 void svc_process_bc(struct rpc_rqst *req, struct svc_rqst *rqstp) 1600 { 1601 struct rpc_timeout timeout = { 1602 .to_increment = 0, 1603 }; 1604 struct rpc_task *task; 1605 int proc_error; 1606 1607 /* Build the svc_rqst used by the common processing routine */ 1608 rqstp->rq_xid = req->rq_xid; 1609 rqstp->rq_prot = req->rq_xprt->prot; 1610 rqstp->rq_bc_net = req->rq_xprt->xprt_net; 1611 1612 rqstp->rq_addrlen = sizeof(req->rq_xprt->addr); 1613 memcpy(&rqstp->rq_addr, &req->rq_xprt->addr, rqstp->rq_addrlen); 1614 memcpy(&rqstp->rq_arg, &req->rq_rcv_buf, sizeof(rqstp->rq_arg)); 1615 memcpy(&rqstp->rq_res, &req->rq_snd_buf, sizeof(rqstp->rq_res)); 1616 1617 /* Adjust the argument buffer length */ 1618 rqstp->rq_arg.len = req->rq_private_buf.len; 1619 if (rqstp->rq_arg.len <= rqstp->rq_arg.head[0].iov_len) { 1620 rqstp->rq_arg.head[0].iov_len = rqstp->rq_arg.len; 1621 rqstp->rq_arg.page_len = 0; 1622 } else if (rqstp->rq_arg.len <= rqstp->rq_arg.head[0].iov_len + 1623 rqstp->rq_arg.page_len) 1624 rqstp->rq_arg.page_len = rqstp->rq_arg.len - 1625 rqstp->rq_arg.head[0].iov_len; 1626 else 1627 rqstp->rq_arg.len = rqstp->rq_arg.head[0].iov_len + 1628 rqstp->rq_arg.page_len; 1629 1630 /* Reset the response buffer */ 1631 rqstp->rq_res.head[0].iov_len = 0; 1632 1633 /* 1634 * Skip the XID and calldir fields because they've already 1635 * been processed by the caller. 1636 */ 1637 svcxdr_init_decode(rqstp); 1638 if (!xdr_inline_decode(&rqstp->rq_arg_stream, XDR_UNIT * 2)) 1639 return; 1640 1641 /* Parse and execute the bc call */ 1642 proc_error = svc_process_common(rqstp); 1643 1644 atomic_dec(&req->rq_xprt->bc_slot_count); 1645 if (!proc_error) { 1646 /* Processing error: drop the request */ 1647 xprt_free_bc_request(req); 1648 return; 1649 } 1650 /* Finally, send the reply synchronously */ 1651 if (rqstp->bc_to_initval > 0) { 1652 timeout.to_initval = rqstp->bc_to_initval; 1653 timeout.to_retries = rqstp->bc_to_retries; 1654 } else { 1655 timeout.to_initval = req->rq_xprt->timeout->to_initval; 1656 timeout.to_retries = req->rq_xprt->timeout->to_retries; 1657 } 1658 timeout.to_maxval = timeout.to_initval; 1659 memcpy(&req->rq_snd_buf, &rqstp->rq_res, sizeof(req->rq_snd_buf)); 1660 task = rpc_run_bc_task(req, &timeout); 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 */ 1677 u32 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 } 1685 EXPORT_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 */ 1694 const 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 */ 1711 int 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 } 1717 EXPORT_SYMBOL_GPL(svc_encode_result_payload); 1718 1719 /** 1720 * svc_fill_write_vector - Construct data argument for VFS write call 1721 * @rqstp: svc_rqst to operate on 1722 * @payload: xdr_buf containing only the write data payload 1723 * 1724 * Fills in rqstp::rq_vec, and returns the number of elements. 1725 */ 1726 unsigned int svc_fill_write_vector(struct svc_rqst *rqstp, 1727 struct xdr_buf *payload) 1728 { 1729 struct page **pages = payload->pages; 1730 struct kvec *first = payload->head; 1731 struct kvec *vec = rqstp->rq_vec; 1732 size_t total = payload->len; 1733 unsigned int i; 1734 1735 /* Some types of transport can present the write payload 1736 * entirely in rq_arg.pages. In this case, @first is empty. 1737 */ 1738 i = 0; 1739 if (first->iov_len) { 1740 vec[i].iov_base = first->iov_base; 1741 vec[i].iov_len = min_t(size_t, total, first->iov_len); 1742 total -= vec[i].iov_len; 1743 ++i; 1744 } 1745 1746 while (total) { 1747 vec[i].iov_base = page_address(*pages); 1748 vec[i].iov_len = min_t(size_t, total, PAGE_SIZE); 1749 total -= vec[i].iov_len; 1750 ++i; 1751 ++pages; 1752 } 1753 1754 WARN_ON_ONCE(i > ARRAY_SIZE(rqstp->rq_vec)); 1755 return i; 1756 } 1757 EXPORT_SYMBOL_GPL(svc_fill_write_vector); 1758 1759 /** 1760 * svc_fill_symlink_pathname - Construct pathname argument for VFS symlink call 1761 * @rqstp: svc_rqst to operate on 1762 * @first: buffer containing first section of pathname 1763 * @p: buffer containing remaining section of pathname 1764 * @total: total length of the pathname argument 1765 * 1766 * The VFS symlink API demands a NUL-terminated pathname in mapped memory. 1767 * Returns pointer to a NUL-terminated string, or an ERR_PTR. Caller must free 1768 * the returned string. 1769 */ 1770 char *svc_fill_symlink_pathname(struct svc_rqst *rqstp, struct kvec *first, 1771 void *p, size_t total) 1772 { 1773 size_t len, remaining; 1774 char *result, *dst; 1775 1776 result = kmalloc(total + 1, GFP_KERNEL); 1777 if (!result) 1778 return ERR_PTR(-ESERVERFAULT); 1779 1780 dst = result; 1781 remaining = total; 1782 1783 len = min_t(size_t, total, first->iov_len); 1784 if (len) { 1785 memcpy(dst, first->iov_base, len); 1786 dst += len; 1787 remaining -= len; 1788 } 1789 1790 if (remaining) { 1791 len = min_t(size_t, remaining, PAGE_SIZE); 1792 memcpy(dst, p, len); 1793 dst += len; 1794 } 1795 1796 *dst = '\0'; 1797 1798 /* Sanity check: Linux doesn't allow the pathname argument to 1799 * contain a NUL byte. 1800 */ 1801 if (strlen(result) != total) { 1802 kfree(result); 1803 return ERR_PTR(-EINVAL); 1804 } 1805 return result; 1806 } 1807 EXPORT_SYMBOL_GPL(svc_fill_symlink_pathname); 1808