aboutsummaryrefslogtreecommitdiff
path: root/zenserver
diff options
context:
space:
mode:
Diffstat (limited to 'zenserver')
-rw-r--r--zenserver/cache/structuredcache.cpp151
-rw-r--r--zenserver/cache/structuredcache.h5
-rw-r--r--zenserver/config.cpp41
-rw-r--r--zenserver/config.h10
-rw-r--r--zenserver/upstream/upstreamcache.cpp24
-rw-r--r--zenserver/upstream/upstreamcache.h4
-rw-r--r--zenserver/zenserver.cpp12
7 files changed, 209 insertions, 38 deletions
diff --git a/zenserver/cache/structuredcache.cpp b/zenserver/cache/structuredcache.cpp
index da2e8850e..c66b1f98d 100644
--- a/zenserver/cache/structuredcache.cpp
+++ b/zenserver/cache/structuredcache.cpp
@@ -25,12 +25,125 @@
#include <queue>
#include <thread>
+#include <gsl/gsl-lite.hpp>
+
namespace zen {
using namespace std::literals;
//////////////////////////////////////////////////////////////////////////
+namespace detail { namespace cacheopt {
+ constexpr std::string_view Local = "local"sv;
+ constexpr std::string_view Remote = "remote"sv;
+ constexpr std::string_view Data = "data"sv;
+ constexpr std::string_view Meta = "meta"sv;
+ constexpr std::string_view Value = "value"sv;
+ constexpr std::string_view Attachments = "attachments"sv;
+}} // namespace detail::cacheopt
+
+//////////////////////////////////////////////////////////////////////////
+
+enum class CachePolicy : uint8_t
+{
+ None = 0,
+ QueryLocal = 1 << 0,
+ QueryRemote = 1 << 1,
+ Query = QueryLocal | QueryRemote,
+ StoreLocal = 1 << 2,
+ StoreRemote = 1 << 3,
+ Store = StoreLocal | StoreRemote,
+ SkipMeta = 1 << 4,
+ SkipValue = 1 << 5,
+ SkipAttachments = 1 << 6,
+ SkipData = SkipMeta | SkipValue | SkipAttachments,
+ SkipLocalCopy = 1 << 7,
+ Local = QueryLocal | StoreLocal,
+ Remote = QueryRemote | StoreRemote,
+ Default = Query | Store,
+ Disable = None,
+};
+
+gsl_DEFINE_ENUM_BITMASK_OPERATORS(CachePolicy);
+
+CachePolicy
+ParseCachePolicy(const zen::HttpServerRequest::QueryParams& QueryParams)
+{
+ CachePolicy QueryPolicy = CachePolicy::Query;
+
+ {
+ std::string_view Opts = QueryParams.GetValue("query"sv);
+ if (!Opts.empty())
+ {
+ QueryPolicy = CachePolicy::None;
+ zen::ForEachStrTok(Opts, ',', [&QueryPolicy](const std::string_view& Opt) {
+ if (Opt == detail::cacheopt::Local)
+ {
+ QueryPolicy |= CachePolicy::QueryLocal;
+ }
+ if (Opt == detail::cacheopt::Remote)
+ {
+ QueryPolicy |= CachePolicy::QueryRemote;
+ }
+ return true;
+ });
+ }
+ }
+
+ CachePolicy StorePolicy = CachePolicy::Store;
+
+ {
+ std::string_view Opts = QueryParams.GetValue("store"sv);
+ if (!Opts.empty())
+ {
+ StorePolicy = CachePolicy::None;
+ zen::ForEachStrTok(Opts, ',', [&StorePolicy](const std::string_view& Opt) {
+ if (Opt == detail::cacheopt::Local)
+ {
+ StorePolicy |= CachePolicy::StoreLocal;
+ }
+ if (Opt == detail::cacheopt::Remote)
+ {
+ StorePolicy |= CachePolicy::StoreRemote;
+ }
+ return true;
+ });
+ }
+ }
+
+ CachePolicy SkipPolicy = CachePolicy::None;
+
+ {
+ std::string_view Opts = QueryParams.GetValue("skip"sv);
+ if (!Opts.empty())
+ {
+ zen::ForEachStrTok(Opts, ',', [&SkipPolicy](const std::string_view& Opt) {
+ if (Opt == detail::cacheopt::Meta)
+ {
+ SkipPolicy |= CachePolicy::SkipMeta;
+ }
+ if (Opt == detail::cacheopt::Value)
+ {
+ SkipPolicy |= CachePolicy::SkipValue;
+ }
+ if (Opt == detail::cacheopt::Attachments)
+ {
+ SkipPolicy |= CachePolicy::SkipAttachments;
+ }
+ if (Opt == detail::cacheopt::Data)
+ {
+ SkipPolicy |= CachePolicy::SkipData;
+ }
+ return true;
+ });
+ }
+ }
+
+ return QueryPolicy | StorePolicy | SkipPolicy;
+}
+
+//////////////////////////////////////////////////////////////////////////
+
HttpStructuredCacheService::HttpStructuredCacheService(::ZenCacheStore& InCacheStore,
zen::CasStore& InStore,
zen::CidStore& InCidStore,
@@ -78,13 +191,16 @@ HttpStructuredCacheService::HandleRequest(zen::HttpServerRequest& Request)
return Request.WriteResponse(zen::HttpResponseCode::BadRequest); // invalid URL
}
+ const auto QueryParams = Request.GetQueryParams();
+ CachePolicy Policy = ParseCachePolicy(QueryParams);
+
if (Ref.PayloadId == IoHash::Zero)
{
- return HandleCacheRecordRequest(Request, Ref);
+ return HandleCacheRecordRequest(Request, Ref, Policy);
}
else
{
- return HandleCachePayloadRequest(Request, Ref);
+ return HandleCachePayloadRequest(Request, Ref, Policy);
}
return;
@@ -121,7 +237,7 @@ HttpStructuredCacheService::HandleCacheBucketRequest(zen::HttpServerRequest& Req
}
void
-HttpStructuredCacheService::HandleCacheRecordRequest(zen::HttpServerRequest& Request, CacheRef& Ref)
+HttpStructuredCacheService::HandleCacheRecordRequest(zen::HttpServerRequest& Request, CacheRef& Ref, CachePolicy Policy)
{
switch (auto Verb = Request.RequestVerb())
{
@@ -136,7 +252,10 @@ HttpStructuredCacheService::HandleCacheRecordRequest(zen::HttpServerRequest& Req
bool Success = m_CacheStore.Get(Ref.BucketSegment, Ref.HashKey, /* out */ Value);
bool InUpstreamCache = false;
- if (!Success && m_UpstreamCache)
+ const bool QueryUpstream =
+ !Success && m_UpstreamCache && (zen::CachePolicy::QueryRemote == (Policy & zen::CachePolicy::QueryRemote));
+
+ if (QueryUpstream)
{
const ZenContentType CacheRecordType = Ref.BucketSegment == "legacy"sv ? ZenContentType::kBinary
: AcceptType == ZenContentType::kCbPackage ? ZenContentType::kCbPackage
@@ -326,13 +445,15 @@ HttpStructuredCacheService::HandleCacheRecordRequest(zen::HttpServerRequest& Req
const HttpContentType ContentType = Request.RequestContentType();
+ const bool StoreUpstream = m_UpstreamCache && (zen::CachePolicy::StoreRemote == (Policy & zen::CachePolicy::StoreRemote));
+
if (ContentType == HttpContentType::kBinary || ContentType == HttpContentType::kUnknownContentType)
{
// TODO: create a cache record and put value in CAS?
m_CacheStore.Put(Ref.BucketSegment, Ref.HashKey, {.Value = Body});
ZEN_DEBUG("PUT - binary '{}/{}' {}", Ref.BucketSegment, Ref.HashKey, NiceBytes(Body.Size()));
- if (m_UpstreamCache)
+ if (StoreUpstream)
{
auto Result = m_UpstreamCache->EnqueueUpstream(
{.Type = ZenContentType::kBinary, .CacheKey = {Ref.BucketSegment, Ref.HashKey}});
@@ -397,15 +518,12 @@ HttpStructuredCacheService::HandleCacheRecordRequest(zen::HttpServerRequest& Req
MissingRefs.size(),
References.size());
- if (MissingRefs.empty())
+ if (MissingRefs.empty() && StoreUpstream)
{
- // Only enqueue valid cache records, i.e. all referenced payloads exists
- if (m_UpstreamCache)
- {
- auto Result = m_UpstreamCache->EnqueueUpstream({.Type = ZenContentType::kCbObject,
- .CacheKey = {Ref.BucketSegment, Ref.HashKey},
- .PayloadIds = std::move(References)});
- }
+ ZEN_ASSERT(m_UpstreamCache);
+ auto Result = m_UpstreamCache->EnqueueUpstream({.Type = ZenContentType::kCbObject,
+ .CacheKey = {Ref.BucketSegment, Ref.HashKey},
+ .PayloadIds = std::move(References)});
return Request.WriteResponse(zen::HttpResponseCode::Created);
}
@@ -502,8 +620,9 @@ HttpStructuredCacheService::HandleCacheRecordRequest(zen::HttpServerRequest& Req
ZenCacheValue CacheValue{.Value = CacheRecordChunk};
m_CacheStore.Put(Ref.BucketSegment, Ref.HashKey, CacheValue);
- if (m_UpstreamCache)
+ if (StoreUpstream)
{
+ ZEN_ASSERT(m_UpstreamCache);
auto Result = m_UpstreamCache->EnqueueUpstream({.Type = ZenContentType::kCbPackage,
.CacheKey = {Ref.BucketSegment, Ref.HashKey},
.PayloadIds = std::move(PayloadIds)});
@@ -536,7 +655,7 @@ HttpStructuredCacheService::HandleCacheRecordRequest(zen::HttpServerRequest& Req
}
void
-HttpStructuredCacheService::HandleCachePayloadRequest(zen::HttpServerRequest& Request, CacheRef& Ref)
+HttpStructuredCacheService::HandleCachePayloadRequest(zen::HttpServerRequest& Request, CacheRef& Ref, CachePolicy Policy)
{
// Note: the URL references the uncompressed payload hash - so this maintains the mapping
// from uncompressed CAS identity (aka CID/Content ID) to the stored payload hash
@@ -544,6 +663,8 @@ HttpStructuredCacheService::HandleCachePayloadRequest(zen::HttpServerRequest& Re
// this is a PITA but a consequence of the fact that the client side code is not able to
// address data by compressed hash
+ ZEN_UNUSED(Policy);
+
switch (auto Verb = Request.RequestVerb())
{
using enum zen::HttpVerb;
diff --git a/zenserver/cache/structuredcache.h b/zenserver/cache/structuredcache.h
index 8289fd700..796b21d1f 100644
--- a/zenserver/cache/structuredcache.h
+++ b/zenserver/cache/structuredcache.h
@@ -17,6 +17,7 @@ namespace zen {
class CasStore;
class CidStore;
class UpstreamCache;
+enum class CachePolicy : uint8_t;
/**
* Structured cache service. Imposes constraints on keys, supports blobs and
@@ -70,8 +71,8 @@ private:
};
[[nodiscard]] bool ValidateKeyUri(zen::HttpServerRequest& Request, CacheRef& OutRef);
- void HandleCacheRecordRequest(zen::HttpServerRequest& Request, CacheRef& Ref);
- void HandleCachePayloadRequest(zen::HttpServerRequest& Request, CacheRef& Ref);
+ void HandleCacheRecordRequest(zen::HttpServerRequest& Request, CacheRef& Ref, CachePolicy Policy);
+ void HandleCachePayloadRequest(zen::HttpServerRequest& Request, CacheRef& Ref, CachePolicy Policy);
void HandleCacheBucketRequest(zen::HttpServerRequest& Request, std::string_view Bucket);
spdlog::logger& Log() { return m_Log; }
diff --git a/zenserver/config.cpp b/zenserver/config.cpp
index 092fc6998..164d2a792 100644
--- a/zenserver/config.cpp
+++ b/zenserver/config.cpp
@@ -55,6 +55,27 @@ PickDefaultStateDirectory()
#endif
+UpstreamCachePolicy
+ParseUpstreamCachePolicy(std::string_view Options)
+{
+ if (Options == "readonly")
+ {
+ return UpstreamCachePolicy::Read;
+ }
+ else if (Options == "writeonly")
+ {
+ return UpstreamCachePolicy::Write;
+ }
+ else if (Options == "disabled")
+ {
+ return UpstreamCachePolicy::Disabled;
+ }
+ else
+ {
+ return UpstreamCachePolicy::ReadWrite;
+ }
+}
+
void
ParseGlobalCliOptions(int argc, char* argv[], ZenServerOptions& GlobalOptions, ZenServiceConfig& ServiceConfig)
{
@@ -113,6 +134,14 @@ ParseGlobalCliOptions(int argc, char* argv[], ZenServerOptions& GlobalOptions, Z
cxxopts::value<bool>(ServiceConfig.ShouldCrash)->default_value("false"),
"");
+ std::string UpstreamCachePolicyOptions;
+ options.add_option("cache",
+ "",
+ "upstream-cache-policy",
+ "",
+ cxxopts::value<std::string>(UpstreamCachePolicyOptions)->default_value(""),
+ "Upstream cache policy (readwrite|readonly|writeonly|disabled)");
+
options.add_option("cache",
"",
"upstream-jupiter-url",
@@ -178,13 +207,6 @@ ParseGlobalCliOptions(int argc, char* argv[], ZenServerOptions& GlobalOptions, Z
options.add_option("cache",
"",
- "upstream-enabled",
- "Whether upstream caching is disabled",
- cxxopts::value<bool>(ServiceConfig.UpstreamCacheConfig.Enabled)->default_value("true"),
- "");
-
- options.add_option("cache",
- "",
"upstream-thread-count",
"Number of threads used for upstream procsssing",
cxxopts::value<int>(ServiceConfig.UpstreamCacheConfig.UpstreamThreadCount)->default_value("4"),
@@ -200,6 +222,8 @@ ParseGlobalCliOptions(int argc, char* argv[], ZenServerOptions& GlobalOptions, Z
exit(0);
}
+
+ ServiceConfig.UpstreamCacheConfig.CachePolicy = ParseUpstreamCachePolicy(UpstreamCachePolicyOptions);
}
catch (cxxopts::OptionParseException& e)
{
@@ -276,7 +300,8 @@ ParseServiceConfig(const std::filesystem::path& DataRoot, ZenServiceConfig& Serv
if (auto UpstreamConfig = StructuredCacheConfig->get<sol::optional<sol::table>>("upstream"))
{
- ServiceConfig.UpstreamCacheConfig.Enabled = UpstreamConfig->get_or("enable", ServiceConfig.UpstreamCacheConfig.Enabled);
+ std::string Policy = UpstreamConfig->get_or("policy", std::string());
+ ServiceConfig.UpstreamCacheConfig.CachePolicy = ParseUpstreamCachePolicy(Policy);
ServiceConfig.UpstreamCacheConfig.UpstreamThreadCount = UpstreamConfig->get_or("upstreamthreadcount", 4);
if (auto JupiterConfig = UpstreamConfig->get<sol::optional<sol::table>>("jupiter"))
diff --git a/zenserver/config.h b/zenserver/config.h
index c06102384..6ade1b401 100644
--- a/zenserver/config.h
+++ b/zenserver/config.h
@@ -36,12 +36,20 @@ struct ZenUpstreamZenConfig
std::string Url;
};
+enum class UpstreamCachePolicy : uint8_t
+{
+ Disabled = 0,
+ Read = 1 << 0,
+ Write = 1 << 1,
+ ReadWrite = Read | Write
+};
+
struct ZenUpstreamCacheConfig
{
ZenUpstreamJupiterConfig JupiterConfig;
ZenUpstreamZenConfig ZenConfig;
int UpstreamThreadCount = 4;
- bool Enabled = false;
+ UpstreamCachePolicy CachePolicy = UpstreamCachePolicy::ReadWrite;
};
struct ZenServiceConfig
diff --git a/zenserver/upstream/upstreamcache.cpp b/zenserver/upstream/upstreamcache.cpp
index 38d30a795..c015ef3e9 100644
--- a/zenserver/upstream/upstreamcache.cpp
+++ b/zenserver/upstream/upstreamcache.cpp
@@ -559,12 +559,15 @@ public:
virtual GetUpstreamCacheResult GetCacheRecord(UpstreamCacheKey CacheKey, ZenContentType Type) override
{
- for (auto& Endpoint : m_Endpoints)
+ if (m_Options.ReadUpstream)
{
- if (GetUpstreamCacheResult Result = Endpoint->GetCacheRecord(CacheKey, Type); Result.Success)
+ for (auto& Endpoint : m_Endpoints)
{
- m_Stats.Add(*Endpoint, Result);
- return Result;
+ if (GetUpstreamCacheResult Result = Endpoint->GetCacheRecord(CacheKey, Type); Result.Success)
+ {
+ m_Stats.Add(*Endpoint, Result);
+ return Result;
+ }
}
}
@@ -573,12 +576,15 @@ public:
virtual GetUpstreamCacheResult GetCachePayload(UpstreamPayloadKey PayloadKey) override
{
- for (auto& Endpoint : m_Endpoints)
+ if (m_Options.ReadUpstream)
{
- if (GetUpstreamCacheResult Result = Endpoint->GetCachePayload(PayloadKey); Result.Success)
+ for (auto& Endpoint : m_Endpoints)
{
- m_Stats.Add(*Endpoint, Result);
- return Result;
+ if (GetUpstreamCacheResult Result = Endpoint->GetCachePayload(PayloadKey); Result.Success)
+ {
+ m_Stats.Add(*Endpoint, Result);
+ return Result;
+ }
}
}
@@ -587,7 +593,7 @@ public:
virtual EnqueueResult EnqueueUpstream(UpstreamCacheRecord CacheRecord) override
{
- if (m_IsRunning.load())
+ if (m_IsRunning.load() && m_Options.WriteUpstream)
{
if (!m_UpstreamThreads.empty())
{
diff --git a/zenserver/upstream/upstreamcache.h b/zenserver/upstream/upstreamcache.h
index 327778452..3b054d815 100644
--- a/zenserver/upstream/upstreamcache.h
+++ b/zenserver/upstream/upstreamcache.h
@@ -36,7 +36,9 @@ struct UpstreamCacheRecord
struct UpstreamCacheOptions
{
- uint32_t ThreadCount = 4;
+ uint32_t ThreadCount = 4;
+ bool ReadUpstream = true;
+ bool WriteUpstream = true;
};
struct GetUpstreamCacheResult
diff --git a/zenserver/zenserver.cpp b/zenserver/zenserver.cpp
index f36cfba48..83580b288 100644
--- a/zenserver/zenserver.cpp
+++ b/zenserver/zenserver.cpp
@@ -166,11 +166,15 @@ public:
m_CacheStore = std::make_unique<ZenCacheStore>(*m_CasStore, m_DataRoot / "cache");
std::unique_ptr<zen::UpstreamCache> UpstreamCache;
- if (ServiceConfig.UpstreamCacheConfig.Enabled)
+ if (ServiceConfig.UpstreamCacheConfig.CachePolicy != UpstreamCachePolicy::Disabled)
{
const ZenUpstreamCacheConfig& UpstreamConfig = ServiceConfig.UpstreamCacheConfig;
zen::UpstreamCacheOptions UpstreamOptions;
+ UpstreamOptions.ReadUpstream =
+ (uint8_t(ServiceConfig.UpstreamCacheConfig.CachePolicy) & uint8_t(UpstreamCachePolicy::Read)) != 0;
+ UpstreamOptions.WriteUpstream =
+ (uint8_t(ServiceConfig.UpstreamCacheConfig.CachePolicy) & uint8_t(UpstreamCachePolicy::Write)) != 0;
if (UpstreamConfig.UpstreamThreadCount < 32)
{
@@ -216,7 +220,11 @@ public:
if (UpstreamCache->Initialize())
{
- ZEN_INFO("upstream cache active");
+ ZEN_INFO("upstream cache active ({})",
+ UpstreamOptions.ReadUpstream && UpstreamOptions.WriteUpstream ? "READ|WRITE"
+ : UpstreamOptions.ReadUpstream ? "READONLY"
+ : UpstreamOptions.WriteUpstream ? "WRITEONLY"
+ : "DISABLED");
}
else
{
generated by cgit v1.2.3 (git 2.25.1) at 2026-04-22 17:43:57 +0000