From 5c2f4dba322b28e8132b21acfa184309de791d12 Mon Sep 17 00:00:00 2001 From: Lucio Franco Date: Thu, 31 Oct 2019 14:09:40 -0400 Subject: [PATCH] feat(transport): Change channel connect to be async (#107) This makes it so you can check if the initial connection is established. Before this we used reconnect which would lazily attempt to connect. So if you were trying to connect to a non existant Server you wouldn't find out until after you attempted your first RPC. This simplifies everything by allowing you connect before creating the RPC client. BREAKING CHANGE: `Endpoint::channel` was removed in favor of an async `Endpoint::connect`. --- tonic-build/src/client.rs | 5 +- tonic-examples/src/authentication/client.rs | 3 +- tonic-examples/src/gcp/client.rs | 3 +- tonic-examples/src/helloworld/client.rs | 2 +- tonic-examples/src/multiplex/client.rs | 4 +- tonic-examples/src/routeguide/client.rs | 2 +- tonic-examples/src/tls/client.rs | 3 +- tonic-examples/src/tls_client_auth/client.rs | 4 +- tonic-interop/src/bin/client.rs | 2 +- tonic/Cargo.toml | 2 - .../benchmarks/compiled_protos/helloworld.rs | 5 +- tonic/src/transport/channel.rs | 10 +- tonic/src/transport/endpoint.rs | 4 +- tonic/src/transport/mod.rs | 3 +- tonic/src/transport/service/connection.rs | 13 +- tonic/src/transport/service/discover.rs | 41 ++++- tonic/src/transport/service/mod.rs | 1 + tonic/src/transport/service/reconnect.rs | 174 ++++++++++++++++++ 18 files changed, 243 insertions(+), 38 deletions(-) create mode 100644 tonic/src/transport/service/reconnect.rs diff --git a/tonic-build/src/client.rs b/tonic-build/src/client.rs index e01da60..e0a6889 100644 --- a/tonic-build/src/client.rs +++ b/tonic-build/src/client.rs @@ -54,12 +54,13 @@ fn generate_connect(service_ident: &syn::Ident) -> TokenStream { quote! { impl #service_ident { /// Attempt to create a new client by connecting to a given endpoint. - pub fn connect(dst: D) -> Result + pub async fn connect(dst: D) -> Result where D: std::convert::TryInto, D::Error: Into, { - tonic::transport::Endpoint::new(dst).map(|c| Self::new(c.channel())) + let conn = tonic::transport::Endpoint::new(dst)?.connect().await?; + Ok(Self::new(conn)) } } } diff --git a/tonic-examples/src/authentication/client.rs b/tonic-examples/src/authentication/client.rs index 729f206..b514773 100644 --- a/tonic-examples/src/authentication/client.rs +++ b/tonic-examples/src/authentication/client.rs @@ -15,7 +15,8 @@ async fn main() -> Result<(), Box> { HeaderValue::from_static("Bearer some-secret-token"), ); }) - .channel(); + .connect() + .await?; let mut client = EchoClient::new(channel); diff --git a/tonic-examples/src/gcp/client.rs b/tonic-examples/src/gcp/client.rs index 8eeaf24..3d0e0af 100644 --- a/tonic-examples/src/gcp/client.rs +++ b/tonic-examples/src/gcp/client.rs @@ -37,7 +37,8 @@ async fn main() -> Result<(), Box> { headers.insert("authorization", header_value.clone()); }) .tls_config(&tls_config) - .channel(); + .connect() + .await?; let mut service = PublisherClient::new(channel); diff --git a/tonic-examples/src/helloworld/client.rs b/tonic-examples/src/helloworld/client.rs index aeeefbe..227be6b 100644 --- a/tonic-examples/src/helloworld/client.rs +++ b/tonic-examples/src/helloworld/client.rs @@ -6,7 +6,7 @@ use hello_world::{client::GreeterClient, HelloRequest}; #[tokio::main] async fn main() -> Result<(), Box> { - let mut client = GreeterClient::connect("http://[::1]:50051")?; + let mut client = GreeterClient::connect("http://[::1]:50051").await?; let request = tonic::Request::new(HelloRequest { name: "Tonic".into(), diff --git a/tonic-examples/src/multiplex/client.rs b/tonic-examples/src/multiplex/client.rs index 6e8f19a..30f72a3 100644 --- a/tonic-examples/src/multiplex/client.rs +++ b/tonic-examples/src/multiplex/client.rs @@ -12,7 +12,9 @@ use tonic::transport::Endpoint; #[tokio::main] async fn main() -> Result<(), Box> { - let channel = Endpoint::from_static("http://[::1]:50051").channel(); + let channel = Endpoint::from_static("http://[::1]:50051") + .connect() + .await?; let mut greeter_client = GreeterClient::new(channel.clone()); let mut echo_client = EchoClient::new(channel); diff --git a/tonic-examples/src/routeguide/client.rs b/tonic-examples/src/routeguide/client.rs index e382c78..66d769d 100644 --- a/tonic-examples/src/routeguide/client.rs +++ b/tonic-examples/src/routeguide/client.rs @@ -91,7 +91,7 @@ async fn run_route_chat(client: &mut RouteGuideClient) -> Result<(), Bo #[tokio::main] async fn main() -> Result<(), Box> { - let mut client = RouteGuideClient::connect("http://[::1]:10000")?; + let mut client = RouteGuideClient::connect("http://[::1]:10000").await?; println!("*** SIMPLE RPC ***"); let response = client diff --git a/tonic-examples/src/tls/client.rs b/tonic-examples/src/tls/client.rs index 6370ba8..02c614c 100644 --- a/tonic-examples/src/tls/client.rs +++ b/tonic-examples/src/tls/client.rs @@ -17,7 +17,8 @@ async fn main() -> Result<(), Box> { let channel = Channel::from_static("http://[::1]:50051") .tls_config(&tls) - .channel(); + .connect() + .await?; let mut client = EchoClient::new(channel); let request = tonic::Request::new(EchoRequest { diff --git a/tonic-examples/src/tls_client_auth/client.rs b/tonic-examples/src/tls_client_auth/client.rs index 2b13cd2..1a1335a 100644 --- a/tonic-examples/src/tls_client_auth/client.rs +++ b/tonic-examples/src/tls_client_auth/client.rs @@ -21,8 +21,8 @@ async fn main() -> Result<(), Box> { let channel = Channel::from_static("http://[::1]:50051") .tls_config(&tls) - .clone() - .channel(); + .connect() + .await?; let mut client = EchoClient::new(channel); diff --git a/tonic-interop/src/bin/client.rs b/tonic-interop/src/bin/client.rs index 0b42830..46ebb9b 100644 --- a/tonic-interop/src/bin/client.rs +++ b/tonic-interop/src/bin/client.rs @@ -41,7 +41,7 @@ async fn main() -> Result<(), Box> { ); } - let channel = endpoint.channel(); + let channel = endpoint.connect().await?; let mut client = client::TestClient::new(channel.clone()); let mut unimplemented_client = client::UnimplementedClient::new(channel); diff --git a/tonic/Cargo.toml b/tonic/Cargo.toml index 662ef9f..527b754 100644 --- a/tonic/Cargo.toml +++ b/tonic/Cargo.toml @@ -29,7 +29,6 @@ transport = [ "hyper", "tokio", "tower", - "tower-reconnect", "tower-balance", "tower-load", ] @@ -68,7 +67,6 @@ hyper = { version = "=0.13.0-alpha.4", features = ["unstable-stream"], optional tokio = { version = "=0.2.0-alpha.6", default-features = false, features = ["tcp"], optional = true } tower = { version = "=0.3.0-alpha.2", optional = true} tower-make = "=0.3.0-alpha.2a" -tower-reconnect = { version = "=0.3.0-alpha.2", optional = true } tower-balance = { version = "=0.3.0-alpha.2", optional = true } tower-load = { version = "=0.3.0-alpha.2", optional = true } diff --git a/tonic/benches/benchmarks/compiled_protos/helloworld.rs b/tonic/benches/benchmarks/compiled_protos/helloworld.rs index bd66076..79afc2b 100755 --- a/tonic/benches/benchmarks/compiled_protos/helloworld.rs +++ b/tonic/benches/benchmarks/compiled_protos/helloworld.rs @@ -20,12 +20,13 @@ pub mod client { } impl GreeterClient { #[doc = r" Attempt to create a new client by connecting to a given endpoint."] - pub fn connect(dst: D) -> Result + pub async fn connect(dst: D) -> Result where D: std::convert::TryInto, D::Error: Into, { - tonic::transport::Endpoint::new(dst).map(|c| Self::new(c.channel())) + let conn = tonic::transport::Endpoint::new(dst)?.connect().await?; + Ok(Self::new(conn)) } } impl GreeterClient diff --git a/tonic/src/transport/channel.rs b/tonic/src/transport/channel.rs index 72ea7a4..3a13308 100644 --- a/tonic/src/transport/channel.rs +++ b/tonic/src/transport/channel.rs @@ -97,18 +97,20 @@ impl Channel { Self::balance(discover, buffer_size, interceptor_headers) } - pub(crate) fn connect(endpoint: Endpoint) -> Self { + pub(crate) async fn connect(endpoint: Endpoint) -> Result { let buffer_size = endpoint.buffer_size.clone().unwrap_or(DEFAULT_BUFFER_SIZE); let interceptor_headers = endpoint.interceptor_headers.clone(); - let svc = Connection::new(endpoint); + let svc = Connection::new(endpoint) + .await + .map_err(|e| super::Error::from_source(super::ErrorKind::Client, e))?; let svc = Buffer::new(Either::A(svc), buffer_size); - Channel { + Ok(Channel { svc, interceptor_headers, - } + }) } pub(crate) fn balance( diff --git a/tonic/src/transport/endpoint.rs b/tonic/src/transport/endpoint.rs index b2112c7..8a5ed4b 100644 --- a/tonic/src/transport/endpoint.rs +++ b/tonic/src/transport/endpoint.rs @@ -142,8 +142,8 @@ impl Endpoint { } /// Create a channel from this config. - pub fn channel(&self) -> Channel { - Channel::connect(self.clone()) + pub async fn connect(&self) -> Result { + Channel::connect(self.clone()).await } } diff --git a/tonic/src/transport/mod.rs b/tonic/src/transport/mod.rs index 961d8ce..718faf5 100644 --- a/tonic/src/transport/mod.rs +++ b/tonic/src/transport/mod.rs @@ -35,7 +35,8 @@ //! .timeout(Duration::from_secs(5)) //! .rate_limit(5, Duration::from_secs(1)) //! .concurrency_limit(256) -//! .channel(); +//! .connect() +//! .await?; //! //! channel.call(Request::new(BoxBody::empty())).await?; //! # Ok(()) diff --git a/tonic/src/transport/service/connection.rs b/tonic/src/transport/service/connection.rs index c308888..47693b3 100644 --- a/tonic/src/transport/service/connection.rs +++ b/tonic/src/transport/service/connection.rs @@ -1,4 +1,4 @@ -use super::{connector, layer::ServiceBuilderExt, AddOrigin}; +use super::{connector, layer::ServiceBuilderExt, reconnect::Reconnect, AddOrigin}; use crate::{body::BoxBody, transport::Endpoint}; use hyper::client::conn::Builder; use hyper::client::service::Connect as HyperConnect; @@ -16,7 +16,6 @@ use tower::{ ServiceBuilder, }; use tower_load::Load; -use tower_reconnect::Reconnect; use tower_service::Service; pub(crate) type Request = http::Request; @@ -27,7 +26,7 @@ pub(crate) struct Connection { } impl Connection { - pub(crate) fn new(endpoint: Endpoint) -> Self { + pub(crate) async fn new(endpoint: Endpoint) -> Result { #[cfg(feature = "tls")] let connector = connector(endpoint.tls.clone()); @@ -47,13 +46,15 @@ impl Connection { .optional_layer(endpoint.rate_limit.map(|(l, d)| RateLimitLayer::new(l, d))) .into_inner(); - let conn = Reconnect::new(HyperConnect::new(connector, settings), endpoint.uri.clone()); + let mut connector = HyperConnect::new(connector, settings); + let initial_conn = connector.call(endpoint.uri.clone()).await?; + let conn = Reconnect::new(initial_conn, connector, endpoint.uri.clone()); let inner = stack.layer(conn); - Self { + Ok(Self { inner: BoxService::new(inner), - } + }) } } diff --git a/tonic/src/transport/service/discover.rs b/tonic/src/transport/service/discover.rs index 1f6e61a..9d83d21 100644 --- a/tonic/src/transport/service/discover.rs +++ b/tonic/src/transport/service/discover.rs @@ -1,13 +1,18 @@ use super::connection::Connection; use crate::transport::Endpoint; -use std::collections::VecDeque; -use std::pin::Pin; -use std::task::{Context, Poll}; +use std::{ + collections::VecDeque, + fmt, + future::Future, + pin::Pin, + task::{Context, Poll}, +}; use tower::discover::{Change, Discover}; -#[derive(Debug)] pub(crate) struct ServiceList { list: VecDeque, + connecting: + Option> + Send + 'static>>>, i: usize, } @@ -15,6 +20,7 @@ impl ServiceList { pub(crate) fn new(list: Vec) -> Self { Self { list: list.into(), + connecting: None, i: 0, } } @@ -27,19 +33,34 @@ impl Discover for ServiceList { fn poll_discover( mut self: Pin<&mut Self>, - _cx: &mut Context<'_>, + cx: &mut Context<'_>, ) -> Poll, Self::Error>> { - match self.list.pop_front() { - Some(endpoint) => { + loop { + if let Some(connecting) = &mut self.connecting { + let svc = futures_core::ready!(Pin::new(connecting).poll(cx))?; + let i = self.i; self.i += 1; - let svc = Connection::new(endpoint); let change = Ok(Change::Insert(i, svc)); - Poll::Ready(change) + return Poll::Ready(change); + } + + if let Some(endpoint) = self.list.pop_front() { + let fut = Connection::new(endpoint); + self.connecting = Some(Box::pin(fut)); + } else { + return Poll::Pending; } - None => Poll::Pending, } } } + +impl fmt::Debug for ServiceList { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("ServiceList") + .field("list", &self.list) + .finish() + } +} diff --git a/tonic/src/transport/service/mod.rs b/tonic/src/transport/service/mod.rs index 1f3c7e9..0819f02 100644 --- a/tonic/src/transport/service/mod.rs +++ b/tonic/src/transport/service/mod.rs @@ -5,6 +5,7 @@ mod discover; mod either; mod io; mod layer; +mod reconnect; mod router; #[cfg(feature = "tls")] mod tls; diff --git a/tonic/src/transport/service/reconnect.rs b/tonic/src/transport/service/reconnect.rs new file mode 100644 index 0000000..344f962 --- /dev/null +++ b/tonic/src/transport/service/reconnect.rs @@ -0,0 +1,174 @@ +use crate::Error; +use pin_project::pin_project; +use std::fmt; +use std::{ + future::Future, + pin::Pin, + task::{Context, Poll}, +}; +use tower_make::MakeService; +use tower_service::Service; +use tracing::trace; + +pub(crate) struct Reconnect +where + M: Service, +{ + mk_service: M, + state: State, + target: Target, +} + +#[derive(Debug)] +enum State { + Idle, + Connecting(F), + Connected(S), +} + +impl Reconnect +where + M: Service, +{ + pub(crate) fn new(initial_connection: S, mk_service: M, target: Target) -> Self + where + M: Service, + S: Service, + Error: From + From, + Target: Clone, + { + Reconnect { + mk_service, + state: State::Connected(initial_connection), + target, + } + } +} + +impl Service for Reconnect +where + M: Service, + S: Service, + M::Future: Unpin, + Error: From + From, + Target: Clone, +{ + type Response = S::Response; + type Error = Error; + type Future = ResponseFuture; + + fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll> { + let ret; + let mut state; + + loop { + match self.state { + State::Idle => { + trace!("poll_ready; idle"); + match self.mk_service.poll_ready(cx) { + Poll::Ready(r) => r?, + Poll::Pending => { + trace!("poll_ready; MakeService not ready"); + return Poll::Pending; + } + } + + let fut = self.mk_service.make_service(self.target.clone()); + self.state = State::Connecting(fut); + continue; + } + State::Connecting(ref mut f) => { + trace!("poll_ready; connecting"); + match Pin::new(f).poll(cx) { + Poll::Ready(Ok(service)) => { + state = State::Connected(service); + } + Poll::Pending => { + trace!("poll_ready; not ready"); + return Poll::Pending; + } + Poll::Ready(Err(e)) => { + trace!("poll_ready; error"); + state = State::Idle; + ret = Err(e.into()); + break; + } + } + } + State::Connected(ref mut inner) => { + trace!("poll_ready; connected"); + match inner.poll_ready(cx) { + Poll::Ready(Ok(())) => { + trace!("poll_ready; ready"); + return Poll::Ready(Ok(())); + } + Poll::Pending => { + trace!("poll_ready; not ready"); + return Poll::Pending; + } + Poll::Ready(Err(_)) => { + trace!("poll_ready; error"); + state = State::Idle; + } + } + } + } + + self.state = state; + } + + self.state = state; + Poll::Ready(ret) + } + + fn call(&mut self, request: Request) -> Self::Future { + let service = match self.state { + State::Connected(ref mut service) => service, + _ => panic!("service not ready; poll_ready must be called first"), + }; + + let fut = service.call(request); + ResponseFuture::new(fut) + } +} + +impl fmt::Debug for Reconnect +where + M: Service + fmt::Debug, + M::Future: fmt::Debug, + M::Response: fmt::Debug, + Target: fmt::Debug, +{ + fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result { + fmt.debug_struct("Reconnect") + .field("mk_service", &self.mk_service) + .field("state", &self.state) + .field("target", &self.target) + .finish() + } +} + +#[pin_project] +#[derive(Debug)] +pub(crate) struct ResponseFuture { + #[pin] + inner: F, +} + +impl ResponseFuture { + pub(crate) fn new(inner: F) -> Self { + ResponseFuture { inner } + } +} + +impl Future for ResponseFuture +where + F: Future>, + E: Into, +{ + type Output = Result; + + fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { + self.project().inner.poll(cx).map_err(Into::into) + } +}