MPIRE

https://github.com/sybrenjansen/mpire
Admin
MPIRE is a Python multiprocessing library designed to provide faster execution, intuitive syntax,...
Tokens:22,671
Snippets:259
Trust Score:7
License:MIT
Update:1 month ago
Show doc for...
Tokens:
Raw
### Install documentation dependencies from source

Source: https://github.com/sybrenjansen/mpire/blob/master/README.rst

Install the documentation dependencies from the current directory.

```bash
pip install .[docs]
```

--------------------------------

### Start dashboard programmatically

Source: https://github.com/sybrenjansen/mpire/blob/master/docs/usage/dashboard.rst

Starts the dashboard and returns a dictionary containing connection details.

```python
from mpire.dashboard import start_dashboard

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

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

--------------------------------

### Working job with spawn start method

Source: https://github.com/sybrenjansen/mpire/blob/master/docs/usage/workerpool/start_method.rst

Example of a job that correctly imports the required module inside the function, allowing it to work with the 'spawn' start method.

```python
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')])
```

--------------------------------

### Install documentation dependencies via pip

Source: https://github.com/sybrenjansen/mpire/blob/master/README.rst

Install the necessary dependencies to build the documentation using pip.

```bash
pip install mpire[docs]
```

--------------------------------

### Failing job with spawn start method

Source: https://github.com/sybrenjansen/mpire/blob/master/docs/usage/workerpool/start_method.rst

Example of a job that fails when using the 'spawn' start method because the 'os' module is not available in the child process scope.

```python
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')])
```

--------------------------------

### Start dashboard via CLI

Source: https://github.com/sybrenjansen/mpire/blob/master/docs/usage/dashboard.rst

Starts the dashboard using the bash script, which is not supported on Windows.

```bash
$ mpire-dashboard
```

```bash
Starting MPIRE dashboard...

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

--------------------------------

### Install rich for progress bars

Source: https://github.com/sybrenjansen/mpire/blob/master/docs/install.rst

Manual installation of the rich dependency to enable rich progress bars.

```bash
pip install rich
```

--------------------------------

### Build documentation using make

Source: https://github.com/sybrenjansen/mpire/blob/master/README.rst

Build the documentation from the docs folder using make, assuming MPIRE is installed in the environment.

```bash
make html
```

--------------------------------

### Install MPIRE via pip

Source: https://github.com/sybrenjansen/mpire/blob/master/docs/install.rst

Standard installation command for the MPIRE library using the Python package manager.

```bash
pip install mpire
```

--------------------------------

### Start dashboard via CLI with custom port range

Source: https://github.com/sybrenjansen/mpire/blob/master/docs/usage/dashboard.rst

Starts the dashboard using the bash script with a specified port range.

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

--------------------------------

### Install MPIRE via conda-forge

Source: https://github.com/sybrenjansen/mpire/blob/master/docs/install.rst

Installation command for the MPIRE library using the conda package manager.

```bash
conda install -c conda-forge mpire
```

--------------------------------

### Install MPIRE with dill support

Source: https://github.com/sybrenjansen/mpire/blob/master/docs/install.rst

Installs the multiprocess package to enable dill serialization. Note that this changes the MPIRE license to BSD.

```bash
pip install mpire[dill]
```

--------------------------------

### Protect setup.py with if __name__ == '__main__'

Source: https://github.com/sybrenjansen/mpire/blob/master/docs/troubleshooting.rst

Surround your setup call with an if-clause to prevent unit test restarts when using 'spawn' or 'forkserver' methods.

```python
from setuptools import setup

if __name__ == '__main__':

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

--------------------------------

### AsyncResult methods

Source: https://github.com/sybrenjansen/mpire/blob/master/docs/usage/apply.rst

Demonstrates common methods available on the AsyncResult object, such as ready, wait, get, and successful.

```python
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()
```

--------------------------------

### Install ipywidgets via pip

Source: https://github.com/sybrenjansen/mpire/blob/master/docs/troubleshooting.rst

Use this command to install ipywidgets if you encounter an 'IProgress not found' error in Jupyter.

```bash
pip install ipywidgets
```

--------------------------------

### Start WorkerPool with context manager

Source: https://github.com/sybrenjansen/mpire/blob/master/docs/usage/workerpool/setup.rst

The recommended approach for initializing a pool, which ensures worker processes are properly joined or terminated after the block completes.

```python
from mpire import WorkerPool

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

