darkirc: add p2p.get_info()

bin/darkirc/src/rpc.rs

@@ -25,6 +25,7 @@ use darkfi::{
     net,
     rpc::{
         jsonrpc::{ErrorCode, JsonError, JsonRequest, JsonResponse, JsonResult, JsonSubscriber},
+        p2p_method::HandlerP2p,
         server::RequestHandler,
     },
 };
@@ -44,6 +45,7 @@ impl RequestHandler for JsonRpcInterface {
             "ping" => self.pong(req.id, req.params).await,
             "dnet.switch" => self.dnet_switch(req.id, req.params).await,
             "dnet.subscribe_events" => self.dnet_subscribe_events(req.id, req.params).await,
+            "p2p.get_info" => self.p2p_get_info(req.id, req.params).await,
             _ => JsonError::new(ErrorCode::MethodNotFound, None, req.id).into(),
         }
     }
@@ -90,3 +92,9 @@ impl JsonRpcInterface {
         self.dnet_sub.clone().into()
     }
 }
+
+impl HandlerP2p for JsonRpcInterface {
+    fn p2p(&self) -> net::P2pPtr {
+        self.p2p.clone()
+    }
+}

script/node_get-info.py

@@ -0,0 +1,125 @@
+# This file is part of DarkFi (https://dark.fi)
+#
+# Copyright (C) 2020-2023 Dyne.org foundation
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as
+# published by the Free Software Foundation, either version 3 of the
+# License, or (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <https://www.gnu.org/licenses/>.
+import asyncio, json, random, sys, time
+
+
+class JsonRpc:
+
+    async def start(self, server, port):
+        reader, writer = await asyncio.open_connection(server, port)
+        self.reader = reader
+        self.writer = writer
+
+    async def stop(self):
+        self.writer.close()
+        await self.writer.wait_closed()
+
+    async def _make_request(self, method, params):
+        ident = random.randint(0, 2**16)
+        print(ident)
+        request = {
+            "jsonrpc": "2.0",
+            "method": method,
+            "params": params,
+            "id": ident,
+        }
+
+        message = json.dumps(request) + "\n"
+        self.writer.write(message.encode())
+        await self.writer.drain()
+
+        data = await self.reader.readline()
+        message = data.decode().strip()
+        response = json.loads(message)
+        print(response)
+        return response
+
+    async def _subscribe(self, method, params):
+        ident = random.randint(0, 2**16)
+        request = {
+            "jsonrpc": "2.0",
+            "method": method,
+            "params": params,
+            "id": ident,
+        }
+
+        message = json.dumps(request) + "\n"
+        self.writer.write(message.encode())
+        await self.writer.drain()
+        print("Subscribed")
+
+    async def ping(self):
+        return await self._make_request("ping", [])
+
+    async def dnet_switch(self, state):
+        return await self._make_request("dnet.switch", [state])
+
+    async def dnet_subscribe_events(self):
+        return await self._subscribe("dnet.subscribe_events", [])
+
+
+async def main(argv):
+    rpc = JsonRpc()
+    while True:
+        try:
+            await rpc.start("localhost", 26660)
+            break
+        except OSError:
+            pass
+    response = await rpc._make_request("p2p.get_info", [])
+    info = response["result"]
+    channels = info["channels"]
+    channel_lookup = {}
+    for channel in channels:
+        id = channel["id"]
+        channel_lookup[id] = channel
+
+    print("inbound:")
+    for channel in channels:
+        if channel["session"] != "inbound":
+            continue
+        url = channel["url"]
+        print(f"  {url}")
+
+    print("outbound:")
+    for i, id in enumerate(info["outbound_slots"]):
+        if id == 0:
+            print(f"  {i}: none")
+            continue
+
+        assert id in channel_lookup
+        url = channel_lookup[id]["url"]
+        print(f"  {i}: {url}")
+
+    print("seed:")
+    for channel in channels:
+        if channel["session"] != "seed":
+            continue
+        url = channel["url"]
+        print(f"  {url}")
+
+    print("manual:")
+    for channel in channels:
+        if channel["session"] != "manual":
+            continue
+        url = channel["url"]
+        print(f"  {url}")
+
+    await rpc.stop()
+
+
+asyncio.run(main(sys.argv))

src/net/session/outbound_session.rs

@@ -27,7 +27,10 @@
 //! same time.
 
 use std::{
-    sync::{Arc, Weak},
+    sync::{
+        atomic::{AtomicU32, Ordering},
+        Arc, Weak,
+    },
     time::{Duration, Instant},
 };
 
@@ -112,6 +115,15 @@ impl OutboundSession {
         self.peer_discovery.clone().stop().await;
     }
 
+    pub async fn slot_info(&self) -> Vec<u32> {
+        let mut info = Vec::new();
+        let slots = &*self.slots.lock().await;
+        for slot in slots {
+            info.push(slot.channel_id.load(Ordering::Relaxed));
+        }
+        info
+    }
+
     fn wakeup_peer_discovery(&self) {
         self.peer_discovery.notify()
     }
