dragon.native.pool

The Dragon native pool manages a pool of child processes that can be used to run python callables.

Functions

`get_logs`(name)
`mapstar`(args)
`rebuild_exc`(exc, tb)

Classes

`ApplyResult`	Returned by `apply_async` and has `get` method to retrieve result.
`AsyncResult`	alias of `ApplyResult`
`ExceptionWithTraceback`
`MapResult`
`Pool`	A Dragon native Pool relying on native Process Group and Queues

Exceptions

`MaybeEncodingError`	Wraps possible unpickleable errors, so they can be safely sent through the socket.
`RemoteTraceback`

get_logs(name)

exception RemoteTraceback

Bases: Exception

__init__(tb)

class ExceptionWithTraceback

Bases: object

__init__(exc, tb)

rebuild_exc(exc, tb)

exception MaybeEncodingError

Bases: Exception

Wraps possible unpickleable errors, so they can be safely sent through the socket.

__init__(exc, value)

class ApplyResult

Bases: object

Returned by apply_async and has get method to retrieve result.

__init__(pool: Pool, callback: callable, error_callback: callable)

Initializes a result

Parameters:

pool (dragon.native.pool.Pool) – the pool where work has been submitted
callback (callable) – function called on returned result
error_callback (callable) – function called on returned result if an error was raised

ready() → bool

Checks whether the result is ready.

Returns:: returns True if the result is ready or False if it isn’t
Return type:: bool

successful() → bool

Checks if the result is ready and returns whether the function call was successful.

Raises:: ValueError – raised if result is not ready
Returns:: returns True if function call was successful
Return type:: bool

wait(timeout: float | None = None) → None

Waits on event for result to be ready

Parameters:: timeout (float, optional) – timeout indicating how long to wait for result to be ready, defaults to None

get(timeout: float | None = None) → Any

Retrieves returned values of work submitted to pool

Parameters:

timeout (float, optional) – timeout for getting result, defaults to None

Raises:

TimeoutError – raised if result is not ready in specified timeout
self._value – raises error if returned work was unsuccessful

Returns:

value returned by func(*args, **kwds)

Return type:

Any

AsyncResult: alias of ApplyResult

class MapResult

Bases: ApplyResult

__init__(pool: Pool, chunksize: int, length: int, callback: callable, error_callback: callable)

Initialization method

Parameters:

pool (dragon.native.pool.Pool) – the pool where work is submitted
chunksize (int) – size that work is supposed to be broken up into
length (int) – number of items in iterable
callback (callable) – function to be called upon return of result
error_callback (callable) – function to be called upon return of result if an error was raised.

mapstar(args)

class Pool

Bases: object

A Dragon native Pool relying on native Process Group and Queues

The interface resembles the Python Multiprocessing.Pool interface and focuses on the most generic functionality. Here we directly support the asynchronous API. The synchronous version of these calls are supported by calling get on the objects returned from the asynchronous functions. By using a Dragon native Process Group to coordinate the worker processes this implementation addresses scalability limitations with the patched base implementation of Multiprocessing.Pool.

At this time, signal.SIGUSR2 is used to signal workers to shutdown. Users should not use or register a handler for this signal if they want to use close to shutdown the pool. Using close guarantees that all work submitted to the pool is finished before the signal is sent while terminate sends the signal immediately. If workers don’t immediately exit, terminate will send escalating signals (SIGINT, SIGTERM, then SIGKILL) and wait patience amount of time for processes to exit before sending the next signal. The user is expected to call join following both of these calls. If join is not called, zombie processes may be left and leave the runtime in a corrupted state.

Init method

Parameters:

processes (int, optional) – number of worker processes, defaults to None
initializer (callable, optional) – initializer function, defaults to None
initargs (tuple, optional) – arguments for initializer function, defaults to ()
maxtasksperchild (int, optional) – maximum tasks each worker will perform, defaults to None
policy (dragon.infrastructure.policy.Policy or list of dragon.infrastructure.policy.Policy) – determines the placement and resources of processes. If a list of policies is given that the number of processes_per_policy must be specified.
processes_per_policy (int) – determines the number of processes to be placed with a specific policy if a list of policies is provided

Raises:

ValueError – raised if number of worker processes is less than 1

terminate(patience: float = 60.0) → None

This sets the threading event immediately and sends signal.SIGINT, signal.SIGTERM, and signal.SIGKILL successively with patience time between the sending of each signal until all processes have exited. Calling this method before blocking on results may lead to some work being undone. If all work should be done before stopping the workers, close should be used.

Parameters:: patience (float, optional) – timeout to wait for processes to join after each signal, defaults to None

kill() → None: This sends the signal.SIGKILL and sets the threading event immediately. This is the most dangerous way to shutdown pool workers and will likely leave the runtime in a corrupted state. Calling this method before blocking on results may lead to some work being undone. If all work should be done before stopping the workers, close should be used.

close() → None

This method starts a thread that waits for all submitted jobs to finish. This thread then sends signal.SIGUSR2 to the process group and sets the threading event to close the handle_results thread. Waiting and then sending the signals within the thread allows close to return without all work needing to be done.

Raises:: ValueError – raised if close or terminate have been previously called

join(patience: float | None = None) → None

Waits for all workers to return. The user must have called close or terminate prior to calling join. By default this blocks indefinitely for processes to exit. If a patience is given, then once the patience has passed signal.SIGINT, signal.SIGTERM, and signal.SIGKILL will be sent successively with patience time between the sending of each signal. It is recommended that if a patience is set to a value other than None that a user assume the runtime is corrupted after the join completes.

Parameters:

patience (float, optional) – timeout to wait for processes to join after each signal, defaults to None

Raises:

ValueError – raised if close or terminate were not previously called
ValueError – raised if join has already been called and the process group is closed

apply_async(func: callable, args: tuple = (), kwds: dict = {}, callback: callable | None = None, error_callback: callable | None = None) → ApplyResult

Equivalent to calling func(*args, **kwds) in a non-blocking way. A result is immediately returned and then updated when func(*args, **kwds) has completed.

Parameters:

func (callable) – user function to be called by worker function
args (tuple, optional) – input args to func, defaults to ()
kwds (dict, optional) – input kwds to func, defaults to {}
callback (callable, optional) – user provided callback function, defaults to None
error_callback (callable, optional) – user provided error callback function, defaults to None

Raises:

ValueError – raised if pool has already been closed or terminated

Returns:

A result that has a get method to retrieve result of func(*args, **kwds)

Return type:

ApplyResult

map_async(func: callable, iterable: Iterable, chunksize: int | None = None, callback: callable | None = None, error_callback: callable | None = None) → MapResult

Apply func to each element in iterable. The results are collected in a list that is returned immediately. It is equivalent to map if the get method is called immediately on the returned result.

Parameters:

func (callable) – user provided function to call on elements of iterable
iterable (iterable) – input args to func
chunksize (int, optional) – size of work elements to be submitted to input work queue, defaults to None
callback (callable, optional) – user provided callback function, defaults to None
error_callback (callable, optional) – user provided error callback function, defaults to None

Returns:

A result that has a get method that returns an iterable of the output from applying func to each element of iterable.

Return type:

MapResult