Changeset b110bcc for libcfa/src/concurrency/actor.hfa

Timestamp:

Apr 21, 2023, 5:36:12 PM (3 years ago)

Author:

JiadaL <j82liang@…>

Branches:

ADT, master

Children:

28f8f15, 6e4c44d

Parents:

2ed94a9 (diff), 699a97d (diff)
Note: this is a merge changeset, the changes displayed below correspond to the merge itself.
Use the (diff) links above to see all the changes relative to each parent.

Message:

Merge branch 'master' of plg.uwaterloo.ca:software/cfa/cfa-cc

File:

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

libcfa/src/concurrency/actor.hfa

@@ -3 +3 @@
 #include <locks.hfa>
 #include <limits.hfa>
-#include <list.hfa>
 #include <kernel.hfa>
+#include <iofwd.hfa>
+#include <virtual_dtor.hfa>
 
 #ifdef __CFA_DEBUG__
@@ -20 +21 @@
 // Define the default number of executor request-queues (mailboxes) written to by actors and serviced by the
 // actor-executor threads. Must be greater than 0.
-#define __DEFAULT_EXECUTOR_RQUEUES__ 2
+#define __DEFAULT_EXECUTOR_RQUEUES__ 4
 
 // Define if executor is created in a separate cluster
 #define __DEFAULT_EXECUTOR_SEPCLUS__ false
 
-// when you flip this make sure to recompile compiler and flip the appropriate flag there too in Actors.cpp
-#define __ALLOC 0
+#define __DEFAULT_EXECUTOR_BUFSIZE__ 10
+
+#define __STEAL 0 // workstealing toggle. Disjoint from toggles above
+
+// workstealing heuristic selection (only set one to be 1)
+// #define RAND 0
+#define SEARCH 1
+
+// show stats
+// #define ACTOR_STATS
 
 // forward decls
 struct actor;
 struct message;
+struct executor;
 
 enum Allocation { Nodelete, Delete, Destroy, Finished }; // allocation status
@@ -40 +50 @@
     __receive_fn fn;
     bool stop;
-    inline dlink(request);
 };
-P9_EMBEDDED( request, dlink(request) )
 
 static inline void ?{}( request & this ) { this.stop = true; } // default ctor makes a sentinel
@@ -58 +66 @@
 }
 
-// hybrid data structure. Copies until buffer is full and then allocates for intrusive list
+// Vector-like data structure that supports O(1) queue operations with no bound on size
+// assumes gulping behaviour (once a remove occurs, removes happen until empty beforw next insert)
 struct copy_queue {
-    dlist( request ) list;
-    #if ! __ALLOC
     request * buffer;
-    size_t count, buffer_size, index;
-    #endif
+    size_t count, buffer_size, index, utilized, last_size;
 };
 static inline void ?{}( copy_queue & this ) {}
 static inline void ?{}( copy_queue & this, size_t buf_size ) with(this) { 
-    list{};
-    #if ! __ALLOC
     buffer_size = buf_size;
     buffer = aalloc( buffer_size );
     count = 0;
+    utilized = 0;
     index = 0;
-    #endif
-}
-static inline void ^?{}( copy_queue & this ) with(this) {
-    #if ! __ALLOC
-    adelete(buffer);
-    #endif
-}
+    last_size = 0;
+}
+static inline void ^?{}( copy_queue & this ) with(this) { adelete(buffer); }
 
 static inline void insert( copy_queue & this, request & elem ) with(this) {
-    #if ! __ALLOC
-    if ( count < buffer_size ) { // fast path ( no alloc )
-        buffer[count]{ elem };
-        count++;
-        return;
-    }
-    request * new_elem = alloc();
-    (*new_elem){ elem };
-    insert_last( list, *new_elem );
-    #else
-    insert_last( list, elem );
-    #endif
+    if ( count >= buffer_size ) { // increase arr size
+        last_size = buffer_size;
+        buffer_size = 2 * buffer_size;
+        buffer = realloc( buffer, sizeof( request ) * buffer_size );
+        /* paranoid */ verify( buffer );
+    }
+    memcpy( &buffer[count], &elem, sizeof(request) );
+    count++;
 }
 
 // once you start removing you need to remove all elements
