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
|
// Copyright Yahoo. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
#pragma once
#include "bootstrapconfig.h"
#include "bootstrapconfigmanager.h"
#include "documentdb.h"
#include "health_adapter.h"
#include "i_proton_configurer_owner.h"
#include "idocumentdbowner.h"
#include "memory_flush_config_updater.h"
#include "proton_config_fetcher.h"
#include "proton_configurer.h"
#include "rpc_hooks.h"
#include "shared_threading_service.h"
#include <vespa/eval/eval/llvm/compile_cache.h>
#include <vespa/searchcore/proton/matching/querylimiter.h>
#include <vespa/searchcore/proton/persistenceengine/i_resource_write_filter.h>
#include <vespa/searchcore/proton/persistenceengine/ipersistenceengineowner.h>
#include <vespa/searchlib/common/fileheadercontext.h>
#include <vespa/searchlib/engine/monitorapi.h>
#include <vespa/vespalib/net/http/component_config_producer.h>
#include <vespa/vespalib/net/http/generic_state_handler.h>
#include <vespa/vespalib/net/http/json_get_handler.h>
#include <vespa/vespalib/net/http/json_handler_repo.h>
#include <vespa/vespalib/net/http/state_explorer.h>
#include <vespa/vespalib/util/cpu_usage.h>
#include <mutex>
#include <shared_mutex>
namespace vespalib { class StateServer; }
namespace search {
namespace attribute { class Interlock; }
namespace transactionlog { class TransLogServerApp; }
}
namespace metrics {
class MetricLockGuard;
class MetricManager;
}
namespace storage::spi { struct PersistenceProvider; }
namespace proton {
class DiskMemUsageSampler;
class FlushEngine;
class IDocumentDBReferenceRegistry;
class IProtonDiskLayout;
class MatchEngine;
class MetricsEngine;
class PersistenceEngine;
class PrepareRestartHandler;
class SummaryEngine;
class Proton : public IProtonConfigurerOwner,
public search::engine::MonitorServer,
public IDocumentDBOwner,
public StatusProducer,
public IPersistenceEngineOwner,
public vespalib::ComponentConfigProducer,
public vespalib::StateExplorer
{
private:
using TLS = search::transactionlog::TransLogServerApp;
using MonitorRequest = search::engine::MonitorRequest;
using MonitorReply = search::engine::MonitorReply;
using MonitorClient = search::engine::MonitorClient;
using DocumentDBMap = std::map<DocTypeName, DocumentDB::SP>;
using InitializeThreads = std::shared_ptr<vespalib::ThreadExecutor>;
using BucketSpace = document::BucketSpace;
class ProtonFileHeaderContext : public search::common::FileHeaderContext
{
vespalib::string _hostName;
vespalib::string _creator;
vespalib::string _cluster;
pid_t _pid;
public:
explicit ProtonFileHeaderContext(const vespalib::string &creator);
~ProtonFileHeaderContext() override;
void addTags(vespalib::GenericHeader &header, const vespalib::string &name) const override;
void setClusterName(const vespalib::string &clusterName, const vespalib::string &baseDir);
};
vespalib::CpuUtil _cpu_util;
HwInfo _hw_info;
FastOS_ThreadPool & _threadPool;
FNET_Transport & _transport;
const config::ConfigUri _configUri;
mutable std::shared_mutex _mutex;
std::unique_ptr<metrics::UpdateHook> _metricsHook;
std::unique_ptr<MetricsEngine> _metricsEngine;
ProtonFileHeaderContext _fileHeaderContext;
std::shared_ptr<search::attribute::Interlock> _attribute_interlock;
std::unique_ptr<TLS> _tls;
std::unique_ptr<DiskMemUsageSampler> _diskMemUsageSampler;
std::unique_ptr<PersistenceEngine> _persistenceEngine;
DocumentDBMap _documentDBMap;
std::unique_ptr<MatchEngine> _matchEngine;
std::unique_ptr<SummaryEngine> _summaryEngine;
MemoryFlushConfigUpdater::UP _memoryFlushConfigUpdater;
std::unique_ptr<FlushEngine> _flushEngine;
std::unique_ptr<PrepareRestartHandler> _prepareRestartHandler;
RPCHooks::UP _rpcHooks;
HealthAdapter _healthAdapter;
vespalib::GenericStateHandler _genericStateHandler;
vespalib::JsonHandlerRepo::Token::UP _customComponentBindToken;
vespalib::JsonHandlerRepo::Token::UP _customComponentRootToken;
std::unique_ptr<vespalib::StateServer> _stateServer;
vespalib::ThreadStackExecutor _executor;
std::unique_ptr<IProtonDiskLayout> _protonDiskLayout;
ProtonConfigurer _protonConfigurer;
ProtonConfigFetcher _protonConfigFetcher;
std::unique_ptr<SharedThreadingService> _shared_service;
vespalib::eval::CompileCache::ExecutorBinding::UP _compile_cache_executor_binding;
matching::QueryLimiter _queryLimiter;
uint32_t _distributionKey;
bool _isInitializing;
bool _abortInit;
bool _initStarted;
bool _initComplete;
bool _initDocumentDbsInSequence;
bool _has_shut_down_config_and_state_components;
std::shared_ptr<IDocumentDBReferenceRegistry> _documentDBReferenceRegistry;
std::mutex _nodeUpLock;
std::set<BucketSpace> _nodeUp; // bucketspaces where node is up
std::shared_ptr<DocumentDBConfigOwner>
addDocumentDB(const DocTypeName & docTypeName, BucketSpace bucketSpace, const vespalib::string & configid,
const BootstrapConfig::SP & bootstrapConfig, const std::shared_ptr<DocumentDBConfig> &documentDBConfig,
InitializeThreads initializeThreads) override;
void removeDocumentDB(const DocTypeName &docTypeName) override;
void applyConfig(const BootstrapConfig::SP & configSnapshot) override;
std::unique_ptr<MonitorReply> ping(std::unique_ptr<MonitorRequest> request, MonitorClient &client) override;
void waitForInitDone();
void waitForOnlineState();
uint32_t getDistributionKey() const override { return _distributionKey; }
BootstrapConfig::SP getActiveConfigSnapshot() const;
std::shared_ptr<IDocumentDBReferenceRegistry> getDocumentDBReferenceRegistry() const override;
SessionManager & session_manager() override;
// Returns true if the node is up in _any_ bucket space
bool updateNodeUp(BucketSpace bucketSpace, bool nodeUpInBucketSpace);
void closeDocumentDBs(vespalib::ThreadStackExecutorBase & executor);
public:
typedef std::unique_ptr<Proton> UP;
typedef std::shared_ptr<Proton> SP;
Proton(FastOS_ThreadPool & threadPool, FNET_Transport & transport, const config::ConfigUri & configUri,
const vespalib::string &progName, vespalib::duration subscribeTimeout);
~Proton() override;
/**
* Called by the metrics update hook (typically in the context of
* the metric manager). Do not call this function in multiple
* threads at once.
**/
void updateMetrics(const metrics::MetricLockGuard &guard);
/**
* This method must be called after the constructor and before the destructor.
* If not I will force a 'core' upon you.
* All relevant initialization is conducted here.
*
* 1st phase init: start cheap clock thread and get initial config
*/
BootstrapConfig::SP init();
/**
* Shuts down metric manager and state server functionality to avoid
* calls to these during service layer component tear-down.
*
* Explicitly noexcept to avoid consistency issues between this and the
* destructor if something throws during shutdown.
*/
void shutdown_config_fetching_and_state_exposing_components_once() noexcept;
// 2nd phase init: setup data structures.
void init(const BootstrapConfig::SP & configSnapshot);
DocumentDB::SP
addDocumentDB(const document::DocumentType &docType, BucketSpace bucketSpace, const BootstrapConfig::SP &configSnapshot,
const std::shared_ptr<DocumentDBConfig> &documentDBConfig, InitializeThreads initializeThreads);
metrics::MetricManager & getMetricManager();
FastOS_ThreadPool & getThreadPool() { return _threadPool; }
bool triggerFlush();
bool prepareRestart();
void getComponentConfig(Consumer &consumer) override;
void setClusterState(BucketSpace bucketSpace, const storage::spi::ClusterState &calc) override;
// Return the oldest active config generation used by proton.
int64_t getConfigGeneration();
size_t getNumDocs() const;
// Active (searchable), and targetActive that will be searchable when idealstate is reached
ActiveDocs getNumActiveDocs() const;
search::engine::SearchServer &get_search_server();
search::engine::DocsumServer &get_docsum_server();
search::engine::MonitorServer &get_monitor_server();
vespalib::string getDelayedConfigs() const;
StatusReport::List getStatusReports() const override;
MatchEngine & getMatchEngine() { return *_matchEngine; }
vespalib::ThreadStackExecutorBase & getExecutor() { return _executor; }
bool isInitializing() const override { return _isInitializing; }
bool hasAbortedInit() const { return _abortInit; }
storage::spi::PersistenceProvider & getPersistence();
void get_state(const vespalib::slime::Inserter &inserter, bool full) const override;
std::vector<vespalib::string> get_children_names() const override;
std::unique_ptr<vespalib::StateExplorer> get_child(vespalib::stringref name) const override;
};
} // namespace proton
|
