Merge branch 'async'

7 files changed, 1688 insertions, 0 deletions

diff --git a/lib/git/async/__init__.py b/lib/git/async/__init__.py
new file mode 100644
index 00000000..e212f1b2
--- /dev/null
+++ b/lib/git/async/__init__.py
@@ -0,0 +1,30 @@
+"""Initialize the multi-processing package"""
+
+#{ Initialization
+def _init_atexit():
+	"""Setup an at-exit job to be sure our workers are shutdown correctly before
+	the interpreter quits"""
+	import atexit
+	import thread
+	atexit.register(thread.do_terminate_threads)
+	
+def _init_signals():
+	"""Assure we shutdown our threads correctly when being interrupted"""
+	import signal
+	import thread
+	
+	prev_handler = signal.getsignal(signal.SIGINT)
+	def thread_interrupt_handler(signum, frame):
+		thread.do_terminate_threads()
+		if callable(prev_handler):
+			prev_handler(signum, frame)
+			raise KeyboardInterrupt()
+		# END call previous handler
+	# END signal handler
+	signal.signal(signal.SIGINT, thread_interrupt_handler)
+
+
+#} END init
+
+_init_atexit()
+_init_signals()
diff --git a/lib/git/async/channel.py b/lib/git/async/channel.py
new file mode 100644
index 00000000..a29ff17c
--- /dev/null
+++ b/lib/git/async/channel.py
@@ -0,0 +1,338 @@
+"""Contains a queue based channel implementation"""
+from Queue import (
+	Empty, 
+	Full
+	)
+
+from util import (
+		AsyncQueue, 
+		SyncQueue,
+		ReadOnly
+		)
+
+from time import time
+import threading
+import sys
+
+__all__ = ('Channel', 'SerialChannel', 'Writer', 'ChannelWriter', 'CallbackChannelWriter',
+			'Reader', 'ChannelReader', 'CallbackChannelReader', 'mkchannel', 'ReadOnly', 
+			'IteratorReader')
+
+#{ Classes 
+class Channel(object):
+	"""A channel is similar to a file like object. It has a write end as well as one or
+	more read ends. If Data is in the channel, it can be read, if not the read operation
+	will block until data becomes available.
+	If the channel is closed, any read operation will result in an exception
+	
+	This base class is not instantiated directly, but instead serves as constructor
+	for Rwriter pairs.
+	
+	Create a new channel """
+	__slots__ = 'queue'
+	
+	# The queue to use to store the actual data
+	QueueCls = AsyncQueue
+	
+	def __init__(self):
+		"""initialize this instance with a queue holding the channel contents""" 
+		self.queue = self.QueueCls()
+
+
+class SerialChannel(Channel):
+	"""A slightly faster version of a Channel, which sacrificed thead-safety for performance"""
+	QueueCls = SyncQueue
+
+
+class Writer(object):
+	"""A writer is an object providing write access to a possibly blocking reading device"""
+	__slots__ = tuple()
+	
+	#{ Interface
+	
+	def __init__(self, device):
+		"""Initialize the instance with the device to write to"""
+	
+	def write(self, item, block=True, timeout=None):
+		"""Write the given item into the device
+		:param block: True if the device may block until space for the item is available
+		:param timeout: The time in seconds to wait for the device to become ready 
+		in blocking mode"""
+		raise NotImplementedError()
+		
+	def size(self):
+		""":return: number of items already in the device, they could be read with a reader"""
+		raise NotImplementedError()
+		
+	def close(self):
+		"""Close the channel. Multiple close calls on a closed channel are no 
+		an error"""
+		raise NotImplementedError()
+		
+	def closed(self):
+		""":return: True if the channel was closed"""
+		raise NotImplementedError()
+		
+	#} END interface
+	
+
+class ChannelWriter(Writer):
+	"""The write end of a channel, a file-like interface for a channel"""
+	__slots__ = ('channel', '_put')
+	
+	def __init__(self, channel):
+		"""Initialize the writer to use the given channel"""
+		self.channel = channel
+		self._put = self.channel.queue.put
+	
+	#{ Interface
+	def write(self, item, block=False, timeout=None):
+		return self._put(item, block, timeout)
+		
+	def size(self):
+		return self.channel.queue.qsize()
+		
+	def close(self):
+		"""Close the channel. Multiple close calls on a closed channel are no 
+		an error"""
+		self.channel.queue.set_writable(False)
+		
+	def closed(self):
+		""":return: True if the channel was closed"""
+		return not self.channel.queue.writable()
+	#} END interface 
+	
+
+class CallbackWriterMixin(object):
+	"""The write end of a channel which allows you to setup a callback to be 
+	called after an item was written to the channel"""
+	# slots don't work with mixin's :(
+	# __slots__ = ('_pre_cb')
+	
+	def __init__(self, *args):
+		super(CallbackWriterMixin, self).__init__(*args)
+		self._pre_cb = None
+	
+	def set_pre_cb(self, fun = lambda item: item):
+		"""Install a callback to be called before the given item is written.
+		It returns a possibly altered item which will be written to the channel
+		instead, making it useful for pre-write item conversions.
+		Providing None uninstalls the current method.
+		:return: the previously installed function or None
+		:note: Must be thread-safe if the channel is used in multiple threads"""
+		prev = self._pre_cb
+		self._pre_cb = fun
+		return prev
+	
+	def write(self, item, block=True, timeout=None):
+		if self._pre_cb:
+			item = self._pre_cb(item)
+		super(CallbackWriterMixin, self).write(item, block, timeout)
+	
+	
+class CallbackChannelWriter(CallbackWriterMixin, ChannelWriter):
+	"""Implements a channel writer with callback functionality"""
+	pass
+	
+
+class Reader(object):
+	"""Allows reading from a device"""
+	__slots__ = tuple()
+	
+	#{ Interface
+	def __init__(self, device):
+		"""Initialize the instance with the device to read from"""
+		
+	def read(self, count=0, block=True, timeout=None):
+		"""read a list of items read from the device. The list, as a sequence
+		of items, is similar to the string of characters returned when reading from 
+		file like objects.
+		:param count: given amount of items to read. If < 1, all items will be read
+		:param block: if True, the call will block until an item is available
+		:param timeout: if positive and block is True, it will block only for the 
+			given amount of seconds, returning the items it received so far.
+			The timeout is applied to each read item, not for the whole operation.
+		:return: single item in a list if count is 1, or a list of count items. 
+			If the device was empty and count was 1, an empty list will be returned.
+			If count was greater 1, a list with less than count items will be 
+			returned.
+			If count was < 1, a list with all items that could be read will be 
+			returned."""
+		raise NotImplementedError()
+		
+
+class ChannelReader(Reader):
+	"""Allows reading from a channel. The reader is thread-safe if the channel is as well"""
+	__slots__ = 'channel'
+	
+	def __init__(self, channel):
+		"""Initialize this instance from its parent write channel"""
+		self.channel = channel
+		
+	#{ Interface
+	
+	def read(self, count=0, block=True, timeout=None):
+		# if the channel is closed for writing, we never block
+		# NOTE: is handled by the queue
+		# We don't check for a closed state here has it costs time - most of 
+		# the time, it will not be closed, and will bail out automatically once
+		# it gets closed
+		
+		
+		# in non-blocking mode, its all not a problem
+		out = list()
+		queue = self.channel.queue
+		if not block:
+			# be as fast as possible in non-blocking mode, hence
+			# its a bit 'unrolled'
+			try:
+				if count == 1:
+					out.append(queue.get(False))
+				elif count < 1:
+					while True:
+						out.append(queue.get(False))
+					# END for each item
+				else:
+					for i in xrange(count):
+						out.append(queue.get(False))
+					# END for each item
+				# END handle count
+			except Empty:
+				pass
+			# END handle exceptions
+		else:
+			# to get everything into one loop, we set the count accordingly
+			if count == 0:
+				count = sys.maxint
+			# END handle count
+			
+			i = 0
+			while i < count:
+				try:
+					out.append(queue.get(block, timeout))
+					i += 1
+				except Empty:
+					# here we are only if 
+					# someone woke us up to inform us about the queue that changed
+					# its writable state
+					# The following branch checks for closed channels, and pulls
+					# as many items as we need and as possible, before 
+					# leaving the loop.
+					if not queue.writable():
+						try:
+							while i < count:
+								out.append(queue.get(False, None))
+								i += 1
+							# END count loop
+						except Empty:
+							break	# out of count loop 
+						# END handle absolutely empty queue
+					# END handle closed channel 
+					
+					# if we are here, we woke up and the channel is not closed
+					# Either the queue became writable again, which currently shouldn't
+					# be able to happen in the channel, or someone read with a timeout
+					# that actually timed out.
+					# As it timed out, which is the only reason we are here, 
+					# we have to abort
+					break
+				# END ignore empty
+				
+			# END for each item
+		# END handle blocking
+		return out
+		
+	#} END interface 
+
+
+class CallbackReaderMixin(object):
+	"""A channel which sends a callback before items are read from the channel"""
+	# unfortunately, slots can only use direct inheritance, have to turn it off :(
+	# __slots__ = "_pre_cb"
+	
+	def __init__(self, *args):
+		super(CallbackReaderMixin, self).__init__(*args)
+		self._pre_cb = None
+	
+	def set_pre_cb(self, fun = lambda count: None):
+		"""Install a callback to call with the item count to be read before any 
+		item is actually read from the channel. 
+		Exceptions will be propagated.
+		If a function is not provided, the call is effectively uninstalled.
+		:return: the previously installed callback or None
+		:note: The callback must be threadsafe if the channel is used by multiple threads."""
+		prev = self._pre_cb
+		self._pre_cb = fun
+		return prev
+	
+	def read(self, count=0, block=True, timeout=None):
+		if self._pre_cb:
+			self._pre_cb(count)
+		return super(CallbackReaderMixin, self).read(count, block, timeout)
+		
+		
+class CallbackChannelReader(CallbackReaderMixin, ChannelReader):
+	"""Implements a channel reader with callback functionality"""
+	pass
+
+
+class IteratorReader(Reader):
+	"""A Reader allowing to read items from an iterator, instead of a channel.
+	Reads will never block. Its thread-safe"""
+	__slots__ = ("_empty", '_iter', '_lock')
+	
+	# the type of the lock to use when reading from the iterator
+	lock_type = threading.Lock
+	
+	def __init__(self, iterator):
+		self._empty = False
+		if not hasattr(iterator, 'next'):
+			raise ValueError("Iterator %r needs a next() function" % iterator)
+		self._iter = iterator
+		self._lock = self.lock_type()
+		
+	def read(self, count=0, block=True, timeout=None):
+		"""Non-Blocking implementation of read"""
+		# not threadsafe, but worst thing that could happen is that 
+		# we try to get items one more time
+		if self._empty:
+			return list()
+		# END early abort
+		
+		self._lock.acquire()
+		try:
+			if count == 0:
+				self._empty = True
+				return list(self._iter)
+			else:
+				out = list()
+				it = self._iter
+				for i in xrange(count):
+					try:
+						out.append(it.next())
+					except StopIteration:
+						self._empty = True
+						break
+					# END handle empty iterator
+				# END for each item to take
+				return out
+			# END handle count
+		finally:
+			self._lock.release()
+		# END handle locking
+		
+
+#} END classes
+
+#{ Constructors
+def mkchannel(ctype = Channel, wtype = ChannelWriter, rtype = ChannelReader):
+	"""Create a channel, with a reader and a writer
+	:return: tuple(reader, writer)
+	:param ctype: Channel to instantiate
+	:param wctype: The type of the write channel to instantiate
+	:param rctype: The type of the read channel to instantiate"""
+	c = ctype()
+	wc = wtype(c)
+	rc = rtype(c)
+	return wc, rc
+#} END constructors
diff --git a/lib/git/async/graph.py b/lib/git/async/graph.py
new file mode 100644
index 00000000..4e14c81e
--- /dev/null
+++ b/lib/git/async/graph.py
@@ -0,0 +1,126 @@
+"""Simplistic implementation of a graph"""
+
+__all__ = ('Node', 'Graph')
+
+class Node(object):
+	"""A Node in the graph. They know their neighbours, and have an id which should 
+	resolve into a string"""
+	__slots__ = ('in_nodes', 'out_nodes', 'id')
+	
+	def __init__(self, id=None):
+		self.id = id
+		self.in_nodes = list()
+		self.out_nodes = list()
+		
+	def __str__(self):
+		return str(self.id)
+		
+	def __repr__(self):
+		return "%s(%s)" % (type(self).__name__, self.id)
+	
+	
+class Graph(object):
+	"""A simple graph implementation, keeping nodes and providing basic access and 
+	editing functions. The performance is only suitable for small graphs of not 
+	more than 10 nodes !"""
+	__slots__ = "nodes"
+	
+	def __init__(self):
+		self.nodes = list()
+
+	def __del__(self):
+		"""Deletes bidericational dependencies"""
+		for node in self.nodes:
+			node.in_nodes = None
+			node.out_nodes = None
+		# END cleanup nodes
+		
+		# otherwise the nodes would keep floating around
+	
+
+	def add_node(self, node):
+		"""Add a new node to the graph
+		:return: the newly added node"""
+		self.nodes.append(node)
+		return node
+	
+	def remove_node(self, node):
+		"""Delete a node from the graph
+		:return: self"""
+		try:
+			del(self.nodes[self.nodes.index(node)])
+		except ValueError:
+			return self
+		# END ignore if it doesn't exist
+		
+		# clear connections
+		for outn in node.out_nodes:
+			del(outn.in_nodes[outn.in_nodes.index(node)])
+		for inn in node.in_nodes:
+			del(inn.out_nodes[inn.out_nodes.index(node)])
+		node.out_nodes = list()
+		node.in_nodes = list()
+		return self
+	
+	def add_edge(self, u, v):
+		"""Add an undirected edge between the given nodes u and v.
+		
+		return: self
+		:raise ValueError: If the new edge would create a cycle"""
+		if u is v:
+			raise ValueError("Cannot connect a node with itself")
+		
+		# are they already connected ?
+		if 	u in v.in_nodes and v in u.out_nodes or \
+			v in u.in_nodes and u in v.out_nodes:
+			return self
+		# END handle connection exists
+		
+		# cycle check - if we can reach any of the two by following either ones 
+		# history, its a cycle
+		for start, end in ((u, v), (v,u)):
+			if not start.in_nodes: 
+				continue
+			nodes = start.in_nodes[:]
+			seen = set()
+			# depth first search - its faster
+			while nodes:
+				n = nodes.pop()
+				if n in seen:
+					continue
+				seen.add(n)
+				if n is end:
+					raise ValueError("Connecting u with v would create a cycle")
+				nodes.extend(n.in_nodes)
+			# END while we are searching
+		# END for each direction to look
+		
+		# connection is valid, set it up
+		u.out_nodes.append(v)
+		v.in_nodes.append(u)
+		
+		return self
+	
+	def input_inclusive_dfirst_reversed(self, node):
+		"""Return all input nodes of the given node, depth first,
+		It will return the actual input node last, as it is required
+		like that by the pool"""
+		stack = [node]
+		seen = set()
+		
+		# depth first
+		out = list()
+		while stack:
+			n = stack.pop()
+			if n in seen:
+				continue
+			seen.add(n)
+			out.append(n)
+			
+			# only proceed in that direction if visitor is fine with it
+			stack.extend(n.in_nodes)
+			# END call visitor
+		# END while walking
+		out.reverse()
+		return out
+		
diff --git a/lib/git/async/pool.py b/lib/git/async/pool.py
new file mode 100644
index 00000000..8f33a029
--- /dev/null
+++ b/lib/git/async/pool.py
@@ -0,0 +1,488 @@
+"""Implementation of a thread-pool working with channels"""
+from thread import (
+		WorkerThread, 
+		StopProcessing,
+		)
+from threading import Lock
+
+from util import (
+		AsyncQueue,
+		DummyLock
+	)
+
+from Queue import (
+	Queue, 
+	Empty
+	)
+
+from graph import Graph 
+from channel import (
+		mkchannel,
+		ChannelWriter, 
+		Channel,
+		SerialChannel,
+		CallbackChannelReader
+	)
+
+import sys
+import weakref
+from time import sleep
+import new
+
+
+__all__ = ('PoolReader', 'Pool', 'ThreadPool')
+
+
+class PoolReader(CallbackChannelReader):
+	"""A reader designed to read from channels which take part in pools
+	It acts like a handle to the underlying task in the pool."""
+	__slots__ = ('_task_ref', '_pool_ref')
+	
+	def __init__(self, channel, task, pool):
+		CallbackChannelReader.__init__(self, channel)
+		self._task_ref = weakref.ref(task)
+		self._pool_ref = weakref.ref(pool)
+		
+	def __del__(self):
+		"""Assures that our task will be deleted if we were the last reader"""
+		task = self._task_ref()
+		if task is None:
+			return
+		
+		pool = self._pool_ref()
+		if pool is None:
+			return
+		
+		# if this is the last reader to the wc we just handled, there 
+		# is no way anyone will ever read from the task again. If so, 
+		# delete the task in question, it will take care of itself and orphans
+		# it might leave
+		# 1 is ourselves, + 1 for the call + 1, and 3 magical ones which 
+		# I can't explain, but appears to be normal in the destructor
+		# On the caller side, getrefcount returns 2, as expected
+		# When just calling remove_task, 
+		# it has no way of knowing that the write channel is about to diminsh.
+		# which is why we pass the info as a private kwarg  - not nice, but 
+		# okay for now
+		if sys.getrefcount(self) < 6:
+			pool.remove_task(task, _from_destructor_ = True)
+		# END handle refcount based removal of task
+
+	#{ Internal
+	def _read(self, count=0, block=True, timeout=None):
+		return CallbackChannelReader.read(self, count, block, timeout)
+	
+	#} END internal
+
+	#{ Interface
+	
+	def pool_ref(self):
+		""":return: reference to the pool we belong to"""
+		return self._pool_ref
+		
+	def task_ref(self):
+		""":return: reference to the task producing our items"""
+		return self._task_ref
+	
+	#} END interface 
+	
+	def read(self, count=0, block=True, timeout=None):
+		"""Read an item that was processed by one of our threads
+		:note: Triggers task dependency handling needed to provide the necessary 
+			input"""
+		# NOTE: we always queue the operation that would give us count items
+		# as tracking the scheduled items or testing the channels size
+		# is in herently unsafe depending on the design of the task network
+		# If we put on tasks onto the queue for every request, we are sure
+		# to always produce enough items, even if the task.min_count actually
+		# provided enough - its better to have some possibly empty task runs 
+		# than having and empty queue that blocks.
+		
+		# if the user tries to use us to read from a done task, we will never 
+		# compute as all produced items are already in the channel
+		task = self._task_ref()
+		if task is None:
+			return list()
+		# END abort if task was deleted
+			
+		skip_compute = task.is_done() or task.error()
+		
+		########## prepare ##############################
+		if not skip_compute:
+			self._pool_ref()._prepare_channel_read(task, count)
+		# END prepare pool scheduling
+		
+		
+		####### read data ########
+		##########################
+		# read actual items, tasks were setup to put their output into our channel ( as well )
+		items = CallbackChannelReader.read(self, count, block, timeout)
+		##########################
+		
+		
+		return items
+		
+	
+	
+class Pool(object):
+	"""A thread pool maintains a set of one or more worker threads, but supports 
+	a fully serial mode in which case the amount of threads is zero.
+	
+	Work is distributed via Channels, which form a dependency graph. The evaluation
+	is lazy, as work will only be done once an output is requested.
+	
+	The thread pools inherent issue is the global interpreter lock that it will hit, 
+	which gets worse considering a few c extensions specifically lock their part
+	globally as well. The only way this will improve is if custom c extensions
+	are written which do some bulk work, but release the GIL once they have acquired
+	their resources.
+	
+	Due to the nature of having multiple objects in git, its easy to distribute 
+	that work cleanly among threads.
+	
+	:note: the current implementation returns channels which are meant to be 
+		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
+					'_num_workers',			# list of workers
+					'_queue', 				# master queue for tasks
+					'_taskorder_cache', 	# map task id -> ordered dependent tasks
+					'_taskgraph_lock',		# lock for accessing the task graph
+				)
+	
+	# CONFIGURATION
+	# The type of worker to create - its expected to provide the Thread interface, 
+	# taking the taskqueue as only init argument
+	# as well as a method called stop_and_join() to terminate it
+	WorkerCls = None
+	
+	# The type of lock to use to protect critical sections, providing the 
+	# threading.Lock interface
+	LockCls = None
+	
+	# the type of the task queue to use - it must provide the Queue interface
+	TaskQueueCls = None
+	
+	
+	def __init__(self, size=0):
+		self._tasks = Graph()
+		self._num_workers = 0
+		self._queue = self.TaskQueueCls()
+		self._taskgraph_lock = self.LockCls()
+		self._taskorder_cache = dict()
+		self.set_size(size)
+		
+	def __del__(self):
+		self.set_size(0)
+	
+	#{ Internal
+		
+	def _prepare_channel_read(self, task, count):
+		"""Process the tasks which depend on the given one to be sure the input 
+		channels are filled with data once we process the actual task
+		
+		Tasks have two important states: either they are done, or they are done 
+		and have an error, so they are likely not to have finished all their work.
+		
+		Either way, we will put them onto a list of tasks to delete them, providng 
+		information about the failed ones.
+		
+		Tasks which are not done will be put onto the queue for processing, which 
+		is fine as we walked them depth-first."""
+		# for the walk, we must make sure the ordering does not change. Even 
+		# when accessing the cache, as it is related to graph changes
+		self._taskgraph_lock.acquire()
+		try:
+			try:
+				dfirst_tasks = self._taskorder_cache[id(task)]
+			except KeyError:
+				# have to retrieve the list from the graph
+				dfirst_tasks = self._tasks.input_inclusive_dfirst_reversed(task)
+				self._taskorder_cache[id(task)] = dfirst_tasks
+			# END handle cached order retrieval
+		finally:
+			self._taskgraph_lock.release()
+		# END handle locking
+		
+		# check the min count on all involved tasks, and be sure that we don't 
+		# have any task which produces less than the maximum min-count of all tasks
+		# The actual_count is used when chunking tasks up for the queue, whereas 
+		# the count is usued to determine whether we still have enough output
+		# on the queue, checking qsize ( ->revise )
+		# ABTRACT: If T depends on T-1, and the client wants 1 item, T produces
+		# at least 10, T-1 goes with 1, then T will block after 1 item, which 
+		# is read by the client. On the next read of 1 item, we would find T's 
+		# queue empty and put in another 10, which could put another thread into 
+		# blocking state. T-1 produces one more item, which is consumed right away
+		# by the two threads running T. Although this works in the end, it leaves
+		# many threads blocking and waiting for input, which is not desired.
+		# Setting the min-count to the max of the mincount of all tasks assures
+		# we have enough items for all.
+		# Addition: in serial mode, we would enter a deadlock if one task would
+		# ever wait for items !
+		actual_count = count
+		min_counts = (((t.min_count is not None and t.min_count) or count) for t in dfirst_tasks)
+		min_count = reduce(lambda m1, m2: max(m1, m2), min_counts)
+		if 0 < count < min_count:
+			actual_count = min_count
+		# END set actual count
+		
+		# the list includes our tasks - the first one to evaluate first, the 
+		# requested one last
+		for task in dfirst_tasks: 
+			# if task.error() or task.is_done():
+				# in theory, the should never be consumed task in the pool, right ?
+				# They delete themselves once they are done. But as we run asynchronously, 
+				# It can be that someone reads, while a task realizes its done, and 
+				# we get here to prepare the read although it already is done.
+				# Its not a problem though, the task wiill not do anything.
+				# Hence we don't waste our time with checking for it
+				# raise AssertionError("Shouldn't have consumed tasks on the pool, they delete themeselves, what happend ?")
+			# END skip processing
+			
+			# but use the actual count to produce the output, we may produce 
+			# more than requested
+			numchunks = 1
+			chunksize = actual_count
+			remainder = 0
+			
+			# we need the count set for this - can't chunk up unlimited items
+			# In serial mode we could do this by checking for empty input channels, 
+			# but in dispatch mode its impossible ( == not easily possible )
+			# Only try it if we have enough demand
+			if task.max_chunksize and actual_count > task.max_chunksize:
+				numchunks = actual_count / task.max_chunksize
+				chunksize = task.max_chunksize
+				remainder = actual_count - (numchunks * chunksize)
+			# END handle chunking
+			
+			# the following loops are kind of unrolled - code duplication
+			# should make things execute faster. Putting the if statements 
+			# into the loop would be less code, but ... slower
+			if self._num_workers:
+				# respect the chunk size, and split the task up if we want 
+				# to process too much. This can be defined per task
+				qput = self._queue.put
+				if numchunks > 1:
+					for i in xrange(numchunks):
+						qput((task.process, chunksize))
+					# END for each chunk to put
+				else:
+					qput((task.process, chunksize))
+				# END try efficient looping
+				
+				if remainder:
+					qput((task.process, remainder))
+				# END handle chunksize
+			else:
+				# no workers, so we have to do the work ourselves
+				if numchunks > 1:
+					for i in xrange(numchunks):
+						task.process(chunksize)
+					# END for each chunk to put
+				else:
+					task.process(chunksize)
+				# END try efficient looping
+				
+				if remainder:
+					task.process(remainder)
+				# END handle chunksize
+			# END handle serial mode
+		# END for each task to process
+		
+		
+	def _remove_task_if_orphaned(self, task, from_destructor):
+		"""Check the task, and delete it if it is orphaned"""
+		# 1 for writer on task, 1 for the getrefcount call + 1 for each other writer/reader
+		# If we are getting here from the destructor of an RPool channel, 
+		# its totally valid to virtually decrement the refcount by 1 as 
+		# we can expect it to drop once the destructor completes, which is when
+		# we finish all recursive calls
+		max_ref_count = 3 + from_destructor
+		if sys.getrefcount(task.writer().channel) < max_ref_count:
+			self.remove_task(task, from_destructor)
+	#} END internal
+	
+	#{ Interface 
+	def size(self):
+		""":return: amount of workers in the pool
+		:note: method is not threadsafe !"""
+		return self._num_workers
+	
+	def set_size(self, size=0):
+		"""Set the amount of workers to use in this pool. When reducing the size, 
+		threads will continue with their work until they are done before effectively
+		being removed.
+		
+		: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.
+			If the size is 0, newly added tasks will use channels which are NOT 
+			threadsafe to optimize item throughput.
+		
+		:note: currently NOT threadsafe !"""
+		assert size > -1, "Size cannot be negative"
+		
+		# either start new threads, or kill existing ones.
+		# If we end up with no threads, we process the remaining chunks on the queue
+		# ourselves
+		cur_count = self._num_workers
+		if cur_count < size:
+			# we can safely increase the size, even from serial mode, as we would
+			# only be able to do this if the serial ( sync ) mode finished processing.
+			# Just adding more workers is not a problem at all.
+			add_count = size - cur_count
+			for i in range(add_count):
+				self.WorkerCls(self._queue).start()
+			# END for each new worker to create
+			self._num_workers += add_count
+		elif cur_count > size:
+			# We don't care which thread exactly gets hit by our stop request
+			# On their way, they will consume remaining tasks, but new ones 
+			# could be added as we speak.
+			del_count = cur_count - size
+			for i in range(del_count):
+				self._queue.put((self.WorkerCls.stop, True))	# arg doesnt matter
+			# END for each thread to stop
+			self._num_workers -= del_count
+		# END handle count
+		
+		if size == 0:
+			# NOTE: we do not preocess any tasks still on the queue, as we ill 
+			# naturally do that once we read the next time, only on the tasks
+			# that are actually required. The queue will keep the tasks, 
+			# and once we are deleted, they will vanish without additional
+			# time spend on them. If there shouldn't be any consumers anyway.
+			# If we should reenable some workers again, they will continue on the 
+			# remaining tasks, probably with nothing to do.
+			# We can't clear the task queue if we have removed workers 
+			# as they will receive the termination signal through it, and if 
+			# we had added workers, we wouldn't be here ;).
+			pass 
+		# END process queue
+		return self
+		
+	def num_tasks(self):
+		""":return: amount of tasks"""
+		self._taskgraph_lock.acquire()
+		try:
+			return len(self._tasks.nodes)
+		finally:
+			self._taskgraph_lock.release()
+		
+	def remove_task(self, task, _from_destructor_ = False):
+		"""Delete the task
+		Additionally we will remove orphaned tasks, which can be identified if their 
+		output channel is only held by themselves, so no one will ever consume 
+		its items.
+		
+		This method blocks until all tasks to be removed have been processed, if 
+		they are currently being processed.
+		:return: self"""
+		self._taskgraph_lock.acquire()
+		try:
+			# it can be that the task is already deleted, but its chunk was on the 
+			# queue until now, so its marked consumed again
+			if not task in self._tasks.nodes:
+				return self
+			# END early abort
+			
+			# the task we are currently deleting could also be processed by 
+			# a thread right now. We don't care about it as its taking care about
+			# its write channel itself, and sends everything it can to it.
+			# For it it doesn't matter that its not part of our task graph anymore.
+		
+			# now delete our actual node - be sure its done to prevent further 
+			# processing in case there are still client reads on their way.
+			task.set_done()
+			
+			# keep its input nodes as we check whether they were orphaned
+			in_tasks = task.in_nodes
+			self._tasks.remove_node(task)
+			self._taskorder_cache.clear()
+		finally:
+			self._taskgraph_lock.release()
+		# END locked deletion
+		
+		for t in in_tasks:
+			self._remove_task_if_orphaned(t, _from_destructor_)
+		# END handle orphans recursively
+		
+		return self
+	
+	def add_task(self, task):
+		"""Add a new task to be processed.
+		:return: a read channel to retrieve processed items. If that handle is lost, 
+			the task will be considered orphaned and will be deleted on the next 
+			occasion."""
+		# create a write channel for it
+		ctype = Channel
+		
+		# adjust the task with our pool ref, if it has the slot and is empty
+		# For now, we don't allow tasks to be used in multiple pools, except
+		# for by their channels
+		if hasattr(task, 'pool'):
+			their_pool = task.pool()
+			if their_pool is None:
+				task.set_pool(self)
+			elif their_pool is not self:
+				raise ValueError("Task %r is already registered to another pool" % task.id)
+			# END handle pool exclusivity
+		# END handle pool aware tasks
+		
+		self._taskgraph_lock.acquire()
+		try:
+			self._taskorder_cache.clear()
+			self._tasks.add_node(task)
+			
+			# Use a non-threadsafe queue
+			# This brings about 15% more performance, but sacrifices thread-safety
+			if self.size() == 0:
+				ctype = SerialChannel
+			# END improve locks
+			
+			# setup the tasks channel - respect the task creators choice though
+			# if it is set.
+			wc = task.writer()
+			ch = None
+			if wc is None:
+				ch = ctype()
+				wc = ChannelWriter(ch)
+				task.set_writer(wc)
+			else:
+				ch = wc.channel
+			# END create write channel ifunset
+			rc = PoolReader(ch, task, self)
+		finally:
+			self._taskgraph_lock.release()
+		# END sync task addition
+		
+		# If the input channel is one of our read channels, we add the relation
+		if hasattr(task, 'reader'):
+			ic = task.reader()
+			if hasattr(ic, 'pool_ref') and ic.pool_ref()() is self:
+				self._taskgraph_lock.acquire()
+				try:
+					self._tasks.add_edge(ic._task_ref(), task)
+					
+					# additionally, bypass ourselves when reading from the 
+					# task, if possible
+					if hasattr(ic, '_read'):
+						task.set_read(ic._read)
+					# END handle read bypass
+				finally:
+					self._taskgraph_lock.release()
+				# END handle edge-adding
+			# END add task relation
+		# END handle input channels for connections
+		
+		return rc
+			
+	#} END interface 
+	
+	
+class ThreadPool(Pool):
+	"""A pool using threads as worker"""
+	WorkerCls = WorkerThread
+	LockCls = Lock
+	TaskQueueCls = AsyncQueue
diff --git a/lib/git/async/task.py b/lib/git/async/task.py
new file mode 100644
index 00000000..ac948dc0
--- /dev/null
+++ b/lib/git/async/task.py
@@ -0,0 +1,237 @@
+from graph import Node
+from util import ReadOnly
+from channel import IteratorReader
+
+
+import threading
+import weakref
+import sys
+import new
+
+__all__ = ('Task', 'ThreadTaskBase', 'IteratorTaskBase', 
+			'IteratorThreadTask', 'ChannelThreadTask')
+
+class Task(Node):
+	"""Abstracts a named task, which contains 
+	additional information on how the task should be queued and processed.
+	
+	Results of the item processing are sent to a writer, which is to be 
+	set by the creator using the ``set_writer`` method.
+	
+	Items are read using the internal ``_read`` callable, subclasses are meant to
+	set this to a callable that supports the Reader interface's read function.
+	
+	* **min_count** assures that not less than min_count items will be processed per call.
+	* **max_chunksize** assures that multi-threading is happening in smaller chunks. If 
+	 someone wants all items to be processed, using read(0), the whole task would go to
+	 one worker, as well as dependent tasks. If you want finer granularity , you can 
+	 specify this here, causing chunks to be no larger than max_chunksize
+	* **apply_single** if True, default True, individual items will be given to the 
+		worker function. If False, a list of possibly multiple items will be passed
+		instead."""
+	__slots__ = (	'_read',			# method to yield items to process 
+					'_out_writer', 			# output write channel
+					'_exc',				# exception caught
+					'_done',			# True if we are done
+					'_num_writers',		# number of concurrent writers
+					'_wlock',			# lock for the above
+					'fun',				# function to call with items read
+					'min_count', 		# minimum amount of items to produce, None means no override
+					'max_chunksize',	# maximium amount of items to process per process call
+					'apply_single'		# apply single items even if multiple where read
+					)
+	
+	def __init__(self, id, fun, apply_single=True, min_count=None, max_chunksize=0, 
+					writer=None):
+		Node.__init__(self, id)
+		self._read = None					# to be set by subclasss 
+		self._out_writer = writer
+		self._exc = None
+		self._done = False
+		self._num_writers = 0
+		self._wlock = threading.Lock()
+		self.fun = fun
+		self.min_count = None
+		self.max_chunksize = 0				# note set
+		self.apply_single = apply_single
+	
+	def is_done(self):
+		""":return: True if we are finished processing"""
+		return self._done
+		
+	def set_done(self):
+		"""Set ourselves to being done, has we have completed the processing"""
+		self._done = True
+		
+	def set_writer(self, writer):
+		"""Set the write channel to the given one"""
+		self._out_writer = writer
+		
+	def writer(self):
+		""":return: a proxy to our write channel or None if non is set
+		:note: you must not hold a reference to our write channel when the 
+			task is being processed. This would cause the write channel never 
+			to be closed as the task will think there is still another instance
+			being processed which can close the channel once it is done.
+			In the worst case, this will block your reads."""
+		if self._out_writer is None:
+			return None
+		return self._out_writer
+		
+	def close(self):
+		"""A closed task will close its channel to assure the readers will wake up
+		:note: its safe to call this method multiple times"""
+		self._out_writer.close()
+		
+	def is_closed(self):
+		""":return: True if the task's write channel is closed"""
+		return self._out_writer.closed()
+		
+	def error(self):
+		""":return: Exception caught during last processing or None"""
+		return self._exc
+
+	def process(self, count=0):
+		"""Process count items and send the result individually to the output channel"""
+		# first thing: increment the writer count - other tasks must be able 
+		# to respond properly ( even if it turns out we don't need it later )
+		self._wlock.acquire()
+		self._num_writers += 1
+		self._wlock.release()
+		
+		items = self._read(count)
+		
+		try:
+			try:
+				if items:
+					write = self._out_writer.write
+					if self.apply_single:
+						for item in items:
+							rval = self.fun(item)
+							write(rval)
+						# END for each item
+					else:
+						# shouldn't apply single be the default anyway ? 
+						# The task designers should chunk them up in advance
+						rvals = self.fun(items)
+						for rval in rvals:
+							write(rval)
+					# END handle single apply
+				# END if there is anything to do
+			finally:
+				self._wlock.acquire()
+				self._num_writers -= 1
+				self._wlock.release()
+			# END handle writer count
+		except Exception, e:
+			# be sure our task is not scheduled again
+			self.set_done()
+			
+			# PROBLEM: We have failed to create at least one item, hence its not 
+			# garantueed that enough items will be produced for a possibly blocking
+			# client on the other end. This is why we have no other choice but
+			# to close the channel, preventing the possibility of blocking.
+			# This implies that dependent tasks will go down with us, but that is
+			# just the right thing to do of course - one loose link in the chain ...
+			# Other chunks of our kind currently being processed will then 
+			# fail to write to the channel and fail as well
+			self.close()
+			
+			# If some other chunk of our Task had an error, the channel will be closed
+			# This is not an issue, just be sure we don't overwrite the original 
+			# exception with the ReadOnly error that would be emitted in that case.
+			# We imply that ReadOnly is exclusive to us, as it won't be an error
+			# if the user emits it
+			if not isinstance(e, ReadOnly):
+				self._exc = e
+			# END set error flag
+		# END exception handling
+		
+		
+		# if we didn't get all demanded items, which is also the case if count is 0
+		# we have depleted the input channel and are done
+		# We could check our output channel for how many items we have and put that 
+		# into the equation, but whats important is that we were asked to produce
+		# count items.
+		if not items or len(items) != count:
+			self.set_done()
+		# END handle done state
+		
+		# If we appear to be the only one left with our output channel, and are 
+		# done ( this could have been set in another thread as well ), make 
+		# sure to close the output channel.
+		# Waiting with this to be the last one helps to keep the 
+		# write-channel writable longer
+		# The count is: 1 = wc itself, 2 = first reader channel, + x for every 
+		# thread having its copy on the stack 
+		# + 1 for the instance we provide to refcount
+		# Soft close, so others can continue writing their results
+		if self.is_done():
+			self._wlock.acquire()
+			try:
+				if self._num_writers == 0:
+					self.close()
+				# END handle writers
+			finally:
+				self._wlock.release()
+			# END assure lock release
+		# END handle channel closure
+	#{ Configuration
+
+
+class ThreadTaskBase(object):
+	"""Describes tasks which can be used with theaded pools"""
+	pass
+
+
+class IteratorTaskBase(Task):
+	"""Implements a task which processes items from an iterable in a multi-processing 
+	safe manner"""
+	__slots__ = tuple()
+	
+	
+	def __init__(self, iterator, *args, **kwargs):
+		Task.__init__(self, *args, **kwargs)
+		self._read = IteratorReader(iterator).read
+		# defaults to returning our items unchanged
+		self.fun = lambda item: item
+				
+		
+class IteratorThreadTask(IteratorTaskBase, ThreadTaskBase):
+	"""An input iterator for threaded pools"""
+	lock_type = threading.Lock
+		
+
+class ChannelThreadTask(Task, ThreadTaskBase):
+	"""Uses an input channel as source for reading items
+	For instantiation, it takes all arguments of its base, the first one needs
+	to be the input channel to read from though."""
+	__slots__ = "_pool_ref"
+	
+	def __init__(self, in_reader, *args, **kwargs):
+		Task.__init__(self, *args, **kwargs)
+		self._read = in_reader.read
+		self._pool_ref = None
+
+	#{ Internal Interface 
+	
+	def reader(self):
+		""":return: input channel from which we read"""
+		# the instance is bound in its instance method - lets use this to keep
+		# the refcount at one ( per consumer )
+		return self._read.im_self
+		
+	def set_read(self, read):
+		"""Adjust the read method to the given one"""
+		self._read = read
+		
+	def set_pool(self, pool):
+		self._pool_ref = weakref.ref(pool)
+		
+	def pool(self):
+		""":return: pool we are attached to, or None"""
+		if self._pool_ref is None:
+			return None
+		return self._pool_ref()
+		
+	#} END intenral interface
diff --git a/lib/git/async/thread.py b/lib/git/async/thread.py
new file mode 100644
index 00000000..96b4f0c4
--- /dev/null
+++ b/lib/git/async/thread.py
@@ -0,0 +1,201 @@
+# -*- coding: utf-8 -*-
+"""Module with threading utilities"""
+__docformat__ = "restructuredtext"
+import threading
+import inspect
+import Queue
+
+import sys
+
+__all__ = ('do_terminate_threads', 'terminate_threads', 'TerminatableThread', 
+			'WorkerThread') 
+		
+
+#{ Decorators
+
+def do_terminate_threads(whitelist=list()):
+	"""Simple function which terminates all of our threads
+	:param whitelist: If whitelist is given, only the given threads will be terminated"""
+	for t in threading.enumerate():
+		if not isinstance(t, TerminatableThread):
+			continue
+		if whitelist and t not in whitelist:
+			continue
+		t.stop_and_join()
+	# END for each thread
+
+def terminate_threads( func ):
+	"""Kills all worker threads the method has created by sending the quit signal.
+	This takes over in case of an error in the main function"""
+	def wrapper(*args, **kwargs):
+		cur_threads = set(threading.enumerate())
+		try:
+			return func(*args, **kwargs)
+		finally:
+			do_terminate_threads(set(threading.enumerate()) - cur_threads)
+		# END finally shutdown threads
+	# END wrapper 
+	wrapper.__name__ = func.__name__
+	return wrapper
+
+#} END decorators
+
+#{ Classes
+	
+class TerminatableThread(threading.Thread):
+	"""A simple thread able to terminate itself on behalf of the user.
+	
+	Terminate a thread as follows:
+	
+	t.stop_and_join()
+	
+	Derived classes call _should_terminate() to determine whether they should 
+	abort gracefully
+	"""
+	__slots__ = '_terminate'
+	
+	def __init__(self):
+		super(TerminatableThread, self).__init__()
+		self._terminate = False
+		
+		
+	#{ Subclass Interface
+	def _should_terminate(self):
+		""":return: True if this thread should terminate its operation immediately"""
+		return self._terminate
+		
+	def _terminated(self):
+		"""Called once the thread terminated. Its called in the main thread
+		and may perform cleanup operations"""
+		pass
+
+	def start(self):
+		"""Start the thread and return self"""
+		super(TerminatableThread, self).start()
+		return self
+	
+	#} END subclass interface
+		
+	#{ Interface 
+		
+	def stop_and_join(self):
+		"""Ask the thread to stop its operation and wait for it to terminate
+		:note: Depending on the implenetation, this might block a moment"""
+		self._terminate = True
+		self.join()
+		self._terminated()
+	#} END interface
+	
+	
+class StopProcessing(Exception):
+	"""If thrown in a function processed by a WorkerThread, it will terminate"""
+	
+
+class WorkerThread(TerminatableThread):
+	""" This base allows to call functions on class instances natively.
+	As it is meant to work with a pool, the result of the call must be 
+	handled by the callee.
+	The thread runs forever unless it receives the terminate signal using 
+	its task queue.
+	
+	Tasks could be anything, but should usually be class methods and arguments to
+	allow the following:
+	
+	inq = Queue()
+	w = WorkerThread(inq)
+	w.start()
+	inq.put((WorkerThread.<method>, args, kwargs))
+	
+	finally we call quit to terminate asap.
+	
+	alternatively, you can make a call more intuitively - the output is the output queue
+	allowing you to get the result right away or later
+	w.call(arg, kwarg='value').get()
+	
+	inq.put(WorkerThread.quit)
+	w.join()
+	
+	You may provide the following tuples as task:
+	t[0] = class method, function or instance method
+	t[1] = optional, tuple or list of arguments to pass to the routine
+	t[2] = optional, dictionary of keyword arguments to pass to the routine
+	"""
+	__slots__ = ('inq')
+	
+	
+	# 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
+		if inq is None:
+			self.inq = Queue.Queue()
+	
+	@classmethod
+	def stop(cls, *args):
+		"""If send via the inq of the thread, it will stop once it processed the function"""
+		raise StopProcessing
+	
+	def run(self):
+		"""Process input tasks until we receive the quit signal"""
+		gettask = self.inq.get
+		while True:
+			if self._should_terminate():
+				break
+			# END check for stop request
+			
+			# note: during shutdown, this turns None in the middle of waiting 
+			# for an item to be put onto it - we can't du anything about it - 
+			# even if we catch everything and break gracefully, the parent 
+			# call will think we failed with an empty exception.
+			# Hence we just don't do anything about it. Alternatively
+			# we could override the start method to get our own bootstrapping, 
+			# which would mean repeating plenty of code in of the threading module.
+			tasktuple = gettask()
+				
+			# needing exactly one function, and one arg
+			routine, arg = tasktuple
+			
+			try:
+				try:
+					rval = None
+					if inspect.ismethod(routine):
+						if routine.im_self is None:
+							rval = routine(self, arg)
+						else:
+							rval = routine(arg)
+					elif inspect.isroutine(routine):
+						rval = routine(arg)
+					else:
+						# ignore unknown items
+						sys.stderr.write("%s: task %s was not understood - terminating\n" % (self.getName(), str(tasktuple)))
+						break
+					# END make routine call
+				finally:
+					# make sure we delete the routine to release the reference as soon
+					# as possible. Otherwise objects might not be destroyed 
+					# while we are waiting
+					del(routine)
+					del(tasktuple)
+			except StopProcessing:
+				break
+			except Exception,e:
+				sys.stderr.write("%s: Task %s raised unhandled exception: %s - this really shouldn't happen !\n" % (self.getName(), str(tasktuple), str(e)))
+				continue	# just continue 
+			# END routine exception handling
+		
+			# END handle routine release
+		# END endless loop
+	
+	def stop_and_join(self):
+		"""Send stop message to ourselves - we don't block, the thread will terminate 
+		once it has finished processing its input queue to receive our termination
+		event"""
+		# DONT call superclass as it will try to join - join's don't work for 
+		# some reason, as python apparently doesn't switch threads (so often)
+		# while waiting ... I don't know, but the threads respond properly, 
+		# but only if dear python switches to them
+		self.inq.put((self.stop, None))
+#} END classes
diff --git a/lib/git/async/util.py b/lib/git/async/util.py
new file mode 100644
index 00000000..4c4f3929
--- /dev/null
+++ b/lib/git/async/util.py
@@ -0,0 +1,268 @@
+"""Module with utilities related to async operations"""
+
+from threading import (
+	Lock,
+	_allocate_lock,
+	_Condition, 
+	_sleep,
+	_time,
+	)
+
+from Queue import (
+		Empty,
+		)
+
+from collections import deque
+import sys
+import os
+
+#{ Routines 
+
+def cpu_count():
+	""":return:number of CPUs in the system
+	:note: inspired by multiprocessing"""
+	num = 0
+	try:
+		if sys.platform == 'win32':
+			num = int(os.environ['NUMBER_OF_PROCESSORS'])
+		elif 'bsd' in sys.platform or sys.platform == 'darwin':
+			num = int(os.popen('sysctl -n hw.ncpu').read())
+		else:
+			num = os.sysconf('SC_NPROCESSORS_ONLN')
+	except (ValueError, KeyError, OSError, AttributeError):
+		pass
+	# END exception handling
+	
+	if num == 0:
+		raise NotImplementedError('cannot determine number of cpus')
+	
+	return num
+	
+#} END routines
+
+
+
+class DummyLock(object):
+	"""An object providing a do-nothing lock interface for use in sync mode"""
+	__slots__ = tuple()
+	
+	def acquire(self):
+		pass
+	
+	def release(self):
+		pass
+	
+
+class SyncQueue(deque):
+	"""Adapter to allow using a deque like a queue, without locking"""
+	def get(self, block=True, timeout=None):
+		try:
+			return self.popleft()
+		except IndexError:
+			raise Empty
+		# END raise empty
+
+	def empty(self):
+		return len(self) == 0
+		
+	def set_writable(self, state):
+		pass
+	
+	def writable(self):
+		return True
+
+	def put(self, item, block=True, timeout=None):
+		self.append(item)
+	
+
+class HSCondition(deque):
+	"""Cleaned up code of the original condition object in order 
+	to make it run and respond faster."""
+	__slots__ = ("_lock")
+	delay = 0.0002					# reduces wait times, but increases overhead
+	
+	def __init__(self, lock=None):
+		if lock is None:
+			lock = Lock()
+		self._lock = lock
+
+	def release(self):
+		self._lock.release()
+		
+	def acquire(self, block=None):
+		if block is None:
+			self._lock.acquire()
+		else:
+			self._lock.acquire(block)
+
+	def wait(self, timeout=None):
+		waiter = _allocate_lock()
+		waiter.acquire()				# get it the first time, no blocking
+		self.append(waiter)
+		
+		
+		try:
+			# restore state no matter what (e.g., KeyboardInterrupt)
+			# now we block, as we hold the lock already
+			# in the momemnt we release our lock, someone else might actually resume
+			self._lock.release()
+			if timeout is None:
+				waiter.acquire()
+			else:
+				# Balancing act:  We can't afford a pure busy loop, because of the 
+				# GIL, so we have to sleep
+				# We try to sleep only tiny amounts of time though to be very responsive
+				# NOTE: this branch is not used by the async system anyway, but 
+				# will be hit when the user reads with timeout 
+				endtime = _time() + timeout
+				delay = self.delay
+				acquire = waiter.acquire
+				while True:
+					gotit = acquire(0)
+					if gotit:
+						break
+					remaining = endtime - _time()
+					if remaining <= 0:
+						break
+					# this makes 4 threads working as good as two, but of course
+					# it causes more frequent micro-sleeping
+					#delay = min(delay * 2, remaining, .05)
+					_sleep(delay)
+				# END endless loop
+				if not gotit:
+					try:
+						self.remove(waiter)
+					except ValueError:
+						pass
+				# END didn't ever get it
+		finally:
+			# reacquire the lock 
+			self._lock.acquire()
+		# END assure release lock
+			
+	def notify(self, n=1):
+		"""Its vital that this method is threadsafe - we absolutely have to 
+		get a lock at the beginning of this method to be sure we get the 
+		correct amount of waiters back. If we bail out, although a waiter
+		is about to be added, it will miss its wakeup notification, and block
+		forever (possibly)"""
+		self._lock.acquire()
+		try:
+			if not self:	# len(self) == 0, but this should be faster
+				return
+			if n == 1:
+				try:
+					self.popleft().release()
+				except IndexError:
+					pass
+			else:
+				for i in range(min(n, len(self))):
+					self.popleft().release()
+				# END for each waiter to resume
+			# END handle n = 1 case faster
+		finally:
+			self._lock.release()
+		# END assure lock is released
+	
+	def notify_all(self):
+		self.notify(len(self))
+		
+
+class ReadOnly(Exception):
+	"""Thrown when trying to write to a read-only queue"""
+
+class AsyncQueue(deque):
+	"""A queue using different condition objects to gain multithreading performance.
+	Additionally it has a threadsafe writable flag, which will alert all readers
+	that there is nothing more to get here.
+	All default-queue code was cleaned up for performance."""
+	__slots__ = ('mutex', 'not_empty', '_writable')
+	
+	def __init__(self, maxsize=0):
+		self.mutex = Lock()
+		self.not_empty = HSCondition(self.mutex)
+		self._writable = True
+		
+	def qsize(self):
+		self.mutex.acquire()
+		try:
+			return len(self)
+		finally:
+			self.mutex.release()
+
+	def writable(self):
+		self.mutex.acquire()
+		try:
+			return self._writable
+		finally:
+			self.mutex.release()
+
+	def set_writable(self, state):
+		"""Set the writable flag of this queue to True or False
+		:return: The previous state"""
+		self.mutex.acquire()
+		try:
+			old = self._writable
+			self._writable = state
+			return old
+		finally:
+			self.mutex.release()
+			# if we won't receive anymore items, inform the getters
+			if not state:
+				self.not_empty.notify_all()
+			# END tell everyone
+		# END handle locking
+
+	def empty(self):
+		self.mutex.acquire()
+		try:
+			return not len(self)
+		finally:
+			self.mutex.release()
+
+	def put(self, item, block=True, timeout=None):
+		self.mutex.acquire()
+		# NOTE: we explicitly do NOT check for our writable state
+		# Its just used as a notification signal, and we need to be able 
+		# to continue writing to prevent threads ( easily ) from failing
+		# to write their computed results, which we want in fact
+		# NO: we want them to fail and stop processing, as the one who caused
+		# the channel to close had a reason and wants the threads to 
+		# stop on the task as soon as possible
+		if not self._writable:
+			self.mutex.release()
+			raise ReadOnly
+		# END handle read-only
+		self.append(item)
+		self.mutex.release()
+		self.not_empty.notify()
+		
+	def get(self, block=True, timeout=None):
+		self.mutex.acquire()
+		try:
+			if block:
+				if timeout is None:
+					while not len(self) and self._writable:
+						self.not_empty.wait()
+				else:
+					endtime = _time() + timeout
+					while not len(self) and self._writable:
+						remaining = endtime - _time()
+						if remaining <= 0.0:
+							raise Empty
+						self.not_empty.wait(remaining)
+				# END handle timeout mode
+			# END handle block
+			
+			# can throw if we woke up because we are not writable anymore
+			try:
+				return self.popleft()
+			except IndexError:
+				raise Empty
+			# END handle unblocking reason
+		finally:
+			self.mutex.release()
+		# END assure lock is released
+
+
+#} END utilities