Changes in / [1df492a:eb5962a]

Files:

• benchmark/io/http/Makefile.am (modified) (2 diffs)
• benchmark/io/http/main.cfa (modified) (16 diffs)
• benchmark/io/http/options.cfa (modified) (3 diffs)
• benchmark/io/http/options.hfa (modified) (1 diff)
• benchmark/io/http/printer.cfa (added)
• benchmark/io/http/printer.hfa (added)
• benchmark/io/http/protocol.cfa (modified) (5 diffs)
• benchmark/io/http/protocol.hfa (modified) (2 diffs)
• benchmark/io/http/socket.cfa (added)
• benchmark/io/http/socket.hfa (added)
• benchmark/io/http/worker.cfa (modified) (3 diffs)
• benchmark/io/http/worker.hfa (modified) (3 diffs)
• benchmark/io/sendfile/producer.cfa (added)
• libcfa/src/bits/locks.hfa (modified) (3 diffs)
• libcfa/src/concurrency/invoke.h (modified) (3 diffs)
• libcfa/src/concurrency/io.cfa (modified) (1 diff)
• libcfa/src/concurrency/io/setup.cfa (modified) (1 diff)
• libcfa/src/concurrency/iofwd.hfa (modified) (1 diff)
• libcfa/src/concurrency/kernel.cfa (modified) (1 diff)
• libcfa/src/concurrency/kernel/fwd.hfa (modified) (5 diffs)
• libcfa/src/concurrency/locks.cfa (modified) (2 diffs)
• libcfa/src/concurrency/locks.hfa (modified) (9 diffs)
• libcfa/src/concurrency/thread.cfa (modified) (2 diffs)
• libcfa/src/containers/queueLockFree.hfa (modified) (1 diff)
• libcfa/src/startup.cfa (modified) (1 diff)
• src/AST/Pass.impl.hpp (modified) (5 diffs)
• src/AST/TypeSubstitution.cpp (modified) (4 diffs)
• src/AST/TypeSubstitution.hpp (modified) (3 diffs)
• src/Concurrency/Waitfor.cc (modified) (1 diff)
• src/Concurrency/Waitfor.h (modified) (1 diff)
• src/Concurrency/WaitforNew.cpp (added)
• src/Concurrency/module.mk (modified) (1 diff)
• src/GenPoly/Specialize.cc (modified) (1 diff)
• src/InitTweak/FixGlobalInit.cc (modified) (1 diff)
• src/InitTweak/FixInit.cc (modified) (1 diff)
• src/InitTweak/InitTweak.cc (modified) (4 diffs)
• src/InitTweak/InitTweak.h (modified) (1 diff)
• src/main.cc (modified) (3 diffs)
• tests/concurrent/futures/.expect/wait_any.txt (added)
• tests/concurrent/futures/wait_any.cfa (added)
• tests/unified_locking/.expect/block_spin_lock.txt (added)
• tests/unified_locking/.expect/clh.txt (added)
• tests/unified_locking/.expect/mcs_block_spin_lock.txt (added)
• tests/unified_locking/.expect/mcs_spin.txt (added)
• tests/unified_locking/.expect/pthread_locks.txt (added)
• tests/unified_locking/.expect/simple_owner_lock.txt (added)
• tests/unified_locking/.expect/spin_queue_lock.txt (added)
• tests/unified_locking/block_spin_lock.cfa (added)
• tests/unified_locking/clh.cfa (added)
• tests/unified_locking/mcs_block_spin_lock.cfa (added)
• tests/unified_locking/mcs_spin.cfa (added)
• tests/unified_locking/mutex_test.hfa (modified) (4 diffs)
• tests/unified_locking/pthread_locks.cfa (added)
• tests/unified_locking/simple_owner_lock.cfa (added)
• tests/unified_locking/spin_queue_lock.cfa (added)
• tools/cfa.nanorc (modified) (1 diff)
• tools/jenkins/setup.sh.in (modified) (2 diffs)

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
@@ -37 +37 @@
         options.cfa \
         options.hfa \
+        printer.cfa \
+        printer.hfa \
         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 "printer.hfa"
 #include "worker.hfa"
 
@@ -30 +35 @@
 
 //=============================================================================================
