服务与监听器

主线位置：服务主线的「add_service → 每个服务 start_service」，也是请求主线的第一步「监听器 accept → TransportStack 建 Stream、set_buffer」。一次 curl 到达代理时，它先经过本页的 accept 循环，再被交给 ServerApp::process_new，后面的 HTTP/1 解析见 HTTP/1 服务端。

sequenceDiagram
  participant S as Server run
  participant L as Service start
  participant A as accept 循环
  participant C as 连接任务
  participant P as ServerApp
  S->>L: run_service 起运行时
  L->>L: Listeners build 绑定或接手 fd
  L->>A: spawn run_endpoint
  A->>A: accept 得到零缓冲 Stream
  A->>C: current_handle spawn
  C->>C: handshake set_buffer TLS
  C->>P: process_new
  P-->>C: Some stream 复用
  C->>P: process_new 下一个请求

代码在 pingora-core/src/services/（mod.rs 738 行，大半是依赖图的测试；listening.rs 339 行；background.rs 120 行）和 listeners/（mod.rs 800 行、l4.rs 789 行，TLS 子目录另算）。实验在 exp-pingora/server-lab（M2 Max、macOS 27、rustc 1.99.0，release，tokio 1.53.2）。

三个 trait

pingora 的「服务」是「跑到停机为止的东西」，模块文档原话是「A service to the pingora server is just something runs forever until the server is shutting down」。0.9.0 里有三个 trait：

• ServiceWithDependents：Server 真正调用的接口。start_service 多一个参数 ready_notifier，服务自己决定什么时候宣布就绪。
• Service：简化版，没有就绪通知。一个 blanket impl 把所有 Service 变成 ServiceWithDependents，做法是一进来就宣布就绪，再调 start_service。
• BackgroundService：不监听端口的后台任务，用 background_service 包成 GenBackgroundService，名字前面自动加 BG ，线程数固定为 Some(1)。

就绪通知是一个 watch::Sender<bool>。ServiceReadyNotifier 的 Drop 里发 true，notify_ready 只是 drop(self)：

impl Drop for ServiceReadyNotifier {
    /// In the event that the notifier is dropped before notifying that the
    /// service is ready, we opt to signal ready anyway
    fn drop(&mut self) {
        // Ignore errors - if there are no receivers, that's fine
        let _ = self.sender.send(true);
    }
}

所以一个服务提前返回或者 panic，依赖它的服务也会被放行，不会永远卡住。依赖关系存在一个 daggy 的有向无环图里：ServiceHandle::add_dependency 加边时如果成环，只打一条 error 日志，不 panic。run 按拓扑序逐个调 run_service，每个服务先在自己的运行时上等完所有依赖的 watch，再 start_service。实验 svc 加了一个监听服务 lab 和一个后台服务 watcher，日志：

[   0.000s INFO pingora_core::server] Starting services in dependency order: ["BG watcher", "lab"]
[   0.000s INFO pingora_core::server] Starting service: BG watcher
[   0.000s INFO pingora_core::server] Starting service: lab

服务名同时是它运行时线程的名字。用 macOS 的 sample 看 threads: 4 的进程，能看到四个 lab、一个 BG watcher、一个 Server，线程数的完整统计见运行时。

监听器：绑定，或者接手旧进程的 fd

listening::Service 持有一个 Listeners，后者是一串 TransportStackBuilder：每个端点一份地址（TCP 或 Unix socket）、可选的 TlsSettings、一份 L4BufferSettings。add_tcp、add_tls、add_uds 最后都落到 add_listener，0.9.0 新增的 ListenerConfig 可以给单个端点配缓冲区大小（c0845a8，「Add per-listener L4 buffer configuration」）。

start_service 里的 listeners.build(fds) 对每个端点调 ListenerEndpointBuilder::listen，fds 是所有服务共享的监听 fd 表（键是地址字符串）：

            let _guard = addr_lock.lock().await;

            let existing_fd = fds_table.lock().get(addr_str).copied();

            if let Some(fd) = existing_fd {
                from_raw_fd(&listen_addr, fd)?
            } else {
                let listener = bind(&listen_addr).await?;
                fds_table
                    .lock()
                    .add(addr_str.to_string(), listener.as_raw_fd());
                listener
            }

