Changeset 5407cdc for libcfa/src/concurrency

Timestamp:

Apr 28, 2021, 4:56:50 PM (5 years ago)

Author:

Thierry Delisle <tdelisle@…>

Branches:

ADT, arm-eh, ast-experimental, enum, forall-pointer-decay, jacob/cs343-translation, master, new-ast-unique-expr, pthread-emulation, qualifiedEnum

Children:

8d66610

Parents:

feacef9 (diff), b7fd2db6 (diff)
Note: this is a merge changeset, the changes displayed below correspond to the merge itself.
Use the (diff) links above to see all the changes relative to each parent.

Message:

Merge branch 'master' of plg.uwaterloo.ca:software/cfa/cfa-cc

Location:

libcfa/src/concurrency

Files:

• alarm.cfa (modified) (4 diffs)
• clib/cfathread.cfa (modified) (3 diffs)
• clib/cfathread.h (modified) (2 diffs)
• coroutine.cfa (modified) (5 diffs)
• coroutine.hfa (modified) (4 diffs)
• future.hfa (modified) (2 diffs)
• invoke.c (modified) (2 diffs)
• invoke.h (modified) (3 diffs)
• io.cfa (modified) (6 diffs)
• io/call.cfa.in (modified) (26 diffs)
• io/setup.cfa (modified) (12 diffs)
• io/types.hfa (modified) (7 diffs)
• iofwd.hfa (modified) (4 diffs)
• kernel.cfa (modified) (33 diffs)
• kernel.hfa (modified) (12 diffs)
• kernel/fwd.hfa (modified) (6 diffs)
• kernel/startup.cfa (modified) (19 diffs)
• kernel_private.hfa (modified) (4 diffs)
• locks.cfa (modified) (9 diffs)
• locks.hfa (modified) (5 diffs)
• monitor.hfa (modified) (1 diff)
• preemption.cfa (modified) (12 diffs)
• ready_queue.cfa (modified) (19 diffs)
• ready_subqueue.hfa (modified) (5 diffs)
• stats.cfa (modified) (3 diffs)
• stats.hfa (modified) (4 diffs)
• thread.cfa (modified) (7 diffs)
• thread.hfa (modified) (4 diffs)

libcfa/src/concurrency/alarm.cfa

@@ -15 +15 @@
 
 #define __cforall_thread__
+// #define __CFA_DEBUG_PRINT_PREEMPTION__
 
 #include <errno.h>
