The new channeldesign actually works, but it also shows that its located at the wrong spot. The channel is nothing more than an adapter allowing to read multiple items from a thread-safe queue, the queue itself though must be 'closable' for writing, or needs something like a writable flag.

Author: Byron

lib/git/async/channel.py

@@ -4,7 +4,11 @@
 	Full
 	)
 
-from util import AsyncQueue
+from util import (
+		AsyncQueue, 
+		DummyLock
+		)
+
 from time import time
 import sys
 
@@ -23,12 +27,9 @@ class Channel(object):
 
 	def __new__(cls, *args):
 		if cls is Channel:
-			max_items = 0
-			if len(args) == 1:
-				max_items = args[0]
-			if len(args) > 1:
-				raise ValueError("Specify not more than the number of items the channel should take")
-			wc = WChannel(max_items)
+			if len(args) > 0:
+				raise ValueError("Cannot take any arguments when creating a new channel")
+			wc = WChannel()
 			rc = RChannel(wc)
 			return wc, rc
 		# END constructor mode
@@ -39,11 +40,11 @@ class WChannel(Channel):
 	"""The write end of a channel"""
 	__slots__ = ('_closed', '_queue')
 
-	def __init__(self, max_items=0):
+	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._closed = False
-		self._queue = AsyncQueue(max_items)
+		self._queue = AsyncQueue()
 
 
 	#{ Interface 
@@ -74,7 +75,21 @@ def size(self):
 	def close(self):
 		"""Close the channel. Multiple close calls on a closed channel are no 
 		an error"""
+		mutex = self._queue.mutex
+		mutex.acquire()
+		# this is atomic already, due to the GIL - no need to get the queue's mutex
+		print "channel.close()"
 		self._closed = True
+		# now make sure that the people waiting for an item are released now
+		# As we it could be that some readers are already on their way to initiate
+		# a blocking get, we must make sure that locks never block before that happens
+		
+		# now we are the only one accessing the queue, so change it
+		self._queue.mutex = DummyLock() 
+		print self._queue.not_empty._waiters
+		self._queue.not_empty.notify_all()
+		print self._queue.not_empty._waiters
+		mutex.release()
 
 	@property
 	def closed(self):
@@ -134,58 +149,47 @@ def read(self, count=0, block=True, timeout=None):
 				pass
 			# END handle exceptions
 		else:
-			# if we have really bad timing, the source of the channel
-			# marks itself closed, but before setting it, the thread 
-			# switches to us. We read it, read False, and try to fetch
-			# something, and never return. The whole closed channel thing
-			# is not atomic ( of course )
-			# This is why we never block for long, to get a chance to recheck
-			# for closed channels.
-			# We blend this into the timeout of the user
-			ourtimeout = 0.25				# the smaller, the more responsive, but the slower 
-			wc = self._wc
-			timeout = (timeout is None and sys.maxint) or timeout		# make sure we can compute with it
-			assert timeout != 0.0, "shouldn't block if timeout is 0"	# okay safe 
-			if timeout and ourtimeout > timeout:
-				ourtimeout = timeout
-			# END setup timeout
-			
 			# to get everything into one loop, we set the count accordingly
 			if count == 0:
 				count = sys.maxint
 			# END handle count
 
+			endtime = sys.maxint		# allows timeout for whole operation
+			if timeout is not None:
+				endtime = time() + timeout
+			# could be improved by a separate: no-endtime branch, saving the time calls
 			for i in xrange(count):
-				have_timeout = False
-				st = time()
-				while True:
+				try:
+					print "about to read", i, count, block, timeout
+					out.append(queue.get(block, timeout))
+					print "got one"
+				except Empty:
+					pass
+				# END ignore empty
+				
+				# if we have been unblocked because the closed state changed 
+				# in the meanwhile, stop trying
+				# NOTE: must NOT cache _wc
+				if self._wc.closed:
+					# its racing time - all threads waiting for the queue
+					# are awake now, and we actually can't be sure its empty
+					# Hence we pop it empty without blocking, getting as much
+					# as we can. This effectively lets us race ( with mutexes )
+					# of the other threads.
+					print "stopped because it was closed"
 					try:
