Python：同时挂起异步协程和同步函数

Question

我想在同步函数执行期间建立一个 SSH SOCKs 隧道（使用asyncssh）。功能完成后，我想拆除隧道并退出。

显然，必须等待一些异步函数来保持隧道正常工作，所以重要的是conn.wait_closed() 和同步函数是同时执行的。所以我很确定我实际上需要第二个线程。 我首先使用ThreadPoolExecutor 和run_in_executor 尝试了一些更理智的事情，但最终得到了下面这个糟糕的多线程变体。

#! /usr/bin/env python3

import traceback
from threading import Thread
from concurrent.futures import ThreadPoolExecutor

import asyncio, asyncssh, sys

_server="127.0.0.1"
_port=22
_proxy_port=8080


async def run_client():
    conn = await asyncio.wait_for(
        asyncssh.connect(
            _server,
            port=_port,
            options=asyncssh.SSHClientConnectionOptions(client_host_keysign=True),
        ),
        10,
    )

    listener = await conn.forward_socks('127.0.0.1', _proxy_port)
    return conn

async def do_stuff(func):
    try:
        conn = await run_client()
        print("SSH tunnel active")

        def start_loop(loop):
            asyncio.set_event_loop(loop)
            try:
                loop.run_forever()
            except Exception as e:
                print(f"worker loop: {e}")

        async def thread_func():
            ret=await func()
            print("Func done - tearing done worker thread and SSH connection")
            conn.close()
            #  asyncio.get_event_loop().stop()
            return ret

        func_loop = asyncio.new_event_loop()
        func_thread = Thread(target=start_loop, args=(func_loop,))
        func_thread.start()
        print("thread started")
        fut = asyncio.run_coroutine_threadsafe(thread_func(), func_loop)
        print(f"fut scheduled: {fut}")

        done = await asyncio.gather(asyncio.wrap_future(fut), conn.wait_closed())
        print("wait done")
        for ret in done:
            print(f"ret={ret}")

        # Canceling pending tasks and stopping the loop
        #  asyncio.gather(*asyncio.Task.all_tasks()).cancel()

        print("stopping func_loop")
        func_loop.call_soon_threadsafe(func_loop.stop())
        print("joining func_thread")
        func_thread.join()
        print("joined func_thread")

    except (OSError, asyncssh.Error) as exc:
        sys.exit('SSH connection failed: ' + str(exc))
    except (Exception) as exc:
        sys.exit('Unhandled exception: ' + str(exc))
        traceback.print_exc()


async def just_wait():
    print("starting just_wait")
    input()
    print("ending just_wait")
    return 42

asyncio.get_event_loop().run_until_complete(do_stuff(just_wait))

它实际上“工作”“正确”，直到我在join工作线程时遇到异常。我想是因为我做的事情不是线程安全的。

Exception in callback None()
handle: <Handle>
Traceback (most recent call last):
  File "/usr/lib/python3.7/asyncio/events.py", line 88, in _run
    self._context.run(self._callback, *self._args)
TypeError: 'NoneType' object is not callable

要测试代码，您必须运行本地 SSH 服务器，并为您的用户设置密钥文件。您可能想要更改 _port 变量。

我正在寻找异常的原因和/或在线程中需要较少人工干预并且可能仅使用单个事件循环的程序版本。当我想await 这两件事（如asyncio.gather 调用）时，我不知道如何实现后者。

Answer 1

你的错误的直接原因是这一行：

# incorrect
func_loop.call_soon_threadsafe(func_loop.stop())

目的是在运行func_loop 事件循环的线程中调用func_loop.stop()。但正如所写，它调用func_loop.stop() 在当前线程中 并将其返回值（None）作为要调用的函数传递给call_soon_threadsafe。这会导致call_soon_threadsafe 抱怨 None 是不可调用的。要解决眼前的问题，您应该删除多余的括号并将方法调用为：

# correct
func_loop.call_soon_threadsafe(func_loop.stop)

但是，代码肯定写得过于复杂了：

• 当您已经在一个事件循环中时，创建一个新的事件循环是没有意义的
• just_wait 不应该是 async def，因为它不等待任何东西，所以它显然不是异步的。
• sys.exit 采用整数退出状态，而不是字符串。此外，尝试在调用sys.exit 之后打印回溯也没有多大意义。

要从 asyncio 运行非异步函数，只需将 run_in_executor 与该函数一起使用，然后按原样将非异步函数传递给它。您不需要额外的线程或额外的事件循环，run_in_executor 将处理线程并将其与您当前的事件循环连接，从而有效地使同步功能可等待。例如（未经测试）：

async def do_stuff(func):
    conn = await run_client()
    print("SSH tunnel active")
    loop = asyncio.get_event_loop()
    ret = await loop.run_in_executor(None, func)
    print(f"ret={ret}")
    conn.close()
    await conn.wait_closed()
    print("wait done")

def just_wait():
    # just_wait is a regular function; it can call blocking code,
    # but it cannot await
    print("starting just_wait")
    input()
    print("ending just_wait")
    return 42

asyncio.get_event_loop().run_until_complete(do_stuff(just_wait))

如果您需要在just_wait 中等待，您可以将其设为async 并使用run_in_executor 作为其中的实际阻塞代码：

async def do_stuff():
    conn = await run_client()
    print("SSH tunnel active")
    loop = asyncio.get_event_loop()
    ret = await just_wait()
    print(f"ret={ret}")
    conn.close()
    await conn.wait_closed()
    print("wait done")

async def just_wait():
    # just_wait is an async function, it can await, but
    # must invoke blocking code through run_in_executor
    print("starting just_wait")
    loop = asyncio.get_event_loop()
    await loop.run_in_executor(None, input)
    print("ending just_wait")
    return 42

asyncio.run(do_stuff())