gui multiprocess - experimental feature (#780)

server/gui/main.py

@@ -1,4 +1,5 @@
 # flake8: noqa F403, F405
+from multiprocessing import Pipe, Process
 from os.path import splitext, basename
 import sys
 import threading
@@ -7,33 +8,48 @@ from cefpython3 import cefpython as cef
 from PySide2.QtCore import *
 from PySide2.QtWidgets import *
 
-from server.app.app import Server
 from server.gui.browser import CefWidget, CefApplication
-from server.gui.workers import DataLoadWorker, ServerRunWorker
+from server.gui.workers import Worker, SiteReadyWorker
-from server.gui.utils import WINDOWS, LINUX, MAC, FileLoadSignals
+from server.gui.utils import WINDOWS, LINUX, MAC, FileLoadSignals, Emitter, WorkerSignals
 from server.utils.constants import MODES
+from server.utils.utils import find_available_port
 
 
 # Configuration
 # TODO remember this or calculate it?
 WIDTH = 1024
 HEIGHT = 768
+GUI_PORT = find_available_port("localhost")
 
-# noinspection PyUnresolvedReferences
 class MainWindow(QMainWindow):
     def __init__(self):
         super(MainWindow, self).__init__(None)
         self.cef_widget = None
         self.data_widget = None
-        self.server = Server()
+        self.parent_conn, self.child_conn = Pipe()
-        self.server.create_app()
+        self.load_emitter = Emitter(self.parent_conn, WorkerSignals)
-        self.runServer()
+        self.load_emitter.signals.error.connect(self.restartOnError)
+        self.emitter_thread = threading.Thread(target=self.load_emitter.run, daemon=True)
+        self.emitter_thread.start()
+        self.worker = None
+        self.url = f"http://localhost:{GUI_PORT}/"
         self.setWindowTitle("cellxgene")
 
         # Strong focus - accepts focus by tab & click
         self.setFocusPolicy(Qt.StrongFocus)
         self.setupLayout()
-        self.setupMenu()
+        # self.setupMenu()
+
+    def restartOnError(self):
+        if self.worker:
+            self.worker.terminate()
+        self.parent_conn.close()
+        # close emitter on error/finished
+        self.parent_conn, self.child_conn = Pipe()
+        self.load_emitter = Emitter(self.parent_conn, WorkerSignals)
+        self.emitter_thread = threading.Thread(target=self.load_emitter.run, daemon=True)
+        self.emitter_thread.start()
+        # send to load with error message?
 
     def setupLayout(self):
         self.resize(WIDTH, HEIGHT)
@@ -67,9 +83,10 @@ class MainWindow(QMainWindow):
             # cef widget in the layout with the container.
             self.container = QWidget.createWindowContainer(
                 self.cef_widget.hidden_window, parent=self)
-            stacked_layout.addWidget(self.container, 1, 0)
+            self.stacked_layout.addWidget(self.container, 1, 0)
 
     def setupMenu(self):
+        # TODO add communication to subprocess on reload
         main_menu = self.menuBar()
         file_menu = main_menu.addMenu('File')
         load_action = QAction("Load file...", self)
@@ -87,11 +104,6 @@ class MainWindow(QMainWindow):
             self.cef_widget.browser.CloseBrowser(True)
             self.clearBrowserReferences()
 
-    def runServer(self):
-        worker = ServerRunWorker(self.server.app, host="127.0.0.1", port=8000)
-        self.httpd = threading.Thread(target=worker.run, daemon=True)
-        self.httpd.start()
-
     def clearBrowserReferences(self):
         # Clear browser references that you keep anywhere in your
         # code. All references must be cleared for CEF to shutdown cleanly.
@@ -117,6 +129,7 @@ class LoadWidget(QFrame):
         message_layout = QHBoxLayout()
         message_layout.setContentsMargins(0, 0, 0, 0)
 
+        self.serverError = False
         self.title = ""
         self.label = QLabel("cellxgene")
         logo_layout.addWidget(self.label)
@@ -131,7 +144,7 @@ class LoadWidget(QFrame):
         self.embeddings = QComboBox(self)
         self.embeddings.currentIndexChanged.connect(self.updateEmbedding)
         self.embeddings.addItems(MODES)
-        self.embedding_selection = MODES[0]
+        self.embedding_selection = [MODES[0]]
         load_layout.addWidget(self.embeddings, 1, 0)
 
         self.load = QPushButton("Open...")
@@ -155,14 +168,19 @@ class LoadWidget(QFrame):
         self.signals.selectedFile.connect(self.createScanpyEngine)
 
     def updateEmbedding(self, idx):
-        self.embedding_selection = MODES[idx]
+        self.embedding_selection = [MODES[idx]]
 
     def createScanpyEngine(self, file_name):
-        worker = DataLoadWorker(file_name, self.embedding_selection)
+        worker = Worker(self.window().parent_conn, self.window().child_conn, file_name, self.title, host="127.0.0.1", port=GUI_PORT,
-        worker.signals.result.connect(self.onDataSuccess)
+                        layout=self.embedding_selection)
-        worker.signals.error.connect(self.onDataError)
+        self.window().load_emitter.signals.ready.connect(self.onDataReady)
-        self.load_worker = threading.Thread(target=worker.run, daemon=True)
+        self.window().load_emitter.signals.engine_error.connect(self.onEngineError)
-        self.load_worker.start()
+        self.window().load_emitter.signals.server_error.connect(self.onServerError)
+        # Error is generic error from emitter
+        self.window().load_emitter.signals.error.connect(self.onServerError)
+        self.window().worker = Process(target=worker.run, daemon=True)
+        self.window().worker.start()
+        self.window().child_conn.close()
 
     def onLoad(self):
         options = QFileDialog.Options()
@@ -172,20 +190,42 @@ class LoadWidget(QFrame):
         self.title = splitext(basename(file_name))[0]
         if file_name:
             self.signals.selectedFile.emit(file_name)
+            # Reset error on reload
+            self.serverError = False
 
+    def onDataReady(self):
+        self.site_ready_worker = SiteReadyWorker(self.window().url)
+        self.site_ready_worker.signals.ready.connect(self.onServerReady)
+        self.site_ready_worker.signals.error.connect(self.onDataError)
 
-    def onDataSuccess(self, data):
+        srw_thread = threading.Thread(target=self.site_ready_worker.run, daemon=True)
-        self.window().server.attach_data(data, self.title)
+        srw_thread.start()
-        self.navigateToLocation()
-        # Reveal browser
-        self.window().stacked_layout.setCurrentIndex(1)
 
-    def onDataError(self, err):
+    def onServerReady(self):
+        if not self.serverError:
+            self.window().cef_widget.browser.Navigate(self.window().url)
+            self.window().stacked_layout.setCurrentIndex(1)
+
+    def onServerError(self, err):
+        # Restart worker
+        self.serverError = True
+        # Report error and switch to load screen
+        self.window().restartOnError()
+        self.window().stacked_layout.setCurrentIndex(0)
         self.error_label.setText(f"Error: {err}")
         self.error_label.resize(self.MAX_CONTENT_WIDTH, self.error_label.height())
 
-    def navigateToLocation(self, location="http://localhost:8000/"):
+    def onEngineError(self, err):
-        self.window().cef_widget.browser.Navigate(location)
+        self.window().restartOnError()
+        self.window().stacked_layout.setCurrentIndex(0)
+        self.error_label.setText(f"Error: {err}")
+        self.error_label.resize(self.MAX_CONTENT_WIDTH, self.error_label.height())
+
+    def onDataError(self, err):
+        self.window().restartOnError()
+        self.window().stacked_layout.setCurrentIndex(0)
+        self.error_label.setText(f"Error: {err}")
+        self.error_label.resize(self.MAX_CONTENT_WIDTH, self.error_label.height())
 
 
 def main():
@@ -203,16 +243,21 @@ def main():
     main_window.show()
     main_window.activateWindow()
     main_window.raise_()
-    app.exec_()
+    try:
+        app.exec_()
+    except Exception as e:
+        raise
+    finally:
+        # Clean up on close
+        if not cef.GetAppSetting("external_message_pump"):
+            app.stopTimer()
 
-    # Clean up on close
+        if main_window.worker:
-    if not cef.GetAppSetting("external_message_pump"):
+            main_window.worker.terminate()
-        app.stopTimer()
+        del main_window  # Just to be safe, similarly to "del app"
-    # TODO clean up threads when we switch threading model
+        del app  # Must destroy app object before calling Shutdown
-    del main_window  # Just to be safe, similarly to "del app"
+        cef.Shutdown()
-    del app  # Must destroy app object before calling Shutdown
+        sys.exit(0)
-    cef.Shutdown()
-    sys.exit(0)
 
 
 if __name__ == '__main__':

server/gui/utils.py

@@ -13,13 +13,55 @@ class WorkerSignals(QObject):
     Defines the signals available from a running worker thread.
     Supported signals are:
         finished
+        ready
         error - `str` error message
         result - `object` data returned from processing, anything
     """
     finished = Signal()
+    engine_error = Signal(str)
+    server_error = Signal(str)
     error = Signal(str)
     result = Signal(object)
+    ready = Signal()
+
+
+class SiteReadySignals(QObject):
+    """
+    Defines the signals available from a running worker thread.
+    Supported signals are:
+        timeout
+        ready
+        error - `str` error message
+    """
+    ready = Signal()
+    timeout = Signal()
+    error = Signal(str)
 
 
 class FileLoadSignals(QObject):
     selectedFile = Signal(str)
+
+
+class Emitter:
+    def __init__(self, transport, signals):
+        self.transport = transport
+        self.signals = signals()
+
+    def _emit(self, signature, args=None):
+        if args is None:
+            getattr(self.signals, signature).emit()
+        else:
+            getattr(self.signals, signature).emit(args)
+
+    def run(self):
+        while True:
+            try:
+                signature = self.transport.recv()
+            except EOFError:
+                # Server done
+                break
+            except Exception as e:
+                self.signals.error.emit(str(e))
+                break
+            else:
+                self._emit(*signature)

server/gui/workers.py

@@ -1,47 +1,92 @@
+from multiprocessing import Process
 import traceback
+import time
 
-from server.gui.utils import WorkerSignals
+import requests
+
+from server.gui.utils import SiteReadySignals
 
 
-class DataLoadWorker():
+class EmittingProcess(Process):
-    def __init__(self, data_file, layout="umap", *args, **kwargs):
+    def __init__(self, parent_conn, child_conn, *arg, **kwargs):
-        super(DataLoadWorker, self).__init__()
+        super(EmittingProcess, self).__init__()
-        self.data_file = data_file
+        self.parent_conn = parent_conn
-        self.layout = layout
+        self.child_conn = child_conn
-        self.signals = WorkerSignals()
 
     def run(self):
-        if not self.data_file:
+        self.parent_conn.close()
-            self.signals.finished.emit()
-            return
 
-        # delayed import to speed load
+    def emit(self, signal_name, *args):
-        from server.app.scanpy_engine.scanpy_engine import ScanpyEngine
+        message = (signal_name, *args)
-        args = {
+        self.child_conn.send(message)
-            "layout": self.layout,
-            "diffexp": "ttest",
-            "max_category_items": 100,
-            "diffexp_lfc_cutoff": 0.01,
-            "obs_names": None,
-            "var_names": None,
-        }
-        try:
-            data_results = ScanpyEngine(self.data_file, args)
-        except Exception as e:
-            traceback.print_exc()
-            self.signals.error.emit(str(e))
-        else:
-            self.signals.result.emit(data_results)
-        finally:
-            self.signals.finished.emit()
 
 
-class ServerRunWorker():
+class Worker(EmittingProcess):
-    def __init__(self, app, host, port, *args, **kwargs):
+    def __init__(self, parent_conn, child_conn, data_file, title, host, port, layout=["umap"], *args, **kwargs):
-        super(ServerRunWorker, self).__init__()
+        super(Worker, self).__init__(parent_conn, child_conn)
-        self.app = app
+        self.data_file = data_file
+        self.layout = layout
+        self.title = title
         self.host = host
         self.port = port
 
     def run(self):
-        self.app.run(host=self.host, debug=False, port=self.port, threaded=True)
+        super(Worker, self).run()
+        if not self.data_file:
+            self.emit("finished")
+            return
+        from server.app.app import Server
+        from server.app.scanpy_engine.scanpy_engine import ScanpyEngine
+        # create server
+        try:
+            server = Server()
+            server.create_app()
+        except Exception as e:
+            self.emit("server_error", str(e))
+            self.emit("finished")
+            return
+        # load data
+        try:
+            args = {
+                "layout": self.layout,
+                "max_category_items": 100,
+                "diffexp_lfc_cutoff": 0.01,
+                "obs_names": None,
+                "var_names": None,
+            }
+            data = ScanpyEngine(self.data_file, args)
+            server.attach_data(data, self.title)
+            self.emit("ready")
+        except Exception as e:
+            self.emit("engine_error", str(e))
+            self.emit("finished")
+            return
+        # launch server
+        try:
+            server.app.run(host=self.host, debug=False, port=self.port, threaded=True)
+        except Exception as e:
+            traceback.print_exc()
+            self.emit("server_error", str(e))
+        finally:
+            self.emit("finished")
+
+
+class SiteReadyWorker:
+    def __init__(self, location):
+        super(SiteReadyWorker, self).__init__()
+        self.signals = SiteReadySignals()
+        self.location = location
+
+    def run(self):
+        session = requests.Session()
+        for i in range(90):
+            try:
+                session.head(self.location)
+                self.signals.ready.emit()
+                break
+            except requests.exceptions.ConnectionError:
+                time.sleep(1)
+            except Exception as e:
+                traceback.print_exc()
+                self.signals.error.emit(str(e))
+        self.signals.timeout.emit()

setup.py

@@ -36,5 +36,5 @@ setup(
         "Topic :: Scientific/Engineering :: Bio-Informatics",
     ],
     entry_points={"console_scripts": ["cellxgene = server.cli.cli:cli"]},
-    extras_require=dict(louvain=["python-igraph", "louvain>=0.6"], gui=["PySide2>=5.12.3", "cefpython3>=66"]),
+    extras_require=dict(louvain=["python-igraph", "louvain>=0.6"], gui=["PySide2>=5.12.3", "cefpython3>=66", "requests"]),
 )