From 68b3382ef7e0f7795b9a601aae73adc2f8ef9873 Mon Sep 17 00:00:00 2001
From: Dan Engelbrecht <dan@engelbrecht.com>
Date: Wed, 29 Nov 2023 09:14:57 -0500
Subject: global thread worker pools (#577)

- Improvement: Use two global worker thread pools instead of ad-hoc creation of worker pools
---
 src/zenserver/cache/cachedisklayer.cpp            | 15 +++++----------
 src/zenserver/projectstore/remoteprojectstore.cpp | 16 ++++------------
 src/zenserver/zenserver.cpp                       |  2 ++
 3 files changed, 11 insertions(+), 22 deletions(-)

(limited to 'src/zenserver')

diff --git a/src/zenserver/cache/cachedisklayer.cpp b/src/zenserver/cache/cachedisklayer.cpp
index c8bc3871a..9117b8820 100644
--- a/src/zenserver/cache/cachedisklayer.cpp
+++ b/src/zenserver/cache/cachedisklayer.cpp
@@ -14,6 +14,7 @@
 #include <zencore/trace.h>
 #include <zencore/workthreadpool.h>
 #include <zencore/xxhash.h>
+#include <zenutil/workerpools.h>
 
 #include <future>
 
@@ -3709,11 +3710,8 @@ ZenCacheDiskLayer::DiscoverBuckets()
 
 	RwLock SyncLock;
 
-	const size_t MaxHwTreadUse		   = std::thread::hardware_concurrency();
-	const int	 WorkerThreadPoolCount = gsl::narrow<int>(Min(MaxHwTreadUse, FoundBucketDirectories.size()));
-
-	WorkerThreadPool Pool(WorkerThreadPoolCount, "CacheBucket::OpenOrCreate");
-	Latch			 WorkLatch(1);
+	WorkerThreadPool& Pool = GetLargeWorkerPool();
+	Latch			  WorkLatch(1);
 	for (auto& BucketPath : FoundBucketDirectories)
 	{
 		WorkLatch.AddCount(1);
@@ -3825,11 +3823,8 @@ ZenCacheDiskLayer::Flush()
 		}
 	}
 	{
-		const size_t MaxHwTreadUse		   = Max((std::thread::hardware_concurrency() / 4u), 1u);
-		const int	 WorkerThreadPoolCount = gsl::narrow<int>(Min(MaxHwTreadUse, Buckets.size()));
-
-		WorkerThreadPool Pool(WorkerThreadPoolCount, "CacheBucket::Flush");
-		Latch			 WorkLatch(1);
+		WorkerThreadPool& Pool = GetSmallWorkerPool();
+		Latch			  WorkLatch(1);
 		for (auto& Bucket : Buckets)
 		{
 			WorkLatch.AddCount(1);
diff --git a/src/zenserver/projectstore/remoteprojectstore.cpp b/src/zenserver/projectstore/remoteprojectstore.cpp
index d5d229e42..826c8ff51 100644
--- a/src/zenserver/projectstore/remoteprojectstore.cpp
+++ b/src/zenserver/projectstore/remoteprojectstore.cpp
@@ -13,6 +13,7 @@
 #include <zencore/timer.h>
 #include <zencore/workthreadpool.h>
 #include <zenstore/cidstore.h>
+#include <zenutil/workerpools.h>
 
 #include <unordered_map>
 
@@ -802,10 +803,7 @@ BuildContainer(CidStore&																	ChunkStore,
 			   const std::function<void(const std::unordered_set<IoHash, IoHash::Hasher>)>& OnBlockChunks,
 			   tsl::robin_map<IoHash, IoBuffer, IoHash::Hasher>*							OutOptionalTempAttachments)
 {
-	// We are creating a worker thread pool here since we are uploading a lot of attachments in one go and we dont want to keep a
-	// WorkerThreadPool alive
-	size_t			 WorkerCount = Min(std::thread::hardware_concurrency(), 16u);
-	WorkerThreadPool WorkerPool(gsl::narrow<int>(WorkerCount));
+	WorkerThreadPool& WorkerPool = GetSmallWorkerPool();
 
 	AsyncRemoteResult RemoteResult;
 	CbObject		  ContainerObject = BuildContainer(ChunkStore,
@@ -1153,10 +1151,7 @@ SaveOplog(CidStore&				 ChunkStore,
 
 	Stopwatch Timer;
 
-	// We are creating a worker thread pool here since we are uploading a lot of attachments in one go
-	// Doing upload is a rare and transient occation so we don't want to keep a WorkerThreadPool alive.
-	size_t			 WorkerCount = Min(std::thread::hardware_concurrency(), 16u);
-	WorkerThreadPool WorkerPool(gsl::narrow<int>(WorkerCount), "oplog_upload"sv);
+	WorkerThreadPool& WorkerPool = GetSmallWorkerPool();
 
 	std::filesystem::path AttachmentTempPath;
 	if (UseTempBlocks)
@@ -1528,10 +1523,7 @@ LoadOplog(CidStore&			   ChunkStore,
 
 	Stopwatch Timer;
 
-	// We are creating a worker thread pool here since we are download a lot of attachments in one go and we dont want to keep a
-	// WorkerThreadPool alive
-	size_t			 WorkerCount = Min(std::thread::hardware_concurrency(), 16u);
-	WorkerThreadPool WorkerPool(gsl::narrow<int>(WorkerCount));
+	WorkerThreadPool& WorkerPool = GetSmallWorkerPool();
 
 	std::unordered_set<IoHash, IoHash::Hasher> Attachments;
 	std::vector<std::vector<IoHash>>		   ChunksInBlocks;
diff --git a/src/zenserver/zenserver.cpp b/src/zenserver/zenserver.cpp
index 841f19295..2430267c1 100644
--- a/src/zenserver/zenserver.cpp
+++ b/src/zenserver/zenserver.cpp
@@ -24,6 +24,7 @@
 #include <zenstore/cidstore.h>
 #include <zenstore/scrubcontext.h>
 #include <zenutil/basicfile.h>
+#include <zenutil/workerpools.h>
 #include <zenutil/zenserverprocess.h>
 
 #if ZEN_PLATFORM_WINDOWS
@@ -659,6 +660,7 @@ ZenServer::Cleanup()
 		m_AuthMgr.reset();
 		m_Http = {};
 		m_JobQueue.reset();
+		ShutdownWorkerPools();
 	}
 	catch (std::exception& Ex)
 	{
-- 
cgit v1.2.3