[gentoo-commits] proj/portage:master commit in: pym/portage/tests/ebuild/, pym/_emerge/

[gentoo-commits] proj/portage:master commit in: pym/portage/tests/ebuild/, pym/_emerge/

[Zac Medico]

commit:     07823ba56f63309da9547e02e96b043005932be0
Author:     Zac Medico <zmedico <AT> gentoo <DOT> org>
AuthorDate: Tue Feb 14 02:35:03 2012 +0000
Commit:     Zac Medico <zmedico <AT> gentoo <DOT> org>
CommitDate: Tue Feb 14 02:35:03 2012 +0000
URL:        http://git.overlays.gentoo.org/gitweb/?p=proj/portage.git;a=commit;h=07823ba5

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 5742cb2..c7b8f83 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 af1c3ff..2c84709 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 d57ccab..a1467b0 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 b4c98dc..07fea73 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 1db6807..6e416c3 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 b84f7bb..4b37026 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 3cd56d2..ebff430 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 83c0cbe..71ac80f 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 edfc058..0efab65 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)
 

[gentoo-commits] proj/portage:master commit in: pym/portage/tests/ebuild/, pym/_emerge/

[Zac Medico]

commit:     71a5a82313226f7be0d966d49392a53139a96f6b
Author:     Zac Medico <zmedico <AT> gentoo <DOT> org>
AuthorDate: Tue Apr 24 02:47:11 2018 +0000
Commit:     Zac Medico <zmedico <AT> gentoo <DOT> org>
CommitDate: Thu Apr 26 03:19:22 2018 +0000
URL:        https://gitweb.gentoo.org/proj/portage.git/commit/?id=71a5a823

AsynchronousTask: add async_wait() method (bug 653856)

Since the AsynchronousTask.wait() method is prone to event loop
recursion, deprecate it, and add an async_wait() method method to
replace it. Instead of using task.wait() in order to implicitly run
the event loop, now loop.run_until_complete(task.async_wait()) will
be used to explicitly run the event loop. This explicit approach will
make it more obvious when code will trigger event loop recursion
which would not be compatible with asyncio's default event loop.

Bug: https://bugs.gentoo.org/653856

 pym/_emerge/AsynchronousTask.py             | 23 +++++++++++++++++++++++
 pym/portage/tests/ebuild/test_ipc_daemon.py |  2 +-
 2 files changed, 24 insertions(+), 1 deletion(-)

diff --git a/pym/_emerge/AsynchronousTask.py b/pym/_emerge/AsynchronousTask.py
index e29324440..7d2e6253b 100644
--- a/pym/_emerge/AsynchronousTask.py
+++ b/pym/_emerge/AsynchronousTask.py
@@ -29,6 +29,26 @@ class AsynchronousTask(SlotObject):
 		self._start_hook()
 		self._start()
 
+	def async_wait(self):
+		"""
+		Wait for returncode asynchronously. Notification is available
+		via the add_done_callback method of the returned Future instance.
+
+		@returns: Future, result is self.returncode
+		"""
+		waiter = self.scheduler.create_future()
+		exit_listener = lambda self: waiter.set_result(self.returncode)
+		self.addExitListener(exit_listener)
+		waiter.add_done_callback(lambda waiter:
+			self.removeExitListener(exit_listener) if waiter.cancelled() else None)
+		if self.returncode is not None:
+			# If the returncode is not None, it means the exit event has already
+			# happened, so use _async_wait() to guarantee that the exit_listener
+			# is called. This does not do any harm because a given exit listener
+			# is never called more than once.
+			self._async_wait()
+		return waiter
+
 	def _start(self):
 		self.returncode = os.EX_OK
 		self.wait()
@@ -47,6 +67,9 @@ class AsynchronousTask(SlotObject):
 		return self.returncode
 
 	def wait(self):
+		"""
+		Deprecated. Use async_wait() instead.
+		"""
 		if self.returncode is None:
 			if not self._waiting:
 				self._waiting = True

diff --git a/pym/portage/tests/ebuild/test_ipc_daemon.py b/pym/portage/tests/ebuild/test_ipc_daemon.py
index bc18cdf64..e6da51a76 100644
--- a/pym/portage/tests/ebuild/test_ipc_daemon.py
+++ b/pym/portage/tests/ebuild/test_ipc_daemon.py
@@ -157,6 +157,6 @@ class IpcDaemonTestCase(TestCase):
 		try:
 			task_scheduler.start()
 			event_loop.run_until_complete(self._run_done)
-			task_scheduler.wait()
+			event_loop.run_until_complete(task_scheduler.async_wait())
 		finally:
 			timeout_handle.cancel()