rust 版本的 peerstore 落地实践

模块地址： github.com/netwarps/li…

在rust-libp2p中，当协议想要获取peer_id所对应的地址时，需要实现NetworkBehaviour的addresses_of_peer方法。与之不同的是，go-libp2p使用peerstore来存储了peer_id与address之间的关系，因此我们可以参考它来实现一个rust版本的peerstore。

实现构想

首先，由于我们的启动核心是swarm，那peerstore会作为其中的一个属性。其次，在go-libp2p-peerstore中，peerstore主要存放的数据为三块：地址信息的AddrBook，公私钥信息的KeyBook，协议信息的ProtoBook；我们可以将三者结合在一起组成一个新的struct，取名为PeerRecord，放在以peer_id为key的Hashmap中。

垃圾回收机制

GC存在的意义就是防止hashmap过度膨胀。由于网络状态时时刻刻都在发生变化，peer之间的连接也可能随之变化，而peerstore有一个重要的作用就是存放peer的地址信息。如果不对已经失效的peer信息进行清理，就会影响到peerstore的工作效率。

目前实现的效果是，在swarm的start方法中使用task::spawn启动一个任务，创建一个mpsc的channel，使用select语法等待管道传来的消息或者 task 等待10分钟的逻辑完成，针对某些地址，如果已经超出ttl的限制，清理当前地址；同时，如果当前peer_id的地址集合中不存在任何的地址信息，就将其从Hashmap中移除。

Pinned

虽然GC机制的存在，使Hashmap不会无限制扩容，良好地帮助了系统的运行。但是仍然有些不足的地方，考虑如下这种情况：

对于KAD协议来说，peer需要不停地进行迭代查询已知节点，逐步填充自己的KBucket，这是一个耗时的过程。如果在gc时，将较早查询到的节点地址信息从peerstore中移除了，那么又需要重新启用迭代查询去获取地址，因此我们在PeerRecord中添加了一个bool值pinned。GC时会判断这条记录的类别，如果pinned为true，就会跳过清理的步骤。

序列化与持久化

对每一个peer来说，因为某些原因需要下线或停机时，存放在peerstore里的节点信息是不应该被丢弃的。而对于需要持久化的数据，也需要进行序列化操作，便于存放。

在libp2p-rs中，主循环的调用也是通过task::spawn启动的。当swarm接收到close的消息时，将会退出事件处理的循环，并向运行peerstore的gc线程发送一个close()的事件，结束gc的过程。接下来调用peerstore的save_data()方法，将数据使用serde序列化成json格式，并使用std::io将序列化后的数据存放到根目录的txt文件中。

方法分析

以GC方法进行解析：

1. swarm主循环spawn运行task，每十分钟触发一次select。
2. Hashmap被Arc包裹，可以通过lock()获取，保证并发安全。
3. 如果该peer的信息不是通过kad获取的，调用retain筛选未超出ttl时限的地址。
4. 如果当前peer的地址数据已经清空，从hashmap中移除这个peer。

    // swarm/lib.rs
    // The GC task is to remove all expired addresses from the peer store
        task::spawn(async move {
            log::info!("starting Peerstore GC...");
            loop {
                let either = future::select(rx.next(), task::sleep(PEERSTORE_GC_PURGE_INTERVAL).boxed()).await;
                match either {
                    Either::Left((_, _)) => break,
                    Either::Right((_, _)) => peer_store.remove_expired_addrs(),
                }
            }
            log::info!("quitting Peerstore GC...");
        });
        
    // core/peerstore.rs
    /// Removes all expired address.
    pub fn remove_expired_addrs(&self) {
        let mut to_remove = vec![];
        let mut guard = self.inner.lock().unwrap();
        for (peer, pr) in guard.iter_mut() {
            if !pr.pinned {
                log::debug!("GC attempt for {:?}", peer);
                pr.addrs.retain(|record| record.expiry.elapsed() < record.ttl);
                // delete this peer if no addr at all
                if pr.addrs.is_empty() {
                    log::debug!("remove {:?} from peerstore", peer);
                    to_remove.push(peer.clone());
                }
            }
        }

        for peer in to_remove {
            guard.remove(&peer);
        }
    }

Netwarps 由国内资深的云计算和分布式技术开发团队组成，该团队在金融、电力、通信及互联网行业有非常丰富的落地经验。Netwarps 目前在深圳、北京均设立了研发中心，团队规模30+，其中大部分为具备十年以上开发经验的技术人员，分别来自互联网、金融、云计算、区块链以及科研机构等专业领域。Netwarps 专注于安全存储技术产品的研发与应用，主要产品有去中心化文件系统（DFS）、去中心化计算平台（DCP），致力于提供基于去中心化网络技术实现的分布式存储和分布式计算平台，具有高可用、低功耗和低网络的技术特点，适用于物联网、工业互联网等场景。公众号：Netwarps