//! Channel padding

//!

//! Tor spec `padding-spec.txt` section 2.

//!

//! # Overview of channel padding control arrangements

//!

//!  1. `tor_chanmgr::mgr::map` collates information about dormancy, netdir,

//!     and overall client configuration, to maintain a

//!     [`ChannelPaddingInstructions`](crate::channel::ChannelPaddingInstructions)

//!     which is to be used for all relevant[^relevant] channels.

//!     This is distributed to channel frontends (`Channel`s)

//!     by calling `Channel::reparameterize`.

//!

//!  2. Circuit and channel `get_or_launch` methods all take a `ChannelUsage`.

//!     This is plumbed through the layers to `AbstractChanMgr::get_or_launch`,

//!     which passes it to the channel frontend via `Channel::note_usage`.

//!

//!  3. The `Channel` collates this information, and maintains an idea

//!     of whether padding is relevant for this channel (`PaddingControlState`).

//!     For channels where it *is* relevant, it sends `CtrlMsg::ConfigUpdate`

//!     to the reactor.

//!

//!  4. The reactor handles `CtrlMsg::ConfigUpdate` by reconfiguring is padding timer;

//!     and by sending PADDING_NEGOTIATE cell(s).

//!

//! [^relevant]: A "relevant" channel is one which is not excluded by the rules about

//! padding in padding-spec 2.2.  Arti does not currently support acting as a relay,

//! so all our channels are client-to-guard or client-to-directory.

use std::pin::Pin;

// TODO, coarsetime maybe?  But see arti#496 and also we want to use the mockable SleepProvider

use std::time::{Duration, Instant};

use derive_builder::Builder;

use educe::Educe;

use futures::future::{self, FusedFuture};

use futures::FutureExt;

use pin_project::pin_project;

use rand::distr::Distribution;

use tracing::error;

use tor_cell::chancell::msg::{Padding, PaddingNegotiate};

use tor_config::impl_standard_builder;

use tor_error::into_internal;

use tor_rtcompat::SleepProvider;

use tor_units::IntegerMilliseconds;

/// Timer that organises wakeups when channel padding should be sent

///

/// Use [`next()`](Timer::next) to find when to send padding, and

/// [`note_cell_sent()`](Timer::note_cell_sent) to reset the timeout when data flows.

///

/// A `Timer` can be in "disabled" state, in which case `next()` never completes.

///

/// `Timer` must be pinned before use

/// (this allows us to avoid involving the allocator when we reschedule).

pub(crate) struct Timer<R: SleepProvider> {

    /// [`SleepProvider`]

    sleep_prov: R,

    /// Parameters controlling distribution of padding time intervals

    ///

    /// Can be `None` to mean the timing parameters are set to infinity.

    parameters: Option<PreparedParameters>,

    /// Gap that we intend to leave between last sent cell, and the padding

    ///

    /// We only resample this (calculating a new random delay) after the previous

    /// timeout actually expired.

    ///

    /// `None` if the timer is disabled.

    /// (This can be done explicitly, but also occurs on time calculation overflow.)

    ///

    /// Invariants: this field may be `Some` or `None` regardless of the values

    /// of other fields.  If this field is `None` then the values in `trigger_at`

    /// and `waker` are unspecified.

    selected_timeout: Option<Duration>,

    /// Absolute time at which we should send padding

    ///

    /// `None` if cells more recently sent than we were polled.

    /// That would mean that we are currently moving data out through this channel.

    /// The absolute timeout will need to be recalculated when the data flow pauses.

    ///

    /// `Some` means our `next` has been demanded recently.

    /// Then `trigger_at` records the absolute timeout at which we should send padding,

    /// which was calculated the first time we were polled (after data).

    ///

    /// Invariants: the value in this field is meaningful only if `selected_timeout`

    /// is `Some`.

    ///

    /// If `selected_timeout` is `Some`, and `trigger_at` is therefore valid,

    /// it is (obviously) no later than `selected_timeout` from now.

    ///

    /// See also `waker`.

    trigger_at: Option<Instant>,

    /// Actual waker from the `SleepProvider`

    ///

    /// This is created and updated lazily, because we suspect that with some runtimes

    /// setting timeouts may be slow.

