TurboPipe

Fast data piping for python

Description

TurboPipe primarily speeds up sending raw bytes from moderngl.Buffer into FFmpeg subprocesses

Features and optimizations:

• Zero-copy: Calls libc::write syscall in pointer math, avoiding intermediate allocations
• Rust: Optimized crates like crossbeam and dashmap for channels and data structures
• Chunks: Write in blocks of 8192 bytes, so the kernel is happy for IPC pipes (Unix)
• Threading: Doesn't block the Python GIL, allows to render next frame async
• Safe: Guarantees order, blocks if the memory is queued on any pipe

_{Note: Also check out ShaderFlow, where TurboPipe shines! 😉}

Installation

Simply add the turbopipe PyPI package to your pyproject.toml:

[project]
dependencies = ["turbopipe"]

Usage

Send any object that implements memoryview() (but not them directly!)¹

Foundations

On its simplest form, the two are equivalent:

# Whatever data you can get a memoryview
data = os.urandom(1000)

with open("/dev/null", "wb") as stream:

    # Native python method (sync)
    stream.write(data)

    # Fast turbopipe method (async)
    turbopipe.pipe(data, stream.fileno())

# Wait for queued writes
turbopipe.sync(data)

Alternatively, for subprocesses:

from subprocess import PIPE, Popen

# Must have stdin or named pipes
process = subprocess.Popen(
    ("sh", "-c", "cat > /dev/null"),
    stdin=PIPE,
)

# Faster than stdin.write(data)
turbopipe.pipe(data, process.stdin.fileno())

ModernGL

Framebuffers expose their data with the internal .mglo object:

import moderngl

ctx = moderngl.create_standalone_context()
buf = ctx.buffer(reserve=1000)

# Send to FFmpeg, named pipes, raw data files
turbopipe.pipe(buf.mglo, fileno)

However, TurboPipe shines in large data transfers for video encoding:

# Pseudocode for a video editor-like
buffer = ctx.buffer(reserve=width*height*3)
scene = Scene()

ffmpeg = subprocess.Popen(...)
fileno = ffmpeg.stdin.fileno()

while not scene.finished:
    scene.render_frame()

    # Waits on all pending pipes in this buffer
    turbopipe.sync(buffer.mglo)

    # Copy data so the next frame can be pre-rendered
    scene.fbo.read_into(buffer)

    # Queue the write into a worker thread
    turbopipe.pipe(buffer.mglo, fileno)

# Sync all buffers, cleanup, etc.
turbopipe.sync(buffer.mglo)
turbopipe.done(buffer.mglo)
ffmpeg.stdin.close()

See the examples directory for more, and ShaderFlow's usage of it!

Future work

Design is compromise:

• Split crate into a pure-rust and pyo3 bindings.
• Are eternal workers heavy on scheduling resources? ²
• Support untracked writes without waitgroup overhead. ³
• Store Py_buffer in Work and release in the worker (correctness).
• Support synchronizing all queued writes in a file descriptor. ⁴