--------------------------------

### Install MPIRE dashboard dependencies

Source: https://github.com/sybrenjansen/mpire/blob/master/docs/install.rst

Installs Flask and other dependencies for the MPIRE dashboard. Note that this changes the MPIRE license to BSD.

```bash
pip install mpire[dashboard]
```

--------------------------------

### Sequential execution example

Source: https://github.com/sybrenjansen/mpire/blob/master/docs/getting_started.rst

A standard sequential implementation of a time-consuming function using a list comprehension.

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

--------------------------------

### Output of manual task chunking

Source: https://github.com/sybrenjansen/mpire/blob/master/docs/usage/map/task_chunking.rst

The expected output from the manual chunk_tasks examples.

```python
[(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,)]
```

--------------------------------

### Start WorkerPool manually

Source: https://github.com/sybrenjansen/mpire/blob/master/docs/usage/workerpool/setup.rst

Manual initialization requires explicit cleanup using stop_and_join() or terminate(). Note that the results queue must be drained before joining to avoid potential deadlocks.

```python
# 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()
```

--------------------------------

### Worker initialization and exit

Source: https://github.com/sybrenjansen/mpire/blob/master/docs/usage/apply.rst

Functions can be specified to run when a worker starts or stops using worker_init and worker_exit.

```python
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)
```

--------------------------------

### Example semaphore tracker error output

Source: https://github.com/sybrenjansen/mpire/blob/master/docs/troubleshooting.rst

This is an example of the semaphore tracker error output that may occur when running individual unit tests.

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

--------------------------------

### Install pywin32 for Windows conda users

Source: https://github.com/sybrenjansen/mpire/blob/master/docs/install.rst

Use this command if you encounter a DLL failed to load error on Windows when using conda.

```bash
conda install pywin32
```

--------------------------------

### Pinning child processes to CPUs with WorkerPool

Source: https://github.com/sybrenjansen/mpire/blob/master/docs/usage/workerpool/cpu_pinning.rst

Examples of using the cpu_ids parameter in the WorkerPool constructor to pin child processes to specific CPU cores or ranges.

```python
# 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:
    ...
```

--------------------------------

### Initialize and clean up workers with worker_init and worker_exit

Source: https://github.com/sybrenjansen/mpire/blob/master/docs/usage/map/worker_init_exit.rst

This example demonstrates using worker_init to set up a counter in worker_state and worker_exit to return that counter after processing. The results are collected using pool.get_exit_results after the map operation completes.

```python
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
```

--------------------------------

### Deadlock example with apply_async

Source: https://github.com/sybrenjansen/mpire/blob/master/docs/usage/apply.rst

Attempting to retrieve results after the pool has exited will cause a deadlock.

```python
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]
```

--------------------------------

### Start dashboard with custom port range

Source: https://github.com/sybrenjansen/mpire/blob/master/docs/usage/dashboard.rst

Overrides the default port range by passing a custom range object to start_dashboard.

```python
dashboard_details = start_dashboard(range(9000, 9100))
```

--------------------------------

### Install ipywidgets via conda

Source: https://github.com/sybrenjansen/mpire/blob/master/docs/troubleshooting.rst

Use this command to install ipywidgets via conda if you encounter an 'IProgress not found' error in Jupyter.

```bash
conda install -c conda-forge ipywidgets
```

--------------------------------

### Configure max_tasks_active in WorkerPool.map

Source: https://github.com/sybrenjansen/mpire/blob/master/docs/usage/map/max_tasks_active.rst

Example of setting the max_tasks_active parameter when calling the map method on a WorkerPool.

```python
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))
```

--------------------------------

### Retrieve and print worker insights

Source: https://github.com/sybrenjansen/mpire/blob/master/docs/usage/workerpool/worker_insights.rst

Use get_insights() to retrieve performance data or print_insights() to output it to the console. The example shows the structure of the returned dictionary containing timing metrics per worker.

```python
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']}
```

--------------------------------

### Passing shared objects via WorkerPool constructor

Source: https://github.com/sybrenjansen/mpire/blob/master/docs/usage/workerpool/shared_objects.rst

Demonstrates passing a large dataset to workers using the shared_objects parameter in the WorkerPool constructor with the 'fork' start method.

```python
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))
```

--------------------------------

### Sharing objects between workers

Source: https://github.com/sybrenjansen/mpire/blob/master/docs/getting_started.rst

