dvcompute_gpss_cons 1.3.4

// Copyright (c) 2020-2022  David Sorokin <david.sorokin@gmail.com>, based in Yoshkar-Ola, Russia
//
// This Source Code Form is subject to the terms of the Mozilla Public
// License, v. 2.0. If a copy of the MPL was not distributed with this
// file, You can obtain one at https://mozilla.org/MPL/2.0/.

use std::rc::Rc;
use std::marker::PhantomData;

use dvcompute::simulation;
use dvcompute::simulation::error::*;
use dvcompute::simulation::Point;
use dvcompute::simulation::ref_comp::RefComp;
use dvcompute::simulation::observable::*;
use dvcompute::simulation::observable::source::*;
use dvcompute::simulation::event::*;
use dvcompute::simulation::process::*;
use dvcompute::simulation::strategy::QueueStorage;

use dvcompute_utils::simulation::stats::*;

use crate::simulation::transact::*;
use crate::simulation::strategy::*;

/// Represents a storage.
pub struct Storage<T> {

    /// The storage capacity.
    capacity: isize,

    /// The content.
    content: RefComp<isize>,

    /// The content statistics.
    content_stats: RefComp<TimingStats<isize>>,

    /// The content observable source.
    content_source: ObservableSource<isize>,

    /// The use count.
    use_count: RefComp<isize>,

    /// The use count observable source.
    use_count_source: ObservableSource<isize>,

    /// The used content.
    used_content: RefComp<isize>,

    /// The used content observable source.
    used_content_source: ObservableSource<isize>,

    /// The utilization.
    util_count: RefComp<isize>,

    /// The utilization statistics.
    util_count_stats: RefComp<TimingStats<isize>>,

    /// The utilization count observable source.
    util_count_source: ObservableSource<isize>,

    /// The queue length.
    queue_count: RefComp<isize>,

    /// The queue count statistics.
    queue_count_stats: RefComp<TimingStats<isize>>,

    /// The queue count observable source.
    queue_count_source: ObservableSource<isize>,

    /// The total wait time.
    total_wait_time: RefComp<f64>,

    /// The wait time.
    wait_time: RefComp<SamplingStats<f64>>,

    /// The wait time observable source.
    wait_time_source: ObservableSource<SamplingStats<f64>>,

    /// The Delay chain.
    delay_chain: FCFSStorage<StorageDelayedItem<T>>
}

impl<T> PartialEq for Storage<T> {

    fn eq(&self, other: &Self) -> bool {
        self.content == other.content
    }
}

impl<T> Eq for Storage<T> {}

/// Identifies a transact item that was delayed.
struct StorageDelayedItem<T> {

    /// The transact.
    transact: Rc<Transact<T>>,

    /// The time of delaying the transact.
    time: f64,

    /// The corresponding content decrement.
    decrement: isize,

    /// The continuation of the corresponding process.
    cont: FrozenProcess<()>
}

impl<T> Storage<T> {

    /// Create a new storage within `Event` computation by the specified capacity.
    #[inline]
    pub fn new(capacity: isize) -> NewStorage<T> {
        NewStorage { capacity: capacity, _phantom: PhantomData }
    }

    /// Return the storage capacity.
    pub fn capacity(&self) -> isize {
        self.capacity
    }

    /// Whether the storage is empty, i.e. completely unused.
    #[inline]
    pub fn is_empty(storage: Rc<Self>) -> impl Event<Item = bool> + Clone {
        cons_event(move |p| {
            Result::Ok(storage.content.read_at(p) == storage.capacity)
        })
    }

    /// Whether the storage is full, i.e. completely used.
    #[inline]
    pub fn is_full(storage: Rc<Self>) -> impl Event<Item = bool> + Clone {
        cons_event(move |p| {
            Result::Ok(storage.content.read_at(p) == 0)
        })
    }

    /// Return the current storage content.
    #[inline]
    pub fn content(storage: Rc<Self>) -> impl Event<Item = isize> + Clone {
        cons_event(move |p| {
            Result::Ok(storage.content.read_at(p))
        })
    }

    /// Return the statistics for the storage content.
    #[inline]
    pub fn content_stats(storage: Rc<Self>) -> impl Event<Item = TimingStats<isize>> + Clone {
        cons_event(move |p| {
            Result::Ok(storage.content_stats.read_at(p))
        })
    }

    /// Notifies when the `content` property changes.
    #[inline]
    pub fn content_changed(&self) -> impl Observable<Message = isize> + Clone {
        self.content_source.publish()
    }

