From f9d97dc33e251050087350b70996b1780f06764b Mon Sep 17 00:00:00 2001 From: GabyUnalaq Date: Thu, 10 Sep 2026 14:25:05 +0300 Subject: [PATCH] Fix race conditions --- ankaios_sdk/_components/control_interface.py | 260 ++++++++++--------- ankaios_sdk/ankaios.py | 10 +- tests/test_control_interface.py | 99 ++++++- 3 files changed, 234 insertions(+), 135 deletions(-) diff --git a/ankaios_sdk/_components/control_interface.py b/ankaios_sdk/_components/control_interface.py index d7a2173..7608eee 100644 --- a/ankaios_sdk/_components/control_interface.py +++ b/ankaios_sdk/_components/control_interface.py @@ -51,7 +51,6 @@ import os import select -import time import threading from typing import Callable from enum import Enum @@ -100,6 +99,9 @@ class ControlInterface: ANKAIOS_CONTROL_INTERFACE_BASE_PATH = DEFAULT_CONTROL_INTERFACE_PATH "(str): The base path for the Ankaios control interface." + AGENT_RECONNECT_INTERVAL_SEC = 1 + "(int): Seconds to wait between hello retries while the agent is gone." + def __init__( self, add_response_callback: Callable, @@ -124,8 +126,17 @@ def __init__( self._output_file = None # The state of the control interface must not be changed directly. # Use the change_state method instead. - self._state_value = ControlInterfaceState.TERMINATED + self._state = ControlInterfaceState.TERMINATED self._state_lock = threading.Lock() + # Serializes connect() / disconnect() so the lifecycle transitions + # and the resources they own cannot interleave. + self._lifecycle_lock = threading.Lock() + # Guards the output file handle and every write to it, so a write + # can never race the handle being closed or another writer. + self._write_lock = threading.Lock() + # Serializes _cleanup() when the reader thread and a caller both + # reach it. + self._cleanup_lock = threading.Lock() self._read_thread = None self._disconnect_event = threading.Event() @@ -135,28 +146,6 @@ def __init__( self._logger = get_logger() - @property - def _state(self) -> ControlInterfaceState: - """ - Get the current state of the control interface. - - :returns: The current state. - :rtype: ControlInterfaceState - """ - with self._state_lock: - return self._state_value - - @_state.setter - def _state(self, value: ControlInterfaceState) -> None: - """ - Set the current state of the control interface. - - :param value: The new state to set. - :type value: ControlInterfaceState - """ - with self._state_lock: - self._state_value = value - @property def connected(self) -> bool: """ @@ -174,79 +163,95 @@ def connect(self) -> None: :raises ControlInterfaceException: If an error occurred. """ - if self._state in [ - ControlInterfaceState.INITIALIZED, - ControlInterfaceState.CONNECTED, - ]: - raise ControlInterfaceException("Already connected.") - - if not os.path.exists( - f"{self.ANKAIOS_CONTROL_INTERFACE_BASE_PATH}/input" - ): - raise ControlInterfaceException( - "Control interface input fifo does not exist." - ) + with self._lifecycle_lock: + if self._state in [ + ControlInterfaceState.INITIALIZED, + ControlInterfaceState.CONNECTED, + ]: + raise ControlInterfaceException("Already connected.") - if not os.path.exists( - f"{self.ANKAIOS_CONTROL_INTERFACE_BASE_PATH}/output" - ): - raise ControlInterfaceException( - "Control interface output fifo does not exist." - ) + if not os.path.exists( + f"{self.ANKAIOS_CONTROL_INTERFACE_BASE_PATH}/input" + ): + raise ControlInterfaceException( + "Control interface input fifo does not exist." + ) - # pylint: disable=consider-using-with - try: - self._output_file = open( - f"{self.ANKAIOS_CONTROL_INTERFACE_BASE_PATH}/output", "ab" + if not os.path.exists( + f"{self.ANKAIOS_CONTROL_INTERFACE_BASE_PATH}/output" + ): + raise ControlInterfaceException( + "Control interface output fifo does not exist." + ) + + # pylint: disable=consider-using-with + try: + self._output_file = open( + f"{self.ANKAIOS_CONTROL_INTERFACE_BASE_PATH}/output", + "ab", + ) + except Exception as e: + self._logger.error( + "Error while opening output fifo: %s", e + ) + raise ControlInterfaceException( + "Error while opening output fifo." + ) from e + + # A previous disconnect leaves the event set; clear it so the + # new reader thread is not stopped immediately. + self._disconnect_event.clear() + self._read_thread = threading.Thread( + target=self._read_from_control_interface, daemon=True ) - except Exception as e: - self._logger.error("Error while opening output fifo: %s", e) - raise ControlInterfaceException( - "Error while opening output fifo." - ) from e + self._read_thread.start() + self.change_state(ControlInterfaceState.INITIALIZED) - self._read_thread = threading.Thread( - target=self._read_from_control_interface, daemon=True - ) - self._read_thread.start() - self.change_state(ControlInterfaceState.INITIALIZED) + # Released the lifecycle lock: the hello write can block on the + # fifo and must not hold off a concurrent disconnect(). self._send_initial_hello() def disconnect(self) -> None: """ Disconnect from the control interface. """ - if self._state not in [ - ControlInterfaceState.INITIALIZED, - ControlInterfaceState.CONNECTED, - ]: - self._logger.debug("Already disconnected.") - return - - self._logger.debug("Disconnecting..") - self._disconnect_event.set() - if self._read_thread is not None: - self._read_thread.join(timeout=2) - if self._read_thread.is_alive(): - self._logger.error("Read thread did not stop.") - self._read_thread = None - self._cleanup() + with self._lifecycle_lock: + if self._state not in [ + ControlInterfaceState.INITIALIZED, + ControlInterfaceState.CONNECTED, + ]: + self._logger.debug("Already disconnected.") + return + + self._logger.debug("Disconnecting..") + self._disconnect_event.set() + if self._read_thread is not None: + self._read_thread.join(timeout=2) + if self._read_thread.is_alive(): + self._logger.error("Read thread did not stop.") + self._read_thread = None + self._cleanup() def _cleanup(self) -> None: """ Clean up the resources. + + Safe to call from both the reader thread and a caller thread: the + lock serializes the two and the None checks make a repeat call a + no-op. """ - self.change_state(ControlInterfaceState.TERMINATED) - if self._output_file is not None: - self._output_file.close() - self._output_file = None - # The input file will be closed by the reading thread. - # If the thread gets terminated or it's stuck, the input file - # will be closed here. No cover because it's an exceptional case. - if self._input_file is not None: # pragma: no cover - self._input_file.close() - self._input_file = None - self._logger.debug("Cleanup happened") + with self._cleanup_lock: + self.change_state(ControlInterfaceState.TERMINATED) + with self._write_lock: + if self._output_file is not None: + self._output_file.close() + self._output_file = None + # Normally the reader thread reaches this via its finally block; + # a stuck or already-gone thread lets a caller close it instead. + if self._input_file is not None: + self._input_file.close() + self._input_file = None + self._logger.debug("Cleanup happened") def change_state( self, state: ControlInterfaceState, info: str = None @@ -259,17 +264,18 @@ def change_state( :param info: Additional information about the state change. :type info: str """ - if state == self._state: - self._logger.debug("State is already %s.", state) - return - if self._state == ControlInterfaceState.CONNECTION_CLOSED: - self._logger.debug("State CONNECTION_CLOSED is unrecoverable.") - return - self._state = state - if info is None: - self._logger.debug("State changed to %s.", state) - else: - self._logger.debug("State changed to %s: %s", state, info) + with self._state_lock: + if state == self._state: + self._logger.debug("State is already %s.", state) + return + if self._state == ControlInterfaceState.CONNECTION_CLOSED: + self._logger.debug("State CONNECTION_CLOSED is unrecoverable.") + return + self._state = state + if info is None: + self._logger.debug("State changed to %s.", state) + else: + self._logger.debug("State changed to %s: %s", state, info) # pylint: disable=too-many-statements, too-many-branches def _read_from_control_interface(self) -> None: @@ -295,7 +301,11 @@ def _read_from_control_interface(self) -> None: ) except Exception as e: self._logger.error("Error while opening input fifo: %s", e) - self.disconnect() + # Runs on the reader thread itself, so it must tear down + # directly instead of calling disconnect() (which would join + # the current thread). + self._disconnect_event.set() + self._cleanup() raise ControlInterfaceException( "Error while opening input fifo." ) from e @@ -352,8 +362,8 @@ def _read_from_control_interface(self) -> None: except Exception as e: # pylint: disable=broad-exception-caught self._logger.error("Error while reading fifo file: %s", e) finally: - self._input_file.close() - self._input_file = None + # _cleanup() closes the input file under its lock so it cannot + # race a concurrent disconnect() touching the same handle. self._cleanup() def _handle_response(self, response: Response) -> None: @@ -442,15 +452,20 @@ def _agent_gone_routine(self) -> None: """ Method will be called when the agent is gone. It will attempt to write the hello message to the agent - until the agent is connected. + until the agent is connected or a disconnect is requested. """ - agent_reconnect_interval = 1 # seconds - while self._state == ControlInterfaceState.AGENT_DISCONNECTED: + while ( + not self._disconnect_event.is_set() + and self._state == ControlInterfaceState.AGENT_DISCONNECTED + ): try: self._send_initial_hello() except BrokenPipeError as _: self._logger.warning("Waiting for the agent..") - time.sleep(agent_reconnect_interval) + # Wait on the event so a disconnect wakes us immediately. + self._disconnect_event.wait( + self.AGENT_RECONNECT_INTERVAL_SEC + ) else: self.change_state(ControlInterfaceState.INITIALIZED) break @@ -465,19 +480,23 @@ def _write_to_pipe(self, to_ankaios: _control_api.ToAnkaios) -> None: :raises ControlInterfaceException: If the output pipe is None. """ - if self._output_file is None: - self._logger.error( - "Could not write to pipe, output file handler is None." - ) - raise ControlInterfaceException( - "Could not write to pipe, output file handler is None." - ) + # Held across the whole write so the length prefix and the payload + # cannot be split by another writer, and so the handle cannot be + # closed mid-write by _cleanup(). + with self._write_lock: + if self._output_file is None: + self._logger.error( + "Could not write to pipe, output file handler is None." + ) + raise ControlInterfaceException( + "Could not write to pipe, output file handler is None." + ) - # Adds the byte length of the proto msg - self._output_file.write(_VarintBytes(to_ankaios.ByteSize())) - # Adds the proto msg itself - self._output_file.write(to_ankaios.SerializeToString()) - self._output_file.flush() + # Adds the byte length of the proto msg + self._output_file.write(_VarintBytes(to_ankaios.ByteSize())) + # Adds the proto msg itself + self._output_file.write(to_ankaios.SerializeToString()) + self._output_file.flush() def write_request(self, request: Request) -> None: """ @@ -489,14 +508,15 @@ def write_request(self, request: Request) -> None: :raises ControlInterfaceException: If not connected. :raises ConnectionClosedException: If the connection is closed. """ - if self._state == ControlInterfaceState.CONNECTION_CLOSED: - raise ConnectionClosedException( - "Could not write to pipe, connection closed." - ) - if self._state != ControlInterfaceState.CONNECTED: - raise ControlInterfaceException( - "Could not write to pipe, not connected." - ) + with self._state_lock: + if self._state == ControlInterfaceState.CONNECTION_CLOSED: + raise ConnectionClosedException( + "Could not write to pipe, connection closed." + ) + if self._state != ControlInterfaceState.CONNECTED: + raise ControlInterfaceException( + "Could not write to pipe, not connected." + ) request_to_ankaios = _control_api.ToAnkaios( request=request._to_proto() diff --git a/ankaios_sdk/ankaios.py b/ankaios_sdk/ankaios.py index bd42d3c..b0e9b4f 100644 --- a/ankaios_sdk/ankaios.py +++ b/ankaios_sdk/ankaios.py @@ -287,9 +287,10 @@ def _add_logs(self, request_id: str, logs: list[LogResponse]) -> None: :param logs: The received logs. :type logs: list[`LogResponse`] """ - if request_id in self._logs_callbacks: + callback = self._logs_callbacks.get(request_id) + if callback is not None: for log in logs: - self._logs_callbacks[request_id](log) + callback(log) else: self.logger.warning( "Received logs for unknown request id %s", request_id @@ -305,8 +306,9 @@ def _add_events(self, request_id: str, event: EventEntry) -> None: :param event: The event entry. :type event: EventEntry """ - if request_id in self._events_callbacks: - self._events_callbacks[request_id](event) + callback = self._events_callbacks.get(request_id) + if callback is not None: + callback(event) else: self.logger.warning( "Received event with unknown request id %s", request_id diff --git a/tests/test_control_interface.py b/tests/test_control_interface.py index c91d2a5..5f31ee3 100644 --- a/tests/test_control_interface.py +++ b/tests/test_control_interface.py @@ -158,6 +158,55 @@ def test_connection(): ci._logger.debug.assert_called_with("Already disconnected.") +def test_connect_clears_disconnect_event(): + """ + connect() must clear a disconnect_event left set by a prior disconnect, + otherwise the freshly started reader thread would stop immediately. + """ + ci = ControlInterface( + add_response_callback=lambda _: None, + add_log_callback=lambda _: None, + add_event_callback=lambda _: None, + ) + ci._disconnect_event.set() + with patch("os.path.exists", return_value=True), patch( + "threading.Thread" + ) as mock_thread, patch("builtins.open"), patch( + "ankaios_sdk.ControlInterface._send_initial_hello" + ): + mock_thread.return_value = MagicMock() + ci.connect() + + assert not ci._disconnect_event.is_set() + assert ci._state == ControlInterfaceState.INITIALIZED + + +def test_cleanup_is_idempotent(): + """ + _cleanup() may be reached by both the reader thread and a caller; + a repeat call must not raise or double-close the handles. + """ + ci = ControlInterface( + add_response_callback=lambda _: None, + add_log_callback=lambda _: None, + add_event_callback=lambda _: None, + ) + ci._state = ControlInterfaceState.CONNECTED + output_file_mock = MagicMock() + input_file_mock = MagicMock() + ci._output_file = output_file_mock + ci._input_file = input_file_mock + + ci._cleanup() + ci._cleanup() + + output_file_mock.close.assert_called_once() + input_file_mock.close.assert_called_once() + assert ci._output_file is None + assert ci._input_file is None + assert ci._state == ControlInterfaceState.TERMINATED + + def test_read_thread_general(): """ Test the _read_from_control_interface method of the ControlInterface class. @@ -169,8 +218,8 @@ def test_read_thread_general(): # Test error while opening input pipe with patch("builtins.open", side_effect=OSError), patch( - "ankaios_sdk.ControlInterface.disconnect" - ) as mock_disconnect: + "ankaios_sdk.ControlInterface._cleanup" + ) as mock_cleanup: ci = ControlInterface( add_response_callback=lambda _: None, add_log_callback=lambda _: None, @@ -180,7 +229,10 @@ def test_read_thread_general(): ControlInterfaceException, match="Error while opening input fifo" ): ci._read_from_control_interface() - mock_disconnect.assert_called_once() + # The reader thread tears down directly instead of calling + # disconnect() (which would join the current thread). + mock_cleanup.assert_called_once() + assert ci._disconnect_event.is_set() # Test success with patch("builtins.open", mock_open()) as mock_file, patch( @@ -483,11 +535,9 @@ def test_agent_gone_routine(): mock_initial_hello.assert_not_called() ci._state = ControlInterfaceState.AGENT_DISCONNECTED - original_sleep = time.sleep - with patch( - "time.sleep", - new=lambda x: original_sleep(x) if x != 1 else original_sleep(0.01), - ) as _, patch( + with patch.object( + ControlInterface, "AGENT_RECONNECT_INTERVAL_SEC", 0.01 + ), patch( "ankaios_sdk.ControlInterface._send_initial_hello" ) as mock_initial_hello: @@ -497,15 +547,42 @@ def test_agent_gone_routine(): target=ci._agent_gone_routine, daemon=True ) agent_gone_thread.start() - time.sleep(0.01) + time.sleep(0.05) mock_initial_hello.side_effect = None - time.sleep(0.01) - agent_gone_thread.join() + time.sleep(0.05) + agent_gone_thread.join(timeout=2) + assert not agent_gone_thread.is_alive() mock_initial_hello.assert_called() assert ci._state == ControlInterfaceState.INITIALIZED +def test_agent_gone_routine_stops_on_disconnect(): + """ + _agent_gone_routine must exit promptly when a disconnect is requested, + even while the agent stays gone. + """ + ci = ControlInterface( + add_response_callback=lambda _: None, + add_log_callback=lambda _: None, + add_event_callback=lambda _: None, + ) + ci._state = ControlInterfaceState.AGENT_DISCONNECTED + with patch( + "ankaios_sdk.ControlInterface._send_initial_hello", + side_effect=BrokenPipeError, + ): + agent_gone_thread = threading.Thread( + target=ci._agent_gone_routine, daemon=True + ) + agent_gone_thread.start() + ci._disconnect_event.set() + agent_gone_thread.join(timeout=2) + + assert not agent_gone_thread.is_alive() + assert ci._state == ControlInterfaceState.AGENT_DISCONNECTED + + def test_write_to_pipe(): """ Test the _write_to_pipe method of the ControlInterface class.