Add support for aborting the worker thread

git-svn-id: https://svn.wxwidgets.org/svn/wx/wxWidgets/trunk@41354 c3d73ce0-8a6f-49c7-b76d-6d57e0e08775

wxPython/demo/DelayedResult.py

@@ -56,7 +56,8 @@ class FrameSimpleDelayed(FrameSimpleDelayedBase):
     
     def __init__(self, *args, **kwargs):
         FrameSimpleDelayedBase.__init__(self, *args, **kwargs)
-        self.jobID = 1
+        self.jobID = 0
+        self.abortEvent = delayedresult.AbortEvent()
         self.Bind(wx.EVT_CLOSE, self.handleClose)
 
     def setLog(self, log):
@@ -67,54 +68,50 @@ class FrameSimpleDelayed(FrameSimpleDelayedBase):
         app, so worker thread continues and sends result to dead frame; normally
         your app would exit so this would not happen."""
         if self.buttonAbort.IsEnabled():
-            self.Hide()
+            self.log( "Exiting: Aborting job %s" % self.jobID )
-            wx.FutureCall(5000, self.Destroy)
+            self.abortEvent.set()
-        else:
         self.Destroy()
             
     def handleGet(self, event): 
         """Compute result in separate thread, doesn't affect GUI response."""
         self.buttonGet.Enable(False)
         self.buttonAbort.Enable(True)
+        self.abortEvent.clear()
+        self.jobID += 1
         
         self.log( "Starting job %s in producer thread: GUI remains responsive"
                   % self.jobID )
         delayedresult.startWorker(self._resultConsumer, self._resultProducer, 
-                                  wargs=(self.jobID,), jobID=self.jobID)
+                                  wargs=(self.jobID,self.abortEvent), jobID=self.jobID)
 
                         
-    def _resultProducer(self, jobID):
+    def _resultProducer(self, jobID, abortEvent):
         """Pretend to be a complex worker function or something that takes 
         long time to run due to network access etc. GUI will freeze if this 
         method is not called in separate thread."""
         import time
-        time.sleep(5)
+        count = 0
+        while not abortEvent() and count < 50:
+            time.sleep(0.1)
+            count += 1
         return jobID
 
 
     def handleAbort(self, event): 
-        """Abort actually just means 'ignore the result when it gets to 
+        """Abort the result computation."""
-        handler, it is no longer relevant'. We just increase the job ID, 
-        this will let handler know that the result has been cancelled."""
         self.log( "Aborting result for job %s" % self.jobID )
         self.buttonGet.Enable(True)
         self.buttonAbort.Enable(False)
-        self.jobID += 1
+        self.abortEvent.set()
 
         
     def _resultConsumer(self, delayedResult):
-        # See if we still want the result for last job started
         jobID = delayedResult.getJobID()
-        if jobID != self.jobID:
+        assert jobID == self.jobID
-            self.log( "Got obsolete result for job %s, ignored" % jobID )
-            return
-
-        # we do, get result:
         try:
             result = delayedResult.get()
         except Exception, exc:
             self.log( "Result for job %s raised exception: %s" % (jobID, exc) )
-            self.jobID += 1
             return
         
         # output result
@@ -124,7 +121,6 @@ class FrameSimpleDelayed(FrameSimpleDelayedBase):
         # get ready for next job:
         self.buttonGet.Enable(True)
         self.buttonAbort.Enable(False)
-        self.jobID += 1
 
 
 class FrameSimpleDirect(FrameSimpleDelayedBase):

wxPython/wx/lib/delayedresult.py

@@ -227,12 +227,20 @@ class DelayedResult:
         return self.__result
 
 
+class AbortedException(Exception):
+    """Raise this in your worker function so that the sender knows 
+    not to send a result to handler."""
+    pass
+    
+
 class Producer(threading.Thread):
     """
     Represent the worker thread that produces delayed results. 
     It causes the given function to run in a separate thread, 
     and a sender to be used to send the return value of the function.
     As with any threading.Thread, instantiate and call start().
+    Note that if the workerFn raises AbortedException, the result is not 
+    sent and the thread terminates gracefully.
     """
     
     def __init__(self, sender, workerFn, args=(), kwargs={}, 
@@ -250,6 +258,8 @@ class Producer(threading.Thread):
         def wrapper():
             try: 
                 result = workerFn(*args, **kwargs)
+            except AbortedException:
+                pass
             except Exception, exc:
                 extraInfo = self._extraInfo(exc)
                 sender.sendException(exc, extraInfo)
@@ -268,6 +278,33 @@ class Producer(threading.Thread):
         return None
 
 
+class AbortEvent:
+    """
+    Convenience class that represents a kind of threading.Event that
+    raises AbortedException when called (see the __call__ method, everything
+    else is just to make it look like threading.Event).
+    """
+    
+    def __init__(self):
+        self.__ev = threading.Event()
+
+    def __call__(self, timeout=None):
+        """See if event has been set (wait at most timeout if given).  If so, 
+        raise AbortedException. Otherwise return None. Allows you to do
+        'while not event():' which will always succeed unless the event 
+        has been set (then AbortedException will cause while to exit)."""
+        if timeout:
+            self.__ev.wait(timeout)
+        if self.__ev.isSet():
+            raise AbortedException()
+        return None
+    
+    def __getattr__(self, name):
+        """This allows us to be a kind of threading.Event."""
+        if name in ('set','clear','wait','isSet'):
+            return getattr(self.__ev, name)
+
+
 def startWorker(
     consumer, workerFn, 
     cargs=(), ckwargs={}, 
@@ -373,4 +410,3 @@ class PreProcessChain:
         handler = self.__chain[0]
         handler( chainTrav )
         
-