From fdda5ae26ace4706cb8faf56ccd35513b5de26c5 Mon Sep 17 00:00:00 2001 From: Quentin Perez Date: Mon, 11 Jan 2021 18:18:34 +0100 Subject: [PATCH] Upgrade to Tokio 1.0.0 ecosystem (#530) * Upgrade Tonic to Tokio 1.0 Work in progress for updating Tonic to Tokio 1.0. Since tower has not been released to crates.io, a git dependency is taken instead. * Upgrade Tonic to Tokio 1.0 phase 2 * tonic: remove tower-* deps * Apply suggestions from code review Co-authored-by: Ed Marshall Co-authored-by: Lucio Franco --- examples/Cargo.toml | 21 +++++----- examples/routeguide-tutorial.md | 6 +-- examples/src/autoreload/server.rs | 6 ++- examples/src/blocking/client.rs | 6 +-- examples/src/dynamic_load_balance/client.rs | 16 ++++---- examples/src/health/server.rs | 3 +- examples/src/routeguide/client.rs | 6 +-- examples/src/routeguide/server.rs | 9 +++-- examples/src/timeout/server.rs | 4 +- examples/src/uds/server.rs | 22 +++++++---- interop/Cargo.toml | 23 +++++------ interop/src/client.rs | 6 ++- interop/src/lib.rs | 5 +-- interop/src/server.rs | 6 +-- tests/ambiguous_methods/Cargo.toml | 2 +- tests/extern_path/my_application/Cargo.toml | 4 +- tests/extern_path/uuid/Cargo.toml | 6 +-- tests/included_service/Cargo.toml | 2 +- tests/integration_tests/Cargo.toml | 6 +-- tests/integration_tests/tests/connection.rs | 8 ++-- tests/integration_tests/tests/status.rs | 4 +- tests/integration_tests/tests/user_agent.rs | 2 +- tests/same_name/Cargo.toml | 2 +- tests/wellknown/Cargo.toml | 4 +- tonic-build/Cargo.toml | 2 +- tonic-health/Cargo.toml | 13 ++++--- tonic-health/src/server.rs | 43 ++++++++++----------- tonic-types/Cargo.toml | 6 +-- tonic/Cargo.toml | 41 +++++++++----------- tonic/benches/decode.rs | 5 +-- tonic/src/body.rs | 2 +- tonic/src/codec/buffer.rs | 18 ++++----- tonic/src/transport/channel/endpoint.rs | 2 +- tonic/src/transport/channel/mod.rs | 8 ++-- tonic/src/transport/server/incoming.rs | 4 +- tonic/src/transport/server/mod.rs | 2 +- tonic/src/transport/service/connection.rs | 2 +- tonic/src/transport/service/connector.rs | 2 +- tonic/src/transport/service/discover.rs | 22 +++++------ tonic/src/transport/service/io.rs | 10 ++--- tonic/src/transport/service/layer.rs | 7 +--- tonic/src/transport/service/reconnect.rs | 2 +- 42 files changed, 185 insertions(+), 185 deletions(-) diff --git a/examples/Cargo.toml b/examples/Cargo.toml index cb46056..321a39b 100644 --- a/examples/Cargo.toml +++ b/examples/Cargo.toml @@ -148,28 +148,29 @@ path = "src/hyper_warp_multiplex/server.rs" [dependencies] tonic = { path = "../tonic", features = ["tls"] } -prost = "0.6" -tokio = { version = "0.2", features = ["rt-threaded", "time", "stream", "fs", "macros", "uds"] } +prost = "0.7" +tokio = { version = "1.0", features = ["rt-multi-thread", "time", "fs", "macros", "net"] } +tokio-stream = { version = "0.1", features = ["net"] } +async-stream = "0.3" futures = { version = "0.3", default-features = false, features = ["alloc"] } -async-stream = "0.2" -tower = "0.3" +tower = { version = "0.4" } # Required for routeguide serde = { version = "1.0", features = ["derive"] } serde_json = "1.0" -rand = "0.7" +rand = "0.8" # Tracing tracing = "0.1.16" tracing-subscriber = { version = "0.2", features = ["tracing-log"] } tracing-attributes = "0.1" tracing-futures = "0.2" # Required for wellknown types -prost-types = "0.6" +prost-types = "0.7" # Hyper example -hyper = "0.13" -warp = { version = "0.2", default-features = false } +hyper = "0.14" +warp = { git = "https://github.com/aknuds1/warp", branch = "chore/upgrade-tokio", default-features = false } http = "0.2" -http-body = "0.3" -pin-project = "0.4.17" +http-body = "0.4" +pin-project = "1.0" # Health example tonic-health = { path = "../tonic-health" } listenfd = "0.3" diff --git a/examples/routeguide-tutorial.md b/examples/routeguide-tutorial.md index 75ae18b..e75b92e 100644 --- a/examples/routeguide-tutorial.md +++ b/examples/routeguide-tutorial.md @@ -702,7 +702,7 @@ use futures_util::stream; ```rust async fn run_record_route(client: &mut RouteGuideClient) -> Result<(), Box> { let mut rng = rand::thread_rng(); - let point_count: i32 = rng.gen_range(2, 100); + let point_count: i32 = rng.gen_range(2..100); let mut points = vec![]; for _ in 0..=point_count { @@ -723,8 +723,8 @@ async fn run_record_route(client: &mut RouteGuideClient) -> Result<(), ```rust fn random_point(rng: &mut ThreadRng) -> Point { - let latitude = (rng.gen_range(0, 180) - 90) * 10_000_000; - let longitude = (rng.gen_range(0, 360) - 180) * 10_000_000; + let latitude = (rng.gen_range(0..180) - 90) * 10_000_000; + let longitude = (rng.gen_range(0..360) - 180) * 10_000_000; Point { latitude, longitude, diff --git a/examples/src/autoreload/server.rs b/examples/src/autoreload/server.rs index a4b713a..acb3621 100644 --- a/examples/src/autoreload/server.rs +++ b/examples/src/autoreload/server.rs @@ -36,9 +36,11 @@ async fn main() -> Result<(), Box> { match listenfd::ListenFd::from_env().take_tcp_listener(0)? { Some(listener) => { - let mut listener = tokio::net::TcpListener::from_std(listener)?; + let listener = tokio_stream::wrappers::TcpListenerStream::new( + tokio::net::TcpListener::from_std(listener)?, + ); - server.serve_with_incoming(listener.incoming()).await?; + server.serve_with_incoming(listener).await?; } None => { server.serve(addr).await?; diff --git a/examples/src/blocking/client.rs b/examples/src/blocking/client.rs index 5326e19..12788e0 100644 --- a/examples/src/blocking/client.rs +++ b/examples/src/blocking/client.rs @@ -24,11 +24,7 @@ impl BlockingClient { D: std::convert::TryInto, D::Error: Into, { - let mut rt = Builder::new() - .basic_scheduler() - .enable_all() - .build() - .unwrap(); + let rt = Builder::new_multi_thread().enable_all().build().unwrap(); let client = rt.block_on(GreeterClient::connect(dst))?; Ok(Self { rt, client }) diff --git a/examples/src/dynamic_load_balance/client.rs b/examples/src/dynamic_load_balance/client.rs index 4ffa6a6..4c8a547 100644 --- a/examples/src/dynamic_load_balance/client.rs +++ b/examples/src/dynamic_load_balance/client.rs @@ -18,42 +18,42 @@ async fn main() -> Result<(), Box> { let e1 = Endpoint::from_static("http://[::1]:50051"); let e2 = Endpoint::from_static("http://[::1]:50052"); - let (channel, mut rx) = Channel::balance_channel(10); + let (channel, rx) = Channel::balance_channel(10); let mut client = EchoClient::new(channel); let done = Arc::new(AtomicBool::new(false)); let demo_done = done.clone(); tokio::spawn(async move { - tokio::time::delay_for(tokio::time::Duration::from_secs(5)).await; + tokio::time::sleep(tokio::time::Duration::from_secs(5)).await; println!("Added first endpoint"); let change = Change::Insert("1", e1); let res = rx.send(change).await; println!("{:?}", res); - tokio::time::delay_for(tokio::time::Duration::from_secs(5)).await; + tokio::time::sleep(tokio::time::Duration::from_secs(5)).await; println!("Added second endpoint"); let change = Change::Insert("2", e2); let res = rx.send(change).await; println!("{:?}", res); - tokio::time::delay_for(tokio::time::Duration::from_secs(5)).await; + tokio::time::sleep(tokio::time::Duration::from_secs(5)).await; println!("Removed first endpoint"); let change = Change::Remove("1"); let res = rx.send(change).await; println!("{:?}", res); - tokio::time::delay_for(tokio::time::Duration::from_secs(5)).await; + tokio::time::sleep(tokio::time::Duration::from_secs(5)).await; println!("Removed second endpoint"); let change = Change::Remove("2"); let res = rx.send(change).await; println!("{:?}", res); - tokio::time::delay_for(tokio::time::Duration::from_secs(5)).await; + tokio::time::sleep(tokio::time::Duration::from_secs(5)).await; println!("Added third endpoint"); let e3 = Endpoint::from_static("http://[::1]:50051"); let change = Change::Insert("3", e3); let res = rx.send(change).await; println!("{:?}", res); - tokio::time::delay_for(tokio::time::Duration::from_secs(5)).await; + tokio::time::sleep(tokio::time::Duration::from_secs(5)).await; println!("Removed third endpoint"); let change = Change::Remove("3"); let res = rx.send(change).await; @@ -62,7 +62,7 @@ async fn main() -> Result<(), Box> { }); while !done.load(SeqCst) { - tokio::time::delay_for(tokio::time::Duration::from_millis(500)).await; + tokio::time::sleep(tokio::time::Duration::from_millis(500)).await; let request = tonic::Request::new(EchoRequest { message: "hello".into(), }); diff --git a/examples/src/health/server.rs b/examples/src/health/server.rs index 625272d..7b12e86 100644 --- a/examples/src/health/server.rs +++ b/examples/src/health/server.rs @@ -3,7 +3,6 @@ use tonic::{transport::Server, Request, Response, Status}; use hello_world::greeter_server::{Greeter, GreeterServer}; use hello_world::{HelloReply, HelloRequest}; use std::time::Duration; -use tokio::time::delay_for; use tonic_health::server::HealthReporter; pub mod hello_world { @@ -34,7 +33,7 @@ async fn twiddle_service_status(mut reporter: HealthReporter) { let mut iter = 0u64; loop { iter += 1; - delay_for(Duration::from_secs(1)).await; + tokio::time::sleep(Duration::from_secs(1)).await; if iter % 2 == 0 { reporter.set_serving::>().await; diff --git a/examples/src/routeguide/client.rs b/examples/src/routeguide/client.rs index 04b8fd1..efa7a86 100644 --- a/examples/src/routeguide/client.rs +++ b/examples/src/routeguide/client.rs @@ -41,7 +41,7 @@ async fn print_features(client: &mut RouteGuideClient) -> Result<(), Bo async fn run_record_route(client: &mut RouteGuideClient) -> Result<(), Box> { let mut rng = rand::thread_rng(); - let point_count: i32 = rng.gen_range(2, 100); + let point_count: i32 = rng.gen_range(2..100); let mut points = vec![]; for _ in 0..=point_count { @@ -115,8 +115,8 @@ async fn main() -> Result<(), Box> { } fn random_point(rng: &mut ThreadRng) -> Point { - let latitude = (rng.gen_range(0, 180) - 90) * 10_000_000; - let longitude = (rng.gen_range(0, 360) - 180) * 10_000_000; + let latitude = (rng.gen_range(0..180) - 90) * 10_000_000; + let longitude = (rng.gen_range(0..360) - 180) * 10_000_000; Point { latitude, longitude, diff --git a/examples/src/routeguide/server.rs b/examples/src/routeguide/server.rs index 2f1c50e..fea8573 100644 --- a/examples/src/routeguide/server.rs +++ b/examples/src/routeguide/server.rs @@ -36,7 +36,8 @@ impl RouteGuide for RouteGuideService { Ok(Response::new(Feature::default())) } - type ListFeaturesStream = mpsc::Receiver>; + type ListFeaturesStream = + Pin> + Send + Sync + 'static>>; async fn list_features( &self, @@ -44,7 +45,7 @@ impl RouteGuide for RouteGuideService { ) -> Result, Status> { println!("ListFeatures = {:?}", request); - let (mut tx, rx) = mpsc::channel(4); + let (tx, rx) = mpsc::channel(4); let features = self.features.clone(); tokio::spawn(async move { @@ -58,7 +59,9 @@ impl RouteGuide for RouteGuideService { println!(" /// done sending"); }); - Ok(Response::new(rx)) + Ok(Response::new(Box::pin( + tokio_stream::wrappers::ReceiverStream::new(rx), + ))) } async fn record_route( diff --git a/examples/src/timeout/server.rs b/examples/src/timeout/server.rs index 7f75d24..2d5b1fa 100644 --- a/examples/src/timeout/server.rs +++ b/examples/src/timeout/server.rs @@ -1,5 +1,5 @@ use std::time::Duration; -use tokio::time::delay_for; +use tokio::time::sleep; use tonic::{transport::Server, Request, Response, Status}; use hello_world::greeter_server::{Greeter, GreeterServer}; @@ -20,7 +20,7 @@ impl Greeter for MyGreeter { ) -> Result, Status> { println!("Got a request from {:?}", request.remote_addr()); - delay_for(Duration::from_millis(5000)).await; + sleep(Duration::from_millis(5000)).await; let reply = hello_world::HelloReply { message: format!("Hello {}!", request.into_inner().name), diff --git a/examples/src/uds/server.rs b/examples/src/uds/server.rs index b744d72..07be063 100644 --- a/examples/src/uds/server.rs +++ b/examples/src/uds/server.rs @@ -1,6 +1,6 @@ #![cfg_attr(not(unix), allow(unused_imports))] -use futures::stream::TryStreamExt; +use futures::TryFutureExt; use std::path::Path; #[cfg(unix)] use tokio::net::UnixListener; @@ -40,13 +40,21 @@ async fn main() -> Result<(), Box> { tokio::fs::create_dir_all(Path::new(path).parent().unwrap()).await?; - let mut uds = UnixListener::bind(path)?; - let greeter = MyGreeter::default(); + let incoming = { + let uds = UnixListener::bind(path)?; + + async_stream::stream! { + while let item = uds.accept().map_ok(|(st, _)| unix::UnixStream(st)).await { + yield item; + } + } + }; + Server::builder() .add_service(GreeterServer::new(greeter)) - .serve_with_incoming(uds.incoming().map_ok(unix::UnixStream)) + .serve_with_incoming(incoming) .await?; Ok(()) @@ -59,7 +67,7 @@ mod unix { task::{Context, Poll}, }; - use tokio::io::{AsyncRead, AsyncWrite}; + use tokio::io::{AsyncRead, AsyncWrite, ReadBuf}; use tonic::transport::server::Connected; #[derive(Debug)] @@ -71,8 +79,8 @@ mod unix { fn poll_read( mut self: Pin<&mut Self>, cx: &mut Context<'_>, - buf: &mut [u8], - ) -> Poll> { + buf: &mut ReadBuf<'_>, + ) -> Poll> { Pin::new(&mut self.0).poll_read(cx, buf) } } diff --git a/interop/Cargo.toml b/interop/Cargo.toml index 58dc3b6..84bdb65 100644 --- a/interop/Cargo.toml +++ b/interop/Cargo.toml @@ -15,23 +15,24 @@ name = "server" path = "src/bin/server.rs" [dependencies] -tokio = { version = "0.2", features = ["rt-threaded", "time", "macros", "stream", "fs"] } +tokio = { version = "1.0", features = ["rt-multi-thread", "time", "macros", "fs"] } +tokio-stream = "0.1" +async-stream = "0.3" tonic = { path = "../tonic", features = ["tls"] } -prost = "0.6" -prost-derive = "0.6" -bytes = "0.5" +prost = "0.7" +prost-derive = "0.7" +bytes = "1.0" http = "0.2" futures-core = "0.3" futures-util = "0.3" -async-stream = "0.2" -tower = "0.3" -http-body = "0.3" -hyper = "0.13" -console = "0.9" +tower = { version = "0.4" } +http-body = "0.4" +hyper = "0.14" +console = "0.14" structopt = "0.3" tracing = "0.1" -tracing-subscriber = "0.2.0-alpha" -tracing-log = "0.1.0" +tracing-subscriber = "0.2" +tracing-log = "0.1" [build-dependencies] tonic-build = { path = "../tonic-build", features = ["prost"] } diff --git a/interop/src/client.rs b/interop/src/client.rs index e79b041..a47441d 100644 --- a/interop/src/client.rs +++ b/interop/src/client.rs @@ -154,7 +154,11 @@ pub async fn ping_pong(client: &mut TestClient, assertions: &mut Vec bool { - match self { - TestAssertion::Failed { .. } => true, - _ => false, - } + matches!(self, TestAssertion::Failed { .. }) } } diff --git a/interop/src/server.rs b/interop/src/server.rs index 154ed50..112ff38 100644 --- a/interop/src/server.rs +++ b/interop/src/server.rs @@ -72,7 +72,7 @@ impl pb::test_service_server::TestService for TestService { let stream = try_stream! { for param in response_parameters { - tokio::time::delay_for(Duration::from_micros(param.interval_us as u64)).await; + tokio::time::sleep(Duration::from_micros(param.interval_us as u64)).await; let payload = crate::server_payload(param.size as usize); yield StreamingOutputCallResponse { payload: Some(payload) }; @@ -90,7 +90,7 @@ impl pb::test_service_server::TestService for TestService { ) -> Result { let mut stream = req.into_inner(); - let mut aggregated_payload_size = 0 as i32; + let mut aggregated_payload_size = 0; while let Some(msg) = stream.try_next().await? { aggregated_payload_size += msg.payload.unwrap().body.len() as i32; } @@ -127,7 +127,7 @@ impl pb::test_service_server::TestService for TestService { } for param in msg.response_parameters { - tokio::time::delay_for(Duration::from_micros(param.interval_us as u64)).await; + tokio::time::sleep(Duration::from_micros(param.interval_us as u64)).await; let payload = crate::server_payload(param.size as usize); yield StreamingOutputCallResponse { payload: Some(payload) }; diff --git a/tests/ambiguous_methods/Cargo.toml b/tests/ambiguous_methods/Cargo.toml index ecfb377..f1a54e3 100644 --- a/tests/ambiguous_methods/Cargo.toml +++ b/tests/ambiguous_methods/Cargo.toml @@ -10,7 +10,7 @@ license = "MIT" [dependencies] tonic = { path= "../../tonic" } -prost = "0.6" +prost = "0.7" [build-dependencies] tonic-build = { path= "../../tonic-build" } diff --git a/tests/extern_path/my_application/Cargo.toml b/tests/extern_path/my_application/Cargo.toml index 801a829..5030afb 100644 --- a/tests/extern_path/my_application/Cargo.toml +++ b/tests/extern_path/my_application/Cargo.toml @@ -10,8 +10,8 @@ license = "MIT" [dependencies] tonic = { path= "../../../tonic" } -prost = "0.6" -prost-types = "0.6" +prost = "0.7" +prost-types = "0.7" uuid = { package = "uuid1", path= "../uuid" } [build-dependencies] diff --git a/tests/extern_path/uuid/Cargo.toml b/tests/extern_path/uuid/Cargo.toml index 2fdfad3..657e9bf 100644 --- a/tests/extern_path/uuid/Cargo.toml +++ b/tests/extern_path/uuid/Cargo.toml @@ -9,7 +9,7 @@ license = "MIT" # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html [dependencies] -prost = "0.6" -bytes = "0.5" +prost = "0.7" +bytes = "1.0" [build-dependencies] -prost-build = "0.6" +prost-build = "0.7" diff --git a/tests/included_service/Cargo.toml b/tests/included_service/Cargo.toml index f51839d..454cd2a 100644 --- a/tests/included_service/Cargo.toml +++ b/tests/included_service/Cargo.toml @@ -10,7 +10,7 @@ license = "MIT" [dependencies] tonic = { path = "../../tonic" } -prost = "0.6" +prost = "0.7" [build-dependencies] tonic-build = { path = "../../tonic-build" } diff --git a/tests/integration_tests/Cargo.toml b/tests/integration_tests/Cargo.toml index fbca006..37ef59a 100644 --- a/tests/integration_tests/Cargo.toml +++ b/tests/integration_tests/Cargo.toml @@ -10,12 +10,12 @@ license = "MIT" [dependencies] tonic = { path = "../../tonic" } -prost = "0.6" +prost = "0.7" futures-util = "0.3" -bytes = "0.5" +bytes = "1.0" [dev-dependencies] -tokio = { version = "0.2", features = ["macros", "rt-core", "tcp"] } +tokio = { version = "1.0", features = ["macros", "rt-multi-thread", "net"] } [build-dependencies] tonic-build = { path = "../../tonic-build" } diff --git a/tests/integration_tests/tests/connection.rs b/tests/integration_tests/tests/connection.rs index 4ffc2c3..c345a35 100644 --- a/tests/integration_tests/tests/connection.rs +++ b/tests/integration_tests/tests/connection.rs @@ -41,14 +41,14 @@ async fn connect_returns_err_via_call_after_connected() { .unwrap(); }); - tokio::time::delay_for(Duration::from_millis(100)).await; + tokio::time::sleep(Duration::from_millis(100)).await; let mut client = TestClient::connect("http://127.0.0.1:1338").await.unwrap(); // First call should pass, then shutdown the server client.unary_call(Request::new(Input {})).await.unwrap(); - tokio::time::delay_for(Duration::from_millis(100)).await; + tokio::time::sleep(Duration::from_millis(100)).await; let res = client.unary_call(Request::new(Input {})).await; @@ -81,11 +81,11 @@ async fn connect_lazy_reconnects_after_first_failure() { .unwrap(); }); - tokio::time::delay_for(Duration::from_millis(100)).await; + tokio::time::sleep(Duration::from_millis(100)).await; client.unary_call(Request::new(Input {})).await.unwrap(); // The server shut down, third call should fail - tokio::time::delay_for(Duration::from_millis(100)).await; + tokio::time::sleep(Duration::from_millis(100)).await; client.unary_call(Request::new(Input {})).await.unwrap_err(); jh.await.unwrap(); diff --git a/tests/integration_tests/tests/status.rs b/tests/integration_tests/tests/status.rs index d3e69f1..6725784 100644 --- a/tests/integration_tests/tests/status.rs +++ b/tests/integration_tests/tests/status.rs @@ -33,7 +33,7 @@ async fn status_with_details() { .unwrap(); }); - tokio::time::delay_for(Duration::from_millis(100)).await; + tokio::time::sleep(Duration::from_millis(100)).await; let mut channel = test_client::TestClient::connect("http://127.0.0.1:1337") .await @@ -87,7 +87,7 @@ async fn status_with_metadata() { .unwrap(); }); - tokio::time::delay_for(Duration::from_millis(100)).await; + tokio::time::sleep(Duration::from_millis(100)).await; let mut channel = test_client::TestClient::connect("http://127.0.0.1:1338") .await diff --git a/tests/integration_tests/tests/user_agent.rs b/tests/integration_tests/tests/user_agent.rs index 12b7b3f..2d6406d 100644 --- a/tests/integration_tests/tests/user_agent.rs +++ b/tests/integration_tests/tests/user_agent.rs @@ -33,7 +33,7 @@ async fn writes_user_agent_header() { .unwrap(); }); - tokio::time::delay_for(Duration::from_millis(100)).await; + tokio::time::sleep(Duration::from_millis(100)).await; let channel = Endpoint::from_static("http://127.0.0.1:1322") .user_agent("my-client") diff --git a/tests/same_name/Cargo.toml b/tests/same_name/Cargo.toml index d166ebc..40d809b 100644 --- a/tests/same_name/Cargo.toml +++ b/tests/same_name/Cargo.toml @@ -10,7 +10,7 @@ license = "MIT" [dependencies] tonic = { path = "../../tonic" } -prost = "0.6" +prost = "0.7" [build-dependencies] tonic-build = { path = "../../tonic-build" } diff --git a/tests/wellknown/Cargo.toml b/tests/wellknown/Cargo.toml index aab66b9..c272ec3 100644 --- a/tests/wellknown/Cargo.toml +++ b/tests/wellknown/Cargo.toml @@ -10,8 +10,8 @@ license = "MIT" [dependencies] tonic = { path = "../../tonic" } -prost = "0.6" -prost-types = "0.6" +prost = "0.7" +prost-types = "0.7" [build-dependencies] tonic-build = { path = "../../tonic-build" } diff --git a/tonic-build/Cargo.toml b/tonic-build/Cargo.toml index 8babc0e..6586e3e 100644 --- a/tonic-build/Cargo.toml +++ b/tonic-build/Cargo.toml @@ -16,7 +16,7 @@ keywords = ["rpc", "grpc", "async", "codegen", "protobuf"] [dependencies] -prost-build = { version = "0.6", optional = true } +prost-build = { version = "0.7", optional = true } syn = "1.0" quote = "1.0" proc-macro2 = "1.0" diff --git a/tonic-health/Cargo.toml b/tonic-health/Cargo.toml index e7de702..988d13b 100644 --- a/tonic-health/Cargo.toml +++ b/tonic-health/Cargo.toml @@ -18,12 +18,15 @@ default = ["transport"] transport = ["tonic/transport"] [dependencies] -tokio = { version = "0.2", features = ["sync", "stream"] } -tonic = { version = "0.3", path = "../tonic", features = ["codegen", "prost"] } -prost = "0.6" +tokio = { version = "1.0", features = ["sync"] } +tonic = { path = "../tonic", features = ["codegen", "prost"] } +bytes = "1.0" +prost = "0.7" +tokio-stream = "0.1" +async-stream = "0.3" [dev-dependencies] -tokio = { version = "0.2", features = ["rt-core", "macros"]} +tokio = { version = "1.0", features = ["rt-multi-thread", "macros"]} [build-dependencies] -tonic-build = { version = "0.3", path = "../tonic-build" } \ No newline at end of file +tonic-build = { path = "../tonic-build" } \ No newline at end of file diff --git a/tonic-health/src/server.rs b/tonic-health/src/server.rs index 1547b8d..7f03233 100644 --- a/tonic-health/src/server.rs +++ b/tonic-health/src/server.rs @@ -6,8 +6,8 @@ use crate::ServingStatus; use std::collections::HashMap; use std::pin::Pin; use std::sync::Arc; -use tokio::stream::{Stream, StreamExt}; use tokio::sync::{watch, RwLock}; +use tokio_stream::Stream; #[cfg(feature = "transport")] use tonic::transport::NamedService; use tonic::{Request, Response, Status}; @@ -80,20 +80,15 @@ impl HealthReporter { let service_name = service_name.as_ref(); let mut writer = self.statuses.write().await; match writer.get(service_name) { - None => { - let _ = writer.insert(service_name.to_string(), watch::channel(status)); - } - Some((tx, rx)) => { - let mut rx = rx.clone(); - if rx.recv().await == Some(status) { - return; - } - + Some((tx, _)) => { // We only ever hand out clones of the receiver, so the originally-created // receiver should always be present, only being dropped when clearing the - // service status. Consequently, `tx.broadcast` should not fail, making use + // service status. Consequently, `tx.send` should not fail, making use // of `expect` here safe. - tx.broadcast(status).expect("channel should not be closed"); + tx.send(status).expect("channel should not be closed"); + } + None => { + writer.insert(service_name.to_string(), watch::channel(status)); } }; } @@ -116,10 +111,7 @@ impl HealthService { async fn service_health(&self, service_name: &str) -> Option { let reader = self.statuses.read().await; - match reader.get(service_name).map(|p| p.1.clone()) { - None => None, - Some(mut receiver) => receiver.recv().await, - } + reader.get(service_name).map(|p| *p.1.borrow()) } } @@ -148,15 +140,21 @@ impl Health for HealthService { request: Request, ) -> Result, Status> { let service_name = request.get_ref().service.as_str(); - let status_rx = match self.statuses.read().await.get(service_name) { + let mut status_rx = match self.statuses.read().await.get(service_name) { None => return Err(Status::not_found("service not registered")), Some(pair) => pair.1.clone(), }; - let output = status_rx.map(|status| { - let status = crate::proto::health_check_response::ServingStatus::from(status) as i32; - Ok(HealthCheckResponse { status }) - }); + let output = async_stream::try_stream! { + // yield the current value + let status = crate::proto::health_check_response::ServingStatus::from(*status_rx.borrow()) as i32; + yield HealthCheckResponse { status }; + + while let Ok(_) = status_rx.changed().await { + let status = crate::proto::health_check_response::ServingStatus::from(*status_rx.borrow()) as i32; + yield HealthCheckResponse { status }; + } + }; Ok(Response::new(Box::pin(output) as Self::WatchStream)) } @@ -170,8 +168,8 @@ mod tests { use crate::ServingStatus; use std::collections::HashMap; use std::sync::Arc; - use tokio::stream::StreamExt; use tokio::sync::{watch, RwLock}; + use tokio_stream::StreamExt; use tonic::{Code, Request, Status}; fn assert_serving_status(wire: i32, expected: ServingStatus) { @@ -269,6 +267,7 @@ mod tests { reporter .set_service_status("TestService", ServingStatus::NotServing) .await; + let item = resp .next() .await diff --git a/tonic-types/Cargo.toml b/tonic-types/Cargo.toml index 7bbf1e4..67ba890 100644 --- a/tonic-types/Cargo.toml +++ b/tonic-types/Cargo.toml @@ -15,8 +15,8 @@ categories = ["web-programming", "network-programming", "asynchronous"] keywords = ["rpc", "grpc", "protobuf"] [dependencies] -prost = "0.6" -prost-types = "0.6" +prost = "0.7" +prost-types = "0.7" [build-dependencies] -prost-build = "0.6" +prost-build = "0.7" diff --git a/tonic/Cargo.toml b/tonic/Cargo.toml index 64ec757..ef1b650 100644 --- a/tonic/Cargo.toml +++ b/tonic/Cargo.toml @@ -30,8 +30,6 @@ transport = [ "hyper", "tokio", "tower", - "tower-balance", - "tower-load", "tracing-futures", ] tls = ["transport", "tokio-rustls"] @@ -43,45 +41,43 @@ prost = ["prost1", "prost-derive"] # harness = false [dependencies] -bytes = "0.5" +bytes = "1.0" futures-core = { version = "0.3", default-features = false } futures-util = { version = "0.3", default-features = false } tracing = "0.1" http = "0.2" -base64 = "0.12" +base64 = "0.13" -percent-encoding = "2.0" +percent-encoding = "2.1" tower-service = "0.3" -tokio-util = { version = "0.3", features = ["codec"] } -async-stream = "0.2" -http-body = "0.3" -pin-project = "0.4.17" +tokio-util = { version = "0.6", features = ["codec"] } +async-stream = "0.3" +http-body = "0.4" +pin-project = "1.0" # prost -prost1 = { package = "prost", version = "0.6", optional = true } -prost-derive = { version = "0.6", optional = true } +prost1 = { package = "prost", version = "0.7", optional = true } +prost-derive = { version = "0.7", optional = true } # codegen async-trait = { version = "0.1.13", optional = true } # transport -h2 = { version = "0.2.2", optional = true } -hyper = { version = "0.13.4", features = ["stream"], optional = true } -tokio = { version = "0.2.13", features = ["tcp"], optional = true } -tower = { version = "0.3", optional = true} -tower-make = { version = "0.3", features = ["connect"] } -tower-balance = { version = "0.3", optional = true } -tower-load = { version = "0.3", optional = true } +h2 = { version = "0.3", optional = true } +hyper = { version = "0.14.2", features = ["full"], optional = true } +tokio = { version = "1.0.1", features = ["net"], optional = true } +tokio-stream = "0.1" +tower = { version = "0.4", features = ["balance", "buffer", "discover", "limit", "load", "make", "timeout", "util"], optional = true} tracing-futures = { version = "0.2", optional = true } # rustls -tokio-rustls = { version = "0.14", optional = true } -rustls-native-certs = { version = "0.4", optional = true } +tokio-rustls = { version = "0.22", optional = true } +rustls-native-certs = { version = "0.5", optional = true } [dev-dependencies] -tokio = { version = "0.2", features = ["rt-core", "macros"] } +tokio = { version = "1.0", features = ["rt", "macros"] } static_assertions = "1.0" -rand = "0.7" +rand = "0.8" bencher = "0.1.5" [package.metadata.docs.rs] @@ -91,4 +87,3 @@ rustdoc-args = ["--cfg", "docsrs"] [[bench]] name = "decode" harness = false - diff --git a/tonic/benches/decode.rs b/tonic/benches/decode.rs index 529e693..41b2496 100644 --- a/tonic/benches/decode.rs +++ b/tonic/benches/decode.rs @@ -11,8 +11,7 @@ use tonic::{codec::DecodeBuf, codec::Decoder, Status, Streaming}; macro_rules! bench { ($name:ident, $message_size:expr, $chunk_size:expr, $message_count:expr) => { fn $name(b: &mut Bencher) { - let mut rt = tokio::runtime::Builder::new() - .basic_scheduler() + let rt = tokio::runtime::Builder::new_multi_thread() .build() .expect("runtime"); @@ -102,7 +101,7 @@ impl Decoder for MockDecoder { type Error = Status; fn decode(&mut self, buf: &mut DecodeBuf<'_>) -> Result, Self::Error> { - let out = Vec::from(buf.bytes()); + let out = Vec::from(buf.chunk()); buf.advance(self.message_size); Ok(Some(out)) } diff --git a/tonic/src/body.rs b/tonic/src/body.rs index 1714cfb..7753086 100644 --- a/tonic/src/body.rs +++ b/tonic/src/body.rs @@ -160,7 +160,7 @@ where Pin::new_unchecked(&mut me.0).poll_data(cx) }; match futures_util::ready!(v) { - Some(Ok(mut i)) => Poll::Ready(Some(Ok(i.to_bytes()))), + Some(Ok(mut i)) => Poll::Ready(Some(Ok(i.copy_to_bytes(i.remaining())))), Some(Err(e)) => { let err = Status::map_error(e.into()); Poll::Ready(Some(Err(err))) diff --git a/tonic/src/codec/buffer.rs b/tonic/src/codec/buffer.rs index 808348c..fcce82e 100644 --- a/tonic/src/codec/buffer.rs +++ b/tonic/src/codec/buffer.rs @@ -1,5 +1,5 @@ +use bytes::buf::UninitSlice; use bytes::{Buf, BufMut, BytesMut}; -use std::mem::MaybeUninit; /// A specialized buffer to decode gRPC messages from. #[derive(Debug)] @@ -27,8 +27,8 @@ impl Buf for DecodeBuf<'_> { } #[inline] - fn bytes(&self) -> &[u8] { - let ret = self.buf.bytes(); + fn chunk(&self) -> &[u8] { + let ret = self.buf.chunk(); if ret.len() > self.len { &ret[..self.len] @@ -63,7 +63,7 @@ impl EncodeBuf<'_> { } } -impl BufMut for EncodeBuf<'_> { +unsafe impl BufMut for EncodeBuf<'_> { #[inline] fn remaining_mut(&self) -> usize { self.buf.remaining_mut() @@ -75,8 +75,8 @@ impl BufMut for EncodeBuf<'_> { } #[inline] - fn bytes_mut(&mut self) -> &mut [MaybeUninit] { - self.buf.bytes_mut() + fn chunk_mut(&mut self) -> &mut UninitSlice { + self.buf.chunk_mut() } } @@ -92,7 +92,7 @@ mod tests { assert_eq!(buf.len, 20); assert_eq!(buf.remaining(), 20); - assert_eq!(buf.bytes().len(), 20); + assert_eq!(buf.chunk().len(), 20); buf.advance(10); assert_eq!(buf.remaining(), 10); @@ -100,9 +100,9 @@ mod tests { let mut out = [0; 5]; buf.copy_to_slice(&mut out); assert_eq!(buf.remaining(), 5); - assert_eq!(buf.bytes().len(), 5); + assert_eq!(buf.chunk().len(), 5); - assert_eq!(buf.to_bytes().len(), 5); + assert_eq!(buf.copy_to_bytes(5).len(), 5); assert!(!buf.has_remaining()); } diff --git a/tonic/src/transport/channel/endpoint.rs b/tonic/src/transport/channel/endpoint.rs index 788fa2e..13eb4b9 100644 --- a/tonic/src/transport/channel/endpoint.rs +++ b/tonic/src/transport/channel/endpoint.rs @@ -15,7 +15,7 @@ use std::{ fmt, time::Duration, }; -use tower_make::MakeConnection; +use tower::make::MakeConnection; /// Channel builder. /// diff --git a/tonic/src/transport/channel/mod.rs b/tonic/src/transport/channel/mod.rs index 4d71752..eca92dd 100644 --- a/tonic/src/transport/channel/mod.rs +++ b/tonic/src/transport/channel/mod.rs @@ -29,13 +29,13 @@ use tokio::{ sync::mpsc::{channel, Sender}, }; +use tower::balance::p2c::Balance; use tower::{ buffer::{self, Buffer}, discover::{Change, Discover}, util::{BoxService, Either}, Service, }; -use tower_balance::p2c::Balance; type Svc = Either, Response, crate::Error>>; @@ -109,7 +109,7 @@ impl Channel { /// This creates a [`Channel`] that will load balance accross all the /// provided endpoints. pub fn balance_list(list: impl Iterator) -> Self { - let (channel, mut tx) = Self::balance_channel(DEFAULT_BUFFER_SIZE); + let (channel, tx) = Self::balance_channel(DEFAULT_BUFFER_SIZE); list.for_each(|endpoint| { tx.try_send(Change::Insert(endpoint.uri.clone(), endpoint)) .unwrap(); @@ -166,9 +166,9 @@ impl Channel { where D: Discover + Unpin + Send + 'static, D::Error: Into, - D::Key: Send + Clone, + D::Key: Hash + Send + Clone, { - let svc = Balance::from_entropy(discover); + let svc = Balance::new(discover); let svc = BoxService::new(svc); let svc = Buffer::new(Either::B(svc), buffer_size); diff --git a/tonic/src/transport/server/incoming.rs b/tonic/src/transport/server/incoming.rs index e33c77e..2e11128 100644 --- a/tonic/src/transport/server/incoming.rs +++ b/tonic/src/transport/server/incoming.rs @@ -105,8 +105,8 @@ where fn poll_read( self: Pin<&mut Self>, cx: &mut Context<'_>, - buf: &mut [u8], - ) -> Poll> { + buf: &mut tokio::io::ReadBuf<'_>, + ) -> Poll> { use std::future::Future; let pin = self.get_mut(); diff --git a/tonic/src/transport/server/mod.rs b/tonic/src/transport/server/mod.rs index 5775165..569be8a 100644 --- a/tonic/src/transport/server/mod.rs +++ b/tonic/src/transport/server/mod.rs @@ -163,7 +163,7 @@ impl Server { /// ``` /// # use tonic::transport::Server; /// # use tower_service::Service; - /// # let mut builder = Server::builder(); + /// # let builder = Server::builder(); /// builder.concurrency_limit_per_connection(32); /// ``` pub fn concurrency_limit_per_connection(self, limit: usize) -> Self { diff --git a/tonic/src/transport/service/connection.rs b/tonic/src/transport/service/connection.rs index 52e82ab..ed23aac 100644 --- a/tonic/src/transport/service/connection.rs +++ b/tonic/src/transport/service/connection.rs @@ -10,6 +10,7 @@ use std::{ task::{Context, Poll}, }; use tokio::io::{AsyncRead, AsyncWrite}; +use tower::load::Load; use tower::{ layer::Layer, limit::{concurrency::ConcurrencyLimitLayer, rate::RateLimitLayer}, @@ -17,7 +18,6 @@ use tower::{ util::BoxService, ServiceBuilder, ServiceExt, }; -use tower_load::Load; use tower_service::Service; pub(crate) type Request = http::Request; diff --git a/tonic/src/transport/service/connector.rs b/tonic/src/transport/service/connector.rs index 25d6a19..8bd82de 100644 --- a/tonic/src/transport/service/connector.rs +++ b/tonic/src/transport/service/connector.rs @@ -4,7 +4,7 @@ use super::io::BoxedIo; use super::tls::TlsConnector; use http::Uri; use std::task::{Context, Poll}; -use tower_make::MakeConnection; +use tower::make::MakeConnection; use tower_service::Service; #[cfg(not(feature = "tls"))] diff --git a/tonic/src/transport/service/discover.rs b/tonic/src/transport/service/discover.rs index a5feccb..4088f8d 100644 --- a/tonic/src/transport/service/discover.rs +++ b/tonic/src/transport/service/discover.rs @@ -7,9 +7,10 @@ use std::{ pin::Pin, task::{Context, Poll}, }; -use tokio::{stream::Stream, sync::mpsc::Receiver}; +use tokio::sync::mpsc::Receiver; -use tower::discover::{Change, Discover}; +use tokio_stream::Stream; +use tower::discover::Change; type DiscoverResult = Result, E>; @@ -23,17 +24,12 @@ impl DynamicServiceStream { } } -impl Discover for DynamicServiceStream { - type Key = K; - type Service = Connection; - type Error = crate::Error; +impl Stream for DynamicServiceStream { + type Item = DiscoverResult; - fn poll_discover( - mut self: Pin<&mut Self>, - cx: &mut Context<'_>, - ) -> Poll> { + fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { let c = &mut self.changes; - match Pin::new(&mut *c).poll_next(cx) { + match Pin::new(&mut *c).poll_recv(cx) { Poll::Pending | Poll::Ready(None) => Poll::Pending, Poll::Ready(Some(change)) => match change { Change::Insert(k, endpoint) => { @@ -48,9 +44,9 @@ impl Discover for DynamicServiceStream { let connector = service::connector(http); let connection = Connection::lazy(connector, endpoint); let change = Ok(Change::Insert(k, connection)); - Poll::Ready(change) + Poll::Ready(Some(change)) } - Change::Remove(k) => Poll::Ready(Ok(Change::Remove(k))), + Change::Remove(k) => Poll::Ready(Some(Ok(Change::Remove(k)))), }, } } diff --git a/tonic/src/transport/service/io.rs b/tonic/src/transport/service/io.rs index 9896150..761c8ec 100644 --- a/tonic/src/transport/service/io.rs +++ b/tonic/src/transport/service/io.rs @@ -4,7 +4,7 @@ use std::io; use std::net::SocketAddr; use std::pin::Pin; use std::task::{Context, Poll}; -use tokio::io::{AsyncRead, AsyncWrite}; +use tokio::io::{AsyncRead, AsyncWrite, ReadBuf}; pub(in crate::transport) trait Io: AsyncRead + AsyncWrite + Send + 'static @@ -33,8 +33,8 @@ impl AsyncRead for BoxedIo { fn poll_read( mut self: Pin<&mut Self>, cx: &mut Context<'_>, - buf: &mut [u8], - ) -> Poll> { + buf: &mut ReadBuf<'_>, + ) -> Poll> { Pin::new(&mut self.0).poll_read(cx, buf) } } @@ -83,8 +83,8 @@ impl AsyncRead for ServerIo { fn poll_read( mut self: Pin<&mut Self>, cx: &mut Context<'_>, - buf: &mut [u8], - ) -> Poll> { + buf: &mut ReadBuf<'_>, + ) -> Poll> { Pin::new(&mut self.0).poll_read(cx, buf) } } diff --git a/tonic/src/transport/service/layer.rs b/tonic/src/transport/service/layer.rs index 765f3b9..b2b1819 100644 --- a/tonic/src/transport/service/layer.rs +++ b/tonic/src/transport/service/layer.rs @@ -1,8 +1,5 @@ -use tower::{ - layer::{Layer, Stack}, - util::Either, - ServiceBuilder, -}; +use tower::layer::util::Stack; +use tower::{layer::Layer, util::Either, ServiceBuilder}; pub(crate) trait ServiceBuilderExt { fn layer_fn Out, S, Out>(self, f: F) -> ServiceBuilder, L>>; diff --git a/tonic/src/transport/service/reconnect.rs b/tonic/src/transport/service/reconnect.rs index c123d96..5d1b501 100644 --- a/tonic/src/transport/service/reconnect.rs +++ b/tonic/src/transport/service/reconnect.rs @@ -6,7 +6,7 @@ use std::{ pin::Pin, task::{Context, Poll}, }; -use tower_make::MakeService; +use tower::make::MakeService; use tower_service::Service; use tracing::trace;