Changeset d95969a for benchmark/io/http/main.cfa

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

File:

• benchmark/io/http/main.cfa (modified) (7 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>
@@ -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";
+        }
+}