// Copyright 2017 TiKV Project Authors. Licensed under Apache-2.0.

// Re-export duration.
pub(crate) use std::time::Duration;
use std::{
    cell::RefCell,
    cmp::Ordering,
    ops::{Add, AddAssign, Sub, SubAssign},
    sync::{
        Once,
        mpsc::{self, Sender},
    },
    thread::{self, Builder, JoinHandle},
    time::{SystemTime, UNIX_EPOCH},
};

use async_speed_limit::clock::{BlockingClock, Clock, StandardClock};
use time::{Duration as TimeDuration, Timespec};

/// Returns the monotonic raw time since some unspecified starting point.
pub(crate) use self::inner::monotonic_raw_now;
pub(crate) use self::inner::{monotonic_coarse_now, monotonic_now};
use crate::thread::StdThreadBuildWrapper;

const NANOSECONDS_PER_SECOND: u64 = 1_000_000_000;
const MILLISECONDS_PER_SECOND: u64 = 1_000;
const MICROSECONDS_PER_SECOND: u64 = 1_000_000;
const NANOSECONDS_PER_MILLISECOND: u64 = 1_000_000;
const NANOSECONDS_PER_MICROSECOND: u64 = 1_000;

/// Converts Duration to milliseconds.
#[inline]
pub fn duration_to_ms(d: Duration) -> u64 {
    let nanos = u64::from(d.subsec_nanos());
    // If Duration is too large, the result may be overflow.
    d.as_secs() * MILLISECONDS_PER_SECOND + (nanos / NANOSECONDS_PER_MILLISECOND)
}

/// Converts Duration to seconds.
#[inline]
pub fn duration_to_sec(d: Duration) -> f64 {
    let nanos = f64::from(d.subsec_nanos());
    d.as_secs() as f64 + (nanos / NANOSECONDS_PER_SECOND as f64)
}

pub fn nanos_to_secs(nanos: u64) -> f64 {
    nanos as f64 / NANOSECONDS_PER_SECOND as f64
}

/// Converts Duration to microseconds.
#[inline]
pub fn duration_to_us(d: Duration) -> u64 {
    let nanos = u64::from(d.subsec_nanos());
    // If Duration is too large, the result may be overflow.
    d.as_secs() * MICROSECONDS_PER_SECOND + (nanos / NANOSECONDS_PER_MICROSECOND)
}

/// Converts TimeSpec to nanoseconds
#[inline]
pub fn timespec_to_ns(t: Timespec) -> u64 {
    (t.sec as u64) * NANOSECONDS_PER_SECOND + t.nsec as u64
}

/// Converts Duration to nanoseconds.
#[inline]
pub fn duration_to_ns(d: Duration) -> u64 {
    let nanos = u64::from(d.subsec_nanos());
    // If Duration is too large, the result may be overflow.
    d.as_secs() * NANOSECONDS_PER_SECOND + nanos
}

pub trait InstantExt {
    fn saturating_elapsed(&self) -> Duration;
}

impl InstantExt for std::time::Instant {
    #[inline]
    fn saturating_elapsed(&self) -> Duration {
        std::time::Instant::now().saturating_duration_since(*self)
    }
}

/// A time in seconds since the start of the Unix epoch.
#[derive(Clone, Copy, Debug, Default, Eq, PartialEq, Ord, PartialOrd, Hash)]
pub struct UnixSecs(u64);

impl UnixSecs {
    pub fn now() -> UnixSecs {
        UnixSecs(
            SystemTime::now()
                .duration_since(UNIX_EPOCH)
                .unwrap()
                .as_secs(),
        )
    }

    pub fn zero() -> UnixSecs {
        UnixSecs(0)
    }

    pub fn into_inner(self) -> u64 {
        self.0
    }

    pub fn is_zero(self) -> bool {
        self.0 == 0
    }
}

pub struct SlowTimer {
    slow_time: Duration,
    t: Instant,
}

impl SlowTimer {
    pub fn new() -> SlowTimer {
        SlowTimer::default()
    }

    pub fn from(slow_time: Duration) -> SlowTimer {
        SlowTimer {
            slow_time,
            t: Instant::now_coarse(),
        }
    }

    pub fn from_secs(secs: u64) -> SlowTimer {
        SlowTimer::from(Duration::from_secs(secs))
    }

    pub fn from_millis(millis: u64) -> SlowTimer {
        SlowTimer::from(Duration::from_millis(millis))
    }

    pub fn saturating_elapsed(&self) -> Duration {
        self.t.saturating_elapsed()
    }

    pub fn is_slow(&self) -> bool {
        self.saturating_elapsed() >= self.slow_time
    }
}

const DEFAULT_SLOW_SECS: u64 = 1;

