net: dnet cleanups and improvements.

Cargo.toml

@@ -211,7 +211,6 @@ net = [
     "x509-parser",
     "semver",
     "serde",
-    "serde_json",
     "socket2",
     "url",
 

src/net/channel.rs

@@ -24,7 +24,6 @@ use futures::{
 };
 use log::{debug, error, info};
 use rand::{rngs::OsRng, Rng};
-use serde_json::json;
 use smol::Executor;
 use url::Url;
 
@@ -45,45 +44,25 @@ use crate::{
 /// Atomic pointer to async channel
 pub type ChannelPtr = Arc<Channel>;
 
-const RINGBUFFER_SIZE: usize = 512;
-
 /// Channel debug info
-struct ChannelInfo {
-    random_id: u32,
-    remote_node_id: String,
-    log: Mutex<RingBuffer<(NanoTimestamp, String, String)>>,
+#[derive(Clone)]
+pub struct ChannelInfo {
+    pub addr: Url,
+    pub random_id: u32,
+    pub remote_node_id: String,
+    pub log: RingBuffer<(NanoTimestamp, String, String), 512>,
 }
 
 impl ChannelInfo {
-    fn new() -> Self {
-        Self {
-            random_id: OsRng.gen(),
-            remote_node_id: String::new(),
-            log: Mutex::new(RingBuffer::new(RINGBUFFER_SIZE)),
-        }
+    fn new(addr: Url) -> Self {
+        Self { addr, random_id: OsRng.gen(), remote_node_id: String::new(), log: RingBuffer::new() }
     }
 
     /// Get available debug info, resets the ringbuffer when called.
-    async fn get_info(&self) -> serde_json::Value {
-        let mut lock = self.log.lock().await;
-        let log = lock.clone();
-        *lock = RingBuffer::new(RINGBUFFER_SIZE);
-        drop(lock);
-
-        let (last_msg, last_status) = {
-            match log.back() {
-                Some((_, m, s)) => (m.clone(), s.clone()),
-                None => (String::new(), String::new()),
-            }
-        };
-
-        json!({
-            "random_id": self.random_id,
-            "remote_node_id": self.remote_node_id,
-            "last_msg": last_msg,
-            "last_status": last_status,
-            "log": log,
-        })
+    fn dnet_info(&mut self) -> Self {
+        let info = self.clone();
+        self.log = RingBuffer::new();
+        info
     }
 }
 
@@ -132,7 +111,7 @@ impl Channel {
         Self::setup_dispatchers(&message_subsystem).await;
 
         let info = if *session.upgrade().unwrap().p2p().dnet_enabled.lock().await {
-            Mutex::new(Some(ChannelInfo::new()))
+            Mutex::new(Some(ChannelInfo::new(address.clone())))
         } else {
             Mutex::new(None)
         };
@@ -160,22 +139,26 @@ impl Channel {
         subsystem.add_dispatch::<message::AddrsMessage>().await;
     }
 
-    /// Fetch debug info, if any
-    pub async fn get_info(&self) -> serde_json::Value {
-        if *self.p2p().dnet_enabled.lock().await {
-            // Maybe here we should panic? It probably should never be
-            // the case that dnet is enabled, but this stuff is empty.
-            // However it's possible that it somehow happens through a
-            // race condition, so let's be safe.
-            match self.info.lock().await.as_ref() {
-                Some(info) => info.get_info().await,
-                None => json!({}),
-            }
-        } else {
-            json!({})
+    /// Fetch dnet info for the channel, if enabled.
+    /// Returns the [`Channel::address`] and [`ChannelInfo`].
+    pub(crate) async fn dnet_info(&self) -> ChannelInfo {
+        // We're unwrapping here because if we get None it means
+        // there's a bug somehwere where we initialized dnet but
+        // ChannelInfo was not created.
+        self.info.lock().await.as_mut().unwrap().dnet_info()
+    }
+
+    pub(crate) async fn dnet_enable(&self) {
+        let mut info = self.info.lock().await;
+        if info.is_none() {
+            *info = Some(ChannelInfo::new(self.address.clone()));
         }
     }
 
+    pub(crate) async fn dnet_disable(&self) {
+        *self.info.lock().await = None;
+    }
+
     /// Starts the channel. Runs a receive loop to start receiving messages
     /// or handles a network failure.
     pub fn start(self: Arc<Self>, executor: Arc<Executor<'_>>) {
@@ -265,9 +248,10 @@ impl Channel {
 
         dnet!(self,
             let time = NanoTimestamp::current_time();
-            let info_lock = self.info.lock().await;
-            let mut log = info_lock.as_ref().unwrap().log.lock().await;
-            log.push((time, "send".to_string(), packet.command.clone()));
+            match self.info.lock().await.as_mut() {
+                Some(info) => info.log.push((time, "send".into(), packet.command.clone())),
+                None => unreachable!(),
+            }
         );
 
         let stream = &mut *self.writer.lock().await;

src/net/p2p.rs

@@ -25,7 +25,6 @@ use async_std::{
 use futures::{stream::FuturesUnordered, TryFutureExt};
 use log::{debug, error, info, warn};
 use rand::{prelude::IteratorRandom, rngs::OsRng};
-use serde_json::json;
 use smol::Executor;
 use url::Url;
 
@@ -35,8 +34,9 @@ use super::{
     message::Message,
     protocol::{protocol_registry::ProtocolRegistry, register_default_protocols},
     session::{
-        InboundSession, InboundSessionPtr, ManualSession, ManualSessionPtr, OutboundSession,
-        OutboundSessionPtr, SeedSyncSession, Session,
+        inbound_session::InboundDnet, outbound_session::OutboundDnet, InboundSession,
+        InboundSessionPtr, ManualSession, ManualSessionPtr, OutboundSession, OutboundSessionPtr,
+        SeedSyncSession, Session,
     },
     settings::{Settings, SettingsPtr},
 };
@@ -66,6 +66,18 @@ enum P2pState {
     Stopped,
 }
 
+/// Types of DnetInfo (used with sessions)
+pub enum DnetInfo {
+    /// Hosts info
+    Hosts(Vec<Url>),
+    /// Outbound Session Info
+    Outbound(OutboundDnet),
+    /// Inbound Session Info
+    Inbound(InboundDnet),
+    // Manual Session Info
+    //Manual(ManualDnet),
+}
+
 impl std::fmt::Display for P2pState {
     fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
         write!(
@@ -155,21 +167,6 @@ impl P2p {
         self_
     }
 
-    pub async fn get_info(&self) -> serde_json::Value {
-        let mut ext = vec![];
-        for addr in &self.settings.external_addrs {
-            ext.push(addr.to_string());
-        }
-
-        json!({
-            "external_addrs": format!("{:?}", ext),
-            //"session_manual": self.session_manual().await.get_info().await,
-            "session_inbound": self.session_inbound().await.get_info().await,
-            "session_outbound": self.session_outbound().await.get_info().await,
-            "state": self.state.lock().await.to_string(),
-        })
-    }
-
     /// Invoke startup and seeding sequence. Call from constructing thread.
     pub async fn start(self: Arc<Self>, ex: Arc<Executor<'_>>) -> Result<()> {
         debug!(target: "net::p2p::start()", "P2P::start() [BEGIN]");
@@ -343,16 +340,41 @@ impl P2p {
     }
 
     /// Enable network debugging
-    pub async fn enable_dnet(&self) {
+    pub async fn dnet_enable(&self) {
+        // Enable log for all connected channels if not enabled already
+        for channel in self.channels().lock().await.values() {
+            channel.dnet_enable().await;
+        }
+
         *self.dnet_enabled.lock().await = true;
         warn!("[P2P] Network debugging enabled!");
     }
 
     /// Disable network debugging
-    pub async fn disable_dnet(&self) {
+    pub async fn dnet_disable(&self) {
         *self.dnet_enabled.lock().await = false;
+
+        // Clear out any held data
+        for channel in self.channels().lock().await.values() {
+            channel.dnet_disable().await;
+        }
+
         warn!("[P2P] Network debugging disabled!");
     }
+
+    /// Gather session dnet info and return it in a vec.
+    /// Returns an empty vec if dnet is disabled.
+    pub async fn dnet_info(&self) -> Vec<DnetInfo> {
+        let mut ret = vec![];
+
+        if *self.dnet_enabled.lock().await {
+            ret.push(self.session_inbound().await.dnet_info().await);
+            ret.push(self.session_outbound().await.dnet_info().await);
+            ret.push(DnetInfo::Hosts(self.hosts.load_all().await));
+        }
+
+        ret
+    }
 }
 
 macro_rules! dnet {

src/net/session/inbound_session.rs

@@ -28,15 +28,14 @@ use std::collections::HashMap;
 use async_std::sync::{Arc, Mutex, Weak};
 use async_trait::async_trait;
 use log::{error, info};
-use serde_json::json;
 use smol::Executor;
 use url::Url;
 
 use super::{
     super::{
         acceptor::{Acceptor, AcceptorPtr},
-        channel::ChannelPtr,
-        p2p::{P2p, P2pPtr},
+        channel::{ChannelInfo, ChannelPtr},
+        p2p::{DnetInfo, P2p, P2pPtr},
     },
     Session, SessionBitFlag, SESSION_INBOUND,
 };
@@ -47,14 +46,26 @@ use crate::{
 
 pub type InboundSessionPtr = Arc<InboundSession>;
 
-/// Channel debug info
-struct InboundInfo {
-    channel: ChannelPtr,
+/// dnet info for an inbound connection
+#[derive(Clone)]
+pub struct InboundInfo {
+    /// Remote address
+    pub addr: Option<Url>,
+    /// Channel info
+    pub channel: Option<ChannelInfo>,
 }
 
 impl InboundInfo {
-    async fn get_info(&self) -> serde_json::Value {
-        self.channel.get_info().await
+    async fn dnet_info(&self, p2p: P2pPtr) -> Option<Self> {
+        let Some(ref addr) = self.addr else {
+            return None
+        };
+
+        let Some(chan) = p2p.channels().lock().await.get(&addr).cloned() else {
+            return None
+        };
+
+        Some(Self { addr: self.addr.clone(), channel: Some(chan.dnet_info().await) })
     }
 }
 
@@ -171,8 +182,9 @@ impl InboundSession {
         self.register_channel(channel.clone(), ex.clone()).await?;
 
         let addr = channel.address().clone();
+
         self.connect_infos.lock().await[index]
-            .insert(addr.clone(), InboundInfo { channel: channel.clone() });
+            .insert(addr.clone(), InboundInfo { addr: Some(addr.clone()), channel: None });
 
         let stop_sub = channel.subscribe_stop().await?;
         stop_sub.receive().await;
@@ -183,6 +195,12 @@ impl InboundSession {
     }
 }
 
+/// Dnet information for the inbound session
+pub struct InboundDnet {
+    /// Slot information
+    pub slots: Vec<Option<InboundInfo>>,
+}
+
 #[async_trait]
 impl Session for InboundSession {
     fn p2p(&self) -> P2pPtr {
@@ -193,17 +211,15 @@ impl Session for InboundSession {
         SESSION_INBOUND
     }
 
-    async fn get_info(&self) -> serde_json::Value {
-        let mut infos = HashMap::new();
-        for (index, accept_addr) in self.p2p().settings().inbound_addrs.iter().enumerate() {
-            let connect_infos = &self.connect_infos.lock().await[index];
-            for (addr, info) in connect_infos {
-                let json_addr = json!({ "accept_addr": accept_addr });
-                let info = vec![json_addr, info.get_info().await];
-                infos.insert(addr.to_string(), info);
+    async fn dnet_info(&self) -> DnetInfo {
+        let mut slots = vec![];
+
+        for listen_addr in (*self.connect_infos.lock().await).iter() {
+            for slot in listen_addr.values() {
+                slots.push(slot.dnet_info(self.p2p()).await);
             }
         }
 
-        json!({ "connected": infos })
+        DnetInfo::Inbound(InboundDnet { slots })
     }
 }

src/net/session/manual_session.rs

@@ -39,7 +39,7 @@ use super::{
     super::{
         channel::ChannelPtr,
         connector::Connector,
-        p2p::{P2p, P2pPtr},
+        p2p::{DnetInfo, P2p, P2pPtr},
     },
     Session, SessionBitFlag, SESSION_MANUAL,
 };
@@ -213,7 +213,7 @@ impl Session for ManualSession {
         SESSION_MANUAL
     }
 
-    async fn get_info(&self) -> serde_json::Value {
+    async fn dnet_info(&self) -> DnetInfo {
         todo!()
     }
 }

src/net/session/mod.rs

@@ -21,7 +21,11 @@ use async_trait::async_trait;
 use log::debug;
 use smol::Executor;
 
-use super::{channel::ChannelPtr, p2p::P2pPtr, protocol::ProtocolVersion};
+use super::{
+    channel::ChannelPtr,
+    p2p::{DnetInfo, P2pPtr},
+    protocol::ProtocolVersion,
+};
 use crate::Result;
 
 pub mod inbound_session;
@@ -152,5 +156,5 @@ pub trait Session: Sync {
     fn type_id(&self) -> SessionBitFlag;
 
     /// Get network debug info
-    async fn get_info(&self) -> serde_json::Value;
+    async fn dnet_info(&self) -> DnetInfo;
 }

src/net/session/outbound_session.rs

@@ -31,16 +31,15 @@ use std::collections::HashSet;
 use async_std::sync::{Arc, Mutex, Weak};
 use async_trait::async_trait;
 use log::{debug, error, info};
-use serde_json::json;
 use smol::Executor;
 use url::Url;
 
 use super::{
     super::{
-        channel::ChannelPtr,
+        channel::{ChannelInfo, ChannelPtr},
         connector::Connector,
         message::GetAddrsMessage,
-        p2p::{P2p, P2pPtr},
+        p2p::{DnetInfo, P2p, P2pPtr},
     },
     Session, SessionBitFlag, SESSION_OUTBOUND,
 };
@@ -52,8 +51,9 @@ use crate::{
 
 pub type OutboundSessionPtr = Arc<OutboundSession>;
 
-#[derive(Clone)]
-enum OutboundState {
+/// Connection state
+#[derive(Copy, Clone)]
+pub enum OutboundState {
     Open,
     Pending,
     Connected,
@@ -73,29 +73,31 @@ impl std::fmt::Display for OutboundState {
     }
 }
 
+/// dnet info for an outbound connection
 #[derive(Clone)]
-struct OutboundInfo {
-    addr: Option<Url>,
-    channel: Option<ChannelPtr>,
-    state: OutboundState,
+pub struct OutboundInfo {
+    /// Remote address
+    pub addr: Option<Url>,
+    /// Channel info
+    pub channel: Option<ChannelInfo>,
+    /// Connection state
+    pub state: OutboundState,
 }
 
 impl OutboundInfo {
-    async fn get_info(&self) -> serde_json::Value {
-        let addr = match self.addr.as_ref() {
-            Some(addr) => serde_json::Value::String(addr.to_string()),
-            None => serde_json::Value::Null,
+    async fn dnet_info(&self, p2p: P2pPtr) -> Option<Self> {
+        let Some(ref addr) = self.addr else {
+            return None
         };
 
-        let channel = match &self.channel {
-            Some(channel) => channel.get_info().await,
-            None => serde_json::Value::Null,
+        let Some(chan) = p2p.channels().lock().await.get(&addr).cloned() else {
+            return None
         };
 
-        json!({
-            "addr": addr,
-            "state": self.state.to_string(),
-            "channel": channel,
+        Some(Self {
+            addr: self.addr.clone(),
+            channel: Some(chan.dnet_info().await),
+            state: self.state,
         })
     }
 }
@@ -108,13 +110,15 @@ impl Default for OutboundInfo {
 
 /// Defines outbound connections session.
 pub struct OutboundSession {
+    /// Weak pointer to parent p2p object
     p2p: Weak<P2p>,
+    /// Outbound connection slots
     connect_slots: Mutex<Vec<StoppableTaskPtr>>,
     /// Subscriber used to signal channels processing
     channel_subscriber: SubscriberPtr<Result<ChannelPtr>>,
     /// Flag to toggle channel_subscriber notifications
     notify: Mutex<bool>,
-    /// Channel debug info
+    /// Channel debug info, corresponds to `connect_slots`
     slot_info: Mutex<Vec<OutboundInfo>>,
 }
 
@@ -137,6 +141,7 @@ impl OutboundSession {
         // Activate mutex lock on connection slots.
         let mut connect_slots = self.connect_slots.lock().await;
 
+        // Create dnet stub
         self.slot_info.lock().await.resize(n_slots, Default::default());
 
         for i in 0..n_slots {
@@ -256,7 +261,6 @@ impl OutboundSession {
 
                 dnet!(self,
                     let info = &mut self.slot_info.lock().await[slot_number];
-                    info.channel = Some(channel.clone());
                     info.state = OutboundState::Connected;
                 );
 
@@ -286,7 +290,6 @@ impl OutboundSession {
         dnet!(self,
             let info = &mut self.slot_info.lock().await[slot_number];
             info.addr = None;
-            info.channel = None;
             info.state = OutboundState::Open;
         );
 
@@ -444,6 +447,12 @@ impl OutboundSession {
     }
 }
 
+/// Dnet information for the outbound session
+pub struct OutboundDnet {
+    /// Slot information
+    pub slots: Vec<Option<OutboundInfo>>,
+}
+
 #[async_trait]
 impl Session for OutboundSession {
     fn p2p(&self) -> P2pPtr {
@@ -454,19 +463,15 @@ impl Session for OutboundSession {
         SESSION_OUTBOUND
     }
 
-    async fn get_info(&self) -> serde_json::Value {
+    async fn dnet_info(&self) -> DnetInfo {
+        // We fetch channel infos for all outbound slots.
+        // If a slot is not connected, it will be `None`.
         let mut slots = vec![];
-        for info in &*self.slot_info.lock().await {
-            slots.push(info.get_info().await);
-        }
 
-        let hosts = self.p2p().hosts().load_all().await;
-        let addrs: Vec<serde_json::Value> =
-            hosts.iter().map(|addr| serde_json::Value::String(addr.to_string())).collect();
+        for slot in self.slot_info.lock().await.iter() {
+            slots.push(slot.dnet_info(self.p2p()).await);
+        }
 
-        json!({
-            "slots": slots,
-            "hosts": serde_json::Value::Array(addrs),
-        })
+        DnetInfo::Outbound(OutboundDnet { slots })
     }
 }

src/net/session/seedsync_session.rs

@@ -50,8 +50,11 @@ use smol::Executor;
 use url::Url;
 
 use super::{
-    super::{connector::Connector, p2p::P2p},
-    P2pPtr, Session, SessionBitFlag, SESSION_SEED,
+    super::{
+        connector::Connector,
+        p2p::{DnetInfo, P2p, P2pPtr},
+    },
+    Session, SessionBitFlag, SESSION_SEED,
 };
 use crate::Result;
 
@@ -199,7 +202,7 @@ impl Session for SeedSyncSession {
         SESSION_SEED
     }
 
-    async fn get_info(&self) -> serde_json::Value {
+    async fn dnet_info(&self) -> DnetInfo {
         todo!()
     }
 }