-// it is not supported to call insert() before the list is fully empty
-// should_delete is an output param
-static inline request & remove( copy_queue & this, bool & should_delete ) with(this) {
-    #if ! __ALLOC
+// it is not supported to call insert() before the array is fully empty
+static inline request & remove( copy_queue & this ) with(this) {
     if ( count > 0 ) {
         count--;
-        should_delete = false;
         size_t old_idx = index;
         index = count == 0 ? 0 : index + 1;
         return buffer[old_idx];
     }
-    #endif
-    should_delete = true;
-    return try_pop_front( list );
-}
-
-static inline bool isEmpty( copy_queue & this ) with(this) {
-    #if ! __ALLOC
-    return count == 0 && list`isEmpty;
-    #else
-    return list`isEmpty;
-    #endif
-}
-
-static size_t __buffer_size = 10; // C_TODO: rework this to be passed from executor through ctors (no need for global)
+    request * ret = 0p;
+    return *0p;
+}
+
+// try to reclaim some memory if less than half of buffer is utilized
+static inline void reclaim( copy_queue & this ) with(this) {
+    if ( utilized >= last_size || buffer_size <= 4 ) { utilized = 0; return; }
+    utilized = 0;
+    buffer_size--;
+    buffer = realloc( buffer, sizeof( request ) * buffer_size ); // try to reclaim some memory
+}
+
+static inline bool isEmpty( copy_queue & this ) with(this) { return count == 0; }
+
 struct work_queue {
     __spinlock_t mutex_lock;
-    copy_queue owned_queue;
-    copy_queue * c_queue; // C_TODO: try putting this on the stack with ptr juggling
-
+    copy_queue * owned_queue;       // copy queue allocated and cleaned up by this work_queue
+    copy_queue * c_queue;           // current queue
+    volatile bool being_processed;  // flag to prevent concurrent processing
+    #ifdef ACTOR_STATS
+    unsigned int id;
+    size_t missed;                  // transfers skipped due to being_processed flag being up
+    #endif
 }; // work_queue
