add disk caching to block move (#661)

* add disk caching to block move

1 files changed, 36 insertions, 23 deletions

diff --git a/src/zenstore/blockstore.cpp b/src/zenstore/blockstore.cpp
index 830ad9e55..69487f9dc 100644
--- a/src/zenstore/blockstore.cpp
+++ b/src/zenstore/blockstore.cpp
@@ -1096,24 +1096,26 @@ BlockStore::CompactBlocks(const BlockStoreCompactState&	  CompactState,
 	uint64_t AddedSize	 = 0;
 	uint64_t RemovedSize = 0;
 
-	Ref<BlockStoreFile> NewBlockFile;
-	auto				NewBlockFileGuard = MakeGuard([&]() {
-		   if (NewBlockFile)
-		   {
-			   {
-				   RwLock::ExclusiveLockScope _l(m_InsertLock);
-				   if (m_ChunkBlocks[NewBlockIndex] == NewBlockFile)
-				   {
-					   m_ChunkBlocks.erase(NewBlockIndex);
-				   }
-			   }
-			   if (NewBlockFile->IsOpen())
-			   {
-				   ZEN_DEBUG("{}Dropping incomplete cas block store file '{}'", LogPrefix, NewBlockFile->GetPath());
-				   NewBlockFile->MarkAsDeleteOnClose();
-			   }
-		   }
-	   });
+	Ref<BlockStoreFile>				 NewBlockFile;
+	std::unique_ptr<BasicFileWriter> TargetFileBuffer;
+	auto							 NewBlockFileGuard = MakeGuard([&]() {
+		TargetFileBuffer.reset();
+		if (NewBlockFile)
+		{
+			{
+				RwLock::ExclusiveLockScope _l(m_InsertLock);
+				if (m_ChunkBlocks[NewBlockIndex] == NewBlockFile)
+				{
+					m_ChunkBlocks.erase(NewBlockIndex);
+				}
+			}
+			if (NewBlockFile->IsOpen())
+			{
+				ZEN_DEBUG("{}Dropping incomplete cas block store file '{}'", LogPrefix, NewBlockFile->GetPath());
+				NewBlockFile->MarkAsDeleteOnClose();
+			}
+		}
+	});
 
 	auto ReportChanges = [&]() -> bool {
 		bool Continue = true;
@@ -1178,9 +1180,14 @@ BlockStore::CompactBlocks(const BlockStoreCompactState&	  CompactState,
 		}
 		else
 		{
+			std::vector<size_t> SortedChunkIndexes(KeepChunkIndexes);
+			std::sort(SortedChunkIndexes.begin(), SortedChunkIndexes.end(), [&ChunkLocations](size_t Lhs, size_t Rhs) {
+				return ChunkLocations[Lhs].Offset < ChunkLocations[Rhs].Offset;
+			});
+			BasicFileBuffer		 SourceFileBuffer(OldBlockFile->GetBasicFile(), Min(65536u, OldBlockSize));
 			uint64_t			 MovedFromBlock = 0;
 			std::vector<uint8_t> Chunk;
-			for (const size_t& ChunkIndex : KeepChunkIndexes)
+			for (const size_t& ChunkIndex : SortedChunkIndexes)
 			{
 				const BlockStoreLocation ChunkLocation = ChunkLocations[ChunkIndex];
 				if (ChunkLocation.Offset + ChunkLocation.Size > OldBlockSize)
@@ -1198,10 +1205,11 @@ BlockStore::CompactBlocks(const BlockStoreCompactState&	  CompactState,
 				}
 
 				Chunk.resize(ChunkLocation.Size);
-				OldBlockFile->Read(Chunk.data(), Chunk.size(), ChunkLocation.Offset);
+				SourceFileBuffer.Read(Chunk.data(), Chunk.size(), ChunkLocation.Offset);
 
 				if ((WriteOffset + Chunk.size()) > m_MaxBlockSize)
 				{
+					TargetFileBuffer.reset();
 					if (NewBlockFile)
 					{
 						ZEN_ASSERT_SLOW(NewBlockFile->IsOpen());
@@ -1283,11 +1291,12 @@ BlockStore::CompactBlocks(const BlockStoreCompactState&	  CompactState,
 								 NiceBytes(Space.Free + ReclaimedSpace));
 					}
 					NewBlockFile->Create(m_MaxBlockSize);
-					NewBlockIndex = NextBlockIndex;
-					WriteOffset	  = 0;
+					NewBlockIndex	 = NextBlockIndex;
+					WriteOffset		 = 0;
+					TargetFileBuffer = std::make_unique<BasicFileWriter>(NewBlockFile->GetBasicFile(), Min(65536u, m_MaxBlockSize));
 				}
 
-				NewBlockFile->Write(Chunk.data(), ChunkLocation.Size, WriteOffset);
+				TargetFileBuffer->Write(Chunk.data(), ChunkLocation.Size, WriteOffset);
 				MovedChunks.push_back(
 					{ChunkIndex, {.BlockIndex = NewBlockIndex, .Offset = gsl::narrow<uint32_t>(WriteOffset), .Size = ChunkLocation.Size}});
 				uint64_t WriteEndOffset = WriteOffset + ChunkLocation.Size;
@@ -1302,6 +1311,10 @@ BlockStore::CompactBlocks(const BlockStoreCompactState&	  CompactState,
 					 GetBlockPath(m_BlocksBasePath, BlockIndex).filename(),
 					 NiceBytes(OldBlockSize - MovedFromBlock));
 		}
+		if (TargetFileBuffer)
+		{
+			TargetFileBuffer->Flush();
+		}
 
 		if (!ReportChanges())
 		{