MPIRE(1)

NAME
mpire - mpire Documentation

MPIRE, short for MultiProcessing Is Really Easy, is a Python package for multiprocessing. MPIRE is faster in most scenarios, packs more features, and is generally more user-friendly than the default multiprocessing package. It combines the convenient map like functions of multiprocessing.Pool with the benefits of using copy-on-write shared objects of multiprocessing.Process, together with easy-to-use worker state, worker insights, worker init and exit functions, timeouts, and progress bar functionality.

FEATURES

    •   Faster execution than other multiprocessing libraries. See benchmarks <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)

    •   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.

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 at Windows <#troubleshooting-windows>.

CONTENTS

  Installation

MPIRE <#secret> 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.

  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/>. dill isn’t installed by default as 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.

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 <#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.

Functions like these are known as embarrassingly parallel <https://en.wikipedia.org/wiki/Embarrassingly_parallel> problems, 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.

MPIRE can be used almost as a drop-in replacement to multiprocessing. We use the mpire.WorkerPool <#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.

MPIRE also offers a dashboard, for which you need to install additional dependencies <#dashboarddep>. See Dashboard <#dashboard> 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.

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 Shared objects <#shared-objects> 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 <#worker-init-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.

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() <#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 Worker insights <#worker-insights> for a more detailed example and expected output.

  Usage

  WorkerPool

This section describes how to setup a mpire.WorkerPool <#mpire.WorkerPool> instance.

  Starting a WorkerPool

  Contents

    •   Nested WorkerPools

The mpire.WorkerPool <#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() <#mpire.WorkerPool.map_unordered>), together with the apply and apply_async functions (see Apply family <#apply-family>).

An mpire.WorkerPool <#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.

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 using 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() <#mpire.WorkerPool.terminate>. For more information, see the warnings in the Python docs here <https://docs.python.org/3/library/multiprocessing.html#pipes-and-queues>.

  Nested WorkerPools

By default, the mpire.WorkerPool <#mpire.WorkerPool> class 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 <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 <https://docs.python.org/3/library/multiprocessing.html#contexts-and-start-methods> and caveats <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:
Start method | Available on Unix | Available on Windows
fork | Yes (default) | No
spawn | Yes | Yes (default)
forkserver | Yes | No
threading | Yes | 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 <#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() <#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 <#mpire.WorkerPool> constructor you can also make use of mpire.WorkerPool.pass_on_worker_id() <#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.

Apart from sharing regular Python objects between workers, you can also share multiprocessing synchronization primitives such as multiprocessing.Lock using this method. Objects like these require to be shared through inheritance, which is exactly how shared objects in MPIRE are passed on.

Important:
        Shared objects are passed on as the second argument, after the worker ID (when enabled), to the provided function.

Instead of passing the shared objects to the mpire.WorkerPool <#mpire.WorkerPool> constructor you can also use the mpire.WorkerPool.set_shared_objects() <#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. Workers are started once the first map call is executed. When keep_alive=True and the workers are reused, changing the shared objects between two consecutive map calls won’t work.

  Copy-on-write alternatives

When copy-on-write is not available for you, you can also use shared objects to share a multiprocessing.Array, multiprocessing.Value, or another object with multiprocessing.Manager. You can then store results in the same object from multiple processes. However, you should keep the amount of synchronization to a minimum when the resources are protected with a lock, or disable locking if your situation allows it as is shown here:

            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)

In the example above we create two results containers, one for squaring and for adding the given value, and disable locking for both. Additionally, we also return a value, even though we use shared objects for storing results. We can safely disable locking here as each task writes to a different index in the array, so no race conditions can occur. Disabling locking is, of course, a lot faster than having it enabled.

  Worker state

  Contents

    •   Combining worker state with worker_init and worker_exit

    •   Combining worker state with keep_alive

If you want to let each worker have its own state you can 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 is passed on as the third argument, after the worker ID and shared objects (when enabled), to the provided function.

