From 0c0699c0460c1b149915729223eec701bde481df Mon Sep 17 00:00:00 2001 From: hozan23 Date: Fri, 14 Jun 2024 02:04:43 +0200 Subject: p2p: WIP implement rpc server for the p2p monitor --- p2p/examples/monitor/Cargo.lock | 43 +++++ p2p/examples/monitor/Cargo.toml | 1 + p2p/examples/monitor/src/client.rs | 36 ++++- p2p/examples/monitor/src/main.rs | 240 +--------------------------- p2p/examples/monitor/src/service.rs | 309 ++++++++++++++++++++++++++++++++++++ 5 files changed, 391 insertions(+), 238 deletions(-) create mode 100644 p2p/examples/monitor/src/service.rs (limited to 'p2p/examples/monitor') diff --git a/p2p/examples/monitor/Cargo.lock b/p2p/examples/monitor/Cargo.lock index d881b93..25defa2 100644 --- a/p2p/examples/monitor/Cargo.lock +++ b/p2p/examples/monitor/Cargo.lock @@ -930,6 +930,21 @@ version = "1.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "42703706b716c37f96a77aea830392ad231f44c9e9a67872fa5548707e11b11c" +[[package]] +name = "futures" +version = "0.3.30" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "645c6916888f6cb6350d2550b80fb63e734897a8498abe35cfb732b6487804b0" +dependencies = [ + "futures-channel", + "futures-core", + "futures-executor", + "futures-io", + "futures-sink", + "futures-task", + "futures-util", +] + [[package]] name = "futures-channel" version = "0.3.30" @@ -937,6 +952,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "eac8f7d7865dcb88bd4373ab671c8cf4508703796caa2b1985a9ca867b3fcb78" dependencies = [ "futures-core", + "futures-sink", ] [[package]] @@ -945,6 +961,17 @@ version = "0.3.30" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "dfc6580bb841c5a68e9ef15c77ccc837b40a7504914d52e47b8b0e9bbda25a1d" +[[package]] +name = "futures-executor" +version = "0.3.30" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a576fc72ae164fca6b9db127eaa9a9dda0d61316034f33a0a0d4eda41f02b01d" +dependencies = [ + "futures-core", + "futures-task", + "futures-util", +] + [[package]] name = "futures-io" version = "0.3.30" @@ -979,6 +1006,17 @@ dependencies = [ "pin-project-lite", ] +[[package]] +name = "futures-macro" +version = "0.3.30" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "87750cf4b7a4c0625b1529e4c543c2182106e4dedc60a2a6455e00d212c489ac" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.66", +] + [[package]] name = "futures-rustls" version = "0.25.1" @@ -1008,9 +1046,13 @@ version = "0.3.30" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3d6401deb83407ab3da39eba7e33987a73c3df0c82b4bb5813ee871c19c41d48" dependencies = [ + "futures-channel", "futures-core", + "futures-io", + "futures-macro", "futures-sink", "futures-task", + "memchr", "pin-project-lite", "pin-utils", "slab", @@ -1386,6 +1428,7 @@ dependencies = [ "ctrlc", "easy-parallel", "env_logger", + "futures", "karyon_core", "karyon_jsonrpc", "karyon_p2p", diff --git a/p2p/examples/monitor/Cargo.toml b/p2p/examples/monitor/Cargo.toml index b323ed1..52bab15 100644 --- a/p2p/examples/monitor/Cargo.toml +++ b/p2p/examples/monitor/Cargo.toml @@ -18,6 +18,7 @@ smol = "2.0.0" serde_json = "1.0.117" easy-parallel = "3.3.1" ringbuffer = "0.15.0" +futures = "0.3.30" [[bin]] name = "client" diff --git a/p2p/examples/monitor/src/client.rs b/p2p/examples/monitor/src/client.rs index b81c286..2f4d00d 100644 --- a/p2p/examples/monitor/src/client.rs +++ b/p2p/examples/monitor/src/client.rs @@ -1,4 +1,5 @@ use clap::Parser; +use serde::{Deserialize, Serialize}; use karyon_jsonrpc::Client; use karyon_p2p::endpoint::Endpoint; @@ -11,6 +12,9 @@ struct Cli { rpc_endpoint: Endpoint, } +#[derive(Deserialize, Serialize)] +struct Pong {} + fn main() { smol::block_on(async { env_logger::init(); @@ -32,18 +36,44 @@ fn main() { .await .expect("Subscribe to peer pool events"); + let (_, sub3) = rpc + .subscribe("MonitorRPC.discovery_subscribe", ()) + .await + .expect("Subscribe to discovery events"); + + smol::spawn(async move { + loop { + let event = sub.recv().await.expect("Receive connection event"); + println!("Receive new connection event: {event}"); + } + }) + .detach(); + smol::spawn(async move { loop { - let _event = sub.recv().await.expect("Receive connection event"); + let event = sub2.recv().await.expect("Receive peer pool event"); + println!("Receive new peerpool event: {event}"); } }) .detach(); smol::spawn(async move { loop { - let _event = sub2.recv().await.expect("Receive peer pool event"); + let event = sub3.recv().await.expect("Receive discovery event"); + println!("Receive new discovery event: {event}"); } }) - .await; + .detach(); + + // start ping-pong loop + loop { + smol::Timer::after(std::time::Duration::from_secs(1)).await; + let _: Pong = rpc + .call("MonitorRPC.ping", ()) + .await + .expect("Receive pong message"); + + println!("Receive pong message"); + } }); } diff --git a/p2p/examples/monitor/src/main.rs b/p2p/examples/monitor/src/main.rs index c57d06c..990f8d2 100644 --- a/p2p/examples/monitor/src/main.rs +++ b/p2p/examples/monitor/src/main.rs @@ -1,28 +1,21 @@ +mod service; mod shared; use std::sync::Arc; use clap::Parser; -use log::error; -use ringbuffer::{AllocRingBuffer, RingBuffer}; -use serde::{Deserialize, Serialize}; -use smol::{channel, lock::Mutex, Executor}; +use smol::{channel, Executor}; -use karyon_core::async_util::{CondWait, TaskGroup, TaskResult}; -use karyon_jsonrpc::{ - message::SubscriptionID, rpc_impl, rpc_pubsub_impl, Channel, Server, Subscription, -}; +use karyon_jsonrpc::Server; use karyon_p2p::{ endpoint::{Endpoint, Port}, keypair::{KeyPair, KeyPairType}, - monitor::{ConnEvent, DiscoveryEvent, PeerPoolEvent}, - ArcBackend, Backend, Config, Error, Result, + Backend, Config, }; +use service::MonitorRPC; use shared::run_executor; -const EVENT_BUFFER_SIZE: usize = 30; - #[derive(Parser)] #[command(author, version, about, long_about = None)] struct Cli { @@ -47,202 +40,6 @@ struct Cli { discovery_port: Option, } -struct MonitorRPC { - backend: ArcBackend, - conn_event_buffer: Arc>>, - pp_event_buffer: Arc>>, - discv_event_buffer: Arc>>, - conn_event_condvar: Arc, - pp_event_condvar: Arc, - discv_event_condvar: Arc, - task_group: TaskGroup, -} - -impl MonitorRPC { - fn new(backend: ArcBackend, ex: Arc>) -> Arc { - Arc::new(MonitorRPC { - backend, - conn_event_buffer: Arc::new(Mutex::new(AllocRingBuffer::new(EVENT_BUFFER_SIZE))), - pp_event_buffer: Arc::new(Mutex::new(AllocRingBuffer::new(EVENT_BUFFER_SIZE))), - discv_event_buffer: Arc::new(Mutex::new(AllocRingBuffer::new(EVENT_BUFFER_SIZE))), - conn_event_condvar: Arc::new(CondWait::new()), - pp_event_condvar: Arc::new(CondWait::new()), - discv_event_condvar: Arc::new(CondWait::new()), - task_group: TaskGroup::with_executor(ex.into()), - }) - } - - async fn run(&self) -> Result<()> { - let conn_events = self.backend.monitor().conn_events().await; - let peer_pool_events = self.backend.monitor().peer_pool_events().await; - let discovery_events = self.backend.monitor().discovery_events().await; - - let conn_event_buffer = self.conn_event_buffer.clone(); - let pp_event_buffer = self.pp_event_buffer.clone(); - let discv_event_buffer = self.discv_event_buffer.clone(); - - let conn_event_condvar = self.conn_event_condvar.clone(); - let pp_event_condvar = self.pp_event_condvar.clone(); - let discv_event_condvar = self.discv_event_condvar.clone(); - - let on_failuer = |res: TaskResult>| async move { - if let TaskResult::Completed(Err(err)) = res { - error!("Event receive loop: {err}") - } - }; - - self.task_group.spawn( - async move { - loop { - let event = conn_events.recv().await?; - conn_event_buffer.lock().await.push(event); - conn_event_condvar.broadcast().await; - } - }, - on_failuer, - ); - - self.task_group.spawn( - async move { - loop { - let event = peer_pool_events.recv().await?; - pp_event_buffer.lock().await.push(event); - pp_event_condvar.broadcast().await; - } - }, - on_failuer, - ); - - self.task_group.spawn( - async move { - loop { - let event = discovery_events.recv().await?; - discv_event_buffer.lock().await.push(event); - discv_event_condvar.broadcast().await; - } - }, - on_failuer, - ); - - Ok(()) - } - - async fn shutdown(&self) { - self.task_group.cancel().await; - } -} - -#[rpc_impl] -impl MonitorRPC { - async fn peer_id( - &self, - _params: serde_json::Value, - ) -> karyon_jsonrpc::Result { - Ok(serde_json::json!(self.backend.peer_id().to_string())) - } - - async fn inbound_connection( - &self, - _params: serde_json::Value, - ) -> karyon_jsonrpc::Result { - Ok(serde_json::json!(self.backend.inbound_slots())) - } - - async fn outbound_connection( - &self, - _params: serde_json::Value, - ) -> karyon_jsonrpc::Result { - Ok(serde_json::json!(self.backend.outbound_slots())) - } -} - -#[rpc_pubsub_impl] -impl MonitorRPC { - async fn conn_subscribe( - &self, - chan: Arc, - method: String, - _params: serde_json::Value, - ) -> karyon_jsonrpc::Result { - let sub = chan.new_subscription(&method).await; - let sub_id = sub.id; - - let cond_wait = self.conn_event_condvar.clone(); - let buffer = self.conn_event_buffer.clone(); - self.task_group - .spawn(notify(sub, cond_wait, buffer), notify_failed); - - Ok(serde_json::json!(sub_id)) - } - - async fn peer_pool_subscribe( - &self, - chan: Arc, - method: String, - _params: serde_json::Value, - ) -> karyon_jsonrpc::Result { - let sub = chan.new_subscription(&method).await; - let sub_id = sub.id; - - let cond_wait = self.pp_event_condvar.clone(); - let buffer = self.pp_event_buffer.clone(); - self.task_group - .spawn(notify(sub, cond_wait, buffer), notify_failed); - - Ok(serde_json::json!(sub_id)) - } - - async fn discovery_subscribe( - &self, - chan: Arc, - method: String, - _params: serde_json::Value, - ) -> karyon_jsonrpc::Result { - let sub = chan.new_subscription(&method).await; - let sub_id = sub.id; - - let cond_wait = self.discv_event_condvar.clone(); - let buffer = self.discv_event_buffer.clone(); - self.task_group - .spawn(notify(sub, cond_wait, buffer), notify_failed); - - Ok(serde_json::json!(sub_id)) - } - - async fn conn_unsubscribe( - &self, - chan: Arc, - _method: String, - params: serde_json::Value, - ) -> karyon_jsonrpc::Result { - let sub_id: SubscriptionID = serde_json::from_value(params)?; - chan.remove_subscription(&sub_id).await; - Ok(serde_json::json!(true)) - } - - async fn peer_pool_unsubscribe( - &self, - chan: Arc, - _method: String, - params: serde_json::Value, - ) -> karyon_jsonrpc::Result { - let sub_id: SubscriptionID = serde_json::from_value(params)?; - chan.remove_subscription(&sub_id).await; - Ok(serde_json::json!(true)) - } - - async fn discovery_unsubscribe( - &self, - chan: Arc, - _method: String, - params: serde_json::Value, - ) -> karyon_jsonrpc::Result { - let sub_id: SubscriptionID = serde_json::from_value(params)?; - chan.remove_subscription(&sub_id).await; - Ok(serde_json::json!(true)) - } -} - fn main() { env_logger::init(); @@ -309,30 +106,3 @@ fn main() { ex, ); } - -async fn notify + Clone>( - sub: Subscription, - cond_wait: Arc, - buffer: Arc>>, -) -> Result<()> { - for event in buffer.lock().await.iter() { - if let Err(err) = sub.notify(serde_json::json!(event)).await { - return Err(Error::Other(format!("failed to notify: {err}"))); - } - } - loop { - cond_wait.wait().await; - cond_wait.reset().await; - if let Some(event) = buffer.lock().await.back().cloned() { - if let Err(err) = sub.notify(serde_json::json!(event)).await { - return Err(Error::Other(format!("failed to notify: {err}"))); - } - } - } -} - -async fn notify_failed(result: TaskResult>) { - if let TaskResult::Completed(Err(err)) = result { - error!("Error: {err}"); - } -} diff --git a/p2p/examples/monitor/src/service.rs b/p2p/examples/monitor/src/service.rs new file mode 100644 index 0000000..15ce8da --- /dev/null +++ b/p2p/examples/monitor/src/service.rs @@ -0,0 +1,309 @@ +use std::{collections::HashMap, sync::Arc}; + +use futures::stream::{FuturesUnordered, StreamExt}; +use log::{debug, error}; +use ringbuffer::{AllocRingBuffer, RingBuffer}; +use serde::{Deserialize, Serialize}; +use smol::{lock::Mutex, Executor}; + +use karyon_core::async_util::{TaskGroup, TaskResult}; +use karyon_jsonrpc::{message::SubscriptionID, rpc_impl, rpc_pubsub_impl, Channel, Subscription}; +use karyon_p2p::{monitor::MonitorTopic, ArcBackend, Result}; + +const EVENT_BUFFER_SIZE: usize = 60; + +pub struct MonitorRPC { + backend: ArcBackend, + subs: Arc, + task_group: TaskGroup, +} + +impl MonitorRPC { + pub fn new(backend: ArcBackend, ex: Arc>) -> Arc { + Arc::new(MonitorRPC { + backend, + task_group: TaskGroup::with_executor(ex.clone().into()), + subs: Subscriptions::new(ex), + }) + } + + pub async fn run(self: &Arc) -> Result<()> { + let conn_events = self.backend.monitor().conn_events().await; + let peer_pool_events = self.backend.monitor().peer_pool_events().await; + let discovery_events = self.backend.monitor().discovery_events().await; + + let on_failuer = |res: TaskResult>| async move { + if let TaskResult::Completed(Err(err)) = res { + error!("Event receive loop: {err}") + } + }; + + let selfc = self.clone(); + self.task_group.spawn( + async move { + loop { + let event = conn_events.recv().await?; + selfc.subs.notify(MonitorTopic::Conn, event).await; + } + }, + on_failuer, + ); + + let selfc = self.clone(); + self.task_group.spawn( + async move { + loop { + let event = peer_pool_events.recv().await?; + selfc.subs.notify(MonitorTopic::PeerPool, event).await; + } + }, + on_failuer, + ); + + let selfc = self.clone(); + self.task_group.spawn( + async move { + loop { + let event = discovery_events.recv().await?; + selfc.subs.notify(MonitorTopic::Discovery, event).await; + } + }, + on_failuer, + ); + + Ok(()) + } + + pub async fn shutdown(&self) { + self.task_group.cancel().await; + self.subs.stop().await; + } +} + +#[rpc_impl] +impl MonitorRPC { + pub async fn ping( + &self, + _params: serde_json::Value, + ) -> karyon_jsonrpc::Result { + Ok(serde_json::json!(Pong {})) + } + + pub async fn peer_id( + &self, + _params: serde_json::Value, + ) -> karyon_jsonrpc::Result { + Ok(serde_json::json!(self.backend.peer_id().to_string())) + } + + pub async fn inbound_connection( + &self, + _params: serde_json::Value, + ) -> karyon_jsonrpc::Result { + Ok(serde_json::json!(self.backend.inbound_slots())) + } + + pub async fn outbound_connection( + &self, + _params: serde_json::Value, + ) -> karyon_jsonrpc::Result { + Ok(serde_json::json!(self.backend.outbound_slots())) + } +} + +#[rpc_pubsub_impl] +impl MonitorRPC { + pub async fn conn_subscribe( + &self, + chan: Arc, + method: String, + _params: serde_json::Value, + ) -> karyon_jsonrpc::Result { + let sub = chan.new_subscription(&method).await; + let sub_id = sub.id; + + self.subs.add(MonitorTopic::Conn, sub).await; + + Ok(serde_json::json!(sub_id)) + } + + pub async fn peer_pool_subscribe( + &self, + chan: Arc, + method: String, + _params: serde_json::Value, + ) -> karyon_jsonrpc::Result { + let sub = chan.new_subscription(&method).await; + let sub_id = sub.id; + + self.subs.add(MonitorTopic::PeerPool, sub).await; + + Ok(serde_json::json!(sub_id)) + } + + pub async fn discovery_subscribe( + &self, + chan: Arc, + method: String, + _params: serde_json::Value, + ) -> karyon_jsonrpc::Result { + let sub = chan.new_subscription(&method).await; + let sub_id = sub.id; + + self.subs.add(MonitorTopic::Discovery, sub).await; + + Ok(serde_json::json!(sub_id)) + } + + pub async fn conn_unsubscribe( + &self, + chan: Arc, + _method: String, + params: serde_json::Value, + ) -> karyon_jsonrpc::Result { + let sub_id: SubscriptionID = serde_json::from_value(params)?; + chan.remove_subscription(&sub_id).await; + Ok(serde_json::json!(true)) + } + + pub async fn peer_pool_unsubscribe( + &self, + chan: Arc, + _method: String, + params: serde_json::Value, + ) -> karyon_jsonrpc::Result { + let sub_id: SubscriptionID = serde_json::from_value(params)?; + chan.remove_subscription(&sub_id).await; + Ok(serde_json::json!(true)) + } + + pub async fn discovery_unsubscribe( + &self, + chan: Arc, + _method: String, + params: serde_json::Value, + ) -> karyon_jsonrpc::Result { + let sub_id: SubscriptionID = serde_json::from_value(params)?; + chan.remove_subscription(&sub_id).await; + Ok(serde_json::json!(true)) + } +} + +#[derive(Deserialize, Serialize)] +struct Pong {} + +struct Subscriptions { + subs: Mutex>>, + buffer: Mutex>>, + task_group: TaskGroup, +} + +impl Subscriptions { + fn new(ex: Arc>) -> Arc { + let mut subs = HashMap::new(); + subs.insert(MonitorTopic::Conn, HashMap::new()); + subs.insert(MonitorTopic::PeerPool, HashMap::new()); + subs.insert(MonitorTopic::Discovery, HashMap::new()); + + let mut buffer = HashMap::new(); + buffer.insert(MonitorTopic::Conn, AllocRingBuffer::new(EVENT_BUFFER_SIZE)); + buffer.insert( + MonitorTopic::PeerPool, + AllocRingBuffer::new(EVENT_BUFFER_SIZE), + ); + buffer.insert( + MonitorTopic::Discovery, + AllocRingBuffer::new(EVENT_BUFFER_SIZE), + ); + + Arc::new(Self { + subs: Mutex::new(subs), + buffer: Mutex::new(buffer), + task_group: TaskGroup::with_executor(ex.into()), + }) + } + + /// Adds the subscription to the subs map according to the given type + async fn add(self: &Arc, ty: MonitorTopic, sub: Subscription) { + match self.subs.lock().await.get_mut(&ty) { + Some(subs) => { + subs.insert(sub.id, sub.clone()); + } + None => todo!(), + } + // Send old events in the buffer to the subscriber + self.send_old_events(ty, sub).await; + } + + /// Notifies all subscribers + async fn notify(&self, ty: MonitorTopic, event: T) { + let event = serde_json::json!(event); + // Add the new event to the ringbuffer + match self.buffer.lock().await.get_mut(&ty) { + Some(events) => events.push(event.clone()), + None => todo!(), + } + + // Notify the subscribers + match self.subs.lock().await.get_mut(&ty) { + Some(subs) => { + let mut fulist = FuturesUnordered::new(); + + for sub in subs.values() { + let fu = async { (sub.id, sub.notify(event.clone()).await) }; + fulist.push(fu) + } + + let mut cleanup_list = vec![]; + while let Some((sub_id, result)) = fulist.next().await { + if let Err(err) = result { + error!("Failed to notify the subscription: {:?} {sub_id} {err}", ty); + cleanup_list.push(sub_id); + } + } + drop(fulist); + + for sub_id in cleanup_list { + subs.remove(&sub_id); + } + } + None => todo!(), + } + } + + /// Sends old events in the ringbuffer to the new subscriber. + async fn send_old_events(self: &Arc, ty: MonitorTopic, sub: Subscription) { + let ty_cloned = ty.clone(); + let sub_id = sub.id; + let on_complete = move |res: TaskResult<()>| async move { + debug!("Send old events: {:?} {:?} {res}", ty_cloned, sub_id); + }; + + let selfc = self.clone(); + self.task_group.spawn( + async move { + match selfc.buffer.lock().await.get_mut(&ty) { + Some(events) => { + let mut fu = FuturesUnordered::new(); + + for event in events.iter().rev() { + fu.push(sub.notify(event.clone())) + } + + while let Some(result) = fu.next().await { + if result.is_err() { + return; + } + } + } + None => todo!(), + } + }, + on_complete, + ); + } + + async fn stop(&self) { + self.task_group.cancel().await; + } +} -- cgit v1.2.3