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

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

commit:     aa1c6dade666ed4e0ffb85fda3c8f0fcb9569d4f
Author:     André Erdmann <dywi <AT> mailerd <DOT> de>
AuthorDate: Thu Jun 21 16:51:32 2012 +0000
Commit:     André Erdmann <dywi <AT> mailerd <DOT> de>
CommitDate: Thu Jun 21 16:51:32 2012 +0000
URL:        http://git.overlays.gentoo.org/gitweb/?p=proj/R_overlay.git;a=commit;h=aa1c6dad

added OverlayCreator

* this is a threaded R packages -> overlay interface

	new file:   roverlay/overlay/creator.py
	new file:   roverlay/overlay/worker.py
	new file:   roverlay/recipe/__init__.py
	new file:   roverlay/recipe/easyresolver.py

---
 roverlay/overlay/creator.py     |  205 +++++++++++++++++++++++++++++++++++++++
 roverlay/overlay/worker.py      |   87 +++++++++++++++++
 roverlay/recipe/__init__.py     |    2 +
 roverlay/recipe/easyresolver.py |   25 +++++
 4 files changed, 319 insertions(+), 0 deletions(-)

diff --git a/roverlay/overlay/creator.py b/roverlay/overlay/creator.py
new file mode 100644
index 0000000..0936ca9
--- /dev/null
+++ b/roverlay/overlay/creator.py
@@ -0,0 +1,205 @@
+# R Overlay -- R package -> overlay interface
+# Copyright 2006-2012 Gentoo Foundation
+# Distributed under the terms of the GNU General Public License v2
+
+import time
+import logging
+import threading
+
+try:
+	import queue
+except ImportError:
+	# python2
+	import Queue as queue
+
+
+from roverlay                    import config
+from roverlay.overlay            import Overlay
+from roverlay.overlay.worker     import OverlayWorker
+from roverlay.packageinfo        import PackageInfo
+
+from roverlay.recipe             import easyresolver
+
+LOGGER = logging.getLogger ( 'OverlayCreator' )
+
+OVERLAY_WRITE_ALLOWED = False
+
+
+class OverlayCreator ( object ):
+
+	def __init__ ( self, logger=None ):
+
+		if logger is None:
+			self.logger = LOGGER
+		else:
+			self.logger = logger.getChild ( 'OverlayCreator' )
+
+		# init overlay using config values
+		self.overlay     = Overlay ( logger=self.logger )
+
+		self.depresolver = easyresolver.setup()
+
+		self.NUMTHREADS  = config.get ( 'EBUILD.jobcount', 0 )
+
+		# --
+		self._pkg_queue  = queue.Queue()
+		self._workers    = None
+		self._runlock    = threading.RLock()
+
+		self.can_write_overlay = OVERLAY_WRITE_ALLOWED
+
+	# --- end of __init__ (...) ---
+
+	def _timestamp ( self, description, start, stop=None ):
+
+		_stop = time.time() if stop is None else stop
+		delta = _stop - start
+
+		self.logger.debug (
+			"timestamp: %s (after %f seconds)" % ( description, delta )
+		)
+		return _stop
+	# --- end of _timestamp (...) ---
+
+	def add ( self, *to_add ):
+		"""Add a directory/package files/distdir to the package queue."""
+		raise Exception ( "method stub" )
+	# --- end of add (...) ---
+
+	def add_package_file ( self, package_file ):
+		"""Adds a single R package."""
+		self._pkg_queue.put ( PackageInfo ( filepath=package_file ) )
+	# --- end of add_package (...) ---
+
+	def add_package_files ( self, *package_files ):
+		"""Adds multiple R packages."""
+		for p in package_files: self.add_package_file ( p )
+	# --- end of add_packages (...) ---
+
+	def add_directory ( self, directory ):
+		"""Adds all packages from a directory to the package queue."""
+		raise Exception ( "method stub" )
+	# --- end of add_directory (...) ---
+
+	def add_distdir ( self, repo_name ):
+		"""Adds a distfiles directory to the package queue."""
+		self.add_directory ( config.get_or_fail ( [ 'DISTFILES', repo_name ] ) )
+	# --- end of add_distdir (...) ---
+
+	def write_overlay ( self, incremental=False ):
+		if self.can_write_overlay:
+			start = time.time()
+			if incremental:
+				# this will fail 'cause not implemented
+				self.overlay.write_incremental()
+			else:
+				self.overlay.write()
+
+			self._timestamp ( "overlay written", start )
+		else:
+			self.logger.warning ( "Not allowed to write overlay!" )
+	# --- end of write_overlay (...) ---
+
+	def show_overlay ( self ):
+		self.overlay.show()
+	# --- end of show_overlay (...) ---
+
+	def run ( self ):
+		self._runlock.acquire()
+		self.start()
+		self.join()
+		self._runlock.release()
+	# --- end of run (...) ---
+
+	def start ( self ):
+		self._runlock.acquire()
+		self.join()
+		self._make_workers()
+		self._runlock.release()
+	# --- end of start (...) ---
+
+	def wait ( self ):
+		self._close_workers()
+	# --- end of wait (...) ---
+
+	join = wait
+
+	def close ( self, write=False ):
+		self._close_workers()
+		self._close_resolver()
+		if write: self.write_overlay()
+	# --- end of close (...) ---
+
+	def _close_resolver ( self ):
+		if self.depresolver is None: return
+
+		self._runlock.acquire()
+		if self.depresolver is None: return
+
+		self.depresolver.close()
+		del self.depresolver
+		self.depresolver = None
+
+		self._runlock.release()
+	# --- end of _close_resolver (...) ---
+
+	def _close_workers ( self ):
+		if self._workers is None: return
+
+		self._runlock.acquire()
+		if self._workers is None: return
+
+		if self.NUMTHREADS > 0: start = time.time()
+
+		for w in self._workers: w.enabled = False
+
+		while True in ( w.active() for w in self._workers ):
+			self._pkg_queue.put ( None )
+
+		del self._workers
+		self._workers = None
+
+		if self.NUMTHREADS > 0:
+			self._timestamp ( "worker threads are done", start )
+
+		self._runlock.release()
+	# --- end of _close_workers (...) ---
+
+	def _pkg_done ( self, package_info ):
+		# ... TODO
+		#  * increase the number of successful/failed packages,
+		#  * request an incremental write to save memory etc.
+
+		# if <>:
+		if package_info ['ebuild'] is not None:
+			self.overlay.add ( package_info )
+	# --- end of _add_to_overlay (...) ---
+
+	def _get_worker ( self, start_now=False, use_threads=True ):
+		w = OverlayWorker (
+			self._pkg_queue, self.depresolver, self.logger, self._pkg_done,
+			use_threads=use_threads
+		)
+		if start_now: w.start()
+		return w
+	# --- end of _get_worker (...) ---
+
+	def _make_workers ( self ):
+		self._close_workers()
+
+		if self.NUMTHREADS > 0:
+			start = time.time()
+			self.logger.warning (
+				"Running in concurrent mode with %i threads." % self.NUMTHREADS
+			)
+			self._workers = frozenset (
+				self._get_worker ( start_now=True ) \
+					for n in range ( self.NUMTHREADS )
+			)
+			self._timestamp ( "worker threads initialized", start )
+		else:
+			self._workers = (
+				self._get_worker ( start_now=True, use_threads=False ),
+			)
+
+	# --- end of _make_workers (...) ---