Instead of passing the flag to the mpire.WorkerPool <#mpire.WorkerPool> constructor you can also make use of mpire.WorkerPool.set_use_worker_state() <#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

The worker state can be combined with the worker_init and worker_exit parameters of each map function, leading to some 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 information about the worker_init and worker_exit parameters can be found at Worker init and exit <#worker-init-exit>.

  Combining worker state with keep_alive

By default, workers are restarted each time a map function is executed. As described in Keep alive <#keep-alive> this can be circumvented by using keep_alive=True. This also ensures 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 supply the worker_init function to the second map call, as the workers will be reused. When worker_lifespan is set, though, this rule doesn’t apply.

  Keep alive

  Contents

    •   Caveats

By default, workers are restarted on each map call. This is done to clean up resources as quickly as possible when the work is done.

Workers can be kept alive in between consecutive map calls using the keep_alive flag. This is useful 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

Instead of passing the flag to the mpire.WorkerPool <#mpire.WorkerPool> constructor you can also make use of mpire.WorkerPool.set_keep_alive() <#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 do require a restart. These include pass_worker_id, shared_objects, and use_worker_state.

Keeping workers alive works even when the function to be called or any other parameter passed on to the map function changes.

However, when you’re changing either the worker_init and/or worker_exit function while keep_alive is enabled, you need to be aware this can have undesired side-effects. worker_init functions are only executed when a worker is started and worker_exit functions when a worker is terminated. When keep_alive is enabled, workers aren’t restarted in between consecutive map calls, so those functions are not 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 above example 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 workers are kept alive. During the second map call init_func_2 isn’t called as well, because the workers are still alive. When exiting the context manager the workers are shut down and exit_func_2 is called.

It gets even trickier when you also enable worker_lifespan. In this scenario during the first map call a worker could’ve reached its maximum lifespan and is forced to restart, while others haven’t. The exit function of the worker to be restarted is called (i.e., exit_func_1). When calling map for the second time and the exit function is changed, the other workers will execute the new exit function when they need to be restarted (i.e., exit_func_2).

  Worker insights

Worker insights gives you insight in your multiprocessing efficiency by tracking worker start up time, waiting time and time spend on executing tasks. Tracking is disabled by default, but can be enabled by setting enable_insights:

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

The overhead is very minimal and you shouldn’t really notice it, even on very small tasks. You can view the tracking results using mpire.WorkerPool.get_insights() <#mpire.WorkerPool.get_insights> or use mpire.WorkerPool.print_insights() <#mpire.WorkerPool.print_insights> to directly print the insights to 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']}

We specified 4 workers, so there are 4 entries in the n_completed_tasks, start_up_time, init_time, waiting_time, working_time, and exit_time containers. They show per worker the number of completed tasks, the total start up time, the total time spend on the worker_init function, the total time waiting for new tasks, total time spend on main function, and the total time spend on the worker_exit function, respectively. The insights also contain mean, standard deviation, and ratio of the tracked time. The ratio is the time for that part divided by the total time. In general, the higher the working ratio the more efficient your multiprocessing setup is. Of course, your setup might still not be optimal because the task itself is inefficient, but timing that is beyond the scope of MPIRE.

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.

When 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.

  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. 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 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.

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.

  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.

            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() <#mpire.utils.make_single_arguments> expects an iterable of arguments and converts them to tuples accordingly. The second argument of this function specifies if you want the function to return a generator or a materialized list. If we would like to return a generator we would need to pass on the iterable length as well.

  Mixing map functions

map functions cannot be used while another map function is still running. 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.

  Not exhausting a lazy imap function

If you don’t exhaust a lazy imap function, but do close the pool, the remaining tasks and results will be lost. 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

  Contents

    •   Progress bar style

        •   Changing the default style

    •   Progress bar options

        •   Progress bar position

Progress bar support is added through the tqdm <https://pypi.python.org/pypi/tqdm> package (installed by default when installing MPIRE). 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).

  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 <#richdep>)

    •   '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 <#troubleshooting-progress-bar>.

  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. 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.

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 <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).

