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 <zhe.wang@wustl.edu>
This commit is contained in:
gxglass
GitHub
parent
dfa51e599b
commit
ff19220787
|
|
@ -94,6 +94,7 @@ flow/coveragetool/obj
|
|||
/.ccls-cache
|
||||
/.clangd
|
||||
/.cache
|
||||
TAGS
|
||||
|
||||
# Temporary and user configuration files
|
||||
*~
|
||||
|
|
|
|||
|
|
@ -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 );
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
|
|
|
|||
|
|
@ -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<struct Peer> 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<NetworkAddress, std::pair<uint64_t, double>> incompatiblePeers;
|
||||
AsyncTrigger incompatiblePeersChanged;
|
||||
|
|
@ -832,9 +812,14 @@ ACTOR Future<Void> connectionKeeper(Reference<Peer> self,
|
|||
when(Reference<IConnection> _conn =
|
||||
wait(INetworkConnections::net()->connect(self->destination))) {
|
||||
conn = _conn;
|
||||
self->transport->countOutgoingConnHandshakeRequested++;
|
||||
static SimpleCounter<int64_t>* countOutgoingConnectionCreated =
|
||||
SimpleCounter<int64_t>::makeCounter("/Transport/TLS/OutgoingConnectionCreated");
|
||||
countOutgoingConnectionCreated->increment(1);
|
||||
wait(conn->connectHandshake());
|
||||
self->transport->countOutgoingConnHandshakeComplete++;
|
||||
static SimpleCounter<int64_t>* countOutgoingConnectionHandshakeComplete =
|
||||
SimpleCounter<int64_t>::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<Void> connectionReader(TransportData* transport,
|
|||
now() + FLOW_KNOBS->CONNECTION_ID_TIMEOUT;
|
||||
}
|
||||
compatible = false;
|
||||
transport->countConnIncompatible++;
|
||||
static SimpleCounter<int64_t>* countConnectionIncompatible =
|
||||
SimpleCounter<int64_t>::makeCounter("/Transport/TLS/ConnectionIncompatible");
|
||||
countConnectionIncompatible->increment(1);
|
||||
if (!protocolVersion.hasInexpensiveMultiVersionClient()) {
|
||||
if (peer) {
|
||||
peer->protocolVersion->set(protocolVersion);
|
||||
}
|
||||
transport->countConnIncompatibleWithOldClient++;
|
||||
static SimpleCounter<int64_t>* countConnectionIncompatibleWithVeryOldClient =
|
||||
SimpleCounter<int64_t>::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<Void> connectionIncoming(TransportData* self, Reference<ICon
|
|||
entry.time = now();
|
||||
entry.addr = conn->getPeerAddress();
|
||||
try {
|
||||
self->countConnHandshakeRequested++;
|
||||
wait(conn->acceptHandshake());
|
||||
self->countConnHandshakeAccepted++;
|
||||
static SimpleCounter<int64_t>* countIncomingConnectionHandshakeAccepted =
|
||||
SimpleCounter<int64_t>::makeCounter("/Transport/TLS/IncomingConnectionHandshakeAccepted");
|
||||
countIncomingConnectionHandshakeAccepted->increment(1);
|
||||
state Promise<Reference<Peer>> onConnected;
|
||||
state Future<Void> reader = connectionReader(self, conn, Reference<Peer>(), onConnected);
|
||||
if (FLOW_KNOBS->LOG_CONNECTION_ATTEMPTS_ENABLED) {
|
||||
|
|
@ -1643,17 +1634,24 @@ ACTOR static Future<Void> connectionIncoming(TransportData* self, Reference<ICon
|
|||
}
|
||||
when(wait(delayJittered(FLOW_KNOBS->CONNECTION_MONITOR_TIMEOUT))) {
|
||||
CODE_PROBE(true, "Incoming connection timed out");
|
||||
self->countIncomingConnectionTimedout++;
|
||||
static SimpleCounter<int64_t>* countIncomingConnectionTimedout =
|
||||
SimpleCounter<int64_t>::makeCounter("/Transport/TLS/IncomingConnectionTimedout");
|
||||
countIncomingConnectionTimedout->increment(1);
|
||||
throw timed_out();
|
||||
}
|
||||
}
|
||||
self->countIncomingConnConnected++;
|
||||
static SimpleCounter<int64_t>* countIncomingConnectionConnected =
|
||||
SimpleCounter<int64_t>::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<int64_t>* countIncomingConnectionFailed =
|
||||
SimpleCounter<int64_t>::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<Void> listen(TransportData* self, NetworkAddress listenAddr)
|
|||
state uint64_t connectionCount = 0;
|
||||
try {
|
||||
loop {
|
||||
self->countIncomingConnRequested++;
|
||||
Reference<IConnection> conn = wait(listener->accept());
|
||||
self->countIncomingConnAccepted++;
|
||||
static SimpleCounter<int64_t>* countIncomingConnectionCreated =
|
||||
SimpleCounter<int64_t>::makeCounter("/Transport/TLS/IncomingConnectionCreated");
|
||||
countIncomingConnectionCreated->increment(1);
|
||||
if (conn) {
|
||||
TraceEvent("ConnectionFrom", conn->getDebugID())
|
||||
.suppressFor(1.0)
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
|
||||
|
|
|
|||
|
|
@ -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<Void> histogramReport() {
|
|||
}
|
||||
}
|
||||
|
||||
ACTOR Future<Void> 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,
|
||||
|
|
|
|||
|
|
@ -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();
|
||||
|
|
|
|||
|
|
@ -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<int>::max());
|
||||
static SimpleCounter<int64_t>* created = SimpleCounter<int64_t>::makeCounter("/flow/arena/arenasCreated");
|
||||
created->increment(1);
|
||||
if (reservedSize) {
|
||||
allowAccess(impl.getPtr());
|
||||
ArenaBlock::create((int)reservedSize, impl);
|
||||
static SimpleCounter<int64_t>* bytes = SimpleCounter<int64_t>::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<int64_t>* calls = SimpleCounter<int64_t>::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<ArenaBlock*>& visited) const {
|
|||
totalSizeEstimate = size();
|
||||
int o = nextBlockOffset;
|
||||
while (o) {
|
||||
static SimpleCounter<int64_t>* count =
|
||||
SimpleCounter<int64_t>::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<int64_t>* bytesWiped = SimpleCounter<int64_t>::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<ArenaBlock>& self, uint32_t
|
|||
}
|
||||
|
||||
void* ArenaBlock::allocate(Reference<ArenaBlock>& self, int bytes, IsSecureMem isSecure) {
|
||||
static SimpleCounter<int64_t>* arenaBlockAllocations =
|
||||
SimpleCounter<int64_t>::makeCounter("/flow/arena/arenaBlockAllocations");
|
||||
static SimpleCounter<int64_t>* arenaBlockBytesAllocated =
|
||||
SimpleCounter<int64_t>::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<ArenaBlock>& self, int bytes, IsSecureMem i
|
|||
// Return an appropriately-sized ArenaBlock to store the given data
|
||||
ArenaBlock* ArenaBlock::create(int dataSize, Reference<ArenaBlock>& next) {
|
||||
ArenaBlock* b;
|
||||
static SimpleCounter<int64_t>* created = SimpleCounter<int64_t>::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<ArenaBlock>& 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<int64_t>* destroyed =
|
||||
SimpleCounter<int64_t>::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");
|
||||
|
|
|
|||
|
|
@ -119,6 +119,9 @@ std::map<std::string, std::pair<int, int64_t>> hugeArenaTraces;
|
|||
|
||||
void hugeArenaSample(int size) {
|
||||
if (TraceEvent::isNetworkThread()) {
|
||||
static SimpleCounter<int64_t>* calls = SimpleCounter<int64_t>::makeCounter("/flow/fastalloc/hugeArenaSample");
|
||||
calls->increment(1);
|
||||
|
||||
auto& info = hugeArenaTraces[platform::get_backtrace()];
|
||||
info.first++;
|
||||
info.second += size;
|
||||
|
|
@ -373,6 +376,15 @@ void* FastAllocator<Size>::allocate() {
|
|||
if (keepalive_allocator::isActive()) [[unlikely]]
|
||||
return keepalive_allocator::allocate(Size);
|
||||
|
||||
// Accounting should mirror release() below.
|
||||
static int size = Size;
|
||||
static SimpleCounter<int64_t>* calls =
|
||||
SimpleCounter<int64_t>::makeCounter(format("/flow/fastalloc/allocateCallsSize%d", size));
|
||||
static SimpleCounter<int64_t>* bytes =
|
||||
SimpleCounter<int64_t>::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<Size>::allocate() {
|
|||
return p;
|
||||
}
|
||||
|
||||
void* countedNew(size_t nbytes) {
|
||||
static SimpleCounter<int64_t>* calls = SimpleCounter<int64_t>::makeCounter("/flow/fastalloc/newCalls");
|
||||
static SimpleCounter<int64_t>* bytes = SimpleCounter<int64_t>::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<int64_t>* calls = SimpleCounter<int64_t>::makeCounter("/flow/fastalloc/deleteCalls");
|
||||
static SimpleCounter<int64_t>* bytes = SimpleCounter<int64_t>::makeCounter("/flow/fastalloc/deleteBytes");
|
||||
calls->increment(1);
|
||||
bytes->increment(nbytes);
|
||||
|
||||
delete[] reinterpret_cast<uint8_t*>(ptr);
|
||||
}
|
||||
|
||||
template <int Size>
|
||||
void FastAllocator<Size>::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<int64_t>* calls =
|
||||
SimpleCounter<int64_t>::makeCounter(format("/flow/fastalloc/releaseCallsSize%d", size));
|
||||
static SimpleCounter<int64_t>* bytes =
|
||||
SimpleCounter<int64_t>::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<Size>::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
|
||||
|
||||
|
|
|
|||
|
|
@ -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<SlowTask> 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<int64_t>* countClientTLSHandshakeThrottled =
|
||||
SimpleCounter<int64_t>::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<int64_t>* countServerTLSHandshakesOnSideThreads =
|
||||
SimpleCounter<int64_t>::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<int64_t>* countServerTLSHandshakesOnMainThread =
|
||||
SimpleCounter<int64_t>::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<int64_t>* countServerTLSHandshakeThrottled =
|
||||
SimpleCounter<int64_t>::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<int64_t>* countServerTLSHandshakeLocked =
|
||||
SimpleCounter<int64_t>::makeCounter("/Net2/TLS/ServerTLSHandshakeLocked");
|
||||
countServerTLSHandshakeLocked->increment(1);
|
||||
|
||||
Promise<Void> connected;
|
||||
doAcceptHandshake(self, connected);
|
||||
|
|
@ -1025,7 +1025,9 @@ public:
|
|||
return Void();
|
||||
}
|
||||
when(wait(delay(FLOW_KNOBS->CONNECTION_MONITOR_TIMEOUT))) {
|
||||
g_net2->countServerTLSHandshakesTimedout++;
|
||||
static SimpleCounter<int64_t>* countServerTLSHandshakesTimedout =
|
||||
SimpleCounter<int64_t>::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<int64_t>* countClientTLSHandshakesOnSideThreads =
|
||||
SimpleCounter<int64_t>::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<int64_t>* countClientTLSHandshakesOnMainThread =
|
||||
SimpleCounter<int64_t>::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<int64_t>* countClientTLSHandshakeLocked =
|
||||
SimpleCounter<int64_t>::makeCounter("/Net2/TLS/ClientTLSHandshakeLocked");
|
||||
countClientTLSHandshakeLocked->increment(1);
|
||||
|
||||
Promise<Void> connected;
|
||||
doConnectHandshake(self, connected);
|
||||
|
|
@ -1106,7 +1114,9 @@ public:
|
|||
return Void();
|
||||
}
|
||||
when(wait(delay(FLOW_KNOBS->CONNECTION_MONITOR_TIMEOUT))) {
|
||||
g_net2->countClientTLSHandshakesTimedout++;
|
||||
static SimpleCounter<int64_t>* countClientTLSHandshakesTimedout =
|
||||
SimpleCounter<int64_t>::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();
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -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<int64_t>* bytes = SimpleCounter<int64_t>::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");
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -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 <thread>
|
||||
|
||||
#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<unsigned char>(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<unsigned char>(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<int64_t>* reportCount = SimpleCounter<int64_t>::makeCounter("/flow/counters/reports");
|
||||
reportCount->increment(1);
|
||||
|
||||
std::vector<SimpleCounter<int64_t>*> intCounters = SimpleCounter<int64_t>::getCounters();
|
||||
std::vector<SimpleCounter<double>*> doubleCounters = SimpleCounter<double>::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<int64_t>* 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<double>* 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<int64_t>* foo = SimpleCounter<int64_t>::makeCounter("/flow/counters/foo");
|
||||
SimpleCounter<int64_t>* bar = SimpleCounter<int64_t>::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<int64_t>* p =
|
||||
SimpleCounter<int64_t>::makeCounter(std::string("/flow/counters/many") + std::to_string(i));
|
||||
p->increment(i);
|
||||
ASSERT(p->get() == i);
|
||||
}
|
||||
|
||||
SimpleCounter<int64_t>* conflict = SimpleCounter<int64_t>::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<std::thread> 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<SimpleCounter<int64_t>*> intCounters = SimpleCounter<int64_t>::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<double>* baz = SimpleCounter<double>::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<std::thread> 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<SimpleCounter<double>*> doubleCounters = SimpleCounter<double>::getCounters();
|
||||
ASSERT(doubleCounters.size() >= 1);
|
||||
|
||||
// Give asserts here a chance to run.
|
||||
simpleCounterReport();
|
||||
|
||||
return Void();
|
||||
}
|
||||
|
||||
void forceLinkSimpleCounterTests() {}
|
||||
|
|
@ -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);
|
||||
|
||||
|
|
|
|||
|
|
@ -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');
|
||||
|
|
|
|||
|
|
@ -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 <algorithm>
|
||||
#include <array>
|
||||
|
|
@ -290,8 +291,6 @@ struct union_like_traits<Optional<T>> : std::true_type {
|
|||
}
|
||||
};
|
||||
|
||||
// #define STANDALONE_ALWAYS_COPY
|
||||
|
||||
template <class T>
|
||||
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<T>'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>& t) : Standalone((T const&)t) {}
|
||||
Standalone(const Standalone<T>&& t) : Standalone((T const&)t) {}
|
||||
Standalone<T>& operator=(const Standalone<T>&& t) {
|
||||
*this = (T const&)t;
|
||||
return *this;
|
||||
}
|
||||
Standalone<T>& operator=(const Standalone<T>& t) {
|
||||
*this = (T const&)t;
|
||||
return *this;
|
||||
}
|
||||
#else
|
||||
Standalone(const T& t, const Arena& arena) : Arena(arena), T(t) {}
|
||||
Standalone(const Standalone<T>&) = default;
|
||||
Standalone<T>& operator=(const Standalone<T>&) = default;
|
||||
Standalone(Standalone<T>&&) = default;
|
||||
Standalone<T>& operator=(Standalone<T>&&) = default;
|
||||
~Standalone() = default;
|
||||
#endif
|
||||
|
||||
template <class U>
|
||||
Standalone<U> castTo() const {
|
||||
|
|
@ -355,11 +336,6 @@ public:
|
|||
serializer(ar, (*(T*)this), arena());
|
||||
}
|
||||
|
||||
/*static Standalone<T> fakeStandalone( const T& t ) {
|
||||
Standalone<T> x;
|
||||
*(T*)&x = t;
|
||||
return x;
|
||||
}*/
|
||||
private:
|
||||
template <class U>
|
||||
Standalone(Standalone<U> const&); // unimplemented
|
||||
|
|
@ -371,22 +347,33 @@ extern std::string format(const char* form, ...);
|
|||
|
||||
#pragma pack(push, 4)
|
||||
class StringRef {
|
||||
private:
|
||||
static SimpleCounter<int64_t>* bytesCopied() {
|
||||
static SimpleCounter<int64_t>* bytesCopied =
|
||||
SimpleCounter<int64_t>::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<int>::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<const char*>(data), length); }
|
||||
std::string toString() const {
|
||||
bytesCopied()->increment(length);
|
||||
return std::string(reinterpret_cast<const char*>(data), length);
|
||||
}
|
||||
|
||||
std::string_view toStringView() const { return std::string_view(reinterpret_cast<const char*>(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<T, VecSerStrategy::String> {
|
|||
void reset() { _cached_size = 0; }
|
||||
};
|
||||
|
||||
// FIXME: consider whether methods in this class should be instrumented to
|
||||
// count calls, bytes processed, etc.
|
||||
template <class T, VecSerStrategy SerStrategy = VecSerStrategy::FlatBuffers>
|
||||
class VectorRef : public ComposedIdentifier<T, 3>, public VectorRefPreserializer<T, SerStrategy> {
|
||||
using VPS = VectorRefPreserializer<T, SerStrategy>;
|
||||
|
|
@ -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 T, int InlineMembers = 1>
|
||||
class SmallVectorRef {
|
||||
static_assert(InlineMembers >= 0);
|
||||
|
|
|
|||
|
|
@ -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<std::pair<const uint8_t*, int>> const& getWipedAreaSet();
|
|||
} // namespace keepalive_allocator
|
||||
|
||||
force_inline uint8_t* allocateAndMaybeKeepalive(size_t size) {
|
||||
static SimpleCounter<int64_t>* calls =
|
||||
SimpleCounter<int64_t>::makeCounter("/flow/fastalloc/AllocateAndMaybeKeepaliveCalls");
|
||||
static SimpleCounter<int64_t>* bytes =
|
||||
SimpleCounter<int64_t>::makeCounter("/flow/fastalloc/AllocateAndMaybeKeepaliveBytes");
|
||||
calls->increment(1);
|
||||
bytes->increment(size);
|
||||
uint8_t* p;
|
||||
if (keepalive_allocator::isActive()) [[unlikely]]
|
||||
p = static_cast<uint8_t*>(keepalive_allocator::allocate(size));
|
||||
|
|
@ -228,6 +243,9 @@ force_inline uint8_t* allocateAndMaybeKeepalive(size_t size) {
|
|||
}
|
||||
|
||||
force_inline void freeOrMaybeKeepalive(void* ptr) {
|
||||
static SimpleCounter<int64_t>* calls =
|
||||
SimpleCounter<int64_t>::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<T>::allocate,
|
||||
// FastAllocated<T>::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 Object>
|
||||
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<sizeof(Object) <= 64 ? 64 : nextFastAllocatedSize(sizeof(Object))>::release(s);
|
||||
} else {
|
||||
delete[] reinterpret_cast<uint8_t*>(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<int64_t>* bytes =
|
||||
SimpleCounter<int64_t>::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<int64_t>* bytes =
|
||||
SimpleCounter<int64_t>::makeCounter("/flow/fastalloc/Fast4AlignedBytesFreed");
|
||||
bytes->increment(size);
|
||||
|
||||
#if !defined(USE_JEMALLOC)
|
||||
// Sizes supported by FastAllocator must be release via FastAllocator
|
||||
if (size <= 4096)
|
||||
|
|
|
|||
|
|
@ -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 <atomic>
|
||||
#include <mutex>
|
||||
#include <vector>
|
||||
|
||||
#include "flow/Error.h"
|
||||
#include "flow/Trace.h"
|
||||
|
||||
// SimpleCounter metrics class for atomic counters of int64_t or
|
||||
// double. Example usage:
|
||||
//
|
||||
// static SimpleCounter<int64_t> *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<T>* 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 T>
|
||||
class SimpleCounter {
|
||||
static_assert(std::is_same_v<T, int64_t> || std::is_same_v<T, double>, "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<T> 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<SimpleCounter<T>*>& counters() {
|
||||
static std::vector<SimpleCounter<T>*> 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<T>* makeCounter(std::string_view name) {
|
||||
SimpleCounter<T>* rv = new SimpleCounter<T>(name);
|
||||
|
||||
std::lock_guard<std::mutex> lock(mutex());
|
||||
std::vector<SimpleCounter<T>*>& v = counters();
|
||||
v.push_back(rv);
|
||||
|
||||
return rv;
|
||||
}
|
||||
|
||||
static inline std::vector<SimpleCounter<T>*> getCounters() {
|
||||
std::vector<SimpleCounter<T>*> rv;
|
||||
std::lock_guard<std::mutex> lock(mutex());
|
||||
rv = counters();
|
||||
return rv;
|
||||
}
|
||||
};
|
||||
|
||||
template <>
|
||||
inline void SimpleCounter<int64_t>::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<double>::increment(double delta) {
|
||||
double old = value.load();
|
||||
while (!value.compare_exchange_weak(old, old + delta)) {
|
||||
;
|
||||
}
|
||||
}
|
||||
|
||||
void simpleCounterReport(Severity severity = SevInfo);
|
||||
|
||||
#endif
|
||||
|
|
@ -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);
|
||||
}
|
||||
};
|
||||
|
||||
|
|
|
|||
|
|
@ -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") {
|
||||
|
|
|
|||
Loading…
Reference in New Issue