-// Stats Printer
-//============================================================================================='
-
-thread StatsPrinter {
-        Worker * workers;
-        int worker_cnt;
-};
-
-void ?{}( StatsPrinter & this, cluster & cl ) {
-        ((thread&)this){ "Stats Printer Thread", cl };
-        this.worker_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 {}
-
-                sleep(10`s);
-
-                print_stats_now( *active_cluster(), CFA_STATS_READY_Q | CFA_STATS_IO );
-                if(this.worker_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.worker_cnt) {
-                                tries += this.workers[i].stats.sendfile.tries;
-                                calls += this.workers[i].stats.sendfile.calls;
-                                header += this.workers[i].stats.sendfile.header;
-                                splcin += this.workers[i].stats.sendfile.splcin;
-                                splcot += this.workers[i].stats.sendfile.splcot;
-                                for(j; zipf_cnts) {
-                                        avgrd[j].calls += this.workers[i].stats.sendfile.avgrd[j].calls;
-                                        avgrd[j].bytes += this.workers[i].stats.sendfile.avgrd[j].bytes;
-                                }
-                        }
-
-                        double ratio = ((double)tries) / calls;
-
-                        sout | "----- Worker 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 Workers!";
-                }
-        }
-}
-
-//=============================================================================================
 // Globals
 //=============================================================================================
-struct ServerCluster {
-        cluster self;
-        processor    * procs;
-        // io_context   * ctxs;
-        StatsPrinter * prnt;
-
-};
-
 void ?{}( ServerCluster & this ) {
         (this.self){ "Server Cluster", options.clopts.params };
@@ -122 +50 @@
                 (this.procs[i]){ "Benchmark Processor", this.self };
 
-                int c = 0;
-                int n = 1 + (i % cnt);
-                for(int j = 0; j < CPU_SETSIZE; j++) {
-                        if(CPU_ISSET(j, &fullset)) n--;
-                        if(n == 0) {
-                                c = j;
-                                break;
-                        }
-                }
-                cpu_set_t localset;
-                CPU_ZERO(&localset);
-                CPU_SET(c, &localset);
-                ret = pthread_setaffinity_np(this.procs[i].kernel_thread, sizeof(localset), &localset);
-                if( ret != 0 ) abort | "sched_getaffinity failed with" | ret | strerror( ret );
+                // int c = 0;
+                // int n = 1 + (i % cnt);
+                // for(int j = 0; j < CPU_SETSIZE; j++) {
+                //      if(CPU_ISSET(j, &fullset)) n--;
+                //      if(n == 0) {
+                //              c = j;
+                //              break;
+                //      }
+                // }
+                // cpu_set_t localset;
+                // CPU_ZERO(&localset);
+                // CPU_SET(c, &localset);
+                // ret = pthread_setaffinity_np(this.procs[i].kernel_thread, sizeof(localset), &localset);
+                // if( ret != 0 ) abort | "sched_getaffinity failed with" | ret | strerror( ret );
 
                 #if !defined(__CFA_NO_STATISTICS__)
@@ -147 +75 @@
         }
 
-        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 );
@@ -163 +84 @@
 
 void ^?{}( ServerCluster & this ) {
-        delete(this.prnt);
-
         for(i; options.clopts.nprocs) {
                 ^(this.procs[i]){};
@@ -175 +94 @@
 extern void init_protocol(void);
 extern void deinit_protocol(void);
+
+//=============================================================================================
+// REUSEPORT
+//=============================================================================================
+
+size_t sockarr_size;
+struct __attribute__((aligned(128))) Q {
+        mpsc_queue(PendingRead) q;
+};
+
+//=============================================================================================
+// 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;
+}
 
 //=============================================================================================
@@ -180 +121 @@
 //============================================================================================='
 int main( int argc, char * argv[] ) {
+        int ret;
         __sighandler_t s = 1p;
         signal(SIGPIPE, s);
@@ -186 +128 @@
         // 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 +153 @@
         // 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;
 
         //===================
@@ -257 +176 @@
 
                 {
+                        // Stats printer makes a copy so this needs to persist longer than normal
+                        connection ** conns;
+                        AcceptWorker  * aworkers = 0p;
+                        ChannelWorker * cworkers = 0p;
+                        Acceptor * acceptors = 0p;
+                        Q * queues = 0p;
                         ServerCluster cl[options.clopts.nclusters];
+
+                        if(options.stats) {
+                                stats_thrd = alloc();
+                                (*stats_thrd){ cl };
+                        } else {
+                                stats_thrd = 0p;
+                        }
 
                         init_protocol();
                         {
-                                Worker * workers = anew(options.clopts.nworkers);
-                                cl[0].prnt->workers = workers;
-                                cl[0].prnt->worker_cnt = 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];
-                                                workers[i].sockfd  = server_fd;
-                                                workers[i].addr    = (struct sockaddr *)&address;
-                                                workers[i].addrlen = (socklen_t*)&addrlen;
-                                                workers[i].flags   = 0;
-                                        }
-                                        unpark( workers[i] );
+                                int nacceptors = options.clopts.nprocs * options.clopts.nclusters;
+                                conns = alloc(options.clopts.nworkers);
+                                if(options.socket.reuseport) {
+                                        queues = alloc(nacceptors);
+                                        acceptors = alloc(nacceptors);
+                                        sout | "Creating" | nacceptors | "Acceptors";
+                                        for(i; nacceptors) {
+                                                (acceptors[i]){ i % options.clopts.nclusters };
+                                        }
+                                        for(i; nacceptors) {
+                                                (queues[i]){};
+                                                {
+                                                        acceptors[i].sockfd  = listener(address, addrlen);
+                                                        acceptors[i].addr    = (struct sockaddr *)&address;
+                                                        acceptors[i].addrlen = (socklen_t*)&addrlen;
+                                                        acceptors[i].flags   = 0;
+                                                        acceptors[i].queue   = &queues[i].q;
+                                                }
+                                                unpark( acceptors[i] );
+                                        }
+
+                                        cworkers = anew(options.clopts.nworkers);
+                                        for(i; options.clopts.nworkers) {
+                                                {
+                                                        cworkers[i].conn.pipe[0] = fds[pipe_off + (i * 2) + 0];
+                                                        cworkers[i].conn.pipe[1] = fds[pipe_off + (i * 2) + 1];
+                                                        cworkers[i].queue = &queues[i % nacceptors].q;
+                                                        conns[i] = &cworkers[i].conn;
+                                                }
+                                                unpark( cworkers[i] );
+                                        }
+                                }
+                                else {
+                                        server_fd = listener(address, addrlen);
+                                        aworkers = anew(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
+                                                {
+                                                        aworkers[i].conn.pipe[0] = fds[pipe_off + (i * 2) + 0];
+                                                        aworkers[i].conn.pipe[1] = fds[pipe_off + (i * 2) + 1];
+                                                        aworkers[i].sockfd = server_fd;
+                                                        aworkers[i].addr    = (struct sockaddr *)&address;
+                                                        aworkers[i].addrlen = (socklen_t*)&addrlen;
+                                                        aworkers[i].flags   = 0;
+                                                        conns[i] = &aworkers[i].conn;
+                                                }
+                                                unpark( aworkers[i] );
+                                        }
                                 }
                                 sout | options.clopts.nworkers | "workers started on" | options.clopts.nprocs | "processors /" | options.clopts.nclusters | "clusters";
@@ -285 +251 @@
                                 }
                                 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 | "'";
                                         }
 
@@ -300 +272 @@
                                 }
 
-                                sout | "Notifying connections..." | nonl; flush( sout );
-                                for(i; options.clopts.nworkers) {
-                                        workers[i].done = true;
-                                }
-                                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) );
-                                }
-                                sout | "done";
-
                                 //===================
-                                // 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) );
-                                }
-                                sout | "done";
-
-                                sout | "Stopping connection threads..." | nonl; flush( sout );
-                                adelete(workers);
+                                // Close Socket and join
+                                if(options.socket.reuseport) {
+                                        sout | "Notifying connections..." | nonl; flush( sout );
+                                        for(i; nacceptors) {
+                                                acceptors[i].done = true;
+                                        }
+                                        for(i; options.clopts.nworkers) {
+                                                cworkers[i].done = true;
+                                        }
+                                        sout | "done";
+
+                                        sout | "Shutting down Socket..." | nonl; flush( sout );
+                                        for(i; nacceptors) {
+                                                ret = shutdown( acceptors[i].sockfd, SHUT_RD );
+                                                if( ret < 0 ) {
+                                                        abort( "shutdown1 error: (%d) %s\n", (int)errno, strerror(errno) );
+                                                }
+                                        }
+                                        sout | "done";
+
+                                        sout | "Closing Socket..." | nonl; flush( sout );
+                                        for(i; nacceptors) {
+                                                ret = close( acceptors[i].sockfd );
+                                                if( ret < 0) {
+                                                        abort( "close socket error: (%d) %s\n", (int)errno, strerror(errno) );
+                                                }
+                                        }
+                                        sout | "done";
+
+                                        sout | "Stopping accept threads..." | nonl; flush( sout );
+                                        for(i; nacceptors) {
+                                                join(acceptors[i]);
+                                        }
+                                        sout | "done";
+
+                                        sout | "Draining worker queues..." | nonl; flush( sout );
+                                        for(i; nacceptors) {
+                                                PendingRead * p = 0p;
+                                                while(p = pop(queues[i].q)) {
+                                                        fulfil(p->f, -ECONNRESET);
+                                                }
+                                        }
+                                        sout | "done";
+
+                                        sout | "Stopping worker threads..." | nonl; flush( sout );
+                                        for(i; options.clopts.nworkers) {
+                                                for(j; 2) {
+                                                        ret = close(cworkers[i].conn.pipe[j]);
+                                                        if(ret < 0) abort( "close pipe %d error: (%d) %s\n", j, (int)errno, strerror(errno) );
+                                                }
+                                                join(cworkers[i]);
+                                        }
+                                }
+                                else {
+                                        sout | "Notifying connections..." | nonl; flush( sout );
+                                        for(i; options.clopts.nworkers) {
+                                                aworkers[i].done = true;
+                                        }
+                                        sout | "done";
+
+                                        sout | "Shutting down Socket..." | nonl; flush( sout );
+                                        ret = shutdown( server_fd, SHUT_RD );
+                                        if( ret < 0 ) {
+                                                abort( "shutdown2 error: (%d) %s\n", (int)errno, strerror(errno) );
+                                        }
+                                        sout | "done";
+
+                                        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 );
+                                        for(i; options.clopts.nworkers) {
+                                                for(j; 2) {
+                                                        ret = close(aworkers[i].conn.pipe[j]);
+                                                        if(ret < 0) abort( "close pipe %d error: (%d) %s\n", j, (int)errno, strerror(errno) );
+                                                }
+                                                join(aworkers[i]);
+                                        }
+                                }
                         }
                         sout | "done";
@@ -331 +363 @@
                         sout | "done";
 
+                        sout | "Stopping printer threads..." | nonl; flush( sout );
+                        if(stats_thrd) {
+                                notify_one(stats_thrd->var);
+                        }
+                        delete(stats_thrd);
+                        sout | "done";
+
+                        // Now that the stats printer is stopped, we can reclaim this
+                        adelete(aworkers);
+                        adelete(cworkers);
+                        adelete(acceptors);
+                        adelete(queues);
+                        free(conns);
+
                         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) );
-                        }
-                }
                 free(fds);
-                sout | "done";
 
                 sout | "Stopping processors..." | nonl; flush( sout );

benchmark/io/http/options.cfa

@@ -35 +35 @@
 
         { // socket
-                8080, // port
-                10,   // backlog
-                1024  // buflen
+                8080,  // port
+                10,    // backlog
+                1024,  // buflen
+                false  // reuseport
         },
 
@@ -52 +53 @@
 
 void parse_options( int argc, char * argv[] ) {
-        // bool fixedfd = false;
-        // bool sqkpoll = false;
-        // bool iokpoll = false;
         unsigned nentries = 0;
         bool isolate = false;
@@ -70 +68 @@
                 {'\0', "shell",          "Disable interactive mode", options.interactive, parse_setfalse},
                 {'\0', "accept-backlog", "Maximum number of pending accepts", options.socket.backlog},
+                {'\0', "reuseport",      "Use acceptor threads with reuse port SO_REUSEPORT", options.socket.reuseport, 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 reuseport;
         } socket;
 

benchmark/io/http/protocol.cfa

@@ -30 +30 @@
 #define PLAINTEXT_NOCOPY
 #define LINKED_IO
+
+static inline __s32 wait_res( io_future_t & this ) {
+        wait( this );
+        if( this.result < 0 ) {{
+                errno = -this.result;
+                return -1;
+        }}
+        return this.result;
+}
 
 struct https_msg_str {
@@ -470 +479 @@
 
                         if(is_error(splice_in.res)) {
+                                if(splice_in.res.error == -EPIPE) return -ECONNRESET;
                                 mutex(serr) serr | "SPLICE IN failed with" | splice_in.res.error;
                                 close(fd);
@@ -503 +513 @@
 }
 
-[HttpCode code, bool closed, * const char file, size_t len] http_read(int fd, []char buffer, size_t len) {
+[HttpCode code, bool closed, * const char file, size_t len] http_read(volatile int & fd, []char buffer, size_t len, io_future_t * f) {
         char * it = buffer;
         size_t count = len - 1;
@@ -509 +519 @@
         READ:
         for() {
-                int ret = cfa_recv(fd, (void*)it, count, 0, CFA_IO_LAZY);
+                int ret;
+                if( f ) {
+                        ret = wait_res(*f);
+                        reset(*f);
+                        f = 0p;
+                } else {
+                        ret = cfa_recv(fd, (void*)it, count, 0, CFA_IO_LAZY);
+                }
                 // int ret = read(fd, (void*)it, count);
                 if(ret == 0 ) return [OK200, true, 0, 0];
@@ -570 +587 @@
 
 void ?{}( DateFormater & this ) {
-        ((thread&)this){ "Server Date Thread", *options.clopts.instance[0] };
+        ((thread&)this){ "Server Date Thread" };
         this.idx = 0;
         memset( &this.buffers[0], 0, sizeof(this.buffers[0]) );

benchmark/io/http/protocol.hfa

@@ -1 +1 @@
 #pragma once
 
+struct io_future_t;
 struct sendfile_stats_t;
 
@@ -22 +23 @@
 int answer_sendfile( int pipe[2], int fd, int ans_fd, size_t count, struct sendfile_stats_t & );
 
-[HttpCode code, bool closed, * const char file, size_t len] http_read(int fd, []char buffer, size_t len);
+[HttpCode code, bool closed, * const char file, size_t len] http_read(volatile int & fd, []char buffer, size_t len, io_future_t * f);

benchmark/io/http/worker.cfa

@@ -8 +8 @@
 #include <fstream.hfa>
 #include <iofwd.hfa>
+#include <mutex_stmt.hfa>
 
 #include "options.hfa"
@@ -14 +15 @@
 
 //=============================================================================================
-// Worker Thread
-//=============================================================================================
-void ?{}( Worker & this ) {
+// Generic connection handling
+//=============================================================================================
+static void handle_connection( connection & this, volatile int & fd, char * buffer, size_t len, io_future_t * f, unsigned long long & last ) {
+        REQUEST:
+        for() {
+                bool closed;
+                HttpCode code;
+                const char * file;
+                size_t name_size;
+
+                // Read the http request
+                if( options.log ) mutex(sout) sout | "=== Reading request ===";
+                [code, closed, file, name_size] = http_read(fd, buffer, len, f);
+                f = 0p;
+
+                // if we are done, break out of the loop
+                if( closed ) break REQUEST;
+
+                // If this wasn't a request retrun 400
+                if( code != OK200 ) {
+                        sout | "=== Invalid Request :" | code_val(code) | "===";
+                        answer_error(fd, code);
+                        continue REQUEST;
+                }
+
+                if(0 == strncmp(file, "plaintext", min(name_size, sizeof("plaintext") ))) {
+                        if( options.log ) mutex(sout) sout | "=== Request for /plaintext ===";
+
+                        int ret = answer_plaintext(fd);
+                        if( ret == -ECONNRESET ) break REQUEST;
+
+                        if( options.log ) mutex(sout) sout | "=== Answer sent ===";
+                        continue REQUEST;
+                }
+
+                if(0 == strncmp(file, "ping", min(name_size, sizeof("ping") ))) {
+                        if( options.log ) mutex(sout) sout | "=== Request for /ping ===";
+
+                        // Send the header
+                        int ret = answer_empty(fd);
+                        if( ret == -ECONNRESET ) break REQUEST;
+
+                        if( options.log ) mutex(sout) sout | "=== Answer sent ===";
+                        continue REQUEST;
+                }
+
+                if( options.log ) {
+                        sout | "=== Request for file " | nonl;
+                        write(sout, file, name_size);
+                        sout | " ===";
+                }
+
+                if( !options.file_cache.path ) {
+                        if( options.log ) {
+                                sout | "=== File Not Found (" | nonl;
+                                write(sout, file, name_size);
+                                sout | ") ===";
+                        }
+                        answer_error(fd, E405);
+                        continue REQUEST;
+                }
+
+                // Get the fd from the file cache
+                int ans_fd;
+                size_t count;
+                [ans_fd, count] = get_file( file, name_size );
+
+                // If we can't find the file, return 404
+                if( ans_fd < 0 ) {
+                        if( options.log ) {
+                                sout | "=== File Not Found (" | nonl;
+                                write(sout, file, name_size);
+                                sout | ") ===";
+                        }
+                        answer_error(fd, E404);
+                        continue REQUEST;
+                }
+
+                // Send the desired file
+                int ret = answer_sendfile( this.pipe, fd, ans_fd, count, this.stats.sendfile );
+                if( ret == -ECONNRESET ) break REQUEST;
+
+                if( options.log ) mutex(sout) sout | "=== Answer sent ===";
+        }
+
+        if (stats_thrd) {
+                unsigned long long next = rdtscl();
+                if(next > (last + 500000000)) {
+                        if(try_lock(stats_thrd->stats.lock __cfaabi_dbg_ctx2)) {
+                                push(this.stats.sendfile, stats_thrd->stats.send);
+                                unlock(stats_thrd->stats.lock);
+                                last = next;
+                        }
+                }
+        }
+}
+
+//=============================================================================================
+// Self Accepting Worker Thread
+//=============================================================================================
+void ?{}( AcceptWorker & this ) {
         size_t cli = rand() % options.clopts.cltr_cnt;
         ((thread&)this){ "Server Worker Thread", *options.clopts.instance[cli], 64000 };
         options.clopts.thrd_cnt[cli]++;
-        this.pipe[0] = -1;
-        this.pipe[1] = -1;
         this.done = false;
-
-        this.stats.sendfile.calls = 0;
-        this.stats.sendfile.tries = 0;
-        this.stats.sendfile.header = 0;
-        this.stats.sendfile.splcin = 0;
-        this.stats.sendfile.splcot = 0;
-        for(i; zipf_cnts) {
-                this.stats.sendfile.avgrd[i].calls = 0;
-                this.stats.sendfile.avgrd[i].bytes = 0;
-        }
-}
-
-extern "C" {
-extern int accept4(int sockfd, struct sockaddr *addr, socklen_t *addrlen, int flags);
-}
-
-void main( Worker & this ) {
+}
+
+void main( AcceptWorker & this ) {
         park();
-        /* paranoid */ assert( this.pipe[0] != -1 );
-        /* paranoid */ assert( this.pipe[1] != -1 );
-
-        CONNECTION:
-        for() {
-                if( options.log ) sout | "=== Accepting connection ===";
-                int fd = cfa_accept4( this.[sockfd, addr, addrlen, flags], CFA_IO_LAZY );
+        unsigned long long last = rdtscl();
+        /* paranoid */ assert( this.conn.pipe[0] != -1 );
+        /* paranoid */ assert( this.conn.pipe[1] != -1 );
+        for() {
+                if( options.log ) mutex(sout) sout | "=== Accepting connection ===";
+                int fd = cfa_accept4( this.sockfd, this.[addr, addrlen, flags], CFA_IO_LAZY );
                 if(fd < 0) {
                         if( errno == ECONNABORTED ) break;
@@ -55 +137 @@
                 if(this.done) break;
 
+                if( options.log ) mutex(sout) sout | "=== New connection" | fd | "" | ", waiting for requests ===";
+                size_t len = options.socket.buflen;
+                char buffer[len];
+                handle_connection( this.conn, fd, buffer, len, 0p, last );
+
+                if( options.log ) mutex(sout) sout | "=== Connection closed ===";
+        }
+}
+
+
+//=============================================================================================
+// Channel Worker Thread
+//=============================================================================================
+void ?{}( ChannelWorker & this ) {
+        size_t cli = rand() % options.clopts.cltr_cnt;
+        ((thread&)this){ "Server Worker Thread", *options.clopts.instance[cli], 64000 };
+        options.clopts.thrd_cnt[cli]++;
+        this.done = false;
+}
+
+void main( ChannelWorker & this ) {
+        park();
+        unsigned long long last = rdtscl();
+        /* paranoid */ assert( this.conn.pipe[0] != -1 );
+        /* paranoid */ assert( this.conn.pipe[1] != -1 );
+        for() {
+                size_t len = options.socket.buflen;
+                char buffer[len];
+                PendingRead p;
+                p.next = 0p;
+                p.in.buf = (void*)buffer;
+                p.in.len = len;
+                push(*this.queue, &p);
+
+                if( options.log ) mutex(sout) sout | "=== Waiting new connection ===";
+                handle_connection( this.conn, p.out.fd, buffer, len, &p.f, last );
+
+                if( options.log ) mutex(sout) sout | "=== Connection closed ===";
+                if(this.done) break;
+        }
+}
+
+extern "C" {
+extern int accept4(int sockfd, struct sockaddr *addr, socklen_t *addrlen, int flags);
+}
+
+void ?{}( Acceptor & this, int cli ) {
+        ((thread&)this){ "Server Acceptor Thread", *options.clopts.instance[cli], 64000 };
+        options.clopts.thrd_cnt[cli]++;
+        this.done = false;
+}
+
+static inline __s32 get_res( io_future_t & this ) {
+        if( this.result < 0 ) {{
+                errno = -this.result;
+                return -1;
+        }}
+        return this.result;
+}
+
+static inline void push_connection( Acceptor & this, int fd ) {
+        PendingRead * p = 0p;
+        for() {
+                if(this.done) return;
+                p = pop(*this.queue);
+                if(p) break;
+                yield();
+                this.stats.creates++;
+        };
+
+        p->out.fd = fd;
+        async_recv(p->f, p->out.fd, p->in.buf, p->in.len, 0, CFA_IO_LAZY);
+}
+
+// #define ACCEPT_SPIN
+#define ACCEPT_MANY
+
+void main( Acceptor & this ) {
+        park();
+        unsigned long long last = rdtscl();
+
+#if defined(ACCEPT_SPIN)
+        if( options.log ) sout | "=== Accepting connection ===";
+        for() {
+                int fd = accept4(this.sockfd, this.[addr, addrlen, flags]);
+                if(fd < 0) {
+                        if( errno == EWOULDBLOCK) {
+                                this.stats.eagains++;
+                                yield();
+                                continue;
+                        }
+                        if( errno == ECONNABORTED ) break;
+                        if( this.done && (errno == EINVAL || errno == EBADF) ) break;
+                        abort( "accept error: (%d) %s\n", (int)errno, strerror(errno) );
+                }
+                this.stats.accepts++;
+
+                if(this.done) return;
+
                 if( options.log ) sout | "=== New connection" | fd | "" | ", waiting for requests ===";
-                REQUEST:
-                for() {
-                        bool closed;
-                        HttpCode code;
-                        const char * file;
-                        size_t name_size;
-
-                        // Read the http request
-                        size_t len = options.socket.buflen;
-                        char buffer[len];
-                        if( options.log ) sout | "=== Reading request ===";
-                        [code, closed, file, name_size] = http_read(fd, buffer, len);
-
-                        // if we are done, break out of the loop
-                        if( closed ) break REQUEST;
-
-                        // If this wasn't a request retrun 400
-                        if( code != OK200 ) {
-                                sout | "=== Invalid Request :" | code_val(code) | "===";
-                                answer_error(fd, code);
-                                continue REQUEST;
-                        }
-
-                        if(0 == strncmp(file, "plaintext", min(name_size, sizeof("plaintext") ))) {
-                                if( options.log ) sout | "=== Request for /plaintext ===";
-
-                                int ret = answer_plaintext(fd);
-                                if( ret == -ECONNRESET ) break REQUEST;
-
-                                if( options.log ) sout | "=== Answer sent ===";
-                                continue REQUEST;
-                        }
-
-                        if(0 == strncmp(file, "ping", min(name_size, sizeof("ping") ))) {
-                                if( options.log ) sout | "=== Request for /ping ===";
-
-                                // Send the header
-                                int ret = answer_empty(fd);
-                                if( ret == -ECONNRESET ) break REQUEST;
-
-                                if( options.log ) sout | "=== Answer sent ===";
-                                continue REQUEST;
-                        }
-
-                        if( options.log ) {
-                                sout | "=== Request for file " | nonl;
-                                write(sout, file, name_size);
-                                sout | " ===";
-                        }
-
-                        if( !options.file_cache.path ) {
-                                if( options.log ) {
-                                        sout | "=== File Not Found (" | nonl;
-                                        write(sout, file, name_size);
-                                        sout | ") ===";
+
+                if(fd) push_connection(this, fd);
+
+                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;
                                 }
-                                answer_error(fd, E405);
-                                continue REQUEST;
-                        }
-
-                        // Get the fd from the file cache
-                        int ans_fd;
-                        size_t count;
-                        [ans_fd, count] = get_file( file, name_size );
-
-                        // If we can't find the file, return 404
-                        if( ans_fd < 0 ) {
-                                if( options.log ) {
-                                        sout | "=== File Not Found (" | nonl;
-                                        write(sout, file, name_size);
-                                        sout | ") ===";
+                        }
+                }
+
+                if( options.log ) sout | "=== Accepting connection ===";
+        }
+
+#elif defined(ACCEPT_MANY)
+        const int nacc = 10;
+        io_future_t results[nacc];
+
+        for(i; nacc) {
+                io_future_t & res = results[i];
+                reset(res);
+                /* paranoid */ assert(!available(res));
+                if( options.log ) mutex(sout) sout | "=== Re-arming accept no" | i | " ===";
+                async_accept4(res, this.sockfd, this.[addr, addrlen, flags], CFA_IO_LAZY);
+        }
+
+        for() {
+                if (stats_thrd) {
+                        unsigned long long next = rdtscl();
+                        if(next > (last + 500000000)) {
+                                if(try_lock(stats_thrd->stats.lock __cfaabi_dbg_ctx2)) {
+                                        push(this.stats, stats_thrd->stats.accpt);
+                                        unlock(stats_thrd->stats.lock);
+                                        last = next;
                                 }
-                                answer_error(fd, E404);
-                                continue REQUEST;
-                        }
-
-                        // Send the desired file
-                        int ret = answer_sendfile( this.pipe, fd, ans_fd, count, this.stats.sendfile );
-                        if( ret == -ECONNRESET ) break REQUEST;
-
-                        if( options.log ) sout | "=== Answer sent ===";
-                }
-
-                if( options.log ) sout | "=== Connection closed ===";
-                continue CONNECTION;
-        }
-}
+                        }
+                }
+
+                for(i; nacc) {
+                        io_future_t & res = results[i];
+                        if(available(res)) {
+                                if( options.log ) mutex(sout) sout | "=== Accept no " | i | "completed with result" | res.result | "===";
+                                int fd = get_res(res);
+                                reset(res);
+                                this.stats.accepts++;
+                                if(fd < 0) {
+                                        if( errno == ECONNABORTED ) continue;
+                                        if( this.done && (errno == EINVAL || errno == EBADF) ) continue;
+                                        abort( "accept error: (%d) %s\n", (int)errno, strerror(errno) );
+                                }
+                                push_connection( this, fd );
+
+                                /* paranoid */ assert(!available(res));
+                                if( options.log ) mutex(sout) sout | "=== Re-arming accept no" | i | " ===";
+                                async_accept4(res, this.sockfd, this.[addr, addrlen, flags], CFA_IO_LAZY);
+                        }
+                }
+                if(this.done) return;
+
+                if( options.log ) mutex(sout) sout | "=== Waiting for any accept ===";
+                this.stats.eagains++;
+                wait_any(results, nacc);
+
+                if( options.log ) mutex(sout) {
+                        sout | "=== Acceptor wake-up ===";
+                        for(i; nacc) {
+                                io_future_t & res = results[i];
+                                sout | i | "available:" | available(res);
+                        }
+                }
+
+        }
+
+        for(i; nacc) {
+                wait(results[i]);
+        }
+#else
+#error no accept algorithm specified
+#endif
+}

benchmark/io/http/worker.hfa

@@ -1 +1 @@
 #pragma once
 
+#include <iofwd.hfa>
+#include <queueLockFree.hfa>
 #include <thread.hfa>
 
@@ -7 +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 connection {
+        int pipe[2];
         struct {
-                volatile uint64_t calls;
-                volatile uint64_t bytes;
-        } avgrd[zipf_cnts];
+                sendfile_stats_t sendfile;
+        } stats;
 };
 
-thread Worker {
-        int pipe[2];
+static inline void ?{}( connection & this ) {
+        this.pipe[0] = -1;
+        this.pipe[1] = -1;
+}
+
+thread AcceptWorker {
+        connection conn;
         int sockfd;
         struct sockaddr * addr;
@@ -33 +34 @@
         int flags;
         volatile bool done;
+};
+void ?{}( AcceptWorker & this);
+void main( AcceptWorker & );
+
+
+struct PendingRead {
+        PendingRead * volatile next;
+        io_future_t f;
         struct {
-                sendfile_stats_t sendfile;
-        } stats;
+                void * buf;
+                size_t len;
+        } in;
+        struct {
+                volatile int fd;
+        } out;
 };
-void ?{}( Worker & this);
-void main( Worker & );
+
+static inline PendingRead * volatile & ?`next ( PendingRead * node ) {
+        return node->next;
+}
+
+thread ChannelWorker {
+        connection conn;
+        volatile bool done;
+        mpsc_queue(PendingRead) * queue;
+};
+void ?{}( ChannelWorker & );
+void main( ChannelWorker & );
+
+thread Acceptor {
+        mpsc_queue(PendingRead) * queue;
+        int sockfd;
+        struct sockaddr * addr;
+        socklen_t * addrlen;
+        int flags;
+        volatile bool done;
+        acceptor_stats_t stats;
+};
+void ?{}( Acceptor &, int cli );
+void main( Acceptor & );

libcfa/src/bits/locks.hfa

@@ -26 +26 @@
         // Wrap in struct to prevent false sharing with debug info
         volatile bool lock;
-        #ifdef __CFA_DEBUG__
-                // previous function to acquire the lock
-                const char * prev_name;
-                // previous thread to acquire the lock
-                void* prev_thrd;
-                // keep track of number of times we had to spin, just in case the number is unexpectedly huge
-                size_t spin_count;
-        #endif
 };
 
@@ -40 +32 @@
                 extern void disable_interrupts() OPTIONAL_THREAD;
                 extern void enable_interrupts( bool poll = true ) OPTIONAL_THREAD;
-
-                #ifdef __CFA_DEBUG__
-                        void __cfaabi_dbg_record_lock(__spinlock_t & this, const char prev_name[]);
-                #else
-                        #define __cfaabi_dbg_record_lock(x, y)
-                #endif
+                #define __cfaabi_dbg_record_lock(x, y)
         }
 
         static inline void ?{}( __spinlock_t & this ) {
                 this.lock = 0;
-                #ifdef __CFA_DEBUG__
-                        this.spin_count = 0;
-                #endif
         }
 
