count fds

Author: benclifford

parsl/dataflow/dflow.py

@@ -86,7 +86,7 @@ def __init__(self, config: Config) -> None:
         self.run_dir = make_rundir(config.run_dir)
 
         if config.initialize_logging:
-            parsl.set_file_logger("{}/parsl.log".format(self.run_dir), level=logging.DEBUG)
+            _, self.logging_handler = parsl.set_file_logger("{}/parsl.log".format(self.run_dir), level=logging.DEBUG)
 
         logger.info("Starting DataFlowKernel with config\n{}".format(config))
 
@@ -1272,7 +1272,13 @@ def cleanup(self) -> None:
             self.monitoring.close()
             logger.info("Terminated monitoring")
 
-        logger.info("DFK cleanup complete")
+        # TODO: do this in parsl/logutils.py
+        logger.info("DFK cleanup complete - removing parsl.log handler")
+        logger_to_remove = logging.getLogger("parsl")
+        logger_to_remove.removeHandler(self.logging_handler)
+        self.logging_handler.close()
+
+        logger.info("handler closed - is this going to break things?")
 
     def checkpoint(self, tasks: Optional[Sequence[TaskRecord]] = None) -> str:
         """Checkpoint the dfk incrementally to a checkpoint file.

parsl/executors/high_throughput/executor.py

@@ -555,6 +555,8 @@ def _start_local_interchange_process(self):
         self.interchange_proc.start()
         try:
             (self.worker_task_port, self.worker_result_port) = comm_q.get(block=True, timeout=120)
+            comm_q.close()
+            comm_q.join_thread()
         except queue.Empty:
             logger.error("Interchange has not completed initialization in 120s. Aborting")
             raise Exception("Interchange failed to start")
@@ -830,6 +832,21 @@ def shutdown(self, timeout: float = 10.0):
         if self._queue_management_thread:
             self._queue_management_thread.join()
 
+        logger.info("closing context sockets")
+        # this might block if there are outstanding messages (eg if the interchange
+        # has gone away... probably something to do with zmq.LINGER sockopt to remove
+        # this hang risk?
+
+        # these should be initialized to none rather than being absent?
+        if hasattr(self, "incoming_q"):
+            self.incoming_q.close()
+
+        if hasattr(self, "outgoing_q"):
+            self.outgoing_q.close()
+
+        if hasattr(self, "command_client"):
+            self.command_client.close()
+
         logger.info("Finished HighThroughputExecutor shutdown attempt")
 
     def get_usage_information(self):

parsl/executors/taskvine/executor.py

@@ -638,11 +638,13 @@ def shutdown(self, *args, **kwargs):
         # Join all processes before exiting
         logger.debug("Joining on submit process")
         self._submit_process.join()
+        self._submit_process.close()
         logger.debug("Joining on collector thread")
         self._collector_thread.join()
         if self.worker_launch_method == 'factory':
             logger.debug("Joining on factory process")
             self._factory_process.join()
+            self._factory_process.close()
 
         # Shutdown multiprocessing queues
         self._ready_task_queue.close()

parsl/executors/workqueue/executor.py

@@ -732,6 +732,8 @@ def shutdown(self, *args, **kwargs):
 
         logger.debug("Joining on submit process")
         self.submit_process.join()
+        self.submit_process.close()
+
         logger.debug("Joining on collector thread")
         self.collector_thread.join()
 

parsl/log_utils.py

@@ -14,7 +14,7 @@
 import logging
 import typeguard
 
-from typing import Optional
+from typing import Optional, Tuple
 
 
 DEFAULT_FORMAT = (
@@ -66,7 +66,7 @@ def set_stream_logger(name: str = 'parsl',
 def set_file_logger(filename: str,
                     name: str = 'parsl',
                     level: int = logging.DEBUG,
-                    format_string: Optional[str] = None) -> logging.Logger:
+                    format_string: Optional[str] = None) -> Tuple[logging.Logger, logging.FileHandler]:
     """Add a file log handler.
 
     Args:
