Honour timeout when muxing and closing an output container

CHANGELOG.rst

@@ -65,7 +65,8 @@ Fixes:
 - Fix crashes from indexes that were turned into C pointer arithmetic without being range checked. ``MotionVectors[i]`` only checked the upper bound, so a negative index read off the front of the buffer (``mvs[-1]`` now returns the last vector, as with any sequence); ``VideoFormatComponent`` and ``AudioPlane`` accepted any index at all; and ``BitmapSubtitlePlane`` and ``VideoBlockParams`` were missing their lower bounds.
 - Frames returned by flushing a codec context directly (``CodecContext.decode()`` with no packet) now carry the stream's ``time_base`` instead of ``None``.
 - ``VideoFrame.reformat()`` (and so ``to_ndarray(format=...)``, ``to_rgb()``, ``to_image()``) now shares one ``SwsContext`` per thread instead of allocating one per frame. FFmpeg 8's swscale retains megabytes of graph state per context, which showed up as large RSS growth when many frames were alive at once.
-- Writing to a network URL no longer blocks every other Python thread, and ``timeout`` now applies to opening an output container. ``avio_open()``, ``avformat_write_header()``, ``av_write_trailer()``, and ``avio_closep()`` held the GIL, so an unreachable RTMP server froze the whole process, and the interrupt callback was only installed for demuxing, so nothing could end the wait. Muxing and closing still ignore ``timeout``, and Muxing and closing still ignore ``timeout``, and :meth:`.OutputContainer.close` now raises rather than freeing a context another thread is still muxing or closing. By :gh-user:`adrianrfreedman` in (:pr:`2412`).
+- Writing to a network URL no longer blocks every other Python thread, and ``timeout`` now applies to opening an output container. ``avio_open()``, ``avformat_write_header()``, ``av_write_trailer()``, and ``avio_closep()`` held the GIL, so an unreachable RTMP server froze the whole process, and the interrupt callback was only installed for demuxing, so nothing could end the wait. :meth:`.OutputContainer.close` now raises rather than freeing a context another thread is still muxing or closing. By :gh-user:`adrianrfreedman` in (:pr:`2412`).
+- ``timeout`` now applies to muxing and closing an output container, not just to opening it. Only opening armed the interrupt callback, so a peer that accepted the connection and then stopped reading left ``av_interleaved_write_frame()`` and ``av_write_trailer()`` blocked forever. Each mux gets the full timeout, and a close shares one across writing the trailer and flushing, so neither can outlast it. Only unseekable outputs are covered: writing a seekable file takes as long as the file is big, so a deadline meant for a peer would abandon it part-written. By :gh-user:`adrianrfreedman` in (:pr:`2414`).
 
 
 18.X and Below

av/container/core.py