    /// Lazy updating means that with intermittent data traffic, we do not keep scheduling,

    /// descheduling, and adjusting, a wakeup time.

    ///

    /// Invariants:

    ///

    /// If `selected_timeout` is `Some`,

    /// the time at which this waker will trigger here is never *later* than `trigger_at`,

    /// and never *later* than `selected_timeout` from now.

    ///

    /// The wakeup time here may well be earlier than `trigger_at`,

    /// and sooner than `selected_timeout` from now.  It may even be in the past.

    /// When we wake up and discover this situation, we reschedule a new waker.

    ///

    /// If `selected_timeout` is `None`, the value is unspecified.

    /// We may retain a `Some` in this case so that if `SleepProvider` is enhanced to

    /// support rescheduling, we can do that without making a new `SleepFuture`

    /// (and without completely reorganising this the `Timer` state structure.)

    #[pin]

    waker: Option<R::SleepFuture>,

}

/// Timing parameters, as described in `padding-spec.txt`

#[builder(build_fn(error = "ParametersError", private, name = "build_inner"))]

pub struct Parameters {

    /// Low end of the distribution of `X`

    #[builder(default = "1500.into()")]

    pub(crate) low: IntegerMilliseconds<u32>,

    /// High end of the distribution of `X` (inclusive)

    #[builder(default = "9500.into()")]

    pub(crate) high: IntegerMilliseconds<u32>,

}

/// An error that occurred whil e constructing padding parameters.

#[derive(Clone, Debug, thiserror::Error)]

#[non_exhaustive]

pub enum ParametersError {

    /// Could not construct a range: there were no members between low and high.

    #[error("Cannot construct padding parameters: low bound was above the high bound.")]

    InvalidRange,

}

impl ParametersBuilder {

    /// Try to construct a [`Parameters`] from this builder.

    ///

    /// returns an error if the distribution is ill-defined.

}

impl_standard_builder! { Parameters: !Deserialize + !Builder + !Default }

impl Parameters {

    /// Return a `PADDING_NEGOTIATE START` cell specifying precisely these parameters

    ///

    /// This function does not take account of the need to avoid sending particular

    /// parameters, and instead sending zeroes, if the requested padding is the consensus

    /// default.  The caller must take care of that.

    /// Make a Parameters containing the specification-defined default parameters

    /// Make a Parameters sentinel value, with both fields set to zero, which means "no padding"

}

/// Timing parameters, "compiled" into a form which can be sampled more efficiently

///

/// According to the docs for [`rand::Rng::gen_range`],

/// it is better to construct a distribution,

/// than to call `gen_range` repeatedly on the same range.

#[derive(Debug, Clone)]

struct PreparedParameters {

    /// The distribution of `X` (not of the ultimate delay, which is `max(X1,X2)`)

    x_distribution_ms: rand::distr::Uniform<u32>,

}

/// Return value from `prepare_to_sleep`: instructions for what caller ought to do

#[derive(Educe)]

#[educe(Debug)]

enum SleepInstructions<'f, R: SleepProvider> {

    /// Caller should send padding immediately

    Immediate {

        /// The current `Instant`, returned so that the caller need not call `now` again

        now: Instant,

    },

    /// Caller should wait forever

    Forever,

    /// Caller should `await` this

    Waker(#[educe(Debug(ignore))] Pin<&'f mut R::SleepFuture>),

}

impl<R: SleepProvider> Timer<R> {

    /// Create a new `Timer`

    #[allow(dead_code)]

