Changes in / [eb5962a:1df492a]

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 (deleted)
• benchmark/io/http/printer.hfa (deleted)
• benchmark/io/http/protocol.cfa (modified) (5 diffs)
• benchmark/io/http/protocol.hfa (modified) (2 diffs)
• benchmark/io/http/socket.cfa (deleted)
• benchmark/io/http/socket.hfa (deleted)
• benchmark/io/http/worker.cfa (modified) (3 diffs)
• benchmark/io/http/worker.hfa (modified) (3 diffs)
• benchmark/io/sendfile/producer.cfa (deleted)
• 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 (deleted)
• 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 (deleted)
• tests/concurrent/futures/wait_any.cfa (deleted)
• tests/unified_locking/.expect/block_spin_lock.txt (deleted)
• tests/unified_locking/.expect/clh.txt (deleted)
• tests/unified_locking/.expect/mcs_block_spin_lock.txt (deleted)
• tests/unified_locking/.expect/mcs_spin.txt (deleted)
• tests/unified_locking/.expect/pthread_locks.txt (deleted)
• tests/unified_locking/.expect/simple_owner_lock.txt (deleted)
• tests/unified_locking/.expect/spin_queue_lock.txt (deleted)
• tests/unified_locking/block_spin_lock.cfa (deleted)
• tests/unified_locking/clh.cfa (deleted)
• tests/unified_locking/mcs_block_spin_lock.cfa (deleted)
• tests/unified_locking/mcs_spin.cfa (deleted)
• tests/unified_locking/mutex_test.hfa (modified) (4 diffs)
• tests/unified_locking/pthread_locks.cfa (deleted)
• tests/unified_locking/simple_owner_lock.cfa (deleted)
• tests/unified_locking/spin_queue_lock.cfa (deleted)
• 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 -g # -Werror
+AM_CFLAGS = -O3 -Wall -Wextra -I$(srcdir) -lrt -pthread # -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>
@@ -9 +8 @@
         #include <sched.h>
         #include <signal.h>
-        #include <sys/eventfd.h>
         #include <sys/socket.h>
         #include <netinet/in.h>
@@ -16 +14 @@
 #include <fstream.hfa>
 #include <kernel.hfa>
-#include <locks.hfa>
 #include <iofwd.hfa>
 #include <stats.hfa>
@@ -24 +21 @@
 #include "filecache.hfa"
 #include "options.hfa"
-#include "socket.hfa"
-#include "printer.hfa"
 #include "worker.hfa"
 
@@ -32 +27 @@
 Duration default_preemption() {
         return 0;
+}
+
+//=============================================================================================
+// 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!";
+                }
+        }
 }
 
@@ -37 +101 @@
 // Globals
 //=============================================================================================
+struct ServerCluster {
+        cluster self;
+        processor    * procs;
+        // io_context   * ctxs;
+        StatsPrinter * prnt;
+
+};
+
 void ?{}( ServerCluster & this ) {
         (this.self){ "Server Cluster", options.clopts.params };
@@ -50 +122 @@
                 (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__)
@@ -75 +147 @@
         }
 
+        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 );
@@ -84 +163 @@
 
 void ^?{}( ServerCluster & this ) {
+        delete(this.prnt);
+
         for(i; options.clopts.nprocs) {
                 ^(this.procs[i]){};
@@ -94 +175 @@
 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;
-}
 
 //=============================================================================================
@@ -121 +180 @@
 //============================================================================================='
 int main( int argc, char * argv[] ) {
-        int ret;
         __sighandler_t s = 1p;
         signal(SIGPIPE, s);
@@ -128 +186 @@
         // 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";
-        }
 
         //===================
@@ -153 +197 @@
         // 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 = prepaddr(address);
-
-        int server_fd;
+        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) );
+        }
 
         //===================
@@ -176 +257 @@
 
                 {
-                        // 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();
                         {
-                                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 };
+                                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;
                                         }
-                                        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] );
-                                        }
+                                        unpark( workers[i] );
                                 }
                                 sout | options.clopts.nworkers | "workers started on" | options.clopts.nprocs | "processors /" | options.clopts.nclusters | "clusters";
