Skip to content

AsyncIOProcessingService class

Bases: ProcessingService

A processing service that utilizes the AsyncIO framework to handle the execution of calls.

Notes:

  • Synchronous callbacks are executed in a blocking manner if there is no active asyncio loop or if force_async is set to False. If force_async is set to True and an asyncio loop is active, these synchronous callbacks are converted to asynchronous using the asyncio.to_thread() function and are processed as background tasks within the existing asyncio loop.

  • Asynchronous callbacks operate independently of the force_async parameter, as they are inherently asynchronous. However, their execution depends on the presence of an active asyncio loop. If a running asyncio loop exists, callbacks are scheduled and executed as background tasks within that loop. If no loop is active, a new asyncio loop is created and automatically closed once the callback is complete.

  • When enforce_submission_order is set to True, new submissions are managed by a queue and processed sequentially within an asyncio loop. If no active asyncio loop is present, a new loop is initiated to handle the queue processing, and it remains open until all callbacks have been executed and the queue is empty, at which point it will be closed automatically. During queue processing, callbacks are executed as before, with the difference that they are always run within an existing asyncio loop and awaited (for async callbacks) instead of being run as background tasks.

  • All active tasks from all instances can be retrieved through the all_tasks() method.

Source code in pyventus/core/processing/asyncio/asyncio_processing_service.py 
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 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
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
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
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
class AsyncIOProcessingService(ProcessingService):
    """
    A processing service that utilizes the `AsyncIO` framework to handle the execution of calls.

    **Notes:**

    -   Synchronous callbacks are executed in a blocking manner if there is no active asyncio loop or
        if `force_async` is set to `False`. If `force_async` is set to `True` and an asyncio loop is
        active, these synchronous callbacks are converted to asynchronous using the `asyncio.to_thread()`
        function and are processed as background tasks within the existing asyncio loop.

    -   Asynchronous callbacks operate independently of the `force_async` parameter, as they are
        inherently asynchronous. However, their execution depends on the presence of an active asyncio
        loop. If a running asyncio loop exists, callbacks are scheduled and executed as background tasks
        within that loop. If no loop is active, a new asyncio loop is created and automatically closed
        once the callback is complete.

    -   When `enforce_submission_order` is set to `True`, new submissions are managed by a queue and
        processed sequentially within an asyncio loop. If no active asyncio loop is present, a new loop
        is initiated to handle the queue processing, and it remains open until all callbacks have been
        executed and the queue is empty, at which point it will be closed automatically. During queue
        processing, callbacks are executed as before, with the difference that they are always run
        within an existing asyncio loop and awaited (for async callbacks) instead of being run as
        background tasks.

    -   All active tasks from all instances can be retrieved through the `all_tasks()` method.
    """

    class _AsyncIOSubmission(NamedTuple):
        """A named tuple for storing relevant information related to an AsyncIO submission."""

        callback: ProcessingServiceCallbackType
        args: tuple[Any, ...]
        kwargs: dict[str, Any]

    __global_thread_lock: Lock = Lock()
    """A global lock for synchronizing access to shared class-level resources."""

    __all_tasks: set[Task[Any]] = set()
    """A set of active background tasks managed across all instances."""

    @staticmethod
    def is_loop_running() -> bool:
        """
        Determine whether there is currently an active `AsyncIO` event loop.

        :return: `True` if an event loop is running; `False` otherwise.
        """
        try:
            get_running_loop()
            return True
        except RuntimeError:
            return False

    @classmethod
    def all_tasks(cls) -> set[Task[Any]]:
        """
        Retrieve all active tasks from all instances.

        :return: A set of currently active tasks across all instances.
        """
        with cls.__global_thread_lock:
            return cls.__all_tasks.copy()

    # Attributes for the AsyncIOProcessingService
    __slots__ = ("__thread_lock", "__tasks", "__force_async", "__is_submission_queue_busy", "__submission_queue")

    def __init__(self, force_async: bool = False, enforce_submission_order: bool = False) -> None:
        """
        Initialize an instance of `AsyncIOProcessingService`.

        :param force_async: A boolean flag that determines whether to force all submitted
            callbacks to run asynchronously.
        :param enforce_submission_order: A boolean flag that determines whether to enforce the order
            of execution for submissions based on their arrival (FIFO: First In, First Out).
        :return: None
        """
        # Ensure that the 'force_async' and 'enforce_submission_order' parameters are of boolean type.
        if not isinstance(force_async, bool):
            raise PyventusException("The 'force_async' argument must be a boolean value.")
        if not isinstance(enforce_submission_order, bool):
            raise PyventusException("The 'enforce_submission_order' argument must be a boolean value.")

        # Create a thread lock to manage concurrent access to shared resources.
        self.__thread_lock: Lock = Lock()

        # Initialize a set to keep track of active background tasks.
        self.__tasks: set[Task[Any]] = set()

        # Store the value of the force_async parameter.
        self.__force_async: bool = force_async

        # Define a flag to indicate whether the submission queue is currently being processed.
        self.__is_submission_queue_busy: bool = False

        # Create a FIFO queue to maintain the order of submissions if execution order is required.
        self.__submission_queue: deque[AsyncIOProcessingService._AsyncIOSubmission] | None = (
            deque[AsyncIOProcessingService._AsyncIOSubmission]() if enforce_submission_order else None
        )

    def __repr__(self) -> str:
        """
        Retrieve a string representation of the instance.

        :return: A string representation of the instance.
        """
        return formatted_repr(
            instance=self,
            info=attributes_repr(
                thread_lock=self.__thread_lock,
                tasks=self.__tasks,
                force_async=self.__force_async,
                is_submission_queue_busy=self.__is_submission_queue_busy,
                submission_queue=self.__submission_queue,
            ),
        )

    @property
    def _thread_lock(self) -> Lock:
        """
        Retrieve the thread lock instance.

        :return: The thread lock instance used to ensure thread-safe operations.
        """
        return self.__thread_lock

    @property
    def _tasks(self) -> set[Task[Any]]:
        """
        Retrieve the set of currently active background tasks.

        :return: A set containing the active background tasks.
        """
        return self.__tasks

    @property
    def _is_submission_queue_busy(self) -> bool:
        """
        Determine whether the submission queue is busy.

        :return: `True` if the submission queue is busy; otherwise, `False`.
        """
        return self.__is_submission_queue_busy

    @property
    def _submission_queue(self) -> deque[_AsyncIOSubmission] | None:
        """
        Retrieve the submission queue used for enforcing submission order.

        :return: The submission queue used for enforcing submission order,
            or None if submission order enforcement is not enabled.
        """
        return self.__submission_queue

    @property
    def task_count(self) -> int:
        """
        Retrieve the count of currently active background tasks.

        :return: The number of active background tasks.
        """
        with self.__thread_lock:
            return len(self.__tasks)

    @property
    def force_async(self) -> bool:
        """
        Determine whether all submitted callbacks are forced to run asynchronously.

        :return: `True` if all submitted callbacks are forced to run asynchronously; otherwise, `False`.
        """
        return self.__force_async

    @property
    def enforce_submission_order(self) -> bool:
        """
        Determine whether submission order enforcement is enabled.

        :return: `True` if submission order enforcement is enabled; otherwise, `False`.
        """
        return self.__submission_queue is not None

    def _remove_task(self, task: Task[Any]) -> None:
        """
        Remove a background task from the local and global tracking sets.

        :param task: The background task to be removed from the local and global sets.
        :return: None.
        """
        # Get the class type to access class-level variables.
        cls: type[AsyncIOProcessingService] = type(self)

        # Acquire locks for thread safety and remove the task.
        with self.__thread_lock, cls.__global_thread_lock:
            cls.__all_tasks.discard(task)
            self.__tasks.discard(task)

    def _add_task(self, task: Task[Any]) -> None:
        """
        Add a background task to the local and global tracking sets.

        :param task: The background task to be added to the local and global sets.
        :return: None.
        """
        # Get the class type to access class-level variables.
        cls: type[AsyncIOProcessingService] = type(self)

        # Acquire locks for thread safety and add the task.
        with self.__thread_lock, cls.__global_thread_lock:
            cls.__all_tasks.add(task)
            self.__tasks.add(task)

    def _schedule_task(self, coroutine: Coroutine[Any, Any, Any]) -> None:
        """
        Schedule a coroutine as a background task and track its execution until completion.

        :param coroutine: The coroutine to be scheduled as a background task.
        :return: None.
        """
        # Create and schedule the coroutine as a background Task.
        task: Task[Any] = create_task(coroutine)

        # Register the cleanup callback to remove the task
        # from the local and global tracking sets upon completion.
        task.add_done_callback(self._remove_task)

        # Add the task to both the local and global tracking sets.
        self._add_task(task)

    async def _process_submission_queue(self) -> None:
        """
        Process each AsyncIO submission in the queue in FIFO order.

        :return: None.
        """
        # Initialize a variable to hold the current submission being processed.
        submission: AsyncIOProcessingService._AsyncIOSubmission | None = None

        # Pop and process submissions with thread
        # safety until the queue is empty.
        while True:
            with self.__thread_lock:
                if not self.__submission_queue:
                    self.__is_submission_queue_busy = False  # Mark processing as complete.
                    break

                # Retrieve the next submission from the front of the queue.
                submission = self.__submission_queue.popleft()

            # Execute the submission's callback accordingly.
            if is_callable_async(submission.callback):
                await submission.callback(*submission.args, **submission.kwargs)
            elif self.__force_async:
                await to_thread(submission.callback, *submission.args, **submission.kwargs)
            else:
                submission.callback(*submission.args, **submission.kwargs)

    @override
    def submit(self, callback: ProcessingServiceCallbackType, *args: Any, **kwargs: Any) -> None:
        # Check if there is an active asyncio event loop.
        is_loop_running: bool = AsyncIOProcessingService.is_loop_running()

        # Process the callback based on whether
        # submission order enforcement is enabled.
        if not self.enforce_submission_order:
            # If submission order enforcement is not required, execute the
            # callback according to its definition and the current context.
            if is_callable_async(callback):
                if is_loop_running:
                    self._schedule_task(callback(*args, **kwargs))
                else:
                    run(callback(*args, **kwargs))
            elif self.__force_async and is_loop_running:
                self._schedule_task(to_thread(callback, *args, **kwargs))
            else:
                callback(*args, **kwargs)
        else:
            # If submission order enforcement is enabled, manage
            # the callback using the submission queue.
            with self.__thread_lock:
                was_submission_queue_busy: bool = self.__is_submission_queue_busy
                self.__is_submission_queue_busy = True
                self.__submission_queue.append(  # type: ignore[union-attr]
                    AsyncIOProcessingService._AsyncIOSubmission(
                        callback=callback,
                        args=args,
                        kwargs=kwargs,
                    )
                )

            # Start processing the queue if it wasn't already active.
            if not was_submission_queue_busy:
                if is_loop_running:
                    self._schedule_task(self._process_submission_queue())
                else:
                    run(self._process_submission_queue())

    async def wait_for_tasks(self) -> None:
        """
        Wait for all background tasks associated with the current service to complete.

        This method ensures that any ongoing tasks are finished before proceeding.

        :return: None.
        """
        # Process all active background tasks until none remain.
        while True:
            with self.__thread_lock:
                # Exit if there are no active tasks.
                if not self.__tasks:
                    break

                # Copy current tasks and clear the registry.
                tasks: set[Task[Any]] = self.__tasks.copy()
                self.__tasks.clear()

            # Await the completion of all tasks.
            await gather(*tasks)

