FreeViz: Offload work to a separate thread

Author: VesnaT

Orange/widgets/tests/base.py

@@ -1013,12 +1013,14 @@ def test_invalidated_embedding(self, timeout=DEFAULT_TIMEOUT):
         self.widget.graph.update_coordinates.assert_not_called()
         self.widget.graph.update_point_props.assert_called_once()
 
-    def test_saved_selection(self):
+    def test_saved_selection(self, timeout=DEFAULT_TIMEOUT):
         self.send_signal(self.widget.Inputs.data, self.data)
+        self.wait_until_stop_blocking()
         self.widget.graph.select_by_indices(list(range(0, len(self.data), 10)))
         settings = self.widget.settingsHandler.pack_data(self.widget)
         w = self.create_widget(self.widget.__class__, stored_settings=settings)
         self.send_signal(self.widget.Inputs.data, self.data, widget=w)
+        self.wait_until_stop_blocking(widget=w)
         self.assertEqual(np.sum(w.graph.selection), 15)
         np.testing.assert_equal(self.widget.graph.selection, w.graph.selection)
 

Orange/widgets/utils/concurrent.py

@@ -3,6 +3,7 @@
 """
 # TODO: Rename the module to something that does not conflict with stdlib
 # concurrent
+from typing import Tuple, Callable, Any
 import os
 import threading
 import atexit
@@ -11,18 +12,15 @@
 import weakref
 from functools import partial
 import concurrent.futures
-
 from concurrent.futures import Future, TimeoutError
 from contextlib import contextmanager
-from typing import Callable, Optional, Any, List
 
 from AnyQt.QtCore import (
     Qt, QObject, QMetaObject, QThreadPool, QThread, QRunnable, QSemaphore,
-    QEventLoop, QCoreApplication, QEvent, Q_ARG
+    QEventLoop, QCoreApplication, QEvent, Q_ARG,
+    pyqtSignal as Signal, pyqtSlot as Slot
 )
 
-from AnyQt.QtCore import pyqtSignal as Signal, pyqtSlot as Slot
-
 _log = logging.getLogger(__name__)
 
 
@@ -784,3 +782,177 @@ def __call__(self, *args):
         args = [Q_ARG(atype, arg) for atype, arg in zip(self.arg_types, args)]
         return QMetaObject.invokeMethod(
             self.obj, self.method, self.conntype, *args)
+
+
+class TaskState(QObject):
+
+    status_changed = Signal(str)
+    _p_status_changed = Signal(str)
+
+    progress_changed = Signal(float)
+    _p_progress_changed = Signal(float)
+
+    partial_result_ready = Signal(object)
+    _p_partial_result_ready = Signal(object)
+
+    stopped = Signal()
+
+    def __init__(self, *args):
+        super().__init__(*args)
+        self.__future = None
+        self.watcher = FutureWatcher()
+        self.__interruption_requested = False
+        self.__progress = 0
+        # Helpers to route the signal emits via a this object's queue.
+        # This ensures 'atomic' disconnect from signals for targets/slots
+        # in the same thread. Requires that the event loop is running in this
+        # object's thread.
+        self._p_status_changed.connect(
+            self.status_changed, Qt.QueuedConnection)
+        self._p_progress_changed.connect(
+            self.progress_changed, Qt.QueuedConnection)
+        self._p_partial_result_ready.connect(
+            self.partial_result_ready, Qt.QueuedConnection)
+
+    @property
+    def future(self):
+        # type: () -> Future
+        return self.__future
+
+    def set_status(self, text):
+        self._p_status_changed.emit(text)
+
+    def set_progress_value(self, value):
+        if round(value, 1) > round(self.__progress, 1):
+            # Only emit progress when it has changed sufficiently
+            self._p_progress_changed.emit(value)
+            self.__progress = value
+
+    def set_partial_results(self, value):
+        self._p_partial_result_ready.emit(value)
+
+    def is_interruption_requested(self):
+        return self.__interruption_requested
+
+    def start(self, executor, func=None):
+        # type: (concurrent.futures.Executor, Callable[[], Any]) -> Future
+        assert self.future is None
+        assert not self.__interruption_requested
+        self.__future = executor.submit(func)
+        self.watcher.setFuture(self.future)
+        return self.future
+
+    def cancel(self):
+        assert not self.__interruption_requested
+        self.__interruption_requested = True
+        if self.future is not None:
+            rval = self.future.cancel()
+        else:
+            # not even scheduled yet
+            rval = True
+        return rval
+
+    def stop(self):
+        self.stopped.emit()
+
+
+class InterruptRequested(BaseException):
+    pass
+
+
+class ConcurrentWidgetMixin:
+    def __init__(self):
+        self.__executor = concurrent.futures.ThreadPoolExecutor(max_workers=1)
+        self.__task = None  # type: Optional[TaskState]
+
+    @property
+    def task(self):
+        return self.__task
+
+    def _start_partial(self, state: TaskState) -> Callable[[], Any]:
+        raise NotImplementedError
+
+    def _set_partial_results(self, result: Tuple[str, Any]) -> None:
+        raise NotImplementedError
+
+    def __set_results(self, results: "Results"):
+        for key in results.__dict__:
+            value = getattr(results, key)
+            if value is not None:
+                setattr(self, key, value)
+
+    def __set_state_ready(self):
+        self.progressBarFinished()
+        self.setBlocking(False)
+        self.setStatusMessage("")
+
+    def __set_state_busy(self):
+        self.progressBarInit()
+        self.setBlocking(True)
+
+    def __start_task(self, task: Callable[[], Any], state: TaskState):
+        assert self.__task is None
+        state.status_changed.connect(self.setStatusMessage)
+        state.progress_changed.connect(self.progressBarSet)
+        state.partial_result_ready.connect(self._set_partial_results)
+        state.stopped.connect(self.canceled)
+        state.watcher.done.connect(self.finish)
+        state.start(self.__executor, task)
+        state.setParent(self)
+        self.__task = state
+
+    def __cancel_task(self, wait=True):
+        if self.__task is not None:
+            state, self.__task = self.__task, None
+            state.cancel()
+            state.partial_result_ready.disconnect(self._set_partial_results)
+            state.status_changed.disconnect(self.setStatusMessage)
+            state.progress_changed.disconnect(self.progressBarSet)
+            state.watcher.done.disconnect(self.finish)
+            if wait:
+                concurrent.futures.wait([state.future])
+                state.deleteLater()
+            else:
+                w = FutureWatcher(state.future, parent=state)
+                w.done.connect(state.deleteLater)
+
+    def start(self):
+        """ Call to start the task. """
+        self.__cancel_task(wait=False)
+
+        if self.data is None:
+            self.__set_state_ready()
+            return
+
+        state = TaskState(self)
+        task = self._start_partial(state)
+        self.__set_state_busy()
+        self.__start_task(task, state)
+
+    def finish(self, future: Future):
+        """ Invoked when task is done. """
+        assert future.done()
+        assert self.__task is not None
+        assert self.__task.future is future
+        assert self.__task.watcher.future() is future
+        self.__task, task = None, self.__task
+        task.deleteLater()
+        self.__set_state_ready()
+        result = future.result()
+        self.__set_results(result)
+
+    def cancel(self):
+        """ Call to stop the task. """
+        self.__cancel_task(wait=False)
+        self.__set_state_ready()
+
+    def canceled(self):
+        """ Invoked when task is stopped.  """
+        pass
+
+    def cancel_task_no_wait(self):
+        self.__cancel_task(wait=False)
+
+    def on_delete_widget(self):
+        self.__cancel_task(wait=True)
+        self.__executor.shutdown(True)