@@ -251 +285 @@
                                 }
                                 sout | nl;
+                                if(!options.interactive) park();
                                 {
-                                        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 | "'";
-                                                }
+                                        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 | "Shutdown received";
                                 }
 
+                                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 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]);
-                                        }
-                                }
+                                // 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);
                         }
                         sout | "done";
@@ -363 +331 @@
                         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
-                false  // reuseport
+                8080, // port
+                10,   // backlog
+                1024  // buflen
         },
 
@@ -53 +52 @@
 
 void parse_options( int argc, char * argv[] ) {
+        // bool fixedfd = false;
+        // bool sqkpoll = false;
+        // bool iokpoll = false;
         unsigned nentries = 0;
         bool isolate = false;
@@ -68 +70 @@
                 {'\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 {
@@ -479 +470 @@
 
                         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);
@@ -513 +503 @@
 }
 
-[HttpCode code, bool closed, * const char file, size_t len] http_read(volatile int & fd, []char buffer, size_t len, io_future_t * f) {
+[HttpCode code, bool closed, * const char file, size_t len] http_read(int fd, []char buffer, size_t len) {
         char * it = buffer;
         size_t count = len - 1;
@@ -519 +509 @@
         READ:
         for() {
-                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 = 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];
@@ -587 +570 @@
 
 void ?{}( DateFormater & this ) {
-        ((thread&)this){ "Server Date Thread" };
+        ((thread&)this){ "Server Date Thread", *options.clopts.instance[0] };
         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;
 
@@ -23 +22 @@
 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(volatile int & fd, []char buffer, size_t len, io_future_t * f);
+[HttpCode code, bool closed, * const char file, size_t len] http_read(int fd, []char buffer, size_t len);

benchmark/io/http/worker.cfa

@@ -8 +8 @@
 #include <fstream.hfa>
 #include <iofwd.hfa>
-#include <mutex_stmt.hfa>
 
 #include "options.hfa"
@@ -15 +14 @@
 
 //=============================================================================================
-// Generic connection handling
+// Worker Thread
 //=============================================================================================
-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;
+void ?{}( Worker & 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;
 
-                // 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;
-                        }
-                }
+        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;
         }
 }
 
-//=============================================================================================
-// 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.done = false;
+extern "C" {
+extern int accept4(int sockfd, struct sockaddr *addr, socklen_t *addrlen, int flags);
 }
 
-void main( AcceptWorker & this ) {
+void main( Worker & this ) {
         park();
-        unsigned long long last = rdtscl();
-        /* paranoid */ assert( this.conn.pipe[0] != -1 );
-        /* paranoid */ assert( this.conn.pipe[1] != -1 );
+        /* paranoid */ assert( this.pipe[0] != -1 );
+        /* paranoid */ assert( this.pipe[1] != -1 );
+
+        CONNECTION:
         for() {
-                if( options.log ) mutex(sout) sout | "=== Accepting connection ===";
-                int fd = cfa_accept4( this.sockfd, this.[addr, addrlen, flags], CFA_IO_LAZY );
+                if( options.log ) sout | "=== Accepting connection ===";
+                int fd = cfa_accept4( this.[sockfd, addr, addrlen, flags], CFA_IO_LAZY );
                 if(fd < 0) {
                         if( errno == ECONNABORTED ) break;
@@ -137 +55 @@
                 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 ) sout | "=== New connection" | fd | "" | ", waiting for requests ===";
+                REQUEST:
+                for() {
+                        bool closed;
+                        HttpCode code;
+                        const char * file;
+                        size_t name_size;
 
-                if( options.log ) mutex(sout) sout | "=== Connection closed ===";
+                        // 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 | ") ===";
+                                }
+                                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 ) sout | "=== Answer sent ===";
+                }
+
+                if( options.log ) sout | "=== Connection closed ===";
+                continue CONNECTION;
         }
 }
-
-
-//=============================================================================================
-// 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 ===";
-
-                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;
-                                }
-                        }
-                }
-
-                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;
-                                }
-                        }
-                }
-
-                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>
 
