source: libcfa/src/concurrency/actor.hfa @ f23d34db

ADTast-experimental
Last change on this file since f23d34db was f23d34db, checked in by caparsons <caparson@…>, 17 months ago

changed some stat collection for actors and attempted to fix matrix.cfa related bug

  • Property mode set to 100644
File size: 26.4 KB
Line 
1#pragma once
2
3#include <locks.hfa>
4#include <limits.hfa>
5#include <kernel.hfa>
6#include <iofwd.hfa>
7
8#ifdef __CFA_DEBUG__
9#define CFA_DEBUG( stmt ) stmt
10#else
11#define CFA_DEBUG( stmt )
12#endif // CFA_DEBUG
13
14// Define the default number of processors created in the executor. Must be greater than 0.
15#define __DEFAULT_EXECUTOR_PROCESSORS__ 2
16
17// Define the default number of threads created in the executor. Must be greater than 0.
18#define __DEFAULT_EXECUTOR_WORKERS__ 2
19
20// Define the default number of executor request-queues (mailboxes) written to by actors and serviced by the
21// actor-executor threads. Must be greater than 0.
22#define __DEFAULT_EXECUTOR_RQUEUES__ 4
23
24// Define if executor is created in a separate cluster
25#define __DEFAULT_EXECUTOR_SEPCLUS__ false
26
27#define __DEFAULT_EXECUTOR_BUFSIZE__ 10
28
29#define __STEAL 0 // workstealing toggle. Disjoint from toggles above
30
31// workstealing heuristic selection (only set one to be 1)
32// #define RAND 0
33#define SEARCH 1
34
35// show stats
36// #define STATS
37
38// forward decls
39struct actor;
40struct message;
41struct executor;
42
43enum Allocation { Nodelete, Delete, Destroy, Finished }; // allocation status
44
45typedef Allocation (*__receive_fn)(actor &, message &);
46struct request {
47    actor * receiver;
48    message * msg;
49    __receive_fn fn;
50    bool stop;
51};
52
53static inline void ?{}( request & this ) { this.stop = true; } // default ctor makes a sentinel
54static inline void ?{}( request & this, actor * receiver, message * msg, __receive_fn fn ) {
55    this.receiver = receiver;
56    this.msg = msg;
57    this.fn = fn;
58    this.stop = false;
59}
60static inline void ?{}( request & this, request & copy ) {
61    this.receiver = copy.receiver;
62    this.msg = copy.msg;
63    this.fn = copy.fn;
64    this.stop = copy.stop;
65}
66
67// Vector-like data structure that supports O(1) queue operations with no bound on size
68// assumes gulping behaviour (once a remove occurs, removes happen until empty beforw next insert)
69struct copy_queue {
70    request * buffer;
71    size_t count, buffer_size, index, utilized, last_size;
72};
73static inline void ?{}( copy_queue & this ) {}
74static inline void ?{}( copy_queue & this, size_t buf_size ) with(this) {
75    buffer_size = buf_size;
76    buffer = aalloc( buffer_size );
77    count = 0;
78    utilized = 0;
79    index = 0;
80    last_size = 0;
81}
82static inline void ^?{}( copy_queue & this ) with(this) { adelete(buffer); }
83
84static inline void insert( copy_queue & this, request & elem ) with(this) {
85    if ( count >= buffer_size ) { // increase arr size
86        last_size = buffer_size;
87        buffer_size = 2 * buffer_size;
88        buffer = realloc( buffer, sizeof( request ) * buffer_size );
89        /* paranoid */ verify( buffer );
90    }
91    memcpy( &buffer[count], &elem, sizeof(request) );
92    count++;
93}
94
95// once you start removing you need to remove all elements
96// it is not supported to call insert() before the array is fully empty
97static inline request & remove( copy_queue & this ) with(this) {
98    if ( count > 0 ) {
99        count--;
100        size_t old_idx = index;
101        index = count == 0 ? 0 : index + 1;
102        return buffer[old_idx];
103    }
104    request * ret = 0p;
105    return *0p;
106}
107
108// try to reclaim some memory if less than half of buffer is utilized
109static inline void reclaim( copy_queue & this ) with(this) {
110    if ( utilized >= last_size || buffer_size <= 4 ) { utilized = 0; return; }
111    utilized = 0;
112    buffer_size--;
113    buffer = realloc( buffer, sizeof( request ) * buffer_size ); // try to reclaim some memory
114}
115
116static inline bool isEmpty( copy_queue & this ) with(this) { return count == 0; }
117
118struct work_queue {
119    __spinlock_t mutex_lock;
120    copy_queue * owned_queue;       // copy queue allocated and cleaned up by this work_queue
121    copy_queue * c_queue;           // current queue
122    volatile bool being_processed;  // flag to prevent concurrent processing
123    #ifdef STATS
124    unsigned int id;
125    size_t missed;                  // transfers skipped due to being_processed flag being up
126    #endif
127}; // work_queue
128static inline void ?{}( work_queue & this, size_t buf_size, unsigned int i ) with(this) {
129    owned_queue = alloc();      // allocated separately to avoid false sharing
130    (*owned_queue){ buf_size };
131    c_queue = owned_queue;
132    being_processed = false;
133    #ifdef STATS
134    id = i;
135    missed = 0;
136    #endif
137}
138
139// clean up copy_queue owned by this work_queue
140static inline void ^?{}( work_queue & this ) with(this) { delete( owned_queue ); }
141
142static inline void insert( work_queue & this, request & elem ) with(this) {
143    lock( mutex_lock __cfaabi_dbg_ctx2 );
144    insert( *c_queue, elem );
145    unlock( mutex_lock );
146} // insert
147
148static inline void transfer( work_queue & this, copy_queue ** transfer_to ) with(this) {
149    lock( mutex_lock __cfaabi_dbg_ctx2 );
150    #ifdef __STEAL
151
152    // check if queue is being processed elsewhere
153    if ( unlikely( being_processed ) ) {
154        #ifdef STATS
155        missed++;
156        #endif
157        unlock( mutex_lock );
158        return;
159    }
160
161    being_processed = c_queue->count != 0;
162    #endif // __STEAL
163
164    c_queue->utilized = c_queue->count;
165
166    // swap copy queue ptrs
167    copy_queue * temp = *transfer_to;
168    *transfer_to = c_queue;
169    c_queue = temp;
170    unlock( mutex_lock );
171} // transfer
172
173// needed since some info needs to persist past worker lifetimes
174struct worker_info {
175    volatile unsigned long long stamp;
176    #ifdef STATS
177    size_t stolen_from, try_steal, stolen, failed_swaps, msgs_stolen;
178    unsigned long long processed;
179    size_t gulps;
180    #endif
181};
182static inline void ?{}( worker_info & this ) {
183    #ifdef STATS
184    this.stolen_from = 0;
185    this.try_steal = 0;                             // attempts to steal
186    this.stolen = 0;                                // successful steals
187    this.processed = 0;                             // requests processed
188    this.gulps = 0;                                 // number of gulps
189    this.failed_swaps = 0;                          // steal swap failures
190    this.msgs_stolen = 0;                           // number of messages stolen
191    #endif
192    this.stamp = rdtscl();
193}
194
195// #ifdef STATS
196// unsigned int * stolen_arr;
197// unsigned int * replaced_queue;
198// #endif
199thread worker {
200    work_queue ** request_queues;
201    copy_queue * current_queue;
202    executor * executor_;
203    unsigned int start, range;
204    int id;
205};
206
207#ifdef STATS
208// aggregate counters for statistics
209size_t __total_tries = 0, __total_stolen = 0, __total_workers, __all_gulps = 0,
210    __total_failed_swaps = 0, __all_processed = 0, __num_actors_stats = 0, __all_msgs_stolen = 0;
211#endif
212static inline void ?{}( worker & this, cluster & clu, work_queue ** request_queues, copy_queue * current_queue, executor * executor_,
213    unsigned int start, unsigned int range, int id ) {
214    ((thread &)this){ clu };
215    this.request_queues = request_queues;           // array of all queues
216    this.current_queue = current_queue;             // currently gulped queue (start with empty queue to use in swap later)
217    this.executor_ = executor_;                     // pointer to current executor
218    this.start = start;                             // start of worker's subrange of request_queues
219    this.range = range;                             // size of worker's subrange of request_queues
220    this.id = id;                                   // worker's id and index in array of workers
221}
222
223static bool no_steal = false;
224struct executor {
225    cluster * cluster;                                                      // if workers execute on separate cluster
226        processor ** processors;                                            // array of virtual processors adding parallelism for workers
227        work_queue * request_queues;                                // master array of work request queues
228    copy_queue * local_queues;                      // array of all worker local queues to avoid deletion race
229        work_queue ** worker_req_queues;                // secondary array of work queues to allow for swapping
230    worker ** workers;                                                          // array of workers executing work requests
231    worker_info * w_infos;                          // array of info about each worker
232        unsigned int nprocessors, nworkers, nrqueues;   // number of processors/threads/request queues
233        bool seperate_clus;                                                             // use same or separate cluster for executor
234}; // executor
235
236// #ifdef STATS
237// __spinlock_t out_lock;
238// #endif
239static inline void ^?{}( worker & mutex this ) with(this) {
240    #ifdef STATS
241    __atomic_add_fetch(&__all_gulps, executor_->w_infos[id].gulps,__ATOMIC_SEQ_CST);
242    __atomic_add_fetch(&__all_processed, executor_->w_infos[id].processed,__ATOMIC_SEQ_CST);
243    __atomic_add_fetch(&__all_msgs_stolen, executor_->w_infos[id].msgs_stolen,__ATOMIC_SEQ_CST);
244    __atomic_add_fetch(&__total_tries, executor_->w_infos[id].try_steal, __ATOMIC_SEQ_CST);
245    __atomic_add_fetch(&__total_stolen, executor_->w_infos[id].stolen, __ATOMIC_SEQ_CST);
246    __atomic_add_fetch(&__total_failed_swaps, executor_->w_infos[id].failed_swaps, __ATOMIC_SEQ_CST);
247
248    // per worker steal stats (uncomment alongside the lock above this routine to print)
249    // lock( out_lock __cfaabi_dbg_ctx2 );
250    // 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) );
251    // int count = 0;
252    // int count2 = 0;
253    // for ( i; range ) {
254    //     if ( replaced_queue[start + i] > 0 ){
255    //         count++;
256    //         // printf("%d: %u, ",i, replaced_queue[i]);
257    //     }
258    //     if (__atomic_add_fetch(&stolen_arr[start + i],0,__ATOMIC_SEQ_CST) > 0)
259    //         count2++;
260    // }
261    // printf("swapped with: %d of %u indices\n", count, executor_->nrqueues / executor_->nworkers );
262    // printf("%d of %u indices were stolen\n", count2, executor_->nrqueues / executor_->nworkers );
263    // unlock( out_lock );
264    #endif
265}
266
267static inline void ?{}( executor & this, unsigned int nprocessors, unsigned int nworkers, unsigned int nrqueues, bool seperate_clus, size_t buf_size ) with(this) {
268    if ( nrqueues < nworkers ) abort( "nrqueues needs to be >= nworkers\n" );
269    this.nprocessors = nprocessors;
270    this.nworkers = nworkers;
271    this.nrqueues = nrqueues;
272    this.seperate_clus = seperate_clus;
273
274    if ( nworkers == nrqueues )
275        no_steal = true;
276   
277    #ifdef STATS
278    // stolen_arr = aalloc( nrqueues );
279    // replaced_queue = aalloc( nrqueues );
280    __total_workers = nworkers;
281    #endif
282
283    if ( seperate_clus ) {
284        cluster = alloc();
285        (*cluster){};
286    } else cluster = active_cluster();
287
288    request_queues = aalloc( nrqueues );
289    worker_req_queues = aalloc( nrqueues );
290    for ( i; nrqueues ) {
291        request_queues[i]{ buf_size, i };
292        worker_req_queues[i] = &request_queues[i];
293    }
294   
295    processors = aalloc( nprocessors );
296    for ( i; nprocessors )
297        (*(processors[i] = alloc())){ *cluster };
298
299    local_queues = aalloc( nworkers );
300    workers = aalloc( nworkers );
301    w_infos = aalloc( nworkers );
302    unsigned int reqPerWorker = nrqueues / nworkers, extras = nrqueues % nworkers;
303
304    for ( i; nworkers ) {
305        w_infos[i]{};
306        local_queues[i]{ buf_size };
307    }
308
309    for ( unsigned int i = 0, start = 0, range; i < nworkers; i += 1, start += range ) {
310        range = reqPerWorker + ( i < extras ? 1 : 0 );
311        (*(workers[i] = alloc())){ *cluster, worker_req_queues, &local_queues[i], &this, start, range, i };
312    } // for
313}
314static 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__ }; }
315static inline void ?{}( executor & this, unsigned int nprocessors, unsigned int nworkers, unsigned int nrqueues ) { this{ nprocessors, nworkers, nrqueues, __DEFAULT_EXECUTOR_SEPCLUS__ }; }
316static inline void ?{}( executor & this, unsigned int nprocessors, unsigned int nworkers ) { this{ nprocessors, nworkers, __DEFAULT_EXECUTOR_RQUEUES__ }; }
317static inline void ?{}( executor & this, unsigned int nprocessors ) { this{ nprocessors, __DEFAULT_EXECUTOR_WORKERS__ }; }
318static inline void ?{}( executor & this ) { this{ __DEFAULT_EXECUTOR_PROCESSORS__ }; }
319
320static inline void ^?{}( executor & this ) with(this) {
321    #ifdef __STEAL
322    request sentinels[nrqueues];
323    for ( unsigned int i = 0; i < nrqueues; i++ ) {
324        insert( request_queues[i], sentinels[i] );              // force eventually termination
325    } // for
326    #else
327    request sentinels[nworkers];
328    unsigned int reqPerWorker = nrqueues / nworkers, extras = nrqueues % nworkers;
329    for ( unsigned int i = 0, step = 0, range; i < nworkers; i += 1, step += range ) {
330        range = reqPerWorker + ( i < extras ? 1 : 0 );
331        insert( request_queues[step], sentinels[i] );           // force eventually termination
332    } // for
333    #endif
334
335    for ( i; nworkers )
336        delete( workers[i] );
337
338    for ( i; nprocessors ) {
339        delete( processors[i] );
340    } // for
341
342    #ifdef STATS
343    size_t misses = 0;
344    for ( i; nrqueues ) {
345        misses += worker_req_queues[i]->missed;
346    }
347    // adelete( stolen_arr );
348    // adelete( replaced_queue );
349    #endif
350
351    adelete( workers );
352    adelete( w_infos );
353    adelete( local_queues );
354    adelete( request_queues );
355    adelete( worker_req_queues );
356    adelete( processors );
357    if ( seperate_clus ) delete( cluster );
358
359    #ifdef STATS
360    printf("    Actor System Stats:\n");
361    printf("\tActors Created:\t\t\t\t%lu\n\tMessages Sent:\t\t\t\t%lu\n", __num_actors_stats, __all_processed);
362    size_t avg_gulps = __all_gulps == 0 ? 0 : __all_processed / __all_gulps;
363    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);
364    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",
365        __total_tries, __total_stolen, __total_tries - __total_stolen - __total_failed_swaps, __total_failed_swaps);
366    size_t avg_steal = __total_stolen == 0 ? 0 : __all_msgs_stolen / __total_stolen;
367    printf("\tMessages stolen:\t\t\t%lu\n\tAverage steal size:\t\t\t%lu\n", __all_msgs_stolen, avg_steal);
368    #endif
369       
370}
371
372// this is a static field of executor but have to forward decl for get_next_ticket
373static unsigned long int __next_ticket = 0;
374
375static inline unsigned long int __get_next_ticket( executor & this ) with(this) {
376    unsigned long int temp = __atomic_fetch_add( &__next_ticket, 1, __ATOMIC_SEQ_CST) % nrqueues;
377
378    // reserve MAX for dead actors
379    if ( temp == MAX ) temp = __atomic_fetch_add( &__next_ticket, 1, __ATOMIC_SEQ_CST) % nrqueues;
380    return temp;
381} // tickets
382
383// TODO: update globals in this file to be static fields once the static fields project is done
384static executor * __actor_executor_ = 0p;
385static bool __actor_executor_passed = false;            // was an executor passed to start_actor_system
386static unsigned long int __num_actors_ = 0;                             // number of actor objects in system
387static struct thread$ * __actor_executor_thd = 0p;              // used to wake executor after actors finish
388struct actor {
389    unsigned long int ticket;                           // executor-queue handle
390    Allocation allocation_;                                         // allocation action
391};
392
393static inline void ?{}( actor & this ) {
394    // Once an actor is allocated it must be sent a message or the actor system cannot stop. Hence, its receive
395    // member must be called to end it
396    verifyf( __actor_executor_, "Creating actor before calling start_actor_system() can cause undefined behaviour.\n" );
397    this.allocation_ = Nodelete;
398    this.ticket = __get_next_ticket( *__actor_executor_ );
399    __atomic_fetch_add( &__num_actors_, 1, __ATOMIC_SEQ_CST );
400    #ifdef STATS
401    __atomic_fetch_add( &__num_actors_stats, 1, __ATOMIC_SEQ_CST );
402    #endif
403}
404static inline void ^?{}( actor & this ) {}
405
406static inline void check_actor( actor & this ) {
407    if ( this.allocation_ != Nodelete ) {
408        switch( this.allocation_ ) {
409            case Delete: delete( &this ); break;
410            case Destroy:
411                CFA_DEBUG( this.ticket = MAX; );        // mark as terminated
412                ^?{}(this);
413                break;
414            case Finished:
415                CFA_DEBUG( this.ticket = MAX; );        // mark as terminated
416                break;
417            default: ;                                                          // stop warning
418        }
419
420        if ( unlikely( __atomic_add_fetch( &__num_actors_, -1, __ATOMIC_SEQ_CST ) == 0 ) ) { // all actors have terminated
421            unpark( __actor_executor_thd );
422        }
423    }
424}
425
426struct message {
427    Allocation allocation_;                     // allocation action
428};
429
430static inline void ?{}( message & this ) { this.allocation_ = Nodelete; }
431static inline void ?{}( message & this, Allocation allocation ) {
432    this.allocation_ = allocation;
433    verifyf( this.allocation_ != Finished, "The Finished Allocation status is not supported for message types.\n");
434}
435static inline void ^?{}( message & this ) {
436    CFA_DEBUG( if ( this.allocation_ == Nodelete ) printf("A message at location %p was allocated but never sent.\n", &this); )
437}
438
439static inline void check_message( message & this ) {
440    CFA_DEBUG( this.allocation_ = Finished; )
441    switch ( this.allocation_ ) {                                               // analyze message status
442        case Nodelete: break;
443        case Delete: delete( &this ); break;
444        case Destroy: ^?{}(this); break;
445        case Finished: break;
446    } // switch
447}
448static inline void set_allocation( message & this, Allocation state ) { this.allocation_ = state; }
449
450static inline void deliver_request( request & this ) {
451    Allocation actor_allocation = this.fn( *this.receiver, *this.msg );
452    this.receiver->allocation_ = actor_allocation;
453    check_actor( *this.receiver );
454    check_message( *this.msg );
455}
456
457// tries to atomically swap two queues and returns 0p if the swap failed
458// returns ptr to newly owned queue if swap succeeds
459static inline work_queue * try_swap_queues( worker & this, unsigned int victim_idx, unsigned int my_idx ) with(this) {
460    work_queue * my_queue = request_queues[my_idx];
461    work_queue * other_queue = request_queues[victim_idx];
462
463    // if either queue is 0p then they are in the process of being stolen
464    if ( other_queue == 0p ) return 0p;
465
466    // try to set our queue ptr to be 0p. If it fails someone moved our queue so return false
467    if ( !__atomic_compare_exchange_n( &request_queues[my_idx], &my_queue, 0p, false, __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST ) )
468        return 0p;
469
470    // 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
471    if ( !__atomic_compare_exchange_n( &request_queues[victim_idx], &other_queue, my_queue, false, __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST ) ) {
472        /* paranoid */ verify( request_queues[my_idx] == 0p );
473        request_queues[my_idx] = my_queue; // reset my queue ptr back to appropriate val
474        return 0p;
475    }
476
477    // we have successfully swapped and since our queue is 0p no one will touch it so write back new queue ptr non atomically
478    request_queues[my_idx] = other_queue; // last write does not need to be atomic
479    return other_queue;
480}
481
482// once a worker to steal from has been chosen, choose queue to steal from
483static inline void choose_queue( worker & this, unsigned int victim_id, unsigned int swap_idx ) with(this) {
484    // have to calculate victim start and range since victim may be deleted before us in shutdown
485    const unsigned int queues_per_worker = executor_->nrqueues / executor_->nworkers;
486    const unsigned int extras = executor_->nrqueues % executor_->nworkers;
487    unsigned int vic_start, vic_range;
488    if ( extras > victim_id  ) {
489        vic_range = queues_per_worker + 1;
490        vic_start = vic_range * victim_id;
491    } else {
492        vic_start = extras + victim_id * queues_per_worker;
493        vic_range = queues_per_worker;
494    }
495    unsigned int start_idx = prng( vic_range );
496
497    unsigned int tries = 0;
498    work_queue * curr_steal_queue;
499
500    for ( unsigned int i = start_idx; tries < vic_range; i = (i + 1) % vic_range ) {
501        tries++;
502        curr_steal_queue = request_queues[ i + vic_start ];
503        // avoid empty queues and queues that are being operated on
504        if ( curr_steal_queue == 0p || curr_steal_queue->being_processed || isEmpty( *curr_steal_queue->c_queue ) )
505            continue;
506
507        #ifdef STATS
508        curr_steal_queue = try_swap_queues( this, i + vic_start, swap_idx );
509        if ( curr_steal_queue ) {
510            executor_->w_infos[id].msgs_stolen += curr_steal_queue->c_queue->count;
511            executor_->w_infos[id].stolen++;
512            // __atomic_add_fetch(&executor_->w_infos[victim_id].stolen_from, 1, __ATOMIC_RELAXED);
513            // replaced_queue[swap_idx]++;
514            // __atomic_add_fetch(&stolen_arr[ i + vic_start ], 1, __ATOMIC_RELAXED);
515        } else {
516            executor_->w_infos[id].failed_swaps++;
517        }
518        #else
519        curr_steal_queue = try_swap_queues( this, i + vic_start, swap_idx );
520        #endif // STATS
521
522        return;
523    }
524
525    return;
526}
527
528// choose a worker to steal from
529static inline void steal_work( worker & this, unsigned int swap_idx ) with(this) {
530    #if RAND
531    unsigned int victim = prng( executor_->nworkers );
532    if ( victim == id ) victim = ( victim + 1 ) % executor_->nworkers;
533    choose_queue( this, victim, swap_idx );
534    #elif SEARCH
535    unsigned long long min = MAX; // smaller timestamp means longer since service
536    int min_id = 0; // use ints not uints to avoid integer underflow without hacky math
537    int n_workers = executor_->nworkers;
538    unsigned long long curr_stamp;
539    int scount = 1;
540    for ( int i = (id + 1) % n_workers; scount < n_workers; i = (i + 1) % n_workers, scount++ ) {
541        curr_stamp = executor_->w_infos[i].stamp;
542        if ( curr_stamp < min ) {
543            min = curr_stamp;
544            min_id = i;
545        }
546    }
547    choose_queue( this, min_id, swap_idx );
548    #endif
549}
550
551void main( worker & this ) with(this) {
552    // #ifdef STATS
553    // for ( i; executor_->nrqueues ) {
554    //     replaced_queue[i] = 0;
555    //     __atomic_store_n( &stolen_arr[i], 0, __ATOMIC_SEQ_CST );
556    // }
557    // #endif
558
559    // threshold of empty queues we see before we go stealing
560    const unsigned int steal_threshold = 2 * range;
561
562    // Store variable data here instead of worker struct to avoid any potential false sharing
563    unsigned int empty_count = 0;
564    request & req;
565    unsigned int curr_idx;
566    work_queue * curr_work_queue;
567
568    Exit:
569    for ( unsigned int i = 0;; i = (i + 1) % range ) { // cycle through set of request buffers
570        curr_idx = i + start;
571        curr_work_queue = request_queues[curr_idx];
572       
573        // check if queue is empty before trying to gulp it
574        if ( isEmpty( *curr_work_queue->c_queue ) ) {
575            #ifdef __STEAL
576            empty_count++;
577            if ( empty_count < steal_threshold ) continue;
578            #else
579            continue;
580            #endif
581        }
582        transfer( *curr_work_queue, &current_queue );
583        #ifdef STATS
584        executor_->w_infos[id].gulps++;
585        #endif // STATS
586        #ifdef __STEAL
587        if ( isEmpty( *current_queue ) ) {
588            if ( unlikely( no_steal ) ) continue;
589            empty_count++;
590            if ( empty_count < steal_threshold ) continue;
591            empty_count = 0;
592
593            __atomic_store_n( &executor_->w_infos[id].stamp, rdtscl(), __ATOMIC_RELAXED );
594           
595            #ifdef STATS
596            executor_->w_infos[id].try_steal++;
597            #endif // STATS
598           
599            steal_work( this, start + prng( range ) );
600            continue;
601        }
602        #endif // __STEAL
603        while ( ! isEmpty( *current_queue ) ) {
604            #ifdef STATS
605            executor_->w_infos[id].processed++;
606            #endif
607            &req = &remove( *current_queue );
608            if ( !&req ) continue;
609            if ( req.stop ) break Exit;
610            deliver_request( req );
611        }
612        #ifdef __STEAL
613        curr_work_queue->being_processed = false; // set done processing
614        empty_count = 0; // we found work so reset empty counter
615        #endif
616       
617        // potentially reclaim some of the current queue's vector space if it is unused
618        reclaim( *current_queue );
619    } // for
620}
621
622static inline void send( executor & this, request & req, unsigned long int ticket ) with(this) {
623    insert( request_queues[ticket], req);
624}
625
626static inline void send( actor & this, request & req ) {
627    verifyf( this.ticket != (unsigned long int)MAX, "Attempted to send message to deleted/dead actor\n" );
628    send( *__actor_executor_, req, this.ticket );
629}
630
631static inline void __reset_stats() {
632    #ifdef STATS
633    __total_tries = 0;
634    __total_stolen = 0;
635    __all_gulps = 0;
636    __total_failed_swaps = 0;
637    __all_processed = 0;
638    __num_actors_stats = 0;
639    __all_msgs_stolen = 0;
640    #endif
641}
642
643static inline void start_actor_system( size_t num_thds ) {
644    __reset_stats();
645    __actor_executor_thd = active_thread();
646    __actor_executor_ = alloc();
647    (*__actor_executor_){ 0, num_thds, num_thds == 1 ? 1 : num_thds * 16 };
648}
649
650// TODO: potentially revisit getting number of processors
651//  ( currently the value stored in active_cluster()->procs.total is often stale
652//  and doesn't reflect how many procs are allocated )
653// static inline void start_actor_system() { start_actor_system( active_cluster()->procs.total ); }
654static inline void start_actor_system() { start_actor_system( 1 ); }
655
656static inline void start_actor_system( executor & this ) {
657    __reset_stats();
658    __actor_executor_thd = active_thread();
659    __actor_executor_ = &this;
660    __actor_executor_passed = true;
661}
662
663static inline void stop_actor_system() {
664    park( ); // will receive signal when actor system is finished
665
666    if ( !__actor_executor_passed ) delete( __actor_executor_ );
667    __actor_executor_ = 0p;
668    __actor_executor_thd = 0p;
669    __next_ticket = 0;
670    __actor_executor_passed = false;
671}
672
673// Default messages to send to any actor to change status
674// assigned at creation to __base_msg_finished to avoid unused message warning
675message __base_msg_finished @= { .allocation_ : Finished };
676struct __DeleteMsg { inline message; } DeleteMsg = __base_msg_finished;
677struct __DestroyMsg { inline message; } DestroyMsg = __base_msg_finished;
678struct __FinishedMsg { inline message; } FinishedMsg = __base_msg_finished;
679void ?{}( __DeleteMsg & this, Allocation status ) { ((message &) this){ status }; }
680void ?{}( __DestroyMsg & this, Allocation status ) { ((message &) this){ status }; }
681void ?{}( __FinishedMsg & this, Allocation status ) { ((message &) this){ status }; }
682
683Allocation receive( actor & this, __DeleteMsg & msg ) { return Delete; }
684Allocation receive( actor & this, __DestroyMsg & msg ) { return Destroy; }
685Allocation receive( actor & this, __FinishedMsg & msg ) { return Finished; }
Note: See TracBrowser for help on using the repository browser.