diff --git a/ankaios_sdk/_components/connection/command_interface.py b/ankaios_sdk/_components/connection/command_interface.py index 1a29914..8e8733d 100644 --- a/ankaios_sdk/_components/connection/command_interface.py +++ b/ankaios_sdk/_components/connection/command_interface.py @@ -49,7 +49,6 @@ import queue import threading -import time from enum import Enum from typing import Callable, Optional @@ -174,17 +173,14 @@ def __init__( # The state of the command interface must not be changed directly. # Use the change_state method instead. self._state = CommandInterfaceState.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 _channel, _call and _write_queue, which are replaced by - # the reconnect loop (reader thread) and read/torn down by - # disconnect() and write_request() (caller thread). - self._stream_lock = threading.Lock() + # Guards _state, _channel, _call, _write_queue and _reader_thread. + # Only connect() holds it across a blocking call. + self._lock = threading.Lock() + # Set by disconnect() to stop the reader thread and its reconnects. + self._stop_event = threading.Event() self._channel: Optional[grpc.Channel] = None self._call = None - self._write_queue: "queue.Queue" = queue.Queue() + self._write_queue: Optional["queue.Queue"] = None self._reader_thread: Optional[threading.Thread] = None @property @@ -205,55 +201,72 @@ def connect(self) -> None: established, a later lost connection is retried transparently every ``RECONNECT_INTERVAL`` seconds. - :raises ConnectionException: If already connected, or if - the connection could not be established. + :raises ConnectionException: If already connected, if the + previous connection is still shutting down, or if the + connection could not be established. """ - with self._lifecycle_lock: - if self._state in ( - CommandInterfaceState.INITIALIZED, - CommandInterfaceState.CONNECTED, - CommandInterfaceState.RECONNECTING, - ): + # Held across the whole attempt (up to CHANNEL_READY_TIMEOUT + # seconds) so a concurrent disconnect() cannot interleave with it. + # No reader thread exists yet that could need the lock. + with self._lock: + if self._state != CommandInterfaceState.TERMINATED: raise ConnectionException("Already connected.") + # Only one reader thread may ever run: a previous one that did + # not stop in time would otherwise reconnect and replace this + # connection's stream. + if ( + self._reader_thread is not None + and self._reader_thread.is_alive() + ): + raise ConnectionException( + "Previous connection is still shutting down." + ) # Only change the state once past the point where connecting # can still fail, so a failed attempt leaves the connection # exactly as it was and free to retry via a plain connect(). - call = self._open_stream() - self.change_state(CommandInterfaceState.INITIALIZED) + channel, call, write_queue = self._open_stream() + self._stop_event.clear() + self._channel = channel + self._call = call + self._write_queue = write_queue + self._set_state(CommandInterfaceState.INITIALIZED) self._reader_thread = threading.Thread( target=self._read_from_grpc, args=(call,), daemon=True ) self._reader_thread.start() - self.change_state(CommandInterfaceState.CONNECTED) + self._set_state(CommandInterfaceState.CONNECTED) self._logger.debug("Connected to the Ankaios server over gRPC.") def disconnect(self) -> None: """ Disconnect from the gRPC connection. """ - with self._lifecycle_lock: + with self._lock: if self._state == CommandInterfaceState.TERMINATED: self._logger.debug("Already disconnected.") return self._logger.debug("Disconnecting..") - self.change_state(CommandInterfaceState.TERMINATED) - with self._stream_lock: - call = self._call - if call is not None: - call.cancel() - if self._reader_thread is not None: - self._reader_thread.join(timeout=2) - if self._reader_thread.is_alive(): - self._logger.error("Reader thread did not stop.") - self._reader_thread = None - with self._stream_lock: - if self._channel is not None: - self._channel.close() - self._channel = None - self._call = None + self._set_state(CommandInterfaceState.TERMINATED) + self._stop_event.set() + channel, self._channel = self._channel, None + call, self._call = self._call, None + self._write_queue = None + # Kept even if it does not stop in time, so connect() can + # refuse to start a second reader while it is still alive. + reader_thread = self._reader_thread + + # The stream is detached, so the slow teardown runs unlocked. + if call is not None: + call.cancel() + if reader_thread is not None: + reader_thread.join(timeout=2) + if reader_thread.is_alive(): + self._logger.error("Reader thread did not stop.") + if channel is not None: + channel.close() def write_request(self, request: Request) -> None: """ @@ -264,7 +277,7 @@ def write_request(self, request: Request) -> None: :raises ConnectionException: If not connected. """ - with self._state_lock: + with self._lock: if self._state != CommandInterfaceState.CONNECTED: self._logger.error( "Could not write to the gRPC connection, not connected." @@ -272,12 +285,8 @@ def write_request(self, request: Request) -> None: raise ConnectionException( "Could not write to the gRPC connection, not connected." ) - # Held so the queue cannot be replaced by a concurrent reconnect - # between reading the reference and putting onto it. - with self._stream_lock: - self._write_queue.put( - _grpc_api.ToServer(request=request._to_proto()) - ) + write_queue = self._write_queue + write_queue.put(_grpc_api.ToServer(request=request._to_proto())) def _read_from_grpc(self, call) -> None: """ @@ -294,7 +303,7 @@ def _read_from_grpc(self, call) -> None: for from_server in call: self._handle_from_server(from_server) except grpc.RpcError as e: - if self._state == CommandInterfaceState.TERMINATED: + if self._stop_event.is_set(): # disconnect() already cancelled the call itself; # this is the expected, self-inflicted result. self._logger.debug( @@ -305,13 +314,12 @@ def _read_from_grpc(self, call) -> None: "Error while reading from the gRPC connection: '%s'", e, ) - else: - # The stream ended without an error; still need to know - # whether that was disconnect()'s doing before retrying. - if self._state == CommandInterfaceState.TERMINATED: - return - self.change_state(CommandInterfaceState.RECONNECTING) + with self._lock: + # The stream may also have ended because of disconnect(). + if self._stop_event.is_set(): + return + self._set_state(CommandInterfaceState.RECONNECTING) self._logger.warning( "Lost connection to the Ankaios server, attempting to " "reconnect every %ss..", @@ -349,18 +357,35 @@ def _reconnect(self): called while reconnecting. :rtype: Optional[grpc.Call] """ - while True: - time.sleep(self.RECONNECT_INTERVAL) - if self._state == CommandInterfaceState.TERMINATED: - return None + # Waits on the event so a disconnect wakes us immediately. + while not self._stop_event.wait(self.RECONNECT_INTERVAL): try: - call = self._open_stream() + channel, call, write_queue = self._open_stream() except ConnectionException as e: self._logger.debug("Reconnect attempt failed: '%s'", e) continue - self.change_state(CommandInterfaceState.CONNECTED) + + old_channel = None + with self._lock: + # disconnect() may have run while the stream was opening; + # it must not be brought back to life then. + stopped = self._stop_event.is_set() + if not stopped: + old_channel = self._channel + self._channel = channel + self._call = call + self._write_queue = write_queue + self._set_state(CommandInterfaceState.CONNECTED) + if stopped: + call.cancel() + channel.close() + return None + if old_channel is not None: + # The channel of the lost connection. + old_channel.close() self._logger.info("Reconnected to the Ankaios server.") return call + return None def change_state(self, state: CommandInterfaceState) -> None: """ @@ -374,32 +399,46 @@ def change_state(self, state: CommandInterfaceState) -> None: :param state: The new state. :type state: CommandInterfaceState """ - with self._state_lock: - if state == self._state: - self._logger.debug("State is already %s.", state) - return - self._state = state - self._logger.debug("State changed to %s.", state) + with self._lock: + self._set_state(state) + + def _set_state(self, state: CommandInterfaceState) -> None: + """ + Change the state of the connection. The caller must hold + the lock. + + :param state: The new state. + :type state: CommandInterfaceState + """ + if state == self._state: + self._logger.debug("State is already %s.", state) + return + self._state = state + self._logger.debug("State changed to %s.", state) def _open_stream(self): """ Builds the gRPC channel, sends the initial CommanderHello and opens the ConnectCommand bidi stream. Used both for the - initial connect and for every reconnect attempt. + initial connect and for every reconnect attempt. The opened + stream is not stored on the connection; the caller does that. - :returns: The opened bidi call, iterable for FromServer - messages. - :rtype: grpc.Call + :returns: The channel, the opened bidi call (iterable for + FromServer messages) and the queue feeding its requests. + :rtype: tuple :raises ConnectionException: If the channel does not become ready in time, or the stream could not be opened. """ + channel = None try: channel = self._build_channel() grpc.channel_ready_future(channel).result( timeout=self.CHANNEL_READY_TIMEOUT ) except (grpc.FutureTimeoutError, grpc.RpcError) as e: + if channel is not None: + channel.close() raise ConnectionException( f"Could not connect to the Ankaios server: '{e}'" ) from e @@ -413,23 +452,19 @@ def _open_stream(self): ) ) ) + call = stub.ConnectCommand(self._request_iterator(write_queue)) + return channel, call, write_queue - # Held across the replacement so disconnect() and write_request() - # never observe a half-replaced write_queue/channel/call. - with self._stream_lock: - self._write_queue = write_queue - call = stub.ConnectCommand(self._request_iterator()) - self._channel = channel - self._call = call - return call - - def _request_iterator(self): + @staticmethod + def _request_iterator(write_queue: "queue.Queue"): """ Generator yielding ToServer messages queued via :func:`write_request` (and the initial hello), until :func:`disconnect` cancels the call. + + :param write_queue: The queue of the stream being fed. + :type write_queue: queue.Queue """ - write_queue = self._write_queue while True: yield write_queue.get() diff --git a/ankaios_sdk/_components/connection/control_interface.py b/ankaios_sdk/_components/connection/control_interface.py index ba9e6fd..bf5c306 100644 --- a/ankaios_sdk/_components/connection/control_interface.py +++ b/ankaios_sdk/_components/connection/control_interface.py @@ -104,6 +104,13 @@ class ControlInterfaceConnection(Connection): AGENT_RECONNECT_INTERVAL_SEC = 1 "(int): Seconds to wait between hello retries while the agent is gone." + _ACTIVE_STATES = ( + ControlInterfaceState.INITIALIZED, + ControlInterfaceState.CONNECTED, + ControlInterfaceState.AGENT_DISCONNECTED, + ) + "(tuple): The states in which a reader thread is running." + def __init__( self, add_response_callback: Callable, @@ -127,21 +134,17 @@ def __init__( super().__init__( add_response_callback, add_log_callback, add_event_callback ) - self._input_file = None - self._output_file = None # The state of the control interface must not be changed directly. # Use the change_state method instead. 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. + # Guards _state, _output_file and _read_thread. Never held across + # a fifo write or a thread join. + self._lock = threading.Lock() + # Held across every framed write and around closing the output + # file, so a write can never be split by another writer or have + # its file closed underneath it. Never taken while holding _lock. self._write_lock = threading.Lock() - # Serializes _cleanup() when the reader thread and a caller both - # reach it. - self._cleanup_lock = threading.Lock() + self._output_file = None self._read_thread = None self._disconnect_event = threading.Event() @@ -162,12 +165,16 @@ def connect(self) -> None: :raises ConnectionException: If an error occurred. """ - with self._lifecycle_lock: - if self._state in [ - ControlInterfaceState.INITIALIZED, - ControlInterfaceState.CONNECTED, - ]: + with self._lock: + if self._state in self._ACTIVE_STATES: raise ConnectionException("Already connected.") + # Only one reader thread may ever run: a previous one that did + # not stop in time would otherwise read the same fifo and tear + # down this connection's resources when it finally exits. + if self._read_thread is not None and self._read_thread.is_alive(): + raise ConnectionException( + "Previous connection is still shutting down." + ) if not os.path.exists( f"{self.ANKAIOS_CONTROL_INTERFACE_BASE_PATH}/input" @@ -200,57 +207,74 @@ def connect(self) -> None: # A previous disconnect leaves the event set; clear it so the # new reader thread is not stopped immediately. self._disconnect_event.clear() + # Set before the reader starts, so a reader that fails right + # away cannot have its TERMINATED overwritten. + self._set_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(). + # Sent without holding the 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. """ - with self._lifecycle_lock: - if self._state not in [ - ControlInterfaceState.INITIALIZED, - ControlInterfaceState.CONNECTED, - ]: + with self._lock: + if self._state not in self._ACTIVE_STATES: 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() + output_file = self._detach_resources() + # Kept even if it does not stop in time, so connect() can + # refuse to start a second reader while it is still alive. + read_thread = self._read_thread + + # The resources are detached, so the slow teardown runs unlocked. + if read_thread is not None: + read_thread.join(timeout=2) + if read_thread.is_alive(): + self._logger.error("Read thread did not stop.") + self._close_output_file(output_file) 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. - """ - 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") + Clean up the resources. Called by the reader thread when it + stops; a no-op if disconnect() already cleaned up. + """ + with self._lock: + output_file = self._detach_resources() + self._close_output_file(output_file) + self._logger.debug("Cleanup happened") + + def _detach_resources(self): + """ + Marks the connection as terminated and detaches the output + file from it. The caller must hold the lock. + + :returns: The detached output file (or None if already + detached), to be closed once the lock is released. + """ + self._set_state(ControlInterfaceState.TERMINATED) + output_file, self._output_file = self._output_file, None + return output_file + + def _close_output_file(self, output_file) -> None: + """ + Closes a detached output file, waiting for any write in + progress on it to finish first. + + :param output_file: The output file to close, or None. + """ + if output_file is None: + return + with self._write_lock: + output_file.close() def change_state( self, state: ControlInterfaceState, info: str = None @@ -263,18 +287,49 @@ def change_state( :param info: Additional information about the state change. :type info: str """ - 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) + with self._lock: + self._set_state(state, info) + + def _change_state_from_reader( + self, state: ControlInterfaceState, info: str = None + ) -> None: + """ + Change the state of the control interface on behalf of the + reader thread, unless a disconnect was requested, so the reader + cannot overwrite the TERMINATED state set by disconnect(). + + :param state: The new state. + :type state: ControlInterfaceState + :param info: Additional information about the state change. + :type info: str + """ + with self._lock: + if not self._disconnect_event.is_set(): + self._set_state(state, info) + + def _set_state( + self, state: ControlInterfaceState, info: str = None + ) -> None: + """ + Change the state of the control interface. The caller must + hold the lock. + + :param state: The new state. + :type state: ControlInterfaceState + :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) # pylint: disable=too-many-statements, too-many-branches def _read_from_control_interface(self) -> None: @@ -282,6 +337,7 @@ def _read_from_control_interface(self) -> None: Reads from the control interface input fifo. This is meant to be run in a separate thread. The responses are then sent to the Ankaios class to be handled. + The input fifo is opened and closed by this thread alone. :raises ConnectionException: If an error occurs while reading the fifo. @@ -295,7 +351,7 @@ def _read_from_control_interface(self) -> None: # pylint: disable=consider-using-with try: - self._input_file = open( + input_file = open( f"{self.ANKAIOS_CONTROL_INTERFACE_BASE_PATH}/input", "rb" ) except Exception as e: @@ -308,14 +364,14 @@ def _read_from_control_interface(self) -> None: raise ConnectionException( "Error while opening input fifo." ) from e - os.set_blocking(self._input_file.fileno(), False) + os.set_blocking(input_file.fileno(), False) try: self._logger.debug("Started reading from the input pipe.") while not self._disconnect_event.is_set(): # The loop continues when data is available or when the # timeout of 1 second is reached. - ready, _, _ = select.select([self._input_file], [], [], 1) + ready, _, _ = select.select([input_file], [], [], 1) if not ready: # pragma: no cover continue @@ -323,7 +379,7 @@ def _read_from_control_interface(self) -> None: varint_buffer = bytearray() while not self._disconnect_event.is_set(): # Consume byte for byte - next_byte = self._input_file.read(1) + next_byte = input_file.read(1) if not next_byte: # pragma: no cover break varint_buffer += next_byte @@ -332,7 +388,9 @@ def _read_from_control_interface(self) -> None: break if not varint_buffer: - self.change_state(ControlInterfaceState.AGENT_DISCONNECTED) + self._change_state_from_reader( + ControlInterfaceState.AGENT_DISCONNECTED + ) self._logger.warning( "Nothing to read from the input fifo pipe." ) @@ -345,7 +403,7 @@ def _read_from_control_interface(self) -> None: msg_buf = bytearray() for _ in range(msg_len): # Read the message according to the length - next_byte = self._input_file.read(1) + next_byte = input_file.read(1) if not next_byte: # pragma: no cover break msg_buf += next_byte @@ -361,8 +419,7 @@ 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: - # _cleanup() closes the input file under its lock so it cannot - # race a concurrent disconnect() touching the same handle. + input_file.close() self._cleanup() @staticmethod @@ -421,10 +478,12 @@ def _handle_response(self, response: Response) -> None: self._logger.debug( "Received control interface accepted response." ) - self.change_state(ControlInterfaceState.CONNECTED) + self._change_state_from_reader( + ControlInterfaceState.CONNECTED + ) elif response.content_type == ResponseType.CONNECTION_CLOSED: self._add_response_callback(response) - self.change_state( + self._change_state_from_reader( ControlInterfaceState.CONNECTION_CLOSED, response.content, ) @@ -444,7 +503,7 @@ def _handle_response(self, response: Response) -> None: # Check if the response is connection closed in order to # terminate the thread. if response.content_type == ResponseType.CONNECTION_CLOSED: - self.change_state( + self._change_state_from_reader( ControlInterfaceState.CONNECTION_CLOSED, response.content, ) @@ -483,7 +542,9 @@ def _agent_gone_routine(self) -> None: self.AGENT_RECONNECT_INTERVAL_SEC ) else: - self.change_state(ControlInterfaceState.INITIALIZED) + self._change_state_from_reader( + ControlInterfaceState.INITIALIZED + ) break def _write_to_pipe(self, to_ankaios: _control_api.ToAnkaios) -> None: @@ -494,25 +555,34 @@ def _write_to_pipe(self, to_ankaios: _control_api.ToAnkaios) -> None: :param to_ankaios: The ToAnkaios proto message. :type to_ankaios: _control_api.ToAnkaios - :raises ConnectionException: If the output pipe is None. + :raises ConnectionException: If the output pipe is None, or + was closed by a disconnect before the write. """ + with self._lock: + output_file = self._output_file + if output_file is None: + self._logger.error( + "Could not write to pipe, output file handler is None." + ) + raise ConnectionException( + "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(). + # cannot be split by another writer, and so the file cannot be + # closed mid-write. with self._write_lock: - if self._output_file is None: - self._logger.error( - "Could not write to pipe, output file handler is None." - ) + try: + # Adds the byte length of the proto msg + output_file.write(_VarintBytes(to_ankaios.ByteSize())) + # Adds the proto msg itself + output_file.write(to_ankaios.SerializeToString()) + output_file.flush() + except ValueError as e: + # A disconnect closed the file after we took it. raise ConnectionException( - "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() + "Could not write to pipe, output file closed." + ) from e def write_request(self, request: Request) -> None: """ @@ -524,7 +594,7 @@ def write_request(self, request: Request) -> None: :raises ConnectionException: If not connected. :raises ConnectionClosedException: If the connection is closed. """ - with self._state_lock: + with self._lock: if self._state == ControlInterfaceState.CONNECTION_CLOSED: raise ConnectionClosedException( "Could not write to pipe, connection closed." diff --git a/examples/apps/follow_state.py b/examples/apps/follow_state.py index 32ec191..d851e62 100644 --- a/examples/apps/follow_state.py +++ b/examples/apps/follow_state.py @@ -12,9 +12,9 @@ # # SPDX-License-Identifier: Apache-2.0 -from ankaios_sdk import Ankaios, ConnectionException, Workload +from ankaios_sdk import Ankaios, ConnectionException from time import sleep -import sys, signal +import signal # Create a new Ankaios object. # The connection to the control interface is automatically done at this step. diff --git a/examples/apps/test_files.py b/examples/apps/test_files.py index f408159..6dea951 100644 --- a/examples/apps/test_files.py +++ b/examples/apps/test_files.py @@ -1,3 +1,17 @@ +# Copyright (c) 2025 Elektrobit Automotive GmbH +# +# This program and the accompanying materials are made available under the +# terms of the Apache License, Version 2.0 which is available at +# https://www.apache.org/licenses/LICENSE-2.0. +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. +# +# SPDX-License-Identifier: Apache-2.0 + import time from ankaios_sdk import ( Ankaios, diff --git a/tests/connection/test_command_interface.py b/tests/connection/test_command_interface.py index 5cc73af..4389d63 100644 --- a/tests/connection/test_command_interface.py +++ b/tests/connection/test_command_interface.py @@ -17,6 +17,7 @@ ankaios_sdk. """ +import queue import threading import time from unittest.mock import MagicMock, patch @@ -257,7 +258,8 @@ def test_connect_write_and_disconnect_success(): conn.disconnect() assert conn.connected is False assert fake_call.cancelled.is_set() - assert conn._reader_thread is None + # Kept so a later connect() can check it has stopped. + assert not conn._reader_thread.is_alive() # Disconnecting again is a no-op. conn.disconnect() @@ -282,6 +284,27 @@ def test_open_stream_channel_not_ready_raises(): conn._open_stream() +def test_open_stream_channel_not_ready_closes_channel(): + """ + Test that _open_stream() closes the channel it built when the + channel does not become ready in time, so failed (re)connect + attempts do not leak channels. + """ + with patch("grpc.insecure_channel") as mock_channel, patch( + "grpc.channel_ready_future" + ) as mock_ready: + channel = MagicMock() + mock_channel.return_value = channel + mock_ready.return_value = MagicMock( + result=MagicMock(side_effect=grpc.FutureTimeoutError()) + ) + conn = _generate_test_connection() + with pytest.raises(ConnectionException): + conn._open_stream() + + channel.close.assert_called_once() + + def test_build_channel_insecure(): """ Test that _build_channel() builds a plaintext channel when no @@ -344,11 +367,11 @@ def test_request_iterator(): Test that _request_iterator() yields exactly the queued messages, in order. """ - conn = _generate_test_connection() - conn._write_queue.put("message_1") - conn._write_queue.put("message_2") + write_queue = queue.Queue() + write_queue.put("message_1") + write_queue.put("message_2") - iterator = conn._request_iterator() + iterator = CommandInterfaceConnection._request_iterator(write_queue) assert next(iterator) == "message_1" assert next(iterator) == "message_2" @@ -457,15 +480,15 @@ def test_handle_from_server_ignores_non_response_messages(): conn._add_event_callback.assert_not_called() -def test_read_from_grpc_stops_on_terminated_state(): +def test_read_from_grpc_stops_on_disconnect(): """ Test that the reader loop returns without attempting a reconnect - if the connection was terminated (disconnect() was called), and - logs the resulting RpcError at debug level, since it's the - expected result of disconnect()'s own call.cancel(). + if disconnect() was called, and logs the resulting RpcError at + debug level, since it's the expected result of disconnect()'s own + call.cancel(). """ conn = _generate_test_connection() - conn._state = CommandInterfaceState.TERMINATED + conn._stop_event.set() conn._logger = MagicMock() fake_call = _FakeCall(error=grpc.RpcError("boom")) @@ -493,7 +516,11 @@ def test_read_from_grpc_reconnects_on_lost_connection(): with patch.object( CommandInterfaceConnection, "RECONNECT_INTERVAL", 0.01 - ), patch.object(conn, "_open_stream", return_value=second_call): + ), patch.object( + conn, + "_open_stream", + return_value=(MagicMock(), second_call, queue.Queue()), + ): reader_thread = threading.Thread( target=conn._read_from_grpc, args=(first_call,), daemon=True ) @@ -509,18 +536,20 @@ def test_read_from_grpc_reconnects_on_lost_connection(): "State changed to %s.", CommandInterfaceState.CONNECTED ) + conn._stop_event.set() second_call.cancel() reader_thread.join(timeout=1) + assert not reader_thread.is_alive() -def test_reconnect_gives_up_when_terminated(): +def test_reconnect_gives_up_when_disconnected(): """ Test that _reconnect() gives up (returns None) as soon as it - notices the connection was terminated, without attempting to - reopen the stream. + notices disconnect() was called, without attempting to reopen + the stream. """ conn = _generate_test_connection() - conn._state = CommandInterfaceState.TERMINATED + conn._stop_event.set() with patch.object( CommandInterfaceConnection, "RECONNECT_INTERVAL", 0.01 @@ -537,9 +566,14 @@ def test_reconnect_retries_until_success(): """ conn = _generate_test_connection() conn._state = CommandInterfaceState.RECONNECTING + channel = MagicMock() fake_call = _FakeCall() + write_queue = queue.Queue() - attempts = [ConnectionException("still down"), fake_call] + attempts = [ + ConnectionException("still down"), + (channel, fake_call, write_queue), + ] def _fake_open_stream(): attempt = attempts.pop(0) @@ -554,6 +588,67 @@ def _fake_open_stream(): assert result is fake_call assert conn._state == CommandInterfaceState.CONNECTED + assert conn._channel is channel + assert conn._call is fake_call + assert conn._write_queue is write_queue + channel.close.assert_not_called() + + +def test_reconnect_closes_replaced_channel(): + """ + Test that a successful reconnect closes the channel of the lost + connection it replaces. + """ + conn = _generate_test_connection() + conn._state = CommandInterfaceState.RECONNECTING + old_channel = MagicMock() + conn._channel = old_channel + new_channel = MagicMock() + + with patch.object( + CommandInterfaceConnection, "RECONNECT_INTERVAL", 0.01 + ), patch.object( + conn, + "_open_stream", + return_value=(new_channel, _FakeCall(), queue.Queue()), + ): + conn._reconnect() + + old_channel.close.assert_called_once() + new_channel.close.assert_not_called() + assert conn._channel is new_channel + + +def test_reconnect_discarded_when_disconnected_while_opening(): + """ + Test that a stream opened by _reconnect() is discarded (call + cancelled, channel closed) instead of being stored when + disconnect() was called while it was being opened, so the + connection is not brought back to life. + """ + conn = _generate_test_connection() + conn._state = CommandInterfaceState.RECONNECTING + channel = MagicMock() + fake_call = _FakeCall() + + def _open_stream_racing_disconnect(): + conn._state = CommandInterfaceState.TERMINATED + conn._stop_event.set() + return channel, fake_call, queue.Queue() + + with patch.object( + CommandInterfaceConnection, "RECONNECT_INTERVAL", 0.01 + ), patch.object( + conn, "_open_stream", side_effect=_open_stream_racing_disconnect + ): + result = conn._reconnect() + + assert result is None + assert fake_call.cancelled.is_set() + channel.close.assert_called_once() + assert conn._call is None + assert conn._channel is None + assert conn._state == CommandInterfaceState.TERMINATED def test_read_from_grpc_dispatches_messages(): @@ -572,7 +667,7 @@ def test_read_from_grpc_dispatches_messages(): reader_thread.start() time.sleep(0.05) - conn._state = CommandInterfaceState.TERMINATED + conn._stop_event.set() fake_call.cancel() reader_thread.join(timeout=1) assert not reader_thread.is_alive() @@ -591,6 +686,23 @@ def test_read_from_grpc_stops_when_reconnect_gives_up(): conn._read_from_grpc(fake_call) +def test_read_from_grpc_stream_end_after_disconnect(): + """ + Test that the reader loop returns without attempting a reconnect + or changing the state when the stream ends without an error after + disconnect() was called. + """ + conn = _generate_test_connection() + conn._stop_event.set() + fake_call = _FakeCall() + fake_call.cancel() + + with patch.object(conn, "_reconnect") as mock_reconnect: + conn._read_from_grpc(fake_call) + mock_reconnect.assert_not_called() + assert conn._state == CommandInterfaceState.TERMINATED + + def test_disconnect_logs_error_when_reader_thread_does_not_stop(): """ Test that disconnect() logs an error if the reader thread does @@ -622,3 +734,25 @@ def test_disconnect_logs_error_when_reader_thread_does_not_stop(): conn._logger.error.assert_called_once_with( "Reader thread did not stop." ) + # Kept so a following connect() is refused. + assert conn._reader_thread is mock_thread_instance + + +def test_connect_refuses_while_previous_reader_alive(): + """ + Test that connect() refuses to start while the reader thread of a + previous connection is still alive, since it could otherwise + reconnect and replace the new connection's stream. + """ + conn = _generate_test_connection() + conn._reader_thread = MagicMock() + conn._reader_thread.is_alive.return_value = True + + with patch.object(conn, "_open_stream") as mock_open_stream, pytest.raises( + ConnectionException, + match="Previous connection is still shutting down.", + ): + conn.connect() + + mock_open_stream.assert_not_called() + assert conn._state == CommandInterfaceState.TERMINATED diff --git a/tests/connection/test_control_interface.py b/tests/connection/test_control_interface.py index 9f094f3..6f57115 100644 --- a/tests/connection/test_control_interface.py +++ b/tests/connection/test_control_interface.py @@ -148,10 +148,10 @@ def test_connection(): assert ci._disconnect_event.is_set() assert ci._state == ControlInterfaceState.TERMINATED mock_thread_instance.join.assert_called_once() - assert ci._read_thread is None + # Kept so a later connect() can check it has stopped. + assert ci._read_thread is mock_thread_instance output_file_mock.close.assert_called_once() assert ci._output_file is None - assert ci._input_file is None # Test disconnect while not connected ci._logger = MagicMock() @@ -183,10 +183,163 @@ def test_connect_clears_disconnect_event(): assert ci._state == ControlInterfaceState.INITIALIZED +def test_connect_refuses_while_previous_reader_alive(): + """ + connect() must refuse to start a second reader thread while the one + of a previous connection has not stopped yet, since both would read + the same fifo. + """ + ci = ControlInterfaceConnection( + add_response_callback=lambda _: None, + add_log_callback=lambda _: None, + add_event_callback=lambda _: None, + ) + ci._read_thread = MagicMock() + ci._read_thread.is_alive.return_value = True + with patch("os.path.exists", return_value=True), patch( + "builtins.open" + ) as mock_open_file, pytest.raises( + ConnectionException, + match="Previous connection is still shutting down.", + ): + ci.connect() + + mock_open_file.assert_not_called() + assert ci._state == ControlInterfaceState.TERMINATED + + +def test_connect_after_previous_reader_stopped(): + """ + connect() succeeds once the reader thread of a previous connection + has stopped. + """ + ci = ControlInterfaceConnection( + add_response_callback=lambda _: None, + add_log_callback=lambda _: None, + add_event_callback=lambda _: None, + ) + ci._read_thread = MagicMock() + ci._read_thread.is_alive.return_value = False + with patch("os.path.exists", return_value=True), patch( + "threading.Thread" + ) as mock_thread, patch("builtins.open"), patch( + "ankaios_sdk.ControlInterfaceConnection._send_initial_hello" + ): + mock_thread.return_value = MagicMock() + ci.connect() + + assert ci._read_thread is mock_thread.return_value + assert ci._state == ControlInterfaceState.INITIALIZED + + +def test_connect_while_agent_disconnected_raises(): + """ + While the agent is gone the reader thread is still running, so + connect() must treat the connection as already connected. + """ + ci = ControlInterfaceConnection( + add_response_callback=lambda _: None, + add_log_callback=lambda _: None, + add_event_callback=lambda _: None, + ) + ci._state = ControlInterfaceState.AGENT_DISCONNECTED + with pytest.raises(ConnectionException, match="Already connected."): + ci.connect() + + +def test_disconnect_while_agent_disconnected(): + """ + disconnect() must stop the reader thread and release the output + file while the agent is gone, too. + """ + ci = ControlInterfaceConnection( + add_response_callback=lambda _: None, + add_log_callback=lambda _: None, + add_event_callback=lambda _: None, + ) + ci._state = ControlInterfaceState.AGENT_DISCONNECTED + read_thread = MagicMock() + read_thread.is_alive.return_value = False + output_file_mock = MagicMock() + ci._read_thread = read_thread + ci._output_file = output_file_mock + + ci.disconnect() + + assert ci._disconnect_event.is_set() + read_thread.join.assert_called_once_with(timeout=2) + output_file_mock.close.assert_called_once() + assert ci._output_file is None + assert ci._state == ControlInterfaceState.TERMINATED + + +def test_disconnect_keeps_read_thread_that_did_not_stop(): + """ + If the reader thread does not stop in time, disconnect() logs an + error and keeps it, so a following connect() is refused. + """ + ci = ControlInterfaceConnection( + add_response_callback=lambda _: None, + add_log_callback=lambda _: None, + add_event_callback=lambda _: None, + ) + ci._logger = MagicMock() + ci._state = ControlInterfaceState.CONNECTED + read_thread = MagicMock() + read_thread.is_alive.return_value = True + ci._read_thread = read_thread + + ci.disconnect() + + ci._logger.error.assert_called_once_with("Read thread did not stop.") + assert ci._read_thread is read_thread + assert ci._state == ControlInterfaceState.TERMINATED + with pytest.raises( + ConnectionException, + match="Previous connection is still shutting down.", + ): + ci.connect() + + +def test_change_state_from_reader(): + """ + The reader thread may change the state while no disconnect was + requested. + """ + ci = ControlInterfaceConnection( + add_response_callback=lambda _: None, + add_log_callback=lambda _: None, + add_event_callback=lambda _: None, + ) + ci._state = ControlInterfaceState.CONNECTED + + ci._change_state_from_reader(ControlInterfaceState.AGENT_DISCONNECTED) + + assert ci._state == ControlInterfaceState.AGENT_DISCONNECTED + + +def test_change_state_from_reader_ignored_after_disconnect(): + """ + Once a disconnect was requested, the reader thread must not + overwrite the TERMINATED state set by disconnect(). + """ + ci = ControlInterfaceConnection( + add_response_callback=lambda _: None, + add_log_callback=lambda _: None, + add_event_callback=lambda _: None, + ) + ci._state = ControlInterfaceState.TERMINATED + ci._disconnect_event.set() + + ci._change_state_from_reader(ControlInterfaceState.AGENT_DISCONNECTED) + + assert ci._state == ControlInterfaceState.TERMINATED + + 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. + a repeat call must not raise or double-close the output file. """ ci = ControlInterfaceConnection( add_response_callback=lambda _: None, @@ -195,17 +348,13 @@ def test_cleanup_is_idempotent(): ) 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 @@ -299,6 +448,7 @@ def test_read_thread_general(): "/run/ankaios/control_interface/input", "rb" ) mock_handle_response.assert_called_once() + mock_file.return_value.close.assert_called_once() def test_read_thread_agent_disconnected(): @@ -384,7 +534,8 @@ def test_read_thread_connection_closed(): mock_file.assert_called_once_with( "/run/ankaios/control_interface/input", "rb" ) - assert ci._input_file is None + mock_file.return_value.close.assert_called_once() + assert ci._state == ControlInterfaceState.TERMINATED def test_handle_response(): @@ -621,22 +772,45 @@ def test_write_to_pipe(): add_log_callback=lambda _: None, add_event_callback=lambda _: None, ) + request = _control_api.FromAnkaios() ci._output_file = None with pytest.raises( ConnectionException, match="Could not write to pipe" ): - ci._write_to_pipe(_control_api.FromAnkaios()) + ci._write_to_pipe(request) output_file = MagicMock() ci._output_file = output_file - ci._write_to_pipe(_control_api.FromAnkaios()) + ci._write_to_pipe(request) output_file.write.assert_called() output_file.flush.assert_called_once() +def test_write_to_pipe_closed_output_file(): + """ + A write whose output file was closed by a concurrent disconnect + must raise a ConnectionException, not a ValueError. + """ + ci = ControlInterfaceConnection( + add_response_callback=lambda _: None, + add_log_callback=lambda _: None, + add_event_callback=lambda _: None, + ) + output_file = MagicMock() + output_file.write.side_effect = ValueError("write to closed file") + ci._output_file = output_file + request = _control_api.FromAnkaios() + + with pytest.raises( + ConnectionException, + match="Could not write to pipe, output file closed.", + ): + ci._write_to_pipe(request) + + def test_write_request(): """ Test the write_request method of the ControlInterfaceConnection class.