Language: Python
Requires: Python >= 3.10 (tested on 3.12 and 3.14)
Dependencies: psutil
Mpop is a single-producer, multiple-consumer task queue for parallel processing in Python.
The main goal is simple: keep workers busy and use memory efficiently.
It does this by packing task data into fixed-size 128-byte slots that sit in shared memory, where workers batch-claim ranges using a bitmap coordination layer to reduce lock contention.
The framework uses a typed handler system where all function parameters must have type annotations.
Types are inspected once at registration time and pre-compiled into pack/unpack functions, so the hot path during task execution has zero per-argument branching.
The model is simply a (supervisor, set of workers) pair.
Supervisor is responsible for spawning, administrating, and logging worker processes. Workers execute tasks.
For efficiency the supervisor can run in minimal mode (display=False).
Workers fetch tasks from a shared queue in batches and copy them into a local queue.
The rationale behind this is:
- To support dynamic request handling and administrative operations
- To let the supervisor efficiently block incoming requests when the shared queue is full
- Reduce lock contention since workers claim ranges atomically instead of dequeueing one by one
The supervisor introduces a potential single point of failure. If supervisor is terminated, workers detect this via periodic health checks and also terminate.
Single universal slot. No variants.
┌──────────┬──────────┬──────────────────┬──────────────┐
│ tsk_id 8B│ fn_id 8B │ args 80B │ meta 32B │ = 128B
│ uint64 │ uint64 │ 10 × int64 │ │
└──────────┴──────────┴──────────────────┴──────────────┘
Each argument occupies exactly 8 bytes regardless of type.
int, float, bool are packed directly into the int64 slot using struct.
str, list, dict, tuple, set, bytes and other types that dont fit in 8 bytes go through an ArgsPool (multiprocessing.Manager dict) and the slot stores the pool_id reference.
This means you get up to 10 arguments per handler function.
DIRECT types (stored in slot, zero overhead):
int→ int64float→ float64bool→ bool (as int64)- Extended:
int8,uint8,int16,uint16,int32,uint32,float32
REFERENCE types (stored in ArgsPool, retrieved by workers):
str,list,dict,tuple,set,bytes, any custom class
The ArgsPool uses lazy initialization.
If your handlers only use DIRECT types, no Manager process is ever spawned and there's zero overhead from it.
from api import MpopApi, TaskResult
def multiply(a: int, b: int) -> int:
return a * b
def power(base: int, exp: int) -> int:
return base ** exp
HANDLERS = {
0x8000: multiply,
0x8001: power,
}
if __name__ == "__main__":
app = MpopApi(
workers=4,
display=True,
handler_module=__name__,
debug_delay=0.05,
)
app.register_handlers(handlers_dict=HANDLERS)
app.validate()
for i in range(100):
app.enqueue(multiply, a=i, b=10, tsk_id=i)
for i in range(2, 6):
app.enqueue(fn_id=0x8001, args=(i, 3), tsk_id=i + 100)
app.run()if __name__ == "__main__":
app = MpopApi(workers=4, display=True)
@app.task(fn_id=0x8000)
def multiply(a: int, b: int) -> int:
return a * b
app.enqueue(multiply, a=5, b=10)
app.run()Two ways to enqueue:
# Pythonic — pass function reference + keyword args
app.enqueue(multiply, a=10, b=3.14)
# Positional — pass fn_id + args tuple
app.enqueue(fn_id=0x8000, args=(10, 3.14))Both are zero-validation at enqueue time. Type checking already happend at registration.
app = MpopApi(
workers=4, # number of worker processes
queue_slots=4096, # shared queue size (MUST be power of 2)
display=True, # TTY supervisor display
auto_terminate=True, # auto send TERMINATE when idle
debug=False, # debug mode
debug_delay=0.0, # artificial delay per task (seconds)
handler_module=None, # module name containing HANDLERS dict
worker_batch_size=16, # max tasks per batch claim
poll_interval=0.05, # supervisor poll rate
idle_check_interval=10, # polls before checking idle state
queue_name="mpop", # shared memory name prefix
)queue_slots must be a power of 2. The supervisor will refuse to allocate otherwise.
When your handler takes str, list, or other non-primitive types, the data flows through ArgsPool automatically:
def process(name: str, items: list, scale: float) -> str:
return f"{name}: {sum(items) * scale}"
HANDLERS = {0x9000: process}
if __name__ == "__main__":
app = MpopApi(workers=2, handler_module=__name__)
app.register_handlers(handlers_dict=HANDLERS)
app.enqueue(process, name="batch", items=[10, 20, 30], scale=2.5)
app.run()The name and items params are REFERENCE type — they get stored in ArgsPool in the main process and retrived by the worker process that picks up the task. scale is DIRECT (float64) so it sits in the slot directly.
Workers automatically clean up pool entries after retrieval.
This is experimental. Things work but there are some fallbacks. Read this section before using.
macOS uses the spawn start method for multiprocessing. This means the entire module gets re-imported in each child process.
If MpopApi() is at module level without the guard, child processes will re-create it and you get an infinite spawn loop.
Linux uses fork so it generally works without the guard, but you should always use it anyway.
All handlers need to be registered before you call app.enqueue() or app.run().
The decorator style (@app.task()) works but its somewhat fragile:
- The decorated function must be defined before any enqueue calls
- If you define handlers in a different module and import them, the decorator might not have ran yet depending on import order
- For anything serious, the HANDLERS dict +
register_handlers()pattern is more reliable
If you enqueue a task with an fn_id that workers dont know about, you'll see [Error] tsk_id=N: Unknown fn_id 0xNNNN in the log. The task is silently dropped.
No exceptions. If even one parameter is missing a type hint, registration will fail with TypeResolutionError.
This is intentional: the type system compiles pack/unpack at registration and needs to know every parameter's type.
# Good
def multiply(a: int, b: int) -> int:
return a * b
# Bad — will fail at registration
def multiply(a, b):
return a * bThe slot has 10 int64 argument slots. If your function has more than 10 parameters it will error at registration.
If you need more arguments, pack related data into a dict or list (those go through ArgsPool as REFERENCE type).
REFERENCE types (str, list, dict, etc.) require inter-process communication through multiprocessing.Manager.\
For performance critical paths, try to stick to int, float, bool parameters when possible.
The Manager is lazily initialized — it only starts when you first enqueue a task with REFERENCE type argument. If you never use REFERENCE types, no Manager process is created.
Handlers can return values and they get wrapped into TaskResult internally. But there is currently no mechanism to send results back to the main process. Return values exist mainly for error handling; if a handler raises, the worker logs it and continues.
# Both work fine
def add(a: int, b: int) -> int:
return a + b
def add_v2(a: int, b: int):
result = a + b
return TaskResult(success=True, value=result)There are two ways handlers reach worker processes:
handlers_map — Built from your registry at run() time, pickled into each Process object. This is how @app.task() and register_handlers() work.
handler_module — Workers import the module by name and look for a HANDLERS dict. This is the handler_module=__name__ parameter.
Both can coexist. If the same fn_id shows up in both, handlers_map takes priority.
For the most reliable setup, use both:
app = MpopApi(handler_module=__name__)
app.register_handlers(handlers_dict=HANDLERS)This way, workers have two chances to discover each handler.
The shared queue is a circular buffer in shared memory. Default size is 4096 slots.
Scheduling uses batching and range-claiming to reduce lock contention.
Workers atomically claim a range of slots from the shared queue, copy them to their local queue, then process everything without holding any locks.
A bitmap (active_batches) tracks which workers currently hold claimed but uncommited ranges.
When the last worker finishes its batch, committed occupancy is updated so the producer can reclaim space.
To change queue size:
app = MpopApi(queue_slots=4096*2) # must be power of 2api/
├── __init__.py — exports
├── slot.py — 128B TaskSlot definition
├── types.py — type classification, pack/unpack codegen
├── errors.py — error codes and exceptions
├── registry.py — ArgsPool + FunctionRegistry
├── tasks.py — TaskResult, TaskDispatcher
├── queues.py — SharedTaskQueue, LocalTaskQueue
├── worker.py — worker process entry point
├── allocation.py — static allocation
├── supervisor.py — supervisor controller + TTY display
└── mpop.py — MpopApi user-facing class
- Result Collection: Returning task results back to the main process via shared memory or queue
- Argument Pool: Better cleanup strategy, capacity monitoring, maybe replacing Manager with SharedMemory-based pool
- Error Recovery: Worker failure detection, health check improvements, respawning
- Supervisor Recovery: Reconnecting a new supervisor with existing workers after failure. This has been challenging because recovery requires reliable method to re-link the supervisor with workers which introduces coordination and state-reconciliation problems.
- Meta Data: Building the 32B meta section to support timestamping, flexible logging levels
- Benchmarking: Comparative analysis against multiprocessing.Pool and concurrent.futures