Merge pull request #2834 from ales-erjavec/fixes/future-set-watcher-empty

[FIX] utils/concurrent: Handle an empty futures list

Author: lanzagar

Orange/widgets/utils/concurrent.py

@@ -21,7 +21,7 @@
     pass
 
 from AnyQt.QtCore import (
-    Qt, QObject, QMetaObject, QThreadPool, QThread, QRunnable,
+    Qt, QObject, QMetaObject, QThreadPool, QThread, QRunnable, QSemaphore,
     QEventLoop, QCoreApplication, QEvent, Q_ARG
 )
 
@@ -570,10 +570,12 @@ class FutureSetWatcher(QObject):
     doneAll = Signal()
 
     def __init__(self, futures=None, *args, **kwargs):
+        # type: (List[Future], ...) -> None
         super().__init__(*args, **kwargs)
         self.__futures = None
+        self.__semaphore = None
         self.__countdone = 0
-        if futures:
+        if futures is not None:
             self.setFutures(futures)
 
     def setFutures(self, futures):
@@ -593,22 +595,33 @@ def setFutures(self, futures):
         selfweakref = weakref.ref(self)
         schedule_emit = methodinvoke(self, "__emitpending", (int, Future))
 
+        # Semaphore counting the number of future that have enqueued
+        # done notifications. Used for the `wait` implementation.
+        self.__semaphore = semaphore = QSemaphore(0)
+
         for i, future in enumerate(futures):
             self.__futures.append(future)
 
             def on_done(index, f):
-                selfref = selfweakref()  # not safe really
-                if selfref is None:
-                    return
                 try:
-                    schedule_emit(index, f)
-                except RuntimeError:
-                    # Ignore RuntimeErrors (when C++ side of QObject is deleted)
-                    # (? Use QObject.destroyed and remove the done callback ?)
-                    pass
+                    selfref = selfweakref()  # not safe really
+                    if selfref is None:  # pragma: no cover
+                        return
+                    try:
+                        schedule_emit(index, f)
+                    except RuntimeError:  # pragma: no cover
+                        # Ignore RuntimeErrors (when C++ side of QObject is deleted)
+                        # (? Use QObject.destroyed and remove the done callback ?)
+                        pass
+                finally:
+                    semaphore.release()
 
             future.add_done_callback(partial(on_done, i))
 
+        if not self.__futures:
+            # `futures` was an empty sequence.
+            methodinvoke(self, "doneAll", ())()
+
     @Slot(int, Future)
     def __emitpending(self, index, future):
         # type: (int, Future) -> None
@@ -640,10 +653,29 @@ def __emitpending(self, index, future):
     def flush(self):
         """
         Flush all pending signal emits currently enqueued.
+
+        Must only ever be called from the thread this object lives in
+        (:func:`QObject.thread()`).
         """
-        assert QThread.currentThread() is self.thread()
+        if QThread.currentThread() is not self.thread():
+            raise RuntimeError("`flush()` called from a wrong thread.")
+        # NOTE: QEvent.MetaCall is the event implementing the
+        # `Qt.QueuedConnection` method invocation.
         QCoreApplication.sendPostedEvents(self, QEvent.MetaCall)
 
+    def wait(self):
+        """
+        Wait for for all the futures to complete and *enqueue* notifications
+        to this object, but do not emit any signals.
+
+        Use `flush()` to emit all signals after a `wait()`
+        """
+        if self.__futures is None:
+            raise RuntimeError("Futures were not set.")
+
+        self.__semaphore.acquire(len(self.__futures))
+        self.__semaphore.release(len(self.__futures))
+
 
 class methodinvoke(object):
     """

Orange/widgets/utils/tests/test_concurrent.py

@@ -1,4 +1,5 @@
 import unittest
+import unittest.mock
 import threading
 import random
 
@@ -201,6 +202,43 @@ def as_set(seq):
         self.assertSetEqual(as_set(spy.resultAt), {(0, True)})
         self.assertSetEqual(as_set(spy.exceptionAt), set())
 
+        # doneAll must always be emitted after the doneAt signals.
+        executor = ThreadPoolExecutor(max_workers=2)
+        futures = [executor.submit(pow, 1000, 1000) for _ in range(100)]
+        watcher = FutureSetWatcher(futures)
+        emithistory = []
+        watcher.doneAt.connect(lambda i, f: emithistory.append(("doneAt", i, f)))
+        watcher.doneAll.connect(lambda: emithistory.append(("doneAll", )))
+
+        spy = spies(watcher)
+        watcher.wait()
+        self.assertEqual(len(spy.doneAll), 0)
+        self.assertEqual(len(spy.doneAt), 0)
+        watcher.flush()
+        self.assertEqual(len(spy.doneAt), 100)
+        self.assertEqual(list(spy.doneAll), [[]])
+        self.assertSetEqual(set(emithistory[:-1]),
+                            {("doneAt", i, f) for i, f in enumerate(futures)})
+        self.assertEqual(emithistory[-1], ("doneAll",))
+
+        # doneAll must be emitted even when on an empty futures list
+        watcher = FutureSetWatcher()
+        watcher.setFutures([])
+        spy = spies(watcher)
+        self.assertTrue(spy.doneAll.wait())
+
+        watcher = FutureSetWatcher()
+        watcher.setFutures([])
+        watcher.wait()
+
+        watcher = FutureSetWatcher()
+        with self.assertRaises(RuntimeError):
+            watcher.wait()
+
+        with unittest.mock.patch.object(watcher, "thread", lambda: 42), \
+                self.assertRaises(RuntimeError):
+            watcher.flush()
+
 
 class TestTask(CoreAppTestCase):