stouputils.parallel.multi module#

multiprocessing(
func: Callable[[...], R] | list[Callable[[...], R]],
args: Iterable,
use_starmap: bool = False,
chunksize: int = 1,
desc: str = '',
max_workers: int | float = 4,
capture_output: bool = False,
delay_first_calls: float = 0,
nice: int | None = None,
color: str = '\x1b[95m',
bar_format: str = '{l_bar}{bar}\x1b[95m| {n_fmt}/{total_fmt} [{rate_fmt}{postfix}, {elapsed}<{remaining}]\x1b[0m',
ascii: bool = False,
smooth_tqdm: bool = True,
**tqdm_kwargs: Any,
) → list[R][source]#

Method to execute a function in parallel using multiprocessing

  • For CPU-bound operations where the GIL (Global Interpreter Lock) is a bottleneck.

  • When the task can be divided into smaller, independent sub-tasks that can be executed concurrently.

  • For computationally intensive tasks like scientific simulations, data analysis, or machine learning workloads.

Parameters:

  • func (Callable | list[Callable]) – Function to execute, or list of functions (one per argument)

  • args (Iterable) – Iterable of arguments to pass to the function(s)

  • use_starmap (bool) – Whether to use starmap or not (Defaults to False): True means the function will be called like func(*args[i]) instead of func(args[i])

  • chunksize (int) – Number of arguments to process at a time (Defaults to 1 for proper progress bar display)

  • desc (str) – Description displayed in the progress bar (if not provided no progress bar will be displayed)

  • max_workers (int | float) – Number of workers to use (Defaults to CPU_COUNT), -1 means CPU_COUNT. If float between 0 and 1, it’s treated as a percentage of CPU_COUNT. If negative float between -1 and 0, it’s treated as a percentage of len(args).

  • capture_output (bool) – Whether to capture stdout/stderr from the worker processes (Defaults to True)

  • delay_first_calls (float) – Apply i*delay_first_calls seconds delay to the first “max_workers” calls. For instance, the first process will be delayed by 0 seconds, the second by 1 second, etc. (Defaults to 0): This can be useful to avoid functions being called in the same second.

  • nice (int | None) – Adjust the priority of worker processes (Defaults to None). Use Unix-style values: -20 (highest priority) to 19 (lowest priority). Positive values reduce priority, negative values increase it. Automatically converted to appropriate priority class on Windows. If None, no priority adjustment is made.

  • color (str) – Color of the progress bar (Defaults to MAGENTA)

  • bar_format (str) – Format of the progress bar (Defaults to BAR_FORMAT)

  • ascii (bool) – Whether to use ASCII or Unicode characters for the progress bar

  • smooth_tqdm (bool) – Whether to enable smooth progress bar updates by setting miniters and mininterval (Defaults to True)

  • **tqdm_kwargs (Any) – Additional keyword arguments to pass to tqdm

Returns:

Results of the function execution

Return type:

list[object]

Examples

> multiprocessing(doctest_square, args=[1, 2, 3])
[1, 4, 9]

> multiprocessing(int.__mul__, [(1,2), (3,4), (5,6)], use_starmap=True)
[2, 12, 30]

> # Using a list of functions (one per argument)
> multiprocessing([doctest_square, doctest_square, doctest_square], [1, 2, 3])
[1, 4, 9]

> # Will process in parallel with progress bar
> multiprocessing(doctest_slow, range(10), desc="Processing")
[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]

> # Will process in parallel with progress bar and delay the first threads
> multiprocessing(
.     doctest_slow,
.     range(10),
.     desc="Processing with delay",
.     max_workers=2,
.     delay_first_calls=0.6
. )
[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
multithreading(
func: Callable[[...], R] | list[Callable[[...], R]],
args: Iterable,
use_starmap: bool = False,
desc: str = '',
max_workers: int | float = 4,
delay_first_calls: float = 0,
color: str = '\x1b[95m',
bar_format: str = '{l_bar}{bar}\x1b[95m| {n_fmt}/{total_fmt} [{rate_fmt}{postfix}, {elapsed}<{remaining}]\x1b[0m',
ascii: bool = False,
smooth_tqdm: bool = True,
**tqdm_kwargs: Any,
) → list[R][source]#

Method to execute a function in parallel using multithreading, you should use it:

  • For I/O-bound operations where the GIL is not a bottleneck, such as network requests or disk operations.

  • When the task involves waiting for external resources, such as network responses or user input.

  • For operations that involve a lot of waiting, such as GUI event handling or handling user input.

Parameters:

  • func (Callable | list[Callable]) – Function to execute, or list of functions (one per argument)

  • args (Iterable) – Iterable of arguments to pass to the function(s)

  • use_starmap (bool) – Whether to use starmap or not (Defaults to False): True means the function will be called like func(*args[i]) instead of func(args[i])

  • desc (str) – Description displayed in the progress bar (if not provided no progress bar will be displayed)

  • max_workers (int | float) – Number of workers to use (Defaults to CPU_COUNT), -1 means CPU_COUNT. If float between 0 and 1, it’s treated as a percentage of CPU_COUNT. If negative float between -1 and 0, it’s treated as a percentage of len(args).

  • delay_first_calls (float) – Apply i*delay_first_calls seconds delay to the first “max_workers” calls. For instance with value to 1, the first thread will be delayed by 0 seconds, the second by 1 second, etc. (Defaults to 0): This can be useful to avoid functions being called in the same second.

  • color (str) – Color of the progress bar (Defaults to MAGENTA)

  • bar_format (str) – Format of the progress bar (Defaults to BAR_FORMAT)

  • ascii (bool) – Whether to use ASCII or Unicode characters for the progress bar

  • smooth_tqdm (bool) – Whether to enable smooth progress bar updates by setting miniters and mininterval (Defaults to True)

  • **tqdm_kwargs (Any) – Additional keyword arguments to pass to tqdm

Returns:

Results of the function execution

Return type:

list[object]

Examples

> multithreading(doctest_square, args=[1, 2, 3])
[1, 4, 9]

> multithreading(int.__mul__, [(1,2), (3,4), (5,6)], use_starmap=True)
[2, 12, 30]

> # Using a list of functions (one per argument)
> multithreading([doctest_square, doctest_square, doctest_square], [1, 2, 3])
[1, 4, 9]

> # Will process in parallel with progress bar
> multithreading(doctest_slow, range(10), desc="Threading")
[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]

> # Will process in parallel with progress bar and delay the first threads
> multithreading(
.     doctest_slow,
.     range(10),
.     desc="Threading with delay",
.     max_workers=2,
.     delay_first_calls=0.6
. )
[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
capture_subprocess_output(
args: tuple[CaptureOutput, Callable[[T], R], T],
) → R[source]#

Wrapper function to execute the target function in a subprocess with optional output capture.

Parameters:

  • tuple[CaptureOutput – Tuple containing: CaptureOutput: Capturer object to redirect stdout/stderr Callable: Target function to execute T: Argument to pass to the target function

  • Callable – Tuple containing: CaptureOutput: Capturer object to redirect stdout/stderr Callable: Target function to execute T: Argument to pass to the target function

  • T] – Tuple containing: CaptureOutput: Capturer object to redirect stdout/stderr Callable: Target function to execute T: Argument to pass to the target function