add batching of CacheStore requests for GetCacheValues/GetCacheChunks (#90)

* cache file size of block on open
* add ability to control size limit for small chunk callback when iterating block
* Add batch fetch of cache values in the GetCacheValues request

1 files changed, 40 insertions, 14 deletions

diff --git a/src/zenstore/blockstore.cpp b/src/zenstore/blockstore.cpp
index aef7b348b..6e289409c 100644
--- a/src/zenstore/blockstore.cpp
+++ b/src/zenstore/blockstore.cpp
@@ -64,7 +64,8 @@ BlockStoreFile::Open()
 		return true;
 	});
 	void* FileHandle = m_File.Handle();
-	m_IoBuffer		 = IoBuffer(IoBuffer::File, FileHandle, 0, m_File.FileSize(), /*IsWholeFile*/ true);
+	m_CachedFileSize = m_File.FileSize();
+	m_IoBuffer		 = IoBuffer(IoBuffer::File, FileHandle, 0, m_CachedFileSize, /*IsWholeFile*/ true);
 }
 
 void
@@ -100,7 +101,7 @@ BlockStoreFile::Create(uint64_t InitialSize)
 uint64_t
 BlockStoreFile::FileSize()
 {
-	return m_File.FileSize();
+	return m_CachedFileSize == 0 ? m_File.FileSize() : m_CachedFileSize;
 }
 
 void
@@ -156,7 +157,8 @@ BlockStoreFile::IsOpen() const
 	return !!m_IoBuffer;
 }
 
