Honour timeout when muxing and closing an output container

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. Arm the read timeout around both.

Each mux gets the full timeout, as each demux already does. A close shares
one deadline across writing the trailer and flushing, so it cannot outlast
the timeout it was given.

The close side has no test of its own. A timed-out write leaves its error
on the AVIO context, so the trailer then fails at once rather than
blocking, and I could not get a peer to stall only between the last mux and
the close: loopback buffers are far larger than the 32KB AVIO buffer, and
AF_UNIX ones are small enough that the header blocks first.

Author: AF-bria

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. By :gh-user:`adrianrfreedman` in (:pr:`2414`).
 
 
 18.X and Below

av/container/core.py

@@ -500,10 +500,10 @@ 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.
     :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

@@ -62,16 +62,23 @@ def close_output(self: OutputContainer) -> cython.void:
             # segmentation fault. Therefore no matter whether it succeeds or not
             # we must absolutely set enum.done.
             ret: cython.int
+            self.set_timeout(self.read_timeout)
             try:
+                self.start_timeout()
                 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 +762,15 @@ def _mux_one(self, packet: Packet) -> cython.void:
         self.err_check(lib.av_packet_ref(self.packet_ptr, packet.ptr))
 
         ret: cython.int
+        self.set_timeout(self.read_timeout)
+        self.start_timeout()
         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,12 +16,22 @@
 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)
         self.port: int = self.sock.getsockname()[1]
@@ -152,3 +163,58 @@ 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,
+        )
+        stream = container.add_stream("mpeg4", rate=30)
+        stream.width = 640
+        stream.height = 480
+        stream.pix_fmt = "yuv420p"
+        return container
+
+    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)