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

2 files changed, 73 insertions, 22 deletions

diff --git a/src/zenserver/storage/projectstore/httpprojectstore.cpp b/src/zenserver/storage/projectstore/httpprojectstore.cpp
index 11231a203..4e947f221 100644
--- a/src/zenserver/storage/projectstore/httpprojectstore.cpp
+++ b/src/zenserver/storage/projectstore/httpprojectstore.cpp
@@ -19,6 +19,7 @@
 #include <zenremotestore/projectstore/jupiterremoteprojectstore.h>
 #include <zenremotestore/projectstore/remoteprojectstore.h>
 #include <zenremotestore/projectstore/zenremoteprojectstore.h>
+#include <zenremotestore/transferthreadworkers.h>
 #include <zenstore/oplogreferencedset.h>
 #include <zenstore/projectstore.h>
 #include <zenstore/zenstore.h>
@@ -250,6 +251,7 @@ namespace {
 											  AuthMgr&					   AuthManager,
 											  size_t					   MaxBlockSize,
 											  size_t					   MaxChunkEmbedSize,
+											  size_t					   MaximumInMemoryDownloadSize,
 											  const std::filesystem::path& TempFilePath)
 	{
 		ZEN_MEMSCOPE(GetProjectHttpTag());
@@ -485,7 +487,8 @@ namespace {
 				ForceDisableTempBlocks,
 				AssumeHttp2,
 				PopulateCache,
-				MetaData};
+				MetaData,
+				MaximumInMemoryDownloadSize};
 			RemoteStore = CreateJupiterBuildsRemoteStore(Log(),
 														 Options,
 														 TempFilePath,
@@ -515,10 +518,17 @@ namespace {
 									  : fmt::format("{}: {}", Result.Reason, Result.Text)};
 	}
 
+	static uint64_t GetMaxMemoryBufferSize(size_t MaxBlockSize, bool BoostWorkerMemory)
+	{
+		return BoostWorkerMemory ? (MaxBlockSize + 16u * 1024u) : 1024u * 1024u;
+	}
+
 }  // namespace
 
 //////////////////////////////////////////////////////////////////////////
 
