Changeset 6e2b04e for benchmark

Timestamp:

Jun 8, 2022, 4:23:41 PM (4 years ago)

Author:

caparsons <caparson@…>

Branches:

ADT, ast-experimental, master, pthread-emulation, qualifiedEnum

Children:

7f0ac12, db7a3ad

Parents:

55422cf (diff), 720f2fe2 (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/http

Files:

• Makefile.am (modified) (2 diffs)
• main.cfa (modified) (15 diffs)
• options.cfa (modified) (2 diffs)
• options.hfa (modified) (1 diff)
• socket.cfa (added)
• socket.hfa (added)
• worker.cfa (modified) (1 diff)

benchmark/io/http/Makefile.am

@@ -21 +21 @@
 include $(top_srcdir)/tools/build/cfa.make
 
-AM_CFLAGS = -O3 -Wall -Wextra -I$(srcdir) -lrt -pthread # -Werror
+AM_CFLAGS = -O3 -Wall -Wextra -I$(srcdir) -lrt -pthread -g # -Werror
 AM_CFAFLAGS = -quiet -nodebug
 AM_LDFLAGS = -quiet -nodebug
@@ -39 +39 @@
         protocol.cfa \
         protocol.hfa \
+        socket.cfa \
+        socket.hfa \
         worker.cfa \
         worker.hfa

benchmark/io/http/main.cfa

@@ -2 +2 @@
 
 #include <errno.h>
+#include <signal.h>
 #include <stdio.h>
 #include <string.h>
@@ -8 +9 @@
         #include <sched.h>
         #include <signal.h>
+        #include <sys/eventfd.h>
         #include <sys/socket.h>
         #include <netinet/in.h>
@@ -14 +16 @@
 #include <fstream.hfa>
 #include <kernel.hfa>
+#include <locks.hfa>
 #include <iofwd.hfa>
 #include <stats.hfa>
@@ -21 +24 @@
 #include "filecache.hfa"
 #include "options.hfa"
+#include "socket.hfa"
 #include "worker.hfa"
 
@@ -36 +40 @@
         Worker * workers;
         int worker_cnt;
+        condition_variable(fast_block_lock) var;
 };
 
@@ -54 +59 @@
                 or else {}
 
-                sleep(10`s);
+                wait(this.var, 10`s);
 
                 print_stats_now( *active_cluster(), CFA_STATS_READY_Q | CFA_STATS_IO );
@@ -177 +182 @@
 
 //=============================================================================================
