pool.consumed_tasks: is now a queue to be thread safe, in preparation for multiple connected pools

Reduced waiting time in tests to make them complete faster

Author: Byron

lib/git/async/pool.py

@@ -1,7 +1,7 @@
 """Implementation of a thread-pool working with channels"""
 from thread import WorkerThread
 from task import InputChannelTask
-from Queue import Queue
+from Queue import Queue, Empty
 
 from graph import (
 		Graph, 
@@ -103,14 +103,14 @@ class ThreadPool(object):
 		used only from the main thread, hence you cannot consume their results 
 		from multiple threads unless you use a task for it."""
 	__slots__ = (	'_tasks',				# a graph of tasks
-					'_consumed_tasks',		# a list with tasks that are done or had an error
+					'_consumed_tasks',		# a queue with tasks that are done or had an error
 					'_workers',				# list of worker threads
 					'_queue', 				# master queue for tasks
 				)
 
 	def __init__(self, size=0):
 		self._tasks = Graph()
-		self._consumed_tasks = list()
+		self._consumed_tasks = Queue()		# make sure its threadsafe
 		self._workers = list()
 		self._queue = Queue()
 		self.set_size(size)
@@ -123,7 +123,7 @@ def _queue_feeder_visitor(self, task, count):
 		"""Walk the graph and find tasks that are done for later cleanup, and 
 		queue all others for processing by our worker threads ( if available )."""
 		if task.error() or task.is_done():
-			self._consumed_tasks.append(task)
+			self._consumed_tasks.put(task)
 			return True
 		# END stop processing
 
@@ -206,16 +206,21 @@ def _prepare_channel_read(self, task, count):
 	def _post_channel_read(self, task):
 		"""Called after we processed a read to cleanup"""
 		# check whether we consumed the task, and schedule it for deletion
+		# This could have happend after the read returned ( even though the pre-read 
+		# checks it as well )
 		if task.error() or task.is_done():
-			self._consumed_tasks.append(task)
+			self._consumed_tasks.put(task)
 		# END handle consumption
 
 		# delete consumed tasks to cleanup
-		for task in self._consumed_tasks:
-			self.del_task(task)
-		# END for each task to delete
-		
-		del(self._consumed_tasks[:])
+		try:
+			while True:
+				ct = self._consumed_tasks.get(False)
+				self.del_task(ct)
+			# END for each task to delete
+		except Empty:
+			pass
+		# END pop queue empty
 
 	def _del_task_if_orphaned(self, task):
 		"""Check the task, and delete it if it is orphaned"""
@@ -236,7 +241,9 @@ def set_size(self, size=0):
 		
 		:return: self
 		:param size: if 0, the pool will do all work itself in the calling thread, 
-			otherwise the work will be distributed among the given amount of threads"""
+			otherwise the work will be distributed among the given amount of threads
+		
+		:note: currently NOT threadsafe !"""
 		# either start new threads, or kill existing ones.
 		# If we end up with no threads, we process the remaining chunks on the queue
 		# ourselves

lib/git/async/thread.py

@@ -115,6 +115,11 @@ class WorkerThread(TerminatableThread):
 	"""
 	__slots__ = ('inq', 'outq')
 
+	
+	# define how often we should check for a shutdown request in case our 
+	# taskqueue is empty
+	shutdown_check_time_s = 0.5
+	
 	def __init__(self, inq = None):
 		super(WorkerThread, self).__init__()
 		self.inq = inq or Queue.Queue()

test/git/async/test_channel.py

@@ -22,14 +22,15 @@ def test_base(self):
 		wc.write(item2)
 
 		# read all - it blocks as its still open for writing
+		to = 0.2
 		st = time.time()
-		assert rc.read(timeout=1) == [item, item2]
-		assert time.time() - st >= 1.0
+		assert rc.read(timeout=to) == [item, item2]
+		assert time.time() - st >= to
 
 		# next read blocks. it waits a second
 		st = time.time()
-		assert len(rc.read(1, True, 1)) == 0
-		assert time.time() - st >= 1.0
+		assert len(rc.read(1, True, to)) == 0
+		assert time.time() - st >= to
 
 		# writing to a closed channel raises
 		assert not wc.closed
@@ -50,10 +51,10 @@ def test_base(self):
 		wc, rc = Channel(1)
 		wc.write(item)			# fine
 
-		# blocks for a second, its full
+		# blocks for a a moment, its full
 		st = time.time()
-		self.failUnlessRaises(EOFError, wc.write, item, True, 1)
-		assert time.time() - st >= 1.0
+		self.failUnlessRaises(EOFError, wc.write, item, True, to)
+		assert time.time() - st >= to
 
 		# get our only one
 		assert rc.read(1)[0] == item