    /// Create a new `Timer` which starts out disabled

        })

    /// Disable this `Timer`

    ///

    /// Idempotent.

    /// Enable this `Timer`

    ///

    /// (If the timer was disabled, the timeout will only start to run when `next()`

    /// is next polled.)

    ///

    /// Idempotent.

    /// Set this `Timer`'s parameters

    ///

    /// Will not enable or disable the timer; that must be done separately if desired.

    ///

    /// The effect may not be immediate: if we are already in a gap between cells,

    /// that existing gap may not be adjusted.

    /// (We don't *restart* the timer since that would very likely result in a gap

    /// longer than either of the configured values.)

    ///

    /// Idempotent.

    /// Enquire whether this `Timer` is currently enabled

    /// Select a fresh timeout (and enable, if possible)

    /// Note that data has been sent (ie, reset the timeout, delaying the next padding)

    /// Calculate when to send padding, and return a suitable waker

    ///

    /// In the usual case returns [`SleepInstructions::Waker`].

            // We need to do this with is_some and expect because we need to consume self

            // to get a return value with the right lifetimes.

                None => {

                }

            }),

        };

    /// Wait until we should next send padding, and then return the padding message

    ///

    /// Should be used as a low-priority branch within `select_biased!`.

    ///

    /// (`next()` has to be selected on, along with other possible events, in the

    /// main loop, so that the padding timer runs concurrently with other processing;

    /// and it should be in a low-priority branch of `select_biased!` as an optimisation:

    /// that avoids calculating timeouts etc. until necessary,

    /// i.e. it calculates them only when the main loop would otherwise block.)

    ///

    /// The returned future is async-cancel-safe,

    /// but once it yields, the padding must actually be sent.

    /// Wait until we should next send padding (not `FusedFuture`)

    ///

    /// Callers wants a [`FusedFuture`] because `select!` needs one.

            }

            // This timer has fired and has therefore been used up.

            // When we go round again we will make a new one.

            //

            // TODO: have SleepProviders provide a reschedule function, and use it.

            // That is likely to be faster where supported.

        };

        // It's time to send padding.

        // Firstly, calculate the new timeout for the *next* padding,

        // so that we leave the `Timer` properly programmed.

}

impl Parameters {

    /// "Compile" the parameters into a form which can be quickly sampled

        })

}

impl PreparedParameters {

    /// Randomly select a timeout (as per `padding-spec.txt`)

}

#[cfg(test)]

mod test {

    // @@ begin test lint list maintained by maint/add_warning @@

    #![allow(clippy::bool_assert_comparison)]

    #![allow(clippy::clone_on_copy)]

    #![allow(clippy::dbg_macro)]

    #![allow(clippy::mixed_attributes_style)]

    #![allow(clippy::print_stderr)]

    #![allow(clippy::print_stdout)]

    #![allow(clippy::single_char_pattern)]

    #![allow(clippy::unwrap_used)]

    #![allow(clippy::unchecked_duration_subtraction)]

    #![allow(clippy::useless_vec)]

    #![allow(clippy::needless_pass_by_value)]

    //! <!-- @@ end test lint list maintained by maint/add_warning @@ -->

    use super::*;

    use futures::future::ready;

    use futures::select_biased;

    use itertools::{izip, Itertools};

    use statrs::distribution::ContinuousCDF;

    use tokio::pin;

    use tokio_crate::{self as tokio};

    use tor_rtcompat::*;

    async fn assert_not_ready<R: Runtime>(timer: &mut Pin<&mut Timer<R>>) {

        select_biased! {

            _ = timer.as_mut().next() => panic!("unexpectedly ready"),

            _ = ready(()) => { },

        };

    }

    async fn assert_is_ready<R: Runtime>(timer: &mut Pin<&mut Timer<R>>) {

        let _: Padding = select_biased! {

            p = timer.as_mut().next() => p,

            _ = ready(()) => panic!("pad timer failed to yield"),

        };

    }

    #[test]

