BodySense Event Loop Blocking 边界

用 BodySense AI Service 的同步 psycopg 与本地 Embedding 实例解释：async def 只是协程接口，真正的非阻塞取决于每一段 I/O/CPU 工作是否能及时把执行权还给 Event Loop。

#type / synthesis #status / growing #tech / dev / backend #tech / lang / python #tech / ai #resource / python #resource / bodysense

[!info] related notes

所属 MOC:

相关概念:

易混淆概念:

相关资源:

BodySense Event Loop Blocking 边界

核心结论

Python 中：

async def f():
    ...

只说明 f() 是 coroutine function，可以被 Event Loop 驱动；它不保证函数体里的每一步都非阻塞。

更可靠的审查问题是：

当前 Task 从一次 await 恢复，到下一次真正 suspend 之间，会不会执行长时间同步 I/O 或重 CPU/GPU 工作？

如果会，这段代码仍会占住 Event Loop 所在线程。

Python asyncio 和 JavaScript Event Loop 到底像不像

一个典型 asyncio Event Loop 通常在一个线程里协作式调度许多 Task，这一点和浏览器/Node 的事件循环心智很接近：

Task A run a little
→ await I/O
→ yield

Task B run a little
→ await I/O
→ yield

但不能把它简化成：

Python = 只有一个线程

因为 Python 进程仍然可以：

创建 OS thread；
使用 asyncio.to_thread() / executor；
使用多进程；
部署多个 Uvicorn worker；
让 PyTorch/BLAS/CUDA 在 native 层并行。

更准确：

一个 asyncio loop 的调度依赖其 loop thread 保持响应，因此不应该在这条线程上直接执行长时间 blocking work。

I/O 并发为什么有效

假设三个远端请求各等待 500ms。

同步串行：

A [wait 500]
B           [wait 500]
C                     [wait 500]
≈1500ms

真正 async I/O：

A [send] -----wait----- [ready]
B [send] -----wait----- [ready]
C [send] -----wait----- [ready]
≈ slowest request + overhead

关键不是“Event Loop 同时计算三件事”，而是：

等待 socket/DB readiness 时不需要 Python CPU
→ 当前 coroutine suspend
→ loop 去运行其他 ready Task

所以 asyncio 最擅长的是大量等待型 I/O 的并发控制流。

为什么同步 psycopg 会破坏这个模型

BodySense 审查时的旧路径：

async def search(query: str):
    embedding = await embedding_generator.generate(query)
    conn = psycopg.connect(...)
    with conn.cursor() as cur:
        cur.execute("SELECT ...")
        rows = cur.fetchall()

cur.execute() 的逻辑可能包含：

serialize SQL
→ write socket
→ wait PostgreSQL
→ wait response socket
→ read rows

同步 psycopg 把“等待数据库”表达成：

block current OS thread until operation completes

如果这个 OS thread 恰好就是 Event Loop thread：

Search-1 sync DB wait
        ↓
loop cannot run Search-2
loop cannot emit SSE promptly
loop cannot process interrupt promptly
loop timeout/cancel callbacks also delayed

问题不是“数据库查询会使用 CPU”，而是同步 driver 把 I/O wait 变成 thread blocking。

为什么 `SentenceTransformer.encode()` 是另一种 Blocking

async def generate(text: str):
    model = get_model()
    return model.encode([text])[0]

这里不是在等网络，而是在本机做：

tokenization；
tensor preparation；
transformer forward；
pooling；
CPU matrix ops / GPU work；
materialization。

即使 PyTorch/native code 内部可能释放 GIL 或使用别的线程，外层 Python 调用仍然是一个同步调用边界：它返回前，这个 coroutine 没有显式 suspend point。

因此从 asyncio 设计角度，local encode 仍然是 blocking section。

三类代码的真实执行属性

Native async remote API

response = await async_client.embeddings.create(...)

主要是网络 I/O；等待期间可以让出 loop。

Sync DB inside async function

cur.execute(...)

阻塞 loop thread 等数据库 I/O。

Sync CPU/GPU inside async function

model.encode(texts)

同步等待本机计算完成，loop 无法及时调度其他 Task。

所以：

都写在 async def 中
≠
执行属性相同

`await` 也不等于“这里一定切走”

另一个常见误解：

写了 await
→ 一定让出 Event Loop

更准确：await 给 awaitable 提供 suspend 的机会。

例如：

async def fake_async():
    time.sleep(1)
    return 1

await fake_async()

调用点有 await，但 fake_async() 内部先执行 time.sleep(1)，这一秒仍然阻塞当前线程。