@@ -107 +108 @@
                 bool first = ! & alarms`first;
 
+                __cfadbg_print_safe( preemption, " KERNEL: alarm inserting %p (%lu).\n", this, this->alarm.tn );
                 insert( &alarms, this );
                 if( first ) {
@@ -114 +116 @@
         unlock( event_kernel->lock );
         this->set = true;
-        enable_interrupts( __cfaabi_dbg_ctx );
+        enable_interrupts();
 }
 
@@ -125 +127 @@
         }
         unlock( event_kernel->lock );
-        enable_interrupts( __cfaabi_dbg_ctx );
+        enable_interrupts();
         this->set = false;
 }

libcfa/src/concurrency/clib/cfathread.cfa

@@ -14 +14 @@
 //
 
+#include "fstream.hfa"
+#include "locks.hfa"
 #include "kernel.hfa"
+#include "stats.hfa"
 #include "thread.hfa"
-
-thread CRunner {
-        void (*themain)( CRunner * );
+#include "time.hfa"
+
+#include "cfathread.h"
+
+extern void ?{}(processor &, const char[], cluster &, $thread *);
+extern "C" {
+      extern void __cfactx_invoke_thread(void (*main)(void *), void * this);
+}
+
+//================================================================================
+// Thread run y the C Interface
+
+struct cfathread_object {
+        $thread self;
+        void * (*themain)( void * );
+        void * arg;
+        void * ret;
 };
-
-static void ?{}( CRunner & this, void (*themain)( CRunner * ) ) {
+void main(cfathread_object & this);
+void ^?{}(cfathread_object & mutex this);
+
+static inline $thread * get_thread( cfathread_object & this ) { return &this.self; }
+
+typedef ThreadCancelled(cfathread_object) cfathread_exception;
+typedef ThreadCancelled_vtable(cfathread_object) cfathread_vtable;
+
+void defaultResumptionHandler(ThreadCancelled(cfathread_object) & except) {
+        abort | "A thread was cancelled";
+}
+
+cfathread_vtable _cfathread_vtable_instance;
+
+cfathread_vtable & const _default_vtable = _cfathread_vtable_instance;
+
+cfathread_vtable const & get_exception_vtable(cfathread_exception *) {
+        return _cfathread_vtable_instance;
+}
+
+static void ?{}( cfathread_object & this, cluster & cl, void *(*themain)( void * ), void * arg ) {
         this.themain = themain;
-}
-
-void main( CRunner & this ) {
-        this.themain( &this );
-}
-
-processor * procs = 0p;
-int proc_cnt = 1;
-
+        this.arg = arg;
+        (this.self){"C-thread", cl};
+        __thrd_start(this, main);
+}
+
+void ^?{}(cfathread_object & mutex this) {
+        ^(this.self){};
+}
+
+void main( cfathread_object & this ) {
+        __attribute__((unused)) void * const thrd_obj = (void*)&this;
+        __attribute__((unused)) void * const thrd_hdl = (void*)active_thread();
+        /* paranoid */ verify( thrd_obj == thrd_hdl );
+
+        this.ret = this.themain( this.arg );
+}
+
+//================================================================================
+// Special Init Thread responsible for the initialization or processors
+struct __cfainit {
+        $thread self;
+        void (*init)( void * );
+        void * arg;
+};
+void main(__cfainit & this);
+void ^?{}(__cfainit & mutex this);
+
+static inline $thread * get_thread( __cfainit & this ) { return &this.self; }
+
+typedef ThreadCancelled(__cfainit) __cfainit_exception;
+typedef ThreadCancelled_vtable(__cfainit) __cfainit_vtable;
+
+void defaultResumptionHandler(ThreadCancelled(__cfainit) & except) {
+        abort | "The init thread was cancelled";
+}
+
+__cfainit_vtable ___cfainit_vtable_instance;
+
+__cfainit_vtable const & get_exception_vtable(__cfainit_exception *) {
+        return ___cfainit_vtable_instance;
+}
+
+static void ?{}( __cfainit & this, void (*init)( void * ), void * arg ) {
+        this.init = init;
+        this.arg = arg;
+        (this.self){"Processir Init"};
+
+        // Don't use __thrd_start! just prep the context manually
+        $thread * this_thrd = get_thread(this);
+        void (*main_p)(__cfainit &) = main;
+
+        disable_interrupts();
+        __cfactx_start(main_p, get_coroutine(this), this, __cfactx_invoke_thread);
+
+        this_thrd->context.[SP, FP] = this_thrd->self_cor.context.[SP, FP];
+        /* paranoid */ verify( this_thrd->context.SP );
+
+        this_thrd->state = Ready;
+        enable_interrupts();
+}
+
+void ^?{}(__cfainit & mutex this) {
+        ^(this.self){};
+}
+
+void main( __cfainit & this ) {
+        __attribute__((unused)) void * const thrd_obj = (void*)&this;
+        __attribute__((unused)) void * const thrd_hdl = (void*)active_thread();
+        /* paranoid */ verify( thrd_obj == thrd_hdl );
+
+        this.init( this.arg );
+}
+
+//================================================================================
+// Main Api
 extern "C" {
-        //--------------------
-        // Basic thread management
-        CRunner * cfathread_create( void (*main)( CRunner * ) ) {
-                return new( main );
-        }
-
-        void cfathread_join( CRunner * thrd ) {
-                delete( thrd );
+        int cfathread_cluster_create(cfathread_cluster_t * cl) __attribute__((nonnull(1))) {
+                *cl = new();
+                return 0;
+        }
+
+        cfathread_cluster_t cfathread_cluster_self(void) {
+                return active_cluster();
+        }
+
+        int cfathread_cluster_print_stats( cfathread_cluster_t cl ) {
+                #if !defined(__CFA_NO_STATISTICS__)
+                        print_stats_at_exit( *cl, CFA_STATS_READY_Q | CFA_STATS_IO );
+                        print_stats_now( *cl, CFA_STATS_READY_Q | CFA_STATS_IO );
+                #endif
+                return 0;
+        }
+
+        int cfathread_cluster_add_worker(cfathread_cluster_t cl, pthread_t* tid, void (*init_routine) (void *), void * arg) {
+                __cfainit * it = 0p;
+                if(init_routine) {
+                        it = alloc();
+                        (*it){init_routine, arg};
+                }
+                processor * proc = alloc();
+                (*proc){ "C-processor", *cl, get_thread(*it) };
+
+                // Wait for the init thread to return before continuing
+                if(it) {
+                        ^(*it){};
+                        free(it);
+                }
+
+                if(tid) *tid = proc->kernel_thread;
+                return 0;
+        }
+
+        int cfathread_cluster_pause (cfathread_cluster_t) {
+                abort | "Pausing clusters is not supported";
+                exit(1);
+        }
+
+        int cfathread_cluster_resume(cfathread_cluster_t) {
+                abort | "Resuming clusters is not supported";
+                exit(1);
+        }
+
+        //--------------------
+        // Thread attributes
+        int cfathread_attr_init(cfathread_attr_t *attr) __attribute__((nonnull (1))) {
+                attr->cl = active_cluster();
+                return 0;
+        }
+
+        //--------------------
+        // Thread
+        int cfathread_create( cfathread_t * handle, const cfathread_attr_t * attr, void *(*main)( void * ), void * arg ) __attribute__((nonnull (1))) {
+                cluster * cl = attr ? attr->cl : active_cluster();
+                cfathread_t thrd = alloc();
+                (*thrd){ *cl, main, arg };
+                *handle = thrd;
+                return 0;
+        }
+
+        int cfathread_join( cfathread_t thrd, void ** retval ) {
+                void * ret = join( *thrd ).ret;
+                ^( *thrd ){};
+                free(thrd);
+                if(retval) {
+                        *retval = ret;
+                }
+                return 0;
+        }
+
+        int cfathread_get_errno(void) {
+                return errno;
+        }
+
+        cfathread_t cfathread_self(void) {
+                return (cfathread_t)active_thread();
+        }
+
+        int cfathread_usleep(useconds_t usecs) {
+                sleep(usecs`us);
+                return 0;
+        }
+
+        int cfathread_sleep(unsigned int secs) {
+                sleep(secs`s);
+                return 0;
         }
 
@@ -47 +230 @@
         }
 
-        void cfathread_unpark( CRunner * thrd ) {
+        void cfathread_unpark( cfathread_t thrd ) {
                 unpark( *thrd );
         }
@@ -55 +238 @@
         }
 
-        //--------------------
-        // Basic kernel features
-        void cfathread_setproccnt( int ncnt ) {
-                assert( ncnt >= 1 );
-                adelete( procs );
-
-                proc_cnt = ncnt - 1;
-                procs = anew(proc_cnt);
-        }
-}
+        typedef struct cfathread_mutex * cfathread_mutex_t;
+
+        //--------------------
+        // Mutex
+        struct cfathread_mutex {
+                fast_lock impl;
+        };
+        int cfathread_mutex_init(cfathread_mutex_t *restrict mut, const cfathread_mutexattr_t *restrict) __attribute__((nonnull (1))) { *mut = new(); return 0; }
+        int cfathread_mutex_destroy(cfathread_mutex_t *mut) __attribute__((nonnull (1))) { delete( *mut ); return 0; }
+        int cfathread_mutex_lock   (cfathread_mutex_t *mut) __attribute__((nonnull (1))) { lock( (*mut)->impl ); return 0; }
+        int cfathread_mutex_unlock (cfathread_mutex_t *mut) __attribute__((nonnull (1))) { unlock( (*mut)->impl ); return 0; }
+        int cfathread_mutex_trylock(cfathread_mutex_t *mut) __attribute__((nonnull (1))) {
+                bool ret = try_lock( (*mut)->impl );
+                if( ret ) return 0;
+                else return EBUSY;
+        }
+
+        //--------------------
+        // Condition
+        struct cfathread_condition {
+                condition_variable(fast_lock) impl;
+        };
+        int cfathread_cond_init(cfathread_cond_t *restrict cond, const cfathread_condattr_t *restrict) __attribute__((nonnull (1))) { *cond = new(); return 0; }
+        int cfathread_cond_signal(cfathread_cond_t *cond) __attribute__((nonnull (1)))  { notify_one( (*cond)->impl ); return 0; }
+        int cfathread_cond_wait(cfathread_cond_t *restrict cond, cfathread_mutex_t *restrict mut) __attribute__((nonnull (1,2))) { wait( (*cond)->impl, (*mut)->impl ); return 0; }
+        int cfathread_cond_timedwait(cfathread_cond_t *restrict cond, cfathread_mutex_t *restrict mut, const struct timespec *restrict abstime) __attribute__((nonnull (1,2,3))) {
+                Time t = { *abstime };
+                if( wait( (*cond)->impl, (*mut)->impl, t ) ) {
+                        return 0;
+                }
+                errno = ETIMEDOUT;
+                return ETIMEDOUT;
+        }
+}
+
+#include <iofwd.hfa>
+
+extern "C" {
+        #include <unistd.h>
+        #include <sys/types.h>
+        #include <sys/socket.h>
+
+        //--------------------
+        // IO operations
+        int cfathread_socket(int domain, int type, int protocol) {
+                return socket(domain, type, protocol);
+        }
+        int cfathread_bind(int socket, const struct sockaddr *address, socklen_t address_len) {
+                return bind(socket, address, address_len);
+        }
+
+        int cfathread_listen(int socket, int backlog) {
+                return listen(socket, backlog);
+        }
+
+        int cfathread_accept(int socket, struct sockaddr *restrict address, socklen_t *restrict address_len) {
+                return cfa_accept4(socket, address, address_len, 0, CFA_IO_LAZY);
+        }
+
+        int cfathread_connect(int socket, const struct sockaddr *address, socklen_t address_len) {
+                return cfa_connect(socket, address, address_len, CFA_IO_LAZY);
+        }
+
+        int cfathread_dup(int fildes) {
+                return dup(fildes);
+        }
+
+        int cfathread_close(int fildes) {
+                return cfa_close(fildes, CFA_IO_LAZY);
+        }
+
+        ssize_t cfathread_sendmsg(int socket, const struct msghdr *message, int flags) {
+                return cfa_sendmsg(socket, message, flags, CFA_IO_LAZY);
+        }
+
+        ssize_t cfathread_write(int fildes, const void *buf, size_t nbyte) {
+                // Use send rather then write for socket since it's faster
+                return cfa_send(fildes, buf, nbyte, 0, CFA_IO_LAZY);
+        }
+
+        ssize_t cfathread_recvfrom(int socket, void *restrict buffer, size_t length, int flags, struct sockaddr *restrict address, socklen_t *restrict address_len)  {
+                struct iovec iov;
+                iov.iov_base = buffer;
+                iov.iov_len = length;
+
+                struct msghdr msg;
+                msg.msg_name = address;
+                msg.msg_namelen = address_len ? (socklen_t)*address_len : (socklen_t)0;
+                msg.msg_iov = &iov;
+                msg.msg_iovlen = 1;
+                msg.msg_control = 0p;
+                msg.msg_controllen = 0;
+
+                ssize_t ret = cfa_recvmsg(socket, &msg, flags, CFA_IO_LAZY);
+
+                if(address_len) *address_len = msg.msg_namelen;
+                return ret;
+        }
+
+        ssize_t cfathread_read(int fildes, void *buf, size_t nbyte) {
+                // Use recv rather then read for socket since it's faster
+                return cfa_recv(fildes, buf, nbyte, 0, CFA_IO_LAZY);
+        }
+
+}

libcfa/src/concurrency/clib/cfathread.h

@@ -14 +14 @@
 //
 
-#include "stddef.h"
-#include "invoke.h"
-
 #if defined(__cforall) || defined(__cplusplus)
 extern "C" {
 #endif
+        #include <asm/types.h>
+        #include <errno.h>
+        #include <unistd.h>
+
+
         //--------------------
         // Basic types
-        struct cfathread_CRunner_t;
-        typedef struct cfathread_CRunner_t * cfathread_t;
+
+        typedef struct cluster * cfathread_cluster_t;
+
+        int cfathread_cluster_create(cfathread_cluster_t * cluster);
+        cfathread_cluster_t cfathread_cluster_self(void);
+        int cfathread_cluster_print_stats(cfathread_cluster_t cluster);
+        int cfathread_cluster_add_worker(cfathread_cluster_t cluster, pthread_t* tid, void (*init_routine) (void *), void * arg);
+        int cfathread_cluster_pause (cfathread_cluster_t cluster);
+        int cfathread_cluster_resume(cfathread_cluster_t cluster);
 
         //--------------------
-        // Basic thread support
-        cfathread_t cfathread_create( void (*main)( cfathread_t ) );
-        void cfathread_join( cfathread_t );
+        // thread attribute
+        typedef struct cfathread_attr {
+                cfathread_cluster_t cl;
+        } cfathread_attr_t;
+
+        int cfathread_attr_init(cfathread_attr_t * attr) __attribute__((nonnull (1)));
+        static inline int cfathread_attr_destroy(cfathread_attr_t * attr) __attribute__((nonnull (1)));
+        static inline int cfathread_attr_destroy(cfathread_attr_t * attr) { return 0; }
+        static inline int cfathread_attr_setbackground(cfathread_attr_t * attr, int background) __attribute__((nonnull (1)));
+        static inline int cfathread_attr_setbackground(cfathread_attr_t * attr, int background) { return 0; }
+        static inline int cfathread_attr_setcluster(cfathread_attr_t * attr, cfathread_cluster_t cl) __attribute__((nonnull (1)));
+        static inline int cfathread_attr_setcluster(cfathread_attr_t * attr, cfathread_cluster_t cl) { attr->cl = cl; return 0; }
+
+        //--------------------
+        // thread type
+        struct cfathread_object;
+        typedef struct cfathread_object * cfathread_t;
+
+        int cfathread_create( cfathread_t * h, const cfathread_attr_t * a, void *(*main)( void * ), void * arg ) __attribute__((nonnull (1)));
+        int cfathread_join( cfathread_t, void ** retval );
+
+        int cfathread_get_errno(void);
+        cfathread_t cfathread_self(void);
+
+        int cfathread_usleep(useconds_t usecs);
+        int cfathread_sleep(unsigned int secs);
 
         void cfathread_park( void );
@@ -35 +67 @@
 
         //--------------------
-        // Basic kernel features
-        void cfathread_setproccnt( int );
+        // mutex and condition
+        struct timespec;
 
+        typedef struct cfathread_mutex_attr {
+        } cfathread_mutexattr_t;
+        typedef struct cfathread_mutex * cfathread_mutex_t;
+        int cfathread_mutex_init(cfathread_mutex_t *restrict mut, const cfathread_mutexattr_t *restrict attr) __attribute__((nonnull (1)));
+        int cfathread_mutex_destroy(cfathread_mutex_t *mut) __attribute__((nonnull (1)));
+        int cfathread_mutex_lock(cfathread_mutex_t *mut) __attribute__((nonnull (1)));
+        int cfathread_mutex_trylock(cfathread_mutex_t *mut) __attribute__((nonnull (1)));
+        int cfathread_mutex_unlock(cfathread_mutex_t *mut) __attribute__((nonnull (1)));
+
+        typedef struct cfathread_cond_attr {
+        } cfathread_condattr_t;
+        typedef struct cfathread_condition * cfathread_cond_t;
+        int cfathread_cond_init(cfathread_cond_t *restrict cond, const cfathread_condattr_t *restrict attr) __attribute__((nonnull (1)));
+        int cfathread_cond_wait(cfathread_cond_t *restrict cond, cfathread_mutex_t *restrict mut) __attribute__((nonnull (1,2)));
+        int cfathread_cond_timedwait(cfathread_cond_t *restrict cond, cfathread_mutex_t *restrict mut, const struct timespec *restrict abstime) __attribute__((nonnull (1,2,3)));
+        int cfathread_cond_signal(cfathread_cond_t *cond) __attribute__((nonnull (1)));
+
+        //--------------------
+        // IO operations
+        struct sockaddr;
+        struct msghdr;
+        int cfathread_socket(int domain, int type, int protocol);
+        int cfathread_bind(int socket, const struct sockaddr *address, socklen_t address_len);
+        int cfathread_listen(int socket, int backlog);
+        int cfathread_accept(int socket, struct sockaddr *restrict address, socklen_t *restrict address_len);
+        int cfathread_connect(int socket, const struct sockaddr *address, socklen_t address_len);
+        int cfathread_dup(int fildes);
+        int cfathread_close(int fildes);
+        ssize_t cfathread_sendmsg(int socket, const struct msghdr *message, int flags);
+        ssize_t cfathread_write(int fildes, const void *buf, size_t nbyte);
+        ssize_t cfathread_recvfrom(int socket, void *restrict buffer, size_t length, int flags, struct sockaddr *restrict address, socklen_t *restrict address_len);
+        ssize_t cfathread_read(int fildes, void *buf, size_t nbyte);
 
 #if defined(__cforall) || defined(__cplusplus)

libcfa/src/concurrency/coroutine.cfa

@@ -46 +46 @@
 
 //-----------------------------------------------------------------------------
-FORALL_DATA_INSTANCE(CoroutineCancelled, (coroutine_t &), (coroutine_t))
-
-forall(T &)
-void mark_exception(CoroutineCancelled(T) *) {}
-
 forall(T &)
 void copy(CoroutineCancelled(T) * dst, CoroutineCancelled(T) * src) {
@@ -65 +60 @@
 // This code should not be inlined. It is the error path on resume.
 forall(T & | is_coroutine(T))
-void __cfaehm_cancelled_coroutine( T & cor, $coroutine * desc ) {
+void __cfaehm_cancelled_coroutine(
+                T & cor, $coroutine * desc, EHM_DEFAULT_VTABLE(CoroutineCancelled, (T)) ) {
         verify( desc->cancellation );
         desc->state = Cancelled;
@@ -72 +68 @@
         // TODO: Remove explitate vtable set once trac#186 is fixed.
         CoroutineCancelled(T) except;
-        except.virtual_table = &get_exception_vtable(&except);
+        except.virtual_table = &_default_vtable;
         except.the_coroutine = &cor;
         except.the_exception = except;
-        throwResume except;
+        // Why does this need a cast?
+        throwResume (CoroutineCancelled(T) &)except;
 
         except->virtual_table->free( except );
@@ -148 +145 @@
 // Part of the Public API
 // Not inline since only ever called once per coroutine
-forall(T & | is_coroutine(T))
+forall(T & | is_coroutine(T) | { EHM_DEFAULT_VTABLE(CoroutineCancelled, (T)); })
 void prime(T& cor) {
         $coroutine* this = get_coroutine(cor);
@@ -196 +193 @@
 
 void __stack_clean  ( __stack_info_t * this ) {
-        size_t size = ((intptr_t)this->storage->base) - ((intptr_t)this->storage->limit) + sizeof(__stack_t);
         void * storage = this->storage->limit;
 
         #if CFA_COROUTINE_USE_MMAP
+                size_t size = ((intptr_t)this->storage->base) - ((intptr_t)this->storage->limit) + sizeof(__stack_t);
                 storage = (void *)(((intptr_t)storage) - __page_size);
                 if(munmap(storage, size + __page_size) == -1) {

libcfa/src/concurrency/coroutine.hfa

@@ -22 +22 @@
 //-----------------------------------------------------------------------------
 // Exception thrown from resume when a coroutine stack is cancelled.
-FORALL_DATA_EXCEPTION(CoroutineCancelled, (coroutine_t &), (coroutine_t)) (
+EHM_FORALL_EXCEPTION(CoroutineCancelled, (coroutine_t &), (coroutine_t)) (
         coroutine_t * the_coroutine;
         exception_t * the_exception;
@@ -60 +60 @@
 //-----------------------------------------------------------------------------
 // Public coroutine API
-forall(T & | is_coroutine(T))
+forall(T & | is_coroutine(T) | { EHM_DEFAULT_VTABLE(CoroutineCancelled, (T)); })
 void prime(T & cor);
 
@@ -130 +130 @@
 
 forall(T & | is_coroutine(T))
-void __cfaehm_cancelled_coroutine( T & cor, $coroutine * desc );
+void __cfaehm_cancelled_coroutine(
+        T & cor, $coroutine * desc, EHM_DEFAULT_VTABLE(CoroutineCancelled, (T)) );
 
 // Resume implementation inlined for performance
-forall(T & | is_coroutine(T))
+forall(T & | is_coroutine(T) | { EHM_DEFAULT_VTABLE(CoroutineCancelled, (T)); })
 static inline T & resume(T & cor) {
         // optimization : read TLS once and reuse it
@@ -163 +164 @@
         $ctx_switch( src, dst );
         if ( unlikely(dst->cancellation) ) {
-                __cfaehm_cancelled_coroutine( cor, dst );
+                __cfaehm_cancelled_coroutine( cor, dst, _default_vtable );
         }
 

libcfa/src/concurrency/future.hfa

@@ -37 +37 @@
 
                 // Fulfil the future, returns whether or not someone was unblocked
-                bool fulfil( future(T) & this, T result ) {
+                $thread * fulfil( future(T) & this, T result ) {
                         this.result = result;
                         return fulfil( (future_t&)this );
@@ -96 +96 @@
                 bool fulfil( multi_future(T) & this, T result ) {
                         this.result = result;
-                        return fulfil( (future_t&)this );
+                        return fulfil( (future_t&)this ) != 0p;
                 }
 

libcfa/src/concurrency/invoke.c

@@ -34 +34 @@
 
 extern void disable_interrupts() OPTIONAL_THREAD;
-extern void enable_interrupts( __cfaabi_dbg_ctx_param );
+extern void enable_interrupts( _Bool poll );
 
 void __cfactx_invoke_coroutine(
@@ -82 +82 @@
 ) {
         // Officially start the thread by enabling preemption
-        enable_interrupts( __cfaabi_dbg_ctx );
+        enable_interrupts( true );
 
         // Call the main of the thread

libcfa/src/concurrency/invoke.h

@@ -148 +148 @@
                 struct $thread * prev;
                 volatile unsigned long long ts;
-                int preferred;
+                unsigned preferred;
         };
 
@@ -200 +200 @@
                 } node;
 
+                struct processor * last_proc;
+
                 #if defined( __CFA_WITH_VERIFY__ )
                         void * canary;
@@ -224 +226 @@
                 }
 
+                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;

libcfa/src/concurrency/io.cfa

@@ -32 +32 @@
         extern "C" {
                 #include <sys/syscall.h>
+                #include <sys/eventfd.h>
 
                 #include <linux/io_uring.h>
@@ -39 +40 @@
         #include "kernel.hfa"
         #include "kernel/fwd.hfa"
+        #include "kernel_private.hfa"
         #include "io/types.hfa"
 
@@ -79 +81 @@
         };
 
-        // returns true of acquired as leader or second leader
-        static inline bool try_lock( __leaderlock_t & this ) {
-                const uintptr_t thrd = 1z | (uintptr_t)active_thread();
-                bool block;
-                disable_interrupts();
-                for() {
-                        struct $thread * expected = this.value;
-                        if( 1p != expected && 0p != expected ) {
-                                /* paranoid */ verify( thrd != (uintptr_t)expected ); // We better not already be the next leader
-                                enable_interrupts( __cfaabi_dbg_ctx );
-                                return false;
-                        }
-                        struct $thread * desired;
-                        if( 0p == expected ) {
-                                // If the lock isn't locked acquire it, no need to block
-                                desired = 1p;
-                                block = false;
-                        }
-                        else {
-                                // If the lock is already locked try becomming the next leader
-                                desired = (struct $thread *)thrd;
-                                block = true;
-                        }
-                        if( __atomic_compare_exchange_n(&this.value, &expected, desired, false, __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST) ) break;
-                }
-                if( block ) {
-                        enable_interrupts( __cfaabi_dbg_ctx );
-                        park();
-                        disable_interrupts();
-                }
-                return true;
-        }
-
-        static inline bool next( __leaderlock_t & this ) {
+        static $io_context * __ioarbiter_allocate( $io_arbiter & this, __u32 idxs[], __u32 want );
+        static void __ioarbiter_submit( $io_context * , __u32 idxs[], __u32 have, bool lazy );
+        static void __ioarbiter_flush ( $io_context & );
+        static inline void __ioarbiter_notify( $io_context & ctx );
+//=============================================================================================
+// I/O Polling
+//=============================================================================================
+        static inline unsigned __flush( struct $io_context & );
+        static inline __u32 __release_sqes( struct $io_context & );
+        extern void __kernel_unpark( $thread * thrd );
+
+        bool __cfa_io_drain( processor * proc ) {
                 /* paranoid */ verify( ! __preemption_enabled() );
-                struct $thread * nextt;
-                for() {
-                        struct $thread * expected = this.value;
-                        /* paranoid */ verify( (1 & (uintptr_t)expected) == 1 ); // The lock better be locked
-
-                        struct $thread * desired;
-                        if( 1p == expected ) {
-                                // No next leader, just unlock
-                                desired = 0p;
-                                nextt   = 0p;
-                        }
-                        else {
-                                // There is a next leader, remove but keep locked
-                                desired = 1p;
-                                nextt   = (struct $thread *)(~1z & (uintptr_t)expected);
-                        }
-                        if( __atomic_compare_exchange_n(&this.value, &expected, desired, false, __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST) ) break;
-                }
-
-                if(nextt) {
-                        unpark( nextt );
-                        enable_interrupts( __cfaabi_dbg_ctx );
-                        return true;
-                }
-                enable_interrupts( __cfaabi_dbg_ctx );
-                return false;
-        }
-
-//=============================================================================================
-// I/O Syscall
-//=============================================================================================
-        static int __io_uring_enter( struct __io_data & ring, unsigned to_submit, bool get ) {
-                bool need_sys_to_submit = false;
-                bool need_sys_to_complete = false;
-                unsigned flags = 0;
-
-                TO_SUBMIT:
-                if( to_submit > 0 ) {
-                        if( !(ring.ring_flags & IORING_SETUP_SQPOLL) ) {
-                                need_sys_to_submit = true;
-                                break TO_SUBMIT;
-                        }
-                        if( (*ring.submit_q.flags) & IORING_SQ_NEED_WAKEUP ) {
-                                need_sys_to_submit = true;
-                                flags |= IORING_ENTER_SQ_WAKEUP;
-                        }
-                }
-
-                if( get && !(ring.ring_flags & IORING_SETUP_SQPOLL) ) {
-                        flags |= IORING_ENTER_GETEVENTS;
-                        if( (ring.ring_flags & IORING_SETUP_IOPOLL) ) {
-                                need_sys_to_complete = true;
-                        }
-                }
-
-                int ret = 0;
-                if( need_sys_to_submit || need_sys_to_complete ) {
-                        __cfadbg_print_safe(io_core, "Kernel I/O : IO_URING enter %d %u %u\n", ring.fd, to_submit, flags);
-                        ret = syscall( __NR_io_uring_enter, ring.fd, to_submit, 0, flags, (sigset_t *)0p, _NSIG / 8);
-                        __cfadbg_print_safe(io_core, "Kernel I/O : IO_URING %d returned %d\n", ring.fd, ret);
-
-                        if( ret < 0 ) {
-                                switch((int)errno) {
-                                case EAGAIN:
-                                case EINTR:
-                                case EBUSY:
-                                        ret = -1;
-                                        break;
-                                default:
-                                        abort( "KERNEL ERROR: IO_URING SYSCALL - (%d) %s\n", (int)errno, strerror(errno) );
-                                }
-                        }
-                }
-
-                // Memory barrier
-                __atomic_thread_fence( __ATOMIC_SEQ_CST );
-                return ret;
-        }
-
-//=============================================================================================
-// I/O Polling
-//=============================================================================================
-        static unsigned __collect_submitions( struct __io_data & ring );
-        static __u32 __release_consumed_submission( struct __io_data & ring );
-        static inline void __clean( volatile struct io_uring_sqe * sqe );
-
-        // Process a single completion message from the io_uring
-        // This is NOT thread-safe
-        static inline void process( volatile struct io_uring_cqe & 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 );
-
-                fulfil( *future, cqe.res );
-        }
-
-        static [int, bool] __drain_io( & struct __io_data ring ) {
-                /* paranoid */ verify( ! __preemption_enabled() );
-
-                unsigned to_submit = 0;
-                if( ring.poller_submits ) {
-                        // If the poller thread also submits, then we need to aggregate the submissions which are ready
-                        to_submit = __collect_submitions( ring );
-                }
-
-                int ret = __io_uring_enter(ring, to_submit, true);
-                if( ret < 0 ) {
-                        return [0, true];
-                }
-
-                // update statistics
-                if (to_submit > 0) {
-                        __STATS__( true,
-                                if( to_submit > 0 ) {
-                                        io.submit_q.submit_avg.rdy += to_submit;
-                                        io.submit_q.submit_avg.csm += ret;
-                                        io.submit_q.submit_avg.cnt += 1;
-                                }
-                        )
-                }
-
-                __atomic_thread_fence( __ATOMIC_SEQ_CST );
-
-                // Release the consumed SQEs
-                __release_consumed_submission( ring );
+                /* paranoid */ verify( ready_schedule_islocked() );
+                /* paranoid */ verify( proc );
+                /* paranoid */ verify( proc->io.ctx );
 
                 // Drain the queue
-                unsigned head = *ring.completion_q.head;
-                unsigned tail = *ring.completion_q.tail;
-                const __u32 mask = *ring.completion_q.mask;
-
-                // Nothing was new return 0
-                if (head == tail) {
-                        return [0, to_submit > 0];
-                }
+                $io_context * ctx = proc->io.ctx;
+                unsigned head = *ctx->cq.head;
+                unsigned tail = *ctx->cq.tail;
+                const __u32 mask = *ctx->cq.mask;
 
                 __u32 count = tail - head;
-                /* paranoid */ verify( count != 0 );
+                __STATS__( false, io.calls.drain++; io.calls.completed += count; )
+
+                if(count == 0) return false;
+
                 for(i; count) {
                         unsigned idx = (head + i) & mask;
-                        volatile struct io_uring_cqe & cqe = ring.completion_q.cqes[idx];
+                        volatile struct io_uring_cqe & cqe = ctx->cq.cqes[idx];
 
                         /* paranoid */ verify(&cqe);
 
-                        process( 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 ) );
+                }
+
+                __cfadbg_print_safe(io, "Kernel I/O : %u completed\n", count);
 
                 // 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_fetch_add( ring.completion_q.head, count, __ATOMIC_SEQ_CST );
-
-                return [count, count > 0 || to_submit > 0];
-        }
-
-        void main( $io_ctx_thread & this ) {
-                __ioctx_register( this );
-
-                __cfadbg_print_safe(io_core, "Kernel I/O : IO poller %d (%p) ready\n", this.ring->fd, &this);
-
-                const int reset_cnt = 5;
-                int reset = reset_cnt;
-                // Then loop until we need to start
-                LOOP:
-                while(!__atomic_load_n(&this.done, __ATOMIC_SEQ_CST)) {
-                        // Drain the io
-                        int count;
-                        bool again;
-                        disable_interrupts();
-                                [count, again] = __drain_io( *this.ring );
-
-                                if(!again) reset--;
-
+                __atomic_store_n( ctx->cq.head, head + count, __ATOMIC_SEQ_CST );
+
+                /* paranoid */ verify( ready_schedule_islocked() );
+                /* paranoid */ verify( ! __preemption_enabled() );
+
+                return true;
+        }
+
+        void __cfa_io_flush( processor * proc ) {
+                /* paranoid */ verify( ! __preemption_enabled() );
+                /* paranoid */ verify( proc );
+                /* paranoid */ verify( proc->io.ctx );
+
+                $io_context & ctx = *proc->io.ctx;
+
+                __ioarbiter_flush( ctx );
+
+                __STATS__( true, io.calls.flush++; )
+                int ret = syscall( __NR_io_uring_enter, ctx.fd, ctx.sq.to_submit, 0, 0, (sigset_t *)0p, _NSIG / 8);
+                if( ret < 0 ) {
+                        switch((int)errno) {
+                        case EAGAIN:
+                        case EINTR:
+                        case EBUSY:
                                 // Update statistics
-                                __STATS__( true,
-                                        io.complete_q.completed_avg.val += count;
-                                        io.complete_q.completed_avg.cnt += 1;
-                                )
-                        enable_interrupts( __cfaabi_dbg_ctx );
-
-                        // If we got something, just yield and check again
-                        if(reset > 1) {
-                                yield();
-                                continue LOOP;
+                                __STATS__( false, io.calls.errors.busy ++; )
+                                return;
+                        default:
+                                abort( "KERNEL ERROR: IO_URING SYSCALL - (%d) %s\n", (int)errno, strerror(errno) );
                         }
-
-                        // We alread failed to find completed entries a few time.
-                        if(reset == 1) {
-                                // Rearm the context so it can block
-                                // but don't block right away
-                                // we need to retry one last time in case
-                                // something completed *just now*
-                                __ioctx_prepare_block( this );
-                                continue LOOP;
-                        }
-
-                                __STATS__( false,
-                                        io.complete_q.blocks += 1;
-                                )
-                                __cfadbg_print_safe(io_core, "Kernel I/O : Parking io poller %d (%p)\n", this.ring->fd, &this);
-
-                                // block this thread
-                                wait( this.sem );
-
-                        // restore counter
-                        reset = reset_cnt;
-                }
-
-                __cfadbg_print_safe(io_core, "Kernel I/O : Fast poller %d (%p) stopping\n", this.ring->fd, &this);
-
-                __ioctx_unregister( this );
+                }
+
+                __cfadbg_print_safe(io, "Kernel I/O : %u submitted to io_uring %d\n", ret, ctx.fd);
+                __STATS__( true, io.calls.submitted += ret; )
+                /* paranoid */ verify( ctx.sq.to_submit <= *ctx.sq.num );
+                /* paranoid */ verify( ctx.sq.to_submit >= ret );
+
+                ctx.sq.to_submit -= ret;
+
+                /* paranoid */ verify( ctx.sq.to_submit <= *ctx.sq.num );
+
+                // Release the consumed SQEs
+                __release_sqes( ctx );
+
+                /* paranoid */ verify( ! __preemption_enabled() );
+
+                ctx.proc->io.pending = false;
         }
 
@@ -344 +193 @@
 //         head and tail must be fully filled and shouldn't ever be touched again.
 //
+        //=============================================================================================
+        // Allocation
+        // for user's convenience fill the sqes from the indexes
+        static inline void __fill(struct io_uring_sqe * out_sqes[], __u32 want, __u32 idxs[], struct $io_context * ctx)  {
+                struct io_uring_sqe * sqes = ctx->sq.sqes;
+                for(i; want) {
+                        __cfadbg_print_safe(io, "Kernel I/O : filling loop\n");
+                        out_sqes[i] = &sqes[idxs[i]];
+                }
+        }
+
+        // Try to directly allocate from the a given context
+        // Not thread-safe
+        static inline bool __alloc(struct $io_context * ctx, __u32 idxs[], __u32 want) {
+                __sub_ring_t & sq = ctx->sq;
+                const __u32 mask  = *sq.mask;
+                __u32 fhead = sq.free_ring.head;    // get the current head of the queue
+                __u32 ftail = sq.free_ring.tail;    // get the current tail of the queue
+
+                // If we don't have enough sqes, fail
+                if((ftail - fhead) < want) { return false; }
+
+                // copy all the indexes we want from the available list
+                for(i; want) {
+                        __cfadbg_print_safe(io, "Kernel I/O : allocating loop\n");
+                        idxs[i] = sq.free_ring.array[(fhead + i) & mask];
+                }
+
+                // Advance the head to mark the indexes as consumed
+                __atomic_store_n(&sq.free_ring.head, fhead + want, __ATOMIC_RELEASE);
+
+                // return success
+                return true;
+        }
 
         // Allocate an submit queue entry.
@@ -350 +233 @@
         // for convenience, return both the index and the pointer to the sqe
         // sqe == &sqes[idx]
-        [* volatile struct io_uring_sqe, __u32] __submit_alloc( struct __io_data & ring, __u64 data ) {
-                /* paranoid */ verify( data != 0 );
-
-                // Prepare the data we need
-                __attribute((unused)) int len   = 0;
-                __attribute((unused)) int block = 0;
-                __u32 cnt = *ring.submit_q.num;
-                __u32 mask = *ring.submit_q.mask;
-
-                __u32 off = thread_rand();
-
-                // Loop around looking for an available spot
-                for() {
-                        // Look through the list starting at some offset
-                        for(i; cnt) {
-                                __u64 expected = 3;
-                                __u32 idx = (i + off) & mask; // Get an index from a random
-                                volatile struct io_uring_sqe * sqe = &ring.submit_q.sqes[idx];
-                                volatile __u64 * udata = &sqe->user_data;
-
-                                // Allocate the entry by CASing the user_data field from 0 to the future address
-                                if( *udata == expected &&
-                                        __atomic_compare_exchange_n( udata, &expected, data, true, __ATOMIC_SEQ_CST, __ATOMIC_RELAXED ) )
-                                {
-                                        // update statistics
-                                        __STATS__( false,
-                                                io.submit_q.alloc_avg.val   += len;
-                                                io.submit_q.alloc_avg.block += block;
-                                                io.submit_q.alloc_avg.cnt   += 1;
-                                        )
-
-                                        // debug log
-                                        __cfadbg_print_safe( io, "Kernel I/O : allocated [%p, %u] for %p (%p)\n", sqe, idx, active_thread(), (void*)data );
-
-                                        // Success return the data
-                                        return [sqe, idx];
-                                }
-                                verify(expected != data);
-
-                                // This one was used
-                                len ++;
-                        }
-
-                        block++;
-
-                        yield();
-                }
-        }
-
-        static inline __u32 __submit_to_ready_array( struct __io_data & ring, __u32 idx, const __u32 mask ) {
-                /* paranoid */ verify( idx <= mask   );
-                /* paranoid */ verify( idx != -1ul32 );
-
-                // We need to find a spot in the ready array
-                __attribute((unused)) int len   = 0;
-                __attribute((unused)) int block = 0;
-                __u32 ready_mask = ring.submit_q.ready_cnt - 1;
-
-                __u32 off = thread_rand();
-
-                __u32 picked;
-                LOOKING: for() {
-                        for(i; ring.submit_q.ready_cnt) {
-                                picked = (i + off) & ready_mask;
-                                __u32 expected = -1ul32;
-                                if( __atomic_compare_exchange_n( &ring.submit_q.ready[picked], &expected, idx, true, __ATOMIC_SEQ_CST, __ATOMIC_RELAXED ) ) {
-                                        break LOOKING;
-                                }
-                                verify(expected != idx);
-
-                                len ++;
-                        }
-
-                        block++;
-
-                        __u32 released = __release_consumed_submission( ring );
-                        if( released == 0 ) {
-                                yield();
-                        }
-                }
-
-                // update statistics
-                __STATS__( false,
-                        io.submit_q.look_avg.val   += len;
-                        io.submit_q.look_avg.block += block;
-                        io.submit_q.look_avg.cnt   += 1;
-                )
-
-                return picked;
-        }
-
-        void __submit( struct io_context * ctx, __u32 idx ) __attribute__((nonnull (1))) {
-                __io_data & ring = *ctx->thrd.ring;
-
+        struct $io_context * cfa_io_allocate(struct io_uring_sqe * sqes[], __u32 idxs[], __u32 want) {
+                __cfadbg_print_safe(io, "Kernel I/O : attempting to allocate %u\n", want);
+
+                disable_interrupts();
+                processor * proc = __cfaabi_tls.this_processor;
+                $io_context * ctx = proc->io.ctx;
+                /* paranoid */ verify( __cfaabi_tls.this_processor );
+                /* paranoid */ verify( ctx );
+
+                __cfadbg_print_safe(io, "Kernel I/O : attempting to fast allocation\n");
+
+                // We can proceed to the fast path
+                if( __alloc(ctx, idxs, want) ) {
+                        // Allocation was successful
+                        __STATS__( true, io.alloc.fast += 1; )
+                        enable_interrupts();
+
+                        __cfadbg_print_safe(io, "Kernel I/O : fast allocation successful from ring %d\n", ctx->fd);
+
+                        __fill( sqes, want, idxs, ctx );
+                        return ctx;
+                }
+                // The fast path failed, fallback
+                __STATS__( true, io.alloc.fail += 1; )
+
+                // Fast path failed, fallback on arbitration
+                __STATS__( true, io.alloc.slow += 1; )
+                enable_interrupts();
+
+                $io_arbiter * ioarb = proc->cltr->io.arbiter;
+                /* paranoid */ verify( ioarb );
+
+                __cfadbg_print_safe(io, "Kernel I/O : falling back on arbiter for allocation\n");
+
+                struct $io_context * ret = __ioarbiter_allocate(*ioarb, idxs, want);
+
+                __cfadbg_print_safe(io, "Kernel I/O : slow allocation completed from ring %d\n", ret->fd);
+
+                __fill( sqes, want, idxs,ret );
+                return ret;
+        }
+
+
+        //=============================================================================================
+        // submission
+        static inline void __submit( struct $io_context * ctx, __u32 idxs[], __u32 have, bool lazy) {
+                // We can proceed to the fast path
+                // Get the right objects
+                __sub_ring_t & sq = ctx->sq;
+                const __u32 mask  = *sq.mask;
+                __u32 tail = *sq.kring.tail;
+
+                // Add the sqes to the array
+                for( i; have ) {
+                        __cfadbg_print_safe(io, "Kernel I/O : __submit loop\n");
+                        sq.kring.array[ (tail + i) & mask ] = idxs[i];
+                }
+
+                // Make the sqes visible to the submitter
+                __atomic_store_n(sq.kring.tail, tail + have, __ATOMIC_RELEASE);
+                sq.to_submit++;
+
+                ctx->proc->io.pending = true;
+                ctx->proc->io.dirty   = true;
+                if(sq.to_submit > 30 || !lazy) {
+                        __cfa_io_flush( ctx->proc );
+                }
+        }
+
+        void cfa_io_submit( struct $io_context * inctx, __u32 idxs[], __u32 have, bool lazy ) __attribute__((nonnull (1))) {
+                __cfadbg_print_safe(io, "Kernel I/O : attempting to submit %u (%s)\n", have, lazy ? "lazy" : "eager");
+
+                disable_interrupts();
+                processor * proc = __cfaabi_tls.this_processor;
+                $io_context * ctx = proc->io.ctx;
+                /* paranoid */ verify( __cfaabi_tls.this_processor );
+                /* paranoid */ verify( ctx );
+
+                // Can we proceed to the fast path
+                if( ctx == inctx )              // We have the right instance?
                 {
-                        __attribute__((unused)) volatile struct io_uring_sqe * sqe = &ring.submit_q.sqes[idx];
-                        __cfadbg_print_safe( io,
-                                "Kernel I/O : submitting %u (%p) for %p\n"
-                                "    data: %p\n"
-                                "    opcode: %s\n"
-                                "    fd: %d\n"
-                                "    flags: %d\n"
-                                "    prio: %d\n"
-                                "    off: %p\n"
-                                "    addr: %p\n"
-                                "    len: %d\n"
-                                "    other flags: %d\n"
-                                "    splice fd: %d\n"
-                                "    pad[0]: %llu\n"
-                                "    pad[1]: %llu\n"
-                                "    pad[2]: %llu\n",
-                                idx, sqe,
-                                active_thread(),
-                                (void*)sqe->user_data,
-                                opcodes[sqe->opcode],
-                                sqe->fd,
-                                sqe->flags,
-                                sqe->ioprio,
-                                (void*)sqe->off,
-                                (void*)sqe->addr,
-                                sqe->len,
-                                sqe->accept_flags,
-                                sqe->splice_fd_in,
-                                sqe->__pad2[0],
-                                sqe->__pad2[1],
-                                sqe->__pad2[2]
-                        );
-                }
-
-
-                // Get now the data we definetely need
-                volatile __u32 * const tail = ring.submit_q.tail;
-                const __u32 mask  = *ring.submit_q.mask;
-
-                // There are 2 submission schemes, check which one we are using
-                if( ring.poller_submits ) {
-                        // If the poller thread submits, then we just need to add this to the ready array
-                        __submit_to_ready_array( ring, idx, mask );
-
-                        post( ctx->thrd.sem );
-
-                        __cfadbg_print_safe( io, "Kernel I/O : Added %u to ready for %p\n", idx, active_thread() );
-                }
-                else if( ring.eager_submits ) {
-                        __attribute__((unused)) __u32 picked = __submit_to_ready_array( ring, idx, mask );
-
-                        #if defined(LEADER_LOCK)
-                                if( !try_lock(ring.submit_q.submit_lock) ) {
-                                        __STATS__( false,
-                                                io.submit_q.helped += 1;
-                                        )
-                                        return;
-                                }
-                                /* paranoid */ verify( ! __preemption_enabled() );
-                                __STATS__( true,
-                                        io.submit_q.leader += 1;
-                                )
-                        #else
-                                for() {
-                                        yield();
-
-                                        if( try_lock(ring.submit_q.submit_lock __cfaabi_dbg_ctx2) ) {
-                                                __STATS__( false,
-                                                        io.submit_q.leader += 1;
-                                                )
-                                                break;
-                                        }
-
-                                        // If some one else collected our index, we are done
-                                        #warning ABA problem
-                                        if( ring.submit_q.ready[picked] != idx ) {
-                                                __STATS__( false,
-                                                        io.submit_q.helped += 1;
-                                                )
-                                                return;
-                                        }
-
-                                        __STATS__( false,
-                                                io.submit_q.busy += 1;
-                                        )
-                                }
-                        #endif
-
-                        // We got the lock
-                        // Collect the submissions
-                        unsigned to_submit = __collect_submitions( ring );
-
-                        // Actually submit
-                        int ret = __io_uring_enter( ring, to_submit, false );
-
-                        #if defined(LEADER_LOCK)
-                                /* paranoid */ verify( ! __preemption_enabled() );
-                                next(ring.submit_q.submit_lock);
-                        #else
-                                unlock(ring.submit_q.submit_lock);
-                        #endif
-                        if( ret < 0 ) {
-                                return;
-                        }
-
-                        // Release the consumed SQEs
-                        __release_consumed_submission( ring );
-
-                        // update statistics
-                        __STATS__( false,
-                                io.submit_q.submit_avg.rdy += to_submit;
-                                io.submit_q.submit_avg.csm += ret;
-                                io.submit_q.submit_avg.cnt += 1;
-                        )
-
-                        __cfadbg_print_safe( io, "Kernel I/O : submitted %u (among %u) for %p\n", idx, ret, active_thread() );
-                }
-                else
-                {
-                        // get mutual exclusion
-                        #if defined(LEADER_LOCK)
-                                while(!try_lock(ring.submit_q.submit_lock));
-                        #else
-                                lock(ring.submit_q.submit_lock __cfaabi_dbg_ctx2);
-                        #endif
-
-                        /* paranoid */ verifyf( ring.submit_q.sqes[ idx ].user_data != 3ul64,
-                        /* paranoid */  "index %u already reclaimed\n"
-                        /* paranoid */  "head %u, prev %u, tail %u\n"
-                        /* paranoid */  "[-0: %u,-1: %u,-2: %u,-3: %u]\n",
-                        /* paranoid */  idx,
-                        /* paranoid */  *ring.submit_q.head, ring.submit_q.prev_head, *tail
-                        /* paranoid */  ,ring.submit_q.array[ ((*ring.submit_q.head) - 0) & (*ring.submit_q.mask) ]
-                        /* paranoid */  ,ring.submit_q.array[ ((*ring.submit_q.head) - 1) & (*ring.submit_q.mask) ]
-                        /* paranoid */  ,ring.submit_q.array[ ((*ring.submit_q.head) - 2) & (*ring.submit_q.mask) ]
-                        /* paranoid */  ,ring.submit_q.array[ ((*ring.submit_q.head) - 3) & (*ring.submit_q.mask) ]
-                        /* paranoid */ );
-
-                        // Append to the list of ready entries
-
-                        /* paranoid */ verify( idx <= mask );
-                        ring.submit_q.array[ (*tail) & mask ] = idx;
-                        __atomic_fetch_add(tail, 1ul32, __ATOMIC_SEQ_CST);
-
-                        // Submit however, many entries need to be submitted
-                        int ret = __io_uring_enter( ring, 1, false );
-                        if( ret < 0 ) {
-                                switch((int)errno) {
-                                default:
-                                        abort( "KERNEL ERROR: IO_URING SUBMIT - %s\n", strerror(errno) );
-                                }
-                        }
-
-                        /* paranoid */ verify(ret == 1);
-
-                        // update statistics
-                        __STATS__( false,
-                                io.submit_q.submit_avg.csm += 1;
-                                io.submit_q.submit_avg.cnt += 1;
-                        )
-
-                        {
-                                __attribute__((unused)) volatile __u32 * const head = ring.submit_q.head;
-                                __attribute__((unused)) __u32 last_idx = ring.submit_q.array[ ((*head) - 1) & mask ];
-                                __attribute__((unused)) volatile struct io_uring_sqe * sqe = &ring.submit_q.sqes[last_idx];
-
-                                __cfadbg_print_safe( io,
-                                        "Kernel I/O : last submitted is %u (%p)\n"
-                                        "    data: %p\n"
-                                        "    opcode: %s\n"
-                                        "    fd: %d\n"
-                                        "    flags: %d\n"
-                                        "    prio: %d\n"
-                                        "    off: %p\n"
-                                        "    addr: %p\n"
-                                        "    len: %d\n"
-                                        "    other flags: %d\n"
-                                        "    splice fd: %d\n"
-                                        "    pad[0]: %llu\n"
-                                        "    pad[1]: %llu\n"
-                                        "    pad[2]: %llu\n",
-                                        last_idx, sqe,
-                                        (void*)sqe->user_data,
-                                        opcodes[sqe->opcode],
-                                        sqe->fd,
-                                        sqe->flags,
-                                        sqe->ioprio,
-                                        (void*)sqe->off,
-                                        (void*)sqe->addr,
-                                        sqe->len,
-                                        sqe->accept_flags,
-                                        sqe->splice_fd_in,
-                                        sqe->__pad2[0],
-                                        sqe->__pad2[1],
-                                        sqe->__pad2[2]
-                                );
-                        }
-
-                        __atomic_thread_fence( __ATOMIC_SEQ_CST );
-                        // Release the consumed SQEs
-
-                        __release_consumed_submission( ring );
-                        // ring.submit_q.sqes[idx].user_data = 3ul64;
-
-                        #if defined(LEADER_LOCK)
-                                next(ring.submit_q.submit_lock);
-                        #else
-                                unlock(ring.submit_q.submit_lock);
-                        #endif
-
-                        __cfadbg_print_safe( io, "Kernel I/O : submitted %u for %p\n", idx, active_thread() );
-                }
-        }
-
-        // #define PARTIAL_SUBMIT 32
-
-        // go through the list of submissions in the ready array and moved them into
-        // the ring's submit queue
-        static unsigned __collect_submitions( struct __io_data & ring ) {
-                /* paranoid */ verify( ring.submit_q.ready != 0p );
-                /* paranoid */ verify( ring.submit_q.ready_cnt > 0 );
-
-                unsigned to_submit = 0;
-                __u32 tail = *ring.submit_q.tail;
-                const __u32 mask = *ring.submit_q.mask;
-                #if defined(PARTIAL_SUBMIT)
-                        #if defined(LEADER_LOCK)
-                                #error PARTIAL_SUBMIT and LEADER_LOCK cannot co-exist
-                        #endif
-                        const __u32 cnt = ring.submit_q.ready_cnt > PARTIAL_SUBMIT ? PARTIAL_SUBMIT : ring.submit_q.ready_cnt;
-                        const __u32 offset = ring.submit_q.prev_ready;
-                        ring.submit_q.prev_ready += cnt;
-                #else
-                        const __u32 cnt = ring.submit_q.ready_cnt;
-                        const __u32 offset = 0;
-                #endif
-
-                // Go through the list of ready submissions
-                for( c; cnt ) {
-                        __u32 i = (offset + c) % ring.submit_q.ready_cnt;
-
-                        // replace any submission with the sentinel, to consume it.
-                        __u32 idx = __atomic_exchange_n( &ring.submit_q.ready[i], -1ul32, __ATOMIC_RELAXED);
-
-                        // If it was already the sentinel, then we are done
-                        if( idx == -1ul32 ) continue;
-
-                        // If we got a real submission, append it to the list
-                        ring.submit_q.array[ (tail + to_submit) & mask ] = idx & mask;
-                        to_submit++;
-                }
-
-                // Increment the tail based on how many we are ready to submit
-                __atomic_fetch_add(ring.submit_q.tail, to_submit, __ATOMIC_SEQ_CST);
-
-                return to_submit;
-        }
-
+                        __submit(ctx, idxs, have, lazy);
+
+                        // Mark the instance as no longer in-use, re-enable interrupts and return
+                        __STATS__( true, io.submit.fast += 1; )
+                        enable_interrupts();
+
+                        __cfadbg_print_safe(io, "Kernel I/O : submitted on fast path\n");
+                        return;
+                }
+
+                // Fast path failed, fallback on arbitration
+                __STATS__( true, io.submit.slow += 1; )
+                enable_interrupts();
+
+                __cfadbg_print_safe(io, "Kernel I/O : falling back on arbiter for submission\n");
+
+                __ioarbiter_submit(inctx, idxs, have, lazy);
+        }
+
+        //=============================================================================================
+        // Flushing
         // Go through the ring's submit queue and release everything that has already been consumed
         // by io_uring
-        static __u32 __release_consumed_submission( struct __io_data & ring ) {
-                const __u32 smask = *ring.submit_q.mask;
-
-                // We need to get the lock to copy the old head and new head
-                if( !try_lock(ring.submit_q.release_lock __cfaabi_dbg_ctx2) ) return 0;
+        // This cannot be done by multiple threads
+        static __u32 __release_sqes( struct $io_context & ctx ) {
+                const __u32 mask = *ctx.sq.mask;
+
                 __attribute__((unused))
-                __u32 ctail = *ring.submit_q.tail;        // get the current tail of the queue
-                __u32 chead = *ring.submit_q.head;              // get the current head of the queue
-                __u32 phead = ring.submit_q.prev_head;  // get the head the last time we were here
-                ring.submit_q.prev_head = chead;                // note up to were we processed
-                unlock(ring.submit_q.release_lock);
+                __u32 ctail = *ctx.sq.kring.tail;    // get the current tail of the queue
+                __u32 chead = *ctx.sq.kring.head;        // get the current head of the queue
+                __u32 phead = ctx.sq.kring.released; // get the head the last time we were here
+
+                __u32 ftail = ctx.sq.free_ring.tail;  // get the current tail of the queue
 
                 // the 3 fields are organized like this diagram
@@ -734 +365 @@
                 __u32 count = chead - phead;
 
+                if(count == 0) {
+                        return 0;
+                }
+
                 // We acquired an previous-head/current-head range
                 // go through the range and release the sqes
                 for( i; count ) {
-                        __u32 idx = ring.submit_q.array[ (phead + i) & smask ];
-
-                        /* paranoid */ verify( 0 != ring.submit_q.sqes[ idx ].user_data );
-                        __clean( &ring.submit_q.sqes[ idx ] );
-                }
+                        __cfadbg_print_safe(io, "Kernel I/O : release loop\n");
+                        __u32 idx = ctx.sq.kring.array[ (phead + i) & mask ];
+                        ctx.sq.free_ring.array[ (ftail + i) & mask ] = idx;
+                }
+
+                ctx.sq.kring.released = chead;          // note up to were we processed
+                __atomic_store_n(&ctx.sq.free_ring.tail, ftail + count, __ATOMIC_SEQ_CST);
+
+                __ioarbiter_notify(ctx);
+
                 return count;
         }
 
-        void __sqe_clean( volatile struct io_uring_sqe * sqe ) {
-                __clean( sqe );
-        }
-
-        static inline void __clean( volatile struct io_uring_sqe * sqe ) {
-                // If we are in debug mode, thrash the fields to make sure we catch reclamation errors
-                __cfaabi_dbg_debug_do(
-                        memset(sqe, 0xde, sizeof(*sqe));
-                        sqe->opcode = (sizeof(opcodes) / sizeof(const char *)) - 1;
-                );
-
-                // Mark the entry as unused
-                __atomic_store_n(&sqe->user_data, 3ul64, __ATOMIC_SEQ_CST);
+//=============================================================================================
+// I/O Arbiter
+//=============================================================================================
+        static inline void block(__outstanding_io_queue & queue, __outstanding_io & item) {
+                // Lock the list, it's not thread safe
+                lock( queue.lock __cfaabi_dbg_ctx2 );
+                {
+                        // Add our request to the list
+                        add( queue.queue, item );
+
+                        // Mark as pending
+                        __atomic_store_n( &queue.empty, false, __ATOMIC_SEQ_CST );
+                }
+                unlock( queue.lock );
+
+                wait( item.sem );
+        }
+
+        static inline bool empty(__outstanding_io_queue & queue ) {
+                return __atomic_load_n( &queue.empty, __ATOMIC_SEQ_CST);
+        }
+
+        static $io_context * __ioarbiter_allocate( $io_arbiter & this, __u32 idxs[], __u32 want ) {
+                __cfadbg_print_safe(io, "Kernel I/O : arbiter allocating\n");
+
+                __STATS__( false, io.alloc.block += 1; )
+
+                // No one has any resources left, wait for something to finish
+                // We need to add ourself to a list of pending allocs and wait for an answer
+                __pending_alloc pa;
+                pa.idxs = idxs;
+                pa.want = want;
+
+                block(this.pending, (__outstanding_io&)pa);
+
+                return pa.ctx;
+
+        }
+
+        static void __ioarbiter_notify( $io_arbiter & this, $io_context * ctx ) {
+                /* paranoid */ verify( !empty(this.pending.queue) );
+
+                lock( this.pending.lock __cfaabi_dbg_ctx2 );
+                {
+                        while( !empty(this.pending.queue) ) {
+                                __cfadbg_print_safe(io, "Kernel I/O : notifying\n");
+                                __u32 have = ctx->sq.free_ring.tail - ctx->sq.free_ring.head;
+                                __pending_alloc & pa = (__pending_alloc&)head( this.pending.queue );
+
+                                if( have > pa.want ) goto DONE;
+                                drop( this.pending.queue );
+
+                                /* paranoid */__attribute__((unused)) bool ret =
+
+                                __alloc(ctx, pa.idxs, pa.want);
+
+                                /* paranoid */ verify( ret );
+
+                                pa.ctx = ctx;
+
+                                post( pa.sem );
+                        }
+
+                        this.pending.empty = true;
+                        DONE:;
+                }
+                unlock( this.pending.lock );
+        }
+
+        static void __ioarbiter_notify( $io_context & ctx ) {
+                if(!empty( ctx.arbiter->pending )) {
+                        __ioarbiter_notify( *ctx.arbiter, &ctx );
+                }
+        }
+
+        // Simply append to the pending
+        static void __ioarbiter_submit( $io_context * ctx, __u32 idxs[], __u32 have, bool lazy ) {
+                __cfadbg_print_safe(io, "Kernel I/O : submitting %u from the arbiter to context %u\n", have, ctx->fd);
+
+                __cfadbg_print_safe(io, "Kernel I/O : waiting to submit %u\n", have);
+
+                __external_io ei;
+                ei.idxs = idxs;
+                ei.have = have;
+                ei.lazy = lazy;
+
+                block(ctx->ext_sq, (__outstanding_io&)ei);
+
+                __cfadbg_print_safe(io, "Kernel I/O : %u submitted from arbiter\n", have);
+        }
+
+        static void __ioarbiter_flush( $io_context & ctx ) {
+                if(!empty( ctx.ext_sq )) {
+                        __STATS__( false, io.flush.external += 1; )
+
+                        __cfadbg_print_safe(io, "Kernel I/O : arbiter flushing\n");
+
+                        lock( ctx.ext_sq.lock __cfaabi_dbg_ctx2 );
+                        {
+                                while( !empty(ctx.ext_sq.queue) ) {
+                                        __external_io & ei = (__external_io&)drop( ctx.ext_sq.queue );
+
+                                        __submit(&ctx, ei.idxs, ei.have, ei.lazy);
+
+                                        post( ei.sem );
+                                }
+
+                                ctx.ext_sq.empty = true;
+                        }
+                        unlock(ctx.ext_sq.lock );
+                }
         }
 #endif

libcfa/src/concurrency/io/call.cfa.in

@@ -54 +54 @@
                         | IOSQE_IO_DRAIN
                 #endif
+                #if defined(CFA_HAVE_IOSQE_IO_LINK)
+                        | IOSQE_IO_LINK
+                #endif
+                #if defined(CFA_HAVE_IOSQE_IO_HARDLINK)
+                        | IOSQE_IO_HARDLINK
+                #endif
                 #if defined(CFA_HAVE_IOSQE_ASYNC)
                         | IOSQE_ASYNC
                 #endif
-        ;
-
-        static const __u32 LINK_FLAGS = 0
-                #if defined(CFA_HAVE_IOSQE_IO_LINK)
-                        | IOSQE_IO_LINK
-                #endif
-                #if defined(CFA_HAVE_IOSQE_IO_HARDLINK)
-                        | IOSQE_IO_HARDLINK
+                #if defined(CFA_HAVE_IOSQE_BUFFER_SELECTED)
+                        | IOSQE_BUFFER_SELECTED
                 #endif
         ;
@@ -74 +74 @@
         ;
 
-        extern [* volatile struct io_uring_sqe, __u32] __submit_alloc( struct __io_data & ring, __u64 data );
-        extern void __submit( struct io_context * ctx, __u32 idx ) __attribute__((nonnull (1)));
-
-        static inline io_context * __get_io_context( void ) {
-                cluster * cltr = active_cluster();
-
-                /* paranoid */ verifyf( cltr, "No active cluster for io operation\\n");
-                assertf( cltr->io.cnt > 0, "Cluster %p has no default io contexts and no context was specified\\n", cltr );
-
-                /* paranoid */ verifyf( cltr->io.ctxs, "default io contexts for cluster %p are missing\\n", cltr);
-                return &cltr->io.ctxs[ thread_rand() % cltr->io.cnt ];
-        }
+        extern struct $io_context * cfa_io_allocate(struct io_uring_sqe * out_sqes[], __u32 out_idxs[], __u32 want)  __attribute__((nonnull (1,2)));
+        extern void cfa_io_submit( struct $io_context * in_ctx, __u32 in_idxs[], __u32 have, bool lazy ) __attribute__((nonnull (1,2)));
 #endif
 
@@ -98 +88 @@
 
 extern "C" {
-        #include <sys/types.h>
+        #include <asm/types.h>
         #include <sys/socket.h>
         #include <sys/syscall.h>
@@ -142 +132 @@
         extern int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event);
 
-        extern ssize_t splice(int fd_in, loff_t *off_in, int fd_out, loff_t *off_out, size_t len, unsigned int flags);
+        extern ssize_t splice(int fd_in, __off64_t *off_in, int fd_out, __off64_t *off_out, size_t len, unsigned int flags);
         extern ssize_t tee(int fd_in, int fd_out, size_t len, unsigned int flags);
 }
@@ -195 +185 @@
                 return ', '.join(args_a)
 
-AsyncTemplate = """inline void async_{name}(io_future_t & future, {params}, int submit_flags, io_cancellation * cancellation, io_context * context) {{
+AsyncTemplate = """inline void async_{name}(io_future_t & future, {params}, __u64 submit_flags) {{
         #if !defined(CFA_HAVE_LINUX_IO_URING_H) || !defined(CFA_HAVE_IORING_OP_{op})
                 ssize_t res = {name}({args});
@@ -205 +195 @@
                 }}
         #else
-                // we don't support LINK yet
-                if( 0 != (submit_flags & LINK_FLAGS) ) {{
-                        errno = ENOTSUP; return -1;
-                }}
-
-                if( !context ) {{
-                        context = __get_io_context();
-                }}
-                if(cancellation) {{
-                        cancellation->target = (__u64)(uintptr_t)&future;
-                }}
-
                 __u8 sflags = REGULAR_FLAGS & submit_flags;
-                struct __io_data & ring = *context->thrd.ring;
-
                 __u32 idx;
                 struct io_uring_sqe * sqe;
-                [(volatile struct io_uring_sqe *) sqe, idx] = __submit_alloc( ring, (__u64)(uintptr_t)&future );
+                struct $io_context * ctx = cfa_io_allocate( &sqe, &idx, 1 );
 
                 sqe->opcode = IORING_OP_{op};
+                sqe->user_data = (uintptr_t)&future;
                 sqe->flags = sflags;
                 sqe->ioprio = 0;
@@ -238 +215 @@
                 asm volatile("": : :"memory");
 
-                verify( sqe->user_data == (__u64)(uintptr_t)&future );
-                __submit( context, idx );
+                verify( sqe->user_data == (uintptr_t)&future );
+                cfa_io_submit( ctx, &idx, 1, 0 != (submit_flags & CFA_IO_LAZY) );
         #endif
 }}"""
 
-SyncTemplate = """{ret} cfa_{name}({params}, int submit_flags, Duration timeout, io_cancellation * cancellation, io_context * context) {{
-        if( timeout >= 0 ) {{
-                errno = ENOTSUP;
-                return -1;
-        }}
+SyncTemplate = """{ret} cfa_{name}({params}, __u64 submit_flags) {{
         io_future_t future;
 
-        async_{name}( future, {args}, submit_flags, cancellation, context );
+        async_{name}( future, {args}, submit_flags );
 
         wait( future );
@@ -265 +238 @@
                 'fd'  : 'fd',
                 'off' : 'offset',
-                'addr': '(__u64)iov',
+                'addr': '(uintptr_t)iov',
                 'len' : 'iovcnt',
         }, define = 'CFA_HAVE_PREADV2'),
@@ -272 +245 @@
                 'fd'  : 'fd',
                 'off' : 'offset',
-                'addr': '(__u64)iov',
+                'addr': '(uintptr_t)iov',
                 'len' : 'iovcnt'
         }, define = 'CFA_HAVE_PWRITEV2'),
@@ -284 +257 @@
                 'addr': 'fd',
                 'len': 'op',
-                'off': '(__u64)event'
+                'off': '(uintptr_t)event'
         }),
         # CFA_HAVE_IORING_OP_SYNC_FILE_RANGE
@@ -296 +269 @@
         Call('SENDMSG', 'ssize_t sendmsg(int sockfd, const struct msghdr *msg, int flags)', {
                 'fd': 'sockfd',
-                'addr': '(__u64)(struct msghdr *)msg',
+                'addr': '(uintptr_t)(struct msghdr *)msg',
                 'len': '1',
                 'msg_flags': 'flags'
@@ -303 +276 @@
         Call('RECVMSG', 'ssize_t recvmsg(int sockfd, struct msghdr *msg, int flags)', {
                 'fd': 'sockfd',
-                'addr': '(__u64)(struct msghdr *)msg',
+                'addr': '(uintptr_t)(struct msghdr *)msg',
                 'len': '1',
                 'msg_flags': 'flags'
@@ -310 +283 @@
         Call('SEND', 'ssize_t send(int sockfd, const void *buf, size_t len, int flags)', {
                 'fd': 'sockfd',
-                'addr': '(__u64)buf',
+                'addr': '(uintptr_t)buf',
                 'len': 'len',
                 'msg_flags': 'flags'
@@ -317 +290 @@
         Call('RECV', 'ssize_t recv(int sockfd, void *buf, size_t len, int flags)', {
                 'fd': 'sockfd',
-                'addr': '(__u64)buf',
+                'addr': '(uintptr_t)buf',
                 'len': 'len',
                 'msg_flags': 'flags'
@@ -324 +297 @@
         Call('ACCEPT', 'int accept4(int sockfd, struct sockaddr *addr, socklen_t *addrlen, int flags)', {
                 'fd': 'sockfd',
-                'addr': '(__u64)addr',
-                'addr2': '(__u64)addrlen',
+                'addr': '(uintptr_t)addr',
+                'addr2': '(uintptr_t)addrlen',
                 'accept_flags': 'flags'
         }),
@@ -331 +304 @@
         Call('CONNECT', 'int connect(int sockfd, const struct sockaddr *addr, socklen_t addrlen)', {
                 'fd': 'sockfd',
-                'addr': '(__u64)addr',
+                'addr': '(uintptr_t)addr',
                 'off': 'addrlen'
         }),
@@ -337 +310 @@
         Call('FALLOCATE', 'int fallocate(int fd, int mode, off_t offset, off_t len)', {
                 'fd': 'fd',
-                'addr': '(__u64)len',
+                'addr': '(uintptr_t)len',
                 'len': 'mode',
                 'off': 'offset'
@@ -350 +323 @@
         # CFA_HAVE_IORING_OP_MADVISE
         Call('MADVISE', 'int madvise(void *addr, size_t length, int advice)', {
-                'addr': '(__u64)addr',
+                'addr': '(uintptr_t)addr',
                 'len': 'length',
                 'fadvise_advice': 'advice'
@@ -357 +330 @@
         Call('OPENAT', 'int openat(int dirfd, const char *pathname, int flags, mode_t mode)', {
                 'fd': 'dirfd',
-                'addr': '(__u64)pathname',
+                'addr': '(uintptr_t)pathname',
                 'len': 'mode',
                 'open_flags': 'flags;'
@@ -366 +339 @@
                 'addr': 'pathname',
                 'len': 'sizeof(*how)',
-                'off': '(__u64)how',
+                'off': '(uintptr_t)how',
         }, define = 'CFA_HAVE_OPENAT2'),
         # CFA_HAVE_IORING_OP_CLOSE
@@ -375 +348 @@
         Call('STATX', 'int statx(int dirfd, const char *pathname, int flags, unsigned int mask, struct statx *statxbuf)', {
                 'fd': 'dirfd',
-                'off': '(__u64)statxbuf',
+                'off': '(uintptr_t)statxbuf',
                 'addr': 'pathname',
                 'len': 'mask',
@@ -383 +356 @@
         Call('READ', 'ssize_t read(int fd, void * buf, size_t count)', {
                 'fd': 'fd',
-                'addr': '(__u64)buf',
+                'addr': '(uintptr_t)buf',
                 'len': 'count'
         }),
@@ -389 +362 @@
         Call('WRITE', 'ssize_t write(int fd, void * buf, size_t count)', {
                 'fd': 'fd',
-                'addr': '(__u64)buf',
+                'addr': '(uintptr_t)buf',
                 'len': 'count'
         }),
         # CFA_HAVE_IORING_OP_SPLICE
-        Call('SPLICE', 'ssize_t splice(int fd_in, loff_t *off_in, int fd_out, loff_t *off_out, size_t len, unsigned int flags)', {
+        Call('SPLICE', 'ssize_t splice(int fd_in, __off64_t *off_in, int fd_out, __off64_t *off_out, size_t len, unsigned int flags)', {
                 'splice_fd_in': 'fd_in',
                 'splice_off_in': 'off_in ? (__u64)*off_in : (__u64)-1',
@@ -415 +388 @@
         if c.define:
                 print("""#if defined({define})
-        {ret} cfa_{name}({params}, int submit_flags, Duration timeout, io_cancellation * cancellation, io_context * context);
+        {ret} cfa_{name}({params}, __u64 submit_flags);
 #endif""".format(define=c.define,ret=c.ret, name=c.name, params=c.params))
         else:
-                print("{ret} cfa_{name}({params}, int submit_flags, Duration timeout, io_cancellation * cancellation, io_context * context);"
+                print("{ret} cfa_{name}({params}, __u64 submit_flags);"
                 .format(ret=c.ret, name=c.name, params=c.params))
 
@@ -426 +399 @@
         if c.define:
                 print("""#if defined({define})
-        void async_{name}(io_future_t & future, {params}, int submit_flags, io_cancellation * cancellation, io_context * context);
+        void async_{name}(io_future_t & future, {params}, __u64 submit_flags);
 #endif""".format(define=c.define,name=c.name, params=c.params))
         else:
-                print("void async_{name}(io_future_t & future, {params}, int submit_flags, io_cancellation * cancellation, io_context * context);"
+                print("void async_{name}(io_future_t & future, {params}, __u64 submit_flags);"
                 .format(name=c.name, params=c.params))
 print("\n")
@@ -474 +447 @@
 
 print("""
-//-----------------------------------------------------------------------------
-bool cancel(io_cancellation & this) {
-        #if !defined(CFA_HAVE_LINUX_IO_URING_H) || !defined(CFA_HAVE_IORING_OP_ASYNC_CANCEL)
-                return false;
-        #else
-                io_future_t future;
-
-                io_context * context = __get_io_context();
-
-                __u8 sflags = 0;
-                struct __io_data & ring = *context->thrd.ring;
-
-                __u32 idx;
-                volatile struct io_uring_sqe * sqe;
-                [sqe, idx] = __submit_alloc( ring, (__u64)(uintptr_t)&future );
-
-                sqe->__pad2[0] = sqe->__pad2[1] = sqe->__pad2[2] = 0;
-                sqe->opcode = IORING_OP_ASYNC_CANCEL;
-                sqe->flags = sflags;
-                sqe->addr = this.target;
-
-                verify( sqe->user_data == (__u64)(uintptr_t)&future );
-                __submit( context, idx );
-
-                wait(future);
-
-                if( future.result == 0 ) return true; // Entry found
-                if( future.result == -EALREADY) return true; // Entry found but in progress
-                if( future.result == -ENOENT ) return false; // Entry not found
-                return false;
-        #endif
-}
-
 //-----------------------------------------------------------------------------
 // Check if a function is has asynchronous

libcfa/src/concurrency/io/setup.cfa

@@ -26 +26 @@
 
 #if !defined(CFA_HAVE_LINUX_IO_URING_H)
-        void __kernel_io_startup() {
-                // Nothing to do without io_uring
-        }
-
-        void __kernel_io_shutdown() {
-                // Nothing to do without io_uring
-        }
-
         void ?{}(io_context_params & this) {}
 
-        void ?{}(io_context & this, struct cluster & cl) {}
-        void ?{}(io_context & this, struct cluster & cl, const io_context_params & params) {}
-
-        void ^?{}(io_context & this) {}
-        void ^?{}(io_context & this, bool cluster_context) {}
-
-        void register_fixed_files( io_context &, int *, unsigned ) {}
-        void register_fixed_files( cluster    &, int *, unsigned ) {}
+        void  ?{}($io_context & this, struct cluster & cl) {}
+        void ^?{}($io_context & this) {}
+
+        void __cfa_io_start( processor * proc ) {}
+        void __cfa_io_flush( processor * proc ) {}
+        void __cfa_io_stop ( processor * proc ) {}
+
+        $io_arbiter * create(void) { return 0p; }
+        void destroy($io_arbiter *) {}
 
 #else
@@ -68 +61 @@
         void ?{}(io_context_params & this) {
                 this.num_entries = 256;
-                this.num_ready = 256;
-                this.submit_aff = -1;
-                this.eager_submits = false;
-                this.poller_submits = false;
-                this.poll_submit = false;
-                this.poll_complete = false;
         }
 
@@ -106 +93 @@
 
 //=============================================================================================
-// I/O Startup / Shutdown logic + Master Poller
-//=============================================================================================
-
-        // IO Master poller loop forward
-        static void * iopoll_loop( __attribute__((unused)) void * args );
-
-        static struct {
-                      pthread_t  thrd;    // pthread handle to io poller thread
-                      void *     stack;   // pthread stack for io poller thread
-                      int        epollfd; // file descriptor to the epoll instance
-                volatile     bool run;     // Whether or not to continue
-                volatile     bool stopped; // Whether the poller has finished running
-                volatile uint64_t epoch;   // Epoch used for memory reclamation
-        } iopoll;
-
-        void __kernel_io_startup(void) {
-                __cfadbg_print_safe(io_core, "Kernel : Creating EPOLL instance\n" );
-
-                iopoll.epollfd = epoll_create1(0);
-                if (iopoll.epollfd == -1) {
-                        abort( "internal error, epoll_create1\n");
-                }
-
-                __cfadbg_print_safe(io_core, "Kernel : Starting io poller thread\n" );
-
-                iopoll.stack   = __create_pthread( &iopoll.thrd, iopoll_loop, 0p );
-                iopoll.run     = true;
-                iopoll.stopped = false;
-                iopoll.epoch   = 0;
-        }
-
-        void __kernel_io_shutdown(void) {
-                // Notify the io poller thread of the shutdown
-                iopoll.run = false;
-                sigval val = { 1 };
-                pthread_sigqueue( iopoll.thrd, SIGUSR1, val );
-
-                // Wait for the io poller thread to finish
-
-                __destroy_pthread( iopoll.thrd, iopoll.stack, 0p );
-
-                int ret = close(iopoll.epollfd);
-                if (ret == -1) {
-                        abort( "internal error, close epoll\n");
-                }
-
-                // Io polling is now fully stopped
-
-                __cfadbg_print_safe(io_core, "Kernel : IO poller stopped\n" );
-        }
-
-        static void * iopoll_loop( __attribute__((unused)) void * args ) {
-                __processor_id_t id;
-                id.full_proc = false;
-                id.id = doregister(&id);
-                __cfaabi_tls.this_proc_id = &id;
-                __cfadbg_print_safe(io_core, "Kernel : IO poller thread starting\n" );
-
-                // Block signals to control when they arrive
-                sigset_t mask;
-                sigfillset(&mask);
-                if ( pthread_sigmask( SIG_BLOCK, &mask, 0p ) == -1 ) {
-                abort( "internal error, pthread_sigmask" );
-                }
-
-                sigdelset( &mask, SIGUSR1 );
-
-                // Create sufficient events
-                struct epoll_event events[10];
-                // Main loop
-                while( iopoll.run ) {
-                        __cfadbg_print_safe(io_core, "Kernel I/O - epoll : waiting on io_uring contexts\n");
-
-                        // increment the epoch to notify any deleters we are starting a new cycle
-                        __atomic_fetch_add(&iopoll.epoch, 1, __ATOMIC_SEQ_CST);
-
-                        // Wait for events
-                        int nfds = epoll_pwait( iopoll.epollfd, events, 10, -1, &mask );
-
-                        __cfadbg_print_safe(io_core, "Kernel I/O - epoll : %d io contexts events, waking up\n", nfds);
-
-                        // Check if an error occured
-                        if (nfds == -1) {
-                                if( errno == EINTR ) continue;
-                                abort( "internal error, pthread_sigmask" );
-                        }
-
-                        for(i; nfds) {
-                                $io_ctx_thread * io_ctx = ($io_ctx_thread *)(uintptr_t)events[i].data.u64;
-                                /* paranoid */ verify( io_ctx );
-                                __cfadbg_print_safe(io_core, "Kernel I/O - epoll : Unparking io poller %d (%p)\n", io_ctx->ring->fd, io_ctx);
-                                #if !defined( __CFA_NO_STATISTICS__ )
-                                        __cfaabi_tls.this_stats = io_ctx->self.curr_cluster->stats;
-                                #endif
-
-                                eventfd_t v;
-                                eventfd_read(io_ctx->ring->efd, &v);
-
-                                post( io_ctx->sem );
-                        }
-                }
-
-                __atomic_store_n(&iopoll.stopped, true, __ATOMIC_SEQ_CST);
-
-                __cfadbg_print_safe(io_core, "Kernel : IO poller thread stopping\n" );
-                unregister(&id);
-                return 0p;
-        }
-
-//=============================================================================================
 // I/O Context Constrution/Destruction
 //=============================================================================================
 
-        void ?{}($io_ctx_thread & this, struct cluster & cl) { (this.self){ "IO Poller", cl }; }
-        void main( $io_ctx_thread & this );
-        static inline $thread * get_thread( $io_ctx_thread & this ) { return &this.self; }
-        void ^?{}( $io_ctx_thread & mutex this ) {}
-
-        static void __io_create ( __io_data & this, const io_context_params & params_in );
-        static void __io_destroy( __io_data & this );
-
-        void ?{}(io_context & this, struct cluster & cl, const io_context_params & params) {
-                (this.thrd){ cl };
-                this.thrd.ring = malloc();
-                __cfadbg_print_safe(io_core, "Kernel I/O : Creating ring for io_context %p\n", &this);
-                __io_create( *this.thrd.ring, params );
-
-                __cfadbg_print_safe(io_core, "Kernel I/O : Starting poller thread for io_context %p\n", &this);
-                this.thrd.done = false;
-                __thrd_start( this.thrd, main );
-
-                __cfadbg_print_safe(io_core, "Kernel I/O : io_context %p ready\n", &this);
-        }
-
-        void ?{}(io_context & this, struct cluster & cl) {
-                io_context_params params;
-                (this){ cl, params };
-        }
-
-        void ^?{}(io_context & this, bool cluster_context) {
-                __cfadbg_print_safe(io_core, "Kernel I/O : tearing down io_context %p\n", &this);
-
-                // Notify the thread of the shutdown
-                __atomic_store_n(&this.thrd.done, true, __ATOMIC_SEQ_CST);
-
-                // If this is an io_context within a cluster, things get trickier
-                $thread & thrd = this.thrd.self;
-                if( cluster_context ) {
-                        // We are about to do weird things with the threads
-                        // we don't need interrupts to complicate everything
-                        disable_interrupts();
-
-                        // Get cluster info
-                        cluster & cltr = *thrd.curr_cluster;
-                        /* paranoid */ verify( cltr.idles.total == 0 || &cltr == mainCluster );
-                        /* paranoid */ verify( !ready_mutate_islocked() );
-
-                        // We need to adjust the clean-up based on where the thread is
-                        if( thrd.state == Ready || thrd.preempted != __NO_PREEMPTION ) {
-                                // This is the tricky case
-                                // The thread was preempted or ready to run and now it is on the ready queue
-                                // but the cluster is shutting down, so there aren't any processors to run the ready queue
-                                // the solution is to steal the thread from the ready-queue and pretend it was blocked all along
-
-                                ready_schedule_lock();
-                                        // The thread should on the list
-                                        /* paranoid */ verify( thrd.link.next != 0p );
-
-                                        // Remove the thread from the ready queue of this cluster
-                                        // The thread should be the last on the list
-                                        __attribute__((unused)) bool removed = remove_head( &cltr, &thrd );
-                                        /* paranoid */ verify( removed );
-                                        thrd.link.next = 0p;
-                                        thrd.link.prev = 0p;
-
-                                        // Fixup the thread state
-                                        thrd.state = Blocked;
-                                        thrd.ticket = TICKET_BLOCKED;
-                                        thrd.preempted = __NO_PREEMPTION;
-
-                                ready_schedule_unlock();
-
-                                // Pretend like the thread was blocked all along
-                        }
-                        // !!! This is not an else if !!!
-                        // Ok, now the thread is blocked (whether we cheated to get here or not)
-                        if( thrd.state == Blocked ) {
-                                // This is the "easy case"
-                                // The thread is parked and can easily be moved to active cluster
-                                verify( thrd.curr_cluster != active_cluster() || thrd.curr_cluster == mainCluster );
-                                thrd.curr_cluster = active_cluster();
-
-                                // unpark the fast io_poller
-                                unpark( &thrd );
-                        }
-                        else {
-                                // The thread is in a weird state
-                                // I don't know what to do here
-                                abort("io_context poller thread is in unexpected state, cannot clean-up correctly\n");
-                        }
-
-                        // The weird thread kidnapping stuff is over, restore interrupts.
-                        enable_interrupts( __cfaabi_dbg_ctx );
-                } else {
-                        post( this.thrd.sem );
-                }
-
-                ^(this.thrd){};
-                __cfadbg_print_safe(io_core, "Kernel I/O : Stopped poller thread for io_context %p\n", &this);
-
-                __io_destroy( *this.thrd.ring );
-                __cfadbg_print_safe(io_core, "Kernel I/O : Destroyed ring for io_context %p\n", &this);
-
-                free(this.thrd.ring);
-        }
-
-        void ^?{}(io_context & this) {
-                ^(this){ false };
+
+
+        static void __io_uring_setup ( $io_context & this, const io_context_params & params_in, int procfd );
+        static void __io_uring_teardown( $io_context & this );
+        static void __epoll_register($io_context & ctx);
+        static void __epoll_unregister($io_context & ctx);
+        void __ioarbiter_register( $io_arbiter & mutex, $io_context & ctx );
+        void __ioarbiter_unregister( $io_arbiter & mutex, $io_context & ctx );
+
+        void ?{}($io_context & this, processor * proc, struct cluster & cl) {
+                /* paranoid */ verify( cl.io.arbiter );
+                this.proc = proc;
+                this.arbiter = cl.io.arbiter;
+                this.ext_sq.empty = true;
+                (this.ext_sq.queue){};
+                __io_uring_setup( this, cl.io.params, proc->idle );
+                __cfadbg_print_safe(io_core, "Kernel I/O : Created ring for io_context %u (%p)\n", this.fd, &this);
+        }
+
+        void ^?{}($io_context & this) {
+                __cfadbg_print_safe(io_core, "Kernel I/O : tearing down io_context %u\n", this.fd);
+
+                __io_uring_teardown( this );
+                __cfadbg_print_safe(io_core, "Kernel I/O : Destroyed ring for io_context %u\n", this.fd);
         }
 
@@ -329 +125 @@
         extern void __enable_interrupts_hard();
 
-        static void __io_create( __io_data & this, const io_context_params & params_in ) {
+        static void __io_uring_setup( $io_context & this, const io_context_params & params_in, int procfd ) {
                 // Step 1 : call to setup
                 struct io_uring_params params;
                 memset(&params, 0, sizeof(params));
-                if( params_in.poll_submit   ) params.flags |= IORING_SETUP_SQPOLL;
-                if( params_in.poll_complete ) params.flags |= IORING_SETUP_IOPOLL;
+                // if( params_in.poll_submit   ) params.flags |= IORING_SETUP_SQPOLL;
+                // if( params_in.poll_complete ) params.flags |= IORING_SETUP_IOPOLL;
 
                 __u32 nentries = params_in.num_entries != 0 ? params_in.num_entries : 256;
@@ -340 +136 @@
                         abort("ERROR: I/O setup 'num_entries' must be a power of 2\n");
                 }
-                if( params_in.poller_submits && params_in.eager_submits ) {
-                        abort("ERROR: I/O setup 'poller_submits' and 'eager_submits' cannot be used together\n");
-                }
 
                 int fd = syscall(__NR_io_uring_setup, nentries, &params );
@@ -350 +143 @@
 
                 // Step 2 : mmap result
-                memset( &this, 0, sizeof(struct __io_data) );
-                struct __submition_data  & sq = this.submit_q;
-                struct __completion_data & cq = this.completion_q;
+                struct __sub_ring_t & sq = this.sq;
+                struct __cmp_ring_t & cq = this.cq;
 
                 // calculate the right ring size
@@ -401 +193 @@
                 // Get the pointers from the kernel to fill the structure
                 // submit queue
-                sq.head    = (volatile __u32 *)(((intptr_t)sq.ring_ptr) + params.sq_off.head);
-                sq.tail    = (volatile __u32 *)(((intptr_t)sq.ring_ptr) + params.sq_off.tail);
-                sq.mask    = (   const __u32 *)(((intptr_t)sq.ring_ptr) + params.sq_off.ring_mask);
-                sq.num     = (   const __u32 *)(((intptr_t)sq.ring_ptr) + params.sq_off.ring_entries);
-                sq.flags   = (         __u32 *)(((intptr_t)sq.ring_ptr) + params.sq_off.flags);
-                sq.dropped = (         __u32 *)(((intptr_t)sq.ring_ptr) + params.sq_off.dropped);
-                sq.array   = (         __u32 *)(((intptr_t)sq.ring_ptr) + params.sq_off.array);
-                sq.prev_head = *sq.head;
-
-                {
-                        const __u32 num = *sq.num;
-                        for( i; num ) {
-                                __sqe_clean( &sq.sqes[i] );
-                        }
-                }
-
-                (sq.submit_lock){};
-                (sq.release_lock){};
-
-                if( params_in.poller_submits || params_in.eager_submits ) {
-                        /* paranoid */ verify( is_pow2( params_in.num_ready ) || (params_in.num_ready < 8) );
-                        sq.ready_cnt = max( params_in.num_ready, 8 );
-                        sq.ready = alloc( sq.ready_cnt, 64`align );
-                        for(i; sq.ready_cnt) {
-                                sq.ready[i] = -1ul32;
-                        }
-                        sq.prev_ready = 0;
-                }
-                else {
-                        sq.ready_cnt = 0;
-                        sq.ready = 0p;
-                        sq.prev_ready = 0;
-                }
+                sq.kring.head  = (volatile __u32 *)(((intptr_t)sq.ring_ptr) + params.sq_off.head);
+                sq.kring.tail  = (volatile __u32 *)(((intptr_t)sq.ring_ptr) + params.sq_off.tail);
+                sq.kring.array = (         __u32 *)(((intptr_t)sq.ring_ptr) + params.sq_off.array);
+                sq.mask        = (   const __u32 *)(((intptr_t)sq.ring_ptr) + params.sq_off.ring_mask);
+                sq.num         = (   const __u32 *)(((intptr_t)sq.ring_ptr) + params.sq_off.ring_entries);
+                sq.flags       = (         __u32 *)(((intptr_t)sq.ring_ptr) + params.sq_off.flags);
+                sq.dropped     = (         __u32 *)(((intptr_t)sq.ring_ptr) + params.sq_off.dropped);
+
+                sq.kring.released = 0;
+
+                sq.free_ring.head = 0;
+                sq.free_ring.tail = *sq.num;
+                sq.free_ring.array = alloc( *sq.num, 128`align );
+                for(i; (__u32)*sq.num) {
+                        sq.free_ring.array[i] = i;
+                }
+
+                sq.to_submit = 0;
 
                 // completion queue
@@ -446 +223 @@
                 // io_uring_register is so f*cking slow on some machine that it
                 // will never succeed if preemption isn't hard blocked
+                __cfadbg_print_safe(io_core, "Kernel I/O : registering %d for completion with ring %d\n", procfd, fd);
+
                 __disable_interrupts_hard();
 
-                int efd = eventfd(0, 0);
-                if (efd < 0) {
-                        abort("KERNEL ERROR: IO_URING EVENTFD - %s\n", strerror(errno));
-                }
-
-                int ret = syscall( __NR_io_uring_register, fd, IORING_REGISTER_EVENTFD, &efd, 1);
+                int ret = syscall( __NR_io_uring_register, fd, IORING_REGISTER_EVENTFD, &procfd, 1);
                 if (ret < 0) {
                         abort("KERNEL ERROR: IO_URING EVENTFD REGISTER - %s\n", strerror(errno));
@@ -459 +233 @@
 
                 __enable_interrupts_hard();
+
+                __cfadbg_print_safe(io_core, "Kernel I/O : registered %d for completion with ring %d\n", procfd, fd);
 
                 // some paranoid checks
@@ -468 +244 @@
                 /* paranoid */ verifyf( (*sq.mask) == ((*sq.num) - 1ul32), "IO_URING Expected mask to be %u (%u entries), was %u", (*sq.num) - 1ul32, *sq.num, *sq.mask );
                 /* paranoid */ verifyf( (*sq.num) >= nentries, "IO_URING Expected %u entries, got %u", nentries, *sq.num );
-                /* paranoid */ verifyf( (*sq.head) == 0, "IO_URING Expected head to be 0, got %u", *sq.head );
-                /* paranoid */ verifyf( (*sq.tail) == 0, "IO_URING Expected tail to be 0, got %u", *sq.tail );
+                /* paranoid */ verifyf( (*sq.kring.head) == 0, "IO_URING Expected head to be 0, got %u", *sq.kring.head );
+                /* paranoid */ verifyf( (*sq.kring.tail) == 0, "IO_URING Expected tail to be 0, got %u", *sq.kring.tail );
 
                 // Update the global ring info
-                this.ring_flags = params.flags;
+                this.ring_flags = 0;
                 this.fd         = fd;
-                this.efd        = efd;
-                this.eager_submits  = params_in.eager_submits;
-                this.poller_submits = params_in.poller_submits;
-        }
-
-        static void __io_destroy( __io_data & this ) {
+        }
+
+        static void __io_uring_teardown( $io_context & this ) {
                 // Shutdown the io rings
-                struct __submition_data  & sq = this.submit_q;
-                struct __completion_data & cq = this.completion_q;
+                struct __sub_ring_t & sq = this.sq;
+                struct __cmp_ring_t & cq = this.cq;
 
                 // unmap the submit queue entries
@@ -497 +270 @@
                 // close the file descriptor
                 close(this.fd);
-                close(this.efd);
-
-                free( this.submit_q.ready ); // Maybe null, doesn't matter
+
+                free( this.sq.free_ring.array ); // Maybe null, doesn't matter
+        }
+
+        void __cfa_io_start( processor * proc ) {
+                proc->io.ctx = alloc();
+                (*proc->io.ctx){proc, *proc->cltr};
+        }
+        void __cfa_io_stop ( processor * proc ) {
+                ^(*proc->io.ctx){};
+                free(proc->io.ctx);
         }
 
@@ -505 +286 @@
 // I/O Context Sleep
 //=============================================================================================
-        static inline void __ioctx_epoll_ctl($io_ctx_thread & ctx, int op, const char * error) {
-                struct epoll_event ev;
-                ev.events = EPOLLIN | EPOLLONESHOT;
-                ev.data.u64 = (__u64)&ctx;
-                int ret = epoll_ctl(iopoll.epollfd, op, ctx.ring->efd, &ev);
-                if (ret < 0) {
-                        abort( "KERNEL ERROR: EPOLL %s - (%d) %s\n", error, (int)errno, strerror(errno) );
-                }
-        }
-
-        void __ioctx_register($io_ctx_thread & ctx) {
-                __ioctx_epoll_ctl(ctx, EPOLL_CTL_ADD, "ADD");
-        }
-
-        void __ioctx_prepare_block($io_ctx_thread & ctx) {
-                __cfadbg_print_safe(io_core, "Kernel I/O - epoll : Re-arming io poller %d (%p)\n", ctx.ring->fd, &ctx);
-                __ioctx_epoll_ctl(ctx, EPOLL_CTL_MOD, "REARM");
-        }
-
-        void __ioctx_unregister($io_ctx_thread & ctx) {
-                // Read the current epoch so we know when to stop
-                size_t curr = __atomic_load_n(&iopoll.epoch, __ATOMIC_SEQ_CST);
-
-                // Remove the fd from the iopoller
-                __ioctx_epoll_ctl(ctx, EPOLL_CTL_DEL, "REMOVE");
-
-                // Notify the io poller thread of the shutdown
-                iopoll.run = false;
-                sigval val = { 1 };
-                pthread_sigqueue( iopoll.thrd, SIGUSR1, val );
-
-                // Make sure all this is done
-                __atomic_thread_fence(__ATOMIC_SEQ_CST);
-
-                // Wait for the next epoch
-                while(curr == iopoll.epoch && !iopoll.stopped) Pause();
-        }
+        // static inline void __epoll_ctl($io_context & ctx, int op, const char * error) {
+        //      struct epoll_event ev;
+        //      ev.events = EPOLLIN | EPOLLONESHOT;
+        //      ev.data.u64 = (__u64)&ctx;
+        //      int ret = epoll_ctl(iopoll.epollfd, op, ctx.efd, &ev);
+        //      if (ret < 0) {
+        //              abort( "KERNEL ERROR: EPOLL %s - (%d) %s\n", error, (int)errno, strerror(errno) );
+        //      }
+        // }
+
+        // static void __epoll_register($io_context & ctx) {
+        //      __epoll_ctl(ctx, EPOLL_CTL_ADD, "ADD");
+        // }
+
+        // static void __epoll_unregister($io_context & ctx) {
+        //      // Read the current epoch so we know when to stop
+        //      size_t curr = __atomic_load_n(&iopoll.epoch, __ATOMIC_SEQ_CST);
+
+        //      // Remove the fd from the iopoller
+        //      __epoll_ctl(ctx, EPOLL_CTL_DEL, "REMOVE");
+
+        //      // Notify the io poller thread of the shutdown
+        //      iopoll.run = false;
+        //      sigval val = { 1 };
+        //      pthread_sigqueue( iopoll.thrd, SIGUSR1, val );
+
+        //      // Make sure all this is done
+        //      __atomic_thread_fence(__ATOMIC_SEQ_CST);
+
+        //      // Wait for the next epoch
+        //      while(curr == iopoll.epoch && !iopoll.stopped) Pause();
+        // }
+
+        // void __ioctx_prepare_block($io_context & ctx) {
+        //      __cfadbg_print_safe(io_core, "Kernel I/O - epoll : Re-arming io poller %d (%p)\n", ctx.fd, &ctx);
+        //      __epoll_ctl(ctx, EPOLL_CTL_MOD, "REARM");
+        // }
+
 
 //=============================================================================================
 // I/O Context Misc Setup
 //=============================================================================================
-        void register_fixed_files( io_context & ctx, int * files, unsigned count ) {
-                int ret = syscall( __NR_io_uring_register, ctx.thrd.ring->fd, IORING_REGISTER_FILES, files, count );
-                if( ret < 0 ) {
-                        abort( "KERNEL ERROR: IO_URING REGISTER - (%d) %s\n", (int)errno, strerror(errno) );
-                }
-
-                __cfadbg_print_safe( io_core, "Kernel I/O : Performed io_register for %p, returned %d\n", active_thread(), ret );
-        }
-
-        void register_fixed_files( cluster & cltr, int * files, unsigned count ) {
-                for(i; cltr.io.cnt) {
-                        register_fixed_files( cltr.io.ctxs[i], files, count );
-                }
-        }
+        void ?{}( $io_arbiter & this ) {
+                this.pending.empty = true;
+        }
+
+        void ^?{}( $io_arbiter & this ) {}
+
+        $io_arbiter * create(void) {
+                return new();
+        }
+        void destroy($io_arbiter * arbiter) {
+                delete(arbiter);
+        }
+
+//=============================================================================================
+// I/O Context Misc Setup
+//=============================================================================================
+
 #endif

libcfa/src/concurrency/io/types.hfa

@@ -22 +22 @@
 
 #include "bits/locks.hfa"
+#include "bits/queue.hfa"
 #include "kernel/fwd.hfa"
 
 #if defined(CFA_HAVE_LINUX_IO_URING_H)
-        #define LEADER_LOCK
-        struct __leaderlock_t {
-                struct $thread * volatile value;        // ($thread) next_leader | (bool:1) is_locked
-        };
+        #include "bits/sequence.hfa"
+        #include "monitor.hfa"
 
-        static inline void ?{}( __leaderlock_t & this ) { this.value = 0p; }
+        struct processor;
+        monitor $io_arbiter;
 
         //-----------------------------------------------------------------------
         // Ring Data structure
-      struct __submition_data {
-                // Head and tail of the ring (associated with array)
-                volatile __u32 * head;
-                volatile __u32 * tail;
-                volatile __u32 prev_head;
+      struct __sub_ring_t {
+                struct {
+                        // Head and tail of the ring (associated with array)
+                        volatile __u32 * head;   // one passed last index consumed by the kernel
+                        volatile __u32 * tail;   // one passed last index visible to the kernel
+                        volatile __u32 released; // one passed last index released back to the free list
 
-                // The actual kernel ring which uses head/tail
-                // indexes into the sqes arrays
-                __u32 * array;
+                        // The actual kernel ring which uses head/tail
+                        // indexes into the sqes arrays
+                        __u32 * array;
+                } kring;
+
+                struct {
+                        volatile __u32 head;
+                        volatile __u32 tail;
+                        // The ring which contains free allocations
+                        // indexes into the sqes arrays
+                        __u32 * array;
+                } free_ring;
+
+                // number of sqes to submit on next system call.
+                __u32 to_submit;
 
                 // number of entries and mask to go with it
@@ -48 +61 @@
                 const __u32 * mask;
 
-                // Submission flags (Not sure what for)
+                // Submission flags, currently only IORING_SETUP_SQPOLL
                 __u32 * flags;
 
-                // number of sqes not submitted (whatever that means)
+                // number of sqes not submitted
+                // From documentation : [dropped] is incremented for each invalid submission queue entry encountered in the ring buffer.
                 __u32 * dropped;
 
-                // Like head/tail but not seen by the kernel
-                volatile __u32 * ready;
-                __u32 ready_cnt;
-                __u32 prev_ready;
-
-                #if defined(LEADER_LOCK)
-                        __leaderlock_t submit_lock;
-                #else
-                        __spinlock_t submit_lock;
-                #endif
-                __spinlock_t  release_lock;
-
                 // A buffer of sqes (not the actual ring)
-                volatile struct io_uring_sqe * sqes;
+                struct io_uring_sqe * sqes;
 
                 // The location and size of the mmaped area
@@ -74 +76 @@
         };
 
-        struct __completion_data {
+        struct __cmp_ring_t {
                 // Head and tail of the ring
                 volatile __u32 * head;
@@ -83 +85 @@
                 const __u32 * num;
 
-                // number of cqes not submitted (whatever that means)
+                // I don't know what this value is for
                 __u32 * overflow;
 
@@ -94 +96 @@
         };
 
-        struct __io_data {
-                struct __submition_data submit_q;
-                struct __completion_data completion_q;
+        struct __outstanding_io {
+                inline Colable;
+                single_sem sem;
+        };
+        static inline __outstanding_io *& Next( __outstanding_io * n ) { return (__outstanding_io *)Next( (Colable *)n ); }
+
+        struct __outstanding_io_queue {
+                __spinlock_t lock;
+                Queue(__outstanding_io) queue;
+                volatile bool empty;
+        };
+
+        struct __external_io {
+                inline __outstanding_io;
+                __u32 * idxs;
+                __u32 have;
+                bool lazy;
+        };
+
+
+        struct __attribute__((aligned(128))) $io_context {
+                $io_arbiter * arbiter;
+                processor * proc;
+
+                __outstanding_io_queue ext_sq;
+
+                struct __sub_ring_t sq;
+                struct __cmp_ring_t cq;
                 __u32 ring_flags;
                 int fd;
-                int efd;
-                bool eager_submits:1;
-                bool poller_submits:1;
+        };
+
+        struct __pending_alloc {
+                inline __outstanding_io;
+                __u32 * idxs;
+                __u32 want;
+                $io_context * ctx;
+        };
+
+        struct __attribute__((aligned(128))) $io_arbiter {
+                __outstanding_io_queue pending;
         };
 
@@ -133 +168 @@
         #endif
 
-        struct $io_ctx_thread;
-        void __ioctx_register($io_ctx_thread & ctx);
-        void __ioctx_unregister($io_ctx_thread & ctx);
-        void __ioctx_prepare_block($io_ctx_thread & ctx);
-        void __sqe_clean( volatile struct io_uring_sqe * sqe );
+        // void __ioctx_prepare_block($io_context & ctx);
 #endif
 
@@ -148 +179 @@
 
 static inline {
-        bool fulfil( io_future_t & this, __s32 result ) {
+        $thread * fulfil( io_future_t & this, __s32 result, bool do_unpark = true ) {
                 this.result = result;
-                return fulfil(this.self);
+                return fulfil(this.self, do_unpark);
         }
 

libcfa/src/concurrency/iofwd.hfa

@@ -18 +18 @@
 #include <unistd.h>
 extern "C" {
-        #include <sys/types.h>
+        #include <asm/types.h>
         #if CFA_HAVE_LINUX_IO_URING_H
                 #include <linux/io_uring.h>
@@ -48 +48 @@
 struct cluster;
 struct io_future_t;
-struct io_context;
-struct io_cancellation;
+struct $io_context;
 
 struct iovec;
@@ -55 +54 @@
 struct sockaddr;
 struct statx;
+struct epoll_event;
+
+struct io_uring_sqe;
+
+//----------
+// underlying calls
+extern struct $io_context * cfa_io_allocate(struct io_uring_sqe * out_sqes[], __u32 out_idxs[], __u32 want)  __attribute__((nonnull (1,2)));
+extern void cfa_io_submit( struct $io_context * in_ctx, __u32 in_idxs[], __u32 have, bool lazy ) __attribute__((nonnull (1,2)));
 
 //----------
 // synchronous calls
 #if defined(CFA_HAVE_PREADV2)
-        extern ssize_t cfa_preadv2(int fd, const struct iovec *iov, int iovcnt, off_t offset, int flags, int submit_flags, Duration timeout, io_cancellation * cancellation, io_context * context);
+        extern ssize_t cfa_preadv2(int fd, const struct iovec *iov, int iovcnt, off_t offset, int flags, __u64 submit_flags);
 #endif
 #if defined(CFA_HAVE_PWRITEV2)
-        extern ssize_t cfa_pwritev2(int fd, const struct iovec *iov, int iovcnt, off_t offset, int flags, int submit_flags, Duration timeout, io_cancellation * cancellation, io_context * context);
+        extern ssize_t cfa_pwritev2(int fd, const struct iovec *iov, int iovcnt, off_t offset, int flags, __u64 submit_flags);
 #endif
-extern int cfa_fsync(int fd, int submit_flags, Duration timeout, io_cancellation * cancellation, io_context * context);
-extern int cfa_epoll_ctl(int epfd, int op, int fd, struct epoll_event *event, int submit_flags, Duration timeout, io_cancellation * cancellation, io_context * context);
-extern int cfa_sync_file_range(int fd, off64_t offset, off64_t nbytes, unsigned int flags, int submit_flags, Duration timeout, io_cancellation * cancellation, io_context * context);
-extern  ssize_t cfa_sendmsg(int sockfd, const struct msghdr *msg, int flags, int submit_flags, Duration timeout, io_cancellation * cancellation, io_context * context);
-extern ssize_t cfa_recvmsg(int sockfd, struct msghdr *msg, int flags, int submit_flags, Duration timeout, io_cancellation * cancellation, io_context * context);
-extern ssize_t cfa_send(int sockfd, const void *buf, size_t len, int flags, int submit_flags, Duration timeout, io_cancellation * cancellation, io_context * context);
-extern ssize_t cfa_recv(int sockfd, void *buf, size_t len, int flags, int submit_flags, Duration timeout, io_cancellation * cancellation, io_context * context);
-extern int cfa_accept4(int sockfd, struct sockaddr *addr, socklen_t *addrlen, int flags, int submit_flags, Duration timeout, io_cancellation * cancellation, io_context * context);
-extern int cfa_connect(int sockfd, const struct sockaddr *addr, socklen_t addrlen, int submit_flags, Duration timeout, io_cancellation * cancellation, io_context * context);
-extern int cfa_fallocate(int fd, int mode, off_t offset, off_t len, int submit_flags, Duration timeout, io_cancellation * cancellation, io_context * context);
-extern int cfa_posix_fadvise(int fd, off_t offset, off_t len, int advice, int submit_flags, Duration timeout, io_cancellation * cancellation, io_context * context);
-extern int cfa_madvise(void *addr, size_t length, int advice, int submit_flags, Duration timeout, io_cancellation * cancellation, io_context * context);
-extern int cfa_openat(int dirfd, const char *pathname, int flags, mode_t mode, int submit_flags, Duration timeout, io_cancellation * cancellation, io_context * context);
+extern int cfa_fsync(int fd, __u64 submit_flags);
+extern int cfa_epoll_ctl(int epfd, int op, int fd, struct epoll_event *event, __u64 submit_flags);
+extern int cfa_sync_file_range(int fd, off64_t offset, off64_t nbytes, unsigned int flags, __u64 submit_flags);
+extern  ssize_t cfa_sendmsg(int sockfd, const struct msghdr *msg, int flags, __u64 submit_flags);
+extern ssize_t cfa_recvmsg(int sockfd, struct msghdr *msg, int flags, __u64 submit_flags);
+extern ssize_t cfa_send(int sockfd, const void *buf, size_t len, int flags, __u64 submit_flags);
+extern ssize_t cfa_recv(int sockfd, void *buf, size_t len, int flags, __u64 submit_flags);
+extern int cfa_accept4(int sockfd, struct sockaddr *addr, socklen_t *addrlen, int flags, __u64 submit_flags);
+extern int cfa_connect(int sockfd, const struct sockaddr *addr, socklen_t addrlen, __u64 submit_flags);
+extern int cfa_fallocate(int fd, int mode, off_t offset, off_t len, __u64 submit_flags);
+extern int cfa_posix_fadvise(int fd, off_t offset, off_t len, int advice, __u64 submit_flags);
+extern int cfa_madvise(void *addr, size_t length, int advice, __u64 submit_flags);
+extern int cfa_openat(int dirfd, const char *pathname, int flags, mode_t mode, __u64 submit_flags);
 #if defined(CFA_HAVE_OPENAT2)
-        extern int cfa_openat2(int dirfd, const char *pathname, struct open_how * how, size_t size, int submit_flags, Duration timeout, io_cancellation * cancellation, io_context * context);
+        extern int cfa_openat2(int dirfd, const char *pathname, struct open_how * how, size_t size, __u64 submit_flags);
 #endif
-extern int cfa_close(int fd, int submit_flags, Duration timeout, io_cancellation * cancellation, io_context * context);
+extern int cfa_close(int fd, __u64 submit_flags);
 #if defined(CFA_HAVE_STATX)
-        extern int cfa_statx(int dirfd, const char *pathname, int flags, unsigned int mask, struct statx *statxbuf, int submit_flags, Duration timeout, io_cancellation * cancellation, io_context * context);
+        extern int cfa_statx(int dirfd, const char *pathname, int flags, unsigned int mask, struct statx *statxbuf, __u64 submit_flags);
 #endif
-extern ssize_t cfa_read(int fd, void * buf, size_t count, int submit_flags, Duration timeout, io_cancellation * cancellation, io_context * context);
-extern ssize_t cfa_write(int fd, void * buf, size_t count, int submit_flags, Duration timeout, io_cancellation * cancellation, io_context * context);
-extern ssize_t cfa_splice(int fd_in, loff_t *off_in, int fd_out, loff_t *off_out, size_t len, unsigned int flags, int submit_flags, Duration timeout, io_cancellation * cancellation, io_context * context);
-extern ssize_t cfa_tee(int fd_in, int fd_out, size_t len, unsigned int flags, int submit_flags, Duration timeout, io_cancellation * cancellation, io_context * context);
+extern ssize_t cfa_read(int fd, void * buf, size_t count, __u64 submit_flags);
+extern ssize_t cfa_write(int fd, void * buf, size_t count, __u64 submit_flags);
+extern ssize_t cfa_splice(int fd_in, __off64_t *off_in, int fd_out, __off64_t *off_out, size_t len, unsigned int flags, __u64 submit_flags);
+extern ssize_t cfa_tee(int fd_in, int fd_out, size_t len, unsigned int flags, __u64 submit_flags);
 
 //----------
 // asynchronous calls
 #if defined(CFA_HAVE_PREADV2)
-        extern void async_preadv2(io_future_t & future, int fd, const struct iovec *iov, int iovcnt, off_t offset, int flags, int submit_flags, io_cancellation * cancellation, io_context * context);
+        extern void async_preadv2(io_future_t & future, int fd, const struct iovec *iov, int iovcnt, off_t offset, int flags, __u64 submit_flags);
 #endif
 #if defined(CFA_HAVE_PWRITEV2)
-        extern void async_pwritev2(io_future_t & future, int fd, const struct iovec *iov, int iovcnt, off_t offset, int flags, int submit_flags, io_cancellation * cancellation, io_context * context);
+        extern void async_pwritev2(io_future_t & future, int fd, const struct iovec *iov, int iovcnt, off_t offset, int flags, __u64 submit_flags);
 #endif
-extern void async_fsync(io_future_t & future, int fd, int submit_flags, io_cancellation * cancellation, io_context * context);
-extern void async_epoll_ctl(io_future_t & future, int epfd, int op, int fd, struct epoll_event *event, int submit_flags, io_cancellation * cancellation, io_context * context);
-extern void async_sync_file_range(io_future_t & future, int fd, off64_t offset, off64_t nbytes, unsigned int flags, int submit_flags, io_cancellation * cancellation, io_context * context);
-extern void async_sendmsg(io_future_t & future, int sockfd, const struct msghdr *msg, int flags, int submit_flags, io_cancellation * cancellation, io_context * context);
-extern void async_recvmsg(io_future_t & future, int sockfd, struct msghdr *msg, int flags, int submit_flags, io_cancellation * cancellation, io_context * context);
-extern void async_send(io_future_t & future, int sockfd, const void *buf, size_t len, int flags, int submit_flags, io_cancellation * cancellation, io_context * context);
-extern void async_recv(io_future_t & future, int sockfd, void *buf, size_t len, int flags, int submit_flags, io_cancellation * cancellation, io_context * context);
-extern void async_accept4(io_future_t & future, int sockfd, struct sockaddr *addr, socklen_t *addrlen, int flags, int submit_flags, io_cancellation * cancellation, io_context * context);
-extern void async_connect(io_future_t & future, int sockfd, const struct sockaddr *addr, socklen_t addrlen, int submit_flags, io_cancellation * cancellation, io_context * context);
-extern void async_fallocate(io_future_t & future, int fd, int mode, off_t offset, off_t len, int submit_flags, io_cancellation * cancellation, io_context * context);
-extern void async_posix_fadvise(io_future_t & future, int fd, off_t offset, off_t len, int advice, int submit_flags, io_cancellation * cancellation, io_context * context);
-extern void async_madvise(io_future_t & future, void *addr, size_t length, int advice, int submit_flags, io_cancellation * cancellation, io_context * context);
-extern void async_openat(io_future_t & future, int dirfd, const char *pathname, int flags, mode_t mode, int submit_flags, io_cancellation * cancellation, io_context * context);
+extern void async_fsync(io_future_t & future, int fd, __u64 submit_flags);
+extern void async_epoll_ctl(io_future_t & future, int epfd, int op, int fd, struct epoll_event *event, __u64 submit_flags);
+extern void async_sync_file_range(io_future_t & future, int fd, off64_t offset, off64_t nbytes, unsigned int flags, __u64 submit_flags);
+extern void async_sendmsg(io_future_t & future, int sockfd, const struct msghdr *msg, int flags, __u64 submit_flags);
+extern void async_recvmsg(io_future_t & future, int sockfd, struct msghdr *msg, int flags, __u64 submit_flags);
+extern void async_send(io_future_t & future, int sockfd, const void *buf, size_t len, int flags, __u64 submit_flags);
+extern void async_recv(io_future_t & future, int sockfd, void *buf, size_t len, int flags, __u64 submit_flags);
+extern void async_accept4(io_future_t & future, int sockfd, struct sockaddr *addr, socklen_t *addrlen, int flags, __u64 submit_flags);
+extern void async_connect(io_future_t & future, int sockfd, const struct sockaddr *addr, socklen_t addrlen, __u64 submit_flags);
+extern void async_fallocate(io_future_t & future, int fd, int mode, off_t offset, off_t len, __u64 submit_flags);
+extern void async_posix_fadvise(io_future_t & future, int fd, off_t offset, off_t len, int advice, __u64 submit_flags);
+extern void async_madvise(io_future_t & future, void *addr, size_t length, int advice, __u64 submit_flags);
+extern void async_openat(io_future_t & future, int dirfd, const char *pathname, int flags, mode_t mode, __u64 submit_flags);
 #if defined(CFA_HAVE_OPENAT2)
-        extern void async_openat2(io_future_t & future, int dirfd, const char *pathname, struct open_how * how, size_t size, int submit_flags, io_cancellation * cancellation, io_context * context);
+        extern void async_openat2(io_future_t & future, int dirfd, const char *pathname, struct open_how * how, size_t size, __u64 submit_flags);
 #endif
-extern void async_close(io_future_t & future, int fd, int submit_flags, io_cancellation * cancellation, io_context * context);
+extern void async_close(io_future_t & future, int fd, __u64 submit_flags);
 #if defined(CFA_HAVE_STATX)
-        extern void async_statx(io_future_t & future, int dirfd, const char *pathname, int flags, unsigned int mask, struct statx *statxbuf, int submit_flags, io_cancellation * cancellation, io_context * context);
+        extern void async_statx(io_future_t & future, int dirfd, const char *pathname, int flags, unsigned int mask, struct statx *statxbuf, __u64 submit_flags);
 #endif
-void async_read(io_future_t & future, int fd, void * buf, size_t count, int submit_flags, io_cancellation * cancellation, io_context * context);
-extern void async_write(io_future_t & future, int fd, void * buf, size_t count, int submit_flags, io_cancellation * cancellation, io_context * context);
-extern void async_splice(io_future_t & future, int fd_in, loff_t *off_in, int fd_out, loff_t *off_out, size_t len, unsigned int flags, int submit_flags, io_cancellation * cancellation, io_context * context);
-extern void async_tee(io_future_t & future, int fd_in, int fd_out, size_t len, unsigned int flags, int submit_flags, io_cancellation * cancellation, io_context * context);
+void async_read(io_future_t & future, int fd, void * buf, size_t count, __u64 submit_flags);
+extern void async_write(io_future_t & future, int fd, void * buf, size_t count, __u64 submit_flags);
+extern void async_splice(io_future_t & future, int fd_in, __off64_t *off_in, int fd_out, __off64_t *off_out, size_t len, unsigned int flags, __u64 submit_flags);
+extern void async_tee(io_future_t & future, int fd_in, int fd_out, size_t len, unsigned int flags, __u64 submit_flags);
 
 
@@ -126 +133 @@
 // Check if a function is blocks a only the user thread
 bool has_user_level_blocking( fptr_t func );
-
-//-----------------------------------------------------------------------------
-void register_fixed_files( io_context & ctx , int * files, unsigned count );
-void register_fixed_files( cluster    & cltr, int * files, unsigned count );

libcfa/src/concurrency/kernel.cfa

@@ -22 +22 @@
 #include <signal.h>
 #include <unistd.h>
+extern "C" {
+        #include <sys/eventfd.h>
+}
 
 //CFA Includes
@@ -31 +34 @@
 #include "invoke.h"
 
+#if !defined(__CFA_NO_STATISTICS__)
+        #define __STATS( ...) __VA_ARGS__
+#else
+        #define __STATS( ...)
+#endif
 
 //-----------------------------------------------------------------------------
@@ -107 +115 @@
 static $thread * __next_thread(cluster * this);
 static $thread * __next_thread_slow(cluster * this);
+static inline bool __must_unpark( $thread * thrd ) __attribute((nonnull(1)));
 static void __run_thread(processor * this, $thread * dst);
 static void __wake_one(cluster * cltr);
 
-static void push  (__cluster_idles & idles, processor & proc);
-static void remove(__cluster_idles & idles, processor & proc);
-static [unsigned idle, unsigned total, * processor] query( & __cluster_idles idles );
-
+static void mark_idle (__cluster_proc_list & idles, processor & proc);
+static void mark_awake(__cluster_proc_list & idles, processor & proc);
+static [unsigned idle, unsigned total, * processor] query_idles( & __cluster_proc_list idles );
+
+extern void __cfa_io_start( processor * );
+extern bool __cfa_io_drain( processor * );
+extern void __cfa_io_flush( processor * );
+extern void __cfa_io_stop ( processor * );
+static inline bool __maybe_io_drain( processor * );
+
+extern void __disable_interrupts_hard();
+extern void __enable_interrupts_hard();
+
+static inline void __disable_interrupts_checked() {
+        /* paranoid */ verify( __preemption_enabled() );
+        disable_interrupts();
+        /* paranoid */ verify( ! __preemption_enabled() );
+}
+
+static inline void __enable_interrupts_checked( bool poll = true ) {
+        /* paranoid */ verify( ! __preemption_enabled() );
+        enable_interrupts( poll );
+        /* paranoid */ verify( __preemption_enabled() );
+}
 
 //=============================================================================================
@@ -129 +158 @@
         verify(this);
 
+        __cfa_io_start( this );
+
         __cfadbg_print_safe(runtime_core, "Kernel : core %p starting\n", this);
         #if !defined(__CFA_NO_STATISTICS__)
@@ -140 +171 @@
                 preemption_scope scope = { this };
 
-                #if !defined(__CFA_NO_STATISTICS__)
-                        unsigned long long last_tally = rdtscl();
-                #endif
-
+                __STATS( unsigned long long last_tally = rdtscl(); )
+
+                // if we need to run some special setup, now is the time to do it.
+                if(this->init.thrd) {
+                        this->init.thrd->curr_cluster = this->cltr;
+                        __run_thread(this, this->init.thrd);
+                }
 
                 __cfadbg_print_safe(runtime_core, "Kernel : core %p started\n", this);
@@ -150 +184 @@
                 MAIN_LOOP:
                 for() {
+                        // Check if there is pending io
+                        __maybe_io_drain( this );
+
                         // Try to get the next thread
                         readyThread = __next_thread( this->cltr );
 
                         if( !readyThread ) {
+                                __cfa_io_flush( this );
                                 readyThread = __next_thread_slow( this->cltr );
                         }
@@ -167 +205 @@
 
                                 // Push self to idle stack
-                                push(this->cltr->idles, * this);
+                                mark_idle(this->cltr->procs, * this);
 
                                 // Confirm the ready-queue is empty
@@ -173 +211 @@
                                 if( readyThread ) {
                                         // A thread was found, cancel the halt
-                                        remove(this->cltr->idles, * this);
+                                        mark_awake(this->cltr->procs, * this);
 
                                         #if !defined(__CFA_NO_STATISTICS__)
@@ -189 +227 @@
                                 #endif
 
-                                wait( this->idle );
+                                __cfadbg_print_safe(runtime_core, "Kernel : core %p waiting on eventfd %d\n", this, this->idle);
+
+                                __disable_interrupts_hard();
+                                eventfd_t val;
+                                eventfd_read( this->idle, &val );
+                                __enable_interrupts_hard();
 
                                 #if !defined(__CFA_NO_STATISTICS__)
@@ -198 +241 @@
 
                                 // We were woken up, remove self from idle
-                                remove(this->cltr->idles, * this);
+                                mark_awake(this->cltr->procs, * this);
 
                                 // DON'T just proceed, start looking again
@@ -205 +248 @@
 
                         /* paranoid */ verify( readyThread );
+
+                        // Reset io dirty bit
+                        this->io.dirty = false;
 
                         // We found a thread run it
@@ -219 +265 @@
                                 }
                         #endif
+
+                        if(this->io.pending && !this->io.dirty) {
+                                __cfa_io_flush( this );
+                        }
+
+                //      SEARCH: {
+                //              /* paranoid */ verify( ! __preemption_enabled() );
+                //              /* paranoid */ verify( kernelTLS().this_proc_id );
+
+                //              // First, lock the scheduler since we are searching for a thread
+
+                //              // Try to get the next thread
+                //              ready_schedule_lock();
+                //              readyThread = pop_fast( this->cltr );
+                //              ready_schedule_unlock();
+                //              if(readyThread) {  break SEARCH; }
+
+                //              // If we can't find a thread, might as well flush any outstanding I/O
+                //              if(this->io.pending) { __cfa_io_flush( this ); }
+
+                //              // Spin a little on I/O, just in case
+                //              for(25) {
+                //                      __maybe_io_drain( this );
+                //                      ready_schedule_lock();
+                //                      readyThread = pop_fast( this->cltr );
+                //                      ready_schedule_unlock();
+                //                      if(readyThread) {  break SEARCH; }
+                //              }
+
+                //              // no luck, try stealing a few times
+                //              for(25) {
+                //                      if( __maybe_io_drain( this ) ) {
+                //                              ready_schedule_lock();
+                //                              readyThread = pop_fast( this->cltr );
+                //                      } else {
+                //                              ready_schedule_lock();
+                //                              readyThread = pop_slow( this->cltr );
+                //                      }
+                //                      ready_schedule_unlock();
+                //                      if(readyThread) {  break SEARCH; }
+                //              }
+
+                //              // still no luck, search for a thread
+                //              ready_schedule_lock();
+                //              readyThread = pop_search( this->cltr );
+                //              ready_schedule_unlock();
+                //              if(readyThread) { break SEARCH; }
+
+                //              // Don't block if we are done
+                //              if( __atomic_load_n(&this->do_terminate, __ATOMIC_SEQ_CST) ) break MAIN_LOOP;
+
+                //              __STATS( __tls_stats()->ready.sleep.halts++; )
+
+                //              // Push self to idle stack
+                //              mark_idle(this->cltr->procs, * this);
+
+                //              // Confirm the ready-queue is empty
+                //              __maybe_io_drain( this );
+                //              ready_schedule_lock();
+                //              readyThread = pop_search( this->cltr );
+                //              ready_schedule_unlock();
+
+                //              if( readyThread ) {
+                //                      // A thread was found, cancel the halt
+                //                      mark_awake(this->cltr->procs, * this);
+
+                //                      __STATS( __tls_stats()->ready.sleep.cancels++; )
+
+                //                      // continue the main loop
+                //                      break SEARCH;
+                //              }
+
+                //              __STATS( if(this->print_halts) __cfaabi_bits_print_safe( STDOUT_FILENO, "PH:%d - %lld 0\n", this->id, rdtscl()); )
+                //              __cfadbg_print_safe(runtime_core, "Kernel : core %p waiting on eventfd %d\n", this, this->idle);
+
+                //              // __disable_interrupts_hard();
+                //              eventfd_t val;
+                //              eventfd_read( this->idle, &val );
+                //              // __enable_interrupts_hard();
+
+                //              __STATS( if(this->print_halts) __cfaabi_bits_print_safe( STDOUT_FILENO, "PH:%d - %lld 1\n", this->id, rdtscl()); )
+
+                //              // We were woken up, remove self from idle
+                //              mark_awake(this->cltr->procs, * this);
+
+                //              // DON'T just proceed, start looking again
+                //              continue MAIN_LOOP;
+                //      }
+
+                // RUN_THREAD:
+                //      /* paranoid */ verify( kernelTLS().this_proc_id );
+                //      /* paranoid */ verify( ! __preemption_enabled() );
+                //      /* paranoid */ verify( readyThread );
+
+                //      // Reset io dirty bit
+                //      this->io.dirty = false;
+
+                //      // We found a thread run it
+                //      __run_thread(this, readyThread);
+
+                //      // Are we done?
+                //      if( __atomic_load_n(&this->do_terminate, __ATOMIC_SEQ_CST) ) break MAIN_LOOP;
+
+                //      #if !defined(__CFA_NO_STATISTICS__)
+                //              unsigned long long curr = rdtscl();
+                //              if(curr > (last_tally + 500000000)) {
+                //                      __tally_stats(this->cltr->stats, __cfaabi_tls.this_stats);
+                //                      last_tally = curr;
+                //              }
+                //      #endif
+
+                //      if(this->io.pending && !this->io.dirty) {
+                //              __cfa_io_flush( this );
+                //      }
+
+                //      // Check if there is pending io
+                //      __maybe_io_drain( this );
                 }
 
@@ -224 +387 @@
         }
 
+        __cfa_io_stop( this );
+
         post( this->terminated );
+
 
         if(this == mainProcessor) {
@@ -247 +413 @@
         /* paranoid */ verifyf( thrd_dst->link.next == 0p, "Expected null got %p", thrd_dst->link.next );
         __builtin_prefetch( thrd_dst->context.SP );
+
+        __cfadbg_print_safe(runtime_core, "Kernel : core %p running thread %p (%s)\n", this, thrd_dst, thrd_dst->self_cor.name);
 
         $coroutine * proc_cor = get_coroutine(this->runner);
@@ -297 +465 @@
                 if(unlikely(thrd_dst->preempted != __NO_PREEMPTION)) {
                         // The thread was preempted, reschedule it and reset the flag
-                        __schedule_thread( thrd_dst );
+                        schedule_thread$( thrd_dst );
                         break RUNNING;
                 }
@@ -318 +486 @@
                                 break RUNNING;
                         case TICKET_UNBLOCK:
+                                #if !defined(__CFA_NO_STATISTICS__)
+                                        __tls_stats()->ready.threads.threads++;
+                                        __push_stat( __tls_stats(), __tls_stats()->ready.threads.threads, false, "Processor", this );
+                                #endif
                                 // This is case 2, the racy case, someone tried to run this thread before it finished blocking
                                 // In this case, just run it again.
@@ -330 +502 @@
         proc_cor->state = Active;
 
+        __cfadbg_print_safe(runtime_core, "Kernel : core %p finished running thread %p\n", this, thrd_dst);
+
+        #if !defined(__CFA_NO_STATISTICS__)
+                __tls_stats()->ready.threads.threads--;
+                __push_stat( __tls_stats(), __tls_stats()->ready.threads.threads, false, "Processor", this );
+        #endif
+
         /* paranoid */ verify( ! __preemption_enabled() );
 }
@@ -339 +518 @@
         $thread * thrd_src = kernelTLS().this_thread;
 
-        #if !defined(__CFA_NO_STATISTICS__)
-                struct processor * last_proc = kernelTLS().this_processor;
-        #endif
+        __STATS( thrd_src->last_proc = kernelTLS().this_processor; )
 
         // Run the thread on this processor
@@ -360 +537 @@
 
         #if !defined(__CFA_NO_STATISTICS__)
-                if(last_proc != kernelTLS().this_processor) {
+                /* paranoid */ verify( thrd_src->last_proc != 0p );
+                if(thrd_src->last_proc != kernelTLS().this_processor) {
                         __tls_stats()->ready.threads.migration++;
                 }
@@ -373 +551 @@
 // Scheduler routines
 // KERNEL ONLY
-void __schedule_thread( $thread * thrd ) {
+static void __schedule_thread( $thread * thrd ) {
         /* paranoid */ verify( ! __preemption_enabled() );
         /* paranoid */ verify( kernelTLS().this_proc_id );
+        /* paranoid */ verify( ready_schedule_islocked());
         /* paranoid */ verify( thrd );
         /* paranoid */ verify( thrd->state != Halted );
@@ -391 +570 @@
         if (thrd->preempted == __NO_PREEMPTION) thrd->state = Ready;
 
+        // Dereference the thread now because once we push it, there is not guaranteed it's still valid.
+        struct cluster * cl = thrd->curr_cluster;
+        __STATS(bool outside = thrd->last_proc && thrd->last_proc != kernelTLS().this_processor; )
+
+        // push the thread to the cluster ready-queue
+        push( cl, thrd );
+
+        // variable thrd is no longer safe to use
+        thrd = 0xdeaddeaddeaddeadp;
+
+        // wake the cluster using the save variable.
+        __wake_one( cl );
+
+        #if !defined(__CFA_NO_STATISTICS__)
+                if( kernelTLS().this_stats ) {
+                        __tls_stats()->ready.threads.threads++;
+                        if(outside) {
+                                __tls_stats()->ready.threads.extunpark++;
+                        }
+                        __push_stat( __tls_stats(), __tls_stats()->ready.threads.threads, false, "Processor", kernelTLS().this_processor );
+                }
+                else {
+                        __atomic_fetch_add(&cl->stats->ready.threads.threads, 1, __ATOMIC_RELAXED);
+                        __atomic_fetch_add(&cl->stats->ready.threads.extunpark, 1, __ATOMIC_RELAXED);
+                        __push_stat( cl->stats, cl->stats->ready.threads.threads, true, "Cluster", cl );
+                }
+        #endif
+
+        /* paranoid */ verify( ready_schedule_islocked());
+        /* paranoid */ verify( ! __preemption_enabled() );
+}
+
+void schedule_thread$( $thread * thrd ) {
         ready_schedule_lock();
-                // Dereference the thread now because once we push it, there is not guaranteed it's still valid.
-                struct cluster * cl = thrd->curr_cluster;
-
-                // push the thread to the cluster ready-queue
-                push( cl, thrd );
-
-                // variable thrd is no longer safe to use
-
-                // wake the cluster using the save variable.
-                __wake_one( cl );
+                __schedule_thread( thrd );
         ready_schedule_unlock();
-
-        /* paranoid */ verify( ! __preemption_enabled() );
 }
 
@@ -413 +614 @@
 
         ready_schedule_lock();
-                $thread * thrd = pop( this );
+                $thread * thrd = pop_fast( this );
         ready_schedule_unlock();
 
@@ -427 +628 @@
 
         ready_schedule_lock();
-                $thread * thrd = pop_slow( this );
+                $thread * thrd;
+                for(25) {
+                        thrd = pop_slow( this );
+                        if(thrd) goto RET;
+                }
+                thrd = pop_search( this );
+
+                RET:
         ready_schedule_unlock();
 
@@ -435 +643 @@
 }
 
-void unpark( $thread * thrd ) {
-        if( !thrd ) return;
-
+static inline bool __must_unpark( $thread * thrd ) {
         int old_ticket = __atomic_fetch_add(&thrd->ticket, 1, __ATOMIC_SEQ_CST);
         switch(old_ticket) {
                 case TICKET_RUNNING:
                         // Wake won the race, the thread will reschedule/rerun itself
-                        break;
+                        return false;
                 case TICKET_BLOCKED:
                         /* paranoid */ verify( ! thrd->preempted != __NO_PREEMPTION );
                         /* paranoid */ verify( thrd->state == Blocked );
-
-                        {
-                                /* paranoid */ verify( publicTLS_get(this_proc_id) );
-                                bool full = publicTLS_get(this_proc_id)->full_proc;
-                                if(full) disable_interrupts();
-
-                                /* paranoid */ verify( ! __preemption_enabled() );
-
-                                // Wake lost the race,
-                                __schedule_thread( thrd );
-
-                                /* paranoid */ verify( ! __preemption_enabled() );
-
-                                if(full) enable_interrupts( __cfaabi_dbg_ctx );
-                                /* paranoid */ verify( publicTLS_get(this_proc_id) );
-                        }
-
-                        break;
+                        return true;
                 default:
                         // This makes no sense, something is wrong abort
@@ -470 +659 @@
 }
 
+void __kernel_unpark( $thread * thrd ) {
+        /* paranoid */ verify( ! __preemption_enabled() );
+        /* paranoid */ verify( ready_schedule_islocked());
+
+        if( !thrd ) return;
+
+        if(__must_unpark(thrd)) {
+                // Wake lost the race,
+                __schedule_thread( thrd );
+        }
+
+        /* paranoid */ verify( ready_schedule_islocked());
+        /* paranoid */ verify( ! __preemption_enabled() );
+}
+
+void unpark( $thread * thrd ) {
+        if( !thrd ) return;
+
+        if(__must_unpark(thrd)) {
+                disable_interrupts();
+                        // Wake lost the race,
+                        schedule_thread$( thrd );
+                enable_interrupts(false);
+        }
+}
+
 void park( void ) {
-        /* paranoid */ verify( __preemption_enabled() );
-        disable_interrupts();
-        /* paranoid */ verify( ! __preemption_enabled() );
-        /* paranoid */ verify( kernelTLS().this_thread->preempted == __NO_PREEMPTION );
-
-        returnToKernel();
-
-        /* paranoid */ verify( ! __preemption_enabled() );
-        enable_interrupts( __cfaabi_dbg_ctx );
-        /* paranoid */ verify( __preemption_enabled() );
+        __disable_interrupts_checked();
+                /* paranoid */ verify( kernelTLS().this_thread->preempted == __NO_PREEMPTION );
+                returnToKernel();
+        __enable_interrupts_checked();
 
 }
@@ -520 +729 @@
 // KERNEL ONLY
 bool force_yield( __Preemption_Reason reason ) {
-        /* paranoid */ verify( __preemption_enabled() );
-        disable_interrupts();
-        /* paranoid */ verify( ! __preemption_enabled() );
-
-        $thread * thrd = kernelTLS().this_thread;
-        /* paranoid */ verify(thrd->state == Active);
-
-        // SKULLDUGGERY: It is possible that we are preempting this thread just before
-        // it was going to park itself. If that is the case and it is already using the
-        // intrusive fields then we can't use them to preempt the thread
-        // If that is the case, abandon the preemption.
-        bool preempted = false;
-        if(thrd->link.next == 0p) {
-                preempted = true;
-                thrd->preempted = reason;
-                returnToKernel();
-        }
-
-        /* paranoid */ verify( ! __preemption_enabled() );
-        enable_interrupts_noPoll();
-        /* paranoid */ verify( __preemption_enabled() );
-
+        __disable_interrupts_checked();
+                $thread * thrd = kernelTLS().this_thread;
+                /* paranoid */ verify(thrd->state == Active);
+
+                // SKULLDUGGERY: It is possible that we are preempting this thread just before
+                // it was going to park itself. If that is the case and it is already using the
+                // intrusive fields then we can't use them to preempt the thread
+                // If that is the case, abandon the preemption.
+                bool preempted = false;
+                if(thrd->link.next == 0p) {
+                        preempted = true;
+                        thrd->preempted = reason;
+                        returnToKernel();
+                }
+        __enable_interrupts_checked( false );
         return preempted;
 }
@@ -557 +759 @@
         unsigned idle;
         unsigned total;
-        [idle, total, p] = query(this->idles);
+        [idle, total, p] = query_idles(this->procs);
 
         // If no one is sleeping, we are done
@@ -563 +765 @@
 
         // We found a processor, wake it up
-        post( p->idle );
+        eventfd_t val;
+        val = 1;
+        eventfd_write( p->idle, val );
 
         #if !defined(__CFA_NO_STATISTICS__)
-                __tls_stats()->ready.sleep.wakes++;
+                if( kernelTLS().this_stats ) {
+                        __tls_stats()->ready.sleep.wakes++;
+                }
+                else {
+                        __atomic_fetch_add(&this->stats->ready.sleep.wakes, 1, __ATOMIC_RELAXED);
+                }
         #endif
 
@@ -579 +788 @@
         __cfadbg_print_safe(runtime_core, "Kernel : waking Processor %p\n", this);
 
-        disable_interrupts();
+        __disable_interrupts_checked();
                 /* paranoid */ verify( ! __preemption_enabled() );
-                post( this->idle );
-        enable_interrupts( __cfaabi_dbg_ctx );
-}
-
-static void push  (__cluster_idles & this, processor & proc) {
+                eventfd_t val;
+                val = 1;
+                eventfd_write( this->idle, val );
+        __enable_interrupts_checked();
+}
+
+static void mark_idle(__cluster_proc_list & this, processor & proc) {
         /* paranoid */ verify( ! __preemption_enabled() );
         lock( this );
                 this.idle++;
                 /* paranoid */ verify( this.idle <= this.total );
-
-                insert_first(this.list, proc);
+                remove(proc);
+                insert_first(this.idles, proc);
         unlock( this );
         /* paranoid */ verify( ! __preemption_enabled() );
 }
 
-static void remove(__cluster_idles & this, processor & proc) {
+static void mark_awake(__cluster_proc_list & this, processor & proc) {
         /* paranoid */ verify( ! __preemption_enabled() );
         lock( this );
                 this.idle--;
                 /* paranoid */ verify( this.idle >= 0 );
-
                 remove(proc);
+                insert_last(this.actives, proc);
         unlock( this );
         /* paranoid */ verify( ! __preemption_enabled() );
 }
 
-static [unsigned idle, unsigned total, * processor] query( & __cluster_idles this ) {
+static [unsigned idle, unsigned total, * processor] query_idles( & __cluster_proc_list this ) {
+        /* paranoid */ verify( ! __preemption_enabled() );
+        /* paranoid */ verify( ready_schedule_islocked() );
+
         for() {
                 uint64_t l = __atomic_load_n(&this.lock, __ATOMIC_SEQ_CST);
@@ -613 +827 @@
                 unsigned idle    = this.idle;
                 unsigned total   = this.total;
-                processor * proc = &this.list`first;
+                processor * proc = &this.idles`first;
                 // Compiler fence is unnecessary, but gcc-8 and older incorrectly reorder code without it
                 asm volatile("": : :"memory");
@@ -619 +833 @@
                 return [idle, total, proc];
         }
+
+        /* paranoid */ verify( ready_schedule_islocked() );
+        /* paranoid */ verify( ! __preemption_enabled() );
 }
 
@@ -664 +881 @@
 // Kernel Utilities
 //=============================================================================================
+#if defined(CFA_HAVE_LINUX_IO_URING_H)
+#include "io/types.hfa"
+#endif
+
+static inline bool __maybe_io_drain( processor * proc ) {
+        bool ret = false;
+        #if defined(CFA_HAVE_LINUX_IO_URING_H)
+                __cfadbg_print_safe(runtime_core, "Kernel : core %p checking io for ring %d\n", proc, proc->io.ctx->fd);
+
+                // Check if we should drain the queue
+                $io_context * ctx = proc->io.ctx;
+                unsigned head = *ctx->cq.head;
+                unsigned tail = *ctx->cq.tail;
+                if(head == tail) return false;
+                ready_schedule_lock();
+                ret = __cfa_io_drain( proc );
+                ready_schedule_unlock();
+        #endif
+        return ret;
+}
+
 //-----------------------------------------------------------------------------
 // Debug
@@ -691 +929 @@
                 __print_stats( this.stats, this.print_stats, "Cluster", this.name, (void*)&this );
         }
-
-        extern int __print_alarm_stats;
-        void print_alarm_stats() {
-                __print_alarm_stats = -1;
-        }
 #endif
 // Local Variables: //

libcfa/src/concurrency/kernel.hfa

@@ -28 +28 @@
 }
 
-//-----------------------------------------------------------------------------
-// Underlying Locks
 #ifdef __CFA_WITH_VERIFY__
         extern bool __cfaabi_dbg_in_kernel();
 #endif
 
-extern "C" {
-        char * strerror(int);
-}
-#define CHECKED(x) { int err = x; if( err != 0 ) abort("KERNEL ERROR: Operation \"" #x "\" return error %d - %s\n", err, strerror(err)); }
-
-struct __bin_sem_t {
-        pthread_mutex_t         lock;
-        pthread_cond_t          cond;
-        int                     val;
-};
-
-static inline void ?{}(__bin_sem_t & this) with( this ) {
-        // Create the mutex with error checking
-        pthread_mutexattr_t mattr;
-        pthread_mutexattr_init( &mattr );
-        pthread_mutexattr_settype( &mattr, PTHREAD_MUTEX_ERRORCHECK_NP);
-        pthread_mutex_init(&lock, &mattr);
-
-        pthread_cond_init (&cond, (const pthread_condattr_t *)0p);  // workaround trac#208: cast should not be required
-        val = 0;
-}
-
-static inline void ^?{}(__bin_sem_t & this) with( this ) {
-        CHECKED( pthread_mutex_destroy(&lock) );
-        CHECKED( pthread_cond_destroy (&cond) );
-}
-
-static inline void wait(__bin_sem_t & this) with( this ) {
-        verify(__cfaabi_dbg_in_kernel());
-        CHECKED( pthread_mutex_lock(&lock) );
-                while(val < 1) {
-                        pthread_cond_wait(&cond, &lock);
-                }
-                val -= 1;
-        CHECKED( pthread_mutex_unlock(&lock) );
-}
-
-static inline bool post(__bin_sem_t & this) with( this ) {
-        bool needs_signal = false;
-
-        CHECKED( pthread_mutex_lock(&lock) );
-                if(val < 1) {
-                        val += 1;
-                        pthread_cond_signal(&cond);
-                        needs_signal = true;
-                }
-        CHECKED( pthread_mutex_unlock(&lock) );
-
-        return needs_signal;
-}
-
-#undef CHECKED
-
+//-----------------------------------------------------------------------------
+// I/O
+struct cluster;
+struct $io_context;
+struct $io_arbiter;
+
+struct io_context_params {
+        int num_entries;
+};
+
+void  ?{}(io_context_params & this);
 
 //-----------------------------------------------------------------------------
@@ -95 +51 @@
 struct __processor_id_t {
         unsigned id:24;
-        bool full_proc:1;
 
         #if !defined(__CFA_NO_STATISTICS__)
@@ -114 +69 @@
         struct cluster * cltr;
 
+        // Ready Queue state per processor
+        struct {
+                unsigned short its;
+                unsigned short itr;
+                unsigned id;
+                unsigned target;
+                unsigned long long int cutoff;
+        } rdq;
+
         // Set to true to notify the processor should terminate
         volatile bool do_terminate;
@@ -125 +89 @@
         // Handle to pthreads
         pthread_t kernel_thread;
+
+        struct {
+                $io_context * ctx;
+                bool pending;
+                bool dirty;
+        } io;
 
         // Preemption data
@@ -134 +104 @@
 
         // Idle lock (kernel semaphore)
-        __bin_sem_t idle;
+        int idle;
 
         // Termination synchronisation (user semaphore)
@@ -144 +114 @@
         // Link lists fields
         DLISTED_MGD_IMPL_IN(processor)
+
+        // special init fields
+        // This is needed for memcached integration
+        // once memcached experiments are done this should probably be removed
+        // it is not a particularly safe scheme as it can make processors less homogeneous
+        struct {
+                $thread * thrd;
+        } init;
 
         #if !defined(__CFA_NO_STATISTICS__)
@@ -159 +137 @@
 void ^?{}(processor & this);
 
-static inline void  ?{}(processor & this)                    { this{ "Anonymous Processor", *mainCluster}; }
-static inline void  ?{}(processor & this, struct cluster & cltr)    { this{ "Anonymous Processor", cltr}; }
-static inline void  ?{}(processor & this, const char name[]) { this{name, *mainCluster }; }
+static inline void  ?{}(processor & this)                        { this{ "Anonymous Processor", *mainCluster}; }
+static inline void  ?{}(processor & this, struct cluster & cltr) { this{ "Anonymous Processor", cltr}; }
+static inline void  ?{}(processor & this, const char name[])     { this{name, *mainCluster}; }
 
 DLISTED_MGD_IMPL_OUT(processor)
 
 //-----------------------------------------------------------------------------
-// I/O
-struct __io_data;
-
-// IO poller user-thread
-// Not using the "thread" keyword because we want to control
-// more carefully when to start/stop it
-struct $io_ctx_thread {
-        struct __io_data * ring;
-        single_sem sem;
-        volatile bool done;
-        $thread self;
-};
-
-
-struct io_context {
-        $io_ctx_thread thrd;
-};
-
-struct io_context_params {
-        int num_entries;
-        int num_ready;
-        int submit_aff;
-        bool eager_submits:1;
-        bool poller_submits:1;
-        bool poll_submit:1;
-        bool poll_complete:1;
-};
-
-void  ?{}(io_context_params & this);
-
-void  ?{}(io_context & this, struct cluster & cl);
-void  ?{}(io_context & this, struct cluster & cl, const io_context_params & params);
-void ^?{}(io_context & this);
-
-struct io_cancellation {
-        __u64 target;
-};
-
-static inline void  ?{}(io_cancellation & this) { this.target = -1u; }
-static inline void ^?{}(io_cancellation &) {}
-bool cancel(io_cancellation & this);
-
-//-----------------------------------------------------------------------------
 // Cluster Tools
 
-// Intrusives lanes which are used by the relaxed ready queue
+// Intrusives lanes which are used by the ready queue
 struct __attribute__((aligned(128))) __intrusive_lane_t;
 void  ?{}(__intrusive_lane_t & this);
 void ^?{}(__intrusive_lane_t & this);
 
-// Counter used for wether or not the lanes are all empty
-struct __attribute__((aligned(128))) __snzi_node_t;
-struct __snzi_t {
-        unsigned mask;
-        int root;
-        __snzi_node_t * nodes;
-};
-
-void  ?{}( __snzi_t & this, unsigned depth );
-void ^?{}( __snzi_t & this );
+// Aligned timestamps which are used by the relaxed ready queue
+struct __attribute__((aligned(128))) __timestamp_t;
+void  ?{}(__timestamp_t & this);
+void ^?{}(__timestamp_t & this);
 
 //TODO adjust cache size to ARCHITECTURE
 // Structure holding the relaxed ready queue
 struct __ready_queue_t {
-        // Data tracking how many/which lanes are used
-        // Aligned to 128 for cache locality
-        __snzi_t snzi;
-
         // Data tracking the actual lanes
         // On a seperate cacheline from the used struct since
@@ -242 +167 @@
                 __intrusive_lane_t * volatile data;
 
+                // Array of times
+                __timestamp_t * volatile tscs;
+
                 // Number of lanes (empty or not)
                 volatile size_t count;
@@ -251 +179 @@
 
 // Idle Sleep
-struct __cluster_idles {
+struct __cluster_proc_list {
         // Spin lock protecting the queue
         volatile uint64_t lock;
@@ -262 +190 @@
 
         // List of idle processors
-        dlist(processor, processor) list;
+        dlist(processor, processor) idles;
+
+        // List of active processors
+        dlist(processor, processor) actives;
 };
 
@@ -278 +209 @@
 
         // List of idle processors
-        __cluster_idles idles;
+        __cluster_proc_list procs;
 
         // List of threads
@@ -292 +223 @@
 
         struct {
-                io_context * ctxs;
-                unsigned cnt;
+                $io_arbiter * arbiter;
+                io_context_params params;
         } io;
 

libcfa/src/concurrency/kernel/fwd.hfa

@@ -108 +108 @@
 
         extern void disable_interrupts();
-        extern void enable_interrupts_noPoll();
-        extern void enable_interrupts( __cfaabi_dbg_ctx_param );
+        extern void enable_interrupts( bool poll = false );
 
         extern "Cforall" {
@@ -220 +219 @@
                         // Mark as fulfilled, wake thread if needed
                         // return true if a thread was unparked
-                        bool post(oneshot & this) {
+                        $thread * post(oneshot & this, bool do_unpark = true) {
                                 struct $thread * got = __atomic_exchange_n( &this.ptr, 1p, __ATOMIC_SEQ_CST);
-                                if( got == 0p ) return false;
-                                unpark( got );
-                                return true;
+                                if( got == 0p ) return 0p;
+                                if(do_unpark) unpark( got );
+                                return got;
                         }
                 }
@@ -336 +335 @@
                         // from the server side, mark the future as fulfilled
                         // delete it if needed
-                        bool fulfil( future_t & this ) {
+                        $thread * fulfil( future_t & this, bool do_unpark = true  ) {
                                 for() {
                                         struct oneshot * expected = this.ptr;
@@ -344 +343 @@
                                                 #pragma GCC diagnostic ignored "-Wfree-nonheap-object"
                                         #endif
-                                                if( expected == 3p ) { free( &this ); return false; }
+                                                if( expected == 3p ) { free( &this ); return 0p; }
                                         #if defined(__GNUC__) && __GNUC__ >= 7
                                                 #pragma GCC diagnostic pop
@@ -356 +355 @@
                                         struct oneshot * want = expected == 0p ? 1p : 2p;
                                         if(__atomic_compare_exchange_n(&this.ptr, &expected, want, false, __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST)) {
-                                                if( expected == 0p ) { /* paranoid */ verify( this.ptr == 1p); return false; }
-                                                bool ret = post( *expected );
+                                                if( expected == 0p ) { /* paranoid */ verify( this.ptr == 1p); return 0p; }
+                                                $thread * ret = post( *expected, do_unpark );
                                                 __atomic_store_n( &this.ptr, 1p, __ATOMIC_SEQ_CST);
                                                 return ret;
@@ -403 +402 @@
                                         __VA_ARGS__ \
                                 } \
-                                if( !(in_kernel) ) enable_interrupts( __cfaabi_dbg_ctx ); \
+                                if( !(in_kernel) ) enable_interrupts(); \
                         }
                 #else

libcfa/src/concurrency/kernel/startup.cfa

@@ -22 +22 @@
 extern "C" {
       #include <limits.h>       // PTHREAD_STACK_MIN
+        #include <sys/eventfd.h>  // eventfd
       #include <sys/mman.h>     // mprotect
       #include <sys/resource.h> // getrlimit
@@ -72 +73 @@
 static void __kernel_first_resume( processor * this );
 static void __kernel_last_resume ( processor * this );
-static void init(processor & this, const char name[], cluster & _cltr);
+static void init(processor & this, const char name[], cluster & _cltr, $thread * initT);
 static void deinit(processor & this);
 static void doregister( struct cluster & cltr );
@@ -89 +90 @@
 extern void __kernel_alarm_startup(void);
 extern void __kernel_alarm_shutdown(void);
-extern void __kernel_io_startup (void);
-extern void __kernel_io_shutdown(void);
 
 //-----------------------------------------------------------------------------
@@ -102 +101 @@
 KERNEL_STORAGE($thread,              mainThread);
 KERNEL_STORAGE(__stack_t,            mainThreadCtx);
-KERNEL_STORAGE(io_context,           mainPollerThread);
 KERNEL_STORAGE(__scheduler_RWLock_t, __scheduler_lock);
 #if !defined(__CFA_NO_STATISTICS__)
@@ -198 +196 @@
 
         void ?{}(processor & this) with( this ) {
-                ( this.idle ){};
                 ( this.terminated ){};
                 ( this.runner ){};
-                init( this, "Main Processor", *mainCluster );
+                init( this, "Main Processor", *mainCluster, 0p );
                 kernel_thread = pthread_self();
 
@@ -226 +223 @@
         __kernel_alarm_startup();
 
-        // Start IO
-        __kernel_io_startup();
-
         // Add the main thread to the ready queue
         // once resume is called on mainProcessor->runner the mainThread needs to be scheduled like any normal thread
-        __schedule_thread(mainThread);
+        schedule_thread$(mainThread);
 
         // SKULLDUGGERY: Force a context switch to the main processor to set the main thread's context to the current UNIX
@@ -241 +235 @@
         // THE SYSTEM IS NOW COMPLETELY RUNNING
 
-
-        // SKULLDUGGERY: The constructor for the mainCluster will call alloc with a dimension of 0
-        // malloc *can* return a non-null value, we should free it if that is the case
-        free( mainCluster->io.ctxs );
-
-        // Now that the system is up, finish creating systems that need threading
-        mainCluster->io.ctxs = (io_context *)&storage_mainPollerThread;
-        mainCluster->io.cnt  = 1;
-        (*mainCluster->io.ctxs){ *mainCluster };
-
         __cfadbg_print_safe(runtime_core, "Kernel : Started\n--------------------------------------------------\n\n");
 
         /* paranoid */ verify( ! __preemption_enabled() );
-        enable_interrupts( __cfaabi_dbg_ctx );
+        enable_interrupts();
         /* paranoid */ verify( __preemption_enabled() );
 
@@ -260 +244 @@
 
 static void __kernel_shutdown(void) {
-        //Before we start shutting things down, wait for systems that need threading to shutdown
-        ^(*mainCluster->io.ctxs){};
-        mainCluster->io.cnt  = 0;
-        mainCluster->io.ctxs = 0p;
-
         /* paranoid */ verify( __preemption_enabled() );
         disable_interrupts();
@@ -283 +262 @@
         __kernel_alarm_shutdown();
 
-        // Stop IO
-        __kernel_io_shutdown();
+        #if !defined( __CFA_NO_STATISTICS__ )
+                __stats_t * st = (__stats_t *)& storage_mainProcStats;
+                __tally_stats(mainCluster->stats, st);
+                if( 0 != mainProcessor->print_stats ) {
+                        __print_stats( st, mainProcessor->print_stats, "Processor ", mainProcessor->name, (void*)mainProcessor );
+                }
+                #if defined(CFA_STATS_ARRAY)
+                        __flush_stat( st, "Processor", mainProcessor );
+                #endif
+        #endif
 
         // Destroy the main processor and its context in reverse order of construction
@@ -364 +351 @@
                         __print_stats( &local_stats, proc->print_stats, "Processor ", proc->name, (void*)proc );
                 }
+                #if defined(CFA_STATS_ARRAY)
+                        __flush_stat( &local_stats, "Processor", proc );
+                #endif
         #endif
 
@@ -457 +447 @@
         link.next = 0p;
         link.prev = 0p;
+        link.preferred = -1u;
+        last_proc = 0p;
         #if defined( __CFA_WITH_VERIFY__ )
                 canary = 0x0D15EA5E0D15EA5Ep;
@@ -476 +468 @@
 }
 
-static void init(processor & this, const char name[], cluster & _cltr) with( this ) {
+static void init(processor & this, const char name[], cluster & _cltr, $thread * initT) with( this ) {
         this.name = name;
         this.cltr = &_cltr;
-        full_proc = true;
+        this.rdq.its = 0;
+        this.rdq.itr = 0;
+        this.rdq.id  = -1u;
+        this.rdq.target = -1u;
+        this.rdq.cutoff = -1ull;
         do_terminate = false;
         preemption_alarm = 0p;
         pending_preemption = false;
 
+        this.io.ctx = 0p;
+        this.io.pending = false;
+        this.io.dirty   = false;
+
+        this.init.thrd = initT;
+
+        this.idle = eventfd(0, 0);
+        if (idle < 0) {
+                abort("KERNEL ERROR: PROCESSOR EVENTFD - %s\n", strerror(errno));
+        }
+
         #if !defined(__CFA_NO_STATISTICS__)
                 print_stats = 0;
@@ -489 +496 @@
         #endif
 
-        lock( this.cltr->idles );
-                int target = this.cltr->idles.total += 1u;
-        unlock( this.cltr->idles );
-
-        id = doregister((__processor_id_t*)&this);
-
+        // Register and Lock the RWlock so no-one pushes/pops while we are changing the queue
+        uint_fast32_t last_size = ready_mutate_register((__processor_id_t*)&this);
+                this.cltr->procs.total += 1u;
+                insert_last(this.cltr->procs.actives, this);
+
+                // Adjust the ready queue size
+                ready_queue_grow( cltr );
+
+        // Unlock the RWlock
+        ready_mutate_unlock( last_size );
+
+        __cfadbg_print_safe(runtime_core, "Kernel : core %p created\n", &this);
+}
+
+// Not a ctor, it just preps the destruction but should not destroy members
+static void deinit(processor & this) {
         // Lock the RWlock so no-one pushes/pops while we are changing the queue
         uint_fast32_t last_size = ready_mutate_lock();
+                this.cltr->procs.total -= 1u;
+                remove(this);
 
                 // Adjust the ready queue size
-                ready_queue_grow( cltr, target );
-
-        // Unlock the RWlock
-        ready_mutate_unlock( last_size );
-
-        __cfadbg_print_safe(runtime_core, "Kernel : core %p created\n", &this);
-}
-
-// Not a ctor, it just preps the destruction but should not destroy members
-static void deinit(processor & this) {
-        lock( this.cltr->idles );
-                int target = this.cltr->idles.total -= 1u;
-        unlock( this.cltr->idles );
-
-        // Lock the RWlock so no-one pushes/pops while we are changing the queue
-        uint_fast32_t last_size = ready_mutate_lock();
-
-                // Adjust the ready queue size
-                ready_queue_shrink( this.cltr, target );
-
-        // Unlock the RWlock
-        ready_mutate_unlock( last_size );
-
-        // Finally we don't need the read_lock any more
-        unregister((__processor_id_t*)&this);
-}
-
-void ?{}(processor & this, const char name[], cluster & _cltr) {
-        ( this.idle ){};
+                ready_queue_shrink( this.cltr );
+
+        // Unlock the RWlock and unregister: we don't need the read_lock any more
+        ready_mutate_unregister((__processor_id_t*)&this, last_size );
+
+        close(this.idle);
+}
+
+void ?{}(processor & this, const char name[], cluster & _cltr, $thread * initT) {
         ( this.terminated ){};
         ( this.runner ){};
 
         disable_interrupts();
-                init( this, name, _cltr );
-        enable_interrupts( __cfaabi_dbg_ctx );
+                init( this, name, _cltr, initT );
+        enable_interrupts();
 
         __cfadbg_print_safe(runtime_core, "Kernel : Starting core %p\n", &this);
 
         this.stack = __create_pthread( &this.kernel_thread, __invoke_processor, (void *)&this );
-
+}
+
+void ?{}(processor & this, const char name[], cluster & _cltr) {
+        (this){name, _cltr, 0p};
 }
 
@@ -557 +559 @@
         disable_interrupts();
                 deinit( this );
-        enable_interrupts( __cfaabi_dbg_ctx );
+        enable_interrupts();
 }
 
 //-----------------------------------------------------------------------------
 // Cluster
-static void ?{}(__cluster_idles & this) {
+static void ?{}(__cluster_proc_list & this) {
         this.lock  = 0;
         this.idle  = 0;
         this.total = 0;
-        (this.list){};
 }
 
@@ -582 +583 @@
         threads{ __get };
 
+        io.arbiter = create();
+        io.params = io_params;
+
         doregister(this);
 
@@ -589 +593 @@
 
                 // Adjust the ready queue size
-                ready_queue_grow( &this, 0 );
+                ready_queue_grow( &this );
 
         // Unlock the RWlock
         ready_mutate_unlock( last_size );
-        enable_interrupts_noPoll(); // Don't poll, could be in main cluster
-
-
-        this.io.cnt  = num_io;
-        this.io.ctxs = aalloc(num_io);
-        for(i; this.io.cnt) {
-                (this.io.ctxs[i]){ this, io_params };
-        }
+        enable_interrupts( false ); // Don't poll, could be in main cluster
 }
 
 void ^?{}(cluster & this) {
-        for(i; this.io.cnt) {
-                ^(this.io.ctxs[i]){ true };
-        }
-        free(this.io.ctxs);
+        destroy(this.io.arbiter);
 
         // Lock the RWlock so no-one pushes/pops while we are changing the queue
@@ -614 +608 @@
 
                 // Adjust the ready queue size
-                ready_queue_shrink( &this, 0 );
+                ready_queue_shrink( &this );
 
         // Unlock the RWlock
         ready_mutate_unlock( last_size );
-        enable_interrupts_noPoll(); // Don't poll, could be in main cluster
+        enable_interrupts( false ); // Don't poll, could be in main cluster
 
         #if !defined(__CFA_NO_STATISTICS__)
@@ -624 +618 @@
                         __print_stats( this.stats, this.print_stats, "Cluster", this.name, (void*)&this );
                 }
+                #if defined(CFA_STATS_ARRAY)
+                        __flush_stat( this.stats, "Cluster", &this );
+                #endif
                 free( this.stats );
         #endif
@@ -736 +733 @@
 }
 
-
 #if defined(__CFA_WITH_VERIFY__)
 static bool verify_fwd_bck_rng(void) {

libcfa/src/concurrency/kernel_private.hfa

@@ -29 +29 @@
 extern "C" {
         void disable_interrupts() OPTIONAL_THREAD;
-        void enable_interrupts_noPoll();
-        void enable_interrupts( __cfaabi_dbg_ctx_param );
-}
-
-void __schedule_thread( $thread * )
-#if defined(NDEBUG) || (!defined(__CFA_DEBUG__) && !defined(__CFA_VERIFY__))
-        __attribute__((nonnull (1)))
-#endif
-;
+        void enable_interrupts( bool poll = true );
+}
+
+void schedule_thread$( $thread * ) __attribute__((nonnull (1)));
 
 extern bool __preemption_enabled();
@@ -77 +72 @@
 //-----------------------------------------------------------------------------
 // I/O
-void ^?{}(io_context & this, bool );
+$io_arbiter * create(void);
+void destroy($io_arbiter *);
 
 //=======================================================================
 // Cluster lock API
 //=======================================================================
-// Cells use by the reader writer lock
-// while not generic it only relies on a opaque pointer
-struct __attribute__((aligned(128))) __scheduler_lock_id_t {
-        // Spin lock used as the underlying lock
-        volatile bool lock;
-
-        // Handle pointing to the proc owning this cell
-        // Used for allocating cells and debugging
-        __processor_id_t * volatile handle;
-
-        #ifdef __CFA_WITH_VERIFY__
-                // Debug, check if this is owned for reading
-                bool owned;
-        #endif
-};
-
-static_assert( sizeof(struct __scheduler_lock_id_t) <= __alignof(struct __scheduler_lock_id_t));
-
 // Lock-Free registering/unregistering of threads
 // Register a processor to a given cluster and get its unique id in return
-unsigned doregister( struct __processor_id_t * proc );
+void register_proc_id( struct __processor_id_t * );
 
 // Unregister a processor from a given cluster using its id, getting back the original pointer
-void     unregister( struct __processor_id_t * proc );
-
-//-----------------------------------------------------------------------
-// Cluster idle lock/unlock
-static inline void lock(__cluster_idles & this) {
-        for() {
-                uint64_t l = this.lock;
-                if(
-                        (0 == (l % 2))
-                        && __atomic_compare_exchange_n(&this.lock, &l, l + 1, false, __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST)
-                ) return;
-                Pause();
-        }
-}
-
-static inline void unlock(__cluster_idles & this) {
-        /* paranoid */ verify( 1 == (this.lock % 2) );
-        __atomic_fetch_add( &this.lock, 1, __ATOMIC_SEQ_CST );
-}
+void unregister_proc_id( struct __processor_id_t * proc );
 
 //=======================================================================
@@ -151 +111 @@
         __atomic_store_n(ll, (bool)false, __ATOMIC_RELEASE);
 }
+
+// Cells use by the reader writer lock
+// while not generic it only relies on a opaque pointer
+struct __attribute__((aligned(128))) __scheduler_lock_id_t {
+        // Spin lock used as the underlying lock
+        volatile bool lock;
+
+        // Handle pointing to the proc owning this cell
+        // Used for allocating cells and debugging
+        __processor_id_t * volatile handle;
+
+        #ifdef __CFA_WITH_VERIFY__
+                // Debug, check if this is owned for reading
+                bool owned;
+        #endif
+};
+
+static_assert( sizeof(struct __scheduler_lock_id_t) <= __alignof(struct __scheduler_lock_id_t));
 
 //-----------------------------------------------------------------------
@@ -246 +224 @@
 void ready_mutate_unlock( uint_fast32_t /* value returned by lock */ );
 
+//-----------------------------------------------------------------------
+// Lock-Free registering/unregistering of threads
+// Register a processor to a given cluster and get its unique id in return
+// For convenience, also acquires the lock
+static inline uint_fast32_t ready_mutate_register( struct __processor_id_t * proc ) {
+        register_proc_id( proc );
+        return ready_mutate_lock();
+}
+
+// Unregister a processor from a given cluster using its id, getting back the original pointer
+// assumes the lock is acquired
+static inline void ready_mutate_unregister( struct __processor_id_t * proc, uint_fast32_t last_s ) {
+        ready_mutate_unlock( last_s );
+        unregister_proc_id( proc );
+}
+
+//-----------------------------------------------------------------------
+// Cluster idle lock/unlock
+static inline void lock(__cluster_proc_list & this) {
+        /* paranoid */ verify( ! __preemption_enabled() );
+
+        // Start by locking the global RWlock so that we know no-one is
+        // adding/removing processors while we mess with the idle lock
+        ready_schedule_lock();
+
+        // Simple counting lock, acquired, acquired by incrementing the counter
+        // to an odd number
+        for() {
+                uint64_t l = this.lock;
+                if(
+                        (0 == (l % 2))
+                        && __atomic_compare_exchange_n(&this.lock, &l, l + 1, false, __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST)
+                ) return;
+                Pause();
+        }
+
+        /* paranoid */ verify( ! __preemption_enabled() );
+}
+
+static inline void unlock(__cluster_proc_list & this) {
+        /* paranoid */ verify( ! __preemption_enabled() );
+
+        /* paranoid */ verify( 1 == (this.lock % 2) );
+        // Simple couting lock, release by incrementing to an even number
+        __atomic_fetch_add( &this.lock, 1, __ATOMIC_SEQ_CST );
+
+        // Release the global lock, which we acquired when locking
+        ready_schedule_unlock();
+
+        /* paranoid */ verify( ! __preemption_enabled() );
+}
+
 //=======================================================================
 // Ready-Queue API
 //-----------------------------------------------------------------------
-// pop thread from the ready queue of a cluster
-// returns 0p if empty
-__attribute__((hot)) bool query(struct cluster * cltr);
-
-//-----------------------------------------------------------------------
 // push thread onto a ready queue for a cluster
 // returns true if the list was previously empty, false otherwise
-__attribute__((hot)) bool push(struct cluster * cltr, struct $thread * thrd);
-
-//-----------------------------------------------------------------------
-// pop thread from the ready queue of a cluster
+__attribute__((hot)) void push(struct cluster * cltr, struct $thread * thrd);
+
+//-----------------------------------------------------------------------
+// pop thread from the local queues of a cluster
 // returns 0p if empty
 // May return 0p spuriously
-__attribute__((hot)) struct $thread * pop(struct cluster * cltr);
-
-//-----------------------------------------------------------------------
-// pop thread from the ready queue of a cluster
+__attribute__((hot)) struct $thread * pop_fast(struct cluster * cltr);
+
+//-----------------------------------------------------------------------
+// pop thread from any ready queue of a cluster
+// returns 0p if empty
+// May return 0p spuriously
+__attribute__((hot)) struct $thread * pop_slow(struct cluster * cltr);
+
+//-----------------------------------------------------------------------
+// search all ready queues of a cluster for any thread
 // returns 0p if empty
 // guaranteed to find any threads added before this call
-__attribute__((hot)) struct $thread * pop_slow(struct cluster * cltr);
-
-//-----------------------------------------------------------------------
-// remove thread from the ready queue of a cluster
-// returns bool if it wasn't found
-bool remove_head(struct cluster * cltr, struct $thread * thrd);
+__attribute__((hot)) struct $thread * pop_search(struct cluster * cltr);
 
 //-----------------------------------------------------------------------
 // Increase the width of the ready queue (number of lanes) by 4
-void ready_queue_grow  (struct cluster * cltr, int target);
+void ready_queue_grow  (struct cluster * cltr);
 
 //-----------------------------------------------------------------------
 // Decrease the width of the ready queue (number of lanes) by 4
-void ready_queue_shrink(struct cluster * cltr, int target);
+void ready_queue_shrink(struct cluster * cltr);
 
 

libcfa/src/concurrency/locks.cfa

@@ -134 +134 @@
         lock( lock __cfaabi_dbg_ctx2 );
         /* paranoid */ verifyf( owner != 0p, "Attempt to release lock %p that isn't held", &this );
-        /* paranoid */ verifyf( owner == active_thread() || !strict_owner, "Thread %p other than the owner %p attempted to release owner lock %p", owner, active_thread(), &this );
+        /* paranoid */ verifyf( owner == active_thread() || !strict_owner , "Thread %p other than the owner %p attempted to release owner lock %p", owner, active_thread(), &this );
+        /* paranoid */ verifyf( recursion_count == 1 || multi_acquisition, "Thread %p attempted to release owner lock %p which is not recursive but has a recursive count of %zu", active_thread(), &this, recursion_count );
 
         // if recursion count is zero release lock and set new owner if one is waiting
@@ -146 +147 @@
 size_t wait_count( blocking_lock & this ) with( this ) {
         return wait_count;
-}
-
-void set_recursion_count( blocking_lock & this, size_t recursion ) with( this ) {
-        recursion_count = recursion;
-}
-
-size_t get_recursion_count( blocking_lock & this ) with( this ) {
-        return recursion_count;
 }
 
@@ -173 +166 @@
 }
 
-void on_wait( blocking_lock & this ) with( this ) {
+size_t on_wait( blocking_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() || !strict_owner, "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 );
         unlock( lock );
+        return ret;
+}
+
+void on_wakeup( blocking_lock & this, size_t recursion ) with( this ) {
+        recursion_count = recursion;
 }
 
@@ -274 +274 @@
         }
 
-        bool empty( condition_variable(L) & this ) with(this) { return empty(blocked_threads); }
+        bool empty( condition_variable(L) & this ) with(this) {
+                lock( lock __cfaabi_dbg_ctx2 );
+                bool ret = empty(blocked_threads);
+                unlock( lock );
+                return ret;
+        }
 
         int counter( condition_variable(L) & this ) with(this) { return count; }
@@ -285 +290 @@
                 if (i->lock) {
                         // if lock was passed get recursion count to reset to after waking thread
-                        recursion_count = get_recursion_count(*i->lock);
-                        on_wait( *i->lock );
+                        recursion_count = on_wait( *i->lock );
                 }
                 return recursion_count;
@@ -301 +305 @@
 
                 // resets recursion count here after waking
-                if (i.lock) set_recursion_count(*i.lock, recursion_count);
+                if (i.lock) on_wakeup(*i.lock, recursion_count);
         }
 
@@ -323 +327 @@
 
                 // resets recursion count here after waking
-                if (info.lock) set_recursion_count(*info.lock, recursion_count);
+                if (info.lock) on_wakeup(*info.lock, recursion_count);
         }
 
@@ -373 +377 @@
 }
 
-bool V(semaphore & this) with( this ) {
+$thread * V (semaphore & this, const bool doUnpark ) with( this ) {
         $thread * thrd = 0p;
         lock( lock __cfaabi_dbg_ctx2 );
@@ -385 +389 @@
 
         // make new owner
-        unpark( thrd );
-
+        if( doUnpark ) unpark( thrd );
+
+        return thrd;
+}
+
+bool V(semaphore & this) with( this ) {
+        $thread * thrd = V(this, true);
         return thrd != 0p;
 }

libcfa/src/concurrency/locks.hfa

@@ -20 +20 @@
 
 #include "bits/weakso_locks.hfa"
+#include "containers/queueLockFree.hfa"
+
+#include "thread.hfa"
 
 #include "time_t.hfa"
 #include "time.hfa"
+
+//-----------------------------------------------------------------------------
+// Semaphores
+
+// '0-nary' semaphore
+// Similar to a counting semaphore except the value of one is never reached
+// as a consequence, a V() that would bring the value to 1 *spins* until
+// a P consumes it
+struct Semaphore0nary {
+        __spinlock_t lock; // needed to protect
+        mpsc_queue($thread) queue;
+};
+
+static inline bool P(Semaphore0nary & this, $thread * thrd) {
+        /* paranoid */ verify(!(thrd->seqable.next));
+        /* paranoid */ verify(!(thrd`next));
+
+        push(this.queue, thrd);
+        return true;
+}
+
+static inline bool P(Semaphore0nary & this) {
+    $thread * thrd = active_thread();
+    P(this, thrd);
+    park();
+    return true;
+}
+
+static inline $thread * V(Semaphore0nary & this, bool doUnpark = true) {
+        $thread * next;
+        lock(this.lock __cfaabi_dbg_ctx2);
+                for (;;) {
+                        next = pop(this.queue);
+                        if (next) break;
+                        Pause();
+                }
+        unlock(this.lock);
+
+        if (doUnpark) unpark(next);
+        return next;
+}
+
+// Wrapper used on top of any sempahore to avoid potential locking
+struct BinaryBenaphore {
+        volatile ssize_t counter;
+};
+
+static inline {
+        void ?{}(BinaryBenaphore & this) { this.counter = 0; }
+        void ?{}(BinaryBenaphore & this, zero_t) { this.counter = 0; }
+        void ?{}(BinaryBenaphore & this, one_t ) { this.counter = 1; }
+
+        // returns true if no blocking needed
+        bool P(BinaryBenaphore & this) {
+                return __atomic_fetch_sub(&this.counter, 1, __ATOMIC_SEQ_CST) > 0;
+        }
+
+        bool tryP(BinaryBenaphore & this) {
+                ssize_t c = this.counter;
+                return (c >= 1) && __atomic_compare_exchange_n(&this.counter, &c, c-1, false, __ATOMIC_SEQ_CST, __ATOMIC_RELAXED);
+        }
+
+        // returns true if notify needed
+        bool V(BinaryBenaphore & this) {
+                ssize_t c = 0;
+                for () {
+                        if (__atomic_compare_exchange_n(&this.counter, &c, c+1, false, __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST)) {
+                                if (c == 0) return true;
+                                /* paranoid */ verify(c < 0);
+                                return false;
+                        } else {
+                                if (c == 1) return true;
+                                /* paranoid */ verify(c < 1);
+                                Pause();
+                        }
+                }
+        }
+}
+
+// Binary Semaphore based on the BinaryBenaphore on top of the 0-nary Semaphore
+struct ThreadBenaphore {
+        BinaryBenaphore ben;
+        Semaphore0nary  sem;
+};
+
+static inline void ?{}(ThreadBenaphore & this) {}
+static inline void ?{}(ThreadBenaphore & this, zero_t) { (this.ben){ 0 }; }
+static inline void ?{}(ThreadBenaphore & this, one_t ) { (this.ben){ 1 }; }
+
+static inline bool P(ThreadBenaphore & this)              { return P(this.ben) ? false : P(this.sem); }
+static inline bool tryP(ThreadBenaphore & this)           { return tryP(this.ben); }
+static inline bool P(ThreadBenaphore & this, bool wait)   { return wait ? P(this) : tryP(this); }
+
+static inline $thread * V(ThreadBenaphore & this, bool doUnpark = true) {
+        if (V(this.ben)) return 0p;
+        return V(this.sem, doUnpark);
+}
+
+//-----------------------------------------------------------------------------
+// Semaphore
+struct semaphore {
+        __spinlock_t lock;
+        int count;
+        __queue_t($thread) waiting;
+};
+
+void  ?{}(semaphore & this, int count = 1);
+void ^?{}(semaphore & this);
+bool   P (semaphore & this);
+bool   V (semaphore & this);
+bool   V (semaphore & this, unsigned count);
+$thread * V (semaphore & this, bool );
 
 //----------
