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, 135 insertions, 10 deletions

diff --git a/src/zenstore/cache/structuredcachestore.cpp b/src/zenstore/cache/structuredcachestore.cpp
index 9eded2a50..d9c2d3e59 100644
--- a/src/zenstore/cache/structuredcachestore.cpp
+++ b/src/zenstore/cache/structuredcachestore.cpp
@@ -137,21 +137,21 @@ ZenCacheNamespace::~ZenCacheNamespace()
 	m_Gc.RemoveGcContributor(this);
 }
 
-struct ZenCacheNamespace::BatchHandle
+struct ZenCacheNamespace::PutBatchHandle
 {
-	ZenCacheDiskLayer::BatchHandle* DiskLayerHandle = nullptr;
+	ZenCacheDiskLayer::PutBatchHandle* DiskLayerHandle = nullptr;
 };
 
-ZenCacheNamespace::BatchHandle*
+ZenCacheNamespace::PutBatchHandle*
 ZenCacheNamespace::BeginPutBatch(std::vector<bool>& OutResult)
 {
-	ZenCacheNamespace::BatchHandle* Handle = new ZenCacheNamespace::BatchHandle;
-	Handle->DiskLayerHandle				   = m_DiskLayer.BeginPutBatch(OutResult);
+	ZenCacheNamespace::PutBatchHandle* Handle = new ZenCacheNamespace::PutBatchHandle;
+	Handle->DiskLayerHandle					  = m_DiskLayer.BeginPutBatch(OutResult);
 	return Handle;
 }
 
 void
-ZenCacheNamespace::EndPutBatch(BatchHandle* Batch) noexcept
+ZenCacheNamespace::EndPutBatch(PutBatchHandle* Batch) noexcept
 {
 	try
 	{
@@ -165,6 +165,50 @@ ZenCacheNamespace::EndPutBatch(BatchHandle* Batch) noexcept
 	}
 }
 
