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

port PR #13243 to release-7.4 branch from release-7.3 branch. DD admission control related. (#13280) 

Originally this was PR #13112. That was ported to release-7.3 as PR #13243. Since release-7.4 is closer to release-7.3 than to main, port 13243 to release-7.4.
This commit is contained in:
gxglass 2026-05-28 11:51:09 -07:00 committed by GitHub
parent a7ed4eb323
commit cf6046297a
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
7 changed files with 99 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
fdbclient/ServerKnobs.cpp
View File

@ -149,6 +149,11 @@ void ServerKnobs::initialize(Randomize randomize, ClientKnobs* clientKnobs, IsSi
init( MERGE_RELOCATION_PARALLELISM_PER_TEAM, 6 ); if (randomize && BUGGIFY ) MERGE_RELOCATION_PARALLELISM_PER_TEAM = 1;
init( DD_QUEUE_MAX_KEY_SERVERS, 100 ); // Do not buggify
init( DD_REBALANCE_PARALLELISM, 50 );
// Hard cap on total relocations DD tracks (queued + in-flight). 1000 corresponds to a 500-server
// cluster with two concurrent shard moves per storage server. We have observed large clusters doing
// 25-30GB in flight, or closer to 100 shards at a time, so this has plenty of margin of safety
// built in.
init( DD_MAX_PIPELINE_MOVES, 1000 ); if( randomize && BUGGIFY ) DD_MAX_PIPELINE_MOVES = 5;
init( DD_REBALANCE_RESET_AMOUNT, 30 );
init( INFLIGHT_PENALTY_HEALTHY, 1.0 );
init( INFLIGHT_PENALTY_UNHEALTHY, 500.0 );

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

@ -137,6 +137,7 @@ public:
double MERGE_RELOCATION_PARALLELISM_PER_TEAM;
int DD_QUEUE_MAX_KEY_SERVERS;
int DD_REBALANCE_PARALLELISM;
int DD_MAX_PIPELINE_MOVES; // Hard cap on total relocations DD tracks (queued + in-flight).
int DD_REBALANCE_RESET_AMOUNT;
double INFLIGHT_PENALTY_HEALTHY;
double INFLIGHT_PENALTY_REDUNDANT;

63
fdbserver/DDRelocationQueue.actor.cpp
View File

@ -34,6 +34,7 @@
#include "fdbserver/DataDistribution.actor.h"
#include "fdbserver/MoveKeys.actor.h"
#include "fdbserver/Knobs.h"
#include "fdbserver/QuietDatabase.h"
#include "fdbrpc/simulator.h"
#include "fdbserver/DDTxnProcessor.h"
#include "flow/DebugTrace.h"
@ -627,7 +628,8 @@ DDQueue::DDQueue(DDQueueInitParams const& params)
finishMoveKeysParallelismLock(SERVER_KNOBS->DD_MOVE_KEYS_PARALLELISM),
cleanUpDataMoveParallelismLock(SERVER_KNOBS->DD_MOVE_KEYS_PARALLELISM),
fetchSourceLock(new FlowLock(SERVER_KNOBS->DD_FETCH_SOURCE_PARALLELISM)), activeRelocations(0),
queuedRelocations(0), bytesWritten(0), teamSize(params.teamSize), singleRegionTeamSize(params.singleRegionTeamSize),
queuedRelocations(0), pendingGateRelocations(0), bytesWritten(0), teamSize(params.teamSize),
singleRegionTeamSize(params.singleRegionTeamSize), pipelineFull(new AsyncVar<bool>(false)),
output(params.relocationProducer), input(params.relocationConsumer), getShardMetrics(params.getShardMetrics),
getTopKMetrics(params.getTopKMetrics), lastInterval(0), suppressIntervals(0),
rawProcessingUnhealthy(new AsyncVar<bool>(false)), rawProcessingWiggle(new AsyncVar<bool>(false)),
@ -636,6 +638,24 @@ DDQueue::DDQueue(DDQueueInitParams const& params)
retryFindDstReasonCount(static_cast<int>(RetryFindDstReason::NumberOfTypes), 0),
moveBytesRate(SERVER_KNOBS->DD_TRACE_MOVE_BYTES_AVERAGE_INTERVAL) {}
void DDQueue::updatePipelineFull() {
if (pipelineSize() >= SERVER_KNOBS->DD_MAX_PIPELINE_MOVES && !pipelineFull->get()) {
pipelineFull->set(true);
TraceEvent("DDPipelineFullSet", distributorId)
.suppressFor(30.0)
.detail("PipelineSize", pipelineSize())
.detail("PendingGateRelocations", pendingGateRelocations)
.detail("PipelineLimit", SERVER_KNOBS->DD_MAX_PIPELINE_MOVES);
} else if (pipelineSize() < SERVER_KNOBS->DD_MAX_PIPELINE_MOVES && pipelineFull->get()) {
pipelineFull->set(false);
TraceEvent("DDPipelineFullCleared", distributorId)
.suppressFor(30.0)
.detail("PipelineSize", pipelineSize())
.detail("PendingGateRelocations", pendingGateRelocations)
.detail("PipelineLimit", SERVER_KNOBS->DD_MAX_PIPELINE_MOVES);
}
}
void DDQueue::startRelocation(int priority, int healthPriority) {
// Although PRIORITY_TEAM_REDUNDANT has lower priority than split and merge shard movement,
// we must count it into unhealthyRelocations; because team removers relies on unhealthyRelocations to
@ -654,6 +674,7 @@ void DDQueue::startRelocation(int priority, int healthPriority) {
rawProcessingWiggle->set(true);
}
priority_relocations[priority]++;
updatePipelineFull();
}
void DDQueue::finishRelocation(int priority, int healthPriority) {
@ -669,6 +690,7 @@ void DDQueue::finishRelocation(int priority, int healthPriority) {
}
}
priority_relocations[priority]--;
updatePipelineFull();
if (priority_relocations[SERVER_KNOBS->PRIORITY_PERPETUAL_STORAGE_WIGGLE] == 0) {
rawProcessingWiggle->set(false);
}
@ -2812,6 +2834,32 @@ ACTOR Future<Void> BgDDLoadRebalance(DDQueue* self, int teamCollectionIndex, Dat
}
}
// Gates the relocation input stream by the pipeline limit. Cancellations and high-priority
// moves (>= PRIORITY_TEAM_UNHEALTHY) always pass through immediately so that failure recovery
// is never blocked by stuck or zombie moves holding pipeline slots. All other relocations are
// held when the pipeline is full, waiting for pipelineFull to become false before forwarding.
// The global isDDPipelineControlEnabled() flag (cleared by disableDDPipelineControl()) also
// bypasses the gate, allowing the test harness to open up the pipeline so DD can quiesce.
// We poll it via delay() rather than AsyncVar to avoid cross-process callbacks in simulation.
ACTOR Future<Void> pipelineGateActor(Reference<DDQueue> self,
FutureStream<RelocateShard> input,
PromiseStream<RelocateShard> output) {
loop {
state RelocateShard rs = waitNext(input);
if (!rs.cancelled && rs.priority < SERVER_KNOBS->PRIORITY_TEAM_UNHEALTHY) {
while (self->pipelineFull->get() && isDDPipelineControlEnabled()) {
TraceEvent("DDPipelineFull", self->distributorId)
.suppressFor(30.0)
.detail("PipelineFull", self->pipelineFull->get());
wait(self->pipelineFull->onChange() || delay(1.0));
}
}
self->pendingGateRelocations++;
self->updatePipelineFull();
output.send(rs);
}
}
struct DDQueueImpl {
ACTOR static Future<Void> run(Reference<DDQueue> self,
Reference<AsyncVar<bool>> processingUnhealthy,
@ -2831,6 +2879,11 @@ struct DDQueueImpl {
state Future<Void> onCleanUpDataMoveActorError =
actorCollection(self->addBackgroundCleanUpDataMoveActor.getFuture());
// Gate the input stream by the pipeline limit so that DD never tracks more
// than DD_MAX_PIPELINE_MOVES relocations at once (queued + in-flight).
state PromiseStream<RelocateShard> gatedRelocationStream;
state Future<Void> pipelineGate = pipelineGateActor(self, self->input, gatedRelocationStream);
for (int i = 0; i < self->teamCollections.size(); i++) {
ddQueueFutures.push_back(
BgDDLoadRebalance(self.getPtr(), i, DataMovementReason::REBALANCE_OVERUTILIZED_TEAM));
@ -2865,7 +2918,9 @@ struct DDQueueImpl {
ASSERT(launchData.startTime == -1 && keysToLaunchFrom.empty());
choose {
when(RelocateShard rs = waitNext(self->input)) {
when(RelocateShard rs = waitNext(gatedRelocationStream.getFuture())) {
self->pendingGateRelocations--;
self->updatePipelineFull();
if (rs.isRestore()) {
ASSERT(rs.dataMove != nullptr);
ASSERT(rs.dataMoveId.isValid());
@ -2932,6 +2987,9 @@ struct DDQueueImpl {
.detail("HighestPriority", highestPriorityRelocation)
.detail("BytesWritten", self->moveBytesRate.getTotal())
.detail("BytesWrittenAverageRate", self->moveBytesRate.getAverage())
.detail("PipelineSize", self->pipelineSize())
.detail("PipelineLimit", SERVER_KNOBS->DD_MAX_PIPELINE_MOVES)
.detail("PendingGateRelocations", self->pendingGateRelocations)
.detail("PriorityRecoverMove",
self->priority_relocations[SERVER_KNOBS->PRIORITY_RECOVER_MOVE])
.detail("PriorityRebalanceUnderutilizedTeam",
@ -3027,6 +3085,7 @@ struct DDQueueImpl {
}
when(wait(self->error.getFuture())) {} // Propagate errors from dataDistributionRelocator
when(wait(waitForAll(ddQueueFutures))) {}
when(wait(pipelineGate)) {} // Propagate errors from pipelineGateActor
when(Promise<int> r = waitNext(getUnhealthyRelocationCount)) {
r.send(self->getUnhealthyRelocationCount());
}

15
fdbserver/QuietDatabase.actor.cpp
View File

@ -40,6 +40,17 @@
#include "fdbclient/ManagementAPI.actor.h"
#include "flow/actorcompiler.h" // This must be the last #include.
static bool g_ddPipelineControlEnabled = true;
bool isDDPipelineControlEnabled() {
return g_ddPipelineControlEnabled;
}
void disableDDPipelineControl() {
TraceEvent("DDPipelineControlDisabled");
g_ddPipelineControlEnabled = false;
}
ACTOR Future<std::vector<WorkerDetails>> getWorkers(Reference<AsyncVar<ServerDBInfo> const> dbInfo, int flags = 0) {
loop {
choose {
@ -1058,6 +1069,10 @@ ACTOR Future<Void> waitForQuietDatabase(Database cx,
state Version version = wait(setPerpetualStorageWiggle(cx, false, LockAware::True));
printf("Set perpetual_storage_wiggle=0 Done.\n");
if (g_network->isSimulated()) {
disableDDPipelineControl();
}
printf("Disabling backup worker ...\n");
wait(disableBackupWorker(cx));
printf("Disabled backup worker.\n");

7
fdbserver/include/fdbserver/DDRelocationQueue.h
View File

@ -256,10 +256,17 @@ public:
int activeRelocations;
int queuedRelocations;
int pendingGateRelocations; // forwarded by pipelineGateActor but not yet consumed by DDQueue
int64_t bytesWritten;
int teamSize;
int singleRegionTeamSize;
int pipelineSize() const { return pendingGateRelocations + activeRelocations + queuedRelocations; }
void updatePipelineFull();
Reference<AsyncVar<bool>> pipelineFull;
std::map<UID, Busyness> busymap; // UID is serverID
std::map<UID, Busyness> destBusymap; // UID is serverID

8
fdbserver/include/fdbserver/DataDistribution.actor.h
View File

@ -151,13 +151,13 @@ struct RelocateShard {
void setParentRange(KeyRange const& parent);
Optional<KeyRange> getParentRange() const;
private:
// If this rs comes from a splitting, parent range is the original range.
Optional<KeyRange> parent_range;
RelocateShard()
: priority(0), cancelled(false), dataMoveId(anonymousShardId), reason(RelocateReason::OTHER),
moveReason(DataMovementReason::INVALID) {}
private:
// If this rs comes from a splitting, parent range is the original range.
Optional<KeyRange> parent_range;
};
struct GetMetricsRequest {

6
fdbserver/include/fdbserver/QuietDatabase.h
View File

@ -68,4 +68,10 @@ Future<std::vector<WorkerInterface>> getCoordWorkers(Database const& cx,
Future<Void> enableConsistencyScanInSim(Database const& db);
Future<Void> disableConsistencyScanInSim(Database const& db, bool const& waitForCompletion);
// Permanently disables DD pipeline control so that all blocked relocations pass through.
// For use by the test harness to allow DD to quiesce after tests complete.
// Uses a plain boolean (not AsyncVar) to avoid cross-process callback issues in simulation.
void disableDDPipelineControl();
bool isDDPipelineControlEnabled();
#endif