@@ -77 +61 @@
                 for ( unsigned int i = 1;; i += 1 ) {
                         if ( (this.lock == 0) && (__atomic_test_and_set( &this.lock, __ATOMIC_ACQUIRE ) == 0) ) break;
-                        #ifdef __CFA_DEBUG__
-                                this.spin_count++;
-                        #endif
                         #ifndef NOEXPBACK
                                 // exponential spin

libcfa/src/concurrency/invoke.h

@@ -195 +195 @@
                 struct __monitor_group_t monitors;
 
-                // used to put threads on user data structures
-                struct {
-                        struct thread$ * next;
-                        struct thread$ * back;
-                } seqable;
-
                 // used to put threads on dlist data structure
                 __cfa_dlink(thread$);
@@ -208 +202 @@
                         struct thread$ * prev;
                 } node;
+
+                // used to store state between clh lock/unlock
+                volatile bool * clh_prev;
+
+                // used to point to this thd's current clh node
+                volatile bool * clh_node;
 
                 struct processor * last_proc;
@@ -240 +240 @@
                 }
 
-                static inline thread$ * volatile & ?`next ( thread$ * this )  __attribute__((const)) {
-                        return this->seqable.next;
-                }
-
-                static inline thread$ *& Back( thread$ * this ) __attribute__((const)) {
-                        return this->seqable.back;
-                }
-
-                static inline thread$ *& Next( thread$ * this ) __attribute__((const)) {
-                                return this->seqable.next;
-                }
-
-                static inline bool listed( thread$ * this ) {
-                        return this->seqable.next != 0p;
-                }
-
                 static inline void ?{}(__monitor_group_t & this) {
                         (this.data){0p};

libcfa/src/concurrency/io.cfa

@@ -159 +159 @@
 
                 const __u32 mask = *ctx->cq.mask;
+                const __u32 num  = *ctx->cq.num;
                 unsigned long long ts_prev = ctx->cq.ts;
-
-                // re-read the head and tail in case it already changed.
-                const __u32 head = *ctx->cq.head;
-                const __u32 tail = *ctx->cq.tail;
-                const __u32 count = tail - head;
-                __STATS__( false, io.calls.drain++; io.calls.completed += count; )
-
-                for(i; count) {
-                        unsigned idx = (head + i) & mask;
-                        volatile struct io_uring_cqe & cqe = ctx->cq.cqes[idx];
-
-                        /* paranoid */ verify(&cqe);
-
-                        struct io_future_t * future = (struct io_future_t *)(uintptr_t)cqe.user_data;
-                        // __cfadbg_print_safe( io, "Kernel I/O : Syscall completed : cqe %p, result %d for %p\n", &cqe, cqe.res, future );
-
-                        __kernel_unpark( fulfil( *future, cqe.res, false ), UNPARK_LOCAL );
-                }
-
-                unsigned long long ts_next = ctx->cq.ts = rdtscl();
-
-                // Mark to the kernel that the cqe has been seen
-                // Ensure that the kernel only sees the new value of the head index after the CQEs have been read.
-                __atomic_store_n( ctx->cq.head, head + count, __ATOMIC_SEQ_CST );
-                ctx->proc->idle_wctx.drain_time = ts_next;
+                unsigned long long ts_next;
+
+                // We might need to do this multiple times if more events completed than can fit in the queue.
+                for() {
+                        // re-read the head and tail in case it already changed.
+                        const __u32 head = *ctx->cq.head;
+                        const __u32 tail = *ctx->cq.tail;
+                        const __u32 count = tail - head;
+                        __STATS__( false, io.calls.drain++; io.calls.completed += count; )
+
+                        for(i; count) {
+                                unsigned idx = (head + i) & mask;
+                                volatile struct io_uring_cqe & cqe = ctx->cq.cqes[idx];
+
+                                /* paranoid */ verify(&cqe);
+
+                                struct io_future_t * future = (struct io_future_t *)(uintptr_t)cqe.user_data;
+                                // __cfadbg_print_safe( io, "Kernel I/O : Syscall completed : cqe %p, result %d for %p\n", &cqe, cqe.res, future );
+
+                                __kernel_unpark( fulfil( *future, cqe.res, false ), UNPARK_LOCAL );
+                        }
+
+                        ts_next = ctx->cq.ts = rdtscl();
+
+                        // Mark to the kernel that the cqe has been seen
+                        // Ensure that the kernel only sees the new value of the head index after the CQEs have been read.
+                        __atomic_store_n( ctx->cq.head, head + count, __ATOMIC_SEQ_CST );
+                        ctx->proc->idle_wctx.drain_time = ts_next;
+
+                        if(likely(count < num)) break;
+
+                        ioring_syscsll( *ctx, 0, IORING_ENTER_GETEVENTS);
+                }
 
                 __cfadbg_print_safe(io, "Kernel I/O : %u completed age %llu\n", count, ts_next);

libcfa/src/concurrency/io/setup.cfa

@@ -138 +138 @@
                 __u32 nentries = params_in.num_entries != 0 ? params_in.num_entries : 256;
                 if( !is_pow2(nentries) ) {
-                        abort("ERROR: I/O setup 'num_entries' must be a power of 2\n");
+                        abort("ERROR: I/O setup 'num_entries' must be a power of 2, was %u\n", nentries);
                 }
 

libcfa/src/concurrency/iofwd.hfa

@@ -76 +76 @@
         void reset    ( io_future_t & this ) { return reset    (this.self); }
         bool available( io_future_t & this ) { return available(this.self); }
+        bool setup    ( io_future_t & this, oneshot & ctx ) { return setup  (this.self, ctx); }
+        bool retract  ( io_future_t & this, oneshot & ctx ) { return retract(this.self, ctx); }
 }
 

libcfa/src/concurrency/kernel.cfa

@@ -834 +834 @@
 #endif
 
-
-
-//-----------------------------------------------------------------------------
-// Debug
-__cfaabi_dbg_debug_do(
-        extern "C" {
-                void __cfaabi_dbg_record_lock(__spinlock_t & this, const char prev_name[]) {
-                        this.prev_name = prev_name;
-                        this.prev_thrd = kernelTLS().this_thread;
-                }
-        }
-)
-
 //-----------------------------------------------------------------------------
 // Debug

libcfa/src/concurrency/kernel/fwd.hfa

@@ -200 +200 @@
                                         struct thread$ * expected = this.ptr;
                                         if(expected == 1p) return false;
-                                        /* paranoid */ verify( expected == 0p );
                                         if(__atomic_compare_exchange_n(&this.ptr, &expected, active_thread(), false, __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST)) {
                                                 park();
@@ -213 +212 @@
                         thread$ * post(oneshot & this, bool do_unpark = true) {
                                 struct thread$ * got = __atomic_exchange_n( &this.ptr, 1p, __ATOMIC_SEQ_CST);
-                                if( got == 0p ) return 0p;
+                                if( got == 0p || got == 1p ) return 0p;
                                 if(do_unpark) unpark( got );
                                 return got;
@@ -263 +262 @@
 
                                         // The future is not fulfilled, try to setup the wait context
-                                        /* paranoid */ verify( expected == 0p );
                                         if(__atomic_compare_exchange_n(&this.ptr, &expected, &wait_ctx, false, __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST)) {
                                                 return true;
@@ -275 +273 @@
                         // should retract the wait ctx
                         // intented to be use by wait, wait_any, waitfor, etc. rather than used directly
-                        void retract( future_t & this, oneshot & wait_ctx ) {
-                                // Remove the wait context
-                                struct oneshot * got = __atomic_exchange_n( &this.ptr, 0p, __ATOMIC_SEQ_CST);
-
-                                // got == 0p: future was never actually setup, just return
-                                if( got == 0p ) return;
-
-                                // got == wait_ctx: since fulfil does an atomic_swap,
-                                // if we got back the original then no one else saw context
-                                // It is safe to delete (which could happen after the return)
-                                if( got == &wait_ctx ) return;
-
-                                // got == 1p: the future is ready and the context was fully consumed
-                                // the server won't use the pointer again
-                                // It is safe to delete (which could happen after the return)
-                                if( got == 1p ) return;
-
-                                // got == 2p: the future is ready but the context hasn't fully been consumed
-                                // spin until it is safe to move on
-                                if( got == 2p ) {
-                                        while( this.ptr != 1p ) Pause();
-                                        return;
-                                }
-
-                                // got == any thing else, something wen't wrong here, abort
-                                abort("Future in unexpected state");
+                        bool retract( future_t & this, oneshot & wait_ctx ) {
+                                for() {
+                                        struct oneshot * expected = this.ptr;
+
+                                        // expected == 0p: future was never actually setup, just return
+                                        if( expected == 0p ) return false;
+
+                                        // expected == 1p: the future is ready and the context was fully consumed
+                                        // the server won't use the pointer again
+                                        // It is safe to delete (which could happen after the return)
+                                        if( expected == 1p ) return true;
+
+                                        // expected == 2p: the future is ready but the context hasn't fully been consumed
+                                        // spin until it is safe to move on
+                                        if( expected == 2p ) {
+                                                while( this.ptr != 1p ) Pause();
+                                                /* paranoid */ verify( this.ptr == 1p );
+                                                return true;
+                                        }
+
+                                        // expected != wait_ctx: the future was setup with a different context ?!?!
+                                        // something went wrong here, abort
+                                        if( expected != &wait_ctx ) abort("Future in unexpected state");
+
+                                        // we still have the original context, then no one else saw it
+                                        // attempt to remove the context so it doesn't get consumed.
+                                        if(__atomic_compare_exchange_n( &this.ptr, &expected, 0p, false, __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST)) {
+                                                return false;
+                                        }
+                                }
                         }
 
@@ -379 +381 @@
                                 return ret;
                         }
+
+                        // Wait for any future to be fulfilled
+                        forall(T& | sized(T) | { bool setup( T&, oneshot & ); bool retract( T&, oneshot & ); })
+                        T & wait_any( T * futures, size_t num_futures ) {
+                                oneshot temp;
+
+                                // setup all futures
+                                // if any are already satisfied return
+                                for ( i; num_futures ) {
+                                        if( !setup(futures[i], temp) ) return futures[i];
+                                }
+
+                                // Wait context is setup, just wait on it
+                                wait( temp );
+
+                                size_t ret;
+                                // attempt to retract all futures
+                                for ( i; num_futures ) {
+                                        if ( retract( futures[i], temp ) ) ret = i;
+                                }
+
+                                return futures[ret];
+                        }
                 }
 

libcfa/src/concurrency/locks.cfa

@@ -219 +219 @@
         // this casts the alarm node to our wrapped type since we used type erasure
         static void alarm_node_wrap_cast( alarm_node_t & a ) { timeout_handler( (alarm_node_wrap(L) &)a ); }
+
+        struct pthread_alarm_node_wrap {
+                alarm_node_t alarm_node;
+                pthread_cond_var(L) * cond;
+                info_thread(L) * info_thd;
+        };
+
+        void ?{}( pthread_alarm_node_wrap(L) & this, Duration alarm, Duration period, Alarm_Callback callback, pthread_cond_var(L) * c, info_thread(L) * i ) {
+                this.alarm_node{ callback, alarm, period };
+                this.cond = c;
+                this.info_thd = i;
+        }
+
+        void ^?{}( pthread_alarm_node_wrap(L) & this ) { }
+
+        static void timeout_handler ( pthread_alarm_node_wrap(L) & this ) with( this ) {
+                // This pthread_cond_var member is called from the kernel, and therefore, cannot block, but it can spin.
+                lock( cond->lock __cfaabi_dbg_ctx2 );
+
+                // this check is necessary to avoid a race condition since this timeout handler
+                //      may still be called after a thread has been removed from the queue but
+                //      before the alarm is unregistered
+                if ( (*info_thd)`isListed ) {   // is thread on queue
+                        info_thd->signalled = false;
+                        // remove this thread O(1)
+                        remove( *info_thd );
+                        on_notify(*info_thd->lock, info_thd->t);
+                }
+                unlock( cond->lock );
+        }
+
+        // this casts the alarm node to our wrapped type since we used type erasure
+        static void pthread_alarm_node_wrap_cast( alarm_node_t & a ) { timeout_handler( (pthread_alarm_node_wrap(L) &)a ); }
 }
 
