graceful wait in parallelwork destructor (#438)

* exception safety when issuing ParallelWork
* add asserts to Latch usage to catch usage errors
* extended error messaging and recovery handling in ParallelWork destructor to help find issues

1 files changed, 47 insertions, 35 deletions

diff --git a/src/zenstore/buildstore/buildstore.cpp b/src/zenstore/buildstore/buildstore.cpp
index c25f762f5..20dc55bca 100644
--- a/src/zenstore/buildstore/buildstore.cpp
+++ b/src/zenstore/buildstore/buildstore.cpp
@@ -528,43 +528,55 @@ BuildStore::GetMetadatas(std::span<const IoHash> BlobHashes, WorkerThreadPool* O
 		std::atomic<bool> PauseFlag;
 		ParallelWork	  Work(AbortFlag, PauseFlag);
 
-		m_MetadataBlockStore.IterateChunks(
-			MetaLocations,
-			[this, OptionalWorkerPool, &Work, &Result, &MetaLocations, &MetaLocationResultIndexes, &ReferencedBlocks, DoOneBlock](
-				uint32_t				BlockIndex,
-				std::span<const size_t> ChunkIndexes) -> bool {
-				ZEN_UNUSED(BlockIndex);
-				if (ChunkIndexes.size() == MetaLocations.size() || OptionalWorkerPool == nullptr || ReferencedBlocks.size() == 1)
-				{
-					return DoOneBlock(MetaLocations, MetaLocationResultIndexes, ChunkIndexes, Result);
-				}
-				else
-				{
-					ZEN_ASSERT(OptionalWorkerPool != nullptr);
-					std::vector<size_t> TmpChunkIndexes(ChunkIndexes.begin(), ChunkIndexes.end());
-					Work.ScheduleWork(
-						*OptionalWorkerPool,
-						[this, &Result, &MetaLocations, &MetaLocationResultIndexes, DoOneBlock, ChunkIndexes = std::move(TmpChunkIndexes)](
-							std::atomic<bool>& AbortFlag) {
-							if (AbortFlag)
-							{
-								return;
-							}
-							try
-							{
-								if (!DoOneBlock(MetaLocations, MetaLocationResultIndexes, ChunkIndexes, Result))
+		try
+		{
+			m_MetadataBlockStore.IterateChunks(
+				MetaLocations,
+				[this, OptionalWorkerPool, &Work, &Result, &MetaLocations, &MetaLocationResultIndexes, &ReferencedBlocks, DoOneBlock](
+					uint32_t				BlockIndex,
+					std::span<const size_t> ChunkIndexes) -> bool {
+					ZEN_UNUSED(BlockIndex);
+					if (ChunkIndexes.size() == MetaLocations.size() || OptionalWorkerPool == nullptr || ReferencedBlocks.size() == 1)
+					{
+						return DoOneBlock(MetaLocations, MetaLocationResultIndexes, ChunkIndexes, Result);
+					}
+					else
+					{
+						ZEN_ASSERT(OptionalWorkerPool != nullptr);
+						std::vector<size_t> TmpChunkIndexes(ChunkIndexes.begin(), ChunkIndexes.end());
+						Work.ScheduleWork(
+							*OptionalWorkerPool,
+							[this,
+							 &Result,
+							 &MetaLocations,
+							 &MetaLocationResultIndexes,
+							 DoOneBlock,
+							 ChunkIndexes = std::move(TmpChunkIndexes)](std::atomic<bool>& AbortFlag) {
+								if (AbortFlag)
 								{
-									AbortFlag.store(true);
+									return;
 								}
-							}
-							catch (const std::exception& Ex)
-							{
-								ZEN_WARN("Failed getting metadata for {} chunks. Reason: {}", ChunkIndexes.size(), Ex.what());
-							}
-						});
-					return !Work.IsAborted();
-				}
-			});
+								try
+								{
+									if (!DoOneBlock(MetaLocations, MetaLocationResultIndexes, ChunkIndexes, Result))
+									{
+										AbortFlag.store(true);
+									}
+								}
+								catch (const std::exception& Ex)
+								{
+									ZEN_WARN("Failed getting metadata for {} chunks. Reason: {}", ChunkIndexes.size(), Ex.what());
+								}
+							});
+						return !Work.IsAborted();
+					}
+				});
+		}
+		catch (const std::exception& Ex)
+		{
+			AbortFlag.store(true);
+			ZEN_WARN("Failed iterating block metadata chunks in {}. Reason: '{}'", m_Config.RootDirectory, Ex.what());
+		}
 
 		Work.Wait();
 	}