About libp2p_stream and quic stream(Need help)

[ISensuiI]

About libp2p_stream and quic stream

1. First of all, I just learned rust and libp2p not long ago. I plan to use libp2p to build a collaboration tool. The problem I encounter now is that the delay between my two computers in the LAN is 1ms. When I use libp2p to connect ping or transmit udp data packets, the delay is 20-30ms, which is unacceptable. I want to reduce the delay to about 5ms. I understand that using libp2p itself has performance loss. But 20-30ms is too high.

2. The problem is that I have several ideas now:

I use opentelemetry to track performance analysis,

pub async fn read_packet(stream: &mut Stream) -> Result<(TLVProtocol, Vec<u8>)> {
// Total time starts
let total_start = Instant::now();
info!(target: "app::net::p2p::stream::packet", "Stream Start reading data");

// 1. Create BufReader
let buf_reader_start = Instant::now();
let mut buffered = BufReader::with_capacity(16384, stream); // 16KB buffer
let buf_reader_time = buf_reader_start.elapsed();
info!(target: "app::sys::performance", "1. BufReader creation time: {:?}", buf_reader_time);

// 2. Buffer allocation
let alloc_start = Instant::now();
let mut buffer = BytesMut::with_capacity(4096);
let alloc_time = alloc_start.elapsed();
info!(target: "app::sys::performance", "2. Buffer allocation time: {:?}", alloc_time);

// 3. Header resize
let resize_header_start = Instant::now();
buffer.resize(4, 0);
let resize_header_time = resize_header_start.elapsed();
info!(target: "app::sys::performance", "3. Header adjustment time: {:?}", resize_header_time);

// 4. Header reading
let read_header_start = Instant::now();
match buffered.read_exact(&mut buffer[..4]).await {

match buffered.read_exact(&mut buffer[..4]).await The delay of this sentence will fluctuate between 10ms-100ms.

pub const TLV_PROTOCOL: StreamProtocol = StreamProtocol::new("/tlv/1.0.0");

let incoming_streams = swarm
.behaviour_mut()
.stream()
.new_control()
.accept(tlv_protocol::TLV_PROTOCOL)?;

My stream is built based on libp2p_Stream lib. Here is to get the stream controller.

stream_result = incoming_streams.next() => {
match stream_result {
Some((peer, stream)) => {

Mode::Dial(_) => {
match control.open_stream(open_peer_id, tlv).await {

Here I open the stream by listening and dialing.

Then I operate the stream to transmit my application data between two nodes.

My question now is, in addition to using libp2p_Stream lib to build custom streams to transmit data, is there a more efficient way to transmit data, such as based on quic_stream.

Because there are too few tutorials related to libp2p, and most of the docs.rs do not have examples, I don’t know what to do now. Is to find another stream to replace my original libp2p_Stream lib, I'd better focus on solving the match buffered.read_exact(&mut buffer[..4]).await problem, which can help me solve the latency problem.

Another problem is that I use iperf3 to test the performance of my two nodes. When they are directly connected, they can reach 30-40Mbps, but when using libp2p, they are only 2Mbps, and packets will be lost as long as they exceed 2M.

Based on the above reasons, I have been wondering if my stream construction method is incorrect.

Below is my Behaviour build and swarm build

pub fn new(
        keypair: &libp2p::identity::Keypair,
        relay_behaviour: relay::client::Behaviour,
    ) -> Result<Self> {
        
        let message_id_fn = |message: &gossipsub::Message| {
            let mut s = DefaultHasher::new();
            message.data.hash(&mut s);
            gossipsub::MessageId::from(s.finish().to_string())
        };
        
        let gossipsub_config = gossipsub::ConfigBuilder::default()
            .heartbeat_interval(Duration::from_secs(10)) // This is set to aid debugging by not cluttering the logs space
            .validation_mode(gossipsub::ValidationMode::Strict) // This sets the kind of message validation. The default is Strict (enforce message signing)
            .message_id_fn(message_id_fn) // content-address messages. No two messages of the same content will be propagated.
            .build()
            .map_err(|msg| std::io::Error::new(std::io::ErrorKind::Other, msg))?; // Temporary hack because "build" does not return a proper "std::error::Error".
        
        let gossipsub = gossipsub::Behaviour::new(
            gossipsub::MessageAuthenticity::Signed(keypair.clone()),
            gossipsub_config,
        )
        .map_err(|e| anyhow::anyhow!("创建 gossipsub 网络行为错误: {}", e))?;

        let relay_client = relay_behaviour;
        let ping = ping::Behaviour::new(ping::Config::new());
        let identify = identify::Behaviour::new(identify::Config::new(
            GlobalState::global_config().get_global_identify_name()?,
            keypair.public(),
        ));
        let dcutr = dcutr::Behaviour::new(keypair.public().to_peer_id());

        let mdns =
            mdns::tokio::Behaviour::new(mdns::Config::default(), keypair.public().to_peer_id())?;

        let stream = libp2p_stream::Behaviour::new();
        
        Ok(ClientBehaviour {
            relay_client,
            ping,
            identify,
            dcutr,
            gossipsub,
            mdns,
            stream,
        })
    }




let mut swarm =
        libp2p::SwarmBuilder::with_existing_identity(helper::generate_ed25519(secret_key_seed))
            .with_tokio()
            .with_tcp(
                tcp::Config::default().nodelay(true),
                noise::Config::new,
                yamux::Config::default,
            )?
            // .with_quic_config(|mut config| {
            //     config.keep_alive_interval = Duration::from_secs(5);
            //     config.max_idle_timeout = Duration::from_millis(10_000);
            //     config.max_concurrent_stream_limit = 256;
            //     // 设置其他字段...
            //     config
            // })
            .with_quic_config(|mut config| {
                // 禁用对 QUIC 协议草案 29 版本的支持
                // 只使用 QUIC 版本 1，保证兼容性和稳定性
                config.support_draft_29 = false;

                // 设置初始握手的超时时间为 5 秒
                // 如果在 5 秒内无法完成握手，连接将被终止，防止握手过程卡住
                config.handshake_timeout = Duration::from_secs(10);

                // 设置最大空闲超时时间为 10,000 毫秒（10 秒）
                // 如果连接在 10 秒内没有任何活动，它将被关闭，以释放资源
                config.max_idle_timeout = 10 * 1000;

                // 设置最大并发双向流的数量为 256
                // 限制单个连接可以同时打开的流的数量，防止资源耗尽
                config.max_concurrent_stream_limit = 1024;

                // 设置保活包发送间隔为 5 秒
                // 每 5 秒发送一个保活包，以保持连接活跃，防止因空闲而被关闭
                config.keep_alive_interval = Duration::from_secs(5);

                // 设置单个连接的最大数据量为 50 MB
                // 限制单个连接可以传输的总数据量，防止单个连接占用过多资源
                config.max_connection_data = 100_000_000;

                // 设置单个流的最大数据量为 30 MB
                // 限制单个流可以传输的数据量，确保资源在多个流之间公平分配
                config.max_stream_data = 100_000_000;

                config

            })
            .with_dns()?
            .with_relay_client(noise::Config::new, yamux::Config::default)?
            .with_behaviour(|keypair, relay_behaviour| {
                let behaviour = behaviour_build::ClientBehaviour::new(keypair, relay_behaviour)?;
                Ok(behaviour)
            })?
            .with_swarm_config(|c| {
                c.with_idle_connection_timeout(Duration::from_secs(60))
                    .with_notify_handler_buffer_size(NonZeroUsize::new(32).expect("eeee"))
                    .with_per_connection_event_buffer_size(usize::from(NonZeroUsize::new(64).expect("eeee")))
            })
            .build();

    swarm.listen_on("/ip4/0.0.0.0/udp/0/quic-v1".parse()?)?;
    swarm.listen_on("/ip4/0.0.0.0/tcp/0".parse()?)?;

[ISensuiI]

done
The implementation of ConnectionHandler and netbehaviour has been solved.

[Tianer1123]

兄弟你好，我以前也尝试使用stream来处理数据流量，我记得好像延迟也非常高，带宽测试也非常低，方便讲下你是如何处理的吗？

我现在改为单独起线程，起udp监听口理数据流量。我以为用stream因为有加解密会导致性能很低。看到你解决了，应该是我哪里处理出现问题了，我是基于example中的steam例子写的。

[ISensuiI]

细节我已经记不清了太久了, 这个工具我用了很久接近大半年 期间花了无数精力在上面, 原本是打算在我一个非常重要的产品作为基础网络服务.
这个实现网络有2个方向, 第一个使用 libp2p stream 库来实现,相对来说简单, example中就有例子, 简单来说就是 2个 node 开启 双向流 互相读写数据即可. 但是 经过我使用 工具测试 (名字记不清了 命令行专门测试网速的 ) 整体 读写性能一般宽带只能跑到 2m 左右 就出现丢包.
后来我又花了大概接近半个月左右反复读源码,因为根本没有任何教程,只能看源码. 然后开始意识到只能自己重新实现 behaviour + substream trait, 实现了 多路子流复用. 并且 从双向流改为单向流, 也就是 每2个节点通信使用2个子流一个读一个写. 这时候 虽说整体快了一点 但是 整体延迟很高.(这次重构大概花了我接近半个月时间 , 因为当时AI CLAUDE 也无法理解LIBP2P库内部细节,可能太冷门了,总之就是问的东西非常不靠谱,只能自己大量读源码)
最后延迟问题我之前没接触过这类底层网络封包解包,以及相关东西, 我又花了很多时间看了路由 封包解包知识. 最后发现我自己手动封包解包过程中有问题 优化了很多个模块节点, 具体就不展开了总之就是 根据数据路径 把之前错误使用同步的改为异步, 丢包率从20%,降低到0%. 整体从最高120ms 到最低5ms(直连 1ms 这4ms是库 以及解包封包带来的损耗) , 带宽从 2mbps增加到 20mbps . 我本地两个物理节点 是局域网 直连带宽是 45mbps . 其实还差一点 具体我就没深究 因为 这个带宽已经足够我的项目用了.

期间比较重要的一个问题当延迟30m的时候我已经没地方优化了,.由于我之前是做web开发方面工作,这方面经验比较少, 我使用opentelemetry 作为数据包追踪, 但是由于 opentelemetry 无法多个节点毫秒级对齐(实现起来比较麻烦 我没时间钻研), 我后来又手写了一套简单的数据包追踪机制(防止opentelemetry不准确,我第一次使用otel 怕不精确),简单来说就是自己封包数据,做好时间标记.然后到每个节点进行解包 打印, 最终查看整个链路到底是哪里提高了延迟. 最终发现是 在 stream.read 和 stream.write 也就是libp2p实例中读写部分之前. 最终判定是粘包问题导致的延迟+丢包. 那经过搜索加上分析得到的结果是,读写之前需要缓冲区 + EOF(数据流+帧边界的问题). 我尝试手写,发现自己写的性能并没有多少提升(这也是我更底层网络机制不熟悉,rust当时我也是刚学). 最后找到了 asynchronous_codec, 可以全自动解决缓冲区+EOF问题. 延迟从 30ms 直接降低到 10ms. 最后又优化了一些细节最终基本稳定在3ms-6ms左右浮动. 应该还有优化空间. 但是暂时对于我来说已经足够了.

这就是大概整体的思路和流程, 你可以按照我的思路去排查构建,整体这些我花了很多时间,后续包括mdns,discover,gossipsub,rendezvous,response/request/kad/hdt,我都花了大量的时间去阅读并且尝试,并且已经构建好了一个MVP,可以完成我多个节点传输数据并且同步. 但是在半年之前我放弃了这个库,原因是它太重了,太复杂了,维护起来非常吃力.并且我多方面分析以及对比,它更加适合做p2p项目而不适合实时性要求高的项目.我做的是同步操作+游戏相关的项目. 当然也可能是我太菜了,比如子流部分,我隔了一个多月回过去在进行调整修改,我完全是一脑子雾水,太复杂那部分,我总是要花几个小时才能想起来整个流程应该怎么走然后才是开始修改逻辑.

我认为作为一个基础底层库,有难度可以理解,但是如果每次维护都需要花几个小时去整理思路,我认为作为一个长期维护项目来说这个成本太高了,应该更多的精力花在业务上而不是底层库上.
我曾经接触过C++, JAVA, 深度使用 php python ts rust kotlin 这个libp2p在各个语言中如果你要让我评分的话,我认为他在这些语言中都算最难一档 框架, 所以除非必要 否则的话如果有替代尽量别用, 如果是大神那另当别论.

后续我将libp2p 替换为了 iroh , 这更加符合我的项目,并且iroh经过了很多跟我同类型项目的验证,并且复杂度直线下降. 尽管也有不少问题, 但iroh 我只花了1天就基本上可以使用. 不过这不代表libp2p不好,它仍旧是非常优秀,但是要看你项目类型,以及你愿意花费多少精力在后续维护上.

整体传输通道使用flume,我认为在异步+性能是达到了一个平衡. 如果你追求极致性能. 最近发布的 https://github.com/frostyplanet/crossfire-rs 这个库,根据数据性能比flume还要快, 如果模块多 使用这个应该还能提高1ms左右.

差不多就这样如果你有其他问题再提问吧,这个库作为初学者来说,我应该了解的算比较全面并且深度使用.

[drHuangMHT]

做游戏客户端、服务端这些还是用其他的库比较好。我个人还是觉得libp2p主要关注的是observability，以找到对等节点异步交换数据为目的而非关注低延迟高吞吐量

[Tianer1123]

好的，非常感谢你得回答，我得项目不得不使用这个库，我使用另外起线程专门处理流量相关得功能，性能上也比较理想，消耗主要在流量得加解密上。但这就要求相关设备能互相通信，限制也比较多。

[ISensuiI]

也是一个思路,我当时想的是能用内置的方法尽量用内置的方法 所以自己实现了子流相关功能 你可以都试试看看哪个更好.

[drHuangMHT]

libp2p默认会使用加密的，有一些延迟上的增加很正常。吞吐量的话尽量减少数据复制吧，比如用Box<T>，还有写入stream之后不要频繁flush之类的，2 Mbps是有点少了。如果感兴趣的话可以看看我的项目。

[ISensuiI]

收到 兄弟 谢谢解答, 我也是刚刚学习, 很多地方不明白, substream 部分已经搞懂了, https://github.com/drHuangMHT/OwlNest 你说的是这个项目吗? 其实我还有一些吞吐的问题没搞懂, 延迟基本上解决了. 吞吐还是很低 还没找到原因. flush 我去尝试一下.

[drHuangMHT]

吞吐量可以参考一下里面的blob协议，虽然我觉得也没写的很好就是了，但是吞吐量应该可以提高一个数量级。

[ISensuiI]

收到兄弟, 这东西教程少的可怜, 大多数我都是配合手册 再加上 看源码 花了不少时间, 我这两天把子流管理器写完了 再仔细读一下你的源码
再次感谢.