    /// Notifies when the `content` property changes.
    #[inline]
    pub fn content_changed_(&self) -> impl Observable<Message = ()> + Clone {
        self.content_changed().map(move |_| { () })
    }

    /// Return the total use count of the storage.
    #[inline]
    pub fn use_count(storage: Rc<Self>) -> impl Event<Item = isize> + Clone {
        cons_event(move |p| {
            Result::Ok(storage.use_count.read_at(p))
        })
    }

    /// Notifies when the `use_count` property changes.
    #[inline]
    pub fn use_count_changed(&self) -> impl Observable<Message = isize> + Clone {
        self.use_count_source.publish()
    }

    /// Notifies when the `use_count` property changes.
    #[inline]
    pub fn use_count_changed_(&self) -> impl Observable<Message = ()> + Clone {
        self.use_count_changed().map(move |_| { () })
    }

    /// Return the total used content of the storage.
    #[inline]
    pub fn used_content(storage: Rc<Self>) -> impl Event<Item = isize> + Clone {
        cons_event(move |p| {
            Result::Ok(storage.used_content.read_at(p))
        })
    }

    /// Notifies when the `used_content` property changes.
    #[inline]
    pub fn used_content_changed(&self) -> impl Observable<Message = isize> + Clone {
        self.used_content_source.publish()
    }

    /// Notifies when the `used_content` property changes.
    #[inline]
    pub fn used_content_changed_(&self) -> impl Observable<Message = ()> + Clone {
        self.used_content_changed().map(move |_| { () })
    }

    /// Return the current utilization count of the storage.
    #[inline]
    pub fn util_count(storage: Rc<Self>) -> impl Event<Item = isize> + Clone {
        cons_event(move |p| {
            Result::Ok(storage.util_count.read_at(p))
        })
    }

    /// Return the statistics for utilization count of the storage.
    #[inline]
    pub fn util_count_stats(storage: Rc<Self>) -> impl Event<Item = TimingStats<isize>> + Clone {
        cons_event(move |p| {
            Result::Ok(storage.util_count_stats.read_at(p))
        })
    }

    /// Notifies when the `util_count` property changes.
    #[inline]
    pub fn util_count_changed(&self) -> impl Observable<Message = isize> + Clone {
        self.util_count_source.publish()
    }

    /// Notifies when the `util_count` property changes.
    #[inline]
    pub fn util_count_changed_(&self) -> impl Observable<Message = ()> + Clone {
        self.util_count_changed().map(move |_| { () })
    }

    /// Return the current queue length of the storage.
    #[inline]
    pub fn queue_count(storage: Rc<Self>) -> impl Event<Item = isize> + Clone {
        cons_event(move |p| {
            Result::Ok(storage.queue_count.read_at(p))
        })
    }

    /// Return the statistics for queue length of the storage.
    #[inline]
    pub fn queue_count_stats(storage: Rc<Self>) -> impl Event<Item = TimingStats<isize>> + Clone {
        cons_event(move |p| {
            Result::Ok(storage.queue_count_stats.read_at(p))
        })
    }

    /// Notifies when the `queue_count` property changes.
    #[inline]
    pub fn queue_count_changed(&self) -> impl Observable<Message = isize> + Clone {
        self.queue_count_source.publish()
    }

    /// Notifies when the `queue_count` property changes.
    #[inline]
    pub fn queue_count_changed_(&self) -> impl Observable<Message = ()> + Clone {
        self.queue_count_changed().map(move |_| { () })
    }

    /// Return the total wait time of the storage.
    #[inline]
    pub fn total_wait_time(storage: Rc<Self>) -> impl Event<Item = f64> + Clone {
        cons_event(move |p| {
            Result::Ok(storage.total_wait_time.read_at(p))
        })
    }

    /// Return the statistics for wait time of the storage.
    #[inline]
    pub fn wait_time(storage: Rc<Self>) -> impl Event<Item = SamplingStats<f64>> + Clone {
        cons_event(move |p| {
            Result::Ok(storage.wait_time.read_at(p))
        })
    }

    /// Notifies when the `wait_time` property changes.
    #[inline]
    pub fn wait_time_changed(&self) -> impl Observable<Message = SamplingStats<f64>> + Clone {
        self.wait_time_source.publish()
    }

    /// Notifies when the `wait_time` property changes.
    #[inline]
    pub fn wait_time_changed_(&self) -> impl Observable<Message = ()> + Clone {
        self.wait_time_changed().map(move |_| { () })
    }

