Implement graceful switch balancer other LB-related changes.

General:

- Add Debug to many traits and derive/impl in structs.
- Pass LB config to LB policies via `Option<LbConfig>` instead of
  `Option<&LbConfig>`.  It should be rare that policies want to store
  a config except for the leaf policy.

Child manager:

The original assumption was that all children would be the same
type/configuration, but several policies (including gracefulswitch)
will not have that property.  So, several changes are made:

- Children are considered unique by both their identifier and their
  LbPolicyBuilder's name().
- Make it so the sharder also can shard LbConfig and provide it via
  the ChildUpdate.child_update field in addition to the
  ResolverUpdate.
- Make ResolverUpdateSharder a generic instead of Box<dyn>.
- Add booleans so users of child manager can easily easily tell
  whether any child policies updated themselves, and which ones did.
- Pass &mut self for sharder so that it can maintain and update its
  state if needed.
- Change the sharder's output ChildUpdate.child_update field to an
  Option; if None then the child will not be called during the
  resolver update, but will remain in the child manager.
- Change child_states into children and provide the whole Child
  struct, exposing the fields it contains.
- Provide mutable access to the sharder.
- Change the LB config to be a flat JSON array to facilitate use
  within another LB policy that should not need a struct to contain on
  the children.
- Minor test cleanups

Graceful switch:

The previous implementation in hyperium#2399 contained a lot of logic to
manage child policy delegation.  It was intended that only
ChildManager should need to have this kind of logic.

- Create a new implementation of this policy that delegates to
  ChildManager.
- Uses a Sharder that simply emits the active policy with no update
  alongside any new policy in the new LbConfig.
- maybe_swap is called after every call into the ChildManager to
  determine if child updates necessitate a swap.
- This logic is simple: if the active policy is not Ready, or if there
  is a new policy and it is not Connecting, then set the new policy as
  the active policy and call resolver_update on the ChildManager.  The
  sharder will see that no LbConfig is provided and just emit the
  active policy with no config, causing the ChildManager to drop the
  previously active policy.  If no swap is needed, update the picker
  of the active policy if it had an update.
- Minor test cleanups/fixes vs. hyperium#2399.

Author: dfawley

grpc/src/client/channel.rs

@@ -453,6 +453,7 @@ impl load_balancing::ChannelController for InternalChannelController {
 }
 
 // A channel that is not idle (connecting, ready, or erroring).
