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, 22 insertions, 19 deletions

diff --git a/src/zen/cmds/cache_cmd.cpp b/src/zen/cmds/cache_cmd.cpp
index ea7ad79ee..aa23373c4 100644
--- a/src/zen/cmds/cache_cmd.cpp
+++ b/src/zen/cmds/cache_cmd.cpp
@@ -478,25 +478,28 @@ CacheGenerateCommand::Run(const ZenCliOptions& GlobalOptions, int argc, char** a
 		std::span<uint64_t> BatchSizes = std::span<uint64_t>(Sizes).subspan(Offset, Min(Max(SizeCount, 1u), Sizes.size() - Offset));
 
 		WorkLatch.AddCount(1);
-		WorkerPool.ScheduleWork([&, BatchSizes, RequestIndex]() {
-			auto	  _ = MakeGuard([&WorkLatch]() { WorkLatch.CountDown(); });
-			CbPackage Package;
-			if (m_MaxAttachmentCount > 0 && SizeCount > 0)
-			{
-				auto Request = GeneratePutCacheRecordRequest(BatchSizes, RequestIndex);
-				ZEN_ASSERT(Request.Format(Package));
-			}
-			else
-			{
-				auto Request = GeneratePutCacheValueRequest(BatchSizes, RequestIndex);
-				ZEN_ASSERT(Request.Format(Package));
-			}
-
-			if (HttpClient::Response Response = Http.Post("/z$/$rpc", Package, HttpClient::Accept(ZenContentType::kCbPackage)); !Response)
-			{
-				ZEN_CONSOLE("{}", Response.ErrorMessage(fmt::format("{}: ", RequestIndex)));
-			}
-		});
+		WorkerPool.ScheduleWork(
+			[&, BatchSizes, RequestIndex]() {
+				auto	  _ = MakeGuard([&WorkLatch]() { WorkLatch.CountDown(); });
+				CbPackage Package;
+				if (m_MaxAttachmentCount > 0 && SizeCount > 0)
+				{
+					auto Request = GeneratePutCacheRecordRequest(BatchSizes, RequestIndex);
+					ZEN_ASSERT(Request.Format(Package));
+				}
+				else
+				{
+					auto Request = GeneratePutCacheValueRequest(BatchSizes, RequestIndex);
+					ZEN_ASSERT(Request.Format(Package));
+				}
+
+				if (HttpClient::Response Response = Http.Post("/z$/$rpc", Package, HttpClient::Accept(ZenContentType::kCbPackage));
+					!Response)
+				{
+					ZEN_CONSOLE("{}", Response.ErrorMessage(fmt::format("{}: ", RequestIndex)));
+				}
+			},
+			WorkerThreadPool::EMode::EnableBacklog);
 		Offset += BatchSizes.size();
 		RequestIndex++;
 	}