aboutsummaryrefslogtreecommitdiff
path: root/zenserver/upstream/upstreamcache.cpp
diff options
context:
space:
mode:
authormattpetersepic <[email protected]>2022-01-27 10:52:07 -0700
committerGitHub <[email protected]>2022-01-27 10:52:07 -0700
commitfb1afc87a436ff6374daeaef5f7682f098a85cd7 (patch)
tree89810fc5d85ab81cce7fdc4e26849ad0d25cf2d1 /zenserver/upstream/upstreamcache.cpp
parentAdd batched CacheRecord put rpc (#38) (diff)
downloadzen-fb1afc87a436ff6374daeaef5f7682f098a85cd7.tar.xz
zen-fb1afc87a436ff6374daeaef5f7682f098a85cd7.zip
Rename Paylod to Value to match the client side. Rename PayloadId to ValueContentId where its a hash instead of an oid.
Diffstat (limited to 'zenserver/upstream/upstreamcache.cpp')
-rw-r--r--zenserver/upstream/upstreamcache.cpp111
1 files changed, 56 insertions, 55 deletions
diff --git a/zenserver/upstream/upstreamcache.cpp b/zenserver/upstream/upstreamcache.cpp
index 091406db3..657cfb729 100644
--- a/zenserver/upstream/upstreamcache.cpp
+++ b/zenserver/upstream/upstreamcache.cpp
@@ -270,14 +270,14 @@ namespace detail {
return Result;
}
- virtual GetUpstreamCacheResult GetCachePayload(const CacheKey&, const IoHash& PayloadId) override
+ virtual GetUpstreamCacheResult GetCacheValue(const CacheKey&, const IoHash& ValueContentId) override
{
- ZEN_TRACE_CPU("Upstream::Horde::GetSingleCachePayload");
+ ZEN_TRACE_CPU("Upstream::Horde::GetSingleCacheValue");
try
{
CloudCacheSession Session(m_Client);
- const CloudCacheResult Result = Session.GetCompressedBlob(PayloadId);
+ const CloudCacheResult Result = Session.GetCompressedBlob(ValueContentId);
m_Status.SetFromErrorCode(Result.ErrorCode, Result.Reason);
@@ -301,11 +301,11 @@ namespace detail {
}
}
- virtual GetUpstreamCacheResult GetCachePayloads(std::span<CacheChunkRequest> CacheChunkRequests,
- std::span<size_t> RequestIndex,
- OnCachePayloadGetComplete&& OnComplete) override final
+ virtual GetUpstreamCacheResult GetCacheValues(std::span<CacheChunkRequest> CacheChunkRequests,
+ std::span<size_t> RequestIndex,
+ OnCacheValueGetComplete&& OnComplete) override final
{
- ZEN_TRACE_CPU("Upstream::Horde::GetCachePayloads");
+ ZEN_TRACE_CPU("Upstream::Horde::GetCacheValues");
CloudCacheSession Session(m_Client);
GetUpstreamCacheResult Result;
@@ -325,7 +325,7 @@ namespace detail {
m_Status.SetFromErrorCode(BlobResult.ErrorCode, BlobResult.Reason);
}
- OnComplete({.Request = Request, .RequestIndex = Index, .Payload = Payload});
+ OnComplete({.Request = Request, .RequestIndex = Index, .Value = Payload});
}
return Result;
@@ -333,11 +333,11 @@ namespace detail {
virtual PutUpstreamCacheResult PutCacheRecord(const UpstreamCacheRecord& CacheRecord,
IoBuffer RecordValue,
- std::span<IoBuffer const> Payloads) override
+ std::span<IoBuffer const> Values) override
{
ZEN_TRACE_CPU("Upstream::Horde::PutCacheRecord");
- ZEN_ASSERT(CacheRecord.PayloadIds.size() == Payloads.size());
+ ZEN_ASSERT(CacheRecord.ValueContentIds.size() == Values.size());
const int32_t MaxAttempts = 3;
try
@@ -371,30 +371,31 @@ namespace detail {
int64_t TotalBytes = 0ull;
double TotalElapsedSeconds = 0.0;
- const auto PutBlobs = [&](std::span<IoHash> PayloadIds, std::string& OutReason) -> bool {
- for (const IoHash& PayloadId : PayloadIds)
+ const auto PutBlobs = [&](std::span<IoHash> ValueContentIds, std::string& OutReason) -> bool {
+ for (const IoHash& ValueContentId : ValueContentIds)
{
- const auto It = std::find(std::begin(CacheRecord.PayloadIds), std::end(CacheRecord.PayloadIds), PayloadId);
+ const auto It =
+ std::find(std::begin(CacheRecord.ValueContentIds), std::end(CacheRecord.ValueContentIds), ValueContentId);
- if (It == std::end(CacheRecord.PayloadIds))
+ if (It == std::end(CacheRecord.ValueContentIds))
{
- OutReason = fmt::format("value '{}' MISSING from local cache", PayloadId);
+ OutReason = fmt::format("value '{}' MISSING from local cache", ValueContentId);
return false;
}
- const size_t Idx = std::distance(std::begin(CacheRecord.PayloadIds), It);
+ const size_t Idx = std::distance(std::begin(CacheRecord.ValueContentIds), It);
CloudCacheResult BlobResult;
for (int32_t Attempt = 0; Attempt < MaxAttempts && !BlobResult.Success; Attempt++)
{
- BlobResult = Session.PutCompressedBlob(CacheRecord.PayloadIds[Idx], Payloads[Idx]);
+ BlobResult = Session.PutCompressedBlob(CacheRecord.ValueContentIds[Idx], Values[Idx]);
}
m_Status.SetFromErrorCode(BlobResult.ErrorCode, BlobResult.Reason);
if (!BlobResult.Success)
{
- OutReason = fmt::format("upload value '{}' FAILED, reason '{}'", PayloadId, BlobResult.Reason);
+ OutReason = fmt::format("upload value '{}' FAILED, reason '{}'", ValueContentId, BlobResult.Reason);
return false;
}
@@ -685,14 +686,14 @@ namespace detail {
return {.Error{.ErrorCode = Result.ErrorCode, .Reason = std::move(Result.Reason)}};
}
- virtual GetUpstreamCacheResult GetCachePayload(const CacheKey& CacheKey, const IoHash& PayloadId) override
+ virtual GetUpstreamCacheResult GetCacheValue(const CacheKey& CacheKey, const IoHash& ValueContentId) override
{
- ZEN_TRACE_CPU("Upstream::Zen::GetSingleCachePayload");
+ ZEN_TRACE_CPU("Upstream::Zen::GetSingleCacheValue");
try
{
ZenStructuredCacheSession Session(*m_Client);
- const ZenCacheResult Result = Session.GetCachePayload(CacheKey.Bucket, CacheKey.Hash, PayloadId);
+ const ZenCacheResult Result = Session.GetCacheValue(CacheKey.Bucket, CacheKey.Hash, ValueContentId);
m_Status.SetFromErrorCode(Result.ErrorCode, Result.Reason);
@@ -716,11 +717,11 @@ namespace detail {
}
}
- virtual GetUpstreamCacheResult GetCachePayloads(std::span<CacheChunkRequest> CacheChunkRequests,
- std::span<size_t> RequestIndex,
- OnCachePayloadGetComplete&& OnComplete) override final
+ virtual GetUpstreamCacheResult GetCacheValues(std::span<CacheChunkRequest> CacheChunkRequests,
+ std::span<size_t> RequestIndex,
+ OnCacheValueGetComplete&& OnComplete) override final
{
- ZEN_TRACE_CPU("Upstream::Zen::GetCachePayloads");
+ ZEN_TRACE_CPU("Upstream::Zen::GetCacheValues");
std::vector<size_t> IndexMap;
IndexMap.reserve(RequestIndex.size());
@@ -744,7 +745,7 @@ namespace detail {
BatchRequest << "Bucket"sv << Request.Key.Bucket;
BatchRequest << "Hash"sv << Request.Key.Hash;
BatchRequest.EndObject();
- BatchRequest.AddObjectId("ValueId"sv, Request.PayloadId);
+ BatchRequest.AddObjectId("ValueId"sv, Request.ValueId);
BatchRequest << "ChunkId"sv << Request.ChunkId;
BatchRequest << "RawOffset"sv << Request.RawOffset;
BatchRequest << "RawSize"sv << Request.RawSize;
@@ -784,7 +785,7 @@ namespace detail {
}
}
- OnComplete({.Request = CacheChunkRequests[Index], .RequestIndex = Index, .Payload = std::move(Payload)});
+ OnComplete({.Request = CacheChunkRequests[Index], .RequestIndex = Index, .Value = std::move(Payload)});
}
return {.Bytes = Result.Bytes, .ElapsedSeconds = Result.ElapsedSeconds, .Success = true};
@@ -793,7 +794,7 @@ namespace detail {
for (size_t Index : RequestIndex)
{
- OnComplete({.Request = CacheChunkRequests[Index], .RequestIndex = Index, .Payload = IoBuffer()});
+ OnComplete({.Request = CacheChunkRequests[Index], .RequestIndex = Index, .Value = IoBuffer()});
}
return {.Error{.ErrorCode = Result.ErrorCode, .Reason = std::move(Result.Reason)}};
@@ -801,11 +802,11 @@ namespace detail {
virtual PutUpstreamCacheResult PutCacheRecord(const UpstreamCacheRecord& CacheRecord,
IoBuffer RecordValue,
- std::span<IoBuffer const> Payloads) override
+ std::span<IoBuffer const> Values) override
{
ZEN_TRACE_CPU("Upstream::Zen::PutCacheRecord");
- ZEN_ASSERT(CacheRecord.PayloadIds.size() == Payloads.size());
+ ZEN_ASSERT(CacheRecord.ValueContentIds.size() == Values.size());
const int32_t MaxAttempts = 3;
try
@@ -820,9 +821,9 @@ namespace detail {
CbPackage Package;
Package.SetObject(CbObject(SharedBuffer(RecordValue)));
- for (const IoBuffer& Payload : Payloads)
+ for (const IoBuffer& Value : Values)
{
- if (CompressedBuffer AttachmentBuffer = CompressedBuffer::FromCompressed(SharedBuffer(Payload)))
+ if (CompressedBuffer AttachmentBuffer = CompressedBuffer::FromCompressed(SharedBuffer(Value)))
{
Package.AddAttachment(CbAttachment(AttachmentBuffer));
}
@@ -848,15 +849,15 @@ namespace detail {
}
else
{
- for (size_t Idx = 0, Count = Payloads.size(); Idx < Count; Idx++)
+ for (size_t Idx = 0, Count = Values.size(); Idx < Count; Idx++)
{
Result.Success = false;
for (uint32_t Attempt = 0; Attempt < MaxAttempts && !Result.Success; Attempt++)
{
- Result = Session.PutCachePayload(CacheRecord.Key.Bucket,
- CacheRecord.Key.Hash,
- CacheRecord.PayloadIds[Idx],
- Payloads[Idx]);
+ Result = Session.PutCacheValue(CacheRecord.Key.Bucket,
+ CacheRecord.Key.Hash,
+ CacheRecord.ValueContentIds[Idx],
+ Values[Idx]);
}
m_Status.SetFromErrorCode(Result.ErrorCode, Result.Reason);
@@ -866,7 +867,7 @@ namespace detail {
if (!Result.Success)
{
- return {.Reason = "Failed to upload payload",
+ return {.Reason = "Failed to upload value",
.Bytes = TotalBytes,
.ElapsedSeconds = TotalElapsedSeconds,
.Success = false};
@@ -1115,11 +1116,11 @@ public:
}
}
- virtual void GetCachePayloads(std::span<CacheChunkRequest> CacheChunkRequests,
- std::span<size_t> RequestIndex,
- OnCachePayloadGetComplete&& OnComplete) override final
+ virtual void GetCacheValues(std::span<CacheChunkRequest> CacheChunkRequests,
+ std::span<size_t> RequestIndex,
+ OnCacheValueGetComplete&& OnComplete) override final
{
- ZEN_TRACE_CPU("Upstream::GetCachePayloads");
+ ZEN_TRACE_CPU("Upstream::GetCacheValues");
std::shared_lock<std::shared_mutex> _(m_EndpointsMutex);
@@ -1145,10 +1146,10 @@ public:
{
metrics::OperationTiming::Scope Scope(Endpoint->Stats().CacheGetRequestTiming);
- Result = Endpoint->GetCachePayloads(CacheChunkRequests, RemainingKeys, [&](CachePayloadGetCompleteParams&& Params) {
- if (Params.Payload)
+ Result = Endpoint->GetCacheValues(CacheChunkRequests, RemainingKeys, [&](CacheValueGetCompleteParams&& Params) {
+ if (Params.Value)
{
- OnComplete(std::forward<CachePayloadGetCompleteParams>(Params));
+ OnComplete(std::forward<CacheValueGetCompleteParams>(Params));
Stats.CacheHitCount.Increment(1);
}
@@ -1166,7 +1167,7 @@ public:
{
Stats.CacheErrorCount.Increment(1);
- ZEN_ERROR("get cache payloads(s) (rpc) FAILED, endpoint '{}', reason '{}', error code '{}'",
+ ZEN_ERROR("get cache values(s) (rpc) FAILED, endpoint '{}', reason '{}', error code '{}'",
Endpoint->GetEndpointInfo().Url,
Result.Error.Reason,
Result.Error.ErrorCode);
@@ -1178,13 +1179,13 @@ public:
for (size_t Index : RemainingKeys)
{
- OnComplete({.Request = CacheChunkRequests[Index], .RequestIndex = Index, .Payload = IoBuffer()});
+ OnComplete({.Request = CacheChunkRequests[Index], .RequestIndex = Index, .Value = IoBuffer()});
}
}
- virtual GetUpstreamCacheResult GetCachePayload(const CacheKey& CacheKey, const IoHash& PayloadId) override
+ virtual GetUpstreamCacheResult GetCacheValue(const CacheKey& CacheKey, const IoHash& ValueContentId) override
{
- ZEN_TRACE_CPU("Upstream::GetCachePayload");
+ ZEN_TRACE_CPU("Upstream::GetCacheValue");
if (m_Options.ReadUpstream)
{
@@ -1200,7 +1201,7 @@ public:
{
metrics::OperationTiming::Scope Scope(Stats.CacheGetRequestTiming);
- Result = Endpoint->GetCachePayload(CacheKey, PayloadId);
+ Result = Endpoint->GetCacheValue(CacheKey, ValueContentId);
}
Stats.CacheGetCount.Increment(1);
@@ -1217,7 +1218,7 @@ public:
{
Stats.CacheErrorCount.Increment(1);
- ZEN_ERROR("get cache payload FAILED, endpoint '{}', reason '{}', error code '{}'",
+ ZEN_ERROR("get cache value FAILED, endpoint '{}', reason '{}', error code '{}'",
Endpoint->GetEndpointInfo().Url,
Result.Error.Reason,
Result.Error.ErrorCode);
@@ -1302,18 +1303,18 @@ private:
return;
}
- for (const IoHash& PayloadId : CacheRecord.PayloadIds)
+ for (const IoHash& ValueContentId : CacheRecord.ValueContentIds)
{
- if (IoBuffer Payload = m_CidStore.FindChunkByCid(PayloadId))
+ if (IoBuffer Payload = m_CidStore.FindChunkByCid(ValueContentId))
{
Payloads.push_back(Payload);
}
else
{
- ZEN_WARN("process upstream FAILED, '{}/{}/{}', payload doesn't exist in CAS",
+ ZEN_WARN("process upstream FAILED, '{}/{}/{}', ValueContentId doesn't exist in CAS",
CacheRecord.Key.Bucket,
CacheRecord.Key.Hash,
- PayloadId);
+ ValueContentId);
return;
}
}