LoginSignup
4
0
@every-monthin株式会社Opening Line

RustのLibp2pでmDNSを使ってピアをみつけてPingしてみた

Last updated at Posted at 2024-05-14

今回の趣旨

P2Pはじめたい！

なので大人気の言語Rustで書かれているほうのLibp2pを使ってまずはかんたんにmDNSで同じマシンのターミナルで動いてるピアを見つけようと思います。

背景の知識の簡単な説明

Tokio

RustのAsyncな関数や入出力や通信を実行するために使うものです。（人気だけで選びました）

mDNS

LAN中でのサービスを見つけるためのものです。
今回は同じマシンのピアを見つけるために使います。
（ゆくゆくはKademliaが使いたいです……）

とにかくやってみる

今回はTokioとmDNSを使いますので、Cargo.tomlはこんな感じにします。

Cargo.toml
[package]
name="qiita-article-libp2p"
version="0.0.0"
edition="2021"

[dependencies.libp2p]
version="^0.53.2"
features=[
	"mdns",
	"noise",
	"tcp",
	"tokio",
	"yamux",
]

[dependencies.tokio]
version="^1.37.0"
features=[
	"rt-multi-thread",
	"macros",
]

src/main.rsはこうしてみます。（こうできるまでが長かった……）

src/main.rs
use libp2p::futures::StreamExt as _;
use std::error::Error;

#[tokio::main]
async fn main(
)->Result<(),Box<dyn Error>>{
	let mut swarm=libp2p::SwarmBuilder::with_new_identity()
		.with_tokio()
		.with_tcp(
			libp2p::tcp::Config::default(),
			libp2p::noise::Config::new,
			||libp2p::yamux::Config::default(),
		)?
		.with_behaviour(
			|keypair|{
				libp2p::mdns::tokio::Behaviour::new(
					libp2p::mdns::Config::default(),
					keypair.public().into(),
				).unwrap()
			}
		)?
		.build();
	swarm.listen_on("/ip4/0.0.0.0/tcp/0".parse()?)?;
	loop{
		println!("{:#?}",swarm.select_next_some().await);
	}
}

まずSwarmはmutにします。あとでlisten_onやselect_next_someする時に必要です。

let mut swarm=libp2p::SwarmBuilder::with_new_identity()

tokio使用のためtokioで、

.with_tokio()

TCPをいい感じに設定します。

.with_tcp(
    libp2p::tcp::Config::default(),
    libp2p::noise::Config::new,
    ||libp2p::yamux::Config::default(),
)?

お目当てのmDNSはデフォのままです。

.with_behaviour(
    |keypair|{
        libp2p::mdns::tokio::Behaviour::new(
            libp2p::mdns::Config::default(),
            keypair.public().into(),
        ).unwrap()
    }
)?

IPv4のアドレスは全部のTCPのポートはおまかせでLISTENします。

swarm.listen_on("/ip4/0.0.0.0/tcp/0".parse()?)?;

イベント全部println!する無限ループにします。

loop{
    println!("{:#?}",swarm.select_next_some().await);
}

結果

cargo runします。無限ループするので２つ動かします。
２つ動かしたものはそれぞれこんなことになっています：

peer id: 12D3KooWKY715KuGyDdQTAL9qhbhvwqhyne7ARGRWWdH7fzhVhjt
NewListenAddr {
    listener_id: ListenerId(
        1,
    ),
    address: "/ip4/127.0.0.1/tcp/40867",
}
NewListenAddr {
    listener_id: ListenerId(
        1,
    ),
    address: "/ip4/10.0.2.15/tcp/40867",
}
Behaviour(
    Discovered(
        [
            (
                PeerId(
                    "12D3KooWJ1tK1vX2yggwwpriAkDaReCjiziY6xzau45u8Ka5ScZw",
                ),
                "/ip4/10.0.2.15/tcp/35731/p2p/12D3KooWJ1tK1vX2yggwwpriAkDaReCjiziY6xzau45u8Ka5ScZw",
            ),
        ],
    ),
)

