//! Capabilities to send data from operators
//!
//! Timely dataflow operators are only able to send data if they possess a "capability",
//! a system-created object which warns the runtime that the operator may still produce
//! output records.
//!
//! The timely dataflow runtime creates a capability and provides it to an operator whenever
//! the operator receives input data. The capabilities allow the operator to respond to the
//! received data, immediately or in the future, for as long as the capability is held.
//!
//! Timely dataflow's progress tracking infrastructure communicates the number of outstanding
//! capabilities across all workers.
//! Each operator may hold on to its capabilities, and may clone, advance, and drop them.
//! Each of these actions informs the timely dataflow runtime of changes to the number of outstanding
//! capabilities, so that the runtime can notice when the count for some capability reaches zero.
//! While an operator can hold capabilities indefinitely, and create as many new copies of them
//! as it would like, the progress tracking infrastructure will not move forward until the
//! operators eventually release their capabilities.
//!
//! Note that these capabilities currently lack the property of "transferability":
//! An operator should not hand its capabilities to some other operator. In the future, we should
//! probably bind capabilities more strongly to a specific operator and output.

use std::{borrow, error::Error, fmt::Display, ops::Deref};
use std::rc::Rc;
use std::cell::RefCell;
use std::fmt::{self, Debug};

use crate::order::PartialOrder;
use crate::progress::Antichain;
use crate::progress::Timestamp;
use crate::progress::ChangeBatch;
use crate::scheduling::Activations;
use crate::dataflow::channels::pullers::counter::ConsumedGuard;

/// An internal trait expressing the capability to send messages with a given timestamp.
pub trait CapabilityTrait<T: Timestamp> {
    /// The timestamp associated with the capability.
    fn time(&self) -> &T;
    fn valid_for_output(&self, query_buffer: &Rc<RefCell<ChangeBatch<T>>>) -> bool;
}

impl<'a, T: Timestamp, C: CapabilityTrait<T>> CapabilityTrait<T> for &'a C {
    fn time(&self) -> &T { (**self).time() }
    fn valid_for_output(&self, query_buffer: &Rc<RefCell<ChangeBatch<T>>>) -> bool {
        (**self).valid_for_output(query_buffer)
    }
}
impl<'a, T: Timestamp, C: CapabilityTrait<T>> CapabilityTrait<T> for &'a mut C {
    fn time(&self) -> &T { (**self).time() }
    fn valid_for_output(&self, query_buffer: &Rc<RefCell<ChangeBatch<T>>>) -> bool {
        (**self).valid_for_output(query_buffer)
    }
}

/// The capability to send data with a certain timestamp on a dataflow edge.
///
/// Capabilities are used by timely dataflow's progress tracking machinery to restrict and track
/// when user code retains the ability to send messages on dataflow edges. All capabilities are
/// constructed by the system, and should eventually be dropped by the user. Failure to drop
/// a capability (for whatever reason) will cause timely dataflow's progress tracking to stall.
pub struct Capability<T: Timestamp> {
    time: T,
    internal: Rc<RefCell<ChangeBatch<T>>>,
}

impl<T: Timestamp> CapabilityTrait<T> for Capability<T> {
    fn time(&self) -> &T { &self.time }
    fn valid_for_output(&self, query_buffer: &Rc<RefCell<ChangeBatch<T>>>) -> bool {
        Rc::ptr_eq(&self.internal, query_buffer)
    }
}

impl<T: Timestamp> Capability<T> {
    /// Creates a new capability at `time` while incrementing (and keeping a reference to) the provided
    /// [`ChangeBatch`].
    pub(crate) fn new(time: T, internal: Rc<RefCell<ChangeBatch<T>>>) -> Self {
        internal.borrow_mut().update(time.clone(), 1);

        Self {
            time,
            internal,
        }
    }

    /// The timestamp associated with this capability.
    pub fn time(&self) -> &T {
        &self.time
    }

