structured compute basics (#714)

this change adds the `zencompute` component, which can be used to distribute work dispatched from UE using the DDB (Derived Data Build) APIs via zenserver

this change also adds a distinct zenserver compute mode (`zenserver compute`) which is intended to be used for leaf compute nodes

to exercise the compute functionality without directly involving UE, a `zen exec` subcommand is also added, which can be used to feed replays through the system

all new functionality is considered *experimental* and disabled by default at this time, behind the `zencompute` option in xmake config

1 files changed, 258 insertions, 0 deletions

diff --git a/src/zencompute/actionrecorder.cpp b/src/zencompute/actionrecorder.cpp
new file mode 100644
index 000000000..04c4b5141
--- /dev/null
+++ b/src/zencompute/actionrecorder.cpp
@@ -0,0 +1,258 @@
+// Copyright Epic Games, Inc. All Rights Reserved.
+
+#include "actionrecorder.h"
+
+#include "functionrunner.h"
+
+#include <zencore/compactbinary.h>
+#include <zencore/compactbinaryfile.h>
+#include <zencore/compactbinaryvalue.h>
+#include <zencore/filesystem.h>
+#include <zencore/fmtutils.h>
+#include <zencore/logging.h>
+
+#if ZEN_PLATFORM_WINDOWS
+#	include <ppl.h>
+#	define ZEN_CONCRT_AVAILABLE 1
+#else
+#	define ZEN_CONCRT_AVAILABLE 0
+#endif
+
+#if ZEN_WITH_COMPUTE_SERVICES
+
+namespace zen::compute {
+
+using namespace std::literals;
+
+//////////////////////////////////////////////////////////////////////////
+
+RecordingFileWriter::RecordingFileWriter()
+{
+}
+
+RecordingFileWriter::~RecordingFileWriter()
+{
+	Close();
+}
+
+void
+RecordingFileWriter::Open(std::filesystem::path FilePath)
+{
+	using namespace std::literals;
+
+	m_File.Open(FilePath, BasicFile::Mode::kTruncate);
+	m_File.Write("----DDC2----DATA", 16, 0);
+	m_FileOffset = 16;
+
+	std::filesystem::path TocPath = FilePath.replace_extension(".ztoc");
+	m_TocFile.Open(TocPath, BasicFile::Mode::kTruncate);
+
+	m_TocWriter << "version"sv << 1;
+	m_TocWriter.BeginArray("toc"sv);
+}
+
+void
+RecordingFileWriter::Close()
+{
+	m_TocWriter.EndArray();
+	CbObject Toc = m_TocWriter.Save();
+
+	std::error_code Ec;
+	m_TocFile.WriteAll(Toc.GetBuffer().AsIoBuffer(), Ec);
+}
+
+void
+RecordingFileWriter::AppendObject(const CbObject& Object, const IoHash& ObjectHash)
+{
+	RwLock::ExclusiveLockScope _(m_FileLock);
+
+	MemoryView ObjectView = Object.GetBuffer().GetView();
+
+	std::error_code Ec;
+	m_File.Write(ObjectView, m_FileOffset, Ec);
+
+	if (Ec)
+	{
+		throw std::system_error(Ec, "failed writing to archive");
+	}
+
+	m_TocWriter.BeginArray();
+	m_TocWriter.AddHash(ObjectHash);
+	m_TocWriter.AddInteger(m_FileOffset);
+	m_TocWriter.AddInteger(gsl::narrow<int>(ObjectView.GetSize()));
+	m_TocWriter.EndArray();
+
+	m_FileOffset += ObjectView.GetSize();
+}
+
+//////////////////////////////////////////////////////////////////////////
+
+ActionRecorder::ActionRecorder(ChunkResolver& InChunkResolver, const std::filesystem::path& RecordingLogPath)
+: m_ChunkResolver(InChunkResolver)
+, m_RecordingLogDir(RecordingLogPath)
+{
+	std::error_code Ec;
+	CreateDirectories(m_RecordingLogDir, Ec);
+
+	if (Ec)
+	{
+		ZEN_WARN("Could not create directory '{}': {}", m_RecordingLogDir, Ec.message());
+	}
+
+	CleanDirectory(m_RecordingLogDir, /* ForceRemoveReadOnlyFiles */ true, Ec);
+
+	if (Ec)
+	{
+		ZEN_WARN("Could not clean directory '{}': {}", m_RecordingLogDir, Ec.message());
+	}
+
+	m_WorkersFile.Open(m_RecordingLogDir / "workers.zdat");
+	m_ActionsFile.Open(m_RecordingLogDir / "actions.zdat");
+
+	CidStoreConfiguration CidConfig;
+	CidConfig.RootDirectory		 = m_RecordingLogDir / "cid";
+	CidConfig.HugeValueThreshold = 128 * 1024 * 1024;
+
+	m_CidStore.Initialize(CidConfig);
+}
+
+ActionRecorder::~ActionRecorder()
+{
+	Shutdown();
+}
+
+void
+ActionRecorder::Shutdown()
+{
+	m_CidStore.Flush();
+}
+
+void
+ActionRecorder::RegisterWorker(const CbPackage& WorkerPackage)
+{
+	const IoHash WorkerId = WorkerPackage.GetObjectHash();
+
+	m_WorkersFile.AppendObject(WorkerPackage.GetObject(), WorkerId);
+
+	std::unordered_set<IoHash> AddedChunks;
+	uint64_t				   AddedBytes = 0;
+
+	// First add all attachments from the worker package itself
+
+	for (const CbAttachment& Attachment : WorkerPackage.GetAttachments())
+	{
+		CompressedBuffer Buffer = Attachment.AsCompressedBinary();
+		IoBuffer		 Data	= Buffer.GetCompressed().Flatten().AsIoBuffer();
+
+		const IoHash ChunkHash = Buffer.DecodeRawHash();
+
+		CidStore::InsertResult Result = m_CidStore.AddChunk(Data, ChunkHash, CidStore::InsertMode::kCopyOnly);
+
+		AddedChunks.insert(ChunkHash);
+
+		if (Result.New)
+		{
+			AddedBytes += Data.GetSize();
+		}
+	}
+
+	// Not all attachments will be present in the worker package, so we need to add
+	// all referenced chunks to ensure that the recording is self-contained and not
+	// referencing data in the main CID store
+
+	CbObject WorkerDescriptor = WorkerPackage.GetObject();
+
+	WorkerDescriptor.IterateAttachments([&](const CbFieldView AttachmentField) {
+		const IoHash AttachmentCid = AttachmentField.GetValue().AsHash();
+
+		if (!AddedChunks.contains(AttachmentCid))
+		{
+			IoBuffer AttachmentData = m_ChunkResolver.FindChunkByCid(AttachmentCid);
+
+			if (AttachmentData)
+			{
+				CidStore::InsertResult Result = m_CidStore.AddChunk(AttachmentData, AttachmentCid, CidStore::InsertMode::kCopyOnly);
+
+				if (Result.New)
+				{
+					AddedBytes += AttachmentData.GetSize();
+				}
+			}
+			else
+			{
+				ZEN_WARN("RegisterWorker: could not resolve attachment chunk {} for worker {}", AttachmentCid, WorkerId);
+			}
+
+			AddedChunks.insert(AttachmentCid);
+		}
+	});
+
+	ZEN_INFO("recorded worker {} with {} attachments ({} bytes)", WorkerId, AddedChunks.size(), AddedBytes);
+}
+
+bool
+ActionRecorder::RecordAction(Ref<RunnerAction> Action)
+{
+	bool AllGood = true;
+
+	Action->ActionObj.IterateAttachments([&](CbFieldView Field) {
+		IoHash	 AttachData = Field.AsHash();
+		IoBuffer ChunkData	= m_ChunkResolver.FindChunkByCid(AttachData);
+
+		if (ChunkData)
+		{
+			if (ChunkData.GetContentType() == ZenContentType::kCompressedBinary)
+			{
+				IoHash			 DecompressedHash;
+				uint64_t		 RawSize = 0;
+				CompressedBuffer Compressed =
+					CompressedBuffer::FromCompressed(SharedBuffer(ChunkData), /* out */ DecompressedHash, /* out*/ RawSize);
+
+				OodleCompressor		  Compressor;
+				OodleCompressionLevel CompressionLevel;
+				uint64_t			  BlockSize = 0;
+				if (Compressed.TryGetCompressParameters(/* out */ Compressor, /* out */ CompressionLevel, /* out */ BlockSize))
+				{
+					if (Compressor == OodleCompressor::NotSet)
+					{
+						CompositeBuffer	 Decompressed  = Compressed.DecompressToComposite();
+						CompressedBuffer NewCompressed = CompressedBuffer::Compress(std::move(Decompressed),
+																					OodleCompressor::Mermaid,
+																					OodleCompressionLevel::Fast,
+																					BlockSize);
+
+						ChunkData = NewCompressed.GetCompressed().Flatten().AsIoBuffer();
+					}
+				}
+			}
+
+			const uint64_t ChunkSize = ChunkData.GetSize();
+
+			m_CidStore.AddChunk(ChunkData, AttachData, CidStore::InsertMode::kCopyOnly);
+			++m_ChunkCounter;
+			m_ChunkBytesCounter.fetch_add(ChunkSize);
+		}
+		else
+		{
+			AllGood = false;
+
+			ZEN_WARN("could not resolve chunk {}", AttachData);
+		}
+	});
+
+	if (AllGood)
+	{
+		m_ActionsFile.AppendObject(Action->ActionObj, Action->ActionId);
+		++m_ActionsCounter;
+
+		return true;
+	}
+	else
+	{
+		return false;
+	}
+}
+
+}  // namespace zen::compute
+
+#endif	// ZEN_WITH_COMPUTE_SERVICES