python - asyncio后台线程: run function in main thread blocking

Question

我有两个线程(主线程和一些后台线程)，并且都有自己的异步事件循环。

现在考虑我在后台线程中，并且我想在主线程中执行某些操作 (func_for_main_thread)。执行异步操作将是这样的:

main_thread_loop.call_soon_threadsafe(func_for_main_thread)

但是，我怎样才能做到同步/阻塞，即等到 func_for_main_thread 执行？

相关的是this问题，它对 Qt 提出相同的问题，并描述 Apple GCD 的相同功能。 ，这基本上是:

dispatch_async(dispatch_get_main_queue(), ^{ /* do sth */ });

对比:

dispatch_sync(dispatch_get_main_queue(), ^{ /* do sth */ });

Answer 1

如果我正确理解你想要什么，没有什么可以阻止你通过 Future到主线程以在 func_for_main_thread 完成后将其设置为完成。在后台线程中，您可以等待这个 future 。

换句话说:

import asyncio
from functools import partial


async def called_threadsafe(loop, func):
    current_loop = asyncio.get_event_loop()
    fut = asyncio.Future()

    def call_and_set():
        try:
            res = func()
        except Exception as exc:
            f = partial(fut.set_exception, exc)
            current_loop.call_soon_threadsafe(f)
        else:
            f = partial(fut.set_result, res)
            current_loop.call_soon_threadsafe(f)
    loop.call_soon_threadsafe(call_and_set)  # submit to execute in other thread

    return await fut  # in current thread await other thread executed func and set future

演示其工作原理的完整代码:

import asyncio
from functools import partial
import threading
import time


async def called_threadsafe(loop, func):
    current_loop = asyncio.get_event_loop()
    fut = asyncio.Future()

    def call_and_set():
        try:
            res = func()
        except Exception as exc:
            f = partial(fut.set_exception, exc)
            current_loop.call_soon_threadsafe(f)
        else:
            f = partial(fut.set_result, res)
            current_loop.call_soon_threadsafe(f)
    loop.call_soon_threadsafe(call_and_set)

    return await fut


# helpers:
_l = threading.Lock()

def info(*args):
    with _l:
        print(*args, threading.get_ident(), flush=True)


def start_bg_loop():
    bg_loop = asyncio.new_event_loop()

    def startup():
        asyncio.set_event_loop(bg_loop)
        bg_loop.run_forever()

    t = threading.Thread(target=startup)
    t.daemon = True
    t.start()

    return bg_loop


# main part:
def func_for_main_thread():
    info('executed in fg thread')
    time.sleep(0.05)
    return 'got result in bg thread'


async def bg_main(fg_loop):
    info('bg_main started')
    await asyncio.sleep(0.1)

    res = await called_threadsafe(fg_loop, func_for_main_thread)
    info(res)

    info('bg_main finished')


async def fg_main(bg_loop):
    info('fg_main started')
    await asyncio.sleep(1)
    info('fg_main finished')


fg_loop = asyncio.get_event_loop()
bg_loop = start_bg_loop()

asyncio.run_coroutine_threadsafe(bg_main(fg_loop), bg_loop)
fg_loop.run_until_complete(fg_main(bg_loop))

输出:

fg_main started 2252
bg_main started 5568
executed in fg thread 2252
got result in bg thread 5568
bg_main finished 5568
fg_main finished 2252