- File:
-
- 1 edited
-
libcfa/src/concurrency/ready_queue.cfa (modified) (26 diffs)
Legend:
- Unmodified
- Added
- Removed
-
libcfa/src/concurrency/ready_queue.cfa
ra2a4566 r078fb05 20 20 21 21 22 //#define USE_RELAXED_FIFO22 #define USE_RELAXED_FIFO 23 23 // #define USE_WORK_STEALING 24 24 // #define USE_CPU_WORK_STEALING 25 #define USE_AWARE_STEALING26 25 27 26 #include "bits/defs.hfa" … … 30 29 31 30 #include "stdlib.hfa" 32 #include "limits.hfa"33 31 #include "math.hfa" 34 32 … … 56 54 #endif 57 55 58 #if defined(USE_AWARE_STEALING) 59 #define READYQ_SHARD_FACTOR 2 60 #define SEQUENTIAL_SHARD 2 61 #elif defined(USE_CPU_WORK_STEALING) 56 #if defined(USE_CPU_WORK_STEALING) 62 57 #define READYQ_SHARD_FACTOR 2 63 58 #elif defined(USE_RELAXED_FIFO) … … 143 138 __kernel_rseq_register(); 144 139 140 __cfadbg_print_safe(ready_queue, "Kernel : Registering proc %p for RW-Lock\n", proc); 145 141 bool * handle = (bool *)&kernelTLS().sched_lock; 146 142 … … 178 174 } 179 175 176 __cfadbg_print_safe(ready_queue, "Kernel : Registering proc %p done, id %lu\n", proc, n); 177 180 178 // Return new spot. 181 179 /* paranoid */ verify(n < ready); … … 192 190 193 191 __atomic_store_n(cell, 0p, __ATOMIC_RELEASE); 192 193 __cfadbg_print_safe(ready_queue, "Kernel : Unregister proc %p\n", proc); 194 194 195 195 __kernel_rseq_unregister(); … … 244 244 245 245 //======================================================================= 246 // caches handling247 248 struct __attribute__((aligned(128))) __ready_queue_caches_t {249 // Count States:250 // - 0 : No one is looking after this cache251 // - 1 : No one is looking after this cache, BUT it's not empty252 // - 2+ : At least one processor is looking after this cache253 volatile unsigned count;254 };255 256 void ?{}(__ready_queue_caches_t & this) { this.count = 0; }257 void ^?{}(__ready_queue_caches_t & this) {}258 259 static inline void depart(__ready_queue_caches_t & cache) {260 /* paranoid */ verify( cache.count > 1);261 __atomic_fetch_add(&cache.count, -1, __ATOMIC_SEQ_CST);262 /* paranoid */ verify( cache.count != 0);263 /* paranoid */ verify( cache.count < 65536 ); // This verify assumes no cluster will have more than 65000 kernel threads mapped to a single cache, which could be correct but is super weird.264 }265 266 static inline void arrive(__ready_queue_caches_t & cache) {267 // for() {268 // unsigned expected = cache.count;269 // unsigned desired = 0 == expected ? 2 : expected + 1;270 // }271 }272 273 //=======================================================================274 246 // Cforall Ready Queue used for scheduling 275 247 //======================================================================= 276 unsigned long long moving_average(unsigned long long currtsc, unsigned long long instsc, unsigned long long old_avg) { 277 /* paranoid */ verifyf( currtsc < 45000000000000000, "Suspiciously large current time: %'llu (%llx)\n", currtsc, currtsc ); 278 /* paranoid */ verifyf( instsc < 45000000000000000, "Suspiciously large insert time: %'llu (%llx)\n", instsc, instsc ); 279 /* paranoid */ verifyf( old_avg < 15000000000000, "Suspiciously large previous average: %'llu (%llx)\n", old_avg, old_avg ); 280 281 const unsigned long long new_val = currtsc > instsc ? currtsc - instsc : 0; 282 const unsigned long long total_weight = 16; 283 const unsigned long long new_weight = 4; 284 const unsigned long long old_weight = total_weight - new_weight; 285 const unsigned long long ret = ((new_weight * new_val) + (old_weight * old_avg)) / total_weight; 286 return ret; 248 unsigned long long moving_average(unsigned long long nval, unsigned long long oval) { 249 const unsigned long long tw = 16; 250 const unsigned long long nw = 4; 251 const unsigned long long ow = tw - nw; 252 return ((nw * nval) + (ow * oval)) / tw; 287 253 } 288 254 … … 304 270 lanes.help[idx].tri = 0; 305 271 } 306 307 caches = alloc( cpu_info.llc_count );308 for( idx; (size_t)cpu_info.llc_count ) {309 (caches[idx]){};310 }311 272 #else 312 lanes.data = 0p; 313 lanes.tscs = 0p; 314 lanes.caches = 0p; 315 lanes.help = 0p; 316 lanes.count = 0; 273 lanes.data = 0p; 274 lanes.tscs = 0p; 275 lanes.help = 0p; 276 lanes.count = 0; 317 277 #endif 318 278 } … … 325 285 free(lanes.data); 326 286 free(lanes.tscs); 327 free(lanes.caches);328 287 free(lanes.help); 329 288 } 330 289 331 290 //----------------------------------------------------------------------- 332 #if defined(USE_AWARE_STEALING)333 __attribute__((hot)) void push(struct cluster * cltr, struct thread$ * thrd, unpark_hint hint) with (cltr->ready_queue) {334 processor * const proc = kernelTLS().this_processor;335 const bool external = (!proc) || (cltr != proc->cltr);336 const bool remote = hint == UNPARK_REMOTE;337 338 unsigned i;339 if( external || remote ) {340 // Figure out where thread was last time and make sure it's valid341 /* paranoid */ verify(thrd->preferred >= 0);342 if(thrd->preferred * READYQ_SHARD_FACTOR < lanes.count) {343 /* paranoid */ verify(thrd->preferred * READYQ_SHARD_FACTOR < lanes.count);344 unsigned start = thrd->preferred * READYQ_SHARD_FACTOR;345 do {346 unsigned r = __tls_rand();347 i = start + (r % READYQ_SHARD_FACTOR);348 /* paranoid */ verify( i < lanes.count );349 // If we can't lock it retry350 } while( !__atomic_try_acquire( &lanes.data[i].lock ) );351 } else {352 do {353 i = __tls_rand() % lanes.count;354 } while( !__atomic_try_acquire( &lanes.data[i].lock ) );355 }356 } else {357 do {358 unsigned r = proc->rdq.its++;359 i = proc->rdq.id + (r % READYQ_SHARD_FACTOR);360 /* paranoid */ verify( i < lanes.count );361 // If we can't lock it retry362 } while( !__atomic_try_acquire( &lanes.data[i].lock ) );363 }364 365 // Actually push it366 push(lanes.data[i], thrd);367 368 // Unlock and return369 __atomic_unlock( &lanes.data[i].lock );370 371 #if !defined(__CFA_NO_STATISTICS__)372 if(unlikely(external || remote)) __atomic_fetch_add(&cltr->stats->ready.push.extrn.success, 1, __ATOMIC_RELAXED);373 else __tls_stats()->ready.push.local.success++;374 #endif375 }376 377 static inline unsigned long long calc_cutoff(const unsigned long long ctsc, const processor * proc, __ready_queue_t & rdq) {378 unsigned start = proc->rdq.id;379 unsigned long long max = 0;380 for(i; READYQ_SHARD_FACTOR) {381 unsigned long long ptsc = ts(rdq.lanes.data[start + i]);382 if(ptsc != -1ull) {383 /* paranoid */ verify( start + i < rdq.lanes.count );384 unsigned long long tsc = moving_average(ctsc, ptsc, rdq.lanes.tscs[start + i].ma);385 if(tsc > max) max = tsc;386 }387 }388 return (max + 2 * max) / 2;389 }390 391 __attribute__((hot)) struct thread$ * pop_fast(struct cluster * cltr) with (cltr->ready_queue) {392 /* paranoid */ verify( lanes.count > 0 );393 /* paranoid */ verify( kernelTLS().this_processor );394 /* paranoid */ verify( kernelTLS().this_processor->rdq.id < lanes.count );395 396 processor * const proc = kernelTLS().this_processor;397 unsigned this = proc->rdq.id;398 /* paranoid */ verify( this < lanes.count );399 __cfadbg_print_safe(ready_queue, "Kernel : pop from %u\n", this);400 401 // Figure out the current cpu and make sure it is valid402 const int cpu = __kernel_getcpu();403 /* paranoid */ verify(cpu >= 0);404 /* paranoid */ verify(cpu < cpu_info.hthrd_count);405 unsigned this_cache = cpu_info.llc_map[cpu].cache;406 __atomic_store_n(&lanes.caches[this / READYQ_SHARD_FACTOR].id, this_cache, __ATOMIC_RELAXED);407 408 const unsigned long long ctsc = rdtscl();409 410 if(proc->rdq.target == MAX) {411 uint64_t chaos = __tls_rand();412 unsigned ext = chaos & 0xff;413 unsigned other = (chaos >> 8) % (lanes.count);414 415 if(ext < 3 || __atomic_load_n(&lanes.caches[other / READYQ_SHARD_FACTOR].id, __ATOMIC_RELAXED) == this_cache) {416 proc->rdq.target = other;417 }418 }419 else {420 const unsigned target = proc->rdq.target;421 __cfadbg_print_safe(ready_queue, "Kernel : %u considering helping %u, tcsc %llu\n", this, target, lanes.tscs[target].tv);422 /* paranoid */ verify( lanes.tscs[target].tv != MAX );423 if(target < lanes.count) {424 const unsigned long long cutoff = calc_cutoff(ctsc, proc, cltr->ready_queue);425 const unsigned long long age = moving_average(ctsc, lanes.tscs[target].tv, lanes.tscs[target].ma);426 __cfadbg_print_safe(ready_queue, "Kernel : Help attempt on %u from %u, age %'llu vs cutoff %'llu, %s\n", target, this, age, cutoff, age > cutoff ? "yes" : "no");427 if(age > cutoff) {428 thread$ * t = try_pop(cltr, target __STATS(, __tls_stats()->ready.pop.help));429 if(t) return t;430 }431 }432 proc->rdq.target = MAX;433 }434 435 for(READYQ_SHARD_FACTOR) {436 unsigned i = this + (proc->rdq.itr++ % READYQ_SHARD_FACTOR);437 if(thread$ * t = try_pop(cltr, i __STATS(, __tls_stats()->ready.pop.local))) return t;438 }439 440 // All lanes where empty return 0p441 return 0p;442 443 }444 __attribute__((hot)) struct thread$ * pop_slow(struct cluster * cltr) with (cltr->ready_queue) {445 unsigned i = __tls_rand() % lanes.count;446 return try_pop(cltr, i __STATS(, __tls_stats()->ready.pop.steal));447 }448 __attribute__((hot)) struct thread$ * pop_search(struct cluster * cltr) {449 return search(cltr);450 }451 #endif452 291 #if defined(USE_CPU_WORK_STEALING) 453 292 __attribute__((hot)) void push(struct cluster * cltr, struct thread$ * thrd, unpark_hint hint) with (cltr->ready_queue) { … … 506 345 } 507 346 508 static inline int pop_getcpu(processor * proc, __ready_queue_caches_t * caches) {509 const int prv = proc->rdq.cpu;510 const int cpu = __kernel_getcpu();511 if( prv != proc->rdq.cpu ) {512 unsigned pidx = cpu_info.llc_map[prv].cache;513 /* paranoid */ verify(pidx < cpu_info.llc_count);514 515 unsigned nidx = cpu_info.llc_map[cpu].cache;516 /* paranoid */ verify(pidx < cpu_info.llc_count);517 518 depart(caches[pidx]);519 arrive(caches[nidx]);520 521 __STATS( /* cpu migs++ */ )522 }523 return proc->rdq.cpu = cpu;524 }525 526 347 // Pop from the ready queue from a given cluster 527 348 __attribute__((hot)) thread$ * pop_fast(struct cluster * cltr) with (cltr->ready_queue) { … … 529 350 /* paranoid */ verify( kernelTLS().this_processor ); 530 351 531 processor * const proc = kernelTLS().this_processor; 532 const int cpu = pop_getcpu( proc, caches ); 533 // const int cpu = __kernel_getcpu(); 352 const int cpu = __kernel_getcpu(); 534 353 /* paranoid */ verify(cpu >= 0); 535 354 /* paranoid */ verify(cpu < cpu_info.hthrd_count); … … 541 360 /* paranoid */ verifyf((map.start + map.count) * READYQ_SHARD_FACTOR <= lanes.count, "have %zu lanes but map can go up to %u", lanes.count, (map.start + map.count) * READYQ_SHARD_FACTOR); 542 361 362 processor * const proc = kernelTLS().this_processor; 543 363 const int start = map.self * READYQ_SHARD_FACTOR; 544 364 const unsigned long long ctsc = rdtscl(); 545 365 546 366 // Did we already have a help target 547 if(proc->rdq.target == MAX) {367 if(proc->rdq.target == -1u) { 548 368 unsigned long long max = 0; 549 369 for(i; READYQ_SHARD_FACTOR) { … … 551 371 if(tsc > max) max = tsc; 552 372 } 553 //proc->rdq.cutoff = (max + 2 * max) / 2;373 proc->rdq.cutoff = (max + 2 * max) / 2; 554 374 /* paranoid */ verify(lanes.count < 65536); // The following code assumes max 65536 cores. 555 375 /* paranoid */ verify(map.count < 65536); // The following code assumes max 65536 cores. … … 564 384 } 565 385 566 /* paranoid */ verify(proc->rdq.target != MAX);386 /* paranoid */ verify(proc->rdq.target != -1u); 567 387 } 568 388 else { … … 575 395 { 576 396 unsigned target = proc->rdq.target; 577 proc->rdq.target = MAX;397 proc->rdq.target = -1u; 578 398 lanes.help[target / READYQ_SHARD_FACTOR].tri++; 579 399 if(moving_average(ctsc - lanes.tscs[target].tv, lanes.tscs[target].ma) > cutoff) { 580 __STATS( __tls_stats()->ready.pop.helped[target]++; )581 400 thread$ * t = try_pop(cltr, target __STATS(, __tls_stats()->ready.pop.help)); 582 401 proc->rdq.last = target; 583 402 if(t) return t; 403 else proc->rdq.target = -1u; 584 404 } 585 proc->rdq.target = MAX;405 else proc->rdq.target = -1u; 586 406 } 587 407 588 408 unsigned last = proc->rdq.last; 589 if(last != MAX && moving_average(ctsc - lanes.tscs[last].tv, lanes.tscs[last].ma) > cutoff) { 590 __STATS( __tls_stats()->ready.pop.helped[last]++; ) 409 if(last != -1u && lanes.tscs[last].tv < cutoff && ts(lanes.data[last]) < cutoff) { 591 410 thread$ * t = try_pop(cltr, last __STATS(, __tls_stats()->ready.pop.help)); 592 411 if(t) return t; 593 412 } 594 413 else { 595 proc->rdq.last = MAX;414 proc->rdq.last = -1u; 596 415 } 597 416 } … … 609 428 processor * const proc = kernelTLS().this_processor; 610 429 unsigned last = proc->rdq.last; 611 if(last != MAX) {430 if(last != -1u) { 612 431 struct thread$ * t = try_pop(cltr, last __STATS(, __tls_stats()->ready.pop.steal)); 613 432 if(t) return t; 614 proc->rdq.last = MAX;433 proc->rdq.last = -1u; 615 434 } 616 435 … … 741 560 #else 742 561 unsigned preferred = thrd->preferred; 743 const bool external = (hint != UNPARK_LOCAL) || (!kernelTLS().this_processor) || preferred == MAX|| thrd->curr_cluster != cltr;562 const bool external = (hint != UNPARK_LOCAL) || (!kernelTLS().this_processor) || preferred == -1u || thrd->curr_cluster != cltr; 744 563 /* paranoid */ verifyf(external || preferred < lanes.count, "Invalid preferred queue %u for %u lanes", preferred, lanes.count ); 745 564 … … 793 612 processor * proc = kernelTLS().this_processor; 794 613 795 if(proc->rdq.target == MAX) {614 if(proc->rdq.target == -1u) { 796 615 unsigned long long min = ts(lanes.data[proc->rdq.id]); 797 616 for(int i = 0; i < READYQ_SHARD_FACTOR; i++) { … … 804 623 else { 805 624 unsigned target = proc->rdq.target; 806 proc->rdq.target = MAX;625 proc->rdq.target = -1u; 807 626 const unsigned long long bias = 0; //2_500_000_000; 808 627 const unsigned long long cutoff = proc->rdq.cutoff > bias ? proc->rdq.cutoff - bias : proc->rdq.cutoff; … … 839 658 // try to pop from a lane given by index w 840 659 static inline struct thread$ * try_pop(struct cluster * cltr, unsigned w __STATS(, __stats_readyQ_pop_t & stats)) with (cltr->ready_queue) { 841 /* paranoid */ verify( w < lanes.count );842 660 __STATS( stats.attempt++; ) 843 661 … … 863 681 // Actually pop the list 864 682 struct thread$ * thrd; 865 #if defined(USE_ AWARE_STEALING) || defined(USE_WORK_STEALING) || defined(USE_CPU_WORK_STEALING)683 #if defined(USE_WORK_STEALING) || defined(USE_CPU_WORK_STEALING) 866 684 unsigned long long tsc_before = ts(lane); 867 685 #endif … … 879 697 __STATS( stats.success++; ) 880 698 881 #if defined(USE_AWARE_STEALING) || defined(USE_WORK_STEALING) || defined(USE_CPU_WORK_STEALING) 882 if (tsv != MAX) { 883 unsigned long long now = rdtscl(); 884 unsigned long long pma = __atomic_load_n(&lanes.tscs[w].ma, __ATOMIC_RELAXED); 885 __atomic_store_n(&lanes.tscs[w].tv, tsv, __ATOMIC_RELAXED); 886 __atomic_store_n(&lanes.tscs[w].ma, moving_average(now, tsc_before, pma), __ATOMIC_RELAXED); 887 } 699 #if defined(USE_WORK_STEALING) || defined(USE_CPU_WORK_STEALING) 700 unsigned long long now = rdtscl(); 701 lanes.tscs[w].tv = tsv; 702 lanes.tscs[w].ma = moving_average(now > tsc_before ? now - tsc_before : 0, lanes.tscs[w].ma); 888 703 #endif 889 704 890 #if defined(USE_ AWARE_STEALING) || defined(USE_CPU_WORK_STEALING)705 #if defined(USE_CPU_WORK_STEALING) 891 706 thrd->preferred = w / READYQ_SHARD_FACTOR; 892 707 #else … … 987 802 /* paranoid */ verifyf( it, "Unexpected null iterator, at index %u of %u\n", i, count); 988 803 it->rdq.id = value; 989 it->rdq.target = MAX;804 it->rdq.target = -1u; 990 805 value += READYQ_SHARD_FACTOR; 991 806 it = &(*it)`next; … … 1000 815 1001 816 static void fix_times( struct cluster * cltr ) with( cltr->ready_queue ) { 1002 #if defined(USE_ AWARE_STEALING) || defined(USE_WORK_STEALING)817 #if defined(USE_WORK_STEALING) 1003 818 lanes.tscs = alloc(lanes.count, lanes.tscs`realloc); 1004 819 for(i; lanes.count) { 1005 lanes.tscs[i].tv = rdtscl(); 1006 lanes.tscs[i].ma = 0; 820 unsigned long long tsc1 = ts(lanes.data[i]); 821 unsigned long long tsc2 = rdtscl(); 822 lanes.tscs[i].tv = min(tsc1, tsc2); 1007 823 } 1008 824 #endif … … 1050 866 // Update original 1051 867 lanes.count = ncount; 1052 1053 lanes.caches = alloc( target, lanes.caches`realloc );1054 868 } 1055 869 … … 1128 942 fix(lanes.data[idx]); 1129 943 } 1130 1131 lanes.caches = alloc( target, lanes.caches`realloc );1132 944 } 1133 945 1134 946 fix_times(cltr); 1135 1136 947 1137 948 reassign_cltr_id(cltr);
Note:
See TracChangeset
for help on using the changeset viewer.