    fn timer_impl() {

        let runtime = tor_rtcompat::tokio::TokioNativeTlsRuntime::create().unwrap();

        #[allow(deprecated)] // TODO #1885

        let runtime = tor_rtmock::MockSleepRuntime::new(runtime);

        let parameters = Parameters {

            low: 1000.into(),

            high: 1000.into(),

        };

        let () = runtime.block_on(async {

            let timer = Timer::new(runtime.clone(), parameters).unwrap();

            pin!(timer);

            assert_eq! { true, timer.is_enabled() }

            // expiry time not yet calculated

            assert_eq! { timer.as_mut().trigger_at, None };

            // ---------- timeout value ----------

            // Just created, not ready yet

            assert_not_ready(&mut timer).await;

            runtime.advance(Duration::from_millis(999)).await;

            // Not quite ready

            assert_not_ready(&mut timer).await;

            runtime.advance(Duration::from_millis(1)).await;

            // Should go off precisely now

            assert_is_ready(&mut timer).await;

            assert_not_ready(&mut timer).await;

            runtime.advance(Duration::from_millis(1001)).await;

            // Should go off 1ms ago, fine

            assert_is_ready(&mut timer).await;

            // ---------- various resets ----------

            runtime.advance(Duration::from_millis(500)).await;

            timer.note_cell_sent();

            assert_eq! { timer.as_mut().trigger_at, None };

            // This ought not to cause us to actually calculate the expiry time

            let () = select_biased! {

                _ = ready(()) => { },

                _ = timer.as_mut().next() => panic!(),

            };

            assert_eq! { timer.as_mut().trigger_at, None };

            // ---------- disable/enable ----------

            timer.disable();

            runtime.advance(Duration::from_millis(2000)).await;

            assert_eq! { timer.as_mut().selected_timeout, None };

            assert_eq! { false, timer.is_enabled() }

            assert_not_ready(&mut timer).await;

            timer.enable();

            runtime.advance(Duration::from_millis(3000)).await;

            assert_eq! { true, timer.is_enabled() }

            // Shouldn't be already ready, since we haven't polled yet

            assert_not_ready(&mut timer).await;

            runtime.advance(Duration::from_millis(1000)).await;

            // *Now*

            assert_is_ready(&mut timer).await;

        });

        let () = runtime.block_on(async {

            let timer = Timer::new(runtime.clone(), parameters).unwrap();

            pin!(timer);

            assert! { timer.as_mut().selected_timeout.is_some() };

            assert! { timer.as_mut().trigger_at.is_none() };

            // Force an overflow by guddling

            *timer.as_mut().project().selected_timeout = Some(Duration::MAX);

            assert_not_ready(&mut timer).await;

            dbg!(timer.as_mut().project().trigger_at);

            assert_eq! { false, timer.is_enabled() }

        });

        let () = runtime.block_on(async {

            let timer = Timer::new_disabled(runtime.clone(), None).unwrap();

            assert! { timer.parameters.is_none() };

            pin!(timer);

            assert_not_ready(&mut timer).await;

            assert! { timer.as_mut().selected_timeout.is_none() };

            assert! { timer.as_mut().trigger_at.is_none() };

        });

        let () = runtime.block_on(async {

            let timer = Timer::new_disabled(runtime.clone(), Some(parameters)).unwrap();

            assert! { timer.parameters.is_some() };

            pin!(timer);

            assert_not_ready(&mut timer).await;

            runtime.advance(Duration::from_millis(3000)).await;

            assert_not_ready(&mut timer).await;

            timer.as_mut().enable();

            assert_not_ready(&mut timer).await;

            runtime.advance(Duration::from_millis(3000)).await;

            assert_is_ready(&mut timer).await;

        });

    }

    #[test]

    #[allow(clippy::print_stderr)]

