Repository: tikv/tikv
File: components/resolved_ts/src/scanner.rs (Rust)

Code before:
            _phantom: PhantomData::default(),
        }
    }

    pub fn spawn_task(&self, mut task: ScanTask, concurrency_semaphore: Arc<Semaphore>) {
        let cdc_handle = self.cdc_handle.clone();
        let fut = async move {
            if let Some(backoff) = task.backoff {
                RTS_INITIAL_SCAN_BACKOFF_DURATION_HISTOGRAM.observe(backoff.as_secs_f64());
                let mut backoff = GLOBAL_TIMER_HANDLE
                    .delay(std::time::Instant::now() + backoff)
                    .compat()
                    .fuse();
                futures::select! {
                    res = backoff => if let Err(e) = res {
                        error!("failed to backoff"; "err" => ?e);
                    },
                    _ = &mut task.cancelled => {}
                }
                if task.is_cancelled() {
                    return;
                }
            }
            let _permit = concurrency_semaphore.acquire().await;
            if task.is_cancelled() {
                return;
            }
            fail::fail_point!("resolved_ts_before_scanner_get_snapshot");
            let snap = match Self::get_snapshot(&mut task, cdc_handle).await {
                Ok(snap) => snap,
                Err(e) => {
                    warn!("resolved_ts scan get snapshot failed"; "err" => ?e);
                    task.on_error(e);
                    return;
                }
            };
            fail::fail_point!("resolved_ts_after_scanner_get_snapshot");
            let start = Instant::now();
            let apply_index = snap.get_apply_index().unwrap();
            let mut reader = MvccReader::new(snap, Some(MvccScanMode::Forward), false);
            let mut done = false;
            let mut start_key = None;
            while !done && !task.is_cancelled() {
                let (locks, has_remaining) =
                    match Self::scan_locks(&mut reader, start_key.as_ref(), task.checkpoint_ts) {
                        Ok(rs) => rs,
                        Err(e) => {
                            warn!("resolved_ts scan lock failed"; "err" => ?e);
                            task.on_error(e);
                            return;
                        }

Review comment (suggestion):
Maybe it would be better to have a metrics about pending task count which is used to see if the concurrency is enough.  Or if we already have pending task count metrics, then we can re-use that one. 

Code after:
        }
    }

    pub fn spawn_task(&self, mut task: ScanTask, concurrency_semaphore: Arc<Semaphore>) {
        let cdc_handle = self.cdc_handle.clone();
        let fut = async move {
            tikv_util::defer!({
                RTS_SCAN_TASKS.with_label_values(&["finish"]).inc();
            });
            if let Some(backoff) = task.backoff {
                RTS_INITIAL_SCAN_BACKOFF_DURATION_HISTOGRAM.observe(backoff.as_secs_f64());
                let mut backoff = GLOBAL_TIMER_HANDLE
                    .delay(std::time::Instant::now() + backoff)
                    .compat()
                    .fuse();
                futures::select! {
                    res = backoff => if let Err(e) = res {
                        error!("failed to backoff"; "err" => ?e);
                    },
                    _ = &mut task.cancelled => {}
                }
                if task.is_cancelled() {
                    return;
                }
            }
            let _permit = concurrency_semaphore.acquire().await;
            if task.is_cancelled() {
                return;
            }
            fail::fail_point!("resolved_ts_before_scanner_get_snapshot");
            let snap = match Self::get_snapshot(&mut task, cdc_handle).await {
                Ok(snap) => snap,
                Err(e) => {
                    warn!("resolved_ts scan get snapshot failed"; "err" => ?e);
                    task.on_error(e);
                    return;
                }
            };
            fail::fail_point!("resolved_ts_after_scanner_get_snapshot");
            let start = Instant::now();
            let apply_index = snap.get_apply_index().unwrap();
            let mut reader = MvccReader::new(snap, Some(MvccScanMode::Forward), false);
            let mut done = false;
            let mut start_key = None;
            while !done && !task.is_cancelled() {
                let (locks, has_remaining) =
                    match Self::scan_locks(&mut reader, start_key.as_ref(), task.checkpoint_ts) {
                        Ok(rs) => rs,
                        Err(e) => {
                            warn!("resolved_ts scan lock failed"; "err" => ?e);
                            task.on_error(e);