-						if wc.closed:
-							have_timeout = True
-							# its about the 'in the meanwhile' :) - get everything
-							# we can in non-blocking mode. This will raise
-							try:
-								while True:
-									out.append(queue.get(False))
-								# END until it raises Empty
-							except Empty:
-								break
-							# END finally, out of here
-						# END don't continue on closed channels
-						
-						# END abort reading if it was closed ( in the meanwhile )
-						out.append(queue.get(block, ourtimeout))
-						break	# breakout right away
+						while True:
+							out.append(queue.get(False))
+						# END pop it empty
 					except Empty:
-						if timeout - (time() - st) <= 0:
-							# hitting timeout
-							have_timeout = True
-							break
-						# END abort if the user wants no more time spent here
-					# END handle timeout
-				# END endless timer loop
-				if have_timeout:
+						pass
+					# END ignore emptyness, we have all
+					
 					break
+				# END handle cloased
+					
+					if time() >= endtime:
+						break
 				# END stop on timeout
 			# END for each item
 		# END handle blocking

lib/git/async/task.py

@@ -140,10 +140,10 @@ def process(self, count=0):
 		# If we appear to be the only one left with our output channel, and are 
 		# closed ( this could have been set in another thread as well ), make 
 		# sure to close the output channel.
-		# The count is: 1 = wc itself, 2 = first reader channel, and we have only 
-		# one, 3 is ours + x for every thread having its copy on the stack 
+		# The count is: 1 = wc itself, 2 = first reader channel, + x for every 
+		# thread having its copy on the stack 
 		# + 1 for the instance we provide to refcount
-		if self.is_done() and sys.getrefcount(self._out_wc) < 5:
+		if self.is_done() and sys.getrefcount(self._out_wc) < 4:
 			self.close()
 		# END handle channel closure
 	#{ Configuration

lib/git/async/util.py

@@ -73,21 +73,22 @@ def empty(self):
 class HSCondition(object):
 	"""An attempt to make conditions less blocking, which gains performance 
 	in return by sleeping less"""
-	__slots__ = ("acquire", "release", "_lock", '_waiters')
+	# __slots__ = ("acquire", "release", "_lock", '_waiters')
+	__slots__ = ("_lock", '_waiters')
 	delay = 0.00002		# reduces wait times, but increases overhead
 
 	def __init__(self, lock=None):
 		if lock is None:
 			lock = Lock()
 		self._lock = lock
-		self.acquire = lock.acquire
-		self.release = lock.release
+		#self.acquire = lock.acquire
+		#self.release = lock.release
 		self._waiters = list()
 
-	def __release(self):
+	def release(self):
 		return self._lock.release()
 
-	def __acquire(self, block=None):
+	def acquire(self, block=None):
 		if block is None:
 			self._lock.acquire()
 		else:
@@ -156,7 +157,7 @@ def notify_all(self):
 		self.notify(len(self._waiters))
 
 
-class AsyncQueue(Queue):
+class _AsyncQueue(Queue):
 	"""A queue using different condition objects to gain multithreading performance"""
 	def __init__(self, maxsize=0):
 		Queue.__init__(self, maxsize)
@@ -166,7 +167,7 @@ def __init__(self, maxsize=0):
 		self.all_tasks_done = HSCondition(self.mutex)
 
 
-class _AsyncQueue(Queue):
+class AsyncQueue(Queue):
 	"""A queue using different condition objects to gain multithreading performance"""
 	__slots__ = ('mutex', 'not_empty', 'queue')
 
@@ -194,9 +195,9 @@ def put(self, item, block=True, timeout=None):
 		self.queue.append(item)
 		self.mutex.release()
 		self.not_empty.notify()
-
+		
 	def get(self, block=True, timeout=None):
-		self.not_empty.acquire()
+		self.not_empty.acquire()	# == self.mutex.acquire in that case
 		q = self.queue
 		try:
 			if not block:
@@ -205,16 +206,23 @@ def get(self, block=True, timeout=None):
 			elif timeout is None:
 				while not len(q):
 					self.not_empty.wait()
-			elif timeout < 0:
-				raise ValueError("'timeout' must be a positive number")
 			else:
+				print "with timeout", timeout
+				import traceback
+				traceback.print_stack()
 				endtime = _time() + timeout
 				while not len(q):
 					remaining = endtime - _time()
 					if remaining <= 0.0:
 						raise Empty
 					self.not_empty.wait(remaining)
