Index: benchmark/io/http/main.cfa =================================================================== --- benchmark/io/http/main.cfa (revision 6e2b04e05bb3d9f5059628fbcd9d7304e2577458) +++ benchmark/io/http/main.cfa (revision 7f0ac120e5b984eb491cf8672d2c267d81986ebd) @@ -38,6 +38,6 @@ thread StatsPrinter { - Worker * workers; - int worker_cnt; + connection ** conns; + volatile int conn_cnt; condition_variable(fast_block_lock) var; }; @@ -45,5 +45,5 @@ void ?{}( StatsPrinter & this, cluster & cl ) { ((thread&)this){ "Stats Printer Thread", cl }; - this.worker_cnt = 0; + this.conn_cnt = 0; } @@ -62,5 +62,5 @@ print_stats_now( *active_cluster(), CFA_STATS_READY_Q | CFA_STATS_IO ); - if(this.worker_cnt != 0) { + if(this.conn_cnt != 0) { uint64_t tries = 0; uint64_t calls = 0; @@ -74,13 +74,13 @@ memset(avgrd, 0, sizeof(avgrd)); - for(i; this.worker_cnt) { - tries += this.workers[i].stats.sendfile.tries; - calls += this.workers[i].stats.sendfile.calls; - header += this.workers[i].stats.sendfile.header; - splcin += this.workers[i].stats.sendfile.splcin; - splcot += this.workers[i].stats.sendfile.splcot; + for(i; this.conn_cnt) { + tries += this.conns[i]->stats.sendfile.tries; + calls += this.conns[i]->stats.sendfile.calls; + header += this.conns[i]->stats.sendfile.header; + splcin += this.conns[i]->stats.sendfile.splcin; + splcot += this.conns[i]->stats.sendfile.splcot; for(j; zipf_cnts) { - avgrd[j].calls += this.workers[i].stats.sendfile.avgrd[j].calls; - avgrd[j].bytes += this.workers[i].stats.sendfile.avgrd[j].bytes; + avgrd[j].calls += this.conns[i]->stats.sendfile.avgrd[j].calls; + avgrd[j].bytes += this.conns[i]->stats.sendfile.avgrd[j].bytes; } } @@ -88,5 +88,5 @@ double ratio = ((double)tries) / calls; - sout | "----- Worker Stats -----"; + sout | "----- Connection Stats -----"; sout | "sendfile : " | calls | "calls," | tries | "tries (" | ratio | " try/call)"; sout | " " | header | "header," | splcin | "splice in," | splcot | "splice out"; @@ -98,5 +98,5 @@ } else { - sout | "No Workers!"; + sout | "No Connections!"; } } @@ -182,4 +182,13 @@ //============================================================================================= +// REUSEPORT +//============================================================================================= + +size_t sockarr_size; +struct __attribute__((aligned(128))) Q { + mpsc_queue(PendingRead) q; +}; + +//============================================================================================= // Termination //============================================================================================= @@ -235,7 +244,4 @@ int server_fd; - if(!options.socket.manyreuse) { - server_fd = listener(address, addrlen); - } //=================== @@ -257,28 +263,63 @@ { // Stats printer makes a copy so this needs to persist longer than normal - Worker * workers; + connection ** conns; + AcceptWorker * aworkers = 0p; + ChannelWorker * cworkers = 0p; + Acceptor * acceptors = 0p; + Q * queues = 0p; ServerCluster cl[options.clopts.nclusters]; init_protocol(); { - workers = anew(options.clopts.nworkers); - cl[0].prnt->workers = workers; - cl[0].prnt->worker_cnt = options.clopts.nworkers; - for(i; options.clopts.nworkers) { - // if( options.file_cache.fixed_fds ) { - // workers[i].pipe[0] = pipe_off + (i * 2) + 0; - // workers[i].pipe[1] = pipe_off + (i * 2) + 1; - // } - // else - { - workers[i].pipe[0] = fds[pipe_off + (i * 2) + 0]; - workers[i].pipe[1] = fds[pipe_off + (i * 2) + 1]; - workers[i].sockfd = options.socket.manyreuse ? listener(address, addrlen) : server_fd; - workers[i].addr = (struct sockaddr *)&address; - workers[i].addrlen = (socklen_t*)&addrlen; - workers[i].flags = 0; - } - unpark( workers[i] ); - } + conns = alloc(options.clopts.nworkers); + if(options.socket.reuseport) { + queues = alloc(options.clopts.nprocs); + acceptors = anew(options.clopts.nprocs); + for(i; options.clopts.nprocs) { + (queues[i]){}; + { + acceptors[i].sockfd = listener(address, addrlen); + acceptors[i].addr = (struct sockaddr *)&address; + acceptors[i].addrlen = (socklen_t*)&addrlen; + acceptors[i].flags = 0; + acceptors[i].queue = &queues[i].q; + } + unpark( acceptors[i] ); + } + + cworkers = anew(options.clopts.nworkers); + for(i; options.clopts.nworkers) { + { + cworkers[i].conn.pipe[0] = fds[pipe_off + (i * 2) + 0]; + cworkers[i].conn.pipe[1] = fds[pipe_off + (i * 2) + 1]; + cworkers[i].queue = &queues[i % options.clopts.nprocs].q; + conns[i] = &cworkers[i].conn; + } + unpark( cworkers[i] ); + } + } + else { + server_fd = listener(address, addrlen); + aworkers = anew(options.clopts.nworkers); + for(i; options.clopts.nworkers) { + // if( options.file_cache.fixed_fds ) { + // workers[i].pipe[0] = pipe_off + (i * 2) + 0; + // workers[i].pipe[1] = pipe_off + (i * 2) + 1; + // } + // else + { + aworkers[i].conn.pipe[0] = fds[pipe_off + (i * 2) + 0]; + aworkers[i].conn.pipe[1] = fds[pipe_off + (i * 2) + 1]; + aworkers[i].sockfd = server_fd; + aworkers[i].addr = (struct sockaddr *)&address; + aworkers[i].addrlen = (socklen_t*)&addrlen; + aworkers[i].flags = 0; + conns[i] = &aworkers[i].conn; + } + unpark( aworkers[i] ); + } + } + cl[0].prnt->conns = conns; + cl[0].prnt->conn_cnt = options.clopts.nworkers; sout | options.clopts.nworkers | "workers started on" | options.clopts.nprocs | "processors /" | options.clopts.nclusters | "clusters"; for(i; options.clopts.nclusters) { @@ -307,49 +348,87 @@ } - sout | "Notifying connections..." | nonl; flush( sout ); - for(i; options.clopts.nworkers) { - workers[i].done = true; - } - sout | "done"; - - sout | "Shutting down socket..." | nonl; flush( sout ); - if(options.socket.manyreuse) { - for(i; options.clopts.nworkers) { - ret = shutdown( workers[i].sockfd, SHUT_RD ); - if(ret < 0) abort( "close socket %d error: (%d) %s\n", i, (int)errno, strerror(errno) ); + //=================== + // Close Socket and join + if(options.socket.reuseport) { + sout | "Notifying connections..." | nonl; flush( sout ); + for(i; options.clopts.nprocs) { + acceptors[i].done = true; + } + for(i; options.clopts.nworkers) { + cworkers[i].done = true; + } + sout | "done"; + + sout | "Shutting down Socket..." | nonl; flush( sout ); + for(i; options.clopts.nprocs) { + ret = shutdown( acceptors[i].sockfd, SHUT_RD ); + if( ret < 0 ) { + abort( "shutdown1 error: (%d) %s\n", (int)errno, strerror(errno) ); + } + } + sout | "done"; + + sout | "Closing Socket..." | nonl; flush( sout ); + for(i; options.clopts.nprocs) { + ret = close( acceptors[i].sockfd ); + if( ret < 0) { + abort( "close socket error: (%d) %s\n", (int)errno, strerror(errno) ); + } + } + sout | "done"; + + sout | "Stopping accept threads..." | nonl; flush( sout ); + for(i; options.clopts.nprocs) { + join(acceptors[i]); + } + sout | "done"; + + sout | "Draining worker queues..." | nonl; flush( sout ); + for(i; options.clopts.nprocs) { + PendingRead * p = 0p; + while(p = pop(queues[i].q)) { + fulfil(p->f, -ECONNRESET); + } + } + sout | "done"; + + sout | "Stopping worker threads..." | nonl; flush( sout ); + for(i; options.clopts.nworkers) { + for(j; 2) { + ret = close(cworkers[i].conn.pipe[j]); + if(ret < 0) abort( "close pipe %d error: (%d) %s\n", j, (int)errno, strerror(errno) ); + } + join(cworkers[i]); } } else { + sout | "Notifying connections..." | nonl; flush( sout ); + for(i; options.clopts.nworkers) { + aworkers[i].done = true; + } + sout | "done"; + + sout | "Shutting down Socket..." | nonl; flush( sout ); ret = shutdown( server_fd, SHUT_RD ); if( ret < 0 ) { - abort( "shutdown error: (%d) %s\n", (int)errno, strerror(errno) ); - } - } - sout | "done"; - - //=================== - // Close Socket - sout | "Closing Socket..." | nonl; flush( sout ); - if(options.socket.manyreuse) { - for(i; options.clopts.nworkers) { - ret = close(workers[i].sockfd); - if(ret < 0) abort( "close socket %d error: (%d) %s\n", i, (int)errno, strerror(errno) ); - } - } - else { + abort( "shutdown2 error: (%d) %s\n", (int)errno, strerror(errno) ); + } + sout | "done"; + + sout | "Closing Socket..." | nonl; flush( sout ); ret = close( server_fd ); if(ret < 0) { abort( "close socket error: (%d) %s\n", (int)errno, strerror(errno) ); } - } - sout | "done"; - - sout | "Stopping connection threads..." | nonl; flush( sout ); - for(i; options.clopts.nworkers) { - for(j; 2) { - ret = close(workers[i].pipe[j]); - if(ret < 0) abort( "close pipe %d error: (%d) %s\n", j, (int)errno, strerror(errno) ); - } - join(workers[i]); + sout | "done"; + + sout | "Stopping connection threads..." | nonl; flush( sout ); + for(i; options.clopts.nworkers) { + for(j; 2) { + ret = close(aworkers[i].conn.pipe[j]); + if(ret < 0) abort( "close pipe %d error: (%d) %s\n", j, (int)errno, strerror(errno) ); + } + join(aworkers[i]); + } } } @@ -371,5 +450,9 @@ // Now that the stats printer is stopped, we can reclaim this - adelete(workers); + adelete(aworkers); + adelete(cworkers); + adelete(acceptors); + adelete(queues); + free(conns); sout | "Stopping processors/clusters..." | nonl; flush( sout ); @@ -377,13 +460,5 @@ sout | "done"; - // sout | "Closing splice fds..." | nonl; flush( sout ); - // for(i; pipe_cnt) { - // ret = close( fds[pipe_off + i] ); - // if(ret < 0) { - // abort( "close pipe error: (%d) %s\n", (int)errno, strerror(errno) ); - // } - // } free(fds); - sout | "done"; sout | "Stopping processors..." | nonl; flush( sout ); Index: benchmark/io/http/options.cfa =================================================================== --- benchmark/io/http/options.cfa (revision 6e2b04e05bb3d9f5059628fbcd9d7304e2577458) +++ benchmark/io/http/options.cfa (revision 7f0ac120e5b984eb491cf8672d2c267d81986ebd) @@ -38,6 +38,5 @@ 10, // backlog 1024, // buflen - false, // onereuse - false // manyreuse + false // reuseport }, @@ -72,6 +71,5 @@ {'\0', "shell", "Disable interactive mode", options.interactive, parse_setfalse}, {'\0', "accept-backlog", "Maximum number of pending accepts", options.socket.backlog}, - {'\0', "reuseport-one", "Create a single listen socket with SO_REUSEPORT", options.socket.onereuse, parse_settrue}, - {'\0', "reuseport", "Use many listen sockets with SO_REUSEPORT", options.socket.manyreuse, parse_settrue}, + {'\0', "reuseport", "Use acceptor threads with reuse port SO_REUSEPORT", options.socket.reuseport, parse_settrue}, {'\0', "request_len", "Maximum number of bytes in the http request, requests with more data will be answered with Http Code 414", options.socket.buflen}, {'\0', "seed", "seed to use for hashing", options.file_cache.hash_seed }, Index: benchmark/io/http/options.hfa =================================================================== --- benchmark/io/http/options.hfa (revision 6e2b04e05bb3d9f5059628fbcd9d7304e2577458) +++ benchmark/io/http/options.hfa (revision 7f0ac120e5b984eb491cf8672d2c267d81986ebd) @@ -27,6 +27,5 @@ int backlog; int buflen; - bool onereuse; - bool manyreuse; + bool reuseport; } socket; Index: benchmark/io/http/protocol.cfa =================================================================== --- benchmark/io/http/protocol.cfa (revision 6e2b04e05bb3d9f5059628fbcd9d7304e2577458) +++ benchmark/io/http/protocol.cfa (revision 7f0ac120e5b984eb491cf8672d2c267d81986ebd) @@ -30,4 +30,13 @@ #define PLAINTEXT_NOCOPY #define LINKED_IO + +static inline __s32 wait_res( io_future_t & this ) { + wait( this ); + if( this.result < 0 ) {{ + errno = -this.result; + return -1; + }} + return this.result; +} struct https_msg_str { @@ -470,4 +479,5 @@ if(is_error(splice_in.res)) { + if(splice_in.res.error == -EPIPE) return -ECONNRESET; mutex(serr) serr | "SPLICE IN failed with" | splice_in.res.error; close(fd); @@ -503,5 +513,5 @@ } -[HttpCode code, bool closed, * const char file, size_t len] http_read(int fd, []char buffer, size_t len) { +[HttpCode code, bool closed, * const char file, size_t len] http_read(volatile int & fd, []char buffer, size_t len, io_future_t * f) { char * it = buffer; size_t count = len - 1; @@ -509,5 +519,12 @@ READ: for() { - int ret = cfa_recv(fd, (void*)it, count, 0, CFA_IO_LAZY); + int ret; + if( f ) { + ret = wait_res(*f); + reset(*f); + f = 0p; + } else { + ret = cfa_recv(fd, (void*)it, count, 0, CFA_IO_LAZY); + } // int ret = read(fd, (void*)it, count); if(ret == 0 ) return [OK200, true, 0, 0]; Index: benchmark/io/http/protocol.hfa =================================================================== --- benchmark/io/http/protocol.hfa (revision 6e2b04e05bb3d9f5059628fbcd9d7304e2577458) +++ benchmark/io/http/protocol.hfa (revision 7f0ac120e5b984eb491cf8672d2c267d81986ebd) @@ -1,4 +1,5 @@ #pragma once +struct io_future_t; struct sendfile_stats_t; @@ -22,3 +23,3 @@ int answer_sendfile( int pipe[2], int fd, int ans_fd, size_t count, struct sendfile_stats_t & ); -[HttpCode code, bool closed, * const char file, size_t len] http_read(int fd, []char buffer, size_t len); +[HttpCode code, bool closed, * const char file, size_t len] http_read(volatile int & fd, []char buffer, size_t len, io_future_t * f); Index: benchmark/io/http/socket.cfa =================================================================== --- benchmark/io/http/socket.cfa (revision 6e2b04e05bb3d9f5059628fbcd9d7304e2577458) +++ benchmark/io/http/socket.cfa (revision 7f0ac120e5b984eb491cf8672d2c267d81986ebd) @@ -26,10 +26,12 @@ int listener(struct sockaddr_in & address, int addrlen) { - int sockfd = socket(AF_INET, SOCK_STREAM, 0); + int type = SOCK_STREAM; + if(options.socket.reuseport) type |= SOCK_NONBLOCK; + int sockfd = socket(AF_INET, type, 0); if(sockfd < 0) { abort( "socket error: (%d) %s\n", (int)errno, strerror(errno) ); } - if(options.socket.onereuse || options.socket.manyreuse) { + if(options.socket.reuseport) { int value = 1; // if (setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, (const void*)&on, sizeof(on))) Index: benchmark/io/http/worker.cfa =================================================================== --- benchmark/io/http/worker.cfa (revision 6e2b04e05bb3d9f5059628fbcd9d7304e2577458) +++ benchmark/io/http/worker.cfa (revision 7f0ac120e5b984eb491cf8672d2c267d81986ebd) @@ -8,4 +8,5 @@ #include #include +#include #include "options.hfa" @@ -13,41 +14,120 @@ #include "filecache.hfa" -//============================================================================================= -// Worker Thread -//============================================================================================= -void ?{}( Worker & this ) { +void ?{}( sendfile_stats_t & this ) { + this.calls = 0; + this.tries = 0; + this.header = 0; + this.splcin = 0; + this.splcot = 0; + for(i; zipf_cnts) { + this.avgrd[i].calls = 0; + this.avgrd[i].bytes = 0; + } +} + +//============================================================================================= +// Generic connection handling +//============================================================================================= +static void handle_connection( connection & this, volatile int & fd, char * buffer, size_t len, io_future_t * f ) { + REQUEST: + for() { + bool closed; + HttpCode code; + const char * file; + size_t name_size; + + // Read the http request + if( options.log ) sout | "=== Reading request ==="; + [code, closed, file, name_size] = http_read(fd, buffer, len, f); + f = 0p; + + // if we are done, break out of the loop + if( closed ) break REQUEST; + + // If this wasn't a request retrun 400 + if( code != OK200 ) { + sout | "=== Invalid Request :" | code_val(code) | "==="; + answer_error(fd, code); + continue REQUEST; + } + + if(0 == strncmp(file, "plaintext", min(name_size, sizeof("plaintext") ))) { + if( options.log ) sout | "=== Request for /plaintext ==="; + + int ret = answer_plaintext(fd); + if( ret == -ECONNRESET ) break REQUEST; + + if( options.log ) sout | "=== Answer sent ==="; + continue REQUEST; + } + + if(0 == strncmp(file, "ping", min(name_size, sizeof("ping") ))) { + if( options.log ) sout | "=== Request for /ping ==="; + + // Send the header + int ret = answer_empty(fd); + if( ret == -ECONNRESET ) break REQUEST; + + if( options.log ) sout | "=== Answer sent ==="; + continue REQUEST; + } + + if( options.log ) { + sout | "=== Request for file " | nonl; + write(sout, file, name_size); + sout | " ==="; + } + + if( !options.file_cache.path ) { + if( options.log ) { + sout | "=== File Not Found (" | nonl; + write(sout, file, name_size); + sout | ") ==="; + } + answer_error(fd, E405); + continue REQUEST; + } + + // Get the fd from the file cache + int ans_fd; + size_t count; + [ans_fd, count] = get_file( file, name_size ); + + // If we can't find the file, return 404 + if( ans_fd < 0 ) { + if( options.log ) { + sout | "=== File Not Found (" | nonl; + write(sout, file, name_size); + sout | ") ==="; + } + answer_error(fd, E404); + continue REQUEST; + } + + // Send the desired file + int ret = answer_sendfile( this.pipe, fd, ans_fd, count, this.stats.sendfile ); + if( ret == -ECONNRESET ) break REQUEST; + + if( options.log ) sout | "=== Answer sent ==="; + } +} + +//============================================================================================= +// Self Accepting Worker Thread +//============================================================================================= +void ?{}( AcceptWorker & this ) { size_t cli = rand() % options.clopts.cltr_cnt; ((thread&)this){ "Server Worker Thread", *options.clopts.instance[cli], 64000 }; options.clopts.thrd_cnt[cli]++; - this.pipe[0] = -1; - this.pipe[1] = -1; this.done = false; - - this.stats.sendfile.calls = 0; - this.stats.sendfile.tries = 0; - this.stats.sendfile.header = 0; - this.stats.sendfile.splcin = 0; - this.stats.sendfile.splcot = 0; - for(i; zipf_cnts) { - this.stats.sendfile.avgrd[i].calls = 0; - this.stats.sendfile.avgrd[i].bytes = 0; - } -} - -extern "C" { -extern int accept4(int sockfd, struct sockaddr *addr, socklen_t *addrlen, int flags); -} - -void main( Worker & this ) { +} + +void main( AcceptWorker & this ) { park(); - /* paranoid */ assert( this.pipe[0] != -1 ); - /* paranoid */ assert( this.pipe[1] != -1 ); - - const bool reuse = options.socket.manyreuse; - - CONNECTION: + /* paranoid */ assert( this.conn.pipe[0] != -1 ); + /* paranoid */ assert( this.conn.pipe[1] != -1 ); for() { if( options.log ) sout | "=== Accepting connection ==="; - int fd = cfa_accept4( this.[sockfd, addr, addrlen, flags], CFA_IO_LAZY ); + int fd = cfa_accept4( this.sockfd, this.[addr, addrlen, flags], CFA_IO_LAZY ); if(fd < 0) { if( errno == ECONNABORTED ) break; @@ -58,89 +138,86 @@ if( options.log ) sout | "=== New connection" | fd | "" | ", waiting for requests ==="; - REQUEST: - for() { - bool closed; - HttpCode code; - const char * file; - size_t name_size; - - // Read the http request - size_t len = options.socket.buflen; - char buffer[len]; - if( options.log ) sout | "=== Reading request ==="; - [code, closed, file, name_size] = http_read(fd, buffer, len); - - // if we are done, break out of the loop - if( closed ) break REQUEST; - - // If this wasn't a request retrun 400 - if( code != OK200 ) { - sout | "=== Invalid Request :" | code_val(code) | "==="; - answer_error(fd, code); - continue REQUEST; + size_t len = options.socket.buflen; + char buffer[len]; + handle_connection( this.conn, fd, buffer, len, 0p ); + + if( options.log ) sout | "=== Connection closed ==="; + } +} + + +//============================================================================================= +// Channel Worker Thread +//============================================================================================= +void ?{}( ChannelWorker & this ) { + size_t cli = rand() % options.clopts.cltr_cnt; + ((thread&)this){ "Server Worker Thread", *options.clopts.instance[cli], 64000 }; + options.clopts.thrd_cnt[cli]++; + this.done = false; +} + +void main( ChannelWorker & this ) { + park(); + /* paranoid */ assert( this.conn.pipe[0] != -1 ); + /* paranoid */ assert( this.conn.pipe[1] != -1 ); + for() { + size_t len = options.socket.buflen; + char buffer[len]; + PendingRead p; + p.in.buf = (void*)buffer; + p.in.len = len; + push(*this.queue, &p); + + if( options.log ) sout | "=== Waiting new connection ==="; + handle_connection( this.conn, p.out.fd, buffer, len, &p.f ); + + if( options.log ) sout | "=== Connection closed ==="; + if(this.done) break; + } +} + +extern "C" { +extern int accept4(int sockfd, struct sockaddr *addr, socklen_t *addrlen, int flags); +} + +void ?{}( Acceptor & this ) { + size_t cli = rand() % options.clopts.cltr_cnt; + ((thread&)this){ "Server Worker Thread", *options.clopts.instance[cli], 64000 }; + options.clopts.thrd_cnt[cli]++; + this.done = false; +} + +void main( Acceptor & this ) { + park(); + if( options.log ) sout | "=== Accepting connection ==="; + for() { + int fd = accept4(this.sockfd, this.[addr, addrlen, flags]); + if(fd < 0) { + if( errno == EWOULDBLOCK) { + yield(); + continue; } - - if(0 == strncmp(file, "plaintext", min(name_size, sizeof("plaintext") ))) { - if( options.log ) sout | "=== Request for /plaintext ==="; - - int ret = answer_plaintext(fd); - if( ret == -ECONNRESET ) break REQUEST; - - if( options.log ) sout | "=== Answer sent ==="; - continue REQUEST; - } - - if(0 == strncmp(file, "ping", min(name_size, sizeof("ping") ))) { - if( options.log ) sout | "=== Request for /ping ==="; - - // Send the header - int ret = answer_empty(fd); - if( ret == -ECONNRESET ) break REQUEST; - - if( options.log ) sout | "=== Answer sent ==="; - continue REQUEST; - } - - if( options.log ) { - sout | "=== Request for file " | nonl; - write(sout, file, name_size); - sout | " ==="; - } - - if( !options.file_cache.path ) { - if( options.log ) { - sout | "=== File Not Found (" | nonl; - write(sout, file, name_size); - sout | ") ==="; - } - answer_error(fd, E405); - continue REQUEST; - } - - // Get the fd from the file cache - int ans_fd; - size_t count; - [ans_fd, count] = get_file( file, name_size ); - - // If we can't find the file, return 404 - if( ans_fd < 0 ) { - if( options.log ) { - sout | "=== File Not Found (" | nonl; - write(sout, file, name_size); - sout | ") ==="; - } - answer_error(fd, E404); - continue REQUEST; - } - - // Send the desired file - int ret = answer_sendfile( this.pipe, fd, ans_fd, count, this.stats.sendfile ); - if( ret == -ECONNRESET ) break REQUEST; - - if( options.log ) sout | "=== Answer sent ==="; - } - - if( options.log ) sout | "=== Connection closed ==="; - continue CONNECTION; - } -} + if( errno == ECONNABORTED ) break; + if( this.done && (errno == EINVAL || errno == EBADF) ) break; + abort( "accept error: (%d) %s\n", (int)errno, strerror(errno) ); + } + if(this.done) return; + + if( options.log ) sout | "=== New connection" | fd | "" | ", waiting for requests ==="; + + if(fd) { + PendingRead * p = 0p; + for() { + if(this.done) return; + p = pop(*this.queue); + if(p) break; + yield(); + }; + + p->out.fd = fd; + async_recv(p->f, p->out.fd, p->in.buf, p->in.len, 0, CFA_IO_LAZY); + } + + if( options.log ) sout | "=== Accepting connection ==="; + } +} Index: benchmark/io/http/worker.hfa =================================================================== --- benchmark/io/http/worker.hfa (revision 6e2b04e05bb3d9f5059628fbcd9d7304e2577458) +++ benchmark/io/http/worker.hfa (revision 7f0ac120e5b984eb491cf8672d2c267d81986ebd) @@ -1,4 +1,6 @@ #pragma once +#include +#include #include @@ -26,6 +28,20 @@ }; -thread Worker { +void ?{}( sendfile_stats_t & this ); + +struct connection { int pipe[2]; + struct { + sendfile_stats_t sendfile; + } stats; +}; + +static inline void ?{}( connection & this ) { + this.pipe[0] = -1; + this.pipe[1] = -1; +} + +thread AcceptWorker { + connection conn; int sockfd; struct sockaddr * addr; @@ -33,8 +49,41 @@ int flags; volatile bool done; +}; +void ?{}( AcceptWorker & this); +void main( AcceptWorker & ); + + +struct PendingRead { + PendingRead * volatile next; + io_future_t f; struct { - sendfile_stats_t sendfile; - } stats; + void * buf; + size_t len; + } in; + struct { + volatile int fd; + } out; }; -void ?{}( Worker & this); -void main( Worker & ); + +static inline PendingRead * volatile & ?`next ( PendingRead * node ) { + return node->next; +} + +thread ChannelWorker { + connection conn; + volatile bool done; + mpsc_queue(PendingRead) * queue; +}; +void ?{}( ChannelWorker & ); +void main( ChannelWorker & ); + +thread Acceptor { + mpsc_queue(PendingRead) * queue; + int sockfd; + struct sockaddr * addr; + socklen_t * addrlen; + int flags; + volatile bool done; +}; +void ?{}( Acceptor & ); +void main( Acceptor & );