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