Changeset d95969a for benchmark

Timestamp:

Jan 25, 2021, 3:45:42 PM (5 years ago)

Author:

Thierry Delisle <tdelisle@…>

Branches:

ADT, arm-eh, ast-experimental, enum, forall-pointer-decay, jacob/cs343-translation, master, new-ast-unique-expr, pthread-emulation, qualifiedEnum

Children:

c292244

Parents:

b6a8b31 (diff), 7158202 (diff)
Note: this is a merge changeset, the changes displayed below correspond to the merge itself.
Use the (diff) links above to see all the changes relative to each parent.

Message:

Merge branch 'master' of plg.uwaterloo.ca:software/cfa/cfa-cc

Location:

benchmark/io

Files:

• http/filecache.cfa (modified) (4 diffs)
• http/http_ring.cpp (added)
• http/main.cfa (modified) (7 diffs)
• http/options.cfa (modified) (8 diffs)
• http/options.hfa (modified) (2 diffs)
• http/protocol.cfa (modified) (11 diffs)
• http/protocol.hfa (modified) (2 diffs)
• http/worker.cfa (modified) (6 diffs)
• setup.sh (added)

benchmark/io/http/filecache.cfa

@@ -4 +4 @@
 #include <string.h>
 
+#include <fstream.hfa>
 #include <stdlib.hfa>
 
@@ -182 +183 @@
                 conflicts += put_file( raw[i], fd );
         }
