fix: Terminate RPyC process pipe-drain watcher threads on stop/kill

[a-gorczew]

fix: Terminate RPyC process pipe-drain watcher threads on stop/kill

Remote stdout/stderr watcher threads were never stopped when a process was terminated, so if a detached child kept the pipe's write-end open they stayed alive and kept pushing output over the single shared RPyC connection.
These leaked threads accumulated across processes and progressively congested the connection, making each subsequent stop()/wait() slower.

Add a remote stop event to _get_process_io_queue and signal it via _stop_pipe_drain() after wait() in kill() and stop().

[mfd-intel-bot]

We don't publish DEVs .whl.
To build .whl, run 'pip install git+https://github.com/intel/mfd-connect@MFD-8972_multi_rpyc'

[Copilot]

Pull request overview

This PR addresses an RPyCProcess resource-leak scenario where remote stdout/stderr watcher threads could outlive the process lifecycle, accumulating across runs and degrading performance over the shared RPyC connection.

Changes:

• Extend the remote IO queue setup to return a remote stop event alongside the queue, and cache that event per stream.
• Add _stop_pipe_drain() and invoke it after successful wait() in kill() and OS-specific stop() implementations.
• Update unit tests for the new queue/stop-event behavior and add coverage for _stop_pipe_drain().

Reviewed changes

Copilot reviewed 5 out of 6 changed files in this pull request and generated 2 comments.

Show a summary per file

File	Description
mfd_connect/process/rpyc/base.py	Adds remote stop events for stdout/stderr watcher threads and introduces _stop_pipe_drain(); calls it after wait() in kill().
mfd_connect/process/rpyc/posix.py	Calls _stop_pipe_drain() after wait() in stop() (POSIX).
mfd_connect/process/rpyc/windows.py	Calls _stop_pipe_drain() after wait() in stop() (Windows).
tests/unit/test_mfd_connect/test_process/test_rpyc/test_base.py	Updates queue-caching tests for the new (queue, stop_event) return and adds tests for _stop_pipe_drain().
log.txt	Adds a log output file to the repository.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Copilot]

Pull request overview

Copilot reviewed 4 out of 5 changed files in this pull request and generated no new comments.

Suppressed comments (2)

mfd_connect/process/rpyc/base.py:120

• The new stop_event doesn’t actually guarantee the watcher thread terminates: for line in process_io (file iteration / readline) can block indefinitely if a detached child keeps the pipe open but isn’t producing newlines. Since the stop flag is only checked after a line is yielded, setting it via _stop_pipe_drain() may never be observed and the thread can remain alive (and the queue iterator may never see its terminating None). Consider also unblocking the read on stop (e.g., close the remote stream / fd as part of the stop signal, or implement a timeout-based read loop that periodically checks the stop condition).

        def _watcher() -> None:
            try:
                with process_io:
                    for line in process_io:
                        if stop_event.is_set():
                            break
                        q.put(line)

mfd_connect/process/rpyc/base.py:120

• In _get_process_io_queue, the watcher checks stop_event before enqueuing the line. If _stop_pipe_drain() signals the event while there is still buffered output (either from the just-finished process or from any process still holding the pipe), the first line read after the signal is dropped entirely because it’s read by the iterator but never put() into the queue. This can truncate captured stdout/stderr unexpectedly.

                    for line in process_io:
                        if stop_event.is_set():
                            break
                        q.put(line)

[Copilot]

Pull request overview

Copilot reviewed 4 out of 4 changed files in this pull request and generated no new comments.

Suppressed comments (1)

mfd_connect/process/rpyc/base.py:120

• The stop_event check inside the for line in process_io loop won’t terminate the watcher thread if it’s blocked waiting for the next line (e.g., pipe write-end kept open but no further output). In that case _stop_pipe_drain() may not actually stop the remote watcher threads, so they can still leak until something is written or the stream closes. Consider also closing process_io when the stop event is set to unblock the read loop immediately.

                with process_io:
                    for line in process_io:
                        if stop_event.is_set():
                            break
                        q.put(line)

[coveralls-official]

coverage: 80.595% (+0.7%) from 79.925% — MFD-8972_multi_rpyc into main

[Copilot]

Pull request overview

Copilot reviewed 6 out of 6 changed files in this pull request and generated no new comments.

Suppressed comments (1)

mfd_connect/process/rpyc/base.py:120

• The new stop_event is only checked after a line is yielded from process_io (for line in process_io:). If the pipe stays open but becomes silent (e.g., detached child keeps the write-end open without further output), the watcher thread can remain blocked inside the iterator and will not observe stop_event, so it won’t actually terminate on stop/kill as intended.

                with process_io:
                    for line in process_io:
                        if stop_event.is_set():
                            break
                        q.put(line)

[Copilot]

Pull request overview

Copilot reviewed 6 out of 6 changed files in this pull request and generated no new comments.

Suppressed comments (1)

mfd_connect/process/rpyc/base.py:106

• In _get_process_io_queue the docstring parameter order doesn’t match the function signature (process_io is first, bq is second), which can confuse readers and generated docs.

        :param bq: BatchQueue class
        :param process_io: IO object to wrap around (stdout or stderr).