diff --git a/roverlay/overlay/worker.py b/roverlay/overlay/worker.py
new file mode 100644
index 0000000..4f03fe2
--- /dev/null
+++ b/roverlay/overlay/worker.py
@@ -0,0 +1,87 @@
+
+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 ):
+		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
+
+	# --- end of __init__ (...) ---
+
+	def start ( self ):
+		if self._thread is not None:
+			self._thread.join()
+			del self._thread
+
+		self.enabled = True
+		if self._use_thread:
+			self._thread = threading.Thread ( target=self._run )
+			self._thread.start()
+		else:
+			self._run_nothread()
+	# --- end of start (...) ---
+
+	def active ( self ):
+		return self.running or self.enabled
+	# --- end of active (...) ---
+
+	def _get_resolver_channel ( self, **channel_kw ):
+		return self.depresolver.register_channel (
+			EbuildJobChannel ( **channel_kw )
+		)
+	# --- end of _get_resolver_channel (...) ---
+
+	def _process ( self, package_info ):
+		job = EbuildCreation (
+			package_info,
+			depres_channel_spawner=self._get_resolver_channel
+		)
+		job.run()
+		self.pkg_done ( package_info )
+	# --- end of _process (...) ---
+
+	def _run ( self ):
+		self.running = True
+		while self.enabled:
+			p = self.pkg_queue.get()
+
+			# drop empty requests that are used to unblock get()
+			if p is not None:
+				self._process ( p )
+
+			self.pkg_queue.task_done()
+		self.running = False
+	# --- end of run (...) ---
+
+	def _run_nothread ( self ):
+		self.running = True
+		while self.enabled and not self.pkg_queue.empty():
+			p = self.pkg_queue.get()
+
+			# drop empty requests that are used to unblock get()
+			if p is not None:
+				self._process ( p )
+
+			self.pkg_queue.task_done()
+		self.running = False
+	# --- end of _run_nothread (...) ---

diff --git a/roverlay/recipe/__init__.py b/roverlay/recipe/__init__.py
new file mode 100644
index 0000000..139597f
--- /dev/null
+++ b/roverlay/recipe/__init__.py
@@ -0,0 +1,2 @@
+
+

diff --git a/roverlay/recipe/easyresolver.py b/roverlay/recipe/easyresolver.py
new file mode 100644
index 0000000..c5e737a
--- /dev/null
+++ b/roverlay/recipe/easyresolver.py
@@ -0,0 +1,25 @@
+
+from roverlay                      import config
+from roverlay.depres               import listeners
+from roverlay.depres.depresolver   import DependencyResolver
+from roverlay.depres.simpledeprule import SimpleDependencyRulePool
+
+
+def setup():
+	res = DependencyResolver()
+	# log everything
+	res.set_logmask ( -1 )
+
+	srule_pool = SimpleDependencyRulePool ( 'default pool', priority=45 )
+
+	srule_files = config.get_or_fail ( 'DEPRES.simple_rules.files' )
+
+	if isinstance ( srule_files, str ):
+		srule_pool.load_rule_file ( srule_files )
+	else:
+		for f in srule_files:
+			srule_pool.load_rule_file ( f )
+
+	res.add_rulepool ( srule_pool )
+	return res
+# --- end of setup (...) ---