WIPde/improved-projectstore-batch-requests

allow control of size for batch iteration
allow adding compositebuffers as attachments directly
add batch2 httpstore api to allow batching of CAS & Oid with range requests
allow responses with file handles from project store

Signed-off-by: Dan Engelbrecht <[email protected]>

6 files changed, 24 insertions, 13 deletions

diff --git a/src/zenstore/cas.cpp b/src/zenstore/cas.cpp
index bff221fc7..4252fc859 100644
--- a/src/zenstore/cas.cpp
+++ b/src/zenstore/cas.cpp
@@ -59,7 +59,8 @@ public:
 	virtual void					  FilterChunks(HashKeySet& InOutChunks) override;
 	virtual bool					  IterateChunks(std::span<IoHash>												  DecompressedIds,
 													const std::function<bool(size_t Index, const IoBuffer& Payload)>& AsyncCallback,
-													WorkerThreadPool*												  OptionalWorkerPool) override;
+													WorkerThreadPool*												  OptionalWorkerPool,
+													uint64_t														  LargeSizeLimit) override;
 	virtual void					  Flush() override;
 	virtual void					  ScrubStorage(ScrubContext& Ctx) override;
 	virtual CidStoreSize			  TotalSize() const override;
@@ -392,7 +393,8 @@ CasImpl::FilterChunks(HashKeySet& InOutChunks)
 bool
 CasImpl::IterateChunks(std::span<IoHash>												 DecompressedIds,
 					   const std::function<bool(size_t Index, const IoBuffer& Payload)>& AsyncCallback,
-					   WorkerThreadPool*												 OptionalWorkerPool)
+					   WorkerThreadPool*												 OptionalWorkerPool,
+					   uint64_t															 LargeSizeLimit)
 {
 	ZEN_TRACE_CPU("CAS::IterateChunks");
 	if (!m_SmallStrategy.IterateChunks(
@@ -402,7 +404,8 @@ CasImpl::IterateChunks(std::span<IoHash>												 DecompressedIds,
 				Chunk.SetContentType(ZenContentType::kCompressedBinary);
 				return AsyncCallback(Index, Payload);
 			},
-			OptionalWorkerPool))
+			OptionalWorkerPool,
+			LargeSizeLimit))
 	{
 		return false;
 	}
@@ -413,7 +416,8 @@ CasImpl::IterateChunks(std::span<IoHash>												 DecompressedIds,
 				Chunk.SetContentType(ZenContentType::kCompressedBinary);
 				return AsyncCallback(Index, Payload);
 			},
-			OptionalWorkerPool))
+			OptionalWorkerPool,
+			LargeSizeLimit))
 	{
 		return false;
 	}
diff --git a/src/zenstore/cas.h b/src/zenstore/cas.h
index bedbc6a9a..e279dd2cc 100644
--- a/src/zenstore/cas.h
+++ b/src/zenstore/cas.h
@@ -47,7 +47,8 @@ public:
 	virtual void					  FilterChunks(HashKeySet& InOutChunks)													   = 0;
 	virtual bool					  IterateChunks(std::span<IoHash>												  DecompressedIds,
 													const std::function<bool(size_t Index, const IoBuffer& Payload)>& AsyncCallback,
-													WorkerThreadPool*												  OptionalWorkerPool)									   = 0;
+													WorkerThreadPool*												  OptionalWorkerPool,
+													uint64_t														  LargeSizeLimit)												   = 0;
 	virtual void					  Flush()																				   = 0;
 	virtual void					  ScrubStorage(ScrubContext& Ctx)														   = 0;
 	virtual CidStoreSize			  TotalSize() const																		   = 0;
diff --git a/src/zenstore/cidstore.cpp b/src/zenstore/cidstore.cpp
index 71fd596f4..2ab769d04 100644
--- a/src/zenstore/cidstore.cpp
+++ b/src/zenstore/cidstore.cpp
@@ -119,9 +119,10 @@ struct CidStore::Impl
 
 	bool IterateChunks(std::span<IoHash>												 DecompressedIds,
 					   const std::function<bool(size_t Index, const IoBuffer& Payload)>& AsyncCallback,
-					   WorkerThreadPool*												 OptionalWorkerPool)
+					   WorkerThreadPool*												 OptionalWorkerPool,
+					   uint64_t															 LargeSizeLimit)
 	{
-		return m_CasStore.IterateChunks(DecompressedIds, AsyncCallback, OptionalWorkerPool);
+		return m_CasStore.IterateChunks(DecompressedIds, AsyncCallback, OptionalWorkerPool, LargeSizeLimit);
 	}
 
 	void Flush() { m_CasStore.Flush(); }
