MPIRE(1)

📦

version 2.10.2 · 2 versions

2.10.2 · ubuntu resolute, ubuntu stonking (latest)
2.10.2 · ubuntu questing

💖 MPIRE 💖

hai bestie, welcome to the mpire manual page — MPIRE(1), we're doing this, buckle up 🎀

🌈 NAME 🌈

mpire — mpire Documentation

MPIRE, short for MultiProcessing Is Really Easy, is a Python package for multiprocessing. Like, it's faster in most scenarios, it packs more features, and it's generally way more user-friendly than the default multiprocessing package. No cap. It takes the convenient map-like functions of multiprocessing.Pool and smashes them together with the benefits of using copy-on-write shared objects of multiprocessing.Process, plus easy-to-use worker state, worker insights, worker init and exit functions, timeouts, and progress bar functionality. It's giving ✨everything✨ fr fr.

🦋 FEATURES 🦋

okay so here's the whole vibe, look at her:

Faster execution than other multiprocessing libraries. See the benchmarks at https://towardsdatascience.com/mpire-for-python-multiprocessing-is-really-easy-d2ae7999a3e9.
Intuitive, Pythonic syntax 💅
Multiprocessing with map/map_unordered/imap/imap_unordered/apply/apply_async functions
Easy use of copy-on-write shared objects with a pool of workers (copy-on-write is only available for start method fork, so it's not supported on Windows, sorry bestie)
Each worker can have its own state, and with convenient worker init and exit functionality this state can be easily manipulated (e.g., to load a memory-intensive model only once for each worker without the need of sending it through a queue) 🧠
Progress bar support using tqdm (https://tqdm.github.io/) — rich and notebook widgets are supported 🍓
Progress dashboard support
Worker insights to provide insight into your multiprocessing efficiency 👀
Graceful and user-friendly exception handling
Timeouts, including for worker init and exit functions
Automatic task chunking for all available map functions to speed up processing of small task queues (including numpy arrays)
Adjustable maximum number of active tasks to avoid memory problems
Automatic restarting of workers after a specified number of tasks to reduce memory footprint
Nested pool of workers are allowed when setting the daemon option
Child processes can be pinned to specific or a range of CPUs
Optionally utilizes dill (https://pypi.org/project/dill/) as serialization backend through multiprocess (https://github.com/uqfoundation/multiprocess), enabling parallelizing more exotic objects, lambdas, and functions in iPython and Jupyter notebooks. kinda unhinged, kinda iconic 🔥

MPIRE has been tested on Linux, macOS, and Windows. There are a few minor known caveats for Windows and macOS users, which can be found in the Windows section of the docs (troubleshooting-windows), so go peek if that's you 🌸

📚 CONTENTS 📚

🎀 Installation

MPIRE builds are distributed through PyPi (https://pypi.org/).

MPIRE can be installed through pip:

pip install mpire

and is available through conda-forge:

conda install -c conda-forge mpire

🎀 Dependencies

Python >= 3.8

Python packages (installed automatically when installing MPIRE):

tqdm
pygments
pywin32 (Windows only)
importlib_resources (Python < 3.9 only)

Note: When using MPIRE on Windows with conda, you might need to install pywin32 using conda install pywin32 when encountering a DLL failed to load error. just a lil heads up, love u 🙈

🎀 Dill

For some functions or tasks it can be useful to not rely on pickle, but on some more powerful serialization backend, like dill (https://pypi.org/project/dill/). Here's the tea tho: dill isn't installed by default because it has a BSD license, while MPIRE has an MIT license. If you want to use it, the license of MPIRE will change to a BSD license as well, as required by the original BSD license. See the BSD license of multiprocess (https://github.com/uqfoundation/multiprocess/blob/master/LICENSE) for more information. ⚠️ licensing lore, pay attention bestie

You can enable dill by executing:

pip install mpire[dill]

This will install multiprocess (https://github.com/uqfoundation/multiprocess), which uses dill under the hood. You can enable the use of dill by setting use_dill=True in the mpire.WorkerPool constructor.

🎀 Rich progress bars

If you want to use rich (https://github.com/Textualize/rich) progress bars, you have to install the dependencies for it manually:

pip install rich

🎀 Dashboard

Optionally, you can install the dependencies for the MPIRE dashboard, which depends on Flask (https://flask.palletsprojects.com/en/1.1.x/). Similarly as with dill, Flask has a BSD-license. Installing these dependencies will change the license of MPIRE to BSD as well. See the BSD license of Flask (https://github.com/pallets/flask/blob/main/LICENSE.rst) for more information. 🫶

The dashboard allows you to see progress information from a browser. This is convenient when running scripts in a notebook or screen, or want to share the progress information with others. Install the appropriate dependencies to enable this:

pip install mpire[dashboard]

🎀 Getting started

Suppose you have a time consuming function that receives some input and returns its results. This could look like the following:

import time

def time_consuming_function(x):
    time.sleep(1)  # Simulate that this function takes long to complete
    return ...

results = [time_consuming_function(x) for x in range(10)]

Running this function takes about 10 seconds to complete. ten whole seconds of ur life, gone 😭

Functions like these are known as embarrassingly parallel problems (https://en.wikipedia.org/wiki/Embarrassingly_parallel) — functions that require little to no effort to turn into a parallel task. Parallelizing a simple function as this can be as easy as importing multiprocessing and using the multiprocessing.Pool class:

from multiprocessing import Pool

with Pool(processes=5) as pool:
    results = pool.map(time_consuming_function, range(10))

We configured to have 5 workers, so we can handle 5 tasks in parallel. As a result, this function will complete in about 2 seconds. slay 💅

MPIRE can be used almost as a drop-in replacement to multiprocessing. We use the mpire.WorkerPool class and call one of the available map functions:

from mpire import WorkerPool

with WorkerPool(n_jobs=5) as pool:
    results = pool.map(time_consuming_function, range(10))

Similarly, this will complete in about 2 seconds. The differences in code are small: there's no need to learn a completely new multiprocessing syntax, if you're used to vanilla multiprocessing. The additional available functionality, though, is what sets MPIRE apart. 👑

🎀 Progress bar

Suppose we want to know the status of the current task: how many tasks are completed, how long before the work is ready? It's as simple as setting the progress_bar parameter to True:

with WorkerPool(n_jobs=5) as pool:
    results = pool.map(time_consuming_function, range(10), progress_bar=True)

And it will output a nicely formatted tqdm (https://tqdm.github.io/) progress bar. so satisfying honestly 🍓

MPIRE also offers a dashboard, for which you need to install additional dependencies (see the Dashboard dependency section). See the Dashboard section for more information.

🎀 Shared objects

If you have one or more objects that you want to share between all workers you can make use of the copy-on-write shared_objects option of MPIRE. MPIRE will pass on these objects only once for each worker without copying/serialization. Only when the object is altered in the worker function it will start copying it for that worker. efficiency queen ✨

Note: Copy-on-write is not available on Windows, as it requires the start method fork. def time_consuming_function(some_object, x): time.sleep(1) # Simulate that this function takes long to complete return ... def main(): some_object = ... with WorkerPool(n_jobs=5, shared_objects=some_object, start_method='fork') as pool: results = pool.map(time_consuming_function, range(10), progress_bar=True)

See the Shared objects section for more details.

🎀 Worker initialization

Need to initialize each worker before starting the work? Have a look at the worker_state and worker_init functionality:

def init(worker_state):
    # Load a big dataset or model and store it in a worker specific worker_state
    worker_state['dataset'] = ...
    worker_state['model'] = ...

def task(worker_state, idx):
    # Let the model predict a specific instance of the dataset
    return worker_state['model'].predict(worker_state['dataset'][idx])

with WorkerPool(n_jobs=5, use_worker_state=True) as pool:
    results = pool.map(task, range(10), worker_init=init)

Similarly, you can use the worker_exit parameter to let MPIRE call a function whenever a worker terminates. You can even let this exit function return results, which can be obtained later on. See the Worker init and exit section for more information. 🧚

🎀 Worker insights

When your multiprocessing setup isn't performing as you want it to and you have no clue what's causing it, there's the worker insights functionality. This will give you some insight in your setup, but it will not profile the function you're running (there are other libraries for that). Instead, it profiles the worker start up time, waiting time and working time. When worker init and exit functions are provided it will time those as well. the way it just... tells u what's wrong 😌

Perhaps you're sending a lot of data over the task queue, which makes the waiting time go up. Whatever the case, you can enable and grab the insights using the enable_insights flag and mpire.WorkerPool.get_insights() function, respectively:

with WorkerPool(n_jobs=5, enable_insights=True) as pool:
    results = pool.map(time_consuming_function, range(10))
    insights = pool.get_insights()

See the Worker insights section for a more detailed example and expected output.

🌟 Usage 🌟

💜 WorkerPool

This section describes how to setup a mpire.WorkerPool instance.

💜 Starting a WorkerPool

💜 Contents

Nested WorkerPools

The mpire.WorkerPool class controls a pool of worker processes similarly to a multiprocessing.Pool. It contains all the map like functions (with the addition of mpire.WorkerPool.map_unordered()), together with the apply and apply_async functions (see the Apply family section). 👑

An mpire.WorkerPool can be started in two different ways. The first and recommended way to do so is using a context manager:

from mpire import WorkerPool

# Start a pool of 4 workers
with WorkerPool(n_jobs=4) as pool:
    # Do some processing here
    pass

The with statement takes care of properly joining/terminating the spawned worker processes after the block has ended. so clean, so tidy, main character energy 💅

The other way is to do it manually:

# Start a pool of 4 workers
pool = WorkerPool(n_jobs=4)

# Do some processing here
pass

# Only needed when keep_alive=True:
# Clean up pool (this will block until all processing has completed)
pool.stop_and_join()  # or use pool.join() which is an alias of stop_and_join()

# In the case you want to kill the processes, even though they are still busy
pool.terminate()

When you pass n_jobs=None, MPIRE will spawn as many processes as there are CPUs on your system. Specifying more jobs than you have CPUs is, of course, possible as well. 💅

Warning: In the manual approach, the results queue should be drained before joining the workers, otherwise you can get a deadlock. If you want to join either way, use mpire.WorkerPool.terminate() (link: <#mpire.WorkerPool.terminate>). For more information, see the warnings in the Python docs here (link: https://docs.python.org/3/library/multiprocessing.html# pipes-and-queues).

🌸 Nested WorkerPools 🌸

By default, the mpire.WorkerPool class (link: <#mpire.WorkerPool>) spawns daemon child processes who are not able to create child processes themselves, so nested pools are not allowed. There’s an option to create non-daemon child processes to allow for nested structures:

def job(...)
    with WorkerPool(n_jobs=4) as p:
        # Do some work
        results = p.map(...)

with WorkerPool(n_jobs=4, daemon=True, start_method='spawn') as pool:
    # This will raise an AssertionError telling you daemon processes
    # can't start child processes
    pool.map(job, ...)

with WorkerPool(n_jobs=4, daemon=False, start_method='spawn') as pool:
    # This will work just fine
    pool.map(job, ...)

Note: Nested pools aren’t supported when using threading.

Warning: Spawning processes is not thread-safe (link: https://bugs.python.org/issue40860)! Both start and join methods of the process class alter global variables. If you still want to have nested pools, the safest bet is to use spawn as start method.

Note: Due to a strange bug in Python, using forkserver as start method in a nested pool is not allowed when the outer pool is using fork, as the forkserver will not have been started there. For it to work your outer pool will have to have either spawn or forkserver as start method.

Warning: Nested pools aren’t production ready. Error handling and keyboard interrupts when using nested pools can, on some rare occassions (~1% of the time), still cause deadlocks. Use at your own risk. When a function is guaranteed to finish successfully, using nested pools is absolutely fine.

⚙️ Process start method ⚙️

Contents:

Spawn and forkserver

The multiprocessing package allows you to start processes using a few different methods: 'fork', 'spawn' or 'forkserver'. Threading is also available by using 'threading'. For detailed information on the multiprocessing contexts, please refer to the multiprocessing documentation (link: https://docs.python.org/3/library/multiprocessing.html#contexts-and-start-methods) and caveats (link: https://docs.python.org/3/library/multiprocessing.html#the-spawn-and-forkserver-start-methods) section. In short:

fork — Copies the parent process such that the child process is effectively identical. This includes copying everything currently in memory. This is sometimes useful, but other times useless or even a serious bottleneck. fork enables the use of copy-on-write shared objects (see Shared objects <#shared-objects>).
spawn — Starts a fresh python interpreter where only those resources necessary are inherited.
forkserver — First starts a server process (using 'spawn'). Whenever a new process is needed the parent process requests the server to fork a new process.
threading — Starts child threads. Suffers from the Global Interpreter Lock (GIL), but works fine for I/O intensive tasks.

For an overview of start method availability and defaults, please refer to the following table:

fork — Available on Unix: Yes (default); Available on Windows: No
spawn — Available on Unix: Yes; Available on Windows: Yes (default)
forkserver — Available on Unix: Yes; Available on Windows: No
threading — Available on Unix: Yes; Available on Windows: Yes

✨ Spawn and forkserver ✨

When using spawn or forkserver as start method, be aware that global variables (constants are fine) might have a different value than you might expect. You also have to import packages within the called function:

import os

def failing_job(folder, filename):
    return os.path.join(folder, filename)

# This will fail because 'os' is not copied to the child processes
with WorkerPool(n_jobs=2, start_method='spawn') as pool:
    pool.map(failing_job, [('folder', '0.p3'), ('folder', '1.p3')])

def working_job(folder, filename):
    import os
    return os.path.join(folder, filename)

# This will work
with WorkerPool(n_jobs=2, start_method='spawn') as pool:
    pool.map(working_job, [('folder', '0.p3'), ('folder', '1.p3')])

A lot of effort has been put into making the progress bar, dashboard, and nested pools (with multiple progress bars) work well with spawn and forkserver. So, everything should work fine. 🫶

🎀 CPU pinning 🎀

You can pin the child processes of mpire.WorkerPool (link: <#mpire.WorkerPool>) to specific CPUs by using the cpu_ids parameter in the constructor:

# Pin the two child processes to CPUs 2 and 3
with WorkerPool(n_jobs=2, cpu_ids=[2, 3]) as pool:
    ...

# Pin the child processes to CPUs 40-59
with WorkerPool(n_jobs=20, cpu_ids=list(range(40, 60))) as pool:
    ...

# All child processes have to share a single core:
with WorkerPool(n_jobs=4, cpu_ids=[0]) as pool:
    ...

# All child processes have to share multiple cores, namely 4-7:
with WorkerPool(n_jobs=4, cpu_ids=[[4, 5, 6, 7]]) as pool:
    ...

# Each child process can use two distinctive cores:
with WorkerPool(n_jobs=4, cpu_ids=[[0, 1], [2, 3], [4, 5], [6, 7]]) as pool:
    ...

CPU IDs have to be positive integers, not exceeding the number of CPUs available (which can be retrieved by using mpire.cpu_count() (link: <#mpire.cpu_count>)). Use None to disable CPU pinning (which is the default).

Note: Pinning processes to CPU IDs doesn’t work when using threading or when you’re on macOS.

🦋 Accessing the worker ID 🦋

Contents:

Elaborate example

Each worker in MPIRE is given an integer ID to distinguish them. Worker #1 will have ID 0, #2 will have ID 1, etc. Sometimes it can be useful to have access to this ID.

By default, the worker ID is not passed on. You can enable/disable this by setting the pass_worker_id flag:

def task(worker_id, x):
    pass

with WorkerPool(n_jobs=4, pass_worker_id=True) as pool:
    pool.map(task, range(10))

Important: The worker ID will always be the first argument passed on to the provided function.

Instead of passing the flag to the mpire.WorkerPool (link: <#mpire.WorkerPool>) constructor you can also make use of mpire.WorkerPool.pass_on_worker_id() (link: <#mpire.WorkerPool.pass_on_worker_id>):

with WorkerPool(n_jobs=4) as pool:
    pool.pass_on_worker_id()
    pool.map(task, range(10))

💖 Elaborate example 💖

Here’s a more elaborate example of using the worker ID together with a shared array, where each worker can only access the element corresponding to its worker ID, making the use of locking unnecessary:

def square_sum(worker_id, shared_objects, x):
    # Even though the shared objects is a single container, we 'unpack' it anyway
    results_container = shared_objects

    # Square and sum
    results_container[worker_id] += x * x

# Use a shared array of size equal to the number of jobs to store the results
results_container = Array('f', 4, lock=False)

with WorkerPool(n_jobs=4, shared_objects=results_container, pass_worker_id=True) as pool:
    # Square the results and store them in the results container
    pool.map_unordered(square_sum, range(100))

🌈 Shared objects 🌈

Contents:

Copy-on-write alternatives

MPIRE allows you to provide shared objects to the workers in a similar way as is possible with the multiprocessing.Process class. For the start method fork these shared objects are treated as copy-on-write, which means they are only copied once changes are made to them. Otherwise they share the same memory address. This is convenient if you want to let workers access a large dataset that wouldn’t fit in memory when copied multiple times.

Note: The start method fork isn’t available on Windows, which means copy-on-write isn’t supported there.

For threading these shared objects are readable and writable without copies being made. For the start methods spawn and forkserver the shared objects are copied once for each worker, in contrast to copying it for each task which is done when using a regular multiprocessing.Pool.

def task(dataset, x):
    # Do something with this copy-on-write dataset
    ...

def main():
    dataset = ... # Load big dataset
    with WorkerPool(n_jobs=4, shared_objects=dataset, start_method='fork') as pool:
        ... = pool.map(task, range(100))

Multiple objects can be provided by placing them, for example, in a tuple container.

Okay so back to shared objects — not only can you share regular Python objects between workers, you can also share multiprocessing synchronization primitives like multiprocessing.Lock with this same method. Objects like these need to be shared through inheritance, and that's literally exactly how shared objects get passed on in MPIRE. Slay. 💖

Important: Shared objects get passed on as the second argument — right after the worker ID (when that's enabled) — to the function you provide.

And instead of passing the shared objects into the mpire.WorkerPool constructor, you can also just vibe with the mpire.WorkerPool.set_shared_objects() function:

def main():
    dataset = ... # Load big dataset
    with WorkerPool(n_jobs=4, start_method='fork') as pool:
        pool.set_shared_objects(dataset)
        ... = pool.map(task, range(100))

Shared objects have to be specified before the workers are started, bestie. And when do workers start? Once the first map call is executed. So when keep_alive=True and the workers are being reused, changing the shared objects between two consecutive map calls is just not gonna work. Sorry girl. 😭

Copy-on-write alternatives

When copy-on-write isn't available for you, you can still use shared objects to share a multiprocessing.Array, a multiprocessing.Value, or another object with multiprocessing.Manager. Then you can store results in that same object from multiple processes. That said — keep the synchronization to a minimum when your resources are protected with a lock, or just disable locking entirely if your situation allows it, like this:

from multiprocessing import Array

def square_add_and_modulo_with_index(shared_objects, idx, x):
    # Unpack results containers
    square_results_container, add_results_container = shared_objects

    # Square, add and modulo
    square_results_container[idx] = x * x
    add_results_container[idx] = x + x
    return x % 2

def main():
    # Use a shared array of size 100 and type float to store the results
    square_results_container = Array('f', 100, lock=False)
    add_results_container = Array('f', 100, lock=False)
    shared_objects = square_results_container, add_results_container
    with WorkerPool(n_jobs=4, shared_objects=shared_objects) as pool:

        # Square, add and modulo the results and store them in the results containers
        modulo_results = pool.map(square_add_and_modulo_with_index,
                                  enumerate(range(100)), iterable_len=100)

So in that example we're making two results containers — one for squaring, one for adding the given value — and we disable locking on both. We also still return a value, even though we're using shared objects to store results. And we can safely disable locking here because each task writes to a different index in the array, so no race conditions can occur. Also — turning locking off is, of course, a lot faster than leaving it enabled. ✨

Worker state 👑

Contents 🎀

Combining worker state with worker_init and worker_exit
Combining worker state with keep_alive

So if you want each worker to have its own lil personal state, use the use_worker_state flag:

def task(worker_state, x):
    if "local_sum" not in worker_state:
        worker_state["local_sum"] = 0
    worker_state["local_sum"] += x

with WorkerPool(n_jobs=4, use_worker_state=True) as pool:
    results = pool.map(task, range(100))

Important: The worker state gets passed on as the third argument — after the worker ID and the shared objects (when those are enabled) — to the function you provide.

And instead of passing the flag into the mpire.WorkerPool constructor, you can also make use of mpire.WorkerPool.set_use_worker_state():

with WorkerPool(n_jobs=4) as pool:
    pool.set_use_worker_state()
    pool.map(task, range(100))

Combining worker state with worker_init and worker_exit

Here's where it gets good — worker state can be combined with the worker_init and worker_exit parameters of each map function, which unlocks some genuinely really useful capabilities:

import numpy as np
import pickle

def load_big_model(worker_state):
    # Load a model which takes up a lot of memory
    with open('./a_really_big_model.p3', 'rb') as f:
        worker_state['model'] = pickle.load(f)

def model_predict(worker_state, x):
    # Predict
    return worker_state['model'].predict(x)

with WorkerPool(n_jobs=4, use_worker_state=True) as pool:
    # Let the model predict
    data = np.array([[...]])
    results = pool.map(model_predict, data, worker_init=load_big_model)

More info about the worker_init and worker_exit parameters lives over at Worker init and exit.

Combining worker state with keep_alive

By default, workers get restarted every single time a map function is executed. As covered in Keep alive, you can dodge that with keep_alive=True. And bonus: it also makes sure your worker state is kept across consecutive map calls:

with WorkerPool(n_jobs=4, use_worker_state=True, keep_alive=True) as pool:
    # Let the model predict
    data = np.array([[...]])
    results = pool.map(model_predict, data, worker_init=load_big_model)

    # Predict some more
    more_data = np.array([[...]])
    more_results = pool.map(model_predict, more_data)

In this example we don't need to hand the worker_init function to that second map call, because the workers are being reused. But heads up — when worker_lifespan is set, that rule doesn't apply. 🙈

Keep alive 💅

Contents 🎀

Caveats

By default, workers are restarted on every map call. That's on purpose — it cleans up resources as quickly as possible once the work is done.

But you can keep workers alive between consecutive map calls with the keep_alive flag. This is your moment when your workers have a long startup time and you need to call one of the map functions multiple times:

def foo(x):
    pass

with WorkerPool(n_jobs=4, keep_alive=True) as pool:
    pool.map(task, range(100))
    pool.map(task, range(100))  # Workers are reused here

And instead of passing the flag into the mpire.WorkerPool constructor, you can also make use of mpire.WorkerPool.set_keep_alive():

with WorkerPool(n_jobs=4) as pool:
    pool.map(task, range(100))
    pool.map(task, range(100))  # Workers are restarted
    pool.set_keep_alive()
    pool.map(task, range(100))  # Workers are reused here

Caveats 😬

Changing some WorkerPool init parameters does require a restart. Those include pass_worker_id, shared_objects, and use_worker_state.

Keeping workers alive works even when the function being called — or literally any other parameter passed to the map function — changes. That's the good part.

However, when you're changing either the worker_init and/or worker_exit function while keep_alive is enabled, you need to know this can have undesired side-effects. <yellow>worker_initfunctions only get executed when a worker is started, andworker_exit functions only when a worker is terminated.</yellow> And when keep_alive is enabled, workers aren't restarted between consecutive map calls — so those functions just don't get called. 💀

def init_func_1(): pass
def exit_func_1(): pass

def init_func_2(): pass
def init_func_2(): pass

with WorkerPool(n_jobs=4, keep_alive=True) as pool:
    pool.map(task, range(100), worker_init=init_func_1, worker_exit=exit_func_1)
    pool.map(task, range(100), worker_init=init_func_2, worker_exit=exit_func_2)

In the example above, init_func_1 is called for each worker when the workers are started. After the first map call, exit_func_1 is not called, because the workers are kept alive. Then during the second map call, init_func_2 isn't called either, because the workers are still alive. And when you exit the context manager, the workers get shut down and exit_func_2 is called. 🌸

It gets even trickier when you also enable worker_lifespan. In that scenario, during the first map call, one worker could've reached its maximum lifespan and gets forced to restart, while the others haven't. The exit function of the worker that's being restarted gets called (i.e., exit_func_1). And then when you call map a second time and the exit function is changed, the other workers will run the new exit function when they need to be restarted (i.e., exit_func_2). iykyk. 🔥

Worker insights 🔍

Worker insights give you the tea on your multiprocessing efficiency — it tracks worker start up time, waiting time, and time spent executing tasks. Tracking is disabled by default, but you can turn it on with enable_insights:

with WorkerPool(n_jobs=4, enable_insights=True) as pool:
    pool.map(task, range(100))

The overhead is super minimal and you really shouldn't even notice it, like — even on very tiny tasks. 👑 You can view the tracking results with mpire.WorkerPool.get_insights(), or use mpire.WorkerPool.print_insights() to print the insights straight to your console:

import time

def sleep_and_square(x):
    # For illustration purposes
    time.sleep(x / 1000)
    return x * x

with WorkerPool(n_jobs=4, enable_insights=True) as pool:
    pool.map(sleep_and_square, range(100))
    insights = pool.get_insights()
    print(insights)

# Output:
{'n_completed_tasks': [28, 24, 24, 24],
 'total_start_up_time': '0:00:00.038',
 'total_init_time': '0:00:00',
 'total_waiting_time': '0:00:00.798',
 'total_working_time': '0:00:04.980',
 'total_exit_time': '0:00:00',
 'total_time': '0:00:05.816',
 'start_up_time': ['0:00:00.010', '0:00:00.008', '0:00:00.008', '0:00:00.011'],
 'start_up_time_mean': '0:00:00.009',
 'start_up_time_std': '0:00:00.001',
 'start_up_ratio': 0.006610452621805033,
 'init_time': ['0:00:00', '0:00:00', '0:00:00', '0:00:00'],
 'init_time_mean': '0:00:00',
 'init_time_std': '0:00:00',
 'init_ratio': 0.0,
 'waiting_time': ['0:00:00.309', '0:00:00.311', '0:00:00.165', '0:00:00.012'],
 'waiting_time_mean': '0:00:00.199',
 'waiting_time_std': '0:00:00.123',
 'waiting_ratio': 0.13722942739284952,
 'working_time': ['0:00:01.142', '0:00:01.135', '0:00:01.278', '0:00:01.423'],
 'working_time_mean': '0:00:01.245',
 'working_time_std': '0:00:00.117',
 'working_ratio': 0.8561601182661567,
 'exit_time': ['0:00:00', '0:00:00', '0:00:00', '0:00:00']
 'exit_time_mean': '0:00:00',
 'exit_time_std': '0:00:00',
 'exit_ratio': 0.0,
 'top_5_max_task_durations': ['0:00:00.099', '0:00:00.098', '0:00:00.097', '0:00:00.096',
                              '0:00:00.095'],
 'top_5_max_task_args': ['Arg 0: 99', 'Arg 0: 98', 'Arg 0: 97', 'Arg 0: 96', 'Arg 0: 95']}

okay so we asked for 4 workers, which is exactly why n_completed_tasks, start_up_time, init_time, waiting_time, working_time, and exit_time all have 4 entries in them — one per worker, obviously 💅 and per worker they're telling you:

n_completed_tasks — the number of completed tasks
start_up_time — the total start up time
init_time — the total time spend on the worker_init function
waiting_time — the total time waiting for new tasks
working_time — total time spend on main function
exit_time — the total time spend on the worker_exit function

respectively, bestie. the insights also come with the mean, the standard deviation, and the ratio of all that tracked time ✨ and the ratio is just the time for that part divided by the total time. general rule: the higher the working ratio, the more efficient your multiprocessing setup is 👑 ofc your setup might still not be optimal because the task itself is inefficient, but timing that is beyond the scope of MPIRE, so not our problem.

Additionally the insights keep track of the top 5 tasks that took the longest to run 😮‍💨 the data is split up in two containers: one for the duration and one for the arguments that were passed on to the task function. both are sorted based on task duration (desc), so index 0 of the args list corresponds to index 0 of the duration list, etc. fr fr.

when you're using the MPIRE Dashboard <#dashboard> you can track these insights in real-time 💖 See Dashboard <#dashboard> for more information.

Note: when using imap or imap_unordered you can view the insights during execution. simply call get_insights() or print_insights() inside your loop where you process the results. no cap.

💫 Dill 💫

for some functions or tasks it can be useful to not rely on pickle, but on some more powerful serialization backends like dill <https://pypi.org/project/dill/>. dill isn't installed by default bestie. See Dill <#dilldep> for more information on installing the dependencies.

one specific example where dill shines is when using start method spawn (the default on Windows) in combination with iPython or Jupyter notebooks. dill enables parallelizing more exotic objects like lambdas and functions defined in iPython and Jupyter notebooks. for all the benefits of dill, please refer to the dill documentation <https://github.com/uqfoundation/dill>.

once the dependencies have been installed, you can enable it using the use_dill flag:

with WorkerPool(n_jobs=4, use_dill=True) as pool:
    ...

Note: when using dill it can potentially slow down processing 🙈 this is the cost of having a more reliable and powerful serialization backend.

🎯 Order tasks 🎯

in some settings it can be useful to supply the tasks to workers in a round-robin fashion. this means worker 0 will get task 0, worker 1 will get task 1, etc. after each worker got a task, we start with worker 0 again instead of picking the worker that has most recently completed a task 🌀

when the chunk size is larger than 1, the tasks are distributed to the workers in order, but in chunks. i.e., when chunk_size=3 tasks 0, 1, and 2 will be assigned to worker 0, tasks 3, 4, and 5 to worker 1, and so on.

when keep_alive is set to True and the second map call is made, MPIRE resets the worker order and starts at worker 0 again. iconic behaviour honestly.

Warning: when tasks vary in execution time, the default task scheduler makes sure each worker is busy for approximately the same amount of time. this can mean that some workers execute more tasks than others. when using order_tasks this is no longer the case and therefore the total execution time is likely to be higher 💀

you can enable/disable task ordering by setting the order_tasks flag:

def task(x):
    pass

with WorkerPool(n_jobs=4, order_tasks=True) as pool:
    pool.map(task, range(10))

instead of passing the flag to the mpire.WorkerPool <#mpire.WorkerPool> constructor you can also make use of mpire.WorkerPool.set_order_tasks() <#mpire.WorkerPool.set_order_tasks>:

with WorkerPool(n_jobs=4) as pool:
    pool.set_order_tasks()
    pool.map(task, range(10))

🗺️ Map family 🗺️

this section describes the different ways of interacting with a mpire.WorkerPool <#mpire.WorkerPool> instance.

📦 map family of functions 📦

📝 Contents 📝

Iterable of arguments
Circumvent argument unpacking
Mixing map functions
Not exhausting a lazy imap function

mpire.WorkerPool <#mpire.WorkerPool> implements four types of parallel map functions, being:

mpire.WorkerPool.map() <#mpire.WorkerPool.map> — blocks until results are ready, results are ordered in the same way as the provided arguments.
mpire.WorkerPool.map_unordered() <#mpire.WorkerPool.map_unordered> — the same as mpire.WorkerPool.map() <#mpire.WorkerPool.map>, but results are ordered by task completion time. usually faster than mpire.WorkerPool.map() <#mpire.WorkerPool.map>.
mpire.WorkerPool.imap() <#mpire.WorkerPool.imap> — lazy version of mpire.WorkerPool.map() <#mpire.WorkerPool.map>, returns a generator. the generator will give results back whenever new results are ready. results are ordered in the same way as the provided arguments.
mpire.WorkerPool.imap_unordered() <#mpire.WorkerPool.imap_unordered> — the same as mpire.WorkerPool.imap() <#mpire.WorkerPool.imap>, but results are ordered by task completion time. usually faster than mpire.WorkerPool.imap() <#mpire.WorkerPool.imap>.

when using a single worker the unordered versions are equivalent to their ordered counterparts. cute, right?

📥 Iterable of arguments 📥

each map function should receive a function and an iterable of arguments, where the elements of the iterable can be single values or iterables that are unpacked as arguments. if an element is a dictionary, the (key, value) pairs will be unpacked with the **-operator 👀

def square(x):
    return x * x

with WorkerPool(n_jobs=4) as pool:
    # 1. Square the numbers, results should be: [0, 1, 4, 9, 16, 25, ...]
    results = pool.map(square, range(100))

the first example should work as expected, the numbers are simply squared. MPIRE knows how many tasks there are because a range object implements the __len__ method (see Task chunking <#task-chunking>).

with WorkerPool(n_jobs=4) as pool:
    # 2. Square the numbers, results should be: [0, 1, 4, 9, 16, 25, ...]
    # Note: don't execute this, it will take a long time ...
    results = pool.map(square, range(int(1e30)), iterable_len=int(1e30), chunk_size=1)

in the second example the 1e30 number is too large for Python: try calling len(range(int(1e30))), this will throw an OverflowError (don't get me started … 😭). therefore, we must use the iterable_len parameter to let MPIRE know how large the tasks list is. we also have to specify a chunk size here as the chunk size should be lower than sys.maxsize.

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

with WorkerPool(n_jobs=4) as pool:
    # 3. Multiply the numbers, results should be [0, 101, 204, 309, 416, ...]
    for result in pool.imap(multiply, zip(range(100), range(100, 200)), iterable_len=100):
        ...

the third example shows an example of using multiple function arguments. note that we use imap in this example, which allows us to process the results whenever they come available, not having to wait for all results to be ready. giving lazy girl energy fr 💅

with WorkerPool(n_jobs=4) as pool:
    # 4. Multiply the numbers, results should be [0, 101, ...]
    for result in pool.imap(multiply, [{'x': 0, 'y': 100}, {'y': 101, 'x': 1}, ...]):
        ...

the final example shows the use of an iterable of dictionaries. the (key, value) pairs are unpacked with the **-operator, as you would expect. so it doesn't matter in what order the keys are stored. this should work for collection.OrderedDict objects as well ✨

🔓 Circumvent argument unpacking 🔓

if you want to avoid unpacking and pass the tuples in example 3 or the dictionaries in example 4 as a whole, you can. we'll continue on example 4, but the workaround for example 3 is similar.

suppose we have the following function which expects a dictionary:

def multiply_dict(d):
    return d['x'] * d['y']

then you would have to convert the list of dictionaries to a list of single argument tuples, where each argument is a dictionary:

with WorkerPool(n_jobs=4) as pool:
    # Multiply the numbers, results should be [0, 101, ...]
    for result in pool.imap(multiply_dict, [({'x': 0, 'y': 100},),
                                            ({'y': 101, 'x': 1},),
                                            ...]):
        ...

there is a utility function available that does this transformation for you 🫶

from mpire.utils import make_single_arguments

with WorkerPool(n_jobs=4) as pool:
    # Multiply the numbers, results should be [0, 101, ...]
    for result in pool.imap(multiply_dict, make_single_arguments([{'x': 0, 'y': 100},
                                                                  {'y': 101, 'x': 1}, ...],
                                                                 generator=False)):
        ...

mpire.utils.make_single_arguments() takes an iterable of arguments and turns them into tuples, no cap, that's literally its job. 💅 The second argument of this function is you telling it whether you want a generator back or an actual materialized list — and if generator is the vibe you're going for, bestie you gotta pass along the iterable length too. ✨

🔀 Mixing map functions

Okay so this is the part where you CANNOT be chaotic. map functions cannot be used while another map function is still running. Like, one at a time, please. 🙅‍♀️ E.g., the following will raise an exception:

with WorkerPool(n_jobs=4) as pool:
    imap_results = pool.imap(multiply, zip(range(100), range(100, 200)), iterable_len=100)
    next(imap_results)  # We actually have to start the imap function

    # Will raise because the imap function is still running
    map_results = pool.map(square, range(100))

Make sure to first finish the imap function before starting a new map function. This holds for all map functions, every single one, periodt.

😭 Not exhausting a lazy imap function

If you don't exhaust a lazy imap function, but you do close the pool, the remaining tasks and results will be lost. Poof. Gone. 💀 E.g., the following will raise an exception:

with WorkerPool(n_jobs=4) as pool:
    imap_results = pool.imap(multiply, zip(range(100), range(100, 200)), iterable_len=100)
    first_result = next(imap_results)  # We actually have to start the imap function
    pool.terminate()

    # This will raise
    results = list(imap_results)

Similarly, exiting the with block terminates the pool as well:

with WorkerPool(n_jobs=4) as pool:
    imap_results = pool.imap(multiply, zip(range(100), range(100, 200)), iterable_len=100)
    first_result = next(imap_results)  # We actually have to start the imap function

# This will raise
results = list(imap_results)

🌈 Progress bar

Progress bar style
- Changing the default style
Progress bar options
- Progress bar position

Progress bar support is added through the tqdm package (installed by default when installing MPIRE) — yes, the link is https://pypi.python.org/pypi/tqdm if you wanna go stare at it. 🫶 The most easy way to include a progress bar is by enabling the progress_bar flag in any of the map functions:

with WorkerPool(n_jobs=4) as pool:
    pool.map(task, range(100), progress_bar=True)

This will display a basic tqdm progress bar displaying the time elapsed and remaining, number of tasks completed (including a percentage value) and the speed (i.e., number of tasks completed per time unit). It's giving productivity. 👑

✨ Progress bar style ✨

You can switch to a different progress bar style by changing the progress_bar_style parameter. For example, when you require a notebook widget use 'notebook' as the style:

with WorkerPool(n_jobs=4) as pool:
    pool.map(task, range(100), progress_bar=True, progress_bar_style='notebook')

The available styles are:

None: use the default style (= 'std', see below)
'std': use the standard tqdm progress bar
'rich': use the rich progress bar (requires the rich package to be installed, see Rich progress bars)
'notebook': use the Jupyter notebook widget
'dashboard': use only the progress bar on the dashboard

When in a terminal and using the 'notebook' style, the progress bar will behave weirdly. This is not recommended. 🙈

Note: If you run into problems with getting the progress bar to work in a Jupyter notebook (with 'notebook' style), have a look at Progress bar issues with Jupyter notebooks.

🎀 Changing the default style 🎀

You can change the default style by setting the mpire.tqdm_utils.PROGRESS_BAR_DEFAULT_STYLE variable:

import mpire.tqdm_utils

mpire.tqdm_utils.PROGRESS_BAR_DEFAULT_STYLE = 'notebook'

💖 Progress bar options 💖

The tqdm progress bar can be configured using the progress_bar_options parameter. This parameter accepts a dictionary with keyword arguments that will be passed to the tqdm constructor.

Some options in tqdm will be overwritten by MPIRE — not up for debate, they're taken. 🚫 These include the iterable, total and leave parameters. The iterable is set to the iterable passed on to the map function. The total parameter is set to the number of tasks to be completed. The leave parameter is always set to True. Some other parameters have a default value assigned to them, but can be overwritten by the user. Those ones are all yours bestie.

Here's an example where we change the description, the units, and the colour of the progress bar:

with WorkerPool(n_jobs=4) as pool:
    pool.map(some_func, some_data, progress_bar=True,
             progress_bar_options={'desc': 'Processing', 'unit': 'items', 'colour': 'green'})

For a complete list of available options, check out the tqdm docs at https://tqdm.github.io/docs/tqdm/#__init__. 📚

🌸 Progress bar position 🌸

You can easily print a progress bar on a different position on the terminal using the position parameter of tqdm, which facilitates the use of multiple progress bars. Here's an example of using multiple progress bars using nested WorkerPools:

def dispatcher(worker_id, X):
    with WorkerPool(n_jobs=4) as nested_pool:
        return nested_pool.map(task, X, progress_bar=True,
                               progress_bar_options={'position': worker_id + 1})

def main():
    with WorkerPool(n_jobs=4, daemon=False, pass_worker_id=True) as pool:
        pool.map(dispatcher, ((range(x, x + 100),) for x in range(100)), iterable_len=100,
                 n_splits=4, progress_bar=True)

main()

We use worker_id + 1 here because the worker IDs start at zero and we reserve position 0 for the progress bar of the main WorkerPool (which is the default). Math, but make it cute. 🧮✨

It goes without saying that you shouldn't specify the same progress bar position multiple times. Like bestie, they'll fight. 💀

Note: When using the rich progress bar style, the position parameter cannot be used. An exception will be raised when trying to do so. 🚨

Note: Most progress bar options are completely ignored when in a Jupyter/IPython notebook session or in the MPIRE dashboard.

👑 Worker init and exit 👑

When you want to initialize a worker you can make use of the worker_init parameter of any map function. This will call the initialization function only once per worker. Similarly, if you need to clean up the worker at the end of its lifecycle you can use the worker_exit parameter. Additionally, the exit function can return anything you like, which can be collected using mpire.WorkerPool.get_exit_results() after the workers are done. 💖

Both init and exit functions receive the worker ID, shared objects, and worker state in the same way as the task function does, given they're enabled.

For example:

def init_func(worker_state):
    # Initialize a counter for each worker
    worker_state['count_even'] = 0

def square_and_count_even(worker_state, x):
    # Count number of even numbers and return the square
    if x % 2 == 0:
        worker_state['count_even'] += 1
    return x * x

def exit_func(worker_state):
    # Return the counter
    return worker_state['count_even']

with WorkerPool(n_jobs=4, use_worker_state=True) as pool:
    pool.map(square_and_count_even, range(100), worker_init=init_func, worker_exit=exit_func)
    print(pool.get_exit_results())  # Output, e.g.: [13, 13, 12, 12]
    print(sum(pool.get_exit_results()))  # Output: 50

Important: When the worker_lifespan option is used to restart workers during execution, the exit function will be called for the worker that's shutting down and the init function will be called again for the new worker. Therefore, the number of elements in the list that's returned from mpire.WorkerPool.get_exit_results() does not always equal n_jobs. Don't be surprised, it's not a bug bestie. 😭

Important: When keep_alive is enabled the workers won't be terminated after a map call. This means the exit function won't be called until it's time for cleaning up the entire pool. You will have to explicitly call mpire.WorkerPool.stop_and_join() to receive the exit results. Periodt. 💅

🍓 Task chunking 🍓

By default, MPIRE chunks the given tasks in to 64 * n_jobs chunks. Each worker is given one chunk of tasks at a time before returning its results. This usually makes processing faster when you have rather small tasks (computation wise) and results are pickled/unpickled when they are send to a worker or main process. Chunking the tasks and results ensures that each process has to pickle/unpickle less often. It's iconic honestly. 🔥

However, to determine the number of tasks in the argument list the iterable should implement the __len__ method, which is available in default containers like list or tuple, but isn't available in most generator objects (the range object is one of the exceptions). To allow working with generators each map function has the option to pass the iterable length:

with WorkerPool(n_jobs=4) as pool:
    # 1. This will issue a warning and sets the chunk size to 1
    results = pool.map(square, ((x,) for x in range(1000)))

    # 2. This will issue a warning as well and sets the chunk size to 1
    results = pool.map(square, ((x,) for x in range(1000)), n_splits=4)

    # 3. Square the numbers using a generator using a specific number of splits
    results = pool.map(square, ((x,) for x in range(1000)), iterable_len=1000, n_splits=4)

    # 4. Square the numbers using a generator using automatic chunking
    results = pool.map(square, ((x,) for x in range(1000)), iterable_len=1000)

    # 5. Square the numbers using a generator using a fixed chunk size
    results = pool.map(square, ((x,) for x in range(1000)), chunk_size=4)

In the first two examples the function call will issue a warning because MPIRE doesn't know how large the chunks should be as the total number of tasks is unknown, therefore it will fall back to a chunk size of 1. The third example should work as expected where 4 chunks are used. The fourth example uses 256 chunks (the default 64 times the number of workers). The last example uses a fixed chunk size of four, so MPIRE doesn't need to know the iterable length. Slay. ✨

You can also call the chunk function manually:

from mpire.utils import chunk_tasks

# Convert to list because chunk_tasks returns a generator
print(list(chunk_tasks(range(10), n_splits=3)))
print(list(chunk_tasks(range(10), chunk_size=2.5)))
print(list(chunk_tasks((x for x in range(10)), iterable_len=10, n_splits=6)))

will output:

[(0, 1, 2, 3), (4, 5, 6), (7, 8, 9)]
[(0, 1, 2), (3, 4), (5, 6, 7), (8, 9)]
[(0, 1), (2, 3), (4,), (5, 6), (7, 8), (9,)]

💾 Maximum number of active tasks

ok so bestie, when your tasks are out here eating up ALL the memory, you've got a couple of moves 👇

Limit the number of jobs (that's the number of tasks currently available to the workers — the tasks just chillin' in the queue, ready to be processed).
Limit the number of active tasks.

The first option is the obvious one when the processes themselves are the memory hogs. The second one? That's your girl when the argument list is what's taking up too much room. Like, picture this: you wanna kick off an absolutely unhinged number of jobs (let's say a billion 🌈) where the args eat up 1 KB per task (big strings, yk), and suddenly that task queue is sitting there using ~1 TB of memory. 💀

For cases like that, a good rule of thumb is having twice the amount of active chunks of tasks as there are jobs. That way, when all the workers finish their task at the exact same time, each one can immediately pick up another task. And when workers take on those new tasks, the task generator gets iterated up to the point where there are, once again, twice as many active chunks of tasks. ✨

In MPIRE, the max number of active tasks by default is n_jobs * chunk_size * 2, so lowkey you don't have to tweak anything for memory optimization. But if you do wanna change that behavior for whatever reason, just set the max_active_tasks parameter:

with WorkerPool(n_jobs=4) as pool:
    results = pool.map(task, range(int(1e300)), iterable_len=int(1e300),
                       chunk_size=int(1e5), max_tasks_active=4 * int(1e5))

⚠️ Note: if you set max_tasks_active lower than n_jobs * chunk_size, some workers might end up with literally nothing to do. 😭

🧚 Worker lifespan

So sometimes workers that process a bunch of memory-intensive tasks don't release their used memory properly, and it just builds up and up and up. This is not a bug in MPIRE — it's just what happens as a consequence of Python's poor garbage collection. 🙈 To dodge this whole vibe, set the worker lifespan: the number of tasks after which a worker should restart.

with WorkerPool(n_jobs=4) as pool:
    results = pool.map(task, range(100), worker_lifespan=1, chunk_size=1)

In this example every single worker gets restarted after finishing one task. iconic behavior honestly.

⚠️ Note: once the worker lifespan is reached, a worker will still finish the current chunk of tasks before it restarts. So depending on your chunk_size, a worker could end up completing more tasks than the worker lifespan technically allows.

⏰ Timeouts

Timeouts can be set separately for the target, worker_init and worker_exit functions. Once a timeout is set and it gets reached, it throws a TimeoutError:

# Will raise TimeoutError, provided that the target function takes longer
# than half a second to complete
with WorkerPool(n_jobs=5) as pool:
    pool.map(time_consuming_function, range(10), task_timeout=0.5)

# Will raise TimeoutError, provided that the worker_init function takes longer
# than 3 seconds to complete or the worker_exit function takes longer than
# 150.5 seconds to complete
with WorkerPool(n_jobs=5) as pool:
    pool.map(time_consuming_function, range(10), worker_init=init, worker_exit=exit_,
             worker_init_timeout=3.0, worker_exit_timeout=150.5)

Use None (=default) to disable timeouts. simple as that 💅

🦋 imap and imap_unordered

When you're using one of the lazy map functions (like imap or imap_unordered), an exception will only be raised when the function is actually running. So like, when you execute:

with WorkerPool(n_jobs=5) as pool:
    results = pool.imap(time_consuming_function, range(10), task_timeout=0.5)

this will never raise. Not ever. Why? Because imap and imap_unordered return a generator object, and that generator stops executing until it gets the trigger to go beyond the yield statement. But when you iterate through the results, it raises exactly like you'd expect:

with WorkerPool(n_jobs=5) as pool:
    results = pool.imap(time_consuming_function, range(10), task_timeout=0.5)
    for result in results:
        ...

🧵 Threading

When using threading as start method, MPIRE won't be able to interrupt certain functions — like time.sleep, for example.

🌈 Numpy arrays

Contents

Chunking
Return value

✨ Chunking

Numpy arrays get treated a lil differently when they're passed on to the map functions. Usually MPIRE uses itertools.islice for chunking, which depends on the __iter__ special function of the container object. But applying that to numpy arrays:

import numpy as np

# Create random array
arr = np.random.rand(10, 3)

# Chunk the array using default chunking
arr_iter = iter(arr)
chunk_size = 3
while True:
    chunk = list(itertools.islice(arr_iter, chunk_size))
    if chunk:
        yield chunk
    else:
        break

would yield:

[array([0.68438994, 0.9701514 , 0.40083965]), array([0.88428556, 0.2083905 , 0.61490443]),
 array([0.89249174, 0.39902235, 0.70762541])]
[array([0.18850964, 0.1022777 , 0.41539432]), array([0.07327858, 0.18608165, 0.75862301]),
 array([0.69215651, 0.4211941 , 0.31029439])]
[array([0.82571272, 0.72257819, 0.86079131]), array([0.91285817, 0.49398461, 0.27863929]),
 array([0.146981  , 0.84671211, 0.30122806])]
[array([0.11783283, 0.12585031, 0.39864368])]

In other words, every row of the array ends up in its own array and each one of them gets handed to the target function individually. which is... not it. 🙈 Instead, MPIRE chunks them into something way more reasonable using numpy slicing:

from mpire.utils import chunk_tasks

for chunk in chunk_tasks(arr, chunk_size=chunk_size):
    print(repr(chunk))

Output:

array([[0.68438994, 0.9701514 , 0.40083965],
       [0.88428556, 0.2083905 , 0.61490443],
       [0.89249174, 0.39902235, 0.70762541]])
array([[0.18850964, 0.1022777 , 0.41539432],
       [0.07327858, 0.18608165, 0.75862301],
       [0.69215651, 0.4211941 , 0.31029439]])
array([[0.82571272, 0.72257819, 0.86079131],
       [0.91285817, 0.49398461, 0.27863929],
       [0.146981  , 0.84671211, 0.30122806]])
array([[0.11783283, 0.12585031, 0.39864368]])

Every chunk is now a single numpy array holding as many rows as the chunk size — except for the last chunk, because there just aren't enough rows left over. 🎀

💜 Return value

When your user provided function returns numpy arrays and you're applying the mpire.WorkerPool.map() function, MPIRE will concatenate the resulting numpy arrays into a single array by default. Like so:

def add_five(x):
    return x + 5

with WorkerPool(n_jobs=4) as pool:
    results = pool.map(add_five, arr, chunk_size=chunk_size)

will return:

array([[5.68438994, 5.9701514 , 5.40083965],
       [5.88428556, 5.2083905 , 5.61490443],
       [5.89249174, 5.39902235, 5.70762541],
       [5.18850964, 5.1022777 , 5.41539432],
       [5.07327858, 5.18608165, 5.75862301],
       [5.69215651, 5.4211941 , 5.31029439],
       [5.82571272, 5.72257819, 5.86079131],
       [5.91285817, 5.49398461, 5.27863929],
       [5.146981  , 5.84671211, 5.30122806],
       [5.11783283, 5.12585031, 5.39864368]])

You can cancel this behavior with the concatenate_numpy_output flag:

with WorkerPool(n_jobs=4) as pool:
    results = pool.map(add_five, arr, chunk_size=chunk_size, concatenate_numpy_output=False)

and this will return individual arrays:

[array([[5.68438994, 5.9701514 , 5.40083965],
        [5.88428556, 5.2083905 , 5.61490443],
        [5.89249174, 5.39902235, 5.70762541]]),
 array([[5.18850964, 5.1022777 , 5.41539432],
        [5.07327858, 5.18608165, 5.75862301],
        [5.69215651, 5.4211941 , 5.31029439]]),
 array([[5.82571272, 5.72257819, 5.86079131],
        [5.91285817, 5.49398461, 5.27863929],
        [5.146981  , 5.84671211, 5.30122806]]),
 array([[5.11783283, 5.12585031, 5.39864368]])]

👑 Apply family

Contents

apply
apply_async
AsyncResult
Callbacks
Worker init and exit
Timeouts

mpire.WorkerPool implements two apply functions, and they're super similar to the ones in the multiprocessing module:

mpire.WorkerPool.apply() — Apply a function to a single task. This is a blocking call.
mpire.WorkerPool.apply_async() — A variant of the above, but non-blocking. This returns an mpire.async_result.AsyncResult object.

🌸 apply

The apply function is a blocking call, which means it will not return until the task is completed. If you wanna run multiple different tasks in parallel, you should use the apply_async function instead. And if you need to run the same function for many tasks in parallel, use the map functions instead.

The apply function takes a function, positional arguments, and keyword arguments — same energy as how multiprocessing does it.

def task(a, b, c, d):
    return a + b + c + d

with WorkerPool(n_jobs=1) as pool:
    result = pool.apply(task, args=(1, 2), kwargs={'d': 4, 'c': 3})
    print(result)

🫶 apply_async

The apply_async function is a non-blocking call, which means it returns immediately. It gives you back an mpire.async_result.AsyncResult object, which you can use to get the result of the task at a later moment in time.

The apply_async function takes the same parameters as the apply function.

def task(a, b):
    return a + b

with WorkerPool(n_jobs=4) as pool:
    async_results = [pool.apply_async(task, args=(i, i)) for i in range(10)]
    results = [async_result.get() for async_result in async_results]

Obtaining the results should happen while the pool is still running! fr fr. Like, the following will deadlock:

with WorkerPool(n_jobs=4) as pool:
    async_results = [pool.apply_async(task, args=(i, i)) for i in range(10)]

# Will wait forever
results = [async_result.get() for async_result in async_results]

You can, however, make use of the mpire.WorkerPool.stop_and_join() function to stop the workers and join the pool. This makes sure all tasks are completed before the pool exits. ✅

with WorkerPool(n_jobs=4) as pool:
    async_results = [pool.apply_async(task, args=(i, i)) for i in range(10)]
    pool.stop_and_join()

# Will not deadlock
results = [async_result.get() for async_result in async_results]

AsyncResult

okay so the mpire.async_result.AsyncResult object? it has these little convenience methods that are lowkey iconic and you're gonna love them 💖

with WorkerPool(n_jobs=1) as pool:
    async_result = pool.apply_async(task, args=(1, 1))

    # Check if the task is completed
    is_completed = async_result.ready()

    # Wait until the task is completed, or until the timeout is reached.
    async_result.wait(timeout=10)

    # Get the result of the task. This will block until the task is completed,
    # or until the timeout is reached.
    result = async_result.get(timeout=None)

    # Check if the task was successful (i.e., did not raise an exception).
    # This will raise an exception if the task is not completed yet.
    is_successful = async_result.successful()

Callbacks

Every single apply function has a callback and an error_callback argument, and no bestie they are not optional aesthetic choices, they're actual functions that get called when the task finishes. 🎀

the callback gets called with the result of the task when it completed successfully
the error_callback gets called with the exception when the task failed

def task(a):
    return a + 1

def callback(result):
    print("Task completed successfully with result:", result)

def error_callback(exception):
    print("Task failed with exception:", exception)

with WorkerPool(n_jobs=1) as pool:
    pool.apply(task, 42, callback=callback, error_callback=error_callback)

Worker init and exit

just like the map family of functions, the apply family also serves worker_init and worker_exit arguments. 🫶 these are functions that get called when a worker is started and stopped, respectively. head over to Worker init and exit for the full lore on these.

def worker_init():
    print("Worker started")

def worker_exit():
    print("Worker stopped")

with WorkerPool(n_jobs=5) as pool:
    pool.apply(task, 42, worker_init=worker_init, worker_exit=worker_exit)

okay but there's a caveat, and it's kind of a big one 💀 When the first apply or apply_async function is executed, the entire pool of workers is started. so in that example above, all five workers get started even though only one was actually needed. which also means the worker_init function is set for all of those workers at once. so you cannot have a different worker_init function for each apply task — a second, different worker_init function will simply be ignored. 🙈

same energy for worker_exit: it can only be set once too. and on top of that, exit functions are only called when a worker exits, which in this case translates to when the pool exits. so if you call apply or apply_async multiple times, the worker_exit function is only called once at the end. use mpire.WorkerPool.stop_and_join() if you want to stop the workers, which will trigger the worker_exit function for each worker.

Timeouts

the apply family of functions also has task_timeout, worker_init_timeout and worker_exit_timeout arguments 💅 and surprise surprise, they're timeouts for the task, the worker_init function and the worker_exit function, respectively. they behave similarly to the ones for the map functions.

when a single task times out, only that task is cancelled. the other tasks keep running, no drama. ✨
when a worker init or exit times out, the entire pool is stopped. 💀

see Timeouts for more.

Dashboard

okay so the dashboard is genuinely the main character here 👑 it lets you see progress information straight from a browser. this is so convenient when you're running scripts in a notebook or screen, when you want to share the progress info with other people, or when you just want real-time worker insight info.

heads up bestie: the dashboard dependencies are not installed by default. check Dashboard for the deets.

Starting the dashboard
Connecting to an existing dashboard
Using the dashboard
Stack level

Starting the dashboard

you can start the dashboard programmatically, like:

from mpire.dashboard import start_dashboard

# Will return a dictionary with dashboard details
dashboard_details = start_dashboard()
print(dashboard_details)

which will print:

{'dashboard_port_nr': 8080,
 'manager_host': 'localhost',
 'manager_port_nr': 8081}

this starts a dashboard on your local machine on port 8080. 🎀 if that port is already in use, MPIRE will try the next one, and the next, until it finds an unused one. in the rare case that no ports are available all the way up to port 8099, the function will raise an OSError. by default, MPIRE tries ports 8080-8100. you can override this range by passing on a custom range object:

dashboard_details = start_dashboard(range(9000, 9100))

the returned dictionary contains the port number that was ultimately chosen, plus info on how to connect to this dashboard remotely. 🦋

another way of starting a dashboard is by using the bash script:

this doesn't work on Windows! 🚫

$ mpire-dashboard

this starts a dashboard with the connection details printed on screen, and it'll say something like:

Starting MPIRE dashboard...

MPIRE dashboard started on http://localhost:8080
Server is listening on localhost:8098
--------------------------------------------------

the server part corresponds to the manager_host and manager_port_nr from the dictionary returned by mpire.dashboard.start_dashboard(). and just like before, you can provide a custom port range:

$ mpire-dashboard --port-range 9000-9100

the benefit of starting a dashboard this way? your dashboard keeps running even if your script errors out. 🔥 so you'll be able to see what the error was, when it occurred, and where it occurred in your code. iconic behavior honestly.

Connecting to an existing dashboard

if you've started a dashboard elsewhere, you can connect to it using:

from mpire.dashboard import connect_to_dashboard

connect_to_dashboard(manager_port_nr=8081, manager_host='localhost')

make sure you use the manager_port_nr, not the dashboard_port_nr from the examples above. this is the classic mix-up and it will not slay. 💀

you can connect to an existing dashboard on the same machine, but also on a remote machine (as long as the ports are open). if manager_host is omitted it will fall back to using 'localhost'.

Using the dashboard

once you're connected to a dashboard you don't need to change anything in your code. 🫶 when you've enabled the use of a progress bar in your map call, the progress bar will automatically register itself to the dashboard server and show up. like here:

from mpire import WorkerPool
from mpire.dashboard import connect_to_dashboard

connect_to_dashboard(8099)

def square(x):
    import time
    time.sleep(0.01)  # To be able to show progress
    return x * x

with WorkerPool(4) as pool:
    pool.map(square, range(10000), progress_bar=True)

this will show something like: [image: ] [image]

you can click on a progress bar row to view details about the function that is called (which has already been done in the screenshot above). 🎀

it'll also let you know when a KeyboardInterrupt signal was sent to the running process: [image: ] [image]

or show the traceback information in case of an exception: [image: ] [image]

and in case you've enabled Worker insights, those insights get shown real-time in the dashboard: [image: ] [image]

click on the Insights (click to expand/collapse) to either expand or collapse the insight details.

the dashboard refreshes itself automatically every 0.5 seconds. ✨

Stack level

by default, the dashboard shows information about the function that is called and where it's called from. but sometimes, when you've wrapped the function in another function, you might be less interested in the wrapper function and more interested in the function that's calling that wrapper. 🙈 in those cases you can use mpire.dashboard.set_stacklevel() to set the stack level. this is the number of levels in the stack to go back in order to find the frame that contains the function that is invoking MPIRE. for example:

from mpire import WorkerPool
from mpire.dashboard import set_stacklevel, start_dashboard

class WorkerPoolWrapper:
    def __init__(self, n_jobs, progress_bar=True):
        self.n_jobs = n_jobs
        self.progress_bar = progress_bar

    def __call__(self, func, data):
        with WorkerPool(self.n_jobs) as pool:
            return pool.map(func, data, progress_bar=self.progress_bar)

def square(x):
    return x * x

if __name__ == '__main__':
    start_dashboard()
    executor = WorkerPoolWrapper(4, progress_bar=True)
    set_stacklevel(1)  # default
    results = executor(square, range(10000))
    set_stacklevel(2)
    results = executor(square, range(10000))

when you run this code you'll see that the dashboard shows two progress bars. in both cases, the dashboard shows the square function as the function that is called. but! in the first case, it shows return pool.map(func, data, progress_bar=self.progress_bar) as the line where it's called from. in the second case, it shows the results = executor(square, range(10000)) line. 💅

Troubleshooting

this section describes some known problems that can arise when using MPIRE. consider it your little emergency kit 🧚

Progress bar issues with Jupyter notebooks
- IProgress not found
- Widget Javascript not detected
Unit tests
Shutting down takes a long time on error
Unpicklable tasks/results
AttributeError: Can't get attribute '<some_function>' on <module '__main__' (built-in)>
Windows
macOS

Progress bar issues with Jupyter notebooks

when you're using the progress bar in a Jupyter notebook you might run into some issues. a few of these are described below, together with possible solutions. 💖

IProgress not found

when you see something like ImportError: IProgress not found. Please update jupyter and ipywidgets., this means ipywidgets is not installed. you can install it using pip:

pip install ipywidgets

or conda:

conda install -c conda-forge ipywidgets

have a look at the ipywidgets documentation for more information. 🌸

when you see something like Widget Javascript not detected. It may not be enabled properly., this means the Javascript extension is not enabled. you can enable it using the following command before starting your notebook:

jupyter nbextension enable --py --sys-prefix widgetsnbextension

🚨 btw bestie — you have to restart your notebook server after enabling the extension. Just restarting the kernel is not gonna cut it, like, at all. 💅

🧪 Unit tests ✨

okay so when you're using the 'spawn' or 'forkserver' method you'll probably run into one or two issues when running unittests in your own package, and it's giving chaos. 🫠

Problem the first: your unittests will restart every single time the piece of code containing such a start method is called, which means your terminal output goes completely funky. To remedy this problem, make sure your setup call in setup.py is surrounded by an if __name__ == '__main__': clause, like so:

from setuptools import setup

if __name__ == '__main__':

    # Call setup and install any dependencies you have inside the if-clause
    setup(...)

Go read the 'Safe importing of main module' section in the caveats over at https://docs.python.org/3/library/multiprocessing.html#the-spawn-and-forkserver-start-methods. 🫶

Problem the second: the semaphore tracker of multiprocessing will complain at you when you run individual (or a selection of) unittests using python setup.py test -s tests.some_test. At the end of the tests you'll see errors like:

Traceback (most recent call last):
  File ".../site-packages/multiprocess/semaphore_tracker.py", line 132, in main
    cache.remove(name)
KeyError: b'/mp-d3i13qd5'
.../site-packages/multiprocess/semaphore_tracker.py:146: UserWarning: semaphore_tracker: There appear to be 58
                                                                                         leaked semaphores to clean up at shutdown
  len(cache))
.../site-packages/multiprocess/semaphore_tracker.py:158: UserWarning: semaphore_tracker: '/mp-f45dt4d6': [Errno 2]
                                                                                         No such file or directory
  warnings.warn('semaphore_tracker: %r: %s' % (name, e))
...

Your unittests will still succeed and run OK, promise. Unfortunately, no remedy to this problem has been found yet when using python setup.py test. 🙈 What you can use instead is something like the following:

python -m unittest tests.some_test

That one will work just fine, slay. Peep the unittest documentation at https://docs.python.org/3.4/library/unittest.html#command-line-interface for more info. 🦋

🐌 Shutting down takes a long time on error 💀

When you issue a KeyboardInterrupt or when an error occurd in the function that's run in parallel, there are situations where MPIRE needs a few seconds to gracefully shutdown. It's not being dramatic, it's just that in these situations the task or results queue can be quite full still. MPIRE drains these queues until they're completely empty, all so she can properly shutdown and clean up every communication channel. Iconic behavior honestly. 👑

To remedy this issue you can use the max_tasks_active parameter and set it to n_jobs * 2, or similar. Aside from the added benefit that the workers can start more quickly, the queues won't get that full anymore and shutting down will be much quicker. See Maximum number of active tasks (at #max-active-tasks) for more information.

When you're using a lazy map function also be sure to iterate through the results — otherwise that queue will be full and draining it will take a longer time. Don't leave her on read bestie. 📵

🥒 Unpicklable tasks/results 🙈

Sometimes you can encounter deadlocks in your code when using MPIRE. When you encounter this, chances are some tasks or results from your script can't be pickled. MPIRE makes use of multiprocessing queues for inter-process communication, and if your function returns unpicklable results the queue will, unfortunately, deadlock. 😭

The only way to remedy this problem in MPIRE would be to manually pickle objects before sending it to a queue and quit gracefully when encountering a pickle error. However, this would mean objects would always be pickled twice. That would add a heavy performance penalty and is therefore not an acceptable solution. periodt.

Instead, the user should make sure their tasks and results are always picklable (which in most cases won't be a problem), or resort to setting use_dill=True. The latter is capable of pickling a lot more exotic types. See Dill (at #use-dill) for more information. 🧚

💀 AttributeError: Can’t get attribute ‘<some_function>’ on <module ‘main’ (built-in)> 😭

This error can occur when inside an iPython or Jupyter notebook session and the function to parallelize is defined in that session. This is often the result of using spawn as start method (the default on Windows), which starts a new process without copying the function in question. Rude of it, honestly.

This error is actually related to the Unpicklable tasks/results problem and can be solved in a similar way. I.e., you can define your function in a file that can be imported by the child process, or you can resort to using dill by setting use_dill=True. See Dill (at #use-dill) for more information. 🌈

🪟 Windows

When using dill and an exception occurs, or when the exception occurs in an exit function, it can print additional OSError messages in the terminal, but they can be safely ignored. 🙈
The mpire-dashboard script does not work on Windows. 💔

🍎 macOS

When encountering OSError: [Errno 24] Too many open files errors, use ulimit -n <number> to increase the limit of the number of open files. This is required because MPIRE uses file-descriptor based synchronization primitives and macOS has a very low default limit. For example, MPIRE uses about 190 file descriptors when using 10 workers. 🫶
Pinning of processes to CPU cores is not supported on macOS. This is because macOS does not support the sched_setaffinity system call. A warning will be printed when trying to use this feature. 😔

📚 API Reference 👑

🗂️ Contents

WorkerPool
AsyncResult
Task chunking
Converting iterable of arguments
Dashboard
Other

🏊 WorkerPool

class mpire.WorkerPool(n_jobs=None, daemon=True, cpu_ids=None, shared_objects=None, pass_worker_id=False, use_worker_state=False, start_method='fork', keep_alive=False, use_dill=False, enable_insights=False, order_tasks=False)

A multiprocessing worker pool which acts like a multiprocessing.Pool, but is faster and has more options. she's not like other pools. 💅

`enter()`

Enable the use of the with statement.

Return type: WorkerPool

`exit(*_)`

Enable the use of the with statement. Gracefully terminates workers, if there are any.

Return type: None

`init(n_jobs=None, daemon=True, cpu_ids=None, shared_objects=None, pass_worker_id=False, use_worker_state=False, start_method='fork', keep_alive=False, use_dill=False, enable_insights=False, order_tasks=False)`

Parameters

n_jobs (int | None) — Number of workers to spawn. If None, will use mpire.cpu_count().
daemon (bool) — Whether to start the child processes as daemon.
cpu_ids (List[int | List[int]]) — List of CPU IDs to use for pinning child processes to specific CPUs. The list must be as long as the number of jobs used (if n_jobs equals None it must be equal to mpire.cpu_count()), or the list must have exactly one element. In the former case, element i specifies the CPU ID(s) to use for child process i. In the latter case the single element specifies the CPU ID(s) for all child processes to use. A single element can be either a single integer specifying a single CPU ID, or a list of integers specifying that a single child process can make use of multiple CPU IDs. If None, CPU pinning will be disabled. ✨
shared_objects (Any) — Objects to be passed on as shared objects to the workers once. It will be passed on to the target, worker_init, and worker_exit functions. shared_objects is only passed on when it's not None. Shared objects will be copy-on-write when using fork as start method. When enabled, functions receive the shared objects as second argument, depending on other settings. The order is: worker_id, shared_objects, worker_state, and finally the arguments passed on from iterable_of_args.
pass_worker_id (bool) — Whether to pass on a worker ID to the target, worker_init, and worker_exit functions. When enabled, functions receive the worker ID as first argument, depending on other settings. The order is: worker_id, shared_objects, worker_state, and finally the arguments passed on from iterable_of_args.
use_worker_state (bool) — Whether to let a worker have a worker state. The worker state will be passed on to the target, worker_init, and worker_exit functions. When enabled, functions receive the worker state as third argument, depending on other settings. The order is: worker_id, shared_objects, worker_state, and finally the arguments passed on from iterable_of_args.
start_method (str) — Which process start method to use. Options for multiprocessing: 'fork' (default, if available), 'forkserver' and 'spawn' (default, if 'fork' isn't available). For multithreading use 'threading'. See https://docs.python.org/3/library/multiprocessing.html#contexts-and-start-methods for more information and https://docs.python.org/3/library/multiprocessing.html#the-spawn-and-forkserver-start-methods for some caveats when using the 'spawn' or 'forkserver' methods.
keep_alive (bool) — When True it will keep workers alive after completing a map call, allowing to reuse workers
use_dill (bool) — should we use dill as our serialization backend? bestie listen, some exotic types (think lambdas, nested functions) just do not vibe with spawn as the start method 😭 so if that's your situation, turn on dill. heads up tho: it can be a bit slower sometimes 🐌
enable_insights (bool) — wanna turn on worker insights? iconic. it might come with a small performance penalty (often neglible, so like, don't stress) ✨
order_tasks (bool) — should tasks be handed to the workers in order? like, worker 0 gets chunk 0, worker 1 gets chunk 1, etc. etc. 🎀

`weakref`

list of weak references to the object. that's it. that's the whole thing. 💅

✨ METHODS ✨

`apply(func, args=(), kwargs=None, callback=None, error_callback=None, worker_init=None, worker_exit=None, task_timeout=None, worker_init_timeout=None, worker_exit_timeout=None)`

Apply a function to a single task. important: this is a blocking call, so it will sit there and hold your hand until it's done 🫶

Parameters

func (Callable) — the function to apply to the task. when you're passing on the worker ID, the function should receive the worker ID as its first argument. if shared objects are provided, the function should receive those as the next argument. and if the worker state has been enabled, it should receive a state variable as the next argument. (yes, the order matters, keep up bestie)
args (Any) — arguments to pass to a worker, which passes it to the function func as func(*args)
kwargs (Dict) — keyword arguments to pass to a worker, which passes it to the function func as func(**kwargs)
callback (Callable | None) — callback function to call when the task is finished. it receives the output of the function func as its argument 🌸
error_callback (Callable | None) — callback function to call when the task has failed 💀 it receives the exception as its argument
worker_init (Callable | None) — function to call each time a new worker starts. when passing on the worker ID, the function should receive the worker ID as its first argument. if shared objects are provided, the function should receive those as the next argument. if the worker state has been enabled, it should receive a state variable as the next argument
worker_exit (Callable | None) — function to call each time a worker exits. return values will be fetched and made available through mpire.WorkerPool.get_exit_results. when passing on the worker ID, the function should receive the worker ID as its first argument. if shared objects are provided, the function should receive those as the next argument. if the worker state has been enabled, it should receive a state variable as the next argument
task_timeout (float | None) — timeout in seconds for a single task. when the timeout is exceeded, MPIRE will raise a TimeoutError 😱 use None to disable (default). note: the timeout doesn't apply to worker_init and worker_exit functions — use worker_init_timeout and worker_exit_timeout for that, respectively
worker_init_timeout (float | None) — timeout in seconds for the worker_init function. when the timeout is exceeded, MPIRE will raise a TimeoutError. use None to disable (default).
worker_exit_timeout (float | None) — timeout in seconds for the worker_exit function. when the timeout is exceeded, MPIRE will raise a TimeoutError. use None to disable (default).

Return type

Any

Returns

result of the function func applied to the task 🎉

`apply_async(func, args=(), kwargs=None, callback=None, error_callback=None, worker_init=None, worker_exit=None, task_timeout=None, worker_init_timeout=None, worker_exit_timeout=None)`

Apply a function to a single task. but this one is a non-blocking call — it does NOT wait around, it's giving main character energy and moving on immediately 👑

Parameters

func (Callable) — the function to apply to the task. when passing on the worker ID the function should receive the worker ID as its first argument. if shared objects are provided the function should receive those as the next argument. if the worker state has been enabled it should receive a state variable as the next argument
args (Any) — arguments to pass to a worker, which passes it to the function func as func(*args)
kwargs (Dict) — keyword arguments to pass to a worker, which passes it to the function func as func(**kwargs)
callback (Callable | None) — callback function to call when the task is finished. it receives the output of the function func as its argument 💖
error_callback (Callable | None) — callback function to call when the task has failed. it receives the exception as its argument 💀
worker_init (Callable | None) — function to call each time a new worker starts. when passing on the worker ID the function should receive the worker ID as its first argument. if shared objects are provided the function should receive those as the next argument. if the worker state has been enabled it should receive a state variable as the next argument
worker_exit (Callable | None) — function to call each time a worker exits. return values will be fetched and made available through mpire.WorkerPool.get_exit_results. when passing on the worker ID the function should receive the worker ID as its first argument. if shared objects are provided the function should receive those as the next argument. if the worker state has been enabled it should receive a state variable as the next argument
task_timeout (float | None) — timeout in seconds for a single task. when the timeout is exceeded, MPIRE will raise a TimeoutError 😱 use None to disable (default). note: the timeout doesn't apply to worker_init and worker_exit functions, use worker_init_timeout and worker_exit_timeout for that, respectively
worker_init_timeout (float | None) — timeout in seconds for the worker_init function. when the timeout is exceeded, MPIRE will raise a TimeoutError. use None to disable (default).
worker_exit_timeout (float | None) — timeout in seconds for the worker_exit function. when the timeout is exceeded, MPIRE will raise a TimeoutError. use None to disable (default).

Return type

AsyncResult

Returns

result of the function func applied to the task 🦋

`get_exit_results()`

Obtain a list of exit results when an exit function is defined.

Return type

List

Returns

exit results list 📋

`get_insights()`

Creates insights from the raw insight data 🔍

Return type

Dict

Returns

dictionary containing worker insights 👑

`imap(func, iterable_of_args, iterable_len=None, max_tasks_active=None, chunk_size=None, n_splits=None, worker_lifespan=None, progress_bar=False, worker_init=None, worker_exit=None, task_timeout=None, worker_init_timeout=None, worker_exit_timeout=None, progress_bar_options=None, progress_bar_style=None)`

Same as multiprocessing.imap_unordered(), but ordered ✨ (yes bestie, order is the whole point). also lets you set the maximum number of tasks available in the queue, which is lowkey handy.

Parameters

func (Callable) — function to call each time new task arguments become available. when passing on the worker ID the function should receive the worker ID as its first argument. if shared objects are provided the function should receive those as the next argument. if the worker state has been enabled it should receive a state variable as the next argument
iterable_of_args (Sized | Iterable) — a numpy array or an iterable containing tuples of arguments to pass to a worker, which passes it to the function func
iterable_len (int | None) — how many elements are chillin' in the iterable_of_args, bestie. When chunk_size is set to None, MPIRE needs to know the number of tasks. You can either give it up by implementing the __len__ function on the iterable object, or just straight-up tell it the number of tasks yourself. Either way, it needs the tea. ✨
max_tasks_active (int | None) — the max number of active tasks kickin' it in the queue. If it's None it gets converted to n_jobs * chunk_size * 2. Yeah, math. 🧮
chunk_size (int | None) — how many simultaneous tasks get handed to a worker at once. When it's None it'll use n_splits instead. 🎀
n_splits (int | None) — the number of splits to use when chunk_size is None. And when both chunk_size and n_splits are None, it goes with n_splits = n_jobs * 64. No cap. 🔥
worker_lifespan (int | None) — how many tasks a worker can handle before it gets restarted (a lil glow-up, if you will 😌). If it's None, workers will stay alive the entire time. Use this when your workers be hoarding too much memory over the course of time — it's giving "please take a break, bestie". 💀
progress_bar (bool) — when True it shows a progress bar. Iconic. ✨
worker_init (Callable | None) — the function to call every single time a new worker starts. When you pass on the worker ID, the function should receive the worker ID as its first argument. If shared objects are provided, the function should receive those as the next argument. And if worker state has been enabled, it should receive a state variable as the next argument. So like, be mindful of the order, okay? 🙈
worker_exit (Callable | None) — the function to call every time a worker exits. Return values get fetched and made available through mpire.WorkerPool.get_exit_results. When you pass on the worker ID, the function should receive the worker ID as its first argument. If shared objects are provided, the function should receive those as the next argument. If worker state has been enabled, it should receive a state variable as the next argument. Same energy. 🫶
task_timeout (float | None) — timeout in seconds for a single task. When the timeout gets exceeded, MPIRE will raise a TimeoutError. Use None to disable it (that's the default). Note: the timeout doesn't apply to worker_init and worker_exit functions — use worker_init_timeout and worker_exit_timeout for that, respectively. Periodt. ⏰
worker_init_timeout (float | None) — timeout in seconds for the worker_init function. When the timeout gets exceeded, MPIRE will raise a TimeoutError. Use None to disable it (default).
worker_exit_timeout (float | None) — timeout in seconds for the worker_exit function. When the timeout gets exceeded, MPIRE will raise a TimeoutError. Use None to disable it (default).
progress_bar_options (Dict[str, Any] | None) — a dictionary holding keyword arguments to pass to the tqdm progress bar. See tqdm.tqdm() for the details. The arguments total and leave will be overwritten by MPIRE, so don't even try it bestie. 🙅‍♀️
progress_bar_style (str | None) — the progress bar style to use. Can be one of None, 'std', or 'notebook'. Pick your fighter. 💅

Return type

Generator[Any, None, None]

Returns

Generator yielding ordered results ✨

`imap_unordered(...)`

imap_unordered(func, iterable_of_args, iterable_len=None,
max_tasks_active=None, chunk_size=None, n_splits=None,
worker_lifespan=None, progress_bar=False, worker_init=None,
worker_exit=None, task_timeout=None, worker_init_timeout=None,
worker_exit_timeout=None, progress_bar_options=None,
progress_bar_style=None)

Same as multiprocessing.imap_unordered(). It also lets you set the maximum number of tasks available in the queue. Bestie-friendly. 🦋

Parameters

func (Callable) — the function to call each time new task arguments become available. When you pass on the worker ID, the function should receive the worker ID as its first argument. If shared objects are provided, the function should receive those as the next argument. If worker state has been enabled, it should receive a state variable as the next argument. 💜
iterable_of_args (Sized | Iterable) — a numpy array, or an iterable containing tuples of arguments to pass to a worker, which then passes them to the function func. 🍓
iterable_len (int | None) — number of elements in the iterable_of_args. When chunk_size is set to None it needs to know the number of tasks. This can either be provided by implementing the __len__ function on the iterable object, or by specifying the number of tasks. ✨
max_tasks_active (int | None) — maximum number of active tasks in the queue. If it's None it will be converted to n_jobs * chunk_size * 2. 🧮
chunk_size (int | None) — number of simultaneous tasks to give to a worker. When None it will use n_splits. 🎀
n_splits (int | None) — number of splits to use when chunk_size is None. When both chunk_size and n_splits are None, it will use n_splits = n_jobs * 64. 🔥
worker_lifespan (int | None) — number of tasks a worker can handle before it is restarted. If None, workers will stay alive the entire time. Use this when workers use up too much memory over the course of time. 💀
progress_bar (bool) — when True it will display a progress bar. ✨
worker_init (Callable | None) — function to call each time a new worker starts. When passing on the worker ID the function should receive the worker ID as its first argument. If shared objects are provided the function should receive those as the next argument. If the worker state has been enabled it should receive a state variable as the next argument. 🙈
worker_exit (Callable | None) — function to call each time a worker exits. Return values will be fetched and made available through mpire.WorkerPool.get_exit_results. When passing on the worker ID the function should receive the worker ID as its first argument. If shared objects are provided the function should receive those as the next argument. If the worker state has been enabled it should receive a state variable as the next argument. 🫶
task_timeout (float | None) — timeout in seconds for a single task. When the timeout is exceeded, MPIRE will raise a TimeoutError. Use None to disable (default). Note: the timeout doesn't apply to worker_init and worker_exit functions, use worker_init_timeout and worker_exit_timeout for that, respectively. ⏰
worker_init_timeout (float | None) — timeout in seconds for the worker_init function. When the timeout is exceeded, MPIRE will raise a TimeoutError. Use None to disable (default).
worker_exit_timeout (float | None) — timeout in seconds for the worker_exit function. When the timeout is exceeded, MPIRE will raise a TimeoutError. Use None to disable (default).
progress_bar_options (Dict[str, Any] | None) — dictionary containing keyword arguments to pass to the tqdm progress bar. See tqdm.tqdm() for details. The arguments total and leave will be overwritten by MPIRE. 🙅‍♀️
progress_bar_style (str | None) — the progress bar style to use. Can be one of None, 'std', or 'notebook'. 💅

Return type

Generator[Any, None, None]

Returns

Generator yielding unordered results 🌈

`join(keep_alive=False)`

When keep_alive=False: it inserts a poison pill 💀, grabs the exit results, waits until the tasks/results queues are done, and waits until all workers are finished. When keep_alive=True: it inserts a non-lethal poison pill 🧚, and waits until the tasks/results queues are done.

Not me reminding you that join and stop_and_join are aliases. 💖

Parameters

keep_alive (bool) — whether to keep the workers alive. 👑

Return type

None

map(func, iterable_of_args, iterable_len=None,
    max_tasks_active=None, chunk_size=None, n_splits=None,
    worker_lifespan=None, progress_bar=False,
    concatenate_numpy_output=True, worker_init=None,
    worker_exit=None, task_timeout=None, worker_init_timeout=None,
    worker_exit_timeout=None, progress_bar_options=None,
    progress_bar_style=None)

Okay so this one is literally just multiprocessing.map() but make it ✨fancy✨. It also lets you cap how many tasks can be sitting in the queue at once. Heads up though: this version can be slower than the unordered one, so if you don't actually need your results in order, bestie… 👀

Parameters 🎀

func (Callable) — the function that gets called every time new task arguments show up. If you're passing on the worker ID, it should take the worker ID as its first argument. If shared objects are provided, it should take those as the next argument. And if worker state is enabled, it should take a state variable as the next argument after that. 💅
iterable_of_args (Sized | Iterable) — a numpy array, or an iterable full of tuples of arguments that get handed to a worker, which then hands them to func. 🫶
iterable_len (int | None) — how many elements are in iterable_of_args. When chunk_size is None, it needs to know the number of tasks. You can give it that either by implementing the __len__ function on your iterable object, or by just telling it the number of tasks. fr fr.
max_tasks_active (int | None) — max number of active tasks chilling in the queue. If it's None, it gets converted to n_jobs * chunk_size * 2. 💖
chunk_size (int | None) — how many tasks get handed to a worker at the same time. When it's None, it uses n_splits instead. ✨
n_splits (int | None) — how many splits to use when chunk_size is None. And when both chunk_size and n_splits are None, it uses n_splits = n_jobs * 64. slay.
worker_lifespan (int | None) — how many tasks a worker can handle before it gets restarted. If it's None, workers stay alive the whole entire time. Use this when your workers are eating up wayyy too much memory over the course of time. 😭
progress_bar (bool) — when True, it shows a progress bar. ✨👑
concatenate_numpy_output (bool) — when True, it concatenates numpy output into a single numpy array. 🧚
worker_init (Callable | None) — function to call every time a new worker starts. If you're passing on the worker ID, it should take the worker ID as its first argument. If shared objects are provided, it should take those as the next argument. If worker state is enabled, it should take a state variable as the next argument. 🌸
worker_exit (Callable | None) — function to call every time a worker exits. Return values get fetched and made available through mpire.WorkerPool.get_exit_results. If you're passing on the worker ID, it should take the worker ID as its first argument. If shared objects are provided, it should take those as the next argument. If worker state is enabled, it should take a state variable as the next argument. 🦋
task_timeout (float | None) — timeout in seconds for a single task. When the timeout is exceeded, MPIRE raises a TimeoutError. Use None to disable (that's the default). ⚠️ Note: this timeout does not apply to worker_init and worker_exit functions — use worker_init_timeout and worker_exit_timeout for those, respectively. iykyk.
worker_init_timeout (float | None) — timeout in seconds for the worker_init function. When the timeout is exceeded, MPIRE raises a TimeoutError. Use None to disable (default). ⏰
worker_exit_timeout (float | None) — timeout in seconds for the worker_exit function. When the timeout is exceeded, MPIRE raises a TimeoutError. Use None to disable (default). ⏰
progress_bar_options (Dict[str, Any] | None) — a dictionary of keyword arguments to pass to the tqdm progress bar. See tqdm.tqdm() for details. The arguments total and leave will be overwritten by MPIRE, so don't even bother bestie. 💀
progress_bar_style (str | None) — the progress bar style to use. Can be one of None, 'std', or 'notebook'. 🎀

Return type Any

Returns List with ordered results ✨

map_unordered(func, iterable_of_args, iterable_len=None,
    max_tasks_active=None, chunk_size=None, n_splits=None,
    worker_lifespan=None, progress_bar=False, worker_init=None,
    worker_exit=None, task_timeout=None, worker_init_timeout=None,
    worker_exit_timeout=None, progress_bar_options=None,
    progress_bar_style=None)

Same as multiprocessing.map(), but unordered 💅. It also lets you set the maximum number of tasks available in the queue. That's it, that's the vibe.

Parameters 🌈

func (Callable) — the function that gets called every time new task arguments show up. If you're passing on the worker ID, it should take the worker ID as its first argument. If shared objects are provided, it should take those as the next argument. If worker state is enabled, it should take a state variable as the next argument. 🔥
iterable_of_args (Sized | Iterable) — a numpy array, or an iterable containing tuples of arguments to pass to a worker, which passes it to the function func. 🫶
iterable_len (int | None) — number of elements in iterable_of_args. When chunk_size is set to None, it needs to know the number of tasks. You can provide that either by implementing the __len__ function on the iterable object, or by specifying the number of tasks. 💖
max_tasks_active (int | None) — maximum number of active tasks in the queue. If None, it will be converted to n_jobs * chunk_size * 2. ✨
chunk_size (int | None) — number of simultaneous tasks to give to a worker. When None, it will use n_splits. 🎀
n_splits (int | None) — number of splits to use when chunk_size is None. When both chunk_size and n_splits are None, it will use n_splits = n_jobs * 64. 👑
worker_lifespan (int | None) — number of tasks a worker can handle before it is restarted. If None, workers will stay alive the entire time. Use this when workers use up too much memory over the course of time. 😭
progress_bar (bool) — when True, it will display a progress bar. ⭐
worker_init (Callable | None) — function to call each time a new worker starts. When passing on the worker ID, the function should receive the worker ID as its first argument. If shared objects are provided, the function should receive those as the next argument. If the worker state has been enabled, it should receive a state variable as the next argument. 🌸
worker_exit (Callable | None) — function to call each time a worker exits. Return values will be fetched and made available through mpire.WorkerPool.get_exit_results. When passing on the worker ID, the function should receive the worker ID as its first argument. If shared objects are provided, the function should receive those as the next argument. If the worker state has been enabled, it should receive a state variable as the next argument. 🦋
task_timeout (float | None) — timeout in seconds for a single task. When the timeout is exceeded, MPIRE will raise a TimeoutError. Use None to disable (default). ⚠️ Note: the timeout doesn't apply to worker_init and worker_exit functions, use worker_init_timeout and worker_exit_timeout for that, respectively. no cap.
worker_init_timeout (float | None) — timeout in seconds for the worker_init function. When the timeout is exceeded, MPIRE will raise a TimeoutError. Use None to disable (default). ⏰
worker_exit_timeout (float | None) — how many seconds we're giving the worker_exit function to do its lil goodbye before MPIRE hits it with a TimeoutError. Pass None to disable it entirely (that's the default, bestie).
- progress_bar_options (Dict[str, Any] | None) — a dict of keyword arguments you wanna hand off to the tqdm progress bar. Peep tqdm.tqdm() for the deets. Heads up: the arguments total and leave get overwritten by MPIRE, so don't get attached 💅
- progress_bar_style (str | None) — which progress bar style you vibe with. One of None, 'std', or 'notebook'.

Return type Any

Returns List with unordered results 🎀

pass_on_worker_id(pass_on=True) : Set whether to pass on the worker ID to the function to be executed or not (default=False).

Parameters

pass_on (bool) — Whether to pass on a worker ID to the target, worker_init, and worker_exit functions. When enabled, functions receive the worker ID depending on other settings. The order is: worker_id, shared_objects, worker_state, and finally the arguments passed on using iterable_of_args

Return type None

print_insights() : Prints insights per worker

Return type None

set_keep_alive(keep_alive=True) : Set whether workers should be kept alive in between consecutive map calls.

Parameters

keep_alive (bool) — When True it will keep workers alive after completing a map call, allowing to reuse workers

Return type None

set_order_tasks(order_tasks=True) : Set whether to provide tasks to the workers in order, such that worker 0 will get chunk 0, worker 1 will get chunk 1, etc.

Parameters

order_tasks (bool) — Whether to provide tasks to the workers in order, such that worker 0 will get chunk 0, worker 1 will get chunk 1, etc.

Return type None

set_shared_objects(shared_objects=None) : Set shared objects to pass to the workers.

Parameters

shared_objects (Any) — Objects to be passed on as shared objects to the workers once. It will be passed on to the target, worker_init, and worker_exit functions. shared_objects is only passed on when it's not None. Shared objects will be copy-on-write when using fork as start method. When enabled, functions receive the shared objects depending on other settings. The order is: worker_id, shared_objects, worker_state, and finally the arguments passed on using iterable_of_args

Return type None

set_use_worker_state(use_worker_state=True) : Set whether or not each worker should have its own state variable. Each worker has its own state, so it's not shared between the workers ✨

Parameters

use_worker_state (bool) — Whether to let a worker have a worker state. The worker state will be passed on to the target, worker_init, and worker_exit functions. When enabled, functions receive the worker state depending on other settings. The order is: worker_id, shared_objects, worker_state, and finally the arguments passed on using iterable_of_args

Return type None

stop_and_join(keep_alive=False) : When keep_alive=False: inserts a poison pill, grabs the exit results, waits until the tasks/results queues are done, and waits until all workers are finished. When keep_alive=True: inserts a non-lethal poison pill, and waits until the tasks/results queues are done.

join and stop_and_join are aliases 🫶

Parameters

keep_alive (bool) — Whether to keep the workers alive

Return type None

terminate() : Tries to do a graceful shutdown of the workers, by interrupting them. In the case processes deadlock it will send a sigkill 💀

Return type None

🌸 AsyncResult 🌸

class mpire.async_result.AsyncResult(cache, callback, error_callback,
job_id=None, delete_from_cache=True, timeout=None)

Adapted from multiprocessing.pool.ApplyResult.

__init__(cache, callback, error_callback, job_id=None,
delete_from_cache=True, timeout=None)

Parameters

cache (Dict) — Cache for storing intermediate results
callback (Callable | None) — Callback function to call when the task is finished. The callback function receives the output of the function as its argument
error_callback (Callable | None) — Callback function to call when the task has failed. The callback function receives the exception as its argument
job_id (int | None) — Job ID of the task. If None, a new job ID is generated
delete_from_cache (bool) — If True, the result is deleted from the cache when the task is finished
timeout (float | None) — Timeout in seconds for a single task. When the timeout is exceeded, MPIRE will raise a TimeoutError. Use None to disable (default)

__weakref__ : list of weak references to the object

get(timeout=None) : Wait until the task is finished and return the output of the function

Parameters

timeout (float | None) — Timeout in seconds. If None, wait indefinitely

Return type Any

Returns Output of the function

Raises TimeoutError if the task is not finished within the timeout. When the task has failed, the exception raised by the function is re-raised 😭

ready()

Return type bool

Returns Returns True if the task is finished

successful()

Return type bool

Returns Returns True if the task has finished successfully

Raises ValueError if the task is not finished yet

wait(timeout=None) : Wait until the task is finished

Parameters

timeout (float | None) — Timeout in seconds. If None, wait indefinitely

Return type None

🧚 Task chunking 🧚

mpire.utils.chunk_tasks(iterable_of_args, iterable_len=None,
chunk_size=None, n_splits=None)

Chunks tasks such that individual workers will receive chunks of tasks rather than individual ones, which can speed up processing drastically 🔥

Parameters

iterable_of_args (Iterable) — A numpy array or an iterable containing tuples of arguments to pass to a worker, which passes it to the function
iterable_len (int | None) — Number of tasks available in iterable_of_args. Only needed when iterable_of_args is a generator
chunk_size (int | float | None) — Number of simultaneous tasks to give to a worker. If None, will use n_splits to determine the chunk size
n_splits (int | None) — Number of splits to use when chunk_size is None

Return type Generator[Collection, None, None]

Returns Generator of chunked task arguments

🌈 Converting iterable of arguments 🌈

mpire.utils.make_single_arguments(iterable_of_args, generator=True)

Converts an iterable of single arguments to an iterable of single argument tuples

Parameters

iterable_of_args (Iterable) — A numpy array or an iterable containing tuples of arguments to pass to a worker, which passes it to the function
generator (bool) — Whether or not to return a generator, otherwise a materialized list will be returned

Return type List | Generator

Returns Iterable of single argument tuples

👑 Dashboard 👑

mpire.dashboard.start_dashboard(port_range=range(8080, 8100))

Starts a new MPIRE dashboard ✨

Parameters

port_range (Sequence) — Port range to try.

Return type Dict[str, int | str]

Returns A dictionary containing the dashboard port number and manager host and port number being used

mpire.dashboard.connect_to_dashboard(manager_port_nr, manager_host=None)

Connects to an existing MPIRE dashboard

Parameters

manager_port_nr (int) — Port to use when connecting to a manager
manager_host (bytes | str | None) — Host to use when connecting to a manager. If None it will use localhost

Return type None
mpire.dashboard.shutdown_dashboard() — closes the dashboard down, bye bestie 👋

Return type None

mpire.dashboard.get_stacklevel() — grabs the stack level that gets used when the dashboard digs up function details

Return type int

Returns Stack level

mpire.dashboard.set_stacklevel(stacklevel) — sets the stack level used when obtaining function details (this is a dashboard thing, iykyk)

Parameters stacklevel (int) – Stack level

Return type None

🌈 Other 🌈

mpire.cpu_count() — returns the number of CPUs in the system. that's it, that's the whole function, lowkey iconic 💅

💖 Contribution guidelines 💖

okay so if you wanna contribute to MPIRE?? slay, we'd literally love that. follow these steps so the whole process stays smooth:

Clone the project.
Create a new branch for your feature or bug fix. Give you branch a meaningful name.
Make your feature addition or bug fix.
Add tests for it and test it yourself. Make sure it both works for Unix and Windows based systems, or make sure to document why it doesn't work for one of the platforms.
Add documentation for it. Don't forget about the changelog:
- Reference the issue number from GitHub in the changelog, if applicable (see current changelog for examples).
- Don't mention a date or a version number here, but use Unreleased instead.
Commit with a meaningful commit message (e.g. the changelog).
Open a pull request.
Resolve any issues or comments by the reviewer.
Merge PR by squashing all your individual commits.

🎀 Making a release 🎀

bestie this one is not for you unless you're the project maintainer. a release is only made by the project maintainer. the following steps are required:

Update the changelog with the release date and version number. Version numbers follow the Semantic Versioning <https://semver.org/> guidelines
Update the version number in setup.py and docs/conf.py.
Commit and push the changes.
Make sure the tests pass on GitHub Actions.
Create a tag for the release by using git tag -a vX.Y.Z -m "vX.Y.Z".
Push the tag to GitHub by using git push origin vX.Y.Z.

✨ Changelog ✨

📌 2.10.2

(2024-05-07)

Function details in progress_bar.py are only obtained when the dashboard is running (#128 <https://github.com/sybrenjansen/mpire/issues/128>)
Obtaining the user name is now put in a try-except block to prevent MPIRE from crashing when the user name cannot be obtained. which can happen when running in a container as a non-root user (#128 <https://github.com/sybrenjansen/mpire/issues/128>)

📌 2.10.1

(2024-03-19)

Fixed a bug in the timeout handler where the cache dictionary could be changed during iteration (#123 <https://github.com/sybrenjansen/mpire/issues/123>)
Fixed an authentication error when using a progress bar or insights in a spawn or forkserver context when using dill (#124 <https://github.com/sybrenjansen/mpire/issues/124>)

📌 2.10.0

(2024-02-19)

Added support for macOS (#27 <https://github.com/sybrenjansen/mpire/issues/27>, #79 <https://github.com/sybrenjansen/mpire/issues/79>, #91 <https://github.com/sybrenjansen/mpire/issues/91>)
- Fixes memory leaks on macOS
- Reduced the amount of semaphores used
- Issues a warning when cpu_ids is used on macOS
Added mpire.dashboard.set_stacklevel() <#mpire.dashboard.set_stacklevel> to set the stack level in the dashboard. This influences what line to display in the 'Invoked on line' section. (#118 <https://github.com/sybrenjansen/mpire/pull/118>)
Use function details from the __call__ method on the dashboard in case the callable being executed is a class instance (#117 <https://github.com/sybrenjansen/mpire/pull/117>)
Use (global) average rate for the estimate on the dashboard when smoothing=0 (#117 <https://github.com/sybrenjansen/mpire/pull/117>)
Make it possible to reuse the same progress_bar_options without raising warnings (#117 <https://github.com/sybrenjansen/mpire/pull/117>)
Removed deprecated progress_bar_position parameter from the map functions. Use progress_bar_options['position'] instead (added since v2.6.0)

📌 2.9.0

(2024-01-08)

Added support for the rich progress bar style (#96 <https://github.com/sybrenjansen/mpire/issues/96>)
Added the option to only show progress on the dashboard. (#107 <https://github.com/sybrenjansen/mpire/pull/107>)
Progress bars are now supported on Windows when using threading as start method.
Insights now also work when using the forkserver and spawn start methods. (#104 <https://github.com/sybrenjansen/mpire/issues/104>)
When using insights on Windows the arguments of the top 5 longest tasks are now available as well.
Fixed deprecated escape import from flask by importing directly from markupsafe. (#106 <https://github.com/sybrenjansen/mpire/issues/106>)
Fixed mpire.dashboard.start_dashboard() <#mpire.dashboard.start_dashboard> freeze when there are no two ports available. (#112 <https://github.com/sybrenjansen/mpire/issues/112>)
Added mpire.dashboard.shutdown_dashboard() <#mpire.dashboard.shutdown_dashboard> to shutdown the dashboard.
Added py.typed file to prompt mypy for type checking. (#108 <https://github.com/sybrenjansen/mpire/pull/108>)

📌 2.8.1

(2023-11-08)

Excluded the tests folder from MPIRE distributions (#89 <https://github.com/sybrenjansen/mpire/issues/89>)
Added a workaround for semaphore leakage on macOS and fixed a bug when working in a fork context while the system default is spawn (#92 <https://github.com/sybrenjansen/mpire/issues/92>)
Fix progressbar percentage on dashboard (#101 <https://github.com/sybrenjansen/mpire/pull/101>)
Fixed a bug where starting multiple apply_async tasks with a task timeout didn't interrupt all tasks when the timeout was reached (#98 <https://github.com/sybrenjansen/mpire/issues/98>)
Add testing python 3.12 to workflow and drop 3.6 and 3.7 (#102 <https://github.com/sybrenjansen/mpire/pull/102>)

📌 2.8.0

(2023-08-16)

Added support for Python 3.11 (#67 <https://github.com/sybrenjansen/mpire/issues/67>)

📌 2.7.1

(2023-04-14)

Transfered ownership of the project from Slimmer AI to sybrenjansen

📌 2.7.0

(2023-03-17)

Added the mpire.WorkerPool.apply() <#mpire.WorkerPool.apply> and mpire.WorkerPool.apply_async() <#mpire.WorkerPool.apply_async> functions (#63 <https://github.com/sybrenjansen/mpire/issues/63>)
When inside a Jupyter notebook, the progress bar will not automatically switch to a widget anymore. tqdm cannot always determine with certainty that someone is in a notebook or, e.g., a Jupyter console. Another reason is to avoid the many errors people get when having widgets or javascript disabled. See Progress bar style <#progress-bar-style> for changing the progress bar to a widget (#71 <https://github.com/sybrenjansen/mpire/issues/71>)
The mpire.dashboard.connect_to_dashboard() <#mpire.dashboard.connect_to_dashboard> function now raises a ConnectionRefused error when the dashboard isn't running, instead of silently failing and deadlocking the next map call with a progress bar (#68 <https://github.com/sybrenjansen/mpire/issues/68>)
Added support for a progress bar without knowing the size of the iterable. It used to disable the progress bar when the size was unknown
Changed how max_tasks_active is handled. It now applies to the number of tasks that are currently being processed, instead of the number of chunks of tasks, as you would expect from the name. Previously, when the chunk size was set to anything other than 1, the number of active tasks could be higher than max_tasks_active
Updated some exception messages and docs (#69 <https://github.com/sybrenjansen/mpire/issues/69>)
Changed how worker results, restarts, timeouts, unexpected deaths, and exceptions are handled. They are now handled by individual threads such that the main thread is more responsive. The API is the same, so no user changes are needed
Mixing multiple map calls now raises an error (see Mixing map functions <#mixing-multiple-map-calls>)
Fixed a bug where calling a map function with a progress bar multiple times in a row didn't display the progress bar correctly
Fixed a bug where the dashboard didn't show an error when an exit function raised an exception

📌 2.6.0

(2022-08-29)

Added Python 3.10 support
The tqdm progress bar can now be customized using the progress_bar_options parameter in the map functions (#57 <https://github.com/sybrenjansen/mpire/issues/57>)
Using progress_bar_position from a map function is now deprecated and will be removed in MPIRE v2.10.0. Use progress_bar_options['position'] instead
Deprecated enable_insights from a map function, use enable_insights in the WorkerPool constructor instead
Fixed a bug where a worker could exit before an exception was entirely sent over the queue, causing a deadlock (#56 <https://github.com/sybrenjansen/mpire/issues/56>)
Fixed a bug where exceptions with init arguments weren't handled correctly (#58 <https://github.com/sybrenjansen/mpire/issues/58>)
Fixed a rare and weird bug in Windows that could cause a deadlock (probably fixes #55 <https://github.com/sybrenjansen/mpire/issues/55>)

📌 2.5.0

(2022-07-25)

Added the option to fix the order of tasks given to the workers (#46 <https://github.com/sybrenjansen/mpire/issues/46>)
Fixed a bug where updated WorkerPool parameters aren't used in subsequent map calls when keep_alive is enabled

📌 2.4.0

(2022-05-25)

A timeout for the target, worker_init, and worker_exit functions can be specified after which a worker is stopped (#36 <https://github.com/sybrenjansen/mpire/issues/36>)
A WorkerPool can now be started within a thread which isn't the main thread (#44 <https://github.com/sybrenjansen/mpire/issues/44>)

📌 2.3.5

(2022-04-25)

MPIRE now handles defunct child processes properly, instead of deadlocking (#34 <https://github.com/sybrenjansen/mpire/issues/34>)
Added benchmark highlights to README (#38 <https://github.com/sybrenjansen/mpire/issues/38>)

📌 2.3.4

(2022-03-29)

Platform specific dependencies are now handled using environment markers as defined in PEP-508 <https://www.python.org/dev/peps/pep-0508/#environment-markers> (#30 <https://github.com/sybrenjansen/mpire/issues/30>)
Fixes hanging WorkerPool when using worker_lifespan and returning results that exceed the pipe capacity (#32 <https://github.com/sybrenjansen/mpire/issues/32>)
Fixes insights unit tests that could sometime fail because it was too fast

2.3.3

(2021-11-29)

Changed progress bar handler process to thread, making it more stable (especially in notebooks)
Changed progress bar tasks completed queue to array, to make it more responsive and faster
Disabled the tqdm monitor thread which, in combination with MPIRE’s own tqdm lock, could result in deadlocks

2.3.2

(2021-11-19)

Included license file in source distribution (#25 <https://github.com/sybrenjansen/mpire/pull/25>)

2.3.1

(2021-11-16)

Made connecting to the tqdm manager more robust (#23 <https://github.com/sybrenjansen/mpire/issues/23>)

2.3.0

(2021-10-15)

Fixed progress bar in a particular setting with iPython and django installed (#13 <https://github.com/sybrenjansen/mpire/pull/13>)
keep_alive now works even when the function to be called or any other parameter passed to the map function is changed (#15 <https://github.com/sybrenjansen/mpire/issues/15>)
Moved enable_insights to the WorkerPool constructor. Using enable_insights from a map function is now deprecated and will be removed in MPIRE v2.6.0.
Restructured docs and updated several sections for Windows users.

2.2.1

(2021-08-31)

Fixed compatibility with newer tqdm versions (>= 4.62.2) (#11 <https://github.com/sybrenjansen/mpire/issues/11>)

2.2.0

(2021-08-30)

Added support for Windows (#6 <https://github.com/sybrenjansen/mpire/issues/6>, #7 <https://github.com/sybrenjansen/mpire/issues/7>). Support has a few caveats:
- When using worker insights the arguments of the top 5 longest tasks are not available
- Progress bar is not supported when using threading as start method
- When using dill and an exception occurs, or when the exception occurs in an exit function, it can print additional OSError messages in the terminal, but these can be safely ignored.

2.1.1

(2021-08-26)

Fixed a bug with newer versions of tqdm. The progress bar would throw an AttributeError when connected to a dashboard.
README and documentation updated

2.1.0

(2021-08-06)

Workers now have their own task queue, which speeds up tasks with bigger payloads
Fixed progress bar showing error information when completed without error
Fixed progress bar and worker insights not displaying properly when using threading
Progress bar handling improved accross several scenarios
Dashboard can now handle progress bars when using spawn or forkserver as start method
Added closing of multiprocessing.JoinableQueue objects, to clean up intermediate junk
Removed numpy dependency
Made dill optional again. In many cases it slows processing down

2.0.0

(2021-07-07)

Worker insights added, providing users insight in multiprocessing efficiency
worker_init and worker_exit parameters added to each map function
max_active_tasks is now set to n_jobs * 2 when max_active_tasks=None, to speed up most jobs
n_splits is now set to n_jobs * 64 when both chunk_size and n_splits are None
Dashboard ports can now be configured
Renamed func_pointer to func in each map function
Fixed a bug with the threading backend not terminating correctly
Fixed a bug with the progress bar not showing correctly in notebooks
Using multiprocess is now the default
Added some debug logging
Refactored a lot of code
Minor bug fixes, which should make things more stable.
Removed Python 3.5 support
Removed add_task, get_result, insert_poison_pill, stop_workers, and join functions from mpire.WorkerPool <#mpire.WorkerPool>. Made start_workers private. There wasn’t any reason to use these functions.

1.2.2

(2021-04-23)

Updated documentation CSS which fixes bullet lists not showing properly

1.2.1

(2021-04-22)

Updated some unittests and fixed some linting issues
Minor improvements in documentation

1.2.0

(2021-04-22)

Workers can be kept alive in between consecutive map calls
Setting CPU affinity is no longer restricted to Linux platforms
README updated to use RST format for better compatibility with PyPI
Added classifiers to the setup file

1.1.3

(2020-09-03)

First public release on Github and PyPi

1.1.2

(2020-08-27)

Added missing typing information
Updated some docstrings
Added license

1.1.1

(2020-02-19)

Changed collections.Iterable to collections.abc.Iterable due to deprecation of the former

1.1.0

(2019-10-31)

Removed custom progress bar support to fix Jupyter notebook support
New progress_bar_position parameter is now available to set the position of the progress bar when using nested worker pools
Screen resizing is now supported when using a progress bar

1.0.0

(2019-10-29)

Added the MPIRE dashboard
Added threading as a possible backend
Progress bar handling now occurs in a separate process, instead of a thread, to improve responsiveness
Refactoring of code and small bug fixes in error handling
Removed deprecated functionality

0.9.0

(2019-03-11)

Added support for using different start methods (‘spawn’ and ‘forkserver’) instead of only the default method ‘fork’
Added optional support for using dill <https://pypi.org/project/dill/> in multiprocessing by utilizing the multiprocess <https://pypi.org/project/multiprocess/> library
The mpire.Worker class is no longer directly available

0.8.1

(2019-02-06)

Fixed bug when process would hang when progress bar was set to True and an empty iterable was provided

0.8.0

(2018-11-01)

Added support for worker state
Chunking numpy arrays is now done using numpy slicing
mpire.WorkerPool.map() <#mpire.WorkerPool.map> now supports automatic concatenation of numpy array output

0.7.2

(2018-06-14)

Small bug fix when not passing on a boolean or tqdm object for the progress_bar parameter

0.7.1

(2017-12-20)

You can now pass on a dictionary as an argument which will be unpacked accordingly using the **-operator.
New function mpire.utils.make_single_arguments() <#mpire.utils.make_single_arguments> added which allows you to create an iterable of single argument tuples out of an iterable of single arguments

0.7.0

(2017-12-11)

mpire.utils.chunk_tasks() <#mpire.utils.chunk_tasks> is now available as a public function
Chunking in above function and map functions now accept a n_splits parameter
iterable_of_args in map functions can now contain single values instead of only iterables
tqdm is now available from the MPIRE package which automatically switches to the Jupyter/IPython notebook widget when available
Small bugfix in cleaning up a worker pool when no map function was called

0.6.2

(2017-11-07)

Fixed a second bug where the main process could get unresponsive when an exception was raised

0.6.1

(2017-11-06)

Fixed bug where sometimes exceptions fail to pickle
Fixed a bug where the main process could get unresponsive when an exception was raised
Child processes are now cleaned up in parallel when an exception was raised

0.6.0

(2017-11-03)

restart_workers parameter is now deprecated and will be removed from v1.0.0
Progress bar functionality added (using tqdm <https://pypi.python.org/pypi/tqdm>)
Improved error handling in user provided functions
Fixed randomly occurring BrokenPipeErrors and deadlocks

0.5.1

(2017-10-12)

Child processes can now also be pinned to a range of CPUs, instead of only a single one. You can also specify a single CPU or range of CPUs that have to be shared between all child processes

0.5.0

(2017-10-06)

Added CPU pinning.
Default number of processes to spawn when using n_jobs=None is now set to the number of CPUs available, instead of cpu_count() - 1

0.4.0

(2017-10-05)

Workers can now be started as normal child processes (non-deamon) such that nested mpire.WorkerPool <#mpire.WorkerPool> s are possible

0.3.0

(2017-09-15)

The worker ID can now be passed on the function to be executed by using the mpire.WorkerPool.pass_on_worker_id() <#mpire.WorkerPool.pass_on_worker_id> function
Removed the use of has_return_value_with_shared_objects when using mpire.WorkerPool.set_shared_objects() <#mpire.WorkerPool.set_shared_objects>. MPIRE now handles both cases out of the box

0.2.0

(2017-06-27)

Added docs

0.1.0

First release

Author

Sybren Jansen

Copyright

2026, Sybren Jansen

And that's a wrap, bestie! This was MPIRE(1), version 2.10.2, dated Jan 31, 2026. Stay iconic, keep slaying those processes! 👑💖✨

📦 where this page comes from

ubuntu resolute: python3-mpire 2.10.2-6 · MIT
ubuntu stonking: python3-mpire 2.10.2-6 · MIT
💖 girlified by deepseek-ai/DeepSeek-V4.1-Flash on 2026-11-02: a language model's rewrite of the original, under its terms