You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@mesos.apache.org by be...@apache.org on 2011/06/05 07:28:11 UTC
svn commit: r1131774 - in /incubator/mesos/trunk/src: master_detector.cpp
master_detector.hpp third_party/libprocess/process.cpp zookeeper.cpp
zookeeper.hpp
Author: benh
Date: Sun Jun 5 05:28:11 2011
New Revision: 1131774
URL: http://svn.apache.org/viewvc?rev=1131774&view=rev
Log:
Updates to fix disconnect bug with ZooKeeper.
Modified:
incubator/mesos/trunk/src/master_detector.cpp
incubator/mesos/trunk/src/master_detector.hpp
incubator/mesos/trunk/src/third_party/libprocess/process.cpp
incubator/mesos/trunk/src/zookeeper.cpp
incubator/mesos/trunk/src/zookeeper.hpp
Modified: incubator/mesos/trunk/src/master_detector.cpp
URL: http://svn.apache.org/viewvc/incubator/mesos/trunk/src/master_detector.cpp?rev=1131774&r1=1131773&r2=1131774&view=diff
==============================================================================
--- incubator/mesos/trunk/src/master_detector.cpp (original)
+++ incubator/mesos/trunk/src/master_detector.cpp Sun Jun 5 05:28:11 2011
@@ -23,7 +23,8 @@ using boost::lexical_cast;
MasterDetector::MasterDetector(const string &_servers, const string &_znode,
const PID &_pid, bool _contend)
- : servers(_servers), znode(_znode), pid(_pid), contend(_contend)
+ : servers(_servers), znode(_znode), pid(_pid),
+ contend(_contend), reconnect(false)
{
zk = new ZooKeeper(servers, 10000, this);
}
@@ -45,46 +46,84 @@ void MasterDetector::process(ZooKeeper *
static const string delimiter = "/";
if ((state == ZOO_CONNECTED_STATE) && (type == ZOO_SESSION_EVENT)) {
- // Create directory path znodes as necessary.
- size_t index = znode.find(delimiter, 0);
- while (index < string::npos) {
- index = znode.find(delimiter, index+1);
- string prefix = znode.substr(0, index);
- ret = zk->create(prefix, "", ZOO_OPEN_ACL_UNSAFE, // ZOO_CREATOR_ALL_ACL, // needs authentication
- 0, &result);
- if (ret != ZOK && ret != ZNODEEXISTS)
- fatal("failed to create ZooKeeper znode! (%s)", zk->error(ret));
- }
-
- // Wierdness in ZooKeeper timing, let's check that everything is created.
- ret = zk->get(znode, false, &result, NULL);
+ // Check if this is a reconnect.
+ if (!reconnect) {
+ // Assume the znode that was created does not end with a "/".
+ CHECK(znode.at(znode.length() - 1) != '/');
+
+ // Create directory path znodes as necessary.
+ size_t index = znode.find(delimiter, 0);
+
+ while (index < string::npos) {
+ // Get out the prefix to create.
+ index = znode.find(delimiter, index + 1);
+ string prefix = znode.substr(0, index);
+
+ // Create the node (even if it already exists).
+ ret = zk->create(prefix, "", ZOO_CREATOR_ALL_ACL,
+ 0, &result);
+
+ if (ret != ZOK && ret != ZNODEEXISTS)
+ fatal("failed to create ZooKeeper znode! (%s)", zk->error(ret));
+ }
- if (ret != ZOK)
- fatal("ZooKeeper not responding correctly (%s). "
- "Make sure ZooKeeper is running on: %s",
- zk->error(ret), servers.c_str());
-
- if (contend) {
- // We use the contend with the pid given in constructor.
- ret = zk->create(znode, pid, ZOO_OPEN_ACL_UNSAFE, // ZOO_CREATOR_ALL_ACL, // needs authentication
- ZOO_SEQUENCE | ZOO_EPHEMERAL, &result);
+ // Wierdness in ZooKeeper timing, let's check that everything is created.
+ ret = zk->get(znode, false, &result, NULL);
if (ret != ZOK)
fatal("ZooKeeper not responding correctly (%s). "
"Make sure ZooKeeper is running on: %s",
zk->error(ret), servers.c_str());
- setMySeq(result);
- LOG(INFO) << "Created ephemeral/sequence:" << getMySeq();
+ if (contend) {
+ // We use the contend with the pid given in constructor.
+ ret = zk->create(znode + "/", pid, ZOO_CREATOR_ALL_ACL,
+ ZOO_SEQUENCE | ZOO_EPHEMERAL, &result);
+
+ if (ret != ZOK)
+ fatal("ZooKeeper not responding correctly (%s). "
+ "Make sure ZooKeeper is running on: %s",
+ zk->error(ret), servers.c_str());
+
+ setMySeq(result);
+ LOG(INFO) << "Created ephemeral/sequence:" << getMySeq();
+
+ const string &s =
+ Tuple<Process>::tupleToString(Tuple<Process>::pack<GOT_MASTER_SEQ>(getMySeq()));
+ Process::post(pid, GOT_MASTER_SEQ, s.data(), s.size());
+ }
- const string &s =
- Tuple<Process>::tupleToString(Tuple<Process>::pack<GOT_MASTER_SEQ>(getMySeq()));
- Process::post(pid, GOT_MASTER_SEQ, s.data(), s.size());
- }
+ // Now determine who the master is (it may be us).
+ detectMaster();
+ } else {
+ // Reconnected. Make sure our ephemeral sequence znode is still there.
+ ret = zk->get(znode + "/" + mySeq, false, &result, NULL);
- detectMaster();
+ // We might no longer be the master! Commit suicide for now
+ // (hoping another master is on standbye), but in the future
+ // it would be nice if we could go back on standbye.
+ if (ret == ZNONODE)
+ fatal("failed to reconnect to ZooKeeper quickly enough "
+ "(our ephemeral sequence znode is gone), commiting suicide!");
+
+ if (ret != ZOK)
+ fatal("ZooKeeper not responding correctly (%s). "
+ "Make sure ZooKeeper is running on: %s",
+ zk->error(ret), servers.c_str());
+
+ // We are still the master!
+ reconnect = false;
+ }
} else if ((state == ZOO_CONNECTED_STATE) && (type == ZOO_CHILD_EVENT)) {
+ // A new master might have showed up and created a sequence
+ // identifier or a master may have died, determine who the master is now!
detectMaster();
+ } else if ((state == ZOO_CONNECTING_STATE) && (type == ZOO_SESSION_EVENT)) {
+ // The client library automatically reconnects, taking into
+ // account failed servers in the connection string,
+ // appropriately handling the "herd effect", etc.
+ LOG(INFO) << "Lost Zookeeper connection. Retrying (automagically).";
+ reconnect = true;
} else {
LOG(INFO) << "Unimplemented watch event: (state is "
<< state << " and type is " << type << ")";
@@ -102,7 +141,7 @@ void MasterDetector::detectMaster()
LOG(ERROR) << "failed to get masters: " << zk->error(ret);
else
LOG(INFO) << "found " << results.size() << " registered masters";
-
+
string masterSeq;
long min = LONG_MAX;
foreach (const string &result, results) {
Modified: incubator/mesos/trunk/src/master_detector.hpp
URL: http://svn.apache.org/viewvc/incubator/mesos/trunk/src/master_detector.hpp?rev=1131774&r1=1131773&r2=1131774&view=diff
==============================================================================
--- incubator/mesos/trunk/src/master_detector.hpp (original)
+++ incubator/mesos/trunk/src/master_detector.hpp Sun Jun 5 05:28:11 2011
@@ -69,10 +69,10 @@ private:
void setMySeq(const string &s)
{
string seq = s;
- // Converts "/nxmaster/000000131" to "000000131".
+ // Converts "/path/to/znode/000000131" to "000000131".
int pos;
if ((pos = seq.find_last_of('/')) != string::npos) {
- mySeq = seq.erase(0, pos+1);
+ mySeq = seq.erase(0, pos + 1);
} else
mySeq = "";
}
@@ -91,6 +91,7 @@ private:
string znode;
PID pid;
bool contend;
+ bool reconnect;
ZooKeeper *zk;
Modified: incubator/mesos/trunk/src/third_party/libprocess/process.cpp
URL: http://svn.apache.org/viewvc/incubator/mesos/trunk/src/third_party/libprocess/process.cpp?rev=1131774&r1=1131773&r2=1131774&view=diff
==============================================================================
--- incubator/mesos/trunk/src/third_party/libprocess/process.cpp (original)
+++ incubator/mesos/trunk/src/third_party/libprocess/process.cpp Sun Jun 5 05:28:11 2011
@@ -3140,12 +3140,19 @@ bool Process::await(int fd, int op, cons
bool Process::await(int fd, int op, const timeval& tv, bool ignore)
{
double secs = tv.tv_sec + (tv.tv_usec * 1e-6);
+
+ if (secs <= 0)
+ return true;
+
return ProcessManager::instance()->await(this, fd, op, secs, ignore);
}
bool Process::ready(int fd, int op)
{
+ if (fd < 0)
+ return false;
+
fd_set rdset;
fd_set wrset;
Modified: incubator/mesos/trunk/src/zookeeper.cpp
URL: http://svn.apache.org/viewvc/incubator/mesos/trunk/src/zookeeper.cpp?rev=1131774&r1=1131773&r2=1131774&view=diff
==============================================================================
--- incubator/mesos/trunk/src/zookeeper.cpp (original)
+++ incubator/mesos/trunk/src/zookeeper.cpp Sun Jun 5 05:28:11 2011
@@ -390,7 +390,7 @@ private:
setCall->zooKeeperProcess->send(setCall->from, COMPLETED);
}
- void prepare(int *fd, int *ops, timeval *tv)
+ bool prepare(int *fd, int *ops, timeval *tv)
{
int interest = 0;
@@ -400,7 +400,7 @@ private:
if (ret == ZINVALIDSTATE ||
ret == ZCONNECTIONLOSS ||
ret == ZOPERATIONTIMEOUT)
- return;
+ return false;
if (ret != ZOK)
fatal("zookeeper_interest failed! (%s)", zerror(ret));
@@ -414,6 +414,8 @@ private:
} else if (interest & ZOOKEEPER_WRITE) {
*ops |= WRONLY;
}
+
+ return true;
}
void process(int fd, int ops)
@@ -456,12 +458,14 @@ protected:
int ops;
timeval tv;
- prepare(&fd, &ops, &tv);
-
- // TODO(benh): If tv is 0, invoke await and ignore queued messages?
+ if (!prepare(&fd, &ops, &tv)) {
+ // Prepare failed, force immediate process.
+ tv.tv_sec = 0;
+ tv.tv_usec = 0;
+ }
if (await(fd, ops, tv, false)) {
- // No enqueued messages and either data available on fd or timer expired.
+ // Either timer expired (might be 0) or data became available on fd.
process(fd, ops);
} else {
// TODO(benh): Don't handle incoming "calls" until we are connected!
Modified: incubator/mesos/trunk/src/zookeeper.hpp
URL: http://svn.apache.org/viewvc/incubator/mesos/trunk/src/zookeeper.hpp?rev=1131774&r1=1131773&r2=1131774&view=diff
==============================================================================
--- incubator/mesos/trunk/src/zookeeper.hpp (original)
+++ incubator/mesos/trunk/src/zookeeper.hpp Sun Jun 5 05:28:11 2011
@@ -95,15 +95,6 @@ public:
*/
class ZooKeeper
{
-private:
- /* ZooKeeper instances are not copyable. */
- ZooKeeper(const ZooKeeper &that);
- ZooKeeper & operator = (const ZooKeeper &that);
-
-protected:
- /* Underlying implementation (pimpl idiom). */
- ZooKeeperImpl *impl;
-
public:
/**
* \brief instantiate new ZooKeeper client.
@@ -276,6 +267,15 @@ public:
* \return string corresponding to the last error
*/
const char * error(int ret) const;
+
+protected:
+ /* Underlying implementation (pimpl idiom). */
+ ZooKeeperImpl *impl;
+
+private:
+ /* ZooKeeper instances are not copyable. */
+ ZooKeeper(const ZooKeeper &that);
+ ZooKeeper & operator = (const ZooKeeper &that);
};