Changeset 137974ae

Timestamp:

Jun 9, 2022, 10:56:34 AM (2 years ago)

Author:

Thierry Delisle <tdelisle@…>

Branches:

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

Children:

430ce61

Parents:

8c58e73

Message:

Moved stats printer to it's own file and now using push-stats rather than pull

Location:

benchmark/io/http

Files:

• Makefile.am (modified) (1 diff)
• main.cfa (modified) (8 diffs)
• printer.cfa (added)
• printer.hfa (added)
• protocol.cfa (modified) (1 diff)
• worker.cfa (modified) (11 diffs)
• worker.hfa (modified) (2 diffs)

benchmark/io/http/Makefile.am

@@ -37 +37 @@
         options.cfa \
         options.hfa \
+        printer.cfa \
+        printer.hfa \
         protocol.cfa \
         protocol.hfa \

benchmark/io/http/main.cfa

@@ -25 +25 @@
 #include "options.hfa"
 #include "socket.hfa"
+#include "printer.hfa"
 #include "worker.hfa"
 
@@ -31 +32 @@
 Duration default_preemption() {
         return 0;
-}
-
-//=============================================================================================
-// Stats Printer
-//============================================================================================='
-
-thread StatsPrinter {
-        connection ** conns;
-        volatile int conn_cnt;
-        condition_variable(fast_block_lock) var;
-};
-
-void ?{}( StatsPrinter & this, cluster & cl ) {
-        ((thread&)this){ "Stats Printer Thread", cl };
-        this.conn_cnt = 0;
-}
-
-void ^?{}( StatsPrinter & mutex this ) {}
-
-#define eng3(X) (ws(3, 3, unit(eng( X ))))
-
-void main(StatsPrinter & this) {
-        LOOP: for() {
-                waitfor( ^?{} : this) {
-                        break LOOP;
-                }
-                or else {}
-
-                wait(this.var, 10`s);
-
-                print_stats_now( *active_cluster(), CFA_STATS_READY_Q | CFA_STATS_IO );
-                if(this.conn_cnt != 0) {
-                        uint64_t tries = 0;
-                        uint64_t calls = 0;
-                        uint64_t header = 0;
-                        uint64_t splcin = 0;
-                        uint64_t splcot = 0;
-                        struct {
-                                volatile uint64_t calls;
-                                volatile uint64_t bytes;
-                        } avgrd[zipf_cnts];
-                        memset(avgrd, 0, sizeof(avgrd));
-
-                        for(i; this.conn_cnt) {
-                                tries += this.conns[i]->stats.sendfile.tries;
-                                calls += this.conns[i]->stats.sendfile.calls;
-                                header += this.conns[i]->stats.sendfile.header;
-                                splcin += this.conns[i]->stats.sendfile.splcin;
-                                splcot += this.conns[i]->stats.sendfile.splcot;
-                                for(j; zipf_cnts) {
-                                        avgrd[j].calls += this.conns[i]->stats.sendfile.avgrd[j].calls;
-                                        avgrd[j].bytes += this.conns[i]->stats.sendfile.avgrd[j].bytes;
-                                }
-                        }
-
-                        double ratio = ((double)tries) / calls;
-
-                        sout | "----- Connection Stats -----";
-                        sout | "sendfile  : " | calls | "calls," | tries | "tries (" | ratio | " try/call)";
-                        sout | "            " | header | "header," | splcin | "splice in," | splcot | "splice out";
-                        sout | " - zipf sizes:";
-                        for(i; zipf_cnts) {
-                                double written = avgrd[i].calls > 0 ? ((double)avgrd[i].bytes) / avgrd[i].calls : 0;
-                                sout | "        " | zipf_sizes[i] | "bytes," | avgrd[i].calls | "shorts," | written | "written";
-                        }
-                }
-                else {
-                        sout | "No Connections!";
-                }
-        }
 }
 
@@ -109 +40 @@
         cluster self;
         processor    * procs;
-        // io_context   * ctxs;
-        StatsPrinter * prnt;
 
 };
@@ -152 +81 @@
         }
 
-        if(options.stats) {
-                this.prnt = alloc();
-                (*this.prnt){ this.self };
-        } else {
-                this.prnt = 0p;
-        }
-
         #if !defined(__CFA_NO_STATISTICS__)
                 print_stats_at_exit( this.self, CFA_STATS_READY_Q | CFA_STATS_IO );
@@ -167 +89 @@
 
 void ^?{}( ServerCluster & this ) {
-        delete(this.prnt);
-
         for(i; options.clopts.nprocs) {
                 ^(this.procs[i]){};
@@ -268 +188 @@
                         Q * queues = 0p;
                         ServerCluster cl;
+
+                        if(options.stats) {
+                                stats_thrd = alloc();
+                                (*stats_thrd){ cl.self };
+                        } else {
+                                stats_thrd = 0p;
+                        }
 
                         init_protocol();
@@ -319 +246 @@
                                         }
                                 }
-                                cl.prnt->conns = conns;
-                                cl.prnt->conn_cnt = options.clopts.nworkers;
+
                                 sout | options.clopts.nworkers | "workers started on" | options.clopts.nprocs | "processors";
                                 sout | nl;
@@ -436 +362 @@
 
                         sout | "Stopping printer threads..." | nonl; flush( sout );
-                        StatsPrinter * p = cl.prnt;
-                        if(p) {
-                                notify_one(p->var);
-                                join(*p);
-                        }
+                        if(stats_thrd) {
+                                notify_one(stats_thrd->var);
+                        }
+                        delete(stats_thrd);
                         sout | "done";
 

benchmark/io/http/protocol.cfa

@@ -587 +587 @@
 
 void ?{}( DateFormater & this ) {
-        ((thread&)this){ "Server Date Thread", *options.clopts.instance[0] };
+        ((thread&)this){ "Server Date Thread", *options.clopts.instance };
         this.idx = 0;
         memset( &this.buffers[0], 0, sizeof(this.buffers[0]) );

benchmark/io/http/worker.cfa

@@ -14 +14 @@
 #include "filecache.hfa"
 
-void ?{}( sendfile_stats_t & this ) {
-        this.calls = 0;
-        this.tries = 0;
-        this.header = 0;
-        this.splcin = 0;
-        this.splcot = 0;
-        for(i; zipf_cnts) {
-                this.avgrd[i].calls = 0;
-                this.avgrd[i].bytes = 0;
-        }
-}
-
 //=============================================================================================
 // Generic connection handling
 //=============================================================================================
-static void handle_connection( connection & this, volatile int & fd, char * buffer, size_t len, io_future_t * f ) {
+static void handle_connection( connection & this, volatile int & fd, char * buffer, size_t len, io_future_t * f, unsigned long long & last ) {
         REQUEST:
         for() {
@@ -111 +99 @@
                 if( options.log ) sout | "=== Answer sent ===";
         }
+
+        if (stats_thrd) {
+                unsigned long long next = rdtscl();
+                if(next > (last + 500000000)) {
+                        if(try_lock(stats_thrd->stats.lock)) {
+                                push(this.stats.sendfile, stats_thrd->stats.send);
+                                unlock(stats_thrd->stats.lock);
+                                last = next;
+                        }
+                }
+        }
 }
 
@@ -124 +123 @@
 void main( AcceptWorker & this ) {
         park();
+        unsigned long long last = rdtscl();
         /* paranoid */ assert( this.conn.pipe[0] != -1 );
         /* paranoid */ assert( this.conn.pipe[1] != -1 );
@@ -139 +139 @@
                 size_t len = options.socket.buflen;
                 char buffer[len];
-                handle_connection( this.conn, fd, buffer, len, 0p );
+                handle_connection( this.conn, fd, buffer, len, 0p, last );
 
                 if( options.log ) sout | "=== Connection closed ===";
@@ -157 +157 @@
 void main( ChannelWorker & this ) {
         park();
+        unsigned long long last = rdtscl();
         /* paranoid */ assert( this.conn.pipe[0] != -1 );
         /* paranoid */ assert( this.conn.pipe[1] != -1 );
@@ -168 +169 @@
 
                 if( options.log ) sout | "=== Waiting new connection ===";
-                handle_connection( this.conn, p.out.fd, buffer, len, &p.f );
+                handle_connection( this.conn, p.out.fd, buffer, len, &p.f, last );
 
                 if( options.log ) sout | "=== Connection closed ===";
@@ -187 +188 @@
 void main( Acceptor & this ) {
         park();
+        unsigned long long last = rdtscl();
         if( options.log ) sout | "=== Accepting connection ===";
         for() {
@@ -192 +194 @@
                 if(fd < 0) {
                         if( errno == EWOULDBLOCK) {
+                                this.stats.eagains++;
                                 yield();
                                 continue;
@@ -199 +202 @@
                         abort( "accept error: (%d) %s\n", (int)errno, strerror(errno) );
                 }
+                this.stats.accepts++;
+
                 if(this.done) return;
 
@@ -210 +215 @@
                                 if(p) break;
                                 yield();
+                                this.stats.creates++;
                         };
 
@@ -216 +222 @@
                 }
 
+                if (stats_thrd) {
+                        unsigned long long next = rdtscl();
+                        if(next > (last + 500000000)) {
+                                if(try_lock(stats_thrd->stats.lock)) {
+                                        push(this.stats, stats_thrd->stats.accpt);
+                                        unlock(stats_thrd->stats.lock);
+                                        last = next;
+                                }
+                        }
+                }
+
                 if( options.log ) sout | "=== Accepting connection ===";
         }

benchmark/io/http/worker.hfa

@@ -9 +9 @@
 }
 
+#include "printer.hfa"
+
 //=============================================================================================
 // Worker Thread
 //=============================================================================================
-
-extern const size_t zipf_sizes[];
-enum { zipf_cnts = 36, };
-
-struct sendfile_stats_t {
-        volatile uint64_t calls;
-        volatile uint64_t tries;
-        volatile uint64_t header;
-        volatile uint64_t splcin;
-        volatile uint64_t splcot;
-        struct {
-                volatile uint64_t calls;
-                volatile uint64_t bytes;
-        } avgrd[zipf_cnts];
-};
-
-void ?{}( sendfile_stats_t & this );
 
 struct connection {
@@ -85 +70 @@
         int flags;
         volatile bool done;
+        acceptor_stats_t stats;
 };
 void ?{}( Acceptor & );