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