Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Support client mode in kademlia #2184

Closed
wants to merge 39 commits into from
Closed
Show file tree
Hide file tree
Changes from 24 commits
Commits
Show all changes
39 commits
Select commit Hold shift + click to select a range
f8cf486
Adding config parameter to support client mode
Aug 7, 2021
1a042f8
Merge branch 'master' into kademlia-client-mode
Aug 7, 2021
3e759a0
example for showing peer interaction in kademlia
Aug 8, 2021
b8cbe62
finally able to discover other peers
Aug 8, 2021
fc218a0
Add `Mode` enum for kademlia client/server mode
Aug 10, 2021
e58e544
Remove confirmation code from `inject_fully_negotiated_inbound`
Aug 10, 2021
af1fcbd
Correct `client` value for network behaviour handler
Aug 10, 2021
c944ecb
Merge branch 'kademlia-client-mode' into kademlia-example
Aug 10, 2021
4be8fca
refactoring
Aug 11, 2021
e213350
Refactoring complete
Aug 12, 2021
e47148d
Add example for running kademlia in `client` mode
Aug 15, 2021
cdaa0a1
Merge commit with master
Aug 15, 2021
1a09907
removing .vscode
Aug 15, 2021
46167a1
Add better documentation and comments
Aug 21, 2021
8bf3ef4
replaced example with test in kad-behaviour
Aug 28, 2021
9ef4c4c
Changes to client mode test
Aug 29, 2021
c387f41
remove kad-example from cargo.toml
Aug 29, 2021
5b21774
Better checks for client mode test
Aug 29, 2021
5ea645e
Fix client mode test
Aug 29, 2021
aa6500f
Refactor kademlia client mode test
Aug 30, 2021
146405f
Rename to check variables
Aug 30, 2021
2cea417
Final fix for `client_mode` test.
Aug 30, 2021
54331e5
remove commented code
Aug 30, 2021
9d23741
Merge branch 'master' into kademlia-client-mode
Sep 3, 2021
b202c5b
Add some basic comments.
Sep 7, 2021
8234a17
Changes to test. Currently failing.
Sep 7, 2021
d5e6509
Changes to `client_mode` test. Currently failing.
Sep 7, 2021
06c1a30
Correct variable name.
Sep 10, 2021
d420fa6
Add checks to see if PUT and GET are working.
Sep 10, 2021
190c3dd
Add correct checks for PUT and GET operations.
Sep 10, 2021
ab97b10
Merge branch `master` into `kademlia-client-mode`
Sep 10, 2021
f3259b1
update prost dependencies
Sep 10, 2021
f308b0c
Replace `any` with `all` for server peer check.
Sep 11, 2021
8f33ba1
Merge branch `master` into `kademlia-client-mode`
Sep 14, 2021
520fb8e
Extra comments
Sep 14, 2021
d52c6af
Fix merge conflict with master
Sep 30, 2021
81eae5f
New test
Oct 3, 2021
f0c2fc3
Remove `allow_listening` logic.
Oct 3, 2021
f0bf7df
Fix gitignore
Oct 3, 2021
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
22 changes: 21 additions & 1 deletion protocols/kad/src/behaviour.rs
Original file line number Diff line number Diff line change
Expand Up @@ -29,7 +29,7 @@ use crate::handler::{
};
use crate::jobs::*;
use crate::kbucket::{self, Distance, KBucketsTable, NodeStatus};
use crate::protocol::{KadConnectionType, KadPeer, KademliaProtocolConfig};
use crate::protocol::{KadConnectionType, KadPeer, KademliaProtocolConfig, Mode};
use crate::query::{Query, QueryConfig, QueryId, QueryPool, QueryPoolState};
use crate::record::{
self,
Expand Down Expand Up @@ -175,6 +175,8 @@ pub struct KademliaConfig {
connection_idle_timeout: Duration,
kbucket_inserts: KademliaBucketInserts,
caching: KademliaCaching,
// Set the peer mode.
mode: Mode,
}

/// The configuration for Kademlia "write-back" caching after successful
Expand Down Expand Up @@ -209,6 +211,8 @@ impl Default for KademliaConfig {
connection_idle_timeout: Duration::from_secs(10),
kbucket_inserts: KademliaBucketInserts::OnConnected,
caching: KademliaCaching::Enabled { max_peers: 1 },
// By default, a peer will not be in client mode.
mode: Mode::default(),
}
}
}
Expand Down Expand Up @@ -390,6 +394,21 @@ impl KademliaConfig {
self.caching = c;
self
}