+// Termination
+//=============================================================================================
+
+int closefd;
+void cleanstop(int) {
+        eventfd_t buffer = 1;
+        char * buffer_s = (char*)&buffer;
+        int ret = write(closefd, buffer_s, sizeof(buffer));
+        if(ret < 0) abort( "eventfd write error: (%d) %s\n", (int)errno, strerror(errno) );
+        return;
+}
+
+//=============================================================================================
 // Main
 //============================================================================================='
 int main( int argc, char * argv[] ) {
+        int ret;
         __sighandler_t s = 1p;
         signal(SIGPIPE, s);
@@ -186 +205 @@
         // Parse args
         parse_options(argc, argv);
+
+        //===================
+        // Setup non-interactive termination
+        if(!options.interactive) {
+                closefd = eventfd(0, 0);
+                if(closefd < 0) abort( "eventfd error: (%d) %s\n", (int)errno, strerror(errno) );
+
+                sighandler_t prev = signal(SIGTERM, cleanstop);
+                intptr_t prev_workaround = (intptr_t) prev;
+                // can't use SIG_ERR it crashes the compiler
+                if(prev_workaround == -1) abort( "signal setup error: (%d) %s\n", (int)errno, strerror(errno) );
+
+                sout | "Signal termination ready";
+        }
 
         //===================
@@ -197 +230 @@
         // Open Socket
         sout | getpid() | ": Listening on port" | options.socket.port;
-        int server_fd = socket(AF_INET, SOCK_STREAM, 0);
-        if(server_fd < 0) {
-                abort( "socket error: (%d) %s\n", (int)errno, strerror(errno) );
-        }
-
-        int ret = 0;
+
         struct sockaddr_in address;
-        int addrlen = sizeof(address);
-        memset( (char *)&address, '\0' );
-        address.sin_family = AF_INET;
-        address.sin_addr.s_addr = htonl(INADDR_ANY);
-        address.sin_port = htons( options.socket.port );
-
-        int waited = 0;
-        for() {
-                int sockfd = server_fd;
-                __CONST_SOCKADDR_ARG addr;
-                addr.__sockaddr__ = (struct sockaddr *)&address;
-                socklen_t addrlen = sizeof(address);
-                ret = bind( sockfd, addr, addrlen );
-                if(ret < 0) {
-                        if(errno == EADDRINUSE) {
-                                if(waited == 0) {
-                                        if(!options.interactive) abort | "Port already in use in non-interactive mode. Aborting";
-                                        sout | "Waiting for port";
-                                } else {
-                                        sout | "\r" | waited | nonl;
-                                        flush( sout );
-                                }
-                                waited ++;
-                                sleep( 1`s );
-                                continue;
-                        }
-                        abort( "bind error: (%d) %s\n", (int)errno, strerror(errno) );
-                }
-                break;
-        }
-
-        ret = listen( server_fd, options.socket.backlog );
-        if(ret < 0) {
-                abort( "listen error: (%d) %s\n", (int)errno, strerror(errno) );
+        int addrlen = prepaddr(address);
+
+        int server_fd;
+        if(!options.socket.manyreuse) {
+                server_fd = listener(address, addrlen);
         }
 
@@ -257 +256 @@
 
                 {
+                        // Stats printer makes a copy so this needs to persist longer than normal
+                        Worker * workers;
                         ServerCluster cl[options.clopts.nclusters];
 
                         init_protocol();
                         {
-                                Worker * workers = anew(options.clopts.nworkers);
+                                workers = anew(options.clopts.nworkers);
                                 cl[0].prnt->workers = workers;
                                 cl[0].prnt->worker_cnt = options.clopts.nworkers;
@@ -273 +274 @@
                                                 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].sockfd  = options.socket.manyreuse ?  listener(address, addrlen) : server_fd;
                                                 workers[i].addr    = (struct sockaddr *)&address;
                                                 workers[i].addrlen = (socklen_t*)&addrlen;
@@ -285 +286 @@
                                 }
                                 sout | nl;
-                                if(!options.interactive) park();
                                 {
-                                        char buffer[128];
-                                        for() {
-                                                int ret = cfa_read(0, buffer, 128, 0);
-                                                if(ret == 0) break;
+                                        if(options.interactive) {
+                                                char buffer[128];
+                                                for() {
+                                                        int ret = cfa_read(0, buffer, 128, 0);
+                                                        if(ret == 0) break;
+                                                        if(ret < 0) abort( "main read error: (%d) %s\n", (int)errno, strerror(errno) );
+                                                        sout | "User wrote '" | "" | nonl;
+                                                        write(sout, buffer, ret - 1);
+                                                        sout | "'";
+                                                }
+                                        }
+                                        else {
+                                                char buffer[sizeof(eventfd_t)];
+                                                int ret = cfa_read(closefd, buffer, sizeof(eventfd_t), 0);
                                                 if(ret < 0) abort( "main read error: (%d) %s\n", (int)errno, strerror(errno) );
-                                                sout | "User wrote '" | "" | nonl;
-                                                write(sout, buffer, ret - 1);
-                                                sout | "'";
                                         }
 
@@ -307 +314 @@
 
                                 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(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) );
+                                        }
+                                }
+                                else {
+                                        ret = shutdown( server_fd, SHUT_RD );
+                                        if( ret < 0 ) {
+                                                abort( "shutdown error: (%d) %s\n", (int)errno, strerror(errno) );
+                                        }
                                 }
                                 sout | "done";
@@ -316 +331 @@
                                 // Close Socket
                                 sout | "Closing Socket..." | nonl; flush( sout );
-                                ret = close( server_fd );
-                                if(ret < 0) {
-                                        abort( "close socket error: (%d) %s\n", (int)errno, strerror(errno) );
+                                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 {
+                                        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 );
-                                adelete(workers);
+                                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";
@@ -331 +360 @@
                         sout | "done";
 
+                        sout | "Stopping printer threads..." | nonl; flush( sout );
+                        for(i; options.clopts.nclusters) {
+                                StatsPrinter * p = cl[i].prnt;
+                                if(p) {
+                                        notify_one(p->var);
+                                        join(*p);
+                                }
+                        }
+                        sout | "done";
+
+                        // Now that the stats printer is stopped, we can reclaim this
+                        adelete(workers);
+
                         sout | "Stopping processors/clusters..." | nonl; flush( sout );
                 }
                 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) );
-                        }
-                }
+                // 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";

benchmark/io/http/options.cfa

@@ -35 +35 @@
 
         { // socket
-                8080, // port
-                10,   // backlog
-                1024  // buflen
+                8080,  // port
+                10,    // backlog
+                1024,  // buflen
+                false, // onereuse
+                false  // manyreuse
         },
 
@@ -70 +72 @@
                 {'\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', "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 },

benchmark/io/http/options.hfa

@@ -27 +27 @@
                 int backlog;
                 int buflen;
+                bool onereuse;
+                bool manyreuse;
         } socket;
 

benchmark/io/http/worker.cfa

@@ -43 +43 @@
         /* paranoid */ assert( this.pipe[0] != -1 );
         /* paranoid */ assert( this.pipe[1] != -1 );
+
+        const bool reuse = options.socket.manyreuse;
 
         CONNECTION: