Changeset e39cfb9

Timestamp:

Jun 19, 2023, 1:55:23 PM (11 months ago)

Author:

caparsons <caparson@…>

Branches:

master

Children:

0388a99

Parents:

ca0c311

Message:

various cleanups and improvements (shutdown flags, automatic processor count)

File:

• libcfa/src/concurrency/actor.hfa (modified) (10 diffs)

libcfa/src/concurrency/actor.hfa

@@ -53 +53 @@
     message * msg;
     __receive_fn fn;
-    // bool stop; // commented from change to termination flag from sentinels C_TODO: remove after confirming no performance degradation
 };
 
@@ -59 +58 @@
     int m;
 };
-static inline void ?{}( request & this ) { 
-    // this.stop = true; // default ctor makes a sentinel
-}
+static inline void ?{}( request & this ) {}
 static inline void ?{}( request & this, actor * base_receiver, actor * receiver, message * base_msg, message * msg, __receive_fn fn ) {
     this.base_receiver = base_receiver;
@@ -68 +65 @@
     this.msg = msg;
     this.fn = fn;
-    // this.stop = false;
 }
 static inline void ?{}( request & this, request & copy ) {
@@ -74 +70 @@
     this.msg = copy.msg;
     this.fn = copy.fn;
-    // this.stop = copy.stop;
 }
 
@@ -97 +92 @@
 }
 
-static inline void insert( copy_queue & this, request & elem ) with(this) { // C_TODO: remove redundant send/insert once decision is made on emplace/copy
+static inline void insert( copy_queue & this, request & elem ) with(this) {
     if ( count >= buffer_size ) { // increase arr size
         last_size = buffer_size;
@@ -338 +333 @@
 
 static inline void ^?{}( executor & this ) with(this) {
-    // #ifdef __STEAL // commented from change to termination flag from sentinels C_TODO: remove after confirming no performance degradation
-    // request sentinels[nrqueues];
-    // for ( unsigned int i = 0; i < nrqueues; i++ ) {
-    //     insert( request_queues[i], sentinels[i] );           // force eventually termination
-    // } // for
-    // #else
-    // request sentinels[nworkers];
-    // unsigned int reqPerWorker = nrqueues / nworkers, extras = nrqueues % nworkers;
-    // for ( unsigned int i = 0, step = 0, range; i < nworkers; i += 1, step += range ) {
-    //     range = reqPerWorker + ( i < extras ? 1 : 0 );
-    //     insert( request_queues[step], sentinels[i] );                // force eventually termination
-    // } // for
-    // #endif
     is_shutdown = true;
 
@@ -614 +596 @@
         #ifdef __STEAL
         if ( is_empty( *current_queue ) ) {
-            if ( unlikely( no_steal ) ) { CHECK_TERMINATION; continue; } // C_TODO: if this impacts static/dynamic perf refactor check
+            if ( unlikely( no_steal ) ) { CHECK_TERMINATION; continue; }
             empty_count++;
             if ( empty_count < steal_threshold ) continue;
@@ -637 +619 @@
             &req = &remove( *current_queue );
             if ( !&req ) continue;
-            // if ( req.stop ) break Exit;
             deliver_request( req );
         }
@@ -679 +660 @@
 }
 
-// TODO: potentially revisit getting number of processors
-//  ( currently the value stored in active_cluster()->procs.total is often stale 
-//  and doesn't reflect how many procs are allocated )
-// static inline void start_actor_system() { start_actor_system( active_cluster()->procs.total ); }
-static inline void start_actor_system() { start_actor_system( 1 ); }
+static inline void start_actor_system() { start_actor_system( get_proc_count( *active_cluster() ) ); }
 
 static inline void start_actor_system( executor & this ) {
@@ -693 +670 @@
 
 static inline void stop_actor_system() {
-    park( ); // will receive signal when actor system is finished
+    park( ); // will be unparked when actor system is finished
 
     if ( !__actor_executor_passed ) delete( __actor_executor_ );