fix child manager

Author: cjqzhao

grpc/src/client/load_balancing/child_manager.rs

@@ -23,6 +23,7 @@
 // policy in use.  Complete tests must be written before it can be used in
 // production.  Also, support for the work scheduler is missing.
 
+use std::{fmt::Debug, sync::atomic::{AtomicUsize, Ordering}};
 use std::collections::HashSet;
 use std::fmt::Display;
 use std::sync::Mutex;
@@ -47,19 +48,14 @@ pub struct ChildManager<T> {
     children: Vec<Child<T>>,
     update_sharder: Box<dyn ResolverUpdateSharder<T>>,
     pending_work: Arc<Mutex<HashSet<usize>>>,
-    updated: bool, // true iff a child has updated its state since the last call to has_updated.
-    // work_requests: Arc<Mutex<HashSet<Arc<T>>>>,
-    // work_scheduler: Arc<dyn WorkScheduler>,
+    updated: bool, // true if a child has updated its state since the last call to has_updated.
     sent_connecting_state: bool,
     aggregated_state: ConnectivityState,
     last_ready_pickers: Vec<Arc<dyn Picker>>,
-    
-}
-
-use std::{sync::{atomic::{AtomicUsize, Ordering}}};
 
-pub trait ChildIdentifier: PartialEq + Hash + Eq + Send + Sync + Display + 'static {}
+}
 
