FetchIterator: add terminate method

Add a terminate method to FetchIterator so that it doesn't
delay termination of emirrordist via SIGINT. Otherwise it
it is possible for the __iter__ method to loop for a very
long time after SIGINT has been delivered. For example,
this could happen if there are many ebuilds with stale
cache and RESTRICT=mirror. This issue was discovered
during testing of changes to the MirrorDistTask.terminate
implementation.

2 files changed, 26 insertions, 1 deletions

diff --git a/pym/portage/_emirrordist/FetchIterator.py b/pym/portage/_emirrordist/FetchIterator.py
index 16a0b04c9..38419799d 100644
--- a/pym/portage/_emirrordist/FetchIterator.py
+++ b/pym/portage/_emirrordist/FetchIterator.py
@@ -1,6 +1,8 @@
 # Copyright 2013 Gentoo Foundation
 # Distributed under the terms of the GNU General Public License v2
 
+import threading
+
 from portage import os
 from portage.checksum import (_apply_hash_filter,
 	_filter_unaccelarated_hashes, _hash_filter)
@@ -13,6 +15,19 @@ class FetchIterator(object):
 	def __init__(self, config):
 		self._config = config
 		self._log_failure = config.log_failure
+		self._terminated = threading.Event()
+
+	def terminate(self):
+		"""
+		Schedules early termination of the __iter__ method, which is
+		useful because under some conditions it's possible for __iter__
+		to loop for a long time without yielding to the caller. For
+		example, it's useful when there are many ebuilds with stale
+		cache and RESTRICT=mirror.
+
+		This method is thread-safe (and safe for signal handlers).
+		"""
+		self._terminated.set()
 
 	def _iter_every_cp(self):
 		# List categories individually, in order to start yielding quicker,
@@ -37,6 +52,9 @@ class FetchIterator(object):
 
 		for cp in self._iter_every_cp():
 
+			if self._terminated.is_set():
+				return
+
 			for tree in portdb.porttrees:
 
 				# Reset state so the Manifest is pulled once
@@ -46,6 +64,9 @@ class FetchIterator(object):
 
 				for cpv in portdb.cp_list(cp, mytree=tree):
 
+					if self._terminated.is_set():
+						return
+
 					try:
 						restrict, = portdb.aux_get(cpv, ("RESTRICT",),
 							mytree=tree)
diff --git a/pym/portage/_emirrordist/MirrorDistTask.py b/pym/portage/_emirrordist/MirrorDistTask.py
index d6b3decc0..a34f2c061 100644
--- a/pym/portage/_emirrordist/MirrorDistTask.py
+++ b/pym/portage/_emirrordist/MirrorDistTask.py
@@ -32,9 +32,11 @@ class MirrorDistTask(CompositeTask):
 		self._config = config
 		self._term_rlock = threading.RLock()
 		self._term_callback_handle = None
+		self._fetch_iterator = None
 
 	def _start(self):
-		fetch = TaskScheduler(iter(FetchIterator(self._config)),
+		self._fetch_iterator = FetchIterator(self._config)
+		fetch = TaskScheduler(iter(self._fetch_iterator),
 			max_jobs=self._config.options.jobs,
 			max_load=self._config.options.load_average,
 			event_loop=self._config.event_loop)
@@ -226,6 +228,8 @@ class MirrorDistTask(CompositeTask):
 					self._term_callback)
 
 	def _term_callback(self):
+		if self._fetch_iterator is not None:
+			self._fetch_iterator.terminate()
 		self.cancel()
 		self.wait()