use crate::handler::{KademliaHandler, KademliaHandlerEvent, KademliaHandlerIn, KademliaRequestId};
use crate::kbucket::{KBucketsTable, Update};
use crate::protocol::{KadConnectionType, KadPeer};
use crate::query::{QueryConfig, QueryState, QueryStatePollOut, QueryTarget};
use fnv::{FnvHashMap, FnvHashSet};
use futures::{prelude::*, stream};
use libp2p_core::swarm::{ConnectedPoint, NetworkBehaviour, NetworkBehaviourAction, PollParameters};
use libp2p_core::{protocols_handler::ProtocolsHandler, Multiaddr, PeerId};
use multihash::Multihash;
use rand;
use smallvec::SmallVec;
use std::{cmp::Ordering, marker::PhantomData, time::Duration, time::Instant};
use tokio_io::{AsyncRead, AsyncWrite};
use tokio_timer::Interval;
pub struct Kademlia<TSubstream> {
kbuckets: KBucketsTable<PeerId, SmallVec<[Multiaddr; 4]>>,
active_queries: FnvHashMap<QueryId, (QueryState, QueryPurpose, Vec<PeerId>)>,
queries_to_starts: SmallVec<[(QueryId, QueryTarget, QueryPurpose); 8]>,
connected_peers: FnvHashSet<PeerId>,
pending_rpcs: SmallVec<[(PeerId, KademliaHandlerIn<QueryId>); 8]>,
next_query_id: QueryId,
remote_requests: SmallVec<[(PeerId, KademliaRequestId, QueryTarget); 4]>,
values_providers: FnvHashMap<Multihash, SmallVec<[PeerId; 20]>>,
providing_keys: FnvHashSet<Multihash>,
refresh_add_providers: stream::Fuse<Interval>,
parallelism: usize,
num_results: usize,
rpc_timeout: Duration,
queued_events: SmallVec<[NetworkBehaviourAction<KademliaHandlerIn<QueryId>, KademliaOut>; 32]>,
add_provider: SmallVec<[(Multihash, PeerId); 32]>,
marker: PhantomData<TSubstream>,
}
#[derive(Debug, Copy, Clone, Hash, PartialEq, Eq)]
pub struct QueryId(usize);
#[derive(Debug, Clone, PartialEq, Eq)]
enum QueryPurpose {
Initialization,
UserRequest,
AddProvider(Multihash),
}
impl<TSubstream> Kademlia<TSubstream> {
#[inline]
pub fn new(local_peer_id: PeerId) -> Self {
Self::new_inner(local_peer_id, true)
}
#[inline]
pub fn without_init(local_peer_id: PeerId) -> Self {
Self::new_inner(local_peer_id, false)
}
pub fn add_address(&mut self, peer_id: &PeerId, address: Multiaddr) {
if let Some(list) = self.kbuckets.entry_mut(peer_id) {
if list.iter().all(|a| *a != address) {
list.push(address);
}
}
}
fn new_inner(local_peer_id: PeerId, initialize: bool) -> Self {
let parallelism = 3;
let mut behaviour = Kademlia {
kbuckets: KBucketsTable::new(local_peer_id, Duration::from_secs(60)),
queued_events: SmallVec::new(),
queries_to_starts: SmallVec::new(),
active_queries: Default::default(),
connected_peers: Default::default(),
pending_rpcs: SmallVec::with_capacity(parallelism),
next_query_id: QueryId(0),
remote_requests: SmallVec::new(),
values_providers: FnvHashMap::default(),
providing_keys: FnvHashSet::default(),
refresh_add_providers: Interval::new_interval(Duration::from_secs(60)).fuse(),
parallelism,
num_results: 20,
rpc_timeout: Duration::from_secs(8),
add_provider: SmallVec::new(),
marker: PhantomData,
};
if initialize {
for n in 0..256 {
let peer_id = match gen_random_id(behaviour.kbuckets.my_id(), n) {
Ok(p) => p,
Err(()) => continue,
};
behaviour.start_query(QueryTarget::FindPeer(peer_id), QueryPurpose::Initialization);
}
}
behaviour
}
fn build_result<TUserData>(&mut self, query: QueryTarget, request_id: KademliaRequestId, parameters: &mut PollParameters)
-> KademliaHandlerIn<TUserData>
{
match query {
QueryTarget::FindPeer(key) => {
let closer_peers = self.kbuckets
.find_closest_with_self(&key)
.take(self.num_results)
.map(|peer_id| build_kad_peer(peer_id, parameters, &self.kbuckets, &self.connected_peers))
.collect();
KademliaHandlerIn::FindNodeRes {
closer_peers,
request_id,
}
},
QueryTarget::GetProviders(key) => {
let closer_peers = self.kbuckets
.find_closest_with_self(&key)
.take(self.num_results)
.map(|peer_id| build_kad_peer(peer_id, parameters, &self.kbuckets, &self.connected_peers))
.collect();
let provider_peers = self.values_providers
.get(&key)
.into_iter()
.flat_map(|peers| peers)
.map(|peer_id| build_kad_peer(peer_id.clone(), parameters, &self.kbuckets, &self.connected_peers))
.collect();
KademliaHandlerIn::GetProvidersRes {
closer_peers,
provider_peers,
request_id,
}
},
}
}
}
impl<TSubstream> Kademlia<TSubstream> {
#[inline]
pub fn find_node(&mut self, peer_id: PeerId) {
self.start_query(QueryTarget::FindPeer(peer_id), QueryPurpose::UserRequest);
}
#[inline]
pub fn get_providers(&mut self, key: Multihash) {
self.start_query(QueryTarget::GetProviders(key), QueryPurpose::UserRequest);
}
pub fn add_providing(&mut self, key: PeerId) {
self.providing_keys.insert(key.clone().into());
let providers = self.values_providers.entry(key.into()).or_insert_with(Default::default);
let my_id = self.kbuckets.my_id();
if !providers.iter().any(|k| k == my_id) {
providers.push(my_id.clone());
}
self.refresh_add_providers = Interval::new(Instant::now(), Duration::from_secs(60)).fuse();
}
pub fn remove_providing(&mut self, key: &Multihash) {
self.providing_keys.remove(key);
let providers = match self.values_providers.get_mut(key) {
Some(p) => p,
None => return,
};
if let Some(position) = providers.iter().position(|k| k == key) {
providers.remove(position);
providers.shrink_to_fit();
}
}
fn start_query(&mut self, target: QueryTarget, purpose: QueryPurpose) {
let query_id = self.next_query_id;
self.next_query_id.0 += 1;
self.queries_to_starts.push((query_id, target, purpose));
}
}
impl<TSubstream> NetworkBehaviour for Kademlia<TSubstream>
where
TSubstream: AsyncRead + AsyncWrite,
{
type ProtocolsHandler = KademliaHandler<TSubstream, QueryId>;
type OutEvent = KademliaOut;
fn new_handler(&mut self) -> Self::ProtocolsHandler {
KademliaHandler::dial_and_listen()
}
fn addresses_of_peer(&mut self, peer_id: &PeerId) -> Vec<Multiaddr> {
self.kbuckets
.get(peer_id)
.map(|l| l.iter().cloned().collect::<Vec<_>>())
.unwrap_or_else(Vec::new)
}
fn inject_connected(&mut self, id: PeerId, endpoint: ConnectedPoint) {
if let Some(pos) = self.pending_rpcs.iter().position(|(p, _)| p == &id) {
let (_, rpc) = self.pending_rpcs.remove(pos);
self.queued_events.push(NetworkBehaviourAction::SendEvent {
peer_id: id.clone(),
event: rpc,
});
}
match self.kbuckets.set_connected(&id) {
Update::Pending(to_ping) => {
self.queued_events.push(NetworkBehaviourAction::DialPeer {
peer_id: to_ping.clone(),
})
},
_ => ()
}
if let ConnectedPoint::Dialer { address } = endpoint {
if let Some(list) = self.kbuckets.entry_mut(&id) {
if list.iter().all(|a| *a != address) {
list.push(address);
}
}
}
self.connected_peers.insert(id);
}
fn inject_disconnected(&mut self, id: &PeerId, _: ConnectedPoint) {
let was_in = self.connected_peers.remove(id);
debug_assert!(was_in);
for (query, _, _) in self.active_queries.values_mut() {
query.inject_rpc_error(id);
}
self.kbuckets.set_disconnected(&id);
}
fn inject_replaced(&mut self, peer_id: PeerId, _: ConnectedPoint, new_endpoint: ConnectedPoint) {
for (query_id, (query, _, _)) in self.active_queries.iter() {
if query.is_waiting(&peer_id) {
self.queued_events.push(NetworkBehaviourAction::SendEvent {
peer_id: peer_id.clone(),
event: query.target().to_rpc_request(*query_id),
});
}
}
if let ConnectedPoint::Dialer { address } = new_endpoint {
if let Some(list) = self.kbuckets.entry_mut(&peer_id) {
if list.iter().all(|a| *a != address) {
list.push(address);
}
}
}
}
fn inject_node_event(&mut self, source: PeerId, event: KademliaHandlerEvent<QueryId>) {
match event {
KademliaHandlerEvent::FindNodeReq { key, request_id } => {
self.remote_requests.push((source, request_id, QueryTarget::FindPeer(key)));
return;
}
KademliaHandlerEvent::FindNodeRes {
closer_peers,
user_data,
} => {
for peer in closer_peers.iter() {
if let Some(entry) = self.kbuckets.entry_mut(&peer.node_id) {
for addr in peer.multiaddrs.iter() {
if entry.iter().all(|a| a != addr) {
entry.push(addr.clone());
}
}
}
self.queued_events.push(NetworkBehaviourAction::GenerateEvent(KademliaOut::Discovered {
peer_id: peer.node_id.clone(),
addresses: peer.multiaddrs.clone(),
ty: peer.connection_ty,
}));
}
if let Some((query, _, _)) = self.active_queries.get_mut(&user_data) {
query.inject_rpc_result(&source, closer_peers.into_iter().map(|kp| kp.node_id))
}
}
KademliaHandlerEvent::GetProvidersReq { key, request_id } => {
self.remote_requests.push((source, request_id, QueryTarget::GetProviders(key)));
return;
}
KademliaHandlerEvent::GetProvidersRes {
closer_peers,
provider_peers,
user_data,
} => {
for peer in closer_peers.iter().chain(provider_peers.iter()) {
if let Some(entry) = self.kbuckets.entry_mut(&peer.node_id) {
for addr in peer.multiaddrs.iter() {
if entry.iter().all(|a| a != addr) {
entry.push(addr.clone());
}
}
}
self.queued_events.push(NetworkBehaviourAction::GenerateEvent(KademliaOut::Discovered {
peer_id: peer.node_id.clone(),
addresses: peer.multiaddrs.clone(),
ty: peer.connection_ty,
}));
}
if let Some((query, _, providers)) = self.active_queries.get_mut(&user_data) {
for peer in provider_peers {
providers.push(peer.node_id);
}
query.inject_rpc_result(&source, closer_peers.into_iter().map(|kp| kp.node_id))
}
}
KademliaHandlerEvent::QueryError { user_data, .. } => {
if let Some((query, _, _)) = self.active_queries.get_mut(&user_data) {
query.inject_rpc_error(&source)
}
}
KademliaHandlerEvent::AddProvider { key, provider_peer } => {
if let Some(entry) = self.kbuckets.entry_mut(&provider_peer.node_id) {
for addr in provider_peer.multiaddrs.iter() {
if entry.iter().all(|a| a != addr) {
entry.push(addr.clone());
}
}
}
self.queued_events.push(NetworkBehaviourAction::GenerateEvent(KademliaOut::Discovered {
peer_id: provider_peer.node_id.clone(),
addresses: provider_peer.multiaddrs.clone(),
ty: provider_peer.connection_ty,
}));
self.add_provider.push((key, provider_peer.node_id));
return;
}
};
}
fn poll(
&mut self,
parameters: &mut PollParameters,
) -> Async<
NetworkBehaviourAction<
<Self::ProtocolsHandler as ProtocolsHandler>::InEvent,
Self::OutEvent,
>,
> {
for (key, provider) in self.add_provider.drain() {
if provider == *self.kbuckets.my_id() {
continue;
}
let providers = self.values_providers.entry(key).or_insert_with(Default::default);
if !providers.iter().any(|k| k == &provider) {
providers.push(provider);
}
}
self.add_provider.shrink_to_fit();
match self.refresh_add_providers.poll() {
Ok(Async::NotReady) => {},
Ok(Async::Ready(Some(_))) => {
for provided in self.providing_keys.clone().into_iter() {
let purpose = QueryPurpose::AddProvider(provided.clone());
if let Ok(key_as_peer) = PeerId::from_multihash(provided) {
self.start_query(QueryTarget::FindPeer(key_as_peer), purpose);
}
}
},
Ok(Async::Ready(None)) | Err(_) => {},
}
for (query_id, query_target, query_purpose) in self.queries_to_starts.drain() {
let known_closest_peers = self.kbuckets
.find_closest(query_target.as_hash())
.take(self.num_results);
self.active_queries.insert(
query_id,
(
QueryState::new(QueryConfig {
target: query_target,
parallelism: self.parallelism,
num_results: self.num_results,
rpc_timeout: self.rpc_timeout,
known_closest_peers,
}),
query_purpose,
Vec::new()
)
);
}
self.queries_to_starts.shrink_to_fit();
if !self.remote_requests.is_empty() {
let (peer_id, request_id, query) = self.remote_requests.remove(0);
let result = self.build_result(query, request_id, parameters);
return Async::Ready(NetworkBehaviourAction::SendEvent {
peer_id,
event: result,
});
}
loop {
if !self.queued_events.is_empty() {
return Async::Ready(self.queued_events.remove(0));
}
self.queued_events.shrink_to_fit();
let mut finished_query = None;
'queries_iter: for (&query_id, (query, _, _)) in self.active_queries.iter_mut() {
loop {
match query.poll() {
Async::Ready(QueryStatePollOut::Finished) => {
finished_query = Some(query_id);
break 'queries_iter;
}
Async::Ready(QueryStatePollOut::SendRpc {
peer_id,
query_target,
}) => {
let rpc = query_target.to_rpc_request(query_id);
if self.connected_peers.contains(&peer_id) {
return Async::Ready(NetworkBehaviourAction::SendEvent {
peer_id: peer_id.clone(),
event: rpc,
});
} else {
self.pending_rpcs.push((peer_id.clone(), rpc));
return Async::Ready(NetworkBehaviourAction::DialPeer {
peer_id: peer_id.clone(),
});
}
}
Async::Ready(QueryStatePollOut::CancelRpc { peer_id }) => {
self.pending_rpcs.retain(|(id, _)| id != peer_id);
}
Async::NotReady => break,
}
}
}
if let Some(finished_query) = finished_query {
let (query, purpose, provider_peers) = self
.active_queries
.remove(&finished_query)
.expect("finished_query was gathered when iterating active_queries; QED.");
match purpose {
QueryPurpose::Initialization => {},
QueryPurpose::UserRequest => {
let event = match query.target().clone() {
QueryTarget::FindPeer(key) => {
debug_assert!(provider_peers.is_empty());
KademliaOut::FindNodeResult {
key,
closer_peers: query.into_closest_peers().collect(),
}
},
QueryTarget::GetProviders(key) => {
KademliaOut::GetProvidersResult {
key,
closer_peers: query.into_closest_peers().collect(),
provider_peers,
}
},
};
break Async::Ready(NetworkBehaviourAction::GenerateEvent(event));
},
QueryPurpose::AddProvider(key) => {
for closest in query.into_closest_peers() {
let event = NetworkBehaviourAction::SendEvent {
peer_id: closest,
event: KademliaHandlerIn::AddProvider {
key: key.clone(),
provider_peer: build_kad_peer(parameters.local_peer_id().clone(), parameters, &self.kbuckets, &self.connected_peers),
},
};
self.queued_events.push(event);
}
},
}
} else {
break Async::NotReady;
}
}
}
}
#[derive(Debug, Clone)]
pub enum KademliaOut {
Discovered {
peer_id: PeerId,
addresses: Vec<Multiaddr>,
ty: KadConnectionType,
},
FindNodeResult {
key: PeerId,
closer_peers: Vec<PeerId>,
},
GetProvidersResult {
key: Multihash,
provider_peers: Vec<PeerId>,
closer_peers: Vec<PeerId>,
},
}
fn gen_random_id(my_id: &PeerId, bucket_num: usize) -> Result<PeerId, ()> {
let my_id_len = my_id.as_bytes().len();
let bits_diff = bucket_num + 1;
if bits_diff > 8 * (my_id_len - 2) {
return Err(());
}
let mut random_id = [0; 64];
for byte in 0..my_id_len {
match byte.cmp(&(my_id_len - bits_diff / 8 - 1)) {
Ordering::Less => {
random_id[byte] = my_id.as_bytes()[byte];
}
Ordering::Equal => {
let mask: u8 = (1 << (bits_diff % 8)) - 1;
random_id[byte] = (my_id.as_bytes()[byte] & !mask) | (rand::random::<u8>() & mask);
}
Ordering::Greater => {
random_id[byte] = rand::random();
}
}
}
let peer_id = PeerId::from_bytes(random_id[..my_id_len].to_owned())
.expect("randomly-generated peer ID should always be valid");
Ok(peer_id)
}
fn build_kad_peer(
peer_id: PeerId,
parameters: &mut PollParameters,
kbuckets: &KBucketsTable<PeerId, SmallVec<[Multiaddr; 4]>>,
connected_peers: &FnvHashSet<PeerId>
) -> KadPeer {
let is_self = peer_id == *parameters.local_peer_id();
let multiaddrs = if is_self {
let mut addrs = parameters
.listened_addresses()
.cloned()
.collect::<Vec<_>>();
addrs.extend(parameters.external_addresses());
addrs
} else {
kbuckets
.get(&peer_id)
.map(|addrs| addrs.iter().cloned().collect())
.unwrap_or_else(Vec::new)
};
let connection_ty = if is_self || connected_peers.contains(&peer_id) {
KadConnectionType::Connected
} else {
KadConnectionType::NotConnected
};
KadPeer {
node_id: peer_id,
multiaddrs,
connection_ty,
}
}