1 /* 2 * linux/net/sunrpc/svc.c 3 * 4 * High-level RPC service routines 5 * 6 * Copyright (C) 1995, 1996 Olaf Kirch <okir@monad.swb.de> 7 * 8 * Multiple threads pools and NUMAisation 9 * Copyright (c) 2006 Silicon Graphics, Inc. 10 * by Greg Banks <gnb@melbourne.sgi.com> 11 */ 12 13 #include <linux/linkage.h> 14 #include <linux/sched.h> 15 #include <linux/errno.h> 16 #include <linux/net.h> 17 #include <linux/in.h> 18 #include <linux/mm.h> 19 #include <linux/interrupt.h> 20 #include <linux/module.h> 21 #include <linux/kthread.h> 22 #include <linux/slab.h> 23 #include <linux/nsproxy.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 #define RPCDBG_FACILITY RPCDBG_SVCDSP 33 34 static void svc_unregister(const struct svc_serv *serv, struct net *net); 35 36 #define svc_serv_is_pooled(serv) ((serv)->sv_function) 37 38 /* 39 * Mode for mapping cpus to pools. 40 */ 41 enum { 42 SVC_POOL_AUTO = -1, /* choose one of the others */ 43 SVC_POOL_GLOBAL, /* no mapping, just a single global pool 44 * (legacy & UP mode) */ 45 SVC_POOL_PERCPU, /* one pool per cpu */ 46 SVC_POOL_PERNODE /* one pool per numa node */ 47 }; 48 #define SVC_POOL_DEFAULT SVC_POOL_GLOBAL 49 50 /* 51 * Structure for mapping cpus to pools and vice versa. 52 * Setup once during sunrpc initialisation. 53 */ 54 static struct svc_pool_map { 55 int count; /* How many svc_servs use us */ 56 int mode; /* Note: int not enum to avoid 57 * warnings about "enumeration value 58 * not handled in switch" */ 59 unsigned int npools; 60 unsigned int *pool_to; /* maps pool id to cpu or node */ 61 unsigned int *to_pool; /* maps cpu or node to pool id */ 62 } svc_pool_map = { 63 .count = 0, 64 .mode = SVC_POOL_DEFAULT 65 }; 66 static DEFINE_MUTEX(svc_pool_map_mutex);/* protects svc_pool_map.count only */ 67 68 static int 69 param_set_pool_mode(const char *val, struct kernel_param *kp) 70 { 71 int *ip = (int *)kp->arg; 72 struct svc_pool_map *m = &svc_pool_map; 73 int err; 74 75 mutex_lock(&svc_pool_map_mutex); 76 77 err = -EBUSY; 78 if (m->count) 79 goto out; 80 81 err = 0; 82 if (!strncmp(val, "auto", 4)) 83 *ip = SVC_POOL_AUTO; 84 else if (!strncmp(val, "global", 6)) 85 *ip = SVC_POOL_GLOBAL; 86 else if (!strncmp(val, "percpu", 6)) 87 *ip = SVC_POOL_PERCPU; 88 else if (!strncmp(val, "pernode", 7)) 89 *ip = SVC_POOL_PERNODE; 90 else 91 err = -EINVAL; 92 93 out: 94 mutex_unlock(&svc_pool_map_mutex); 95 return err; 96 } 97 98 static int 99 param_get_pool_mode(char *buf, struct kernel_param *kp) 100 { 101 int *ip = (int *)kp->arg; 102 103 switch (*ip) 104 { 105 case SVC_POOL_AUTO: 106 return strlcpy(buf, "auto", 20); 107 case SVC_POOL_GLOBAL: 108 return strlcpy(buf, "global", 20); 109 case SVC_POOL_PERCPU: 110 return strlcpy(buf, "percpu", 20); 111 case SVC_POOL_PERNODE: 112 return strlcpy(buf, "pernode", 20); 113 default: 114 return sprintf(buf, "%d", *ip); 115 } 116 } 117 118 module_param_call(pool_mode, param_set_pool_mode, param_get_pool_mode, 119 &svc_pool_map.mode, 0644); 120 121 /* 122 * Detect best pool mapping mode heuristically, 123 * according to the machine's topology. 124 */ 125 static int 126 svc_pool_map_choose_mode(void) 127 { 128 unsigned int node; 129 130 if (nr_online_nodes > 1) { 131 /* 132 * Actually have multiple NUMA nodes, 133 * so split pools on NUMA node boundaries 134 */ 135 return SVC_POOL_PERNODE; 136 } 137 138 node = first_online_node; 139 if (nr_cpus_node(node) > 2) { 140 /* 141 * Non-trivial SMP, or CONFIG_NUMA on 142 * non-NUMA hardware, e.g. with a generic 143 * x86_64 kernel on Xeons. In this case we 144 * want to divide the pools on cpu boundaries. 145 */ 146 return SVC_POOL_PERCPU; 147 } 148 149 /* default: one global pool */ 150 return SVC_POOL_GLOBAL; 151 } 152 153 /* 154 * Allocate the to_pool[] and pool_to[] arrays. 155 * Returns 0 on success or an errno. 156 */ 157 static int 158 svc_pool_map_alloc_arrays(struct svc_pool_map *m, unsigned int maxpools) 159 { 160 m->to_pool = kcalloc(maxpools, sizeof(unsigned int), GFP_KERNEL); 161 if (!m->to_pool) 162 goto fail; 163 m->pool_to = kcalloc(maxpools, sizeof(unsigned int), GFP_KERNEL); 164 if (!m->pool_to) 165 goto fail_free; 166 167 return 0; 168 169 fail_free: 170 kfree(m->to_pool); 171 m->to_pool = NULL; 172 fail: 173 return -ENOMEM; 174 } 175 176 /* 177 * Initialise the pool map for SVC_POOL_PERCPU mode. 178 * Returns number of pools or <0 on error. 179 */ 180 static int 181 svc_pool_map_init_percpu(struct svc_pool_map *m) 182 { 183 unsigned int maxpools = nr_cpu_ids; 184 unsigned int pidx = 0; 185 unsigned int cpu; 186 int err; 187 188 err = svc_pool_map_alloc_arrays(m, maxpools); 189 if (err) 190 return err; 191 192 for_each_online_cpu(cpu) { 193 BUG_ON(pidx > maxpools); 194 m->to_pool[cpu] = pidx; 195 m->pool_to[pidx] = cpu; 196 pidx++; 197 } 198 /* cpus brought online later all get mapped to pool0, sorry */ 199 200 return pidx; 201 }; 202 203 204 /* 205 * Initialise the pool map for SVC_POOL_PERNODE mode. 206 * Returns number of pools or <0 on error. 207 */ 208 static int 209 svc_pool_map_init_pernode(struct svc_pool_map *m) 210 { 211 unsigned int maxpools = nr_node_ids; 212 unsigned int pidx = 0; 213 unsigned int node; 214 int err; 215 216 err = svc_pool_map_alloc_arrays(m, maxpools); 217 if (err) 218 return err; 219 220 for_each_node_with_cpus(node) { 221 /* some architectures (e.g. SN2) have cpuless nodes */ 222 BUG_ON(pidx > maxpools); 223 m->to_pool[node] = pidx; 224 m->pool_to[pidx] = node; 225 pidx++; 226 } 227 /* nodes brought online later all get mapped to pool0, sorry */ 228 229 return pidx; 230 } 231 232 233 /* 234 * Add a reference to the global map of cpus to pools (and 235 * vice versa). Initialise the map if we're the first user. 236 * Returns the number of pools. 237 */ 238 static unsigned int 239 svc_pool_map_get(void) 240 { 241 struct svc_pool_map *m = &svc_pool_map; 242 int npools = -1; 243 244 mutex_lock(&svc_pool_map_mutex); 245 246 if (m->count++) { 247 mutex_unlock(&svc_pool_map_mutex); 248 return m->npools; 249 } 250 251 if (m->mode == SVC_POOL_AUTO) 252 m->mode = svc_pool_map_choose_mode(); 253 254 switch (m->mode) { 255 case SVC_POOL_PERCPU: 256 npools = svc_pool_map_init_percpu(m); 257 break; 258 case SVC_POOL_PERNODE: 259 npools = svc_pool_map_init_pernode(m); 260 break; 261 } 262 263 if (npools < 0) { 264 /* default, or memory allocation failure */ 265 npools = 1; 266 m->mode = SVC_POOL_GLOBAL; 267 } 268 m->npools = npools; 269 270 mutex_unlock(&svc_pool_map_mutex); 271 return m->npools; 272 } 273 274 275 /* 276 * Drop a reference to the global map of cpus to pools. 277 * When the last reference is dropped, the map data is 278 * freed; this allows the sysadmin to change the pool 279 * mode using the pool_mode module option without 280 * rebooting or re-loading sunrpc.ko. 281 */ 282 static void 283 svc_pool_map_put(void) 284 { 285 struct svc_pool_map *m = &svc_pool_map; 286 287 mutex_lock(&svc_pool_map_mutex); 288 289 if (!--m->count) { 290 kfree(m->to_pool); 291 m->to_pool = NULL; 292 kfree(m->pool_to); 293 m->pool_to = NULL; 294 m->npools = 0; 295 } 296 297 mutex_unlock(&svc_pool_map_mutex); 298 } 299 300 301 static int svc_pool_map_get_node(unsigned int pidx) 302 { 303 const struct svc_pool_map *m = &svc_pool_map; 304 305 if (m->count) { 306 if (m->mode == SVC_POOL_PERCPU) 307 return cpu_to_node(m->pool_to[pidx]); 308 if (m->mode == SVC_POOL_PERNODE) 309 return m->pool_to[pidx]; 310 } 311 return NUMA_NO_NODE; 312 } 313 /* 314 * Set the given thread's cpus_allowed mask so that it 315 * will only run on cpus in the given pool. 316 */ 317 static inline void 318 svc_pool_map_set_cpumask(struct task_struct *task, unsigned int pidx) 319 { 320 struct svc_pool_map *m = &svc_pool_map; 321 unsigned int node = m->pool_to[pidx]; 322 323 /* 324 * The caller checks for sv_nrpools > 1, which 325 * implies that we've been initialized. 326 */ 327 WARN_ON_ONCE(m->count == 0); 328 if (m->count == 0) 329 return; 330 331 switch (m->mode) { 332 case SVC_POOL_PERCPU: 333 { 334 set_cpus_allowed_ptr(task, cpumask_of(node)); 335 break; 336 } 337 case SVC_POOL_PERNODE: 338 { 339 set_cpus_allowed_ptr(task, cpumask_of_node(node)); 340 break; 341 } 342 } 343 } 344 345 /* 346 * Use the mapping mode to choose a pool for a given CPU. 347 * Used when enqueueing an incoming RPC. Always returns 348 * a non-NULL pool pointer. 349 */ 350 struct svc_pool * 351 svc_pool_for_cpu(struct svc_serv *serv, int cpu) 352 { 353 struct svc_pool_map *m = &svc_pool_map; 354 unsigned int pidx = 0; 355 356 /* 357 * An uninitialised map happens in a pure client when 358 * lockd is brought up, so silently treat it the 359 * same as SVC_POOL_GLOBAL. 360 */ 361 if (svc_serv_is_pooled(serv)) { 362 switch (m->mode) { 363 case SVC_POOL_PERCPU: 364 pidx = m->to_pool[cpu]; 365 break; 366 case SVC_POOL_PERNODE: 367 pidx = m->to_pool[cpu_to_node(cpu)]; 368 break; 369 } 370 } 371 return &serv->sv_pools[pidx % serv->sv_nrpools]; 372 } 373 374 int svc_rpcb_setup(struct svc_serv *serv, struct net *net) 375 { 376 int err; 377 378 err = rpcb_create_local(net); 379 if (err) 380 return err; 381 382 /* Remove any stale portmap registrations */ 383 svc_unregister(serv, net); 384 return 0; 385 } 386 EXPORT_SYMBOL_GPL(svc_rpcb_setup); 387 388 void svc_rpcb_cleanup(struct svc_serv *serv, struct net *net) 389 { 390 svc_unregister(serv, net); 391 rpcb_put_local(net); 392 } 393 EXPORT_SYMBOL_GPL(svc_rpcb_cleanup); 394 395 static int svc_uses_rpcbind(struct svc_serv *serv) 396 { 397 struct svc_program *progp; 398 unsigned int i; 399 400 for (progp = serv->sv_program; progp; progp = progp->pg_next) { 401 for (i = 0; i < progp->pg_nvers; i++) { 402 if (progp->pg_vers[i] == NULL) 403 continue; 404 if (progp->pg_vers[i]->vs_hidden == 0) 405 return 1; 406 } 407 } 408 409 return 0; 410 } 411 412 int svc_bind(struct svc_serv *serv, struct net *net) 413 { 414 if (!svc_uses_rpcbind(serv)) 415 return 0; 416 return svc_rpcb_setup(serv, net); 417 } 418 EXPORT_SYMBOL_GPL(svc_bind); 419 420 /* 421 * Create an RPC service 422 */ 423 static struct svc_serv * 424 __svc_create(struct svc_program *prog, unsigned int bufsize, int npools, 425 void (*shutdown)(struct svc_serv *serv, struct net *net)) 426 { 427 struct svc_serv *serv; 428 unsigned int vers; 429 unsigned int xdrsize; 430 unsigned int i; 431 432 if (!(serv = kzalloc(sizeof(*serv), GFP_KERNEL))) 433 return NULL; 434 serv->sv_name = prog->pg_name; 435 serv->sv_program = prog; 436 serv->sv_nrthreads = 1; 437 serv->sv_stats = prog->pg_stats; 438 if (bufsize > RPCSVC_MAXPAYLOAD) 439 bufsize = RPCSVC_MAXPAYLOAD; 440 serv->sv_max_payload = bufsize? bufsize : 4096; 441 serv->sv_max_mesg = roundup(serv->sv_max_payload + PAGE_SIZE, PAGE_SIZE); 442 serv->sv_shutdown = shutdown; 443 xdrsize = 0; 444 while (prog) { 445 prog->pg_lovers = prog->pg_nvers-1; 446 for (vers=0; vers<prog->pg_nvers ; vers++) 447 if (prog->pg_vers[vers]) { 448 prog->pg_hivers = vers; 449 if (prog->pg_lovers > vers) 450 prog->pg_lovers = vers; 451 if (prog->pg_vers[vers]->vs_xdrsize > xdrsize) 452 xdrsize = prog->pg_vers[vers]->vs_xdrsize; 453 } 454 prog = prog->pg_next; 455 } 456 serv->sv_xdrsize = xdrsize; 457 INIT_LIST_HEAD(&serv->sv_tempsocks); 458 INIT_LIST_HEAD(&serv->sv_permsocks); 459 init_timer(&serv->sv_temptimer); 460 spin_lock_init(&serv->sv_lock); 461 462 serv->sv_nrpools = npools; 463 serv->sv_pools = 464 kcalloc(serv->sv_nrpools, sizeof(struct svc_pool), 465 GFP_KERNEL); 466 if (!serv->sv_pools) { 467 kfree(serv); 468 return NULL; 469 } 470 471 for (i = 0; i < serv->sv_nrpools; i++) { 472 struct svc_pool *pool = &serv->sv_pools[i]; 473 474 dprintk("svc: initialising pool %u for %s\n", 475 i, serv->sv_name); 476 477 pool->sp_id = i; 478 INIT_LIST_HEAD(&pool->sp_threads); 479 INIT_LIST_HEAD(&pool->sp_sockets); 480 INIT_LIST_HEAD(&pool->sp_all_threads); 481 spin_lock_init(&pool->sp_lock); 482 } 483 484 if (svc_uses_rpcbind(serv) && (!serv->sv_shutdown)) 485 serv->sv_shutdown = svc_rpcb_cleanup; 486 487 return serv; 488 } 489 490 struct svc_serv * 491 svc_create(struct svc_program *prog, unsigned int bufsize, 492 void (*shutdown)(struct svc_serv *serv, struct net *net)) 493 { 494 return __svc_create(prog, bufsize, /*npools*/1, shutdown); 495 } 496 EXPORT_SYMBOL_GPL(svc_create); 497 498 struct svc_serv * 499 svc_create_pooled(struct svc_program *prog, unsigned int bufsize, 500 void (*shutdown)(struct svc_serv *serv, struct net *net), 501 svc_thread_fn func, struct module *mod) 502 { 503 struct svc_serv *serv; 504 unsigned int npools = svc_pool_map_get(); 505 506 serv = __svc_create(prog, bufsize, npools, shutdown); 507 508 if (serv != NULL) { 509 serv->sv_function = func; 510 serv->sv_module = mod; 511 } 512 513 return serv; 514 } 515 EXPORT_SYMBOL_GPL(svc_create_pooled); 516 517 void svc_shutdown_net(struct svc_serv *serv, struct net *net) 518 { 519 /* 520 * The set of xprts (contained in the sv_tempsocks and 521 * sv_permsocks lists) is now constant, since it is modified 522 * only by accepting new sockets (done by service threads in 523 * svc_recv) or aging old ones (done by sv_temptimer), or 524 * configuration changes (excluded by whatever locking the 525 * caller is using--nfsd_mutex in the case of nfsd). So it's 526 * safe to traverse those lists and shut everything down: 527 */ 528 svc_close_net(serv, net); 529 530 if (serv->sv_shutdown) 531 serv->sv_shutdown(serv, net); 532 } 533 EXPORT_SYMBOL_GPL(svc_shutdown_net); 534 535 /* 536 * Destroy an RPC service. Should be called with appropriate locking to 537 * protect the sv_nrthreads, sv_permsocks and sv_tempsocks. 538 */ 539 void 540 svc_destroy(struct svc_serv *serv) 541 { 542 dprintk("svc: svc_destroy(%s, %d)\n", 543 serv->sv_program->pg_name, 544 serv->sv_nrthreads); 545 546 if (serv->sv_nrthreads) { 547 if (--(serv->sv_nrthreads) != 0) { 548 svc_sock_update_bufs(serv); 549 return; 550 } 551 } else 552 printk("svc_destroy: no threads for serv=%p!\n", serv); 553 554 del_timer_sync(&serv->sv_temptimer); 555 556 /* 557 * The last user is gone and thus all sockets have to be destroyed to 558 * the point. Check this. 559 */ 560 BUG_ON(!list_empty(&serv->sv_permsocks)); 561 BUG_ON(!list_empty(&serv->sv_tempsocks)); 562 563 cache_clean_deferred(serv); 564 565 if (svc_serv_is_pooled(serv)) 566 svc_pool_map_put(); 567 568 kfree(serv->sv_pools); 569 kfree(serv); 570 } 571 EXPORT_SYMBOL_GPL(svc_destroy); 572 573 /* 574 * Allocate an RPC server's buffer space. 575 * We allocate pages and place them in rq_argpages. 576 */ 577 static int 578 svc_init_buffer(struct svc_rqst *rqstp, unsigned int size, int node) 579 { 580 unsigned int pages, arghi; 581 582 /* bc_xprt uses fore channel allocated buffers */ 583 if (svc_is_backchannel(rqstp)) 584 return 1; 585 586 pages = size / PAGE_SIZE + 1; /* extra page as we hold both request and reply. 587 * We assume one is at most one page 588 */ 589 arghi = 0; 590 WARN_ON_ONCE(pages > RPCSVC_MAXPAGES); 591 if (pages > RPCSVC_MAXPAGES) 592 pages = RPCSVC_MAXPAGES; 593 while (pages) { 594 struct page *p = alloc_pages_node(node, GFP_KERNEL, 0); 595 if (!p) 596 break; 597 rqstp->rq_pages[arghi++] = p; 598 pages--; 599 } 600 return pages == 0; 601 } 602 603 /* 604 * Release an RPC server buffer 605 */ 606 static void 607 svc_release_buffer(struct svc_rqst *rqstp) 608 { 609 unsigned int i; 610 611 for (i = 0; i < ARRAY_SIZE(rqstp->rq_pages); i++) 612 if (rqstp->rq_pages[i]) 613 put_page(rqstp->rq_pages[i]); 614 } 615 616 struct svc_rqst * 617 svc_prepare_thread(struct svc_serv *serv, struct svc_pool *pool, int node) 618 { 619 struct svc_rqst *rqstp; 620 621 rqstp = kzalloc_node(sizeof(*rqstp), GFP_KERNEL, node); 622 if (!rqstp) 623 goto out_enomem; 624 625 init_waitqueue_head(&rqstp->rq_wait); 626 627 serv->sv_nrthreads++; 628 spin_lock_bh(&pool->sp_lock); 629 pool->sp_nrthreads++; 630 list_add(&rqstp->rq_all, &pool->sp_all_threads); 631 spin_unlock_bh(&pool->sp_lock); 632 rqstp->rq_server = serv; 633 rqstp->rq_pool = pool; 634 635 rqstp->rq_argp = kmalloc_node(serv->sv_xdrsize, GFP_KERNEL, node); 636 if (!rqstp->rq_argp) 637 goto out_thread; 638 639 rqstp->rq_resp = kmalloc_node(serv->sv_xdrsize, GFP_KERNEL, node); 640 if (!rqstp->rq_resp) 641 goto out_thread; 642 643 if (!svc_init_buffer(rqstp, serv->sv_max_mesg, node)) 644 goto out_thread; 645 646 return rqstp; 647 out_thread: 648 svc_exit_thread(rqstp); 649 out_enomem: 650 return ERR_PTR(-ENOMEM); 651 } 652 EXPORT_SYMBOL_GPL(svc_prepare_thread); 653 654 /* 655 * Choose a pool in which to create a new thread, for svc_set_num_threads 656 */ 657 static inline struct svc_pool * 658 choose_pool(struct svc_serv *serv, struct svc_pool *pool, unsigned int *state) 659 { 660 if (pool != NULL) 661 return pool; 662 663 return &serv->sv_pools[(*state)++ % serv->sv_nrpools]; 664 } 665 666 /* 667 * Choose a thread to kill, for svc_set_num_threads 668 */ 669 static inline struct task_struct * 670 choose_victim(struct svc_serv *serv, struct svc_pool *pool, unsigned int *state) 671 { 672 unsigned int i; 673 struct task_struct *task = NULL; 674 675 if (pool != NULL) { 676 spin_lock_bh(&pool->sp_lock); 677 } else { 678 /* choose a pool in round-robin fashion */ 679 for (i = 0; i < serv->sv_nrpools; i++) { 680 pool = &serv->sv_pools[--(*state) % serv->sv_nrpools]; 681 spin_lock_bh(&pool->sp_lock); 682 if (!list_empty(&pool->sp_all_threads)) 683 goto found_pool; 684 spin_unlock_bh(&pool->sp_lock); 685 } 686 return NULL; 687 } 688 689 found_pool: 690 if (!list_empty(&pool->sp_all_threads)) { 691 struct svc_rqst *rqstp; 692 693 /* 694 * Remove from the pool->sp_all_threads list 695 * so we don't try to kill it again. 696 */ 697 rqstp = list_entry(pool->sp_all_threads.next, struct svc_rqst, rq_all); 698 list_del_init(&rqstp->rq_all); 699 task = rqstp->rq_task; 700 } 701 spin_unlock_bh(&pool->sp_lock); 702 703 return task; 704 } 705 706 /* 707 * Create or destroy enough new threads to make the number 708 * of threads the given number. If `pool' is non-NULL, applies 709 * only to threads in that pool, otherwise round-robins between 710 * all pools. Caller must ensure that mutual exclusion between this and 711 * server startup or shutdown. 712 * 713 * Destroying threads relies on the service threads filling in 714 * rqstp->rq_task, which only the nfs ones do. Assumes the serv 715 * has been created using svc_create_pooled(). 716 * 717 * Based on code that used to be in nfsd_svc() but tweaked 718 * to be pool-aware. 719 */ 720 int 721 svc_set_num_threads(struct svc_serv *serv, struct svc_pool *pool, int nrservs) 722 { 723 struct svc_rqst *rqstp; 724 struct task_struct *task; 725 struct svc_pool *chosen_pool; 726 int error = 0; 727 unsigned int state = serv->sv_nrthreads-1; 728 int node; 729 730 if (pool == NULL) { 731 /* The -1 assumes caller has done a svc_get() */ 732 nrservs -= (serv->sv_nrthreads-1); 733 } else { 734 spin_lock_bh(&pool->sp_lock); 735 nrservs -= pool->sp_nrthreads; 736 spin_unlock_bh(&pool->sp_lock); 737 } 738 739 /* create new threads */ 740 while (nrservs > 0) { 741 nrservs--; 742 chosen_pool = choose_pool(serv, pool, &state); 743 744 node = svc_pool_map_get_node(chosen_pool->sp_id); 745 rqstp = svc_prepare_thread(serv, chosen_pool, node); 746 if (IS_ERR(rqstp)) { 747 error = PTR_ERR(rqstp); 748 break; 749 } 750 751 __module_get(serv->sv_module); 752 task = kthread_create_on_node(serv->sv_function, rqstp, 753 node, serv->sv_name); 754 if (IS_ERR(task)) { 755 error = PTR_ERR(task); 756 module_put(serv->sv_module); 757 svc_exit_thread(rqstp); 758 break; 759 } 760 761 rqstp->rq_task = task; 762 if (serv->sv_nrpools > 1) 763 svc_pool_map_set_cpumask(task, chosen_pool->sp_id); 764 765 svc_sock_update_bufs(serv); 766 wake_up_process(task); 767 } 768 /* destroy old threads */ 769 while (nrservs < 0 && 770 (task = choose_victim(serv, pool, &state)) != NULL) { 771 send_sig(SIGINT, task, 1); 772 nrservs++; 773 } 774 775 return error; 776 } 777 EXPORT_SYMBOL_GPL(svc_set_num_threads); 778 779 /* 780 * Called from a server thread as it's exiting. Caller must hold the BKL or 781 * the "service mutex", whichever is appropriate for the service. 782 */ 783 void 784 svc_exit_thread(struct svc_rqst *rqstp) 785 { 786 struct svc_serv *serv = rqstp->rq_server; 787 struct svc_pool *pool = rqstp->rq_pool; 788 789 svc_release_buffer(rqstp); 790 kfree(rqstp->rq_resp); 791 kfree(rqstp->rq_argp); 792 kfree(rqstp->rq_auth_data); 793 794 spin_lock_bh(&pool->sp_lock); 795 pool->sp_nrthreads--; 796 list_del(&rqstp->rq_all); 797 spin_unlock_bh(&pool->sp_lock); 798 799 kfree(rqstp); 800 801 /* Release the server */ 802 if (serv) 803 svc_destroy(serv); 804 } 805 EXPORT_SYMBOL_GPL(svc_exit_thread); 806 807 /* 808 * Register an "inet" protocol family netid with the local 809 * rpcbind daemon via an rpcbind v4 SET request. 810 * 811 * No netconfig infrastructure is available in the kernel, so 812 * we map IP_ protocol numbers to netids by hand. 813 * 814 * Returns zero on success; a negative errno value is returned 815 * if any error occurs. 816 */ 817 static int __svc_rpcb_register4(struct net *net, const u32 program, 818 const u32 version, 819 const unsigned short protocol, 820 const unsigned short port) 821 { 822 const struct sockaddr_in sin = { 823 .sin_family = AF_INET, 824 .sin_addr.s_addr = htonl(INADDR_ANY), 825 .sin_port = htons(port), 826 }; 827 const char *netid; 828 int error; 829 830 switch (protocol) { 831 case IPPROTO_UDP: 832 netid = RPCBIND_NETID_UDP; 833 break; 834 case IPPROTO_TCP: 835 netid = RPCBIND_NETID_TCP; 836 break; 837 default: 838 return -ENOPROTOOPT; 839 } 840 841 error = rpcb_v4_register(net, program, version, 842 (const struct sockaddr *)&sin, netid); 843 844 /* 845 * User space didn't support rpcbind v4, so retry this 846 * registration request with the legacy rpcbind v2 protocol. 847 */ 848 if (error == -EPROTONOSUPPORT) 849 error = rpcb_register(net, program, version, protocol, port); 850 851 return error; 852 } 853 854 #if IS_ENABLED(CONFIG_IPV6) 855 /* 856 * Register an "inet6" protocol family netid with the local 857 * rpcbind daemon via an rpcbind v4 SET request. 858 * 859 * No netconfig infrastructure is available in the kernel, so 860 * we map IP_ protocol numbers to netids by hand. 861 * 862 * Returns zero on success; a negative errno value is returned 863 * if any error occurs. 864 */ 865 static int __svc_rpcb_register6(struct net *net, const u32 program, 866 const u32 version, 867 const unsigned short protocol, 868 const unsigned short port) 869 { 870 const struct sockaddr_in6 sin6 = { 871 .sin6_family = AF_INET6, 872 .sin6_addr = IN6ADDR_ANY_INIT, 873 .sin6_port = htons(port), 874 }; 875 const char *netid; 876 int error; 877 878 switch (protocol) { 879 case IPPROTO_UDP: 880 netid = RPCBIND_NETID_UDP6; 881 break; 882 case IPPROTO_TCP: 883 netid = RPCBIND_NETID_TCP6; 884 break; 885 default: 886 return -ENOPROTOOPT; 887 } 888 889 error = rpcb_v4_register(net, program, version, 890 (const struct sockaddr *)&sin6, netid); 891 892 /* 893 * User space didn't support rpcbind version 4, so we won't 894 * use a PF_INET6 listener. 895 */ 896 if (error == -EPROTONOSUPPORT) 897 error = -EAFNOSUPPORT; 898 899 return error; 900 } 901 #endif /* IS_ENABLED(CONFIG_IPV6) */ 902 903 /* 904 * Register a kernel RPC service via rpcbind version 4. 905 * 906 * Returns zero on success; a negative errno value is returned 907 * if any error occurs. 908 */ 909 static int __svc_register(struct net *net, const char *progname, 910 const u32 program, const u32 version, 911 const int family, 912 const unsigned short protocol, 913 const unsigned short port) 914 { 915 int error = -EAFNOSUPPORT; 916 917 switch (family) { 918 case PF_INET: 919 error = __svc_rpcb_register4(net, program, version, 920 protocol, port); 921 break; 922 #if IS_ENABLED(CONFIG_IPV6) 923 case PF_INET6: 924 error = __svc_rpcb_register6(net, program, version, 925 protocol, port); 926 #endif 927 } 928 929 if (error < 0) 930 printk(KERN_WARNING "svc: failed to register %sv%u RPC " 931 "service (errno %d).\n", progname, version, -error); 932 return error; 933 } 934 935 /** 936 * svc_register - register an RPC service with the local portmapper 937 * @serv: svc_serv struct for the service to register 938 * @net: net namespace for the service to register 939 * @family: protocol family of service's listener socket 940 * @proto: transport protocol number to advertise 941 * @port: port to advertise 942 * 943 * Service is registered for any address in the passed-in protocol family 944 */ 945 int svc_register(const struct svc_serv *serv, struct net *net, 946 const int family, const unsigned short proto, 947 const unsigned short port) 948 { 949 struct svc_program *progp; 950 unsigned int i; 951 int error = 0; 952 953 WARN_ON_ONCE(proto == 0 && port == 0); 954 if (proto == 0 && port == 0) 955 return -EINVAL; 956 957 for (progp = serv->sv_program; progp; progp = progp->pg_next) { 958 for (i = 0; i < progp->pg_nvers; i++) { 959 if (progp->pg_vers[i] == NULL) 960 continue; 961 962 dprintk("svc: svc_register(%sv%d, %s, %u, %u)%s\n", 963 progp->pg_name, 964 i, 965 proto == IPPROTO_UDP? "udp" : "tcp", 966 port, 967 family, 968 progp->pg_vers[i]->vs_hidden? 969 " (but not telling portmap)" : ""); 970 971 if (progp->pg_vers[i]->vs_hidden) 972 continue; 973 974 error = __svc_register(net, progp->pg_name, progp->pg_prog, 975 i, family, proto, port); 976 if (error < 0) 977 break; 978 } 979 } 980 981 return error; 982 } 983 984 /* 985 * If user space is running rpcbind, it should take the v4 UNSET 986 * and clear everything for this [program, version]. If user space 987 * is running portmap, it will reject the v4 UNSET, but won't have 988 * any "inet6" entries anyway. So a PMAP_UNSET should be sufficient 989 * in this case to clear all existing entries for [program, version]. 990 */ 991 static void __svc_unregister(struct net *net, const u32 program, const u32 version, 992 const char *progname) 993 { 994 int error; 995 996 error = rpcb_v4_register(net, program, version, NULL, ""); 997 998 /* 999 * User space didn't support rpcbind v4, so retry this 1000 * request with the legacy rpcbind v2 protocol. 1001 */ 1002 if (error == -EPROTONOSUPPORT) 1003 error = rpcb_register(net, program, version, 0, 0); 1004 1005 dprintk("svc: %s(%sv%u), error %d\n", 1006 __func__, progname, version, error); 1007 } 1008 1009 /* 1010 * All netids, bind addresses and ports registered for [program, version] 1011 * are removed from the local rpcbind database (if the service is not 1012 * hidden) to make way for a new instance of the service. 1013 * 1014 * The result of unregistration is reported via dprintk for those who want 1015 * verification of the result, but is otherwise not important. 1016 */ 1017 static void svc_unregister(const struct svc_serv *serv, struct net *net) 1018 { 1019 struct svc_program *progp; 1020 unsigned long flags; 1021 unsigned int i; 1022 1023 clear_thread_flag(TIF_SIGPENDING); 1024 1025 for (progp = serv->sv_program; progp; progp = progp->pg_next) { 1026 for (i = 0; i < progp->pg_nvers; i++) { 1027 if (progp->pg_vers[i] == NULL) 1028 continue; 1029 if (progp->pg_vers[i]->vs_hidden) 1030 continue; 1031 1032 dprintk("svc: attempting to unregister %sv%u\n", 1033 progp->pg_name, i); 1034 __svc_unregister(net, progp->pg_prog, i, progp->pg_name); 1035 } 1036 } 1037 1038 spin_lock_irqsave(¤t->sighand->siglock, flags); 1039 recalc_sigpending(); 1040 spin_unlock_irqrestore(¤t->sighand->siglock, flags); 1041 } 1042 1043 /* 1044 * Printk the given error with the address of the client that caused it. 1045 */ 1046 static __printf(2, 3) 1047 void svc_printk(struct svc_rqst *rqstp, const char *fmt, ...) 1048 { 1049 struct va_format vaf; 1050 va_list args; 1051 char buf[RPC_MAX_ADDRBUFLEN]; 1052 1053 va_start(args, fmt); 1054 1055 vaf.fmt = fmt; 1056 vaf.va = &args; 1057 1058 net_warn_ratelimited("svc: %s: %pV", 1059 svc_print_addr(rqstp, buf, sizeof(buf)), &vaf); 1060 1061 va_end(args); 1062 } 1063 1064 /* 1065 * Common routine for processing the RPC request. 1066 */ 1067 static int 1068 svc_process_common(struct svc_rqst *rqstp, struct kvec *argv, struct kvec *resv) 1069 { 1070 struct svc_program *progp; 1071 struct svc_version *versp = NULL; /* compiler food */ 1072 struct svc_procedure *procp = NULL; 1073 struct svc_serv *serv = rqstp->rq_server; 1074 kxdrproc_t xdr; 1075 __be32 *statp; 1076 u32 prog, vers, proc; 1077 __be32 auth_stat, rpc_stat; 1078 int auth_res; 1079 __be32 *reply_statp; 1080 1081 rpc_stat = rpc_success; 1082 1083 if (argv->iov_len < 6*4) 1084 goto err_short_len; 1085 1086 /* Will be turned off only in gss privacy case: */ 1087 rqstp->rq_splice_ok = 1; 1088 /* Will be turned off only when NFSv4 Sessions are used */ 1089 rqstp->rq_usedeferral = 1; 1090 rqstp->rq_dropme = false; 1091 1092 /* Setup reply header */ 1093 rqstp->rq_xprt->xpt_ops->xpo_prep_reply_hdr(rqstp); 1094 1095 svc_putu32(resv, rqstp->rq_xid); 1096 1097 vers = svc_getnl(argv); 1098 1099 /* First words of reply: */ 1100 svc_putnl(resv, 1); /* REPLY */ 1101 1102 if (vers != 2) /* RPC version number */ 1103 goto err_bad_rpc; 1104 1105 /* Save position in case we later decide to reject: */ 1106 reply_statp = resv->iov_base + resv->iov_len; 1107 1108 svc_putnl(resv, 0); /* ACCEPT */ 1109 1110 rqstp->rq_prog = prog = svc_getnl(argv); /* program number */ 1111 rqstp->rq_vers = vers = svc_getnl(argv); /* version number */ 1112 rqstp->rq_proc = proc = svc_getnl(argv); /* procedure number */ 1113 1114 progp = serv->sv_program; 1115 1116 for (progp = serv->sv_program; progp; progp = progp->pg_next) 1117 if (prog == progp->pg_prog) 1118 break; 1119 1120 /* 1121 * Decode auth data, and add verifier to reply buffer. 1122 * We do this before anything else in order to get a decent 1123 * auth verifier. 1124 */ 1125 auth_res = svc_authenticate(rqstp, &auth_stat); 1126 /* Also give the program a chance to reject this call: */ 1127 if (auth_res == SVC_OK && progp) { 1128 auth_stat = rpc_autherr_badcred; 1129 auth_res = progp->pg_authenticate(rqstp); 1130 } 1131 switch (auth_res) { 1132 case SVC_OK: 1133 break; 1134 case SVC_GARBAGE: 1135 goto err_garbage; 1136 case SVC_SYSERR: 1137 rpc_stat = rpc_system_err; 1138 goto err_bad; 1139 case SVC_DENIED: 1140 goto err_bad_auth; 1141 case SVC_CLOSE: 1142 if (test_bit(XPT_TEMP, &rqstp->rq_xprt->xpt_flags)) 1143 svc_close_xprt(rqstp->rq_xprt); 1144 case SVC_DROP: 1145 goto dropit; 1146 case SVC_COMPLETE: 1147 goto sendit; 1148 } 1149 1150 if (progp == NULL) 1151 goto err_bad_prog; 1152 1153 if (vers >= progp->pg_nvers || 1154 !(versp = progp->pg_vers[vers])) 1155 goto err_bad_vers; 1156 1157 procp = versp->vs_proc + proc; 1158 if (proc >= versp->vs_nproc || !procp->pc_func) 1159 goto err_bad_proc; 1160 rqstp->rq_procinfo = procp; 1161 1162 /* Syntactic check complete */ 1163 serv->sv_stats->rpccnt++; 1164 1165 /* Build the reply header. */ 1166 statp = resv->iov_base +resv->iov_len; 1167 svc_putnl(resv, RPC_SUCCESS); 1168 1169 /* Bump per-procedure stats counter */ 1170 procp->pc_count++; 1171 1172 /* Initialize storage for argp and resp */ 1173 memset(rqstp->rq_argp, 0, procp->pc_argsize); 1174 memset(rqstp->rq_resp, 0, procp->pc_ressize); 1175 1176 /* un-reserve some of the out-queue now that we have a 1177 * better idea of reply size 1178 */ 1179 if (procp->pc_xdrressize) 1180 svc_reserve_auth(rqstp, procp->pc_xdrressize<<2); 1181 1182 /* Call the function that processes the request. */ 1183 if (!versp->vs_dispatch) { 1184 /* Decode arguments */ 1185 xdr = procp->pc_decode; 1186 if (xdr && !xdr(rqstp, argv->iov_base, rqstp->rq_argp)) 1187 goto err_garbage; 1188 1189 *statp = procp->pc_func(rqstp, rqstp->rq_argp, rqstp->rq_resp); 1190 1191 /* Encode reply */ 1192 if (rqstp->rq_dropme) { 1193 if (procp->pc_release) 1194 procp->pc_release(rqstp, NULL, rqstp->rq_resp); 1195 goto dropit; 1196 } 1197 if (*statp == rpc_success && 1198 (xdr = procp->pc_encode) && 1199 !xdr(rqstp, resv->iov_base+resv->iov_len, rqstp->rq_resp)) { 1200 dprintk("svc: failed to encode reply\n"); 1201 /* serv->sv_stats->rpcsystemerr++; */ 1202 *statp = rpc_system_err; 1203 } 1204 } else { 1205 dprintk("svc: calling dispatcher\n"); 1206 if (!versp->vs_dispatch(rqstp, statp)) { 1207 /* Release reply info */ 1208 if (procp->pc_release) 1209 procp->pc_release(rqstp, NULL, rqstp->rq_resp); 1210 goto dropit; 1211 } 1212 } 1213 1214 /* Check RPC status result */ 1215 if (*statp != rpc_success) 1216 resv->iov_len = ((void*)statp) - resv->iov_base + 4; 1217 1218 /* Release reply info */ 1219 if (procp->pc_release) 1220 procp->pc_release(rqstp, NULL, rqstp->rq_resp); 1221 1222 if (procp->pc_encode == NULL) 1223 goto dropit; 1224 1225 sendit: 1226 if (svc_authorise(rqstp)) 1227 goto dropit; 1228 return 1; /* Caller can now send it */ 1229 1230 dropit: 1231 svc_authorise(rqstp); /* doesn't hurt to call this twice */ 1232 dprintk("svc: svc_process dropit\n"); 1233 return 0; 1234 1235 err_short_len: 1236 svc_printk(rqstp, "short len %Zd, dropping request\n", 1237 argv->iov_len); 1238 1239 goto dropit; /* drop request */ 1240 1241 err_bad_rpc: 1242 serv->sv_stats->rpcbadfmt++; 1243 svc_putnl(resv, 1); /* REJECT */ 1244 svc_putnl(resv, 0); /* RPC_MISMATCH */ 1245 svc_putnl(resv, 2); /* Only RPCv2 supported */ 1246 svc_putnl(resv, 2); 1247 goto sendit; 1248 1249 err_bad_auth: 1250 dprintk("svc: authentication failed (%d)\n", ntohl(auth_stat)); 1251 serv->sv_stats->rpcbadauth++; 1252 /* Restore write pointer to location of accept status: */ 1253 xdr_ressize_check(rqstp, reply_statp); 1254 svc_putnl(resv, 1); /* REJECT */ 1255 svc_putnl(resv, 1); /* AUTH_ERROR */ 1256 svc_putnl(resv, ntohl(auth_stat)); /* status */ 1257 goto sendit; 1258 1259 err_bad_prog: 1260 dprintk("svc: unknown program %d\n", prog); 1261 serv->sv_stats->rpcbadfmt++; 1262 svc_putnl(resv, RPC_PROG_UNAVAIL); 1263 goto sendit; 1264 1265 err_bad_vers: 1266 svc_printk(rqstp, "unknown version (%d for prog %d, %s)\n", 1267 vers, prog, progp->pg_name); 1268 1269 serv->sv_stats->rpcbadfmt++; 1270 svc_putnl(resv, RPC_PROG_MISMATCH); 1271 svc_putnl(resv, progp->pg_lovers); 1272 svc_putnl(resv, progp->pg_hivers); 1273 goto sendit; 1274 1275 err_bad_proc: 1276 svc_printk(rqstp, "unknown procedure (%d)\n", proc); 1277 1278 serv->sv_stats->rpcbadfmt++; 1279 svc_putnl(resv, RPC_PROC_UNAVAIL); 1280 goto sendit; 1281 1282 err_garbage: 1283 svc_printk(rqstp, "failed to decode args\n"); 1284 1285 rpc_stat = rpc_garbage_args; 1286 err_bad: 1287 serv->sv_stats->rpcbadfmt++; 1288 svc_putnl(resv, ntohl(rpc_stat)); 1289 goto sendit; 1290 } 1291 EXPORT_SYMBOL_GPL(svc_process); 1292 1293 /* 1294 * Process the RPC request. 1295 */ 1296 int 1297 svc_process(struct svc_rqst *rqstp) 1298 { 1299 struct kvec *argv = &rqstp->rq_arg.head[0]; 1300 struct kvec *resv = &rqstp->rq_res.head[0]; 1301 struct svc_serv *serv = rqstp->rq_server; 1302 u32 dir; 1303 1304 /* 1305 * Setup response xdr_buf. 1306 * Initially it has just one page 1307 */ 1308 rqstp->rq_resused = 1; 1309 resv->iov_base = page_address(rqstp->rq_respages[0]); 1310 resv->iov_len = 0; 1311 rqstp->rq_res.pages = rqstp->rq_respages + 1; 1312 rqstp->rq_res.len = 0; 1313 rqstp->rq_res.page_base = 0; 1314 rqstp->rq_res.page_len = 0; 1315 rqstp->rq_res.buflen = PAGE_SIZE; 1316 rqstp->rq_res.tail[0].iov_base = NULL; 1317 rqstp->rq_res.tail[0].iov_len = 0; 1318 1319 rqstp->rq_xid = svc_getu32(argv); 1320 1321 dir = svc_getnl(argv); 1322 if (dir != 0) { 1323 /* direction != CALL */ 1324 svc_printk(rqstp, "bad direction %d, dropping request\n", dir); 1325 serv->sv_stats->rpcbadfmt++; 1326 svc_drop(rqstp); 1327 return 0; 1328 } 1329 1330 /* Returns 1 for send, 0 for drop */ 1331 if (svc_process_common(rqstp, argv, resv)) 1332 return svc_send(rqstp); 1333 else { 1334 svc_drop(rqstp); 1335 return 0; 1336 } 1337 } 1338 1339 #if defined(CONFIG_SUNRPC_BACKCHANNEL) 1340 /* 1341 * Process a backchannel RPC request that arrived over an existing 1342 * outbound connection 1343 */ 1344 int 1345 bc_svc_process(struct svc_serv *serv, struct rpc_rqst *req, 1346 struct svc_rqst *rqstp) 1347 { 1348 struct kvec *argv = &rqstp->rq_arg.head[0]; 1349 struct kvec *resv = &rqstp->rq_res.head[0]; 1350 1351 /* Build the svc_rqst used by the common processing routine */ 1352 rqstp->rq_xprt = serv->sv_bc_xprt; 1353 rqstp->rq_xid = req->rq_xid; 1354 rqstp->rq_prot = req->rq_xprt->prot; 1355 rqstp->rq_server = serv; 1356 1357 rqstp->rq_addrlen = sizeof(req->rq_xprt->addr); 1358 memcpy(&rqstp->rq_addr, &req->rq_xprt->addr, rqstp->rq_addrlen); 1359 memcpy(&rqstp->rq_arg, &req->rq_rcv_buf, sizeof(rqstp->rq_arg)); 1360 memcpy(&rqstp->rq_res, &req->rq_snd_buf, sizeof(rqstp->rq_res)); 1361 1362 /* reset result send buffer "put" position */ 1363 resv->iov_len = 0; 1364 1365 if (rqstp->rq_prot != IPPROTO_TCP) { 1366 printk(KERN_ERR "No support for Non-TCP transports!\n"); 1367 BUG(); 1368 } 1369 1370 /* 1371 * Skip the next two words because they've already been 1372 * processed in the trasport 1373 */ 1374 svc_getu32(argv); /* XID */ 1375 svc_getnl(argv); /* CALLDIR */ 1376 1377 /* Returns 1 for send, 0 for drop */ 1378 if (svc_process_common(rqstp, argv, resv)) { 1379 memcpy(&req->rq_snd_buf, &rqstp->rq_res, 1380 sizeof(req->rq_snd_buf)); 1381 return bc_send(req); 1382 } else { 1383 /* drop request */ 1384 xprt_free_bc_request(req); 1385 return 0; 1386 } 1387 } 1388 EXPORT_SYMBOL_GPL(bc_svc_process); 1389 #endif /* CONFIG_SUNRPC_BACKCHANNEL */ 1390 1391 /* 1392 * Return (transport-specific) limit on the rpc payload. 1393 */ 1394 u32 svc_max_payload(const struct svc_rqst *rqstp) 1395 { 1396 u32 max = rqstp->rq_xprt->xpt_class->xcl_max_payload; 1397 1398 if (rqstp->rq_server->sv_max_payload < max) 1399 max = rqstp->rq_server->sv_max_payload; 1400 return max; 1401 } 1402 EXPORT_SYMBOL_GPL(svc_max_payload); 1403