[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 { |
---|
[b9c06b98] | 50 | actor * receiver; |
---|
| 51 | message * msg; |
---|
| 52 | __receive_fn fn; |
---|
[c042d79] | 53 | }; |
---|
| 54 | |
---|
[1e940de0] | 55 | struct a_msg { |
---|
[b9c06b98] | 56 | int m; |
---|
[1e940de0] | 57 | }; |
---|
[e39cfb9] | 58 | static inline void ?{}( request & this ) {} |
---|
[b065dbb] | 59 | static inline void ?{}( request & this, actor * receiver, message * msg, __receive_fn fn ) { |
---|
[b9c06b98] | 60 | this.receiver = receiver; |
---|
| 61 | this.msg = msg; |
---|
| 62 | this.fn = fn; |
---|
[c042d79] | 63 | } |
---|
[ecfe574] | 64 | static inline void ?{}( request & this, request & copy ) { |
---|
[b9c06b98] | 65 | this.receiver = copy.receiver; |
---|
| 66 | this.msg = copy.msg; |
---|
| 67 | this.fn = copy.fn; |
---|
[ecfe574] | 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 { |
---|
[b9c06b98] | 73 | request * buffer; |
---|
| 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) { |
---|
[b9c06b98] | 78 | buffer_size = buf_size; |
---|
| 79 | buffer = aalloc( buffer_size ); |
---|
| 80 | count = 0; |
---|
| 81 | utilized = 0; |
---|
| 82 | index = 0; |
---|
| 83 | last_size = 0; |
---|
[ecfe574] | 84 | } |
---|
[1e940de0] | 85 | static inline void ^?{}( copy_queue & this ) with(this) { |
---|
[b9c06b98] | 86 | DEBUG_ABORT( count != 0, "Actor system terminated with messages sent but not received\n" ); |
---|
| 87 | adelete(buffer); |
---|
[1e940de0] | 88 | } |
---|
[ecfe574] | 89 | |
---|
[e39cfb9] | 90 | static inline void insert( copy_queue & this, request & elem ) with(this) { |
---|
[b9c06b98] | 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 ); |
---|
| 96 | } |
---|
| 97 | memcpy( &buffer[count], &elem, sizeof(request) ); |
---|
| 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) { |
---|
[b9c06b98] | 104 | if ( count > 0 ) { |
---|
| 105 | count--; |
---|
| 106 | size_t old_idx = index; |
---|
| 107 | index = count == 0 ? 0 : index + 1; |
---|
| 108 | return buffer[old_idx]; |
---|
| 109 | } |
---|
| 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) { |
---|
[b9c06b98] | 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 { |
---|
[b9c06b98] | 125 | __spinlock_t mutex_lock; |
---|
| 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 |
---|
| 129 | #ifdef ACTOR_STATS |
---|
| 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) { |
---|
[b9c06b98] | 135 | owned_queue = alloc(); // allocated separately to avoid false sharing |
---|
| 136 | (*owned_queue){ buf_size }; |
---|
| 137 | c_queue = owned_queue; |
---|
| 138 | being_processed = false; |
---|
| 139 | #ifdef ACTOR_STATS |
---|
| 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) { |
---|
[b9c06b98] | 149 | lock( mutex_lock __cfaabi_dbg_ctx2 ); |
---|
| 150 | insert( *c_queue, elem ); |
---|
| 151 | unlock( mutex_lock ); |
---|
[c042d79] | 152 | } // insert |
---|
| 153 | |
---|
[1e38178] | 154 | static inline void transfer( work_queue & this, copy_queue ** transfer_to ) with(this) { |
---|
[b9c06b98] | 155 | lock( mutex_lock __cfaabi_dbg_ctx2 ); |
---|
| 156 | #ifdef __STEAL |
---|
| 157 | |
---|
| 158 | // check if queue is being processed elsewhere |
---|
| 159 | if ( unlikely( being_processed ) ) { |
---|
| 160 | #ifdef ACTOR_STATS |
---|
| 161 | missed++; |
---|
| 162 | #endif |
---|
| 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 | |
---|
| 172 | // swap copy queue ptrs |
---|
| 173 | copy_queue * temp = *transfer_to; |
---|
| 174 | *transfer_to = c_queue; |
---|
| 175 | c_queue = temp; |
---|
| 176 | unlock( mutex_lock ); |
---|
[c042d79] | 177 | } // transfer |
---|
| 178 | |
---|
[1e38178] | 179 | // needed since some info needs to persist past worker lifetimes |
---|
| 180 | struct worker_info { |
---|
[b9c06b98] | 181 | volatile unsigned long long stamp; |
---|
| 182 | #ifdef ACTOR_STATS |
---|
| 183 | size_t stolen_from, try_steal, stolen, empty_stolen, failed_swaps, msgs_stolen; |
---|
| 184 | unsigned long long processed; |
---|
| 185 | size_t gulps; |
---|
| 186 | #endif |
---|
[1e38178] | 187 | }; |
---|
| 188 | static inline void ?{}( worker_info & this ) { |
---|
[b9c06b98] | 189 | #ifdef ACTOR_STATS |
---|
| 190 | this.stolen_from = 0; |
---|
| 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 |
---|
| 196 | this.empty_stolen = 0; // queues empty after steal |
---|
| 197 | this.msgs_stolen = 0; // number of messages stolen |
---|
| 198 | #endif |
---|
| 199 | this.stamp = rdtscl(); |
---|
[1e38178] | 200 | } |
---|
| 201 | |
---|
[efdd18c] | 202 | // #ifdef ACTOR_STATS |
---|
[f23d34db] | 203 | // unsigned int * stolen_arr; |
---|
| 204 | // unsigned int * replaced_queue; |
---|
| 205 | // #endif |
---|
[c042d79] | 206 | thread worker { |
---|
[b9c06b98] | 207 | work_queue ** request_queues; |
---|
| 208 | copy_queue * current_queue; |
---|
| 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, |
---|
[b9c06b98] | 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_, |
---|
[b9c06b98] | 220 | unsigned int start, unsigned int range, int id ) { |
---|
| 221 | ((thread &)this){ clu }; |
---|
| 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 { |
---|
[b9c06b98] | 232 | cluster * cluster; // if workers execute on separate cluster |
---|
| 233 | processor ** processors; // array of virtual processors adding parallelism for workers |
---|
| 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 |
---|
| 238 | worker_info * w_infos; // array of info about each worker |
---|
| 239 | unsigned int nprocessors, nworkers, nrqueues; // number of processors/threads/request queues |
---|
| 240 | bool seperate_clus; // use same or separate cluster for executor |
---|
| 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) { |
---|
[b9c06b98] | 248 | #ifdef ACTOR_STATS |
---|
| 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); |
---|
| 255 | __atomic_add_fetch(&__total_empty_stolen, executor_->w_infos[id].empty_stolen, __ATOMIC_SEQ_CST); |
---|
| 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 |
---|
[1e38178] | 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) { |
---|
[b9c06b98] | 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; |
---|
| 282 | this.is_shutdown = false; |
---|
| 283 | |
---|
| 284 | if ( nworkers == nrqueues ) |
---|
| 285 | no_steal = true; |
---|
| 286 | |
---|
| 287 | #ifdef ACTOR_STATS |
---|
| 288 | // stolen_arr = aalloc( nrqueues ); |
---|
| 289 | // replaced_queue = aalloc( nrqueues ); |
---|
| 290 | __total_workers = nworkers; |
---|
| 291 | #endif |
---|
| 292 | |
---|
| 293 | if ( seperate_clus ) { |
---|
| 294 | cluster = alloc(); |
---|
| 295 | (*cluster){}; |
---|
| 296 | } else cluster = active_cluster(); |
---|
| 297 | |
---|
| 298 | request_queues = aalloc( nrqueues ); |
---|
| 299 | worker_req_queues = aalloc( nrqueues ); |
---|
| 300 | for ( i; nrqueues ) { |
---|
| 301 | request_queues[i]{ buf_size, i }; |
---|
| 302 | worker_req_queues[i] = &request_queues[i]; |
---|
| 303 | } |
---|
| 304 | |
---|
| 305 | processors = aalloc( nprocessors ); |
---|
| 306 | for ( i; nprocessors ) |
---|
| 307 | (*(processors[i] = alloc())){ *cluster }; |
---|
| 308 | |
---|
| 309 | local_queues = aalloc( nworkers ); |
---|
| 310 | workers = aalloc( nworkers ); |
---|
| 311 | w_infos = aalloc( nworkers ); |
---|
| 312 | unsigned int reqPerWorker = nrqueues / nworkers, extras = nrqueues % nworkers; |
---|
| 313 | |
---|
| 314 | for ( i; nworkers ) { |
---|
| 315 | w_infos[i]{}; |
---|
| 316 | local_queues[i]{ buf_size }; |
---|
| 317 | } |
---|
| 318 | |
---|
| 319 | for ( unsigned int i = 0, start = 0, range; i < nworkers; i += 1, start += range ) { |
---|
| 320 | range = reqPerWorker + ( i < extras ? 1 : 0 ); |
---|
| 321 | (*(workers[i] = alloc())){ *cluster, worker_req_queues, &local_queues[i], &this, start, range, i }; |
---|
| 322 | } // for |
---|
[c042d79] | 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) { |
---|
[b9c06b98] | 331 | is_shutdown = true; |
---|
| 332 | |
---|
| 333 | for ( i; nworkers ) |
---|
| 334 | delete( workers[i] ); |
---|
| 335 | |
---|
| 336 | for ( i; nprocessors ) { |
---|
| 337 | delete( processors[i] ); |
---|
| 338 | } // for |
---|
| 339 | |
---|
| 340 | #ifdef ACTOR_STATS |
---|
| 341 | size_t misses = 0; |
---|
| 342 | for ( i; nrqueues ) { |
---|
| 343 | misses += worker_req_queues[i]->missed; |
---|
| 344 | } |
---|
| 345 | // adelete( stolen_arr ); |
---|
| 346 | // adelete( replaced_queue ); |
---|
| 347 | #endif |
---|
| 348 | |
---|
| 349 | adelete( workers ); |
---|
| 350 | adelete( w_infos ); |
---|
| 351 | adelete( local_queues ); |
---|
| 352 | adelete( request_queues ); |
---|
| 353 | adelete( worker_req_queues ); |
---|
| 354 | adelete( processors ); |
---|
| 355 | if ( seperate_clus ) delete( cluster ); |
---|
| 356 | |
---|
| 357 | #ifdef ACTOR_STATS // print formatted stats |
---|
| 358 | printf(" Actor System Stats:\n"); |
---|
| 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); |
---|
| 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); |
---|
| 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); |
---|
| 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) { |
---|
[b9c06b98] | 374 | #ifdef __CFA_DEBUG__ |
---|
| 375 | size_t temp = __atomic_fetch_add( &__next_ticket, 1, __ATOMIC_SEQ_CST) % nrqueues; |
---|
| 376 | |
---|
| 377 | // reserve MAX for dead actors |
---|
| 378 | if ( unlikely( temp == MAX ) ) temp = __atomic_fetch_add( &__next_ticket, 1, __ATOMIC_SEQ_CST) % nrqueues; |
---|
| 379 | return temp; |
---|
| 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; |
---|
[b9c06b98] | 387 | static bool __actor_executor_passed = false; // was an executor passed to start_actor_system |
---|
| 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 { |
---|
[b9c06b98] | 391 | size_t ticket; // executor-queue handle |
---|
| 392 | allocation alloc; // allocation action |
---|
| 393 | inline virtual_dtor; |
---|
[c042d79] | 394 | }; |
---|
| 395 | |
---|
[e23169b] | 396 | static inline void ?{}( actor & this ) with(this) { |
---|
[b9c06b98] | 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 |
---|
| 399 | DEBUG_ABORT( __actor_executor_ == 0p, "Creating actor before calling start_actor_system() can cause undefined behaviour.\n" ); |
---|
| 400 | alloc = Nodelete; |
---|
| 401 | ticket = __get_next_ticket( *__actor_executor_ ); |
---|
| 402 | __atomic_fetch_add( &__num_actors_, 1, __ATOMIC_RELAXED ); |
---|
| 403 | #ifdef ACTOR_STATS |
---|
| 404 | __atomic_fetch_add( &__num_actors_stats, 1, __ATOMIC_SEQ_CST ); |
---|
| 405 | #endif |
---|
[c042d79] | 406 | } |
---|
| 407 | |
---|
| 408 | static inline void check_actor( actor & this ) { |
---|
[b9c06b98] | 409 | if ( this.alloc != Nodelete ) { |
---|
| 410 | switch( this.alloc ) { |
---|
| 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 | |
---|
| 422 | if ( unlikely( __atomic_add_fetch( &__num_actors_, -1, __ATOMIC_RELAXED ) == 0 ) ) { // all actors have terminated |
---|
| 423 | unpark( __actor_executor_thd ); |
---|
| 424 | } |
---|
| 425 | } |
---|
[c042d79] | 426 | } |
---|
| 427 | |
---|
| 428 | struct message { |
---|
[b9c06b98] | 429 | allocation alloc; // allocation action |
---|
| 430 | inline virtual_dtor; |
---|
[c042d79] | 431 | }; |
---|
| 432 | |
---|
[8512a2f] | 433 | static inline void ?{}( message & this ) { |
---|
[b9c06b98] | 434 | this.alloc = Nodelete; |
---|
[8512a2f] | 435 | } |
---|
[0794365] | 436 | static inline void ?{}( message & this, allocation alloc ) { |
---|
[b9c06b98] | 437 | memcpy( &this.alloc, &alloc, sizeof(allocation) ); // optimization to elide ctor |
---|
| 438 | CFA_DEBUG( if ( this.alloc == Finished ) this.alloc = Nodelete; ); |
---|
[1e38178] | 439 | } |
---|
[8512a2f] | 440 | static inline void ^?{}( message & this ) with(this) { |
---|
[b9c06b98] | 441 | CFA_DEBUG( |
---|
[1fbf481] | 442 | if ( alloc == Nodelete ) { |
---|
| 443 | printf( "CFA warning (UNIX pid:%ld) : program terminating with message %p allocated but never sent.\n", |
---|
| 444 | (long int)getpid(), &this ); |
---|
| 445 | } |
---|
| 446 | ) |
---|
[1e38178] | 447 | } |
---|
[c042d79] | 448 | |
---|
| 449 | static inline void check_message( message & this ) { |
---|
[b9c06b98] | 450 | switch ( this.alloc ) { // analyze message status |
---|
| 451 | case Nodelete: CFA_DEBUG( this.alloc = Finished ); break; |
---|
| 452 | case Delete: delete( &this ); break; |
---|
| 453 | case Destroy: ^?{}( this ); break; |
---|
| 454 | case Finished: break; |
---|
| 455 | } // switch |
---|
| 456 | } |
---|
| 457 | static inline allocation set_allocation( message & this, allocation state ) { |
---|
| 458 | CFA_DEBUG( if ( state == Nodelete ) state = Finished; ); |
---|
| 459 | allocation prev = this.alloc; |
---|
| 460 | this.alloc = state; |
---|
| 461 | return prev; |
---|
[c042d79] | 462 | } |
---|
[b9c06b98] | 463 | static inline allocation get_allocation( message & this ) { |
---|
| 464 | return this.alloc; |
---|
[8512a2f] | 465 | } |
---|
[c042d79] | 466 | |
---|
[ecfe574] | 467 | static inline void deliver_request( request & this ) { |
---|
[b9c06b98] | 468 | DEBUG_ABORT( this.receiver->ticket == (unsigned long int)MAX, "Attempted to send message to deleted/dead actor\n" ); |
---|
| 469 | actor * base_actor; |
---|
| 470 | message * base_msg; |
---|
| 471 | allocation temp = this.fn( *this.receiver, *this.msg, &base_actor, &base_msg ); |
---|
| 472 | memcpy( &base_actor->alloc, &temp, sizeof(allocation) ); // optimization to elide ctor |
---|
| 473 | check_message( *base_msg ); |
---|
| 474 | check_actor( *base_actor ); |
---|
[c042d79] | 475 | } |
---|
| 476 | |
---|
[1e38178] | 477 | // tries to atomically swap two queues and returns 0p if the swap failed |
---|
| 478 | // returns ptr to newly owned queue if swap succeeds |
---|
| 479 | static inline work_queue * try_swap_queues( worker & this, unsigned int victim_idx, unsigned int my_idx ) with(this) { |
---|
[b9c06b98] | 480 | work_queue * my_queue = request_queues[my_idx]; |
---|
| 481 | work_queue * other_queue = request_queues[victim_idx]; |
---|
| 482 | |
---|
| 483 | // if either queue is 0p then they are in the process of being stolen |
---|
| 484 | if ( other_queue == 0p ) return 0p; |
---|
| 485 | |
---|
| 486 | // try to set our queue ptr to be 0p. If it fails someone moved our queue so return false |
---|
| 487 | if ( !__atomic_compare_exchange_n( &request_queues[my_idx], &my_queue, 0p, false, __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST ) ) |
---|
| 488 | return 0p; |
---|
| 489 | |
---|
| 490 | // 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 |
---|
| 491 | if ( !__atomic_compare_exchange_n( &request_queues[victim_idx], &other_queue, my_queue, false, __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST ) ) { |
---|
| 492 | /* paranoid */ verify( request_queues[my_idx] == 0p ); |
---|
| 493 | request_queues[my_idx] = my_queue; // reset my queue ptr back to appropriate val |
---|
| 494 | return 0p; |
---|
| 495 | } |
---|
| 496 | |
---|
| 497 | // we have successfully swapped and since our queue is 0p no one will touch it so write back new queue ptr non atomically |
---|
| 498 | request_queues[my_idx] = other_queue; // last write does not need to be atomic |
---|
| 499 | return other_queue; |
---|
[2d028039] | 500 | } |
---|
| 501 | |
---|
| 502 | // once a worker to steal from has been chosen, choose queue to steal from |
---|
[1e38178] | 503 | static inline void choose_queue( worker & this, unsigned int victim_id, unsigned int swap_idx ) with(this) { |
---|
[b9c06b98] | 504 | // have to calculate victim start and range since victim may be deleted before us in shutdown |
---|
| 505 | const unsigned int queues_per_worker = executor_->nrqueues / executor_->nworkers; |
---|
| 506 | const unsigned int extras = executor_->nrqueues % executor_->nworkers; |
---|
| 507 | unsigned int vic_start, vic_range; |
---|
| 508 | if ( extras > victim_id ) { |
---|
| 509 | vic_range = queues_per_worker + 1; |
---|
| 510 | vic_start = vic_range * victim_id; |
---|
| 511 | } else { |
---|
| 512 | vic_start = extras + victim_id * queues_per_worker; |
---|
| 513 | vic_range = queues_per_worker; |
---|
| 514 | } |
---|
| 515 | unsigned int start_idx = prng( vic_range ); |
---|
| 516 | |
---|
| 517 | unsigned int tries = 0; |
---|
| 518 | work_queue * curr_steal_queue; |
---|
| 519 | |
---|
| 520 | for ( unsigned int i = start_idx; tries < vic_range; i = (i + 1) % vic_range ) { |
---|
| 521 | tries++; |
---|
| 522 | curr_steal_queue = request_queues[ i + vic_start ]; |
---|
| 523 | // avoid empty queues and queues that are being operated on |
---|
| 524 | if ( curr_steal_queue == 0p || curr_steal_queue->being_processed || is_empty( *curr_steal_queue->c_queue ) ) |
---|
| 525 | continue; |
---|
| 526 | |
---|
| 527 | #ifdef ACTOR_STATS |
---|
| 528 | curr_steal_queue = try_swap_queues( this, i + vic_start, swap_idx ); |
---|
| 529 | if ( curr_steal_queue ) { |
---|
| 530 | executor_->w_infos[id].msgs_stolen += curr_steal_queue->c_queue->count; |
---|
| 531 | executor_->w_infos[id].stolen++; |
---|
| 532 | if ( is_empty( *curr_steal_queue->c_queue ) ) executor_->w_infos[id].empty_stolen++; |
---|
| 533 | // __atomic_add_fetch(&executor_->w_infos[victim_id].stolen_from, 1, __ATOMIC_RELAXED); |
---|
| 534 | // replaced_queue[swap_idx]++; |
---|
| 535 | // __atomic_add_fetch(&stolen_arr[ i + vic_start ], 1, __ATOMIC_RELAXED); |
---|
| 536 | } else { |
---|
| 537 | executor_->w_infos[id].failed_swaps++; |
---|
| 538 | } |
---|
| 539 | #else |
---|
| 540 | curr_steal_queue = try_swap_queues( this, i + vic_start, swap_idx ); |
---|
| 541 | #endif // ACTOR_STATS |
---|
| 542 | |
---|
| 543 | return; |
---|
| 544 | } |
---|
| 545 | |
---|
| 546 | return; |
---|
[2d028039] | 547 | } |
---|
| 548 | |
---|
| 549 | // choose a worker to steal from |
---|
[1e38178] | 550 | static inline void steal_work( worker & this, unsigned int swap_idx ) with(this) { |
---|
[b9c06b98] | 551 | #if RAND |
---|
| 552 | unsigned int victim = prng( executor_->nworkers ); |
---|
| 553 | if ( victim == id ) victim = ( victim + 1 ) % executor_->nworkers; |
---|
| 554 | choose_queue( this, victim, swap_idx ); |
---|
| 555 | #elif SEARCH |
---|
| 556 | unsigned long long min = MAX; // smaller timestamp means longer since service |
---|
| 557 | int min_id = 0; // use ints not uints to avoid integer underflow without hacky math |
---|
| 558 | int n_workers = executor_->nworkers; |
---|
| 559 | unsigned long long curr_stamp; |
---|
| 560 | int scount = 1; |
---|
| 561 | for ( int i = (id + 1) % n_workers; scount < n_workers; i = (i + 1) % n_workers, scount++ ) { |
---|
| 562 | curr_stamp = executor_->w_infos[i].stamp; |
---|
| 563 | if ( curr_stamp < min ) { |
---|
| 564 | min = curr_stamp; |
---|
| 565 | min_id = i; |
---|
| 566 | } |
---|
| 567 | } |
---|
| 568 | choose_queue( this, min_id, swap_idx ); |
---|
| 569 | #endif |
---|
[2d028039] | 570 | } |
---|
| 571 | |
---|
[1e940de0] | 572 | #define CHECK_TERMINATION if ( unlikely( executor_->is_shutdown ) ) break Exit |
---|
[c042d79] | 573 | void main( worker & this ) with(this) { |
---|
[b9c06b98] | 574 | // #ifdef ACTOR_STATS |
---|
| 575 | // for ( i; executor_->nrqueues ) { |
---|
| 576 | // replaced_queue[i] = 0; |
---|
| 577 | // __atomic_store_n( &stolen_arr[i], 0, __ATOMIC_SEQ_CST ); |
---|
| 578 | // } |
---|
| 579 | // #endif |
---|
[1e38178] | 580 | |
---|
[b9c06b98] | 581 | // threshold of empty queues we see before we go stealing |
---|
| 582 | const unsigned int steal_threshold = 2 * range; |
---|
[1e38178] | 583 | |
---|
[b9c06b98] | 584 | // Store variable data here instead of worker struct to avoid any potential false sharing |
---|
| 585 | unsigned int empty_count = 0; |
---|
| 586 | request & req; |
---|
| 587 | work_queue * curr_work_queue; |
---|
[1e38178] | 588 | |
---|
[b9c06b98] | 589 | Exit: |
---|
| 590 | for ( unsigned int i = 0;; i = (i + 1) % range ) { // cycle through set of request buffers |
---|
| 591 | curr_work_queue = request_queues[i + start]; |
---|
[508671e] | 592 | |
---|
| 593 | #ifndef __STEAL |
---|
| 594 | CHECK_TERMINATION; |
---|
| 595 | #endif |
---|
[b9c06b98] | 596 | |
---|
| 597 | // check if queue is empty before trying to gulp it |
---|
| 598 | if ( is_empty( *curr_work_queue->c_queue ) ) { |
---|
| 599 | #ifdef __STEAL |
---|
| 600 | empty_count++; |
---|
| 601 | if ( empty_count < steal_threshold ) continue; |
---|
| 602 | #else |
---|
| 603 | continue; |
---|
| 604 | #endif |
---|
| 605 | } |
---|
| 606 | transfer( *curr_work_queue, ¤t_queue ); |
---|
| 607 | #ifdef ACTOR_STATS |
---|
| 608 | executor_->w_infos[id].gulps++; |
---|
| 609 | #endif // ACTOR_STATS |
---|
| 610 | #ifdef __STEAL |
---|
| 611 | if ( is_empty( *current_queue ) ) { |
---|
| 612 | if ( unlikely( no_steal ) ) { CHECK_TERMINATION; continue; } |
---|
| 613 | empty_count++; |
---|
| 614 | if ( empty_count < steal_threshold ) continue; |
---|
| 615 | empty_count = 0; |
---|
| 616 | |
---|
| 617 | CHECK_TERMINATION; // check for termination |
---|
| 618 | |
---|
| 619 | __atomic_store_n( &executor_->w_infos[id].stamp, rdtscl(), __ATOMIC_RELAXED ); |
---|
| 620 | |
---|
| 621 | #ifdef ACTOR_STATS |
---|
| 622 | executor_->w_infos[id].try_steal++; |
---|
| 623 | #endif // ACTOR_STATS |
---|
| 624 | |
---|
| 625 | steal_work( this, start + prng( range ) ); |
---|
| 626 | continue; |
---|
| 627 | } |
---|
| 628 | #endif // __STEAL |
---|
| 629 | while ( ! is_empty( *current_queue ) ) { |
---|
| 630 | #ifdef ACTOR_STATS |
---|
| 631 | executor_->w_infos[id].processed++; |
---|
| 632 | #endif |
---|
| 633 | &req = &remove( *current_queue ); |
---|
| 634 | if ( !&req ) continue; |
---|
| 635 | deliver_request( req ); |
---|
| 636 | } |
---|
| 637 | #ifdef __STEAL |
---|
| 638 | curr_work_queue->being_processed = false; // set done processing |
---|
| 639 | empty_count = 0; // we found work so reset empty counter |
---|
| 640 | #endif |
---|
| 641 | |
---|
| 642 | CHECK_TERMINATION; |
---|
| 643 | |
---|
| 644 | // potentially reclaim some of the current queue's vector space if it is unused |
---|
| 645 | reclaim( *current_queue ); |
---|
| 646 | } // for |
---|
[c042d79] | 647 | } |
---|
| 648 | |
---|
| 649 | static inline void send( executor & this, request & req, unsigned long int ticket ) with(this) { |
---|
[b9c06b98] | 650 | insert( request_queues[ticket], req); |
---|
[c042d79] | 651 | } |
---|
| 652 | |
---|
| 653 | static inline void send( actor & this, request & req ) { |
---|
[b9c06b98] | 654 | DEBUG_ABORT( this.ticket == (unsigned long int)MAX, "Attempted to send message to deleted/dead actor\n" ); |
---|
| 655 | send( *__actor_executor_, req, this.ticket ); |
---|
[c042d79] | 656 | } |
---|
| 657 | |
---|
[f23d34db] | 658 | static inline void __reset_stats() { |
---|
[b9c06b98] | 659 | #ifdef ACTOR_STATS |
---|
| 660 | __total_tries = 0; |
---|
| 661 | __total_stolen = 0; |
---|
| 662 | __all_gulps = 0; |
---|
| 663 | __total_failed_swaps = 0; |
---|
| 664 | __total_empty_stolen = 0; |
---|
| 665 | __all_processed = 0; |
---|
| 666 | __num_actors_stats = 0; |
---|
| 667 | __all_msgs_stolen = 0; |
---|
| 668 | #endif |
---|
[f23d34db] | 669 | } |
---|
| 670 | |
---|
[c042d79] | 671 | static inline void start_actor_system( size_t num_thds ) { |
---|
[b9c06b98] | 672 | __reset_stats(); |
---|
| 673 | __actor_executor_thd = active_thread(); |
---|
| 674 | __actor_executor_ = alloc(); |
---|
| 675 | (*__actor_executor_){ 0, num_thds, num_thds == 1 ? 1 : num_thds * 16 }; |
---|
[c042d79] | 676 | } |
---|
| 677 | |
---|
[e39cfb9] | 678 | static inline void start_actor_system() { start_actor_system( get_proc_count( *active_cluster() ) ); } |
---|
[c042d79] | 679 | |
---|
| 680 | static inline void start_actor_system( executor & this ) { |
---|
[b9c06b98] | 681 | __reset_stats(); |
---|
| 682 | __actor_executor_thd = active_thread(); |
---|
| 683 | __actor_executor_ = &this; |
---|
| 684 | __actor_executor_passed = true; |
---|
[c042d79] | 685 | } |
---|
| 686 | |
---|
| 687 | static inline void stop_actor_system() { |
---|
[b9c06b98] | 688 | park(); // unparked when actor system is finished |
---|
[c042d79] | 689 | |
---|
[b9c06b98] | 690 | if ( !__actor_executor_passed ) delete( __actor_executor_ ); |
---|
| 691 | __actor_executor_ = 0p; |
---|
| 692 | __actor_executor_thd = 0p; |
---|
| 693 | __next_ticket = 0; |
---|
| 694 | __actor_executor_passed = false; |
---|
[c042d79] | 695 | } |
---|
[1e38178] | 696 | |
---|
| 697 | // Default messages to send to any actor to change status |
---|
[858350a] | 698 | // assigned at creation to __base_msg_finished to avoid unused message warning |
---|
[1fbf481] | 699 | message __base_msg_finished @= { .alloc : Finished }; |
---|
[b9c06b98] | 700 | struct delete_msg_t { inline message; } delete_msg = __base_msg_finished; |
---|
[1fbf481] | 701 | struct destroy_msg_t { inline message; } destroy_msg = __base_msg_finished; |
---|
| 702 | struct finished_msg_t { inline message; } finished_msg = __base_msg_finished; |
---|
| 703 | |
---|
[b9c06b98] | 704 | allocation receive( actor & this, delete_msg_t & msg ) { return Delete; } |
---|
[1fbf481] | 705 | allocation receive( actor & this, destroy_msg_t & msg ) { return Destroy; } |
---|
| 706 | allocation receive( actor & this, finished_msg_t & msg ) { return Finished; } |
---|