diff --git a/scripts/nvx_tools/openvmm_process.py b/scripts/nvx_tools/openvmm_process.py index 7e418458..ff4385b2 100644 --- a/scripts/nvx_tools/openvmm_process.py +++ b/scripts/nvx_tools/openvmm_process.py @@ -83,7 +83,15 @@ def __init__( process_environment["OPENVMM_LOG"] = "off" if environment is not None: process_environment.update(environment) - self._interaction = InteractiveProcess(command, process_environment) + # The transcript is open before OpenVMM starts, so a failure to create it + # stops the run here, and every consumed byte reaches it as it is read. + log_path.parent.mkdir(parents=True, exist_ok=True) + self._log = log_path.open("wb") + try: + self._interaction = InteractiveProcess(command, process_environment) + except BaseException: + self._log.close() + raise self._chunks: queue.Queue[bytes | None] = queue.Queue() self._reader = threading.Thread( target=self._interaction.read_output, @@ -97,6 +105,7 @@ def __init__( except BaseException: terminate(self._interaction.process) self._interaction.close() + self._log.close() raise self._output = bytearray() self._search_offset = 0 @@ -154,7 +163,7 @@ def _query_status_first(self) -> None: self._consume(chunk) def _consume(self, chunk: bytes) -> None: - self._output.extend(chunk) + self._record(chunk) try: self._monitor.feed(chunk) except TimeAbiFailure as error: @@ -266,7 +275,7 @@ def wait(self, timeout: float) -> OpenvmmProcessResult: except TimeAbiFailure as error: self._fail(error) self._finished = True - self._write_log() + self._close_log() return OpenvmmProcessResult(returncode, bytes(self._output)) def close(self) -> None: @@ -279,24 +288,52 @@ def close(self) -> None: """ if not self._finished and self.process.poll() is None: terminate(self.process) - self._drain_available() - self._write_log() - self._interaction.close() - self._finished = True + try: + # The reader queues its last chunks while the interaction shuts down, so + # drain only after it has stopped. Drain even if that shutdown fails, and + # release the transcript either way. + try: + self._interaction.close() + finally: + if not self._log.closed: + self._drain_available() + finally: + self._close_log() + self._finished = True def _drain_available(self) -> None: - """Keep the output still queued; failure paths only, so no scan.""" + """Keep the output still queued; failure paths only, so no scan. + + A transcript write error does not stop the drain: every queued chunk is + kept in memory, and the first error is re-raised once the queue is empty. + """ + error: Exception | None = None while True: try: chunk = self._chunks.get_nowait() except queue.Empty: - return - if chunk is not None: - self._output.extend(chunk) + break + if chunk is None: + continue + self._output.extend(chunk) + if error is None: + try: + self._log.write(chunk) + self._log.flush() + except Exception as write_error: + error = write_error + if error is not None: + raise error + + def _record(self, chunk: bytes) -> None: + """Keep a consumed chunk in memory and in the transcript, flushed for readers.""" + self._output.extend(chunk) + self._log.write(chunk) + self._log.flush() - def _write_log(self) -> None: - self._log_path.parent.mkdir(parents=True, exist_ok=True) - self._log_path.write_bytes(self._output) + def _close_log(self) -> None: + if not self._log.closed: + self._log.close() def _fail(self, error: Exception): self.close() diff --git a/scripts/test_microvm_tests.py b/scripts/test_microvm_tests.py index 6a073364..ff6cb3bc 100644 --- a/scripts/test_microvm_tests.py +++ b/scripts/test_microvm_tests.py @@ -4827,6 +4827,139 @@ def test_process_wait_reads_final_chunks_after_process_exit(self): self.assertEqual(log_path.read_bytes(), result.output) self.assertEqual(queues.return_value.get.call_count, 4) + def test_process_transcript_grows_while_the_process_runs(self): + with tempfile.TemporaryDirectory() as temporary: + log_path = Path(temporary) / "output.log" + with ( + patch.object(openvmm_process, "InteractiveProcess") as interaction, + patch.object(openvmm_process.threading, "Thread"), + patch.object(openvmm_process.queue, "Queue") as queues, + ): + interaction.return_value.process.poll.return_value = None + interaction.return_value.process.wait.return_value = 0 + queues.return_value.get.side_effect = [b"BOOT\n", None] + queues.return_value.get_nowait.side_effect = queue.Empty + with openvmm_process.OpenvmmProcess(["openvmm"], log_path) as process: + process.wait_for(b"BOOT", 1) + # Consumed bytes are on disk while OpenVMM is still running. + self.assertEqual(log_path.read_bytes(), b"BOOT\n") + process.wait(1) + self.assertEqual(log_path.read_bytes(), b"BOOT\n") + + def test_process_transcript_open_failure_surfaces_before_start(self): + with tempfile.TemporaryDirectory() as temporary: + log_path = Path(temporary) / "output.log" + log_path.mkdir() + with patch.object(openvmm_process, "InteractiveProcess") as interaction: + with self.assertRaises(OSError): + openvmm_process.OpenvmmProcess(["openvmm"], log_path) + interaction.assert_not_called() + + def test_process_close_writes_output_delivered_after_the_last_wait(self): + with tempfile.TemporaryDirectory() as temporary: + log_path = Path(temporary) / "output.log" + with ( + patch.object(openvmm_process, "InteractiveProcess") as interaction, + patch.object(openvmm_process.threading, "Thread"), + patch.object(openvmm_process.queue, "Queue") as queues, + ): + interaction.return_value.process.poll.return_value = None + queues.return_value.get_nowait.side_effect = [b"LATE\n", queue.Empty] + process = openvmm_process.OpenvmmProcess(["openvmm"], log_path) + process.close() + self.assertEqual(log_path.read_bytes(), b"LATE\n") + + def test_process_close_keeps_output_queued_during_interaction_close(self): + with tempfile.TemporaryDirectory() as temporary: + log_path = Path(temporary) / "output.log" + closed: list[bool] = [] + with ( + patch.object(openvmm_process, "InteractiveProcess") as interaction, + patch.object(openvmm_process.threading, "Thread"), + patch.object(openvmm_process.queue, "Queue") as queues, + ): + interaction.return_value.process.poll.return_value = None + interaction.return_value.close.side_effect = lambda: closed.append(True) + + pending = [b"LAST\n"] + + def get_nowait() -> bytes: + # The reader queues its last chunk only while the interaction shuts down. + if closed and pending: + return pending.pop() + raise queue.Empty + + queues.return_value.get_nowait.side_effect = get_nowait + process = openvmm_process.OpenvmmProcess(["openvmm"], log_path) + process.close() + self.assertEqual(process.output, b"LAST\n") + self.assertEqual(log_path.read_bytes(), b"LAST\n") + + def test_process_close_keeps_queued_output_when_interaction_close_fails(self): + with tempfile.TemporaryDirectory() as temporary: + log_path = Path(temporary) / "output.log" + with ( + patch.object(openvmm_process, "InteractiveProcess") as interaction, + patch.object(openvmm_process.threading, "Thread"), + patch.object(openvmm_process.queue, "Queue") as queues, + ): + interaction.return_value.process.poll.return_value = None + interaction.return_value.close.side_effect = OSError("reader stuck") + queues.return_value.get_nowait.side_effect = [b"QUEUED\n", queue.Empty] + process = openvmm_process.OpenvmmProcess(["openvmm"], log_path) + with self.assertRaises(OSError): + process.close() + self.assertEqual(process.output, b"QUEUED\n") + self.assertEqual(log_path.read_bytes(), b"QUEUED\n") + + def test_process_close_releases_resources_when_the_transcript_write_fails(self): + with tempfile.TemporaryDirectory() as temporary: + log_path = Path(temporary) / "output.log" + with ( + patch.object(openvmm_process, "InteractiveProcess") as interaction, + patch.object(openvmm_process.threading, "Thread"), + patch.object(openvmm_process.queue, "Queue") as queues, + ): + interaction.return_value.process.poll.return_value = None + queues.return_value.get_nowait.side_effect = [b"LATE\n", queue.Empty] + process = openvmm_process.OpenvmmProcess(["openvmm"], log_path) + failing = MagicMock() + failing.closed = False + failing.write.side_effect = OSError("no space left on device") + process._log.close() + process._log = failing + with self.assertRaises(OSError): + process.close() + interaction.return_value.close.assert_called_once_with() + failing.close.assert_called_once_with() + + def test_process_close_keeps_every_queued_chunk_when_the_transcript_write_fails(self): + with tempfile.TemporaryDirectory() as temporary: + log_path = Path(temporary) / "output.log" + with ( + patch.object(openvmm_process, "InteractiveProcess") as interaction, + patch.object(openvmm_process.threading, "Thread"), + patch.object(openvmm_process.queue, "Queue") as queues, + ): + interaction.return_value.process.poll.return_value = None + queues.return_value.get_nowait.side_effect = [ + b"FIRST\n", + b"SECOND\n", + b"THIRD\n", + queue.Empty, + ] + process = openvmm_process.OpenvmmProcess(["openvmm"], log_path) + failing = MagicMock() + failing.closed = False + failing.write.side_effect = OSError("no space left on device") + process._log.close() + process._log = failing + with self.assertRaises(OSError): + process.close() + self.assertEqual(bytes(process._output), b"FIRST\nSECOND\nTHIRD\n") + interaction.return_value.close.assert_called_once_with() + failing.close.assert_called_once_with() + def test_process_wait_clamps_elapsed_process_timeout(self): with tempfile.TemporaryDirectory() as temporary: with (