wbtiger
/
foundationdb
mirror of https://github.com/apple/foundationdb.git
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

[Release-7.4] Cherry-pick Avoid Source Storage Server Being Overloaded by Data Movements with Replica Consistency Check (#12176) 

* Avoid Source Storage Server Being Overloaded by Data Movements with Replica Consistency Check (#12164)

* add ss metrics for fetch key

* bug fix

* revert checkTimeSpanSec

* fix adjustRelocationParallelismForSrc

* code cleanup

* fix replicaComparison

* remove unnecessary counters

* fix large storage server data structure

* address comments

* address comments

* address comments

* code cleanup

* bug fix

* fix bug

* remove taskID in get range requests
This commit is contained in:
Zhe Wang 2025-06-02 12:34:34 -07:00 committed by GitHub
parent 37e3964113
commit f60e95c50b
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
13 changed files with 191 additions and 107 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
fdbclient/BackupAgentBase.actor.cpp
View File

@ -567,7 +567,7 @@ ACTOR Future<Void> readCommitted(Database cx,
tr.setOption(FDBTransactionOptions::LOCK_AWARE);
if (CLIENT_KNOBS->ENABLE_REPLICA_CONSISTENCY_CHECK_ON_BACKUP_READS) {
tr.setOption(FDBTransactionOptions::ENABLE_REPLICA_CONSISTENCY_CHECK);
int64_t requiredReplicas = CLIENT_KNOBS->CONSISTENCY_CHECK_REQUIRED_REPLICAS;
int64_t requiredReplicas = CLIENT_KNOBS->BACKUP_CONSISTENCY_CHECK_REQUIRED_REPLICAS;
tr.setOption(FDBTransactionOptions::CONSISTENCY_CHECK_REQUIRED_REPLICAS,
StringRef((uint8_t*)&requiredReplicas, sizeof(int64_t)));
}

2
fdbclient/ClientKnobs.cpp
View File

@ -200,7 +200,7 @@ void ClientKnobs::initialize(Randomize randomize) {
init( BLOB_GRANULE_RESTORE_CHECK_INTERVAL, 10 );
init( BACKUP_CONTAINER_LOCAL_ALLOW_RELATIVE_PATH, false );
init( ENABLE_REPLICA_CONSISTENCY_CHECK_ON_BACKUP_READS, false ); if( randomize && BUGGIFY ) { ENABLE_REPLICA_CONSISTENCY_CHECK_ON_BACKUP_READS = true; }
init( CONSISTENCY_CHECK_REQUIRED_REPLICAS, -2 ); // Do consistency check based on all available storage replicas
init( BACKUP_CONSISTENCY_CHECK_REQUIRED_REPLICAS, -2 ); // Do consistency check based on all available storage replicas
init( BULKLOAD_JOB_HISTORY_COUNT_MAX, 10 ); if (randomize && BUGGIFY) BULKLOAD_JOB_HISTORY_COUNT_MAX = deterministicRandom()->randomInt(1, 10);
init( BULKLOAD_VERBOSE_LEVEL, 10 );
init( S3CLIENT_VERBOSE_LEVEL, 10 );

4
fdbclient/ServerKnobs.cpp
View File

@ -177,8 +177,8 @@ void ServerKnobs::initialize(Randomize randomize, ClientKnobs* clientKnobs, IsSi
init( FETCH_KEYS_THROTTLE_PRIORITY_THRESHOLD, 0 ); if( randomize && BUGGIFY ) FETCH_KEYS_THROTTLE_PRIORITY_THRESHOLD = 700;
init( ENABLE_REPLICA_CONSISTENCY_CHECK_ON_DATA_MOVEMENT, false ); ENABLE_REPLICA_CONSISTENCY_CHECK_ON_DATA_MOVEMENT = isSimulated;
init( CONSISTENCY_CHECK_REQUIRED_REPLICAS, 1 );
init( DATAMOVE_CONSISTENCY_CHECK_REQUIRED_REPLICAS, 1 );
init( ENABLE_CONSERVATIVE_RELOCATION_WHEN_REPLICA_CONSISTENCY_CHECK, false ); if (isSimulated) ENABLE_CONSERVATIVE_RELOCATION_WHEN_REPLICA_CONSISTENCY_CHECK = deterministicRandom()->coinflip();
init( PROBABILITY_TEAM_REDUNDANT_DATAMOVE_CHOOSE_TRUE_BEST_DEST, 0.0 ); if (isSimulated) PROBABILITY_TEAM_REDUNDANT_DATAMOVE_CHOOSE_TRUE_BEST_DEST = deterministicRandom()->random01();
init( PROBABILITY_TEAM_UNHEALTHY_DATAMOVE_CHOOSE_TRUE_BEST_DEST, 0.0 ); if (isSimulated) PROBABILITY_TEAM_UNHEALTHY_DATAMOVE_CHOOSE_TRUE_BEST_DEST = deterministicRandom()->random01();

2
fdbclient/include/fdbclient/ClientKnobs.h
View File

@ -200,7 +200,7 @@ public:
int BLOB_GRANULE_RESTORE_CHECK_INTERVAL;
bool BACKUP_CONTAINER_LOCAL_ALLOW_RELATIVE_PATH;
bool ENABLE_REPLICA_CONSISTENCY_CHECK_ON_BACKUP_READS;
int CONSISTENCY_CHECK_REQUIRED_REPLICAS;
int BACKUP_CONSISTENCY_CHECK_REQUIRED_REPLICAS;
int BULKLOAD_JOB_HISTORY_COUNT_MAX; // the max number of bulk load job history to keep. The oldest job history will
// be removed when the count exceeds this value. Set to 0 to disable history.
// Do not set the value to a large number, e.g. <= 10.

7
fdbclient/include/fdbclient/ServerKnobs.h
View File

@ -199,8 +199,11 @@ public:
// STORAGE_FETCH_KEYS_RATE_LIMIT.
int FETCH_KEYS_THROTTLE_PRIORITY_THRESHOLD;
bool ENABLE_REPLICA_CONSISTENCY_CHECK_ON_DATA_MOVEMENT;
int CONSISTENCY_CHECK_REQUIRED_REPLICAS;
bool ENABLE_REPLICA_CONSISTENCY_CHECK_ON_DATA_MOVEMENT; // Enable to check replica consistency on data movement
int DATAMOVE_CONSISTENCY_CHECK_REQUIRED_REPLICAS; // The number of extra replicas to check for replica consistency
// on data movement read range requests by fetchKeys
bool ENABLE_CONSERVATIVE_RELOCATION_WHEN_REPLICA_CONSISTENCY_CHECK; // Enable to slow down relocation when replica
// consistency check on data movement is enabled
// Probability that a team redundant data move set TrueBest when get destination team
double PROBABILITY_TEAM_REDUNDANT_DATAMOVE_CHOOSE_TRUE_BEST_DEST;

87
fdbrpc/include/fdbrpc/LoadBalance.actor.h
View File

@ -311,31 +311,55 @@ Future<Void> replicaComparison(Req req,
}
} else if (!srcLB.present() || !srcLB.get().error.present()) {
// Verify that the other SS servers in the team have the same data.
state std::vector<Future<Optional<ErrorOr<Resp>>>> restOfTeamFutures;
restOfTeamFutures.reserve(ssTeam->size() - 1);
std::vector<uint64_t> candidates;
// candidates includes all healthy SS endpoints in the team except the one we already
// have a response from
for (int i = 0; i < ssTeam->size(); i++) {
RequestStream<Req, P> const* si = &ssTeam->get(i, channel);
if (si->getEndpoint().token.first() !=
srcEndpointId) { // don't re-request to SS we already have a response from
if (!IFailureMonitor::failureMonitor().getState(si->getEndpoint()).failed) {
resetReply(req);
restOfTeamFutures.push_back((
requiredReplicas == BEST_EFFORT
? timeout(si->tryGetReply(req), FLOW_KNOBS->LOAD_BALANCE_FETCH_REPLICA_TIMEOUT)
: timeout(errorOr(si->getReply(req)), FLOW_KNOBS->LOAD_BALANCE_FETCH_REPLICA_TIMEOUT)));
} else if (requiredReplicas == ALL_REPLICAS) {
TraceEvent(SevWarnAlways, "UnreachableStorageServer")
.detail("SSID", ssTeam->getInterface(i).id());
throw unreachable_storage_replica();
}
if (si->getEndpoint().token.first() == srcEndpointId) {
// Don't re-request to SS we already have a response from
continue;
}
if (!IFailureMonitor::failureMonitor().getState(si->getEndpoint()).failed) {
candidates.push_back(si->getEndpoint().token.first());
} else if (requiredReplicas == ALL_REPLICAS) {
TraceEvent(SevWarnAlways, "UnreachableStorageServer").detail("SSID", ssTeam->getInterface(i).id());
throw unreachable_storage_replica();
}
}
if (requiredReplicas == BEST_EFFORT || requiredReplicas == ALL_REPLICAS) {
wait(waitForAllReady(restOfTeamFutures));
} else {
wait(waitForQuorumReplies(&restOfTeamFutures, requiredReplicas));
int numReplicaToRead = candidates.size();
if (requiredReplicas != BEST_EFFORT && requiredReplicas != ALL_REPLICAS) {
ASSERT(requiredReplicas > 0);
numReplicaToRead = std::min((int)candidates.size(), requiredReplicas);
if (FLOW_KNOBS->ENABLE_WARNING_READ_CONSISTENCY_CHECK_NOT_ENOUGH_REPLICA &&
candidates.size() < requiredReplicas) {
TraceEvent(SevWarn, "ReplicaConsistencyCheckNotEnoughReplica")
.suppressFor(5.0)
.detail("RequiredReplicas", requiredReplicas)
.detail("AvailableReplicas", candidates.size());
}
}
state std::vector<Future<Optional<ErrorOr<Resp>>>> restOfTeamFutures;
restOfTeamFutures.reserve(numReplicaToRead);
// Randomly select numReplicaToRead SSes to read from
deterministicRandom()->randomShuffle(candidates);
candidates.erase(candidates.begin() + numReplicaToRead, candidates.end());
std::unordered_set<uint64_t> ssToRead(candidates.begin(), candidates.end());
for (int i = 0; i < ssTeam->size(); i++) {
RequestStream<Req, P> const* si = &ssTeam->get(i, channel);
if (!ssToRead.contains(si->getEndpoint().token.first())) {
// Only send requests to the SSes that we randomly selected
continue;
}
resetReply(req);
restOfTeamFutures.push_back(
(requiredReplicas == BEST_EFFORT
? timeout(si->tryGetReply(req), FLOW_KNOBS->LOAD_BALANCE_FETCH_REPLICA_TIMEOUT)
: timeout(errorOr(si->getReply(req)), FLOW_KNOBS->LOAD_BALANCE_FETCH_REPLICA_TIMEOUT)));
}
wait(waitForAllReady(restOfTeamFutures));
int numError = 0;
int numMismatch = 0;
@ -480,16 +504,31 @@ struct RequestData : NonCopyable {
int requiredReplicas) {
if (model && (compareReplicas || FLOW_KNOBS->ENABLE_REPLICA_CONSISTENCY_CHECK_ON_READS)) {
ASSERT(requestStream != nullptr);
int requiredReplicaCount =
compareReplicas ? requiredReplicas : FLOW_KNOBS->CONSISTENCY_CHECK_REQUIRED_REPLICAS;
if (compareReplicas) {
// In case compareReplicas == true, we may read extra requiredReplicas replica.
// The value of compareReplicas is decided by the caller and the knobs.
// If the caller is fetchKeys, when ENABLE_REPLICA_CONSISTENCY_CHECK_ON_DATA_MOVEMENT is on,
// the value is DATAMOVE_CONSISTENCY_CHECK_REQUIRED_REPLICAS.
// If the caller is backup agents, when ENABLE_REPLICA_CONSISTENCY_CHECK_ON_BACKUP_READS is on,
// the value is BACKUP_CONSISTENCY_CHECK_REQUIRED_REPLICAS.
// Otherwise, the value is 0.
return replicaComparison(request,
response,
requestStream->getEndpoint().token.first(),
alternatives,
channel,
requiredReplicas);
}
// In case ENABLE_REPLICA_CONSISTENCY_CHECK_ON_READS is on, we read extra
// READ_CONSISTENCY_CHECK_REQUIRED_REPLICAS replica and conduct consistency
// check among replica for any read request.
return replicaComparison(request,
response,
requestStream->getEndpoint().token.first(),
alternatives,
channel,
requiredReplicaCount);
FLOW_KNOBS->READ_CONSISTENCY_CHECK_REQUIRED_REPLICAS);
}
return Void();
}

42
fdbserver/DDRelocationQueue.actor.cpp
View File

@ -428,25 +428,55 @@ std::string Busyness::toString() {
return result;
}
double adjustRelocationParallelismForSrc(double srcParallelism) {
double res = srcParallelism;
if (SERVER_KNOBS->ENABLE_CONSERVATIVE_RELOCATION_WHEN_REPLICA_CONSISTENCY_CHECK &&
SERVER_KNOBS->ENABLE_REPLICA_CONSISTENCY_CHECK_ON_DATA_MOVEMENT &&
srcParallelism >= 1.0 + SERVER_KNOBS->DATAMOVE_CONSISTENCY_CHECK_REQUIRED_REPLICAS) {
// DATAMOVE_CONSISTENCY_CHECK_REQUIRED_REPLICAS is the number of extra replicas that the destination
// servers will read from the source team.
res = res / (1.0 + SERVER_KNOBS->DATAMOVE_CONSISTENCY_CHECK_REQUIRED_REPLICAS);
}
ASSERT(res > 0);
return res;
}
// find the "workFactor" for this, were it launched now
int getSrcWorkFactor(RelocateData const& relocation, int singleRegionTeamSize) {
// RELOCATION_PARALLELISM_PER_SOURCE_SERVER is the number of concurrent replications that can be launched on a
// single storage server at a time, given the team size is 1 --- only this storage server is available to serve
// fetchKey read requests from the dest team.
// The real parallelism is adjusted by the number of source servers of a source team that can serve
// fetchKey requests.
// When ENABLE_REPLICA_CONSISTENCY_CHECK_ON_DATA_MOVEMENT is enabled, the fetchKeys on
// destination servers will read from DATAMOVE_CONSISTENCY_CHECK_REQUIRED_REPLICAS + 1 replicas from the source team
// (suppose the team size is large enough). As a result it is possible that the source team can be overloaded by the
// fetchKey read requests. This is especially true when the shard split data movements are launched. So, we
// introduce ENABLE_CONSERVATIVE_RELOCATION_WHEN_REPLICA_CONSISTENCY_CHECK knob to adjust the relocation parallelism
// accordingly. The adjustment is to reduce the relocation parallelism by a factor of
// (1 + DATAMOVE_CONSISTENCY_CHECK_REQUIRED_REPLICAS).
if (relocation.bulkLoadTask.present())
return 0;
else if (relocation.healthPriority == SERVER_KNOBS->PRIORITY_TEAM_1_LEFT ||
relocation.healthPriority == SERVER_KNOBS->PRIORITY_TEAM_0_LEFT)
return WORK_FULL_UTILIZATION / SERVER_KNOBS->RELOCATION_PARALLELISM_PER_SOURCE_SERVER;
return WORK_FULL_UTILIZATION /
adjustRelocationParallelismForSrc(SERVER_KNOBS->RELOCATION_PARALLELISM_PER_SOURCE_SERVER);
else if (relocation.healthPriority == SERVER_KNOBS->PRIORITY_TEAM_2_LEFT)
return WORK_FULL_UTILIZATION / 2 / SERVER_KNOBS->RELOCATION_PARALLELISM_PER_SOURCE_SERVER;
return WORK_FULL_UTILIZATION /
adjustRelocationParallelismForSrc(2 * SERVER_KNOBS->RELOCATION_PARALLELISM_PER_SOURCE_SERVER);
else if (relocation.healthPriority == SERVER_KNOBS->PRIORITY_PERPETUAL_STORAGE_WIGGLE)
// we want to set PRIORITY_PERPETUAL_STORAGE_WIGGLE to a reasonably large value
// to make this parallelism take effect
return WORK_FULL_UTILIZATION / SERVER_KNOBS->WIGGLING_RELOCATION_PARALLELISM_PER_SOURCE_SERVER;
return WORK_FULL_UTILIZATION /
adjustRelocationParallelismForSrc(SERVER_KNOBS->WIGGLING_RELOCATION_PARALLELISM_PER_SOURCE_SERVER);
else if (relocation.priority == SERVER_KNOBS->PRIORITY_MERGE_SHARD)
return WORK_FULL_UTILIZATION / SERVER_KNOBS->MERGE_RELOCATION_PARALLELISM_PER_TEAM;
return WORK_FULL_UTILIZATION /
adjustRelocationParallelismForSrc(SERVER_KNOBS->MERGE_RELOCATION_PARALLELISM_PER_TEAM);
else { // for now we assume that any message at a lower priority can best be assumed to have a full team left for
// work
return WORK_FULL_UTILIZATION / singleRegionTeamSize / SERVER_KNOBS->RELOCATION_PARALLELISM_PER_SOURCE_SERVER;
return WORK_FULL_UTILIZATION /
adjustRelocationParallelismForSrc(singleRegionTeamSize *
SERVER_KNOBS->RELOCATION_PARALLELISM_PER_SOURCE_SERVER);
}
}

2
fdbserver/StorageMetrics.actor.cpp
View File

@ -28,7 +28,7 @@ CommonStorageCounters::CommonStorageCounters(const std::string& name,
: cc(name, id), finishedQueries("FinishedQueries", cc), bytesQueried("BytesQueried", cc),
bytesFetched("BytesFetched", cc), bytesInput("BytesInput", cc), mutationBytes("MutationBytes", cc),
kvFetched("KVFetched", cc), mutations("Mutations", cc), setMutations("SetMutations", cc),
clearRangeMutations("ClearRangeMutations", cc) {
clearRangeMutations("ClearRangeMutations", cc), fetchKeyErrors("FetchKeyErrors", cc) {
if (metrics) {
specialCounter(cc, "BytesStored", [metrics]() { return metrics->byteSample.getEstimate(allKeys); });
specialCounter(cc, "BytesReadSampleCount", [metrics]() { return metrics->bytesReadSample.queue.size(); });

3
fdbserver/include/fdbserver/StorageMetrics.actor.h
View File

@ -206,6 +206,9 @@ struct CommonStorageCounters {
// The number of key-value pairs fetched by fetchKeys()
Counter kvFetched;
// The number of fetchKeys errors
Counter fetchKeyErrors;
// name and id are the inputs to CounterCollection initialization. If metrics provided, the caller should guarantee
// the lifetime of metrics is longer than this counter
CommonStorageCounters(const std::string& name,

138
fdbserver/storageserver.actor.cpp
View File

@ -1546,15 +1546,15 @@ public:
// expensive.
Counter pTreeClearSplits;
LatencySample readLatencySample;
LatencySample readKeyLatencySample;
LatencySample readValueLatencySample;
LatencySample readRangeLatencySample;
LatencySample readVersionWaitSample;
LatencySample readQueueWaitSample;
LatencySample kvReadRangeLatencySample;
LatencySample updateLatencySample;
LatencySample updateEncryptionLatencySample;
std::unique_ptr<LatencySample> readLatencySample;
std::unique_ptr<LatencySample> readKeyLatencySample;
std::unique_ptr<LatencySample> readValueLatencySample;
std::unique_ptr<LatencySample> readRangeLatencySample;
std::unique_ptr<LatencySample> readVersionWaitSample;
std::unique_ptr<LatencySample> readQueueWaitSample;
std::unique_ptr<LatencySample> kvReadRangeLatencySample;
std::unique_ptr<LatencySample> updateLatencySample;
std::unique_ptr<LatencySample> updateEncryptionLatencySample;
LatencyBands readLatencyBands;
std::unique_ptr<LatencySample> mappedRangeSample; // Samples getMappedRange latency
std::unique_ptr<LatencySample> mappedRangeRemoteSample; // Samples getMappedRange remote subquery latency
@ -1594,42 +1594,43 @@ public:
changeServerKeysAssigned("ChangeServerKeysAssigned", cc),
changeServerKeysUnassigned("ChangeServerKeysUnassigned", cc),
kvClearRangesInFetchKeys("KvClearRangesInFetchKeys", cc),
readLatencySample("ReadLatencyMetrics",
self->thisServerID,
SERVER_KNOBS->LATENCY_METRICS_LOGGING_INTERVAL,
SERVER_KNOBS->LATENCY_SKETCH_ACCURACY),
readKeyLatencySample("GetKeyMetrics",
self->thisServerID,
SERVER_KNOBS->LATENCY_METRICS_LOGGING_INTERVAL,
SERVER_KNOBS->LATENCY_SKETCH_ACCURACY),
readValueLatencySample("GetValueMetrics",
self->thisServerID,
SERVER_KNOBS->LATENCY_METRICS_LOGGING_INTERVAL,
SERVER_KNOBS->LATENCY_SKETCH_ACCURACY),
readRangeLatencySample("GetRangeMetrics",
self->thisServerID,
SERVER_KNOBS->LATENCY_METRICS_LOGGING_INTERVAL,
SERVER_KNOBS->LATENCY_SKETCH_ACCURACY),
readVersionWaitSample("ReadVersionWaitMetrics",
self->thisServerID,
SERVER_KNOBS->LATENCY_METRICS_LOGGING_INTERVAL,
SERVER_KNOBS->LATENCY_SKETCH_ACCURACY),
readQueueWaitSample("ReadQueueWaitMetrics",
self->thisServerID,
SERVER_KNOBS->LATENCY_METRICS_LOGGING_INTERVAL,
SERVER_KNOBS->LATENCY_SKETCH_ACCURACY),
kvReadRangeLatencySample("KVGetRangeMetrics",
self->thisServerID,
SERVER_KNOBS->LATENCY_METRICS_LOGGING_INTERVAL,
SERVER_KNOBS->LATENCY_SKETCH_ACCURACY),
updateLatencySample("UpdateLatencyMetrics",
self->thisServerID,
SERVER_KNOBS->LATENCY_METRICS_LOGGING_INTERVAL,
SERVER_KNOBS->LATENCY_SKETCH_ACCURACY),
updateEncryptionLatencySample("UpdateEncryptionLatencyMetrics",
self->thisServerID,
SERVER_KNOBS->LATENCY_METRICS_LOGGING_INTERVAL,
SERVER_KNOBS->LATENCY_SKETCH_ACCURACY),
readLatencySample(std::make_unique<LatencySample>("ReadLatencyMetrics",
self->thisServerID,
SERVER_KNOBS->LATENCY_METRICS_LOGGING_INTERVAL,
SERVER_KNOBS->LATENCY_SKETCH_ACCURACY)),
readKeyLatencySample(std::make_unique<LatencySample>("GetKeyMetrics",
self->thisServerID,
SERVER_KNOBS->LATENCY_METRICS_LOGGING_INTERVAL,
SERVER_KNOBS->LATENCY_SKETCH_ACCURACY)),
readValueLatencySample(std::make_unique<LatencySample>("GetValueMetrics",
self->thisServerID,
SERVER_KNOBS->LATENCY_METRICS_LOGGING_INTERVAL,
SERVER_KNOBS->LATENCY_SKETCH_ACCURACY)),
readRangeLatencySample(std::make_unique<LatencySample>("GetRangeMetrics",
self->thisServerID,
SERVER_KNOBS->LATENCY_METRICS_LOGGING_INTERVAL,
SERVER_KNOBS->LATENCY_SKETCH_ACCURACY)),
readVersionWaitSample(std::make_unique<LatencySample>("ReadVersionWaitMetrics",
self->thisServerID,
SERVER_KNOBS->LATENCY_METRICS_LOGGING_INTERVAL,
SERVER_KNOBS->LATENCY_SKETCH_ACCURACY)),
readQueueWaitSample(std::make_unique<LatencySample>("ReadQueueWaitMetrics",
self->thisServerID,
SERVER_KNOBS->LATENCY_METRICS_LOGGING_INTERVAL,
SERVER_KNOBS->LATENCY_SKETCH_ACCURACY)),
kvReadRangeLatencySample(std::make_unique<LatencySample>("KVGetRangeMetrics",
self->thisServerID,
SERVER_KNOBS->LATENCY_METRICS_LOGGING_INTERVAL,
SERVER_KNOBS->LATENCY_SKETCH_ACCURACY)),
updateLatencySample(std::make_unique<LatencySample>("UpdateLatencyMetrics",
self->thisServerID,
SERVER_KNOBS->LATENCY_METRICS_LOGGING_INTERVAL,
SERVER_KNOBS->LATENCY_SKETCH_ACCURACY)),
updateEncryptionLatencySample(
std::make_unique<LatencySample>("UpdateEncryptionLatencyMetrics",
self->thisServerID,
SERVER_KNOBS->LATENCY_METRICS_LOGGING_INTERVAL,
SERVER_KNOBS->LATENCY_SKETCH_ACCURACY)),
readLatencyBands("ReadLatencyBands", self->thisServerID, SERVER_KNOBS->STORAGE_LOGGING_DELAY),
mappedRangeSample(std::make_unique<LatencySample>("GetMappedRangeMetrics",
self->thisServerID,
@ -1647,7 +1648,6 @@ public:
self->thisServerID,
SERVER_KNOBS->LATENCY_METRICS_LOGGING_INTERVAL,
SERVER_KNOBS->LATENCY_SKETCH_ACCURACY)) {
specialCounter(cc, "LastTLogVersion", [self]() { return self->lastTLogVersion; });
specialCounter(cc, "Version", [self]() { return self->version.get(); });
specialCounter(cc, "StorageVersion", [self]() { return self->storageVersion(); });
@ -2541,7 +2541,7 @@ ACTOR Future<Void> getValueQ(StorageServer* data, GetValueRequest req) {
// Track time from requestTime through now as read queueing wait time
state double queueWaitEnd = g_network->timer();
data->counters.readQueueWaitSample.addMeasurement(queueWaitEnd - req.requestTime());
data->counters.readQueueWaitSample->addMeasurement(queueWaitEnd - req.requestTime());
if (req.options.present() && req.options.get().debugID.present())
g_traceBatch.addEvent("GetValueDebug",
@ -2551,7 +2551,7 @@ ACTOR Future<Void> getValueQ(StorageServer* data, GetValueRequest req) {
state Optional<Value> v;
Version commitVersion = getLatestCommitVersion(req.ssLatestCommitVersions, data->tag);
state Version version = wait(waitForVersion(data, commitVersion, req.version, req.spanContext));
data->counters.readVersionWaitSample.addMeasurement(g_network->timer() - queueWaitEnd);
data->counters.readVersionWaitSample->addMeasurement(g_network->timer() - queueWaitEnd);
if (req.options.present() && req.options.get().debugID.present())
g_traceBatch.addEvent("GetValueDebug",
@ -2649,8 +2649,8 @@ ACTOR Future<Void> getValueQ(StorageServer* data, GetValueRequest req) {
++data->counters.finishedQueries;
double duration = g_network->timer() - req.requestTime();
data->counters.readLatencySample.addMeasurement(duration);
data->counters.readValueLatencySample.addMeasurement(duration);
data->counters.readLatencySample->addMeasurement(duration);
data->counters.readValueLatencySample->addMeasurement(duration);
if (data->latencyBandConfig.present()) {
int maxReadBytes =
data->latencyBandConfig.get().readConfig.maxReadBytes.orDefault(std::numeric_limits<int>::max());
@ -4747,7 +4747,7 @@ ACTOR Future<Void> getKeyValuesQ(StorageServer* data, GetKeyValuesRequest req)
// Track time from requestTime through now as read queueing wait time
state double queueWaitEnd = g_network->timer();
data->counters.readQueueWaitSample.addMeasurement(queueWaitEnd - req.requestTime());
data->counters.readQueueWaitSample->addMeasurement(queueWaitEnd - req.requestTime());
try {
if (req.options.present() && req.options.get().debugID.present())
@ -4764,7 +4764,7 @@ ACTOR Future<Void> getKeyValuesQ(StorageServer* data, GetKeyValuesRequest req)
.detail("DebugID",
req.options.present() && req.options.get().debugID.present() ? req.options.get().debugID.get()
: UID());
data->counters.readVersionWaitSample.addMeasurement(g_network->timer() - queueWaitEnd);
data->counters.readVersionWaitSample->addMeasurement(g_network->timer() - queueWaitEnd);
data->checkTenantEntry(version, req.tenantInfo, req.options.present() ? req.options.get().lockAware : false);
if (req.tenantInfo.hasTenant()) {
@ -4871,7 +4871,7 @@ ACTOR Future<Void> getKeyValuesQ(StorageServer* data, GetKeyValuesRequest req)
req.options,
req.tenantInfo.prefix));
const double duration = g_network->timer() - kvReadRange;
data->counters.kvReadRangeLatencySample.addMeasurement(duration);
data->counters.kvReadRangeLatencySample->addMeasurement(duration);
GetKeyValuesReply r = _r;
if (req.options.present() && req.options.get().debugID.present())
@ -4900,6 +4900,7 @@ ACTOR Future<Void> getKeyValuesQ(StorageServer* data, GetKeyValuesRequest req)
for (int i = 0; i < r.data.size(); i++) {
totalByteSize += r.data[i].expectedSize();
}
if (totalByteSize > 0 && SERVER_KNOBS->READ_SAMPLING_ENABLED) {
int64_t bytesReadPerKSecond = std::max(totalByteSize, SERVER_KNOBS->EMPTY_READ_PENALTY) / 2;
data->metrics.notifyBytesReadPerKSecond(addPrefix(r.data[0].key, req.tenantInfo.prefix, req.arena),
@ -4931,8 +4932,8 @@ ACTOR Future<Void> getKeyValuesQ(StorageServer* data, GetKeyValuesRequest req)
++data->counters.finishedQueries;
double duration = g_network->timer() - req.requestTime();
data->counters.readLatencySample.addMeasurement(duration);
data->counters.readRangeLatencySample.addMeasurement(duration);
data->counters.readLatencySample->addMeasurement(duration);
data->counters.readRangeLatencySample->addMeasurement(duration);
if (data->latencyBandConfig.present()) {
int maxReadBytes =
data->latencyBandConfig.get().readConfig.maxReadBytes.orDefault(std::numeric_limits<int>::max());
@ -6709,7 +6710,7 @@ ACTOR Future<Void> getMappedKeyValuesQ(StorageServer* data, GetMappedKeyValuesRe
// Track time from requestTime through now as read queueing wait time
state double queueWaitEnd = g_network->timer();
data->counters.readQueueWaitSample.addMeasurement(queueWaitEnd - req.requestTime());
data->counters.readQueueWaitSample->addMeasurement(queueWaitEnd - req.requestTime());
try {
if (req.options.present() && req.options.get().debugID.present())
@ -6718,7 +6719,7 @@ ACTOR Future<Void> getMappedKeyValuesQ(StorageServer* data, GetMappedKeyValuesRe
// VERSION_VECTOR change
Version commitVersion = getLatestCommitVersion(req.ssLatestCommitVersions, data->tag);
state Version version = wait(waitForVersion(data, commitVersion, req.version, span.context));
data->counters.readVersionWaitSample.addMeasurement(g_network->timer() - queueWaitEnd);
data->counters.readVersionWaitSample->addMeasurement(g_network->timer() - queueWaitEnd);
data->checkTenantEntry(version, req.tenantInfo, req.options.present() ? req.options.get().lockAware : false);
if (req.tenantInfo.hasTenant()) {
@ -6877,7 +6878,7 @@ ACTOR Future<Void> getMappedKeyValuesQ(StorageServer* data, GetMappedKeyValuesRe
++data->counters.finishedGetMappedRangeQueries;
double duration = g_network->timer() - req.requestTime();
data->counters.readLatencySample.addMeasurement(duration);
data->counters.readLatencySample->addMeasurement(duration);
data->counters.mappedRangeSample->addMeasurement(duration);
if (data->latencyBandConfig.present()) {
int maxReadBytes =
@ -7124,12 +7125,12 @@ ACTOR Future<Void> getKeyQ(StorageServer* data, GetKeyRequest req) {
// Track time from requestTime through now as read queueing wait time
state double queueWaitEnd = g_network->timer();
data->counters.readQueueWaitSample.addMeasurement(queueWaitEnd - req.requestTime());
data->counters.readQueueWaitSample->addMeasurement(queueWaitEnd - req.requestTime());
try {
Version commitVersion = getLatestCommitVersion(req.ssLatestCommitVersions, data->tag);
state Version version = wait(waitForVersion(data, commitVersion, req.version, req.spanContext));
data->counters.readVersionWaitSample.addMeasurement(g_network->timer() - queueWaitEnd);
data->counters.readVersionWaitSample->addMeasurement(g_network->timer() - queueWaitEnd);
data->checkTenantEntry(version, req.tenantInfo, req.options.map(&ReadOptions::lockAware).orDefault(false));
if (req.tenantInfo.hasTenant()) {
@ -7192,8 +7193,8 @@ ACTOR Future<Void> getKeyQ(StorageServer* data, GetKeyRequest req) {
++data->counters.finishedQueries;
double duration = g_network->timer() - req.requestTime();
data->counters.readLatencySample.addMeasurement(duration);
data->counters.readKeyLatencySample.addMeasurement(duration);
data->counters.readLatencySample->addMeasurement(duration);
data->counters.readKeyLatencySample->addMeasurement(duration);
if (data->latencyBandConfig.present()) {
int maxReadBytes =
@ -9131,7 +9132,7 @@ ACTOR Future<Void> fetchKeys(StorageServer* data, AddingShard* shard) {
tr.setOption(FDBTransactionOptions::LOCK_AWARE);
if (!isFullRestore && SERVER_KNOBS->ENABLE_REPLICA_CONSISTENCY_CHECK_ON_DATA_MOVEMENT) {
tr.setOption(FDBTransactionOptions::ENABLE_REPLICA_CONSISTENCY_CHECK);
int64_t requiredReplicas = SERVER_KNOBS->CONSISTENCY_CHECK_REQUIRED_REPLICAS;
int64_t requiredReplicas = SERVER_KNOBS->DATAMOVE_CONSISTENCY_CHECK_REQUIRED_REPLICAS;
tr.setOption(FDBTransactionOptions::CONSISTENCY_CHECK_REQUIRED_REPLICAS,
StringRef((uint8_t*)&requiredReplicas, sizeof(int64_t)));
}
@ -9320,7 +9321,9 @@ ACTOR Future<Void> fetchKeys(StorageServer* data, AddingShard* shard) {
data->thisServerID);
}
}
metricReporter.addFetchedBytes(expectedBlockSize, this_block.size());
if (!conductBulkLoad) {
metricReporter.addFetchedBytes(expectedBlockSize, this_block.size());
}
totalBytes += expectedBlockSize;
if (shard->reason != DataMovementReason::INVALID &&
@ -9375,6 +9378,9 @@ ACTOR Future<Void> fetchKeys(StorageServer* data, AddingShard* shard) {
if (!fetchKeyCanRetry(e)) {
throw e;
}
if (!conductBulkLoad) {
data->counters.fetchKeyErrors += 1;
}
lastError = e;
if (lastError.code() == error_code_storage_replica_comparison_error) {
// The inconsistency could be because of the inclusion of a rolled back
@ -12616,8 +12622,8 @@ ACTOR Future<Void> update(StorageServer* data, bool* pReceivedUpdate) {
data->behind = false;
}
const double duration = g_network->timer() - updateStart;
data->counters.updateEncryptionLatencySample.addMeasurement(decryptionTime);
data->counters.updateLatencySample.addMeasurement(duration);
data->counters.updateEncryptionLatencySample->addMeasurement(decryptionTime);
data->counters.updateLatencySample->addMeasurement(duration);
return Void(); // update will get called again ASAP
} catch (Error& err) {

3
flow/Knobs.cpp
View File

@ -309,7 +309,8 @@ void FlowKnobs::initialize(Randomize randomize, IsSimulated isSimulated) {
init( TSS_LARGE_TRACE_SIZE, 50000 );
init( LOAD_BALANCE_FETCH_REPLICA_TIMEOUT, 5.0 );
init( ENABLE_REPLICA_CONSISTENCY_CHECK_ON_READS, false ); if( randomize && BUGGIFY ) ENABLE_REPLICA_CONSISTENCY_CHECK_ON_READS = true;
init( CONSISTENCY_CHECK_REQUIRED_REPLICAS, -2 ); // Do consistency check based on all available storage replicas
init( READ_CONSISTENCY_CHECK_REQUIRED_REPLICAS, -2 ); // Do consistency check based on all available storage replicas
init( ENABLE_WARNING_READ_CONSISTENCY_CHECK_NOT_ENOUGH_REPLICA, false); if (randomize && BUGGIFY) { ENABLE_WARNING_READ_CONSISTENCY_CHECK_NOT_ENOUGH_REPLICA = true; }
// Health Monitor
init( FAILURE_DETECTION_DELAY, 4.0 ); if( randomize && BUGGIFY ) FAILURE_DETECTION_DELAY = 1.0;

3
flow/include/flow/FastAlloc.h
View File

@ -22,6 +22,7 @@
#define FLOW_FASTALLOC_H
#pragma once
#include "flow/Error.h"
#include "flow/Platform.h"
#include "flow/config.h"
@ -234,7 +235,7 @@ force_inline void freeOrMaybeKeepalive(void* ptr) {
}
inline constexpr int nextFastAllocatedSize(int x) {
assert(x > 0 && x <= 16384);
ASSERT(x > 0 && x <= 16384);
if (x <= 16)
return 16;
else if (x <= 32)

3
flow/include/flow/Knobs.h
View File

@ -376,7 +376,8 @@ public:
int TSS_LARGE_TRACE_SIZE;
double LOAD_BALANCE_FETCH_REPLICA_TIMEOUT;
bool ENABLE_REPLICA_CONSISTENCY_CHECK_ON_READS;
int CONSISTENCY_CHECK_REQUIRED_REPLICAS;
int READ_CONSISTENCY_CHECK_REQUIRED_REPLICAS;
bool ENABLE_WARNING_READ_CONSISTENCY_CHECK_NOT_ENOUGH_REPLICA;
// Health Monitor
int FAILURE_DETECTION_DELAY;