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,

process_title: str | 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.
process_title (str | None) – If provided, sets the process title for worker processes.
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

process_title_wrapper( args: tuple[str, int, Callable[[T], R], T], ) → R[source]#

Wrapper function to set the process title before executing the target function.

Parameters:

tuple[str – Tuple containing: str: Process title to set int: Worker index to append to title Callable: Target function to execute T: Argument to pass to the target function
int – Tuple containing: str: Process title to set int: Worker index to append to title Callable: Target function to execute T: Argument to pass to the target function
Callable – Tuple containing: str: Process title to set int: Worker index to append to title Callable: Target function to execute T: Argument to pass to the target function
T] – Tuple containing: str: Process title to set int: Worker index to append to title Callable: Target function to execute T: Argument to pass to the target function