From 86979abdb061e5f2d09e10f26df25e39a5d7c261 Mon Sep 17 00:00:00 2001 From: Zhe Wang Date: Tue, 19 Nov 2024 18:59:15 -0800 Subject: [PATCH] add parallelism check --- fdbclient/ServerKnobs.cpp | 3 ++- fdbclient/include/fdbclient/ServerKnobs.h | 3 ++- fdbserver/DataDistribution.actor.cpp | 22 ++++++++++----- .../include/fdbserver/BulkDumpUtil.actor.h | 27 +++++++++++++++++++ fdbserver/storageserver.actor.cpp | 6 ++--- 5 files changed, 49 insertions(+), 12 deletions(-) diff --git a/fdbclient/ServerKnobs.cpp b/fdbclient/ServerKnobs.cpp index 403d8c5519..83de8e1156 100644 --- a/fdbclient/ServerKnobs.cpp +++ b/fdbclient/ServerKnobs.cpp @@ -384,7 +384,8 @@ void ServerKnobs::initialize(Randomize randomize, ClientKnobs* clientKnobs, IsSi // BulkDumping init( DD_BULKDUMP_TASK_METADATA_READ_SIZE, 100 ); if( randomize && BUGGIFY ) DD_BULKDUMP_TASK_METADATA_READ_SIZE = deterministicRandom()->randomInt(2, 100); init( DD_BULKDUMP_SCHEDULE_MIN_INTERVAL_SEC, 2.0 ); if( randomize && BUGGIFY ) DD_BULKDUMP_SCHEDULE_MIN_INTERVAL_SEC = deterministicRandom()->random01() * 10 + 1; - init( SS_SERVE_BULK_DUMP_PARALLELISM, 1 ); // TODO(BulkDump): Do not set to 1 after SS can resolve the file folder conflict + init( SS_SERVE_BULKDUMP_PARALLELISM, 1 ); // TODO(BulkDump): Do not set to 1 after SS can resolve the file folder conflict + init( DD_BULKDUMP_PARALLELISM, 50 ); if ( randomize && BUGGIFY ) DD_BULKDUMP_PARALLELISM = deterministicRandom()->randomInt(1, 5); // TeamRemover init( TR_LOW_SPACE_PIVOT_DELAY_SEC, 0 ); if (isSimulated) TR_LOW_SPACE_PIVOT_DELAY_SEC = deterministicRandom()->randomInt(0, 3); diff --git a/fdbclient/include/fdbclient/ServerKnobs.h b/fdbclient/include/fdbclient/ServerKnobs.h index b19c0a4c0f..eff9af53d7 100644 --- a/fdbclient/include/fdbclient/ServerKnobs.h +++ b/fdbclient/include/fdbclient/ServerKnobs.h @@ -403,7 +403,8 @@ public: int DD_BULKDUMP_TASK_METADATA_READ_SIZE; // the number of bulk dump tasks read from metadata at a time double DD_BULKDUMP_SCHEDULE_MIN_INTERVAL_SEC; // the minimal seconds that the bulk dump scheduler has to wait // between two rounds - int SS_SERVE_BULK_DUMP_PARALLELISM; // the number of bulk dump tasks that can concurrently happen at a SS + int SS_SERVE_BULKDUMP_PARALLELISM; // the number of bulk dump tasks that can concurrently happen at a SS + int DD_BULKDUMP_PARALLELISM; // the max number of concurrent bulk dump tasks in DD // Run storage engine on a child process on the same machine with storage process bool REMOTE_KV_STORE; diff --git a/fdbserver/DataDistribution.actor.cpp b/fdbserver/DataDistribution.actor.cpp index 448abbea09..697eef7d97 100644 --- a/fdbserver/DataDistribution.actor.cpp +++ b/fdbserver/DataDistribution.actor.cpp @@ -425,9 +425,10 @@ public: ActorCollection bulkLoadActors; bool bulkLoadEnabled = false; - ActorCollection bulkDumpActors; + bool bulkDumpEnabled = false; KeyRangeActorMap ongoingBulkDumpActors; + ParallelismLimitor bulkDumpParallelismLimitor; DataDistributor(Reference const> const& db, UID id, Reference context) : dbInfo(db), context(context), ddId(id), txnProcessor(nullptr), lock(context->lock), @@ -438,7 +439,8 @@ public: teamCollection(nullptr), bulkLoadTaskCollection(nullptr), auditStorageHaLaunchingLock(1), auditStorageReplicaLaunchingLock(1), auditStorageLocationMetadataLaunchingLock(1), auditStorageSsShardLaunchingLock(1), auditStorageInitStarted(false), bulkLoadActors(false), - bulkLoadEnabled(false), bulkDumpEnabled(false) {} + bulkLoadEnabled(false), bulkDumpEnabled(false), + bulkDumpParallelismLimitor(SERVER_KNOBS->DD_BULKDUMP_PARALLELISM) {} // bootstrap steps @@ -1357,6 +1359,7 @@ ACTOR Future doBulkDumpTask(Reference self, throw e; } } + self->bulkDumpParallelismLimitor.decrementTaskCounter(); return Void(); } @@ -1374,7 +1377,7 @@ ACTOR Future scheduleBulkDumpTasks(Reference self) { state int rangeLocationIndex = 0; state std::vector rangeLocations; - + state KeyRange taskRange; state bool allComplete = true; while (beginKey < endKey) { @@ -1403,9 +1406,16 @@ ACTOR Future scheduleBulkDumpTasks(Reference self) { rangeLocationIndex = 0; for (; rangeLocationIndex < rangeLocations.size(); ++rangeLocationIndex) { // Spawn task per shard - KeyRange taskRange = rangeLocations[rangeLocationIndex].range; + taskRange = rangeLocations[rangeLocationIndex].range; ASSERT(!taskRange.empty()); if (!self->ongoingBulkDumpActors.liveActorAt(taskRange.begin)) { + // Limit parallelism + loop { + if (self->bulkDumpParallelismLimitor.tryIncrementTaskCounter()) { + break; + } + wait(self->bulkDumpParallelismLimitor.waitUntilCounterChanged()); + } // In case no ongoing task on the same range SSBulkDumpTask task = getSSBulkDumpTask(rangeLocations[rangeLocationIndex].servers, bulkDumpState.spawn(taskRange)); @@ -1453,8 +1463,7 @@ ACTOR Future bulkDumpingCore(Reference self, Future state Database cx = self->txnProcessor->context(); loop { try { - self->bulkDumpActors.add(bulkDumpTaskScheduler(self)); - wait(self->bulkDumpActors.getResult()); + wait(bulkDumpTaskScheduler(self)); } catch (Error& e) { if (e.code() == error_code_actor_cancelled) { throw e; @@ -1464,7 +1473,6 @@ ACTOR Future bulkDumpingCore(Reference self, Future throw e; } } - self->bulkDumpActors.clear(false); wait(delay(SERVER_KNOBS->DD_BULKDUMP_SCHEDULE_MIN_INTERVAL_SEC)); } } diff --git a/fdbserver/include/fdbserver/BulkDumpUtil.actor.h b/fdbserver/include/fdbserver/BulkDumpUtil.actor.h index f248bb93d2..0e8da0d8c7 100644 --- a/fdbserver/include/fdbserver/BulkDumpUtil.actor.h +++ b/fdbserver/include/fdbserver/BulkDumpUtil.actor.h @@ -101,5 +101,32 @@ ACTOR Future uploadFiles(BulkDumpTransportMethod transportMethod, ACTOR Future persistCompleteBulkDumpRange(Database cx, BulkDumpState bulkDumpState); +class ParallelismLimitor { +public: + ParallelismLimitor(int maxParallelism) : maxParallelism(maxParallelism) {} + + inline void decrementTaskCounter() { + ASSERT(numRunningTasks.get() <= maxParallelism); + numRunningTasks.set(numRunningTasks.get() - 1); + ASSERT(numRunningTasks.get() >= 0); + } + + // return true if succeed + inline bool tryIncrementTaskCounter() { + if (numRunningTasks.get() < maxParallelism) { + numRunningTasks.set(numRunningTasks.get() + 1); + return true; + } else { + return false; + } + } + + inline Future waitUntilCounterChanged() const { return numRunningTasks.onChange(); } + +private: + AsyncVar numRunningTasks; + int maxParallelism; +}; + #include "flow/unactorcompiler.h" #endif diff --git a/fdbserver/storageserver.actor.cpp b/fdbserver/storageserver.actor.cpp index 4b00164b92..097c291b74 100644 --- a/fdbserver/storageserver.actor.cpp +++ b/fdbserver/storageserver.actor.cpp @@ -1715,7 +1715,7 @@ public: ssLock(makeReference(SERVER_KNOBS->STORAGE_SERVER_READ_CONCURRENCY, SERVER_KNOBS->STORAGESERVER_READ_PRIORITIES)), serveAuditStorageParallelismLock(SERVER_KNOBS->SERVE_AUDIT_STORAGE_PARALLELISM), - serveBulkDumpParallelismLock(SERVER_KNOBS->SS_SERVE_BULK_DUMP_PARALLELISM), + serveBulkDumpParallelismLock(SERVER_KNOBS->SS_SERVE_BULKDUMP_PARALLELISM), instanceID(deterministicRandom()->randomUniqueID().first()), shuttingDown(false), behind(false), versionBehind(false), debug_inApplyUpdate(false), debug_lastValidateTime(0), lastBytesInputEBrake(0), lastDurableVersionEBrake(0), maxQueryQueue(0), @@ -5993,8 +5993,8 @@ ACTOR Future getRangeDataToDump(StorageServer* data, KeyRange ran localReq.begin = firstGreaterOrEqual(range.begin); localReq.end = firstGreaterOrEqual(range.end); localReq.version = version; - localReq.limit = 1e6; // TODO(BulkDump): make this configurable - localReq.limitBytes = 1e8; + localReq.limit = SERVER_KNOBS->MOVE_SHARD_KRM_ROW_LIMIT; + localReq.limitBytes = SERVER_KNOBS->MOVE_SHARD_KRM_BYTE_LIMIT; localReq.tags = TagSet(); data->actors.add(getKeyValuesQ(data, localReq)); state ErrorOr rep = wait(errorOr(localReq.reply.getFuture()));