proxy -s socks -s -tcppm -s 8080 192.168.1.1 80 +tcppm -s3 8080 192.168.1.1 80--s0 turns it off explicitly. In builds without it (CMake option 3PROXY_USE_SPLICE=OFF, +-s alone splices both directions where it pays: an HTTP body of known length if at +least 16K of it is left to read, a stream (CONNECT, SOCKS, tcppm) once it carries +large reads, going back to copying when they turn small. -s1 always splices what the +server sends to the client (downloads), -s2 what the client sends to the server +(uploads), -s3 both; -s5, -s6 and -s7 (the same as -s) do it for those directions +only where it pays. -s0, the default, turns it off. In builds without it (CMake option 3PROXY_USE_SPLICE=OFF, or -DWITHSPLICE removed from the Makefile) -s is accepted and ignored.
splice() pays off when most of the traffic is large transfers: in the benchmarks it moved 1.4 to 1.7 times more data than the read/write path at about half the CPU per gigabyte. It does not help small requests and short connections, which go slower with it, and -every spliced connection needs two more pipes (file descriptors) and more kernel +every spliced direction needs a pipe (two file descriptors) and more kernel buffer memory. Data filters are not applied to spliced data.
TCP_NODELAY and splice() do not conflict and can be combined on high-speed connections. diff --git a/man/3proxy.cfg.5 b/man/3proxy.cfg.5 index 19668e3..c6819a4 100644 --- a/man/3proxy.cfg.5 +++ b/man/3proxy.cfg.5 @@ -203,12 +203,17 @@ delay GRACE_DELAY milliseconds before polling if average polling size is below G view without ability to reset. .br (for TCP services, Linux) use splice() to pass data between the client and the -server without copying it through 3proxy. It is off unless -s is given (-s0 turns -it off explicitly). splice() suits large transfers: they go faster and use less -CPU. Small requests and short connections go slower with it, every connection -needs two more pipes (file descriptors) and more kernel memory, and data filters -are not applied to spliced data. Available in builds with WITHSPLICE, the default -on Linux. +server without copying it through 3proxy. \fB-s\fR alone splices both directions where +it pays: a body of known length if at least 16K of it is left to read, a stream +once it carries large reads, back to copying when they turn small. \fB-s1\fR always +splices what the server sends to the client (downloads), \fB-s2\fR what the client +sends (uploads), \fB-s3\fR both; adding 4 (\fB-s5\fR, \fB-s6\fR, \fB-s7\fR, the last the same as +\fB-s\fR) splices those directions only where it pays. \fB-s0\fR, the default, turns it +off. splice() suits large transfers: they go faster and use less CPU; small +requests go slower with it. A spliced direction needs a pipe (two file +descriptors), made when first needed and kept for the connection, and more +kernel memory; data filters are not applied to spliced data. Available in builds +with WITHSPLICE, the default on Linux. .br .B -C (for TCP services) a connection closed by any of the sides terminates the diff --git a/src/proxy.h b/src/proxy.h index 19dbccc..1197254 100644 --- a/src/proxy.h +++ b/src/proxy.h @@ -239,6 +239,11 @@ extern struct extparam conf; extern int timeouts[12]; +/* directions sockmap() passes with splice(): -s1, -s2, -s3; with + SPLICE_AUTO (-s alone, -s5 to -s7) only when it pays */ +#define SPLICE_DOWNLOAD 1 +#define SPLICE_UPLOAD 2 +#define SPLICE_AUTO 4 int sockmap(struct clientparam * param, int timeo, int usesplice); int udpsockmap(struct clientparam * param, int timeo); int udpbind(struct clientparam * param); @@ -481,7 +486,7 @@ extern struct datatype datatypes[64]; extern struct commands commandhandlers[]; #ifdef WITHSPLICE -#define mapsocket(a,b) ((a->srv->usesplice && !a->ndatfilterssrv && !a->ndatfilterscli)?sockmap(a,b,1):sockmap(a,b,0)) +#define mapsocket(a,b) ((a->srv->usesplice && !a->ndatfilterssrv && !a->ndatfilterscli)?sockmap(a,b,a->srv->usesplice):sockmap(a,b,0)) #else #define mapsocket(a,b) sockmap(a,b, 0) #endif diff --git a/src/proxymain.c b/src/proxymain.c index bda76bd..a9c4b92 100644 --- a/src/proxymain.c +++ b/src/proxymain.c @@ -366,7 +366,9 @@ int MODULEMAINFUNC (int argc, char** argv){ " -De(DEVICENAME) bind external interface to device, e.g. eth1\n" #endif #ifdef WITHSPLICE - " -s Use splice() - no filtering for data, off by default\n" + " -s[N] Use splice(): -s1 server to client, -s2 client to server, -s3 both\n" + " directions; -s (-s7) both, -s5 or -s6 one, only where it pays (large\n" + " transfers) - no filtering for data, off by default\n" " -C connection closed by any of the sides terminates the session, by default\n" " the session is kept until both sides close it (TCP half-close)\n" " -U(0-3) (for socks) what to do when the destination changes within an UDP\n" @@ -377,7 +379,7 @@ int MODULEMAINFUNC (int argc, char** argv){ " -fFORMAT logging format (see documentation)\n" " -l log to stderr\n" " -lFILENAME log to FILENAME\n" - " -b(BUFSIZE) size of network buffer (default 4096 for TCP, 16384 for UDP)\n" + " -b(BUFSIZE) size of network buffer (default 65536 for TCP, 16384 for UDP)\n" " -S(STACKSIZE) value to add to default client thread stack size\n" " -t be silent (do not log service start/stop)\n" "\n" @@ -698,7 +700,9 @@ int MODULEMAINFUNC (int argc, char** argv){ srv.s_option = 1 + atoi(argv[i]+2); #ifdef WITHSPLICE else - srv.usesplice = *(argv[i]+2)? atoi(argv[i]+2) : 1; + srv.usesplice = (*(argv[i]+2)? atoi(argv[i]+2) : + SPLICE_DOWNLOAD|SPLICE_UPLOAD|SPLICE_AUTO) & + (SPLICE_DOWNLOAD|SPLICE_UPLOAD|SPLICE_AUTO); #endif break; case 'o': @@ -1325,6 +1329,7 @@ void srvinit(struct srvparam * srv, struct clientparam *param){ param->version = srv->version; param->paused = srv->paused; param->remsock = param->clisock = param->ctrlsock = param->ctrlsocksrv = INVALID_SOCKET; + param->pipecli[0] = param->pipecli[1] = param->pipesrv[0] = param->pipesrv[1] = -1; *SAFAMILY(¶m->req) = *SAFAMILY(¶m->sinsl) = *SAFAMILY(¶m->sinsr) = *SAFAMILY(¶m->sincr) = *SAFAMILY(¶m->sincl) = AF_INET; _3proxy_mutex_init(&srv->counter_mutex); (void)_3proxy_sem_init_timed(srv->childslots, srv->maxchild, srv->maxchild); @@ -1447,6 +1452,12 @@ void freeparam(struct clientparam * param) { } if(param->clibuf) free(param->clibuf); if(param->srvbuf) free(param->srvbuf); +#ifdef WITHSPLICE + if(param->pipecli[0] >= 0) close(param->pipecli[0]); + if(param->pipecli[1] >= 0) close(param->pipecli[1]); + if(param->pipesrv[0] >= 0) close(param->pipesrv[0]); + if(param->pipesrv[1] >= 0) close(param->pipesrv[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: diff --git a/src/sockmap.c b/src/sockmap.c index 0e993bb..6813833 100644 --- a/src/sockmap.c +++ b/src/sockmap.c @@ -39,6 +39,18 @@ ssize_t splice(int fd_in, loff_t *off_in, int fd_out, loff_t *off_out, size_t le #define MAXSPLICE 65536 +/* With SPLICE_AUTO, when a direction allowed to splice actually does + (without it, an allowed direction always does). + A body of known length is spliced if at least SPLICE_MIN of it is left + to read once what is already buffered is counted: a smaller one, or one + read whole with its headers, is cheaper to copy. + A stream of unknown length starts copied and is spliced after + SPLICE_STREAK reads in a row of at least SPLICE_MIN each, and goes back to + copying after SPLICE_STREAK+1 splices in a row of less than SPLICE_SMALL. */ +#define SPLICE_MIN 16384 +#define SPLICE_SMALL 4096 +#define SPLICE_STREAK 2 + #endif #define MIN(a,b) ((a>b)?b:a) @@ -49,6 +61,9 @@ ssize_t splice(int fd_in, loff_t *off_in, int fd_out, loff_t *off_out, size_t le before the sockets are closed. */ #define SESSIONEND (!halfclose && (CLIENTTERMREAD || SERVERTERMREAD)) +/* usesplice is a mask of the directions passed with splice() rather than + through a buffer: SPLICE_DOWNLOAD, server to client, and SPLICE_UPLOAD, + client to server. */ int sockmap(struct clientparam * param, int timeo, int usesplice){ int halfclose = param->srv->halfclose; uint64_t fromclient=0x7fffffffffffffff, fromserver =0x7fffffffffffffff; @@ -76,16 +91,30 @@ int sockmap(struct clientparam * param, int timeo, int usesplice){ #ifdef WITHSPLICE uint64_t inclientpipe = 0, inserverpipe = 0; int TOCLIENTPIPE = 0, FROMCLIENTPIPE = 0, TOSERVERPIPE = 0, FROMSERVERPIPE = 0; - int pipesrv[2] = {-1,-1}; - int pipecli[2] = {-1,-1}; + /* The client's data goes through pipecli, the server's through pipesrv. + They belong to the connection: made when first needed, kept for the + next mapping, closed with the connection. */ + int *pipecli = param->pipecli, *pipesrv = param->pipesrv; + /* allowed to splice (-s), splicing now, and a run of reads counted + towards switching; wantbuf: go back to copying once the pipe is empty */ + int allowcli = 0, allowsrv = 0, splicecli = 0, splicesrv = 0, autosplice; + int streakcli = 0, streaksrv = 0, wantbufcli = 0, wantbufsrv = 0; if((!param->nolongdatfilter && (param->ndatfilterscli > 0 || param->ndatfilterssrv))) usesplice = 0; - if(usesplice){ - TOCLIENTPIPE = FROMCLIENTPIPE = TOSERVERPIPE = FROMSERVERPIPE = 1; - TOCLIENTBUF = TOSERVERBUF = 0; - if(pipe2(pipecli, O_NONBLOCK) < 0) RETURN (21); - if(pipe2(pipesrv, O_NONBLOCK) < 0) RETURN (21); + allowcli = (usesplice & SPLICE_UPLOAD) != 0; + allowsrv = (usesplice & SPLICE_DOWNLOAD) != 0; + autosplice = (usesplice & SPLICE_AUTO) != 0; + if(!autosplice){ + splicecli = allowcli; + splicesrv = allowsrv; } + TOCLIENTPIPE = FROMCLIENTPIPE = allowcli; + TOSERVERPIPE = FROMSERVERPIPE = allowsrv; +#else +#define splicecli 0 +#define splicesrv 0 +#define wantbufcli 0 +#define wantbufsrv 0 #endif inserverbuf = param->srvinbuf - param->srvoffset; @@ -108,16 +137,16 @@ int sockmap(struct clientparam * param, int timeo, int usesplice){ if(inserverbuf >= fromserver) FROMSERVER = 0; if(inclientbuf >= fromclient) FROMCLIENT = 0; #ifdef WITHSPLICE - if(!usesplice) + /* a body of known length large enough is spliced from the start */ + if(autosplice && allowsrv && param->waitserver64 && fromserver > inserverbuf && fromserver - inserverbuf >= SPLICE_MIN) splicesrv = 1; + if(autosplice && allowcli && param->waitclient64 && fromclient > inclientbuf && fromclient - inclientbuf >= SPLICE_MIN) splicecli = 1; #endif - { - if(fromserver && !param->srvbuf && (!(param->srvbuf=malloc(SRVBUFSIZE)) || !(param->srvbufsize = SRVBUFSIZE))){ - RETURN (21); - } - if(fromclient && !param->clibuf && (!(param->clibuf=malloc(SRVBUFSIZE)) || !(param->clibufsize = SRVBUFSIZE))){ - RETURN (21); - } - + /* a spliced direction needs no buffer, only what was read before */ + if(!splicesrv && fromserver && !param->srvbuf && (!(param->srvbuf=malloc(SRVBUFSIZE)) || !(param->srvbufsize = SRVBUFSIZE))){ + RETURN (21); + } + if(!splicecli && fromclient && !param->clibuf && (!(param->clibuf=malloc(SRVBUFSIZE)) || !(param->clibufsize = SRVBUFSIZE))){ + RETURN (21); } if(param->srvinbuf == param->srvoffset) param->srvinbuf =param->srvoffset = 0; if(param->cliinbuf == param->clioffset) param->cliinbuf =param->clioffset = 0; @@ -292,8 +321,21 @@ log("done send to client from buf"); } } #ifdef WITHSPLICE - if(usesplice){ - if(inclientpipe && !inclientbuf && FROMCLIENTPIPE && TOSERVER){ + /* a direction going back to copying does once its pipe is empty */ + if(wantbufcli && !inclientpipe){ + splicecli = wantbufcli = streakcli = 0; + if(!param->clibuf && (!(param->clibuf=malloc(SRVBUFSIZE)) || !(param->clibufsize = SRVBUFSIZE))) RETURN (21); + if(param->cliinbuf < param->clibufsize) TOCLIENTBUF = 1; + } + if(wantbufsrv && !inserverpipe){ + splicesrv = wantbufsrv = streaksrv = 0; + if(!param->srvbuf && (!(param->srvbuf=malloc(SRVBUFSIZE)) || !(param->srvbufsize = SRVBUFSIZE))) RETURN (21); + if(param->srvinbuf < param->srvbufsize) TOSERVERBUF = 1; + } + { + /* each direction goes through its pipe if spliced, through its + buffer otherwise; what was in the buffer before is sent first */ + if(splicecli && inclientpipe && !inclientbuf && FROMCLIENTPIPE && TOSERVER){ #ifdef WITHLOG log("send to server from pipe"); #endif @@ -330,7 +372,7 @@ log(logbuf); } } } - if(inserverpipe && !inserverbuf && FROMSERVERPIPE && TOCLIENT){ + if(splicesrv && inserverpipe && !inserverbuf && FROMSERVERPIPE && TOCLIENT){ #ifdef WITHLOG log("send to client from pipe"); #endif @@ -360,12 +402,17 @@ log(logbuf); } } } - if(fromclient>inclientpipe && FROMCLIENT && TOCLIENTPIPE && !SESSIONEND){ - int error; - socklen_t len=sizeof(error); + if(splicecli && !wantbufcli && fromclient>inclientpipe && FROMCLIENT && TOCLIENTPIPE && !SESSIONEND){ #ifdef WITHLOG log("read from client to pipe"); #endif + if(pipecli[0] < 0 && pipe2(pipecli, O_NONBLOCK) < 0){ + /* no pipe to be had: copy instead */ + allowcli = splicecli = TOCLIENTPIPE = FROMCLIENTPIPE = 0; + if(!param->clibuf && (!(param->clibuf=malloc(SRVBUFSIZE)) || !(param->clibufsize = SRVBUFSIZE))) RETURN (21); + if(param->cliinbuf < param->clibufsize) TOCLIENTBUF = 1; + continue; + } errno = 0; res = splice(param->clisock, NULL, pipecli[1], NULL, (int)MIN((uint64_t)MAXSPLICE - inclientpipe, (uint64_t)fromclient-inclientpipe), SPLICE_F_NONBLOCK|SPLICE_F_MOVE); #ifdef WITHLOG @@ -396,19 +443,27 @@ log("done read from client to pipe"); graceclinum++; graceclitraf += res; gracesrvnum = gracesrvtraf = 0; + if(autosplice && !param->waitclient64){ + if(res >= SPLICE_SMALL) streakcli = 0; + else if(++streakcli > SPLICE_STREAK) wantbufcli = 1; + } inclientpipe += res; if(inclientpipe >= MAXSPLICE) TOCLIENTPIPE = 0; needaction = 0; continue; } } - if(fromserver > inserverpipe && FROMSERVER && TOSERVERPIPE && !SESSIONEND){ - int error; - socklen_t len=sizeof(error); - errno = 0; + if(splicesrv && !wantbufsrv && fromserver > inserverpipe && FROMSERVER && TOSERVERPIPE && !SESSIONEND){ #ifdef WITHLOG log("read from server to pipe\n"); #endif + if(pipesrv[0] < 0 && pipe2(pipesrv, O_NONBLOCK) < 0){ + allowsrv = splicesrv = TOSERVERPIPE = FROMSERVERPIPE = 0; + if(!param->srvbuf && (!(param->srvbuf=malloc(SRVBUFSIZE)) || !(param->srvbufsize = SRVBUFSIZE))) RETURN (21); + if(param->srvinbuf < param->srvbufsize) TOSERVERBUF = 1; + continue; + } + errno = 0; res = splice(param->remsock, NULL, pipesrv[1], NULL, MIN(MAXSPLICE - inserverpipe, fromserver - inserverpipe), SPLICE_F_NONBLOCK|SPLICE_F_MOVE); #ifdef WITHLOG log("server to pipe splice finished\n"); @@ -438,6 +493,10 @@ log("done read from server to pipe\n"); gracesrvnum++; gracesrvtraf += res; graceclinum = graceclitraf = 0; + if(autosplice && !param->waitserver64){ + if(res >= SPLICE_SMALL) streaksrv = 0; + else if(++streaksrv > SPLICE_STREAK) wantbufsrv = 1; + } param->nreads++; param->statssrv64 += res; inserverpipe += res; @@ -452,10 +511,9 @@ log("done read from server to pipe\n"); } } } - else #endif { - if(fromclient > inclientbuf && FROMCLIENT && TOCLIENTBUF && !SESSIONEND){ + if(!splicecli && fromclient > inclientbuf && FROMCLIENT && TOCLIENTBUF && !SESSIONEND){ #ifdef WITHLOG log("read from client to buf"); #endif @@ -480,6 +538,17 @@ log("done read from client to buf"); graceclinum++; graceclitraf += res; gracesrvnum = gracesrvtraf = 0; +#ifdef WITHSPLICE + if(autosplice && allowcli && !param->waitclient64){ + if(res < SPLICE_MIN) streakcli = 0; + else if(++streakcli >= SPLICE_STREAK){ + /* a stream: splice from now on, after what is + buffered has been sent */ + splicecli = TOCLIENTPIPE = FROMCLIENTPIPE = 1; + streakcli = 0; + } + } +#endif inclientbuf += res; param->cliinbuf += res; if(param->clibufsize == param->cliinbuf) TOCLIENTBUF = 0; @@ -488,7 +557,7 @@ log("done read from client to buf"); } } - if(fromserver > inserverbuf && FROMSERVER && TOSERVERBUF && !SESSIONEND){ + if(!splicesrv && fromserver > inserverbuf && FROMSERVER && TOSERVERBUF && !SESSIONEND){ #ifdef WITHLOG log("read from server to buf"); #endif @@ -513,6 +582,15 @@ log("done read from server to buf"); gracesrvnum++; gracesrvtraf += res; graceclinum = graceclitraf = 0; +#ifdef WITHSPLICE + if(autosplice && allowsrv && !param->waitserver64){ + if(res < SPLICE_MIN) streaksrv = 0; + else if(++streaksrv >= SPLICE_STREAK){ + splicesrv = TOSERVERPIPE = FROMSERVERPIPE = 1; + streaksrv = 0; + } + } +#endif param->nreads++; param->statssrv64 += res; inserverbuf += res; @@ -553,15 +631,8 @@ log("done read from server to buf"); // if(!CLIENTTERMREAD || !CLIENTTERMWRITE){ if(!after){ fds[fdsc].fd = param->clisock; - if(fromclient && !CLIENTTERMREAD && !FROMCLIENT && !SESSIONEND && (( -#ifdef WITHSPLICE - !usesplice && -#endif - TOCLIENTBUF) -#ifdef WITHSPLICE - || (usesplice) -#endif - )) + if(fromclient && !CLIENTTERMREAD && !FROMCLIENT && !SESSIONEND && + ((!splicecli && TOCLIENTBUF) || (splicecli && !wantbufcli))) cli_events |= POLLIN; if(!TOCLIENT && !CLIENTTERMWRITE && (inserverbuf #ifdef WITHSPLICE @@ -613,15 +684,8 @@ log("ready to write to client"); // if(!SERVERTERMREAD || !SERVERTERMWRITE){ if(!after){ fds[fdsc].fd = param->remsock; - if(fromserver && !SERVERTERMREAD && !FROMSERVER && !SESSIONEND && (( -#ifdef WITHSPLICE - !usesplice && -#endif - TOSERVERBUF) -#ifdef WITHSPLICE - || (usesplice) -#endif - )) + if(fromserver && !SERVERTERMREAD && !FROMSERVER && !SESSIONEND && + ((!splicesrv && TOSERVERBUF) || (splicesrv && !wantbufsrv))) srv_events |= POLLIN; if(!TOSERVER && !SERVERTERMWRITE && (inclientbuf #ifdef WITHSPLICE @@ -678,8 +742,8 @@ log("server terminated connection"); fdsc++; // } #ifdef WITHSPLICE - if(usesplice){ - if(fromclient>inclientpipe && !TOCLIENTPIPE && inclientpipe < MAXSPLICE){ + { + if(splicecli && fromclient>inclientpipe && !TOCLIENTPIPE && inclientpipe < MAXSPLICE){ if(!after){ #ifdef WITHLOG log("wait writing to client pipe"); @@ -700,7 +764,7 @@ log("ready to write to client pipe"); } fdsc++; } - if(inclientpipe && !FROMCLIENTPIPE){ + if(splicecli && inclientpipe && !FROMCLIENTPIPE){ if(!after){ #ifdef WITHLOG log("wait reading from client pipe"); @@ -719,7 +783,7 @@ log("ready reading from client pipe"); } fdsc++; } - if(fromserver>inserverpipe && !TOSERVERPIPE && inserverpipe < MAXSPLICE){ + if(splicesrv && fromserver>inserverpipe && !TOSERVERPIPE && inserverpipe < MAXSPLICE){ if(!after){ #ifdef WITHLOG log("wait writing to server pipe"); @@ -738,7 +802,7 @@ log("ready writing to server pipe"); } fdsc++; } - if(inserverpipe && !FROMSERVERPIPE){ + if(splicesrv && inserverpipe && !FROMSERVERPIPE){ if(!after){ #ifdef WITHLOG log("wait reading from server pipe"); @@ -822,12 +886,26 @@ log("timeout"); CLEANRET: #ifdef WITHSPLICE - if(pipecli[0] >= 0) close(pipecli[0]); - if(pipecli[1] >= 0) close(pipecli[1]); - if(pipesrv[0] >= 0) close(pipesrv[0]); - if(pipesrv[1] >= 0) close(pipesrv[1]); + /* empty pipes stay with the connection for its next mapping; one still + holding data, left by an error or a timeout, cannot be used again */ + if(inclientpipe && pipecli[0] >= 0){ + close(pipecli[0]); + close(pipecli[1]); + pipecli[0] = pipecli[1] = -1; + } + if(inserverpipe && pipesrv[0] >= 0){ + close(pipesrv[0]); + close(pipesrv[1]); + pipesrv[0] = pipesrv[1] = -1; + } #endif return res; } #undef SESSIONEND +#ifndef WITHSPLICE +#undef splicecli +#undef splicesrv +#undef wantbufcli +#undef wantbufsrv +#endif diff --git a/src/structures.h b/src/structures.h index 472e44a..5b9957a 100644 --- a/src/structures.h +++ b/src/structures.h @@ -743,6 +743,11 @@ struct clientparam { *clibuf, *srvbuf; + /* splice() pipes for the client's and the server's data, -1 until + made; kept for the life of the connection */ + int pipecli[2], + pipesrv[2]; + unsigned cliinbuf, srvinbuf, clioffset,