所以代码审查不能只搜调用点有没有 await，而要继续追到真正的 I/O/CPU boundary。

`asyncio.to_thread()` 到底做了什么

result = await asyncio.to_thread(blocking_fn, arg)

心智模型：

Event Loop thread
→ submit blocking_fn to thread pool
→ coroutine awaits Future
→ loop free to schedule other Tasks

Worker thread
→ blocking_fn(arg)
→ complete
→ notify Future ready

它解决的是：

不要让 blocking_fn 占住 Event Loop thread。

但它不自动解决：

blocking_fn 自身很慢；
同时启动太多重计算；
CPU oversubscription；
GPU contention/OOM；
thread pool saturation；
thread 中的同步函数无法被 coroutine cancel 强制杀掉。

所以 to_thread 是 execution boundary，不是容量管理方案。

`to_thread` 对 I/O 和 CPU 的意义不同

同步第三方 I/O SDK

如果没有 async client：

await asyncio.to_thread(sync_sdk_call)

常是一个合理兼容桥，因为 worker thread 大部分时间在等待外部 I/O。

重纯 Python CPU

GIL 下多个 thread 未必带来 CPU parallel speedup；to_thread 仍可以保护 Event Loop responsiveness，但吞吐不一定提高。

Native / GPU 计算

SentenceTransformer/PyTorch 的实际并发要靠 benchmark；即使可以并行，也必须限制并发，避免底层线程/GPU 资源被同时打爆。

为什么“单次只有 40ms”仍可能是生产问题

单请求看：

sync DB = 40ms

好像很短。

但并发下多个 blocking section 会形成 loop starvation：

20 × 40ms
→ 可能制造大量调度延迟

用户看到的可能不是 DB API 报错，而是：

SSE token 卡顿；
ask_user 响应慢；
health check tail latency 上升；
cancellation 不及时；
unrelated Agent run 延迟变长。

因此 async 系统要看：

concurrency + scheduling latency + tail latency

不能只看单次函数 benchmark。

Event Loop Blocking 与 GIL 不是同一个问题

这两个概念经常混在一起。

Event Loop Blocking

问：

当前 loop thread 能不能及时回到 scheduler？

GIL

问：

多个 Python threads 如何同时执行 Python bytecode？

即使某个 native library 释放 GIL，外层同步调用也可能让当前 coroutine 长时间不能回到 Event Loop；反过来，一个纯 Python CPU loop 即使没有 I/O，也会让 loop starvation。

所以：

GIL problem
≠
Event Loop blocking problem

它们可能同时存在，但不能互相替代解释。

FastAPI `async def` Route 为什么特别容易被 Sync SDK 拖死

如果 route 是：

@app.get(...)
async def handler():
    return requests.get(...)

框架不会因为 route 是 async 就自动把 requests.get() 搬去 thread pool。

结果：

one sync call
→ blocks that worker's event loop
→ unrelated async requests in same worker suffer

所以异步框架里最危险的不是“看起来同步”的 def，而是async call path 里偷偷出现 blocking SDK。

怎样审查一条 Async Call Path

不要只看最上层函数。

可以沿调用链标注：

A async route
↓
B async service
↓
C generate embedding
↓
D local model encode   [BLOCKING CPU/GPU]
↓
E DB query             [SYNC I/O]

给每个耗时步骤分类：

native async I/O
sync I/O
affordable inline CPU
heavy blocking CPU/GPU
long durable job

然后选择执行模型。

BodySense 中应该特别检查的 Blocking Source

Network / DB

sync psycopg.Connection / cursor；
requests；
同步第三方 SDK；
大文件同步读写；
blocking subprocess。

CPU/GPU

sentence-transformer encode；
OCR/image preprocessing；
tokenizer；
reranker model；
大量 hashing/compression；
大 JSON transform。

Resource / Lock

thread lock 长持有；
async semaphore 容量太小导致排队；
DB pool exhaustion；
concurrent lazy model initialization。

注意后两类不一定“阻塞 loop”，但都会影响 latency/capacity，需要区分 root cause。

Heartbeat Test：验证“其他 Task 还能不能活着”

一个比“源码里有没有 async”更强的测试：

async def heartbeat(samples):
    while ...:
        samples.append(loop.time())
        await asyncio.sleep(0.01)

同时运行被测工作。

Blocking

time.sleep(0.3)

heartbeat 会出现约 300ms gap。

Offloaded

await asyncio.to_thread(time.sleep, 0.3)

