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