Prześlij zadanie do pętli zdarzeń asyncio

Question

Chciałbym przesłać zadania z wątku do pętli zdarzeń asyncio (podobnie jak run_in_executor, ale na odwrót).

Oto co dokumentacja asyncio mówi o concurrency and multithreading:

Aby zaplanować zwrotnego z innego wątku, należy stosować metodę BaseEventLoop.call_soon_threadsafe(). przykład zaplanować współprogram z innego wątku: loop.call_soon_threadsafe(asyncio.async, coro_func())

Działa to dobrze, ale rezultatem współprogram stracone.

Zamiast tego można użyć funkcji dodającej wykonane wywołanie zwrotne do przyszłości zwrócone przez async (lub ensure_future), aby wątek mógł uzyskać dostęp do wyniku przez concurrent.futures.Future.

Czy istnieje szczególny powód, dla którego taka funkcja nie jest zaimplementowana w standardowej bibliotece? A może brakowało mi prostszego sposobu, aby to osiągnąć?

Answer 1

Moja prośba została zrealizowana, a funkcja run_coroutine_threadsafe została zaimplementowana jako run_coroutine_threadsafe. Została zaimplementowana funkcja run_coroutine_threadsafehere.

Przykład:

def target(loop, timeout=None): 
    future = asyncio.run_coroutine_threadsafe(add(1, b=2), loop) 
    return future.result(timeout) 

async def add(a, b): 
    await asyncio.sleep(1) 
    return a + b 

loop = asyncio.get_event_loop() 
future = loop.run_in_executor(None, target, loop) 
assert loop.run_until_complete(future) == 3 

---

pierwotnie napisali podklasę concurrent.futures.Executor które mogą być nadal realizowane jako:

class LoopExecutor(concurrent.futures.Executor): 
    """An Executor subclass that uses an event loop 
    to execute calls asynchronously.""" 

    def __init__(self, loop=None): 
     """Initialize the executor with a given loop.""" 
     self.loop = loop or asyncio.get_event_loop() 

    def submit(self, fn, *args, **kwargs): 
     """Schedule the callable, fn, to be executed as fn(*args **kwargs). 
     Return a Future object representing the execution of the callable.""" 
     coro = asyncio.coroutine(fn)(*args, **kwargs) 
     return asyncio.run_coroutine_threadsafe(coro, self.loop)