Put RT as associate type of ClientTransport & ServerTransport

Let FailoverPool store the runtime handle (so that we can reduce one
generic parameter). update_addr does not need rt argument as well.

Author: frostyplanet

docs/api_design.md

@@ -55,7 +55,7 @@ A Service in `razor-rpc` follows these principles:
 - Client and server share the same trait definition for compile-time checks
 - Compatible with GRPC naming conventions (`service` in PascalCase, `method` in snake_case)
 - Methods should be `async fn` or return `impl Future`
-- Methods can return **custom error type**. All method should return `Result<T, RpcError<E>>` where `E: RpcErrCodec`, refer to doc: [error module](crate::error). 
+- Methods can return **custom error type**. All method should return `Result<T, RpcError<E>>` where `E: RpcErrCodec`, refer to doc: [error module](crate::error).
 
 We supports rust 1.75 `AFIT` (Async fn in Traits) `RPITIT` (Return Position Impl Trait in Traits), and legacy `#[async_trait]`.
 
@@ -198,7 +198,7 @@ async fn use_client(server_addr: &str) {
     let rt = RT::new_multi_thread(8);
     let factory = APIFact::<Codec>::new(client_config);
     // 9. Create client connection pool
-    let pool: APIConnPool<Codec, ClientProto> = factory.new_conn_pool::<ClientProto, RT>(&rt, server_addr);
+    let pool: APIConnPool<Codec, ClientProto> = factory.new_conn_pool::<ClientProto>(&rt, server_addr);
     let client = CalculatorClient::new(pool);
     //  You will have to import CalculatorService trait to call its methods
     use CalculatorService;
@@ -282,9 +282,9 @@ endpoint_client!(KVClient);
 #[endpoint_async(KVClient)]
 pub trait KVService {
     // Note: endpoint_async macro requires exactly one parameter besides &self
-    fn put(&self, kv: (String, String)) 
+    fn put(&self, kv: (String, String))
         -> impl Future<Output = Result<(), RpcError<ClusterErr>>> + Send;
-    fn get(&self, key: String) 
+    fn get(&self, key: String)
         -> impl Future<Output = Result<Option<String>, RpcError<String>>> + Send;
 }
 
@@ -295,13 +295,13 @@ type FailoverCaller = razor_rpc::client::APIFailoverPool<Codec, TcpClient<RT>>;
 
 impl KVClient<FailoverCaller> {
     pub fn new_cluster_client(
-        config: ClientConfig, 
-        addrs: Vec<String>, 
+        config: ClientConfig,
+        addrs: Vec<String>,
         rt: &RT
     ) -> Self {
         let fact = APIFact::<Codec>::new(config);
         // stateless=false: maintain leader affinity for stateful service
-        let pool = fact.new_failover::<TcpClient<RT>, RT>(rt, addrs, false, 3);
+        let pool = fact.new_failover::<TcpClient<RT>>(rt, addrs, false, 3);
         KVClient::new(pool)
     }
 }
@@ -315,12 +315,12 @@ async fn example() {
         "127.0.0.1:8081".to_string(),
         "127.0.0.1:8082".to_string(),
     ];
-    
+
     let client = KVClient::new_cluster_client(config, addrs, &rt);
-    
+
     // Write goes to leader (with automatic redirect if needed)
     client.put(("key1".to_string(), "value1".to_string())).await.unwrap();
-    
+
     // Read can go to any node
     let value = client.get("key1".to_string()).await.unwrap();
 }

src/client/mod.rs

@@ -1,7 +1,6 @@
 pub mod task;
 use captains_log::filter::LogFilter;
 use crossfire::oneshot::oneshot;
-use orb::AsyncRuntime;
 pub use razor_rpc_macros::{endpoint_async, endpoint_client};
 pub use razor_stream::client::ClientCaller;
 pub use task::*;
@@ -33,16 +32,16 @@ impl<C: Codec> APIFact<C> {
         self.logger.set_level(level);
     }
 
