feat(grpc): add round_robin LB policy (#2451)

This change is based heavily on #2405, especially the tests.

Author: dfawley

grpc/src/client/load_balancing/child_manager.rs

@@ -30,6 +30,7 @@
 // production.
 
 use std::collections::HashSet;
+use std::error::Error;
 use std::fmt::Debug;
 use std::sync::Mutex;
 use std::{collections::HashMap, hash::Hash, mem, sync::Arc};
@@ -287,9 +288,10 @@ where
         &mut self,
         child_updates: impl IntoIterator<Item = ChildUpdate<T>>,
         channel_controller: &mut dyn ChannelController,
-    ) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+    ) -> Result<(), Box<dyn Error + Send + Sync>> {
         // Split the child updates into the IDs and builders, and the
         // ResolverUpdates/LbConfigs.
+        let mut errs = vec![];
         let (ids_builders, updates): (Vec<_>, Vec<_>) = child_updates
             .into_iter()
             .map(|e| ((e.child_identifier, e.child_policy_builder), e.child_update))
@@ -306,14 +308,59 @@ where
                 continue;
             };
             let mut channel_controller = WrappedController::new(channel_controller);
-            let _ = child.policy.resolver_update(
+            if let Err(err) = child.policy.resolver_update(
                 resolver_update,
                 config.as_ref(),
                 &mut channel_controller,
-            );
+            ) {
+                errs.push(err);
+            }
             self.resolve_child_controller(channel_controller, child_idx);
         }
-        Ok(())
+        if errs.is_empty() {
+            Ok(())
+        } else {
+            let err = errs
+                .into_iter()
+                .map(|e| e.to_string())
+                .collect::<Vec<_>>()
+                .join("; ");
+            Err(err.into())
+        }
+    }
+
+    /// Forwards the `resolver_update` and `config` to all current children.
+    ///
+    /// Returns the Result from calling into each child.
+    pub fn resolver_update(
+        &mut self,
+        resolver_update: ResolverUpdate,
+        config: Option<&LbConfig>,
+        channel_controller: &mut dyn ChannelController,
+    ) -> Result<(), Box<dyn Error + Send + Sync>> {
+        let mut errs = Vec::with_capacity(self.children.len());
+        for child_idx in 0..self.children.len() {
+            let child = &mut self.children[child_idx];
+            let mut channel_controller = WrappedController::new(channel_controller);
+            if let Err(err) = child.policy.resolver_update(
+                resolver_update.clone(),
+                config,
+                &mut channel_controller,
+            ) {
+                errs.push(err);
+            }
+            self.resolve_child_controller(channel_controller, child_idx);
+        }
+        if errs.is_empty() {
+            Ok(())
+        } else {
+            let err = errs
+                .into_iter()
+                .map(|e| e.to_string())
+                .collect::<Vec<_>>()
+                .join("; ");
+            Err(err.into())
+        }
     }
 
     /// Forwards the incoming subchannel_update to the child that created the
@@ -434,6 +481,7 @@ mod test {
     use crate::client::ConnectivityState;
     use crate::rt::default_runtime;
     use std::collections::HashMap;
+    use std::error::Error;
     use std::panic;
     use std::sync::Arc;
     use std::sync::Mutex;
@@ -498,7 +546,7 @@ mod test {
         endpoints: Vec<Endpoint>,
         builder: Arc<dyn LbPolicyBuilder>,
         tcc: &mut dyn ChannelController,
-    ) {
+    ) -> Result<(), Box<dyn Error + Send + Sync>> {
         let updates = endpoints.iter().map(|e| ChildUpdate {
             child_identifier: e.clone(),
             child_policy_builder: builder.clone(),
@@ -513,7 +561,7 @@ mod test {
             )),
         });
 
-        assert!(child_manager.update(updates, tcc).is_ok());
+        child_manager.update(updates, tcc)
     }
 
     fn move_subchannel_to_state(
@@ -595,7 +643,8 @@ mod test {
             endpoints.clone(),
             builder,
             tcc.as_mut(),
-        );
+        )
+        .unwrap();
         let mut subchannels = vec![];
         for endpoint in endpoints {
             subchannels.push(
@@ -648,7 +697,8 @@ mod test {
             endpoints.clone(),
             builder,
             tcc.as_mut(),
-        );
+        )
+        .unwrap();
         let mut subchannels = vec![];
         for endpoint in endpoints {
             subchannels.push(
@@ -699,7 +749,8 @@ mod test {
             endpoints.clone(),
             builder,
             tcc.as_mut(),
-        );
+        )
+        .unwrap();
         let mut subchannels = vec![];
         for endpoint in endpoints {
             subchannels.push(
@@ -740,7 +791,8 @@ mod test {
             endpoints.clone(),
             builder,
             tcc.as_mut(),
-        );
+        )
+        .unwrap();
         let mut subchannels = vec![];
         for endpoint in endpoints {
             subchannels.push(

grpc/src/client/load_balancing/graceful_switch.rs

@@ -26,7 +26,7 @@ struct GracefulSwitchLbConfig {
 /// to to active and tear down the previously active policy.
 #[derive(Debug)]
 pub(crate) struct GracefulSwitchPolicy {
-    child_manager: ChildManager<()>, // Child ID is the name of the child policy.
+    child_manager: ChildManager<()>, // Child ID empty - only the name of the child LB policy matters.
     last_update: Option<LbState>, // Saves the last output LbState to determine if an update is needed.
     active_child_builder: Option<Arc<dyn LbPolicyBuilder>>,
 }
@@ -69,11 +69,9 @@ impl LbPolicy for GracefulSwitchPolicy {
             });
         }
 
-        let res = self
-            .child_manager
-            .update(children.into_iter(), channel_controller)?;
+        let res = self.child_manager.update(children, channel_controller);
         self.update_picker(channel_controller);
-        Ok(())
+        res
     }
 
     fn subchannel_update(

grpc/src/client/load_balancing/mod.rs

@@ -49,6 +49,7 @@ use crate::client::{
 pub(crate) mod child_manager;
 pub(crate) mod graceful_switch;
 pub(crate) mod pick_first;
+pub(crate) mod round_robin;
 
 #[cfg(test)]
 pub(crate) mod test_utils;
@@ -606,11 +607,11 @@ impl Picker for QueuingPicker {
 }
 
 #[derive(Debug)]
-pub(crate) struct Failing {
+pub(crate) struct FailingPicker {
     pub error: String,
 }
 
-impl Picker for Failing {
+impl Picker for FailingPicker {
     fn pick(&self, _: &Request) -> PickResult {
         PickResult::Fail(Status::unavailable(self.error.clone()))
     }