diff options
Diffstat (limited to 'p2p/examples')
| -rw-r--r-- | p2p/examples/chat.rs | 44 | ||||
| -rw-r--r-- | p2p/examples/monitor.rs | 60 | ||||
| -rw-r--r-- | p2p/examples/peer.rs | 43 | ||||
| -rw-r--r-- | p2p/examples/shared/mod.rs | 33 |
4 files changed, 106 insertions, 74 deletions
diff --git a/p2p/examples/chat.rs b/p2p/examples/chat.rs index 4358362..907ba06 100644 --- a/p2p/examples/chat.rs +++ b/p2p/examples/chat.rs @@ -1,9 +1,11 @@ +mod shared; + use std::sync::Arc; use async_std::io; use async_trait::async_trait; use clap::Parser; -use smol::{channel, future, Executor}; +use smol::{channel, Executor}; use karyons_net::{Endpoint, Port}; @@ -12,6 +14,8 @@ use karyons_p2p::{ ArcPeer, Backend, Config, P2pError, PeerID, Version, }; +use shared::run_executor; + #[derive(Parser)] #[command(author, version, about, long_about = None)] struct Cli { @@ -109,33 +113,33 @@ fn main() { ..Default::default() }; + // Create a new Executor + let ex = Arc::new(Executor::new()); + // Create a new Backend - let backend = Backend::new(peer_id, config); + let backend = Backend::new(peer_id, config, ex.clone()); let (ctrlc_s, ctrlc_r) = channel::unbounded(); let handle = move || ctrlc_s.try_send(()).unwrap(); ctrlc::set_handler(handle).unwrap(); - // Create a new Executor - let ex = Arc::new(Executor::new()); - - let ex_cloned = ex.clone(); - let task = ex.spawn(async { - let username = cli.username; - - // Attach the ChatProtocol - let c = move |peer| ChatProtocol::new(&username, peer); - backend.attach_protocol::<ChatProtocol>(c).await.unwrap(); + run_executor( + async { + let username = cli.username; - // Run the backend - backend.run(ex_cloned).await.unwrap(); + // Attach the ChatProtocol + let c = move |peer| ChatProtocol::new(&username, peer); + backend.attach_protocol::<ChatProtocol>(c).await.unwrap(); - // Wait for ctrlc signal - ctrlc_r.recv().await.unwrap(); + // Run the backend + backend.run().await.unwrap(); - // Shutdown the backend - backend.shutdown().await; - }); + // Wait for ctrlc signal + ctrlc_r.recv().await.unwrap(); - future::block_on(ex.run(task)); + // Shutdown the backend + backend.shutdown().await; + }, + ex, + ); } diff --git a/p2p/examples/monitor.rs b/p2p/examples/monitor.rs index cd4defc..fc48c2f 100644 --- a/p2p/examples/monitor.rs +++ b/p2p/examples/monitor.rs @@ -1,13 +1,16 @@ +mod shared; + use std::sync::Arc; use clap::Parser; -use easy_parallel::Parallel; -use smol::{channel, future, Executor}; +use smol::{channel, Executor}; use karyons_net::{Endpoint, Port}; use karyons_p2p::{Backend, Config, PeerID}; +use shared::run_executor; + #[derive(Parser)] #[command(author, version, about, long_about = None)] struct Cli { @@ -50,44 +53,39 @@ fn main() { ..Default::default() }; + // Create a new Executor + let ex = Arc::new(Executor::new()); + // Create a new Backend - let backend = Backend::new(peer_id, config); + let backend = Backend::new(peer_id, config, ex.clone()); let (ctrlc_s, ctrlc_r) = channel::unbounded(); let handle = move || ctrlc_s.try_send(()).unwrap(); ctrlc::set_handler(handle).unwrap(); - let (signal, shutdown) = channel::unbounded::<()>(); - - // Create a new Executor - let ex = Arc::new(Executor::new()); - - let task = async { - let monitor = backend.monitor().await; + let exc = ex.clone(); + run_executor( + async { + let monitor = backend.monitor().await; - let monitor_task = ex.spawn(async move { - loop { - let event = monitor.recv().await.unwrap(); - println!("{}", event); - } - }); + let monitor_task = exc.spawn(async move { + loop { + let event = monitor.recv().await.unwrap(); + println!("{}", event); + } + }); - // Run the backend - backend.run(ex.clone()).await.unwrap(); + // Run the backend + backend.run().await.unwrap(); - // Wait for ctrlc signal - ctrlc_r.recv().await.unwrap(); + // Wait for ctrlc signal + ctrlc_r.recv().await.unwrap(); - // Shutdown the backend - backend.shutdown().await; - - monitor_task.cancel().await; - - drop(signal); - }; + // Shutdown the backend + backend.shutdown().await; - // Run four executor threads. - Parallel::new() - .each(0..4, |_| future::block_on(ex.run(shutdown.recv()))) - .finish(|| future::block_on(task)); + monitor_task.cancel().await; + }, + ex, + ); } diff --git a/p2p/examples/peer.rs b/p2p/examples/peer.rs index f805d68..5ff365d 100644 --- a/p2p/examples/peer.rs +++ b/p2p/examples/peer.rs @@ -1,13 +1,16 @@ +mod shared; + use std::sync::Arc; use clap::Parser; -use easy_parallel::Parallel; -use smol::{channel, future, Executor}; +use smol::{channel, Executor}; use karyons_net::{Endpoint, Port}; use karyons_p2p::{Backend, Config, PeerID}; +use shared::run_executor; + #[derive(Parser)] #[command(author, version, about, long_about = None)] struct Cli { @@ -50,33 +53,27 @@ fn main() { ..Default::default() }; + // Create a new Executor + let ex = Arc::new(Executor::new()); + // Create a new Backend - let backend = Backend::new(peer_id, config); + let backend = Backend::new(peer_id, config, ex.clone()); let (ctrlc_s, ctrlc_r) = channel::unbounded(); let handle = move || ctrlc_s.try_send(()).unwrap(); ctrlc::set_handler(handle).unwrap(); - let (signal, shutdown) = channel::unbounded::<()>(); - - // Create a new Executor - let ex = Arc::new(Executor::new()); - - let task = async { - // Run the backend - backend.run(ex.clone()).await.unwrap(); + run_executor( + async { + // Run the backend + backend.run().await.unwrap(); - // Wait for ctrlc signal - ctrlc_r.recv().await.unwrap(); - - // Shutdown the backend - backend.shutdown().await; - - drop(signal); - }; + // Wait for ctrlc signal + ctrlc_r.recv().await.unwrap(); - // Run four executor threads. - Parallel::new() - .each(0..4, |_| future::block_on(ex.run(shutdown.recv()))) - .finish(|| future::block_on(task)); + // Shutdown the backend + backend.shutdown().await; + }, + ex, + ); } diff --git a/p2p/examples/shared/mod.rs b/p2p/examples/shared/mod.rs new file mode 100644 index 0000000..9a8e387 --- /dev/null +++ b/p2p/examples/shared/mod.rs @@ -0,0 +1,33 @@ +use std::{num::NonZeroUsize, thread}; + +use easy_parallel::Parallel; +use smol::{channel, future, future::Future}; + +use karyons_core::Executor; + +/// Returns an estimate of the default amount of parallelism a program should use. +/// see `std::thread::available_parallelism` +fn available_parallelism() -> usize { + thread::available_parallelism() + .map(NonZeroUsize::get) + .unwrap_or(1) +} + +/// Run a multi-threaded executor +pub fn run_executor<T>(main_future: impl Future<Output = T>, ex: Executor<'_>) { + let (signal, shutdown) = channel::unbounded::<()>(); + + let num_threads = available_parallelism(); + + Parallel::new() + .each(0..(num_threads), |_| { + future::block_on(ex.run(shutdown.recv())) + }) + // Run the main future on the current thread. + .finish(|| { + future::block_on(async { + main_future.await; + drop(signal); + }) + }); +} |