一次被 ws.read() 卡死的发送:同步 WebSocket 客户端的"读阻塞 vs 写饥饿"之殇

[收] 42["join_ack",{"code":1,...}]          ← join 成功
[发] 42 ["preset_chat_message",{...}]       ← 预设消息正常
[发] 42 ["get_text2Img_remaining_times"]    ← 正常
[发] 42 ["chat",{"chat_type":"chat_v3",...}] ← 消息发出(入队)
(此后 20 秒内服务器零帧)
[超时] AI 未开始回复

impl<S> WebSocket<S> {
    pub fn send(&mut self, msg: Message) -> Result<()>;
    pub fn read(&mut self) -> Result<Message>;   // ← 阻塞
}

┌─────────────┐   mpsc channel   ┌─────────────────────┐
│ 任意调用线程 │ ───────────────▶ │        读线程        │
│  send_event  │                  │  rx.recv_timeout()  │ ① 取待发消息
└─────────────┘                  │  ws.send(msg)        │ ② 发出
                                 │  ws.read()           │ ③ 读入站 ← 阻塞点
                                 └─────────────────────┘

fn read_loop(inner: Arc<CloudInner>, mut ws: Ws, rx: mpsc::Receiver<Message>) {
    'outer: loop {
        if inner.stopping.load(Ordering::Acquire) {
            break;
        }
        // ① 先处理积压的待发消息(最多等 100ms,连续 16 条封顶)
        let mut sent = 0;
        loop {
            match rx.recv_timeout(Duration::from_millis(100)) {
                Ok(msg) => {
                    if let Err(e) = ws.send(msg) {
                        info!("发送失败: {e}");
                        break 'outer;
                    }
                    sent += 1;
                    if sent >= 16 {
                        break;
                    }
                }
                Err(mpsc::RecvTimeoutError::Timeout) => break,
                Err(mpsc::RecvTimeoutError::Disconnected) => break 'outer,
            }
        }
        // ② 读取入站消息 —— 这里会无限阻塞
        match ws.read() {
            // ...
            Err(e) => {
                info!("云存储读取结束: {e}");
                break;
            }
        }
    }
    drop(rx);
    on_connection_lost(inner);
}

t=0   服务器静默(握手帧已发完,下一帧是 25s 后的 ping)
t=0.1 调用线程把 chat 帧投入 channel
t=0.1 读线程正阻塞在 ws.read()  ← chat 帧躺在 channel 里
t=20  等待超时,测试失败
t=25  服务器发 ping,read() 终于返回,chat 帧这才发出去(为时已晚)

// 发送线程
let mut ws = ws_mutex.lock().unwrap();
ws.send(msg);          // ← 读线程正持锁 read(),这里会等锁

pub enum MaybeTlsStream<S: Read + Write> {
    Plain(S),                                  // 裸 TCP
    Rustls(rustls::StreamOwned<rustls::ClientConnection, S>),  // TLS
}

/// 设置 WebSocket 底层流的读取超时(Plain 或 rustls 两种形态)。
fn set_stream_read_timeout(stream: &mut WsStream, timeout: Duration) -> std::io::Result<()> {
    match stream {
        MaybeTlsStream::Plain(s) => s.set_read_timeout(Some(timeout)),
        MaybeTlsStream::Rustls(owned) => owned.sock.set_read_timeout(Some(timeout)),
        _ => Ok(()),
    }
}

let (mut ws, response) = connect(request)?;
// WebSocket 升级成功返回 HTTP 101 Switching Protocols
if response.status() != http::StatusCode::SWITCHING_PROTOCOLS {
    return Err(CloudError::Handshake(format!("HTTP 状态: {}", response.status())));
}
// 设置底层流读取超时:read 周期性苏醒,避免服务器静默时发送通道饥饿
let _ = set_stream_read_timeout(ws.get_mut(), Duration::from_millis(200));

match ws.read() {
    // ... 正常帧分支略 ...
    // 读取超时:回到循环顶部处理待发送消息
    Err(tungstenite::Error::Io(ref e)) if e.kind() == std::io::ErrorKind::WouldBlock => {
        continue;
    }
    Err(e) => {
        info!("云存储读取结束: {e}");
        break;
    }
}

① rx.recv_timeout(100ms) 处理待发消息
② ws.read()              最多阻塞 200ms
   ├─ 有帧 → 处理,回 ①
   └─ WouldBlock → 回 ①   ← 新路径:哪怕服务器静默,每 200ms 回来一次

[云变量] 私有 4 公有 9 列表 1, 在线 1          ← 只读流程正常
[AI回复] > 思考如何满足用户需求……              ← chat 帧成功发出并收到流式回复
test result: ok. 2 passed; 0 failed           ← 两个模块真机测试全过