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:24:07 UTC
svn commit: r1131739 - in /incubator/mesos/trunk/src: Makefile.in
config.hpp.in ft_messaging.cpp ft_messaging.hpp master.cpp master.hpp
messages.hpp nexus_sched.cpp slave.cpp slave.hpp
third_party/libprocess/process.cpp third_party/libprocess/process.hpp
Author: benh
Date: Sun Jun 5 05:24:06 2011
New Revision: 1131739
URL: http://svn.apache.org/viewvc?rev=1131739&view=rev
Log:
Updates to ft branch to get it ready for deploy.
Modified:
incubator/mesos/trunk/src/Makefile.in
incubator/mesos/trunk/src/config.hpp.in
incubator/mesos/trunk/src/ft_messaging.cpp
incubator/mesos/trunk/src/ft_messaging.hpp
incubator/mesos/trunk/src/master.cpp
incubator/mesos/trunk/src/master.hpp
incubator/mesos/trunk/src/messages.hpp
incubator/mesos/trunk/src/nexus_sched.cpp
incubator/mesos/trunk/src/slave.cpp
incubator/mesos/trunk/src/slave.hpp
incubator/mesos/trunk/src/third_party/libprocess/process.cpp
incubator/mesos/trunk/src/third_party/libprocess/process.hpp
Modified: incubator/mesos/trunk/src/Makefile.in
URL: http://svn.apache.org/viewvc/incubator/mesos/trunk/src/Makefile.in?rev=1131739&r1=1131738&r2=1131739&view=diff
==============================================================================
--- incubator/mesos/trunk/src/Makefile.in (original)
+++ incubator/mesos/trunk/src/Makefile.in Sun Jun 5 05:24:06 2011
@@ -53,7 +53,7 @@ LDFLAGS += -L$(LIBEV)/.libs
# Add glog and gtest and zookeeper to include and lib paths
CXXFLAGS += -I$(GLOG)/src
CXXFLAGS += -I$(GTEST)/include
-CXXFLAGS += -I$(ZOOKEEPER)/include -I$(ZOOKEEPER)/generated -DTHREADED
+CXXFLAGS += -I$(ZOOKEEPER)/include -I$(ZOOKEEPER)/generated
LDFLAGS += -L$(GLOG)/.libs
LDFLAGS += -L$(GTEST)/lib/.libs
LDFLAGS += -L$(ZOOKEEPER)/.libs
@@ -71,7 +71,7 @@ CFLAGS += -DBUILD_USER="\"$$USER\""
CXXFLAGS += -DBUILD_USER="\"$$USER\""
# Add libev, libprocess, pthread, and dl to LIBS.
-LIBS += -lglog -lgtest -lprocess -lev -lpthread -ldl -lzookeeper_mt
+LIBS += -lglog -lgtest -lprocess -lev -lpthread -ldl -lzookeeper_st
NEXUS_EXES = nexus-master nexus-slave nexus-local nexus-launcher alltests \
test-framework test-executor cpp-test-framework cpp-test-executor \
@@ -92,7 +92,7 @@ NEXUS_LIBS = $(SCHED_LIB) $(EXEC_LIB) $(
MASTER_OBJ = master.o allocator_factory.o simple_allocator.o
SLAVE_OBJ = slave.o launcher.o isolation_module_factory.o \
process_based_isolation_module.o
-COMMON_OBJ = fatal.o hash_pid.o messages.o lock.o leader_detector.o url_processor.o ft_messaging.o
+COMMON_OBJ = fatal.o hash_pid.o messages.o lock.o master_detector.o url_processor.o ft_messaging.o zookeeper.o
EXEC_LIB_OBJ = nexus_exec.o
SCHED_LIB_OBJ = nexus_sched.o nexus_local.o params.o
TEST_OBJ = tests/main.o tests/test_master.o tests/test_resources.o
@@ -133,7 +133,7 @@ $(NEXUS_LIBS): $(COMMON_OBJ) third_party
$(NEXUS_EXES): $(COMMON_OBJ) third_party/libprocess/libprocess.a
$(SCHED_LIB): $(SCHED_LIB_OBJ) $(MASTER_OBJ) $(SLAVE_OBJ) $(COMMON_OBJ)
- $(AR) rcs $@ $^ third_party/libprocess/libprocess.a $(GLOG)/.libs/libglog.a $(LIBEV)/.libs/libev.a $(ZOOKEEPER)/.libs/libzookeeper_mt.a
+ $(AR) rcs $@ $^ third_party/libprocess/libprocess.a $(GLOG)/.libs/libglog.a $(LIBEV)/.libs/libev.a $(ZOOKEEPER)/.libs/libzookeeper_st.a
$(EXEC_LIB): $(EXEC_LIB_OBJ) $(COMMON_OBJ)
$(AR) rcs $@ $^
Modified: incubator/mesos/trunk/src/config.hpp.in
URL: http://svn.apache.org/viewvc/incubator/mesos/trunk/src/config.hpp.in?rev=1131739&r1=1131738&r2=1131739&view=diff
==============================================================================
--- incubator/mesos/trunk/src/config.hpp.in (original)
+++ incubator/mesos/trunk/src/config.hpp.in Sun Jun 5 05:24:06 2011
@@ -1,7 +1,12 @@
#ifndef __CONFIG_HPP__
#define __CONFIG_HPP__
-/* This header is generated by configure */
+/* This header gets completed by configure. */
+
+#include <string>
+
+/* Constants. */
+const std::string ZNODE = "/nxmaster";
#if @WITH_WEBUI@
#define NEXUS_WEBUI
Modified: incubator/mesos/trunk/src/ft_messaging.cpp
URL: http://svn.apache.org/viewvc/incubator/mesos/trunk/src/ft_messaging.cpp?rev=1131739&r1=1131738&r2=1131739&view=diff
==============================================================================
--- incubator/mesos/trunk/src/ft_messaging.cpp (original)
+++ incubator/mesos/trunk/src/ft_messaging.cpp Sun Jun 5 05:24:06 2011
@@ -6,7 +6,6 @@
#include <zookeeper.h>
#include <glog/logging.h>
#include "messages.hpp"
-#include "leader_detector.hpp"
#include "ft_messaging.hpp"
namespace nexus { namespace internal {
Modified: incubator/mesos/trunk/src/ft_messaging.hpp
URL: http://svn.apache.org/viewvc/incubator/mesos/trunk/src/ft_messaging.hpp?rev=1131739&r1=1131738&r2=1131739&view=diff
==============================================================================
--- incubator/mesos/trunk/src/ft_messaging.hpp (original)
+++ incubator/mesos/trunk/src/ft_messaging.hpp Sun Jun 5 05:24:06 2011
@@ -1,8 +1,6 @@
#ifndef _FT_MESSAGING_HPP_
#define _FT_MESSAGING_HPP_
-#define FT_MAX_RESENDS 30 // 5 minutes
-
#include <boost/lexical_cast.hpp>
#include <boost/unordered_set.hpp>
#include <boost/unordered_map.hpp>
@@ -10,11 +8,12 @@
#include <ctime>
#include <process.hpp>
-#include "leader_detector.hpp"
#include "messages.hpp"
#include "tuple.hpp"
#include "foreach.hpp"
+#define FT_TIMEOUT 10
+#define FT_MAX_RESENDS 30 // 5 minutes
namespace nexus { namespace internal {
Modified: incubator/mesos/trunk/src/master.cpp
URL: http://svn.apache.org/viewvc/incubator/mesos/trunk/src/master.cpp?rev=1131739&r1=1131738&r2=1131739&view=diff
==============================================================================
--- incubator/mesos/trunk/src/master.cpp (original)
+++ incubator/mesos/trunk/src/master.cpp Sun Jun 5 05:24:06 2011
@@ -1,15 +1,10 @@
-#include "config.hpp" // Need to define first to get USING_ZOOKEEPER
-
-#include <glog/logging.h>
-
-#ifdef USING_ZOOKEEPER
-#include <zookeeper.hpp>
-#endif
-
#include "allocator.hpp"
-#include "allocator_factory.hpp"
#include "master.hpp"
#include "master_webui.hpp"
+#include "allocator_factory.hpp"
+#include "ft_messaging.hpp"
+
+#include <glog/logging.h>
using std::endl;
using std::max;
@@ -31,88 +26,8 @@ using namespace nexus::internal;
using namespace nexus::internal::master;
-/* List of ZooKeeper host:port pairs (from master_main.cpp/local.cpp). */
-extern string zookeeper;
-
namespace {
-#ifdef USING_ZOOKEEPER
-class MasterWatcher : public Watcher
-{
-private:
- Master *master;
-
-public:
- void process(ZooKeeper *zk, int type, int state, const string &path)
- {
- string znode = "/home/nexus/master";
- string dirname = "/home/nexus";
- string delimiter = "/";
- string contents = "";
-
- int ret;
- string result;
-
- if ((state == ZOO_CONNECTED_STATE) &&
- (type == ZOO_SESSION_EVENT)) {
- // Create directory path znodes as necessary.
- size_t index = dirname.find(delimiter, 0);
- while (index < string::npos) {
- index = dirname.find(delimiter, index+1);
- string prefix = dirname.substr(0, index);
- ret = zk->create(prefix, contents, ZOO_CREATOR_ALL_ACL,
- 0, &result);
- if (ret != ZOK && ret != ZNODEEXISTS)
- fatal("failed to create ZooKeeper znode! (%s)", zk->error(ret));
- }
-
- // Now create znode.
- ret = zk->create(znode, master->getPID(), ZOO_CREATOR_ALL_ACL,
- ZOO_EPHEMERAL, &result);
-
- // If the node already exists, wait for it to get deleted.
- if (ret == ZNODEEXISTS)
- ret = zk->exists(znode, true, NULL);
-
- if (ret != ZOK)
- fatal("failed to create ZooKeeper znode! (%s)", zk->error(ret));
- } else if ((state == ZOO_CONNECTED_STATE) &&
- (type == ZOO_DELETED_EVENT) &&
- (path.compare(znode) == 0)) {
- // Now (re)create znode.
- ret = zk->create(znode, master->getPID(), ZOO_CREATOR_ALL_ACL,
- ZOO_EPHEMERAL, &result);
-
- if (ret != ZOK)
- fatal("failed to create ZooKeeper znode! (%s)", zk->error(ret));
-
- // And set the value to our pid.
- ret = zk->set(znode, master->getPID(), -1);
-
- if (ret != ZOK)
- fatal("failed to create ZooKeeper znode! (%s)", zk->error(ret));
- } else if ((state == ZOO_EXPIRED_SESSION_STATE) &&
- (type == ZOO_SESSION_EVENT)) {
- // TODO(benh): Reconnect if session expires. Note the Zookeeper
- // C library retries in the case of connection timeouts,
- // connection loss, etc. Only expired sessions require
- // explicitly reconnecting.
- fatal("connection to ZooKeeper expired!");
- } 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).";
- } else {
- fatal("unhandled ZooKeeper event!");
- }
- }
-
- MasterWatcher(Master *_master) : master(_master) {}
-};
-#endif
-
// A process that periodically pings the master to check filter expiries, etc
class AllocatorTimer : public Tuple<Process>
{
@@ -195,24 +110,50 @@ public:
}
-Master::Master()
- : nextFrameworkId(0), nextSlaveId(0), nextSlotOfferId(0),
- allocatorType("simple")
-{}
+Master::Master(const string &zk)
+ : masterDetector(NULL), nextFrameworkId(0), nextSlaveId(0),
+ nextSlotOfferId(0), allocatorType("simple"), masterId(0)
+{
+ if (zk != "") {
+ pair<UrlProcessor::URLType, string> urlPair = UrlProcessor::process(zk);
+ if (urlPair.first == UrlProcessor::ZOO) {
+ isFT = true;
+ zkServers = urlPair.second;
+ } else {
+ LOG(ERROR) << "Failed to parse URL for ZooKeeper servers. URL must start with zoo:// or zoofile://";
+ exit(1);
+ }
+ }
+ ftMsg = FTMessaging::getInstance();
+}
-Master::Master(const string& _allocatorType)
- : nextFrameworkId(0), nextSlaveId(0), nextSlotOfferId(0),
- allocatorType(_allocatorType)
-{}
+Master::Master(const string& _allocatorType, const string &zk)
+ : masterDetector(NULL), nextFrameworkId(0), nextSlaveId(0),
+ nextSlotOfferId(0), allocatorType(_allocatorType), masterId(0)
+{
+ if (zk != "") {
+ pair<UrlProcessor::URLType, string> urlPair = UrlProcessor::process(zk);
+ if (urlPair.first == UrlProcessor::ZOO) {
+ isFT = true;
+ zkServers = urlPair.second;
+ } else {
+ LOG(ERROR) << "Failed to parse URL for ZooKeeper servers. URL must start with zoo:// or zoofile://";
+ exit(1);
+ }
+ }
+ ftMsg = FTMessaging::getInstance();
+}
Master::~Master()
{
+ if (isFT && masterDetector != NULL)
+ delete masterDetector;
LOG(INFO) << "Shutting down master";
delete allocator;
foreachpair (_, Framework *framework, frameworks) {
- foreachpair(_, Task *task, framework->tasks)
+ foreachpair(_, TaskInfo *task, framework->tasks)
delete task;
delete framework;
}
@@ -230,7 +171,7 @@ state::MasterState * Master::getState()
std::ostringstream oss;
oss << self();
state::MasterState *state =
- new state::MasterState(BUILD_DATE, BUILD_USER, oss.str());
+ new state::MasterState(BUILD_DATE, BUILD_USER, oss.str(), isFT);
foreachpair (_, Slave *s, slaves) {
state::Slave *slave = new state::Slave(s->id, s->hostname, s->publicDns,
@@ -243,7 +184,7 @@ state::MasterState * Master::getState()
f->executorInfo.uri, f->resources.cpus, f->resources.mem,
f->connectTime);
state->frameworks.push_back(framework);
- foreachpair (_, Task *t, f->tasks) {
+ foreachpair (_, TaskInfo *t, f->tasks) {
state::Task *task = new state::Task(t->id, t->name, t->frameworkId,
t->slaveId, t->state, t->resources.cpus, t->resources.mem);
framework->tasks.push_back(task);
@@ -317,11 +258,32 @@ SlotOffer * Master::lookupSlotOffer(Offe
return NULL;
}
+void Master::updateFrameworkTasks() {
+ foreachpair (SlaveID sid, Slave *slave, slaves) {
+ foreachpair (_, TaskInfo *task, slave->tasks) {
+ updateFrameworkTasks(task);
+ }
+ }
+}
+
+void Master::updateFrameworkTasks(TaskInfo *task) {
+ Framework *fwrk = lookupFramework(task->frameworkId);
+ if (fwrk != NULL) {
+ if (fwrk->tasks.find(task->id) == fwrk->tasks.end()) {
+ fwrk->tasks[task->id] = task;
+ fwrk->resources += task->resources;
+ }
+ }
+}
+
void Master::operator () ()
{
- LOG(INFO) << "Master started at " << self();
+ LOG(INFO) << "Master started at nexus://" << self();
+ LOG(INFO) << "Connecting to ZooKeeper at " << zkServers;
+ masterDetector = new MasterDetector(zkServers, ZNODE, self(), true);
+
allocator = createAllocator();
if (!allocator)
LOG(FATAL) << "Unrecognized allocator type: " << allocatorType;
@@ -329,21 +291,25 @@ void Master::operator () ()
link(spawn(new AllocatorTimer(self())));
//link(spawn(new SharesPrinter(self())));
-#ifdef USING_ZOOKEEPER
- ZooKeeper *zk;
- if (!zookeeper.empty())
- zk = new ZooKeeper(zookeeper, 10000, new MasterWatcher(this));
-#endif
-
while (true) {
switch (receive()) {
+ case GOT_MASTER_SEQ: {
+ // TODO(benh|alig): NEED TO GET SEQ BEFORE ANYONE ELSE CONNECTS!
+ string mySeq;
+ unpack<GOT_MASTER_SEQ>(mySeq);
+ masterId = lexical_cast<long>(mySeq);
+ LOG(INFO) << "Master ID:" << masterId;
+ break;
+ }
+
case F2M_REGISTER_FRAMEWORK: {
- FrameworkID fid = nextFrameworkId++;
+ FrameworkID fid = lexical_cast<string>(masterId) + "-" + lexical_cast<string>(nextFrameworkId++);
+
Framework *framework = new Framework(from(), fid);
unpack<F2M_REGISTER_FRAMEWORK>(framework->name,
- framework->user,
- framework->executorInfo);
+ framework->user,
+ framework->executorInfo);
LOG(INFO) << "Registering " << framework << " at " << framework->pid;
frameworks[fid] = framework;
pidToFid[framework->pid] = fid;
@@ -355,13 +321,78 @@ void Master::operator () ()
break;
}
+ case F2M_REREGISTER_FRAMEWORK: {
+
+ Framework *framework = new Framework(from());
+ unpack<F2M_REREGISTER_FRAMEWORK>(framework->id,
+ framework->name,
+ framework->user,
+ framework->executorInfo);
+
+ if (framework->id == "") {
+ DLOG(INFO) << "Framework reconnecting without a FrameworkID, generating new id";
+ framework->id = lexical_cast<string>(masterId) + "-" + lexical_cast<string>(nextFrameworkId++);
+ }
+
+ LOG(INFO) << "Registering " << framework << " at " << framework->pid;
+ frameworks[framework->id] = framework;
+ pidToFid[framework->pid] = framework->id;
+
+ updateFrameworkTasks();
+
+ link(framework->pid);
+ send(framework->pid, pack<M2F_REGISTER_REPLY>(framework->id));
+ allocator->frameworkAdded(framework);
+ if (framework->executorInfo.uri == "")
+ terminateFramework(framework, 1, "No executor URI given");
+
+ timeval tv;
+ gettimeofday(&tv, NULL);
+
+ DLOG(INFO) << tv.tv_sec << "." << tv.tv_usec << " STAT: Slave count: " << slaves.size() << " Framework count: " << frameworks.size();
+
+ break;
+ }
+
case F2M_UNREGISTER_FRAMEWORK: {
FrameworkID fid;
unpack<F2M_UNREGISTER_FRAMEWORK>(fid);
LOG(INFO) << "Asked to unregister framework " << fid;
Framework *framework = lookupFramework(fid);
if (framework != NULL)
- removeFramework(framework);
+ removeFramework(framework);
+ break;
+ }
+
+ case F2M_FT_SLOT_OFFER_REPLY: {
+ FrameworkID fid;
+ OfferID oid;
+ vector<TaskDescription> tasks;
+ Params params;
+ string ftId, senderStr;
+ unpack<F2M_FT_SLOT_OFFER_REPLY>(ftId, senderStr, fid, oid, tasks, params);
+ PID senderPid;
+ istringstream ss(senderStr);
+ ss >> senderPid;
+ if (!ftMsg->acceptMessageAckTo(senderPid, ftId, senderStr)) {
+ LOG(WARNING) << "FT: Locally ignoring duplicate message with id:" << ftId;
+ break;
+ }
+ Framework *framework = lookupFramework(fid);
+ if (framework != NULL) {
+ SlotOffer *offer = lookupSlotOffer(oid);
+ if (offer != NULL) {
+ processOfferReply(offer, tasks, params);
+ } else {
+ // The slot offer is gone, meaning that we rescinded it or that
+ // the slave was lost; immediately report any tasks in it as lost
+ foreach (TaskDescription &t, tasks) {
+ send(framework->pid,
+ pack<M2F_STATUS_UPDATE>(t.taskId, TASK_LOST, ""));
+ }
+ }
+ } else
+ DLOG(INFO) << "F2M_FT_SLOT_OFFER_REPLY error: couldn't lookup framework id" << fid;
break;
}
@@ -373,17 +404,17 @@ void Master::operator () ()
unpack<F2M_SLOT_OFFER_REPLY>(fid, oid, tasks, params);
Framework *framework = lookupFramework(fid);
if (framework != NULL) {
- SlotOffer *offer = lookupSlotOffer(oid);
- if (offer != NULL) {
- processOfferReply(offer, tasks, params);
- } else {
- // The slot offer is gone, meaning that we rescinded it or that
- // the slave was lost; immediately report any tasks in it as lost
- foreach (TaskDescription &t, tasks) {
- send(framework->pid,
- pack<M2F_STATUS_UPDATE>(t.taskId, TASK_LOST, ""));
- }
- }
+ SlotOffer *offer = lookupSlotOffer(oid);
+ if (offer != NULL) {
+ processOfferReply(offer, tasks, params);
+ } else {
+ // The slot offer is gone, meaning that we rescinded it or that
+ // the slave was lost; immediately report any tasks in it as lost
+ foreach (TaskDescription &t, tasks) {
+ send(framework->pid,
+ pack<M2F_STATUS_UPDATE>(t.taskId, TASK_LOST, ""));
+ }
+ }
}
break;
}
@@ -393,9 +424,9 @@ void Master::operator () ()
unpack<F2M_REVIVE_OFFERS>(fid);
Framework *framework = lookupFramework(fid);
if (framework != NULL) {
- LOG(INFO) << "Reviving offers for " << framework;
- framework->slaveFilter.clear();
- allocator->offersRevived(framework);
+ LOG(INFO) << "Reviving offers for " << framework;
+ framework->slaveFilter.clear();
+ allocator->offersRevived(framework);
}
break;
}
@@ -406,34 +437,39 @@ void Master::operator () ()
unpack<F2M_KILL_TASK>(fid, tid);
Framework *framework = lookupFramework(fid);
if (framework != NULL) {
- Task *task = framework->lookupTask(tid);
- if (task != NULL) {
- LOG(INFO) << "Asked to kill " << task << " by its framework";
- killTask(task);
- }
+ TaskInfo *task = framework->lookupTask(tid);
+ if (task != NULL) {
+ LOG(INFO) << "Asked to kill " << task << " by its framework";
+ killTask(task);
+ }
}
break;
}
- case F2M_FRAMEWORK_MESSAGE: {
+ case F2M_FT_FRAMEWORK_MESSAGE: {
FrameworkID fid;
FrameworkMessage message;
- unpack<F2M_FRAMEWORK_MESSAGE>(fid, message);
+ string ftId, senderStr;
+ unpack<F2M_FT_FRAMEWORK_MESSAGE>(ftId, senderStr, fid, message);
Framework *framework = lookupFramework(fid);
if (framework != NULL) {
- Slave *slave = lookupSlave(message.slaveId);
- if (slave != NULL) {
- LOG(INFO) << "Sending framework message to " << slave;
- send(slave->pid, pack<M2S_FRAMEWORK_MESSAGE>(fid, message));
- }
- }
+ Slave *slave = lookupSlave(message.slaveId);
+ if (slave != NULL) {
+ LOG(INFO) << "Sending framework message to " << slave;
+ send(slave->pid, pack<M2S_FT_FRAMEWORK_MESSAGE>(ftId, senderStr, fid, message));
+ } else
+ DLOG(INFO) << "S2M_FT_FRAMEWORK_MESSAGE error: couldn't lookup framework id" << fid;
+ } else
+ DLOG(INFO) << "S2M_FT_FRAMEWORK_MESSAGE error: couldn't lookup slave id" << message.slaveId;
break;
}
case S2M_REGISTER_SLAVE: {
- Slave *slave = new Slave(from(), nextSlaveId++);
+ string slaveId = lexical_cast<string>(masterId) + "-" + lexical_cast<string>(nextSlaveId++);
+
+ Slave *slave = new Slave(from(), slaveId);
unpack<S2M_REGISTER_SLAVE>(slave->hostname, slave->publicDns,
- slave->resources);
+ slave->resources);
LOG(INFO) << "Registering " << slave << " at " << slave->pid;
slaves[slave->id] = slave;
pidToSid[slave->pid] = slave->id;
@@ -443,29 +479,73 @@ void Master::operator () ()
break;
}
+ case S2M_REREGISTER_SLAVE: {
+ Slave *slave = new Slave(from());
+ vector<TaskInfo> taskVec;
+
+ unpack<S2M_REREGISTER_SLAVE>(slave->id, slave->hostname, slave->publicDns,
+ slave->resources, taskVec);
+
+ if (slave->id == "") {
+ slave->id = lexical_cast<string>(masterId) + "-" + lexical_cast<string>(nextSlaveId++);
+ DLOG(WARNING) << "Slave re-registered without a SlaveID, generating a new id for it.";
+ }
+
+ foreach(TaskInfo &ti, taskVec) {
+ TaskInfo *tip = new TaskInfo(ti);
+ slave->addTask(tip);
+ updateFrameworkTasks(tip);
+ }
+
+ //alibandali
+ LOG(INFO) << "Re-registering " << slave << " at " << slave->pid;
+ slaves[slave->id] = slave;
+ pidToSid[slave->pid] = slave->id;
+ link(slave->pid);
+ send(slave->pid, pack<M2S_REREGISTER_REPLY>(slave->id));
+ allocator->slaveAdded(slave);
+
+ timeval tv;
+ gettimeofday(&tv, NULL);
+
+ DLOG(INFO) << tv.tv_sec << "." << tv.tv_usec << " STAT: Slave count: " << slaves.size() << " Framework count: " << frameworks.size();
+
+ break;
+ }
+
case S2M_UNREGISTER_SLAVE: {
SlaveID sid;
unpack<S2M_UNREGISTER_SLAVE>(sid);
LOG(INFO) << "Asked to unregister slave " << sid;
Slave *slave = lookupSlave(sid);
if (slave != NULL)
- removeSlave(slave);
+ removeSlave(slave);
break;
}
- case S2M_STATUS_UPDATE: {
+ case S2M_FT_STATUS_UPDATE: {
SlaveID sid;
FrameworkID fid;
TaskID tid;
TaskState state;
string data;
- unpack<S2M_STATUS_UPDATE>(sid, fid, tid, state, data);
+ string ftId, senderStr;
+
+ unpack<S2M_FT_STATUS_UPDATE>(ftId, senderStr, sid, fid, tid, state, data);
+ DLOG(INFO) << "FT: prepare relay ftId:"<< ftId << " from: "<< senderStr;
if (Slave *slave = lookupSlave(sid)) {
- if (Framework *framework = lookupFramework(fid)) {
- // Pass on the status update to the framework
- send(framework->pid, pack<M2F_STATUS_UPDATE>(tid, state, data));
+ if (Framework *framework = lookupFramework(fid)) {
+ // Pass on the status update to the framework
+
+ DLOG(INFO) << "FT: relaying ftId:"<< ftId << " from: "<< senderStr;
+ send(framework->pid, pack<M2F_FT_STATUS_UPDATE>(ftId, senderStr, tid, state, data));
+
+ if (!ftMsg->acceptMessage(ftId, senderStr)) {
+ LOG(WARNING) << "FT: Locally ignoring duplicate message with id:" << ftId;
+ break;
+ }
// Update the task state locally
- Task *task = slave->lookupTask(fid, tid);
+ TaskInfo *task = slave->lookupTask(fid, tid);
if (task != NULL) {
LOG(INFO) << "Status update: " << task << " is in state " << state;
task->state = state;
@@ -476,11 +556,65 @@ void Master::operator () ()
removeTask(task, TRR_TASK_ENDED);
}
}
- }
- break;
- }
+ } else
+ DLOG(INFO) << "S2M_STATUS_UPDATE error: couldn't lookup framework id" << fid;
+ } else
+ DLOG(INFO) << "S2M_STATUS_UPDATE error: couldn't lookup slave id" << sid;
+
+ break;
+ }
+
+ case S2M_STATUS_UPDATE: {
+ SlaveID sid;
+ FrameworkID fid;
+ TaskID tid;
+ TaskState state;
+ string data;
+ unpack<S2M_STATUS_UPDATE>(sid, fid, tid, state, data);
+ if (Slave *slave = lookupSlave(sid)) {
+ if (Framework *framework = lookupFramework(fid)) {
+ // Pass on the status update to the framework
+ send(framework->pid, pack<M2F_STATUS_UPDATE>(tid, state, data));
+ // Update the task state locally
+ TaskInfo *task = slave->lookupTask(fid, tid);
+ if (task != NULL) {
+ LOG(INFO) << "Status update: " << task << " is in state " << state;
+ task->state = state;
+ // Remove the task if it finished or failed
+ if (state == TASK_FINISHED || state == TASK_FAILED ||
+ state == TASK_KILLED || state == TASK_LOST) {
+ LOG(INFO) << "Removing " << task << " because it's done";
+ removeTask(task, TRR_TASK_ENDED);
+ }
+ }
+ } else
+ DLOG(INFO) << "S2M_STATUS_UPDATE error: couldn't lookup framework id" << fid;
+ } else
+ DLOG(INFO) << "S2M_STATUS_UPDATE error: couldn't lookup slave id" << sid;
+ break;
}
+ case S2M_FT_FRAMEWORK_MESSAGE: {
+ SlaveID sid;
+ FrameworkID fid;
+ FrameworkMessage message;
+ string ftId, senderStr;
+ unpack<S2M_FT_FRAMEWORK_MESSAGE>(ftId, senderStr, sid, fid, message);
+ Slave *slave = lookupSlave(sid);
+ if (slave != NULL) {
+ Framework *framework = lookupFramework(fid);
+ if (framework != NULL) {
+
+ send(framework->pid, pack<M2F_FT_FRAMEWORK_MESSAGE>(ftId, senderStr, message));
+
+ } else
+ DLOG(INFO) << "S2M_FT_FRAMEWORK_MESSAGE error: couldn't lookup framework id" << fid;
+ } else
+ DLOG(INFO) << "S2M_FT_FRAMEWORK_MESSAGE error: couldn't lookup slave id" << sid;
+
+ break;
+ }
+
case S2M_FRAMEWORK_MESSAGE: {
SlaveID sid;
FrameworkID fid;
@@ -488,9 +622,9 @@ void Master::operator () ()
unpack<S2M_FRAMEWORK_MESSAGE>(sid, fid, message);
Slave *slave = lookupSlave(sid);
if (slave != NULL) {
- Framework *framework = lookupFramework(fid);
- if (framework != NULL)
- send(framework->pid, pack<M2F_FRAMEWORK_MESSAGE>(message));
+ Framework *framework = lookupFramework(fid);
+ if (framework != NULL)
+ send(framework->pid, pack<M2F_FRAMEWORK_MESSAGE>(message));
}
break;
}
@@ -514,13 +648,13 @@ void Master::operator () ()
}
// Collect all the lost tasks for this framework.
- set<Task*> tasks;
- foreachpair (_, Task* task, framework->tasks)
+ set<TaskInfo*> tasks;
+ foreachpair (_, TaskInfo* task, framework->tasks)
if (task->slaveId == slave->id)
tasks.insert(task);
// Tell the framework they have been lost and remove them.
- foreach (Task* task, tasks) {
+ foreach (TaskInfo* task, tasks) {
send(framework->pid, pack<M2F_STATUS_UPDATE>(task->id, TASK_LOST,
task->message));
@@ -562,6 +696,30 @@ void Master::operator () ()
// Do allocations!
allocator->timerTick();
+
+ // int cnts = 0;
+ // foreachpair(_, Framework *framework, frameworks) {
+ // DLOG(INFO) << (cnts++) << " resourceInUse:" << framework->resources;
+ // }
+ break;
+ }
+
+ case FT_RELAY_ACK: {
+ string ftId;
+ string origPidStr;
+ unpack<FT_RELAY_ACK>(ftId, origPidStr);
+
+ DLOG(INFO) << "FT_RELAY_ACK for " << ftId << " forwarding it to " << origPidStr;
+
+ PID origPid;
+ istringstream iss(origPidStr);
+ if (!(iss >> origPid)) {
+ cerr << "FT: Failed to resolve PID for originator: " << origPidStr << endl;
+ break;
+ }
+
+ send(origPid, pack<FT_RELAY_ACK>(ftId, origPidStr));
+
break;
}
@@ -607,7 +765,8 @@ void Master::operator () ()
OfferID Master::makeOffer(Framework *framework,
const vector<SlaveResources>& resources)
{
- OfferID oid = nextSlotOfferId++;
+ OfferID oid = lexical_cast<string>(masterId) + "-" + lexical_cast<string>(nextSlotOfferId++);
+
SlotOffer *offer = new SlotOffer(oid, framework->id, resources);
slotOffers[offer->id] = offer;
framework->addOffer(offer);
@@ -621,7 +780,7 @@ OfferID Master::makeOffer(Framework *fra
Params params;
params.set("cpus", r.resources.cpus);
params.set("mem", r.resources.mem);
- SlaveOffer offer(r.slave->id, r.slave->hostname, params.getMap());
+ SlaveOffer offer(r.slave->id, r.slave->hostname, params.getMap(), r.slave->pid);
offers.push_back(offer);
}
send(framework->pid, pack<M2F_SLOT_OFFER>(oid, offers));
@@ -720,13 +879,13 @@ void Master::launchTask(Framework *f, co
Resources res(params.getInt32("cpus", -1),
params.getInt64("mem", -1));
Slave *slave = lookupSlave(t.slaveId);
- Task *task = f->addTask(t.taskId, t.name, slave->id, res);
+ TaskInfo *task = f->addTask(t.taskId, t.name, slave->id, res);
LOG(INFO) << "Launching " << task << " on " << slave;
slave->addTask(task);
allocator->taskAdded(task);
send(slave->pid, pack<M2S_RUN_TASK>(
f->id, t.taskId, f->name, f->user, f->executorInfo,
- t.name, t.arg, t.params));
+ t.name, t.arg, t.params, (string)f->pid));
}
@@ -736,7 +895,7 @@ void Master::rescindOffer(SlotOffer *off
}
-void Master::killTask(Task *task)
+void Master::killTask(TaskInfo *task)
{
LOG(INFO) << "Killing " << task;
Framework *framework = lookupFramework(task->frameworkId);
@@ -805,8 +964,8 @@ void Master::removeFramework(Framework *
send(slave->pid, pack<M2S_KILL_FRAMEWORK>(framework->id));
// Remove pointers to the framework's tasks in slaves
- unordered_map<TaskID, Task *> tasksCopy = framework->tasks;
- foreachpair (_, Task *task, tasksCopy) {
+ unordered_map<TaskID, TaskInfo *> tasksCopy = framework->tasks;
+ foreachpair (_, TaskInfo *task, tasksCopy) {
Slave *slave = lookupSlave(task->slaveId);
CHECK(slave != NULL);
removeTask(task, TRR_FRAMEWORK_LOST);
@@ -835,8 +994,8 @@ void Master::removeSlave(Slave *slave)
// TODO: Notify allocator that a slave removal is beginning?
// Remove pointers to slave's tasks in frameworks, and send status updates
- unordered_map<pair<FrameworkID, TaskID>, Task *> tasksCopy = slave->tasks;
- foreachpair (_, Task *task, tasksCopy) {
+ unordered_map<pair<FrameworkID, TaskID>, TaskInfo *> tasksCopy = slave->tasks;
+ foreachpair (_, TaskInfo *task, tasksCopy) {
Framework *framework = lookupFramework(task->frameworkId);
CHECK(framework != NULL);
send(framework->pid, pack<M2F_STATUS_UPDATE>(task->id, TASK_LOST,
@@ -877,7 +1036,7 @@ void Master::removeSlave(Slave *slave)
// Remove a slot offer (because it was replied or we lost a framework or slave)
-void Master::removeTask(Task *task, TaskRemovalReason reason)
+void Master::removeTask(TaskInfo *task, TaskRemovalReason reason)
{
Framework *framework = lookupFramework(task->frameworkId);
Slave *slave = lookupSlave(task->slaveId);
Modified: incubator/mesos/trunk/src/master.hpp
URL: http://svn.apache.org/viewvc/incubator/mesos/trunk/src/master.hpp?rev=1131739&r1=1131738&r2=1131739&view=diff
==============================================================================
--- incubator/mesos/trunk/src/master.hpp (original)
+++ incubator/mesos/trunk/src/master.hpp Sun Jun 5 05:24:06 2011
@@ -27,6 +27,10 @@
#include "messages.hpp"
#include "params.hpp"
#include "resources.hpp"
+#include "master_detector.hpp"
+#include "task_info.hpp"
+#include "url_processor.hpp"
+#include "ft_messaging.hpp"
namespace nexus { namespace internal { namespace master {
@@ -93,22 +97,6 @@ struct SlotOffer
: id(i), frameworkId(f), resources(r) {}
};
-
-// An active task. The master only keeps track of tasks that are active.
-struct Task
-{
- TaskID id;
- FrameworkID frameworkId; // Which framework we belong to
- SlaveID slaveId; // Which slave we're on
- Resources resources;
- TaskState state;
- string name;
- string message;
-
- Task(TaskID _id, Resources _res) : id(_id), resources(_res) {}
-};
-
-
// An connected framework.
struct Framework
{
@@ -120,7 +108,7 @@ struct Framework
ExecutorInfo executorInfo;
time_t connectTime;
- unordered_map<TaskID, Task *> tasks;
+ unordered_map<TaskID, TaskInfo *> tasks;
unordered_set<SlotOffer *> slotOffers; // Active offers given to this framework
Resources resources; // Total resources owned by framework (tasks + offers)
@@ -129,26 +117,26 @@ struct Framework
// or 0 for slaves that we want to keep filtered forever
unordered_map<Slave *, time_t> slaveFilter;
- Framework(const PID &_pid, FrameworkID _id)
+ Framework(const PID &_pid, FrameworkID _id = "")
: pid(_pid), id(_id), active(true)
{
time(&connectTime);
}
- Task * lookupTask(TaskID tid)
+ TaskInfo * lookupTask(TaskID tid)
{
- unordered_map<TaskID, Task *>::iterator it = tasks.find(tid);
+ unordered_map<TaskID, TaskInfo *>::iterator it = tasks.find(tid);
if (it != tasks.end())
return it->second;
else
return NULL;
}
- Task * addTask(TaskID tid, const std::string& name,
+ TaskInfo * addTask(TaskID tid, const std::string& name,
SlaveID location, Resources resources)
{
CHECK(tasks.find(tid) == tasks.end());
- Task *task = new Task(tid, resources);
+ TaskInfo *task = new TaskInfo(tid, resources);
task->frameworkId = id;
task->state = TASK_STARTING;
task->name = name;
@@ -161,7 +149,7 @@ struct Framework
void removeTask(TaskID tid)
{
CHECK(tasks.find(tid) != tasks.end());
- unordered_map<TaskID, Task *>::iterator it = tasks.find(tid);
+ unordered_map<TaskID, TaskInfo *>::iterator it = tasks.find(tid);
this->resources -= it->second->resources;
tasks.erase(it);
}
@@ -216,30 +204,30 @@ struct Slave
Resources resourcesOffered; // Resources currently in offers
Resources resourcesInUse; // Resources currently used by tasks
- unordered_map<pair<FrameworkID, TaskID>, Task *> tasks;
+ unordered_map<pair<FrameworkID, TaskID>, TaskInfo *> tasks;
unordered_set<SlotOffer *> slotOffers; // Active offers of slots on this slave
- Slave(const PID &_pid, SlaveID _id) : pid(_pid), id(_id), active(true) {
+ Slave(const PID &_pid, SlaveID _id = "") : pid(_pid), id(_id), active(true) {
connectTime = lastHeartbeat = time(NULL);
}
- Task * lookupTask(FrameworkID fid, TaskID tid)
+ TaskInfo * lookupTask(FrameworkID fid, TaskID tid)
{
- foreachpair (_, Task *task, tasks)
+ foreachpair (_, TaskInfo *task, tasks)
if (task->frameworkId == fid && task->id == tid)
return task;
return NULL;
}
- void addTask(Task *task)
+ void addTask(TaskInfo *task)
{
CHECK(tasks.find(make_pair(task->frameworkId, task->id)) == tasks.end());
tasks[make_pair(task->frameworkId, task->id)] = task;
resourcesInUse += task->resources;
}
- void removeTask(Task *task)
+ void removeTask(TaskInfo *task)
{
CHECK(tasks.find(make_pair(task->frameworkId, task->id)) != tasks.end());
tasks.erase(make_pair(task->frameworkId, task->id));
@@ -276,6 +264,9 @@ enum TaskRemovalReason
class Master : public Tuple<Process>
{
protected:
+ bool isFT;
+ string zkServers;
+ MasterDetector *masterDetector;
unordered_map<FrameworkID, Framework *> frameworks;
unordered_map<SlaveID, Slave *> slaves;
unordered_map<OfferID, SlotOffer *> slotOffers;
@@ -283,17 +274,21 @@ protected:
unordered_map<PID, FrameworkID> pidToFid;
unordered_map<PID, SlaveID> pidToSid;
- FrameworkID nextFrameworkId; // Used to give each framework a unique ID.
- SlaveID nextSlaveId; // Used to give each slave a unique ID.
- OfferID nextSlotOfferId; // Used to give each slot offer a unique ID.
+ long nextFrameworkId; // Used to give each framework a unique ID.
+ long nextSlaveId; // Used to give each slave a unique ID.
+ long nextSlotOfferId; // Used to give each slot offer a unique ID.
+
string allocatorType;
Allocator *allocator;
+ long masterId; // Used to differentiate different master in FT mode, will be ephemeral id
+ FTMessaging *ftMsg;
+
public:
- Master();
+ Master(const string &zk = "");
- Master(const string& _allocatorType);
+ Master(const string& _allocatorType, const string &zk = "");
~Master();
@@ -304,7 +299,7 @@ public:
void rescindOffer(SlotOffer *offer);
- void killTask(Task *task);
+ void killTask(TaskInfo *task);
Framework * lookupFramework(FrameworkID fid);
@@ -312,6 +307,13 @@ public:
SlotOffer * lookupSlotOffer(OfferID soid);
+ // Used in FT mode. Ensures that task is also registered in frameworks->tasks
+ void updateFrameworkTasks(TaskInfo *task);
+
+ // Used in FT mode. Traverses all slaves' tasks t and calls updateFrameworkTasks(t)
+ void updateFrameworkTasks();
+
+
// Return connected frameworks that are not in the process of being removed
vector<Framework *> getActiveFrameworks();
@@ -346,7 +348,7 @@ protected:
OfferReturnReason reason,
const vector<SlaveResources>& resourcesLeft);
- void removeTask(Task *task, TaskRemovalReason reason);
+ void removeTask(TaskInfo *task, TaskRemovalReason reason);
// Kill all of a framework's tasks, delete the framework object, and
// reschedule slot offers for slots that were assigned to this framework
@@ -382,7 +384,7 @@ inline std::ostream& operator << (std::o
}
-inline std::ostream& operator << (std::ostream& stream, const Task *t)
+inline std::ostream& operator << (std::ostream& stream, const TaskInfo *t)
{
stream << "task " << t->frameworkId << ":" << t->id;
return stream;
Modified: incubator/mesos/trunk/src/messages.hpp
URL: http://svn.apache.org/viewvc/incubator/mesos/trunk/src/messages.hpp?rev=1131739&r1=1131738&r2=1131739&view=diff
==============================================================================
--- incubator/mesos/trunk/src/messages.hpp (original)
+++ incubator/mesos/trunk/src/messages.hpp Sun Jun 5 05:24:06 2011
@@ -60,8 +60,10 @@ enum MessageType {
/* From slave heart to master. */
SH2M_HEARTBEAT,
- /* From leader election listener to slave */
- LE_NEWLEADER,
+ /* From master detector to processes */
+ NEW_MASTER_DETECTED,
+ NO_MASTER_DETECTED,
+ GOT_MASTER_SEQ,
/* From master to slave. */
M2S_REGISTER_REPLY,
@@ -271,8 +273,15 @@ TUPLE(S2M_LOST_EXECUTOR,
TUPLE(SH2M_HEARTBEAT,
(SlaveID));
-TUPLE(LE_NEWLEADER,
- (std::string /*PID of new leader*/));
+TUPLE(NEW_MASTER_DETECTED,
+ (std::string, /* master seq */
+ PID /* master PID */));
+
+TUPLE(NO_MASTER_DETECTED,
+ ());
+
+TUPLE(GOT_MASTER_SEQ,
+ (std::string /* seq */));
TUPLE(M2S_REGISTER_REPLY,
(SlaveID));
Modified: incubator/mesos/trunk/src/nexus_sched.cpp
URL: http://svn.apache.org/viewvc/incubator/mesos/trunk/src/nexus_sched.cpp?rev=1131739&r1=1131738&r2=1131739&view=diff
==============================================================================
--- incubator/mesos/trunk/src/nexus_sched.cpp (original)
+++ incubator/mesos/trunk/src/nexus_sched.cpp Sun Jun 5 05:24:06 2011
@@ -26,6 +26,9 @@
#include "messages.hpp"
#include "nexus_local.hpp"
#include "nexus_sched.hpp"
+#include "url_processor.hpp"
+#include "master_detector.hpp"
+#include "ft_messaging.hpp"
using std::cerr;
@@ -54,6 +57,8 @@ namespace nexus { namespace internal {
* any synchronization necessary is performed.
*/
+
+
class SchedulerProcess : public Tuple<Process>
{
public:
@@ -66,22 +71,68 @@ private:
FrameworkID fid;
string frameworkName;
ExecutorInfo execInfo;
+ bool isFT;
+ string zkServers;
+ MasterDetector *masterDetector;
+ FTMessaging *ftMsg;
+
+ typedef unordered_map< SlaveID, PID > SidToPidMap;
+ unordered_map< OfferID, SidToPidMap > savedOffers;
+ SidToPidMap sidToPidMap;
volatile bool terminate;
+ class TimeoutListener;
+ friend class TimeoutListener;
+
+ class TimeoutListener : public FTCallback {
+ public:
+ TimeoutListener(SchedulerProcess *s, const vector<TaskDescription> t) : parent(s), tasks(t) {}
+
+ virtual void timeout() {
+ foreach (const TaskDescription &t, tasks) {
+ DLOG(INFO) << "FT: faking M2F_STATUS_UPDATE due to timeout to server during ReplyToOffer";
+ parent->send( parent->self(),
+ pack<M2F_STATUS_UPDATE>(t.taskId, TASK_LOST, ""));
+ }
+ }
+
+ private:
+ SchedulerProcess *parent;
+ vector<TaskDescription> tasks;
+ };
+
public:
- SchedulerProcess(const PID &_master,
+ SchedulerProcess(const string &_master,
NexusSchedulerDriver* _driver,
Scheduler* _sched,
const string& _frameworkName,
const ExecutorInfo& _execInfo)
- : master(_master),
- driver(_driver),
+ : driver(_driver),
sched(_sched),
- fid(-1),
+ fid(""),
terminate(false),
frameworkName(_frameworkName),
- execInfo(_execInfo) {}
+ execInfo(_execInfo),
+ masterDetector(NULL)
+{
+ pair<UrlProcessor::URLType, string> urlPair = UrlProcessor::process(_master);
+ if (urlPair.first == UrlProcessor::ZOO) {
+ isFT = true;
+ zkServers = urlPair.second;
+ // } else if (urlPair.first == UrlProcessor::NEXUS) {
+ } else {
+ isFT = false;
+ istringstream ss(urlPair.second); // the case nexus://
+ istringstream ss2(_master); // in case nexus:// is missing
+ if (!((ss >> master) || (ss2 >> master))) {
+ cerr << "Failed to parse URL for master: " << _master <<endl;
+ exit(1);
+ }
+ }
+
+ ftMsg = FTMessaging::getInstance();
+}
protected:
void operator () ()
@@ -92,8 +143,8 @@ protected:
fatal("failed to get username information");
string user(passwd->pw_name);
- link(master);
- send(master, pack<F2M_REGISTER_FRAMEWORK>(frameworkName, user, execInfo));
+ LOG(INFO) << "Connecting to ZooKeeper at " << zkServers;
+ masterDetector = new MasterDetector(zkServers, ZNODE, self(), false);
while(true) {
// Rather than send a message to this process when it is time to
@@ -111,11 +162,39 @@ protected:
if (terminate)
return;
+ switch(receive(FT_TIMEOUT)) {
// TODO(benh): We need to break the receive loop every so often
// to check if 'terminate' has been set .. but rather than use a
// timeout in receive, maybe we should send a message.
- switch(receive(1)) {
+ case NEW_MASTER_DETECTED: {
+ string masterSeq;
+ PID masterPid;
+ unpack<NEW_MASTER_DETECTED>(masterSeq, masterPid);
+
+ LOG(INFO) << "New master at " << masterPid
+ << " with ephemeral id:" << masterSeq;
+
+ // TODO(alig|benh): Use new API -> redirect(master, masterPid);
+ master = masterPid;
+ ftMsg->setMasterPid(master);
+ link(master);
+
+ if (fid.empty()) {
+ // Touched for the very first time.
+ send(master, pack<F2M_REGISTER_FRAMEWORK>(frameworkName, user, execInfo));
+ } else {
+ // Not the first time.
+ send(master, pack<F2M_REREGISTER_FRAMEWORK>(fid, frameworkName, user, execInfo));
+ }
+ break;
+ }
+
+ case NO_MASTER_DETECTED: {
+ // TODO(alig): Do we want to do anything here?
+ break;
+ }
+
case M2F_REGISTER_REPLY: {
unpack<M2F_REGISTER_REPLY>(fid);
invoke(bind(&Scheduler::registered, sched, driver, fid));
@@ -126,17 +205,86 @@ protected:
OfferID oid;
vector<SlaveOffer> offs;
unpack<M2F_SLOT_OFFER>(oid, offs);
+
+ SidToPidMap &tmpMap = savedOffers[ oid ];
+ foreach(const SlaveOffer &offer, offs) {
+ tmpMap[ offer.slaveId ] = offer.slavePid;
+ }
+
invoke(bind(&Scheduler::resourceOffer, sched, driver, oid, ref(offs)));
break;
}
+ case F2F_SLOT_OFFER_REPLY: {
+ OfferID oid;
+ vector<TaskDescription> tasks;
+ Params params;
+ unpack<F2F_SLOT_OFFER_REPLY>(oid, tasks, params);
+
+ // Save only the PIDs we need for sending framework messages.
+ foreach(const TaskDescription &task, tasks) {
+ sidToPidMap[ task.slaveId ] = savedOffers[ oid ][ task.slaveId ];
+
+ }
+ savedOffers.erase( oid );
+
+ // TODO(alig|benh): Use new API -> rsend(master, pack<F2M_SLOT_OFFER_REPLY>(fid, oid, tasks, params));
+ // TODO(alig): Improve the following comment.
+ // Do a reliable send here because ...
+ if (isFT) {
+ TimeoutListener *tListener =
+ new TimeoutListener(this, tasks);
+
+ string ftId = ftMsg->getNextId();
+ DLOG(INFO) << "Sending reliably reply to slot offer for msg " << ftId;
+ ftMsg->reliableSend( ftId,
+ pack<F2M_FT_SLOT_OFFER_REPLY>(ftId, self(), fid, oid, tasks, params),
+ tListener);
+ } else
+ send(master, pack<F2M_SLOT_OFFER_REPLY>(fid, oid, tasks, params));
+
+ break;
+ }
+
+ case F2F_FRAMEWORK_MESSAGE: {
+ FrameworkMessage msg;
+ unpack<F2F_FRAMEWORK_MESSAGE>(msg);
+
+ /*
+ if (isFT) {
+ string ftId = ftMsg->getNextId();
+ ftMsg->reliableSend( ftId, pack<F2M_FT_FRAMEWORK_MESSAGE>(ftId, self(), fid, msg));
+ } else
+ send(master, pack<F2M_FRAMEWORK_MESSAGE>(fid, msg));
+ */
+ send( sidToPidMap[ msg.slaveId ], pack<M2S_FRAMEWORK_MESSAGE>(fid, msg));
+ break;
+ }
+
case M2F_RESCIND_OFFER: {
OfferID oid;
unpack<M2F_RESCIND_OFFER>(oid);
+ savedOffers.erase(oid);
invoke(bind(&Scheduler::offerRescinded, sched, driver, oid));
break;
}
+ case M2F_FT_STATUS_UPDATE: {
+ TaskID tid;
+ TaskState state;
+ string data;
+ string ftId, origPid;
+ unpack<M2F_FT_STATUS_UPDATE>(ftId, origPid, tid, state, data);
+ if (!ftMsg->acceptMessageAck(ftId, origPid))
+ break;
+ DLOG(INFO) << "FT: Received message with id: " << ftId;
+
+ TaskStatus status(tid, state, data);
+ invoke(bind(&Scheduler::statusUpdate, sched, driver, ref(status)));
+ break;
+ }
+
+
case M2F_STATUS_UPDATE: {
TaskID tid;
TaskState state;
@@ -147,6 +295,20 @@ protected:
break;
}
+ case M2F_FT_FRAMEWORK_MESSAGE: {
+ FrameworkMessage msg;
+ string ftId, origPid;
+ unpack<M2F_FT_FRAMEWORK_MESSAGE>(ftId, origPid, msg);
+
+ if (!ftMsg->acceptMessageAck(ftId, origPid))
+ break;
+
+ DLOG(INFO) << "FT: Received message with id: " << ftId;
+
+ invoke(bind(&Scheduler::frameworkMessage, sched, driver, ref(msg)));
+ break;
+ }
+
case M2F_FRAMEWORK_MESSAGE: {
FrameworkMessage msg;
unpack<M2F_FRAMEWORK_MESSAGE>(msg);
@@ -157,6 +319,7 @@ protected:
case M2F_LOST_SLAVE: {
SlaveID sid;
unpack<M2F_LOST_SLAVE>(sid);
+ sidToPidMap.erase(sid);
invoke(bind(&Scheduler::slaveLost, sched, driver, sid));
break;
}
@@ -170,12 +333,29 @@ protected:
}
case PROCESS_EXIT: {
- const char* message = "connection to master failed";
- invoke(bind(&Scheduler::error, sched, driver, -1, message));
+ const char* message = "Connection to master failed";
+ if (isFT)
+ LOG(WARNING) << "Connection to master failed. Waiting for a new master to be elected.";
+ else
+ {
+ LOG(ERROR) << "Connection to master failed. Exiting. Consider running Nexus in FT mode!";
+ invoke(bind(&Scheduler::error, sched, driver, -1, message));
+ }
+ break;
+ }
+
+ case FT_RELAY_ACK: {
+ string ftId, senderStr;
+ unpack<FT_RELAY_ACK>(ftId, senderStr);
+
+ DLOG(INFO) << "FT: got final ack for " << ftId;
+
+ ftMsg->gotAck(ftId);
break;
}
case PROCESS_TIMEOUT: {
+ ftMsg->sendOutstanding();
break;
}
@@ -275,51 +455,53 @@ NexusSchedulerDriver::~NexusSchedulerDri
}
-int NexusSchedulerDriver::start()
+void NexusSchedulerDriver::start()
{
Lock lock(&mutex);
if (running) {
- //error(1, "cannot call start - scheduler is already running");
- return - 1;
+ error(1, "cannot call start - scheduler is already running");
+ return;
}
PID pid;
+
+ string initServer = master;
if (master == string("localquiet")) {
// TODO(benh): Look up resources in environment variables.
pid = run_nexus(1, 1, 1073741824, true, true);
+ ostringstream ss;
+ ss << pid;
+ initServer = ss.str();
} else if (master == string("local")) {
// TODO(benh): Look up resources in environment variables.
pid = run_nexus(1, 1, 1073741824, true, false);
- } else {
- std::istringstream iss(master);
- if (!(iss >> pid)) {
- //error(errno, "register failed - bad master PID");
- return -1;
- }
- }
+ ostringstream ss;
+ ss << pid;
+ initServer = ss.str();
+ }
const string& frameworkName = sched->getFrameworkName(this);
const ExecutorInfo& executorInfo = sched->getExecutorInfo(this);
- process = new SchedulerProcess(pid, this, sched, frameworkName, executorInfo);
+ process = new SchedulerProcess(initServer, this, sched, frameworkName, executorInfo);
+
Process::spawn(process);
running = true;
-
- return 0;
}
-int NexusSchedulerDriver::stop()
+void NexusSchedulerDriver::stop()
{
Lock lock(&mutex);
if (!running) {
// Don't issue an error (could lead to an infinite loop).
- return -1;
+ // TODO(benh): It would be much cleaner to return success or failure!
+ return;
}
// TODO(benh): Do a Process::post instead?
@@ -331,115 +513,97 @@ int NexusSchedulerDriver::stop()
running = false;
pthread_cond_signal(&cond);
-
- return 0;
}
-int NexusSchedulerDriver::join()
+void NexusSchedulerDriver::join()
{
Lock lock(&mutex);
while (running)
pthread_cond_wait(&cond, &mutex);
-
- return 0;
}
-int NexusSchedulerDriver::run()
+void NexusSchedulerDriver::run()
{
- int ret = start();
- return ret != 0 ? ret : join();
+ start();
+ join();
}
-int NexusSchedulerDriver::killTask(TaskID tid)
+void NexusSchedulerDriver::killTask(TaskID tid)
{
Lock lock(&mutex);
if (!running) {
- //error(1, "cannot call killTask - scheduler is not running");
- return -1;
+ error(1, "cannot call killTask - scheduler is not running");
+ return;
}
// TODO(benh): Do a Process::post instead?
process->send(process->master,
process->pack<F2M_KILL_TASK>(process->fid, tid));
-
- return 0;
}
-int NexusSchedulerDriver::replyToOffer(OfferID offerId,
- const vector<TaskDescription> &tasks,
- const string_map ¶ms)
+void NexusSchedulerDriver::replyToOffer(OfferID offerId,
+ const vector<TaskDescription> &tasks,
+ const string_map ¶ms)
{
Lock lock(&mutex);
if (!running) {
- //error(1, "cannot call replyToOffer - scheduler is not running");
- return -1;
+ error(1, "cannot call replyToOffer - scheduler is not running");
+ return;
}
// TODO(benh): Do a Process::post instead?
-
- process->send(process->master,
- process->pack<F2M_SLOT_OFFER_REPLY>(process->fid,
- offerId,
- tasks,
- Params(params)));
-
- return 0;
+
+ process->send( process->self(), process->pack<F2F_SLOT_OFFER_REPLY>(offerId, tasks, Params(params)));
}
-int NexusSchedulerDriver::reviveOffers()
+void NexusSchedulerDriver::reviveOffers()
{
Lock lock(&mutex);
if (!running) {
- //error(1, "cannot call reviveOffers - scheduler is not running");
- return -1;
+ error(1, "cannot call reviveOffers - scheduler is not running");
+ return;
}
// TODO(benh): Do a Process::post instead?
process->send(process->master,
process->pack<F2M_REVIVE_OFFERS>(process->fid));
-
- return 0;
}
-int NexusSchedulerDriver::sendFrameworkMessage(const FrameworkMessage &message)
+void NexusSchedulerDriver::sendFrameworkMessage(const FrameworkMessage &message)
{
Lock lock(&mutex);
if (!running) {
- //error(1, "cannot call sendFrameworkMessage - scheduler is not running");
- return -1;
+ error(1, "cannot call sendFrameworkMessage - scheduler is not running");
+ return;
}
- process->send(process->master,
- process->pack<F2M_FRAMEWORK_MESSAGE>(process->fid, message));
-
- return 0;
+ process->send( process->self(), process->pack<F2F_FRAMEWORK_MESSAGE>(message) );
}
-int NexusSchedulerDriver::sendHints(const string_map& hints)
+void NexusSchedulerDriver::sendHints(const string_map& hints)
{
Lock lock(&mutex);
if (!running) {
- //error(1, "cannot call sendHints - scheduler is not running");
- return -1;
+ error(1, "cannot call sendHints - scheduler is not running");
+ return;
}
// TODO(*): Send the hints; for now, we do nothing
- //error(1, "sendHints is not yet implemented");
- return -1;
+ error(1, "sendHints is not yet implemented");
}
@@ -496,7 +660,7 @@ public:
virtual void registered(SchedulerDriver*, FrameworkID frameworkId)
{
- sched->registered(sched, frameworkId);
+ sched->registered(sched, frameworkId.c_str());
}
virtual void resourceOffer(SchedulerDriver*,
@@ -515,19 +679,19 @@ public:
// Create C offer structs
nexus_slot* c_offers = new nexus_slot[offers.size()];
for (size_t i = 0; i < offers.size(); i++) {
- nexus_slot offer = { offers[i].slaveId,
+ nexus_slot offer = { offers[i].slaveId.c_str(),
offers[i].host.c_str(),
paramStrs[i].c_str() };
c_offers[i] = offer;
}
- sched->slot_offer(sched, offerId, c_offers, offers.size());
+ sched->slot_offer(sched, offerId.c_str(), c_offers, offers.size());
delete[] c_offers;
}
virtual void offerRescinded(SchedulerDriver*, OfferID offerId)
{
- sched->slot_offer_rescinded(sched, offerId);
+ sched->slot_offer_rescinded(sched, offerId.c_str());
}
virtual void statusUpdate(SchedulerDriver*, TaskStatus &status)
@@ -541,7 +705,7 @@ public:
virtual void frameworkMessage(SchedulerDriver*, FrameworkMessage &message)
{
- nexus_framework_message c_message = { message.slaveId,
+ nexus_framework_message c_message = { message.slaveId.c_str(),
message.taskId,
message.data.data(),
message.data.size() };
@@ -550,7 +714,7 @@ public:
virtual void slaveLost(SchedulerDriver*, SlaveID sid)
{
- sched->slave_lost(sched, sid);
+ sched->slave_lost(sched, sid.c_str());
}
virtual void error(SchedulerDriver*, int code, const std::string &message)
@@ -697,7 +861,7 @@ int nexus_sched_send_message(struct nexu
return -1;
}
- FrameworkMessage message(msg->sid, msg->tid,
+ FrameworkMessage message(string(msg->sid), msg->tid,
string((char*) msg->data, msg->data_len));
CScheduler* cs = lookupCScheduler(sched);
@@ -761,7 +925,7 @@ int nexus_sched_reply_to_offer(struct ne
string taskArg((char*) tasks[i].arg, tasks[i].arg_len);
wrapped_tasks[i] = TaskDescription(tasks[i].tid,
- tasks[i].sid,
+ string(tasks[i].sid),
string(tasks[i].name),
params,
taskArg);
Modified: incubator/mesos/trunk/src/slave.cpp
URL: http://svn.apache.org/viewvc/incubator/mesos/trunk/src/slave.cpp?rev=1131739&r1=1131738&r2=1131739&view=diff
==============================================================================
--- incubator/mesos/trunk/src/slave.cpp (original)
+++ incubator/mesos/trunk/src/slave.cpp Sun Jun 5 05:24:06 2011
@@ -1,19 +1,17 @@
-#include "config.hpp" // Need to define first to get USING_ZOOKEEPER
-
#include <getopt.h>
-#ifdef USING_ZOOKEEPER
-#include <zookeeper.hpp>
-#endif
-
-#include "isolation_module_factory.hpp"
+#include <fstream>
+#include <algorithm>
#include "slave.hpp"
#include "slave_webui.hpp"
+#include "isolation_module_factory.hpp"
+#include "url_processor.hpp"
+
+#define FT_TIMEOUT 10
using std::list;
using std::make_pair;
using std::ostringstream;
-using std::istringstream;
using std::pair;
using std::queue;
using std::string;
@@ -27,70 +25,13 @@ using namespace nexus;
using namespace nexus::internal;
using namespace nexus::internal::slave;
-
// There's no gethostbyname2 on Solaris, so fake it by calling gethostbyname
#ifdef __sun__
#define gethostbyname2(name, _) gethostbyname(name)
#endif
-
-/* List of ZooKeeper host:port pairs (from slave_main.cpp/local.cpp). */
-extern string zookeeper;
-
namespace {
-#ifdef USING_ZOOKEEPER
-class SlaveWatcher : public Watcher
-{
-private:
- Slave *slave;
-
-public:
- void process(ZooKeeper *zk, int type, int state, const string &path)
- {
- string znode = "/home/nexus/master";
- int ret;
-
- if ((state == ZOO_CONNECTED_STATE) &&
- ((type == ZOO_SESSION_EVENT) || (type == ZOO_CREATED_EVENT))) {
- // Check if znode exists, if not, just return and wait.
- ret = zk->exists(znode, true, NULL);
-
- if (ret == ZNONODE)
- return;
-
- if (ret != ZOK)
- fatal("failed to get %s! (%s)", znode.c_str(), zerror(ret));
-
- string result;
- ret = zk->get(znode, false, &result, NULL);
-
- if (ret != ZOK)
- fatal("failed to get %s! (%s)", znode.c_str(), zerror(ret));
-
- PID master;
-
- istringstream iss(result);
- if (!(iss >> master))
- fatal("bad data at %s!", znode.c_str());
-
- // TODO(benh): HACK! Don't just set field in slave instance!
- slave->master = master;
-
- Process::post(slave->getPID(), S2S_GOT_MASTER);
- } else if ((state == ZOO_CONNECTED_STATE) &&
- (type == ZOO_DELETED_EVENT) &&
- (path.compare(znode) == 0)) {
- // Master gone, we should get a PROCESS_EXIT and commit suicide.
- } else {
- fatal("unhandled ZooKeeper event!");
- }
- }
-
- SlaveWatcher(Slave *_slave) : slave(_slave) {}
-};
-#endif
-
// Periodically sends heartbeats to the master
class Heart : public Tuple<Process>
{
@@ -122,20 +63,52 @@ public:
} /* namespace */
-Slave::Slave(const PID &_master, Resources _resources, bool _local)
- : master(_master), resources(_resources), local(_local), id(-1),
+Slave::Slave(const string &_master, Resources _resources, bool _local)
+ : masterDetector(NULL), resources(_resources), local(_local), id(""),
isolationType("process"), isolationModule(NULL)
-{}
+{
+ ftMsg = FTMessaging::getInstance();
+ pair<UrlProcessor::URLType, string> urlPair = UrlProcessor::process(_master);
+ if (urlPair.first == UrlProcessor::ZOO) {
+ isFT = true;
+ zkServers = urlPair.second;
+ } else {
+ isFT = false;
+ istringstream iss(urlPair.second);
+ istringstream iss2(_master);
+ if (!((iss >> master) || (iss2 >> master) )) {
+ cerr << "Failed to resolve master PID " << urlPair.second << endl;
+ exit(1);
+ }
+ }
+}
-Slave::Slave(const PID &_master, Resources _resources, bool _local,
- const string& _isolationType)
- : master(_master), resources(_resources), local(_local), id(-1),
+Slave::Slave(const string &_master, Resources _resources, bool _local,
+ const string &_isolationType)
+ : masterDetector(NULL), resources(_resources), local(_local), id(""),
isolationType(_isolationType), isolationModule(NULL)
-{}
-
+{
+ ftMsg = FTMessaging::getInstance();
+ pair<UrlProcessor::URLType, string> urlPair = UrlProcessor::process(_master);
+ if (urlPair.first == UrlProcessor::ZOO) {
+ isFT = true;
+ zkServers = urlPair.second;
+ } else {
+ isFT = false;
+ istringstream iss(urlPair.second);
+ istringstream iss2(_master);
+ if (!((iss >> master) || (iss2 >> master) )) {
+ cerr << "Failed to resolve master PID " << urlPair.second << endl;
+ exit(1);
+ }
+ }
+}
-Slave::~Slave() {}
+Slave::~Slave()
+{
+ delete masterDetector;
+}
state::SlaveState *Slave::getState()
@@ -153,7 +126,7 @@ state::SlaveState *Slave::getState()
f->executorInfo.uri, f->executorStatus, f->resources.cpus,
f->resources.mem);
state->frameworks.push_back(framework);
- foreachpair(_, Task *t, f->tasks) {
+ foreachpair(_, TaskInfo *t, f->tasks) {
state::Task *task = new state::Task(t->id, t->name, t->state,
t->resources.cpus, t->resources.mem);
framework->tasks.push_back(task);
@@ -168,6 +141,9 @@ void Slave::operator () ()
{
LOG(INFO) << "Slave started at " << self();
+ LOG(INFO) << "Connecting to ZooKeeper at " << zkServers;
+ masterDetector = new MasterDetector(zkServers, ZNODE, self(), false);
+
// Get our hostname
char buf[256];
gethostname(buf, sizeof(buf));
@@ -181,16 +157,6 @@ void Slave::operator () ()
if (!publicDns) {
publicDns = hostname;
}
-
-#ifdef USING_ZOOKEEPER
- ZooKeeper *zk;
- if (!zookeeper.empty())
- zk = new ZooKeeper(zookeeper, 10000, new SlaveWatcher(this));
- else
- send(self(), pack<S2S_GOT_MASTER>());
-#else
- send(self(), pack<S2S_GOT_MASTER>());
-#endif
FrameworkID fid;
TaskID tid;
@@ -200,11 +166,42 @@ void Slave::operator () ()
string data;
while (true) {
- switch (receive()) {
- case S2S_GOT_MASTER: {
- LOG(INFO) << "Connecting to Nexus master at " << master;
+ switch (receive(FT_TIMEOUT)) {
+ case NEW_MASTER_DETECTED: {
+ string masterSeq;
+ PID masterPid;
+ LOG(INFO) << "New master at " << masterPid
+ << " with ephemeral id:" << masterSeq;
+
+ // TODO(alig|benh): Use new API -> redirect(master, masterPid);
+ master = masterPid;
+ ftMsg->setMasterPid(master);
link(master);
- send(master, pack<S2M_REGISTER_SLAVE>(hostname, publicDns, resources));
+
+ if (id.empty()) {
+ // Slave started before master.
+ send(master, pack<S2M_REGISTER_SLAVE>(hostname, publicDns, resources));
+ } else {
+ // Reconnecting, so reconstruct resourcesInUse for the master.
+ Resources resourcesInUse;
+ vector<TaskInfo> taskVec;
+
+ foreachpair(_, Framework *framework, frameworks) {
+ foreachpair(_, TaskInfo *task, framework->tasks) {
+ resourcesInUse += task->resources;
+ TaskInfo ti = *task;
+ ti.slaveId = id;
+ taskVec.push_back(ti);
+ }
+ }
+
+ send(master, pack<S2M_REREGISTER_SLAVE>(id, hostname, publicDns, resources, taskVec));
+ }
+ break;
+ }
+
+ case NO_MASTER_DETECTED: {
+ // TODO(alig): Do anything here?
break;
}
@@ -218,11 +215,22 @@ void Slave::operator () ()
break;
}
+ case M2S_REREGISTER_REPLY: {
+ FrameworkID tmpfid;
+ unpack<M2S_REGISTER_REPLY>(tmpfid);
+ LOG(INFO) << "RE-registered with master; given slave ID " << tmpfid << " had "<< this->id;
+ if (this->id == "")
+ this->id = tmpfid;
+ link(spawn(new Heart(master, getPID(), this->id)));
+ break;
+ }
+
case M2S_RUN_TASK: {
- string fwName, user, taskName, taskArg;
+ FrameworkID fid;
+ string fwName, user, taskName, taskArg, fwPidStr;
ExecutorInfo execInfo;
unpack<M2S_RUN_TASK>(fid, tid, fwName, user, execInfo,
- taskName, taskArg, params);
+ taskName, taskArg, params, fwPidStr);
LOG(INFO) << "Got assigned task " << fid << ":" << tid;
Resources res;
res.cpus = params.getInt32("cpus", -1);
@@ -230,12 +238,15 @@ void Slave::operator () ()
Framework *framework = getFramework(fid);
if (framework == NULL) {
// Framework not yet created on this node - create it
- framework = new Framework(fid, fwName, user, execInfo);
+ PID fwPid;
+ istringstream ss(fwPidStr);
+ ss >> fwPid;
+ framework = new Framework(fid, fwName, user, execInfo, fwPid);
frameworks[fid] = framework;
isolationModule->frameworkAdded(framework);
isolationModule->startExecutor(framework);
}
- Task *task = framework->addTask(tid, taskName, res);
+ TaskInfo *task = framework->addTask(tid, taskName, res);
Executor *executor = getExecutor(fid);
if (executor) {
send(executor->pid,
@@ -265,12 +276,30 @@ void Slave::operator () ()
break;
}
+ case M2S_FT_FRAMEWORK_MESSAGE: {
+ string ftId, origPid;
+ unpack<M2S_FT_FRAMEWORK_MESSAGE>(ftId, origPid, fid, message);
+
+ if (!ftMsg->acceptMessageAck(ftId, origPid))
+ break;
+
+ DLOG(INFO) << "FT: Received message with id: " << ftId;
+
+ if (Executor *ex = getExecutor(fid)) {
+ send(ex->pid, pack<S2E_FRAMEWORK_MESSAGE>(message));
+ }
+ // TODO(*): If executor is not started, queue framework message?
+ // (It's probably okay to just drop it since frameworks can have
+ // the executor send a message to the master to say when it's ready.)
+ break;
+ }
+
case M2S_FRAMEWORK_MESSAGE: {
unpack<M2S_FRAMEWORK_MESSAGE>(fid, message);
if (Executor *ex = getExecutor(fid)) {
send(ex->pid, pack<S2E_FRAMEWORK_MESSAGE>(message));
}
- // TODO(matei): If executor is not started, queue framework message?
+ // TODO(*): If executor is not started, queue framework message?
// (It's probably okay to just drop it since frameworks can have
// the executor send a message to the master to say when it's ready.)
break;
@@ -324,7 +353,11 @@ void Slave::operator () ()
}
}
// Pass on the update to the master
- send(master, pack<S2M_STATUS_UPDATE>(id, fid, tid, taskState, data));
+ if (isFT) {
+ string ftId = ftMsg->getNextId();
+ ftMsg->reliableSend(ftId, pack<S2M_FT_STATUS_UPDATE>(ftId, self(), id, fid, tid, taskState, data));
+ } else
+ send(master, pack<S2M_STATUS_UPDATE>(id, fid, tid, taskState, data));
break;
}
@@ -332,7 +365,14 @@ void Slave::operator () ()
unpack<E2S_FRAMEWORK_MESSAGE>(fid, message);
// Set slave ID in case framework omitted it
message.slaveId = this->id;
- send(master, pack<S2M_FRAMEWORK_MESSAGE>(id, fid, message));
+ /*
+ if (isFT) {
+ string ftId = ftMsg->getNextId();
+ ftMsg->reliableSend(ftId, pack<S2M_FT_FRAMEWORK_MESSAGE>(ftId, self(), id, fid, message));
+ } else
+ send(master, pack<S2M_FRAMEWORK_MESSAGE>(id, fid, message));
+ */
+ send(getFramework(fid)->fwPid, pack<M2F_FRAMEWORK_MESSAGE>(message));
break;
}
@@ -346,12 +386,16 @@ void Slave::operator () ()
if (from() == master) {
// TODO: Fault tolerance!
- LOG(ERROR) << "Master disconnected! Committing suicide ...";
- // TODO(matei): Add support for factory style destroy of objects!
- if (isolationModule != NULL)
- delete isolationModule;
- // TODO: Shut down executors?
- return;
+ if (isFT)
+ LOG(WARNING) << "FT: Master disconnected! Waiting for a new master to be elected.";
+ else
+ {
+ LOG(ERROR) << "Master disconnected! Exiting. Consider running Nexus in FT mode!";
+ if (isolationModule != NULL)
+ delete isolationModule;
+ // TODO: Shut down executors?
+ return;
+ }
}
foreachpair (_, Executor *ex, executors) {
@@ -379,6 +423,7 @@ void Slave::operator () ()
return;
}
+
case S2S_SHUTDOWN: {
LOG(INFO) << "Asked to shut down by " << from();
// TODO(matei): Add support for factory style destroy of objects!
@@ -388,6 +433,20 @@ void Slave::operator () ()
return;
}
+ case FT_RELAY_ACK: {
+ string ftId, senderStr;
+ unpack<FT_RELAY_ACK>(ftId, senderStr);
+
+ DLOG(INFO) << "FT: got final ack for " << ftId;
+
+ ftMsg->gotAck(ftId);
+ break;
+ }
+
+ case PROCESS_TIMEOUT: {
+ ftMsg->sendOutstanding();
+ break;
+ }
default: {
LOG(ERROR) << "Received unknown message ID " << msgid()
<< " from " << from();
Modified: incubator/mesos/trunk/src/slave.hpp
URL: http://svn.apache.org/viewvc/incubator/mesos/trunk/src/slave.hpp?rev=1131739&r1=1131738&r2=1131739&view=diff
==============================================================================
--- incubator/mesos/trunk/src/slave.hpp (original)
+++ incubator/mesos/trunk/src/slave.hpp Sun Jun 5 05:24:06 2011
@@ -34,7 +34,7 @@
#include "params.hpp"
#include "resources.hpp"
#include "slave_state.hpp"
-#include "leader_detector.hpp"
+#include "master_detector.hpp"
#include "task_info.hpp"
#include "ft_messaging.hpp"
@@ -167,7 +167,7 @@ public:
bool isFT;
string zkServers;
- LeaderDetector *leaderDetector;
+ MasterDetector *masterDetector;
PID master;
SlaveID id;
Resources resources;
@@ -178,31 +178,6 @@ public:
IsolationModule *isolationModule;
FTMessaging *ftMsg;
-
- class SlaveLeaderListener;
- friend class SlaveLeaderListener;
-
- class SlaveLeaderListener : public LeaderListener {
- public:
- // Need to be thread safe. Currently does not use any shared variables.
- SlaveLeaderListener(Slave *s, PID pp) : parent(s), parentPID(pp) {}
-
- virtual void newLeaderElected(const string &zkId, const string &pidStr) {
- if (zkId != "") {
- LOG(INFO) << "Leader listener detected leader at " << pidStr <<" with ephemeral id:" << zkId;
-
- LOG(INFO) << "Sending message to parent " << parentPID << " about new leader";
- parent->send(parentPID, parent->pack<LE_NEWLEADER>(pidStr));
-
- }
- }
-
- private:
- Slave *parent;
- PID parentPID;
- } slaveLeaderListener;
-
-
public:
Slave(const string &_master, Resources resources, bool _local);
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=1131739&r1=1131738&r2=1131739&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:24:06 2011
@@ -1085,7 +1085,7 @@ public:
/* Free any pending messages. */
while (!process->msgs.empty()) {
struct msg *msg = process->msgs.front();
- process->msgs.pop();
+ process->msgs.pop_front();
free(msg);
}
@@ -2863,7 +2863,7 @@ void Process::enqueue(struct msg *msg)
{
if (state != EXITED) {
//cout << "enqueing pending message: " << msg << endl;
- msgs.push(msg);
+ msgs.push_back(msg);
if (state == RECEIVING) {
state = READY;
@@ -2895,7 +2895,7 @@ struct msg * Process::dequeue()
assert (state == RUNNING);
if (!msgs.empty()) {
msg = msgs.front();
- msgs.pop();
+ msgs.pop_front();
//cout << "dequeueing pending message: " << msg << endl;
}
}
@@ -2905,19 +2905,51 @@ struct msg * Process::dequeue()
}
-PID Process::self()
+PID Process::self() const
{
return pid;
}
-PID Process::from()
+PID Process::from() const
{
PID pid = { 0, 0, 0 };
return current != NULL ? current->from : pid;
}
+void Process::inject(const PID &from, MSGID id, const char *data, size_t length)
+{
+ if (replaying)
+ return;
+
+ /* Disallow sending messages using an internal id. */
+ if (id < PROCESS_MSGID)
+ return;
+
+ /* Allocate/Initialize outgoing message. */
+ struct msg *msg = (struct msg *) malloc(sizeof(struct msg) + length);
+
+ msg->from.pipe = from.pipe;
+ msg->from.ip = from.ip;
+ msg->from.port = from.port;
+ msg->to.pipe = pid.pipe;
+ msg->to.ip = pid.ip;
+ msg->to.port = pid.port;
+ msg->id = id;
+ msg->len = length;
+
+ if (length > 0)
+ memcpy((char *) msg + sizeof(struct msg), data, length);
+
+ lock();
+ {
+ msgs.push_front(msg);
+ }
+ unlock();
+}
+
+
void Process::send(const PID &to, MSGID id, const char *data, size_t length)
{
//cout << "Process::send" << endl;
@@ -3045,7 +3077,7 @@ MSGID Process::call(const PID &to, MSGID
}
-const char * Process::body(size_t *length)
+const char * Process::body(size_t *length) const
{
if (current != NULL && current->len > 0) {
if (length != NULL)
Modified: incubator/mesos/trunk/src/third_party/libprocess/process.hpp
URL: http://svn.apache.org/viewvc/incubator/mesos/trunk/src/third_party/libprocess/process.hpp?rev=1131739&r1=1131738&r2=1131739&view=diff
==============================================================================
--- incubator/mesos/trunk/src/third_party/libprocess/process.hpp (original)
+++ incubator/mesos/trunk/src/third_party/libprocess/process.hpp Sun Jun 5 05:24:06 2011
@@ -5,8 +5,6 @@
#include <stdlib.h>
#include <ucontext.h>
-#include <sys/time.h>
-
#ifdef USE_LITHE
#include <lithe.hh>
@@ -113,7 +111,7 @@ private:
EXITED } state;
/* Queue of messages received. */
- std::queue<struct msg *> msgs;
+ std::deque<struct msg *> msgs;
/* Current message. */
struct msg *current;
@@ -160,66 +158,67 @@ protected:
virtual void operator() () = 0;
/* Returns the PID describing this process. */
- PID self();
+ PID self() const;
/* Returns the sender's PID of the last dequeued (current) message. */
- PID from();
+ PID from() const;
+
+ /* Returns the id of the current message. */
+ MSGID msgid() const;
+
+ /* Returns pointer and length of body of last dequeued (current) message. */
+ const char * body(size_t *length) const;
- MSGID msgid();
+ /* Put a message at front of queue (will not reschedule process). */
+ virtual void inject(const PID &from, MSGID id, const char *data, size_t length);
/* Sends a message to PID. */
- void send(const PID &, MSGID);
+ virtual void send(const PID &to , MSGID);
/* Sends a message with data to PID. */
- void send(const PID &, MSGID, const char *data, size_t length);
+ virtual void send(const PID &to, MSGID id, const char *data, size_t length);
/* Blocks for message indefinitely. */
- MSGID receive();
+ virtual MSGID receive();
/* Blocks for message at most specified seconds. */
- MSGID receive(double sec);
+ virtual MSGID receive(time_t secs);
/* Sends a message to PID and then blocks for a message indefinitely. */
- MSGID call(const PID &, MSGID);
+ virtual MSGID call(const PID &to , MSGID id);
/* Sends a message with data to PID and then blocks for a message. */
- MSGID call(const PID &, MSGID, const char *data, size_t length);
+ virtual MSGID call(const PID &to, MSGID id, const char *data, size_t length);
/* Sends, and then blocks for a message at most specified seconds. */
- MSGID call(const PID &, MSGID, const char *data, size_t length, double secs);
-
- /* Returns pointer and length of body of last dequeued (current) message. */
- const char * body(size_t *length);
+ virtual MSGID call(const PID &to, MSGID id, const char *data, size_t length, time_t secs);
/* Blocks at least specified seconds (may block longer). */
- void pause(double secs);
+ virtual void pause(time_t secs);
/* Links with the specified PID. */
- PID link(const PID &);
+ virtual PID link(const PID &pid);
- /* IO events for awaiting. */
+ /* IO operations for awaiting. */
enum { RDONLY = 01, WRONLY = 02, RDWR = 03 };
/* Wait until operation is ready for file descriptor (or message received). */
- bool await(int fd, int op, const timeval& tv);
-
- /* Wait until operation is ready for file descriptor (or message received). */
- bool await(int fd, int op, const timeval& tv, bool ignore);
+ virtual bool await(int fd, int op);
/* Returns true if operation on file descriptor is ready. */
- bool ready(int fd, int op);
+ virtual bool ready(int fd, int op);
public:
virtual ~Process();
/* Returns pid of process; valid even before calling spawn. */
- PID getPID();
+ PID getPID() const;
/* Sends a message to PID without a return address. */
- static void post(const PID &, MSGID);
+ static void post(const PID &to, MSGID id);
/* Sends a message with data to PID without a return address. */
- static void post(const PID &, MSGID, const char *data, size_t length);
+ static void post(const PID &to, MSGID id, const char *data, size_t length);
/* Spawn a new process. */
static PID spawn(Process *process);
@@ -235,7 +234,7 @@ public:
};
-inline MSGID Process::msgid()
+inline MSGID Process::msgid() const
{
return current != NULL ? current->id : PROCESS_ERROR;
}
@@ -256,7 +255,7 @@ inline MSGID Process::call(const PID &to
inline MSGID Process::call(const PID &to, MSGID id,
const char *data, size_t length)
{
- return call(to, id, data, length, 0);
+ return call(to, id, NULL, 0, 0);
}
@@ -266,7 +265,7 @@ inline MSGID Process::receive()
}
-inline PID Process::getPID()
+inline PID Process::getPID() const
{
return self();
}