    /// Return the average holding time per unit.
    #[inline]
    pub fn average_holding_time(storage: Rc<Self>) -> impl Event<Item = f64> + Clone {
        cons_event(move |p| {
            let s = storage.util_count_stats.read_at(p);
            let n = storage.util_count.read_at(p);
            let m = storage.used_content.read_at(p);
            let t = p.time;
            let s2 = s.add(t, n);
            Result::Ok(s2.sum / (m as f64))
        })
    }

    /// Update the storage content.
    #[inline]
    fn update_content(&self, delta: isize, p: &Point) -> simulation::Result<()> {
        let a  = self.content.read_at(p);
        let a2 = a + delta;
        let stats  = self.content_stats.read_at(p);
        let stats2 = stats.add(p.time, a2);
        self.content.write_at(a2, p);
        self.content_stats.write_at(stats2, p);
        self.content_source.trigger_at(&a2, p)
    }

    /// Update the total storage use count.
    #[inline]
    fn update_use_count(&self, delta: isize, p: &Point) -> simulation::Result<()> {
        let a  = self.use_count.read_at(p);
        let a2 = a + delta;
        self.use_count.write_at(a2, p);
        self.use_count_source.trigger_at(&a2, p)
    }

    /// Update the total used storage content.
    #[inline]
    fn update_used_content(&self, delta: isize, p: &Point) -> simulation::Result<()> {
        let a  = self.used_content.read_at(p);
        let a2 = a + delta;
        self.used_content.write_at(a2, p);
        self.used_content_source.trigger_at(&a2, p)
    }

    /// Update the queue count.
    #[inline]
    fn update_queue_count(&self, delta: isize, p: &Point) -> simulation::Result<()> {
        let a  = self.queue_count.read_at(p);
        let a2 = a + delta;
        let stats  = self.queue_count_stats.read_at(p);
        let stats2 = stats.add(p.time, a2);
        self.queue_count.write_at(a2, p);
        self.queue_count_stats.write_at(stats2, p);
        self.queue_count_source.trigger_at(&a2, p)
    }

    /// Update the utilization count.
    #[inline]
    fn update_util_count(&self, delta: isize, p: &Point) -> simulation::Result<()> {
        let a  = self.util_count.read_at(p);
        let a2 = a + delta;
        let stats  = self.util_count_stats.read_at(p);
        let stats2 = stats.add(p.time, a2);
        self.util_count.write_at(a2, p);
        self.util_count_stats.write_at(stats2, p);
        self.util_count_source.trigger_at(&a2, p)
    }

    /// Update the wait time.
    #[inline]
    fn update_wait_time(&self, delta: f64, p: &Point) -> simulation::Result<()> {
        let a  = self.total_wait_time.read_at(p);
        let a2 = a + delta;
        let stats  = self.wait_time.read_at(p);
        let stats2 = stats.add(delta);
        self.total_wait_time.write_at(a2, p);
        self.wait_time.write_at(stats2, p);
        self.wait_time_source.trigger_at(&stats2, p)
    }

    /// Triggered when one of the `Storage` properties changes.
    #[inline]
    pub fn changed_(&self) -> impl Observable<Message = ()> + Clone {
        self.content_changed_()
            .merge(self.used_content_changed_())
            .merge(self.util_count_changed_())
            .merge(self.queue_count_changed_())
    }

    /// Reset the statistics.
    #[inline]
    pub fn reset(storage: Rc<Self>) -> impl Event<Item = ()> + Clone {
        cons_event(move |p| {
            let t = p.time;
            let content = storage.content.read_at(p);
            let used_content = storage.capacity - content;
            let util_count = storage.util_count.read_at(p);
            let queue_count = storage.queue_count.read_at(p);
            storage.content_stats.write_at(TimingStats::from_sample(t, content), p);
            storage.use_count.write_at(0, p);
            storage.used_content.write_at(used_content, p);
            storage.util_count_stats.write_at(TimingStats::from_sample(t, util_count), p);
            storage.queue_count_stats.write_at(TimingStats::from_sample(t, queue_count), p);
            storage.total_wait_time.write_at(0.0, p);
            storage.wait_time.write_at(SamplingStats::empty(), p);
            storage.content_source.trigger_at(&content, p)?;
            storage.use_count_source.trigger_at(&0, p)?;
            storage.used_content_source.trigger_at(&used_content, p)?;
            storage.util_count_source.trigger_at(&util_count, p)?;
            storage.queue_count_source.trigger_at(&queue_count, p)?;
            storage.wait_time_source.trigger_at(&SamplingStats::empty(), p)
        })
    }
}

