feat(grpc): Add tonic transport (#2339)

Author: arjan-bal

codegen/src/main.rs

@@ -68,6 +68,19 @@ fn main() {
         false,
     );
 
+    // grpc
+    codegen(
+        &PathBuf::from(std::env!("CARGO_MANIFEST_DIR"))
+            .parent()
+            .unwrap()
+            .join("grpc"),
+        &["proto/echo/echo.proto"],
+        &["proto"],
+        &PathBuf::from("src/generated"),
+        &PathBuf::from("src/generated/echo_fds.rs"),
+        true,
+        true,
+    );
     println!("Codgen completed: {}ms", start.elapsed().as_millis());
 }
 

examples/Cargo.toml

@@ -263,7 +263,6 @@ tracing = ["dep:tracing", "dep:tracing-subscriber"]
 uds = ["dep:tokio-stream", "tokio-stream?/net", "dep:tower", "dep:hyper", "dep:hyper-util"]
 streaming = ["dep:tokio-stream", "dep:h2"]
 mock = ["dep:tokio-stream", "dep:tower", "dep:hyper-util"]
-tower = ["dep:tower", "dep:http"]
 json-codec = ["dep:serde", "dep:serde_json", "dep:bytes"]
 compression = ["tonic/gzip"]
 tls = ["tonic/tls-ring"]
@@ -273,7 +272,7 @@ types = ["dep:tonic-types"]
 h2c = ["dep:hyper", "dep:tower", "dep:http", "dep:hyper-util"]
 cancellation = ["dep:tokio-util"]
 
-full = ["gcp", "routeguide", "reflection", "autoreload", "health", "grpc-web", "tracing", "uds", "streaming", "mock", "tower", "json-codec", "compression", "tls", "tls-rustls", "tls-client-auth", "types", "cancellation", "h2c"]
+full = ["gcp", "routeguide", "reflection", "autoreload", "health", "grpc-web", "tracing", "uds", "streaming", "mock", "json-codec", "compression", "tls", "tls-rustls", "tls-client-auth", "types", "cancellation", "h2c"]
 default = ["full"]
 
 [dependencies]

grpc/Cargo.toml

@@ -6,39 +6,59 @@ authors = ["gRPC Authors"]
 license = "MIT"
 rust-version = "1.86"
 
+[package.metadata.cargo_check_external_types]
+allowed_external_types = [
+    "tonic::*",
+    "futures_core::stream::Stream",
+    "tokio::sync::oneshot::Sender",
+]
+
+[features]
+default = ["dns", "_runtime-tokio"]
+dns = ["dep:hickory-resolver", "_runtime-tokio"]
+# The following feature is used to ensure all modules use the runtime
+# abstraction instead of using tokio directly.
+# Using tower/buffer enables tokio's rt feature even though it's possible to
+# create Buffers with a user provided executor.
+_runtime-tokio = [
+    "tokio/rt",
+    "tokio/net",
+    "tokio/time",
+    "dep:socket2",
+    "dep:tower",
+]
+
 [dependencies]
 bytes = "1.10.1"
 hickory-resolver = { version = "0.25.1", optional = true }
 http = "1.1.0"
 http-body = "1.0.1"
 hyper = { version = "1.6.0", features = ["client", "http2"] }
-hyper-util = "0.1.14"
 parking_lot = "0.12.4"
 pin-project-lite = "0.2.16"
 rand = "0.9"
 serde = { version = "1.0.219", features = ["derive"] }
 serde_json = "1.0.140"
-socket2 = "0.5.10"
-tokio = { version = "1.37.0", features = ["sync", "rt", "net", "time", "macros"] }
-tokio-stream = "0.1.17"
-tonic = { version = "0.14.0", path = "../tonic", default-features = false, features = ["codegen", "transport"] }
-tower = "0.5.2"
+socket2 = { version = "0.5.10", optional = true }
+tokio = { version = "1.37.0", features = ["sync", "macros"] }
+tokio-stream = { version = "0.1.17", default-features = false }
+tonic = { version = "0.14.0", path = "../tonic", default-features = false, features = [
+    "codegen",
+] }
+tower = { version = "0.5.2", features = [
+    "limit",
+    "util",
+    "buffer",
+], optional = true }
 tower-service = "0.3.3"
 url = "2.5.0"
 
 [dev-dependencies]
 async-stream = "0.3.6"