peer id: 12D3KooWJ1tK1vX2yggwwpriAkDaReCjiziY6xzau45u8Ka5ScZw
NewListenAddr {
    listener_id: ListenerId(
        1,
    ),
    address: "/ip4/127.0.0.1/tcp/35731",
}
NewListenAddr {
    listener_id: ListenerId(
        1,
    ),
    address: "/ip4/10.0.2.15/tcp/35731",
}
Behaviour(
    Discovered(
        [
            (
                PeerId(
                    "12D3KooWKY715KuGyDdQTAL9qhbhvwqhyne7ARGRWWdH7fzhVhjt",
                ),
                "/ip4/10.0.2.15/tcp/40867/p2p/12D3KooWKY715KuGyDdQTAL9qhbhvwqhyne7ARGRWWdH7fzhVhjt",
            ),
        ],
    ),
)

ちゃんとお互いのノードを発見しているようです。

Behaviour(
    Discovered(
        [
            (
                PeerId(
                    "12D3KooWKY715KuGyDdQTAL9qhbhvwqhyne7ARGRWWdH7fzhVhjt",
                ),
                "/ip4/10.0.2.15/tcp/40867/p2p/12D3KooWKY715KuGyDdQTAL9qhbhvwqhyne7ARGRWWdH7fzhVhjt",
            ),
        ],
    ),
)

こんな感じの出力のところに（自分のではなく）相手のIPやポート番号やPeerIdがしっかり出ています。

（一旦）やったね✨🎉

次はPingする

Cargo.tomlにはmacrosとpingを追加です。

Cargo.toml
[package]
name="handson-libp2p-cource-generated-by-ai"
version="0.0.0"
edition="2021"

[dependencies.libp2p]
version="^0.53.2"
features=[
	"macros",
	"mdns",
	"noise",
	"ping",
	"tcp",
	"tokio",
	"yamux",
]

[dependencies.tokio]
version="^1.37.0"
features=[
	"rt-multi-thread",
	"macros",
]

src/main.rsはちょっと変更が多めです。

src/main.rs

use libp2p::futures::StreamExt as _;
use std::error::Error;
use std::time::Duration;

#[derive(libp2p::swarm::NetworkBehaviour)]
struct Behaviour{
	mdns:libp2p::mdns::tokio::Behaviour,
	ping:libp2p::ping::Behaviour,
}

impl Behaviour{

	pub fn new(
		mdns:libp2p::mdns::tokio::Behaviour,
		ping:libp2p::ping::Behaviour,
	)->Self{
		Self{
			mdns,
			ping,
		}
	}
}

#[tokio::main]
async fn main(
)->Result<(),Box<dyn Error>>{
	let mut swarm=libp2p::SwarmBuilder::with_new_identity()
		.with_tokio()
		.with_tcp(
			libp2p::tcp::Config::default(),
			libp2p::noise::Config::new,
			||libp2p::yamux::Config::default(),
		)?
		.with_behaviour(
			|keypair|{
				Behaviour::new(
					libp2p::mdns::tokio::Behaviour::new(
						libp2p::mdns::Config::default(),
						keypair.public().into(),
					).unwrap(),
					libp2p::ping::Behaviour::new(
						libp2p::ping::Config::new()
							.with_timeout(Duration::from_secs(5))
							.with_interval(Duration::from_secs(1)),
					),
				)
			}
		)?
		.with_swarm_config(|config|config.with_idle_connection_timeout(Duration::from_secs(5)))
		.build();
	println!("peer id: {}",swarm.local_peer_id().to_string());
	swarm.listen_on("/ip4/0.0.0.0/tcp/0".parse()?)?;
	loop{
		let ev=swarm
			.select_next_some()
			.await;
		println!("{:#?}",ev);
		if let libp2p::swarm::SwarmEvent::Behaviour(BehaviourEvent::Mdns(libp2p::mdns::Event::Discovered(e)))=ev{
			for peer in e{
				swarm.dial(peer.1)?;
			}
		}
	}
}

やってることとしてはまず、NetworkBehaviourを２つ使うのでderive macroで１つにくっつけてます。

ここでderiveを使うためにCargo.tomlでlibp2pのmacrosを使っています

#[derive(libp2p::swarm::NetworkBehaviour)]
struct Behaviour{
	mdns:libp2p::mdns::tokio::Behaviour,
	ping:libp2p::ping::Behaviour,
}

タイムアウト５秒の１秒周期でPingして

libp2p::ping::Behaviour::new(
    libp2p::ping::Config::new()
        .with_timeout(Duration::from_secs(5))
        .with_interval(Duration::from_secs(1)),
),

なにもしてないとすぐ切断されるのでそのタイムアウトを５秒に伸ばします。

.with_swarm_config(|config|config.with_idle_connection_timeout(Duration::from_secs(5)))