It goes without saying that you shouldn’t specify the same progress bar position multiple times.

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() <#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() <#mpire.WorkerPool.get_exit_results> does not always equal n_jobs.

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() <#mpire.WorkerPool.stop_and_join> to receive the exit results.

  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.

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.

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

When you have tasks that take up a lot of memory you can do a few things:

    •   Limit the number of jobs (i.e., the number of tasks currently being available to the workers, tasks that are in the queue ready to be processed).

    •   Limit the number of active tasks

The first option is the most obvious one to save memory when the processes themselves use up much memory. The second is convenient when the argument list takes up too much memory. For example, suppose you want to kick off an enormous amount of jobs (let’s say a billion) of which the arguments take up 1 KB per task (e.g., large strings), then that task queue would take up ~1 TB of memory!

In such cases, a good rule of thumb would be to have twice the amount of active chunks of tasks than there are jobs. This means that when all workers complete their task at the same time each would directly be able to continue with another task. When workers take on their new tasks the generator of tasks is iterated to the point that again there would be twice the amount of active chunks of tasks.

In MPIRE, the maximum number of active tasks by default is set to n_jobs * chunk_size * 2, so you don’t have to tweak it for memory optimization. If, for whatever reason, you want to change this behavior, you can do so by setting 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:
        Setting the max_tasks_active parameter to a value lower than n_jobs * chunk_size can result in some workers not being able to do anything.

  Worker lifespan

Occasionally, workers that process multiple, memory intensive tasks do not release their used up memory properly, which results in memory usage building up. This is not a bug in MPIRE, but a consequence of Python’s poor garbage collection. To avoid this type of problem you can 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 each worker is restarted after finishing a single task.

Note:
        When the worker lifespan has been reached, a worker will finish the current chunk of tasks before restarting. I.e., based on the chunk_size a worker could end up completing more tasks than is allowed by the worker lifespan.

  Timeouts

Timeouts can be set separately for the target, worker_init and worker_exit functions. When a timeout has been set and reached, it will throw 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.

  imap and imap_unordered

When you’re using one of the lazy map functions (e.g., imap or imap_unordered) then an exception will only be raised when the function is actually running. E.g. when executing:

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

this will never raise. This is because imap and imap_unordered return a generator object, which stops executing until it gets the trigger to go beyond the yield statement. When iterating through the results, it will raise as expected:

            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.

  Numpy arrays

  Contents

    •   Chunking

    •   Return value

  Chunking

Numpy arrays are treated a little bit differently when 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, each row of the array is now in its own array and each one of them is given to the target function individually. Instead, MPIRE will chunk them in to something more reasonable using numpy slicing instead:

            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]])

Each chunk is now a single numpy array containing as many rows as the chunk size, except for the last chunk as there aren’t enough rows left.

  Return value

When the user provided function returns numpy arrays and you’re applying the mpire.WorkerPool.map() <#mpire.WorkerPool.map> function MPIRE will concatenate the resulting numpy arrays to a single array by default. For example:

            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]])

This behavior can be cancelled by using 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)

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 <#mpire.WorkerPool> implements two apply functions, which are very similar to the ones in the multiprocessing module:

    mpire.WorkerPool.apply() <#mpire.WorkerPool.apply>
        Apply a function to a single task. This is a blocking call.

    mpire.WorkerPool.apply_async() <#mpire.WorkerPool.apply_async>
        A variant of the above, but which is non-blocking. This returns an mpire.async_result.AsyncResult <#mpire.async_result.AsyncResult> object.

  apply

The apply function is a blocking call, which means that it will not return until the task is completed. If you want to run multiple different tasks in parallel, you should use the apply_async function instead. If you require 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, similar to 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 that it will return immediately. It returns an mpire.async_result.AsyncResult <#mpire.async_result.AsyncResult> object, which can be used 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! E.g., 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() <#mpire.WorkerPool.stop_and_join> function to stop the workers and join the pool. This will make sure that 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

