push rtmp stream block the main thread

[zhaoxinyuan996]

When the target network is unreachable, the main thread will be blocked
python3.13
pyav==15.1.0（many versions will be blocked）
pyside6==6.8.2

import sys
import time
import threading
from fractions import Fraction

import av
import numpy as np
from PySide6.QtCore import QTimer
from PySide6.QtWidgets import QApplication, QWidget, QVBoxLayout, QPushButton, QLabel


RTMP_URL = 'rtmp://127.0.0.1:1935/live/test'

VIDEO_WIDTH = 1280
VIDEO_HEIGHT = 720
VIDEO_FPS = 30
VIDEO_TIME_BASE = Fraction(1, VIDEO_FPS)

AUDIO_RATE = 24000
AUDIO_CHANNELS = 1
AUDIO_SAMPLES = 1024
AUDIO_TIME_BASE = Fraction(1, AUDIO_RATE)

VIDEO_CODEC = 'h264_nvenc'


class RTMPStreamer:
    def __init__(self, url):
        self.url = url
        self.running = False
        self.thread = None
        self.container = None

    def start(self):
        if self.running:
            return
        self.running = True
        self.thread = threading.Thread(target=self._run, name='RTMPStreamer', daemon=True)
        self.thread.start()

    def stop(self):
        self.running = False

    def _run(self):
        print('子线程启动')
        try:
            print('准备 av.open()')
            self.container = av.open(
                self.url,
                'w',
                format='flv',
                timeout=(3.0, 3.0),
            )
            print('av.open() 成功')

            video_stream = self.container.add_stream(VIDEO_CODEC, rate=VIDEO_FPS)
            video_stream.width = VIDEO_WIDTH
            video_stream.height = VIDEO_HEIGHT
            video_stream.pix_fmt = 'nv12'
            video_stream.time_base = VIDEO_TIME_BASE
            video_stream.options = {
                'preset': 'p1',
                'tune': 'll',
                'bf': '0',
                'b': '2500k',
            }

            audio_stream = self.container.add_stream('aac', rate=AUDIO_RATE)
            audio_stream.layout = 'mono'
            audio_stream.time_base = AUDIO_TIME_BASE
            audio_stream.options = {
                'b': '64k',
            }

            video = np.zeros((VIDEO_HEIGHT, VIDEO_WIDTH, 3), dtype=np.uint8)
            audio = np.zeros(AUDIO_SAMPLES, dtype=np.int16)

            video_pts = 0
            audio_pts = 0
            next_video_time = time.monotonic()

            while self.running:
                now = time.monotonic()

                if now >= next_video_time:
                    frame = av.VideoFrame.from_ndarray(video, format='bgr24')
                    frame.pts = video_pts
                    frame.time_base = VIDEO_TIME_BASE
                    video_pts += 1

                    for packet in video_stream.encode(frame):
                        self.container.mux(packet)

                    next_video_time += 1.0 / VIDEO_FPS

                audio_frame = av.AudioFrame.from_ndarray(
                    audio.reshape(1, -1),
                    format='s16',
                    layout='mono',
                )
                audio_frame.sample_rate = AUDIO_RATE
                audio_frame.pts = audio_pts
                audio_frame.time_base = AUDIO_TIME_BASE
                audio_pts += AUDIO_SAMPLES

                for packet in audio_stream.encode(audio_frame):
                    self.container.mux(packet)

                sleep_time = next_video_time - time.monotonic()
                if sleep_time > 0:
                    time.sleep(min(sleep_time, 0.01))

            print('开始 flush')

            for packet in video_stream.encode():
                self.container.mux(packet)

            for packet in audio_stream.encode():
                self.container.mux(packet)

            print('推流线程正常结束')

        except Exception as e:
            import traceback
            traceback.print_exc()
            print('推流异常:', repr(e))

        finally:
            container = self.container
            self.container = None

            if container is not None:
                try:
                    container.close()
                except Exception as e:
                    print('关闭异常:', repr(e))

            self.running = False
            print('子线程退出')


