You are an expert in writing code in many different languages. Your goal is to perform code completion for the following code, keeping in mind the rest of the code and the file meta data.
    Metadata details:
    filename: scanner.rs
    path of file: ./repos/tikv/components/tidb_query_common/src/storage
    
    the code of the file until where you have to start completion: // Copyright 2019 TiKV Project Authors. Licensed under Apache-2.0.

use std::{marker::PhantomData, time::Duration};

use api_version::KvFormat;
use tikv_util::time::Instant;
use yatp::task::future::reschedule;

use super::{range::*, ranges_iter::*, OwnedKvPair, Storage};
use crate::error::StorageError;

const KEY_BUFFER_CAPACITY: usize = 64;
/// Batch executors are run in coroutines. `MAX_TIME_SLICE` is the maximum time
/// a coroutine can run without being yielded.
const MAX_TIME_SLICE: Duration = Duration::from_millis(1);
/// the number of scanned keys that should trigger a reschedule.
const CHECK_KEYS: usize = 32;

/// A scanner that scans over multiple ranges. Each range can be a point range
/// containing only one row, or an interval range containing multiple rows.
pub struct RangesScanner<T, F> {
    storage: T,
    ranges_iter: RangesIterator,

    scan_backward_in_range: bool,
    is_key_only: bool,

    scanned_rows_per_range: Vec<usize>,

    // The following fields are only used for calculating scanned range. Scanned range is only
    // useful in streaming mode, where the client need to know the underlying physical data range
    // of each response slice, so that partial retry can be non-overlapping.
    is_scanned_range_aware: bool,
    current_range: IntervalRange,
    working_range_begin_key: Vec<u8>,
    working_range_end_key: Vec<u8>,
    rescheduler: RescheduleChecker,

    _phantom: PhantomData<F>,
}

// TODO: maybe it's better to make it generic to avoid directly depending
// on yatp's rescheduler.
struct RescheduleChecker {
    prev_start: Instant,
    prev_key_count: usize,
}

impl RescheduleChecker {
    fn new() -> Self {
        Self {
            prev_start: Instant::now(),
            prev_key_count: 0,
        }
    }

    #[inline(always)]
    async fn check_reschedule(&mut self, force_check: bool) {
        self.prev_key_count += 1;
        if (force_check || self.prev_key_count % CHECK_KEYS == 0)
            && self.prev_start.saturating_elapsed() > MAX_TIME_SLICE
        {
            reschedule().await;
            self.prev_start = Instant::now();
            self.prev_key_count = 0;
        }
    }
}

pub struct RangesScannerOptions<T> {
    pub storage: T,
    pub ranges: Vec<Range>,
    pub scan_backward_in_range: bool, // TODO: This can be const generics
    pub is_key_only: bool,            // TODO: This can be const generics
    pub is_scanned_range_aware: bool, // TODO: This can be const generics
}

impl<T: Storage, F: KvFormat> RangesScanner<T, F> {
    pub fn new(
        RangesScannerOptions {
            storage,
    
    fn test_scanned_range_forward2() {
        let storage = create_storage();
        // Filled interval range
        let ranges = vec![IntervalRange::from(("foo", "foo_8")).into()];
        let mut scanner = RangesScanner::<_, ApiV1>::new(RangesScannerOptions {
            storage: storage.clone(),
            ranges,
            scan_backward_in_range: false,
            is_key_only: false,
            is_scanned_range_aware: true,
        });

        // Only lower_inclusive is updated.
        assert_eq!(
            &block_on(scanner.next_opt(false)).unwrap().unwrap().key(),
            b"foo"
        );
        assert_eq!(&scanner.working_range_begin_key, b"foo");
        assert_eq!(&scanner.working_range_end_key, b"");

        // Upper_exclusive is updated.
        assert_eq!(
            &block_on(scanner.next_opt(true)).unwrap().unwrap().key(),
            b"foo_2"
        );
        assert_eq!(&scanner.working_range_begin_key, b"foo");
        assert_eq!(&scanner.working_range_end_key, b"foo_2\0");

        // Upper_exclusive is not updated.
        assert_eq!(
            &block_on(scanner.next_opt(false)).unwrap().unwrap().key(),
            b"foo_3"
        );
        assert_eq!(&scanner.working_range_begin_key, b"foo");
        assert_eq!(&scanner.working_range_end_key, b"foo_2\0");

        // Drained.
        assert_eq!(block_on(scanner.next_opt(false)).unwrap(), None);
        assert_eq!(&scanner.working_range_begin_key, b"foo");
        assert_eq!(&scanner.working_range_end_key, b"foo_8");

        let r = scanner.take_scanned_range();
        assert_eq!(&r.lower_inclusive, b"foo");
        assert_eq!(&r.upper_exclusive, b"foo_8");

        // Multiple ranges
        // TODO: caller should not pass in unordered ranges otherwise scanned ranges
        // would be unsound.
        let ranges = vec![
            IntervalRange::from(("foo", "foo_3")).into(),
            IntervalRange::from(("foo_5", "foo_50")).into(),
            IntervalRange::from(("bar", "bar_")).into(),
            PointRange::from("bar_2").into(),
            PointRange::from("bar_3").into(),
            IntervalRange::from(("bar_4", "box")).into(),
        ];
        let mut scanner = RangesScanner::<_, ApiV1>::new(RangesScannerOptions {
            storage,
            ranges,
            scan_backward_in_range: false,
            is_key_only: false,
            is_scanned_range_aware: true,
        });

        // Only lower_inclusive is updated.
        assert_eq!(
            &block_on(scanner.next_opt(false)).unwrap().unwrap().key(),
            b"foo"
        );
        assert_eq!(&scanner.working_range_begin_key, b"foo");
        assert_eq!(&scanner.working_range_end_key, b"");

        // Upper_exclusive is updated. Updated by scanned row.
        assert_eq!(
            &block_on(scanner.next_opt(true)).unwrap().unwrap().key(),
            b"foo_2"
        );
        assert_eq!(&scanner.working_range_begin_key, b"foo");
        assert_eq!(&scanner.working_range_end_key, b"foo_2\0");

        // Upper_exclusive is not updated.
        assert_eq!(
            &block_on(scanner.next_opt(false)).unwrap().unwrap().key(),
            b"bar"
        );
        assert_eq!(&scanner.working_range_begin_key, b"foo");
        assert_eq!(&scanner.working_range_end_key, b"foo_2\0");

        // Upper_exclusive is not updated.
        assert_eq!(
            &block_on(scanner.next_opt(false)).unwrap().unwrap().key(),
            b"bar_2"
        );
        assert_eq!(&scanner.working_range_begin_key, b"foo");
        assert_eq!(&scanner.working_range_end_key, b"foo_2\0");

        // Drain.
        assert_eq!(block_on(scanner.next_opt(false)).unwrap(), None);
        assert_eq!(&scanner.working_range_begin_key, b"foo");
        assert_eq!(&scanner.working_range_end_key, b"box");

        let r = scanner.take_scanned_range();
        assert_eq!(&r.lower_inclusive, b"foo");
        assert_eq!(&r.upper_exclusive, b"box");
    }