Context Navigation

-                      r487198c
+                      r807a632
         int value;
+        int id;
         Node() { creates++; }
 …
 template<>
+relaxed_list<Node>::intrusive_queue_t::stat::Dif relaxed_list<Node>::intrusive_queue_t::stat::dif = {};
+relaxed_list<Node> * relaxed_list<Node>::head = nullptr;
+#ifndef NO_STATS
+template<>
+relaxed_list<Node>::GlobalStats relaxed_list<Node>::global_stats = {};
+#endif
 // ================================================================================================
 …
         size_t crc_in  = 0;
         size_t crc_out = 0;
+        size_t valmax = 0;
+        size_t valmin = 100000000ul;
 };
 …
         std::atomic_size_t crc_in  = { 0 };
         std::atomic_size_t crc_out = { 0 };
+        struct {
+                struct {
+                        std::atomic_size_t attempt = { 0 };
+                        std::atomic_size_t success = { 0 };
+                } push;
+                struct {
+                        std::atomic_size_t attempt = { 0 };
+                        std::atomic_size_t success = { 0 };
+                        std::atomic_size_t mask_attempt = { 0 };
+                } pop;
+        } pick;
+        struct {
+                struct {
+                        std::atomic_size_t value = { 0 };
+                        std::atomic_size_t count = { 0 };
+                } push;
+                struct {
+                        std::atomic_size_t value = { 0 };
+                        std::atomic_size_t count = { 0 };
+                } pop;
+        } qstat;
+        std::atomic_size_t valmax = { 0 };
+        std::atomic_size_t valmin = { 100000000ul };
 };
+void atomic_max(std::atomic_size_t & target, size_t value) {
+        for(;;) {
+                size_t expect = target.load(std::memory_order_relaxed);
+                if(value <= expect) return;
+                bool success = target.compare_exchange_strong(expect, value);
+                if(success) return;
+        }
+}
+void atomic_min(std::atomic_size_t & target, size_t value) {
+        for(;;) {
+                size_t expect = target.load(std::memory_order_relaxed);
+                if(value >= expect) return;
+                bool success = target.compare_exchange_strong(expect, value);
+                if(success) return;
+        }
+}
 void tally_stats(global_stat_t & global, local_stat_t & local) {
         global.in    += local.in;
         global.out   += local.out;
 …
         global.crc_out += local.crc_out;
+        global.pick.push.attempt += relaxed_list<Node>::tls.pick.push.attempt;
+        global.pick.push.success += relaxed_list<Node>::tls.pick.push.success;
+        global.pick.pop .attempt += relaxed_list<Node>::tls.pick.pop.attempt;
+        global.pick.pop .success += relaxed_list<Node>::tls.pick.pop.success;
+        global.pick.pop .mask_attempt += relaxed_list<Node>::tls.pick.pop.mask_attempt;
+        global.qstat.push.value += relaxed_list<Node>::tls.empty.push.value;
+        global.qstat.push.count += relaxed_list<Node>::tls.empty.push.count;
+        global.qstat.pop .value += relaxed_list<Node>::tls.empty.pop .value;
+        global.qstat.pop .count += relaxed_list<Node>::tls.empty.pop .count;
+        atomic_max(global.valmax, local.valmax);
+        atomic_min(global.valmin, local.valmin);
+        relaxed_list<Node>::stats_tls_tally();
+}
 …
