Mizu——基于共享内存的零拷贝并行 R 与 Python
Mizu – zero-copy parallel R and Python over shared memory

原始链接: https://github.com/shikokuchuo/pymizu

pymizu 是一个基于 libmizu 的实验性 Python 库,用于基于共享内存的快速异步与并行执行。它实现亚微秒级任务往返,以及在进程之间甚至 R 之间对列表、字典、NumPy 数组和 Arrow 表格的零拷贝共享。它提供工作窃取池、无锁共享内存通道、可直接替代的 concurrent.futures 执行器、真实 Future、asyncio 支持、可复现的并行随机数生成器和无死锁的嵌套 map。Python 与 R 可通过同一核心互操作并进行零拷贝数据交换。基准测试显示其相对 ProcessPoolExecutor 和 ThreadPoolExecutor 有大幅加速。可从 PyPI 安装,适用于 64 位平台上的 Python 3.10+;可选 numpy 和 cloudpickle 扩展。API 为实验性,可能发生变化。

Mizu(R)和 pymizu(Python)通过 libmizu 提供基于共享内存的无锁异步并行计算。它们将任务往返时间降低到 1 微秒以下,而基于套接字的往返时间约为 100 微秒。工作进程从无锁队列中拉取任务,无需调度器或热路径系统调用。pymizu 提供可直接使用的 concurrent.futures 执行器、asyncio 支持,以及 NumPy 数组和 pandas、polars、pyarrow 数据帧的零拷贝传输;共享格式还支持 R 和 Python 池之间的零拷贝交换。它仅限单机且处于实验阶段,作者欢迎反馈和错误报告。
相关文章

原文

pymizu logo

PyPI ci codecov License: MIT Ruff Python 3.10+

Thread-speed async and parallel Python over shared memory.

A task round trip drops from ~100 µs over sockets to < 1 µs, so you can divide work as finely as you would with threads (see Benchmarks). Common data structures like lists, dicts, numpy arrays, and Arrow tables can be shared between Python and R as zero-copy views.

In the default CPython build, the GIL runs CPU-bound threads on one core at a time, so compute parallelism in Python usually means multiple processes. Processes also isolate failures: a worker that crashes does not take down the host with it.

Channels and work-stealing task pools run over POSIX shared memory (Linux, macOS) or Win32 file mappings (Windows): lock-free single-producer single-consumer rings with batched publication, spin-then-park waiting that stays out of the kernel on the hot path, and OS notification when a peer dies.

pymizu is built on libmizu, a C library for lock-free shared-memory IPC. mizu binds the same core for R, so Python and R can talk to each other.

  • Zero-copy numpy arrays and Arrow interchange (pyarrow, polars, pandas, duckdb)
  • Drop-in concurrent.futures.Executor — real Futures, wait, as_completed, asyncio.wrap_future
  • Event-driven asyncio — await tasks and maps, async for and recv_async() on channels — wake-ups as data arrives, no polling threads
  • Reproducible parallel random numbers (seed=) — identical results for any worker count or steal order
  • Nested parallel maps — a task can map over its own pool, deadlock-free

Lifecycle: experimental. The API is not yet stable and we may make changes in response to feedback.

Install from PyPI:

Or the development version from GitHub:

pip install git+https://github.com/shikokuchuo/pymizu

Requires Python 3.10 or later on a 64-bit platform. The extension compiles the vendored C core, so no system library is necessary.

Optional extras:

  • pymizu[numpy]: zero-copy array views and raw-tier array staging
  • pymizu[cloudpickle]: lambdas, closures, and local functions as pool tasks

To request an extra with the GitHub install, use pip install "pymizu[numpy] @ git+https://github.com/shikokuchuo/pymizu".

import pymizu

with pymizu.Pool.create(4) as pool:
    task = pool.submit(pow, 2, 16)
    print(task.collect(timeout=5))

Communication overhead against the stdlib concurrent.futures pools (Apple M4 Pro, from benchmarks/mizu-stdlib-bench.py):

Against ProcessPoolExecutor (tasks run in separate processes, with pickled payloads):

Benchmark pymizu ProcessPoolExecutor Speedup
Trivial task round trip 0.7 µs 94.2 µs 135x
Pipelined throughput, 1 worker 2,390,000 tasks/s 18,500 tasks/s 129x
Parallel map overhead, trivial function, 4 workers 0.4 µs/elt* 70.3 µs/elt 176x
Parallel map of 2,000 ~5 µs tasks, 4 workers 4.3 ms 129 ms 30x

Against ThreadPoolExecutor (tasks share one process, so the GIL caps CPU-bound work at a single core):

Benchmark pymizu ThreadPoolExecutor Speedup
Trivial task round trip 0.7 µs 9.5 µs 14x
Pipelined throughput, 1 worker 2,390,000 tasks/s 405,000 tasks/s 5.9x
Parallel map overhead, trivial function, 4 workers 0.4 µs/elt* 2.6 µs/elt 6.5x
Parallel map of 2,000 ~5 µs tasks, 4 workers 4.3 ms 50 ms 12x

benchmarks/mizu-bench.py runs the pymizu rows standalone. See how pymizu compares with Ray, Dask, joblib, and modern CPython.

* elt = element; microseconds of wall time per map element.

Drop-in concurrent.futures

PoolExecutor adapts a pool to the stdlib concurrent.futures.Executor interface, so code written against ProcessPoolExecutor drops in. submit returns real Future objects, so wait / as_completed / asyncio.wrap_future all work unchanged:

with pymizu.PoolExecutor.create(4) as ex:
    futures = [ex.submit(pow, 2, i) for i in range(4)]
    print([f.result() for f in futures])

Pool.map(fn, x) maps fn over x on the pool and returns a Map handle immediately, just as Pool.submit() returns a task. collect() on the handle waits and returns the values in order. The function and data cross into shared memory once. Workers divide the elements among themselves, claiming ranges from a shared cursor.

A map also nests: the mapped function can map over its own pool with pymizu.current_pool(), deadlock-free.

with pymizu.Pool.create(4) as pool:
    m = pool.map(abs, range(-5, 5))
    print(m.collect())

Pool workers can be R processes that run the mizu package, the R binding of the same core. pymizu.r_launcher() spawns them and pymizu.call() specifies the task. Here, R's built-in mtcars is summarized in R and read back into polars:

import polars as pl
import pymizu

with pymizu.Pool.create(4, launcher=pymizu.r_launcher()) as pool:
    task = pool.submit(
        pymizu.call(source="aggregate(mpg ~ cyl, data = mtcars, FUN = mean)")
    )
    print(pl.DataFrame(task.collect()))
  • A numpy array arrives in R as a numeric vector, and a polars, pyarrow, or pandas frame as a data frame
  • A vector or data frame arrives back as a numpy array or a pymizu.Frame
  • Arrow data (pyarrow tables and record batch readers, or any __arrow_c_stream__ producer) crosses the same way

The R interop page has the full contract: the portable subset, pymizu.Frame, zero-copy frames, and the dtype matrix. The mizu README shows the reverse direction: an R host that drives Python workers with mizu_py_launcher().

The documentation site covers the full package: channels and what crosses them, task pools (batch operations, nested tasks, the concurrent.futures executor, watching a running pool), the parallel map (scheduling, reproducible randomness, templates, prepared maps), R interop and the dtype matrix, deployment on Linux, and the API reference.

See CONTRIBUTING.md for the repository layout and contributor notes.

MIT. LICENSE.note has the RngStreams attribution for the vendored libmizu core.


Please note that this project is released with a Contributor Code of Conduct. By participating in this project you agree to abide by its terms.

联系我们 contact @ memedata.com