-        printf("Filled cache from path \"%s\" with %zu files\n", path, fcount);
+        sout | "Filled cache from path \"" | path | "\" with" | fcount | "files";
         if( conflicts > 0 ) {
-                printf("Found %d conflicts (seed: %u)\n", conflicts, options.file_cache.hash_seed);
+                sout | "Found" | conflicts | "conflicts (seed: " | options.file_cache.hash_seed | ")";
                 #if defined(REJECT_CONFLICTS)
                         abort("Conflicts found in the cache");
@@ -191 +192 @@
 
         if(options.file_cache.list) {
-                printf("Listing files and exiting\n");
+                sout | "Listing files and exiting";
                 for(i; fcount) {
                         int s; char u;
                         [s, u] = human_size(raw[i].size);
-                        printf("%4d%c - %s\n", s, u, raw[i].file);
+                        sout | s | u | "-" | raw[i].file;
                         free(raw[i].file);
                 }
@@ -208 +209 @@
 
 [int *, int] filefds(int extra) {
+        if(!options.file_cache.path) {
+                int * data = alloc(extra);
+                return [data, 0];
+        }
+
         if(!file_cache.entries) {
                 abort("File cache not filled!\n");

benchmark/io/http/main.cfa

@@ -6 +6 @@
 #include <unistd.h>
 extern "C" {
+        #include <signal.h>
         #include <sys/socket.h>
         #include <netinet/in.h>
 }
 
+#include <fstream.hfa>
 #include <kernel.hfa>
+#include <iofwd.hfa>
 #include <stats.hfa>
 #include <time.hfa>
@@ -50 +53 @@
 
 //=============================================================================================
+// Stats Printer
+//============================================================================================='
+
+thread StatsPrinter {};
+
+void ?{}( StatsPrinter & this ) {
+        ((thread&)this){ "Stats Printer Thread" };
+}
+
+void main(StatsPrinter & this) {
+        LOOP: for() {
+                waitfor( ^?{} : this) {
+                        break LOOP;
+                }
+                or else {}
+
+                sleep(10`s);
+
+                print_stats_now( *options.clopts.instance, CFA_STATS_READY_Q | CFA_STATS_IO );
+        }
+}
+
+//=============================================================================================
 // Main
 //============================================================================================='
 int main( int argc, char * argv[] ) {
+        __sighandler_t s = 1p;
+        signal(SIGPIPE, s);
+
         //===================
         // Parse args
-        const char * path = parse_options(argc, argv);
+        parse_options(argc, argv);
 
         //===================
         // Open Files
-        printf("Filling cache from %s\n", path);
-        fill_cache( path );
+        if( options.file_cache.path ) {
+                sout | "Filling cache from" | options.file_cache.path;
+                fill_cache( options.file_cache.path );
+        }
 
         //===================
         // Open Socket
-        printf("%ld : Listening on port %d\n", getpid(), options.socket.port);
+        sout | getpid() | ": Listening on port" | options.socket.port;
         int server_fd = socket(AF_INET, SOCK_STREAM, 0);
         if(server_fd < 0) {
@@ -84 +115 @@
                         if(errno == EADDRINUSE) {
                                 if(waited == 0) {
-                                        printf("Waiting for port\n");
+                                        sout | "Waiting for port";
                                 } else {
-                                        printf("\r%d", waited);
-                                        fflush(stdout);
+                                        sout | "\r" | waited | nonl;
+                                        flush( sout );
                                 }
                                 waited ++;
@@ -122 +153 @@
                 }
 
-                if(options.file_cache.fixed_fds) {
+                if(options.file_cache.path && options.file_cache.fixed_fds) {
                         register_fixed_files(cl, fds, pipe_off);
                 }
@@ -128 +159 @@
                 {
                         ServerProc procs[options.clopts.nprocs];
+                        StatsPrinter printer;
 
                         init_protocol();
@@ -148 +180 @@
                                         unpark( workers[i] );
                                 }
-                                printf("%d workers started on %d processors\n", options.clopts.nworkers, options.clopts.nprocs);
+                                sout | options.clopts.nworkers | "workers started on" | options.clopts.nprocs | "processors";
                                 {
                                         char buffer[128];
-                                        while(!feof(stdin)) {
-                                                fgets(buffer, 128, stdin);
+                                        while(int ret = cfa_read(0, buffer, 128, 0, -1`s, 0p, 0p); ret != 0) {
+                                                if(ret < 0) abort( "main read error: (%d) %s\n", (int)errno, strerror(errno) );
                                         }
 
-                                        printf("Shutting Down\n");
-                                }
-
+                                        sout | "Shutdown received";
+                                }
+
+                                sout | "Notifying connections..." | nonl; flush( sout );
                                 for(i; options.clopts.nworkers) {
-                                        printf("Cancelling %p\n", (void*)workers[i].cancel.target);
                                         workers[i].done = true;
                                         cancel(workers[i].cancel);
                                 }
-
-                                printf("Shutting down socket\n");
+                                sout | "done";
+
+                                sout | "Shutting down socket..." | nonl; flush( sout );
                                 int ret = shutdown( server_fd, SHUT_RD );
-                                if( ret < 0 ) { abort( "shutdown error: (%d) %s\n", (int)errno, strerror(errno) ); }
+                                if( ret < 0 ) {
+                                        abort( "shutdown error: (%d) %s\n", (int)errno, strerror(errno) );
+                                }
+                                sout | "done";
 
                                 //===================
                                 // Close Socket
-                                printf("Closing Socket\n");
+                                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 );
                         }
-                        printf("Workers Closed\n");
-
+                        sout | "done";
+
+                        sout | "Stopping protocol threads..." | nonl; flush( sout );
                         deinit_protocol();
-                }
-
+                        sout | "done";
+
+                        sout | "Stopping processors..." | nonl; flush( sout );
+                }
+                sout | "done";
+
+                sout | "Closing splice fds..." | nonl; flush( sout );
                 for(i; pipe_cnt) {
                         ret = close( fds[pipe_off + i] );
@@ -188 +233 @@
                 }
                 free(fds);
-
-        }
+                sout | "done";
+
+                sout | "Stopping processors..." | nonl; flush( sout );
+        }
+        sout | "done";
 
         //===================
         // Close Files
-        printf("Closing Files\n");
-        close_cache();
-}
+        if( options.file_cache.path ) {
+                sout | "Closing open files..." | nonl; flush( sout );
+                close_cache();
+                sout | "done";
+        }
+}

benchmark/io/http/options.cfa

@@ -9 +9 @@
 }
 
+#include <bitmanip.hfa>
+#include <fstream.hfa>
 #include <kernel.hfa>
 #include <parseargs.hfa>
 
+#include <stdlib.h>
 #include <string.h>
 
@@ -18 +21 @@
 
         { // file_cache
+                0,     // path
                 0,     // open_flags;
                 42u,   // hash_seed;
@@ -34 +38 @@
                 1,     // nprocs;
                 1,     // nworkers;
-                0,     // flags;
+                {},     // params;
                 false, // procstats
                 false, // viewhalts
@@ -41 +45 @@
 };
 