The mpire.async_result.AsyncResult <#mpire.async_result.AsyncResult> object has the following convenient methods:

            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

Each apply function has a callback and error_callback argument. These are functions which are called when the task is finished. The callback function is called with the result of the task when the task was completed successfully, and the error_callback is 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

As with the map family of functions, the apply family of functions also has worker_init and worker_exit arguments. These are functions which are called when a worker is started and stopped, respectively. See Worker init and exit <#worker-init-exit> for more information on these functions.

            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)

There’s a caveat though. When the first apply or apply_async function is executed, the entire pool of workers is started. This means that in the above example all five workers are started, while only one was needed. This also means that the worker_init function is set for all those workers at once. This means you cannot have a different worker_init function for each apply task. A second, different worker_init function will simply be ignored.

Similarly, the worker_exit function can only be set once as well. Additionally, exit functions are only called when a worker exits, which in this case translates to when the pool exits. This means that 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() <#mpire.WorkerPool.stop_and_join> to stop the workers, which will cause the worker_exit function to be triggered for each worker.

  Timeouts

The apply family of functions also has task_timeout, worker_init_timeout and worker_exit_timeout arguments. These are timeouts for the task, the worker_init function and the worker_exit function, respectively. They work similarly as those for the map functions.

When a single task times out, only that task is cancelled. The other tasks will continue to run. When a worker init or exit times out, the entire pool is stopped.

See Timeouts <#timeouts> for more information.

  Dashboard

The dashboard allows you to see progress information from a browser. This is convenient when running scripts in a notebook or screen, if you want to share the progress information with others, or if you want to get real-time worker insight information.

The dashboard dependencies are not installed by default. See Dashboard <#dashboarddep> for more information.

  Contents

    •   Starting the dashboard

    •   Connecting to an existing dashboard

    •   Using the dashboard

    •   Stack level

  Starting the dashboard

You can start the dashboard programmatically:

            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 will start a dashboard on your local machine on port 8080. When the port is already in use MPIRE will try the next, until it finds an unused one. In the rare case that no ports are available 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 is ultimately chosen. It also contains information 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 will start a dashboard with the connection details printed on screen. It will 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() <#mpire.dashboard.start_dashboard>. Similarly to earlier, a custom port range can be provided:

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

The benefit of starting a dashboard this way is that your dashboard keeps running in case of errors in your script. You will be able to see what the error was, when it occurred and where it occurred in your code.

  Connecting to an existing dashboard

If you have 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 in the examples above.

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

  Using the dashboard

Once connected to a dashboard you don’t need to change anything to your code. When you have 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 will let you know when a KeyboardInterrupt signal was send to the running process: [image: ] [image]

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

In case you have enabled Worker insights <#worker-insights> these insights will be 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 will refresh automatically every 0.5 seconds.

  Stack level

By default, the dashboard will show information about the function that is called and where it is called from. However, in some cases where you have wrapped the function in another function, you might be less interested in the wrapper function and more interested in the function that is calling this wrapper. In such cases you can use mpire.dashboard.set_stacklevel() <#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 will see that the dashboard will show two progress bars. In both cases, the dashboard will show the square function as the function that is called. However, in the first case, it will show return pool.map(func, data, progress_bar=self.progress_bar) as the line where it is called from. In the second case, it will show the results = executor(square, range(10000)) line.

  Troubleshooting

This section describes some known problems that can arise when using MPIRE.

  Contents

    •   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 using the progress bar in a Jupyter notebook you might encounter some issues. A few of these are described below, together with possible solutions.

  IProgress not found

When you 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 <https://ipywidgets.readthedocs.io/en/stable/user_install.html> for more information.

  Widget Javascript not detected

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

Note that you have to restart your notebook server after enabling the extension, simply restarting the kernel won’t be enough.

  Unit tests

