tweak worker pools for builds (#595)

* update logic for number of network and disk threads for `zen builds` commands

2 files changed, 120 insertions, 17 deletions

diff --git a/src/zen/cmds/builds_cmd.cpp b/src/zen/cmds/builds_cmd.cpp
index d220eefb2..1f748400a 100644
--- a/src/zen/cmds/builds_cmd.cpp
+++ b/src/zen/cmds/builds_cmd.cpp
@@ -222,21 +222,6 @@ namespace {
 		}
 	}
 
-	const double DefaultLatency		  = 0;	// .0010;
-	const double DefaultDelayPerKBSec = 0;	// 0.00005;
-
-	const bool SingleThreaded	  = false;
-	bool	   BoostWorkerThreads = false;
-	bool	   UseSparseFiles	  = false;
-
-	WorkerThreadPool& GetIOWorkerPool()
-	{
-		return SingleThreaded		? GetSyncWorkerPool()
-			   : BoostWorkerThreads ? GetLargeWorkerPool(EWorkloadType::Burst)
-									: GetMediumWorkerPool(EWorkloadType::Burst);
-	}
-	WorkerThreadPool& GetNetworkPool() { return SingleThreaded ? GetSyncWorkerPool() : GetSmallWorkerPool(EWorkloadType::Burst); }
-
 	const std::string	  ZenFolderName = ".zen";
 	std::filesystem::path ZenStateFilePath(const std::filesystem::path& ZenFolderPath) { return ZenFolderPath / "current_state.cbo"; }
 	// std::filesystem::path ZenStateFileJsonPath(const std::filesystem::path& ZenFolderPath) { return ZenFolderPath / "current_state.json";