    /// Makes a new capability for a timestamp `new_time` greater or equal to the timestamp of
    /// the source capability (`self`).
    ///
    /// This method panics if `self.time` is not less or equal to `new_time`.
    pub fn delayed(&self, new_time: &T) -> Capability<T> {
        /// Makes the panic branch cold & outlined to decrease code bloat & give
        /// the inner function the best chance possible of being inlined with
        /// minimal code bloat
        #[cold]
        #[inline(never)]
        fn delayed_panic(capability: &dyn Debug, invalid_time: &dyn Debug) -> ! {
            // Formatting & panic machinery is relatively expensive in terms of code bloat, so
            // we outline it
            panic!(
                "Attempted to delay {:?} to {:?}, which is not beyond the capability's time.",
                capability,
                invalid_time,
            )
        }

        self.try_delayed(new_time)
            .unwrap_or_else(|| delayed_panic(self, new_time))
    }

    /// Attempts to make a new capability for a timestamp `new_time` that is
    /// greater or equal to the timestamp of the source capability (`self`).
    ///
    /// Returns [`None`] `self.time` is not less or equal to `new_time`.
    pub fn try_delayed(&self, new_time: &T) -> Option<Capability<T>> {
        if self.time.less_equal(new_time) {
            Some(Self::new(new_time.clone(), self.internal.clone()))
        } else {
            None
        }
    }

    /// Downgrades the capability to one corresponding to `new_time`.
    ///
    /// This method panics if `self.time` is not less or equal to `new_time`.
    pub fn downgrade(&mut self, new_time: &T) {
        /// Makes the panic branch cold & outlined to decrease code bloat & give
        /// the inner function the best chance possible of being inlined with
        /// minimal code bloat
        #[cold]
        #[inline(never)]
        fn downgrade_panic(capability: &dyn Debug, invalid_time: &dyn Debug) -> ! {
            // Formatting & panic machinery is relatively expensive in terms of code bloat, so
            // we outline it
            panic!(
                "Attempted to downgrade {:?} to {:?}, which is not beyond the capability's time.",
                capability,
                invalid_time,
            )
        }

        self.try_downgrade(new_time)
            .unwrap_or_else(|_| downgrade_panic(self, new_time))
    }

    /// Attempts to downgrade the capability to one corresponding to `new_time`.
    ///
    /// Returns a [DowngradeError] if `self.time` is not less or equal to `new_time`.
    pub fn try_downgrade(&mut self, new_time: &T) -> Result<(), DowngradeError> {
        if let Some(new_capability) = self.try_delayed(new_time) {
            *self = new_capability;
            Ok(())
        } else {
            Err(DowngradeError(()))
        }
    }
}

// Necessary for correctness. When a capability is dropped, the "internal" `ChangeBatch` needs to be
// updated accordingly to inform the rest of the system that the operator has released its permit
// to send data and request notification at the associated timestamp.
impl<T: Timestamp> Drop for Capability<T> {
    fn drop(&mut self) {
        self.internal.borrow_mut().update(self.time.clone(), -1);
    }
}

impl<T: Timestamp> Clone for Capability<T> {
    fn clone(&self) -> Capability<T> {
        Self::new(self.time.clone(), self.internal.clone())
    }
}

impl<T: Timestamp> Deref for Capability<T> {
    type Target = T;

    fn deref(&self) -> &T {
        &self.time
    }
}

impl<T: Timestamp> Debug for Capability<T> {
    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
        f.debug_struct("Capability")
            .field("time", &self.time)
            .field("internal", &"...")
            .finish()
    }
}

impl<T: Timestamp> PartialEq for Capability<T> {
    fn eq(&self, other: &Self) -> bool {
        self.time() == other.time() && Rc::ptr_eq(&self.internal, &other.internal)
    }
}
impl<T: Timestamp> Eq for Capability<T> { }

impl<T: Timestamp> PartialOrder for Capability<T> {
    fn less_equal(&self, other: &Self) -> bool {
        self.time().less_equal(other.time()) && Rc::ptr_eq(&self.internal, &other.internal)
    }
}

impl<T: Timestamp> ::std::hash::Hash for Capability<T> {
    fn hash<H: ::std::hash::Hasher>(&self, state: &mut H) {
        self.time.hash(state);
    }
}

