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, 98 insertions, 90 deletions

diff --git a/src/zenstore/gc.cpp b/src/zenstore/gc.cpp
index 5023695b2..b08e6a3ca 100644
--- a/src/zenstore/gc.cpp
+++ b/src/zenstore/gc.cpp
@@ -869,7 +869,8 @@ GcManager::CollectGarbage(const GcSettings& Settings)
 											  Ex.what());
 									SetCancelGC(true);
 								}
-							});
+							},
+							WorkerThreadPool::EMode::DisableBacklog);
 					}
 					WorkLeft.CountDown();
 					WorkLeft.Wait();
@@ -981,7 +982,8 @@ GcManager::CollectGarbage(const GcSettings& Settings)
 												SetCancelGC(true);
 											}
 										}
-									});
+									},
+									WorkerThreadPool::EMode::DisableBacklog);
 							}
 							WorkLeft.CountDown();
 							WorkLeft.Wait();
@@ -1021,77 +1023,80 @@ GcManager::CollectGarbage(const GcSettings& Settings)
 							GcReferencer*							   Referencer	   = m_GcReferencers[Index];
 							std::pair<std::string, GcReferencerStats>* ReferemcerStats = &Result.ReferencerStats[Index];
 							WorkLeft.AddCount(1);
-							ParallelWorkThreadPool.ScheduleWork([this,
-																 &Ctx,
-																 &WorkLeft,
-																 Referencer,
-																 Index,
-																 Result = &Result,
-																 ReferemcerStats,
-																 &ReferenceValidatorsLock,
-																 &ReferenceValidators]() {
-								ZEN_MEMSCOPE(GetGcTag());
+							ParallelWorkThreadPool.ScheduleWork(
+								[this,
+								 &Ctx,
+								 &WorkLeft,
+								 Referencer,
+								 Index,
+								 Result = &Result,
+								 ReferemcerStats,
+								 &ReferenceValidatorsLock,
+								 &ReferenceValidators]() {
+									ZEN_MEMSCOPE(GetGcTag());
 
-								auto							   _ = MakeGuard([&WorkLeft]() { WorkLeft.CountDown(); });
-								std::vector<GcReferenceValidator*> Validators;
-								auto							   __ = MakeGuard([&Validators]() {
-									  while (!Validators.empty())
-									  {
-										  delete Validators.back();
-										  Validators.pop_back();
-									  }
-								  });
-								try
-								{
+									auto							   _ = MakeGuard([&WorkLeft]() { WorkLeft.CountDown(); });
+									std::vector<GcReferenceValidator*> Validators;
+									auto							   __ = MakeGuard([&Validators]() {
+										  while (!Validators.empty())
+										  {
+											  delete Validators.back();
+											  Validators.pop_back();
+										  }
+									  });
+									try
 									{
-										SCOPED_TIMER(ReferemcerStats->second.CreateReferenceValidatorsMS =
-														 std::chrono::milliseconds(Timer.GetElapsedTimeMs()););
-										Validators = Referencer->CreateReferenceValidators(Ctx);
+										{
+											SCOPED_TIMER(ReferemcerStats->second.CreateReferenceValidatorsMS =
+															 std::chrono::milliseconds(Timer.GetElapsedTimeMs()););
+											Validators = Referencer->CreateReferenceValidators(Ctx);
+										}
+										if (!Validators.empty())
+										{
+											RwLock::ExclusiveLockScope __(ReferenceValidatorsLock);
+											for (auto& ReferenceValidator : Validators)
+											{
+												size_t ReferencesStatsIndex = Result->ReferenceValidatorStats.size();
+												Result->ReferenceValidatorStats.push_back({ReferenceValidator->GetGcName(Ctx), {}});
+												ReferenceValidators.insert_or_assign(
+													std::unique_ptr<GcReferenceValidator>(ReferenceValidator),
+													ReferencesStatsIndex);
+												ReferenceValidator = nullptr;
+											}
+										}
 									}
-									if (!Validators.empty())
+									catch (const std::system_error& Ex)
 									{
-										RwLock::ExclusiveLockScope __(ReferenceValidatorsLock);
-										for (auto& ReferenceValidator : Validators)
+										if (IsOOD(Ex) || IsOOM(Ex))
 										{
-											size_t ReferencesStatsIndex = Result->ReferenceValidatorStats.size();
-											Result->ReferenceValidatorStats.push_back({ReferenceValidator->GetGcName(Ctx), {}});
-											ReferenceValidators.insert_or_assign(std::unique_ptr<GcReferenceValidator>(ReferenceValidator),
-																				 ReferencesStatsIndex);
-											ReferenceValidator = nullptr;
+											ZEN_WARN("GCV2: Failed creating reference validators for {}. Reason: '{}'",
+													 Referencer->GetGcName(Ctx),
+													 Ex.what());
+										}
+										else
+										{
+											ZEN_ERROR("GCV2: Failed creating reference validators for {}. Reason: '{}'",
+													  Referencer->GetGcName(Ctx),
+													  Ex.what());
 										}
+										SetCancelGC(true);
 									}
-								}
-								catch (const std::system_error& Ex)
-								{
-									if (IsOOD(Ex) || IsOOM(Ex))
+									catch (const std::bad_alloc& Ex)
 									{
-										ZEN_WARN("GCV2: Failed creating reference validators for {}. Reason: '{}'",
-												 Referencer->GetGcName(Ctx),
-												 Ex.what());
+										ZEN_ERROR("GCV2: Failed creating reference validators for {}. Reason: '{}'",
+												  Referencer->GetGcName(Ctx),
+												  Ex.what());
+										SetCancelGC(true);
 									}
-									else
+									catch (const std::exception& Ex)
 									{
 										ZEN_ERROR("GCV2: Failed creating reference validators for {}. Reason: '{}'",
 												  Referencer->GetGcName(Ctx),
 												  Ex.what());
+										SetCancelGC(true);
 									}
-									SetCancelGC(true);
-								}
-								catch (const std::bad_alloc& Ex)
-								{
-									ZEN_ERROR("GCV2: Failed creating reference validators for {}. Reason: '{}'",
-											  Referencer->GetGcName(Ctx),
-											  Ex.what());
-									SetCancelGC(true);
-								}
-								catch (const std::exception& Ex)
-								{
-									ZEN_ERROR("GCV2: Failed creating reference validators for {}. Reason: '{}'",
-											  Referencer->GetGcName(Ctx),
-											  Ex.what());
-									SetCancelGC(true);
-								}
-							});
+								},
+								WorkerThreadPool::EMode::DisableBacklog);
 						}
 						WorkLeft.CountDown();
 						WorkLeft.Wait();