    fn timeout_distribution() {

        // Test that the distribution of padding intervals is as we expect.  This is not so

        // straightforward.  We need to deal with true randomness (since we can't plumb a

        // testing RNG into the padding timer, and perhaps don't even *want* to make that a

        // mockable interface).  Measuring a distribution of random variables involves some

        // statistics.

        // The overall approach is:

        //    Use a fixed (but nontrivial) low to high range

        //    Sample N times into n equal sized buckets

        //    Calculate the expected number of samples in each bucket

        //    Do a chi^2 test.  If it doesn't spot a potential difference, declare OK.

        //    If the chi^2 test does definitely declare a difference, declare failure.

        //    Otherwise increase N and go round again.

        //

        // This allows most runs to be fast without having an appreciable possibility of a

        // false test failure and while being able to detect even quite small deviations.

        // Notation from

        // https://en.wikipedia.org/wiki/Pearson%27s_chi-squared_test#Calculating_the_test-statistic

        // I haven't done a formal power calculation but empirically

        // this detects the following most of the time:

        //  deviation of the CDF power from B^2 to B^1.98

        //  wrong minimum value by 25ms out of 12s, low_ms = min + 25

        //  wrong maximum value by 10ms out of 12s, high_ms = max -1 - 10

        #[allow(non_snake_case)]

        let mut N = 100_0000;

        #[allow(non_upper_case_globals)]

        const n: usize = 100;

        const P_GOOD: f64 = 0.05; // Investigate further 5% of times (if all is actually well)

        const P_BAD: f64 = 1e-12;

        loop {

            eprintln!("padding distribution test, n={} N={}", n, N);

            let min = 5000;

            let max = 17000; // Exclusive

            assert_eq!(0, (max - min) % (n as u32)); // buckets must match up to integer boundaries

            let cdf = (0..=n)

                .map(|bi| {

                    let b = (bi as f64) / (n as f64);

                    // expected distribution:

                    // with B = bi / n

                    //   P(X) < B == B

                    //   P(max(X1,X1)) < B = B^2

                    b.powi(2)

                })

                .collect_vec();

            let pdf = cdf

                .iter()

                .cloned()

                .tuple_windows()

                .map(|(p, q)| q - p)

                .collect_vec();

            let exp = pdf.iter().cloned().map(|p| p * f64::from(N)).collect_vec();

            // chi-squared test only valid if every cell expects at least 5

            assert!(exp.iter().cloned().all(|ei| ei >= 5.));

            let mut obs = [0_u32; n];

            let params = Parameters {

                low: min.into(),

                high: (max - 1).into(), // convert exclusive to inclusive

            }

            .prepare()

            .unwrap();

            for _ in 0..N {

                let xx = params.select_timeout();

                let ms = xx.as_millis();

                let ms = u32::try_from(ms).unwrap();

                assert!(ms >= min);

                assert!(ms < max);

                // Integer arithmetic ensures that we classify exactly

                let bi = ((ms - min) * (n as u32)) / (max - min);

                obs[bi as usize] += 1;

            }

            let chi2 = izip!(&obs, &exp)

                .map(|(&oi, &ei)| (f64::from(oi) - ei).powi(2) / ei)

                .sum::<f64>();

            // n degrees of freedom, one-tailed test

            // (since distro parameters are all fixed, not estimated from the sample)

            let chi2_distr = statrs::distribution::ChiSquared::new(n as _).unwrap();

            // probability of good code generating a result at least this bad

            let p = 1. - chi2_distr.cdf(chi2);

            eprintln!(

                "padding distribution test, n={} N={} chi2={} p={}",

                n, N, chi2, p

            );

            if p >= P_GOOD {

                break;

            }

            for (i, (&oi, &ei)) in izip!(&obs, &exp).enumerate() {

                eprintln!("bi={:4} OI={:4} EI={}", i, oi, ei);

            }

            if p < P_BAD {

                panic!("distribution is wrong (p < {:e})", P_BAD);

            }

            // This is statistically rather cheaty: we keep trying until we get a definite

            // answer!  But we radically increase the power of the test each time.

            // If the distribution is really wrong, this test ought to find it soon enough,

            // especially since we run this repeatedly in CI.

            N *= 10;

        }

    }

    #[test]

    fn parameters_range() {

        let ms100 = IntegerMilliseconds::new(100);

        let ms1000 = IntegerMilliseconds::new(1000);

        let ms1500 = IntegerMilliseconds::new(1500);

        let ms9500 = IntegerMilliseconds::new(9500);

        // default

        let p = Parameters::builder().build().unwrap();

        assert_eq!(

            p,

            Parameters {

                low: ms1500,

                high: ms9500

            }

        );

        assert!(p.prepare().is_ok());

        // low < high

        let mut pb = Parameters::builder();

        pb.low(ms100);

        pb.high(ms1000);

        let p = pb.build().unwrap();

        assert_eq!(

            p,

            Parameters {

                low: ms100,

                high: ms1000

            }

        );

        let p = p.prepare().unwrap();

        let range = Duration::try_from(ms100).unwrap()..=Duration::try_from(ms1000).unwrap();

        for _ in 1..100 {

            assert!(range.contains(&p.select_timeout()));

        }

        // low == high

        let mut pb = Parameters::builder();

        pb.low(ms1000);

        pb.high(ms1000);

        let p = pb.build().unwrap();

        assert!(p.prepare().is_ok());

        // low > high (error case)

        let mut pb = Parameters::builder();

        pb.low(ms1000);

        pb.high(ms100);

        let e = pb.build().unwrap_err();

        assert!(matches!(e, ParametersError::InvalidRange));

    }

}