diff --git a/Cargo.toml b/Cargo.toml index 8f40df0..e9f3fda 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -30,6 +30,8 @@ serde = { version = "1.0", features = ["derive"], optional = true } [lib] +# [profile.release] +# debug = true [[example]] name = "channel_sync" @@ -39,6 +41,18 @@ required-features = ["channel"] name = "channel_async" required-features = ["channel"] +[[example]] +name = "channel_sync_benchmark" +required-features = ["channel"] + +[[example]] +name = "channel_async_benchmark" +required-features = ["channel"] + [[example]] name = "tcp" required-features = ["serde"] + +[[example]] +name = "tcp_benchmark" +required-features = ["serde"] diff --git a/README.md b/README.md index 0425b14..dfa1355 100644 --- a/README.md +++ b/README.md @@ -14,7 +14,10 @@ The Server implements read-write locking, there can be many reads at once, but o The Client object cannot be cloned. Instead all function calls are immutable, so a reference can be shared to all. -Currently only implements local thread messaging. Serialized TCP transport is unfinished. +Current status: +- channel_sync : Implementation OK +- channel_async : Implementation OK, but beware of performance bottlenecks. If you are making lots of small calls, channel_sync is more fitting +- tcp : Functional, but server close handling uninmplemented yet Example: ```rs diff --git a/examples/channel_async.rs b/examples/channel_async.rs index fd0b930..0de34d0 100644 --- a/examples/channel_async.rs +++ b/examples/channel_async.rs @@ -16,23 +16,16 @@ async fn main() -> Result<(), Box> { }); let client = TestDataClient::new(c); - let clientref = &client; + + let start_time = std::time::Instant::now(); let job1 = async { let now = std::time::Instant::now(); - assert_eq!(DATASIZE, client.len().await.unwrap()); - println!("len: {}", dur_to_str(now.elapsed())); - - let now = std::time::Instant::now(); - assert_eq!(267914296, clientref.fib(42).await.unwrap()); + assert_eq!(267914296, client.fib(42).await.unwrap()); println!("fib1: {}", dur_to_str(now.elapsed())); }; let job2 = async { - let now = std::time::Instant::now(); - assert_eq!(DATASIZE, client.len().await.unwrap()); - println!("len: {}", dur_to_str(now.elapsed())); - let now = std::time::Instant::now(); assert_eq!(4, client.calc().await.unwrap().unwrap_or(0)); println!("calc: {}", dur_to_str(now.elapsed())); @@ -47,7 +40,11 @@ async fn main() -> Result<(), Box> { println!("fib2: {}", dur_to_str(now.elapsed())); let now = std::time::Instant::now(); - clientref.push((false, TestEnum::NoValue)).await.unwrap(); + assert_eq!(DATASIZE, client.len().await.unwrap()); + println!("len: {}", dur_to_str(now.elapsed())); + + let now = std::time::Instant::now(); + client.push((false, TestEnum::NoValue)).await.unwrap(); println!("push: {}", dur_to_str(now.elapsed())); let now = std::time::Instant::now(); @@ -56,6 +53,9 @@ async fn main() -> Result<(), Box> { }; join!(job1, job2); + + println!("total: {}", dur_to_str(start_time.elapsed())); + client.stop().await.unwrap(); srv_thread.await.unwrap().unwrap(); diff --git a/examples/channel_async_benchmark.rs b/examples/channel_async_benchmark.rs new file mode 100644 index 0000000..8986f6c --- /dev/null +++ b/examples/channel_async_benchmark.rs @@ -0,0 +1,39 @@ +use futures::{stream::FuturesUnordered, StreamExt}; +use rspc_dev_utilities::test_data::{ + dur_to_str, make_test_data, TestData, TestDataClient, TestDataServer, CALLS_PER_THREAD, + DATASIZE, THREADS, +}; + +use rspc::transport; + +#[tokio::main] +async fn main() -> Result<(), Box> { + let (c, s) = transport::channel::new_async(); + let data: TestData = make_test_data(DATASIZE); + + let srv_thread = tokio::spawn(async move { + let mut server = TestDataServer::from(data); + server.listen(s).await + }); + + let client = TestDataClient::new(c); + + let now = std::time::Instant::now(); + { + let set = FuturesUnordered::new(); + for _ in 0..THREADS { + set.push(async { + for _ in 0..CALLS_PER_THREAD { + client.heavy_calc().await.unwrap(); + } + }); + } + let _: Vec<_> = set.collect().await; + } + println!("time: {}", dur_to_str(now.elapsed())); + + client.stop().await.unwrap(); + srv_thread.await.unwrap().unwrap(); + + Ok(()) +} diff --git a/examples/channel_sync.rs b/examples/channel_sync.rs index 3a057e4..fb38b0e 100644 --- a/examples/channel_sync.rs +++ b/examples/channel_sync.rs @@ -1,7 +1,7 @@ +use futures::{stream::FuturesUnordered, StreamExt}; use rspc_dev_utilities::test_data::{ dur_to_str, make_test_data, TestData, TestDataClient, TestDataServer, TestEnum, DATASIZE, }; -use tokio::join; use rspc::transport; @@ -16,46 +16,54 @@ async fn main() -> Result<(), Box> { }); let client = TestDataClient::new(c); - let clientref = &client; - let job1 = async { - let now = std::time::Instant::now(); - assert_eq!(DATASIZE, client.len().await.unwrap()); - println!("len: {}", dur_to_str(now.elapsed())); + let start_time = std::time::Instant::now(); - let now = std::time::Instant::now(); - assert_eq!(267914296, clientref.fib(42).await.unwrap()); - println!("fib1: {}", dur_to_str(now.elapsed())); - }; + let now = std::time::Instant::now(); + assert_eq!(DATASIZE, client.len().await.unwrap()); + println!("len: {}", dur_to_str(now.elapsed())); - let job2 = async { - let now = std::time::Instant::now(); - assert_eq!(DATASIZE, client.len().await.unwrap()); - println!("len: {}", dur_to_str(now.elapsed())); + let now = std::time::Instant::now(); + client.push((false, TestEnum::NoValue)).await.unwrap(); + println!("push: {}", dur_to_str(now.elapsed())); - let now = std::time::Instant::now(); - assert_eq!(4, client.calc().await.unwrap().unwrap_or(0)); - println!("calc: {}", dur_to_str(now.elapsed())); + let now = std::time::Instant::now(); + assert_eq!(DATASIZE + 1, client.len().await.unwrap()); + println!("len: {}", dur_to_str(now.elapsed())); - let cdat = make_test_data(DATASIZE); - let now = std::time::Instant::now(); - assert_eq!(8, client.calc_add(cdat).await.unwrap().unwrap_or(0)); - println!("calc_add: {}", dur_to_str(now.elapsed())); + let now = std::time::Instant::now(); + assert_eq!(267914296, client.fib(42).await.unwrap()); + println!("fib1: {}", dur_to_str(now.elapsed())); - let now = std::time::Instant::now(); - assert_eq!(267914296, client.fib(42).await.unwrap()); - println!("fib2: {}", dur_to_str(now.elapsed())); + let now = std::time::Instant::now(); + assert_eq!(267914296, client.fib(42).await.unwrap()); + println!("fib2: {}", dur_to_str(now.elapsed())); - let now = std::time::Instant::now(); - clientref.push((false, TestEnum::NoValue)).await.unwrap(); - println!("push: {}", dur_to_str(now.elapsed())); + let now = std::time::Instant::now(); + assert_eq!(4, client.calc().await.unwrap().unwrap_or(0)); + println!("calc: {}", dur_to_str(now.elapsed())); - let now = std::time::Instant::now(); - assert_eq!(DATASIZE + 1, client.len().await.unwrap()); - println!("len: {}", dur_to_str(now.elapsed())); - }; + let cdat = make_test_data(DATASIZE); + let now = std::time::Instant::now(); + assert_eq!(8, client.calc_add(cdat).await.unwrap().unwrap_or(0)); + println!("calc_add: {}", dur_to_str(now.elapsed())); + + let now = std::time::Instant::now(); + { + let set = FuturesUnordered::new(); + for _ in 0..128 { + set.push(async { + for _ in 0..100000 { + client.calc().await.unwrap(); + } + }); + } + let _: Vec<_> = set.collect().await; + } + println!("benchmark: {}", dur_to_str(now.elapsed())); + + println!("total: {}", dur_to_str(start_time.elapsed())); - join!(job1, job2); client.stop().await.unwrap(); srv_thread.await.unwrap().unwrap(); diff --git a/examples/channel_sync_benchmark.rs b/examples/channel_sync_benchmark.rs new file mode 100644 index 0000000..ab26eb0 --- /dev/null +++ b/examples/channel_sync_benchmark.rs @@ -0,0 +1,39 @@ +use futures::{stream::FuturesUnordered, StreamExt}; +use rspc_dev_utilities::test_data::{ + dur_to_str, make_test_data, TestData, TestDataClient, TestDataServer, CALLS_PER_THREAD, + DATASIZE, THREADS, +}; + +use rspc::transport; + +#[tokio::main] +async fn main() -> Result<(), Box> { + let (c, s) = transport::channel::new_sync(); + let data: TestData = make_test_data(DATASIZE); + + let srv_thread = tokio::spawn(async move { + let mut server = TestDataServer::from(data); + server.listen(s).await + }); + + let client = TestDataClient::new(c); + + let now = std::time::Instant::now(); + { + let set = FuturesUnordered::new(); + for _ in 0..THREADS { + set.push(async { + for _ in 0..CALLS_PER_THREAD { + client.heavy_calc().await.unwrap(); + } + }); + } + let _: Vec<_> = set.collect().await; + } + println!("time: {}", dur_to_str(now.elapsed())); + + client.stop().await.unwrap(); + srv_thread.await.unwrap().unwrap(); + + Ok(()) +} diff --git a/examples/tcp.rs b/examples/tcp.rs index d1d0c69..2b63100 100644 --- a/examples/tcp.rs +++ b/examples/tcp.rs @@ -1,38 +1,33 @@ +use futures::{stream::FuturesUnordered, StreamExt}; use rspc_dev_utilities::test_data::{ dur_to_str, make_test_data, TestData, TestDataClient, TestDataServer, TestEnum, DATASIZE, }; use tokio::join; -use rspc::transport::serde::TcpClient; - -use rspc::transport; +use rspc::transport::serde::{TcpClient, TcpServer}; #[tokio::main] async fn main() -> Result<(), Box> { + let data: TestData = make_test_data(DATASIZE); + let mut server = TestDataServer::from(data); + let t = TcpServer::new(&"127.0.0.1:6543").await.unwrap(); + let srv_thread = tokio::spawn(async move { server.listen(t).await }); + + tokio::time::sleep(tokio::time::Duration::from_millis(100)).await; + let t = TcpClient::connect("127.0.0.1:6543").await.unwrap(); let client = t.spawn().await; - - // todo : server - let client = TestDataClient::new(client); - let clientref = &client; + let start_time = std::time::Instant::now(); let job1 = async { let now = std::time::Instant::now(); - assert_eq!(DATASIZE, client.len().await.unwrap()); - println!("len: {}", dur_to_str(now.elapsed())); - - let now = std::time::Instant::now(); - assert_eq!(267914296, clientref.fib(42).await.unwrap()); + assert_eq!(267914296, client.fib(42).await.unwrap()); println!("fib1: {}", dur_to_str(now.elapsed())); }; let job2 = async { - let now = std::time::Instant::now(); - assert_eq!(DATASIZE, client.len().await.unwrap()); - println!("len: {}", dur_to_str(now.elapsed())); - let now = std::time::Instant::now(); assert_eq!(4, client.calc().await.unwrap().unwrap_or(0)); println!("calc: {}", dur_to_str(now.elapsed())); @@ -47,7 +42,11 @@ async fn main() -> Result<(), Box> { println!("fib2: {}", dur_to_str(now.elapsed())); let now = std::time::Instant::now(); - clientref.push((false, TestEnum::NoValue)).await.unwrap(); + assert_eq!(DATASIZE, client.len().await.unwrap()); + println!("len: {}", dur_to_str(now.elapsed())); + + let now = std::time::Instant::now(); + client.push((false, TestEnum::NoValue)).await.unwrap(); println!("push: {}", dur_to_str(now.elapsed())); let now = std::time::Instant::now(); @@ -56,7 +55,27 @@ async fn main() -> Result<(), Box> { }; join!(job1, job2); + + let now = std::time::Instant::now(); + { + let set = FuturesUnordered::new(); + for _ in 0..128 { + set.push(async { + for _ in 0..100000 { + client.calc().await.unwrap(); + } + }); + } + let _: Vec<_> = set.collect().await; + } + println!("benchmark: {}", dur_to_str(now.elapsed())); + + println!("total: {}", dur_to_str(start_time.elapsed())); + client.stop().await.unwrap(); + // todo : server stop + srv_thread.await; + Ok(()) } diff --git a/examples/tcp_benchmark.rs b/examples/tcp_benchmark.rs new file mode 100644 index 0000000..f840850 --- /dev/null +++ b/examples/tcp_benchmark.rs @@ -0,0 +1,40 @@ +use futures::{stream::FuturesUnordered, StreamExt}; +use rspc_dev_utilities::test_data::{ + dur_to_str, make_test_data, TestData, TestDataClient, TestDataServer, CALLS_PER_THREAD, + DATASIZE, THREADS, +}; + +use rspc::transport::serde::{TcpClient, TcpServer}; + +#[tokio::main] +async fn main() -> Result<(), Box> { + let data: TestData = make_test_data(DATASIZE); + let mut server = TestDataServer::from(data); + let t = TcpServer::new(&"127.0.0.1:6543").await.unwrap(); + let srv_thread = tokio::spawn(async move { server.listen(t).await }); + + tokio::time::sleep(tokio::time::Duration::from_millis(10)).await; + + let t = TcpClient::connect("127.0.0.1:6543").await.unwrap(); + let client = t.spawn().await; + let client = TestDataClient::new(client); + + let now = std::time::Instant::now(); + { + let set = FuturesUnordered::new(); + for _ in 0..THREADS { + set.push(async { + for _ in 0..CALLS_PER_THREAD { + client.heavy_calc().await.unwrap(); + } + }); + } + let _: Vec<_> = set.collect().await; + } + println!("time: {}", dur_to_str(now.elapsed())); + + client.stop().await.unwrap(); + srv_thread.await.unwrap().unwrap(); + + Ok(()) +} diff --git a/macros/src/lib.rs b/macros/src/lib.rs index f4b6690..c538b3c 100644 --- a/macros/src/lib.rs +++ b/macros/src/lib.rs @@ -316,7 +316,7 @@ pub fn service(_attr: TokenStream, mut input: TokenStream) -> TokenStream { impl #server { pub async fn listen(&mut self, mut transport: Tr) -> Result<(), Tr::Error> where - Tr: ServerTransporter<#transport_request,#transport_response> + Send + Tr: ::rspc::transport::ServerTransporter<#transport_request,#transport_response> + Send { { transport.listen( |v,obj| { @@ -339,14 +339,14 @@ pub fn service(_attr: TokenStream, mut input: TokenStream) -> TokenStream { pub struct #client where - Tr: ClientTransporter<#transport_request,#transport_response>, + Tr: ::rspc::transport::ClientTransporter<#transport_request,#transport_response>, { transporter: Tr, } impl #client where - Tr: ClientTransporter<#transport_request,#transport_response> + Tr: ::rspc::transport::ClientTransporter<#transport_request,#transport_response> { pub fn new(transporter: Tr) -> Self { #client { diff --git a/src/transport/mod.rs b/src/transport/mod.rs index ad43acf..c2b919b 100644 --- a/src/transport/mod.rs +++ b/src/transport/mod.rs @@ -37,14 +37,14 @@ pub trait ServerTransporter { } pub async fn async_listener( - listener: &mut L, + mut listener: L, listener_recv: LF, - sender: &mut S, + mut sender: S, sender_send: SF, - handler: F, - stop_response: Option, + handler: &F, + stop_response: &Option, userdata: &D, -) -> Result<(), E> +) -> Result where T: Send + Sync, R: Send + Sync + 'static, @@ -55,10 +55,11 @@ where SF: Fn(&mut S, (C, R)) -> BoxFuture> + Send + Sync + 'static, LF: Fn(&mut L) -> BoxFuture, E>> + 'static, { + let mut stop = false; let mut pending = FuturesUnordered::new(); loop { tokio::select! { - rcv = listener_recv(listener) => { + rcv = listener_recv(&mut listener) => { match rcv? { Some((id, data)) => { pending.push( @@ -76,12 +77,14 @@ where Some(r) = pending.next() => { match r { (id,Some(r)) => { - sender_send(sender, (id,r)).await?; + sender_send(&mut sender, (id,r)).await?; }, (id,None) => { if let Some(v) = stop_response { - sender_send(sender, (id,v)).await?; + todo!() // clone issue on R + // sender_send(&mut sender, (id,v)).await?; } + stop = true; break; }, } @@ -94,11 +97,11 @@ where for it in results { match it { (id, Some(r)) => { - sender_send(sender, (id, r)).await?; + sender_send(&mut sender, (id, r)).await?; } _ => (), } } - Ok(()) + Ok(stop) } diff --git a/src/transport/serde.rs b/src/transport/serde.rs index 1ac9f3a..6aea2d7 100644 --- a/src/transport/serde.rs +++ b/src/transport/serde.rs @@ -1,7 +1,7 @@ use futures::future::BoxFuture; use futures::prelude::*; +use futures::stream::FuturesUnordered; use serde::{Deserialize, Serialize}; -use std::net::Ipv4Addr; use std::sync::atomic::AtomicUsize; use thiserror::Error; use tokio::net::{TcpListener, TcpStream}; @@ -115,6 +115,8 @@ where }) } + /// Turn the TCP Client into a background process and a channel client multiplexer + #[cfg(feature = "channel")] pub async fn multiplex( self, ) -> ( @@ -133,6 +135,8 @@ where (client, Box::pin(fut)) } + /// Spawn a background client multiplexer and give a channel client + #[cfg(feature = "channel")] pub async fn spawn(self) -> ChannelClient { let (client, job) = self.multiplex().await; tokio::spawn(job); @@ -140,14 +144,28 @@ where } } +impl ClientTransporter for TcpClient +where + T: for<'a> Deserialize<'a> + Serialize + Send + Sync + Unpin + 'static, + R: for<'a> Deserialize<'a> + Serialize + Send + Sync + Unpin + 'static, +{ + type Error = Error; + + async fn request(&self, data: T) -> Result { + todo!() + } +} + impl TcpServer where T: for<'a> Deserialize<'a> + Serialize, R: for<'a> Deserialize<'a> + Serialize, { - pub async fn new(address: &Ipv4Addr, port: u16) -> Result, Error> { - let address = format!("{}:{}", address, port); - let listener = TcpListener::bind(&address).await.map_err(Error::IO)?; + pub async fn new(address: &A) -> Result, Error> + where + A: tokio::net::ToSocketAddrs + std::fmt::Display, + { + let listener = TcpListener::bind(address).await.map_err(Error::IO)?; Ok(TcpServer { listener, ghost: Default::default(), @@ -156,15 +174,14 @@ where async fn accept(&mut self) -> Result { let (socket, address) = self.listener.accept().await.map_err(Error::IO)?; - println!("connection accepted: {:?}", address); Ok(socket) } } impl ServerTransporter for TcpServer where - T: for<'a> Deserialize<'a> + Serialize + Send + Sync + Unpin, - R: for<'a> Deserialize<'a> + Serialize + Send + Sync + Unpin, + T: for<'a> Deserialize<'a> + Serialize + Send + Sync + Unpin + 'static, + R: for<'a> Deserialize<'a> + Serialize + Send + Sync + Unpin + 'static, { type Error = Error; @@ -176,34 +193,45 @@ where ) -> Result<(), Self::Error> where FR: Future> + Send + 'static, - F: Fn(T, &D) -> FR + Send + Sync, - D: Send + Sync, + F: Fn(T, &D) -> FR + Send + Sync + 'static, + D: Send + Sync + 'static, { - let (client, fut) = channel::new_multiplexer::(); - - // super::async_listener( - // &mut receiver, |_self| { Box::pin(async { - // _self.recv().await - // }) }, - // &mut sender, |_self, data| { Box::pin(async { - // _self.send(data).await - // }) }, - // handler, stop_response, &userdata); - - while let Ok(mut stream) = self.accept().await { - let (sender, receiver) = split::(stream).await; - // tokio::spawn(async move { - // super::async_listener( - // &mut receiver, |_self| { Box::pin(async { - // _self.recv().await - // }) }, - // &mut sender, |_self, data| { Box::pin(async { - // _self.send(data).await - // }) }, - // handler, stop_response, &userdata) - // }); + let mut connections = FuturesUnordered::new(); + loop { + tokio::select! { + connect = self.accept() => { + match connect { + Ok(stream) => { + let (sender, receiver) = split::(stream).await; + let handler = &handler; + let stop_response = &stop_response; + let userdata = &userdata; + connections.push(async move { + super::async_listener( + receiver, + |_self| Box::pin(async { _self.recv().await }), + sender, + |_self, data| Box::pin(async { _self.send(data).await }), + handler, + stop_response, + userdata, + ).await + }); + }, + Err(e) => { + println!("WARN: connection accept error: {e}"); + } + } + }, + Some(s) = connections.next() => { + if let Ok(true) = s { + break; + } + // poll connected threads to get them running + } + } } - todo!() + Ok(()) } } diff --git a/tests/tcp.rs b/tests/tcp.rs new file mode 100644 index 0000000..342e4a8 --- /dev/null +++ b/tests/tcp.rs @@ -0,0 +1,46 @@ +#[cfg(test)] +#[cfg(feature = "serde")] +mod tests { + // use rspc::transport::{channel, ClientTransporter, ServerTransporter}; + + pub struct MyStruct { + my_vec: Vec, + } + + #[rspc::service] + impl MyStruct { + pub fn len(&self) -> usize { + self.my_vec.len() + } + + pub fn push(&mut self, val: String) { + self.my_vec.push(val) + } + + pub fn pop(&mut self) -> Option { + self.my_vec.pop() + } + } + + #[tokio::test] + async fn test() { + // let my_data = MyStruct { my_vec: Vec::new() }; + // let (c, s) = channel::new_sync(); + + // let srv_thread = tokio::spawn(async move { + // let mut server = MyStructServer::from(my_data); + // server.listen(s).await + // }); + + // let client = MyStructClient::new(c); + // assert_eq!(client.len().await.unwrap(), 0); + // client.push("Hello world!".to_string()).await.unwrap(); + // assert_eq!(client.len().await.unwrap(), 1); + // assert_eq!( + // client.pop().await.unwrap(), + // Some("Hello world!".to_string()) + // ); + // client.stop().await.unwrap(); + // srv_thread.await.unwrap().unwrap(); + } +} diff --git a/utilities/src/test_data.rs b/utilities/src/test_data.rs index ea5ce4e..ae3701a 100644 --- a/utilities/src/test_data.rs +++ b/utilities/src/test_data.rs @@ -1,7 +1,5 @@ use std::time::Duration; -use rspc::transport::{ClientTransporter, ServerTransporter}; - use serde::{Deserialize, Serialize}; pub fn dur_to_num(dur: Duration) -> (u128, &'static str) { @@ -21,7 +19,9 @@ pub fn dur_to_str(dur: Duration) -> String { n.to_string() + " " + s } -pub const DATASIZE: usize = 1000000; +pub const DATASIZE: usize = 10000; +pub const THREADS: usize = 8; +pub const CALLS_PER_THREAD: usize = 1000; const TEST_STRINGS: [&str; 4] = ["toto", "tata", "titi", "tutu"]; @@ -39,6 +39,10 @@ pub struct TestData { #[rspc::service] impl TestData { + pub fn mem_size(&self) -> usize { + std::mem::size_of_val(&self.vec) + } + pub fn len(&self) -> usize { self.vec.len() } @@ -52,6 +56,16 @@ impl TestData { return None; } + pub fn heavy_calc(&self) -> usize { + let mut res = 0; + for v in &self.vec { + if let (true, TestEnum::Num(n)) = v { + res += n; + } + } + return res; + } + pub async fn slow_fct(&self) -> Option { tokio::time::sleep(Duration::from_secs(1)).await; self.calc()