HSCondition: Fixed terrible bug which it inherited from its default python Condition implementation, related to the notify method not being treadsafe. Although I was aware of it, I missed the first check which tests for the size - the result could be incorrect if the whole method wasn't locked.

Testing runs stable now, allowing to move on \!

3 files changed, 21 insertions, 35 deletions

diff --git a/lib/git/async/pool.py b/lib/git/async/pool.py
index 7ed6fd8e..66a2a105 100644
--- a/lib/git/async/pool.py
+++ b/lib/git/async/pool.py
@@ -58,14 +58,17 @@ class RPoolChannel(RChannel):
 	
 	def set_pre_cb(self, fun = lambda count: None):
 		"""Install a callback to call with the item count to be read before any 
-		item is actually  read from the channel.
+		item is actually  read from the channel. The call must be threadsafe if
+		the channel is passed to more than one tasks.
 		If it fails, the read will fail with an IOError
 		If a function is not provided, the call is effectively uninstalled."""
 		self._pre_cb = fun
 	
 	def set_post_cb(self, fun = lambda item: item):
 		"""Install a callback to call after the items were read. The function
-		returns a possibly changed item list. If it raises, the exception will be propagated.
+		returns a possibly changed item list.The call must be threadsafe if
+		the channel is passed to more than one tasks. 
+		If it raises, the exception will be propagated.
 		If a function is not provided, the call is effectively uninstalled."""
 		self._post_cb = fun
 	
diff --git a/lib/git/async/task.py b/lib/git/async/task.py
index dd2bd351..d18cedca 100644
--- a/lib/git/async/task.py
+++ b/lib/git/async/task.py
@@ -138,7 +138,7 @@ class OutputChannelTask(Node):
 			# just the right thing to do of course - one loose link in the chain ...
 			# Other chunks of our kind currently being processed will then 
 			# fail to write to the channel and fail as well
-			# self.close()
+			self.close()
 			
 			# If some other chunk of our Task had an error, the channel will be closed
 			# This is not an issue, just be sure we don't overwrite the original 
diff --git a/lib/git/async/util.py b/lib/git/async/util.py
index 008e60a3..2f46d55f 100644
--- a/lib/git/async/util.py
+++ b/lib/git/async/util.py
@@ -133,45 +133,28 @@ class HSCondition(deque):
 		# END assure release lock
 			
 	def notify(self, n=1):
-		"""Its vital that this method is threadsafe - to be fast we don'd get a lock, 
-		but instead rely on pseudo-atomic operations that come with the GIL.
-		Hence we use pop in the n=1 case to be truly atomic.
-		In the multi-notify case, we acquire a lock just for safety, as otherwise
-		we might pop too much of someone else notifies n waiters as well, which 
-		would in the worst case lead to double-releases of locks."""
-		if not self:
-			return
-		if n == 1:
-			# so here we assume this is thead-safe ! It wouldn't be in any other
-			# language, but python it is.
-			# But ... its two objects here - first the popleft, then the relasecall.
-			# If the timing is really really bad, and that happens if you let it 
-			# run often enough ( its a matter of statistics ), this will fail, 
-			# which is why we lock it.
-			# And yes, this causes some slow down, as single notifications happen
-			# alot
-			self._lock.acquire()
-			try:
+		"""Its vital that this method is threadsafe - we absolutely have to 
+		get a lock at the beginning of this method to be sure we get the 
+		correct amount of waiters back. If we bail out, although a waiter
+		is about to be added, it will miss its wakeup notification, and block
+		forever (possibly)"""
+		self._lock.acquire()
+		try:
+			if not self:	# len(self) == 0, but this should be faster
+				return
+			if n == 1:
 				try:
 					self.popleft().release()
 				except IndexError:
 					pass
-			finally:
-				self._lock.release()
-			# END assure lock is released
-		else:
-			self._lock.acquire()
-			# once the waiter resumes, he will want to acquire the lock
-			# and waits again, but only until we are done, which is important
-			# to do that in a thread-safe fashion
-			try:
+			else:
 				for i in range(min(n, len(self))):
 					self.popleft().release()
 				# END for each waiter to resume
-			finally:
-				self._lock.release()
-			# END assure we release our lock
-		# END handle n = 1 case faster
+			# END handle n = 1 case faster
+		finally:
+			self._lock.release()
+		# END assure lock is released
 	
 	def notify_all(self):
 		self.notify(len(self))