使用tokio::time::timeout无法捕获TCPReadStream延迟响应的问题

问题 #

我正在编写一个与比特币服务器进行P2P节点握手的PoC。向目标节点发送version消息后，对方会返回对应的version消息，这部分逻辑正常，但偶尔目标节点的version消息需要两分钟以上才会到达。虽然最终能成功响应，但为了和其他网络操作的5秒超时保持一致，我希望在此处也设置超时。

以下是未添加超时检查的代码，可正常运行但有时速度极慢：

use bitcoin::{
    consensus::Decodable,
    p2p::message::{self},
};
use std::{
    io::BufReader,
    net::{IpAddr, SocketAddr, TcpStream},
    time::{Duration, SystemTime},
};

pub async fn handshake(to_address: IpAddr, port: u16, timeout: u64) -> Result<()> {
    let target_node = SocketAddr::new(to_address, port);

    let mut write_stream = TcpStream::connect_timeout(&target_node, Duration::from_secs(timeout))?;

    // Build and send version message
    send_msg_version(&target_node, &mut write_stream)?;

    let read_stream = write_stream.try_clone()?;
    let mut stream_reader = BufReader::new(read_stream);

    let response1 = message::RawNetworkMessage::consensus_decode(&mut stream_reader)?;

    match response1.payload() {
        // Do stuff here
    }

    // More stuff here...
}

其中consensus_decode()调用有时会耗时极久，因此我尝试将其包裹在future::ready()中，再放入tokio::time::timeout()中，代码如下：

let response1 = if let Ok(net_msg) = tokio::time::timeout(
    Duration::from_secs(timeout),
    future::ready(message::RawNetworkMessage::consensus_decode(&mut stream_reader)?),
)
.await
{
    net_msg
} else {
    return Err(CustomError(format!(
        "TIMEOUT: {} failed to respond with VERSION message within {} seconds",
        to_address, timeout
    )));
};

(&mut stream_reader)?部分可以成功捕获解码错误，但如果消息成功到达但耗时过长，tokio::time::timeout却无法捕获这种超时情况。请问我忽略了什么？

解决方案 #

问题根源 #

核心错误有两点：

• 同步IO阻塞线程：你使用的std::net::TcpStream和std::io::BufReader是同步组件，调用consensus_decode时会直接阻塞Tokio的工作线程，直到数据读取完成，期间Tokio无法触发超时逻辑。
• future::ready提前执行同步调用：future::ready(...)会在创建Future的瞬间就执行内部的consensus_decode同步方法，等await时Future已经完成，tokio::time::timeout根本没机会触发超时。

修复步骤 #

1. 替换为异步IO组件：用tokio::net::TcpStream和tokio::io::BufReader替代标准库的同步组件，确保网络操作是非阻塞的。
2. 将同步解码包装为异步任务：由于consensus_decode是同步方法，需要用tokio::spawn_blocking将其放到单独的阻塞线程池执行，避免占用Tokio的异步工作线程，让超时逻辑正常生效。

修改后的代码示例 #

use bitcoin::{
    consensus::Decodable,
    p2p::message::{self},
};
use tokio::{
    io::BufReader,
    net::{IpAddr, SocketAddr, TcpStream},
    time::{self, Duration},
};
use std::error::Error;

// 适配异步TcpStream的send_msg_version实现
async fn send_msg_version(target_node: &SocketAddr, stream: &mut TcpStream) -> Result<(), Box<dyn Error>> {
    // 实现发送逻辑...
    Ok(())
}

pub async fn handshake(to_address: IpAddr, port: u16, timeout: u64) -> Result<(), Box<dyn Error>> {
    let target_node = SocketAddr::new(to_address, port);

    // 异步连接+超时控制
    let mut stream = time::timeout(
        Duration::from_secs(timeout),
        TcpStream::connect(target_node)
    ).await??;

    // 异步发送version消息
    send_msg_version(&target_node, &mut stream).await?;

    let read_stream = stream.try_clone()?;
    let mut stream_reader = BufReader::new(read_stream);

    // 同步解码包装为异步任务+超时控制
    let response1 = time::timeout(
        Duration::from_secs(timeout),
        tokio::task::spawn_blocking(move || {
            message::RawNetworkMessage::consensus_decode(&mut stream_reader)
        })
    ).await??;

    match response1.payload() {
        // 处理消息逻辑...
    }

    // 后续逻辑...
    Ok(())
}

关键说明 #

• tokio::net::TcpStream::connect是异步方法，配合timeout可实现连接超时。
• spawn_blocking将同步解码操作隔离到阻塞线程池，避免影响Tokio事件循环，确保timeout能在指定时间后终止任务。
• await??的用法：第一个?处理超时错误，第二个?处理解码或任务执行的错误。

内容的提问来源于stack exchange，提问作者Chris W