Attributes

task_count property 

task_count: int

Retrieve the count of currently active background tasks.

RETURNS DESCRIPTION
int

The number of active background tasks.

force_async property 

force_async: bool

Determine whether all submitted callbacks are forced to run asynchronously.

RETURNS DESCRIPTION
bool

True if all submitted callbacks are forced to run asynchronously; otherwise, False.

enforce_submission_order property 

enforce_submission_order: bool

Determine whether submission order enforcement is enabled.

RETURNS DESCRIPTION
bool

True if submission order enforcement is enabled; otherwise, False.

Functions

is_loop_running staticmethod 

is_loop_running() -> bool

Determine whether there is currently an active AsyncIO event loop.

RETURNS DESCRIPTION
bool

True if an event loop is running; False otherwise.
Source code in pyventus/core/processing/asyncio/asyncio_processing_service.py 
55
56
57
58
59
60
61
62
63
64
65
66
@staticmethod
def is_loop_running() -> bool:
    """
    Determine whether there is currently an active `AsyncIO` event loop.

    :return: `True` if an event loop is running; `False` otherwise.
    """
    try:
        get_running_loop()
        return True
    except RuntimeError:
        return False

all_tasks classmethod 

all_tasks() -> set[Task[Any]]

Retrieve all active tasks from all instances.

