L4 流

主线位置：请求主线的第一步。监听器 accept 出一个 TcpStream，From<TcpStream> 把它包成 l4::Stream，每连接任务里的 handshake 调 set_buffer 换上真正的缓冲区，再装箱成 Box<dyn IO> 交给 HTTP 层。上游方向的连接器也用同一个类型，但不调 set_buffer。

flowchart TB
  tcp["TcpStream"] --> raw["RawStream::Tcp"]
  raw --> wrap["RawStreamWrapper"]
  wrap --> bs0["BufStream 0/0"]
  bs0 --> acc{"哪条路"}
  acc -->|"监听器 accept"| hs["handshake: set_buffer"]
  hs --> bs1["BufStream 64K/1460"]
  acc -->|"连接器 connect"| keep["保持 0/0"]
  bs1 --> boxed["Box dyn IO"]
  keep --> boxed
  boxed --> http["HTTP/1、h2、TLS"]

代码在 pingora-core/src/protocols/l4/：stream.rs 1049 行（其中最后 176 行是只在 Linux 上编译的测试），ext.rs 864 行，socket.rs 301 行，virt.rs 161 行；摘要类型在 protocols/digest.rs（286 行）。TCP 选项（keepalive、TCP_FASTOPEN、DSCP）在建连一侧怎么设，见 L4 连接；监听器怎么建，见 服务与监听。

实验在 exp-pingora/proto-main（M2 Max、macOS 27、rustc 1.99.0，debug，tokio 1.53.2）：一个代理进程在 127.0.0.1:18300 等几个端口上监听，上游是同一个二进制里的原始 TCP 服务（18301），把每次 read 收到的字节原样打进日志。

上层只看见 Box<dyn IO>

HTTP 层拿到的不是 l4::Stream，而是 protocols::Stream，一个 trait object。IO 把九个 trait 捆在一起：

pub trait IO:
    AsyncRead
    + AsyncWrite
    + Shutdown
    + UniqueID
    + Ssl
    + GetTimingDigest
    + GetProxyDigest
    + GetSocketDigest
    + Peek
    + Unpin
    + Debug
    + Send
    + Sync
{
    /// helper to cast as the reference of the concrete type
    fn as_any(&self) -> &dyn Any;
    /// helper to cast back of the concrete type
    fn into_any(self: Box<Self>) -> Box<dyn Any>;
}
// ...
/// The type of any established transport layer connection
pub type Stream = Box<dyn IO>;

满足这些 trait 的类型自动实现 IO。明文连接是 l4::Stream，TLS 连接是包着 l4::Stream 的 SslStream 或 rustls 的 TlsStream，测试里还有 tokio::io::DuplexStream 和 Cursor。Ssl、Peek 这几个 trait 都有默认实现（「没有 TLS」「不支持 peek」），所以 HTTP/1 服务端、h2 握手、日志里的 digest() 都只写一份，不用对每种连接泛型展开。代价是每次读写多一次虚调用。hyper 走的是另一条路：连接类型是泛型参数，只要求 hyper::rt::Read + Write，见 IO trait。

三层包装

RawStream 是三种底层连接的枚举：

#[derive(Debug)]
enum RawStream {
    Tcp(TcpStream),
    #[cfg(unix)]
    Unix(UnixStream),
    Virtual(virt::VirtualSocketStream),
}

Virtual 包一个用户实现的 VirtualSocket（AsyncRead + AsyncWrite 再加一个 set_socket_option），没有真的 fd，as_raw_fd 返回 -1。AsyncRead、AsyncWrite 的实现是手写的 pin 投影：unsafe { Pin::new_unchecked(s) } 按分支转发。

RawStreamWrapper 在外面加一个 rx_ts: Option<SystemTime>。在 Linux 上调过 set_rx_timestamp 之后，TCP 读走 recvmsg，从控制消息里取 SO_TIMESTAMPING 的内核接收时间戳；其他平台的 poll_read 只是转发，set_rx_timestamp 是空函数。这条路只在 #[cfg(target_os = "linux")] 下编译，本页没法在 macOS 上跑，只读了源码。

