multiloop: Multi-Event-Loop Engine & Concurrency Toolkit for Python 3.14t

⚡ A high-performance multi-event-loop concurrency toolkit and Web server engine for Python 3.14t (Free-Threaded / No-GIL), powered by an ultra-fast Rust SIMD core.

中文版 (Chinese)

Table of Contents

• 1. Installation
• 2. Verified Performance & Benchmarks
• 3. Web Server CLI
• 4. Python API Usage
• 5. Architecture & Developer Guide
• 6. Miscellaneous & Community

1. Installation

Prerequisites

• Python 3.14+: Free-Threaded (no-GIL) build, e.g. python3.14t.
• Rust Toolchain: Stable (via rustup) to compile the _multiloop_core Rust extension.

Install with uv (Recommended)

# Add multiloop to your project
uv add multiloop

# Or install into the current active virtual environment
uv pip install multiloop

Install with pip

pip install multiloop

Build from source

# Clone and compile optimized release extension via maturin
git clone https://github.com/hankaihong1/multiloop.git
cd multiloop
maturin develop --release

2. Verified Performance & Benchmarks

Multi-Core Throughput (Python 3.14t)

multiloop achieves linear physical multi-core scalability without GIL bottlenecks.

3-round average benchmark metrics on Apple M1 (8 Cores, 8GB RAM, Python 3.14.6 Free-Threaded No-GIL):

Reproducing Benchmarks

Run the built-in cross-platform pure-Python socket benchmark suite directly on your system (zero external dependencies like ab required):

uv run python benchmarks/bench_asgi_throughput.py

3. Web Server CLI

multiloop includes a high-performance CLI server runner. Unlike traditional multi-process process managers (gunicorn -w 4 or uvicorn --workers 4), multiloop run operates across multi-threaded isolated event loops in a single process with shared memory and zero IPC serialization overhead.

Run a FastAPI Application

Create main.py:

from fastapi import FastAPI

app = FastAPI(title="My multiloop API")


@app.get("/")
def read_root():
    return {"message": "Hello from FastAPI on multiloop!"}

Start the multi-worker server with 1 command:

multiloop run main:app --port 8000 --workers 4 --reload

Open http://127.0.0.1:8000/docs to view the live interactive Swagger UI!

Run a Flask or Django Application

multiloop run automatically detects WSGI applications (PEP 3333). Create app.py:

from flask import Flask

app = Flask(__name__)


@app.route("/")
def index():
    return "Hello from Flask running on multiloop WSGI thread pool!"

Launch the WSGI application with multi-thread pool offloading:

multiloop run app:app --port 5000 --workers 4

CLI Parameter Reference

multiloop run <module:app> [OPTIONS]

4. Python API Usage

1. Multi-Core Thread Pool (asyncssh style)

import asyncio
import multiloop


async def heavy_task(x: int) -> int:
    await asyncio.sleep(0.01)
    return x * 2


async def main() -> None:
    # async with manages the pool lifecycle automatically
    async with multiloop.EventLoopThreadPool(num_threads=4) as pool:
        # Submit an async coroutine (shared queue + work-stealing scheduler)
        fut1 = pool.submit(heavy_task, 21)

        # Target a specific worker loop (stateful connection affinity)
        fut2 = pool.submit(heavy_task, 21, pin_to=0)

        # Await results computed across physical CPU cores
        print("Results:", await fut1, await fut2)  # Output: 42 42


if __name__ == "__main__":
    asyncio.run(main())

2. Go-Style Channel & select_channel

import asyncio
import multiloop


async def main() -> None:
    ch1: multiloop.Channel[str] = multiloop.Channel()
    ch2: multiloop.Channel[str] = multiloop.Channel()

    async def producer() -> None:
        await ch1.send("Data from Channel 1")
        ch1.close()

    asyncio.create_task(producer())

    # select_channel waits for the first channel that becomes ready
    selected_ch, val = await multiloop.select_channel(ch1, ch2)
    print(f"Received: {val}")


