You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@mesos.apache.org by bm...@apache.org on 2013/10/09 03:37:10 UTC
[2/2] git commit: Fixed a crash in the master during framework
re-registration.
Fixed a crash in the master during framework re-registration.
Review: https://reviews.apache.org/r/14548
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/8077ad40
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/8077ad40
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/8077ad40
Branch: refs/heads/master
Commit: 8077ad40f2e2ec1912425c3cfc2a64e32b86881d
Parents: 0eb61a3
Author: Benjamin Mahler <bm...@twitter.com>
Authored: Tue Oct 8 16:47:56 2013 -0700
Committer: Benjamin Mahler <bm...@twitter.com>
Committed: Tue Oct 8 18:36:02 2013 -0700
----------------------------------------------------------------------
src/master/master.cpp | 22 +++----
src/slave/slave.cpp | 45 ++++++-------
src/tests/fault_tolerance_tests.cpp | 109 ++++++++++++++++++++++++++++++-
3 files changed, 137 insertions(+), 39 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/8077ad40/src/master/master.cpp
----------------------------------------------------------------------
diff --git a/src/master/master.cpp b/src/master/master.cpp
index cdfae1d..2fd48a6 100644
--- a/src/master/master.cpp
+++ b/src/master/master.cpp
@@ -794,18 +794,16 @@ void Master::reregisterFramework(const FrameworkInfo& frameworkInfo,
foreachvalue (Task* task, slave->tasks[frameworkId]) {
if (framework->id == task->framework_id()) {
framework->addTask(task);
- // Also add the task's executor for resource accounting.
- if (task->has_executor_id()) {
- if (!framework->hasExecutor(slave->id, task->executor_id())) {
- CHECK(slave->hasExecutor(framework->id, task->executor_id()))
- << "Unknown executor " << task->executor_id()
- << " of framework " << framework->id
- << " for the task " << task->task_id();
-
- const ExecutorInfo& executorInfo =
- slave->executors[framework->id][task->executor_id()];
- framework->addExecutor(slave->id, executorInfo);
- }
+
+ // Also add the task's executor for resource accounting
+ // if it's still alive on the slave and we've not yet
+ // added it to the framework.
+ if (task->has_executor_id() &&
+ slave->hasExecutor(framework->id, task->executor_id()) &&
+ !framework->hasExecutor(slave->id, task->executor_id())) {
+ const ExecutorInfo& executorInfo =
+ slave->executors[framework->id][task->executor_id()];
+ framework->addExecutor(slave->id, executorInfo);
}
}
}
http://git-wip-us.apache.org/repos/asf/mesos/blob/8077ad40/src/slave/slave.cpp
----------------------------------------------------------------------
diff --git a/src/slave/slave.cpp b/src/slave/slave.cpp
index 1dc2189..debb2f4 100644
--- a/src/slave/slave.cpp
+++ b/src/slave/slave.cpp
@@ -723,9 +723,28 @@ void Slave::doReliableRegistration()
continue;
}
+ // Add launched, terminated, and queued tasks.
+ foreach (Task* task, executor->launchedTasks.values()) {
+ message.add_tasks()->CopyFrom(*task);
+ }
+ foreach (Task* task, executor->terminatedTasks.values()) {
+ message.add_tasks()->CopyFrom(*task);
+ }
+ foreach (const TaskInfo& task, executor->queuedTasks.values()) {
+ message.add_tasks()->CopyFrom(protobuf::createTask(
+ task, TASK_STAGING, executor->id, framework->id));
+ }
+
// Do not re-register with Command Executors because the
// master doesn't store them; they are generated by the slave.
- if (!executor->commandExecutor) {
+ if (executor->commandExecutor) {
+ // NOTE: We have to unset the executor id here for the task
+ // because the master uses the absence of task.executor_id()
+ // to detect command executors.
+ for (int i = 0; i < message.tasks_size(); ++i) {
+ message.mutable_tasks(i)->clear_executor_id();
+ }
+ } else {
ExecutorInfo* executorInfo = message.add_executor_infos();
executorInfo->MergeFrom(executor->info);
@@ -735,30 +754,6 @@ void Slave::doReliableRegistration()
// it a required field.
executorInfo->mutable_framework_id()->MergeFrom(framework->id);
}
-
- // Add launched tasks.
- // TODO(vinod): Use foreachvalue instead once LinkedHashmap
- // supports it.
- foreach (Task* task, executor->launchedTasks.values()) {
- message.add_tasks()->CopyFrom(*task);
- }
-
- // Add queued tasks.
- // TODO(vinod): Use foreachvalue instead once LinkedHashmap
- // supports it.
- foreach (const TaskInfo& task, executor->queuedTasks.values()) {
- const Task& t = protobuf::createTask(
- task, TASK_STAGING, executor->id, framework->id);
-
- message.add_tasks()->CopyFrom(t);
- }
-
- // Add terminated tasks.
- // TODO(vinod): Use foreachvalue instead once LinkedHashmap
- // supports it.
- foreach (Task* task, executor->terminatedTasks.values()) {
- message.add_tasks()->CopyFrom(*task);
- }
}
}
send(master, message);
http://git-wip-us.apache.org/repos/asf/mesos/blob/8077ad40/src/tests/fault_tolerance_tests.cpp
----------------------------------------------------------------------
diff --git a/src/tests/fault_tolerance_tests.cpp b/src/tests/fault_tolerance_tests.cpp
index 254eae4..35b743a 100644
--- a/src/tests/fault_tolerance_tests.cpp
+++ b/src/tests/fault_tolerance_tests.cpp
@@ -998,8 +998,7 @@ TEST_F(FaultToleranceTest, SchedulerFailoverStatusUpdate)
vector<TaskInfo> tasks;
tasks.push_back(task);
- EXPECT_CALL(exec, registered(_, _, _, _))
- .Times(1);
+ EXPECT_CALL(exec, registered(_, _, _, _));
EXPECT_CALL(exec, launchTask(_, _))
.WillOnce(SendStatusUpdateFromTask(TASK_RUNNING));
@@ -1061,6 +1060,112 @@ TEST_F(FaultToleranceTest, SchedulerFailoverStatusUpdate)
}
+// This test was added to ensure MESOS-420 is fixed.
+// We need to make sure that the master correctly handles non-terminal
+// tasks with exited executors upon framework re-registration. This is
+// possible because the ExitedExecutor message can arrive before the
+// terminal status update(s) of its task(s).
+TEST_F(FaultToleranceTest, ReregisterFrameworkExitedExecutor)
+{
+ // First we'll start a master and slave, then register a framework
+ // so we can launch a task.
+ Try<PID<Master> > master = StartMaster();
+ ASSERT_SOME(master);
+
+ MockExecutor exec(DEFAULT_EXECUTOR_ID);
+ TestingIsolator isolator(&exec);
+ Try<PID<Slave> > slave = StartSlave(&isolator);
+ ASSERT_SOME(slave);
+
+ MockScheduler sched;
+ MesosSchedulerDriver driver(&sched, DEFAULT_FRAMEWORK_INFO, master.get());
+
+ Future<process::Message> frameworkRegisteredMessage =
+ FUTURE_MESSAGE(Eq(FrameworkRegisteredMessage().GetTypeName()), _, _);
+
+ FrameworkID frameworkId;
+ EXPECT_CALL(sched, registered(&driver, _, _))
+ .WillOnce(SaveArg<1>(&frameworkId));
+
+ EXPECT_CALL(sched, resourceOffers(&driver, _))
+ .WillOnce(LaunchTasks(1, 1, 16, "*"));
+
+ Future<Nothing> statusUpdate;
+ EXPECT_CALL(sched, statusUpdate(&driver, _))
+ .WillOnce(FutureSatisfy(&statusUpdate)); // TASK_RUNNING.
+
+ EXPECT_CALL(exec, registered(_, _, _, _));
+
+ EXPECT_CALL(exec, launchTask(_, _))
+ .WillOnce(SendStatusUpdateFromTask(TASK_RUNNING));
+
+ driver.start();
+
+ AWAIT_READY(frameworkRegisteredMessage);
+
+ // Wait until TASK_RUNNING of the task is received.
+ AWAIT_READY(statusUpdate);
+
+ EXPECT_CALL(sched, disconnected(&driver));
+
+ // Now that the task is launched, we need to induce the following:
+ // 1. ExitedExecutorMessage received by the master prior to a
+ // terminal status update for the corresponding task. This
+ // means we need to drop the status update coming from the
+ // slave.
+ // 2. Framework re-registration.
+ //
+ // To achieve this, we need to:
+ // 1. Restart the master (the slave / framework will not detect
+ // the new master automatically using the BasicMasterDetector).
+ // 2. Notify the slave of the new master.
+ // 3. Kill the executor.
+ // 4. Drop the status update, but allow the ExitedExecutorMessage.
+ // 5. Notify the framework of the new master.
+ Stop(master.get());
+ master = StartMaster();
+ ASSERT_SOME(master);
+
+ // Simulate a new master detected message to the slave.
+ NewMasterDetectedMessage newMasterDetectedMsg;
+ newMasterDetectedMsg.set_pid(master.get());
+
+ Future<SlaveReregisteredMessage> slaveReregisteredMessage =
+ FUTURE_PROTOBUF(SlaveReregisteredMessage(), _, _);
+
+ process::post(slave.get(), newMasterDetectedMsg);
+
+ // Wait for the slave to re-register.
+ AWAIT_READY(slaveReregisteredMessage);
+
+ // Allow the executor exited message and drop the status update.
+ Future<ExitedExecutorMessage> executorExitedMessage =
+ FUTURE_PROTOBUF(ExitedExecutorMessage(), _, _);
+ Future<StatusUpdateMessage> statusUpdateMessage =
+ DROP_PROTOBUF(StatusUpdateMessage(), _, _);
+
+ // Now kill the executor.
+ dispatch(isolator, &Isolator::killExecutor, frameworkId, DEFAULT_EXECUTOR_ID);
+
+ AWAIT_READY(executorExitedMessage);
+ AWAIT_READY(statusUpdateMessage);
+
+ // Now notify the framework of the new master.
+ Future<FrameworkRegisteredMessage> frameworkRegisteredMessage2 =
+ FUTURE_PROTOBUF(FrameworkRegisteredMessage(), _, _);
+
+ EXPECT_CALL(sched, registered(&driver, _, _));
+
+ process::post(frameworkRegisteredMessage.get().to, newMasterDetectedMsg);
+
+ AWAIT_READY(frameworkRegisteredMessage2);
+
+ driver.stop();
+ driver.join();
+ Shutdown();
+}
+
+
TEST_F(FaultToleranceTest, ForwardStatusUpdateUnknownExecutor)
{
Try<PID<Master> > master = StartMaster();