import io
import fractions
import logging
logging.basicConfig(level=logging.DEBUG)
import av
import av.logging
av.logging.set_level(av.logging.DEBUG)
import numpy as np
def make_rgb_frame(width: int, height: int, t: int) -> av.VideoFrame:
y, x = np.mgrid[0:height, 0:width]
r = ((x + 3 * t) % 256).astype(np.uint8)
g = ((y + 2 * t) % 256).astype(np.uint8)
b = (((x + y) // 2 + 5 * t) % 256).astype(np.uint8)
img = np.stack([r, g, b], axis=-1)
return av.VideoFrame.from_ndarray(img, format="rgb24")
def run_mux1by1_segfault():
out_path = io.BytesIO()
width = 224
height = 240
fps = 30
num_frames = 90
container = av.open(out_path, format="mp4", mode="w")
container.add_stream("ffv1", rate=fps)
container.add_stream("ffv1", rate=fps)
for i in range(num_frames):
for vstream in container.streams.video:
if i == 0:
vstream.format = av.VideoFormat("bgr0", width=width, height=height)
frame = make_rgb_frame(width, height, i).reformat(format=vstream.format)
frame.pts = i
frame.time_base = fractions.Fraction(1, fps)
print("encoding frame", frame)
container.mux(vstream.encode(frame))
for vstream in container.streams.video:
container.mux(vstream.encode())
container.close()
print(f"Wrote {out_path}")
def run_muxdelayed_nosegfault():
out_path = io.BytesIO()
width = 224
height = 240
fps = 30
num_frames = 90
container = av.open(out_path, format="mp4", mode="w")
container.add_stream("ffv1", rate=fps)
container.add_stream("ffv1", rate=fps)
for i in range(num_frames):
packets = []
for vstream in container.streams.video:
if i == 0:
vstream.format = av.VideoFormat("bgr0", width=width, height=height)
frame = make_rgb_frame(width, height, i).reformat(format=vstream.format)
frame.pts = i
frame.time_base = fractions.Fraction(1, fps)
print("encoding frame", frame)
packets.extend(vstream.encode(frame))
container.mux(packets)
for vstream in container.streams.video:
container.mux(vstream.encode())
container.close()
print(f"Wrote {out_path}")
# NOTE segfault
# run_mux1by1_segfault()
# NOTE does not segfault
# run_muxdelayed_nosegfault()
# python3 -X faulthandler ...
# PYTHONMALLOC=malloc valgrind --leak-check=no --track-origins=yes python3 ...
PyAV version: 16.1.0
Steps to reproduce:
Create script:
Script content
Run
run_muxdelayed_nosegfault(), script finishes without segfault.Run
run_mux1by1_segfault(), script segfaults:Logs produced by `PYTHONMALLOC=malloc valgrind --leak-check=no --track-origins=yes python3 ...`