2.2 消息边界、Deadline 与 Backpressure

同一连接连续调用两次 send，对端应用只会读到一条连续字节流。TCP 负责可靠、有序地交付字节，却不承诺保留两次调用的消息边界。把一次 recv 当成一条消息，测试环境也许暂时配合，生产环境通常很快翻脸。

这一课实现四字节长度前缀协议。先用本地 socket 连续写入，再用可控的输入适配器强制短读；随后解释 timeout、deadline、pipelining 和 backpressure 怎样影响连接状态。

1. 四种常见 framing

固定长度

每条消息恰好 N 字节。解析最简单，但短消息浪费空间，变长字段需要额外规则。适合硬件记录或已知尺寸 block。

分隔符

以换行或特殊序列结束：

PING\r\n
SET key value\r\n

实现必须处理分隔符跨越两次 recv、payload 中转义、最大行长和未结束行。不能无限等待并增长 buffer。

长度前缀

[4-byte length][payload]

可以传二进制和空 payload。读取 header 后必须先验证上限，再分配和读取 body；否则一个 0xffffffff 就可能触发内存耗尽。

语法与协议定义边界

JSON 对象可以靠完整语法识别结束，但字符串转义、嵌套和顶层数字都需要明确的解析规则；HTTP 则有自己的消息长度规则。Protobuf wire format 本身不能确定连续消息的分界，仍需长度前缀等外层 framing。数据“有格式”不等于一读就是一条完整消息。

2. 正确读取固定字节数

recv_exact 必须区分：

• 收满 N 字节；
• 开始新 header 前遇到 EOF，表示消息流正常结束；
• header 内部或已声明的 body 未读满时遇到 EOF，表示截断 frame。

import socket
import struct
import threading

HEADER = struct.Struct("!I")
MAX_FRAME = 1024 * 1024

def recv_exact(connection, length, allow_clean_eof=False):
    if length < 0:
        raise ValueError("length must be nonnegative")
    data = bytearray()
    while len(data) < length:
        chunk = connection.recv(length - len(data))
        if chunk == b"":
            if allow_clean_eof and len(data) == 0:
                return None
            raise EOFError(
                f"stream ended after {len(data)} of {length} bytes"
            )
        data.extend(chunk)
    return bytes(data)

def encode_frame(payload):
    if len(payload) > MAX_FRAME:
        raise ValueError("frame too large")
    return HEADER.pack(len(payload)) + payload

def recv_frame(connection):
    header = recv_exact(
        connection,
        HEADER.size,
        allow_clean_eof=True,
    )
    if header is None:
        return None

    (length,) = HEADER.unpack(header)
    if length > MAX_FRAME:
        raise ValueError(f"declared frame too large: {length}")
    return recv_exact(connection, length)

def fragmented_writer(connection, frames):
    with connection:
        encoded = b"".join(encode_frame(frame) for frame in frames)
        for byte in encoded:
            connection.sendall(bytes([byte]))
        connection.shutdown(socket.SHUT_WR)

left, right = socket.socketpair()
left.settimeout(2)
right.settimeout(2)
expected = [b"alpha", b"", b"omega"]
writer = threading.Thread(
    target=fragmented_writer,
    args=(left, expected),
)
writer.start()

with right:
    actual = []
    while True:
        frame = recv_frame(right)
        if frame is None:
            break
        actual.append(frame)

writer.join(timeout=3)
assert not writer.is_alive()
assert actual == expected
print(actual)

这个 Python 3.8+ 例子使用 socketpair 创建本机连接；在常见 Unix 上默认是 AF_UNIX 字节流，不经过 TCP。其他平台的实现可能不同。逐字节 sendall 仍可能被接收端合并，不能证明所有短读路径都测到了。!I 是 network byte order 的无符号 32 位整数，长度按字节计，不是字符数。

接着运行下面的适配器，强制每次返回不超过指定字节数。它复用上面的 encode_frame 和 recv_frame，把“刚好读短了”变成可重复输入：

class ChunkReader:
    def __init__(self, data, max_chunk):
        if max_chunk <= 0:
            raise ValueError("max_chunk must be positive")
        self.data = data
        self.max_chunk = max_chunk

    def recv(self, length):
        count = min(length, self.max_chunk)
        part, self.data = self.data[:count], self.data[count:]
        return part

wire = b"".join(encode_frame(p) for p in [b"alpha", b"", b"omega"])
for chunk_size in [1, 2, 3, 4, len(wire)]:
    reader = ChunkReader(wire, chunk_size)
    frames = []
    while (frame := recv_frame(reader)) is not None:
        frames.append(frame)
    assert frames == [b"alpha", b"", b"omega"]
print("all chunk sizes passed")

None 表示新 header 之前的 EOF；b"" 是有效空 payload。若混用 if not frame，空消息会被错当成结束。适配器只验证解析逻辑，不能代替真实 socket 的超时和关闭测试。

