FreeViz: Offload work to a separate thread

Author: VesnaT

Orange/projection/base.py

@@ -1,3 +1,4 @@
+import copy
 import inspect
 import threading
 
@@ -135,6 +136,7 @@ def proj_variable(i, name):
 
         super().__init__(proj=proj)
         self.orig_domain = domain
+        self.n_components = n_components
         var_names = self._get_var_names(n_components)
         self.domain = Orange.data.Domain(
             [proj_variable(i, var_names[i]) for i in range(n_components)],
@@ -145,6 +147,13 @@ def _get_var_names(self, n):
         names = [f"{self.var_prefix}-{postfix}" for postfix in postfixes]
         return get_unique_names(self.orig_domain, names)
 
+    def copy(self):
+        proj = copy.deepcopy(self.proj)
+        model = type(self)(proj, self.domain.copy(), self.n_components)
+        model.pre_domain = self.pre_domain.copy()
+        model.name = self.name
+        return model
+
 
 class LinearProjector(Projector):
     name = "Linear Projection"

Orange/widgets/tests/base.py

@@ -1013,12 +1013,18 @@ 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)
+        if self.widget.isBlocking():
+            spy = QSignalSpy(self.widget.blockingStateChanged)
+            self.assertTrue(spy.wait(timeout))
         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)
+        if w.isBlocking():
+            spy = QSignalSpy(w.blockingStateChanged)
+            self.assertTrue(spy.wait(timeout))
         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 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,187 @@ 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)
+
+    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) -> Future:
+        return self.__future
+
+    def set_status(self, text: str):
+        self._p_status_changed.emit(text)
+
+    def set_progress_value(self, value: float):
+        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_result(self, value: Any):
+        self._p_partial_result_ready.emit(value)
+
+    def is_interruption_requested(self) -> bool:
+        return self.__interruption_requested
+
+    def start(self, executor: concurrent.futures.Executor,
+              func: Callable[[], Any] = None) -> 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) -> bool:
+        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
+
+
+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) -> TaskState:
+        return self.__task
+
+    def on_partial_result(self, result: Any) -> None:
+        """ Invoked from runner (by state) to send the partial results
+        The method should handle partial results, i.e. show them in the plot.
+
+        :param result: any data structure to hold temporary result
+        """
+        raise NotImplementedError
+
+    def on_done(self, result: Any) -> None:
+        """ Invoked when task is done.
+        The method should re-set the result (to double check it) and
+        perform operations with obtained results, eg. send data to the output.
+
+        :param result: any data structure to hold temporary result
+        """
+        raise NotImplementedError
+
+    def on_exception(self, ex: Exception):
+        """ Invoked when an exception occurs during the calculation.
+        Override in order to handle exceptions, eg. show an error
+        message in the widget.
+
+        :param ex: exception
+        """
+        raise ex
+
+    def start(self, task: Callable, *args, **kwargs):
+        """ Call from derived class to start the task.
+        :param task: runner - a method to run in a thread - should accept
+        `state` parameter
+        """
+        self.__cancel_task(wait=False)
+
+        if self.data is None:
+            self.__set_state_ready()
+            return
+
+        assert callable(task), "`task` must be callable!"
+        state = TaskState(self)
+        task = partial(task, *(args + (state,)), **kwargs)
+
+        self.__set_state_busy()
+        self.__start_task(task, state)
+
+    def cancel(self):
+        """ Call from derived class to stop the task. """
+        self.__cancel_task(wait=False)
+        self.__set_state_ready()
+
+    def shutdown(self):
+        """ Call from derived class when the widget is deleted
+         (in onDeleteWidget).
+        """
+        self.__cancel_task(wait=True)
+        self.__executor.shutdown(True)
+
+    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.on_partial_result)
+        state.watcher.done.connect(self.__on_task_done)
+        state.start(self.__executor, task)
+        state.setParent(self)
+        self.__task = state
+
+    def __cancel_task(self, wait: bool = True):
+        if self.__task is not None:
+            state, self.__task = self.__task, None
+            state.cancel()
+            state.partial_result_ready.disconnect(self.on_partial_result)
+            state.status_changed.disconnect(self.setStatusMessage)
+            state.progress_changed.disconnect(self.progressBarSet)
+            state.watcher.done.disconnect(self.__on_task_done)
+            if wait:
+                concurrent.futures.wait([state.future])
+                state.deleteLater()
+            else:
+                w = FutureWatcher(state.future, parent=state)
+                w.done.connect(state.deleteLater)
+
+    def __on_task_done(self, future: Future):
+        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()
+        ex = future.exception()
+        if ex is not None:
+            self.on_exception(ex)
+        else:
+            self.on_done(future.result())