diff --git a/include/tsutil/Metrics.h b/include/tsutil/Metrics.h index 40cdef9522f..68665dd2de7 100644 --- a/include/tsutil/Metrics.h +++ b/include/tsutil/Metrics.h @@ -308,7 +308,7 @@ class Metrics static constexpr MetricType _extractType(IdType value) { - return MetricType{value >> METRIC_TYPE_BITS}; + return MetricType{static_cast((static_cast(value) >> METRIC_TYPE_BITS) & 0x1)}; } static constexpr IdType @@ -320,11 +320,16 @@ class Metrics class Storage { - BlobStorage _blobs; - uint16_t _cur_blob = 0; - uint16_t _cur_off = 0; - LookupTable _lookups; - mutable std::mutex _mutex; + /* _cur_blob and _cur_off are release stored last, after whatever they publish: the blob pointer + * for _cur_blob, the slot's name for _cur_off. Readers acquire load them, _cur_blob first. + * _blobs needs no atomic because it is only read at an index no greater than _cur_blob. + * Writers hold _mutex and load relaxed. + */ + BlobStorage _blobs; + std::atomic _cur_blob{0}; + std::atomic _cur_off{0}; + LookupTable _lookups; + mutable std::mutex _mutex; public: Storage(const Storage &) = delete; @@ -354,15 +359,37 @@ class Metrics current() const { std::lock_guard lock(_mutex); - return {_cur_blob, _cur_off}; + return {_cur_blob.load(std::memory_order_relaxed), _cur_off.load(std::memory_order_relaxed)}; } + /** Whether @a id names an allocated slot. + * + * The gate for every id based accessor, since ids from the @c TSStat* API are untrusted. An id + * qualifies when it is non-negative, its offset is one @c _makeId could produce, and its slot + * has been handed out. + */ bool - valid(IdType id) const + _is_allocated(IdType id) const { - auto [blob, entry] = _splitID(id); + if (id < 0) { + return false; + } + + auto [blob_ix, offset] = _splitID(id); - return (id >= 0 && ((blob < _cur_blob && entry < MAX_SIZE) || (blob == _cur_blob && entry <= _cur_off))); + // _cur_blob first: acquiring it also makes visible everything published under it. + auto const cur_blob = _cur_blob.load(std::memory_order_acquire); + auto const cur_off = _cur_off.load(std::memory_order_acquire); + + // A non-null blob past cur_blob is allocated but not yet published, hence <= and < rather + // than a test for "not the current blob". + return offset < MAX_SIZE && blob_ix <= cur_blob && _blobs[blob_ix] != nullptr && (blob_ix < cur_blob || offset < cur_off); + } + + bool + valid(IdType id) const + { + return _is_allocated(id); } }; diff --git a/src/tsutil/Metrics.cc b/src/tsutil/Metrics.cc index c339eea2df8..0c45722b4d0 100644 --- a/src/tsutil/Metrics.cc +++ b/src/tsutil/Metrics.cc @@ -58,12 +58,17 @@ Metrics::Storage::addBlob() // The mutex must be held before calling this! { auto blob = std::make_unique(); + auto const cur_blob = _cur_blob.load(std::memory_order_relaxed); + debug_assert(blob); - // The write below is to _blobs[_cur_blob + 1], so the last usable blob index is MAX_BLOBS - 1. - release_assert(_cur_blob < MAX_BLOBS - 1); + // The write below is to _blobs[cur_blob + 1], so the last usable blob index is MAX_BLOBS - 1. + release_assert(cur_blob < MAX_BLOBS - 1); + + _blobs[cur_blob + 1] = std::move(blob); + _cur_off.store(0, std::memory_order_relaxed); - _blobs[++_cur_blob] = std::move(blob); - _cur_off = 0; + // Publishes the blob; both writes above are sequenced before it. + _cur_blob.store(cur_blob + 1, std::memory_order_release); } Metrics::IdType @@ -79,18 +84,24 @@ Metrics::Storage::create(std::string_view name, const MetricType type) // The slot is written below and the bookkeeping only then advances, calling addBlob() once // _cur_off reaches MAX_SIZE. Refusing the final slot of the final blob keeps addBlob() from // ever being reached in an exhausted store, at a cost of one slot out of MAX_BLOBS * MAX_SIZE. - if (_cur_blob >= MAX_BLOBS - 1 && _cur_off >= MAX_SIZE - 1) { + auto const cur_blob = _cur_blob.load(std::memory_order_relaxed); + auto const cur_off = _cur_off.load(std::memory_order_relaxed); + + if (cur_blob >= MAX_BLOBS - 1 && cur_off >= MAX_SIZE - 1) { return 0; // Slot 0 is the reserved bad_id. Cannot grow further. } - Metrics::IdType id = _makeId(_cur_blob, _cur_off, type); - Metrics::NamesAndAtomics *blob = _blobs[_cur_blob].get(); + Metrics::IdType id = _makeId(cur_blob, cur_off, type); + Metrics::NamesAndAtomics *blob = _blobs[cur_blob].get(); Metrics::NameStorage &names = std::get<0>(*blob); - names[_cur_off] = std::make_tuple(std::string(name), id); - _lookups.emplace(std::get<0>(names[_cur_off]), id); + names[cur_off] = std::make_tuple(std::string(name), id); + _lookups.emplace(std::get<0>(names[cur_off]), id); - if (++_cur_off >= MAX_SIZE) { + // Publishes the slot; the name write above is sequenced before it. + _cur_off.store(cur_off + 1, std::memory_order_release); + + if (cur_off + 1 >= MAX_SIZE) { addBlob(); // This resets _cur_off to 0 as well } @@ -113,15 +124,16 @@ Metrics::Storage::lookup(const std::string_view name) const Metrics::AtomicType * Metrics::Storage::lookup(Metrics::IdType id, std::string_view *out_name, Metrics::MetricType *out_type) const { - auto [blob_ix, offset] = _splitID(id); - Metrics::NamesAndAtomics *blob = _blobs[blob_ix].get(); + auto [blob_ix, offset] = _splitID(id); - // Do a sanity check on the ID, to make sure we don't index outside of the realm of possibility. - if (!blob || (blob_ix == _cur_blob && offset > _cur_off)) { - blob = _blobs[0].get(); - offset = 0; + // Anything not naming an allocated slot resolves to the reserved bad_id slot. + if (!_is_allocated(id)) { + blob_ix = 0; + offset = 0; } + Metrics::NamesAndAtomics *blob = _blobs[blob_ix].get(); + if (out_name) { *out_name = std::get<0>(std::get<0>(*blob)[offset]); } @@ -159,15 +171,16 @@ Metrics::Storage::lookup(const std::string_view name, Metrics::IdType *out_id, M std::string_view Metrics::Storage::name(Metrics::IdType id) const { - auto [blob_ix, offset] = _splitID(id); - Metrics::NamesAndAtomics *blob = _blobs[blob_ix].get(); + auto [blob_ix, offset] = _splitID(id); - // Do a sanity check on the ID, to make sure we don't index outside of the realm of possibility. - if (!blob || (blob_ix == _cur_blob && offset > _cur_off)) { - blob = _blobs[0].get(); - offset = 0; + // Anything not naming an allocated slot resolves to the reserved bad_id slot. + if (!_is_allocated(id)) { + blob_ix = 0; + offset = 0; } + Metrics::NamesAndAtomics *blob = _blobs[blob_ix].get(); + const std::string &result = std::get<0>(std::get<0>(*blob)[offset]); return result; @@ -188,7 +201,7 @@ Metrics::Storage::createSpan(size_t size, Metrics::MetricType type, Metrics::IdT // On the final blob there is nowhere left to grow, so refuse a span that would fill or overflow // it rather than letting addBlob() assert. Same intent as the guard in create(), and the same // cost: some slots of the last blob go unused. - if (_cur_blob >= MAX_BLOBS - 1 && _cur_off + size >= MAX_SIZE) { + if (_cur_blob.load(std::memory_order_relaxed) >= MAX_BLOBS - 1 && _cur_off.load(std::memory_order_relaxed) + size >= MAX_SIZE) { if (id) { *id = 0; // Slot 0 is the reserved bad_id. } @@ -196,26 +209,31 @@ Metrics::Storage::createSpan(size_t size, Metrics::MetricType type, Metrics::IdT } // A span has to be contiguous, so one that does not fit in the current blob starts a new one. - if (_cur_off + size > MAX_SIZE) { + if (_cur_off.load(std::memory_order_relaxed) + size > MAX_SIZE) { addBlob(); } - Metrics::IdType span_start = _makeId(_cur_blob, _cur_off, type); - Metrics::NamesAndAtomics *blob = _blobs[_cur_blob].get(); + // Re-read: addBlob() above may have moved both. + auto const cur_blob = _cur_blob.load(std::memory_order_relaxed); + auto const cur_off = _cur_off.load(std::memory_order_relaxed); + + Metrics::IdType span_start = _makeId(cur_blob, cur_off, type); + Metrics::NamesAndAtomics *blob = _blobs[cur_blob].get(); Metrics::AtomicStorage &atomics = std::get<1>(*blob); - Metrics::SpanType span = Metrics::SpanType(&atomics[_cur_off], size); + Metrics::SpanType span = Metrics::SpanType(&atomics[cur_off], size); if (id) { *id = span_start; } - _cur_off += size; + // Publishes the span's slots. + _cur_off.store(cur_off + size, std::memory_order_release); // create() grows as soon as it consumes the last slot; do the same here. Otherwise a span ending // exactly on the boundary leaves _cur_off at MAX_SIZE, and the next create() writes one past the // end of the blob's name array. It also makes end() unreachable for iterator::next(), which // wraps on ++offset == MAX_SIZE. - if (_cur_off >= MAX_SIZE) { + if (cur_off + size >= MAX_SIZE) { addBlob(); } @@ -225,14 +243,14 @@ Metrics::Storage::createSpan(size_t size, Metrics::MetricType type, Metrics::IdT bool Metrics::Storage::rename(Metrics::IdType id, std::string_view name) { - auto [blob_ix, offset] = _splitID(id); - Metrics::NamesAndAtomics *blob = _blobs[blob_ix].get(); - // We can only rename Metrics that are already allocated - if (!blob || (blob_ix == _cur_blob && offset > _cur_off)) { + if (!_is_allocated(id)) { return false; } + auto [blob_ix, offset] = _splitID(id); + Metrics::NamesAndAtomics *blob = _blobs[blob_ix].get(); + std::string &cur = std::get<0>(std::get<0>(*blob)[offset]); std::lock_guard lock(_mutex); diff --git a/src/tsutil/unit_tests/test_Metrics.cc b/src/tsutil/unit_tests/test_Metrics.cc index 960324997b0..5d2ec771f9e 100644 --- a/src/tsutil/unit_tests/test_Metrics.cc +++ b/src/tsutil/unit_tests/test_Metrics.cc @@ -24,6 +24,7 @@ #include #include +#include #include #include #include @@ -610,14 +611,9 @@ TEST_CASE("Metrics blob growth boundary", "[libtsapi][Metrics]") TEST_CASE("Metrics span lands exactly on a blob boundary", "[libtsapi][Metrics]") { - // A span has to be contiguous, so createSpan(MAX_SIZE) always starts a fresh blob and then fills - // it completely, whatever the current offset was. That makes this the one span size that reaches - // the boundary case deterministically: the offset ends up at MAX_SIZE, and unlike create(), - // createSpan used not to grow a new blob afterwards. The next create() then indexed one past the - // end of the blob's name array, and end() became an id that iterator::next() can never reach - // because it wraps at ++offset == MAX_SIZE. - // - // createSpan only ever targets the published store, so this necessarily allocates there. + // A span of MAX_SIZE always lands at offset 0 of an empty blob and fills it, whatever the current + // offset was, so it reaches the blob boundary deterministically. createSpan only targets the + // published store, so this allocates there. Metrics::IdType span_id = Metrics::NOT_FOUND; auto span = Metrics::Counter::createSpan(Metrics::MAX_SIZE, &span_id); @@ -640,3 +636,85 @@ TEST_CASE("Metrics span lands exactly on a blob boundary", "[libtsapi][Metrics]" REQUIRE(Metrics::Counter::load(p) == 7); REQUIRE(Metrics::Counter::createPtr("span.boundary.after") == p); } + +TEST_CASE("Metrics malformed id offsets resolve to bad_id", "[libtsapi][Metrics]") +{ + // An id's offset field is 16 bits but a real offset is below MAX_SIZE, so a malformed one must + // not index past a blob's arrays. Two blobs are needed for the offset check to be what rejects + // it; with one, the null blob check would. + auto &h = Metrics::hidden_instance(); + + for (int i = 0; i < Metrics::MAX_SIZE + 8; ++i) { + REQUIRE(Metrics::Counter::createHiddenPtr("f1.fill." + std::to_string(i)) != nullptr); + } + + auto const *bad = h.lookup(Metrics::IdType{0}); // the reserved bad_id slot + REQUIRE(bad != nullptr); + + // blob 0 is allocated, so the null check does not fire; only the MAX_SIZE test stands between + // this and atomics[65535]. + for (Metrics::IdType id : {Metrics::IdType{0x0000FFFF}, Metrics::IdType{0x00000400}, Metrics::IdType{0x0001FFFF}}) { + REQUIRE(h.valid(id) == false); + REQUIRE(h.lookup(id) == bad); + REQUIRE(h.name(id) == h.name(Metrics::IdType{0})); + } +} + +TEST_CASE("Metrics id lookup is safe against concurrent creation", "[libtsapi][Metrics]") +{ + // The id based read paths take no lock, so resolving an id races a concurrent create. Run both + // sides at once, across enough metrics to cross several blob boundaries. Under the tsan preset a + // non-atomic allocation counter reports a data race here; relaxing the memory orders does not, + // since atomics are race free at any ordering. + constexpr int N_READERS = 4; + constexpr int N_CREATE = Metrics::MAX_SIZE * 2 + 64; + auto &h = Metrics::hidden_instance(); + std::atomic stop{false}; + std::atomic mismatches{0}; + + std::vector readers; + + for (int t = 0; t < N_READERS; ++t) { + readers.emplace_back([&]() { + while (!stop.load(std::memory_order_relaxed)) { + for (Metrics::IdType id = 0; id < N_CREATE; ++id) { + if (!h.valid(id)) { + continue; + } + + // valid() said this id names an allocated slot, so lookup() must agree and hand back a + // real metric rather than clamping to the reserved bad_id slot. A publication ordering + // mistake shows up here as a name that is still empty. + std::string_view name; + Metrics::MetricType type; + auto *m = h.lookup(id, &name, &type); + + if (m == nullptr || (id != 0 && name.empty())) { + mismatches.fetch_add(1, std::memory_order_relaxed); + } + } + } + }); + } + + for (int i = 0; i < N_CREATE; ++i) { + REQUIRE(Metrics::Counter::createHiddenPtr("pub.order." + std::to_string(i)) != nullptr); + } + + stop.store(true, std::memory_order_relaxed); + for (auto &r : readers) { + r.join(); + } + + CHECK(mismatches.load() == 0); + + // Everything the writer created must be resolvable by name and by id afterwards. + for (int i = 0; i < N_CREATE; ++i) { + auto const nm = "pub.order." + std::to_string(i); + auto const id = h.lookup(nm); + + REQUIRE(id != Metrics::NOT_FOUND); + REQUIRE(h.valid(id)); + REQUIRE(h.name(id) == nm); + } +} diff --git a/tools/benchmark/CMakeLists.txt b/tools/benchmark/CMakeLists.txt index e58e8dcf26b..ba3816cf6be 100644 --- a/tools/benchmark/CMakeLists.txt +++ b/tools/benchmark/CMakeLists.txt @@ -33,6 +33,9 @@ target_link_libraries(benchmark_ProxyAllocator PRIVATE Catch2::Catch2WithMain ts add_executable(benchmark_SharedMutex benchmark_SharedMutex.cc) target_link_libraries(benchmark_SharedMutex PRIVATE Catch2::Catch2 ts::tscore libswoc::libswoc) +add_executable(benchmark_Metrics benchmark_Metrics.cc) +target_link_libraries(benchmark_Metrics PRIVATE Catch2::Catch2 ts::tsutil libswoc::libswoc) + add_executable(benchmark_Random benchmark_Random.cc) target_link_libraries(benchmark_Random PRIVATE Catch2::Catch2WithMain ts::tscore) diff --git a/tools/benchmark/benchmark_Metrics.cc b/tools/benchmark/benchmark_Metrics.cc new file mode 100644 index 00000000000..e42b092a7f4 --- /dev/null +++ b/tools/benchmark/benchmark_Metrics.cc @@ -0,0 +1,190 @@ +/** @file + + Micro benchmark tool for ts::Metrics + + Metric values are lock free atomics; reaching one from an id is not. Four cases, scaled by thread + count: + + increment(id) valid() then lookup(id) then fetch_add, as TSStatIntIncrement does + increment(ptr) a bare fetch_add on a cached pointer, as core does + lookup(id) lock free id resolution + lookup(name) the same resolution through the mutex guarded name map + + increment(id) against increment(ptr) is what an id costs a plugin. lookup(id) against + lookup(name) isolates the mutex, and serves as a control: it must degrade with thread count, or + the harness is not loading the machine. + + @section license License + + Licensed to the Apache Software Foundation (ASF) under one + or more contributor license agreements. See the NOTICE file + distributed with this work for additional information + regarding copyright ownership. The ASF licenses this file + to you under the Apache License, Version 2.0 (the + "License"); you may not use this file except in compliance + with the License. You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. + */ + +#define CATCH_CONFIG_ENABLE_BENCHMARKING + +#include +#include +#include + +#include "tsutil/Metrics.h" + +#include +#include +#include +#include + +using ts::Metrics; + +namespace +{ +// Args +struct Conf { + int nthreads = 1; + int nops = 1000; + int nmetrics = 64; +}; + +Conf conf; + +/// The metrics every case operates on, created once. +struct Fixture { + std::vector ids; + std::vector ptrs; + std::vector names; + + Fixture() + { + auto &m = Metrics::instance(); + + ids.reserve(conf.nmetrics); + ptrs.reserve(conf.nmetrics); + names.reserve(conf.nmetrics); + + for (int i = 0; i < conf.nmetrics; ++i) { + names.push_back("benchmark.metrics." + std::to_string(i)); + + // Registers the name too, so the id and name lookups resolve to the same metric. + ptrs.push_back(Metrics::Counter::createPtr(names.back())); + ids.push_back(m.lookup(names.back())); + } + } +}; + +Fixture *fixture = nullptr; + +/// Run @a op on every thread, @c nops times each. The return value only defeats optimization. +template +int64_t +run(F &&op) +{ + std::vector threads; + std::atomic sink{0}; + + threads.reserve(conf.nthreads); + + for (int t = 0; t < conf.nthreads; ++t) { + threads.emplace_back([t, &sink, &op]() { + int64_t local = 0; + + for (int i = 0; i < conf.nops; ++i) { + // Stride per thread, or this measures cacheline ping-pong on one atomic. + local += op((t + i) % conf.nmetrics); + } + sink.fetch_add(local, std::memory_order_relaxed); + }); + } + + for (auto &th : threads) { + th.join(); + } + + return sink.load(); +} + +} // namespace + +TEST_CASE("Micro benchmark of ts::Metrics", "") +{ + auto &m = Metrics::instance(); + + SECTION("increment by id") + { + BENCHMARK("increment(id)") + { + return run([&m](int i) -> int64_t { + auto id = fixture->ids[i]; + + return m.valid(id) ? m.increment(id, 1) : 0; + }); + }; + } + + SECTION("increment by cached pointer") + { + // The floor: no resolution at all. + BENCHMARK("increment(ptr)") + { + return run([](int i) -> int64_t { + Metrics::Counter::increment(fixture->ptrs[i], 1); + + return 1; + }); + }; + } + + SECTION("lookup by id") + { + BENCHMARK("lookup(id)") + { + return run([&m](int i) -> int64_t { return m.lookup(fixture->ids[i]) != nullptr; }); + }; + } + + SECTION("lookup by name") + { + BENCHMARK("lookup(name)") + { + return run([&m](int i) -> int64_t { return m.lookup(fixture->names[i]) != Metrics::NOT_FOUND; }); + }; + } +} + +int +main(int argc, char *argv[]) +{ + Catch::Session session; + + using namespace Catch::Clara; + + // clang-format off + auto cli = session.cli() | + Opt(conf.nthreads, "")["--ts-nthreads"]("number of threads (default: 1)") | + Opt(conf.nops, "")["--ts-nops"]("operations per thread per run (default: 1000)") | + Opt(conf.nmetrics, "")["--ts-nmetrics"]("distinct metrics to spread across (default: 64)"); + // clang-format on + + session.cli(cli); + + int returnCode = session.applyCommandLine(argc, argv); + if (returnCode != 0) { + return returnCode; + } + + Fixture f; + fixture = &f; + + return session.run(); +}