Threads pool added

This commit is contained in:
Vladimir Dubrovin 2026-10-12 22:40:41 +03:00
parent 3d5da0105d
commit bdee95c7e0
9 changed files with 275 additions and 33 deletions

View File

@ -1395,6 +1395,21 @@ PAM (\fI/etc/pam.d\fR and the modules it names, read for every authentication),
LDAP (\fIldap.conf\fR and, for ldaps, the CA certificates), LDAP (\fIldap.conf\fR and, for ldaps, the CA certificates),
ODBC (\fIodbc.ini\fR, \fIodbcinst.ini\fR and the driver libraries, loaded again whenever the connection is reopened). ODBC (\fIodbc.ini\fR, \fIodbcinst.ini\fR and the driver libraries, loaded again whenever the connection is reopened).
.br
.BR threadpool
\fI<max_idle>\fR [\fI<idle_seconds>\fR]
.br
A thread whose connection has ended waits for the next one, of any service,
instead of ending, and keeps the memory it had for its connection: a new
connection starts without creating a thread. At most \fImax_idle\fR threads
wait, each at most \fIidle_seconds\fR (60 by default; 0 waits without a limit).
By default up to 32 threads wait for each running service, but no more than
the services' maxchild together. \fBthreadpool 0\fR ends every thread with its
connection. A thread serves a service only if its stack is at least as large as
the service needs (see \fBstacksize\fR), otherwise a new one is created.
The admin service shows how many threads wait, how many were created and how
many connections were given to a waiting one on its Running Services page.
.br .br
.BR stacksize .BR stacksize
\fI<value_to_add_to_default_stack_size>\fR \fI<value_to_add_to_default_stack_size>\fR
@ -1443,7 +1458,7 @@ as
data filtering will be performed through filtering plugins to avoid data data filtering will be performed through filtering plugins to avoid data
corruption and/or Content-Length changing. A body up to this size is read corruption and/or Content-Length changing. A body up to this size is read
whole into memory and filtered at once. 0 removes the limit, but then a body whole into memory and filtered at once. 0 removes the limit, but then a body
is only filtered whole if it fits in the connection's 16K buffer. is only filtered whole if it fits in the connection's 32K buffer.
Default is 1MB (1048576). Default is 1MB (1048576).
.SH SSL/TLS SUPPORT .SH SSL/TLS SUPPORT

View File

