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, 44 insertions, 15 deletions

diff --git a/src/zenserver/projectstore/projectstore.cpp b/src/zenserver/projectstore/projectstore.cpp
index 3a7922aaf..3c281275e 100644
--- a/src/zenserver/projectstore/projectstore.cpp
+++ b/src/zenserver/projectstore/projectstore.cpp
@@ -1448,19 +1448,36 @@ ProjectStore::Oplog::AppendNewOplogEntry(CbPackage OpPackage)
 
 	auto Attachments = OpPackage.GetAttachments();
 
-	for (const auto& Attach : Attachments)
+	if (!Attachments.empty())
 	{
-		ZEN_ASSERT(Attach.IsCompressedBinary());
+		std::vector<IoBuffer> WriteAttachmentBuffers;
+		std::vector<IoHash>	  WriteRawHashes;
+		std::vector<uint64_t> WriteRawSizes;
 
-		CompressedBuffer	   AttachmentData = Attach.AsCompressedBinary();
-		const uint64_t		   AttachmentSize = AttachmentData.DecodeRawSize();
-		CidStore::InsertResult InsertResult = m_CidStore.AddChunk(AttachmentData.GetCompressed().Flatten().AsIoBuffer(), Attach.GetHash());
+		WriteAttachmentBuffers.reserve(Attachments.size());
+		WriteRawHashes.reserve(Attachments.size());
+		WriteRawSizes.reserve(Attachments.size());
 
-		if (InsertResult.New)
+		for (const auto& Attach : Attachments)
 		{
-			NewAttachmentBytes += AttachmentSize;
+			ZEN_ASSERT(Attach.IsCompressedBinary());
+
+			CompressedBuffer AttachmentData = Attach.AsCompressedBinary();
+			const uint64_t	 AttachmentSize = AttachmentData.DecodeRawSize();
+			WriteAttachmentBuffers.push_back(AttachmentData.GetCompressed().Flatten().AsIoBuffer());
+			WriteRawHashes.push_back(Attach.GetHash());
+			WriteRawSizes.push_back(AttachmentSize);
+			AttachmentBytes += AttachmentSize;
+		}
+
+		std::vector<CidStore::InsertResult> InsertResults = m_CidStore.AddChunks(WriteAttachmentBuffers, WriteRawHashes);
+		for (size_t Index = 0; Index < InsertResults.size(); Index++)
+		{
+			if (InsertResults[Index].New)
+			{
+				NewAttachmentBytes += WriteRawSizes[Index];
+			}
 		}
-		AttachmentBytes += AttachmentSize;
 	}
 
 	ZEN_DEBUG("oplog entry #{} attachments: {} new, {} total", EntryId, NiceBytes(NewAttachmentBytes), NiceBytes(AttachmentBytes));
@@ -3354,12 +3371,25 @@ ProjectStore::Rpc(HttpServerRequest&	 HttpReq,
 			HttpReq.WriteResponse(HttpResponseCode::InsufficientStorage);
 			return true;
 		}
+
 		std::span<const CbAttachment> Attachments = Package.GetAttachments();
-		for (const CbAttachment& Attachment : Attachments)
+		if (!Attachments.empty())
 		{
-			IoHash			 RawHash	= Attachment.GetHash();
-			CompressedBuffer Compressed = Attachment.AsCompressedBinary();
-			m_CidStore.AddChunk(Compressed.GetCompressed().Flatten().AsIoBuffer(), RawHash, CidStore::InsertMode::kCopyOnly);
+			std::vector<IoBuffer> WriteAttachmentBuffers;
+			std::vector<IoHash>	  WriteRawHashes;
+
+			WriteAttachmentBuffers.reserve(Attachments.size());
+			WriteRawHashes.reserve(Attachments.size());
+
+			for (const CbAttachment& Attachment : Attachments)
+			{
+				IoHash			 RawHash	= Attachment.GetHash();
+				CompressedBuffer Compressed = Attachment.AsCompressedBinary();
+				WriteAttachmentBuffers.push_back(Compressed.GetCompressed().Flatten().AsIoBuffer());
+				WriteRawHashes.push_back(RawHash);
+			}
+
+			m_CidStore.AddChunks(WriteAttachmentBuffers, WriteRawHashes, CidStore::InsertMode::kCopyOnly);
 		}
 		HttpReq.WriteResponse(HttpResponseCode::OK);
 		return true;
@@ -4703,9 +4733,8 @@ TEST_CASE("project.store.block")
 											return CompositeBuffer(SharedBuffer(Buffer));
 										}));
 	}
-	CompressedBuffer Block		 = GenerateBlock(std::move(Chunks));
-	IoBuffer		 BlockBuffer = Block.GetCompressed().Flatten().AsIoBuffer();
-	CHECK(IterateBlock(Block.DecodeRawHash(), std::move(BlockBuffer), [](CompressedBuffer&&, const IoHash&) {}));
+	CompressedBuffer Block = GenerateBlock(std::move(Chunks));
+	CHECK(IterateBlock(Block.Decompress(), [](CompressedBuffer&&, const IoHash&) {}));
 }
 
 #endif