RETURNS DESCRIPTION
set[Task[Any]]

A set of currently active tasks across all instances.
Source code in pyventus/core/processing/asyncio/asyncio_processing_service.py 
68
69
70
71
72
73
74
75
76
@classmethod
def all_tasks(cls) -> set[Task[Any]]:
    """
    Retrieve all active tasks from all instances.

    :return: A set of currently active tasks across all instances.
    """
    with cls.__global_thread_lock:
        return cls.__all_tasks.copy()

__init__ 

__init__(force_async: bool = False, enforce_submission_order: bool = False) -> None

Initialize an instance of AsyncIOProcessingService.

PARAMETER DESCRIPTION
force_async

A boolean flag that determines whether to force all submitted callbacks to run asynchronously.

TYPE: bool DEFAULT: False

enforce_submission_order

A boolean flag that determines whether to enforce the order of execution for submissions based on their arrival (FIFO: First In, First Out).

TYPE: bool DEFAULT: False

RETURNS DESCRIPTION
None

None
Source code in pyventus/core/processing/asyncio/asyncio_processing_service.py 
 81
 82
 83
 84
 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
def __init__(self, force_async: bool = False, enforce_submission_order: bool = False) -> None:
    """
    Initialize an instance of `AsyncIOProcessingService`.

    :param force_async: A boolean flag that determines whether to force all submitted
        callbacks to run asynchronously.
    :param enforce_submission_order: A boolean flag that determines whether to enforce the order
        of execution for submissions based on their arrival (FIFO: First In, First Out).
    :return: None
    """
    # Ensure that the 'force_async' and 'enforce_submission_order' parameters are of boolean type.
    if not isinstance(force_async, bool):
        raise PyventusException("The 'force_async' argument must be a boolean value.")
    if not isinstance(enforce_submission_order, bool):
        raise PyventusException("The 'enforce_submission_order' argument must be a boolean value.")

    # Create a thread lock to manage concurrent access to shared resources.
    self.__thread_lock: Lock = Lock()

    # Initialize a set to keep track of active background tasks.
    self.__tasks: set[Task[Any]] = set()

    # Store the value of the force_async parameter.
    self.__force_async: bool = force_async

    # Define a flag to indicate whether the submission queue is currently being processed.
    self.__is_submission_queue_busy: bool = False

    # Create a FIFO queue to maintain the order of submissions if execution order is required.
    self.__submission_queue: deque[AsyncIOProcessingService._AsyncIOSubmission] | None = (
        deque[AsyncIOProcessingService._AsyncIOSubmission]() if enforce_submission_order else None
    )