@@ -31 +146 @@
 static inline void  ?{}( single_acquisition_lock & this ) {((blocking_lock &)this){ false, false };}
 static inline void ^?{}( single_acquisition_lock & this ) {}
-static inline void   lock      ( single_acquisition_lock & this ) { lock   ( (blocking_lock &)this ); }
-static inline void   unlock    ( single_acquisition_lock & this ) { unlock ( (blocking_lock &)this ); }
-static inline void   on_wait   ( single_acquisition_lock & this ) { on_wait( (blocking_lock &)this ); }
-static inline void   on_notify ( single_acquisition_lock & this, struct $thread * t ) { on_notify( (blocking_lock &)this, t ); }
-static inline void   set_recursion_count( single_acquisition_lock & this, size_t recursion ) { set_recursion_count( (blocking_lock &)this, recursion ); }
-static inline size_t get_recursion_count( single_acquisition_lock & this ) { return get_recursion_count( (blocking_lock &)this ); }
+static inline void   lock     ( single_acquisition_lock & this ) { lock    ( (blocking_lock &)this ); }
+static inline bool   try_lock ( single_acquisition_lock & this ) { return try_lock( (blocking_lock &)this ); }
+static inline void   unlock   ( single_acquisition_lock & this ) { unlock  ( (blocking_lock &)this ); }
+static inline size_t on_wait  ( single_acquisition_lock & this ) { return on_wait ( (blocking_lock &)this ); }
+static inline void   on_wakeup( single_acquisition_lock & this, size_t v ) { on_wakeup ( (blocking_lock &)this, v ); }
+static inline void   on_notify( single_acquisition_lock & this, struct $thread * t ) { on_notify( (blocking_lock &)this, t ); }
 
 //----------
