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

Code before:
        // 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
            // 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"; "processed_regions" => processed_regions);

                    processed_regions = 0;
                    // Set the metric to zero to show that rewinding has happened.
                    AUTO_GC_PROCESSED_REGIONS_GAUGE_VEC
                        .with_label_values(&[PROCESS_TYPE_GC])

Review comment (suggestion):
Actually, I think, there should be some way to block it on the task sender which is consumed by the GCWorker or GCRunner, in which way the code should be able to be more clear 🤔

Code after:
        // 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);

                    scheduled_regions = 0;
                    // Set the metric to zero to show that rewinding has happened.
                    AUTO_GC_PROCESSED_REGIONS_GAUGE_VEC
                        .with_label_values(&[PROCESS_TYPE_GC])
                        .set(0);
                    AUTO_GC_PROCESSED_REGIONS_GAUGE_VEC
                        .with_label_values(&[PROCESS_TYPE_SCAN])
                        .set(0);
                }
            } else {
                // We are not going to rewind, So we will stop if `progress` reaches `end`.