Using the shared_objects parameter for copy-on-write memory sharing. Note that this requires the 'fork' start method and is not available on Windows.

```python
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)
```

--------------------------------

### Build documentation using setup.py

Source: https://github.com/sybrenjansen/mpire/blob/master/README.rst

Build the documentation using setup.py, requiring Python 3.9 or lower.

```bash
python setup.py build_docs
```

--------------------------------

### Run unit tests with python -m unittest

Source: https://github.com/sybrenjansen/mpire/blob/master/docs/troubleshooting.rst

Use this command as an alternative to python setup.py test to avoid semaphore tracker errors during unit testing.

```bash
python -m unittest tests.some_test
```

--------------------------------

### Configure progress bar options

Source: https://github.com/sybrenjansen/mpire/blob/master/docs/usage/map/progress_bar.rst

Pass a dictionary of tqdm keyword arguments to the progress_bar_options parameter to customize appearance like description, units, and color.

```python
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'})
```

--------------------------------

### Manual argument wrapping to avoid unpacking

Source: https://github.com/sybrenjansen/mpire/blob/master/docs/usage/map/map.rst

Demonstrates how to pass dictionaries as a single argument by wrapping them in a tuple.

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

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},),
                                            ...]):
        ...
```

--------------------------------

### Set task, worker_init, and worker_exit timeouts

Source: https://github.com/sybrenjansen/mpire/blob/master/docs/usage/map/timeouts.rst

Demonstrates setting task_timeout, worker_init_timeout, and worker_exit_timeout within the WorkerPool map method.

```python
# 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)
```

--------------------------------

### Enable basic progress bar

Source: https://github.com/sybrenjansen/mpire/blob/master/docs/usage/map/progress_bar.rst

Enable a basic tqdm progress bar by setting the progress_bar flag to True in a map function.

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

--------------------------------

### Using pool.imap with multiple arguments

Source: https://github.com/sybrenjansen/mpire/blob/master/docs/usage/map/map.rst

Demonstrates passing multiple arguments to a function using zip and imap for lazy processing.

```python
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):
        ...
```

--------------------------------

### Worker initialization and state management

Source: https://github.com/sybrenjansen/mpire/blob/master/docs/getting_started.rst

Using worker_init to prepare worker-specific state, such as loading models or datasets, accessed via worker_state.

```python
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)
```

--------------------------------

### Manual task chunking with chunk_tasks

Source: https://github.com/sybrenjansen/mpire/blob/master/docs/usage/map/task_chunking.rst

Shows how to use the chunk_tasks utility function to manually partition iterables.

```python
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)))
```

--------------------------------

### Configuring map with generators

Source: https://github.com/sybrenjansen/mpire/blob/master/docs/usage/map/task_chunking.rst

Demonstrates various ways to use map with generators, including providing iterable_len or chunk_size to control chunking behavior.

```python
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)
```

--------------------------------

### Set Timeouts for Tasks and Initialization

Source: https://github.com/sybrenjansen/mpire/blob/master/README.rst

Configure timeouts for target functions, worker initialization, and worker exit functions. Exceeding these limits raises a TimeoutError.

```python
def init():
    ...

def exit_():
    ...

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

--------------------------------

### Enable Jupyter widget extension

Source: https://github.com/sybrenjansen/mpire/blob/master/docs/troubleshooting.rst

Run this command to enable the widget Javascript extension if you see a 'Widget Javascript not detected' error. You must restart your notebook server after running this.

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

--------------------------------

### Basic usage of pool.map with a range

Source: https://github.com/sybrenjansen/mpire/blob/master/docs/usage/map/map.rst

Demonstrates squaring numbers using pool.map with a standard range object.

```python
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))
```

--------------------------------

### Initialize WorkerPool with Insights

Source: https://github.com/sybrenjansen/mpire/blob/master/docs/getting_started.rst

Configures a WorkerPool with 5 jobs and enables insights tracking. Insights can be retrieved after the map operation using pool.get_insights().

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

--------------------------------

### Combine worker state with worker_init

Source: https://github.com/sybrenjansen/mpire/blob/master/docs/usage/workerpool/worker_state.rst

Initialize worker state by loading large resources like models using the worker_init parameter.

```python
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)
```

--------------------------------

### Using make_single_arguments utility

Source: https://github.com/sybrenjansen/mpire/blob/master/docs/usage/map/map.rst