最外层是 l4::Stream：

pub struct Stream {
    // Use `Option` to be able to swap to adjust the buffer size. Always safe to unwrap
    stream: Option<BufStream<RawStreamWrapper>>,
    // the data put back at the front of the read buffer, in order to replay the read
    rewind_read_buf: Vec<Vec<u8>>,
    buffer_write: bool,
    proxy_digest: Option<Arc<ProxyDigest>>,
    socket_digest: Option<Arc<SocketDigest>>,
    /// When this connection is established
    pub established_ts: SystemTime,
    establishment_duration: Option<Duration>,
    offload_wait_duration: Option<Duration>,
    /// The distributed tracing object for this stream
    pub tracer: Option<Tracer>,
    read_pending_time: AccumulatedDuration,
    write_pending_time: AccumulatedDuration,
    /// Last rx timestamp associated with the last recvmsg call.
    pub rx_ts: Option<SystemTime>,
}

BufStream 是 tokio 的类型，内部是 BufReader<BufWriter<RW>>。其余字段分三类：回退栈 rewind_read_buf（下文），摘要（proxy_digest、socket_digest、几个时间），统计（read_pending_time / write_pending_time 累计读写处于 Pending 的时长，由 AccumulatedDuration 在每次 poll 后按结果开表、停表）。

缓冲区：先给 0，握手时再换

impl From<TcpStream> for Stream 建的 BufStream 两边容量都是 0：

impl From<TcpStream> for Stream {
    fn from(s: TcpStream) -> Self {
        Stream {
            stream: Some(BufStream::with_capacity(
                0,
                0,
                RawStreamWrapper::new(RawStream::Tcp(s)),
            )),
            rewind_read_buf: Vec::new(),
            buffer_write: true,
            // ...
        }
    }
}

UnixStream 和 VirtualSocketStream 的 From 也一样。真正的缓冲区由 set_buffer 换上去：

    /// Set the buffer of BufStream
    /// It is only set later because of the malloc overhead in critical accept() path
    pub(crate) fn set_buffer(&mut self, buffer: L4BufferSettings) {
        use std::mem;
        // Since BufStream doesn't provide an API to adjust the buf directly,
        // we take the raw stream out of it and put it in a new BufStream with the size we want
        let stream = mem::take(&mut self.stream);
        let stream = stream.map(|s| {
            BufStream::with_capacity(
                buffer.read_capacity(),
                buffer.write_capacity(),
                s.into_inner(),
            )
        });
        let _ = mem::replace(&mut self.stream, stream);
    }

stream 字段是 Option 就是为了这次 mem::take。调用它的只有一处：UninitializedStream::handshake 的第一行，这时已经在每连接的任务里，不在 accept 循环里了。默认容量是 DEFAULT_L4_READ_BUFFER_SIZE（64 KiB，注释说 TLS 层按 16k 的记录读，L4 读缓冲能省很多系统调用）和 DEFAULT_L4_WRITE_BUFFER_SIZE（1460，一个典型 MSS，「something similar to Nagle's algorithm, but user space can control when to flush」）。0.9.0 起每个监听地址可以用 L4BufferSettings 改这两个数（c0845a8），Some(0) 表示这个方向不缓冲。

上游连接的 Stream 也是 From<TcpStream> 建的（l4::connect 里的 socket.into()），但连接器那边没有任何地方调 set_buffer。Stream 被丢弃时 Drop 会打一行 debug 日志，把整个 BufStream 用 Debug 打出来，正好能看到两边的容量。代理跑一轮 curl，按连接方向整理日志里的 buffer: 字段：

