multiprocessing#

max_workers(hint: int = -1) → int#
get_context(method: Literal['fork', 'forkserver', 'spawn'] | None = None) → BaseContext#

Return a multiprocessing context.

  • If method is None: return the process-wide default context (whatever was set by initialize()).

  • If method is provided: return that specific context.

This is the safest way to avoid touching private symbols like multiprocessing.context._default_context.

default_start_method() → str#

Best-effort reporting of the active default start method.

recommended_start_method() → Literal['fork', 'forkserver', 'spawn']#

Mirror your initialize() logic but just return the recommendation (does not mutate global state).

initialize() → None#
class SimpleQueue(ctx: BaseContext | None = None)#

Bases: SimpleQueue

class Queue(maxsize: int = 0, ctx: BaseContext | None = None)#

Bases: Queue

class ErrorFromWorker(exc_cls, exc, tb)#

Bases: object

Wrapper class to report an error from a worker process

property stacktrace#
class Task(func: Callable)#

Bases: object

Wrapped task that trap every Exception and return it as an ErrorFromWorker object.

We are using a wrapper class instead of a decorator since the class is pickleable, while a decorator with an inner closure is not.

raise_if_errors(*results, debug=False)#

Analyze results from worker Processes to search for ErrorFromWorker objects. If found print all of them and raise an exception.

Parameters:

  • *results – results from worker processes

  • debug – if True show complete stacktraces

Raises:

RuntimeError – if ErrorFromWorker objects are in the results

pool(*args, **kwargs)#

Context manager to start and terminate a pool of processes

Arguments are forwarded to the multiprocessing.Pool.__init__ method.

num_processes(max_processes: int | None = None, allow_oversubscription: bool = False) → int#

Return the number of processes in a pool.

Currently the function return the minimum between the maximum number of processes and the cpus available.

When a maximum number of processes is not specified return the cpus available.

Parameters:

max_processes (int or None) – maximum number of processes allowed

map(func: Callable, args: Sequence, processes: int | None = None, debug: bool = False, initializer: Callable[[...], Any] | None = None, initargs: Iterable[Any] = ()) → Any#

Map a func to the list of arguments, return the list of results.

Parameters:

  • func – user defined task object

  • args – iterator of arguments for the task

  • processes – maximum number of processes allowed

  • debug – if False, raise an exception containing just the error messages from workers, if True an exception with complete stacktraces

Raises:

RuntimeError – if any error occurred in the worker processes

starmap(func: Callable, args: Sequence, processes: int | None = None, debug: bool = False, initializer: Callable[[...], Any] | None = None, initargs: Iterable[Any] = ()) → Any#

Map a func to the list of arguments, return the list of results.

Parameters:

  • func – user defined task object

  • args – list of arguments for the task

  • processes – maximum number of processes allowed

  • debug – if False, raise an exception containing just the error messages from workers, if True an exception with complete stacktraces

Raises:

RuntimeError – if any error occurred in the worker processes

parent_process() → BaseProcess | None#
class FSQueue(root: Path)#

Bases: object

A file-system-backed queue with caching.

Items are pickled to disk, multiple processes can safely put items, and a listener can consume them in FIFO order.

put(obj: Any) → None#

Serialize and atomically store an object in the queue.

empty() → bool#
get() → Any#

Get the oldest item from the queue.

Raises:

Empty – if the queue is empty.

drain() → list[Any]#

Yield items from the queue until empty.