+}
+void waitfor(double & duration, barrier_t & barrier, const std::atomic_size_t & count) {
+        std::cout << "Starting" << std::endl;
+        auto before = Clock::now();
+        barrier.wait(0);
+        while(true) {
+                usleep(100000);
+                size_t c = count.load();
+                if( c == 0 ) {
+                        break;
+                }
+                std::cout << "\r" << c;
+                std::cout.flush();
+        }
+        barrier.wait(0);
+        auto after = Clock::now();
+        duration_t durr = after - before;
+        duration = durr.count();
+        std::cout << "\rClosing down" << std::endl;
+}
 void print_stats(double duration, unsigned nthread, global_stat_t & global) {
         assert(Node::creates == Node::destroys);
 …
         std::cout << "Ops/sec       : " << ops_sec << "\n";
         std::cout << "Total ops     : " << ops << "(" << global.in << "i, " << global.out << "o, " << global.empty << "e)\n";
+        if(global.valmax != 0) {
+                std::cout << "Max runs      : " << global.valmax << "\n";
+                std::cout << "Min runs      : " << global.valmin << "\n";
+        }
         #ifndef NO_STATS
+                double push_sur = (100.0 * double(global.pick.push.success) / global.pick.push.attempt);
+                double pop_sur  = (100.0 * double(global.pick.pop .success) / global.pick.pop .attempt);
+                std::cout << "Push Pick %   : " << push_sur << "(" << global.pick.push.success << " / " << global.pick.push.attempt << ")\n";
+                std::cout << "Pop  Pick %   : " << pop_sur  << "(" << global.pick.pop .success << " / " << global.pick.pop .attempt << ")\n";
+                std::cout << "Pop mask trys : " << global.pick.pop.mask_attempt << std::endl;
+                double avgQ_push = double(global.qstat.push.value) / global.qstat.push.count;
+                double avgQ_pop  = double(global.qstat.pop .value) / global.qstat.pop .count;
+                double avgQ      = double(global.qstat.push.value + global.qstat.pop .value) / (global.qstat.push.count + global.qstat.pop .count);
+                std::cout << "Push   Avg Qs : " << avgQ_push << " (" << global.qstat.push.count << "ops)\n";
+                std::cout << "Pop    Avg Qs : " << avgQ_pop  << " (" << global.qstat.pop .count << "ops)\n";
+                std::cout << "Global Avg Qs : " << avgQ      << " (" << (global.qstat.push.count + global.qstat.pop .count) << "ops)\n";
+                relaxed_list<Node>::stats_print(std::cout);
         #endif
+}
+void save_fairness(const int data[], int factor, unsigned nthreads, size_t columns, size_t rows, const std::string & output);
 // ================================================================================================
 …
         assert(nnodes <= nslots);
         // List being tested
-        relaxed_list<Node> list = { nthread * nqueues };
         // Barrier for synchronization
 …
         std::cout << "Initializing ";
         size_t npushed = 0;