-constexpr uint64_t IterateSmallChunkWindowSize = 2 * 1024 * 1024;
+constexpr uint64_t DefaultIterateSmallChunkWindowSize = 2 * 1024 * 1024;
+constexpr uint64_t IterateSmallChunkMaxGapSize		  = 4 * 1024;
 
 BlockStore::BlockStore()
 {
@@ -1041,20 +1043,33 @@ bool
 BlockStore::IterateBlock(std::span<const BlockStoreLocation>   ChunkLocations,
 						 std::span<const size_t>			   InChunkIndexes,
 						 const IterateChunksSmallSizeCallback& SmallSizeCallback,
-						 const IterateChunksLargeSizeCallback& LargeSizeCallback)
+						 const IterateChunksLargeSizeCallback& LargeSizeCallback,
+						 uint64_t							   LargeSizeLimit)
 {
 	if (InChunkIndexes.empty())
 	{
 		return true;
 	}
 
+	uint64_t IterateSmallChunkWindowSize = Max(DefaultIterateSmallChunkWindowSize, LargeSizeLimit);
+	if (LargeSizeLimit == 0u)
+	{
+		LargeSizeLimit = IterateSmallChunkWindowSize;
+	}
+	else
+	{
+		IterateSmallChunkWindowSize =
+			Min((LargeSizeLimit + IterateSmallChunkMaxGapSize) * ChunkLocations.size(), IterateSmallChunkWindowSize);
+	}
+
 	uint32_t			BlockIndex = ChunkLocations[InChunkIndexes[0]].BlockIndex;
 	std::vector<size_t> ChunkIndexes(InChunkIndexes.begin(), InChunkIndexes.end());
 	std::sort(ChunkIndexes.begin(), ChunkIndexes.end(), [&](size_t IndexA, size_t IndexB) -> bool {
 		return ChunkLocations[IndexA].Offset < ChunkLocations[IndexB].Offset;
 	});
 
-	auto GetNextRange = [&ChunkLocations](std::span<const size_t> ChunkIndexes, size_t StartIndexOffset) -> size_t {
+	auto GetNextRange = [LargeSizeLimit, IterateSmallChunkWindowSize, &ChunkLocations](std::span<const size_t> ChunkIndexes,
+																					   size_t				   StartIndexOffset) -> size_t {
 		size_t					  ChunkCount	= 0;
 		size_t					  StartIndex	= ChunkIndexes[StartIndexOffset];
 		const BlockStoreLocation& StartLocation = ChunkLocations[StartIndex];
@@ -1065,7 +1080,11 @@ BlockStore::IterateBlock(std::span<const BlockStoreLocation>   ChunkLocations,
 			size_t					  NextIndex = ChunkIndexes[StartIndexOffset + ChunkCount];
 			const BlockStoreLocation& Location	= ChunkLocations[NextIndex];
 			ZEN_ASSERT(Location.BlockIndex == StartLocation.BlockIndex);
-			if (Location.Offset >= (LastEnd + (4u * 1024u)))
+			if (Location.Size > LargeSizeLimit)
+			{
+				break;
+			}
+			if (Location.Offset >= (LastEnd + IterateSmallChunkMaxGapSize))
 			{
 				break;
 			}
@@ -1097,8 +1116,9 @@ BlockStore::IterateBlock(std::span<const BlockStoreLocation>   ChunkLocations,
 	}
 	else
 	{
-		const Ref<BlockStoreFile>& BlockFile = FindBlockIt->second;
+		Ref<BlockStoreFile> BlockFile = FindBlockIt->second;
 		ZEN_ASSERT(BlockFile);
+		InsertLock.ReleaseNow();
 
 		IoBuffer ReadBuffer{IterateSmallChunkWindowSize};
 		void*	 BufferBase = ReadBuffer.MutableData();
@@ -1129,11 +1149,17 @@ BlockStore::IterateBlock(std::span<const BlockStoreLocation>   ChunkLocations,
 									  BlockIndex,
 									  BlockSize);
 
-						SmallSizeCallback(NextChunkIndex, nullptr, 0);
+						if (!SmallSizeCallback(NextChunkIndex, nullptr, 0))
+						{
+							return false;
+						}
 						continue;
 					}
 					void* BufferPtr = &((char*)BufferBase)[ChunkLocation.Offset - FirstLocation.Offset];
-					SmallSizeCallback(NextChunkIndex, BufferPtr, ChunkLocation.Size);
+					if (!SmallSizeCallback(NextChunkIndex, BufferPtr, ChunkLocation.Size))
+					{
+						return false;
+					}
 				}
 				LocationIndexOffset += RangeCount;
 				continue;
@@ -1167,7 +1193,7 @@ BlockStore::IterateChunks(const std::span<const BlockStoreLocation>& ChunkLocati
 
 	Stopwatch Timer;
 	auto	  _ = MakeGuard([&]() {
-		 ZEN_INFO("Iterated {} chunks from '{}' in {}", ChunkLocations.size(), m_BlocksBasePath, NiceTimeSpanMs(Timer.GetElapsedTimeMs()));
+		 ZEN_DEBUG("Iterated {} chunks from '{}' in {}", ChunkLocations.size(), m_BlocksBasePath, NiceTimeSpanMs(Timer.GetElapsedTimeMs()));
 	 });
 
 	ZEN_LOG_SCOPE("iterating chunks from '{}'", m_BlocksBasePath);
@@ -1838,7 +1864,7 @@ TEST_CASE("blockstore.iterate.chunks")
 	auto					 RootDirectory = TempDir.Path();
 
 	BlockStore Store;
-	Store.Initialize(RootDirectory / "store", IterateSmallChunkWindowSize * 2, 1024);
+	Store.Initialize(RootDirectory / "store", DefaultIterateSmallChunkWindowSize * 2, 1024);
 	IoBuffer BadChunk = Store.TryGetChunk({.BlockIndex = 0, .Offset = 0, .Size = 512});
 	CHECK(!BadChunk);
 
@@ -1849,13 +1875,13 @@ TEST_CASE("blockstore.iterate.chunks")
 	BlockStoreLocation SecondChunkLocation = WriteStringAsChunk(Store, SecondChunkData, 4);
 	Store.Flush(/*ForceNewBlock*/ false);
 
-	std::string		   VeryLargeChunk(IterateSmallChunkWindowSize * 2, 'L');
+	std::string		   VeryLargeChunk(DefaultIterateSmallChunkWindowSize * 2, 'L');
 	BlockStoreLocation VeryLargeChunkLocation = WriteStringAsChunk(Store, VeryLargeChunk, 4);
 
 	BlockStoreLocation BadLocationZeroSize	 = {.BlockIndex = 0, .Offset = 0, .Size = 0};
 	BlockStoreLocation BadLocationOutOfRange = {.BlockIndex = 0,
-												.Offset		= IterateSmallChunkWindowSize,
-												.Size		= IterateSmallChunkWindowSize * 2};
+												.Offset		= DefaultIterateSmallChunkWindowSize,
+												.Size		= DefaultIterateSmallChunkWindowSize * 2};
 	BlockStoreLocation BadBlockIndex		 = {.BlockIndex = 0xfffff, .Offset = 1024, .Size = 1024};
 
 	WorkerThreadPool WorkerPool(4);