@@ -45 +160 @@
 static inline void  ?{}( owner_lock & this ) {((blocking_lock &)this){ true, true };}
 static inline void ^?{}( owner_lock & this ) {}
-static inline void   lock     ( owner_lock & this ) { lock   ( (blocking_lock &)this ); }
-static inline void   unlock   ( owner_lock & this ) { unlock ( (blocking_lock &)this ); }
-static inline void   on_wait  ( owner_lock & this ) { on_wait( (blocking_lock &)this ); }
+static inline void   lock     ( owner_lock & this ) { lock    ( (blocking_lock &)this ); }
+static inline bool   try_lock ( owner_lock & this ) { return try_lock( (blocking_lock &)this ); }
+static inline void   unlock   ( owner_lock & this ) { unlock  ( (blocking_lock &)this ); }
+static inline size_t on_wait  ( owner_lock & this ) { return on_wait ( (blocking_lock &)this ); }
+static inline void   on_wakeup( owner_lock & this, size_t v ) { on_wakeup ( (blocking_lock &)this, v ); }
 static inline void   on_notify( owner_lock & this, struct $thread * t ) { on_notify( (blocking_lock &)this, t ); }
-static inline void   set_recursion_count( owner_lock & this, size_t recursion ) { set_recursion_count( (blocking_lock &)this, recursion ); }
-static inline size_t get_recursion_count( owner_lock & this ) { return get_recursion_count( (blocking_lock &)this ); }
+
+struct fast_lock {
+        $thread * volatile owner;
+        ThreadBenaphore sem;
+};
+
+static inline bool $try_lock(fast_lock & this, $thread * thrd) {
+    $thread * exp = 0p;
+    return __atomic_compare_exchange_n(&this.owner, &exp, thrd, false, __ATOMIC_SEQ_CST, __ATOMIC_RELAXED);
+}
+
+static inline void lock( fast_lock & this ) __attribute__((artificial));
+static inline void lock( fast_lock & this ) {
+        $thread * thrd = active_thread();
+        /* paranoid */verify(thrd != this.owner);
+
+        for (;;) {
+                if ($try_lock(this, thrd)) return;
+                P(this.sem);
+        }
+}
+
+static inline bool try_lock( fast_lock & this ) __attribute__((artificial));
+static inline bool try_lock ( fast_lock & this ) {
+        $thread * thrd = active_thread();
+        /* paranoid */ verify(thrd != this.owner);
+        return $try_lock(this, thrd);
+}
+
+static inline $thread * unlock( fast_lock & this ) __attribute__((artificial));
+static inline $thread * unlock( fast_lock & this ) {
+        /* paranoid */ verify(active_thread() == this.owner);
+
+        // open 'owner' before unlocking anyone
+        // so new and unlocked threads don't park incorrectly.
+        // This may require additional fencing on ARM.
+        this.owner = 0p;
+
+        return V(this.sem);
+}
+
+static inline size_t on_wait( fast_lock & this ) { unlock(this); return 0; }
+static inline void on_wakeup( fast_lock & this, size_t ) { lock(this); }
+static inline void on_notify( fast_lock &, struct $thread * t ) { unpark(t); }
+
+struct mcs_node {
+        mcs_node * volatile next;
+        single_sem sem;
+};
+
+static inline void ?{}(mcs_node & this) { this.next = 0p; }
+
+static inline mcs_node * volatile & ?`next ( mcs_node * node ) {
+        return node->next;
+}
+
+struct mcs_lock {
+        mcs_queue(mcs_node) queue;
+};
+
+static inline void lock(mcs_lock & l, mcs_node & n) {
+        if(push(l.queue, &n))
+                wait(n.sem);
+}
+
+static inline void unlock(mcs_lock & l, mcs_node & n) {
+        mcs_node * next = advance(l.queue, &n);
+        if(next) post(next->sem);
+}
 
 //-----------------------------------------------------------------------------
@@ -59 +243 @@
 
         // For synchronization locks to use when releasing
-        void on_wait( L & );
-
-        // to get recursion count for cond lock to reset after waking
-        size_t get_recursion_count( L & );
+        size_t on_wait( L & );
 
         // to set recursion count after getting signalled;
-        void set_recursion_count( L &, size_t recursion );
+        void on_wakeup( L &, size_t recursion );
 };
 
@@ -119 +300 @@
         bool wait( condition_variable(L) & this, L & l, uintptr_t info, Time time );
 }
-
-//-----------------------------------------------------------------------------
-// Semaphore
-struct semaphore {
-        __spinlock_t lock;
-        int count;
-        __queue_t($thread) waiting;
-};
-
-void  ?{}(semaphore & this, int count = 1);
-void ^?{}(semaphore & this);
-bool   P (semaphore & this);
-bool   V (semaphore & this);
-bool   V (semaphore & this, unsigned count);

libcfa/src/concurrency/monitor.hfa

@@ -61 +61 @@
 static inline forall( T & | sized(T) | { void ^?{}( T & mutex ); } )
 void delete( T * th ) {
-        ^(*th){};
+        if(th) ^(*th){};
         free( th );
 }

libcfa/src/concurrency/preemption.cfa

@@ -15 +15 @@
 
 #define __cforall_thread__
+// #define __CFA_DEBUG_PRINT_PREEMPTION__
 
 #include "preemption.hfa"
@@ -28 +29 @@
 #include "kernel_private.hfa"
 
+
 #if !defined(__CFA_DEFAULT_PREEMPTION__)
 #define __CFA_DEFAULT_PREEMPTION__ 10`ms
 #endif
 
