33high-throughput system for delegating Parsl tasks to thousands of remote machines
44"""
55
6- import atexit
76import threading
87import multiprocessing
98import logging
@@ -298,24 +297,6 @@ def __init__(self,
298297 if self .init_command != "" :
299298 self .launch_cmd = self .init_command + "; " + self .launch_cmd
300299
301- # register atexit handler to cleanup when Python shuts down
302- atexit .register (self .atexit_cleanup )
303-
304- # Attribute indicating whether this executor was started to shut it down properly.
305- # This safeguards cases where an object of this executor is created but
306- # the executor never starts, so it shouldn't be shutdowned.
307- self .is_started = False
308-
309- # Attribute indicating whether this executor was shutdown before.
310- # This safeguards cases where this object is automatically shut down (e.g.,
311- # via atexit) and the user also explicitly calls shut down. While this is
312- # permitted, the effect of an executor shutdown should happen only once.
313- self .is_shutdown = False
314-
315- def atexit_cleanup (self ):
316- # Calls this executor's shutdown method upon Python exiting the process.
317- self .shutdown ()
318-
319300 def _get_launch_command (self , block_id ):
320301 # this executor uses different terminology for worker/launch
321302 # commands than in htex
@@ -325,8 +306,6 @@ def start(self):
325306 """Create submit process and collector thread to create, send, and
326307 retrieve Parsl tasks within the Work Queue system.
327308 """
328- # Mark this executor object as started
329- self .is_started = True
330309 self .tasks_lock = threading .Lock ()
331310
332311 # Create directories for data and results
@@ -713,14 +692,6 @@ def shutdown(self, *args, **kwargs):
713692 """Shutdown the executor. Sets flag to cancel the submit process and
714693 collector thread, which shuts down the Work Queue system submission.
715694 """
716- if not self .is_started :
717- # Don't shutdown if the executor never starts.
718- return
719-
720- if self .is_shutdown :
721- # Don't shutdown this executor again.
722- return
723-
724695 logger .debug ("Work Queue shutdown started" )
725696 self .should_stop .value = True
726697
@@ -741,7 +712,6 @@ def shutdown(self, *args, **kwargs):
741712 self .collector_queue .close ()
742713 self .collector_queue .join_thread ()
743714
744- self .is_shutdown = True
745715 logger .debug ("Work Queue shutdown completed" )
746716
747717 @wrap_with_logs
0 commit comments