From 07823ba56f63309da9547e02e96b043005932be0 Mon Sep 17 00:00:00 2001 From: Zac Medico Date: Mon, 13 Feb 2012 18:35:03 -0800 Subject: AsynchronousTask: don't wait for exit status Synchronous waiting for status is not supported, since it would be vulnerable to hitting the recursion limit when a large number of tasks need to be terminated simultaneously, like in bug #402335. --- pym/_emerge/AbstractEbuildProcess.py | 3 +-- pym/_emerge/AbstractPollTask.py | 1 + pym/_emerge/AsynchronousTask.py | 9 ++++++++- pym/_emerge/MetadataRegen.py | 4 ++-- pym/_emerge/PollScheduler.py | 3 +++ pym/_emerge/Scheduler.py | 4 ++-- pym/_emerge/SequentialTaskQueue.py | 15 +++++++++++---- pym/_emerge/TaskScheduler.py | 1 + pym/portage/tests/ebuild/test_ipc_daemon.py | 10 ++++++---- 9 files changed, 35 insertions(+), 15 deletions(-) diff --git a/pym/_emerge/AbstractEbuildProcess.py b/pym/_emerge/AbstractEbuildProcess.py index 5742cb2a7..c7b8f83ca 100644 --- a/pym/_emerge/AbstractEbuildProcess.py +++ b/pym/_emerge/AbstractEbuildProcess.py @@ -167,8 +167,7 @@ class AbstractEbuildProcess(SpawnProcess): # of time, kill it (solves bug #278895). We try to avoid # this when possible since it makes sandbox complain about # being killed by a signal. - self.cancelled = True - self._cancel() + self.cancel() self._exit_timeout_id = \ self.scheduler.timeout_add(self._cancel_timeout, self._cancel_timeout_cb) diff --git a/pym/_emerge/AbstractPollTask.py b/pym/_emerge/AbstractPollTask.py index af1c3ffe8..2c8470925 100644 --- a/pym/_emerge/AbstractPollTask.py +++ b/pym/_emerge/AbstractPollTask.py @@ -123,6 +123,7 @@ class AbstractPollTask(AsynchronousTask): self._log_poll_exception(event) self._unregister() self.cancel() + self.wait() elif event & self.scheduler.IO_HUP: self._unregister() self.wait() diff --git a/pym/_emerge/AsynchronousTask.py b/pym/_emerge/AsynchronousTask.py index d57ccab2b..a1467b0b5 100644 --- a/pym/_emerge/AsynchronousTask.py +++ b/pym/_emerge/AsynchronousTask.py @@ -56,10 +56,17 @@ class AsynchronousTask(SlotObject): return self.returncode def cancel(self): + """ + Cancel the task, but do not wait for exit status. If asynchronous exit + notification is desired, then use addExitListener to add a listener + before calling this method. + NOTE: Synchronous waiting for status is not supported, since it would + be vulnerable to hitting the recursion limit when a large number of + tasks need to be terminated simultaneously, like in bug #402335. + """ if not self.cancelled: self.cancelled = True self._cancel() - self.wait() def _cancel(self): """ diff --git a/pym/_emerge/MetadataRegen.py b/pym/_emerge/MetadataRegen.py index b4c98dc7e..07fea73c4 100644 --- a/pym/_emerge/MetadataRegen.py +++ b/pym/_emerge/MetadataRegen.py @@ -37,8 +37,8 @@ class MetadataRegen(PollScheduler): self._remaining_tasks = True def _terminate_tasks(self): - while self._running_tasks: - self._running_tasks.pop().cancel() + for task in list(self._running_tasks): + task.cancel() def _iter_every_cp(self): portage.writemsg_stdout("Listing available packages...\n") diff --git a/pym/_emerge/PollScheduler.py b/pym/_emerge/PollScheduler.py index 1db68072c..6e416c300 100644 --- a/pym/_emerge/PollScheduler.py +++ b/pym/_emerge/PollScheduler.py @@ -86,6 +86,9 @@ class PollScheduler(object): implementation is running, in order to avoid potential interference. All tasks should be cleaned up at the earliest opportunity, but not necessarily before this method returns. + Typically, this method will send kill signals and return without + waiting for exit status. This allows basic cleanup to occur, such as + flushing of buffered output to logs. """ raise NotImplementedError() diff --git a/pym/_emerge/Scheduler.py b/pym/_emerge/Scheduler.py index b84f7bb66..4b3702667 100644 --- a/pym/_emerge/Scheduler.py +++ b/pym/_emerge/Scheduler.py @@ -313,8 +313,7 @@ class Scheduler(PollScheduler): def _terminate_tasks(self): self._status_display.quiet = True - while self._running_tasks: - task_id, task = self._running_tasks.popitem() + for task in list(self._running_tasks.values()): task.cancel() for q in self._task_queues.values(): q.clear() @@ -904,6 +903,7 @@ class Scheduler(PollScheduler): finally: if current_task is not None and current_task.isAlive(): current_task.cancel() + current_task.wait() clean_phase = EbuildPhase(background=False, phase='clean', scheduler=sched_iface, settings=settings) clean_phase.start() diff --git a/pym/_emerge/SequentialTaskQueue.py b/pym/_emerge/SequentialTaskQueue.py index 3cd56d2d6..ebff430e3 100644 --- a/pym/_emerge/SequentialTaskQueue.py +++ b/pym/_emerge/SequentialTaskQueue.py @@ -55,13 +55,20 @@ class SequentialTaskQueue(SlotObject): self.schedule() def clear(self): + """ + Clear the task queue and asynchronously terminate any running tasks. + """ self._task_queue.clear() - running_tasks = self.running_tasks - while running_tasks: - task = running_tasks.pop() - task.removeExitListener(self._task_exit) + for task in list(self.running_tasks): task.cancel() + def wait(self): + """ + Synchronously wait for all running tasks to exit. + """ + while self.running_tasks: + next(iter(self.running_tasks)).wait() + def __bool__(self): return bool(self._task_queue or self.running_tasks) diff --git a/pym/_emerge/TaskScheduler.py b/pym/_emerge/TaskScheduler.py index 83c0cbe96..71ac80f14 100644 --- a/pym/_emerge/TaskScheduler.py +++ b/pym/_emerge/TaskScheduler.py @@ -18,6 +18,7 @@ class TaskScheduler(object): self.sched_iface = self._scheduler.sched_iface self.run = self._scheduler.run self.clear = self._scheduler.clear + self.wait = self._queue.wait self._scheduler.add(self._queue) def add(self, task): diff --git a/pym/portage/tests/ebuild/test_ipc_daemon.py b/pym/portage/tests/ebuild/test_ipc_daemon.py index edfc058d7..0efab6584 100644 --- a/pym/portage/tests/ebuild/test_ipc_daemon.py +++ b/pym/portage/tests/ebuild/test_ipc_daemon.py @@ -71,8 +71,8 @@ class IpcDaemonTestCase(TestCase): self.received_command = False def exit_command_callback(): self.received_command = True - proc.cancel() - daemon.cancel() + task_scheduler.clear() + task_scheduler.wait() exit_command.reply_hook = exit_command_callback start_time = time.time() @@ -80,6 +80,7 @@ class IpcDaemonTestCase(TestCase): task_scheduler.add(proc) task_scheduler.run(timeout=self._SCHEDULE_TIMEOUT) task_scheduler.clear() + task_scheduler.wait() hardlock_cleanup(env['PORTAGE_BUILDDIR'], remove_all_locks=True) @@ -108,8 +109,8 @@ class IpcDaemonTestCase(TestCase): self.received_command = False def exit_command_callback(): self.received_command = True - proc.cancel() - daemon.cancel() + task_scheduler.clear() + task_scheduler.wait() exit_command.reply_hook = exit_command_callback start_time = time.time() @@ -117,6 +118,7 @@ class IpcDaemonTestCase(TestCase): task_scheduler.add(proc) task_scheduler.run(timeout=short_timeout_ms) task_scheduler.clear() + task_scheduler.wait() hardlock_cleanup(env['PORTAGE_BUILDDIR'], remove_all_locks=True) -- cgit v1.2.3-1-g7c22