-    pub fn new_conn_pool<P: ClientTransport, RT: AsyncRuntime + Clone>(
-        self: Arc<Self>, rt: &RT, addr: &str,
+    pub fn new_conn_pool<P: ClientTransport>(
+        self: Arc<Self>, rt: &P::RT, addr: &str,
     ) -> APIConnPool<C, P> {
-        ConnPool::<APIFact<C>, P>::new::<RT>(self.clone(), rt, addr, 0)
+        ConnPool::<APIFact<C>, P>::new(self.clone(), rt, addr, 0)
     }
 
-    pub fn new_failover<P: ClientTransport, RT: AsyncRuntime + Clone>(
-        self: Arc<Self>, rt: &RT, addrs: Vec<String>, stateless: bool, retry_limit: usize,
+    pub fn new_failover<P: ClientTransport>(
+        self: Arc<Self>, rt: &P::RT, addrs: Vec<String>, stateless: bool, retry_limit: usize,
     ) -> APIFailoverPool<C, P> {
-        FailoverPool::<APIFact<C>, P>::new::<RT>(self.clone(), rt, addrs, stateless, retry_limit, 0)
+        FailoverPool::<APIFact<C>, P>::new(self.clone(), rt, addrs, stateless, retry_limit, 0)
     }
 }
 

stream/src/client/failover.rs

@@ -11,7 +11,7 @@ use ahash::AHashMap;
 use arc_swap::ArcSwap;
 use captains_log::filter::LogFilter;
 use crossfire::{AsyncRx, MTx, SendError, mpsc};
-use orb::AsyncRuntime;
+use orb::prelude::AsyncExec;
 use std::fmt;
 use std::sync::{
     Arc, Weak,
@@ -57,6 +57,7 @@ where
     next_node: AtomicUsize,
     pool_channel_size: usize,
     facts: Arc<FailoverFacts<F>>,
+    rt: P::RT,
 }
 
 struct ClusterConfig<F, P>
@@ -76,8 +77,8 @@ where
     /// Initiate the pool with multiple addresses.
     /// When stateless == true, all addresses in the pool will be selected with equal chance (round-robin);
     /// When stateless == false, the leader address will always be picked unless error happens.
-    pub fn new<RT: AsyncRuntime + Clone>(
-        facts: Arc<F>, rt: &RT, addrs: Vec<String>, stateless: bool, retry_limit: usize,
+    pub fn new(
+        facts: Arc<F>, rt: &P::RT, addrs: Vec<String>, stateless: bool, retry_limit: usize,
         pool_channel_size: usize,
     ) -> Self {
         let (retry_tx, retry_rx) = mpsc::unbounded_async();
@@ -86,7 +87,7 @@ where
             Arc::new(FailoverFacts { retry_limit, retry_tx, logger: facts.new_logger(), facts });
         let mut pools = Vec::with_capacity(addrs.len());
         for addr in addrs.iter() {
-            let pool = ConnPool::new::<RT>(wrapped_facts.clone(), rt, addr, pool_channel_size);
+            let pool = ConnPool::new(wrapped_facts.clone(), rt, addr, pool_channel_size);
             pools.push(pool);
         }
         // NOTE: the ConnPool has cycle reference with FailoverPoolInner
@@ -97,6 +98,7 @@ where
             ver: AtomicU64::new(1),
             next_node: AtomicUsize::new(0),
             pool_channel_size,
+            rt: rt.clone(),
         });
         let weak_self = Arc::downgrade(&inner);
         rt.spawn_detach(async move {
@@ -174,7 +176,7 @@ where
         }
     }
 
-    pub fn update_addrs<RT: AsyncRuntime + Clone>(&self, addrs: Vec<String>, rt: &RT) {
+    pub fn update_addrs(&self, addrs: Vec<String>) {
         let inner = &self.inner;
         let old_pools = inner.pools.load_full();
         let mut new_pools: Vec<ConnPool<FailoverFacts<F>, P>> = Vec::with_capacity(addrs.len());
@@ -190,7 +192,7 @@ where
             } else {
                 // Create a new pool for the new address
                 let new_pool =
-                    ConnPool::new::<RT>(inner.facts.clone(), rt, &addr, inner.pool_channel_size);
+                    ConnPool::new(inner.facts.clone(), &inner.rt, &addr, inner.pool_channel_size);
                 new_pools.push(new_pool);
             }
         }

stream/src/client/mod.rs

@@ -3,6 +3,7 @@
 use crate::{Codec, error::RpcIntErr};
 use captains_log::filter::LogFilter;
 use crossfire::{AsyncRx, mpsc};
+use orb::AsyncRuntime;
 use std::future::Future;
 use std::sync::Arc;
 use std::time::{Duration, SystemTime, UNIX_EPOCH};
@@ -146,6 +147,8 @@ impl<C: ClientCallerBlocking + Send + Sync> ClientCallerBlocking for Arc<C> {
 ///
 /// - [razor-rpc-tcp](https://docs.rs/razor-rpc-tcp): For TCP and Unix socket
 pub trait ClientTransport: fmt::Debug + Send + Sized + 'static {
+    type RT: AsyncRuntime + Clone;
+
     /// How to establish an async connection.
     ///
     /// conn_id: used for log fmt, can by the same of addr.

stream/src/client/pool.rs

@@ -5,7 +5,7 @@ use crate::client::{
 use crate::error::RpcIntErr;
 use captains_log::filter::LogFilter;
 use crossfire::{MAsyncRx, MAsyncTx, MTx, RecvTimeoutError, mpmc};
-use orb::AsyncRuntime;
+use orb::prelude::{AsyncExec, AsyncTime};
 use std::fmt;
 use std::marker::PhantomData;
 use std::sync::Arc;
@@ -65,9 +65,7 @@ struct ConnPoolInner<F: ClientFacts, P: ClientTransport> {
 const ONE_SEC: Duration = Duration::from_secs(1);
 
 impl<F: ClientFacts, P: ClientTransport> ConnPool<F, P> {
-    pub fn new<RT: AsyncRuntime + Clone>(
-        facts: Arc<F>, rt: &RT, addr: &str, mut channel_size: usize,
-    ) -> Self {
+    pub fn new(facts: Arc<F>, rt: &P::RT, addr: &str, mut channel_size: usize) -> Self {
         let config = facts.get_config();
         if config.thresholds > 0 {
             if channel_size < config.thresholds {
@@ -91,7 +89,7 @@ impl<F: ClientFacts, P: ClientTransport> ConnPool<F, P> {
             _phan: Default::default(),
         });
         let s = Self { tx_async, tx, inner };
-        s.spawn::<RT>(rt);
+        s.spawn(rt);
         s
     }
 
@@ -118,7 +116,7 @@ impl<F: ClientFacts, P: ClientTransport> ConnPool<F, P> {
     /// by default there's one worker thread after initiation, but you can pre-spawn more thread if
     /// the connection is not enough to achieve desired throughput.
     #[inline]
-    pub fn spawn<RT: AsyncRuntime + Clone>(&self, rt: &RT) {
+    pub fn spawn(&self, rt: &P::RT) {
         let worker_id = self.inner.worker_count.fetch_add(1, Acquire);
         self.inner.clone().spawn_worker(rt, worker_id);
     }
@@ -155,7 +153,7 @@ impl<F: ClientFacts, P: ClientTransport> fmt::Display for ConnPoolInner<F, P> {
 }
 
 impl<F: ClientFacts, P: ClientTransport> ConnPoolInner<F, P> {
-    fn spawn_worker<RT: AsyncRuntime + Clone>(self: Arc<Self>, rt: &RT, worker_id: usize) {
+    fn spawn_worker(self: Arc<Self>, rt: &P::RT, worker_id: usize) {
         let _rt = rt.clone();
         rt.spawn_detach(async move {
             logger_trace!(&self.logger, "{} worker_id={} running", self, worker_id);
@@ -181,7 +179,7 @@ impl<F: ClientFacts, P: ClientTransport> ConnPoolInner<F, P> {
     }
 
     #[inline]
-    async fn connect<RT: AsyncRuntime>(&self, rt: &RT) -> Result<ClientStream<F, P>, RpcIntErr> {
+    async fn connect(&self, rt: &P::RT) -> Result<ClientStream<F, P>, RpcIntErr> {
         ClientStream::connect(self.facts.clone(), rt, &self.addr, &self.conn_id, None).await
     }
 
@@ -220,23 +218,27 @@ impl<F: ClientFacts, P: ClientTransport> ConnPoolInner<F, P> {
     /// connection attempts happens after we spawn.
     /// If the address is dead, the thread might exit after multiple attempts, and later re-spawn
     /// when the needs arrives.
-    async fn run<RT: AsyncRuntime + Clone>(self: &Arc<Self>, rt: RT, mut worker_id: usize) {
+    async fn run(self: &Arc<Self>, rt: P::RT, mut worker_id: usize) {
         'CONN_LOOP: loop {
-            match self.connect::<RT>(&rt).await {
+            match self.connect(&rt).await {
                 Ok(mut stream) => {
                     logger_trace!(self.logger, "{} worker={} connected", self, worker_id);
                     if worker_id == 0 {
                         // act as monitor
                         'MONITOR: loop {
                             if self.get_workers() > 1 {
-                                RT::sleep(ONE_SEC).await;
+                                <P::RT as AsyncTime>::sleep(ONE_SEC).await;
                                 if stream.ping().await.is_err() {
                                     self.set_err();
                                     // don't cleanup the channel unless only one worker left
                                     continue 'CONN_LOOP;
                                 }
                             } else {
-                                match self.rx.recv_with_timer(RT::sleep(ONE_SEC)).await {
+                                match self
+                                    .rx
+                                    .recv_with_timer(<P::RT as AsyncTime>::sleep(ONE_SEC))
+                                    .await
+                                {
                                     Err(RecvTimeoutError::Disconnected) => {
                                         return;
                                     }
@@ -258,13 +260,13 @@ impl<F: ClientFacts, P: ClientTransport> ConnPoolInner<F, P> {
                                             // there's might be a lag to connect,
                                             // so we are spawning identity with new worker,
                                             worker_id = 1;
-                                            self.clone().spawn_worker::<RT>(&rt, 0);
+                                            self.clone().spawn_worker(&rt, 0);
                                         }
                                         if stream.send_task(task, true).await.is_err() {
                                             self.set_err();
                                             if worker_id == 0 {
                                                 self.cleanup();
-                                                RT::sleep(ONE_SEC).await;
+                                                <P::RT as AsyncTime>::sleep(ONE_SEC).await;
                                                 continue 'CONN_LOOP;
                                             } else {
                                                 return;
@@ -297,7 +299,7 @@ impl<F: ClientFacts, P: ClientTransport> ConnPoolInner<F, P> {
                     self.set_err();
                     error!("connect failed to {}: {}", self.addr, e);
                     self.cleanup();
-                    RT::sleep(ONE_SEC).await;
+                    <P::RT as AsyncTime>::sleep(ONE_SEC).await;
                 }
             }
         }

stream/src/server/mod.rs

@@ -65,6 +65,8 @@ pub trait ServerFacts: Sync + Send + 'static + Sized {
 ///
 /// - [razor-rpc-tcp](https://docs.rs/razor-rpc-tcp): For TCP and Unix socket
 pub trait ServerTransport: Send + Sync + Sized + 'static + fmt::Debug {
+    type RT: AsyncRuntime + Clone;
+
     type Listener: AsyncListener;
 
     fn bind(addr: &str) -> impl Future<Output = io::Result<Self::Listener>> + Send;

test-suite/src/api/basic/test_failover_redirect.rs

@@ -1,4 +1,4 @@
-use orb::prelude::{AsyncExec, AsyncTime};
+use orb::prelude::AsyncExec;
 use parking_lot::Mutex;
 use razor_rpc::client::{APIFact, ClientConfig, endpoint_async, endpoint_client};
 use razor_rpc::error::{EncodedErr, RpcErrCodec, RpcError};
@@ -13,7 +13,6 @@ use std::collections::HashMap;
 use std::fmt;
 use std::future::Future;
 use std::sync::Arc;
-use std::sync::atomic::{AtomicUsize, Ordering};
 
 use crate::{TestRunner, logfn, runner};
 
@@ -179,7 +178,7 @@ pub type FailoverCaller = razor_rpc::client::APIFailoverPool<MsgpCodec, TcpClien
 impl DataClient<FailoverCaller> {
     pub fn new_failover_client(config: ClientConfig, addrs: Vec<String>, rt: &crate::RT) -> Self {
         let fact = APIFact::<MsgpCodec>::new(config);
-        let pool = fact.new_failover::<TcpClient<crate::RT>, crate::RT>(rt, addrs, false, 3);
+        let pool = fact.new_failover::<TcpClient<crate::RT>>(rt, addrs, false, 3);
         DataClient::new(pool)
     }
 }

test-suite/src/api/basic/test_multi_service.rs

@@ -38,7 +38,7 @@ pub type PoolCaller = APIConnPool<MsgpCodec, TcpClient<crate::RT>>;
 impl MyClient<PoolCaller> {
     pub fn new_client(config: ClientConfig, addr: &str, rt: &crate::RT) -> Self {
         let facts = APIFact::<MsgpCodec>::new(config);
-        let pool = facts.new_conn_pool::<TcpClient<crate::RT>, crate::RT>(rt, addr);
+        let pool = facts.new_conn_pool::<TcpClient<crate::RT>>(rt, addr);
         MyClient::new(pool)
     }
 }

test-suite/src/stream/client.rs

@@ -165,7 +165,7 @@ pub async fn init_failover_client(
     // since client may be drop by test logic, it's not allow
     // to drop a tokio runtime inside async code.
     let facts = MyClient::new(config);
-    FailoverPool::new::<crate::RT>(
+    FailoverPool::new(
         facts,
         rt,
         addrs,

transport/tcp/src/client.rs

@@ -203,7 +203,9 @@ impl<RT: AsyncRuntime> TcpClient<RT> {
     }
 }
 
-impl<RT: AsyncRuntime> ClientTransport for TcpClient<RT> {
+impl<RT: AsyncRuntime + Clone> ClientTransport for TcpClient<RT> {
+    type RT = RT;
+
     async fn connect(addr: &str, conn_id: &str, config: &ClientConfig) -> Result<Self, RpcIntErr> {
         let stream: UnifyStream<RT> =
             match UnifyStream::<RT>::connect_timeout(addr, config.connect_timeout).await {