impl Default for SlowTimer {
    fn default() -> SlowTimer {
        SlowTimer::from_secs(DEFAULT_SLOW_SECS)
    }
}

const DEFAULT_WAIT_MS: u64 = 100;

pub struct Monitor {
    tx: Sender<bool>,
    handle: Option<JoinHandle<()>>,
}

impl Monitor {
    pub fn new<D, N>(on_jumped: D, now: N) -> Monitor
    where
        D: Fn() + Send + 'static,
        N: Fn() -> SystemTime + Send + 'static,
    {
        let props = crate::thread_group::current_properties();
        let (tx, rx) = mpsc::channel();
        let h = Builder::new()
            .name(thd_name!("time-monitor"))
            .spawn_wrapper(move || {
                crate::thread_group::set_properties(props);

                while rx.try_recv().is_err() {
                    let before = now();
                    thread::sleep(Duration::from_millis(DEFAULT_WAIT_MS));

                    let after = now();
                    if let Err(e) = after.duration_since(before) {
                        log::error!("system time jumped back, {:?}", e);
                        on_jumped()
                    }
                }
            })
            .unwrap();

        Monitor {
            tx,
            handle: Some(h),
        }
    }
}

impl Default for Monitor {
    fn default() -> Monitor {
        Monitor::new(|| {}, SystemTime::now)
    }
}

impl Drop for Monitor {
    fn drop(&mut self) {
        let h = self.handle.take();
        if h.is_none() {
            return;
        }

        if let Err(e) = self.tx.send(true) {
            log::error!("send quit message for time monitor worker failed, {:?}", e);
            return;
        }

        if let Err(e) = h.unwrap().join() {
            log::error!("join time monitor worker failed, {:?}", e);
        }
    }
}

#[cfg(not(target_os = "linux"))]
mod inner {
    use time::{self, Timespec};

    use super::NANOSECONDS_PER_SECOND;

    pub fn monotonic_raw_now() -> Timespec {
        // TODO Add monotonic raw clock time impl for macos and windows
        // Currently use `time::get_precise_ns()` instead.
        let ns = time::precise_time_ns();
        let s = ns / NANOSECONDS_PER_SECOND;
        let ns = ns % NANOSECONDS_PER_SECOND;
        Timespec::new(s as i64, ns as i32)
    }

    pub fn monotonic_now() -> Timespec {
        // TODO Add monotonic clock time impl for macos and windows
        monotonic_raw_now()
    }

    pub fn monotonic_coarse_now() -> Timespec {
        // TODO Add monotonic coarse clock time impl for macos and windows
        monotonic_raw_now()
    }
}

#[cfg(target_os = "linux")]
mod inner {
    use std::io;

    use time::Timespec;

    #[inline]
    pub fn monotonic_raw_now() -> Timespec {
        get_time(libc::CLOCK_MONOTONIC_RAW)
    }

    #[inline]
    pub fn monotonic_now() -> Timespec {
        get_time(libc::CLOCK_MONOTONIC)
    }

    #[inline]
    pub fn monotonic_coarse_now() -> Timespec {
        get_time(libc::CLOCK_MONOTONIC_COARSE)
    }

    #[inline]
    fn get_time(clock: libc::clockid_t) -> Timespec {
        let mut t = libc::timespec {
            tv_sec: 0,
            tv_nsec: 0,
        };
        let errno = unsafe { libc::clock_gettime(clock, &mut t) };
        if errno != 0 {
            panic!(
                "failed to get clocktime, err {}",
                io::Error::last_os_error()
            );
        }
        Timespec::new(t.tv_sec, t.tv_nsec as _)
    }
}

/// A measurement of a monotonically increasing clock.
/// It's similar and meant to replace `std::time::Instant`,
/// for providing extra features.
#[derive(Copy, Clone, Debug, Eq)]
pub enum Instant {
    Monotonic(Timespec),
    MonotonicCoarse(Timespec),
}

impl Instant {
    pub fn now() -> Instant {
        Instant::Monotonic(monotonic_now())
    }

    pub fn now_coarse() -> Instant {
        Instant::MonotonicCoarse(monotonic_coarse_now())
    }

    pub fn saturating_elapsed(&self) -> Duration {
        match *self {
            Instant::Monotonic(t) => {
                let now = monotonic_now();
                Instant::saturating_elapsed_duration(now, t)
            }
            Instant::MonotonicCoarse(t) => {
                let now = monotonic_coarse_now();
                Instant::saturating_elapsed_duration_coarse(now, t)
            }
        }
    }

    // This function may panic if the current time is earlier than this
    // instant. Deprecated.
    // pub fn elapsed_secs(&self) -> f64;

