[gentoo-commits] proj/R_overlay:master commit in: roverlay/overlay/

["André Erdmann" <dywi@mailerd.de>]

commit:     9646d809ea212ce68a04904a2d3f2d10d3fed718
Author:     André Erdmann <dywi <AT> mailerd <DOT> de>
AuthorDate: Fri Jun 22 18:04:31 2012 +0000
Commit:     André Erdmann <dywi <AT> mailerd <DOT> de>
CommitDate: Fri Jun 22 18:04:31 2012 +0000
URL:        http://git.overlays.gentoo.org/gitweb/?p=proj/R_overlay.git;a=commit;h=9646d809

fix wait() in overlay creation

* workers now run until the queue is empty after receiving stop_when_empty()
  instead of stopping after the next package (which is what close() does)

* added some comments in creator,worker

	modified:   overlay/creator.py
	modified:   overlay/worker.py

---
 roverlay/overlay/creator.py |   71 ++++++++++++++++++++++++++++++++++++++-----
 roverlay/overlay/worker.py  |   71 ++++++++++++++++++++++++++++++++++--------
 2 files changed, 120 insertions(+), 22 deletions(-)

diff --git a/roverlay/overlay/creator.py b/roverlay/overlay/creator.py
index 0936ca9..bbaaa8e 100644
--- a/roverlay/overlay/creator.py
+++ b/roverlay/overlay/creator.py
@@ -26,6 +26,7 @@ OVERLAY_WRITE_ALLOWED = False
 
 
 class OverlayCreator ( object ):
+	"""This is a 'R packages -> Overlay' interface."""
 
 	def __init__ ( self, logger=None ):
 
@@ -51,7 +52,13 @@ class OverlayCreator ( object ):
 	# --- end of __init__ (...) ---
 
 	def _timestamp ( self, description, start, stop=None ):
+		"""Logs a timestamp, used for testing.
 
+		arguments:
+		* description -- timestamp text
+		* start       -- when measuring for this timestamp has been started
+		* stop        -- stop time; defaults to now (time.time()) if unset
+		"""
 		_stop = time.time() if stop is None else stop
 		delta = _stop - start
 
@@ -87,6 +94,11 @@ class OverlayCreator ( object ):
 	# --- end of add_distdir (...) ---
 
 	def write_overlay ( self, incremental=False ):
+		"""Writes the overlay.
+
+		arguments:
+		* incremental -- (TODO)
+		"""
 		if self.can_write_overlay:
 			start = time.time()
 			if incremental:
@@ -101,10 +113,12 @@ class OverlayCreator ( object ):
 	# --- end of write_overlay (...) ---
 
 	def show_overlay ( self ):
+		"""Prints the overlay to the console. Does not create Manifest files."""
 		self.overlay.show()
 	# --- end of show_overlay (...) ---
 
 	def run ( self ):
+		"""Starts ebuild creation and waits until done."""
 		self._runlock.acquire()
 		self.start()
 		self.join()
@@ -112,25 +126,30 @@ class OverlayCreator ( object ):
 	# --- end of run (...) ---
 
 	def start ( self ):
+		"""Starts ebuild creation."""
 		self._runlock.acquire()
 		self.join()
 		self._make_workers()
 		self._runlock.release()
 	# --- end of start (...) ---
 
-	def wait ( self ):
-		self._close_workers()
+	def join ( self ):
+		"""Waits until ebuild creation is done."""
+		self._join_workers()
 	# --- end of wait (...) ---
 
-	join = wait
-
 	def close ( self, write=False ):
+		"""Closes this OverlayCreator."""
 		self._close_workers()
 		self._close_resolver()
 		if write: self.write_overlay()
 	# --- end of close (...) ---
 
 	def _close_resolver ( self ):
+		"""Tells the dependency resolver to close.
+		This is useful 'cause certain depres listener modules will write files
+		when told to exit.
+		"""
 		if self.depresolver is None: return
 
 		self._runlock.acquire()
@@ -143,7 +162,12 @@ class OverlayCreator ( object ):
 		self._runlock.release()
 	# --- end of _close_resolver (...) ---
 
-	def _close_workers ( self ):
+	def _waitfor_workers ( self, do_close ):
+		"""Waits until the workers are done.
+
+		arguments:
+		* do_close -- close (exit) workers if True, else wait until done.
+		"""
 		if self._workers is None: return
 
 		self._runlock.acquire()
@@ -151,7 +175,10 @@ class OverlayCreator ( object ):
 
 		if self.NUMTHREADS > 0: start = time.time()
 
-		for w in self._workers: w.enabled = False
+		if do_close:
+			for w in self._workers: w.enabled = False
+		else:
+			for w in self._workers: w.stop_when_empty()
 
 		while True in ( w.active() for w in self._workers ):
 			self._pkg_queue.put ( None )
@@ -159,13 +186,34 @@ class OverlayCreator ( object ):
 		del self._workers
 		self._workers = None
 
-		if self.NUMTHREADS > 0:
+		self._runlock.release()
+		if self.NUMTHREADS > 0: return start
+	# --- end of _waitfor_workers (...) ---
+
+	def _join_workers ( self ):
+		"""Waits until all workers are done."""
+		start = self._waitfor_workers ( False )
+		if start is not None:
 			self._timestamp ( "worker threads are done", start )
+	# --- end of _join_workers (...) ---
 
