iterate cas chunks (#59)

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

2 files changed, 30 insertions, 28 deletions

diff --git a/src/zenserver/projectstore/projectstore.cpp b/src/zenserver/projectstore/projectstore.cpp
index 874715818..d177b0b2b 100644
--- a/src/zenserver/projectstore/projectstore.cpp
+++ b/src/zenserver/projectstore/projectstore.cpp
@@ -885,6 +885,14 @@ ProjectStore::Oplog::GetChunkByRawHash(const IoHash& RawHash)
 	return Chunk;
 }
 
+bool
+ProjectStore::Oplog::IterateChunks(std::span<IoHash>												 RawHashes,
+								   const std::function<bool(size_t Index, const IoBuffer& Payload)>& AsyncCallback,
+								   WorkerThreadPool*												 OptionalWorkerPool)
+{
+	return m_CidStore.IterateChunks(RawHashes, AsyncCallback, OptionalWorkerPool);
+}
+
 IoBuffer
 ProjectStore::Oplog::FindChunk(const Oid& ChunkId)
 {
@@ -2769,36 +2777,27 @@ ProjectStore::GetProjectChunkInfos(const std::string_view				  ProjectId,
 		}
 
 		WorkerThreadPool& WorkerPool = GetSmallWorkerPool();  // GetSyncWorkerPool();
-		Latch			  WorkLatch(1);
-
-		for (size_t Index = 0; Index < Hashes.size(); Index++)
-		{
-			WorkLatch.AddCount(1);
-			WorkerPool.ScheduleWork(
-				[&WorkLatch, FoundLog, WantsSizeField, WantsRawSizeField, &Sizes, &RawSizes, Index, RawHash = Hashes[Index]]() {
-					auto _ = MakeGuard([&WorkLatch]() { WorkLatch.CountDown(); });
-					if (IoBuffer Chunk = FoundLog->GetChunkByRawHash(RawHash))
+		(void)FoundLog->IterateChunks(
+			Hashes,
+			[&](size_t Index, const IoBuffer& Chunk) -> bool {
+				uint64_t Size = Chunk.GetSize();
+				if (WantsRawSizeField)
+				{
+					uint64_t RawSize = Size;
+					if (Chunk.GetContentType() == ZenContentType::kCompressedBinary)
 					{
-						uint64_t Size = Chunk.GetSize();
-						if (WantsRawSizeField)
-						{
-							uint64_t RawSize = Size;
-							if (Chunk.GetContentType() == ZenContentType::kCompressedBinary)
-							{
-								IoHash __;
-								(void)CompressedBuffer::FromCompressed(SharedBuffer(Chunk), __, RawSize);
-							}
-							RawSizes[Index] = RawSize;
-						}
-						if (WantsSizeField)
-						{
-							Sizes[Index] = Size;
-						}
+						IoHash __;
+						(void)CompressedBuffer::FromCompressed(SharedBuffer(Chunk), __, RawSize);
 					}
-				});
-		}
-		WorkLatch.CountDown();
-		WorkLatch.Wait();
+					RawSizes[Index] = RawSize;
+				}
+				if (WantsSizeField)
+				{
+					Sizes[Index] = Size;
+				}
+				return true;
+			},
+			&WorkerPool);
 	}
 
 	CbObjectWriter Response;
diff --git a/src/zenserver/projectstore/projectstore.h b/src/zenserver/projectstore/projectstore.h
index 269fe7336..9bae0382b 100644
--- a/src/zenserver/projectstore/projectstore.h
+++ b/src/zenserver/projectstore/projectstore.h
@@ -104,6 +104,9 @@ public:
 
 		IoBuffer FindChunk(const Oid& ChunkId);
 		IoBuffer GetChunkByRawHash(const IoHash& RawHash);
+		bool	 IterateChunks(std::span<IoHash>												 RawHashes,
+							   const std::function<bool(size_t Index, const IoBuffer& Payload)>& AsyncCallback,
+							   WorkerThreadPool*												 OptionalWorkerPool);
 
 		inline static const uint32_t kInvalidOp = ~0u;