improved testing to test the actual async handling of the pool. there are still inconsistencies that need to be fixed, but it already improved, especially the 4-thread performance which now is as fast as the dual-threaded performance

Author: Byron

lib/git/async/pool.py

@@ -67,20 +67,32 @@ def read(self, count=0, block=True, timeout=None):
 		# if we have count items, don't do any queue preparation - if someone
 		# depletes the queue in the meanwhile, the channel will close and 
 		# we will unblock naturally
+		# PROBLEM: If there are multiple consumer of this channel, we might 
+		# run out of items without being replenished == block forever in the 
+		# worst case. task.min_count could have triggered to produce more ...
+		# usually per read with n items, we put n items on to the queue, 
+		# so we wouldn't check this
+		# Even if we have just one consumer ( we could determine that with 
+		# the reference count ), it could be that in one moment we don't yet
+		# have an item, but its currently being produced by some worker.
+		# This is why we: 
+		# * make no assumptions if there are multiple consumers
+		# * 
 		have_enough = False
 		if count > 0:
-			# explicitly > count, as we want a certain safe range
-			have_enough = self._wc._queue.qsize() > count
+			have_enough = self._wc._queue.qsize() >= count
 		# END risky game
 
 		########## prepare ##############################
 		if not have_enough:
 			self._pool._prepare_channel_read(self._task, count)
 
 
-		######### read data ######
+		####### read data ########
+		##########################
 		# read actual items, tasks were setup to put their output into our channel ( as well )
 		items = RChannel.read(self, count, block, timeout)
+		##########################
 
 		if self._post_cb:
 			items = self._post_cb(items)

lib/git/async/task.py

@@ -178,6 +178,17 @@ def process(self, count=1):
 		# make sure we don't trigger the pool if we read from a pool channel which 
 		# belongs to our own pool. Channels from different pools are fine though, 
 		# there we want to trigger its computation
+		# PROBLEM: if the user keeps an end, but decides to put the same end into
+		# a task of this pool, then all items might deplete without new ones being 
+		# produced, causing a deadlock. Just triggering the pool would be better, 
+		# but cost's more, unnecessarily if there is just one consumer, which is 
+		# the user.
+		# * could encode usage in the channel type, and fail if the refcount on 
+		#   the read-pool channel is too high
+		# * maybe keep track of the elements that are requested or in-production 
+		#  for each task, which would allow to precisely determine whether 
+		#  the pool as to be triggered, and bail out early. Problem would 
+		#	be the 
 		if isinstance(self._in_rc, RPoolChannel) and self._in_rc._pool is self._pool_ref():
 			self._read = self._in_rc._read
 

lib/git/async/util.py

@@ -59,7 +59,7 @@ def empty(self):
 class HSCondition(_Condition):
 	"""An attempt to make conditions less blocking, which gains performance 
 	in return by sleeping less"""
-	delay = 0.00002		# reduces wait times, but increases overhead
+	delay = 0.00005		# reduces wait times, but increases overhead
 
 	def wait(self, timeout=None):
 		waiter = Lock()
@@ -85,7 +85,9 @@ def wait(self, timeout=None):
 					remaining = endtime - _time()
 					if remaining <= 0:
 						break
-					delay = min(delay * 2, remaining, .05)
+					# 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:

test/git/async/test_pool.py

@@ -12,9 +12,13 @@ def __init__(self, *args, **kwargs):
 		super(TestThreadTaskNode, self).__init__(*args, **kwargs)
 		self.reset(self._iterator)
 		self.should_fail = False
+		self.lock = threading.Lock()		# yes, can't safely do x = x + 1 :)
+		self.plock = threading.Lock()
 
 	def do_fun(self, item):
+		self.lock.acquire()
 		self.item_count += 1
+		self.lock.release()
 		if self.should_fail:
 			raise AssertionError("I am failing just for the fun of it")
 		return item
@@ -25,14 +29,26 @@ def reset(self, iterator):
 		self._iterator = iterator
 
 	def process(self, count=1):
-		super(TestThreadTaskNode, self).process(count)
+		# must do it first, otherwise we might read and check results before
+		# the thread gets here :). Its a lesson !
+		self.plock.acquire()
 		self.process_count += 1
+		self.plock.release()
+		super(TestThreadTaskNode, self).process(count)
 
 	def _assert(self, pc, fc):
 		"""Assert for num process counts (pc) and num function counts (fc)
 		:return: self"""
+		self.plock.acquire()
+		if self.process_count != pc:
+			print self.process_count, pc
 		assert self.process_count == pc
+		self.plock.release()
+		self.lock.acquire()
+		if self.item_count != fc:
+			print self.item_count, fc
 		assert self.item_count == fc
+		self.lock.release()
 		assert not self.error()
 		return self
 
@@ -103,15 +119,17 @@ def make_iter():
 		# 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			# processes ni / 2
-		assert len(rc.read(1)) == 1			# processes nothing
+		items = rc.read(1)
+		assert len(items) == 1 and items[0] == 0			# processes ni / 2
+		items = rc.read(1)
+		assert len(items) == 1 and items[0] == 1			# processes nothing
 		# rest - it has ni/2 - 2 on the queue, and pulls ni-2
 		# It wants too much, so the task realizes its done. The task
 		# doesn't care about the items in its output channel
 		items = rc.read(ni-2)
 		assert len(items) == ni - 2
 		assert p.num_tasks() == null_tasks
-		task._assert(2, ni)						# two chunks, 20 calls ( all items )
+		task._assert(2, ni)						# two chunks, ni calls
 
 		# its already done, gives us no more
 		assert len(rc.read()) == 0
@@ -246,7 +264,8 @@ def test_base(self):
 		p.set_size(2)
 		self._assert_single_task(p, True)
 
-		# kill it
+		# real stress test-  should be native on every dual-core cpu with 2 hardware
+		# threads per core
 		p.set_size(4)
 		self._assert_single_task(p, True)