-		self._runlock.release()
+	def _close_workers ( self ):
+		"""Tells the workers to exit.
+		This is done by disabling them and inserting empty requests (None as
+		PackageInfo) to unblock them.
+		"""
+		start = self._waitfor_workers ( True )
+		if start is not None:
+			self._timestamp ( "worker threads are closed", start )
 	# --- end of _close_workers (...) ---
 
 	def _pkg_done ( self, package_info ):
+		"""This is an event method used by worker threads when they have
+		processed a package info.
+
+		arguments:
+		* package_info --
+		"""
 		# ... TODO
 		#  * increase the number of successful/failed packages,
 		#  * request an incremental write to save memory etc.
@@ -176,6 +224,12 @@ class OverlayCreator ( object ):
 	# --- end of _add_to_overlay (...) ---
 
 	def _get_worker ( self, start_now=False, use_threads=True ):
+		"""Creates and returns a worker.
+
+		arguments:
+		* start_now   -- if set and True: start the worker immediately
+		* use_threads -- if set and False: disable threads
+		"""
 		w = OverlayWorker (
 			self._pkg_queue, self.depresolver, self.logger, self._pkg_done,
 			use_threads=use_threads
@@ -185,6 +239,7 @@ class OverlayCreator ( object ):
 	# --- end of _get_worker (...) ---
 
 	def _make_workers ( self ):
+		"""Creates and starts workers."""
 		self._close_workers()
 
 		if self.NUMTHREADS > 0:

diff --git a/roverlay/overlay/worker.py b/roverlay/overlay/worker.py
index 4f03fe2..471cb94 100644
--- a/roverlay/overlay/worker.py
+++ b/roverlay/overlay/worker.py
@@ -1,34 +1,44 @@
+# R Overlay -- R package -> overlay interface, PackageInfo worker
+# Copyright 2006-2012 Gentoo Foundation
+# Distributed under the terms of the GNU General Public License v2
 
-import time
+#import time
 import threading
 
-try:
-	import queue
-except ImportError:
-	# python2
-	import Queue as queue
-
-
 from roverlay.depres.channels    import EbuildJobChannel
 from roverlay.ebuild.creation    import EbuildCreation
 
 class OverlayWorker ( object ):
-
-	def __init__ ( self, pkg_queue, depresolver, logger, pkg_done, use_threads ):
+	"""Overlay package queue worker."""
+
+	def __init__ ( self,
+		pkg_queue, depresolver, logger, pkg_done, use_threads
+	):
+		"""Initializes a worker.
+
+		arguments:
+		* pkg_queue   -- queue with PackageInfos
+		* depresolver -- dependency resolver to use
+		* logger      -- logger to use
+		* pkg_done    -- method to call when a PackageInfo has been
+		                 processed
+		* use_threads -- whether to run this worker as a thread or not
+		"""
 		self.logger      = logger
 		self.pkg_queue   = pkg_queue
 		self.pkg_done    = pkg_done
 		self.depresolver = depresolver
 
-
-		self.enabled     = True
-		self.running     = False
 		self._use_thread = use_threads
 		self._thread     = None
 
+		self.enabled     = True
+		self.running     = False
+		self.halting     = False
 	# --- end of __init__ (...) ---
 
 	def start ( self ):
+		"""Starts the worker."""
 		if self._thread is not None:
 			self._thread.join()
 			del self._thread
@@ -41,17 +51,34 @@ class OverlayWorker ( object ):
 			self._run_nothread()
 	# --- end of start (...) ---
 
+	def stop_when_empty ( self ):
+		"""Tells the worker thread to exit when the queue is empty."""
+		self.enabled = False
+		self.halting = True
+	# --- end of stop_when_empty (...) ---
+
 	def active ( self ):
+		"""Returns True if this worker is active (running or enabled)."""
 		return self.running or self.enabled
 	# --- end of active (...) ---
 
 	def _get_resolver_channel ( self, **channel_kw ):
+		"""Returns a resolver channel.
+
+		arguments:
+		* **channel_kw -- keywords for EbuildJobChannel.__init__
+		"""
 		return self.depresolver.register_channel (
 			EbuildJobChannel ( **channel_kw )
 		)
 	# --- end of _get_resolver_channel (...) ---
 
 	def _process ( self, package_info ):
+		"""Processes a PackageInfo taken from the queue.
+
+		arguments:
+		* package_info --
+		"""
 		job = EbuildCreation (
 			package_info,
 			depres_channel_spawner=self._get_resolver_channel
@@ -61,21 +88,37 @@ class OverlayWorker ( object ):
 	# --- end of _process (...) ---
 
 	def _run ( self ):
+		"""Runs the worker (thread mode)."""
 		self.running = True
-		while self.enabled:
+		self.halting = False
+		while self.enabled or (
+			self.halting and not self.pkg_queue.empty()
+		):
+			if not self.running:
+				# exit now
+				break
 			p = self.pkg_queue.get()
 
 			# drop empty requests that are used to unblock get()
 			if p is not None:
 				self._process ( p )
+			elif self.halting:
+				# receiving an empty request while halting means 'stop now',
+				self.enabled = False
+				self.halting = False
 
 			self.pkg_queue.task_done()
 		self.running = False
 	# --- end of run (...) ---
 
 	def _run_nothread ( self ):
+		"""Runs the worker (no-thread mode)."""
 		self.running = True
 		while self.enabled and not self.pkg_queue.empty():
+			if not self.running:
+				# exit now
+				break
+
 			p = self.pkg_queue.get()
 
 			# drop empty requests that are used to unblock get()