add parallelism check

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);

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;

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<AsyncVar<ServerDBInfo> const> const& db, UID id, Reference<DDSharedContext> 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<Void> doBulkDumpTask(Reference<DataDistributor> self,
 			throw e;
 		}
 	}
+	self->bulkDumpParallelismLimitor.decrementTaskCounter();
 	return Void();
 }
 
@@ -1374,7 +1377,7 @@ ACTOR Future<bool> scheduleBulkDumpTasks(Reference<DataDistributor> self) {
 
 	state int rangeLocationIndex = 0;
 	state std::vector<IDDTxnProcessor::DDRangeLocations> rangeLocations;
-
+	state KeyRange taskRange;
 	state bool allComplete = true;
 
 	while (beginKey < endKey) {
@@ -1403,9 +1406,16 @@ ACTOR Future<bool> scheduleBulkDumpTasks(Reference<DataDistributor> 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<Void> bulkDumpingCore(Reference<DataDistributor> self, Future<Void>
 	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<Void> bulkDumpingCore(Reference<DataDistributor> self, Future<Void>
 				throw e;
 			}
 		}
-		self->bulkDumpActors.clear(false);
 		wait(delay(SERVER_KNOBS->DD_BULKDUMP_SCHEDULE_MIN_INTERVAL_SEC));
 	}
 }

fdbserver/include/fdbserver/BulkDumpUtil.actor.h

@@ -101,5 +101,32 @@ ACTOR Future<Void> uploadFiles(BulkDumpTransportMethod transportMethod,
 
 ACTOR Future<Void> 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<Void> waitUntilCounterChanged() const { return numRunningTasks.onChange(); }
+
+private:
+	AsyncVar<int> numRunningTasks;
+	int maxParallelism;
+};
+
 #include "flow/unactorcompiler.h"
 #endif

fdbserver/storageserver.actor.cpp

@@ -1715,7 +1715,7 @@ public:
 	    ssLock(makeReference<PriorityMultiLock>(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<RangeDumpData> 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<GetKeyValuesReply> rep = wait(errorOr(localReq.reply.getFuture()));