Repository: tikv/tikv
File: src/server/gc_worker/gc_manager.rs (Rust)

Code before:
    /// ```
    ///
    /// Then GC finishes.
    /// If safe point updates again at some time, it will still try to GC all
    /// regions with the latest safe point. If safe point always updates
    /// before `gc_a_round` finishes, `gc_a_round` may never stop, but it
    /// doesn't matter.
    fn gc_a_round(&mut self) -> GcManagerResult<()> {
        let mut need_rewind = false;
        // Represents where we should stop doing GC. `None` means the very end of the
        // TiKV.
        let mut end = None;
        // Represents where we have GC-ed to. `None` means the very end of the TiKV.
        let mut progress = Some(Key::from_encoded(BEGIN_KEY.to_vec()));

        // Records how many region we have GC-ed.
        let mut processed_regions = 0;

        info!("gc_worker: auto gc starts"; "safe_point" => self.curr_safe_point());

        // The following loop iterates all regions whose leader is on this TiKV and does
        // GC on them. At the same time, check whether safe_point is updated
        // periodically. If it's updated, rewinding will happen.

        let (tx, rx) = mpsc::channel();
        let cb = Box::new(move |res| {
            // send must be successful.
            tx.send(res).unwrap();
        });
        let mut running_tasks = 0;
        loop {
            self.gc_manager_ctx.check_stopped()?;
            if is_compaction_filter_allowed(&self.cfg_tracker.value(), &self.feature_gate) {
                return Ok(());
            }

            if running_tasks >= self.max_concurrent_tasks {
                let res = rx.recv().unwrap();
                running_tasks -= 1;
                if let Err(e) = res {
                    // Ignore the error and continue, since it's useless to retry this.
                    // TODO: Find a better way to handle errors. Maybe we should retry.
                    warn!("failed gc"; "err" => ?e);
                }
                processed_regions += 1;
                AUTO_GC_PROCESSED_REGIONS_GAUGE_VEC
                    .with_label_values(&[PROCESS_TYPE_GC])
                    .inc();
            }

            // Check the current GC progress and determine if we are going to rewind or we

Review comment (suggestion):
how about using https://docs.rs/crossbeam/latest/crossbeam/sync/struct.WaitGroup.html or `CallbackWaitGroup` of backup-stream util

Code after:
    /// ```
    ///
    /// Then GC finishes.
    /// If safe point updates again at some time, it will still try to GC all
    /// regions with the latest safe point. If safe point always updates
    /// before `gc_a_round` finishes, `gc_a_round` may never stop, but it
    /// doesn't matter.
    fn gc_a_round(&mut self) -> GcManagerResult<()> {
        let mut need_rewind = false;
        // Represents where we should stop doing GC. `None` means the very end of the
        // TiKV.
        let mut end = None;
        // Represents where we have GC-ed to. `None` means the very end of the TiKV.
        let mut progress = Some(Key::from_encoded(BEGIN_KEY.to_vec()));

        // Records how many region we have GC-ed.
        let mut scheduled_regions = 0;
        let task_controller = Arc::new((Mutex::new(0), Condvar::new()));
        // the task_controller is the <mutex,Condvar> combination to control the number
        // of tasks The mutex is used for protecting the number of current
        // tasks, while the condvar is used for notifying/get notified when the
        // number of current tasks is changed.
        let (lock, cvar) = &*task_controller;
        let maybe_wait = |max_tasks| {
            let mut current_tasks: std::sync::MutexGuard<'_, usize> = lock.lock().unwrap();
            while *current_tasks > max_tasks {
                // Wait until the number of current tasks is below the limit
                current_tasks = cvar.wait(current_tasks).unwrap();
            }
        };

        info!("gc_worker: auto gc starts"; "safe_point" => self.curr_safe_point());

        // The following loop iterates all regions whose leader is on this TiKV and does
        // GC on them. At the same time, check whether safe_point is updated
        // periodically. If it's updated, rewinding will happen.

        loop {
            self.gc_manager_ctx.check_stopped()?;
            if is_compaction_filter_allowed(&self.cfg_tracker.value(), &self.feature_gate) {
                return Ok(());
            }

            // Check the current GC progress and determine if we are going to rewind or we
            // have finished the round of GC.
            if need_rewind {
                if progress.is_none() {
                    // We have worked to the end and we need to rewind. Restart from beginning.
                    progress = Some(Key::from_encoded(BEGIN_KEY.to_vec()));
                    need_rewind = false;
                    info!("gc_worker: auto gc rewinds"; "scheduled_regions" => scheduled_regions);