/// Sets the [`Mode`] the node is operating in.
///
/// The default is [`Mode::Server`].
///
/// In [`Mode::Client`] the node does not advertise support for the Kademlia
/// protocol, nor does it accept incoming Kademlia streams. Peers will thus not
/// include the local node in their routing table. The node can still make
/// outbound requests.
///
/// Use [`Mode::Client`] when not publicly reachable.
pub fn set_mode(&mut self, mode: Mode) -> &mut Self {
self.mode = mode;
self
}
}

impl<TStore> Kademlia<TStore>
Expand Down Expand Up @@ -1710,6 +1729,7 @@ where
protocol_config: self.protocol_config.clone(),
allow_listening: true,
idle_timeout: self.connection_idle_timeout,
mode: Mode::default(),
})
}

Expand Down
64 changes: 64 additions & 0 deletions protocols/kad/src/behaviour/test.rs
Original file line number Diff line number Diff line change
Expand Up @@ -1317,3 +1317,67 @@ fn network_behaviour_inject_address_change() {
kademlia.addresses_of_peer(&remote_peer_id),
);
}

#[test]
fn client_mode() {
let mut cfg = KademliaConfig::default();
cfg.set_mode(Mode::Client);

// Create a server and client peer.
let client_swarm = build_node_with_config(cfg);
let server_swarm = build_node_with_config(KademliaConfig::default());

let mut swarms = vec![client_swarm, server_swarm];

// Collect addresses and peer IDs.
let addrs: Vec<_> = swarms.iter().map(|(addr, _)| addr.clone()).collect();
let peers: Vec<_> = swarms
.iter()
.map(|(_, swarm)| swarm.local_peer_id().clone())
.collect();

swarms[0].1.dial_addr(addrs[1].clone()).unwrap();
swarms[1].1.dial_addr(addrs[0].clone()).unwrap();
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why does the server dial the client?

Copy link
Author

@whereistejas whereistejas Sep 6, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It's there to guarantee that server tried to reach out to the client, before we check if the client is absent from the server's routing table.


block_on(poll_fn(move |ctx| {
whereistejas marked this conversation as resolved.
Show resolved Hide resolved
// Flag variables to store the result of various checks.
let mut is_server_present = false;
let mut is_client_absent = false;

for (_, swarm) in swarms.iter_mut() {
loop {
match swarm.poll_next_unpin(ctx) {
Poll::Ready(Some(SwarmEvent::Behaviour(KademliaEvent::RoutingUpdated {
peer,
..
}))) => {
// Check if the server peer is present in the client peer's routing table.
if swarm.local_peer_id().clone() == peers[0] {
is_server_present = peer == peers[1];
}
return Poll::Ready(());
Copy link
Author

@whereistejas whereistejas Aug 30, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

There is one part of this code, that I don't understand. I noticed that the test fails if we don't add return Poll::Ready(()) in every match arm. I'm still a newbie at async, could you tell me why we need to return this and what its effect is?

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I suggest reading the async book: https://rust-lang.github.io/async-book/

With the early return we might not check whether the server stores the client in its routing table.

}
Poll::Ready(_) => {
whereistejas marked this conversation as resolved.
Show resolved Hide resolved
// Check if the client peer is NOT present in the server peer's routing table.
if swarm.local_peer_id().clone() == peers[1] {
is_client_absent = swarm.behaviour_mut().kbucket(peers[0]).is_none();
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It is kind of hard to reliably check for absence of something, especially in an async setting.

How about:

  1. The server stores a value Kademlia::put_value, waiting for a KademliaEvent::OutboundQueryCompleted.
  2. The client connects to the server.
  3. The client does a Kademlia::get_record, waiting for a KademliaEvent::OutboundQueryCompleted.
  4. Check that the client has the server in its routing table and check that the server does not have the client in its routing table.

Copy link
Author

@whereistejas whereistejas Sep 6, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Sure. I will implement this. I will try to finish before the same time, tomorrow. Please, keep an eye out for any notifications from this PR. I want to get this PR closed, as early as possible (ofcourse, without sacrificing code quality).

Are we finished with the other parts of this PR except the test?

}
return Poll::Ready(());
}
Poll::Pending => break,
}
}
}
// The assert statements are present outside the match block, to make sure that all the
// necessary events occur.
assert!(
is_server_present,
"The client peer does not have the server peer in its routing table."
);
assert!(
is_client_absent,
"The server peer has the client peer in its routing table."
);
Poll::Pending
}))
}
46 changes: 30 additions & 16 deletions protocols/kad/src/handler.rs
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,7 @@