@@ -9 +7 @@
 }
 
-#include "printer.hfa"
-
 //=============================================================================================
 // Worker Thread
 //=============================================================================================
 
-struct connection {
-        int pipe[2];
+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 {
-                sendfile_stats_t sendfile;
-        } stats;
+                volatile uint64_t calls;
+                volatile uint64_t bytes;
+        } avgrd[zipf_cnts];
 };
 
-static inline void ?{}( connection & this ) {
-        this.pipe[0] = -1;
-        this.pipe[1] = -1;
-}
-
-thread AcceptWorker {
-        connection conn;
+thread Worker {
+        int pipe[2];
         int sockfd;
         struct sockaddr * addr;
@@ -34 +33 @@
         int flags;
         volatile bool done;
+        struct {
+                sendfile_stats_t sendfile;
+        } stats;
 };
-void ?{}( AcceptWorker & this);
-void main( AcceptWorker & );
-
-
-struct PendingRead {
-        PendingRead * volatile next;
-        io_future_t f;
-        struct {
-                void * buf;
-                size_t len;
-        } in;
-        struct {
-                volatile int fd;
-        } out;
-};
-
-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 & );
+void ?{}( Worker & this);
+void main( Worker & );

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
 };
 
@@ -32 +40 @@
                 extern void disable_interrupts() OPTIONAL_THREAD;
                 extern void enable_interrupts( bool poll = true ) OPTIONAL_THREAD;
-                #define __cfaabi_dbg_record_lock(x, y)
+
+                #ifdef __CFA_DEBUG__
+                        void __cfaabi_dbg_record_lock(__spinlock_t & this, const char prev_name[]);
+                #else
+                        #define __cfaabi_dbg_record_lock(x, y)
+                #endif
         }
 
         static inline void ?{}( __spinlock_t & this ) {
                 this.lock = 0;
+                #ifdef __CFA_DEBUG__
+                        this.spin_count = 0;
+                #endif
         }
 
@@ -61 +77 @@
                 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$);
@@ -202 +208 @@
                         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;
-                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);
-                }
+
+                // 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;
 
                 __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, was %u\n", nentries);
+                        abort("ERROR: I/O setup 'num_entries' must be a power of 2\n");
                 }
 

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();
@@ -212 +213 @@
                         thread$ * post(oneshot & this, bool do_unpark = true) {
                                 struct thread$ * got = __atomic_exchange_n( &this.ptr, 1p, __ATOMIC_SEQ_CST);
-                                if( got == 0p || got == 1p ) return 0p;
+                                if( got == 0p ) return 0p;
                                 if(do_unpark) unpark( got );
                                 return got;
@@ -262 +263 @@
 
                                         // 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;
@@ -273 +275 @@
                         // should retract the wait ctx
                         // intented to be use by wait, wait_any, waitfor, etc. rather than used directly
-                        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;
-                                        }
-                                }
+                        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");
                         }
 
@@ -381 +379 @@
                                 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 ); }
 }
 
@@ -421 +388 @@
                 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

@@ -98 +98 @@
         mcs_node * next = advance(l.queue, &n);
         if(next) post(next->sem);
-}
-
-//-----------------------------------------------------------------------------
-// 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;
 }
 
@@ -270 +205 @@
 // Fast Block Lock
 
-// minimal blocking lock
+// High efficiency minimal blocking lock
 // - No reacquire for cond var
 // - No recursive acquisition
 // - No ownership
 struct fast_block_lock {
+        // Spin lock used for mutual exclusion
+        __spinlock_t lock;
+
         // List of blocked threads
         dlist( thread$ ) blocked_threads;
 
-        // Spin lock used for mutual exclusion
-        __spinlock_t lock;
-
-        // flag showing if lock is held
         bool held:1;
-
-        #ifdef __CFA_DEBUG__
-        // for deadlock detection
-        struct thread$ * owner;
-        #endif
 };
 
@@ -302 +231 @@
 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() );
@@ -313 +238 @@
         }
         held = true;