class Demo(QWidget):
    def __init__(self):
        super().__init__()
        self.setWindowTitle('PyAV RTMP 阻塞测试')
        self.resize(400, 180)

        self.label = QLabel('状态：未推流')
        self.button = QPushButton('推流')
        self.button.clicked.connect(self.on_push)

        layout = QVBoxLayout(self)
        layout.addWidget(self.label)
        layout.addWidget(self.button)

        self.streamer = RTMPStreamer(RTMP_URL)

        self.timer = QTimer(self)
        self.timer.timeout.connect(self.update_ui)
        self.timer.start(100)

        self.ui_tick = 0

    def on_push(self):
        print('主线程：点击推流')
        self.label.setText('状态：正在连接...')
        self.button.setEnabled(False)
        self.streamer.start()

    def update_ui(self):
        self.ui_tick += 1
        self.label.setText(f'状态：UI正常运行  {self.ui_tick}')

    def closeEvent(self, event):
        self.streamer.stop()
        event.accept()


if __name__ == '__main__':
    app = QApplication(sys.argv)
    window = Demo()
    window.show()
    sys.exit(app.exec())

[zhaoxinyuan996]

same result

import time
from fractions import Fraction

import av
import numpy as np


RTMP_URL = 'rtmp://127.0.0.1/live/test'

WIDTH = 1280
HEIGHT = 720
FPS = Fraction(30, 1)
SAMPLE_RATE = 24000


def streamer():
    print('[STREAM] 开始打开 RTMP')

    try:
        container = av.open(
            RTMP_URL,
            'w',
            format='flv',
        )

        print('[STREAM] av.open 完成')

        video_stream = container.add_stream(
            'h264_nvenc',
            rate=FPS
        )
        video_stream.width = WIDTH
        video_stream.height = HEIGHT
        video_stream.pix_fmt = 'nv12'
        video_stream.time_base = Fraction(1, 30)

        video_stream.options = {
            'preset': 'p1',
            'tune': 'll',
            'bf': '0',
            'b': '2000k',
        }

        audio_stream = container.add_stream(
            'aac',
            rate=SAMPLE_RATE
        )
        audio_stream.layout = 'mono'
        audio_stream.time_base = Fraction(1, SAMPLE_RATE)
        audio_stream.options = {
            'b': '64k',
        }

        print('[STREAM] stream 创建完成')

        video_pts = 0
        audio_pts = 0

        # 黑屏
        image = np.zeros(
            (HEIGHT, WIDTH, 3),
            dtype=np.uint8
        )

        # 静音
        silence = np.zeros(
            1024,
            dtype=np.int16
        )

        next_video = time.monotonic()
        next_audio = time.monotonic()

        while True:
            now = time.monotonic()

            # -------------------------
            # 视频
            # -------------------------
            if now >= next_video:
                frame = av.VideoFrame.from_ndarray(
                    image,
                    format='bgr24'
                )

                frame.pts = video_pts
                frame.time_base = Fraction(1, 30)

                video_pts += 1

                for packet in video_stream.encode(frame):
                    print('[STREAM] before video mux')

                    container.mux(packet)

                    print('[STREAM] after video mux')

                next_video += 1 / 30

            # -------------------------
            # 音频
            # -------------------------
            if now >= next_audio:
                frame = av.AudioFrame.from_ndarray(
                    silence.reshape(1, -1),
                    format='s16',
                    layout='mono'
                )

                frame.sample_rate = SAMPLE_RATE
                frame.pts = audio_pts
                frame.time_base = Fraction(1, SAMPLE_RATE)

                audio_pts += len(silence)

                for packet in audio_stream.encode(frame):
                    print('[STREAM] before audio mux')

                    container.mux(packet)

                    print('[STREAM] after audio mux')

                next_audio += len(silence) / SAMPLE_RATE

            time.sleep(0.001)

    except Exception as e:
        print('[STREAM] 异常:', repr(e))

    finally:
        print('[STREAM] streamer 结束')


def main():
    threading.Thread(
        target=streamer
    ).start()

    i = 0

    while True:
        i += 1
        print(
            f'[MAIN] {i}  '
            f'{time.strftime("%H:%M:%S")}',
            flush=True
        )
        time.sleep(1)


if __name__ == '__main__':
    main()```

[zhaoxinyuan996]

PyAV/av/container/output.py

Line 595 in be473b3

err_check(

I won't block the main thread now, but I don't know if this will cause freezing or other serious problems, please help me

        if self.ptr.pb == cython.NULL and not self.ptr.oformat.flags & lib.AVFMT_NOFILE:
            with cython.nogil:
                ret = lib.avio_open(cython.address(self.ptr.pb), name, lib.AVIO_FLAG_WRITE)
            err_check(ret)