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
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
|
// Copyright Epic Games, Inc. All Rights Reserved.
#pragma once
#include "hubinstancestate.h"
#include "resourcemetrics.h"
#include "storageserverinstance.h"
#include <zencore/system.h>
#include <zenutil/zenserverprocess.h>
#include <deque>
#include <filesystem>
#include <functional>
#include <memory>
#include <thread>
#include <unordered_map>
#include <unordered_set>
namespace zen {
class WorkerThreadPool;
/**
* Hub
*
* Core logic for managing storage server instances on behalf of external clients.
*/
struct HubProvisionedInstanceInfo
{
std::string BaseUri;
uint16_t Port;
};
class Hub
{
public:
struct Configuration
{
/** Enable or disable the use of a Windows Job Object for child process management.
* When enabled, all spawned child processes are assigned to a job object with
* JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE, ensuring children are terminated if the hub
* crashes or is force-killed.
*/
bool UseJobObject = true;
uint16_t BasePortNumber = 21000;
int InstanceLimit = 1000;
uint32_t InstanceHttpThreadCount = 0; // Automatic
int InstanceCoreLimit = 0; // Automatic
std::filesystem::path InstanceConfigPath;
std::string HydrationTargetSpecification;
};
typedef std::function<
void(std::string_view ModuleId, const HubProvisionedInstanceInfo& Info, HubInstanceState OldState, HubInstanceState NewState)>
AsyncModuleStateChangeCallbackFunc;
Hub(const Configuration& Config,
ZenServerEnvironment&& RunEnvironment,
WorkerThreadPool* OptionalWorkerPool = nullptr,
AsyncModuleStateChangeCallbackFunc&& ModuleStateChangeCallback = {});
~Hub();
Hub(const Hub&) = delete;
Hub& operator=(const Hub&) = delete;
struct InstanceInfo
{
HubInstanceState State = HubInstanceState::Unprovisioned;
std::chrono::system_clock::time_point ProvisionTime;
ProcessMetrics Metrics;
uint16_t Port = 0;
};
/**
* Deprovision all running instances
*/
void Shutdown();
enum class EResponseCode
{
NotFound,
Rejected,
Accepted,
Completed
};
struct Response
{
EResponseCode ResponseCode = EResponseCode::Rejected;
std::string Message;
};
/**
* Provision a storage server instance for the given module ID.
*
* @param ModuleId The ID of the module to provision.
* @param OutInfo On success, information about the provisioned instance is returned here.
*/
Response Provision(std::string_view ModuleId, HubProvisionedInstanceInfo& OutInfo);
/**
* Deprovision a storage server instance for the given module ID.
*
* @param ModuleId The ID of the module to deprovision.
*/
Response Deprovision(const std::string& ModuleId);
/**
* Hibernate a storage server instance for the given module ID.
* The instance is shut down but its data is preserved; it can be woken later.
*
* @param ModuleId The ID of the module to hibernate.
*/
Response Hibernate(const std::string& ModuleId);
/**
* Wake a hibernated storage server instance for the given module ID.
*
* @param ModuleId The ID of the module to wake.
*/
Response Wake(const std::string& ModuleId);
/**
* Find info about storage server instance for the given module ID.
*
* @param ModuleId The ID of the module to find.
* @param OutInstanceInfo If found, the instance info will be returned here.
* @return true if the instance was found, false otherwise.
*/
bool Find(std::string_view ModuleId, InstanceInfo* OutInstanceInfo = nullptr);
/**
* Enumerate a snapshot of all storage server instances.
*
* @param Callback The callback to invoke for each instance.
*/
void EnumerateModules(std::function<void(std::string_view ModuleId, const InstanceInfo&)> Callback);
int GetInstanceCount();
int GetMaxInstanceCount() const { return m_MaxInstanceCount.load(); }
const Configuration& GetConfig() const { return m_Config; }
#if ZEN_WITH_TESTS
void TerminateModuleForTesting(const std::string& ModuleId);
#endif
private:
const Configuration m_Config;
ZenServerEnvironment m_RunEnvironment;
WorkerThreadPool* m_WorkerPool = nullptr;
Latch m_BackgroundWorkLatch;
std::atomic<bool> m_ShutdownFlag = false;
AsyncModuleStateChangeCallbackFunc m_ModuleStateChangeCallback;
std::string m_HydrationTargetSpecification;
std::filesystem::path m_HydrationTempPath;
#if ZEN_PLATFORM_WINDOWS
JobObject m_JobObject;
#endif
RwLock m_Lock;
std::unordered_map<std::string, size_t> m_InstanceLookup;
std::unordered_map<std::string, uint16_t> m_DeprovisioningModules;
std::unordered_map<std::string, uint16_t> m_ProvisioningModules;
std::unordered_map<std::string, uint16_t> m_HibernatingModules;
std::unordered_map<std::string, uint16_t> m_WakingModules;
std::unordered_set<std::string> m_RecoveringModules;
std::vector<std::unique_ptr<StorageServerInstance>> m_ActiveInstances;
std::vector<size_t> m_FreeActiveInstanceIndexes;
ResourceMetrics m_ResourceLimits;
SystemMetrics m_HostMetrics;
std::atomic<int> m_MaxInstanceCount = 0;
std::deque<uint16_t> m_FreePorts;
std::thread m_WatchDog;
Event m_WatchDogEvent;
void WatchDog();
void AttemptRecoverInstance(std::string_view ModuleId);
void UpdateStats();
void UpdateCapacityMetrics();
bool CanProvisionInstance(std::string_view ModuleId, std::string& OutReason);
bool IsModuleInFlightLocked(std::string_view ModuleId) const;
Response InternalDeprovision(const std::string& ModuleId);
void CompleteProvision(StorageServerInstance::ExclusiveLockedPtr& Instance, uint16_t AllocatedPort, bool IsNewInstance);
void AbortProvision(std::string_view ModuleId);
void CompleteDeprovision(StorageServerInstance::ExclusiveLockedPtr& Instance);
void CompleteHibernate(StorageServerInstance::ExclusiveLockedPtr& Instance);
void CompleteWake(StorageServerInstance::ExclusiveLockedPtr& Instance);
class InstanceStateUpdateGuard
{
public:
InstanceStateUpdateGuard(Hub& InHub,
std::string_view ModuleId,
HubInstanceState OldState,
HubInstanceState& NewState,
uint16_t BasePort,
const std::string& BaseUri)
: m_Hub(InHub)
, m_ModuleId(ModuleId)
, m_OldState(OldState)
, m_NewState(NewState)
, m_BasePort(BasePort)
, m_BaseUri(BaseUri)
{
}
~InstanceStateUpdateGuard() { m_Hub.OnStateUpdate(m_ModuleId, m_OldState, m_NewState, m_BasePort, m_BaseUri); }
private:
Hub& m_Hub;
const std::string m_ModuleId;
HubInstanceState m_OldState;
HubInstanceState& m_NewState;
uint16_t m_BasePort;
const std::string m_BaseUri;
};
void OnStateUpdate(std::string_view ModuleId,
HubInstanceState OldState,
HubInstanceState& NewState,
uint16_t BasePort,
std::string_view BaseUri);
friend class InstanceStateUpdateGuard;
};
#if ZEN_WITH_TESTS
void hub_forcelink();
#endif // ZEN_WITH_TESTS
} // namespace zen
|