if __name__ == "__main__":
    asyncio.run(main())

3. Task Synchronization with AsyncWaitGroup

import asyncio
import multiloop


async def worker(name: str, wg: multiloop.AsyncWaitGroup) -> None:
    try:
        await asyncio.sleep(0.02)  # simulated asynchronous work
        print(f"worker {name} done")
    finally:
        wg.done()  # decrement counter safely on completion or error


async def main() -> None:
    wg = multiloop.AsyncWaitGroup()

    # Dispatch 5 tasks across a 4-thread pool
    async with multiloop.EventLoopThreadPool(num_threads=4) as pool:
        for i in range(5):
            wg.add()  # increment counter
            pool.submit(worker, f"task-{i}", wg)

        # Block until all tasks finish (counter reaches zero)
        await wg.wait()
        print("All workers finished cleanly!")


if __name__ == "__main__":
    asyncio.run(main())

4. Structured Concurrency with TaskGroup

import asyncio
import multiloop


async def fetch(name: str, delay: float) -> str:
    await asyncio.sleep(delay)  # simulated network latency
    return f"{name}: ok"


async def main() -> None:
    try:
        # fail_after sets an overall deadline for the entire taskgroup
        async with multiloop.fail_after(0.1):
            async with multiloop.TaskGroup() as tg:
                h1 = tg.start_soon(fetch, "fast", 0.01)
                h2 = tg.start_soon(fetch, "slow", 0.5)

            print(await h1, "|", await h2)
    except TimeoutError:
        print("Timed out: child tasks cancelled safely")


if __name__ == "__main__":
    asyncio.run(main())

Explore more standalone scripts in examples/.

5. Architecture & Developer Guide

Core Architecture

multiloop assigns an isolated asyncio event loop to each worker OS thread, backed by lock-free task queues and padded atomic metrics in Rust:

graph TD
    UserApp[User Application / FastAPI / ASGI 3.0] -->|pool.submit| Scheduler[Round-Robin Scheduler]
    
    subgraph multiloop Core Engine
        Scheduler -->|Round-Robin Notify| W1[Worker Loop Thread 1]
        Scheduler -->|Round-Robin Notify| W2[Worker Loop Thread 2]
        Scheduler -->|Round-Robin Notify| W3[Worker Loop Thread 3]
        
        W1 <-->|Atomic Metrics| RustCore[Rust C-Extension _multiloop_core]
        W2 <-->|Atomic Metrics| RustCore
        W3 <-->|Atomic Metrics| RustCore
    end
    
    subgraph Golang Concurrency Toolkit
        RustCore <--> FastChan[Channel / flume]
        FastChan <--> Select[multiloop.select_channel]
        FastChan <--> Context[AsyncContext]
        RustCore <--> WaitGroup[AsyncWaitGroup]
    end

Local Development & Testing Gates

# 1. Build and install release Rust extension
make develop

# 2. Run all linter & type checks (0 warnings, strict typing)
make lint

# 3. Run complete test suite (355+ tests)
make test

6. Miscellaneous & Community

Known Limits & Invariants

For formal invariants and concurrency design principles, see docs/CONCURRENCY.md.

Live Demo

Want to test a live production setup without writing code? multiloop-fastapi-demo is a real FastAPI application served directly by multiloop run / MultiloopASGIWorker without uvicorn:

git clone https://github.com/hankaihong1/multiloop-fastapi-demo
cd multiloop-fastapi-demo
uv sync
uv run python app.py        # then open http://127.0.0.1:8000

Community & License

• Complete API Reference: docs/API.md
• Choosing Primitives Guide: docs/CHOOSING.md
• CONTRIBUTING.md — Contribution Guide
• CHANGELOG.md — Changelog
• SECURITY.md — Security Policy
• CODE_OF_CONDUCT.md — Code of Conduct
• AGENTS.md — AI Development Guide
• License: MIT License. See LICENSE for details.