-tonic = { version = "0.14.0", path = "../tonic", default-features = false, features = ["server", "router"] }
-hickory-server =  "0.25.2"
-prost = "0.14"
-
-[features]
-default = ["dns"]
-dns = ["dep:hickory-resolver"]
-
-[package.metadata.cargo_check_external_types]
-allowed_external_types = [
-    "tonic::*",
-    "futures_core::stream::Stream",
-    "tokio::sync::oneshot::Sender",
-]
+hickory-server = "0.25.2"
+prost = "0.14.0"
+tonic = { version = "0.14.0", path = "../tonic", default-features = false, features = [
+    "server",
+    "router",
+] }
+tonic-prost = { version = "0.14.0", path = "../tonic-prost" }

grpc/examples/inmemory.rs

@@ -10,11 +10,8 @@ struct Handler {}
 #[derive(Debug)]
 struct MyReqMessage(String);
 
-impl Message for MyReqMessage {}
-
 #[derive(Debug)]
 struct MyResMessage(String);
-impl Message for MyResMessage {}
 
 #[async_trait]
 impl Service for Handler {

grpc/examples/multiaddr.rs

@@ -12,11 +12,8 @@ struct Handler {
 #[derive(Debug)]
 struct MyReqMessage(String);
 
-impl Message for MyReqMessage {}
-
 #[derive(Debug)]
 struct MyResMessage(String);
-impl Message for MyResMessage {}
 
 #[async_trait]
 impl Service for Handler {

grpc/proto/echo/echo.proto

@@ -0,0 +1,43 @@
+/*
+ *
+ * Copyright 2018 gRPC authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+syntax = "proto3";
+
+package grpc.examples.echo;
+
+// EchoRequest is the request for echo.
+message EchoRequest {
+  string message = 1;
+}
+
+// EchoResponse is the response for echo.
+message EchoResponse {
+  string message = 1;
+}
+
+// Echo is the echo service.
+service Echo {
+  // UnaryEcho is unary echo.
+  rpc UnaryEcho(EchoRequest) returns (EchoResponse) {}
+  // ServerStreamingEcho is server side streaming.
+  rpc ServerStreamingEcho(EchoRequest) returns (stream EchoResponse) {}
+  // ClientStreamingEcho is client side streaming.
+  rpc ClientStreamingEcho(stream EchoRequest) returns (EchoResponse) {}
+  // BidirectionalStreamingEcho is bidi streaming.
+  rpc BidirectionalStreamingEcho(stream EchoRequest) returns (stream EchoResponse) {}
+}

grpc/src/client/channel.rs

@@ -37,17 +37,16 @@ use std::{
 };
 
 use tokio::sync::{mpsc, oneshot, watch, Notify};
-use tokio::task::AbortHandle;
 
 use serde_json::json;
 use tonic::async_trait;
 use url::Url; // NOTE: http::Uri requires non-empty authority portion of URI
 
-use crate::credentials::Credentials;
+use crate::attributes::Attributes;
 use crate::rt;
 use crate::service::{Request, Response, Service};
-use crate::{attributes::Attributes, rt::tokio::TokioRuntime};
 use crate::{client::ConnectivityState, rt::Runtime};
+use crate::{credentials::Credentials, rt::default_runtime};
 
 use super::service_config::ServiceConfig;
 use super::transport::{TransportRegistry, GLOBAL_TRANSPORT_REGISTRY};
@@ -156,7 +155,7 @@ impl Channel {
             inner: Arc::new(PersistentChannel::new(
                 target,
                 credentials,
-                Arc::new(rt::tokio::TokioRuntime {}),
+                default_runtime(),
                 options,
             )),
         }
@@ -262,6 +261,7 @@ impl ActiveChannel {
             tx.clone(),
             picker.clone(),
             connectivity_state.clone(),
+            runtime.clone(),
         );
 
         let resolver_helper = Box::new(tx.clone());
@@ -279,7 +279,7 @@ impl ActiveChannel {
         let resolver_opts = name_resolution::ResolverOptions {
             authority,
             work_scheduler,
-            runtime: Arc::new(TokioRuntime {}),
+            runtime: runtime.clone(),
         };
         let resolver = rb.build(&target, resolver_opts);
 
@@ -360,6 +360,7 @@ pub(crate) struct InternalChannelController {
     wqtx: WorkQueueTx,
     picker: Arc<Watcher<Arc<dyn Picker>>>,
     connectivity_state: Arc<Watcher<ConnectivityState>>,
+    runtime: Arc<dyn Runtime>,
 }
 
 impl InternalChannelController {
@@ -369,8 +370,9 @@ impl InternalChannelController {
         wqtx: WorkQueueTx,
         picker: Arc<Watcher<Arc<dyn Picker>>>,
         connectivity_state: Arc<Watcher<ConnectivityState>>,
+        runtime: Arc<dyn Runtime>,
     ) -> Self {
-        let lb = Arc::new(GracefulSwitchBalancer::new(wqtx.clone()));
+        let lb = Arc::new(GracefulSwitchBalancer::new(wqtx.clone(), runtime.clone()));
 
         Self {
             lb,
@@ -380,6 +382,7 @@ impl InternalChannelController {
             wqtx,
             picker,
             connectivity_state,
+            runtime,
         }
     }
 
@@ -429,6 +432,7 @@ impl load_balancing::ChannelController for InternalChannelController {
             Box::new(move |k: SubchannelKey| {
                 scp.unregister_subchannel(&k);
             }),
+            self.runtime.clone(),
         );
         let _ = self.subchannel_pool.register_subchannel(&key, isc.clone());
         self.new_esc_for_isc(isc)
@@ -454,6 +458,7 @@ pub(super) struct GracefulSwitchBalancer {
     policy_builder: Mutex<Option<Arc<dyn LbPolicyBuilder>>>,
     work_scheduler: WorkQueueTx,
     pending: Mutex<bool>,
+    runtime: Arc<dyn Runtime>,
 }
 
 impl WorkScheduler for GracefulSwitchBalancer {
@@ -478,12 +483,13 @@ impl WorkScheduler for GracefulSwitchBalancer {
 }
 
 impl GracefulSwitchBalancer {
-    fn new(work_scheduler: WorkQueueTx) -> Self {
+    fn new(work_scheduler: WorkQueueTx, runtime: Arc<dyn Runtime>) -> Self {
         Self {
             policy_builder: Mutex::default(),
             policy: Mutex::default(), // new(None::<Box<dyn LbPolicy>>),
             work_scheduler,
             pending: Mutex::default(),
+            runtime,
         }
     }
 
@@ -501,6 +507,7 @@ impl GracefulSwitchBalancer {
             let builder = GLOBAL_LB_REGISTRY.get_policy(policy_name).unwrap();
             let newpol = builder.build(LbPolicyOptions {
                 work_scheduler: self.clone(),
+                runtime: self.runtime.clone(),
             });
             *self.policy_builder.lock().unwrap() = Some(builder);
             *p = Some(newpol);

grpc/src/client/load_balancing/child_manager.rs

@@ -38,6 +38,7 @@ use crate::client::load_balancing::{
     WeakSubchannel, WorkScheduler,
 };
 use crate::client::name_resolution::{Address, ResolverUpdate};
+use crate::rt::Runtime;
 
 use super::{Subchannel, SubchannelState};
 
@@ -47,6 +48,7 @@ pub struct ChildManager<T> {
     children: Vec<Child<T>>,
     update_sharder: Box<dyn ResolverUpdateSharder<T>>,
     pending_work: Arc<Mutex<HashSet<usize>>>,
+    runtime: Arc<dyn Runtime>,
 }
 
 struct Child<T> {
@@ -81,12 +83,16 @@ pub trait ResolverUpdateSharder<T>: Send {
 impl<T> ChildManager<T> {
     /// Creates a new ChildManager LB policy.  shard_update is called whenever a
     /// resolver_update operation occurs.
-    pub fn new(update_sharder: Box<dyn ResolverUpdateSharder<T>>) -> Self {
+    pub fn new(
+        update_sharder: Box<dyn ResolverUpdateSharder<T>>,
+        runtime: Arc<dyn Runtime>,
+    ) -> Self {
         Self {
             update_sharder,
             subchannel_child_map: Default::default(),
             children: Default::default(),
             pending_work: Default::default(),
+            runtime,
         }
     }
 
@@ -197,6 +203,7 @@ impl<T: PartialEq + Hash + Eq + Send + Sync + 'static> LbPolicy for ChildManager
                 });
                 let policy = builder.build(LbPolicyOptions {
                     work_scheduler: work_scheduler.clone(),
+                    runtime: self.runtime.clone(),
                 });
                 let state = LbState::initial();
                 self.children.push(Child {

grpc/src/client/load_balancing/mod.rs

@@ -41,6 +41,7 @@ use tonic::{metadata::MetadataMap, Status};
 
 use crate::{
     client::channel::WorkQueueTx,
+    rt::Runtime,
     service::{Request, Response, Service},
 };
 
@@ -66,6 +67,7 @@ pub struct LbPolicyOptions {
     /// A hook into the channel's work scheduler that allows the LbPolicy to
     /// request the ability to perform operations on the ChannelController.
     pub work_scheduler: Arc<dyn WorkScheduler>,
+    pub runtime: Arc<dyn Runtime>,
 }
 
 /// Used to asynchronously request a call into the LbPolicy's work method if