From 51287faaac6989386ccd06b60a3edfdc05feacc6 Mon Sep 17 00:00:00 2001 From: chenBright Date: Tue, 22 Sep 2026 23:46:55 +0800 Subject: [PATCH 1/2] Support higher performance histogram --- BUILD.bazel | 5 +- src/bvar/detail/sampler.h | 35 ++++--- src/bvar/histogram.cpp | 16 +++ src/bvar/histogram.h | 149 +++++++++++++++++++++++++-- src/bvar/passive_status.h | 7 +- test/bvar_histogram_unittest.cpp | 167 +++++++++++++++++++++++++++++++ 6 files changed, 350 insertions(+), 29 deletions(-) diff --git a/BUILD.bazel b/BUILD.bazel index db73605e6e..23e433c76e 100644 --- a/BUILD.bazel +++ b/BUILD.bazel @@ -424,7 +424,10 @@ cc_library( deps = [ ":butil", ] + select({ - "//bazel/config:with_babylon_counter": ["@babylon//:concurrent_counter"], + "//bazel/config:with_babylon_counter": [ + "@babylon//:concurrent_counter", + "@babylon//:concurrent_thread_local", + ], "//conditions:default": [], }), ) diff --git a/src/bvar/detail/sampler.h b/src/bvar/detail/sampler.h index 06533e4e0c..7c92413d51 100644 --- a/src/bvar/detail/sampler.h +++ b/src/bvar/detail/sampler.h @@ -198,7 +198,7 @@ class ReducerSampler : public Sampler { // would not be ignored take_sample(); } - ~ReducerSampler() {} + ~ReducerSampler() override = default; void take_sample() override { // Make _q ready. @@ -222,21 +222,7 @@ class ReducerSampler : public Sampler { } Sample latest; - if (butil::is_same::value) { - // The operator can't be inversed. - // We reset the reducer and save the result as a sample. - // Suming up samples gives the result within a window. - // In this case, get_value() of _reducer gives wrong answer and - // should not be called. - latest.data = _source.reset(); - } else { - // The operator can be inversed. - // We save the result as a sample. - // Inversed operation between latest and oldest sample within a - // window gives result. - // get_value() of _reducer can still be called. - latest.data = _source.get_value(); - } + latest.data = take_sample_of(butil::is_same()); latest.time_us = butil::cpuwide_time_us(); _q.elim_push(latest); } @@ -313,6 +299,23 @@ class ReducerSampler : public Sampler { } private: + // Tag dispatch instead of a runtime branch on is_same, so + // that only the taken branch is instantiated. + + // The operator can't be inversed. + // We reset the reducer and save the result as a sample. + // Suming up samples gives the result within a window. + // In this case, get_value() of `_source` gives wrong answer and + // should not be called. + T take_sample_of(butil::true_type) { return _source.reset(); } + + // The operator can be inversed. + // We save the result as a sample. + // Inversed operation between latest and oldest sample within a + // window gives result. + // get_value() of `_source` can still be called. + T take_sample_of(butil::false_type) { return _source.get_value(); } + source_type _source; time_t _window_size; butil::BoundedQueue > _q; diff --git a/src/bvar/histogram.cpp b/src/bvar/histogram.cpp index 2288bd2493..af4fcb69d7 100644 --- a/src/bvar/histogram.cpp +++ b/src/bvar/histogram.cpp @@ -100,10 +100,14 @@ std::ostream& operator<<(std::ostream& os, const Histogram::Value& v) { Histogram::Histogram(const BucketSchema& schema) : _schema(schema) +#if WITH_BABYLON_COUNTER + , _storage(std::make_shared(schema.num_buckets())) +#else // Both identities carry `num_buckets` so that a value combined out of no // agent at all still knows how wide it is. , _combiner(std::make_shared(value_type(schema.num_buckets()), value_type(schema.num_buckets()))) +#endif // WITH_BABYLON_COUNTER , _sampler(nullptr) { } @@ -133,6 +137,9 @@ Histogram& Histogram::operator<<(double value) { << " recorded into Histogram(" << name() << ')'; return *this; } +#if WITH_BABYLON_COUNTER + _storage->add(_schema.index_of(value), value); +#else agent_type* agent = _combiner->get_or_create_tls_agent(); if (BAIDU_UNLIKELY(agent == nullptr)) { LOG(FATAL) << "Fail to create agent"; @@ -140,9 +147,18 @@ Histogram& Histogram::operator<<(double value) { } // `_schema` outlives the call, the op only borrows it to find the bucket. agent->element.modify(detail::AddSampleToHistogram(&_schema), value); +#endif // WITH_BABYLON_COUNTER return *this; } +Histogram::value_type Histogram::get_value() const { +#if WITH_BABYLON_COUNTER + return _storage->combine_agents(); +#else + return _combiner->combine_agents(); +#endif // WITH_BABYLON_COUNTER +} + Histogram::sampler_type* Histogram::get_sampler() { if (_sampler == nullptr) { _sampler = new sampler_type(this); diff --git a/src/bvar/histogram.h b/src/bvar/histogram.h index d262384196..27dd5cd1cd 100644 --- a/src/bvar/histogram.h +++ b/src/bvar/histogram.h @@ -19,15 +19,22 @@ #define BVAR_HISTOGRAM_H #include // int64_t, uint64_t -#include // std::lower_bound +#include // std::max #include // std::initializer_list +#include // std::shared_ptr #include // std::string #include // std::vector #include "butil/strings/string_piece.h" // butil::StringPiece #include "bvar/variable.h" // Variable -#include "bvar/detail/combiner.h" // AgentCombiner #include "bvar/detail/sampler.h" // ReducerSampler #include "bvar/detail/series.h" // HasPlottableSeries +#if WITH_BABYLON_COUNTER +#include "babylon/concurrent/thread_local.h" // EnumerableThreadLocal +#include "butil/atomicops.h" // butil::atomic +#include "butil/synchronization/seqlock.h" // butil::Seqlock +#else +#include "bvar/detail/combiner.h" // AgentCombiner +#endif // WITH_BABYLON_COUNTER namespace bvar { @@ -36,6 +43,13 @@ namespace bvar { // bounds. static const size_t MAX_HISTOGRAM_BUCKETS = 32; +#if WITH_BABYLON_COUNTER +namespace detail { +// Defined below, once Histogram::Value is complete. +class HistogramStorage; +} // namespace detail +#endif // WITH_BABYLON_COUNTER + // Bucketed distribution of the recorded values. // // Each bucket count accumulates since construction and never decreases. The @@ -79,9 +93,17 @@ class Histogram : public Variable { BucketSchema(std::initializer_list bounds); explicit BucketSchema(const std::vector& bounds); + // Linear rather than a std::lower_bound: a schema holds at most + // MAX_HISTOGRAM_BUCKETS - 1 bounds, which is a couple of cache lines + // scanned straight through instead of jumped around in, and the + // branch of the scan predicts far better than the one of a binary + // search, whose direction is a coin flip at every step. size_t index_of(double value) const { - return std::lower_bound(_bounds.begin(), _bounds.end(), value) - - _bounds.begin(); + size_t index = 0; + while (index < _bounds.size() && _bounds[index] < value) { + ++index; + } + return index; } size_t num_buckets() const { return _bounds.size() + 1; } @@ -148,9 +170,13 @@ class Histogram : public Variable { typedef Value value_type; typedef detail::ReducerSampler sampler_type; +#if WITH_BABYLON_COUNTER + typedef std::shared_ptr shared_combiner_type; +#else typedef detail::AgentCombiner combiner_type; typedef combiner_type::self_shared_type shared_combiner_type; typedef combiner_type::Agent agent_type; +#endif // WITH_BABYLON_COUNTER explicit Histogram(const BucketSchema& schema); Histogram(const butil::StringPiece& name, const BucketSchema& schema); @@ -168,7 +194,11 @@ class Histogram : public Variable { const BucketSchema& schema() const { return _schema; } bool valid() const { +#if WITH_BABYLON_COUNTER + return _storage != nullptr; +#else return _combiner != nullptr && _combiner->valid(); +#endif // WITH_BABYLON_COUNTER } void describe(std::ostream& os, bool quote_string) const override; @@ -192,7 +222,13 @@ class Histogram : public Variable { // Expose the shared data carrier, so that ReducerSampler holds it instead // of `this`. Sampling then keeps reading valid memory even if this // Percentile is destructed before the sampler is recycled. - shared_combiner_type share_combiner() const { return _combiner; } + shared_combiner_type share_combiner() const { +#if WITH_BABYLON_COUNTER + return _storage; +#else + return _combiner; +#endif // WITH_BABYLON_COUNTER + } sampler_type* get_sampler(); private: @@ -202,10 +238,14 @@ class Histogram : public Variable { // Snapshot of all the values recorded so far. Walks through every thread // that ever recorded into this Histogram. - value_type get_value() const { return _combiner->combine_agents(); } + value_type get_value() const; BucketSchema _schema; +#if WITH_BABYLON_COUNTER + shared_combiner_type _storage; +#else shared_combiner_type _combiner; +#endif // WITH_BABYLON_COUNTER sampler_type* _sampler; }; @@ -219,6 +259,101 @@ namespace detail { template <> struct HasPlottableSeries : butil::false_type {}; +#if WITH_BABYLON_COUNTER + +// One thread's slice of a Histogram. +// +// Only the thread owning the slot writes it, so the counters are updated with +// a relaxed load plus a relaxed store rather than an atomic read-modify-write. +// They are atomic all the same because the sampling thread reads them while +// they are being written, which the seqlock allows but does not by itself make +// race free. The seqlock is what keeps the buckets, the sum and the count of one +// slot mutually consistent. +class HistogramSlot { +public: + HistogramSlot() { + for (size_t i = 0; i < MAX_HISTOGRAM_BUCKETS; ++i) { + _counts[i].store(0, butil::memory_order_relaxed); + } + } + + DISALLOW_COPY_AND_ASSIGN(HistogramSlot); + + void add(size_t bucket_index, double value) { + _seqlock.store([&] { + relaxed_add(&_counts[bucket_index], (uint64_t)1); + relaxed_add(&_sum, value); + relaxed_add(&_num, (int64_t)1); + }); + } + + Histogram::Value load(size_t num_buckets) const { + return _seqlock.load([&] { + Histogram::Value v(num_buckets); + for (size_t i = 0; i < num_buckets; ++i) { + v.counts[i] = _counts[i].load(butil::memory_order_relaxed); + } + v.sum = _sum.load(butil::memory_order_relaxed); + v.num = _num.load(butil::memory_order_relaxed); + return v; + }); + } + +private: + template + static void relaxed_add(butil::atomic* target, U delta) { + target->store(target->load(butil::memory_order_relaxed) + delta, + butil::memory_order_relaxed); + } + + butil::Seqlock<> _seqlock; + butil::atomic _counts[MAX_HISTOGRAM_BUCKETS]; + butil::atomic _sum{0.0}; + butil::atomic _num{0}; +}; + +// The per thread slices of one Histogram and their aggregation. +// +// babylon's EnumerableThreadLocal hands out one slot per thread and walks every +// slot any thread ever took, which is how a thread that has exited keeps +// contributing what it recorded, the way AgentCombiner commits a dying agent +// into its global result. babylon recycles the thread id of an exited thread, +// so a later thread inherits the slot and accumulates on top of it: correct +// here because a Histogram only ever adds to its buckets and never clears one. +class HistogramStorage { +public: + explicit HistogramStorage(size_t num_buckets) : _num_buckets(num_buckets) {} + + DISALLOW_COPY_AND_ASSIGN(HistogramStorage); + + // Records one value into the slot of the calling thread. + void add(size_t bucket_index, double value) { + _slots.local().add(bucket_index, value); + } + + // [Threadsafe] Everything recorded so far by every thread that ever + // recorded into this Histogram. Named after AgentCombiner::combine_agents() + // so that detail::CombinerSampleSource fits either backend. + Histogram::Value combine_agents() const { + Histogram::Value result(_num_buckets); + _slots.for_each([&](const HistogramSlot* iter, const HistogramSlot* end) { + for (; iter != end; ++iter) { + result += iter->load(_num_buckets); + } + }); + return result; + } + +private: + // Leaky: the id allocator behind the thread ids is never destroyed, which + // is what a Histogram of static storage duration needs. The thread ids + // themselves are still recycled when a thread exits. + babylon::EnumerableThreadLocal _slots; + size_t _num_buckets; +}; + +#else + // The op of the writing path takes a recorded value rather than another // Histogram::Value, and needs the schema to find its bucket. struct AddSampleToHistogram { @@ -231,6 +366,8 @@ struct AddSampleToHistogram { const Histogram::BucketSchema* schema; }; +#endif // WITH_BABYLON_COUNTER + } // namespace detail } // namespace bvar diff --git a/src/bvar/passive_status.h b/src/bvar/passive_status.h index 7122e74113..c6252c7134 100644 --- a/src/bvar/passive_status.h +++ b/src/bvar/passive_status.h @@ -102,7 +102,7 @@ class PassiveStatus : public Variable { , _series_sampler(nullptr) { } - ~PassiveStatus() { + ~PassiveStatus() override { hide(); if (_sampler) { _sampler->destroy(); @@ -162,11 +162,6 @@ class PassiveStatus : public Variable { return 0; } - Tp reset() { - CHECK(false) << "PassiveStatus::reset() should never be called, abort"; - abort(); - } - protected: int expose_impl(const butil::StringPiece& prefix, const butil::StringPiece& n, diff --git a/test/bvar_histogram_unittest.cpp b/test/bvar_histogram_unittest.cpp index 80b53c75c3..44ba8c5ccf 100644 --- a/test/bvar_histogram_unittest.cpp +++ b/test/bvar_histogram_unittest.cpp @@ -21,6 +21,7 @@ #include // snprintf #include // memset #include // std::max +#include // std::setprecision #include #include #include // std::make_shared @@ -30,6 +31,7 @@ #include #include #include +#include #include #include #include "bvar/bvar.h" @@ -44,6 +46,54 @@ namespace { // below call private members such as Histogram::get_value() directly. class HistogramTest : public testing::Test {}; +#if WITH_BABYLON_COUNTER + +// One thread's slice of a Histogram, the babylon backed replacement of the +// ElementContainer tested below. +TEST_F(HistogramTest, histogram_slot) { + bvar::Histogram::BucketSchema schema({10, 20, 30}); + bvar::detail::HistogramSlot slot; + // Freshly constructed, before any add(). + bvar::Histogram::Value v = slot.load(schema.num_buckets()); + ASSERT_EQ(0, v.num); + ASSERT_DOUBLE_EQ(0.0, v.sum); + ASSERT_EQ(0u, v.counts[0]); + + slot.add(schema.index_of(5.25), 5.25); + slot.add(schema.index_of(25.5), 25.5); + slot.add(schema.index_of(1000.75), 1000.75); + v = slot.load(schema.num_buckets()); + ASSERT_EQ(3, v.num); + ASSERT_DOUBLE_EQ(1031.5, v.sum); + ASSERT_EQ(4u, v.num_buckets); + ASSERT_EQ(1u, v.counts[0]); // 5.25 -> (-inf, 10] + ASSERT_EQ(0u, v.counts[1]); + ASSERT_EQ(1u, v.counts[2]); // 25.5 -> (20, 30] + ASSERT_EQ(1u, v.counts[3]); // 1000.75 -> +Inf +} + +// A slot is taken lazily, on the first record of a thread, and the storage +// aggregates every slot ever taken. +TEST_F(HistogramTest, histogram_storage) { + bvar::Histogram::BucketSchema schema({10, 20, 30}); + bvar::detail::HistogramStorage storage(schema.num_buckets()); + // No thread has recorded anything, yet the combined value already knows + // how wide the histogram is. + bvar::Histogram::Value v = storage.combine_agents(); + ASSERT_EQ(0, v.num); + ASSERT_EQ(4u, v.num_buckets); + + storage.add(schema.index_of(5.25), 5.25); + storage.add(schema.index_of(25.5), 25.5); + v = storage.combine_agents(); + ASSERT_EQ(2, v.num); + ASSERT_DOUBLE_EQ(30.75, v.sum); + ASSERT_EQ(1u, v.counts[0]); + ASSERT_EQ(1u, v.counts[2]); +} + +#else + // The element container of a Histogram::Value, which is the generic mutex one: // the value is far too wide to be atomical. TEST_F(HistogramTest, element_container) { @@ -76,6 +126,8 @@ TEST_F(HistogramTest, element_container) { ASSERT_EQ(0u, v.counts[3]); } +#endif // WITH_BABYLON_COUNTER + // Fixed workload for export and performance tests, not a library default. static bvar::Histogram::BucketSchema test_latency_schema() { return {10, 20, 40, 80, 160, 320, 640, 1280, 2560, 5120, @@ -724,4 +776,119 @@ TEST_F(HistogramTest, multithreaded) { } ASSERT_DOUBLE_EQ((double)expected_sum * nrecords / 40, v.sum); } + +// What a thread recorded outlives it: the AgentCombiner backend commits a +// dying agent into its global result, the babylon one keeps the slot of an +// exited thread around for whichever thread inherits its id later. +TEST_F(HistogramTest, values_of_dead_threads_are_kept) { + int64_t nvalues = 100; + bvar::Histogram h(bvar::Histogram::BucketSchema({10, 20, 30})); + AddArgs args = {&h, nvalues}; + for (int64_t round = 1; round <= 3; ++round) { + pthread_t th; + ASSERT_EQ(0, pthread_create(&th, nullptr, add_values, &args)); + ASSERT_EQ(0, pthread_join(th, nullptr)); + // Every round runs on a thread which is gone by the time count() reads + // what it recorded, and the next one may well reuse its slot. + ASSERT_EQ(nvalues * round, h.count()); + } +} + +static void check_snapshots_while_recording(bvar::Histogram* h, + int64_t nrecords) { + bvar::Histogram::Value v; + uint64_t last_counts[4] = {}; + for (int i = 0; i < 1000 || v.num < nrecords; ++i) { + v = h->get_value(); + uint64_t total = 0; + for (size_t b = 0; b < arraysize(last_counts); ++b) { + total += v.counts[b]; + // Every sample is taken after the previous one returned and a + // bucket count only ever grows, so this snapshot cannot hold less + // than the last one did. + ASSERT_LE(last_counts[b], v.counts[b]) << "i=" << i << " bucket=" << b; + last_counts[b] = v.counts[b]; + } + ASSERT_EQ(v.num, (int64_t)total) << "i=" << i; + } +} + +// The invariant the per thread seqlock buys, the mutex of the ElementContainer +// without WITH_BABYLON_COUNTER: a snapshot never mixes a bucket that has +// already been incremented with a `num` that has not. It holds for the slice +// of one thread, and summing consistent slices keeps it, so the whole snapshot +// still satisfies the `+Inf bucket == _count` rule of the prometheus format +// while other threads are recording. +TEST_F(HistogramTest, snapshot_is_self_consistent_under_contention) { + int64_t nvalues = 50000; + bvar::Histogram h(bvar::Histogram::BucketSchema({10, 20, 30})); + + pthread_t threads[4]; + AddArgs args = {&h, nvalues}; + for (size_t i = 0; i < arraysize(threads); ++i) { + ASSERT_EQ(0, pthread_create(&threads[i], nullptr, add_values, &args)); + } + + int64_t nrecords = (int64_t)arraysize(threads) * nvalues; + check_snapshots_while_recording(&h, nrecords); + + for (size_t i = 0; i < arraysize(threads); ++i) { + ASSERT_EQ(0, pthread_join(threads[i], nullptr)); + } + ASSERT_EQ(nrecords, h.count()); +} + +static const size_t PERF_OPS_PER_THREAD = 500000; + +struct PerfArgs { + bvar::Histogram* h; + long elapsed_ns; +}; + +static void* record_into_histogram(void* arg) { + PerfArgs* args = (PerfArgs*)arg; + butil::Timer timer; + timer.start(); + for (size_t i = 0; i < PERF_OPS_PER_THREAD; ++i) { + *args->h << (double)(i % 40); + } + timer.stop(); + args->elapsed_ns = timer.n_elapsed(); + return nullptr; +} + +static double time_records(bvar::Histogram* h, size_t nthread) { + PerfArgs proto = {h, 0}; + std::vector args(nthread, proto); + std::vector threads(nthread); + for (size_t i = 0; i < nthread; ++i) { + EXPECT_EQ(0, pthread_create(&threads[i], nullptr, + record_into_histogram, &args[i])); + } + long total_ns = 0; + for (size_t i = 0; i < nthread; ++i) { + EXPECT_EQ(0, pthread_join(threads[i], nullptr)); + total_ns += args[i].elapsed_ns; + } + return (double)total_ns / (double)(PERF_OPS_PER_THREAD * nthread); +} + +TEST_F(HistogramTest, write_perf) { +#if WITH_BABYLON_COUNTER + const char* backend = "babylon"; +#else + const char* backend = "combiner"; +#endif // WITH_BABYLON_COUNTER + std::ostringstream oss; + oss << std::fixed << std::setprecision(2) + << "threads\t" << backend << " (ns per record)\n"; + for (size_t nthread = 1; nthread <= 8; nthread *= 2) { + bvar::Histogram h(bvar::Histogram::BucketSchema({10, 20, 30})); + double ns = time_records(&h, nthread); + ASSERT_EQ((int64_t)(PERF_OPS_PER_THREAD * nthread), h.count()); + oss << nthread << '\t' << ns << '\n'; + } + LOG(INFO) << "Histogram write performance:\n" << oss.str(); +} + } // namespace From e4d8557ac9e9204df7ef8700ec4bc1740a4edee5 Mon Sep 17 00:00:00 2001 From: chenBright Date: Tue, 22 Sep 2026 23:36:57 +0800 Subject: [PATCH 2/2] Fix some issues --- src/bvar/detail/sampler.h | 2 +- src/bvar/histogram.h | 6 +++--- test/bvar_histogram_unittest.cpp | 10 +++++----- 3 files changed, 9 insertions(+), 9 deletions(-) diff --git a/src/bvar/detail/sampler.h b/src/bvar/detail/sampler.h index 7c92413d51..d120458d18 100644 --- a/src/bvar/detail/sampler.h +++ b/src/bvar/detail/sampler.h @@ -304,7 +304,7 @@ class ReducerSampler : public Sampler { // The operator can't be inversed. // We reset the reducer and save the result as a sample. - // Suming up samples gives the result within a window. + // Summing up samples gives the result within a window. // In this case, get_value() of `_source` gives wrong answer and // should not be called. T take_sample_of(butil::true_type) { return _source.reset(); } diff --git a/src/bvar/histogram.h b/src/bvar/histogram.h index 27dd5cd1cd..468805f4a1 100644 --- a/src/bvar/histogram.h +++ b/src/bvar/histogram.h @@ -219,9 +219,9 @@ class Histogram : public Variable { // The contract of Window<>/ReducerSampler Op op() const { return Op(); } InvOp inv_op() const { return InvOp(); } - // Expose the shared data carrier, so that ReducerSampler holds it instead - // of `this`. Sampling then keeps reading valid memory even if this - // Percentile is destructed before the sampler is recycled. + // Expose the shared data carrier, so that ReducerSampler holds it + // instead of `this`. Sampling then keeps reading valid memory even + // if this Histogram is destructed before the sampler is recycled. shared_combiner_type share_combiner() const { #if WITH_BABYLON_COUNTER return _storage; diff --git a/test/bvar_histogram_unittest.cpp b/test/bvar_histogram_unittest.cpp index 44ba8c5ccf..f5a83c5d1c 100644 --- a/test/bvar_histogram_unittest.cpp +++ b/test/bvar_histogram_unittest.cpp @@ -796,12 +796,12 @@ TEST_F(HistogramTest, values_of_dead_threads_are_kept) { static void check_snapshots_while_recording(bvar::Histogram* h, int64_t nrecords) { - bvar::Histogram::Value v; - uint64_t last_counts[4] = {}; + bvar::Histogram::Value v = h->get_value(); + std::vector last_counts(v.num_buckets, 0); for (int i = 0; i < 1000 || v.num < nrecords; ++i) { v = h->get_value(); uint64_t total = 0; - for (size_t b = 0; b < arraysize(last_counts); ++b) { + for (size_t b = 0; b < last_counts.size(); ++b) { total += v.counts[b]; // Every sample is taken after the previous one returned and a // bucket count only ever grows, so this snapshot cannot hold less @@ -842,7 +842,7 @@ static const size_t PERF_OPS_PER_THREAD = 500000; struct PerfArgs { bvar::Histogram* h; - long elapsed_ns; + int64_t elapsed_ns; }; static void* record_into_histogram(void* arg) { @@ -865,7 +865,7 @@ static double time_records(bvar::Histogram* h, size_t nthread) { EXPECT_EQ(0, pthread_create(&threads[i], nullptr, record_into_histogram, &args[i])); } - long total_ns = 0; + int64_t total_ns = 0; for (size_t i = 0; i < nthread; ++i) { EXPECT_EQ(0, pthread_join(threads[i], nullptr)); total_ns += args[i].elapsed_ns;