diff --git a/src/borg/remote.py b/src/borg/remote.py index 637f4d111..ec9d490c4 100644 --- a/src/borg/remote.py +++ b/src/borg/remote.py @@ -13,7 +13,7 @@ import tempfile import textwrap import time import traceback -from subprocess import Popen, PIPE +from subprocess import Popen, PIPE, TimeoutExpired from . import __version__ from .compress import Compressor @@ -44,6 +44,12 @@ MAX_INFLIGHT = 100 RATELIMIT_PERIOD = 0.1 +# when shutting down the connection, we still want to receive (and log) the remote's stderr output, +# but we must not wait for it forever if the remote is stuck or gone: +STDERR_DRAIN_IDLE_TIMEOUT = 30.0 # seconds without any stderr data before we stop draining it +# after that, wait for the child process to exit - and make it exit if it does not do so voluntarily: +CHILD_EXIT_TIMEOUT = 10.0 # seconds to wait for the child, before SIGTERM and (again) before SIGKILL + def os_write(fd, data): """os.write wrapper so we do not lose data for partial writes.""" @@ -857,17 +863,7 @@ This problem will go away as soon as the server has been upgraded to 1.0.7+. data = os.read(fd, 32768) if not data: raise ConnectionClosed() - self.rx_bytes += len(data) - # deal with incomplete lines (may appear due to block buffering) - if self.stderr_received: - data = self.stderr_received + data - self.stderr_received = b'' - lines = data.splitlines(keepends=True) - if lines and not lines[-1].endswith((b'\r', b'\n')): - self.stderr_received = lines.pop() - # now we have complete lines in and any partial line in self.stderr_received. - for line in lines: - handle_remote_line(line.decode()) # decode late, avoid partial utf-8 sequences + self.handle_stderr_data(data) if w: while (len(self.to_send) <= maximum_to_send) and (calls or self.preload_ids) and len(waiting_for) < MAX_INFLIGHT: if calls: @@ -973,11 +969,56 @@ This problem will go away as soon as the server has been upgraded to 1.0.7+. def break_lock(self): """actual remoting is done via self.call in the @api decorator""" + def handle_stderr_data(self, data): + """Log the remote's stderr output (given as bytes, maybe not ending on a line boundary).""" + self.rx_bytes += len(data) + # deal with incomplete lines (may appear due to block buffering) + if self.stderr_received: + data = self.stderr_received + data + self.stderr_received = b'' + lines = data.splitlines(keepends=True) + if lines and not lines[-1].endswith((b'\r', b'\n')): + self.stderr_received = lines.pop() + # now we have complete lines in and any partial line in self.stderr_received. + for line in lines: + handle_remote_line(line.decode()) # decode late, avoid partial utf-8 sequences + + def drain_stderr(self): + """Read and log the remote's stderr until EOF (or until it is idle for too long). + + We are the only reader of that pipe, so we must keep reading it: if we did not, the child + could block forever writing into a full pipe buffer while we wait for it to terminate. + Also, the remote's stderr transports its error messages, so we want to see them. + """ + while True: + r, w, x = select.select([self.stderr_fd], [], [], STDERR_DRAIN_IDLE_TIMEOUT) + if not r: + logger.debug('drain_stderr: remote stderr idle for %.1fs, giving up on it.', + STDERR_DRAIN_IDLE_TIMEOUT) + break + data = os.read(self.stderr_fd, 32768) + if not data: + break # EOF + self.handle_stderr_data(data) + if self.stderr_received: # log a trailing partial line, if any + self.handle_stderr_data(b'\n') + def close(self): if self.p: self.p.stdin.close() self.p.stdout.close() - self.p.wait() + self.drain_stderr() + self.p.stderr.close() + # the child should terminate now, but just in case it does not, do not wait forever: + for make_it_terminate in (self.p.terminate, self.p.kill): + try: + self.p.wait(timeout=CHILD_EXIT_TIMEOUT) + break + except TimeoutExpired: + logger.warning('Remote process %d did not terminate, sending it a signal.', self.p.pid) + make_it_terminate() + else: + self.p.wait() # SIGKILLed, so this will not block for long self.p = None def async_response(self, wait=True): diff --git a/src/borg/testsuite/remote.py b/src/borg/testsuite/remote.py index 034ebb4ba..75e886203 100644 --- a/src/borg/testsuite/remote.py +++ b/src/borg/testsuite/remote.py @@ -1,12 +1,15 @@ import errno import os import io +import sys +import threading import time +from subprocess import Popen, PIPE from unittest.mock import patch import pytest -from ..remote import SleepingBandwidthLimiter, RepositoryCache, cache_if_remote +from ..remote import RemoteRepository, SleepingBandwidthLimiter, RepositoryCache, cache_if_remote from ..repository import Repository from ..crypto.key import PlaintextKey from ..compress import CompressionSpec @@ -199,3 +202,34 @@ class TestRepositoryCache: with pytest.raises(IntegrityError): assert next(iterator) == (7, b'5678') + + +@pytest.mark.skipif(sys.platform == 'win32', reason='remote repositories are not supported on Windows') +def test_close_drains_stderr(): + """RemoteRepository.close() must not deadlock on a child that still writes to stderr, see #10165. + + borg is the only reader of the child's stderr pipe, so if it stopped reading it while waiting + for the child to terminate, the child would block forever writing into the full pipe buffer. + """ + # the child waits for EOF on stdin (like the remote does when we close our end), + # then writes way more than a pipe buffer (64kiB) worth of stderr output: + code = 'import sys; sys.stdin.read(); sys.stderr.write(("E" * 4095 + "\\n") * 64)' + p = Popen([sys.executable, '-c', code], bufsize=0, stdin=PIPE, stdout=PIPE, stderr=PIPE) + # set up just enough of a RemoteRepository for close() to work on: + rr = RemoteRepository.__new__(RemoteRepository) + rr.p = p + rr.stderr_fd = p.stderr.fileno() + rr.stderr_received = b'' + rr.rx_bytes = 0 + rr.responses = {} # only used by RemoteRepository.__del__ + os.set_blocking(rr.stderr_fd, False) + # if close() deadlocks, it never returns - so do not call it in the main thread: + closer = threading.Thread(target=rr.close, daemon=True) + closer.start() + closer.join(timeout=30) + if closer.is_alive(): + p.kill() # unblock the deadlocked close(), so the child does not stay around + pytest.fail('RemoteRepository.close() did not return, see #10165.') + assert rr.p is None + assert p.returncode == 0 # the child could write all its output and terminated by itself + assert rr.rx_bytes == 64 * 4096 # we have received all of the child's stderr output