You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@mesos.apache.org by vi...@apache.org on 2013/10/24 23:01:30 UTC
git commit: Fixed slave to properly recover latest run of completed
executors.
Updated Branches:
refs/heads/master 4bbcbae08 -> 7549350f5
Fixed slave to properly recover latest run of completed executors.
Review: https://reviews.apache.org/r/14910
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/7549350f
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/7549350f
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/7549350f
Branch: refs/heads/master
Commit: 7549350f52f7c02f5629abb11f09df083b17b293
Parents: 4bbcbae
Author: Vinod Kone <vi...@twitter.com>
Authored: Thu Oct 24 10:56:19 2013 -0700
Committer: Vinod Kone <vi...@twitter.com>
Committed: Thu Oct 24 13:54:08 2013 -0700
----------------------------------------------------------------------
src/slave/slave.cpp | 75 +++++++++++++++++++++---------------------------
1 file changed, 32 insertions(+), 43 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/7549350f/src/slave/slave.cpp
----------------------------------------------------------------------
diff --git a/src/slave/slave.cpp b/src/slave/slave.cpp
index 58b16d4..5272600 100644
--- a/src/slave/slave.cpp
+++ b/src/slave/slave.cpp
@@ -853,7 +853,6 @@ void Slave::_runTask(
return;
}
-
if (!future.isReady()) {
LOG(ERROR) << "Failed to unschedule directories scheduled for gc: "
<< (future.isFailed() ? future.failure() : "future discarded");
@@ -1145,9 +1144,6 @@ void Slave::shutdownFramework(const FrameworkID& frameworkId)
framework->state = Framework::TERMINATING;
// Shut down all executors of this framework.
- // NOTE: If there are no executors but 'pending' tasks, the
- // framework will be removed and all its tasks are appropriately
- // handled in '_runTask()'.
// NOTE: We use 'executors.keys()' here because 'shutdownExecutor'
// and 'removeExecutor' can remove an executor from 'executors'.
foreach (const ExecutorID& executorId, framework->executors.keys()) {
@@ -1170,6 +1166,11 @@ void Slave::shutdownFramework(const FrameworkID& frameworkId)
// Executor is terminating. Ignore.
}
}
+
+ // Remove this framework if it has no pending executors and tasks.
+ if (framework->executors.empty() && framework->pending.empty()) {
+ removeFramework(framework);
+ }
break;
default:
LOG(FATAL) << "Framework " << frameworkId
@@ -1384,6 +1385,11 @@ void Slave::_statusUpdateAcknowledgement(
if (executor->state == Executor::TERMINATED && !executor->incompleteTasks()) {
removeExecutor(framework, executor);
}
+
+ // Remove this framework if it has no pending executors and tasks.
+ if (framework->executors.empty() && framework->pending.empty()) {
+ removeFramework(framework);
+ }
}
@@ -2220,6 +2226,11 @@ void Slave::executorTerminated(
!executor->incompleteTasks()) {
removeExecutor(framework, executor);
}
+
+ // Remove this framework if it has no pending executors and tasks.
+ if (framework->executors.empty() && framework->pending.empty()) {
+ removeFramework(framework);
+ }
break;
}
default:
@@ -2300,11 +2311,6 @@ void Slave::removeExecutor(Framework* framework, Executor* executor)
}
framework->destroyExecutor(executor->id);
-
- // Remove this framework if it has no pending executors and tasks.
- if (framework->executors.empty() && framework->pending.empty()) {
- removeFramework(framework);
- }
}
@@ -2817,6 +2823,11 @@ void Slave::recoverFramework(const FrameworkState& state)
&Self::fileAttached,
lambda::_1,
executor->directory));
+
+ // Remove the executor if it's terminated.
+ if (executor->state == Executor::TERMINATED) {
+ removeExecutor(framework, executor);
+ }
}
// Remove the framework in case we didn't recover any executors.
@@ -2998,22 +3009,23 @@ Executor* Framework::recoverExecutor(const ExecutorState& state)
// We are only interested in the latest run of the executor!
// So, we GC all the old runs.
+ // NOTE: We don't schedule the top level executor work and meta
+ // directories for GC here, because they will be scheduled when
+ // the latest executor run terminates.
const UUID& uuid = state.latest.get();
foreachvalue (const RunState& run, state.runs) {
CHECK_SOME(run.id);
const UUID& runId = run.id.get();
if (uuid != runId) {
// GC the executor run's work directory.
+ // TODO(vinod): Expose this directory to webui by recovering the
+ // tasks and doing a 'files->attach()'.
slave->garbageCollect(paths::getExecutorRunPath(
slave->flags.work_dir, slave->info.id(), id, state.id, runId));
// GC the executor run's meta directory.
slave->garbageCollect(paths::getExecutorRunPath(
slave->metaDir, slave->info.id(), id, state.id, runId));
-
- // NOTE: We don't schedule the top level executor work and meta
- // directories for GC here, because they will be scheduled when
- // the latest executor run terminates.
}
}
@@ -3023,36 +3035,6 @@ Executor* Framework::recoverExecutor(const ExecutorState& state)
const RunState& run = state.runs.get(uuid).get();
- // If the latest run of the executor was completed (i.e., terminated
- // and all updates are acknowledged) in the previous run, we don't
- // recover it.
- if (run.completed) {
- VLOG(1) << "Skipping recovery of executor " << state.id
- << " of framework " << id
- << " because its latest run " << uuid << " is completed";
-
- CHECK_SOME(run.id);
- const UUID& runId = run.id.get();
-
- // GC the executor run's work directory.
- slave->garbageCollect(paths::getExecutorRunPath(
- slave->flags.work_dir, slave->info.id(), id, state.id, runId));
-
- // GC the executor run's meta directory.
- slave->garbageCollect(paths::getExecutorRunPath(
- slave->metaDir, slave->info.id(), id, state.id, runId));
-
- // GC the top level executor work directory.
- slave->garbageCollect(paths::getExecutorPath(
- slave->flags.work_dir, slave->info.id(), id, state.id));
-
- // GC the top level executor meta directory.
- slave->garbageCollect(paths::getExecutorPath(
- slave->metaDir, slave->info.id(), id, state.id));
-
- return NULL;
- }
-
// Create executor.
const string& directory = paths::getExecutorRunPath(
slave->flags.work_dir, slave->info.id(), id, state.id, uuid);
@@ -3074,6 +3056,13 @@ Executor* Framework::recoverExecutor(const ExecutorState& state)
executor->pid = run.libprocessPid.get();
}
+ // If the latest run of the executor was completed (i.e., terminated
+ // and all updates are acknowledged) in the previous run, we
+ // transition its state to 'TERMINATED'.
+ if (run.completed) {
+ executor->state = Executor::TERMINATED;
+ }
+
// And finally recover all the executor's tasks.
foreachvalue (const TaskState& taskState, run.tasks) {
executor->recoverTask(taskState);