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 05:25:53 UTC
svn commit: r1131582 - in /incubator/mesos/trunk: ./ src/
src/third_party/libprocess/
Author: benh
Date: Sun Jun 5 03:25:52 2011
New Revision: 1131582
URL: http://svn.apache.org/viewvc?rev=1131582&view=rev
Log:
FTMessaging class added. STATUS messages from slaves are reliable, they're acked, and resent if no ack is received, redundant messages are dropped. A libprocess abstraction is added that converts a Tuple to a string, such that pending messages are stored in a container by FTMessaging.
Added:
incubator/mesos/trunk/src/ft_messaging.cpp
incubator/mesos/trunk/src/ft_messaging.hpp
Modified:
incubator/mesos/trunk/configure
incubator/mesos/trunk/configure.ac
incubator/mesos/trunk/src/Makefile.in
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/tuple-impl.hpp
Modified: incubator/mesos/trunk/configure
URL: http://svn.apache.org/viewvc/incubator/mesos/trunk/configure?rev=1131582&r1=1131581&r2=1131582&view=diff
==============================================================================
--- incubator/mesos/trunk/configure (original)
+++ incubator/mesos/trunk/configure Sun Jun 5 03:25:52 2011
@@ -1784,8 +1784,7 @@ export CFLAGS
export CXXFLAGS
PYTHON=python
-ac_configure_args_back = ac_configure_args
-ac_configure_args="$ac_configure_args_back --enable-shared=no"
+ac_configure_args="--enable-shared=no --with-pic"
ac_config_files="$ac_config_files Makefile src/Makefile src/config.hpp"
@@ -1820,7 +1819,6 @@ subdirs="$subdirs src/third_party/gtest-
subdirs="$subdirs src/third_party/glog-0.3.0"
-ac_configure_args="$ac_configure_args_back --enable-shared=yes"
subdirs="$subdirs src/third_party/zookeeper-3.2.2/src/c"
Modified: incubator/mesos/trunk/configure.ac
URL: http://svn.apache.org/viewvc/incubator/mesos/trunk/configure.ac?rev=1131582&r1=1131581&r2=1131582&view=diff
==============================================================================
--- incubator/mesos/trunk/configure.ac (original)
+++ incubator/mesos/trunk/configure.ac Sun Jun 5 03:25:52 2011
@@ -10,14 +10,12 @@ export CFLAGS
export CXXFLAGS
PYTHON=python
-ac_configure_args_back = ac_configure_args
-ac_configure_args="$ac_configure_args_back --enable-shared=no"
+ac_configure_args="--enable-shared=no --with-pic"
AC_CONFIG_FILES([Makefile src/Makefile src/config.hpp])
AC_CONFIG_SUBDIRS([src/third_party/libprocess])
AC_CONFIG_SUBDIRS([src/third_party/gtest-1.4.0-patched])
AC_CONFIG_SUBDIRS([src/third_party/glog-0.3.0])
-ac_configure_args="$ac_configure_args_back --enable-shared=yes"
AC_CONFIG_SUBDIRS([src/third_party/zookeeper-3.2.2/src/c])
AC_CANONICAL_SYSTEM
Modified: incubator/mesos/trunk/src/Makefile.in
URL: http://svn.apache.org/viewvc/incubator/mesos/trunk/src/Makefile.in?rev=1131582&r1=1131581&r2=1131582&view=diff
==============================================================================
--- incubator/mesos/trunk/src/Makefile.in (original)
+++ incubator/mesos/trunk/src/Makefile.in Sun Jun 5 03:25:52 2011
@@ -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
+COMMON_OBJ = fatal.o hash_pid.o messages.o lock.o leader_detector.o url_processor.o ft_messaging.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
@@ -281,5 +281,6 @@ clean:
rm -f webui/master/swig/master_wrap.o
rm -f webui/master/swig/master.py
rm -f webui/master/swig/master.pyc
+ rm -f TAGS
.PHONY: default third_party all clean java python ruby test
Added: incubator/mesos/trunk/src/ft_messaging.cpp
URL: http://svn.apache.org/viewvc/incubator/mesos/trunk/src/ft_messaging.cpp?rev=1131582&view=auto
==============================================================================
--- incubator/mesos/trunk/src/ft_messaging.cpp (added)
+++ incubator/mesos/trunk/src/ft_messaging.cpp Sun Jun 5 03:25:52 2011
@@ -0,0 +1,70 @@
+#include <iostream>
+#include <unistd.h>
+#include <climits>
+#include <cstdlib>
+#include <cstdio>
+#include <zookeeper.h>
+#include <glog/logging.h>
+#include "messages.hpp"
+#include "leader_detector.hpp"
+#include "ft_messaging.hpp"
+
+namespace nexus { namespace internal {
+
+using namespace nexus;
+using namespace nexus::internal;
+using namespace std;
+
+FTMessaging *FTMessaging::getInstance() {
+ if (instance==NULL)
+ instance = new FTMessaging();
+ return instance;
+}
+
+FTMessaging *FTMessaging::getInstance(PID _master) {
+ if (instance==NULL)
+ instance = new FTMessaging(_master);
+ return instance;
+}
+
+FTMessaging *FTMessaging::getInstance(string _master) {
+ if (instance==NULL) {
+ PID mPid;
+ istringstream ss(_master);
+ if (!(ss >> mPid)) {
+ LOG(ERROR) << "Couldn't create PID out of string in constructor.";
+ return NULL;
+ }
+ instance = new FTMessaging(mPid);
+ }
+ return instance;
+}
+
+FTMessaging::FTMessaging(PID _master) :
+ master(_master), msgId(0)
+{
+ srand(time(0));
+ char s[50];
+ sprintf(s, "%09i", (int)rand());
+ uniqPrefix = s;
+ DLOG(INFO) << "FT: Created unique FT TAG: " << s;
+}
+
+FTMessaging::FTMessaging() :
+ msgId(0)
+{
+ srand(time(0));
+ char s[50];
+ sprintf(s, "%09i", (int)rand());
+ uniqPrefix = s;
+ DLOG(INFO) << "FT: Created unique FT TAG: " << s;
+ master = {0,0,0};
+}
+
+string FTMessaging::getNextId() {
+ return uniqPrefix + ":" + lexical_cast<string>(msgId++);
+}
+
+FTMessaging *FTMessaging::instance = NULL;
+
+}}
Added: incubator/mesos/trunk/src/ft_messaging.hpp
URL: http://svn.apache.org/viewvc/incubator/mesos/trunk/src/ft_messaging.hpp?rev=1131582&view=auto
==============================================================================
--- incubator/mesos/trunk/src/ft_messaging.hpp (added)
+++ incubator/mesos/trunk/src/ft_messaging.hpp Sun Jun 5 03:25:52 2011
@@ -0,0 +1,177 @@
+#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>
+#include <process.hpp>
+#include <iostream>
+#include <unistd.h>
+#include <climits>
+#include <cstdlib>
+#include <zookeeper.h>
+#include <glog/logging.h>
+#include <ctime>
+#include <cstdlib>
+#include "leader_detector.hpp"
+
+
+#include <dirent.h>
+#include <libgen.h>
+#include <netdb.h>
+#include <pwd.h>
+#include <signal.h>
+#include <stdio.h>
+#include <strings.h>
+
+#include <iostream>
+#include <list>
+#include <sstream>
+#include <vector>
+
+#include <arpa/inet.h>
+
+#include <boost/lexical_cast.hpp>
+#include <boost/unordered_set.hpp>
+#include <boost/unordered_map.hpp>
+
+#include <glog/logging.h>
+
+#include <sys/stat.h>
+#include <sys/types.h>
+#include <sys/wait.h>
+
+#include <process.hpp>
+
+#include "messages.hpp"
+#include "tuple.hpp"
+#include "foreach.hpp"
+
+
+namespace nexus { namespace internal {
+
+using namespace nexus;
+using namespace nexus::internal;
+using namespace std;
+using boost::lexical_cast;
+using boost::unordered_map;
+using boost::unordered_set;
+
+class EmptyClass {
+};
+
+struct StoredMsg {
+ StoredMsg(const string &_ftId, const string &_data, const MSGID &_id) :
+ ftId(_ftId), data(_data), id(_id), count(0) {}
+ StoredMsg() : ftId(""), data(""), id(), count(0) {}
+ string ftId;
+ string data;
+ MSGID id;
+ long count;
+};
+
+
+class FTMessaging {
+public:
+ static FTMessaging *getInstance();
+ static FTMessaging *getInstance(PID master);
+ static FTMessaging *getInstance(string masterStr);
+
+ template<MSGID ID> void reliableSend(const string &ftId, const tuple<ID> &msgTuple)
+ {
+ DLOG(INFO) << "FT: sending " << ftId;
+ string msgStr = Tuple<EmptyClass>::tupleToString(msgTuple);
+ StoredMsg sm(ftId, msgStr, ID);
+ outMsgs[ftId] = sm;
+ if (!master) {
+ DLOG(INFO) << "FT: Not RE-resending due to NULL master PID";
+ return;
+ } else
+ Process::post(master, ID, msgStr.data(), msgStr.size());
+ }
+
+ void gotAck(string ftId) {
+ DLOG(INFO) << "FT: Got ack, deleting outstanding msg " << ftId;
+ outMsgs.erase(ftId);
+ }
+
+ void sendOutstanding() {
+ if (!master) {
+ DLOG(INFO) << "FT: Not RE-resending due to NULL master PID";
+ return;
+ }
+
+ foreachpair( const string &ftId, struct StoredMsg &msg, outMsgs) {
+ if (msg.count < FT_MAX_RESENDS) {
+ DLOG(INFO) << "FT: RE-sending " << msg.ftId << " attempt:" << msg.count;
+ Process::post(master, msg.id, msg.data.data(), msg.data.size());
+ msg.count++;
+ } else {
+ DLOG(INFO) << "FT: Not RE-sending " << msg.ftId << " reached limit " << FT_MAX_RESENDS;
+ outMsgs.erase(ftId);
+ }
+ }
+ }
+
+ // Careful: not idempotent function.
+ bool acceptMessage(string from, string ftId) {
+ if (inMsgs.find(from)==inMsgs.end()) {
+ DLOG(INFO) << "FT: new msgs seq: " << ftId;
+ inMsgs[from] = ftId;
+ return true;
+ } else {
+ string oldSeq = inMsgs[from];
+ string oldRnd = oldSeq;
+ int pos;
+ if ((pos=oldSeq.find_last_of(':'))!=string::npos ) {
+ oldSeq.erase(0,pos+1);
+ oldRnd.erase(pos,255);
+ long seqNr = lexical_cast<long>(oldSeq);
+ string nextFtId = oldRnd+":"+lexical_cast<string>(seqNr+1);
+ if (nextFtId==ftId) {
+ DLOG(INFO) << "FT: match - got ftId:" << ftId << " expecting " << nextFtId;
+ inMsgs[from] = nextFtId;
+ return true;
+ } else {
+ DLOG(INFO) << "FT: mismatch - got ftId:" << ftId << " expecting " << nextFtId;
+ return false;
+ }
+ } else {
+ DLOG(INFO) << "FT: Error parsing ftId in acceptMessage for ftId:" << ftId;
+ return false;
+ }
+ }
+ }
+
+ string getNextId();
+
+ void setMasterPid(const PID &mPid) {
+ master = mPid;
+ }
+
+private:
+
+ PID master;
+
+ unordered_map<string, StoredMsg> outMsgs;
+
+ unordered_map<string, string> inMsgs;
+
+ string uniqPrefix;
+
+ long msgId;
+
+ static FTMessaging *instance;
+
+ FTMessaging();
+ FTMessaging(PID _master);
+
+ FTMessaging(FTMessaging const ©) {}
+ FTMessaging &operator= (FTMessaging const ©) {}
+};
+
+}}
+
+#endif /* _FT_MESSAGING_HPP_ */
Modified: incubator/mesos/trunk/src/master.cpp
URL: http://svn.apache.org/viewvc/incubator/mesos/trunk/src/master.cpp?rev=1131582&r1=1131581&r2=1131582&view=diff
==============================================================================
--- incubator/mesos/trunk/src/master.cpp (original)
+++ incubator/mesos/trunk/src/master.cpp Sun Jun 5 03:25:52 2011
@@ -2,6 +2,7 @@
#include "master.hpp"
#include "master_webui.hpp"
#include "allocator_factory.hpp"
+#include "ft_messaging.hpp"
#include <glog/logging.h>
@@ -124,6 +125,7 @@ Master::Master(const string zk)
exit(1);
}
}
+ ftMsg = FTMessaging::getInstance();
}
@@ -141,6 +143,7 @@ Master::Master(const string& _allocatorT
exit(1);
}
}
+ ftMsg = FTMessaging::getInstance();
}
@@ -281,7 +284,7 @@ void Master::operator () ()
LOG(INFO) << "Master started at nexus://" << self();
if (isFT) {
- LOG(INFO) << "Conencting to ZooKeeper at "<<zkservers;
+ LOG(INFO) << "Connecting to ZooKeeper at "<<zkservers;
ostringstream lpid;
lpid<<self();
leaderDetector = new LeaderDetector(zkservers, true, lpid.str());
@@ -463,6 +466,48 @@ void Master::operator () ()
break;
}
+ case S2M_FT_STATUS_UPDATE: {
+ string ftId;
+ SlaveID sid;
+ FrameworkID fid;
+ TaskID tid;
+ TaskState state;
+ string data;
+ string 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
+
+ DLOG(INFO) << "FT: relaying ftId:"<< ftId << " from: "<< senderStr;
+ send(framework->pid, pack<M2F_FT_STATUS_UPDATE>(ftId, senderStr, tid, state, data));
+
+ if (!ftMsg->acceptMessage(senderStr, ftId)) {
+ LOG(WARNING) << "Locally ignoring duplicate message with id:" << ftId;
+ } else {
+ // 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_STATUS_UPDATE: {
SlaveID sid;
FrameworkID fid;
@@ -488,8 +533,8 @@ void Master::operator () ()
}
} else
DLOG(INFO) << "S2M_STATUS_UPDATE error: couldn't lookup slave id" << sid;
- break;
}
+ break;
}
case S2M_FRAMEWORK_MESSAGE: {
@@ -548,10 +593,29 @@ void Master::operator () ()
framework->removeExpiredFilters();
allocator->timerTick();
- int cnts=0;
- foreachpair(_, Framework *framework, frameworks) {
- LOG(INFO)<<(cnts++)<<" resourceInUse:"<<framework->resources;
+ // 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 << " FT_ACK sent 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_ACK>(ftId));
+
break;
}
Modified: incubator/mesos/trunk/src/master.hpp
URL: http://svn.apache.org/viewvc/incubator/mesos/trunk/src/master.hpp?rev=1131582&r1=1131581&r2=1131582&view=diff
==============================================================================
--- incubator/mesos/trunk/src/master.hpp (original)
+++ incubator/mesos/trunk/src/master.hpp Sun Jun 5 03:25:52 2011
@@ -30,6 +30,7 @@
#include "leader_detector.hpp"
#include "task_info.hpp"
#include "url_processor.hpp"
+#include "ft_messaging.hpp"
namespace nexus { namespace internal { namespace master {
@@ -278,11 +279,12 @@ protected:
Allocator *allocator;
long masterId; // Used to differentiate different master in FT mode, will be ephemeral id
+ FTMessaging *ftMsg;
public:
- Master(const string ="");
+ Master(const string zk="");
- Master(const string& _allocatorType, const string ="");
+ Master(const string& _allocatorType, const string zk="");
~Master();
Modified: incubator/mesos/trunk/src/messages.hpp
URL: http://svn.apache.org/viewvc/incubator/mesos/trunk/src/messages.hpp?rev=1131582&r1=1131581&r2=1131582&view=diff
==============================================================================
--- incubator/mesos/trunk/src/messages.hpp (original)
+++ incubator/mesos/trunk/src/messages.hpp Sun Jun 5 03:25:52 2011
@@ -32,15 +32,21 @@ enum MessageType {
M2F_SLOT_OFFER,
M2F_RESCIND_OFFER,
M2F_STATUS_UPDATE,
+ M2F_FT_STATUS_UPDATE,
M2F_LOST_SLAVE,
M2F_FRAMEWORK_MESSAGE,
M2F_ERROR,
+ /* Used for FT. */
+ FT_RELAY_ACK,
+ FT_ACK,
+
/* From slave to master. */
S2M_REGISTER_SLAVE,
S2M_REREGISTER_SLAVE,
S2M_UNREGISTER_SLAVE,
S2M_STATUS_UPDATE,
+ S2M_FT_STATUS_UPDATE,
S2M_FRAMEWORK_MESSAGE,
S2M_LOST_EXECUTOR,
@@ -152,6 +158,13 @@ TUPLE(M2F_STATUS_UPDATE,
TaskState,
std::string));
+TUPLE(M2F_FT_STATUS_UPDATE,
+ (std::string, /* FT ID */
+ std::string, /* original sender */
+ TaskID,
+ TaskState,
+ std::string));
+
TUPLE(M2F_LOST_SLAVE,
(SlaveID));
@@ -162,6 +175,19 @@ TUPLE(M2F_ERROR,
(int32_t /*code*/,
std::string /*msg*/));
+
+
+TUPLE(FT_RELAY_ACK,
+ (std::string, /* FT ID */
+ std::string /* PID of orig */
+ ));
+
+TUPLE(FT_ACK,
+ (std::string /* FT ID */
+ ));
+
+
+
TUPLE(S2M_REGISTER_SLAVE,
(std::string /*name*/,
std::string /*publicDns*/,
@@ -184,6 +210,15 @@ TUPLE(S2M_STATUS_UPDATE,
TaskState,
std::string));
+TUPLE(S2M_FT_STATUS_UPDATE,
+ (std::string, /* unique msgId */
+ std::string, /* senders PID */
+ SlaveID,
+ FrameworkID,
+ TaskID,
+ TaskState,
+ std::string));
+
TUPLE(S2M_FRAMEWORK_MESSAGE,
(SlaveID,
FrameworkID,
Modified: incubator/mesos/trunk/src/nexus_sched.cpp
URL: http://svn.apache.org/viewvc/incubator/mesos/trunk/src/nexus_sched.cpp?rev=1131582&r1=1131581&r2=1131582&view=diff
==============================================================================
--- incubator/mesos/trunk/src/nexus_sched.cpp (original)
+++ incubator/mesos/trunk/src/nexus_sched.cpp Sun Jun 5 03:25:52 2011
@@ -28,6 +28,7 @@
#include "nexus_sched.hpp"
#include "url_processor.hpp"
#include "leader_detector.hpp"
+#include "ft_messaging.hpp"
using std::cerr;
using std::cout;
@@ -70,6 +71,7 @@ private:
bool isFT;
string zkservers;
LeaderDetector *leaderDetector;
+ FTMessaging *ftMsg;
volatile bool terminate;
@@ -126,6 +128,7 @@ public:
exit(1);
}
}
+ ftMsg = FTMessaging::getInstance();
}
protected:
@@ -193,6 +196,27 @@ protected:
break;
}
+ case M2F_FT_STATUS_UPDATE: {
+ string ftId;
+ string origPid;
+ TaskID tid;
+ TaskState state;
+ string data;
+ unpack<M2F_FT_STATUS_UPDATE>(ftId, origPid, tid, state, data);
+ if (!ftMsg->acceptMessage(origPid, ftId)) {
+ LOG(WARNING) << "Dropping duplicate message with id:" << ftId;
+ break;
+ }
+ DLOG(INFO) << "Received fault tolerant message with id: " << ftId;
+ DLOG(INFO) << "Sending FT_RELAY_ACK for : " << ftId;
+ send(from(), pack<FT_RELAY_ACK>(ftId, origPid));
+
+ TaskStatus status(tid, state, data);
+ invoke(bind(&Scheduler::statusUpdate, sched, driver, ref(status)));
+ break;
+ }
+
+
case M2F_STATUS_UPDATE: {
TaskID tid;
TaskState state;
Modified: incubator/mesos/trunk/src/slave.cpp
URL: http://svn.apache.org/viewvc/incubator/mesos/trunk/src/slave.cpp?rev=1131582&r1=1131581&r2=1131582&view=diff
==============================================================================
--- incubator/mesos/trunk/src/slave.cpp (original)
+++ incubator/mesos/trunk/src/slave.cpp Sun Jun 5 03:25:52 2011
@@ -7,6 +7,8 @@
#include "isolation_module_factory.hpp"
#include "url_processor.hpp"
+#define FT_TIMEOUT 10
+
using std::list;
using std::make_pair;
using std::ostringstream;
@@ -66,6 +68,7 @@ Slave::Slave(const string &_master, Reso
resources(_resources), local(_local), id("-1"),
isolationType("process"), isolationModule(NULL), slaveLeaderListener(this, getPID())
{
+ ftMsg = FTMessaging::getInstance();
pair<UrlProcessor::URLType, string> urlPair = UrlProcessor::process(_master);
if (urlPair.first == UrlProcessor::ZOO) {
isFT=true;
@@ -90,6 +93,7 @@ Slave::Slave(const string &_master, Reso
resources(_resources), local(_local), id("-1"),
isolationType(_isolationType), isolationModule(NULL), slaveLeaderListener(this, getPID())
{
+ ftMsg = FTMessaging::getInstance();
pair<UrlProcessor::URLType, string> urlPair = UrlProcessor::process(_master);
if (urlPair.first == UrlProcessor::ZOO) {
isFT=true;
@@ -171,6 +175,9 @@ void Slave::operator () ()
LOG(INFO) << "Connecting to Nexus master at " << master;
link(master);
+
+ ftMsg->setMasterPid(master);
+
send(master, pack<S2M_REGISTER_SLAVE>(hostname, publicDns, resources));
FrameworkID fid;
@@ -181,7 +188,7 @@ void Slave::operator () ()
string data;
while (true) {
- switch (receive()) {
+ switch (receive(FT_TIMEOUT)) {
case M2S_REGISTER_REPLY: {
unpack<M2S_REGISTER_REPLY>(this->id);
LOG(INFO) << "Registered with master; given slave ID " << this->id;
@@ -306,7 +313,9 @@ void Slave::operator () ()
}
}
// Pass on the update to the master
- send(master, pack<S2M_STATUS_UPDATE>(id, fid, tid, taskState, data));
+ string ftId = ftMsg->getNextId();
+ ftMsg->reliableSend(ftId, pack<S2M_FT_STATUS_UPDATE>(ftId, self(), id, fid, tid, taskState, data));
+ // send(master, pack<S2M_STATUS_UPDATE>(id, fid, tid, taskState, data));
break;
}
@@ -381,6 +390,8 @@ void Slave::operator () ()
LOG(INFO) << "Connecting to Nexus master at " << master;
link(master);
+ ftMsg->setMasterPid(master);
+
// reconstruct resourcesInUse for the master
// alig: do I need to include queuedTasks in this number? Don't think so.
Resources resourcesInUse;
@@ -400,7 +411,21 @@ void Slave::operator () ()
break;
}
+
+ case FT_ACK: {
+ string ftId;
+ unpack<FT_ACK>(ftId);
+
+ 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=1131582&r1=1131581&r2=1131582&view=diff
==============================================================================
--- incubator/mesos/trunk/src/slave.hpp (original)
+++ incubator/mesos/trunk/src/slave.hpp Sun Jun 5 03:25:52 2011
@@ -36,6 +36,7 @@
#include "slave_state.hpp"
#include "leader_detector.hpp"
#include "task_info.hpp"
+#include "ft_messaging.hpp"
namespace nexus { namespace internal { namespace slave {
@@ -174,6 +175,7 @@ public:
ExecutorMap executors; // Invariant: framework will exist if executor exists
string isolationType;
IsolationModule *isolationModule;
+ FTMessaging *ftMsg;
class SlaveLeaderListener;
Modified: incubator/mesos/trunk/src/third_party/libprocess/tuple-impl.hpp
URL: http://svn.apache.org/viewvc/incubator/mesos/trunk/src/third_party/libprocess/tuple-impl.hpp?rev=1131582&r1=1131581&r2=1131582&view=diff
==============================================================================
--- incubator/mesos/trunk/src/third_party/libprocess/tuple-impl.hpp (original)
+++ incubator/mesos/trunk/src/third_party/libprocess/tuple-impl.hpp Sun Jun 5 03:25:52 2011
@@ -868,7 +868,7 @@ protected:
public:
template <MSGID ID>
- static void post(const PID &, const tuple<ID> &r)
+ static void post(const PID &to, const tuple<ID> &r)
{
std::ostringstream os;
@@ -929,4 +929,64 @@ public:
Process::post(to, ID, data.data(), data.size());
}
+ template <MSGID ID>
+ static string tupleToString(const tuple<ID> &r)
+ {
+ std::ostringstream os;
+
+ process::serialization::serializer s(os);
+
+ if (size<ID>::value >= 1) {
+ const typename field<0, ID>::type &temp = at<0>(r);
+ s & temp;
+ }
+
+ if (size<ID>::value >= 2) {
+ const typename field<1, ID>::type &temp = at<1>(r);
+ s & temp;
+ }
+
+ if (size<ID>::value >= 3) {
+ const typename field<2, ID>::type &temp = at<2>(r);
+ s & temp;
+ }
+
+ if (size<ID>::value >= 4) {
+ const typename field<3, ID>::type &temp = at<3>(r);
+ s & temp;
+ }
+
+ if (size<ID>::value >= 5) {
+ const typename field<4, ID>::type &temp = at<4>(r);
+ s & temp;
+ }
+
+ if (size<ID>::value >= 6) {
+ const typename field<5, ID>::type &temp = at<5>(r);
+ s & temp;
+ }
+
+ if (size<ID>::value >= 7) {
+ const typename field<6, ID>::type &temp = at<6>(r);
+ s & temp;
+ }
+
+ if (size<ID>::value >= 8) {
+ const typename field<7, ID>::type &temp = at<7>(r);
+ s & temp;
+ }
+
+ if (size<ID>::value >= 9) {
+ const typename field<8, ID>::type &temp = at<8>(r);
+ s & temp;
+ }
+
+ if (size<ID>::value >= 10) {
+ const typename field<9, ID>::type &temp = at<9>(r);
+ s & temp;
+ }
+
+ return os.str();
+
+ }
};