}
FAutoConsoleCommand StartWorkTestCommand(TEXT("StartWorkTest"), TEXT(""), FConsoleCommandDelegate::CreateStatic(&StartWorkTest));
FAutoConsoleCommand AddWorkTestCommand(TEXT("AddWork"), TEXT(""), FConsoleCommandDelegate::CreateStatic(&AddWork));
FAutoConsoleCommand StopWorkTestCommand(TEXT("StopWorkTest"), TEXT(""), FConsoleCommandDelegate::CreateStatic(&StopWorkTest));

file
class        class FVirtualHeightfieldMeshRendererExtension : public FRenderResource

	/** Call once per frame for each mesh/view that has relevance. This allocates the buffers to use for the frame and adds the work to fill the buffers to the queue. */
	VirtualHeightfieldMesh::FDrawInstanceBuffers& AddWork(FRHICommandListBase& RHICmdList, FVirtualHeightfieldMeshSceneProxy const* InProxy, FSceneView const* InMainView, FSceneView const* InCullView);
	/** Submit all the work added by AddWork(). The work fills all of the buffers ready for use by the referencing mesh batches. */
	void SubmitWork(FRDGBuilder& GraphBuilder);

protected:
	//~ Begin FRenderResource Interface
	virtual void ReleaseRHI() override;
	//~ End FRenderResource Interface

file
function     VirtualHeightfieldMesh::FDrawInstanceBuffers& FVirtualHeightfieldMeshRendererExtension::AddWork

}

