Changeset 8e4aa05 for benchmark/io/http/main.cfa

Timestamp:

Mar 4, 2021, 7:40:25 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:

77d601f

Parents:

342af53 (diff), a5040fe (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

@@ -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>
@@ -26 +29 @@
 
 //=============================================================================================
+// Stats Printer
+//============================================================================================='
+
+thread StatsPrinter {};
+
+void ?{}( StatsPrinter & this, cluster & cl ) {
+        ((thread&)this){ "Stats Printer Thread", cl };
+}
+
+void ^?{}( StatsPrinter & mutex this ) {}
+
+void main(StatsPrinter & this) {
+        LOOP: for() {
+                waitfor( ^?{} : this) {
+                        break LOOP;
+                }
+                or else {}
+
+                sleep(10`s);
+
+                print_stats_now( *active_cluster(), CFA_STATS_READY_Q | CFA_STATS_IO );
+        }
+}
+
+//=============================================================================================
 // Globals
 //=============================================================================================
-struct ServerProc {
-        processor self;
+struct ServerCluster {
+        cluster self;
+        processor    * procs;
+        // io_context   * ctxs;
+        StatsPrinter * prnt;
+
 };
 
-void ?{}( ServerProc & this ) {
-        /* paranoid */ assert( options.clopts.instance != 0p );
-        (this.self){ "Benchmark Processor", *options.clopts.instance };
+void ?{}( ServerCluster & this ) {
+        (this.self){ "Server Cluster", options.clopts.params };
+
+        this.procs = alloc(options.clopts.nprocs);
+        for(i; options.clopts.nprocs) {
+                (this.procs[i]){ "Benchmark Processor", this.self };
+
+                #if !defined(__CFA_NO_STATISTICS__)
+                        if( options.clopts.procstats ) {
+                                print_stats_at_exit( *this.procs, this.self.print_stats );
+                        }
+                        if( options.clopts.viewhalts ) {
+                                print_halts( *this.procs );
+                        }
+                #endif
+        }
+
+        if(options.stats) {
+                this.prnt = alloc();
+                (*this.prnt){ this.self };
+        } else {
+                this.prnt = 0p;
+        }
 
         #if !defined(__CFA_NO_STATISTICS__)
-                if( options.clopts.procstats ) {
-                        print_stats_at_exit( this.self, options.clopts.instance->print_stats );
-                }
-                if( options.clopts.viewhalts ) {
-                        print_halts( this.self );
-                }
+                print_stats_at_exit( this.self, CFA_STATS_READY_Q | CFA_STATS_IO );
         #endif
+
+        options.clopts.instance[options.clopts.cltr_cnt] = &this.self;
+        options.clopts.cltr_cnt++;
+}
+
+void ^?{}( ServerCluster & this ) {
+        delete(this.prnt);
+
+        for(i; options.clopts.nprocs) {
+                ^(this.procs[i]){};
+        }
+        free(this.procs);
+
+        ^(this.self){};
 }
 
@@ -53 +114 @@
 //============================================================================================='
 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 +150 @@
                         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 ++;
@@ -106 +172 @@
         // Run Server Cluster
         {
-                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;
-
-
                 int pipe_cnt = options.clopts.nworkers * 2;
                 int pipe_off;
@@ -122 +181 @@
                 }
 
-                if(options.file_cache.fixed_fds) {
-                        register_fixed_files(cl, fds, pipe_off);
-                }
+                // if(options.file_cache.path && options.file_cache.fixed_fds) {
+                //      register_fixed_files(cl, fds, pipe_off);
+                // }
 
                 {
-                        ServerProc procs[options.clopts.nprocs];
+                        ServerCluster cl[options.clopts.nclusters];
 
                         init_protocol();
@@ -148 +207 @@
                                         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 /" | options.clopts.nclusters | "clusters";
+                                for(i; options.clopts.nclusters) {
+                                        sout | options.clopts.thrd_cnt[i] | nonl;
+                                }
+                                sout | nl;
                                 {
                                         char buffer[128];
-                                        while(!feof(stdin)) {
-                                                fgets(buffer, 128, stdin);
+                                        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 | "'";
                                         }
 
-                                        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) );
                                 }
-                        }
-                        printf("Workers Closed\n");
-
+                                sout | "done";
+
+                                sout | "Stopping connection threads..." | nonl; flush( sout );
+                        }
+                        sout | "done";
+
+                        sout | "Stopping protocol threads..." | nonl; flush( sout );
                         deinit_protocol();
-                }
-
+                        sout | "done";
+
+                        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] );
@@ -188 +268 @@
                 }
                 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";
+        }
+}