bughz/3proxy
1
0
Fork 0
mirror of https://github.com/3proxy/3proxy.git synced 2026-11-05 09:25:49 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

Event loops: bodies of known length relayed by the callbacks too

Browse Source 
A body of known length (waitclient64, waitserver64), the HTTP proxy's
requests and answers, was left to sockmap(). The relay of the callbacks
moves it as sockmap() does: that way only, as much as it has left, the
other way not at all, and ends with 98 or 99 once it is all through.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Vladimir Dubrovin 2026-10-25 21:03:58 +03:00
parent 82387caf2f
commit 875e47a230
1 changed files with 47 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

64
src/mainloop.c
View File

@ -361,7 +361,7 @@ DONE:
/* ---- sockmap() as callbacks (-L2, a draft) ------------------------------- */ /* ---- sockmap() as callbacks (-L2, a draft) ------------------------------- */
/* The data of a connection mapped by a fiber of a loop with neither filters, /* The data of a connection mapped by a fiber of a loop with neither filters,
limits, nor anything but the socket functions of the system between the limits but that of a body's length, nor anything but the socket functions of the system between the
sides, is relayed by the loop itself: the sockets call it back, on the sides, is relayed by the loop itself: the sockets call it back, on the
loop's own stack, and the fiber, parked meanwhile, is resumed only once loop's own stack, and the fiber, parked meanwhile, is resumed only once
the mapping is over. No splice() yet: data is copied. */ the mapping is over. No splice() yet: data is copied. */
@ -379,6 +379,7 @@ struct cbmap {
uint64_t last; /* looptime() of the last data moved */ uint64_t last; /* looptime() of the last data moved */
uint64_t timeo, linger; /* ms */ uint64_t timeo, linger; /* ms */
int halfclose; int halfclose;
uint64_t fromcli, fromsrv; /* what is left to move each way */
int cterm_r, cterm_w, sterm_r, sterm_w, haserror; int cterm_r, cterm_w, sterm_r, sterm_w, haserror;
int res; int res;
}; };
@ -403,11 +404,14 @@ static int cbmapend(struct cbmap *m, int res){
} }
/* What the sockets are ready for, moved: 1 once the mapping is over, as /* What the sockets are ready for, moved: 1 once the mapping is over, as
sockmap() would end it. */ sockmap() would end it. As sockmap() too, a body of known length
(waitclient64, waitserver64) is all of what is moved: that way only, as
much as it has left (fromcli, fromsrv), the other way not at all. */
static int cbmaprun(struct cbmap *m){ static int cbmaprun(struct cbmap *m){
struct clientparam *param = m->param; struct clientparam *param = m->param;
struct sockfuncs *sf = &param->srv->so; struct sockfuncs *sf = &param->srv->so;
int round, moved = 0; int round, moved = 0;
uint64_t incli, insrv, room;
ssize_t n; ssize_t n;
/* a reload: sockmap() takes the connection back, which has it go by /* a reload: sockmap() takes the connection back, which has it go by
@ -417,12 +421,15 @@ static int cbmaprun(struct cbmap *m){
int progress = 0, end = !m->halfclose && (m->cterm_r || m->sterm_r); int progress = 0, end = !m->halfclose && (m->cterm_r || m->sterm_r);
/* the client to the server */ /* the client to the server */
if(!m->cterm_r && !end && CBROOM(param->clibufsize, param->cliinbuf) > 0 && incli = m->fromcli? param->cliinbuf - param->clioffset : 0;
room = CBROOM(param->clibufsize, param->cliinbuf);
if(m->fromcli > incli && !m->cterm_r && !end && room > 0 &&
(m->cr->ready & (POLLIN | POLLHUP | POLLERR))){ (m->cr->ready & (POLLIN | POLLHUP | POLLERR))){
n = sf->_recvfrom(param->sostate, param->clisock, (char *)param->clibuf + param->cliinbuf, n = sf->_recvfrom(param->sostate, param->clisock, (char *)param->clibuf + param->cliinbuf,
(int)CBROOM(param->clibufsize, param->cliinbuf), 0, NULL, NULL); (int)(room < m->fromcli - incli? room : m->fromcli - incli), 0, NULL, NULL);
if(n > 0){ if(n > 0){
param->cliinbuf += n; param->cliinbuf += n;
incli += n;
progress = 1; progress = 1;
} }
else if(!n || !CBAGAIN(errno)){ else if(!n || !CBAGAIN(errno)){
@ -431,13 +438,15 @@ static int cbmaprun(struct cbmap *m){
progress = 1; progress = 1;
} }
} }
if(param->cliinbuf > param->clioffset && !m->sterm_w && (m->sr->ready & (POLLOUT | POLLHUP | POLLERR))){ if(incli && m->fromcli && !m->sterm_w && (m->sr->ready & (POLLOUT | POLLHUP | POLLERR))){
n = sf->_sendto(param->sostate, param->remsock, (char *)param->clibuf + param->clioffset, n = sf->_sendto(param->sostate, param->remsock, (char *)param->clibuf + param->clioffset,
(int)(param->cliinbuf - param->clioffset), 0, NULL, 0); (int)(incli < m->fromcli? incli : m->fromcli), 0, NULL, 0);
if(n > 0){ if(n > 0){
param->nwrites++; param->nwrites++;
param->statscli64 += n; param->statscli64 += n;
param->clioffset += n; param->clioffset += n;
m->fromcli -= n;
incli -= n;
if(param->clioffset == param->cliinbuf) param->clioffset = param->cliinbuf = 0; if(param->clioffset == param->cliinbuf) param->clioffset = param->cliinbuf = 0;
progress = 1; progress = 1;
} }
@ -447,20 +456,23 @@ static int cbmaprun(struct cbmap *m){
progress = 1; progress = 1;
} }
} }
if(m->cterm_r && param->cliinbuf == param->clioffset && !m->sterm_w){ if(m->cterm_r && !incli && !m->sterm_w){
m->sterm_w = 1; m->sterm_w = 1;
sf->_shutdown(param->sostate, param->remsock, SHUT_WR); sf->_shutdown(param->sostate, param->remsock, SHUT_WR);
} }
/* the server to the client */ /* the server to the client */
if(!m->sterm_r && !end && CBROOM(param->srvbufsize, param->srvinbuf) > 0 && insrv = m->fromsrv? param->srvinbuf - param->srvoffset : 0;
room = CBROOM(param->srvbufsize, param->srvinbuf);
if(m->fromsrv > insrv && !m->sterm_r && !end && room > 0 &&
(m->sr->ready & (POLLIN | POLLHUP | POLLERR))){ (m->sr->ready & (POLLIN | POLLHUP | POLLERR))){
n = sf->_recvfrom(param->sostate, param->remsock, (char *)param->srvbuf + param->srvinbuf, n = sf->_recvfrom(param->sostate, param->remsock, (char *)param->srvbuf + param->srvinbuf,
(int)CBROOM(param->srvbufsize, param->srvinbuf), 0, NULL, NULL); (int)(room < m->fromsrv - insrv? room : m->fromsrv - insrv), 0, NULL, NULL);
if(n > 0){ if(n > 0){
param->nreads++; param->nreads++;
param->statssrv64 += n; param->statssrv64 += n;
param->srvinbuf += n; param->srvinbuf += n;
insrv += n;
progress = 1; progress = 1;
} }
else if(!n || !CBAGAIN(errno)){ else if(!n || !CBAGAIN(errno)){
@ -469,11 +481,13 @@ static int cbmaprun(struct cbmap *m){
progress = 1; progress = 1;
} }
} }
if(param->srvinbuf > param->srvoffset && !m->cterm_w && (m->cr->ready & (POLLOUT | POLLHUP | POLLERR))){ if(insrv && m->fromsrv && !m->cterm_w && (m->cr->ready & (POLLOUT | POLLHUP | POLLERR))){
n = sf->_sendto(param->sostate, param->clisock, (char *)param->srvbuf + param->srvoffset, n = sf->_sendto(param->sostate, param->clisock, (char *)param->srvbuf + param->srvoffset,
(int)(param->srvinbuf - param->srvoffset), 0, NULL, 0); (int)(insrv < m->fromsrv? insrv : m->fromsrv), 0, NULL, 0);
if(n > 0){ if(n > 0){
param->srvoffset += n; param->srvoffset += n;
m->fromsrv -= n;
insrv -= n;
if(param->srvoffset == param->srvinbuf) param->srvoffset = param->srvinbuf = 0; if(param->srvoffset == param->srvinbuf) param->srvoffset = param->srvinbuf = 0;
progress = 1; progress = 1;
} }
@ -483,17 +497,24 @@ static int cbmaprun(struct cbmap *m){
progress = 1; progress = 1;
} }
} }
if(m->sterm_r && param->srvinbuf == param->srvoffset && !m->cterm_w){ if(m->sterm_r && !insrv && !m->cterm_w){
m->cterm_w = 1; m->cterm_w = 1;
sf->_shutdown(param->sostate, param->clisock, SHUT_WR); sf->_shutdown(param->sostate, param->clisock, SHUT_WR);
} }
/* sockmap()'s condition to go on */ /* sockmap()'s condition to go on */
end = !m->halfclose && (m->cterm_r || m->sterm_r); end = !m->halfclose && (m->cterm_r || m->sterm_r);
if(!((!m->cterm_w && (param->srvinbuf > param->srvoffset || (!m->sterm_r && !end))) || incli = m->fromcli? param->cliinbuf - param->clioffset : 0;
(!m->sterm_w && (param->cliinbuf > param->clioffset || (!m->cterm_r && !end))))){ insrv = m->fromsrv? param->srvinbuf - param->srvoffset : 0;
return cbmapend(m, m->haserror? 94 + m->haserror : if(!((!m->cterm_w && m->fromsrv && (insrv || (!m->sterm_r && !end))) ||
(param->cliinbuf > param->clioffset || param->srvinbuf > param->srvoffset)? 94 : 0); (!m->sterm_w && m->fromcli && (incli || (!m->cterm_r && !end))))){
int res;
if(!m->fromsrv && param->waitserver64) res = 98;
else if(!m->fromcli && param->waitclient64) res = 99;
else if(m->haserror) res = 94 + m->haserror;
else res = (incli || insrv)? 94 : 0;
return cbmapend(m, res);
} }
if(!progress) break; if(!progress) break;
moved = 1; moved = 1;
@ -542,7 +563,7 @@ static int fibermap(struct clientparam *param, int timeo){
struct cbmap m; struct cbmap m;
if(!lp || !fibercurrent() || param->srv->loopmode != 2) return -2; if(!lp || !fibercurrent() || param->srv->loopmode != 2) return -2;
if(param->bandlimfunc || param->waitclient64 || param->waitserver64 || if(param->bandlimfunc ||
param->ctrlsock != INVALID_SOCKET || param->ctrlsocksrv != INVALID_SOCKET || param->ctrlsock != INVALID_SOCKET || param->ctrlsocksrv != INVALID_SOCKET ||
param->npredatfilters || param->ndatfilterscli || param->ndatfilterssrv || param->npredatfilters || param->ndatfilterscli || param->ndatfilterssrv ||
param->maxtrafin64 || param->maxtrafout64 || param->srv->gracedelay || param->maxtrafin64 || param->maxtrafout64 || param->srv->gracedelay ||
@ -560,6 +581,15 @@ static int fibermap(struct clientparam *param, int timeo){
m.halfclose = param->srv->halfclose; m.halfclose = param->srv->halfclose;
m.timeo = (uint64_t)timeo * 1000; m.timeo = (uint64_t)timeo * 1000;
m.linger = (uint64_t)param->srv->timeouts[LINGER_TO] * 1000; m.linger = (uint64_t)param->srv->timeouts[LINGER_TO] * 1000;
m.fromcli = m.fromsrv = 0x7fffffffffffffffULL;
if(param->waitclient64){
m.fromcli = param->waitclient64;
m.fromsrv = 0;
}
if(param->waitserver64){
m.fromsrv = param->waitserver64;
m.fromcli = 0;
}
m.last = looptime(); m.last = looptime();
m.idle.func = cbmapidle; m.idle.func = cbmapidle;
m.idle.data = &m; m.idle.data = &m;