diff options
| author | Dan Engelbrecht <[email protected]> | 2025-08-26 11:43:37 +0200 |
|---|---|---|
| committer | GitHub Enterprise <[email protected]> | 2025-08-26 11:43:37 +0200 |
| commit | fb137cf9c8b7a9d1659b03472c9591c4863e9173 (patch) | |
| tree | bbd0df09d4b425ef668d22f3b12ea2cb3482bf66 /src | |
| parent | Merge pull request #139 from ue-foundation/de/zen-service-command (diff) | |
| download | zen-fb137cf9c8b7a9d1659b03472c9591c4863e9173.tar.xz zen-fb137cf9c8b7a9d1659b03472c9591c4863e9173.zip | |
revert multi-cid store (#475)
Diffstat (limited to 'src')
| -rw-r--r-- | src/zenserver/cache/httpstructuredcache.cpp | 286 | ||||
| -rw-r--r-- | src/zenserver/cache/httpstructuredcache.h | 5 | ||||
| -rw-r--r-- | src/zenserver/projectstore/httpprojectstore.cpp | 167 | ||||
| -rw-r--r-- | src/zenserver/projectstore/httpprojectstore.h | 7 | ||||
| -rw-r--r-- | src/zenserver/projectstore/projectstore.cpp | 169 | ||||
| -rw-r--r-- | src/zenserver/projectstore/projectstore.h | 8 | ||||
| -rw-r--r-- | src/zenserver/upstream/upstreamcache.cpp | 38 | ||||
| -rw-r--r-- | src/zenserver/upstream/upstreamcache.h | 3 | ||||
| -rw-r--r-- | src/zenserver/zenserver.cpp | 38 | ||||
| -rw-r--r-- | src/zenserver/zenserver.h | 28 | ||||
| -rw-r--r-- | src/zenstore/cache/cacherpc.cpp | 54 | ||||
| -rw-r--r-- | src/zenstore/include/zenstore/cache/cacherpc.h | 8 | ||||
| -rw-r--r-- | src/zenstore/include/zenstore/cache/upstreamcacheclient.h | 2 |
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 |