From cf6046297a167a45e53a11b75f63b831c7efdff2 Mon Sep 17 00:00:00 2001 From: gxglass Date: Thu, 28 May 2026 11:51:09 -0700 Subject: [PATCH] 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. --- fdbclient/ServerKnobs.cpp | 5 ++ fdbclient/include/fdbclient/ServerKnobs.h | 1 + fdbserver/DDRelocationQueue.actor.cpp | 63 ++++++++++++++++++- fdbserver/QuietDatabase.actor.cpp | 15 +++++ .../include/fdbserver/DDRelocationQueue.h | 7 +++ .../fdbserver/DataDistribution.actor.h | 8 +-- fdbserver/include/fdbserver/QuietDatabase.h | 6 ++ 7 files changed, 99 insertions(+), 6 deletions(-) diff --git a/fdbclient/ServerKnobs.cpp b/fdbclient/ServerKnobs.cpp index 1154f1359d..50d66add3f 100644 --- a/fdbclient/ServerKnobs.cpp +++ b/fdbclient/ServerKnobs.cpp @@ -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 ); diff --git a/fdbclient/include/fdbclient/ServerKnobs.h b/fdbclient/include/fdbclient/ServerKnobs.h index aff5082dd4..d26ec026ff 100644 --- a/fdbclient/include/fdbclient/ServerKnobs.h +++ b/fdbclient/include/fdbclient/ServerKnobs.h @@ -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; diff --git a/fdbserver/DDRelocationQueue.actor.cpp b/fdbserver/DDRelocationQueue.actor.cpp index 926b6c349d..eb12065672 100644 --- a/fdbserver/DDRelocationQueue.actor.cpp +++ b/fdbserver/DDRelocationQueue.actor.cpp @@ -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(false)), output(params.relocationProducer), input(params.relocationConsumer), getShardMetrics(params.getShardMetrics), getTopKMetrics(params.getTopKMetrics), lastInterval(0), suppressIntervals(0), rawProcessingUnhealthy(new AsyncVar(false)), rawProcessingWiggle(new AsyncVar(false)), @@ -636,6 +638,24 @@ DDQueue::DDQueue(DDQueueInitParams const& params) retryFindDstReasonCount(static_cast(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 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 pipelineGateActor(Reference self, + FutureStream input, + PromiseStream 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 run(Reference self, Reference> processingUnhealthy, @@ -2831,6 +2879,11 @@ struct DDQueueImpl { state Future 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 gatedRelocationStream; + state Future 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 r = waitNext(getUnhealthyRelocationCount)) { r.send(self->getUnhealthyRelocationCount()); } diff --git a/fdbserver/QuietDatabase.actor.cpp b/fdbserver/QuietDatabase.actor.cpp index b9a2f2b590..af269f0063 100644 --- a/fdbserver/QuietDatabase.actor.cpp +++ b/fdbserver/QuietDatabase.actor.cpp @@ -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> getWorkers(Reference const> dbInfo, int flags = 0) { loop { choose { @@ -1058,6 +1069,10 @@ ACTOR Future 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"); diff --git a/fdbserver/include/fdbserver/DDRelocationQueue.h b/fdbserver/include/fdbserver/DDRelocationQueue.h index aed4f295e6..98b339ae00 100644 --- a/fdbserver/include/fdbserver/DDRelocationQueue.h +++ b/fdbserver/include/fdbserver/DDRelocationQueue.h @@ -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> pipelineFull; + std::map busymap; // UID is serverID std::map destBusymap; // UID is serverID diff --git a/fdbserver/include/fdbserver/DataDistribution.actor.h b/fdbserver/include/fdbserver/DataDistribution.actor.h index d410e697cb..5e4ef4505a 100644 --- a/fdbserver/include/fdbserver/DataDistribution.actor.h +++ b/fdbserver/include/fdbserver/DataDistribution.actor.h @@ -151,13 +151,13 @@ struct RelocateShard { void setParentRange(KeyRange const& parent); Optional getParentRange() const; -private: - // If this rs comes from a splitting, parent range is the original range. - Optional 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 parent_range; }; struct GetMetricsRequest { diff --git a/fdbserver/include/fdbserver/QuietDatabase.h b/fdbserver/include/fdbserver/QuietDatabase.h index 2b05947561..8157887b2f 100644 --- a/fdbserver/include/fdbserver/QuietDatabase.h +++ b/fdbserver/include/fdbserver/QuietDatabase.h @@ -68,4 +68,10 @@ Future> getCoordWorkers(Database const& cx, Future enableConsistencyScanInSim(Database const& db); Future 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