mirror of
https://github.com/3proxy/3proxy.git
synced 2026-11-05 09:25:49 +08:00
Clients of TCP services on event loops, served by fibers
With -L, or loopmode on for the services which follow, the clients of a TCP service are served on event loops: loops threads (by default one for each CPU) each run a loop, and each client is given in turn to one of them, which serves it on a fiber with the same code a thread runs. Its waits in sockpoll() wait in the loop, its sockets and a timer registered there, and the fiber goes on once one is ready or the time is out; a sleep is a timer alone. A client costs a stack, as large as its thread's would be, and no thread. The loops start with the first client served on one and run as long as the process; each keeps up to 256 idle fibers for its next clients. srvspawn() posts the client to its loop once it is counted, what is posted made before so that posting cannot fail. A service whose clients cannot share a thread (threadperclient), or listening for connections back (-R), all of its clients waiting for one socket, which a loop waits for once, keeps a thread for each client, as does every service of a build or a processor without fibers, and the standalone services. tests/run.py --loopmode runs every configuration with loopmode on, on two loops. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
parent
e250f98598
commit
c55f736a0d
@ -246,6 +246,11 @@ authorization and to drop the per-destination log records only.
|
||||
.br
|
||||
(for udppm) singlepacket, expect only one packet from both client and server
|
||||
.br
|
||||
.B -L
|
||||
(for TCP services) serve the clients on event loops (see \fBloopmode\fR): a
|
||||
few threads each serve many clients, instead of a thread for each.
|
||||
\fB-L0\fR gives each client a thread of its own whatever \fBloopmode\fR says.
|
||||
.br
|
||||
.B -u
|
||||
Never ask for username/password
|
||||
.br
|
||||
@ -1510,6 +1515,38 @@ 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 loopmode
|
||||
\fIon\fR|\fIoff\fR
|
||||
.br
|
||||
Whether the clients of the TCP services which follow (until the next
|
||||
\fBloopmode\fR) are served on event loops: a few threads (see \fBloops\fR)
|
||||
each wait for the sockets and timers of many clients and switch between
|
||||
them, each client running on a stack of its own as large as its thread's
|
||||
would be. A client costs a stack and no thread, and waking it costs no
|
||||
switch of threads, so many connections take less memory and CPU. Off by
|
||||
default; the \fB-L\fR option of a service overrides it, \fB-L0\fR for a
|
||||
thread for each client.
|
||||
.br
|
||||
Clients keep a thread each, whatever is asked, where something they use
|
||||
would hold up the others sharing their thread: a service with FilePlugin,
|
||||
with PAM, LDAP or Windows authentication, resolving names through the
|
||||
system (\fBsysresolve\fR, or no name server given or found), logging
|
||||
through ODBC, using a \fB-ne\fR network namespace, or listening for
|
||||
connections back (\fB-R\fR); and where there are no event loops: on a
|
||||
processor other than x86, x86-64, ARM, AArch64 and MIPS (on Windows, any),
|
||||
where the system checks returns against a shadow stack, or in the
|
||||
standalone services (\fBproxy\fR(8) and others).
|
||||
|
||||
.br
|
||||
.BR loops
|
||||
\fI<number>\fR
|
||||
.br
|
||||
The number of threads running event loops for the services of
|
||||
\fBloopmode\fR, by default (0) one for each CPU. The loops start with the
|
||||
first client served on one and run as long as 3proxy does: a change takes
|
||||
effect only on a restart. Clients are given to the loops in turn.
|
||||
|
||||
.br
|
||||
.BR stacksize
|
||||
\fI<value_to_add_to_default_stack_size>\fR
|
||||
|
||||
16
src/common.c
16
src/common.c
@ -470,27 +470,35 @@ struct sockfuncs so = {
|
||||
for, and events are given back as they were. With no fds it only
|
||||
sleeps. */
|
||||
#define POLLFAIL (POLLERR|POLLHUP|POLLNVAL)
|
||||
/* set once clients are served by fibers on event loops: the wait of a
|
||||
fiber, -2 for anything else */
|
||||
int (*fiberpollfunc)(struct pollfd *fds, unsigned nfds, int timeout) = NULL;
|
||||
|
||||
int sockpoll(struct sockfuncs *sf, void *state, struct pollfd *fds, unsigned nfds, int timeout){
|
||||
short own[64], *events = own;
|
||||
unsigned i;
|
||||
int res;
|
||||
|
||||
if(!nfds){
|
||||
if(timeout > 0) usleep(timeout * SLEEPTIME);
|
||||
if(timeout > 0 && (!fiberpollfunc || fiberpollfunc(NULL, 0, timeout) == -2))
|
||||
usleep(timeout * SLEEPTIME);
|
||||
return 0;
|
||||
}
|
||||
if(nfds > sizeof(own)/sizeof(own[0]) && !(events = malloc(nfds * sizeof(*events)))) return -1;
|
||||
for(i = 0; i < nfds; i++) events[i] = fds[i].events;
|
||||
if(!(res = sf->_prepoll(state, fds, nfds))){
|
||||
/* a fiber waits in its loop */
|
||||
if(!fiberpollfunc || (res = fiberpollfunc(fds, nfds, timeout)) == -2){
|
||||
#ifndef WITH_POLL
|
||||
#ifndef WITH_WSAPOLL
|
||||
res = mypoll(fds, nfds, timeout);
|
||||
res = mypoll(fds, nfds, timeout);
|
||||
#else
|
||||
res = WSAPoll(fds, nfds, timeout);
|
||||
res = WSAPoll(fds, nfds, timeout);
|
||||
#endif
|
||||
#else
|
||||
res = poll(fds, (nfds_t)nfds, timeout);
|
||||
res = poll(fds, (nfds_t)nfds, timeout);
|
||||
#endif
|
||||
}
|
||||
for(i = 0; res > 0 && i < nfds; i++)
|
||||
if(fds[i].events != events[i] && (fds[i].revents & ~POLLFAIL))
|
||||
fds[i].revents = events[i] | (fds[i].revents & POLLFAIL);
|
||||
|
||||
15
src/conf.c
15
src/conf.c
@ -418,6 +418,19 @@ static int h_stacksize(int argc, unsigned char **argv){
|
||||
return 0;
|
||||
}
|
||||
|
||||
static int h_loops(int argc, unsigned char **argv){
|
||||
conf->loops = atoi((char *)argv[1]);
|
||||
if(conf->loops < 0) conf->loops = 0;
|
||||
return 0;
|
||||
}
|
||||
|
||||
static int h_loopmode(int argc, unsigned char **argv){
|
||||
if(!strcmp((char *)argv[1], "on")) conf->loopmode = 1;
|
||||
else if(!strcmp((char *)argv[1], "off")) conf->loopmode = 0;
|
||||
else return 1;
|
||||
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]);
|
||||
@ -2133,6 +2146,8 @@ struct commands commandhandlers[]={
|
||||
{NULL, "delimchar",h_delimchar, 2, 2},
|
||||
{NULL, "authnserver", h_authnserver, 2, 2},
|
||||
{NULL, "stacksize", h_stacksize, 2, 2},
|
||||
{NULL, "loops", h_loops, 2, 2},
|
||||
{NULL, "loopmode", h_loopmode, 2, 2},
|
||||
{NULL, "minlogqueue", h_minlogqueue, 2, 2},
|
||||
{NULL, "maxlogqueue", h_maxlogqueue, 2, 2},
|
||||
{NULL, "threadpool", h_threadpool, 2, 3},
|
||||
|
||||
@ -113,9 +113,12 @@ static void serveparam(struct clientparam *param){
|
||||
|
||||
#ifndef WITHMAIN
|
||||
#ifndef _WIN32
|
||||
sigset_t mask;
|
||||
sigfillset(&mask);
|
||||
pthread_sigmask(SIG_SETMASK, &mask, NULL);
|
||||
/* a loop's thread blocked them all as it started */
|
||||
if(!onlooper()){
|
||||
sigset_t mask;
|
||||
sigfillset(&mask);
|
||||
pthread_sigmask(SIG_SETMASK, &mask, NULL);
|
||||
}
|
||||
#endif
|
||||
#endif
|
||||
|
||||
@ -246,8 +249,8 @@ void * threadfunc (void *p) {
|
||||
#endif
|
||||
}
|
||||
|
||||
/* A thread for the client, client_sock for TCP. 0 once it runs, the slot
|
||||
in hand going with it. */
|
||||
/* A thread for the client, client_sock for TCP, or a fiber of an event
|
||||
loop. 0 once it runs, the slot in hand going with it. */
|
||||
static int srvspawn(struct srvstart *s, SOCKET client_sock){
|
||||
struct srvparam *srv = &s->srv;
|
||||
struct clientparam *defparam = &s->defparam;
|
||||
@ -256,6 +259,7 @@ static int srvspawn(struct srvstart *s, SOCKET client_sock){
|
||||
unsigned char buf[256];
|
||||
int error = 0, reused;
|
||||
size_t need;
|
||||
struct loopjob *job = NULL;
|
||||
#ifdef _WIN32
|
||||
HANDLE h;
|
||||
|
||||
@ -264,9 +268,13 @@ static int srvspawn(struct srvstart *s, SOCKET client_sock){
|
||||
need = threadstacksize(srv->stacksize);
|
||||
#endif
|
||||
|
||||
/* on a loop: what is posted to it made now, so that posting it, done
|
||||
once the client is counted, cannot fail */
|
||||
if(loopwanted(srv)) job = loopjobnew(need);
|
||||
|
||||
/* an idle worker if one has the stack this service needs */
|
||||
_3proxy_mutex_lock(&tpool.mutex);
|
||||
for(pp = &tpool.idle; *pp; pp = &(*pp)->poolnext){
|
||||
for(pp = job? NULL : &tpool.idle; pp && *pp; pp = &(*pp)->poolnext){
|
||||
if((*pp)->stacksize >= need){
|
||||
newparam = *pp;
|
||||
*pp = newparam->poolnext;
|
||||
@ -278,8 +286,9 @@ static int srvspawn(struct srvstart *s, SOCKET client_sock){
|
||||
}
|
||||
_3proxy_mutex_unlock(&tpool.mutex);
|
||||
reused = newparam != NULL;
|
||||
if(!reused && (!(newparam = malloc (sizeof(*defparam))) || _3proxy_sem_init_timed(newparam->wake, 0, 1))){
|
||||
if(!reused && (!(newparam = malloc (sizeof(*defparam))) || (!job && _3proxy_sem_init_timed(newparam->wake, 0, 1)))){
|
||||
if(newparam) free(newparam);
|
||||
if(job) free(job);
|
||||
srv->so._closesocket(srv->so.state, client_sock);
|
||||
defparam->res = 21;
|
||||
if(!srv->silent)dolog(defparam, (unsigned char *)"Memory Allocation Failed");
|
||||
@ -306,8 +315,9 @@ static int srvspawn(struct srvstart *s, SOCKET client_sock){
|
||||
newparam->next = srv->child;
|
||||
srv->child = srv->child->prev = newparam;
|
||||
}
|
||||
/* a worker from the pool is woken below, once it is one of the children */
|
||||
if(!reused){
|
||||
/* a worker from the pool is woken below, once it is one of the children,
|
||||
a loop given its client once counted */
|
||||
if(!reused && !job){
|
||||
#ifdef _WIN32
|
||||
#ifndef _WINCE
|
||||
h = (HANDLE)_beginthreadex((LPSECURITY_ATTRIBUTES )NULL, (unsigned)(16384 + srv->stacksize), (void *)threadfunc, (void *) newparam, 0, &thread);
|
||||
@ -346,13 +356,21 @@ static int srvspawn(struct srvstart *s, SOCKET client_sock){
|
||||
#ifndef STDMAIN
|
||||
listenercount(srv, 1);
|
||||
#endif
|
||||
if(!reused){
|
||||
newparam->workerid = (uint64_t)thread;
|
||||
_3proxy_mutex_lock(&tpool.mutex);
|
||||
tpool.created++;
|
||||
_3proxy_mutex_unlock(&tpool.mutex);
|
||||
if(job){
|
||||
/* the loop's thread is its thread: the loop sets it. Posted last,
|
||||
the client may run at once */
|
||||
newparam->workerid = newparam->threadid = 0;
|
||||
loopspawn(job, newparam);
|
||||
}
|
||||
else {
|
||||
if(!reused){
|
||||
newparam->workerid = (uint64_t)thread;
|
||||
_3proxy_mutex_lock(&tpool.mutex);
|
||||
tpool.created++;
|
||||
_3proxy_mutex_unlock(&tpool.mutex);
|
||||
}
|
||||
newparam->threadid = newparam->workerid;
|
||||
}
|
||||
newparam->threadid = newparam->workerid;
|
||||
}
|
||||
_3proxy_mutex_unlock(&srv->counter_mutex);
|
||||
if(reused) _3proxy_sem_unlock(newparam->wake);
|
||||
|
||||
@ -230,6 +230,7 @@ static int srvprepare(struct srvstart *s, int argc, char **argv){
|
||||
#endif
|
||||
" -b(BUFSIZE) size of network buffer, at most and by default 16384\n"
|
||||
" -S(STACKSIZE) value to add to default client thread stack size\n"
|
||||
" -L serve clients on event loops, a thread serving many; -L0 a thread each\n"
|
||||
" -t be silent (do not log service start/stop)\n"
|
||||
"\n"
|
||||
" -iIP ip address or internal interface (clients are expected to connect)\n"
|
||||
@ -477,6 +478,9 @@ static int srvprepare(struct srvstart *s, int argc, char **argv){
|
||||
case 'S':
|
||||
srv->stacksize = atoi(argv[i]+2);
|
||||
break;
|
||||
case 'L':
|
||||
srv->loopmode = argv[i][2] != '0';
|
||||
break;
|
||||
case 'a':
|
||||
srv->anonymous = 1 + atoi(argv[i]+2);
|
||||
break;
|
||||
@ -738,6 +742,7 @@ void srvinit(struct srvparam * srv, struct clientparam *param){
|
||||
srv->maxchild = conf->maxchild;
|
||||
srv->backlog = conf->backlog;
|
||||
srv->stacksize = conf->stacksize;
|
||||
srv->loopmode = conf->loopmode;
|
||||
srv->config = conf;
|
||||
srv->resolvmode = conf->resolvmode;
|
||||
memcpy(srv->timeouts, conf->timeouts, sizeof(srv->timeouts));
|
||||
|
||||
369
src/mainloop.c
Normal file
369
src/mainloop.c
Normal file
@ -0,0 +1,369 @@
|
||||
/*
|
||||
3APA3A simplest proxy server
|
||||
(c) 2002-2026 by Vladimir Dubrovin <vlad@3proxy.org>
|
||||
|
||||
please read License Agreement
|
||||
|
||||
*/
|
||||
|
||||
/* Clients served on event loops: a few threads, a CPU each by default
|
||||
(loops), each running a loop, and each client of a service given -L (or
|
||||
loopmode on) served by a fiber of one of them, with the same code as a
|
||||
client with a thread of its own: its waits in sockpoll() wait in the loop.
|
||||
A service whose clients cannot share a thread (threadperclient), one
|
||||
listening for connections back (-R), all of its clients waiting for the
|
||||
same socket, which a loop waits for once, or a build or processor without
|
||||
fibers gives each client a thread as before.
|
||||
Part of proxymain.c, which includes it; only for 3proxy itself. */
|
||||
|
||||
#if defined MODULEMAINFUNC && !defined STDMAIN
|
||||
#include "loop.h"
|
||||
#include "fiber.h"
|
||||
#endif
|
||||
|
||||
#if defined MODULEMAINFUNC && !defined STDMAIN && defined WITH_FIBERS
|
||||
|
||||
#ifdef _MSC_VER
|
||||
#define LTLS __declspec(thread)
|
||||
#else
|
||||
#define LTLS __thread
|
||||
#endif
|
||||
|
||||
/* fibers kept idle by a loop for its next clients */
|
||||
#define LOOPFIBERS 256
|
||||
/* at most so many sockets waited for at once without allocating */
|
||||
#define FIBERWAITS 8
|
||||
|
||||
struct fiberslot {
|
||||
struct fiber f;
|
||||
size_t stacksize;
|
||||
struct fiberslot *next;
|
||||
};
|
||||
|
||||
struct looper {
|
||||
struct loop l;
|
||||
uint64_t threadid;
|
||||
struct fiberslot *idle; /* the loop's thread only */
|
||||
unsigned nidle;
|
||||
};
|
||||
|
||||
/* what srvspawn() posts to a loop */
|
||||
struct loopjob {
|
||||
struct loopmsg m;
|
||||
struct clientparam *param;
|
||||
size_t stacksize;
|
||||
};
|
||||
|
||||
static struct looper *loopers;
|
||||
static unsigned nloopers, nextlooper;
|
||||
static int loopsfailed;
|
||||
static _3proxy_mutex_t loopers_mutex;
|
||||
static int loopers_mutex_init;
|
||||
|
||||
/* the loop of this thread, NULL but in a loop's thread */
|
||||
static LTLS struct looper *curlooper;
|
||||
|
||||
static size_t fiberstack(size_t need){
|
||||
#ifdef _WIN32
|
||||
/* only reserved: Windows commits as it is used, and takes some for
|
||||
itself */
|
||||
if(need < 262144) need = 262144;
|
||||
#endif
|
||||
return need;
|
||||
}
|
||||
|
||||
static struct fiberslot *fibertake(struct looper *lp, size_t stacksize){
|
||||
struct fiberslot **pp, *fs;
|
||||
|
||||
for(pp = &lp->idle; (fs = *pp); pp = &fs->next){
|
||||
if(fs->stacksize >= stacksize){
|
||||
*pp = fs->next;
|
||||
lp->nidle--;
|
||||
return fs;
|
||||
}
|
||||
}
|
||||
if(!(fs = malloc(sizeof(*fs)))) return NULL;
|
||||
if(fiberinit(&fs->f, stacksize)){
|
||||
free(fs);
|
||||
return NULL;
|
||||
}
|
||||
fs->stacksize = stacksize;
|
||||
return fs;
|
||||
}
|
||||
|
||||
static void fibergive(struct looper *lp, struct fiberslot *fs){
|
||||
if(lp->nidle >= LOOPFIBERS || timetoexit){
|
||||
fiberfree(&fs->f);
|
||||
free(fs);
|
||||
return;
|
||||
}
|
||||
fs->next = lp->idle;
|
||||
lp->idle = fs;
|
||||
lp->nidle++;
|
||||
}
|
||||
|
||||
/* Runs a fiber of this loop until it waits, or gives it back once its
|
||||
client is done. */
|
||||
static void fiberrun(struct looper *lp, struct fiber *f){
|
||||
if(!fiberresume(f)) fibergive(lp, (struct fiberslot *)f);
|
||||
}
|
||||
|
||||
/* A client, on a fiber: as its thread would serve it. */
|
||||
static void fiberclient(void *arg){
|
||||
struct clientparam *param = (struct clientparam *)arg;
|
||||
|
||||
serveparam(param);
|
||||
freeparam(param);
|
||||
}
|
||||
|
||||
/* posted by srvspawn(): the client gets a fiber of this loop */
|
||||
static void loopclient(struct loop *l, struct loopmsg *m){
|
||||
struct loopjob *job = (struct loopjob *)m;
|
||||
struct looper *lp = curlooper;
|
||||
struct clientparam *param = job->param;
|
||||
struct fiberslot *fs;
|
||||
|
||||
fs = fibertake(lp, job->stacksize);
|
||||
free(job);
|
||||
param->threadid = lp->threadid;
|
||||
if(!fs){
|
||||
param->res = 21;
|
||||
dolog(param, (unsigned char *)"Memory Allocation Failed");
|
||||
freeparam(param);
|
||||
return;
|
||||
}
|
||||
fiberstart(&fs->f, fiberclient, param);
|
||||
fiberrun(lp, &fs->f);
|
||||
}
|
||||
|
||||
#ifdef _WIN32
|
||||
static DWORD WINAPI looperthread(LPVOID arg){
|
||||
#else
|
||||
static void *looperthread(void *arg){
|
||||
sigset_t mask;
|
||||
|
||||
sigfillset(&mask);
|
||||
pthread_sigmask(SIG_SETMASK, &mask, NULL);
|
||||
#endif
|
||||
curlooper = (struct looper *)arg;
|
||||
looprun(&curlooper->l);
|
||||
return 0;
|
||||
}
|
||||
|
||||
|
||||
/* ---- a fiber waiting in its loop ---------------------------------------- */
|
||||
|
||||
struct fiberwait {
|
||||
struct looptimer t;
|
||||
struct looper *lp;
|
||||
struct fiber *f;
|
||||
struct loopio *io;
|
||||
short *rev;
|
||||
unsigned n;
|
||||
int woken;
|
||||
};
|
||||
|
||||
static void fiberwaitio(struct loop *l, struct loopio *io, short revents){
|
||||
struct fiberwait *w = (struct fiberwait *)io->data;
|
||||
|
||||
w->rev[io - w->io] |= revents;
|
||||
/* the fiber deletes what it waited for before it goes on: nothing of
|
||||
this wait is called again, and w, on its stack, is not touched
|
||||
after it runs */
|
||||
if(!w->woken){
|
||||
w->woken = 1;
|
||||
fiberrun(w->lp, w->f);
|
||||
}
|
||||
}
|
||||
|
||||
static void fiberwaittimer(struct loop *l, struct looptimer *t){
|
||||
struct fiberwait *w = (struct fiberwait *)t->data;
|
||||
|
||||
if(!w->woken){
|
||||
w->woken = 1;
|
||||
fiberrun(w->lp, w->f);
|
||||
}
|
||||
}
|
||||
|
||||
/* sockpoll() in a fiber of a loop: what poll() would return, waiting in the
|
||||
loop. -2 elsewhere, or for no wait at all, for sockpoll() to poll. */
|
||||
static int fiberpoll(struct pollfd *fds, unsigned nfds, int timeout){
|
||||
struct looper *lp = curlooper;
|
||||
struct fiberwait w;
|
||||
struct loopio ownio[FIBERWAITS];
|
||||
short ownrev[FIBERWAITS];
|
||||
unsigned i;
|
||||
int res = 0;
|
||||
|
||||
if(!lp || !fibercurrent() || !timeout) return -2;
|
||||
memset(&w, 0, sizeof(w));
|
||||
w.lp = lp;
|
||||
w.f = fibercurrent();
|
||||
w.n = nfds;
|
||||
w.io = ownio;
|
||||
w.rev = ownrev;
|
||||
if(nfds > FIBERWAITS){
|
||||
w.io = malloc(nfds * sizeof(*w.io));
|
||||
w.rev = malloc(nfds * sizeof(*w.rev));
|
||||
if(!w.io || !w.rev){
|
||||
if(w.io != ownio) free(w.io);
|
||||
if(w.rev != ownrev) free(w.rev);
|
||||
return -1;
|
||||
}
|
||||
}
|
||||
for(i = 0; i < nfds; i++){
|
||||
fds[i].revents = 0;
|
||||
w.rev[i] = 0;
|
||||
memset(w.io + i, 0, sizeof(w.io[i]));
|
||||
w.io[i].fd = fds[i].fd;
|
||||
w.io[i].events = fds[i].events & (POLLIN | POLLOUT);
|
||||
w.io[i].func = fiberwaitio;
|
||||
w.io[i].data = &w;
|
||||
if(loopadd(&lp->l, w.io + i)){
|
||||
while(i--) loopdel(&lp->l, w.io + i);
|
||||
res = -1;
|
||||
goto DONE;
|
||||
}
|
||||
}
|
||||
if(timeout > 0){
|
||||
w.t.func = fiberwaittimer;
|
||||
w.t.data = &w;
|
||||
looptimerset(&lp->l, &w.t, looptime() + (uint64_t)timeout);
|
||||
}
|
||||
fiberyield();
|
||||
if(timeout > 0) looptimerdel(&lp->l, &w.t);
|
||||
for(i = 0; i < nfds; i++){
|
||||
loopdel(&lp->l, w.io + i);
|
||||
fds[i].revents = w.rev[i];
|
||||
if(w.rev[i]) res++;
|
||||
}
|
||||
DONE:
|
||||
if(w.io != ownio) free(w.io);
|
||||
if(w.rev != ownrev) free(w.rev);
|
||||
return res;
|
||||
}
|
||||
|
||||
|
||||
/* ---- the loops, and clients given to them ------------------------------- */
|
||||
|
||||
static unsigned ncpus(void){
|
||||
#ifdef _WIN32
|
||||
SYSTEM_INFO si;
|
||||
|
||||
GetSystemInfo(&si);
|
||||
return si.dwNumberOfProcessors? (unsigned)si.dwNumberOfProcessors : 1;
|
||||
#else
|
||||
long n = sysconf(_SC_NPROCESSORS_ONLN);
|
||||
|
||||
return n > 0? (unsigned)n : 1;
|
||||
#endif
|
||||
}
|
||||
|
||||
/* The loops, started the first time a client is to be served on one, for
|
||||
the life of the process: as many as loops said then. 0 if they run. */
|
||||
static int loopsstart(void){
|
||||
unsigned i, n;
|
||||
|
||||
if(nloopers) return 0;
|
||||
if(loopsfailed) return 1;
|
||||
if(!loopers_mutex_init){
|
||||
/* the first services start from the configuration's thread */
|
||||
_3proxy_mutex_init(&loopers_mutex);
|
||||
loopers_mutex_init = 1;
|
||||
}
|
||||
_3proxy_mutex_lock(&loopers_mutex);
|
||||
if(nloopers || loopsfailed){
|
||||
_3proxy_mutex_unlock(&loopers_mutex);
|
||||
return loopsfailed;
|
||||
}
|
||||
n = runconf->loops? (unsigned)runconf->loops : ncpus();
|
||||
if(!fiberusable() || !(loopers = calloc(n, sizeof(*loopers)))){
|
||||
loopsfailed = 1;
|
||||
_3proxy_mutex_unlock(&loopers_mutex);
|
||||
return 1;
|
||||
}
|
||||
for(i = 0; i < n; i++){
|
||||
#ifdef _WIN32
|
||||
HANDLE h;
|
||||
DWORD tid;
|
||||
#else
|
||||
pthread_t t;
|
||||
pthread_attr_t pa;
|
||||
#endif
|
||||
|
||||
if(loopinit(&loopers[i].l)) break;
|
||||
#ifdef _WIN32
|
||||
if(!(h = CreateThread(NULL, 0, looperthread, loopers + i, 0, &tid))){
|
||||
loopfree(&loopers[i].l);
|
||||
break;
|
||||
}
|
||||
loopers[i].threadid = (uint64_t)tid;
|
||||
CloseHandle(h);
|
||||
#else
|
||||
pthread_attr_init(&pa);
|
||||
pthread_attr_setdetachstate(&pa, PTHREAD_CREATE_DETACHED);
|
||||
if(pthread_create(&t, &pa, looperthread, loopers + i)){
|
||||
pthread_attr_destroy(&pa);
|
||||
loopfree(&loopers[i].l);
|
||||
break;
|
||||
}
|
||||
pthread_attr_destroy(&pa);
|
||||
loopers[i].threadid = (uint64_t)(uintptr_t)t;
|
||||
#endif
|
||||
}
|
||||
if(!i) loopsfailed = 1;
|
||||
nloopers = i;
|
||||
if(i) fiberpollfunc = fiberpoll;
|
||||
_3proxy_mutex_unlock(&loopers_mutex);
|
||||
return !i;
|
||||
}
|
||||
|
||||
/* Whether this thread is a loop's. */
|
||||
static int onlooper(void){
|
||||
return curlooper != NULL;
|
||||
}
|
||||
|
||||
/* Whether the clients of srv are to be served on loops. */
|
||||
static int loopwanted(struct srvparam *srv){
|
||||
return srv->loopmode && !srv->threadperclient && srv->cbsock == INVALID_SOCKET && !loopsstart();
|
||||
}
|
||||
|
||||
/* What srvspawn() hands to a loop, made before: posting cannot fail. */
|
||||
static struct loopjob *loopjobnew(size_t stacksize){
|
||||
struct loopjob *job = malloc(sizeof(*job));
|
||||
|
||||
if(job){
|
||||
memset(job, 0, sizeof(*job));
|
||||
job->m.func = loopclient;
|
||||
job->stacksize = fiberstack(stacksize);
|
||||
}
|
||||
return job;
|
||||
}
|
||||
|
||||
/* The client goes to the next loop. */
|
||||
static void loopspawn(struct loopjob *job, struct clientparam *param){
|
||||
struct looper *lp;
|
||||
|
||||
_3proxy_mutex_lock(&loopers_mutex);
|
||||
lp = loopers + (nextlooper++ % nloopers);
|
||||
_3proxy_mutex_unlock(&loopers_mutex);
|
||||
job->param = param;
|
||||
looppost(&lp->l, &job->m);
|
||||
}
|
||||
|
||||
#else
|
||||
|
||||
struct loopjob;
|
||||
static int onlooper(void){
|
||||
return 0;
|
||||
}
|
||||
static int loopwanted(struct srvparam *srv){
|
||||
return 0;
|
||||
}
|
||||
static struct loopjob *loopjobnew(size_t stacksize){
|
||||
return NULL;
|
||||
}
|
||||
static void loopspawn(struct loopjob *job, struct clientparam *param){
|
||||
}
|
||||
|
||||
#endif
|
||||
@ -333,6 +333,7 @@ int nsbroken(void);
|
||||
lines among its arguments */
|
||||
SOCKET nssocket(struct srvparam *srv, struct sockfuncs *sf, void *state, int side, int domain, int type, int protocol);
|
||||
int sockpoll(struct sockfuncs *sf, void *state, struct pollfd *fds, unsigned nfds, int timeout);
|
||||
extern int (*fiberpollfunc)(struct pollfd *fds, unsigned nfds, int timeout);
|
||||
void nonblock(SOCKET sock);
|
||||
int sysresolves(int mode);
|
||||
int acceptagain(void);
|
||||
|
||||
@ -50,6 +50,7 @@ struct udpstate;
|
||||
mainclient.c threads for the clients of TCP services, and filters
|
||||
maintcp.c the main loop of a TCP service
|
||||
mainudp.c the main loop of a UDP service, in its one thread
|
||||
mainloop.c clients of TCP services served on event loops, by fibers
|
||||
Their functions of file scope are declared here, so that they can be in
|
||||
any order. */
|
||||
static void srvdiscard(struct srvstart *s);
|
||||
@ -81,6 +82,11 @@ static void freeworker(struct clientparam *param);
|
||||
static unsigned poollimit(void);
|
||||
static int poolpark(struct clientparam *param);
|
||||
static int srvspawn(struct srvstart *s, SOCKET client_sock);
|
||||
struct loopjob;
|
||||
static int loopwanted(struct srvparam *srv);
|
||||
static struct loopjob *loopjobnew(size_t stacksize);
|
||||
static void loopspawn(struct loopjob *job, struct clientparam *param);
|
||||
static int onlooper(void);
|
||||
static void filterswap(struct clientparam *param, unsigned char **cur, unsigned char *now);
|
||||
static int cblisten(struct srvstart *s);
|
||||
static int tcpnext(struct srvstart *s, SOCKET *client, int *nlog);
|
||||
@ -117,6 +123,7 @@ static int srvrun(struct srvstart *s);
|
||||
#include "mainconf.c"
|
||||
#include "mainlisten.c"
|
||||
#include "mainclient.c"
|
||||
#include "mainloop.c"
|
||||
#include "maintcp.c"
|
||||
#include "mainudp.c"
|
||||
|
||||
|
||||
@ -751,6 +751,7 @@ struct srvparam {
|
||||
/* a thread for each client: what serves them waits other than in
|
||||
sockpoll(), or must not be left by a thread to serve another */
|
||||
int threadperclient;
|
||||
int loopmode; /* its clients served on event loops */
|
||||
struct extparam *config; /* the configuration it was started with */
|
||||
int timeouts[12]; /* conf.timeouts as they were when the service started */
|
||||
int filtermaxsize, parentretries, maxseg; /* and these */
|
||||
@ -1000,6 +1001,8 @@ struct extparam {
|
||||
unsigned authcachetype, authcachetime;
|
||||
int filtermaxsize;
|
||||
int poolmax, pooltimeout; /* threadpool: idle workers kept, -1 automatic */
|
||||
int loops; /* threads serving clients on event loops, 0: a CPU each */
|
||||
int loopmode; /* services from here on serve clients on loops */
|
||||
int gracetraf, gracenum, gracedelay;
|
||||
int maxseg;
|
||||
unsigned char **archiver;
|
||||
|
||||
@ -179,7 +179,9 @@ class Tester:
|
||||
|
||||
def write_config(self, name, config):
|
||||
path = os.path.join(self.tmpdir, name + ".cfg")
|
||||
text = textwrap.dedent(config).strip() + "\n"
|
||||
# run.py --loopmode: every configuration starts with what it gives
|
||||
text = os.environ.get("THREEPROXY_TESTCONFIG", "") + \
|
||||
textwrap.dedent(config).strip() + "\n"
|
||||
# newline="" keeps the line endings as written, rather than letting
|
||||
# Windows turn them into CRLF behind the parser's back
|
||||
with open(path, "w", newline="") as fp:
|
||||
|
||||
@ -58,7 +58,11 @@ def main():
|
||||
help="keep the temporary directory")
|
||||
parser.add_argument("-v", "--verbose", action="store_true",
|
||||
help="print every check, not just the failures")
|
||||
parser.add_argument("--loopmode", action="store_true",
|
||||
help="serve the clients of every service on event loops")
|
||||
args = parser.parse_args()
|
||||
if args.loopmode:
|
||||
os.environ["THREEPROXY_TESTCONFIG"] = "loops 2\nloopmode on\n"
|
||||
|
||||
binary = args.binary or os.environ.get("BIN") or default_binary()
|
||||
binary = os.path.abspath(binary)
|
||||
|
||||
Loading…
Reference in New Issue
Block a user