• 表里有这个地址（平滑升级时旧进程传过来的），就 from_raw_fd 接手，再调一次 listen 把 backlog 改成 65535。
• 没有就自己 bind，并把新 fd 记回表里：下次 SIGQUIT 时，发给新进程的就是这张表。
• BIND_LOCKS 是一张按地址分的 tokio 互斥锁表，防止两个服务同时「查表没有、各自 bind」。

bind_tcp 总是设 SO_REUSEADDR；bind 失败如果是 AddrInUse，打一条 is in use, will try again，睡 1 秒重试，最多 30 次（TCP_LISTENER_MAX_TRY）。地址解析用的是同步的 to_socket_addrs，注释写着「this could invoke a blocking network lookup」。

Service::listen_addresses 返回这个服务认领的地址。0.9.0 的 f82478a（「Close unclaimed inherited listening sockets on graceful upgrade」）用它在升级后关掉旧进程传来、但新进程没有任何服务认领的 fd：提交说明解释，开了 SO_REUSEPORT 时这种没人 accept 的 socket 会「black-hole connections」。只要有一个服务返回 None（trait 的默认值），整个清理就跳过。

accept 循环

start_service 为每个端点 spawn listener_tasks_per_fd 个 run_endpoint：

        loop {
            let new_io = tokio::select! { // TODO: consider biased for perf reason?
                new_io = stack.accept() => new_io,
                shutdown_signal = shutdown.changed() => {
                    // ...
                            info!("Shutting down {}", stack.as_str());
                            break;
                    // ...
                }
            };
            match new_io {
                Ok(io) => {
                    let app = app_logic.clone();
                    let shutdown = shutdown.clone();
                    current_handle().spawn(async move {
                        let peer_addr = io.peer_addr();
                        match timeout(Duration::from_secs(60), io.handshake()).await {
                            Ok(handshake) => {
                                match handshake {
                                    Ok(io) => Self::handle_event(io, app, shutdown).await,
                                    // ...

• 停机信号一到，循环 break，监听 socket 随 TransportStack 一起被 drop。平滑升级与停机里 SIGTERM 之后 0.3 秒的新连接得到 curl_exit=7（connection refused），就是这里。
• accept 本身只做最少的事：Listener::accept 把 TcpStream 转成 Stream、填上对端地址；apply_stream_settings 对每个连接设 TCP_NODELAY，再按 TcpSocketOptions 设 keepalive、DSCP、收发缓冲区。开了 connection_filter feature 时，被过滤掉的连接在这里直接 drop，不进入 TLS 握手。
• 剩下的工作都在新 spawn 的连接任务里：UninitializedStream::handshake 先 set_buffer，有 TLS 时再跑 pre-TLS 回调和 TLS 握手，整个握手限时 60 秒，超时只记一条 Downstream handshake timeout。
• accept 返回错误码 24（EMFILE，fd 用完）时睡 1 秒，因为这时 accept 会不停地立刻返回同一个错误。

current_handle() 在 Steal 运行时上就是当前运行时，在 NoSteal 上是随机挑一个线程；这一点决定了连接落在哪个线程上，见运行时。

为什么先建零容量缓冲区

Stream 里是一个 BufStream（读缓冲叠写缓冲）。accept 时 From<TcpStream> for Stream 建的是两个容量都为 0 的缓冲：

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

到了连接任务里，set_buffer 再把裸流取出来，套进真正大小的 BufStream：

    /// 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);
    }

理由就在注释里：accept 循环是一个任务串行地接所有连接，默认的 64 KiB 读缓冲（DEFAULT_L4_READ_BUFFER_SIZE，注释说 TLS 层总是按 16k 的记录读，大的 L4 读缓冲能省系统调用）和 1460 字节的写缓冲（一个典型 MSS）如果在 accept 时分配，这次 malloc 就落在所有新连接都要排队经过的路径上。挪到连接任务里以后，分配发生在连接自己的线程上，和 accept 并行。BufStream 没有调整容量的 API，只能拆开重建，所以字段是 Option。

这也是 tokio 1.54 让 pingora 0.9.0 panic 的位置：tokio 1.54.0 的 #8494 让 BufReader::with_capacity(0, ..) 直接 assert!，accept 时那个零容量的读缓冲就过不去了。来龙去脉见怎么读这一卷，Stream 本身的结构见 L4 流。同样的问题还藏在另一个地方：L4BufferSettings 的文档说「Setting either side to Some(0) disables BufStream buffering」，unbuffered() 就是 new(0, 0)；在 tokio 1.54 上，这个配置会让 set_buffer 里的 BufStream::with_capacity(0, ..) 同样 panic。main 上的修复 5b30d71 两处一起改了。

连接交给 ServerApp

握手完成后，handle_event 调 ServerApp::process_new；返回 Some(stream) 表示连接还能复用，就在同一个任务里再调一次：

    pub async fn handle_event(event: Stream, app_logic: Arc<A>, shutdown: ShutdownWatch) {
        debug!("new event!");
        let mut reuse_event = app_logic.process_new(event, &shutdown).await;
        while let Some(event) = reuse_event {
            // TODO: with no steal runtime, consider spawn() the next event on
            // another thread for more evenly load balancing
            debug!("new reusable event!");
            reuse_event = app_logic.process_new(event, &shutdown).await;
        }
    }

那条 TODO 说的是：在 NoSteal 运行时上，一个 keep-alive 连接从第一个请求到最后一个请求都在同一个线程上。

所有 HttpServerApp（HttpProxy、ServeHttp 的实现）通过一个 blanket impl 得到 ServerApp::process_new：按 TLS 的 ALPN 或明文的 h2c 前言选 HTTP/2，否则建 HTTP/1 会话；如果此时停机信号已经是 true，新会话直接关掉 keep-alive，否则 keep-alive 设成 60 秒。HTTP/1 的会话在这个函数内部循环复用，HTTP/2 的每个流再 current_handle().spawn 成独立任务。协议细节在 HTTP/1 服务端和 HTTP/2。

动脑筋答案

第二个进程每秒重试一次，第 29 秒放弃。放弃时只有 lab 这个服务的线程 panic，进程不退出，直到被 runq 在第 38 秒 SIGKILL（release）：

[  26.073s WARN pingora_core::listeners::l4] 127.0.0.1:18200 is in use, will try again
[  27.075s WARN pingora_core::listeners::l4] 127.0.0.1:18200 is in use, will try again
[  28.079s WARN pingora_core::listeners::l4] 127.0.0.1:18200 is in use, will try again

thread 'lab' (5572571) panicked at ~/.cargo/registry/src/index.crates.io-1949cf8c6b5b557f/pingora-core-0.9.0/src/services/listening.rs:293:14:
Failed to build listeners: Error { etype: BindError, esource: Unset, retry: Decided(false), cause: Some(Os { code: 48, kind: AddrInUse, message: "Address already in use" }), context: Some(Owned("bind() failed, after retries, 127.0.0.1:18200 still in use")) }
note: run with `RUST_BACKTRACE=1` environment variable to display a backtrace
[runq: killed after 38s]

• 30 次尝试之间睡 29 次 1 秒，所以最后一条 warn 在第 28 秒，第 29 秒报错。两边都设了 SO_REUSEADDR，但 macOS 上它不允许两个 socket 同时监听同一个地址，bind 返回 EADDRINUSE（errno 48）。
• panic 来自 start_service 里的 .expect("Failed to build listeners")。它发生在服务运行时的一个任务里，tokio 把任务里的 panic 截在 JoinHandle 里，Server 的主线程仍然在 main_loop 里等信号。结果是一个活着、但没有监听任何端口的进程；只看进程是否存在的监控会认为它正常。
• 如果旧进程在这 29 秒内退出，新进程会在下一次重试时绑上，这就是 macOS 上能用的「土办法升级」，平滑升级与停机里量了它的空窗。

小结

• Server 只认 ServiceWithDependents；Service 一进来就宣布就绪，BackgroundService 固定 1 个线程；就绪通知 drop 即发，依赖方不会卡死。
• 监听器先查共享 fd 表，有就接手旧进程的 fd，没有就 bind 并把 fd 记回表里；AddrInUse 每秒重试，30 次后服务线程 panic，进程不退出。
• accept 循环只 accept 和设 TCP_NODELAY，握手（set_buffer、TLS，限时 60 秒）放在每个连接自己的任务里。
• 零容量缓冲是 0.5.0 为了不在 accept 路径上 malloc 加的；tokio 1.54 的断言让它和 L4BufferSettings::unbuffered() 一起 panic。
• NoSteal 上所有 accept 任务都在 start_service 取到的同一个线程上，listener_tasks_per_fd 不能把它们分开。