Newbie kademlia question. How to connect to other peers?

[whereistejas]

Hi,
For the past week, I have been trying to write a program that creates four peers, each of which supports the Kademlia behavior. I put all of them in one swarm, and then listen on an arbitrary ip address. I want all the four peers connect to one another, once that happens I should be able to see all the peers in each other's kbucket.

use std::iter::FromIterator;

use libp2p_dns as dns;
use libp2p_noise as noise;
use libp2p_tcp as tcp;
use libp2p_yamux as yamux;

use libp2p_core::{
    identity, muxing::StreamMuxerBox, transport::Boxed, upgrade, Multiaddr, PeerId, Transport,
};
use libp2p_swarm::{Swarm, SwarmEvent};

use libp2p_kad::record::store::MemoryStore;
use libp2p_kad::{GetClosestPeersError, Kademlia, KademliaConfig, KademliaEvent, QueryResult};

use async_std::task;
use futures::StreamExt;
use std::str::FromStr;

#[async_std::test]
async fn kademlia_client_peer() {
    // generate four keys for four peers
    let local_key = vec![
        identity::Keypair::generate_ed25519(),
        identity::Keypair::generate_ed25519(),
        identity::Keypair::generate_ed25519(),
        identity::Keypair::generate_ed25519(),
    ];

    //PeerId::from_public_key(local_key.public());
    let local_peer_id = Vec::from_iter(
        local_key
            .iter()
            .map(|key| PeerId::from_public_key(key.public())),
    );

    let mut transport = Vec::new();

    for key in local_key {
        transport.push(
            make_transport(key.clone())
                .await
                .expect("Failed to create transport"),
        );
    }

    let mut swarm = Vec::from_iter(
        local_peer_id
            .iter()
            .zip(transport)
            .map(|(peer, transport)| {
                let cfg = KademliaConfig::default();
                let store = MemoryStore::new(peer.clone());
                let mut behaviour = Kademlia::with_config(peer.clone(), store, cfg);

                for other_peer in local_peer_id.clone() {
                    if peer.to_string() != other_peer.to_string() {
                        behaviour.add_address(&other_peer, Multiaddr::from_str("").unwrap());
                    }
                }

                Swarm::new(transport, behaviour, peer.clone())
            }),
    );

    dbg!(
        "{} is searching for {}",
        &local_peer_id[3],
        &local_peer_id[0]
    );

    swarm[3].listen_on("/ip4/127.0.0.1/tcp/8080".parse().unwrap()).unwrap();
    swarm[0].dial(&local_peer_id[3]).unwrap();

    swarm[3].behaviour_mut().get_closest_peers(local_peer_id[0]);

    task::block_on(async move {
        loop {
            let event = swarm[3].select_next_some().await;
            if let SwarmEvent::Behaviour(KademliaEvent::OutboundQueryCompleted {
                result: QueryResult::GetClosestPeers(result),
                ..
            }) = event
            {
                match result {
                    Ok(ok) => {
                        if !ok.peers.is_empty() {
                            println!("Query finished with closest peers: {:#?}", ok.peers)
                        } else {
                            // The example is considered failed as there
                            // should always be at least 1 reachable peer.
                            println!("Query finished with no closest peers.")
                        }
                    }
                    Err(GetClosestPeersError::Timeout { peers, .. }) => {
                        if !peers.is_empty() {
                            println!("Query timed out with closest peers: {:#?}", peers)
                        } else {
                            // The example is considered failed as there
                            // should always be at least 1 reachable peer.
                            println!("Query timed out with no closest peers.");
                        }
                    }
                };
                break;
            };
        }
    });

    assert!(false);
}

async fn make_transport(
    keypair: identity::Keypair,
) -> std::io::Result<Boxed<(PeerId, StreamMuxerBox)>> {
    let transport = {
        let tcp = tcp::TcpConfig::new().nodelay(true);
        let dns_tcp = dns::DnsConfig::system(tcp).await?;
        dns_tcp
    };

    let noise_keys = noise::Keypair::<noise::X25519Spec>::new()
        .into_authentic(&keypair)
        .expect("Signing noise keypair failed");

    Ok(transport
        .upgrade(upgrade::Version::V1)
        .authenticate(noise::NoiseConfig::xx(noise_keys).into_authenticated())
        .multiplex(yamux::YamuxConfig::default())
        .timeout(std::time::Duration::from_secs(20))
        .boxed())
}

However this code always fails to find peers.

