Changeset a2a4566


Ignore:
Timestamp:
Jan 14, 2022, 7:24:45 PM (4 months ago)
Author:
Thierry Delisle <tdelisle@…>
Branches:
enum, forall-pointer-decay, master
Children:
0fc447c
Parents:
c90db2d
Message:

Added new ready-queue that uses per-thread queues but with some cpu awarness

Location:
libcfa/src/concurrency
Files:
2 edited

Legend:

Unmodified
Added
Removed
  • libcfa/src/concurrency/kernel.hfa

    rc90db2d ra2a4566  
    6767                unsigned target;
    6868                unsigned last;
    69                 unsigned cnt;
    70                 unsigned long long int cutoff;
     69                signed   cpu;
     70                // unsigned long long int cutoff;
    7171        } rdq;
    7272
     
    152152        volatile unsigned long long tv;
    153153        volatile unsigned long long ma;
     154};
     155
     156struct __attribute__((aligned(128))) __cache_id_t {
     157        volatile unsigned id;
    154158};
    155159
     
    164168static inline void ^?{}(__timestamp_t & this) {}
    165169
     170struct __attribute__((aligned(128))) __ready_queue_caches_t;
     171void  ?{}(__ready_queue_caches_t & this);
     172void ^?{}(__ready_queue_caches_t & this);
     173
    166174//TODO adjust cache size to ARCHITECTURE
    167 // Structure holding the relaxed ready queue
     175// Structure holding the ready queue
    168176struct __ready_queue_t {
    169177        // Data tracking the actual lanes
     
    177185                // Array of times
    178186                __timestamp_t * volatile tscs;
     187
     188                __cache_id_t * volatile caches;
    179189
    180190                // Array of stats
  • libcfa/src/concurrency/ready_queue.cfa

    rc90db2d ra2a4566  
    2020
    2121
    22 #define USE_RELAXED_FIFO
     22// #define USE_RELAXED_FIFO
    2323// #define USE_WORK_STEALING
    2424// #define USE_CPU_WORK_STEALING
     25#define USE_AWARE_STEALING
    2526
    2627#include "bits/defs.hfa"
     
    2930
    3031#include "stdlib.hfa"
     32#include "limits.hfa"
    3133#include "math.hfa"
    3234
     
    5456#endif
    5557
    56 #if   defined(USE_CPU_WORK_STEALING)
     58#if   defined(USE_AWARE_STEALING)
     59        #define READYQ_SHARD_FACTOR 2
     60        #define SEQUENTIAL_SHARD 2
     61#elif defined(USE_CPU_WORK_STEALING)
    5762        #define READYQ_SHARD_FACTOR 2
    5863#elif defined(USE_RELAXED_FIFO)
     
    138143        __kernel_rseq_register();
    139144
    140         __cfadbg_print_safe(ready_queue, "Kernel : Registering proc %p for RW-Lock\n", proc);
    141145        bool * handle = (bool *)&kernelTLS().sched_lock;
    142146
     
    174178        }
    175179
    176         __cfadbg_print_safe(ready_queue, "Kernel : Registering proc %p done, id %lu\n", proc, n);
    177 
    178180        // Return new spot.
    179181        /* paranoid */ verify(n < ready);
     
    190192
    191193        __atomic_store_n(cell, 0p, __ATOMIC_RELEASE);
    192 
    193         __cfadbg_print_safe(ready_queue, "Kernel : Unregister proc %p\n", proc);
    194194
    195195        __kernel_rseq_unregister();
     
    244244
    245245//=======================================================================
     246// caches handling
     247
     248struct __attribute__((aligned(128))) __ready_queue_caches_t {
     249        // Count States:
     250        // - 0  : No one is looking after this cache
     251        // - 1  : No one is looking after this cache, BUT it's not empty
     252        // - 2+ : At least one processor is looking after this cache
     253        volatile unsigned count;
     254};
     255
     256void  ?{}(__ready_queue_caches_t & this) { this.count = 0; }
     257void ^?{}(__ready_queue_caches_t & this) {}
     258
     259static inline void depart(__ready_queue_caches_t & cache) {
     260        /* paranoid */ verify( cache.count > 1);
     261        __atomic_fetch_add(&cache.count, -1, __ATOMIC_SEQ_CST);
     262        /* paranoid */ verify( cache.count != 0);
     263        /* 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.
     264}
     265
     266static inline void arrive(__ready_queue_caches_t & cache) {
     267        // for() {
     268        //      unsigned expected = cache.count;
     269        //      unsigned desired  = 0 == expected ? 2 : expected + 1;
     270        // }
     271}
     272
     273//=======================================================================
    246274// Cforall Ready Queue used for scheduling
    247275//=======================================================================
    248 unsigned long long moving_average(unsigned long long nval, unsigned long long oval) {
    249         const unsigned long long tw = 16;
    250         const unsigned long long nw = 4;
    251         const unsigned long long ow = tw - nw;
    252         return ((nw * nval) + (ow * oval)) / tw;
     276unsigned long long moving_average(unsigned long long currtsc, unsigned long long instsc, unsigned long long old_avg) {
     277        /* paranoid */ verifyf( currtsc < 45000000000000000, "Suspiciously large current time: %'llu (%llx)\n", currtsc, currtsc );
     278        /* paranoid */ verifyf( instsc  < 45000000000000000, "Suspiciously large insert time: %'llu (%llx)\n", instsc, instsc );
     279        /* paranoid */ verifyf( old_avg < 15000000000000, "Suspiciously large previous average: %'llu (%llx)\n", old_avg, old_avg );
     280
     281        const unsigned long long new_val = currtsc > instsc ? currtsc - instsc : 0;
     282        const unsigned long long total_weight = 16;
     283        const unsigned long long new_weight   = 4;
     284        const unsigned long long old_weight = total_weight - new_weight;
     285        const unsigned long long ret = ((new_weight * new_val) + (old_weight * old_avg)) / total_weight;
     286        return ret;
    253287}
    254288
     
    270304                        lanes.help[idx].tri = 0;
    271305                }
     306
     307                caches = alloc( cpu_info.llc_count );
     308                for( idx; (size_t)cpu_info.llc_count ) {
     309                        (caches[idx]){};
     310                }
    272311        #else
    273                 lanes.data  = 0p;
    274                 lanes.tscs  = 0p;
    275                 lanes.help  = 0p;
    276                 lanes.count = 0;
     312                lanes.data   = 0p;
     313                lanes.tscs   = 0p;
     314                lanes.caches = 0p;
     315                lanes.help   = 0p;
     316                lanes.count  = 0;
    277317        #endif
    278318}
     
    285325        free(lanes.data);
    286326        free(lanes.tscs);
     327        free(lanes.caches);
    287328        free(lanes.help);
    288329}
    289330
    290331//-----------------------------------------------------------------------
     332#if defined(USE_AWARE_STEALING)
     333        __attribute__((hot)) void push(struct cluster * cltr, struct thread$ * thrd, unpark_hint hint) with (cltr->ready_queue) {
     334                processor * const proc = kernelTLS().this_processor;
     335                const bool external = (!proc) || (cltr != proc->cltr);
     336                const bool remote   = hint == UNPARK_REMOTE;
     337
     338                unsigned i;
     339                if( external || remote ) {
     340                        // Figure out where thread was last time and make sure it's valid
     341                        /* paranoid */ verify(thrd->preferred >= 0);
     342                        if(thrd->preferred * READYQ_SHARD_FACTOR < lanes.count) {
     343                                /* paranoid */ verify(thrd->preferred * READYQ_SHARD_FACTOR < lanes.count);
     344                                unsigned start = thrd->preferred * READYQ_SHARD_FACTOR;
     345                                do {
     346                                        unsigned r = __tls_rand();
     347                                        i = start + (r % READYQ_SHARD_FACTOR);
     348                                        /* paranoid */ verify( i < lanes.count );
     349                                        // If we can't lock it retry
     350                                } while( !__atomic_try_acquire( &lanes.data[i].lock ) );
     351                        } else {
     352                                do {
     353                                        i = __tls_rand() % lanes.count;
     354                                } while( !__atomic_try_acquire( &lanes.data[i].lock ) );
     355                        }
     356                } else {
     357                        do {
     358                                unsigned r = proc->rdq.its++;
     359                                i = proc->rdq.id + (r % READYQ_SHARD_FACTOR);
     360                                /* paranoid */ verify( i < lanes.count );
     361                                // If we can't lock it retry
     362                        } while( !__atomic_try_acquire( &lanes.data[i].lock ) );
     363                }
     364
     365                // Actually push it
     366                push(lanes.data[i], thrd);
     367
     368                // Unlock and return
     369                __atomic_unlock( &lanes.data[i].lock );
     370
     371                #if !defined(__CFA_NO_STATISTICS__)
     372                        if(unlikely(external || remote)) __atomic_fetch_add(&cltr->stats->ready.push.extrn.success, 1, __ATOMIC_RELAXED);
     373                        else __tls_stats()->ready.push.local.success++;
     374                #endif
     375        }
     376
     377        static inline unsigned long long calc_cutoff(const unsigned long long ctsc, const processor * proc, __ready_queue_t & rdq) {
     378                unsigned start = proc->rdq.id;
     379                unsigned long long max = 0;
     380                for(i; READYQ_SHARD_FACTOR) {
     381                        unsigned long long ptsc = ts(rdq.lanes.data[start + i]);
     382                        if(ptsc != -1ull) {
     383                                /* paranoid */ verify( start + i < rdq.lanes.count );
     384                                unsigned long long tsc = moving_average(ctsc, ptsc, rdq.lanes.tscs[start + i].ma);
     385                                if(tsc > max) max = tsc;
     386                        }
     387                }
     388                return (max + 2 * max) / 2;
     389        }
     390
     391        __attribute__((hot)) struct thread$ * pop_fast(struct cluster * cltr) with (cltr->ready_queue) {
     392                /* paranoid */ verify( lanes.count > 0 );
     393                /* paranoid */ verify( kernelTLS().this_processor );
     394                /* paranoid */ verify( kernelTLS().this_processor->rdq.id < lanes.count );
     395
     396                processor * const proc = kernelTLS().this_processor;
     397                unsigned this = proc->rdq.id;
     398                /* paranoid */ verify( this < lanes.count );
     399                __cfadbg_print_safe(ready_queue, "Kernel : pop from %u\n", this);
     400
     401                // Figure out the current cpu and make sure it is valid
     402                const int cpu = __kernel_getcpu();
     403                /* paranoid */ verify(cpu >= 0);
     404                /* paranoid */ verify(cpu < cpu_info.hthrd_count);
     405                unsigned this_cache = cpu_info.llc_map[cpu].cache;
     406                __atomic_store_n(&lanes.caches[this / READYQ_SHARD_FACTOR].id, this_cache, __ATOMIC_RELAXED);
     407
     408                const unsigned long long ctsc = rdtscl();
     409
     410                if(proc->rdq.target == MAX) {
     411                        uint64_t chaos = __tls_rand();
     412                        unsigned ext = chaos & 0xff;
     413                        unsigned other  = (chaos >> 8) % (lanes.count);
     414
     415                        if(ext < 3 || __atomic_load_n(&lanes.caches[other / READYQ_SHARD_FACTOR].id, __ATOMIC_RELAXED) == this_cache) {
     416                                proc->rdq.target = other;
     417                        }
     418                }
     419                else {
     420                        const unsigned target = proc->rdq.target;
     421                        __cfadbg_print_safe(ready_queue, "Kernel : %u considering helping %u, tcsc %llu\n", this, target, lanes.tscs[target].tv);
     422                        /* paranoid */ verify( lanes.tscs[target].tv != MAX );
     423                        if(target < lanes.count) {
     424                                const unsigned long long cutoff = calc_cutoff(ctsc, proc, cltr->ready_queue);
     425                                const unsigned long long age = moving_average(ctsc, lanes.tscs[target].tv, lanes.tscs[target].ma);
     426                                __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");
     427                                if(age > cutoff) {
     428                                        thread$ * t = try_pop(cltr, target __STATS(, __tls_stats()->ready.pop.help));
     429                                        if(t) return t;
     430                                }
     431                        }
     432                        proc->rdq.target = MAX;
     433                }
     434
     435                for(READYQ_SHARD_FACTOR) {
     436                        unsigned i = this + (proc->rdq.itr++ % READYQ_SHARD_FACTOR);
     437                        if(thread$ * t = try_pop(cltr, i __STATS(, __tls_stats()->ready.pop.local))) return t;
     438                }
     439
     440                // All lanes where empty return 0p
     441                return 0p;
     442
     443        }
     444        __attribute__((hot)) struct thread$ * pop_slow(struct cluster * cltr) with (cltr->ready_queue) {
     445                unsigned i = __tls_rand() % lanes.count;
     446                return try_pop(cltr, i __STATS(, __tls_stats()->ready.pop.steal));
     447        }
     448        __attribute__((hot)) struct thread$ * pop_search(struct cluster * cltr) {
     449                return search(cltr);
     450        }
     451#endif
    291452#if defined(USE_CPU_WORK_STEALING)
    292453        __attribute__((hot)) void push(struct cluster * cltr, struct thread$ * thrd, unpark_hint hint) with (cltr->ready_queue) {
     
    345506        }
    346507
     508        static inline int pop_getcpu(processor * proc, __ready_queue_caches_t * caches) {
     509                const int prv = proc->rdq.cpu;
     510                const int cpu = __kernel_getcpu();
     511                if( prv != proc->rdq.cpu ) {
     512                        unsigned pidx = cpu_info.llc_map[prv].cache;
     513                        /* paranoid */ verify(pidx < cpu_info.llc_count);
     514
     515                        unsigned nidx = cpu_info.llc_map[cpu].cache;
     516                        /* paranoid */ verify(pidx < cpu_info.llc_count);
     517
     518                        depart(caches[pidx]);
     519                        arrive(caches[nidx]);
     520
     521                        __STATS( /* cpu migs++ */ )
     522                }
     523                return proc->rdq.cpu = cpu;
     524        }
     525
    347526        // Pop from the ready queue from a given cluster
    348527        __attribute__((hot)) thread$ * pop_fast(struct cluster * cltr) with (cltr->ready_queue) {
     
    350529                /* paranoid */ verify( kernelTLS().this_processor );
    351530
    352                 const int cpu = __kernel_getcpu();
     531                processor * const proc = kernelTLS().this_processor;
     532                const int cpu = pop_getcpu( proc, caches );
     533                // const int cpu = __kernel_getcpu();
    353534                /* paranoid */ verify(cpu >= 0);
    354535                /* paranoid */ verify(cpu < cpu_info.hthrd_count);
     
    360541                /* 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);
    361542
    362                 processor * const proc = kernelTLS().this_processor;
    363543                const int start = map.self * READYQ_SHARD_FACTOR;
    364544                const unsigned long long ctsc = rdtscl();
    365545
    366546                // Did we already have a help target
    367                 if(proc->rdq.target == -1u) {
     547                if(proc->rdq.target == MAX) {
    368548                        unsigned long long max = 0;
    369549                        for(i; READYQ_SHARD_FACTOR) {
     
    371551                                if(tsc > max) max = tsc;
    372552                        }
    373                         proc->rdq.cutoff = (max + 2 * max) / 2;
     553                        // proc->rdq.cutoff = (max + 2 * max) / 2;
    374554                        /* paranoid */ verify(lanes.count < 65536); // The following code assumes max 65536 cores.
    375555                        /* paranoid */ verify(map.count < 65536); // The following code assumes max 65536 cores.
     
    384564                        }
    385565
    386                         /* paranoid */ verify(proc->rdq.target != -1u);
     566                        /* paranoid */ verify(proc->rdq.target != MAX);
    387567                }
    388568                else {
     
    395575                        {
    396576                                unsigned target = proc->rdq.target;
    397                                 proc->rdq.target = -1u;
     577                                proc->rdq.target = MAX;
    398578                                lanes.help[target / READYQ_SHARD_FACTOR].tri++;
    399579                                if(moving_average(ctsc - lanes.tscs[target].tv, lanes.tscs[target].ma) > cutoff) {
     580                                        __STATS( __tls_stats()->ready.pop.helped[target]++; )
    400581                                        thread$ * t = try_pop(cltr, target __STATS(, __tls_stats()->ready.pop.help));
    401582                                        proc->rdq.last = target;
    402583                                        if(t) return t;
    403                                         else proc->rdq.target = -1u;
    404584                                }
    405                                 else proc->rdq.target = -1u;
     585                                proc->rdq.target = MAX;
    406586                        }
    407587
    408588                        unsigned last = proc->rdq.last;
    409                         if(last != -1u && lanes.tscs[last].tv < cutoff && ts(lanes.data[last]) < cutoff) {
     589                        if(last != MAX && moving_average(ctsc - lanes.tscs[last].tv, lanes.tscs[last].ma) > cutoff) {
     590                                __STATS( __tls_stats()->ready.pop.helped[last]++; )
    410591                                thread$ * t = try_pop(cltr, last __STATS(, __tls_stats()->ready.pop.help));
    411592                                if(t) return t;
    412593                        }
    413594                        else {
    414                                 proc->rdq.last = -1u;
     595                                proc->rdq.last = MAX;
    415596                        }
    416597                }
     
    428609                processor * const proc = kernelTLS().this_processor;
    429610                unsigned last = proc->rdq.last;
    430                 if(last != -1u) {
     611                if(last != MAX) {
    431612                        struct thread$ * t = try_pop(cltr, last __STATS(, __tls_stats()->ready.pop.steal));
    432613                        if(t) return t;
    433                         proc->rdq.last = -1u;
     614                        proc->rdq.last = MAX;
    434615                }
    435616
     
    560741                #else
    561742                        unsigned preferred = thrd->preferred;
    562                         const bool external = (hint != UNPARK_LOCAL) || (!kernelTLS().this_processor) || preferred == -1u || thrd->curr_cluster != cltr;
     743                        const bool external = (hint != UNPARK_LOCAL) || (!kernelTLS().this_processor) || preferred == MAX || thrd->curr_cluster != cltr;
    563744                        /* paranoid */ verifyf(external || preferred < lanes.count, "Invalid preferred queue %u for %u lanes", preferred, lanes.count );
    564745
     
    612793                processor * proc = kernelTLS().this_processor;
    613794
    614                 if(proc->rdq.target == -1u) {
     795                if(proc->rdq.target == MAX) {
    615796                        unsigned long long min = ts(lanes.data[proc->rdq.id]);
    616797                        for(int i = 0; i < READYQ_SHARD_FACTOR; i++) {
     
    623804                else {
    624805                        unsigned target = proc->rdq.target;
    625                         proc->rdq.target = -1u;
     806                        proc->rdq.target = MAX;
    626807                        const unsigned long long bias = 0; //2_500_000_000;
    627808                        const unsigned long long cutoff = proc->rdq.cutoff > bias ? proc->rdq.cutoff - bias : proc->rdq.cutoff;
     
    658839// try to pop from a lane given by index w
    659840static inline struct thread$ * try_pop(struct cluster * cltr, unsigned w __STATS(, __stats_readyQ_pop_t & stats)) with (cltr->ready_queue) {
     841        /* paranoid */ verify( w < lanes.count );
    660842        __STATS( stats.attempt++; )
    661843
     
    681863        // Actually pop the list
    682864        struct thread$ * thrd;
    683         #if defined(USE_WORK_STEALING) || defined(USE_CPU_WORK_STEALING)
     865        #if defined(USE_AWARE_STEALING) || defined(USE_WORK_STEALING) || defined(USE_CPU_WORK_STEALING)
    684866                unsigned long long tsc_before = ts(lane);
    685867        #endif
     
    697879        __STATS( stats.success++; )
    698880
    699         #if defined(USE_WORK_STEALING) || defined(USE_CPU_WORK_STEALING)
    700                 unsigned long long now = rdtscl();
    701                 lanes.tscs[w].tv = tsv;
    702                 lanes.tscs[w].ma = moving_average(now > tsc_before ? now - tsc_before : 0, lanes.tscs[w].ma);
     881        #if defined(USE_AWARE_STEALING) || defined(USE_WORK_STEALING) || defined(USE_CPU_WORK_STEALING)
     882                if (tsv != MAX) {
     883                        unsigned long long now = rdtscl();
     884                        unsigned long long pma = __atomic_load_n(&lanes.tscs[w].ma, __ATOMIC_RELAXED);
     885                        __atomic_store_n(&lanes.tscs[w].tv, tsv, __ATOMIC_RELAXED);
     886                        __atomic_store_n(&lanes.tscs[w].ma, moving_average(now, tsc_before, pma), __ATOMIC_RELAXED);
     887                }
    703888        #endif
    704889
    705         #if defined(USE_CPU_WORK_STEALING)
     890        #if defined(USE_AWARE_STEALING) || defined(USE_CPU_WORK_STEALING)
    706891                thrd->preferred = w / READYQ_SHARD_FACTOR;
    707892        #else
     
    802987                /* paranoid */ verifyf( it, "Unexpected null iterator, at index %u of %u\n", i, count);
    803988                it->rdq.id = value;
    804                 it->rdq.target = -1u;
     989                it->rdq.target = MAX;
    805990                value += READYQ_SHARD_FACTOR;
    806991                it = &(*it)`next;
     
    8151000
    8161001static void fix_times( struct cluster * cltr ) with( cltr->ready_queue ) {
    817         #if defined(USE_WORK_STEALING)
     1002        #if defined(USE_AWARE_STEALING) || defined(USE_WORK_STEALING)
    8181003                lanes.tscs = alloc(lanes.count, lanes.tscs`realloc);
    8191004                for(i; lanes.count) {
    820                         unsigned long long tsc1 = ts(lanes.data[i]);
    821                         unsigned long long tsc2 = rdtscl();
    822                         lanes.tscs[i].tv = min(tsc1, tsc2);
     1005                        lanes.tscs[i].tv = rdtscl();
     1006                        lanes.tscs[i].ma = 0;
    8231007                }
    8241008        #endif
     
    8661050                        // Update original
    8671051                        lanes.count = ncount;
     1052
     1053                        lanes.caches = alloc( target, lanes.caches`realloc );
    8681054                }
    8691055
     
    9421128                                fix(lanes.data[idx]);
    9431129                        }
     1130
     1131                        lanes.caches = alloc( target, lanes.caches`realloc );
    9441132                }
    9451133
    9461134                fix_times(cltr);
     1135
    9471136
    9481137                reassign_cltr_id(cltr);
Note: See TracChangeset for help on using the changeset viewer.