Change Agent.abort() again. This time, it runs through its queue of AgentTasks, aborting them until it reaches one that ignores the abort (or exhausts the queue). With the previous logic, we might have an Agent with a GatherLogsTasks that should ignore the abort, but if the Agent got aborted before starting it would never run the task. I hope I've really got it right this time.
To help simplify things, I reorganized the AgentTask logic a bit, making AgentTask.poll() call AgentTask.start() itself so that the Agent wouldn't have to explicitly call AgentTask.start(). I also got rid of Agent.start(), which was unused.
Signed-off-by: Steve Howard <showard@google.com>
git-svn-id: http://test.kernel.org/svn/autotest/trunk@3089 592f7852-d20e-0410-864c-8624ca9c26a4
diff --git a/scheduler/monitor_db.py b/scheduler/monitor_db.py
index ce1e431..6e7c24b 100755
--- a/scheduler/monitor_db.py
+++ b/scheduler/monitor_db.py
@@ -1189,7 +1189,7 @@
def tick(self):
while not self.is_done():
- if self.active_task and not self.active_task.is_done():
+ if self.active_task:
self.active_task.poll()
if not self.active_task.is_done():
return
@@ -1198,16 +1198,14 @@
def _next_task(self):
logging.info("agent picking task")
- if self.active_task:
+ if self.active_task is not None:
assert self.active_task.is_done()
if not self.active_task.success:
self.on_task_failure()
+ self.active_task = None
- self.active_task = None
if not self.is_done():
self.active_task = self.queue.get_nowait()
- if self.active_task:
- self.active_task.start()
def on_task_failure(self):
@@ -1226,21 +1224,17 @@
return self.active_task is None and self.queue.empty()
- def start(self):
- assert self.dispatcher
- self._next_task()
-
-
def abort(self):
- if self.active_task:
+ # abort tasks until the queue is empty or a task ignores the abort
+ while not self.is_done():
+ if not self.active_task:
+ self._next_task()
self.active_task.abort()
- if not self.active_task.aborted: # tasks can choose to ignore aborts
+ if not self.active_task.aborted:
+ # tasks can choose to ignore aborts
return
self.active_task = None
- self._clear_queue()
-
-
class AgentTask(object):
def __init__(self, cmd, working_directory=None, failure_tasks=[],
@@ -1270,17 +1264,17 @@
def poll(self):
+ if not self.started:
+ self.start()
+ self.tick()
+
+
+ def tick(self):
if self.monitor:
- self.tick(self.monitor.exit_code())
- else:
- self.finished(False)
-
-
- def tick(self, exit_code):
- if exit_code is None:
- return
- if exit_code == 0:
- success = True
+ exit_code = self.monitor.exit_code()
+ if exit_code is None:
+ return
+ success = (exit_code == 0)
else:
success = False
@@ -1292,6 +1286,8 @@
def finished(self, success):
+ if self.done:
+ return
self.done = True
self.success = success
self.epilog()
@@ -1577,6 +1573,9 @@
def _finish_task(self):
+ if not self.monitor:
+ return
+
self._write_job_finished()
# both of these conditionals can be true, iff the process ran, wrote a
@@ -1886,11 +1885,11 @@
results_dir]
- def poll(self):
- # override poll to keep trying to start until the parse count goes down
+ def tick(self):
+ # override tick to keep trying to start until the parse count goes down
# and we can, at which point we revert to default behavior
if self._parse_started:
- super(FinalReparseTask, self).poll()
+ super(FinalReparseTask, self).tick()
else:
self._try_starting_parse()
diff --git a/scheduler/monitor_db_unittest.py b/scheduler/monitor_db_unittest.py
index 86602ed..342ab7d 100644
--- a/scheduler/monitor_db_unittest.py
+++ b/scheduler/monitor_db_unittest.py
@@ -1240,15 +1240,14 @@
task1 = self._create_mock_task('task1')
task2 = self._create_mock_task('task2')
task3 = self._create_mock_task('task3')
-
- task1.start.expect_call()
+ task1.poll.expect_call()
task1.is_done.expect_call().and_return(False)
task1.poll.expect_call()
task1.is_done.expect_call().and_return(True)
task1.is_done.expect_call().and_return(True)
task1.success = True
- task2.start.expect_call()
+ task2.poll.expect_call()
task2.is_done.expect_call().and_return(True)
task2.is_done.expect_call().and_return(True)
task2.success = False
@@ -1257,7 +1256,6 @@
self._dispatcher.add_agent.expect_call(IsAgentWithTask(task3))
agent = self._create_agent([task1, task2])
- agent.start()
self._finish_agent(agent)
self.god.check_playback()
@@ -1265,27 +1263,27 @@
def _test_agent_abort_helper(self, ignore_abort=False):
task1 = self._create_mock_task('task1')
task2 = self._create_mock_task('task2')
- task1.start.expect_call()
- task1.is_done.expect_call().and_return(False)
task1.poll.expect_call()
task1.is_done.expect_call().and_return(False)
task1.abort.expect_call()
if ignore_abort:
task1.aborted = False # task ignores abort; execution continues
+ task1.poll.expect_call()
task1.is_done.expect_call().and_return(True)
task1.is_done.expect_call().and_return(True)
task1.success = True
- task2.start.expect_call()
+ task2.poll.expect_call()
task2.is_done.expect_call().and_return(True)
task2.is_done.expect_call().and_return(True)
task2.success = True
else:
- task1.aborted = True # execution halts, no further expectations
+ task1.aborted = True
+ task2.abort.expect_call()
+ task2.aborted = True
agent = self._create_agent([task1, task2])
- agent.start()
agent.tick()
agent.abort()
self._finish_agent(agent)
@@ -1297,15 +1295,29 @@
self._test_agent_abort_helper(True)
- def test_agent_abort_before_started(self):
+ def _test_agent_abort_before_started_helper(self, ignore_abort=False):
task = self._create_mock_task('task')
+ task.abort.expect_call()
+ if ignore_abort:
+ task.aborted = False
+ task.poll.expect_call()
+ task.is_done.expect_call().and_return(True)
+ task.is_done.expect_call().and_return(True)
+ task.success = True
+ else:
+ task.aborted = True
+
agent = self._create_agent([task])
agent.abort()
- agent.start()
self._finish_agent(agent)
self.god.check_playback()
+ def test_agent_abort_before_started(self):
+ self._test_agent_abort_before_started_helper()
+ self._test_agent_abort_before_started_helper(True)
+
+
class AgentTasksTest(unittest.TestCase):
TEMP_DIR = '/abspath/tempdir'
RESULTS_DIR = '/results/dir'
@@ -1369,7 +1381,6 @@
"""
if not getattr(task, 'agent', None):
task.agent = object()
- task.start()
count = 0
while not task.is_done():
count += 1
@@ -1672,7 +1683,7 @@
def test_gather_logs_task_successful_autoserv(self):
- """When Autoserv exits successful, no collect_crashinfo stage runs."""
+ # When Autoserv exits successful, no collect_crashinfo stage runs
self._setup_gather_logs_expects(autoserv_success=True)
self.job.reboot_after = models.RebootAfter.NEVER
self.host.set_status.expect_call('Ready')