Skip to content

academy.executor

EventLoopExecutor 

EventLoopExecutor(factory: ExchangeFactory[Any])

Bases: Executor

Modified Executor that packs multiple agents into one event loop.

This executor spends a single inner worker on a hidden host agent, then runs every subsequent Agent submission onto that host's own event loop.

Parameters:

Source code in academy/executor.py 
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
def __init__(
    self,
    factory: ae.ExchangeFactory[Any],
):
    self._loop: asyncio.AbstractEventLoop | None = None
    self._loop_ready: threading.Event = threading.Event()
    self._factory = factory
    self._client: ae.UserExchangeClient[Any] | None = None
    self._host: Handle[Any] | None = None
    self._shutdown = False
    self._host_lock: asyncio.Lock = asyncio.Lock()
    self._pending_futures: set[Future[Any]] = set()
    self._host_task: asyncio.Task[None] | None = None

    self._thread: threading.Thread = Thread(target=self._thread_main)
    self._thread.start()
    self._loop_ready.wait()

submit 

submit(
    fn: Callable[..., Any], /, *args: Any, **kwargs: Any
) -> Future[Any]

Run a callable on the host agent's event loop.

Parameters:

  • fn (Callable[..., Any]) –

    Callable to run on the host.

  • *args (Any, default: () ) –

    Positional arguments for submitted function.

  • **kwargs (Any, default: {} ) –

    Keyword arguments for submitted function.

Returns: Future resolving when fn finishes on host.

Source code in academy/executor.py 
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
def submit(
    self,
    fn: Callable[..., Any],
    /,
    *args: Any,
    **kwargs: Any,
) -> Future[Any]:
    """Run a callable on the host agent's event loop.

    Args:
        fn: Callable to run on the host.
        *args: Positional arguments for submitted function.
        **kwargs: Keyword arguments for submitted function.

    Returns: Future resolving when fn finishes on host.
    """
    if self._shutdown:
        raise RuntimeError('Cannot submit after host shutdown')

    if fn is _run_agent_on_worker:
        spec = args[0]
        fn, args, kwargs = _run_agent_async, (spec,), {}

    else:
        raise ValueError(
            'Only functions of the type _run_agent_on_worker '
            'are allowed to be submitted.',
        )

    assert self._loop is not None
    task_future = asyncio.run_coroutine_threadsafe(
        self._submit_async(fn, args, kwargs),
        self._loop,
    )
    self._pending_futures.add(task_future)
    task_future.add_done_callback(self._pending_futures.discard)

    return task_future

shutdown 

shutdown(
    wait: bool = True, cancel_futures: bool = True
) -> None

Shut down the owned thread and the host agent.

Args: wait: Wait for the inner executor to finish before returning. cancel_futures: Cancel pending futures before shutting down.

Source code in academy/executor.py 
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
def shutdown(
    self,
    wait: bool = True,
    cancel_futures: bool = True,
) -> None:
    """Shut down the owned thread and the host agent.

    Args:
    wait: Wait for the inner executor to finish before returning.
    cancel_futures: Cancel pending futures before shutting down.
    """
    self._shutdown = True
    assert self._loop is not None

    if cancel_futures:
        for future in list(self._pending_futures):
            future.cancel()

    if self._host is not None:
        future_host = asyncio.run_coroutine_threadsafe(
            self._host.shutdown(),
            self._loop,
        )
        future_host.result()

    if self._client is not None:
        future_client = asyncio.run_coroutine_threadsafe(
            self._client.close(),
            self._loop,
        )
        future_client.result()

    self._loop.call_soon_threadsafe(self._loop.stop)

    if wait:
        self._thread.join()