iterate cas chunks (#59)

- Improvement: Reworked GetChunkInfos in oplog store to reduce disk thrashing and improve performance

1 files changed, 43 insertions, 10 deletions

diff --git a/src/zenstore/compactcas.cpp b/src/zenstore/compactcas.cpp
index f3c67eb9a..7b11200a5 100644
--- a/src/zenstore/compactcas.cpp
+++ b/src/zenstore/compactcas.cpp
@@ -299,6 +299,38 @@ CasContainerStrategy::FilterChunks(HashKeySet& InOutChunks)
 	InOutChunks.RemoveHashesIf([&](const IoHash& Hash) { return HaveChunk(Hash); });
 }
 
+bool
+CasContainerStrategy::IterateChunks(std::span<IoHash>												  ChunkHashes,
+									const std::function<bool(size_t Index, const IoBuffer& Payload)>& AsyncCallback,
+									WorkerThreadPool*												  OptionalWorkerPool)
+{
+	std::vector<size_t>				FoundChunkIndexes;
+	std::vector<BlockStoreLocation> FoundChunkLocations;
+	RwLock::SharedLockScope			_(m_LocationMapLock);
+	for (size_t ChunkIndex = 0; ChunkIndex < ChunkHashes.size(); ChunkIndex++)
+	{
+		if (auto KeyIt = m_LocationMap.find(ChunkHashes[ChunkIndex]); KeyIt != m_LocationMap.end())
+		{
+			FoundChunkIndexes.push_back(ChunkIndex);
+			FoundChunkLocations.push_back(m_Locations[KeyIt->second].Get(m_PayloadAlignment));
+		}
+	}
+	bool Continue = true;
+	m_BlockStore.IterateChunks(
+		FoundChunkLocations,
+		[&](size_t ChunkIndex, const void* Data, uint64_t Size) {
+			if (Data != nullptr)
+			{
+				Continue = Continue && AsyncCallback(FoundChunkIndexes[ChunkIndex], IoBuffer(IoBuffer::Wrap, Data, Size));
+			}
+		},
+		[&](size_t ChunkIndex, BlockStoreFile& File, uint64_t Offset, uint64_t Size) {
+			Continue = Continue && AsyncCallback(FoundChunkIndexes[ChunkIndex], File.GetChunk(Offset, Size));
+		},
+		OptionalWorkerPool);
+	return Continue;
+}
+
 void
 CasContainerStrategy::Flush()
 {
@@ -321,8 +353,9 @@ CasContainerStrategy::ScrubStorage(ScrubContext& Ctx)
 
 	ZEN_INFO("scrubbing '{}'", m_BlocksBasePath);
 
+	RwLock							BadKeysLock;
 	std::vector<IoHash>				BadKeys;
-	uint64_t						ChunkCount{0}, ChunkBytes{0};
+	std::atomic_uint64_t			ChunkCount{0}, ChunkBytes{0};
 	std::vector<BlockStoreLocation> ChunkLocations;
 	std::vector<IoHash>				ChunkIndexToChunkHash;
 
@@ -346,14 +379,14 @@ CasContainerStrategy::ScrubStorage(ScrubContext& Ctx)
 		}
 
 		const auto ValidateSmallChunk = [&](size_t ChunkIndex, const void* Data, uint64_t Size) {
-			++ChunkCount;
-			ChunkBytes += Size;
+			ChunkCount.fetch_add(1);
+			ChunkBytes.fetch_add(Size);
 
 			const IoHash& Hash = ChunkIndexToChunkHash[ChunkIndex];
 			if (!Data)
 			{
 				// ChunkLocation out of range of stored blocks
-				BadKeys.push_back(Hash);
+				BadKeysLock.WithExclusiveLock([&]() { BadKeys.push_back(Hash); });
 				return;
 			}
 
@@ -368,14 +401,14 @@ CasContainerStrategy::ScrubStorage(ScrubContext& Ctx)
 					return;
 				}
 			}
-			BadKeys.push_back(Hash);
+			BadKeysLock.WithExclusiveLock([&]() { BadKeys.push_back(Hash); });
 		};
 
 		const auto ValidateLargeChunk = [&](size_t ChunkIndex, BlockStoreFile& File, uint64_t Offset, uint64_t Size) {
 			Ctx.ThrowIfDeadlineExpired();
 
-			++ChunkCount;
-			ChunkBytes += Size;
+			ChunkCount.fetch_add(1);
+			ChunkBytes.fetch_add(Size);
 
 			const IoHash& Hash = ChunkIndexToChunkHash[ChunkIndex];
 			IoBuffer	  Buffer(IoBuffer::BorrowedFile, File.GetBasicFile().Handle(), Offset, Size);
@@ -391,10 +424,10 @@ CasContainerStrategy::ScrubStorage(ScrubContext& Ctx)
 					return;
 				}
 			}
-			BadKeys.push_back(Hash);
+			BadKeysLock.WithExclusiveLock([&]() { BadKeys.push_back(Hash); });
 		};
 
-		m_BlockStore.IterateChunks(ChunkLocations, ValidateSmallChunk, ValidateLargeChunk);
+		m_BlockStore.IterateChunks(ChunkLocations, ValidateSmallChunk, ValidateLargeChunk, nullptr);
 	}
 	catch (const ScrubDeadlineExpiredException&)
 	{
@@ -443,7 +476,7 @@ CasContainerStrategy::ScrubStorage(ScrubContext& Ctx)
 		Ctx.ReportBadCidChunks(BadKeys);
 	}
 
-	ZEN_INFO("scrubbed {} chunks ({}) in '{}'", ChunkCount, NiceBytes(ChunkBytes), m_RootDirectory / m_ContainerBaseName);
+	ZEN_INFO("scrubbed {} chunks ({}) in '{}'", ChunkCount.load(), NiceBytes(ChunkBytes.load()), m_RootDirectory / m_ContainerBaseName);
 }
 
 void