merge mtexp to trunk - a more robust threaded shell implementation that works with macros, and does not hang IPython if worker thread dies

IPython/Extensions/ipy_bzr.py

@@ -8,7 +8,7 b' are at the bottom of the file, the rest is left untouched.'
 Must be loaded with ip.load('ipy_bzr') 
 
 """ 
-a 
+
 # Copyright (C) 2004, 2005 Aaron Bentley
 # <[e-mail]>
 #

IPython/Shell.py

@@ -358,18 +358,13 b' class MTInteractiveShell(InteractiveShell):'
         InteractiveShell.__init__(self,name,usage,rc,user_ns,
                                   user_global_ns,banner2)
 
-        # Locking control variable.
-        self.thread_ready = threading.Condition(threading.RLock())
 
-        # A queue to hold the code to be executed.  A scalar variable is NOT
-        # enough, because uses like macros cause reentrancy.
+        # A queue to hold the code to be executed. 
         self.code_queue = Queue.Queue()
 
         # Stuff to do at closing time
-        self._kill = False
-        on_kill = kw.get('on_kill')
-        if on_kill is None:
-            on_kill = []
+        self._kill = None
+        on_kill = kw.get('on_kill', [])
         # Check that all things to kill are callable:
         for t in on_kill:
             if not callable(t):
@@ -377,6 +372,7 b' class MTInteractiveShell(InteractiveShell):'
         self.on_kill = on_kill
         # thread identity of the "worker thread" (that may execute code directly)
         self.worker_ident = None
+        
     def runsource(self, source, filename="<input>", symbol="single"):
         """Compile and run some source in the interpreter.
 
@@ -390,6 +386,10 b' class MTInteractiveShell(InteractiveShell):'
             KBINT = False
             return False
         
+        if self._kill:
+            # can't queue new code if we are being killed
+            return True
+        
         try:
             code = self.compile(source, filename, symbol)
         except (OverflowError, SyntaxError, ValueError):
@@ -401,13 +401,6 b' class MTInteractiveShell(InteractiveShell):'
             # Case 2
             return True
 
-        # Case 3
-        # Store code in queue, so the execution thread can handle it.
-
-        # Note that with macros and other applications, we MAY re-enter this
-        # section, so we have to acquire the lock with non-blocking semantics,
-        # else we deadlock.
-        
         # shortcut - if we are in worker thread, or the worker thread is not running, 
         # execute directly (to allow recursion and prevent deadlock if code is run early 
         # in IPython construction)
@@ -416,12 +409,21 b' class MTInteractiveShell(InteractiveShell):'
             InteractiveShell.runcode(self,code)
             return
         
-        got_lock = self.thread_ready.acquire(blocking=False)
-        self.code_queue.put(code)
-        if got_lock:
-           self.thread_ready.wait()  # Wait until processed in timeout interval
-        self.thread_ready.release()
+        # Case 3
+        # Store code in queue, so the execution thread can handle it.
+
+        completed_ev, received_ev = threading.Event(), threading.Event() 
 
+        self.code_queue.put((code,completed_ev, received_ev))
+        # first make sure the message was received, with timeout
+        received_ev.wait(5)
+        if not received_ev.isSet():
+            # the mainloop is dead, start executing code directly
+            print "Warning: Timeout for mainloop thread exceeded"
+            print "switching to nonthreaded mode (until mainloop wakes up again)"
+            self.worker_ident = None
+        else:
+            completed_ev.wait()
         return False
 
     def runcode(self):
@@ -430,9 +432,9 b' class MTInteractiveShell(InteractiveShell):'
         Multithreaded wrapper around IPython's runcode()."""
 
         global CODE_RUN
-        # lock thread-protected stuff
+        
+        # we are in worker thread, stash out the id for runsource() 
         self.worker_ident = thread.get_ident()
-        got_lock = self.thread_ready.acquire()
 
         if self._kill:
             print >>Term.cout, 'Closing threads...',
@@ -440,6 +442,9 b' class MTInteractiveShell(InteractiveShell):'
             for tokill in self.on_kill:
                 tokill()
             print >>Term.cout, 'Done.'
+            # allow kill() to return
+            self._kill.set()
+            return True
 
         # Install sigint handler.  We do it every time to ensure that if user
         # code modifies it, we restore our own handling.
@@ -455,9 +460,11 b' class MTInteractiveShell(InteractiveShell):'
         code_to_run = None
         while 1:
             try:
-                code_to_run = self.code_queue.get_nowait()
+                code_to_run, completed_ev, received_ev = self.code_queue.get_nowait()                
             except Queue.Empty:
                 break
+            received_ev.set()
+            
             # Exceptions need to be raised differently depending on which
             # thread is active.  This convoluted try/except is only there to
             # protect against asynchronous exceptions, to ensure that a KBINT
@@ -471,28 +478,23 b' class MTInteractiveShell(InteractiveShell):'
                 except KeyboardInterrupt:
                    print "Keyboard interrupted in mainloop"
                    while not self.code_queue.empty():
-                      self.code_queue.get_nowait()
+                      code, ev1,ev2 = self.code_queue.get_nowait()
+                      ev1.set()
+                      ev2.set()                      
                    break
             finally:
-               if got_lock:
                   CODE_RUN = False
+                # allow runsource() return from wait
+                completed_ev.set()
 
-        # We're done with thread-protected variables
-        if code_to_run is not None:
-           self.thread_ready.notify()
-        self.thread_ready.release()
 
-        # We're done...
-        CODE_RUN = False
         # This MUST return true for gtk threading to work
         return True
 
     def kill(self):
         """Kill the thread, returning when it has been shut down."""
-        got_lock = self.thread_ready.acquire(False)
-        self._kill = True
-        if got_lock:
-            self.thread_ready.release()
+        self._kill = threading.Event()
+        self._kill.wait()
 
 class MatplotlibShellBase:
     """Mixin class to provide the necessary modifications to regular IPython