3. Header 也可能被拆开

不少错误实现会写：

length = struct.unpack("!I", connection.recv(4))[0]

recv(4) 允许只返回 1 至 3 字节，并不会替你等待长度字段完整。header 和 body 都要走精确读取循环；不能用“一次调用在本机总能读满”的观察替代接口约定。

另一种错误是把 EOF 当成“暂时没数据”：

while True:
    if connection.recv(4096) == b"":
        continue

这会在已关闭连接上忙循环。blocking socket 暂时没数据时会阻塞或 timeout；返回空字节表示有序 EOF。

4. 长度字段属于不可信输入

读取到长度后，在分配前检查：

• 是否超过协议最大 frame；
• 是否允许零长度；
• 长度计算在所用整数类型中是否可能溢出；
• 解压后的大小是否另有上限；
• 一个连接允许多少在途 frame；
• 总 buffer 是否受全局预算控制。

“最大 1 MiB”只是示例。真实上限应由业务语义、内存预算与代理链限制共同决定。

压缩协议还要防高压缩比 payload。wire length 很小不等于解压内存小；应同时限制压缩输入、解压输出和 CPU 工作量。

5. Per-operation timeout 与总 deadline

给每次 recv 设置 5 秒 timeout，不代表整个请求最多 5 秒。对端可以每 4 秒发一个字节，让一条 1 MiB 消息持续数周。

下面沿用第 2 节的 HEADER 和 MAX_FRAME，用 monotonic clock 计算剩余预算。同一个绝对 deadline 同时传给 header 和 body，不能在进入 body 时重新加 5 秒：

import time

def recv_exact_before(connection, length, deadline, allow_clean_eof=False):
    if length < 0:
        raise ValueError("length must be nonnegative")
    data = bytearray()
    previous_timeout = connection.gettimeout()
    try:
        while len(data) < length:
            remaining = deadline - time.monotonic()
            if remaining <= 0:
                raise TimeoutError("message deadline exceeded")
            connection.settimeout(remaining)
            chunk = connection.recv(length - len(data))
            if chunk == b"":
                if allow_clean_eof and not data:
                    return None
                raise EOFError("truncated message")
            data.extend(chunk)
        return bytes(data)
    finally:
        connection.settimeout(previous_timeout)

def recv_frame_before(connection, deadline):
    header = recv_exact_before(connection, HEADER.size, deadline, True)
    if header is None:
        return None
    (length,) = HEADER.unpack(header)
    if length > MAX_FRAME:
        raise ValueError("frame too large")
    return recv_exact_before(connection, length, deadline)

# Use one budget for both header and body:
# frame = recv_frame_before(connection, time.monotonic() + 5)

函数会恢复原来的 socket timeout，前提是这个 socket 的读写与超时设置由当前操作独占。并发任务不能相互改 timeout。截止时间限制的是这里的读取等待；若要包含解析、排队和下游调用，调用链也必须传递同一预算。

还需要决定：

• timeout 后连接中的半条 frame 如何处理？
• 是否关闭连接，还是协议支持安全 resynchronize？
• request deadline 是否包含排队和下游调用？
• response 已部分发送时能否重试？
• 取消怎样传递给 worker 和 I/O？

deadline 是状态机的一部分，不是给 socket 加一个数字。

6. Backpressure：接收缓冲区满了以后

长度前缀解决消息边界；慢客户端还会让缓冲区堆积。简化后的发送路径是：

1. 字节先进入本机 socket send buffer；
2. TCP 按接收窗口与拥塞窗口发送；
3. 对端内核放入 receive buffer；
4. 对端应用最终 recv。

若对端读得慢，buffer 会逐级填满。blocking sendall 最终阻塞，nonblocking send 返回 would-block，async writer 的待发送队列会增长。

错误做法是每个 producer 无限制地把响应追加进用户态列表。这样把网络 backpressure 变成进程 OOM。

需要明确：

• 每连接最大 queued bytes；
• 全局最大 queued bytes；
• 超限时暂停读取上游、拒绝新请求还是关闭慢连接；
• 哪些消息可以丢弃或合并；
• 写入 deadline；
• 公平调度，避免一个大响应饿死小响应。

Python asyncio 的 StreamWriter.write 会尝试发送，未能立即发送的部分进入 transport buffer；await drain() 在写缓冲达到 high-water mark 后等待其降到 low-water mark。下面的 Python 3.11+ 片段限制每次提交量，并让一个异步数据源随下游速度推进：

import asyncio

async def send_chunks(writer, chunks):
    async with asyncio.timeout(5):
        async for chunk in chunks:
            if len(chunk) > 64 * 1024:
                raise ValueError("producer chunk exceeds 64 KiB")
            writer.write(chunk)
            await writer.drain()

