1 /* $NetBSD: svc.c,v 1.21 2000/07/06 03:10:35 christos Exp $ */
4 * Sun RPC is a product of Sun Microsystems, Inc. and is provided for
5 * unrestricted use provided that this legend is included on all tape
6 * media and as a part of the software program in whole or part. Users
7 * may copy or modify Sun RPC without charge, but are not authorized
8 * to license or distribute it to anyone else except as part of a product or
9 * program developed by the user.
11 * SUN RPC IS PROVIDED AS IS WITH NO WARRANTIES OF ANY KIND INCLUDING THE
12 * WARRANTIES OF DESIGN, MERCHANTIBILITY AND FITNESS FOR A PARTICULAR
13 * PURPOSE, OR ARISING FROM A COURSE OF DEALING, USAGE OR TRADE PRACTICE.
15 * Sun RPC is provided with no support and without any obligation on the
16 * part of Sun Microsystems, Inc. to assist in its use, correction,
17 * modification or enhancement.
19 * SUN MICROSYSTEMS, INC. SHALL HAVE NO LIABILITY WITH RESPECT TO THE
20 * INFRINGEMENT OF COPYRIGHTS, TRADE SECRETS OR ANY PATENTS BY SUN RPC
21 * OR ANY PART THEREOF.
23 * In no event will Sun Microsystems, Inc. be liable for any lost revenue
24 * or profits or other special, indirect and consequential damages, even if
25 * Sun has been advised of the possibility of such damages.
27 * Sun Microsystems, Inc.
29 * Mountain View, California 94043
32 #if defined(LIBC_SCCS) && !defined(lint)
33 static char *sccsid2 = "@(#)svc.c 1.44 88/02/08 Copyr 1984 Sun Micro";
34 static char *sccsid = "@(#)svc.c 2.4 88/08/11 4.0 RPCSRC";
36 #include <sys/cdefs.h>
37 __FBSDID("$FreeBSD$");
40 * svc.c, Server-side remote procedure call interface.
42 * There are two sets of procedures here. The xprt routines are
43 * for handling transport handles. The svc routines handle the
44 * list of service routines.
46 * Copyright (C) 1984, Sun Microsystems, Inc.
49 #include <sys/param.h>
51 #include <sys/kernel.h>
52 #include <sys/kthread.h>
53 #include <sys/malloc.h>
55 #include <sys/mutex.h>
57 #include <sys/queue.h>
58 #include <sys/socketvar.h>
59 #include <sys/systm.h>
60 #include <sys/ucred.h>
63 #include <rpc/rpcb_clnt.h>
64 #include <rpc/replay.h>
66 #include <rpc/rpc_com.h>
68 #define SVC_VERSQUIET 0x0001 /* keep quiet about vers mismatch */
69 #define version_keepquiet(xp) (SVC_EXT(xp)->xp_flags & SVC_VERSQUIET)
71 static struct svc_callout *svc_find(SVCPOOL *pool, rpcprog_t, rpcvers_t,
73 static void svc_new_thread(SVCPOOL *pool);
74 static void xprt_unregister_locked(SVCXPRT *xprt);
76 /* *************** SVCXPRT related stuff **************** */
78 static int svcpool_minthread_sysctl(SYSCTL_HANDLER_ARGS);
79 static int svcpool_maxthread_sysctl(SYSCTL_HANDLER_ARGS);
82 svcpool_create(const char *name, struct sysctl_oid_list *sysctl_base)
86 pool = malloc(sizeof(SVCPOOL), M_RPC, M_WAITOK|M_ZERO);
88 mtx_init(&pool->sp_lock, "sp_lock", NULL, MTX_DEF);
90 pool->sp_state = SVCPOOL_INIT;
92 TAILQ_INIT(&pool->sp_xlist);
93 TAILQ_INIT(&pool->sp_active);
94 TAILQ_INIT(&pool->sp_callouts);
95 LIST_INIT(&pool->sp_threads);
96 LIST_INIT(&pool->sp_idlethreads);
97 pool->sp_minthreads = 1;
98 pool->sp_maxthreads = 1;
99 pool->sp_threadcount = 0;
102 * Don't use more than a quarter of mbuf clusters or more than
103 * 45Mb buffering requests.
105 pool->sp_space_high = nmbclusters * MCLBYTES / 4;
106 if (pool->sp_space_high > 45 << 20)
107 pool->sp_space_high = 45 << 20;
108 pool->sp_space_low = 2 * pool->sp_space_high / 3;
110 sysctl_ctx_init(&pool->sp_sysctl);
112 SYSCTL_ADD_PROC(&pool->sp_sysctl, sysctl_base, OID_AUTO,
113 "minthreads", CTLTYPE_INT | CTLFLAG_RW,
114 pool, 0, svcpool_minthread_sysctl, "I", "");
115 SYSCTL_ADD_PROC(&pool->sp_sysctl, sysctl_base, OID_AUTO,
116 "maxthreads", CTLTYPE_INT | CTLFLAG_RW,
117 pool, 0, svcpool_maxthread_sysctl, "I", "");
118 SYSCTL_ADD_INT(&pool->sp_sysctl, sysctl_base, OID_AUTO,
119 "threads", CTLFLAG_RD, &pool->sp_threadcount, 0, "");
121 SYSCTL_ADD_UINT(&pool->sp_sysctl, sysctl_base, OID_AUTO,
122 "request_space_used", CTLFLAG_RD,
123 &pool->sp_space_used, 0,
124 "Space in parsed but not handled requests.");
126 SYSCTL_ADD_UINT(&pool->sp_sysctl, sysctl_base, OID_AUTO,
127 "request_space_used_highest", CTLFLAG_RD,
128 &pool->sp_space_used_highest, 0,
129 "Highest space used since reboot.");
131 SYSCTL_ADD_UINT(&pool->sp_sysctl, sysctl_base, OID_AUTO,
132 "request_space_high", CTLFLAG_RW,
133 &pool->sp_space_high, 0,
134 "Maximum space in parsed but not handled requests.");
136 SYSCTL_ADD_UINT(&pool->sp_sysctl, sysctl_base, OID_AUTO,
137 "request_space_low", CTLFLAG_RW,
138 &pool->sp_space_low, 0,
139 "Low water mark for request space.");
141 SYSCTL_ADD_INT(&pool->sp_sysctl, sysctl_base, OID_AUTO,
142 "request_space_throttled", CTLFLAG_RD,
143 &pool->sp_space_throttled, 0,
144 "Whether nfs requests are currently throttled");
146 SYSCTL_ADD_INT(&pool->sp_sysctl, sysctl_base, OID_AUTO,
147 "request_space_throttle_count", CTLFLAG_RD,
148 &pool->sp_space_throttle_count, 0,
149 "Count of times throttling based on request space has occurred");
156 svcpool_destroy(SVCPOOL *pool)
158 SVCXPRT *xprt, *nxprt;
159 struct svc_callout *s;
160 struct svcxprt_list cleanup;
162 TAILQ_INIT(&cleanup);
163 mtx_lock(&pool->sp_lock);
165 while (TAILQ_FIRST(&pool->sp_xlist)) {
166 xprt = TAILQ_FIRST(&pool->sp_xlist);
167 xprt_unregister_locked(xprt);
168 TAILQ_INSERT_TAIL(&cleanup, xprt, xp_link);
171 while (TAILQ_FIRST(&pool->sp_callouts)) {
172 s = TAILQ_FIRST(&pool->sp_callouts);
173 mtx_unlock(&pool->sp_lock);
174 svc_unreg(pool, s->sc_prog, s->sc_vers);
175 mtx_lock(&pool->sp_lock);
177 mtx_unlock(&pool->sp_lock);
179 TAILQ_FOREACH_SAFE(xprt, &cleanup, xp_link, nxprt) {
183 mtx_destroy(&pool->sp_lock);
186 replay_freecache(pool->sp_rcache);
188 sysctl_ctx_free(&pool->sp_sysctl);
193 svcpool_active(SVCPOOL *pool)
195 enum svcpool_state state = pool->sp_state;
197 if (state == SVCPOOL_INIT || state == SVCPOOL_CLOSING)
203 * Sysctl handler to set the minimum thread count on a pool
206 svcpool_minthread_sysctl(SYSCTL_HANDLER_ARGS)
209 int newminthreads, error, n;
211 pool = oidp->oid_arg1;
212 newminthreads = pool->sp_minthreads;
213 error = sysctl_handle_int(oidp, &newminthreads, 0, req);
214 if (error == 0 && newminthreads != pool->sp_minthreads) {
215 if (newminthreads > pool->sp_maxthreads)
217 mtx_lock(&pool->sp_lock);
218 if (newminthreads > pool->sp_minthreads
219 && svcpool_active(pool)) {
221 * If the pool is running and we are
222 * increasing, create some more threads now.
224 n = newminthreads - pool->sp_threadcount;
226 mtx_unlock(&pool->sp_lock);
228 svc_new_thread(pool);
229 mtx_lock(&pool->sp_lock);
232 pool->sp_minthreads = newminthreads;
233 mtx_unlock(&pool->sp_lock);
239 * Sysctl handler to set the maximum thread count on a pool
242 svcpool_maxthread_sysctl(SYSCTL_HANDLER_ARGS)
246 int newmaxthreads, error;
248 pool = oidp->oid_arg1;
249 newmaxthreads = pool->sp_maxthreads;
250 error = sysctl_handle_int(oidp, &newmaxthreads, 0, req);
251 if (error == 0 && newmaxthreads != pool->sp_maxthreads) {
252 if (newmaxthreads < pool->sp_minthreads)
254 mtx_lock(&pool->sp_lock);
255 if (newmaxthreads < pool->sp_maxthreads
256 && svcpool_active(pool)) {
258 * If the pool is running and we are
259 * decreasing, wake up some idle threads to
260 * encourage them to exit.
262 LIST_FOREACH(st, &pool->sp_idlethreads, st_ilink)
263 cv_signal(&st->st_cond);
265 pool->sp_maxthreads = newmaxthreads;
266 mtx_unlock(&pool->sp_lock);
272 * Activate a transport handle.
275 xprt_register(SVCXPRT *xprt)
277 SVCPOOL *pool = xprt->xp_pool;
280 mtx_lock(&pool->sp_lock);
281 xprt->xp_registered = TRUE;
282 xprt->xp_active = FALSE;
283 TAILQ_INSERT_TAIL(&pool->sp_xlist, xprt, xp_link);
284 mtx_unlock(&pool->sp_lock);
288 * De-activate a transport handle. Note: the locked version doesn't
289 * release the transport - caller must do that after dropping the pool
293 xprt_unregister_locked(SVCXPRT *xprt)
295 SVCPOOL *pool = xprt->xp_pool;
297 KASSERT(xprt->xp_registered == TRUE,
298 ("xprt_unregister_locked: not registered"));
299 if (xprt->xp_active) {
300 TAILQ_REMOVE(&pool->sp_active, xprt, xp_alink);
301 xprt->xp_active = FALSE;
303 TAILQ_REMOVE(&pool->sp_xlist, xprt, xp_link);
304 xprt->xp_registered = FALSE;
308 xprt_unregister(SVCXPRT *xprt)
310 SVCPOOL *pool = xprt->xp_pool;
312 mtx_lock(&pool->sp_lock);
313 if (xprt->xp_registered == FALSE) {
314 /* Already unregistered by another thread */
315 mtx_unlock(&pool->sp_lock);
318 xprt_unregister_locked(xprt);
319 mtx_unlock(&pool->sp_lock);
325 xprt_assignthread(SVCXPRT *xprt)
327 SVCPOOL *pool = xprt->xp_pool;
331 * Attempt to assign a service thread to this
334 LIST_FOREACH(st, &pool->sp_idlethreads, st_ilink) {
335 if (st->st_xprt == NULL && STAILQ_EMPTY(&st->st_reqs))
340 xprt->xp_thread = st;
342 cv_signal(&st->st_cond);
345 * See if we can create a new thread. The
346 * actual thread creation happens in
347 * svc_run_internal because our locking state
348 * is poorly defined (we are typically called
349 * from a socket upcall). Don't create more
350 * than one thread per second.
352 if (pool->sp_state == SVCPOOL_ACTIVE
353 && pool->sp_lastcreatetime < time_uptime
354 && pool->sp_threadcount < pool->sp_maxthreads) {
355 pool->sp_state = SVCPOOL_THREADWANTED;
361 xprt_active(SVCXPRT *xprt)
363 SVCPOOL *pool = xprt->xp_pool;
365 mtx_lock(&pool->sp_lock);
367 if (!xprt->xp_registered) {
369 * Race with xprt_unregister - we lose.
371 mtx_unlock(&pool->sp_lock);
375 if (!xprt->xp_active) {
376 TAILQ_INSERT_TAIL(&pool->sp_active, xprt, xp_alink);
377 xprt->xp_active = TRUE;
378 xprt_assignthread(xprt);
381 mtx_unlock(&pool->sp_lock);
385 xprt_inactive_locked(SVCXPRT *xprt)
387 SVCPOOL *pool = xprt->xp_pool;
389 if (xprt->xp_active) {
390 TAILQ_REMOVE(&pool->sp_active, xprt, xp_alink);
391 xprt->xp_active = FALSE;
396 xprt_inactive(SVCXPRT *xprt)
398 SVCPOOL *pool = xprt->xp_pool;
400 mtx_lock(&pool->sp_lock);
401 xprt_inactive_locked(xprt);
402 mtx_unlock(&pool->sp_lock);
406 * Add a service program to the callout list.
407 * The dispatch routine will be called when a rpc request for this
408 * program number comes in.
411 svc_reg(SVCXPRT *xprt, const rpcprog_t prog, const rpcvers_t vers,
412 void (*dispatch)(struct svc_req *, SVCXPRT *),
413 const struct netconfig *nconf)
415 SVCPOOL *pool = xprt->xp_pool;
416 struct svc_callout *s;
420 /* VARIABLES PROTECTED BY svc_lock: s, svc_head */
422 if (xprt->xp_netid) {
423 netid = strdup(xprt->xp_netid, M_RPC);
425 } else if (nconf && nconf->nc_netid) {
426 netid = strdup(nconf->nc_netid, M_RPC);
428 } /* must have been created with svc_raw_create */
429 if ((netid == NULL) && (flag == 1)) {
433 mtx_lock(&pool->sp_lock);
434 if ((s = svc_find(pool, prog, vers, netid)) != NULL) {
437 if (s->sc_dispatch == dispatch)
438 goto rpcb_it; /* he is registering another xptr */
439 mtx_unlock(&pool->sp_lock);
442 s = malloc(sizeof (struct svc_callout), M_RPC, M_NOWAIT);
446 mtx_unlock(&pool->sp_lock);
452 s->sc_dispatch = dispatch;
454 TAILQ_INSERT_TAIL(&pool->sp_callouts, s, sc_link);
456 if ((xprt->xp_netid == NULL) && (flag == 1) && netid)
457 ((SVCXPRT *) xprt)->xp_netid = strdup(netid, M_RPC);
460 mtx_unlock(&pool->sp_lock);
461 /* now register the information with the local binder service */
464 struct netconfig tnc;
467 nb.buf = &xprt->xp_ltaddr;
468 nb.len = xprt->xp_ltaddr.ss_len;
469 dummy = rpcb_set(prog, vers, &tnc, &nb);
476 * Remove a service program from the callout list.
479 svc_unreg(SVCPOOL *pool, const rpcprog_t prog, const rpcvers_t vers)
481 struct svc_callout *s;
483 /* unregister the information anyway */
484 (void) rpcb_unset(prog, vers, NULL);
485 mtx_lock(&pool->sp_lock);
486 while ((s = svc_find(pool, prog, vers, NULL)) != NULL) {
487 TAILQ_REMOVE(&pool->sp_callouts, s, sc_link);
489 mem_free(s->sc_netid, sizeof (s->sc_netid) + 1);
490 mem_free(s, sizeof (struct svc_callout));
492 mtx_unlock(&pool->sp_lock);
495 /* ********************** CALLOUT list related stuff ************* */
498 * Search the callout list for a program number, return the callout
501 static struct svc_callout *
502 svc_find(SVCPOOL *pool, rpcprog_t prog, rpcvers_t vers, char *netid)
504 struct svc_callout *s;
506 mtx_assert(&pool->sp_lock, MA_OWNED);
507 TAILQ_FOREACH(s, &pool->sp_callouts, sc_link) {
508 if (s->sc_prog == prog && s->sc_vers == vers
509 && (netid == NULL || s->sc_netid == NULL ||
510 strcmp(netid, s->sc_netid) == 0))
517 /* ******************* REPLY GENERATION ROUTINES ************ */
520 svc_sendreply_common(struct svc_req *rqstp, struct rpc_msg *rply,
523 SVCXPRT *xprt = rqstp->rq_xprt;
526 if (rqstp->rq_args) {
527 m_freem(rqstp->rq_args);
528 rqstp->rq_args = NULL;
531 if (xprt->xp_pool->sp_rcache)
532 replay_setreply(xprt->xp_pool->sp_rcache,
533 rply, svc_getrpccaller(rqstp), body);
535 if (!SVCAUTH_WRAP(&rqstp->rq_auth, &body))
538 ok = SVC_REPLY(xprt, rply, rqstp->rq_addr, body);
539 if (rqstp->rq_addr) {
540 free(rqstp->rq_addr, M_SONAME);
541 rqstp->rq_addr = NULL;
548 * Send a reply to an rpc request
551 svc_sendreply(struct svc_req *rqstp, xdrproc_t xdr_results, void * xdr_location)
558 rply.rm_xid = rqstp->rq_xid;
559 rply.rm_direction = REPLY;
560 rply.rm_reply.rp_stat = MSG_ACCEPTED;
561 rply.acpted_rply.ar_verf = rqstp->rq_verf;
562 rply.acpted_rply.ar_stat = SUCCESS;
563 rply.acpted_rply.ar_results.where = NULL;
564 rply.acpted_rply.ar_results.proc = (xdrproc_t) xdr_void;
566 MGET(m, M_WAITOK, MT_DATA);
569 xdrmbuf_create(&xdrs, m, XDR_ENCODE);
570 ok = xdr_results(&xdrs, xdr_location);
574 return (svc_sendreply_common(rqstp, &rply, m));
582 svc_sendreply_mbuf(struct svc_req *rqstp, struct mbuf *m)
586 rply.rm_xid = rqstp->rq_xid;
587 rply.rm_direction = REPLY;
588 rply.rm_reply.rp_stat = MSG_ACCEPTED;
589 rply.acpted_rply.ar_verf = rqstp->rq_verf;
590 rply.acpted_rply.ar_stat = SUCCESS;
591 rply.acpted_rply.ar_results.where = NULL;
592 rply.acpted_rply.ar_results.proc = (xdrproc_t) xdr_void;
594 return (svc_sendreply_common(rqstp, &rply, m));
598 * No procedure error reply
601 svcerr_noproc(struct svc_req *rqstp)
603 SVCXPRT *xprt = rqstp->rq_xprt;
606 rply.rm_xid = rqstp->rq_xid;
607 rply.rm_direction = REPLY;
608 rply.rm_reply.rp_stat = MSG_ACCEPTED;
609 rply.acpted_rply.ar_verf = rqstp->rq_verf;
610 rply.acpted_rply.ar_stat = PROC_UNAVAIL;
612 if (xprt->xp_pool->sp_rcache)
613 replay_setreply(xprt->xp_pool->sp_rcache,
614 &rply, svc_getrpccaller(rqstp), NULL);
616 svc_sendreply_common(rqstp, &rply, NULL);
620 * Can't decode args error reply
623 svcerr_decode(struct svc_req *rqstp)
625 SVCXPRT *xprt = rqstp->rq_xprt;
628 rply.rm_xid = rqstp->rq_xid;
629 rply.rm_direction = REPLY;
630 rply.rm_reply.rp_stat = MSG_ACCEPTED;
631 rply.acpted_rply.ar_verf = rqstp->rq_verf;
632 rply.acpted_rply.ar_stat = GARBAGE_ARGS;
634 if (xprt->xp_pool->sp_rcache)
635 replay_setreply(xprt->xp_pool->sp_rcache,
636 &rply, (struct sockaddr *) &xprt->xp_rtaddr, NULL);
638 svc_sendreply_common(rqstp, &rply, NULL);
645 svcerr_systemerr(struct svc_req *rqstp)
647 SVCXPRT *xprt = rqstp->rq_xprt;
650 rply.rm_xid = rqstp->rq_xid;
651 rply.rm_direction = REPLY;
652 rply.rm_reply.rp_stat = MSG_ACCEPTED;
653 rply.acpted_rply.ar_verf = rqstp->rq_verf;
654 rply.acpted_rply.ar_stat = SYSTEM_ERR;
656 if (xprt->xp_pool->sp_rcache)
657 replay_setreply(xprt->xp_pool->sp_rcache,
658 &rply, svc_getrpccaller(rqstp), NULL);
660 svc_sendreply_common(rqstp, &rply, NULL);
664 * Authentication error reply
667 svcerr_auth(struct svc_req *rqstp, enum auth_stat why)
669 SVCXPRT *xprt = rqstp->rq_xprt;
672 rply.rm_xid = rqstp->rq_xid;
673 rply.rm_direction = REPLY;
674 rply.rm_reply.rp_stat = MSG_DENIED;
675 rply.rjcted_rply.rj_stat = AUTH_ERROR;
676 rply.rjcted_rply.rj_why = why;
678 if (xprt->xp_pool->sp_rcache)
679 replay_setreply(xprt->xp_pool->sp_rcache,
680 &rply, svc_getrpccaller(rqstp), NULL);
682 svc_sendreply_common(rqstp, &rply, NULL);
686 * Auth too weak error reply
689 svcerr_weakauth(struct svc_req *rqstp)
692 svcerr_auth(rqstp, AUTH_TOOWEAK);
696 * Program unavailable error reply
699 svcerr_noprog(struct svc_req *rqstp)
701 SVCXPRT *xprt = rqstp->rq_xprt;
704 rply.rm_xid = rqstp->rq_xid;
705 rply.rm_direction = REPLY;
706 rply.rm_reply.rp_stat = MSG_ACCEPTED;
707 rply.acpted_rply.ar_verf = rqstp->rq_verf;
708 rply.acpted_rply.ar_stat = PROG_UNAVAIL;
710 if (xprt->xp_pool->sp_rcache)
711 replay_setreply(xprt->xp_pool->sp_rcache,
712 &rply, svc_getrpccaller(rqstp), NULL);
714 svc_sendreply_common(rqstp, &rply, NULL);
718 * Program version mismatch error reply
721 svcerr_progvers(struct svc_req *rqstp, rpcvers_t low_vers, rpcvers_t high_vers)
723 SVCXPRT *xprt = rqstp->rq_xprt;
726 rply.rm_xid = rqstp->rq_xid;
727 rply.rm_direction = REPLY;
728 rply.rm_reply.rp_stat = MSG_ACCEPTED;
729 rply.acpted_rply.ar_verf = rqstp->rq_verf;
730 rply.acpted_rply.ar_stat = PROG_MISMATCH;
731 rply.acpted_rply.ar_vers.low = (uint32_t)low_vers;
732 rply.acpted_rply.ar_vers.high = (uint32_t)high_vers;
734 if (xprt->xp_pool->sp_rcache)
735 replay_setreply(xprt->xp_pool->sp_rcache,
736 &rply, svc_getrpccaller(rqstp), NULL);
738 svc_sendreply_common(rqstp, &rply, NULL);
742 * Allocate a new server transport structure. All fields are
743 * initialized to zero and xp_p3 is initialized to point at an
744 * extension structure to hold various flags and authentication
753 xprt = mem_alloc(sizeof(SVCXPRT));
754 memset(xprt, 0, sizeof(SVCXPRT));
755 ext = mem_alloc(sizeof(SVCXPRT_EXT));
756 memset(ext, 0, sizeof(SVCXPRT_EXT));
758 refcount_init(&xprt->xp_refs, 1);
764 * Free a server transport structure.
771 mem_free(xprt->xp_p3, sizeof(SVCXPRT_EXT));
772 mem_free(xprt, sizeof(SVCXPRT));
775 /* ******************* SERVER INPUT STUFF ******************* */
778 * Read RPC requests from a transport and queue them to be
779 * executed. We handle authentication and replay cache replies here.
780 * Actually dispatching the RPC is deferred till svc_executereq.
782 static enum xprt_stat
783 svc_getreq(SVCXPRT *xprt, struct svc_req **rqstp_ret)
785 SVCPOOL *pool = xprt->xp_pool;
791 /* now receive msgs from xprtprt (support batch calls) */
792 r = malloc(sizeof(*r), M_RPC, M_WAITOK|M_ZERO);
794 msg.rm_call.cb_cred.oa_base = r->rq_credarea;
795 msg.rm_call.cb_verf.oa_base = &r->rq_credarea[MAX_AUTH_BYTES];
796 r->rq_clntcred = &r->rq_credarea[2*MAX_AUTH_BYTES];
797 if (SVC_RECV(xprt, &msg, &r->rq_addr, &args)) {
801 * Handle replays and authenticate before queuing the
802 * request to be executed.
806 if (pool->sp_rcache) {
807 struct rpc_msg repmsg;
808 struct mbuf *repbody;
809 enum replay_state rs;
810 rs = replay_find(pool->sp_rcache, &msg,
811 svc_getrpccaller(r), &repmsg, &repbody);
816 SVC_REPLY(xprt, &repmsg, r->rq_addr,
819 free(r->rq_addr, M_SONAME);
831 r->rq_xid = msg.rm_xid;
832 r->rq_prog = msg.rm_call.cb_prog;
833 r->rq_vers = msg.rm_call.cb_vers;
834 r->rq_proc = msg.rm_call.cb_proc;
835 r->rq_size = sizeof(*r) + m_length(args, NULL);
837 if ((why = _authenticate(r, &msg)) != AUTH_OK) {
839 * RPCSEC_GSS uses this return code
840 * for requests that form part of its
841 * context establishment protocol and
842 * should not be dispatched to the
845 if (why != RPCSEC_GSS_NODISPATCH)
850 if (!SVCAUTH_UNWRAP(&r->rq_auth, &r->rq_args)) {
856 * Everything checks out, return request to caller.
866 if ((stat = SVC_STAT(xprt)) == XPRT_DIED) {
867 xprt_unregister(xprt);
874 svc_executereq(struct svc_req *rqstp)
876 SVCXPRT *xprt = rqstp->rq_xprt;
877 SVCPOOL *pool = xprt->xp_pool;
881 struct svc_callout *s;
883 /* now match message with a registered service*/
885 low_vers = (rpcvers_t) -1L;
886 high_vers = (rpcvers_t) 0L;
887 TAILQ_FOREACH(s, &pool->sp_callouts, sc_link) {
888 if (s->sc_prog == rqstp->rq_prog) {
889 if (s->sc_vers == rqstp->rq_vers) {
891 * We hand ownership of r to the
892 * dispatch method - they must call
895 (*s->sc_dispatch)(rqstp, xprt);
897 } /* found correct version */
899 if (s->sc_vers < low_vers)
900 low_vers = s->sc_vers;
901 if (s->sc_vers > high_vers)
902 high_vers = s->sc_vers;
903 } /* found correct program */
907 * if we got here, the program or version
911 svcerr_progvers(rqstp, low_vers, high_vers);
913 svcerr_noprog(rqstp);
919 svc_checkidle(SVCPOOL *pool)
921 SVCXPRT *xprt, *nxprt;
923 struct svcxprt_list cleanup;
925 TAILQ_INIT(&cleanup);
926 TAILQ_FOREACH_SAFE(xprt, &pool->sp_xlist, xp_link, nxprt) {
928 * Only some transports have idle timers. Don't time
929 * something out which is just waking up.
931 if (!xprt->xp_idletimeout || xprt->xp_thread)
934 timo = xprt->xp_lastactive + xprt->xp_idletimeout;
935 if (time_uptime > timo) {
936 xprt_unregister_locked(xprt);
937 TAILQ_INSERT_TAIL(&cleanup, xprt, xp_link);
941 mtx_unlock(&pool->sp_lock);
942 TAILQ_FOREACH_SAFE(xprt, &cleanup, xp_link, nxprt) {
945 mtx_lock(&pool->sp_lock);
950 svc_assign_waiting_sockets(SVCPOOL *pool)
954 TAILQ_FOREACH(xprt, &pool->sp_active, xp_alink) {
955 if (!xprt->xp_thread) {
956 xprt_assignthread(xprt);
962 svc_request_space_available(SVCPOOL *pool)
965 mtx_assert(&pool->sp_lock, MA_OWNED);
967 if (pool->sp_space_throttled) {
969 * Below the low-water yet? If so, assign any waiting sockets.
971 if (pool->sp_space_used < pool->sp_space_low) {
972 pool->sp_space_throttled = FALSE;
973 svc_assign_waiting_sockets(pool);
979 if (pool->sp_space_used
980 >= pool->sp_space_high) {
981 pool->sp_space_throttled = TRUE;
982 pool->sp_space_throttle_count++;
991 svc_run_internal(SVCPOOL *pool, bool_t ismaster)
993 SVCTHREAD *st, *stpref;
996 struct svc_req *rqstp;
999 st = mem_alloc(sizeof(*st));
1001 STAILQ_INIT(&st->st_reqs);
1002 cv_init(&st->st_cond, "rpcsvc");
1004 mtx_lock(&pool->sp_lock);
1005 LIST_INSERT_HEAD(&pool->sp_threads, st, st_link);
1008 * If we are a new thread which was spawned to cope with
1009 * increased load, set the state back to SVCPOOL_ACTIVE.
1011 if (pool->sp_state == SVCPOOL_THREADSTARTING)
1012 pool->sp_state = SVCPOOL_ACTIVE;
1014 while (pool->sp_state != SVCPOOL_CLOSING) {
1016 * Check for idle transports once per second.
1018 if (time_uptime > pool->sp_lastidlecheck) {
1019 pool->sp_lastidlecheck = time_uptime;
1020 svc_checkidle(pool);
1024 if (!xprt && STAILQ_EMPTY(&st->st_reqs)) {
1026 * Enforce maxthreads count.
1028 if (pool->sp_threadcount > pool->sp_maxthreads)
1032 * Before sleeping, see if we can find an
1033 * active transport which isn't being serviced
1036 if (svc_request_space_available(pool)) {
1037 TAILQ_FOREACH(xprt, &pool->sp_active,
1039 if (!xprt->xp_thread) {
1041 xprt->xp_thread = st;
1050 LIST_INSERT_HEAD(&pool->sp_idlethreads, st, st_ilink);
1051 error = cv_timedwait_sig(&st->st_cond, &pool->sp_lock,
1053 LIST_REMOVE(st, st_ilink);
1056 * Reduce worker thread count when idle.
1058 if (error == EWOULDBLOCK) {
1060 && (pool->sp_threadcount
1061 > pool->sp_minthreads)
1063 && STAILQ_EMPTY(&st->st_reqs))
1066 if (error == EWOULDBLOCK)
1069 if (pool->sp_state != SVCPOOL_CLOSING) {
1070 mtx_unlock(&pool->sp_lock);
1072 mtx_lock(&pool->sp_lock);
1077 if (pool->sp_state == SVCPOOL_THREADWANTED) {
1078 pool->sp_state = SVCPOOL_THREADSTARTING;
1079 pool->sp_lastcreatetime = time_uptime;
1080 mtx_unlock(&pool->sp_lock);
1081 svc_new_thread(pool);
1082 mtx_lock(&pool->sp_lock);
1089 * Drain the transport socket and queue up any
1092 xprt->xp_lastactive = time_uptime;
1095 if (!svc_request_space_available(pool))
1098 mtx_unlock(&pool->sp_lock);
1099 stat = svc_getreq(xprt, &rqstp);
1100 mtx_lock(&pool->sp_lock);
1103 * See if the application has
1104 * a preference for some other
1108 if (pool->sp_assign)
1109 stpref = pool->sp_assign(st,
1112 pool->sp_space_used +=
1114 if (pool->sp_space_used
1115 > pool->sp_space_used_highest)
1116 pool->sp_space_used_highest =
1117 pool->sp_space_used;
1118 rqstp->rq_thread = stpref;
1119 STAILQ_INSERT_TAIL(&stpref->st_reqs,
1121 stpref->st_reqcount++;
1124 * If we assigned the request
1125 * to another thread, make
1126 * sure its awake and continue
1128 * socket. Otherwise, try to
1129 * find some other thread to
1130 * read from the socket and
1131 * execute the request
1135 cv_signal(&stpref->st_cond);
1141 } while (stat == XPRT_MOREREQS
1142 && pool->sp_state != SVCPOOL_CLOSING);
1145 * Move this transport to the end of the
1146 * active list to ensure fairness when
1147 * multiple transports are active. If this was
1148 * the last queued request, svc_getreq will
1149 * end up calling xprt_inactive to remove from
1152 xprt->xp_thread = NULL;
1154 if (xprt->xp_active) {
1155 xprt_assignthread(xprt);
1156 TAILQ_REMOVE(&pool->sp_active, xprt, xp_alink);
1157 TAILQ_INSERT_TAIL(&pool->sp_active, xprt,
1160 mtx_unlock(&pool->sp_lock);
1162 mtx_lock(&pool->sp_lock);
1166 * Execute what we have queued.
1168 while ((rqstp = STAILQ_FIRST(&st->st_reqs)) != NULL) {
1169 size_t sz = rqstp->rq_size;
1170 mtx_unlock(&pool->sp_lock);
1171 svc_executereq(rqstp);
1172 mtx_lock(&pool->sp_lock);
1173 pool->sp_space_used -= sz;
1183 KASSERT(STAILQ_EMPTY(&st->st_reqs), ("stray reqs on exit"));
1184 LIST_REMOVE(st, st_link);
1185 pool->sp_threadcount--;
1187 mtx_unlock(&pool->sp_lock);
1189 cv_destroy(&st->st_cond);
1190 mem_free(st, sizeof(*st));
1197 svc_thread_start(void *arg)
1200 svc_run_internal((SVCPOOL *) arg, FALSE);
1205 svc_new_thread(SVCPOOL *pool)
1209 pool->sp_threadcount++;
1210 kthread_add(svc_thread_start, pool,
1211 pool->sp_proc, &td, 0, 0,
1212 "%s: service", pool->sp_name);
1216 svc_run(SVCPOOL *pool)
1224 snprintf(td->td_name, sizeof(td->td_name),
1225 "%s: master", pool->sp_name);
1226 pool->sp_state = SVCPOOL_ACTIVE;
1228 pool->sp_lastcreatetime = time_uptime;
1229 pool->sp_threadcount = 1;
1231 for (i = 1; i < pool->sp_minthreads; i++) {
1232 svc_new_thread(pool);
1235 svc_run_internal(pool, TRUE);
1237 mtx_lock(&pool->sp_lock);
1238 while (pool->sp_threadcount > 0)
1239 msleep(pool, &pool->sp_lock, 0, "svcexit", 0);
1240 mtx_unlock(&pool->sp_lock);
1244 svc_exit(SVCPOOL *pool)
1248 mtx_lock(&pool->sp_lock);
1250 pool->sp_state = SVCPOOL_CLOSING;
1251 LIST_FOREACH(st, &pool->sp_idlethreads, st_ilink)
1252 cv_signal(&st->st_cond);
1254 mtx_unlock(&pool->sp_lock);
1258 svc_getargs(struct svc_req *rqstp, xdrproc_t xargs, void *args)
1265 rqstp->rq_args = NULL;
1267 xdrmbuf_create(&xdrs, m, XDR_DECODE);
1268 stat = xargs(&xdrs, args);
1275 svc_freeargs(struct svc_req *rqstp, xdrproc_t xargs, void *args)
1279 if (rqstp->rq_addr) {
1280 free(rqstp->rq_addr, M_SONAME);
1281 rqstp->rq_addr = NULL;
1284 xdrs.x_op = XDR_FREE;
1285 return (xargs(&xdrs, args));
1289 svc_freereq(struct svc_req *rqstp)
1295 st = rqstp->rq_thread;
1296 xprt = rqstp->rq_xprt;
1298 pool = xprt->xp_pool;
1302 mtx_lock(&pool->sp_lock);
1303 KASSERT(rqstp == STAILQ_FIRST(&st->st_reqs),
1304 ("Freeing request out of order"));
1305 STAILQ_REMOVE_HEAD(&st->st_reqs, rq_link);
1308 pool->sp_done(st, rqstp);
1309 mtx_unlock(&pool->sp_lock);
1312 if (rqstp->rq_auth.svc_ah_ops)
1313 SVCAUTH_RELEASE(&rqstp->rq_auth);
1315 if (rqstp->rq_xprt) {
1316 SVC_RELEASE(rqstp->rq_xprt);
1320 free(rqstp->rq_addr, M_SONAME);
1323 m_freem(rqstp->rq_args);