-static inline void ?{}( work_queue & this ) with(this) { 
-    // c_queue = alloc();
-    // (*c_queue){ __buffer_size };
-    owned_queue{ __buffer_size };
-    c_queue = &owned_queue;
-}
-// static inline void ^?{}( work_queue & this ) with(this) { delete( c_queue ); }
+static inline void ?{}( work_queue & this, size_t buf_size, unsigned int i ) with(this) { 
+    owned_queue = alloc();      // allocated separately to avoid false sharing
+    (*owned_queue){ buf_size };
+    c_queue = owned_queue;
+    being_processed = false;
+    #ifdef ACTOR_STATS
+    id = i;
+    missed = 0;
+    #endif
+}
+
+// clean up copy_queue owned by this work_queue
+static inline void ^?{}( work_queue & this ) with(this) { delete( owned_queue ); }
 
 static inline void insert( work_queue & this, request & elem ) with(this) {
@@ -145 +149 @@
 static inline void transfer( work_queue & this, copy_queue ** transfer_to ) with(this) {
     lock( mutex_lock __cfaabi_dbg_ctx2 );
+    #ifdef __STEAL
+
+    // check if queue is being processed elsewhere
+    if ( unlikely( being_processed ) ) {
+        #ifdef ACTOR_STATS
+        missed++;
+        #endif
+        unlock( mutex_lock );
+        return;
+    }
+
+    being_processed = c_queue->count != 0;
+    #endif // __STEAL
+
+    c_queue->utilized = c_queue->count;
+
     // swap copy queue ptrs
     copy_queue * temp = *transfer_to;
@@ -152 +172 @@
 } // transfer
 
+// needed since some info needs to persist past worker lifetimes
+struct worker_info {
+    volatile unsigned long long stamp;
+    #ifdef ACTOR_STATS
+    size_t stolen_from, try_steal, stolen, failed_swaps, msgs_stolen;
+    unsigned long long processed;
+    size_t gulps;
+    #endif
+};
+static inline void ?{}( worker_info & this ) {
+    #ifdef ACTOR_STATS
+    this.stolen_from = 0;
+    this.try_steal = 0;                             // attempts to steal
+    this.stolen = 0;                                // successful steals
+    this.processed = 0;                             // requests processed
+    this.gulps = 0;                                 // number of gulps
+    this.failed_swaps = 0;                          // steal swap failures
+    this.msgs_stolen = 0;                           // number of messages stolen
+    #endif
+    this.stamp = rdtscl();
+}
+
+// #ifdef ACTOR_STATS
+// unsigned int * stolen_arr;
+// unsigned int * replaced_queue;
+// #endif
 thread worker {
-    copy_queue owned_queue;
-    work_queue * request_queues;
+    work_queue ** request_queues;
     copy_queue * current_queue;
-        request & req;
+    executor * executor_;
     unsigned int start, range;
+    int id;
 };
 
-static inline void ?{}( worker & this, cluster & clu, work_queue * request_queues, unsigned int start, unsigned int range ) {
+#ifdef ACTOR_STATS
+// aggregate counters for statistics
+size_t __total_tries = 0, __total_stolen = 0, __total_workers, __all_gulps = 0,
+    __total_failed_swaps = 0, __all_processed = 0, __num_actors_stats = 0, __all_msgs_stolen = 0;
+#endif
+static inline void ?{}( worker & this, cluster & clu, work_queue ** request_queues, copy_queue * current_queue, executor * executor_,
+    unsigned int start, unsigned int range, int id ) {
     ((thread &)this){ clu };
-    this.request_queues = request_queues;
-    // this.current_queue = alloc();
-    // (*this.current_queue){ __buffer_size };
-    this.owned_queue{ __buffer_size };
-    this.current_queue = &this.owned_queue;
-    this.start = start;
-    this.range = range;
-}
-// static inline void ^?{}( worker & mutex this ) with(this) { delete( current_queue ); }
-
+    this.request_queues = request_queues;           // array of all queues
+    this.current_queue = current_queue;             // currently gulped queue (start with empty queue to use in swap later)
+    this.executor_ = executor_;                     // pointer to current executor
+    this.start = start;                             // start of worker's subrange of request_queues
+    this.range = range;                             // size of worker's subrange of request_queues
+    this.id = id;                                   // worker's id and index in array of workers
+}
+
+static bool no_steal = false;
 struct executor {
     cluster * cluster;                                                      // if workers execute on separate cluster
         processor ** processors;                                            // array of virtual processors adding parallelism for workers
-        work_queue * request_queues;                                // master list of work request queues
-        worker ** workers;                                                              // array of workers executing work requests
+        work_queue * request_queues;                                // master array of work request queues
+    copy_queue * local_queues;                      // array of all worker local queues to avoid deletion race
+        work_queue ** worker_req_queues;                // secondary array of work queues to allow for swapping
+    worker ** workers;                                                          // array of workers executing work requests
+    worker_info * w_infos;                          // array of info about each worker 
         unsigned int nprocessors, nworkers, nrqueues;   // number of processors/threads/request queues
         bool seperate_clus;                                                             // use same or separate cluster for executor
 }; // executor
 
+// #ifdef ACTOR_STATS
+// __spinlock_t out_lock;
+// #endif
+static inline void ^?{}( worker & mutex this ) with(this) { 
+    #ifdef ACTOR_STATS
+    __atomic_add_fetch(&__all_gulps, executor_->w_infos[id].gulps,__ATOMIC_SEQ_CST);
+    __atomic_add_fetch(&__all_processed, executor_->w_infos[id].processed,__ATOMIC_SEQ_CST);
+    __atomic_add_fetch(&__all_msgs_stolen, executor_->w_infos[id].msgs_stolen,__ATOMIC_SEQ_CST);
+    __atomic_add_fetch(&__total_tries, executor_->w_infos[id].try_steal, __ATOMIC_SEQ_CST);
+    __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);
+
+    // per worker steal stats (uncomment alongside the lock above this routine to print)
+    // lock( out_lock __cfaabi_dbg_ctx2 );
+    // printf("Worker id: %d, processed: %llu messages, attempted %lu, stole: %lu, stolen from: %lu\n", id, processed, try_steal, stolen, __atomic_add_fetch(&executor_->w_infos[id].stolen_from, 0, __ATOMIC_SEQ_CST) );
+    // int count = 0;
+    // int count2 = 0;
+    // for ( i; range ) {
+    //     if ( replaced_queue[start + i] > 0 ){
+    //         count++;
+    //         // printf("%d: %u, ",i, replaced_queue[i]);
+    //     }
+    //     if (__atomic_add_fetch(&stolen_arr[start + i],0,__ATOMIC_SEQ_CST) > 0)
+    //         count2++;
+    // }
+    // printf("swapped with: %d of %u indices\n", count, executor_->nrqueues / executor_->nworkers );
+    // printf("%d of %u indices were stolen\n", count2, executor_->nrqueues / executor_->nworkers );
+    // unlock( out_lock );
+    #endif
+}
+
 static inline void ?{}( executor & this, unsigned int nprocessors, unsigned int nworkers, unsigned int nrqueues, bool seperate_clus, size_t buf_size ) with(this) {
     if ( nrqueues < nworkers ) abort( "nrqueues needs to be >= nworkers\n" );
-    __buffer_size = buf_size;
     this.nprocessors = nprocessors;
     this.nworkers = nworkers;
@@ -189 +273 @@
     this.seperate_clus = seperate_clus;
 
+    if ( nworkers == nrqueues )
+        no_steal = true;
+    
+    #ifdef ACTOR_STATS
+    // stolen_arr = aalloc( nrqueues );
+    // replaced_queue = aalloc( nrqueues );
+    __total_workers = nworkers;
+    #endif
+
     if ( seperate_clus ) {
         cluster = alloc();
@@ -195 +288 @@
 
     request_queues = aalloc( nrqueues );
-    for ( i; nrqueues )
-        request_queues[i]{};
+    worker_req_queues = aalloc( nrqueues );
+    for ( i; nrqueues ) {
+        request_queues[i]{ buf_size, i };
+        worker_req_queues[i] = &request_queues[i];
+    }
     
     processors = aalloc( nprocessors );
@@ -202 +298 @@
         (*(processors[i] = alloc())){ *cluster };
 
-    workers = alloc( nworkers );
+    local_queues = aalloc( nworkers );
+    workers = aalloc( nworkers );
+    w_infos = aalloc( nworkers );
     unsigned int reqPerWorker = nrqueues / nworkers, extras = nrqueues % nworkers;
+
+    for ( i; nworkers ) {
+        w_infos[i]{};
+        local_queues[i]{ buf_size };
+    }
+
     for ( unsigned int i = 0, start = 0, range; i < nworkers; i += 1, start += range ) {
         range = reqPerWorker + ( i < extras ? 1 : 0 );
-        (*(workers[i] = alloc())){ *cluster, request_queues, start, range };
+        (*(workers[i] = alloc())){ *cluster, worker_req_queues, &local_queues[i], &this, start, range, i };
     } // for
 }
-static inline void ?{}( executor & this, unsigned int nprocessors, unsigned int nworkers, unsigned int nrqueues, bool seperate_clus ) { this{ nprocessors, nworkers, nrqueues, seperate_clus, __buffer_size }; }
+static inline void ?{}( executor & this, unsigned int nprocessors, unsigned int nworkers, unsigned int nrqueues, bool seperate_clus ) { this{ nprocessors, nworkers, nrqueues, seperate_clus, __DEFAULT_EXECUTOR_BUFSIZE__ }; }
 static inline void ?{}( executor & this, unsigned int nprocessors, unsigned int nworkers, unsigned int nrqueues ) { this{ nprocessors, nworkers, nrqueues, __DEFAULT_EXECUTOR_SEPCLUS__ }; }
 static inline void ?{}( executor & this, unsigned int nprocessors, unsigned int nworkers ) { this{ nprocessors, nworkers, __DEFAULT_EXECUTOR_RQUEUES__ }; }
@@ -216 +320 @@
 
 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;
-    for ( unsigned int i = 0, step = 0; i < nworkers; i += 1, step += reqPerWorker ) {
+    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
 
     for ( i; nworkers )
@@ -229 +341 @@
     } // for
 
+    #ifdef ACTOR_STATS
+    size_t misses = 0;
+    for ( i; nrqueues ) {
+        misses += worker_req_queues[i]->missed;
+    }
+    // adelete( stolen_arr );
+    // adelete( replaced_queue );
+    #endif
+
     adelete( workers );
+    adelete( w_infos );
+    adelete( local_queues );
     adelete( request_queues );
+    adelete( worker_req_queues );
     adelete( processors );
     if ( seperate_clus ) delete( cluster );
+
+    #ifdef ACTOR_STATS // print formatted stats
+    printf("    Actor System Stats:\n");
+    printf("\tActors Created:\t\t\t\t%lu\n\tMessages Sent:\t\t\t\t%lu\n", __num_actors_stats, __all_processed);
+    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);
+    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);
+    #endif
+        
 }
 
 // this is a static field of executor but have to forward decl for get_next_ticket
