pause, resume and abort running builds cmd (#421)

- Feature: `zen builds pause`, `zen builds resume` and `zen builds abort` commands to control a running `zen builds` command
  - `--process-id` the process id to control, if omitted it tries to find a running process using the same executable as itself
- Improvement: Process report now indicates if it is pausing or aborting

4 files changed, 14 insertions, 7 deletions

diff --git a/src/zenstore/buildstore/buildstore.cpp b/src/zenstore/buildstore/buildstore.cpp
index 41c747e08..afb7e4bee 100644
--- a/src/zenstore/buildstore/buildstore.cpp
+++ b/src/zenstore/buildstore/buildstore.cpp
@@ -483,7 +483,8 @@ BuildStore::GetMetadatas(std::span<const IoHash> BlobHashes, WorkerThreadPool* O
 	if (!MetaLocations.empty())
 	{
 		std::atomic<bool> AbortFlag;
-		ParallelWork	  Work(AbortFlag);
+		std::atomic<bool> PauseFlag;
+		ParallelWork	  Work(AbortFlag, PauseFlag);
 
 		m_MetadataBlockStore.IterateChunks(
 			MetaLocations,
diff --git a/src/zenstore/cache/cachedisklayer.cpp b/src/zenstore/cache/cachedisklayer.cpp
index e973cee77..3f1f0e34a 100644
--- a/src/zenstore/cache/cachedisklayer.cpp
+++ b/src/zenstore/cache/cachedisklayer.cpp
@@ -3978,7 +3978,8 @@ ZenCacheDiskLayer::DiscoverBuckets()
 
 	WorkerThreadPool& Pool = GetLargeWorkerPool(EWorkloadType::Burst);
 	std::atomic<bool> AbortFlag;
-	ParallelWork	  Work(AbortFlag);
+	std::atomic<bool> PauseFlag;
+	ParallelWork	  Work(AbortFlag, PauseFlag);
 	for (auto& BucketPath : FoundBucketDirectories)
 	{
 		Work.ScheduleWork(Pool, [this, &SyncLock, BucketPath](std::atomic<bool>&) {
@@ -4141,7 +4142,8 @@ ZenCacheDiskLayer::Flush()
 	{
 		WorkerThreadPool& Pool = GetMediumWorkerPool(EWorkloadType::Burst);
 		std::atomic<bool> AbortFlag;
-		ParallelWork	  Work(AbortFlag);
+		std::atomic<bool> PauseFlag;
+		ParallelWork	  Work(AbortFlag, PauseFlag);
 		try
 		{
 			for (auto& Bucket : Buckets)
@@ -4164,8 +4166,10 @@ ZenCacheDiskLayer::Flush()
 		{
 			ZEN_ERROR("Failed to flush buckets at '{}'. Reason: '{}'", m_RootDir, Ex.what());
 		}
-		Work.Wait(1000,
-				  [&](std::ptrdiff_t Remaining, bool) { ZEN_DEBUG("Waiting for {} buckets at '{}' to flush", Remaining, m_RootDir); });
+		Work.Wait(1000, [&](bool IsAborted, bool IsPaused, std::ptrdiff_t RemainingWork) {
+			ZEN_UNUSED(IsAborted, IsPaused);
+			ZEN_DEBUG("Waiting for {} buckets at '{}' to flush", RemainingWork, m_RootDir);
+		});
 	}
 }
 
diff --git a/src/zenstore/compactcas.cpp b/src/zenstore/compactcas.cpp
index 75562176e..0c9302ec8 100644
--- a/src/zenstore/compactcas.cpp
+++ b/src/zenstore/compactcas.cpp
@@ -402,7 +402,8 @@ CasContainerStrategy::IterateChunks(std::span<const IoHash>											  ChunkHas
 	std::atomic<bool> AsyncContinue = true;
 	{
 		std::atomic<bool> AbortFlag;
-		ParallelWork	  Work(AbortFlag);
+		std::atomic<bool> PauseFlag;
+		ParallelWork	  Work(AbortFlag, PauseFlag);
 		const bool		  Continue = m_BlockStore.IterateChunks(
 			   FoundChunkLocations,
 			   [this,
diff --git a/src/zenstore/filecas.cpp b/src/zenstore/filecas.cpp
index 56979f267..539b5e95b 100644
--- a/src/zenstore/filecas.cpp
+++ b/src/zenstore/filecas.cpp
@@ -664,7 +664,8 @@ FileCasStrategy::IterateChunks(std::span<IoHash>												 ChunkHashes,
 		};
 
 		std::atomic<bool> AbortFlag;
-		ParallelWork	  Work(AbortFlag);
+		std::atomic<bool> PauseFlag;
+		ParallelWork	  Work(AbortFlag, PauseFlag);
 		for (size_t Index = 0; Index < FoundChunkIndexes.size(); Index++)
 		{
 			if (!AsyncContinue)