+#[derive(Debug)]
 pub(super) struct GracefulSwitchBalancer {
     pub(super) policy: Mutex<Option<Box<dyn LbPolicy>>>,
     policy_builder: Mutex<Option<Arc<dyn LbPolicyBuilder>>>,
@@ -529,7 +530,7 @@ impl GracefulSwitchBalancer {
 
         p.as_mut()
             .unwrap()
-            .resolver_update(update, config.as_ref(), controller)
+            .resolver_update(update, config, controller)
 
         // TODO: close old LB policy gracefully vs. drop?
     }

grpc/src/client/load_balancing/child_manager.rs

@@ -75,7 +75,7 @@ pub struct LbPolicyOptions {
 /// Used to asynchronously request a call into the LbPolicy's work method if
 /// the LbPolicy needs to provide an update without waiting for an update
 /// from the channel first.
-pub trait WorkScheduler: Send + Sync {
+pub trait WorkScheduler: Send + Sync + Debug {
     // Schedules a call into the LbPolicy's work method.  If there is already a
     // pending work call that has not yet started, this may not schedule another
     // call.
@@ -123,7 +123,7 @@ impl ParsedJsonLbConfig {
 
 /// An LB policy factory that produces LbPolicy instances used by the channel
 /// to manage connections and pick connections for RPCs.
-pub(crate) trait LbPolicyBuilder: Send + Sync {
+pub(crate) trait LbPolicyBuilder: Send + Sync + Debug {
     /// Builds and returns a new LB policy instance.
     ///
     /// Note that build must not fail.  Any optional configuration is delivered
@@ -153,13 +153,13 @@ pub(crate) trait LbPolicyBuilder: Send + Sync {
 /// LB policies are responsible for creating connections (modeled as
 /// Subchannels) and producing Picker instances for picking connections for
 /// RPCs.
-pub trait LbPolicy: Send {
+pub trait LbPolicy: Send + Debug {
     /// Called by the channel when the name resolver produces a new set of
     /// resolved addresses or a new service config.
     fn resolver_update(
         &mut self,
         update: ResolverUpdate,
-        config: Option<&LbConfig>,
+        config: Option<LbConfig>,
         channel_controller: &mut dyn ChannelController,
     ) -> Result<(), Box<dyn Error + Send + Sync>>;
 
@@ -246,7 +246,7 @@ impl Display for SubchannelState {
 ///
 /// If the ConnectivityState is TransientFailure, the Picker should return an
 /// Err with an error that describes why connections are failing.
-pub trait Picker: Send + Sync {
+pub trait Picker: Send + Sync + Debug {
     /// Picks a connection to use for the request.
     ///
     /// This function should not block.  If the Picker needs to do blocking or
@@ -256,6 +256,7 @@ pub trait Picker: Send + Sync {
     fn pick(&self, request: &Request) -> PickResult;
 }
 
+#[derive(Debug)]
 pub enum PickResult {
     /// Indicates the Subchannel in the Pick should be used for the request.
     Pick(Pick),
@@ -317,7 +318,7 @@ impl Display for PickResult {
     }
 }
 /// Data provided by the LB policy.
-#[derive(Clone)]
+#[derive(Clone, Debug)]
 pub struct LbState {
     pub connectivity_state: super::ConnectivityState,
     pub picker: Arc<dyn Picker>,
@@ -347,6 +348,16 @@ pub struct Pick {
     pub on_complete: Option<CompletionCallback>,
 }
 
+impl Debug for Pick {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("Pick")
+            .field("subchannel", &self.subchannel)
+            .field("metadata", &self.metadata)
+            .field("on_complete", &format_args!("{:p}", &self.on_complete))
+            .finish()
+    }
+}
+
 pub trait DynHash {
     #[allow(clippy::redundant_allocation)]
     fn dyn_hash(&self, state: &mut Box<&mut dyn Hasher>);
@@ -438,6 +449,7 @@ impl Display for dyn Subchannel {
     }
 }
 
+#[derive(Debug)]
 struct WeakSubchannel(Weak<dyn Subchannel>);
 
 impl From<Arc<dyn Subchannel>> for WeakSubchannel {
@@ -580,6 +592,7 @@ impl<T: ForwardingSubchannel> private::Sealed for T {}
 
 /// QueuingPicker always returns Queue.  LB policies that are not actively
 /// Connecting should not use this picker.
+#[derive(Debug)]
 pub struct QueuingPicker {}
 
 impl Picker for QueuingPicker {
@@ -588,6 +601,7 @@ impl Picker for QueuingPicker {
     }
 }
 
+#[derive(Debug)]
 pub struct Failing {
     pub error: String,
 }

grpc/src/client/load_balancing/graceful_switch.rs

@@ -23,6 +23,7 @@ use super::{
 
 pub static POLICY_NAME: &str = "pick_first";
 
+#[derive(Debug)]
 struct Builder {}
 
 impl LbPolicyBuilder for Builder {
@@ -44,6 +45,7 @@ pub fn reg() {
     super::GLOBAL_LB_REGISTRY.add_builder(Builder {})
 }
 
+#[derive(Debug)]
 struct PickFirstPolicy {
     work_scheduler: Arc<dyn WorkScheduler>,
     subchannel: Option<Arc<dyn Subchannel>>,
@@ -55,7 +57,7 @@ impl LbPolicy for PickFirstPolicy {
     fn resolver_update(
         &mut self,
         update: ResolverUpdate,
-        config: Option<&LbConfig>,
+        config: Option<LbConfig>,
         channel_controller: &mut dyn ChannelController,
     ) -> Result<(), Box<dyn Error + Send + Sync>> {
         let mut addresses = update
@@ -108,6 +110,7 @@ impl LbPolicy for PickFirstPolicy {
     }
 }
 
+#[derive(Debug)]
 struct OneSubchannelPicker {
     sc: Arc<dyn Subchannel>,
 }
@@ -116,8 +119,9 @@ impl Picker for OneSubchannelPicker {
     fn pick(&self, request: &Request) -> PickResult {
         PickResult::Pick(Pick {
             subchannel: self.sc.clone(),
-            on_complete: None,
+            // on_complete: None,
             metadata: MetadataMap::new(),
+            on_complete: None,
         })
     }
 }

grpc/src/client/load_balancing/mod.rs

@@ -149,6 +149,7 @@ impl ChannelController for TestChannelController {
     }
 }
 
+#[derive(Debug)]
 pub(crate) struct TestWorkScheduler {
     pub(crate) tx_events: mpsc::UnboundedSender<TestEvent>,
 }
@@ -164,7 +165,7 @@ type ResolverUpdateFn = Arc<
     dyn Fn(
             &mut StubPolicyData,
             ResolverUpdate,
-            Option<&LbConfig>,
+            Option<LbConfig>,
             &mut dyn ChannelController,
         ) -> Result<(), Box<dyn Error + Send + Sync>>
         + Send
@@ -186,7 +187,14 @@ pub struct StubPolicyFuncs {
     pub subchannel_update: Option<SubchannelUpdateFn>,
 }
 
+impl Debug for StubPolicyFuncs {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "stub funcs")
+    }
+}
+
 /// Data holds test data that will be passed all to functions in PolicyFuncs
+#[derive(Debug)]
 pub struct StubPolicyData {
     pub test_data: Option<Box<dyn Any + Send + Sync>>,
 }
@@ -199,6 +207,7 @@ impl StubPolicyData {
 }
 
 /// The stub `LbPolicy` that calls the provided functions.
+#[derive(Debug)]
 pub struct StubPolicy {
     funcs: StubPolicyFuncs,
     data: StubPolicyData,
@@ -208,7 +217,7 @@ impl LbPolicy for StubPolicy {
     fn resolver_update(
         &mut self,
         update: ResolverUpdate,
-        config: Option<&LbConfig>,
+        config: Option<LbConfig>,
         channel_controller: &mut dyn ChannelController,
     ) -> Result<(), Box<dyn Error + Send + Sync>> {
         if let Some(f) = &mut self.funcs.resolver_update {
@@ -238,6 +247,7 @@ impl LbPolicy for StubPolicy {
 }
 
 /// StubPolicyBuilder builds a StubLbPolicy.
+#[derive(Debug)]
 pub struct StubPolicyBuilder {
     name: &'static str,
     funcs: StubPolicyFuncs,

grpc/src/client/load_balancing/pick_first.rs

@@ -252,7 +252,7 @@ pub async fn invalid_target() {
         .contains(&target.to_string()));
 }
 
-#[derive(Clone)]
+#[derive(Clone, Debug)]
 struct FakeDns {
     latency: Duration,
     lookup_result: Result<Vec<std::net::IpAddr>, String>,
@@ -270,6 +270,7 @@ impl rt::DnsResolver for FakeDns {
     }
 }
 
+#[derive(Debug)]
 struct FakeRuntime {
     inner: TokioRuntime,
     dns: FakeDns,

grpc/src/client/load_balancing/test_utils.rs

@@ -29,26 +29,21 @@ use std::{any::Any, error::Error, sync::Arc};
 pub(crate) struct ServiceConfig;
 
 /// A convenience wrapper for an LB policy's configuration object.
-#[derive(Debug)]
+#[derive(Debug, Clone)]
 pub(crate) struct LbConfig {
     config: Arc<dyn Any + Send + Sync>,
 }
 
 impl LbConfig {
     /// Create a new LbConfig wrapper containing the provided config.
-    pub fn new<T: 'static + Send + Sync>(config: T) -> Self {
+    pub fn new(config: impl Any + Send + Sync) -> Self {
         LbConfig {
             config: Arc::new(config),
         }
     }
 
     /// Convenience method to extract the LB policy's configuration object.
-    pub fn convert_to<T: 'static + Send + Sync>(
-        &self,
-    ) -> Result<Arc<T>, Box<dyn Error + Send + Sync>> {
-        match self.config.clone().downcast::<T>() {
-            Ok(c) => Ok(c),
-            Err(e) => Err("failed to downcast to config type".into()),
-        }
+    pub fn convert_to<T: 'static + Send + Sync>(&self) -> Option<Arc<T>> {
+        self.config.clone().downcast::<T>().ok()
     }
 }

grpc/src/client/name_resolution/dns/test.rs

@@ -23,6 +23,7 @@
  */
 
 use ::tokio::io::{AsyncRead, AsyncWrite};
+use std::fmt::Debug;
 use std::{future::Future, net::SocketAddr, pin::Pin, sync::Arc, time::Duration};
 
 pub(crate) mod hyper_wrapper;
@@ -39,7 +40,7 @@ pub(crate) type BoxedTaskHandle = Box<dyn TaskHandle>;
 /// time-based operations such as sleeping. It provides a uniform interface
 /// that can be implemented for various async runtimes, enabling pluggable
 /// and testable infrastructure.
-pub(super) trait Runtime: Send + Sync {
+pub(super) trait Runtime: Send + Sync + Debug {
     /// Spawns the given asynchronous task to run in the background.
     fn spawn(&self, task: Pin<Box<dyn Future<Output = ()> + Send + 'static>>) -> BoxedTaskHandle;
 
@@ -98,7 +99,7 @@ pub(crate) trait TcpStream: AsyncRead + AsyncWrite + Send + Unpin {}
 /// # Panics
 ///
 /// Panics if any of its functions are called.
-#[derive(Default)]
+#[derive(Default, Debug)]
 pub(crate) struct NoOpRuntime {}
 
 impl Runtime for NoOpRuntime {

grpc/src/client/service_config.rs

@@ -64,6 +64,7 @@ impl DnsResolver for TokioDefaultDnsResolver {
     }
 }
 
+#[derive(Debug)]
 pub(crate) struct TokioRuntime {}
 
 impl TaskHandle for JoinHandle<()> {