Make OpenHTF TestState, PhaseExecutor, and KillableThread thread-safe for concurrent phases.

openhtf/core/monitors.py

@@ -71,6 +71,11 @@ def __init__(self, measurement_name: Text,
     self.measurement_name = measurement_name
     self.monitor_desc = monitor_desc
     self.test_state = test_state
+    # Capture the monitored phase's PhaseState on the phase thread so this
+    # thread can bind to it. TestState tracks the running phase per thread, so
+    # without this the monitor thread would have no running phase when other
+    # phases execute concurrently.
+    self._phase_state = test_state.running_phase_state
     self.interval_ms = interval_ms
     self.extra_kwargs = extra_kwargs
 
@@ -91,6 +96,7 @@ def get_value(self) -> Any:
     return self.monitor_desc.with_args(**kwargs)(self.test_state)
 
   def _thread_proc(self):
+    self.test_state.running_phase_state = self._phase_state
     measurement = getattr(self.test_state.test_api.measurements,
                           self.measurement_name)
     start_time = time.time()

openhtf/core/phase_executor.py

@@ -36,7 +36,7 @@
 import time
 import traceback
 import types
-from typing import Any, Dict, Optional, Text, Tuple, Type, TYPE_CHECKING, Union
+from typing import Any, Dict, Optional, Set, TYPE_CHECKING, Text, Tuple, Type, Union
 
 import attr
 from openhtf import util
@@ -170,10 +170,12 @@ def __init__(self, phase_desc: phase_descriptor.PhaseDescriptor,
     self._phase_desc = phase_desc
     self._test_state = test_state
     self._subtest_rec = subtest_rec
+    self._phase_state = test_state.running_phase_state
     self._phase_execution_outcome = None  # type: Optional[PhaseExecutionOutcome]
 
   def _thread_proc(self) -> None:
     """Execute the encompassed phase and save the result."""
+    self._test_state.running_phase_state = self._phase_state
     # Call the phase, save the return value, or default it to CONTINUE.
     phase_return = self._phase_desc(self._test_state)
     if phase_return is None:
@@ -239,22 +241,36 @@ def __init__(self, test_state: 'htf_test_state.TestState'):
     # _execute_phase_once is setting up the next phase thread.
     self._current_phase_thread_lock = threading.Lock()
     self._current_phase_thread = None  # type: Optional[PhaseExecutorThread]
+    self._active_phase_threads = set()  # type: Set[PhaseExecutorThread]
     self._stopping = threading.Event()
 
-  def _should_repeat(self, phase: phase_descriptor.PhaseDescriptor,
-                     phase_execution_outcome: PhaseExecutionOutcome, is_last_repeat: bool) -> bool:
+  def _should_repeat(
+      self,
+      phase: phase_descriptor.PhaseDescriptor,
+      phase_execution_outcome: PhaseExecutionOutcome,
+      is_last_repeat: bool,
+  ) -> bool:
     """Returns whether a phase should be repeated."""
+    # Never repeat a phase that was killed due to stop or abort.
+    if phase_execution_outcome.is_aborted:
+      return False
     if phase_execution_outcome.is_timeout and phase.options.repeat_on_timeout:
       return True
     elif phase_execution_outcome.is_repeat:
       return True
     elif phase.options.force_repeat:
       return True
     elif phase.options.repeat_on_measurement_fail:
-      last_phase_outcome = self.test_state.test_record.phases[-1].outcome
-      last_repeat_failed = last_phase_outcome == test_record.PhaseOutcome.FAIL
+      # Set by running_phase_context on this thread; only None if the phase
+      # was short-circuited (e.g. by run_if) before any phase ran here.
+      last_phase_rec = self.test_state.last_phase_record
+      if last_phase_rec is None:
+        return False
+      last_repeat_failed = (
+          last_phase_rec.outcome == test_record.PhaseOutcome.FAIL
+      )
       if last_repeat_failed and not is_last_repeat:
-        self.test_state.test_record.phases[-1].outcome = test_record.PhaseOutcome.SKIP
+        last_phase_rec.outcome = test_record.PhaseOutcome.SKIP
       return last_repeat_failed
     return False
 
@@ -320,7 +336,6 @@ def _execute_phase_once(
                           phase_desc.name)
         return PhaseExecutionOutcome(phase_descriptor.PhaseResult.SKIP), None
 
-
     override_result = None
     with self.test_state.running_phase_context(phase_desc) as phase_state:
       if subtest_rec:
@@ -347,14 +362,18 @@ def _execute_phase_once(
                                            run_with_profiling, subtest_rec)
         phase_thread.start()
         self._current_phase_thread = phase_thread
+        self._active_phase_threads.add(phase_thread)
 
       phase_state.result = phase_thread.join_or_die()
       if phase_state.result.is_repeat and is_last_repeat:  # pyrefly: ignore[missing-attribute]
         self.logger.error('Phase returned REPEAT, exceeding repeat_limit.')
         phase_state.hit_repeat_limit = True
         override_result = PhaseExecutionOutcome(
             phase_descriptor.PhaseResult.STOP)
-      self._current_phase_thread = None
+      with self._current_phase_thread_lock:
+        self._active_phase_threads.discard(phase_thread)
+        if self._current_phase_thread == phase_thread:
+          self._current_phase_thread = None
 
     # Refresh the result in case a validation for a partially set measurement
     # or phase diagnoser raised an exception.
@@ -433,18 +452,23 @@ def stop(
     """
     self._stopping.set()
     with self._current_phase_thread_lock:
-      phase_thread = self._current_phase_thread
-      if not phase_thread:
-        return
-
-    if phase_thread.is_alive():
-      phase_thread.kill()
-
-      self.logger.debug('Waiting for cancelled phase to exit: %s', phase_thread)
-      timeout = timeouts.PolledTimeout.from_seconds(timeout_s)
-      while phase_thread.is_alive() and not timeout.has_expired():
-        time.sleep(0.1)
-      self.logger.debug('Cancelled phase %s exit',
-                        "didn't" if phase_thread.is_alive() else 'did')
+      threads_to_kill = list(self._active_phase_threads)
+
+    timeout = timeouts.PolledTimeout.from_seconds(timeout_s)
+    for phase_thread in threads_to_kill:
+      if phase_thread.is_alive():
+        phase_thread.kill()
+
+    for phase_thread in threads_to_kill:
+      if phase_thread.is_alive():
+        self.logger.debug(
+            'Waiting for cancelled phase to exit: %s', phase_thread
+        )
+        while phase_thread.is_alive() and not timeout.has_expired():
+          time.sleep(0.1)
+        self.logger.debug(
+            'Cancelled phase %s exit',
+            "didn't" if phase_thread.is_alive() else 'did',
+        )
     # Clear the currently running phase, whether it finished or timed out.
     self.test_state.stop_running_phase()

openhtf/core/test_descriptor.py

@@ -20,6 +20,7 @@
 
 import argparse
 import collections
+import copy
 import logging
 import os
 import sys
@@ -589,6 +590,14 @@ def get_measurement(
     Returns:
       an ImmutableMeasurement or None if the measurement cannot be found.
     """
+    # Check the phase this TestApi is bound to first. This keeps lookups
+    # correct from threads started by the phase (which have no running phase of
+    # their own) while other phases execute concurrently.
+    current_measurements = self._running_phase_state.measurements
+    if measurement_name in current_measurements:
+      return htf_measurements.ImmutableMeasurement.from_measurement(
+          current_measurements[measurement_name]
+      )
     return self._running_test_state.get_measurement(measurement_name)
 
   def get_measurement_strict(
@@ -608,7 +617,7 @@ def get_measurement_strict(
     Raises:
       MeasurementNotFoundError: Thrown when the test measurement is not found.
     """
-    measurement = self._running_test_state.get_measurement(measurement_name)
+    measurement = self.get_measurement(measurement_name)
     if measurement is None:
       raise MeasurementNotFoundError(
           f'Failed to find test measurement {measurement_name}')
@@ -627,6 +636,10 @@ def get_attachment(
     Returns:
       A copy of the attachment or None if the attachment cannot be found.
     """
+    # See get_measurement() for why the bound phase is checked first.
+    current_attachments = self._running_phase_state.phase_record.attachments
+    if attachment_name in current_attachments:
+      return copy.deepcopy(current_attachments[attachment_name])
     return self._running_test_state.get_attachment(attachment_name)
 
   def get_attachment_strict(