parallelbar

Parallelbar

Table of contents

• Instalation
• Usage
• Exception handling
• Changelog
  • New in version 2.6
  • New in version 2.5
  • New in version 2.4
  • New in version 2.3
  • New in version 1.3
  • New in version 1.2
  • New in version 1.1
  • New in version 1.0
  • New in version 0.3
• Problems of the naive approach
• License

Parallelbar displays the progress of tasks in the process pool for Pool class methods such as map, starmap (since 1.2 version), imap and imap_unordered. Parallelbar is based on the tqdm module and the standard python multiprocessing library. Also, it is possible to handle exceptions that occur within a separate process, as well as set a timeout for the execution of a task by a process.

Installation

pip install parallelbar

or

pip install --user git+https://github.com/dubovikmaster/parallelbar.git

Usage

from parallelbar import progress_imap, progress_map, progress_imapu
from parallelbar.tools import cpu_bench, fibonacci

Let’s create a list of 100 numbers and test progress_map with default parameters on a toy function cpu_bench:

tasks = range(10000)

%%time
list(map(cpu_bench, tasks))

Wall time: 52.6 s

Ok, by default this works on one core of my i7-9700F and it took 52 seconds. Let’s parallelize the calculations for all 8 cores and look at the progress. This can be easily done by replacing standart function map with progress_map.

if __name__=='__main__':
    progress_map(cpu_bench, tasks)

Core progress:

You can also easily use progress_imap and progress_imapu analogs of the imap and imap_unordered methods of the Pool() class

%%time
if __name__=='__main__':
    tasks = [20 + i for i in range(15)]
    result = progress_imap(fibonacci, tasks, chunk_size=1)

Reusable (warm) pool: Parallel (since 2.6)

Every call to progress_map/progress_imap/… creates a fresh process pool (and a multiprocessing.Manager), which spawns processes each time. If you run many parallel operations in a row, this overhead dominates. The Parallel class creates the workers once and reuses them:

from parallelbar import Parallel
from parallelbar.tools import cpu_bench


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


def risky(n):
    if n % 10 == 0:
        raise ValueError(f'bad value: {n}')
    return n * n


if __name__ == '__main__':
    # Create the pool once and reuse it for several operations.
    with Parallel(n_cpu=4) as par:
        # 1) plain map
        squares = par.map(cpu_bench, range(1000))

        # 2) starmap (each task is a tuple of arguments)
        sums = par.starmap(add, [(1, 2), (3, 4), (5, 6)])
        # -> [3, 7, 11]

        # 3) imap (ordered) / imapu (unordered) also available
        ordered = par.imap(cpu_bench, range(1000))

        # 4) collect exceptions instead of raising, and get the failed tasks
        results, failed = par.map(
            risky, range(50),
            error_behavior='coerce',
            return_failed_tasks=True,
        )
        # results[i] is either n*n or a ValueError instance (with .traceback)
        # failed == [0, 10, 20, 30, 40]

        # 5) automatically retry transient failures up to 3 times
        recovered = par.map(risky, range(1, 10), retries=3)
    # leaving the `with` block closes the workers and the manager process

You can also manage the lifetime manually instead of using the context manager:

par = Parallel(n_cpu=4)
try:
    result = par.map(cpu_bench, range(1000))
finally:
    par.close()   # or par.terminate() to stop immediately

For a workload of many small sequential map calls this is dramatically faster (an order of magnitude in a simple benchmark) because process startup happens only once. Parallel supports map, starmap, imap, imapu with the same options as the functional API (error_behavior, set_error_value, return_failed_tasks, process_timeout, need_serialize, retries, …).

Note: Parallel uses the process pool only (no thread executor and no add_progress decorator path). retries is available for map/starmap/imap (not imapu).

Retrying failed tasks (since 2.6)

Pass retries=N to automatically re-submit tasks that raised an exception (including tasks that hit process_timeout). Retry attempts are shown as separate RETRY N bars:

if __name__ == '__main__':
    # each failing task is retried up to 3 more times before giving up
    result = progress_map(unstable_task, tasks, retries=3)

After the retries are exhausted, the remaining failures are handled according to error_behavior (raise re-raises the last exception; coerce puts set_error_value/the exception into the result). Combine with return_failed_tasks=True to also get the list of tasks that still failed after all attempts. retries works for progress_map, progress_starmap, progress_imap and the corresponding Parallel methods. It is not available for imapu/imap_unordered, because retried results cannot be positionally aligned with the input (for that case use a per-call retry inside the worker function instead).