/// Computation that creates a new storage.
#[derive(Clone)]
pub struct NewStorage<T> {

    /// The storage capacity.
    capacity: isize,

    /// To keep the data type.
    _phantom: PhantomData<T>
}

impl<T> Event for NewStorage<T> {

    type Item = Storage<T>;

    #[doc(hidden)]
    #[inline]
    fn call_event(self, p: &Point) -> simulation::Result<Self::Item> {
        let t = p.time;
        let capacity = self.capacity;
        Result::Ok(Storage {
            capacity: capacity,
            content: RefComp::new(capacity),
            content_stats: RefComp::new(TimingStats::from_sample(t, capacity)),
            content_source: ObservableSource::new(),
            use_count: RefComp::new(0),
            use_count_source: ObservableSource::new(),
            used_content: RefComp::new(0),
            used_content_source: ObservableSource::new(),
            util_count: RefComp::new(0),
            util_count_stats: RefComp::new(TimingStats::from_sample(t, 0)),
            util_count_source: ObservableSource::new(),
            queue_count: RefComp::new(0),
            queue_count_stats: RefComp::new(TimingStats::from_sample(t, 0)),
            queue_count_source: ObservableSource::new(),
            total_wait_time: RefComp::new(0.0),
            wait_time: RefComp::new(SamplingStats::empty()),
            wait_time_source: ObservableSource::new(),
            delay_chain: FCFSStorage::new()
        })
    }
}

/// Enter the storage.
#[must_use = "computations are lazy and do nothing unless to be run"]
#[derive(Clone)]
pub struct Enter<T> {

    /// The storage.
    storage: Rc<Storage<T>>,

    /// The transact.
    transact: Rc<Transact<T>>,

    /// The decrement.
    decrement: isize
}

impl<T> Process for Enter<T>
    where T: 'static
{
    type Item = ();

    #[doc(hidden)]
    #[inline]
    fn call_process<C>(self, cont: C, pid: Rc<ProcessId>, p: &Point) -> simulation::Result<()>
        where C: FnOnce(simulation::Result<Self::Item>, Rc<ProcessId>, &Point) -> simulation::Result<()> + 'static
    {
        let cont = ProcessBoxCont::new(cont);
        self.call_process_boxed(cont, pid, p)
    }

    #[doc(hidden)]
    fn call_process_boxed(self, cont: ProcessBoxCont<Self::Item>, pid: Rc<ProcessId>, p: &Point) -> simulation::Result<()> {
        let Enter { storage, transact, decrement } = self;
        let t = p.time;
        let f = storage.delay_chain.is_empty(p);
        if f {
            enter_free_storage_boxed(storage, transact, decrement, cont, pid, p)
        } else {
            let comp = enter_storage(storage.clone(), transact.clone(), decrement);
            let cont = FrozenProcess::with_reentering(cont, pid, (), comp, p)?;
            let priority = Transact::priority_at(&transact, p);
            let item = StorageDelayedItem {
                transact: transact,
                time: t,
                decrement: decrement,
                cont: cont
            };
            storage.delay_chain.push_with_priority(priority, item, p);
            storage.update_queue_count(1, p)
        }
    }
}

/// Enter the storage within `Process` computation.
#[inline]
pub fn enter_storage<T>(storage: Rc<Storage<T>>, transact: Rc<Transact<T>>, decrement: isize) -> Enter<T> {
    Enter { storage: storage, transact: transact, decrement: decrement }
}

/// Enter the storage that has an empty chain.
fn enter_free_storage_boxed<T>(storage: Rc<Storage<T>>, transact: Rc<Transact<T>>, decrement: isize,
    cont: ProcessBoxCont<()>, pid: Rc<ProcessId>, p: &Point) -> simulation::Result<()>
        where T: 'static
{
    let t = p.time;
    let a = storage.content.read_at(p);
    if a < decrement {
        let comp = enter_storage(storage.clone(), transact.clone(), decrement);
        let cont = FrozenProcess::with_reentering(cont, pid, (), comp, p)?;
        let priority = Transact::priority_at(&transact, p);
        let item = StorageDelayedItem {
            transact: transact,
            time: t,
            decrement: decrement,
            cont: cont
        };
        storage.delay_chain.push_with_priority(priority, item, p);
        storage.update_queue_count(1, p)
    } else {
        storage.update_wait_time(0.0, p)?;
        storage.update_content(- decrement, p)?;
        storage.update_use_count(1, p)?;
        storage.update_used_content(decrement, p)?;
        storage.update_util_count(decrement, p)?;
        resume_process_boxed(cont, pid, (), p)
    }
}

