separate worker pools into burst/background to avoid background jobs blocking client requests (#134)

2 files changed, 57 insertions, 73 deletions

diff --git a/src/zenutil/include/zenutil/workerpools.h b/src/zenutil/include/zenutil/workerpools.h
index 78a8e5c5e..9683ad720 100644
--- a/src/zenutil/include/zenutil/workerpools.h
+++ b/src/zenutil/include/zenutil/workerpools.h
@@ -6,14 +6,20 @@
 
 namespace zen {
 
+enum class EWorkloadType
+{
+	Burst,		// Used when you want to respond quickly, such as requests from a client
+	Background	// Used for background jobs such as GC/Scrub/oplog import-export
+};
+
 // Worker pool with std::thread::hardware_concurrency() worker threads, but at least one thread
-WorkerThreadPool& GetLargeWorkerPool();
+WorkerThreadPool& GetLargeWorkerPool(EWorkloadType WorkloadType);
 
 // Worker pool with std::thread::hardware_concurrency() / 4 worker threads, but at least one thread
-WorkerThreadPool& GetMediumWorkerPool();
+WorkerThreadPool& GetMediumWorkerPool(EWorkloadType WorkloadType);
 
 // Worker pool with std::thread::hardware_concurrency() / 8 worker threads, but at least one thread
-WorkerThreadPool& GetSmallWorkerPool();
+WorkerThreadPool& GetSmallWorkerPool(EWorkloadType WorkloadType);
 
 // Special worker pool that does not use worker thread but issues all scheduled work on the calling thread
 // This is useful for debugging when multiple async thread can make stepping in debugger complicated
diff --git a/src/zenutil/workerpools.cpp b/src/zenutil/workerpools.cpp
index 144ef6817..e3165e838 100644
--- a/src/zenutil/workerpools.cpp
+++ b/src/zenutil/workerpools.cpp
@@ -19,90 +19,65 @@ namespace {
 
 	RwLock PoolLock;
 
-	std::unique_ptr<WorkerThreadPool> LargeWorkerPool;
-	std::unique_ptr<WorkerThreadPool> MediumWorkerPool;
-	std::unique_ptr<WorkerThreadPool> SmallWorkerPool;
-	std::unique_ptr<WorkerThreadPool> SyncWorkerPool;
-}  // namespace
+	struct WorkerPool
+	{
+		std::unique_ptr<WorkerThreadPool> Pool;
+		const int						  TreadCount;
+		const std::string_view			  Name;
+	};
 
-WorkerThreadPool&
-GetLargeWorkerPool()
-{
+	WorkerPool BurstLargeWorkerPool		 = {.TreadCount = LargeWorkerThreadPoolTreadCount, .Name = "LargeThreadPool(burst)"};
+	WorkerPool BackgroundLargeWorkerPool = {.TreadCount = LargeWorkerThreadPoolTreadCount, .Name = "LargeThreadPool(bkg)"};
+
+	WorkerPool BurstMediumWorkerPool	  = {.TreadCount = MediumWorkerThreadPoolTreadCount, .Name = "MediumThreadPool(burst)"};
+	WorkerPool BackgroundMediumWorkerPool = {.TreadCount = MediumWorkerThreadPoolTreadCount, .Name = "MediumThreadPool(bkg)"};
+
+	WorkerPool BurstSmallWorkerPool		 = {.TreadCount = SmallWorkerThreadPoolTreadCount, .Name = "SmallThreadPool(burst)"};
+	WorkerPool BackgroundSmallWorkerPool = {.TreadCount = SmallWorkerThreadPoolTreadCount, .Name = "SmallThreadPool(bkg)"};
+
+	WorkerPool SyncWorkerPool = {.TreadCount = 0, .Name = "SyncThreadPool"};
+
+	WorkerThreadPool& EnsurePoolPtr(WorkerPool& Pool)
 	{
-		RwLock::SharedLockScope _(PoolLock);
-		if (LargeWorkerPool)
 		{
-			return *LargeWorkerPool;
+			RwLock::SharedLockScope _(PoolLock);
+			if (Pool.Pool)
+			{
+				return *Pool.Pool;
+			}
 		}
+		RwLock::ExclusiveLockScope _(PoolLock);
+		ZEN_ASSERT(!IsShutDown);
+		if (!Pool.Pool)
+		{
+			Pool.Pool.reset(new WorkerThreadPool(Pool.TreadCount, Pool.Name));
+		}
+		return *Pool.Pool;
 	}
-	RwLock::ExclusiveLockScope _(PoolLock);
-	ZEN_ASSERT(!IsShutDown);
-	if (LargeWorkerPool)
-	{
-		return *LargeWorkerPool;
-	}
-	LargeWorkerPool.reset(new WorkerThreadPool(LargeWorkerThreadPoolTreadCount, "LargeThreadPool"));
-	return *LargeWorkerPool;
+}  // namespace
+
+WorkerThreadPool&
+GetLargeWorkerPool(EWorkloadType WorkloadType)
+{
+	return EnsurePoolPtr(WorkloadType == EWorkloadType::Burst ? BurstLargeWorkerPool : BackgroundLargeWorkerPool);
 }
 
 WorkerThreadPool&
-GetMediumWorkerPool()
+GetMediumWorkerPool(EWorkloadType WorkloadType)
 {
-	{
-		RwLock::SharedLockScope _(PoolLock);
-		if (MediumWorkerPool)
-		{
-			return *MediumWorkerPool;
-		}
-	}
-	RwLock::ExclusiveLockScope _(PoolLock);
-	ZEN_ASSERT(!IsShutDown);
-	if (MediumWorkerPool)
-	{
-		return *MediumWorkerPool;
-	}
-	MediumWorkerPool.reset(new WorkerThreadPool(MediumWorkerThreadPoolTreadCount, "MediumThreadPool"));
-	return *MediumWorkerPool;
+	return EnsurePoolPtr(WorkloadType == EWorkloadType::Burst ? BurstMediumWorkerPool : BackgroundMediumWorkerPool);
 }
 
 WorkerThreadPool&
-GetSmallWorkerPool()
+GetSmallWorkerPool(EWorkloadType WorkloadType)
 {
-	{
-		RwLock::SharedLockScope _(PoolLock);
-		if (SmallWorkerPool)
-		{
-			return *SmallWorkerPool;
-		}
-	}
-	RwLock::ExclusiveLockScope _(PoolLock);
-	ZEN_ASSERT(!IsShutDown);
-	if (SmallWorkerPool)
-	{
-		return *SmallWorkerPool;
-	}
-	SmallWorkerPool.reset(new WorkerThreadPool(SmallWorkerThreadPoolTreadCount, "SmallThreadPool"));
-	return *SmallWorkerPool;
+	return EnsurePoolPtr(WorkloadType == EWorkloadType::Burst ? BurstSmallWorkerPool : BackgroundSmallWorkerPool);
 }
 
 WorkerThreadPool&
 GetSyncWorkerPool()
 {
-	{
-		RwLock::SharedLockScope _(PoolLock);
-		if (SyncWorkerPool)
-		{
-			return *SyncWorkerPool;
-		}
-	}
-	RwLock::ExclusiveLockScope _(PoolLock);
-	ZEN_ASSERT(!IsShutDown);
-	if (SyncWorkerPool)
-	{
-		return *SyncWorkerPool;
-	}
-	SyncWorkerPool.reset(new WorkerThreadPool(0, "SyncThreadPool"));
-	return *SyncWorkerPool;
+	return EnsurePoolPtr(SyncWorkerPool);
 }
 
 void
@@ -110,9 +85,12 @@ ShutdownWorkerPools()
 {
 	RwLock::ExclusiveLockScope _(PoolLock);
 	IsShutDown = true;
-	LargeWorkerPool.reset();
-	MediumWorkerPool.reset();
-	SmallWorkerPool.reset();
-	SyncWorkerPool.reset();
+	BurstLargeWorkerPool.Pool.reset();
+	BackgroundLargeWorkerPool.Pool.reset();
+	BurstMediumWorkerPool.Pool.reset();
+	BackgroundMediumWorkerPool.Pool.reset();
+	BurstSmallWorkerPool.Pool.reset();
+	BackgroundSmallWorkerPool.Pool.reset();
+	SyncWorkerPool.Pool.reset();
 }
 }  // namespace zen