-Duration default_preemption() __attribute__((weak)) {
-        return __CFA_DEFAULT_PREEMPTION__;
+__attribute__((weak)) Duration default_preemption() {
+        const char * preempt_rate_s = getenv("CFA_DEFAULT_PREEMPTION");
+        if(!preempt_rate_s) {
+                __cfadbg_print_safe(preemption, "No CFA_DEFAULT_PREEMPTION in ENV\n");
+                return __CFA_DEFAULT_PREEMPTION__;
+        }
+
+        char * endptr = 0p;
+        long int preempt_rate_l = strtol(preempt_rate_s, &endptr, 10);
+        if(preempt_rate_l < 0 || preempt_rate_l > 65535) {
+                __cfadbg_print_safe(preemption, "CFA_DEFAULT_PREEMPTION out of range : %ld\n", preempt_rate_l);
+                return __CFA_DEFAULT_PREEMPTION__;
+        }
+        if('\0' != *endptr) {
+                __cfadbg_print_safe(preemption, "CFA_DEFAULT_PREEMPTION not a decimal number : %s\n", preempt_rate_s);
+                return __CFA_DEFAULT_PREEMPTION__;
+        }
+
+        return preempt_rate_l`ms;
 }
 
@@ -98 +117 @@
         //Loop throught every thing expired
         while( node = get_expired( alarms, currtime ) ) {
-                // __cfaabi_dbg_print_buffer_decl( " KERNEL: preemption tick.\n" );
+                __cfadbg_print_buffer_decl( preemption, " KERNEL: preemption tick %lu\n", currtime.tn);
                 Duration period = node->period;
                 if( period == 0) {
@@ -104 +123 @@
                 }
 
+                __cfadbg_print_buffer_local( preemption, " KERNEL: alarm ticking node %p.\n", node );
+
+
                 // Check if this is a kernel
                 if( node->type == Kernel ) {
@@ -109 +131 @@
                 }
                 else if( node->type == User ) {
+                        __cfadbg_print_buffer_local( preemption, " KERNEL: alarm unparking %p.\n", node->thrd );
                         timeout( node->thrd );
                 }
@@ -117 +140 @@
                 // Check if this is a periodic alarm
                 if( period > 0 ) {
-                        // __cfaabi_dbg_print_buffer_local( " KERNEL: alarm period is %lu.\n", period.tv );
+                        __cfadbg_print_buffer_local( preemption, " KERNEL: alarm period is %lu.\n", period`ns );
                         node->alarm = currtime + period;    // Alarm is periodic, add currtime to it (used cached current time)
                         insert( alarms, node );             // Reinsert the node for the next time it triggers
@@ -125 +148 @@
         // If there are still alarms pending, reset the timer
         if( & (*alarms)`first ) {
-                __cfadbg_print_buffer_decl(preemption, " KERNEL: @%ju(%ju) resetting alarm to %ju.\n", currtime.tv, __kernel_get_time().tv, (alarms->head->alarm - currtime).tv);
                 Duration delta = (*alarms)`first.alarm - currtime;
                 Duration capped = max(delta, 50`us);
-                // itimerval tim  = { caped };
-                // __cfaabi_dbg_print_buffer_local( "    Values are %lu, %lu, %lu %lu.\n", delta.tv, caped.tv, tim.it_value.tv_sec, tim.it_value.tv_usec);
-
                 __kernel_set_timer( capped );
         }
@@ -296 +315 @@
         // Enable interrupts by decrementing the counter
         // If counter reaches 0, execute any pending __cfactx_switch
-        void enable_interrupts( __cfaabi_dbg_ctx_param ) {
+        void enable_interrupts( bool poll ) {
                 // Cache the processor now since interrupts can start happening after the atomic store
                 processor   * proc = __cfaabi_tls.this_processor;
-                /* paranoid */ verify( proc );
+                /* paranoid */ verify( !poll || proc );
 
                 with( __cfaabi_tls.preemption_state ){
@@ -321 +340 @@
                                 // Signal the compiler that a fence is needed but only for signal handlers
                                 __atomic_signal_fence(__ATOMIC_RELEASE);
-                                if( proc->pending_preemption ) {
+                                if( poll && proc->pending_preemption ) {
                                         proc->pending_preemption = false;
                                         force_yield( __POLL_PREEMPTION );
                                 }
                         }
-                }
-
-                // For debugging purposes : keep track of the last person to enable the interrupts
-                __cfaabi_dbg_debug_do( proc->last_enable = caller; )
-        }
-
-        // Disable interrupts by incrementint the counter
-        // Don't execute any pending __cfactx_switch even if counter reaches 0
-        void enable_interrupts_noPoll() {
-                unsigned short prev = __cfaabi_tls.preemption_state.disable_count;
-                __cfaabi_tls.preemption_state.disable_count -= 1;
-                // If this triggers someone is enabled already enabled interrupts
-                /* paranoid */ verifyf( prev != 0u, "Incremented from %u\n", prev );
-                if( prev == 1 ) {
-                        #if GCC_VERSION > 50000
-                                static_assert(__atomic_always_lock_free(sizeof(__cfaabi_tls.preemption_state.enabled), &__cfaabi_tls.preemption_state.enabled), "Must be lock-free");
-                        #endif
-                        // Set enabled flag to true
-                        // should be atomic to avoid preemption in the middle of the operation.
-                        // use memory order RELAXED since there is no inter-thread on this variable requirements
-                        __atomic_store_n(&__cfaabi_tls.preemption_state.enabled, true, __ATOMIC_RELAXED);
-
-                        // Signal the compiler that a fence is needed but only for signal handlers
-                        __atomic_signal_fence(__ATOMIC_RELEASE);
                 }
         }
@@ -585 +580 @@
 
         // Setup proper signal handlers
-        __cfaabi_sigaction( SIGUSR1, sigHandler_ctxSwitch, SA_SIGINFO | SA_RESTART ); // __cfactx_switch handler
-        __cfaabi_sigaction( SIGALRM, sigHandler_alarm    , SA_SIGINFO | SA_RESTART ); // debug handler
+        __cfaabi_sigaction( SIGUSR1, sigHandler_ctxSwitch, SA_SIGINFO ); // __cfactx_switch handler
+        __cfaabi_sigaction( SIGALRM, sigHandler_alarm    , SA_SIGINFO ); // debug handler
 
         signal_block( SIGALRM );
@@ -689 +684 @@
 }
 
-#if !defined(__CFA_NO_STATISTICS__)
-        int __print_alarm_stats = 0;
-#endif
-
 // Main of the alarm thread
 // Waits on SIGALRM and send SIGUSR1 to whom ever needs it
 static void * alarm_loop( __attribute__((unused)) void * args ) {
         __processor_id_t id;
-        id.full_proc = false;
-        id.id = doregister(&id);
+        register_proc_id(&id);
         __cfaabi_tls.this_proc_id = &id;
 
-        #if !defined(__CFA_NO_STATISTICS__)
-                struct __stats_t local_stats;
-                __cfaabi_tls.this_stats = &local_stats;
-                __init_stats( &local_stats );
-        #endif
 
         // Block sigalrms to control when they arrive
@@ -764 +749 @@
 EXIT:
         __cfaabi_dbg_print_safe( "Kernel : Preemption thread stopping\n" );
-        unregister(&id);
-
-        #if !defined(__CFA_NO_STATISTICS__)
-                if( 0 != __print_alarm_stats ) {
-                        __print_stats( &local_stats, __print_alarm_stats, "Alarm", "Thread", 0p );
-                }
-        #endif
+        register_proc_id(&id);
+
         return 0p;
 }

libcfa/src/concurrency/ready_queue.cfa

@@ -17 +17 @@
 // #define __CFA_DEBUG_PRINT_READY_QUEUE__
 
-// #define USE_SNZI
+// #define USE_MPSC
+
+#define USE_RELAXED_FIFO
+// #define USE_WORK_STEALING
 
 #include "bits/defs.hfa"
@@ -28 +31 @@
 #include <unistd.h>
 
-#include "snzi.hfa"
 #include "ready_subqueue.hfa"
 
 static const size_t cache_line_size = 64;
+
+#if !defined(__CFA_NO_STATISTICS__)
+        #define __STATS(...) __VA_ARGS__
+#else
+        #define __STATS(...)
+#endif
 
 // No overriden function, no environment variable, no define
@@ -39 +47 @@
 #endif
 
-#define BIAS 16
+#if   defined(USE_RELAXED_FIFO)
+        #define BIAS 4
+        #define READYQ_SHARD_FACTOR 4
+        #define SEQUENTIAL_SHARD 1
+#elif defined(USE_WORK_STEALING)
+        #define READYQ_SHARD_FACTOR 2
+        #define SEQUENTIAL_SHARD 2
+#else
+        #error no scheduling strategy selected
+#endif
+
+static inline struct $thread * try_pop(struct cluster * cltr, unsigned w __STATS(, __stats_readyQ_pop_t & stats));
+static inline struct $thread * try_pop(struct cluster * cltr, unsigned i, unsigned j __STATS(, __stats_readyQ_pop_t & stats));
+static inline struct $thread * search(struct cluster * cltr);
+static inline [unsigned, bool] idx_from_r(unsigned r, unsigned preferred);
+
 
 // returns the maximum number of processors the RWLock support
@@ -93 +116 @@
 //=======================================================================
 // Lock-Free registering/unregistering of threads
-unsigned doregister( struct __processor_id_t * proc ) with(*__scheduler_lock) {
+void register_proc_id( struct __processor_id_t * proc ) with(*__scheduler_lock) {
         __cfadbg_print_safe(ready_queue, "Kernel : Registering proc %p for RW-Lock\n", proc);
 
@@ -107 +130 @@
                         /*paranoid*/ verify(0 == (__alignof__(data[i]) % cache_line_size));
                         /*paranoid*/ verify((((uintptr_t)&data[i]) % cache_line_size) == 0);
-                        return i;
+                        proc->id = i;
                 }
         }
@@ -134 +157 @@
         /*paranoid*/ verify(__alignof__(data[n]) == (2 * cache_line_size));
         /*paranoid*/ verify((((uintptr_t)&data[n]) % cache_line_size) == 0);
-        return n;
-}
-
-void unregister( struct __processor_id_t * proc ) with(*__scheduler_lock) {
+        proc->id = n;
+}
+
+void unregister_proc_id( struct __processor_id_t * proc ) with(*__scheduler_lock) {
         unsigned id = proc->id;
         /*paranoid*/ verify(id < ready);
@@ -192 +215 @@
 
 //=======================================================================
-// Cforall Reqdy Queue used for scheduling
+// Cforall Ready Queue used for scheduling
 //=======================================================================
 void ?{}(__ready_queue_t & this) with (this) {
         lanes.data  = 0p;
+        lanes.tscs  = 0p;
         lanes.count = 0;
 }
 
 void ^?{}(__ready_queue_t & this) with (this) {
-        verify( 1 == lanes.count );
-        #ifdef USE_SNZI
-                verify( !query( snzi ) );
-        #endif
+        verify( SEQUENTIAL_SHARD == lanes.count );
         free(lanes.data);
+        free(lanes.tscs);
 }
 
 //-----------------------------------------------------------------------
-__attribute__((hot)) bool query(struct cluster * cltr) {
-        #ifdef USE_SNZI
-                return query(cltr->ready_queue.snzi);
-        #endif
-        return true;
-}
-
-static inline [unsigned, bool] idx_from_r(unsigned r, unsigned preferred) {
-        unsigned i;
-        bool local;
-        #if defined(BIAS)
+#if defined(USE_RELAXED_FIFO)
+        //-----------------------------------------------------------------------
+        // get index from random number with or without bias towards queues
+        static inline [unsigned, bool] idx_from_r(unsigned r, unsigned preferred) {
+                unsigned i;
+                bool local;
                 unsigned rlow  = r % BIAS;
                 unsigned rhigh = r / BIAS;
@@ -224 +241 @@
                         // (BIAS - 1) out of BIAS chances
                         // Use perferred queues
-                        i = preferred + (rhigh % 4);
+                        i = preferred + (rhigh % READYQ_SHARD_FACTOR);
                         local = true;
                 }
@@ -233 +250 @@
                         local = false;
                 }
-        #else
-                i = r;
-                local = false;
+                return [i, local];
+        }
+
+        __attribute__((hot)) void push(struct cluster * cltr, struct $thread * thrd) with (cltr->ready_queue) {
+                __cfadbg_print_safe(ready_queue, "Kernel : Pushing %p on cluster %p\n", thrd, cltr);
+
+                const bool external = (!kernelTLS().this_processor) || (cltr != kernelTLS().this_processor->cltr);
+                /* paranoid */ verify(external || kernelTLS().this_processor->rdq.id < lanes.count );
+
+                // write timestamp
+                thrd->link.ts = rdtscl();
+
+                bool local;
+                int preferred = external ? -1 : kernelTLS().this_processor->rdq.id;
+
+                // Try to pick a lane and lock it
+                unsigned i;
+                do {
+                        // Pick the index of a lane
+                        unsigned r = __tls_rand_fwd();
+                        [i, local] = idx_from_r(r, preferred);
+
+                        i %= __atomic_load_n( &lanes.count, __ATOMIC_RELAXED );
+
+                        #if !defined(__CFA_NO_STATISTICS__)
+                                if(unlikely(external)) __atomic_fetch_add(&cltr->stats->ready.push.extrn.attempt, 1, __ATOMIC_RELAXED);
+                                else if(local) __tls_stats()->ready.push.local.attempt++;
+                                else __tls_stats()->ready.push.share.attempt++;
+                        #endif
+
+                #if defined(USE_MPSC)
+                        // mpsc always succeeds
+                } while( false );
+                #else
+                        // If we can't lock it retry
+                } while( !__atomic_try_acquire( &lanes.data[i].lock ) );
+                #endif
+
+                // Actually push it
+                push(lanes.data[i], thrd);
+
+                #if !defined(USE_MPSC)
+                        // Unlock and return
+                        __atomic_unlock( &lanes.data[i].lock );
+                #endif
+
+                // Mark the current index in the tls rng instance as having an item
+                __tls_rand_advance_bck();
+
+                __cfadbg_print_safe(ready_queue, "Kernel : Pushed %p on cluster %p (idx: %u, mask %llu, first %d)\n", thrd, cltr, i, used.mask[0], lane_first);
+
+                // Update statistics
+                #if !defined(__CFA_NO_STATISTICS__)
+                        if(unlikely(external)) __atomic_fetch_add(&cltr->stats->ready.push.extrn.success, 1, __ATOMIC_RELAXED);
+                        else if(local) __tls_stats()->ready.push.local.success++;
+                        else __tls_stats()->ready.push.share.success++;
+                #endif
+        }
+
+        // Pop from the ready queue from a given cluster
+        __attribute__((hot)) $thread * pop_fast(struct cluster * cltr) with (cltr->ready_queue) {
+                /* paranoid */ verify( lanes.count > 0 );
+                /* paranoid */ verify( kernelTLS().this_processor );
+                /* paranoid */ verify( kernelTLS().this_processor->rdq.id < lanes.count );
+
+                unsigned count = __atomic_load_n( &lanes.count, __ATOMIC_RELAXED );
+                int preferred = kernelTLS().this_processor->rdq.id;
+
+
+                // As long as the list is not empty, try finding a lane that isn't empty and pop from it
+                for(25) {
+                        // Pick two lists at random
+                        unsigned ri = __tls_rand_bck();
+                        unsigned rj = __tls_rand_bck();
+
+                        unsigned i, j;
+                        __attribute__((unused)) bool locali, localj;
+                        [i, locali] = idx_from_r(ri, preferred);
+                        [j, localj] = idx_from_r(rj, preferred);
+
+                        i %= count;
+                        j %= count;
+
+                        // try popping from the 2 picked lists
+                        struct $thread * thrd = try_pop(cltr, i, j __STATS(, *(locali || localj ? &__tls_stats()->ready.pop.local : &__tls_stats()->ready.pop.help)));
+                        if(thrd) {
+                                return thrd;
+                        }
+                }
+
+                // All lanes where empty return 0p
+                return 0p;
+        }
+
+        __attribute__((hot)) struct $thread * pop_slow(struct cluster * cltr) { return pop_fast(cltr); }
+        __attribute__((hot)) struct $thread * pop_search(struct cluster * cltr) {
+                return search(cltr);
+        }
+#endif
+#if defined(USE_WORK_STEALING)
+        __attribute__((hot)) void push(struct cluster * cltr, struct $thread * thrd) with (cltr->ready_queue) {
+                __cfadbg_print_safe(ready_queue, "Kernel : Pushing %p on cluster %p\n", thrd, cltr);
+
+                const bool external = (!kernelTLS().this_processor) || (cltr != kernelTLS().this_processor->cltr);
+                /* paranoid */ verify(external || kernelTLS().this_processor->rdq.id < lanes.count );
+
+                // write timestamp
+                thrd->link.ts = rdtscl();
+
+                // Try to pick a lane and lock it
+                unsigned i;
+                do {
+                        #if !defined(__CFA_NO_STATISTICS__)
+                                if(unlikely(external)) __atomic_fetch_add(&cltr->stats->ready.push.extrn.attempt, 1, __ATOMIC_RELAXED);
+                                else __tls_stats()->ready.push.local.attempt++;
+                        #endif
+
+                        if(unlikely(external)) {
+                                i = __tls_rand() % lanes.count;
+                        }
+                        else {
+                                processor * proc = kernelTLS().this_processor;
+                                unsigned r = proc->rdq.its++;
+                                i =  proc->rdq.id + (r % READYQ_SHARD_FACTOR);
+                        }
+
+
+                #if defined(USE_MPSC)
+                        // mpsc always succeeds
+                } while( false );
+                #else
+                        // If we can't lock it retry
+                } while( !__atomic_try_acquire( &lanes.data[i].lock ) );
+                #endif
+
+                // Actually push it
+                push(lanes.data[i], thrd);
+
+                #if !defined(USE_MPSC)
+                        // Unlock and return
+                        __atomic_unlock( &lanes.data[i].lock );
+                #endif
+
+                #if !defined(__CFA_NO_STATISTICS__)
+                        if(unlikely(external)) __atomic_fetch_add(&cltr->stats->ready.push.extrn.success, 1, __ATOMIC_RELAXED);
+                        else __tls_stats()->ready.push.local.success++;
+                #endif
+
+                __cfadbg_print_safe(ready_queue, "Kernel : Pushed %p on cluster %p (idx: %u, mask %llu, first %d)\n", thrd, cltr, i, used.mask[0], lane_first);
+        }
+
+        // Pop from the ready queue from a given cluster
+        __attribute__((hot)) $thread * pop_fast(struct cluster * cltr) with (cltr->ready_queue) {
+                /* paranoid */ verify( lanes.count > 0 );
+                /* paranoid */ verify( kernelTLS().this_processor );
+                /* paranoid */ verify( kernelTLS().this_processor->rdq.id < lanes.count );
+
+                processor * proc = kernelTLS().this_processor;
+
+                if(proc->rdq.target == -1u) {
+                        proc->rdq.target = __tls_rand() % lanes.count;
+                        unsigned it1  = proc->rdq.itr;
+                        unsigned it2  = proc->rdq.itr + 1;
+                        unsigned idx1 = proc->rdq.id + (it1 % READYQ_SHARD_FACTOR);
+                        unsigned idx2 = proc->rdq.id + (it2 % READYQ_SHARD_FACTOR);
+                        unsigned long long tsc1 = ts(lanes.data[idx1]);
+                        unsigned long long tsc2 = ts(lanes.data[idx2]);
+                        proc->rdq.cutoff = min(tsc1, tsc2);
+                        if(proc->rdq.cutoff == 0) proc->rdq.cutoff = -1ull;
+                }
+                else {
+                        unsigned target = proc->rdq.target;
+                        proc->rdq.target = -1u;
+                        if(lanes.tscs[target].tv < proc->rdq.cutoff) {
+                                $thread * t = try_pop(cltr, target __STATS(, __tls_stats()->ready.pop.help));
+                                if(t) return t;
+                        }
+                }
+
+                for(READYQ_SHARD_FACTOR) {
+                        unsigned i = proc->rdq.id + (--proc->rdq.itr % READYQ_SHARD_FACTOR);
+                        if($thread * t = try_pop(cltr, i __STATS(, __tls_stats()->ready.pop.local))) return t;
+                }
+                return 0p;
+        }
+
+        __attribute__((hot)) struct $thread * pop_slow(struct cluster * cltr) with (cltr->ready_queue) {
+                unsigned i = __tls_rand() % lanes.count;
+                return try_pop(cltr, i __STATS(, __tls_stats()->ready.pop.steal));
+        }
+
+        __attribute__((hot)) struct $thread * pop_search(struct cluster * cltr) with (cltr->ready_queue) {
+                return search(cltr);
+        }
+#endif
+
+//=======================================================================
+// Various Ready Queue utilities
+//=======================================================================
+// these function work the same or almost the same
+// whether they are using work-stealing or relaxed fifo scheduling
+
+//-----------------------------------------------------------------------
+// try to pop from a lane given by index w
+static inline struct $thread * try_pop(struct cluster * cltr, unsigned w __STATS(, __stats_readyQ_pop_t & stats)) with (cltr->ready_queue) {
+        __STATS( stats.attempt++; )
+
+        // Get relevant elements locally
+        __intrusive_lane_t & lane = lanes.data[w];
+
+        // If list looks empty retry
+        if( is_empty(lane) ) {
+                __STATS( stats.espec++; )
+                return 0p;
+        }
+
+        // If we can't get the lock retry
+        if( !__atomic_try_acquire(&lane.lock) ) {
+                __STATS( stats.elock++; )
+                return 0p;
+        }
+
+        // If list is empty, unlock and retry
+        if( is_empty(lane) ) {
+                __atomic_unlock(&lane.lock);
+                __STATS( stats.eempty++; )
+                return 0p;
+        }
+
+        // Actually pop the list
+        struct $thread * thrd;
+        thrd = pop(lane);
+
+        /* paranoid */ verify(thrd);
+        /* paranoid */ verify(lane.lock);
+
+        // Unlock and return
+        __atomic_unlock(&lane.lock);
+
+        // Update statistics
+        __STATS( stats.success++; )
+
+        #if defined(USE_WORK_STEALING)
+                lanes.tscs[w].tv = thrd->link.ts;
         #endif
-        return [i, local];
+
+        // return the popped thread
+        return thrd;
 }
 
 //-----------------------------------------------------------------------
-__attribute__((hot)) bool push(struct cluster * cltr, struct $thread * thrd) with (cltr->ready_queue) {
-        __cfadbg_print_safe(ready_queue, "Kernel : Pushing %p on cluster %p\n", thrd, cltr);
-
-        // write timestamp
-        thrd->link.ts = rdtscl();
-
-        __attribute__((unused)) bool local;
-        __attribute__((unused)) int preferred;
-        #if defined(BIAS)
-                preferred =
-                        //*
-                        kernelTLS().this_processor ? kernelTLS().this_processor->id * 4 : -1;
-                        /*/
-                        thrd->link.preferred * 4;
-                        //*/
-        #endif
-
-        // Try to pick a lane and lock it
-        unsigned i;
-        do {
-                // Pick the index of a lane
-                // unsigned r = __tls_rand();
-                unsigned r = __tls_rand_fwd();
-                [i, local] = idx_from_r(r, preferred);
-
-                #if !defined(__CFA_NO_STATISTICS__)
-                        if(local) {
-                                __tls_stats()->ready.pick.push.local++;
-                        }
-                #endif
-
-                i %= __atomic_load_n( &lanes.count, __ATOMIC_RELAXED );
-
-                #if !defined(__CFA_NO_STATISTICS__)
-                        __tls_stats()->ready.pick.push.attempt++;
-                #endif
-
-                // If we can't lock it retry
-        } while( !__atomic_try_acquire( &lanes.data[i].lock ) );
-
-        bool first = false;
-
-        // Actually push it
-        #ifdef USE_SNZI
-                bool lane_first =
-        #endif
-
-        push(lanes.data[i], thrd);
-
-        #ifdef USE_SNZI
-                // If this lane used to be empty we need to do more
-                if(lane_first) {
-                        // Check if the entire queue used to be empty
-                        first = !query(snzi);
-
-                        // Update the snzi
-                        arrive( snzi, i );
-                }
-        #endif
-
-        __tls_rand_advance_bck();
-
-        // Unlock and return
-        __atomic_unlock( &lanes.data[i].lock );
-
-        __cfadbg_print_safe(ready_queue, "Kernel : Pushed %p on cluster %p (idx: %u, mask %llu, first %d)\n", thrd, cltr, i, used.mask[0], lane_first);
-
-        // Update statistics
-        #if !defined(__CFA_NO_STATISTICS__)
-                #if defined(BIAS)
-                        if( local ) __tls_stats()->ready.pick.push.lsuccess++;
-                #endif
-                __tls_stats()->ready.pick.push.success++;
-        #endif
-
-        // return whether or not the list was empty before this push
-        return first;
-}
-
-static struct $thread * try_pop(struct cluster * cltr, unsigned i, unsigned j);
-static struct $thread * try_pop(struct cluster * cltr, unsigned i);
-
-// Pop from the ready queue from a given cluster
-__attribute__((hot)) $thread * pop(struct cluster * cltr) with (cltr->ready_queue) {
-        /* paranoid */ verify( lanes.count > 0 );
-        unsigned count = __atomic_load_n( &lanes.count, __ATOMIC_RELAXED );
-        int preferred;
-        #if defined(BIAS)
-                // Don't bother trying locally too much
-                int local_tries = 8;
-                preferred = kernelTLS().this_processor->id * 4;
-        #endif
-
-
-        // As long as the list is not empty, try finding a lane that isn't empty and pop from it
-        #ifdef USE_SNZI
-                while( query(snzi) ) {
-        #else
-                for(25) {
-        #endif
-                // Pick two lists at random
-                // unsigned ri = __tls_rand();
-                // unsigned rj = __tls_rand();
-                unsigned ri = __tls_rand_bck();
-                unsigned rj = __tls_rand_bck();
-
-                unsigned i, j;
-                __attribute__((unused)) bool locali, localj;
-                [i, locali] = idx_from_r(ri, preferred);
-                [j, localj] = idx_from_r(rj, preferred);
-
-                #if !defined(__CFA_NO_STATISTICS__)
-                        if(locali) {
-                                __tls_stats()->ready.pick.pop.local++;
-                        }
-                        if(localj) {
-                                __tls_stats()->ready.pick.pop.local++;
-                        }
-                #endif
-
-                i %= count;
-                j %= count;
-
-                // try popping from the 2 picked lists
-                struct $thread * thrd = try_pop(cltr, i, j);
-                if(thrd) {
-                        #if defined(BIAS) && !defined(__CFA_NO_STATISTICS__)
-                                if( locali || localj ) __tls_stats()->ready.pick.pop.lsuccess++;
-                        #endif
-                        return thrd;
-                }
-        }
-
-        // All lanes where empty return 0p
-        return 0p;
-}
-
-__attribute__((hot)) struct $thread * pop_slow(struct cluster * cltr) with (cltr->ready_queue) {
+// try to pop from any lanes making sure you don't miss any threads push
+// before the start of the function
+static inline struct $thread * search(struct cluster * cltr) with (cltr->ready_queue) {
         /* paranoid */ verify( lanes.count > 0 );
         unsigned count = __atomic_load_n( &lanes.count, __ATOMIC_RELAXED );
@@ -384 +509 @@
         for(i; count) {
                 unsigned idx = (offset + i) % count;
-                struct $thread * thrd = try_pop(cltr, idx);
+                struct $thread * thrd = try_pop(cltr, idx __STATS(, __tls_stats()->ready.pop.search));
                 if(thrd) {
                         return thrd;
@@ -394 +519 @@
 }
 
-
 //-----------------------------------------------------------------------
-// Given 2 indexes, pick the list with the oldest push an try to pop from it
-static inline struct $thread * try_pop(struct cluster * cltr, unsigned i, unsigned j) with (cltr->ready_queue) {
-        #if !defined(__CFA_NO_STATISTICS__)
-                __tls_stats()->ready.pick.pop.attempt++;
-        #endif
-
-        // Pick the bet list
-        int w = i;
-        if( __builtin_expect(!is_empty(lanes.data[j]), true) ) {
-                w = (ts(lanes.data[i]) < ts(lanes.data[j])) ? i : j;
-        }
-
-        return try_pop(cltr, w);
-}
-
-static inline struct $thread * try_pop(struct cluster * cltr, unsigned w) with (cltr->ready_queue) {
-        // Get relevant elements locally
-        __intrusive_lane_t & lane = lanes.data[w];
-
-        // If list looks empty retry
-        if( is_empty(lane) ) return 0p;
-
-        // If we can't get the lock retry
-        if( !__atomic_try_acquire(&lane.lock) ) return 0p;
-
-
-        // If list is empty, unlock and retry
-        if( is_empty(lane) ) {
-                __atomic_unlock(&lane.lock);
-                return 0p;
-        }
-
-        // Actually pop the list
-        struct $thread * thrd;
-        thrd = pop(lane);
-
-        /* paranoid */ verify(thrd);
-        /* paranoid */ verify(lane.lock);
-
-        #ifdef USE_SNZI
-                // If this was the last element in the lane
-                if(emptied) {
-                        depart( snzi, w );
-                }
-        #endif
-
-        // Unlock and return
-        __atomic_unlock(&lane.lock);
-
-        // Update statistics
-        #if !defined(__CFA_NO_STATISTICS__)
-                __tls_stats()->ready.pick.pop.success++;
-        #endif
-
-        // Update the thread bias
-        thrd->link.preferred = w / 4;
-
-        // return the popped thread
-        return thrd;
-}
-//-----------------------------------------------------------------------
-
-bool remove_head(struct cluster * cltr, struct $thread * thrd) with (cltr->ready_queue) {
-        for(i; lanes.count) {
-                __intrusive_lane_t & lane = lanes.data[i];
-
-                bool removed = false;
-
-                __atomic_acquire(&lane.lock);
-                        if(head(lane)->link.next == thrd) {
-                                $thread * pthrd;
-                                pthrd = pop(lane);
-
-                                /* paranoid */ verify( pthrd == thrd );
-
-                                removed = true;
-                                #ifdef USE_SNZI
-                                        if(emptied) {
-                                                depart( snzi, i );
-                                        }
-                                #endif
-                        }
-                __atomic_unlock(&lane.lock);
-
-                if( removed ) return true;
-        }
-        return false;
-}
-
-//-----------------------------------------------------------------------
-
+// Check that all the intrusive queues in the data structure are still consistent
 static void check( __ready_queue_t & q ) with (q) {
-        #if defined(__CFA_WITH_VERIFY__)
+        #if defined(__CFA_WITH_VERIFY__) && !defined(USE_MPSC)
                 {
                         for( idx ; lanes.count ) {
@@ -499 +533 @@
                                 assert(tail(sl)->link.prev->link.next == tail(sl) );
 
-                                if(sl.before.link.ts == 0l) {
+                                if(is_empty(sl)) {
                                         assert(tail(sl)->link.prev == head(sl));
                                         assert(head(sl)->link.next == tail(sl));
@@ -511 +545 @@
 }
 
+//-----------------------------------------------------------------------
+// Given 2 indexes, pick the list with the oldest push an try to pop from it
+static inline struct $thread * try_pop(struct cluster * cltr, unsigned i, unsigned j __STATS(, __stats_readyQ_pop_t & stats)) with (cltr->ready_queue) {
+        // Pick the bet list
+        int w = i;
+        if( __builtin_expect(!is_empty(lanes.data[j]), true) ) {
+                w = (ts(lanes.data[i]) < ts(lanes.data[j])) ? i : j;
+        }
+
+        return try_pop(cltr, w __STATS(, stats));
+}
+
 // Call this function of the intrusive list was moved using memcpy
 // fixes the list so that the pointers back to anchors aren't left dangling
 static inline void fix(__intrusive_lane_t & ll) {
-        // if the list is not empty then follow he pointer and fix its reverse
-        if(!is_empty(ll)) {
-                head(ll)->link.next->link.prev = head(ll);
-                tail(ll)->link.prev->link.next = tail(ll);
-        }
-        // Otherwise just reset the list
-        else {
-                verify(tail(ll)->link.next == 0p);
-                tail(ll)->link.prev = head(ll);
-                head(ll)->link.next = tail(ll);
-                verify(head(ll)->link.prev == 0p);
-        }
+        #if !defined(USE_MPSC)
+                // if the list is not empty then follow he pointer and fix its reverse
+                if(!is_empty(ll)) {
+                        head(ll)->link.next->link.prev = head(ll);
+                        tail(ll)->link.prev->link.next = tail(ll);
+                }
+                // Otherwise just reset the list
+                else {
+                        verify(tail(ll)->link.next == 0p);
+                        tail(ll)->link.prev = head(ll);
+                        head(ll)->link.next = tail(ll);
+                        verify(head(ll)->link.prev == 0p);
+                }
+        #endif
+}
+
+static void assign_list(unsigned & value, dlist(processor, processor) & list, unsigned count) {
+        processor * it = &list`first;
+        for(unsigned i = 0; i < count; i++) {
+                /* paranoid */ verifyf( it, "Unexpected null iterator, at index %u of %u\n", i, count);
+                it->rdq.id = value;
+                it->rdq.target = -1u;
+                value += READYQ_SHARD_FACTOR;
+                it = &(*it)`next;
+        }
+}
+
+static void reassign_cltr_id(struct cluster * cltr) {
+        unsigned preferred = 0;
+        assign_list(preferred, cltr->procs.actives, cltr->procs.total - cltr->procs.idle);
+        assign_list(preferred, cltr->procs.idles  , cltr->procs.idle );
+}
+
+static void fix_times( struct cluster * cltr ) with( cltr->ready_queue ) {
+        #if defined(USE_WORK_STEALING)
+                lanes.tscs = alloc(lanes.count, lanes.tscs`realloc);
+                for(i; lanes.count) {
+                        lanes.tscs[i].tv = ts(lanes.data[i]);
+                }
+        #endif
 }
 
 // Grow the ready queue
-void ready_queue_grow  (struct cluster * cltr, int target) {
+void ready_queue_grow(struct cluster * cltr) {
+        size_t ncount;
+        int target = cltr->procs.total;
+
         /* paranoid */ verify( ready_mutate_islocked() );
         __cfadbg_print_safe(ready_queue, "Kernel : Growing ready queue\n");
@@ -538 +615 @@
         // grow the ready queue
         with( cltr->ready_queue ) {
-                #ifdef USE_SNZI
-                        ^(snzi){};
-                #endif
-
                 // Find new count
                 // Make sure we always have atleast 1 list
-                size_t ncount = target >= 2 ? target * 4: 1;
+                if(target >= 2) {
+                        ncount = target * READYQ_SHARD_FACTOR;
+                } else {
+                        ncount = SEQUENTIAL_SHARD;
+                }
 
                 // Allocate new array (uses realloc and memcpies the data)
@@ -561 +638 @@
                 // Update original
                 lanes.count = ncount;
-
-                #ifdef USE_SNZI
-                        // Re-create the snzi
-                        snzi{ log2( lanes.count / 8 ) };
-                        for( idx; (size_t)lanes.count ) {
-                                if( !is_empty(lanes.data[idx]) ) {
-                                        arrive(snzi, idx);
-                                }
-                        }
-                #endif
-        }
+        }
+
+        fix_times(cltr);
+
+        reassign_cltr_id(cltr);
 
         // Make sure that everything is consistent
@@ -582 +653 @@
 
 // Shrink the ready queue
-void ready_queue_shrink(struct cluster * cltr, int target) {
+void ready_queue_shrink(struct cluster * cltr) {
         /* paranoid */ verify( ready_mutate_islocked() );
         __cfadbg_print_safe(ready_queue, "Kernel : Shrinking ready queue\n");
@@ -589 +660 @@
         /* paranoid */ check( cltr->ready_queue );
 
+        int target = cltr->procs.total;
+
         with( cltr->ready_queue ) {
-                #ifdef USE_SNZI
-                        ^(snzi){};
-                #endif
-
                 // Remember old count
                 size_t ocount = lanes.count;
@@ -599 +668 @@
                 // Find new count
                 // Make sure we always have atleast 1 list
-                lanes.count = target >= 2 ? target * 4: 1;
+                lanes.count = target >= 2 ? target * READYQ_SHARD_FACTOR: SEQUENTIAL_SHARD;
                 /* paranoid */ verify( ocount >= lanes.count );
-                /* paranoid */ verify( lanes.count == target * 4 || target < 2 );
+                /* paranoid */ verify( lanes.count == target * READYQ_SHARD_FACTOR || target < 2 );
 
                 // for printing count the number of displaced threads
@@ -644 +713 @@
                         fix(lanes.data[idx]);
                 }
-
-                #ifdef USE_SNZI
-                        // Re-create the snzi
-                        snzi{ log2( lanes.count / 8 ) };
-                        for( idx; (size_t)lanes.count ) {
-                                if( !is_empty(lanes.data[idx]) ) {
-                                        arrive(snzi, idx);
-                                }
-                        }
-                #endif
-        }
+        }
+
+        fix_times(cltr);
+
+        reassign_cltr_id(cltr);
 
         // Make sure that everything is consistent

libcfa/src/concurrency/ready_subqueue.hfa

@@ -2 +2 @@
 
 #define __CFA_NO_SCHED_STATS__
+
+#include "containers/queueLockFree.hfa"
 
 // Intrusives lanes which are used by the relaxed ready queue
 struct __attribute__((aligned(128))) __intrusive_lane_t {
 
-        // anchor for the head and the tail of the queue
-        __attribute__((aligned(128))) struct __sentinel_t {
-                // Link lists fields
-                // instrusive link field for threads
-                // must be exactly as in $thread
-                __thread_desc_link link;
-        } before, after;
+        #if defined(USE_MPSC)
+                mpsc_queue($thread) queue;
+                __attribute__((aligned(128)))
+        #else
+                // anchor for the head and the tail of the queue
+                __attribute__((aligned(128))) struct __sentinel_t {
+                        // Link lists fields
+                        // instrusive link field for threads
+                        // must be exactly as in $thread
+                        __thread_desc_link link;
+                } before, after;
+        #endif
 
         // spin lock protecting the queue
@@ -35 +42 @@
 // Get the head pointer (one before the first element) from the anchor
 static inline $thread * head(const __intrusive_lane_t & this) {
-        $thread * rhead = ($thread *)(
-                (uintptr_t)( &this.before ) - offsetof( $thread, link )
-        );
-        /* paranoid */ verify(rhead);
-        return rhead;
+        #if defined(USE_MPSC)
+                return this.queue.head;
+        #else
+                $thread * rhead = ($thread *)(
+                        (uintptr_t)( &this.before ) - offsetof( $thread, link )
+                );
+                /* paranoid */ verify(rhead);
+                return rhead;
+        #endif
 }
 
 // Get the tail pointer (one after the last element) from the anchor
 static inline $thread * tail(const __intrusive_lane_t & this) {
-        $thread * rtail = ($thread *)(
-                (uintptr_t)( &this.after ) - offsetof( $thread, link )
-        );
-        /* paranoid */ verify(rtail);
-        return rtail;
+        #if defined(USE_MPSC)
+                return this.queue.tail;
+        #else
+                $thread * rtail = ($thread *)(
+                        (uintptr_t)( &this.after ) - offsetof( $thread, link )
+                );
+                /* paranoid */ verify(rtail);
+                return rtail;
+        #endif
 }
 
@@ -55 +70 @@
         this.lock = false;
 
-        this.before.link.prev = 0p;
-        this.before.link.next = tail(this);
-        this.before.link.ts   = 0;
-
-        this.after .link.prev = head(this);
-        this.after .link.next = 0p;
-        this.after .link.ts   = 0;
-
-        #if !defined(__CFA_NO_SCHED_STATS__)
-                this.stat.diff = 0;
-                this.stat.push = 0;
-                this.stat.pop  = 0;
-        #endif
-
-        // We add a boat-load of assertions here because the anchor code is very fragile
-        /* paranoid */ verify(((uintptr_t)( head(this) ) + offsetof( $thread, link )) == (uintptr_t)(&this.before));
-        /* paranoid */ verify(((uintptr_t)( tail(this) ) + offsetof( $thread, link )) == (uintptr_t)(&this.after ));
-        /* paranoid */ verify(head(this)->link.prev == 0p );
-        /* paranoid */ verify(head(this)->link.next == tail(this) );
-        /* paranoid */ verify(tail(this)->link.next == 0p );
-        /* paranoid */ verify(tail(this)->link.prev == head(this) );
-        /* paranoid */ verify(&head(this)->link.prev == &this.before.link.prev );
-        /* paranoid */ verify(&head(this)->link.next == &this.before.link.next );
-        /* paranoid */ verify(&tail(this)->link.prev == &this.after .link.prev );
-        /* paranoid */ verify(&tail(this)->link.next == &this.after .link.next );
-        /* paranoid */ verify(__alignof__(__intrusive_lane_t) == 128);
-        /* paranoid */ verify(__alignof__(this) == 128);
-        /* paranoid */ verifyf(((intptr_t)(&this) % 128) == 0, "Expected address to be aligned %p %% 128 == %zd", &this, ((intptr_t)(&this) % 128));
+        #if !defined(USE_MPSC)
+                this.before.link.prev = 0p;
+                this.before.link.next = tail(this);
+                this.before.link.ts   = 0;
+
+                this.after .link.prev = head(this);
+                this.after .link.next = 0p;
+                this.after .link.ts   = 0;
+
+                #if !defined(__CFA_NO_SCHED_STATS__)
+                        this.stat.diff = 0;
+                        this.stat.push = 0;
+                        this.stat.pop  = 0;
+                #endif
+
+                // We add a boat-load of assertions here because the anchor code is very fragile
+                /* paranoid */ verify(((uintptr_t)( head(this) ) + offsetof( $thread, link )) == (uintptr_t)(&this.before));
+                /* paranoid */ verify(((uintptr_t)( tail(this) ) + offsetof( $thread, link )) == (uintptr_t)(&this.after ));
+                /* paranoid */ verify(head(this)->link.prev == 0p );
+                /* paranoid */ verify(head(this)->link.next == tail(this) );
+                /* paranoid */ verify(tail(this)->link.next == 0p );
+                /* paranoid */ verify(tail(this)->link.prev == head(this) );
+                /* paranoid */ verify(&head(this)->link.prev == &this.before.link.prev );
+                /* paranoid */ verify(&head(this)->link.next == &this.before.link.next );
+                /* paranoid */ verify(&tail(this)->link.prev == &this.after .link.prev );
+                /* paranoid */ verify(&tail(this)->link.next == &this.after .link.next );
+                /* paranoid */ verify(__alignof__(__intrusive_lane_t) == 128);
+                /* paranoid */ verify(__alignof__(this) == 128);
+                /* paranoid */ verifyf(((intptr_t)(&this) % 128) == 0, "Expected address to be aligned %p %% 128 == %zd", &this, ((intptr_t)(&this) % 128));
+        #endif
 }
 
 // Dtor is trivial
 void ^?{}( __intrusive_lane_t & this ) {
-        // Make sure the list is empty
-        /* paranoid */ verify(head(this)->link.prev == 0p );
-        /* paranoid */ verify(head(this)->link.next == tail(this) );
-        /* paranoid */ verify(tail(this)->link.next == 0p );
-        /* paranoid */ verify(tail(this)->link.prev == head(this) );
+        #if !defined(USE_MPSC)
+                // Make sure the list is empty
+                /* paranoid */ verify(head(this)->link.prev == 0p );
+                /* paranoid */ verify(head(this)->link.next == tail(this) );
+                /* paranoid */ verify(tail(this)->link.next == 0p );
+                /* paranoid */ verify(tail(this)->link.prev == head(this) );
+        #endif
 }
 
@@ -97 +116 @@
 // returns true of lane was empty before push, false otherwise
 bool push(__intrusive_lane_t & this, $thread * node) {
-        #if defined(__CFA_WITH_VERIFY__)
-                /* paranoid */ verify(this.lock);
-                /* paranoid */ verify(node->link.ts != 0);
-                /* paranoid */ verify(node->link.next == 0p);
-                /* paranoid */ verify(node->link.prev == 0p);
-                /* paranoid */ verify(tail(this)->link.next == 0p);
-                /* paranoid */ verify(head(this)->link.prev == 0p);
-
+        #if defined(USE_MPSC)
+                inline $thread * volatile & ?`next ( $thread * this )  __attribute__((const)) {
+                        return this->link.next;
+                }
+                push(this.queue, node);
+        #else
+                #if defined(__CFA_WITH_VERIFY__)
+                        /* paranoid */ verify(this.lock);
+                        /* paranoid */ verify(node->link.ts != 0);
+                        /* paranoid */ verify(node->link.next == 0p);
+                        /* paranoid */ verify(node->link.prev == 0p);
+                        /* paranoid */ verify(tail(this)->link.next == 0p);
+                        /* paranoid */ verify(head(this)->link.prev == 0p);
+
+                        if(this.before.link.ts == 0l) {
+                                /* paranoid */ verify(tail(this)->link.prev == head(this));
+                                /* paranoid */ verify(head(this)->link.next == tail(this));
+                        } else {
+                                /* paranoid */ verify(tail(this)->link.prev != head(this));
+                                /* paranoid */ verify(head(this)->link.next != tail(this));
+                        }
+                #endif
+
+                // Get the relevant nodes locally
+                $thread * tail = tail(this);
+                $thread * prev = tail->link.prev;
+
+                // Do the push
+                node->link.next = tail;
+                node->link.prev = prev;
+                prev->link.next = node;
+                tail->link.prev = node;
+
+                // Update stats
+                #if !defined(__CFA_NO_SCHED_STATS__)
+                        this.stat.diff++;
+                        this.stat.push++;
+                #endif
+
+                verify(node->link.next == tail(this));
+
+                // Check if the queue used to be empty
                 if(this.before.link.ts == 0l) {
-                        /* paranoid */ verify(tail(this)->link.prev == head(this));
-                        /* paranoid */ verify(head(this)->link.next == tail(this));
-                } else {
-                        /* paranoid */ verify(tail(this)->link.prev != head(this));
-                        /* paranoid */ verify(head(this)->link.next != tail(this));
-                }
-        #endif
-
-        // Get the relevant nodes locally
-        $thread * tail = tail(this);
-        $thread * prev = tail->link.prev;
-
-        // Do the push
-        node->link.next = tail;
-        node->link.prev = prev;
-        prev->link.next = node;
-        tail->link.prev = node;
-
-        // Update stats
-        #if !defined(__CFA_NO_SCHED_STATS__)
-                this.stat.diff++;
-                this.stat.push++;
-        #endif
-
-        verify(node->link.next == tail(this));
-
-        // Check if the queue used to be empty
-        if(this.before.link.ts == 0l) {
-                this.before.link.ts = node->link.ts;
-                /* paranoid */ verify(node->link.prev == head(this));
-                return true;
-        }
-        return false;
+                        this.before.link.ts = node->link.ts;
+                        /* paranoid */ verify(node->link.prev == head(this));
+                        return true;
+                }
+                return false;
+        #endif
 }
 
@@ -146 +172 @@
 $thread * pop(__intrusive_lane_t & this) {
         /* paranoid */ verify(this.lock);
-        /* paranoid */ verify(this.before.link.ts != 0ul);
-
-        // Get anchors locally
-        $thread * head = head(this);
-        $thread * tail = tail(this);
-
-        // Get the relevant nodes locally
-        $thread * node = head->link.next;
-        $thread * next = node->link.next;
-
-        /* paranoid */ verify(node != tail);
-        /* paranoid */ verify(node);
-
-        // Do the pop
-        head->link.next = next;
-        next->link.prev = head;
-        node->link.next = 0p;
-        node->link.prev = 0p;
-
-        // Update head time stamp
-        this.before.link.ts = next->link.ts;
-
-        // Update stats
-        #ifndef __CFA_NO_SCHED_STATS__
-                this.stat.diff--;
-                this.stat.pop ++;
-        #endif
-
-        // Check if we emptied list and return accordingly
-        /* paranoid */ verify(tail(this)->link.next == 0p);
-        /* paranoid */ verify(head(this)->link.prev == 0p);
-        if(next == tail) {
-                /* paranoid */ verify(this.before.link.ts == 0);
-                /* paranoid */ verify(tail(this)->link.prev == head(this));
-                /* paranoid */ verify(head(this)->link.next == tail(this));
-                return node;
-        }
-        else {
-                /* paranoid */ verify(next->link.ts != 0);
-                /* paranoid */ verify(tail(this)->link.prev != head(this));
-                /* paranoid */ verify(head(this)->link.next != tail(this));
-                /* paranoid */ verify(this.before.link.ts != 0);
-                return node;
-        }
+        #if defined(USE_MPSC)
+                inline $thread * volatile & ?`next ( $thread * this )  __attribute__((const)) {
+                        return this->link.next;
+                }
+                return pop(this.queue);
+        #else
+                /* paranoid */ verify(this.before.link.ts != 0ul);
+
+                // Get anchors locally
+                $thread * head = head(this);
+                $thread * tail = tail(this);
+
+                // Get the relevant nodes locally
+                $thread * node = head->link.next;
+                $thread * next = node->link.next;
+
+                /* paranoid */ verify(node != tail);
+                /* paranoid */ verify(node);
+
+                // Do the pop
+                head->link.next = next;
+                next->link.prev = head;
+                node->link.next = 0p;
+                node->link.prev = 0p;
+
+                // Update head time stamp
+                this.before.link.ts = next->link.ts;
+
+                // Update stats
+                #ifndef __CFA_NO_SCHED_STATS__
+                        this.stat.diff--;
+                        this.stat.pop ++;
+                #endif
+
+                // Check if we emptied list and return accordingly
+                /* paranoid */ verify(tail(this)->link.next == 0p);
+                /* paranoid */ verify(head(this)->link.prev == 0p);
+                if(next == tail) {
+                        /* paranoid */ verify(this.before.link.ts == 0);
+                        /* paranoid */ verify(tail(this)->link.prev == head(this));
+                        /* paranoid */ verify(head(this)->link.next == tail(this));
+                        return node;
+                }
+                else {
+                        /* paranoid */ verify(next->link.ts != 0);
+                        /* paranoid */ verify(tail(this)->link.prev != head(this));
+                        /* paranoid */ verify(head(this)->link.next != tail(this));
+                        /* paranoid */ verify(this.before.link.ts != 0);
+                        return node;
+                }
+        #endif
 }
 
 // Check whether or not list is empty
 static inline bool is_empty(__intrusive_lane_t & this) {
-        // Cannot verify here since it may not be locked
-        return this.before.link.ts == 0;
+        #if defined(USE_MPSC)
+                return this.queue.head == 0p;
+        #else
+                // Cannot verify here since it may not be locked
+                return this.before.link.ts == 0;
+        #endif
 }
 
 // Return the timestamp
 static inline unsigned long long ts(__intrusive_lane_t & this) {
-        // Cannot verify here since it may not be locked
-        return this.before.link.ts;
-}
+        #if defined(USE_MPSC)
+                $thread * tl = this.queue.head;
+                if(!tl) return -1ull;
+                return tl->link.ts;
+        #else
+                // Cannot verify here since it may not be locked
+                return this.before.link.ts;
+        #endif
+}
+
+// Aligned timestamps which are used by the relaxed ready queue
+struct __attribute__((aligned(128))) __timestamp_t {
+        volatile unsigned long long tv;
+};
+
+void  ?{}(__timestamp_t & this) { this.tv = 0; }
+void ^?{}(__timestamp_t & this) {}

libcfa/src/concurrency/stats.cfa

@@ -5 +5 @@
 #include <inttypes.h>
 #include "bits/debug.hfa"
+#include "bits/locks.hfa"
 #include "stats.hfa"
+#include "strstream.hfa"
 
 #if !defined(__CFA_NO_STATISTICS__)
         void __init_stats( struct __stats_t * stats ) {
-                stats->ready.pick.push.attempt  = 0;
-                stats->ready.pick.push.success  = 0;
-                stats->ready.pick.push.local    = 0;
-                stats->ready.pick.push.lsuccess = 0;
-                stats->ready.pick.pop .probe    = 0;
-                stats->ready.pick.pop .attempt  = 0;
-                stats->ready.pick.pop .success  = 0;
-                stats->ready.pick.pop .local    = 0;
-                stats->ready.pick.pop .lsuccess = 0;
+                stats->ready.push.local.attempt = 0;
+                stats->ready.push.local.success = 0;
+                stats->ready.push.share.attempt = 0;
+                stats->ready.push.share.success = 0;
+                stats->ready.push.extrn.attempt = 0;
+                stats->ready.push.extrn.success = 0;
+                stats->ready.pop.local .attempt = 0;
+                stats->ready.pop.local .success = 0;
+                stats->ready.pop.local .elock   = 0;
+                stats->ready.pop.local .eempty  = 0;
+                stats->ready.pop.local .espec   = 0;
+                stats->ready.pop.help  .attempt = 0;
+                stats->ready.pop.help  .success = 0;
+                stats->ready.pop.help  .elock   = 0;
+                stats->ready.pop.help  .eempty  = 0;
+                stats->ready.pop.help  .espec   = 0;
+                stats->ready.pop.steal .attempt = 0;
+                stats->ready.pop.steal .success = 0;
+                stats->ready.pop.steal .elock   = 0;
+                stats->ready.pop.steal .eempty  = 0;
+                stats->ready.pop.steal .espec   = 0;
+                stats->ready.pop.search.attempt = 0;
+                stats->ready.pop.search.success = 0;
+                stats->ready.pop.search.elock   = 0;
+                stats->ready.pop.search.eempty  = 0;
+                stats->ready.pop.search.espec   = 0;
                 stats->ready.threads.migration = 0;
+                stats->ready.threads.extunpark = 0;
+                stats->ready.threads.threads   = 0;
                 stats->ready.sleep.halts   = 0;
                 stats->ready.sleep.cancels = 0;
@@ -25 +46 @@
 
                 #if defined(CFA_HAVE_LINUX_IO_URING_H)
-                        stats->io.submit_q.submit_avg.rdy = 0;
-                        stats->io.submit_q.submit_avg.csm = 0;
-                        stats->io.submit_q.submit_avg.cnt = 0;
-                        stats->io.submit_q.look_avg.val   = 0;
-                        stats->io.submit_q.look_avg.cnt   = 0;
-                        stats->io.submit_q.look_avg.block = 0;
-                        stats->io.submit_q.alloc_avg.val   = 0;
-                        stats->io.submit_q.alloc_avg.cnt   = 0;
-                        stats->io.submit_q.alloc_avg.block = 0;
-                        stats->io.submit_q.helped = 0;
-                        stats->io.submit_q.leader = 0;
-                        stats->io.submit_q.busy   = 0;
-                        stats->io.complete_q.completed_avg.val = 0;
-                        stats->io.complete_q.completed_avg.cnt = 0;
-                        stats->io.complete_q.blocks = 0;
+                        stats->io.alloc.fast        = 0;
+                        stats->io.alloc.slow        = 0;
+                        stats->io.alloc.fail        = 0;
+                        stats->io.alloc.revoke      = 0;
+                        stats->io.alloc.block       = 0;
+                        stats->io.submit.fast       = 0;
+                        stats->io.submit.slow       = 0;
+                        stats->io.flush.external    = 0;
+                        stats->io.calls.flush       = 0;
+                        stats->io.calls.submitted   = 0;
+                        stats->io.calls.drain       = 0;
+                        stats->io.calls.completed   = 0;
+                        stats->io.calls.errors.busy = 0;
+                        stats->io.poller.sleeps     = 0;
+                #endif
+
+                #if defined(CFA_STATS_ARRAY)
+                        stats->array.values = alloc(CFA_STATS_ARRAY);
+                        stats->array.cnt = 0;
                 #endif
         }
 
         void __tally_stats( struct __stats_t * cltr, struct __stats_t * proc ) {
-                __atomic_fetch_add( &cltr->ready.pick.push.attempt , proc->ready.pick.push.attempt , __ATOMIC_SEQ_CST ); proc->ready.pick.push.attempt  = 0;
-                __atomic_fetch_add( &cltr->ready.pick.push.success , proc->ready.pick.push.success , __ATOMIC_SEQ_CST ); proc->ready.pick.push.success  = 0;
-                __atomic_fetch_add( &cltr->ready.pick.push.local   , proc->ready.pick.push.local   , __ATOMIC_SEQ_CST ); proc->ready.pick.push.local    = 0;
-                __atomic_fetch_add( &cltr->ready.pick.push.lsuccess, proc->ready.pick.push.lsuccess, __ATOMIC_SEQ_CST ); proc->ready.pick.push.lsuccess = 0;
-                __atomic_fetch_add( &cltr->ready.pick.pop .probe   , proc->ready.pick.pop .probe   , __ATOMIC_SEQ_CST ); proc->ready.pick.pop .probe    = 0;
-                __atomic_fetch_add( &cltr->ready.pick.pop .attempt , proc->ready.pick.pop .attempt , __ATOMIC_SEQ_CST ); proc->ready.pick.pop .attempt  = 0;
-                __atomic_fetch_add( &cltr->ready.pick.pop .success , proc->ready.pick.pop .success , __ATOMIC_SEQ_CST ); proc->ready.pick.pop .success  = 0;
-                __atomic_fetch_add( &cltr->ready.pick.pop .local   , proc->ready.pick.pop .local   , __ATOMIC_SEQ_CST ); proc->ready.pick.pop .local    = 0;
-                __atomic_fetch_add( &cltr->ready.pick.pop .lsuccess, proc->ready.pick.pop .lsuccess, __ATOMIC_SEQ_CST ); proc->ready.pick.pop .lsuccess = 0;
+                __atomic_fetch_add( &cltr->ready.push.local.attempt, proc->ready.push.local.attempt, __ATOMIC_SEQ_CST ); proc->ready.push.local.attempt = 0;
+                __atomic_fetch_add( &cltr->ready.push.local.success, proc->ready.push.local.success, __ATOMIC_SEQ_CST ); proc->ready.push.local.success = 0;
+                __atomic_fetch_add( &cltr->ready.push.share.attempt, proc->ready.push.share.attempt, __ATOMIC_SEQ_CST ); proc->ready.push.share.attempt = 0;
+                __atomic_fetch_add( &cltr->ready.push.share.success, proc->ready.push.share.success, __ATOMIC_SEQ_CST ); proc->ready.push.share.success = 0;
+                __atomic_fetch_add( &cltr->ready.push.extrn.attempt, proc->ready.push.extrn.attempt, __ATOMIC_SEQ_CST ); proc->ready.push.extrn.attempt = 0;
+                __atomic_fetch_add( &cltr->ready.push.extrn.success, proc->ready.push.extrn.success, __ATOMIC_SEQ_CST ); proc->ready.push.extrn.success = 0;
+                __atomic_fetch_add( &cltr->ready.pop.local .attempt, proc->ready.pop.local .attempt, __ATOMIC_SEQ_CST ); proc->ready.pop.local .attempt = 0;
+                __atomic_fetch_add( &cltr->ready.pop.local .success, proc->ready.pop.local .success, __ATOMIC_SEQ_CST ); proc->ready.pop.local .success = 0;
+                __atomic_fetch_add( &cltr->ready.pop.local .elock  , proc->ready.pop.local .elock  , __ATOMIC_SEQ_CST ); proc->ready.pop.local .elock   = 0;
+                __atomic_fetch_add( &cltr->ready.pop.local .eempty , proc->ready.pop.local .eempty , __ATOMIC_SEQ_CST ); proc->ready.pop.local .eempty  = 0;
+                __atomic_fetch_add( &cltr->ready.pop.local .espec  , proc->ready.pop.local .espec  , __ATOMIC_SEQ_CST ); proc->ready.pop.local .espec   = 0;
+                __atomic_fetch_add( &cltr->ready.pop.help  .attempt, proc->ready.pop.help  .attempt, __ATOMIC_SEQ_CST ); proc->ready.pop.help  .attempt = 0;
+                __atomic_fetch_add( &cltr->ready.pop.help  .success, proc->ready.pop.help  .success, __ATOMIC_SEQ_CST ); proc->ready.pop.help  .success = 0;
+                __atomic_fetch_add( &cltr->ready.pop.help  .elock  , proc->ready.pop.help  .elock  , __ATOMIC_SEQ_CST ); proc->ready.pop.help  .elock   = 0;
+                __atomic_fetch_add( &cltr->ready.pop.help  .eempty , proc->ready.pop.help  .eempty , __ATOMIC_SEQ_CST ); proc->ready.pop.help  .eempty  = 0;
+                __atomic_fetch_add( &cltr->ready.pop.help  .espec  , proc->ready.pop.help  .espec  , __ATOMIC_SEQ_CST ); proc->ready.pop.help  .espec   = 0;
+                __atomic_fetch_add( &cltr->ready.pop.steal .attempt, proc->ready.pop.steal .attempt, __ATOMIC_SEQ_CST ); proc->ready.pop.steal .attempt = 0;
+                __atomic_fetch_add( &cltr->ready.pop.steal .success, proc->ready.pop.steal .success, __ATOMIC_SEQ_CST ); proc->ready.pop.steal .success = 0;
+                __atomic_fetch_add( &cltr->ready.pop.steal .elock  , proc->ready.pop.steal .elock  , __ATOMIC_SEQ_CST ); proc->ready.pop.steal .elock   = 0;
+                __atomic_fetch_add( &cltr->ready.pop.steal .eempty , proc->ready.pop.steal .eempty , __ATOMIC_SEQ_CST ); proc->ready.pop.steal .eempty  = 0;
+                __atomic_fetch_add( &cltr->ready.pop.steal .espec  , proc->ready.pop.steal .espec  , __ATOMIC_SEQ_CST ); proc->ready.pop.steal .espec   = 0;
+                __atomic_fetch_add( &cltr->ready.pop.search.attempt, proc->ready.pop.search.attempt, __ATOMIC_SEQ_CST ); proc->ready.pop.search.attempt = 0;
+                __atomic_fetch_add( &cltr->ready.pop.search.success, proc->ready.pop.search.success, __ATOMIC_SEQ_CST ); proc->ready.pop.search.success = 0;
+                __atomic_fetch_add( &cltr->ready.pop.search.elock  , proc->ready.pop.search.elock  , __ATOMIC_SEQ_CST ); proc->ready.pop.search.elock   = 0;
+                __atomic_fetch_add( &cltr->ready.pop.search.eempty , proc->ready.pop.search.eempty , __ATOMIC_SEQ_CST ); proc->ready.pop.search.eempty  = 0;
+                __atomic_fetch_add( &cltr->ready.pop.search.espec  , proc->ready.pop.search.espec  , __ATOMIC_SEQ_CST ); proc->ready.pop.search.espec   = 0;
                 __atomic_fetch_add( &cltr->ready.threads.migration , proc->ready.threads.migration , __ATOMIC_SEQ_CST ); proc->ready.threads.migration  = 0;
+                __atomic_fetch_add( &cltr->ready.threads.extunpark , proc->ready.threads.extunpark , __ATOMIC_SEQ_CST ); proc->ready.threads.extunpark  = 0;
+                __atomic_fetch_add( &cltr->ready.threads.threads   , proc->ready.threads.threads   , __ATOMIC_SEQ_CST ); proc->ready.threads.threads    = 0;
                 __atomic_fetch_add( &cltr->ready.sleep.halts       , proc->ready.sleep.halts       , __ATOMIC_SEQ_CST ); proc->ready.sleep.halts        = 0;
                 __atomic_fetch_add( &cltr->ready.sleep.cancels     , proc->ready.sleep.cancels     , __ATOMIC_SEQ_CST ); proc->ready.sleep.cancels      = 0;
@@ -60 +104 @@
 
                 #if defined(CFA_HAVE_LINUX_IO_URING_H)
-                        __atomic_fetch_add( &cltr->io.submit_q.submit_avg.rdy     , proc->io.submit_q.submit_avg.rdy     , __ATOMIC_SEQ_CST ); proc->io.submit_q.submit_avg.rdy      = 0;
-                        __atomic_fetch_add( &cltr->io.submit_q.submit_avg.csm     , proc->io.submit_q.submit_avg.csm     , __ATOMIC_SEQ_CST ); proc->io.submit_q.submit_avg.csm      = 0;
-                        __atomic_fetch_add( &cltr->io.submit_q.submit_avg.avl     , proc->io.submit_q.submit_avg.avl     , __ATOMIC_SEQ_CST ); proc->io.submit_q.submit_avg.avl      = 0;
-                        __atomic_fetch_add( &cltr->io.submit_q.submit_avg.cnt     , proc->io.submit_q.submit_avg.cnt     , __ATOMIC_SEQ_CST ); proc->io.submit_q.submit_avg.cnt      = 0;
-                        __atomic_fetch_add( &cltr->io.submit_q.look_avg.val       , proc->io.submit_q.look_avg.val       , __ATOMIC_SEQ_CST ); proc->io.submit_q.look_avg.val        = 0;
-                        __atomic_fetch_add( &cltr->io.submit_q.look_avg.cnt       , proc->io.submit_q.look_avg.cnt       , __ATOMIC_SEQ_CST ); proc->io.submit_q.look_avg.cnt        = 0;
-                        __atomic_fetch_add( &cltr->io.submit_q.look_avg.block     , proc->io.submit_q.look_avg.block     , __ATOMIC_SEQ_CST ); proc->io.submit_q.look_avg.block      = 0;
-                        __atomic_fetch_add( &cltr->io.submit_q.alloc_avg.val      , proc->io.submit_q.alloc_avg.val      , __ATOMIC_SEQ_CST ); proc->io.submit_q.alloc_avg.val       = 0;
-                        __atomic_fetch_add( &cltr->io.submit_q.alloc_avg.cnt      , proc->io.submit_q.alloc_avg.cnt      , __ATOMIC_SEQ_CST ); proc->io.submit_q.alloc_avg.cnt       = 0;
-                        __atomic_fetch_add( &cltr->io.submit_q.alloc_avg.block    , proc->io.submit_q.alloc_avg.block    , __ATOMIC_SEQ_CST ); proc->io.submit_q.alloc_avg.block     = 0;
-                        __atomic_fetch_add( &cltr->io.submit_q.helped             , proc->io.submit_q.helped             , __ATOMIC_SEQ_CST ); proc->io.submit_q.helped              = 0;
-                        __atomic_fetch_add( &cltr->io.submit_q.leader             , proc->io.submit_q.leader             , __ATOMIC_SEQ_CST ); proc->io.submit_q.leader              = 0;
-                        __atomic_fetch_add( &cltr->io.submit_q.busy               , proc->io.submit_q.busy               , __ATOMIC_SEQ_CST ); proc->io.submit_q.busy                = 0;
-                        __atomic_fetch_add( &cltr->io.complete_q.completed_avg.val, proc->io.complete_q.completed_avg.val, __ATOMIC_SEQ_CST ); proc->io.complete_q.completed_avg.val = 0;
-                        __atomic_fetch_add( &cltr->io.complete_q.completed_avg.cnt, proc->io.complete_q.completed_avg.cnt, __ATOMIC_SEQ_CST ); proc->io.complete_q.completed_avg.cnt = 0;
-                        __atomic_fetch_add( &cltr->io.complete_q.blocks           , proc->io.complete_q.blocks           , __ATOMIC_SEQ_CST ); proc->io.complete_q.blocks            = 0;
+                        __atomic_fetch_add( &cltr->io.alloc.fast       , proc->io.alloc.fast       , __ATOMIC_SEQ_CST ); proc->io.alloc.fast        = 0;
+                        __atomic_fetch_add( &cltr->io.alloc.slow       , proc->io.alloc.slow       , __ATOMIC_SEQ_CST ); proc->io.alloc.slow        = 0;
+                        __atomic_fetch_add( &cltr->io.alloc.fail       , proc->io.alloc.fail       , __ATOMIC_SEQ_CST ); proc->io.alloc.fail        = 0;
+                        __atomic_fetch_add( &cltr->io.alloc.revoke     , proc->io.alloc.revoke     , __ATOMIC_SEQ_CST ); proc->io.alloc.revoke      = 0;
+                        __atomic_fetch_add( &cltr->io.alloc.block      , proc->io.alloc.block      , __ATOMIC_SEQ_CST ); proc->io.alloc.block       = 0;
+                        __atomic_fetch_add( &cltr->io.submit.fast      , proc->io.submit.fast      , __ATOMIC_SEQ_CST ); proc->io.submit.fast       = 0;
+                        __atomic_fetch_add( &cltr->io.submit.slow      , proc->io.submit.slow      , __ATOMIC_SEQ_CST ); proc->io.submit.slow       = 0;
+                        __atomic_fetch_add( &cltr->io.flush.external   , proc->io.flush.external   , __ATOMIC_SEQ_CST ); proc->io.flush.external    = 0;
+                        __atomic_fetch_add( &cltr->io.calls.flush      , proc->io.calls.flush      , __ATOMIC_SEQ_CST ); proc->io.calls.flush       = 0;
+                        __atomic_fetch_add( &cltr->io.calls.submitted  , proc->io.calls.submitted  , __ATOMIC_SEQ_CST ); proc->io.calls.submitted   = 0;
+                        __atomic_fetch_add( &cltr->io.calls.drain      , proc->io.calls.drain      , __ATOMIC_SEQ_CST ); proc->io.calls.drain       = 0;
+                        __atomic_fetch_add( &cltr->io.calls.completed  , proc->io.calls.completed  , __ATOMIC_SEQ_CST ); proc->io.calls.completed   = 0;
+                        __atomic_fetch_add( &cltr->io.calls.errors.busy, proc->io.calls.errors.busy, __ATOMIC_SEQ_CST ); proc->io.calls.errors.busy = 0;
+                        __atomic_fetch_add( &cltr->io.poller.sleeps    , proc->io.poller.sleeps    , __ATOMIC_SEQ_CST ); proc->io.poller.sleeps     = 0;
                 #endif
         }
 
+        #define eng3(X) (ws(3, 3, unit(eng( X ))))
+
         void __print_stats( struct __stats_t * stats, int flags, const char * type, const char * name, void * id ) with( *stats ) {
 
+                char buf[1024];
+                ostrstream sstr = { buf, 1024 };
+
                 if( flags & CFA_STATS_READY_Q ) {
-                        double push_sur = (100.0 * ((double)ready.pick.push.success) / ready.pick.push.attempt);
-                        double pop_sur  = (100.0 * ((double)ready.pick.pop .success) / ready.pick.pop .attempt);
-
-                        double push_len = ((double)ready.pick.push.attempt) / ready.pick.push.success;
-                        double pop_len  = ((double)ready.pick.pop .attempt) / ready.pick.pop .success;
-
-                        double lpush_sur = (100.0 * ((double)ready.pick.push.lsuccess) / ready.pick.push.local);
-                        double lpop_sur  = (100.0 * ((double)ready.pick.pop .lsuccess) / ready.pick.pop .local);
-
-                        double lpush_len = ((double)ready.pick.push.local) / ready.pick.push.lsuccess;
-                        double lpop_len  = ((double)ready.pick.pop .local) / ready.pick.pop .lsuccess;
-
-                        __cfaabi_bits_print_safe( STDOUT_FILENO,
-                                "----- %s \"%s\" (%p) - Ready Q Stats -----\n"
-                                "- total threads run      : %'15" PRIu64 "\n"
-                                "- total threads scheduled: %'15" PRIu64 "\n"
-                                "- push average probe len : %'18.2lf, %'18.2lf%% (%'15" PRIu64 " attempts)\n"
-                                "- pop  average probe len : %'18.2lf, %'18.2lf%% (%'15" PRIu64 " attempts)\n"
-                                "- local push avg prb len : %'18.2lf, %'18.2lf%% (%'15" PRIu64 " attempts)\n"
-                                "- local pop  avg prb len : %'18.2lf, %'18.2lf%% (%'15" PRIu64 " attempts)\n"
-                                "- thread migrations      : %'15" PRIu64 "\n"
-                                "- Idle Sleep -\n"
-                                "-- halts                 : %'15" PRIu64 "\n"
-                                "-- cancelled halts       : %'15" PRIu64 "\n"
-                                "-- schedule wake         : %'15" PRIu64 "\n"
-                                "-- wake on exit          : %'15" PRIu64 "\n"
-                                "\n"
-                                , type, name, id
-                                , ready.pick.pop.success
-                                , ready.pick.push.success
-                                , push_len, push_sur, ready.pick.push.attempt
-                                , pop_len , pop_sur , ready.pick.pop .attempt
-                                , lpush_len, lpush_sur, ready.pick.push.local
-                                , lpop_len , lpop_sur , ready.pick.pop .local
-                                , ready.threads.migration
-                                , ready.sleep.halts, ready.sleep.cancels, ready.sleep.wakes, ready.sleep.exits
-                        );
+
+                        sstr | "----- " | type | "\"" | name | "\" (" | "" | id | "" | ") - Ready Q Stats -----";
+
+                        uint64_t totalR = ready.pop.local.success + ready.pop.help.success + ready.pop.steal.success + ready.pop.search.success;
+                        uint64_t totalS = ready.push.local.success + ready.push.share.success + ready.push.extrn.success;
+                        sstr | "- totals   : " | eng3(totalR) | "run," | eng3(totalS) | "schd (" | eng3(ready.push.extrn.success) | "ext," | eng3(ready.threads.migration) | "mig," | eng3(ready.threads.extunpark) | " eupk)";
+
+                        double push_len = ((double)ready.push.local.attempt + ready.push.share.attempt + ready.push.extrn.attempt) / totalS;
+                        double sLcl_len = ready.push.local.success ? ((double)ready.push.local.attempt) / ready.push.local.success : 0;
+                        double sOth_len = ready.push.share.success ? ((double)ready.push.share.attempt) / ready.push.share.success : 0;
+                        double sExt_len = ready.push.extrn.success ? ((double)ready.push.extrn.attempt) / ready.push.extrn.success : 0;
+                        sstr | "- push avg : " | ws(3, 3, push_len)
+                             | "- l: " | eng3(ready.push.local.attempt) | " (" | ws(3, 3, sLcl_len) | ")"
+                             | ", s: " | eng3(ready.push.share.attempt) | " (" | ws(3, 3, sOth_len) | ")"
+                             | ", e: " | eng3(ready.push.extrn.attempt) | " (" | ws(3, 3, sExt_len) | ")";
+
+                        double rLcl_pc = (100.0 * (double)ready.pop.local .success) / totalR;
+                        sstr | "- local    : " | eng3(ready.pop.local .success) | "-"| ws(3, 3, rLcl_pc) | '%'
+                             | " (" | eng3(ready.pop.local .attempt) | " try," | eng3(ready.pop.local .espec) | " spc," | eng3(ready.pop.local .elock) | " lck," | eng3(ready.pop.local .eempty) | " ept)";
+                        double rHlp_pc = (100.0 * (double)ready.pop.help  .success) / totalR;
+                        sstr | "- help     : " | eng3(ready.pop.help  .success) | "-"| ws(3, 3, rHlp_pc) | '%'
+                             | " (" | eng3(ready.pop.help  .attempt) | " try," | eng3(ready.pop.help  .espec) | " spc," | eng3(ready.pop.help  .elock) | " lck," | eng3(ready.pop.help  .eempty) | " ept)";
+                        double rStl_pc = (100.0 * (double)ready.pop.steal .success) / totalR;
+                        sstr | "- steal    : " | eng3(ready.pop.steal .success) | "-"| ws(3, 3, rStl_pc) | '%'
+                             | " (" | eng3(ready.pop.steal .attempt) | " try," | eng3(ready.pop.steal .espec) | " spc," | eng3(ready.pop.steal .elock) | " lck," | eng3(ready.pop.steal .eempty) | " ept)";
+                        double rSch_pc = (100.0 * (double)ready.pop.search.success) / totalR;
+                        sstr | "- search   : " | eng3(ready.pop.search.success) | "-"| ws(3, 3, rSch_pc) | '%'
+                             | " (" | eng3(ready.pop.search.attempt) | " try," | eng3(ready.pop.search.espec) | " spc," | eng3(ready.pop.search.elock) | " lck," | eng3(ready.pop.search.eempty) | " ept)";
+
+                        sstr | "- Idle Slp : " | eng3(ready.sleep.halts) | "halt," | eng3(ready.sleep.cancels) | "cancel," | eng3(ready.sleep.wakes) | "wake," | eng3(ready.sleep.exits) | "exit";
+                        sstr | nl;
                 }
 
                 #if defined(CFA_HAVE_LINUX_IO_URING_H)
                         if( flags & CFA_STATS_IO ) {
-                                double avgrdy = ((double)io.submit_q.submit_avg.rdy) / io.submit_q.submit_avg.cnt;
-                                double avgcsm = ((double)io.submit_q.submit_avg.csm) / io.submit_q.submit_avg.cnt;
-
-                                double lavgv = 0;
-                                double lavgb = 0;
-                                if(io.submit_q.look_avg.cnt != 0) {
-                                        lavgv = ((double)io.submit_q.look_avg.val  ) / io.submit_q.look_avg.cnt;
-                                        lavgb = ((double)io.submit_q.look_avg.block) / io.submit_q.look_avg.cnt;
-                                }
-
-                                double aavgv = 0;
-                                double aavgb = 0;
-                                if(io.submit_q.alloc_avg.cnt != 0) {
-                                        aavgv = ((double)io.submit_q.alloc_avg.val  ) / io.submit_q.alloc_avg.cnt;
-                                        aavgb = ((double)io.submit_q.alloc_avg.block) / io.submit_q.alloc_avg.cnt;
-                                }
-
-                                __cfaabi_bits_print_safe( STDOUT_FILENO,
-                                        "----- %s \"%s\" (%p) - I/O Stats -----\n"
-                                        "- total submit calls     : %'15" PRIu64 "\n"
-                                        "- avg ready entries      : %'18.2lf\n"
-                                        "- avg submitted entries  : %'18.2lf\n"
-                                        "- total helped entries   : %'15" PRIu64 "\n"
-                                        "- total leader entries   : %'15" PRIu64 "\n"
-                                        "- total busy submit      : %'15" PRIu64 "\n"
-                                        "- total ready search     : %'15" PRIu64 "\n"
-                                        "- avg ready search len   : %'18.2lf\n"
-                                        "- avg ready search block : %'18.2lf\n"
-                                        "- total alloc search     : %'15" PRIu64 "\n"
-                                        "- avg alloc search len   : %'18.2lf\n"
-                                        "- avg alloc search block : %'18.2lf\n"
-                                        "- total wait calls       : %'15" PRIu64 "\n"
-                                        "- avg completion/wait    : %'18.2lf\n"
-                                        "- total completion blocks: %'15" PRIu64 "\n"
-                                        "\n"
-                                        , type,  name, id
-                                        , io.submit_q.submit_avg.cnt
-                                        , avgrdy, avgcsm
-                                        , io.submit_q.helped, io.submit_q.leader, io.submit_q.busy
-                                        , io.submit_q.look_avg.cnt
-                                        , lavgv, lavgb
-                                        , io.submit_q.alloc_avg.cnt
-                                        , aavgv, aavgb
-                                        , io.complete_q.completed_avg.cnt
-                                        , ((double)io.complete_q.completed_avg.val) / io.complete_q.completed_avg.cnt
-                                        , io.complete_q.blocks
-                                );
+                                sstr | "----- " | type | "\"" | name | "\" (" | "" | id | "" | ") - I/O Stats -----";
+
+                                uint64_t total_allocs = io.alloc.fast + io.alloc.slow;
+                                double avgfasta = (100.0 * (double)io.alloc.fast) / total_allocs;
+                                sstr | "- total allocations : " | eng3(io.alloc.fast) | "fast," | eng3(io.alloc.slow) | "slow (" | ws(3, 3, avgfasta) | "%)";
+                                sstr | "-     failures      : " | eng3(io.alloc.fail) | "oom, " | eng3(io.alloc.revoke) | "rvk, " | eng3(io.alloc.block) | "blk";
+
+                                uint64_t total_submits = io.submit.fast + io.submit.slow;
+                                double avgfasts = (100.0 * (double)io.submit.fast) / total_submits;
+                                sstr | "- total submits     : " | eng3(io.submit.fast) | "fast," | eng3(io.submit.slow) | "slow (" | ws(3, 3, avgfasts) | "%)";
+                                sstr | "- flush external    : " | eng3(io.flush.external);
+
+                                sstr | "- io_uring_enter    : " | eng3(io.calls.flush) | " (" | eng3(io.calls.drain) | ", " | eng3(io.calls.errors.busy) | " EBUSY)";
+
+                                double avgsubs = ((double)io.calls.submitted) / io.calls.flush;
+                                double avgcomp = ((double)io.calls.completed) / io.calls.drain;
+                                sstr | "-     submits       : " | eng3(io.calls.submitted) | "(" | ws(3, 3, avgsubs) | "/flush)";
+                                sstr | "-     completes     : " | eng3(io.calls.completed) | "(" | ws(3, 3, avgcomp) | "/drain)";
+
+                                sstr | "- poller sleeping   : " | eng3(io.poller.sleeps);
+                                sstr | nl;
                         }
                 #endif
+
+                if(flags) write( sstr, stdout );
         }
+
+        #if defined(CFA_STATS_ARRAY)
+                extern "C" {
+                        #include <stdio.h>
+                        #include <errno.h>
+                        #include <sys/stat.h>
+                        #include <fcntl.h>
+                }
+
+                void __flush_stat( struct __stats_t * this, const char * name, void * handle) {
+                        int ret = mkdir(".cfadata", 0755);
+                        if(ret < 0 && errno != EEXIST) abort("Failed to create directory .cfadata: %d\n", errno);
+
+                        char filename[100];
+                        snprintf(filename, 100, ".cfadata/%s%p.data", name, handle);
+
+                        int fd = open(filename, O_WRONLY | O_APPEND | O_CREAT, 0644);
+                        if(fd < 0) abort("Failed to create file %s: %d\n", filename, errno);
+
+                        for(i; this->array.cnt) {
+                                char line[100];
+                                size_t n = snprintf(line, 100, "%llu, %lld\n", this->array.values[i].ts, this->array.values[i].value);
+                                write(fd, line, n);
+                        }
+
+                        this->array.cnt = 0;
+                        close(fd);
+                }
+
+                static __spinlock_t stats_lock;
+
+                void __push_stat( struct __stats_t * this, int64_t value, bool external, const char * name, void * handle ) {
+                        if(external) lock(stats_lock __cfaabi_dbg_ctx2);
+
+                        if( this->array.cnt >= CFA_STATS_ARRAY ) __flush_stat( this, name, handle );
+
+                        size_t idx = this->array.cnt;
+                        this->array.cnt++;
+
+                        if(external) unlock(stats_lock);
+
+                        this->array.values[idx].ts = rdtscl();
+                        this->array.values[idx].value = value;
+                }
+        #endif
 #endif

libcfa/src/concurrency/stats.hfa

@@ -1 +1 @@
 #pragma once
+
+// #define CFA_STATS_ARRAY 10000
 
 #include <stdint.h>
@@ -14 +16 @@
         static inline void __print_stats( struct __stats_t *, int, const char *, const char *, void * ) {}
 #else
+        struct __stats_readyQ_pop_t {
+                // number of attemps at poping something
+                volatile uint64_t attempt;
 
-        struct __attribute__((aligned(64))) __stats_readQ_t {
+                // number of successes at poping
+                volatile uint64_t success;
+
+                // number of attempts failed due to the lock being held
+                volatile uint64_t elock;
+
+                // number of attempts failed due to the queue being empty (lock held)
+                volatile uint64_t eempty;
+
+                // number of attempts failed due to the queue looking empty (lock not held)
+                volatile uint64_t espec;
+        };
+
+        struct __attribute__((aligned(64))) __stats_readyQ_t {
+                // Push statistic
                 struct {
-                        // Push statistic
                         struct {
-                                // number of attemps at pushing something
+                                // number of attemps at pushing something to preferred queues
                                 volatile uint64_t attempt;
 
-                                // number of successes at pushing
+                                // number of successes at pushing to preferred queues
                                 volatile uint64_t success;
+                        }
+                        // Stats for local queue within cluster
+                        local,
 
-                                // number of attemps at pushing something to preferred queues
-                                volatile uint64_t local;
+                        // Stats for non-local queues within cluster
+                        share,
 
-                                // number of successes at pushing to preferred queues
-                                volatile uint64_t lsuccess;
-                        } push;
+                        // Stats from outside cluster
+                        extrn;
+                } push;
 
-                        // Pop statistic
-                        struct {
-                                // number of reads of the mask
-                                // picking an empty __cfa_readyQ_mask_t counts here
-                                // but not as an attempt
-                                volatile uint64_t probe;
+                // Pop statistic
+                struct {
+                        // pop from local queue
+                        __stats_readyQ_pop_t local;
 
-                                // number of attemps at poping something
-                                volatile uint64_t attempt;
+                        // pop before looking at local queue
+                        __stats_readyQ_pop_t help;
 
-                                // number of successes at poping
-                                volatile uint64_t success;
+                        // pop from some other queue
+                        __stats_readyQ_pop_t steal;
 
-                                // number of attemps at poping something to preferred queues
-                                volatile uint64_t local;
+                        // pop when searching queues sequentially
+                        __stats_readyQ_pop_t search;
+                } pop;
 
-                                // number of successes at poping to preferred queues
-                                volatile uint64_t lsuccess;
-                        } pop;
-                } pick;
                 struct {
                         volatile uint64_t migration;
+                        volatile uint64_t extunpark;
+                        volatile  int64_t threads; // number of threads in the system, includes only local change
                 } threads;
                 struct {
@@ -66 +84 @@
                 struct __attribute__((aligned(64))) __stats_io_t{
                         struct {
+                                volatile uint64_t fast;
+                                volatile uint64_t slow;
+                                volatile uint64_t fail;
+                                volatile uint64_t revoke;
+                                volatile uint64_t block;
+                        } alloc;
+                        struct {
+                                volatile uint64_t fast;
+                                volatile uint64_t slow;
+                        } submit;
+                        struct {
+                                volatile uint64_t external;
+                        } flush;
+                        struct {
+                                volatile uint64_t drain;
+                                volatile uint64_t completed;
+                                volatile uint64_t flush;
+                                volatile uint64_t submitted;
                                 struct {
-                                        volatile uint64_t rdy;
-                                        volatile uint64_t csm;
-                                        volatile uint64_t avl;
-                                        volatile uint64_t cnt;
-                                } submit_avg;
-                                struct {
-                                        volatile uint64_t val;
-                                        volatile uint64_t cnt;
-                                        volatile uint64_t block;
-                                } look_avg;
-                                struct {
-                                        volatile uint64_t val;
-                                        volatile uint64_t cnt;
-                                        volatile uint64_t block;
-                                } alloc_avg;
-                                volatile uint64_t helped;
-                                volatile uint64_t leader;
-                                volatile uint64_t busy;
-                        } submit_q;
+                                        volatile uint64_t busy;
+                                } errors;
+                        } calls;
                         struct {
-                                struct {
-                                        volatile uint64_t val;
-                                        volatile uint64_t cnt;
-                                } completed_avg;
-                                volatile uint64_t blocks;
-                        } complete_q;
+                                volatile uint64_t sleeps;
+                        } poller;
+                };
+        #endif
+
+        #if defined(CFA_STATS_ARRAY)
+                struct __stats_elem_t {
+                        long long int ts;
+                        int64_t value;
                 };
         #endif
 
         struct __attribute__((aligned(128))) __stats_t {
-                __stats_readQ_t ready;
+                __stats_readyQ_t ready;
                 #if defined(CFA_HAVE_LINUX_IO_URING_H)
                         __stats_io_t    io;
                 #endif
+
+                #if defined(CFA_STATS_ARRAY)
+                        struct {
+                                __stats_elem_t * values;
+                                volatile size_t cnt;
+                        } array;
+                #endif
+
         };
 
