avoid big ioworker backlog (#733)

* add ability to override scheduling mode in ParallelWork
* Don't increase buffering size when copying from local cache with --boost-worker-memory enabled
* Rework scheduling writes of downloaded data to reduce memory usage

1 files changed, 40 insertions, 35 deletions

diff --git a/src/zenremotestore/builds/buildstorageoperations.cpp b/src/zenremotestore/builds/buildstorageoperations.cpp
index 485973e2b..76cbe6540 100644
--- a/src/zenremotestore/builds/buildstorageoperations.cpp
+++ b/src/zenremotestore/builds/buildstorageoperations.cpp
@@ -1302,37 +1302,39 @@ BuildsOperationUpdateFolder::Execute(FolderContent& OutLocalFolderState)
 					}
 				}
 
-				Work.ScheduleWork(m_IOWorkerPool,
-								  [this,
-								   &SequenceIndexChunksLeftToWriteCounters,
-								   &Work,
-								   &ExistsResult,
-								   &WritePartsComplete,
-								   &LooseChunkHashWorks,
-								   LooseChunkHashWorkIndex,
-								   TotalRequestCount,
-								   TotalPartWriteCount,
-								   &WriteCache,
-								   &FilteredDownloadedBytesPerSecond,
-								   &FilteredWrittenBytesPerSecond](std::atomic<bool>&) mutable {
-									  ZEN_TRACE_CPU("Async_ReadPreDownloadedChunk");
-									  if (!m_AbortFlag)
-									  {
-										  LooseChunkHashWorkData& LooseChunkHashWork = LooseChunkHashWorks[LooseChunkHashWorkIndex];
-										  const uint32_t RemoteChunkIndex = LooseChunkHashWorks[LooseChunkHashWorkIndex].RemoteChunkIndex;
-										  WriteLooseChunk(RemoteChunkIndex,
-														  ExistsResult,
-														  SequenceIndexChunksLeftToWriteCounters,
-														  WritePartsComplete,
-														  std::move(LooseChunkHashWork.ChunkTargetPtrs),
-														  WriteCache,
-														  Work,
-														  TotalRequestCount,
-														  TotalPartWriteCount,
-														  FilteredDownloadedBytesPerSecond,
-														  FilteredWrittenBytesPerSecond);
-									  }
-								  });
+				Work.ScheduleWork(
+					m_IOWorkerPool,
+					[this,
+					 &SequenceIndexChunksLeftToWriteCounters,
+					 &Work,
+					 &ExistsResult,
+					 &WritePartsComplete,
+					 &LooseChunkHashWorks,
+					 LooseChunkHashWorkIndex,
+					 TotalRequestCount,
+					 TotalPartWriteCount,
+					 &WriteCache,
+					 &FilteredDownloadedBytesPerSecond,
+					 &FilteredWrittenBytesPerSecond](std::atomic<bool>&) mutable {
+						ZEN_TRACE_CPU("Async_ReadPreDownloadedChunk");
+						if (!m_AbortFlag)
+						{
+							LooseChunkHashWorkData& LooseChunkHashWork = LooseChunkHashWorks[LooseChunkHashWorkIndex];
+							const uint32_t			RemoteChunkIndex   = LooseChunkHashWorks[LooseChunkHashWorkIndex].RemoteChunkIndex;
+							WriteLooseChunk(RemoteChunkIndex,
+											ExistsResult,
+											SequenceIndexChunksLeftToWriteCounters,
+											WritePartsComplete,
+											std::move(LooseChunkHashWork.ChunkTargetPtrs),
+											WriteCache,
+											Work,
+											TotalRequestCount,
+											TotalPartWriteCount,
+											FilteredDownloadedBytesPerSecond,
+											FilteredWrittenBytesPerSecond);
+						}
+					},
+					WorkerThreadPool::EMode::DisableBacklog);
 			}
 
 			std::unique_ptr<CloneQueryInterface> CloneQuery;
@@ -1592,7 +1594,8 @@ BuildsOperationUpdateFolder::Execute(FolderContent& OutLocalFolderState)
 														FilteredWrittenBytesPerSecond.Stop();
 													}
 												}
-											});
+											},
+											WorkerThreadPool::EMode::DisableBacklog);
 									}
 								});
 						}
@@ -1780,7 +1783,8 @@ BuildsOperationUpdateFolder::Execute(FolderContent& OutLocalFolderState)
 														FilteredWrittenBytesPerSecond.Stop();
 													}
 												}
-											});
+											},
+											WorkerThreadPool::EMode::DisableBacklog);
 									}
 								}
 							}
@@ -3750,7 +3754,7 @@ BuildsOperationUpdateFolder::WriteLocalChunkToCache(CloneQueryInterface*					  C
 					break;
 				}
 				const uint64_t NextChunkLength = m_RemoteContent.ChunkedContent.ChunkRawSizes[NextOp.ChunkIndex];
-				if (ReadLength + NextChunkLength > m_Options.MaximumInMemoryPayloadSize)
+				if (ReadLength + NextChunkLength > BufferedOpenFile::BlockSize)
 				{
 					break;
 				}
@@ -4421,7 +4425,8 @@ BuildsOperationUpdateFolder::AsyncWriteDownloadedChunk(const std::filesystem::pa
 					}
 				}
 			}
-		});
+		},
+		WorkerThreadPool::EMode::DisableBacklog);
 }
 
 void