-        #ifdef __CFA_DEBUG__
-        owner = active_thread();
-        #endif
         unlock( lock );
 }
@@ -324 +246 @@
         thread$ * t = &try_pop_front( blocked_threads );
         held = ( t ? true : false );
-        #ifdef __CFA_DEBUG__
-        owner = ( t ? t : 0p );
-        #endif
         unpark( t );
         unlock( lock );
@@ -334 +253 @@
 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 ) { }
 
 //-----------------------------------------------------------------------------
@@ -683 +332 @@
         // - 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;
@@ -692 +341 @@
         };
 
+
         void  ?{}( fast_cond_var(L) & this );
         void ^?{}( fast_cond_var(L) & this );
@@ -699 +349 @@
 
         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;
+        doregister(curr_cluster, this);
 
-        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
@@ -2114 +2116 @@
         if ( __visit_children() ) {
                 bool mutated = false;
-                ast::TypeSubstitution::TypeMap new_map;
-                for ( const auto & p : node->typeMap ) {
+                std::unordered_map< ast::TypeInstType::TypeEnvKey, ast::ptr< ast::Type > > new_map;
+                for ( const auto & p : node->typeEnv ) {
                         guard_symtab guard { *this };
                         auto new_node = p.second->accept( *this );
@@ -2123 +2125 @@
                 if (mutated) {
                         auto new_node = __pass::mutate<core_t>( node );
-                        new_node->typeMap.swap( new_map );
+                        new_node->typeEnv.swap( new_map );
                         node = new_node;
                 }

src/AST/TypeSubstitution.cpp

@@ -38 +38 @@
 
 void TypeSubstitution::initialize( const TypeSubstitution &src, TypeSubstitution &dest ) {
-        dest.typeMap.clear();
+        dest.typeEnv.clear();
         dest.add( src );
 }
 
 void TypeSubstitution::add( const TypeSubstitution &other ) {
-        for ( TypeMap::const_iterator i = other.typeMap.begin(); i != other.typeMap.end(); ++i ) {
-                typeMap[ i->first ] = i->second;
+        for ( TypeEnvType::const_iterator i = other.typeEnv.begin(); i != other.typeEnv.end(); ++i ) {
+                typeEnv[ i->first ] = i->second;
         } // for
 }
 
 void TypeSubstitution::add( const TypeInstType * formalType, const Type *actualType ) {
-        typeMap[ *formalType ] = actualType;
+        typeEnv[ *formalType ] = actualType;
 }
 
 void TypeSubstitution::add( const TypeInstType::TypeEnvKey & key, const Type * actualType) {
-        typeMap[ key ] = actualType;
+        typeEnv[ key ] = actualType;
 }
 
 void TypeSubstitution::remove( const TypeInstType * formalType ) {
-        TypeMap::iterator i = typeMap.find( *formalType );
-        if ( i != typeMap.end() ) {
-                typeMap.erase( *formalType );
+        TypeEnvType::iterator i = typeEnv.find( *formalType );
+        if ( i != typeEnv.end() ) {
+                typeEnv.erase( *formalType );
         } // if
 }
 
-const Type *TypeSubstitution::lookup(
-                const TypeInstType::TypeEnvKey & formalType ) const {
-        TypeMap::const_iterator i = typeMap.find( formalType );
+const Type *TypeSubstitution::lookup( const TypeInstType * formalType ) const {
+        TypeEnvType::const_iterator i = typeEnv.find( *formalType );
 
         // break on not in substitution set
-        if ( i == typeMap.end() ) return 0;
+        if ( i == typeEnv.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 = typeMap.find( *actualType );
-                if ( i == typeMap.end() ) return actualType;
+                i = typeEnv.find( *actualType );
+                if ( i == typeEnv.end() ) return actualType;
         }
 
@@ -84 +83 @@
 }
 
-const Type *TypeSubstitution::lookup( const TypeInstType * formalType ) const {
-        return lookup( ast::TypeInstType::TypeEnvKey( *formalType ) );
-}
-
 bool TypeSubstitution::empty() const {
-        return typeMap.empty();
+        return typeEnv.empty();
 }
 
@@ -124 +119 @@
                 sub.core.subCount = 0;
                 sub.core.freeOnly = true;
-                for ( TypeMap::iterator i = typeMap.begin(); i != typeMap.end(); ++i ) {
+                for ( TypeEnvType::iterator i = typeEnv.begin(); i != typeEnv.end(); ++i ) {
                         i->second = i->second->accept( sub );
                 }
@@ -134 +129 @@
         if ( bound != boundVars.end() ) return inst;
 
-        TypeMap::const_iterator i = sub.typeMap.find( *inst );
-        if ( i == sub.typeMap.end() ) {
+        TypeEnvType::const_iterator i = sub.typeEnv.find( *inst );
+        if ( i == sub.typeEnv.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;
@@ -105 +104 @@
         friend class Pass;
 
-        typedef std::unordered_map< TypeInstType::TypeEnvKey, ptr<Type> > TypeMap;
-        TypeMap typeMap;
+        typedef std::unordered_map< TypeInstType::TypeEnvKey, ptr<Type> > TypeEnvType;
+        TypeEnvType typeEnv;
 
   public:
-        // 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(); }
+        // 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(); }
 
 };
@@ -145 +144 @@
                         if ( const TypeExpr *actual = actualIt->template as<TypeExpr>() ) {
                                 if ( formal->name != "" ) {
-                                        typeMap[ formal ] = actual->type;
+                                        typeEnv[ 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 ) {
-                                addDataSectionAttribute( objDecl );
+                                addDataSectonAttribute( 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.
-                                                addDataSectionAttribute( objDecl );
+                                                addDataSectonAttribute( 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 @@
         }
 
-        #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;
+        void addDataSectonAttribute( ObjectDecl * objDecl ) {
                 objDecl->attributes.push_back(new Attribute("section", {
-                        new ConstantExpr( Constant::from_string( 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
+                                ))}));
         }
 
         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, 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
+                                )}));
         }
 

src/InitTweak/InitTweak.h

@@ -127 +127 @@
         ///    .section .data#,"a"
         /// to avoid assembler warning "ignoring changed section attributes for .data"
-        void addDataSectionAttribute( ObjectDecl * objDecl );
+        void addDataSectonAttribute( 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 : Tue Jun  7 13:29:00 2022
-// Update Count     : 674
+// Last Modified On : Fri Apr 29  9:52:00 2022
+// Update Count     : 673
 //
 
@@ -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( "Gen Waitfor", Concurrency::generateWaitFor( transUnit ) );
+                        PASS( "Translate Tries" , ControlStruct::translateTries( transUnit ) );
 
                         translationUnit = convert( move( transUnit ) );
@@ -518 +517 @@
 
                         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( "Gen Waitfor", Concurrency::generateWaitFor( translationUnit ) );
+
+                        PASS( "Translate Tries" , ControlStruct::translateTries( 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 & entries) {
+uint32_t cs() {
         thread$ * me = active_thread();
         uint32_t value;
         lock(mo.l);
         {
-                entries++;
                 uint32_t tsum = mo.sum;
                 uint32_t cnt = mo.cnt;
@@ -43 +42 @@
 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.entries );
+                this.sum += cs();
                 trash();
                 yield(random(10));
@@ -61 +58 @@
         mo.sum = -32;
         mo.cnt = 0;
-        uint32_t real_entries = 0;
         processor p[2];
         sout | "Starting";
@@ -67 +63 @@
                 LockCheck checkers[13];
                 for(i;13) {
-                        LockCheck & curr = join(checkers[i]);
-                        sum += curr.sum;
-                        real_entries += curr.entries;
+                        sum += join(checkers[i]).sum;
                 }
         }
         sout | "Done!";
-        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(mo.cnt != (13 * num_times)) sout | "Invalid 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|vtable)\>"
+color green "\<(static|const|extern|(un)?signed|inline)\>" "\<(sizeof)\>"
 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|exception)\>"
+color brightgreen "\<(monitor|thread|with)\>"
 
 # Control Flow Structures

tools/jenkins/setup.sh.in

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