+struct ZenCacheNamespace::GetBatchHandle
+{
+	GetBatchHandle(std::vector<ZenCacheValue>& OutResult) : Results(OutResult) {}
+	std::vector<ZenCacheValue>&		   Results;
+	ZenCacheDiskLayer::GetBatchHandle* DiskLayerHandle = nullptr;
+};
+
+ZenCacheNamespace::GetBatchHandle*
+ZenCacheNamespace::BeginGetBatch(std::vector<ZenCacheValue>& OutResult)
+{
+	ZenCacheNamespace::GetBatchHandle* Handle = new ZenCacheNamespace::GetBatchHandle(OutResult);
+	Handle->DiskLayerHandle					  = m_DiskLayer.BeginGetBatch(OutResult);
+	return Handle;
+}
+
+void
+ZenCacheNamespace::EndGetBatch(GetBatchHandle* Batch) noexcept
+{
+	try
+	{
+		ZEN_ASSERT(Batch);
+		m_DiskLayer.EndGetBatch(Batch->DiskLayerHandle);
+
+		metrics::RequestStats::Scope StatsScope(m_GetOps, 0);
+		for (const ZenCacheValue& Result : Batch->Results)
+		{
+			if (Result.Value)
+			{
+				m_HitCount++;
+				StatsScope.SetBytes(Result.Value.Size());
+			}
+			else
+			{
+				m_MissCount++;
+			}
+		}
+		delete Batch;
+	}
+	catch (std::exception& Ex)
+	{
+		ZEN_ERROR("Exception in cache namespace layer when ending batch put operation: '{}'", Ex.what());
+	}
+}
+
 bool
 ZenCacheNamespace::Get(std::string_view InBucket, const IoHash& HashKey, ZenCacheValue& OutValue)
 {
@@ -173,7 +217,6 @@ ZenCacheNamespace::Get(std::string_view InBucket, const IoHash& HashKey, ZenCach
 	metrics::RequestStats::Scope StatsScope(m_GetOps, 0);
 
 	bool Ok = m_DiskLayer.Get(InBucket, HashKey, OutValue);
-
 	if (Ok)
 	{
 		ZEN_ASSERT(OutValue.Value.Size());
@@ -188,11 +231,22 @@ ZenCacheNamespace::Get(std::string_view InBucket, const IoHash& HashKey, ZenCach
 }
 
 void
+ZenCacheNamespace::Get(std::string_view InBucket, const IoHash& HashKey, GetBatchHandle& BatchHandle)
+{
+	ZEN_TRACE_CPU("Z$::Namespace::GetBatched");
+
+	metrics::RequestStats::Scope StatsScope(m_GetOps, 0);
+
+	m_DiskLayer.Get(InBucket, HashKey, *BatchHandle.DiskLayerHandle);
+	return;
+}
+
+void
 ZenCacheNamespace::Put(std::string_view		InBucket,
 					   const IoHash&		HashKey,
 					   const ZenCacheValue& Value,
 					   std::span<IoHash>	References,
-					   BatchHandle*			OptionalBatchHandle)
+					   PutBatchHandle*		OptionalBatchHandle)
 {
 	ZEN_TRACE_CPU("Z$::Namespace::Put");
 
@@ -202,7 +256,8 @@ ZenCacheNamespace::Put(std::string_view		InBucket,
 
 	ZEN_ASSERT(Value.Value.Size());
 
-	m_DiskLayer.Put(InBucket, HashKey, Value, References, OptionalBatchHandle ? OptionalBatchHandle->DiskLayerHandle : nullptr);
+	ZenCacheDiskLayer::PutBatchHandle* DiskLayerBatchHandle = OptionalBatchHandle ? OptionalBatchHandle->DiskLayerHandle : nullptr;
+	m_DiskLayer.Put(InBucket, HashKey, Value, References, DiskLayerBatchHandle);
 	m_WriteCount++;
 }
 
@@ -500,6 +555,43 @@ ZenCacheStore::PutBatch::~PutBatch()
 	}
 }
 
+ZenCacheStore::GetBatch::GetBatch(ZenCacheStore& CacheStore, std::string_view InNamespace, std::vector<ZenCacheValue>& OutResult)
+: m_CacheStore(CacheStore)
+, Results(OutResult)
+{
+	if (m_Store = m_CacheStore.GetNamespace(InNamespace); m_Store)
+	{
+		m_NamespaceBatchHandle = m_Store->BeginGetBatch(OutResult);
+	}
+}
+
+ZenCacheStore::GetBatch::~GetBatch()
+{
+	try
+	{
+		if (m_Store)
+		{
+			ZEN_ASSERT(m_NamespaceBatchHandle);
+			m_Store->EndGetBatch(m_NamespaceBatchHandle);
+
+			metrics::RequestStats::Scope OpScope(m_CacheStore.m_GetOps, 0);
+			for (const ZenCacheValue& Result : Results)
+			{
+				if (Result.Value)
+				{
+					m_CacheStore.m_HitCount++;
+					OpScope.SetBytes(Result.Value.GetSize());
+				}
+				m_CacheStore.m_MissCount++;
+			}
+		}
+	}
+	catch (std::exception& Ex)
+	{
+		ZEN_ERROR("Exception in cache store when ending batch get operation: '{}'", Ex.what());
+	}
+}
+
 bool
 ZenCacheStore::Get(const CacheRequestContext& Context,
 				   std::string_view			  Namespace,
@@ -562,6 +654,39 @@ ZenCacheStore::Get(const CacheRequestContext& Context,
 }
 
 void
+ZenCacheStore::Get(const CacheRequestContext& Context,
+				   std::string_view			  Namespace,
+				   std::string_view			  Bucket,
+				   const IoHash&			  HashKey,
+				   GetBatch&				  BatchHandle)
+{
+	// Ad hoc rejection of known bad usage patterns for DDC bucket names
+
+	if (IsKnownBadBucketName(Bucket))
+	{
+		m_RejectedReadCount++;
+		return;
+	}
+	ZEN_TRACE_CPU("Z$::Get");
+
+	metrics::RequestStats::Scope OpScope(m_GetOps, 0);
+
+	if (ZenCacheNamespace* Store = GetNamespace(Namespace); Store)
+	{
+		Store->Get(Bucket, HashKey, *BatchHandle.m_NamespaceBatchHandle);
+		return;
+	}
+
+	ZEN_WARN("request for unknown namespace '{}' in ZenCacheStore::Get [{}], bucket '{}', key '{}'",
+			 Context,
+			 Namespace,
+			 Bucket,
+			 HashKey.ToHexString());
+
+	m_MissCount++;
+}
+
+void
 ZenCacheStore::Put(const CacheRequestContext& Context,
 				   std::string_view			  Namespace,
 				   std::string_view			  Bucket,
@@ -603,7 +728,7 @@ ZenCacheStore::Put(const CacheRequestContext& Context,
 
 	if (ZenCacheNamespace* Store = GetNamespace(Namespace); Store)
 	{
-		ZenCacheNamespace::BatchHandle* BatchHandle = OptionalBatchHandle ? OptionalBatchHandle->m_NamespaceBatchHandle : nullptr;
+		ZenCacheNamespace::PutBatchHandle* BatchHandle = OptionalBatchHandle ? OptionalBatchHandle->m_NamespaceBatchHandle : nullptr;
 		Store->Put(Bucket, HashKey, Value, References, BatchHandle);
 		m_WriteCount++;
 		return;