+//////////////////////////////////////////////////////////////////////////
+
 HttpProjectService::HttpProjectService(CidStore&		  Store,
 									   ProjectStore*	  Projects,
 									   HttpStatusService& StatusService,
@@ -2639,9 +2649,16 @@ HttpProjectService::HandleRpcRequest(HttpRouterRequest& Req)
 				bool		 Force			   = Params["force"sv].AsBool(false);
 				bool		 IgnoreMissingAttachments = Params["ignoremissingattachments"sv].AsBool(false);
 				bool		 CleanOplog				  = Params["clean"].AsBool(false);
+				bool		 BoostWorkerCount		  = Params["boostworkercount"].AsBool(false);
+				bool		 BoostWorkerMemory		  = Params["boostworkermemory"sv].AsBool(false);
 
-				CreateRemoteStoreResult RemoteStoreResult =
-					CreateRemoteStore(Log(), Params, m_AuthMgr, MaxBlockSize, MaxChunkEmbedSize, Oplog->TempPath());
+				CreateRemoteStoreResult RemoteStoreResult = CreateRemoteStore(Log(),
+																			  Params,
+																			  m_AuthMgr,
+																			  MaxBlockSize,
+																			  MaxChunkEmbedSize,
+																			  GetMaxMemoryBufferSize(MaxBlockSize, BoostWorkerMemory),
+																			  Oplog->TempPath());
 
 				if (RemoteStoreResult.Store == nullptr)
 				{
@@ -2658,14 +2675,19 @@ HttpProjectService::HandleRpcRequest(HttpRouterRequest& Req)
 					 Oplog,
 					 Force,
 					 IgnoreMissingAttachments,
-					 CleanOplog](JobContext& Context) {
+					 CleanOplog,
+					 BoostWorkerCount](JobContext& Context) {
 						Context.ReportMessage(fmt::format("Loading oplog '{}/{}' from {}",
 														  Oplog->GetOuterProjectIdentifier(),
 														  Oplog->OplogId(),
 														  ActualRemoteStore->GetInfo().Description));
 
-						WorkerThreadPool& WorkerPool		= GetLargeWorkerPool(EWorkloadType::Background);
-						WorkerThreadPool& NetworkWorkerPool = GetMediumWorkerPool(EWorkloadType::Background);
+						Ref<TransferThreadWorkers> Workers = GetThreadWorkers(BoostWorkerCount, /*SingleThreaded*/ false);
+
+						WorkerThreadPool& WorkerPool		= Workers->GetIOWorkerPool();
+						WorkerThreadPool& NetworkWorkerPool = Workers->GetNetworkPool();
+
+						Context.ReportMessage(fmt::format("{}", Workers->GetWorkersInfo()));
 
 						RemoteProjectStore::Result Result	= LoadOplog(m_CidStore,
 																		*ActualRemoteStore,
@@ -2697,9 +2719,16 @@ HttpProjectService::HandleRpcRequest(HttpRouterRequest& Req)
 				bool		 Force				= Params["force"sv].AsBool(false);
 				bool		 IgnoreMissingAttachments = Params["ignoremissingattachments"sv].AsBool(false);
 				bool		 EmbedLooseFile			  = Params["embedloosefiles"sv].AsBool(false);
+				bool		 BoostWorkerCount		  = Params["boostworkercount"].AsBool(false);
+				bool		 BoostWorkerMemory		  = Params["boostworkermemory"sv].AsBool(false);
 
-				CreateRemoteStoreResult RemoteStoreResult =
-					CreateRemoteStore(Log(), Params, m_AuthMgr, MaxBlockSize, MaxChunkEmbedSize, Oplog->TempPath());
+				CreateRemoteStoreResult RemoteStoreResult = CreateRemoteStore(Log(),
+																			  Params,
+																			  m_AuthMgr,
+																			  MaxBlockSize,
+																			  MaxChunkEmbedSize,
+																			  GetMaxMemoryBufferSize(MaxBlockSize, BoostWorkerMemory),
+																			  Oplog->TempPath());
 
 				if (RemoteStoreResult.Store == nullptr)
 				{
@@ -2720,7 +2749,8 @@ HttpProjectService::HandleRpcRequest(HttpRouterRequest& Req)
 					 ChunkFileSizeLimit,
 					 EmbedLooseFile,
 					 Force,
-					 IgnoreMissingAttachments](JobContext& Context) {
+					 IgnoreMissingAttachments,
+					 BoostWorkerCount](JobContext& Context) {
 						Context.ReportMessage(fmt::format("Saving oplog '{}/{}' to {}, maxblocksize {}, maxchunkembedsize {}",
 														  Project->Identifier,
 														  Oplog->OplogId(),
@@ -2728,8 +2758,12 @@ HttpProjectService::HandleRpcRequest(HttpRouterRequest& Req)
 														  NiceBytes(MaxBlockSize),
 														  NiceBytes(MaxChunkEmbedSize)));
 
-						WorkerThreadPool& WorkerPool		= GetLargeWorkerPool(EWorkloadType::Background);
-						WorkerThreadPool& NetworkWorkerPool = GetMediumWorkerPool(EWorkloadType::Background);
+						Ref<TransferThreadWorkers> Workers = GetThreadWorkers(BoostWorkerCount, /*SingleThreaded*/ false);
+
+						Context.ReportMessage(fmt::format("{}", Workers->GetWorkersInfo()));
+
+						WorkerThreadPool& WorkerPool		= Workers->GetIOWorkerPool();
+						WorkerThreadPool& NetworkWorkerPool = Workers->GetNetworkPool();
 
 						RemoteProjectStore::Result Result	= SaveOplog(m_CidStore,
 																		*ActualRemoteStore,
@@ -3322,4 +3356,16 @@ HttpProjectService::HandleOplogOpDetailsRequest(HttpRouterRequest& Req)
 	}
 }
 
+Ref<TransferThreadWorkers>
+HttpProjectService::GetThreadWorkers(bool BoostWorkers, bool SingleThreaded)
+{
+	RwLock::ExclusiveLockScope _(m_ThreadWorkersLock);
+	if (m_ThreadWorkers && m_ThreadWorkers->IsBoostWorkers() == BoostWorkers && m_ThreadWorkers->IsSingleThreaded() == SingleThreaded)
+	{
+		return m_ThreadWorkers;
+	}
+	m_ThreadWorkers = new TransferThreadWorkers(BoostWorkers, SingleThreaded);
+	return m_ThreadWorkers;
+}
+
 }  // namespace zen
diff --git a/src/zenserver/storage/projectstore/httpprojectstore.h b/src/zenserver/storage/projectstore/httpprojectstore.h
index f6fe63614..b742102a5 100644
--- a/src/zenserver/storage/projectstore/httpprojectstore.h
+++ b/src/zenserver/storage/projectstore/httpprojectstore.h
@@ -14,6 +14,7 @@ class AuthMgr;
 class JobQueue;
 class OpenProcessCache;
 class ProjectStore;
+class TransferThreadWorkers;
 
 //////////////////////////////////////////////////////////////////////////
 //
@@ -95,17 +96,21 @@ private:
 
 	inline LoggerRef Log() { return m_Log; }
 
-	LoggerRef				 m_Log;
-	CidStore&				 m_CidStore;
-	HttpRequestRouter		 m_Router;
-	Ref<ProjectStore>		 m_ProjectStore;
-	HttpStatusService&		 m_StatusService;
-	HttpStatsService&		 m_StatsService;
-	AuthMgr&				 m_AuthMgr;
-	OpenProcessCache&		 m_OpenProcessCache;
-	JobQueue&				 m_JobQueue;
-	ProjectStats			 m_ProjectStats;
-	metrics::OperationTiming m_HttpRequests;
+	LoggerRef				   m_Log;
+	CidStore&				   m_CidStore;
+	HttpRequestRouter		   m_Router;
+	Ref<ProjectStore>		   m_ProjectStore;
+	HttpStatusService&		   m_StatusService;
+	HttpStatsService&		   m_StatsService;
+	AuthMgr&				   m_AuthMgr;
+	OpenProcessCache&		   m_OpenProcessCache;
+	JobQueue&				   m_JobQueue;
+	ProjectStats			   m_ProjectStats;
+	metrics::OperationTiming   m_HttpRequests;
+	RwLock					   m_ThreadWorkersLock;
+	Ref<TransferThreadWorkers> m_ThreadWorkers;
+
+	Ref<TransferThreadWorkers> GetThreadWorkers(bool BoostWorkers, bool SingleThreaded);
 };
 
 }  // namespace zen