3proxy/src/mainclient.c
Vladimir Dubrovin 7963494b1c Event loops: a relayed client's buffers given back while empty
A connection's two data buffers are malloc()'d with it (bufalloc())
instead of being part of struct clientparam, which is CLIENTBUFSIZE*2
smaller. The callbacks' relay of a client whose fiber gave up its stack
frees a side's buffer while it is empty and makes it again to read, and
gives both back before sockmap() or the child's tail goes on: a pool of
idle relayed connections holds no buffers. Fibers and threads keep
theirs as before.

mix scenarios, callbacks (-L -L2), peak memory before/after, throughput
the same:
  Intel: relay 972/632 MB, connect 981/581, socks 992/594
  ARM:   relay 1000/673 MB, connect 1000/615, socks 1020/629

struct clientparam's clibufdata and srvbufdata are pointers now:
plugins built against the old structure are not compatible.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-26 23:47:24 +03:00

664 lines
21 KiB
C

/*
3APA3A simplest proxy server
(c) 2002-2026 by Vladimir Dubrovin <vlad@3proxy.org>
please read License Agreement
*/
/* Clients of TCP services: a thread for each, from a pool, from the filters
to the end of the child serving it, and the filters' hooks.
Part of proxymain.c, which includes it. */
/* 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.
The child which completed the request returns NULL. param is logged by
the child and released here, but not freed: its thread may keep it.
*/
void * childfunc(struct clientparam * param){
PROXYFUNC pf = param->srv->pf;
int i;
for(i = 0; pf && i < MAXCHILDREDIRECTS; i++) pf = (PROXYFUNC)(*pf)(param);
if(pf){
param->res = 101;
dolog(param, (unsigned char *)"Redirection loop");
}
releaseparam(param);
return NULL;
}
/* One connection, from the filters to the end of the child which answers
it. The structure is released, not freed, when it returns. */
static void serveparam(struct clientparam *param){
int i = -1;
#ifdef MODULEMAINFUNC
if(makefilters(param->srv, param) > CONTINUE){
releaseparam(param);
return;
}
#endif
if(param->srv->cbsock != INVALID_SOCKET){
SASIZETYPE size = sizeof(param->sinsr);
struct pollfd fds;
/* closed connections passed over free: the listen queue's length */
int skips = 1 + (param->srv->maxchild>>4);
fds.fd = param->srv->cbsock;
fds.events = POLLIN;
fds.revents = 0;
for(i=5+(param->srv->maxchild>>10); i; i--){
if(sockpoll(&param->srv->so, param->sostate, &fds, 1, 1000*CONNBACK_TO)!=1){
dolog(param, (unsigned char *)"Connect back not received, check connback client");
i = 0;
break;
}
param->remsock = param->srv->so._accept(param->sostate, param->srv->cbsock, (struct sockaddr*)&param->sinsr, &size);
if(param->remsock == INVALID_SOCKET) {
/* another client of the service took it: not a try */
if(acceptagain()) { i++; continue; }
dolog(param, (unsigned char *)"Connect back accept() failed");
continue;
}
{
#ifdef _WIN32
unsigned long ul=1;
ioctlsocket(param->remsock, FIONBIO, &ul);
#else
fcntl(param->remsock,F_SETFL,O_NONBLOCK | fcntl(param->remsock,F_GETFL));
#endif
}
#ifndef WITHMAIN
param->req = param->sinsr;
if(param->srv->acl) param->res = checkACL(param);
if(param->res){
dolog(param, (unsigned char *)"Connect back ACL failed");
param->srv->so._closesocket(param->sostate, param->remsock);
param->remsock = INVALID_SOCKET;
continue;
}
#endif
{
/* the side connecting back sends nothing before the go-ahead:
readable, the connection was closed there, by a service
which stopped waiting (a reload, its time out); the next
one is taken, and this one does not count as a try, for as
many as the queue holds: past that it does */
struct pollfd rf;
rf.fd = param->remsock;
rf.events = POLLIN;
rf.revents = 0;
if(sockpoll(&param->srv->so, param->sostate, &rf, 1, 0) == 1){
param->srv->so._closesocket(param->sostate, param->remsock);
param->remsock = INVALID_SOCKET;
if(skips > 0){ skips--; i++; }
continue;
}
}
if(socksend(param, param->remsock, (unsigned char *)"C", 1, CONNBACK_TO) != 1){
dolog(param, (unsigned char *)"Connect back sending command failed");
param->srv->so._closesocket(param->sostate, param->remsock);
param->remsock = INVALID_SOCKET;
continue;
}
break;
}
}
if(!i){
param->res = 13;
releaseparam(param);
}
else {
#ifndef WITHMAIN
#ifndef _WIN32
/* 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
if(param->srv->haproxy){
char buf[128];
int i;
i = sockgetlinebuf(param, CLIENT, (unsigned char *)buf, sizeof(buf)-1, '\n', param->srv->timeouts[STRING_S]);
if(i > 12 && !strncasecmp(buf, "PROXY TCP", 9)){
char *token, *token2=NULL;
unsigned short u1=0, u2=0;
buf[i] = 0;
token = strchr(buf, ' ');
if(token) token = strchr(token+1, ' ');
if(token) token++;
if(token) token2 = strchr(token+1, ' ');
if(token2) {
*token2 = 0;
getip46(46, (unsigned char*) token, (struct sockaddr *)&param->sincr);
token = token2+1;
token2 = strchr(token, ' ');
}
if(token2) {
*token2 = 0;
getip46(46, (unsigned char *) token, (struct sockaddr *)&param->sincl);
token = token2+1;
token2 = strchr(token, ' ');
}
if(token){
sscanf(token,"%hu%hu", &u1, &u2);
if(u1) *SAPORT(&param->sincr) = htons(u1);
if(u2) *SAPORT(&param->sincl) = htons(u2);
}
}
}
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);
buffree(param);
free(param);
}
/* The automatic cap: POOLPERSERVICE idle workers for each running service,
but never more than all of them could have connections. */
static unsigned poollimit(void){
unsigned limit;
if(runconf->poolmax >= 0) limit = (unsigned)runconf->poolmax;
else {
limit = POOLPERSERVICE * 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(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 = runconf->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
return 0;
#else
return NULL;
#endif
}
/* 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;
struct clientparam * newparam = NULL, **pp;
pthread_t thread;
unsigned char buf[256];
int error = 0, reused;
size_t need;
struct loopjob *job = NULL;
#ifdef _WIN32
HANDLE h;
need = 16384 + srv->stacksize;
#else
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 = job? NULL : &tpool.idle; pp && *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)))){
newparam->clibufdata = newparam->srvbufdata = NULL;
newparam->clibuf = newparam->srvbuf = NULL;
if(bufalloc(newparam) || (!job && _3proxy_sem_init_timed(newparam->wake, 0, 1))){
buffree(newparam);
free(newparam);
newparam = NULL;
}
}
if(!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");
usleep(SLEEPTIME);
return 1;
};
if(!reused){
newparam->stacksize = need;
newparam->pooled = newparam->poolexit = 0;
newparam->poolnext = NULL;
}
memcpy(newparam, defparam, offsetof(struct clientparam, filters));
newparam->clibuf = newparam->clibufdata;
newparam->srvbuf = newparam->srvbufdata;
newparam->clibufsize = newparam->srvbufsize = SRVBUFSIZE(srv);
clearstat(newparam);
if(!s->isudp) newparam->clisock = client_sock;
newparam->prev = newparam->next = NULL;
_3proxy_mutex_lock(&srv->counter_mutex);
if(!srv->child){
srv->child = newparam;
}
else {
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,
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);
#else
h = (HANDLE)CreateThread((LPSECURITY_ATTRIBUTES )NULL, (unsigned)(16384 + srv->stacksize), (void *)threadfunc, (void *) newparam, 0, &thread);
#endif
if (h) {
CloseHandle(h);
}
else {
sprintf((char *)buf, "_beginthreadex(): %s", _strerror(NULL));
error = 1;
}
#else
if ((error = pthread_create(&thread, &s->pa, threadfunc, (void *)newparam))){
sprintf((char *)buf, "pthread_create(): %s", strerror(error));
}
#endif
}
if(error){
if(!srv->silent)dolog(defparam, buf);
if(newparam->prev) newparam->prev->next = newparam->next;
else srv->child = newparam->next;
if(newparam->next) newparam->next->prev = newparam->prev;
if(newparam->clisock != INVALID_SOCKET){
srv->so._shutdown(srv->so.state, newparam->clisock, SHUT_RDWR);
srv->so._closesocket(srv->so.state, newparam->clisock);
newparam->clisock = INVALID_SOCKET;
}
newparam->srv = NULL;
freeworker(newparam);
}
else {
srv->childcount++;
#ifndef STDMAIN
listenercount(srv, 1);
#endif
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;
}
}
_3proxy_mutex_unlock(&srv->counter_mutex);
if(reused) _3proxy_sem_unlock(newparam->wake);
memset(&defparam->sincl, 0, sizeof(defparam->sincl));
memset(&defparam->sincr, 0, sizeof(defparam->sincr));
return error;
}
/* 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) {
param->maptail = NULL;
if(param->maptaildata){
free(param->maptaildata);
param->maptaildata = NULL;
}
if(param->srv){
if(param->srv->so.freefunc) param->srv->so.freefunc(param->sostate);
}
if(param->clibuf != param->clibufdata) free(param->clibuf);
if(param->srvbuf != param->srvbufdata) free(param->srvbuf);
param->clibuf = param->clibufdata;
param->srvbuf = param->srvbufdata;
#ifdef WITHSPLICE
if(param->pipecli[0] >= 0) fdclose(param->pipecli[0]);
if(param->pipecli[1] >= 0) fdclose(param->pipecli[1]);
if(param->pipesrv[0] >= 0) fdclose(param->pipesrv[0]);
if(param->pipesrv[1] >= 0) fdclose(param->pipesrv[1]);
param->pipecli[0] = param->pipecli[1] = param->pipesrv[0] = param->pipesrv[1] = -1;
#endif
/* Not shut down on Windows: there shutting down receiving resets the
connection as soon as the peer sends anything more, and close()
alone sends whatever is queued before its FIN. Elsewhere it stays:
on the BSDs it discards what is left unread, so the close that
follows is not turned into a reset. */
if(param->srv) {
if(param->ctrlsocksrv != INVALID_SOCKET && param->ctrlsocksrv != param->remsock) {
#ifndef _WIN32
param->srv->so._shutdown(param->sostate, param->ctrlsocksrv, SHUT_RDWR);
#endif
param->srv->so._closesocket(param->sostate, param->ctrlsocksrv);
}
if(param->ctrlsock != INVALID_SOCKET && param->ctrlsock != param->clisock) {
#ifndef _WIN32
param->srv->so._shutdown(param->sostate, param->ctrlsock, SHUT_RDWR);
#endif
param->srv->so._closesocket(param->sostate, param->ctrlsock);
}
if(param->remsock != INVALID_SOCKET) {
#ifndef _WIN32
param->srv->so._shutdown(param->sostate, param->remsock, SHUT_RDWR);
#endif
param->srv->so._closesocket(param->sostate, param->remsock);
}
if(param->clisock != INVALID_SOCKET) {
#ifndef _WIN32
param->srv->so._shutdown(param->sostate, param->clisock, SHUT_RDWR);
#endif
param->srv->so._closesocket(param->sostate, param->clisock);
}
param->ctrlsocksrv = param->ctrlsock = param->remsock = param->clisock = INVALID_SOCKET;
}
#ifndef STDMAIN
while(param->nfilters > 0){
param->nfilters--;
if(param->filters[param->nfilters].filter->filter_clear)
(*param->filters[param->nfilters].filter->filter_clear)(param->filters[param->nfilters].data);
}
if(param->connlim) stopconnlims(param);
/* a slot held in the newest service on the listener after a reload */
if(param->newsrv){
slotrelease(param->newsrv);
param->newsrv = NULL;
}
#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()
frees the service along with its socket functions, filters and mutex. */
if(param->srv){
_3proxy_mutex_lock(&param->srv->counter_mutex);
if(param->prev || param->next || param->srv->child == param){
if(param->prev){
param->prev->next = param->next;
}
else
param->srv->child = param->next;
if(param->next){
param->next->prev = param->prev;
}
(param->srv->childcount)--;
#ifndef STDMAIN
listenercount(param->srv, -1);
#endif
/* under the mutex too: srvfree() may destroy the semaphore as
soon as the mutex shows the last child gone */
_3proxy_sem_unlock(param->srv->childslots);
}
_3proxy_mutex_unlock(&param->srv->counter_mutex);
param->prev = param->next = NULL;
param->srv = NULL;
}
}
void freeparam(struct clientparam * param) {
releaseparam(param);
buffree(param);
free(param);
}
FILTER_ACTION handleconnectflt(struct clientparam *cparam){
#ifndef STDMAIN
FILTER_ACTION action;
int i;
for(i=0; i<cparam->nconnectfilters ;i++){
action = (*cparam->connectfilters[i]->filter->filter_connect)(cparam->connectfilters[i]->data, cparam);
if(action!=CONTINUE) return action;
}
#endif
return PASS;
}
/* A buffer filter given a buffer too small for what it makes of it hands
back one of its own, allocated with malloc, and leaves the one it was given
alone: nothing refers to that one any more, so it is freed here, unless it
is, or is in, one of the connection's own buffers: headers are worked on
in clibuf or srvbuf, which may have been grown to a bigger one of
malloc's. The caller owns whatever *buf_p points to on return. */
static int connbuf(struct clientparam *param, unsigned char *p){
return (p >= param->clibuf && p < param->clibuf + param->clibufsize) ||
(p >= param->srvbuf && p < param->srvbuf + param->srvbufsize) ||
p == param->clibufdata || p == param->srvbufdata || p == param->tmpbuf;
}
static void filterswap(struct clientparam *param, unsigned char **cur, unsigned char *now){
if(now != *cur){
if(!connbuf(param, *cur)) free(*cur);
*cur = now;
}
}
#ifndef STDMAIN
FILTER_ACTION makefilters (struct srvparam *srv, struct clientparam *param){
FILTER_ACTION res=PASS;
FILTER_ACTION action;
int i;
if(!srv->nfilters) return PASS;
for(i=0; i<srv->nfilters; i++){
if(!srv->filter[i].filter_client)continue;
/* the slot may hold what the connection before left in it */
param->filters[param->nfilters].data = NULL;
action = (*srv->filter[i].filter_client)(srv->filter[i].data, param, &param->filters[param->nfilters].data);
if(action == PASS) continue;
if(action > CONTINUE) {
/* a filter refusing the connection is not kept, but it may
have made a state for it */
if(srv->filter[i].filter_clear && param->filters[param->nfilters].data)
(*srv->filter[i].filter_clear)(param->filters[param->nfilters].data);
return action;
}
param->filters[param->nfilters].filter = srv->filter + i;
if(srv->filter[i].filter_request)param->reqfilters[param->nreqfilters++] = param->filters + param->nfilters;
if(srv->filter[i].filter_connect)param->connectfilters[param->nconnectfilters++] = param->filters + param->nfilters;
if(srv->filter[i].filter_afterauth)param->afterauthfilters[param->nafterauthfilters++] = param->filters + param->nfilters;
if(srv->filter[i].filter_header_cli)param->hdrfilterscli[param->nhdrfilterscli++] = param->filters + param->nfilters;
if(srv->filter[i].filter_header_srv)param->hdrfilterssrv[param->nhdrfilterssrv++] = param->filters + param->nfilters;
if(srv->filter[i].filter_predata)param->predatfilters[param->npredatfilters++] = param->filters + param->nfilters;
if(srv->filter[i].filter_data_cli)param->datfilterscli[param->ndatfilterscli++] = param->filters + param->nfilters;
if(srv->filter[i].filter_data_srv)param->datfilterssrv[param->ndatfilterssrv++] = param->filters + param->nfilters;
param->nfilters++;
}
return res;
}
#endif
#ifndef STDMAIN
FILTER_ACTION handleafterauthflt(struct clientparam *cparam){
#ifndef STDMAIN
FILTER_ACTION action;
int i;
for(i=0; i<cparam->nafterauthfilters ;i++){
action = (*cparam->afterauthfilters[i]->filter->filter_afterauth)(cparam->afterauthfilters[i]->data, cparam);
if(action!=CONTINUE) return action;
}
#endif
return PASS;
}
FILTER_ACTION handlereqfilters(struct clientparam *param, unsigned char ** buf_p, int * bufsize_p, int offset, int * length_p){
FILTER_ACTION action;
int i;
unsigned char *cur = *buf_p;
for(i=0; i<param->nreqfilters; i++){
action = (*param->reqfilters[i]->filter->filter_request)(param->reqfilters[i]->data, param, buf_p, bufsize_p, offset, length_p);
filterswap(param, &cur, *buf_p);
if(action!=CONTINUE) return action;
}
return PASS;
}
FILTER_ACTION handlehdrfilterssrv(struct clientparam *param, unsigned char ** buf_p, int * bufsize_p, int offset, int * length_p){
FILTER_ACTION action;
int i;
unsigned char *cur = *buf_p;
for(i=0; i<param->nhdrfilterssrv; i++){
action = (*param->hdrfilterssrv[i]->filter->filter_header_srv)(param->hdrfilterssrv[i]->data, param, buf_p, bufsize_p, offset, length_p);
filterswap(param, &cur, *buf_p);
if(action!=CONTINUE) return action;
}
return PASS;
}
FILTER_ACTION handlehdrfilterscli(struct clientparam *param, unsigned char ** buf_p, int * bufsize_p, int offset, int * length_p){
FILTER_ACTION action;
int i;
unsigned char *cur = *buf_p;
for(i = 0; i < param->nhdrfilterscli; i++){
action = (*param->hdrfilterscli[i]->filter->filter_header_cli)(param->hdrfilterscli[i]->data, param, buf_p, bufsize_p, offset, length_p);
filterswap(param, &cur, *buf_p);
if(action!=CONTINUE) return action;
}
return PASS;
}
#endif
FILTER_ACTION handlepredatflt(struct clientparam *cparam){
#ifndef STDMAIN
FILTER_ACTION action;
int i;
if(cparam->predatdone) return PASS;
cparam->predatdone = 1;
for(i=0; i<cparam->npredatfilters ;i++){
action = (*cparam->predatfilters[i]->filter->filter_predata)(cparam->predatfilters[i]->data, cparam);
if(action!=CONTINUE) return action;
}
#endif
return PASS;
}
FILTER_ACTION handledatfltcli(struct clientparam *cparam, unsigned char ** buf_p, int * bufsize_p, int offset, int * length_p){
#ifndef STDMAIN
FILTER_ACTION action;
int i;
unsigned char *cur = *buf_p;
for(i=0; i<cparam->ndatfilterscli ;i++){
action = (*cparam->datfilterscli[i]->filter->filter_data_cli)(cparam->datfilterscli[i]->data, cparam, buf_p, bufsize_p, offset, length_p);
filterswap(cparam, &cur, *buf_p);
if(action!=CONTINUE) return action;
}
#endif
return PASS;
}
FILTER_ACTION handledatfltsrv(struct clientparam *cparam, unsigned char ** buf_p, int * bufsize_p, int offset, int * length_p){
FILTER_ACTION action;
int i;
unsigned char *cur = *buf_p;
for(i=0; i<cparam->ndatfilterssrv; i++){
action = (*cparam->datfilterssrv[i]->filter->filter_data_srv)(cparam->datfilterssrv[i]->data, cparam, buf_p, bufsize_p, offset, length_p);
filterswap(cparam, &cur, *buf_p);
if(action!=CONTINUE) return action;
}
return PASS;
}