Merge branch 'channel' into taskdep

Author: Byron

lib/git/async/channel.py

@@ -21,61 +21,57 @@ class Channel(object):
 	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 RWChannel pairs.
+	for Rwriter pairs.
 	
 	Create a new channel """
-	__slots__ = tuple()
-
-
-class WChannel(Channel):
-	"""The write end of a channel - it is thread-safe"""
-	__slots__ = ('_queue')
+	__slots__ = 'queue'
 
 	# The queue to use to store the actual data
 	QueueCls = AsyncQueue
 
 	def __init__(self):
-		"""initialize this instance, able to hold max_items at once
-		Write calls will block if the channel is full, until someone reads from it"""
-		self._queue = self.QueueCls()
-	
-	#{ Interface 
-	def write(self, item, block=True, timeout=None):
-		"""Send an item into the channel, it can be read from the read end of the 
-		channel accordingly
-		:param item: Item to send
-		:param block: If True, the call will block until there is free space in the 
-			channel
-		:param timeout: timeout in seconds for blocking calls.
-		:raise ReadOnly: when writing into closed channel"""
-		# let the queue handle the 'closed' attribute, we write much more often 
-		# to an open channel than to a closed one, saving a few cycles
-		self._queue.put(item, block, timeout)
-		
+		"""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):
+	"""The write end of a channel, a file-like interface for a channel"""
+	__slots__ = ('write', 'channel')
+	
+	def __init__(self, channel):
+		"""Initialize the writer to use the given channel"""
+		self.channel = channel
+		self.write = channel.queue.put
+	
+	#{ Interface
 	def size(self):
-		""":return: approximate number of items that could be read from the read-ends
-			of this channel"""
-		return self._queue.qsize()
+		return self.channel.queue.qsize()
 
 	def close(self):
 		"""Close the channel. Multiple close calls on a closed channel are no 
 		an error"""
-		self._queue.set_writable(False)
+		self.channel.queue.set_writable(False)
 
 	def closed(self):
 		""":return: True if the channel was closed"""
-		return not self._queue.writable()
+		return not self.channel.queue.writable()
 	#} END interface 
 
 
-class CallbackWChannel(WChannel):
+class CallbackWriter(Writer):
 	"""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__ = ('_pre_cb')
 
-	def __init__(self):
-		WChannel.__init__(self)
+	def __init__(self, channel):
+		Writer.__init__(self, channel)
 		self._pre_cb = None
+		self.write = self._write
 
 	def set_pre_cb(self, fun = lambda item: item):
 		"""Install a callback to be called before the given item is written.
@@ -88,25 +84,19 @@ def set_pre_cb(self, fun = lambda item: item):
 		self._pre_cb = fun
 		return prev
 
-	def write(self, item, block=True, timeout=None):
+	def _write(self, item, block=True, timeout=None):
 		if self._pre_cb:
 			item = self._pre_cb(item)
-		WChannel.write(self, item, block, timeout)
+		self.channel.queue.put(item, block, timeout)
 
-	
-class SerialWChannel(WChannel):
-	"""A slightly faster version of a WChannel, which sacrificed thead-safety for
-	performance"""
-	QueueCls = SyncQueue
-
 
-class RChannel(Channel):
-	"""The read-end of a corresponding write channel"""
-	__slots__ = '_wc'
+class Reader(object):
+	"""Allows reading from a channel"""
+	__slots__ = 'channel'
 
-	def __init__(self, wchannel):
+	def __init__(self, channel):
 		"""Initialize this instance from its parent write channel"""
-		self._wc = wchannel
+		self.channel = channel
 
 
 	#{ Interface
@@ -135,7 +125,7 @@ def read(self, count=0, block=True, timeout=None):
 
 		# in non-blocking mode, its all not a problem
 		out = list()
-		queue = self._wc._queue
+		queue = self.channel.queue
 		if not block:
 			# be as fast as possible in non-blocking mode, hence
 			# its a bit 'unrolled'
@@ -198,12 +188,12 @@ def read(self, count=0, block=True, timeout=None):
 
 	#} END interface 
 
-class CallbackRChannel(RChannel):
+class CallbackReader(Reader):
 	"""A channel which sends a callback before items are read from the channel"""
 	__slots__ = "_pre_cb"
 
-	def __init__(self, wc):
-		RChannel.__init__(self, wc)
+	def __init__(self, channel):
+		Reader.__init__(self, channel)
 		self._pre_cb = None
 
 	def set_pre_cb(self, fun = lambda count: None):
@@ -220,18 +210,20 @@ def set_pre_cb(self, fun = lambda count: None):
 	def read(self, count=0, block=True, timeout=None):
 		if self._pre_cb:
 			self._pre_cb(count)
-		return RChannel.read(self, count, block, timeout)
+		return Reader.read(self, count, block, timeout)
 
 
 #} END classes
 
 #{ Constructors