    pub fn saturating_elapsed_secs(&self) -> f64 {
        duration_to_sec(self.saturating_elapsed())
    }

    pub fn duration_since(&self, earlier: Instant) -> Duration {
        match (*self, earlier) {
            (Instant::Monotonic(later), Instant::Monotonic(earlier)) => {
                Instant::elapsed_duration(later, earlier)
            }
            (Instant::MonotonicCoarse(later), Instant::MonotonicCoarse(earlier)) => {
                Instant::saturating_elapsed_duration_coarse(later, earlier)
            }
            _ => {
                panic!("duration between different types of Instants");
            }
        }
    }

    pub fn saturating_duration_since(&self, earlier: Instant) -> Duration {
        match (*self, earlier) {
            (Instant::Monotonic(later), Instant::Monotonic(earlier)) => {
                Instant::saturating_elapsed_duration(later, earlier)
            }
            (Instant::MonotonicCoarse(later), Instant::MonotonicCoarse(earlier)) => {
                Instant::saturating_elapsed_duration_coarse(later, earlier)
            }
            _ => {
                panic!("duration between different types of Instants");
            }
        }
    }

    /// It is similar to `duration_since`, but it won't panic when `self` is
    /// less than `other`, and `None` will be returned in this case.
    ///
    /// Callers need to ensure that `self` and `other` are same type of
    /// Instants.
    pub fn checked_sub(&self, other: Instant) -> Option<Duration> {
        if *self >= other {
            Some(self.duration_since(other))
        } else {
            None
        }
    }

    pub(crate) fn elapsed_duration(later: Timespec, earlier: Timespec) -> Duration {
        if later >= earlier {
            (later - earlier).to_std().unwrap()
        } else {
            panic!(
                "monotonic time jumped back, {:.9} -> {:.9}",
                earlier.sec as f64 + f64::from(earlier.nsec) / NANOSECONDS_PER_SECOND as f64,
                later.sec as f64 + f64::from(later.nsec) / NANOSECONDS_PER_SECOND as f64
            );
        }
    }

    pub(crate) fn saturating_elapsed_duration(later: Timespec, earlier: Timespec) -> Duration {
        if later >= earlier {
            (later - earlier).to_std().unwrap()
        } else {
            log::error!(
                "monotonic time jumped back, {:.3} -> {:.3}",
                earlier.sec as f64 + f64::from(earlier.nsec) / NANOSECONDS_PER_SECOND as f64,
                later.sec as f64 + f64::from(later.nsec) / NANOSECONDS_PER_SECOND as f64
            );
            Duration::from_millis(0)
        }
    }

    // It is different from `elapsed_duration`, the resolution here is millisecond.
    // The processors in an SMP system do not start all at exactly the same time
    // and therefore the timer registers are typically running at an offset.
    // Use millisecond resolution for ignoring the error.
    // See more: https://linux.die.net/man/2/clock_gettime
    pub(crate) fn saturating_elapsed_duration_coarse(
        later: Timespec,
        earlier: Timespec,
    ) -> Duration {
        let later_ms = later.sec * MILLISECONDS_PER_SECOND as i64
            + i64::from(later.nsec) / NANOSECONDS_PER_MILLISECOND as i64;
        let earlier_ms = earlier.sec * MILLISECONDS_PER_SECOND as i64
            + i64::from(earlier.nsec) / NANOSECONDS_PER_MILLISECOND as i64;
        let dur = later_ms - earlier_ms;
        if dur >= 0 {
            Duration::from_millis(dur as u64)
        } else {
            log::debug!(
                "coarse time jumped back, {:.3} -> {:.3}",
                earlier.sec as f64 + f64::from(earlier.nsec) / NANOSECONDS_PER_SECOND as f64,
                later.sec as f64 + f64::from(later.nsec) / NANOSECONDS_PER_SECOND as f64
            );
            Duration::from_millis(0)
        }
    }
}

impl PartialEq for Instant {
    fn eq(&self, other: &Instant) -> bool {
        match (*self, *other) {
            (Instant::Monotonic(this), Instant::Monotonic(other))
            | (Instant::MonotonicCoarse(this), Instant::MonotonicCoarse(other)) => this.eq(&other),
            _ => false,
        }
    }
}

impl PartialOrd for Instant {
    fn partial_cmp(&self, other: &Instant) -> Option<Ordering> {
        match (*self, *other) {
            (Instant::Monotonic(this), Instant::Monotonic(other))
            | (Instant::MonotonicCoarse(this), Instant::MonotonicCoarse(other)) => {
                this.partial_cmp(&other)
            }
            // The Order of different types of Instants is meaningless.
            _ => None,
        }
    }
}

impl Add<Duration> for Instant {
    type Output = Instant;