/// An error produced when trying to downgrade a capability with a time
/// that's not less than or equal to the current capability
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub struct DowngradeError(());

impl Display for DowngradeError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.write_str("could not downgrade the given capability")
    }
}

impl Error for DowngradeError {}

/// A shared list of shared output capability buffers.
type CapabilityUpdates<T> = Rc<RefCell<Vec<Rc<RefCell<ChangeBatch<T>>>>>>;

/// An capability of an input port. Holding onto this capability will implicitly holds onto a
/// capability for all the outputs ports this input is connected to, after the connection summaries
/// have been applied.
///
/// This input capability supplies a `retain_for_output(self)` method which consumes the input
/// capability and turns it into a [Capability] for a specific output port.
pub struct InputCapability<T: Timestamp> {
    /// Output capability buffers, for use in minting capabilities.
    internal: CapabilityUpdates<T>,
    /// Timestamp summaries for each output.
    summaries: Rc<RefCell<Vec<Antichain<T::Summary>>>>,
    /// A drop guard that updates the consumed capability this InputCapability refers to on drop
    consumed_guard: ConsumedGuard<T>,
}

impl<T: Timestamp> CapabilityTrait<T> for InputCapability<T> {
    fn time(&self) -> &T { self.time() }
    fn valid_for_output(&self, query_buffer: &Rc<RefCell<ChangeBatch<T>>>) -> bool {
        let borrow = self.summaries.borrow();
        self.internal.borrow().iter().enumerate().any(|(index, rc)| {
            // To be valid, the output buffer must match and the timestamp summary needs to be the default.
            Rc::ptr_eq(rc, query_buffer) && borrow[index].len() == 1 && borrow[index][0] == Default::default()
        })
    }
}

impl<T: Timestamp> InputCapability<T> {
    /// Creates a new capability reference at `time` while incrementing (and keeping a reference to)
    /// the provided [`ChangeBatch`].
    pub(crate) fn new(internal: CapabilityUpdates<T>, summaries: Rc<RefCell<Vec<Antichain<T::Summary>>>>, guard: ConsumedGuard<T>) -> Self {
        InputCapability {
            internal,
            summaries,
            consumed_guard: guard,
        }
    }

    /// The timestamp associated with this capability.
    pub fn time(&self) -> &T {
        self.consumed_guard.time()
    }

    /// Makes a new capability for a timestamp `new_time` greater or equal to the timestamp of
    /// the source capability (`self`).
    ///
    /// This method panics if `self.time` is not less or equal to `new_time`.
    pub fn delayed(&self, new_time: &T) -> Capability<T> {
        self.delayed_for_output(new_time, 0)
    }

    /// Delays capability for a specific output port.
    pub fn delayed_for_output(&self, new_time: &T, output_port: usize) -> Capability<T> {
        use crate::progress::timestamp::PathSummary;
        if self.summaries.borrow()[output_port].iter().flat_map(|summary| summary.results_in(self.time())).any(|time| time.less_equal(new_time)) {
            Capability::new(new_time.clone(), self.internal.borrow()[output_port].clone())
        } else {
            panic!("Attempted to delay to a time ({:?}) not greater or equal to the operators input-output summary ({:?}) applied to the capabilities time ({:?})", new_time, self.summaries.borrow()[output_port], self.time());
        }
    }

    /// Transform to an owned capability.
    ///
    /// This method produces an owned capability which must be dropped to release the
    /// capability. Users should take care that these capabilities are only stored for
    /// as long as they are required, as failing to drop them may result in livelock.
    ///
    /// This method panics if the timestamp summary to output zero strictly advances the time.
    pub fn retain(self) -> Capability<T> {
        self.retain_for_output(0)
    }

