Hawkin
/
arangodb
mirror of https://gitee.com/bigwinds/arangodb
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

_preparing is enough for both persistent restart and leadership change in agency

This commit is contained in:
Kaveh Vahedipour 2017-06-09 14:03:47 +02:00
parent 1503ca2510
commit 72721f5b2b
2 changed files with 15 additions and 44 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

58
arangod/Agency/Agent.cpp
View File

@ -56,8 +56,7 @@ Agent::Agent(config_t const& config)
_activator(nullptr), _activator(nullptr),
_compactor(this), _compactor(this),
_ready(false), _ready(false),
_preparing(false), _preparing(false) {
_startup(true) {
_state.configure(this); _state.configure(this);
_constituent.configure(this); _constituent.configure(this);
} }
@ -490,7 +489,7 @@ void Agent::sendAppendEntriesRPC() {
// Body // Body
Builder builder; Builder builder;
builder.add(VPackValue(VPackValueType::Array)); builder.add(VPackValue(VPackValueType::Array));
if (!_preparing && if (
((system_clock::now() - _earliestPackage[followerId]).count() > 0)) { ((system_clock::now() - _earliestPackage[followerId]).count() > 0)) {
if (needSnapshot) { if (needSnapshot) {
{ VPackObjectBuilder guard(&builder); { VPackObjectBuilder guard(&builder);
@ -704,8 +703,7 @@ void Agent::load() {
if (size() > 1) { if (size() > 1) {
_inception->start(); _inception->start();
} else { } else {
rebuildDBs(); _spearhead = _readDB;
_startup = false;
activateAgency(); activateAgency();
} }
} }
@ -763,19 +761,10 @@ trans_ret_t Agent::transact(query_t const& queries) {
{ {
CONDITION_LOCKER(guard, _waitForCV); CONDITION_LOCKER(guard, _waitForCV);
while (_startup) {
_waitForCV.wait(100);
MUTEX_LOCKER(ioLocker, _ioLock);
_startup = (_commitIndex != _state.lastIndex());
if (!_startup) {
_spearhead = _readDB;
}
}
}
while (_preparing) { while (_preparing) {
_waitForCV.wait(100); _waitForCV.wait(100);
} }
}
// Apply to spearhead and get indices for log entries // Apply to spearhead and get indices for log entries
auto qs = queries->slice(); auto qs = queries->slice();
@ -833,19 +822,10 @@ trans_ret_t Agent::transient(query_t const& queries) {
{ {
CONDITION_LOCKER(guard, _waitForCV); CONDITION_LOCKER(guard, _waitForCV);
while (_startup) {
_waitForCV.wait(100);
MUTEX_LOCKER(ioLocker, _ioLock);
_startup = (_commitIndex != _state.lastIndex());
if (!_startup) {
_spearhead = _readDB;
}
}
}
while (_preparing) { while (_preparing) {
_waitForCV.wait(100); _waitForCV.wait(100);
} }
}
// Apply to spearhead and get indices for log entries // Apply to spearhead and get indices for log entries
{ {
@ -936,14 +916,6 @@ write_ret_t Agent::write(query_t const& query, bool discardStartup) {
if (!discardStartup) { if (!discardStartup) {
CONDITION_LOCKER(guard, _waitForCV); CONDITION_LOCKER(guard, _waitForCV);
while (_startup) {
_waitForCV.wait(100);
MUTEX_LOCKER(ioLocker, _ioLock);
_startup = (_commitIndex != _state.lastIndex());
if (!_startup) {
_spearhead = _readDB;
}
}
while (_preparing) { while (_preparing) {
_waitForCV.wait(100); _waitForCV.wait(100);
} }
@ -1010,19 +982,10 @@ read_ret_t Agent::read(query_t const& query) {
{ {
CONDITION_LOCKER(guard, _waitForCV); CONDITION_LOCKER(guard, _waitForCV);
while (_startup) {
_waitForCV.wait(100);
MUTEX_LOCKER(ioLocker, _ioLock);
_startup = (_commitIndex != _state.lastIndex());
if (!_startup) {
_spearhead = _readDB;
}
}
}
while (_preparing) { while (_preparing) {
_waitForCV.wait(100); _waitForCV.wait(100);
} }
}
MUTEX_LOCKER(ioLocker, _ioLock); MUTEX_LOCKER(ioLocker, _ioLock);
// Only leader else redirect // Only leader else redirect
@ -1261,6 +1224,15 @@ void Agent::lead() {
// Notify inactive pool // Notify inactive pool
notifyInactive(); notifyInactive();
{
CONDITION_LOCKER(guard, _waitForCV);
while(_commitIndex != _state.lastIndex()) {
_waitForCV.wait(10000);
}
}
_spearhead = _readDB;
} }
// When did we take on leader ship? // When did we take on leader ship?

1
arangod/Agency/Agent.h
View File

@ -351,7 +351,6 @@ class Agent : public arangodb::Thread,
/// @brief Agent is ready for RAFT /// @brief Agent is ready for RAFT
std::atomic<bool> _ready; std::atomic<bool> _ready;
std::atomic<bool> _preparing; std::atomic<bool> _preparing;
std::atomic<bool> _startup;
/// @brief Keep track of when I last took on leadership /// @brief Keep track of when I last took on leadership
TimePoint _leaderSince; TimePoint _leaderSince;