-static unsigned int __next_ticket = 0; 
-
-static inline unsigned int get_next_ticket( executor & this ) with(this) {
-    return __atomic_fetch_add( &__next_ticket, 1, __ATOMIC_SEQ_CST) % nrqueues;
+static size_t __next_ticket = 0; 
+
+static inline size_t __get_next_ticket( executor & this ) with(this) {
+    #ifdef __CFA_DEBUG__
+    size_t temp = __atomic_fetch_add( &__next_ticket, 1, __ATOMIC_SEQ_CST) % nrqueues;
+
+    // reserve MAX for dead actors
+    if ( unlikely( temp == MAX ) ) temp = __atomic_fetch_add( &__next_ticket, 1, __ATOMIC_SEQ_CST) % nrqueues;
+    return temp;
+    #else
+    return __atomic_fetch_add( &__next_ticket, 1, __ATOMIC_RELAXED) % nrqueues;
+    #endif
 } // tickets
 
-// C_TODO: update globals in this file to be static fields once the project is done
+// TODO: update globals in this file to be static fields once the static fields project is done
 static executor * __actor_executor_ = 0p;
-static bool __actor_executor_passed = false;        // was an executor passed to start_actor_system
-static unsigned long int __num_actors_;                         // number of actor objects in system
+static bool __actor_executor_passed = false;            // was an executor passed to start_actor_system
+static size_t __num_actors_ = 0;                                        // number of actor objects in system
 static struct thread$ * __actor_executor_thd = 0p;              // used to wake executor after actors finish
 struct actor {
-    unsigned long int ticket;           // executor-queue handle to provide FIFO message execution
-    Allocation allocation_;                     // allocation action
+    size_t ticket;                                          // executor-queue handle
+    Allocation allocation_;                                         // allocation action
+    inline virtual_dtor;
 };
 
-static inline void ?{}( actor & this ) {
+static inline void ?{}( actor & this ) with(this) {
     // Once an actor is allocated it must be sent a message or the actor system cannot stop. Hence, its receive
     // member must be called to end it
-    verifyf( __actor_executor_, "Creating actor before calling start_actor_system()." ); 
-    this.allocation_ = Nodelete;
-    this.ticket = get_next_ticket( *__actor_executor_ );
-    __atomic_fetch_add( &__num_actors_, 1, __ATOMIC_SEQ_CST );
-}
-static inline void ^?{}( actor & this ) {}
+    verifyf( __actor_executor_, "Creating actor before calling start_actor_system() can cause undefined behaviour.\n" ); 
+    allocation_ = Nodelete;
+    ticket = __get_next_ticket( *__actor_executor_ );
+    __atomic_fetch_add( &__num_actors_, 1, __ATOMIC_RELAXED );
+    #ifdef ACTOR_STATS
+    __atomic_fetch_add( &__num_actors_stats, 1, __ATOMIC_SEQ_CST );
+    #endif
+}
 
 static inline void check_actor( actor & this ) {
@@ -276 +423 @@
         }
 
-        if ( unlikely( __atomic_add_fetch( &__num_actors_, -1, __ATOMIC_SEQ_CST ) == 0 ) ) { // all actors have terminated
+        if ( unlikely( __atomic_add_fetch( &__num_actors_, -1, __ATOMIC_RELAXED ) == 0 ) ) { // all actors have terminated
             unpark( __actor_executor_thd );
         }
@@ -284 +431 @@
 struct message {
     Allocation allocation_;                     // allocation action
+    inline virtual_dtor;
 };
 
-static inline void ?{}( message & this ) { this.allocation_ = Nodelete; }
-static inline void ?{}( message & this, Allocation allocation ) { this.allocation_ = allocation; }
-static inline void ^?{}( message & this ) {}
+static inline void ?{}( message & this ) {
+    this.allocation_ = Nodelete;
+}
+static inline void ?{}( message & this, Allocation allocation ) {
+    memcpy( &this.allocation_, &allocation, sizeof(allocation) ); // optimization to elide ctor
+    verifyf( this.allocation_ != Finished, "The Finished Allocation status is not supported for message types.\n");
+}
+static inline void ^?{}( message & this ) with(this) {
+    CFA_DEBUG( if ( allocation_ == Nodelete ) printf("A message at location %p was allocated but never sent.\n", &this); )
+}
 
 static inline void check_message( message & this ) {
     switch ( this.allocation_ ) {                                               // analyze message status
-        case Nodelete: break;
+        case Nodelete: CFA_DEBUG(this.allocation_ = Finished); break;
         case Delete: delete( &this ); break;
         case Destroy: ^?{}(this); break;
@@ -298 +453 @@
     } // switch
 }
+static inline void set_allocation( message & this, Allocation state ) {
+    this.allocation_ = state;
+}
 
 static inline void deliver_request( request & this ) {
-    Allocation actor_allocation = this.fn( *this.receiver, *this.msg );
-    this.receiver->allocation_ = actor_allocation;
+    this.receiver->allocation_ = this.fn( *this.receiver, *this.msg );
+    check_message( *this.msg );
     check_actor( *this.receiver );
-    check_message( *this.msg );
+}
+
+// tries to atomically swap two queues and returns 0p if the swap failed
+// returns ptr to newly owned queue if swap succeeds
+static inline work_queue * try_swap_queues( worker & this, unsigned int victim_idx, unsigned int my_idx ) with(this) {
+    work_queue * my_queue = request_queues[my_idx];
+    work_queue * other_queue = request_queues[victim_idx];
+
+    // if either queue is 0p then they are in the process of being stolen
+    if ( other_queue == 0p ) return 0p;
+
+    // try to set our queue ptr to be 0p. If it fails someone moved our queue so return false
+    if ( !__atomic_compare_exchange_n( &request_queues[my_idx], &my_queue, 0p, false, __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST ) )
+        return 0p;
+
+    // try to set other queue ptr to be our queue ptr. If it fails someone moved the other queue so fix up then return false
+    if ( !__atomic_compare_exchange_n( &request_queues[victim_idx], &other_queue, my_queue, false, __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST ) ) {
+        /* paranoid */ verify( request_queues[my_idx] == 0p );
+        request_queues[my_idx] = my_queue; // reset my queue ptr back to appropriate val
+        return 0p;
+    }
+
+    // we have successfully swapped and since our queue is 0p no one will touch it so write back new queue ptr non atomically
+    request_queues[my_idx] = other_queue; // last write does not need to be atomic
+    return other_queue;
+}
+
+// once a worker to steal from has been chosen, choose queue to steal from
+static inline void choose_queue( worker & this, unsigned int victim_id, unsigned int swap_idx ) with(this) {
+    // have to calculate victim start and range since victim may be deleted before us in shutdown
+    const unsigned int queues_per_worker = executor_->nrqueues / executor_->nworkers;
+    const unsigned int extras = executor_->nrqueues % executor_->nworkers;
+    unsigned int vic_start, vic_range;
+    if ( extras > victim_id  ) {
+        vic_range = queues_per_worker + 1;
+        vic_start = vic_range * victim_id;
+    } else {
+        vic_start = extras + victim_id * queues_per_worker;
+        vic_range = queues_per_worker;
+    }
+    unsigned int start_idx = prng( vic_range );
+
+    unsigned int tries = 0;
+    work_queue * curr_steal_queue;
+
+    for ( unsigned int i = start_idx; tries < vic_range; i = (i + 1) % vic_range ) {
+        tries++;
+        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 ) )
+            continue;
+
+        #ifdef ACTOR_STATS
+        curr_steal_queue = try_swap_queues( this, i + vic_start, swap_idx );
+        if ( curr_steal_queue ) {
+            executor_->w_infos[id].msgs_stolen += curr_steal_queue->c_queue->count;
+            executor_->w_infos[id].stolen++;
+            // __atomic_add_fetch(&executor_->w_infos[victim_id].stolen_from, 1, __ATOMIC_RELAXED);
+            // replaced_queue[swap_idx]++;
+            // __atomic_add_fetch(&stolen_arr[ i + vic_start ], 1, __ATOMIC_RELAXED);
+        } else {
+            executor_->w_infos[id].failed_swaps++;
+        }
+        #else
+        curr_steal_queue = try_swap_queues( this, i + vic_start, swap_idx );
+        #endif // ACTOR_STATS
+
+        return;
+    }
+
+    return;
+}
+
+// choose a worker to steal from
+static inline void steal_work( worker & this, unsigned int swap_idx ) with(this) {
+    #if RAND
+    unsigned int victim = prng( executor_->nworkers );
+    if ( victim == id ) victim = ( victim + 1 ) % executor_->nworkers;
+    choose_queue( this, victim, swap_idx );
+    #elif SEARCH
+    unsigned long long min = MAX; // smaller timestamp means longer since service
+    int min_id = 0; // use ints not uints to avoid integer underflow without hacky math
+    int n_workers = executor_->nworkers;
+    unsigned long long curr_stamp;
+    int scount = 1;
+    for ( int i = (id + 1) % n_workers; scount < n_workers; i = (i + 1) % n_workers, scount++ ) {
+        curr_stamp = executor_->w_infos[i].stamp;
+        if ( curr_stamp < min ) {
+            min = curr_stamp;
+            min_id = i;
+        }
+    } 
+    choose_queue( this, min_id, swap_idx );
+    #endif
 }
 
 void main( worker & this ) with(this) {
-    bool should_delete;
+    // #ifdef ACTOR_STATS
+    // for ( i; executor_->nrqueues ) {
+    //     replaced_queue[i] = 0;
+    //     __atomic_store_n( &stolen_arr[i], 0, __ATOMIC_SEQ_CST );
+    // }
+    // #endif
+
+    // threshold of empty queues we see before we go stealing
+    const unsigned int steal_threshold = 2 * range;
+
+    // Store variable data here instead of worker struct to avoid any potential false sharing
+    unsigned int empty_count = 0;
+    request & req;
+    work_queue * curr_work_queue;
+
     Exit:
     for ( unsigned int i = 0;; i = (i + 1) % range ) { // cycle through set of request buffers
-        // C_TODO: potentially check queue count instead of immediately trying to transfer
-        transfer( request_queues[i + start], &current_queue );
+        curr_work_queue = request_queues[i + start];
+        
+        // check if queue is empty before trying to gulp it
+        if ( isEmpty( *curr_work_queue->c_queue ) ) {
+            #ifdef __STEAL
+            empty_count++;
+            if ( empty_count < steal_threshold ) continue;
+            #else
+            continue;
+            #endif
+        }
+        transfer( *curr_work_queue, &current_queue );
+        #ifdef ACTOR_STATS
+        executor_->w_infos[id].gulps++;
+        #endif // ACTOR_STATS
+        #ifdef __STEAL
+        if ( isEmpty( *current_queue ) ) {
+            if ( unlikely( no_steal ) ) continue;
+            empty_count++;
+            if ( empty_count < steal_threshold ) continue;
+            empty_count = 0;
+
+            __atomic_store_n( &executor_->w_infos[id].stamp, rdtscl(), __ATOMIC_RELAXED );
+            
+            #ifdef ACTOR_STATS
+            executor_->w_infos[id].try_steal++;
+            #endif // ACTOR_STATS
+            
+            steal_work( this, start + prng( range ) );
+            continue;
+        }
+        #endif // __STEAL
         while ( ! isEmpty( *current_queue ) ) {
-            &req = &remove( *current_queue, should_delete );
-            if ( !&req ) continue; // possibly add some work stealing/idle sleep here
+            #ifdef ACTOR_STATS
+            executor_->w_infos[id].processed++;
+            #endif
+            &req = &remove( *current_queue );
+            if ( !&req ) continue;
             if ( req.stop ) break Exit;
             deliver_request( req );
-
-            if ( should_delete ) delete( &req );
-        } // while
+        }
+        #ifdef __STEAL
+        curr_work_queue->being_processed = false; // set done processing
+        empty_count = 0; // we found work so reset empty counter
+        #endif
+        
+        // potentially reclaim some of the current queue's vector space if it is unused
+        reclaim( *current_queue );
     } // for
 }
