d/freebsd-skq
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity
3718535142
freebsd-skq/sys/rpc/svc.c
wollman 299467e040 Fix overflow bugs in and remove obsolete limit from kernel RPC 
implementation.

The kernel RPC code, which is responsible for the low-level scheduling
of incoming NFS requests, contains a throttling mechanism that
prevents too much kernel memory from being tied up by NFS requests
that are being serviced.  When the throttle is engaged, the RPC layer
stops servicing incoming NFS sockets, resulting ultimately in
backpressure on the clients (if they're using TCP).  However, this is
a very heavy-handed mechanism as it prevents all clients from making
any requests, regardless of how heavy or light they are.  (Thus, when
engaged, the throttle often prevents clients from even mounting the
filesystem.)  The throttle mechanism applies specifically to requests
that have been received by the RPC layer (from a TCP or UDP socket)
and are queued waiting to be serviced by one of the nfsd threads; it
does not limit the amount of backlog in the socket buffers.

The original implementation limited the total bytes of queued requests
to the minimum of a quarter of (nmbclusters * MCLBYTES) and 45 MiB.
The former limit seems reasonable, since requests queued in the socket
buffers and replies being constructed to the requests in progress will
all require some amount of network memory, but the 45 MiB limit is
plainly ridiculous for modern memory sizes: when running 256 service
threads on a busy server, 45 MiB would result in just a single
maximum-sized NFS3PROC_WRITE queued per thread before throttling.

Removing this limit exposed integer-overflow bugs in the original
computation, and related bugs in the routines that actually account
for the amount of traffic enqueued for service threads.  The old
implementation also attempted to reduce accounting overhead by
batching updates until each queue is fully drained, but this is prone
to livelock, resulting in repeated accumulate-throttle-drain cycles on
a busy server.  Various data types are changed to long or unsigned
long; explicit 64-bit types are not used due to the unavailability of
64-bit atomics on many 32-bit platforms, but those platforms also
cannot support nmbclusters large enough to cause overflow.

This code (in a 10.1 kernel) is presently running on production NFS
servers at CSAIL.

Summary of this revision:
* Removes 45 MiB limit on requests queued for nfsd service threads
* Fixes integer-overflow and signedness bugs
* Avoids unnecessary throttling by not deferring accounting for
  completed requests

Differential Revision:	https://reviews.freebsd.org/D2165
Reviewed by:	rmacklem, mav
MFC after:	30 days
Relnotes:	yes
Sponsored by:	MIT Computer Science & Artificial Intelligence Laboratory
2015-04-01 00:45:47 +00:00

1436 lines
34 KiB
C
Raw Blame History

/* $NetBSD: svc.c,v 1.21 2000/07/06 03:10:35 christos Exp $ */
/*-
* Copyright (c) 2009, Sun Microsystems, Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
* modification, are permitted provided that the following conditions are met:
* - Redistributions of source code must retain the above copyright notice,
* this list of conditions and the following disclaimer.
* - Redistributions in binary form must reproduce the above copyright notice,
* this list of conditions and the following disclaimer in the documentation
* and/or other materials provided with the distribution.
* - Neither the name of Sun Microsystems, Inc. nor the names of its
* contributors may be used to endorse or promote products derived
* from this software without specific prior written permission.
*
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
* AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
* IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
* ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
* LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
* CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
* SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
* INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
* CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
* ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
* POSSIBILITY OF SUCH DAMAGE.
*/
#if defined(LIBC_SCCS) && !defined(lint)
static char *sccsid2 = "@(#)svc.c 1.44 88/02/08 Copyr 1984 Sun Micro";
static char *sccsid = "@(#)svc.c 2.4 88/08/11 4.0 RPCSRC";
#endif
#include <sys/cdefs.h>
__FBSDID("$FreeBSD$");
/*
* svc.c, Server-side remote procedure call interface.
*
* There are two sets of procedures here. The xprt routines are
* for handling transport handles. The svc routines handle the
* list of service routines.
*
* Copyright (C) 1984, Sun Microsystems, Inc.
*/
#include <sys/param.h>
#include <sys/lock.h>
#include <sys/kernel.h>
#include <sys/kthread.h>
#include <sys/malloc.h>
#include <sys/mbuf.h>
#include <sys/mutex.h>
#include <sys/proc.h>
#include <sys/queue.h>
#include <sys/socketvar.h>
#include <sys/systm.h>
#include <sys/smp.h>
#include <sys/sx.h>
#include <sys/ucred.h>
#include <rpc/rpc.h>
#include <rpc/rpcb_clnt.h>
#include <rpc/replay.h>
#include <rpc/rpc_com.h>
#define SVC_VERSQUIET 0x0001 /* keep quiet about vers mismatch */
#define version_keepquiet(xp) (SVC_EXT(xp)->xp_flags & SVC_VERSQUIET)
static struct svc_callout *svc_find(SVCPOOL *pool, rpcprog_t, rpcvers_t,
char *);
static void svc_new_thread(SVCGROUP *grp);
static void xprt_unregister_locked(SVCXPRT *xprt);
static void svc_change_space_used(SVCPOOL *pool, long delta);
static bool_t svc_request_space_available(SVCPOOL *pool);
/* *************** SVCXPRT related stuff **************** */
static int svcpool_minthread_sysctl(SYSCTL_HANDLER_ARGS);
static int svcpool_maxthread_sysctl(SYSCTL_HANDLER_ARGS);
static int svcpool_threads_sysctl(SYSCTL_HANDLER_ARGS);
SVCPOOL*
svcpool_create(const char *name, struct sysctl_oid_list *sysctl_base)
{
SVCPOOL *pool;
SVCGROUP *grp;
int g;
pool = malloc(sizeof(SVCPOOL), M_RPC, M_WAITOK|M_ZERO);
mtx_init(&pool->sp_lock, "sp_lock", NULL, MTX_DEF);
pool->sp_name = name;
pool->sp_state = SVCPOOL_INIT;
pool->sp_proc = NULL;
TAILQ_INIT(&pool->sp_callouts);
TAILQ_INIT(&pool->sp_lcallouts);
pool->sp_minthreads = 1;
pool->sp_maxthreads = 1;
pool->sp_groupcount = 1;
for (g = 0; g < SVC_MAXGROUPS; g++) {
grp = &pool->sp_groups[g];
mtx_init(&grp->sg_lock, "sg_lock", NULL, MTX_DEF);
grp->sg_pool = pool;
grp->sg_state = SVCPOOL_ACTIVE;
TAILQ_INIT(&grp->sg_xlist);
TAILQ_INIT(&grp->sg_active);
LIST_INIT(&grp->sg_idlethreads);
grp->sg_minthreads = 1;
grp->sg_maxthreads = 1;
}
/*
* Don't use more than a quarter of mbuf clusters. Nota bene:
* nmbclusters is an int, but nmbclusters*MCLBYTES may overflow
* on LP64 architectures, so cast to u_long to avoid undefined
* behavior. (ILP32 architectures cannot have nmbclusters
* large enough to overflow for other reasons.)
*/
pool->sp_space_high = (u_long)nmbclusters * MCLBYTES / 4;
pool->sp_space_low = (pool->sp_space_high / 3) * 2;
sysctl_ctx_init(&pool->sp_sysctl);
if (sysctl_base) {
SYSCTL_ADD_PROC(&pool->sp_sysctl, sysctl_base, OID_AUTO,
"minthreads", CTLTYPE_INT | CTLFLAG_RW,
pool, 0, svcpool_minthread_sysctl, "I",
"Minimal number of threads");
SYSCTL_ADD_PROC(&pool->sp_sysctl, sysctl_base, OID_AUTO,
"maxthreads", CTLTYPE_INT | CTLFLAG_RW,
pool, 0, svcpool_maxthread_sysctl, "I",
"Maximal number of threads");
SYSCTL_ADD_PROC(&pool->sp_sysctl, sysctl_base, OID_AUTO,
"threads", CTLTYPE_INT | CTLFLAG_RD,
pool, 0, svcpool_threads_sysctl, "I",
"Current number of threads");
SYSCTL_ADD_INT(&pool->sp_sysctl, sysctl_base, OID_AUTO,
"groups", CTLFLAG_RD, &pool->sp_groupcount, 0,
"Number of thread groups");
SYSCTL_ADD_ULONG(&pool->sp_sysctl, sysctl_base, OID_AUTO,
"request_space_used", CTLFLAG_RD,
&pool->sp_space_used,
"Space in parsed but not handled requests.");
SYSCTL_ADD_ULONG(&pool->sp_sysctl, sysctl_base, OID_AUTO,
"request_space_used_highest", CTLFLAG_RD,
&pool->sp_space_used_highest,
"Highest space used since reboot.");
SYSCTL_ADD_ULONG(&pool->sp_sysctl, sysctl_base, OID_AUTO,
"request_space_high", CTLFLAG_RW,
&pool->sp_space_high,
"Maximum space in parsed but not handled requests.");
SYSCTL_ADD_ULONG(&pool->sp_sysctl, sysctl_base, OID_AUTO,
"request_space_low", CTLFLAG_RW,
&pool->sp_space_low,
"Low water mark for request space.");
SYSCTL_ADD_INT(&pool->sp_sysctl, sysctl_base, OID_AUTO,
"request_space_throttled", CTLFLAG_RD,
&pool->sp_space_throttled, 0,
"Whether nfs requests are currently throttled");
SYSCTL_ADD_INT(&pool->sp_sysctl, sysctl_base, OID_AUTO,
"request_space_throttle_count", CTLFLAG_RD,
&pool->sp_space_throttle_count, 0,
"Count of times throttling based on request space has occurred");
}
return pool;
}
void
svcpool_destroy(SVCPOOL *pool)
{
SVCGROUP *grp;
SVCXPRT *xprt, *nxprt;
struct svc_callout *s;
struct svc_loss_callout *sl;
struct svcxprt_list cleanup;
int g;
TAILQ_INIT(&cleanup);
for (g = 0; g < SVC_MAXGROUPS; g++) {
grp = &pool->sp_groups[g];
mtx_lock(&grp->sg_lock);
while ((xprt = TAILQ_FIRST(&grp->sg_xlist)) != NULL) {
xprt_unregister_locked(xprt);
TAILQ_INSERT_TAIL(&cleanup, xprt, xp_link);
}
mtx_unlock(&grp->sg_lock);
}
TAILQ_FOREACH_SAFE(xprt, &cleanup, xp_link, nxprt) {
SVC_RELEASE(xprt);
}
mtx_lock(&pool->sp_lock);
while ((s = TAILQ_FIRST(&pool->sp_callouts)) != NULL) {
mtx_unlock(&pool->sp_lock);
svc_unreg(pool, s->sc_prog, s->sc_vers);
mtx_lock(&pool->sp_lock);
}
while ((sl = TAILQ_FIRST(&pool->sp_lcallouts)) != NULL) {
mtx_unlock(&pool->sp_lock);
svc_loss_unreg(pool, sl->slc_dispatch);
mtx_lock(&pool->sp_lock);
}
mtx_unlock(&pool->sp_lock);
for (g = 0; g < SVC_MAXGROUPS; g++) {
grp = &pool->sp_groups[g];
mtx_destroy(&grp->sg_lock);
}
mtx_destroy(&pool->sp_lock);
if (pool->sp_rcache)
replay_freecache(pool->sp_rcache);
sysctl_ctx_free(&pool->sp_sysctl);
free(pool, M_RPC);
}
/*
* Sysctl handler to get the present thread count on a pool
*/
static int
svcpool_threads_sysctl(SYSCTL_HANDLER_ARGS)
{
SVCPOOL *pool;
int threads, error, g;
pool = oidp->oid_arg1;
threads = 0;
mtx_lock(&pool->sp_lock);
for (g = 0; g < pool->sp_groupcount; g++)
threads += pool->sp_groups[g].sg_threadcount;
mtx_unlock(&pool->sp_lock);
error = sysctl_handle_int(oidp, &threads, 0, req);
return (error);
}
/*
* Sysctl handler to set the minimum thread count on a pool
*/
static int
svcpool_minthread_sysctl(SYSCTL_HANDLER_ARGS)
{
SVCPOOL *pool;
int newminthreads, error, g;
pool = oidp->oid_arg1;
newminthreads = pool->sp_minthreads;
error = sysctl_handle_int(oidp, &newminthreads, 0, req);
if (error == 0 && newminthreads != pool->sp_minthreads) {
if (newminthreads > pool->sp_maxthreads)
return (EINVAL);
mtx_lock(&pool->sp_lock);
pool->sp_minthreads = newminthreads;
for (g = 0; g < pool->sp_groupcount; g++) {
pool->sp_groups[g].sg_minthreads = max(1,
pool->sp_minthreads / pool->sp_groupcount);
}
mtx_unlock(&pool->sp_lock);
}
return (error);
}
/*
* Sysctl handler to set the maximum thread count on a pool
*/
static int
svcpool_maxthread_sysctl(SYSCTL_HANDLER_ARGS)
{
SVCPOOL *pool;
int newmaxthreads, error, g;
pool = oidp->oid_arg1;
newmaxthreads = pool->sp_maxthreads;
error = sysctl_handle_int(oidp, &newmaxthreads, 0, req);
if (error == 0 && newmaxthreads != pool->sp_maxthreads) {
if (newmaxthreads < pool->sp_minthreads)
return (EINVAL);
mtx_lock(&pool->sp_lock);
pool->sp_maxthreads = newmaxthreads;
for (g = 0; g < pool->sp_groupcount; g++) {
pool->sp_groups[g].sg_maxthreads = max(1,
pool->sp_maxthreads / pool->sp_groupcount);
}
mtx_unlock(&pool->sp_lock);
}
return (error);
}
/*
* Activate a transport handle.
*/
void
xprt_register(SVCXPRT *xprt)
{
SVCPOOL *pool = xprt->xp_pool;
SVCGROUP *grp;
int g;
SVC_ACQUIRE(xprt);
g = atomic_fetchadd_int(&pool->sp_nextgroup, 1) % pool->sp_groupcount;
xprt->xp_group = grp = &pool->sp_groups[g];
mtx_lock(&grp->sg_lock);
xprt->xp_registered = TRUE;
xprt->xp_active = FALSE;
TAILQ_INSERT_TAIL(&grp->sg_xlist, xprt, xp_link);
mtx_unlock(&grp->sg_lock);
}
/*
* De-activate a transport handle. Note: the locked version doesn't
* release the transport - caller must do that after dropping the pool
* lock.
*/
static void
xprt_unregister_locked(SVCXPRT *xprt)
{
SVCGROUP *grp = xprt->xp_group;
mtx_assert(&grp->sg_lock, MA_OWNED);
KASSERT(xprt->xp_registered == TRUE,
("xprt_unregister_locked: not registered"));
xprt_inactive_locked(xprt);
TAILQ_REMOVE(&grp->sg_xlist, xprt, xp_link);
xprt->xp_registered = FALSE;
}
void
xprt_unregister(SVCXPRT *xprt)
{
SVCGROUP *grp = xprt->xp_group;
mtx_lock(&grp->sg_lock);
if (xprt->xp_registered == FALSE) {
/* Already unregistered by another thread */
mtx_unlock(&grp->sg_lock);
return;
}
xprt_unregister_locked(xprt);
mtx_unlock(&grp->sg_lock);
SVC_RELEASE(xprt);
}
/*
* Attempt to assign a service thread to this transport.
*/
static int
xprt_assignthread(SVCXPRT *xprt)
{
SVCGROUP *grp = xprt->xp_group;
SVCTHREAD *st;
mtx_assert(&grp->sg_lock, MA_OWNED);
st = LIST_FIRST(&grp->sg_idlethreads);
if (st) {
LIST_REMOVE(st, st_ilink);
SVC_ACQUIRE(xprt);
xprt->xp_thread = st;
st->st_xprt = xprt;
cv_signal(&st->st_cond);
return (TRUE);
} else {
/*
* See if we can create a new thread. The
* actual thread creation happens in
* svc_run_internal because our locking state
* is poorly defined (we are typically called
* from a socket upcall). Don't create more
* than one thread per second.
*/
if (grp->sg_state == SVCPOOL_ACTIVE
&& grp->sg_lastcreatetime < time_uptime
&& grp->sg_threadcount < grp->sg_maxthreads) {
grp->sg_state = SVCPOOL_THREADWANTED;
}
}
return (FALSE);
}
void
xprt_active(SVCXPRT *xprt)
{
SVCGROUP *grp = xprt->xp_group;
mtx_lock(&grp->sg_lock);
if (!xprt->xp_registered) {
/*
* Race with xprt_unregister - we lose.
*/
mtx_unlock(&grp->sg_lock);
return;
}
if (!xprt->xp_active) {
xprt->xp_active = TRUE;
if (xprt->xp_thread == NULL) {
if (!svc_request_space_available(xprt->xp_pool) ||
!xprt_assignthread(xprt))
TAILQ_INSERT_TAIL(&grp->sg_active, xprt,
xp_alink);
}
}
mtx_unlock(&grp->sg_lock);
}
void
xprt_inactive_locked(SVCXPRT *xprt)
{
SVCGROUP *grp = xprt->xp_group;
mtx_assert(&grp->sg_lock, MA_OWNED);
if (xprt->xp_active) {
if (xprt->xp_thread == NULL)
TAILQ_REMOVE(&grp->sg_active, xprt, xp_alink);
xprt->xp_active = FALSE;
}
}
void
xprt_inactive(SVCXPRT *xprt)
{
SVCGROUP *grp = xprt->xp_group;
mtx_lock(&grp->sg_lock);
xprt_inactive_locked(xprt);
mtx_unlock(&grp->sg_lock);
}
/*
* Variant of xprt_inactive() for use only when sure that port is
* assigned to thread. For example, withing receive handlers.
*/
void
xprt_inactive_self(SVCXPRT *xprt)
{
KASSERT(xprt->xp_thread != NULL,
("xprt_inactive_self(%p) with NULL xp_thread", xprt));
xprt->xp_active = FALSE;
}
/*
* Add a service program to the callout list.
* The dispatch routine will be called when a rpc request for this
* program number comes in.
*/
bool_t
svc_reg(SVCXPRT *xprt, const rpcprog_t prog, const rpcvers_t vers,
void (*dispatch)(struct svc_req *, SVCXPRT *),
const struct netconfig *nconf)
{
SVCPOOL *pool = xprt->xp_pool;
struct svc_callout *s;
char *netid = NULL;
int flag = 0;
/* VARIABLES PROTECTED BY svc_lock: s, svc_head */
if (xprt->xp_netid) {
netid = strdup(xprt->xp_netid, M_RPC);
flag = 1;
} else if (nconf && nconf->nc_netid) {
netid = strdup(nconf->nc_netid, M_RPC);
flag = 1;
} /* must have been created with svc_raw_create */
if ((netid == NULL) && (flag == 1)) {
return (FALSE);
}
mtx_lock(&pool->sp_lock);
if ((s = svc_find(pool, prog, vers, netid)) != NULL) {
if (netid)
free(netid, M_RPC);
if (s->sc_dispatch == dispatch)
goto rpcb_it; /* he is registering another xptr */
mtx_unlock(&pool->sp_lock);
return (FALSE);
}
s = malloc(sizeof (struct svc_callout), M_RPC, M_NOWAIT);
if (s == NULL) {
if (netid)
free(netid, M_RPC);
mtx_unlock(&pool->sp_lock);
return (FALSE);
}
s->sc_prog = prog;
s->sc_vers = vers;
s->sc_dispatch = dispatch;
s->sc_netid = netid;
TAILQ_INSERT_TAIL(&pool->sp_callouts, s, sc_link);
if ((xprt->xp_netid == NULL) && (flag == 1) && netid)
((SVCXPRT *) xprt)->xp_netid = strdup(netid, M_RPC);
rpcb_it:
mtx_unlock(&pool->sp_lock);
/* now register the information with the local binder service */
if (nconf) {
bool_t dummy;
struct netconfig tnc;
struct netbuf nb;
tnc = *nconf;
nb.buf = &xprt->xp_ltaddr;
nb.len = xprt->xp_ltaddr.ss_len;
dummy = rpcb_set(prog, vers, &tnc, &nb);
return (dummy);
}
return (TRUE);
}
/*
* Remove a service program from the callout list.
*/
void
svc_unreg(SVCPOOL *pool, const rpcprog_t prog, const rpcvers_t vers)
{
struct svc_callout *s;
/* unregister the information anyway */
(void) rpcb_unset(prog, vers, NULL);
mtx_lock(&pool->sp_lock);
while ((s = svc_find(pool, prog, vers, NULL)) != NULL) {
TAILQ_REMOVE(&pool->sp_callouts, s, sc_link);
if (s->sc_netid)
mem_free(s->sc_netid, sizeof (s->sc_netid) + 1);
mem_free(s, sizeof (struct svc_callout));
}
mtx_unlock(&pool->sp_lock);
}
/*
* Add a service connection loss program to the callout list.
* The dispatch routine will be called when some port in ths pool die.
*/
bool_t
svc_loss_reg(SVCXPRT *xprt, void (*dispatch)(SVCXPRT *))
{
SVCPOOL *pool = xprt->xp_pool;
struct svc_loss_callout *s;
mtx_lock(&pool->sp_lock);
TAILQ_FOREACH(s, &pool->sp_lcallouts, slc_link) {
if (s->slc_dispatch == dispatch)
break;
}
if (s != NULL) {
mtx_unlock(&pool->sp_lock);
return (TRUE);
}
s = malloc(sizeof (struct svc_callout), M_RPC, M_NOWAIT);
if (s == NULL) {
mtx_unlock(&pool->sp_lock);
return (FALSE);
}
s->slc_dispatch = dispatch;
TAILQ_INSERT_TAIL(&pool->sp_lcallouts, s, slc_link);
mtx_unlock(&pool->sp_lock);
return (TRUE);
}
/*
* Remove a service connection loss program from the callout list.
*/
void
svc_loss_unreg(SVCPOOL *pool, void (*dispatch)(SVCXPRT *))
{
struct svc_loss_callout *s;
mtx_lock(&pool->sp_lock);
TAILQ_FOREACH(s, &pool->sp_lcallouts, slc_link) {
if (s->slc_dispatch == dispatch) {
TAILQ_REMOVE(&pool->sp_lcallouts, s, slc_link);
free(s, M_RPC);
break;
}
}
mtx_unlock(&pool->sp_lock);
}
/* ********************** CALLOUT list related stuff ************* */
/*
* Search the callout list for a program number, return the callout
* struct.
*/
static struct svc_callout *
svc_find(SVCPOOL *pool, rpcprog_t prog, rpcvers_t vers, char *netid)
{
struct svc_callout *s;
mtx_assert(&pool->sp_lock, MA_OWNED);
TAILQ_FOREACH(s, &pool->sp_callouts, sc_link) {
if (s->sc_prog == prog && s->sc_vers == vers
&& (netid == NULL || s->sc_netid == NULL ||
strcmp(netid, s->sc_netid) == 0))
break;
}
return (s);
}
/* ******************* REPLY GENERATION ROUTINES ************ */
static bool_t
svc_sendreply_common(struct svc_req *rqstp, struct rpc_msg *rply,
struct mbuf *body)
{
SVCXPRT *xprt = rqstp->rq_xprt;
bool_t ok;
if (rqstp->rq_args) {
m_freem(rqstp->rq_args);
rqstp->rq_args = NULL;
}
if (xprt->xp_pool->sp_rcache)
replay_setreply(xprt->xp_pool->sp_rcache,
rply, svc_getrpccaller(rqstp), body);
if (!SVCAUTH_WRAP(&rqstp->rq_auth, &body))
return (FALSE);
ok = SVC_REPLY(xprt, rply, rqstp->rq_addr, body, &rqstp->rq_reply_seq);
if (rqstp->rq_addr) {
free(rqstp->rq_addr, M_SONAME);
rqstp->rq_addr = NULL;
}
return (ok);
}
/*
* Send a reply to an rpc request
*/
bool_t
svc_sendreply(struct svc_req *rqstp, xdrproc_t xdr_results, void * xdr_location)
{
struct rpc_msg rply;
struct mbuf *m;
XDR xdrs;
bool_t ok;
rply.rm_xid = rqstp->rq_xid;
rply.rm_direction = REPLY;
rply.rm_reply.rp_stat = MSG_ACCEPTED;
rply.acpted_rply.ar_verf = rqstp->rq_verf;
rply.acpted_rply.ar_stat = SUCCESS;
rply.acpted_rply.ar_results.where = NULL;
rply.acpted_rply.ar_results.proc = (xdrproc_t) xdr_void;
m = m_getcl(M_WAITOK, MT_DATA, 0);
xdrmbuf_create(&xdrs, m, XDR_ENCODE);
ok = xdr_results(&xdrs, xdr_location);
XDR_DESTROY(&xdrs);
if (ok) {
return (svc_sendreply_common(rqstp, &rply, m));
} else {
m_freem(m);
return (FALSE);
}
}
bool_t
svc_sendreply_mbuf(struct svc_req *rqstp, struct mbuf *m)
{
struct rpc_msg rply;
rply.rm_xid = rqstp->rq_xid;
rply.rm_direction = REPLY;
rply.rm_reply.rp_stat = MSG_ACCEPTED;
rply.acpted_rply.ar_verf = rqstp->rq_verf;
rply.acpted_rply.ar_stat = SUCCESS;
rply.acpted_rply.ar_results.where = NULL;
rply.acpted_rply.ar_results.proc = (xdrproc_t) xdr_void;
return (svc_sendreply_common(rqstp, &rply, m));
}
/*
* No procedure error reply
*/
void
svcerr_noproc(struct svc_req *rqstp)
{
SVCXPRT *xprt = rqstp->rq_xprt;
struct rpc_msg rply;
rply.rm_xid = rqstp->rq_xid;
rply.rm_direction = REPLY;
rply.rm_reply.rp_stat = MSG_ACCEPTED;
rply.acpted_rply.ar_verf = rqstp->rq_verf;
rply.acpted_rply.ar_stat = PROC_UNAVAIL;
if (xprt->xp_pool->sp_rcache)
replay_setreply(xprt->xp_pool->sp_rcache,
&rply, svc_getrpccaller(rqstp), NULL);
svc_sendreply_common(rqstp, &rply, NULL);
}
/*
* Can't decode args error reply
*/
void
svcerr_decode(struct svc_req *rqstp)
{
SVCXPRT *xprt = rqstp->rq_xprt;
struct rpc_msg rply;
rply.rm_xid = rqstp->rq_xid;
rply.rm_direction = REPLY;
rply.rm_reply.rp_stat = MSG_ACCEPTED;
rply.acpted_rply.ar_verf = rqstp->rq_verf;
rply.acpted_rply.ar_stat = GARBAGE_ARGS;
if (xprt->xp_pool->sp_rcache)
replay_setreply(xprt->xp_pool->sp_rcache,
&rply, (struct sockaddr *) &xprt->xp_rtaddr, NULL);
svc_sendreply_common(rqstp, &rply, NULL);
}
/*
* Some system error
*/
void
svcerr_systemerr(struct svc_req *rqstp)
{
SVCXPRT *xprt = rqstp->rq_xprt;
struct rpc_msg rply;
rply.rm_xid = rqstp->rq_xid;
rply.rm_direction = REPLY;
rply.rm_reply.rp_stat = MSG_ACCEPTED;
rply.acpted_rply.ar_verf = rqstp->rq_verf;
rply.acpted_rply.ar_stat = SYSTEM_ERR;
if (xprt->xp_pool->sp_rcache)
replay_setreply(xprt->xp_pool->sp_rcache,
&rply, svc_getrpccaller(rqstp), NULL);
svc_sendreply_common(rqstp, &rply, NULL);
}
/*
* Authentication error reply
*/
void
svcerr_auth(struct svc_req *rqstp, enum auth_stat why)
{
SVCXPRT *xprt = rqstp->rq_xprt;
struct rpc_msg rply;
rply.rm_xid = rqstp->rq_xid;
rply.rm_direction = REPLY;
rply.rm_reply.rp_stat = MSG_DENIED;
rply.rjcted_rply.rj_stat = AUTH_ERROR;
rply.rjcted_rply.rj_why = why;
if (xprt->xp_pool->sp_rcache)
replay_setreply(xprt->xp_pool->sp_rcache,
&rply, svc_getrpccaller(rqstp), NULL);
svc_sendreply_common(rqstp, &rply, NULL);
}
/*
* Auth too weak error reply
*/
void
svcerr_weakauth(struct svc_req *rqstp)
{
svcerr_auth(rqstp, AUTH_TOOWEAK);
}
/*
* Program unavailable error reply
*/
void
svcerr_noprog(struct svc_req *rqstp)
{
SVCXPRT *xprt = rqstp->rq_xprt;
struct rpc_msg rply;
rply.rm_xid = rqstp->rq_xid;
rply.rm_direction = REPLY;
rply.rm_reply.rp_stat = MSG_ACCEPTED;
rply.acpted_rply.ar_verf = rqstp->rq_verf;
rply.acpted_rply.ar_stat = PROG_UNAVAIL;
if (xprt->xp_pool->sp_rcache)
replay_setreply(xprt->xp_pool->sp_rcache,
&rply, svc_getrpccaller(rqstp), NULL);
svc_sendreply_common(rqstp, &rply, NULL);
}
/*
* Program version mismatch error reply
*/
void
svcerr_progvers(struct svc_req *rqstp, rpcvers_t low_vers, rpcvers_t high_vers)
{
SVCXPRT *xprt = rqstp->rq_xprt;
struct rpc_msg rply;
rply.rm_xid = rqstp->rq_xid;
rply.rm_direction = REPLY;
rply.rm_reply.rp_stat = MSG_ACCEPTED;
rply.acpted_rply.ar_verf = rqstp->rq_verf;
rply.acpted_rply.ar_stat = PROG_MISMATCH;
rply.acpted_rply.ar_vers.low = (uint32_t)low_vers;
rply.acpted_rply.ar_vers.high = (uint32_t)high_vers;
if (xprt->xp_pool->sp_rcache)
replay_setreply(xprt->xp_pool->sp_rcache,
&rply, svc_getrpccaller(rqstp), NULL);
svc_sendreply_common(rqstp, &rply, NULL);
}
/*
* Allocate a new server transport structure. All fields are
* initialized to zero and xp_p3 is initialized to point at an
* extension structure to hold various flags and authentication
* parameters.
*/
SVCXPRT *
svc_xprt_alloc()
{
SVCXPRT *xprt;
SVCXPRT_EXT *ext;
xprt = mem_alloc(sizeof(SVCXPRT));
memset(xprt, 0, sizeof(SVCXPRT));
ext = mem_alloc(sizeof(SVCXPRT_EXT));
memset(ext, 0, sizeof(SVCXPRT_EXT));
xprt->xp_p3 = ext;
refcount_init(&xprt->xp_refs, 1);
return (xprt);
}
/*
* Free a server transport structure.
*/
void
svc_xprt_free(xprt)
SVCXPRT *xprt;
{
mem_free(xprt->xp_p3, sizeof(SVCXPRT_EXT));
mem_free(xprt, sizeof(SVCXPRT));
}
/* ******************* SERVER INPUT STUFF ******************* */
/*
* Read RPC requests from a transport and queue them to be
* executed. We handle authentication and replay cache replies here.
* Actually dispatching the RPC is deferred till svc_executereq.
*/
static enum xprt_stat
svc_getreq(SVCXPRT *xprt, struct svc_req **rqstp_ret)
{
SVCPOOL *pool = xprt->xp_pool;
struct svc_req *r;
struct rpc_msg msg;
struct mbuf *args;
struct svc_loss_callout *s;
enum xprt_stat stat;
/* now receive msgs from xprtprt (support batch calls) */
r = malloc(sizeof(*r), M_RPC, M_WAITOK|M_ZERO);
msg.rm_call.cb_cred.oa_base = r->rq_credarea;
msg.rm_call.cb_verf.oa_base = &r->rq_credarea[MAX_AUTH_BYTES];
r->rq_clntcred = &r->rq_credarea[2*MAX_AUTH_BYTES];
if (SVC_RECV(xprt, &msg, &r->rq_addr, &args)) {
enum auth_stat why;
/*
* Handle replays and authenticate before queuing the
* request to be executed.
*/
SVC_ACQUIRE(xprt);
r->rq_xprt = xprt;
if (pool->sp_rcache) {
struct rpc_msg repmsg;
struct mbuf *repbody;
enum replay_state rs;
rs = replay_find(pool->sp_rcache, &msg,
svc_getrpccaller(r), &repmsg, &repbody);
switch (rs) {
case RS_NEW:
break;
case RS_DONE:
SVC_REPLY(xprt, &repmsg, r->rq_addr,
repbody, &r->rq_reply_seq);
if (r->rq_addr) {
free(r->rq_addr, M_SONAME);
r->rq_addr = NULL;
}
m_freem(args);
goto call_done;
default:
m_freem(args);
goto call_done;
}
}
r->rq_xid = msg.rm_xid;
r->rq_prog = msg.rm_call.cb_prog;
r->rq_vers = msg.rm_call.cb_vers;
r->rq_proc = msg.rm_call.cb_proc;
r->rq_size = sizeof(*r) + m_length(args, NULL);
r->rq_args = args;
if ((why = _authenticate(r, &msg)) != AUTH_OK) {
/*
* RPCSEC_GSS uses this return code
* for requests that form part of its
* context establishment protocol and
* should not be dispatched to the
* application.
*/
if (why != RPCSEC_GSS_NODISPATCH)
svcerr_auth(r, why);
goto call_done;
}
if (!SVCAUTH_UNWRAP(&r->rq_auth, &r->rq_args)) {
svcerr_decode(r);
goto call_done;
}
/*
* Everything checks out, return request to caller.
*/
*rqstp_ret = r;
r = NULL;
}
call_done:
if (r) {
svc_freereq(r);
r = NULL;
}
if ((stat = SVC_STAT(xprt)) == XPRT_DIED) {
TAILQ_FOREACH(s, &pool->sp_lcallouts, slc_link)
(*s->slc_dispatch)(xprt);
xprt_unregister(xprt);
}
return (stat);
}
static void
svc_executereq(struct svc_req *rqstp)
{
SVCXPRT *xprt = rqstp->rq_xprt;
SVCPOOL *pool = xprt->xp_pool;
int prog_found;
rpcvers_t low_vers;
rpcvers_t high_vers;
struct svc_callout *s;
/* now match message with a registered service*/
prog_found = FALSE;
low_vers = (rpcvers_t) -1L;
high_vers = (rpcvers_t) 0L;
TAILQ_FOREACH(s, &pool->sp_callouts, sc_link) {
if (s->sc_prog == rqstp->rq_prog) {
if (s->sc_vers == rqstp->rq_vers) {
/*
* We hand ownership of r to the
* dispatch method - they must call
* svc_freereq.
*/
(*s->sc_dispatch)(rqstp, xprt);
return;
} /* found correct version */
prog_found = TRUE;
if (s->sc_vers < low_vers)
low_vers = s->sc_vers;
if (s->sc_vers > high_vers)
high_vers = s->sc_vers;
} /* found correct program */
}
/*
* if we got here, the program or version
* is not served ...
*/
if (prog_found)
svcerr_progvers(rqstp, low_vers, high_vers);
else
svcerr_noprog(rqstp);
svc_freereq(rqstp);
}
static void
svc_checkidle(SVCGROUP *grp)
{
SVCXPRT *xprt, *nxprt;
time_t timo;
struct svcxprt_list cleanup;
TAILQ_INIT(&cleanup);
TAILQ_FOREACH_SAFE(xprt, &grp->sg_xlist, xp_link, nxprt) {
/*
* Only some transports have idle timers. Don't time
* something out which is just waking up.
*/
if (!xprt->xp_idletimeout || xprt->xp_thread)
continue;
timo = xprt->xp_lastactive + xprt->xp_idletimeout;
if (time_uptime > timo) {
xprt_unregister_locked(xprt);
TAILQ_INSERT_TAIL(&cleanup, xprt, xp_link);
}
}
mtx_unlock(&grp->sg_lock);
TAILQ_FOREACH_SAFE(xprt, &cleanup, xp_link, nxprt) {
SVC_RELEASE(xprt);
}
mtx_lock(&grp->sg_lock);
}
static void
svc_assign_waiting_sockets(SVCPOOL *pool)
{
SVCGROUP *grp;
SVCXPRT *xprt;
int g;
for (g = 0; g < pool->sp_groupcount; g++) {
grp = &pool->sp_groups[g];
mtx_lock(&grp->sg_lock);
while ((xprt = TAILQ_FIRST(&grp->sg_active)) != NULL) {
if (xprt_assignthread(xprt))
TAILQ_REMOVE(&grp->sg_active, xprt, xp_alink);
else
break;
}
mtx_unlock(&grp->sg_lock);
}
}
static void
svc_change_space_used(SVCPOOL *pool, long delta)
{
unsigned long value;
value = atomic_fetchadd_long(&pool->sp_space_used, delta) + delta;
if (delta > 0) {
if (value >= pool->sp_space_high && !pool->sp_space_throttled) {
pool->sp_space_throttled = TRUE;
pool->sp_space_throttle_count++;
}
if (value > pool->sp_space_used_highest)
pool->sp_space_used_highest = value;
} else {
if (value < pool->sp_space_low && pool->sp_space_throttled) {
pool->sp_space_throttled = FALSE;
svc_assign_waiting_sockets(pool);
}
}
}
static bool_t
svc_request_space_available(SVCPOOL *pool)
{
if (pool->sp_space_throttled)
return (FALSE);
return (TRUE);
}
static void
svc_run_internal(SVCGROUP *grp, bool_t ismaster)
{
SVCPOOL *pool = grp->sg_pool;
SVCTHREAD *st, *stpref;
SVCXPRT *xprt;
enum xprt_stat stat;
struct svc_req *rqstp;
struct proc *p;
long sz;
int error;
st = mem_alloc(sizeof(*st));
mtx_init(&st->st_lock, "st_lock", NULL, MTX_DEF);
st->st_pool = pool;
st->st_xprt = NULL;
STAILQ_INIT(&st->st_reqs);
cv_init(&st->st_cond, "rpcsvc");
mtx_lock(&grp->sg_lock);
/*
* If we are a new thread which was spawned to cope with
* increased load, set the state back to SVCPOOL_ACTIVE.
*/
if (grp->sg_state == SVCPOOL_THREADSTARTING)
grp->sg_state = SVCPOOL_ACTIVE;
while (grp->sg_state != SVCPOOL_CLOSING) {
/*
* Create new thread if requested.
*/
if (grp->sg_state == SVCPOOL_THREADWANTED) {
grp->sg_state = SVCPOOL_THREADSTARTING;
grp->sg_lastcreatetime = time_uptime;
mtx_unlock(&grp->sg_lock);
svc_new_thread(grp);
mtx_lock(&grp->sg_lock);
continue;
}
/*
* Check for idle transports once per second.
*/
if (time_uptime > grp->sg_lastidlecheck) {
grp->sg_lastidlecheck = time_uptime;
svc_checkidle(grp);
}
xprt = st->st_xprt;
if (!xprt) {
/*
* Enforce maxthreads count.
*/
if (grp->sg_threadcount > grp->sg_maxthreads)
break;
/*
* Before sleeping, see if we can find an
* active transport which isn't being serviced
* by a thread.
*/
if (svc_request_space_available(pool) &&
(xprt = TAILQ_FIRST(&grp->sg_active)) != NULL) {
TAILQ_REMOVE(&grp->sg_active, xprt, xp_alink);
SVC_ACQUIRE(xprt);
xprt->xp_thread = st;
st->st_xprt = xprt;
continue;
}
LIST_INSERT_HEAD(&grp->sg_idlethreads, st, st_ilink);
if (ismaster || (!ismaster &&
grp->sg_threadcount > grp->sg_minthreads))
error = cv_timedwait_sig(&st->st_cond,
&grp->sg_lock, 5 * hz);
else
error = cv_wait_sig(&st->st_cond,
&grp->sg_lock);
if (st->st_xprt == NULL)
LIST_REMOVE(st, st_ilink);
/*
* Reduce worker thread count when idle.
*/
if (error == EWOULDBLOCK) {
if (!ismaster
&& (grp->sg_threadcount
> grp->sg_minthreads)
&& !st->st_xprt)
break;
} else if (error != 0) {
KASSERT(error == EINTR || error == ERESTART,
("non-signal error %d", error));
mtx_unlock(&grp->sg_lock);
p = curproc;
PROC_LOCK(p);
if (P_SHOULDSTOP(p) ||
(p->p_flag & P_TOTAL_STOP) != 0) {
thread_suspend_check(0);
PROC_UNLOCK(p);
mtx_lock(&grp->sg_lock);
} else {
PROC_UNLOCK(p);
svc_exit(pool);
mtx_lock(&grp->sg_lock);
break;
}
}
continue;
}
mtx_unlock(&grp->sg_lock);
/*
* Drain the transport socket and queue up any RPCs.
*/
xprt->xp_lastactive = time_uptime;
do {
if (!svc_request_space_available(pool))
break;
rqstp = NULL;
stat = svc_getreq(xprt, &rqstp);
if (rqstp) {
svc_change_space_used(pool, rqstp->rq_size);
/*
* See if the application has a preference
* for some other thread.
*/
if (pool->sp_assign) {
stpref = pool->sp_assign(st, rqstp);
rqstp->rq_thread = stpref;
STAILQ_INSERT_TAIL(&stpref->st_reqs,
rqstp, rq_link);
mtx_unlock(&stpref->st_lock);
if (stpref != st)
rqstp = NULL;
} else {
rqstp->rq_thread = st;
STAILQ_INSERT_TAIL(&st->st_reqs,
rqstp, rq_link);
}
}
} while (rqstp == NULL && stat == XPRT_MOREREQS
&& grp->sg_state != SVCPOOL_CLOSING);
/*
* Move this transport to the end of the active list to
* ensure fairness when multiple transports are active.
* If this was the last queued request, svc_getreq will end
* up calling xprt_inactive to remove from the active list.
*/
mtx_lock(&grp->sg_lock);
xprt->xp_thread = NULL;
st->st_xprt = NULL;
if (xprt->xp_active) {
if (!svc_request_space_available(pool) ||
!xprt_assignthread(xprt))
TAILQ_INSERT_TAIL(&grp->sg_active,
xprt, xp_alink);
}
mtx_unlock(&grp->sg_lock);
SVC_RELEASE(xprt);
/*
* Execute what we have queued.
*/
mtx_lock(&st->st_lock);
while ((rqstp = STAILQ_FIRST(&st->st_reqs)) != NULL) {
STAILQ_REMOVE_HEAD(&st->st_reqs, rq_link);
mtx_unlock(&st->st_lock);
sz = (long)rqstp->rq_size;
svc_executereq(rqstp);
svc_change_space_used(pool, -sz);
mtx_lock(&st->st_lock);
}
mtx_unlock(&st->st_lock);
mtx_lock(&grp->sg_lock);
}
if (st->st_xprt) {
xprt = st->st_xprt;
st->st_xprt = NULL;
SVC_RELEASE(xprt);
}
KASSERT(STAILQ_EMPTY(&st->st_reqs), ("stray reqs on exit"));
mtx_destroy(&st->st_lock);
cv_destroy(&st->st_cond);
mem_free(st, sizeof(*st));
grp->sg_threadcount--;
if (!ismaster)
wakeup(grp);
mtx_unlock(&grp->sg_lock);
}
static void
svc_thread_start(void *arg)
{
svc_run_internal((SVCGROUP *) arg, FALSE);
kthread_exit();
}
static void
svc_new_thread(SVCGROUP *grp)
{
SVCPOOL *pool = grp->sg_pool;
struct thread *td;
grp->sg_threadcount++;
kthread_add(svc_thread_start, grp, pool->sp_proc, &td, 0, 0,
"%s: service", pool->sp_name);
}
void
svc_run(SVCPOOL *pool)
{
int g, i;
struct proc *p;
struct thread *td;
SVCGROUP *grp;
p = curproc;
td = curthread;
snprintf(td->td_name, sizeof(td->td_name),
"%s: master", pool->sp_name);
pool->sp_state = SVCPOOL_ACTIVE;
pool->sp_proc = p;
/* Choose group count based on number of threads and CPUs. */
pool->sp_groupcount = max(1, min(SVC_MAXGROUPS,
min(pool->sp_maxthreads / 2, mp_ncpus) / 6));
for (g = 0; g < pool->sp_groupcount; g++) {
grp = &pool->sp_groups[g];
grp->sg_minthreads = max(1,
pool->sp_minthreads / pool->sp_groupcount);
grp->sg_maxthreads = max(1,
pool->sp_maxthreads / pool->sp_groupcount);
grp->sg_lastcreatetime = time_uptime;
}
/* Starting threads */
for (g = 0; g < pool->sp_groupcount; g++) {
grp = &pool->sp_groups[g];
for (i = ((g == 0) ? 1 : 0); i < grp->sg_minthreads; i++)
svc_new_thread(grp);
}
pool->sp_groups[0].sg_threadcount++;
svc_run_internal(&pool->sp_groups[0], TRUE);
/* Waiting for threads to stop. */
for (g = 0; g < pool->sp_groupcount; g++) {
grp = &pool->sp_groups[g];
mtx_lock(&grp->sg_lock);
while (grp->sg_threadcount > 0)
msleep(grp, &grp->sg_lock, 0, "svcexit", 0);
mtx_unlock(&grp->sg_lock);
}
}
void
svc_exit(SVCPOOL *pool)
{
SVCGROUP *grp;
SVCTHREAD *st;
int g;
pool->sp_state = SVCPOOL_CLOSING;
for (g = 0; g < pool->sp_groupcount; g++) {
grp = &pool->sp_groups[g];
mtx_lock(&grp->sg_lock);
if (grp->sg_state != SVCPOOL_CLOSING) {
grp->sg_state = SVCPOOL_CLOSING;
LIST_FOREACH(st, &grp->sg_idlethreads, st_ilink)
cv_signal(&st->st_cond);
}
mtx_unlock(&grp->sg_lock);
}
}
bool_t
svc_getargs(struct svc_req *rqstp, xdrproc_t xargs, void *args)
{
struct mbuf *m;
XDR xdrs;
bool_t stat;
m = rqstp->rq_args;
rqstp->rq_args = NULL;
xdrmbuf_create(&xdrs, m, XDR_DECODE);
stat = xargs(&xdrs, args);
XDR_DESTROY(&xdrs);
return (stat);
}
bool_t
svc_freeargs(struct svc_req *rqstp, xdrproc_t xargs, void *args)
{
XDR xdrs;
if (rqstp->rq_addr) {
free(rqstp->rq_addr, M_SONAME);
rqstp->rq_addr = NULL;
}
xdrs.x_op = XDR_FREE;
return (xargs(&xdrs, args));
}
void
svc_freereq(struct svc_req *rqstp)
{
SVCTHREAD *st;
SVCPOOL *pool;
st = rqstp->rq_thread;
if (st) {
pool = st->st_pool;
if (pool->sp_done)
pool->sp_done(st, rqstp);
}
if (rqstp->rq_auth.svc_ah_ops)
SVCAUTH_RELEASE(&rqstp->rq_auth);
if (rqstp->rq_xprt) {
SVC_RELEASE(rqstp->rq_xprt);
}
if (rqstp->rq_addr)
free(rqstp->rq_addr, M_SONAME);
if (rqstp->rq_args)
m_freem(rqstp->rq_args);
free(rqstp, M_RPC);
}
Reference in New Issue View Git Blame Copy Permalink