    fn add(self, other: Duration) -> Instant {
        match self {
            Instant::Monotonic(t) => Instant::Monotonic(t + TimeDuration::from_std(other).unwrap()),
            Instant::MonotonicCoarse(t) => {
                Instant::MonotonicCoarse(t + TimeDuration::from_std(other).unwrap())
            }
        }
    }
}

impl AddAssign<Duration> for Instant {
    fn add_assign(&mut self, rhs: Duration) {
        *self = self.add(rhs)
    }
}

impl Sub<Duration> for Instant {
    type Output = Instant;

    fn sub(self, other: Duration) -> Instant {
        match self {
            Instant::Monotonic(t) => Instant::Monotonic(t - TimeDuration::from_std(other).unwrap()),
            Instant::MonotonicCoarse(t) => {
                Instant::MonotonicCoarse(t - TimeDuration::from_std(other).unwrap())
            }
        }
    }
}

impl SubAssign<Duration> for Instant {
    fn sub_assign(&mut self, rhs: Duration) {
        *self = self.sub(rhs)
    }
}

impl Sub<Instant> for Instant {
    type Output = Duration;

    // TODO: For safety in production code, `sub` actually does saturating_sub.
    // We should remove this operator from public scope.
    fn sub(self, other: Instant) -> Duration {
        self.saturating_duration_since(other)
    }
}

/// A coarse clock for `async_speed_limit`.
#[derive(Copy, Clone, Default, Debug)]
pub struct CoarseClock;

impl Clock for CoarseClock {
    type Instant = Instant;
    type Delay = <StandardClock as Clock>::Delay;

    fn now(&self) -> Self::Instant {
        Instant::now_coarse()
    }

    fn sleep(&self, dur: Duration) -> Self::Delay {
        StandardClock.sleep(dur)
    }
}

impl BlockingClock for CoarseClock {
    fn blocking_sleep(&self, dur: Duration) {
        StandardClock.blocking_sleep(dur);
    }
}

/// A limiter which uses the coarse clock for measurement.
pub type Limiter = async_speed_limit::Limiter<CoarseClock>;
pub type Consume = async_speed_limit::limiter::Consume<CoarseClock, ()>;

/// ReadId to judge whether the read requests come from the same GRPC stream.
#[derive(PartialEq, Clone, Debug)]
pub struct ThreadReadId {
    sequence: u64,
    pub create_time: Timespec,
}

thread_local!(static READ_SEQUENCE: RefCell<u64> = const { RefCell::new(0) });

impl ThreadReadId {
    pub fn new() -> ThreadReadId {
        let sequence = READ_SEQUENCE.with(|s| {
            let seq = *s.borrow() + 1;
            *s.borrow_mut() = seq;
            seq
        });
        ThreadReadId {
            sequence,
            create_time: monotonic_raw_now(),
        }
    }
}

impl Default for ThreadReadId {
    fn default() -> Self {
        Self::new()
    }
}

/// Default duration cost for spinning one round.
///
/// Heuristically, spin duration for one round is about 3～4ns.
static mut DEFAULT_DURATION_SPIN_ONE_ROUND: u64 = 1;

/// Setup the default ratio for spin duration.
pub fn setup_for_spin_interval() {
    static INIT: Once = Once::new();
    INIT.call_once(|| {
        let inspect_duration = Duration::from_millis(10);
        let start = Instant::now();
        let mut count = 0;
        // Spin for a while to get the duration for one round.
        for _ in 0..2_097_152 {
            count += 1;
            if count % 1024 == 0 && start.saturating_elapsed() >= inspect_duration {
                break;
            }
        }
        let elapsed_one_round = start.saturating_elapsed().as_nanos() as u64 / count;
        if elapsed_one_round > 0 {
            unsafe {
                DEFAULT_DURATION_SPIN_ONE_ROUND = elapsed_one_round;
            }
        }
        log::debug!("setup duration for spinning one round: {}ns", unsafe {
            DEFAULT_DURATION_SPIN_ONE_ROUND
        });
    });
}

/// Wait for at least `elaspsed` duration synchronously by looping.
///
/// Attention, this function is only suitable for short-time spinning, so
/// the `elaspsed` should be small, like 1ms. And the caller should not
/// rely on it to guarantee the exact time to sleep.
pub fn spin_at_least(elaspsed: Duration) {
    // Initialize default spin loop interval.
    setup_for_spin_interval();

    let rounds = unsafe { elaspsed.as_nanos() as u64 / DEFAULT_DURATION_SPIN_ONE_ROUND };
    let now = Instant::now();
    for i in 1..=rounds {
        if i % 100 == 0 && now.saturating_elapsed() >= elaspsed {
            return;
        }
    }
}