ClickHouse/libs/libzkutil/include/zkutil/LeaderElection.h

129 lines
2.9 KiB
C++
Raw Normal View History

2014-04-04 10:37:33 +00:00
#pragma once
#include <zkutil/ZooKeeper.h>
#include <functional>
#include <memory>
2015-09-29 19:19:54 +00:00
#include <common/logger_useful.h>
2014-04-04 10:37:33 +00:00
namespace zkutil
{
/** Реализует метод выбора лидера, описанный здесь: http://zookeeper.apache.org/doc/r3.4.5/recipes.html#sc_leaderElection
*/
class LeaderElection
{
public:
using LeadershipHandler = std::function<void()>;
2014-04-04 10:37:33 +00:00
/** handler вызывается, когда этот экземпляр становится лидером.
*/
LeaderElection(const std::string & path_, ZooKeeper & zookeeper_, LeadershipHandler handler_, const std::string & identifier_ = "")
: path(path_), zookeeper(zookeeper_), handler(handler_), identifier(identifier_),
2016-03-29 15:31:24 +00:00
log(&Logger::get("LeaderElection"))
2014-04-04 10:37:33 +00:00
{
node = EphemeralNodeHolder::createSequential(path + "/leader_election-", zookeeper, identifier);
std::string node_path = node->getPath();
node_name = node_path.substr(node_path.find_last_of('/') + 1);
thread = std::thread(&LeaderElection::threadFunction, this);
}
enum State
{
WAITING_LEADERSHIP,
LEADER,
LEADERSHIP_LOST
};
/// если возвращает LEADER, то еще sessionTimeoutMs мы будем лидером, даже если порвется соединение с zookeeper
2014-04-17 13:59:40 +00:00
State getState()
{
if (state == LEADER)
{
try
{
/// возможно, если сессия разорвалась и заново был вызван init
if (!zookeeper.exists(node->getPath()))
{
LOG_WARNING(log, "Leadership lost. Node " << node->getPath() << " doesn't exist.");
state = LEADERSHIP_LOST;
}
}
catch (const KeeperException & e)
{
LOG_WARNING(log, "Leadership lost. e.message() = " << e.message());
state = LEADERSHIP_LOST;
}
}
return state;
}
2014-04-04 10:37:33 +00:00
~LeaderElection()
{
shutdown = true;
event->set();
2014-04-04 10:37:33 +00:00
thread.join();
}
private:
std::string path;
ZooKeeper & zookeeper;
LeadershipHandler handler;
std::string identifier;
EphemeralNodeHolderPtr node;
std::string node_name;
std::thread thread;
2016-03-29 15:31:24 +00:00
volatile bool shutdown = false;
zkutil::EventPtr event = std::make_shared<Poco::Event>();
2014-04-04 10:37:33 +00:00
2016-03-29 15:31:24 +00:00
State state = WAITING_LEADERSHIP;
2014-04-04 10:37:33 +00:00
Logger * log;
void threadFunction()
{
while (!shutdown)
2014-04-04 10:37:33 +00:00
{
2014-07-01 15:58:25 +00:00
bool success = false;
try
2014-04-04 10:37:33 +00:00
{
Strings children = zookeeper.getChildren(path);
std::sort(children.begin(), children.end());
auto it = std::lower_bound(children.begin(), children.end(), node_name);
if (it == children.end() || *it != node_name)
throw Poco::Exception("Assertion failed in LeaderElection");
if (it == children.begin())
{
state = LEADER;
2014-04-04 10:37:33 +00:00
handler();
return;
}
if (zookeeper.exists(path + "/" + *(it - 1), nullptr, event))
2014-07-14 11:45:34 +00:00
event->wait();
2014-07-01 15:58:25 +00:00
success = true;
}
catch (...)
{
2016-03-29 15:31:24 +00:00
DB::tryLogCurrentException("LeaderElection");
2014-04-04 10:37:33 +00:00
}
2014-07-01 15:58:25 +00:00
if (!success)
2014-07-24 16:29:37 +00:00
event->tryWait(10 * 1000);
2014-04-04 10:37:33 +00:00
}
}
};
using LeaderElectionPtr = std::shared_ptr<LeaderElection>;
2014-04-04 10:37:33 +00:00
}