use crate::protocol::{
KadInStreamSink, KadOutStreamSink, KadPeer, KadRequestMsg, KadResponseMsg,
KademliaProtocolConfig,
KademliaProtocolConfig, Mode,
};
use crate::record::{self, Record};
use futures::prelude::*;
Expand Down Expand Up @@ -62,10 +62,15 @@ impl<T: Clone + fmt::Debug + Send + 'static> IntoProtocolsHandler for KademliaHa
}

fn inbound_protocol(&self) -> <Self::Handler as ProtocolsHandler>::InboundProtocol {
if self.config.allow_listening {
upgrade::EitherUpgrade::A(self.config.protocol_config.clone())
} else {
upgrade::EitherUpgrade::B(upgrade::DeniedUpgrade)
match self.config.mode {
Mode::Client => upgrade::EitherUpgrade::B(upgrade::DeniedUpgrade),
Mode::Server => {
if self.config.allow_listening {
whereistejas marked this conversation as resolved.
Show resolved Hide resolved
upgrade::EitherUpgrade::A(self.config.protocol_config.clone())
} else {
upgrade::EitherUpgrade::B(upgrade::DeniedUpgrade)
}
}
}
}
}
Expand Down Expand Up @@ -123,6 +128,9 @@ pub struct KademliaHandlerConfig {

/// Time after which we close an idle connection.
pub idle_timeout: Duration,

/// [`Mode`] the handler is operating in.
pub mode: Mode,
}

/// State of an active substream, opened either by us or by the remote.
Expand Down Expand Up @@ -480,11 +488,20 @@ where
type InboundOpenInfo = ();

fn listen_protocol(&self) -> SubstreamProtocol<Self::InboundProtocol, Self::InboundOpenInfo> {
if self.config.allow_listening {
SubstreamProtocol::new(self.config.protocol_config.clone(), ())
.map_upgrade(upgrade::EitherUpgrade::A)
} else {
SubstreamProtocol::new(upgrade::EitherUpgrade::B(upgrade::DeniedUpgrade), ())
match self.config.mode {
Mode::Server => {
if self.config.allow_listening {
SubstreamProtocol::new(self.config.protocol_config.clone(), ())
.map_upgrade(upgrade::EitherUpgrade::A)
} else {
SubstreamProtocol::new(upgrade::EitherUpgrade::B(upgrade::DeniedUpgrade), ())
}
}
// If we are in client mode, I don't want to advertise Kademlia so that other peers will
// not send me Kademlia requests.
Mode::Client => {
SubstreamProtocol::new(upgrade::EitherUpgrade::B(upgrade::DeniedUpgrade), ())
}
}
}

Expand Down Expand Up @@ -520,12 +537,8 @@ where
self.next_connec_unique_id.0 += 1;
self.substreams
.push(SubstreamState::InWaitingMessage(connec_unique_id, protocol));
if let ProtocolStatus::Unconfirmed = self.protocol_status {
// Upon the first successfully negotiated substream, we know that the
// remote is configured with the same protocol name and we want
// the behaviour to add this peer to the routing table, if possible.
self.protocol_status = ProtocolStatus::Confirmed;
}
// Just because another peer is sending us Kademlia requests, doesn't necessarily
// mean that it will answer the Kademlia requests that we send to it.
}

fn inject_event(&mut self, message: KademliaHandlerIn<TUserData>) {
Expand Down Expand Up @@ -763,6 +776,7 @@ impl Default for KademliaHandlerConfig {
protocol_config: Default::default(),
allow_listening: true,
idle_timeout: Duration::from_secs(10),
mode: Mode::default(),
}
}
}
Expand Down
13 changes: 13 additions & 0 deletions protocols/kad/src/protocol.rs
Original file line number Diff line number Diff line change
Expand Up @@ -46,6 +46,19 @@ pub const DEFAULT_PROTO_NAME: &[u8] = b"/ipfs/kad/1.0.0";
/// The default maximum size for a varint length-delimited packet.
pub const DEFAULT_MAX_PACKET_SIZE: usize = 16 * 1024;

#[derive(Debug, Clone)]
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
#[derive(Debug, Clone)]
/// See [`crate::KademliaConfig::set_mode`].
#[derive(Debug, Clone)]

/// See [`crate::KademliaConfig::set_mode`].
pub enum Mode {
Client,
Server,
}

impl Default for Mode {
fn default() -> Self {
Mode::Server
}
}

/// Status of our connection to a node reported by the Kademlia protocol.
#[derive(Copy, Clone, PartialEq, Eq, Debug, Hash)]
pub enum KadConnectionType {
Expand Down