Changeset 04b73b6 for benchmark/io/http/main.cfa

Timestamp:

Jul 23, 2020, 3:37:05 PM (6 years ago)

Author:

Thierry Delisle <tdelisle@…>

Branches:

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

Children:

f4ec4a90

Parents:

f0c3120 (diff), e262b5e (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

File:

• benchmark/io/http/main.cfa (modified) (8 diffs)

benchmark/io/http/main.cfa

@@ -17 +17 @@
 #include "filecache.hfa"
 #include "options.hfa"
-#include "parseargs.hfa"
 #include "worker.hfa"
 
@@ -23 +22 @@
 // Globals
 //=============================================================================================
-Options options @= {
-        0,
-        42u,
-        0,
-        false,
-        false,
-        0
-};
-
 channel & wait_connect;
 
@@ -39 +29 @@
 
 void ?{}( ServerProc & this ) {
-        /* paranoid */ assert( options.the_cluster != 0p );
-        (this.self){ "Benchmark Processor", *options.the_cluster };
+        /* paranoid */ assert( options.clopts.instance != 0p );
+        (this.self){ "Benchmark Processor", *options.clopts.instance };
 
         #if !defined(__CFA_NO_STATISTICS__)
-                if( options.procstats ) {
-                        print_stats_at_exit( this.self, options.the_cluster->print_stats );
+                if( options.clopts.procstats ) {
+                        print_stats_at_exit( this.self, options.clopts.instance->print_stats );
                 }
-                if( options.viewhalts ) {
+                if( options.clopts.viewhalts ) {
                         print_halts( this.self );
                 }
@@ -56 +46 @@
 //============================================================================================='
 int main( int argc, char * argv[] ) {
-        int port      = 8080;
-        int backlog   = 10;
-        int nprocs    = 1;
-        int nworkers  = 1;
-        int cl_flags  = 0;
-        int chan_size = 10;
-        const char * path = ".";
         //===================
         // Parse args
-        static cfa_option opt[] = {
-                {'p', "port", "Port the server will listen on", port},
-                {'c', "cpus", "Number of processors to use", nprocs},
-                {'t', "threads", "Number of worker threads to use", nworkers},
-                {'b', "accept-backlog", "Maximum number of pending accepts", backlog},
-                {'B', "channel-size", "Maximum number of accepted connection pending", chan_size}
-        };
-        int opt_cnt = sizeof(opt) / sizeof(cfa_option);
-
-        char **left;
-      parse_args( argc, argv, opt, opt_cnt, "[OPTIONS] [PATH]  -- cforall http server", left );
-
+        const char * path = parse_options(argc, argv);
 
         //===================
@@ -85 +57 @@
         //===================
         // Open Socket
-        printf("Listening on port %d\n", port);
+        printf("Listening on port %d\n", options.socket.port);
         int server_fd = socket(AF_INET, SOCK_STREAM, 0);
         if(server_fd < 0) {
@@ -97 +69 @@
         address.sin_family = AF_INET;
         address.sin_addr.s_addr = htonl(INADDR_ANY);
-        address.sin_port = htons( port );
+        address.sin_port = htons( options.socket.port );
 
         ret = bind( server_fd, (struct sockaddr *)&address, sizeof(address) );
@@ -104 +76 @@
         }
 
-        ret = listen( server_fd, backlog );
+        ret = listen( server_fd, options.socket.backlog );
         if(ret < 0) {
                 abort( "listen error: (%d) %s\n", (int)errno, strerror(errno) );
@@ -112 +84 @@
         // Run Server Cluster
         {
-                cluster cl = { "Server Cluster", cl_flags };
+                cluster cl = { "Server Cluster", options.clopts.flags };
                 #if !defined(__CFA_NO_STATISTICS__)
                         print_stats_at_exit( cl, CFA_STATS_READY_Q | CFA_STATS_IO );
                 #endif
-                options.the_cluster = &cl;
+                options.clopts.instance = &cl;
 
-                channel chan = { chan_size };
+                channel chan = { options.clopts.chan_size };
                 &wait_connect = &chan;
 
+                int pipe_cnt = options.clopts.nworkers * 2;
+                int pipe_off;
+                int * fds;
+                [fds, pipe_off] = filefds( pipe_cnt );
+                for(i; 0 ~ pipe_cnt ~ 2) {
+                        int ret = pipe(&fds[pipe_off + i]);
+                        if( ret < 0 ) { abort( "pipe error: (%d) %s\n", (int)errno, strerror(errno) ); }
+                }
+
                 {
-                        ServerProc procs[nprocs];
+                        ServerProc procs[options.clopts.nprocs];
                         {
-                                Worker workers[nworkers];
-                                printf("%d workers started on %d processors\n", nworkers, nprocs);
+                                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 {
+                                                workers[i].pipe[0] = fds[pipe_off + (i * 2) + 0];
+                                                workers[i].pipe[1] = fds[pipe_off + (i * 2) + 1];
+                                        }
+                                        unpark( workers[i] __cfaabi_dbg_ctx2 );
+                                }
+                                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)) {
+                                                fgets(buffer, 128, stdin);
+                                        }
+
+                                        printf("Shutting Down\n");
                                 }
-                                printf("Shutting Down\n");
+                                printf("Acceptor Closed\n");
 
                                 // Clean-up the workers
-                                for(nworkers) {
+                                for(options.clopts.nworkers) {
                                         put( wait_connect, -1 );
                                 }
                         }
+                        printf("Workers Closed\n");
                 }
+
+                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);
         }