@@ -217,9 +218,10 @@ CidStore::ContainsChunk(const IoHash& DecompressedId)
 bool
 CidStore::IterateChunks(std::span<IoHash>												  DecompressedIds,
 						const std::function<bool(size_t Index, const IoBuffer& Payload)>& AsyncCallback,
-						WorkerThreadPool*												  OptionalWorkerPool)
+						WorkerThreadPool*												  OptionalWorkerPool,
+						uint64_t														  LargeSizeLimit)
 {
-	return m_Impl->IterateChunks(DecompressedIds, AsyncCallback, OptionalWorkerPool);
+	return m_Impl->IterateChunks(DecompressedIds, AsyncCallback, OptionalWorkerPool, LargeSizeLimit);
 }
 
 void
diff --git a/src/zenstore/compactcas.cpp b/src/zenstore/compactcas.cpp
index 7f1300177..f479a7173 100644
--- a/src/zenstore/compactcas.cpp
+++ b/src/zenstore/compactcas.cpp
@@ -305,7 +305,8 @@ CasContainerStrategy::FilterChunks(HashKeySet& InOutChunks)
 bool
 CasContainerStrategy::IterateChunks(std::span<IoHash>												  ChunkHashes,
 									const std::function<bool(size_t Index, const IoBuffer& Payload)>& AsyncCallback,
-									WorkerThreadPool*												  OptionalWorkerPool)
+									WorkerThreadPool*												  OptionalWorkerPool,
+									uint64_t														  LargeSizeLimit)
 {
 	std::vector<size_t>				FoundChunkIndexes;
 	std::vector<BlockStoreLocation> FoundChunkLocations;
@@ -332,7 +333,8 @@ CasContainerStrategy::IterateChunks(std::span<IoHash>												  ChunkHashes,
 			},
 			[&](size_t ChunkIndex, BlockStoreFile& File, uint64_t Offset, uint64_t Size) {
 				return AsyncCallback(FoundChunkIndexes[ChunkIndex], File.GetChunk(Offset, Size));
-			});
+			},
+			LargeSizeLimit);
 	};
 
 	Latch			 WorkLatch(1);
diff --git a/src/zenstore/compactcas.h b/src/zenstore/compactcas.h
index 44567e7a0..07e620086 100644
--- a/src/zenstore/compactcas.h
+++ b/src/zenstore/compactcas.h
@@ -58,7 +58,8 @@ struct CasContainerStrategy final : public GcStorage, public GcReferenceStore
 	void								FilterChunks(HashKeySet& InOutChunks);
 	bool								IterateChunks(std::span<IoHash>													ChunkHashes,
 													  const std::function<bool(size_t Index, const IoBuffer& Payload)>& AsyncCallback,
-													  WorkerThreadPool*													OptionalWorkerPool);
+													  WorkerThreadPool*													OptionalWorkerPool,
+													  uint64_t															LargeSizeLimit);
 	void								Initialize(const std::filesystem::path& RootDirectory,
 												   const std::string_view		ContainerBaseName,
 												   uint32_t						MaxBlockSize,
diff --git a/src/zenstore/include/zenstore/cidstore.h b/src/zenstore/include/zenstore/cidstore.h
index d95fa7cd4..b3d00fec0 100644
--- a/src/zenstore/include/zenstore/cidstore.h
+++ b/src/zenstore/include/zenstore/cidstore.h
@@ -82,7 +82,8 @@ public:
 	virtual IoBuffer		  FindChunkByCid(const IoHash& DecompressedId) override;
 	bool					  IterateChunks(std::span<IoHash>												  DecompressedIds,
 											const std::function<bool(size_t Index, const IoBuffer& Payload)>& AsyncCallback,
-											WorkerThreadPool*												  OptionalWorkerPool);
+											WorkerThreadPool*												  OptionalWorkerPool,
+											uint64_t														  LargeSizeLimit);
 	bool					  ContainsChunk(const IoHash& DecompressedId);
 	void					  FilterChunks(HashKeySet& InOutChunks);
 	void					  Flush();