diff options
Diffstat (limited to 'karyons_p2p/src')
26 files changed, 0 insertions, 3702 deletions
diff --git a/karyons_p2p/src/backend.rs b/karyons_p2p/src/backend.rs deleted file mode 100644 index 290e3e7..0000000 --- a/karyons_p2p/src/backend.rs +++ /dev/null @@ -1,139 +0,0 @@ -use std::sync::Arc; - -use log::info; - -use karyons_core::{pubsub::Subscription, Executor}; - -use crate::{ - config::Config, - discovery::{ArcDiscovery, Discovery}, - monitor::{Monitor, MonitorEvent}, - net::ConnQueue, - peer_pool::PeerPool, - protocol::{ArcProtocol, Protocol}, - ArcPeer, PeerID, Result, -}; - -pub type ArcBackend = Arc<Backend>; - -/// Backend serves as the central entry point for initiating and managing -/// the P2P network. -/// -/// -/// # Example -/// ``` -/// use std::sync::Arc; -/// -/// use easy_parallel::Parallel; -/// use smol::{channel as smol_channel, future, Executor}; -/// -/// use karyons_p2p::{Backend, Config, PeerID}; -/// -/// let peer_id = PeerID::random(); -/// -/// // Create the configuration for the backend. -/// let mut config = Config::default(); -/// -/// // Create a new Backend -/// let backend = Backend::new(peer_id, config); -/// -/// // Create a new Executor -/// let ex = Arc::new(Executor::new()); -/// -/// let task = async { -/// // Run the backend -/// backend.run(ex.clone()).await.unwrap(); -/// -/// // .... -/// -/// // Shutdown the backend -/// backend.shutdown().await; -/// }; -/// -/// future::block_on(ex.run(task)); -/// -/// ``` -pub struct Backend { - /// The Configuration for the P2P network. - config: Arc<Config>, - - /// Peer ID. - id: PeerID, - - /// Responsible for network and system monitoring. - monitor: Arc<Monitor>, - - /// Discovery instance. - discovery: ArcDiscovery, - - /// PeerPool instance. - peer_pool: Arc<PeerPool>, -} - -impl Backend { - /// Creates a new Backend. - pub fn new(id: PeerID, config: Config) -> ArcBackend { - let config = Arc::new(config); - let monitor = Arc::new(Monitor::new()); - - let conn_queue = ConnQueue::new(); - - let peer_pool = PeerPool::new(&id, conn_queue.clone(), config.clone(), monitor.clone()); - let discovery = Discovery::new(&id, conn_queue, config.clone(), monitor.clone()); - - Arc::new(Self { - id: id.clone(), - monitor, - discovery, - config, - peer_pool, - }) - } - - /// Run the Backend, starting the PeerPool and Discovery instances. - pub async fn run(self: &Arc<Self>, ex: Executor<'_>) -> Result<()> { - info!("Run the backend {}", self.id); - self.peer_pool.start(ex.clone()).await?; - self.discovery.start(ex.clone()).await?; - Ok(()) - } - - /// Attach a custom protocol to the network - pub async fn attach_protocol<P: Protocol>( - &self, - c: impl Fn(ArcPeer) -> ArcProtocol + Send + Sync + 'static, - ) -> Result<()> { - self.peer_pool.attach_protocol::<P>(Box::new(c)).await - } - - /// Returns the number of currently connected peers. - pub async fn peers(&self) -> usize { - self.peer_pool.peers_len().await - } - - /// Returns the `Config`. - pub fn config(&self) -> Arc<Config> { - self.config.clone() - } - - /// Returns the number of occupied inbound slots. - pub fn inbound_slots(&self) -> usize { - self.discovery.inbound_slots.load() - } - - /// Returns the number of occupied outbound slots. - pub fn outbound_slots(&self) -> usize { - self.discovery.outbound_slots.load() - } - - /// Subscribes to the monitor to receive network events. - pub async fn monitor(&self) -> Subscription<MonitorEvent> { - self.monitor.subscribe().await - } - - /// Shuts down the Backend. - pub async fn shutdown(&self) { - self.discovery.shutdown().await; - self.peer_pool.shutdown().await; - } -} diff --git a/karyons_p2p/src/config.rs b/karyons_p2p/src/config.rs deleted file mode 100644 index ebecbf0..0000000 --- a/karyons_p2p/src/config.rs +++ /dev/null @@ -1,105 +0,0 @@ -use karyons_net::{Endpoint, Port}; - -use crate::utils::Version; - -/// the Configuration for the P2P network. -pub struct Config { - /// Represents the network version. - pub version: Version, - - ///////////////// - // PeerPool - //////////////// - /// Timeout duration for the handshake with new peers, in seconds. - pub handshake_timeout: u64, - /// Interval at which the ping protocol sends ping messages to a peer to - /// maintain connections, in seconds. - pub ping_interval: u64, - /// Timeout duration for receiving the pong message corresponding to the - /// sent ping message, in seconds. - pub ping_timeout: u64, - /// The maximum number of retries for outbound connection establishment. - pub max_connect_retries: usize, - - ///////////////// - // DISCOVERY - //////////////// - /// A list of bootstrap peers for the seeding process. - pub bootstrap_peers: Vec<Endpoint>, - /// An optional listening endpoint to accept incoming connections. - pub listen_endpoint: Option<Endpoint>, - /// A list of endpoints representing peers that the `Discovery` will - /// manually connect to. - pub peer_endpoints: Vec<Endpoint>, - /// The number of available inbound slots for incoming connections. - pub inbound_slots: usize, - /// The number of available outbound slots for outgoing connections. - pub outbound_slots: usize, - /// TCP/UDP port for lookup and refresh processes. - pub discovery_port: Port, - /// Time interval, in seconds, at which the Discovery restarts the - /// seeding process. - pub seeding_interval: u64, - - ///////////////// - // LOOKUP - //////////////// - /// The number of available inbound slots for incoming connections during - /// the lookup process. - pub lookup_inbound_slots: usize, - /// The number of available outbound slots for outgoing connections during - /// the lookup process. - pub lookup_outbound_slots: usize, - /// Timeout duration for a peer response during the lookup process, in - /// seconds. - pub lookup_response_timeout: u64, - /// Maximum allowable time for a live connection with a peer during the - /// lookup process, in seconds. - pub lookup_connection_lifespan: u64, - /// The maximum number of retries for outbound connection establishment - /// during the lookup process. - pub lookup_connect_retries: usize, - - ///////////////// - // REFRESH - //////////////// - /// Interval at which the table refreshes its entries, in seconds. - pub refresh_interval: u64, - /// Timeout duration for a peer response during the table refresh process, - /// in seconds. - pub refresh_response_timeout: u64, - /// The maximum number of retries for outbound connection establishment - /// during the refresh process. - pub refresh_connect_retries: usize, -} - -impl Default for Config { - fn default() -> Self { - Config { - version: "0.1.0".parse().unwrap(), - - handshake_timeout: 2, - ping_interval: 20, - ping_timeout: 2, - - bootstrap_peers: vec![], - listen_endpoint: None, - peer_endpoints: vec![], - inbound_slots: 12, - outbound_slots: 12, - max_connect_retries: 3, - discovery_port: 0, - seeding_interval: 60, - - lookup_inbound_slots: 20, - lookup_outbound_slots: 20, - lookup_response_timeout: 1, - lookup_connection_lifespan: 3, - lookup_connect_retries: 3, - - refresh_interval: 1800, - refresh_response_timeout: 1, - refresh_connect_retries: 3, - } - } -} diff --git a/karyons_p2p/src/discovery/lookup.rs b/karyons_p2p/src/discovery/lookup.rs deleted file mode 100644 index f404133..0000000 --- a/karyons_p2p/src/discovery/lookup.rs +++ /dev/null @@ -1,366 +0,0 @@ -use std::{sync::Arc, time::Duration}; - -use futures_util::{stream::FuturesUnordered, StreamExt}; -use log::{error, trace}; -use rand::{rngs::OsRng, seq::SliceRandom, RngCore}; -use smol::lock::{Mutex, RwLock}; - -use karyons_core::{async_utils::timeout, utils::decode, Executor}; - -use karyons_net::{Conn, Endpoint}; - -use crate::{ - io_codec::IOCodec, - message::{ - get_msg_payload, FindPeerMsg, NetMsg, NetMsgCmd, PeerMsg, PeersMsg, PingMsg, PongMsg, - ShutdownMsg, - }, - monitor::{ConnEvent, DiscoveryEvent, Monitor}, - net::{ConnectionSlots, Connector, Listener}, - routing_table::RoutingTable, - utils::version_match, - Config, Error, PeerID, Result, -}; - -/// Maximum number of peers that can be returned in a PeersMsg. -pub const MAX_PEERS_IN_PEERSMSG: usize = 10; - -pub struct LookupService { - /// Peer's ID - id: PeerID, - - /// Routing Table - table: Arc<Mutex<RoutingTable>>, - - /// Listener - listener: Arc<Listener>, - /// Connector - connector: Arc<Connector>, - - /// Outbound slots. - outbound_slots: Arc<ConnectionSlots>, - - /// Resolved listen endpoint - listen_endpoint: Option<RwLock<Endpoint>>, - - /// Holds the configuration for the P2P network. - config: Arc<Config>, - - /// Responsible for network and system monitoring. - monitor: Arc<Monitor>, -} - -impl LookupService { - /// Creates a new lookup service - pub fn new( - id: &PeerID, - table: Arc<Mutex<RoutingTable>>, - config: Arc<Config>, - monitor: Arc<Monitor>, - ) -> Self { - let inbound_slots = Arc::new(ConnectionSlots::new(config.lookup_inbound_slots)); - let outbound_slots = Arc::new(ConnectionSlots::new(config.lookup_outbound_slots)); - - let listener = Listener::new(inbound_slots.clone(), monitor.clone()); - let connector = Connector::new( - config.lookup_connect_retries, - outbound_slots.clone(), - monitor.clone(), - ); - - let listen_endpoint = config - .listen_endpoint - .as_ref() - .map(|endpoint| RwLock::new(endpoint.clone())); - - Self { - id: id.clone(), - table, - listener, - connector, - outbound_slots, - listen_endpoint, - config, - monitor, - } - } - - /// Start the lookup service. - pub async fn start(self: &Arc<Self>, ex: Executor<'_>) -> Result<()> { - self.start_listener(ex).await?; - Ok(()) - } - - /// Set the resolved listen endpoint. - pub async fn set_listen_endpoint(&self, resolved_endpoint: &Endpoint) { - if let Some(endpoint) = &self.listen_endpoint { - *endpoint.write().await = resolved_endpoint.clone(); - } - } - - /// Shuts down the lookup service. - pub async fn shutdown(&self) { - self.connector.shutdown().await; - self.listener.shutdown().await; - } - - /// Starts iterative lookup and populate the routing table. - /// - /// This method begins by generating a random peer ID and connecting to the - /// provided endpoint. It then sends a FindPeer message containing the - /// randomly generated peer ID. Upon receiving peers from the initial lookup, - /// it starts connecting to these received peers and sends them a FindPeer - /// message that contains our own peer ID. - pub async fn start_lookup(&self, endpoint: &Endpoint) -> Result<()> { - trace!("Lookup started {endpoint}"); - self.monitor - .notify(&DiscoveryEvent::LookupStarted(endpoint.clone()).into()) - .await; - - let mut random_peers = vec![]; - if let Err(err) = self.random_lookup(endpoint, &mut random_peers).await { - self.monitor - .notify(&DiscoveryEvent::LookupFailed(endpoint.clone()).into()) - .await; - return Err(err); - }; - - let mut peer_buffer = vec![]; - self.self_lookup(&random_peers, &mut peer_buffer).await; - - while peer_buffer.len() < MAX_PEERS_IN_PEERSMSG { - match random_peers.pop() { - Some(p) => peer_buffer.push(p), - None => break, - } - } - - for peer in peer_buffer.iter() { - let mut table = self.table.lock().await; - let result = table.add_entry(peer.clone().into()); - trace!("Add entry {:?}", result); - } - - self.monitor - .notify(&DiscoveryEvent::LookupSucceeded(endpoint.clone(), peer_buffer.len()).into()) - .await; - - Ok(()) - } - - /// Starts a random lookup - /// - /// This will perfom lookup on a random generated PeerID - async fn random_lookup( - &self, - endpoint: &Endpoint, - random_peers: &mut Vec<PeerMsg>, - ) -> Result<()> { - for _ in 0..2 { - let peer_id = PeerID::random(); - let peers = self.connect(&peer_id, endpoint.clone()).await?; - for peer in peers { - if random_peers.contains(&peer) - || peer.peer_id == self.id - || self.table.lock().await.contains_key(&peer.peer_id.0) - { - continue; - } - - random_peers.push(peer); - } - } - - Ok(()) - } - - /// Starts a self lookup - async fn self_lookup(&self, random_peers: &Vec<PeerMsg>, peer_buffer: &mut Vec<PeerMsg>) { - let mut tasks = FuturesUnordered::new(); - for peer in random_peers.choose_multiple(&mut OsRng, random_peers.len()) { - let endpoint = Endpoint::Tcp(peer.addr.clone(), peer.discovery_port); - tasks.push(self.connect(&self.id, endpoint)) - } - - while let Some(result) = tasks.next().await { - match result { - Ok(peers) => peer_buffer.extend(peers), - Err(err) => { - error!("Failed to do self lookup: {err}"); - } - } - } - } - - /// Connects to the given endpoint - async fn connect(&self, peer_id: &PeerID, endpoint: Endpoint) -> Result<Vec<PeerMsg>> { - let conn = self.connector.connect(&endpoint).await?; - let io_codec = IOCodec::new(conn); - let result = self.handle_outbound(io_codec, peer_id).await; - - self.monitor - .notify(&ConnEvent::Disconnected(endpoint).into()) - .await; - self.outbound_slots.remove().await; - - result - } - - /// Handles outbound connection - async fn handle_outbound(&self, io_codec: IOCodec, peer_id: &PeerID) -> Result<Vec<PeerMsg>> { - trace!("Send Ping msg"); - self.send_ping_msg(&io_codec).await?; - - trace!("Send FindPeer msg"); - let peers = self.send_findpeer_msg(&io_codec, peer_id).await?; - - if peers.0.len() >= MAX_PEERS_IN_PEERSMSG { - return Err(Error::Lookup("Received too many peers in PeersMsg")); - } - - trace!("Send Peer msg"); - if let Some(endpoint) = &self.listen_endpoint { - self.send_peer_msg(&io_codec, endpoint.read().await.clone()) - .await?; - } - - trace!("Send Shutdown msg"); - self.send_shutdown_msg(&io_codec).await?; - - Ok(peers.0) - } - - /// Start a listener. - async fn start_listener(self: &Arc<Self>, ex: Executor<'_>) -> Result<()> { - let addr = match &self.listen_endpoint { - Some(a) => a.read().await.addr()?.clone(), - None => return Ok(()), - }; - - let endpoint = Endpoint::Tcp(addr, self.config.discovery_port); - - let selfc = self.clone(); - let callback = |conn: Conn| async move { - let t = Duration::from_secs(selfc.config.lookup_connection_lifespan); - timeout(t, selfc.handle_inbound(conn)).await??; - Ok(()) - }; - - self.listener.start(ex, endpoint.clone(), callback).await?; - Ok(()) - } - - /// Handles inbound connection - async fn handle_inbound(self: &Arc<Self>, conn: Conn) -> Result<()> { - let io_codec = IOCodec::new(conn); - loop { - let msg: NetMsg = io_codec.read().await?; - trace!("Receive msg {:?}", msg.header.command); - - if let NetMsgCmd::Shutdown = msg.header.command { - return Ok(()); - } - - match &msg.header.command { - NetMsgCmd::Ping => { - let (ping_msg, _) = decode::<PingMsg>(&msg.payload)?; - if !version_match(&self.config.version.req, &ping_msg.version) { - return Err(Error::IncompatibleVersion("system: {}".into())); - } - self.send_pong_msg(ping_msg.nonce, &io_codec).await?; - } - NetMsgCmd::FindPeer => { - let (findpeer_msg, _) = decode::<FindPeerMsg>(&msg.payload)?; - let peer_id = findpeer_msg.0; - self.send_peers_msg(&peer_id, &io_codec).await?; - } - NetMsgCmd::Peer => { - let (peer, _) = decode::<PeerMsg>(&msg.payload)?; - let result = self.table.lock().await.add_entry(peer.clone().into()); - trace!("Add entry result: {:?}", result); - } - c => return Err(Error::InvalidMsg(format!("Unexpected msg: {:?}", c))), - } - } - } - - /// Sends a Ping msg and wait to receive the Pong message. - async fn send_ping_msg(&self, io_codec: &IOCodec) -> Result<()> { - trace!("Send Pong msg"); - - let mut nonce: [u8; 32] = [0; 32]; - RngCore::fill_bytes(&mut OsRng, &mut nonce); - - let ping_msg = PingMsg { - version: self.config.version.v.clone(), - nonce, - }; - io_codec.write(NetMsgCmd::Ping, &ping_msg).await?; - - let t = Duration::from_secs(self.config.lookup_response_timeout); - let recv_msg: NetMsg = io_codec.read_timeout(t).await?; - - let payload = get_msg_payload!(Pong, recv_msg); - let (pong_msg, _) = decode::<PongMsg>(&payload)?; - - if ping_msg.nonce != pong_msg.0 { - return Err(Error::InvalidPongMsg); - } - - Ok(()) - } - - /// Sends a Pong msg - async fn send_pong_msg(&self, nonce: [u8; 32], io_codec: &IOCodec) -> Result<()> { - trace!("Send Pong msg"); - io_codec.write(NetMsgCmd::Pong, &PongMsg(nonce)).await?; - Ok(()) - } - - /// Sends a FindPeer msg and wait to receivet the Peers msg. - async fn send_findpeer_msg(&self, io_codec: &IOCodec, peer_id: &PeerID) -> Result<PeersMsg> { - trace!("Send FindPeer msg"); - io_codec - .write(NetMsgCmd::FindPeer, &FindPeerMsg(peer_id.clone())) - .await?; - - let t = Duration::from_secs(self.config.lookup_response_timeout); - let recv_msg: NetMsg = io_codec.read_timeout(t).await?; - - let payload = get_msg_payload!(Peers, recv_msg); - let (peers, _) = decode(&payload)?; - - Ok(peers) - } - - /// Sends a Peers msg. - async fn send_peers_msg(&self, peer_id: &PeerID, io_codec: &IOCodec) -> Result<()> { - trace!("Send Peers msg"); - let table = self.table.lock().await; - let entries = table.closest_entries(&peer_id.0, MAX_PEERS_IN_PEERSMSG); - let peers: Vec<PeerMsg> = entries.into_iter().map(|e| e.into()).collect(); - drop(table); - io_codec.write(NetMsgCmd::Peers, &PeersMsg(peers)).await?; - Ok(()) - } - - /// Sends a Peer msg. - async fn send_peer_msg(&self, io_codec: &IOCodec, endpoint: Endpoint) -> Result<()> { - trace!("Send Peer msg"); - let peer_msg = PeerMsg { - addr: endpoint.addr()?.clone(), - port: *endpoint.port()?, - discovery_port: self.config.discovery_port, - peer_id: self.id.clone(), - }; - io_codec.write(NetMsgCmd::Peer, &peer_msg).await?; - Ok(()) - } - - /// Sends a Shutdown msg. - async fn send_shutdown_msg(&self, io_codec: &IOCodec) -> Result<()> { - trace!("Send Shutdown msg"); - io_codec.write(NetMsgCmd::Shutdown, &ShutdownMsg(0)).await?; - Ok(()) - } -} diff --git a/karyons_p2p/src/discovery/mod.rs b/karyons_p2p/src/discovery/mod.rs deleted file mode 100644 index 94b350b..0000000 --- a/karyons_p2p/src/discovery/mod.rs +++ /dev/null @@ -1,262 +0,0 @@ -mod lookup; -mod refresh; - -use std::sync::Arc; - -use log::{error, info}; -use rand::{rngs::OsRng, seq::SliceRandom}; -use smol::lock::Mutex; - -use karyons_core::{ - async_utils::{Backoff, TaskGroup, TaskResult}, - Executor, -}; - -use karyons_net::{Conn, Endpoint}; - -use crate::{ - config::Config, - monitor::Monitor, - net::ConnQueue, - net::{ConnDirection, ConnectionSlots, Connector, Listener}, - routing_table::{ - Entry, EntryStatusFlag, RoutingTable, CONNECTED_ENTRY, DISCONNECTED_ENTRY, PENDING_ENTRY, - UNREACHABLE_ENTRY, UNSTABLE_ENTRY, - }, - Error, PeerID, Result, -}; - -use lookup::LookupService; -use refresh::RefreshService; - -pub type ArcDiscovery = Arc<Discovery>; - -pub struct Discovery { - /// Routing table - table: Arc<Mutex<RoutingTable>>, - - /// Lookup Service - lookup_service: Arc<LookupService>, - - /// Refresh Service - refresh_service: Arc<RefreshService>, - - /// Connector - connector: Arc<Connector>, - /// Listener - listener: Arc<Listener>, - - /// Connection queue - conn_queue: Arc<ConnQueue>, - - /// Inbound slots. - pub(crate) inbound_slots: Arc<ConnectionSlots>, - /// Outbound slots. - pub(crate) outbound_slots: Arc<ConnectionSlots>, - - /// Managing spawned tasks. - task_group: TaskGroup, - - /// Holds the configuration for the P2P network. - config: Arc<Config>, -} - -impl Discovery { - /// Creates a new Discovery - pub fn new( - peer_id: &PeerID, - conn_queue: Arc<ConnQueue>, - config: Arc<Config>, - monitor: Arc<Monitor>, - ) -> ArcDiscovery { - let inbound_slots = Arc::new(ConnectionSlots::new(config.inbound_slots)); - let outbound_slots = Arc::new(ConnectionSlots::new(config.outbound_slots)); - - let table_key = peer_id.0; - let table = Arc::new(Mutex::new(RoutingTable::new(table_key))); - - let refresh_service = RefreshService::new(config.clone(), table.clone(), monitor.clone()); - let lookup_service = - LookupService::new(peer_id, table.clone(), config.clone(), monitor.clone()); - - let connector = Connector::new( - config.max_connect_retries, - outbound_slots.clone(), - monitor.clone(), - ); - let listener = Listener::new(inbound_slots.clone(), monitor.clone()); - - Arc::new(Self { - refresh_service: Arc::new(refresh_service), - lookup_service: Arc::new(lookup_service), - conn_queue, - table, - inbound_slots, - outbound_slots, - connector, - listener, - task_group: TaskGroup::new(), - config, - }) - } - - /// Start the Discovery - pub async fn start(self: &Arc<Self>, ex: Executor<'_>) -> Result<()> { - // Check if the listen_endpoint is provided, and if so, start a listener. - if let Some(endpoint) = &self.config.listen_endpoint { - // Return an error if the discovery port is set to 0. - if self.config.discovery_port == 0 { - return Err(Error::Config( - "Please add a valid discovery port".to_string(), - )); - } - - let resolved_endpoint = self.start_listener(endpoint, ex.clone()).await?; - - if endpoint.addr()? != resolved_endpoint.addr()? { - info!("Resolved listen endpoint: {resolved_endpoint}"); - self.lookup_service - .set_listen_endpoint(&resolved_endpoint) - .await; - self.refresh_service - .set_listen_endpoint(&resolved_endpoint) - .await; - } - } - - // Start the lookup service - self.lookup_service.start(ex.clone()).await?; - // Start the refresh service - self.refresh_service.start(ex.clone()).await?; - - // Attempt to manually connect to peer endpoints provided in the Config. - for endpoint in self.config.peer_endpoints.iter() { - let _ = self.connect(endpoint, None, ex.clone()).await; - } - - // Start connect loop - let selfc = self.clone(); - self.task_group - .spawn(ex.clone(), selfc.connect_loop(ex), |res| async move { - if let TaskResult::Completed(Err(err)) = res { - error!("Connect loop stopped: {err}"); - } - }); - - Ok(()) - } - - /// Shuts down the discovery - pub async fn shutdown(&self) { - self.task_group.cancel().await; - self.connector.shutdown().await; - self.listener.shutdown().await; - - self.refresh_service.shutdown().await; - self.lookup_service.shutdown().await; - } - - /// Start a listener and on success, return the resolved endpoint. - async fn start_listener( - self: &Arc<Self>, - endpoint: &Endpoint, - ex: Executor<'_>, - ) -> Result<Endpoint> { - let selfc = self.clone(); - let callback = |conn: Conn| async move { - selfc.conn_queue.handle(conn, ConnDirection::Inbound).await; - Ok(()) - }; - - let resolved_endpoint = self.listener.start(ex, endpoint.clone(), callback).await?; - Ok(resolved_endpoint) - } - - /// This method will attempt to connect to a peer in the routing table. - /// If the routing table is empty, it will start the seeding process for - /// finding new peers. - /// - /// This will perform a backoff to prevent getting stuck in the loop - /// if the seeding process couldn't find any peers. - async fn connect_loop(self: Arc<Self>, ex: Executor<'_>) -> Result<()> { - let backoff = Backoff::new(500, self.config.seeding_interval * 1000); - loop { - let random_entry = self.random_entry(PENDING_ENTRY).await; - match random_entry { - Some(entry) => { - backoff.reset(); - let endpoint = Endpoint::Tcp(entry.addr, entry.port); - self.connect(&endpoint, Some(entry.key.into()), ex.clone()) - .await; - } - None => { - backoff.sleep().await; - self.start_seeding().await; - } - } - } - } - - /// Connect to the given endpoint using the connector - async fn connect(self: &Arc<Self>, endpoint: &Endpoint, pid: Option<PeerID>, ex: Executor<'_>) { - let selfc = self.clone(); - let pid_cloned = pid.clone(); - let cback = |conn: Conn| async move { - selfc.conn_queue.handle(conn, ConnDirection::Outbound).await; - if let Some(pid) = pid_cloned { - selfc.update_entry(&pid, DISCONNECTED_ENTRY).await; - } - Ok(()) - }; - - let res = self.connector.connect_with_cback(ex, endpoint, cback).await; - - if let Some(pid) = &pid { - match res { - Ok(_) => { - self.update_entry(pid, CONNECTED_ENTRY).await; - } - Err(_) => { - self.update_entry(pid, UNREACHABLE_ENTRY).await; - } - } - } - } - - /// Starts seeding process. - /// - /// This method randomly selects a peer from the routing table and - /// attempts to connect to that peer for the initial lookup. If the routing - /// table doesn't have an available entry, it will connect to one of the - /// provided bootstrap endpoints in the `Config` and initiate the lookup. - async fn start_seeding(&self) { - match self.random_entry(PENDING_ENTRY | CONNECTED_ENTRY).await { - Some(entry) => { - let endpoint = Endpoint::Tcp(entry.addr, entry.discovery_port); - if let Err(err) = self.lookup_service.start_lookup(&endpoint).await { - self.update_entry(&entry.key.into(), UNSTABLE_ENTRY).await; - error!("Failed to do lookup: {endpoint}: {err}"); - } - } - None => { - let peers = &self.config.bootstrap_peers; - for endpoint in peers.choose_multiple(&mut OsRng, peers.len()) { - if let Err(err) = self.lookup_service.start_lookup(endpoint).await { - error!("Failed to do lookup: {endpoint}: {err}"); - } - } - } - } - } - - /// Returns a random entry from routing table. - async fn random_entry(&self, entry_flag: EntryStatusFlag) -> Option<Entry> { - self.table.lock().await.random_entry(entry_flag).cloned() - } - - /// Update the entry status - async fn update_entry(&self, pid: &PeerID, entry_flag: EntryStatusFlag) { - let table = &mut self.table.lock().await; - table.update_entry(&pid.0, entry_flag); - } -} diff --git a/karyons_p2p/src/discovery/refresh.rs b/karyons_p2p/src/discovery/refresh.rs deleted file mode 100644 index 7582c84..0000000 --- a/karyons_p2p/src/discovery/refresh.rs +++ /dev/null @@ -1,289 +0,0 @@ -use std::{sync::Arc, time::Duration}; - -use bincode::{Decode, Encode}; -use log::{error, info, trace}; -use rand::{rngs::OsRng, RngCore}; -use smol::{ - lock::{Mutex, RwLock}, - stream::StreamExt, - Timer, -}; - -use karyons_core::{ - async_utils::{timeout, Backoff, TaskGroup, TaskResult}, - utils::{decode, encode}, - Executor, -}; - -use karyons_net::{dial_udp, listen_udp, Addr, Connection, Endpoint, NetError, Port, UdpConn}; - -/// Maximum failures for an entry before removing it from the routing table. -pub const MAX_FAILURES: u32 = 3; - -/// Ping message size -const PINGMSG_SIZE: usize = 32; - -use crate::{ - monitor::{ConnEvent, DiscoveryEvent, Monitor}, - routing_table::{BucketEntry, Entry, RoutingTable, PENDING_ENTRY, UNREACHABLE_ENTRY}, - Config, Error, Result, -}; - -#[derive(Decode, Encode, Debug, Clone)] -pub struct PingMsg(pub [u8; 32]); - -#[derive(Decode, Encode, Debug)] -pub struct PongMsg(pub [u8; 32]); - -pub struct RefreshService { - /// Routing table - table: Arc<Mutex<RoutingTable>>, - - /// Resolved listen endpoint - listen_endpoint: Option<RwLock<Endpoint>>, - - /// Managing spawned tasks. - task_group: TaskGroup, - - /// Holds the configuration for the P2P network. - config: Arc<Config>, - - /// Responsible for network and system monitoring. - monitor: Arc<Monitor>, -} - -impl RefreshService { - /// Creates a new refresh service - pub fn new( - config: Arc<Config>, - table: Arc<Mutex<RoutingTable>>, - monitor: Arc<Monitor>, - ) -> Self { - let listen_endpoint = config - .listen_endpoint - .as_ref() - .map(|endpoint| RwLock::new(endpoint.clone())); - - Self { - table, - listen_endpoint, - task_group: TaskGroup::new(), - config, - monitor, - } - } - - /// Start the refresh service - pub async fn start(self: &Arc<Self>, ex: Executor<'_>) -> Result<()> { - if let Some(endpoint) = &self.listen_endpoint { - let endpoint = endpoint.read().await; - let addr = endpoint.addr()?; - let port = self.config.discovery_port; - - let selfc = self.clone(); - self.task_group.spawn( - ex.clone(), - selfc.listen_loop(addr.clone(), port), - |res| async move { - if let TaskResult::Completed(Err(err)) = res { - error!("Listen loop stopped: {err}"); - } - }, - ); - } - - let selfc = self.clone(); - self.task_group.spawn( - ex.clone(), - selfc.refresh_loop(ex.clone()), - |res| async move { - if let TaskResult::Completed(Err(err)) = res { - error!("Refresh loop stopped: {err}"); - } - }, - ); - - Ok(()) - } - - /// Set the resolved listen endpoint. - pub async fn set_listen_endpoint(&self, resolved_endpoint: &Endpoint) { - if let Some(endpoint) = &self.listen_endpoint { - *endpoint.write().await = resolved_endpoint.clone(); - } - } - - /// Shuts down the refresh service - pub async fn shutdown(&self) { - self.task_group.cancel().await; - } - - /// Initiates periodic refreshing of the routing table. This function will - /// select 8 random entries from each bucket in the routing table and start - /// sending Ping messages to the entries. - async fn refresh_loop(self: Arc<Self>, ex: Executor<'_>) -> Result<()> { - let mut timer = Timer::interval(Duration::from_secs(self.config.refresh_interval)); - loop { - timer.next().await; - trace!("Start refreshing the routing table..."); - - self.monitor - .notify(&DiscoveryEvent::RefreshStarted.into()) - .await; - - let table = self.table.lock().await; - let mut entries: Vec<BucketEntry> = vec![]; - for bucket in table.iter() { - for entry in bucket.random_iter(8) { - entries.push(entry.clone()) - } - } - drop(table); - - self.clone().do_refresh(&entries, ex.clone()).await; - } - } - - /// Iterates over the entries and spawns a new task for each entry to - /// initiate a connection attempt to that entry. - async fn do_refresh(self: Arc<Self>, entries: &[BucketEntry], ex: Executor<'_>) { - for chunk in entries.chunks(16) { - let mut tasks = Vec::new(); - for bucket_entry in chunk { - if bucket_entry.is_connected() { - continue; - } - - if bucket_entry.failures >= MAX_FAILURES { - self.table - .lock() - .await - .remove_entry(&bucket_entry.entry.key); - return; - } - - tasks.push(ex.spawn(self.clone().refresh_entry(bucket_entry.clone()))) - } - - for task in tasks { - task.await; - } - } - } - - /// Initiates refresh for a specific entry within the routing table. It - /// updates the routing table according to the result. - async fn refresh_entry(self: Arc<Self>, bucket_entry: BucketEntry) { - let key = &bucket_entry.entry.key; - match self.connect(&bucket_entry.entry).await { - Ok(_) => { - self.table.lock().await.update_entry(key, PENDING_ENTRY); - } - Err(err) => { - trace!("Failed to refresh entry {:?}: {err}", key); - let table = &mut self.table.lock().await; - if bucket_entry.failures >= MAX_FAILURES { - table.remove_entry(key); - return; - } - table.update_entry(key, UNREACHABLE_ENTRY); - } - } - } - - /// Initiates a UDP connection with the entry and attempts to send a Ping - /// message. If it fails, it retries according to the allowed retries - /// specified in the Config, with backoff between each retry. - async fn connect(&self, entry: &Entry) -> Result<()> { - let mut retry = 0; - let conn = dial_udp(&entry.addr, &entry.discovery_port).await?; - let backoff = Backoff::new(100, 5000); - while retry < self.config.refresh_connect_retries { - match self.send_ping_msg(&conn).await { - Ok(()) => return Ok(()), - Err(Error::KaryonsNet(NetError::Timeout)) => { - retry += 1; - backoff.sleep().await; - } - Err(err) => { - return Err(err); - } - } - } - - Err(NetError::Timeout.into()) - } - - /// Set up a UDP listener and start listening for Ping messages from other - /// peers. - async fn listen_loop(self: Arc<Self>, addr: Addr, port: Port) -> Result<()> { - let endpoint = Endpoint::Udp(addr.clone(), port); - let conn = match listen_udp(&addr, &port).await { - Ok(c) => { - self.monitor - .notify(&ConnEvent::Listening(endpoint.clone()).into()) - .await; - c - } - Err(err) => { - self.monitor - .notify(&ConnEvent::ListenFailed(endpoint.clone()).into()) - .await; - return Err(err.into()); - } - }; - info!("Start listening on {endpoint}"); - - loop { - let res = self.listen_to_ping_msg(&conn).await; - if let Err(err) = res { - trace!("Failed to handle ping msg {err}"); - self.monitor.notify(&ConnEvent::AcceptFailed.into()).await; - } - } - } - - /// Listen to receive a Ping message and respond with a Pong message. - async fn listen_to_ping_msg(&self, conn: &UdpConn) -> Result<()> { - let mut buf = [0; PINGMSG_SIZE]; - let (_, endpoint) = conn.recv_from(&mut buf).await?; - - self.monitor - .notify(&ConnEvent::Accepted(endpoint.clone()).into()) - .await; - - let (ping_msg, _) = decode::<PingMsg>(&buf)?; - - let pong_msg = PongMsg(ping_msg.0); - let buffer = encode(&pong_msg)?; - - conn.send_to(&buffer, &endpoint).await?; - - self.monitor - .notify(&ConnEvent::Disconnected(endpoint.clone()).into()) - .await; - Ok(()) - } - - /// Sends a Ping msg and wait to receive the Pong message. - async fn send_ping_msg(&self, conn: &UdpConn) -> Result<()> { - let mut nonce: [u8; 32] = [0; 32]; - RngCore::fill_bytes(&mut OsRng, &mut nonce); - - let ping_msg = PingMsg(nonce); - let buffer = encode(&ping_msg)?; - conn.send(&buffer).await?; - - let buf = &mut [0; PINGMSG_SIZE]; - let t = Duration::from_secs(self.config.refresh_response_timeout); - timeout(t, conn.recv(buf)).await??; - - let (pong_msg, _) = decode::<PongMsg>(buf)?; - - if ping_msg.0 != pong_msg.0 { - return Err(Error::InvalidPongMsg); - } - - Ok(()) - } -} diff --git a/karyons_p2p/src/error.rs b/karyons_p2p/src/error.rs deleted file mode 100644 index 945e90a..0000000 --- a/karyons_p2p/src/error.rs +++ /dev/null @@ -1,82 +0,0 @@ -use thiserror::Error as ThisError; - -pub type Result<T> = std::result::Result<T, Error>; - -/// Represents Karyons's p2p Error. -#[derive(ThisError, Debug)] -pub enum Error { - #[error("IO Error: {0}")] - IO(#[from] std::io::Error), - - #[error("Unsupported protocol error: {0}")] - UnsupportedProtocol(String), - - #[error("Invalid message error: {0}")] - InvalidMsg(String), - - #[error("Parse error: {0}")] - ParseError(String), - - #[error("Incompatible version error: {0}")] - IncompatibleVersion(String), - - #[error("Config error: {0}")] - Config(String), - - #[error("Peer shutdown")] - PeerShutdown, - - #[error("Invalid Pong Msg")] - InvalidPongMsg, - - #[error("Discovery error: {0}")] - Discovery(&'static str), - - #[error("Lookup error: {0}")] - Lookup(&'static str), - - #[error("Peer already connected")] - PeerAlreadyConnected, - - #[error("Channel Send Error: {0}")] - ChannelSend(String), - - #[error("Channel Receive Error: {0}")] - ChannelRecv(String), - - #[error("CORE::ERROR : {0}")] - KaryonsCore(#[from] karyons_core::error::Error), - - #[error("NET::ERROR : {0}")] - KaryonsNet(#[from] karyons_net::NetError), -} - -impl<T> From<smol::channel::SendError<T>> for Error { - fn from(error: smol::channel::SendError<T>) -> Self { - Error::ChannelSend(error.to_string()) - } -} - -impl From<smol::channel::RecvError> for Error { - fn from(error: smol::channel::RecvError) -> Self { - Error::ChannelRecv(error.to_string()) - } -} - -impl From<std::num::ParseIntError> for Error { - fn from(error: std::num::ParseIntError) -> Self { - Error::ParseError(error.to_string()) - } -} - -impl From<std::num::ParseFloatError> for Error { - fn from(error: std::num::ParseFloatError) -> Self { - Error::ParseError(error.to_string()) - } -} - -impl From<semver::Error> for Error { - fn from(error: semver::Error) -> Self { - Error::ParseError(error.to_string()) - } -} diff --git a/karyons_p2p/src/io_codec.rs b/karyons_p2p/src/io_codec.rs deleted file mode 100644 index 4515832..0000000 --- a/karyons_p2p/src/io_codec.rs +++ /dev/null @@ -1,102 +0,0 @@ -use std::time::Duration; - -use bincode::{Decode, Encode}; - -use karyons_core::{ - async_utils::timeout, - utils::{decode, encode, encode_into_slice}, -}; - -use karyons_net::{Connection, NetError}; - -use crate::{ - message::{NetMsg, NetMsgCmd, NetMsgHeader, MAX_ALLOWED_MSG_SIZE, MSG_HEADER_SIZE}, - Error, Result, -}; - -pub trait CodecMsg: Decode + Encode + std::fmt::Debug {} -impl<T: Encode + Decode + std::fmt::Debug> CodecMsg for T {} - -/// I/O codec working with generic network connections. -/// -/// It is responsible for both decoding data received from the network and -/// encoding data before sending it. -pub struct IOCodec { - conn: Box<dyn Connection>, -} - -impl IOCodec { - /// Creates a new IOCodec. - pub fn new(conn: Box<dyn Connection>) -> Self { - Self { conn } - } - - /// Reads a message of type `NetMsg` from the connection. - /// - /// It reads the first 6 bytes as the header of the message, then reads - /// and decodes the remaining message data based on the determined header. - pub async fn read(&self) -> Result<NetMsg> { - // Read 6 bytes to get the header of the incoming message - let mut buf = [0; MSG_HEADER_SIZE]; - self.conn.recv(&mut buf).await?; - - // Decode the header from bytes to NetMsgHeader - let (header, _) = decode::<NetMsgHeader>(&buf)?; - - if header.payload_size > MAX_ALLOWED_MSG_SIZE { - return Err(Error::InvalidMsg( - "Message exceeds the maximum allowed size".to_string(), - )); - } - - // Create a buffer to hold the message based on its length - let mut payload = vec![0; header.payload_size as usize]; - self.conn.recv(&mut payload).await?; - - Ok(NetMsg { header, payload }) - } - - /// Writes a message of type `T` to the connection. - /// - /// Before appending the actual message payload, it calculates the length of - /// the encoded message in bytes and appends this length to the message header. - pub async fn write<T: CodecMsg>(&self, command: NetMsgCmd, msg: &T) -> Result<()> { - let payload = encode(msg)?; - - // Create a buffer to hold the message header (6 bytes) - let header_buf = &mut [0; MSG_HEADER_SIZE]; - let header = NetMsgHeader { - command, - payload_size: payload.len() as u32, - }; - encode_into_slice(&header, header_buf)?; - - let mut buffer = vec![]; - // Append the header bytes to the buffer - buffer.extend_from_slice(header_buf); - // Append the message payload to the buffer - buffer.extend_from_slice(&payload); - - self.conn.send(&buffer).await?; - Ok(()) - } - - /// Reads a message of type `NetMsg` with the given timeout. - pub async fn read_timeout(&self, duration: Duration) -> Result<NetMsg> { - timeout(duration, self.read()) - .await - .map_err(|_| NetError::Timeout)? - } - - /// Writes a message of type `T` with the given timeout. - pub async fn write_timeout<T: CodecMsg>( - &self, - command: NetMsgCmd, - msg: &T, - duration: Duration, - ) -> Result<()> { - timeout(duration, self.write(command, msg)) - .await - .map_err(|_| NetError::Timeout)? - } -} diff --git a/karyons_p2p/src/lib.rs b/karyons_p2p/src/lib.rs deleted file mode 100644 index 08ba059..0000000 --- a/karyons_p2p/src/lib.rs +++ /dev/null @@ -1,27 +0,0 @@ -mod backend; -mod config; -mod discovery; -mod error; -mod io_codec; -mod message; -mod net; -mod peer; -mod peer_pool; -mod protocols; -mod routing_table; -mod utils; - -/// Responsible for network and system monitoring. -/// [`Read More`](./monitor/struct.Monitor.html) -pub mod monitor; -/// Defines the protocol trait. -/// [`Read More`](./protocol/trait.Protocol.html) -pub mod protocol; - -pub use backend::{ArcBackend, Backend}; -pub use config::Config; -pub use error::Error as P2pError; -pub use peer::{ArcPeer, PeerID}; -pub use utils::Version; - -use error::{Error, Result}; diff --git a/karyons_p2p/src/message.rs b/karyons_p2p/src/message.rs deleted file mode 100644 index 833f6f4..0000000 --- a/karyons_p2p/src/message.rs +++ /dev/null @@ -1,133 +0,0 @@ -use std::collections::HashMap; - -use bincode::{Decode, Encode}; - -use karyons_net::{Addr, Port}; - -use crate::{protocol::ProtocolID, routing_table::Entry, utils::VersionInt, PeerID}; - -/// The size of the message header, in bytes. -pub const MSG_HEADER_SIZE: usize = 6; - -/// The maximum allowed size for a message in bytes. -pub const MAX_ALLOWED_MSG_SIZE: u32 = 1000000; - -/// Defines the main message in the Karyon P2P network. -/// -/// This message structure consists of a header and payload, where the header -/// typically contains essential information about the message, and the payload -/// contains the actual data being transmitted. -#[derive(Decode, Encode, Debug, Clone)] -pub struct NetMsg { - pub header: NetMsgHeader, - pub payload: Vec<u8>, -} - -/// Represents the header of a message. -#[derive(Decode, Encode, Debug, Clone)] -pub struct NetMsgHeader { - pub command: NetMsgCmd, - pub payload_size: u32, -} - -/// Defines message commands. -#[derive(Decode, Encode, Debug, Clone)] -#[repr(u8)] -pub enum NetMsgCmd { - Version, - Verack, - Protocol, - Shutdown, - - // NOTE: The following commands are used during the lookup process. - Ping, - Pong, - FindPeer, - Peer, - Peers, -} - -/// Defines a message related to a specific protocol. -#[derive(Decode, Encode, Debug, Clone)] -pub struct ProtocolMsg { - pub protocol_id: ProtocolID, - pub payload: Vec<u8>, -} - -/// Version message, providing information about a peer's capabilities. -#[derive(Decode, Encode, Debug, Clone)] -pub struct VerMsg { - pub peer_id: PeerID, - pub version: VersionInt, - pub protocols: HashMap<ProtocolID, VersionInt>, -} - -/// VerAck message acknowledging the receipt of a Version message. -#[derive(Decode, Encode, Debug, Clone)] -pub struct VerAckMsg(pub PeerID); - -/// Shutdown message. -#[derive(Decode, Encode, Debug, Clone)] -pub struct ShutdownMsg(pub u8); - -/// Ping message with a nonce and version information. -#[derive(Decode, Encode, Debug, Clone)] -pub struct PingMsg { - pub nonce: [u8; 32], - pub version: VersionInt, -} - -/// Ping message with a nonce. -#[derive(Decode, Encode, Debug)] -pub struct PongMsg(pub [u8; 32]); - -/// FindPeer message used to find a specific peer. -#[derive(Decode, Encode, Debug)] -pub struct FindPeerMsg(pub PeerID); - -/// PeerMsg containing information about a peer. -#[derive(Decode, Encode, Debug, Clone, PartialEq, Eq)] -pub struct PeerMsg { - pub peer_id: PeerID, - pub addr: Addr, - pub port: Port, - pub discovery_port: Port, -} - -/// PeersMsg a list of `PeerMsg`. -#[derive(Decode, Encode, Debug)] -pub struct PeersMsg(pub Vec<PeerMsg>); - -macro_rules! get_msg_payload { - ($a:ident, $b:expr) => { - if let NetMsgCmd::$a = $b.header.command { - $b.payload - } else { - return Err(Error::InvalidMsg(format!("Unexpected msg{:?}", $b))); - } - }; -} - -pub(super) use get_msg_payload; - -impl From<Entry> for PeerMsg { - fn from(entry: Entry) -> PeerMsg { - PeerMsg { - peer_id: PeerID(entry.key), - addr: entry.addr, - port: entry.port, - discovery_port: entry.discovery_port, - } - } -} - -impl From<PeerMsg> for Entry { - fn from(peer: PeerMsg) -> Entry { - Entry { - key: peer.peer_id.0, - addr: peer.addr, - port: peer.port, - discovery_port: peer.discovery_port, - } - } -} diff --git a/karyons_p2p/src/monitor.rs b/karyons_p2p/src/monitor.rs deleted file mode 100644 index ee0bf44..0000000 --- a/karyons_p2p/src/monitor.rs +++ /dev/null @@ -1,154 +0,0 @@ -use std::fmt; - -use crate::PeerID; - -use karyons_core::pubsub::{ArcPublisher, Publisher, Subscription}; - -use karyons_net::Endpoint; - -/// Responsible for network and system monitoring. -/// -/// It use pub-sub pattern to notify the subscribers with new events. -/// -/// # Example -/// -/// ``` -/// use karyons_p2p::{Config, Backend, PeerID}; -/// async { -/// -/// let backend = Backend::new(PeerID::random(), Config::default()); -/// -/// // Create a new Subscription -/// let sub = backend.monitor().await; -/// -/// let event = sub.recv().await; -/// }; -/// ``` -pub struct Monitor { - inner: ArcPublisher<MonitorEvent>, -} - -impl Monitor { - /// Creates a new Monitor - pub(crate) fn new() -> Monitor { - Self { - inner: Publisher::new(), - } - } - - /// Sends a new monitor event to all subscribers. - pub async fn notify(&self, event: &MonitorEvent) { - self.inner.notify(event).await; - } - - /// Subscribes to listen to new events. - pub async fn subscribe(&self) -> Subscription<MonitorEvent> { - self.inner.subscribe().await - } -} - -/// Defines various type of event that can be monitored. -#[derive(Clone, Debug)] -pub enum MonitorEvent { - Conn(ConnEvent), - PeerPool(PeerPoolEvent), - Discovery(DiscoveryEvent), -} - -/// Defines connection-related events. -#[derive(Clone, Debug)] -pub enum ConnEvent { - Connected(Endpoint), - ConnectRetried(Endpoint), - ConnectFailed(Endpoint), - Accepted(Endpoint), - AcceptFailed, - Disconnected(Endpoint), - Listening(Endpoint), - ListenFailed(Endpoint), -} - -/// Defines `PeerPool` events. -#[derive(Clone, Debug)] -pub enum PeerPoolEvent { - NewPeer(PeerID), - RemovePeer(PeerID), -} - -/// Defines `Discovery` events. -#[derive(Clone, Debug)] -pub enum DiscoveryEvent { - LookupStarted(Endpoint), - LookupFailed(Endpoint), - LookupSucceeded(Endpoint, usize), - RefreshStarted, -} - -impl fmt::Display for MonitorEvent { - fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { - let val = match self { - MonitorEvent::Conn(e) => format!("Connection Event: {e}"), - MonitorEvent::PeerPool(e) => format!("PeerPool Event: {e}"), - MonitorEvent::Discovery(e) => format!("Discovery Event: {e}"), - }; - write!(f, "{}", val) - } -} - -impl fmt::Display for ConnEvent { - fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { - let val = match self { - ConnEvent::Connected(endpoint) => format!("Connected: {endpoint}"), - ConnEvent::ConnectFailed(endpoint) => format!("ConnectFailed: {endpoint}"), - ConnEvent::ConnectRetried(endpoint) => format!("ConnectRetried: {endpoint}"), - ConnEvent::AcceptFailed => "AcceptFailed".to_string(), - ConnEvent::Accepted(endpoint) => format!("Accepted: {endpoint}"), - ConnEvent::Disconnected(endpoint) => format!("Disconnected: {endpoint}"), - ConnEvent::Listening(endpoint) => format!("Listening: {endpoint}"), - ConnEvent::ListenFailed(endpoint) => format!("ListenFailed: {endpoint}"), - }; - write!(f, "{}", val) - } -} - -impl fmt::Display for PeerPoolEvent { - fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { - let val = match self { - PeerPoolEvent::NewPeer(pid) => format!("NewPeer: {pid}"), - PeerPoolEvent::RemovePeer(pid) => format!("RemovePeer: {pid}"), - }; - write!(f, "{}", val) - } -} - -impl fmt::Display for DiscoveryEvent { - fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { - let val = match self { - DiscoveryEvent::LookupStarted(endpoint) => format!("LookupStarted: {endpoint}"), - DiscoveryEvent::LookupFailed(endpoint) => format!("LookupFailed: {endpoint}"), - DiscoveryEvent::LookupSucceeded(endpoint, len) => { - format!("LookupSucceeded: {endpoint} {len}") - } - DiscoveryEvent::RefreshStarted => "RefreshStarted".to_string(), - }; - write!(f, "{}", val) - } -} - -impl From<ConnEvent> for MonitorEvent { - fn from(val: ConnEvent) -> Self { - MonitorEvent::Conn(val) - } -} - -impl From<PeerPoolEvent> for MonitorEvent { - fn from(val: PeerPoolEvent) -> Self { - MonitorEvent::PeerPool(val) - } -} - -impl From<DiscoveryEvent> for MonitorEvent { - fn from(val: DiscoveryEvent) -> Self { - MonitorEvent::Discovery(val) - } -} diff --git a/karyons_p2p/src/net/connection_queue.rs b/karyons_p2p/src/net/connection_queue.rs deleted file mode 100644 index fbc4bfc..0000000 --- a/karyons_p2p/src/net/connection_queue.rs +++ /dev/null @@ -1,52 +0,0 @@ -use std::sync::Arc; - -use smol::{channel::Sender, lock::Mutex}; - -use karyons_core::async_utils::CondVar; - -use karyons_net::Conn; - -use crate::net::ConnDirection; - -pub struct NewConn { - pub direction: ConnDirection, - pub conn: Conn, - pub disconnect_signal: Sender<()>, -} - -/// Connection queue -pub struct ConnQueue { - queue: Mutex<Vec<NewConn>>, - conn_available: CondVar, -} - -impl ConnQueue { - pub fn new() -> Arc<Self> { - Arc::new(Self { - queue: Mutex::new(Vec::new()), - conn_available: CondVar::new(), - }) - } - - /// Push a connection into the queue and wait for the disconnect signal - pub async fn handle(&self, conn: Conn, direction: ConnDirection) { - let (disconnect_signal, chan) = smol::channel::bounded(1); - let new_conn = NewConn { - direction, - conn, - disconnect_signal, - }; - self.queue.lock().await.push(new_conn); - self.conn_available.signal(); - let _ = chan.recv().await; - } - - /// Receive the next connection in the queue - pub async fn next(&self) -> NewConn { - let mut queue = self.queue.lock().await; - while queue.is_empty() { - queue = self.conn_available.wait(queue).await; - } - queue.pop().unwrap() - } -} diff --git a/karyons_p2p/src/net/connector.rs b/karyons_p2p/src/net/connector.rs deleted file mode 100644 index 72dc0d8..0000000 --- a/karyons_p2p/src/net/connector.rs +++ /dev/null @@ -1,125 +0,0 @@ -use std::{future::Future, sync::Arc}; - -use log::{trace, warn}; - -use karyons_core::{ - async_utils::{Backoff, TaskGroup, TaskResult}, - Executor, -}; -use karyons_net::{dial, Conn, Endpoint, NetError}; - -use crate::{ - monitor::{ConnEvent, Monitor}, - Result, -}; - -use super::slots::ConnectionSlots; - -/// Responsible for creating outbound connections with other peers. -pub struct Connector { - /// Managing spawned tasks. - task_group: TaskGroup, - - /// Manages available outbound slots. - connection_slots: Arc<ConnectionSlots>, - - /// The maximum number of retries allowed before successfully - /// establishing a connection. - max_retries: usize, - - /// Responsible for network and system monitoring. - monitor: Arc<Monitor>, -} - -impl Connector { - /// Creates a new Connector - pub fn new( - max_retries: usize, - connection_slots: Arc<ConnectionSlots>, - monitor: Arc<Monitor>, - ) -> Arc<Self> { - Arc::new(Self { - task_group: TaskGroup::new(), - monitor, - connection_slots, - max_retries, - }) - } - - /// Shuts down the connector - pub async fn shutdown(&self) { - self.task_group.cancel().await; - } - - /// Establish a connection to the specified `endpoint`. If the connection - /// attempt fails, it performs a backoff and retries until the maximum allowed - /// number of retries is exceeded. On a successful connection, it returns a - /// `Conn` instance. - /// - /// This method will block until it finds an available slot. - pub async fn connect(&self, endpoint: &Endpoint) -> Result<Conn> { - self.connection_slots.wait_for_slot().await; - self.connection_slots.add(); - - let mut retry = 0; - let backoff = Backoff::new(500, 2000); - while retry < self.max_retries { - let conn_result = dial(endpoint).await; - - if let Ok(conn) = conn_result { - self.monitor - .notify(&ConnEvent::Connected(endpoint.clone()).into()) - .await; - return Ok(conn); - } - - self.monitor - .notify(&ConnEvent::ConnectRetried(endpoint.clone()).into()) - .await; - - backoff.sleep().await; - - warn!("try to reconnect {endpoint}"); - retry += 1; - } - - self.monitor - .notify(&ConnEvent::ConnectFailed(endpoint.clone()).into()) - .await; - - self.connection_slots.remove().await; - Err(NetError::Timeout.into()) - } - - /// Establish a connection to the given `endpoint`. For each new connection, - /// it invokes the provided `callback`, and pass the connection to the callback. - pub async fn connect_with_cback<'a, Fut>( - self: &Arc<Self>, - ex: Executor<'a>, - endpoint: &Endpoint, - callback: impl FnOnce(Conn) -> Fut + Send + 'a, - ) -> Result<()> - where - Fut: Future<Output = Result<()>> + Send + 'a, - { - let conn = self.connect(endpoint).await?; - - let selfc = self.clone(); - let endpoint = endpoint.clone(); - let on_disconnect = |res| async move { - if let TaskResult::Completed(Err(err)) = res { - trace!("Outbound connection dropped: {err}"); - } - selfc - .monitor - .notify(&ConnEvent::Disconnected(endpoint.clone()).into()) - .await; - selfc.connection_slots.remove().await; - }; - - self.task_group - .spawn(ex.clone(), callback(conn), on_disconnect); - - Ok(()) - } -} diff --git a/karyons_p2p/src/net/listener.rs b/karyons_p2p/src/net/listener.rs deleted file mode 100644 index d1a7bfb..0000000 --- a/karyons_p2p/src/net/listener.rs +++ /dev/null @@ -1,142 +0,0 @@ -use std::{future::Future, sync::Arc}; - -use log::{error, info, trace}; - -use karyons_core::{ - async_utils::{TaskGroup, TaskResult}, - Executor, -}; - -use karyons_net::{listen, Conn, Endpoint, Listener as NetListener}; - -use crate::{ - monitor::{ConnEvent, Monitor}, - Result, -}; - -use super::slots::ConnectionSlots; - -/// Responsible for creating inbound connections with other peers. -pub struct Listener { - /// Managing spawned tasks. - task_group: TaskGroup, - - /// Manages available inbound slots. - connection_slots: Arc<ConnectionSlots>, - - /// Responsible for network and system monitoring. - monitor: Arc<Monitor>, -} - -impl Listener { - /// Creates a new Listener - pub fn new(connection_slots: Arc<ConnectionSlots>, monitor: Arc<Monitor>) -> Arc<Self> { - Arc::new(Self { - connection_slots, - task_group: TaskGroup::new(), - monitor, - }) - } - - /// Starts a listener on the given `endpoint`. For each incoming connection - /// that is accepted, it invokes the provided `callback`, and pass the - /// connection to the callback. - /// - /// Returns the resloved listening endpoint. - pub async fn start<'a, Fut>( - self: &Arc<Self>, - ex: Executor<'a>, - endpoint: Endpoint, - // https://github.com/rust-lang/rfcs/pull/2132 - callback: impl FnOnce(Conn) -> Fut + Clone + Send + 'a, - ) -> Result<Endpoint> - where - Fut: Future<Output = Result<()>> + Send + 'a, - { - let listener = match listen(&endpoint).await { - Ok(listener) => { - self.monitor - .notify(&ConnEvent::Listening(endpoint.clone()).into()) - .await; - listener - } - Err(err) => { - error!("Failed to listen on {endpoint}: {err}"); - self.monitor - .notify(&ConnEvent::ListenFailed(endpoint).into()) - .await; - return Err(err.into()); - } - }; - - let resolved_endpoint = listener.local_endpoint()?; - - info!("Start listening on {endpoint}"); - - let selfc = self.clone(); - self.task_group.spawn( - ex.clone(), - selfc.listen_loop(ex.clone(), listener, callback), - |res| async move { - if let TaskResult::Completed(Err(err)) = res { - error!("Listen loop stopped: {endpoint} {err}"); - } - }, - ); - Ok(resolved_endpoint) - } - - /// Shuts down the listener - pub async fn shutdown(&self) { - self.task_group.cancel().await; - } - - async fn listen_loop<'a, Fut>( - self: Arc<Self>, - ex: Executor<'a>, - listener: Box<dyn NetListener>, - callback: impl FnOnce(Conn) -> Fut + Clone + Send + 'a, - ) -> Result<()> - where - Fut: Future<Output = Result<()>> + Send + 'a, - { - loop { - // Wait for an available inbound slot. - self.connection_slots.wait_for_slot().await; - let result = listener.accept().await; - - let conn = match result { - Ok(c) => { - self.monitor - .notify(&ConnEvent::Accepted(c.peer_endpoint()?).into()) - .await; - c - } - Err(err) => { - error!("Failed to accept a new connection: {err}"); - self.monitor.notify(&ConnEvent::AcceptFailed.into()).await; - return Err(err.into()); - } - }; - - self.connection_slots.add(); - - let selfc = self.clone(); - let endpoint = conn.peer_endpoint()?; - let on_disconnect = |res| async move { - if let TaskResult::Completed(Err(err)) = res { - trace!("Inbound connection dropped: {err}"); - } - selfc - .monitor - .notify(&ConnEvent::Disconnected(endpoint).into()) - .await; - selfc.connection_slots.remove().await; - }; - - let callback = callback.clone(); - self.task_group - .spawn(ex.clone(), callback(conn), on_disconnect); - } - } -} diff --git a/karyons_p2p/src/net/mod.rs b/karyons_p2p/src/net/mod.rs deleted file mode 100644 index 9cdc748..0000000 --- a/karyons_p2p/src/net/mod.rs +++ /dev/null @@ -1,27 +0,0 @@ -mod connection_queue; -mod connector; -mod listener; -mod slots; - -pub use connection_queue::ConnQueue; -pub use connector::Connector; -pub use listener::Listener; -pub use slots::ConnectionSlots; - -use std::fmt; - -/// Defines the direction of a network connection. -#[derive(Clone, Debug)] -pub enum ConnDirection { - Inbound, - Outbound, -} - -impl fmt::Display for ConnDirection { - fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { - match self { - ConnDirection::Inbound => write!(f, "Inbound"), - ConnDirection::Outbound => write!(f, "Outbound"), - } - } -} diff --git a/karyons_p2p/src/net/slots.rs b/karyons_p2p/src/net/slots.rs deleted file mode 100644 index 99f0a78..0000000 --- a/karyons_p2p/src/net/slots.rs +++ /dev/null @@ -1,54 +0,0 @@ -use std::sync::atomic::{AtomicUsize, Ordering}; - -use karyons_core::async_utils::CondWait; - -/// Manages available inbound and outbound slots. -pub struct ConnectionSlots { - /// A condvar for notifying when a slot become available. - signal: CondWait, - /// The number of occupied slots - slots: AtomicUsize, - /// The maximum number of slots. - max_slots: usize, -} - -impl ConnectionSlots { - /// Creates a new ConnectionSlots - pub fn new(max_slots: usize) -> Self { - Self { - signal: CondWait::new(), - slots: AtomicUsize::new(0), - max_slots, - } - } - - /// Returns the number of occupied slots - pub fn load(&self) -> usize { - self.slots.load(Ordering::SeqCst) - } - - /// Increases the occupied slots by one. - pub fn add(&self) { - self.slots.fetch_add(1, Ordering::SeqCst); - } - - /// Decreases the occupied slots by one and notifies the waiting signal - /// to start accepting/connecting new connections. - pub async fn remove(&self) { - self.slots.fetch_sub(1, Ordering::SeqCst); - if self.slots.load(Ordering::SeqCst) < self.max_slots { - self.signal.signal().await; - } - } - - /// Waits for a slot to become available. - pub async fn wait_for_slot(&self) { - if self.slots.load(Ordering::SeqCst) < self.max_slots { - return; - } - - // Wait for a signal - self.signal.wait().await; - self.signal.reset().await; - } -} diff --git a/karyons_p2p/src/peer/mod.rs b/karyons_p2p/src/peer/mod.rs deleted file mode 100644 index ee0fdc4..0000000 --- a/karyons_p2p/src/peer/mod.rs +++ /dev/null @@ -1,237 +0,0 @@ -mod peer_id; - -pub use peer_id::PeerID; - -use std::sync::Arc; - -use log::{error, trace}; -use smol::{ - channel::{self, Receiver, Sender}, - lock::RwLock, -}; - -use karyons_core::{ - async_utils::{select, Either, TaskGroup, TaskResult}, - event::{ArcEventSys, EventListener, EventSys}, - utils::{decode, encode}, - Executor, -}; - -use karyons_net::Endpoint; - -use crate::{ - io_codec::{CodecMsg, IOCodec}, - message::{NetMsgCmd, ProtocolMsg, ShutdownMsg}, - net::ConnDirection, - peer_pool::{ArcPeerPool, WeakPeerPool}, - protocol::{Protocol, ProtocolEvent, ProtocolID}, - Config, Error, Result, -}; - -pub type ArcPeer = Arc<Peer>; - -pub struct Peer { - /// Peer's ID - id: PeerID, - - /// A weak pointer to `PeerPool` - peer_pool: WeakPeerPool, - - /// Holds the IOCodec for the peer connection - io_codec: IOCodec, - - /// Remote endpoint for the peer - remote_endpoint: Endpoint, - - /// The direction of the connection, either `Inbound` or `Outbound` - conn_direction: ConnDirection, - - /// A list of protocol IDs - protocol_ids: RwLock<Vec<ProtocolID>>, - - /// `EventSys` responsible for sending events to the protocols. - protocol_events: ArcEventSys<ProtocolID>, - - /// This channel is used to send a stop signal to the read loop. - stop_chan: (Sender<Result<()>>, Receiver<Result<()>>), - - /// Managing spawned tasks. - task_group: TaskGroup, -} - -impl Peer { - /// Creates a new peer - pub fn new( - peer_pool: WeakPeerPool, - id: &PeerID, - io_codec: IOCodec, - remote_endpoint: Endpoint, - conn_direction: ConnDirection, - ) -> ArcPeer { - Arc::new(Peer { - id: id.clone(), - peer_pool, - io_codec, - protocol_ids: RwLock::new(Vec::new()), - remote_endpoint, - conn_direction, - protocol_events: EventSys::new(), - task_group: TaskGroup::new(), - stop_chan: channel::bounded(1), - }) - } - - /// Run the peer - pub async fn run(self: Arc<Self>, ex: Executor<'_>) -> Result<()> { - self.start_protocols(ex.clone()).await; - self.read_loop().await - } - - /// Send a message to the peer connection using the specified protocol. - pub async fn send<T: CodecMsg>(&self, protocol_id: &ProtocolID, msg: &T) -> Result<()> { - let payload = encode(msg)?; - - let proto_msg = ProtocolMsg { - protocol_id: protocol_id.to_string(), - payload: payload.to_vec(), - }; - - self.io_codec.write(NetMsgCmd::Protocol, &proto_msg).await?; - Ok(()) - } - - /// Broadcast a message to all connected peers using the specified protocol. - pub async fn broadcast<T: CodecMsg>(&self, protocol_id: &ProtocolID, msg: &T) { - self.peer_pool().broadcast(protocol_id, msg).await; - } - - /// Shuts down the peer - pub async fn shutdown(&self) { - trace!("peer {} start shutting down", self.id); - - // Send shutdown event to all protocols - for protocol_id in self.protocol_ids.read().await.iter() { - self.protocol_events - .emit_by_topic(protocol_id, &ProtocolEvent::Shutdown) - .await; - } - - // Send a stop signal to the read loop - // - // No need to handle the error here; a dropped channel and - // sending a stop signal have the same effect. - let _ = self.stop_chan.0.try_send(Ok(())); - - // No need to handle the error here - let _ = self - .io_codec - .write(NetMsgCmd::Shutdown, &ShutdownMsg(0)) - .await; - - // Force shutting down - self.task_group.cancel().await; - } - - /// Check if the connection is Inbound - #[inline] - pub fn is_inbound(&self) -> bool { - match self.conn_direction { - ConnDirection::Inbound => true, - ConnDirection::Outbound => false, - } - } - - /// Returns the direction of the connection, which can be either `Inbound` - /// or `Outbound`. - #[inline] - pub fn direction(&self) -> &ConnDirection { - &self.conn_direction - } - - /// Returns the remote endpoint for the peer - #[inline] - pub fn remote_endpoint(&self) -> &Endpoint { - &self.remote_endpoint - } - - /// Return the peer's ID - #[inline] - pub fn id(&self) -> &PeerID { - &self.id - } - - /// Returns the `Config` instance. - pub fn config(&self) -> Arc<Config> { - self.peer_pool().config.clone() - } - - /// Registers a listener for the given Protocol `P`. - pub async fn register_listener<P: Protocol>(&self) -> EventListener<ProtocolID, ProtocolEvent> { - self.protocol_events.register(&P::id()).await - } - - /// Start a read loop to handle incoming messages from the peer connection. - async fn read_loop(&self) -> Result<()> { - loop { - let fut = select(self.stop_chan.1.recv(), self.io_codec.read()).await; - let result = match fut { - Either::Left(stop_signal) => { - trace!("Peer {} received a stop signal", self.id); - return stop_signal?; - } - Either::Right(result) => result, - }; - - let msg = result?; - - match msg.header.command { - NetMsgCmd::Protocol => { - let msg: ProtocolMsg = decode(&msg.payload)?.0; - - if !self.protocol_ids.read().await.contains(&msg.protocol_id) { - return Err(Error::UnsupportedProtocol(msg.protocol_id)); - } - - let proto_id = &msg.protocol_id; - let msg = ProtocolEvent::Message(msg.payload); - self.protocol_events.emit_by_topic(proto_id, &msg).await; - } - NetMsgCmd::Shutdown => { - return Err(Error::PeerShutdown); - } - command => return Err(Error::InvalidMsg(format!("Unexpected msg {:?}", command))), - } - } - } - - /// Start running the protocols for this peer connection. - async fn start_protocols(self: &Arc<Self>, ex: Executor<'_>) { - for (protocol_id, constructor) in self.peer_pool().protocols.read().await.iter() { - trace!("peer {} start protocol {protocol_id}", self.id); - let protocol = constructor(self.clone()); - - self.protocol_ids.write().await.push(protocol_id.clone()); - - let selfc = self.clone(); - let exc = ex.clone(); - let proto_idc = protocol_id.clone(); - - let on_failure = |result: TaskResult<Result<()>>| async move { - if let TaskResult::Completed(res) = result { - if res.is_err() { - error!("protocol {} stopped", proto_idc); - } - // Send a stop signal to read loop - let _ = selfc.stop_chan.0.try_send(res); - } - }; - - self.task_group - .spawn(ex.clone(), protocol.start(exc), on_failure); - } - } - - fn peer_pool(&self) -> ArcPeerPool { - self.peer_pool.upgrade().unwrap() - } -} diff --git a/karyons_p2p/src/peer/peer_id.rs b/karyons_p2p/src/peer/peer_id.rs deleted file mode 100644 index c8aec7d..0000000 --- a/karyons_p2p/src/peer/peer_id.rs +++ /dev/null @@ -1,41 +0,0 @@ -use bincode::{Decode, Encode}; -use rand::{rngs::OsRng, RngCore}; -use sha2::{Digest, Sha256}; - -/// Represents a unique identifier for a peer. -#[derive(Clone, Debug, Eq, PartialEq, Hash, Decode, Encode)] -pub struct PeerID(pub [u8; 32]); - -impl std::fmt::Display for PeerID { - fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result { - let id = self.0[0..8] - .iter() - .map(|b| format!("{:x}", b)) - .collect::<Vec<String>>() - .join(""); - - write!(f, "{}", id) - } -} - -impl PeerID { - /// Creates a new PeerID. - pub fn new(src: &[u8]) -> Self { - let mut hasher = Sha256::new(); - hasher.update(src); - Self(hasher.finalize().into()) - } - - /// Generates a random PeerID. - pub fn random() -> Self { - let mut id: [u8; 32] = [0; 32]; - OsRng.fill_bytes(&mut id); - Self(id) - } -} - -impl From<[u8; 32]> for PeerID { - fn from(b: [u8; 32]) -> Self { - PeerID(b) - } -} diff --git a/karyons_p2p/src/peer_pool.rs b/karyons_p2p/src/peer_pool.rs deleted file mode 100644 index eac4d3d..0000000 --- a/karyons_p2p/src/peer_pool.rs +++ /dev/null @@ -1,337 +0,0 @@ -use std::{ - collections::HashMap, - sync::{Arc, Weak}, - time::Duration, -}; - -use log::{error, info, trace, warn}; -use smol::{ - channel::Sender, - lock::{Mutex, RwLock}, -}; - -use karyons_core::{ - async_utils::{TaskGroup, TaskResult}, - utils::decode, - Executor, -}; - -use karyons_net::Conn; - -use crate::{ - config::Config, - io_codec::{CodecMsg, IOCodec}, - message::{get_msg_payload, NetMsg, NetMsgCmd, VerAckMsg, VerMsg}, - monitor::{Monitor, PeerPoolEvent}, - net::ConnDirection, - net::ConnQueue, - peer::{ArcPeer, Peer, PeerID}, - protocol::{Protocol, ProtocolConstructor, ProtocolID}, - protocols::PingProtocol, - utils::{version_match, Version, VersionInt}, - Error, Result, -}; - -pub type ArcPeerPool = Arc<PeerPool>; -pub type WeakPeerPool = Weak<PeerPool>; - -pub struct PeerPool { - /// Peer's ID - pub id: PeerID, - - /// Connection queue - conn_queue: Arc<ConnQueue>, - - /// Holds the running peers. - peers: Mutex<HashMap<PeerID, ArcPeer>>, - - /// Hashmap contains protocol constructors. - pub(crate) protocols: RwLock<HashMap<ProtocolID, Box<ProtocolConstructor>>>, - - /// Hashmap contains protocol IDs and their versions. - protocol_versions: Arc<RwLock<HashMap<ProtocolID, Version>>>, - - /// Managing spawned tasks. - task_group: TaskGroup, - - /// The Configuration for the P2P network. - pub config: Arc<Config>, - - /// Responsible for network and system monitoring. - monitor: Arc<Monitor>, -} - -impl PeerPool { - /// Creates a new PeerPool - pub fn new( - id: &PeerID, - conn_queue: Arc<ConnQueue>, - config: Arc<Config>, - monitor: Arc<Monitor>, - ) -> Arc<Self> { - let protocols = RwLock::new(HashMap::new()); - let protocol_versions = Arc::new(RwLock::new(HashMap::new())); - - Arc::new(Self { - id: id.clone(), - conn_queue, - peers: Mutex::new(HashMap::new()), - protocols, - protocol_versions, - task_group: TaskGroup::new(), - monitor, - config, - }) - } - - /// Start - pub async fn start(self: &Arc<Self>, ex: Executor<'_>) -> Result<()> { - self.setup_protocols().await?; - let selfc = self.clone(); - self.task_group - .spawn(ex.clone(), selfc.listen_loop(ex.clone()), |_| async {}); - Ok(()) - } - - /// Listens to a new connection from the connection queue - pub async fn listen_loop(self: Arc<Self>, ex: Executor<'_>) { - loop { - let new_conn = self.conn_queue.next().await; - let disconnect_signal = new_conn.disconnect_signal; - - let result = self - .new_peer( - new_conn.conn, - &new_conn.direction, - disconnect_signal.clone(), - ex.clone(), - ) - .await; - - if result.is_err() { - let _ = disconnect_signal.send(()).await; - } - } - } - - /// Shuts down - pub async fn shutdown(&self) { - for (_, peer) in self.peers.lock().await.iter() { - peer.shutdown().await; - } - - self.task_group.cancel().await; - } - - /// Attach a custom protocol to the network - pub async fn attach_protocol<P: Protocol>(&self, c: Box<ProtocolConstructor>) -> Result<()> { - let protocol_versions = &mut self.protocol_versions.write().await; - let protocols = &mut self.protocols.write().await; - - protocol_versions.insert(P::id(), P::version()?); - protocols.insert(P::id(), Box::new(c) as Box<ProtocolConstructor>); - Ok(()) - } - - /// Returns the number of currently connected peers. - pub async fn peers_len(&self) -> usize { - self.peers.lock().await.len() - } - - /// Broadcast a message to all connected peers using the specified protocol. - pub async fn broadcast<T: CodecMsg>(&self, proto_id: &ProtocolID, msg: &T) { - for (pid, peer) in self.peers.lock().await.iter() { - if let Err(err) = peer.send(proto_id, msg).await { - error!("failed to send msg to {pid}: {err}"); - continue; - } - } - } - - /// Add a new peer to the peer list. - pub async fn new_peer( - self: &Arc<Self>, - conn: Conn, - conn_direction: &ConnDirection, - disconnect_signal: Sender<()>, - ex: Executor<'_>, - ) -> Result<PeerID> { - let endpoint = conn.peer_endpoint()?; - let io_codec = IOCodec::new(conn); - - // Do a handshake with a connection before creating a new peer. - let pid = self.do_handshake(&io_codec, conn_direction).await?; - - // TODO: Consider restricting the subnet for inbound connections - if self.contains_peer(&pid).await { - return Err(Error::PeerAlreadyConnected); - } - - // Create a new peer - let peer = Peer::new( - Arc::downgrade(self), - &pid, - io_codec, - endpoint.clone(), - conn_direction.clone(), - ); - - // Insert the new peer - self.peers.lock().await.insert(pid.clone(), peer.clone()); - - let selfc = self.clone(); - let pid_c = pid.clone(); - let on_disconnect = |result| async move { - if let TaskResult::Completed(_) = result { - if let Err(err) = selfc.remove_peer(&pid_c).await { - error!("Failed to remove peer {pid_c}: {err}"); - } - let _ = disconnect_signal.send(()).await; - } - }; - - self.task_group - .spawn(ex.clone(), peer.run(ex.clone()), on_disconnect); - - info!("Add new peer {pid}, direction: {conn_direction}, endpoint: {endpoint}"); - - self.monitor - .notify(&PeerPoolEvent::NewPeer(pid.clone()).into()) - .await; - Ok(pid) - } - - /// Checks if the peer list contains a peer with the given peer id - pub async fn contains_peer(&self, pid: &PeerID) -> bool { - self.peers.lock().await.contains_key(pid) - } - - /// Shuts down the peer and remove it from the peer list. - async fn remove_peer(&self, pid: &PeerID) -> Result<()> { - let mut peers = self.peers.lock().await; - let result = peers.remove(pid); - - drop(peers); - - let peer = match result { - Some(p) => p, - None => return Ok(()), - }; - - peer.shutdown().await; - - self.monitor - .notify(&PeerPoolEvent::RemovePeer(pid.clone()).into()) - .await; - - let endpoint = peer.remote_endpoint(); - let direction = peer.direction(); - - warn!("Peer {pid} removed, direction: {direction}, endpoint: {endpoint}",); - Ok(()) - } - - /// Attach the core protocols. - async fn setup_protocols(&self) -> Result<()> { - self.attach_protocol::<PingProtocol>(Box::new(PingProtocol::new)) - .await - } - - /// Initiate a handshake with a connection. - async fn do_handshake( - &self, - io_codec: &IOCodec, - conn_direction: &ConnDirection, - ) -> Result<PeerID> { - match conn_direction { - ConnDirection::Inbound => { - let pid = self.wait_vermsg(io_codec).await?; - self.send_verack(io_codec).await?; - Ok(pid) - } - ConnDirection::Outbound => { - self.send_vermsg(io_codec).await?; - self.wait_verack(io_codec).await - } - } - } - - /// Send a Version message - async fn send_vermsg(&self, io_codec: &IOCodec) -> Result<()> { - let pids = self.protocol_versions.read().await; - let protocols = pids.iter().map(|p| (p.0.clone(), p.1.v.clone())).collect(); - drop(pids); - - let vermsg = VerMsg { - peer_id: self.id.clone(), - protocols, - version: self.config.version.v.clone(), - }; - - trace!("Send VerMsg"); - io_codec.write(NetMsgCmd::Version, &vermsg).await?; - Ok(()) - } - - /// Wait for a Version message - /// - /// Returns the peer's ID upon successfully receiving the Version message. - async fn wait_vermsg(&self, io_codec: &IOCodec) -> Result<PeerID> { - let timeout = Duration::from_secs(self.config.handshake_timeout); - let msg: NetMsg = io_codec.read_timeout(timeout).await?; - - let payload = get_msg_payload!(Version, msg); - let (vermsg, _) = decode::<VerMsg>(&payload)?; - - if !version_match(&self.config.version.req, &vermsg.version) { - return Err(Error::IncompatibleVersion("system: {}".into())); - } - - self.protocols_match(&vermsg.protocols).await?; - - trace!("Received VerMsg from: {}", vermsg.peer_id); - Ok(vermsg.peer_id) - } - - /// Send a Verack message - async fn send_verack(&self, io_codec: &IOCodec) -> Result<()> { - let verack = VerAckMsg(self.id.clone()); - - trace!("Send VerAckMsg"); - io_codec.write(NetMsgCmd::Verack, &verack).await?; - Ok(()) - } - - /// Wait for a Verack message - /// - /// Returns the peer's ID upon successfully receiving the Verack message. - async fn wait_verack(&self, io_codec: &IOCodec) -> Result<PeerID> { - let timeout = Duration::from_secs(self.config.handshake_timeout); - let msg: NetMsg = io_codec.read_timeout(timeout).await?; - - let payload = get_msg_payload!(Verack, msg); - let (verack, _) = decode::<VerAckMsg>(&payload)?; - - trace!("Received VerAckMsg from: {}", verack.0); - Ok(verack.0) - } - - /// Check if the new connection has compatible protocols. - async fn protocols_match(&self, protocols: &HashMap<ProtocolID, VersionInt>) -> Result<()> { - for (n, pv) in protocols.iter() { - let pids = self.protocol_versions.read().await; - - match pids.get(n) { - Some(v) => { - if !version_match(&v.req, pv) { - return Err(Error::IncompatibleVersion(format!("{n} protocol: {pv}"))); - } - } - None => { - return Err(Error::UnsupportedProtocol(n.to_string())); - } - } - } - Ok(()) - } -} diff --git a/karyons_p2p/src/protocol.rs b/karyons_p2p/src/protocol.rs deleted file mode 100644 index 515efc6..0000000 --- a/karyons_p2p/src/protocol.rs +++ /dev/null @@ -1,113 +0,0 @@ -use std::sync::Arc; - -use async_trait::async_trait; - -use karyons_core::{event::EventValue, Executor}; - -use crate::{peer::ArcPeer, utils::Version, Result}; - -pub type ArcProtocol = Arc<dyn Protocol>; - -pub type ProtocolConstructor = dyn Fn(ArcPeer) -> Arc<dyn Protocol> + Send + Sync; - -pub type ProtocolID = String; - -/// Protocol event -#[derive(Debug, Clone)] -pub enum ProtocolEvent { - /// Message event, contains a vector of bytes. - Message(Vec<u8>), - /// Shutdown event signals the protocol to gracefully shut down. - Shutdown, -} - -impl EventValue for ProtocolEvent { - fn id() -> &'static str { - "ProtocolEvent" - } -} - -/// The Protocol trait defines the interface for core protocols -/// and custom protocols. -/// -/// # Example -/// ``` -/// use std::sync::Arc; -/// -/// use async_trait::async_trait; -/// use smol::Executor; -/// -/// use karyons_p2p::{ -/// protocol::{ArcProtocol, Protocol, ProtocolID, ProtocolEvent}, -/// Backend, PeerID, Config, Version, P2pError, ArcPeer}; -/// -/// pub struct NewProtocol { -/// peer: ArcPeer, -/// } -/// -/// impl NewProtocol { -/// fn new(peer: ArcPeer) -> ArcProtocol { -/// Arc::new(Self { -/// peer, -/// }) -/// } -/// } -/// -/// #[async_trait] -/// impl Protocol for NewProtocol { -/// async fn start(self: Arc<Self>, ex: Arc<Executor<'_>>) -> Result<(), P2pError> { -/// let listener = self.peer.register_listener::<Self>().await; -/// loop { -/// let event = listener.recv().await.unwrap(); -/// -/// match event { -/// ProtocolEvent::Message(msg) => { -/// println!("{:?}", msg); -/// } -/// ProtocolEvent::Shutdown => { -/// break; -/// } -/// } -/// } -/// -/// listener.cancel().await; -/// Ok(()) -/// } -/// -/// fn version() -> Result<Version, P2pError> { -/// "0.2.0, >0.1.0".parse() -/// } -/// -/// fn id() -> ProtocolID { -/// "NEWPROTOCOLID".into() -/// } -/// } -/// -/// async { -/// let peer_id = PeerID::random(); -/// let config = Config::default(); -/// -/// // Create a new Backend -/// let backend = Backend::new(peer_id, config); -/// -/// // Attach the NewProtocol -/// let c = move |peer| NewProtocol::new(peer); -/// backend.attach_protocol::<NewProtocol>(c).await.unwrap(); -/// }; -/// -/// ``` -#[async_trait] -pub trait Protocol: Send + Sync { - /// Start the protocol - async fn start(self: Arc<Self>, ex: Executor<'_>) -> Result<()>; - - /// Returns the version of the protocol. - fn version() -> Result<Version> - where - Self: Sized; - - /// Returns the unique ProtocolID associated with the protocol. - fn id() -> ProtocolID - where - Self: Sized; -} diff --git a/karyons_p2p/src/protocols/mod.rs b/karyons_p2p/src/protocols/mod.rs deleted file mode 100644 index 4a8f6b9..0000000 --- a/karyons_p2p/src/protocols/mod.rs +++ /dev/null @@ -1,3 +0,0 @@ -mod ping; - -pub use ping::PingProtocol; diff --git a/karyons_p2p/src/protocols/ping.rs b/karyons_p2p/src/protocols/ping.rs deleted file mode 100644 index b337494..0000000 --- a/karyons_p2p/src/protocols/ping.rs +++ /dev/null @@ -1,173 +0,0 @@ -use std::{sync::Arc, time::Duration}; - -use async_trait::async_trait; -use bincode::{Decode, Encode}; -use log::trace; -use rand::{rngs::OsRng, RngCore}; -use smol::{ - channel, - channel::{Receiver, Sender}, - stream::StreamExt, - Timer, -}; - -use karyons_core::{ - async_utils::{select, timeout, Either, TaskGroup, TaskResult}, - event::EventListener, - utils::decode, - Executor, -}; - -use karyons_net::NetError; - -use crate::{ - peer::ArcPeer, - protocol::{ArcProtocol, Protocol, ProtocolEvent, ProtocolID}, - utils::Version, - Result, -}; - -const MAX_FAILUERS: u32 = 3; - -#[derive(Clone, Debug, Encode, Decode)] -enum PingProtocolMsg { - Ping([u8; 32]), - Pong([u8; 32]), -} - -pub struct PingProtocol { - peer: ArcPeer, - ping_interval: u64, - ping_timeout: u64, - task_group: TaskGroup, -} - -impl PingProtocol { - #[allow(clippy::new_ret_no_self)] - pub fn new(peer: ArcPeer) -> ArcProtocol { - let ping_interval = peer.config().ping_interval; - let ping_timeout = peer.config().ping_timeout; - Arc::new(Self { - peer, - ping_interval, - ping_timeout, - task_group: TaskGroup::new(), - }) - } - - async fn recv_loop( - &self, - listener: &EventListener<ProtocolID, ProtocolEvent>, - pong_chan: Sender<[u8; 32]>, - ) -> Result<()> { - loop { - let event = listener.recv().await?; - let msg_payload = match event.clone() { - ProtocolEvent::Message(m) => m, - ProtocolEvent::Shutdown => { - break; - } - }; - - let (msg, _) = decode::<PingProtocolMsg>(&msg_payload)?; - - match msg { - PingProtocolMsg::Ping(nonce) => { - trace!("Received Ping message {:?}", nonce); - self.peer - .send(&Self::id(), &PingProtocolMsg::Pong(nonce)) - .await?; - trace!("Send back Pong message {:?}", nonce); - } - PingProtocolMsg::Pong(nonce) => { - pong_chan.send(nonce).await?; - } - } - } - Ok(()) - } - - async fn ping_loop(self: Arc<Self>, chan: Receiver<[u8; 32]>) -> Result<()> { - let mut timer = Timer::interval(Duration::from_secs(self.ping_interval)); - let rng = &mut OsRng; - let mut retry = 0; - - while retry < MAX_FAILUERS { - timer.next().await; - - let mut ping_nonce: [u8; 32] = [0; 32]; - rng.fill_bytes(&mut ping_nonce); - - trace!("Send Ping message {:?}", ping_nonce); - self.peer - .send(&Self::id(), &PingProtocolMsg::Ping(ping_nonce)) - .await?; - - let d = Duration::from_secs(self.ping_timeout); - - // Wait for Pong message - let pong_msg = match timeout(d, chan.recv()).await { - Ok(m) => m?, - Err(_) => { - retry += 1; - continue; - } - }; - - trace!("Received Pong message {:?}", pong_msg); - - if pong_msg != ping_nonce { - retry += 1; - continue; - } - } - - Err(NetError::Timeout.into()) - } -} - -#[async_trait] -impl Protocol for PingProtocol { - async fn start(self: Arc<Self>, ex: Executor<'_>) -> Result<()> { - trace!("Start Ping protocol"); - let (pong_chan, pong_chan_recv) = channel::bounded(1); - let (stop_signal_s, stop_signal) = channel::bounded::<Result<()>>(1); - - let selfc = self.clone(); - self.task_group.spawn( - ex.clone(), - selfc.clone().ping_loop(pong_chan_recv.clone()), - |res| async move { - if let TaskResult::Completed(result) = res { - let _ = stop_signal_s.send(result).await; - } - }, - ); - - let listener = self.peer.register_listener::<Self>().await; - - let result = select(self.recv_loop(&listener, pong_chan), stop_signal.recv()).await; - listener.cancel().await; - self.task_group.cancel().await; - - match result { - Either::Left(res) => { - trace!("Receive loop stopped {:?}", res); - res - } - Either::Right(res) => { - let res = res?; - trace!("Ping loop stopped {:?}", res); - res - } - } - } - - fn version() -> Result<Version> { - "0.1.0".parse() - } - - fn id() -> ProtocolID { - "PING".into() - } -} diff --git a/karyons_p2p/src/routing_table/bucket.rs b/karyons_p2p/src/routing_table/bucket.rs deleted file mode 100644 index 13edd24..0000000 --- a/karyons_p2p/src/routing_table/bucket.rs +++ /dev/null @@ -1,123 +0,0 @@ -use super::{Entry, Key}; - -use rand::{rngs::OsRng, seq::SliceRandom}; - -/// BITFLAGS represent the status of an Entry within a bucket. -pub type EntryStatusFlag = u16; - -/// The entry is connected. -pub const CONNECTED_ENTRY: EntryStatusFlag = 0b00001; - -/// The entry is disconnected. This will increase the failure counter. -pub const DISCONNECTED_ENTRY: EntryStatusFlag = 0b00010; - -/// The entry is ready to reconnect, meaning it has either been added and -/// has no connection attempts, or it has been refreshed. -pub const PENDING_ENTRY: EntryStatusFlag = 0b00100; - -/// The entry is unreachable. This will increase the failure counter. -pub const UNREACHABLE_ENTRY: EntryStatusFlag = 0b01000; - -/// The entry is unstable. This will increase the failure counter. -pub const UNSTABLE_ENTRY: EntryStatusFlag = 0b10000; - -#[allow(dead_code)] -pub const ALL_ENTRY: EntryStatusFlag = 0b11111; - -/// A BucketEntry represents a peer in the routing table. -#[derive(Clone, Debug)] -pub struct BucketEntry { - pub status: EntryStatusFlag, - pub entry: Entry, - pub failures: u32, - pub last_seen: i64, -} - -impl BucketEntry { - pub fn is_connected(&self) -> bool { - self.status ^ CONNECTED_ENTRY == 0 - } - - pub fn is_unreachable(&self) -> bool { - self.status ^ UNREACHABLE_ENTRY == 0 - } - - pub fn is_unstable(&self) -> bool { - self.status ^ UNSTABLE_ENTRY == 0 - } -} - -/// The number of entries that can be stored within a single bucket. -pub const BUCKET_SIZE: usize = 20; - -/// A Bucket represents a group of entries in the routing table. -#[derive(Debug, Clone)] -pub struct Bucket { - entries: Vec<BucketEntry>, -} - -impl Bucket { - /// Creates a new empty Bucket - pub fn new() -> Self { - Self { - entries: Vec::with_capacity(BUCKET_SIZE), - } - } - - /// Add an entry to the bucket. - pub fn add(&mut self, entry: &Entry) { - self.entries.push(BucketEntry { - status: PENDING_ENTRY, - entry: entry.clone(), - failures: 0, - last_seen: chrono::Utc::now().timestamp(), - }) - } - - /// Get the number of entries in the bucket. - pub fn len(&self) -> usize { - self.entries.len() - } - - /// Returns an iterator over the entries in the bucket. - pub fn iter(&self) -> impl Iterator<Item = &BucketEntry> { - self.entries.iter() - } - - /// Remove an entry. - pub fn remove(&mut self, key: &Key) { - let position = self.entries.iter().position(|e| &e.entry.key == key); - if let Some(i) = position { - self.entries.remove(i); - } - } - - /// Returns an iterator of entries in random order. - pub fn random_iter(&self, amount: usize) -> impl Iterator<Item = &BucketEntry> { - self.entries.choose_multiple(&mut OsRng, amount) - } - - /// Updates the status of an entry in the bucket identified by the given key. - /// - /// If the key is not found in the bucket, no action is taken. - /// - /// This will also update the last_seen field and increase the failures - /// counter for the bucket entry according to the new status. - pub fn update_entry(&mut self, key: &Key, entry_flag: EntryStatusFlag) { - if let Some(e) = self.entries.iter_mut().find(|e| &e.entry.key == key) { - e.status = entry_flag; - if e.is_unreachable() || e.is_unstable() { - e.failures += 1; - } - - if !e.is_unreachable() { - e.last_seen = chrono::Utc::now().timestamp(); - } - } - } - - /// Check if the bucket contains the given key. - pub fn contains_key(&self, key: &Key) -> bool { - self.entries.iter().any(|e| &e.entry.key == key) - } -} diff --git a/karyons_p2p/src/routing_table/entry.rs b/karyons_p2p/src/routing_table/entry.rs deleted file mode 100644 index b3f219f..0000000 --- a/karyons_p2p/src/routing_table/entry.rs +++ /dev/null @@ -1,41 +0,0 @@ -use bincode::{Decode, Encode}; - -use karyons_net::{Addr, Port}; - -/// Specifies the size of the key, in bytes. -pub const KEY_SIZE: usize = 32; - -/// An Entry represents a peer in the routing table. -#[derive(Encode, Decode, Clone, Debug)] -pub struct Entry { - /// The unique key identifying the peer. - pub key: Key, - /// The IP address of the peer. - pub addr: Addr, - /// TCP port - pub port: Port, - /// UDP/TCP port - pub discovery_port: Port, -} - -impl PartialEq for Entry { - fn eq(&self, other: &Self) -> bool { - // XXX this should also compare both addresses (the self.addr == other.addr) - self.key == other.key - } -} - -/// The unique key identifying the peer. -pub type Key = [u8; KEY_SIZE]; - -/// Calculates the XOR distance between two provided keys. -/// -/// The XOR distance is a metric used in Kademlia to measure the closeness -/// of keys. -pub fn xor_distance(key: &Key, other: &Key) -> Key { - let mut res = [0; 32]; - for (i, (k, o)) in key.iter().zip(other.iter()).enumerate() { - res[i] = k ^ o; - } - res -} diff --git a/karyons_p2p/src/routing_table/mod.rs b/karyons_p2p/src/routing_table/mod.rs deleted file mode 100644 index abf9a08..0000000 --- a/karyons_p2p/src/routing_table/mod.rs +++ /dev/null @@ -1,461 +0,0 @@ -mod bucket; -mod entry; -pub use bucket::{ - Bucket, BucketEntry, EntryStatusFlag, CONNECTED_ENTRY, DISCONNECTED_ENTRY, PENDING_ENTRY, - UNREACHABLE_ENTRY, UNSTABLE_ENTRY, -}; -pub use entry::{xor_distance, Entry, Key}; - -use rand::{rngs::OsRng, seq::SliceRandom}; - -use crate::utils::subnet_match; - -use bucket::BUCKET_SIZE; -use entry::KEY_SIZE; - -/// The total number of buckets in the routing table. -const TABLE_SIZE: usize = 32; - -/// The distance limit for the closest buckets. -const DISTANCE_LIMIT: usize = 32; - -/// The maximum number of matched subnets allowed within a single bucket. -const MAX_MATCHED_SUBNET_IN_BUCKET: usize = 1; - -/// The maximum number of matched subnets across the entire routing table. -const MAX_MATCHED_SUBNET_IN_TABLE: usize = 6; - -/// Represents the possible result when adding a new entry. -#[derive(Debug)] -pub enum AddEntryResult { - /// The entry is added. - Added, - /// The entry is already exists. - Exists, - /// The entry is ignored. - Ignored, - /// The entry is restricted and not allowed. - Restricted, -} - -/// This is a modified version of the Kademlia Distributed Hash Table (DHT). -/// https://en.wikipedia.org/wiki/Kademlia -#[derive(Debug)] -pub struct RoutingTable { - key: Key, - buckets: Vec<Bucket>, -} - -impl RoutingTable { - /// Creates a new RoutingTable - pub fn new(key: Key) -> Self { - let buckets: Vec<Bucket> = (0..TABLE_SIZE).map(|_| Bucket::new()).collect(); - Self { key, buckets } - } - - /// Adds a new entry to the table and returns a result indicating success, - /// failure, or restrictions. - pub fn add_entry(&mut self, entry: Entry) -> AddEntryResult { - // Determine the index of the bucket where the entry should be placed. - let bucket_idx = match self.bucket_index(&entry.key) { - Some(i) => i, - None => return AddEntryResult::Ignored, - }; - - let bucket = &self.buckets[bucket_idx]; - - // Check if the entry already exists in the bucket. - if bucket.contains_key(&entry.key) { - return AddEntryResult::Exists; - } - - // Check if the entry is restricted. - if self.subnet_restricted(bucket_idx, &entry) { - return AddEntryResult::Restricted; - } - - let bucket = &mut self.buckets[bucket_idx]; - - // If the bucket has free space, add the entry and return success. - if bucket.len() < BUCKET_SIZE { - bucket.add(&entry); - return AddEntryResult::Added; - } - - // If the bucket is full, the entry is ignored. - AddEntryResult::Ignored - } - - /// Check if the table contains the given key. - pub fn contains_key(&self, key: &Key) -> bool { - // Determine the bucket index for the given key. - let bucket_idx = match self.bucket_index(key) { - Some(bi) => bi, - None => return false, - }; - - let bucket = &self.buckets[bucket_idx]; - bucket.contains_key(key) - } - - /// Updates the status of an entry in the routing table identified - /// by the given key. - /// - /// If the key is not found, no action is taken. - pub fn update_entry(&mut self, key: &Key, entry_flag: EntryStatusFlag) { - // Determine the bucket index for the given key. - let bucket_idx = match self.bucket_index(key) { - Some(bi) => bi, - None => return, - }; - - let bucket = &mut self.buckets[bucket_idx]; - bucket.update_entry(key, entry_flag); - } - - /// Returns a list of bucket indexes that are closest to the given target key. - pub fn bucket_indexes(&self, target_key: &Key) -> Vec<usize> { - let mut indexes = vec![]; - - // Determine the primary bucket index for the target key. - let bucket_idx = self.bucket_index(target_key).unwrap_or(0); - - indexes.push(bucket_idx); - - // Add additional bucket indexes within a certain distance limit. - for i in 1..DISTANCE_LIMIT { - if bucket_idx >= i && bucket_idx - i >= 1 { - indexes.push(bucket_idx - i); - } - - if bucket_idx + i < (TABLE_SIZE - 1) { - indexes.push(bucket_idx + i); - } - } - - indexes - } - - /// Returns a list of the closest entries to the given target key, limited by max_entries. - pub fn closest_entries(&self, target_key: &Key, max_entries: usize) -> Vec<Entry> { - let mut entries: Vec<Entry> = vec![]; - - // Collect entries - 'outer: for idx in self.bucket_indexes(target_key) { - let bucket = &self.buckets[idx]; - for bucket_entry in bucket.iter() { - if bucket_entry.is_unreachable() || bucket_entry.is_unstable() { - continue; - } - - entries.push(bucket_entry.entry.clone()); - if entries.len() == max_entries { - break 'outer; - } - } - } - - // Sort the entries by their distance to the target key. - entries.sort_by(|a, b| { - xor_distance(target_key, &a.key).cmp(&xor_distance(target_key, &b.key)) - }); - - entries - } - - /// Removes an entry with the given key from the routing table, if it exists. - pub fn remove_entry(&mut self, key: &Key) { - // Determine the bucket index for the given key. - let bucket_idx = match self.bucket_index(key) { - Some(bi) => bi, - None => return, - }; - - let bucket = &mut self.buckets[bucket_idx]; - bucket.remove(key); - } - - /// Returns an iterator of entries. - pub fn iter(&self) -> impl Iterator<Item = &Bucket> { - self.buckets.iter() - } - - /// Returns a random entry from the routing table. - pub fn random_entry(&self, entry_flag: EntryStatusFlag) -> Option<&Entry> { - for bucket in self.buckets.choose_multiple(&mut OsRng, self.buckets.len()) { - for entry in bucket.random_iter(bucket.len()) { - if entry.status & entry_flag == 0 { - continue; - } - return Some(&entry.entry); - } - } - - None - } - - // Returns the bucket index for a given key in the table. - fn bucket_index(&self, key: &Key) -> Option<usize> { - // Calculate the XOR distance between the self key and the provided key. - let distance = xor_distance(&self.key, key); - - for (i, b) in distance.iter().enumerate() { - if *b != 0 { - let lz = i * 8 + b.leading_zeros() as usize; - let bits = KEY_SIZE * 8 - 1; - let idx = (bits - lz) / 8; - return Some(idx); - } - } - None - } - - /// This function iterate through the routing table and counts how many - /// entries in the same subnet as the given Entry are already present. - /// - /// If the number of matching entries in the same bucket exceeds a - /// threshold (MAX_MATCHED_SUBNET_IN_BUCKET), or if the total count of - /// matching entries in the entire table exceeds a threshold - /// (MAX_MATCHED_SUBNET_IN_TABLE), the addition of the Entry - /// is considered restricted and returns true. - fn subnet_restricted(&self, idx: usize, entry: &Entry) -> bool { - let mut bucket_count = 0; - let mut table_count = 0; - - // Iterate through the routing table's buckets and entries to check - // for subnet matches. - for (i, bucket) in self.buckets.iter().enumerate() { - for e in bucket.iter() { - // If there is a subnet match, update the counts. - let matched = subnet_match(&e.entry.addr, &entry.addr); - if matched { - if i == idx { - bucket_count += 1; - } - table_count += 1; - } - - // If the number of matched entries in the same bucket exceeds - // the limit, return true - if bucket_count >= MAX_MATCHED_SUBNET_IN_BUCKET { - return true; - } - } - - // If the total matched entries in the table exceed the limit, - // return true. - if table_count >= MAX_MATCHED_SUBNET_IN_TABLE { - return true; - } - } - - // If no subnet restrictions are encountered, return false. - false - } -} - -#[cfg(test)] -mod tests { - use super::bucket::ALL_ENTRY; - use super::*; - - use karyons_net::Addr; - - struct Setup { - local_key: Key, - keys: Vec<Key>, - } - - fn new_entry(key: &Key, addr: &Addr, port: u16, discovery_port: u16) -> Entry { - Entry { - key: key.clone(), - addr: addr.clone(), - port, - discovery_port, - } - } - - impl Setup { - fn new() -> Self { - let keys = vec![ - [ - 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, - 0, 0, 0, 0, 0, 1, - ], - [ - 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, - 1, 1, 0, 1, 1, 2, - ], - [ - 0, 0, 0, 0, 0, 20, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, - 0, 0, 0, 0, 0, 3, - ], - [ - 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 30, 1, 18, 0, 0, 0, - 0, 0, 0, 0, 0, 4, - ], - [ - 223, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, - 0, 0, 0, 0, 0, 5, - ], - [ - 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 50, 1, 18, 0, 0, 0, - 0, 0, 0, 0, 0, 6, - ], - [ - 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 10, 50, 1, 18, 0, 0, - 0, 0, 0, 0, 0, 0, 7, - ], - [ - 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 10, 10, 50, 1, 18, 0, 0, - 0, 0, 0, 0, 0, 0, 8, - ], - [ - 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 10, 10, 10, 50, 1, 18, 0, 0, - 0, 0, 0, 0, 0, 0, 9, - ], - ]; - - Self { - local_key: [ - 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, - 0, 0, 0, 0, 0, 0, - ], - keys, - } - } - - fn entries(&self) -> Vec<Entry> { - let mut entries = vec![]; - for (i, key) in self.keys.iter().enumerate() { - entries.push(new_entry( - key, - &Addr::Ip(format!("127.0.{i}.1").parse().unwrap()), - 3000, - 3010, - )); - } - entries - } - - fn table(&self) -> RoutingTable { - let mut table = RoutingTable::new(self.local_key.clone()); - - for entry in self.entries() { - let res = table.add_entry(entry); - assert!(matches!(res, AddEntryResult::Added)); - } - - table - } - } - - #[test] - fn test_bucket_index() { - let setup = Setup::new(); - let table = setup.table(); - - assert_eq!(table.bucket_index(&setup.local_key), None); - assert_eq!(table.bucket_index(&setup.keys[0]), Some(0)); - assert_eq!(table.bucket_index(&setup.keys[1]), Some(5)); - assert_eq!(table.bucket_index(&setup.keys[2]), Some(26)); - assert_eq!(table.bucket_index(&setup.keys[3]), Some(11)); - assert_eq!(table.bucket_index(&setup.keys[4]), Some(31)); - assert_eq!(table.bucket_index(&setup.keys[5]), Some(11)); - assert_eq!(table.bucket_index(&setup.keys[6]), Some(12)); - assert_eq!(table.bucket_index(&setup.keys[7]), Some(13)); - assert_eq!(table.bucket_index(&setup.keys[8]), Some(14)); - } - - #[test] - fn test_closest_entries() { - let setup = Setup::new(); - let table = setup.table(); - let entries = setup.entries(); - - assert_eq!( - table.closest_entries(&setup.keys[5], 8), - vec![ - entries[5].clone(), - entries[3].clone(), - entries[1].clone(), - entries[6].clone(), - entries[7].clone(), - entries[8].clone(), - entries[2].clone(), - ] - ); - - assert_eq!( - table.closest_entries(&setup.keys[4], 2), - vec![entries[4].clone(), entries[2].clone()] - ); - } - - #[test] - fn test_random_entry() { - let setup = Setup::new(); - let mut table = setup.table(); - let entries = setup.entries(); - - let entry = table.random_entry(ALL_ENTRY); - assert!(matches!(entry, Some(&_))); - - let entry = table.random_entry(CONNECTED_ENTRY); - assert!(matches!(entry, None)); - - for entry in entries { - table.remove_entry(&entry.key); - } - - let entry = table.random_entry(ALL_ENTRY); - assert!(matches!(entry, None)); - } - - #[test] - fn test_add_entries() { - let setup = Setup::new(); - let mut table = setup.table(); - - let key = [ - 0, 0, 0, 0, 0, 0, 0, 1, 3, 4, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, - 0, 0, 5, - ]; - - let key2 = [ - 0, 0, 0, 0, 0, 0, 0, 1, 2, 4, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, - 0, 0, 5, - ]; - - let entry1 = new_entry(&key, &Addr::Ip("240.120.3.1".parse().unwrap()), 3000, 3010); - assert!(matches!( - table.add_entry(entry1.clone()), - AddEntryResult::Added - )); - - assert!(matches!(table.add_entry(entry1), AddEntryResult::Exists)); - - let entry2 = new_entry(&key2, &Addr::Ip("240.120.3.2".parse().unwrap()), 3000, 3010); - assert!(matches!( - table.add_entry(entry2), - AddEntryResult::Restricted - )); - - let mut key: [u8; 32] = [0; 32]; - - for i in 0..BUCKET_SIZE { - key[i] += 1; - let entry = new_entry( - &key, - &Addr::Ip(format!("127.0.{i}.1").parse().unwrap()), - 3000, - 3010, - ); - table.add_entry(entry); - } - - key[BUCKET_SIZE] += 1; - let entry = new_entry(&key, &Addr::Ip("125.20.0.1".parse().unwrap()), 3000, 3010); - assert!(matches!(table.add_entry(entry), AddEntryResult::Ignored)); - } -} diff --git a/karyons_p2p/src/utils/mod.rs b/karyons_p2p/src/utils/mod.rs deleted file mode 100644 index e8ff9d0..0000000 --- a/karyons_p2p/src/utils/mod.rs +++ /dev/null @@ -1,21 +0,0 @@ -mod version; - -pub use version::{version_match, Version, VersionInt}; - -use std::net::IpAddr; - -use karyons_net::Addr; - -/// Check if two addresses belong to the same subnet. -pub fn subnet_match(addr: &Addr, other_addr: &Addr) -> bool { - match (addr, other_addr) { - (Addr::Ip(IpAddr::V4(ip)), Addr::Ip(IpAddr::V4(other_ip))) => { - // XXX Consider moving this to a different location - if other_ip.is_loopback() && ip.is_loopback() { - return false; - } - ip.octets()[0..3] == other_ip.octets()[0..3] - } - _ => false, - } -} diff --git a/karyons_p2p/src/utils/version.rs b/karyons_p2p/src/utils/version.rs deleted file mode 100644 index 4986495..0000000 --- a/karyons_p2p/src/utils/version.rs +++ /dev/null @@ -1,93 +0,0 @@ -use std::str::FromStr; - -use bincode::{Decode, Encode}; -use semver::VersionReq; - -use crate::{Error, Result}; - -/// Represents the network version and protocol version used in Karyons p2p. -/// -/// # Example -/// -/// ``` -/// use karyons_p2p::Version; -/// -/// let version: Version = "0.2.0, >0.1.0".parse().unwrap(); -/// -/// let version: Version = "0.2.0".parse().unwrap(); -/// -/// ``` -#[derive(Debug, Clone)] -pub struct Version { - pub v: VersionInt, - pub req: VersionReq, -} - -impl Version { - /// Creates a new Version - pub fn new(v: VersionInt, req: VersionReq) -> Self { - Self { v, req } - } -} - -#[derive(Debug, Decode, Encode, Clone)] -pub struct VersionInt { - major: u64, - minor: u64, - patch: u64, -} - -impl FromStr for Version { - type Err = Error; - - fn from_str(s: &str) -> Result<Self> { - let v: Vec<&str> = s.split(", ").collect(); - if v.is_empty() || v.len() > 2 { - return Err(Error::ParseError(format!("Invalid version{s}"))); - } - - let version: VersionInt = v[0].parse()?; - let req: VersionReq = if v.len() > 1 { v[1] } else { v[0] }.parse()?; - - Ok(Self { v: version, req }) - } -} - -impl FromStr for VersionInt { - type Err = Error; - - fn from_str(s: &str) -> Result<Self> { - let v: Vec<&str> = s.split('.').collect(); - if v.len() < 2 || v.len() > 3 { - return Err(Error::ParseError(format!("Invalid version{s}"))); - } - - let major = v[0].parse::<u64>()?; - let minor = v[1].parse::<u64>()?; - let patch = v.get(2).unwrap_or(&"0").parse::<u64>()?; - - Ok(Self { - major, - minor, - patch, - }) - } -} - -impl std::fmt::Display for VersionInt { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - write!(f, "{}.{}.{}", self.major, self.minor, self.patch) - } -} - -impl From<VersionInt> for semver::Version { - fn from(v: VersionInt) -> Self { - semver::Version::new(v.major, v.minor, v.patch) - } -} - -/// Check if a version satisfies a version request. -pub fn version_match(version_req: &VersionReq, version: &VersionInt) -> bool { - let version: semver::Version = version.clone().into(); - version_req.matches(&version) -} |