Skip to content

always increment finishedQueries counter if incrementing allQueries #13786

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
jkt-signal wants to merge 2 commits into
base: release-7.3
Choose a base branch
from
+97 −89
Open

always increment finishedQueries counter if incrementing allQueries #13786

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
988b399
always increment finishedQueries counter if incrementing allQueries
jkt-signal Jul 24, 2026
aead434
move counters before actors in StorageServer class body to correct de…
jkt-signal Jul 30, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 13 additions & 0 deletions fdbrpc/include/fdbrpc/Stats.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -158,6 +158,19 @@ struct Traceable<Counter> : std::true_type {
}
};

// Increments one counter at construction, and another at destruction;
// intended to be used to mark the beginning and end of some scope, without
// the possibility of forgetting to increment the exit counter on exceptions.
class CountedSection final : public NonCopyable {
public:
CountedSection() : end(nullptr) {}
CountedSection(Counter &start, Counter &end) : end(&end) { ++start; }
CountedSection &operator=(CountedSection &&other) { std::swap(end, other.end); return *this; }
~CountedSection() { if (end) ++*end; }
private:
Counter *end;
};

template <class F>
struct SpecialCounter final : ICounter, FastAllocated<SpecialCounter<F>>, NonCopyable {
SpecialCounter(CounterCollection& collection, std::string const& name, F&& f) : name(name), f(f) {
Expand Down
173 changes: 84 additions & 89 deletions fdbserver/storageserver.actor.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1310,83 +1310,6 @@ struct StorageServer : public IStorageMetricsService {
Key sk;
Reference<AsyncVar<ServerDBInfo> const> db;
Database cx;
ActorCollection actors;

CoalescedKeyRangeMap<bool, int64_t, KeyBytesMetric<int64_t>> byteSampleClears;
AsyncVar<bool> byteSampleClearsTooLarge;
Future<Void> byteSampleRecovery;
Future<Void> durableInProgress;

AsyncMap<Key, bool> watches;
AsyncMap<int64_t, bool> tenantWatches;
int64_t watchBytes;
int64_t numWatches;
AsyncVar<bool> noRecentUpdates;
double lastUpdate;

std::string folder;
std::string checkpointFolder;
std::string fetchedCheckpointFolder;

// defined only during splitMutations()/addMutation()
UpdateEagerReadInfo* updateEagerReads;

FlowLock durableVersionLock;
FlowLock fetchKeysParallelismLock;
// Extra lock that prevents too much post-initial-fetch work from building up, such as mutation applying and change
// feed tail fetching
FlowLock fetchKeysParallelismChangeFeedLock;
int64_t fetchKeysBytesBudget;
AsyncVar<bool> fetchKeysBudgetUsed;
int64_t fetchKeysTotalCommitBytes;
std::vector<Promise<FetchInjectionInfo*>> readyFetchKeys;

FlowLock serveFetchCheckpointParallelismLock;

std::unordered_map<UID, std::shared_ptr<MoveInShard>> moveInShards;

Reference<PriorityMultiLock> ssLock;
std::vector<int> readPriorityRanks;

Future<PriorityMultiLock::Lock> getReadLock(const Optional<ReadOptions>& options) {
int readType = (int)(options.present() ? options.get().type : ReadType::NORMAL);
readType = std::clamp<int>(readType, 0, readPriorityRanks.size() - 1);
return ssLock->lock(readPriorityRanks[readType]);
}

FlowLock serveAuditStorageParallelismLock;

int64_t instanceID;

Promise<Void> otherError;
Promise<Void> coreStarted;
bool shuttingDown;

Promise<Void> registerInterfaceAcceptingRequests;
Future<Void> interfaceRegistered;

bool behind;
bool versionBehind;

bool debug_inApplyUpdate;
double debug_lastValidateTime;

int64_t lastBytesInputEBrake;
Version lastDurableVersionEBrake;

int maxQueryQueue;
int getAndResetMaxQueryQueueSize() {
int val = maxQueryQueue;
maxQueryQueue = 0;
return val;
}

TransactionTagCounter transactionTagCounter;
BusiestWriteTagContext busiestWriteTagContext;

Optional<LatencyBandConfig> latencyBandConfig;

Optional<EncryptionAtRestMode> encryptionMode;

struct Counters {
CounterCollection cc;
Expand Down Expand Up @@ -1616,6 +1539,84 @@ struct StorageServer : public IStorageMetricsService {
}
} counters;

ActorCollection actors;

CoalescedKeyRangeMap<bool, int64_t, KeyBytesMetric<int64_t>> byteSampleClears;
AsyncVar<bool> byteSampleClearsTooLarge;
Future<Void> byteSampleRecovery;
Future<Void> durableInProgress;

AsyncMap<Key, bool> watches;
AsyncMap<int64_t, bool> tenantWatches;
int64_t watchBytes;
int64_t numWatches;
AsyncVar<bool> noRecentUpdates;
double lastUpdate;

std::string folder;
std::string checkpointFolder;
std::string fetchedCheckpointFolder;

// defined only during splitMutations()/addMutation()
UpdateEagerReadInfo* updateEagerReads;

FlowLock durableVersionLock;
FlowLock fetchKeysParallelismLock;
// Extra lock that prevents too much post-initial-fetch work from building up, such as mutation applying and change
// feed tail fetching
FlowLock fetchKeysParallelismChangeFeedLock;
int64_t fetchKeysBytesBudget;
AsyncVar<bool> fetchKeysBudgetUsed;
int64_t fetchKeysTotalCommitBytes;
std::vector<Promise<FetchInjectionInfo*>> readyFetchKeys;

FlowLock serveFetchCheckpointParallelismLock;

std::unordered_map<UID, std::shared_ptr<MoveInShard>> moveInShards;

Reference<PriorityMultiLock> ssLock;
std::vector<int> readPriorityRanks;

Future<PriorityMultiLock::Lock> getReadLock(const Optional<ReadOptions>& options) {
int readType = (int)(options.present() ? options.get().type : ReadType::NORMAL);
readType = std::clamp<int>(readType, 0, readPriorityRanks.size() - 1);
return ssLock->lock(readPriorityRanks[readType]);
}

FlowLock serveAuditStorageParallelismLock;

int64_t instanceID;

Promise<Void> otherError;
Promise<Void> coreStarted;
bool shuttingDown;

Promise<Void> registerInterfaceAcceptingRequests;
Future<Void> interfaceRegistered;

bool behind;
bool versionBehind;

bool debug_inApplyUpdate;
double debug_lastValidateTime;

int64_t lastBytesInputEBrake;
Version lastDurableVersionEBrake;

int maxQueryQueue;
int getAndResetMaxQueryQueueSize() {
int val = maxQueryQueue;
maxQueryQueue = 0;
return val;
}

TransactionTagCounter transactionTagCounter;
BusiestWriteTagContext busiestWriteTagContext;

Optional<LatencyBandConfig> latencyBandConfig;

Optional<EncryptionAtRestMode> encryptionMode;

// Bytes read from storage engine when a storage server starts.
int64_t bytesRestored = 0;

Expand Down Expand Up @@ -2416,12 +2417,12 @@ std::shared_ptr<MoveInShard> StorageServer::getMoveInShard(const UID& dataMoveId
ACTOR Future<Void> getValueQ(StorageServer* data, GetValueRequest req) {
state int64_t resultSize = 0;
Span span("SS:getValue"_loc, req.spanContext);
state CountedSection cs(data->counters.allQueries, data->counters.finishedQueries);
// Temporarily disabled -- this path is hit a lot
// getCurrentLineage()->modify(&TransactionLineage::txID) = req.spanContext.first();

try {
++data->counters.getValueQueries;
++data->counters.allQueries;
if (req.key.startsWith(systemKeys.begin)) {
++data->counters.systemKeyQueries;
}
Expand Down Expand Up @@ -2540,8 +2541,6 @@ ACTOR Future<Void> getValueQ(StorageServer* data, GetValueRequest req) {
// so it must be accounted for here.
data->transactionTagCounter.addRequest(req.tags, req.key.size() + resultSize);

++data->counters.finishedQueries;

double duration = g_network->timer() - req.requestTime();
data->counters.readLatencySample.addMeasurement(duration);
data->counters.readValueLatencySample.addMeasurement(duration);
Expand Down Expand Up @@ -4535,8 +4534,8 @@ ACTOR Future<Void> getKeyValuesQ(StorageServer* data, GetKeyValuesRequest req)

getCurrentLineage()->modify(&TransactionLineage::txID) = req.spanContext.traceID;

state CountedSection cs(data->counters.allQueries, data->counters.finishedQueries);
++data->counters.getRangeQueries;
++data->counters.allQueries;
if (req.begin.getKey().startsWith(systemKeys.begin)) {
++data->counters.systemKeyQueries;
++data->counters.getRangeSystemKeyQueries;
Expand Down Expand Up @@ -4711,7 +4710,6 @@ ACTOR Future<Void> getKeyValuesQ(StorageServer* data, GetKeyValuesRequest req)
}

data->transactionTagCounter.addRequest(req.tags, resultSize);
++data->counters.finishedQueries;

double duration = g_network->timer() - req.requestTime();
data->counters.readLatencySample.addMeasurement(duration);
Expand Down Expand Up @@ -6224,8 +6222,8 @@ ACTOR Future<Void> getMappedKeyValuesQ(StorageServer* data, GetMappedKeyValuesRe

getCurrentLineage()->modify(&TransactionLineage::txID) = req.spanContext.traceID;

++data->counters.getMappedRangeQueries;
++data->counters.allQueries;
state CountedSection csAll(data->counters.allQueries, data->counters.finishedQueries);
state CountedSection csRangeMapped(data->counters.getMappedRangeQueries, data->counters.finishedGetMappedRangeQueries);
if (req.begin.getKey().startsWith(systemKeys.begin)) {
++data->counters.systemKeyQueries;
}
Expand Down Expand Up @@ -6404,8 +6402,6 @@ ACTOR Future<Void> getMappedKeyValuesQ(StorageServer* data, GetMappedKeyValuesRe
}

data->transactionTagCounter.addRequest(req.tags, resultSize);
++data->counters.finishedQueries;
++data->counters.finishedGetMappedRangeQueries;

double duration = g_network->timer() - req.requestTime();
data->counters.readLatencySample.addMeasurement(duration);
Expand Down Expand Up @@ -6433,8 +6429,8 @@ ACTOR Future<Void> getKeyValuesStreamQ(StorageServer* data, GetKeyValuesStreamRe
state int64_t resultSize = 0;

req.reply.setByteLimit(SERVER_KNOBS->RANGESTREAM_LIMIT_BYTES);
state CountedSection cs(data->counters.allQueries, data->counters.finishedQueries);
++data->counters.getRangeStreamQueries;
++data->counters.allQueries;
if (req.begin.getKey().startsWith(systemKeys.begin)) {
++data->counters.systemKeyQueries;
}
Expand Down Expand Up @@ -6632,7 +6628,6 @@ ACTOR Future<Void> getKeyValuesStreamQ(StorageServer* data, GetKeyValuesStreamRe
}

data->transactionTagCounter.addRequest(req.tags, resultSize);
++data->counters.finishedQueries;

return Void();
}
Expand All @@ -6643,8 +6638,8 @@ ACTOR Future<Void> getKeyQ(StorageServer* data, GetKeyRequest req) {

getCurrentLineage()->modify(&TransactionLineage::txID) = req.spanContext.traceID;

state CountedSection cs(data->counters.allQueries, data->counters.finishedQueries);
++data->counters.getKeyQueries;
++data->counters.allQueries;
data->maxQueryQueue = std::max<int>(
data->maxQueryQueue, data->counters.allQueries.getValue() - data->counters.finishedQueries.getValue());

Expand Down
Loading