When using the 'spawn' or 'forkserver' method you’ll probably run into one or two issues when running unittests in your own package. One problem that might occur is that your unittests will restart whenever the piece of code containing such a start method is called, leading to very funky terminal output. To remedy this problem make sure your setup call in setup.py is surrounded by an if __name__ == '__main__': clause:

            from setuptools import setup

            if __name__ == '__main__':

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

See the ‘Safe importing of main module’ section at caveats <https://docs.python.org/3/library/multiprocessing.html#the-spawn-and-forkserver-start-methods>.

The second problem you might encounter is that the semaphore tracker of multiprocessing will complain 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 will 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. Unfortunately, I’ve not found a remedy to this problem using python setup.py test yet. What you can use instead is something like the following:

            python -m unittest tests.some_test

This will work just fine. See the unittest <https://docs.python.org/3.4/library/unittest.html#command-line-interface> documentation for more information.

  Shutting down takes a long time on error

When you issue a KeyboardInterrupt or when an error occured in the function that’s run in parallel, there are situations where MPIRE needs a few seconds to gracefully shutdown. This has to do with the fact that in these situations the task or results queue can be quite full, still. MPIRE drains these queues until they’re completely empty, as to properly shutdown and clean up every communication channel.

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 <#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.

  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. This would add a heavy performance penalty and is therefore not an acceptable solution.

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 <#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.

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 <#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.

        __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 (Optional[int]) – 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[Union[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) – Whether to use dill as serialization backend. Some exotic types (e.g., lambdas, nested functions) don’t work well when using spawn as start method. In such cased, use dill (can be a bit slower sometimes)

                •   enable_insights (bool) – Whether to enable worker insights. Might come at a small performance penalty (often neglible)

                •   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.

        __weakref__
            list of weak references to the object

        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. This is a blocking call.

            Parameters

                •   func (Callable) – 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 (Optional[Callable]) – Callback function to call when the task is finished. The callback function receives the output of the function func as its argument

                •   error_callback (Optional[Callable]) – Callback function to call when the task has failed. The callback function receives the exception as its argument

                •   worker_init (Optional[Callable]) – 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 (Optional[Callable]) – 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 (Optional[float]) – 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 (Optional[float]) – 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 (Optional[float]) – 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. This is a non-blocking call.

            Parameters

                •   func (Callable) – 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 (Optional[Callable]) – Callback function to call when the task is finished. The callback function receives the output of the function func as its argument

                •   error_callback (Optional[Callable]) – Callback function to call when the task has failed. The callback function receives the exception as its argument

                •   worker_init (Optional[Callable]) – 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 (Optional[Callable]) – 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 (Optional[float]) – 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 (Optional[float]) – 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 (Optional[float]) – 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. Also allows a user to set the maximum number of tasks available in the queue.

            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 (Union[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 (Optional[int]) – 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 (Optional[int]) – Maximum number of active tasks in the queue. If None it will be converted to n_jobs * chunk_size * 2

                •   chunk_size (Optional[int]) – Number of simultaneous tasks to give to a worker. When None it will use n_splits.

                •   n_splits (Optional[int]) – 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 (Optional[int]) – 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 (Optional[Callable]) – 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 (Optional[Callable]) – 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 (Optional[float]) – 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 (Optional[float]) – 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 (Optional[float]) – 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 (Optional[Dict[str, Any]]) – 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 (Optional[str]) – The progress bar style to use. Can be one of None, 'std', or 'notebook'

            Return type
                Generator[Any, None, None]

            Returns
                Generator yielding ordered results

        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(). Also allows a user to set the maximum number of tasks available in the queue.

            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 (Union[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 (Optional[int]) – 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 (Optional[int]) – Maximum number of active tasks in the queue. If None it will be converted to n_jobs * chunk_size * 2

                •   chunk_size (Optional[int]) – Number of simultaneous tasks to give to a worker. When None it will use n_splits.

                •   n_splits (Optional[int]) – 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 (Optional[int]) – 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 (Optional[Callable]) – 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 (Optional[Callable]) – 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 (Optional[float]) – 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 (Optional[float]) – 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 (Optional[float]) – 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 (Optional[Dict[str, Any]]) – 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 (Optional[str]) – 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: 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

        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)
            Same as multiprocessing.map(). Also allows a user to set the maximum number of tasks available in the queue. Note that this function can be slower than the unordered version.

            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 (Union[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 (Optional[int]) – 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 (Optional[int]) – Maximum number of active tasks in the queue. If None it will be converted to n_jobs * chunk_size * 2

                •   chunk_size (Optional[int]) – Number of simultaneous tasks to give to a worker. When None it will use n_splits.

                •   n_splits (Optional[int]) – 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 (Optional[int]) – 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

                •   concatenate_numpy_output (bool) – When True it will concatenate numpy output to a single numpy array

                •   worker_init (Optional[Callable]) – 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 (Optional[Callable]) – 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 (Optional[float]) – 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 (Optional[float]) – 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 (Optional[float]) – 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 (Optional[Dict[str, Any]]) – 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 (Optional[str]) – 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. Also allows a user to set the maximum number of tasks available in the queue.

            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 (Union[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 (Optional[int]) – 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 (Optional[int]) – Maximum number of active tasks in the queue. If None it will be converted to n_jobs * chunk_size * 2

                •   chunk_size (Optional[int]) – Number of simultaneous tasks to give to a worker. When None it will use n_splits.

                •   n_splits (Optional[int]) – 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 (Optional[int]) – 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 (Optional[Callable]) – 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 (Optional[Callable]) – 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 (Optional[float]) – 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 (Optional[float]) – 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 (Optional[float]) – 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 (Optional[Dict[str, Any]]) – 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 (Optional[str]) – The progress bar style to use. Can be 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 (Optional[Callable]) – Callback function to call when the task is finished. The callback function receives the output of the function as its argument

                •   error_callback (Optional[Callable]) – Callback function to call when the task has failed. The callback function receives the exception as its argument

                •   job_id (Optional[int]) – 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 (Optional[float]) – 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 (Optional[float]) – 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 (Optional[float]) – 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 (Optional[int]) – Number of tasks available in iterable_of_args. Only needed when iterable_of_args is a generator

            •   chunk_size (Union[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 (Optional[int]) – 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
            Union[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, Union[str, int]]

        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 (Union[bytes, str, None]) – Host to use when connecting to a manager. If None it will use localhost

        Return type
            None

    mpire.dashboard.shutdown_dashboard()
        Shuts down the dashboard

        Return type
            None

    mpire.dashboard.get_stacklevel()
        Gets the stack level to use when obtaining function details (used for the dashboard)

        Return type
            int

        Returns
            Stack level

    mpire.dashboard.set_stacklevel(stacklevel)
        Sets the stack level to use when obtaining function details (used for the dashboard)

        Parameters
            stacklevel (int) – Stack level

        Return type
            None

  Other

    mpire.cpu_count()
        Returns the number of CPUs in the system

  Contribution guidelines

If you want to contribute to MPIRE, great! Please follow the steps below to ensure a smooth process:

    1.  Clone the project.

    2.  Create a new branch for your feature or bug fix. Give you branch a meaningful name.

    3.  Make your feature addition or bug fix.

    4.  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.

    5.  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.

    6.  Commit with a meaningful commit message (e.g. the changelog).

    7.  Open a pull request.

    8.  Resolve any issues or comments by the reviewer.

    9.  Merge PR by squashing all your individual commits.

  Making a release

A release is only made by the project maintainer. The following steps are required:

    1.  Update the changelog with the release date and version number. Version numbers follow the Semantic Versioning <https://semver.org/> guidelines

    2.  Update the version number in setup.py and docs/conf.py.

    3.  Commit and push the changes.

    4.  Make sure the tests pass on GitHub Actions.

    5.  Create a tag for the release by using git tag -a vX.Y.Z -m "vX.Y.Z".

    6.  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
2025, Sybren Jansen