Added dependency-task tests, and fixed plenty of ref-count related bugs, as well as concurrency issues. Now it works okay, but the thread-shutdown is still an issue, as it causes incorrect behaviour making the tests fail. Its good, as it hints at additional issues that need to be solved. There is just a little more left on the feature side, but its nearly there

Byron · Byron · commit fbe062bf6dac · 2010-06-10T23:55:50.000+02:00
diff --git a/lib/git/async/graph.py b/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):
diff --git a/lib/git/async/pool.py b/lib/git/async/pool.py
@@ -402,7 +402,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()
diff --git a/lib/git/async/task.py b/lib/git/async/task.py
@@ -82,7 +82,8 @@ def error(self):
 
 	def process(self, count=0):
 		"""Process count items and send the result individually to the output channel"""
-		# first thing: increment the writer count
+		# 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()
@@ -191,7 +192,11 @@ def __init__(self, iterator, *args, **kwargs):
 			raise ValueError("Iterator %r needs a next() function" % iterator)
 		self._iterator = iterator
 		self._lock = self.lock_type()
-		self._read = self.__read
+		
+		# this is necessary to prevent a cyclic ref, preventing us from 
+		# getting deleted ( and collected )
+		weakself = weakref.ref(self)
+		self._read = lambda count: weakself().__read(count)
 		self._empty = False
 		
 	def __read(self, count=0):
@@ -201,6 +206,7 @@ def __read(self, count=0):
 		if self._empty:
 			return list()
 		# END early abort
+		
 		self._lock.acquire()
 		try:
 			if count == 0:
diff --git a/lib/git/async/thread.py b/lib/git/async/thread.py
@@ -116,7 +116,7 @@ class WorkerThread(TerminatableThread):
 	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', '_current_routine')
+	__slots__ = ('inq')
 	
 	
 	# define how often we should check for a shutdown request in case our 
@@ -128,7 +128,6 @@ def __init__(self, inq = None):
 		self.inq = inq
 		if inq is None:
 			self.inq = Queue.Queue()
-		self._current_routine = None				# routine we execute right now
 	
 	@classmethod
 	def stop(cls, *args):
@@ -141,7 +140,6 @@ def run(self):
 		
 		gettask = self.inq.get
 		while True:
-			self._current_routine = None
 			if self._should_terminate():
 				break
 			# END check for stop request
@@ -153,35 +151,38 @@ def run(self):
 			assert len(tasktuple) == 2, "Need tuple of function, arg - it could be more flexible, but its reduced to what we need"
 			routine, arg = tasktuple
 			
-			self._current_routine = routine
-			
 			try:
-				rval = None
-				if inspect.ismethod(routine):
-					if routine.im_self is None:
-						rval = routine(self, arg)
-					else:
+				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)
-				elif inspect.isroutine(routine):
-					rval = routine(arg)
-				else:
-					# ignore unknown items
-					print >> sys.stderr, "%s: task %s was not understood - terminating" % (self.getName(), str(tasktuple))
-					break
-				# END make routine call
+					else:
+						# ignore unknown items
+						print >> sys.stderr, "%s: task %s was not understood - terminating" % (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:
 				print self.name, "stops processing"	# DEBUG
 				break
 			except Exception,e:
 				print >> sys.stderr, "%s: Task %s raised unhandled exception: %s - this really shouldn't happen !" % (self.getName(), str(tasktuple), str(e))
 				continue	# just continue 
 			# END routine exception handling
+		
+			# END handle routine release
 		# END endless loop
 	
-	def routine(self):
-		""":return: routine we are currently executing, or None if we have no task"""
-		return self._current_routine
-	
 	def stop_and_join(self):
 		"""Send stop message to ourselves"""
 		self.inq.put((self.stop, None))
diff --git a/test/git/async/test_graph.py b/test/git/async/test_graph.py
@@ -3,6 +3,7 @@
 from git.async.graph import *
 
 import time
+import sys
 
 class TestGraph(TestBase):
 	
@@ -19,7 +20,7 @@ def test_base(self):
 		
 		# delete unconnected nodes
 		for n in g.nodes[:]:
-			g.del_node(n)
+			g.remove_node(n)
 		# END del nodes
 		
 		# add a chain of connected nodes
@@ -54,8 +55,8 @@ def test_base(self):
 			
 		# deleting a connected node clears its neighbour connections
 		assert n3.in_nodes[0] is n2
-		assert g.del_node(n2) is g
-		assert g.del_node(n2) is g					# multi-deletion okay
+		assert g.remove_node(n2) is g
+		assert g.remove_node(n2) is g					# multi-deletion okay
 		assert len(g.nodes) == nn - 1
 		assert len(n3.in_nodes) == 0
 		assert len(n1.out_nodes) == 0
@@ -68,3 +69,12 @@ def test_base(self):
 		assert dfirst_nodes[-1] == end and dfirst_nodes[-2].id == end.id-1
 		
 		
+		# test cleanup
+		# its at least kept by its graph
+		assert sys.getrefcount(end) > 3
+		del(g)
+		del(n1); del(n2); del(n3)
+		del(dfirst_nodes)
+		del(last)
+		del(n)
+		assert sys.getrefcount(end) == 2
diff --git a/test/git/async/test_pool.py b/test/git/async/test_pool.py
