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
|
// **********************************************************************
//
// Copyright (c) 2003-2017 ZeroC, Inc. All rights reserved.
//
// This copy of Ice is licensed to you under the terms described in the
// ICE_LICENSE file included in this distribution.
//
// **********************************************************************
#ifndef ELECTION_I_H
#define ELECTION_I_H
#include <IceUtil/IceUtil.h>
#include <Ice/Ice.h>
#include <IceStorm/Election.h>
#include <IceStorm/Replica.h>
#include <IceStorm/Instance.h>
#include <IceUtil/Timer.h>
#include <set>
namespace IceStormElection
{
class Observers;
typedef IceUtil::Handle<Observers> ObserversPtr;
class NodeI : public Node, public IceUtil::Monitor<IceUtil::RecMutex>
{
public:
NodeI(const IceStorm::InstancePtr&, const ReplicaPtr&, const Ice::ObjectPrx&,
int, const std::map<int, NodePrx>&);
void start();
void check();
void timeout();
void merge(const std::set<int>&);
void mergeContinue();
virtual void invitation(int, const std::string&, const Ice::Current&);
virtual void ready(int, const std::string&, const Ice::ObjectPrx&, int, Ice::Long, const Ice::Current&);
virtual void accept(int, const std::string&, const Ice::IntSeq&, const Ice::ObjectPrx&, const LogUpdate&, int,
const Ice::Current&);
virtual bool areYouCoordinator(const Ice::Current&) const;
virtual bool areYouThere(const std::string&, int, const Ice::Current&) const;
virtual Ice::ObjectPrx sync(const Ice::Current&) const;
virtual NodeInfoSeq nodes(const Ice::Current&) const;
virtual QueryInfo query(const Ice::Current&) const;
void recovery(Ice::Long = -1);
void destroy();
// Notify the node that we're about to start an update.
void checkObserverInit(Ice::Long);
Ice::ObjectPrx startUpdate(Ice::Long&, const char*, int);
Ice::ObjectPrx startCachedRead(Ice::Long&, const char*, int);
void startObserverUpdate(Ice::Long, const char*, int);
bool updateMaster(const char*, int);
// The node has completed the update.
void finishUpdate();
private:
void setState(NodeState);
const IceUtil::TimerPtr _timer;
const IceStorm::TraceLevelsPtr _traceLevels;
const IceStormElection::ObserversPtr _observers;
const ReplicaPtr _replica; // The replica.
const Ice::ObjectPrx _replicaProxy; // A proxy to the individual replica.
const int _id; // My node id.
const std::map<int, NodePrx> _nodes; // The nodes indexed by their id.
const std::map<int, NodePrx> _nodesOneway; // The nodes indexed by their id (as oneway proxies).
const IceUtil::Time _masterTimeout;
const IceUtil::Time _electionTimeout;
const IceUtil::Time _mergeTimeout;
NodeState _state;
int _updateCounter;
int _coord; // Id of the coordinator.
std::string _group; // My group id.
std::set<GroupNodeInfo> _up; // Set of nodes in my group.
std::set<int> _invitesIssued; // The issued invitations.
std::set<int> _invitesAccepted; // The accepted invitations.
unsigned int _max; // The highest group count I've seen.
Ice::Long _generation; // The current generation (or -1 if not set).
Ice::ObjectPrx _coordinatorProxy;
bool _destroy;
// Various timers.
IceUtil::TimerTaskPtr _mergeTask;
IceUtil::TimerTaskPtr _timeoutTask;
IceUtil::TimerTaskPtr _checkTask;
IceUtil::TimerTaskPtr _mergeContinueTask;
};
typedef IceUtil::Handle<NodeI> NodeIPtr;
class FinishUpdateHelper
{
public:
FinishUpdateHelper(const NodeIPtr& node) :
_node(node)
{
}
~FinishUpdateHelper()
{
if(_node)
{
_node->finishUpdate();
}
}
private:
const NodeIPtr _node;
};
class CachedReadHelper
{
public:
CachedReadHelper(const NodeIPtr& node, const char* file, int line) :
_node(node)
{
if(_node)
{
_master = _node->startCachedRead(_generation, file, line);
}
}
~CachedReadHelper()
{
if(_node)
{
_node->finishUpdate();
}
}
Ice::ObjectPrx
getMaster() const
{
return _master;
}
Ice::Long
generation() const
{
return _generation;
}
bool
observerPrecondition(Ice::Long generation) const
{
return generation == _generation && _master;
}
private:
const NodeIPtr _node;
Ice::ObjectPrx _master;
Ice::Long _generation;
};
class ObserverUpdateHelper
{
public:
ObserverUpdateHelper(const NodeIPtr& node, Ice::Long generation, const char* file, int line) :
_node(node)
{
if(_node)
{
_node->startObserverUpdate(generation, file, line);
}
}
~ObserverUpdateHelper()
{
if(_node)
{
_node->finishUpdate();
}
}
private:
const NodeIPtr _node;
};
}
#endif // ELECTION_I_H
|
