TurboPipe

Fast data piping for python

🔥 Description

TurboPipe speeds up sending raw bytes from moderngl.Buffer objects primarily to FFmpeg subprocess

The optimizations involved are:

• Zero-copy: Avoid unnecessary memory copies or allocation (intermediate buffer.read())
• Chunks: Write in blocks of 8192 bytes (RAM page size), so the hardware is happy (Unix)
• Threaded:
  • Doesn't block the Python GIL, allows to render next frame
  • Decouples the main thread from the I/O thread for performance
• Rust: The core of TurboPipe is written in Rust for speed, efficiency and low-level control
• 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)
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. ↩