Plenty of fixes in the chunking routine, made possible by a serialized chunking test. Next up, actual async processing

Author: Byron

lib/git/async/graph.py

@@ -35,12 +35,17 @@ def add_node(self, node):
 	def del_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)])
-		del(self.nodes[self.nodes.index(node)]) 
 		return self
 
 	def add_edge(self, u, v):

lib/git/async/pool.py

@@ -117,36 +117,72 @@ def _queue_feeder_visitor(self, task, count):
 		queue all others for processing by our worker threads ( if available )."""
 		if task.error() or task.is_done():
 			self._consumed_tasks.append(task)
+			return True
+		# END stop processing
 
 		# allow min-count override. This makes sure we take at least min-count
 		# items off the input queue ( later )
-		if task.min_count is not None:
+		if task.min_count is not None and count != 0 and count < task.min_count:
 			count = task.min_count
 		# END handle min-count
 
 		# if the task does not have the required output on its queue, schedule
 		# it for processing. If we should process all, we don't care about the 
 		# amount as it should process until its all done.
-		if self._workers:
-			if count < 1 or task._out_wc.size() < count:
+		if count < 1 or task._out_wc.size() < count:
+			numchunks = 1
+			chunksize = 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 count > task.max_chunksize:
+				numchunks = count / task.max_chunksize
+				chunksize = task.max_chunksize
+				remainder = count - (numchunks * chunksize)
+			# END handle chunking
+			
+			print count, numchunks, chunksize, remainder
+			# 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._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
-				if task.max_chunksize:
-					chunksize = count / task.max_chunksize
-					remainder = count - (chunksize * task.max_chunksize)
-					for i in xrange(chunksize):
+				if numchunks > 1:
+					for i in xrange(numchunks):
 						queue.put((task.process, chunksize))
-					if remainder:
-						queue.put((task.process, remainder))
+					# END for each chunk to put
+				else:
+					queue.put((task.process, chunksize))
+				# END try efficient looping
+				
+				if remainder:
+					queue.put((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:
-					self._queue.put((task.process, count))
+					task.process(chunksize)
+				# END try efficient looping
+				
+				if remainder:
+					task.process(remainder)
 				# END handle chunksize
-			# END handle queuing
-		else:
-			# no workers, so we have to do the work ourselves
-			task.process(count)
-		# END handle serial mode 
+				
+				# as we are serial, we can check for consumption right away
+				if task.error() or task.is_done():
+					self._consumed_tasks.append(task)
+				# END handle consumption
+			# END handle serial mode
+		# END handle queuing 
 
 		# always walk the whole graph, we want to find consumed tasks
 		return True

lib/git/async/task.py

@@ -58,7 +58,7 @@ def process(self, count=0):
 
 		# 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
-		if len(items) != count:
+		if not items or len(items) != count:
 			self.set_done()
 		# END handle done state
 	#{ Configuration

test/git/async/test_graph.py

@@ -54,7 +54,8 @@ def test_base(self):
 
 		# deleting a connected node clears its neighbour connections
 		assert n3.in_nodes[0] is n2
-		g.del_node(n2)
+		assert g.del_node(n2) is g
+		assert g.del_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

test/git/async/test_pool.py

@@ -3,21 +3,22 @@
 from git.async.pool import *
 from git.async.task import *
 from git.async.util import cpu_count
-
+import threading
 import time
 
 class TestThreadTaskNode(InputIteratorThreadTask):
 	def __init__(self, *args, **kwargs):
 		super(TestThreadTaskNode, self).__init__(*args, **kwargs)
-		self.reset()
+		self.reset(self._iterator)
 
 	def do_fun(self, item):
 		self.item_count += 1
 		return item
 
-	def reset(self):
+	def reset(self, iterator):
 		self.process_count = 0
 		self.item_count = 0
+		self._iterator = iterator
 
 	def process(self, count=1):
 		super(TestThreadTaskNode, self).process(count)
@@ -36,6 +37,111 @@ class TestThreadPool(TestBase):
 
 	max_threads = cpu_count()
 
+	def _assert_sync_single_task(self, p):
+		"""Performs testing in a synchronized environment"""
+		null_tasks = p.num_tasks()		# in case we had some before
+		
+		# add a simple task
+		# it iterates n items
+		ni = 20
+		assert ni % 2 == 0, "ni needs to be dividable by 2"
+		
+		def make_iter():
+			return iter(range(ni))
+		# END utility
+		
+		task = TestThreadTaskNode(make_iter(), 'iterator', None)
+		task.fun = task.do_fun
+		
+		assert p.num_tasks() == null_tasks
+		rc = p.add_task(task)
+		assert p.num_tasks() == 1 + null_tasks
+		assert isinstance(rc, RPoolChannel)
+		assert task._out_wc is not None
+		
+		# pull the result completely - we should get one task, which calls its 
+		# function once. In serial mode, the order matches
+		items = rc.read()
+		task._assert(1, ni).reset(make_iter())
+		assert len(items) == ni
+		assert items[0] == 0 and items[-1] == ni-1
+		
+		# as the task is done, it should have been removed - we have read everything
+		assert task.is_done()
+		assert p.num_tasks() == null_tasks
+		
+		# pull individual items
+		rc = p.add_task(task)
+		assert p.num_tasks() == 1 + null_tasks
+		for i in range(ni):
+			items = rc.read(1)
+			assert len(items) == 1
+			assert i == items[0]
+		# END for each item
+		# it couldn't yet notice that the input is depleted as we pulled exaclty 
+		# ni items - the next one would remove it. Instead, we delete our channel 
+		# which triggers orphan handling
+		assert p.num_tasks() == 1 + null_tasks
+		del(rc)
+		assert p.num_tasks() == null_tasks
+		
+		task.reset(make_iter())
+		
+		# test min count
+		# if we query 1 item, it will prepare ni / 2
+		task.min_count = ni / 2
+		rc = p.add_task(task)
+		assert len(rc.read(1)) == 1			# 1
+		assert len(rc.read(1)) == 1
+		assert len(rc.read(ni-2)) == ni - 2	# rest - it has ni/2 - 2 on the queue, and pulls ni-2
+		task._assert(2, ni)						# two chunks, 20 calls ( all items )
+		assert p.num_tasks() == 1 + null_tasks	# it still doesn't know, didn't read too much
+		assert len(rc.read()) == 0				# now we read too much and its done
+		assert p.num_tasks() == null_tasks
+		
+		# test chunking
+		# we always want 4 chunks, these could go to individual nodes
+		task.reset(make_iter())
+		task.max_chunksize = ni / 4			# 4 chunks
+		rc = p.add_task(task)
+		# must read a specific item count
+		# count is still at ni / 2 - here we want more than that
+		assert len(rc.read(ni / 2 + 2)) == ni / 2 + 2		# make sure its uneven ;)
+		assert len(rc.read(ni / 2 - 2)) == ni / 2 - 2
+		
+		# END read chunks
+		task._assert(ni / 4, ni)			# read two times, got 4 processing steps
+		assert p.num_tasks() == null_tasks	# depleted
+		
+		# but this only hits if we want too many items, if we want less, it could 
+		# still do too much - hence we set the min_count to the same number to enforce
+		# at least ni / 4 items to be preocessed, no matter what we request
+		task.reset(make_iter())
+		task.min_count = None
+		rc = p.add_task(task)
+		for i in range(ni):
+			assert rc.read(1)[0] == i
+		# END pull individual items
+		# too many processing counts ;)
+		task._assert(ni, ni)
+		assert p.num_tasks() == 1 + null_tasks
+		assert p.del_task(task) is p		# del manually this time
+		assert p.num_tasks() == null_tasks
+		
+		# now with we set the minimum count to reduce the number of processing counts
+		task.reset(make_iter())
+		task.min_count = ni / 4
+		rc = p.add_task(task)
+		for i in range(ni):
+			assert rc.read(1)[0] == i
+		# END for each item
+		task._assert(ni / 4, ni)
+		del(rc)
+		assert p.num_tasks() == null_tasks
+		
+	def _assert_async_dependent_tasks(self, p):
+		pass
+	
 	def test_base(self):
 		p = ThreadPool()
 
@@ -50,30 +156,49 @@ def test_base(self):
 			p.set_size(i)
 			assert p.size() == i
 
-		# currently in serial mode !
+		# SINGLE TASK SERIAL SYNC MODE
+		##############################
+		# put a few unrelated tasks that we forget about
+		urc1 = p.add_task(TestThreadTaskNode(iter(list()), "nothing", None))
+		urc2 = p.add_task(TestThreadTaskNode(iter(list()), "nothing", None))
+		assert p.num_tasks() == 2
+		self._assert_sync_single_task(p)
+		assert p.num_tasks() == 2
+		del(urc1)
+		del(urc2)
+		assert p.num_tasks() == 0
 
-		# add a simple task
-		# it iterates n items
-		ni = 20
-		task = TestThreadTaskNode(iter(range(ni)), 'iterator', None)
-		task.fun = task.do_fun
 
-		assert p.num_tasks() == 0
-		rc = p.add_task(task)
-		assert p.num_tasks() == 1
-		assert isinstance(rc, RPoolChannel)
-		assert task._out_wc is not None
+		# DEPENDENT TASKS SERIAL
+		########################
+		self._assert_async_dependent_tasks(p)
+		
+		
+		# SINGLE TASK THREADED SYNC MODE
+		################################
+		# step one gear up - just one thread for now.
+		num_threads = len(threading.enumerate())
+		p.set_size(1)
+		assert len(threading.enumerate()) == num_threads + 1
+		# deleting the pool stops its threads - just to be sure ;)
+		del(p)
+		assert len(threading.enumerate()) == num_threads
+		
+		p = ThreadPool(1)
+		assert len(threading.enumerate()) == num_threads + 1
+		
+		# here we go
+		self._assert_sync_single_task(p)
+		
 
-		# pull the result completely - we should get one task, which calls its 
-		# function once. In serial mode, the order matches
-		items = rc.read()
-		task._assert(1, ni).reset()
-		assert len(items) == ni
-		assert items[0] == 0 and items[-1] == ni-1
 
+		# SINGLE TASK ASYNC MODE
+		########################
+		# two threads to compete for a single task
 
-		# switch to threaded mode - just one thread for now
 
-		# two threads to compete for tasks
+		# DEPENDENT TASK ASYNC MODE
+		###########################
+		# self._assert_async_dependent_tasks(p)