add boost-worker oplog import export options (#693)

- Feature: `zen oplog-export`, `zen oplog-import` and `zen oplog-download` now has options to boost workers
  - `--boost-worker-count` - Increase the number of worker threads - may cause computer to be less responsive
  - `--boost-worker-memory` - Increase the limit where we write downloaded data to temporary storage to conserve space - may cause computer to be less responsive due to high memory usage
  - `--boost-workers` - Enables both 'boost-worker-count' and 'boost-worker-memory' - may cause computer to be less responsive
 - Improvement: Refactored boost options for `zen builds` operations `upload`, `download`, `diff`, `prime-cache`, `fetch-blob` and `validate-part`
  - `--boost-worker-count` - Increase the number of worker threads - may cause computer to be less responsive
  - `--boost-worker-memory` - Increase the limit where we write downloaded data to temporary storage to conserve space - may cause computer to be less responsive due to high memory usage
  - `--boost-workers` - Enables both 'boost-worker-count' and 'boost-worker-memory' - may cause computer to be less responsive

5 files changed, 154 insertions, 16 deletions

diff --git a/src/zenremotestore/include/zenremotestore/builds/buildstorageoperations.h b/src/zenremotestore/include/zenremotestore/builds/buildstorageoperations.h
index 223c668cd..eca654223 100644
--- a/src/zenremotestore/include/zenremotestore/builds/buildstorageoperations.h
+++ b/src/zenremotestore/include/zenremotestore/builds/buildstorageoperations.h
@@ -477,14 +477,18 @@ struct GenerateBlocksStatistics
 	}
 };
 
