Stop output containers blocking the process on a stalled network URL

CHANGELOG.rst

@@ -65,6 +65,7 @@ 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`).
 
 
 18.X and Below

av/container/core.py

@@ -292,16 +292,20 @@ def __cinit__(
             # We need the context before we open the input AND setup Python IO.
             self.ptr = lib.avformat_alloc_context()
 
-            # Setup interrupt callback
-            if self.open_timeout is not None or self.read_timeout is not None:
-                self.ptr.interrupt_callback.callback = interrupt_cb
-                self.ptr.interrupt_callback.opaque = cython.address(
-                    self.interrupt_callback_info
-                )
-
             if acodec is not None:
                 self.ptr.audio_codec_id = getattr(AudioCodec, acodec)
 
+        # Setup interrupt callback. Muxing needs it as much as demuxing does,
+        # since writing the header to a network URL can block indefinitely.
+        if self.open_timeout is not None or self.read_timeout is not None:
+            # Start disarmed, so nothing between here and the first
+            # start_timeout() can be interrupted by a zeroed deadline.
+            self.set_timeout(None)
+            self.ptr.interrupt_callback.callback = interrupt_cb
+            self.ptr.interrupt_callback.opaque = cython.address(
+                self.interrupt_callback_info
+            )
+
         self.ptr.flags |= lib.AVFMT_FLAG_GENPTS
         self.ptr.opaque = cython.cast(cython.p_void, self)
 
@@ -495,7 +499,11 @@ def open(
     :param int buffer_size: Size of buffer for Python input/output operations in bytes.
         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.
+        ``(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.
     :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.pxd

@@ -7,6 +7,9 @@ from av.stream cimport Stream
 
 cdef class OutputContainer(Container):
     cdef lib.AVPacket *packet_ptr
+    # How many nogil libav calls are in flight, so close() can refuse
+    # to free the context while another thread is still inside one.
+    cdef int _blocking_depth
     cdef dict _extradata_bsfs
     cdef list[Packet] _buffered_packets
     cdef _buffer_for_extradata(self, Packet packet)

av/container/output.py

@@ -49,6 +49,7 @@ def close_output(self: OutputContainer) -> cython.void:
             self._mux_one(packet)
 
     self.streams = StreamContainer()
+    self._blocking_depth += 1
     try:
         if self._myflag & 12 == 4:  # enum.started and not enum.done
             # If the underlying Python IO file was already closed (e.g. during
@@ -60,13 +61,17 @@ def close_output(self: OutputContainer) -> cython.void:
             # We must only ever call av_write_trailer *once*, otherwise we get a
             # segmentation fault. Therefore no matter whether it succeeds or not
             # we must absolutely set enum.done.
+            ret: cython.int
             try:
-                self.err_check(lib.av_write_trailer(self.ptr))
+                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
                 ):
-                    lib.avio_closep(cython.address(self.ptr.pb))
+                    with cython.nogil:
+                        lib.avio_closep(cython.address(self.ptr.pb))
                 self._myflag |= 8  # enum.done = True
     finally:
         # Drop the context so a closed output reports itself as closed:
@@ -76,6 +81,7 @@ def close_output(self: OutputContainer) -> cython.void:
         with cython.nogil:
             lib.avformat_free_context(self.ptr)
             self.ptr = cython.NULL
+        self._blocking_depth -= 1
 
 
 @cython.final
@@ -591,17 +597,54 @@ def start_encoding(self):
         # Open the output file, if needed.
         name_obj: bytes = os.fsencode(self.name if self.file is None else "")
         name: cython.p_char = name_obj
-        if self.ptr.pb == cython.NULL and not self.ptr.oformat.flags & lib.AVFMT_NOFILE:
-            err_check(
-                lib.avio_open(cython.address(self.ptr.pb), name, lib.AVIO_FLAG_WRITE)
-            )
+        ret: cython.int
+        opened_pb: cython.bint = False
+        all_options: Dictionary
+        options: Dictionary
+        options_ptr: cython.pointer[cython.pointer[lib.AVDictionary]]
+
+        self.set_timeout(self.open_timeout)
+        self.start_timeout()
+        self._blocking_depth += 1
+        try:
+            if (
+                self.ptr.pb == cython.NULL
+                and not self.ptr.oformat.flags & lib.AVFMT_NOFILE
+            ):
+                # avio_open() would pass the protocol a NULL interrupt
+                # callback, so a stalled connect could never be timed out.
+                with cython.nogil:
+                    ret = lib.avio_open2(
+                        cython.address(self.ptr.pb),
+                        name,
+                        lib.AVIO_FLAG_WRITE,
+                        cython.address(self.ptr.interrupt_callback),
+                        cython.NULL,
+                    )
+                err_check(ret)
+                opened_pb = True
 
-        # Copy the metadata dict.
-        dict_to_avdict(cython.address(self.ptr.metadata), self.metadata)
+            # Copy the metadata dict.
+            dict_to_avdict(cython.address(self.ptr.metadata), self.metadata)
 
-        all_options: Dictionary = Dictionary(self.options, self.container_options)
-        options: Dictionary = all_options.copy()
-        self.err_check(lib.avformat_write_header(self.ptr, cython.address(options.ptr)))
+            all_options = Dictionary(self.options, self.container_options)
+            options = all_options.copy()
+            options_ptr = cython.address(options.ptr)
+            with cython.nogil:
+                ret = lib.avformat_write_header(self.ptr, options_ptr)
+            try:
+                self.err_check(ret)
+            except Exception:
+                # started is never set, so close_output() will not close pb.
+                # Nothing else will either, and a stalled header write is an
+                # expected path now that it can time out.
+                if opened_pb:
+                    with cython.nogil:
+                        lib.avio_closep(cython.address(self.ptr.pb))
+                raise
+        finally:
+            self._blocking_depth -= 1
+            self.set_timeout(None)
 
         # Track option usage...
         for k in all_options:
@@ -668,6 +711,13 @@ def default_subtitle_codec(self):
         return lib.avcodec_get_name(self.format.optr.subtitle_codec)
 
     def close(self):
+        if self._blocking_depth:
+            # Another thread is inside libav without the GIL, so freeing the
+            # context here would be a use-after-free. Pass ``timeout`` to
+            # :func:`av.open` to give up on an open that never connects.
+            raise RuntimeError(
+                "Cannot close an OutputContainer while another thread is writing to it"
+            )
         close_output(self)
 
     def mux(self, packets):
@@ -704,8 +754,13 @@ def _mux_one(self, packet: Packet) -> cython.void:
         # takes ownership of the reference.
         self.err_check(lib.av_packet_ref(self.packet_ptr, packet.ptr))
 
-        with cython.nogil:
-            ret: cython.int = lib.av_interleaved_write_frame(self.ptr, self.packet_ptr)
+        ret: cython.int
+        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.err_check(ret)
 
     @cython.cfunc

include/avformat.pxd

@@ -179,6 +179,10 @@ cdef extern from "libavformat/avformat.h" nogil:
     cdef int av_interleaved_write_frame(AVFormatContext *ctx, AVPacket *pkt)
     cdef int av_write_frame(AVFormatContext *ctx, AVPacket *pkt)
     cdef int avio_open(AVIOContext **s, const char *url, int flags)
+    cdef int avio_open2(
+        AVIOContext **s, const char *url, int flags,
+        const AVIOInterruptCB *int_cb, AVDictionary **options
+    )
     cdef int64_t avio_size(AVIOContext *s)
     cdef const AVOutputFormat* av_guess_format(
         const char *short_name, const char *filename, const char *mime_type

tests/test_output_blocking.py

@@ -0,0 +1,154 @@
+import socket
+import threading
+import time
+
+import pytest
+
+import av
+
+from .common import TestCase
+
+# The main thread sleeps in 1 ms slices while another thread is stuck in the
+# RTMP handshake. It wakes roughly a thousand times if the GIL is free and a
+# handful of times if it is not, so this threshold sits well clear of both.
+WINDOW = 1.0
+MIN_TICKS = 100
+
+
+class SilentServer:
+    """Accepts connections and then says nothing, so the handshake never ends."""
+
+    def __init__(self) -> None:
+        self.sock = socket.socket()
+        self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+        self.sock.bind(("127.0.0.1", 0))
+        self.sock.listen(4)
+        self.port: int = self.sock.getsockname()[1]
+        self.accepted: list[socket.socket] = []
+        self.thread = threading.Thread(target=self._accept, daemon=True)
+        self.thread.start()
+
+    def _accept(self) -> None:
+        while True:
+            try:
+                conn, _ = self.sock.accept()
+            except OSError:
+                return
+            self.accepted.append(conn)
+
+    def close(self) -> None:
+        self.sock.close()
+        for conn in self.accepted:
+            conn.close()
+
+
+def has_rtmp() -> bool:
+    """Whether FFmpeg was built with the RTMP protocol.
+
+    Port 1 on loopback refuses at once, so the probe either fails looking the
+    protocol up, before any connect, or fails connecting.
+    """
+    try:
+        with av.open("rtmp://127.0.0.1:1/x", "w", format="flv", timeout=1) as container:
+            container.start_encoding()
+    except av.error.ProtocolNotFoundError:
+        return False
+    except Exception:
+        pass
+    return True
+
+
+@pytest.mark.skipif(not has_rtmp(), reason="FFmpeg was built without RTMP")
+class TestOutputBlocking(TestCase):
+    def setUp(self) -> None:
+        self.server = SilentServer()
+
+    def tearDown(self) -> None:
+        self.server.close()
+
+    def _push(
+        self, timeout: float, containers: list | None = None
+    ) -> tuple[threading.Thread, list[BaseException]]:
+        raised: list[BaseException] = []
+
+        def run() -> None:
+            try:
+                container = av.open(
+                    f"rtmp://127.0.0.1:{self.server.port}/live/x",
+                    "w",
+                    format="flv",
+                    timeout=timeout,
+                )
+                if containers is not None:
+                    containers.append(container)
+                stream = container.add_stream("h264", rate=30)
+                stream.width = 320
+                stream.height = 240
+                stream.pix_fmt = "yuv420p"
+                container.start_encoding()
+            except BaseException as e:
+                raised.append(e)
+
+        thread = threading.Thread(target=run, daemon=True)
+        thread.start()
+        return thread, raised
+
+    def test_start_encoding_releases_the_gil(self) -> None:
+        thread, _ = self._push(WINDOW * 3)
+
+        ticks = 0
+        deadline = time.monotonic() + WINDOW
+        while time.monotonic() < deadline:
+            ticks += 1
+            time.sleep(0.001)
+
+        assert thread.is_alive(), "the handshake completed, so nothing was blocking"
+        assert ticks > MIN_TICKS, f"main thread only ran {ticks} times"
+        thread.join(WINDOW * 8)
+
+    def test_start_encoding_honours_the_timeout(self) -> None:
+        thread, raised = self._push(WINDOW)
+        thread.join(WINDOW * 8)
+        assert not thread.is_alive(), "timeout did not interrupt the handshake"
+        assert raised, "the handshake returned instead of timing out"
+
+    def test_close_refuses_to_free_a_container_in_use(self) -> None:
+        containers: list[av.container.OutputContainer] = []
+        thread, _ = self._push(WINDOW * 2, containers)
+
+        # The server only accepts once the writing thread is inside the
+        # connect, which is where the context stops being ours to free.
+        deadline = time.monotonic() + WINDOW
+        while not self.server.accepted and time.monotonic() < deadline:
+            time.sleep(0.001)
+        assert self.server.accepted, "the writing thread never connected"
+
+        with pytest.raises(RuntimeError, match="another thread"):
+            containers[0].close()
+        thread.join(WINDOW * 8)
+
+
+class TestFailedHeaderWrite(TestCase):
+    def test_a_failed_header_write_closes_the_connection(self) -> None:
+        """mp4 cannot carry PCM, so the muxer rejects it after the connect."""
+        server = SilentServer()
+        self.addCleanup(server.close)
+
+        container = av.open(f"tcp://127.0.0.1:{server.port}", "w", format="mp4")
+        container.add_stream("pcm_s16le")
+        with pytest.raises(av.error.ArgumentError):
+            container.start_encoding()
+
+        deadline = time.monotonic() + WINDOW
+        while not server.accepted and time.monotonic() < deadline:
+            time.sleep(0.001)
+        assert server.accepted, "the writer never connected"
+
+        # started was never set, so nothing downstream would close pb.
+        conn = server.accepted[0]
+        conn.settimeout(WINDOW)
+        try:
+            while conn.recv(4096):
+                pass  # Drain whatever the muxer wrote before it gave up.
+        except TimeoutError:
+            raise AssertionError("the connection was left open") from None