@@ -106 +137 @@
         void __tally_stats( struct __stats_t *, struct __stats_t * );
         void __print_stats( struct __stats_t *, int, const char *, const char *, void * );
+        #if defined(CFA_STATS_ARRAY)
+                void __push_stat ( struct __stats_t *, int64_t value, bool external, const char * name, void * handle);
+                void __flush_stat( struct __stats_t *, const char *, void * );
+        #else
+                static inline void __push_stat ( struct __stats_t *, int64_t, bool, const char *, void * ) {}
+                static inline void __flush_stat( struct __stats_t *, const char *, void * ) {}
+        #endif
 #endif
 

libcfa/src/concurrency/thread.cfa

@@ -39 +39 @@
         link.next = 0p;
         link.prev = 0p;
-        link.preferred = -1;
+        link.preferred = -1u;
+        last_proc = 0p;
         #if defined( __CFA_WITH_VERIFY__ )
                 canary = 0x0D15EA5E0D15EA5Ep;
@@ -62 +63 @@
 }
 
-FORALL_DATA_INSTANCE(ThreadCancelled, (thread_t &), (thread_t))
-
 forall(T &)
 void copy(ThreadCancelled(T) * dst, ThreadCancelled(T) * src) {
@@ -73 +72 @@
 forall(T &)
 const char * msg(ThreadCancelled(T) *) {
-        return "ThreadCancelled";
+        return "ThreadCancelled(...)";
 }
 
 forall(T &)
 static void default_thread_cancel_handler(ThreadCancelled(T) & ) {
+        // Improve this error message, can I do formatting?
         abort( "Unhandled thread cancellation.\n" );
 }
 
-forall(T & | is_thread(T) | IS_EXCEPTION(ThreadCancelled, (T)))
+forall(T & | is_thread(T) | IS_EXCEPTION(ThreadCancelled, (T))
+    | { EHM_DEFAULT_VTABLE(ThreadCancelled, (T)); })
 void ?{}( thread_dtor_guard_t & this,
                 T & thrd, void(*cancelHandler)(ThreadCancelled(T) &)) {
-        $monitor * m = get_monitor(thrd);
+        $monitor * m = get_monitor(thrd);
         $thread * desc = get_thread(thrd);
 
@@ -103 +104 @@
         }
         desc->state = Cancelled;
-        void(*defaultResumptionHandler)(ThreadCancelled(T) &) = 
+        void(*defaultResumptionHandler)(ThreadCancelled(T) &) =
                 join ? cancelHandler : default_thread_cancel_handler;
 
+        // TODO: Remove explitate vtable set once trac#186 is fixed.
         ThreadCancelled(T) except;
-        // TODO: Remove explitate vtable set once trac#186 is fixed.
-        except.virtual_table = &get_exception_vtable(&except);
+        except.virtual_table = &_default_vtable;
         except.the_thread = &thrd;
         except.the_exception = __cfaehm_cancellation_exception( cancellation );
-        throwResume except;
+        // Why is this cast required?
+        throwResume (ThreadCancelled(T) &)except;
 
         except.the_exception->virtual_table->free( except.the_exception );
@@ -134 +136 @@
         /* paranoid */ verify( this_thrd->context.SP );
 
-        __schedule_thread( this_thrd );
-        enable_interrupts( __cfaabi_dbg_ctx );
+        schedule_thread$( this_thrd );
+        enable_interrupts();
 }
 