VirtualHeightfieldMesh::FDrawInstanceBuffers& FVirtualHeightfieldMeshRendererExtension::AddWork(FRHICommandListBase& RHICmdList, FVirtualHeightfieldMeshSceneProxy const* InProxy, FSceneView const* InMainView, FSceneView const* InCullView)
{
	UE::TScopeLock Lock(Mutex);

	// If we hit this then BegineFrame()/EndFrame() logic needs fixing in the Scene Renderer.
	if (!ensure(!bInFrame))
	{

file
function     void FVirtualHeightfieldMeshSceneProxy::GetDynamicMeshElements

	{
		// Can't add new work while bInFrame.
		// In UE5 we need to AddWork()/SubmitWork() in two phases: InitViews() and InitViewsAfterPrepass()
		// The main renderer hooks for that don't exist in UE5.0 and are only added in UE5.1
		// That means that for UE5.0 we always hit this for shadow drawing and shadows will not be rendered.
		// Not earlying out here can lead to crashes from buffers being released too soon.
		return;
	}

file
function     void FVirtualHeightfieldMeshSceneProxy::GetDynamicMeshElements

		if (VisibilityMap & (1 << ViewIndex))
		{
			VirtualHeightfieldMesh::FDrawInstanceBuffers& Buffers = GVirtualHeightfieldMeshViewRendererExtension.AddWork(Collector.GetRHICommandList(), this, ViewFamily.Views[0], Views[ViewIndex]);

			FMeshBatch& Mesh = Collector.AllocateMesh();
			Mesh.bWireframe = AllowDebugViewmodes() && ViewFamily.EngineShowFlags.Wireframe;
			Mesh.bUseWireframeSelectionColoring = IsSelected();
			Mesh.VertexFactory = VertexFactory;
			Mesh.MaterialRenderProxy = Material;

file
function     static void AddWork

}

static void AddWork()
{
	int32 TaskIndex = FMath::Rand() % TestProgressNotificationTasks.Num();
	TUniquePtr<FTestProgressNotification>& TestProgressNotificationTask = TestProgressNotificationTasks[TaskIndex];
	//if (TestProgressNotificationTask)
	{
		TestProgressNotificationTask->RemainingJobs += 100;

file
namespace    uba
function     int WrappedMain
lambda-function

			storageServer.TraverseAllCasFiles([&](const CasKey& casKey)
				{
					workManager.AddWork([&, casKey]()
						{
							Storage::RetrieveResult res;
							storageServer.EnsureCasFile(casKey, TC("Dummy"));
							CasKey casKey2 = AsCompressed(casKey, false);
							if (!storageClient.RetrieveCasFile(res, casKey2, TC("")))
								success = false;

file
namespace    uba
function     void FileMappingBuffer::UnmapView

			};
		if (m_workManager && newSize == InvalidValue)
			m_workManager->AddWork([=, h = TString(hint_)]() { unmap(h.c_str()); }, 1, TC("UnmapView"));
		else
			unmap(hint_);
	}

	void FileMappingBuffer::GetSizeAndCount(FileMappingType type, u64& outSize, u32& outCount)
	{

file
namespace    uba
function     void NetworkServer::Worker::DoAdditionalWorkAndSignalAvailable

			// Both locks needs to be taken to verify if additional work
			// is present before making ourself available to avoid
			// a race where AddWork would not see this thread in the
			// available list after adding some work.
			SCOPED_WRITE_LOCK(server.m_availableWorkersLock, lock1);
			SCOPED_READ_LOCK(server.m_additionalWorkLock, lock2);
			// Verify there is not additional work while we hold both lock
			// and only add ourself as available if no additional work is present.
			if (!server.m_additionalWork.empty())

file
namespace    uba
function     void NetworkServer::AddWork

	}

	void NetworkServer::AddWork(const Function<void()>& work, u32 count, const tchar* desc)
	{
		SCOPED_WRITE_LOCK(m_additionalWorkLock, lock);
		for (u32 i = 0; i != count; ++i)
		{
			m_additionalWork.push_back({ work });
			if (m_workTracker)

file
function     bool Session::WriteFilesToDisk
lambda-function

			if (IsRarelyReadAfterWritten(process, file.name.c_str(), file.name.size()) || file.mappingWritten > m_keepOutputFileMemoryMapsThreshold)
			{
				m_workManager->AddWork([mh = file.mappingHandle]()
					{
						CloseFileMapping(mh);
					}, 1, TC("CFM"));
				//CloseFileMapping(file.mappingHandle);
			}
			else

file
function     bool Session::WriteFilesToDisk

			events[i].Create(true);

			m_workManager->AddWork([&, ii = i]()
				{
					writeFile(*files[ii]);
					events[ii].Set();
				}, 1, TC(""));
		}
		for (u32 i=0; i!=fileCount; ++i)
			events[i].IsSet();
		*/
		return true;
	}

	bool Session::AllocFailed(Process& process, const tchar* allocType, u32 error)
	{
		m_logger.Warning(TC("Allocation failed in %s (%s).. process will sleep and try again"), allocType, LastErrorToText(error).data);
		return true;
	}

	bool Session::GetNextProcess(Process& process, bool& outNewProcess, NextProcessInfo& outNextProcess, u32 prevExitCode, BinaryReader& statsReader)
	{
		if (!m_getNextProcessFunction)
		{
			outNewProcess = false;
			return true;

file
namespace    uba
function     bool SessionClient::EnsureApplicationEnvironment

			for (auto& m : modules)
			{
				m_client.AddWork([&handledCount, &m, this, &success, &keyStr]()
					{
						++handledCount;
						auto g = MakeGuard([&]() { m.done.Set(); });
						CasKey newCasKey;
						bool storeUncompressed = true;
						u64 fileSize;

file
namespace    uba
function     bool SessionClient::SendProcessAvailable

					knownInputKey = AsCompressed(knownInputKey, false);

				m_client.AddWork([knownInputKey, this]()
					{
						Storage::RetrieveResult result;
						bool allowProxy = true;
						bool res = m_storage.RetrieveCasFile(result, knownInputKey, TC("KnownInput"), nullptr, 1, allowProxy);
						(void)res;
					}, 1, TC("KnownInput"));

file
namespace    uba
function     bool StorageImpl::WriteCompressed
lambda-function

					{
						#if UBA_EXPERIMENTAL
						m_workManager->AddWork([=, f = TString(from)]() { UnmapViewOfFile(uncompressedData, fileSize, f.c_str()); }, 1, TC("UnmapFile"));
						#else
						UnmapViewOfFile(uncompressedData, fileSize, from);
						#endif
					});

				if (!WriteMemToCompressedFile(destinationFile, workCount, uncompressedData, fileSize, maxUncompressedBlock, totalWritten))

file
namespace    uba
function     bool StorageImpl::WriteMemToCompressedFile

		rec->refCount = workerCount + 1; // We need to keep refcount up 1 to make sure it is not deleted before we read rec->written
		m_workManager->AddWork(work, workerCount-1, TC("Compress")); // We are a worker ourselves
		work();
		rec->events[rec->workCount - 1].IsSet();

		totalWritten += rec->written;
		bool error = rec->error;

file
namespace    uba
function     bool StorageImpl::DecompressMemoryToMemory

				workerCount = Min(workerCount, MaxWorkItemsPerAction); // Cap this to not starve other things
				rec->refCount += workerCount;
				m_workManager->AddWork(work, workerCount, TC("DecompressMemToMem"));
			}

			TimerScope ts(stats.decompressToMem);
			work();
			rec->done.IsSet();
			bool success = !rec->error;

file
namespace    uba
function     bool StorageImpl::CheckCasContent
lambda-function

			{
				++entryCount;
				workManager.AddWork([&, filePath = TString(fullPath.data), name = TString(e.name), lastWritten = e.lastWritten]()
					{
						StringBuffer<> timeStr;
						writeTimeAgo(timeStr, lastWritten);

						
						//m_logger.Info(TC("Validating %s (%s ago)"), name.c_str(), timeStr.data);

file
namespace    uba
function     bool StorageImpl::DeleteAllCas
lambda-function

					if (!IsDirectory(e.attributes))
						return;
					workManager.AddWork([&, name = TString(e.name)]()
						{
							StringBuffer<> fullPath;
							fullPath.Append(m_rootDir).Append(name);
							u32 deleteCountTemp = 0;
							DeleteAllFiles(m_logger, fullPath.data, true, &deleteCountTemp);
							atomicDeleteCount += deleteCountTemp;

file
namespace    uba
function     CasKey StorageImpl::CalculateCasKey

				workerCount = Min(workerCount, MaxWorkItemsPerAction); // Cap this to not starve other things
				rec->refCount += workerCount;
				m_workManager->AddWork(work, workerCount, TC("CalculateKey"));
			}

			work();
			rec->done.IsSet();

			hasher.Update(rec->keys.data(), rec->keys.size()*sizeof(CasKey));

file
namespace    uba
function     CasKey StorageImpl::CalculateCasKey
lambda-function

			auto udg = MakeGuard([&]() { 
					#if UBA_EXPERIMENTAL
					m_workManager->AddWork([=, fn = TString(fileName)]() { UnmapViewOfFile(fileData, fileSize, fn.c_str()); }, 1, TC("UnmapFile"));
					#else
					UnmapViewOfFile(fileData, fileSize, fileName);
					#endif
				});

			struct WorkRec

file
namespace    uba
function     CasKey StorageImpl::CalculateCasKey

			rec->refCount = workerCount + 1; // We need to keep refcount up 1 to make sure it is not deleted before we read rec->written
			m_workManager->AddWork(work, workerCount-1, TC("CalculateKey")); // We are a worker ourselves
			work();
			rec->done.IsSet();

			hasher.Update(rec->keys.data(), rec->keys.size()*sizeof(CasKey));

			bool error = rec->error;

file
namespace    uba
function     bool StorageClient::PopulateCasFromDirsRecursive
lambda-function

				}

				workManager.AddWork([&, filePath = TString(fullPath.data), name = TString(e.name)]()
					{
						FileInformation info;
						if (!GetFileInformation(info, m_logger, filePath.c_str()))
						{
							m_logger.Error(TC("Failed to get information for file %s"), filePath.c_str());
							return;

file
namespace    uba
function     bool StorageProxy::HandleMessage

						// Move the additional messages to a job to be able to return this one quickly.
						m_server.AddWork([f = &file, segmentCount, this]()
							{
								SCOPED_WRITE_LOCK(m_largeFileLock, lock);
								//TrackWorkScope tws(m_client, TC("SEGMENTS"));
								auto& file = *f;
								for (u32 i=0; i!=segmentCount; ++i)
								{

file
lambda-function
lambda-function

											auto mif = (MessageInFlight*)userData;
											mif->error = error;
											mif->proxy.m_server.AddWork([mif]() { mif->proxy.HandleReceivedData(*mif); }, 1, TC(""));
										}, mif);
									if (!res)
									{
										// TODO: Don't leak mif
										mif->error = true;
									}

file
namespace    uba
function     void WorkManagerImpl::AddWork

	void WorkManagerImpl::AddWork(const Function<void()>& work, u32 count, const tchar* desc)
	{
		SCOPED_WRITE_LOCK(m_workLock, lock);
		for (u32 i = 0; i != count; ++i)
		{
			m_work.push_back({ work });
			if (m_workTracker.load())

file
namespace    uba
class        class NetworkServer : public WorkManager

		//void UnregisterOnConnection(u8 id);

		virtual void AddWork(const Function<void()>& work, u32 count, const tchar* desc) override final;
		virtual u32 GetWorkerCount() override final;

		struct ClientStats
		{
			u64 send = 0;
			u64 recv = 0;

file
namespace    uba
class        class WorkManager

	{
	public:
		virtual void AddWork(const Function<void()>& work, u32 count, const tchar* desc) = 0;
		virtual u32 GetWorkerCount() = 0;

		u32 TrackWorkStart(const tchar* desc);
		void TrackWorkEnd(u32 id);

		void SetWorkTracker(WorkTracker* workTracker) { m_workTracker = workTracker; }

file
namespace    uba
class        class WorkManagerImpl : public WorkManager

		WorkManagerImpl(u32 workerCount);
		virtual ~WorkManagerImpl();
		virtual void AddWork(const Function<void()>& work, u32 count, const tchar* desc) override;
		virtual u32 GetWorkerCount() override;
		void DoWork(u32 count = 1);
		void FlushWork();

	private:
		struct Worker;