Run task workers in separate process bus

Running Kolibri tasks cannot be cancelled[1] and shutting down the
process bus running them will block until the running tasks complete.
This means the Kolibri server can't be stopped quickly or reliably
during the activity's `onActivityStopped` hook.

To workaround that, move the task workers spawned from the
`ServicesPlugin` into a separate process bus. This bus will only be
stopped when the activity is destroyed (or killed). Besides making the
server lifecycle much more reliable, it allows tasks to continue running
and starting in the background as long as the activity process is alive.

Ideally the server and workers would be separate services outside of the
main webview activity, but python-for-android makes that really hard to
accomplish.

1. learningequality/kolibri#10249

Helps: endlessm/kolibri-explore-plugin#740

src/kolibri_android/main_activity/activity.py

@@ -45,21 +45,37 @@ def set_app_key_cookie(url, app_key):
     KolibriAndroidHelper.getInstance().setAppKeyCookie(url, app_key)
 
 
-def _build_kolibri_process_bus(application):
+def _build_server_process_bus(application):
     from .kolibri_bus import AppPlugin
-    from .kolibri_bus import KolibriAppProcessBus
+    from .kolibri_bus import KolibriServerProcessBus
 
     AppPlugin.register_share_file_interface(share_by_intent)
 
-    kolibri_bus = KolibriAppProcessBus(enable_zeroconf=True)
+    kolibri_bus = KolibriServerProcessBus(enable_zeroconf=True)
     AppPlugin(kolibri_bus, application).subscribe()
 
     return kolibri_bus
 
 
+def _build_worker_process_bus(application):
+    from .kolibri_bus import KolibriWorkerProcessBus
+
+    return KolibriWorkerProcessBus()
+
+
 class MainActivity(BaseActivity):
     TO_RUN_IN_MAIN = None
-    _kolibri_bus = None
+
+    # Kolibri process buses for the server and the task workers. Two
+    # buses are used so that the server bus can be quickly stopped when
+    # the activity is stopped. The worker bus is only stopped when the
+    # activity is destroyed.
+    #
+    # FIXME: Ideally these would be completely separate Android
+    # services, but pythonforandroid makes that hard to do.
+    _server_bus = None
+    _worker_bus = None
+
     _saved_kolibri_path = None
     _last_kolibri_path = None
 
@@ -74,40 +90,52 @@ def __init__(self):
     def on_activity_stopped(self, activity):
         super().on_activity_stopped(activity)
 
-        if self._kolibri_bus is None:
+        if self._server_bus is None:
             return
 
-        if self._kolibri_bus.can_transition("IDLE"):
+        # Only the server bus is stopped here since the worker bus will
+        # block until any running tasks complete. This also allows
+        # running tasks to continue in the background.
+        if self._server_bus.can_transition("IDLE"):
             # With some versions of Android, the onSaveInstanceState hook will
             # run after thsi one, so we need to keep track of the webview's
             # URL before switching to the loading screen.
             self._last_kolibri_path = self._get_current_kolibri_path()
             show_loading_page(LOADING_PAGE_URL)
-            self._kolibri_bus.transition("IDLE")
-        elif self._kolibri_bus.state != "IDLE":
+            self._server_bus.transition("IDLE")
+        elif self._server_bus.state != "IDLE":
             logging.warning(
-                f"Kolibri is unable to stop because its state is '{self._kolibri_bus.state}"
+                f"Kolibri is unable to stop because its state is '{self._server_bus.state}"
             )
 
+    def on_activity_destroyed(self, activity):
+        super().on_activity_destroyed(activity)
+
+        if self._worker_bus:
+            self._worker_bus.transition("EXITED")
+
+        if self._server_bus:
+            self._server_bus.transition("EXITED")
+
     def on_activity_resumed(self, activity):
         super().on_activity_resumed(activity)
 
-        if self._kolibri_bus is None:
+        if self._server_bus is None:
             return
 
-        if self._kolibri_bus.can_transition("START"):
+        if self._server_bus.can_transition("START"):
             self._last_kolibri_path = None
             show_loading_page(LOADING_PAGE_URL)
-            self._kolibri_bus.transition("START")
-        elif self._kolibri_bus.state != "START":
+            self._server_bus.transition("START")
+        elif self._server_bus.state != "START":
             logging.warning(
-                f"Kolibri is unable to start because its state is '{self._kolibri_bus.state}'"
+                f"Kolibri is unable to start because its state is '{self._server_bus.state}'"
             )
 
     def on_activity_save_instance_state(self, activity, out_state_bundle):
         super().on_activity_save_instance_state(activity, out_state_bundle)
 
-        if self._kolibri_bus is None:
+        if self._server_bus is None:
             return
 
         kolibri_path = self._last_kolibri_path or self._get_current_kolibri_path()
@@ -128,7 +156,7 @@ def _get_current_kolibri_path(self):
         current_url = KolibriAndroidHelper.getInstance().getUrl()
         self._last_url = None
 
-        if self._kolibri_bus.is_kolibri_url(current_url):
+        if self._server_bus.is_kolibri_url(current_url):
             return urlparse(current_url)._replace(scheme="", netloc="").geturl()
         else:
             return None
@@ -157,19 +185,25 @@ def start_kolibri(self):
 
         init_kolibri(debug=True)
 
-        self._kolibri_bus = _build_kolibri_process_bus(self)
-        app_key = self._kolibri_bus.get_app_key()
+        self._server_bus = _build_server_process_bus(self)
+        app_key = self._server_bus.get_app_key()
         logging.info(f"Setting app key cookie: {app_key}")
         # Android's CookieManager.setCookie awkwardly asks for a full URL, but
         # cookies generally apply across all ports for a given hostname, so it
         # is okay that we give it the expected hostname without specifying a
         # port.
         set_app_key_cookie("http://127.0.0.1", app_key)
 
+        self._worker_bus = _build_worker_process_bus(self)
+
+        # Start the worker bus but don't block on it.
+        logging.info("Starting kolibri workers.")
+        self._worker_bus.graceful()
+
         # start kolibri server
         logging.info("Starting kolibri server.")
 
-        self._kolibri_bus.run()
+        self._server_bus.run()
 
     def _on_start_with_network(self):
         self.TO_RUN_IN_MAIN = self.start_kolibri

src/kolibri_android/main_activity/kolibri_bus.py

@@ -12,11 +12,11 @@
 from magicbus.plugins import SimplePlugin
 
 
-class KolibriAppProcessBus(BaseKolibriProcessBus):
-    def __init__(self, *args, enable_zeroconf=True, **kwargs):
-        super(KolibriAppProcessBus, self).__init__(*args, **kwargs)
+class KolibriServerProcessBus(BaseKolibriProcessBus):
+    """Process bus running Kolibri servers"""
 
-        ServicesPlugin(self).subscribe()
+    def __init__(self, *args, enable_zeroconf=True, **kwargs):
+        super().__init__(*args, **kwargs)
 
         if enable_zeroconf:
             ZeroConfPlugin(self, self.port).subscribe()
@@ -56,6 +56,15 @@ def can_transition(self, to_state: str) -> bool:
         return (self.state, to_state) in self.transitions
 
 
+class KolibriWorkerProcessBus(BaseKolibriProcessBus):
+    """Process bus running Kolibri workers"""
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+
+        ServicesPlugin(self).subscribe()
+
+
 class AppPlugin(SimplePlugin):
     def __init__(self, bus, application):
         self.application = application