add EMode to WorkerTheadPool to avoid thread starvation (#492)

- Improvement: Add a new mode to worker thread pools to avoid starvation of workers which could cause long stalls due to other work begin queued up. UE-305498

1 files changed, 50 insertions, 45 deletions

diff --git a/src/zenserver-test/zenserver-test.cpp b/src/zenserver-test/zenserver-test.cpp
index 77ed87cb1..79e5db554 100644
--- a/src/zenserver-test/zenserver-test.cpp
+++ b/src/zenserver-test/zenserver-test.cpp
@@ -2142,20 +2142,23 @@ TEST_CASE("zcache.failing.upstream")
 	for (size_t I = 0; I < ThreadCount * KeyMultiplier; I++)
 	{
 		size_t Iteration = I;
-		Pool.ScheduleWork([&] {
-			std::vector<CacheKey> NewKeys = PutCacheRecords(LocalUri, "ue4.ddc"sv, "mastodon"sv, RecordsPerRequest, I * RecordsPerRequest);
-			if (NewKeys.size() != RecordsPerRequest)
-			{
-				ZEN_DEBUG("PutCacheRecords iteration {} failed", Iteration);
+		Pool.ScheduleWork(
+			[&] {
+				std::vector<CacheKey> NewKeys =
+					PutCacheRecords(LocalUri, "ue4.ddc"sv, "mastodon"sv, RecordsPerRequest, I * RecordsPerRequest);
+				if (NewKeys.size() != RecordsPerRequest)
+				{
+					ZEN_DEBUG("PutCacheRecords iteration {} failed", Iteration);
+					Completed.fetch_add(1);
+					return;
+				}
+				{
+					RwLock::ExclusiveLockScope _(KeysLock);
+					Keys[Iteration].swap(NewKeys);
+				}
 				Completed.fetch_add(1);
-				return;
-			}
-			{
-				RwLock::ExclusiveLockScope _(KeysLock);
-				Keys[Iteration].swap(NewKeys);
-			}
-			Completed.fetch_add(1);
-		});
+			},
+			WorkerThreadPool::EMode::EnableBacklog);
 	}
 	bool UseUpstream1 = false;
 	while (Completed < ThreadCount * KeyMultiplier)
@@ -2206,48 +2209,50 @@ TEST_CASE("zcache.failing.upstream")
 			Completed.fetch_add(1);
 			continue;
 		}
-		Pool.ScheduleWork([&] {
-			GetCacheRecordResult Result = GetCacheRecords(LocalUri, "ue4.ddc"sv, LocalKeys, Policy);
+		Pool.ScheduleWork(
+			[&] {
+				GetCacheRecordResult Result = GetCacheRecords(LocalUri, "ue4.ddc"sv, LocalKeys, Policy);
 
-			if (!Result.Success)
-			{
-				ZEN_DEBUG("GetCacheRecords iteration {} failed", Iteration);
-				Completed.fetch_add(1);
-				return;
-			}
-
-			if (Result.Result.Results.size() != LocalKeys.size())
-			{
-				ZEN_DEBUG("GetCacheRecords iteration {} empty records", Iteration);
-				Completed.fetch_add(1);
-				return;
-			}
-			for (size_t Index = 0; const std::optional<cacherequests::GetCacheRecordResult>& Record : Result.Result.Results)
-			{
-				const CacheKey& ExpectedKey = LocalKeys[Index++];
-				if (!Record)
-				{
-					continue;
-				}
-				if (Record->Key != ExpectedKey)
+				if (!Result.Success)
 				{
-					continue;
+					ZEN_DEBUG("GetCacheRecords iteration {} failed", Iteration);
+					Completed.fetch_add(1);
+					return;
 				}
-				if (Record->Values.size() != 1)
+
+				if (Result.Result.Results.size() != LocalKeys.size())
 				{
-					continue;
+					ZEN_DEBUG("GetCacheRecords iteration {} empty records", Iteration);
+					Completed.fetch_add(1);
+					return;
 				}
-
-				for (const cacherequests::GetCacheRecordResultValue& Value : Record->Values)
+				for (size_t Index = 0; const std::optional<cacherequests::GetCacheRecordResult>& Record : Result.Result.Results)
 				{
-					if (!Value.Body)
+					const CacheKey& ExpectedKey = LocalKeys[Index++];
+					if (!Record)
 					{
 						continue;
 					}
+					if (Record->Key != ExpectedKey)
+					{
+						continue;
+					}
+					if (Record->Values.size() != 1)
+					{
+						continue;
+					}
+
+					for (const cacherequests::GetCacheRecordResultValue& Value : Record->Values)
+					{
+						if (!Value.Body)
+						{
+							continue;
+						}
+					}
 				}
-			}
-			Completed.fetch_add(1);
-		});
+				Completed.fetch_add(1);
+			},
+			WorkerThreadPool::EMode::EnableBacklog);
 	}
 	while (Completed < ThreadCount * KeyMultiplier)
 	{