localstack · MEPalma · Oct 16, 2024 · Sep 27, 2024 · Sep 27, 2024 · Oct 10, 2024
@@ -72,7 +72,7 @@ def _get_state(self, state_name: str) -> CommonStateField:
     def eval(self, env: Environment) -> None:
         timeout = self.timeout_seconds.timeout_seconds if self.timeout_seconds else None
         env.next_state_name = self.start_at.start_at_name
-        worker_thread = threading.Thread(target=super().eval, args=(env,))
+        worker_thread = threading.Thread(target=super().eval, args=(env,), daemon=True)
         TMP_THREADS.append(worker_thread)
         worker_thread.start()
         worker_thread.join(timeout=timeout)

@@ -35,6 +35,7 @@
 from localstack.services.stepfunctions.asl.component.state.state_props import StateProps
 from localstack.services.stepfunctions.asl.eval.environment import Environment
 from localstack.services.stepfunctions.asl.eval.event.event_detail import EventDetails
+from localstack.utils.common import TMP_THREADS
 
 LOG = logging.getLogger(__name__)
 
@@ -183,8 +184,10 @@ def _exec_and_notify():
                 execution_exceptions.append(ex)
             terminated_event.set()
 
-        thread = Thread(target=_exec_and_notify)
+        thread = Thread(target=_exec_and_notify, daemon=True)
+        TMP_THREADS.append(thread)
         thread.start()
+
         finished_on_time: bool = terminated_event.wait(timeout_seconds)
         frame.set_ended()
         env.close_frame(frame)
@@ -213,7 +216,7 @@ def _eval_state(self, env: Environment) -> None:
         env.context_object_manager.context_object["State"]["RetryCount"] = 0
 
         # Attempt to evaluate the state's logic through until it's successful, caught, or retries have run out.
-        while True:
+        while env.is_running():
             try:
                 self._evaluate_with_timeout(env)
                 break

@@ -77,7 +77,7 @@ def _create_worker(self, env: Environment) -> IterationWorker: ...
 
     def _launch_worker(self, env: Environment) -> IterationWorker:
         worker = self._create_worker(env=env)
-        worker_thread = threading.Thread(target=worker.eval)
+        worker_thread = threading.Thread(target=worker.eval, daemon=True)
         TMP_THREADS.append(worker_thread)
         worker_thread.start()
         return worker

@@ -240,7 +240,7 @@ def _eval_state(self, env: Environment) -> None:
         env.context_object_manager.context_object["State"]["RetryCount"] = 0
 
         # Attempt to evaluate the state's logic through until it's successful, caught, or retries have run out.
-        while True:
+        while env.is_running():
             try:
                 self._evaluate_with_timeout(env)
                 break

@@ -38,7 +38,7 @@ def start(self):
             raise RuntimeError(f"Attempted to rerun BranchWorker for program ${self._program}.")
 
         self._worker_thread = threading.Thread(
-            target=self._thread_routine, name=f"BranchWorker_${self._program}"
+            target=self._thread_routine, name=f"BranchWorker_${self._program}", daemon=True
         )
         TMP_THREADS.append(self._worker_thread)
         self._worker_thread.start()

@@ -65,7 +65,7 @@ def _eval_state(self, env: Environment) -> None:
         input_value = copy.deepcopy(env.stack.pop())
 
         # Attempt to evaluate the state's logic through until it's successful, caught, or retries have run out.
-        while True:
+        while env.is_running():
             try:
                 env.stack.append(input_value)
                 self._evaluate_with_timeout(env)

@@ -120,6 +120,7 @@ def _local_update_wait_for_task_token():
             thread_wait_for_task_token = threading.Thread(
                 target=_local_update_wait_for_task_token,
                 name=f"WaitForTaskToken_SyncTask_{self.resource.resource_arn}",
+                daemon=True,
             )
             TMP_THREADS.append(thread_wait_for_task_token)
             thread_wait_for_task_token.start()

@@ -36,7 +36,7 @@ def __init__(
 
     def eval(self, env: TestStateEnvironment) -> None:
         env.next_state_name = self.test_state.name
-        worker_thread = threading.Thread(target=super().eval, args=(env,))
+        worker_thread = threading.Thread(target=super().eval, args=(env,), daemon=True)
         TMP_THREADS.append(worker_thread)
         worker_thread.start()
         worker_thread.join(timeout=TEST_CASE_EXECUTION_TIMEOUT_SECONDS)

@@ -30,6 +30,7 @@
 from localstack.services.stepfunctions.backend.execution_worker_comm import (
     ExecutionWorkerCommunication,
 )
+from localstack.utils.common import TMP_THREADS
 
 
 class ExecutionWorker:
@@ -104,7 +105,9 @@ def _execution_logic(self):
         self._exec_comm.terminated()
 
     def start(self):
-        Thread(target=self._execution_logic).start()
+        execution_logic_thread = Thread(target=self._execution_logic, daemon=True)
+        TMP_THREADS.append(execution_logic_thread)
+        execution_logic_thread.start()
 
     def stop(self, stop_date: datetime.datetime, error: Optional[str], cause: Optional[str]):
         self.env.set_stop(stop_date=stop_date, cause=cause, error=error)