Changes in libcfa/src/concurrency/actor.hfa [0794365:70d8e2f2]

File:

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

libcfa/src/concurrency/actor.hfa

@@ -30 +30 @@
 #define __DEFAULT_EXECUTOR_BUFSIZE__ 10
 
-#define __STEAL 0 // workstealing toggle. Disjoint from toggles above
+#define __STEAL 1 // workstealing toggle. Disjoint from toggles above
 
 // workstealing heuristic selection (only set one to be 1)
@@ -46 +46 @@
 enum allocation { Nodelete, Delete, Destroy, Finished }; // allocation status
 
-typedef allocation (*__receive_fn)(actor &, message &);
+typedef allocation (*__receive_fn)(actor &, message &, actor **, message **);
 struct request {
     actor * receiver;
     message * msg;
     __receive_fn fn;
-    bool stop;
 };
 
-static inline void ?{}( request & this ) { this.stop = true; } // default ctor makes a sentinel
+struct a_msg {
+    int m;
+};
+static inline void ?{}( request & this ) {}
 static inline void ?{}( request & this, actor * receiver, message * msg, __receive_fn fn ) {
     this.receiver = receiver;
     this.msg = msg;
     this.fn = fn;
-    this.stop = false;
 }
 static inline void ?{}( request & this, request & copy ) {
@@ -65 +66 @@
     this.msg = copy.msg;
     this.fn = copy.fn;
-    this.stop = copy.stop;
 }
 
@@ -83 +83 @@
     last_size = 0;
 }
-static inline void ^?{}( copy_queue & this ) with(this) { adelete(buffer); }
+static inline void ^?{}( copy_queue & this ) with(this) {
+    DEBUG_ABORT( count != 0, "Actor system terminated with messages sent but not received\n" );
+    adelete(buffer);
+}
 
 static inline void insert( copy_queue & this, request & elem ) with(this) {
@@ -117 +120 @@
 }
 
-static inline bool isEmpty( copy_queue & this ) with(this) { return count == 0; }
+static inline bool is_empty( copy_queue & this ) with(this) { return count == 0; }
 
 struct work_queue {
@@ -178 +181 @@
     volatile unsigned long long stamp;
     #ifdef ACTOR_STATS
-    size_t stolen_from, try_steal, stolen, failed_swaps, msgs_stolen;
+    size_t stolen_from, try_steal, stolen, empty_stolen, failed_swaps, msgs_stolen;
     unsigned long long processed;
     size_t gulps;
@@ -191 +194 @@
     this.gulps = 0;                                 // number of gulps
     this.failed_swaps = 0;                          // steal swap failures
+    this.empty_stolen = 0;                          // queues empty after steal
     this.msgs_stolen = 0;                           // number of messages stolen
     #endif
@@ -210 +214 @@
 #ifdef ACTOR_STATS
 // aggregate counters for statistics
-size_t __total_tries = 0, __total_stolen = 0, __total_workers, __all_gulps = 0,
+size_t __total_tries = 0, __total_stolen = 0, __total_workers, __all_gulps = 0, __total_empty_stolen = 0,
     __total_failed_swaps = 0, __all_processed = 0, __num_actors_stats = 0, __all_msgs_stolen = 0;
 #endif
@@ -235 +239 @@
         unsigned int nprocessors, nworkers, nrqueues;   // number of processors/threads/request queues
         bool seperate_clus;                                                             // use same or separate cluster for executor
+    volatile bool is_shutdown;                      // flag to communicate shutdown to worker threads
 }; // executor
 
@@ -248 +253 @@
     __atomic_add_fetch(&__total_stolen, executor_->w_infos[id].stolen, __ATOMIC_SEQ_CST);
     __atomic_add_fetch(&__total_failed_swaps, executor_->w_infos[id].failed_swaps, __ATOMIC_SEQ_CST);
+    __atomic_add_fetch(&__total_empty_stolen, executor_->w_infos[id].empty_stolen, __ATOMIC_SEQ_CST);
 
     // per worker steal stats (uncomment alongside the lock above this routine to print)
@@ -274 +280 @@
     this.nrqueues = nrqueues;
     this.seperate_clus = seperate_clus;
+    this.is_shutdown = false;
 
     if ( nworkers == nrqueues )
