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/03/21 20:25:09 UTC
svn commit: r1459483 - in /incubator/mesos/trunk/src: slave/slave.cpp
slave/slave.hpp slave/state.cpp slave/state.hpp
slave/status_update_manager.cpp slave/status_update_manager.hpp
tests/gc_tests.cpp tests/slave_recovery_tests.cpp
Author: vinodkone
Date: Thu Mar 21 19:25:09 2013
New Revision: 1459483
URL: http://svn.apache.org/r1459483
Log:
Fixed slave to delete an executor after all its pending updates have
been acknowledged.
Review: https://reviews.apache.org/r/10019
Modified:
incubator/mesos/trunk/src/slave/slave.cpp
incubator/mesos/trunk/src/slave/slave.hpp
incubator/mesos/trunk/src/slave/state.cpp
incubator/mesos/trunk/src/slave/state.hpp
incubator/mesos/trunk/src/slave/status_update_manager.cpp
incubator/mesos/trunk/src/slave/status_update_manager.hpp
incubator/mesos/trunk/src/tests/gc_tests.cpp
incubator/mesos/trunk/src/tests/slave_recovery_tests.cpp
Modified: incubator/mesos/trunk/src/slave/slave.cpp
URL: http://svn.apache.org/viewvc/incubator/mesos/trunk/src/slave/slave.cpp?rev=1459483&r1=1459482&r2=1459483&view=diff
==============================================================================
--- incubator/mesos/trunk/src/slave/slave.cpp (original)
+++ incubator/mesos/trunk/src/slave/slave.cpp Thu Mar 21 19:25:09 2013
@@ -992,13 +992,14 @@ void Slave::statusUpdateAcknowledgement(
<< " for task " << taskId
<< " of framework " << frameworkId;
- statusUpdateManager->acknowledgement(taskId, frameworkId, uuid)
+ statusUpdateManager->acknowledgement(
+ taskId, frameworkId, UUID::fromBytes(uuid))
.onAny(defer(self(),
&Slave::_statusUpdateAcknowledgement,
params::_1,
taskId,
frameworkId,
- uuid));
+ UUID::fromBytes(uuid)));
}
@@ -1006,10 +1007,10 @@ void Slave::_statusUpdateAcknowledgement
const Future<Try<Nothing> >& future,
const TaskID& taskId,
const FrameworkID& frameworkId,
- const string& uuid)
+ const UUID& uuid)
{
if (!future.isReady()) {
- LOG(FATAL) << "Failed to handle status update acknowledgement " << uuid
+ LOG(FATAL) << "Failed to handle status update acknowledgement"
<< " for task " << taskId
<< " of framework " << frameworkId
<< (future.isFailed() ? future.failure() : "future discarded");
@@ -1017,12 +1018,36 @@ void Slave::_statusUpdateAcknowledgement
}
if (future.get().isError()) {
- LOG(ERROR) << "Failed to handle the status update acknowledgement " << uuid
- << " for task " << taskId
- << " of framework " << frameworkId
- << future.get().error();
+ LOG(ERROR) << "Failed to handle the status update acknowledgement"
+ << " for task " << taskId
+ << " of framework " << frameworkId
+ << future.get().error();
+ return;
+ }
+
+ LOG(INFO) << "Status update manager successfully handled status update"
+ << " acknowledgement for task " << taskId
+ << " of framework " << frameworkId;
+
+ Framework* framework = getFramework(frameworkId);
+ if (framework == NULL) {
+ LOG(ERROR) << "Status update acknowledgement for task " << taskId
+ << " of unknown framework " << frameworkId;
+ return;
+ }
+
+ // Find the executor that has this update.
+ Executor* executor = framework->getExecutor(taskId);
+ if (executor == NULL) {
+ LOG(ERROR) << "Status update acknowledgement for task " << taskId
+ << " of unknown executor";
return;
}
+
+ executor->updates.remove(taskId, uuid);
+
+ // Cleanup the executor and framework, if possible.
+ cleanup(framework, executor);
}
@@ -1147,12 +1172,15 @@ void Slave::reregisterExecutor(
send(executor->pid, message);
// Handle all the pending updates.
- // NOTE: The status update manager might have already checkpointed some
- // of these pending updates (for e.g: if the slave died right after it
- // checkpointed the update but before it could send the ACK to the executor).
- // This is ok because, the status update manager simply re-ACKs the executor.
foreach (const StatusUpdate& update, updates) {
- statusUpdate(update); // This also updates the executor's resources!
+ // The status update manager might have already checkpointed some
+ // of these pending updates (for e.g: if the slave died right
+ // after it checkpointed the update but before it could send the
+ // ACK to the executor). If so, we can just ignore those updates.
+ if (!executor->updates.contains(
+ update.status().task_id(), UUID::fromBytes(update.uuid()))) {
+ statusUpdate(update); // This also updates the executor's resources!
+ }
}
// Now, if there is any task still in STAGING state and not in 'tasks' known
@@ -1223,6 +1251,7 @@ void Slave::statusUpdate(const StatusUpd
if (executor != NULL) {
executor->updateTaskState(status.task_id(), status.state());
+ executor->updates.put(status.task_id(), UUID::fromBytes(update.uuid()));
// Handle the task appropriately if it's terminated.
if (protobuf::isTerminalState(status.state())) {
@@ -1488,6 +1517,8 @@ void Slave::executorTerminated(
return;
}
+ executor->terminated = true;
+
bool isCommandExecutor = false;
// Transition all live tasks to TASK_LOST/TASK_FAILED.
@@ -1562,11 +1593,29 @@ void Slave::executorTerminated(
send(master, message);
}
- // Schedule the executor directory to get garbage collected.
- gc.schedule(flags.gc_delay, executor->directory)
- .onAny(defer(self(), &Self::detachFile, params::_1, executor->directory));
+ // Cleanup the executor and framework, if possible.
+ cleanup(framework, executor);
+}
- framework->destroyExecutor(executor->id);
+
+void Slave::cleanup(Framework* framework, Executor* executor)
+{
+ CHECK_NOTNULL(framework);
+ CHECK_NOTNULL(executor);
+
+ // Cleanup this executor if it has terminated and either has no
+ // pending updates or the framework is shutting down. We don't
+ // care for pending updates when framework is shutting down
+ // because the framework cannot ACK them.
+ if (executor->terminated &&
+ (executor->updates.empty() || framework->shutdown)) {
+
+ // Schedule the executor directory to get garbage collected.
+ gc.schedule(flags.gc_delay, executor->directory)
+ .onAny(defer(self(), &Self::detachFile, params::_1, executor->directory));
+
+ framework->destroyExecutor(executor->id);
+ }
// Cleanup if this framework has no executors running.
// TODO(vinod): If the framework is not being shutdown, remove
@@ -1578,9 +1627,8 @@ void Slave::executorTerminated(
completedFrameworks.push_back(std::tr1::shared_ptr<Framework>(framework));
}
- // If this slave is in 'recover=cleanup' mode, exit after all executors
- // have exited.
- // TODO(vinod): Ensure all status updates have been acknowledged.
+ // If this slave is in 'recover=cleanup' mode, exit after all
+ // executors have been removed.
if (flags.recover == "cleanup" && frameworks.size() == 0) {
cleanup();
}
@@ -1913,10 +1961,17 @@ Future<Nothing> Slave::recoverExecutors(
// Read updates to get the latest state of the task.
foreach (const StatusUpdate& update, task.updates) {
executor->updateTaskState(task.id, update.status().state());
+ executor->updates.put(task.id, UUID::fromBytes(update.uuid()));
- // Remove the task if it terminated.
+ // Remove the task if it received a terminal update.
if (protobuf::isTerminalState(update.status().state())) {
executor->removeTask(task.id);
+
+ // If the terminal update has been acknowledged, remove
+ // it from pending tasks.
+ if (task.acks.contains(UUID::fromBytes(update.uuid()))) {
+ executor->updates.remove(task.id, UUID::fromBytes(update.uuid()));
+ }
break;
}
}
Modified: incubator/mesos/trunk/src/slave/slave.hpp
URL: http://svn.apache.org/viewvc/incubator/mesos/trunk/src/slave/slave.hpp?rev=1459483&r1=1459482&r2=1459483&view=diff
==============================================================================
--- incubator/mesos/trunk/src/slave/slave.hpp (original)
+++ incubator/mesos/trunk/src/slave/slave.hpp Thu Mar 21 19:25:09 2013
@@ -32,6 +32,7 @@
#include <process/protobuf.hpp>
#include <stout/hashmap.hpp>
+#include <stout/multihashmap.hpp>
#include <stout/os.hpp>
#include <stout/path.hpp>
#include <stout/uuid.hpp>
@@ -153,7 +154,7 @@ public:
const Future<Try<Nothing> >& future,
const TaskID& taskId,
const FrameworkID& frameworkId,
- const std::string& uuid);
+ const UUID& uuid);
void executorStarted(
const FrameworkID& frameworkId,
@@ -230,6 +231,12 @@ protected:
const state::SlaveState& state,
bool reconnect);
+ // Called when an executor terminates or a status update
+ // acknowledgement is handled by the status update manager.
+ // This potentially removes the executor and framework and
+ // schedules them for garbage collection.
+ void cleanup(Framework* framework, Executor* executor);
+
// Called when the slave is started in 'cleanup' recovery mode and
// all the executors have terminated.
void cleanup();
@@ -314,6 +321,7 @@ struct Executor
uuid(_uuid),
pid(UPID()),
shutdown(false),
+ terminated(false),
resources(_info.resources()),
completedTasks(MAX_COMPLETED_TASKS_PER_EXECUTOR) {}
@@ -378,11 +386,18 @@ struct Executor
bool shutdown; // Indicates if executor is being shut down.
+ // Indicates if the executor has terminated. We need this
+ // because a terminated executor might still have pending
+ // status updates that are not yet acknowledged.
+ bool terminated;
+
Resources resources; // Currently consumed resources.
hashmap<TaskID, TaskInfo> queuedTasks;
hashmap<TaskID, Task*> launchedTasks;
+ multihashmap<TaskID, UUID> updates; // Pending updates.
+
boost::circular_buffer<Task> completedTasks;
private:
@@ -510,7 +525,8 @@ struct Framework
{
foreachvalue (Executor* executor, executors) {
if (executor->queuedTasks.contains(taskId) ||
- executor->launchedTasks.contains(taskId)) {
+ executor->launchedTasks.contains(taskId) ||
+ executor->updates.contains(taskId)) {
return executor;
}
}
@@ -531,10 +547,6 @@ struct Framework
// Up to MAX_COMPLETED_EXECUTORS_PER_FRAMEWORK completed executors.
boost::circular_buffer<std::tr1::shared_ptr<Executor> > completedExecutors;
-
- // Status updates keyed by uuid.
- hashmap<UUID, StatusUpdate> updates;
-
private:
Framework(const Framework&); // No copying.
Framework& operator = (const Framework&); // No assigning.
Modified: incubator/mesos/trunk/src/slave/state.cpp
URL: http://svn.apache.org/viewvc/incubator/mesos/trunk/src/slave/state.cpp?rev=1459483&r1=1459482&r2=1459483&view=diff
==============================================================================
--- incubator/mesos/trunk/src/slave/state.cpp (original)
+++ incubator/mesos/trunk/src/slave/state.cpp Thu Mar 21 19:25:09 2013
@@ -443,7 +443,7 @@ Try<TaskState> TaskState::recover(
if (record.get().type() == StatusUpdateRecord::UPDATE) {
state.updates.push_back(record.get().update());
} else {
- state.acks.insert(record.get().uuid());
+ state.acks.insert(UUID::fromBytes(record.get().uuid()));
}
}
Modified: incubator/mesos/trunk/src/slave/state.hpp
URL: http://svn.apache.org/viewvc/incubator/mesos/trunk/src/slave/state.hpp?rev=1459483&r1=1459482&r2=1459483&view=diff
==============================================================================
--- incubator/mesos/trunk/src/slave/state.hpp (original)
+++ incubator/mesos/trunk/src/slave/state.hpp Thu Mar 21 19:25:09 2013
@@ -134,7 +134,7 @@ struct TaskState
TaskID id;
Option<Task> info;
std::vector<StatusUpdate> updates;
- hashset<std::string> acks;
+ hashset<UUID> acks;
};
Modified: incubator/mesos/trunk/src/slave/status_update_manager.cpp
URL: http://svn.apache.org/viewvc/incubator/mesos/trunk/src/slave/status_update_manager.cpp?rev=1459483&r1=1459482&r2=1459483&view=diff
==============================================================================
--- incubator/mesos/trunk/src/slave/status_update_manager.cpp (original)
+++ incubator/mesos/trunk/src/slave/status_update_manager.cpp Thu Mar 21 19:25:09 2013
@@ -27,6 +27,7 @@
#include "stout/hashset.hpp"
#include "stout/protobuf.hpp"
#include "stout/utils.hpp"
+#include "stout/uuid.hpp"
#include "slave/constants.hpp"
#include "slave/slave.hpp"
@@ -67,7 +68,7 @@ public:
Try<Nothing> acknowledgement(
const TaskID& taskId,
const FrameworkID& frameworkId,
- const string& uuid);
+ const UUID& uuid);
Try<Nothing> recover(const string& rootDir, const SlaveState& state);
@@ -308,7 +309,7 @@ Timeout StatusUpdateManagerProcess::forw
Try<Nothing> StatusUpdateManagerProcess::acknowledgement(
const TaskID& taskId,
const FrameworkID& frameworkId,
- const string& uuid)
+ const UUID& uuid)
{
LOG(INFO) << "Received status update acknowledgement"
<< " for task " << taskId
@@ -488,7 +489,7 @@ Future<Try<Nothing> > StatusUpdateManage
Future<Try<Nothing> > StatusUpdateManager::acknowledgement(
const TaskID& taskId,
const FrameworkID& frameworkId,
- const std::string& uuid)
+ const UUID& uuid)
{
return dispatch(
process,
Modified: incubator/mesos/trunk/src/slave/status_update_manager.hpp
URL: http://svn.apache.org/viewvc/incubator/mesos/trunk/src/slave/status_update_manager.hpp?rev=1459483&r1=1459482&r2=1459483&view=diff
==============================================================================
--- incubator/mesos/trunk/src/slave/status_update_manager.hpp (original)
+++ incubator/mesos/trunk/src/slave/status_update_manager.hpp Thu Mar 21 19:25:09 2013
@@ -86,7 +86,7 @@ public:
process::Future<Try<Nothing> > acknowledgement(
const TaskID& taskId,
const FrameworkID& frameworkId,
- const std::string& uuid);
+ const UUID& uuid);
// Recover status updates.
process::Future<Try<Nothing> > recover(
@@ -169,7 +169,7 @@ struct StatusUpdateStream
// Check that this status update has not already been acknowledged.
// This could happen in the rare case when the slave received the ACK
// from the framework, died, but slave's ACK to the executor never made it!
- if (acknowledged.contains(update.uuid())) {
+ if (acknowledged.contains(UUID::fromBytes(update.uuid()))) {
LOG(WARNING) << "Ignoring status update " << update
<< " that has already been acknowledged by the framework!";
return Nothing();
@@ -178,7 +178,7 @@ struct StatusUpdateStream
// Check that this update hasn't already been received.
// This could happen if the slave receives a status update from an executor,
// then crashes after it writes it to disk but before it sends an ack.
- if (received.contains(update.uuid())) {
+ if (received.contains(UUID::fromBytes(update.uuid()))) {
LOG(WARNING) << "Ignoring duplicate status update " << update;
return Nothing();
}
@@ -190,16 +190,16 @@ struct StatusUpdateStream
Try<Nothing> acknowledgement(
const TaskID& taskId,
const FrameworkID& frameworkId,
- const std::string& uuid,
+ const UUID& uuid,
const StatusUpdate& update)
{
if (error.isSome()) {
return Error(error.get());
}
- CHECK(uuid.compare(update.uuid()) == 0)
+ CHECK(uuid == UUID::fromBytes(update.uuid()))
<< "Unexpected UUID mismatch! (received " << uuid
- << ", expecting " << update.uuid()
+ << ", expecting " << UUID::fromBytes(update.uuid()).toString()
<< ") for update " << stringify(update);
// Handle the ACK, checkpointing if necessary.
@@ -224,7 +224,7 @@ struct StatusUpdateStream
// corresponding ACK, if present.
Try<Nothing> replay(
const std::vector<StatusUpdate>& updates,
- const hashset<std::string>& acks)
+ const hashset<UUID>& acks)
{
if (error.isSome()) {
return Error(error.get());
@@ -237,7 +237,7 @@ struct StatusUpdateStream
_handle(update, StatusUpdateRecord::UPDATE);
// Check if the update has an ACK too.
- if (acks.contains(update.uuid())) {
+ if (acks.contains(UUID::fromBytes(update.uuid()))) {
_handle(update, StatusUpdateRecord::ACK);
}
}
@@ -315,13 +315,13 @@ private:
if (type == StatusUpdateRecord::UPDATE) {
// Record this update.
- received.insert(update.uuid());
+ received.insert(UUID::fromBytes(update.uuid()));
// Add it to the pending updates queue.
pending.push(update);
} else {
// Record this ACK.
- acknowledged.insert(update.uuid());
+ acknowledged.insert(UUID::fromBytes(update.uuid()));
// Remove the corresponding update from the pending queue.
pending.pop();
@@ -338,8 +338,8 @@ private:
bool checkpoint;
bool terminated_;
- hashset<std::string> received;
- hashset<std::string> acknowledged;
+ hashset<UUID> received;
+ hashset<UUID> acknowledged;
const Option<std::string> path; // File path of the update stream.
Option<int> fd; // File descriptor to the update stream.
Modified: incubator/mesos/trunk/src/tests/gc_tests.cpp
URL: http://svn.apache.org/viewvc/incubator/mesos/trunk/src/tests/gc_tests.cpp?rev=1459483&r1=1459482&r2=1459483&view=diff
==============================================================================
--- incubator/mesos/trunk/src/tests/gc_tests.cpp (original)
+++ incubator/mesos/trunk/src/tests/gc_tests.cpp Thu Mar 21 19:25:09 2013
@@ -398,12 +398,12 @@ TEST_F(GarbageCollectorTest, DiskUsage)
// message back to itself.
WAIT_UNTIL(exitedExecutorMsg);
+ sleep(1); // Wait for slave to schedule for gc.
+
// Simulate a disk full message to the slave.
process::dispatch(slave, &Slave::_checkDiskUsage, Try<double>::some(1));
- // TODO(vinod): As above, we need to wait until GarbageCollectorProcess has
- // dispatched remove message back to itself.
- sleep(1);
+ sleep(1); // Wait for gc to act.
Clock::settle();
Modified: incubator/mesos/trunk/src/tests/slave_recovery_tests.cpp
URL: http://svn.apache.org/viewvc/incubator/mesos/trunk/src/tests/slave_recovery_tests.cpp?rev=1459483&r1=1459482&r2=1459483&view=diff
==============================================================================
--- incubator/mesos/trunk/src/tests/slave_recovery_tests.cpp (original)
+++ incubator/mesos/trunk/src/tests/slave_recovery_tests.cpp Thu Mar 21 19:25:09 2013
@@ -31,6 +31,7 @@
#include <stout/numify.hpp>
#include <stout/option.hpp>
#include <stout/path.hpp>
+#include <stout/uuid.hpp>
#include "common/process_utils.hpp"
#include "common/protobuf_utils.hpp"
@@ -402,7 +403,7 @@ TYPED_TEST(SlaveRecoveryTest, RecoverSla
.executors[executorId]
.runs[uuid.get()]
.tasks[task.task_id()]
- .acks.contains(ack.uuid()));
+ .acks.contains(UUID::fromBytes(ack.uuid())));
// Shut down the executor.
process::post(libprocessPid, ShutdownExecutorMessage());