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, 72 insertions, 65 deletions

diff --git a/src/zenstore/compactcas.cpp b/src/zenstore/compactcas.cpp
index 2ab5752ff..b00abb2cb 100644
--- a/src/zenstore/compactcas.cpp
+++ b/src/zenstore/compactcas.cpp
@@ -396,89 +396,96 @@ CasContainerStrategy::IterateChunks(std::span<const IoHash>											  ChunkHas
 		return m_BlockStore.IterateBlock(
 			FoundChunkLocations,
 			ChunkIndexes,
-			[AsyncCallback, FoundChunkIndexes, LargeSizeLimit](size_t ChunkIndex, const void* Data, uint64_t Size) {
+			[AsyncCallback, FoundChunkIndexes](size_t ChunkIndex, const void* Data, uint64_t Size) {
 				if (Data == nullptr)
 				{
 					return AsyncCallback(FoundChunkIndexes[ChunkIndex], IoBuffer());
 				}
 				return AsyncCallback(FoundChunkIndexes[ChunkIndex], IoBuffer(IoBuffer::Wrap, Data, Size));
 			},
-			[&](size_t ChunkIndex, BlockStoreFile& File, uint64_t Offset, uint64_t Size) {
+			[AsyncCallback, FoundChunkIndexes](size_t ChunkIndex, BlockStoreFile& File, uint64_t Offset, uint64_t Size) {
 				return AsyncCallback(FoundChunkIndexes[ChunkIndex], File.GetChunk(Offset, Size));
 			},
 			LargeSizeLimit);
 	};
 
-	std::atomic<bool> AsyncContinue = true;
+	std::atomic<bool> AbortFlag;
 	{
-		std::atomic<bool> AbortFlag;
 		std::atomic<bool> PauseFlag;
 		ParallelWork	  Work(AbortFlag, PauseFlag);
-		const bool		  Continue = m_BlockStore.IterateChunks(
-			   FoundChunkLocations,
-			   [this,
-				&Work,
-				&AsyncContinue,
-				&AsyncCallback,
-				LargeSizeLimit,
-				DoOneBlock,
-				&FoundChunkIndexes,
-				&FoundChunkLocations,
-				OptionalWorkerPool](uint32_t BlockIndex, std::span<const size_t> ChunkIndexes) {
-				   if (OptionalWorkerPool && (ChunkIndexes.size() > 3))
-				   {
-					   std::vector<size_t> TmpChunkIndexes(ChunkIndexes.begin(), ChunkIndexes.end());
-					   Work.ScheduleWork(
-						   *OptionalWorkerPool,
-						   [this,
-							&AsyncContinue,
-							&AsyncCallback,
-							LargeSizeLimit,
-							DoOneBlock,
-							BlockIndex,
-							&FoundChunkIndexes,
-							&FoundChunkLocations,
-							ChunkIndexes = std::move(TmpChunkIndexes)](std::atomic<bool>& AbortFlag) {
-							   if (AbortFlag)
-							   {
-								   AsyncContinue.store(false);
-							   }
-							   if (!AsyncContinue)
-							   {
-								   return;
-							   }
-							   try
-							   {
-								   bool Continue =
-									   DoOneBlock(AsyncCallback, LargeSizeLimit, FoundChunkIndexes, FoundChunkLocations, ChunkIndexes);
-								   if (!Continue)
-								   {
-									   AsyncContinue.store(false);
-								   }
-							   }
-							   catch (const std::exception& Ex)
-							   {
-								   ZEN_WARN("Failed iterating chunks for cas root path {}, block {}. Reason: '{}'",
-											m_RootDirectory,
-											BlockIndex,
-											Ex.what());
-								   AsyncContinue.store(false);
-							   }
-						   });
-					   return AsyncContinue.load();
-				   }
-				   else
-				   {
-					   return DoOneBlock(AsyncCallback, LargeSizeLimit, FoundChunkIndexes, FoundChunkLocations, ChunkIndexes);
-				   }
-			   });
-		if (!Continue)
+		try
 		{
-			AsyncContinue.store(false);
+			const bool Continue = m_BlockStore.IterateChunks(
+				FoundChunkLocations,
+				[this,
+				 &Work,
+				 &AbortFlag,
+				 &AsyncCallback,
+				 LargeSizeLimit,
+				 DoOneBlock,
+				 &FoundChunkIndexes,
+				 &FoundChunkLocations,
+				 OptionalWorkerPool](uint32_t BlockIndex, std::span<const size_t> ChunkIndexes) {
+					if (OptionalWorkerPool && (ChunkIndexes.size() > 3))
+					{
+						std::vector<size_t> TmpChunkIndexes(ChunkIndexes.begin(), ChunkIndexes.end());
+						Work.ScheduleWork(
+							*OptionalWorkerPool,
+							[this,
+							 &AsyncCallback,
+							 LargeSizeLimit,
+							 DoOneBlock,
+							 BlockIndex,
+							 &FoundChunkIndexes,
+							 &FoundChunkLocations,
+							 ChunkIndexes = std::move(TmpChunkIndexes)](std::atomic<bool>& AbortFlag) {
+								if (AbortFlag)
+								{
+									return;
+								}
+								try
+								{
+									bool Continue =
+										DoOneBlock(AsyncCallback, LargeSizeLimit, FoundChunkIndexes, FoundChunkLocations, ChunkIndexes);
+									if (!Continue)
+									{
+										AbortFlag.store(true);
+									}
+								}
+								catch (const std::exception& Ex)
+								{
+									ZEN_WARN("Failed iterating chunks for cas root path {}, block {}. Reason: '{}'",
+											 m_RootDirectory,
+											 BlockIndex,
+											 Ex.what());
+									AbortFlag.store(true);
+								}
+							});
+						return !AbortFlag.load();
+					}
+					else
+					{
+						if (!DoOneBlock(AsyncCallback, LargeSizeLimit, FoundChunkIndexes, FoundChunkLocations, ChunkIndexes))
+						{
+							AbortFlag.store(true);
+						}
+						return !AbortFlag.load();
+					}
+				});
+			if (!Continue)
+			{
+				AbortFlag.store(true);
+			}
 		}
+		catch (const std::exception& Ex)
+		{
+			AbortFlag.store(true);
+			ZEN_WARN("Failed iterating chunks for cas root path {}. Reason: '{}'", m_RootDirectory, Ex.what());
+		}
+
 		Work.Wait();
 	}
-	return AsyncContinue.load();
+	return !AbortFlag.load();
 }
 
 void