source: libcfa/src/concurrency/ready_queue.cfa @ c42b8a1

//
// Cforall Version 1.0.0 Copyright (C) 2019 University of Waterloo
//
// The contents of this file are covered under the licence agreement in the
// file "LICENCE" distributed with Cforall.
//
// ready_queue.cfa --
//
// Author           : Thierry Delisle
// Created On       : Mon Nov dd 16:29:18 2019
// Last Modified By :
// Last Modified On :
// Update Count     :
//

#define __cforall_thread__
#define _GNU_SOURCE

// #define __CFA_DEBUG_PRINT_READY_QUEUE__


#define USE_AWARE_STEALING

#include "bits/defs.hfa"
#include "device/cpu.hfa"
#include "kernel_private.hfa"

#include "limits.hfa"

// #include <errno.h>
// #include <unistd.h>

#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

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);

//=======================================================================
// Cforall Ready Queue used for scheduling
//=======================================================================
void ?{}(__ready_queue_t & this) with (this) {
        lanes.data   = 0p;
        lanes.tscs   = 0p;
        lanes.caches = 0p;
        lanes.help   = 0p;
        lanes.count  = 0;
}

void ^?{}(__ready_queue_t & this) with (this) {
        free(lanes.data);
        free(lanes.tscs);
        free(lanes.caches);
        free(lanes.help);
}

//-----------------------------------------------------------------------
__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);
}

//=======================================================================
// 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) {
        /* paranoid */ verify( w < lanes.count );
        __STATS( stats.attempt++; )

        // 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;
        unsigned long long tsc_before = ts(lane);
        unsigned long long tsv;
        [thrd, tsv] = pop(lane);

        /* paranoid */ verify(thrd);
        /* paranoid */ verify(tsv);
        /* paranoid */ verify(lane.lock);

        // Unlock and return
        __atomic_unlock(&lane.lock);

        // Update statistics
        __STATS( stats.success++; )

        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);
        }

        thrd->preferred = w / __readyq_shard_factor;

        // return the popped thread
        return thrd;
}

//-----------------------------------------------------------------------
// 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 );
        unsigned offset = __tls_rand();
        for(i; count) {
                unsigned idx = (offset + i) % count;
                struct thread$ * thrd = try_pop(cltr, idx __STATS(, __tls_stats()->ready.pop.search));
                if(thrd) {
                        return thrd;
                }
        }

        // All lanes where empty return 0p
        return 0p;
}

//-----------------------------------------------------------------------
// get preferred ready for new thread
unsigned ready_queue_new_preferred() {
        unsigned pref = MAX;
        if(struct thread$ * thrd = publicTLS_get( this_thread )) {
                pref = thrd->preferred;
        }

        return pref;
}

//-----------------------------------------------------------------------
// 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));
}