Exception handling

You can handle exceptions and set timeouts for the execution of tasks by the process.
Consider the following toy example:

def foo(n):
    if n==5 or n==17:
        1/0
    elif n==10:
        time.sleep(2)
    else:
        time.sleep(1)
    return n
if __name__=='__main__':
	res = progress_map(foo, range(20), process_timeout=5, n_cpu=8, error_behavior='coerce')

As you can see, under the main progress bar, another progress bar has appeared that displays the number of tasks that ended unsuccessfully. At the same time, the main bar turned orange, as if signaling something went wrong

print(res)
	[0, 1, 2, 3, 4, ZeroDivisionError('division by zero'), 6, 7, 8, 9, 10, 11, 12,
     13, 14, 15, 16, ZeroDivisionError('division by zero'), 18, 19]

In the resulting array, we have exceptions in the corresponding places (to collect exceptions into the result instead of raising, use error_behavior='coerce'). The worker-side traceback of each coerced exception is available as a string via .traceback:

print(res[5].traceback)
Traceback (most recent call last):
  File ".../parallelbar/parallelbar.py", line ..., in _func_wrapped
    result = func(task)
  File "<stdin>", line 3, in foo
    1/0
ZeroDivisionError: division by zero

From which you can tell where in the code the exception occurred. Let’s add a timeout of 1.5 seconds for each task. If a task runs longer than 1.5 seconds, a TimeoutError will be raised and handled:

if __name__=='__main__':
	res = progress_map(foo, range(20), process_timeout=1.5, n_cpu=8, error_behavior='coerce')

print(res)
	[0, 1, 2, 3, 4, ZeroDivisionError('division by zero'), 6, 7, 8, 9, 'function foo took longer than 1.5 s.', 
	11, 12, 13, 14, 15, 16, ZeroDivisionError('division by zero'), 18, 19]

Exception handling has also been added to methods progress_imap and progress_imapu.

Note on process_timeout. The timeout is implemented in-process: it interrupts the running Python code by raising an exception (via a signal). This is a soft timeout — if a task is stuck inside a C extension or a blocking system call that does not return control to the Python interpreter, the timeout cannot fire until the call returns. It does not forcibly kill and restart the worker process. For CPU-bound pure-Python tasks it works as expected.

Changelog

Version 2.6

• Coerced exceptions now carry the worker-side traceback as a string in exc.traceback, so result[i].traceback can be inspected when error_behavior='coerce' (works for the functional API and the add_progress decorator).
• KeyboardInterrupt (Ctrl+C) now reliably stops the status thread, re-raises, and tears down the worker pool and manager process instead of leaving orphaned processes behind.
• Added the Parallel class — a reusable (“warm”) pool of workers. Creating a pool spawns processes on every call; Parallel creates them once and reuses them across many map/starmap/imap/imapu calls, giving a large speedup for many sequential parallel operations (see the Usage section).
• Added the retries parameter to progress_map, progress_starmap and progress_imap (and to the matching Parallel methods). Tasks that raise an exception (including process_timeout) are automatically re-submitted up to retries times; the retry attempts are shown as separate RETRY N bars. Not available for imapu/imap_unordered, since retried results can’t be positionally aligned.
• Added separate_bar (plus desc and total) parameters to the add_progress decorator. When separate_bar=True, each decorated function gets its own progress bar in addition to the aggregate DONE bar (see the Usage section).
• Bug fixes in the internal engine:
  • Replaced a busy-wait loop that stopped the status thread (it spun a CPU core at 100% and flooded the queue) with a single sentinel and thread.join().
  • Wrapped the multiprocessing.Manager in a context manager and reduced it to a single manager for both queues, fixing a lingering manager-process leak on every call.
  • Replaced the unreliable queue.qsize() drain of failed tasks (which can raise NotImplementedError on macOS) with a safe get_nowait() loop.

Version 2.5

• Introduced an optional timeout parameter to progress_map and progress_starmap for managing execution time limits.

Version 2.4

• fixed issue
• For Windows OS, when using the add_progress decorator, the function being decorated no longer needs the worker_queue keyword argument.

New in version 2.3

• added wrappers module with which contains decorators:
  • stop_it_after_timeout - stops the function execution after the specified time (in seconds)
  • add_progress - adds a progress bar to the function execution, exception handling and timeout.

Usage example for UNIX systems:

from parallelbar.wrappers import add_progress
from parallelbar import progress_map
import time


