FreeBSD/Linux Kernel Cross Reference
sys/rpc/svc.c
1 /* $NetBSD: svc.c,v 1.21 2000/07/06 03:10:35 christos Exp $ */
2
3 /*
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.
10 *
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.
14 *
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.
18 *
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.
22 *
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.
26 *
27 * Sun Microsystems, Inc.
28 * 2550 Garcia Avenue
29 * Mountain View, California 94043
30 */
31
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";
35 #endif
36 #include <sys/cdefs.h>
37 __FBSDID("$FreeBSD$");
38
39 /*
40 * svc.c, Server-side remote procedure call interface.
41 *
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.
45 *
46 * Copyright (C) 1984, Sun Microsystems, Inc.
47 */
48
49 #include <sys/param.h>
50 #include <sys/lock.h>
51 #include <sys/kernel.h>
52 #include <sys/kthread.h>
53 #include <sys/malloc.h>
54 #include <sys/mbuf.h>
55 #include <sys/mutex.h>
56 #include <sys/proc.h>
57 #include <sys/queue.h>
58 #include <sys/socketvar.h>
59 #include <sys/systm.h>
60 #include <sys/ucred.h>
61
62 #include <rpc/rpc.h>
63 #include <rpc/rpcb_clnt.h>
64 #include <rpc/replay.h>
65
66 #include <rpc/rpc_com.h>
67
68 #define SVC_VERSQUIET 0x0001 /* keep quiet about vers mismatch */
69 #define version_keepquiet(xp) (SVC_EXT(xp)->xp_flags & SVC_VERSQUIET)
70
71 static struct svc_callout *svc_find(SVCPOOL *pool, rpcprog_t, rpcvers_t,
72 char *);
73 static void svc_new_thread(SVCPOOL *pool);
74 static void xprt_unregister_locked(SVCXPRT *xprt);
75
76 /* *************** SVCXPRT related stuff **************** */
77
78 static int svcpool_minthread_sysctl(SYSCTL_HANDLER_ARGS);
79 static int svcpool_maxthread_sysctl(SYSCTL_HANDLER_ARGS);
80
81 SVCPOOL*
82 svcpool_create(const char *name, struct sysctl_oid_list *sysctl_base)
83 {
84 SVCPOOL *pool;
85
86 pool = malloc(sizeof(SVCPOOL), M_RPC, M_WAITOK|M_ZERO);
87
88 mtx_init(&pool->sp_lock, "sp_lock", NULL, MTX_DEF);
89 pool->sp_name = name;
90 pool->sp_state = SVCPOOL_INIT;
91 pool->sp_proc = NULL;
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;
100
101 /*
102 * Don't use more than a quarter of mbuf clusters or more than
103 * 45Mb buffering requests.
104 */
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;
109
110 sysctl_ctx_init(&pool->sp_sysctl);
111 if (sysctl_base) {
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, "");
120
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.");
125
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.");
130
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.");
135
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.");
140
141 SYSCTL_ADD_UINT(&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");
145
146 SYSCTL_ADD_UINT(&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");
150 }
151
152 return pool;
153 }
154
155 void
156 svcpool_destroy(SVCPOOL *pool)
157 {
158 SVCXPRT *xprt, *nxprt;
159 struct svc_callout *s;
160 struct svcxprt_list cleanup;
161
162 TAILQ_INIT(&cleanup);
163 mtx_lock(&pool->sp_lock);
164
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);
169 }
170
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);
176 }
177 mtx_unlock(&pool->sp_lock);
178
179 TAILQ_FOREACH_SAFE(xprt, &cleanup, xp_link, nxprt) {
180 SVC_RELEASE(xprt);
181 }
182
183 mtx_destroy(&pool->sp_lock);
184
185 if (pool->sp_rcache)
186 replay_freecache(pool->sp_rcache);
187
188 sysctl_ctx_free(&pool->sp_sysctl);
189 free(pool, M_RPC);
190 }
191
192 static bool_t
193 svcpool_active(SVCPOOL *pool)
194 {
195 enum svcpool_state state = pool->sp_state;
196
197 if (state == SVCPOOL_INIT || state == SVCPOOL_CLOSING)
198 return (FALSE);
199 return (TRUE);
200 }
201
202 /*
203 * Sysctl handler to set the minimum thread count on a pool
204 */
205 static int
206 svcpool_minthread_sysctl(SYSCTL_HANDLER_ARGS)
207 {
208 SVCPOOL *pool;
209 int newminthreads, error, n;
210
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)
216 return (EINVAL);
217 mtx_lock(&pool->sp_lock);
218 if (newminthreads > pool->sp_minthreads
219 && svcpool_active(pool)) {
220 /*
221 * If the pool is running and we are
222 * increasing, create some more threads now.
223 */
224 n = newminthreads - pool->sp_threadcount;
225 if (n > 0) {
226 mtx_unlock(&pool->sp_lock);
227 while (n--)
228 svc_new_thread(pool);
229 mtx_lock(&pool->sp_lock);
230 }
231 }
232 pool->sp_minthreads = newminthreads;
233 mtx_unlock(&pool->sp_lock);
234 }
235 return (error);
236 }
237
238 /*
239 * Sysctl handler to set the maximum thread count on a pool
240 */
241 static int
242 svcpool_maxthread_sysctl(SYSCTL_HANDLER_ARGS)
243 {
244 SVCPOOL *pool;
245 SVCTHREAD *st;
246 int newmaxthreads, error;
247
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)
253 return (EINVAL);
254 mtx_lock(&pool->sp_lock);
255 if (newmaxthreads < pool->sp_maxthreads
256 && svcpool_active(pool)) {
257 /*
258 * If the pool is running and we are
259 * decreasing, wake up some idle threads to
260 * encourage them to exit.
261 */
262 LIST_FOREACH(st, &pool->sp_idlethreads, st_ilink)
263 cv_signal(&st->st_cond);
264 }
265 pool->sp_maxthreads = newmaxthreads;
266 mtx_unlock(&pool->sp_lock);
267 }
268 return (error);
269 }
270
271 /*
272 * Activate a transport handle.
273 */
274 void
275 xprt_register(SVCXPRT *xprt)
276 {
277 SVCPOOL *pool = xprt->xp_pool;
278
279 SVC_ACQUIRE(xprt);
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);
285 }
286
287 /*
288 * De-activate a transport handle. Note: the locked version doesn't
289 * release the transport - caller must do that after dropping the pool
290 * lock.
291 */
292 static void
293 xprt_unregister_locked(SVCXPRT *xprt)
294 {
295 SVCPOOL *pool = xprt->xp_pool;
296
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;
302 }
303 TAILQ_REMOVE(&pool->sp_xlist, xprt, xp_link);
304 xprt->xp_registered = FALSE;
305 }
306
307 void
308 xprt_unregister(SVCXPRT *xprt)
309 {
310 SVCPOOL *pool = xprt->xp_pool;
311
312 mtx_lock(&pool->sp_lock);
313 if (xprt->xp_registered == FALSE) {
314 /* Already unregistered by another thread */
315 mtx_unlock(&pool->sp_lock);
316 return;
317 }
318 xprt_unregister_locked(xprt);
319 mtx_unlock(&pool->sp_lock);
320
321 SVC_RELEASE(xprt);
322 }
323
324 static void
325 xprt_assignthread(SVCXPRT *xprt)
326 {
327 SVCPOOL *pool = xprt->xp_pool;
328 SVCTHREAD *st;
329
330 /*
331 * Attempt to assign a service thread to this
332 * transport.
333 */
334 LIST_FOREACH(st, &pool->sp_idlethreads, st_ilink) {
335 if (st->st_xprt == NULL && STAILQ_EMPTY(&st->st_reqs))
336 break;
337 }
338 if (st) {
339 SVC_ACQUIRE(xprt);
340 xprt->xp_thread = st;
341 st->st_xprt = xprt;
342 cv_signal(&st->st_cond);
343 } else {
344 /*
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.
351 */
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;
356 }
357 }
358 }
359
360 void
361 xprt_active(SVCXPRT *xprt)
362 {
363 SVCPOOL *pool = xprt->xp_pool;
364
365 mtx_lock(&pool->sp_lock);
366
367 if (!xprt->xp_registered) {
368 /*
369 * Race with xprt_unregister - we lose.
370 */
371 mtx_unlock(&pool->sp_lock);
372 return;
373 }
374
375 if (!xprt->xp_active) {
376 TAILQ_INSERT_TAIL(&pool->sp_active, xprt, xp_alink);
377 xprt->xp_active = TRUE;
378 xprt_assignthread(xprt);
379 }
380
381 mtx_unlock(&pool->sp_lock);
382 }
383
384 void
385 xprt_inactive_locked(SVCXPRT *xprt)
386 {
387 SVCPOOL *pool = xprt->xp_pool;
388
389 if (xprt->xp_active) {
390 TAILQ_REMOVE(&pool->sp_active, xprt, xp_alink);
391 xprt->xp_active = FALSE;
392 }
393 }
394
395 void
396 xprt_inactive(SVCXPRT *xprt)
397 {
398 SVCPOOL *pool = xprt->xp_pool;
399
400 mtx_lock(&pool->sp_lock);
401 xprt_inactive_locked(xprt);
402 mtx_unlock(&pool->sp_lock);
403 }
404
405 /*
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.
409 */
410 bool_t
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)
414 {
415 SVCPOOL *pool = xprt->xp_pool;
416 struct svc_callout *s;
417 char *netid = NULL;
418 int flag = 0;
419
420 /* VARIABLES PROTECTED BY svc_lock: s, svc_head */
421
422 if (xprt->xp_netid) {
423 netid = strdup(xprt->xp_netid, M_RPC);
424 flag = 1;
425 } else if (nconf && nconf->nc_netid) {
426 netid = strdup(nconf->nc_netid, M_RPC);
427 flag = 1;
428 } /* must have been created with svc_raw_create */
429 if ((netid == NULL) && (flag == 1)) {
430 return (FALSE);
431 }
432
433 mtx_lock(&pool->sp_lock);
434 if ((s = svc_find(pool, prog, vers, netid)) != NULL) {
435 if (netid)
436 free(netid, M_RPC);
437 if (s->sc_dispatch == dispatch)
438 goto rpcb_it; /* he is registering another xptr */
439 mtx_unlock(&pool->sp_lock);
440 return (FALSE);
441 }
442 s = malloc(sizeof (struct svc_callout), M_RPC, M_NOWAIT);
443 if (s == NULL) {
444 if (netid)
445 free(netid, M_RPC);
446 mtx_unlock(&pool->sp_lock);
447 return (FALSE);
448 }
449
450 s->sc_prog = prog;
451 s->sc_vers = vers;
452 s->sc_dispatch = dispatch;
453 s->sc_netid = netid;
454 TAILQ_INSERT_TAIL(&pool->sp_callouts, s, sc_link);
455
456 if ((xprt->xp_netid == NULL) && (flag == 1) && netid)
457 ((SVCXPRT *) xprt)->xp_netid = strdup(netid, M_RPC);
458
459 rpcb_it:
460 mtx_unlock(&pool->sp_lock);
461 /* now register the information with the local binder service */
462 if (nconf) {
463 bool_t dummy;
464 struct netconfig tnc;
465 struct netbuf nb;
466 tnc = *nconf;
467 nb.buf = &xprt->xp_ltaddr;
468 nb.len = xprt->xp_ltaddr.ss_len;
469 dummy = rpcb_set(prog, vers, &tnc, &nb);
470 return (dummy);
471 }
472 return (TRUE);
473 }
474
475 /*
476 * Remove a service program from the callout list.
477 */
478 void
479 svc_unreg(SVCPOOL *pool, const rpcprog_t prog, const rpcvers_t vers)
480 {
481 struct svc_callout *s;
482
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);
488 if (s->sc_netid)
489 mem_free(s->sc_netid, sizeof (s->sc_netid) + 1);
490 mem_free(s, sizeof (struct svc_callout));
491 }
492 mtx_unlock(&pool->sp_lock);
493 }
494
495 /* ********************** CALLOUT list related stuff ************* */
496
497 /*
498 * Search the callout list for a program number, return the callout
499 * struct.
500 */
501 static struct svc_callout *
502 svc_find(SVCPOOL *pool, rpcprog_t prog, rpcvers_t vers, char *netid)
503 {
504 struct svc_callout *s;
505
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))
511 break;
512 }
513
514 return (s);
515 }
516
517 /* ******************* REPLY GENERATION ROUTINES ************ */
518
519 static bool_t
520 svc_sendreply_common(struct svc_req *rqstp, struct rpc_msg *rply,
521 struct mbuf *body)
522 {
523 SVCXPRT *xprt = rqstp->rq_xprt;
524 bool_t ok;
525
526 if (rqstp->rq_args) {
527 m_freem(rqstp->rq_args);
528 rqstp->rq_args = NULL;
529 }
530
531 if (xprt->xp_pool->sp_rcache)
532 replay_setreply(xprt->xp_pool->sp_rcache,
533 rply, svc_getrpccaller(rqstp), body);
534
535 if (!SVCAUTH_WRAP(&rqstp->rq_auth, &body))
536 return (FALSE);
537
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;
542 }
543
544 return (ok);
545 }
546
547 /*
548 * Send a reply to an rpc request
549 */
550 bool_t
551 svc_sendreply(struct svc_req *rqstp, xdrproc_t xdr_results, void * xdr_location)
552 {
553 struct rpc_msg rply;
554 struct mbuf *m;
555 XDR xdrs;
556 bool_t ok;
557
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;
565
566 MGET(m, M_WAIT, MT_DATA);
567 MCLGET(m, M_WAIT);
568 m->m_len = 0;
569 xdrmbuf_create(&xdrs, m, XDR_ENCODE);
570 ok = xdr_results(&xdrs, xdr_location);
571 XDR_DESTROY(&xdrs);
572
573 if (ok) {
574 return (svc_sendreply_common(rqstp, &rply, m));
575 } else {
576 m_freem(m);
577 return (FALSE);
578 }
579 }
580
581 bool_t
582 svc_sendreply_mbuf(struct svc_req *rqstp, struct mbuf *m)
583 {
584 struct rpc_msg rply;
585
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;
593
594 return (svc_sendreply_common(rqstp, &rply, m));
595 }
596
597 /*
598 * No procedure error reply
599 */
600 void
601 svcerr_noproc(struct svc_req *rqstp)
602 {
603 SVCXPRT *xprt = rqstp->rq_xprt;
604 struct rpc_msg rply;
605
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;
611
612 if (xprt->xp_pool->sp_rcache)
613 replay_setreply(xprt->xp_pool->sp_rcache,
614 &rply, svc_getrpccaller(rqstp), NULL);
615
616 svc_sendreply_common(rqstp, &rply, NULL);
617 }
618
619 /*
620 * Can't decode args error reply
621 */
622 void
623 svcerr_decode(struct svc_req *rqstp)
624 {
625 SVCXPRT *xprt = rqstp->rq_xprt;
626 struct rpc_msg rply;
627
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;
633
634 if (xprt->xp_pool->sp_rcache)
635 replay_setreply(xprt->xp_pool->sp_rcache,
636 &rply, (struct sockaddr *) &xprt->xp_rtaddr, NULL);
637
638 svc_sendreply_common(rqstp, &rply, NULL);
639 }
640
641 /*
642 * Some system error
643 */
644 void
645 svcerr_systemerr(struct svc_req *rqstp)
646 {
647 SVCXPRT *xprt = rqstp->rq_xprt;
648 struct rpc_msg rply;
649
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;
655
656 if (xprt->xp_pool->sp_rcache)
657 replay_setreply(xprt->xp_pool->sp_rcache,
658 &rply, svc_getrpccaller(rqstp), NULL);
659
660 svc_sendreply_common(rqstp, &rply, NULL);
661 }
662
663 /*
664 * Authentication error reply
665 */
666 void
667 svcerr_auth(struct svc_req *rqstp, enum auth_stat why)
668 {
669 SVCXPRT *xprt = rqstp->rq_xprt;
670 struct rpc_msg rply;
671
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;
677
678 if (xprt->xp_pool->sp_rcache)
679 replay_setreply(xprt->xp_pool->sp_rcache,
680 &rply, svc_getrpccaller(rqstp), NULL);
681
682 svc_sendreply_common(rqstp, &rply, NULL);
683 }
684
685 /*
686 * Auth too weak error reply
687 */
688 void
689 svcerr_weakauth(struct svc_req *rqstp)
690 {
691
692 svcerr_auth(rqstp, AUTH_TOOWEAK);
693 }
694
695 /*
696 * Program unavailable error reply
697 */
698 void
699 svcerr_noprog(struct svc_req *rqstp)
700 {
701 SVCXPRT *xprt = rqstp->rq_xprt;
702 struct rpc_msg rply;
703
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;
709
710 if (xprt->xp_pool->sp_rcache)
711 replay_setreply(xprt->xp_pool->sp_rcache,
712 &rply, svc_getrpccaller(rqstp), NULL);
713
714 svc_sendreply_common(rqstp, &rply, NULL);
715 }
716
717 /*
718 * Program version mismatch error reply
719 */
720 void
721 svcerr_progvers(struct svc_req *rqstp, rpcvers_t low_vers, rpcvers_t high_vers)
722 {
723 SVCXPRT *xprt = rqstp->rq_xprt;
724 struct rpc_msg rply;
725
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;
733
734 if (xprt->xp_pool->sp_rcache)
735 replay_setreply(xprt->xp_pool->sp_rcache,
736 &rply, svc_getrpccaller(rqstp), NULL);
737
738 svc_sendreply_common(rqstp, &rply, NULL);
739 }
740
741 /*
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
745 * parameters.
746 */
747 SVCXPRT *
748 svc_xprt_alloc()
749 {
750 SVCXPRT *xprt;
751 SVCXPRT_EXT *ext;
752
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));
757 xprt->xp_p3 = ext;
758 refcount_init(&xprt->xp_refs, 1);
759
760 return (xprt);
761 }
762
763 /*
764 * Free a server transport structure.
765 */
766 void
767 svc_xprt_free(xprt)
768 SVCXPRT *xprt;
769 {
770
771 mem_free(xprt->xp_p3, sizeof(SVCXPRT_EXT));
772 mem_free(xprt, sizeof(SVCXPRT));
773 }
774
775 /* ******************* SERVER INPUT STUFF ******************* */
776
777 /*
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.
781 */
782 static enum xprt_stat
783 svc_getreq(SVCXPRT *xprt, struct svc_req **rqstp_ret)
784 {
785 SVCPOOL *pool = xprt->xp_pool;
786 struct svc_req *r;
787 struct rpc_msg msg;
788 struct mbuf *args;
789 enum xprt_stat stat;
790
791 /* now receive msgs from xprtprt (support batch calls) */
792 r = malloc(sizeof(*r), M_RPC, M_WAITOK|M_ZERO);
793
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)) {
798 enum auth_stat why;
799
800 /*
801 * Handle replays and authenticate before queuing the
802 * request to be executed.
803 */
804 SVC_ACQUIRE(xprt);
805 r->rq_xprt = xprt;
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);
812 switch (rs) {
813 case RS_NEW:
814 break;
815 case RS_DONE:
816 SVC_REPLY(xprt, &repmsg, r->rq_addr,
817 repbody);
818 if (r->rq_addr) {
819 free(r->rq_addr, M_SONAME);
820 r->rq_addr = NULL;
821 }
822 m_freem(args);
823 goto call_done;
824
825 default:
826 m_freem(args);
827 goto call_done;
828 }
829 }
830
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);
836 r->rq_args = args;
837 if ((why = _authenticate(r, &msg)) != AUTH_OK) {
838 /*
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
843 * application.
844 */
845 if (why != RPCSEC_GSS_NODISPATCH)
846 svcerr_auth(r, why);
847 goto call_done;
848 }
849
850 if (!SVCAUTH_UNWRAP(&r->rq_auth, &r->rq_args)) {
851 svcerr_decode(r);
852 goto call_done;
853 }
854
855 /*
856 * Everything checks out, return request to caller.
857 */
858 *rqstp_ret = r;
859 r = NULL;
860 }
861 call_done:
862 if (r) {
863 svc_freereq(r);
864 r = NULL;
865 }
866 if ((stat = SVC_STAT(xprt)) == XPRT_DIED) {
867 xprt_unregister(xprt);
868 }
869
870 return (stat);
871 }
872
873 static void
874 svc_executereq(struct svc_req *rqstp)
875 {
876 SVCXPRT *xprt = rqstp->rq_xprt;
877 SVCPOOL *pool = xprt->xp_pool;
878 int prog_found;
879 rpcvers_t low_vers;
880 rpcvers_t high_vers;
881 struct svc_callout *s;
882
883 /* now match message with a registered service*/
884 prog_found = FALSE;
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) {
890 /*
891 * We hand ownership of r to the
892 * dispatch method - they must call
893 * svc_freereq.
894 */
895 (*s->sc_dispatch)(rqstp, xprt);
896 return;
897 } /* found correct version */
898 prog_found = TRUE;
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 */
904 }
905
906 /*
907 * if we got here, the program or version
908 * is not served ...
909 */
910 if (prog_found)
911 svcerr_progvers(rqstp, low_vers, high_vers);
912 else
913 svcerr_noprog(rqstp);
914
915 svc_freereq(rqstp);
916 }
917
918 static void
919 svc_checkidle(SVCPOOL *pool)
920 {
921 SVCXPRT *xprt, *nxprt;
922 time_t timo;
923 struct svcxprt_list cleanup;
924
925 TAILQ_INIT(&cleanup);
926 TAILQ_FOREACH_SAFE(xprt, &pool->sp_xlist, xp_link, nxprt) {
927 /*
928 * Only some transports have idle timers. Don't time
929 * something out which is just waking up.
930 */
931 if (!xprt->xp_idletimeout || xprt->xp_thread)
932 continue;
933
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);
938 }
939 }
940
941 mtx_unlock(&pool->sp_lock);
942 TAILQ_FOREACH_SAFE(xprt, &cleanup, xp_link, nxprt) {
943 SVC_RELEASE(xprt);
944 }
945 mtx_lock(&pool->sp_lock);
946
947 }
948
949 static void
950 svc_assign_waiting_sockets(SVCPOOL *pool)
951 {
952 SVCXPRT *xprt;
953
954 TAILQ_FOREACH(xprt, &pool->sp_active, xp_alink) {
955 if (!xprt->xp_thread) {
956 xprt_assignthread(xprt);
957 }
958 }
959 }
960
961 static bool_t
962 svc_request_space_available(SVCPOOL *pool)
963 {
964
965 mtx_assert(&pool->sp_lock, MA_OWNED);
966
967 if (pool->sp_space_throttled) {
968 /*
969 * Below the low-water yet? If so, assign any waiting sockets.
970 */
971 if (pool->sp_space_used < pool->sp_space_low) {
972 pool->sp_space_throttled = FALSE;
973 svc_assign_waiting_sockets(pool);
974 return TRUE;
975 }
976
977 return FALSE;
978 } else {
979 if (pool->sp_space_used
980 >= pool->sp_space_high) {
981 pool->sp_space_throttled = TRUE;
982 pool->sp_space_throttle_count++;
983 return FALSE;
984 }
985
986 return TRUE;
987 }
988 }
989
990 static void
991 svc_run_internal(SVCPOOL *pool, bool_t ismaster)
992 {
993 SVCTHREAD *st, *stpref;
994 SVCXPRT *xprt;
995 enum xprt_stat stat;
996 struct svc_req *rqstp;
997 int error;
998
999 st = mem_alloc(sizeof(*st));
1000 st->st_xprt = NULL;
1001 STAILQ_INIT(&st->st_reqs);
1002 cv_init(&st->st_cond, "rpcsvc");
1003
1004 mtx_lock(&pool->sp_lock);
1005 LIST_INSERT_HEAD(&pool->sp_threads, st, st_link);
1006
1007 /*
1008 * If we are a new thread which was spawned to cope with
1009 * increased load, set the state back to SVCPOOL_ACTIVE.
1010 */
1011 if (pool->sp_state == SVCPOOL_THREADSTARTING)
1012 pool->sp_state = SVCPOOL_ACTIVE;
1013
1014 while (pool->sp_state != SVCPOOL_CLOSING) {
1015 /*
1016 * Check for idle transports once per second.
1017 */
1018 if (time_uptime > pool->sp_lastidlecheck) {
1019 pool->sp_lastidlecheck = time_uptime;
1020 svc_checkidle(pool);
1021 }
1022
1023 xprt = st->st_xprt;
1024 if (!xprt && STAILQ_EMPTY(&st->st_reqs)) {
1025 /*
1026 * Enforce maxthreads count.
1027 */
1028 if (pool->sp_threadcount > pool->sp_maxthreads)
1029 break;
1030
1031 /*
1032 * Before sleeping, see if we can find an
1033 * active transport which isn't being serviced
1034 * by a thread.
1035 */
1036 if (svc_request_space_available(pool)) {
1037 TAILQ_FOREACH(xprt, &pool->sp_active,
1038 xp_alink) {
1039 if (!xprt->xp_thread) {
1040 SVC_ACQUIRE(xprt);
1041 xprt->xp_thread = st;
1042 st->st_xprt = xprt;
1043 break;
1044 }
1045 }
1046 }
1047 if (st->st_xprt)
1048 continue;
1049
1050 LIST_INSERT_HEAD(&pool->sp_idlethreads, st, st_ilink);
1051 error = cv_timedwait_sig(&st->st_cond, &pool->sp_lock,
1052 5 * hz);
1053 LIST_REMOVE(st, st_ilink);
1054
1055 /*
1056 * Reduce worker thread count when idle.
1057 */
1058 if (error == EWOULDBLOCK) {
1059 if (!ismaster
1060 && (pool->sp_threadcount
1061 > pool->sp_minthreads)
1062 && !st->st_xprt
1063 && STAILQ_EMPTY(&st->st_reqs))
1064 break;
1065 }
1066 if (error == EWOULDBLOCK)
1067 continue;
1068 if (error) {
1069 if (pool->sp_state != SVCPOOL_CLOSING) {
1070 mtx_unlock(&pool->sp_lock);
1071 svc_exit(pool);
1072 mtx_lock(&pool->sp_lock);
1073 }
1074 break;
1075 }
1076
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);
1083 }
1084 continue;
1085 }
1086
1087 if (xprt) {
1088 /*
1089 * Drain the transport socket and queue up any
1090 * RPCs.
1091 */
1092 xprt->xp_lastactive = time_uptime;
1093 stat = XPRT_IDLE;
1094 do {
1095 if (!svc_request_space_available(pool))
1096 break;
1097 rqstp = NULL;
1098 mtx_unlock(&pool->sp_lock);
1099 stat = svc_getreq(xprt, &rqstp);
1100 mtx_lock(&pool->sp_lock);
1101 if (rqstp) {
1102 /*
1103 * See if the application has
1104 * a preference for some other
1105 * thread.
1106 */
1107 stpref = st;
1108 if (pool->sp_assign)
1109 stpref = pool->sp_assign(st,
1110 rqstp);
1111
1112 pool->sp_space_used +=
1113 rqstp->rq_size;
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,
1120 rqstp, rq_link);
1121 stpref->st_reqcount++;
1122
1123 /*
1124 * If we assigned the request
1125 * to another thread, make
1126 * sure its awake and continue
1127 * reading from the
1128 * socket. Otherwise, try to
1129 * find some other thread to
1130 * read from the socket and
1131 * execute the request
1132 * immediately.
1133 */
1134 if (stpref != st) {
1135 cv_signal(&stpref->st_cond);
1136 continue;
1137 } else {
1138 break;
1139 }
1140 }
1141 } while (stat == XPRT_MOREREQS
1142 && pool->sp_state != SVCPOOL_CLOSING);
1143
1144 /*
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
1150 * the active list.
1151 */
1152 xprt->xp_thread = NULL;
1153 st->st_xprt = 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,
1158 xp_alink);
1159 }
1160 mtx_unlock(&pool->sp_lock);
1161 SVC_RELEASE(xprt);
1162 mtx_lock(&pool->sp_lock);
1163 }
1164
1165 /*
1166 * Execute what we have queued.
1167 */
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;
1174 }
1175 }
1176
1177 if (st->st_xprt) {
1178 xprt = st->st_xprt;
1179 st->st_xprt = NULL;
1180 SVC_RELEASE(xprt);
1181 }
1182
1183 KASSERT(STAILQ_EMPTY(&st->st_reqs), ("stray reqs on exit"));
1184 LIST_REMOVE(st, st_link);
1185 pool->sp_threadcount--;
1186
1187 mtx_unlock(&pool->sp_lock);
1188
1189 cv_destroy(&st->st_cond);
1190 mem_free(st, sizeof(*st));
1191
1192 if (!ismaster)
1193 wakeup(pool);
1194 }
1195
1196 static void
1197 svc_thread_start(void *arg)
1198 {
1199
1200 svc_run_internal((SVCPOOL *) arg, FALSE);
1201 kthread_exit();
1202 }
1203
1204 static void
1205 svc_new_thread(SVCPOOL *pool)
1206 {
1207 struct thread *td;
1208
1209 pool->sp_threadcount++;
1210 kthread_add(svc_thread_start, pool,
1211 pool->sp_proc, &td, 0, 0,
1212 "%s: service", pool->sp_name);
1213 }
1214
1215 void
1216 svc_run(SVCPOOL *pool)
1217 {
1218 int i;
1219 struct proc *p;
1220 struct thread *td;
1221
1222 p = curproc;
1223 td = curthread;
1224 snprintf(td->td_name, sizeof(td->td_name),
1225 "%s: master", pool->sp_name);
1226 pool->sp_state = SVCPOOL_ACTIVE;
1227 pool->sp_proc = p;
1228 pool->sp_lastcreatetime = time_uptime;
1229 pool->sp_threadcount = 1;
1230
1231 for (i = 1; i < pool->sp_minthreads; i++) {
1232 svc_new_thread(pool);
1233 }
1234
1235 svc_run_internal(pool, TRUE);
1236
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);
1241 }
1242
1243 void
1244 svc_exit(SVCPOOL *pool)
1245 {
1246 SVCTHREAD *st;
1247
1248 mtx_lock(&pool->sp_lock);
1249
1250 pool->sp_state = SVCPOOL_CLOSING;
1251 LIST_FOREACH(st, &pool->sp_idlethreads, st_ilink)
1252 cv_signal(&st->st_cond);
1253
1254 mtx_unlock(&pool->sp_lock);
1255 }
1256
1257 bool_t
1258 svc_getargs(struct svc_req *rqstp, xdrproc_t xargs, void *args)
1259 {
1260 struct mbuf *m;
1261 XDR xdrs;
1262 bool_t stat;
1263
1264 m = rqstp->rq_args;
1265 rqstp->rq_args = NULL;
1266
1267 xdrmbuf_create(&xdrs, m, XDR_DECODE);
1268 stat = xargs(&xdrs, args);
1269 XDR_DESTROY(&xdrs);
1270
1271 return (stat);
1272 }
1273
1274 bool_t
1275 svc_freeargs(struct svc_req *rqstp, xdrproc_t xargs, void *args)
1276 {
1277 XDR xdrs;
1278
1279 if (rqstp->rq_addr) {
1280 free(rqstp->rq_addr, M_SONAME);
1281 rqstp->rq_addr = NULL;
1282 }
1283
1284 xdrs.x_op = XDR_FREE;
1285 return (xargs(&xdrs, args));
1286 }
1287
1288 void
1289 svc_freereq(struct svc_req *rqstp)
1290 {
1291 SVCTHREAD *st;
1292 SVCXPRT *xprt;
1293 SVCPOOL *pool;
1294
1295 st = rqstp->rq_thread;
1296 xprt = rqstp->rq_xprt;
1297 if (xprt)
1298 pool = xprt->xp_pool;
1299 else
1300 pool = NULL;
1301 if (st) {
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);
1306 st->st_reqcount--;
1307 if (pool->sp_done)
1308 pool->sp_done(st, rqstp);
1309 mtx_unlock(&pool->sp_lock);
1310 }
1311
1312 if (rqstp->rq_auth.svc_ah_ops)
1313 SVCAUTH_RELEASE(&rqstp->rq_auth);
1314
1315 if (rqstp->rq_xprt) {
1316 SVC_RELEASE(rqstp->rq_xprt);
1317 }
1318
1319 if (rqstp->rq_addr)
1320 free(rqstp->rq_addr, M_SONAME);
1321
1322 if (rqstp->rq_args)
1323 m_freem(rqstp->rq_args);
1324
1325 free(rqstp, M_RPC);
1326 }
Cache object: b444d233427822806e7fb610e2227d56
|