submit 

submit(callback: ProcessingServiceCallbackType, *args: Any, **kwargs: Any) -> None
Source code in pyventus/core/processing/asyncio/asyncio_processing_service.py 
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
@override
def submit(self, callback: ProcessingServiceCallbackType, *args: Any, **kwargs: Any) -> None:
    # Check if there is an active asyncio event loop.
    is_loop_running: bool = AsyncIOProcessingService.is_loop_running()

    # Process the callback based on whether
    # submission order enforcement is enabled.
    if not self.enforce_submission_order:
        # If submission order enforcement is not required, execute the
        # callback according to its definition and the current context.
        if is_callable_async(callback):
            if is_loop_running:
                self._schedule_task(callback(*args, **kwargs))
            else:
                run(callback(*args, **kwargs))
        elif self.__force_async and is_loop_running:
            self._schedule_task(to_thread(callback, *args, **kwargs))
        else:
            callback(*args, **kwargs)
    else:
        # If submission order enforcement is enabled, manage
        # the callback using the submission queue.
        with self.__thread_lock:
            was_submission_queue_busy: bool = self.__is_submission_queue_busy
            self.__is_submission_queue_busy = True
            self.__submission_queue.append(  # type: ignore[union-attr]
                AsyncIOProcessingService._AsyncIOSubmission(
                    callback=callback,
                    args=args,
                    kwargs=kwargs,
                )
            )

        # Start processing the queue if it wasn't already active.
        if not was_submission_queue_busy:
            if is_loop_running:
                self._schedule_task(self._process_submission_queue())
            else:
                run(self._process_submission_queue())

wait_for_tasks async 

wait_for_tasks() -> None

Wait for all background tasks associated with the current service to complete.

This method ensures that any ongoing tasks are finished before proceeding.

RETURNS DESCRIPTION
None

None.
Source code in pyventus/core/processing/asyncio/asyncio_processing_service.py 
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
async def wait_for_tasks(self) -> None:
    """
    Wait for all background tasks associated with the current service to complete.

    This method ensures that any ongoing tasks are finished before proceeding.

    :return: None.
    """
    # Process all active background tasks until none remain.
    while True:
        with self.__thread_lock:
            # Exit if there are no active tasks.
            if not self.__tasks:
                break

            # Copy current tasks and clear the registry.
            tasks: set[Task[Any]] = self.__tasks.copy()
            self.__tasks.clear()

        # Await the completion of all tasks.
        await gather(*tasks)