-			return q.popleft()
+			# END handle block
+			# can happen if someone else woke us up
+			try:
+				return q.popleft()
+			except IndexError:
+				raise Empty
+			# END handle unblocking reason
 		finally:
 			self.not_empty.release()
 

test/git/async/test_channel.py

@@ -43,26 +43,4 @@ def test_base(self):
 
 		# reading from a closed channel never blocks
 		assert len(rc.read()) == 0
-		
-		
-		
-		# TEST LIMITED SIZE CHANNEL
-		# channel with max-items set
-		wc, rc = Channel(1)
-		wc.write(item)			# fine
-		
-		# blocks for a a moment, its full
-		st = time.time()
-		self.failUnlessRaises(EOFError, wc.write, item, True, to)
-		assert time.time() - st >= to
-		
-		# get our only one
-		assert rc.read(1)[0] == item
-		
-		# its empty,can put one again
-		wc.write(item2)
-		wc.close()
-		
-		# reading 10 will only yield one, it will not block as its closed
-		assert rc.read(10, timeout=1)[0] == item2
-		
+				

test/git/async/test_pool.py

@@ -92,6 +92,7 @@ def make_iter():
 
 		# pull the result completely - we should get one task, which calls its 
 		# function once. In sync mode, the order matches
+		print "read(0)"
 		items = rc.read()
 		assert len(items) == ni
 		task._assert(1, ni).reset(make_iter())
@@ -105,6 +106,7 @@ def make_iter():
 		rc = p.add_task(task)
 		assert p.num_tasks() == 1 + null_tasks
 		st = time.time()
+		print "read(1) * %i" % ni
 		for i in range(ni):
 			items = rc.read(1)
 			assert len(items) == 1
@@ -129,20 +131,24 @@ def make_iter():
 		# if we query 1 item, it will prepare ni / 2
 		task.min_count = ni / 2
 		rc = p.add_task(task)
+		print "read(1)"
 		items = rc.read(1)
 		assert len(items) == 1 and items[0] == 0			# processes ni / 2
+		print "read(1)"
 		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
 		nri = ni-2
+		print "read(%i)" % nri
 		items = rc.read(nri)
 		assert len(items) == nri
 		assert p.num_tasks() == null_tasks
 		task._assert(2, ni)						# two chunks, ni calls
 
 		# its already done, gives us no more
+		print "read(0) on closed"
 		assert len(rc.read()) == 0
 
 		# test chunking
@@ -154,11 +160,13 @@ def make_iter():
 		# count is still at ni / 2 - here we want more than that
 		# 2 steps with n / 4 items, + 1 step with n/4 items to get + 2
 		nri = ni / 2 + 2
+		print "read(%i)" % nri
 		items = rc.read(nri)
 		assert len(items) == nri
 		# have n / 4 - 2 items on queue, want n / 4 in first chunk, cause 1 processing
 		# ( 4 in total ). Still want n / 4 - 2 in second chunk, causing another processing
 		nri = ni / 2 - 2
+		print "read(%i)" % nri
 		items = rc.read(nri)
 		assert len(items) == nri
 
@@ -172,6 +180,7 @@ def make_iter():
 		task.min_count = None
 		rc = p.add_task(task)
 		st = time.time()
+		print "read(1) * %i, chunksize set" % ni
 		for i in range(ni):
 			if async:
 				assert len(rc.read(1)) == 1
@@ -192,6 +201,7 @@ def make_iter():
 		task.reset(make_iter())
 		task.min_count = ni / 4
 		rc = p.add_task(task)
+		print "read(1) * %i, min_count%i + chunksize" % (ni, task.min_count)
 		for i in range(ni):
 			if async:
 				assert len(rc.read(1)) == 1
@@ -208,10 +218,13 @@ def make_iter():
 		task.reset(make_iter())
 		task.should_fail = True
 		rc = p.add_task(task)
+		print "read(0) with failure"
 		assert len(rc.read()) == 0		# failure on first item
+		print "done with everything"
 		assert isinstance(task.error(), AssertionError)
 		assert p.num_tasks() == null_tasks
 
+		
 	def _assert_async_dependent_tasks(self, p):
 		# includes failure in center task, 'recursive' orphan cleanup
 		# This will also verify that the channel-close mechanism works