@@ -322 +329 @@
 
 static inline void ^?{}( executor & this ) with(this) {
-    #ifdef __STEAL
-    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;
 
     for ( i; nworkers )
@@ -365 +360 @@
     size_t avg_gulps = __all_gulps == 0 ? 0 : __all_processed / __all_gulps;
     printf("\tGulps:\t\t\t\t\t%lu\n\tAverage Gulp Size:\t\t\t%lu\n\tMissed gulps:\t\t\t\t%lu\n", __all_gulps, avg_gulps, misses);
-    printf("\tSteal attempts:\t\t\t\t%lu\n\tSteals:\t\t\t\t\t%lu\n\tSteal failures (no candidates):\t\t%lu\n\tSteal failures (failed swaps):\t\t%lu\n", 
-        __total_tries, __total_stolen, __total_tries - __total_stolen - __total_failed_swaps, __total_failed_swaps);
+    printf("\tSteal attempts:\t\t\t\t%lu\n\tSteals:\t\t\t\t\t%lu\n\tSteal failures (no candidates):\t\t%lu\n\tSteal failures (failed swaps):\t\t%lu\t Empty steals:\t\t%lu\n", 
+        __total_tries, __total_stolen, __total_tries - __total_stolen - __total_failed_swaps, __total_failed_swaps, __total_empty_stolen);
     size_t avg_steal = __total_stolen == 0 ? 0 : __all_msgs_stolen / __total_stolen;
     printf("\tMessages stolen:\t\t\t%lu\n\tAverage steal size:\t\t\t%lu\n", __all_msgs_stolen, avg_steal);
@@ -449 +444 @@
 static inline void check_message( message & this ) {
     switch ( this.allocation_ ) {                                               // analyze message status
-        case Nodelete: CFA_DEBUG(this.allocation_ = Finished); break;
+        case Nodelete: CFA_DEBUG( this.allocation_ = Finished ); break;
         case Delete: delete( &this ); break;
-        case Destroy: ^?{}(this); break;
+        case Destroy: ^?{}( this ); break;
         case Finished: break;
     } // switch
@@ -461 +456 @@
 static inline void deliver_request( request & this ) {
     DEBUG_ABORT( this.receiver->ticket == (unsigned long int)MAX, "Attempted to send message to deleted/dead actor\n" );
-    this.receiver->allocation_ = this.fn( *this.receiver, *this.msg );
-    check_message( *this.msg );
-    check_actor( *this.receiver );
+    actor * base_actor;
+    message * base_msg;
+    allocation temp = this.fn( *this.receiver, *this.msg, &base_actor, &base_msg );
+    base_actor->allocation_ = temp;
+    check_message( *base_msg );
+    check_actor( *base_actor );
 }
 
@@ -513 +511 @@
         curr_steal_queue = request_queues[ i + vic_start ];
         // avoid empty queues and queues that are being operated on
-        if ( curr_steal_queue == 0p || curr_steal_queue->being_processed || isEmpty( *curr_steal_queue->c_queue ) )
+        if ( curr_steal_queue == 0p || curr_steal_queue->being_processed || is_empty( *curr_steal_queue->c_queue ) )
             continue;
 
@@ -521 +519 @@
             executor_->w_infos[id].msgs_stolen += curr_steal_queue->c_queue->count;
             executor_->w_infos[id].stolen++;
+            if ( is_empty( *curr_steal_queue->c_queue ) ) executor_->w_infos[id].empty_stolen++;
             // __atomic_add_fetch(&executor_->w_infos[victim_id].stolen_from, 1, __ATOMIC_RELAXED);
             // replaced_queue[swap_idx]++;
@@ -560 +559 @@
 }
 
+#define CHECK_TERMINATION if ( unlikely( executor_->is_shutdown ) ) break Exit
 void main( worker & this ) with(this) {
     // #ifdef ACTOR_STATS
@@ -581 +581 @@
         
         // check if queue is empty before trying to gulp it
-        if ( isEmpty( *curr_work_queue->c_queue ) ) {
+        if ( is_empty( *curr_work_queue->c_queue ) ) {
             #ifdef __STEAL
             empty_count++;
@@ -594 +594 @@
         #endif // ACTOR_STATS
         #ifdef __STEAL
-        if ( isEmpty( *current_queue ) ) {
-            if ( unlikely( no_steal ) ) continue;
+        if ( is_empty( *current_queue ) ) {
+            if ( unlikely( no_steal ) ) { CHECK_TERMINATION; continue; }
             empty_count++;
             if ( empty_count < steal_threshold ) continue;
             empty_count = 0;
+
+            CHECK_TERMINATION; // check for termination
 
             __atomic_store_n( &executor_->w_infos[id].stamp, rdtscl(), __ATOMIC_RELAXED );
@@ -610 +612 @@
         }
         #endif // __STEAL
-        while ( ! isEmpty( *current_queue ) ) {
+        while ( ! is_empty( *current_queue ) ) {
             #ifdef ACTOR_STATS
             executor_->w_infos[id].processed++;
@@ -616 +618 @@
             &req = &remove( *current_queue );
             if ( !&req ) continue;
-            if ( req.stop ) break Exit;
             deliver_request( req );
         }
@@ -623 +624 @@
         empty_count = 0; // we found work so reset empty counter
         #endif
+
+        CHECK_TERMINATION;
         
         // potentially reclaim some of the current queue's vector space if it is unused
@@ -644 +647 @@
     __all_gulps = 0;
     __total_failed_swaps = 0;
+    __total_empty_stolen = 0;
     __all_processed = 0;
     __num_actors_stats = 0;
@@ -657 +661 @@
 }
 
-// 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 ) {
@@ -671 +671 @@
 
 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_ );