@@ -259,11 +244,106 @@ namespace {
 	const std::vector<std::string> DefaultExcludeFolders({UnsyncFolderName, ZenFolderName, UGSFolderName, LegacyZenTempFolderName});
 	const std::vector<std::string> DefaultExcludeExtensions({});
 
+	const double DefaultLatency		  = 0;	// .0010;
+	const double DefaultDelayPerKBSec = 0;	// 0.00005;
+
+	const bool SingleThreaded = false;
+	bool	   UseSparseFiles = false;
+
 	static bool				 IsVerbose		= false;
 	static bool				 IsQuiet		= false;
 	static ProgressBar::Mode ProgressMode	= ProgressBar::Mode::Pretty;
 	static bool				 AllowFileClone = true;
 
+	static std::unique_ptr<WorkerThreadPool> NetworkPoolInstance;
+	static std::unique_ptr<WorkerThreadPool> IOPoolInstance;
+
+	WorkerThreadPool* NetworkPool = nullptr;
+	WorkerThreadPool* IOPool	  = nullptr;
+
+	void InitializeWorkerPools(bool BoostWorkers)
+	{
+		if (SingleThreaded)
+		{
+			NetworkPool = &GetSyncWorkerPool();
+			IOPool		= &GetSyncWorkerPool();
+			ZEN_CONSOLE("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);
+
+				NetworkPoolInstance = std::make_unique<WorkerThreadPool>(NetworkThreadCount, "net");
+				IOPoolInstance		= std::make_unique<WorkerThreadPool>(IOThreadCount, "disk");
+
+				if (!IsQuiet)
+				{
+					ZEN_CONSOLE("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);
+
+				NetworkPoolInstance = std::make_unique<WorkerThreadPool>(NetworkThreadCount, "net");
+				IOPoolInstance		= std::make_unique<WorkerThreadPool>(IOThreadCount, "disk");
+
+				if (!IsQuiet)
+				{
+					ZEN_CONSOLE("Using {} network threads and {} worker/IO threads", NetworkThreadCount, IOThreadCount);
+				}
+			}
+			NetworkPool = NetworkPoolInstance.get();
+			IOPool		= IOPoolInstance.get();
+		}
+	}
+
+	WorkerThreadPool& GetIOWorkerPool()
+	{
+		ZEN_ASSERT(IOPool);
+		return *IOPool;
+	}
+	WorkerThreadPool& GetNetworkPool()
+	{
+		ZEN_ASSERT(NetworkPool);
+		return *NetworkPool;
+	}
+
 	uint32_t GetUpdateDelayMS(ProgressBar::Mode InMode)
 	{
 		switch (InMode)
@@ -3568,8 +3648,7 @@ BuildsCommand::Run(const ZenCliOptions& GlobalOptions, int argc, char** argv)
 		return {};
 	};
 
-	BoostWorkerThreads = m_BoostWorkerThreads;
-	UseSparseFiles	   = m_UseSparseFiles;
+	UseSparseFiles = m_UseSparseFiles;
 
 	if (SubOption == &m_ListNamespacesOptions)
 	{
@@ -3800,6 +3879,8 @@ BuildsCommand::Run(const ZenCliOptions& GlobalOptions, int argc, char** argv)
 			LogExecutableVersionAndPid();
 		}
 
+		InitializeWorkerPools(m_BoostWorkerThreads);
+
 		ZenState InstanceState;
 
 		ParsePath();
@@ -3920,6 +4001,8 @@ BuildsCommand::Run(const ZenCliOptions& GlobalOptions, int argc, char** argv)
 			LogExecutableVersionAndPid();
 		}
 
+		InitializeWorkerPools(m_BoostWorkerThreads);
+
 		ZenState InstanceState;
 
 		ParsePath();
@@ -4036,6 +4119,13 @@ BuildsCommand::Run(const ZenCliOptions& GlobalOptions, int argc, char** argv)
 
 	if (SubOption == &m_DiffOptions)
 	{
+		if (!IsQuiet)
+		{
+			LogExecutableVersionAndPid();
+		}
+
+		InitializeWorkerPools(m_BoostWorkerThreads);
+
 		ParsePath();
 		ParseDiffPath();
 
@@ -4053,6 +4143,8 @@ BuildsCommand::Run(const ZenCliOptions& GlobalOptions, int argc, char** argv)
 			LogExecutableVersionAndPid();
 		}
 
+		InitializeWorkerPools(m_BoostWorkerThreads);
+
 		BuildStorage::Statistics	  StorageStats;
 		BuildStorageCache::Statistics StorageCacheStats;
 
@@ -4105,6 +4197,8 @@ BuildsCommand::Run(const ZenCliOptions& GlobalOptions, int argc, char** argv)
 			LogExecutableVersionAndPid();
 		}
 
+		InitializeWorkerPools(m_BoostWorkerThreads);
+
 		ZenState InstanceState;
 
 		BuildStorage::Statistics	  StorageStats;
@@ -4152,6 +4246,8 @@ BuildsCommand::Run(const ZenCliOptions& GlobalOptions, int argc, char** argv)
 
 	if (SubOption == &m_MultiTestDownloadOptions)
 	{
+		InitializeWorkerPools(m_BoostWorkerThreads);
+
 		m_SystemRootDir = (GetRunningExecutablePath().parent_path() / ".tmpzensystem").make_preferred();
 		CreateDirectories(m_SystemRootDir);
 		CleanDirectory(m_SystemRootDir, {});
@@ -4256,6 +4352,8 @@ BuildsCommand::Run(const ZenCliOptions& GlobalOptions, int argc, char** argv)
 
 	if (SubOption == &m_TestOptions)
 	{
+		InitializeWorkerPools(m_BoostWorkerThreads);
+
 		m_SystemRootDir = (GetRunningExecutablePath().parent_path() / ".tmpzensystem").make_preferred();
 		CreateDirectories(m_SystemRootDir);
 		CleanDirectory(m_SystemRootDir, {});
diff --git a/src/zen/zen.cpp b/src/zen/zen.cpp
index e60e5451b..ad4bb9013 100644
--- a/src/zen/zen.cpp
+++ b/src/zen/zen.cpp
@@ -907,6 +907,10 @@ main(int argc, char** argv)
 	Options.add_options()("help", "Show command line help");
 	Options.add_options()("c, command", "Sub command", cxxopts::value<std::string>(SubCommand));
 
+	int CoreLimit = 0;
+
+	Options.add_options()("corelimit", "Limit concurrency", cxxopts::value(CoreLimit));
+
 #if ZEN_WITH_TRACE
 	// We only have this in options for command line help purposes - we parse these argument separately earlier using
 	// GetTraceOptionsFromCommandline()
@@ -992,6 +996,7 @@ main(int argc, char** argv)
 			return (int)ReturnCode::kSuccess;
 		}
 
+		LimitHardwareConcurrency(CoreLimit);
 #if ZEN_USE_SENTRY
 
 		{