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

ADTarm-ehast-experimentalenumforall-pointer-decayjacob/cs343-translationnew-astnew-ast-unique-exprpthread-emulationqualifiedEnum
Last change on this file since db62eef was 343d10e, checked in by Thierry Delisle <tdelisle@…>, 4 years ago

Removed code broken in new-ast out of libcfa

  • Property mode set to 100644
File size: 17.7 KB
Line 
1//
2// Cforall Version 1.0.0 Copyright (C) 2019 University of Waterloo
3//
4// The contents of this file are covered under the licence agreement in the
5// file "LICENCE" distributed with Cforall.
6//
7// ready_queue.cfa --
8//
9// Author           : Thierry Delisle
10// Created On       : Mon Nov dd 16:29:18 2019
11// Last Modified By :
12// Last Modified On :
13// Update Count     :
14//
15
16#define __cforall_thread__
17// #define __CFA_DEBUG_PRINT_READY_QUEUE__
18
19// #define USE_SNZI
20
21#include "bits/defs.hfa"
22#include "kernel_private.hfa"
23
24#define _GNU_SOURCE
25#include "stdlib.hfa"
26#include "math.hfa"
27
28#include <unistd.h>
29
30#include "snzi.hfa"
31#include "ready_subqueue.hfa"
32
33static const size_t cache_line_size = 64;
34
35// No overriden function, no environment variable, no define
36// fall back to a magic number
37#ifndef __CFA_MAX_PROCESSORS__
38        #define __CFA_MAX_PROCESSORS__ 1024
39#endif
40
41#define BIAS 16
42
43// returns the maximum number of processors the RWLock support
44__attribute__((weak)) unsigned __max_processors() {
45        const char * max_cores_s = getenv("CFA_MAX_PROCESSORS");
46        if(!max_cores_s) {
47                __cfadbg_print_nolock(ready_queue, "No CFA_MAX_PROCESSORS in ENV\n");
48                return __CFA_MAX_PROCESSORS__;
49        }
50
51        char * endptr = 0p;
52        long int max_cores_l = strtol(max_cores_s, &endptr, 10);
53        if(max_cores_l < 1 || max_cores_l > 65535) {
54                __cfadbg_print_nolock(ready_queue, "CFA_MAX_PROCESSORS out of range : %ld\n", max_cores_l);
55                return __CFA_MAX_PROCESSORS__;
56        }
57        if('\0' != *endptr) {
58                __cfadbg_print_nolock(ready_queue, "CFA_MAX_PROCESSORS not a decimal number : %s\n", max_cores_s);
59                return __CFA_MAX_PROCESSORS__;
60        }
61
62        return max_cores_l;
63}
64
65//=======================================================================
66// Cluster wide reader-writer lock
67//=======================================================================
68void  ?{}(__scheduler_RWLock_t & this) {
69        this.max   = __max_processors();
70        this.alloc = 0;
71        this.ready = 0;
72        this.lock  = false;
73        this.data  = alloc(this.max);
74
75        /*paranoid*/ verify( 0 == (((uintptr_t)(this.data    )) % 64) );
76        /*paranoid*/ verify( 0 == (((uintptr_t)(this.data + 1)) % 64) );
77        /*paranoid*/ verify(__atomic_is_lock_free(sizeof(this.alloc), &this.alloc));
78        /*paranoid*/ verify(__atomic_is_lock_free(sizeof(this.ready), &this.ready));
79
80}
81void ^?{}(__scheduler_RWLock_t & this) {
82        free(this.data);
83}
84
85void ?{}( __scheduler_lock_id_t & this, __processor_id_t * proc ) {
86        this.handle = proc;
87        this.lock   = false;
88        #ifdef __CFA_WITH_VERIFY__
89                this.owned  = false;
90        #endif
91}
92
93//=======================================================================
94// Lock-Free registering/unregistering of threads
95unsigned doregister( struct __processor_id_t * proc ) with(*__scheduler_lock) {
96        __cfadbg_print_safe(ready_queue, "Kernel : Registering proc %p for RW-Lock\n", proc);
97
98        // Step - 1 : check if there is already space in the data
99        uint_fast32_t s = ready;
100
101        // Check among all the ready
102        for(uint_fast32_t i = 0; i < s; i++) {
103                __processor_id_t * null = 0p; // Re-write every loop since compare thrashes it
104                if( __atomic_load_n(&data[i].handle, (int)__ATOMIC_RELAXED) == null
105                        && __atomic_compare_exchange_n( &data[i].handle, &null, proc, false, __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST)) {
106                        /*paranoid*/ verify(i < ready);
107                        /*paranoid*/ verify(0 == (__alignof__(data[i]) % cache_line_size));
108                        /*paranoid*/ verify((((uintptr_t)&data[i]) % cache_line_size) == 0);
109                        return i;
110                }
111        }
112
113        if(max <= alloc) abort("Trying to create more than %ud processors", __scheduler_lock->max);
114
115        // Step - 2 : F&A to get a new spot in the array.
116        uint_fast32_t n = __atomic_fetch_add(&alloc, 1, __ATOMIC_SEQ_CST);
117        if(max <= n) abort("Trying to create more than %ud processors", __scheduler_lock->max);
118
119        // Step - 3 : Mark space as used and then publish it.
120        __scheduler_lock_id_t * storage = (__scheduler_lock_id_t *)&data[n];
121        (*storage){ proc };
122        while(true) {
123                unsigned copy = n;
124                if( __atomic_load_n(&ready, __ATOMIC_RELAXED) == n
125                        && __atomic_compare_exchange_n(&ready, &copy, n + 1, true, __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST))
126                        break;
127                asm volatile("pause");
128        }
129
130        __cfadbg_print_safe(ready_queue, "Kernel : Registering proc %p done, id %lu\n", proc, n);
131
132        // Return new spot.
133        /*paranoid*/ verify(n < ready);
134        /*paranoid*/ verify(__alignof__(data[n]) == (2 * cache_line_size));
135        /*paranoid*/ verify((((uintptr_t)&data[n]) % cache_line_size) == 0);
136        return n;
137}
138
139void unregister( struct __processor_id_t * proc ) with(*__scheduler_lock) {
140        unsigned id = proc->id;
141        /*paranoid*/ verify(id < ready);
142        /*paranoid*/ verify(proc == __atomic_load_n(&data[id].handle, __ATOMIC_RELAXED));
143        __atomic_store_n(&data[id].handle, 0p, __ATOMIC_RELEASE);
144
145        __cfadbg_print_safe(ready_queue, "Kernel : Unregister proc %p\n", proc);
146}
147
148//-----------------------------------------------------------------------
149// Writer side : acquire when changing the ready queue, e.g. adding more
150//  queues or removing them.
151uint_fast32_t ready_mutate_lock( void ) with(*__scheduler_lock) {
152        // Step 1 : lock global lock
153        // It is needed to avoid processors that register mid Critical-Section
154        //   to simply lock their own lock and enter.
155        __atomic_acquire( &lock );
156
157        // Step 2 : lock per-proc lock
158        // Processors that are currently being registered aren't counted
159        //   but can't be in read_lock or in the critical section.
160        // All other processors are counted
161        uint_fast32_t s = ready;
162        for(uint_fast32_t i = 0; i < s; i++) {
163                __atomic_acquire( &data[i].lock );
164        }
165
166        return s;
167}
168
169void ready_mutate_unlock( uint_fast32_t last_s ) with(*__scheduler_lock) {
170        // Step 1 : release local locks
171        // This must be done while the global lock is held to avoid
172        //   threads that where created mid critical section
173        //   to race to lock their local locks and have the writer
174        //   immidiately unlock them
175        // Alternative solution : return s in write_lock and pass it to write_unlock
176        for(uint_fast32_t i = 0; i < last_s; i++) {
177                verify(data[i].lock);
178                __atomic_store_n(&data[i].lock, (bool)false, __ATOMIC_RELEASE);
179        }
180
181        // Step 2 : release global lock
182        /*paranoid*/ assert(true == lock);
183        __atomic_store_n(&lock, (bool)false, __ATOMIC_RELEASE);
184}
185
186//=======================================================================
187// Cforall Reqdy Queue used for scheduling
188//=======================================================================
189void ?{}(__ready_queue_t & this) with (this) {
190        lanes.data  = 0p;
191        lanes.count = 0;
192}
193
194void ^?{}(__ready_queue_t & this) with (this) {
195        verify( 1 == lanes.count );
196        #ifdef USE_SNZI
197                verify( !query( snzi ) );
198        #endif
199        free(lanes.data);
200}
201
202//-----------------------------------------------------------------------
203__attribute__((hot)) bool query(struct cluster * cltr) {
204        #ifdef USE_SNZI
205                return query(cltr->ready_queue.snzi);
206        #endif
207        return true;
208}
209
210//-----------------------------------------------------------------------
211__attribute__((hot)) bool push(struct cluster * cltr, struct $thread * thrd) with (cltr->ready_queue) {
212        __cfadbg_print_safe(ready_queue, "Kernel : Pushing %p on cluster %p\n", thrd, cltr);
213
214        // write timestamp
215        thrd->link.ts = rdtscl();
216
217        #if defined(BIAS) && !defined(__CFA_NO_STATISTICS__)
218                bool local = false;
219                int preferred =
220                        //*
221                        kernelTLS.this_processor ? kernelTLS.this_processor->id * 4 : -1;
222                        /*/
223                        thrd->link.preferred * 4;
224                        //*/
225
226
227        #endif
228
229        // Try to pick a lane and lock it
230        unsigned i;
231        do {
232                // Pick the index of a lane
233                #if defined(BIAS)
234                        unsigned r = __tls_rand();
235                        unsigned rlow  = r % BIAS;
236                        unsigned rhigh = r / BIAS;
237                        if((0 != rlow) && preferred >= 0) {
238                                // (BIAS - 1) out of BIAS chances
239                                // Use perferred queues
240                                i = preferred + (rhigh % 4);
241
242                                #if !defined(__CFA_NO_STATISTICS__)
243                                        local = true;
244                                        __tls_stats()->ready.pick.push.local++;
245                                #endif
246                        }
247                        else {
248                                // 1 out of BIAS chances
249                                // Use all queues
250                                i = rhigh;
251                                local = false;
252                        }
253                #else
254                        i = __tls_rand();
255                #endif
256
257                i %= __atomic_load_n( &lanes.count, __ATOMIC_RELAXED );
258
259                #if !defined(__CFA_NO_STATISTICS__)
260                        __tls_stats()->ready.pick.push.attempt++;
261                #endif
262
263                // If we can't lock it retry
264        } while( !__atomic_try_acquire( &lanes.data[i].lock ) );
265
266        bool first = false;
267
268        // Actually push it
269        bool lane_first = push(lanes.data[i], thrd);
270
271        #ifdef USE_SNZI
272                // If this lane used to be empty we need to do more
273                if(lane_first) {
274                        // Check if the entire queue used to be empty
275                        first = !query(snzi);
276
277                        // Update the snzi
278                        arrive( snzi, i );
279                }
280        #endif
281
282        // Unlock and return
283        __atomic_unlock( &lanes.data[i].lock );
284
285        __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);
286
287        // Update statistics
288        #if !defined(__CFA_NO_STATISTICS__)
289                #if defined(BIAS)
290                        if( local ) __tls_stats()->ready.pick.push.lsuccess++;
291                #endif
292                __tls_stats()->ready.pick.push.success++;
293        #endif
294
295        // return whether or not the list was empty before this push
296        return first;
297}
298
299static struct $thread * try_pop(struct cluster * cltr, unsigned i, unsigned j);
300static struct $thread * try_pop(struct cluster * cltr, unsigned i);
301
302// Pop from the ready queue from a given cluster
303__attribute__((hot)) $thread * pop(struct cluster * cltr) with (cltr->ready_queue) {
304        /* paranoid */ verify( lanes.count > 0 );
305        unsigned count = __atomic_load_n( &lanes.count, __ATOMIC_RELAXED );
306        #if defined(BIAS)
307                // Don't bother trying locally too much
308                int local_tries = 8;
309        #endif
310
311        // As long as the list is not empty, try finding a lane that isn't empty and pop from it
312        #ifdef USE_SNZI
313                while( query(snzi) ) {
314        #else
315                for(25) {
316        #endif
317                // Pick two lists at random
318                unsigned i,j;
319                #if defined(BIAS)
320                        #if !defined(__CFA_NO_STATISTICS__)
321                                bool local = false;
322                        #endif
323                        uint64_t r = __tls_rand();
324                        unsigned rlow  = r % BIAS;
325                        uint64_t rhigh = r / BIAS;
326                        if(local_tries && 0 != rlow) {
327                                // (BIAS - 1) out of BIAS chances
328                                // Use perferred queues
329                                unsigned pid = kernelTLS.this_processor->id * 4;
330                                i = pid + (rhigh % 4);
331                                j = pid + ((rhigh >> 32ull) % 4);
332
333                                // count the tries
334                                local_tries--;
335
336                                #if !defined(__CFA_NO_STATISTICS__)
337                                        local = true;
338                                        __tls_stats()->ready.pick.pop.local++;
339                                #endif
340                        }
341                        else {
342                                // 1 out of BIAS chances
343                                // Use all queues
344                                i = rhigh;
345                                j = rhigh >> 32ull;
346                        }
347                #else
348                        i = __tls_rand();
349                        j = __tls_rand();
350                #endif
351
352                i %= count;
353                j %= count;
354
355                // try popping from the 2 picked lists
356                struct $thread * thrd = try_pop(cltr, i, j);
357                if(thrd) {
358                        #if defined(BIAS) && !defined(__CFA_NO_STATISTICS__)
359                                if( local ) __tls_stats()->ready.pick.pop.lsuccess++;
360                        #endif
361                        return thrd;
362                }
363        }
364
365        // All lanes where empty return 0p
366        return 0p;
367}
368
369__attribute__((hot)) struct $thread * pop_slow(struct cluster * cltr) with (cltr->ready_queue) {
370        /* paranoid */ verify( lanes.count > 0 );
371        unsigned count = __atomic_load_n( &lanes.count, __ATOMIC_RELAXED );
372        unsigned offset = __tls_rand();
373        for(i; count) {
374                unsigned idx = (offset + i) % count;
375                struct $thread * thrd = try_pop(cltr, idx);
376                if(thrd) {
377                        return thrd;
378                }
379        }
380
381        // All lanes where empty return 0p
382        return 0p;
383}
384
385
386//-----------------------------------------------------------------------
387// Given 2 indexes, pick the list with the oldest push an try to pop from it
388static inline struct $thread * try_pop(struct cluster * cltr, unsigned i, unsigned j) with (cltr->ready_queue) {
389        #if !defined(__CFA_NO_STATISTICS__)
390                __tls_stats()->ready.pick.pop.attempt++;
391        #endif
392
393        // Pick the bet list
394        int w = i;
395        if( __builtin_expect(!is_empty(lanes.data[j]), true) ) {
396                w = (ts(lanes.data[i]) < ts(lanes.data[j])) ? i : j;
397        }
398
399        return try_pop(cltr, w);
400}
401
402static inline struct $thread * try_pop(struct cluster * cltr, unsigned w) with (cltr->ready_queue) {
403        // Get relevant elements locally
404        __intrusive_lane_t & lane = lanes.data[w];
405
406        // If list looks empty retry
407        if( is_empty(lane) ) return 0p;
408
409        // If we can't get the lock retry
410        if( !__atomic_try_acquire(&lane.lock) ) return 0p;
411
412
413        // If list is empty, unlock and retry
414        if( is_empty(lane) ) {
415                __atomic_unlock(&lane.lock);
416                return 0p;
417        }
418
419        // Actually pop the list
420        struct $thread * thrd;
421        thrd = pop(lane);
422
423        /* paranoid */ verify(thrd);
424        /* paranoid */ verify(lane.lock);
425
426        #ifdef USE_SNZI
427                // If this was the last element in the lane
428                if(emptied) {
429                        depart( snzi, w );
430                }
431        #endif
432
433        // Unlock and return
434        __atomic_unlock(&lane.lock);
435
436        // Update statistics
437        #if !defined(__CFA_NO_STATISTICS__)
438                __tls_stats()->ready.pick.pop.success++;
439        #endif
440
441        // Update the thread bias
442        thrd->link.preferred = w / 4;
443
444        // return the popped thread
445        return thrd;
446}
447//-----------------------------------------------------------------------
448
449bool remove_head(struct cluster * cltr, struct $thread * thrd) with (cltr->ready_queue) {
450        for(i; lanes.count) {
451                __intrusive_lane_t & lane = lanes.data[i];
452
453                bool removed = false;
454
455                __atomic_acquire(&lane.lock);
456                        if(head(lane)->link.next == thrd) {
457                                $thread * pthrd;
458                                pthrd = pop(lane);
459
460                                /* paranoid */ verify( pthrd == thrd );
461
462                                removed = true;
463                                #ifdef USE_SNZI
464                                        if(emptied) {
465                                                depart( snzi, i );
466                                        }
467                                #endif
468                        }
469                __atomic_unlock(&lane.lock);
470
471                if( removed ) return true;
472        }
473        return false;
474}
475
476//-----------------------------------------------------------------------
477
478static void check( __ready_queue_t & q ) with (q) {
479        #if defined(__CFA_WITH_VERIFY__)
480                {
481                        for( idx ; lanes.count ) {
482                                __intrusive_lane_t & sl = lanes.data[idx];
483                                assert(!lanes.data[idx].lock);
484
485                                assert(head(sl)->link.prev == 0p );
486                                assert(head(sl)->link.next->link.prev == head(sl) );
487                                assert(tail(sl)->link.next == 0p );
488                                assert(tail(sl)->link.prev->link.next == tail(sl) );
489
490                                if(sl.before.link.ts == 0l) {
491                                        assert(tail(sl)->link.prev == head(sl));
492                                        assert(head(sl)->link.next == tail(sl));
493                                } else {
494                                        assert(tail(sl)->link.prev != head(sl));
495                                        assert(head(sl)->link.next != tail(sl));
496                                }
497                        }
498                }
499        #endif
500}
501
502// Call this function of the intrusive list was moved using memcpy
503// fixes the list so that the pointers back to anchors aren't left dangling
504static inline void fix(__intrusive_lane_t & ll) {
505        // if the list is not empty then follow he pointer and fix its reverse
506        if(!is_empty(ll)) {
507                head(ll)->link.next->link.prev = head(ll);
508                tail(ll)->link.prev->link.next = tail(ll);
509        }
510        // Otherwise just reset the list
511        else {
512                verify(tail(ll)->link.next == 0p);
513                tail(ll)->link.prev = head(ll);
514                head(ll)->link.next = tail(ll);
515                verify(head(ll)->link.prev == 0p);
516        }
517}
518
519// Grow the ready queue
520void ready_queue_grow  (struct cluster * cltr, int target) {
521        /* paranoid */ verify( ready_mutate_islocked() );
522        __cfadbg_print_safe(ready_queue, "Kernel : Growing ready queue\n");
523
524        // Make sure that everything is consistent
525        /* paranoid */ check( cltr->ready_queue );
526
527        // grow the ready queue
528        with( cltr->ready_queue ) {
529                #ifdef USE_SNZI
530                        ^(snzi){};
531                #endif
532
533                // Find new count
534                // Make sure we always have atleast 1 list
535                size_t ncount = target >= 2 ? target * 4: 1;
536
537                // Allocate new array (uses realloc and memcpies the data)
538                lanes.data = alloc(lanes.data, ncount);
539
540                // Fix the moved data
541                for( idx; (size_t)lanes.count ) {
542                        fix(lanes.data[idx]);
543                }
544
545                // Construct new data
546                for( idx; (size_t)lanes.count ~ ncount) {
547                        (lanes.data[idx]){};
548                }
549
550                // Update original
551                lanes.count = ncount;
552
553                #ifdef USE_SNZI
554                        // Re-create the snzi
555                        snzi{ log2( lanes.count / 8 ) };
556                        for( idx; (size_t)lanes.count ) {
557                                if( !is_empty(lanes.data[idx]) ) {
558                                        arrive(snzi, idx);
559                                }
560                        }
561                #endif
562        }
563
564        // Make sure that everything is consistent
565        /* paranoid */ check( cltr->ready_queue );
566
567        __cfadbg_print_safe(ready_queue, "Kernel : Growing ready queue done\n");
568
569        /* paranoid */ verify( ready_mutate_islocked() );
570}
571
572// Shrink the ready queue
573void ready_queue_shrink(struct cluster * cltr, int target) {
574        /* paranoid */ verify( ready_mutate_islocked() );
575        __cfadbg_print_safe(ready_queue, "Kernel : Shrinking ready queue\n");
576
577        // Make sure that everything is consistent
578        /* paranoid */ check( cltr->ready_queue );
579
580        with( cltr->ready_queue ) {
581                #ifdef USE_SNZI
582                        ^(snzi){};
583                #endif
584
585                // Remember old count
586                size_t ocount = lanes.count;
587
588                // Find new count
589                // Make sure we always have atleast 1 list
590                lanes.count = target >= 2 ? target * 4: 1;
591                /* paranoid */ verify( ocount >= lanes.count );
592                /* paranoid */ verify( lanes.count == target * 4 || target < 2 );
593
594                // for printing count the number of displaced threads
595                #if defined(__CFA_DEBUG_PRINT__) || defined(__CFA_DEBUG_PRINT_READY_QUEUE__)
596                        __attribute__((unused)) size_t displaced = 0;
597                #endif
598
599                // redistribute old data
600                for( idx; (size_t)lanes.count ~ ocount) {
601                        // Lock is not strictly needed but makes checking invariants much easier
602                        __attribute__((unused)) bool locked = __atomic_try_acquire(&lanes.data[idx].lock);
603                        verify(locked);
604
605                        // As long as we can pop from this lane to push the threads somewhere else in the queue
606                        while(!is_empty(lanes.data[idx])) {
607                                struct $thread * thrd;
608                                thrd = pop(lanes.data[idx]);
609
610                                push(cltr, thrd);
611
612                                // for printing count the number of displaced threads
613                                #if defined(__CFA_DEBUG_PRINT__) || defined(__CFA_DEBUG_PRINT_READY_QUEUE__)
614                                        displaced++;
615                                #endif
616                        }
617
618                        // Unlock the lane
619                        __atomic_unlock(&lanes.data[idx].lock);
620
621                        // TODO print the queue statistics here
622
623                        ^(lanes.data[idx]){};
624                }
625
626                __cfadbg_print_safe(ready_queue, "Kernel : Shrinking ready queue displaced %zu threads\n", displaced);
627
628                // Allocate new array (uses realloc and memcpies the data)
629                lanes.data = alloc(lanes.data, lanes.count);
630
631                // Fix the moved data
632                for( idx; (size_t)lanes.count ) {
633                        fix(lanes.data[idx]);
634                }
635
636                #ifdef USE_SNZI
637                        // Re-create the snzi
638                        snzi{ log2( lanes.count / 8 ) };
639                        for( idx; (size_t)lanes.count ) {
640                                if( !is_empty(lanes.data[idx]) ) {
641                                        arrive(snzi, idx);
642                                }
643                        }
644                #endif
645        }
646
647        // Make sure that everything is consistent
648        /* paranoid */ check( cltr->ready_queue );
649
650        __cfadbg_print_safe(ready_queue, "Kernel : Shrinking ready queue done\n");
651        /* paranoid */ verify( ready_mutate_islocked() );
652}
Note: See TracBrowser for help on using the repository browser.