From 4b8d4c0e375c729e38bdaadfebf0eaf14f08f5f9 Mon Sep 17 00:00:00 2001 From: Per Larsson Date: Tue, 2 Nov 2021 10:50:18 +0100 Subject: Added upstream batch API. --- zenserver/upstream/upstreamcache.cpp | 41 ++++++++++++++++++++++++++++++++++++ 1 file changed, 41 insertions(+) (limited to 'zenserver/upstream/upstreamcache.cpp') diff --git a/zenserver/upstream/upstreamcache.cpp b/zenserver/upstream/upstreamcache.cpp index c0cd858b6..437b29cd7 100644 --- a/zenserver/upstream/upstreamcache.cpp +++ b/zenserver/upstream/upstreamcache.cpp @@ -780,6 +780,47 @@ public: return {}; } + virtual GetUpstreamCacheBatchResult GetCacheRecords(std::span CacheKeys, + std::span KeyIndex, + OnCacheGetComplete OnComplete) override + { + if (!m_Options.ReadUpstream) + { + return {.Missing = std::vector(KeyIndex.begin(), KeyIndex.end())}; + } + + GetUpstreamCacheBatchResult Result; + + for (size_t Idx : KeyIndex) + { + const UpstreamCacheKey CacheKey = {CacheKeys[Idx].Bucket, CacheKeys[Idx].Hash}; + + GetUpstreamCacheResult CacheResult; + for (auto& Endpoint : m_Endpoints) + { + if (Endpoint->IsHealthy()) + { + CacheResult = Endpoint->GetCacheRecord(CacheKey, ZenContentType::kCbPackage); + if (CacheResult.Success) + { + break; + } + } + } + + if (CacheResult.Success) + { + OnComplete(Idx, CacheResult.Value); + } + else + { + Result.Missing.push_back(Idx); + } + } + + return Result; + } + virtual GetUpstreamCacheResult GetCachePayload(UpstreamPayloadKey PayloadKey) override { if (m_Options.ReadUpstream) -- cgit v1.2.3 From e0d54396fa3ba0f5466a4ea1f2810721c18fa55f Mon Sep 17 00:00:00 2001 From: Per Larsson Date: Tue, 9 Nov 2021 13:20:00 +0100 Subject: Sort cache keys when resolving payload ID's. --- zenserver/upstream/upstreamcache.cpp | 48 +++++++++++++++++++++++++++++++++++- 1 file changed, 47 insertions(+), 1 deletion(-) (limited to 'zenserver/upstream/upstreamcache.cpp') diff --git a/zenserver/upstream/upstreamcache.cpp b/zenserver/upstream/upstreamcache.cpp index 437b29cd7..7ef0caf62 100644 --- a/zenserver/upstream/upstreamcache.cpp +++ b/zenserver/upstream/upstreamcache.cpp @@ -782,7 +782,7 @@ public: virtual GetUpstreamCacheBatchResult GetCacheRecords(std::span CacheKeys, std::span KeyIndex, - OnCacheGetComplete OnComplete) override + OnCacheGetComplete OnComplete) override final { if (!m_Options.ReadUpstream) { @@ -801,6 +801,52 @@ public: if (Endpoint->IsHealthy()) { CacheResult = Endpoint->GetCacheRecord(CacheKey, ZenContentType::kCbPackage); + m_Stats.Add(m_Log, *Endpoint, CacheResult, m_Endpoints); + + if (CacheResult.Success) + { + break; + } + } + } + + if (CacheResult.Success) + { + OnComplete(Idx, CacheResult.Value); + } + else + { + Result.Missing.push_back(Idx); + } + } + + return Result; + } + + virtual GetUpstreamCacheBatchResult GetCachePayloads(std::span CacheChunkRequests, + std::span ChunkIndex, + OnCacheGetComplete OnComplete) override final + { + if (!m_Options.ReadUpstream) + { + return {.Missing = std::vector(ChunkIndex.begin(), ChunkIndex.end())}; + } + + GetUpstreamCacheBatchResult Result; + + for (size_t Idx : ChunkIndex) + { + const CacheChunkRequest& Chunk = CacheChunkRequests[Idx]; + UpstreamPayloadKey PayloadKey{{Chunk.Key.Bucket, Chunk.Key.Hash}, Chunk.ChunkId}; + + GetUpstreamCacheResult CacheResult; + for (auto& Endpoint : m_Endpoints) + { + if (Endpoint->IsHealthy()) + { + CacheResult = Endpoint->GetCachePayload(PayloadKey); + m_Stats.Add(m_Log, *Endpoint, CacheResult, m_Endpoints); + if (CacheResult.Success) { break; -- cgit v1.2.3 From 2c0e2ab5de21b13dcd25758ca3b96af889db7137 Mon Sep 17 00:00:00 2001 From: Per Larsson Date: Thu, 11 Nov 2021 11:19:17 +0100 Subject: Added batch API to upstream endpoints. --- zenserver/upstream/upstreamcache.cpp | 346 ++++++++++++++++++++++++++++------- 1 file changed, 276 insertions(+), 70 deletions(-) (limited to 'zenserver/upstream/upstreamcache.cpp') diff --git a/zenserver/upstream/upstreamcache.cpp b/zenserver/upstream/upstreamcache.cpp index 7ef0caf62..0a0706656 100644 --- a/zenserver/upstream/upstreamcache.cpp +++ b/zenserver/upstream/upstreamcache.cpp @@ -70,7 +70,7 @@ namespace detail { virtual std::string_view DisplayName() const override { return m_DisplayName; } - virtual GetUpstreamCacheResult GetCacheRecord(UpstreamCacheKey CacheKey, ZenContentType Type) override + virtual GetUpstreamCacheResult GetCacheRecord(CacheKey CacheKey, ZenContentType Type) override { try { @@ -144,12 +144,54 @@ namespace detail { } } - virtual GetUpstreamCacheResult GetCachePayload(UpstreamPayloadKey PayloadKey) override + virtual GetUpstreamCacheBatchResult GetCacheRecords(std::span CacheKeys, + std::span KeyIndex, + const CacheRecordPolicy& Policy, + OnCacheRecordGetComplete&& OnComplete) override + { + ZEN_UNUSED(Policy); + + CloudCacheSession Session(m_Client); + + for (size_t Index : KeyIndex) + { + const CacheKey& CacheKey = CacheKeys[Index]; + CloudCacheResult Result = Session.GetRef(CacheKey.Bucket, CacheKey.Hash, ZenContentType::kCbObject); + + CbPackage Package; + CbObjectView Record; + + if (Result.Success) + { + const CbValidateError ValidationResult = ValidateCompactBinary(Result.Response, CbValidateMode::All); + if (ValidationResult == CbValidateError::None) + { + Record = CbObjectView(Result.Response.GetData()); + Record.IterateAttachments([&Session, &Result, &Package](CbFieldView AttachmentHash) { + CloudCacheResult AttachmentResult = Session.GetCompressedBlob(AttachmentHash.AsHash()); + if (AttachmentResult.Success) + { + if (CompressedBuffer Chunk = CompressedBuffer::FromCompressed(SharedBuffer(AttachmentResult.Response))) + { + Package.AddAttachment(CbAttachment(Chunk)); + } + } + }); + } + } + + OnComplete({.CacheKey = CacheKey, .KeyIndex = Index, .Record = Record, .Package = Package}); + } + + return {}; + } + + virtual GetUpstreamCacheResult GetCachePayload(const CacheKey&, const IoHash& PayloadId) override { try { CloudCacheSession Session(m_Client); - const CloudCacheResult Result = Session.GetCompressedBlob(PayloadKey.PayloadId); + const CloudCacheResult Result = Session.GetCompressedBlob(PayloadId); if (Result.ErrorCode == 0) { @@ -171,6 +213,29 @@ namespace detail { } } + virtual GetUpstreamCacheBatchResult GetCachePayloads(std::span CacheChunkRequests, + std::span RequestIndex, + OnCachePayloadGetComplete&& OnComplete) override final + { + CloudCacheSession Session(m_Client); + + for (size_t Index : RequestIndex) + { + const CacheChunkRequest& Request = CacheChunkRequests[Index]; + const CloudCacheResult Result = Session.GetCompressedBlob(Request.ChunkId); + + OnComplete({.Request = Request, .RequestIndex = Index, .Payload = Result.Response}); + + if (Result.ErrorCode) + { + m_HealthOk = false; + break; + } + } + + return {}; + } + virtual PutUpstreamCacheResult PutCacheRecord(const UpstreamCacheRecord& CacheRecord, IoBuffer RecordValue, std::span Payloads) override @@ -419,7 +484,7 @@ namespace detail { virtual std::string_view DisplayName() const override { return m_DisplayName; } - virtual GetUpstreamCacheResult GetCacheRecord(UpstreamCacheKey CacheKey, ZenContentType Type) override + virtual GetUpstreamCacheResult GetCacheRecord(CacheKey CacheKey, ZenContentType Type) override { try { @@ -446,13 +511,81 @@ namespace detail { } } - virtual GetUpstreamCacheResult GetCachePayload(UpstreamPayloadKey PayloadKey) override + virtual GetUpstreamCacheBatchResult GetCacheRecords(std::span CacheKeys, + std::span KeyIndex, + const CacheRecordPolicy& Policy, + OnCacheRecordGetComplete&& OnComplete) override + { + std::vector IndexMap; + IndexMap.reserve(KeyIndex.size()); + + CbObjectWriter BatchRequest; + BatchRequest << "Method"sv + << "GetCacheRecords"; + + BatchRequest.BeginObject("Params"sv); + { + BatchRequest.BeginArray("CacheKeys"sv); + for (size_t Index : KeyIndex) + { + const CacheKey& Key = CacheKeys[Index]; + IndexMap.push_back(Index); + + BatchRequest.BeginObject(); + BatchRequest << "Bucket"sv << Key.Bucket; + BatchRequest << "Hash"sv << Key.Hash; + BatchRequest.EndObject(); + } + BatchRequest.EndArray(); + + BatchRequest.BeginObject("Policy"sv); + CacheRecordPolicy::Save(Policy, BatchRequest); + BatchRequest.EndObject(); + } + BatchRequest.EndObject(); + + CbPackage BatchResponse; + bool Success = false; + + { + ZenStructuredCacheSession Session(*m_Client); + ZenCacheResult Result = Session.InvokeRpc(BatchRequest.Save()); + if (Result.Success) + { + Success = BatchResponse.TryLoad(Result.Response); + } + else if (Result.ErrorCode) + { + Success = m_HealthOk = false; + } + } + + if (!Success) + { + for (size_t Index : KeyIndex) + { + OnComplete({.CacheKey = CacheKeys[Index], .KeyIndex = Index, .Record = CbObjectView(), .Package = CbPackage()}); + } + + return {}; + } + + for (size_t LocalIndex = 0; CbFieldView Record : BatchResponse.GetObject()["Result"sv]) + { + const size_t Index = IndexMap[LocalIndex++]; + OnComplete({.CacheKey = CacheKeys[Index], .KeyIndex = Index, .Record = Record.AsObjectView(), .Package = BatchResponse}); + } + + return {}; + } + + virtual GetUpstreamCacheResult GetCachePayload(const CacheKey& CacheKey, const IoHash& PayloadId) override { try { ZenStructuredCacheSession Session(*m_Client); const ZenCacheResult Result = - Session.GetCachePayload(PayloadKey.CacheKey.Bucket, PayloadKey.CacheKey.Hash, PayloadKey.PayloadId); + Session.GetCachePayload(CacheKey.Bucket, CacheKey.Hash, PayloadId); if (Result.ErrorCode == 0) { @@ -474,6 +607,91 @@ namespace detail { } } + virtual GetUpstreamCacheBatchResult GetCachePayloads(std::span CacheChunkRequests, + std::span RequestIndex, + OnCachePayloadGetComplete&& OnComplete) override final + { + std::vector IndexMap; + IndexMap.reserve(RequestIndex.size()); + + CbObjectWriter BatchRequest; + BatchRequest << "Method"sv + << "GetCachePayloads"; + + BatchRequest.BeginObject("Params"sv); + { + BatchRequest.BeginArray("ChunkRequests"sv); + { + for (size_t Index : RequestIndex) + { + const CacheChunkRequest& Request = CacheChunkRequests[Index]; + IndexMap.push_back(Index); + + BatchRequest.BeginObject(); + { + BatchRequest.BeginObject("Key"sv); + BatchRequest << "Bucket"sv << Request.Key.Bucket; + BatchRequest << "Hash"sv << Request.Key.Hash; + BatchRequest.EndObject(); + + BatchRequest.AddObjectId("PayloadId"sv, Request.PayloadId); + BatchRequest << "ChunkId"sv << Request.ChunkId; + BatchRequest << "RawOffset"sv << Request.RawOffset; + BatchRequest << "RawSize"sv << Request.RawSize; + BatchRequest << "Policy"sv << static_cast(Request.Policy); + } + BatchRequest.EndObject(); + } + } + BatchRequest.EndArray(); + } + BatchRequest.EndObject(); + + CbPackage BatchResponse; + bool Success = false; + + { + ZenStructuredCacheSession Session(*m_Client); + ZenCacheResult Result = Session.InvokeRpc(BatchRequest.Save()); + if (Result.Success) + { + Success = BatchResponse.TryLoad(Result.Response); + } + else if (Result.ErrorCode) + { + m_HealthOk = false; + } + } + + if (!Success) + { + for (size_t Index : RequestIndex) + { + OnComplete({.Request = CacheChunkRequests[Index], .RequestIndex = Index, .Payload = IoBuffer()}); + } + + return {}; + } + + for (int32_t LocalIndex = 0; CbFieldView AttachmentHash : BatchResponse.GetObject()["Result"sv]) + { + const size_t Index = IndexMap[LocalIndex++]; + IoBuffer Payload; + + if (const CbAttachment* Attachment = BatchResponse.FindAttachment(AttachmentHash.AsHash())) + { + if (const CompressedBuffer& Compressed = Attachment->AsCompressedBinary()) + { + Payload = Compressed.GetCompressed().Flatten().AsIoBuffer(); + } + } + + OnComplete({.Request = CacheChunkRequests[Index], .RequestIndex = Index, .Payload = std::move(Payload)}); + } + + return {}; + } + virtual PutUpstreamCacheResult PutCacheRecord(const UpstreamCacheRecord& CacheRecord, IoBuffer RecordValue, std::span Payloads) override @@ -758,7 +976,7 @@ public: virtual void RegisterEndpoint(std::unique_ptr Endpoint) override { m_Endpoints.emplace_back(std::move(Endpoint)); } - virtual GetUpstreamCacheResult GetCacheRecord(UpstreamCacheKey CacheKey, ZenContentType Type) override + virtual GetUpstreamCacheResult GetCacheRecord(CacheKey CacheKey, ZenContentType Type) override { if (m_Options.ReadUpstream) { @@ -780,94 +998,82 @@ public: return {}; } - virtual GetUpstreamCacheBatchResult GetCacheRecords(std::span CacheKeys, - std::span KeyIndex, - OnCacheGetComplete OnComplete) override final + virtual void GetCacheRecords(std::span CacheKeys, + std::span KeyIndex, + const CacheRecordPolicy& Policy, + OnCacheRecordGetComplete&& OnComplete) override final { - if (!m_Options.ReadUpstream) - { - return {.Missing = std::vector(KeyIndex.begin(), KeyIndex.end())}; - } + std::vector MissingKeys(KeyIndex.begin(), KeyIndex.end()); - GetUpstreamCacheBatchResult Result; - - for (size_t Idx : KeyIndex) + if (m_Options.ReadUpstream) { - const UpstreamCacheKey CacheKey = {CacheKeys[Idx].Bucket, CacheKeys[Idx].Hash}; - - GetUpstreamCacheResult CacheResult; for (auto& Endpoint : m_Endpoints) { - if (Endpoint->IsHealthy()) + if (Endpoint->IsHealthy() && !MissingKeys.empty()) { - CacheResult = Endpoint->GetCacheRecord(CacheKey, ZenContentType::kCbPackage); - m_Stats.Add(m_Log, *Endpoint, CacheResult, m_Endpoints); + std::vector Missing; - if (CacheResult.Success) - { - break; - } - } - } + auto EndpointResult = + Endpoint->GetCacheRecords(CacheKeys, MissingKeys, Policy, [&](CacheRecordGetCompleteParams&& Params) { + if (Params.Record) + { + OnComplete(std::forward(Params)); + } + else + { + Missing.push_back(Params.KeyIndex); + } + }); - if (CacheResult.Success) - { - OnComplete(Idx, CacheResult.Value); - } - else - { - Result.Missing.push_back(Idx); + MissingKeys = std::move(Missing); + } } } - return Result; - } - - virtual GetUpstreamCacheBatchResult GetCachePayloads(std::span CacheChunkRequests, - std::span ChunkIndex, - OnCacheGetComplete OnComplete) override final - { - if (!m_Options.ReadUpstream) + for (size_t Index : MissingKeys) { - return {.Missing = std::vector(ChunkIndex.begin(), ChunkIndex.end())}; + OnComplete({.CacheKey = CacheKeys[Index], .KeyIndex = Index, .Record = CbObjectView(), .Package = CbPackage()}); } + } - GetUpstreamCacheBatchResult Result; + virtual void GetCachePayloads(std::span CacheChunkRequests, + std::span RequestIndex, + OnCachePayloadGetComplete&& OnComplete) override final + { + std::vector MissingPayloads(RequestIndex.begin(), RequestIndex.end()); - for (size_t Idx : ChunkIndex) + if (m_Options.ReadUpstream) { - const CacheChunkRequest& Chunk = CacheChunkRequests[Idx]; - UpstreamPayloadKey PayloadKey{{Chunk.Key.Bucket, Chunk.Key.Hash}, Chunk.ChunkId}; - - GetUpstreamCacheResult CacheResult; for (auto& Endpoint : m_Endpoints) { - if (Endpoint->IsHealthy()) + if (Endpoint->IsHealthy() && !MissingPayloads.empty()) { - CacheResult = Endpoint->GetCachePayload(PayloadKey); - m_Stats.Add(m_Log, *Endpoint, CacheResult, m_Endpoints); + std::vector Missing; - if (CacheResult.Success) - { - break; - } - } - } + auto EndpointResult = + Endpoint->GetCachePayloads(CacheChunkRequests, MissingPayloads, [&](CachePayloadGetCompleteParams&& Params) { + if (Params.Payload) + { + OnComplete(std::forward(Params)); + } + else + { + Missing.push_back(Params.RequestIndex); + } + }); - if (CacheResult.Success) - { - OnComplete(Idx, CacheResult.Value); - } - else - { - Result.Missing.push_back(Idx); + MissingPayloads = std::move(Missing); + } } } - return Result; + for (size_t Index : MissingPayloads) + { + OnComplete({.Request = CacheChunkRequests[Index], .RequestIndex = Index, .Payload = IoBuffer()}); + } } - virtual GetUpstreamCacheResult GetCachePayload(UpstreamPayloadKey PayloadKey) override + virtual GetUpstreamCacheResult GetCachePayload(const CacheKey& CacheKey, const IoHash& PayloadId) override { if (m_Options.ReadUpstream) { @@ -875,7 +1081,7 @@ public: { if (Endpoint->IsHealthy()) { - const GetUpstreamCacheResult Result = Endpoint->GetCachePayload(PayloadKey); + const GetUpstreamCacheResult Result = Endpoint->GetCachePayload(CacheKey, PayloadId); m_Stats.Add(m_Log, *Endpoint, Result, m_Endpoints); if (Result.Success) -- cgit v1.2.3 From 31ba344167677f175ec79ce7e579552a9811245d Mon Sep 17 00:00:00 2001 From: Per Larsson Date: Thu, 11 Nov 2021 11:21:31 +0100 Subject: Format and remove unused type. --- zenserver/upstream/upstreamcache.cpp | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) (limited to 'zenserver/upstream/upstreamcache.cpp') diff --git a/zenserver/upstream/upstreamcache.cpp b/zenserver/upstream/upstreamcache.cpp index 0a0706656..a221c7d8e 100644 --- a/zenserver/upstream/upstreamcache.cpp +++ b/zenserver/upstream/upstreamcache.cpp @@ -584,8 +584,7 @@ namespace detail { try { ZenStructuredCacheSession Session(*m_Client); - const ZenCacheResult Result = - Session.GetCachePayload(CacheKey.Bucket, CacheKey.Hash, PayloadId); + const ZenCacheResult Result = Session.GetCachePayload(CacheKey.Bucket, CacheKey.Hash, PayloadId); if (Result.ErrorCode == 0) { -- cgit v1.2.3 From efd0c3c0896c97c6cd37e6a6ac2edf331e5b4a3f Mon Sep 17 00:00:00 2001 From: Per Larsson Date: Thu, 11 Nov 2021 11:37:31 +0100 Subject: Removed batch result. --- zenserver/upstream/upstreamcache.cpp | 34 +++++++++++++++++++--------------- 1 file changed, 19 insertions(+), 15 deletions(-) (limited to 'zenserver/upstream/upstreamcache.cpp') diff --git a/zenserver/upstream/upstreamcache.cpp b/zenserver/upstream/upstreamcache.cpp index a221c7d8e..d5414daf1 100644 --- a/zenserver/upstream/upstreamcache.cpp +++ b/zenserver/upstream/upstreamcache.cpp @@ -144,10 +144,10 @@ namespace detail { } } - virtual GetUpstreamCacheBatchResult GetCacheRecords(std::span CacheKeys, - std::span KeyIndex, - const CacheRecordPolicy& Policy, - OnCacheRecordGetComplete&& OnComplete) override + virtual GetUpstreamCacheResult GetCacheRecords(std::span CacheKeys, + std::span KeyIndex, + const CacheRecordPolicy& Policy, + OnCacheRecordGetComplete&& OnComplete) override { ZEN_UNUSED(Policy); @@ -161,7 +161,7 @@ namespace detail { CbPackage Package; CbObjectView Record; - if (Result.Success) + if (Result.ErrorCode == 0) { const CbValidateError ValidationResult = ValidateCompactBinary(Result.Response, CbValidateMode::All); if (ValidationResult == CbValidateError::None) @@ -179,6 +179,10 @@ namespace detail { }); } } + else + { + m_HealthOk = false; + } OnComplete({.CacheKey = CacheKey, .KeyIndex = Index, .Record = Record, .Package = Package}); } @@ -213,9 +217,9 @@ namespace detail { } } - virtual GetUpstreamCacheBatchResult GetCachePayloads(std::span CacheChunkRequests, - std::span RequestIndex, - OnCachePayloadGetComplete&& OnComplete) override final + virtual GetUpstreamCacheResult GetCachePayloads(std::span CacheChunkRequests, + std::span RequestIndex, + OnCachePayloadGetComplete&& OnComplete) override final { CloudCacheSession Session(m_Client); @@ -511,10 +515,10 @@ namespace detail { } } - virtual GetUpstreamCacheBatchResult GetCacheRecords(std::span CacheKeys, - std::span KeyIndex, - const CacheRecordPolicy& Policy, - OnCacheRecordGetComplete&& OnComplete) override + virtual GetUpstreamCacheResult GetCacheRecords(std::span CacheKeys, + std::span KeyIndex, + const CacheRecordPolicy& Policy, + OnCacheRecordGetComplete&& OnComplete) override { std::vector IndexMap; IndexMap.reserve(KeyIndex.size()); @@ -606,9 +610,9 @@ namespace detail { } } - virtual GetUpstreamCacheBatchResult GetCachePayloads(std::span CacheChunkRequests, - std::span RequestIndex, - OnCachePayloadGetComplete&& OnComplete) override final + virtual GetUpstreamCacheResult GetCachePayloads(std::span CacheChunkRequests, + std::span RequestIndex, + OnCachePayloadGetComplete&& OnComplete) override final { std::vector IndexMap; IndexMap.reserve(RequestIndex.size()); -- cgit v1.2.3 From 7940ae2ff24e1b708d1d6a0bccf266213ea7316b Mon Sep 17 00:00:00 2001 From: Per Larsson Date: Thu, 11 Nov 2021 13:28:52 +0100 Subject: Fixed stats. --- zenserver/upstream/upstreamcache.cpp | 211 ++++++++++++++++++++--------------- 1 file changed, 119 insertions(+), 92 deletions(-) (limited to 'zenserver/upstream/upstreamcache.cpp') diff --git a/zenserver/upstream/upstreamcache.cpp b/zenserver/upstream/upstreamcache.cpp index d5414daf1..b486b751c 100644 --- a/zenserver/upstream/upstreamcache.cpp +++ b/zenserver/upstream/upstreamcache.cpp @@ -151,43 +151,54 @@ namespace detail { { ZEN_UNUSED(Policy); - CloudCacheSession Session(m_Client); + CloudCacheSession Session(m_Client); + GetUpstreamCacheResult Result; for (size_t Index : KeyIndex) { - const CacheKey& CacheKey = CacheKeys[Index]; - CloudCacheResult Result = Session.GetRef(CacheKey.Bucket, CacheKey.Hash, ZenContentType::kCbObject); + const CacheKey& CacheKey = CacheKeys[Index]; + CbPackage Package; + CbObjectView Record; - CbPackage Package; - CbObjectView Record; - - if (Result.ErrorCode == 0) + if (!Result.Error) { - const CbValidateError ValidationResult = ValidateCompactBinary(Result.Response, CbValidateMode::All); - if (ValidationResult == CbValidateError::None) + CloudCacheResult RefResult = Session.GetRef(CacheKey.Bucket, CacheKey.Hash, ZenContentType::kCbObject); + AppendResult(RefResult, Result); + + if (RefResult.ErrorCode == 0) { - Record = CbObjectView(Result.Response.GetData()); - Record.IterateAttachments([&Session, &Result, &Package](CbFieldView AttachmentHash) { - CloudCacheResult AttachmentResult = Session.GetCompressedBlob(AttachmentHash.AsHash()); - if (AttachmentResult.Success) - { - if (CompressedBuffer Chunk = CompressedBuffer::FromCompressed(SharedBuffer(AttachmentResult.Response))) + const CbValidateError ValidationResult = ValidateCompactBinary(RefResult.Response, CbValidateMode::All); + if (ValidationResult == CbValidateError::None) + { + Record = CbObjectView(RefResult.Response.GetData()); + Record.IterateAttachments([this, &Session, &Result, &Package](CbFieldView AttachmentHash) { + CloudCacheResult BlobResult = Session.GetCompressedBlob(AttachmentHash.AsHash()); + AppendResult(BlobResult, Result); + + if (BlobResult.ErrorCode == 0) { - Package.AddAttachment(CbAttachment(Chunk)); + if (CompressedBuffer Chunk = CompressedBuffer::FromCompressed(SharedBuffer(BlobResult.Response))) + { + Package.AddAttachment(CbAttachment(Chunk)); + } } - } - }); + else + { + m_HealthOk = false; + } + }); + } + } + else + { + m_HealthOk = false; } - } - else - { - m_HealthOk = false; } OnComplete({.CacheKey = CacheKey, .KeyIndex = Index, .Record = Record, .Package = Package}); } - return {}; + return Result; } virtual GetUpstreamCacheResult GetCachePayload(const CacheKey&, const IoHash& PayloadId) override @@ -221,23 +232,27 @@ namespace detail { std::span RequestIndex, OnCachePayloadGetComplete&& OnComplete) override final { - CloudCacheSession Session(m_Client); + CloudCacheSession Session(m_Client); + GetUpstreamCacheResult Result; for (size_t Index : RequestIndex) { const CacheChunkRequest& Request = CacheChunkRequests[Index]; - const CloudCacheResult Result = Session.GetCompressedBlob(Request.ChunkId); - - OnComplete({.Request = Request, .RequestIndex = Index, .Payload = Result.Response}); + IoBuffer Payload; - if (Result.ErrorCode) + if (!Result.Error) { - m_HealthOk = false; - break; + const CloudCacheResult BlobResult = Session.GetCompressedBlob(Request.ChunkId); + Payload = BlobResult.Response; + + AppendResult(BlobResult, Result); + m_HealthOk = BlobResult.ErrorCode == 0; } + + OnComplete({.Request = Request, .RequestIndex = Index, .Payload = Payload}); } - return {}; + return Result; } virtual PutUpstreamCacheResult PutCacheRecord(const UpstreamCacheRecord& CacheRecord, @@ -392,6 +407,18 @@ namespace detail { virtual UpstreamEndpointStats& Stats() override { return m_Stats; } private: + static void AppendResult(const CloudCacheResult& Result, GetUpstreamCacheResult& Out) + { + Out.Success &= Result.Success; + Out.Bytes += Result.Bytes; + Out.ElapsedSeconds += Result.ElapsedSeconds; + + if (Result.ErrorCode) + { + Out.Error = {.ErrorCode = Result.ErrorCode, .Reason = std::move(Result.Reason)}; + } + }; + spdlog::logger& Log() { return m_Log; } spdlog::logger& m_Log; @@ -548,39 +575,39 @@ namespace detail { } BatchRequest.EndObject(); - CbPackage BatchResponse; - bool Success = false; + CbPackage BatchResponse; + ZenCacheResult Result; { ZenStructuredCacheSession Session(*m_Client); - ZenCacheResult Result = Session.InvokeRpc(BatchRequest.Save()); - if (Result.Success) - { - Success = BatchResponse.TryLoad(Result.Response); - } - else if (Result.ErrorCode) - { - Success = m_HealthOk = false; - } + Result = Session.InvokeRpc(BatchRequest.Save()); } - if (!Success) + if (Result.Success) { - for (size_t Index : KeyIndex) + if (BatchResponse.TryLoad(Result.Response)) { - OnComplete({.CacheKey = CacheKeys[Index], .KeyIndex = Index, .Record = CbObjectView(), .Package = CbPackage()}); - } + for (size_t LocalIndex = 0; CbFieldView Record : BatchResponse.GetObject()["Result"sv]) + { + const size_t Index = IndexMap[LocalIndex++]; + OnComplete( + {.CacheKey = CacheKeys[Index], .KeyIndex = Index, .Record = Record.AsObjectView(), .Package = BatchResponse}); + } - return {}; + return {.Bytes = Result.Bytes, .ElapsedSeconds = Result.ElapsedSeconds, .Success = true}; + } + } + else if (Result.ErrorCode) + { + m_HealthOk = false; } - for (size_t LocalIndex = 0; CbFieldView Record : BatchResponse.GetObject()["Result"sv]) + for (size_t Index : KeyIndex) { - const size_t Index = IndexMap[LocalIndex++]; - OnComplete({.CacheKey = CacheKeys[Index], .KeyIndex = Index, .Record = Record.AsObjectView(), .Package = BatchResponse}); + OnComplete({.CacheKey = CacheKeys[Index], .KeyIndex = Index, .Record = CbObjectView(), .Package = CbPackage()}); } - return {}; + return {.Error{.ErrorCode = Result.ErrorCode, .Reason = std::move(Result.Reason)}}; } virtual GetUpstreamCacheResult GetCachePayload(const CacheKey& CacheKey, const IoHash& PayloadId) override @@ -650,49 +677,48 @@ namespace detail { } BatchRequest.EndObject(); - CbPackage BatchResponse; - bool Success = false; + CbPackage BatchResponse; + ZenCacheResult Result; { ZenStructuredCacheSession Session(*m_Client); - ZenCacheResult Result = Session.InvokeRpc(BatchRequest.Save()); - if (Result.Success) - { - Success = BatchResponse.TryLoad(Result.Response); - } - else if (Result.ErrorCode) - { - m_HealthOk = false; - } + Result = Session.InvokeRpc(BatchRequest.Save()); } - if (!Success) + if (Result.Success) { - for (size_t Index : RequestIndex) + if (BatchResponse.TryLoad(Result.Response)) { - OnComplete({.Request = CacheChunkRequests[Index], .RequestIndex = Index, .Payload = IoBuffer()}); - } - - return {}; - } + for (size_t LocalIndex = 0; CbFieldView AttachmentHash : BatchResponse.GetObject()["Result"sv]) + { + const size_t Index = IndexMap[LocalIndex++]; + IoBuffer Payload; - for (int32_t LocalIndex = 0; CbFieldView AttachmentHash : BatchResponse.GetObject()["Result"sv]) - { - const size_t Index = IndexMap[LocalIndex++]; - IoBuffer Payload; + if (const CbAttachment* Attachment = BatchResponse.FindAttachment(AttachmentHash.AsHash())) + { + if (const CompressedBuffer& Compressed = Attachment->AsCompressedBinary()) + { + Payload = Compressed.GetCompressed().Flatten().AsIoBuffer(); + } + } - if (const CbAttachment* Attachment = BatchResponse.FindAttachment(AttachmentHash.AsHash())) - { - if (const CompressedBuffer& Compressed = Attachment->AsCompressedBinary()) - { - Payload = Compressed.GetCompressed().Flatten().AsIoBuffer(); + OnComplete({.Request = CacheChunkRequests[Index], .RequestIndex = Index, .Payload = std::move(Payload)}); } + + return {.Bytes = Result.Bytes, .ElapsedSeconds = Result.ElapsedSeconds, .Success = true}; } + } + else if (Result.ErrorCode) + { + m_HealthOk = false; + } - OnComplete({.Request = CacheChunkRequests[Index], .RequestIndex = Index, .Payload = std::move(Payload)}); + for (size_t Index : RequestIndex) + { + OnComplete({.Request = CacheChunkRequests[Index], .RequestIndex = Index, .Payload = IoBuffer()}); } - return {}; + return {.Error{.ErrorCode = Result.ErrorCode, .Reason = std::move(Result.Reason)}}; } virtual PutUpstreamCacheResult PutCacheRecord(const UpstreamCacheRecord& CacheRecord, @@ -1016,18 +1042,18 @@ public: { std::vector Missing; - auto EndpointResult = - Endpoint->GetCacheRecords(CacheKeys, MissingKeys, Policy, [&](CacheRecordGetCompleteParams&& Params) { - if (Params.Record) - { - OnComplete(std::forward(Params)); - } - else - { - Missing.push_back(Params.KeyIndex); - } - }); + auto Result = Endpoint->GetCacheRecords(CacheKeys, MissingKeys, Policy, [&](CacheRecordGetCompleteParams&& Params) { + if (Params.Record) + { + OnComplete(std::forward(Params)); + } + else + { + Missing.push_back(Params.KeyIndex); + } + }); + m_Stats.Add(m_Log, *Endpoint, Result, m_Endpoints); MissingKeys = std::move(Missing); } } @@ -1053,7 +1079,7 @@ public: { std::vector Missing; - auto EndpointResult = + auto Result = Endpoint->GetCachePayloads(CacheChunkRequests, MissingPayloads, [&](CachePayloadGetCompleteParams&& Params) { if (Params.Payload) { @@ -1065,6 +1091,7 @@ public: } }); + m_Stats.Add(m_Log, *Endpoint, Result, m_Endpoints); MissingPayloads = std::move(Missing); } } -- cgit v1.2.3 From 033fd189b876197e263e669c81703bdb550b3ec0 Mon Sep 17 00:00:00 2001 From: Per Larsson Date: Sat, 13 Nov 2021 15:33:22 +0100 Subject: Fixed bug in upstream jupiter endpoint. --- zenserver/upstream/upstreamcache.cpp | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) (limited to 'zenserver/upstream/upstreamcache.cpp') diff --git a/zenserver/upstream/upstreamcache.cpp b/zenserver/upstream/upstreamcache.cpp index 174c3d17a..085932685 100644 --- a/zenserver/upstream/upstreamcache.cpp +++ b/zenserver/upstream/upstreamcache.cpp @@ -158,7 +158,7 @@ namespace detail { { const CacheKey& CacheKey = CacheKeys[Index]; CbPackage Package; - CbObjectView Record; + CbObject Record; if (!Result.Error) { @@ -170,7 +170,7 @@ namespace detail { const CbValidateError ValidationResult = ValidateCompactBinary(RefResult.Response, CbValidateMode::All); if (ValidationResult == CbValidateError::None) { - Record = CbObjectView(RefResult.Response.GetData()); + Record = LoadCompactBinaryObject(RefResult.Response); Record.IterateAttachments([this, &Session, &Result, &Package](CbFieldView AttachmentHash) { CloudCacheResult BlobResult = Session.GetCompressedBlob(AttachmentHash.AsHash()); AppendResult(BlobResult, Result); -- cgit v1.2.3