Uses the make_single_arguments utility to automatically wrap arguments for functions that expect a single dictionary.

```python
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)):
        ...
```

--------------------------------

### Unpacking dictionaries with pool.imap

Source: https://github.com/sybrenjansen/mpire/blob/master/docs/usage/map/map.rst

Shows how dictionaries passed as arguments are automatically unpacked using the **-operator.

```python
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}, ...]):
        ...
```

--------------------------------

### Setting shared objects using set_shared_objects

Source: https://github.com/sybrenjansen/mpire/blob/master/docs/usage/workerpool/shared_objects.rst

Shows how to define shared objects after the WorkerPool has been initialized using the set_shared_objects method.

```python
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))
```

--------------------------------

### Enable dill in WorkerPool

Source: https://github.com/sybrenjansen/mpire/blob/master/docs/usage/workerpool/dill.rst

Use the use_dill flag when initializing the WorkerPool to enable the dill serialization backend.

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

--------------------------------

### Connect to an existing dashboard

Source: https://github.com/sybrenjansen/mpire/blob/master/docs/usage/dashboard.rst

Connects to a dashboard running elsewhere using the manager port and host.

```python
from mpire.dashboard import connect_to_dashboard

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

--------------------------------

### Parallel execution with MPIRE WorkerPool

Source: https://github.com/sybrenjansen/mpire/blob/master/docs/getting_started.rst

Using MPIRE's WorkerPool as a drop-in replacement for multiprocessing.

```python
from mpire import WorkerPool

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

--------------------------------

### Enable task ordering via WorkerPool constructor

Source: https://github.com/sybrenjansen/mpire/blob/master/docs/usage/workerpool/order_tasks.rst

Sets the order_tasks flag to True during the initialization of the WorkerPool to enable round-robin task distribution.

```python
def task(x):
    pass

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

--------------------------------

### Use dashboard with WorkerPool

Source: https://github.com/sybrenjansen/mpire/blob/master/docs/usage/dashboard.rst

Connects to a dashboard and enables a progress bar in a WorkerPool map call.

```python
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)
```

--------------------------------

### Enable task ordering via set_order_tasks method

Source: https://github.com/sybrenjansen/mpire/blob/master/docs/usage/workerpool/order_tasks.rst

Uses the set_order_tasks method on an existing WorkerPool instance to enable round-robin task distribution.

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

--------------------------------

### Using callbacks and error_callbacks

Source: https://github.com/sybrenjansen/mpire/blob/master/docs/usage/apply.rst

Callbacks are triggered upon task completion, with separate functions for success and failure.

```python
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)
```

--------------------------------

### Enabling a progress bar

Source: https://github.com/sybrenjansen/mpire/blob/master/docs/getting_started.rst

Adding a tqdm progress bar to the WorkerPool map function by setting progress_bar to True.

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

--------------------------------

### Enable worker state via set_use_worker_state

Source: https://github.com/sybrenjansen/mpire/blob/master/docs/usage/workerpool/worker_state.rst

Alternatively, enable worker state dynamically using the set_use_worker_state method on an existing pool instance.

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

--------------------------------

### Enable worker state via WorkerPool constructor

Source: https://github.com/sybrenjansen/mpire/blob/master/docs/usage/workerpool/worker_state.rst

Use the use_worker_state flag in the WorkerPool constructor to allow each worker to maintain its own state.

```python
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))
```

--------------------------------

### mpire.cpu_count

Source: https://github.com/sybrenjansen/mpire/blob/master/docs/reference/index.rst

Utility to retrieve the number of available CPUs.

```APIDOC
## mpire.cpu_count

### Description
Returns the number of CPUs available for parallel processing.
```

--------------------------------

### Configure stack level with set_stacklevel

Source: https://github.com/sybrenjansen/mpire/blob/master/docs/usage/dashboard.rst

Demonstrates using set_stacklevel to adjust the stack depth when using a WorkerPoolWrapper. Setting the level to 2 allows the dashboard to attribute the call to the line where the executor is invoked rather than the internal pool.map call.

```python
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))
```

--------------------------------

### Nested WorkerPool configuration

Source: https://github.com/sybrenjansen/mpire/blob/master/docs/usage/workerpool/setup.rst

Demonstrates that nested pools require daemon=False to function. Using daemon=True will raise an AssertionError.

```python
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, ...)
```

--------------------------------

### Handling large iterables with iterable_len

Source: https://github.com/sybrenjansen/mpire/blob/master/docs/usage/map/map.rst

Shows how to use iterable_len and chunk_size when the iterable length cannot be determined automatically.

```python
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)
```

--------------------------------

### Change default progress bar style

Source: https://github.com/sybrenjansen/mpire/blob/master/docs/usage/map/progress_bar.rst

Set the global default progress bar style by modifying the mpire.tqdm_utils.PROGRESS_BAR_DEFAULT_STYLE variable.

```python
import mpire.tqdm_utils