-const char * parse_options( int argc, char * argv[] ) {
+void parse_options( int argc, char * argv[] ) {
         bool subthrd = false;
         bool eagrsub = false;
@@ -48 +52 @@
         bool iokpoll = false;
         unsigned sublen = 16;
+        unsigned nentries = 16;
+
 
         static cfa_option opt[] = {
-                {'p', "port",           "Port the server will listen on", options.socket.port},
-                {'c', "cpus",           "Number of processors to use", options.clopts.nprocs},
-                {'L', "log",            "Enable logs", options.log, parse_settrue},
-                {'t', "threads",        "Number of worker threads to use", options.clopts.nworkers},
-                {'b', "accept-backlog", "Maximum number of pending accepts", options.socket.backlog},
-                {'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 },
-                {'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},
-                {'f', "fixed-fds",      "If set, files are open eagerly and pre-registered with the cluster", fixedfd, parse_settrue},
-                {'k', "kpollsubmit",    "If set, cluster uses IORING_SETUP_SQPOLL, implies -f", sqkpoll, parse_settrue },
-                {'i', "kpollcomplete",  "If set, cluster uses IORING_SETUP_IOPOLL", iokpoll, parse_settrue },
-                {'L', "submitlength",   "Max number of submitions that can be submitted together", sublen },
+                { 'p', "port",           "Port the server will listen on", options.socket.port},
+                { 'c', "cpus",           "Number of processors to use", options.clopts.nprocs},
+                { 't', "threads",        "Number of worker threads to use", options.clopts.nworkers},
+                {'\0', "log",            "Enable logs", options.log, parse_settrue},
+                {'\0', "accept-backlog", "Maximum number of pending accepts", options.socket.backlog},
+                {'\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 },
+                {'\0', "cache-size",     "Size of the cache to use, if set to small, will uses closes power of 2", options.file_cache.size },
+                {'\0', "list-files",     "List the files in the specified path and exit", options.file_cache.list, 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},
+                { 'f', "fixed-fds",      "If set, files are open eagerly and pre-registered with the cluster", fixedfd, parse_settrue},
+                { 'k', "kpollsubmit",    "If set, cluster uses IORING_SETUP_SQPOLL, implies -f", sqkpoll, parse_settrue },
+                { 'i', "kpollcomplete",  "If set, cluster uses IORING_SETUP_IOPOLL", iokpoll, parse_settrue },
+                {'\0', "submitlength",   "Max number of submitions that can be submitted together", sublen },
+                {'\0', "numentries",     "Number of I/O entries", nentries },
 
         };
@@ -71 +78 @@
         char **left;
         parse_args( argc, argv, opt, opt_cnt, "[OPTIONS]... [PATH]\ncforall http server", left );
+
+        if( !is_pow2(nentries) ) {
+                unsigned v = nentries;
+                v--;
+                v |= v >> 1;
+                v |= v >> 2;
+                v |= v >> 4;
+                v |= v >> 8;
+                v |= v >> 16;
+                v++;
+                serr | "Warning: num_entries not a power of 2" | '(' | nentries | ')' | "raising to " | v;
+                nentries = v;
+        }
+        options.clopts.params.num_entries = nentries;
 
         options.clopts.params.poller_submits = subthrd;
@@ -91 +112 @@
         options.clopts.params.num_ready = sublen;
 
-        if( left[0] == 0p ) { return "."; }
+        if( left[0] == 0p ) { return; }
 
         const char * path = left[0];
@@ -97 +118 @@
 
         if( left[0] != 0p ) {
-                abort("Too many trailing arguments!\n");
+                serr | "Too many trailing arguments!" | '\'' | path | '\'';
+                while(left[0] != 0p) {
+                        serr | " - " | left[0];
+                        left++;
+                }
+                exit(EXIT_FAILURE);
         }
 
-        return path;
+        options.file_cache.path = path;
 }

benchmark/io/http/options.hfa

@@ -11 +11 @@
 
         struct {
+                const char * path;
                 int open_flags;
                 uint32_t hash_seed;
@@ -36 +37 @@
 extern Options options;
 
-const char * parse_options( int argc, char * argv[] );
+void parse_options( int argc, char * argv[] );

benchmark/io/http/protocol.cfa

@@ -5 +5 @@
         #include <fcntl.h>
 }
+
+#include <fstream.hfa>
 #include <iofwd.hfa>
 
@@ -11 +13 @@
 extern "C" {
       int snprintf ( char * s, size_t n, const char * format, ... );
-        #include <linux/io_uring.h>
+        // #include <linux/io_uring.h>
 }
 #include <string.h>
@@ -24 +26 @@
         "HTTP/1.1 400 Bad Request\nServer: HttoForall\nDate: %s \nContent-Type: text/plain\nContent-Length: 0 \n\n",
         "HTTP/1.1 404 Not Found\nServer: HttoForall\nDate: %s \nContent-Type: text/plain\nContent-Length: 0 \n\n",
+        "HTTP/1.1 405 Method Not Allowed\nServer: HttoForall\nDate: %s \nContent-Type: text/plain\nContent-Length: 0 \n\n",
+        "HTTP/1.1 408 Request Timeout\nServer: HttoForall\nDate: %s \nContent-Type: text/plain\nContent-Length: 0 \n\n",
         "HTTP/1.1 413 Payload Too Large\nServer: HttoForall\nDate: %s \nContent-Type: text/plain\nContent-Length: 0 \n\n",
         "HTTP/1.1 414 URI Too Long\nServer: HttoForall\nDate: %s \nContent-Type: text/plain\nContent-Length: 0 \n\n",
@@ -34 +38 @@
         400,
         404,
+        405,
+        408,
         413,
         414,
@@ -49 +55 @@
                 int ret = cfa_write(fd, it, len, 0, -1`s, 0p, 0p);
                 // int ret = write(fd, it, len);
-                if( ret < 0 ) { if( errno != EAGAIN && errno != EWOULDBLOCK) abort( "'answer error' error: (%d) %s\n", (int)errno, strerror(errno) ); }
+                if( ret < 0 ) {
+                        if( errno == ECONNRESET || errno == EPIPE ) return -ECONNRESET;
+                        if( errno == EAGAIN || errno == EWOULDBLOCK) return -EAGAIN;
+
+                        abort( "'answer error' error: (%d) %s\n", (int)errno, strerror(errno) );
+                }
 
                 // update it/len
@@ -94 +105 @@
                 if(ret < 0 ) {
                         if( errno == EAGAIN || errno == EWOULDBLOCK) continue READ;
-                        // if( errno == EINVAL ) return [E400, true, 0, 0];
+                        if( errno == ECONNRESET ) return [E408, true, 0, 0];
+                        if( errno == EPIPE ) return [E408, true, 0, 0];
                         abort( "read error: (%d) %s\n", (int)errno, strerror(errno) );
                 }
@@ -108 +120 @@
         }
 
-        if( options.log ) printf("%.*s\n", rlen, buffer);
+        if( options.log ) {
+                write(sout, buffer, rlen);
+                sout | nl;
+        }
 
         it = buffer;
@@ -119 +134 @@
 }
 
-void sendfile( int pipe[2], int fd, int ans_fd, size_t count ) {
+int sendfile( int pipe[2], int fd, int ans_fd, size_t count ) {
         unsigned sflags = SPLICE_F_MOVE; // | SPLICE_F_MORE;
         off_t offset = 0;
@@ -128 +143 @@
                 if( ret < 0 ) {
                         if( errno != EAGAIN && errno != EWOULDBLOCK) continue SPLICE1;
+                        if( errno == ECONNRESET ) return -ECONNRESET;
+                        if( errno == EPIPE ) return -EPIPE;
                         abort( "splice [0] error: (%d) %s\n", (int)errno, strerror(errno) );
                 }
@@ -139 +156 @@
                         if( ret < 0 ) {
                                 if( errno != EAGAIN && errno != EWOULDBLOCK) continue SPLICE2;
+                                if( errno == ECONNRESET ) return -ECONNRESET;
+                                if( errno == EPIPE ) return -EPIPE;
                                 abort( "splice [1] error: (%d) %s\n", (int)errno, strerror(errno) );
                         }
@@ -145 +164 @@
 
         }
+        return count;
 }
 

benchmark/io/http/protocol.hfa

@@ -7 +7 @@
         E400,
         E404,
+        E405,
+        E408,
         E413,
         E414,
@@ -21 +23 @@
 [HttpCode code, bool closed, * const char file, size_t len] http_read(int fd, []char buffer, size_t len, io_cancellation *);
 
-void sendfile( int pipe[2], int fd, int ans_fd, size_t count );
+int sendfile( int pipe[2], int fd, int ans_fd, size_t count );

benchmark/io/http/worker.cfa

@@ -6 +6 @@
 #include <unistd.h>
 
+#include <fstream.hfa>
 #include <iofwd.hfa>
 
@@ -33 +34 @@
         CONNECTION:
         for() {
-                if( options.log ) printf("=== Accepting connection ===\n");
+                if( options.log ) sout | "=== Accepting connection ===";
                 int fd = cfa_accept4( this.[sockfd, addr, addrlen, flags], 0, -1`s, &this.cancel, 0p );
                 // int fd = accept4( this.[sockfd, addr, addrlen, flags] );
                 if(fd < 0) {
                         if( errno == ECONNABORTED ) break;
-                        if( errno == EINVAL && this.done ) break;
+                        if( this.done && (errno == EINVAL || errno == EBADF) ) break;
                         abort( "accept error: (%d) %s\n", (int)errno, strerror(errno) );
                 }
 
-                if( options.log ) printf("=== New connection %d, waiting for requests ===\n", fd);
+                if( options.log ) sout | "=== New connection" | fd | "" | ", waiting for requests ===";
                 REQUEST:
                 for() {
@@ -53 +54 @@
                         size_t len = options.socket.buflen;
                         char buffer[len];
-                        if( options.log ) printf("=== Reading request ===\n");
+                        if( options.log ) sout | "=== Reading request ===";
                         [code, closed, file, name_size] = http_read(fd, buffer, len, &this.cancel);
 
                         // if we are done, break out of the loop
-                        if( closed ) {
-                                if( options.log ) printf("=== Connection closed ===\n");
-                                close(fd);
-                                continue CONNECTION;
-                        }
+                        if( closed ) break REQUEST;
 
                         // If this wasn't a request retrun 400
                         if( code != OK200 ) {
-                                printf("=== Invalid Request : %d ===\n", code_val(code));
+                                sout | "=== Invalid Request :" | code_val(code) | "===";
                                 answer_error(fd, code);
                                 continue REQUEST;
@@ -71 +68 @@
 
                         if(0 == strncmp(file, "plaintext", min(name_size, sizeof("plaintext") ))) {
-                                if( options.log ) printf("=== Request for /plaintext ===\n");
+                                if( options.log ) sout | "=== Request for /plaintext ===";
 
                                 char text[] = "Hello, World!\n";
 
                                 // Send the header
-                                answer_plain(fd, text, sizeof(text));
+                                int ret = answer_plain(fd, text, sizeof(text));
+                                if( ret == -ECONNRESET ) break REQUEST;
 
-                                if( options.log ) printf("=== Answer sent ===\n");
+                                if( options.log ) sout | "=== Answer sent ===";
                                 continue REQUEST;
                         }
 
                         if(0 == strncmp(file, "ping", min(name_size, sizeof("ping") ))) {
-                                if( options.log ) printf("=== Request for /ping ===\n");
+                                if( options.log ) sout | "=== Request for /ping ===";
 
                                 // Send the header
-                                answer_empty(fd);
+                                int ret = answer_empty(fd);
+                                if( ret == -ECONNRESET ) break REQUEST;
 
-                                if( options.log ) printf("=== Answer sent ===\n");
+                                if( options.log ) sout | "=== Answer sent ===";
                                 continue REQUEST;
                         }
 
-                        if( options.log ) printf("=== Request for file %.*s ===\n", (int)name_size, file);
+                        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
@@ -101 +114 @@
                         // If we can't find the file, return 404
                         if( ans_fd < 0 ) {
-                                printf("=== File Not Found ===\n");
+                                if( options.log ) {
+                                        sout | "=== File Not Found (" | nonl;
+                                        write(sout, file, name_size);
+                                        sout | ") ===";
+                                }
                                 answer_error(fd, E404);
                                 continue REQUEST;
@@ -107 +124 @@
 
                         // Send the header
-                        answer_header(fd, count);
+                        int ret = answer_header(fd, count);
+                        if( ret == -ECONNRESET ) break REQUEST;
 
                         // Send the desired file
-                        sendfile( this.pipe, fd, ans_fd, count);
+                        ret = sendfile( this.pipe, fd, ans_fd, count);
+                        if( ret == -ECONNRESET ) break REQUEST;
 
-                        if( options.log ) printf("=== Answer sent ===\n");
+                        if( options.log ) sout | "=== Answer sent ===";
                 }
+
+                if( options.log ) sout | "=== Connection closed ===";
+                close(fd);
+                continue CONNECTION;
         }
 }