Changeset 97fed44 for libcfa/src/concurrency

Timestamp:

Jan 25, 2022, 4:54:35 PM (4 years ago)

Author:

caparsons <caparson@…>

Branches:

ADT, ast-experimental, enum, forall-pointer-decay, master, pthread-emulation, qualifiedEnum

Children:

6b2d444, a488783, f681823

Parents:

f57f6ea0 (diff), 4fcbf26 (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:

• clib/cfathread.cfa (modified) (2 diffs)
• io.cfa (modified) (1 diff)
• kernel.cfa (modified) (7 diffs)
• kernel.hfa (modified) (7 diffs)
• kernel/fwd.hfa (modified) (2 diffs)
• kernel/startup.cfa (modified) (6 diffs)
• locks.hfa (modified) (2 diffs)
• ready_queue.cfa (modified) (26 diffs)
• stats.cfa (modified) (3 diffs)
• stats.hfa (modified) (1 diff)
• thread.cfa (modified) (3 diffs)

libcfa/src/concurrency/clib/cfathread.cfa

@@ -22 +22 @@
 #include "thread.hfa"
 #include "time.hfa"
+#include "stdlib.hfa"
 
 #include "cfathread.h"
@@ -195 +196 @@
                                 eevent.data.u64 = (uint64_t)active_thread();
 
-                                int id = thread_rand() % poller_cnt;
+                                int id = prng() % poller_cnt;
                                 if(0 != epoll_ctl(poller_fds[id], EPOLL_CTL_ADD, fd, &eevent))
                                 {

libcfa/src/concurrency/io.cfa

@@ -144 +144 @@
                 __ioarbiter_flush( ctx );
 
-                __STATS__( true, io.calls.flush++; )
-                int ret = syscall( __NR_io_uring_enter, ctx.fd, ctx.sq.to_submit, min_comp, min_comp > 0 ? IORING_ENTER_GETEVENTS : 0, (sigset_t *)0p, _NSIG / 8);
-                if( ret < 0 ) {
-                        switch((int)errno) {
-                        case EAGAIN:
-                        case EINTR:
-                        case EBUSY:
-                                // Update statistics
-                                __STATS__( false, io.calls.errors.busy ++; )
-                                return false;
-                        default:
-                                abort( "KERNEL ERROR: IO_URING SYSCALL - (%d) %s\n", (int)errno, strerror(errno) );
+                if(ctx.sq.to_submit != 0 || min_comp > 0) {
+
+                        __STATS__( true, io.calls.flush++; )
+                        int ret = syscall( __NR_io_uring_enter, ctx.fd, ctx.sq.to_submit, min_comp, min_comp > 0 ? IORING_ENTER_GETEVENTS : 0, (sigset_t *)0p, _NSIG / 8);
+                        if( ret < 0 ) {
+                                switch((int)errno) {
+                                case EAGAIN:
+                                case EINTR:
+                                case EBUSY:
+                                        // Update statistics
+                                        __STATS__( false, io.calls.errors.busy ++; )
+                                        return false;
+                                default:
+                                        abort( "KERNEL ERROR: IO_URING SYSCALL - (%d) %s\n", (int)errno, strerror(errno) );
+                                }
                         }
-                }
-
-                __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;
+
+                        __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;
+                }
+
                 ready_schedule_lock();
                 bool ret = __cfa_io_drain( proc );

libcfa/src/concurrency/kernel.cfa

@@ -205 +205 @@
                                 // Don't block if we are done
                                 if( __atomic_load_n(&this->do_terminate, __ATOMIC_SEQ_CST) ) break MAIN_LOOP;
-
-                                #if !defined(__CFA_NO_STATISTICS__)
-                                        __tls_stats()->ready.sleep.halts++;
-                                #endif
 
                                 // Push self to idle stack
@@ -732 +728 @@
 // Wake a thread from the front if there are any
 static void __wake_one(cluster * this) {
+        eventfd_t val;
+
         /* paranoid */ verify( ! __preemption_enabled() );
         /* paranoid */ verify( ready_schedule_islocked() );
 
         // Check if there is a sleeping processor
-        // int fd = __atomic_load_n(&this->procs.fd, __ATOMIC_SEQ_CST);
-        int fd = 0;
-        if( __atomic_load_n(&this->procs.fd, __ATOMIC_SEQ_CST) != 0 ) {
-                fd = __atomic_exchange_n(&this->procs.fd, 0, __ATOMIC_RELAXED);
-        }
-
-        // If no one is sleeping, we are done
-        if( fd == 0 ) return;
-
-        // We found a processor, wake it up
-        eventfd_t val;
-        val = 1;
-        eventfd_write( fd, val );
-
-        #if !defined(__CFA_NO_STATISTICS__)
-                if( kernelTLS().this_stats ) {
-                        __tls_stats()->ready.sleep.wakes++;
-                }
-                else {
-                        __atomic_fetch_add(&this->stats->ready.sleep.wakes, 1, __ATOMIC_RELAXED);
-                }
-        #endif
+        struct __fd_waitctx * fdp = __atomic_load_n(&this->procs.fdw, __ATOMIC_SEQ_CST);
+
+        // If no one is sleeping: we are done
+        if( fdp == 0p ) return;
+
+        int fd = 1;
+        if( __atomic_load_n(&fdp->fd, __ATOMIC_SEQ_CST) != 1 ) {
+                fd = __atomic_exchange_n(&fdp->fd, 1, __ATOMIC_RELAXED);
+        }
+
+        switch(fd) {
+        case 0:
+                // If the processor isn't ready to sleep then the exchange will already wake it up
+                #if !defined(__CFA_NO_STATISTICS__)
+                        if( kernelTLS().this_stats ) { __tls_stats()->ready.sleep.early++;
+                        } else { __atomic_fetch_add(&this->stats->ready.sleep.early, 1, __ATOMIC_RELAXED); }
+                #endif
+                break;
+        case 1:
+                // If someone else already said they will wake them: we are done
+                #if !defined(__CFA_NO_STATISTICS__)
+                        if( kernelTLS().this_stats ) { __tls_stats()->ready.sleep.seen++;
+                        } else { __atomic_fetch_add(&this->stats->ready.sleep.seen, 1, __ATOMIC_RELAXED); }
+                #endif
+                break;
+        default:
+                // If the processor was ready to sleep, we need to wake it up with an actual write
+                val = 1;
+                eventfd_write( fd, val );
+
+                #if !defined(__CFA_NO_STATISTICS__)
+                        if( kernelTLS().this_stats ) { __tls_stats()->ready.sleep.wakes++;
+                        } else { __atomic_fetch_add(&this->stats->ready.sleep.wakes, 1, __ATOMIC_RELAXED); }
+                #endif
+                break;
+        }
 
         /* paranoid */ verify( ready_schedule_islocked() );
@@ -770 +782 @@
 
         __cfadbg_print_safe(runtime_core, "Kernel : waking Processor %p\n", this);
+
+        this->idle_wctx.fd = 1;
 
         eventfd_t val;
@@ -779 +793 @@
 
 static void idle_sleep(processor * this, io_future_t & future, iovec & iov) {
+        // Tell everyone we are ready to go do sleep
+        for() {
+                int expected = this->idle_wctx.fd;
+
+                // Someone already told us to wake-up! No time for a nap.
+                if(expected == 1) { return; }
+
+                // Try to mark that we are going to sleep
+                if(__atomic_compare_exchange_n(&this->idle_wctx.fd, &expected, this->idle_fd, false,  __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST) ) {
+                        // Every one agreed, taking a nap
+                        break;
+                }
+        }
+
+
         #if !defined(CFA_WITH_IO_URING_IDLE)
                 #if !defined(__CFA_NO_STATISTICS__)
@@ -825 +854 @@
 
 static bool mark_idle(__cluster_proc_list & this, processor & proc) {
+        #if !defined(__CFA_NO_STATISTICS__)
+                __tls_stats()->ready.sleep.halts++;
+        #endif
+
+        proc.idle_wctx.fd = 0;
+
         /* paranoid */ verify( ! __preemption_enabled() );
         if(!try_lock( this )) return false;
@@ -832 +867 @@
                 insert_first(this.idles, proc);
 
-                __atomic_store_n(&this.fd, proc.idle_fd, __ATOMIC_SEQ_CST);
+                __atomic_store_n(&this.fdw, &proc.idle_wctx, __ATOMIC_SEQ_CST);
         unlock( this );
         /* paranoid */ verify( ! __preemption_enabled() );
@@ -848 +883 @@
 
                 {
-                        int fd = 0;
-                        if(!this.idles`isEmpty) fd = this.idles`first.idle_fd;
-                        __atomic_store_n(&this.fd, fd, __ATOMIC_SEQ_CST);
+                        struct __fd_waitctx * wctx = 0;
+                        if(!this.idles`isEmpty) wctx = &this.idles`first.idle_wctx;
+                        __atomic_store_n(&this.fdw, wctx, __ATOMIC_SEQ_CST);
                 }
 

libcfa/src/concurrency/kernel.hfa

@@ -55 +55 @@
 };
 
+
+struct __fd_waitctx {
+        volatile int fd;
+};
+
 // Wrapper around kernel threads
 struct __attribute__((aligned(128))) processor {
@@ -67 +72 @@
                 unsigned target;
                 unsigned last;
-                unsigned cnt;
-                unsigned long long int cutoff;
+                signed   cpu;
         } rdq;
 
@@ -102 +106 @@
         int idle_fd;
 
+        // Idle waitctx
+        struct __fd_waitctx idle_wctx;
+
         // Termination synchronisation (user semaphore)
         oneshot terminated;
@@ -152 +159 @@
         volatile unsigned long long tv;
         volatile unsigned long long ma;
+};
+
+struct __attribute__((aligned(16))) __cache_id_t {
+        volatile unsigned id;
 };
 
@@ -164 +175 @@
 static inline void ^?{}(__timestamp_t & this) {}
 
+struct __attribute__((aligned(128))) __ready_queue_caches_t;
+void  ?{}(__ready_queue_caches_t & this);
+void ^?{}(__ready_queue_caches_t & this);
+
 //TODO adjust cache size to ARCHITECTURE
-// Structure holding the relaxed ready queue
+// Structure holding the ready queue
 struct __ready_queue_t {
         // Data tracking the actual lanes
@@ -178 +193 @@
                 __timestamp_t * volatile tscs;
 
+                __cache_id_t * volatile caches;
+
                 // Array of stats
                 __help_cnts_t * volatile help;
@@ -198 +215 @@
 
         // FD to use to wake a processor
-        volatile int fd;
+        struct __fd_waitctx * volatile fdw;
 
         // Total number of processors

libcfa/src/concurrency/kernel/fwd.hfa

@@ -79 +79 @@
                         return
                         #if defined(__SIZEOF_INT128__)
-                                __lehmer64( kernelTLS().rand_seed );
+                                lehmer64( kernelTLS().rand_seed );
                         #else
-                                __xorshift64( kernelTLS().rand_seed );
+                                xorshift_13_7_17( kernelTLS().rand_seed );
                         #endif
                 }
 
-                #define M  (1_l64u << 48_l64u)
-                #define A  (25214903917_l64u)
-                #define AI (18446708753438544741_l64u)
-                #define C  (11_l64u)
-                #define D  (16_l64u)
-
                 static inline unsigned __tls_rand_fwd() {
-                        kernelTLS().ready_rng.fwd_seed = (A * kernelTLS().ready_rng.fwd_seed + C) & (M - 1);
-                        return kernelTLS().ready_rng.fwd_seed >> D;
+                        return LCGBI_fwd( kernelTLS().ready_rng.fwd_seed );
                 }
 
                 static inline unsigned __tls_rand_bck() {
-                        unsigned int r = kernelTLS().ready_rng.bck_seed >> D;
-                        kernelTLS().ready_rng.bck_seed = AI * (kernelTLS().ready_rng.bck_seed - C) & (M - 1);
-                        return r;
-                }
-
-                #undef M
-                #undef A
-                #undef AI
-                #undef C
-                #undef D
+                        return LCGBI_bck( kernelTLS().ready_rng.bck_seed );
+                }
 
                 static inline void __tls_rand_advance_bck(void) {
@@ -140 +125 @@
                         }
                 }
-
-                extern uint64_t thread_rand();
 
                 // Semaphore which only supports a single thread

libcfa/src/concurrency/kernel/startup.cfa

@@ -34 +34 @@
 #include "kernel_private.hfa"
 #include "startup.hfa"          // STARTUP_PRIORITY_XXX
+#include "limits.hfa"
 #include "math.hfa"
 
@@ -177 +178 @@
 
 
-
 //=============================================================================================
 // Kernel Setup logic
@@ -515 +515 @@
         this.rdq.its = 0;
         this.rdq.itr = 0;
-        this.rdq.id  = -1u;
-        this.rdq.target = -1u;
-        this.rdq.last = -1u;
-        this.rdq.cutoff = 0ull;
+        this.rdq.id  = MAX;
+        this.rdq.target = MAX;
+        this.rdq.last = MAX;
+        this.rdq.cpu = 0;
+        // this.rdq.cutoff = 0ull;
         do_terminate = false;
         preemption_alarm = 0p;
@@ -536 +537 @@
         }
 
+        this.idle_wctx.fd = 0;
+
+        // I'm assuming these two are reserved for standard input and output
+        // so I'm using them as sentinels with idle_wctx.
+        /* paranoid */ verify( this.idle_fd != 0 );
+        /* paranoid */ verify( this.idle_fd != 1 );
+
         #if !defined(__CFA_NO_STATISTICS__)
                 print_stats = 0;
@@ -589 +597 @@
 // Cluster
 static void ?{}(__cluster_proc_list & this) {
-        this.fd    = 0;
+        this.fdw   = 0p;
         this.idle  = 0;
         this.total = 0;
@@ -686 +694 @@
         uint_fast32_t last_size;
         [this->unique_id, last_size] = ready_mutate_register();
+
+                this->rdq.cpu = __kernel_getcpu();
 
                 this->cltr->procs.total += 1u;

libcfa/src/concurrency/locks.hfa

@@ -29 +29 @@
 #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`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;
-                /* paranoid */ verify( c > MIN );
-                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 () {
-                        /* paranoid */ verify( this.counter < MAX );
-                        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);
-}
 
 //-----------------------------------------------------------------------------
@@ -171 +72 @@
 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 ); }
-
-struct fast_lock {
-        thread$ * volatile owner;
-        ThreadBenaphore sem;
-};
-
-static inline void ?{}(fast_lock & this) __attribute__((deprecated("use linear_backoff_then_block_lock instead")));
-static inline void ?{}(fast_lock & this) { this.owner = 0p; }
-
-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__((deprecated("use linear_backoff_then_block_lock instead"), 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__((deprecated("use linear_backoff_then_block_lock instead"), 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__((deprecated("use linear_backoff_then_block_lock instead"), 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 {

libcfa/src/concurrency/ready_queue.cfa

@@ -20 +20 @@
 
 
-#define USE_RELAXED_FIFO
+// #define USE_RELAXED_FIFO
 // #define USE_WORK_STEALING
 // #define USE_CPU_WORK_STEALING
+#define USE_AWARE_STEALING
 
 #include "bits/defs.hfa"
@@ -29 +30 @@
 
 #include "stdlib.hfa"
+#include "limits.hfa"
 #include "math.hfa"
 
@@ -54 +56 @@
 #endif
 
-#if   defined(USE_CPU_WORK_STEALING)
+#if   defined(USE_AWARE_STEALING)
+        #define READYQ_SHARD_FACTOR 2
+        #define SEQUENTIAL_SHARD 2
+#elif defined(USE_CPU_WORK_STEALING)
         #define READYQ_SHARD_FACTOR 2
 #elif defined(USE_RELAXED_FIFO)
@@ -138 +143 @@
         __kernel_rseq_register();
 
-        __cfadbg_print_safe(ready_queue, "Kernel : Registering proc %p for RW-Lock\n", proc);
         bool * handle = (bool *)&kernelTLS().sched_lock;
 
@@ -174 +178 @@
         }
 
-        __cfadbg_print_safe(ready_queue, "Kernel : Registering proc %p done, id %lu\n", proc, n);
-
         // Return new spot.
         /* paranoid */ verify(n < ready);
@@ -190 +192 @@
 
         __atomic_store_n(cell, 0p, __ATOMIC_RELEASE);
-
-        __cfadbg_print_safe(ready_queue, "Kernel : Unregister proc %p\n", proc);
 
         __kernel_rseq_unregister();
@@ -201 +201 @@
 uint_fast32_t ready_mutate_lock( void ) with(*__scheduler_lock) {
         /* paranoid */ verify( ! __preemption_enabled() );
-        /* paranoid */ verify( ! kernelTLS().sched_lock );
 
         // Step 1 : lock global lock
@@ -207 +206 @@
         //   to simply lock their own lock and enter.
         __atomic_acquire( &write_lock );
+
+        // Make sure we won't deadlock ourself
+        // Checking before acquiring the writer lock isn't safe
+        // because someone else could have locked us.
+        /* paranoid */ verify( ! kernelTLS().sched_lock );
 
         // Step 2 : lock per-proc lock
@@ -244 +248 @@
 
 //=======================================================================
+// caches handling
+
+struct __attribute__((aligned(128))) __ready_queue_caches_t {
+        // Count States:
+        // - 0  : No one is looking after this cache
+        // - 1  : No one is looking after this cache, BUT it's not empty
+        // - 2+ : At least one processor is looking after this cache
+        volatile unsigned count;
+};
+
+void  ?{}(__ready_queue_caches_t & this) { this.count = 0; }
+void ^?{}(__ready_queue_caches_t & this) {}
+
+static inline void depart(__ready_queue_caches_t & cache) {
+        /* paranoid */ verify( cache.count > 1);
+        __atomic_fetch_add(&cache.count, -1, __ATOMIC_SEQ_CST);
+        /* paranoid */ verify( cache.count != 0);
+        /* paranoid */ verify( cache.count < 65536 ); // This verify assumes no cluster will have more than 65000 kernel threads mapped to a single cache, which could be correct but is super weird.
+}
+
+static inline void arrive(__ready_queue_caches_t & cache) {
+        // for() {
+        //      unsigned expected = cache.count;
+        //      unsigned desired  = 0 == expected ? 2 : expected + 1;
+        // }
+}
+
+//=======================================================================
 // Cforall Ready Queue used for scheduling
 //=======================================================================
-unsigned long long moving_average(unsigned long long nval, unsigned long long oval) {
-        const unsigned long long tw = 16;
-        const unsigned long long nw = 4;
-        const unsigned long long ow = tw - nw;
-        return ((nw * nval) + (ow * oval)) / tw;
+unsigned long long moving_average(unsigned long long currtsc, unsigned long long instsc, unsigned long long old_avg) {
+        /* paranoid */ verifyf( currtsc < 45000000000000000, "Suspiciously large current time: %'llu (%llx)\n", currtsc, currtsc );
+        /* paranoid */ verifyf( instsc  < 45000000000000000, "Suspiciously large insert time: %'llu (%llx)\n", instsc, instsc );
+        /* paranoid */ verifyf( old_avg < 15000000000000, "Suspiciously large previous average: %'llu (%llx)\n", old_avg, old_avg );
+
+        const unsigned long long new_val = currtsc > instsc ? currtsc - instsc : 0;
+        const unsigned long long total_weight = 16;
+        const unsigned long long new_weight   = 4;
+        const unsigned long long old_weight = total_weight - new_weight;
+        const unsigned long long ret = ((new_weight * new_val) + (old_weight * old_avg)) / total_weight;
+        return ret;
 }
 
@@ -271 +309 @@
                 }
         #else
-                lanes.data  = 0p;
-                lanes.tscs  = 0p;
-                lanes.help  = 0p;
-                lanes.count = 0;
+                lanes.data   = 0p;
+                lanes.tscs   = 0p;
+                lanes.caches = 0p;
+                lanes.help   = 0p;
+                lanes.count  = 0;
         #endif
 }
@@ -285 +324 @@
         free(lanes.data);
         free(lanes.tscs);
+        free(lanes.caches);
         free(lanes.help);
 }
 
 //-----------------------------------------------------------------------
+#if defined(USE_AWARE_STEALING)
+        __attribute__((hot)) void push(struct cluster * cltr, struct thread$ * thrd, unpark_hint hint) with (cltr->ready_queue) {
+                processor * const proc = kernelTLS().this_processor;
+                const bool external = (!proc) || (cltr != proc->cltr);
+                const bool remote   = hint == UNPARK_REMOTE;
+
+                unsigned i;
+                if( external || remote ) {
+                        // Figure out where thread was last time and make sure it's valid
+                        /* paranoid */ verify(thrd->preferred >= 0);
+                        if(thrd->preferred * READYQ_SHARD_FACTOR < lanes.count) {
+                                /* paranoid */ verify(thrd->preferred * READYQ_SHARD_FACTOR < lanes.count);
+                                unsigned start = thrd->preferred * READYQ_SHARD_FACTOR;
+                                do {
+                                        unsigned r = __tls_rand();
+                                        i = start + (r % READYQ_SHARD_FACTOR);
+                                        /* paranoid */ verify( i < lanes.count );
+                                        // If we can't lock it retry
+                                } while( !__atomic_try_acquire( &lanes.data[i].lock ) );
+                        } else {
+                                do {
+                                        i = __tls_rand() % lanes.count;
+                                } while( !__atomic_try_acquire( &lanes.data[i].lock ) );
+                        }
+                } else {
+                        do {
+                                unsigned r = proc->rdq.its++;
+                                i = proc->rdq.id + (r % READYQ_SHARD_FACTOR);
+                                /* paranoid */ verify( i < lanes.count );
+                                // If we can't lock it retry
+                        } while( !__atomic_try_acquire( &lanes.data[i].lock ) );
+                }
+
+                // Actually push it
+                push(lanes.data[i], thrd);
+
+                // Unlock and return
+                __atomic_unlock( &lanes.data[i].lock );
+
+                #if !defined(__CFA_NO_STATISTICS__)
+                        if(unlikely(external || remote)) __atomic_fetch_add(&cltr->stats->ready.push.extrn.success, 1, __ATOMIC_RELAXED);
+                        else __tls_stats()->ready.push.local.success++;
+                #endif
+        }
+
+        static inline unsigned long long calc_cutoff(const unsigned long long ctsc, const processor * proc, __ready_queue_t & rdq) {
+                unsigned start = proc->rdq.id;
+                unsigned long long max = 0;
+                for(i; READYQ_SHARD_FACTOR) {
+                        unsigned long long ptsc = ts(rdq.lanes.data[start + i]);
+                        if(ptsc != -1ull) {
+                                /* paranoid */ verify( start + i < rdq.lanes.count );
+                                unsigned long long tsc = moving_average(ctsc, ptsc, rdq.lanes.tscs[start + i].ma);
+                                if(tsc > max) max = tsc;
+                        }
+                }
+                return (max + 2 * max) / 2;
+        }
+
+        __attribute__((hot)) struct 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 * const proc = kernelTLS().this_processor;
+                unsigned this = proc->rdq.id;
+                /* paranoid */ verify( this < lanes.count );
+                __cfadbg_print_safe(ready_queue, "Kernel : pop from %u\n", this);
+
+                // Figure out the current cpu and make sure it is valid
+                const int cpu = __kernel_getcpu();
+                /* paranoid */ verify(cpu >= 0);
+                /* paranoid */ verify(cpu < cpu_info.hthrd_count);
+                unsigned this_cache = cpu_info.llc_map[cpu].cache;
+
+                // Super important: don't write the same value over and over again
+                // We want to maximise our chances that his particular values stays in cache
+                if(lanes.caches[this / READYQ_SHARD_FACTOR].id != this_cache)
+                        __atomic_store_n(&lanes.caches[this / READYQ_SHARD_FACTOR].id, this_cache, __ATOMIC_RELAXED);
+
+                const unsigned long long ctsc = rdtscl();
+
+                if(proc->rdq.target == MAX) {
+                        uint64_t chaos = __tls_rand();
+                        unsigned ext = chaos & 0xff;
+                        unsigned other  = (chaos >> 8) % (lanes.count);
+
+                        if(ext < 3 || __atomic_load_n(&lanes.caches[other / READYQ_SHARD_FACTOR].id, __ATOMIC_RELAXED) == this_cache) {
+                                proc->rdq.target = other;
+                        }
+                }
+                else {
+                        const unsigned target = proc->rdq.target;
+                        __cfadbg_print_safe(ready_queue, "Kernel : %u considering helping %u, tcsc %llu\n", this, target, lanes.tscs[target].tv);
+                        /* paranoid */ verify( lanes.tscs[target].tv != MAX );
+                        if(target < lanes.count) {
+                                const unsigned long long cutoff = calc_cutoff(ctsc, proc, cltr->ready_queue);
+                                const unsigned long long age = moving_average(ctsc, lanes.tscs[target].tv, lanes.tscs[target].ma);
+                                __cfadbg_print_safe(ready_queue, "Kernel : Help attempt on %u from %u, age %'llu vs cutoff %'llu, %s\n", target, this, age, cutoff, age > cutoff ? "yes" : "no");
+                                if(age > cutoff) {
+                                        thread$ * t = try_pop(cltr, target __STATS(, __tls_stats()->ready.pop.help));
+                                        if(t) return t;
+                                }
+                        }
+                        proc->rdq.target = MAX;
+                }
+
+                for(READYQ_SHARD_FACTOR) {
+                        unsigned i = this + (proc->rdq.itr++ % READYQ_SHARD_FACTOR);
+                        if(thread$ * t = try_pop(cltr, i __STATS(, __tls_stats()->ready.pop.local))) return t;
+                }
+
+                // All lanes where empty return 0p
+                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) {
+                return search(cltr);
+        }
+#endif
 #if defined(USE_CPU_WORK_STEALING)
         __attribute__((hot)) void push(struct cluster * cltr, struct thread$ * thrd, unpark_hint hint) with (cltr->ready_queue) {
@@ -350 +514 @@
                 /* paranoid */ verify( kernelTLS().this_processor );
 
+                processor * const proc = kernelTLS().this_processor;
                 const int cpu = __kernel_getcpu();
                 /* paranoid */ verify(cpu >= 0);
@@ -360 +525 @@
                 /* paranoid */ verifyf((map.start + map.count) * READYQ_SHARD_FACTOR <= lanes.count, "have %zu lanes but map can go up to %u", lanes.count, (map.start + map.count) * READYQ_SHARD_FACTOR);
 
-                processor * const proc = kernelTLS().this_processor;
                 const int start = map.self * READYQ_SHARD_FACTOR;
                 const unsigned long long ctsc = rdtscl();
 
                 // Did we already have a help target
-                if(proc->rdq.target == -1u) {
+                if(proc->rdq.target == MAX) {
                         unsigned long long max = 0;
                         for(i; READYQ_SHARD_FACTOR) {
-                                unsigned long long tsc = moving_average(ctsc - ts(lanes.data[start + i]), lanes.tscs[start + i].ma);
+                                unsigned long long tsc = moving_average(ctsc, ts(lanes.data[start + i]), lanes.tscs[start + i].ma);
                                 if(tsc > max) max = tsc;
                         }
-                         proc->rdq.cutoff = (max + 2 * max) / 2;
+                        //  proc->rdq.cutoff = (max + 2 * max) / 2;
                         /* paranoid */ verify(lanes.count < 65536); // The following code assumes max 65536 cores.
                         /* paranoid */ verify(map.count < 65536); // The following code assumes max 65536 cores.
@@ -384 +548 @@
                         }
 
-                        /* paranoid */ verify(proc->rdq.target != -1u);
+                        /* paranoid */ verify(proc->rdq.target != MAX);
                 }
                 else {
                         unsigned long long max = 0;
                         for(i; READYQ_SHARD_FACTOR) {
-                                unsigned long long tsc = moving_average(ctsc - ts(lanes.data[start + i]), lanes.tscs[start + i].ma);
+                                unsigned long long tsc = moving_average(ctsc, ts(lanes.data[start + i]), lanes.tscs[start + i].ma);
                                 if(tsc > max) max = tsc;
                         }
@@ -395 +559 @@
                         {
                                 unsigned target = proc->rdq.target;
-                                proc->rdq.target = -1u;
+                                proc->rdq.target = MAX;
                                 lanes.help[target / READYQ_SHARD_FACTOR].tri++;
-                                if(moving_average(ctsc - lanes.tscs[target].tv, lanes.tscs[target].ma) > cutoff) {
+                                if(moving_average(ctsc, lanes.tscs[target].tv, lanes.tscs[target].ma) > cutoff) {
                                         thread$ * t = try_pop(cltr, target __STATS(, __tls_stats()->ready.pop.help));
                                         proc->rdq.last = target;
                                         if(t) return t;
-                                        else proc->rdq.target = -1u;
                                 }
-                                else proc->rdq.target = -1u;
+                                proc->rdq.target = MAX;
                         }
 
                         unsigned last = proc->rdq.last;
-                        if(last != -1u && lanes.tscs[last].tv < cutoff && ts(lanes.data[last]) < cutoff) {
+                        if(last != MAX && moving_average(ctsc, lanes.tscs[last].tv, lanes.tscs[last].ma) > cutoff) {
                                 thread$ * t = try_pop(cltr, last __STATS(, __tls_stats()->ready.pop.help));
                                 if(t) return t;
                         }
                         else {
-                                proc->rdq.last = -1u;
+                                proc->rdq.last = MAX;
                         }
                 }
@@ -428 +591 @@
                 processor * const proc = kernelTLS().this_processor;
                 unsigned last = proc->rdq.last;
-                if(last != -1u) {
+                if(last != MAX) {
                         struct thread$ * t = try_pop(cltr, last __STATS(, __tls_stats()->ready.pop.steal));
                         if(t) return t;
-                        proc->rdq.last = -1u;
+                        proc->rdq.last = MAX;
                 }
 
@@ -560 +723 @@
                 #else
                         unsigned preferred = thrd->preferred;
-                        const bool external = (hint != UNPARK_LOCAL) || (!kernelTLS().this_processor) || preferred == -1u || thrd->curr_cluster != cltr;
+                        const bool external = (hint != UNPARK_LOCAL) || (!kernelTLS().this_processor) || preferred == MAX || thrd->curr_cluster != cltr;
                         /* paranoid */ verifyf(external || preferred < lanes.count, "Invalid preferred queue %u for %u lanes", preferred, lanes.count );
 
@@ -612 +775 @@
                 processor * proc = kernelTLS().this_processor;
 
-                if(proc->rdq.target == -1u) {
+                if(proc->rdq.target == MAX) {
                         unsigned long long min = ts(lanes.data[proc->rdq.id]);
                         for(int i = 0; i < READYQ_SHARD_FACTOR; i++) {
@@ -623 +786 @@
                 else {
                         unsigned target = proc->rdq.target;
-                        proc->rdq.target = -1u;
+                        proc->rdq.target = MAX;
                         const unsigned long long bias = 0; //2_500_000_000;
                         const unsigned long long cutoff = proc->rdq.cutoff > bias ? proc->rdq.cutoff - bias : proc->rdq.cutoff;
@@ -658 +821 @@
 // 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) {
+        /* paranoid */ verify( w < lanes.count );
         __STATS( stats.attempt++; )
 
@@ -681 +845 @@
         // Actually pop the list
         struct thread$ * thrd;
-        #if defined(USE_WORK_STEALING) || defined(USE_CPU_WORK_STEALING)
+        #if defined(USE_AWARE_STEALING) || defined(USE_WORK_STEALING) || defined(USE_CPU_WORK_STEALING)
                 unsigned long long tsc_before = ts(lane);
         #endif
@@ -697 +861 @@
         __STATS( stats.success++; )
 
-        #if defined(USE_WORK_STEALING) || defined(USE_CPU_WORK_STEALING)
-                unsigned long long now = rdtscl();
-                lanes.tscs[w].tv = tsv;
-                lanes.tscs[w].ma = moving_average(now > tsc_before ? now - tsc_before : 0, lanes.tscs[w].ma);
+        #if defined(USE_AWARE_STEALING) || defined(USE_WORK_STEALING) || defined(USE_CPU_WORK_STEALING)
+                if (tsv != MAX) {
+                        unsigned long long now = rdtscl();
+                        unsigned long long pma = __atomic_load_n(&lanes.tscs[w].ma, __ATOMIC_RELAXED);
+                        __atomic_store_n(&lanes.tscs[w].tv, tsv, __ATOMIC_RELAXED);
+                        __atomic_store_n(&lanes.tscs[w].ma, moving_average(now, tsc_before, pma), __ATOMIC_RELAXED);
+                }
         #endif
 
-        #if defined(USE_CPU_WORK_STEALING)
+        #if defined(USE_AWARE_STEALING) || defined(USE_CPU_WORK_STEALING)
                 thrd->preferred = w / READYQ_SHARD_FACTOR;
         #else
@@ -802 +969 @@
                 /* paranoid */ verifyf( it, "Unexpected null iterator, at index %u of %u\n", i, count);
                 it->rdq.id = value;
-                it->rdq.target = -1u;
+                it->rdq.target = MAX;
                 value += READYQ_SHARD_FACTOR;
                 it = &(*it)`next;
@@ -815 +982 @@
 
 static void fix_times( struct cluster * cltr ) with( cltr->ready_queue ) {
-        #if defined(USE_WORK_STEALING)
+        #if defined(USE_AWARE_STEALING) || defined(USE_WORK_STEALING)
                 lanes.tscs = alloc(lanes.count, lanes.tscs`realloc);
                 for(i; lanes.count) {
-                        unsigned long long tsc1 = ts(lanes.data[i]);
-                        unsigned long long tsc2 = rdtscl();
-                        lanes.tscs[i].tv = min(tsc1, tsc2);
+                        lanes.tscs[i].tv = rdtscl();
+                        lanes.tscs[i].ma = 0;
                 }
         #endif
@@ -866 +1032 @@
                         // Update original
                         lanes.count = ncount;
+
+                        lanes.caches = alloc( target, lanes.caches`realloc );
                 }
 
@@ -942 +1110 @@
                                 fix(lanes.data[idx]);
                         }
+
+                        lanes.caches = alloc( target, lanes.caches`realloc );
                 }
 
                 fix_times(cltr);
+
 
                 reassign_cltr_id(cltr);

libcfa/src/concurrency/stats.cfa

@@ -31 +31 @@
                 stats->ready.sleep.halts   = 0;
                 stats->ready.sleep.cancels = 0;
+                stats->ready.sleep.early   = 0;
                 stats->ready.sleep.wakes   = 0;
+                stats->ready.sleep.seen    = 0;
                 stats->ready.sleep.exits   = 0;
 
@@ -91 +93 @@
                 tally_one( &cltr->ready.sleep.halts       , &proc->ready.sleep.halts        );
                 tally_one( &cltr->ready.sleep.cancels     , &proc->ready.sleep.cancels      );
+                tally_one( &cltr->ready.sleep.early       , &proc->ready.sleep.early        );
                 tally_one( &cltr->ready.sleep.wakes       , &proc->ready.sleep.wakes        );
+                tally_one( &cltr->ready.sleep.seen        , &proc->ready.sleep.wakes        );
                 tally_one( &cltr->ready.sleep.exits       , &proc->ready.sleep.exits        );
 
@@ -153 +157 @@
                              | " (" | eng3(ready.pop.search.attempt) | " try)";
 
-                        sstr | "- Idle Slp : " | eng3(ready.sleep.halts) | "halt," | eng3(ready.sleep.cancels) | "cancel," | eng3(ready.sleep.wakes) | "wake," | eng3(ready.sleep.exits) | "exit";
+                        sstr | "- Idle Slp : " | eng3(ready.sleep.halts) | "halt," | eng3(ready.sleep.cancels) | "cancel,"
+                             | eng3(ready.sleep.wakes + ready.sleep.early) | '(' | eng3(ready.sleep.early) | ',' | eng3(ready.sleep.seen) | ')' | " wake(early, seen),"
+                             | eng3(ready.sleep.exits) | "exit";
                         sstr | nl;
                 }

libcfa/src/concurrency/stats.hfa

@@ -69 +69 @@
                         volatile uint64_t halts;
                         volatile uint64_t cancels;
+                        volatile uint64_t early;
                         volatile uint64_t wakes;
+                        volatile uint64_t seen;
                         volatile uint64_t exits;
                 } sleep;

libcfa/src/concurrency/thread.cfa

@@ -10 +10 @@
 // Created On       : Tue Jan 17 12:27:26 2017
 // Last Modified By : Peter A. Buhr
-// Last Modified On : Wed Jan 12 18:46:48 2022
-// Update Count     : 36
+// Last Modified On : Thu Jan 13 20:11:55 2022
+// Update Count     : 42
 //
 
@@ -24 +24 @@
 #define __CFA_INVOKE_PRIVATE__
 #include "invoke.h"
-
-uint64_t thread_rand();
 
 extern uint32_t __global_random_seed;
@@ -174 +172 @@
 }
 
-uint64_t thread_rand() {
-        disable_interrupts();
-        uint64_t ret = __tls_rand();
-        enable_interrupts();
-        return ret;
-}
- 
+//-----------------------------------------------------------------------------
 #define GENERATOR LCG
-
-static inline uint32_t MarsagliaXor( uint32_t & state ) {
-        uint32_t ret = state;
-        state ^= state << 6;
-        state ^= state >> 21;
-        state ^= state << 7;
-        return ret;
-} // MarsagliaXor
-
-static inline uint32_t LCG( uint32_t & state ) {                // linear congruential generator
-        uint32_t ret = state;
-        state = 36969 * (state & 65535) + (state >> 16);        // 36969 is NOT prime! No not change it!
-        return ret;
-} // LCG
 
 void set_seed( uint32_t seed ) {