Metrics: close the id lookup race and bounds gaps left by the lock revert

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<int>((static_cast<uint32_t>(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<uint16_t> _cur_blob{0};
+    std::atomic<uint16_t> _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);
     }
   };
 

src/tsutil/Metrics.cc

@@ -58,12 +58,17 @@ Metrics::Storage::addBlob() // The mutex must be held before calling this!
 {
   auto blob = std::make_unique<Metrics::NamesAndAtomics>();
 
+  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,34 +201,39 @@ 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.
     }
     return {};
   }
 
   // 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);
 

src/tsutil/unit_tests/test_Metrics.cc

@@ -24,6 +24,7 @@
 #include <catch2/catch_test_macros.hpp>
 
 #include <algorithm>
+#include <atomic>
 #include <array>
 #include <iterator>
 #include <memory>
@@ -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<bool> stop{false};
+  std::atomic<int>  mismatches{0};
+
+  std::vector<std::thread> 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);
+  }
+}

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)