diff --git a/fdbserver/RestoreLoader.actor.cpp b/fdbserver/RestoreLoader.actor.cpp index 3abfa1f060..d6f6379859 100644 --- a/fdbserver/RestoreLoader.actor.cpp +++ b/fdbserver/RestoreLoader.actor.cpp @@ -336,6 +336,8 @@ ACTOR Future handleLoadFileRequest(RestoreLoadFileRequest req, ReferencefinishedBatch.set(req.batchIndex); + ASSERT(self->finishedBatch.get() < req.batchIndex); wait(isSchedulable(self, req.batchIndex, __FUNCTION__)); @@ -376,6 +378,8 @@ ACTOR Future handleSendMutationsRequest(RestoreSendMutationsToAppliersRequ .detail("BatchIndex", req.batchIndex) .detail("UseRangeFile", req.useRangeFile) .detail("LoaderSendStatus", batchStatus->toString()); + // Loader destroy batchData once the batch finishes and self->finishedBatch.set(req.batchIndex); + ASSERT(self->finishedBatch.get() < req.batchIndex); // Ensure each file is sent exactly once by using batchStatus->sendAllLogs and batchStatus->sendAllRanges if (!req.useRangeFile) { @@ -945,6 +949,9 @@ ACTOR Future handleFinishVersionBatchRequest(RestoreVersionBatchRequest re wait(self->finishedBatch.whenAtLeast(req.batchIndex - 1)); if (self->finishedBatch.get() == req.batchIndex - 1) { self->finishedBatch.set(req.batchIndex); + // Clean up batchData + self->batch.erase(req.batchIndex); + self->status.erase(req.batchIndex); } if (self->delayedActors > 0) { self->checkMemory.trigger(); diff --git a/fdbserver/RestoreLoader.actor.h b/fdbserver/RestoreLoader.actor.h index d2469fbcaa..f6fc772819 100644 --- a/fdbserver/RestoreLoader.actor.h +++ b/fdbserver/RestoreLoader.actor.h @@ -177,6 +177,7 @@ struct RestoreLoaderData : RestoreRoleData, public ReferenceCounted handleInitVersionBatchRequest(RestoreVersionBatchRequest req, .detail("BatchIndex", req.batchIndex) .detail("Role", getRoleStr(self->role)) .detail("VersionBatchNotifiedVersion", self->versionBatchId.get()); + // Loader destroy batchData once the batch finishes and self->finishedBatch.set(req.batchIndex); + ASSERT(self->finishedBatch.get() < req.batchIndex); + // batchId is continuous. (req.batchIndex-1) is the id of the just finished batch. wait(self->versionBatchId.whenAtLeast(req.batchIndex - 1));