1 /* 2 * linux/net/sunrpc/svc_xprt.c 3 * 4 * Author: Tom Tucker <tom@opengridcomputing.com> 5 */ 6 7 #include <linux/sched.h> 8 #include <linux/errno.h> 9 #include <linux/freezer.h> 10 #include <linux/kthread.h> 11 #include <linux/slab.h> 12 #include <net/sock.h> 13 #include <linux/sunrpc/stats.h> 14 #include <linux/sunrpc/svc_xprt.h> 15 #include <linux/sunrpc/svcsock.h> 16 #include <linux/sunrpc/xprt.h> 17 #include <linux/module.h> 18 19 #define RPCDBG_FACILITY RPCDBG_SVCXPRT 20 21 static struct svc_deferred_req *svc_deferred_dequeue(struct svc_xprt *xprt); 22 static int svc_deferred_recv(struct svc_rqst *rqstp); 23 static struct cache_deferred_req *svc_defer(struct cache_req *req); 24 static void svc_age_temp_xprts(unsigned long closure); 25 static void svc_delete_xprt(struct svc_xprt *xprt); 26 27 /* apparently the "standard" is that clients close 28 * idle connections after 5 minutes, servers after 29 * 6 minutes 30 * http://www.connectathon.org/talks96/nfstcp.pdf 31 */ 32 static int svc_conn_age_period = 6*60; 33 34 /* List of registered transport classes */ 35 static DEFINE_SPINLOCK(svc_xprt_class_lock); 36 static LIST_HEAD(svc_xprt_class_list); 37 38 /* SMP locking strategy: 39 * 40 * svc_pool->sp_lock protects most of the fields of that pool. 41 * svc_serv->sv_lock protects sv_tempsocks, sv_permsocks, sv_tmpcnt. 42 * when both need to be taken (rare), svc_serv->sv_lock is first. 43 * BKL protects svc_serv->sv_nrthread. 44 * svc_sock->sk_lock protects the svc_sock->sk_deferred list 45 * and the ->sk_info_authunix cache. 46 * 47 * The XPT_BUSY bit in xprt->xpt_flags prevents a transport being 48 * enqueued multiply. During normal transport processing this bit 49 * is set by svc_xprt_enqueue and cleared by svc_xprt_received. 50 * Providers should not manipulate this bit directly. 51 * 52 * Some flags can be set to certain values at any time 53 * providing that certain rules are followed: 54 * 55 * XPT_CONN, XPT_DATA: 56 * - Can be set or cleared at any time. 57 * - After a set, svc_xprt_enqueue must be called to enqueue 58 * the transport for processing. 59 * - After a clear, the transport must be read/accepted. 60 * If this succeeds, it must be set again. 61 * XPT_CLOSE: 62 * - Can set at any time. It is never cleared. 63 * XPT_DEAD: 64 * - Can only be set while XPT_BUSY is held which ensures 65 * that no other thread will be using the transport or will 66 * try to set XPT_DEAD. 67 */ 68 69 int svc_reg_xprt_class(struct svc_xprt_class *xcl) 70 { 71 struct svc_xprt_class *cl; 72 int res = -EEXIST; 73 74 dprintk("svc: Adding svc transport class '%s'\n", xcl->xcl_name); 75 76 INIT_LIST_HEAD(&xcl->xcl_list); 77 spin_lock(&svc_xprt_class_lock); 78 /* Make sure there isn't already a class with the same name */ 79 list_for_each_entry(cl, &svc_xprt_class_list, xcl_list) { 80 if (strcmp(xcl->xcl_name, cl->xcl_name) == 0) 81 goto out; 82 } 83 list_add_tail(&xcl->xcl_list, &svc_xprt_class_list); 84 res = 0; 85 out: 86 spin_unlock(&svc_xprt_class_lock); 87 return res; 88 } 89 EXPORT_SYMBOL_GPL(svc_reg_xprt_class); 90 91 void svc_unreg_xprt_class(struct svc_xprt_class *xcl) 92 { 93 dprintk("svc: Removing svc transport class '%s'\n", xcl->xcl_name); 94 spin_lock(&svc_xprt_class_lock); 95 list_del_init(&xcl->xcl_list); 96 spin_unlock(&svc_xprt_class_lock); 97 } 98 EXPORT_SYMBOL_GPL(svc_unreg_xprt_class); 99 100 /* 101 * Format the transport list for printing 102 */ 103 int svc_print_xprts(char *buf, int maxlen) 104 { 105 struct svc_xprt_class *xcl; 106 char tmpstr[80]; 107 int len = 0; 108 buf[0] = '\0'; 109 110 spin_lock(&svc_xprt_class_lock); 111 list_for_each_entry(xcl, &svc_xprt_class_list, xcl_list) { 112 int slen; 113 114 sprintf(tmpstr, "%s %d\n", xcl->xcl_name, xcl->xcl_max_payload); 115 slen = strlen(tmpstr); 116 if (len + slen > maxlen) 117 break; 118 len += slen; 119 strcat(buf, tmpstr); 120 } 121 spin_unlock(&svc_xprt_class_lock); 122 123 return len; 124 } 125 126 static void svc_xprt_free(struct kref *kref) 127 { 128 struct svc_xprt *xprt = 129 container_of(kref, struct svc_xprt, xpt_ref); 130 struct module *owner = xprt->xpt_class->xcl_owner; 131 if (test_bit(XPT_CACHE_AUTH, &xprt->xpt_flags)) 132 svcauth_unix_info_release(xprt); 133 put_net(xprt->xpt_net); 134 /* See comment on corresponding get in xs_setup_bc_tcp(): */ 135 if (xprt->xpt_bc_xprt) 136 xprt_put(xprt->xpt_bc_xprt); 137 xprt->xpt_ops->xpo_free(xprt); 138 module_put(owner); 139 } 140 141 void svc_xprt_put(struct svc_xprt *xprt) 142 { 143 kref_put(&xprt->xpt_ref, svc_xprt_free); 144 } 145 EXPORT_SYMBOL_GPL(svc_xprt_put); 146 147 /* 148 * Called by transport drivers to initialize the transport independent 149 * portion of the transport instance. 150 */ 151 void svc_xprt_init(struct net *net, struct svc_xprt_class *xcl, 152 struct svc_xprt *xprt, struct svc_serv *serv) 153 { 154 memset(xprt, 0, sizeof(*xprt)); 155 xprt->xpt_class = xcl; 156 xprt->xpt_ops = xcl->xcl_ops; 157 kref_init(&xprt->xpt_ref); 158 xprt->xpt_server = serv; 159 INIT_LIST_HEAD(&xprt->xpt_list); 160 INIT_LIST_HEAD(&xprt->xpt_ready); 161 INIT_LIST_HEAD(&xprt->xpt_deferred); 162 INIT_LIST_HEAD(&xprt->xpt_users); 163 mutex_init(&xprt->xpt_mutex); 164 spin_lock_init(&xprt->xpt_lock); 165 set_bit(XPT_BUSY, &xprt->xpt_flags); 166 rpc_init_wait_queue(&xprt->xpt_bc_pending, "xpt_bc_pending"); 167 xprt->xpt_net = get_net(net); 168 } 169 EXPORT_SYMBOL_GPL(svc_xprt_init); 170 171 static struct svc_xprt *__svc_xpo_create(struct svc_xprt_class *xcl, 172 struct svc_serv *serv, 173 struct net *net, 174 const int family, 175 const unsigned short port, 176 int flags) 177 { 178 struct sockaddr_in sin = { 179 .sin_family = AF_INET, 180 .sin_addr.s_addr = htonl(INADDR_ANY), 181 .sin_port = htons(port), 182 }; 183 #if IS_ENABLED(CONFIG_IPV6) 184 struct sockaddr_in6 sin6 = { 185 .sin6_family = AF_INET6, 186 .sin6_addr = IN6ADDR_ANY_INIT, 187 .sin6_port = htons(port), 188 }; 189 #endif 190 struct sockaddr *sap; 191 size_t len; 192 193 switch (family) { 194 case PF_INET: 195 sap = (struct sockaddr *)&sin; 196 len = sizeof(sin); 197 break; 198 #if IS_ENABLED(CONFIG_IPV6) 199 case PF_INET6: 200 sap = (struct sockaddr *)&sin6; 201 len = sizeof(sin6); 202 break; 203 #endif 204 default: 205 return ERR_PTR(-EAFNOSUPPORT); 206 } 207 208 return xcl->xcl_ops->xpo_create(serv, net, sap, len, flags); 209 } 210 211 int svc_create_xprt(struct svc_serv *serv, const char *xprt_name, 212 struct net *net, const int family, 213 const unsigned short port, int flags) 214 { 215 struct svc_xprt_class *xcl; 216 217 dprintk("svc: creating transport %s[%d]\n", xprt_name, port); 218 spin_lock(&svc_xprt_class_lock); 219 list_for_each_entry(xcl, &svc_xprt_class_list, xcl_list) { 220 struct svc_xprt *newxprt; 221 unsigned short newport; 222 223 if (strcmp(xprt_name, xcl->xcl_name)) 224 continue; 225 226 if (!try_module_get(xcl->xcl_owner)) 227 goto err; 228 229 spin_unlock(&svc_xprt_class_lock); 230 newxprt = __svc_xpo_create(xcl, serv, net, family, port, flags); 231 if (IS_ERR(newxprt)) { 232 module_put(xcl->xcl_owner); 233 return PTR_ERR(newxprt); 234 } 235 236 clear_bit(XPT_TEMP, &newxprt->xpt_flags); 237 spin_lock_bh(&serv->sv_lock); 238 list_add(&newxprt->xpt_list, &serv->sv_permsocks); 239 spin_unlock_bh(&serv->sv_lock); 240 newport = svc_xprt_local_port(newxprt); 241 clear_bit(XPT_BUSY, &newxprt->xpt_flags); 242 return newport; 243 } 244 err: 245 spin_unlock(&svc_xprt_class_lock); 246 dprintk("svc: transport %s not found\n", xprt_name); 247 248 /* This errno is exposed to user space. Provide a reasonable 249 * perror msg for a bad transport. */ 250 return -EPROTONOSUPPORT; 251 } 252 EXPORT_SYMBOL_GPL(svc_create_xprt); 253 254 /* 255 * Copy the local and remote xprt addresses to the rqstp structure 256 */ 257 void svc_xprt_copy_addrs(struct svc_rqst *rqstp, struct svc_xprt *xprt) 258 { 259 memcpy(&rqstp->rq_addr, &xprt->xpt_remote, xprt->xpt_remotelen); 260 rqstp->rq_addrlen = xprt->xpt_remotelen; 261 262 /* 263 * Destination address in request is needed for binding the 264 * source address in RPC replies/callbacks later. 265 */ 266 memcpy(&rqstp->rq_daddr, &xprt->xpt_local, xprt->xpt_locallen); 267 rqstp->rq_daddrlen = xprt->xpt_locallen; 268 } 269 EXPORT_SYMBOL_GPL(svc_xprt_copy_addrs); 270 271 /** 272 * svc_print_addr - Format rq_addr field for printing 273 * @rqstp: svc_rqst struct containing address to print 274 * @buf: target buffer for formatted address 275 * @len: length of target buffer 276 * 277 */ 278 char *svc_print_addr(struct svc_rqst *rqstp, char *buf, size_t len) 279 { 280 return __svc_print_addr(svc_addr(rqstp), buf, len); 281 } 282 EXPORT_SYMBOL_GPL(svc_print_addr); 283 284 /* 285 * Queue up an idle server thread. Must have pool->sp_lock held. 286 * Note: this is really a stack rather than a queue, so that we only 287 * use as many different threads as we need, and the rest don't pollute 288 * the cache. 289 */ 290 static void svc_thread_enqueue(struct svc_pool *pool, struct svc_rqst *rqstp) 291 { 292 list_add(&rqstp->rq_list, &pool->sp_threads); 293 } 294 295 /* 296 * Dequeue an nfsd thread. Must have pool->sp_lock held. 297 */ 298 static void svc_thread_dequeue(struct svc_pool *pool, struct svc_rqst *rqstp) 299 { 300 list_del(&rqstp->rq_list); 301 } 302 303 static bool svc_xprt_has_something_to_do(struct svc_xprt *xprt) 304 { 305 if (xprt->xpt_flags & ((1<<XPT_CONN)|(1<<XPT_CLOSE))) 306 return true; 307 if (xprt->xpt_flags & ((1<<XPT_DATA)|(1<<XPT_DEFERRED))) 308 return xprt->xpt_ops->xpo_has_wspace(xprt); 309 return false; 310 } 311 312 /* 313 * Queue up a transport with data pending. If there are idle nfsd 314 * processes, wake 'em up. 315 * 316 */ 317 void svc_xprt_enqueue(struct svc_xprt *xprt) 318 { 319 struct svc_serv *serv = xprt->xpt_server; 320 struct svc_pool *pool; 321 struct svc_rqst *rqstp; 322 int cpu; 323 324 if (!svc_xprt_has_something_to_do(xprt)) 325 return; 326 327 cpu = get_cpu(); 328 pool = svc_pool_for_cpu(xprt->xpt_server, cpu); 329 put_cpu(); 330 331 spin_lock_bh(&pool->sp_lock); 332 333 if (!list_empty(&pool->sp_threads) && 334 !list_empty(&pool->sp_sockets)) 335 printk(KERN_ERR 336 "svc_xprt_enqueue: " 337 "threads and transports both waiting??\n"); 338 339 pool->sp_stats.packets++; 340 341 /* Mark transport as busy. It will remain in this state until 342 * the provider calls svc_xprt_received. We update XPT_BUSY 343 * atomically because it also guards against trying to enqueue 344 * the transport twice. 345 */ 346 if (test_and_set_bit(XPT_BUSY, &xprt->xpt_flags)) { 347 /* Don't enqueue transport while already enqueued */ 348 dprintk("svc: transport %p busy, not enqueued\n", xprt); 349 goto out_unlock; 350 } 351 352 if (!list_empty(&pool->sp_threads)) { 353 rqstp = list_entry(pool->sp_threads.next, 354 struct svc_rqst, 355 rq_list); 356 dprintk("svc: transport %p served by daemon %p\n", 357 xprt, rqstp); 358 svc_thread_dequeue(pool, rqstp); 359 if (rqstp->rq_xprt) 360 printk(KERN_ERR 361 "svc_xprt_enqueue: server %p, rq_xprt=%p!\n", 362 rqstp, rqstp->rq_xprt); 363 rqstp->rq_xprt = xprt; 364 svc_xprt_get(xprt); 365 rqstp->rq_reserved = serv->sv_max_mesg; 366 atomic_add(rqstp->rq_reserved, &xprt->xpt_reserved); 367 pool->sp_stats.threads_woken++; 368 wake_up(&rqstp->rq_wait); 369 } else { 370 dprintk("svc: transport %p put into queue\n", xprt); 371 list_add_tail(&xprt->xpt_ready, &pool->sp_sockets); 372 pool->sp_stats.sockets_queued++; 373 } 374 375 out_unlock: 376 spin_unlock_bh(&pool->sp_lock); 377 } 378 EXPORT_SYMBOL_GPL(svc_xprt_enqueue); 379 380 /* 381 * Dequeue the first transport. Must be called with the pool->sp_lock held. 382 */ 383 static struct svc_xprt *svc_xprt_dequeue(struct svc_pool *pool) 384 { 385 struct svc_xprt *xprt; 386 387 if (list_empty(&pool->sp_sockets)) 388 return NULL; 389 390 xprt = list_entry(pool->sp_sockets.next, 391 struct svc_xprt, xpt_ready); 392 list_del_init(&xprt->xpt_ready); 393 394 dprintk("svc: transport %p dequeued, inuse=%d\n", 395 xprt, atomic_read(&xprt->xpt_ref.refcount)); 396 397 return xprt; 398 } 399 400 /* 401 * svc_xprt_received conditionally queues the transport for processing 402 * by another thread. The caller must hold the XPT_BUSY bit and must 403 * not thereafter touch transport data. 404 * 405 * Note: XPT_DATA only gets cleared when a read-attempt finds no (or 406 * insufficient) data. 407 */ 408 void svc_xprt_received(struct svc_xprt *xprt) 409 { 410 BUG_ON(!test_bit(XPT_BUSY, &xprt->xpt_flags)); 411 /* As soon as we clear busy, the xprt could be closed and 412 * 'put', so we need a reference to call svc_xprt_enqueue with: 413 */ 414 svc_xprt_get(xprt); 415 clear_bit(XPT_BUSY, &xprt->xpt_flags); 416 svc_xprt_enqueue(xprt); 417 svc_xprt_put(xprt); 418 } 419 EXPORT_SYMBOL_GPL(svc_xprt_received); 420 421 /** 422 * svc_reserve - change the space reserved for the reply to a request. 423 * @rqstp: The request in question 424 * @space: new max space to reserve 425 * 426 * Each request reserves some space on the output queue of the transport 427 * to make sure the reply fits. This function reduces that reserved 428 * space to be the amount of space used already, plus @space. 429 * 430 */ 431 void svc_reserve(struct svc_rqst *rqstp, int space) 432 { 433 space += rqstp->rq_res.head[0].iov_len; 434 435 if (space < rqstp->rq_reserved) { 436 struct svc_xprt *xprt = rqstp->rq_xprt; 437 atomic_sub((rqstp->rq_reserved - space), &xprt->xpt_reserved); 438 rqstp->rq_reserved = space; 439 440 svc_xprt_enqueue(xprt); 441 } 442 } 443 EXPORT_SYMBOL_GPL(svc_reserve); 444 445 static void svc_xprt_release(struct svc_rqst *rqstp) 446 { 447 struct svc_xprt *xprt = rqstp->rq_xprt; 448 449 rqstp->rq_xprt->xpt_ops->xpo_release_rqst(rqstp); 450 451 kfree(rqstp->rq_deferred); 452 rqstp->rq_deferred = NULL; 453 454 svc_free_res_pages(rqstp); 455 rqstp->rq_res.page_len = 0; 456 rqstp->rq_res.page_base = 0; 457 458 /* Reset response buffer and release 459 * the reservation. 460 * But first, check that enough space was reserved 461 * for the reply, otherwise we have a bug! 462 */ 463 if ((rqstp->rq_res.len) > rqstp->rq_reserved) 464 printk(KERN_ERR "RPC request reserved %d but used %d\n", 465 rqstp->rq_reserved, 466 rqstp->rq_res.len); 467 468 rqstp->rq_res.head[0].iov_len = 0; 469 svc_reserve(rqstp, 0); 470 rqstp->rq_xprt = NULL; 471 472 svc_xprt_put(xprt); 473 } 474 475 /* 476 * External function to wake up a server waiting for data 477 * This really only makes sense for services like lockd 478 * which have exactly one thread anyway. 479 */ 480 void svc_wake_up(struct svc_serv *serv) 481 { 482 struct svc_rqst *rqstp; 483 unsigned int i; 484 struct svc_pool *pool; 485 486 for (i = 0; i < serv->sv_nrpools; i++) { 487 pool = &serv->sv_pools[i]; 488 489 spin_lock_bh(&pool->sp_lock); 490 if (!list_empty(&pool->sp_threads)) { 491 rqstp = list_entry(pool->sp_threads.next, 492 struct svc_rqst, 493 rq_list); 494 dprintk("svc: daemon %p woken up.\n", rqstp); 495 /* 496 svc_thread_dequeue(pool, rqstp); 497 rqstp->rq_xprt = NULL; 498 */ 499 wake_up(&rqstp->rq_wait); 500 } 501 spin_unlock_bh(&pool->sp_lock); 502 } 503 } 504 EXPORT_SYMBOL_GPL(svc_wake_up); 505 506 int svc_port_is_privileged(struct sockaddr *sin) 507 { 508 switch (sin->sa_family) { 509 case AF_INET: 510 return ntohs(((struct sockaddr_in *)sin)->sin_port) 511 < PROT_SOCK; 512 case AF_INET6: 513 return ntohs(((struct sockaddr_in6 *)sin)->sin6_port) 514 < PROT_SOCK; 515 default: 516 return 0; 517 } 518 } 519 520 /* 521 * Make sure that we don't have too many active connections. If we have, 522 * something must be dropped. It's not clear what will happen if we allow 523 * "too many" connections, but when dealing with network-facing software, 524 * we have to code defensively. Here we do that by imposing hard limits. 525 * 526 * There's no point in trying to do random drop here for DoS 527 * prevention. The NFS clients does 1 reconnect in 15 seconds. An 528 * attacker can easily beat that. 529 * 530 * The only somewhat efficient mechanism would be if drop old 531 * connections from the same IP first. But right now we don't even 532 * record the client IP in svc_sock. 533 * 534 * single-threaded services that expect a lot of clients will probably 535 * need to set sv_maxconn to override the default value which is based 536 * on the number of threads 537 */ 538 static void svc_check_conn_limits(struct svc_serv *serv) 539 { 540 unsigned int limit = serv->sv_maxconn ? serv->sv_maxconn : 541 (serv->sv_nrthreads+3) * 20; 542 543 if (serv->sv_tmpcnt > limit) { 544 struct svc_xprt *xprt = NULL; 545 spin_lock_bh(&serv->sv_lock); 546 if (!list_empty(&serv->sv_tempsocks)) { 547 if (net_ratelimit()) { 548 /* Try to help the admin */ 549 printk(KERN_NOTICE "%s: too many open " 550 "connections, consider increasing %s\n", 551 serv->sv_name, serv->sv_maxconn ? 552 "the max number of connections." : 553 "the number of threads."); 554 } 555 /* 556 * Always select the oldest connection. It's not fair, 557 * but so is life 558 */ 559 xprt = list_entry(serv->sv_tempsocks.prev, 560 struct svc_xprt, 561 xpt_list); 562 set_bit(XPT_CLOSE, &xprt->xpt_flags); 563 svc_xprt_get(xprt); 564 } 565 spin_unlock_bh(&serv->sv_lock); 566 567 if (xprt) { 568 svc_xprt_enqueue(xprt); 569 svc_xprt_put(xprt); 570 } 571 } 572 } 573 574 /* 575 * Receive the next request on any transport. This code is carefully 576 * organised not to touch any cachelines in the shared svc_serv 577 * structure, only cachelines in the local svc_pool. 578 */ 579 int svc_recv(struct svc_rqst *rqstp, long timeout) 580 { 581 struct svc_xprt *xprt = NULL; 582 struct svc_serv *serv = rqstp->rq_server; 583 struct svc_pool *pool = rqstp->rq_pool; 584 int len, i; 585 int pages; 586 struct xdr_buf *arg; 587 DECLARE_WAITQUEUE(wait, current); 588 long time_left; 589 590 dprintk("svc: server %p waiting for data (to = %ld)\n", 591 rqstp, timeout); 592 593 if (rqstp->rq_xprt) 594 printk(KERN_ERR 595 "svc_recv: service %p, transport not NULL!\n", 596 rqstp); 597 if (waitqueue_active(&rqstp->rq_wait)) 598 printk(KERN_ERR 599 "svc_recv: service %p, wait queue active!\n", 600 rqstp); 601 602 /* now allocate needed pages. If we get a failure, sleep briefly */ 603 pages = (serv->sv_max_mesg + PAGE_SIZE) / PAGE_SIZE; 604 for (i = 0; i < pages ; i++) 605 while (rqstp->rq_pages[i] == NULL) { 606 struct page *p = alloc_page(GFP_KERNEL); 607 if (!p) { 608 set_current_state(TASK_INTERRUPTIBLE); 609 if (signalled() || kthread_should_stop()) { 610 set_current_state(TASK_RUNNING); 611 return -EINTR; 612 } 613 schedule_timeout(msecs_to_jiffies(500)); 614 } 615 rqstp->rq_pages[i] = p; 616 } 617 rqstp->rq_pages[i++] = NULL; /* this might be seen in nfs_read_actor */ 618 BUG_ON(pages >= RPCSVC_MAXPAGES); 619 620 /* Make arg->head point to first page and arg->pages point to rest */ 621 arg = &rqstp->rq_arg; 622 arg->head[0].iov_base = page_address(rqstp->rq_pages[0]); 623 arg->head[0].iov_len = PAGE_SIZE; 624 arg->pages = rqstp->rq_pages + 1; 625 arg->page_base = 0; 626 /* save at least one page for response */ 627 arg->page_len = (pages-2)*PAGE_SIZE; 628 arg->len = (pages-1)*PAGE_SIZE; 629 arg->tail[0].iov_len = 0; 630 631 try_to_freeze(); 632 cond_resched(); 633 if (signalled() || kthread_should_stop()) 634 return -EINTR; 635 636 /* Normally we will wait up to 5 seconds for any required 637 * cache information to be provided. 638 */ 639 rqstp->rq_chandle.thread_wait = 5*HZ; 640 641 spin_lock_bh(&pool->sp_lock); 642 xprt = svc_xprt_dequeue(pool); 643 if (xprt) { 644 rqstp->rq_xprt = xprt; 645 svc_xprt_get(xprt); 646 rqstp->rq_reserved = serv->sv_max_mesg; 647 atomic_add(rqstp->rq_reserved, &xprt->xpt_reserved); 648 649 /* As there is a shortage of threads and this request 650 * had to be queued, don't allow the thread to wait so 651 * long for cache updates. 652 */ 653 rqstp->rq_chandle.thread_wait = 1*HZ; 654 } else { 655 /* No data pending. Go to sleep */ 656 svc_thread_enqueue(pool, rqstp); 657 658 /* 659 * We have to be able to interrupt this wait 660 * to bring down the daemons ... 661 */ 662 set_current_state(TASK_INTERRUPTIBLE); 663 664 /* 665 * checking kthread_should_stop() here allows us to avoid 666 * locking and signalling when stopping kthreads that call 667 * svc_recv. If the thread has already been woken up, then 668 * we can exit here without sleeping. If not, then it 669 * it'll be woken up quickly during the schedule_timeout 670 */ 671 if (kthread_should_stop()) { 672 set_current_state(TASK_RUNNING); 673 spin_unlock_bh(&pool->sp_lock); 674 return -EINTR; 675 } 676 677 add_wait_queue(&rqstp->rq_wait, &wait); 678 spin_unlock_bh(&pool->sp_lock); 679 680 time_left = schedule_timeout(timeout); 681 682 try_to_freeze(); 683 684 spin_lock_bh(&pool->sp_lock); 685 remove_wait_queue(&rqstp->rq_wait, &wait); 686 if (!time_left) 687 pool->sp_stats.threads_timedout++; 688 689 xprt = rqstp->rq_xprt; 690 if (!xprt) { 691 svc_thread_dequeue(pool, rqstp); 692 spin_unlock_bh(&pool->sp_lock); 693 dprintk("svc: server %p, no data yet\n", rqstp); 694 if (signalled() || kthread_should_stop()) 695 return -EINTR; 696 else 697 return -EAGAIN; 698 } 699 } 700 spin_unlock_bh(&pool->sp_lock); 701 702 len = 0; 703 if (test_bit(XPT_CLOSE, &xprt->xpt_flags)) { 704 dprintk("svc_recv: found XPT_CLOSE\n"); 705 svc_delete_xprt(xprt); 706 /* Leave XPT_BUSY set on the dead xprt: */ 707 goto out; 708 } 709 if (test_bit(XPT_LISTENER, &xprt->xpt_flags)) { 710 struct svc_xprt *newxpt; 711 newxpt = xprt->xpt_ops->xpo_accept(xprt); 712 if (newxpt) { 713 /* 714 * We know this module_get will succeed because the 715 * listener holds a reference too 716 */ 717 __module_get(newxpt->xpt_class->xcl_owner); 718 svc_check_conn_limits(xprt->xpt_server); 719 spin_lock_bh(&serv->sv_lock); 720 set_bit(XPT_TEMP, &newxpt->xpt_flags); 721 list_add(&newxpt->xpt_list, &serv->sv_tempsocks); 722 serv->sv_tmpcnt++; 723 if (serv->sv_temptimer.function == NULL) { 724 /* setup timer to age temp transports */ 725 setup_timer(&serv->sv_temptimer, 726 svc_age_temp_xprts, 727 (unsigned long)serv); 728 mod_timer(&serv->sv_temptimer, 729 jiffies + svc_conn_age_period * HZ); 730 } 731 spin_unlock_bh(&serv->sv_lock); 732 svc_xprt_received(newxpt); 733 } 734 } else if (xprt->xpt_ops->xpo_has_wspace(xprt)) { 735 dprintk("svc: server %p, pool %u, transport %p, inuse=%d\n", 736 rqstp, pool->sp_id, xprt, 737 atomic_read(&xprt->xpt_ref.refcount)); 738 rqstp->rq_deferred = svc_deferred_dequeue(xprt); 739 if (rqstp->rq_deferred) 740 len = svc_deferred_recv(rqstp); 741 else 742 len = xprt->xpt_ops->xpo_recvfrom(rqstp); 743 dprintk("svc: got len=%d\n", len); 744 } 745 svc_xprt_received(xprt); 746 747 /* No data, incomplete (TCP) read, or accept() */ 748 if (len == 0 || len == -EAGAIN) 749 goto out; 750 751 clear_bit(XPT_OLD, &xprt->xpt_flags); 752 753 rqstp->rq_secure = svc_port_is_privileged(svc_addr(rqstp)); 754 rqstp->rq_chandle.defer = svc_defer; 755 756 if (serv->sv_stats) 757 serv->sv_stats->netcnt++; 758 return len; 759 out: 760 rqstp->rq_res.len = 0; 761 svc_xprt_release(rqstp); 762 return -EAGAIN; 763 } 764 EXPORT_SYMBOL_GPL(svc_recv); 765 766 /* 767 * Drop request 768 */ 769 void svc_drop(struct svc_rqst *rqstp) 770 { 771 dprintk("svc: xprt %p dropped request\n", rqstp->rq_xprt); 772 svc_xprt_release(rqstp); 773 } 774 EXPORT_SYMBOL_GPL(svc_drop); 775 776 /* 777 * Return reply to client. 778 */ 779 int svc_send(struct svc_rqst *rqstp) 780 { 781 struct svc_xprt *xprt; 782 int len; 783 struct xdr_buf *xb; 784 785 xprt = rqstp->rq_xprt; 786 if (!xprt) 787 return -EFAULT; 788 789 /* release the receive skb before sending the reply */ 790 rqstp->rq_xprt->xpt_ops->xpo_release_rqst(rqstp); 791 792 /* calculate over-all length */ 793 xb = &rqstp->rq_res; 794 xb->len = xb->head[0].iov_len + 795 xb->page_len + 796 xb->tail[0].iov_len; 797 798 /* Grab mutex to serialize outgoing data. */ 799 mutex_lock(&xprt->xpt_mutex); 800 if (test_bit(XPT_DEAD, &xprt->xpt_flags)) 801 len = -ENOTCONN; 802 else 803 len = xprt->xpt_ops->xpo_sendto(rqstp); 804 mutex_unlock(&xprt->xpt_mutex); 805 rpc_wake_up(&xprt->xpt_bc_pending); 806 svc_xprt_release(rqstp); 807 808 if (len == -ECONNREFUSED || len == -ENOTCONN || len == -EAGAIN) 809 return 0; 810 return len; 811 } 812 813 /* 814 * Timer function to close old temporary transports, using 815 * a mark-and-sweep algorithm. 816 */ 817 static void svc_age_temp_xprts(unsigned long closure) 818 { 819 struct svc_serv *serv = (struct svc_serv *)closure; 820 struct svc_xprt *xprt; 821 struct list_head *le, *next; 822 LIST_HEAD(to_be_aged); 823 824 dprintk("svc_age_temp_xprts\n"); 825 826 if (!spin_trylock_bh(&serv->sv_lock)) { 827 /* busy, try again 1 sec later */ 828 dprintk("svc_age_temp_xprts: busy\n"); 829 mod_timer(&serv->sv_temptimer, jiffies + HZ); 830 return; 831 } 832 833 list_for_each_safe(le, next, &serv->sv_tempsocks) { 834 xprt = list_entry(le, struct svc_xprt, xpt_list); 835 836 /* First time through, just mark it OLD. Second time 837 * through, close it. */ 838 if (!test_and_set_bit(XPT_OLD, &xprt->xpt_flags)) 839 continue; 840 if (atomic_read(&xprt->xpt_ref.refcount) > 1 || 841 test_bit(XPT_BUSY, &xprt->xpt_flags)) 842 continue; 843 svc_xprt_get(xprt); 844 list_move(le, &to_be_aged); 845 set_bit(XPT_CLOSE, &xprt->xpt_flags); 846 set_bit(XPT_DETACHED, &xprt->xpt_flags); 847 } 848 spin_unlock_bh(&serv->sv_lock); 849 850 while (!list_empty(&to_be_aged)) { 851 le = to_be_aged.next; 852 /* fiddling the xpt_list node is safe 'cos we're XPT_DETACHED */ 853 list_del_init(le); 854 xprt = list_entry(le, struct svc_xprt, xpt_list); 855 856 dprintk("queuing xprt %p for closing\n", xprt); 857 858 /* a thread will dequeue and close it soon */ 859 svc_xprt_enqueue(xprt); 860 svc_xprt_put(xprt); 861 } 862 863 mod_timer(&serv->sv_temptimer, jiffies + svc_conn_age_period * HZ); 864 } 865 866 static void call_xpt_users(struct svc_xprt *xprt) 867 { 868 struct svc_xpt_user *u; 869 870 spin_lock(&xprt->xpt_lock); 871 while (!list_empty(&xprt->xpt_users)) { 872 u = list_first_entry(&xprt->xpt_users, struct svc_xpt_user, list); 873 list_del(&u->list); 874 u->callback(u); 875 } 876 spin_unlock(&xprt->xpt_lock); 877 } 878 879 /* 880 * Remove a dead transport 881 */ 882 static void svc_delete_xprt(struct svc_xprt *xprt) 883 { 884 struct svc_serv *serv = xprt->xpt_server; 885 struct svc_deferred_req *dr; 886 887 /* Only do this once */ 888 if (test_and_set_bit(XPT_DEAD, &xprt->xpt_flags)) 889 BUG(); 890 891 dprintk("svc: svc_delete_xprt(%p)\n", xprt); 892 xprt->xpt_ops->xpo_detach(xprt); 893 894 spin_lock_bh(&serv->sv_lock); 895 if (!test_and_set_bit(XPT_DETACHED, &xprt->xpt_flags)) 896 list_del_init(&xprt->xpt_list); 897 BUG_ON(!list_empty(&xprt->xpt_ready)); 898 if (test_bit(XPT_TEMP, &xprt->xpt_flags)) 899 serv->sv_tmpcnt--; 900 spin_unlock_bh(&serv->sv_lock); 901 902 while ((dr = svc_deferred_dequeue(xprt)) != NULL) 903 kfree(dr); 904 905 call_xpt_users(xprt); 906 svc_xprt_put(xprt); 907 } 908 909 void svc_close_xprt(struct svc_xprt *xprt) 910 { 911 set_bit(XPT_CLOSE, &xprt->xpt_flags); 912 if (test_and_set_bit(XPT_BUSY, &xprt->xpt_flags)) 913 /* someone else will have to effect the close */ 914 return; 915 /* 916 * We expect svc_close_xprt() to work even when no threads are 917 * running (e.g., while configuring the server before starting 918 * any threads), so if the transport isn't busy, we delete 919 * it ourself: 920 */ 921 svc_delete_xprt(xprt); 922 } 923 EXPORT_SYMBOL_GPL(svc_close_xprt); 924 925 static void svc_close_list(struct list_head *xprt_list) 926 { 927 struct svc_xprt *xprt; 928 929 list_for_each_entry(xprt, xprt_list, xpt_list) { 930 set_bit(XPT_CLOSE, &xprt->xpt_flags); 931 set_bit(XPT_BUSY, &xprt->xpt_flags); 932 } 933 } 934 935 void svc_close_all(struct svc_serv *serv) 936 { 937 struct svc_pool *pool; 938 struct svc_xprt *xprt; 939 struct svc_xprt *tmp; 940 int i; 941 942 svc_close_list(&serv->sv_tempsocks); 943 svc_close_list(&serv->sv_permsocks); 944 945 for (i = 0; i < serv->sv_nrpools; i++) { 946 pool = &serv->sv_pools[i]; 947 948 spin_lock_bh(&pool->sp_lock); 949 while (!list_empty(&pool->sp_sockets)) { 950 xprt = list_first_entry(&pool->sp_sockets, struct svc_xprt, xpt_ready); 951 list_del_init(&xprt->xpt_ready); 952 } 953 spin_unlock_bh(&pool->sp_lock); 954 } 955 /* 956 * At this point the sp_sockets lists will stay empty, since 957 * svc_enqueue will not add new entries without taking the 958 * sp_lock and checking XPT_BUSY. 959 */ 960 list_for_each_entry_safe(xprt, tmp, &serv->sv_tempsocks, xpt_list) 961 svc_delete_xprt(xprt); 962 list_for_each_entry_safe(xprt, tmp, &serv->sv_permsocks, xpt_list) 963 svc_delete_xprt(xprt); 964 965 BUG_ON(!list_empty(&serv->sv_permsocks)); 966 BUG_ON(!list_empty(&serv->sv_tempsocks)); 967 } 968 969 /* 970 * Handle defer and revisit of requests 971 */ 972 973 static void svc_revisit(struct cache_deferred_req *dreq, int too_many) 974 { 975 struct svc_deferred_req *dr = 976 container_of(dreq, struct svc_deferred_req, handle); 977 struct svc_xprt *xprt = dr->xprt; 978 979 spin_lock(&xprt->xpt_lock); 980 set_bit(XPT_DEFERRED, &xprt->xpt_flags); 981 if (too_many || test_bit(XPT_DEAD, &xprt->xpt_flags)) { 982 spin_unlock(&xprt->xpt_lock); 983 dprintk("revisit canceled\n"); 984 svc_xprt_put(xprt); 985 kfree(dr); 986 return; 987 } 988 dprintk("revisit queued\n"); 989 dr->xprt = NULL; 990 list_add(&dr->handle.recent, &xprt->xpt_deferred); 991 spin_unlock(&xprt->xpt_lock); 992 svc_xprt_enqueue(xprt); 993 svc_xprt_put(xprt); 994 } 995 996 /* 997 * Save the request off for later processing. The request buffer looks 998 * like this: 999 * 1000 * <xprt-header><rpc-header><rpc-pagelist><rpc-tail> 1001 * 1002 * This code can only handle requests that consist of an xprt-header 1003 * and rpc-header. 1004 */ 1005 static struct cache_deferred_req *svc_defer(struct cache_req *req) 1006 { 1007 struct svc_rqst *rqstp = container_of(req, struct svc_rqst, rq_chandle); 1008 struct svc_deferred_req *dr; 1009 1010 if (rqstp->rq_arg.page_len || !rqstp->rq_usedeferral) 1011 return NULL; /* if more than a page, give up FIXME */ 1012 if (rqstp->rq_deferred) { 1013 dr = rqstp->rq_deferred; 1014 rqstp->rq_deferred = NULL; 1015 } else { 1016 size_t skip; 1017 size_t size; 1018 /* FIXME maybe discard if size too large */ 1019 size = sizeof(struct svc_deferred_req) + rqstp->rq_arg.len; 1020 dr = kmalloc(size, GFP_KERNEL); 1021 if (dr == NULL) 1022 return NULL; 1023 1024 dr->handle.owner = rqstp->rq_server; 1025 dr->prot = rqstp->rq_prot; 1026 memcpy(&dr->addr, &rqstp->rq_addr, rqstp->rq_addrlen); 1027 dr->addrlen = rqstp->rq_addrlen; 1028 dr->daddr = rqstp->rq_daddr; 1029 dr->argslen = rqstp->rq_arg.len >> 2; 1030 dr->xprt_hlen = rqstp->rq_xprt_hlen; 1031 1032 /* back up head to the start of the buffer and copy */ 1033 skip = rqstp->rq_arg.len - rqstp->rq_arg.head[0].iov_len; 1034 memcpy(dr->args, rqstp->rq_arg.head[0].iov_base - skip, 1035 dr->argslen << 2); 1036 } 1037 svc_xprt_get(rqstp->rq_xprt); 1038 dr->xprt = rqstp->rq_xprt; 1039 rqstp->rq_dropme = true; 1040 1041 dr->handle.revisit = svc_revisit; 1042 return &dr->handle; 1043 } 1044 1045 /* 1046 * recv data from a deferred request into an active one 1047 */ 1048 static int svc_deferred_recv(struct svc_rqst *rqstp) 1049 { 1050 struct svc_deferred_req *dr = rqstp->rq_deferred; 1051 1052 /* setup iov_base past transport header */ 1053 rqstp->rq_arg.head[0].iov_base = dr->args + (dr->xprt_hlen>>2); 1054 /* The iov_len does not include the transport header bytes */ 1055 rqstp->rq_arg.head[0].iov_len = (dr->argslen<<2) - dr->xprt_hlen; 1056 rqstp->rq_arg.page_len = 0; 1057 /* The rq_arg.len includes the transport header bytes */ 1058 rqstp->rq_arg.len = dr->argslen<<2; 1059 rqstp->rq_prot = dr->prot; 1060 memcpy(&rqstp->rq_addr, &dr->addr, dr->addrlen); 1061 rqstp->rq_addrlen = dr->addrlen; 1062 /* Save off transport header len in case we get deferred again */ 1063 rqstp->rq_xprt_hlen = dr->xprt_hlen; 1064 rqstp->rq_daddr = dr->daddr; 1065 rqstp->rq_respages = rqstp->rq_pages; 1066 return (dr->argslen<<2) - dr->xprt_hlen; 1067 } 1068 1069 1070 static struct svc_deferred_req *svc_deferred_dequeue(struct svc_xprt *xprt) 1071 { 1072 struct svc_deferred_req *dr = NULL; 1073 1074 if (!test_bit(XPT_DEFERRED, &xprt->xpt_flags)) 1075 return NULL; 1076 spin_lock(&xprt->xpt_lock); 1077 if (!list_empty(&xprt->xpt_deferred)) { 1078 dr = list_entry(xprt->xpt_deferred.next, 1079 struct svc_deferred_req, 1080 handle.recent); 1081 list_del_init(&dr->handle.recent); 1082 } else 1083 clear_bit(XPT_DEFERRED, &xprt->xpt_flags); 1084 spin_unlock(&xprt->xpt_lock); 1085 return dr; 1086 } 1087 1088 /** 1089 * svc_find_xprt - find an RPC transport instance 1090 * @serv: pointer to svc_serv to search 1091 * @xcl_name: C string containing transport's class name 1092 * @af: Address family of transport's local address 1093 * @port: transport's IP port number 1094 * 1095 * Return the transport instance pointer for the endpoint accepting 1096 * connections/peer traffic from the specified transport class, 1097 * address family and port. 1098 * 1099 * Specifying 0 for the address family or port is effectively a 1100 * wild-card, and will result in matching the first transport in the 1101 * service's list that has a matching class name. 1102 */ 1103 struct svc_xprt *svc_find_xprt(struct svc_serv *serv, const char *xcl_name, 1104 const sa_family_t af, const unsigned short port) 1105 { 1106 struct svc_xprt *xprt; 1107 struct svc_xprt *found = NULL; 1108 1109 /* Sanity check the args */ 1110 if (serv == NULL || xcl_name == NULL) 1111 return found; 1112 1113 spin_lock_bh(&serv->sv_lock); 1114 list_for_each_entry(xprt, &serv->sv_permsocks, xpt_list) { 1115 if (strcmp(xprt->xpt_class->xcl_name, xcl_name)) 1116 continue; 1117 if (af != AF_UNSPEC && af != xprt->xpt_local.ss_family) 1118 continue; 1119 if (port != 0 && port != svc_xprt_local_port(xprt)) 1120 continue; 1121 found = xprt; 1122 svc_xprt_get(xprt); 1123 break; 1124 } 1125 spin_unlock_bh(&serv->sv_lock); 1126 return found; 1127 } 1128 EXPORT_SYMBOL_GPL(svc_find_xprt); 1129 1130 static int svc_one_xprt_name(const struct svc_xprt *xprt, 1131 char *pos, int remaining) 1132 { 1133 int len; 1134 1135 len = snprintf(pos, remaining, "%s %u\n", 1136 xprt->xpt_class->xcl_name, 1137 svc_xprt_local_port(xprt)); 1138 if (len >= remaining) 1139 return -ENAMETOOLONG; 1140 return len; 1141 } 1142 1143 /** 1144 * svc_xprt_names - format a buffer with a list of transport names 1145 * @serv: pointer to an RPC service 1146 * @buf: pointer to a buffer to be filled in 1147 * @buflen: length of buffer to be filled in 1148 * 1149 * Fills in @buf with a string containing a list of transport names, 1150 * each name terminated with '\n'. 1151 * 1152 * Returns positive length of the filled-in string on success; otherwise 1153 * a negative errno value is returned if an error occurs. 1154 */ 1155 int svc_xprt_names(struct svc_serv *serv, char *buf, const int buflen) 1156 { 1157 struct svc_xprt *xprt; 1158 int len, totlen; 1159 char *pos; 1160 1161 /* Sanity check args */ 1162 if (!serv) 1163 return 0; 1164 1165 spin_lock_bh(&serv->sv_lock); 1166 1167 pos = buf; 1168 totlen = 0; 1169 list_for_each_entry(xprt, &serv->sv_permsocks, xpt_list) { 1170 len = svc_one_xprt_name(xprt, pos, buflen - totlen); 1171 if (len < 0) { 1172 *buf = '\0'; 1173 totlen = len; 1174 } 1175 if (len <= 0) 1176 break; 1177 1178 pos += len; 1179 totlen += len; 1180 } 1181 1182 spin_unlock_bh(&serv->sv_lock); 1183 return totlen; 1184 } 1185 EXPORT_SYMBOL_GPL(svc_xprt_names); 1186 1187 1188 /*----------------------------------------------------------------------------*/ 1189 1190 static void *svc_pool_stats_start(struct seq_file *m, loff_t *pos) 1191 { 1192 unsigned int pidx = (unsigned int)*pos; 1193 struct svc_serv *serv = m->private; 1194 1195 dprintk("svc_pool_stats_start, *pidx=%u\n", pidx); 1196 1197 if (!pidx) 1198 return SEQ_START_TOKEN; 1199 return (pidx > serv->sv_nrpools ? NULL : &serv->sv_pools[pidx-1]); 1200 } 1201 1202 static void *svc_pool_stats_next(struct seq_file *m, void *p, loff_t *pos) 1203 { 1204 struct svc_pool *pool = p; 1205 struct svc_serv *serv = m->private; 1206 1207 dprintk("svc_pool_stats_next, *pos=%llu\n", *pos); 1208 1209 if (p == SEQ_START_TOKEN) { 1210 pool = &serv->sv_pools[0]; 1211 } else { 1212 unsigned int pidx = (pool - &serv->sv_pools[0]); 1213 if (pidx < serv->sv_nrpools-1) 1214 pool = &serv->sv_pools[pidx+1]; 1215 else 1216 pool = NULL; 1217 } 1218 ++*pos; 1219 return pool; 1220 } 1221 1222 static void svc_pool_stats_stop(struct seq_file *m, void *p) 1223 { 1224 } 1225 1226 static int svc_pool_stats_show(struct seq_file *m, void *p) 1227 { 1228 struct svc_pool *pool = p; 1229 1230 if (p == SEQ_START_TOKEN) { 1231 seq_puts(m, "# pool packets-arrived sockets-enqueued threads-woken threads-timedout\n"); 1232 return 0; 1233 } 1234 1235 seq_printf(m, "%u %lu %lu %lu %lu\n", 1236 pool->sp_id, 1237 pool->sp_stats.packets, 1238 pool->sp_stats.sockets_queued, 1239 pool->sp_stats.threads_woken, 1240 pool->sp_stats.threads_timedout); 1241 1242 return 0; 1243 } 1244 1245 static const struct seq_operations svc_pool_stats_seq_ops = { 1246 .start = svc_pool_stats_start, 1247 .next = svc_pool_stats_next, 1248 .stop = svc_pool_stats_stop, 1249 .show = svc_pool_stats_show, 1250 }; 1251 1252 int svc_pool_stats_open(struct svc_serv *serv, struct file *file) 1253 { 1254 int err; 1255 1256 err = seq_open(file, &svc_pool_stats_seq_ops); 1257 if (!err) 1258 ((struct seq_file *) file->private_data)->private = serv; 1259 return err; 1260 } 1261 EXPORT_SYMBOL(svc_pool_stats_open); 1262 1263 /*----------------------------------------------------------------------------*/ 1264