+        Node** all_nodes[nthread];
+        for(auto & nodes : all_nodes) {
+                nodes = new __attribute__((aligned(64))) Node*[nslots + 8];
+                Random rand(rdtscl());
+                for(unsigned i = 0; i < nnodes; i++) {
+                        nodes[i] = new Node(rand.next() % 100);
+                }
+                for(unsigned i = nnodes; i < nslots; i++) {
+                        nodes[i] = nullptr;
+                }
+                for(int i = 0; i < 10 && i < (int)nslots; i++) {
+                        int idx = rand.next() % nslots;
+                        if (auto node = nodes[idx]) {
+                                global.crc_in += node->value;
+                                list.push(node);
+                                npushed++;
+                                nodes[idx] = nullptr;
+        relaxed_list<Node> list = { nthread * nqueues };
+        {
+                Node** all_nodes[nthread];
+                for(auto & nodes : all_nodes) {
+                        nodes = new __attribute__((aligned(64))) Node*[nslots + 8];
+                        Random rand(rdtscl());
+                        for(unsigned i = 0; i < nnodes; i++) {
+                                nodes[i] = new Node(rand.next() % 100);
+                        }
+                }
+        }
+        std::cout << nnodes << " nodes (" << nslots << " slots)" << std::endl;
+        enable_stats = true;
+        std::thread * threads[nthread];
+        unsigned i = 1;
+        for(auto & t : threads) {
+                auto & my_nodes = all_nodes[i - 1];
+                t = new std::thread([&done, &list, &barrier, &global, &my_nodes, nslots](unsigned tid) {
+                        Random rand(tid + rdtscl());
+                        local_stat_t local;
+                        // affinity(tid);
+                        barrier.wait(tid);
+                        // EXPERIMENT START
+                        runChurn_body(done, rand, my_nodes, nslots, local, list);
+                        // EXPERIMENT END
+                        barrier.wait(tid);
+                        tally_stats(global, local);
+                        for(unsigned i = 0; i < nslots; i++) {
+                                delete my_nodes[i];
+                        for(unsigned i = nnodes; i < nslots; i++) {
+                                nodes[i] = nullptr;
+                        }
+                }, i++);
+        }
+        waitfor(duration, barrier, done);
+        for(auto t : threads) {
+                t->join();
+                delete t;
+        }
+        enable_stats = false;
+        while(auto node = list.pop()) {
+                global.crc_out += node->value;
+                delete node;
+        }
+        for(auto nodes : all_nodes) {
+                delete[] nodes;
+                        for(int i = 0; i < 10 && i < (int)nslots; i++) {
+                                int idx = rand.next() % nslots;
+                                if (auto node = nodes[idx]) {
+                                        global.crc_in += node->value;
+                                        list.push(node);
+                                        npushed++;
+                                        nodes[idx] = nullptr;
+                                }
+                        }
+                }
+                std::cout << nnodes << " nodes (" << nslots << " slots)" << std::endl;
+                enable_stats = true;
+                std::thread * threads[nthread];
+                unsigned i = 1;
+                for(auto & t : threads) {
+                        auto & my_nodes = all_nodes[i - 1];
+                        t = new std::thread([&done, &list, &barrier, &global, &my_nodes, nslots](unsigned tid) {
+                                Random rand(tid + rdtscl());
+                                local_stat_t local;
+                                // affinity(tid);
+                                barrier.wait(tid);
+                                // EXPERIMENT START
+                                runChurn_body(done, rand, my_nodes, nslots, local, list);
+                                // EXPERIMENT END
+                                barrier.wait(tid);
+                                tally_stats(global, local);
+                                for(unsigned i = 0; i < nslots; i++) {
+                                        delete my_nodes[i];
+                                }
+                        }, i++);
+                }
+                waitfor(duration, barrier, done);
+                for(auto t : threads) {
+                        t->join();
+                        delete t;
+                }
+                enable_stats = false;
+                while(auto node = list.pop()) {
+                        global.crc_out += node->value;
+                        delete node;
+                }
+                for(auto nodes : all_nodes) {
+                        delete[] nodes;
+                }
+        }
 …
         std::cout << "PingPong Benchmark" << std::endl;
+        // Barrier for synchronization
+        barrier_t barrier(nthread + 1);
+        // Data to check everything is OK
+        global_stat_t global;
+        // Flag to signal termination
+        std::atomic_bool done  = { false };
+        std::cout << "Initializing ";
         // List being tested
         relaxed_list<Node> list = { nthread * nqueues };
+        {
+                enable_stats = true;
+                std::thread * threads[nthread];
+                unsigned i = 1;
+                for(auto & t : threads) {
+                        t = new std::thread([&done, &list, &barrier, &global, nnodes](unsigned tid) {
+                                Random rand(tid + rdtscl());
+                                Node nodes[nnodes];
+                                for(auto & n : nodes) {
+                                        n.value = (int)rand.next() % 100;
+                                }
+                                local_stat_t local;
+                                // affinity(tid);
+                                barrier.wait(tid);
+                                // EXPERIMENT START
+                                runPingPong_body(done, nodes, nnodes, local, list);
+                                // EXPERIMENT END
+                                barrier.wait(tid);
+                                tally_stats(global, local);
+                        }, i++);
+                }
+                waitfor(duration, barrier, done);
+                for(auto t : threads) {
+                        t->join();
+                        delete t;
+                }
+                enable_stats = false;
+        }
+        print_stats(duration, nthread, global);
+}
+// ================================================================================================
+__attribute__((noinline)) void runFairness_body(
+        unsigned tid,
+        size_t width,
+        size_t length,
+        int output[],
+        std::atomic_size_t & count,
+        Node initial_nodes[],
+        unsigned nnodes,
+        local_stat_t & local,
+        relaxed_list<Node> & list
+) {
+        Node * nodes[nnodes];
+        {
+                unsigned i = 0;
+                for(auto & n : nodes) {
+                        n = &initial_nodes[i++];
+                }
+        }
+        while(__builtin_expect(0 != count.load(std::memory_order_relaxed), true)) {
+                for(Node * & node : nodes) {
+                        local.crc_in += node->id;
+                        list.push(node);
+                        local.in++;
+                }
+                // -----
+                for(Node * & node : nodes) {
+                        node = list.pop();
+                        assert(node);
+                        if (unsigned(node->value) < length) {
+                                size_t idx = (node->value * width) + node->id;
+                                assert(idx < (width * length));
+                                output[idx] = tid;
+                        }
+                        node->value++;
+                        if(unsigned(node->value) == length) count--;
+                        local.crc_out += node->id;
+                        local.out++;
+                }
+        }
+}
+void runFairness(unsigned nthread, unsigned nqueues, double duration, unsigned nnodes, const std::string & output) {
+        std::cout << "Fairness Benchmark, outputing to : " << output << std::endl;
         // Barrier for synchronization
 …
         global_stat_t global;
+        std::cout << "Initializing ";
+        // Check fairness by creating a png of where the threads ran
+        size_t width = nthread * nnodes;
+        size_t length = 100000;
+        std::unique_ptr<int[]> data_out { new int[width * length] };
         // Flag to signal termination
+        std::atomic_bool done  = { false };
+        std::cout << "Initializing ";
+        enable_stats = true;
+        std::thread * threads[nthread];
+        unsigned i = 1;
+        for(auto & t : threads) {
+                t = new std::thread([&done, &list, &barrier, &global, nnodes](unsigned tid) {
+                        Random rand(tid + rdtscl());
+                        Node nodes[nnodes];
+                        for(auto & n : nodes) {
+                                n.value = (int)rand.next() % 100;
+                        }
+                        local_stat_t local;
+                        // affinity(tid);
+                        barrier.wait(tid);
+                        // EXPERIMENT START
+                        runPingPong_body(done, nodes, nnodes, local, list);
+                        // EXPERIMENT END
+                        barrier.wait(tid);
+                        tally_stats(global, local);
+                }, i++);
+        }
+        waitfor(duration, barrier, done);
+        for(auto t : threads) {
+                t->join();
+                delete t;
+        }
+        enable_stats = false;
+        std::atomic_size_t count = width;
+        // List being tested
+        relaxed_list<Node> list = { nthread * nqueues };
+        {
+                enable_stats = true;
+                std::thread * threads[nthread];
+                unsigned i = 1;
+                for(auto & t : threads) {
+                        t = new std::thread([&count, &list, &barrier, &global, nnodes, width, length, data_out = data_out.get()](unsigned tid) {
+                                unsigned int start = (tid - 1) * nnodes;
+                                Node nodes[nnodes];
+                                for(auto & n : nodes) {
+                                        n.id = start;
+                                        n.value = 0;
+                                        start++;
+                                }
+                                local_stat_t local;
+                                // affinity(tid);
+                                barrier.wait(tid);
+                                // EXPERIMENT START
+                                runFairness_body(tid, width, length, data_out, count, nodes, nnodes, local, list);
+                                // EXPERIMENT END
+                                barrier.wait(tid);
+                                for(const auto & n : nodes) {
+                                        local.valmax = max(local.valmax, size_t(n.value));
+                                        local.valmin = min(local.valmin, size_t(n.value));
+                                }
+                                tally_stats(global, local);
+                        }, i++);
+                }
+                waitfor(duration, barrier, count);
+                for(auto t : threads) {
+                        t->join();
+                        delete t;
+                }
+                enable_stats = false;
+        }
         print_stats(duration, nthread, global);
+}
+        save_fairness(data_out.get(), 100, nthread, width, length, output);
+}
+// ================================================================================================
 bool iequals(const std::string& a, const std::string& b)
 …
         unsigned nnodes   = 100;
         unsigned nslots   = 100;
+        std::string out   = "fairness.png";
         enum {
                 Churn,
                 PingPong,
+                Fairness,
                 NONE
         } benchmark = NONE;
 …
+                                        }
                                         break;