@ -372,6 +372,7 @@ void cyclestep(void){
conf.paused++; conf.paused++;
doschedule(); doschedule();
waitservices(); waitservices();
pooldrain();
return; return;
} }
@ -558,6 +559,7 @@ int WinMain(HINSTANCE hInstance, HINSTANCE hPrevInstance, LPWSTR lpCmdLine, int
_3proxy_mutex_init(&connlim_mutex); _3proxy_mutex_init(&connlim_mutex);
_3proxy_mutex_init(&tc_mutex); _3proxy_mutex_init(&tc_mutex);
_3proxy_mutex_init(&log_mutex); _3proxy_mutex_init(&log_mutex);
_3proxy_mutex_init(&tpool.mutex);
#ifndef NORADIUS #ifndef NORADIUS
_3proxy_mutex_init(&rad_mutex); _3proxy_mutex_init(&rad_mutex);
#endif #endif

View File

@ -51,6 +51,12 @@ int socks5_setaddr(int family, int atyp, const unsigned char *addr, PROXYSOCKADD
thread does without making a socket of its own, a name looked up through thread does without making a socket of its own, a name looked up through
the system, a log line, RADIUS, happens in the main namespace. */ the system, a log line, RADIUS, happens in the main namespace. */
int mainnsfd = -1; int mainnsfd = -1;
/* set in a thread left outside the main namespace, which is not reused */
static __thread int nsfailed;
int nsbroken(void){
return nsfailed;
}
/* Opens the main namespace, once and while /proc can still be reached: a /* Opens the main namespace, once and while /proc can still be reached: a
chroot may take it away. */ chroot may take it away. */
@ -72,6 +78,7 @@ SOCKET nssocket(struct srvparam *srv, struct sockfuncs *sf, void *state, int sid
/* Left in the other namespace: no socket from here, and the /* Left in the other namespace: no socket from here, and the
connection fails. The thread ends with it. */ connection fails. The thread ends with it. */
if(s != INVALID_SOCKET) sf->_closesocket(state, s); if(s != INVALID_SOCKET) sf->_closesocket(state, s);
nsfailed = 1;
return INVALID_SOCKET; return INVALID_SOCKET;
} }
errno = err; errno = err;
@ -221,6 +228,8 @@ size_t threadstacksize(int extra){
} }
#endif #endif
struct threadpool tpool;
struct extparam conf = { struct extparam conf = {
.timeouts = timeouts, .timeouts = timeouts,
.acl = NULL, .acl = NULL,
@ -231,6 +240,8 @@ struct extparam conf = {
.trafcounter = NULL, .trafcounter = NULL,
.services = NULL, .services = NULL,
.stacksize = 0, .stacksize = 0,
.poolmax = -1,
.pooltimeout = 60,
.counterd = -1, .counterd = -1,
.haveerror = 0, .haveerror = 0,
.rotate = 0, .rotate = 0,

View File

@ -425,6 +425,12 @@ static int h_stacksize(int argc, unsigned char **argv){
return 0; return 0;
} }
static int h_threadpool(int argc, unsigned char **argv){
conf.poolmax = atoi((char *)argv[1]);
if(argc > 2) conf.pooltimeout = atoi((char *)argv[2]);
return 0;
}
static int h_force(int argc, unsigned char **argv){ static int h_force(int argc, unsigned char **argv){
conf.noforce = 0; conf.noforce = 0;
@ -2095,6 +2101,7 @@ struct commands commandhandlers[]={
{NULL, "delimchar",h_delimchar, 2, 2}, {NULL, "delimchar",h_delimchar, 2, 2},
{NULL, "authnserver", h_authnserver, 2, 2}, {NULL, "authnserver", h_authnserver, 2, 2},
{NULL, "stacksize", h_stacksize, 2, 2}, {NULL, "stacksize", h_stacksize, 2, 2},
{NULL, "threadpool", h_threadpool, 2, 3},
{NULL, "force", h_force, 1, 1}, {NULL, "force", h_force, 1, 1},
{NULL, "noforce", h_noforce, 1, 1}, {NULL, "noforce", h_noforce, 1, 1},
{NULL, "parentretries", h_parentretries, 2, 2}, {NULL, "parentretries", h_parentretries, 2, 2},
@ -2391,6 +2398,8 @@ void freeconf(struct extparam *confp){
confp->rotate = 0; confp->rotate = 0;
confp->logtype = NONE; confp->logtype = NONE;
confp->logtime = confp->time = 0; confp->logtime = confp->time = 0;
confp->poolmax = -1;
confp->pooltimeout = 60;
archiverc = confp->archiverc; archiverc = confp->archiverc;
confp->archiverc = 0; confp->archiverc = 0;

View File

@ -534,6 +534,26 @@ static void * ef_server_childcount(struct node * node){
return &((struct srvparam *)node->value) -> childcount; return &((struct srvparam *)node->value) -> childcount;
} }
static void * ef_pool_idle(struct node * node){
return &((struct threadpool *)node->value) -> nidle;
}
static void * ef_pool_limit(struct node * node){
return &((struct threadpool *)node->value) -> limit;
}
static void * ef_pool_timeout(struct node * node){
return &conf.pooltimeout;
}
static void * ef_pool_created(struct node * node){
return &((struct threadpool *)node->value) -> created;
}
static void * ef_pool_reused(struct node * node){
return &((struct threadpool *)node->value) -> reused;
}
static void * ef_server_maxchild(struct node * node){ static void * ef_server_maxchild(struct node * node){
return &((struct srvparam *)node->value) -> maxchild; return &((struct srvparam *)node->value) -> maxchild;
} }
@ -892,6 +912,14 @@ static struct property prop_server[] = {
}; };
static struct property prop_threadpool[] = {
{"idle", ef_pool_idle, TYPE_UNSIGNED, "threads waiting for a connection"},
{"limit", ef_pool_limit, TYPE_UNSIGNED, "most threads kept waiting"},
{"timeout", ef_pool_timeout, TYPE_INTEGER, "seconds a thread waits before it ends"},
{"created", ef_pool_created, TYPE_UNSIGNED64, "threads started"},
{"reused", ef_pool_reused, TYPE_UNSIGNED64, "connections given to a waiting thread"}
};
static struct property prop_client[] = { static struct property prop_client[] = {
{"servicetype", ef_client_type, TYPE_STRING, "type of the client"}, {"servicetype", ef_client_type, TYPE_STRING, "type of the client"},
{"threadid", ef_client_threadid, TYPE_UNSIGNED64, "process thread id"}, {"threadid", ef_client_threadid, TYPE_UNSIGNED64, "process thread id"},
@ -950,5 +978,6 @@ struct datatype datatypes[64] = {
{"period", ef_period_next, NULL, prop_period, sizeof(prop_period)/sizeof(struct property)}, {"period", ef_period_next, NULL, prop_period, sizeof(prop_period)/sizeof(struct property)},
{"server", ef_server_next, NULL, prop_server, sizeof(prop_server)/sizeof(struct property)}, {"server", ef_server_next, NULL, prop_server, sizeof(prop_server)/sizeof(struct property)},
{"password", NULL, pr_password, NULL, 0}, {"password", NULL, pr_password, NULL, 0},
{"hostname", ef_hostname_next, pr_hostname, prop_hostname, sizeof(prop_hostname)/sizeof(struct property)} {"hostname", ef_hostname_next, pr_hostname, prop_hostname, sizeof(prop_hostname)/sizeof(struct property)},
{"threadpool", NULL, NULL, prop_threadpool, sizeof(prop_threadpool)/sizeof(struct property)}
}; };

View File

@ -245,6 +245,9 @@ int socks5_setaddr(int family, int atyp, const unsigned char *addr, PROXYSOCKADD
#ifdef __linux__ #ifdef __linux__
extern int mainnsfd; extern int mainnsfd;
int nsopen(void); int nsopen(void);
int nsbroken(void);
#else
#define nsbroken() 0
#endif #endif
/* a function everywhere, never a macro: its callers have preprocessor /* a function everywhere, never a macro: its callers have preprocessor
lines among its arguments */ lines among its arguments */
@ -302,6 +305,8 @@ uint32_t getip46(int family, unsigned char *name, struct sockaddr *sa);
int afdetect(unsigned char *name); int afdetect(unsigned char *name);
uint32_t myresolver(int, unsigned char *, unsigned char *); uint32_t myresolver(int, unsigned char *, unsigned char *);
uint32_t fakeresolver (int, unsigned char *, unsigned char*); uint32_t fakeresolver (int, unsigned char *, unsigned char*);
void releaseparam(struct clientparam * param);
void pooldrain(void);
void freeparam(struct clientparam * param); void freeparam(struct clientparam * param);
void clearstat(struct clientparam * param); void clearstat(struct clientparam * param);
void dumpcounters(struct trafcount *tl, int counterd); void dumpcounters(struct trafcount *tl, int counterd);

View File

@ -14,7 +14,7 @@
/* Child functions do not call each other, a child requesting redirection to /* Child functions do not call each other, a child requesting redirection to
another child returns it instead of calling it, to keep the stack flat. another child returns it instead of calling it, to keep the stack flat.
The child which completed the request returns NULL. param is logged by The child which completed the request returns NULL. param is logged by
the child and released here. the child and released here, but not freed: its thread may keep it.
*/ */
void * childfunc(struct clientparam * param){ void * childfunc(struct clientparam * param){
PROXYFUNC pf = param->srv->pf; PROXYFUNC pf = param->srv->pf;
@ -25,28 +25,21 @@ void * childfunc(struct clientparam * param){
param->res = 101; param->res = 101;
dolog(param, (unsigned char *)"Redirection loop"); dolog(param, (unsigned char *)"Redirection loop");
} }
freeparam(param); releaseparam(param);
return NULL; return NULL;
} }
#define param ((struct clientparam *) p) /* One connection, from the filters to the end of the child which answers
#ifdef _WIN32 it. The structure is released, not freed, when it returns. */
DWORD WINAPI threadfunc(LPVOID p) { static void serveparam(struct clientparam *param){
#else
void * threadfunc (void *p) {
#endif
int i = -1; int i = -1;
#ifdef MODULEMAINFUNC #ifdef MODULEMAINFUNC
if(makefilters(param->srv, param) > CONTINUE){ if(makefilters(param->srv, param) > CONTINUE){
#ifndef NOUDPMAIN #ifndef NOUDPMAIN
if(param->srv->service == S_UDPPM) _3proxy_sem_unlock(udpinit); if(param->srv->service == S_UDPPM) _3proxy_sem_unlock(udpinit);
#endif #endif
freeparam(param); releaseparam(param);
#ifdef _WIN32 return;
return 0;
#else
return NULL;
#endif
} }
#endif #endif
if(param->srv->cbsock != INVALID_SOCKET){ if(param->srv->cbsock != INVALID_SOCKET){
@ -96,7 +89,7 @@ void * threadfunc (void *p) {
} }
if(!i){ if(!i){
param->res = 13; param->res = 13;
freeparam(param); releaseparam(param);
} }
else { else {
@ -139,15 +132,101 @@ void * threadfunc (void *p) {
} }
} }
} }
childfunc((struct clientparam *)p); childfunc(param);
} }
}
/* Frees a worker's structure, once its thread is done with it. */
static void freeworker(struct clientparam *param){
releaseparam(param);
_3proxy_sem_destroy(param->wake);
free(param);
}
/* The automatic cap: 32 idle workers for each running service, but never
more than all of them could have connections. */
static unsigned poollimit(void){
unsigned limit;
if(conf.poolmax >= 0) limit = (unsigned)conf.poolmax;
else {
limit = 32 * tpool.services;
if(limit > tpool.maxchild) limit = tpool.maxchild;
}
tpool.limit = limit;
return limit;
}
/* Parks a worker whose connection has ended and waits for the next one.
Returns 0 when the thread is to end instead: on exit, when the pool is
full, when the thread was left in another network namespace, or when it
waited longer than the idle timeout. */
static int poolpark(struct clientparam *param){
int timeout;
_3proxy_mutex_lock(&tpool.mutex);
if(conf.timetoexit || nsbroken() || tpool.nidle >= poollimit()){
_3proxy_mutex_unlock(&tpool.mutex);
return 0;
}
param->poolnext = tpool.idle;
tpool.idle = param;
tpool.nidle++;
param->pooled = 1;
timeout = conf.pooltimeout;
_3proxy_mutex_unlock(&tpool.mutex);
if(timeout <= 0) _3proxy_sem_lock(param->wake);
else if(_3proxy_sem_timedlock(param->wake, (unsigned)timeout * 1000)){
struct clientparam **pp;
_3proxy_mutex_lock(&tpool.mutex);
if(param->pooled){
for(pp = &tpool.idle; *pp != param; pp = &(*pp)->poolnext);
*pp = param->poolnext;
tpool.nidle--;
param->pooled = 0;
_3proxy_mutex_unlock(&tpool.mutex);
return 0;
}
_3proxy_mutex_unlock(&tpool.mutex);
/* taken from the pool as the wait ran out: the connection follows */
_3proxy_sem_lock(param->wake);
}
return !param->poolexit;
}
/* Wakes every idle worker to end, when the process exits. */
void pooldrain(void){
struct clientparam *param;
_3proxy_mutex_lock(&tpool.mutex);
while((param = tpool.idle)){
tpool.idle = param->poolnext;
tpool.nidle--;
param->pooled = 0;
param->poolexit = 1;
_3proxy_sem_unlock(param->wake);
}
_3proxy_mutex_unlock(&tpool.mutex);
}
#ifdef _WIN32
DWORD WINAPI threadfunc(LPVOID p) {
#else
void * threadfunc (void *p) {
#endif
struct clientparam *param = (struct clientparam *)p;
do serveparam(param);
while(poolpark(param));
freeworker(param);
#ifdef _WIN32 #ifdef _WIN32
return 0; return 0;
#else #else
return NULL; return NULL;
#endif #endif
} }
#undef param
#ifdef _WIN32 #ifdef _WIN32
/* Present since Windows 7 (SO_PORT_SCALABILITY) and Windows 10 / Server 2019 /* Present since Windows 7 (SO_PORT_SCALABILITY) and Windows 10 / Server 2019
@ -367,7 +446,7 @@ static int srvprepare(struct srvstart *s, int argc, char **argv){
" -fFORMAT logging format (see documentation)\n" " -fFORMAT logging format (see documentation)\n"
" -l log to stderr\n" " -l log to stderr\n"
" -lFILENAME log to FILENAME\n" " -lFILENAME log to FILENAME\n"
" -b(BUFSIZE) size of network buffer (default 65536 for TCP, 16384 for UDP)\n" " -b(BUFSIZE) size of network buffer, at most and by default 32768\n"
" -S(STACKSIZE) value to add to default client thread stack size\n" " -S(STACKSIZE) value to add to default client thread stack size\n"
" -t be silent (do not log service start/stop)\n" " -t be silent (do not log service start/stop)\n"
"\n" "\n"
@ -1046,15 +1125,35 @@ static int udpnext(struct srvstart *s){
static int srvspawn(struct srvstart *s, SOCKET client_sock){ static int srvspawn(struct srvstart *s, SOCKET client_sock){
struct srvparam *srv = &s->srv; struct srvparam *srv = &s->srv;
struct clientparam *defparam = &s->defparam; struct clientparam *defparam = &s->defparam;
struct clientparam * newparam; struct clientparam * newparam = NULL, **pp;
pthread_t thread; pthread_t thread;
unsigned char buf[256]; unsigned char buf[256];
int error = 0; int error = 0, reused;
size_t need;
#ifdef _WIN32 #ifdef _WIN32
HANDLE h; HANDLE h;
need = 16384 + srv->stacksize;
#else
need = threadstacksize(srv->stacksize);
#endif #endif
if(! (newparam = malloc (sizeof(*defparam)))){ /* an idle worker if one has the stack this service needs */
_3proxy_mutex_lock(&tpool.mutex);
for(pp = &tpool.idle; *pp; pp = &(*pp)->poolnext){
if((*pp)->stacksize >= need){
newparam = *pp;
*pp = newparam->poolnext;
newparam->pooled = 0;
tpool.nidle--;
tpool.reused++;
break;
}
}
_3proxy_mutex_unlock(&tpool.mutex);
reused = newparam != NULL;
if(!reused && (!(newparam = malloc (sizeof(*defparam))) || _3proxy_sem_init_timed(newparam->wake, 0, 1))){
if(newparam) free(newparam);
if(!s->isudp) srv->so._closesocket(srv->so.state, client_sock); if(!s->isudp) srv->so._closesocket(srv->so.state, client_sock);
#ifndef NOUDPMAIN #ifndef NOUDPMAIN
else { else {
@ -1066,6 +1165,11 @@ static int srvspawn(struct srvstart *s, SOCKET client_sock){
usleep(SLEEPTIME); usleep(SLEEPTIME);
return 1; return 1;
}; };
if(!reused){
newparam->stacksize = need;
newparam->pooled = newparam->poolexit = 0;
newparam->poolnext = NULL;
}
memcpy(newparam, defparam, offsetof(struct clientparam, filters)); memcpy(newparam, defparam, offsetof(struct clientparam, filters));
newparam->clibuf = newparam->clibufdata; newparam->clibuf = newparam->clibufdata;
newparam->srvbuf = newparam->srvbufdata; newparam->srvbuf = newparam->srvbufdata;
@ -1081,6 +1185,8 @@ static int srvspawn(struct srvstart *s, SOCKET client_sock){
newparam->next = srv->child; newparam->next = srv->child;
srv->child = srv->child->prev = newparam; srv->child = srv->child->prev = newparam;
} }
/* a worker from the pool is woken below, once it is one of the children */
if(!reused){
#ifdef _WIN32 #ifdef _WIN32
#ifndef _WINCE #ifndef _WINCE
h = (HANDLE)_beginthreadex((LPSECURITY_ATTRIBUTES )NULL, (unsigned)(16384 + srv->stacksize), (void *)threadfunc, (void *) newparam, 0, &thread); h = (HANDLE)_beginthreadex((LPSECURITY_ATTRIBUTES )NULL, (unsigned)(16384 + srv->stacksize), (void *)threadfunc, (void *) newparam, 0, &thread);
@ -1100,6 +1206,7 @@ static int srvspawn(struct srvstart *s, SOCKET client_sock){
sprintf((char *)buf, "pthread_create(): %s", strerror(error)); sprintf((char *)buf, "pthread_create(): %s", strerror(error));
} }
#endif #endif
}
if(error){ if(error){
if(!srv->silent)dolog(defparam, buf); if(!srv->silent)dolog(defparam, buf);
if(newparam->prev) newparam->prev->next = newparam->next; if(newparam->prev) newparam->prev->next = newparam->next;
@ -1116,13 +1223,20 @@ static int srvspawn(struct srvstart *s, SOCKET client_sock){
_3proxy_sem_unlock(udpinit); _3proxy_sem_unlock(udpinit);
} }
#endif #endif
freeparam(newparam); freeworker(newparam);
} }
else { else {
srv->childcount++; srv->childcount++;
newparam->threadid = (uint64_t)thread; if(!reused){
newparam->workerid = (uint64_t)thread;
_3proxy_mutex_lock(&tpool.mutex);
tpool.created++;
_3proxy_mutex_unlock(&tpool.mutex);
}
newparam->threadid = newparam->workerid;
} }
_3proxy_mutex_unlock(&srv->counter_mutex); _3proxy_mutex_unlock(&srv->counter_mutex);
if(reused) _3proxy_sem_unlock(newparam->wake);
memset(&defparam->sincl, 0, sizeof(defparam->sincl)); memset(&defparam->sincl, 0, sizeof(defparam->sincl));
memset(&defparam->sincr, 0, sizeof(defparam->sincr)); memset(&defparam->sincr, 0, sizeof(defparam->sincr));
return error; return error;
@ -1158,6 +1272,11 @@ static int srvrun(struct srvstart *s){
if(!s->isudp && s->iscbl && (res = cblisten(s))) goto EXIT; if(!s->isudp && s->iscbl && (res = cblisten(s))) goto EXIT;
#endif #endif
/* counted for the automatic cap of the thread pool while it runs */
_3proxy_mutex_lock(&tpool.mutex);
tpool.services++;
tpool.maxchild += srv->maxchild;
_3proxy_mutex_unlock(&tpool.mutex);
for (;;) { for (;;) {
for(;;){ for(;;){
/* A connection is only taken with a slot of maxchild in hand, which /* A connection is only taken with a slot of maxchild in hand, which
@ -1198,6 +1317,10 @@ static int srvrun(struct srvstart *s){
if(next == NEXT_RETRY) continue; if(next == NEXT_RETRY) continue;
if(!srvspawn(s, new_sock)) haveslot = 0; if(!srvspawn(s, new_sock)) haveslot = 0;
} }
_3proxy_mutex_lock(&tpool.mutex);
tpool.services--;
tpool.maxchild -= srv->maxchild;
_3proxy_mutex_unlock(&tpool.mutex);
EXIT: EXIT:
#ifndef STDMAIN #ifndef STDMAIN
@ -1262,6 +1385,7 @@ int WinMain(HINSTANCE hInstance, HINSTANCE hPrevInstance, LPWSTR lpCmdLine, int
WSAStartup(MAKEWORD( 1, 1 ), &wd); WSAStartup(MAKEWORD( 1, 1 ), &wd);
#endif #endif
_3proxy_mutex_init(&tpool.mutex);
if((res = srvprepare(&s, argc, argv))) return res; if((res = srvprepare(&s, argc, argv))) return res;
#ifndef _WIN32 #ifndef _WIN32
@ -1277,12 +1401,19 @@ int WinMain(HINSTANCE hInstance, HINSTANCE hPrevInstance, LPWSTR lpCmdLine, int
if(! (newparam = malloc (sizeof(s.defparam)))){ if(! (newparam = malloc (sizeof(s.defparam)))){
return 2; return 2;
}; };
*newparam = s.defparam; memcpy(newparam, &s.defparam, offsetof(struct clientparam, filters));
return(childfunc(newparam)? 1:0); newparam->clibuf = newparam->clibufdata;
newparam->srvbuf = newparam->srvbufdata;
newparam->clibufsize = newparam->srvbufsize = SRVBUFSIZE(&s.srv);
childfunc(newparam);
free(newparam);
return 0;
} }
#endif #endif
return srvrun(&s); res = srvrun(&s);
pooldrain();
return res;
} }
#else #else
@ -1478,7 +1609,11 @@ void srvfree(struct srvparam * srv){
} }
void freeparam(struct clientparam * param) { /* Ends a connection: everything it holds is released and it leaves its
service, but the structure itself stays, with nothing in it pointing to
what was freed. After this the service may be gone at any moment, so
param->srv is cleared and nothing here touches it again. */
void releaseparam(struct clientparam * param) {
if(param->srv){ if(param->srv){
if(param->srv->so.freefunc) param->srv->so.freefunc(param->sostate); if(param->srv->so.freefunc) param->srv->so.freefunc(param->sostate);
#ifndef STDMAIN #ifndef STDMAIN
@ -1491,11 +1626,14 @@ void freeparam(struct clientparam * param) {
} }
if(param->clibuf != param->clibufdata) free(param->clibuf); if(param->clibuf != param->clibufdata) free(param->clibuf);
if(param->srvbuf != param->srvbufdata) free(param->srvbuf); if(param->srvbuf != param->srvbufdata) free(param->srvbuf);
param->clibuf = param->clibufdata;
param->srvbuf = param->srvbufdata;
#ifdef WITHSPLICE #ifdef WITHSPLICE
if(param->pipecli[0] >= 0) close(param->pipecli[0]); if(param->pipecli[0] >= 0) close(param->pipecli[0]);
if(param->pipecli[1] >= 0) close(param->pipecli[1]); if(param->pipecli[1] >= 0) close(param->pipecli[1]);
if(param->pipesrv[0] >= 0) close(param->pipesrv[0]); if(param->pipesrv[0] >= 0) close(param->pipesrv[0]);
if(param->pipesrv[1] >= 0) close(param->pipesrv[1]); if(param->pipesrv[1] >= 0) close(param->pipesrv[1]);
param->pipecli[0] = param->pipecli[1] = param->pipesrv[0] = param->pipesrv[1] = -1;
#endif #endif
/* Not shut down on Windows: there shutting down receiving resets the /* Not shut down on Windows: there shutting down receiving resets the
connection as soon as the peer sends anything more, and close() connection as soon as the peer sends anything more, and close()
@ -1527,6 +1665,7 @@ void freeparam(struct clientparam * param) {
#endif #endif
param->srv->so._closesocket(param->sostate, param->clisock); param->srv->so._closesocket(param->sostate, param->clisock);
} }
param->ctrlsocksrv = param->ctrlsock = param->remsock = param->clisock = INVALID_SOCKET;
} }
#ifndef STDMAIN #ifndef STDMAIN
while(param->nfilters > 0){ while(param->nfilters > 0){
@ -1536,6 +1675,8 @@ void freeparam(struct clientparam * param) {
} }
if(param->connlim) stopconnlims(param); if(param->connlim) stopconnlims(param);
#endif #endif
/* the sockets were closed and the filters cleared with it */
param->sostate = NULL;
/* Leaving the service comes last. Once its last child is gone, srvfree() /* Leaving the service comes last. Once its last child is gone, srvfree()
frees the service along with its socket functions, filters and mutex, frees the service along with its socket functions, filters and mutex,
and a close above may have lingered for seconds before getting here. */ and a close above may have lingered for seconds before getting here. */
@ -1556,7 +1697,13 @@ void freeparam(struct clientparam * param) {
_3proxy_sem_unlock(param->srv->childslots); _3proxy_sem_unlock(param->srv->childslots);
} }
_3proxy_mutex_unlock(&param->srv->counter_mutex); _3proxy_mutex_unlock(&param->srv->counter_mutex);
param->prev = param->next = NULL;
param->srv = NULL;
} }
}
void freeparam(struct clientparam * param) {
releaseparam(param);
free(param); free(param);
} }

View File

@ -199,9 +199,7 @@ int
#define MAX_HASH_SIZE (64) #define MAX_HASH_SIZE (64)
#define UDPBUFSIZE 16384 #define UDPBUFSIZE 16384
/* A connection's buffer for each side: 16K, and the 5 byte header of a TLS #define CLIENTBUFSIZE 32768
record that big, which tlspr reads whole. */
#define CLIENTBUFSIZE (16384 + 64)
#define CLIENTTMPBUFSIZE 8192 #define CLIENTTMPBUFSIZE 8192
#define HOSTNAMELEN 256 #define HOSTNAMELEN 256
#define USERNAMELEN 128 #define USERNAMELEN 128
@ -805,6 +803,14 @@ struct clientparam {
*hdrfilterscli[MAX_FILTERS], *hdrfilterssrv[MAX_FILTERS], *hdrfilterscli[MAX_FILTERS], *hdrfilterssrv[MAX_FILTERS],
*predatfilters[MAX_FILTERS], *datfilterscli[MAX_FILTERS], *predatfilters[MAX_FILTERS], *datfilterscli[MAX_FILTERS],
*datfilterssrv[MAX_FILTERS]; *datfilterssrv[MAX_FILTERS];
/* The thread this structure belongs to, which keeps it from one
connection to the next while it waits in the pool. */
struct clientparam *poolnext; /* next idle worker */
_3proxy_sem_t wake; /* posted when given a connection */
size_t stacksize; /* of its thread */
uint64_t workerid; /* the thread's id */
int pooled, /* waiting in the pool */
poolexit; /* woken to end instead */
/* The connection's scratch buffer, for a log line and the like: for /* The connection's scratch buffer, for a log line and the like: for
something done at once, never for what has to outlive a call into something done at once, never for what has to outlive a call into
anything that could log. */ anything that could log. */
@ -815,6 +821,19 @@ struct clientparam {
srvbufdata[CLIENTBUFSIZE]; srvbufdata[CLIENTBUFSIZE];
}; };
/* Threads whose connections have ended wait here for the next one, with
the clientparam they keep, for any service whose stack fits theirs. */
struct threadpool {
_3proxy_mutex_t mutex;
struct clientparam *idle; /* most recently parked first */
unsigned nidle;
unsigned limit; /* the cap last applied */
unsigned services; /* running services and their maxchild, */
unsigned maxchild; /* for the automatic cap */
uint64_t created, reused;
};
extern struct threadpool tpool;
struct filemon { struct filemon {
char * path; char * path;
struct stat sb; struct stat sb;
@ -838,6 +857,7 @@ struct extparam {
demon, maxchild, backlog, needreload, timetoexit, version, noforce, bandlimver, parentretries; demon, maxchild, backlog, needreload, timetoexit, version, noforce, bandlimver, parentretries;
unsigned authcachetype, authcachetime; unsigned authcachetype, authcachetime;
int filtermaxsize; int filtermaxsize;
int poolmax, pooltimeout; /* threadpool: idle workers kept, -1 automatic */
int gracetraf, gracenum, gracedelay; int gracetraf, gracenum, gracedelay;
int maxseg; int maxseg;
unsigned char *logname, **archiver; unsigned char *logname, **archiver;
@ -1051,7 +1071,8 @@ typedef enum {
TYPE_PERIOD, TYPE_PERIOD,
TYPE_SERVER, TYPE_SERVER,
TYPE_PASSWORD, TYPE_PASSWORD,
TYPE_HOSTNAME TYPE_HOSTNAME,
TYPE_THREADPOOL
}DATA_TYPE; }DATA_TYPE;
#ifdef __cplusplus #ifdef __cplusplus

View File

@ -519,6 +519,9 @@ int op_admin_services(struct httpreq *r, const unsigned char *params)
if(sel[0] == 'X') printstr(&pp, style); if(sel[0] == 'X') printstr(&pp, style);
else { else {
printstr(&pp, xml); printstr(&pp, xml);
printstr(&pp, "<threadpool>");
printval(&tpool, TYPE_THREADPOOL, 0, &pp);
printstr(&pp, "</threadpool>\r\n");
printval(conf.services, TYPE_SERVER, 0, &pp); printval(conf.services, TYPE_SERVER, 0, &pp);
printstr(&pp, postxml); printstr(&pp, postxml);
} }