@@ -158 +160 @@
 
 //-----------------------------------------------------------------------------
-forall(T & | is_thread(T) | IS_RESUMPTION_EXCEPTION(ThreadCancelled, (T)))
+forall(T & | is_thread(T) | IS_RESUMPTION_EXCEPTION(ThreadCancelled, (T))
+    | { EHM_DEFAULT_VTABLE(ThreadCancelled, (T)); })
 T & join( T & this ) {
         thread_dtor_guard_t guard = { this, defaultResumptionHandler };
@@ -167 +170 @@
         disable_interrupts();
         uint64_t ret = __tls_rand();
-        enable_interrupts( __cfaabi_dbg_ctx );
+        enable_interrupts();
         return ret;
 }

libcfa/src/concurrency/thread.hfa

@@ -32 +32 @@
 };
 
-FORALL_DATA_EXCEPTION(ThreadCancelled, (thread_t &), (thread_t)) (
+EHM_FORALL_EXCEPTION(ThreadCancelled, (thread_t &), (thread_t)) (
         thread_t * the_thread;
         exception_t * the_exception;
@@ -42 +42 @@
 forall(T &)
 const char * msg(ThreadCancelled(T) *);
-
-// define that satisfies the trait without using the thread keyword
-#define DECL_THREAD(X) $thread* get_thread(X& this) __attribute__((const)) { return &this.__thrd; } void main(X& this)
 
 // Inline getters for threads/coroutines/monitors
@@ -82 +79 @@
 };
 
-forall( T & | is_thread(T) | IS_EXCEPTION(ThreadCancelled, (T)) )
+forall( T & | is_thread(T) | IS_EXCEPTION(ThreadCancelled, (T))
+    | { EHM_DEFAULT_VTABLE(ThreadCancelled, (T)); })
 void ?{}( thread_dtor_guard_t & this, T & thrd, void(*)(ThreadCancelled(T) &) );
 void ^?{}( thread_dtor_guard_t & this );
@@ -128 +126 @@
 //----------
 // join
-forall( T & | is_thread(T) | IS_RESUMPTION_EXCEPTION(ThreadCancelled, (T)) )
+forall( T & | is_thread(T) | IS_RESUMPTION_EXCEPTION(ThreadCancelled, (T))
+    | { EHM_DEFAULT_VTABLE(ThreadCancelled, (T)); })
 T & join( T & this );