@@ -388 +421 @@
                 on_wakeup(*i.lock, recursion_count);
         }
-}
-
+
+        //-----------------------------------------------------------------------------
+        // pthread_cond_var
+
+        void  ?{}( pthread_cond_var(L) & this ) with(this) {
+                blocked_threads{};
+                lock{};
+        }
+
+        void ^?{}( pthread_cond_var(L) & this ) { }
+
+        bool notify_one( pthread_cond_var(L) & this ) with(this) { 
+                lock( lock __cfaabi_dbg_ctx2 );
+                bool ret = ! blocked_threads`isEmpty;
+                if ( ret ) {
+                        info_thread(L) & popped = try_pop_front( blocked_threads );
+                        on_notify(*popped.lock, popped.t);
+                }
+                unlock( lock );
+                return ret;
+        }
+
+        bool notify_all( pthread_cond_var(L) & this ) with(this) { 
+                lock( lock __cfaabi_dbg_ctx2 );
+                bool ret = ! blocked_threads`isEmpty;
+                while( ! blocked_threads`isEmpty ) {
+                        info_thread(L) & popped = try_pop_front( blocked_threads );
+                        on_notify(*popped.lock, popped.t);
+                }
+                unlock( lock );
+                return ret;
+        }
+
+        uintptr_t front( pthread_cond_var(L) & this ) with(this) { return blocked_threads`isEmpty ? NULL : blocked_threads`first.info; }
+        bool empty ( pthread_cond_var(L) & this ) with(this) { return blocked_threads`isEmpty; }
+
+        static size_t queue_and_get_recursion( pthread_cond_var(L) & this, info_thread(L) * i ) with(this) {
+                // add info_thread to waiting queue
+                insert_last( blocked_threads, *i );
+                size_t recursion_count = 0;
+                recursion_count = on_wait( *i->lock );
+                return recursion_count;
+        }
+        
+        static void queue_info_thread_timeout( pthread_cond_var(L) & this, info_thread(L) & info, Duration t, Alarm_Callback callback ) with(this) {
+                lock( lock __cfaabi_dbg_ctx2 );
+                size_t recursion_count = queue_and_get_recursion(this, &info);
+                pthread_alarm_node_wrap(L) node_wrap = { t, 0`s, callback, &this, &info };
+                register_self( &node_wrap.alarm_node );
+                unlock( lock );
+
+                // blocks here
+                park();
+
+                // unregisters alarm so it doesn't go off if this happens first
+                unregister_self( &node_wrap.alarm_node );
+
+                // resets recursion count here after waking
+                if (info.lock) on_wakeup(*info.lock, recursion_count);
+        }
+
+        void wait( pthread_cond_var(L) & this, L & l ) with(this) {
+                wait( this, l, 0 );
+        }
+
+        void wait( pthread_cond_var(L) & this, L & l, uintptr_t info ) with(this) {
+                lock( lock __cfaabi_dbg_ctx2 );
+                info_thread( L ) i = { active_thread(), info, &l };
+                size_t recursion_count = queue_and_get_recursion(this, &i);
+                unlock( lock );
+                park( );
+                on_wakeup(*i.lock, recursion_count);
+        }
+
+        #define PTHREAD_WAIT_TIME( u, l, t ) \
+                info_thread( L ) i = { active_thread(), u, l }; \
+                queue_info_thread_timeout(this, i, t, pthread_alarm_node_wrap_cast ); \
+                return i.signalled;
+
+        bool wait( pthread_cond_var(L) & this, L & l, timespec t ) {
+                Duration d = { t };
+                WAIT_TIME( 0, &l , d )
+        }
+        
+        bool wait( pthread_cond_var(L) & this, L & l, uintptr_t info, timespec t  ) {
+                Duration d = { t };
+                WAIT_TIME( info, &l , d )
+        }
+}
 //-----------------------------------------------------------------------------
 // Semaphore

libcfa/src/concurrency/locks.hfa

@@ -101 +101 @@
 
 //-----------------------------------------------------------------------------
+// MCS Spin Lock
+// - No recursive acquisition
+// - Needs to be released by owner
+
+struct mcs_spin_node {
+        mcs_spin_node * volatile next;
+        volatile bool locked;
+};
+
+struct mcs_spin_queue {
+        mcs_spin_node * volatile tail;
+};
+
+static inline void ?{}(mcs_spin_node & this) { this.next = 0p; this.locked = true; }
+
+static inline mcs_spin_node * volatile & ?`next ( mcs_spin_node * node ) {
+        return node->next;
+}
+
+struct mcs_spin_lock {
+        mcs_spin_queue queue;
+};
+
+static inline void lock(mcs_spin_lock & l, mcs_spin_node & n) {
+        mcs_spin_node * prev = __atomic_exchange_n(&l.queue.tail, &n, __ATOMIC_SEQ_CST);
+        n.locked = true;
+        if(prev == 0p) return;
+        prev->next = &n;
+        while(__atomic_load_n(&n.locked, __ATOMIC_RELAXED)) Pause();
+}
+
+static inline void unlock(mcs_spin_lock & l, mcs_spin_node & n) {
+        mcs_spin_node * n_ptr = &n;
+        if (__atomic_compare_exchange_n(&l.queue.tail, &n_ptr, 0p, false, __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST)) return;
+        while (__atomic_load_n(&n.next, __ATOMIC_RELAXED) == 0p) {}
+        n.next->locked = false;
+}
+
+//-----------------------------------------------------------------------------
+// CLH Spinlock
+// - No recursive acquisition
+// - Needs to be released by owner
+
+struct clh_lock {
+        volatile bool * volatile tail;
+};
+
+static inline void  ?{}( clh_lock & this ) { this.tail = malloc(); *this.tail = true; }
+static inline void ^?{}( clh_lock & this ) { free(this.tail); }
+
+static inline void lock(clh_lock & l) {
+        thread$ * curr_thd = active_thread();
+        *(curr_thd->clh_node) = false;
+        volatile bool * prev = __atomic_exchange_n((bool **)(&l.tail), (bool *)(curr_thd->clh_node), __ATOMIC_SEQ_CST);
+        while(!__atomic_load_n(prev, __ATOMIC_ACQUIRE)) Pause();
+        curr_thd->clh_prev = prev;
+}
+
+static inline void unlock(clh_lock & l) {
+        thread$ * curr_thd = active_thread();
+        __atomic_store_n(curr_thd->clh_node, true, __ATOMIC_RELEASE);
+        curr_thd->clh_node = curr_thd->clh_prev;
+}
+
+//-----------------------------------------------------------------------------
 // Linear backoff Spinlock
 struct linear_backoff_then_block_lock {
@@ -205 +270 @@
 // Fast Block Lock
 
-// High efficiency minimal blocking lock
+// minimal blocking lock
 // - No reacquire for cond var
 // - No recursive acquisition
 // - No ownership
 struct fast_block_lock {
+        // List of blocked threads
+        dlist( thread$ ) blocked_threads;
+
         // Spin lock used for mutual exclusion
         __spinlock_t lock;
 
-        // List of blocked threads
-        dlist( thread$ ) blocked_threads;
-
+        // flag showing if lock is held
         bool held:1;
+
+        #ifdef __CFA_DEBUG__
+        // for deadlock detection
+        struct thread$ * owner;
+        #endif
 };
 
@@ -231 +302 @@
 static inline void lock(fast_block_lock & this) with(this) {
         lock( lock __cfaabi_dbg_ctx2 );
+
+        #ifdef __CFA_DEBUG__
+        assert(!(held && owner == active_thread()));
+        #endif
         if (held) {
                 insert_last( blocked_threads, *active_thread() );
@@ -238 +313 @@
         }
         held = true;
+        #ifdef __CFA_DEBUG__
+        owner = active_thread();
+        #endif
         unlock( lock );
 }
@@ -246 +324 @@
         thread$ * t = &try_pop_front( blocked_threads );
         held = ( t ? true : false );
+        #ifdef __CFA_DEBUG__
+        owner = ( t ? t : 0p );
+        #endif
         unpark( t );
         unlock( lock );
@@ -253 +334 @@
 static inline size_t on_wait(fast_block_lock & this) { unlock(this); return 0; }
 static inline void on_wakeup(fast_block_lock & this, size_t recursion ) { }
+
+//-----------------------------------------------------------------------------
+// simple_owner_lock
+
+// pthread owner lock
+// - reacquire for cond var
+// - recursive acquisition
+// - ownership
+struct simple_owner_lock {
+        // List of blocked threads
+        dlist( thread$ ) blocked_threads;
+
+        // Spin lock used for mutual exclusion
+        __spinlock_t lock;
+
+        // owner showing if lock is held
+        struct thread$ * owner;
+
+        size_t recursion_count;
+};
+
+static inline void  ?{}( simple_owner_lock & this ) with(this) {
+        lock{};
+        blocked_threads{};
+        owner = 0p;
+        recursion_count = 0;
+}
+static inline void ^?{}( simple_owner_lock & this ) {}
+static inline void ?{}( simple_owner_lock & this, simple_owner_lock this2 ) = void;
+static inline void ?=?( simple_owner_lock & this, simple_owner_lock this2 ) = void;
+
+static inline void lock(simple_owner_lock & this) with(this) {
+        if (owner == active_thread()) {
+                recursion_count++;
+                return;
+        }
+        lock( lock __cfaabi_dbg_ctx2 );
+
+        if (owner != 0p) {
+                insert_last( blocked_threads, *active_thread() );
+                unlock( lock );
+                park( );
+                return;
+        }
+        owner = active_thread();
+        recursion_count = 1;
+        unlock( lock );
+}
+
+// TODO: fix duplicate def issue and bring this back
+// void pop_and_set_new_owner( simple_owner_lock & this ) with( this ) {
+        // thread$ * t = &try_pop_front( blocked_threads );
+        // owner = t;
+        // recursion_count = ( t ? 1 : 0 );
+        // unpark( t );
+// }
+
+static inline void unlock(simple_owner_lock & this) with(this) {
+        lock( lock __cfaabi_dbg_ctx2 );
+        /* paranoid */ verifyf( owner != 0p, "Attempt to release lock %p that isn't held", &this );
+        /* paranoid */ verifyf( owner == active_thread(), "Thread %p other than the owner %p attempted to release owner lock %p", owner, active_thread(), &this );
+        // if recursion count is zero release lock and set new owner if one is waiting
+        recursion_count--;
+        if ( recursion_count == 0 ) {
+                // pop_and_set_new_owner( this );
+                thread$ * t = &try_pop_front( blocked_threads );
+                owner = t;
+                recursion_count = ( t ? 1 : 0 );
+                unpark( t );
+        }
+        unlock( lock );
+}
+
+static inline void on_notify(simple_owner_lock & this, struct thread$ * t ) with(this) {
+        lock( lock __cfaabi_dbg_ctx2 );
+        // lock held
+        if ( owner != 0p ) {
+                insert_last( blocked_threads, *t );
+                unlock( lock );
+        }
+        // lock not held
+        else {
+                owner = t;
+                recursion_count = 1;
+                unpark( t );
+                unlock( lock );
+        }
+}
+
+static inline size_t on_wait(simple_owner_lock & this) with(this) { 
+        lock( lock __cfaabi_dbg_ctx2 );
+        /* paranoid */ verifyf( owner != 0p, "Attempt to release lock %p that isn't held", &this );
+        /* paranoid */ verifyf( owner == active_thread(), "Thread %p other than the owner %p attempted to release owner lock %p", owner, active_thread(), &this );
+
+        size_t ret = recursion_count;
+
+        // pop_and_set_new_owner( this );
+
+        thread$ * t = &try_pop_front( blocked_threads );
+        owner = t;
+        recursion_count = ( t ? 1 : 0 );
+        unpark( t );
+
+        unlock( lock );
+        return ret;
+}
+
+static inline void on_wakeup(simple_owner_lock & this, size_t recursion ) with(this) { recursion_count = recursion; }
+
+//-----------------------------------------------------------------------------
+// Spin Queue Lock
+
+// - No reacquire for cond var
+// - No recursive acquisition
+// - No ownership
+// - spin lock with no locking/atomics in unlock
+struct spin_queue_lock {
+        // Spin lock used for mutual exclusion
+        mcs_spin_lock lock;
+
+        // flag showing if lock is held
+        volatile bool held;
+
+        #ifdef __CFA_DEBUG__
+        // for deadlock detection
+        struct thread$ * owner;
+        #endif
+};
+
+static inline void  ?{}( spin_queue_lock & this ) with(this) {
+        lock{};
+        held = false;
+}
+static inline void ^?{}( spin_queue_lock & this ) {}
+static inline void ?{}( spin_queue_lock & this, spin_queue_lock this2 ) = void;
+static inline void ?=?( spin_queue_lock & this, spin_queue_lock this2 ) = void;
+
+// if this is called recursively IT WILL DEADLOCK!!!!!
+static inline void lock(spin_queue_lock & this) with(this) {
+        mcs_spin_node node;
+        #ifdef __CFA_DEBUG__
+        assert(!(held && owner == active_thread()));
+        #endif
+        lock( lock, node );
+        while(__atomic_load_n(&held, __ATOMIC_SEQ_CST)) Pause();
+        __atomic_store_n(&held, true, __ATOMIC_SEQ_CST);
+        unlock( lock, node );
+        #ifdef __CFA_DEBUG__
+        owner = active_thread();
+        #endif
+}
+
+static inline void unlock(spin_queue_lock & this) with(this) {
+        #ifdef __CFA_DEBUG__
+        owner = 0p;
+        #endif
+        __atomic_store_n(&held, false, __ATOMIC_RELEASE);
+}
+
+static inline void on_notify(spin_queue_lock & this, struct thread$ * t ) { unpark(t); }
+static inline size_t on_wait(spin_queue_lock & this) { unlock(this); return 0; }
+static inline void on_wakeup(spin_queue_lock & this, size_t recursion ) { }
+
+
+//-----------------------------------------------------------------------------
+// MCS Block Spin Lock
+
+// - No reacquire for cond var
+// - No recursive acquisition
+// - No ownership
+// - Blocks but first node spins (like spin queue but blocking for not first thd)
+struct mcs_block_spin_lock {
+        // Spin lock used for mutual exclusion
+        mcs_lock lock;
+
+        // flag showing if lock is held
+        volatile bool held;
+
+        #ifdef __CFA_DEBUG__
+        // for deadlock detection
+        struct thread$ * owner;
+        #endif
+};
+
+static inline void  ?{}( mcs_block_spin_lock & this ) with(this) {
+        lock{};
+        held = false;
+}
+static inline void ^?{}( mcs_block_spin_lock & this ) {}
+static inline void ?{}( mcs_block_spin_lock & this, mcs_block_spin_lock this2 ) = void;
+static inline void ?=?( mcs_block_spin_lock & this, mcs_block_spin_lock this2 ) = void;
+
+// if this is called recursively IT WILL DEADLOCK!!!!!
+static inline void lock(mcs_block_spin_lock & this) with(this) {
+        mcs_node node;
+        #ifdef __CFA_DEBUG__
+        assert(!(held && owner == active_thread()));
+        #endif
+        lock( lock, node );
+        while(held) Pause();
+        held = true;
+        unlock( lock, node );
+        #ifdef __CFA_DEBUG__
+        owner = active_thread();
+        #endif
+}
+
+static inline void unlock(mcs_block_spin_lock & this) with(this) {
+        #ifdef __CFA_DEBUG__
+        owner = 0p;
+        #endif
+        held = false;
+}
+
+static inline void on_notify(mcs_block_spin_lock & this, struct thread$ * t ) { unpark(t); }
+static inline size_t on_wait(mcs_block_spin_lock & this) { unlock(this); return 0; }
+static inline void on_wakeup(mcs_block_spin_lock & this, size_t recursion ) { }
+
+//-----------------------------------------------------------------------------
+// Block Spin Lock
+
+// - No reacquire for cond var
+// - No recursive acquisition
+// - No ownership
+// - Blocks but first node spins (like spin queue but blocking for not first thd)
+struct block_spin_lock {
+        // Spin lock used for mutual exclusion
+        fast_block_lock lock;
+
+        // flag showing if lock is held
+        volatile bool held;
+
+        #ifdef __CFA_DEBUG__
+        // for deadlock detection
+        struct thread$ * owner;
+        #endif
+};
+
+static inline void  ?{}( block_spin_lock & this ) with(this) {
+        lock{};
+        held = false;
+}
+static inline void ^?{}( block_spin_lock & this ) {}
+static inline void ?{}( block_spin_lock & this, block_spin_lock this2 ) = void;
+static inline void ?=?( block_spin_lock & this, block_spin_lock this2 ) = void;
+
+// if this is called recursively IT WILL DEADLOCK!!!!!
+static inline void lock(block_spin_lock & this) with(this) {
+        #ifdef __CFA_DEBUG__
+        assert(!(held && owner == active_thread()));
+        #endif
+        lock( lock );
+        while(held) Pause();
+        held = true;
+        unlock( lock );
+        #ifdef __CFA_DEBUG__
+        owner = active_thread();
+        #endif
+}
+
+static inline void unlock(block_spin_lock & this) with(this) {
+        #ifdef __CFA_DEBUG__
+        owner = 0p;
+        #endif
+        held = false;
+}
+
+static inline void on_notify(block_spin_lock & this, struct thread$ * t ) { unpark(t); }
+static inline size_t on_wait(block_spin_lock & this) { unlock(this); return 0; }
+static inline void on_wakeup(block_spin_lock & this, size_t recursion ) { }
 
 //-----------------------------------------------------------------------------
@@ -332 +683 @@
         // - signalling without holding branded lock is UNSAFE!
         // - only allows usage of one lock, cond var is branded after usage
+
         struct fast_cond_var {
                 // List of blocked threads
                 dlist( info_thread(L) ) blocked_threads;
-
                 #ifdef __CFA_DEBUG__
                 L * lock_used;
@@ -341 +692 @@
         };
 
-
         void  ?{}( fast_cond_var(L) & this );
         void ^?{}( fast_cond_var(L) & this );
@@ -349 +699 @@
 
         uintptr_t front( fast_cond_var(L) & this );
-
         bool empty  ( fast_cond_var(L) & this );
 
         void wait( fast_cond_var(L) & this, L & l );
         void wait( fast_cond_var(L) & this, L & l, uintptr_t info );
-}
+
+
+        //-----------------------------------------------------------------------------
+        // pthread_cond_var
+        //
+        // - cond var with minimal footprint
+        // - supports operations needed for phthread cond
+
+        struct pthread_cond_var {
+                dlist( info_thread(L) ) blocked_threads;
+                __spinlock_t lock;
+        };
+
+        void  ?{}( pthread_cond_var(L) & this );
+        void ^?{}( pthread_cond_var(L) & this );
+
+        bool notify_one( pthread_cond_var(L) & this );
+        bool notify_all( pthread_cond_var(L) & this );
+
+        uintptr_t front( pthread_cond_var(L) & this );
+        bool empty ( pthread_cond_var(L) & this );
+
+        void wait( pthread_cond_var(L) & this, L & l );
+        void wait( pthread_cond_var(L) & this, L & l, uintptr_t info );
+        bool wait( pthread_cond_var(L) & this, L & l, timespec t );
+        bool wait( pthread_cond_var(L) & this, L & l, uintptr_t info, timespec t );
+}

libcfa/src/concurrency/thread.cfa

@@ -53 +53 @@
         #endif
 
-        seqable.next = 0p;
-        seqable.back = 0p;
-
         node.next = 0p;
         node.prev = 0p;
+
+        clh_node = malloc( );
+        *clh_node = false;
+
         doregister(curr_cluster, this);
-
         monitors{ &self_mon_p, 1, (fptr_t)0 };
 }
@@ -67 +67 @@
                 canary = 0xDEADDEADDEADDEADp;
         #endif
+        free(clh_node);
         unregister(curr_cluster, this);
         ^self_cor{};

libcfa/src/containers/queueLockFree.hfa

@@ -2 +2 @@
 
 #include <assert.h>
+
+#include <bits/defs.hfa>
 
 forall( T &) {

libcfa/src/startup.cfa

@@ -63 +63 @@
 
 struct __spinlock_t;
-extern "C" {
-        void __cfaabi_dbg_record_lock(struct __spinlock_t & this, const char prev_name[]) __attribute__(( weak )) libcfa_public {}
-}
 
 // Local Variables: //

src/AST/Pass.impl.hpp

@@ -182 +182 @@
 
                 // get the stmts/decls that will need to be spliced in
-                auto stmts_before = __pass::stmtsToAddBefore( core, 0);
-                auto stmts_after  = __pass::stmtsToAddAfter ( core, 0);
-                auto decls_before = __pass::declsToAddBefore( core, 0);
-                auto decls_after  = __pass::declsToAddAfter ( core, 0);
+                auto stmts_before = __pass::stmtsToAddBefore( core, 0 );
+                auto stmts_after  = __pass::stmtsToAddAfter ( core, 0 );
+                auto decls_before = __pass::declsToAddBefore( core, 0 );
+                auto decls_after  = __pass::declsToAddAfter ( core, 0 );
 
                 // These may be modified by subnode but most be restored once we exit this statemnet.
@@ -287 +287 @@
 
                 // get the stmts/decls that will need to be spliced in
-                auto stmts_before = __pass::stmtsToAddBefore( core, 0);
-                auto stmts_after  = __pass::stmtsToAddAfter ( core, 0);
-                auto decls_before = __pass::declsToAddBefore( core, 0);
-                auto decls_after  = __pass::declsToAddAfter ( core, 0);
+                auto stmts_before = __pass::stmtsToAddBefore( core, 0 );
+                auto stmts_after  = __pass::stmtsToAddAfter ( core, 0 );
+                auto decls_before = __pass::declsToAddBefore( core, 0 );
+                auto decls_after  = __pass::declsToAddAfter ( core, 0 );
 
                 // These may be modified by subnode but most be restored once we exit this statemnet.
@@ -317 +317 @@
                                 assert(( empty( stmts_before ) && empty( stmts_after ))
                                     || ( empty( decls_before ) && empty( decls_after )) );
-
-
 
                                 // Take all the statements which should have gone after, N/A for first iteration
@@ -2116 +2114 @@
         if ( __visit_children() ) {
                 bool mutated = false;
-                std::unordered_map< ast::TypeInstType::TypeEnvKey, ast::ptr< ast::Type > > new_map;
-                for ( const auto & p : node->typeEnv ) {
+                ast::TypeSubstitution::TypeMap new_map;
+                for ( const auto & p : node->typeMap ) {
                         guard_symtab guard { *this };
                         auto new_node = p.second->accept( *this );
@@ -2125 +2123 @@
                 if (mutated) {
                         auto new_node = __pass::mutate<core_t>( node );
-                        new_node->typeEnv.swap( new_map );
+                        new_node->typeMap.swap( new_map );
                         node = new_node;
                 }

src/AST/TypeSubstitution.cpp

@@ -38 +38 @@
 
 void TypeSubstitution::initialize( const TypeSubstitution &src, TypeSubstitution &dest ) {
-        dest.typeEnv.clear();
+        dest.typeMap.clear();
         dest.add( src );
 }
 
 void TypeSubstitution::add( const TypeSubstitution &other ) {
-        for ( TypeEnvType::const_iterator i = other.typeEnv.begin(); i != other.typeEnv.end(); ++i ) {
-                typeEnv[ i->first ] = i->second;
+        for ( TypeMap::const_iterator i = other.typeMap.begin(); i != other.typeMap.end(); ++i ) {
+                typeMap[ i->first ] = i->second;
         } // for
 }
 
 void TypeSubstitution::add( const TypeInstType * formalType, const Type *actualType ) {
-        typeEnv[ *formalType ] = actualType;
+        typeMap[ *formalType ] = actualType;
 }
 
 void TypeSubstitution::add( const TypeInstType::TypeEnvKey & key, const Type * actualType) {
-        typeEnv[ key ] = actualType;
+        typeMap[ key ] = actualType;
 }
 
 void TypeSubstitution::remove( const TypeInstType * formalType ) {
-        TypeEnvType::iterator i = typeEnv.find( *formalType );
-        if ( i != typeEnv.end() ) {
-                typeEnv.erase( *formalType );
-        } // if
-}
-
-const Type *TypeSubstitution::lookup( const TypeInstType * formalType ) const {
-        TypeEnvType::const_iterator i = typeEnv.find( *formalType );
+        TypeMap::iterator i = typeMap.find( *formalType );
+        if ( i != typeMap.end() ) {
+                typeMap.erase( *formalType );
+        } // if
+}
+
+const Type *TypeSubstitution::lookup(
+                const TypeInstType::TypeEnvKey & formalType ) const {
+        TypeMap::const_iterator i = typeMap.find( formalType );
 
         // break on not in substitution set
-        if ( i == typeEnv.end() ) return 0;
+        if ( i == typeMap.end() ) return 0;
 
         // attempt to transitively follow TypeInstType links.
         while ( const TypeInstType *actualType = i->second.as<TypeInstType>()) {
                 // break cycles in the transitive follow
-                if ( *formalType == *actualType ) break;
+                if ( formalType == *actualType ) break;
 
                 // Look for the type this maps to, returning previous mapping if none-such
-                i = typeEnv.find( *actualType );
-                if ( i == typeEnv.end() ) return actualType;
+                i = typeMap.find( *actualType );
+                if ( i == typeMap.end() ) return actualType;
         }
 
@@ -83 +84 @@
 }
 
+const Type *TypeSubstitution::lookup( const TypeInstType * formalType ) const {
+        return lookup( ast::TypeInstType::TypeEnvKey( *formalType ) );
+}
+
 bool TypeSubstitution::empty() const {
-        return typeEnv.empty();
+        return typeMap.empty();
 }
 
@@ -119 +124 @@
                 sub.core.subCount = 0;
                 sub.core.freeOnly = true;
-                for ( TypeEnvType::iterator i = typeEnv.begin(); i != typeEnv.end(); ++i ) {
+                for ( TypeMap::iterator i = typeMap.begin(); i != typeMap.end(); ++i ) {
                         i->second = i->second->accept( sub );
                 }
@@ -129 +134 @@
         if ( bound != boundVars.end() ) return inst;
 
-        TypeEnvType::const_iterator i = sub.typeEnv.find( *inst );
-        if ( i == sub.typeEnv.end() ) {
+        TypeMap::const_iterator i = sub.typeMap.find( *inst );
+        if ( i == sub.typeMap.end() ) {
                 return inst;
         } else {

src/AST/TypeSubstitution.hpp

@@ -75 +75 @@
         void add( const TypeSubstitution &other );
         void remove( const TypeInstType * formalType );
+        const Type *lookup( const TypeInstType::TypeEnvKey & formalType ) const;
         const Type *lookup( const TypeInstType * formalType ) const;
         bool empty() const;
@@ -104 +105 @@
         friend class Pass;
 
-        typedef std::unordered_map< TypeInstType::TypeEnvKey, ptr<Type> > TypeEnvType;
-        TypeEnvType typeEnv;
+        typedef std::unordered_map< TypeInstType::TypeEnvKey, ptr<Type> > TypeMap;
+        TypeMap typeMap;
 
   public:
-        // has to come after declaration of typeEnv
-        auto begin()       -> decltype( typeEnv.begin() ) { return typeEnv.begin(); }
-        auto   end()       -> decltype( typeEnv.  end() ) { return typeEnv.  end(); }
-        auto begin() const -> decltype( typeEnv.begin() ) { return typeEnv.begin(); }
-        auto   end() const -> decltype( typeEnv.  end() ) { return typeEnv.  end(); }
+        // has to come after declaration of typeMap
+        auto begin()       -> decltype( typeMap.begin() ) { return typeMap.begin(); }
+        auto   end()       -> decltype( typeMap.  end() ) { return typeMap.  end(); }
+        auto begin() const -> decltype( typeMap.begin() ) { return typeMap.begin(); }
+        auto   end() const -> decltype( typeMap.  end() ) { return typeMap.  end(); }
 
 };
@@ -144 +145 @@
                         if ( const TypeExpr *actual = actualIt->template as<TypeExpr>() ) {
                                 if ( formal->name != "" ) {
-                                        typeEnv[ formal ] = actual->type;
+                                        typeMap[ formal ] = actual->type;
                                 } // if
                         } else {

src/Concurrency/Waitfor.cc

@@ -56 +56 @@
                       |  |
                       |  |
-                            |  |
+                      |  |
                       |  |
                       |  |

src/Concurrency/Waitfor.h

@@ -19 +19 @@
 
 class Declaration;
+namespace ast {
+        class TranslationUnit;
+}
 
 namespace Concurrency {
         void generateWaitFor( std::list< Declaration * > & translationUnit );
+
+void generateWaitFor( ast::TranslationUnit & translationUnit );
 };
 

src/Concurrency/module.mk

@@ -19 +19 @@
         Concurrency/Keywords.cc \
         Concurrency/Keywords.h \
+        Concurrency/WaitforNew.cpp \
         Concurrency/Waitfor.cc \
         Concurrency/Waitfor.h

src/GenPoly/Specialize.cc

@@ -247 +247 @@
                         structureArg( (*actualBegin)->get_type(), argBegin, argEnd, back_inserter( appExpr->get_args() ) );
                 }
+                assertf( argBegin == argEnd, "Did not structure all arguments." );
 
                 appExpr->env = TypeSubstitution::newFromExpr( appExpr, env );

src/InitTweak/FixGlobalInit.cc

@@ -162 +162 @@
                         } // if
                         if ( Statement * ctor = ctorInit->ctor ) {
-                                addDataSectonAttribute( objDecl );
+                                addDataSectionAttribute( objDecl );
                                 initStatements.push_back( ctor );
                                 objDecl->init = nullptr;

src/InitTweak/FixInit.cc

@@ -806 +806 @@
                                                 // The attribute works, and is meant to apply, both for leaving the static local alone,
                                                 // and for hoisting it out as a static global.
-                                                addDataSectonAttribute( objDecl );
+                                                addDataSectionAttribute( objDecl );
 
                                                 // originally wanted to take advantage of gcc nested functions, but

src/InitTweak/InitTweak.cc

@@ -587 +587 @@
 
         bool isConstructable( const ast::Type * type ) {
-                return ! dynamic_cast< const ast::VarArgsType * >( type ) && ! dynamic_cast< const ast::ReferenceType * >( type ) 
+                return ! dynamic_cast< const ast::VarArgsType * >( type ) && ! dynamic_cast< const ast::ReferenceType * >( type )
                 && ! dynamic_cast< const ast::FunctionType * >( type ) && ! Tuples::isTtype( type );
         }
@@ -1025 +1025 @@
                 if (!assign) {
                         auto td = new ast::TypeDecl({}, "T", {}, nullptr, ast::TypeDecl::Dtype, true);
-                        assign = new ast::FunctionDecl({}, "?=?", {}, 
+                        assign = new ast::FunctionDecl({}, "?=?", {},
                         { new ast::ObjectDecl({}, "_dst", new ast::ReferenceType(new ast::TypeInstType("T", td))),
                           new ast::ObjectDecl({}, "_src", new ast::TypeInstType("T", td))},
@@ -1095 +1095 @@
 
                         // address of a variable or member expression is constexpr
-                        if ( ! dynamic_cast< const ast::NameExpr * >( arg ) 
-                        && ! dynamic_cast< const ast::VariableExpr * >( arg ) 
-                        && ! dynamic_cast< const ast::MemberExpr * >( arg ) 
+                        if ( ! dynamic_cast< const ast::NameExpr * >( arg )
+                        && ! dynamic_cast< const ast::VariableExpr * >( arg )
+                        && ! dynamic_cast< const ast::MemberExpr * >( arg )
                         && ! dynamic_cast< const ast::UntypedMemberExpr * >( arg ) ) result = false;
                 }
@@ -1241 +1241 @@
         }
 
-        void addDataSectonAttribute( ObjectDecl * objDecl ) {
+        #if defined( __x86_64 ) || defined( __i386 ) // assembler comment to prevent assembler warning message
+                #define ASM_COMMENT "#"
+        #else // defined( __ARM_ARCH )
+                #define ASM_COMMENT "//"
+        #endif
+        static const char * const data_section =  ".data" ASM_COMMENT;
+        static const char * const tlsd_section = ".tdata" ASM_COMMENT;
+        void addDataSectionAttribute( ObjectDecl * objDecl ) {
+                const bool is_tls = objDecl->get_storageClasses().is_threadlocal;
+                const char * section = is_tls ? tlsd_section : data_section;
                 objDecl->attributes.push_back(new Attribute("section", {
-                        new ConstantExpr( Constant::from_string(".data"
-#if defined( __x86_64 ) || defined( __i386 ) // assembler comment to prevent assembler warning message
-                                        "#"
-#else // defined( __ARM_ARCH )
-                                        "//"
-#endif
-                                ))}));
+                        new ConstantExpr( Constant::from_string( section ) )
+                }));
         }
 
         void addDataSectionAttribute( ast::ObjectDecl * objDecl ) {
+                const bool is_tls = objDecl->storage.is_threadlocal;
+                const char * section = is_tls ? tlsd_section : data_section;
                 objDecl->attributes.push_back(new ast::Attribute("section", {
-                        ast::ConstantExpr::from_string(objDecl->location, ".data"
-#if defined( __x86_64 ) || defined( __i386 ) // assembler comment to prevent assembler warning message
-                                        "#"
-#else // defined( __ARM_ARCH )
-                                        "//"
-#endif
-                                )}));
+                        ast::ConstantExpr::from_string(objDecl->location, section)
+                }));
         }
 

src/InitTweak/InitTweak.h

@@ -127 +127 @@
         ///    .section .data#,"a"
         /// to avoid assembler warning "ignoring changed section attributes for .data"
-        void addDataSectonAttribute( ObjectDecl * objDecl );
+        void addDataSectionAttribute( ObjectDecl * objDecl );
 
         void addDataSectionAttribute( ast::ObjectDecl * objDecl );

src/main.cc

@@ -10 +10 @@
 // Created On       : Fri May 15 23:12:02 2015
 // Last Modified By : Andrew Beach
-// Last Modified On : Fri Apr 29  9:52:00 2022
-// Update Count     : 673
+// Last Modified On : Tue Jun  7 13:29:00 2022
+// Update Count     : 674
 //
 
@@ -447 +447 @@
                         PASS( "Expand Unique Expr", Tuples::expandUniqueExpr( transUnit ) ); // xxx - is this the right place for this? want to expand ASAP so tha, sequent passes don't need to worry about double-visiting a unique expr - needs to go after InitTweak::fix so that copy constructed return declarations are reused
 
-                        PASS( "Translate Tries" , ControlStruct::translateTries( transUnit ) );
+                        PASS( "Translate Tries", ControlStruct::translateTries( transUnit ) );
+                        PASS( "Gen Waitfor", Concurrency::generateWaitFor( transUnit ) );
 
                         translationUnit = convert( move( transUnit ) );
@@ -517 +518 @@
 
                         PASS( "Expand Unique Expr", Tuples::expandUniqueExpr( translationUnit ) ); // xxx - is this the right place for this? want to expand ASAP so tha, sequent passes don't need to worry about double-visiting a unique expr - needs to go after InitTweak::fix so that copy constructed return declarations are reused
-
-                        PASS( "Translate Tries" , ControlStruct::translateTries( translationUnit ) );
+                        PASS( "Translate Tries", ControlStruct::translateTries( translationUnit ) );
+                        PASS( "Gen Waitfor", Concurrency::generateWaitFor( translationUnit ) );
                 }
-
-                PASS( "Gen Waitfor" , Concurrency::generateWaitFor( translationUnit ) );
 
                 PASS( "Convert Specializations",  GenPoly::convertSpecializations( translationUnit ) ); // needs to happen before tuple types are expanded

tests/unified_locking/mutex_test.hfa

@@ -22 +22 @@
 }
 
-uint32_t cs() {
+uint32_t cs(uint32_t & entries) {
         thread$ * me = active_thread();
         uint32_t value;
         lock(mo.l);
         {
+                entries++;
                 uint32_t tsum = mo.sum;
                 uint32_t cnt = mo.cnt;
@@ -42 +43 @@
 thread LockCheck {
         uint32_t sum;
+        uint32_t entries;
 };
 
 void main(LockCheck & this) {
         this.sum = 0;
+        this.entries = 0;
         for(num_times) {
                 trash();
-                this.sum += cs();
+                this.sum += cs( this.entries );
                 trash();
                 yield(random(10));
@@ -58 +61 @@
         mo.sum = -32;
         mo.cnt = 0;
+        uint32_t real_entries = 0;
         processor p[2];
         sout | "Starting";
@@ -63 +67 @@
                 LockCheck checkers[13];
                 for(i;13) {
-                        sum += join(checkers[i]).sum;
+                        LockCheck & curr = join(checkers[i]);
+                        sum += curr.sum;
+                        real_entries += curr.entries;
                 }
         }
         sout | "Done!";
-        if(mo.cnt != (13 * num_times)) sout | "Invalid cs count!" | mo.cnt | "vs "| (13 * num_times) | "(13 *" | num_times | ')';
+        if(real_entries != (13 * num_times)) sout | "Invalid real cs count!" | mo.cnt | "vs "| (13 * num_times) | "(13 *" | num_times | ')';
+        if(mo.cnt != (13 * num_times)) sout | "Invalid concurrent cs count!" | mo.cnt | "vs "| (13 * num_times) | "(13 *" | num_times | ')';
         if(sum == mo.sum) sout | "Match!";
         else sout | "No Match!" | sum | "vs" | mo.sum;

tools/cfa.nanorc

@@ -10 +10 @@
 color green "\<(forall|trait|(o|d|f|t)type|mutex|_Bool|volatile|virtual)\>"
 color green "\<(float|double|bool|char|int|short|long|enum|void|auto)\>"
-color green "\<(static|const|extern|(un)?signed|inline)\>" "\<(sizeof)\>"
+color green "\<(static|const|extern|(un)?signed|inline|sizeof|vtable)\>"
 color green "\<((s?size)|one|zero|((u_?)?int(8|16|32|64|ptr)))_t\>"
 
 # Declarations
 color brightgreen "\<(struct|union|typedef|trait|coroutine|generator)\>"
-color brightgreen "\<(monitor|thread|with)\>"
+color brightgreen "\<(monitor|thread|with|exception)\>"
 
 # Control Flow Structures

tools/jenkins/setup.sh.in

@@ -29 +29 @@
 function getrunpath()
 {
-        local elfout=$(readelf -d $1 | grep "RUNPATH")
+        local elfout=$(readelf -d $1 | grep -E "RPATH|RUNPATH")
         regex='\[/([[:alpha:][:digit:]@/_.-]+)\]'
         if [[ $elfout =~ $regex ]]; then
@@ -43 +43 @@
 {
         local deps=$(ldd $1)
+        retlcldeps=()
         retsysdeps=()
-        retlcldeps=()
         while IFS= read -r line; do
-                regex1='/([[:alpha:][:digit:]@/_.-]+)'
-                regex2='(libcfa[[:alpha:][:digit:].]+) => not found'
+                regex1='(libcfa[[:alpha:][:digit:].]+)'
+                regex2='/([[:alpha:][:digit:]@/_.-]+)'
                 regex3='linux-vdso.so.1|linux-gate.so.1'
                 if [[ $line =~ $regex1 ]]; then
+                        retlcldeps+=(${BASH_REMATCH[1]})
+                elif [[ $line =~ $regex2 ]]; then
                         retsysdeps+=(${BASH_REMATCH[1]})
-                elif [[ $line =~ $regex2 ]]; then
-                        retlcldeps+=(${BASH_REMATCH[1]})
                 elif [[ $line =~ $regex3 ]]; then
                         # echo "ignoring '$line}': intrinsic"