+static constexpr size_t DefaultMaxChunkBlockSize	  = 64u * 1024u * 1024u;
+static constexpr size_t DefaultMaxChunksPerChunkBlock = 4u * 1000u;
+static constexpr size_t DefaultMaxChunkBlockEmbedSize = 3u * 512u * 1024u;
+
 class BuildsOperationUploadFolder
 {
 public:
 	struct ChunksBlockParameters
 	{
-		size_t MaxBlockSize		 = 64u * 1024u * 1024u;
-		size_t MaxChunksPerBlock = 4u * 1000u;
-		size_t MaxChunkEmbedSize = 3u * 512u * 1024u;
+		size_t MaxBlockSize		 = DefaultMaxChunkBlockSize;
+		size_t MaxChunksPerBlock = DefaultMaxChunksPerChunkBlock;
+		size_t MaxChunkEmbedSize = DefaultMaxChunkBlockEmbedSize;
 	};
 
 	struct Options
diff --git a/src/zenremotestore/include/zenremotestore/projectstore/buildsremoteprojectstore.h b/src/zenremotestore/include/zenremotestore/projectstore/buildsremoteprojectstore.h
index bb26f55a0..e8b7c15c0 100644
--- a/src/zenremotestore/include/zenremotestore/projectstore/buildsremoteprojectstore.h
+++ b/src/zenremotestore/include/zenremotestore/projectstore/buildsremoteprojectstore.h
@@ -25,6 +25,7 @@ struct BuildsRemoteStoreOptions : RemoteStoreOptions
 	bool				  AssumeHttp2			 = false;
 	bool				  PopulateCache			 = true;
 	IoBuffer			  MetaData;
+	size_t				  MaximumInMemoryDownloadSize = 1024u * 1024u;
 };
 
 std::shared_ptr<RemoteProjectStore> CreateJupiterBuildsRemoteStore(LoggerRef					   InLog,
diff --git a/src/zenremotestore/include/zenremotestore/transferthreadworkers.h b/src/zenremotestore/include/zenremotestore/transferthreadworkers.h
new file mode 100644
index 000000000..a7faacfd5
--- /dev/null
+++ b/src/zenremotestore/include/zenremotestore/transferthreadworkers.h
@@ -0,0 +1,49 @@
+// Copyright Epic Games, Inc. All Rights Reserved.
+
+#pragma once
+
+#include <zenbase/refcount.h>
+#include <zencore/timer.h>
+#include <zencore/zencore.h>
+
+#include <memory>
+#include <string>
+
+namespace zen {
+
+class WorkerThreadPool;
+
+class TransferThreadWorkers : public RefCounted
+{
+public:
+	TransferThreadWorkers(bool BoostWorkers, bool SingleThreaded);
+
+	WorkerThreadPool& GetIOWorkerPool()
+	{
+		ZEN_ASSERT(m_IOPool);
+		return *m_IOPool;
+	}
+	WorkerThreadPool& GetNetworkPool()
+	{
+		ZEN_ASSERT(m_NetworkPool);
+		return *m_NetworkPool;
+	}
+
+	const std::string& GetWorkersInfo() const { return WorkersInfo; }
+
+	bool IsBoostWorkers() const { return BoostWorkers; }
+	bool IsSingleThreaded() const { return SingleThreaded; }
+
+private:
+	WorkerThreadPool* m_NetworkPool = nullptr;
+	WorkerThreadPool* m_IOPool		= nullptr;
+
+	std::unique_ptr<WorkerThreadPool> m_NetworkPoolInstance;
+	std::unique_ptr<WorkerThreadPool> m_IOPoolInstance;
+
+	std::string WorkersInfo;
+	const bool	BoostWorkers;
+	const bool	SingleThreaded;
+};
+
+}  // namespace zen
diff --git a/src/zenremotestore/projectstore/buildsremoteprojectstore.cpp b/src/zenremotestore/projectstore/buildsremoteprojectstore.cpp
index bd793b745..a8e883dde 100644
--- a/src/zenremotestore/projectstore/buildsremoteprojectstore.cpp
+++ b/src/zenremotestore/projectstore/buildsremoteprojectstore.cpp
@@ -690,24 +690,27 @@ CreateJupiterBuildsRemoteStore(LoggerRef					   InLog,
 			ResolveBuildStorage(*Output, ClientSettings, Host, OverrideUrl, ZenHost, ZenCacheResolveMode::Discovery, /*Verbose*/ false);
 	}
 
-	HttpClientSettings ClientSettings{.LogCategory		   = "httpbuildsclient",
-									  .ConnectTimeout	   = std::chrono::milliseconds(3000),
-									  .Timeout			   = std::chrono::milliseconds(1800000),
-									  .AccessTokenProvider = std::move(TokenProvider),
-									  .AssumeHttp2		   = ResolveRes.HostAssumeHttp2,
-									  .AllowResume		   = true,
-									  .RetryCount		   = 4};
+	HttpClientSettings ClientSettings{.LogCategory				   = "httpbuildsclient",
+									  .ConnectTimeout			   = std::chrono::milliseconds(3000),
+									  .Timeout					   = std::chrono::milliseconds(1800000),
+									  .AccessTokenProvider		   = std::move(TokenProvider),
+									  .AssumeHttp2				   = ResolveRes.HostAssumeHttp2,
+									  .AllowResume				   = true,
+									  .RetryCount				   = 4,
+									  .MaximumInMemoryDownloadSize = Options.MaximumInMemoryDownloadSize};
 
 	std::unique_ptr<HttpClientSettings> CacheClientSettings;
 
 	if (!ResolveRes.CacheUrl.empty())
 	{
-		CacheClientSettings = std::make_unique<HttpClientSettings>(HttpClientSettings{.LogCategory	  = "httpcacheclient",
-																					  .ConnectTimeout = std::chrono::milliseconds{3000},
-																					  .Timeout		  = std::chrono::milliseconds{30000},
-																					  .AssumeHttp2	  = ResolveRes.CacheAssumeHttp2,
-																					  .AllowResume	  = true,
-																					  .RetryCount	  = 0});
+		CacheClientSettings =
+			std::make_unique<HttpClientSettings>(HttpClientSettings{.LogCategory				 = "httpcacheclient",
+																	.ConnectTimeout				 = std::chrono::milliseconds{3000},
+																	.Timeout					 = std::chrono::milliseconds{30000},
+																	.AssumeHttp2				 = ResolveRes.CacheAssumeHttp2,
+																	.AllowResume				 = true,
+																	.RetryCount					 = 0,
+																	.MaximumInMemoryDownloadSize = Options.MaximumInMemoryDownloadSize});
 	}
 
 	std::shared_ptr<RemoteProjectStore> RemoteStore = std::make_shared<BuildsRemoteStore>(InLog,
diff --git a/src/zenremotestore/transferthreadworkers.cpp b/src/zenremotestore/transferthreadworkers.cpp
new file mode 100644
index 000000000..4f863027d
--- /dev/null
+++ b/src/zenremotestore/transferthreadworkers.cpp
@@ -0,0 +1,81 @@
+// Copyright Epic Games, Inc. All Rights Reserved.
+
+#include <zenremotestore/transferthreadworkers.h>
+
+#include <zencore/intmath.h>
+#include <zencore/logging.h>
+#include <zencore/thread.h>
+#include <zenutil/workerpools.h>
+
+//////////////////////////////////////////////////////////////////////////
+
+namespace zen {
+
+TransferThreadWorkers::TransferThreadWorkers(bool InBoostWorkers, bool InSingleThreaded)
+: BoostWorkers(InBoostWorkers)
+, SingleThreaded(InSingleThreaded)
+{
+	if (SingleThreaded)
+	{
+		m_NetworkPool = &GetSyncWorkerPool();
+		m_IOPool	  = &GetSyncWorkerPool();
+		WorkersInfo	  = "Not using thread workers for network or I/O, expect no progress reporting";
+	}
+	else
+	{
+		unsigned int Cores = GetHardwareConcurrency();
+		if (BoostWorkers)
+		{
+			// Cores    Net     Disk
+			// 2        4       3
+			// 4        6       3
+			// 8        11      6
+			// 16       16      12
+			// 32       16      24
+			// 64       16      32
+			// n        16      32
+
+			unsigned int NetworkThreadCount = Cores + (Cores / 4u) + Max(1u, (Cores / 8u));
+			NetworkThreadCount				= Min(NetworkThreadCount, 16u);
+			NetworkThreadCount				= Max(NetworkThreadCount, 4u);
+
+			unsigned int IOThreadCount = Cores - (Cores / 4u);
+			IOThreadCount			   = Min(IOThreadCount, 32u);
+			IOThreadCount			   = Max(IOThreadCount, 3u);
+
+			m_NetworkPoolInstance = std::make_unique<WorkerThreadPool>(NetworkThreadCount, "net");
+			m_IOPoolInstance	  = std::make_unique<WorkerThreadPool>(IOThreadCount, "disk");
+
+			WorkersInfo =
+				fmt::format("Worker boost enabled, using {} network threads and {} worker/IO threads", NetworkThreadCount, IOThreadCount);
+		}
+		else
+		{
+			// Cores    Net     Disk
+			// 2        2       2
+			// 4        3       2
+			// 8        5       4
+			// 16       10      9
+			// 32       12      18
+			// 64       12      20
+			// n        12      20
+
+			unsigned int NetworkThreadCount = (Cores / 2u) + Max(1u, (Cores / 8u));
+			NetworkThreadCount				= Min(NetworkThreadCount, 12u);
+			NetworkThreadCount				= Max(NetworkThreadCount, 2u);
+
+			unsigned int IOThreadCount = Cores / 2u + Cores / 16u;
+			IOThreadCount			   = Min(IOThreadCount, 20u);
+			IOThreadCount			   = Max(IOThreadCount, 2u);
+
+			m_NetworkPoolInstance = std::make_unique<WorkerThreadPool>(NetworkThreadCount, "net");
+			m_IOPoolInstance	  = std::make_unique<WorkerThreadPool>(IOThreadCount, "disk");
+
+			WorkersInfo = fmt::format("Using {} network threads and {} worker/IO threads", NetworkThreadCount, IOThreadCount);
+		}
+		m_NetworkPool = m_NetworkPoolInstance.get();
+		m_IOPool	  = m_IOPoolInstance.get();
+	}
+}
+
+}  // namespace zen