Improved shutdown handling - although its impossible to prevent some stderr printing thanks to the underlying threading implementation, we can at least make sure that the interpreter doesn't block during shutdown. Now it appears to be running smoothly

Byron · Byron · commit 1873db442dc7 · 2010-06-11T16:29:55.000+02:00
diff --git a/lib/git/async/__init__.py b/lib/git/async/__init__.py
@@ -1 +1,30 @@
 """Initialize the multi-processing package"""
+
+#{ Initialization
+def _init_atexit():
+	"""Setup an at-exit job to be sure our workers are shutdown correctly before
+	the interpreter quits"""
+	import atexit
+	import thread
+	atexit.register(thread.do_terminate_threads)
+	
+def _init_signals():
+	"""Assure we shutdown our threads correctly when being interrupted"""
+	import signal
+	import thread
+	
+	prev_handler = signal.getsignal(signal.SIGINT)
+	def thread_interrupt_handler(signum, frame):
+		thread.do_terminate_threads()
+		if callable(prev_handler):
+			prev_handler(signum, frame)
+			raise KeyboardInterrupt()
+		# END call previous handler
+	# END signal handler
+	signal.signal(signal.SIGINT, thread_interrupt_handler)
+
+
+#} END init
+
+_init_atexit()
+_init_signals()
diff --git a/lib/git/async/thread.py b/lib/git/async/thread.py
@@ -136,16 +136,21 @@ def stop(cls, *args):
 	
 	def run(self):
 		"""Process input tasks until we receive the quit signal"""
-		print self.name, "starts processing"	# DEBUG
-		
 		gettask = self.inq.get
 		while True:
 			if self._should_terminate():
 				break
 			# END check for stop request
 			
-			# we wait and block - to terminate, send the 'stop' method
+			# note: during shutdown, this turns None in the middle of waiting 
+			# for an item to be put onto it - we can't du anything about it - 
+			# even if we catch everything and break gracefully, the parent 
+			# call will think we failed with an empty exception.
+			# Hence we just don't do anything about it. Alternatively
+			# we could override the start method to get our own bootstrapping, 
+			# which would mean repeating plenty of code in of the threading module.
 			tasktuple = gettask()
+				
 			# needing exactly one function, and one arg
 			routine, arg = tasktuple
 			
@@ -161,7 +166,7 @@ def run(self):
 						rval = routine(arg)
 					else:
 						# ignore unknown items
-						print >> sys.stderr, "%s: task %s was not understood - terminating" % (self.getName(), str(tasktuple))
+						sys.stderr.write("%s: task %s was not understood - terminating\n" % (self.getName(), str(tasktuple)))
 						break
 					# END make routine call
 				finally:
@@ -171,18 +176,22 @@ def run(self):
 					del(routine)
 					del(tasktuple)
 			except StopProcessing:
-				print self.name, "stops processing"	# DEBUG
 				break
 			except Exception,e:
-				print >> sys.stderr, "%s: Task %s raised unhandled exception: %s - this really shouldn't happen !" % (self.getName(), str(tasktuple), str(e))
+				sys.stderr.write("%s: Task %s raised unhandled exception: %s - this really shouldn't happen !\n" % (self.getName(), str(tasktuple), str(e)))
 				continue	# just continue 
 			# END routine exception handling
 		
 			# END handle routine release
 		# END endless loop
 	
 	def stop_and_join(self):
-		"""Send stop message to ourselves"""
+		"""Send stop message to ourselves - we don't block, the thread will terminate 
+		once it has finished processing its input queue to receive our termination
+		event"""
+		# DONT call superclass as it will try to join - join's don't work for 
+		# some reason, as python apparently doesn't switch threads (so often)
+		# while waiting ... I don't know, but the threads respond properly, 
+		# but only if dear python switches to them
 		self.inq.put((self.stop, None))
-		super(WorkerThread, self).stop_and_join()
 #} END classes
diff --git a/test/git/async/test_pool.py b/test/git/async/test_pool.py
@@ -373,10 +373,7 @@ def _assert_async_dependent_tasks(self, pool):
 		
 	
 	
-	
-	# for some reason, sometimes it has multiple workerthreads already when he 
-	# enters the method ... dunno yet, pools should clean up themselvess
-	#@terminate_threads
+	@terminate_threads
 	def test_base(self):
 		assert len(threading.enumerate()) == 1
 		
@@ -463,10 +460,11 @@ def test_base(self):
 		# threads per core
 		p.set_size(4)
 		self._assert_single_task(p, True)
+		
+		
 		# DEPENDENT TASK ASYNC MODE
 		###########################
 		self._assert_async_dependent_tasks(p)
 		
 		print >> sys.stderr, "Done with everything"
 		
-		# TODO: test multi-pool connections 
