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") {