Changeset 884f3f67 for libcfa/src/concurrency/kernel

Timestamp:

Mar 14, 2022, 2:24:51 PM (3 years ago)

Author:

Thierry Delisle <tdelisle@…>

Branches:

ADT, ast-experimental, enum, master, pthread-emulation, qualifiedEnum

Children:

bfb9bf5

Parents:

c42b8a1

Message:

Change how the ready queue is initialized to make it common with I/O

Location:

libcfa/src/concurrency/kernel

Files:

• cluster.cfa (modified) (10 diffs)
• startup.cfa (modified) (3 diffs)

libcfa/src/concurrency/kernel/cluster.cfa

@@ -221 +221 @@
 //-----------------------------------------------------------------------
 // Check that all the intrusive queues in the data structure are still consistent
-static void check( __ready_queue_t & q ) with (q) {
+static void check_readyQ( cluster * cltr ) with (cltr->sched) {
         #if defined(__CFA_WITH_VERIFY__)
                 {
-                        for( idx ; lanes.count ) {
-                                __intrusive_lane_t & sl = lanes.data[idx];
-                                assert(!lanes.data[idx].lock);
+                        const unsigned lanes_count = readyQ.count;
+                        for( idx ; lanes_count ) {
+                                __intrusive_lane_t & sl = readyQ.data[idx];
+                                assert(!readyQ.data[idx].lock);
 
                                         if(is_empty(sl)) {
@@ -257 +258 @@
                 it->rdq.id = value;
                 it->rdq.target = MAX;
-                value += __readyq_shard_factor;
+                value += __shard_factor.readyq;
                 it = &(*it)`next;
         }
@@ -268 +269 @@
 }
 
-static void fix_times( struct cluster * cltr ) with( cltr->ready_queue ) {
-        lanes.tscs = alloc(lanes.count, lanes.tscs`realloc);
-        for(i; lanes.count) {
-                lanes.tscs[i].tv = rdtscl();
-                lanes.tscs[i].ma = 0;
+static void fix_times( __timestamp_t * volatile & tscs, unsigned count ) {
+        tscs = alloc(count, tscs`realloc);
+        for(i; count) {
+                tscs[i].tv = rdtscl();
+                tscs[i].ma = 0;
         }
 }
@@ -278 +279 @@
 // Grow the ready queue
 void ready_queue_grow(struct cluster * cltr) {
-        size_t ncount;
         int target = cltr->procs.total;
 
@@ -285 +285 @@
 
         // Make sure that everything is consistent
-        /* paranoid */ check( cltr->ready_queue );
-
-        // grow the ready queue
-        with( cltr->ready_queue ) {
-                // Find new count
-                // Make sure we always have atleast 1 list
-                if(target >= 2) {
-                        ncount = target * __readyq_shard_factor;
-                } else {
-                        ncount = __readyq_single_shard;
-                }
-
-                // Allocate new array (uses realloc and memcpies the data)
-                lanes.data = alloc( ncount, lanes.data`realloc );
-
-                // Fix the moved data
-                for( idx; (size_t)lanes.count ) {
-                        fix(lanes.data[idx]);
-                }
-
-                // Construct new data
-                for( idx; (size_t)lanes.count ~ ncount) {
-                        (lanes.data[idx]){};
-                }
-
-                // Update original
-                lanes.count = ncount;
-
-                lanes.caches = alloc( target, lanes.caches`realloc );
-        }
-
-        fix_times(cltr);
-
+        /* paranoid */ check_readyQ( cltr );
+
+
+        // Find new count
+        // Make sure we always have atleast 1 list
+        size_t ocount = cltr->sched.readyQ.count;
+        size_t ncount = max(target * __shard_factor.readyq, __readyq_single_shard);
+
+        // Do we have to do anything?
+        if( ocount != ncount ) {
+
+                // grow the ready queue
+                with( cltr->sched ) {
+
+                        // Allocate new array (uses realloc and memcpies the data)
+                        readyQ.data = alloc( ncount, readyQ.data`realloc );
+
+                        // Fix the moved data
+                        for( idx; ocount ) {
+                                fix(readyQ.data[idx]);
+                        }
+
+                        // Construct new data
+                        for( idx; ocount ~ ncount) {
+                                (readyQ.data[idx]){};
+                        }
+
+                        // Update original count
+                        readyQ.count = ncount;
+                }
+
+
+                fix_times(cltr->sched.readyQ.tscs, cltr->sched.readyQ.count);
+        }
+
+        // realloc the caches
+        cltr->sched.caches = alloc( target, cltr->sched.caches`realloc );
+
+        // reassign the clusters.
         reassign_cltr_id(cltr);
 
         // Make sure that everything is consistent
-        /* paranoid */ check( cltr->ready_queue );
+        /* paranoid */ check_readyQ( cltr );
+        /* paranoid */ verify( (target == 0) == (cltr->sched.caches == 0p) );
 
         __cfadbg_print_safe(ready_queue, "Kernel : Growing ready queue done\n");
@@ -334 +341 @@
 
         // Make sure that everything is consistent
-        /* paranoid */ check( cltr->ready_queue );
+        /* paranoid */ check_readyQ( cltr );
 
         int target = cltr->procs.total;
 
-        with( cltr->ready_queue ) {
+        with( cltr->sched ) {
                 // Remember old count
-                size_t ocount = lanes.count;
+                size_t ocount = readyQ.count;
 
                 // Find new count
                 // Make sure we always have atleast 1 list
-                lanes.count = target >= 2 ? target * __readyq_shard_factor: __readyq_single_shard;
-                /* paranoid */ verify( ocount >= lanes.count );
-                /* paranoid */ verify( lanes.count == target * __readyq_shard_factor || target < 2 );
+                size_t ncount = max(target * __shard_factor.readyq, __readyq_single_shard);
+                /* paranoid */ verifyf( ocount >= ncount, "Error in shrinking size calculation, %zu >= %zu", ocount, ncount );
+                /* paranoid */ verifyf( ncount == target * __shard_factor.readyq || ncount == __readyq_single_shard,
+                /* paranoid */          "Error in shrinking size calculation, expected %u or %u, got %zu", target * __shard_factor.readyq, ncount );
+
+                readyQ.count = ncount;
 
                 // for printing count the number of displaced threads
@@ -354 +364 @@
 
                 // redistribute old data
-                for( idx; (size_t)lanes.count ~ ocount) {
+                for( idx; ncount ~ ocount) {
                         // Lock is not strictly needed but makes checking invariants much easier
-                        __attribute__((unused)) bool locked = __atomic_try_acquire(&lanes.data[idx].lock);
+                        __attribute__((unused)) bool locked = __atomic_try_acquire(&readyQ.data[idx].lock);
                         verify(locked);
 
                         // As long as we can pop from this lane to push the threads somewhere else in the queue
-                        while(!is_empty(lanes.data[idx])) {
+                        while(!is_empty(readyQ.data[idx])) {
                                 struct thread$ * thrd;
                                 unsigned long long _;
-                                [thrd, _] = pop(lanes.data[idx]);
+                                [thrd, _] = pop(readyQ.data[idx]);
 
                                 push(cltr, thrd, true);
@@ -374 +384 @@
 
                         // Unlock the lane
-                        __atomic_unlock(&lanes.data[idx].lock);
+                        __atomic_unlock(&readyQ.data[idx].lock);
 
                         // TODO print the queue statistics here
 
-                        ^(lanes.data[idx]){};
+                        ^(readyQ.data[idx]){};
                 }
 
@@ -384 +394 @@
 
                 // Allocate new array (uses realloc and memcpies the data)
-                lanes.data = alloc( lanes.count, lanes.data`realloc );
+                readyQ.data = alloc( ncount, readyQ.data`realloc );
 
                 // Fix the moved data
-                for( idx; (size_t)lanes.count ) {
-                        fix(lanes.data[idx]);
-                }
-
-                lanes.caches = alloc( target, lanes.caches`realloc );
-        }
-
-        fix_times(cltr);
+                for( idx; ncount ) {
+                        fix(readyQ.data[idx]);
+                }
+
+                fix_times(readyQ.tscs, ncount);
+        }
+        cltr->sched.caches = alloc( target, cltr->sched.caches`realloc );
+
 
 
@@ -400 +410 @@
 
         // Make sure that everything is consistent
-        /* paranoid */ check( cltr->ready_queue );
+        /* paranoid */ verify( (target == 0) == (cltr->sched.caches == 0p) );
+        /* paranoid */ check_readyQ( cltr );
 
         __cfadbg_print_safe(ready_queue, "Kernel : Shrinking ready queue done\n");
         /* paranoid */ verify( ready_mutate_islocked() );
+}
+
+void ready_queue_close(struct cluster * cltr) {
+        free( cltr->sched.readyQ.data );
+        free( cltr->sched.readyQ.tscs );
+        cltr->sched.readyQ.data = 0p;
+        cltr->sched.readyQ.tscs = 0p;
+        cltr->sched.readyQ.count = 0;
+
+        free( cltr->sched.io.tscs );
+        free( cltr->sched.caches );
 }
 

libcfa/src/concurrency/kernel/startup.cfa

@@ -515 +515 @@
         this.rdq.its = 0;
         this.rdq.itr = 0;
-        this.rdq.id  = MAX;
+        this.rdq.id  = 0;
         this.rdq.target = MAX;
         this.rdq.last = MAX;
@@ -605 +605 @@
         this.name = name;
         this.preemption_rate = preemption_rate;
-        ready_queue{};
+        this.sched.readyQ.data = 0p;
+        this.sched.readyQ.tscs = 0p;
+        this.sched.readyQ.count = 0;
+        this.sched.io.tscs = 0p;
+        this.sched.caches = 0p;
 
         #if !defined(__CFA_NO_STATISTICS__)
@@ -644 +648 @@
         // Unlock the RWlock
         ready_mutate_unlock( last_size );
+
+        ready_queue_close( &this );
+        /* paranoid */ verify( this.sched.readyQ.data == 0p );
+        /* paranoid */ verify( this.sched.readyQ.tscs == 0p );
+        /* paranoid */ verify( this.sched.readyQ.count == 0 );
+        /* paranoid */ verify( this.sched.io.tscs == 0p );
+        /* paranoid */ verify( this.sched.caches == 0p );
+
         enable_interrupts( false ); // Don't poll, could be in main cluster
+
 
         #if !defined(__CFA_NO_STATISTICS__)