@add_progress(error_handling='coerce', timeout=.5)
def foo(n):
    if n==5 or n==17:
        1/0
    elif n==10:
        time.sleep(1)
    else:
        time.sleep(.1)
    return n

def bar(x):
    return [foo(i) for i in range(x)]

if __name__=='__main__':
    # you must specify the total number of tasks
    res = progress_map(bar, [10, 20, 30, 40], n_cpu=4, total=100)

Out:

For Windows systems you need to add the worker_queue parameter to the functions foo and bar and use the used_add_progress_decorator parameter in the progress_map function:

@add_progress(error_handling='coerce', timeout=.5)
def foo(n):
    if n==5 or n==17:
        1/0
    elif n==10:
        time.sleep(1)
    else:
        time.sleep(.1)
    return n

def bar(x, worker_queue=None):
    return [foo(i, worker_queue=worker_queue) for i in range(x)]

if __name__=='__main__':
    res = progress_map(bar, [10, 20, 30, 40], n_cpu=4, total=100, used_add_progress_decorator=True)

Out:

Separate progress bar per function (since 2.6)

By default every @add_progress function feeds the single aggregate DONE bar. If your worker calls several functions and you want to track each of them independently, pass separate_bar=True. Each decorated function then gets its own bar in addition to the aggregate DONE bar:

• desc — label of the separate bar (defaults to the function name);
• total — total number of calls of that function across all tasks. If omitted, the bar is shown without a percentage (as a plain counter).

import time
from parallelbar import progress_map
from parallelbar.wrappers import add_progress


# foo is called 10 + 20 + 30 + 40 = 100 times across all tasks
@add_progress(separate_bar=True, total=100, desc='foo')
def foo(n):
    time.sleep(0.02)
    return n


# baz is called 10 times per task * 4 tasks = 40 times
@add_progress(separate_bar=True, total=40, desc='baz')
def baz(n):
    time.sleep(0.02)
    return n


def pipeline(x):
    a = [foo(i) for i in range(x)]
    b = [baz(i) for i in range(10)]
    return len(a) + len(b)


if __name__ == '__main__':
    # aggregate DONE total = foo (100) + baz (40) = 140
    progress_map(pipeline, [10, 20, 30, 40], n_cpu=4,
                 total=140, used_add_progress_decorator=True)

This renders three live bars: DONE (aggregate, 140), foo (100) and baz (40). Note that errors from all functions are still collected into a single shared ERROR bar.

You can also use the stopit_after_timeout decorator separately:

from parallelbar.wrappers import stopit_after_timeout
from parallelbar import progress_map
import time


@stopit_after_timeout(.5, raise_exception=True)
def foo(n):
    if n==5:
        time.sleep(1)
    else:
        time.sleep(.1)
    return n

if __name__=='__main__':
    print(f'first result is: {foo(3)}')
    print(f'second result is: {foo(5)}')

Out:

first result is: 3

TimeoutError                              Traceback (most recent call last)
Cell In[7], line 16
     14 if __name__=='__main__':
     15     print(foo(3))
---> 16     print(foo(5))

File /opt/conda/envs/user_response/lib/python3.10/site-packages/parallelbar/wrappers.py:38, in stopit_after_timeout.<locals>.actual_decorator.<locals>.wrapper(*args, **kwargs)
     36     msg = f'function took longer than {s} s.'
     37     if raise_exception:
---> 38         raise TimeoutError(msg)
     39     result = msg
     40 finally:

TimeoutError: function took longer than 0.5 s.

• added return_failed_tasks keyword parameter to the progress_map/starmap/imap/imapu function (default=False) - if True then the result will include the tasks that failed with an exception.

New in version 1.3

• added maxtaskperchild keyword parameter to the progress_map/starmap/imap/imapu function (default=None)

New in version 1.2

• Added progress_starmap function. An extension of the starmap method of the Pool class.
• Improved documentation.

New in version 1.1

1. The bar_step keyword argument is no longer used and will be removed in a future version
2. Added need_serialize boolean keyword argument to the progress_map/imap/imapu function (default False). Requires dill to be installed. If True the target function is serialized using dill library. Thus, as a target function, you can now use lambda functions, class methods and other callable objects that pickle cannot serialize
3. Added dynamic optimization of the progress bar refresh rate. This can significantly improve the performance of the progress_map/imap/imapu functions ror very long iterables and small execution time of one task by the objective function.

New in version 1.0

