为何Django视图中用asyncio.gather调用sync_to_async函数会挂起？

我有一个Django视图，调用asgiref.sync.async_to_sync函数查询远程API，用于批量更新员工薪资单数据。在此调用中需再次访问数据库，因此使用asgiref.sync.sync_to_async装饰的函数来获取数据。当我用asyncio.gather并发执行员工更新函数调用时，程序会无限挂起，甚至不会进入数据库调用——即使gather列表中只有一个调用也会出现此问题。但直接await该函数却能正常运行。所有内容都在Uvicorn下以ASGI应用运行。

最小复现代码 #

@sync_to_async
def _get_database_employee_payslips(employee_data):
    print(f"Entering DB function")
    return employee_data

@async_to_sync
async def get_full_employees_data(_, __):
    print(f"This works")
    ret = await _get_database_employee_payslips({'id': 1625841})
    print(f"Got {ret}")

    print(f"This doesn't")
    ret = await asyncio.gather(*[
        _get_database_employee_payslips({'id': employee_id})
        for employee_id in [1625841]
    ], return_exceptions=True)    
    print(f"We're hung in the above call")

    return ret

运行结果 #

INFO 2022-09-27 14:20:20,234 on 10776 140271592661440 Application startup complete.
DEBUG 2022-09-27 14:20:22,163 middleware 10776 140271484180032 process_request
DEBUG 2022-09-27 14:20:22,163 model_factory 10776 140271484180032 No maximum request body size is set, continuing.
DEBUG 2022-09-27 14:20:22,183 model_factory 10776 140271484180032 Created new request model with pk 161dfa90-4082-4ef1-8ab0-84d613c25550
This works
Entering DB function
Got {'id': 1625841}
This doesn't

环境信息 #

• WSL2
• Ubuntu 22.04.1 LTS
• Python 3.10.4

相关依赖包 #

aiohttp==3.8.3
aiosignal==1.2.0
anyio==3.6.1
asgiref==3.5.2
async-timeout==4.0.2
asyncio==3.4.3
Django==4.1.1
uvicorn==0.18.3
uvloop==0.17.0
uWSGI==2.0.20

这是线程锁竞争引发的死锁：

1. async_to_sync会把异步函数放到线程池的线程中执行，该线程绑定事件循环并持有锁。
2. 默认情况下sync_to_async的thread_sensitive为True，它会尝试复用当前线程（即async_to_sync所在线程）执行同步函数，这需要获取当前线程的事件循环锁。
3. 当用asyncio.gather并发调用时，async_to_sync所在线程正等待gather完成，无法释放锁；而sync_to_async调用又在等待获取该锁执行逻辑，最终形成互相等待的死锁，导致程序挂起。

哪怕gather里只有一个调用，死锁依然会发生——因为asyncio.gather的调度逻辑会触发sync_to_async以需要重新获取锁的方式执行，而原线程的锁已被async_to_sync持有。

方案1：修改sync_to_async的线程敏感参数 #

给sync_to_async添加thread_sensitive=False，让它使用独立线程池执行同步函数，避免复用当前线程的锁：

@sync_to_async(thread_sensitive=False)
def _get_database_employee_payslips(employee_data):
    print(f"Entering DB function")
    return employee_data

这样每个sync_to_async调用会在新线程运行，不会和async_to_sync的线程锁产生竞争，asyncio.gather即可正常执行。

方案2：重构为纯异步逻辑 #

如果业务允许，将整个视图逻辑改为纯异步模式，直接使用Django 4.0+原生支持的异步ORM，彻底绕开线程锁问题。

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