parallel/benchmark.cpp

配套源码，运行方法见同目录 README。返回实验总览。

// Original scheduling benchmark: uneven independent jobs, exact comparison.
#include <algorithm>
#include <atomic>
#include <chrono>
#include <cstdint>
#include <cstdlib>
#include <iomanip>
#include <iostream>
#include <stdexcept>
#include <string>
#include <thread>
#include <vector>

using Clock = std::chrono::steady_clock;
using Values = std::vector<std::uint64_t>;

static std::size_t argument(const char *text, std::size_t maximum) {
    std::string value(text);
    if (value.empty() || value.find_first_not_of("0123456789") != std::string::npos)
        throw std::invalid_argument("arguments must be positive decimal integers");
    std::size_t used = 0;
    auto n = std::stoull(value, &used);
    if (used != value.size() || n == 0 || n > maximum)
        throw std::invalid_argument("argument outside supported range");
    return static_cast<std::size_t>(n);
}

static std::uint64_t work(std::size_t i, std::size_t n) {
    // The first quarter is deliberately heavier: static contiguous chunks
    // give one worker much more work even though item counts are equal.
    unsigned steps = i < n / 4 ? 512 : 16;
    std::uint64_t x = static_cast<std::uint64_t>(i) + UINT64_C(0x9e3779b97f4a7c15);
    for (unsigned k = 0; k < steps; ++k) {
        x ^= x >> 12;
        x ^= x << 25;
        x ^= x >> 27;
        x *= UINT64_C(2685821657736338717);
    }
    return x;
}

static void range(Values &output, std::size_t begin, std::size_t end) {
    for (std::size_t i = begin; i < end; ++i) output[i] = work(i, output.size());
}

static void execute(Values &output, const std::string &policy,
                    std::size_t workers, std::size_t grain) {
    if (policy == "serial") { range(output, 0, output.size()); return; }
    std::atomic<std::size_t> next{0};
    std::vector<std::thread> threads;
    threads.reserve(workers);
    try {
        for (std::size_t id = 0; id < workers; ++id) {
            threads.emplace_back([&, id] {
                if (policy == "static") {
                    range(output, output.size() * id / workers,
                          output.size() * (id + 1) / workers);
                } else {
                    for (;;) {
                        std::size_t begin = next.fetch_add(grain, std::memory_order_relaxed);
                        if (begin >= output.size()) break;
                        range(output, begin, std::min(begin + grain, output.size()));
                    }
                }
            });
        }
    } catch (...) {
        for (auto &thread : threads) thread.join();
        throw;
    }
    for (auto &thread : threads) thread.join();
}

static std::uint64_t checksum(const Values &values) {
    std::uint64_t hash = UINT64_C(1469598103934665603);
    for (std::uint64_t value : values) { hash ^= value; hash *= UINT64_C(1099511628211); }
    return hash;
}

static void measure(const Values &expected, const std::string &policy,
                    std::size_t workers, std::size_t grain, std::size_t repeats) {
    Values output(expected.size());
    execute(output, policy, workers, grain); // One untimed warmup.
    if (output != expected) throw std::runtime_error("warmup mismatch: " + policy);
    std::vector<double> times;
    for (std::size_t trial = 0; trial < repeats; ++trial) {
        std::fill(output.begin(), output.end(), 0); // Not timed.
        const auto begin = Clock::now();
        execute(output, policy, workers, grain); // Includes thread creation/join.
        const auto end = Clock::now();
        if (output != expected) throw std::runtime_error("result mismatch: " + policy);
        times.push_back(std::chrono::duration<double, std::milli>(end - begin).count());
    }
    std::sort(times.begin(), times.end());
    double median = times[times.size()/2];
    if (times.size() % 2 == 0) median = (times[times.size()/2-1] + median) / 2;
    std::cout << policy << ',' << workers << ',' << grain << ',' << expected.size() << ','
              << repeats << ',' << times.front() << ',' << median << ',' << times.back()
              << ',' << checksum(output) << '\n';
}

int main(int argc, char **argv) {
    try {
        if (argc > 5) throw std::invalid_argument("usage: benchmark [items [threads [grain [repeats]]]]");
        std::size_t count = argc > 1 ? argument(argv[1], 5000000) : 200000;
        std::size_t workers = argc > 2 ? argument(argv[2], 128) : 4;
        std::size_t grain = argc > 3 ? argument(argv[3], 5000000) : 64;
        std::size_t repeats = argc > 4 ? argument(argv[4], 31) : 5;
        Values expected(count);
        range(expected, 0, count);
        std::cerr << "compiler=" << __VERSION__ << " hardware_concurrency_hint="
                  << std::thread::hardware_concurrency() << '\n';
        std::cout << "policy,threads,grain,items,repeats,min_ms,median_ms,max_ms,checksum\n";
        std::cout << std::fixed << std::setprecision(6);
        measure(expected, "serial", 1, count, repeats);
        measure(expected, "static", workers, grain, repeats);
        measure(expected, "dynamic", workers, grain, repeats);
        std::cerr << "all output elements matched serial reference\n";
    } catch (const std::exception &error) {
        std::cerr << error.what() << '\n';
        return 1;
    }
}