+pub trait ChildIdentifier: PartialEq + Hash + Eq + Send + Sync + Debug + 'static {}
 
 struct Child<T> {
     identifier: T,
@@ -114,13 +110,11 @@ impl<T: ChildIdentifier> ChildManager<T> {
             .iter()
             .map(|child| (&child.identifier, &child.state))
     }
-    
+
     pub fn has_updated(&mut self) -> bool {
         mem::take(&mut self.updated)
     }
 
-
-
     // Called to update all accounting in the ChildManager from operations
     // performed by a child policy on the WrappedController that was created for
     // it.  child_idx is an index into the children map for the relevant child.
@@ -130,7 +124,7 @@ impl<T: ChildIdentifier> ChildManager<T> {
     // which way is better.
     fn resolve_child_controller(
         &mut self,
-        channel_controller: &mut WrappedController,        
+        channel_controller: &mut WrappedController,
         child_idx: usize,
     ) {
         // Add all created subchannels into the subchannel_child_map.
@@ -143,51 +137,47 @@ impl<T: ChildIdentifier> ChildManager<T> {
             self.updated = true;
         };
 
-  
         if self.has_updated() {
             self.aggregate_states(channel_controller);
-
         }
-        
-
     }
+
     // Called to aggregate states from pick first children. Sends a picker to
     // the channel based on aggregation
     fn aggregate_states (& mut self, channel_controller: &mut WrappedController) {
         let current_connectivity_state = self.aggregated_state.clone();
         let child_states_vec: Vec<_> = self.child_states().collect();
         // Constructing pickers to return
-        let mut transient_failure_picker = TransientFailurePickers::new("error string I guess".to_string());
-        let mut connecting_pickers = ConnectingPickers::new();
+        // let mut transient_failure_picker = TransientFailurePickers::new("error string I guess".to_string());
+        // let mut connecting_pickers = ConnectingPickers::new();
         let mut ready_pickers = ReadyPickers::new();
 
         let mut has_idle = false;
+        let mut has_connecting = false;
         let mut is_transient_failure = true;
 
         for (child_id, state) in &child_states_vec {
             match state.connectivity_state {
                 ConnectivityState::Idle => {
-                    connecting_pickers.add_picker(state.picker.clone());
                     has_idle = true;
+                    has_connecting = true;
                     is_transient_failure = false;
                 }
                 ConnectivityState::Connecting  => {
-                    connecting_pickers.add_picker(state.picker.clone());
+                    has_connecting = true;
                     is_transient_failure = false;
                 }
                 ConnectivityState::Ready => {
                     ready_pickers.add_picker(state.picker.clone());
                     is_transient_failure = false;
                 }
-                ConnectivityState::TransientFailure =>{
-                    transient_failure_picker.add_picker(state.picker.clone());
+                _ =>{
+
                 }
             }
         }
 
         let has_ready = ready_pickers.has_any();
-        let has_connecting = connecting_pickers.pickers.len() >= 1;
-        
 
         // Decide the new aggregate state
         let new_state = if has_ready {
@@ -202,16 +192,13 @@ impl<T: ChildIdentifier> ChildManager<T> {
 
         self.aggregated_state = new_state;
 
-       
-
         // Now update state and send picker as appropriate
         match new_state {
             ConnectivityState::Ready => {
                 let pickers_vec = ready_pickers.pickers.clone();
                 let picker: Arc<dyn Picker> = Arc::new(ready_pickers);
                 let should_update = !self.compare_prev_to_new_pickers(&self.last_ready_pickers, &pickers_vec);
 
-              
                 if should_update || self.aggregated_state != ConnectivityState::Ready {
                     println!("child manager sends ready picker update");
                     self.aggregated_state = ConnectivityState::Ready;
@@ -234,14 +221,13 @@ impl<T: ChildIdentifier> ChildManager<T> {
                 }
                 if !self.sent_connecting_state {
                     println!("child manager sends connecting picker update");
-                    
+
                     channel_controller.need_to_reach();
                     let picker = Arc::new(QueuingPicker{});
                     self.move_to_connecting(has_idle, channel_controller, picker);
                 } else{
                     println!("child manager is not sending connecting picker as it has alrady sent connecting state");
                 }
-                // Don't send another Connecting picker if already sent
             }
             ConnectivityState::Idle => {
                 self.aggregated_state = ConnectivityState::Connecting;
@@ -251,57 +237,48 @@ impl<T: ChildIdentifier> ChildManager<T> {
                     connectivity_state: ConnectivityState::Connecting,
                     picker,
                 });
-                
                 self.sent_connecting_state = true;
             }
             ConnectivityState::TransientFailure => {
                 if current_connectivity_state != ConnectivityState::TransientFailure{
                     println!("child manager sends transient failure picker update");
-                    let picker = Arc::new(transient_failure_picker);
                     channel_controller.need_to_reach();
+                    let picker = Arc::new(Failing {error: "No children available".to_string()});
                     self.move_to_transient_failure(channel_controller, picker);
                 }
             }
-   
+
         }
     }
-    
-    
 }
 
 impl<T: ChildIdentifier> ChildManager<T> {
     fn move_to_transient_failure(&mut self, channel_controller: &mut dyn ChannelController, picker: Arc<dyn Picker>) {
         self.aggregated_state = ConnectivityState::TransientFailure;
         channel_controller.update_picker(LbState {
-                connectivity_state: ConnectivityState::TransientFailure,
+            connectivity_state: ConnectivityState::TransientFailure,
                picker: picker,
             });
         println!("requesting resolution");
         channel_controller.request_resolution();
         self.sent_connecting_state = false;
     }
 
-
-
     fn move_to_connecting(&mut self, is_idle: bool, channel_controller: &mut dyn ChannelController, picker: Arc<dyn Picker>) {
         self.aggregated_state = ConnectivityState::Connecting;
         channel_controller.update_picker(LbState {
             connectivity_state: ConnectivityState::Connecting,
             picker: picker,
         });
         self.sent_connecting_state = true;
-        // if is_idle {
-        //     println!("requesting resolution");
-        //     channel_controller.request_resolution();
-        // }
     }
 
     fn compare_prev_to_new_pickers(& self, old_pickers: &[Arc<dyn Picker>], new_pickers: &[Arc<dyn Picker>]) -> bool {
         //if length is different, then definitely not the same picker
         if old_pickers.len() != new_pickers.len() {
             return false;
         }
-        //compares two vectors of pickers by pointer equality and returns true if all pickers are the same 
+        //compares two vectors of pickers by pointer equality and returns true if all pickers are the same
         for (x, y) in old_pickers.iter().zip(new_pickers.iter()) {
             if !Arc::ptr_eq(x, y) {
                 return false;
@@ -423,7 +400,6 @@ impl<T: ChildIdentifier> LbPolicy for ChildManager<T> {
         }
         Ok(())
     }
-    
 
     fn subchannel_update(
         &mut self,
@@ -454,14 +430,16 @@ impl<T: ChildIdentifier> LbPolicy for ChildManager<T> {
             self.resolve_child_controller(&mut channel_controller, child_idx);
         }
     }
-    
+
     fn exit_idle(&mut self, channel_controller: &mut dyn ChannelController) {
-        todo!()
-        // let policy = &mut self.children.get_mut(&child_id.clone()).unwrap().policy;
-        // let mut channel_controller = WrappedController::new(channel_controller);
-        // // Call the proper child.
-        // policy.exit_idle(&mut channel_controller);
-        // self.resolve_child_controller(channel_controller, child_id.clone());
+        let child_idxes = mem::take(&mut *self.pending_work.lock().unwrap());
+        for child_idx in child_idxes {
+            let mut channel_controller = WrappedController::new(channel_controller);
+            self.children[child_idx]
+                .policy
+                .exit_idle(&mut channel_controller);
+            self.resolve_child_controller(&mut channel_controller, child_idx);
+        }
     }
 }
 
@@ -495,12 +473,12 @@ impl ChannelController for WrappedController<'_> {
     }
 
     fn update_picker(&mut self, update: LbState) {
-            self.picker_update = Some(update.clone());
-            if self.need_to_reach{
-                self.channel_controller.update_picker(update);
-                self.need_to_reach = false;
-            }
+        self.picker_update = Some(update.clone());
+        if self.need_to_reach{
+            self.channel_controller.update_picker(update);
+            self.need_to_reach = false;
         }
+    }
 
     fn request_resolution(&mut self) {
         self.channel_controller.request_resolution();
@@ -526,8 +504,6 @@ struct ReadyPickers {
     next: AtomicUsize,
 }
 
-
-
 impl ReadyPickers {
     pub fn new() -> Self {
         Self {
@@ -543,8 +519,6 @@ impl ReadyPickers {
     pub fn has_any(&self) -> bool {
         !self.pickers.is_empty()
     }
-
-    
 }
 
 impl Picker for ReadyPickers {
@@ -556,74 +530,5 @@ impl Picker for ReadyPickers {
         let idx = self.next.fetch_add(1, Ordering::Relaxed) % len;
         self.pickers[idx].pick(request)
     }
-
-    
-}
-struct ConnectingPickers {
-    pickers: Vec<Arc<dyn Picker>>,
-    // next: AtomicUsize,
-}
-
-impl Picker for ConnectingPickers {
-    fn pick(&self, request: &Request) -> PickResult {
-        
-        return PickResult::Queue;
-        
-    }
-}
-
-impl ConnectingPickers {
-    pub fn new() -> Self {
-        Self {
-            pickers: vec![],
-        }
-    }
-
-    fn add_picker(&mut self, picker: Arc<dyn Picker>)  {
-        self.pickers.push(picker);
-    }
-}
-
-struct TransientFailurePickers {
-    pickers: Vec<Arc<dyn Picker>>,
-    error: String,    
-
-}
-
-impl Picker for TransientFailurePickers {
-    fn pick(&self, request: &Request) -> PickResult {
-        
-        return PickResult::Fail(Status::unavailable(self.error.clone()))
-        
-    }
-}
-
-impl TransientFailurePickers {
-    pub fn new(error: String) -> Self {
-        Self {
-            pickers: vec![],
-            error,
-        }
-    }
-
-    fn add_picker(&mut self, picker: Arc<dyn Picker>)  {
-        self.pickers.push(picker);
-    }
-}
-struct IdlePickers {
-    pickers: Vec<Arc<dyn Picker>>,
-    work_scheduler: Arc<dyn WorkScheduler>,
-    // next: AtomicUsize,
-}
-
-struct SchedulingIdlePicker {
-    work_scheduler: Arc<dyn WorkScheduler>,
-}
-
-impl Picker for SchedulingIdlePicker {
-    fn pick(&self, _request: &Request) -> PickResult {
-        self.work_scheduler.schedule_work();
-        PickResult::Queue
-    }
 }