You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@mesos.apache.org by id...@apache.org on 2014/06/16 19:19:19 UTC
[1/3] git commit: Wait on Containerizer::update to complete before
sending status update.
Repository: mesos
Updated Branches:
refs/heads/master 6215725d2 -> 65bc1f88d
Wait on Containerizer::update to complete before sending status update.
Review: https://reviews.apache.org/r/22253
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/d8f57cef
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/d8f57cef
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/d8f57cef
Branch: refs/heads/master
Commit: d8f57cef9be8c6c199d537c2139bde9f19e7c77d
Parents: 75798d4
Author: Ian Downes <id...@twitter.com>
Authored: Wed Jun 4 16:23:02 2014 -0700
Committer: Ian Downes <id...@twitter.com>
Committed: Mon Jun 16 10:01:44 2014 -0700
----------------------------------------------------------------------
src/slave/slave.cpp | 71 ++++++++++++++-----
src/slave/slave.hpp | 12 +++-
src/tests/slave_tests.cpp | 95 ++++++++++++++++++++++++++
src/tests/status_update_manager_tests.cpp | 18 ++---
4 files changed, 169 insertions(+), 27 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/d8f57cef/src/slave/slave.cpp
----------------------------------------------------------------------
diff --git a/src/slave/slave.cpp b/src/slave/slave.cpp
index 423974c..bc976b7 100644
--- a/src/slave/slave.cpp
+++ b/src/slave/slave.cpp
@@ -2012,7 +2012,7 @@ void Slave::reregisterExecutorTimeout()
// This can be called in two ways:
// 1) When a status update from the executor is received.
// 2) When slave generates task updates (e.g LOST/KILLED/FAILED).
-// NOTE: We set the pid in 'Slave::_statusUpdate()' to 'pid' so that
+// NOTE: We set the pid in 'Slave::__statusUpdate()' to 'pid' so that
// whoever sent this update will get an ACK. This is important because
// we allow executors to send updates for tasks that belong to other
// executors. Currently we allow this because we cannot guarantee
@@ -2072,7 +2072,7 @@ void Slave::statusUpdate(const StatusUpdate& update, const UPID& pid)
// corresponding to this task because the task has been moved to
// 'Executor::completedTasks'.
statusUpdateManager->update(update, info.id())
- .onAny(defer(self(), &Slave::_statusUpdate, lambda::_1, update, pid));
+ .onAny(defer(self(), &Slave::__statusUpdate, lambda::_1, update, pid));
return;
}
@@ -2105,37 +2105,74 @@ void Slave::statusUpdate(const StatusUpdate& update, const UPID& pid)
executor->launchedTasks.contains(status.task_id()))) {
executor->terminateTask(status.task_id(), status.state());
- // Tell the isolator to update the resources.
- // TODO(idownes): Wait until this completes.
+ // Wait until the container's resources have been updated before
+ // sending the status update.
CHECK_SOME(executor->resources);
- containerizer->update(executor->containerId, executor->resources.get());
+ containerizer->update(executor->containerId, executor->resources.get())
+ .onAny(defer(self(),
+ &Slave::_statusUpdate,
+ lambda::_1,
+ update,
+ pid,
+ executor->id,
+ executor->containerId,
+ executor->checkpoint));
+ } else {
+ // Immediately send the status update.
+ _statusUpdate(None(),
+ update,
+ pid,
+ executor->id,
+ executor->containerId,
+ executor->checkpoint);
}
+}
- if (executor->checkpoint) {
+
+void Slave::_statusUpdate(
+ const Option<Future<Nothing> >& future,
+ const StatusUpdate& update,
+ const UPID& pid,
+ const ExecutorID& executorId,
+ const ContainerID& containerId,
+ bool checkpoint)
+{
+ if (future.isSome() && !future.get().isReady()) {
+ LOG(ERROR) << "Failed to update resources for container " << containerId
+ << " of executor " << executorId
+ << " running task " << update.status().task_id()
+ << " on status update for terminal task, destroying container:"
+ << (future.get().isFailed() ? future.get().failure()
+ : "discarded");
+
+ containerizer->destroy(containerId);
+ }
+
+ if (checkpoint) {
// Ask the status update manager to checkpoint and reliably send the update.
statusUpdateManager->update(
update,
info.id(),
- executor->id,
- executor->containerId)
+ executorId,
+ containerId)
.onAny(defer(self(),
- &Slave::_statusUpdate,
- lambda::_1,
- update,
- pid));
+ &Slave::__statusUpdate,
+ lambda::_1,
+ update,
+ pid));
} else {
// Ask the status update manager to just retry the update.
statusUpdateManager->update(update, info.id())
.onAny(defer(self(),
- &Slave::_statusUpdate,
- lambda::_1,
- update,
- pid));
+ &Slave::__statusUpdate,
+ lambda::_1,
+ update,
+ pid));
}
}
-void Slave::_statusUpdate(
+void Slave::__statusUpdate(
const Future<Nothing>& future,
const StatusUpdate& update,
const UPID& pid)
http://git-wip-us.apache.org/repos/asf/mesos/blob/d8f57cef/src/slave/slave.hpp
----------------------------------------------------------------------
diff --git a/src/slave/slave.hpp b/src/slave/slave.hpp
index 34687e5..3227e73 100644
--- a/src/slave/slave.hpp
+++ b/src/slave/slave.hpp
@@ -164,10 +164,20 @@ public:
// status updates it generated (e.g., TASK_LOST).
void statusUpdate(const StatusUpdate& update, const process::UPID& pid);
+ // Continue handling the status update after optionally updating the
+ // container's resources.
+ void _statusUpdate(
+ const Option<Future<Nothing> >& future,
+ const StatusUpdate& update,
+ const UPID& pid,
+ const ExecutorID& executorId,
+ const ContainerID& containerId,
+ bool checkpoint);
+
// This is called when the status update manager finishes
// handling the update. If the handling is successful, an
// acknowledgment is sent to the executor.
- void _statusUpdate(
+ void __statusUpdate(
const process::Future<Nothing>& future,
const StatusUpdate& update,
const process::UPID& pid);
http://git-wip-us.apache.org/repos/asf/mesos/blob/d8f57cef/src/tests/slave_tests.cpp
----------------------------------------------------------------------
diff --git a/src/tests/slave_tests.cpp b/src/tests/slave_tests.cpp
index 974293b..aaf509d 100644
--- a/src/tests/slave_tests.cpp
+++ b/src/tests/slave_tests.cpp
@@ -805,3 +805,98 @@ TEST_F(SlaveTest, TerminatingSlaveDoesNotReregister)
Shutdown();
}
+
+
+// This test verifies the slave will destroy a container if, when
+// receiving a terminal status task update, updating the container's
+// resources fails.
+TEST_F(SlaveTest, TerminalTaskContainerizerUpdateFails)
+{
+ // Start a master.
+ Try<PID<Master> > master = StartMaster();
+ ASSERT_SOME(master);
+
+ MockExecutor exec(DEFAULT_EXECUTOR_ID);
+ EXPECT_CALL(exec, registered(_, _, _, _));
+
+ TestContainerizer containerizer(&exec);
+
+ // Start a slave.
+ Try<PID<Slave> > slave = StartSlave(&containerizer);
+ ASSERT_SOME(slave);
+
+ MockScheduler sched;
+ MesosSchedulerDriver driver(
+ &sched, DEFAULT_FRAMEWORK_INFO, master.get(), DEFAULT_CREDENTIAL);
+
+ EXPECT_CALL(sched, registered(_, _, _));
+
+ Future<vector<Offer> > offers;
+
+ EXPECT_CALL(sched, resourceOffers(&driver, _))
+ .WillOnce(FutureArg<1>(&offers))
+ .WillRepeatedly(Return()); // Ignore subsequent offers.
+
+ driver.start();
+
+ AWAIT_READY(offers);
+ EXPECT_NE(0u, offers.get().size());
+ Offer offer = offers.get()[0];
+
+ // Start two tasks.
+ vector<TaskInfo> tasks;
+
+ tasks.push_back(createTask(
+ offer.slave_id(),
+ Resources::parse("cpus:0.1;mem:32").get(),
+ "sleep 1000",
+ exec.id));
+
+ tasks.push_back(createTask(
+ offer.slave_id(),
+ Resources::parse("cpus:0.1;mem:32").get(),
+ "sleep 1000",
+ exec.id));
+
+ EXPECT_CALL(exec, launchTask(_, _))
+ .WillOnce(SendStatusUpdateFromTask(TASK_RUNNING))
+ .WillOnce(SendStatusUpdateFromTask(TASK_RUNNING));
+
+ Future<TaskStatus> status1, status2, status3, status4;
+ EXPECT_CALL(sched, statusUpdate(&driver, _))
+ .WillOnce(FutureArg<1>(&status1))
+ .WillOnce(FutureArg<1>(&status2))
+ .WillOnce(FutureArg<1>(&status3))
+ .WillOnce(FutureArg<1>(&status4));
+
+ driver.launchTasks(offer.id(), tasks);
+
+ AWAIT_READY(status1);
+ EXPECT_EQ(TASK_RUNNING, status1.get().state());
+
+ AWAIT_READY(status2);
+ EXPECT_EQ(TASK_RUNNING, status2.get().state());
+
+ // Set up the containerizer so the next update() will fail.
+ EXPECT_CALL(containerizer, update(_, _))
+ .WillOnce(Return(process::Failure("update() failed")))
+ .WillRepeatedly(Return(Nothing()));
+
+ EXPECT_CALL(exec, killTask(_, _))
+ .WillOnce(SendStatusUpdateFromTaskID(TASK_KILLED));
+
+ // Kill one of the tasks. The failed update should result in the
+ // second task going lost when the container is destroyed.
+ driver.killTask(tasks[0].task_id());
+
+ AWAIT_READY(status3);
+ EXPECT_EQ(TASK_KILLED, status3.get().state());
+
+ AWAIT_READY(status4);
+ EXPECT_EQ(TASK_LOST, status4.get().state());
+
+ driver.stop();
+ driver.join();
+
+ Shutdown();
+}
http://git-wip-us.apache.org/repos/asf/mesos/blob/d8f57cef/src/tests/status_update_manager_tests.cpp
----------------------------------------------------------------------
diff --git a/src/tests/status_update_manager_tests.cpp b/src/tests/status_update_manager_tests.cpp
index c753921..e9ef1e2 100644
--- a/src/tests/status_update_manager_tests.cpp
+++ b/src/tests/status_update_manager_tests.cpp
@@ -522,8 +522,8 @@ TEST_F(StatusUpdateManagerTest, DuplicateTerminalUpdateBeforeAck)
Future<StatusUpdateAcknowledgementMessage> statusUpdateAckMessage =
DROP_PROTOBUF(StatusUpdateAcknowledgementMessage(), _, slave.get());
- Future<Nothing> _statusUpdate =
- FUTURE_DISPATCH(slave.get(), &Slave::_statusUpdate);
+ Future<Nothing> __statusUpdate =
+ FUTURE_DISPATCH(slave.get(), &Slave::__statusUpdate);
Clock::pause();
@@ -537,10 +537,10 @@ TEST_F(StatusUpdateManagerTest, DuplicateTerminalUpdateBeforeAck)
// At this point the status update manager has enqueued
// TASK_FINISHED update.
- AWAIT_READY(_statusUpdate);
+ AWAIT_READY(__statusUpdate);
- Future<Nothing> _statusUpdate2 =
- FUTURE_DISPATCH(slave.get(), &Slave::_statusUpdate);
+ Future<Nothing> __statusUpdate2 =
+ FUTURE_DISPATCH(slave.get(), &Slave::__statusUpdate);
// Now send a TASK_KILLED update for the same task.
TaskStatus status2 = status.get();
@@ -549,7 +549,7 @@ TEST_F(StatusUpdateManagerTest, DuplicateTerminalUpdateBeforeAck)
// At this point the status update manager has enqueued
// TASK_FINISHED and TASK_KILLED updates.
- AWAIT_READY(_statusUpdate2);
+ AWAIT_READY(__statusUpdate2);
// After we advance the clock, the scheduler should receive
// the retried TASK_FINISHED update and acknowledge it. The
@@ -748,15 +748,15 @@ TEST_F(StatusUpdateManagerTest, DuplicateUpdateBeforeAck)
AWAIT_READY(statusUpdateAckMessage);
- Future<Nothing> _statusUpdate =
- FUTURE_DISPATCH(slave.get(), &Slave::_statusUpdate);
+ Future<Nothing> __statusUpdate =
+ FUTURE_DISPATCH(slave.get(), &Slave::__statusUpdate);
// Now resend the TASK_RUNNING update.
process::post(slave.get(), statusUpdateMessage.get());
// At this point the status update manager has handled
// the duplicate status update.
- AWAIT_READY(_statusUpdate);
+ AWAIT_READY(__statusUpdate);
// After we advance the clock, the status update manager should
// retry the TASK_RUNING update and the scheduler should receive
[2/3] git commit: Give the command executor some cpu and memory.
Posted by id...@apache.org.
Give the command executor some cpu and memory.
Review: https://reviews.apache.org/r/22251
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/75798d43
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/75798d43
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/75798d43
Branch: refs/heads/master
Commit: 75798d437f5f1f7d08b61388c9f1018a54553908
Parents: 6215725
Author: Ian Downes <id...@twitter.com>
Authored: Wed Jun 4 16:19:54 2014 -0700
Committer: Ian Downes <id...@twitter.com>
Committed: Mon Jun 16 10:01:44 2014 -0700
----------------------------------------------------------------------
src/slave/constants.hpp | 6 ++++++
src/slave/slave.cpp | 7 +++++++
2 files changed, 13 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/75798d43/src/slave/constants.hpp
----------------------------------------------------------------------
diff --git a/src/slave/constants.hpp b/src/slave/constants.hpp
index ace4590..c65a62d 100644
--- a/src/slave/constants.hpp
+++ b/src/slave/constants.hpp
@@ -80,6 +80,12 @@ extern const Bytes DEFAULT_DISK;
// Default ports range offered by the slave.
extern const std::string DEFAULT_PORTS;
+// Default cpu resource given to a command executor.
+const double DEFAULT_EXECUTOR_CPUS = 0.1;
+
+// Default memory resource given to a command executor.
+const Bytes DEFAULT_EXECUTOR_MEM = Megabytes(32);
+
} // namespace slave {
} // namespace internal {
} // namespace mesos {
http://git-wip-us.apache.org/repos/asf/mesos/blob/75798d43/src/slave/slave.cpp
----------------------------------------------------------------------
diff --git a/src/slave/slave.cpp b/src/slave/slave.cpp
index 643c088..423974c 100644
--- a/src/slave/slave.cpp
+++ b/src/slave/slave.cpp
@@ -2297,6 +2297,13 @@ ExecutorInfo Slave::getExecutorInfo(
"'; exit 1");
}
+ // Add an allowance for the command executor. This does lead to a
+ // small overcommit of resources.
+ executor.mutable_resources()->MergeFrom(
+ Resources::parse(
+ "cpus:" + stringify(DEFAULT_EXECUTOR_CPUS) + ";" +
+ "mem:" + stringify(DEFAULT_EXECUTOR_MEM.megabytes())).get());
+
return executor;
}
[3/3] git commit: Fix some clang compiler issues in perf code.
Posted by id...@apache.org.
Fix some clang compiler issues in perf code.
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/65bc1f88
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/65bc1f88
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/65bc1f88
Branch: refs/heads/master
Commit: 65bc1f88d7d1f4e6c92449e2522810e3f7257b0b
Parents: d8f57ce
Author: Ian Downes <id...@twitter.com>
Authored: Mon Jun 16 10:17:28 2014 -0700
Committer: Ian Downes <id...@twitter.com>
Committed: Mon Jun 16 10:17:28 2014 -0700
----------------------------------------------------------------------
src/linux/perf.cpp | 4 ++--
src/tests/cgroups_tests.cpp | 4 ++--
2 files changed, 4 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/65bc1f88/src/linux/perf.cpp
----------------------------------------------------------------------
diff --git a/src/linux/perf.cpp b/src/linux/perf.cpp
index c0f2e92..bc31db5 100644
--- a/src/linux/perf.cpp
+++ b/src/linux/perf.cpp
@@ -196,7 +196,7 @@ private:
{
if (!status.isReady()) {
promise.fail("Failed to get exit status of perf process: " +
- status.isFailed() ? status.failure() : "discarded");
+ (status.isFailed() ? status.failure() : "discarded"));
terminate(self());
return;
}
@@ -217,7 +217,7 @@ private:
{
if (!future.isReady()) {
promise.fail("Failed to collect output of perf process: " +
- future.isFailed() ? future.failure() : "discarded");
+ (future.isFailed() ? future.failure() : "discarded"));
terminate(self());
return;
}
http://git-wip-us.apache.org/repos/asf/mesos/blob/65bc1f88/src/tests/cgroups_tests.cpp
----------------------------------------------------------------------
diff --git a/src/tests/cgroups_tests.cpp b/src/tests/cgroups_tests.cpp
index 73510fb..adcf46b 100644
--- a/src/tests/cgroups_tests.cpp
+++ b/src/tests/cgroups_tests.cpp
@@ -884,7 +884,7 @@ TEST_F(CgroupsAnyHierarchyWithPerfEventTest, ROOT_CGROUPS_Perf)
ssize_t len;
while ((len = ::read(pipes[0], &dummy, sizeof(dummy))) == -1 &&
errno == EINTR);
- ASSERT_EQ(sizeof(dummy), len);
+ ASSERT_EQ((ssize_t) sizeof(dummy), len);
::close(pipes[0]);
while (true) { sleep(1); }
@@ -901,7 +901,7 @@ TEST_F(CgroupsAnyHierarchyWithPerfEventTest, ROOT_CGROUPS_Perf)
ssize_t len;
while ((len = ::write(pipes[1], &dummy, sizeof(dummy))) == -1 &&
errno == EINTR);
- ASSERT_EQ(sizeof(dummy), len);
+ ASSERT_EQ((ssize_t) sizeof(dummy), len);
::close(pipes[1]);
std::set<std::string> events;