@@ -328 +631 @@
 
 static inline void send( actor & this, request & req ) {
+    verifyf( this.ticket != (unsigned long int)MAX, "Attempted to send message to deleted/dead actor\n" );
     send( *__actor_executor_, req, this.ticket );
 }
 
+static inline void __reset_stats() {
+    #ifdef ACTOR_STATS
+    __total_tries = 0;
+    __total_stolen = 0;
+    __all_gulps = 0;
+    __total_failed_swaps = 0;
+    __all_processed = 0;
+    __num_actors_stats = 0;
+    __all_msgs_stolen = 0;
+    #endif
+}
+
 static inline void start_actor_system( size_t num_thds ) {
+    __reset_stats();
     __actor_executor_thd = active_thread();
     __actor_executor_ = alloc();
@@ -337 +654 @@
 }
 
-static inline void start_actor_system() { start_actor_system( active_cluster()->procs.total ); }
+// 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( executor & this ) {
+    __reset_stats();
     __actor_executor_thd = active_thread();
     __actor_executor_ = &this;
@@ -354 +676 @@
     __actor_executor_passed = false;
 }
+
+// Default messages to send to any actor to change status
+// assigned at creation to __base_msg_finished to avoid unused message warning
+message __base_msg_finished @= { .allocation_ : Finished };
+struct __DeleteMsg { inline message; } DeleteMsg = __base_msg_finished;
+struct __DestroyMsg { inline message; } DestroyMsg = __base_msg_finished;
+struct __FinishedMsg { inline message; } FinishedMsg = __base_msg_finished;
+
+Allocation receive( actor & this, __DeleteMsg & msg ) { return Delete; }
+Allocation receive( actor & this, __DestroyMsg & msg ) { return Destroy; }
+Allocation receive( actor & this, __FinishedMsg & msg ) { return Finished; }
+