+                                case Fairness:
+                                        nnodes = 1;
+                                        switch(argc - optind) {
+                                        case 0: break;
+                                        case 1:
+                                                arg = optarg = argv[optind];
+                                                out = arg;
+                                                break;
+                                        default:
+                                                std::cerr << "'Churn' benchmark doesn't accept more than 2 extra arguments" << std::endl;
+                                                goto usage;
+                                        }
+                                }
                                 goto run;
 …
                                 if(iequals(arg, "pingpong")) {
                                         benchmark = PingPong;
+                                        break;
+                                }
+                                if(iequals(arg, "fairness")) {
+                                        benchmark = Fairness;
                                         break;
+                                }
 …
                         runPingPong(nthreads, nqueues, duration, nnodes);
                         break;
+                case Fairness:
+                        runFairness(nthreads, nqueues, duration, nnodes, out);
+                        break;
                 default:
                         abort();
 …
 const char * __my_progname = "Relaxed List";
+struct rgb_t {
+    double r;       // a fraction between 0 and 1
+    double g;       // a fraction between 0 and 1
+    double b;       // a fraction between 0 and 1
+};
+struct hsv_t {
+    double h;       // angle in degrees
+    double s;       // a fraction between 0 and 1
+    double v;       // a fraction between 0 and 1
+};
+rgb_t hsv2rgb(hsv_t in) {
+        double hh, p, q, t, ff;
+        long   i;
+        rgb_t  out;
+        if(in.s <= 0.0) {       // < is bogus, just shuts up warnings
+                out.r = in.v;
+                out.g = in.v;
+                out.b = in.v;
+                return out;
+        }
+        hh = in.h;
+        if(hh >= 360.0) hh = 0.0;
+        hh /= 60.0;
+        i = (long)hh;
+        ff = hh - i;
+        p = in.v * (1.0 - in.s);
+        q = in.v * (1.0 - (in.s * ff));
+        t = in.v * (1.0 - (in.s * (1.0 - ff)));
+        switch(i) {
+        case 0:
+                out.r = in.v;
+                out.g = t;
+                out.b = p;
+                break;
+        case 1:
+                out.r = q;
+                out.g = in.v;
+                out.b = p;
+                break;
+        case 2:
+                out.r = p;
+                out.g = in.v;
+                out.b = t;
+                break;
+        case 3:
+                out.r = p;
+                out.g = q;
+                out.b = in.v;
+                break;
+        case 4:
+                out.r = t;
+                out.g = p;
+                out.b = in.v;
+                break;
+        case 5:
+        default:
+                out.r = in.v;
+                out.g = p;
+                out.b = q;
+                break;
+        }
+        return out;
+}
+void save_fairness(const int data[], int factor, unsigned nthreads, size_t columns, size_t rows, const std::string & output) {
+        std::ofstream os(output);
+        os << "<html>\n";
+        os << "<head>\n";
+        os << "<style>\n";
+        os << "</style>\n";
+        os << "</head>\n";
+        os << "<body>\n";
+        os << "<table style=\"width=100%\">\n";
+        size_t idx = 0;
+        for(size_t r = 0ul; r < rows; r++) {
+                os << "<tr>\n";
+                for(size_t c = 0ul; c < columns; c++) {
+                        os << "<td class=\"custom custom" << data[idx] << "\"></td>\n";
+                        idx++;
+                }
+                os << "</tr>\n";
+        }
+        os << "</table>\n";
+        os << "</body>\n";
+        os << "</html>\n";
+        os << std::endl;
+}
+#include <png.h>
+#include <setjmp.h>
+/*
+void save_fairness(const int data[], int factor, unsigned nthreads, size_t columns, size_t rows, const std::string & output) {
+        int width  = columns * factor;
+        int height = rows / factor;
+        int code = 0;
+        int idx = 0;
+        FILE *fp = NULL;
+        png_structp png_ptr = NULL;
+        png_infop info_ptr = NULL;
+        png_bytep row = NULL;
+        // Open file for writing (binary mode)
+        fp = fopen(output.c_str(), "wb");
+        if (fp == NULL) {
+                fprintf(stderr, "Could not open file %s for writing\n", output.c_str());
+                code = 1;
+                goto finalise;
+        }
+           // Initialize write structure
+        png_ptr = png_create_write_struct(PNG_LIBPNG_VER_STRING, NULL, NULL, NULL);
+        if (png_ptr == NULL) {
+                fprintf(stderr, "Could not allocate write struct\n");
+                code = 1;
+                goto finalise;
+        }
+        // Initialize info structure
+        info_ptr = png_create_info_struct(png_ptr);
+        if (info_ptr == NULL) {
+                fprintf(stderr, "Could not allocate info struct\n");
+                code = 1;
+                goto finalise;
+        }
+        // Setup Exception handling
+        if (setjmp(png_jmpbuf(png_ptr))) {
+                fprintf(stderr, "Error during png creation\n");
+                code = 1;
+                goto finalise;
+        }
+        png_init_io(png_ptr, fp);
+        // Write header (8 bit colour depth)
+        png_set_IHDR(png_ptr, info_ptr, width, height,
+, PNG_COLOR_TYPE_RGB, PNG_INTERLACE_NONE,
+                PNG_COMPRESSION_TYPE_BASE, PNG_FILTER_TYPE_BASE);
+        png_write_info(png_ptr, info_ptr);
+        // Allocate memory for one row (3 bytes per pixel - RGB)
+        row = (png_bytep) malloc(3 * width * sizeof(png_byte));
+        // Write image data
+        int x, y;
+        for (y=0 ; y<height ; y++) {
+                for (x=0 ; x<width ; x++) {
+                        auto & r = row[(x * 3) + 0];
+                        auto & g = row[(x * 3) + 1];
+                        auto & b = row[(x * 3) + 2];
+                        assert(idx < (rows * columns));
+                        int color = data[idx] - 1;
+                        assert(color < nthreads);
+                        assert(color >= 0);
+                        idx++;
+                        double angle = double(color) / double(nthreads);
+                        auto c = hsv2rgb({ 360.0 * angle, 0.8, 0.8 });
+                        r = char(c.r * 255.0);
+                        g = char(c.g * 255.0);
+                        b = char(c.b * 255.0);
+                }
+                png_write_row(png_ptr, row);
+        }
+        assert(idx == (rows * columns));
+        // End write
+        png_write_end(png_ptr, NULL);
+        finalise:
+        if (fp != NULL) fclose(fp);
+        if (info_ptr != NULL) png_free_data(png_ptr, info_ptr, PNG_FREE_ALL, -1);
+        if (png_ptr != NULL) png_destroy_write_struct(&png_ptr, (png_infopp)NULL);
+        if (row != NULL) free(row);
+}
+*/

Note: See TracChangeset for help on using the changeset viewer.

Context Navigation

Changeset 807a632 for doc/theses/thierry_delisle_PhD/code/relaxed_list.cpp

Legend:

doc/theses/thierry_delisle_PhD/code/relaxed_list.cpp

Download in other formats: