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, 112 insertions, 0 deletions

diff --git a/src/zencompute/functionrunner.cpp b/src/zencompute/functionrunner.cpp
new file mode 100644
index 000000000..8e7c12b2b
--- /dev/null
+++ b/src/zencompute/functionrunner.cpp
@@ -0,0 +1,112 @@
+// Copyright Epic Games, Inc. All Rights Reserved.
+
+#include "functionrunner.h"
+
+#if ZEN_WITH_COMPUTE_SERVICES
+
+#	include <zencore/compactbinary.h>
+#	include <zencore/filesystem.h>
+
+#	include <fmt/format.h>
+#	include <vector>
+
+namespace zen::compute {
+
+FunctionRunner::FunctionRunner(std::filesystem::path BasePath) : m_ActionsPath(BasePath / "actions")
+{
+}
+
+FunctionRunner::~FunctionRunner() = default;
+
+size_t
+FunctionRunner::QueryCapacity()
+{
+	return 1;
+}
+
+std::vector<SubmitResult>
+FunctionRunner::SubmitActions(const std::vector<Ref<RunnerAction>>& Actions)
+{
+	std::vector<SubmitResult> Results;
+	Results.reserve(Actions.size());
+
+	for (const Ref<RunnerAction>& Action : Actions)
+	{
+		Results.push_back(SubmitAction(Action));
+	}
+
+	return Results;
+}
+
+void
+FunctionRunner::MaybeDumpAction(int ActionLsn, const CbObject& ActionObject)
+{
+	if (m_DumpActions)
+	{
+		std::string			  UniqueId = fmt::format("{}.ddb", ActionLsn);
+		std::filesystem::path Path	   = m_ActionsPath / UniqueId;
+
+		zen::WriteFile(Path, IoBuffer(ActionObject.GetBuffer().AsIoBuffer()));
+	}
+}
+
+//////////////////////////////////////////////////////////////////////////
+
+RunnerAction::RunnerAction(FunctionServiceSession* OwnerSession) : m_OwnerSession(OwnerSession)
+{
+	this->Timestamps[static_cast<int>(State::New)] = DateTime::Now().GetTicks();
+}
+
+RunnerAction::~RunnerAction()
+{
+}
+
+void
+RunnerAction::SetActionState(State NewState)
+{
+	ZEN_ASSERT(NewState < State::_Count);
+	this->Timestamps[static_cast<int>(NewState)] = DateTime::Now().GetTicks();
+
+	do
+	{
+		if (State CurrentState = m_ActionState.load(); CurrentState == NewState)
+		{
+			// No state change
+			return;
+		}
+		else
+		{
+			if (NewState <= CurrentState)
+			{
+				// Cannot transition to an earlier or same state
+				return;
+			}
+
+			if (m_ActionState.compare_exchange_strong(CurrentState, NewState))
+			{
+				// Successful state change
+
+				m_OwnerSession->PostUpdate(this);
+
+				return;
+			}
+		}
+	} while (true);
+}
+
+void
+RunnerAction::SetResult(CbPackage&& Result)
+{
+	m_Result = std::move(Result);
+}
+
+CbPackage&
+RunnerAction::GetResult()
+{
+	ZEN_ASSERT(IsCompleted());
+	return m_Result;
+}
+
+}  // namespace zen::compute
+
+#endif	// ZEN_WITH_COMPUTE_SERVICES
\ No newline at end of file