aea.helpers.multiple_executor

This module contains the helpers to run multiple stoppable tasks in different modes: async, threaded, multiprocess .

ExecutorExceptionPolicies Objects

class ExecutorExceptionPolicies(Enum)

Runner exception policy modes.

AbstractExecutorTask Objects

class AbstractExecutorTask(ABC)

Abstract task class to create Task classes.

__init__

 | __init__() -> None

Init task.

future

 | @property
 | future() -> Optional[TaskAwaitable]

Return awaitable to get result of task execution.

future

 | @future.setter
 | future(future: TaskAwaitable) -> None

Set awaitable to get result of task execution.

start

 | @abstractmethod
 | start() -> Tuple[Callable, Sequence[Any]]

Implement start task function here.

stop

 | @abstractmethod
 | stop() -> None

Implement stop task function here.

create_async_task

 | @abstractmethod
 | create_async_task(loop: AbstractEventLoop) -> TaskAwaitable

Create asyncio task for task run in asyncio loop.

Arguments:

• loop: the event loop

Returns:

task to run in asyncio loop.

id

 | @property
 | id() -> Any

Return task id.

failed

 | @property
 | failed() -> bool

Return was exception failed or not.

If it's running it's not failed.

Returns:

bool

AbstractMultiprocessExecutorTask Objects

class AbstractMultiprocessExecutorTask(AbstractExecutorTask)

Task for multiprocess executor.

start

 | @abstractmethod
 | start() -> Tuple[Callable, Sequence[Any]]

Return function and arguments to call within subprocess.

create_async_task

 | create_async_task(loop: AbstractEventLoop) -> TaskAwaitable

Create asyncio task for task run in asyncio loop.

Raise error, cause async mode is not supported, cause this task for multiprocess executor only.

Arguments:

• loop: the event loop

Raises:

• ValueError: async task construction not possible

AbstractMultipleExecutor Objects

class AbstractMultipleExecutor(ABC)

Abstract class to create multiple executors classes.

__init__

 | __init__(tasks: Sequence[AbstractExecutorTask], task_fail_policy: ExecutorExceptionPolicies = ExecutorExceptionPolicies.propagate) -> None

Init executor.

Arguments:

• tasks: sequence of AbstractExecutorTask instances to run.
• task_fail_policy: the exception policy of all the tasks

is_running

 | @property
 | is_running() -> bool

Return running state of the executor.

start

 | start() -> None

Start tasks.

stop

 | stop() -> None

Stop tasks.

num_failed

 | @property
 | num_failed() -> int

Return number of failed tasks.

failed_tasks

 | @property
 | failed_tasks() -> Sequence[AbstractExecutorTask]

Return sequence failed tasks.

not_failed_tasks

 | @property
 | not_failed_tasks() -> Sequence[AbstractExecutorTask]

Return sequence successful tasks.

ThreadExecutor Objects

class ThreadExecutor(AbstractMultipleExecutor)

Thread based executor to run multiple agents in threads.

ProcessExecutor Objects

class ProcessExecutor(ThreadExecutor)

Subprocess based executor to run multiple agents in threads.

AsyncExecutor Objects

class AsyncExecutor(AbstractMultipleExecutor)

Thread based executor to run multiple agents in threads.

AbstractMultipleRunner Objects

class AbstractMultipleRunner()

Abstract multiple runner to create classes to launch tasks with selected mode.

__init__

 | __init__(mode: str, fail_policy: ExecutorExceptionPolicies = ExecutorExceptionPolicies.propagate) -> None

Init with selected executor mode.

Arguments:

• mode: one of supported executor modes
• fail_policy: one of ExecutorExceptionPolicies to be used with Executor

is_running

 | @property
 | is_running() -> bool

Return state of the executor.

start

 | start(threaded: bool = False) -> None

Run agents.

Arguments:

• threaded: run in dedicated thread without blocking current thread.

stop

 | stop(timeout: Optional[float] = None) -> None

Stop agents.

Arguments:

• timeout: timeout in seconds to wait thread stopped, only if started in thread mode.

num_failed

 | @property
 | num_failed() -> int

Return number of failed tasks.

failed

 | @property
 | failed() -> Sequence[Task]

Return sequence failed tasks.

not_failed

 | @property
 | not_failed() -> Sequence[Task]

Return sequence successful tasks.

try_join_thread

 | try_join_thread() -> None

Try to join thread if running in thread mode.