aboutsummaryrefslogtreecommitdiff
path: root/src
diff options
context:
space:
mode:
authorDan Engelbrecht <[email protected]>2025-08-26 11:43:37 +0200
committerGitHub Enterprise <[email protected]>2025-08-26 11:43:37 +0200
commitfb137cf9c8b7a9d1659b03472c9591c4863e9173 (patch)
treebbd0df09d4b425ef668d22f3b12ea2cb3482bf66 /src
parentMerge pull request #139 from ue-foundation/de/zen-service-command (diff)
downloadzen-fb137cf9c8b7a9d1659b03472c9591c4863e9173.tar.xz
zen-fb137cf9c8b7a9d1659b03472c9591c4863e9173.zip
revert multi-cid store (#475)
Diffstat (limited to 'src')
-rw-r--r--src/zenserver/cache/httpstructuredcache.cpp286
-rw-r--r--src/zenserver/cache/httpstructuredcache.h5
-rw-r--r--src/zenserver/projectstore/httpprojectstore.cpp167
-rw-r--r--src/zenserver/projectstore/httpprojectstore.h7
-rw-r--r--src/zenserver/projectstore/projectstore.cpp169
-rw-r--r--src/zenserver/projectstore/projectstore.h8
-rw-r--r--src/zenserver/upstream/upstreamcache.cpp38
-rw-r--r--src/zenserver/upstream/upstreamcache.h3
-rw-r--r--src/zenserver/zenserver.cpp38
-rw-r--r--src/zenserver/zenserver.h28
-rw-r--r--src/zenstore/cache/cacherpc.cpp54
-rw-r--r--src/zenstore/include/zenstore/cache/cacherpc.h8
-rw-r--r--src/zenstore/include/zenstore/cache/upstreamcacheclient.h2
13 files changed, 260 insertions, 553 deletions
diff --git a/src/zenserver/cache/httpstructuredcache.cpp b/src/zenserver/cache/httpstructuredcache.cpp
index 19ac3a216..68f1c602e 100644
--- a/src/zenserver/cache/httpstructuredcache.cpp
+++ b/src/zenserver/cache/httpstructuredcache.cpp
@@ -85,7 +85,7 @@ namespace {
//////////////////////////////////////////////////////////////////////////
HttpStructuredCacheService::HttpStructuredCacheService(ZenCacheStore& InCacheStore,
- GetCidStoreFunc&& GetCidStore,
+ CidStore& InCidStore,
HttpStatsService& StatsService,
HttpStatusService& StatusService,
UpstreamCache& UpstreamCache,
@@ -95,10 +95,11 @@ HttpStructuredCacheService::HttpStructuredCacheService(ZenCacheStore& InCach
, m_CacheStore(InCacheStore)
, m_StatsService(StatsService)
, m_StatusService(StatusService)
+, m_CidStore(InCidStore)
, m_UpstreamCache(UpstreamCache)
, m_DiskWriteBlocker(InDiskWriteBlocker)
, m_OpenProcessCache(InOpenProcessCache)
-, m_RpcHandler(m_Log, m_CacheStats, UpstreamCache, InCacheStore, std::move(GetCidStore), InDiskWriteBlocker)
+, m_RpcHandler(m_Log, m_CacheStats, UpstreamCache, InCacheStore, InCidStore, InDiskWriteBlocker)
{
m_StatsService.RegisterHandler("z$", *this);
m_StatusService.RegisterHandler("z$", *this);
@@ -224,9 +225,6 @@ HttpStructuredCacheService::HandleDetailsRequest(HttpServerRequest& Request)
for (const auto& NamespaceIt : ValueDetails.Namespaces)
{
const std::string& Namespace = NamespaceIt.first;
-
- CidStore& ChunkStore = m_RpcHandler.GetCidStore(Namespace);
-
for (const auto& BucketIt : NamespaceIt.second.Buckets)
{
const std::string& Bucket = BucketIt.first;
@@ -236,7 +234,7 @@ HttpStructuredCacheService::HandleDetailsRequest(HttpServerRequest& Request)
{
for (const IoHash& Hash : ValueIt.second.Attachments)
{
- IoBuffer Payload = ChunkStore.FindChunkByCid(Hash);
+ IoBuffer Payload = m_CidStore.FindChunkByCid(Hash);
CSVWriter << "\r\n"
<< Namespace << "," << Bucket << "," << ValueIt.first.ToHexString() << ", " << Hash.ToHexString()
<< ", " << gsl::narrow<uint64_t>(Payload.GetSize());
@@ -254,7 +252,7 @@ HttpStructuredCacheService::HandleDetailsRequest(HttpServerRequest& Request)
size_t AttachmentsSize = 0;
for (const IoHash& Hash : ValueIt.second.Attachments)
{
- IoBuffer Payload = ChunkStore.FindChunkByCid(Hash);
+ IoBuffer Payload = m_CidStore.FindChunkByCid(Hash);
AttachmentsSize += Payload.GetSize();
}
CSVWriter << ", " << gsl::narrow<uint64_t>(AttachmentsSize);
@@ -276,9 +274,6 @@ HttpStructuredCacheService::HandleDetailsRequest(HttpServerRequest& Request)
for (const auto& NamespaceIt : ValueDetails.Namespaces)
{
const std::string& Namespace = NamespaceIt.first;
-
- CidStore& ChunkStore = m_RpcHandler.GetCidStore(Namespace);
-
Cbo.BeginObject();
{
Cbo.AddString("name", Namespace);
@@ -321,7 +316,7 @@ HttpStructuredCacheService::HandleDetailsRequest(HttpServerRequest& Request)
{
Cbo.BeginObject();
Cbo.AddHash("cid", Hash);
- IoBuffer Payload = ChunkStore.FindChunkByCid(Hash);
+ IoBuffer Payload = m_CidStore.FindChunkByCid(Hash);
Cbo.AddInteger("size", gsl::narrow<uint64_t>(Payload.GetSize()));
Cbo.EndObject();
}
@@ -335,7 +330,7 @@ HttpStructuredCacheService::HandleDetailsRequest(HttpServerRequest& Request)
size_t AttachmentsSize = 0;
for (const IoHash& Hash : ValueIt.second.Attachments)
{
- IoBuffer Payload = ChunkStore.FindChunkByCid(Hash);
+ IoBuffer Payload = m_CidStore.FindChunkByCid(Hash);
AttachmentsSize += Payload.GetSize();
}
Cbo.AddInteger("attachmentssize", gsl::narrow<uint64_t>(AttachmentsSize));
@@ -610,8 +605,6 @@ HttpStructuredCacheService::HandleCacheNamespaceRequest(HttpServerRequest& Reque
ResponseWriter.AddInteger("EntryCount", Info->DiskLayerInfo.EntryCount);
- CidStore& ChunkStore = m_RpcHandler.GetCidStore(NamespaceName);
-
if (auto Buckets = HttpServerRequest::Decode(Request.GetQueryParams().GetValue("bucketsizes")); !Buckets.empty())
{
ResponseWriter.BeginObject("BucketSizes");
@@ -670,7 +663,7 @@ HttpStructuredCacheService::HandleCacheNamespaceRequest(HttpServerRequest& Reque
uint64_t AttachmentsSize = 0;
- ChunkStore.IterateChunks(
+ m_CidStore.IterateChunks(
AllAttachments,
[&](size_t Index, const IoBuffer& Payload) {
ZEN_UNUSED(Index);
@@ -738,8 +731,6 @@ HttpStructuredCacheService::HandleCacheBucketRequest(HttpServerRequest& Request,
ResponseWriter.AddInteger("DiskEntryCount", Info->DiskLayerInfo.EntryCount);
- CidStore& ChunkStore = m_RpcHandler.GetCidStore(NamespaceName);
-
if (auto GetBucketSize = Request.GetQueryParams().GetValue("bucketsize"); GetBucketSize == "true")
{
CacheContentStats ContentStats;
@@ -766,7 +757,7 @@ HttpStructuredCacheService::HandleCacheBucketRequest(HttpServerRequest& Request,
WorkerThreadPool& WorkerPool = GetMediumWorkerPool(EWorkloadType::Background);
- ChunkStore.IterateChunks(
+ m_CidStore.IterateChunks(
ContentStats.Attachments,
[&](size_t Index, const IoBuffer& Payload) {
ZEN_UNUSED(Index);
@@ -841,8 +832,6 @@ HttpStructuredCacheService::HandleGetCacheRecord(HttpServerRequest& Request, con
CacheRequestContext RequestContext = {.SessionId = Request.SessionId(), .RequestId = Request.RequestId()};
Stopwatch Timer;
- CidStore& ChunkStore = m_RpcHandler.GetCidStore(Ref.Namespace);
-
if (EnumHasAllFlags(PolicyFromUrl, CachePolicy::QueryLocal) &&
m_CacheStore.Get(RequestContext, Ref.Namespace, Ref.BucketSegment, Ref.HashKey, ClientResultValue))
{
@@ -857,17 +846,17 @@ HttpStructuredCacheService::HandleGetCacheRecord(HttpServerRequest& Request, con
uint32_t MissingCount = 0;
CbObjectView CacheRecord(ClientResultValue.Value.Data());
- CacheRecord.IterateAttachments([this, &ChunkStore, &MissingCount, &Package, SkipData](CbFieldView AttachmentHash) {
+ CacheRecord.IterateAttachments([this, &MissingCount, &Package, SkipData](CbFieldView AttachmentHash) {
if (SkipData)
{
- if (!ChunkStore.ContainsChunk(AttachmentHash.AsHash()))
+ if (!m_CidStore.ContainsChunk(AttachmentHash.AsHash()))
{
MissingCount++;
}
}
else
{
- if (IoBuffer Chunk = ChunkStore.FindChunkByCid(AttachmentHash.AsHash()))
+ if (IoBuffer Chunk = m_CidStore.FindChunkByCid(AttachmentHash.AsHash()))
{
CompressedBuffer Compressed = CompressedBuffer::FromCompressedNoValidate(std::move(Chunk));
if (Compressed)
@@ -967,8 +956,6 @@ HttpStructuredCacheService::HandleGetCacheRecord(HttpServerRequest& Request, con
{
Success = true;
- CidStore& ChunkStore = m_RpcHandler.GetCidStore(Ref.Namespace);
-
ClientResultValue.Value = UpstreamResult.Value;
ClientResultValue.Value.SetContentType(AcceptType);
@@ -1019,7 +1006,6 @@ HttpStructuredCacheService::HandleGetCacheRecord(HttpServerRequest& Request, con
CacheRecord.IterateAttachments([this,
&Package,
&Ref,
- &ChunkStore,
&WriteAttachmentBuffers,
&WriteRawHashes,
&ReferencedAttachments,
@@ -1054,12 +1040,12 @@ HttpStructuredCacheService::HandleGetCacheRecord(HttpServerRequest& Request, con
{
if (SkipData)
{
- if (ChunkStore.ContainsChunk(Hash))
+ if (m_CidStore.ContainsChunk(Hash))
{
Count.Valid++;
}
}
- else if (IoBuffer Chunk = ChunkStore.FindChunkByCid(Hash))
+ else if (IoBuffer Chunk = m_CidStore.FindChunkByCid(Hash))
{
CompressedBuffer Compressed = CompressedBuffer::FromCompressedNoValidate(std::move(Chunk));
if (Compressed)
@@ -1101,7 +1087,7 @@ HttpStructuredCacheService::HandleGetCacheRecord(HttpServerRequest& Request, con
if (!WriteAttachmentBuffers.empty())
{
std::vector<CidStore::InsertResult> InsertResults =
- ChunkStore.AddChunks(WriteAttachmentBuffers, WriteRawHashes);
+ m_CidStore.AddChunks(WriteAttachmentBuffers, WriteRawHashes);
for (const CidStore::InsertResult& Result : InsertResults)
{
if (Result.New)
@@ -1268,10 +1254,7 @@ HttpStructuredCacheService::HandlePutCacheRecord(HttpServerRequest& Request, con
if (HasUpstream && EnumHasAllFlags(PolicyFromUrl, CachePolicy::StoreRemote))
{
- CidStore& ChunkStore = m_RpcHandler.GetCidStore(Ref.Namespace);
- m_UpstreamCache.EnqueueUpstream(
- {.Type = ContentType, .Namespace = Ref.Namespace, .Key = {Ref.BucketSegment, Ref.HashKey}},
- [ChunkStore = &ChunkStore](const IoHash& ValueHash) { return ChunkStore->FindChunkByCid(ValueHash); });
+ m_UpstreamCache.EnqueueUpstream({.Type = ContentType, .Namespace = Ref.Namespace, .Key = {Ref.BucketSegment, Ref.HashKey}});
}
ZEN_DEBUG("PUTCACHERECORD - '{}/{}/{}' {} '{}' in {}",
@@ -1305,18 +1288,15 @@ HttpStructuredCacheService::HandlePutCacheRecord(HttpServerRequest& Request, con
std::vector<IoHash> ReferencedAttachments;
int32_t TotalCount = 0;
- CidStore& ChunkStore = m_RpcHandler.GetCidStore(Ref.Namespace);
-
- CacheRecord.IterateAttachments(
- [this, &ChunkStore, &TotalCount, &ValidAttachments, &ReferencedAttachments](CbFieldView AttachmentHash) {
- const IoHash Hash = AttachmentHash.AsHash();
- ReferencedAttachments.push_back(Hash);
- if (ChunkStore.ContainsChunk(Hash))
- {
- ValidAttachments.emplace_back(Hash);
- }
- TotalCount++;
- });
+ CacheRecord.IterateAttachments([this, &TotalCount, &ValidAttachments, &ReferencedAttachments](CbFieldView AttachmentHash) {
+ const IoHash Hash = AttachmentHash.AsHash();
+ ReferencedAttachments.push_back(Hash);
+ if (m_CidStore.ContainsChunk(Hash))
+ {
+ ValidAttachments.emplace_back(Hash);
+ }
+ TotalCount++;
+ });
const bool Overwrite = !EnumHasAllFlags(PolicyFromUrl, CachePolicy::QueryLocal);
@@ -1350,12 +1330,10 @@ HttpStructuredCacheService::HandlePutCacheRecord(HttpServerRequest& Request, con
CachePolicy Policy = PolicyFromUrl;
if (HasUpstream && EnumHasAllFlags(Policy, CachePolicy::StoreRemote) && !IsPartialRecord)
{
- m_UpstreamCache.EnqueueUpstream(
- {.Type = ZenContentType::kCbObject,
- .Namespace = Ref.Namespace,
- .Key = {Ref.BucketSegment, Ref.HashKey},
- .ValueContentIds = std::move(ValidAttachments)},
- [ChunkStore = &ChunkStore](const IoHash& ValueHash) { return ChunkStore->FindChunkByCid(ValueHash); });
+ m_UpstreamCache.EnqueueUpstream({.Type = ZenContentType::kCbObject,
+ .Namespace = Ref.Namespace,
+ .Key = {Ref.BucketSegment, Ref.HashKey},
+ .ValueContentIds = std::move(ValidAttachments)});
}
Request.WriteResponse(HttpResponseCode::Created);
@@ -1388,46 +1366,38 @@ HttpStructuredCacheService::HandlePutCacheRecord(HttpServerRequest& Request, con
WriteAttachmentBuffers.reserve(NumAttachments);
WriteRawHashes.reserve(NumAttachments);
- CidStore& ChunkStore = m_RpcHandler.GetCidStore(Ref.Namespace);
-
- CacheRecord.IterateAttachments([this,
- &Ref,
- &Package,
- &ChunkStore,
- &WriteAttachmentBuffers,
- &WriteRawHashes,
- &ValidAttachments,
- &ReferencedAttachments,
- &Count](CbFieldView HashView) {
- const IoHash Hash = HashView.AsHash();
- ReferencedAttachments.push_back(Hash);
- if (const CbAttachment* Attachment = Package.FindAttachment(Hash))
- {
- if (Attachment->IsCompressedBinary())
+ CacheRecord.IterateAttachments(
+ [this, &Ref, &Package, &WriteAttachmentBuffers, &WriteRawHashes, &ValidAttachments, &ReferencedAttachments, &Count](
+ CbFieldView HashView) {
+ const IoHash Hash = HashView.AsHash();
+ ReferencedAttachments.push_back(Hash);
+ if (const CbAttachment* Attachment = Package.FindAttachment(Hash))
{
- WriteAttachmentBuffers.emplace_back(Attachment->AsCompressedBinary().GetCompressed().Flatten().AsIoBuffer());
- WriteRawHashes.push_back(Hash);
- ValidAttachments.emplace_back(Hash);
- Count.Valid++;
+ if (Attachment->IsCompressedBinary())
+ {
+ WriteAttachmentBuffers.emplace_back(Attachment->AsCompressedBinary().GetCompressed().Flatten().AsIoBuffer());
+ WriteRawHashes.push_back(Hash);
+ ValidAttachments.emplace_back(Hash);
+ Count.Valid++;
+ }
+ else
+ {
+ ZEN_WARN("PUTCACHERECORD - '{}/{}/{}' '{}' FAILED, attachment '{}' is not compressed",
+ Ref.Namespace,
+ Ref.BucketSegment,
+ Ref.HashKey,
+ ToString(HttpContentType::kCbPackage),
+ Hash);
+ Count.Invalid++;
+ }
}
- else
+ else if (m_CidStore.ContainsChunk(Hash))
{
- ZEN_WARN("PUTCACHERECORD - '{}/{}/{}' '{}' FAILED, attachment '{}' is not compressed",
- Ref.Namespace,
- Ref.BucketSegment,
- Ref.HashKey,
- ToString(HttpContentType::kCbPackage),
- Hash);
- Count.Invalid++;
+ ValidAttachments.emplace_back(Hash);
+ Count.Valid++;
}
- }
- else if (ChunkStore.ContainsChunk(Hash))
- {
- ValidAttachments.emplace_back(Hash);
- Count.Valid++;
- }
- Count.Total++;
- });
+ Count.Total++;
+ });
if (Count.Invalid > 0)
{
@@ -1451,7 +1421,7 @@ HttpStructuredCacheService::HandlePutCacheRecord(HttpServerRequest& Request, con
if (!WriteAttachmentBuffers.empty())
{
- std::vector<CidStore::InsertResult> InsertResults = ChunkStore.AddChunks(WriteAttachmentBuffers, WriteRawHashes);
+ std::vector<CidStore::InsertResult> InsertResults = m_CidStore.AddChunks(WriteAttachmentBuffers, WriteRawHashes);
for (const CidStore::InsertResult& InsertResult : InsertResults)
{
if (InsertResult.New)
@@ -1478,12 +1448,10 @@ HttpStructuredCacheService::HandlePutCacheRecord(HttpServerRequest& Request, con
if (HasUpstream && EnumHasAllFlags(Policy, CachePolicy::StoreRemote) && !IsPartialRecord)
{
- m_UpstreamCache.EnqueueUpstream(
- {.Type = ZenContentType::kCbPackage,
- .Namespace = Ref.Namespace,
- .Key = {Ref.BucketSegment, Ref.HashKey},
- .ValueContentIds = std::move(ValidAttachments)},
- [ChunkStore = &ChunkStore](const IoHash& ValueHash) { return ChunkStore->FindChunkByCid(ValueHash); });
+ m_UpstreamCache.EnqueueUpstream({.Type = ZenContentType::kCbPackage,
+ .Namespace = Ref.Namespace,
+ .Key = {Ref.BucketSegment, Ref.HashKey},
+ .ValueContentIds = std::move(ValidAttachments)});
}
Request.WriteResponse(HttpResponseCode::Created);
@@ -1517,9 +1485,7 @@ HttpStructuredCacheService::HandleGetCacheChunk(HttpServerRequest& Request, cons
{
Stopwatch Timer;
- CidStore& ChunkStore = m_RpcHandler.GetCidStore(Ref.Namespace);
-
- IoBuffer Value = ChunkStore.FindChunkByCid(Ref.ValueContentId);
+ IoBuffer Value = m_CidStore.FindChunkByCid(Ref.ValueContentId);
const UpstreamEndpointInfo* Source = nullptr;
CachePolicy Policy = PolicyFromUrl;
@@ -1541,7 +1507,7 @@ HttpStructuredCacheService::HandleGetCacheChunk(HttpServerRequest& Request, cons
{
if (AreDiskWritesAllowed())
{
- ChunkStore.AddChunk(UpstreamResult.Value, RawHash);
+ m_CidStore.AddChunk(UpstreamResult.Value, RawHash);
}
Source = UpstreamResult.Source;
}
@@ -1635,9 +1601,7 @@ HttpStructuredCacheService::HandlePutCacheChunk(HttpServerRequest& Request, cons
"ValueContentId does not match attachment hash"sv);
}
- CidStore& ChunkStore = m_RpcHandler.GetCidStore(Ref.Namespace);
-
- CidStore::InsertResult Result = ChunkStore.AddChunk(Body, RawHash);
+ CidStore::InsertResult Result = m_CidStore.AddChunk(Body, RawHash);
ZEN_DEBUG("PUTCACHECHUNK - '{}/{}/{}/{}' {} '{}' ({}) in {}",
Ref.Namespace,
@@ -1852,51 +1816,16 @@ HttpStructuredCacheService::HandleStatsRequest(HttpServerRequest& Request)
EmitSnapshot("requests", m_HttpRequests, Cbo);
- const uint64_t HitCount = m_CacheStats.HitCount;
- const uint64_t UpstreamHitCount = m_CacheStats.UpstreamHitCount;
- const uint64_t MissCount = m_CacheStats.MissCount;
- const uint64_t WriteCount = m_CacheStats.WriteCount;
- const uint64_t BadRequestCount = m_CacheStats.BadRequestCount;
-
- uint64_t TotalChunkHitCount = 0;
- uint64_t TotalChunkMissCount = 0;
- uint64_t TotalChunkWriteCount = 0;
- CidStoreSize TotalCidSize;
-
- tsl::robin_map<CidStore*, std::string> UniqueStores;
- {
- std::vector<std::string> NamespaceNames = m_CacheStore.GetNamespaces();
-
- for (const std::string& NamespaceName : NamespaceNames)
- {
- CidStore* Store = &m_RpcHandler.GetCidStore(NamespaceName);
- if (auto It = UniqueStores.find(Store); It == UniqueStores.end())
- {
- UniqueStores.insert_or_assign(Store, NamespaceName);
- }
- else
- {
- UniqueStores.insert_or_assign(Store, std::string{});
- }
- }
-
- for (auto It : UniqueStores)
- {
- CidStore* ChunkStore = It.first;
-
- CidStoreStats StoreStats = ChunkStore->Stats();
- CidStoreSize StoreSize = ChunkStore->TotalSize();
-
- TotalChunkHitCount += StoreStats.HitCount;
- TotalChunkMissCount += StoreStats.MissCount;
- TotalChunkWriteCount += StoreStats.WriteCount;
-
- TotalCidSize.TinySize += StoreSize.TinySize;
- TotalCidSize.SmallSize += StoreSize.SmallSize;
- TotalCidSize.LargeSize += StoreSize.LargeSize;
- TotalCidSize.TotalSize += StoreSize.TotalSize;
- }
- }
+ const uint64_t HitCount = m_CacheStats.HitCount;
+ const uint64_t UpstreamHitCount = m_CacheStats.UpstreamHitCount;
+ const uint64_t MissCount = m_CacheStats.MissCount;
+ const uint64_t WriteCount = m_CacheStats.WriteCount;
+ const uint64_t BadRequestCount = m_CacheStats.BadRequestCount;
+ struct CidStoreStats StoreStats = m_CidStore.Stats();
+ const uint64_t ChunkHitCount = StoreStats.HitCount;
+ const uint64_t ChunkMissCount = StoreStats.MissCount;
+ const uint64_t ChunkWriteCount = StoreStats.WriteCount;
+ const uint64_t TotalCount = HitCount + MissCount;
const uint64_t RpcRequests = m_CacheStats.RpcRequests;
const uint64_t RpcRecordRequests = m_CacheStats.RpcRecordRequests;
@@ -1906,11 +1835,17 @@ HttpStructuredCacheService::HandleStatsRequest(HttpServerRequest& Request)
const uint64_t RpcChunkRequests = m_CacheStats.RpcChunkRequests;
const uint64_t RpcChunkBatchRequests = m_CacheStats.RpcChunkBatchRequests;
+ const CidStoreSize CidSize = m_CidStore.TotalSize();
const CacheStoreSize CacheSize = m_CacheStore.TotalSize();
bool ShowCidStoreStats = Request.GetQueryParams().GetValue("cidstorestats") == "true";
bool ShowCacheStoreStats = Request.GetQueryParams().GetValue("cachestorestats") == "true";
+ CidStoreStats CidStoreStats = {};
+ if (ShowCidStoreStats)
+ {
+ CidStoreStats = m_CidStore.Stats();
+ }
ZenCacheStore::CacheStoreStats CacheStoreStats = {};
if (ShowCacheStoreStats)
{
@@ -1945,7 +1880,6 @@ HttpStructuredCacheService::HandleStatsRequest(HttpServerRequest& Request)
Cbo.EndObject();
Cbo << "hits" << HitCount << "misses" << MissCount << "writes" << WriteCount;
- const uint64_t TotalCount = HitCount + MissCount;
Cbo << "hit_ratio" << (TotalCount > 0 ? (double(HitCount) / double(TotalCount)) : 0.0);
if (m_UpstreamCache.IsActive())
@@ -1956,9 +1890,7 @@ HttpStructuredCacheService::HandleStatsRequest(HttpServerRequest& Request)
Cbo << "upstream_ratio" << (HitCount > 0 ? (double(UpstreamHitCount) / double(HitCount)) : 0.0);
}
- Cbo << "cidhits" << TotalChunkHitCount << "cidmisses" << TotalChunkMissCount << "cidwrites" << TotalChunkWriteCount;
- const uint64_t TotalChunkCount = TotalChunkHitCount + TotalChunkMissCount;
- Cbo << "cidhit_ratio" << (TotalChunkHitCount ? (double(TotalChunkCount) / double(TotalChunkHitCount)) : 0.0);
+ Cbo << "cidhits" << ChunkHitCount << "cidmisses" << ChunkMissCount << "cidwrites" << ChunkWriteCount;
if (ShowCacheStoreStats)
{
@@ -2067,58 +1999,20 @@ HttpStructuredCacheService::HandleStatsRequest(HttpServerRequest& Request)
{
Cbo.BeginObject("size");
{
- Cbo << "tiny" << TotalCidSize.TinySize;
- Cbo << "small" << TotalCidSize.SmallSize;
- Cbo << "large" << TotalCidSize.LargeSize;
- Cbo << "total" << TotalCidSize.TotalSize;
+ Cbo << "tiny" << CidSize.TinySize;
+ Cbo << "small" << CidSize.SmallSize;
+ Cbo << "large" << CidSize.LargeSize;
+ Cbo << "total" << CidSize.TotalSize;
}
Cbo.EndObject();
if (ShowCidStoreStats)
{
Cbo.BeginObject("store");
-
- auto OutputStats = [&](CidStore& ChunkStore) {
- CidStoreStats StoreStats = ChunkStore.Stats();
- Cbo << "hits" << StoreStats.HitCount << "misses" << StoreStats.MissCount << "writes" << StoreStats.WriteCount;
- const uint64_t Count = StoreStats.HitCount + StoreStats.MissCount;
- Cbo << "hit_ratio" << (Count ? (double(StoreStats.HitCount) / double(Count)) : 0.0);
- EmitSnapshot("read", StoreStats.FindChunkOps, Cbo);
- EmitSnapshot("write", StoreStats.AddChunkOps, Cbo);
- };
-
- if (UniqueStores.size() > 1)
- {
- Cbo.BeginArray("namespaces");
- for (auto It : UniqueStores)
- {
- CidStore* ChunkStore = It.first;
- const std::string& Namespace = It.second;
- CidStoreSize ChunkStoreSize = ChunkStore->TotalSize();
- Cbo.BeginObject();
- {
- Cbo << "namespace" << Namespace;
- Cbo.BeginObject("stats");
- OutputStats(*ChunkStore);
- Cbo.EndObject();
-
- Cbo.BeginObject("size");
- {
- Cbo << "tiny" << ChunkStoreSize.TinySize;
- Cbo << "small" << ChunkStoreSize.SmallSize;
- Cbo << "large" << ChunkStoreSize.LargeSize;
- Cbo << "total" << ChunkStoreSize.TotalSize;
- }
- Cbo.EndObject();
- }
- Cbo.EndObject();
- }
- Cbo.EndArray(); // namespaces
- }
- else if (UniqueStores.size() != 0)
- {
- OutputStats(*UniqueStores.begin()->first);
- }
+ Cbo << "hits" << CidStoreStats.HitCount << "misses" << CidStoreStats.MissCount << "writes" << CidStoreStats.WriteCount;
+ EmitSnapshot("read", CidStoreStats.FindChunkOps, Cbo);
+ EmitSnapshot("write", CidStoreStats.AddChunkOps, Cbo);
+ // EmitSnapshot("exists", CidStoreStats.ContainChunkOps, Cbo);
Cbo.EndObject();
}
}
diff --git a/src/zenserver/cache/httpstructuredcache.h b/src/zenserver/cache/httpstructuredcache.h
index d46ca145d..cb822f117 100644
--- a/src/zenserver/cache/httpstructuredcache.h
+++ b/src/zenserver/cache/httpstructuredcache.h
@@ -70,10 +70,8 @@ namespace cache {
class HttpStructuredCacheService : public HttpService, public IHttpStatsProvider, public IHttpStatusProvider
{
public:
- typedef std::function<CidStore&(std::string_view Context)> GetCidStoreFunc;
-
HttpStructuredCacheService(ZenCacheStore& InCacheStore,
- GetCidStoreFunc&& GetCidStore,
+ CidStore& InCidStore,
HttpStatsService& StatsService,
HttpStatusService& StatusService,
UpstreamCache& UpstreamCache,
@@ -117,6 +115,7 @@ private:
ZenCacheStore& m_CacheStore;
HttpStatsService& m_StatsService;
HttpStatusService& m_StatusService;
+ CidStore& m_CidStore;
UpstreamCache& m_UpstreamCache;
metrics::OperationTiming m_HttpRequests;
metrics::OperationTiming m_UpstreamGetRequestTiming;
diff --git a/src/zenserver/projectstore/httpprojectstore.cpp b/src/zenserver/projectstore/httpprojectstore.cpp
index 9600133f3..317a419eb 100644
--- a/src/zenserver/projectstore/httpprojectstore.cpp
+++ b/src/zenserver/projectstore/httpprojectstore.cpp
@@ -235,11 +235,13 @@ namespace {
//////////////////////////////////////////////////////////////////////////
-HttpProjectService::HttpProjectService(ProjectStore* Projects,
+HttpProjectService::HttpProjectService(CidStore& Store,
+ ProjectStore* Projects,
HttpStatusService& StatusService,
HttpStatsService& StatsService,
AuthMgr& AuthMgr)
: m_Log(logging::Get("project"))
+, m_CidStore(Store)
, m_ProjectStore(Projects)
, m_StatusService(StatusService)
, m_StatsService(StatsService)
@@ -405,45 +407,8 @@ HttpProjectService::HandleStatsRequest(HttpServerRequest& HttpReq)
{
ZEN_TRACE_CPU("ProjectService::Stats");
- bool ShowCidStoreStats = HttpReq.GetQueryParams().GetValue("cidstorestats") == "true";
-
- const GcStorageSize StoreSize = m_ProjectStore->StorageSize();
- uint64_t TotalChunkHitCount = 0;
- uint64_t TotalChunkMissCount = 0;
- uint64_t TotalChunkWriteCount = 0;
- CidStoreSize TotalCidSize;
-
- tsl::robin_map<CidStore*, std::string> UniqueStores;
- {
- m_ProjectStore->IterateProjects([&UniqueStores](ProjectStore::Project& Project) {
- CidStore* Store = &Project.GetCidStore();
- if (auto It = UniqueStores.find(Store); It == UniqueStores.end())
- {
- UniqueStores.insert_or_assign(Store, Project.Identifier);
- }
- else
- {
- UniqueStores.insert_or_assign(Store, std::string{});
- }
- });
-
- for (auto It : UniqueStores)
- {
- CidStore* ChunkStore = It.first;
-
- CidStoreStats ChunkStoreStats = ChunkStore->Stats();
- CidStoreSize ChunkStoreSize = ChunkStore->TotalSize();
-
- TotalChunkHitCount += ChunkStoreStats.HitCount;
- TotalChunkMissCount += ChunkStoreStats.MissCount;
- TotalChunkWriteCount += ChunkStoreStats.WriteCount;
-
- TotalCidSize.TinySize += ChunkStoreSize.TinySize;
- TotalCidSize.SmallSize += ChunkStoreSize.SmallSize;
- TotalCidSize.LargeSize += ChunkStoreSize.LargeSize;
- TotalCidSize.TotalSize += ChunkStoreSize.TotalSize;
- }
- }
+ const GcStorageSize StoreSize = m_ProjectStore->StorageSize();
+ const CidStoreSize CidSize = m_CidStore.TotalSize();
CbObjectWriter Cbo;
@@ -495,66 +460,12 @@ HttpProjectService::HandleStatsRequest(HttpServerRequest& HttpReq)
{
Cbo.BeginObject("size");
{
- Cbo << "tiny" << TotalCidSize.TinySize;
- Cbo << "small" << TotalCidSize.SmallSize;
- Cbo << "large" << TotalCidSize.LargeSize;
- Cbo << "total" << TotalCidSize.TotalSize;
+ Cbo << "tiny" << CidSize.TinySize;
+ Cbo << "small" << CidSize.SmallSize;
+ Cbo << "large" << CidSize.LargeSize;
+ Cbo << "total" << CidSize.TotalSize;
}
Cbo.EndObject();
-
- if (ShowCidStoreStats)
- {
- Cbo << "cidhits" << TotalChunkHitCount << "cidmisses" << TotalChunkMissCount << "cidwrites" << TotalChunkWriteCount;
- const uint64_t TotalChunkCount = TotalChunkHitCount + TotalChunkMissCount;
- Cbo << "cidhit_ratio" << (TotalChunkHitCount ? (double(TotalChunkCount) / double(TotalChunkHitCount)) : 0.0);
-
- Cbo.BeginObject("store");
-
- auto OutputStats = [&](CidStore& ChunkStore) {
- CidStoreStats StoreStats = ChunkStore.Stats();
- Cbo << "hits" << StoreStats.HitCount << "misses" << StoreStats.MissCount << "writes" << StoreStats.WriteCount;
- const uint64_t Count = StoreStats.HitCount + StoreStats.MissCount;
- Cbo << "hit_ratio" << (Count ? (double(StoreStats.HitCount) / double(Count)) : 0.0);
- EmitSnapshot("read", StoreStats.FindChunkOps, Cbo);
- EmitSnapshot("write", StoreStats.AddChunkOps, Cbo);
- };
-
- if (UniqueStores.size() > 1)
- {
- Cbo.BeginArray("projects");
- for (auto It : UniqueStores)
- {
- CidStore* ChunkStore = It.first;
- const std::string& ProjectId = It.second;
- CidStoreSize ChunkStoreSize = ChunkStore->TotalSize();
-
- Cbo.BeginObject();
- {
- Cbo << "project" << ProjectId;
- Cbo.BeginObject("stats");
- OutputStats(*ChunkStore);
- Cbo.EndObject();
-
- Cbo.BeginObject("size");
- {
- Cbo << "tiny" << ChunkStoreSize.TinySize;
- Cbo << "small" << ChunkStoreSize.SmallSize;
- Cbo << "large" << ChunkStoreSize.LargeSize;
- Cbo << "total" << ChunkStoreSize.TotalSize;
- }
- Cbo.EndObject();
- }
- Cbo.EndObject();
- }
- Cbo.EndArray(); // projects
- }
- else if (UniqueStores.size() != 0)
- {
- CidStore& ChunkStore = *UniqueStores.begin()->first;
- OutputStats(ChunkStore);
- }
- Cbo.EndObject();
- }
}
Cbo.EndObject();
@@ -1214,8 +1125,6 @@ HttpProjectService::HandleOplogOpNewRequest(HttpRouterRequest& Req)
}
Project->TouchOplog(OplogId);
- CidStore& ChunkStore = Project->GetCidStore();
-
ProjectStore::Oplog& Oplog = *FoundLog;
IoBuffer Payload = HttpReq.ReadPayload();
@@ -1228,7 +1137,7 @@ HttpProjectService::HandleOplogOpNewRequest(HttpRouterRequest& Req)
std::vector<IoHash> MissingChunks;
CbPackage::AttachmentResolver Resolver = [&](const IoHash& Hash) -> SharedBuffer {
- if (ChunkStore.ContainsChunk(Hash))
+ if (m_CidStore.ContainsChunk(Hash))
{
// Return null attachment as we already have it, no point in reading it and storing it again
return {};
@@ -1484,8 +1393,6 @@ HttpProjectService::HandleOpLogOpRequest(HttpRouterRequest& Req)
}
Project->TouchOplog(OplogId);
- CidStore& ChunkStore = Project->GetCidStore();
-
ProjectStore::Oplog& Oplog = *FoundLog;
if (const std::optional<int32_t> OpId = ParseInt<uint32_t>(OpIdString))
@@ -1500,7 +1407,7 @@ HttpProjectService::HandleOpLogOpRequest(HttpRouterRequest& Req)
Op.IterateAttachments([&](CbFieldView FieldView) {
const IoHash AttachmentHash = FieldView.AsAttachment();
- IoBuffer Payload = ChunkStore.FindChunkByCid(AttachmentHash);
+ IoBuffer Payload = m_CidStore.FindChunkByCid(AttachmentHash);
if (Payload)
{
switch (Payload.GetContentType())
@@ -2129,14 +2036,11 @@ HttpProjectService::HandleDetailsRequest(HttpRouterRequest& Req)
CSVHeader(Details, AttachmentDetails, CSVWriter);
m_ProjectStore->IterateProjects([&](ProjectStore::Project& Project) {
- CidStore& ChunkStore = Project.GetCidStore();
-
Project.IterateOplogs([&](const RwLock::SharedLockScope&, ProjectStore::Oplog& Oplog) {
- Oplog.IterateOplogWithKey([this, &Project, &Oplog, &ChunkStore, &CSVWriter, Details, AttachmentDetails](uint32_t LSN,
- const Oid& Key,
- CbObjectView Op) {
- CSVWriteOp(ChunkStore, Project.Identifier, Oplog.OplogId(), Details, AttachmentDetails, LSN, Key, Op, CSVWriter);
- });
+ Oplog.IterateOplogWithKey(
+ [this, &Project, &Oplog, &CSVWriter, Details, AttachmentDetails](uint32_t LSN, const Oid& Key, CbObjectView Op) {
+ CSVWriteOp(m_CidStore, Project.Identifier, Oplog.OplogId(), Details, AttachmentDetails, LSN, Key, Op, CSVWriter);
+ });
});
});
@@ -2150,9 +2054,8 @@ HttpProjectService::HandleDetailsRequest(HttpRouterRequest& Req)
m_ProjectStore->DiscoverProjects();
m_ProjectStore->IterateProjects([&](ProjectStore::Project& Project) {
- CidStore& ChunkStore = Project.GetCidStore();
- std::vector<std::string> OpLogs = Project.ScanForOplogs();
- CbWriteProject(ChunkStore, Project, OpLogs, Details, OpDetails, AttachmentDetails, Cbo);
+ std::vector<std::string> OpLogs = Project.ScanForOplogs();
+ CbWriteProject(m_CidStore, Project, OpLogs, Details, OpDetails, AttachmentDetails, Cbo);
});
}
Cbo.EndArray();
@@ -2181,8 +2084,7 @@ HttpProjectService::HandleProjectDetailsRequest(HttpRouterRequest& Req)
{
return HttpReq.WriteResponse(HttpResponseCode::NotFound);
}
- ProjectStore::Project& Project = *FoundProject.Get();
- CidStore& ChunkStore = Project.GetCidStore();
+ ProjectStore::Project& Project = *FoundProject.Get();
if (CSV)
{
@@ -2190,11 +2092,10 @@ HttpProjectService::HandleProjectDetailsRequest(HttpRouterRequest& Req)
CSVHeader(Details, AttachmentDetails, CSVWriter);
FoundProject->IterateOplogs([&](const RwLock::SharedLockScope&, ProjectStore::Oplog& Oplog) {
- Oplog.IterateOplogWithKey([this, &Project, &Oplog, &ChunkStore, &CSVWriter, Details, AttachmentDetails](uint32_t LSN,
- const Oid& Key,
- CbObjectView Op) {
- CSVWriteOp(ChunkStore, Project.Identifier, Oplog.OplogId(), Details, AttachmentDetails, LSN, Key, Op, CSVWriter);
- });
+ Oplog.IterateOplogWithKey(
+ [this, &Project, &Oplog, &CSVWriter, Details, AttachmentDetails](uint32_t LSN, const Oid& Key, CbObjectView Op) {
+ CSVWriteOp(m_CidStore, Project.Identifier, Oplog.OplogId(), Details, AttachmentDetails, LSN, Key, Op, CSVWriter);
+ });
});
HttpReq.WriteResponse(HttpResponseCode::OK, HttpContentType::kText, CSVWriter.ToView());
}
@@ -2204,7 +2105,7 @@ HttpProjectService::HandleProjectDetailsRequest(HttpRouterRequest& Req)
std::vector<std::string> OpLogs = FoundProject->ScanForOplogs();
Cbo.BeginArray("projects");
{
- CbWriteProject(ChunkStore, Project, OpLogs, Details, OpDetails, AttachmentDetails, Cbo);
+ CbWriteProject(m_CidStore, Project, OpLogs, Details, OpDetails, AttachmentDetails, Cbo);
}
Cbo.EndArray();
HttpReq.WriteResponse(HttpResponseCode::OK, Cbo.Save());
@@ -2240,17 +2141,16 @@ HttpProjectService::HandleOplogDetailsRequest(HttpRouterRequest& Req)
return HttpReq.WriteResponse(HttpResponseCode::NotFound);
}
- ProjectStore::Project& Project = *FoundProject.Get();
- CidStore& ChunkStore = Project.GetCidStore();
- ProjectStore::Oplog& Oplog = *FoundLog;
+ ProjectStore::Project& Project = *FoundProject.Get();
+ ProjectStore::Oplog& Oplog = *FoundLog;
if (CSV)
{
ExtendableStringBuilder<4096> CSVWriter;
CSVHeader(Details, AttachmentDetails, CSVWriter);
Oplog.IterateOplogWithKey(
- [this, &Project, &Oplog, &ChunkStore, &CSVWriter, Details, AttachmentDetails](uint32_t LSN, const Oid& Key, CbObjectView Op) {
- CSVWriteOp(ChunkStore, Project.Identifier, Oplog.OplogId(), Details, AttachmentDetails, LSN, Key, Op, CSVWriter);
+ [this, &Project, &Oplog, &CSVWriter, Details, AttachmentDetails](uint32_t LSN, const Oid& Key, CbObjectView Op) {
+ CSVWriteOp(m_CidStore, Project.Identifier, Oplog.OplogId(), Details, AttachmentDetails, LSN, Key, Op, CSVWriter);
});
HttpReq.WriteResponse(HttpResponseCode::OK, HttpContentType::kText, CSVWriter.ToView());
}
@@ -2259,7 +2159,7 @@ HttpProjectService::HandleOplogDetailsRequest(HttpRouterRequest& Req)
CbObjectWriter Cbo;
Cbo.BeginArray("oplogs");
{
- CbWriteOplog(ChunkStore, Oplog, Details, OpDetails, AttachmentDetails, Cbo);
+ CbWriteOplog(m_CidStore, Oplog, Details, OpDetails, AttachmentDetails, Cbo);
}
Cbo.EndArray();
HttpReq.WriteResponse(HttpResponseCode::OK, Cbo.Save());
@@ -2304,10 +2204,9 @@ HttpProjectService::HandleOplogOpDetailsRequest(HttpRouterRequest& Req)
fmt::format("Chunk info request for invalid chunk id '{}/{}'/'{}'", ProjectId, OplogId, OpId));
}
- const Oid ObjId = Oid::FromHexString(OpId);
- ProjectStore::Project& Project = *FoundProject.Get();
- CidStore& ChunkStore = Project.GetCidStore();
- ProjectStore::Oplog& Oplog = *FoundLog;
+ const Oid ObjId = Oid::FromHexString(OpId);
+ ProjectStore::Project& Project = *FoundProject.Get();
+ ProjectStore::Oplog& Oplog = *FoundLog;
std::optional<CbObject> Op = Oplog.GetOpByKey(ObjId);
if (!Op.has_value())
@@ -2325,7 +2224,7 @@ HttpProjectService::HandleOplogOpDetailsRequest(HttpRouterRequest& Req)
ExtendableStringBuilder<4096> CSVWriter;
CSVHeader(Details, AttachmentDetails, CSVWriter);
- CSVWriteOp(ChunkStore, Project.Identifier, Oplog.OplogId(), Details, AttachmentDetails, LSN.value(), ObjId, Op.value(), CSVWriter);
+ CSVWriteOp(m_CidStore, Project.Identifier, Oplog.OplogId(), Details, AttachmentDetails, LSN.value(), ObjId, Op.value(), CSVWriter);
HttpReq.WriteResponse(HttpResponseCode::OK, HttpContentType::kText, CSVWriter.ToView());
}
else
@@ -2333,7 +2232,7 @@ HttpProjectService::HandleOplogOpDetailsRequest(HttpRouterRequest& Req)
CbObjectWriter Cbo;
Cbo.BeginArray("ops");
{
- CbWriteOp(ChunkStore, Details, OpDetails, AttachmentDetails, LSN.value(), ObjId, Op.value(), Cbo);
+ CbWriteOp(m_CidStore, Details, OpDetails, AttachmentDetails, LSN.value(), ObjId, Op.value(), Cbo);
}
Cbo.EndArray();
HttpReq.WriteResponse(HttpResponseCode::OK, Cbo.Save());
diff --git a/src/zenserver/projectstore/httpprojectstore.h b/src/zenserver/projectstore/httpprojectstore.h
index 5782188e6..295defa5c 100644
--- a/src/zenserver/projectstore/httpprojectstore.h
+++ b/src/zenserver/projectstore/httpprojectstore.h
@@ -35,7 +35,11 @@ class ProjectStore;
class HttpProjectService : public HttpService, public IHttpStatusProvider, public IHttpStatsProvider
{
public:
- HttpProjectService(ProjectStore* InProjectStore, HttpStatusService& StatusService, HttpStatsService& StatsService, AuthMgr& AuthMgr);
+ HttpProjectService(CidStore& Store,
+ ProjectStore* InProjectStore,
+ HttpStatusService& StatusService,
+ HttpStatsService& StatsService,
+ AuthMgr& AuthMgr);
~HttpProjectService();
virtual const char* BaseUri() const override;
@@ -88,6 +92,7 @@ private:
inline LoggerRef Log() { return m_Log; }
LoggerRef m_Log;
+ CidStore& m_CidStore;
HttpRequestRouter m_Router;
Ref<ProjectStore> m_ProjectStore;
HttpStatusService& m_StatusService;
diff --git a/src/zenserver/projectstore/projectstore.cpp b/src/zenserver/projectstore/projectstore.cpp
index 7463e9caa..a5ab24cfb 100644
--- a/src/zenserver/projectstore/projectstore.cpp
+++ b/src/zenserver/projectstore/projectstore.cpp
@@ -3832,7 +3832,7 @@ ProjectStore::Project::LastOplogAccessTime(std::string_view Oplog) const
//////////////////////////////////////////////////////////////////////////
-ProjectStore::ProjectStore(GetCidStoreFunc&& GetCidStore,
+ProjectStore::ProjectStore(CidStore& Store,
std::filesystem::path BasePath,
GcManager& Gc,
JobQueue& JobQueue,
@@ -3840,7 +3840,7 @@ ProjectStore::ProjectStore(GetCidStoreFunc&& GetCidStore,
const Configuration& Config)
: m_Log(logging::Get("project"))
, m_Gc(Gc)
-, m_GetCidStore(std::move(GetCidStore))
+, m_CidStore(Store)
, m_JobQueue(JobQueue)
, m_OpenProcessCache(InOpenProcessCache)
, m_ProjectBasePath(BasePath)
@@ -4025,8 +4025,6 @@ ProjectStore::OpenProject(std::string_view ProjectId)
}
}
- CidStore& ChunkStore = m_GetCidStore(ProjectId);
-
RwLock::ExclusiveLockScope _(m_ProjectsLock);
if (auto ProjIt = m_Projects.find(std::string{ProjectId}); ProjIt != m_Projects.end())
{
@@ -4043,7 +4041,7 @@ ProjectStore::OpenProject(std::string_view ProjectId)
Ref<Project>& Prj =
m_Projects
- .try_emplace(std::string{ProjectId}, Ref<ProjectStore::Project>(new ProjectStore::Project(this, ChunkStore, BasePath)))
+ .try_emplace(std::string{ProjectId}, Ref<ProjectStore::Project>(new ProjectStore::Project(this, m_CidStore, BasePath)))
.first->second;
Prj->Identifier = ProjectId;
Prj->Read();
@@ -4070,14 +4068,12 @@ ProjectStore::NewProject(const std::filesystem::path& BasePath,
ZEN_MEMSCOPE(GetProjectstoreTag());
ZEN_TRACE_CPU("Store::NewProject");
- CidStore& ChunkStore = m_GetCidStore(ProjectId);
-
RwLock::ExclusiveLockScope _(m_ProjectsLock);
ZEN_INFO("project '{}': creating project at '{}'", ProjectId, BasePath);
Ref<Project>& Prj =
- m_Projects.try_emplace(std::string{ProjectId}, Ref<ProjectStore::Project>(new ProjectStore::Project(this, ChunkStore, BasePath)))
+ m_Projects.try_emplace(std::string{ProjectId}, Ref<ProjectStore::Project>(new ProjectStore::Project(this, m_CidStore, BasePath)))
.first->second;
Prj->Identifier = ProjectId;
Prj->RootDir = RootDir;
@@ -4806,7 +4802,7 @@ ProjectStore::GetChunk(const std::string_view ProjectId,
}
const IoHash Hash = IoHash::FromHexString(Cid);
- OutChunk = Project->GetCidStore().FindChunkByCid(Hash);
+ OutChunk = m_CidStore.FindChunkByCid(Hash);
if (!OutChunk)
{
@@ -4869,7 +4865,7 @@ ProjectStore::PutChunk(const std::string_view ProjectId,
}
FoundLog->CaptureAddedAttachments(std::vector<IoHash>{Hash});
- CidStore::InsertResult Result = Project->GetCidStore().AddChunk(Chunk, Hash);
+ CidStore::InsertResult Result = m_CidStore.AddChunk(Chunk, Hash);
return {Result.New ? HttpResponseCode::Created : HttpResponseCode::OK, {}};
}
@@ -4898,19 +4894,18 @@ ProjectStore::GetChunks(const std::string_view ProjectId,
}
Project->TouchOplog(OplogId);
- CidStore& ChunkStore = Project->GetCidStore();
-
if (RequestObject["chunks"sv].IsArray())
{
// Legacy full chunks only by rawhash
- CbArrayView ChunksArray = RequestObject["chunks"sv].AsArrayView();
+ CbArrayView ChunksArray = RequestObject["chunks"sv].AsArrayView();
+
CbObjectWriter ResponseWriter;
ResponseWriter.BeginArray("chunks"sv);
for (CbFieldView FieldView : ChunksArray)
{
IoHash RawHash = FieldView.AsHash();
- IoBuffer ChunkBuffer = ChunkStore.FindChunkByCid(RawHash);
+ IoBuffer ChunkBuffer = m_CidStore.FindChunkByCid(RawHash);
if (ChunkBuffer)
{
CompressedBuffer Compressed = CompressedBuffer::FromCompressedNoValidate(std::move(ChunkBuffer));
@@ -5062,7 +5057,7 @@ ProjectStore::GetChunks(const std::string_view ProjectId,
if (ChunkRequest.Input.Id.index() == 0)
{
const IoHash& ChunkHash = std::get<IoHash>(ChunkRequest.Input.Id);
- IoBuffer Payload = ChunkStore.FindChunkByCid(ChunkHash);
+ IoBuffer Payload = m_CidStore.FindChunkByCid(ChunkHash);
if (Payload)
{
ChunkRequest.Output.Exists = true;
@@ -5249,7 +5244,7 @@ ProjectStore::WriteOplog(const std::string_view ProjectId, const std::string_vie
return {HttpResponseCode::BadRequest, "Invalid payload format"};
}
- CidStore& ChunkStore = Project->GetCidStore();
+ CidStore& ChunkStore = m_CidStore;
RwLock AttachmentsLock;
tsl::robin_set<IoHash, IoHash::Hasher> Attachments;
@@ -5355,7 +5350,7 @@ ProjectStore::ReadOplog(const std::string_view ProjectId,
}
}
- CidStore& ChunkStore = Project->GetCidStore();
+ CidStore& ChunkStore = m_CidStore;
RemoteProjectStore::LoadContainerResult ContainerResult = BuildContainer(
ChunkStore,
@@ -5467,8 +5462,6 @@ ProjectStore::Rpc(HttpServerRequest& HttpReq,
}
Project->TouchOplog(OplogId);
- CidStore& ChunkStore = Project->GetCidStore();
-
if (Method == "import"sv)
{
if (!AreDiskWritesAllowed())
@@ -5550,7 +5543,7 @@ ProjectStore::Rpc(HttpServerRequest& HttpReq,
}
Oplog->CaptureAddedAttachments(WriteRawHashes);
- ChunkStore.AddChunks(WriteAttachmentBuffers, WriteRawHashes, CidStore::InsertMode::kCopyOnly);
+ m_CidStore.AddChunks(WriteAttachmentBuffers, WriteRawHashes, CidStore::InsertMode::kCopyOnly);
}
HttpReq.WriteResponse(HttpResponseCode::OK);
return true;
@@ -5723,14 +5716,14 @@ ProjectStore::Rpc(HttpServerRequest& HttpReq,
ResponseObj.EndArray();
}
- // Ops that have moved chunks to a compressed buffer for storage in ChunkStore have been rewritten with references to the new
- // chunk(s). Make sure we add the chunks to ChunkStore, and do it after we update the oplog so GC doesn't think we have
+ // Ops that have moved chunks to a compressed buffer for storage in m_CidStore have been rewritten with references to the new
+ // chunk(s). Make sure we add the chunks to m_CidStore, and do it after we update the oplog so GC doesn't think we have
// unreferenced chunks.
for (auto It : AddedChunks)
{
const IoHash& RawHash = It.first;
AddedChunk& Chunk = It.second;
- CidStore::InsertResult Result = ChunkStore.AddChunk(Chunk.Buffer, RawHash);
+ CidStore::InsertResult Result = m_CidStore.AddChunk(Chunk.Buffer, RawHash);
if (Result.New)
{
InlinedBytes += Chunk.RawSize;
@@ -5793,7 +5786,7 @@ ProjectStore::Export(Ref<ProjectStore::Project> Project, ProjectStore::Oplog& Op
NiceBytes(MaxBlockSize),
NiceBytes(MaxChunkEmbedSize)));
- RemoteProjectStore::Result Result = SaveOplog(Project->GetCidStore(),
+ RemoteProjectStore::Result Result = SaveOplog(m_CidStore,
*ActualRemoteStore,
*Project.Get(),
*OplogPtr,
@@ -5838,12 +5831,11 @@ ProjectStore::Import(ProjectStore::Project& Project, ProjectStore::Oplog& Oplog,
}
std::shared_ptr<RemoteProjectStore> RemoteStore = std::move(RemoteStoreResult.Store);
RemoteProjectStore::RemoteStoreInfo StoreInfo = RemoteStore->GetInfo();
- CidStore& ChunkStore = Project.GetCidStore();
JobId JobId = m_JobQueue.QueueJob(
fmt::format("Import oplog '{}/{}'", Project.Identifier, Oplog.OplogId()),
[this,
- ChunkStore = &ChunkStore,
+ ChunkStore = &m_CidStore,
ActualRemoteStore = std::move(RemoteStore),
OplogPtr = &Oplog,
Force,
@@ -5855,7 +5847,7 @@ ProjectStore::Import(ProjectStore::Project& Project, ProjectStore::Oplog& Oplog,
ActualRemoteStore->GetInfo().Description));
RemoteProjectStore::Result Result =
- LoadOplog(*ChunkStore, *ActualRemoteStore, *OplogPtr, Force, IgnoreMissingAttachments, CleanOplog, &Context);
+ LoadOplog(m_CidStore, *ActualRemoteStore, *OplogPtr, Force, IgnoreMissingAttachments, CleanOplog, &Context);
auto Response = ConvertResult(Result);
ZEN_INFO("LoadOplog: Status: {} '{}'", ToString(Response.first), Response.second);
if (!IsHttpSuccessCode(Response.first))
@@ -6894,11 +6886,6 @@ namespace testutils {
return BuildChunksRequest<IoHash>(SkipData, "RawHash", Chunks, Ranges, ModTags);
}
- ProjectStore::GetCidStoreFunc SingleChunkStore(CidStore& ChunkStore)
- {
- return [ChunkStore = &ChunkStore](std::string_view) -> CidStore& { return *ChunkStore; };
- }
-
} // namespace testutils
TEST_CASE("project.opkeys")
@@ -6961,18 +6948,13 @@ TEST_CASE("project.store.create")
auto JobQueue = MakeJobQueue(1, ""sv);
OpenProcessCache ProcessCache;
GcManager Gc;
- CidStore ChunkStore(Gc);
+ CidStore CidStore(Gc);
CidStoreConfiguration CidConfig = {.RootDirectory = TempDir.Path() / "cas", .TinyValueThreshold = 1024, .HugeValueThreshold = 4096};
- ChunkStore.Initialize(CidConfig);
+ CidStore.Initialize(CidConfig);
std::string_view ProjectName("proj1"sv);
std::filesystem::path BasePath = TempDir.Path() / "projectstore";
- ProjectStore ProjectStore(testutils::SingleChunkStore(ChunkStore),
- BasePath,
- Gc,
- *JobQueue,
- ProcessCache,
- ProjectStore::Configuration{});
+ ProjectStore ProjectStore(CidStore, BasePath, Gc, *JobQueue, ProcessCache, ProjectStore::Configuration{});
std::filesystem::path RootDir = TempDir.Path() / "root";
std::filesystem::path EngineRootDir = TempDir.Path() / "engine";
std::filesystem::path ProjectRootDir = TempDir.Path() / "game";
@@ -6997,17 +6979,12 @@ TEST_CASE("project.store.lifetimes")
auto JobQueue = MakeJobQueue(1, ""sv);
OpenProcessCache ProcessCache;
GcManager Gc;
- CidStore ChunkStore(Gc);
+ CidStore CidStore(Gc);
CidStoreConfiguration CidConfig = {.RootDirectory = TempDir.Path() / "cas", .TinyValueThreshold = 1024, .HugeValueThreshold = 4096};
- ChunkStore.Initialize(CidConfig);
+ CidStore.Initialize(CidConfig);
std::filesystem::path BasePath = TempDir.Path() / "projectstore";
- ProjectStore ProjectStore(testutils::SingleChunkStore(ChunkStore),
- BasePath,
- Gc,
- *JobQueue,
- ProcessCache,
- ProjectStore::Configuration{});
+ ProjectStore ProjectStore(CidStore, BasePath, Gc, *JobQueue, ProcessCache, ProjectStore::Configuration{});
std::filesystem::path RootDir = TempDir.Path() / "root";
std::filesystem::path EngineRootDir = TempDir.Path() / "engine";
std::filesystem::path ProjectRootDir = TempDir.Path() / "game";
@@ -7065,17 +7042,12 @@ TEST_CASE_TEMPLATE("project.store.export",
auto JobQueue = MakeJobQueue(1, ""sv);
OpenProcessCache ProcessCache;
GcManager Gc;
- CidStore ChunkStore(Gc);
+ CidStore CidStore(Gc);
CidStoreConfiguration CidConfig = {.RootDirectory = TempDir.Path() / "cas", .TinyValueThreshold = 1024, .HugeValueThreshold = 4096};
- ChunkStore.Initialize(CidConfig);
+ CidStore.Initialize(CidConfig);
std::filesystem::path BasePath = TempDir.Path() / "projectstore";
- ProjectStore ProjectStore(testutils::SingleChunkStore(ChunkStore),
- BasePath,
- Gc,
- *JobQueue,
- ProcessCache,
- ProjectStore::Configuration{});
+ ProjectStore ProjectStore(CidStore, BasePath, Gc, *JobQueue, ProcessCache, ProjectStore::Configuration{});
std::filesystem::path RootDir = TempDir.Path() / "root";
std::filesystem::path EngineRootDir = TempDir.Path() / "engine";
std::filesystem::path ProjectRootDir = TempDir.Path() / "game";
@@ -7109,7 +7081,7 @@ TEST_CASE_TEMPLATE("project.store.export",
std::shared_ptr<RemoteProjectStore> RemoteStore = CreateFileRemoteStore(Options);
RemoteProjectStore::RemoteStoreInfo StoreInfo = RemoteStore->GetInfo();
- RemoteProjectStore::Result ExportResult = SaveOplog(ChunkStore,
+ RemoteProjectStore::Result ExportResult = SaveOplog(CidStore,
*RemoteStore,
*Project.Get(),
*Oplog,
@@ -7126,7 +7098,7 @@ TEST_CASE_TEMPLATE("project.store.export",
ProjectStore::Oplog* OplogImport = Project->NewOplog("oplog2", {});
CHECK(OplogImport != nullptr);
- RemoteProjectStore::Result ImportResult = LoadOplog(ChunkStore,
+ RemoteProjectStore::Result ImportResult = LoadOplog(CidStore,
*RemoteStore,
*OplogImport,
/*Force*/ false,
@@ -7135,7 +7107,7 @@ TEST_CASE_TEMPLATE("project.store.export",
nullptr);
CHECK(ImportResult.ErrorCode == 0);
- RemoteProjectStore::Result ImportForceResult = LoadOplog(ChunkStore,
+ RemoteProjectStore::Result ImportForceResult = LoadOplog(CidStore,
*RemoteStore,
*OplogImport,
/*Force*/ true,
@@ -7144,7 +7116,7 @@ TEST_CASE_TEMPLATE("project.store.export",
nullptr);
CHECK(ImportForceResult.ErrorCode == 0);
- RemoteProjectStore::Result ImportCleanResult = LoadOplog(ChunkStore,
+ RemoteProjectStore::Result ImportCleanResult = LoadOplog(CidStore,
*RemoteStore,
*OplogImport,
/*Force*/ false,
@@ -7153,7 +7125,7 @@ TEST_CASE_TEMPLATE("project.store.export",
nullptr);
CHECK(ImportCleanResult.ErrorCode == 0);
- RemoteProjectStore::Result ImportForceCleanResult = LoadOplog(ChunkStore,
+ RemoteProjectStore::Result ImportForceCleanResult = LoadOplog(CidStore,
*RemoteStore,
*OplogImport,
/*Force*/ true,
@@ -7173,17 +7145,12 @@ TEST_CASE("project.store.gc")
auto JobQueue = MakeJobQueue(1, ""sv);
OpenProcessCache ProcessCache;
GcManager Gc;
- CidStore ChunkStore(Gc);
+ CidStore CidStore(Gc);
CidStoreConfiguration CidConfig = {.RootDirectory = TempDir.Path() / "cas", .TinyValueThreshold = 1024, .HugeValueThreshold = 4096};
- ChunkStore.Initialize(CidConfig);
+ CidStore.Initialize(CidConfig);
std::filesystem::path BasePath = TempDir.Path() / "projectstore";
- ProjectStore ProjectStore(testutils::SingleChunkStore(ChunkStore),
- BasePath,
- Gc,
- *JobQueue,
- ProcessCache,
- ProjectStore::Configuration{});
+ ProjectStore ProjectStore(CidStore, BasePath, Gc, *JobQueue, ProcessCache, ProjectStore::Configuration{});
std::filesystem::path RootDir = TempDir.Path() / "root";
std::filesystem::path EngineRootDir = TempDir.Path() / "engine";
@@ -7379,17 +7346,12 @@ TEST_CASE("project.store.gc.prep")
auto JobQueue = MakeJobQueue(1, ""sv);
OpenProcessCache ProcessCache;
GcManager Gc;
- CidStore ChunkStore(Gc);
+ CidStore CidStore(Gc);
CidStoreConfiguration CidConfig = {.RootDirectory = TempDir.Path() / "cas", .TinyValueThreshold = 1024, .HugeValueThreshold = 4096};
- ChunkStore.Initialize(CidConfig);
+ CidStore.Initialize(CidConfig);
std::filesystem::path BasePath = TempDir.Path() / "projectstore";
- ProjectStore ProjectStore(testutils::SingleChunkStore(ChunkStore),
- BasePath,
- Gc,
- *JobQueue,
- ProcessCache,
- ProjectStore::Configuration{});
+ ProjectStore ProjectStore(CidStore, BasePath, Gc, *JobQueue, ProcessCache, ProjectStore::Configuration{});
std::filesystem::path RootDir = TempDir.Path() / "root";
std::filesystem::path EngineRootDir = TempDir.Path() / "engine";
@@ -7432,7 +7394,7 @@ TEST_CASE("project.store.gc.prep")
// Equivalent of a `prep` existance check call
for (auto Attachment : OpAttachments)
{
- CHECK(ChunkStore.ContainsChunk(Attachment.second.DecodeRawHash()));
+ CHECK(CidStore.ContainsChunk(Attachment.second.DecodeRawHash()));
}
{
@@ -7446,7 +7408,7 @@ TEST_CASE("project.store.gc.prep")
// If a gc comes in between our prep and op write the chunks will be removed
for (auto Attachment : OpAttachments)
{
- CHECK(!ChunkStore.ContainsChunk(Attachment.second.DecodeRawHash()));
+ CHECK(!CidStore.ContainsChunk(Attachment.second.DecodeRawHash()));
}
{
@@ -7466,7 +7428,7 @@ TEST_CASE("project.store.gc.prep")
for (auto Attachment : OpAttachments)
{
- CHECK(ChunkStore.ContainsChunk(Attachment.second.DecodeRawHash()));
+ CHECK(CidStore.ContainsChunk(Attachment.second.DecodeRawHash()));
}
{
@@ -7480,7 +7442,7 @@ TEST_CASE("project.store.gc.prep")
// Attachments should now be retained
for (auto Attachment : OpAttachments)
{
- CHECK(ChunkStore.ContainsChunk(Attachment.second.DecodeRawHash()));
+ CHECK(CidStore.ContainsChunk(Attachment.second.DecodeRawHash()));
}
{
@@ -7494,7 +7456,7 @@ TEST_CASE("project.store.gc.prep")
// Attachments should now be retained across multiple GCs if retain time is still valud
for (auto Attachment : OpAttachments)
{
- CHECK(ChunkStore.ContainsChunk(Attachment.second.DecodeRawHash()));
+ CHECK(CidStore.ContainsChunk(Attachment.second.DecodeRawHash()));
}
{
@@ -7506,7 +7468,7 @@ TEST_CASE("project.store.gc.prep")
}
for (auto Attachment : OpAttachments)
{
- CHECK(ChunkStore.ContainsChunk(Attachment.second.DecodeRawHash()));
+ CHECK(CidStore.ContainsChunk(Attachment.second.DecodeRawHash()));
}
{
Ref<ProjectStore::Project> Project1 = ProjectStore.OpenProject("proj1"sv);
@@ -7523,7 +7485,7 @@ TEST_CASE("project.store.gc.prep")
for (auto Attachment : OpAttachments)
{
- CHECK(!ChunkStore.ContainsChunk(Attachment.second.DecodeRawHash()));
+ CHECK(!CidStore.ContainsChunk(Attachment.second.DecodeRawHash()));
}
{
@@ -7556,7 +7518,7 @@ TEST_CASE("project.store.gc.prep")
}
for (auto Attachment : OpAttachments)
{
- CHECK(ChunkStore.ContainsChunk(Attachment.second.DecodeRawHash()));
+ CHECK(CidStore.ContainsChunk(Attachment.second.DecodeRawHash()));
}
Sleep(200);
@@ -7571,7 +7533,7 @@ TEST_CASE("project.store.gc.prep")
}
for (auto Attachment : OpAttachments)
{
- CHECK(ChunkStore.ContainsChunk(Attachment.second.DecodeRawHash()));
+ CHECK(CidStore.ContainsChunk(Attachment.second.DecodeRawHash()));
}
// This pass the retention time has expired and the last GC pass cleared the entries
@@ -7585,7 +7547,7 @@ TEST_CASE("project.store.gc.prep")
for (auto Attachment : OpAttachments)
{
- CHECK(!ChunkStore.ContainsChunk(Attachment.second.DecodeRawHash()));
+ CHECK(!CidStore.ContainsChunk(Attachment.second.DecodeRawHash()));
}
}
@@ -7599,17 +7561,12 @@ TEST_CASE("project.store.rpc.getchunks")
auto JobQueue = MakeJobQueue(1, ""sv);
OpenProcessCache ProcessCache;
GcManager Gc;
- CidStore ChunkStore(Gc);
+ CidStore CidStore(Gc);
CidStoreConfiguration CidConfig = {.RootDirectory = TempDir.Path() / "cas"sv, .TinyValueThreshold = 1024, .HugeValueThreshold = 4096};
- ChunkStore.Initialize(CidConfig);
+ CidStore.Initialize(CidConfig);
std::filesystem::path BasePath = TempDir.Path() / "projectstore"sv;
- ProjectStore ProjectStore(testutils::SingleChunkStore(ChunkStore),
- BasePath,
- Gc,
- *JobQueue,
- ProcessCache,
- ProjectStore::Configuration{});
+ ProjectStore ProjectStore(CidStore, BasePath, Gc, *JobQueue, ProcessCache, ProjectStore::Configuration{});
std::filesystem::path RootDir = TempDir.Path() / "root"sv;
std::filesystem::path EngineRootDir = TempDir.Path() / "engine"sv;
@@ -8526,17 +8483,12 @@ TEST_CASE("project.store.partial.read")
auto JobQueue = MakeJobQueue(1, ""sv);
OpenProcessCache ProcessCache;
GcManager Gc;
- CidStore ChunkStore(Gc);
+ CidStore CidStore(Gc);
CidStoreConfiguration CidConfig = {.RootDirectory = TempDir.Path() / "cas"sv, .TinyValueThreshold = 1024, .HugeValueThreshold = 4096};
- ChunkStore.Initialize(CidConfig);
+ CidStore.Initialize(CidConfig);
std::filesystem::path BasePath = TempDir.Path() / "projectstore"sv;
- ProjectStore ProjectStore(testutils::SingleChunkStore(ChunkStore),
- BasePath,
- Gc,
- *JobQueue,
- ProcessCache,
- ProjectStore::Configuration{});
+ ProjectStore ProjectStore(CidStore, BasePath, Gc, *JobQueue, ProcessCache, ProjectStore::Configuration{});
std::filesystem::path RootDir = TempDir.Path() / "root"sv;
std::filesystem::path EngineRootDir = TempDir.Path() / "engine"sv;
@@ -8709,17 +8661,12 @@ TEST_CASE("project.store.iterateoplog")
auto JobQueue = MakeJobQueue(1, ""sv);
OpenProcessCache ProcessCache;
GcManager Gc;
- CidStore ChunkStore(Gc);
+ CidStore CidStore(Gc);
CidStoreConfiguration CidConfig = {.RootDirectory = TempDir.Path() / "cas"sv, .TinyValueThreshold = 1024, .HugeValueThreshold = 4096};
- ChunkStore.Initialize(CidConfig);
+ CidStore.Initialize(CidConfig);
std::filesystem::path BasePath = TempDir.Path() / "projectstore"sv;
- ProjectStore ProjectStore(testutils::SingleChunkStore(ChunkStore),
- BasePath,
- Gc,
- *JobQueue,
- ProcessCache,
- ProjectStore::Configuration{});
+ ProjectStore ProjectStore(CidStore, BasePath, Gc, *JobQueue, ProcessCache, ProjectStore::Configuration{});
std::filesystem::path RootDir = TempDir.Path() / "root"sv;
std::filesystem::path EngineRootDir = TempDir.Path() / "enginesv";
diff --git a/src/zenserver/projectstore/projectstore.h b/src/zenserver/projectstore/projectstore.h
index eb27665f9..2595d7198 100644
--- a/src/zenserver/projectstore/projectstore.h
+++ b/src/zenserver/projectstore/projectstore.h
@@ -68,9 +68,7 @@ public:
{
};
- typedef std::function<CidStore&(std::string_view Context)> GetCidStoreFunc;
-
- ProjectStore(GetCidStoreFunc&& GetCidStore,
+ ProjectStore(CidStore& Store,
std::filesystem::path BasePath,
GcManager& Gc,
JobQueue& JobQueue,
@@ -328,8 +326,6 @@ public:
Project(ProjectStore* PrjStore, CidStore& Store, std::filesystem::path BasePath);
virtual ~Project();
- CidStore& GetCidStore() { return m_CidStore; };
-
void Read();
void Write();
[[nodiscard]] static bool Exists(const std::filesystem::path& BasePath);
@@ -503,7 +499,7 @@ public:
private:
LoggerRef m_Log;
GcManager& m_Gc;
- GetCidStoreFunc m_GetCidStore;
+ CidStore& m_CidStore;
JobQueue& m_JobQueue;
OpenProcessCache& m_OpenProcessCache;
std::filesystem::path m_ProjectBasePath;
diff --git a/src/zenserver/upstream/upstreamcache.cpp b/src/zenserver/upstream/upstreamcache.cpp
index a1c460bc0..744b861dd 100644
--- a/src/zenserver/upstream/upstreamcache.cpp
+++ b/src/zenserver/upstream/upstreamcache.cpp
@@ -1475,17 +1475,12 @@ namespace detail {
class UpstreamCacheImpl final : public UpstreamCache
{
- struct EnqueuedRequest
- {
- UpstreamCacheRecord Record;
- std::function<IoBuffer(const IoHash& ChunkHash)> GetValueFunc;
- };
-
public:
- UpstreamCacheImpl(const UpstreamCacheOptions& Options, ZenCacheStore& CacheStore)
+ UpstreamCacheImpl(const UpstreamCacheOptions& Options, ZenCacheStore& CacheStore, CidStore& CidStore)
: m_Log(logging::Get("upstream"))
, m_Options(Options)
, m_CacheStore(CacheStore)
+ , m_CidStore(CidStore)
{
}
@@ -1841,17 +1836,17 @@ public:
}
}
- virtual void EnqueueUpstream(UpstreamCacheRecord CacheRecord, std::function<IoBuffer(const IoHash&)>&& GetValueFunc) override
+ virtual void EnqueueUpstream(UpstreamCacheRecord CacheRecord) override
{
if (m_RunState.IsRunning && m_Options.WriteUpstream && m_Endpoints.size() > 0)
{
if (!m_UpstreamThreads.empty())
{
- m_UpstreamQueue.Enqueue(EnqueuedRequest{.Record = std::move(CacheRecord), .GetValueFunc = GetValueFunc});
+ m_UpstreamQueue.Enqueue(std::move(CacheRecord));
}
else
{
- ProcessCacheRecord(std::move(CacheRecord), std::move(GetValueFunc));
+ ProcessCacheRecord(std::move(CacheRecord));
}
}
}
@@ -1905,7 +1900,7 @@ public:
}
private:
- void ProcessCacheRecord(const UpstreamCacheRecord& CacheRecord, std::function<IoBuffer(const IoHash& ChunkHash)>&& GetValueFunc)
+ void ProcessCacheRecord(UpstreamCacheRecord CacheRecord)
{
ZEN_TRACE_CPU("Upstream::ProcessCacheRecord");
@@ -1923,7 +1918,7 @@ private:
for (const IoHash& ValueContentId : CacheRecord.ValueContentIds)
{
- if (IoBuffer Payload = GetValueFunc(ValueContentId))
+ if (IoBuffer Payload = m_CidStore.FindChunkByCid(ValueContentId))
{
Payloads.push_back(Payload);
}
@@ -1975,19 +1970,19 @@ private:
for (;;)
{
- EnqueuedRequest Request;
- if (m_UpstreamQueue.WaitAndDequeue(Request))
+ UpstreamCacheRecord CacheRecord;
+ if (m_UpstreamQueue.WaitAndDequeue(CacheRecord))
{
try
{
- ProcessCacheRecord(Request.Record, std::move(Request.GetValueFunc));
+ ProcessCacheRecord(std::move(CacheRecord));
}
catch (const std::exception& Err)
{
ZEN_ERROR("upload cache record '{}/{}/{}' FAILED, reason '{}'",
- Request.Record.Namespace,
- Request.Record.Key.Bucket,
- Request.Record.Key.Hash,
+ CacheRecord.Namespace,
+ CacheRecord.Key.Bucket,
+ CacheRecord.Key.Hash,
Err.what());
}
}
@@ -2081,7 +2076,7 @@ private:
LoggerRef Log() { return m_Log; }
- using UpstreamQueue = BlockingQueue<EnqueuedRequest>;
+ using UpstreamQueue = BlockingQueue<UpstreamCacheRecord>;
struct RunState
{
@@ -2107,6 +2102,7 @@ private:
LoggerRef m_Log;
UpstreamCacheOptions m_Options;
ZenCacheStore& m_CacheStore;
+ CidStore& m_CidStore;
UpstreamQueue m_UpstreamQueue;
std::shared_mutex m_EndpointsMutex;
std::vector<std::unique_ptr<UpstreamEndpoint>> m_Endpoints;
@@ -2130,9 +2126,9 @@ UpstreamEndpoint::CreateJupiterEndpoint(const JupiterClientOptions& Options, con
}
std::unique_ptr<UpstreamCache>
-CreateUpstreamCache(const UpstreamCacheOptions& Options, ZenCacheStore& CacheStore)
+CreateUpstreamCache(const UpstreamCacheOptions& Options, ZenCacheStore& CacheStore, CidStore& CidStore)
{
- return std::make_unique<UpstreamCacheImpl>(Options, CacheStore);
+ return std::make_unique<UpstreamCacheImpl>(Options, CacheStore, CidStore);
}
} // namespace zen
diff --git a/src/zenserver/upstream/upstreamcache.h b/src/zenserver/upstream/upstreamcache.h
index e4b9a73ad..26e5decac 100644
--- a/src/zenserver/upstream/upstreamcache.h
+++ b/src/zenserver/upstream/upstreamcache.h
@@ -24,6 +24,7 @@ class AuthMgr;
class CbObjectView;
class CbPackage;
class CbObjectWriter;
+class CidStore;
class ZenCacheStore;
struct JupiterClientOptions;
class JupiterAccessTokenProvider;
@@ -161,6 +162,6 @@ struct UpstreamCacheOptions
bool WriteUpstream = true;
};
-std::unique_ptr<UpstreamCache> CreateUpstreamCache(const UpstreamCacheOptions& Options, ZenCacheStore& CacheStore);
+std::unique_ptr<UpstreamCache> CreateUpstreamCache(const UpstreamCacheOptions& Options, ZenCacheStore& CacheStore, CidStore& CidStore);
} // namespace zen
diff --git a/src/zenserver/zenserver.cpp b/src/zenserver/zenserver.cpp
index 71b52817c..48a2dad95 100644
--- a/src/zenserver/zenserver.cpp
+++ b/src/zenserver/zenserver.cpp
@@ -242,18 +242,18 @@ ZenServer::Initialize(const ZenServerOptions& ServerOptions, ZenServerState::Zen
CidStoreConfiguration Config;
Config.RootDirectory = m_DataRoot / "cas";
- m_CidStores.insert_or_assign({}, std::make_unique<CidStore>(m_GcManager));
- m_CidStores.at({})->Initialize(Config);
+ m_CidStore = std::make_unique<CidStore>(m_GcManager);
+ m_CidStore->Initialize(Config);
ZEN_INFO("instantiating project service");
- m_ProjectStore = new ProjectStore([this](std::string_view) -> CidStore& { return *m_CidStores.at({}).get(); },
+ m_ProjectStore = new ProjectStore(*m_CidStore,
m_DataRoot / "projects",
m_GcManager,
*m_JobQueue,
*m_OpenProcessCache,
ProjectStore::Configuration{});
- m_HttpProjectService.reset(new HttpProjectService(m_ProjectStore, m_StatusService, m_StatsService, *m_AuthMgr));
+ m_HttpProjectService.reset(new HttpProjectService{*m_CidStore, m_ProjectStore, m_StatusService, m_StatsService, *m_AuthMgr});
if (ServerOptions.WorksSpacesConfig.Enabled)
{
@@ -600,7 +600,7 @@ ZenServer::InitializeStructuredCache(const ZenServerOptions& ServerOptions)
UpstreamOptions.ThreadCount = static_cast<uint32_t>(UpstreamConfig.UpstreamThreadCount);
}
- m_UpstreamCache = CreateUpstreamCache(UpstreamOptions, *m_CacheStore);
+ m_UpstreamCache = CreateUpstreamCache(UpstreamOptions, *m_CacheStore, *m_CidStore);
m_UpstreamService = std::make_unique<HttpUpstreamService>(*m_UpstreamCache, *m_AuthMgr);
m_UpstreamCache->Initialize();
@@ -664,23 +664,19 @@ ZenServer::InitializeStructuredCache(const ZenServerOptions& ServerOptions)
}
}
- m_StructuredCacheService = std::make_unique<HttpStructuredCacheService>(
- *m_CacheStore,
- [this](std::string_view) -> CidStore& { return *m_CidStores.at({}).get(); },
- m_StatsService,
- m_StatusService,
- *m_UpstreamCache,
- m_GcManager.GetDiskWriteBlocker(),
- *m_OpenProcessCache);
+ m_StructuredCacheService = std::make_unique<HttpStructuredCacheService>(*m_CacheStore,
+ *m_CidStore,
+ m_StatsService,
+ m_StatusService,
+ *m_UpstreamCache,
+ m_GcManager.GetDiskWriteBlocker(),
+ *m_OpenProcessCache);
m_Http->RegisterService(*m_StructuredCacheService);
m_Http->RegisterService(*m_UpstreamService);
m_StatsReporter.AddProvider(m_CacheStore.Get());
- for (const auto& It : m_CidStores)
- {
- m_StatsReporter.AddProvider(It.second.get());
- }
+ m_StatsReporter.AddProvider(m_CidStore.get());
m_StatsReporter.AddProvider(m_BuildCidStore.get());
}
@@ -865,7 +861,7 @@ ZenServer::Cleanup()
m_Workspaces.reset();
m_HttpProjectService.reset();
m_ProjectStore = {};
- m_CidStores.clear();
+ m_CidStore.reset();
m_AuthService.reset();
m_AuthMgr.reset();
m_Http = {};
@@ -1053,10 +1049,8 @@ ZenServer::Flush()
{
ZEN_TRACE_CPU("ZenServer::Flush");
- for (auto& It : m_CidStores)
- {
- It.second->Flush();
- }
+ if (m_CidStore)
+ m_CidStore->Flush();
if (m_StructuredCacheService)
m_StructuredCacheService->Flush();
diff --git a/src/zenserver/zenserver.h b/src/zenserver/zenserver.h
index 0a17446ae..bcb02d336 100644
--- a/src/zenserver/zenserver.h
+++ b/src/zenserver/zenserver.h
@@ -115,20 +115,20 @@ private:
inline void SetNewState(ServerState NewState) { m_CurrentState = NewState; }
static std::string_view ToString(ServerState Value);
- StatsReporter m_StatsReporter;
- Ref<HttpServer> m_Http;
- std::unique_ptr<AuthMgr> m_AuthMgr;
- std::unique_ptr<HttpAuthService> m_AuthService;
- HttpStatusService m_StatusService;
- HttpStatsService m_StatsService;
- GcManager m_GcManager;
- GcScheduler m_GcScheduler{m_GcManager};
- tsl::robin_map<std::string, std::unique_ptr<CidStore>> m_CidStores;
- Ref<ZenCacheStore> m_CacheStore;
- std::unique_ptr<OpenProcessCache> m_OpenProcessCache;
- HttpTestService m_TestService;
- std::unique_ptr<CidStore> m_BuildCidStore;
- std::unique_ptr<BuildStore> m_BuildStore;
+ StatsReporter m_StatsReporter;
+ Ref<HttpServer> m_Http;
+ std::unique_ptr<AuthMgr> m_AuthMgr;
+ std::unique_ptr<HttpAuthService> m_AuthService;
+ HttpStatusService m_StatusService;
+ HttpStatsService m_StatsService;
+ GcManager m_GcManager;
+ GcScheduler m_GcScheduler{m_GcManager};
+ std::unique_ptr<CidStore> m_CidStore;
+ Ref<ZenCacheStore> m_CacheStore;
+ std::unique_ptr<OpenProcessCache> m_OpenProcessCache;
+ HttpTestService m_TestService;
+ std::unique_ptr<CidStore> m_BuildCidStore;
+ std::unique_ptr<BuildStore> m_BuildStore;
#if ZEN_WITH_TESTS
HttpTestingService m_TestingService;
diff --git a/src/zenstore/cache/cacherpc.cpp b/src/zenstore/cache/cacherpc.cpp
index ff21d1ede..5d9a68919 100644
--- a/src/zenstore/cache/cacherpc.cpp
+++ b/src/zenstore/cache/cacherpc.cpp
@@ -153,13 +153,13 @@ CacheRpcHandler::CacheRpcHandler(LoggerRef InLog,
CacheStats& InCacheStats,
UpstreamCacheClient& InUpstreamCache,
ZenCacheStore& InCacheStore,
- GetCidStoreFunc&& InGetCidStore,
+ CidStore& InCidStore,
const DiskWriteBlocker* InDiskWriteBlocker)
: m_Log(InLog)
, m_CacheStats(InCacheStats)
, m_UpstreamCache(InUpstreamCache)
, m_CacheStore(InCacheStore)
-, m_GetCidStore(std::move(InGetCidStore))
+, m_CidStore(InCidStore)
, m_DiskWriteBlocker(InDiskWriteBlocker)
{
}
@@ -174,12 +174,6 @@ CacheRpcHandler::AreDiskWritesAllowed() const
return (m_DiskWriteBlocker == nullptr || m_DiskWriteBlocker->AreDiskWritesAllowed());
}
-CidStore&
-CacheRpcHandler::GetCidStore(std::string_view Namespace)
-{
- return m_GetCidStore(Namespace);
-}
-
CacheRpcHandler::RpcResponseCode
CacheRpcHandler::HandleRpcRequest(const CacheRequestContext& Context,
std::string_view UriNamespace,
@@ -387,12 +381,9 @@ CacheRpcHandler::PutCacheRecord(PutRequestData& Request, const CbPackage* Packag
Stopwatch Timer;
- CidStore& ChunkStore = m_GetCidStore(Request.Namespace);
-
Request.RecordObject.IterateAttachments([this,
&Request,
Package,
- &ChunkStore,
&WriteAttachmentBuffers,
&WriteRawHashes,
&ValidAttachments,
@@ -421,7 +412,7 @@ CacheRpcHandler::PutCacheRecord(PutRequestData& Request, const CbPackage* Packag
Count.Invalid++;
}
}
- else if (ChunkStore.ContainsChunk(ValueHash))
+ else if (m_CidStore.ContainsChunk(ValueHash))
{
ValidAttachments.emplace_back(ValueHash);
Count.Valid++;
@@ -457,7 +448,7 @@ CacheRpcHandler::PutCacheRecord(PutRequestData& Request, const CbPackage* Packag
if (!WriteAttachmentBuffers.empty())
{
- std::vector<CidStore::InsertResult> InsertResults = ChunkStore.AddChunks(WriteAttachmentBuffers, WriteRawHashes);
+ std::vector<CidStore::InsertResult> InsertResults = m_CidStore.AddChunks(WriteAttachmentBuffers, WriteRawHashes);
for (size_t Index = 0; Index < InsertResults.size(); Index++)
{
if (InsertResults[Index].New)
@@ -484,12 +475,10 @@ CacheRpcHandler::PutCacheRecord(PutRequestData& Request, const CbPackage* Packag
if (HasUpstream && EnumHasAllFlags(Request.Policy.GetRecordPolicy(), CachePolicy::StoreRemote) && !IsPartialRecord)
{
- m_UpstreamCache.EnqueueUpstream(
- {.Type = ZenContentType::kCbPackage,
- .Namespace = Request.Namespace,
- .Key = Request.Key,
- .ValueContentIds = std::move(ValidAttachments)},
- [ChunkStore = &ChunkStore](const IoHash& ValueHash) { return ChunkStore->FindChunkByCid(ValueHash); });
+ m_UpstreamCache.EnqueueUpstream({.Type = ZenContentType::kCbPackage,
+ .Namespace = Request.Namespace,
+ .Key = Request.Key,
+ .ValueContentIds = std::move(ValidAttachments)});
}
return PutStatus::Success;
}
@@ -532,8 +521,6 @@ CacheRpcHandler::HandleRpcGetCacheRecords(const CacheRequestContext& Context, Cb
return CbPackage{};
}
- CidStore& ChunkStore = m_GetCidStore(Namespace.value());
-
const bool HasUpstream = m_UpstreamCache.IsActive();
eastl::fixed_vector<RecordRequestData, 16> Requests;
@@ -633,7 +620,7 @@ CacheRpcHandler::HandleRpcGetCacheRecords(const CacheRequestContext& Context, Cb
}
else if (EnumHasAllFlags(ValuePolicy, CachePolicy::SkipData))
{
- if (ChunkStore.ContainsChunk(Value.ContentId))
+ if (m_CidStore.ContainsChunk(Value.ContentId))
{
Value.Exists = true;
}
@@ -654,7 +641,7 @@ CacheRpcHandler::HandleRpcGetCacheRecords(const CacheRequestContext& Context, Cb
}
else
{
- if (IoBuffer Chunk = ChunkStore.FindChunkByCid(Value.ContentId))
+ if (IoBuffer Chunk = m_CidStore.FindChunkByCid(Value.ContentId))
{
if (Chunk.GetSize() > 0)
{
@@ -678,7 +665,7 @@ CacheRpcHandler::HandleRpcGetCacheRecords(const CacheRequestContext& Context, Cb
}
if (!RequestValueIndexes.empty())
{
- ChunkStore.IterateChunks(
+ m_CidStore.IterateChunks(
CidHashes,
[this, &Request, &RequestValueIndexes](size_t Index, const IoBuffer& Payload) -> bool {
try
@@ -771,7 +758,7 @@ CacheRpcHandler::HandleRpcGetCacheRecords(const CacheRequestContext& Context, Cb
}
}
- const auto OnCacheRecordGetComplete = [this, Namespace, &ChunkStore, &ParseValues, Context](CacheRecordGetCompleteParams&& Params) {
+ const auto OnCacheRecordGetComplete = [this, Namespace, &ParseValues, Context](CacheRecordGetCompleteParams&& Params) {
if (!Params.Record)
{
return;
@@ -840,7 +827,7 @@ CacheRpcHandler::HandleRpcGetCacheRecords(const CacheRequestContext& Context, Cb
Value.Exists = true;
if (StoreLocal)
{
- ChunkStore.AddChunk(Compressed.GetCompressed().Flatten().AsIoBuffer(), Attachment->GetHash());
+ m_CidStore.AddChunk(Compressed.GetCompressed().Flatten().AsIoBuffer(), Attachment->GetHash());
}
if (!EnumHasAllFlags(ValuePolicy, CachePolicy::SkipData))
{
@@ -957,8 +944,6 @@ CacheRpcHandler::HandleRpcPutCacheValues(const CacheRequestContext& Context, con
const bool HasUpstream = m_UpstreamCache.IsActive();
CbArrayView RequestsArray = Params["Requests"sv].AsArrayView();
- CidStore& ChunkStore = m_GetCidStore(Namespace.value());
-
std::vector<ZenCacheStore::PutResult> BatchResults;
eastl::fixed_vector<size_t, 32> BatchResultIndexes;
eastl::fixed_vector<ZenCacheStore::PutResult, 32> Results;
@@ -1109,8 +1094,7 @@ CacheRpcHandler::HandleRpcPutCacheValues(const CacheRequestContext& Context, con
if ((Results[Index].Status == zen::PutStatus::Success) && UpstreamCacheKeys[Index] != CacheKey::Empty)
{
m_UpstreamCache.EnqueueUpstream(
- {.Type = ZenContentType::kCompressedBinary, .Namespace = *Namespace, .Key = UpstreamCacheKeys[Index]},
- [ChunkStore = &ChunkStore](const IoHash& ValueHash) { return ChunkStore->FindChunkByCid(ValueHash); });
+ {.Type = ZenContentType::kCompressedBinary, .Namespace = *Namespace, .Key = UpstreamCacheKeys[Index]});
}
}
{
@@ -1562,8 +1546,6 @@ CacheRpcHandler::GetLocalCacheRecords(const CacheRequestContext& Context,
using namespace cache::detail;
const bool HasUpstream = m_UpstreamCache.IsActive();
- CidStore& ChunkStore = m_GetCidStore(Namespace);
-
// TODO: BatchGet records?
std::vector<CacheKeyRequest*> UpstreamRecordRequests;
for (size_t RecordIndex = 0; RecordIndex < Records.size(); ++RecordIndex)
@@ -1702,12 +1684,12 @@ CacheRpcHandler::GetLocalCacheRecords(const CacheRequestContext& Context,
{
if (EnumHasAllFlags(Request->DownstreamPolicy, CachePolicy::SkipData) && Request->RawSizeKnown)
{
- if (ChunkStore.ContainsChunk(Request->Key->ChunkId))
+ if (m_CidStore.ContainsChunk(Request->Key->ChunkId))
{
Request->Exists = true;
}
}
- else if (IoBuffer Payload = ChunkStore.FindChunkByCid(Request->Key->ChunkId))
+ else if (IoBuffer Payload = m_CidStore.FindChunkByCid(Request->Key->ChunkId))
{
if (!EnumHasAllFlags(Request->DownstreamPolicy, CachePolicy::SkipData))
{
@@ -1840,8 +1822,6 @@ CacheRpcHandler::GetUpstreamCacheChunks(const CacheRequestContext& Context,
return;
}
- CidStore& ChunkStore = m_GetCidStore(Namespace);
-
CacheChunkRequest& Key = Params.Request;
size_t RequestIndex = std::distance(RequestKeys.data(), &Key);
ChunkRequest& Request = Requests[RequestIndex];
@@ -1861,7 +1841,7 @@ CacheRpcHandler::GetUpstreamCacheChunks(const CacheRequestContext& Context,
bool Overwrite = !EnumHasAllFlags(Request.DownstreamPolicy, CachePolicy::QueryLocal);
if (Request.IsRecordRequest)
{
- ChunkStore.AddChunk(Params.Value, Params.RawHash);
+ m_CidStore.AddChunk(Params.Value, Params.RawHash);
}
else
{
diff --git a/src/zenstore/include/zenstore/cache/cacherpc.h b/src/zenstore/include/zenstore/cache/cacherpc.h
index 80340d72c..104746aba 100644
--- a/src/zenstore/include/zenstore/cache/cacherpc.h
+++ b/src/zenstore/include/zenstore/cache/cacherpc.h
@@ -70,13 +70,11 @@ IsCompressedBinary(ZenContentType Type)
struct CacheRpcHandler
{
- typedef std::function<CidStore&(std::string_view Context)> GetCidStoreFunc;
-
CacheRpcHandler(LoggerRef InLog,
CacheStats& InCacheStats,
UpstreamCacheClient& InUpstreamCache,
ZenCacheStore& InCacheStore,
- GetCidStoreFunc&& InGetCidStore,
+ CidStore& InCidStore,
const DiskWriteBlocker* InDiskWriteBlocker);
~CacheRpcHandler();
@@ -96,8 +94,6 @@ struct CacheRpcHandler
int& OutTargetProcessId,
CbPackage& OutPackage);
- CidStore& GetCidStore(std::string_view Namespace);
-
private:
CbPackage HandleRpcPutCacheRecords(const CacheRequestContext& Context, const CbPackage& BatchRequest);
CbPackage HandleRpcGetCacheRecords(const CacheRequestContext& Context, CbObjectView BatchRequest);
@@ -146,7 +142,7 @@ private:
CacheStats& m_CacheStats;
UpstreamCacheClient& m_UpstreamCache;
ZenCacheStore& m_CacheStore;
- GetCidStoreFunc m_GetCidStore;
+ CidStore& m_CidStore;
const DiskWriteBlocker* m_DiskWriteBlocker = nullptr;
bool AreDiskWritesAllowed() const;
diff --git a/src/zenstore/include/zenstore/cache/upstreamcacheclient.h b/src/zenstore/include/zenstore/cache/upstreamcacheclient.h
index c3993c028..152031c3a 100644
--- a/src/zenstore/include/zenstore/cache/upstreamcacheclient.h
+++ b/src/zenstore/include/zenstore/cache/upstreamcacheclient.h
@@ -113,7 +113,7 @@ public:
std::span<CacheChunkRequest*> CacheChunkRequests,
OnCacheChunksGetComplete&& OnComplete) = 0;
- virtual void EnqueueUpstream(UpstreamCacheRecord CacheRecord, std::function<IoBuffer(const IoHash&)>&& GetValueFunc) = 0;
+ virtual void EnqueueUpstream(UpstreamCacheRecord CacheRecord) = 0;
};
} // namespace zen