From ff19220787e5326ffbe67e242bcb9201f4341f8f Mon Sep 17 00:00:00 2001 From: gxglass Date: Mon, 15 Sep 2025 17:07:07 -0700 Subject: [PATCH] SimpleCounter implementation and application to Arena, FastAlloc, and Net2 cherrypick to 7.4 (#12372) * Arena/FastAlloc: add comments where potential metrics can be added (#12306) * Arena/FastAlloc: add comments where potential metrics can be incremented. The intent is to count allocations and bytes. Remove a commented-out ifdef block that has been disabled for many years and which does not work (per the explanation in the comment). We don't need to keep reading about the results of a small failed experiment from many years ago. * add more one FIXME comment * add one more METRICS-FIXME * Add a SimpleCounter template for counter metrics (#12326) * ignore TAGS (from etags/ctags) * Add initial SimpleCounter interface/implementation/unit tests * Add initial SimpleCounter interface/implementation/unit tests * Fix unit test * Improve clarity on unit test * Update FIXME comments * Address review comments. Must use function local static mutex * update comment * Go back to template specializations to handle older C++ versions * SimpleCounter: periodically log the counters to TraceEvent (#12329) * SimpleCounter: periodically log the counters to TraceEvent. Muck with hierarchical names to comply with random rules. * Update doc about Prometheus metric names * relax assertion about counter count, because unit test is actually running in fdbserver and that causes a unrelated counter to be created * Update SimpleCounter unit tests not to use metric names that break Trace.cpp simulation-only checks (#12333) * Add a pointed comment in UnitTest.h about some weaknesses * Use counter names that will get converted to field names that TraceEvent does not complain about * unit test: do not use a counter name that will cause Trace.cpp to emit errors in simulation * update comment about caveats with unittests breaking simulation * yet another field name fix * just call validateField() directly from simple counters * run report loop in unit tests * fix build, fix comment * blah blah blah * always be munging metric names * Instrument Arena, FastAlloc, Platform.cpp with SimpleCounter metrics to count allocations and bytes (#12339) * emit a simpleCounterReport when we declare out of memory * FastAlloc.h: initial pass of adding byte/object allocation/deallocation metrics * Avoid conflict over the name SimpleCounter by eliminating this private definition of a name which is too valuable for this one random file to claim for its own use * FastAlloc.cpp, Platform.actor.cpp: initial pass at adding SimpleCounter metrics to count allocations and bytes * rename wrapper calls and update comments * Count bytes copied in StringRef * Arena.cpp: instrument allocations and some other stuff * Arena.cpp: simplify use of SimpleCounter * simpleCounterReport: generate TraceEvent in batches of MAX_TRACE_EVENT_LENGTH / 100 counters to avoid trace buffer overflow * Eliminate poorly motivated trace field name validation, and change SimpleCounter to emit Prometheus-compatible metric names (#12356) Trace.cpp does not provide a rationale for validateField() and validateFormat(). It appears to be some kind of XML related validation. Why we should care about this is not clear. The output is going to Splunk. As far as I know, Splunk is supposed to be pretty liberal in what it accepts as input. Add logic in SimpleCounter.cpp to convert hierarchical metric names to Prometheus compatible metric names by the simple rule of converting intermediate '/' chars into '_', i.e. something like /flow/arena/bytesAllocated becomes flow_arena_bytesAllocated. I feel hierarchical names are still slightly better, and very easy to reason about when creating new metric names on the fly, but ensuring that they are at least Prometheus compatible should allow targeting to future metrics platforms down the road. Testing: 20250905-010257-gglass-25f3ef43ccc1c130 compressed=True data_size=41538755 duration=6389116 ended=100000 fail_fast=10 max_runs=100000 pass=100000 priority=100 remaining=0 runtime=1:00:34 sanity=False started=100000 stopped=20250905-020331 submitted=20250905-010257 timeout=5400 username=gglass * replace undocumented trace event field name rules with a rule that enforces that field names must be valid Prometheus metric names. No idea why the old code declines to even state what it is trying to be compatible with * move Prometheus metric name validation to SimpleCounter.cpp. Remove validation from Trace.cpp. This stuff is going to Splunk. Splunk takes what we give it. * Use simple counter to replace recently added net2 counters (#12358) * use simple counter to replace recent added net2 counters * allow unit test to use SimpleCounter Trace event --------- Co-authored-by: Zhe Wang --- .gitignore | 1 + fdbclient/ServerKnobs.cpp | 1 + fdbclient/include/fdbclient/ServerKnobs.h | 1 + fdbrpc/FlowTransport.actor.cpp | 59 +++---- fdbserver/VersionedBTree.actor.cpp | 16 +- fdbserver/fdbserver.actor.cpp | 21 ++- fdbserver/workloads/UnitTests.actor.cpp | 2 + flow/Arena.cpp | 29 ++- flow/FastAlloc.cpp | 43 +++++ flow/Net2.actor.cpp | 60 +++---- flow/Platform.actor.cpp | 8 + flow/SimpleCounter.cpp | 205 ++++++++++++++++++++++ flow/SystemMonitor.cpp | 74 -------- flow/Trace.cpp | 34 ---- flow/include/flow/Arena.h | 88 +++++----- flow/include/flow/FastAlloc.h | 42 ++++- flow/include/flow/SimpleCounter.h | 150 ++++++++++++++++ flow/include/flow/SystemMonitor.h | 49 ------ flow/include/flow/UnitTest.h | 12 +- 19 files changed, 613 insertions(+), 282 deletions(-) create mode 100644 flow/SimpleCounter.cpp create mode 100644 flow/include/flow/SimpleCounter.h diff --git a/.gitignore b/.gitignore index 0fe8a7c92a..d4f3a0248d 100644 --- a/.gitignore +++ b/.gitignore @@ -94,6 +94,7 @@ flow/coveragetool/obj /.ccls-cache /.clangd /.cache +TAGS # Temporary and user configuration files *~ diff --git a/fdbclient/ServerKnobs.cpp b/fdbclient/ServerKnobs.cpp index 81800c29fb..8fdf97d3f7 100644 --- a/fdbclient/ServerKnobs.cpp +++ b/fdbclient/ServerKnobs.cpp @@ -1182,6 +1182,7 @@ void ServerKnobs::initialize(Randomize randomize, ClientKnobs* clientKnobs, IsSi init( CONFIGURATION_ROWS_TO_FETCH, 20000 ); init( DISABLE_DUPLICATE_LOG_WARNING, false ); init( HISTOGRAM_REPORT_INTERVAL, 300.0 ); + init( GENERIC_METRICS_REPORT_INTERVAL, isSimulated ? 10.0 : 300.0 ); // Timekeeper init( TIME_KEEPER_DELAY, 10 ); diff --git a/fdbclient/include/fdbclient/ServerKnobs.h b/fdbclient/include/fdbclient/ServerKnobs.h index b46d1ee9e1..c66ebc9fd3 100644 --- a/fdbclient/include/fdbclient/ServerKnobs.h +++ b/fdbclient/include/fdbclient/ServerKnobs.h @@ -1240,6 +1240,7 @@ public: int CONFIGURATION_ROWS_TO_FETCH; bool DISABLE_DUPLICATE_LOG_WARNING; double HISTOGRAM_REPORT_INTERVAL; + double GENERIC_METRICS_REPORT_INTERVAL; // Timekeeper int64_t TIME_KEEPER_DELAY; diff --git a/fdbrpc/FlowTransport.actor.cpp b/fdbrpc/FlowTransport.actor.cpp index 9144c90723..47365b9cb5 100644 --- a/fdbrpc/FlowTransport.actor.cpp +++ b/fdbrpc/FlowTransport.actor.cpp @@ -315,16 +315,6 @@ public: countConnEstablished.init("Net2.CountConnEstablished"_sr); countConnClosedWithError.init("Net2.CountConnClosedWithError"_sr); countConnClosedWithoutError.init("Net2.CountConnClosedWithoutError"_sr); - countConnIncompatible.init("Net2.CountConnIncompatible"_sr); - countConnIncompatibleWithOldClient.init("Net2.CountConnIncompatibleWithOldClient"_sr); - countConnHandshakeAccepted.init("Net2.CountConnHandshakeAccepted"_sr); - countConnHandshakeRequested.init("Net2.CountConnHandshakeRequested"_sr); - countIncomingConnRequested.init("Net2.CountIncomingConnRequested"_sr); - countIncomingConnAccepted.init("Net2.CountIncomingConnAccepted"_sr); - countOutgoingConnHandshakeComplete.init("Net2.CountOutgoingConnHandshakeComplete"_sr); - countOutgoingConnHandshakeRequested.init("Net2.CountOutgoingConnHandshakeRequested"_sr); - countIncomingConnectionTimedout.init("Net2.CountIncomingConnectionTimedout"_sr); - countIncomingConnConnected.init("Net2.CountIncomingConnConnected"_sr); } Reference getPeer(NetworkAddress const& address); @@ -353,16 +343,6 @@ public: Int64MetricHandle countConnEstablished; Int64MetricHandle countConnClosedWithError; Int64MetricHandle countConnClosedWithoutError; - Int64MetricHandle countConnIncompatible; - Int64MetricHandle countConnIncompatibleWithOldClient; - Int64MetricHandle countConnHandshakeAccepted; - Int64MetricHandle countConnHandshakeRequested; - Int64MetricHandle countIncomingConnRequested; - Int64MetricHandle countIncomingConnAccepted; - Int64MetricHandle countOutgoingConnHandshakeComplete; - Int64MetricHandle countOutgoingConnHandshakeRequested; - Int64MetricHandle countIncomingConnectionTimedout; - Int64MetricHandle countIncomingConnConnected; std::map> incompatiblePeers; AsyncTrigger incompatiblePeersChanged; @@ -832,9 +812,14 @@ ACTOR Future connectionKeeper(Reference self, when(Reference _conn = wait(INetworkConnections::net()->connect(self->destination))) { conn = _conn; - self->transport->countOutgoingConnHandshakeRequested++; + static SimpleCounter* countOutgoingConnectionCreated = + SimpleCounter::makeCounter("/Transport/TLS/OutgoingConnectionCreated"); + countOutgoingConnectionCreated->increment(1); wait(conn->connectHandshake()); - self->transport->countOutgoingConnHandshakeComplete++; + static SimpleCounter* countOutgoingConnectionHandshakeComplete = + SimpleCounter::makeCounter( + "/Transport/TLS/OutgoingConnectionHandshakeComplete"); + countOutgoingConnectionHandshakeComplete->increment(1); self->connectLatencies.addSample(now() - self->lastConnectTime); if (FlowTransport::isClient()) { IFailureMonitor::failureMonitor().setStatus(self->destination, FailureStatus(false)); @@ -1520,12 +1505,17 @@ ACTOR static Future connectionReader(TransportData* transport, now() + FLOW_KNOBS->CONNECTION_ID_TIMEOUT; } compatible = false; - transport->countConnIncompatible++; + static SimpleCounter* countConnectionIncompatible = + SimpleCounter::makeCounter("/Transport/TLS/ConnectionIncompatible"); + countConnectionIncompatible->increment(1); if (!protocolVersion.hasInexpensiveMultiVersionClient()) { if (peer) { peer->protocolVersion->set(protocolVersion); } - transport->countConnIncompatibleWithOldClient++; + static SimpleCounter* countConnectionIncompatibleWithVeryOldClient = + SimpleCounter::makeCounter( + "/Transport/TLS/ConnectionIncompatibleWithVeryOldClient"); + countConnectionIncompatibleWithVeryOldClient->increment(1); // Older versions expected us to hang up. It may work even if we don't hang up here, but // it's safer to keep the old behavior. throw incompatible_protocol_version(); @@ -1624,9 +1614,10 @@ ACTOR static Future connectionIncoming(TransportData* self, ReferencegetPeerAddress(); try { - self->countConnHandshakeRequested++; wait(conn->acceptHandshake()); - self->countConnHandshakeAccepted++; + static SimpleCounter* countIncomingConnectionHandshakeAccepted = + SimpleCounter::makeCounter("/Transport/TLS/IncomingConnectionHandshakeAccepted"); + countIncomingConnectionHandshakeAccepted->increment(1); state Promise> onConnected; state Future reader = connectionReader(self, conn, Reference(), onConnected); if (FLOW_KNOBS->LOG_CONNECTION_ATTEMPTS_ENABLED) { @@ -1643,17 +1634,24 @@ ACTOR static Future connectionIncoming(TransportData* self, ReferenceCONNECTION_MONITOR_TIMEOUT))) { CODE_PROBE(true, "Incoming connection timed out"); - self->countIncomingConnectionTimedout++; + static SimpleCounter* countIncomingConnectionTimedout = + SimpleCounter::makeCounter("/Transport/TLS/IncomingConnectionTimedout"); + countIncomingConnectionTimedout->increment(1); throw timed_out(); } } - self->countIncomingConnConnected++; + static SimpleCounter* countIncomingConnectionConnected = + SimpleCounter::makeCounter("/Transport/TLS/IncomingConnectionConnected"); + countIncomingConnectionConnected->increment(1); } catch (Error& e) { if (e.code() != error_code_actor_cancelled) { TraceEvent("IncomingConnectionError", conn->getDebugID()) .errorUnsuppressed(e) .suppressFor(1.0) .detail("FromAddress", conn->getPeerAddress()); + static SimpleCounter* countIncomingConnectionFailed = + SimpleCounter::makeCounter("/Transport/TLS/IncomingConnectionFailed"); + countIncomingConnectionFailed->increment(1); if (FLOW_KNOBS->LOG_CONNECTION_ATTEMPTS_ENABLED) { entry.failed = true; self->connectionHistory.push_back(entry); @@ -1678,9 +1676,10 @@ ACTOR static Future listen(TransportData* self, NetworkAddress listenAddr) state uint64_t connectionCount = 0; try { loop { - self->countIncomingConnRequested++; Reference conn = wait(listener->accept()); - self->countIncomingConnAccepted++; + static SimpleCounter* countIncomingConnectionCreated = + SimpleCounter::makeCounter("/Transport/TLS/IncomingConnectionCreated"); + countIncomingConnectionCreated->increment(1); if (conn) { TraceEvent("ConnectionFrom", conn->getDebugID()) .suppressFor(1.0) diff --git a/fdbserver/VersionedBTree.actor.cpp b/fdbserver/VersionedBTree.actor.cpp index fbaac87c51..e19d48d7dd 100644 --- a/fdbserver/VersionedBTree.actor.cpp +++ b/fdbserver/VersionedBTree.actor.cpp @@ -10037,8 +10037,8 @@ TEST_CASE("Lredwood/correctness/unit/deltaTree/IntIntPair") { return Void(); } -struct SimpleCounter { - SimpleCounter() : x(0), t(timer()), start(t), xt(0) {} +struct ReallySimpleCounter { + ReallySimpleCounter() : x(0), t(timer()), start(t), xt(0) {} void operator+=(int n) { x += n; } void operator++() { x++; } int64_t get() { return x; } @@ -10302,12 +10302,12 @@ TEST_CASE("Lredwood/correctness/btree") { state Version version = lastVer + 1; - state SimpleCounter mutationBytes; - state SimpleCounter keyBytesInserted; - state SimpleCounter valueBytesInserted; - state SimpleCounter sets; - state SimpleCounter rangeClears; - state SimpleCounter keyBytesCleared; + state ReallySimpleCounter mutationBytes; + state ReallySimpleCounter keyBytesInserted; + state ReallySimpleCounter valueBytesInserted; + state ReallySimpleCounter sets; + state ReallySimpleCounter rangeClears; + state ReallySimpleCounter keyBytesCleared; state int mutationBytesThisCommit = 0; state int mutationBytesTargetThisCommit = randomSize(maxCommitSize); diff --git a/fdbserver/fdbserver.actor.cpp b/fdbserver/fdbserver.actor.cpp index 070466e450..5dba899073 100644 --- a/fdbserver/fdbserver.actor.cpp +++ b/fdbserver/fdbserver.actor.cpp @@ -87,6 +87,7 @@ #include "flow/FaultInjection.h" #include "flow/flow.h" #include "flow/network.h" +#include "flow/SimpleCounter.h" #include "flow/swift.h" #include "flow/swift_concurrency_hooks.h" @@ -410,6 +411,14 @@ ACTOR Future histogramReport() { } } +ACTOR Future metricsReport() { + loop { + wait(delay(SERVER_KNOBS->GENERIC_METRICS_REPORT_INTERVAL)); + + simpleCounterReport(); + } +} + void testSerializationSpeed() { double tstart; double build = 0, serialize = 0, deserialize = 0, copy = 0, deallocate = 0; @@ -2008,11 +2017,6 @@ int main(int argc, char* argv[]) { // Enables profiling on this thread (but does not start it) registerThreadForProfiling(); -#ifdef _WIN32 - // Windows needs a gentle nudge to format floats correctly - //_set_output_format(_TWO_DIGIT_EXPONENT); -#endif - auto opts = CLIOptions::parseArgs(argc, argv); const auto role = opts.role; @@ -2085,8 +2089,8 @@ int main(int argc, char* argv[]) { flushAndExit(FDB_EXIT_SUCCESS); } - // Initialize the thread pool CoroThreadPool::init(); + // Ordinarily, this is done when the network is run. However, network thread should be set before TraceEvents // are logged. This thread will eventually run the network, so call it now. TraceEvent::setNetworkThread(); @@ -2252,6 +2256,7 @@ int main(int argc, char* argv[]) { TraceEvent("Simulation").detail("TestFile", opts.testFile); auto histogramReportActor = histogramReport(); + auto metricsReportActor = metricsReport(); CLIENT_KNOBS->trace(); FLOW_KNOBS->trace(); @@ -2441,7 +2446,8 @@ int main(int argc, char* argv[]) { opts.configDBType, opts.consistencyCheckUrgentMode)); actors.push_back(histogramReport()); - // actors.push_back( recurring( []{}, .001 ) ); // for ASIO latency measurement + actors.push_back(metricsReport()); + #ifdef FLOW_GRPC_ENABLED if (opts.grpcAddressStrs.size() > 0) { FlowGrpc::init(&opts.tlsConfig, NetworkAddress::parse(opts.grpcAddressStrs[0])); @@ -2498,6 +2504,7 @@ int main(int argc, char* argv[]) { setupRunLoopProfiler(); auto m = startSystemMonitor(opts.dataFolder, opts.dcId, opts.zoneId, opts.zoneId, opts.localities.dataHallId()); + auto metricsReportActor = metricsReport(); f = stopAfter(runTests(opts.connectionFile, TEST_TYPE_UNIT_TESTS, TEST_HERE, diff --git a/fdbserver/workloads/UnitTests.actor.cpp b/fdbserver/workloads/UnitTests.actor.cpp index 66652ca2be..00edf59a88 100644 --- a/fdbserver/workloads/UnitTests.actor.cpp +++ b/fdbserver/workloads/UnitTests.actor.cpp @@ -57,6 +57,7 @@ void forceLinkRESTSimKmsVaultTest(); void forceLinkActorFuzzUnitTests(); void forceLinkGrpcTests(); void forceLinkGrpcTests2(); +void forceLinkSimpleCounterTests(); struct UnitTestWorkload : TestWorkload { static constexpr auto NAME = "UnitTests"; @@ -132,6 +133,7 @@ struct UnitTestWorkload : TestWorkload { forceLinkSimKmsVaultTests(); forceLinkRESTSimKmsVaultTest(); forceLinkActorFuzzUnitTests(); + forceLinkSimpleCounterTests(); #ifdef FLOW_GRPC_ENABLED forceLinkGrpcTests(); diff --git a/flow/Arena.cpp b/flow/Arena.cpp index 377e3bb298..e85fc99154 100644 --- a/flow/Arena.cpp +++ b/flow/Arena.cpp @@ -19,9 +19,9 @@ */ #include "flow/Arena.h" - -#include "flow/UnitTest.h" #include "flow/ScopeExit.h" +#include "flow/SimpleCounter.h" +#include "flow/UnitTest.h" #include "flow/config.h" @@ -110,9 +110,13 @@ void makeUndefined(void*, size_t) {} Arena::Arena() : impl(nullptr) {} Arena::Arena(size_t reservedSize) : impl(0) { UNSTOPPABLE_ASSERT(reservedSize < std::numeric_limits::max()); + static SimpleCounter* created = SimpleCounter::makeCounter("/flow/arena/arenasCreated"); + created->increment(1); if (reservedSize) { allowAccess(impl.getPtr()); ArenaBlock::create((int)reservedSize, impl); + static SimpleCounter* bytes = SimpleCounter::makeCounter("/flow/arena/arenaBytesReserved"); + bytes->increment(reservedSize); disallowAccess(impl.getPtr()); } } @@ -138,6 +142,8 @@ void* Arena::allocate4kAlignedBuffer(uint32_t size) { size_t Arena::getSize(FastInaccurateEstimate fastInaccurateEstimate) const { if (impl) { + static SimpleCounter* calls = SimpleCounter::makeCounter("/flow/arena/getSizeCalls"); + calls->increment(1); allowAccess(impl.getPtr()); size_t result; if (fastInaccurateEstimate) { @@ -219,6 +225,9 @@ size_t ArenaBlock::totalSize(std::unordered_set& visited) const { totalSizeEstimate = size(); int o = nextBlockOffset; while (o) { + static SimpleCounter* count = + SimpleCounter::makeCounter("/flow/arena/totalSizeBlocksExamined"); + count->increment(1); ArenaBlockRef* r = (ArenaBlockRef*)((char*)getData() + o); makeDefined(r, sizeof(ArenaBlockRef)); if (r->aligned4kBufferSize != 0) { @@ -247,6 +256,8 @@ void ArenaBlock::wipeUsed() { int dataOffset = isTiny() ? TINY_HEADER : sizeof(ArenaBlock); void* dataBegin = (char*)getData() + dataOffset; int dataSize = used() - dataOffset; + static SimpleCounter* bytesWiped = SimpleCounter::makeCounter("/flow/arena/bytesWiped"); + bytesWiped->increment(dataSize); makeDefined(dataBegin, dataSize); ::memset(dataBegin, 0, dataSize); makeNoAccess(dataBegin, dataSize); @@ -332,6 +343,13 @@ void* ArenaBlock::dependOn4kAlignedBuffer(Reference& self, uint32_t } void* ArenaBlock::allocate(Reference& self, int bytes, IsSecureMem isSecure) { + static SimpleCounter* arenaBlockAllocations = + SimpleCounter::makeCounter("/flow/arena/arenaBlockAllocations"); + static SimpleCounter* arenaBlockBytesAllocated = + SimpleCounter::makeCounter("/flow/arena/arenaBlockBytesAllocated"); + arenaBlockAllocations->increment(1); + arenaBlockBytesAllocated->increment(bytes); + ArenaBlock* b = self.getPtr(); allowAccess(b); if (!self || self->unused() < bytes) { @@ -351,6 +369,8 @@ void* ArenaBlock::allocate(Reference& self, int bytes, IsSecureMem i // Return an appropriately-sized ArenaBlock to store the given data ArenaBlock* ArenaBlock::create(int dataSize, Reference& next) { ArenaBlock* b; + static SimpleCounter* created = SimpleCounter::makeCounter("/flow/arena/arenaBlocksCreated"); + created->increment(1); // all blocks are initialized with no-wipe by default. allocate() sets it, if needed. if (dataSize <= SMALL - TINY_HEADER && !next) { static_assert(sizeof(ArenaBlock) <= 32); // Need to allocate at least sizeof(ArenaBlock) for an ArenaBlock*. See @@ -381,6 +401,7 @@ ArenaBlock* ArenaBlock::create(int dataSize, Reference& next) { } if (reqSize < LARGE) { + // NOTE: FastAlloc.* maintains metrics on its ::allocate calls. if (reqSize <= 128) { b = (ArenaBlock*)FastAllocator<128>::allocate(); b->bigSize = 128; @@ -487,6 +508,9 @@ void ArenaBlock::destroy() { } } b->destroyLeaf(); + static SimpleCounter* destroyed = + SimpleCounter::makeCounter("/flow/arena/arenaBlocksDestroyed"); + destroyed->increment(1); } } @@ -503,6 +527,7 @@ void ArenaBlock::destroyLeaf() { INSTRUMENT_RELEASE("Arena64"); } } else { + // NOTE: FastAlloc.* maintains counters on ::release calls/bytes. if (bigSize <= 128) { FastAllocator<128>::release(this); INSTRUMENT_RELEASE("Arena128"); diff --git a/flow/FastAlloc.cpp b/flow/FastAlloc.cpp index 9bbd8c58de..88181b8ba3 100644 --- a/flow/FastAlloc.cpp +++ b/flow/FastAlloc.cpp @@ -119,6 +119,9 @@ std::map> hugeArenaTraces; void hugeArenaSample(int size) { if (TraceEvent::isNetworkThread()) { + static SimpleCounter* calls = SimpleCounter::makeCounter("/flow/fastalloc/hugeArenaSample"); + calls->increment(1); + auto& info = hugeArenaTraces[platform::get_backtrace()]; info.first++; info.second += size; @@ -373,6 +376,15 @@ void* FastAllocator::allocate() { if (keepalive_allocator::isActive()) [[unlikely]] return keepalive_allocator::allocate(Size); + // Accounting should mirror release() below. + static int size = Size; + static SimpleCounter* calls = + SimpleCounter::makeCounter(format("/flow/fastalloc/allocateCallsSize%d", size)); + static SimpleCounter* bytes = + SimpleCounter::makeCounter(format("/flow/fastalloc/allocateBytesSize%d", size)); + calls->increment(1); + bytes->increment(size); + #if defined(USE_GPERFTOOLS) || defined(ADDRESS_SANITIZER) // Some usages of FastAllocator require 4096 byte alignment. return aligned_alloc(Size >= 4096 ? 4096 : alignof(void*), Size); @@ -423,11 +435,40 @@ void* FastAllocator::allocate() { return p; } +void* countedNew(size_t nbytes) { + static SimpleCounter* calls = SimpleCounter::makeCounter("/flow/fastalloc/newCalls"); + static SimpleCounter* bytes = SimpleCounter::makeCounter("/flow/fastalloc/newBytes"); + calls->increment(1); + bytes->increment(nbytes); + + void* p = new uint8_t[nbytes]; + return p; +} + +void countedDelete(size_t nbytes, void* ptr) { + static SimpleCounter* calls = SimpleCounter::makeCounter("/flow/fastalloc/deleteCalls"); + static SimpleCounter* bytes = SimpleCounter::makeCounter("/flow/fastalloc/deleteBytes"); + calls->increment(1); + bytes->increment(nbytes); + + delete[] reinterpret_cast(ptr); +} + template void FastAllocator::release(void* ptr) { if (keepalive_allocator::isActive()) [[unlikely]] return keepalive_allocator::invalidate(ptr); + // Accounting should mirror allocate() above. Only count paths where we + // allocated/free memory from lower levels or from our magazine cache. + static int size = Size; + static SimpleCounter* calls = + SimpleCounter::makeCounter(format("/flow/fastalloc/releaseCallsSize%d", size)); + static SimpleCounter* bytes = + SimpleCounter::makeCounter(format("/flow/fastalloc/releaseBytesSize%d", size)); + calls->increment(1); + bytes->increment(size); + #if defined(USE_GPERFTOOLS) || defined(ADDRESS_SANITIZER) return aligned_free(ptr); #endif @@ -592,6 +633,8 @@ void FastAllocator::getMagazine() { #else const bool includeGuardPages = true; #endif + // NOTE: rely on lower level metrics in allocate() (and whatever it calls) + // for accounting the allocations it does. block = (void**)::allocate(magazine_size * Size, /*allowLargePages*/ false, includeGuardPages); #endif diff --git a/flow/Net2.actor.cpp b/flow/Net2.actor.cpp index 1de1d784e5..19975b486a 100644 --- a/flow/Net2.actor.cpp +++ b/flow/Net2.actor.cpp @@ -321,16 +321,6 @@ public: DoubleMetricHandle countLaunchTime; DoubleMetricHandle countReactTime; BoolMetricHandle awakeMetric; - Int64MetricHandle countClientTLSHandshakesOnSideThreads; - Int64MetricHandle countClientTLSHandshakesOnMainThread; - Int64MetricHandle countServerTLSHandshakesOnSideThreads; - Int64MetricHandle countServerTLSHandshakesOnMainThread; - Int64MetricHandle countClientTLSHandshakesTimedout; - Int64MetricHandle countServerTLSHandshakesTimedout; - Int64MetricHandle countServerTLSHandshakeThrottled; - Int64MetricHandle countClientTLSHandshakeThrottled; - Int64MetricHandle countServerTLSHandshakeLocked; - Int64MetricHandle countClientTLSHandshakeLocked; EventMetricHandle slowTaskMetric; @@ -915,7 +905,9 @@ public: if (iter->second.first >= FLOW_KNOBS->TLS_CLIENT_CONNECTION_THROTTLE_ATTEMPTS) { TraceEvent("TLSOutgoingConnectionThrottlingWarning").suppressFor(1.0).detail("PeerIP", addr); wait(delay(FLOW_KNOBS->CONNECTION_MONITOR_TIMEOUT)); - g_net2->countClientTLSHandshakeThrottled++; + static SimpleCounter* countClientTLSHandshakeThrottled = + SimpleCounter::makeCounter("/Net2/TLS/ClientTLSHandshakeThrottled"); + countClientTLSHandshakeThrottled->increment(1); throw connection_failed(); } } else { @@ -968,7 +960,9 @@ public: // FIXME: see comment elsewhere about making this the only path. if ((FLOW_KNOBS->DISABLE_MAINTHREAD_TLS_HANDSHAKE && N2::g_net2->sslHandshakerThreadsStarted > 0) || N2::g_net2->sslPoolHandshakesInProgress < N2::g_net2->sslHandshakerThreadsStarted) { - g_net2->countServerTLSHandshakesOnSideThreads++; + static SimpleCounter* countServerTLSHandshakesOnSideThreads = + SimpleCounter::makeCounter("/Net2/TLS/ServerTLSHandshakesOnSideThreads"); + countServerTLSHandshakesOnSideThreads->increment(1); holder = Hold(&N2::g_net2->sslPoolHandshakesInProgress); auto handshake = new SSLHandshakerThread::Handshake(self->ssl_sock, boost::asio::ssl::stream_base::server); @@ -977,7 +971,9 @@ public: N2::g_net2->sslHandshakerPool->post(handshake); } else { // Otherwise use flow network thread - g_net2->countServerTLSHandshakesOnMainThread++; + static SimpleCounter* countServerTLSHandshakesOnMainThread = + SimpleCounter::makeCounter("/Net2/TLS/ServerTLSHandshakesOnMainThread"); + countServerTLSHandshakesOnMainThread->increment(1); BindPromise p("N2_AcceptHandshakeError"_audit, self->id); p.setPeerAddr(self->getPeerAddress()); onHandshook = p.getFuture(); @@ -1004,7 +1000,9 @@ public: .detail("PeerIP", peerIP.first.toString()); wait(delay(FLOW_KNOBS->CONNECTION_MONITOR_TIMEOUT)); self->closeSocket(); - g_net2->countServerTLSHandshakeThrottled++; + static SimpleCounter* countServerTLSHandshakeThrottled = + SimpleCounter::makeCounter("/Net2/TLS/ServerTLSHandshakeThrottled"); + countServerTLSHandshakeThrottled->increment(1); throw connection_failed(); } } else { @@ -1015,7 +1013,9 @@ public: wait(g_network->networkInfo.handshakeLock->take( getTaskPriorityFromInt(FLOW_KNOBS->TLS_HANDSHAKE_FLOWLOCK_PRIORITY))); state FlowLock::Releaser releaser(*g_network->networkInfo.handshakeLock); - g_net2->countServerTLSHandshakeLocked++; + static SimpleCounter* countServerTLSHandshakeLocked = + SimpleCounter::makeCounter("/Net2/TLS/ServerTLSHandshakeLocked"); + countServerTLSHandshakeLocked->increment(1); Promise connected; doAcceptHandshake(self, connected); @@ -1025,7 +1025,9 @@ public: return Void(); } when(wait(delay(FLOW_KNOBS->CONNECTION_MONITOR_TIMEOUT))) { - g_net2->countServerTLSHandshakesTimedout++; + static SimpleCounter* countServerTLSHandshakesTimedout = + SimpleCounter::makeCounter("/Net2/TLS/ServerTLSHandshakesTimedout"); + countServerTLSHandshakesTimedout->increment(1); throw connection_failed(); } } @@ -1068,7 +1070,9 @@ public: // thousand incremental threads. if ((FLOW_KNOBS->DISABLE_MAINTHREAD_TLS_HANDSHAKE && N2::g_net2->sslHandshakerThreadsStarted > 0) || N2::g_net2->sslPoolHandshakesInProgress < N2::g_net2->sslHandshakerThreadsStarted) { - g_net2->countClientTLSHandshakesOnSideThreads++; + static SimpleCounter* countClientTLSHandshakesOnSideThreads = + SimpleCounter::makeCounter("/Net2/TLS/ClientTLSHandshakesOnSideThreads"); + countClientTLSHandshakesOnSideThreads->increment(1); holder = Hold(&N2::g_net2->sslPoolHandshakesInProgress); auto handshake = new SSLHandshakerThread::Handshake(self->ssl_sock, boost::asio::ssl::stream_base::client); @@ -1077,7 +1081,9 @@ public: N2::g_net2->sslHandshakerPool->post(handshake); } else { // Otherwise use flow network thread - g_net2->countClientTLSHandshakesOnMainThread++; + static SimpleCounter* countClientTLSHandshakesOnMainThread = + SimpleCounter::makeCounter("/Net2/TLS/ClientTLSHandshakesOnMainThread"); + countClientTLSHandshakesOnMainThread->increment(1); BindPromise p("N2_ConnectHandshakeError"_audit, self->id); p.setPeerAddr(self->getPeerAddress()); onHandshook = p.getFuture(); @@ -1096,7 +1102,9 @@ public: wait(g_network->networkInfo.handshakeLock->take( getTaskPriorityFromInt(FLOW_KNOBS->TLS_HANDSHAKE_FLOWLOCK_PRIORITY))); state FlowLock::Releaser releaser(*g_network->networkInfo.handshakeLock); - g_net2->countClientTLSHandshakeLocked++; + static SimpleCounter* countClientTLSHandshakeLocked = + SimpleCounter::makeCounter("/Net2/TLS/ClientTLSHandshakeLocked"); + countClientTLSHandshakeLocked->increment(1); Promise connected; doConnectHandshake(self, connected); @@ -1106,7 +1114,9 @@ public: return Void(); } when(wait(delay(FLOW_KNOBS->CONNECTION_MONITOR_TIMEOUT))) { - g_net2->countClientTLSHandshakesTimedout++; + static SimpleCounter* countClientTLSHandshakesTimedout = + SimpleCounter::makeCounter("/Net2/TLS/ClientTLSHandshakesTimedout"); + countClientTLSHandshakesTimedout->increment(1); throw connection_failed(); } } @@ -1496,16 +1506,6 @@ void Net2::initMetrics() { slowTaskMetric.init("Net2.SlowTask"_sr); countLaunchTime.init("Net2.CountLaunchTime"_sr); countReactTime.init("Net2.CountReactTime"_sr); - countClientTLSHandshakesOnSideThreads.init("Net2.CountClientTLSHandshakesOnSideThreads"_sr); - countClientTLSHandshakesOnMainThread.init("Net2.CountClientTLSHandshakesOnMainThread"_sr); - countServerTLSHandshakesOnSideThreads.init("Net2.CountServerTLSHandshakesOnSideThreads"_sr); - countServerTLSHandshakesOnMainThread.init("Net2.CountServerTLSHandshakesOnMainThread"_sr); - countClientTLSHandshakesTimedout.init("Net2.CountClientTLSHandshakesTimedout"_sr); - countServerTLSHandshakesTimedout.init("Net2.CountServerTLSHandshakesTimedout"_sr); - countServerTLSHandshakeThrottled.init("Net2.CountServerTLSHandshakeThrottled"_sr); - countClientTLSHandshakeThrottled.init("Net2.CountClientTLSHandshakeThrottled"_sr); - countServerTLSHandshakeLocked.init("Net2.CountServerTLSHandshakeLocked"_sr); - countClientTLSHandshakeLocked.init("Net2.CountClientTLSHandshakeLocked"_sr); taskQueue.initMetrics(); } diff --git a/flow/Platform.actor.cpp b/flow/Platform.actor.cpp index a5d330f28c..b3e3fbd6a3 100644 --- a/flow/Platform.actor.cpp +++ b/flow/Platform.actor.cpp @@ -54,6 +54,7 @@ #include "flow/Knobs.h" #include "flow/Platform.actor.h" #include "flow/ScopeExit.h" +#include "flow/SimpleCounter.h" #include "flow/StreamCipher.h" #include "flow/Trace.h" #include "flow/Trace.h" @@ -2140,17 +2141,21 @@ static void mprotectSafe(void* p, size_t s, int prot) { } static void* mmapInternal(size_t length, int flags, bool guardPages) { + static SimpleCounter* bytes = SimpleCounter::makeCounter("/flow/platform/mmapBytes"); + if (guardPages && FLOW_KNOBS->FAST_ALLOC_ALLOW_GUARD_PAGES) { static size_t pageSize = sysconf(_SC_PAGESIZE); length = RightAlign(length, pageSize); length += 2 * pageSize; // Map enough for the guard pages void* resultWithGuardPages = mmapSafe(nullptr, length, PROT_READ | PROT_WRITE, flags, -1, 0); + bytes->increment(length); // left guard page mprotectSafe(resultWithGuardPages, pageSize, PROT_NONE); // right guard page mprotectSafe((void*)(uintptr_t(resultWithGuardPages) + length - pageSize), pageSize, PROT_NONE); return (void*)(uintptr_t(resultWithGuardPages) + pageSize); } else { + bytes->increment(length); return mmapSafe(nullptr, length, PROT_READ | PROT_WRITE, flags, -1, 0); } } @@ -3294,6 +3299,9 @@ void outOfMemory() { g_traceBatch.dump(); #endif + TraceEvent(SevWarn, "OutOfMemorySimpleCounterReportFollows"); + simpleCounterReport(SevWarn); + criticalError(FDB_EXIT_NO_MEM, "OutOfMemory", "Out of memory"); } diff --git a/flow/SimpleCounter.cpp b/flow/SimpleCounter.cpp new file mode 100644 index 0000000000..6f6b54b551 --- /dev/null +++ b/flow/SimpleCounter.cpp @@ -0,0 +1,205 @@ +/* + * SimpleCounter.cpp + * + * This source file is part of the FoundationDB open source project + * + * Copyright 2013-2025 Apple Inc. and the FoundationDB project authors + * + * Licensed 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. + */ + +#include + +#include "flow/SimpleCounter.h" +#include "flow/UnitTest.h" + +// Convert hierarchical metric names into Prometheus-compatible metric +// names. Do this by a) removing initial '/' chars, and b) converting +// remaining '/' chars to '_' chars. +static std::string hierarchicalToPrometheus(const std::string input) { + std::string output; + for (char ch : input) { + if (ch == '/') { + if (output.size() > 0) { + output += '_'; + } + } else { + output += ch; + } + } + return output; +} + +// ChatGPT generated code to return true iff `name` is a valid Prometheus +// metric name. Below we call this on trace event fields for the reason +// that we are contemplating converting fields in trace events to metrics +// in downstream Prometheus-compatible metrics systems. +static bool isValidPrometheusMetricName(std::string_view name) { + if (name.empty()) { + return false; + } + + // First character: [a-zA-Z_:] + char first = name.front(); + if (!(std::isalpha(static_cast(first)) || first == '_' || first == ':')) { + return false; + } + + // Rest: [a-zA-Z0-9_:]* + for (size_t i = 1; i < name.size(); ++i) { + char c = name[i]; + if (!(std::isalnum(static_cast(c)) || c == '_' || c == ':')) { + return false; + } + } + + // Reserved prefix check: names starting with "__" are reserved + if (name.size() >= 2 && name[0] == '_' && name[1] == '_') { + return false; + } + + return true; +} + +// This should be called periodically by higher level code somewhere. +void simpleCounterReport(Severity severity) { + static SimpleCounter* reportCount = SimpleCounter::makeCounter("/flow/counters/reports"); + reportCount->increment(1); + + std::vector*> intCounters = SimpleCounter::getCounters(); + std::vector*> doubleCounters = SimpleCounter::getCounters(); + + int i = 0; + // Avoid trace buffer overflow by assuming average field is O(100) bytes or less. + int countersPerTrace = FLOW_KNOBS->MAX_TRACE_EVENT_LENGTH / 100; + while (i < intCounters.size()) { + auto traceEvent = TraceEvent(severity, "SimpleCounters"); + int c = 0; + do { + SimpleCounter* ic = intCounters[i]; + std::string n = ic->name(); + n = hierarchicalToPrometheus(n); + ASSERT(isValidPrometheusMetricName(n)); + traceEvent.detail(std::move(n), ic->get()); + i++; + c++; + } while (i < intCounters.size() && c < countersPerTrace); + } + i = 0; + while (i < doubleCounters.size()) { + auto traceEvent = TraceEvent(severity, "SimpleCounters"); + int c = 0; + do { + SimpleCounter* dc = doubleCounters[i]; + std::string n = dc->name(); + n = hierarchicalToPrometheus(n); + ASSERT(isValidPrometheusMetricName(n)); + traceEvent.detail(std::move(n), dc->get()); + i++; + c++; + } while (i < doubleCounters.size() && c < countersPerTrace); + } +} + +TEST_CASE("/flow/simplecounter/int64") { + SimpleCounter* foo = SimpleCounter::makeCounter("/flow/counters/foo"); + SimpleCounter* bar = SimpleCounter::makeCounter("/flow/counters/bar"); + + foo->increment(5); + foo->increment(1); + + bar->increment(10); + + ASSERT(foo->get() == 6); + ASSERT(bar->get() == 10); + + for (int i = 0; i < 100; i++) { + SimpleCounter* p = + SimpleCounter::makeCounter(std::string("/flow/counters/many") + std::to_string(i)); + p->increment(i); + ASSERT(p->get() == i); + } + + SimpleCounter* conflict = SimpleCounter::makeCounter("/flow/counters/lots"); + + // Increment by all values in [1, 1000000] across 10 threads. + // Expected sum: 10 * (min + max) * (num entries in series)/2 + // ==> 10 * ( 1 + 1M ) * 500K + int64_t expectedSum = int64_t{ 10 } * int64_t{ 1'000'000 + 1 } * uint64_t{ 500'000 }; + + auto inclots = [conflict]() { + for (int i = 1; i <= 1'000'000; i++) { + conflict->increment(i); + } + }; + + std::vector threads; + for (int i = 0; i < 10; i++) { + threads.emplace_back(inclots); + } + for (int i = 0; i < 10; i++) { + threads[i].join(); + } + + ASSERT(conflict->get() == expectedSum); + + std::vector*> intCounters = SimpleCounter::getCounters(); + + // NOTE: the following is written as >= 103 and not == 103 because + // "unit tests" actually run in fdbserver, so any background + // logic, like for example simpleCounterReport() being called + // above from fdbserver.actor.cpp, will affect the execution + // environment. + ASSERT(intCounters.size() >= 103); + + // Give asserts here a chance to run. + simpleCounterReport(); + + return Void(); +} + +TEST_CASE("/flow/simplecounter/double") { + SimpleCounter* baz = SimpleCounter::makeCounter("/flow/counters/baz"); + + // We intend to compute a floating point sum with an exact representation. + // A way to do this is to only add values with exact representations. + // Integers and powers of two (within the limits of the number of mantissa + // and exponent bits) do have exact representations. Hence the 0.5 increment. + double expectedSum = 10.0 * (0.5 + 1'000'000.0) * 1'000'000.0; + + auto double_inclots = [baz]() { + for (double i = 0.5; i <= 1'000'000; i += 0.5) { + baz->increment(i); + } + }; + + std::vector threads; + for (int i = 0; i < 10; i++) { + threads.emplace_back(double_inclots); + } + for (int i = 0; i < 10; i++) { + threads[i].join(); + } + + ASSERT(baz->get() == expectedSum); + + std::vector*> doubleCounters = SimpleCounter::getCounters(); + ASSERT(doubleCounters.size() >= 1); + + // Give asserts here a chance to run. + simpleCounterReport(); + + return Void(); +} + +void forceLinkSimpleCounterTests() {} diff --git a/flow/SystemMonitor.cpp b/flow/SystemMonitor.cpp index 8719253b0a..96ad8e3c40 100644 --- a/flow/SystemMonitor.cpp +++ b/flow/SystemMonitor.cpp @@ -182,80 +182,6 @@ SystemStatistics customSystemMonitor(std::string const& eventName, StatisticsSta .detail("TLSPolicyFailures", (netData.countTLSPolicyFailures - statState->networkState.countTLSPolicyFailures) / currentStats.elapsed) - .detail("ClientTLSHandshakesOnSideThreads", - (netData.countClientTLSHandshakesOnSideThreads - - statState->networkState.countClientTLSHandshakesOnSideThreads) / - currentStats.elapsed) - .detail("ClientTLSHandshakesOnMainThread", - (netData.countClientTLSHandshakesOnMainThread - - statState->networkState.countClientTLSHandshakesOnMainThread) / - currentStats.elapsed) - .detail("ServerTLSHandshakesOnSideThreads", - (netData.countServerTLSHandshakesOnSideThreads - - statState->networkState.countServerTLSHandshakesOnSideThreads) / - currentStats.elapsed) - .detail("ServerTLSHandshakesOnMainThread", - (netData.countServerTLSHandshakesOnMainThread - - statState->networkState.countServerTLSHandshakesOnMainThread) / - currentStats.elapsed) - .detail("ConnectionIncompatible", - (netData.countConnIncompatible - statState->networkState.countConnIncompatible) / - currentStats.elapsed) - .detail("ConnectionIncompatibleWithOldClient", - (netData.countConnIncompatibleWithOldClient - - statState->networkState.countConnIncompatibleWithOldClient) / - currentStats.elapsed) - .detail("ClientTLSHandshakesTimedout", - (netData.countClientTLSHandshakesTimedout - - statState->networkState.countClientTLSHandshakesTimedout) / - currentStats.elapsed) - .detail("ServerTLSHandshakesTimedout", - (netData.countServerTLSHandshakesTimedout - - statState->networkState.countServerTLSHandshakesTimedout) / - currentStats.elapsed) - .detail("ConnectionHandshakeAccepted", - (netData.countConnHandshakeAccepted - statState->networkState.countConnHandshakeAccepted) / - currentStats.elapsed) - .detail("ConnectionHandshakeRequested", - (netData.countConnHandshakeRequested - statState->networkState.countConnHandshakeRequested) / - currentStats.elapsed) - .detail("IncomingConnRequested", - (netData.countIncomingConnRequested - statState->networkState.countIncomingConnRequested) / - currentStats.elapsed) - .detail("IncomingConnAccepted", - (netData.countIncomingConnAccepted - statState->networkState.countIncomingConnAccepted) / - currentStats.elapsed) - .detail("ServerTLSHandshakeThrottled", - (netData.countServerTLSHandshakeThrottled - - statState->networkState.countServerTLSHandshakeThrottled) / - currentStats.elapsed) - .detail("ClientTLSHandshakeThrottled", - (netData.countClientTLSHandshakeThrottled - - statState->networkState.countClientTLSHandshakeThrottled) / - currentStats.elapsed) - .detail("OutgoingConnHandshakeComplete", - (netData.countOutgoingConnHandshakeComplete - - statState->networkState.countOutgoingConnHandshakeComplete) / - currentStats.elapsed) - .detail("OutgoingConnHandshakeRequested", - (netData.countOutgoingConnHandshakeRequested - - statState->networkState.countOutgoingConnHandshakeRequested) / - currentStats.elapsed) - .detail("IncomingConnectionTimedout", - (netData.countIncomingConnectionTimedout - - statState->networkState.countIncomingConnectionTimedout) / - currentStats.elapsed) - .detail( - "ServerTLSHandshakeLocked", - (netData.countServerTLSHandshakeLocked - statState->networkState.countServerTLSHandshakeLocked) / - currentStats.elapsed) - .detail( - "ClientTLSHandshakeLocked", - (netData.countClientTLSHandshakeLocked - statState->networkState.countClientTLSHandshakeLocked) / - currentStats.elapsed) - .detail("IncomingConnConnected", - (netData.countIncomingConnConnected - statState->networkState.countIncomingConnConnected) / - currentStats.elapsed) .trackLatest(eventName); diff --git a/flow/Trace.cpp b/flow/Trace.cpp index 0cd158e2db..149ae9ca65 100644 --- a/flow/Trace.cpp +++ b/flow/Trace.cpp @@ -265,7 +265,6 @@ public: }; void action(WriteBuffer& a) { for (const auto& event : a.events) { - event.validateFormat(); logWriter->write(formatter->formatEvent(event)); } @@ -1785,39 +1784,6 @@ std::string TraceEventFields::toString() const { return str; } -bool validateField(const char* key, bool allowUnderscores) { - if ((key[0] < 'A' || key[0] > 'Z') && key[0] != '_') { - return false; - } - - const char* underscore = strchr(key, '_'); - while (underscore) { - if (!allowUnderscores || ((underscore[1] < 'A' || underscore[1] > 'Z') && key[0] != '_' && key[0] != '\0')) { - return false; - } - - underscore = strchr(&underscore[1], '_'); - } - - return true; -} - -void TraceEventFields::validateFormat() const { - if (g_network && g_network->isSimulated()) { - for (Field field : fields) { - if (!validateField(field.first.c_str(), false)) { - fprintf(stderr, - "Trace event detail name `%s' is invalid in:\n\t%s\n", - field.first.c_str(), - toString().c_str()); - } - if (field.first == "Type" && !validateField(field.second.c_str(), true)) { - fprintf(stderr, "Trace event detail Type `%s' is invalid\n", field.second.c_str()); - } - } - } -} - std::string traceableStringToString(const char* value, size_t S) { if (g_network) { ASSERT_WE_THINK(S > 0 && value[S - 1] == '\0'); diff --git a/flow/include/flow/Arena.h b/flow/include/flow/Arena.h index 8c0d9d2693..b7666d93a2 100644 --- a/flow/include/flow/Arena.h +++ b/flow/include/flow/Arena.h @@ -31,6 +31,7 @@ #include "flow/FileIdentifier.h" #include "flow/swift_support.h" #include "flow/Optional.h" +#include "flow/SimpleCounter.h" #include "flow/Traceable.h" #include #include @@ -290,8 +291,6 @@ struct union_like_traits> : std::true_type { } }; -// #define STANDALONE_ALWAYS_COPY - template class Standalone : private Arena, public T { public: @@ -316,30 +315,12 @@ public: return *this; } -// Always-copy mode was meant to make alloc instrumentation more useful by making allocations occur at the final resting -// place of objects leaked It doesn't actually work because some uses of Standalone things assume the object's memory -// will not change on copy or assignment -#ifdef STANDALONE_ALWAYS_COPY - // Treat Standalone's as T's in construction and assignment so the memory is copied - Standalone(const T& t, const Arena& arena) : Standalone(t) {} - Standalone(const Standalone& t) : Standalone((T const&)t) {} - Standalone(const Standalone&& t) : Standalone((T const&)t) {} - Standalone& operator=(const Standalone&& t) { - *this = (T const&)t; - return *this; - } - Standalone& operator=(const Standalone& t) { - *this = (T const&)t; - return *this; - } -#else Standalone(const T& t, const Arena& arena) : Arena(arena), T(t) {} Standalone(const Standalone&) = default; Standalone& operator=(const Standalone&) = default; Standalone(Standalone&&) = default; Standalone& operator=(Standalone&&) = default; ~Standalone() = default; -#endif template Standalone castTo() const { @@ -355,11 +336,6 @@ public: serializer(ar, (*(T*)this), arena()); } - /*static Standalone fakeStandalone( const T& t ) { - Standalone x; - *(T*)&x = t; - return x; - }*/ private: template Standalone(Standalone const&); // unimplemented @@ -371,22 +347,33 @@ extern std::string format(const char* form, ...); #pragma pack(push, 4) class StringRef { +private: + static SimpleCounter* bytesCopied() { + static SimpleCounter* bytesCopied = + SimpleCounter::makeCounter("/flow/arena/stringRefBytesCopied"); + return bytesCopied; + } + public: constexpr static FileIdentifier file_identifier = 13300811; StringRef() : data(0), length(0) {} StringRef(Arena& p, const StringRef& toCopy) : data(new(p) uint8_t[toCopy.size()]), length(toCopy.size()) { if (length > 0) { + bytesCopied()->increment(length); memcpy((void*)data, toCopy.data, length); } } StringRef(Arena& p, const std::string& toCopy) : length((int)toCopy.size()) { UNSTOPPABLE_ASSERT(toCopy.size() <= std::numeric_limits::max()); data = new (p) uint8_t[toCopy.size()]; - if (length) + if (length) { + bytesCopied()->increment(length); memcpy((void*)data, &toCopy[0], length); + } } StringRef(Arena& p, const uint8_t* toCopy, int length) : data(new(p) uint8_t[length]), length(length) { if (length > 0) { + bytesCopied()->increment(length); memcpy((void*)data, toCopy, length); } } @@ -422,7 +409,10 @@ public: } StringRef withPrefix(const StringRef& prefix, Arena& arena) const { - uint8_t* s = new (arena) uint8_t[prefix.size() + size()]; + size_t len = prefix.size() + size(); + uint8_t* s = new (arena) uint8_t[len]; + bytesCopied()->increment(len); + if (prefix.size() > 0) { memcpy(s, prefix.begin(), prefix.size()); } @@ -433,7 +423,9 @@ public: } StringRef withSuffix(const StringRef& suffix, Arena& arena) const { - uint8_t* s = new (arena) uint8_t[suffix.size() + size()]; + size_t len = suffix.size() + size(); + uint8_t* s = new (arena) uint8_t[len]; + bytesCopied()->increment(len); if (size() > 0) { memcpy(s, begin(), size()); } @@ -467,7 +459,10 @@ public: return substr(0, size() - s.size()); } - std::string toString() const { return std::string(reinterpret_cast(data), length); } + std::string toString() const { + bytesCopied()->increment(length); + return std::string(reinterpret_cast(data), length); + } std::string_view toStringView() const { return std::string_view(reinterpret_cast(data), length); } @@ -477,6 +472,7 @@ public: std::string toHexString(int limit = -1) const { if (limit < 0) limit = length; + std::string rv; if (length > limit) { // If limit is high enough split it so that 2/3 of limit is used to show prefix bytes and the rest is used // for suffix bytes @@ -485,21 +481,21 @@ public: return substr(0, limit - suffix).toHexString() + "..." + substr(length - suffix, suffix).toHexString() + format(" [%d bytes]", length); } - return substr(0, limit).toHexString() + format("...[%d]", length); + rv = substr(0, limit).toHexString() + format("...[%d]", length); + } else { + rv.reserve(length * 7); + for (int i = 0; i < length; i++) { + uint8_t b = (*this)[i]; + if (isalnum(b)) + rv.append(format("%02x (%c) ", b, b)); + else + rv.append(format("%02x ", b)); + } + if (rv.size() > 0) + rv.resize(rv.size() - 1); } - - std::string s; - s.reserve(length * 7); - for (int i = 0; i < length; i++) { - uint8_t b = (*this)[i]; - if (isalnum(b)) - s.append(format("%02x (%c) ", b, b)); - else - s.append(format("%02x ", b)); - } - if (s.size() > 0) - s.resize(s.size() - 1); - return s; + bytesCopied()->increment(rv.length()); + return rv; } // Get string with full content in hex format. Different digits are splitted by a space. @@ -513,6 +509,7 @@ public: } if (s.size() > 0) s.resize(s.size() - 1); + bytesCopied()->increment(s.length()); return s; } @@ -600,6 +597,7 @@ public: // Copies string contents to dst and returns a pointer to the next byte after uint8_t* copyTo(uint8_t* dst) const { if (length > 0) { + bytesCopied()->increment(length); memcpy(dst, data, length); } return dst + length; @@ -953,6 +951,8 @@ struct VectorRefPreserializer { void reset() { _cached_size = 0; } }; +// FIXME: consider whether methods in this class should be instrumented to +// count calls, bytes processed, etc. template class VectorRef : public ComposedIdentifier, public VectorRefPreserializer { using VPS = VectorRefPreserializer; @@ -1219,6 +1219,8 @@ protected: // that all of them are always copied. This should be faster // when you expect the vector to be usually very small as it // won't need allocations in these cases. +// FIXME: assess whether methods in this class should be instrumented +// with metrics. Currently this appears to be thinly used. template class SmallVectorRef { static_assert(InlineMembers >= 0); diff --git a/flow/include/flow/FastAlloc.h b/flow/include/flow/FastAlloc.h index 7ccb94a088..a0e7aa0144 100644 --- a/flow/include/flow/FastAlloc.h +++ b/flow/include/flow/FastAlloc.h @@ -24,6 +24,7 @@ #include "flow/Error.h" #include "flow/Platform.h" +#include "flow/SimpleCounter.h" #include "flow/config.h" // ALLOC_INSTRUMENTATION_STDOUT enables non-sampled logging of all allocations and deallocations to stdout to be @@ -33,6 +34,9 @@ // #define ALLOC_INSTRUMENTATION ENABLED(NOT_IN_CLEAN) // The form "(1==1)" in this context is used to satisfy both clang and vc++ with a single syntax. Clang rejects "1" // and vc++ rejects "true". +// FIXME: this has been set to true for 4+ years. We probably do not need the "not thread safe" +// version of the code. Consider removing this and just making it thread safe. +// Also, explain why thread safety is required here and not elsewhere (e.g. Arena and ArenaBlock). #define FASTALLOC_THREAD_SAFE (FLOW_THREAD_SAFE || (1 == 1)) #if VALGRIND @@ -181,6 +185,11 @@ void hugeArenaSample(int size); void releaseAllThreadMagazines(); int64_t getTotalUnusedAllocatedMemory(); +// These are thin wrappers around operator new and operator delete +// and exist so that we can update metrics inside them. +void* countedNew(size_t nbytes); +void countedDelete(size_t nbytes, void* ptr); + // Allow temporary overriding of default allocators used by arena to let memory survive deallocation and test // correctness of memory policy (e.g. zeroing out sensitive contents after use) namespace keepalive_allocator { @@ -214,6 +223,12 @@ std::vector> const& getWipedAreaSet(); } // namespace keepalive_allocator force_inline uint8_t* allocateAndMaybeKeepalive(size_t size) { + static SimpleCounter* calls = + SimpleCounter::makeCounter("/flow/fastalloc/AllocateAndMaybeKeepaliveCalls"); + static SimpleCounter* bytes = + SimpleCounter::makeCounter("/flow/fastalloc/AllocateAndMaybeKeepaliveBytes"); + calls->increment(1); + bytes->increment(size); uint8_t* p; if (keepalive_allocator::isActive()) [[unlikely]] p = static_cast(keepalive_allocator::allocate(size)); @@ -228,6 +243,9 @@ force_inline uint8_t* allocateAndMaybeKeepalive(size_t size) { } force_inline void freeOrMaybeKeepalive(void* ptr) { + static SimpleCounter* calls = + SimpleCounter::makeCounter("/flow/fastalloc/FreeOrMaybeKeepaliveCalls"); + calls->increment(1); if (keepalive_allocator::isActive()) [[unlikely]] keepalive_allocator::invalidate(ptr); else @@ -262,6 +280,13 @@ inline constexpr int nextFastAllocatedSize(int x) { return 16384; } +// NOTE: for maintaining metrics on objects/bytes allocated and +// deleted, we rely on FastAllocated::allocate, +// FastAllocated::release, countedNew(), and countedDelete() to +// update metrics for code paths that do cause allocations or frees. +// Do not add uninstrumented operator new or operator delete +// invocations. + template class FastAllocated { public: @@ -274,7 +299,7 @@ public: void* p = FastAllocator < sizeof(Object) <= 64 ? 64 : nextFastAllocatedSize(sizeof(Object)) > ::allocate(); return p; } else { - void* p = new uint8_t[nextFastAllocatedSize(sizeof(Object))]; + void* p = countedNew(nextFastAllocatedSize(sizeof(Object))); return p; } } @@ -285,7 +310,7 @@ public: if constexpr (sizeof(Object) <= 256) { FastAllocator::release(s); } else { - delete[] reinterpret_cast(s); + countedDelete(nextFastAllocatedSize(sizeof(Object)), s); } } // Redefine placement new so you can still use it @@ -306,7 +331,7 @@ public: return FastAllocator<128>::allocate(); if (size <= 256) return FastAllocator<256>::allocate(); - return new uint8_t[size]; + return countedNew(size); } inline void freeFast(int size, void* ptr) { @@ -322,12 +347,16 @@ inline void freeFast(int size, void* ptr) { return FastAllocator<128>::release(ptr); if (size <= 256) return FastAllocator<256>::release(ptr); - delete[] (uint8_t*)ptr; + countedDelete(size, ptr); } // Allocate a block of memory aligned to 4096 bytes. Size must be a multiple of // 4096. Guaranteed not to return null. Use freeFast4kAligned to free. [[nodiscard]] inline void* allocateFast4kAligned(int size) { + static SimpleCounter* bytes = + SimpleCounter::makeCounter("/flow/fastalloc/Fast4AlignedBytesAllocated"); + bytes->increment(size); + #if !defined(USE_JEMALLOC) // Use FastAllocator for sizes it supports to avoid internal fragmentation in some implementations of aligned_alloc if (size <= 4096) @@ -337,6 +366,7 @@ inline void freeFast(int size, void* ptr) { if (size <= 16384) return FastAllocator<16384>::allocate(); #endif + auto* result = aligned_alloc(4096, size); if (result == nullptr) { platform::outOfMemory(); @@ -346,6 +376,10 @@ inline void freeFast(int size, void* ptr) { // Free a pointer returned from allocateFast4kAligned(size) inline void freeFast4kAligned(int size, void* ptr) { + static SimpleCounter* bytes = + SimpleCounter::makeCounter("/flow/fastalloc/Fast4AlignedBytesFreed"); + bytes->increment(size); + #if !defined(USE_JEMALLOC) // Sizes supported by FastAllocator must be release via FastAllocator if (size <= 4096) diff --git a/flow/include/flow/SimpleCounter.h b/flow/include/flow/SimpleCounter.h new file mode 100644 index 0000000000..c20bc93f09 --- /dev/null +++ b/flow/include/flow/SimpleCounter.h @@ -0,0 +1,150 @@ +/* + * SimpleCounter.h + * + * This source file is part of the FoundationDB open source project + * + * Copyright 2013-2025 Apple Inc. and the FoundationDB project authors + * + * Licensed 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. + */ + +#ifndef FLOW_SIMPLECOUNTER_H +#define FLOW_SIMPLECOUNTER_H +#pragma once + +#include +#include +#include + +#include "flow/Error.h" +#include "flow/Trace.h" + +// SimpleCounter metrics class for atomic counters of int64_t or +// double. Example usage: +// +// static SimpleCounter *foo = SimpleCounter::makeCounter("/mymodule/foo"); +// +// if (...) { +// // condition of interest +// foo->increment(1); +// ... +// } +// +// This class is thread safe, i.e. can be used by code which does limited-scope +// synchronous work in side threads, but is intended to generally be very +// light weight. `makeCounter` can be called in constructors of global objects. +// +// If you want to use hierarchical metric names (e.g., '/'-separated +// components), please use ALL LOWER CASE METRIC NAMES AS PER THE EXAMPLE ABOVE. +// This enables the implementation to smuggle path component +// separators into the trace output by replacing path separaters like '/' with +// carefully chosen capital letters. This obtains compatibility with current FDB +// "field name" naming conventions. +// +// In the future we might replace '/' with '_' to obtain Prometheus-compatible +// metric names that don't actually look terrible. +// +// If you don't want to use hierarchical metric names, then your counter +// names should be ReallyVerboseConcatenatedNamesWithCaps and must be globally +// unique. +// +// SimpleCounter* returned by `makeCounter` are intended to live for the +// duration of the process, i.e. they are not intended to be freed/destroyed. +// +// Counters are periodically logged as "SimpleCounters". +// +// More background: https://quip-apple.com/PyfZA6Qkbc7w +// +// Caveat: if you allocate two different counters with the same name, they will +// accumulate updates independently. You probably don't want to do that. +// +// FIXME: add support for metric labels. This can be done by letting the +// template take 0 or more additional string typed arguments which represent +// label dimension names. The increment() API below would require that the +// same number of string-valued arguments (or arguments convertable to string) +// be provided and would remember those as labels. + +template +class SimpleCounter { + static_assert(std::is_same_v || std::is_same_v, "T must be int64_t or double"); + +private: + SimpleCounter(std::string_view n) : value(T(0)), name_(n) {} + + // Not copyable or movable. + SimpleCounter(const SimpleCounter&) = delete; + SimpleCounter& operator=(const SimpleCounter&) = delete; + SimpleCounter(SimpleCounter&&) = delete; + SimpleCounter& operator=(SimpleCounter&&) = delete; + +private: + std::atomic value; + std::string name_; + + // Protects the static object returned by counters() below. + // https://chatgpt.com/share/68acec3c-21d4-800b-b315-ff6fc45ec806 + // explains why this is necessary. + static inline std::mutex& mutex() { + static std::mutex m; + return m; + } + static inline std::vector*>& counters() { + static std::vector*> v; + return v; + } + +public: + // Defined in template instantiations below. + inline void increment(T delta); + + inline T get(void) const { return value.load(); } + + inline const std::string& name(void) const { return name_; } + + static inline SimpleCounter* makeCounter(std::string_view name) { + SimpleCounter* rv = new SimpleCounter(name); + + std::lock_guard lock(mutex()); + std::vector*>& v = counters(); + v.push_back(rv); + + return rv; + } + + static inline std::vector*> getCounters() { + std::vector*> rv; + std::lock_guard lock(mutex()); + rv = counters(); + return rv; + } +}; + +template <> +inline void SimpleCounter::increment(int64_t delta) { + value.fetch_add(delta, std::memory_order_relaxed); +} + +// Newer versions of C++ allow `fetch_add` on double, but older +// versions don't. This is not expected to cause performance issues in +// practice. +template <> +inline void SimpleCounter::increment(double delta) { + double old = value.load(); + while (!value.compare_exchange_weak(old, old + delta)) { + ; + } +} + +void simpleCounterReport(Severity severity = SevInfo); + +#endif diff --git a/flow/include/flow/SystemMonitor.h b/flow/include/flow/SystemMonitor.h index 4ff067b2a5..6dae467ce3 100644 --- a/flow/include/flow/SystemMonitor.h +++ b/flow/include/flow/SystemMonitor.h @@ -95,28 +95,6 @@ struct NetworkData { int64_t countTLSPolicyFailures; double countLaunchTime; double countReactTime; - int64_t countClientTLSHandshakesOnSideThreads; - int64_t countClientTLSHandshakesOnMainThread; - int64_t countServerTLSHandshakesOnSideThreads; - int64_t countServerTLSHandshakesOnMainThread; - int64_t countConnIncompatible; - int64_t countConnIncompatibleWithOldClient; // Increments when a very old client connects to fdbserver with - // incompatible protocol version error. Please check the definition of - // hasInexpensiveMultiVersionClient. - int64_t countClientTLSHandshakesTimedout; - int64_t countServerTLSHandshakesTimedout; - int64_t countConnHandshakeAccepted; - int64_t countConnHandshakeRequested; - int64_t countIncomingConnRequested; - int64_t countIncomingConnAccepted; - int64_t countServerTLSHandshakeThrottled; - int64_t countClientTLSHandshakeThrottled; - int64_t countOutgoingConnHandshakeComplete; - int64_t countOutgoingConnHandshakeRequested; - int64_t countIncomingConnectionTimedout; - int64_t countServerTLSHandshakeLocked; - int64_t countClientTLSHandshakeLocked; - int64_t countIncomingConnConnected; void init() { bytesSent = Int64Metric::getValueOrDefault("Net2.BytesSent"_sr); @@ -159,33 +137,6 @@ struct NetworkData { countFilePageCacheHits = Int64Metric::getValueOrDefault("AsyncFile.CountCachePageReadsHit"_sr); countFilePageCacheMisses = Int64Metric::getValueOrDefault("AsyncFile.CountCachePageReadsMissed"_sr); countFilePageCacheEvictions = Int64Metric::getValueOrDefault("EvictablePageCache.CacheEvictions"_sr); - countClientTLSHandshakesOnSideThreads = - Int64Metric::getValueOrDefault("Net2.CountClientTLSHandshakesOnSideThreads"_sr); - countClientTLSHandshakesOnMainThread = - Int64Metric::getValueOrDefault("Net2.CountClientTLSHandshakesOnMainThread"_sr); - countServerTLSHandshakesOnSideThreads = - Int64Metric::getValueOrDefault("Net2.CountServerTLSHandshakesOnSideThreads"_sr); - countServerTLSHandshakesOnMainThread = - Int64Metric::getValueOrDefault("Net2.CountServerTLSHandshakesOnMainThread"_sr); - countConnIncompatible = Int64Metric::getValueOrDefault("Net2.CountConnIncompatible"_sr); - countConnIncompatibleWithOldClient = - Int64Metric::getValueOrDefault("Net2.CountConnIncompatibleWithOldClient"_sr); - countClientTLSHandshakesTimedout = Int64Metric::getValueOrDefault("Net2.CountClientTLSHandshakesTimedout"_sr); - countServerTLSHandshakesTimedout = Int64Metric::getValueOrDefault("Net2.CountServerTLSHandshakesTimedout"_sr); - countConnHandshakeAccepted = Int64Metric::getValueOrDefault("Net2.CountConnHandshakeAccepted"_sr); - countConnHandshakeRequested = Int64Metric::getValueOrDefault("Net2.CountConnHandshakeRequested"_sr); - countIncomingConnRequested = Int64Metric::getValueOrDefault("Net2.CountIncomingConnRequested"_sr); - countIncomingConnAccepted = Int64Metric::getValueOrDefault("Net2.CountIncomingConnAccepted"_sr); - countServerTLSHandshakeThrottled = Int64Metric::getValueOrDefault("Net2.CountServerTLSHandshakeThrottled"_sr); - countClientTLSHandshakeThrottled = Int64Metric::getValueOrDefault("Net2.CountClientTLSHandshakeThrottled"_sr); - countOutgoingConnHandshakeComplete = - Int64Metric::getValueOrDefault("Net2.CountOutgoingConnHandshakeComplete"_sr); - countOutgoingConnHandshakeRequested = - Int64Metric::getValueOrDefault("Net2.CountOutgoingConnHandshakeRequested"_sr); - countIncomingConnectionTimedout = Int64Metric::getValueOrDefault("Net2.CountIncomingConnectionTimedout"_sr); - countServerTLSHandshakeLocked = Int64Metric::getValueOrDefault("Net2.CountServerTLSHandshakeLocked"_sr); - countClientTLSHandshakeLocked = Int64Metric::getValueOrDefault("Net2.CountClientTLSHandshakeLocked"_sr); - countIncomingConnConnected = Int64Metric::getValueOrDefault("Net2.CountIncomingConnConnected"_sr); } }; diff --git a/flow/include/flow/UnitTest.h b/flow/include/flow/UnitTest.h index c0fdd0d221..51626116da 100644 --- a/flow/include/flow/UnitTest.h +++ b/flow/include/flow/UnitTest.h @@ -25,9 +25,19 @@ /* * Flow unit testing framework * - * This is an *extremely* lightweight framework for writing optionally asynchronous, + * This is a simple framework for writing optionally asynchronous, * optionally randomized unit tests. * + * This framework is not trivial to use correctly. For example, your + * unit tests will affect the global execution environment of a + * fdbserver process. If things done in your unit test are not in + * accordance with global expectations that are only enabled in + * simulation, then you may break simulation even though your unit + * test itself runs fine via fdbserver -r unittests. As a result, to test + * that your unit tests themselves do not break simulation, you should + * also run a 100k simulation run. If you think this sounds + * backwards, you may be right. + * * Usage: * * TEST_CASE("/product/module/testcase") {