Repository: tikv/tikv
File: src/storage/txn/flow_controller/singleton_flow_controller.rs (Rust)

Code before:
impl EngineFlowController {
    // only for test
    pub fn empty() -> Self {
        Self {
            discard_ratio: Arc::new(AtomicU32::new(0)),
            limiter: Arc::new(Limiter::new(f64::INFINITY)),
            tx: None,
            handle: None,
            config_tracker: Arc::new(VersionTrack::default()),
        }
    }

    pub fn new<E: CfNamesExt + FlowControlFactorsExt + Send + 'static>(
        config: &FlowControlConfig,
        engine: E,
        flow_info_receiver: Receiver<FlowInfo>,
    ) -> Self {
        let limiter = Arc::new(
            <Limiter>::builder(f64::INFINITY)
                .refill(Duration::from_millis(1))
                // min_wait is used in the `consume_duration` method of `Limiter`.
                // The final delay duration is calculated by the `consume` method of `Bucket`.
                // Specifically, the delay is computed as: self.min_wait - self.value / self.speed_limit.
                // Here, `self.value` represents the current accumulated tokens (i.e., the amount of traffic that has exceeded the speed limit and needs to be throttled).
                // By setting min_wait to 0, we let TiKV's speed_limit parameter fully control the delay duration, making the flow control logic simpler and the write speed smoother.
                .min_wait(Duration::from_millis(0))
                .build(),
        );
        let discard_ratio = Arc::new(AtomicU32::new(0));
        let config_tracker = Arc::new(VersionTrack::new(config.clone()));
        let checker = FlowChecker::new(
            config_tracker.clone(),
            engine,
            discard_ratio.clone(),
            limiter.clone(),
        );
        let (tx, rx) = mpsc::sync_channel(5);

        Self {
            discard_ratio,
            limiter,
            tx: Some(tx),
            handle: Some(checker.start(rx, flow_info_receiver)),
            config_tracker,
        }
    }
}

impl EngineFlowController {
    pub fn should_drop(&self, _region_id: u64) -> bool {
        let ratio = self.discard_ratio.load(Ordering::Relaxed);

Review comment (performance):
I suggest set the minimum duration to `1ms` to potential corner case the frequent small wait may impact the overall performance when the cpu usage is very high.

Code after:
impl EngineFlowController {
    // only for test
    pub fn empty() -> Self {
        Self {
            discard_ratio: Arc::new(AtomicU32::new(0)),
            limiter: Arc::new(Limiter::new(f64::INFINITY)),
            tx: None,
            handle: None,
            config_tracker: Arc::new(VersionTrack::default()),
        }
    }

    pub fn new<E: CfNamesExt + FlowControlFactorsExt + Send + 'static>(
        config: &FlowControlConfig,
        engine: E,
        flow_info_receiver: Receiver<FlowInfo>,
    ) -> Self {
        let limiter = Arc::new(
            <Limiter>::builder(f64::INFINITY)
                .refill(Duration::from_millis(1))
                // `min_wait` is used in the `consume_duration` method of `Limiter`. The final delay
                // duration is calculated in the `consume` method of `Bucket`. The formula is:
                //     self.min_wait - self.value / self.speed_limit.
                // where `self.value`` represents the current byte quota for write operations; it is
                // initialized to the maximum allowed value and is decreased by the size of each 
                // write. When the value becomes negative, write requests will be delayed.
                // 
                // By setting min_wait to 1, we can ensure flow control more smooth while avoid the
                // overhead caused by many short sleep calls.
                .min_wait(Duration::from_millis(1))
                .build(),
        );
        let discard_ratio = Arc::new(AtomicU32::new(0));
        let config_tracker = Arc::new(VersionTrack::new(config.clone()));
        let checker = FlowChecker::new(
            config_tracker.clone(),
            engine,
            discard_ratio.clone(),
            limiter.clone(),
        );
        let (tx, rx) = mpsc::sync_channel(5);

        Self {
            discard_ratio,
            limiter,
            tx: Some(tx),
            handle: Some(checker.start(rx, flow_info_receiver)),
            config_tracker,
        }
    }
}