mpire.tqdm_utils.PROGRESS_BAR_DEFAULT_STYLE = 'notebook'
```

--------------------------------

### Change progress bar style

Source: https://github.com/sybrenjansen/mpire/blob/master/docs/usage/map/progress_bar.rst

Switch the progress bar style using the progress_bar_style parameter. The 'notebook' style is shown here.

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

--------------------------------

### Enable worker insights in WorkerPool

Source: https://github.com/sybrenjansen/mpire/blob/master/docs/usage/workerpool/worker_insights.rst

Enable tracking of worker performance metrics by setting enable_insights to True when initializing the WorkerPool.

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

--------------------------------

### Default chunking behavior with itertools.islice

Source: https://github.com/sybrenjansen/mpire/blob/master/docs/usage/map/numpy.rst

Demonstrates how standard iteration over a numpy array splits it into individual rows.

```python
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
```

```python
[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])]
```

--------------------------------

### Efficient chunking with mpire.utils.chunk_tasks

Source: https://github.com/sybrenjansen/mpire/blob/master/docs/usage/map/numpy.rst

Uses MPIRE's utility to chunk numpy arrays into blocks of rows instead of individual rows.

```python
from mpire.utils import chunk_tasks

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

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

--------------------------------

### mpire.dashboard

Source: https://github.com/sybrenjansen/mpire/blob/master/docs/reference/index.rst

Functions to manage the MPIRE dashboard for monitoring.

```APIDOC
## mpire.dashboard Functions

### Description
- start_dashboard: Starts the MPIRE dashboard server.
- connect_to_dashboard: Connects to an existing dashboard.
- shutdown_dashboard: Shuts down the running dashboard.
- get_stacklevel: Retrieves the current stack level.
- set_stacklevel: Sets the stack level for the dashboard.
```

--------------------------------

### Avoiding deadlock with stop_and_join

Source: https://github.com/sybrenjansen/mpire/blob/master/docs/usage/apply.rst

Using stop_and_join ensures all tasks are completed before the pool exits, preventing deadlocks.

```python
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]
```

--------------------------------

### Mixing map functions exception

Source: https://github.com/sybrenjansen/mpire/blob/master/docs/usage/map/map.rst

Demonstrates that calling a new map function while an imap function is still running will raise an exception.

```python
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))
```

--------------------------------

### mpire.utils.chunk_tasks

Source: https://github.com/sybrenjansen/mpire/blob/master/docs/reference/index.rst

Utility function to chunk tasks for efficient processing.

```APIDOC
## mpire.utils.chunk_tasks

### Description
Splits an iterable of tasks into smaller chunks to be processed by workers.
```

--------------------------------

### Lazy map timeout behavior with imap

Source: https://github.com/sybrenjansen/mpire/blob/master/docs/usage/map/timeouts.rst

Shows that imap does not raise a timeout exception upon initialization, but does so during iteration.

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

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

--------------------------------

### Enable worker ID via pass_on_worker_id method

Source: https://github.com/sybrenjansen/mpire/blob/master/docs/usage/workerpool/worker_id.rst

Alternatively, enable the worker ID passing dynamically using the pass_on_worker_id method on an existing WorkerPool instance.

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

--------------------------------

### Enable keep_alive using set_keep_alive method

Source: https://github.com/sybrenjansen/mpire/blob/master/docs/usage/workerpool/keep_alive.rst

Use the set_keep_alive method to enable worker reuse dynamically after the pool has been initialized.

```python
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
```

--------------------------------

### Exiting with block before exhausting imap

Source: https://github.com/sybrenjansen/mpire/blob/master/docs/usage/map/map.rst

Demonstrates that exiting the with block terminates the pool, causing an exception if the imap results are accessed afterwards.

```python
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)
```

--------------------------------

### Use multiple progress bars with position

