controlpool 1.1.0

Multiprocessing controllable pool

controlpool

Multiprocessing controllable pool for Python.

controlpool provides a Pool of workers (processes or threads) to execute tasks in parallel. What sets it apart is the optional controller process that can monitor and dynamically add/restart workers while the pool is running.

• ✅ Simple parallel map and streaming map_iterable
• ✅ Automatic task reassignment when a worker fails
• ✅ Per‑worker parameters via worker_params
• ✅ External controller for self‑healing and dynamic scaling
• ✅ Works with both multiprocessing and threading
• ✅ Context manager support

Installation

pip install controlpool

Quick Start

import controlpool as cp

def square(x):
    return x * x

with cp.Pool(square, processes=4) as pool:
    results = pool.map([1, 2, 3, 4, 5])

print(results)   # [1, 4, 9, 16, 25]

Core Concepts

Pool – a fixed number of workers (processes or threads) that apply a given function to each task.

Controller – an optional callable running in a separate process that can inspect and change the pool state (e.g., restart failed workers, add new ones).

Worker parameters – an iterable of per‑worker data that is passed as the first argument to the worker function.

Usage Examples

1. Basic parallel mapping

import controlpool as cp
import random

def sqrt(x):
    return x ** 0.5

rnd = random.Random(42)
data = [rnd.random() for _ in range(20)]

with cp.Pool(sqrt, 3) as pool:
    results = pool.map(data)

2. Worker parameters

Pass a list of parameters, one per worker. The function receives (param, arg) for each task.

import controlpool as cp

def multiply(param, x):
    return param * x

pool = cp.Pool(multiply, processes=3, worker_params=[2, 3, 4])
res = pool.map([10, 20, 30])
pool.close()

# Each worker uses its own multiplier

3. Automatic recovery with a controller

When a worker raises an exception, the pool reassigns the task to another worker. A controller can monitor for failures and restart dead workers.

import controlpool as cp
import random
import time

def fragile_sqrt(x):
    if random.randrange(2):    # fail 50% of the time
        raise ValueError("worker error")
    return x ** 0.5

def controller(term):
    try:
        while True:
            time.sleep(0.01)
            for wid, info in enumerate(term.get_info().workers_info):
                if info.state == cp.WorkerState.EXCEPTION:
                    term.change_worker_state(wid, cp.WorkerState.RUN)
    except cp.TerminateException:
        pass

with cp.Pool(fragile_sqrt, processes=4,
             controller=controller,
             raise_if_fail=False) as pool:
    results = pool.map(range(100))

4. Streaming results with map_iterable

Use map_iterable to get results as they are computed, without waiting for the whole list.

import controlpool as cp

def sqrt(x):
    return x**0.5

with cp.Pool(sqrt, 3) as pool:
    for result in pool.map_iterable(range(10)):
        print(result)

The generator behaves like a normal iterator but raises an error if you try to run multiple iterators simultaneously.

5. Using threads instead of processes

For I/O‑bound tasks (or when you need shared state) just set use_threading=True.

import controlpool as cp

def io_task(url):
    ...

urls = ["example.com", "example.org"]

with cp.Pool(io_task, processes=8, use_threading=True) as pool:
    results = pool.map(urls)

6. Passing marks to workers

Marks are labels that are not passed to the function but can be inspected through the pool info.

import controlpool as cp

def f(x):
    ...

pool = cp.Pool(f, processes=2, marks=['worker-A', 'worker-B'])

API Reference

The complete API documentation is available at controlpool Sphinx documentation

Why controlpool?

• Self‑healing – with a controller your pool can survive transient worker errors.
• Transparent – inspect every worker's state and task at runtime.
• Flexible – each worker can have its own initialisation parameters.
• Lightweight – zero dependencies beyond the standard library.

Contributing

We welcome your contributions! Here’s how you can participate:

🐛 Report a Bug or Suggest a Feature

Work Items: Our primary channel for tracking work. Go to issue tracker to create a new item. Choose the type Issue for bugs or feature requests. If you’re unsure, just use the default type.

Service Desk: Prefer to stay in your email client? Send your report or suggestion to our dedicated Service Desk address: [e-mail]. GitLab will automatically create a issue from your email, and you’ll get updates directly in your inbox.

💡 Propose a Code Change

Merge Request (MR): The standard GitLab way to contribute code. Fork the repository, create a branch for your changes, and submit a Merge Request. In the MR description, link the related issue (e.g., Closes #123) to connect your code with the discussion. See Gitlab documentation

⚙️ CI/CD Integration

Every Merge Request automatically triggers our CI/CD pipeline. Check the Pipelines tab to see the status of your changes (tests, linters, etc.). A green pipeline is a strong signal that your contribution is ready to be reviewed.

We appreciate your help in making controlpool better!

License

This project is licensed under the GNU General Public License v3.0 (GPLv3). You can find the full license text in the LICENSE file or at https://www.gnu.org/licenses/gpl-3.0.html.