@@ -500,10 +500,15 @@ def open(
         Honored only when ``file`` is a file-like object. Defaults to 32768 (32k).
     :param timeout: How many seconds to wait for data before giving up, as a float, or a
         ``(open timeout, read timeout)`` tuple. The open timeout covers both connecting
-        and reading or writing the header. Writing honours it only while opening, so it
-        is the supported way to give up on an output that never connects. Muxing and
-        closing still block indefinitely, and calling :meth:`.OutputContainer.close`
-        from another thread to break out of them raises instead.
+        and reading or writing the header. The read timeout covers each subsequent
+        demux, mux, or close, so a stalled peer gives up rather than blocking forever.
+        Each demux and mux gets the full timeout; a close shares one across writing
+        the trailer and flushing, so it cannot outlast the timeout either. On output,
+        it applies only where the destination is unseekable, such as a socket or a
+        pipe. A seekable file is left alone, since the time a write takes there scales
+        with the file rather than with a peer, and giving up part-way would leave it
+        unreadable. A mux that does time out leaves the error on the I/O context, so
+        every later mux and the trailer fail at once. There is no retrying after one.
     :param callable io_open: Custom I/O callable for opening files/streams.
         This option is intended for formats that need to open additional
         file-like objects to ``file`` using custom I/O.

av/container/output.py

@@ -36,6 +36,23 @@ def _set_codecpar_extradata(
     stream.codecpar.extradata_size = size
 
 
+@cython.cfunc
+def arm_write_timeout(self: OutputContainer) -> cython.void:
+    """Start the write deadline, unless the output is a seekable file.
+
+    FFmpeg checks the interrupt callback before every write, local files
+    included, and a file's write time scales with its size: with
+    ``movflags=faststart`` the trailer rewrites the whole thing. A deadline
+    sized for a stalled peer would abandon a large file part-written rather
+    than protect it, so only unseekable outputs, which are the ones that can
+    stall indefinitely, get one.
+    """
+    if self.ptr.pb == cython.NULL or self.ptr.pb.seekable & lib.AVIO_SEEKABLE_NORMAL:
+        return
+    self.set_timeout(self.read_timeout)
+    self.start_timeout()
+
+
 @cython.cfunc
 def close_output(self: OutputContainer) -> cython.void:
     if self.ptr == cython.NULL:
@@ -63,15 +80,21 @@ def close_output(self: OutputContainer) -> cython.void:
             # we must absolutely set enum.done.
             ret: cython.int
             try:
+                arm_write_timeout(self)
                 with cython.nogil:
                     ret = lib.av_write_trailer(self.ptr)
                 self.err_check(ret)
             finally:
                 if self.file is None and not (
                     self.ptr.oformat.flags & lib.AVFMT_NOFILE
                 ):
+                    # No fresh deadline: the trailer and this flush share one,
+                    # so closing cannot outlast the timeout. The point here is
+                    # to stop the flush hanging, not to report on it, so its
+                    # return goes unchecked as it always has.
                     with cython.nogil:
                         lib.avio_closep(cython.address(self.ptr.pb))
+                self.set_timeout(None)
                 self._myflag |= 8  # enum.done = True
     finally:
         # Drop the context so a closed output reports itself as closed:
@@ -755,12 +778,14 @@ def _mux_one(self, packet: Packet) -> cython.void:
         self.err_check(lib.av_packet_ref(self.packet_ptr, packet.ptr))
 
         ret: cython.int
+        arm_write_timeout(self)
         self._blocking_depth += 1
         try:
             with cython.nogil:
                 ret = lib.av_interleaved_write_frame(self.ptr, self.packet_ptr)
         finally:
             self._blocking_depth -= 1
+            self.set_timeout(None)
         self.err_check(ret)
 
     @cython.cfunc

tests/test_output_blocking.py

@@ -2,6 +2,7 @@
 import threading
 import time
 
+import numpy as np
 import pytest
 
 import av
@@ -15,28 +16,48 @@
 MIN_TICKS = 100
 
 
+# A writer that never blocks has nothing to time out, so give up once the
+# socket has swallowed more than any plausible buffer.
+MAX_FRAMES = 500
+
+
 class SilentServer:
-    """Accepts connections and then says nothing, so the handshake never ends."""
+    """Accepts connections and then neither reads nor writes.
+
+    An RTMP handshake never completes against it, and a socket written to it
+    fills up and stays full.
+    """
 
     def __init__(self) -> None:
         self.sock = socket.socket()
         self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+        self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 1024)
         self.sock.bind(("127.0.0.1", 0))
         self.sock.listen(4)
+        # Poll rather than block, so close() can stop the thread itself.
+        # Closing the socket under a blocked accept() is not guaranteed to
+        # wake it, and a thread still sitting in accept() outlives the test.
+        self.sock.settimeout(0.1)
         self.port: int = self.sock.getsockname()[1]
         self.accepted: list[socket.socket] = []
+        self.stopped = threading.Event()
         self.thread = threading.Thread(target=self._accept, daemon=True)
         self.thread.start()
 
     def _accept(self) -> None:
-        while True:
+        while not self.stopped.is_set():
             try:
                 conn, _ = self.sock.accept()
+            except TimeoutError:
+                continue
             except OSError:
                 return
             self.accepted.append(conn)
 
     def close(self) -> None:
+        self.stopped.set()
+        self.thread.join(WINDOW)
+        assert not self.thread.is_alive(), "the accept thread outlived the server"
         self.sock.close()
         for conn in self.accepted:
             conn.close()
@@ -152,3 +173,105 @@ def test_a_failed_header_write_closes_the_connection(self) -> None:
                 pass  # Drain whatever the muxer wrote before it gave up.
         except TimeoutError:
             raise AssertionError("the connection was left open") from None
+
+
+class TestOutputWriteTimeout(TestCase):
+    """Muxing and closing over a peer that has stopped reading."""
+
+    def setUp(self) -> None:
+        self.server = SilentServer()
+
+    def tearDown(self) -> None:
+        self.server.close()
+
+    def _open(self) -> av.container.OutputContainer:
+        container = av.open(
+            f"tcp://127.0.0.1:{self.server.port}",
+            "w",
+            format="mpegts",
+            timeout=WINDOW,
+        )
+        # Closing from __del__ instead would write the trailer at collection
+        # time, where the failure surfaces as an unraisable exception.
+        self.addCleanup(self._close_quietly, container)
+        stream = container.add_stream("mpeg4", rate=30)
+        stream.width = 640
+        stream.height = 480
+        stream.pix_fmt = "yuv420p"
+        return container
+
+    @staticmethod
+    def _close_quietly(container: av.container.OutputContainer) -> None:
+        try:
+            container.close()
+        except av.error.ExitError:
+            pass  # A stalled write is the point of these tests.
+
+    def _fill(self, container: av.container.OutputContainer) -> None:
+        """Mux noise, which compresses badly, until the socket blocks."""
+        stream = container.streams.video[0]
+        rgb = np.random.randint(0, 256, (480, 640, 3), dtype=np.uint8)
+        frame = av.VideoFrame.from_ndarray(rgb, format="rgb24")
+        for i in range(MAX_FRAMES):
+            frame.pts = i
+            for packet in stream.encode(frame):
+                container.mux(packet)
+        raise AssertionError("the socket swallowed everything without blocking")
+
+    def test_mux_honours_the_timeout(self) -> None:
+        container = self._open()
+        start = time.monotonic()
+        with pytest.raises(av.error.ExitError):
+            self._fill(container)
+        assert time.monotonic() - start >= WINDOW, "the write gave up early"
+
+    def test_close_frees_the_container_after_a_stalled_write(self) -> None:
+        container = self._open()
+        with pytest.raises(av.error.ExitError):
+            self._fill(container)
+
+        # A timed-out write leaves its error on the AVIO context, so the
+        # trailer fails straight away rather than blocking. That makes this a
+        # test of the teardown, not of the close timeout: close() must report
+        # the failure and still free the context.
+        with pytest.raises(av.error.ExitError):
+            container.close()
+        with pytest.raises(AssertionError, match="not open"):
+            container.add_stream("mpeg4", rate=30)
+
+
+class TestSeekableOutputIgnoresTheTimeout(TestCase):
+    """A local file is written on its own schedule, not a peer's."""
+
+    def test_faststart_close_is_not_cut_short(self) -> None:
+        """``movflags=faststart`` rewrites the file when the trailer is written.
+
+        How long that takes grows with the file, so a timeout meant for a
+        stalled peer would abandon it part-written. The file has to survive
+        both the write and a read back.
+        """
+        path = self.sandboxed("faststart.mp4")
+        with av.open(
+            path,
+            "w",
+            # Small enough that any interruptible write trips it, so the test
+            # turns on whether a file is subject to the timeout at all rather
+            # than on how fast the disk is.
+            timeout=(None, 1e-9),
+            container_options={"movflags": "faststart"},
+        ) as container:
+            stream = container.add_stream("mpeg4", rate=30)
+            stream.width = 640
+            stream.height = 480
+            stream.pix_fmt = "yuv420p"
+            rgb = np.random.randint(0, 256, (480, 640, 3), dtype=np.uint8)
+            frame = av.VideoFrame.from_ndarray(rgb, format="rgb24")
+            for i in range(120):
+                frame.pts = i
+                for packet in stream.encode(frame):
+                    container.mux(packet)
+            for packet in stream.encode():
+                container.mux(packet)
+
+        with av.open(path) as container:
+            assert sum(1 for _ in container.decode(video=0)) == 120