@@ -139,11 +151,19 @@ pub struct Slot {
     process: StoppableTaskPtr,
     wakeup_self: CondVar,
     session: Weak<OutboundSession>,
+    // For debugging
+    channel_id: AtomicU32,
 }
 
 impl Slot {
     fn new(session: Weak<OutboundSession>, slot: u32) -> Arc<Self> {
-        Arc::new(Self { slot, process: StoppableTask::new(), wakeup_self: CondVar::new(), session })
+        Arc::new(Self {
+            slot,
+            process: StoppableTask::new(),
+            wakeup_self: CondVar::new(),
+            session,
+            channel_id: AtomicU32::new(0),
+        })
     }
 
     async fn start(self: Arc<Self>) {
@@ -229,6 +249,7 @@ impl Slot {
                         err: err.to_string()
                     });
 
+                    self.channel_id.store(0, Ordering::Relaxed);
                     continue
                 }
             };
@@ -258,10 +279,16 @@ impl Slot {
                     slot: self.slot,
                     err: err.to_string()
                 });
+
+                self.channel_id.store(0, Ordering::Relaxed);
                 continue
             }
+
+            self.channel_id.store(channel.info.id, Ordering::Relaxed);
+
             // Wait for channel to close
             stop_sub.receive().await;
+            self.channel_id.store(0, Ordering::Relaxed);
         }
     }
 

src/rpc/from_impl.rs

@@ -16,19 +16,9 @@
  * along with this program.  If not, see <https://www.gnu.org/licenses/>.
  */
 
-use std::collections::HashMap;
-use tinyjson::JsonValue::{self, Number as JsonNum, Object as JsonObj, String as JsonStr};
-
+use super::util::*;
 use crate::net;
 
-// helper functions
-fn json_map<const N: usize>(vals: [(&str, JsonValue); N]) -> JsonValue {
-    JsonObj(HashMap::from(vals.map(|(k, v)| (k.to_string(), v))))
-}
-fn json_str(val: &str) -> JsonValue {
-    JsonStr(val.to_string())
-}
-
 #[cfg(feature = "net")]
 impl From<net::channel::ChannelInfo> for JsonValue {
     fn from(info: net::channel::ChannelInfo) -> JsonValue {

src/rpc/mod.rs

@@ -33,3 +33,9 @@ pub mod clock_sync;
 
 /// Various `From` implementations
 pub mod from_impl;
+
+/// Provides optional `p2p.get_info()` method
+pub mod p2p_method;
+
+/// Json helper methods and types
+pub mod util;

src/rpc/p2p_method.rs

@@ -0,0 +1,39 @@
+use async_trait::async_trait;
+
+use super::{
+    jsonrpc::{JsonResponse, JsonResult},
+    util::*,
+};
+use crate::net;
+
+#[async_trait]
+pub trait HandlerP2p: Sync + Send {
+    async fn p2p_get_info(&self, id: u16, _params: JsonValue) -> JsonResult {
+        let mut channels = Vec::new();
+        for (url, channel) in self.p2p().channels().lock().await.iter() {
+            let session = match channel.session_type_id() {
+                net::session::SESSION_INBOUND => "inbound",
+                net::session::SESSION_OUTBOUND => "outbound",
+                net::session::SESSION_MANUAL => "manual",
+                net::session::SESSION_SEED => "seed",
+                _ => panic!("invalid result from channel.session_type_id()"),
+            };
+            channels.push(json_map([
+                ("url", JsonStr(url.clone().into())),
+                ("session", json_str(session)),
+                ("id", JsonNum(channel.info.id.into())),
+            ]));
+        }
+
+        let mut slots = Vec::new();
+        for channel_id in self.p2p().session_outbound().slot_info().await {
+            slots.push(JsonNum(channel_id.into()));
+        }
+
+        let result =
+            json_map([("channels", JsonArray(channels)), ("outbound_slots", JsonArray(slots))]);
+        JsonResponse::new(result, id).into()
+    }
+
+    fn p2p(&self) -> net::P2pPtr;
+}

src/rpc/util.rs

@@ -0,0 +1,12 @@
+use std::collections::HashMap;
+pub use tinyjson::JsonValue::{
+    self, Array as JsonArray, Number as JsonNum, Object as JsonObj, String as JsonStr,
+};
+
+// helper functions
+pub fn json_map<const N: usize>(vals: [(&str, JsonValue); N]) -> JsonValue {
+    JsonObj(HashMap::from(vals.map(|(k, v)| (k.to_string(), v))))
+}
+pub fn json_str(val: &str) -> JsonValue {
+    JsonStr(val.to_string())
+}

src/util/cli.rs

@@ -199,7 +199,7 @@ macro_rules! async_daemonize {
             }
 
             // https://docs.rs/smol/latest/smol/struct.Executor.html#examples
-            let n_threads = std::thread::available_parallelism().unwrap().get();
+            let n_threads = 1;
             let ex = std::sync::Arc::new(smol::Executor::new());
             let (signal, shutdown) = smol::channel::unbounded::<()>();
             let (_, result) = easy_parallel::Parallel::new()