    /// Transforms to an owned capability for a specific output port.
    ///
    /// This method panics if the timestamp summary to `output_port` strictly advances the time.
    pub fn retain_for_output(self, output_port: usize) -> Capability<T> {
        use crate::progress::timestamp::PathSummary;
        let self_time = self.time().clone();
        if self.summaries.borrow()[output_port].iter().flat_map(|summary| summary.results_in(&self_time)).any(|time| time.less_equal(&self_time)) {
            Capability::new(self_time, self.internal.borrow()[output_port].clone())
        }
        else {
            panic!("Attempted to retain a time ({:?}) not greater or equal to the operators input-output summary ({:?}) applied to the capabilities time ({:?})", self_time, self.summaries.borrow()[output_port], self_time);
        }
    }
}

impl<T: Timestamp> Deref for InputCapability<T> {
    type Target = T;

    fn deref(&self) -> &T {
        self.time()
    }
}

impl<T: Timestamp> Debug for InputCapability<T> {
    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
        f.debug_struct("InputCapability")
            .field("time", self.time())
            .field("internal", &"...")
            .finish()
    }
}

/// Capability that activates on drop.
#[derive(Clone, Debug)]
pub struct ActivateCapability<T: Timestamp> {
    pub(crate) capability: Capability<T>,
    pub(crate) address: Rc<[usize]>,
    pub(crate) activations: Rc<RefCell<Activations>>,
}

impl<T: Timestamp> CapabilityTrait<T> for ActivateCapability<T> {
    fn time(&self) -> &T { self.capability.time() }
    fn valid_for_output(&self, query_buffer: &Rc<RefCell<ChangeBatch<T>>>) -> bool {
        self.capability.valid_for_output(query_buffer)
    }
}

impl<T: Timestamp> ActivateCapability<T> {
    /// Creates a new activating capability.
    pub fn new(capability: Capability<T>, address: Rc<[usize]>, activations: Rc<RefCell<Activations>>) -> Self {
        Self {
            capability,
            address,
            activations,
        }
    }

    /// The timestamp associated with this capability.
    pub fn time(&self) -> &T {
        self.capability.time()
    }

    /// Creates a new delayed capability.
    pub fn delayed(&self, time: &T) -> Self {
        ActivateCapability {
            capability: self.capability.delayed(time),
            address: self.address.clone(),
            activations: self.activations.clone(),
        }
    }

    /// Downgrades this capability.
    pub fn downgrade(&mut self, time: &T) {
        self.capability.downgrade(time);
        self.activations.borrow_mut().activate(&self.address);
    }
}

impl<T: Timestamp> Drop for ActivateCapability<T> {
    fn drop(&mut self) {
        self.activations.borrow_mut().activate(&self.address);
    }
}

/// A set of capabilities, for possibly incomparable times.
#[derive(Clone, Debug)]
pub struct CapabilitySet<T: Timestamp> {
    elements: Vec<Capability<T>>,
}

impl<T: Timestamp> CapabilitySet<T> {

    /// Allocates an empty capability set.
    pub fn new() -> Self {
        Self { elements: Vec::new() }
    }

    /// Allocates an empty capability set with space for `capacity` elements
    pub fn with_capacity(capacity: usize) -> Self {
        Self { elements: Vec::with_capacity(capacity) }
    }

    /// Allocates a capability set containing a single capability.
    ///
    /// # Examples
    /// ```
    /// use std::collections::HashMap;
    /// use timely::dataflow::{
    ///     operators::{ToStream, generic::Operator},
    ///     channels::pact::Pipeline,
    /// };
    /// use timely::dataflow::operators::CapabilitySet;
    ///
    /// timely::example(|scope| {
    ///     vec![()].into_iter().to_stream(scope)
    ///         .unary_frontier(Pipeline, "example", |default_cap, _info| {
    ///             let mut cap = CapabilitySet::from_elem(default_cap);
    ///             move |input, output| {
    ///                 cap.downgrade(&input.frontier().frontier());
    ///                 while let Some((time, data)) = input.next() {
    ///                 }
    ///                 let a_cap = cap.first();
    ///                 if let Some(a_cap) = a_cap.as_ref() {
    ///                     output.session(a_cap).give(());
    ///                 }
    ///             }
    ///         })
    ///         .container::<Vec<_>>();
    /// });
    /// ```
    pub fn from_elem(cap: Capability<T>) -> Self {
        Self { elements: vec![cap] }
    }