1. The “ignore” value of the error_behavior key parameter is no longer supported.
2. Default value of key parameter error_behavior changed to “raise”.
3. The pebble module is no longer used.
4. Added key parameter executor in the functions progress_map, progress_imap and progress_imapu. Must be one of the values:
   • “threads” - use thread pool
   • “processes” - use processes pool (default)

New in version 0.3.0

1. The error_behavior keyword argument has been added to the progress_map, progress_imap and progress_imapu methods. Must be one of the values: “raise”, “ignore”, “coerce”.
   • “raise” - raise an exception thrown in the process pool.
   • “ignore” - ignore the exceptions that occur. Do not add anything to the result
   • “coerce” - handle the exception. The result will include the value set by the parameter set_error_value (by default None - the traceback of the raised exception will be added to the result)
2. The set_error_value keyword argument has been added to the progress_map, progress_imap and progress_imapu methods.

Example of usage

import time
import resource as rs
from parallelbar import progress_imap


def memory_limit(limit):
    soft, hard = rs.getrlimit(rs.RLIMIT_AS)
    rs.setrlimit(rs.RLIMIT_AS, (limit, hard))


def my_awesome_foo(n):
    if n == 0:
        s = 'a' * 10000000
    elif n == 20:
        time.sleep(100)
    else:
        time.sleep(1)
    return n


if __name__ == '__main__':
    tasks = range(30)
    start = time.monotonic()
    result = progress_imap(my_awesome_foo, tasks, 
                           process_timeout=1.5, 
                           initializer=memory_limit, 
                           initargs=(100,),
                           n_cpu=4,
                           error_behavior='coerce',
                           set_error_value=None,
                           )
    print(f'time took: {time.monotonic() - start:.1f}')
    print(result)

time took: 8.2
[MemoryError(), 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 
16, 17, 18, 19, TimeoutError('function "my_awesome_foo" took longer than 1.5 s.'), 21, 22, 23, 24, 25, 26, 27, 28, 29]

Set NaN instead of tracebacks to the result of the pool operation:

if __name__ == '__main__':
    tasks = range(30)
    start = time.monotonic()
    result = progress_imap(my_awesome_foo, tasks, 
                           process_timeout=1.5, 
                           initializer=memory_limit, 
                           initargs=(100,),
                           n_cpu=4,
                           error_behavior='coerce',
                           set_error_value=float('nan'),
                           )
    print(f'time took: {time.monotonic() - start:.1f}')
    print(result)

time took: 8.0
[nan, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 
16, 17, 18, 19, nan, 21, 22, 23, 24, 25, 26, 27, 28, 29]

Let’s ignore exception:

if __name__ == '__main__':
    tasks = range(30)
    start = time.monotonic()
    result = progress_imap(my_awesome_foo, tasks, 
                           process_timeout=1.5, 
                           initializer=memory_limit, 
                           initargs=(100,),
                           n_cpu=4,
                           error_behavior='ignore',
                           set_error_value=None,
                           )
    print(f'time took: {time.monotonic() - start:.1f}')
    print(result)

time took: 8.0
[1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 
16, 17, 18, 19, 21, 22, 23, 24, 25, 26, 27, 28, 29]

Problems of the naive approach

Why can’t I do something simpler? Let’s take the standard imap method and run through it in a loop with tqdm and take the results from the processes:

from multiprocessing import Pool
from tqdm.auto import tqdm

if __name__=='__main__':
    with Pool() as p:
        tasks = [20 + i for i in range(15)]
        pool = p.imap(fibonacci, tasks)
        result = []
        for i in tqdm(pool, total=len(tasks)):
            result.append(i)

It looks good, doesn’t it? But let’s do the following, make the first task very difficult for the core. To do this, I will insert the number 38 at the beginning of the tasks list. Let’s see what happens

if __name__=='__main__':
    with Pool() as p:
        tasks = [20 + i for i in range(15)]
        tasks.insert(0, 39)
        pool = p.imap_unordered(fibonacci, tasks)
        result = []
        for i in tqdm(pool, total=len(tasks)):
            result.append(i)

This is a fiasco. Our progress hung on the completion of the first task and then at the end showed 100% progress. Let’s try to do the same experiment only for the progress_imap function:

if __name__=='__main__':
    tasks = [20 + i for i in range(15)]
    tasks.insert(0, 39)
    result = progress_imap(fibonacci, tasks)

The progress_imap function takes care of collecting the result and closing the process pool for you. In fact, the naive approach described above will work for the standard imap_unordered method. But it does not guarantee the order of the returned result. This is often critically important.

License

MIT license

This site is open source. Improve this page.