ピアを見つけたらとにかく接続です。

if let libp2p::swarm::SwarmEvent::Behaviour(BehaviourEvent::Mdns(libp2p::mdns::Event::Discovered(e)))=ev{
    for peer in e{
        let _=swarm.dial(peer.1);
    }
}

結果

cargo runします。今回もやはり無限ループするので２つ動かします。
２つ動かしたものはそれぞれこんなことになっています：

peer id: 12D3KooWHbnn6as4CnjqkWdUeAeTS9BNPcufQfqxm7dtctFw5WWy
NewListenAddr {
    listener_id: ListenerId(
        1,
    ),
    address: "/ip4/127.0.0.1/tcp/38257",
}
NewListenAddr {
    listener_id: ListenerId(
        1,
    ),
    address: "/ip4/10.0.2.15/tcp/38257",
}
Behaviour(
    BehaviourEvent: Discovered([(PeerId("12D3KooWEQDUXZAA2QUSV4YLRCfjhHzf1kGPinxExnGDTtEy2Q9a"), "/ip4/10.0.2.15/tcp/37901/p2p/12D3KooWEQDUXZAA2QUSV4YLRCfjhHzf1kGPinxExnGDTtEy2Q9a")]),
)
IncomingConnection {
    connection_id: ConnectionId(
        2,
    ),
    local_addr: "/ip4/10.0.2.15/tcp/38257",
    send_back_addr: "/ip4/10.0.2.15/tcp/41440",
}
ConnectionEstablished {
    peer_id: PeerId(
        "12D3KooWEQDUXZAA2QUSV4YLRCfjhHzf1kGPinxExnGDTtEy2Q9a",
    ),
    connection_id: ConnectionId(
        1,
    ),
    endpoint: Dialer {
        address: "/ip4/10.0.2.15/tcp/37901/p2p/12D3KooWEQDUXZAA2QUSV4YLRCfjhHzf1kGPinxExnGDTtEy2Q9a",
        role_override: Dialer,
    },
    num_established: 1,
    concurrent_dial_errors: Some(
        [],
    ),
    established_in: 12.930926ms,
}
ConnectionEstablished {
    peer_id: PeerId(
        "12D3KooWEQDUXZAA2QUSV4YLRCfjhHzf1kGPinxExnGDTtEy2Q9a",
    ),
    connection_id: ConnectionId(
        2,
    ),
    endpoint: Listener {
        local_addr: "/ip4/10.0.2.15/tcp/38257",
        send_back_addr: "/ip4/10.0.2.15/tcp/41440",
    },
    num_established: 2,
    concurrent_dial_errors: None,
    established_in: 41.065112ms,
}
Behaviour(
    BehaviourEvent: Event { peer: PeerId("12D3KooWEQDUXZAA2QUSV4YLRCfjhHzf1kGPinxExnGDTtEy2Q9a"), connection: ConnectionId(1), result: Ok(44.038478ms) },
)
Behaviour(
    BehaviourEvent: Event { peer: PeerId("12D3KooWEQDUXZAA2QUSV4YLRCfjhHzf1kGPinxExnGDTtEy2Q9a"), connection: ConnectionId(2), result: Ok(44.330714ms) },
)
Behaviour(
    BehaviourEvent: Event { peer: PeerId("12D3KooWEQDUXZAA2QUSV4YLRCfjhHzf1kGPinxExnGDTtEy2Q9a"), connection: ConnectionId(1), result: Ok(918.144µs) },
)
Behaviour(
    BehaviourEvent: Event { peer: PeerId("12D3KooWEQDUXZAA2QUSV4YLRCfjhHzf1kGPinxExnGDTtEy2Q9a"), connection: ConnectionId(2), result: Ok(1.02102ms) },
)
Behaviour(
    BehaviourEvent: Event { peer: PeerId("12D3KooWEQDUXZAA2QUSV4YLRCfjhHzf1kGPinxExnGDTtEy2Q9a"), connection: ConnectionId(1), result: Ok(851.047µs) },
)
Behaviour(
    BehaviourEvent: Event { peer: PeerId("12D3KooWEQDUXZAA2QUSV4YLRCfjhHzf1kGPinxExnGDTtEy2Q9a"), connection: ConnectionId(2), result: Ok(1.389371ms) },
)
Behaviour(
    BehaviourEvent: Event { peer: PeerId("12D3KooWEQDUXZAA2QUSV4YLRCfjhHzf1kGPinxExnGDTtEy2Q9a"), connection: ConnectionId(1), result: Ok(1.024088ms) },
)
Behaviour(
    BehaviourEvent: Event { peer: PeerId("12D3KooWEQDUXZAA2QUSV4YLRCfjhHzf1kGPinxExnGDTtEy2Q9a"), connection: ConnectionId(2), result: Ok(900.088µs) },
)
Behaviour(
    BehaviourEvent: Event { peer: PeerId("12D3KooWEQDUXZAA2QUSV4YLRCfjhHzf1kGPinxExnGDTtEy2Q9a"), connection: ConnectionId(1), result: Ok(506.884µs) },
)
Behaviour(
    BehaviourEvent: Event { peer: PeerId("12D3KooWEQDUXZAA2QUSV4YLRCfjhHzf1kGPinxExnGDTtEy2Q9a"), connection: ConnectionId(2), result: Ok(1.833147ms) },
)
ConnectionClosed {
    peer_id: PeerId(
        "12D3KooWEQDUXZAA2QUSV4YLRCfjhHzf1kGPinxExnGDTtEy2Q9a",
    ),
    connection_id: ConnectionId(
        2,
    ),
    endpoint: Listener {
        local_addr: "/ip4/10.0.2.15/tcp/38257",
        send_back_addr: "/ip4/10.0.2.15/tcp/41440",
    },
    num_established: 1,
    cause: Some(
        IO(
            Custom {
                kind: Other,
                error: Error(
                    Right(
                        Closed,
                    ),
                ),
            },
        ),
    ),
}
ConnectionClosed {
    peer_id: PeerId(
        "12D3KooWEQDUXZAA2QUSV4YLRCfjhHzf1kGPinxExnGDTtEy2Q9a",
    ),
    connection_id: ConnectionId(
        1,
    ),
    endpoint: Dialer {
        address: "/ip4/10.0.2.15/tcp/37901/p2p/12D3KooWEQDUXZAA2QUSV4YLRCfjhHzf1kGPinxExnGDTtEy2Q9a",
        role_override: Dialer,
    },
    num_established: 0,
    cause: Some(
        KeepAliveTimeout,
    ),
}