Dropping socket ... addr: 127.0.0.1:18300, peer: 127.0.0.1:62930 ... buffer: 0/1460, ... buffer: 0/65536
Dropping socket ... addr: 127.0.0.1:62931, peer: 127.0.0.1:18301 ... buffer: 0/0, ... buffer: 0/0

第一行是下游连接（本地端口 18300），写缓冲 1460、读缓冲 65536；第二行是去上游 18301 的连接，两边都是 0。上游服务的日志也对得上：一个 chunked 请求的请求头、每个块、结尾的 0\r\n\r\n 是分开的几次 read：

[up#1] read 59: POST /b01 HTTP/1.1\r\nHost: a\r\nTransfer-Encoding: chunked\r\n\r\n
[up#1] read 10: 5\r\nhello\r\n
[up#1] read 11: 6\r\n world\r\n
[up#1] read 5: 0\r\n\r\n

为什么 0 能用

容量为 0 的 BufReader 能工作，靠的是 tokio 1.53.2 BufReader::poll_read 开头的旁路：

        // If we don't have any buffered data and we're doing a massive read
        // (larger than our internal buffer), bypass our internal buffer
        // entirely.
        if self.pos == self.cap && buf.remaining() >= self.buf.len() {
            let res = ready!(self.as_mut().get_pin_mut().poll_read(cx, buf));
            self.discard_buffer();
            return Poll::Ready(res);
        }

内部缓冲长度是 0，任何读都算「massive read」，直接读到调用方的缓冲里。BufWriter 同理：buf.len() >= capacity 就直接写。但 AsyncBufRead 那一面是坏的：fill_buf 只能交出内部缓冲，长度永远是 0，调用方会把它当成 EOF。一个只依赖 tokio 的小程序（exp-pingora/proto-bufzero）对 &b"line one\nline two\n"[..] 套一个零容量 BufReader：

let mut r = BufReader::with_capacity(0, &b"line one\nline two\n"[..]);
let n = r.read(&mut buf).await.unwrap();             // buf: [u8; 4]
let filled = r.fill_buf().await.unwrap().len();
let n = r.read_line(&mut line).await.unwrap();
let n = r.read_to_end(&mut rest).await.unwrap();

tokio 1.53.2 上的输出：

read() -> 4 bytes "line"
fill_buf() -> 0 bytes
read_line() -> 0 bytes ""
read_to_end() -> 14 bytes " one\nline two\n"

read_line 读到 0 字节就返回了，数据其实还在。pingora 的 Stream 只走 poll_read，从不调 fill_buf，所以没踩到这一面。

tokio 1.54 与 5b30d71

tokio 的 #8494（dd826ef1，「io: reject zero capacity in BufReader」，首次进入 tokio 1.54.0）就是为了上面那个 EOF：提交说明写「A BufReader built with a capacity of zero holds a zero-length buffer, so poll_fill_buf can never make progress and every AsyncBufRead operation reports end-of-file with Ok instead of returning the data」。它给 BufReader::with_capacity 加了 assert!(capacity > 0, "capacity must be greater than zero")，并给它和 BufStream::with_capacity 都标上 #[track_caller]。同一个小程序换到 tokio 1.54.1：

thread 'main' (5572853) panicked at src/main.rs:6:15:
capacity must be greater than zero

panic 的位置是调用方那一行，所以 pingora 0.9.0 在 tokio 1.54 上报的是 pingora-core-0.9.0/src/protocols/l4/stream.rs:549:26，正是 From<TcpStream> 里 BufStream::with_capacity( 那一列。这个 panic 在代理上的样子和本卷为什么钉 tokio 1.53.2，见怎么读这一卷。

main 上的 5b30d71（2026-10-19，「Keep zero-capacity L4 read buffers working on tokio 1.54」）不再用 BufStream，自己写了一个两态的枚举：

enum BufferedStream {
    UnbufferedRead(BufWriter<RawStreamWrapper>),
    BufferedRead(BufReader<BufWriter<RawStreamWrapper>>),
}

impl BufferedStream {
    fn with_capacity(read: usize, write: usize, stream: RawStreamWrapper) -> Self {
        let writer = BufWriter::with_capacity(write, stream);
        if read == 0 {
            BufferedStream::UnbufferedRead(writer)
        } else {
            BufferedStream::BufferedRead(BufReader::with_capacity(read, writer))
        }
    }
    // ...
}

层次和 BufStream 一样（读缓冲套在写缓冲外面），只是读容量为 0 时干脆不要 BufReader。BufWriter 接受容量 0，不用改。写方向的四个方法直接找到里面的 BufWriter，不经过 BufReader 转发。提交说明的论证是「An empty BufReader never buffered anything here. Stream only reads through poll_read, and BufReader hands a read straight to the stream under it when its own buffer is empty」，也就是上一节的旁路；它还顺带修了 L4BufferSettings::unbuffered()：0.9.0 里这个设置在 tokio 1.54 上同样会 panic，提交加的监听器测试就是用它服务一个请求。git tag --contains 5b30d71 是空的，这个修复还没有发版。

回退：rewind 与 try_peek

有些协议要先看几个字节再决定怎么解析：明文端口上分辨 HTTP/1 和 h2c，TLS 之前读 PROXY protocol 的头。读过的字节要能「退回去」。rewind 把一段字节压到 rewind_read_buf 上：

    pub fn rewind(&mut self, data: &[u8]) {
        if !data.is_empty() {
            self.rewind_read_buf.push(data.to_vec());
        }
    }

poll_read 先看这个栈：

        let result = if !self.rewind_read_buf.is_empty() {
            let data_to_read = self.rewind_read_buf.pop().unwrap(); // safe
            let mut data_to_read = data_to_read.as_slice();
            let result = Pin::new(&mut data_to_read).poll_read(cx, buf);
            // return the remaining data back to the head of rewind_read_buf
            if !data_to_read.is_empty() {
                let remaining_buf = Vec::from(data_to_read);
                self.rewind_read_buf.push(remaining_buf);
            }
            result
        } else {
            Pin::new(&mut self.stream_mut()).poll_read(cx, buf)
        };

它是一个栈：后退回去的先读出来，符合「退回到流的最前面」的意思。每次 poll_read 只从栈顶那一段里拷，不把几段拼起来，也不顺手再读 socket。try_peek 就是 read_exact 再 rewind：

    async fn try_peek(&mut self, buf: &mut [u8]) -> std::io::Result<bool> {
        use tokio::io::AsyncReadExt;
        self.read_exact(buf).await?;
        // rewind regardless of what is read
        self.rewind(buf);
        Ok(true)
    }

read_exact 要凑满 buf 才返回。HTTP 层的 h2c 检测用它读 24 字节的 h2 连接前言，比 24 字节短的明文请求会因此卡住，实验见 HTTP/2。TLS 之前的回调 PreTlsProcess 拿到的就是这个 l4::Stream 的 &mut，可以读完 PROXY 头再把多读的部分 rewind 回去。hyper-util 做 h2c 检测时用的是一个独立的 Rewind 包装类型，见 auto。

exp-pingora/proto-main 的 rewind 子命令在一条真实的 TCP 连接上验证这个顺序：对端 100 ms 后发 hello，本端先 rewind(b"abc")、再 rewind(b"xyz")，try_peek 2 个字节，然后用 16 字节的缓冲连读 4 次。

st.rewind(b"abc");
st.rewind(b"xyz");
let mut peek = [0u8; 2];
let ok = st.try_peek(&mut peek).await.unwrap();
for i in 0..4 {
  let mut b = [0u8; 16];
  let n = st.read(&mut b).await.unwrap();
  // ...
}

摘要：连接的身份和时间

日志和钩子里要知道「这个请求从哪来、连接建了多久、TLS 用的什么」，这些都放在 Digest 里：

pub struct Digest {
    /// Information regarding the TLS of this connection if any
    pub ssl_digest: Option<Arc<SslDigest>>,
    /// Timing information
    pub timing_digest: Vec<Option<TimingDigest>>,
    /// information regarding the CONNECT proxy this connection uses.
    pub proxy_digest: Option<Arc<ProxyDigest>>,
    /// Information about underlying socket/fd of this connection
    pub socket_digest: Option<Arc<SocketDigest>>,
}

每一层连接各自实现 Get*Digest，HTTP/1 的 HttpSession::new 和 h2 的握手各把它们收集一次，存成会话的 Digest。timing_digest 是 Vec：l4::Stream 推一项，TLS 层再推一项（GetTimingDigest 的 Vec::with_capacity(2) 注释写「expect to have both L4 stream and TLS layer」）。

SocketDigest 只存 fd，peer_addr、local_addr、original_dst 都是 OnceCell，第一次问时才做系统调用。监听器 accept 时已经拿到对端地址，直接填进 peer_addr；local_addr 留着，源码里一条 TODO 说以后可以省掉这次查询。tcp_info 对 TCP 连接调 getsockopt(TCP_INFO)，但这一份只在 Linux 上是真的：其他 Unix 上的 get_tcp_info 返回一个全零的 TCP_INFO，set_keepalive 也是空函数。实验代理在 response_filter 里把下游连接的摘要写进一个响应头：

x-l4: peer=Some("127.0.0.1:59468") rtt=Some(0) est=Some(None) tls=None

macOS 上 tcpi_rtt 是 0，不是 None：调用方分不出「没测到」和「真的是 0」。est 是 TimingDigest 的 establishment_duration（0.9.0 新加），下游连接上是 None：这个字段只在连接器建连时填，注释也写了「None means that the producer did not measure this operation」。

Drop 里还有一个小检查：先对 socket 调一次 nodelay()（UDS 调 local_addr()）。返回 EBADF（errno 9）说明 fd 已经被别人关了，打一条 Crit: socket ... is being double closed；正常的话用 now_or_never() 试着把写缓冲 flush 掉，因为 Drop 里不能 .await。

动脑筋答案

try_peek -> true "xy"
read #0: 2 bytes "xy"
read #1: 1 bytes "z"
read #2: 3 bytes "abc"
read #3: 5 bytes "hello"

• try_peek 的 read_exact 要 2 个字节：弹出栈顶 xyz，拷走 xy，剩下的 z 压回去；然后 rewind(b"xy") 再压一层。栈从底到顶是 abc、z、xy。
• 4 次 read 各弹一段：xy、z、abc，每次只拷一段，所以 16 字节的缓冲每次都没读满。
• 栈空了才读 socket，读到对端 100 ms 后发来的 hello。

小结

• HTTP 层看到的是 Box<dyn IO>；l4::Stream 是 RawStream（TCP / UDS / 虚拟 socket）→ RawStreamWrapper（Linux 接收时间戳）→ BufStream 三层，外加回退栈和摘要。
• From<TcpStream> 建 0/0 的 BufStream，下游连接在 handshake 里由 set_buffer 换成 64 KiB 读、1460 写；上游连接从不调 set_buffer，实测一直是 0/0。这个分工从 0.5.0 的 ef234f5 开始。
• 0 能用是因为 tokio BufReader 在内部缓冲为空时把读直接交给下层；fill_buf 那一面会报假 EOF，这正是 tokio #8494 在 1.54.0 里加 assert 的原因。5b30d71 用一个两态枚举替掉 BufStream，未发版。
• rewind 是栈，后退的先读，每次 poll_read 只交一段；try_peek = read_exact + rewind，凑不满就一直等。
• SocketDigest 的地址懒加载；TCP_INFO、keepalive、接收时间戳只在 Linux 上有实现，macOS 上 tcp_info() 返回全零。