Terminate htex result queue thread by polling for an exit flag

Before this PR, this thread stays running forever

this *requires* socket to be closed at exit -- and this PR introduces code to do that

context:
see recent flux PR for same problem.
because stopping this thread is now allowing garbage collection to happen, it looks like? or something similar... see PR #3517 for the same problem in Flux


counts:
before this PR, on parsl/tests/test_monitoring/

451 fds, 32 threads

after this PR, 48 fds, 1 thread

Author: benclifford

parsl/executors/high_throughput/executor.py

@@ -347,6 +347,9 @@ def __init__(self,
             interchange_launch_cmd = DEFAULT_INTERCHANGE_LAUNCH_CMD
         self.interchange_launch_cmd = interchange_launch_cmd
 
+        self._result_queue_thread_exit = threading.Event()
+        self._result_queue_thread: Optional[threading.Thread] = None
+
     radio_mode = "htex"
 
     def _warn_deprecated(self, old: str, new: str):
@@ -465,9 +468,11 @@ def _result_queue_worker(self):
         """
         logger.debug("Result queue worker starting")
 
-        while not self.bad_state_is_set:
+        while not self.bad_state_is_set and not self._result_queue_thread_exit.is_set():
             try:
-                msgs = self.incoming_q.get()
+                msgs = self.incoming_q.get(timeout_ms=1000)
+                if msgs is None:  # timeout
+                    continue
 
             except IOError as e:
                 logger.exception("Caught broken queue with exception code {}: {}".format(e.errno, e))
@@ -504,7 +509,6 @@ def _result_queue_worker(self):
                         if 'result' in msg:
                             result = deserialize(msg['result'])
                             task_fut.set_result(result)
-
                         elif 'exception' in msg:
                             try:
                                 s = deserialize(msg['exception'])
@@ -527,6 +531,8 @@ def _result_queue_worker(self):
                     else:
                         raise BadMessage("Message received with unknown type {}".format(msg['type']))
 
+        logger.info("Closing result ZMQ pipe")
+        self.incoming_q.close()
         logger.info("Result queue worker finished")
 
     def _start_local_interchange_process(self) -> None:
@@ -827,6 +833,8 @@ def shutdown(self, timeout: float = 10.0):
 
         logger.info("Attempting HighThroughputExecutor shutdown")
 
+        logger.info("Terminating interchange and result queue thread")
+        self._result_queue_thread_exit.set()
         self.interchange_proc.terminate()
         try:
             self.interchange_proc.wait(timeout=timeout)
@@ -851,6 +859,10 @@ def shutdown(self, timeout: float = 10.0):
             logger.info("Closing command client")
             self.command_client.close()
 
+        logger.info("Waiting for result queue thread exit")
+        if self._result_queue_thread:
+            self._result_queue_thread.join()
+
         logger.info("Finished HighThroughputExecutor shutdown attempt")
 
     def get_usage_information(self):

parsl/executors/high_throughput/zmq_pipes.py

@@ -205,12 +205,21 @@ def __init__(self, ip_address, port_range, cert_dir: Optional[str] = None):
         self.port = self.results_receiver.bind_to_random_port("tcp://{}".format(ip_address),
                                                               min_port=port_range[0],
                                                               max_port=port_range[1])
+        self.poller = zmq.Poller()
+        self.poller.register(self.results_receiver, zmq.POLLIN)
 
-    def get(self):
+    def get(self, timeout_ms=None):
+        """Get a message from the queue, returning None if timeout expires
+        without a message. timeout is measured in milliseconds.
+        """
         logger.debug("Waiting for ResultsIncoming message")
-        m = self.results_receiver.recv_multipart()
-        logger.debug("Received ResultsIncoming message")
-        return m
+        socks = dict(self.poller.poll(timeout=timeout_ms))
+        if self.results_receiver in socks and socks[self.results_receiver] == zmq.POLLIN:
+            m = self.results_receiver.recv_multipart()
+            logger.debug("Received ResultsIncoming message")
+            return m
+        else:
+            return None
 
     def close(self):
         self.results_receiver.close()

parsl/tests/test_htex/test_disconnected_blocks.py

@@ -9,6 +9,8 @@
 from parsl.jobs.states import JobState, JobStatus
 from parsl.providers import LocalProvider
 
+logger = logging.getLogger(__name__)
+
 
 def local_config():
     """Config to simulate failing blocks without connecting"""
@@ -41,25 +43,39 @@ def double(x):
 @pytest.mark.local
 def test_disconnected_blocks():
     """Test reporting of blocks that fail to connect from HTEX"""
+    logger.info("BENC: 1")
     dfk = parsl.dfk()
+    logger.info("BENC: 2")
     executor = dfk.executors["HTEX"]
+    logger.info("BENC: 3")
 
     connected_blocks = executor.connected_blocks()
+    logger.info("BENC: 4")
     assert not connected_blocks, "Expected 0 blocks"
+    logger.info("BENC: 5")
 
     future = double(5)
+    logger.info("BENC: 6")
     with pytest.raises(BadStateException):
+        logger.info("BENC: 7")
         future.result()
+        logger.info("BENC: 8")
+    logger.info("BENC: 9")
 
     assert isinstance(future.exception(), BadStateException)
+    logger.info("BENC: 10")
     exception_body = str(future.exception())
     assert "EXIT CODE: 0" in exception_body
+    logger.info("BENC: 11")
 
     status_dict = executor.status()
+    logger.info("BENC: 12")
     assert len(status_dict) == 1, "Expected only 1 block"
     for status in status_dict.values():
         assert isinstance(status, JobStatus)
         assert status.state == JobState.MISSING
 
+    logger.info("BENC: 13")
     connected_blocks = executor.connected_blocks()
+    logger.info("BENC: 14")
     assert not connected_blocks, "Expected 0 blocks"