From 203d3f03f0d0ef51f414b5344462bde0a8fcaf1b Mon Sep 17 00:00:00 2001
From: Dan Engelbrecht <dan.engelbrecht@epicgames.com>
Date: Thu, 22 Aug 2024 16:03:01 +0200
Subject: separate worker pools into burst/background to avoid background jobs
 blocking client requests (#134)

---
 src/zenutil/workerpools.cpp | 118 ++++++++++++++++++--------------------------
 1 file changed, 48 insertions(+), 70 deletions(-)

(limited to 'src/zenutil/workerpools.cpp')

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
-- 
cgit v1.2.3