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
|
// Copyright 2017 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
/**
* @class storage::StateManager
* @ingroup storageserver
*
* @brief Keeps and updates node and system states.
*
* This component implements the NodeStateUpdater interface to handle states
* for all components. See that interface for documentation.
*
* In addition, this manager is a storage link such that it can handle the
* various commands for setting and retrieving states.
*/
#pragma once
#include <vespa/storage/common/hostreporter/hostinfo.h>
#include <vespa/storage/common/nodestateupdater.h>
#include <vespa/storage/common/storagelink.h>
#include <vespa/storage/common/storagecomponent.h>
#include <vespa/storageframework/generic/status/htmlstatusreporter.h>
#include <vespa/storageapi/message/state.h>
#include <vespa/storageapi/messageapi/storagemessage.h>
#include <vespa/vespalib/util/sync.h>
#include <vespa/vespalib/objects/floatingpointtype.h>
#include <deque>
#include <map>
#include <list>
#include <atomic>
namespace metrics {
class MetricManager;
}
namespace storage {
class StateManager : public NodeStateUpdater,
public StorageLink,
public framework::HtmlStatusReporter,
private framework::Runnable,
private vespalib::JsonStreamTypes
{
bool _noThreadTestMode;
StorageComponent _component;
metrics::MetricManager& _metricManager;
vespalib::Monitor _stateLock;
vespalib::Lock _listenerLock;
bool _grabbedExternalLock;
std::atomic<bool> _notifyingListeners;
std::shared_ptr<lib::NodeState> _nodeState;
std::shared_ptr<lib::NodeState> _nextNodeState;
std::shared_ptr<lib::ClusterState> _systemState;
std::shared_ptr<lib::ClusterState> _nextSystemState;
std::list<StateListener*> _stateListeners;
typedef std::pair<framework::MilliSecTime,
api::GetNodeStateCommand::SP> TimeStatePair;
std::list<TimeStatePair> _queuedStateRequests;
mutable vespalib::Monitor _threadMonitor;
framework::MilliSecTime _lastProgressUpdateCausingSend;
vespalib::Double _progressLastInitStateSend;
typedef std::pair<framework::MilliSecTime,
lib::ClusterState::SP> TimeSysStatePair;
std::deque<TimeSysStatePair> _systemStateHistory;
uint32_t _systemStateHistorySize;
std::unique_ptr<HostInfo> _hostInfo;
framework::Thread::UP _thread;
public:
explicit StateManager(StorageComponentRegister&, metrics::MetricManager&,
std::unique_ptr<HostInfo>, bool testMode = false);
~StateManager();
void onOpen() override;
void onClose() override;
void tick();
void print(std::ostream& out, bool verbose, const std::string& indent) const override;
void reportHtmlStatus(std::ostream&, const framework::HttpUrlPath&) const override;
lib::NodeState::CSP getReportedNodeState() const override;
lib::NodeState::CSP getCurrentNodeState() const override;
lib::ClusterState::CSP getSystemState() const override;
void addStateListener(StateListener&) override;
void removeStateListener(StateListener&) override;
Lock::SP grabStateChangeLock() override;
void setReportedNodeState(const lib::NodeState& state) override;
void setClusterState(const lib::ClusterState& c);
HostInfo& getHostInfo() { return *_hostInfo; }
private:
class ExternalStateLock;
friend class ExternalStateLock;
friend class StateManagerTest;
void notifyStateListeners();
bool sendGetNodeStateReplies(
framework::MilliSecTime olderThanTime = framework::MilliSecTime(0),
uint16_t index = 0xffff);
lib::Node thisNode() const;
/**
* Overwrite the current cluster state with the one that is currently
* pending.
*
* Appends the pending cluster state to a circular buffer of historic
* states.
*
* Preconditions:
* - _stateLock is held
* - _systemState.get() != nullptr
* - _nextSystemState.get() != nullptr
* Postconditions:
* - _systemState = old(_nextSystemState)
* - _nextSystemState.get() == nullptr
*/
void enableNextClusterState();
/**
* Log this node's state transition as given by the cluster state iff the
* state differs between currentState and newState.
*/
void logNodeClusterStateTransition(
const lib::ClusterState& currentState,
const lib::ClusterState& newState) const;
bool onGetNodeState(const std::shared_ptr<api::GetNodeStateCommand>&) override;
bool onSetSystemState(const std::shared_ptr<api::SetSystemStateCommand>&) override;
/**
* _stateLock MUST NOT be held while calling.
*/
std::string getNodeInfo() const;
void run(framework::ThreadHandle&) override;
};
} // storage
|
