Repository: tikv/tikv
File: components/tikv_util/src/timer.rs (Rust)

Code before:
}

impl SteadyTimer {
    /// Creates a delay future that will be notified after the given duration.
    pub fn delay(&self, dur: Duration) -> Delay {
        self.handle.delay(self.clock.now() + dur)
    }
}

lazy_static! {
    static ref GLOBAL_STEADY_TIMER: SteadyTimer = start_global_steady_timer();
}

impl Default for SteadyTimer {
    #[inline]
    fn default() -> SteadyTimer {
        GLOBAL_STEADY_TIMER.clone()
    }
}

fn start_global_steady_timer() -> SteadyTimer {
    let (tx, rx) = mpsc::channel();
    let clock = SteadyClock::default();
    let clock_ = clock.clone();
    Builder::new()
        .name(thd_name!("steady-timer"))
        .spawn_wrapper(move || {
            let c = Clock::new_with_now(clock_);
            let mut timer = tokio_timer::Timer::new_with_now(ParkThread::new(), c);
            tx.send(timer.handle()).unwrap();
            loop {
                timer.turn(None).unwrap();
            }
        })
        .unwrap();
    SteadyTimer {
        clock,
        handle: rx.recv().unwrap(),
    }
}

#[cfg(test)]
mod tests {
    use futures::{compat::Future01CompatExt, executor::block_on};

    use super::*;

    #[derive(Debug, PartialEq, Copy, Clone)]
    enum Task {
        A,
        B,

Review comment (performance):
While I still think a `Cell<u64>` is enough here, using a Mutex here is acceptable to me as there should be no performance impact.

Code after:
}

impl Default for SteadyTimer {
    #[inline]
    fn default() -> SteadyTimer {
        GLOBAL_STEADY_TIMER.clone()
    }
}

fn start_global_steady_timer() -> SteadyTimer {
    let clock = SteadyClock::default();
    let handle = start_timer_thread("steady-timer", clock.clone());
    SteadyTimer { clock, handle }
}

/// A clock that ratchets forward.
///
/// It is used to workaround a panic[^1] in tokio-timer when the clock goes
/// backward, which is possible in some environments, e.g., aliyun ECS g7.
///
/// [^1]: https://github.com/tokio-rs/tokio/pull/515
struct RatchetClock<T: Now> {
    // Although the RatchetClock has to be thread-safe (Now requires Sync+Send),
    // the [`tokio_timer::timer::Clock`] that wraps it can only be accessed by
    // the timer thread. Therefore, performance penalty of using a Mutex is negligible.
    state: Mutex<RatchetClockState<T>>,
}

struct RatchetClockState<T> {
    start: std::time::Instant,
    clock: T,

    // The last recorded time in milliseconds.
    //
    // Using u64 is sufficient here because tokio-timer operates with millisecond
    // precision, and `u64::MAX` milliseconds represent an extremely large time span
    // (several million years).
    last_now_ms: u64,
}

impl<T: Now> Now for RatchetClock<T> {
    fn now(&self) -> std::time::Instant {
        let mut state = self.state.lock().unwrap();
        let now = state.clock.now();
        let now_ms = now.saturating_duration_since(state.start).as_millis() as u64;

        if now_ms > state.last_now_ms {
            state.last_now_ms = now_ms;
            state.start + Duration::from_millis(now_ms)
        } else {
            state.start + Duration::from_millis(state.last_now_ms)