    /// Inserts `capability` into the set, discarding redundant capabilities.
    pub fn insert(&mut self, capability: Capability<T>) {
        if !self.elements.iter().any(|c| c.less_equal(&capability)) {
            self.elements.retain(|c| !capability.less_equal(c));
            self.elements.push(capability);
        }
    }

    /// Creates a new capability to send data at `time`.
    ///
    /// This method panics if there does not exist a capability in `self.elements` less or equal to `time`.
    pub fn delayed(&self, time: &T) -> Capability<T> {
        /// Makes the panic branch cold & outlined to decrease code bloat & give
        /// the inner function the best chance possible of being inlined with
        /// minimal code bloat
        #[cold]
        #[inline(never)]
        fn delayed_panic(invalid_time: &dyn Debug) -> ! {
            // Formatting & panic machinery is relatively expensive in terms of code bloat, so
            // we outline it
            panic!(
                "failed to create a delayed capability, the current set does not \
                have an element less than or equal to {:?}",
                invalid_time,
            )
        }

        self.try_delayed(time)
            .unwrap_or_else(|| delayed_panic(time))
    }

    /// Attempts to create a new capability to send data at `time`.
    ///
    /// Returns [`None`] if there does not exist a capability in `self.elements` less or equal to `time`.
    pub fn try_delayed(&self, time: &T) -> Option<Capability<T>> {
        self.elements
            .iter()
            .find(|capability| capability.time().less_equal(time))
            .and_then(|capability| capability.try_delayed(time))
    }

    /// Downgrades the set of capabilities to correspond with the times in `frontier`.
    ///
    /// This method panics if any element of `frontier` is not greater or equal to some element of `self.elements`.
    pub fn downgrade<B, F>(&mut self, frontier: F)
    where
        B: borrow::Borrow<T>,
        F: IntoIterator<Item = B>,
    {
        /// Makes the panic branch cold & outlined to decrease code bloat & give
        /// the inner function the best chance possible of being inlined with
        /// minimal code bloat
        #[cold]
        #[inline(never)]
        fn downgrade_panic() -> ! {
            // Formatting & panic machinery is relatively expensive in terms of code bloat, so
            // we outline it
            panic!(
                "Attempted to downgrade a CapabilitySet with a frontier containing an element \
                that was not beyond an element within the set"
            )
        }

        self.try_downgrade(frontier)
            .unwrap_or_else(|_| downgrade_panic())
    }

    /// Attempts to downgrade the set of capabilities to correspond with the times in `frontier`.
    ///
    /// Returns [`None`] if any element of `frontier` is not greater or equal to some element of `self.elements`.
    ///
    /// **Warning**: If an error is returned the capability set may be in an inconsistent state and can easily
    /// cause logic errors within the program if not properly handled.
    ///
    pub fn try_downgrade<B, F>(&mut self, frontier: F) -> Result<(), DowngradeError>
    where
        B: borrow::Borrow<T>,
        F: IntoIterator<Item = B>,
    {
        let count = self.elements.len();
        for time in frontier.into_iter() {
            let capability = self.try_delayed(time.borrow()).ok_or(DowngradeError(()))?;
            self.elements.push(capability);
        }
        self.elements.drain(..count);

        Ok(())
    }
}

impl<T> From<Vec<Capability<T>>> for CapabilitySet<T>
where
    T: Timestamp,
{
    fn from(capabilities: Vec<Capability<T>>) -> Self {
        let mut this = Self::with_capacity(capabilities.len());
        for capability in capabilities {
            this.insert(capability);
        }

        this
    }
}

impl<T: Timestamp> Default for CapabilitySet<T> {
    fn default() -> Self {
        Self::new()
    }
}

impl<T: Timestamp> Deref for CapabilitySet<T> {
    type Target=[Capability<T>];

    fn deref(&self) -> &[Capability<T>] {
        &self.elements
    }
}