PyCharm中trio_asyncio脚本调试失败原因及解决方法咨询

问题背景 #

以下是基于 Trio 异步库编写的简化程序，通过 trio_asyncio 调用 asyncio 方法：

import asyncio
import trio
import trio_asyncio


async def async_main(*args):
    print('async_main start')
    async with trio_asyncio.open_loop() as loop:
        print('async_main before trio sleep')
        await trio.sleep(1)
        print('async_main before asyncio sleep')
        await trio_asyncio.aio_as_trio(asyncio.sleep)(2)
        print('async_main after sleeps')
    print('async_main stop')


if __name__ == '__main__':
    print('main start')
    trio.run(async_main)
    print('main stop')

正常运行输出 #

main start
async_main start
async_main before trio sleep
async_main before asyncio sleep
async_main after sleeps
async_main stop
main stop

调试模式下的异常输出 #

Connected to pydev debugger (build 232.9559.58)
main start
async_main start
async_main before trio sleep
async_main before asyncio sleep
Traceback (most recent call last):
  File "/home/vaclav/.config/JetBrains/PyCharmCE2023.2/scratches/scratch_3.py", line 12, in async_main
    await trio_asyncio.aio_as_trio(asyncio.sleep)(2)
  File "/home/vaclav/.cache/pypoetry/virtualenvs/maybankwithoutselenium-RhkLw-zs-py3.11/lib/python3.11/site-packages/trio_asyncio/_adapter.py", line 54, in __call__
    return await self.loop.run_aio_coroutine(f)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/vaclav/.cache/pypoetry/virtualenvs/maybankwithoutselenium-RhkLw-zs-py3.11/lib/python3.11/site-packages/trio_asyncio/_base.py", line 214, in run_aio_coroutine
    fut = asyncio.ensure_future(coro, loop=self)
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/opt/pycharm-community/plugins/python-ce/helpers/pydev/_pydevd_asyncio_util/pydevd_nest_asyncio.py", line 156, in ensure_future
    return loop.create_task(coro_or_future)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/lib/python3.11/asyncio/base_events.py", line 436, in create_task
    task = tasks.Task(coro, loop=self, name=name, context=context)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/opt/pycharm-community/plugins/python-ce/helpers/pydev/_pydevd_asyncio_util/pydevd_nest_asyncio.py", line 390, in task_new_init
    self._loop.call_soon(self, context=self._context)
  File "/home/vaclav/.cache/pypoetry/virtualenvs/maybankwithoutselenium-RhkLw-zs-py3.11/lib/python3.11/site-packages/trio_asyncio/_base.py", line 312, in call_soon
    self._check_callback(callback, 'call_soon')
  File "/usr/lib/python3.11/asyncio/base_events.py", line 776, in _check_callback
    raise TypeError(
TypeError: a callable object was expected by call_soon(), got <Task pending name='Task-1' coro=<__call_defer() running at /home/vaclav/.cache/pypoetry/virtualenvs/maybankwithoutselenium-RhkLw-zs-py3.11/lib/python3.11/site-packages/trio_asyncio/_adapter.py:16>>
python-BaseException
sys:1: RuntimeWarning: coroutine '_call_defer' was never awaited
Exception in default exception handler
Traceback (most recent call last):
  File "/usr/lib/python3.11/asyncio/base_events.py", line 1797, in call_exception_handler
    self.default_exception_handler(context)
  File "/home/vaclav/.cache/pypoetry/virtualenvs/maybankwithoutselenium-RhkLw-zs-py3.11/lib/python3.11/site-packages/trio_asyncio/_async.py", line 42, in default_exception_handler
    raise RuntimeError(message)
RuntimeError: Task was destroyed but it is pending!

Process finished with exit code 1

疑问 #

1. 为何代码正常运行时无问题，调试模式下却失败？
2. 如何修改代码，使其既能调用trio和asyncio方法，又能支持调试模式运行？

问题分析与解决方案 #

1. 调试模式失败原因 #

PyCharm调试器会通过pydevd_nest_asyncio补丁修改asyncio的核心方法（如create_task、ensure_future），以实现异步调试功能。但该补丁与trio_asyncio的内部实现存在兼容性冲突：

• 调试器补丁在创建Task后，会调用loop.call_soon(self)，直接将Task实例作为回调函数传入；
• 而trio_asyncio的call_soon方法会严格校验回调必须是可调用对象，Task本身不具备可调用性，因此触发TypeError；
• 后续的未等待协程警告、任务销毁异常均为初始错误引发的连锁反应。

2. 兼容调试模式的修改方案 #

方法一：禁用PyCharm异步调试补丁

打开PyCharm调试配置，找到Python Debugger选项，取消勾选Asyncio debugging。调试器将不再修改asyncio行为，trio_asyncio可正常工作。

方法二：代码适配调试器

若需保留异步调试功能，可替换trio_asyncio.aio_as_trio为兼容性更好的调用方式：

方案A：使用trio_asyncio.run_asyncio

import asyncio
import trio
import trio_asyncio


async def async_main(*args):
    print('async_main start')
    async with trio_asyncio.open_loop():
        print('async_main before trio sleep')
        await trio.sleep(1)
        print('async_main before asyncio sleep')
        # 用run_asyncio替代aio_as_trio
        await trio_asyncio.run_asyncio(asyncio.sleep, 2)
        print('async_main after sleeps')
    print('async_main stop')


if __name__ == '__main__':
    print('main start')
    trio.run(async_main)
    print('main stop')

方案B：手动通过线程安全接口调度

import asyncio
import trio
import trio_asyncio


async def async_main(*args):
    print('async_main start')
    async with trio_asyncio.open_loop() as loop:
        print('async_main before trio sleep')
        await trio.sleep(1)
        print('async_main before asyncio sleep')
        # 用run_coroutine_threadsafe配合trio.wrap_future
        fut = asyncio.run_coroutine_threadsafe(asyncio.sleep(2), loop)
        await trio.wrap_future(fut)
        print('async_main after sleeps')
    print('async_main stop')


if __name__ == '__main__':
    print('main start')
    trio.run(async_main)
    print('main stop')

内容的提问来源于stack exchange，提问作者eNca