diff --git a/parsl/executors/workqueue/executor.py b/parsl/executors/workqueue/executor.py index b08cd3edfe..67628dacb2 100644 --- a/parsl/executors/workqueue/executor.py +++ b/parsl/executors/workqueue/executor.py @@ -3,7 +3,6 @@ high-throughput system for delegating Parsl tasks to thousands of remote machines """ -import atexit import threading import multiprocessing import logging @@ -298,24 +297,6 @@ def __init__(self, if self.init_command != "": self.launch_cmd = self.init_command + "; " + self.launch_cmd - # register atexit handler to cleanup when Python shuts down - atexit.register(self.atexit_cleanup) - - # Attribute indicating whether this executor was started to shut it down properly. - # This safeguards cases where an object of this executor is created but - # the executor never starts, so it shouldn't be shutdowned. - self.is_started = False - - # Attribute indicating whether this executor was shutdown before. - # This safeguards cases where this object is automatically shut down (e.g., - # via atexit) and the user also explicitly calls shut down. While this is - # permitted, the effect of an executor shutdown should happen only once. - self.is_shutdown = False - - def atexit_cleanup(self): - # Calls this executor's shutdown method upon Python exiting the process. - self.shutdown() - def _get_launch_command(self, block_id): # this executor uses different terminology for worker/launch # commands than in htex @@ -325,8 +306,6 @@ def start(self): """Create submit process and collector thread to create, send, and retrieve Parsl tasks within the Work Queue system. """ - # Mark this executor object as started - self.is_started = True self.tasks_lock = threading.Lock() # Create directories for data and results @@ -713,14 +692,6 @@ def shutdown(self, *args, **kwargs): """Shutdown the executor. Sets flag to cancel the submit process and collector thread, which shuts down the Work Queue system submission. """ - if not self.is_started: - # Don't shutdown if the executor never starts. - return - - if self.is_shutdown: - # Don't shutdown this executor again. - return - logger.debug("Work Queue shutdown started") self.should_stop.value = True @@ -741,7 +712,6 @@ def shutdown(self, *args, **kwargs): self.collector_queue.close() self.collector_queue.join_thread() - self.is_shutdown = True logger.debug("Work Queue shutdown completed") @wrap_with_logs