Source: https://github.com/sybrenjansen/mpire/blob/master/docs/usage/map/progress_bar.rst

Use the position parameter within progress_bar_options to manage multiple progress bars, such as in nested WorkerPools.

```python
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()
```

--------------------------------

### Parallel execution with multiprocessing.Pool

Source: https://github.com/sybrenjansen/mpire/blob/master/docs/getting_started.rst

Parallelizing the function using the standard library's multiprocessing.Pool class.

```python
from multiprocessing import Pool

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

--------------------------------

### mpire.async_result.AsyncResult

Source: https://github.com/sybrenjansen/mpire/blob/master/docs/reference/index.rst

Represents the result of an asynchronous operation.

```APIDOC
## mpire.async_result.AsyncResult

### Description
AsyncResult provides a way to access the results of tasks submitted to the WorkerPool asynchronously.
```

--------------------------------

### Sharing multiprocessing.Array with disabled locking

Source: https://github.com/sybrenjansen/mpire/blob/master/docs/usage/workerpool/shared_objects.rst

Illustrates using shared multiprocessing.Array objects to store results from multiple processes, with locking disabled for performance when race conditions are avoided by index-based writing.

```python
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)
```

--------------------------------

### Non-blocking apply_async function

Source: https://github.com/sybrenjansen/mpire/blob/master/docs/usage/apply.rst

The apply_async function returns immediately with an AsyncResult object, allowing for parallel task submission.

```python
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]
```

--------------------------------

### mpire.WorkerPool

Source: https://github.com/sybrenjansen/mpire/blob/master/docs/reference/index.rst

The primary class for managing a pool of workers for parallel processing.

```APIDOC
## mpire.WorkerPool

### Description
WorkerPool is the main entry point for creating and managing a pool of worker processes in MPIRE.
```

--------------------------------

### Use worker ID with shared objects

Source: https://github.com/sybrenjansen/mpire/blob/master/docs/usage/workerpool/worker_id.rst

Demonstrates using the worker ID to index into a shared array, allowing workers to update specific elements without requiring explicit locking.

```python
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))
```

--------------------------------

### Caveats with worker_init and worker_exit functions

Source: https://github.com/sybrenjansen/mpire/blob/master/docs/usage/workerpool/keep_alive.rst

When keep_alive is enabled, worker_init and worker_exit functions are not re-executed between map calls, which can lead to unexpected behavior if these functions are changed.

```python
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)
```

--------------------------------

### Blocking apply function

Source: https://github.com/sybrenjansen/mpire/blob/master/docs/usage/apply.rst

The apply function is a blocking call that waits for the task to complete before returning.

```python
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)
```

--------------------------------

### Setting worker_lifespan in WorkerPool

Source: https://github.com/sybrenjansen/mpire/blob/master/docs/usage/map/worker_lifespan.rst

Configures the worker pool to restart workers after completing a single task to manage memory usage.

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

--------------------------------

### Enable keep_alive in WorkerPool constructor

Source: https://github.com/sybrenjansen/mpire/blob/master/docs/usage/workerpool/keep_alive.rst

Pass keep_alive=True to the WorkerPool constructor to reuse workers across multiple map calls.

```python
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
```

--------------------------------

### Persist worker state across map calls with keep_alive

Source: https://github.com/sybrenjansen/mpire/blob/master/docs/usage/workerpool/worker_state.rst

Use keep_alive=True to maintain worker state across multiple consecutive map calls without re-initializing.

```python
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)
```

--------------------------------

### Enable worker ID via WorkerPool constructor

Source: https://github.com/sybrenjansen/mpire/blob/master/docs/usage/workerpool/worker_id.rst

Pass the worker ID to the task function by setting pass_worker_id=True in the WorkerPool constructor. The worker ID is always the first argument provided to the function.

```python
def task(worker_id, x):
    pass

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

--------------------------------

### Disabling result concatenation

Source: https://github.com/sybrenjansen/mpire/blob/master/docs/usage/map/numpy.rst

Use the concatenate_numpy_output flag to return individual arrays instead of a single concatenated array.

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

```python
[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]])]
```

--------------------------------

### Terminating pool before exhausting imap

Source: https://github.com/sybrenjansen/mpire/blob/master/docs/usage/map/map.rst

Shows that calling pool.terminate() while an imap function is still active will cause subsequent attempts to access results to raise an exception.

```python
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)
```
Tokens:
Raw