这里假设只有一个任务向 writer 写入，chunks 是按需产生字节块的异步迭代器。drain 不是对端已读确认，high-water mark 也不是硬内存上限：最后一次 write 可以超出阈值一个块。数据源预取、其他连接和 kernel buffer 仍要单独预算。超时或取消后，由连接所有者决定关闭及清理，不能无条件重发。

7. Request/response 顺序与 request ID

最简单协议每连接一次只允许一个请求：

send request A
receive response A
send request B
receive response B

pipelining 允许 A、B 同时在途。如果 response 必须按请求顺序返回，慢 A 会阻塞已完成的 B；如果允许乱序返回，每条 frame 就要带 request ID：

[length][request-id][type][payload]

还要定义：

• ID 是否可复用，何时可复用；
• response、error、cancel 怎样关联；
• server push 使用哪个 ID 空间；
• 同一 ID 重复出现如何处理；
• 连接重建后旧 ID 是否仍有意义。

HTTP/2、gRPC 与 QUIC stream 已经解决了大量多路复用状态，普通应用不应轻易重造。

8. Blocking、nonblocking 与 readiness

nonblocking socket 在暂时不能完成时返回 would-block。epoll/kqueue 等 readiness API 告诉 event loop 某 fd “现在可能可读/可写”，不保证下一次操作一定完成全部请求。

事件处理器仍要：

• 循环 read 到 would-block 或达到公平预算；
• 保存半个 header/body 的 parser 状态；
• 只在有待发送数据时关注 writable；
• 处理 hangup/error 与剩余可读数据；
• 避免一个永远活跃的连接垄断 loop；
• 在关闭前取消 timer 和业务任务。

edge-triggered 模式通常要求 drain 到 would-block，否则可能等不到下一次边沿。level-triggered 会在条件仍成立时继续通知，实现更直观但也可能反复唤醒。

若 edge-triggered 处理器为了公平提前停下，必须把仍可能有数据的连接留在应用 ready queue，或使用正确的重新注册策略继续处理，不能直接等新边沿。

async/await 把状态机藏进编译器/runtime，不会消除这些协议规则。

9. TLS 会再增加一层 framing

这里讨论 TLS over TCP：TLS 把应用字节包装成 record。一次应用 write 可能对应多个 TLS record 和 TCP segment；一次 TCP recv 也可能只含半个 TLS record。QUIC 使用 TLS 握手与派生密钥，但不使用这套 TLS record 承载。

应用应通过 TLS library 的 read/write API，让库维护 record、认证和重组状态，不要对加密 socket 的底层 TCP 字节自行切片。TLS clean shutdown 与 TCP EOF 也有区别，安全敏感协议需要判断是否收到规范的 close notification。

10. UDP framing 的不同边界

UDP 保留 datagram boundary，所以一个长度前缀通常不用于跨多个 recvfrom 重组同一个 datagram。但应用仍可能在一个 datagram 内放多条 record，或把大消息分成多个 datagram。

自定义 UDP 分片需要 message ID、片号、总数、timeout、去重、内存上限和拥塞控制。丢一个片可能让整个消息不可用。若需求是可靠、安全、多路流传输，优先评估 QUIC 而不是逐步复刻它。

11. 协议测试矩阵

不要只测“正常发送一条消息”。至少覆盖：

输入	预期
header 每次只到 1 字节	正确重组
body 分成任意 chunk	正确重组
两个 frame 合并到一次 recv	解析两条
零长度 frame	按协议接受或拒绝
header 中途 EOF	截断错误
body 中途 EOF	截断错误
length 超上限	分配前拒绝
慢速逐字节输入	总 deadline 生效
对端不读 response	queued bytes 有界
cancel 与 close 同时发生	只释放一次，状态一致

property-based test 可以随机切分同一编码字节串，检查每种已测切分是否得到相同消息序列。随机样本能扩大覆盖，但不能证明所有切分都已测过。

12. 小结

TCP 协议设计必须主动补上字节流没有提供的东西：

• framing 定义消息边界；
• recv_exact 处理 header/body 的任意切分；
• 长度和解压结果必须在分配前受限；
• operation timeout 不能代替 end-to-end deadline；
• backpressure 让慢下游限制上游，而不是无限堆内存；
• pipelining 需要取消和响应顺序规则，乱序响应还需要 request ID 等关联机制；
• readiness/async 仍要维护半包、错误与关闭状态；
• 测试要随机化 chunk boundary，并覆盖截断与慢客户端。

到这里，协议已经给连续字节划出消息边界，也能在缓冲区达到上限时向上游传播压力。下一章先看 TCP 连接怎样建立和关闭，再追踪序号、ACK、流量控制与重传。

参考资料

• Python 3.12: socket
• Python 3.12: asyncio Streams
• Protocol Buffers: Streaming Multiple Messages
• RFC 9001: Using TLS to Secure QUIC
• Linux man-pages: epoll(7)