-def mkchannel(wctype = WChannel, rctype = RChannel):
-	"""Create a channel, which consists of one write end and one read end
-	:return: tuple(write_channel, read_channel)
+def mkchannel(ctype = Channel, wtype = Writer, rtype = Reader):
+	"""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"""
-	wc = wctype()
-	rc = rctype(wc)
+	c = ctype()
+	wc = wtype(c)
+	rc = rtype(c)
 	return wc, rc
 #} END constructors

lib/git/async/graph.py

@@ -25,14 +25,24 @@ class Graph(object):
 
 	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 del_node(self, node):
+	def remove_node(self, node):
 		"""Delete a node from the graph
 		:return: self"""
 		try:
@@ -46,6 +56,8 @@ def del_node(self, node):
 			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):

lib/git/async/pool.py

@@ -18,25 +18,25 @@
 from graph import Graph 
 from channel import (
 		mkchannel,
-		WChannel, 
-		SerialWChannel,
-		CallbackRChannel
+		Writer, 
+		Channel,
+		SerialChannel,
+		CallbackReader
 	)
 
 import sys
 import weakref
 from time import sleep
+import new
 
 
-class RPoolChannel(CallbackRChannel):
-	""" A read-only pool channel may not be wrapped or derived from, but it provides slots to call
-	before and after an item is to be read.
-	
+class PoolReader(CallbackReader):
+	"""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, wchannel, task, pool):
-		CallbackRChannel.__init__(self, wchannel)
+	def __init__(self, channel, task, pool):
+		CallbackReader.__init__(self, channel)
 		self._task_ref = weakref.ref(task)
 		self._pool_ref = weakref.ref(pool)
 
@@ -61,17 +61,16 @@ def __del__(self):
 		# 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
-		# TODO: Fix this - private/public method
 		if sys.getrefcount(self) < 6:
-			pool.remove_task(task, _from_destructor_=True)
+			pool.remove_task(task, _from_destructor_ = True)
 		# END handle refcount based removal of task
 
 	#{ Internal
 	def _read(self, count=0, block=True, timeout=None):
-		"""Direct read, bypassing the pool handling"""
-		return CallbackRChannel.read(self, count, block, timeout)
-	#} END internal
+		return CallbackReader.read(self, count, block, timeout)
 
+	#} END internal
+
 	#{ Interface
 
 	def pool_ref(self):
@@ -118,7 +117,7 @@ def read(self, count=0, block=True, timeout=None):
 		####### read data ########
 		##########################
 		# read actual items, tasks were setup to put their output into our channel ( as well )
-		items = CallbackRChannel.read(self, count, block, timeout)
+		items = CallbackReader.read(self, count, block, timeout)
 		##########################
 
 
@@ -262,21 +261,21 @@ def _prepare_channel_read(self, task, count):
 			# should make things execute faster. Putting the if statements 
 			# into the loop would be less code, but ... slower
 			# DEBUG
-			# print actual_count, numchunks, chunksize, remainder, task._out_wc.size()
+			# print actual_count, numchunks, chunksize, remainder, task._out_writer.size()
 			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
-				queue = self._queue
+				qput = self._queue.put
 				if numchunks > 1:
 					for i in xrange(numchunks):
-						queue.put((task.process, chunksize))
+						qput((task.process, chunksize))
 					# END for each chunk to put
 				else:
-					queue.put((task.process, chunksize))
+					qput((task.process, chunksize))
 				# END try efficient looping
 
 				if remainder:
-					queue.put((task.process, remainder))
+					qput((task.process, remainder))
 				# END handle chunksize
 			else:
 				# no workers, so we have to do the work ourselves
@@ -297,13 +296,13 @@ def _prepare_channel_read(self, task, count):
 
 	def _remove_task_if_orphaned(self, task, from_destructor):
 		"""Check the task, and delete it if it is orphaned"""
-		# 1 as its stored on the task, 1 for the getrefcount call
+		# 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.wchannel()) < max_ref_count:
+		if sys.getrefcount(task.writer().channel) < max_ref_count:
 			self.remove_task(task, from_destructor)
 	#} END internal
 
@@ -325,7 +324,6 @@ def set_size(self, size=0):
 			threadsafe to optimize item throughput.
 		
 		:note: currently NOT threadsafe !"""
-		print "set_size", size
 		assert size > -1, "Size cannot be negative"
 
 		# either start new threads, or kill existing ones.
@@ -375,7 +373,7 @@ def num_tasks(self):
 		finally:
 			self._taskgraph_lock.release()
 
-	def remove_task(self, task, _from_destructor_=False):
+	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 
@@ -403,7 +401,7 @@ def remove_task(self, task, _from_destructor_=False):
 
 			# keep its input nodes as we check whether they were orphaned
 			in_tasks = task.in_nodes
-			self._tasks.del_node(task)
+			self._tasks.remove_node(task)
 			self._taskorder_cache.clear()
 		finally:
 			self._taskgraph_lock.release()
@@ -421,7 +419,7 @@ def add_task(self, task):
 			the task will be considered orphaned and will be deleted on the next 
 			occasion."""
 		# create a write channel for it
-		wctype = WChannel
+		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
@@ -442,26 +440,29 @@ def add_task(self, task):
 
 			# Use a non-threadsafe queue
 			# This brings about 15% more performance, but sacrifices thread-safety
-			# when reading from multiple threads.
 			if self.size() == 0:
-				wctype = SerialWChannel
+				ctype = SerialChannel
 			# END improve locks
 
 			# setup the tasks channel - respect the task creators choice though
 			# if it is set.
-			wc = task.wchannel()
+			wc = task.writer()
+			ch = None
 			if wc is None:
-				wc = wctype()
+				ch = ctype()
+				wc = Writer(ch)
+				task.set_writer(wc)
+			else:
+				ch = wc.channel
 			# END create write channel ifunset
-			rc = RPoolChannel(wc, task, self)
-			task.set_wchannel(wc)
+			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, 'rchannel'):
-			ic = task.rchannel()
+		if hasattr(task, 'reader'):
+			ic = task.reader()
 			if hasattr(ic, 'pool_ref') and ic.pool_ref()() is self:
 				self._taskgraph_lock.acquire()
 				try: