Remove ZeroMQ subscription threading

Author: dwsutherland

cylc/uiserver/app.py

@@ -491,6 +491,12 @@ def initialize_settings(self):
             self.scan_interval * 1000
         ).start()
 
+        # process incoming ZeroMQ deltas from the scheduler(s)
+        ioloop.PeriodicCallback(
+            self.data_store_mgr.process_incoming_deltas,
+            1 * 1000
+        ).start()
+
     def initialize_handlers(self):
         self.authobj = self.set_auth()
         self.set_sub_server()

cylc/uiserver/data_store_mgr.py

@@ -32,12 +32,13 @@
 """
 
 import asyncio
-from concurrent.futures import ThreadPoolExecutor
 from copy import deepcopy
 from pathlib import Path
 import time
 from typing import TYPE_CHECKING, Dict, Optional, Set, cast
 
+import zmq
+
 from cylc.flow.exceptions import WorkflowStopped
 from cylc.flow.id import Tokens
 from cylc.flow.network.server import PB_METHOD_MAP
@@ -105,7 +106,6 @@ def __init__(self, workflows_mgr, log, max_threads=10):
         self.w_subs: Dict[str, WorkflowSubscriber] = {}
         self.topics = {ALL_DELTAS.encode('utf-8'), b'shutdown'}
         self.loop = None
-        self.executor = ThreadPoolExecutor(max_threads)
         self.delta_queues = {}
 
     @log_call
@@ -163,10 +163,7 @@ async def connect_workflow(self, w_id, contact_data):
 
         self.delta_queues[w_id] = {}
 
-        # Might be options other than threads to achieve
-        # non-blocking subscriptions, but this works.
-        self.executor.submit(
-            self._start_subscription,
+        self._start_subscription(
             w_id,
             contact_data['name'],
             contact_data[CFF.HOST],
@@ -256,11 +253,27 @@ def _start_subscription(self, w_id, reg, host, port):
             context=self.workflows_mgr.context,
             topics=self.topics
         )
-        self.w_subs[w_id].loop.run_until_complete(
-            self.w_subs[w_id].subscribe(
-                process_delta_msg,
-                func=self._update_workflow_data,
-                w_id=w_id))
+
+    async def process_incoming_deltas(self):
+        w_empty = set()
+        while set(self.w_subs.keys()).difference(w_empty):
+            for w_id, sub in self.w_subs.items():
+                if w_id in w_empty:
+                    continue
+                try:
+                    [topic, delta] = await sub.socket.recv_multipart(
+                        flags=zmq.NOBLOCK
+                    )
+                except zmq.ZMQError:
+                    w_empty.add(w_id)
+                    continue
+
+                process_delta_msg(
+                    topic,
+                    delta,
+                    func=self._update_workflow_data,
+                    w_id=w_id
+                )
 
     def _update_workflow_data(self, topic, delta, w_id):
         """Manage and apply incoming data-store deltas.
@@ -346,16 +359,15 @@ def _reconcile_update(self, topic, delta, w_id):
             self.log.debug(
                 f'Out of sync with {topic} of {w_id}... Reconciling.')
             try:
-                # use threadsafe as client socket is in main loop thread.
-                future = asyncio.run_coroutine_threadsafe(
+                task = asyncio.create_task(
                     workflow_request(
                         self.workflows_mgr.workflows[w_id]['req_client'],
                         'pb_data_elements',
                         args={'element_type': topic}
                     ),
                     self.loop
                 )
-                new_delta_msg = future.result(self.RECONCILE_TIMEOUT)
+                new_delta_msg = task.result(self.RECONCILE_TIMEOUT)
                 new_delta = DELTAS_MAP[topic]()
                 new_delta.ParseFromString(new_delta_msg)
                 self._clear_data_field(w_id, topic)
@@ -366,7 +378,7 @@ def _reconcile_update(self, topic, delta, w_id):
                     f'The reconcile update coroutine {w_id} {topic}'
                     f'took too long, cancelling the subscription/sync.'
                 )
-                future.cancel()
+                task.cancel()
             except Exception as exc:
                 self.log.exception(exc)