peer id: 12D3KooWEQDUXZAA2QUSV4YLRCfjhHzf1kGPinxExnGDTtEy2Q9a
NewListenAddr {
    listener_id: ListenerId(
        1,
    ),
    address: "/ip4/127.0.0.1/tcp/37901",
}
NewListenAddr {
    listener_id: ListenerId(
        1,
    ),
    address: "/ip4/10.0.2.15/tcp/37901",
}
Behaviour(
    BehaviourEvent: Discovered([(PeerId("12D3KooWHbnn6as4CnjqkWdUeAeTS9BNPcufQfqxm7dtctFw5WWy"), "/ip4/10.0.2.15/tcp/38257/p2p/12D3KooWHbnn6as4CnjqkWdUeAeTS9BNPcufQfqxm7dtctFw5WWy")]),
)
IncomingConnection {
    connection_id: ConnectionId(
        2,
    ),
    local_addr: "/ip4/10.0.2.15/tcp/37901",
    send_back_addr: "/ip4/10.0.2.15/tcp/47314",
}
ConnectionEstablished {
    peer_id: PeerId(
        "12D3KooWHbnn6as4CnjqkWdUeAeTS9BNPcufQfqxm7dtctFw5WWy",
    ),
    connection_id: ConnectionId(
        1,
    ),
    endpoint: Dialer {
        address: "/ip4/10.0.2.15/tcp/38257/p2p/12D3KooWHbnn6as4CnjqkWdUeAeTS9BNPcufQfqxm7dtctFw5WWy",
        role_override: Dialer,
    },
    num_established: 1,
    concurrent_dial_errors: Some(
        [],
    ),
    established_in: 6.793408ms,
}
ConnectionEstablished {
    peer_id: PeerId(
        "12D3KooWHbnn6as4CnjqkWdUeAeTS9BNPcufQfqxm7dtctFw5WWy",
    ),
    connection_id: ConnectionId(
        2,
    ),
    endpoint: Listener {
        local_addr: "/ip4/10.0.2.15/tcp/37901",
        send_back_addr: "/ip4/10.0.2.15/tcp/47314",
    },
    num_established: 2,
    concurrent_dial_errors: None,
    established_in: 51.313413ms,
}
Behaviour(
    BehaviourEvent: Event { peer: PeerId("12D3KooWHbnn6as4CnjqkWdUeAeTS9BNPcufQfqxm7dtctFw5WWy"), connection: ConnectionId(1), result: Ok(44.250221ms) },
)
Behaviour(
    BehaviourEvent: Event { peer: PeerId("12D3KooWHbnn6as4CnjqkWdUeAeTS9BNPcufQfqxm7dtctFw5WWy"), connection: ConnectionId(2), result: Ok(44.474519ms) },
)
Behaviour(
    BehaviourEvent: Event { peer: PeerId("12D3KooWHbnn6as4CnjqkWdUeAeTS9BNPcufQfqxm7dtctFw5WWy"), connection: ConnectionId(1), result: Ok(1.071157ms) },
)
Behaviour(
    BehaviourEvent: Event { peer: PeerId("12D3KooWHbnn6as4CnjqkWdUeAeTS9BNPcufQfqxm7dtctFw5WWy"), connection: ConnectionId(2), result: Ok(1.016736ms) },
)
Behaviour(
    BehaviourEvent: Event { peer: PeerId("12D3KooWHbnn6as4CnjqkWdUeAeTS9BNPcufQfqxm7dtctFw5WWy"), connection: ConnectionId(1), result: Ok(1.020467ms) },
)
Behaviour(
    BehaviourEvent: Event { peer: PeerId("12D3KooWHbnn6as4CnjqkWdUeAeTS9BNPcufQfqxm7dtctFw5WWy"), connection: ConnectionId(2), result: Ok(1.291714ms) },
)
Behaviour(
    BehaviourEvent: Event { peer: PeerId("12D3KooWHbnn6as4CnjqkWdUeAeTS9BNPcufQfqxm7dtctFw5WWy"), connection: ConnectionId(1), result: Ok(1.086732ms) },
)
Behaviour(
    BehaviourEvent: Event { peer: PeerId("12D3KooWHbnn6as4CnjqkWdUeAeTS9BNPcufQfqxm7dtctFw5WWy"), connection: ConnectionId(2), result: Ok(895.066µs) },
)
Behaviour(
    BehaviourEvent: Event { peer: PeerId("12D3KooWHbnn6as4CnjqkWdUeAeTS9BNPcufQfqxm7dtctFw5WWy"), connection: ConnectionId(1), result: Ok(1.078993ms) },
)
Behaviour(
    BehaviourEvent: Event { peer: PeerId("12D3KooWHbnn6as4CnjqkWdUeAeTS9BNPcufQfqxm7dtctFw5WWy"), connection: ConnectionId(2), result: Ok(903.222µs) },
)
ConnectionClosed {
    peer_id: PeerId(
        "12D3KooWHbnn6as4CnjqkWdUeAeTS9BNPcufQfqxm7dtctFw5WWy",
    ),
    connection_id: ConnectionId(
        1,
    ),
    endpoint: Dialer {
        address: "/ip4/10.0.2.15/tcp/38257/p2p/12D3KooWHbnn6as4CnjqkWdUeAeTS9BNPcufQfqxm7dtctFw5WWy",
        role_override: Dialer,
    },
    num_established: 1,
    cause: Some(
        KeepAliveTimeout,
    ),
}
ConnectionClosed {
    peer_id: PeerId(
        "12D3KooWHbnn6as4CnjqkWdUeAeTS9BNPcufQfqxm7dtctFw5WWy",
    ),
    connection_id: ConnectionId(
        2,
    ),
    endpoint: Listener {
        local_addr: "/ip4/10.0.2.15/tcp/37901",
        send_back_addr: "/ip4/10.0.2.15/tcp/47314",
    },
    num_established: 0,
    cause: Some(
        IO(
            Custom {
                kind: Other,
                error: Error(
                    Right(
                        Closed,
                    ),
                ),
            },
        ),
    ),
}

Pingのイベントが生じているようです。

Behaviour(
    BehaviourEvent: Event { peer: PeerId("12D3KooWEQDUXZAA2QUSV4YLRCfjhHzf1kGPinxExnGDTtEy2Q9a"), connection: ConnectionId(2), result: Ok(1.02102ms) },
)

↑こういうのがいっぱい出てます。そしてOkとあります。

やったね✨🎉🥳

4
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
4
0