InsertChunks for CAS store (#55)

- Improvement: Add batching when writing multiple small chunks to block store - decreases I/O load significantly on oplog import

1 files changed, 50 insertions, 0 deletions

diff --git a/src/zenstore/cidstore.cpp b/src/zenstore/cidstore.cpp
index 67b7e95ac..68bccd06b 100644
--- a/src/zenstore/cidstore.cpp
+++ b/src/zenstore/cidstore.cpp
@@ -45,6 +45,50 @@ struct CidStore::Impl
 		return {.New = Result.New};
 	}
 
+	std::vector<CidStore::InsertResult> AddChunks(std::span<IoBuffer> ChunkDatas, std::span<IoHash> RawHashes, CidStore::InsertMode Mode)
+	{
+		if (ChunkDatas.size() == 1)
+		{
+			std::vector<CidStore::InsertResult> Result(1);
+			Result[0] = AddChunk(ChunkDatas[0], RawHashes[0], Mode);
+			return Result;
+		}
+		ZEN_ASSERT(ChunkDatas.size() == RawHashes.size());
+		std::vector<IoBuffer> Chunks;
+		Chunks.reserve(ChunkDatas.size());
+#if ZEN_BUILD_DEBUG
+		size_t Offset = 0;
+#endif
+		uint64_t TotalSize = 0;
+		for (const IoBuffer& ChunkData : ChunkDatas)
+		{
+			TotalSize += ChunkData.GetSize();
+#if ZEN_BUILD_DEBUG
+			IoHash	 VerifyRawHash;
+			uint64_t _;
+			ZEN_ASSERT(CompressedBuffer::ValidateCompressedHeader(ChunkData, VerifyRawHash, _) && RawHashes[Offset++] == VerifyRawHash);
+#endif
+			Chunks.push_back(ChunkData);
+			Chunks.back().SetContentType(ZenContentType::kCompressedBinary);
+		}
+
+		metrics::RequestStats::Scope $(m_AddChunkOps, TotalSize);
+
+		std::vector<CasStore::InsertResult> CasResults =
+			m_CasStore.InsertChunks(Chunks, RawHashes, static_cast<CasStore::InsertMode>(Mode));
+		ZEN_ASSERT(CasResults.size() == ChunkDatas.size());
+		std::vector<CidStore::InsertResult> Result;
+		for (const CasStore::InsertResult& CasResult : CasResults)
+		{
+			if (CasResult.New)
+			{
+				m_WriteCount++;
+			}
+			Result.emplace_back(CidStore::InsertResult{.New = CasResult.New});
+		}
+		return Result;
+	}
+
 	IoBuffer FindChunkByCid(const IoHash& DecompressedId)
 	{
 		metrics::RequestStats::Scope StatsScope(m_FindChunkOps, 0);
@@ -145,6 +189,12 @@ CidStore::AddChunk(const IoBuffer& ChunkData, const IoHash& RawHash, InsertMode
 	return m_Impl->AddChunk(ChunkData, RawHash, Mode);
 }
 
+std::vector<CidStore::InsertResult>
+CidStore::AddChunks(std::span<IoBuffer> ChunkDatas, std::span<IoHash> RawHashes, InsertMode Mode)
+{
+	return m_Impl->AddChunks(ChunkDatas, RawHashes, Mode);
+}
+
 IoBuffer
 CidStore::FindChunkByCid(const IoHash& DecompressedId)
 {