bughz/3proxy
1
0
Fork 0
mirror of https://github.com/3proxy/3proxy.git synced 2025-02-23 18:45:40 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

Add splice() support on Linux to proxy, socks, ftppt, smtpp, pop3p

Browse Source 
usage example:
proxy -s
proxy -s2
-s - enable splice() for sockets mapping to avoid data to be copied to user space
-s2 - enable splice() and attempt to free user space buffer ASAP
This commit is contained in:
z3apa3a 2016-12-19 22:07:34 +03:00
parent 34876b7616
commit d159c57e1f
6 changed files with 98 additions and 23 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
src/ftppr.c
View File

@ -242,7 +242,7 @@ void * ftpprchild(struct clientparam* param) {
so._setsockopt(param->remsock, SOL_SOCKET, SO_LINGER, (char *)&lg, sizeof(lg)); so._setsockopt(param->remsock, SOL_SOCKET, SO_LINGER, (char *)&lg, sizeof(lg));
so._setsockopt(clidatasock, SOL_SOCKET, SO_LINGER, (char *)&lg, sizeof(lg)); so._setsockopt(clidatasock, SOL_SOCKET, SO_LINGER, (char *)&lg, sizeof(lg));
param->clisock = clidatasock; param->clisock = clidatasock;
res = sockmap(param, conf.timeouts[CONNECTION_S]); res = mapsocket(param, conf.timeouts[CONNECTION_S]);
if(param->remsock != INVALID_SOCKET) { if(param->remsock != INVALID_SOCKET) {
so._shutdown (param->remsock, SHUT_RDWR); so._shutdown (param->remsock, SHUT_RDWR);
so._closesocket(param->remsock); so._closesocket(param->remsock);

2
src/pop3p.c
View File

@ -44,7 +44,7 @@ void * pop3pchild(struct clientparam* param) {
{RETURN(623);} {RETURN(623);}
param->statscli64 += (uint64_t)(strlen((char *)param->extusername) + 7); param->statscli64 += (uint64_t)(strlen((char *)param->extusername) + 7);
param->nwrites++; param->nwrites++;
RETURN (sockmap(param, 180)); RETURN (mapsocket(param, 180));
CLEANRET: CLEANRET:
if(param->hostname&&param->extusername) { if(param->hostname&&param->extusername) {

12
src/proxy.c
View File

@ -616,7 +616,7 @@ for(;;){
s = param->remsock; s = param->remsock;
param->remsock = ftps; param->remsock = ftps;
if((param->operation == FTP_PUT) && (contentlength64 > 0)) param->waitclient64 = contentlength64; if((param->operation == FTP_PUT) && (contentlength64 > 0)) param->waitclient64 = contentlength64;
res = sockmap(param, conf.timeouts[CONNECTION_L]); res = mapsocket(param, conf.timeouts[CONNECTION_L]);
if (res == 99) res = 0; if (res == 99) res = 0;
so._closesocket(ftps); so._closesocket(ftps);
ftps = INVALID_SOCKET; ftps = INVALID_SOCKET;
@ -803,7 +803,7 @@ for(;;){
if(isconnect && param->redirtype != R_HTTP) { if(isconnect && param->redirtype != R_HTTP) {
socksend(param->clisock, (unsigned char *)proxy_stringtable[8], (int)strlen(proxy_stringtable[8]), conf.timeouts[STRING_S]); socksend(param->clisock, (unsigned char *)proxy_stringtable[8], (int)strlen(proxy_stringtable[8]), conf.timeouts[STRING_S]);
if(param->redirectfunc) return (*param->redirectfunc)(param); if(param->redirectfunc) return (*param->redirectfunc)(param);
param->res = sockmap(param, conf.timeouts[CONNECTION_L]); param->res = mapsocket(param, conf.timeouts[CONNECTION_L]);
if(param->redirectfunc) return (*param->redirectfunc)(param); if(param->redirectfunc) return (*param->redirectfunc)(param);
RETURN(param->res); RETURN(param->res);
} }
@ -866,7 +866,7 @@ for(;;){
if(contentlength64 > 0){ if(contentlength64 > 0){
param->nolongdatfilter = 0; param->nolongdatfilter = 0;
param->waitclient64 = contentlength64; param->waitclient64 = contentlength64;
res = sockmap(param, conf.timeouts[CONNECTION_S]); res = mapsocket(param, conf.timeouts[CONNECTION_S]);
param->waitclient64 = 0; param->waitclient64 = 0;
if(res != 99) { if(res != 99) {
RETURN(res); RETURN(res);
@ -1034,7 +1034,7 @@ for(;;){
} }
if(param->chunked != 2){ if(param->chunked != 2){
param->waitserver64 = contentlength64; param->waitserver64 = contentlength64;
if((res = sockmap(param, conf.timeouts[CONNECTION_S])) != 98){ if((res = mapsocket(param, conf.timeouts[CONNECTION_S])) != 98){
RETURN(res); RETURN(res);
} }
param->waitserver64 = 0; param->waitserver64 = 0;
@ -1042,13 +1042,13 @@ for(;;){
} while(param->chunked); } while(param->chunked);
} }
if(isconnect && res == 200 && param->operation){ if(isconnect && res == 200 && param->operation){
RETURN (sockmap(param, conf.timeouts[CONNECTION_S])); RETURN (mapsocket(param, conf.timeouts[CONNECTION_S]));
} }
else if(isconnect){ else if(isconnect){
ckeepalive = keepalive = 1; ckeepalive = keepalive = 1;
} }
else if(!hascontent && !param->chunked) { else if(!hascontent && !param->chunked) {
RETURN(sockmap(param, conf.timeouts[CONNECTION_S])); RETURN(mapsocket(param, conf.timeouts[CONNECTION_S]));
} }
contentlength64 = 0; contentlength64 = 0;
REQUESTEND: REQUESTEND:

4
src/smtpp.c
View File

@ -272,7 +272,7 @@ void * smtppchild(struct clientparam* param) {
if(param->nhdrfilterscli || param->nhdrfilterssrv || param->ndatfilterscli || param->ndatfilterssrv){ if(param->nhdrfilterscli || param->nhdrfilterssrv || param->ndatfilterscli || param->ndatfilterssrv){
do { do {
if(res == 22) RETURN (sockmap(param, 180)); if(res == 22) RETURN (mapsocket(param, 180));
if(res != 2 && (res = readreply(param)) <= 0) break; if(res != 2 && (res = readreply(param)) <= 0) break;
if(res == 221) RETURN(0); if(res == 221) RETURN(0);
if(res == 354) res = readdata(param); if(res == 354) res = readdata(param);
@ -286,7 +286,7 @@ void * smtppchild(struct clientparam* param) {
#endif #endif
RETURN (sockmap(param, 180)); RETURN (mapsocket(param, 180));
CLEANRET: CLEANRET:

97
src/sockmap.c
View File

@ -37,6 +37,7 @@ int splicemap(struct clientparam * param, int timeo){
int pipesrv[2] = {-1,-1}; int pipesrv[2] = {-1,-1};
int pipecli[2] = {-1,-1}; int pipecli[2] = {-1,-1};
uint64_t sent=0, received=0; uint64_t sent=0, received=0;
SASIZETYPE sasize;
int res = 0, stop = 0; int res = 0, stop = 0;
int srvstate = 0, clistate = 0; int srvstate = 0, clistate = 0;
int insrvpipe = 0, inclipipe = 0; int insrvpipe = 0, inclipipe = 0;
@ -44,6 +45,23 @@ int splicemap(struct clientparam * param, int timeo){
int sleeptime = 0; int sleeptime = 0;
if(param->srv->usesplice > 1 && !param->waitserver64 && !param->waitclient64){
if(param->clioffset == param->cliinbuf){
param->clioffset = param->cliinbuf = 0;
if(param->clibuf){
myfree(param->clibuf);
param->clibuf = NULL;
}
}
if(param->srvoffset == param->srvinbuf){
param->srvoffset = param->srvinbuf = 0;
if(param->srvbuf){
myfree(param->srvbuf);
param->srvbuf = NULL;
}
}
}
param->res = 0; param->res = 0;
if(pipe(pipecli) < 0) RETURN(21); if(pipe(pipecli) < 0) RETURN(21);
if(pipe(pipesrv) < 0) RETURN(21); if(pipe(pipesrv) < 0) RETURN(21);
@ -53,30 +71,37 @@ int splicemap(struct clientparam * param, int timeo){
while(!stop && !conf.timetoexit){ while(!stop && !conf.timetoexit){
#ifdef NOIPV6
sasize = sizeof(struct sockaddr_in);
#else
sasize = sizeof(struct sockaddr_in6);
#endif
fds[0].events = fds[1].events = 0; fds[0].events = fds[1].events = 0;
if(srvstate && !param->waitclient64){ if((srvstate || param->srvinbuf > param->srvoffset) && !param->waitclient64){
#if DEBUGLEVEL > 2 #if DEBUGLEVEL > 2
(*param->srv->logfunc)(param, "splice: will send to client"); (*param->srv->logfunc)(param, "splice: will send to client");
#endif #endif
fds[0].events |= POLLOUT; fds[0].events |= POLLOUT;
} }
rfromserver = MAXSPLICE; rfromserver = MAXSPLICE;
if(param->waitserver64) rfromserver = MIN(MAXSPLICE, param->waitserver64 - (received + insrvpipe)); if(param->srvinbuf > param->srvoffset) rfromserver = 0;
else if(param->waitserver64) rfromserver = MIN(MAXSPLICE, param->waitserver64 - (received + insrvpipe));
if(srvstate < 2 && rfromserver > 0) { if(srvstate < 2 && rfromserver > 0) {
#if DEBUGLEVEL > 2 #if DEBUGLEVEL > 2
(*param->srv->logfunc)(param, "splice: will recv from server"); (*param->srv->logfunc)(param, "splice: will recv from server");
#endif #endif
fds[1].events |= POLLIN; fds[1].events |= POLLIN;
} }
if(clistate && !param->waitserver64){ if((clistate || param->cliinbuf > param->clioffset)&& !param->waitserver64){
#if DEBUGLEVEL > 2 #if DEBUGLEVEL > 2
(*param->srv->logfunc)(param, "splice: will send to server"); (*param->srv->logfunc)(param, "splice: will send to server");
#endif #endif
fds[1].events |= POLLOUT; fds[1].events |= POLLOUT;
} }
rfromclient = MAXSPLICE; rfromclient = MAXSPLICE;
if(param->waitclient64) rfromclient = MIN(MAXSPLICE, param->waitclient64 - (sent + inclipipe)); if(param->cliinbuf > param->clioffset) rfromclient = 0;
else if(param->waitclient64) rfromclient = MIN(MAXSPLICE, param->waitclient64 - (sent + inclipipe));
if(clistate < 2 && rfromclient > 0) { if(clistate < 2 && rfromclient > 0) {
#if DEBUGLEVEL > 2 #if DEBUGLEVEL > 2
(*param->srv->logfunc)(param, "splice :will recv from client"); (*param->srv->logfunc)(param, "splice :will recv from client");
@ -104,17 +129,37 @@ int splicemap(struct clientparam * param, int timeo){
param->res = 90; param->res = 90;
} }
if((fds[0].revents & POLLOUT)){ if((fds[0].revents & POLLOUT)){
if (param->srvinbuf > param->srvoffset) {
#if DEBUGLEVEL > 2 #if DEBUGLEVEL > 2
(*param->srv->logfunc)(param, "splice: send to client"); (*param->srv->logfunc)(param, "splice: non-spliced send to client");
#endif #endif
res = splice(pipesrv[0], NULL, param->clisock, NULL, MIN(MAXSPLICE, insrvpipe), SPLICE_F_NONBLOCK | SPLICE_F_MORE | SPLICE_F_MOVE); res = so._sendto(param->clisock, (char *)param->srvbuf + param->srvoffset,(!param->waitserver64 || (param->waitserver64 - received) > (param->srvinbuf - param->srvoffset))? param->srvinbuf - param->srvoffset : (int)(param->waitserver64 - received), 0, (struct sockaddr*)&param->sincr, sasize);
}
else {
#if DEBUGLEVEL > 2
(*param->srv->logfunc)(param, "splice: spliced send to client");
#endif
res = splice(pipesrv[0], NULL, param->clisock, NULL, MIN(MAXSPLICE, insrvpipe), SPLICE_F_NONBLOCK | SPLICE_F_MORE | SPLICE_F_MOVE);
}
if(res < 0) { if(res < 0) {
#if DEBUGLEVEL > 2
(*param->srv->logfunc)(param, "splice: send to client error");
#endif
if(errno != EAGAIN && errno != EINTR) RETURN(96); if(errno != EAGAIN && errno != EINTR) RETURN(96);
if(errno == EINTR) usleep(SLEEPTIME); if(errno == EINTR) usleep(SLEEPTIME);
continue; continue;
} }
if(res){ if(res){
insrvpipe -= res; if (param->srvinbuf > param->srvoffset){
param->srvinbuf = param->srvoffset = 0;
if(param->srv->usesplice > 1 && !param->waitclient64 && !param->waitserver64){
if(param->srvbuf){
myfree(param->srvbuf);
param->srvbuf = NULL;
}
}
}
else insrvpipe -= res;
received += res; received += res;
if(param->bandlimfunc) { if(param->bandlimfunc) {
@ -128,17 +173,40 @@ int splicemap(struct clientparam * param, int timeo){
} }
} }
if((fds[1].revents & POLLOUT)){ if((fds[1].revents & POLLOUT)){
if(param->cliinbuf > param->clioffset){
#if DEBUGLEVEL > 2 #if DEBUGLEVEL > 2
(*param->srv->logfunc)(param, "splice: send to server"); (*param->srv->logfunc)(param, "splice: non-spliced send to server");
#endif #endif
res = splice(pipecli[0], NULL, param->remsock, NULL, MIN(MAXSPLICE, inclipipe), SPLICE_F_NONBLOCK | SPLICE_F_MORE | SPLICE_F_MOVE); res = so._sendto(param->remsock, (char *)param->clibuf + param->clioffset, (!param->waitclient64 || (param->waitclient64 - sent) > (param->cliinbuf - param->clioffset))? param->cliinbuf - param->clioffset : (int)(param->waitclient64 - sent), 0, (struct sockaddr*)&param->sinsr, sasize);
}
else {
#if DEBUGLEVEL > 2
(*param->srv->logfunc)(param, "splice: spliced send to server");
#endif
res = splice(pipecli[0], NULL, param->remsock, NULL, MIN(MAXSPLICE, inclipipe), SPLICE_F_NONBLOCK | SPLICE_F_MORE | SPLICE_F_MOVE);
}
if(res < 0) { if(res < 0) {
#if DEBUGLEVEL > 2
(*param->srv->logfunc)(param, "splice: send to server error");
#endif
if(errno != EAGAIN && errno != EINTR) RETURN(97); if(errno != EAGAIN && errno != EINTR) RETURN(97);
if(errno == EINTR) usleep(SLEEPTIME); if(errno == EINTR) usleep(SLEEPTIME);
continue; continue;
} }
if(res){ if(res){
inclipipe -= res; if(param->cliinbuf > param->clioffset){
param->clioffset += res;
if(param->clioffset == param->cliinbuf){
param->clioffset = param->cliinbuf = 0;
if(param->srv->usesplice > 1 && !param->waitclient64 && !param->waitserver64){
if(param->clibuf){
myfree(param->clibuf);
param->clibuf = NULL;
}
}
}
}
else inclipipe -= res;
sent += res; sent += res;
param->nwrites++; param->nwrites++;
param->statscli64 += res; param->statscli64 += res;
@ -150,7 +218,6 @@ int splicemap(struct clientparam * param, int timeo){
} }
clistate = 0; clistate = 0;
} }
else clistate = 2;
if(param->waitclient64 && param->waitclient64 <= sent){ if(param->waitclient64 && param->waitclient64 <= sent){
RETURN (99); RETURN (99);
} }
@ -282,7 +349,11 @@ int sockmap(struct clientparam * param, int timeo){
while (!stop&&!conf.timetoexit){ while (!stop&&!conf.timetoexit){
#ifdef NOIPV6
sasize = sizeof(struct sockaddr_in); sasize = sizeof(struct sockaddr_in);
#else
sasize = sizeof(struct sockaddr_in6);
#endif
if(param->version < conf.version){ if(param->version < conf.version){
if((res = (*param->srv->authfunc)(param)) && res != 2 && !param->srv->noforce) {return(res);} if((res = (*param->srv->authfunc)(param)) && res != 2 && !param->srv->noforce) {return(res);}
param->paused = conf.paused; param->paused = conf.paused;
@ -417,7 +488,11 @@ int sockmap(struct clientparam * param, int timeo){
} }
} }
if (!stop && (fds[1].revents & POLLIN)) { if (!stop && (fds[1].revents & POLLIN)) {
#ifdef NOIPV6
struct sockaddr_in sin; struct sockaddr_in sin;
#else
struct sockaddr_in6 sin;
#endif
#if DEBUGLEVEL > 2 #if DEBUGLEVEL > 2
(*param->srv->logfunc)(param, "recv from server"); (*param->srv->logfunc)(param, "recv from server");
#endif #endif

4
src/socks.c
View File

@ -279,7 +279,7 @@ fflush(stderr);
if(buf)myfree(buf); if(buf)myfree(buf);
return (*param->redirectfunc)(param); return (*param->redirectfunc)(param);
} }
param->res = sockmap(param, conf.timeouts[CONNECTION_L]); param->res = mapsocket(param, conf.timeouts[CONNECTION_L]);
break; break;
case 2: case 2:
so._listen (param->remsock, 1); so._listen (param->remsock, 1);
@ -325,7 +325,7 @@ fflush(stderr);
socksend(param->clisock, buf, 8, conf.timeouts[STRING_S]); socksend(param->clisock, buf, 8, conf.timeouts[STRING_S]);
} }
param->res = sockmap(param, conf.timeouts[CONNECTION_S]); param->res = mapsocket(param, conf.timeouts[CONNECTION_S]);
break; break;
case 3: case 3:
param->sinsr = param->req; param->sinsr = param->req;