-
-
Notifications
You must be signed in to change notification settings - Fork 1.6k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
ehn: use qt-async-thread to run coro in parallel thread safe
- Loading branch information
Showing
3 changed files
with
169 additions
and
70 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,153 @@ | ||
| import asyncio | ||
| import atexit | ||
| import functools | ||
| import threading | ||
| from collections.abc import Coroutine | ||
| from typing import Callable | ||
| from typing import Any | ||
|
|
||
| from qt_async_threads import QtAsyncRunner | ||
|
|
||
|
|
||
| class AsSync: | ||
| """Decorator to convert a coroutine to a sync function. | ||
| Helper class to facilitate the conversion of coroutines to sync functions | ||
| or to run a coroutine as a sync function without the need to call the event | ||
| loop method. | ||
| Usage | ||
| ------ | ||
| As a decorator: | ||
| ``` | ||
| @AsSync | ||
| async def my_coroutine(): | ||
| pass | ||
| my_coroutine() | ||
| ``` | ||
| As a class wrapper: | ||
| ``` | ||
| sync_coroutine = AsSync(my_coroutine) | ||
| sync_coroutine() | ||
| ``` | ||
| """ | ||
| def __init__(self, coro, loop=None): | ||
| """Initialize the decorator. | ||
| Parameters | ||
| ---------- | ||
| coro : coroutine | ||
| The coroutine to be wrapped. | ||
| loop : asyncio.AbstractEventLoop, optional | ||
| The event loop to be used, by default get the current event loop. | ||
| """ | ||
| self.__coro = coro | ||
| self.__loop = loop or asyncio.get_event_loop() | ||
| functools.update_wrapper(self, coro) | ||
|
|
||
| def __call__(self, *args, **kwargs): | ||
| return self.__loop.run_until_complete(self.__coro(*args, **kwargs)) | ||
|
|
||
| def __get__(self, instance, owner): | ||
| if instance is None: | ||
| return self | ||
| else: | ||
| bound_method = self.__coro.__get__(instance, owner) | ||
| return functools.partial(self.__class__(bound_method, self.__loop)) | ||
|
|
||
|
|
||
| class SpyderQAsyncRunner(QtAsyncRunner): | ||
| """Reimplement QtAsyncRunner as a singleton.""" | ||
|
|
||
| _instance = None | ||
| _rlock = threading.RLock() | ||
| __inside_instance = False | ||
|
|
||
| @classmethod | ||
| def instance(cls, *args, **kwargs): | ||
| """Get *the* class instance. | ||
| Return the instance of the class. If it did not exist yet, create it | ||
| by calling the "constructor" with whatever arguments and keyword | ||
| arguments provided. | ||
| Returns | ||
| ------- | ||
| instance(object) | ||
| Class Singleton Instance | ||
| """ | ||
| if cls._instance is not None: | ||
| return cls._instance | ||
| with cls._rlock: | ||
| # Re-check, perhaps it was created in the meantime... | ||
| if cls._instance is None: | ||
| cls.__inside_instance = True | ||
| try: | ||
| cls._instance = cls(*args, **kwargs) | ||
| finally: | ||
| cls.__inside_instance = False | ||
| return cls._instance | ||
|
|
||
| def __new__(cls, *args, **kwargs): | ||
| """Class constructor. | ||
| Ensures that this class isn't created without | ||
| the ``instance`` class method. | ||
| Returns | ||
| ------- | ||
| object | ||
| Class instance | ||
| Raises | ||
| ------ | ||
| RuntimeError | ||
| Exception when not called from the ``instance`` class method. | ||
| """ | ||
| if cls._instance is None: | ||
| with cls._rlock: | ||
| if cls._instance is None and cls.__inside_instance: | ||
| return super().__new__(cls) | ||
|
|
||
| raise RuntimeError( | ||
| f"Attempt to create a {cls.__qualname__} instance outside of instance()" | ||
| ) | ||
|
|
||
| async def run( | ||
| self, func: Callable | Coroutine, *args: Any, **kwargs: Any | ||
| ) -> Any: | ||
| """ | ||
| Updated to handle both functions and coroutines. If `func` is a coroutine, | ||
| it is scheduled to run in the thread pool. | ||
| """ | ||
| if asyncio.iscoroutinefunction(func) or isinstance(func, Coroutine): | ||
| # Wrap the coroutine in a function that sets up an event loop | ||
| async_func = func if asyncio.iscoroutinefunction(func) else functools.partial(lambda x: x, func) | ||
| loop = asyncio.get_event_loop() | ||
| def run_coro_in_thread(): | ||
| coroutine = async_func(*args, **kwargs) | ||
| return asyncio.run_coroutine_threadsafe(coroutine, loop) | ||
|
|
||
| func_to_run = run_coro_in_thread | ||
| else: | ||
| func_to_run = functools.partial(func, *args, **kwargs) | ||
|
|
||
| async for result in self.run_parallel([func_to_run]): | ||
| return result | ||
| assert False, "should never be reached" | ||
|
|
||
| @atexit.register | ||
| def __atexit(): | ||
| SpyderQAsyncRunner.instance().close() | ||
|
|
||
| @classmethod | ||
| def run_async(cls, coro): | ||
| """Decorator to run a coroutine asynchronously.""" | ||
| @functools.wraps(coro) | ||
| def wrapper(*args, **kwargs): | ||
| return cls.instance().start_coroutine(cls.instance().run(coro, *args, **kwargs)) | ||
| return wrapper | ||
|
|
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters