Index: benchmark/readyQ/cycle.cpp =================================================================== --- benchmark/readyQ/cycle.cpp (revision b6460bfa54ce852d257710855d08250e61ae24d7) +++ benchmark/readyQ/cycle.cpp (revision c4241b65cfa6efe7a40dc30450a6a74a14491a08) @@ -71,5 +71,5 @@ { 'r', "ringsize", "Number of threads in a cycle", ring_size } }; - BENCH_OPT_PARSE("cforall cycle benchmark"); + BENCH_OPT_PARSE("libfibre cycle benchmark"); { Index: benchmark/readyQ/locality.cfa =================================================================== --- benchmark/readyQ/locality.cfa (revision b6460bfa54ce852d257710855d08250e61ae24d7) +++ benchmark/readyQ/locality.cfa (revision c4241b65cfa6efe7a40dc30450a6a74a14491a08) @@ -8,5 +8,5 @@ // ================================================== -thread MyThread { +thread __attribute__((aligned(128))) MyThread { struct MyData * volatile data; @@ -35,5 +35,5 @@ // ================================================== -struct MyData { +struct __attribute__((aligned(128))) MyData { uint64_t _p1[16]; // padding uint64_t * data; @@ -71,5 +71,5 @@ // Atomic object where a single thread can wait // May exchanges data -struct MySpot { +struct __attribute__((aligned(128))) MySpot { MyThread * volatile ptr; size_t id; Index: benchmark/readyQ/locality.cpp =================================================================== --- benchmark/readyQ/locality.cpp (revision c4241b65cfa6efe7a40dc30450a6a74a14491a08) +++ benchmark/readyQ/locality.cpp (revision c4241b65cfa6efe7a40dc30450a6a74a14491a08) @@ -0,0 +1,329 @@ +#include "rq_bench.hpp" +#pragma GCC diagnostic push +#pragma GCC diagnostic ignored "-Wunused-parameter" + #include +#pragma GCC diagnostic pop + +struct Result { + uint64_t count = 0; + uint64_t dmigs = 0; + uint64_t gmigs = 0; +}; + +class __attribute__((aligned(128))) bench_sem { + Fibre * volatile ptr = nullptr; +public: + inline bool wait() { + static Fibre * const ready = reinterpret_cast(1ull); + for(;;) { + Fibre * expected = this->ptr; + if(expected == ready) { + if(__atomic_compare_exchange_n(&this->ptr, &expected, nullptr, false, __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST)) { + return false; + } + } + else { + /* paranoid */ assert( expected == nullptr ); + if(__atomic_compare_exchange_n(&this->ptr, &expected, fibre_self(), false, __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST)) { + fibre_park(); + return true; + } + } + + } + } + + inline bool post() { + static Fibre * const ready = reinterpret_cast(1ull); + for(;;) { + Fibre * expected = this->ptr; + if(expected == ready) return false; + if(expected == nullptr) { + if(__atomic_compare_exchange_n(&this->ptr, &expected, ready, false, __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST)) { + return false; + } + } + else { + if(__atomic_compare_exchange_n(&this->ptr, &expected, nullptr, false, __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST)) { + fibre_unpark( expected ); + return true; + } + } + } + } +}; + +// ================================================== +struct __attribute__((aligned(128))) MyData { + uint64_t _p1[16]; // padding + uint64_t * data; + size_t len; + BaseProcessor * ttid; + size_t id; + uint64_t _p2[16]; // padding + + MyData(size_t id, size_t size) + : data( (uintptr_t *)aligned_alloc(128, size * sizeof(uint64_t)) ) + , len( size ) + , ttid( &Context::CurrProcessor() ) + , id( id ) + { + for(size_t i = 0; i < this->len; i++) { + this->data[i] = 0; + } + } + + uint64_t moved(BaseProcessor * ttid) { + if(this->ttid == ttid) { + return 0; + } + this->ttid = ttid; + return 1; + } + + __attribute__((noinline)) void access(size_t idx) { + size_t l = this->len; + this->data[idx % l] += 1; + } +}; + +// ================================================== +struct __attribute__((aligned(128))) MyCtx { + struct MyData * volatile data; + + struct { + struct MySpot ** ptr; + size_t len; + } spots; + + bench_sem sem; + + Result result; + + bool share; + size_t cnt; + BaseProcessor * ttid; + size_t id; + + MyCtx(MyData * d, MySpot ** spots, size_t len, size_t cnt, bool share, size_t id) + : data( d ) + , spots{ .ptr = spots, .len = len } + , share( share ) + , cnt( cnt ) + , ttid( &Context::CurrProcessor() ) + , id( id ) + {} + + uint64_t moved(BaseProcessor * ttid) { + if(this->ttid == ttid) { + return 0; + } + this->ttid = ttid; + return 1; + } +}; + +// ================================================== +// Atomic object where a single thread can wait +// May exchanges data +struct __attribute__((aligned(128))) MySpot { + MyCtx * volatile ptr; + size_t id; + uint64_t _p1[16]; // padding + + MySpot(size_t id) : ptr( nullptr ), id( id ) {} + + + static inline MyCtx * one() { + return reinterpret_cast(1); + } + + // Main handshake of the code + // Single seat, first thread arriving waits + // Next threads unblocks current one and blocks in its place + // if share == true, exchange data in the process + bool put( MyCtx & ctx, MyData * data, bool share) { + // Attempt to CAS our context into the seat + for(;;) { + MyCtx * expected = this->ptr; + if (expected == one()) { // Seat is closed, return + return true; + } + + if (__atomic_compare_exchange_n(&this->ptr, &expected, &ctx, false, __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST)) { + if(expected) { + if(share) { + expected->data = data; + } + expected->sem.post(); + } + break; // We got the seat + } + } + + // Block once on the seat + ctx.sem.wait(); + + // Someone woke us up, get the new data + return false; + } + + // Shutdown the spot + // Wake current thread and mark seat as closed + void release() { + struct MyCtx * val = __atomic_exchange_n(&this->ptr, one(), __ATOMIC_SEQ_CST); + if (!val) { + return; + } + + // Someone was there, release them + val->sem.post(); + } +}; + +// ================================================== +// Random number generator, Go's native one is to slow and global +uint64_t __xorshift64( uint64_t & state ) { + uint64_t x = state; + x ^= x << 13; + x ^= x >> 7; + x ^= x << 17; + return state = x; +} + +// ================================================== +// Do some work by accessing 'cnt' cells in the array +__attribute__((noinline)) void work(MyData & data, size_t cnt, uint64_t & state) { + for (size_t i = 0; i < cnt; i++) { + data.access(__xorshift64(state)); + } +} + +void thread_main( MyCtx & ctx ) { + uint64_t state; + + // Wait for start + ctx.sem.wait(); + + // Main loop + for(;;) { + // Touch our current data, write to invalidate remote cache lines + work( *ctx.data, ctx.cnt, state ); + + // Wait on a random spot + uint64_t idx = __xorshift64(state) % ctx.spots.len; + bool closed = ctx.spots.ptr[idx]->put(ctx, ctx.data, ctx.share); + + // Check if the experiment is over + if (closed) break; + if ( clock_mode && stop) break; + if (!clock_mode && ctx.result.count >= stop_count) break; + + // Check everything is consistent + assert( ctx.data ); + + // write down progress and check migrations + BaseProcessor * ttid = &Context::CurrProcessor(); + ctx.result.count += 1; + ctx.result.gmigs += ctx.moved(ttid); + ctx.result.dmigs += ctx.data->moved(ttid); + } + + __atomic_fetch_add(&threads_left, -1, __ATOMIC_SEQ_CST); +} + +// ================================================== +int main(int argc, char * argv[]) { + unsigned wsize = 2; + unsigned wcnt = 2; + bool share = false; + option_t opt[] = { + BENCH_OPT, + { 'w', "worksize", "Size of the array for each threads, in words (64bit)", wsize}, + { 'c', "workcnt" , "Number of words to touch when working (random pick, cells can be picked more than once)", wcnt }, + { 's', "share" , "Pass the work data to the next thread when blocking", share, parse_truefalse } + }; + BENCH_OPT_PARSE("libfibre cycle benchmark"); + + std::cout.imbue(std::locale("")); + setlocale(LC_ALL, ""); + + unsigned long long global_count = 0; + unsigned long long global_gmigs = 0; + unsigned long long global_dmigs = 0; + + uint64_t start, end; + { + FibreInit(1, nprocs); + MyData * data_arrays[nthreads]; + for(size_t i = 0; i < nthreads; i++) { + data_arrays[i] = new MyData( i, wsize ); + } + + MySpot * spots[nthreads - nprocs]; + for(size_t i = 0; i < (nthreads - nprocs); i++) { + spots[i] = new MySpot{ i }; + } + + threads_left = nthreads; + Fibre * threads[nthreads]; + MyCtx * thddata[nthreads]; + { + for(size_t i = 0; i < nthreads; i++) { + thddata[i] = new MyCtx( + data_arrays[i], + spots, + nthreads - nprocs, + wcnt, + share, + i + ); + threads[i] = new Fibre( reinterpret_cast(thread_main), thddata[i] ); + } + + bool is_tty = isatty(STDOUT_FILENO); + start = getTimeNsec(); + + for(size_t i = 0; i < nthreads; i++) { + thddata[i]->sem.post(); + } + wait(start, is_tty); + + stop = true; + end = getTimeNsec(); + printf("\nDone\n"); + + for(size_t i = 0; i < nthreads; i++) { + thddata[i]->sem.post(); + fibre_join( threads[i], nullptr ); + global_count += thddata[i]->result.count; + global_gmigs += thddata[i]->result.gmigs; + global_dmigs += thddata[i]->result.dmigs; + } + } + + for(size_t i = 0; i < nthreads; i++) { + delete( data_arrays[i] ); + } + + for(size_t i = 0; i < (nthreads - nprocs); i++) { + delete( spots[i] ); + } + } + + printf("Duration (ms) : %'ld\n", to_miliseconds(end - start)); + printf("Number of processors : %'d\n", nprocs); + printf("Number of threads : %'d\n", nthreads); + printf("Total Operations(ops) : %'15llu\n", global_count); + printf("Work size (64bit words): %'15u\n", wsize); + printf("Total Operations(ops) : %'15llu\n", global_count); + printf("Total G Migrations : %'15llu\n", global_gmigs); + printf("Total D Migrations : %'15llu\n", global_dmigs); + printf("Ops per second : %'18.2lf\n", ((double)global_count) / to_fseconds(end - start)); + printf("ns per ops : %'18.2lf\n", ((double)(end - start)) / global_count); + printf("Ops per threads : %'15llu\n", global_count / nthreads); + printf("Ops per procs : %'15llu\n", global_count / nprocs); + printf("Ops/sec/procs : %'18.2lf\n", (((double)global_count) / nprocs) / to_fseconds(end - start)); + printf("ns per ops/procs : %'18.2lf\n", ((double)(end - start)) / (global_count / nprocs)); + fflush(stdout); +} Index: benchmark/readyQ/rq_bench.hpp =================================================================== --- benchmark/readyQ/rq_bench.hpp (revision b6460bfa54ce852d257710855d08250e61ae24d7) +++ benchmark/readyQ/rq_bench.hpp (revision c4241b65cfa6efe7a40dc30450a6a74a14491a08) @@ -97,4 +97,18 @@ } +bool parse_truefalse(const char * arg, bool & value) { + if(strcmp(arg, "true") == 0) { + value = true; + return true; + } + + if(strcmp(arg, "false") == 0) { + value = false; + return true; + } + + return false; +} + bool parse_settrue (const char *, bool & value ) { value = true; @@ -226,5 +240,5 @@ { int idx = 0; - for(int i = 0; i < opt_count; i++) { + for(size_t i = 0; i < opt_count; i++) { if(options[i].long_name) { optarr[idx].name = options[i].long_name; @@ -256,5 +270,5 @@ { int idx = 0; - for(int i = 0; i < opt_count; i++) { + for(size_t i = 0; i < opt_count; i++) { optstring[idx] = options[i].short_name; idx++; @@ -279,10 +293,12 @@ case 'h': out = stdout; + [[fallthrough]]; case '?': usage(argv[0], options, opt_count, usage_msg, out); default: - for(int i = 0; i < opt_count; i++) { + for(size_t i = 0; i < opt_count; i++) { if(opt == options[i].short_name) { const char * arg = optarg ? optarg : ""; + if( arg[0] == '=' ) { arg++; } bool success = options[i].parse_fun( arg, options[i].variable ); if(success) goto NEXT_ARG; @@ -319,5 +335,5 @@ int width = 0; { - for(int i = 0; i < opt_count; i++) { + for(size_t i = 0; i < opt_count; i++) { if(options[i].long_name) { int w = strlen(options[i].long_name); @@ -338,5 +354,5 @@ fprintf(out, "Usage:\n %s %s\n", cmd, help); - for(int i = 0; i < opt_count; i++) { + for(size_t i = 0; i < opt_count; i++) { printopt(out, width, max_width, options[i].short_name, options[i].long_name, options[i].help); }