I tried to run the ipfs-kad example code, but it always fails to find the closest peers, even if I'm supplying it with one of the peers from the boot nodes.

I think, I'm missing the code to announce each peer to one another. Can someone please help me? I would like to know how to connect the peers to each other.

[thomaseizinger]

Two things:

1. The only one of your swarms that is listening on an address is swarm[3], hence that is the only one someone can connect to. You probably want all swarms to listen.
2. You are setting an empty address for each peer: behaviour.add_address(&other_peer, Multiaddr::from_str("").unwrap()); You need to let each peer know, what the actual address of each other peer is to establish a connection via Swarm::dial. Swarm::dial assumes that you have the address of that peer stored somewhere on your NetworkBehaviour. Alternative, you can use Swarm::dial_addr to connect to a specific address.

[whereistejas]

Hi, I have managed to get the code working, now. I'm able to connect to other peers and see other peers in the their kbuckets. The code can be found here.

I'm trying to refactor my code and instead of creating a seperate method that serves as the eventloop, I'm simply passing certain events to certain methods, like this:

loop {
    futures::select! {
        // handle input from the command line
        line = stdin.read_line(&mut buffer).fuse() => handle_command(buffer),
        // handle events from the Swarm
        event = swarm.next() => handle_event(event.unwrap()),
    }
}

I'm having trouble figuring out the type of the event in the the handle_event method.

fn handle_event(event: SwarmEvent<(), EitherError>) {
    match event {
        SwarmEvent::NewListenAddr { address, .. } => println!("Listening on: {}", address),
        _ => {}
    }
}

I see that SwarmEvent expects certains types but I cannot figure out what they should be.

In correlation to this, I am a little confused about the difference between SwarmEvents and NetworkEvents. Could anyone please help me with that a little?

[thomaseizinger]

I see that SwarmEvent expects certains types but I cannot figure out what they should be.

If you are not using event_process = false on your NetworkBehaviour, then SwarmEvent will be quite meaningless, at least in regards to Behaviour because it will simply be just the unit type (). I find myself always defaulting to event_process = false because it plays nicely with handling all the other variants of SwarmEvent in addition to SwarmEvent::Behaviour.

I usually don't bother writing these types out and do the main branching logic within the futures::select!. You can always delegate from within a match there to individual functions.

I'm trying to refactor my code and instead of creating a seperate method that serves as the eventloop, I'm simply passing certain events to certain methods, like this:

Just be aware that - as with every eventloop-based design regardless of language or tooling - you MUST NOT stall the event-loop for too long, otherwise your application will become non-responsive. In other words, don't do anything computationally or IO expensive as part of reacting to events.

In correlation to this, I am a little confused about the difference between SwarmEvents and NetworkEvents. Could anyone please help me with that a little?

Unless you are interacting with libp2p-core directly, you shouldn't need to concern yourself with NetworkEvent. Where is that surfacing for you?

[whereistejas]

Unless you are interacting with libp2p-core directly, you shouldn't need to concern yourself with NetworkEvent. Where is that surfacing for you?

It is not surfacing for me, as such. I came across NetworkEvent while reading other discussions.

I find myself always defaulting to event_process = false because it plays nicely with handling all the other variants of SwarmEvent in addition to SwarmEvent::Behaviour.

So, if my understanding is correct, by setting event_process = false I don't have to implement NetworkBehaviourEventProcess for any of the protocols in my custom behaviour. And, by setting an out_event type, I can wrap all my protocol events in an enum, making it easier to handle and pass them around in methods. Am I correct?

[thomaseizinger]

I find myself always defaulting to event_process = false because it plays nicely with handling all the other variants of SwarmEvent in addition to SwarmEvent::Behaviour.

So, if my understanding is correct, by setting event_process = false I don't have to implement NetworkBehaviourEventProcess for any of the protocols in my custom behaviour. And, by setting an out_event type, I can wrap all my protocol events in an enum, making it easier to handle and pass them around in methods. Am I correct?

Basically yes.

To understand exactly what happens, you can use cargo expand and look at the generated code from the derived NetworkBehaviour.

In a nutshell though: when polling the individual behaviours that are composed in your NetworkBehaviour, something needs to happen to the events that are emitted.

By default, NetworkBehaviourEventProcess::inject_event is called with the event but that forces you to put all your logic in this callback function.

If you set event_process = false, a From implementation is required to convert the inner event to your custom outevent. Any event generated by the inner behaviours will be converted to your custom out-event and returned as part of SwarmEvent::Behaviour.