Index: benchmark/io/http/filecache.cfa =================================================================== --- benchmark/io/http/filecache.cfa (revision 280182991c166aec83754649cbd34177b1061aa5) +++ benchmark/io/http/filecache.cfa (revision ce55a81847ece86766c279874bcc6dccb9aff995) @@ -73,4 +73,6 @@ cache_line * entries; size_t size; + int * rawfds; + int nfds; } file_cache; @@ -98,5 +100,5 @@ } -int put_file( cache_line & entry ) { +int put_file( cache_line & entry, int fd ) { uint32_t idx = murmur3_32( (const uint8_t *)entry.file, strlen(entry.file), options.file_cache.hash_seed ) % file_cache.size; @@ -108,4 +110,5 @@ file_cache.entries[idx] = entry; + file_cache.entries[idx].fd = fd; return i > 0 ? 1 : 0; } @@ -121,6 +124,5 @@ size_t fcount = 0; size_t fsize = 16; - cache_line * raw = 0p; - raw = alloc(raw, fsize, true); + cache_line * raw = alloc(fsize); // Step 1 get a dense array of all files int walk(const char *fpath, const struct stat *sb, int typeflag) { @@ -131,5 +133,5 @@ if(fcount > fsize) { fsize *= 2; - raw = alloc(raw, fsize, true); + raw = alloc(fsize, raw`realloc); } @@ -162,8 +164,21 @@ file_cache.entries = anew(file_cache.size); + if(options.file_cache.fixed_fds) { + file_cache.nfds = fcount; + file_cache.rawfds = alloc(fcount); + } + // Step 3 fill the cache int conflicts = 0; for(i; fcount) { - conflicts += put_file( raw[i] ); + int fd; + if(options.file_cache.fixed_fds) { + file_cache.rawfds[i] = raw[i].fd; + fd = i; + } + else { + fd = raw[i].fd; + } + conflicts += put_file( raw[i], fd ); } printf("Filled cache from path \"%s\" with %zu files\n", path, fcount); @@ -197,5 +212,7 @@ } - return [aalloc(extra), 0]; + size_t s = file_cache.nfds + extra; + int * data = alloc(s, file_cache.rawfds`realloc); + return [data, file_cache.nfds]; } Index: benchmark/io/http/main.cfa =================================================================== --- benchmark/io/http/main.cfa (revision 280182991c166aec83754649cbd34177b1061aa5) +++ benchmark/io/http/main.cfa (revision ce55a81847ece86766c279874bcc6dccb9aff995) @@ -12,16 +12,20 @@ #include #include +#include #include -#include "channel.hfa" #include "filecache.hfa" #include "options.hfa" #include "worker.hfa" +extern void register_fixed_files( cluster &, int *, unsigned count ); + +Duration default_preemption() { + return 0; +} + //============================================================================================= // Globals //============================================================================================= -channel & wait_connect; - struct ServerProc { processor self; @@ -84,12 +88,9 @@ // Run Server Cluster { - cluster cl = { "Server Cluster", options.clopts.flags }; + cluster cl = { "Server Cluster", options.clopts.params }; #if !defined(__CFA_NO_STATISTICS__) print_stats_at_exit( cl, CFA_STATS_READY_Q | CFA_STATS_IO ); #endif options.clopts.instance = &cl; - - channel chan = { options.clopts.chan_size }; - &wait_connect = &chan; int pipe_cnt = options.clopts.nworkers * 2; @@ -102,4 +103,8 @@ } + if(options.file_cache.fixed_fds) { + register_fixed_files(cl, fds, pipe_off); + } + { ServerProc procs[options.clopts.nprocs]; @@ -107,11 +112,16 @@ Worker workers[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 { + // 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 = server_fd; + workers[i].addr = (struct sockaddr *)&address; + workers[i].addrlen = (socklen_t*)&addrlen; + workers[i].flags = 0; } unpark( workers[i] __cfaabi_dbg_ctx2 ); @@ -119,6 +129,4 @@ printf("%d workers started on %d processors\n", options.clopts.nworkers, options.clopts.nprocs); { - Acceptor acceptor = { server_fd, (struct sockaddr *)&address, (socklen_t*)&addrlen, 0 }; - char buffer[128]; while(!feof(stdin)) { @@ -127,10 +135,4 @@ printf("Shutting Down\n"); - } - printf("Acceptor Closed\n"); - - // Clean-up the workers - for(options.clopts.nworkers) { - put( wait_connect, -1 ); } } Index: benchmark/io/http/options.cfa =================================================================== --- benchmark/io/http/options.cfa (revision 280182991c166aec83754649cbd34177b1061aa5) +++ benchmark/io/http/options.cfa (revision ce55a81847ece86766c279874bcc6dccb9aff995) @@ -31,5 +31,4 @@ 1, // nworkers; 0, // flags; - 10, // chan_size; false, // procstats false, // viewhalts @@ -39,5 +38,4 @@ const char * parse_options( int argc, char * argv[] ) { - bool uthrdpo = false; bool subthrd = false; bool eagrsub = false; @@ -52,10 +50,8 @@ {'t', "threads", "Number of worker threads to use", options.clopts.nworkers}, {'b', "accept-backlog", "Maximum number of pending accepts", options.socket.backlog}, - {'B', "channel-size", "Maximum number of accepted connection pending", options.clopts.chan_size}, {'r', "request_len", "Maximum number of bytes in the http request, requests with more data will be answered with Http Code 414", options.socket.buflen}, {'S', "seed", "seed to use for hashing", options.file_cache.hash_seed }, {'C', "cache-size", "Size of the cache to use, if set to small, will uses closes power of 2", options.file_cache.size }, {'l', "list-files", "List the files in the specified path and exit", options.file_cache.list, parse_settrue }, - {'u', "userthread", "If set, cluster uses user-thread to poll I/O", uthrdpo, parse_settrue }, {'s', "submitthread", "If set, cluster uses polling thread to submit I/O", subthrd, parse_settrue }, {'e', "eagersubmit", "If set, cluster submits I/O eagerly but still aggregates submits", eagrsub, parse_settrue}, @@ -71,15 +67,6 @@ parse_args( argc, argv, opt, opt_cnt, "[OPTIONS]... [PATH]\ncforall http server", left ); - if( uthrdpo ) { - options.clopts.flags |= CFA_CLUSTER_IO_POLLER_USER_THREAD; - } - - if( subthrd ) { - options.clopts.flags |= CFA_CLUSTER_IO_POLLER_THREAD_SUBMITS; - } - - if( eagrsub ) { - options.clopts.flags |= CFA_CLUSTER_IO_EAGER_SUBMITS; - } + options.clopts.params.poller_submits = subthrd; + options.clopts.params.eager_submits = eagrsub; if( fixedfd ) { @@ -88,14 +75,14 @@ if( sqkpoll ) { - options.clopts.flags |= CFA_CLUSTER_IO_KERNEL_POLL_SUBMITS; + options.clopts.params.poll_submit = true; options.file_cache.fixed_fds = true; } if( iokpoll ) { - options.clopts.flags |= CFA_CLUSTER_IO_KERNEL_POLL_COMPLETES; + options.clopts.params.poll_complete = true; options.file_cache.open_flags |= O_DIRECT; } - options.clopts.flags |= (sublen << CFA_CLUSTER_IO_BUFFLEN_OFFSET); + options.clopts.params.num_ready = sublen; if( left[0] == 0p ) { return "."; } Index: benchmark/io/http/options.hfa =================================================================== --- benchmark/io/http/options.hfa (revision 280182991c166aec83754649cbd34177b1061aa5) +++ benchmark/io/http/options.hfa (revision ce55a81847ece86766c279874bcc6dccb9aff995) @@ -2,4 +2,6 @@ #include + +#include struct cluster; @@ -23,6 +25,5 @@ int nprocs; int nworkers; - int flags; - int chan_size; + io_context_params params; bool procstats; bool viewhalts; Index: benchmark/io/http/protocol.cfa =================================================================== --- benchmark/io/http/protocol.cfa (revision 280182991c166aec83754649cbd34177b1061aa5) +++ benchmark/io/http/protocol.cfa (revision ce55a81847ece86766c279874bcc6dccb9aff995) @@ -11,9 +11,10 @@ extern "C" { int snprintf ( char * s, size_t n, const char * format, ... ); + #include } #include - #include +#include "options.hfa" const char * http_msgs[] = { @@ -74,6 +75,6 @@ READ: for() { - int ret = cfa_read(fd, it, count); - if(ret == 0 ) return [OK200, true, 0p, 0]; + int ret = cfa_read(fd, (void*)it, count, 0, -1`s, 0p, 0p); + if(ret == 0 ) return [OK200, true, 0, 0]; if(ret < 0 ) { if( errno == EAGAIN || errno == EWOULDBLOCK) continue READ; @@ -88,5 +89,5 @@ count -= ret; - if( count < 1 ) return [E414, false, 0p, 0]; + if( count < 1 ) return [E414, false, 0, 0]; } @@ -95,5 +96,5 @@ it = buffer; int ret = memcmp(it, "GET /", 5); - if( ret != 0 ) return [E400, false, 0p, 0]; + if( ret != 0 ) return [E400, false, 0, 0]; it += 5; @@ -106,5 +107,5 @@ ssize_t ret; SPLICE1: while(count > 0) { - ret = cfa_splice(ans_fd, &offset, pipe[1], 0p, count, SPLICE_F_MOVE | SPLICE_F_MORE); + ret = cfa_splice(ans_fd, &offset, pipe[1], 0p, count, SPLICE_F_MOVE | SPLICE_F_MORE, 0, -1`s, 0p, 0p); if( ret < 0 ) { if( errno != EAGAIN && errno != EWOULDBLOCK) continue SPLICE1; @@ -116,5 +117,5 @@ size_t in_pipe = ret; SPLICE2: while(in_pipe > 0) { - ret = cfa_splice(pipe[0], 0p, fd, 0p, in_pipe, SPLICE_F_MOVE | SPLICE_F_MORE); + ret = cfa_splice(pipe[0], 0p, fd, 0p, in_pipe, SPLICE_F_MOVE | SPLICE_F_MORE, 0, -1`s, 0p, 0p); if( ret < 0 ) { if( errno != EAGAIN && errno != EWOULDBLOCK) continue SPLICE2; Index: benchmark/io/http/worker.cfa =================================================================== --- benchmark/io/http/worker.cfa (revision 280182991c166aec83754649cbd34177b1061aa5) +++ benchmark/io/http/worker.cfa (revision ce55a81847ece86766c279874bcc6dccb9aff995) @@ -28,6 +28,9 @@ CONNECTION: for() { - int fd = take(wait_connect); - if (fd < 0) break; + int fd = cfa_accept4( this.[sockfd, addr, addrlen, flags], 0, -1`s, 0p, 0p ); + if(fd < 0) { + if( errno == ECONNABORTED ) break; + abort( "accept error: (%d) %s\n", (int)errno, strerror(errno) ); + } printf("New connection %d, waiting for requests\n", fd); @@ -82,26 +85,2 @@ } } - -//============================================================================================= -// Acceptor Thread -//============================================================================================= -void ?{}( Acceptor & this, int sockfd, struct sockaddr * addr, socklen_t * addrlen, int flags ) { - ((thread&)this){ "Acceptor Thread", *options.clopts.instance }; - this.sockfd = sockfd; - this.addr = addr; - this.addrlen = addrlen; - this.flags = flags; -} - -void main( Acceptor & this ) { - for() { - int ret = cfa_accept4( this.[sockfd, addr, addrlen, flags] ); - if(ret < 0) { - if( errno == ECONNABORTED ) break; - abort( "accept error: (%d) %s\n", (int)errno, strerror(errno) ); - } - - printf("New connection accepted\n"); - put( wait_connect, ret ); - } -} Index: benchmark/io/http/worker.hfa =================================================================== --- benchmark/io/http/worker.hfa (revision 280182991c166aec83754649cbd34177b1061aa5) +++ benchmark/io/http/worker.hfa (revision ce55a81847ece86766c279874bcc6dccb9aff995) @@ -7,8 +7,4 @@ } -#include "channel.hfa" - -extern channel & wait_connect; - //============================================================================================= // Worker Thread @@ -17,12 +13,4 @@ thread Worker { int pipe[2]; -}; -void ?{}( Worker & this ); -void main( Worker & ); - -//============================================================================================= -// Acceptor Thread -//============================================================================================= -thread Acceptor { int sockfd; struct sockaddr * addr; @@ -30,5 +18,4 @@ int flags; }; - -void ?{}( Acceptor & this, int sockfd, struct sockaddr * addr, socklen_t * addrlen, int flags ); -void main( Acceptor & this ); +void ?{}( Worker & this); +void main( Worker & );