/// Leave the storage.
#[must_use = "computations are lazy and do nothing unless to be run"]
#[derive(Clone)]
pub struct Leave<T> {

    /// The storage.
    storage: Rc<Storage<T>>,

    /// The increment.
    increment: isize
}

impl<T> Process for Leave<T>
    where T: 'static
{
    type Item = ();

    #[doc(hidden)]
    #[inline]
    fn call_process<C>(self, cont: C, pid: Rc<ProcessId>, p: &Point) -> simulation::Result<()>
        where C: FnOnce(simulation::Result<Self::Item>, Rc<ProcessId>, &Point) -> simulation::Result<()> + 'static
    {
        let Leave { storage, increment } = self;
        leave_storage_within_event(storage, increment).call_event(p)?;
        resume_process(cont, pid, (), p)
    }

    #[doc(hidden)]
    fn call_process_boxed(self, cont: ProcessBoxCont<Self::Item>, pid: Rc<ProcessId>, p: &Point) -> simulation::Result<()> {
        let Leave { storage, increment } = self;
        leave_storage_within_event(storage, increment).call_event(p)?;
        resume_process_boxed(cont, pid, (), p)
    }
}

/// Leave the storage within `Process` computation.
#[inline]
pub fn leave_storage<T>(storage: Rc<Storage<T>>, increment: isize) -> Leave<T> {
    Leave { storage: storage, increment: increment }
}

/// Leave the storage within `Event` computation.
#[must_use = "computations are lazy and do nothing unless to be run"]
#[derive(Clone)]
pub struct LeaveWithinEvent<T> {

    /// The storage.
    storage: Rc<Storage<T>>,

    /// The increment.
    increment: isize
}

impl<T> Event for LeaveWithinEvent<T>
    where T: 'static
{
    type Item = ();

    #[doc(hidden)]
    fn call_event(self, p: &Point) -> simulation::Result<Self::Item> {
        let LeaveWithinEvent { storage, increment } = self;
        let t = p.time;
        storage.update_util_count(- increment, p)?;
        storage.update_content(increment, p)?;
        enqueue_event(t, {
            cons_event(move |p| {
                try_enter_storage(storage, p)
            }).into_boxed()
        }).call_event(p)
    }
}

/// Leave the storage within `Event` computation.
#[inline]
pub fn leave_storage_within_event<T>(storage: Rc<Storage<T>>, increment: isize) -> LeaveWithinEvent<T> {
    LeaveWithinEvent { storage: storage, increment: increment }
}

/// Try to enter the storage.
#[inline]
fn try_enter_storage<T>(storage: Rc<Storage<T>>, p: &Point) -> simulation::Result<()> {
    let a = storage.content.read_at(p);
    if a > 0 {
        let_enter_storage(storage, p)
    } else {
        Result::Ok(())
    }
}

/// Let enter the storage.
#[inline]
fn let_enter_storage<T>(storage: Rc<Storage<T>>, p: &Point) -> simulation::Result<()> {
    let t = p.time;
    let a = storage.content.read_at(p);
    if a > storage.capacity {
        let msg = String::from("The storage content cannot exceed the limited capacity");
        let err = Error::retry(msg);
        Result::Err(err)
    } else {
        match storage.delay_chain.remove_by(move |i| { i.decrement <= a }, p) {
            None => Result::Ok(()),
            Some(StorageDelayedItem { transact: transact0, time: t0, decrement: decrement0, cont: cont0 }) => {
                storage.update_queue_count(-1, p)?;
                match cont0.unfreeze(p)? {
                    None => let_enter_storage(storage, p),
                    Some(cont) => {
                        let pid = transact0.require_process_id(p)?;
                        storage.update_content(- decrement0, p)?;
                        storage.update_wait_time(t - t0, p)?;
                        storage.update_util_count(decrement0, p)?;
                        storage.update_use_count(1, p)?;
                        storage.update_used_content(decrement0, p)?;
                        enqueue_event(t, {
                            cons_event(move |p| {
                                reenter_process(cont, pid, (), p)
                            }).into_boxed()
                        }).call_event(p)
                    }
                }
            }
        }
    }
}