1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
|
// Copyright Epic Games, Inc. All Rights Reserved.
#pragma once
#include <zencore/stats.h>
#include <zenhttp/httpserver.h>
#include <zenhttp/websocket.h>
#include "monitoring/httpstats.h"
#include "monitoring/httpstatus.h"
#include <memory>
#include <vector>
namespace spdlog {
class logger;
}
namespace zen {
struct CacheChunkRequest;
struct CacheKeyRequest;
class CasStore;
class CidStore;
class CbObjectView;
struct PutRequestData;
class ScrubContext;
class UpstreamCache;
class ZenCacheStore;
enum class CachePolicy : uint32_t;
namespace cache::detail {
struct RecordBody;
struct ChunkRequest;
} // namespace cache::detail
/**
* Structured cache service. Imposes constraints on keys, supports blobs and
* structured values
*
* Keys are structured as:
*
* {BucketId}/{KeyHash}
*
* Where BucketId is a lower-case alphanumeric string, and KeyHash is a 40-character
* hexadecimal sequence. The hash value may be derived in any number of ways, it's
* up to the application to pick an approach.
*
* Values may be structured or unstructured. Structured values are encoded using Unreal
* Engine's compact binary encoding (see CbObject)
*
* Additionally, attachments may be addressed as:
*
* {BucketId}/{KeyHash}/{ValueHash}
*
* Where the two initial components are the same as for the main endpoint
*
* The storage strategy is as follows:
*
* - Structured values are stored in a dedicated backing store per bucket
* - Unstructured values and attachments are stored in the CAS pool
*
*/
class HttpStructuredCacheService : public HttpService, public IHttpStatsProvider, public IHttpStatusProvider, public WebSocketService
{
public:
HttpStructuredCacheService(ZenCacheStore& InCacheStore,
CidStore& InCidStore,
HttpStatsService& StatsService,
HttpStatusService& StatusService,
UpstreamCache& UpstreamCache);
~HttpStructuredCacheService();
virtual const char* BaseUri() const override;
virtual void HandleRequest(zen::HttpServerRequest& Request) override;
void Flush();
void Scrub(ScrubContext& Ctx);
private:
struct CacheRef
{
std::string BucketSegment;
IoHash HashKey;
IoHash ValueContentId;
};
struct CacheStats
{
std::atomic_uint64_t HitCount{};
std::atomic_uint64_t UpstreamHitCount{};
std::atomic_uint64_t MissCount{};
};
enum class PutResult
{
Success,
Fail,
Invalid,
};
[[nodiscard]] bool ValidateKeyUri(zen::HttpServerRequest& Request, CacheRef& OutRef);
void HandleCacheRecordRequest(zen::HttpServerRequest& Request, const CacheRef& Ref, CachePolicy PolicyFromURL);
void HandleGetCacheRecord(zen::HttpServerRequest& Request, const CacheRef& Ref, CachePolicy PolicyFromUrl);
void HandlePutCacheRecord(zen::HttpServerRequest& Request, const CacheRef& Ref, CachePolicy PolicyFromURL);
void HandleCacheValueRequest(zen::HttpServerRequest& Request, const CacheRef& Ref, CachePolicy PolicyFromURL);
void HandleGetCacheValue(zen::HttpServerRequest& Request, const CacheRef& Ref, CachePolicy PolicyFromURL);
void HandlePutCacheValue(zen::HttpServerRequest& Request, const CacheRef& Ref, CachePolicy PolicyFromURL);
void HandleRpcRequest(zen::HttpServerRequest& Request);
void HandleRpcPutCacheRecords(zen::HttpServerRequest& Request, const CbPackage& BatchRequest);
void HandleRpcGetCacheRecords(zen::HttpServerRequest& Request, CbObjectView BatchRequest);
void HandleRpcPutCacheValues(zen::HttpServerRequest& Request, const CbPackage& BatchRequest);
void HandleRpcGetCacheValues(zen::HttpServerRequest& Request, CbObjectView BatchRequest);
void HandleRpcGetCacheChunks(zen::HttpServerRequest& Request, CbObjectView BatchRequest);
void HandleCacheBucketRequest(zen::HttpServerRequest& Request, std::string_view Bucket);
virtual void HandleStatsRequest(zen::HttpServerRequest& Request) override;
virtual void HandleStatusRequest(zen::HttpServerRequest& Request) override;
PutResult PutCacheRecord(PutRequestData& Request, const CbPackage* Package);
/** HandleRpcGetCacheChunks Helper: Parse the Body object into RecordValue Requests and Value Requests. */
bool ParseGetCacheChunksRequest(std::vector<CacheKeyRequest>& RecordKeys,
std::vector<cache::detail::RecordBody>& Records,
std::vector<CacheChunkRequest>& RequestKeys,
std::vector<cache::detail::ChunkRequest>& Requests,
std::vector<cache::detail::ChunkRequest*>& RecordRequests,
std::vector<cache::detail::ChunkRequest*>& ValueRequests,
CbObjectView RpcRequest);
/** HandleRpcGetCacheChunks Helper: Load records to get ContentId for RecordRequests, and load their payloads if they exist locally. */
void GetLocalCacheRecords(std::vector<CacheKeyRequest>& RecordKeys,
std::vector<cache::detail::RecordBody>& Records,
std::vector<cache::detail::ChunkRequest*>& RecordRequests,
std::vector<CacheChunkRequest*>& OutUpstreamChunks);
/** HandleRpcGetCacheChunks Helper: For ValueRequests, load their payloads if they exist locally. */
void GetLocalCacheValues(std::vector<cache::detail::ChunkRequest*>& ValueRequests, std::vector<CacheChunkRequest*>& OutUpstreamChunks);
/** HandleRpcGetCacheChunks Helper: Load payloads from upstream that did not exist locally. */
void GetUpstreamCacheChunks(std::vector<CacheChunkRequest*>& UpstreamChunks,
std::vector<CacheChunkRequest>& RequestKeys,
std::vector<cache::detail::ChunkRequest>& Requests);
/** HandleRpcGetCacheChunks Helper: Send response message containing all chunk results. */
void WriteGetCacheChunksResponse(std::vector<cache::detail::ChunkRequest>& Requests, zen::HttpServerRequest& HttpRequest);
virtual void RegisterHandlers(WebSocketServer& Server) override;
virtual bool HandleRequest(const WebSocketMessage& RequestMessage) override;
spdlog::logger& Log() { return m_Log; }
spdlog::logger& m_Log;
ZenCacheStore& m_CacheStore;
HttpStatsService& m_StatsService;
HttpStatusService& m_StatusService;
CidStore& m_CidStore;
UpstreamCache& m_UpstreamCache;
uint64_t m_LastScrubTime = 0;
metrics::OperationTiming m_HttpRequests;
metrics::OperationTiming m_UpstreamGetRequestTiming;
CacheStats m_CacheStats;
};
/** Recognize both kBinary and kCompressedBinary as kCompressedBinary for structured cache value keys.
* We need this until the content type is preserved for kCompressedBinary when passing to and from upstream servers. */
inline bool
IsCompressedBinary(ZenContentType Type)
{
return Type == ZenContentType::kBinary || Type == ZenContentType::kCompressedBinary;
}
} // namespace zen
|