zawz/rspc
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat: tcp transport

Browse source 
- add tcp transport
- refactor examples
- improve proc macro
- add benchmark examples
This commit is contained in:
zawz 2026-07-04 11:14:25 +02:00
parent b619f79095
commit bc23b1a9db
13 changed files with 363 additions and 110 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

14
Cargo.toml
View file

@ -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"]

5
README.md
View file

@ -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

22
examples/channel_async.rs
View file

@ -16,23 +16,16 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
});
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<dyn std::error::Error>> {
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<dyn std::error::Error>> {
};
join!(job1, job2);
println!("total: {}", dur_to_str(start_time.elapsed()));
client.stop().await.unwrap();
srv_thread.await.unwrap().unwrap();

39
examples/channel_async_benchmark.rs Normal file
View file

@ -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<dyn std::error::Error>> {
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(())
}

72
examples/channel_sync.rs
View file

@ -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<dyn std::error::Error>> {
});
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();

39
examples/channel_sync_benchmark.rs Normal file
View file

@ -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<dyn std::error::Error>> {
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(())
}

53
examples/tcp.rs
View file

@ -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<dyn std::error::Error>> {
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<dyn std::error::Error>> {
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<dyn std::error::Error>> {
};
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(())
}

40
examples/tcp_benchmark.rs Normal file
View file

@ -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<dyn std::error::Error>> {
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(())
}

6
macros/src/lib.rs
View file

@ -316,7 +316,7 @@ pub fn service(_attr: TokenStream, mut input: TokenStream) -> TokenStream {
impl #server {
pub async fn listen<Tr>(&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<Tr>
where
Tr: ClientTransporter<#transport_request,#transport_response>,
Tr: ::rspc::transport::ClientTransporter<#transport_request,#transport_response>,
{
transporter: Tr,
}
impl<Tr> #client<Tr>
where
Tr: ClientTransporter<#transport_request,#transport_response>
Tr: ::rspc::transport::ClientTransporter<#transport_request,#transport_response>
{
pub fn new(transporter: Tr) -> Self {
#client {

23
src/transport/mod.rs
View file

@ -37,14 +37,14 @@ pub trait ServerTransporter<T, R> {
}
pub async fn async_listener<T, R, C, L, LF, S, SF, F, FR, D, E>(
listener: &mut L,
mut listener: L,
listener_recv: LF,
sender: &mut S,
mut sender: S,
sender_send: SF,
handler: F,
stop_response: Option<R>,
handler: &F,
stop_response: &Option<R>,
userdata: &D,
) -> Result<(), E>
) -> Result<bool, E>
where
T: Send + Sync,
R: Send + Sync + 'static,
@ -55,10 +55,11 @@ where
SF: Fn(&mut S, (C, R)) -> BoxFuture<Result<(), E>> + Send + Sync + 'static,
LF: Fn(&mut L) -> BoxFuture<Result<Option<(C, T)>, 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)
}

94
src/transport/serde.rs
View file

@ -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<T, R> {
let (client, job) = self.multiplex().await;
tokio::spawn(job);
@ -140,14 +144,28 @@ where
}
}
impl<T, R> ClientTransporter<T, R> for TcpClient<T, R>
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<R, Self::Error> {
todo!()
}
}
impl<T, R> TcpServer<T, R>
where
T: for<'a> Deserialize<'a> + Serialize,
R: for<'a> Deserialize<'a> + Serialize,
{
pub async fn new(address: &Ipv4Addr, port: u16) -> Result<TcpServer<T, R>, Error> {
let address = format!("{}:{}", address, port);
let listener = TcpListener::bind(&address).await.map_err(Error::IO)?;
pub async fn new<A>(address: &A) -> Result<TcpServer<T, R>, 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<TcpStream, Error> {
let (socket, address) = self.listener.accept().await.map_err(Error::IO)?;
println!("connection accepted: {:?}", address);
Ok(socket)
}
}
impl<T, R> ServerTransporter<T, R> for TcpServer<T, R>
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<Output = Option<R>> + 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::<R, T>();
// 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::<R, T>(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::<R, T>(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(())
}
}

46
tests/tcp.rs Normal file
View file

@ -0,0 +1,46 @@
#[cfg(test)]
#[cfg(feature = "serde")]
mod tests {
// use rspc::transport::{channel, ClientTransporter, ServerTransporter};
pub struct MyStruct {
my_vec: Vec<String>,
}
#[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<String> {
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();
}
}

20
utilities/src/test_data.rs
View file

@ -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<usize> {
tokio::time::sleep(Duration::from_secs(1)).await;
self.calc()