Added performance test, improved iterator task which will now be usable by default. It shows that there must be the notion of a producer, which can work if there are no items read

Author: Byron

lib/git/async/task.py

@@ -196,6 +196,9 @@ def __init__(self, iterator, *args, **kwargs):
 		self._read = lambda count: weakself().__read(count)
 		self._empty = False
 
+		# defaults to returning our items unchanged
+		self.fun = lambda item: item
+		
 	def __read(self, count=0):
 		"""Read count items from the iterator, and return them"""
 		# not threadsafe, but worst thing that could happen is that 

test/git/async/task.py

@@ -102,6 +102,14 @@ def do_fun(self, item):
 		# END handle tuple
 
 
+class TestThreadPerformanceTaskNode(InputChannelTask):
+	"""Applies no operation to the item, and does not lock, measuring
+	the actual throughput of the system"""
+	
+	def do_fun(self, item):
+		return item
+
+
 class TestThreadInputChannelVerifyTaskNode(_TestTaskBase, InputChannelTask):
 	"""An input channel task, which verifies the result of its input channels, 
 	should be last in the chain.
@@ -121,15 +129,16 @@ def do_fun(self, item):
 
 		return item
 
-	
 #{ Utilities
 
 def make_proxy_method(t):
 	"""required to prevent binding self into the method we call"""
 	wt = weakref.proxy(t)
 	return lambda item: wt.do_fun(item)
 
-def add_task_chain(p, ni, count=1, fail_setup=list(), feeder_channel=None, id_offset=0):
+def add_task_chain(p, ni, count=1, fail_setup=list(), feeder_channel=None, id_offset=0, 
+					feedercls=TestThreadTaskNode, transformercls=TestThreadInputChannelTaskNode, 
+					include_verifier=True):
 	"""Create a task chain of feeder, count transformers and order verifcator 
 	to the pool p, like t1 -> t2 -> t3
 	:param fail_setup: a list of pairs, task_id, fail_after, i.e. [(2, 20)] would 
@@ -145,7 +154,7 @@ def add_task_chain(p, ni, count=1, fail_setup=list(), feeder_channel=None, id_of
 	feeder = None
 	frc = feeder_channel
 	if feeder_channel is None:
-		feeder = make_iterator_task(ni)
+		feeder = make_iterator_task(ni, taskcls=feedercls)
 		frc = p.add_task(feeder)
 	# END handle specific feeder
 
@@ -154,7 +163,7 @@ def add_task_chain(p, ni, count=1, fail_setup=list(), feeder_channel=None, id_of
 
 	inrc = frc
 	for tc in xrange(count):
-		t = TestThreadInputChannelTaskNode(inrc, tc+id_offset, None)
+		t = transformercls(inrc, tc+id_offset, None)
 
 		t.fun = make_proxy_method(t)
 		#t.fun = t.do_fun
@@ -169,22 +178,25 @@ def add_task_chain(p, ni, count=1, fail_setup=list(), feeder_channel=None, id_of
 		tasks[1+id].fail_after = fail_after
 	# END setup failure 
 
-	verifier = TestThreadInputChannelVerifyTaskNode(inrc, 'verifier', None)
-	#verifier.fun = verifier.do_fun
-	verifier.fun = make_proxy_method(verifier)
-	vrc = p.add_task(verifier)
-	
-	
-	tasks.append(verifier)
-	rcs.append(vrc)
+	if include_verifier:
+		verifier = TestThreadInputChannelVerifyTaskNode(inrc, 'verifier', None)
+		#verifier.fun = verifier.do_fun
+		verifier.fun = make_proxy_method(verifier)
+		vrc = p.add_task(verifier)
+		
+		
+		tasks.append(verifier)
+		rcs.append(vrc)
+	# END handle include verifier
 	return tasks, rcs
 
 def make_iterator_task(ni, taskcls=TestThreadTaskNode, **kwargs):
 	""":return: task which yields ni items
 	:param taskcls: the actual iterator type to use
 	:param **kwargs: additional kwargs to be passed to the task"""
 	t = taskcls(iter(range(ni)), 'iterator', None, **kwargs)
-	t.fun = make_proxy_method(t)
+	if isinstance(t, _TestTaskBase):
+		t.fun = make_proxy_method(t)
 	return t
 
 #} END utilities

test/git/async/test_performance.py

@@ -0,0 +1,51 @@
+"""Channel testing"""
+from test.testlib import *
+from task import *
+
+from git.async.pool import *
+from git.async.thread import terminate_threads
+from git.async.util import cpu_count
+
+import time
+import sys
+
+
+
+class TestThreadPoolPerformance(TestBase):
+	
+	max_threads = cpu_count()
+	
+	def test_base(self):
+		# create a dependency network, and see how the performance changes
+		# when adjusting the amount of threads
+		pool = ThreadPool(0)
+		ni = 1000				# number of items to process
+		print self.max_threads
+		for num_threads in range(self.max_threads*2 + 1):
+			pool.set_size(num_threads)
+			for num_transformers in (1, 5, 10):
+				for read_mode in range(2):
+					ts, rcs = add_task_chain(pool, ni, count=num_transformers, 
+												feedercls=InputIteratorThreadTask, 
+												transformercls=TestThreadPerformanceTaskNode, 
+												include_verifier=False)
+					
+					mode_info = "read(0)"
+					if read_mode == 1:
+						mode_info = "read(1) * %i" % ni
+					# END mode info
+					fmt = "Threadcount=%%i: Produced %%i items using %s in %%i transformations in %%f s (%%f items / s)" % mode_info
+					reader = rcs[-1]
+					st = time.time()
+					if read_mode == 1:
+						for i in xrange(ni):
+							assert len(reader.read(1)) == 1
+						# END for each item to read
+					else:
+						assert len(reader.read(0)) == ni
+					# END handle read mode
+					elapsed = time.time() - st
+					print >> sys.stderr, fmt % (num_threads, ni, num_transformers, elapsed, ni / elapsed)
+				# END for each read-mode
+			# END for each amount of processors
+		# END for each thread count