@@ -1221,47 +1226,49 @@ GcManager::CollectGarbage(const GcSettings& Settings)
 									size_t									   Index   = It.second;
 									std::pair<std::string, GcReferencerStats>* Stats   = &Result.ReferencerStats[Index];
 									WorkLeft.AddCount(1);
-									LockedPhaseThreadPool.ScheduleWork([this, &Ctx, Checker, Index, Stats, &WorkLeft]() {
-										ZEN_MEMSCOPE(GetGcTag());
+									LockedPhaseThreadPool.ScheduleWork(
+										[this, &Ctx, Checker, Index, Stats, &WorkLeft]() {
+											ZEN_MEMSCOPE(GetGcTag());
 
-										auto _ = MakeGuard([&WorkLeft]() { WorkLeft.CountDown(); });
-										try
-										{
-											SCOPED_TIMER(Stats->second.UpdateLockedStateMS =
-															 std::chrono::milliseconds(Timer.GetElapsedTimeMs()););
-											Checker->UpdateLockedState(Ctx);
-										}
-										catch (const std::system_error& Ex)
-										{
-											if (IsOOD(Ex) || IsOOM(Ex))
+											auto _ = MakeGuard([&WorkLeft]() { WorkLeft.CountDown(); });
+											try
+											{
+												SCOPED_TIMER(Stats->second.UpdateLockedStateMS =
+																 std::chrono::milliseconds(Timer.GetElapsedTimeMs()););
+												Checker->UpdateLockedState(Ctx);
+											}
+											catch (const std::system_error& Ex)
+											{
+												if (IsOOD(Ex) || IsOOM(Ex))
+												{
+													ZEN_WARN("GCV2: Failed Updating locked state for {}. Reason: '{}'",
+															 Checker->GetGcName(Ctx),
+															 Ex.what());
+												}
+												else
+												{
+													ZEN_ERROR("GCV2: Failed Updating locked state for {}. Reason: '{}'",
+															  Checker->GetGcName(Ctx),
+															  Ex.what());
+												}
+												SetCancelGC(true);
+											}
+											catch (const std::bad_alloc& Ex)
 											{
 												ZEN_WARN("GCV2: Failed Updating locked state for {}. Reason: '{}'",
 														 Checker->GetGcName(Ctx),
 														 Ex.what());
+												SetCancelGC(true);
 											}
-											else
+											catch (const std::exception& Ex)
 											{
 												ZEN_ERROR("GCV2: Failed Updating locked state for {}. Reason: '{}'",
 														  Checker->GetGcName(Ctx),
 														  Ex.what());
+												SetCancelGC(true);
 											}
-											SetCancelGC(true);
-										}
-										catch (const std::bad_alloc& Ex)
-										{
-											ZEN_WARN("GCV2: Failed Updating locked state for {}. Reason: '{}'",
-													 Checker->GetGcName(Ctx),
-													 Ex.what());
-											SetCancelGC(true);
-										}
-										catch (const std::exception& Ex)
-										{
-											ZEN_ERROR("GCV2: Failed Updating locked state for {}. Reason: '{}'",
-													  Checker->GetGcName(Ctx),
-													  Ex.what());
-											SetCancelGC(true);
-										}
-									});
+										},
+										WorkerThreadPool::EMode::EnableBacklog);
 								}
 								WorkLeft.CountDown();
 								WorkLeft.Wait();
@@ -1373,7 +1380,8 @@ GcManager::CollectGarbage(const GcSettings& Settings)
 												  Ex.what());
 										SetCancelGC(true);
 									}
-								});
+								},
+								WorkerThreadPool::EMode::EnableBacklog);
 						}
 						WorkLeft.CountDown();
 						WorkLeft.Wait();