aboutsummaryrefslogtreecommitdiff
path: root/zenserver/upstream/upstreamapply.cpp
diff options
context:
space:
mode:
Diffstat (limited to 'zenserver/upstream/upstreamapply.cpp')
-rw-r--r--zenserver/upstream/upstreamapply.cpp1321
1 files changed, 19 insertions, 1302 deletions
diff --git a/zenserver/upstream/upstreamapply.cpp b/zenserver/upstream/upstreamapply.cpp
index 17a6bb3cf..9758e7565 100644
--- a/zenserver/upstream/upstreamapply.cpp
+++ b/zenserver/upstream/upstreamapply.cpp
@@ -4,1303 +4,26 @@
#if ZEN_WITH_COMPUTE_SERVICES
-# include "jupiter.h"
-# include "zen.h"
-
# include <zencore/compactbinary.h>
# include <zencore/compactbinarybuilder.h>
-# include <zencore/compactbinarypackage.h>
-# include <zencore/compactbinaryvalidation.h>
-# include <zencore/compress.h>
# include <zencore/fmtutils.h>
-# include <zencore/session.h>
-# include <zencore/stats.h>
# include <zencore/stream.h>
-# include <zencore/thread.h>
# include <zencore/timer.h>
# include <zencore/workthreadpool.h>
# include <zenstore/cas.h>
# include <zenstore/cidstore.h>
-# include <auth/authmgr.h>
-# include <upstream/upstreamcache.h>
-
-# include "cache/structuredcachestore.h"
# include "diag/logging.h"
# include <fmt/format.h>
-# include <algorithm>
# include <atomic>
-# include <set>
-# include <stack>
-# include <thread>
-# include <unordered_map>
namespace zen {
using namespace std::literals;
-static const IoBuffer EmptyBuffer;
-static const IoHash EmptyBufferId = IoHash::HashBuffer(EmptyBuffer);
-
-namespace detail {
-
- class HordeUpstreamApplyEndpoint final : public UpstreamApplyEndpoint
- {
- public:
- HordeUpstreamApplyEndpoint(const CloudCacheClientOptions& ComputeOptions,
- const UpstreamAuthConfig& ComputeAuthConfig,
- const CloudCacheClientOptions& StorageOptions,
- const UpstreamAuthConfig& StorageAuthConfig,
- CasStore& CasStore,
- CidStore& CidStore,
- AuthMgr& Mgr)
- : m_Log(logging::Get("upstream-apply"))
- , m_CasStore(CasStore)
- , m_CidStore(CidStore)
- , m_AuthMgr(Mgr)
- {
- m_DisplayName = fmt::format("{} - '{}'+'{}'", ComputeOptions.Name, ComputeOptions.ServiceUrl, StorageOptions.ServiceUrl);
- m_ChannelId = fmt::format("zen-{}", zen::GetSessionIdString());
-
- {
- std::unique_ptr<CloudCacheTokenProvider> TokenProvider;
-
- if (ComputeAuthConfig.OAuthUrl.empty() == false)
- {
- TokenProvider =
- CloudCacheTokenProvider::CreateFromOAuthClientCredentials({.Url = ComputeAuthConfig.OAuthUrl,
- .ClientId = ComputeAuthConfig.OAuthClientId,
- .ClientSecret = ComputeAuthConfig.OAuthClientSecret});
- }
- else if (ComputeAuthConfig.OpenIdProvider.empty() == false)
- {
- TokenProvider =
- CloudCacheTokenProvider::CreateFromCallback([this, ProviderName = std::string(ComputeAuthConfig.OpenIdProvider)]() {
- AuthMgr::OpenIdAccessToken Token = m_AuthMgr.GetOpenIdAccessToken(ProviderName);
- return CloudCacheAccessToken{.Value = Token.AccessToken, .ExpireTime = Token.ExpireTime};
- });
- }
- else
- {
- CloudCacheAccessToken AccessToken{.Value = std::string(ComputeAuthConfig.AccessToken),
- .ExpireTime = CloudCacheAccessToken::TimePoint::max()};
- TokenProvider = CloudCacheTokenProvider::CreateFromStaticToken(AccessToken);
- }
-
- m_Client = new CloudCacheClient(ComputeOptions, std::move(TokenProvider));
- }
-
- {
- std::unique_ptr<CloudCacheTokenProvider> TokenProvider;
-
- if (StorageAuthConfig.OAuthUrl.empty() == false)
- {
- TokenProvider =
- CloudCacheTokenProvider::CreateFromOAuthClientCredentials({.Url = StorageAuthConfig.OAuthUrl,
- .ClientId = StorageAuthConfig.OAuthClientId,
- .ClientSecret = StorageAuthConfig.OAuthClientSecret});
- }
- else if (StorageAuthConfig.OpenIdProvider.empty() == false)
- {
- TokenProvider =
- CloudCacheTokenProvider::CreateFromCallback([this, ProviderName = std::string(StorageAuthConfig.OpenIdProvider)]() {
- AuthMgr::OpenIdAccessToken Token = m_AuthMgr.GetOpenIdAccessToken(ProviderName);
- return CloudCacheAccessToken{.Value = Token.AccessToken, .ExpireTime = Token.ExpireTime};
- });
- }
- else
- {
- CloudCacheAccessToken AccessToken{.Value = std::string(StorageAuthConfig.AccessToken),
- .ExpireTime = CloudCacheAccessToken::TimePoint::max()};
- TokenProvider = CloudCacheTokenProvider::CreateFromStaticToken(AccessToken);
- }
-
- m_StorageClient = new CloudCacheClient(StorageOptions, std::move(TokenProvider));
- }
- }
-
- virtual ~HordeUpstreamApplyEndpoint() = default;
-
- virtual UpstreamEndpointHealth Initialize() override { return CheckHealth(); }
-
- virtual bool IsHealthy() const override { return m_HealthOk.load(); }
-
- virtual UpstreamEndpointHealth CheckHealth() override
- {
- try
- {
- CloudCacheSession Session(m_Client);
- CloudCacheResult Result = Session.Authenticate();
-
- m_HealthOk = Result.ErrorCode == 0;
-
- return {.Reason = std::move(Result.Reason), .Ok = Result.Success};
- }
- catch (std::exception& Err)
- {
- return {.Reason = Err.what(), .Ok = false};
- }
- }
-
- virtual std::string_view DisplayName() const override { return m_DisplayName; }
-
- virtual PostUpstreamApplyResult PostApply(UpstreamApplyRecord ApplyRecord) override
- {
- int64_t Bytes{};
- double ElapsedSeconds{};
-
- try
- {
- UpstreamData UpstreamData;
- if (!ProcessApplyKey(ApplyRecord, UpstreamData))
- {
- return {.Error{.ErrorCode = -1, .Reason = "Failed to generate task data"}};
- }
-
- {
- std::scoped_lock Lock(m_TaskMutex);
- if (m_PendingTasks.contains(UpstreamData.TaskId))
- {
- // Pending task is already queued, return success
- return {.Bytes = Bytes, .ElapsedSeconds = ElapsedSeconds, .Success = true};
- }
- m_PendingTasks[UpstreamData.TaskId] = std::move(ApplyRecord);
- }
-
- CloudCacheSession ComputeSession(m_Client);
- CloudCacheSession StorageSession(m_StorageClient);
-
- {
- CloudCacheResult Result = BatchPutBlobsIfMissing(StorageSession, UpstreamData.Blobs);
- Bytes += Result.Bytes;
- ElapsedSeconds += Result.ElapsedSeconds;
- if (!Result.Success)
- {
- return {.Error{.ErrorCode = Result.ErrorCode ? Result.ErrorCode : -1,
- .Reason = !Result.Reason.empty() ? std::move(Result.Reason) : "Failed to upload blobs"},
- .Bytes = Bytes,
- .ElapsedSeconds = ElapsedSeconds};
- }
- UpstreamData.Blobs.clear();
- }
-
- {
- CloudCacheResult Result = BatchPutObjectsIfMissing(StorageSession, UpstreamData.Objects);
- Bytes += Result.Bytes;
- ElapsedSeconds += Result.ElapsedSeconds;
- if (!Result.Success)
- {
- return {.Error{.ErrorCode = Result.ErrorCode ? Result.ErrorCode : -1,
- .Reason = !Result.Reason.empty() ? std::move(Result.Reason) : "Failed to upload objects"},
- .Bytes = Bytes,
- .ElapsedSeconds = ElapsedSeconds};
- }
-
- PutRefResult RefResult = StorageSession.PutRef("requests"sv,
- UpstreamData.TaskId,
- UpstreamData.Objects[UpstreamData.TaskId].GetBuffer().AsIoBuffer(),
- ZenContentType::kCbObject);
- Log().debug("Put ref {} Need={} Bytes={} Duration={}s Result={}",
- UpstreamData.TaskId,
- RefResult.Needs.size(),
- RefResult.Bytes,
- RefResult.ElapsedSeconds,
- RefResult.Success);
-
- Bytes += RefResult.Bytes;
- ElapsedSeconds += RefResult.ElapsedSeconds;
- if (!RefResult.Success)
- {
- return {.Error{.ErrorCode = RefResult.ErrorCode ? RefResult.ErrorCode : -1,
- .Reason = !RefResult.Reason.empty() ? std::move(RefResult.Reason) : "Failed to add task ref"},
- .Bytes = Bytes,
- .ElapsedSeconds = ElapsedSeconds};
- }
- UpstreamData.Objects.clear();
- }
-
- CbObjectWriter Writer;
- Writer.AddString("c"sv, m_ChannelId);
- Writer.AddObjectAttachment("r"sv, UpstreamData.RequirementsId);
- Writer.BeginArray("t"sv);
- Writer.AddObjectAttachment(UpstreamData.TaskId);
- Writer.EndArray();
- CbObject TasksObject = Writer.Save();
- IoBuffer TasksData = TasksObject.GetBuffer().AsIoBuffer();
-
- CloudCacheResult Result = ComputeSession.PostComputeTasks(TasksData);
- Log().debug("Post compute task {} Bytes={} Duration={}s Result={}",
- TasksObject.GetHash(),
- Result.Bytes,
- Result.ElapsedSeconds,
- Result.Success);
- Bytes += Result.Bytes;
- ElapsedSeconds += Result.ElapsedSeconds;
- if (!Result.Success)
- {
- {
- std::scoped_lock Lock(m_TaskMutex);
- m_PendingTasks.erase(UpstreamData.TaskId);
- }
-
- return {.Error{.ErrorCode = Result.ErrorCode ? Result.ErrorCode : -1,
- .Reason = !Result.Reason.empty() ? std::move(Result.Reason) : "Failed to post compute task"},
- .Bytes = Bytes,
- .ElapsedSeconds = ElapsedSeconds};
- }
-
- Log().info("Task posted {}", UpstreamData.TaskId);
- return {.Bytes = Bytes, .ElapsedSeconds = ElapsedSeconds, .Success = true};
- }
- catch (std::exception& Err)
- {
- m_HealthOk = false;
- return {.Error{.ErrorCode = -1, .Reason = Err.what()}, .Bytes = Bytes, .ElapsedSeconds = ElapsedSeconds};
- }
- }
-
- [[nodiscard]] CloudCacheResult BatchPutBlobsIfMissing(CloudCacheSession& Session, const std::map<IoHash, IoBuffer>& Blobs)
- {
- if (Blobs.size() == 0)
- {
- return {.Success = true};
- }
-
- int64_t Bytes{};
- double ElapsedSeconds{};
-
- // Batch check for missing blobs
- std::set<IoHash> Keys;
- std::transform(Blobs.begin(), Blobs.end(), std::inserter(Keys, Keys.end()), [](const auto& It) { return It.first; });
-
- CloudCacheExistsResult ExistsResult = Session.BlobExists(Keys);
- Log().debug("Queried {} missing blobs Need={} Duration={}s Result={}",
- Keys.size(),
- ExistsResult.Needs.size(),
- ExistsResult.ElapsedSeconds,
- ExistsResult.Success);
- ElapsedSeconds += ExistsResult.ElapsedSeconds;
- if (!ExistsResult.Success)
- {
- return {.Bytes = Bytes,
- .ElapsedSeconds = ElapsedSeconds,
- .ErrorCode = ExistsResult.ErrorCode ? ExistsResult.ErrorCode : -1,
- .Reason = !ExistsResult.Reason.empty() ? std::move(ExistsResult.Reason) : "Failed to check if blobs exist"};
- }
-
- // TODO: Batch upload missing blobs
-
- for (const auto& Hash : ExistsResult.Needs)
- {
- CloudCacheResult Result = Session.PutBlob(Hash, Blobs.at(Hash));
- Log().debug("Put blob {} Bytes={} Duration={}s Result={}", Hash, Result.Bytes, Result.ElapsedSeconds, Result.Success);
- Bytes += Result.Bytes;
- ElapsedSeconds += Result.ElapsedSeconds;
- if (!Result.Success)
- {
- return {.Bytes = Bytes,
- .ElapsedSeconds = ElapsedSeconds,
- .ErrorCode = Result.ErrorCode ? Result.ErrorCode : -1,
- .Reason = !Result.Reason.empty() ? std::move(Result.Reason) : "Failed to put blobs"};
- }
- }
-
- return {.Bytes = Bytes, .ElapsedSeconds = ElapsedSeconds, .Success = true};
- }
-
- [[nodiscard]] CloudCacheResult BatchPutObjectsIfMissing(CloudCacheSession& Session, const std::map<IoHash, CbObject>& Objects)
- {
- if (Objects.size() == 0)
- {
- return {.Success = true};
- }
-
- int64_t Bytes{};
- double ElapsedSeconds{};
-
- // Batch check for missing objects
- std::set<IoHash> Keys;
- std::transform(Objects.begin(), Objects.end(), std::inserter(Keys, Keys.end()), [](const auto& It) { return It.first; });
-
- // Todo: Endpoint doesn't exist for objects
- CloudCacheExistsResult ExistsResult = Session.ObjectExists(Keys);
- Log().debug("Queried {} missing objects Need={} Duration={}s Result={}",
- Keys.size(),
- ExistsResult.Needs.size(),
- ExistsResult.ElapsedSeconds,
- ExistsResult.Success);
- ElapsedSeconds += ExistsResult.ElapsedSeconds;
- if (!ExistsResult.Success)
- {
- return {.Bytes = Bytes,
- .ElapsedSeconds = ElapsedSeconds,
- .ErrorCode = ExistsResult.ErrorCode ? ExistsResult.ErrorCode : -1,
- .Reason = !ExistsResult.Reason.empty() ? std::move(ExistsResult.Reason) : "Failed to check if objects exist"};
- }
-
- // TODO: Batch upload missing objects
-
- for (const auto& Hash : ExistsResult.Needs)
- {
- CloudCacheResult Result = Session.PutObject(Hash, Objects.at(Hash).GetBuffer().AsIoBuffer());
- Log().debug("Put object {} Bytes={} Duration={}s Result={}", Hash, Result.Bytes, Result.ElapsedSeconds, Result.Success);
- Bytes += Result.Bytes;
- ElapsedSeconds += Result.ElapsedSeconds;
- if (!Result.Success)
- {
- return {.Bytes = Bytes,
- .ElapsedSeconds = ElapsedSeconds,
- .ErrorCode = Result.ErrorCode ? Result.ErrorCode : -1,
- .Reason = !Result.Reason.empty() ? std::move(Result.Reason) : "Failed to put objects"};
- }
- }
-
- return {.Bytes = Bytes, .ElapsedSeconds = ElapsedSeconds, .Success = true};
- }
-
- enum class ComputeTaskState : int32_t
- {
- Queued = 0,
- Executing = 1,
- Complete = 2,
- };
-
- enum class ComputeTaskOutcome : int32_t
- {
- Success = 0,
- Failed = 1,
- Cancelled = 2,
- NoResult = 3,
- Exipred = 4,
- BlobNotFound = 5,
- Exception = 6,
- };
-
- [[nodiscard]] static std::string_view ComputeTaskStateToString(const ComputeTaskState Outcome)
- {
- switch (Outcome)
- {
- case ComputeTaskState::Queued:
- return "Queued"sv;
- case ComputeTaskState::Executing:
- return "Executing"sv;
- case ComputeTaskState::Complete:
- return "Complete"sv;
- };
- return "Unknown"sv;
- }
-
- [[nodiscard]] static std::string_view ComputeTaskOutcomeToString(const ComputeTaskOutcome Outcome)
- {
- switch (Outcome)
- {
- case ComputeTaskOutcome::Success:
- return "Success"sv;
- case ComputeTaskOutcome::Failed:
- return "Failed"sv;
- case ComputeTaskOutcome::Cancelled:
- return "Cancelled"sv;
- case ComputeTaskOutcome::NoResult:
- return "NoResult"sv;
- case ComputeTaskOutcome::Exipred:
- return "Exipred"sv;
- case ComputeTaskOutcome::BlobNotFound:
- return "BlobNotFound"sv;
- case ComputeTaskOutcome::Exception:
- return "Exception"sv;
- };
- return "Unknown"sv;
- }
-
- virtual GetUpstreamApplyUpdatesResult GetUpdates(WorkerThreadPool& ThreadPool) override
- {
- int64_t Bytes{};
- double ElapsedSeconds{};
-
- {
- std::scoped_lock Lock(m_TaskMutex);
- if (m_PendingTasks.empty())
- {
- if (m_CompletedTasks.empty())
- {
- // Nothing to do.
- return {.Success = true};
- }
-
- UpstreamApplyCompleted CompletedTasks;
- std::swap(CompletedTasks, m_CompletedTasks);
- return {.Bytes = Bytes, .ElapsedSeconds = ElapsedSeconds, .Completed = std::move(CompletedTasks), .Success = true};
- }
- }
-
- try
- {
- CloudCacheSession ComputeSession(m_Client);
-
- CloudCacheResult UpdatesResult = ComputeSession.GetComputeUpdates(m_ChannelId);
- Log().debug("Get compute updates Bytes={} Duration={}s Result={}",
- UpdatesResult.Bytes,
- UpdatesResult.ElapsedSeconds,
- UpdatesResult.Success);
- Bytes += UpdatesResult.Bytes;
- ElapsedSeconds += UpdatesResult.ElapsedSeconds;
- if (!UpdatesResult.Success)
- {
- return {.Error{.ErrorCode = UpdatesResult.ErrorCode, .Reason = std::move(UpdatesResult.Reason)},
- .Bytes = Bytes,
- .ElapsedSeconds = ElapsedSeconds};
- }
-
- if (!UpdatesResult.Success)
- {
- return {.Error{.ErrorCode = -1, .Reason = "Failed get task updates"}, .Bytes = Bytes, .ElapsedSeconds = ElapsedSeconds};
- }
-
- CbObject TaskStatus = LoadCompactBinaryObject(std::move(UpdatesResult.Response));
-
- for (auto& It : TaskStatus["u"sv])
- {
- CbObjectView Status = It.AsObjectView();
- IoHash TaskId = Status["h"sv].AsHash();
- const ComputeTaskState State = (ComputeTaskState)Status["s"sv].AsInt32();
- const ComputeTaskOutcome Outcome = (ComputeTaskOutcome)Status["o"sv].AsInt32();
-
- Log().info("Task {} State={}", TaskId, ComputeTaskStateToString(State));
-
- // Only completed tasks need to be processed
- if (State != ComputeTaskState::Complete)
- {
- continue;
- }
-
- IoHash WorkerId{};
- IoHash ActionId{};
- UpstreamApplyType ApplyType{};
-
- {
- std::scoped_lock Lock(m_TaskMutex);
- auto TaskIt = m_PendingTasks.find(TaskId);
- if (TaskIt != m_PendingTasks.end())
- {
- WorkerId = TaskIt->second.WorkerDescriptor.GetHash();
- ActionId = TaskIt->second.Action.GetHash();
- ApplyType = TaskIt->second.Type;
- m_PendingTasks.erase(TaskIt);
- }
- }
-
- if (WorkerId == IoHash::Zero)
- {
- Log().warn("Task {} missing from pending tasks", TaskId);
- continue;
- }
-
- if (Outcome != ComputeTaskOutcome::Success)
- {
- const std::string_view Detail = Status["d"sv].AsString();
- {
- std::scoped_lock Lock(m_TaskMutex);
- m_CompletedTasks[WorkerId][ActionId] = {
- .Error{.ErrorCode = -1, .Reason = fmt::format("Task {} {}", ComputeTaskOutcomeToString(Outcome), Detail)}};
- }
- continue;
- }
-
- ThreadPool.ScheduleWork([this,
- ApplyType,
- ResultHash = Status["r"sv].AsHash(),
- TaskId = std::move(TaskId),
- WorkerId = std::move(WorkerId),
- ActionId = std::move(ActionId)]() {
- GetUpstreamApplyResult Result = ProcessTaskStatus(ApplyType, ResultHash);
- Log().debug("Task Processed {} Files={} Attachments={} ExitCode={}",
- TaskId,
- Result.OutputFiles.size(),
- Result.OutputPackage.GetAttachments().size(),
- Result.Error.ErrorCode);
- {
- std::scoped_lock Lock(m_TaskMutex);
- m_CompletedTasks[WorkerId][ActionId] = std::move(Result);
- }
- });
- }
-
- {
- std::scoped_lock Lock(m_TaskMutex);
- if (m_CompletedTasks.empty())
- {
- // Nothing to do.
- return {.Bytes = Bytes, .ElapsedSeconds = ElapsedSeconds, .Success = true};
- }
- UpstreamApplyCompleted CompletedTasks;
- std::swap(CompletedTasks, m_CompletedTasks);
- return {.Bytes = Bytes, .ElapsedSeconds = ElapsedSeconds, .Completed = std::move(CompletedTasks), .Success = true};
- }
- }
- catch (std::exception& Err)
- {
- m_HealthOk = false;
- return {
- .Error{.ErrorCode = -1, .Reason = Err.what()},
- .Bytes = Bytes,
- .ElapsedSeconds = ElapsedSeconds,
- };
- }
- }
-
- virtual UpstreamApplyEndpointStats& Stats() override { return m_Stats; }
-
- private:
- spdlog::logger& Log() { return m_Log; }
-
- spdlog::logger& m_Log;
- CasStore& m_CasStore;
- CidStore& m_CidStore;
- AuthMgr& m_AuthMgr;
- std::string m_DisplayName;
- RefPtr<CloudCacheClient> m_Client;
- RefPtr<CloudCacheClient> m_StorageClient;
- UpstreamApplyEndpointStats m_Stats;
- std::atomic_bool m_HealthOk{false};
- std::string m_ChannelId;
-
- std::mutex m_TaskMutex;
- std::unordered_map<IoHash, UpstreamApplyRecord> m_PendingTasks;
- UpstreamApplyCompleted m_CompletedTasks;
-
- struct UpstreamData
- {
- std::map<IoHash, IoBuffer> Blobs;
- std::map<IoHash, CbObject> Objects;
- IoHash TaskId;
- IoHash RequirementsId;
- };
-
- struct UpstreamDirectory
- {
- std::filesystem::path Path;
- std::map<std::string, UpstreamDirectory> Directories;
- std::set<std::string> Files;
- };
-
- [[nodiscard]] GetUpstreamApplyResult ProcessTaskStatus(const UpstreamApplyType ApplyType, const IoHash& ResultHash)
- {
- try
- {
- CloudCacheSession Session(m_StorageClient);
-
- int64_t Bytes{};
- double ElapsedSeconds{};
-
- // Get Result object and all Object Attachments + Binary Attachment IDs
- CloudCacheResult ObjectRefResult = Session.GetRef("responses"sv, ResultHash, ZenContentType::kCbObject);
- Log().debug("Get ref {} Bytes={} Duration={}s Result={}",
- ResultHash,
- ObjectRefResult.Bytes,
- ObjectRefResult.ElapsedSeconds,
- ObjectRefResult.Success);
- Bytes += ObjectRefResult.Bytes;
- ElapsedSeconds += ObjectRefResult.ElapsedSeconds;
-
- if (!ObjectRefResult.Success)
- {
- return {.Error{.ErrorCode = -1, .Reason = "Failed to get result object data"},
- .Bytes = Bytes,
- .ElapsedSeconds = ElapsedSeconds};
- }
-
- std::vector<IoHash> ObjectsToIterate;
- std::map<IoHash, IoBuffer> ObjectData;
- std::map<IoHash, IoBuffer> BinaryData;
-
- ObjectData[ResultHash] = ObjectRefResult.Response;
- CbObject Object = LoadCompactBinaryObject(ObjectData[ResultHash]);
- Object.IterateAttachments([&](CbFieldView Field) {
- if (Field.IsObjectAttachment())
- {
- const IoHash AttachmentHash = Field.AsObjectAttachment();
- if (!ObjectData.contains(AttachmentHash))
- {
- ObjectsToIterate.push_back(AttachmentHash);
- }
- }
- else if (Field.IsBinaryAttachment())
- {
- const IoHash AttachmentHash = Field.AsBinaryAttachment();
- BinaryData[AttachmentHash] = {};
- }
- });
-
- while (!ObjectsToIterate.empty())
- {
- const IoHash Hash = ObjectsToIterate.back();
- ObjectsToIterate.pop_back();
-
- CloudCacheResult ObjectResult = Session.GetObject(Hash);
- Log().debug("Get object {} Bytes={} Duration={}s Result={}",
- Hash,
- ObjectResult.Bytes,
- ObjectResult.ElapsedSeconds,
- ObjectResult.Success);
- Bytes += ObjectRefResult.Bytes;
- ElapsedSeconds += ObjectRefResult.ElapsedSeconds;
- if (!ObjectResult.Success)
- {
- return {.Error{.ErrorCode = ObjectResult.ErrorCode, .Reason = std::move(ObjectResult.Reason)},
- .Bytes = Bytes,
- .ElapsedSeconds = ElapsedSeconds};
- }
- ObjectData[Hash] = std::move(ObjectResult.Response);
-
- CbObject IterateObject = LoadCompactBinaryObject(ObjectData[Hash]);
- IterateObject.IterateAttachments([&](CbFieldView Field) {
- if (Field.IsObjectAttachment())
- {
- const IoHash AttachmentHash = Field.AsObjectAttachment();
- if (!ObjectData.contains(AttachmentHash))
- {
- ObjectsToIterate.push_back(AttachmentHash);
- }
- }
- else if (Field.IsBinaryAttachment())
- {
- const IoHash AttachmentHash = Field.AsBinaryAttachment();
- BinaryData[AttachmentHash] = {};
- }
- });
- }
-
- // Batch load all binary data
- for (auto& It : BinaryData)
- {
- CloudCacheResult BlobResult = Session.GetBlob(It.first);
- Log().debug("Get blob {} Bytes={} Duration={}s Result={}",
- It.first,
- BlobResult.Bytes,
- BlobResult.ElapsedSeconds,
- BlobResult.Success);
- Bytes += ObjectRefResult.Bytes;
- ElapsedSeconds += ObjectRefResult.ElapsedSeconds;
- if (!BlobResult.Success)
- {
- return {.Error{.ErrorCode = BlobResult.ErrorCode, .Reason = std::move(BlobResult.Reason)},
- .Bytes = Bytes,
- .ElapsedSeconds = ElapsedSeconds};
- }
- It.second = std::move(BlobResult.Response);
- }
-
- CbObject ResultObject = LoadCompactBinaryObject(ObjectData[ResultHash]);
- int32_t ExitCode = ResultObject["e"sv].AsInt32();
- IoHash StdOutHash = ResultObject["so"sv].AsBinaryAttachment();
- IoHash StdErrHash = ResultObject["se"sv].AsBinaryAttachment();
- IoHash OutputHash = ResultObject["o"sv].AsObjectAttachment();
-
- std::string StdOut = std::string((const char*)BinaryData[StdOutHash].GetData(), BinaryData[StdOutHash].GetSize());
- std::string StdErr = std::string((const char*)BinaryData[StdErrHash].GetData(), BinaryData[StdErrHash].GetSize());
-
- if (OutputHash == IoHash::Zero)
- {
- return {.Error{.ErrorCode = ExitCode, .Reason = "Task completed with no output object"},
- .Bytes = Bytes,
- .ElapsedSeconds = ElapsedSeconds,
- .StdOut = std::move(StdOut),
- .StdErr = std::move(StdErr)};
- }
-
- CbObject OutputObject = LoadCompactBinaryObject(ObjectData[OutputHash]);
-
- switch (ApplyType)
- {
- case UpstreamApplyType::Simple:
- {
- std::map<std::filesystem::path, IoHash> OutputFiles;
-
- ResolveMerkleTreeDirectory(""sv, OutputHash, ObjectData, OutputFiles);
-
- return {.OutputFiles = std::move(OutputFiles),
- .FileData = std::move(BinaryData),
- .Bytes = Bytes,
- .ElapsedSeconds = ElapsedSeconds,
- .StdOut = std::move(StdOut),
- .StdErr = std::move(StdErr),
- .Success = true};
- }
- break;
- case UpstreamApplyType::Asset:
- {
- if (ExitCode != 0)
- {
- return {.Error{.ErrorCode = ExitCode, .Reason = "Task completed with errors"},
- .Bytes = Bytes,
- .ElapsedSeconds = ElapsedSeconds,
- .StdOut = std::move(StdOut),
- .StdErr = std::move(StdErr)};
- }
-
- // Get build.output
- IoHash BuildOutputId;
- IoBuffer BuildOutput;
- for (auto& It : OutputObject["f"sv])
- {
- const CbObjectView FileObject = It.AsObjectView();
- if (FileObject["n"sv].AsString() == "Build.output"sv)
- {
- BuildOutputId = FileObject["h"sv].AsBinaryAttachment();
- BuildOutput = BinaryData[BuildOutputId];
- break;
- }
- }
-
- if (BuildOutput.GetSize() == 0)
- {
- return {.Error{.ErrorCode = ExitCode, .Reason = "Build.output file not found in task results"},
- .Bytes = Bytes,
- .ElapsedSeconds = ElapsedSeconds,
- .StdOut = std::move(StdOut),
- .StdErr = std::move(StdErr)};
- }
-
- // Get Output directory node
- IoBuffer OutputDirectoryTree;
- for (auto& It : OutputObject["d"sv])
- {
- const CbObjectView DirectoryObject = It.AsObjectView();
- if (DirectoryObject["n"sv].AsString() == "Outputs"sv)
- {
- OutputDirectoryTree = ObjectData[DirectoryObject["h"sv].AsObjectAttachment()];
- break;
- }
- }
-
- if (OutputDirectoryTree.GetSize() == 0)
- {
- return {.Error{.ErrorCode = ExitCode, .Reason = "Outputs directory not found in task results"},
- .Bytes = Bytes,
- .ElapsedSeconds = ElapsedSeconds,
- .StdOut = std::move(StdOut),
- .StdErr = std::move(StdErr)};
- }
-
- // load build.output as CbObject
-
- // Move Outputs from Horde to CbPackage
-
- std::unordered_map<IoHash, IoHash> CidToCompressedId;
- CbPackage OutputPackage;
- CbObject OutputDirectoryTreeObject = LoadCompactBinaryObject(OutputDirectoryTree);
- int64_t TotalAttachmentBytes = 0;
- int64_t TotalRawAttachmentBytes = 0;
-
- for (auto& It : OutputDirectoryTreeObject["f"sv])
- {
- CbObjectView FileObject = It.AsObjectView();
- // Name is the uncompressed hash
- IoHash DecompressedId = IoHash::FromHexString(FileObject["n"sv].AsString());
- // Hash is the compressed data hash, and how it is stored in Horde
- IoHash CompressedId = FileObject["h"sv].AsBinaryAttachment();
-
- if (!BinaryData.contains(CompressedId))
- {
- Log().warn("Object attachment chunk not retrieved from Horde {}", CompressedId);
- return {.Error{.ErrorCode = -1, .Reason = "Object attachment chunk not retrieved from Horde"},
- .Bytes = Bytes,
- .ElapsedSeconds = ElapsedSeconds};
- }
- CidToCompressedId[DecompressedId] = CompressedId;
- }
-
- // Iterate attachments, verify all chunks exist, and add to CbPackage
- bool AnyErrors = false;
- CbObject BuildOutputObject = LoadCompactBinaryObject(BuildOutput);
- BuildOutputObject.IterateAttachments([&](CbFieldView Field) {
- const IoHash DecompressedId = Field.AsHash();
- if (!CidToCompressedId.contains(DecompressedId))
- {
- Log().warn("Attachment not found {}", DecompressedId);
- AnyErrors = true;
- return;
- }
- const IoHash& CompressedId = CidToCompressedId.at(DecompressedId);
-
- if (!BinaryData.contains(CompressedId))
- {
- Log().warn("Missing output {} compressed {} uncompressed", CompressedId, DecompressedId);
- AnyErrors = true;
- return;
- }
-
- CompressedBuffer AttachmentBuffer =
- CompressedBuffer::FromCompressed(SharedBuffer(BinaryData[CompressedId]));
-
- if (!AttachmentBuffer)
- {
- Log().warn(
- "Invalid output encountered (not valid CompressedBuffer format) {} compressed {} uncompressed",
- CompressedId,
- DecompressedId);
- AnyErrors = true;
- return;
- }
-
- TotalAttachmentBytes += AttachmentBuffer.GetCompressedSize();
- TotalRawAttachmentBytes += AttachmentBuffer.GetRawSize();
-
- CbAttachment Attachment(AttachmentBuffer);
- OutputPackage.AddAttachment(Attachment);
- });
-
- if (AnyErrors)
- {
- return {.Error{.ErrorCode = -1, .Reason = "Failed to get result object attachment data"},
- .Bytes = Bytes,
- .ElapsedSeconds = ElapsedSeconds,
- .StdOut = std::move(StdOut),
- .StdErr = std::move(StdErr)};
- }
-
- OutputPackage.SetObject(BuildOutputObject);
-
- return {.OutputPackage = std::move(OutputPackage),
- .TotalAttachmentBytes = TotalAttachmentBytes,
- .TotalRawAttachmentBytes = TotalRawAttachmentBytes,
- .Bytes = Bytes,
- .ElapsedSeconds = ElapsedSeconds,
- .StdOut = std::move(StdOut),
- .StdErr = std::move(StdErr),
- .Success = true};
- }
- break;
- }
-
- return {.Error{.ErrorCode = ExitCode, .Reason = "Unknown apply type"},
- .Bytes = Bytes,
- .ElapsedSeconds = ElapsedSeconds,
- .StdOut = std::move(StdOut),
- .StdErr = std::move(StdErr)};
- }
- catch (std::exception& Err)
- {
- return {.Error{.ErrorCode = -1, .Reason = Err.what()}};
- }
- }
-
- [[nodiscard]] bool ProcessApplyKey(const UpstreamApplyRecord& ApplyRecord, UpstreamData& Data)
- {
- std::string ExecutablePath;
- std::string WorkingDirectory;
- std::vector<std::string> Arguments;
- std::map<std::string, std::string> Environment;
- std::set<std::filesystem::path> InputFiles;
- std::set<std::string> Outputs;
- std::map<std::filesystem::path, IoHash> InputFileHashes;
-
- ExecutablePath = ApplyRecord.WorkerDescriptor["path"sv].AsString();
- if (ExecutablePath.empty())
- {
- Log().warn("process apply upstream FAILED, '{}', path missing from worker descriptor",
- ApplyRecord.WorkerDescriptor.GetHash());
- return false;
- }
-
- WorkingDirectory = ApplyRecord.WorkerDescriptor["workdir"sv].AsString();
-
- for (auto& It : ApplyRecord.WorkerDescriptor["executables"sv])
- {
- CbObjectView FileEntry = It.AsObjectView();
- if (!ProcessFileEntry(FileEntry, InputFiles, InputFileHashes, Data.Blobs))
- {
- return false;
- }
- }
-
- for (auto& It : ApplyRecord.WorkerDescriptor["files"sv])
- {
- CbObjectView FileEntry = It.AsObjectView();
- if (!ProcessFileEntry(FileEntry, InputFiles, InputFileHashes, Data.Blobs))
- {
- return false;
- }
- }
-
- for (auto& It : ApplyRecord.WorkerDescriptor["dirs"sv])
- {
- std::string_view Directory = It.AsString();
- std::string DummyFile = fmt::format("{}/.zen_empty_file", Directory);
- InputFiles.insert(DummyFile);
- Data.Blobs[EmptyBufferId] = EmptyBuffer;
- InputFileHashes[DummyFile] = EmptyBufferId;
- }
-
- for (auto& It : ApplyRecord.WorkerDescriptor["environment"sv])
- {
- std::string_view Env = It.AsString();
- auto Index = Env.find('=');
- if (Index == std::string_view::npos)
- {
- Log().warn("process apply upstream FAILED, environment '{}' malformed", Env);
- return false;
- }
-
- Environment[std::string(Env.substr(0, Index))] = Env.substr(Index + 1);
- }
-
- switch (ApplyRecord.Type)
- {
- case UpstreamApplyType::Simple:
- {
- for (auto& It : ApplyRecord.WorkerDescriptor["arguments"sv])
- {
- Arguments.push_back(std::string(It.AsString()));
- }
-
- for (auto& It : ApplyRecord.WorkerDescriptor["outputs"sv])
- {
- Outputs.insert(std::string(It.AsString()));
- }
- }
- break;
- case UpstreamApplyType::Asset:
- {
- static const std::filesystem::path BuildActionPath = "Build.action"sv;
- static const std::filesystem::path InputPath = "Inputs"sv;
- const IoHash ActionId = ApplyRecord.Action.GetHash();
-
- Arguments.push_back("-Build=build.action");
- Outputs.insert("Build.output");
- Outputs.insert("Outputs");
-
- InputFiles.insert(BuildActionPath);
- InputFileHashes[BuildActionPath] = ActionId;
- Data.Blobs[ActionId] = IoBufferBuilder::MakeCloneFromMemory(ApplyRecord.Action.GetBuffer().GetData(),
- ApplyRecord.Action.GetBuffer().GetSize());
-
- bool AnyErrors = false;
- ApplyRecord.Action.IterateAttachments([&](CbFieldView Field) {
- const IoHash Cid = Field.AsHash();
- const std::filesystem::path FilePath = {InputPath / Cid.ToHexString()};
- IoBuffer DataBuffer = m_CidStore.FindChunkByCid(Cid);
-
- if (!DataBuffer)
- {
- Log().warn("process apply upstream FAILED, input CID chunk '{}' missing", Cid);
- AnyErrors = true;
- return;
- }
-
- if (InputFiles.contains(FilePath))
- {
- return;
- }
-
- const IoHash CompressedId = IoHash::HashBuffer(DataBuffer.GetData(), DataBuffer.GetSize());
-
- InputFiles.insert(FilePath);
- InputFileHashes[FilePath] = CompressedId;
- Data.Blobs[CompressedId] = std::move(DataBuffer);
- });
-
- if (AnyErrors)
- {
- return false;
- }
- }
- break;
- }
-
- const UpstreamDirectory RootDirectory = BuildDirectoryTree(InputFiles);
-
- CbObject Sandbox = BuildMerkleTreeDirectory(RootDirectory, InputFileHashes, Data.Blobs, Data.Objects);
- const IoHash SandboxHash = Sandbox.GetHash();
- Data.Objects[SandboxHash] = std::move(Sandbox);
-
- {
- std::string_view HostPlatform = ApplyRecord.WorkerDescriptor["host"sv].AsString();
- if (HostPlatform.empty())
- {
- Log().warn("process apply upstream FAILED, 'host' platform not provided");
- return false;
- }
-
- int32_t LogicalCores = ApplyRecord.WorkerDescriptor["cores"sv].AsInt32();
- int64_t Memory = ApplyRecord.WorkerDescriptor["memory"sv].AsInt64();
- bool Exclusive = ApplyRecord.WorkerDescriptor["exclusive"sv].AsBool();
-
- std::string Condition = fmt::format("Platform == '{}'", HostPlatform);
- if (HostPlatform == "Win64")
- {
- // TODO
- // Condition += " && Pool == 'Win-RemoteExec'";
- }
-
- std::map<std::string_view, int64_t> Resources;
- if (LogicalCores > 0)
- {
- Resources["LogicalCores"sv] = LogicalCores;
- }
- if (Memory > 0)
- {
- Resources["RAM"sv] = std::max(Memory / 1024LL / 1024LL / 1024LL, 1LL);
- }
-
- CbObject Requirements = BuildRequirements(Condition, Resources, Exclusive);
- const IoHash RequirementsId = Requirements.GetHash();
- Data.Objects[RequirementsId] = std::move(Requirements);
- Data.RequirementsId = RequirementsId;
- }
-
- CbObject Task = BuildTask(ExecutablePath, Arguments, Environment, WorkingDirectory, SandboxHash, Data.RequirementsId, Outputs);
-
- const IoHash TaskId = Task.GetHash();
- Data.Objects[TaskId] = std::move(Task);
- Data.TaskId = TaskId;
-
- return true;
- }
-
- [[nodiscard]] bool ProcessFileEntry(const CbObjectView& FileEntry,
- std::set<std::filesystem::path>& InputFiles,
- std::map<std::filesystem::path, IoHash>& InputFileHashes,
- std::map<IoHash, IoBuffer>& Blobs)
- {
- const std::filesystem::path FilePath = FileEntry["name"sv].AsString();
- const IoHash ChunkId = FileEntry["hash"sv].AsHash();
- const uint64_t Size = FileEntry["size"sv].AsUInt64();
- IoBuffer DataBuffer = m_CasStore.FindChunk(ChunkId);
-
- if (!DataBuffer)
- {
- Log().warn("process apply upstream FAILED, worker CAS chunk '{}' missing", ChunkId);
- return false;
- }
-
- if (DataBuffer.Size() != Size)
- {
- Log().warn("process apply upstream FAILED, worker CAS chunk '{}' size: {}, action spec expected {}",
- ChunkId,
- DataBuffer.Size(),
- Size);
- return false;
- }
-
- if (InputFiles.contains(FilePath))
- {
- Log().warn("process apply upstream FAILED, worker CAS chunk '{}' size: {} duplicate filename {}", ChunkId, Size, FilePath);
- return false;
- }
-
- InputFiles.insert(FilePath);
- InputFileHashes[FilePath] = ChunkId;
- Blobs[ChunkId] = std::move(DataBuffer);
- return true;
- }
-
- [[nodiscard]] UpstreamDirectory BuildDirectoryTree(const std::set<std::filesystem::path>& InputFiles)
- {
- static const std::filesystem::path RootPath;
- std::map<std::filesystem::path, UpstreamDirectory*> AllDirectories;
- UpstreamDirectory RootDirectory = {.Path = RootPath};
-
- AllDirectories[RootPath] = &RootDirectory;
-
- // Build tree from flat list
- for (const auto& Path : InputFiles)
- {
- if (Path.has_parent_path())
- {
- if (!AllDirectories.contains(Path.parent_path()))
- {
- std::stack<std::string> PathSplit;
- {
- std::filesystem::path ParentPath = Path.parent_path();
- PathSplit.push(ParentPath.filename().string());
- while (ParentPath.has_parent_path())
- {
- ParentPath = ParentPath.parent_path();
- PathSplit.push(ParentPath.filename().string());
- }
- }
- UpstreamDirectory* ParentPtr = &RootDirectory;
- while (!PathSplit.empty())
- {
- if (!ParentPtr->Directories.contains(PathSplit.top()))
- {
- std::filesystem::path NewParentPath = {ParentPtr->Path / PathSplit.top()};
- ParentPtr->Directories[PathSplit.top()] = {.Path = NewParentPath};
- AllDirectories[NewParentPath] = &ParentPtr->Directories[PathSplit.top()];
- }
- ParentPtr = &ParentPtr->Directories[PathSplit.top()];
- PathSplit.pop();
- }
- }
-
- AllDirectories[Path.parent_path()]->Files.insert(Path.filename().string());
- }
- else
- {
- RootDirectory.Files.insert(Path.filename().string());
- }
- }
-
- return RootDirectory;
- }
-
- [[nodiscard]] CbObject BuildMerkleTreeDirectory(const UpstreamDirectory& RootDirectory,
- const std::map<std::filesystem::path, IoHash>& InputFileHashes,
- const std::map<IoHash, IoBuffer>& Blobs,
- std::map<IoHash, CbObject>& Objects)
- {
- CbObjectWriter DirectoryTreeWriter;
-
- if (!RootDirectory.Files.empty())
- {
- DirectoryTreeWriter.BeginArray("f"sv);
- for (const auto& File : RootDirectory.Files)
- {
- const std::filesystem::path FilePath = {RootDirectory.Path / File};
- const IoHash& FileHash = InputFileHashes.at(FilePath);
- const uint64_t FileSize = Blobs.at(FileHash).Size();
- DirectoryTreeWriter.BeginObject();
- DirectoryTreeWriter.AddString("n"sv, File);
- DirectoryTreeWriter.AddBinaryAttachment("h"sv, FileHash);
- DirectoryTreeWriter.AddInteger("s"sv, FileSize); // Size
- // DirectoryTreeWriter.AddInteger("a"sv, 0); // Attributes Currently unneeded
- DirectoryTreeWriter.EndObject();
- }
- DirectoryTreeWriter.EndArray();
- }
-
- if (!RootDirectory.Directories.empty())
- {
- DirectoryTreeWriter.BeginArray("d"sv);
- for (const auto& Item : RootDirectory.Directories)
- {
- CbObject Directory = BuildMerkleTreeDirectory(Item.second, InputFileHashes, Blobs, Objects);
- const IoHash DirectoryHash = Directory.GetHash();
- Objects[DirectoryHash] = std::move(Directory);
-
- DirectoryTreeWriter.BeginObject();
- DirectoryTreeWriter.AddString("n"sv, Item.first);
- DirectoryTreeWriter.AddObjectAttachment("h"sv, DirectoryHash);
- DirectoryTreeWriter.EndObject();
- }
- DirectoryTreeWriter.EndArray();
- }
-
- return DirectoryTreeWriter.Save();
- }
-
- void ResolveMerkleTreeDirectory(const std::filesystem::path& ParentDirectory,
- const IoHash& DirectoryHash,
- const std::map<IoHash, IoBuffer>& Objects,
- std::map<std::filesystem::path, IoHash>& OutputFiles)
- {
- CbObject Directory = LoadCompactBinaryObject(Objects.at(DirectoryHash));
-
- for (auto& It : Directory["f"sv])
- {
- const CbObjectView FileObject = It.AsObjectView();
- const std::filesystem::path Path = ParentDirectory / FileObject["n"sv].AsString();
-
- OutputFiles[Path] = FileObject["h"sv].AsBinaryAttachment();
- }
-
- for (auto& It : Directory["d"sv])
- {
- const CbObjectView DirectoryObject = It.AsObjectView();
-
- ResolveMerkleTreeDirectory(ParentDirectory / DirectoryObject["n"sv].AsString(),
- DirectoryObject["h"sv].AsObjectAttachment(),
- Objects,
- OutputFiles);
- }
- }
-
- [[nodiscard]] CbObject BuildRequirements(const std::string_view Condition,
- const std::map<std::string_view, int64_t>& Resources,
- const bool Exclusive)
- {
- CbObjectWriter Writer;
- Writer.AddString("c", Condition);
- if (!Resources.empty())
- {
- Writer.BeginArray("r");
- for (const auto& Resource : Resources)
- {
- Writer.BeginArray();
- Writer.AddString(Resource.first);
- Writer.AddInteger(Resource.second);
- Writer.EndArray();
- }
- Writer.EndArray();
- }
- Writer.AddBool("e", Exclusive);
- return Writer.Save();
- }
-
- [[nodiscard]] CbObject BuildTask(const std::string_view Executable,
- const std::vector<std::string>& Arguments,
- const std::map<std::string, std::string>& Environment,
- const std::string_view WorkingDirectory,
- const IoHash& SandboxHash,
- const IoHash& RequirementsId,
- const std::set<std::string>& Outputs)
- {
- CbObjectWriter TaskWriter;
- TaskWriter.AddString("e"sv, Executable);
-
- if (!Arguments.empty())
- {
- TaskWriter.BeginArray("a"sv);
- for (const auto& Argument : Arguments)
- {
- TaskWriter.AddString(Argument);
- }
- TaskWriter.EndArray();
- }
-
- if (!Environment.empty())
- {
- TaskWriter.BeginArray("v"sv);
- for (const auto& Env : Environment)
- {
- TaskWriter.BeginArray();
- TaskWriter.AddString(Env.first);
- TaskWriter.AddString(Env.second);
- TaskWriter.EndArray();
- }
- TaskWriter.EndArray();
- }
-
- if (!WorkingDirectory.empty())
- {
- TaskWriter.AddString("w"sv, WorkingDirectory);
- }
-
- TaskWriter.AddObjectAttachment("s"sv, SandboxHash);
- TaskWriter.AddObjectAttachment("r"sv, RequirementsId);
-
- // Outputs
- if (!Outputs.empty())
- {
- TaskWriter.BeginArray("o"sv);
- for (const auto& Output : Outputs)
- {
- TaskWriter.AddString(Output);
- }
- TaskWriter.EndArray();
- }
-
- return TaskWriter.Save();
- }
- };
-} // namespace detail
-
-//////////////////////////////////////////////////////////////////////////
-
struct UpstreamApplyStats
{
static constexpr uint64_t MaxSampleCount = 1000ull;
@@ -1360,7 +83,8 @@ public:
, m_CasStore(CasStore)
, m_CidStore(CidStore)
, m_Stats(Options.StatsEnabled)
- , m_AsyncWorkPool(Options.ThreadCount)
+ , m_UpstreamAsyncWorkPool(Options.UpstreamThreadCount)
+ , m_DownstreamAsyncWorkPool(Options.DownstreamThreadCount)
{
}
@@ -1423,7 +147,9 @@ public:
m_ApplyTasks[WorkerId][ActionId] = {.State = UpstreamApplyState::Queued, .Result{}, .ExpireTime = std::move(ExpireTime)};
}
- m_AsyncWorkPool.ScheduleWork([this, ApplyRecord = std::move(ApplyRecord)]() { ProcessApplyRecord(std::move(ApplyRecord)); });
+ ApplyRecord.Timepoints["zen-queue-added"] = DateTime::NowTicks();
+ m_UpstreamAsyncWorkPool.ScheduleWork(
+ [this, ApplyRecord = std::move(ApplyRecord)]() { ProcessApplyRecord(std::move(ApplyRecord)); });
return {.ApplyId = ActionId, .Success = true};
}
@@ -1447,8 +173,10 @@ public:
virtual void GetStatus(CbObjectWriter& Status) override
{
- Status << "worker_threads" << m_Options.ThreadCount;
- Status << "queue_count" << m_AsyncWorkPool.PendingWork();
+ Status << "upstream_worker_threads" << m_Options.UpstreamThreadCount;
+ Status << "upstream_queue_count" << m_UpstreamAsyncWorkPool.PendingWork();
+ Status << "downstream_worker_threads" << m_Options.DownstreamThreadCount;
+ Status << "downstream_queue_count" << m_DownstreamAsyncWorkPool.PendingWork();
Status.BeginArray("endpoints");
for (const auto& Ep : m_Endpoints)
@@ -1501,11 +229,14 @@ private:
{
if (Endpoint->IsHealthy())
{
- PostUpstreamApplyResult Result = Endpoint->PostApply(std::move(ApplyRecord));
+ ApplyRecord.Timepoints["zen-queue-dispatched"] = DateTime::NowTicks();
+ PostUpstreamApplyResult Result = Endpoint->PostApply(std::move(ApplyRecord));
{
std::scoped_lock Lock(m_ApplyTasksMutex);
if (auto Status = FindStatus(WorkerId, ActionId); Status != nullptr)
{
+ Status->Timepoints.merge(Result.Timepoints);
+
if (Result.Success)
{
Status->State = UpstreamApplyState::Executing;
@@ -1553,7 +284,7 @@ private:
{
if (Endpoint->IsHealthy())
{
- GetUpstreamApplyUpdatesResult Result = Endpoint->GetUpdates(m_AsyncWorkPool);
+ GetUpstreamApplyUpdatesResult Result = Endpoint->GetUpdates(m_DownstreamAsyncWorkPool);
m_Stats.Add(*Endpoint, Result);
if (!Result.Success)
@@ -1572,6 +303,9 @@ private:
{
Status->State = UpstreamApplyState::Complete;
Status->Result = std::move(It2.second);
+ Status->Result.Timepoints.merge(Status->Timepoints);
+ Status->Result.Timepoints["zen-queue-complete"] = DateTime::NowTicks();
+ Status->Timepoints.clear();
}
}
}
@@ -1686,7 +420,8 @@ private:
std::mutex m_ApplyTasksMutex;
std::vector<std::unique_ptr<UpstreamApplyEndpoint>> m_Endpoints;
Event m_ShutdownEvent;
- WorkerThreadPool m_AsyncWorkPool;
+ WorkerThreadPool m_UpstreamAsyncWorkPool;
+ WorkerThreadPool m_DownstreamAsyncWorkPool;
std::thread m_UpstreamUpdatesThread;
std::thread m_EndpointMonitorThread;
RunState m_RunState;
@@ -1700,24 +435,6 @@ UpstreamApply::Create(const UpstreamApplyOptions& Options, CasStore& CasStore, C
return std::make_unique<UpstreamApplyImpl>(Options, CasStore, CidStore);
}
-std::unique_ptr<UpstreamApplyEndpoint>
-UpstreamApplyEndpoint::CreateHordeEndpoint(const CloudCacheClientOptions& ComputeOptions,
- const UpstreamAuthConfig& ComputeAuthConfig,
- const CloudCacheClientOptions& StorageOptions,
- const UpstreamAuthConfig& StorageAuthConfig,
- CasStore& CasStore,
- CidStore& CidStore,
- AuthMgr& Mgr)
-{
- return std::make_unique<detail::HordeUpstreamApplyEndpoint>(ComputeOptions,
- ComputeAuthConfig,
- StorageOptions,
- StorageAuthConfig,
- CasStore,
- CidStore,
- Mgr);
-}
-
} // namespace zen
#endif // ZEN_WITH_COMPUTE_SERVICES
generated by cgit v1.2.3 (git 2.25.1) at 2026-04-15 15:41:20 +0000