heartbeat 仍会持续调度。

真正测试的是行为：

被测 operation 执行期间，无关 lightweight Task 仍能在合理时间窗口获得 Event Loop 调度。

Thread Identity Test

对 local encode 还可以让 fake model 记录：

threading.get_ident()

并比较：

encode thread != event loop thread

但注意：这是实现层测试；heartbeat 更接近用户可观察 contract。

最稳可以同时有：

implementation smoke
+
behavioral heartbeat

Async Correctness 不只等于 Performance

如果 DB/Embedding block loop，可能造成：

Evidence acquisition timeout
→ critical Gap unresolved
→ DecisionAuthority ABSTAIN

从业务上看是“系统没有足够证据”，但根因可能是 resource starvation。

所以 async resource boundary 直接影响：

Agent evidence quality；
streaming correctness；
HITL responsiveness；
timeout semantics；
production failure attribution。

L4 不是孤立性能课。

Offload 也不能破坏 Domain Invariants

把 sync DB 改 async 后，如果拆坏 transaction：错。

把 encode 下沉 thread 后，如果并发首次加载多个 GPU model：错。

把 retrieval 改 background job 后，如果破坏 interactive interrupt/latency contract：也错。

因此 async refactor 的完整目标：

preserve business invariants
+
change execution resource boundary
+
add concurrency behavior tests
+
preserve cancellation/transaction/provenance

速查表

代码	对 Event Loop	说明
`await asyncio.sleep()`	可让出	timer awaitable
`time.sleep()`	阻塞	当前线程睡眠
`await AsyncOpenAI...`	通常可让出	native async HTTP
`requests.get()` inside async	阻塞	sync network I/O
sync psycopg cursor	阻塞	sync DB I/O
async psycopg cursor + await	可让出	native async DB wait
`model.encode()` direct	高风险阻塞	sync CPU/GPU boundary
`await to_thread(model.encode, ...)`	loop 可响应	仍需容量限制
`await semaphore.acquire()`	不阻塞 loop	async backpressure

自测题

为什么 async def 不能证明函数内部 non-blocking？
为什么 await fake_async() 仍可能堵 loop？
sync DB 与 local encode 分别属于哪类 blocking？
Python asyncio 与 JavaScript Event Loop 的相似点和不同点是什么？
to_thread 解决什么，为什么不保证 CPU throughput？
Event Loop blocking 与 GIL 为什么不能混为一个问题？
为什么单次 40ms sync call 在高并发下也可能显著伤害 SSE/HITL？
Heartbeat test 真正验证的 contract 是什么？
Semaphore 排队为什么属于 backpressure，而不是 loop blocking？
为什么 async refactor 仍然必须保护 transaction/domain invariants？

最终心智

不要问：

“这个函数是不是 async？”

要问：

“这条调用链上有哪些显著 I/O/计算，它们分别在哪条线程/哪个 pool/executor 中执行；发生这些工作时，Event Loop 是否还能调度其他 Task？”

这才是从 async 语法走向生产异步工程的关键转变。

BodySense Event Loop Blocking 边界

核心结论

Python asyncio 和 JavaScript Event Loop 到底像不像

I/O 并发为什么有效

为什么同步 psycopg 会破坏这个模型

为什么 SentenceTransformer.encode() 是另一种 Blocking

三类代码的真实执行属性

Native async remote API

Sync DB inside async function

Sync CPU/GPU inside async function

await 也不等于“这里一定切走”

asyncio.to_thread() 到底做了什么

to_thread 对 I/O 和 CPU 的意义不同

同步第三方 I/O SDK

重纯 Python CPU

Native / GPU 计算

为什么“单次只有 40ms”仍可能是生产问题

Event Loop Blocking 与 GIL 不是同一个问题

Event Loop Blocking

GIL

FastAPI async def Route 为什么特别容易被 Sync SDK 拖死

怎样审查一条 Async Call Path

BodySense 中应该特别检查的 Blocking Source

Network / DB

CPU/GPU

Resource / Lock

Heartbeat Test：验证“其他 Task 还能不能活着”

Blocking

Offloaded

Thread Identity Test

Async Correctness 不只等于 Performance

Offload 也不能破坏 Domain Invariants

速查表

自测题

最终心智

为什么 `SentenceTransformer.encode()` 是另一种 Blocking

`await` 也不等于“这里一定切走”

`asyncio.to_thread()` 到底做了什么

`to_thread` 对 I/O 和 CPU 的意义不同

FastAPI `async def` Route 为什么特别容易被 Sync SDK 拖死