You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@mesos.apache.org by an...@apache.org on 2016/09/12 22:18:45 UTC
[1/8] mesos git commit: Added tests for launching task groups on the
agent.
Repository: mesos
Updated Branches:
refs/heads/master 00815ee93 -> 170c01f1a
Added tests for launching task groups on the agent.
This change adds tests for testing the `runTaskGroup()` handler
on the agent.
Review: https://reviews.apache.org/r/51480/
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/170c01f1
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/170c01f1
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/170c01f1
Branch: refs/heads/master
Commit: 170c01f1a184e7fdc6cd057e5ef7dbf2fbc658af
Parents: f8734b7
Author: Anand Mazumdar <an...@apache.org>
Authored: Mon Sep 12 14:23:00 2016 -0700
Committer: Anand Mazumdar <an...@apache.org>
Committed: Mon Sep 12 15:17:02 2016 -0700
----------------------------------------------------------------------
src/tests/slave_tests.cpp | 318 +++++++++++++++++++++++++++++++++++++++++
1 file changed, 318 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/170c01f1/src/tests/slave_tests.cpp
----------------------------------------------------------------------
diff --git a/src/tests/slave_tests.cpp b/src/tests/slave_tests.cpp
index 55dba3f..2f3fa5f 100644
--- a/src/tests/slave_tests.cpp
+++ b/src/tests/slave_tests.cpp
@@ -77,6 +77,9 @@ using mesos::internal::protobuf::createLabel;
using mesos::master::detector::MasterDetector;
using mesos::master::detector::StandaloneMasterDetector;
+using mesos::v1::scheduler::Call;
+using mesos::v1::scheduler::Mesos;
+
using process::Clock;
using process::Future;
using process::Owned;
@@ -4304,6 +4307,321 @@ TEST_F(SlaveTest, ExecutorShutdownGracePeriod)
driver.join();
}
+
+// This test verifies that the agent can forward a task group to an
+// executor atomically via the `LAUNCH_GROUP` event.
+TEST_F(SlaveTest, RunTaskGroup)
+{
+ Try<Owned<cluster::Master>> master = StartMaster();
+ ASSERT_SOME(master);
+
+ auto scheduler = std::make_shared<MockV1HTTPScheduler>();
+ auto executor = std::make_shared<MockV1HTTPExecutor>();
+
+ Resources resources =
+ Resources::parse("cpus:0.1;mem:32;disk:32").get();
+
+ ExecutorInfo executorInfo = DEFAULT_EXECUTOR_INFO;
+ executorInfo.set_type(ExecutorInfo::CUSTOM);
+
+ executorInfo.mutable_resources()->CopyFrom(resources);
+
+ const ExecutorID& executorId = executorInfo.executor_id();
+ TestContainerizer containerizer(executorId, executor);
+
+ Owned<MasterDetector> detector = master.get()->createDetector();
+ Try<Owned<cluster::Slave>> slave = StartSlave(detector.get(), &containerizer);
+ ASSERT_SOME(slave);
+
+ Future<Nothing> connected;
+ EXPECT_CALL(*scheduler, connected(_))
+ .WillOnce(FutureSatisfy(&connected));
+
+ scheduler::TestV1Mesos mesos(
+ master.get()->pid, ContentType::PROTOBUF, scheduler);
+
+ AWAIT_READY(connected);
+
+ Future<v1::scheduler::Event::Subscribed> subscribed;
+ EXPECT_CALL(*scheduler, subscribed(_, _))
+ .WillOnce(FutureArg<1>(&subscribed));
+
+ Future<v1::scheduler::Event::Offers> offers;
+ EXPECT_CALL(*scheduler, offers(_, _))
+ .WillOnce(FutureArg<1>(&offers));
+
+ EXPECT_CALL(*scheduler, heartbeat(_))
+ .WillRepeatedly(Return()); // Ignore heartbeats.
+
+ {
+ Call call;
+ call.set_type(Call::SUBSCRIBE);
+ Call::Subscribe* subscribe = call.mutable_subscribe();
+ subscribe->mutable_framework_info()->CopyFrom(DEFAULT_V1_FRAMEWORK_INFO);
+
+ mesos.send(call);
+ }
+
+ AWAIT_READY(subscribed);
+
+ v1::FrameworkID frameworkId(subscribed->framework_id());
+
+ // Update `executorInfo` with the subscribed `frameworkId`.
+ executorInfo.mutable_framework_id()->CopyFrom(devolve(frameworkId));
+
+ AWAIT_READY(offers);
+ EXPECT_NE(0, offers->offers().size());
+
+ EXPECT_CALL(*executor, connected(_))
+ .WillOnce(executor::SendSubscribe(frameworkId, evolve(executorId)));
+
+ EXPECT_CALL(*executor, subscribed(_, _));
+
+ EXPECT_CALL(*executor, launch(_, _))
+ .Times(0);
+
+ Future<v1::executor::Event::LaunchGroup> launchGroupEvent;
+ EXPECT_CALL(*executor, launchGroup(_, _))
+ .WillOnce(FutureArg<1>(&launchGroupEvent));
+
+ const v1::Offer& offer = offers->offers(0);
+ const SlaveID slaveId = devolve(offer.agent_id());
+
+ v1::TaskInfo taskInfo1 =
+ evolve(createTask(slaveId, resources, ""));
+
+ v1::TaskInfo taskInfo2 =
+ evolve(createTask(slaveId, resources, ""));
+
+ v1::TaskGroupInfo taskGroup;
+ taskGroup.add_tasks()->CopyFrom(taskInfo1);
+ taskGroup.add_tasks()->CopyFrom(taskInfo2);
+
+ {
+ Call call;
+ call.mutable_framework_id()->CopyFrom(frameworkId);
+ call.set_type(Call::ACCEPT);
+
+ Call::Accept* accept = call.mutable_accept();
+ accept->add_offer_ids()->CopyFrom(offer.id());
+
+ v1::Offer::Operation* operation = accept->add_operations();
+ operation->set_type(v1::Offer::Operation::LAUNCH_GROUP);
+
+ v1::Offer::Operation::LaunchGroup* launchGroup =
+ operation->mutable_launch_group();
+
+ launchGroup->mutable_executor()->CopyFrom(evolve(executorInfo));
+ launchGroup->mutable_task_group()->CopyFrom(taskGroup);
+
+ mesos.send(call);
+ }
+
+ AWAIT_READY(launchGroupEvent);
+
+ ASSERT_EQ(2, launchGroupEvent->task_group().tasks().size());
+ EXPECT_EQ(taskInfo1.task_id(),
+ launchGroupEvent->task_group().tasks(0).task_id());
+ EXPECT_EQ(taskInfo2.task_id(),
+ launchGroupEvent->task_group().tasks(1).task_id());
+
+ EXPECT_CALL(*executor, shutdown(_))
+ .Times(AtMost(1));
+}
+
+
+// This test ensures that a `killTask()` can happen between `runTask()`
+// and `_run()` and then gets "handled properly" for a task group.
+// This should result in TASK_KILLED updates for all the tasks in the
+// task group.
+TEST_F(SlaveTest, KillTaskGroupBetweenRunTaskParts)
+{
+ Try<Owned<cluster::Master>> master = StartMaster();
+ ASSERT_SOME(master);
+
+ auto scheduler = std::make_shared<MockV1HTTPScheduler>();
+ auto executor = std::make_shared<MockV1HTTPExecutor>();
+
+ Resources resources =
+ Resources::parse("cpus:0.1;mem:32;disk:32").get();
+
+ ExecutorInfo executorInfo = DEFAULT_EXECUTOR_INFO;
+ executorInfo.set_type(ExecutorInfo::CUSTOM);
+
+ executorInfo.mutable_resources()->CopyFrom(resources);
+
+ const ExecutorID& executorId = executorInfo.executor_id();
+ TestContainerizer containerizer(executorId, executor);
+
+ StandaloneMasterDetector detector(master.get()->pid);
+
+ MockSlave slave(CreateSlaveFlags(), &detector, &containerizer);
+ spawn(slave);
+
+ Future<Nothing> connected;
+ EXPECT_CALL(*scheduler, connected(_))
+ .WillOnce(FutureSatisfy(&connected));
+
+ scheduler::TestV1Mesos mesos(
+ master.get()->pid, ContentType::PROTOBUF, scheduler);
+
+ AWAIT_READY(connected);
+
+ Future<v1::scheduler::Event::Subscribed> subscribed;
+ EXPECT_CALL(*scheduler, subscribed(_, _))
+ .WillOnce(FutureArg<1>(&subscribed));
+
+ Future<v1::scheduler::Event::Offers> offers;
+ EXPECT_CALL(*scheduler, offers(_, _))
+ .WillOnce(FutureArg<1>(&offers))
+ .WillRepeatedly(Return());
+
+ EXPECT_CALL(*scheduler, heartbeat(_))
+ .WillRepeatedly(Return()); // Ignore heartbeats.
+
+ {
+ Call call;
+ call.set_type(Call::SUBSCRIBE);
+ Call::Subscribe* subscribe = call.mutable_subscribe();
+ subscribe->mutable_framework_info()->CopyFrom(DEFAULT_V1_FRAMEWORK_INFO);
+
+ mesos.send(call);
+ }
+
+ AWAIT_READY(subscribed);
+
+ v1::FrameworkID frameworkId(subscribed->framework_id());
+
+ // Update `executorInfo` with the subscribed `frameworkId`.
+ executorInfo.mutable_framework_id()->CopyFrom(devolve(frameworkId));
+
+ AWAIT_READY(offers);
+ EXPECT_NE(0, offers->offers().size());
+
+ EXPECT_CALL(*executor, connected(_))
+ .Times(0);
+
+ EXPECT_CALL(*executor, subscribed(_, _))
+ .Times(0);
+
+ EXPECT_CALL(*executor, shutdown(_))
+ .Times(0);
+
+ EXPECT_CALL(*executor, launchGroup(_, _))
+ .Times(0);
+
+ EXPECT_CALL(*executor, launch(_, _))
+ .Times(0);
+
+ Future<v1::scheduler::Event::Update> update1;
+ Future<v1::scheduler::Event::Update> update2;
+
+ EXPECT_CALL(*scheduler, update(_, _))
+ .WillOnce(FutureArg<1>(&update1))
+ .WillOnce(FutureArg<1>(&update2))
+ .WillRepeatedly(Return());
+
+ EXPECT_CALL(slave, runTaskGroup(_, _, _, _))
+ .WillOnce(Invoke(&slave, &MockSlave::unmocked_runTaskGroup));
+
+ // Saved arguments from `Slave::_run()`.
+ Future<bool> future;
+ FrameworkInfo frameworkInfo;
+ ExecutorInfo executorInfo_;
+ Option<TaskGroupInfo> taskGroup_;
+ Option<TaskInfo> task_;
+
+ // Skip what `Slave::_run()` normally does, save its arguments for
+ // later, till reaching the critical moment when to kill the task
+ // in the future.
+ Future<Nothing> _run;
+ EXPECT_CALL(slave, _run(_, _, _, _, _))
+ .WillOnce(DoAll(FutureSatisfy(&_run),
+ SaveArg<0>(&future),
+ SaveArg<1>(&frameworkInfo),
+ SaveArg<2>(&executorInfo_),
+ SaveArg<3>(&task_),
+ SaveArg<4>(&taskGroup_)));
+
+ const v1::Offer& offer = offers->offers(0);
+ const SlaveID slaveId = devolve(offer.agent_id());
+
+ v1::TaskInfo taskInfo1 =
+ evolve(createTask(slaveId, resources, ""));
+
+ v1::TaskInfo taskInfo2 =
+ evolve(createTask(slaveId, resources, ""));
+
+ v1::TaskGroupInfo taskGroup;
+ taskGroup.add_tasks()->CopyFrom(taskInfo1);
+ taskGroup.add_tasks()->CopyFrom(taskInfo2);
+
+ {
+ Call call;
+ call.mutable_framework_id()->CopyFrom(frameworkId);
+ call.set_type(Call::ACCEPT);
+
+ Call::Accept* accept = call.mutable_accept();
+ accept->add_offer_ids()->CopyFrom(offer.id());
+
+ v1::Offer::Operation* operation = accept->add_operations();
+ operation->set_type(v1::Offer::Operation::LAUNCH_GROUP);
+
+ v1::Offer::Operation::LaunchGroup* launchGroup =
+ operation->mutable_launch_group();
+
+ launchGroup->mutable_executor()->CopyFrom(evolve(executorInfo));
+ launchGroup->mutable_task_group()->CopyFrom(taskGroup);
+
+ mesos.send(call);
+ }
+
+ AWAIT_READY(_run);
+
+ Future<Nothing> killTask;
+ EXPECT_CALL(slave, killTask(_, _))
+ .WillOnce(DoAll(Invoke(&slave, &MockSlave::unmocked_killTask),
+ FutureSatisfy(&killTask)));
+
+ {
+ Call call;
+ call.mutable_framework_id()->CopyFrom(frameworkId);
+ call.set_type(Call::KILL);
+
+ Call::Kill* kill = call.mutable_kill();
+ kill->mutable_task_id()->CopyFrom(taskInfo1.task_id());
+ kill->mutable_agent_id()->CopyFrom(offer.agent_id());
+
+ mesos.send(call);
+ }
+
+ AWAIT_READY(killTask);
+
+ // Since this is the only task group for this framework, the
+ // framework should get removed in `Slave::_run()`.
+ Future<Nothing> removeFramework;
+ EXPECT_CALL(slave, removeFramework(_))
+ .WillOnce(DoAll(Invoke(&slave, &MockSlave::unmocked_removeFramework),
+ FutureSatisfy(&removeFramework)));
+
+ slave.unmocked__run(
+ future, frameworkInfo, executorInfo_, task_, taskGroup_);
+
+ AWAIT_READY(removeFramework);
+
+ AWAIT_READY(update1);
+ AWAIT_READY(update2);
+
+ EXPECT_EQ(TASK_KILLED, update1->status().state());
+ EXPECT_EQ(taskInfo1.task_id(), update1->status().task_id());
+
+ EXPECT_EQ(TASK_KILLED, update2->status().state());
+ EXPECT_EQ(taskInfo2.task_id(), update2->status().task_id());
+
+ terminate(slave);
+ wait(slave);
+}
+
} // namespace tests {
} // namespace internal {
} // namespace mesos {
[4/8] mesos git commit: Implemented `RunTaskGroupMessage` handler on
the agent.
Posted by an...@apache.org.
Implemented `RunTaskGroupMessage` handler on the agent.
This changes implements the `runTaskGroup()` handler on the
agent ensuring that task group is sent atomically to the executor
via the `LAUNCH_GROUP` event. It also refactors `runTask()`/`_runTask()`
to go through a common handler function. Also, it ensures that all
tasks in `framework->pending`/`queuedTasks` that are killed before
running the task group result in all the tasks being killed.
Review: https://reviews.apache.org/r/51477/
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/3a4c3f14
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/3a4c3f14
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/3a4c3f14
Branch: refs/heads/master
Commit: 3a4c3f14c1b88650b6c1ed8329ff44869de53024
Parents: 3e9ea3b
Author: Anand Mazumdar <an...@apache.org>
Authored: Mon Sep 12 14:21:28 2016 -0700
Committer: Anand Mazumdar <an...@apache.org>
Committed: Mon Sep 12 15:17:02 2016 -0700
----------------------------------------------------------------------
src/slave/slave.cpp | 819 ++++++++++++++++++++++++++++++++++-------------
src/slave/slave.hpp | 52 ++-
2 files changed, 636 insertions(+), 235 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/3a4c3f14/src/slave/slave.cpp
----------------------------------------------------------------------
diff --git a/src/slave/slave.cpp b/src/slave/slave.cpp
index fa3610d..c7ab7f2 100644
--- a/src/slave/slave.cpp
+++ b/src/slave/slave.cpp
@@ -25,6 +25,7 @@
#include <map>
#include <memory>
#include <set>
+#include <sstream>
#include <string>
#include <vector>
@@ -109,8 +110,10 @@ using mesos::slave::QoSController;
using mesos::slave::QoSCorrection;
using mesos::slave::ResourceEstimator;
+using std::find;
using std::list;
using std::map;
+using std::ostringstream;
using std::set;
using std::string;
using std::tuple;
@@ -568,6 +571,12 @@ void Slave::initialize()
&RunTaskMessage::pid,
&RunTaskMessage::task);
+ install<RunTaskGroupMessage>(
+ &Slave::runTaskGroup,
+ &RunTaskGroupMessage::framework,
+ &RunTaskGroupMessage::executor,
+ &RunTaskGroupMessage::task_group);
+
install<KillTaskMessage>(
&Slave::killTask);
@@ -1481,7 +1490,7 @@ void Slave::runTask(
const FrameworkInfo& frameworkInfo,
const FrameworkID& frameworkId_,
const UPID& pid,
- TaskInfo task)
+ const TaskInfo& task)
{
if (master != from) {
LOG(WARNING) << "Ignoring run task message from " << from
@@ -1496,17 +1505,44 @@ void Slave::runTask(
return;
}
- // Create frameworkId alias to use in the rest of the function.
- const FrameworkID frameworkId = frameworkInfo.id();
+ const ExecutorInfo executorInfo = getExecutorInfo(frameworkInfo, task);
+
+ run(frameworkInfo, executorInfo, task, None(), pid);
+}
+
+
+void Slave::run(
+ const FrameworkInfo& frameworkInfo,
+ const ExecutorInfo& executorInfo,
+ Option<TaskInfo> task,
+ Option<TaskGroupInfo> taskGroup,
+ const UPID& pid)
+{
+ CHECK_NE(task.isSome(), taskGroup.isSome())
+ << "Either task or task group should be set but not both";
+
+ vector<TaskInfo> tasks;
+ if (task.isSome()) {
+ tasks.push_back(task.get());
+ } else {
+ foreach (const TaskInfo& task, taskGroup->tasks()) {
+ tasks.push_back(task);
+ }
+ }
+
+ const FrameworkID& frameworkId = frameworkInfo.id();
- LOG(INFO) << "Got assigned task " << task.task_id()
+ LOG(INFO) << "Got assigned " << taskOrTaskGroup(task, taskGroup)
<< " for framework " << frameworkId;
- if (!(task.slave_id() == info.id())) {
- LOG(WARNING)
- << "Agent " << info.id() << " ignoring task " << task.task_id()
- << " because it was intended for old agent " << task.slave_id();
- return;
+ foreach (const TaskInfo& task, tasks) {
+ if (!(task.slave_id() == info.id())) {
+ LOG(WARNING)
+ << "Agent " << info.id() << " ignoring running "
+ << taskOrTaskGroup(task, taskGroup) << " because "
+ << "it was intended for old agent " << task.slave_id();
+ return;
+ }
}
CHECK(state == RECOVERING || state == DISCONNECTED ||
@@ -1515,8 +1551,9 @@ void Slave::runTask(
// TODO(bmahler): Also ignore if we're DISCONNECTED.
if (state == RECOVERING || state == TERMINATING) {
- LOG(WARNING) << "Ignoring task " << task.task_id()
+ LOG(WARNING) << "Ignoring running " << taskOrTaskGroup(task, taskGroup)
<< " because the agent is " << state;
+
// TODO(vinod): Consider sending a TASK_LOST here.
// Currently it is tricky because 'statusUpdate()'
// ignores updates for unknown frameworks.
@@ -1570,20 +1607,35 @@ void Slave::runTask(
}
}
- const ExecutorInfo executorInfo = getExecutorInfo(frameworkInfo, task);
const ExecutorID& executorId = executorInfo.executor_id();
if (HookManager::hooksAvailable()) {
// Set task labels from run task label decorator.
- task.mutable_labels()->CopyFrom(HookManager::slaveRunTaskLabelDecorator(
- task, executorInfo, frameworkInfo, info));
+ for (auto it = tasks.begin(); it != tasks.end(); ++it) {
+ (*it).mutable_labels()->CopyFrom(
+ HookManager::slaveRunTaskLabelDecorator(
+ (*it), executorInfo, frameworkInfo, info));
+ }
+
+ // Update `task`/`taskGroup` to reflect the task label updates.
+ if (task.isSome()) {
+ CHECK_EQ(1u, tasks.size());
+ task->mutable_labels()->CopyFrom(tasks[0].labels());
+ } else {
+ for (int i = 0; i < taskGroup->tasks().size(); ++i) {
+ taskGroup->mutable_tasks(i)->mutable_labels()->
+ CopyFrom(tasks[i].labels());
+ }
+ }
}
- // We add the task to 'pending' to ensure the framework is not
+ // We add the task/task group to 'pending' to ensure the framework is not
// removed and the framework and top level executor directories
- // are not scheduled for deletion before '_runTask()' is called.
+ // are not scheduled for deletion before '_run()' is called.
CHECK_NOTNULL(framework);
- framework->pending[executorId][task.task_id()] = task;
+ foreach (const TaskInfo& task, tasks) {
+ framework->pending[executorId][task.task_id()] = task;
+ }
// If we are about to create a new executor, unschedule the top
// level work and meta directories from getting gc'ed.
@@ -1606,59 +1658,107 @@ void Slave::runTask(
}
// Run the task after the unschedules are done.
- unschedule.onAny(
- defer(self(), &Self::_runTask, lambda::_1, frameworkInfo, task));
+ unschedule.onAny(defer(
+ self(),
+ &Self::_run,
+ lambda::_1,
+ frameworkInfo,
+ executorInfo,
+ task,
+ taskGroup));
}
-void Slave::_runTask(
+void Slave::_run(
const Future<bool>& future,
const FrameworkInfo& frameworkInfo,
- const TaskInfo& task)
+ const ExecutorInfo& executorInfo,
+ const Option<TaskInfo>& task,
+ const Option<TaskGroupInfo>& taskGroup)
{
- const FrameworkID frameworkId = frameworkInfo.id();
+ CHECK_NE(task.isSome(), taskGroup.isSome())
+ << "Either task or task group should be set but not both";
+
+ vector<TaskInfo> tasks;
+ if (task.isSome()) {
+ tasks.push_back(task.get());
+ } else {
+ foreach (const TaskInfo& task, taskGroup->tasks()) {
+ tasks.push_back(task);
+ }
+ }
+
+ const FrameworkID& frameworkId = frameworkInfo.id();
- LOG(INFO) << "Launching task " << task.task_id()
+ LOG(INFO) << "Launching " << taskOrTaskGroup(task, taskGroup)
<< " for framework " << frameworkId;
Framework* framework = getFramework(frameworkId);
if (framework == nullptr) {
- LOG(WARNING) << "Ignoring run task " << task.task_id()
+ LOG(WARNING) << "Ignoring running " << taskOrTaskGroup(task, taskGroup)
<< " because the framework " << frameworkId
<< " does not exist";
return;
}
- const ExecutorInfo executorInfo = getExecutorInfo(frameworkInfo, task);
const ExecutorID& executorId = executorInfo.executor_id();
- if (framework->pending.contains(executorId) &&
- framework->pending[executorId].contains(task.task_id())) {
- framework->pending[executorId].erase(task.task_id());
- if (framework->pending[executorId].empty()) {
- framework->pending.erase(executorId);
- // NOTE: Ideally we would perform the following check here:
- //
- // if (framework->executors.empty() &&
- // framework->pending.empty()) {
- // removeFramework(framework);
- // }
- //
- // However, we need 'framework' to stay valid for the rest of
- // this function. As such, we perform the check before each of
- // the 'return' statements below.
+ // Remove the task/task group from being pending. If any of the
+ // tasks in the task group have been killed in the interim, we
+ // send a TASK_KILLED for all the other tasks in the group.
+ bool killed = false;
+ foreach (const TaskInfo& task, tasks) {
+ if (framework->pending.contains(executorId) &&
+ framework->pending[executorId].contains(task.task_id())) {
+ framework->pending[executorId].erase(task.task_id());
+ if (framework->pending[executorId].empty()) {
+ framework->pending.erase(executorId);
+ // NOTE: Ideally we would perform the following check here:
+ //
+ // if (framework->executors.empty() &&
+ // framework->pending.empty()) {
+ // removeFramework(framework);
+ // }
+ //
+ // However, we need 'framework' to stay valid for the rest of
+ // this function. As such, we perform the check before each of
+ // the 'return' statements below.
+ }
+ } else {
+ killed = true;
}
- } else {
- LOG(WARNING) << "Ignoring run task " << task.task_id()
+ }
+
+ if (killed) {
+ LOG(WARNING) << "Ignoring running " << taskOrTaskGroup(task, taskGroup)
<< " of framework " << frameworkId
- << " because the task has been killed in the meantime";
+ << " because it has been killed in the meantime";
+
+ foreach (const TaskInfo& task, tasks) {
+ const StatusUpdate update = protobuf::createStatusUpdate(
+ frameworkId,
+ info.id(),
+ task.task_id(),
+ TASK_KILLED,
+ TaskStatus::SOURCE_SLAVE,
+ UUID::random(),
+ "Task killed before it was launched");
+ statusUpdate(update, UPID());
+ }
+
+ // Refer to the comment after 'framework->pending.erase' above
+ // for why we need this.
+ if (framework->executors.empty() && framework->pending.empty()) {
+ removeFramework(framework);
+ }
+
return;
}
// We don't send a status update here because a terminating
// framework cannot send acknowledgements.
if (framework->state == Framework::TERMINATING) {
- LOG(WARNING) << "Ignoring run task " << task.task_id()
+ LOG(WARNING) << "Ignoring running " << taskOrTaskGroup(task, taskGroup)
<< " of framework " << frameworkId
<< " because the framework is terminating";
@@ -1675,22 +1775,24 @@ void Slave::_runTask(
LOG(ERROR) << "Failed to unschedule directories scheduled for gc: "
<< (future.isFailed() ? future.failure() : "future discarded");
- const StatusUpdate update = protobuf::createStatusUpdate(
- frameworkId,
- info.id(),
- task.task_id(),
- TASK_LOST,
- TaskStatus::SOURCE_SLAVE,
- UUID::random(),
- "Could not launch the task because we failed to unschedule directories"
- " scheduled for gc",
- TaskStatus::REASON_GC_ERROR);
-
- // TODO(vinod): Ensure that the status update manager reliably
- // delivers this update. Currently, we don't guarantee this
- // because removal of the framework causes the status update
- // manager to stop retrying for its un-acked updates.
- statusUpdate(update, UPID());
+ foreach (const TaskInfo& task, tasks) {
+ const StatusUpdate update = protobuf::createStatusUpdate(
+ frameworkId,
+ info.id(),
+ task.task_id(),
+ TASK_LOST,
+ TaskStatus::SOURCE_SLAVE,
+ UUID::random(),
+ "Could not launch the task because we failed to unschedule"
+ " directories scheduled for gc",
+ TaskStatus::REASON_GC_ERROR);
+
+ // TODO(vinod): Ensure that the status update manager reliably
+ // delivers this update. Currently, we don't guarantee this
+ // because removal of the framework causes the status update
+ // manager to stop retrying for its un-acked updates.
+ statusUpdate(update, UPID());
+ }
// Refer to the comment after 'framework->pending.erase' above
// for why we need this.
@@ -1701,22 +1803,32 @@ void Slave::_runTask(
return;
}
- // NOTE: If the task or executor uses resources that are
+ // NOTE: If the task/task group or executor uses resources that are
// checkpointed on the slave (e.g. persistent volumes), we should
// already know about it. If the slave doesn't know about them (e.g.
// CheckpointResourcesMessage was dropped or came out of order),
// we send TASK_LOST status updates here since restarting the task
// may succeed in the event that CheckpointResourcesMessage arrives
// out of order.
- Resources checkpointedTaskResources =
- Resources(task.resources()).filter(needCheckpointing);
+ bool kill = false;
+ foreach (const TaskInfo& task, tasks) {
+ Resources checkpointedTaskResources =
+ Resources(task.resources()).filter(needCheckpointing);
- foreach (const Resource& resource, checkpointedTaskResources) {
- if (!checkpointedResources.contains(resource)) {
- LOG(WARNING) << "Unknown checkpointed resource " << resource
- << " for task " << task.task_id()
- << " of framework " << frameworkId;
+ foreach (const Resource& resource, checkpointedTaskResources) {
+ if (!checkpointedResources.contains(resource)) {
+ LOG(WARNING) << "Unknown checkpointed resource " << resource
+ << " for task " << task
+ << " of framework " << frameworkId;
+
+ kill = true;
+ break;
+ }
+ }
+ }
+ if (kill) {
+ foreach (const TaskInfo& task, tasks) {
const StatusUpdate update = protobuf::createStatusUpdate(
frameworkId,
info.id(),
@@ -1724,64 +1836,70 @@ void Slave::_runTask(
TASK_LOST,
TaskStatus::SOURCE_SLAVE,
UUID::random(),
- "The checkpointed resources being used by the task are unknown to "
- "the agent",
+ "The checkpointed resources being used by the task or task group are "
+ "unknown to the agent",
TaskStatus::REASON_RESOURCES_UNKNOWN);
statusUpdate(update, UPID());
+ }
- // Refer to the comment after 'framework->pending.erase' above
- // for why we need this.
- if (framework->executors.empty() && framework->pending.empty()) {
- removeFramework(framework);
- }
-
- return;
+ // Refer to the comment after 'framework->pending.erase' above
+ // for why we need this.
+ if (framework->executors.empty() && framework->pending.empty()) {
+ removeFramework(framework);
}
+
+ return;
}
- if (task.has_executor()) {
- Resources checkpointedExecutorResources =
- Resources(task.executor().resources()).filter(needCheckpointing);
+ CHECK_EQ(kill, false);
+ Resources checkpointedExecutorResources =
+ Resources(executorInfo.resources()).filter(needCheckpointing);
- foreach (const Resource& resource, checkpointedExecutorResources) {
- if (!checkpointedResources.contains(resource)) {
- LOG(WARNING) << "Unknown checkpointed resource " << resource
- << " for executor '" << task.executor().executor_id()
- << "' of framework " << frameworkId;
+ foreach (const Resource& resource, checkpointedExecutorResources) {
+ if (!checkpointedResources.contains(resource)) {
+ LOG(WARNING) << "Unknown checkpointed resource " << resource
+ << " for executor '" << executorId
+ << "' of framework " << frameworkId;
- const StatusUpdate update = protobuf::createStatusUpdate(
- frameworkId,
- info.id(),
- task.task_id(),
- TASK_LOST,
- TaskStatus::SOURCE_SLAVE,
- UUID::random(),
- "The checkpointed resources being used by the executor are unknown "
- "to the agent",
- TaskStatus::REASON_RESOURCES_UNKNOWN,
- task.executor().executor_id());
+ kill = true;
+ break;
+ }
+ }
- statusUpdate(update, UPID());
+ if (kill) {
+ foreach (const TaskInfo& task, tasks) {
+ const StatusUpdate update = protobuf::createStatusUpdate(
+ frameworkId,
+ info.id(),
+ task.task_id(),
+ TASK_LOST,
+ TaskStatus::SOURCE_SLAVE,
+ UUID::random(),
+ "The checkpointed resources being used by the executor are unknown "
+ "to the agent",
+ TaskStatus::REASON_RESOURCES_UNKNOWN,
+ executorId);
- // Refer to the comment after 'framework->pending.erase' above
- // for why we need this.
- if (framework->executors.empty() && framework->pending.empty()) {
- removeFramework(framework);
- }
+ statusUpdate(update, UPID());
+ }
- return;
- }
+ // Refer to the comment after 'framework->pending.erase' above
+ // for why we need this.
+ if (framework->executors.empty() && framework->pending.empty()) {
+ removeFramework(framework);
}
+
+ return;
}
// NOTE: The slave cannot be in 'RECOVERING' because the task would
- // have been rejected in 'runTask()' in that case.
+ // have been rejected in 'run()' in that case.
CHECK(state == DISCONNECTED || state == RUNNING || state == TERMINATING)
<< state;
if (state == TERMINATING) {
- LOG(WARNING) << "Ignoring run task " << task.task_id()
+ LOG(WARNING) << "Ignoring running " << taskOrTaskGroup(task, taskGroup)
<< " of framework " << frameworkId
<< " because the agent is terminating";
@@ -1798,12 +1916,14 @@ void Slave::_runTask(
CHECK(framework->state == Framework::RUNNING) << framework->state;
- // Either send the task to an executor or start a new executor
- // and queue the task until the executor has started.
+ // Either send the task/task group to an executor or start a new executor
+ // and queue it until the executor has started.
Executor* executor = framework->getExecutor(executorId);
if (executor == nullptr) {
- executor = framework->launchExecutor(executorInfo, task);
+ executor = framework->launchExecutor(
+ executorInfo,
+ taskGroup.isNone() ? task.get() : Option<TaskInfo>::none());
}
CHECK_NOTNULL(executor);
@@ -1819,48 +1939,67 @@ void Slave::_runTask(
executorState = "terminated";
}
- LOG(WARNING) << "Asked to run task '" << task.task_id()
+ LOG(WARNING) << "Asked to run " << taskOrTaskGroup(task, taskGroup)
<< "' for framework " << frameworkId
<< " with executor '" << executorId
<< "' which is " << executorState;
- const StatusUpdate update = protobuf::createStatusUpdate(
- frameworkId,
- info.id(),
- task.task_id(),
- TASK_LOST,
- TaskStatus::SOURCE_SLAVE,
- UUID::random(),
- "Executor " + executorState,
- TaskStatus::REASON_EXECUTOR_TERMINATED);
+ foreach (const TaskInfo& task, tasks) {
+ const StatusUpdate update = protobuf::createStatusUpdate(
+ frameworkId,
+ info.id(),
+ task.task_id(),
+ TASK_LOST,
+ TaskStatus::SOURCE_SLAVE,
+ UUID::random(),
+ "Executor " + executorState,
+ TaskStatus::REASON_EXECUTOR_TERMINATED);
+
+ statusUpdate(update, UPID());
+ }
- statusUpdate(update, UPID());
break;
}
case Executor::REGISTERING:
- // Checkpoint the task before we do anything else.
- if (executor->checkpoint) {
- executor->checkpointTask(task);
+ foreach (const TaskInfo& task, tasks) {
+ // Checkpoint the task before we do anything else.
+ if (executor->checkpoint) {
+ executor->checkpointTask(task);
+ }
+
+ // Queue task if the executor has not yet registered.
+ executor->queuedTasks[task.task_id()] = task;
+ }
+
+ if (taskGroup.isSome()) {
+ // Queue task group if the executor has not yet registered.
+ executor->queuedTaskGroups.push_back(taskGroup.get());
}
- // Queue task if the executor has not yet registered.
- LOG(INFO) << "Queuing task '" << task.task_id()
- << "' for executor " << *executor;
+ LOG(INFO) << "Queued " << taskOrTaskGroup(task, taskGroup)
+ << " for executor " << *executor;
- executor->queuedTasks[task.task_id()] = task;
break;
case Executor::RUNNING: {
- // Checkpoint the task before we do anything else.
- if (executor->checkpoint) {
- executor->checkpointTask(task);
+ foreach (const TaskInfo& task, tasks) {
+ // Checkpoint the task before we do anything else.
+ if (executor->checkpoint) {
+ executor->checkpointTask(task);
+ }
+
+ // Queue task until the containerizer is updated with new
+ // resource limits (MESOS-998).
+ executor->queuedTasks[task.task_id()] = task;
}
- // Queue task until the containerizer is updated with new
- // resource limits (MESOS-998).
- LOG(INFO) << "Queuing task '" << task.task_id()
- << "' for executor " << *executor;
+ if (taskGroup.isSome()) {
+ // Queue task group until the containerizer is updated with new
+ // resource limits (MESOS-998).
+ executor->queuedTaskGroups.push_back(taskGroup.get());
+ }
- executor->queuedTasks[task.task_id()] = task;
+ LOG(INFO) << "Queued " << taskOrTaskGroup(task, taskGroup)
+ << " for executor " << *executor;
// Update the resource limits for the container. Note that the
// resource limits include the currently queued tasks because we
@@ -1876,12 +2015,16 @@ void Slave::_runTask(
containerizer->update(executor->containerId, resources)
.onAny(defer(self(),
- &Self::runTasks,
+ &Self::__run,
lambda::_1,
frameworkId,
executorId,
executor->containerId,
- list<TaskInfo>({task})));
+ task.isSome() ? list<TaskInfo>({task.get()})
+ : list<TaskInfo>(),
+ taskGroup.isSome() ? list<TaskGroupInfo>({taskGroup.get()})
+ : list<TaskGroupInfo>()));
+
break;
}
default:
@@ -1897,19 +2040,14 @@ void Slave::_runTask(
}
-void Slave::runTasks(
+void Slave::__run(
const Future<Nothing>& future,
const FrameworkID& frameworkId,
const ExecutorID& executorId,
const ContainerID& containerId,
- const list<TaskInfo>& tasks)
+ const list<TaskInfo>& tasks,
+ const list<TaskGroupInfo>& taskGroups)
{
- // Store all task IDs for logging.
- vector<TaskID> taskIds;
- foreach (const TaskInfo& task, tasks) {
- taskIds.push_back(task.task_id());
- }
-
if (!future.isReady()) {
LOG(ERROR) << "Failed to update resources for container " << containerId
<< " of executor '" << executorId
@@ -1936,9 +2074,42 @@ void Slave::runTasks(
return;
}
+ // Needed for logging.
+ auto tasksAndTaskGroups = [&tasks, &taskGroups]() {
+ ostringstream out;
+ if (!tasks.empty()) {
+ vector<TaskID> taskIds;
+ foreach (const TaskInfo& task, tasks) {
+ taskIds.push_back(task.task_id());
+ }
+ out << "tasks " << stringify(taskIds);
+ }
+
+ if (!taskGroups.empty()) {
+ if (!tasks.empty()) {
+ out << " and ";
+ }
+
+ out << "task groups ";
+
+ vector<vector<TaskID>> taskIds;
+ for (auto it = taskGroups.begin(); it != taskGroups.end(); it++) {
+ vector<TaskID> taskIds_;
+ foreach (const TaskInfo& task, (*it).tasks()) {
+ taskIds_.push_back(task.task_id());
+ }
+ taskIds.push_back(taskIds_);
+ }
+
+ out << stringify(taskIds);
+ }
+
+ return out.str();
+ };
+
Framework* framework = getFramework(frameworkId);
if (framework == nullptr) {
- LOG(WARNING) << "Ignoring sending queued tasks " << taskIds
+ LOG(WARNING) << "Ignoring sending queued " << tasksAndTaskGroups()
<< " to executor '" << executorId
<< "' of framework " << frameworkId
<< " because the framework does not exist";
@@ -1949,7 +2120,7 @@ void Slave::runTasks(
// being shutdown. No need to send status update for the task as
// well because the framework is terminating!
if (framework->state == Framework::TERMINATING) {
- LOG(WARNING) << "Ignoring sending queued tasks " << taskIds
+ LOG(WARNING) << "Ignoring sending queued " << tasksAndTaskGroups()
<< " to executor '" << executorId
<< "' of framework " << frameworkId
<< " because the framework is terminating";
@@ -1958,7 +2129,7 @@ void Slave::runTasks(
Executor* executor = framework->getExecutor(executorId);
if (executor == nullptr) {
- LOG(WARNING) << "Ignoring sending queued tasks " << taskIds
+ LOG(WARNING) << "Ignoring sending queued " << tasksAndTaskGroups()
<< " to executor '" << executorId
<< "' of framework " << frameworkId
<< " because the executor does not exist";
@@ -1970,7 +2141,7 @@ void Slave::runTasks(
// status update as well because it should have already been sent
// when the original instance of the executor was shutting down.
if (executor->containerId != containerId) {
- LOG(WARNING) << "Ignoring sending queued tasks '" << taskIds
+ LOG(WARNING) << "Ignoring sending queued " << tasksAndTaskGroups()
<< "' to executor " << *executor
<< " because the target container " << containerId
<< " has exited";
@@ -1987,7 +2158,7 @@ void Slave::runTasks(
// for the task as well because it will be properly handled by
// 'executorTerminated'.
if (executor->state != Executor::RUNNING) {
- LOG(WARNING) << "Ignoring sending queued tasks " << taskIds
+ LOG(WARNING) << "Ignoring sending queued " << tasksAndTaskGroups()
<< " to executor " << *executor
<< " because the executor is in "
<< executor->state << " state";
@@ -2022,6 +2193,68 @@ void Slave::runTasks(
executor->send(message);
}
+
+ foreach (const TaskGroupInfo& taskGroup, taskGroups) {
+ auto it = find(
+ executor->queuedTaskGroups.begin(),
+ executor->queuedTaskGroups.end(),
+ taskGroup);
+
+ if (it == executor->queuedTaskGroups.end()) {
+ // This is the case where the task group is killed. No need to send
+ // status update because it should be handled in 'killTask'.
+ LOG(WARNING) << "Ignoring sending queued task group "
+ << taskOrTaskGroup(None(), taskGroup) << " to executor "
+ << *executor << " because the task group has been killed";
+ continue;
+ }
+
+ // Add the tasks and send the task group to the executor.
+ foreach (const TaskInfo& task, it->tasks()) {
+ executor->addTask(task);
+ }
+
+ executor->queuedTaskGroups.erase(it);
+
+ executor::Event event;
+ event.set_type(executor::Event::LAUNCH_GROUP);
+
+ executor::Event::LaunchGroup* launchGroup = event.mutable_launch_group();
+ launchGroup->mutable_task_group()->CopyFrom(taskGroup);
+
+ executor->send(event);
+ }
+}
+
+
+void Slave::runTaskGroup(
+ const UPID& from,
+ const FrameworkInfo& frameworkInfo,
+ const ExecutorInfo& executorInfo,
+ const TaskGroupInfo& taskGroupInfo)
+{
+ if (master != from) {
+ LOG(WARNING) << "Ignoring run task group message from " << from
+ << " because it is not the expected master: "
+ << (master.isSome() ? stringify(master.get()) : "None");
+ return;
+ }
+
+ if (!frameworkInfo.has_id()) {
+ LOG(ERROR) << "Ignoring run task group message from " << from
+ << " because it does not have a framework ID";
+ return;
+ }
+
+ if (taskGroupInfo.tasks().empty()) {
+ LOG(ERROR) << "Ignoring run task group message from " << from
+ << " for framework " << frameworkInfo.id()
+ << " because it has no tasks";
+ return;
+ }
+
+ // TODO(anand): Populate command info correctly for the `DEFAULT` executor.
+ run(frameworkInfo, executorInfo, None(), taskGroupInfo, UPID());
}
@@ -2084,24 +2317,16 @@ void Slave::killTask(
<< " of framework " << frameworkId
<< " before it was launched";
- const StatusUpdate update = protobuf::createStatusUpdate(
- frameworkId,
- info.id(),
- taskId,
- TASK_KILLED,
- TaskStatus::SOURCE_SLAVE,
- UUID::random(),
- "Task killed before it was launched");
- statusUpdate(update, UPID());
-
- framework->pending[executorId].erase(taskId);
- if (framework->pending[executorId].empty()) {
- framework->pending.erase(executorId);
- if (framework->pending.empty() && framework->executors.empty()) {
- removeFramework(framework);
- }
- }
- return;
+ // We send the TASK_KILLED status update in `_run()` as the
+ // task being killed could be part of a task group and we
+ // don't store this information in `framework->pending`.
+ // We don't invoke `removeFramework()` here since we need the
+ // framework to be valid for sending the status update later.
+ framework->pending[executorId].erase(taskId);
+ if (framework->pending[executorId].empty()) {
+ framework->pending.erase(executorId);
+ }
+ return;
}
}
@@ -2132,22 +2357,45 @@ void Slave::killTask(
<< " of framework " << frameworkId
<< " to TASK_KILLED because the executor is not registered";
- // The executor hasn't registered yet.
- const StatusUpdate update = protobuf::createStatusUpdate(
- frameworkId,
- info.id(),
- taskId,
- TASK_KILLED,
- TaskStatus::SOURCE_SLAVE,
- UUID::random(),
- "Unregistered executor",
- TaskStatus::REASON_EXECUTOR_UNREGISTERED,
- executor->id);
+ // This task might be part of a task group. If so, we need to
+ // send a TASK_KILLED update for all tasks in the group.
+ Option<TaskGroupInfo> taskGroup = executor->getQueuedTaskGroup(taskId);
+
+ std::list<StatusUpdate> updates;
+ if (taskGroup.isSome()) {
+ foreach (const TaskInfo& task, taskGroup->tasks()) {
+ updates.push_back(protobuf::createStatusUpdate(
+ frameworkId,
+ info.id(),
+ task.task_id(),
+ TASK_KILLED,
+ TaskStatus::SOURCE_SLAVE,
+ UUID::random(),
+ "Unregistered executor",
+ TaskStatus::REASON_EXECUTOR_UNREGISTERED,
+ executor->id));
+ }
+ } else {
+ updates.push_back(protobuf::createStatusUpdate(
+ frameworkId,
+ info.id(),
+ taskId,
+ TASK_KILLED,
+ TaskStatus::SOURCE_SLAVE,
+ UUID::random(),
+ "Unregistered executor",
+ TaskStatus::REASON_EXECUTOR_UNREGISTERED,
+ executor->id));
+ }
+
+ foreach (const StatusUpdate& update, updates) {
+ // NOTE: Sending a terminal update (TASK_KILLED) removes the
+ // task/task group from 'executor->queuedTasks' and
+ // 'executor->queuedTaskGroup', so that if the executor registers at
+ // a later point in time, it won't get this task or task group.
+ statusUpdate(update, UPID());
+ }
- // NOTE: Sending a terminal update (TASK_KILLED) removes the
- // task from 'executor->queuedTasks', so that if the executor
- // registers at a later point in time, it won't get this task.
- statusUpdate(update, UPID());
break;
}
case Executor::TERMINATING:
@@ -2165,20 +2413,45 @@ void Slave::killTask(
// This is the case where the task has not yet been sent to
// the executor (e.g., waiting for containerizer update to
// finish).
- const StatusUpdate update = protobuf::createStatusUpdate(
- frameworkId,
- info.id(),
- taskId,
- TASK_KILLED,
- TaskStatus::SOURCE_SLAVE,
- UUID::random(),
- "Task killed when it was queued",
- None(),
- executor->id);
- // NOTE: Sending a terminal update (TASK_KILLED) removes the
- // task from 'executor->queuedTasks'.
- statusUpdate(update, UPID());
+ // This task might be part of a task group. If so, we need to
+ // send a TASK_KILLED update for all the other tasks.
+ Option<TaskGroupInfo> taskGroup = executor->getQueuedTaskGroup(taskId);
+
+ std::list<StatusUpdate> updates;
+ if (taskGroup.isSome()) {
+ foreach (const TaskInfo& task, taskGroup->tasks()) {
+ updates.push_back(protobuf::createStatusUpdate(
+ frameworkId,
+ info.id(),
+ task.task_id(),
+ TASK_KILLED,
+ TaskStatus::SOURCE_SLAVE,
+ UUID::random(),
+ "Task killed while it was queued",
+ None(),
+ executor->id));
+ }
+ } else {
+ updates.push_back(protobuf::createStatusUpdate(
+ frameworkId,
+ info.id(),
+ taskId,
+ TASK_KILLED,
+ TaskStatus::SOURCE_SLAVE,
+ UUID::random(),
+ "Task killed while it was queued",
+ None(),
+ executor->id));
+ }
+
+ foreach (const StatusUpdate& update, updates) {
+ // NOTE: Sending a terminal update (TASK_KILLED) removes the
+ // task/task group from 'executor->queuedTasks' and
+ // 'executor->queuedTaskGroup', so that if the executor registers at
+ // a later point in time, it won't get this task.
+ statusUpdate(update, UPID());
+ }
} else {
// Send a message to the executor and wait for
// it to send us a status update.
@@ -2786,7 +3059,8 @@ void Slave::subscribe(
CHECK_SOME(os::touch(path));
}
- // Tell executor it's registered and give it any queued tasks.
+ // Tell executor it's registered and give it any queued tasks
+ // or task groups.
ExecutorRegisteredMessage message;
message.mutable_executor_info()->MergeFrom(executor->info);
message.mutable_framework_id()->MergeFrom(framework->id());
@@ -2822,14 +3096,32 @@ void Slave::subscribe(
resources += task.resources();
}
+ // We maintain a copy of the tasks in `queuedTaskGroups` also in
+ // `queuedTasks`. Hence, we need to ensure that we don't send the same
+ // tasks to the executor twice.
+ LinkedHashMap<TaskID, TaskInfo> queuedTasks;
+ foreach (const TaskID& taskId, executor->queuedTasks.keys()) {
+ queuedTasks[taskId] = executor->queuedTasks[taskId];
+ }
+
+ foreach (const TaskGroupInfo& taskGroup, executor->queuedTaskGroups) {
+ foreach (const TaskInfo& task, taskGroup.tasks()) {
+ const TaskID& taskId = task.task_id();
+ if (queuedTasks.contains(taskId)) {
+ queuedTasks.erase(taskId);
+ }
+ }
+ }
+
containerizer->update(executor->containerId, resources)
.onAny(defer(self(),
- &Self::runTasks,
+ &Self::__run,
lambda::_1,
framework->id(),
executor->id,
executor->containerId,
- executor->queuedTasks.values()));
+ queuedTasks.values(),
+ executor->queuedTaskGroups));
hashmap<TaskID, TaskInfo> unackedTasks;
foreach (const TaskInfo& task, subscribe.unacknowledged_tasks()) {
@@ -2997,7 +3289,8 @@ void Slave::registerExecutor(
return;
}
- // Tell executor it's registered and give it any queued tasks.
+ // Tell executor it's registered and give it any queued tasks
+ // or task groups.
ExecutorRegisteredMessage message;
message.mutable_executor_info()->MergeFrom(executor->info);
message.mutable_framework_id()->MergeFrom(framework->id());
@@ -3018,14 +3311,33 @@ void Slave::registerExecutor(
resources += task.resources();
}
+ // We maintain a copy of the tasks in `queuedTaskGroups` also in
+ // `queuedTasks`. Hence, we need to ensure that we don't send the same
+ // tasks to the executor twice.
+ LinkedHashMap<TaskID, TaskInfo> queuedTasks;
+ foreach (const TaskID& taskId, executor->queuedTasks.keys()) {
+ queuedTasks[taskId] = executor->queuedTasks[taskId];
+ }
+
+ foreach (const TaskGroupInfo& taskGroup, executor->queuedTaskGroups) {
+ foreach (const TaskInfo& task, taskGroup.tasks()) {
+ const TaskID& taskId = task.task_id();
+ if (queuedTasks.contains(taskId)) {
+ queuedTasks.erase(taskId);
+ }
+ }
+ }
+
containerizer->update(executor->containerId, resources)
.onAny(defer(self(),
- &Self::runTasks,
+ &Self::__run,
lambda::_1,
frameworkId,
executorId,
executor->containerId,
- executor->queuedTasks.values()));
+ queuedTasks.values(),
+ executor->queuedTaskGroups));
+
break;
}
default:
@@ -3354,7 +3666,7 @@ void Slave::statusUpdate(StatusUpdate update, const Option<UPID>& pid)
// NOTE: We forward the update here because this update could be
// generated by the slave when the executor is unknown to it
- // (e.g., killTask(), _runTask()) or sent by an executor for a
+ // (e.g., killTask(), _run()) or sent by an executor for a
// task that belongs to another executor.
// We also end up here if 1) the previous slave died after
// checkpointing a _terminal_ update but before it could send an
@@ -6231,27 +6543,43 @@ Try<Nothing> Executor::updateTaskState(const TaskStatus& status)
bool terminal = protobuf::isTerminalState(status.state());
const TaskID& taskId = status.task_id();
- Task* task = nullptr;
+ std::list<Task*> tasks;
if (queuedTasks.contains(taskId)) {
if (terminal) {
- task = new Task(protobuf::createTask(
+ Task* task = new Task(protobuf::createTask(
queuedTasks.at(taskId),
status.state(),
frameworkId));
+
queuedTasks.erase(taskId);
- terminatedTasks[taskId] = task;
+
+ // This might be a queued task belonging to a task group.
+ // If so, we need to update the other tasks belonging to this task group.
+ Option<TaskGroupInfo> taskGroup = getQueuedTaskGroup(taskId);
+
+ if (taskGroup.isSome()) {
+ foreach (const TaskInfo& task_, taskGroup->tasks()) {
+ Task* task = new Task(
+ protobuf::createTask(task_, status.state(), frameworkId));
+
+ tasks.push_back(task);
+ }
+ } else {
+ tasks.push_back(task);
+ }
} else {
return Error("Cannot send non-terminal update for queued task");
}
} else if (launchedTasks.contains(taskId)) {
- task = launchedTasks.at(status.task_id());
+ Task* task = launchedTasks.at(status.task_id());
if (terminal) {
resources -= task->resources(); // Release the resources.
launchedTasks.erase(taskId);
- terminatedTasks[taskId] = task;
}
+
+ tasks.push_back(task);
} else if (terminatedTasks.contains(taskId)) {
return Error("Task is already terminated with state"
" " + stringify(terminatedTasks.at(taskId)->state()));
@@ -6259,27 +6587,31 @@ Try<Nothing> Executor::updateTaskState(const TaskStatus& status)
return Error("Task is unknown");
}
- CHECK_NOTNULL(task);
+ foreach (Task* task, tasks) {
+ CHECK_NOTNULL(task);
+ // TODO(brenden): Consider wiping the `data` and `message` fields?
+ if (task->statuses_size() > 0 &&
+ task->statuses(task->statuses_size() - 1).state() == status.state()) {
+ task->mutable_statuses()->RemoveLast();
+ }
- // TODO(brenden): Consider wiping the `data` and `message` fields?
- if (task->statuses_size() > 0 &&
- task->statuses(task->statuses_size() - 1).state() == status.state()) {
- task->mutable_statuses()->RemoveLast();
- }
- task->add_statuses()->CopyFrom(status);
- task->set_state(status.state());
+ task->add_statuses()->CopyFrom(status);
+ task->set_state(status.state());
- // TODO(bmahler): This only increments the state when the update
- // can be handled. Should we always increment the state?
- if (terminal) {
- switch (status.state()) {
- case TASK_FINISHED: ++slave->metrics.tasks_finished; break;
- case TASK_FAILED: ++slave->metrics.tasks_failed; break;
- case TASK_KILLED: ++slave->metrics.tasks_killed; break;
- case TASK_LOST: ++slave->metrics.tasks_lost; break;
- default:
- LOG(ERROR) << "Unexpected terminal task state " << status.state();
- break;
+ // TODO(bmahler): This only increments the state when the update
+ // can be handled. Should we always increment the state?
+ if (terminal) {
+ terminatedTasks[task->task_id()] = task;
+
+ switch (status.state()) {
+ case TASK_FINISHED: ++slave->metrics.tasks_finished; break;
+ case TASK_FAILED: ++slave->metrics.tasks_failed; break;
+ case TASK_KILLED: ++slave->metrics.tasks_killed; break;
+ case TASK_LOST: ++slave->metrics.tasks_lost; break;
+ default:
+ LOG(ERROR) << "Unexpected terminal task state " << status.state();
+ break;
+ }
}
}
@@ -6313,6 +6645,20 @@ void Executor::closeHttpConnection()
}
+Option<TaskGroupInfo> Executor::getQueuedTaskGroup(const TaskID& taskId)
+{
+ foreach (const TaskGroupInfo& taskGroup, queuedTaskGroups) {
+ foreach (const TaskInfo& taskInfo, taskGroup.tasks()) {
+ if (taskInfo.task_id() == taskId) {
+ return taskGroup;
+ }
+ }
+ }
+
+ return None();
+}
+
+
map<string, string> executorEnvironment(
const Flags& flags,
const ExecutorInfo& executorInfo,
@@ -6494,6 +6840,27 @@ std::ostream& operator<<(std::ostream& stream, Executor::State state)
}
}
+
+string taskOrTaskGroup(
+ const Option<TaskInfo>& task,
+ const Option<TaskGroupInfo>& taskGroup)
+{
+ ostringstream out;
+ if (task.isSome()) {
+ out << "task '" << task->task_id() << "'";
+ } else {
+ CHECK_SOME(taskGroup);
+
+ vector<TaskID> taskIds;
+ foreach (const TaskInfo& task, taskGroup->tasks()) {
+ taskIds.push_back(task.task_id());
+ }
+ out << "task group containing tasks " << taskIds;
+ }
+
+ return out.str();
+}
+
} // namespace slave {
} // namespace internal {
} // namespace mesos {
http://git-wip-us.apache.org/repos/asf/mesos/blob/3a4c3f14/src/slave/slave.hpp
----------------------------------------------------------------------
diff --git a/src/slave/slave.hpp b/src/slave/slave.hpp
index ec02f63..903a48d 100644
--- a/src/slave/slave.hpp
+++ b/src/slave/slave.hpp
@@ -138,13 +138,29 @@ public:
const FrameworkInfo& frameworkInfo,
const FrameworkID& frameworkId,
const process::UPID& pid,
- TaskInfo task);
+ const TaskInfo& task);
+
+ void run(
+ const FrameworkInfo& frameworkInfo,
+ const ExecutorInfo& executorInfo,
+ Option<TaskInfo> task,
+ Option<TaskGroupInfo> taskGroup,
+ const UPID& pid);
// Made 'virtual' for Slave mocking.
- virtual void _runTask(
- const process::Future<bool>& future,
+ virtual void _run(
+ const process::Future<bool>& future,
+ const FrameworkInfo& frameworkInfo,
+ const ExecutorInfo& executorInfo,
+ const Option<TaskInfo>& task,
+ const Option<TaskGroupInfo>& taskGroup);
+
+ // Made 'virtual' for Slave mocking.
+ virtual void runTaskGroup(
+ const process::UPID& upid,
const FrameworkInfo& frameworkInfo,
- const TaskInfo& task);
+ const ExecutorInfo& executorInfo,
+ const TaskGroupInfo& taskGroupInfo);
process::Future<bool> unschedule(const std::string& path);
@@ -306,17 +322,19 @@ public:
virtual void exited(const process::UPID& pid);
// This is called when the resource limits of the container have
- // been updated for the given tasks. If the update is successful, we
- // flush the given tasks to the executor by sending RunTaskMessages.
+ // been updated for the given tasks and task groups. If the update is
+ // successful, we flush the given tasks to the executor by sending
+ // RunTaskMessages or `LAUNCH_GROUP` events.
// TODO(jieyu): Consider renaming it to '__runTasks' once the slave
// starts to support launching multiple tasks in one call (i.e.,
// multi-tasks version of 'runTask').
- void runTasks(
+ void __run(
const process::Future<Nothing>& future,
const FrameworkID& frameworkId,
const ExecutorID& executorId,
const ContainerID& containerId,
- const std::list<TaskInfo>& tasks);
+ const std::list<TaskInfo>& tasks,
+ const std::list<TaskGroupInfo>& taskGroups);
void fileAttached(const process::Future<Nothing>& result,
const std::string& path);
@@ -824,6 +842,9 @@ struct Executor
// Closes the HTTP connection.
void closeHttpConnection();
+ // Returns the task group associated with the task.
+ Option<TaskGroupInfo> getQueuedTaskGroup(const TaskID& taskId);
+
friend std::ostream& operator<<(
std::ostream& stream,
const Executor& executor);
@@ -882,9 +903,16 @@ struct Executor
// Tasks can be found in one of the following four data structures:
- // Not yet launched.
+ // Not yet launched tasks. This also includes tasks from `queuedTaskGroups`.
LinkedHashMap<TaskID, TaskInfo> queuedTasks;
+ // Not yet launched task groups. This is needed for correctly sending
+ // TASK_KILLED status updates for all tasks in the group if any of the
+ // tasks were killed before the executor could register with the agent.
+ //
+ // TODO(anand): Replace this with `LinkedHashSet` when it is available.
+ std::list<TaskGroupInfo> queuedTaskGroups;
+
// Running.
LinkedHashMap<TaskID, Task*> launchedTasks;
@@ -995,6 +1023,12 @@ std::ostream& operator<<(std::ostream& stream, Slave::State state);
std::ostream& operator<<(std::ostream& stream, Framework::State state);
std::ostream& operator<<(std::ostream& stream, Executor::State state);
+
+// Needed for logging task/task group.
+std::string taskOrTaskGroup(
+ const Option<TaskInfo>& task,
+ const Option<TaskGroupInfo>& taskGroup);
+
} // namespace slave {
} // namespace internal {
} // namespace mesos {
[7/8] mesos git commit: Added mock overloads for
`runTaskGroup()`/`launchGroup()`.
Posted by an...@apache.org.
Added mock overloads for `runTaskGroup()`/`launchGroup()`.
Review: https://reviews.apache.org/r/51479/
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/f8734b7a
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/f8734b7a
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/f8734b7a
Branch: refs/heads/master
Commit: f8734b7a2572ac48fbd1cb763e732d68d7752529
Parents: 8f9f17b
Author: Anand Mazumdar <an...@apache.org>
Authored: Mon Sep 12 14:22:54 2016 -0700
Committer: Anand Mazumdar <an...@apache.org>
Committed: Mon Sep 12 15:17:02 2016 -0700
----------------------------------------------------------------------
src/tests/mesos.cpp | 12 ++++++++++++
src/tests/mesos.hpp | 15 ++++++++++++++-
2 files changed, 26 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/f8734b7a/src/tests/mesos.cpp
----------------------------------------------------------------------
diff --git a/src/tests/mesos.cpp b/src/tests/mesos.cpp
index 4ec21e7..539634e 100644
--- a/src/tests/mesos.cpp
+++ b/src/tests/mesos.cpp
@@ -543,6 +543,8 @@ MockSlave::MockSlave(
.WillRepeatedly(Invoke(this, &MockSlave::unmocked_runTask));
EXPECT_CALL(*this, _run(_, _, _, _, _))
.WillRepeatedly(Invoke(this, &MockSlave::unmocked__run));
+ EXPECT_CALL(*this, runTaskGroup(_, _, _, _))
+ .WillRepeatedly(Invoke(this, &MockSlave::unmocked_runTaskGroup));
EXPECT_CALL(*this, killTask(_, _))
.WillRepeatedly(Invoke(this, &MockSlave::unmocked_killTask));
EXPECT_CALL(*this, removeFramework(_))
@@ -585,6 +587,16 @@ void MockSlave::unmocked__run(
}
+void MockSlave::unmocked_runTaskGroup(
+ const process::UPID& from,
+ const FrameworkInfo& frameworkInfo,
+ const ExecutorInfo& executorInfo,
+ const TaskGroupInfo& taskGroup)
+{
+ slave::Slave::runTaskGroup(from, frameworkInfo, executorInfo, taskGroup);
+}
+
+
void MockSlave::unmocked_killTask(
const UPID& from,
const KillTaskMessage& killTaskMessage)
http://git-wip-us.apache.org/repos/asf/mesos/blob/f8734b7a/src/tests/mesos.hpp
----------------------------------------------------------------------
diff --git a/src/tests/mesos.hpp b/src/tests/mesos.hpp
index 946e03d..92d7d94 100644
--- a/src/tests/mesos.hpp
+++ b/src/tests/mesos.hpp
@@ -1140,6 +1140,7 @@ public:
MOCK_METHOD1_T(disconnected, void(Mesos*));
MOCK_METHOD2_T(subscribed, void(Mesos*, const typename Event::Subscribed&));
MOCK_METHOD2_T(launch, void(Mesos*, const typename Event::Launch&));
+ MOCK_METHOD2_T(launchGroup, void(Mesos*, const typename Event::LaunchGroup&));
MOCK_METHOD2_T(kill, void(Mesos*, const typename Event::Kill&));
MOCK_METHOD2_T(message, void(Mesos*, const typename Event::Message&));
MOCK_METHOD1_T(shutdown, void(Mesos*));
@@ -1157,7 +1158,7 @@ public:
launch(mesos, event.launch());
break;
case Event::LAUNCH_GROUP:
- // TODO(vinod): Implement this.
+ launchGroup(mesos, event.launch_group());
break;
case Event::KILL:
kill(mesos, event.kill());
@@ -1380,6 +1381,18 @@ public:
const Option<TaskInfo>& task,
const Option<TaskGroupInfo>& taskGroup);
+ MOCK_METHOD4(runTaskGroup, void(
+ const process::UPID& from,
+ const FrameworkInfo& frameworkInfo,
+ const ExecutorInfo& executorInfo,
+ const TaskGroupInfo& taskGroup));
+
+ void unmocked_runTaskGroup(
+ const process::UPID& from,
+ const FrameworkInfo& frameworkInfo,
+ const ExecutorInfo& executorInfo,
+ const TaskGroupInfo& taskGroup);
+
MOCK_METHOD2(killTask, void(
const process::UPID& from,
const KillTaskMessage& killTaskMessage));
[2/8] mesos git commit: Modified tests to account for
`runTask()`/`_runTask()` refactoring.
Posted by an...@apache.org.
Modified tests to account for `runTask()`/`_runTask()` refactoring.
This change modified the agent mocks/tests to account for the
\`runTask()\`/\`\_runTask()\` refactoring.
Review: https://reviews.apache.org/r/51478/
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/8f9f17bb
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/8f9f17bb
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/8f9f17bb
Branch: refs/heads/master
Commit: 8f9f17bbdbcd08b46037816bc3d23e66aa3d2f16
Parents: aaa353a
Author: Anand Mazumdar <an...@apache.org>
Authored: Mon Sep 12 14:22:45 2016 -0700
Committer: Anand Mazumdar <an...@apache.org>
Committed: Mon Sep 12 15:17:02 2016 -0700
----------------------------------------------------------------------
src/tests/master_slave_reconciliation_tests.cpp | 6 +--
src/tests/mesos.cpp | 15 ++++---
src/tests/mesos.hpp | 16 +++++---
src/tests/slave_tests.cpp | 43 ++++++++++++--------
4 files changed, 47 insertions(+), 33 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/8f9f17bb/src/tests/master_slave_reconciliation_tests.cpp
----------------------------------------------------------------------
diff --git a/src/tests/master_slave_reconciliation_tests.cpp b/src/tests/master_slave_reconciliation_tests.cpp
index 7dda6f6..2983c1b 100644
--- a/src/tests/master_slave_reconciliation_tests.cpp
+++ b/src/tests/master_slave_reconciliation_tests.cpp
@@ -426,9 +426,9 @@ TEST_F(MasterSlaveReconciliationTest, SlaveReregisterPendingTask)
EXPECT_CALL(sched, statusUpdate(&driver, _))
.Times(0);
- // We drop the _runTask dispatch to ensure the task remains
+ // We drop the _run dispatch to ensure the task remains
// pending in the slave.
- Future<Nothing> _runTask = DROP_DISPATCH(slave.get()->pid, &Slave::_runTask);
+ Future<Nothing> _run = DROP_DISPATCH(slave.get()->pid, &Slave::_run);
TaskInfo task1;
task1.set_name("test task");
@@ -439,7 +439,7 @@ TEST_F(MasterSlaveReconciliationTest, SlaveReregisterPendingTask)
driver.launchTasks(offers.get()[0].id(), {task1});
- AWAIT_READY(_runTask);
+ AWAIT_READY(_run);
Future<SlaveReregisteredMessage> slaveReregisteredMessage =
FUTURE_PROTOBUF(SlaveReregisteredMessage(), _, _);
http://git-wip-us.apache.org/repos/asf/mesos/blob/8f9f17bb/src/tests/mesos.cpp
----------------------------------------------------------------------
diff --git a/src/tests/mesos.cpp b/src/tests/mesos.cpp
index f5034f9..4ec21e7 100644
--- a/src/tests/mesos.cpp
+++ b/src/tests/mesos.cpp
@@ -541,8 +541,8 @@ MockSlave::MockSlave(
// Set up default behaviors, calling the original methods.
EXPECT_CALL(*this, runTask(_, _, _, _, _))
.WillRepeatedly(Invoke(this, &MockSlave::unmocked_runTask));
- EXPECT_CALL(*this, _runTask(_, _, _))
- .WillRepeatedly(Invoke(this, &MockSlave::unmocked__runTask));
+ EXPECT_CALL(*this, _run(_, _, _, _, _))
+ .WillRepeatedly(Invoke(this, &MockSlave::unmocked__run));
EXPECT_CALL(*this, killTask(_, _))
.WillRepeatedly(Invoke(this, &MockSlave::unmocked_killTask));
EXPECT_CALL(*this, removeFramework(_))
@@ -567,18 +567,21 @@ void MockSlave::unmocked_runTask(
const FrameworkInfo& frameworkInfo,
const FrameworkID& frameworkId,
const UPID& pid,
- TaskInfo task)
+ const TaskInfo& task)
{
slave::Slave::runTask(from, frameworkInfo, frameworkInfo.id(), pid, task);
}
-void MockSlave::unmocked__runTask(
+void MockSlave::unmocked__run(
const Future<bool>& future,
const FrameworkInfo& frameworkInfo,
- const TaskInfo& task)
+ const ExecutorInfo& executorInfo,
+ const Option<TaskInfo>& taskInfo,
+ const Option<TaskGroupInfo>& taskGroup)
{
- slave::Slave::_runTask(future, frameworkInfo, task);
+ slave::Slave::_run(
+ future, frameworkInfo, executorInfo, taskInfo, taskGroup);
}
http://git-wip-us.apache.org/repos/asf/mesos/blob/8f9f17bb/src/tests/mesos.hpp
----------------------------------------------------------------------
diff --git a/src/tests/mesos.hpp b/src/tests/mesos.hpp
index 9bcdaf1..946e03d 100644
--- a/src/tests/mesos.hpp
+++ b/src/tests/mesos.hpp
@@ -1357,24 +1357,28 @@ public:
const FrameworkInfo& frameworkInfo,
const FrameworkID& frameworkId,
const process::UPID& pid,
- TaskInfo task));
+ const TaskInfo& task));
void unmocked_runTask(
const process::UPID& from,
const FrameworkInfo& frameworkInfo,
const FrameworkID& frameworkId,
const process::UPID& pid,
- TaskInfo task);
+ const TaskInfo& task);
- MOCK_METHOD3(_runTask, void(
+ MOCK_METHOD5(_run, void(
const process::Future<bool>& future,
const FrameworkInfo& frameworkInfo,
- const TaskInfo& task));
+ const ExecutorInfo& executorInfo,
+ const Option<TaskInfo>& task,
+ const Option<TaskGroupInfo>& taskGroup));
- void unmocked__runTask(
+ void unmocked__run(
const process::Future<bool>& future,
const FrameworkInfo& frameworkInfo,
- const TaskInfo& task);
+ const ExecutorInfo& executorInfo,
+ const Option<TaskInfo>& task,
+ const Option<TaskGroupInfo>& taskGroup);
MOCK_METHOD2(killTask, void(
const process::UPID& from,
http://git-wip-us.apache.org/repos/asf/mesos/blob/8f9f17bb/src/tests/slave_tests.cpp
----------------------------------------------------------------------
diff --git a/src/tests/slave_tests.cpp b/src/tests/slave_tests.cpp
index 701fab2..55dba3f 100644
--- a/src/tests/slave_tests.cpp
+++ b/src/tests/slave_tests.cpp
@@ -2807,7 +2807,7 @@ TEST_F(SlaveTest, CancelSlaveRemoval)
// This test ensures that a killTask() can happen between runTask()
-// and _runTask() and then gets "handled properly". This means that
+// and _run() and then gets "handled properly". This means that
// the task never gets started, but also does not get lost. The end
// result is status TASK_KILLED. Essentially, killing the task is
// realized while preparing to start it. See MESOS-947. This test
@@ -2866,30 +2866,27 @@ TEST_F(SlaveTest, KillTaskBetweenRunTaskParts)
EXPECT_CALL(slave, runTask(_, _, _, _, _))
.WillOnce(Invoke(&slave, &MockSlave::unmocked_runTask));
- // Saved arguments from Slave::_runTask().
+ // Saved arguments from Slave::_run().
Future<bool> future;
FrameworkInfo frameworkInfo;
-
- // Skip what Slave::_runTask() normally does, save its arguments for
+ ExecutorInfo executorInfo;
+ Option<TaskGroupInfo> taskGroup;
+ Option<TaskInfo> task_;
+ // Skip what Slave::_run() normally does, save its arguments for
// later, tie reaching the critical moment when to kill the task to
// a future.
- Future<Nothing> _runTask;
- EXPECT_CALL(slave, _runTask(_, _, _))
- .WillOnce(DoAll(FutureSatisfy(&_runTask),
+ Future<Nothing> _run;
+ EXPECT_CALL(slave, _run(_, _, _, _, _))
+ .WillOnce(DoAll(FutureSatisfy(&_run),
SaveArg<0>(&future),
- SaveArg<1>(&frameworkInfo)));
+ SaveArg<1>(&frameworkInfo),
+ SaveArg<2>(&executorInfo),
+ SaveArg<3>(&task_),
+ SaveArg<4>(&taskGroup)));
driver.launchTasks(offers.get()[0].id(), {task});
- AWAIT_READY(_runTask);
-
- // Since this is the only task ever for this framework, the
- // framework should get removed in Slave::killTask().
- // Thus we can observe that this happens before Shutdown().
- Future<Nothing> removeFramework;
- EXPECT_CALL(slave, removeFramework(_))
- .WillOnce(DoAll(Invoke(&slave, &MockSlave::unmocked_removeFramework),
- FutureSatisfy(&removeFramework)));
+ AWAIT_READY(_run);
Future<Nothing> killTask;
EXPECT_CALL(slave, killTask(_, _))
@@ -2899,7 +2896,17 @@ TEST_F(SlaveTest, KillTaskBetweenRunTaskParts)
driver.killTask(task.task_id());
AWAIT_READY(killTask);
- slave.unmocked__runTask(future, frameworkInfo, task);
+
+ // Since this is the only task ever for this framework, the
+ // framework should get removed in Slave::_run().
+ // Thus we can observe that this happens before Shutdown().
+ Future<Nothing> removeFramework;
+ EXPECT_CALL(slave, removeFramework(_))
+ .WillOnce(DoAll(Invoke(&slave, &MockSlave::unmocked_removeFramework),
+ FutureSatisfy(&removeFramework)));
+
+ slave.unmocked__run(
+ future, frameworkInfo, executorInfo, task_, taskGroup);
AWAIT_READY(removeFramework);
[3/8] mesos git commit: Made the `TaskInfo` argument in
`launchExecutor()` optional.
Posted by an...@apache.org.
Made the `TaskInfo` argument in `launchExecutor()` optional.
Launching a task group requires just the `ExecutorInfo`. We had
already refactored the containerizer `launch()` interface to make
`TaskInfo` as an optional. This change funnels it through.
Review: https://reviews.apache.org/r/51476/
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/3e9ea3b2
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/3e9ea3b2
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/3e9ea3b2
Branch: refs/heads/master
Commit: 3e9ea3b2090236514bfd5c6a47b9d42e6e306aad
Parents: 9482755
Author: Anand Mazumdar <an...@apache.org>
Authored: Mon Sep 12 14:21:23 2016 -0700
Committer: Anand Mazumdar <an...@apache.org>
Committed: Mon Sep 12 15:17:02 2016 -0700
----------------------------------------------------------------------
src/slave/slave.cpp | 12 ++++++++----
src/slave/slave.hpp | 4 +++-
2 files changed, 11 insertions(+), 5 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/3e9ea3b2/src/slave/slave.cpp
----------------------------------------------------------------------
diff --git a/src/slave/slave.cpp b/src/slave/slave.cpp
index 2da5a29..fa3610d 100644
--- a/src/slave/slave.cpp
+++ b/src/slave/slave.cpp
@@ -5727,7 +5727,7 @@ Framework::~Framework()
// Create and launch an executor.
Executor* Framework::launchExecutor(
const ExecutorInfo& executorInfo,
- const TaskInfo& taskInfo)
+ const Option<TaskInfo>& taskInfo)
{
// Generate an ID for the executor's container.
// TODO(idownes) This should be done by the containerizer but we
@@ -5800,13 +5800,17 @@ Executor* Framework::launchExecutor(
.onAny(defer(slave, &Slave::fileAttached, lambda::_1, executor->directory));
// Tell the containerizer to launch the executor.
+ ExecutorInfo executorInfo_ = executor->info;
+ Resources resources = executorInfo_.resources();
+
// NOTE: We modify the ExecutorInfo to include the task's
// resources when launching the executor so that the containerizer
// has non-zero resources to work with when the executor has
// no resources. This should be revisited after MESOS-600.
- ExecutorInfo executorInfo_ = executor->info;
- Resources resources = executorInfo_.resources();
- resources += taskInfo.resources();
+ if (taskInfo.isSome()) {
+ resources += taskInfo->resources();
+ }
+
executorInfo_.mutable_resources()->CopyFrom(resources);
// Prepare environment variables for the executor.
http://git-wip-us.apache.org/repos/asf/mesos/blob/3e9ea3b2/src/slave/slave.hpp
----------------------------------------------------------------------
diff --git a/src/slave/slave.hpp b/src/slave/slave.hpp
index 4add4c0..ec02f63 100644
--- a/src/slave/slave.hpp
+++ b/src/slave/slave.hpp
@@ -922,9 +922,11 @@ struct Framework
~Framework();
+ // If an executor is launched for a task group, `taskInfo` would
+ // not be set.
Executor* launchExecutor(
const ExecutorInfo& executorInfo,
- const TaskInfo& taskInfo);
+ const Option<TaskInfo>& taskInfo);
void destroyExecutor(const ExecutorID& executorId);
Executor* getExecutor(const ExecutorID& executorId);
Executor* getExecutor(const TaskID& taskId);
[8/8] mesos git commit: Minor fix to avoid creating an unnecessary
temporary when looping.
Posted by an...@apache.org.
Minor fix to avoid creating an unnecessary temporary when looping.
This change is done for making the looping style consistent
through the code.
Review: https://reviews.apache.org/r/51710/
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/aaa353ac
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/aaa353ac
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/aaa353ac
Branch: refs/heads/master
Commit: aaa353acc515c0435a859113c9ee236247b51169
Parents: 2b61afe
Author: Anand Mazumdar <an...@apache.org>
Authored: Mon Sep 12 14:21:46 2016 -0700
Committer: Anand Mazumdar <an...@apache.org>
Committed: Mon Sep 12 15:17:02 2016 -0700
----------------------------------------------------------------------
src/slave/slave.cpp | 3 +--
1 file changed, 1 insertion(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/aaa353ac/src/slave/slave.cpp
----------------------------------------------------------------------
diff --git a/src/slave/slave.cpp b/src/slave/slave.cpp
index c7ab7f2..2c7b5ad 100644
--- a/src/slave/slave.cpp
+++ b/src/slave/slave.cpp
@@ -1596,8 +1596,7 @@ void Slave::run(
// executors to this framework and remove it from that list.
// TODO(brenden): Consider using stout/cache.hpp instead of boost
// circular_buffer.
- for (auto it = completedFrameworks.begin(), end = completedFrameworks.end();
- it != end;
+ for (auto it = completedFrameworks.begin(); it != completedFrameworks.end();
++it) {
if ((*it)->id() == frameworkId) {
framework->completedExecutors = (*it)->completedExecutors;
[6/8] mesos git commit: Removed a no longer needed comment after the
`runTasks()` refactoring.
Posted by an...@apache.org.
Removed a no longer needed comment after the `runTasks()` refactoring.
Review: https://reviews.apache.org/r/51709/
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/2b61afe5
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/2b61afe5
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/2b61afe5
Branch: refs/heads/master
Commit: 2b61afe547209020a61387c7dba0b9a7fa3f8b33
Parents: 3a4c3f1
Author: Anand Mazumdar <an...@apache.org>
Authored: Mon Sep 12 14:21:34 2016 -0700
Committer: Anand Mazumdar <an...@apache.org>
Committed: Mon Sep 12 15:17:02 2016 -0700
----------------------------------------------------------------------
src/slave/slave.hpp | 3 ---
1 file changed, 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/2b61afe5/src/slave/slave.hpp
----------------------------------------------------------------------
diff --git a/src/slave/slave.hpp b/src/slave/slave.hpp
index 903a48d..e659c44 100644
--- a/src/slave/slave.hpp
+++ b/src/slave/slave.hpp
@@ -325,9 +325,6 @@ public:
// been updated for the given tasks and task groups. If the update is
// successful, we flush the given tasks to the executor by sending
// RunTaskMessages or `LAUNCH_GROUP` events.
- // TODO(jieyu): Consider renaming it to '__runTasks' once the slave
- // starts to support launching multiple tasks in one call (i.e.,
- // multi-tasks version of 'runTask').
void __run(
const process::Future<Nothing>& future,
const FrameworkID& frameworkId,
[5/8] mesos git commit: Added an equality operator for
`TaskGroupInfo`/`TaskInfo`.
Posted by an...@apache.org.
Added an equality operator for `TaskGroupInfo`/`TaskInfo`.
This is needed for storing \`TaskGroupInfo\` in a list and then
invoking \`std::find()\`/\`std::remove()\` on it. Also added
a basic test in `src/tests/common/type_utils_tests.cpp`.
Review: https://reviews.apache.org/r/51475/
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/94827559
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/94827559
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/94827559
Branch: refs/heads/master
Commit: 94827559f3727ef6dd0058068e7a4c976000a0f9
Parents: 00815ee
Author: Anand Mazumdar <an...@apache.org>
Authored: Mon Sep 12 14:21:09 2016 -0700
Committer: Anand Mazumdar <an...@apache.org>
Committed: Mon Sep 12 15:17:02 2016 -0700
----------------------------------------------------------------------
include/mesos/type_utils.hpp | 2 +
src/Makefile.am | 1 +
src/common/type_utils.cpp | 31 ++++++++++++
src/tests/common/type_utils_tests.cpp | 78 ++++++++++++++++++++++++++++++
4 files changed, 112 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/94827559/include/mesos/type_utils.hpp
----------------------------------------------------------------------
diff --git a/include/mesos/type_utils.hpp b/include/mesos/type_utils.hpp
index d10b663..4f4f46e 100644
--- a/include/mesos/type_utils.hpp
+++ b/include/mesos/type_utils.hpp
@@ -62,6 +62,8 @@ bool operator==(
bool operator==(const SlaveInfo& left, const SlaveInfo& right);
bool operator==(const Task& left, const Task& right);
+bool operator==(const TaskGroupInfo& left, const TaskGroupInfo& right);
+bool operator==(const TaskInfo& left, const TaskInfo& right);
bool operator==(const TaskStatus& left, const TaskStatus& right);
bool operator==(const URL& left, const URL& right);
bool operator==(const Volume& left, const Volume& right);
http://git-wip-us.apache.org/repos/asf/mesos/blob/94827559/src/Makefile.am
----------------------------------------------------------------------
diff --git a/src/Makefile.am b/src/Makefile.am
index fa71381..2dd7913 100644
--- a/src/Makefile.am
+++ b/src/Makefile.am
@@ -2141,6 +2141,7 @@ mesos_tests_SOURCES = \
tests/common/command_utils_tests.cpp \
tests/common/http_tests.cpp \
tests/common/recordio_tests.cpp \
+ tests/common/type_utils_tests.cpp \
tests/containerizer/appc_spec_tests.cpp \
tests/containerizer/composing_containerizer_tests.cpp \
tests/containerizer/docker_containerizer_tests.cpp \
http://git-wip-us.apache.org/repos/asf/mesos/blob/94827559/src/common/type_utils.cpp
----------------------------------------------------------------------
diff --git a/src/common/type_utils.cpp b/src/common/type_utils.cpp
index c7d5033..9cb6274 100644
--- a/src/common/type_utils.cpp
+++ b/src/common/type_utils.cpp
@@ -388,6 +388,37 @@ bool operator==(const Task& left, const Task& right)
}
+bool operator==(const TaskGroupInfo& left, const TaskGroupInfo& right)
+{
+ // Order of tasks in a task group is not important.
+ if (left.tasks().size() != right.tasks().size()) {
+ return false;
+ }
+
+ for (int i = 0; i < left.tasks().size(); i++) {
+ bool found = false;
+ for (int j = 0; j < right.tasks().size(); j++) {
+ if (left.tasks().Get(i) == right.tasks().Get(j)) {
+ found = true;
+ break;
+ }
+ }
+ if (!found) {
+ return false;
+ }
+ }
+
+ return true;
+}
+
+
+// TODO(anand): Consider doing a field by field comparison instead.
+bool operator==(const TaskInfo& left, const TaskInfo& right)
+{
+ return left.SerializeAsString() == right.SerializeAsString();
+}
+
+
// TODO(bmahler): Use SerializeToString here?
bool operator==(const TaskStatus& left, const TaskStatus& right)
{
http://git-wip-us.apache.org/repos/asf/mesos/blob/94827559/src/tests/common/type_utils_tests.cpp
----------------------------------------------------------------------
diff --git a/src/tests/common/type_utils_tests.cpp b/src/tests/common/type_utils_tests.cpp
new file mode 100644
index 0000000..a962b39
--- /dev/null
+++ b/src/tests/common/type_utils_tests.cpp
@@ -0,0 +1,78 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#include <gtest/gtest.h>
+
+#include <mesos/mesos.hpp>
+#include <mesos/resources.hpp>
+
+#include "tests/mesos.hpp"
+
+using namespace mesos;
+using namespace mesos::internal::tests;
+
+
+TEST(TypeUtilsTest, TaskGroupEquality)
+{
+ SlaveID slaveId;
+ slaveId.set_value("default-agent");
+
+ Resources resources = Resources::parse("cpus:0.1;mem:32;disk:32").get();
+
+ TaskInfo task1 =
+ createTask(slaveId, resources, "default-command1");
+
+ TaskInfo task2 =
+ createTask(slaveId, resources, "default-command2");
+
+ // Compare task groups with identical tasks.
+ {
+ TaskGroupInfo taskGroup1;
+ taskGroup1.add_tasks()->CopyFrom(task1);
+ taskGroup1.add_tasks()->CopyFrom(task2);
+
+ TaskGroupInfo taskGroup2;
+ taskGroup2.add_tasks()->CopyFrom(task1);
+ taskGroup2.add_tasks()->CopyFrom(task2);
+
+ EXPECT_EQ(taskGroup1, taskGroup2);
+ }
+
+ // Compare task groups with identical tasks but ordered differently.
+ {
+ TaskGroupInfo taskGroup1;
+ taskGroup1.add_tasks()->CopyFrom(task1);
+ taskGroup1.add_tasks()->CopyFrom(task2);
+
+ TaskGroupInfo taskGroup2;
+ taskGroup2.add_tasks()->CopyFrom(task2);
+ taskGroup2.add_tasks()->CopyFrom(task1);
+
+ EXPECT_EQ(taskGroup1, taskGroup2);
+ }
+
+ // Compare task groups with unequal tasks.
+ {
+ TaskGroupInfo taskGroup1;
+ taskGroup1.add_tasks()->CopyFrom(task1);
+ taskGroup1.add_tasks()->CopyFrom(task2);
+
+ TaskGroupInfo taskGroup2;
+ taskGroup2.add_tasks()->CopyFrom(task1);
+
+ EXPECT_FALSE(taskGroup1 == taskGroup2);
+ }
+}