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 memoryviewdata=os.urandom(1000)
withopen("/dev/null", "wb") asstream:
# Native python method (sync)stream.write(data)
# Fast turbopipe method (async)turbopipe.pipe(data, stream.fileno())
# Wait for queued writesturbopipe.sync(data)

Alternatively, for subprocesses:

fromsubprocessimportPIPE, Popen# Must have stdin or named pipesprocess=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:

importmodernglctx=moderngl.create_standalone_context()
buf=ctx.buffer(reserve=1000)
# Send to FFmpeg, named pipes, raw data filesturbopipe.pipe(buf.mglo, fileno)

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

# Pseudocode for a video editor-likebuffer=ctx.buffer(reserve=width*height*3)
scene=Scene()
ffmpeg=subprocess.Popen(...)
fileno=ffmpeg.stdin.fileno()
whilenotscene.finished:
scene.render_frame()
# Waits on all pending pipes in this bufferturbopipe.sync(buffer.mglo)
# Copy data so the next frame can be pre-renderedscene.fbo.read_into(buffer)
# Queue the write into a worker threadturbopipe.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. ⁴

Footnotes

1. According to the Python docs on buffers, the view->obj is a reference to the exporter, so using pipe(memoryview(data), file) there is no way for turbopipe to know who is the original data object for synchronization methods (only the ephemeral memoryview). ↩

2. Stopping workers midway has non-trivial concurrency problems, like needing a new WaitGroup per file descriptor to block .pipe() creating a new thread while due exiting (unecessary overhead). ↩

3. Simple to implement, but most realistic usage needs to sync at some point, and reutilize buffers. Strong argument is to not create an ephemeral waitgroup overhead. ↩

4. Nice for minimal code or rotating/untrackable/dangling data sources, however the decision was know your data-first, controlling the truth for pipes and syncs. ↩