various optimizations (#704)

- Improvement: Validate chunk hashes when dechunking files in oplog import
- Improvement: Use stream decompression when dechunking files
- Improvement: When assembling blocks for oplog export, make sure we keep under/at block size limit
- Improvement: Make cancelling of oplog import more responsive
- Improvement: Use decompress to composite to avoid allocating a new memory buffer for uncompressed chunks during oplog import
- Improvement: Reduce memory buffer size and allocate it on demand when writing multiple chunks to block store
- Improvement: Reduce lock contention when fetching/checking existence of chunks in block store

3 files changed, 25 insertions, 11 deletions

diff --git a/src/zenstore/blockstore.cpp b/src/zenstore/blockstore.cpp
index f97c98e08..0542d1171 100644
--- a/src/zenstore/blockstore.cpp
+++ b/src/zenstore/blockstore.cpp
@@ -762,7 +762,7 @@ BlockStore::WriteChunks(std::span<const IoBuffer> Datas, uint32_t Alignment, con
 		LargestSize = Max(LargestSize, Size);
 	}
 
-	const uint64_t		 MinSize	= Max(LargestSize, 8u * 1024u * 1024u);
+	const uint64_t		 MinSize	= Max(LargestSize, 512u * 1024u);
 	const uint64_t		 BufferSize = Min(TotalSize, MinSize);
 	std::vector<uint8_t> Buffer(BufferSize);
 
@@ -815,7 +815,12 @@ BlockStore::WriteChunks(std::span<const IoBuffer> Datas, uint32_t Alignment, con
 
 		auto _ = MakeGuard([this, WriteBlockIndex]() { RemoveActiveWriteBlock(WriteBlockIndex); });
 
+		if (Count > 1)
 		{
+			if (Buffer.empty())
+			{
+				Buffer.resize(BufferSize);
+			}
 			MutableMemoryView WriteBuffer(Buffer.data(), RangeSize);
 			for (size_t Index = 0; Index < Count; Index++)
 			{
@@ -824,9 +829,14 @@ BlockStore::WriteChunks(std::span<const IoBuffer> Datas, uint32_t Alignment, con
 				WriteBuffer.MidInline(RoundUp(SourceBuffer.GetSize(), Alignment));
 			}
 			WriteBlock->Write(Buffer.data(), RangeSize, AlignedInsertOffset);
+			m_TotalSize.fetch_add(RangeSize, std::memory_order::relaxed);
+		}
+		else
+		{
+			MemoryView SourceBuffer = Datas[Offset];
+			WriteBlock->Write(SourceBuffer.GetData(), SourceBuffer.GetSize(), AlignedInsertOffset);
+			m_TotalSize.fetch_add(SourceBuffer.GetSize(), std::memory_order::relaxed);
 		}
-
-		m_TotalSize.fetch_add(RangeSize, std::memory_order::relaxed);
 
 		uint32_t						ChunkOffset = AlignedInsertOffset;
 		std::vector<BlockStoreLocation> Locations(Count);
@@ -845,11 +855,11 @@ BlockStore::WriteChunks(std::span<const IoBuffer> Datas, uint32_t Alignment, con
 bool
 BlockStore::HasChunk(const BlockStoreLocation& Location) const
 {
-	ZEN_TRACE_CPU("BlockStore::TryGetChunk");
+	ZEN_TRACE_CPU("BlockStore::HasChunk");
 	RwLock::SharedLockScope InsertLock(m_InsertLock);
 	if (auto BlockIt = m_ChunkBlocks.find(Location.BlockIndex); BlockIt != m_ChunkBlocks.end())
 	{
-		if (const Ref<BlockStoreFile>& Block = BlockIt->second; Block)
+		if (Ref<BlockStoreFile> Block = BlockIt->second; Block)
 		{
 			InsertLock.ReleaseNow();
 
@@ -878,8 +888,10 @@ BlockStore::TryGetChunk(const BlockStoreLocation& Location) const
 	RwLock::SharedLockScope InsertLock(m_InsertLock);
 	if (auto BlockIt = m_ChunkBlocks.find(Location.BlockIndex); BlockIt != m_ChunkBlocks.end())
 	{
-		if (const Ref<BlockStoreFile>& Block = BlockIt->second; Block)
+		if (Ref<BlockStoreFile> Block = BlockIt->second; Block)
 		{
+			InsertLock.ReleaseNow();
+
 			IoBuffer Chunk = Block->GetChunk(Location.Offset, Location.Size);
 			if (Chunk.GetSize() == Location.Size)
 			{
diff --git a/src/zenstore/compactcas.cpp b/src/zenstore/compactcas.cpp
index a5de5c448..37a8c36b8 100644
--- a/src/zenstore/compactcas.cpp
+++ b/src/zenstore/compactcas.cpp
@@ -301,13 +301,14 @@ CasContainerStrategy::FindChunk(const IoHash& ChunkHash)
 {
 	ZEN_TRACE_CPU("CasContainer::FindChunk");
 
-	RwLock::SharedLockScope _(m_LocationMapLock);
+	RwLock::SharedLockScope Lock(m_LocationMapLock);
 	auto					KeyIt = m_LocationMap.find(ChunkHash);
 	if (KeyIt == m_LocationMap.end())
 	{
 		return IoBuffer();
 	}
-	const BlockStoreLocation& Location = m_Locations[KeyIt->second].Get(m_PayloadAlignment);
+	const BlockStoreLocation Location = m_Locations[KeyIt->second].Get(m_PayloadAlignment);
+	Lock.ReleaseNow();
 
 	IoBuffer Chunk = m_BlockStore.TryGetChunk(Location);
 	return Chunk;
@@ -316,10 +317,11 @@ CasContainerStrategy::FindChunk(const IoHash& ChunkHash)
 bool
 CasContainerStrategy::HaveChunk(const IoHash& ChunkHash)
 {
-	RwLock::SharedLockScope _(m_LocationMapLock);
+	RwLock::SharedLockScope Lock(m_LocationMapLock);
 	if (auto KeyIt = m_LocationMap.find(ChunkHash); KeyIt != m_LocationMap.end())
 	{
-		const BlockStoreLocation& Location = m_Locations[KeyIt->second].Get(m_PayloadAlignment);
+		const BlockStoreLocation Location = m_Locations[KeyIt->second].Get(m_PayloadAlignment);
+		Lock.ReleaseNow();
 		return m_BlockStore.HasChunk(Location);
 	}
 	return false;
diff --git a/src/zenstore/projectstore.cpp b/src/zenstore/projectstore.cpp
index f1001f665..c5b27c1ea 100644
--- a/src/zenstore/projectstore.cpp
+++ b/src/zenstore/projectstore.cpp
@@ -3917,7 +3917,7 @@ ProjectStore::Project::Scrub(ScrubContext& Ctx)
 {
 	ZEN_MEMSCOPE(GetProjectstoreTag());
 
-	ZEN_INFO("scrubbing '{}'", ProjectRootDir);
+	ZEN_INFO("scrubbing '{}'", m_OplogStoragePath);
 
 	// Scrubbing needs to check all existing oplogs
 	std::vector<std::string> OpLogs = ScanForOplogs();