@@ -94,4 +94,4 @@ def set_file_logger(filename: str,
     futures_logger = logging.getLogger("concurrent.futures")
     futures_logger.addHandler(handler)
 
-    return logger
+    return (logger, handler)

parsl/monitoring/monitoring.py

@@ -252,6 +252,7 @@ def close(self) -> None:
             self.router_exit_event.set()
             logger.info("Waiting for router to terminate")
             self.router_proc.join()
+            self.router_proc.close()
             logger.debug("Finished waiting for router termination")
             if len(exception_msgs) == 0:
                 logger.debug("Sending STOP to DBM")
@@ -260,13 +261,15 @@ def close(self) -> None:
                 logger.debug("Not sending STOP to DBM, because there were DBM exceptions")
             logger.debug("Waiting for DB termination")
             self.dbm_proc.join()
+            self.dbm_proc.close()
             logger.debug("Finished waiting for DBM termination")
 
             # should this be message based? it probably doesn't need to be if
             # we believe we've received all messages
             logger.info("Terminating filesystem radio receiver process")
             self.filesystem_proc.terminate()
             self.filesystem_proc.join()
+            self.filesystem_proc.close()
 
             logger.info("Closing monitoring multiprocessing queues")
             self.exception_q.close()
@@ -284,9 +287,9 @@ def close(self) -> None:
 
 @wrap_with_logs
 def filesystem_receiver(logdir: str, q: "queue.Queue[AddressedMonitoringMessage]", run_dir: str) -> None:
-    logger = set_file_logger("{}/monitoring_filesystem_radio.log".format(logdir),
-                             name="monitoring_filesystem_radio",
-                             level=logging.INFO)
+    logger, _ = set_file_logger("{}/monitoring_filesystem_radio.log".format(logdir),
+                                name="monitoring_filesystem_radio",
+                                level=logging.INFO)
 
     logger.info("Starting filesystem radio receiver")
     setproctitle("parsl: monitoring filesystem receiver")

parsl/monitoring/router.py

@@ -58,9 +58,9 @@ def __init__(self,
 
         """
         os.makedirs(logdir, exist_ok=True)
-        self.logger = set_file_logger("{}/monitoring_router.log".format(logdir),
-                                      name="monitoring_router",
-                                      level=logging_level)
+        self.logger, _ = set_file_logger("{}/monitoring_router.log".format(logdir),
+                                         name="monitoring_router",
+                                         level=logging_level)
         self.logger.debug("Monitoring router starting")
 
         self.hub_address = hub_address

parsl/tests/conftest.py

@@ -3,6 +3,7 @@
 import logging
 import os
 import pathlib
+import psutil
 import random
 import re
 import shutil
@@ -228,6 +229,9 @@ def load_dfk_local_module(request, pytestconfig, tmpd_cwd_session):
     config = pytestconfig.getoption('config')[0]
 
     if config == 'local':
+        this_process = psutil.Process()
+        start_fds = this_process.num_fds()
+        logger.error(f"BENC: open fds: {start_fds}")
         assert threading.active_count() == 1, "precondition: only one thread can be running before this test"
         local_setup = getattr(request.module, "local_setup", None)
         local_teardown = getattr(request.module, "local_teardown", None)
@@ -262,6 +266,12 @@ def load_dfk_local_module(request, pytestconfig, tmpd_cwd_session):
             parsl.clear()
 
         assert threading.active_count() == 1, "test left threads running: " + repr(threading.enumerate())
+        end_fds = this_process.num_fds()
+        logger.error(f"BENC: open fds END: {end_fds}")
+        if end_fds > start_fds:
+            logger.error(f"Open files (not all fds, though?): {this_process.open_files()!r}")
+            os.system(f"ls -l /proc/{os.getpid()}/fd")
+            pytest.fail("BENC: number of open fds increased across test")
 
     else:
         yield

parsl/tests/test_providers/test_local_provider.py

@@ -104,6 +104,7 @@ def test_ssh_channel():
 
 def _stop_sshd(sshd_thread):
     sshd_thread.stop()
+    sshd_thread.join()
 
 
 class SSHDThread(threading.Thread):