dvcompute_cons/simulation/queue/stats/

mod.rs

// Copyright (c) 2020-2022  David Sorokin <davsor@mail.ru>, 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::marker::PhantomData;

use crate::simulation;
use crate::simulation::error::*;
use crate::simulation::Point;
use crate::simulation::ref_comp::RefComp;
use crate::simulation::observable::*;
use crate::simulation::observable::source::*;
use crate::simulation::simulation::*;
use crate::simulation::event::*;
use crate::simulation::process::*;
use crate::simulation::strategy::*;
use crate::simulation::resource::*;

use dvcompute_utils::simulation::stats::*;
use dvcompute_utils::grc::Grc;

/// A type synonym for the ordinary FIFO queue, also known as the FCFS
/// (First Come - First Serviced) queue.
pub type FCFSQueue<T> = Queue<FCFSStrategy, FCFSStrategy, FCFSStrategy, T>;

/// A type synonym for the ordinary LIFO queue, also known as the LCFS
/// (Last Come - First Serviced) queue.
pub type LCFSQueue<T> = Queue<FCFSStrategy, LCFSStrategy, FCFSStrategy, T>;

/// Represents a bounded queue by using the specified strategies for enqueueing (input), `SI`,
/// internal storing (in memory), `SM`, and dequeueing (output), `SO`, where `T` denotes
/// the type of items stored in the queue.
pub struct Queue<SI, SM, SO, T>
    where SI: QueueStrategy + 'static,
          SM: QueueStrategy,
          SO: QueueStrategy + 'static,
          T: 'static
{
    /// The queue capacity.
    max_count: isize,

    /// The enqueue resource.
    enqueue_resource: Grc<Resource<SI>>,

    /// The queue store.
    queue_store: QueueStorageBox<QueueItem<T>, SM::Priority>,

    /// The dequeue resource.
    dequeue_resource: Grc<Resource<SO>>,

    /// The queue size.
    count: RefComp<isize>,

    /// The size statistics.
    count_stats: RefComp<TimingStats<isize>>,

    /// The enqueue count.
    enqueue_count: RefComp<isize>,

    /// The count of lost items.
    enqueue_lost_count: RefComp<isize>,

    /// The count of stored items.
    enqueue_store_count: RefComp<isize>,

    /// The dequeue count.
    dequeue_count: RefComp<isize>,

    /// The count of extracted items.
    dequeue_extract_count: RefComp<isize>,

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

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

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

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

    /// The observable source when the enqueue is initiated.
    enqueue_initiated_source: ObservableSource<T>,

    /// The observable source when the item is lost.
    enqueue_lost_source: ObservableSource<T>,

    /// The observable source when the item is stored.
    enqueue_stored_source: ObservableSource<T>,

    /// The observable source when the item is requested for.
    dequeue_requested_source: ObservableSource<()>,

    /// The observable source when the item is extracted.
    dequeue_extracted_source: ObservableSource<T>
}

/// Stores the item and a time of its enqueueing.
#[derive(Clone)]
struct QueueItem<T> {

    /// The item value.
    value: T,

    /// The time of enqueueing the item.
    input_time: f64,

    /// The time of storing the item.
    storing_time: f64
}

/// Create a new bounded FCFS (a.k.a FIFO) queue by the specified capacity.
#[inline]
pub fn new_fcfs_queue<T>(max_count: isize) -> NewQueue<FCFSStrategy, FCFSStrategy, FCFSStrategy, T>
    where T: 'static
{
    NewQueue {
        enqueue_strategy: FCFSStrategy::Instance,
        storing_strategy: FCFSStrategy::Instance,
        dequeue_strategy: FCFSStrategy::Instance,
        max_count: max_count,
        _phantom: PhantomData
    }
}

/// Create a new bounded LCFS (a.k.a LIFO) queue by the specified capacity.
#[inline]
pub fn new_lcfs_queue<T>(max_count: isize) -> NewQueue<FCFSStrategy, LCFSStrategy, FCFSStrategy, T>
    where T: 'static
{
    NewQueue {
        enqueue_strategy: FCFSStrategy::Instance,
        storing_strategy: LCFSStrategy::Instance,
        dequeue_strategy: FCFSStrategy::Instance,
        max_count: max_count,
        _phantom: PhantomData
    }
}

impl<SI, SM, SO, T> Queue<SI, SM, SO, T>
    where SI: QueueStrategy + 'static,
          SM: QueueStrategy + 'static,
          SO: QueueStrategy + 'static,
          T: Clone + 'static
{
    /// Create a new bounded queue by the specified strategies and capacity.
    #[inline]
    pub fn new(enqueue_strategy: SI,
        storing_strategy: SM,
        dequeue_strategy: SO,
        max_count: isize) -> NewQueue<SI, SM, SO, T>
    {
        NewQueue {
            enqueue_strategy: enqueue_strategy,
            storing_strategy: storing_strategy,
            dequeue_strategy: dequeue_strategy,
            max_count: max_count,
            _phantom: PhantomData
        }
    }

    /// Return the queue capacity, i.e. its maximum size.
    #[inline]
    pub fn max_count(&self) -> isize {
        self.max_count
    }

    /// Test whether the queue is empty.
    #[inline]
    pub fn is_empty(queue: Grc<Self>) -> impl Event<Item = bool> + Clone {
        cons_event(move |p| {
            Result::Ok(queue.count.read_at(p) == 0)
        })
    }

    /// Notifies when the `is_empty` property changes.
    #[inline]
    pub fn is_empty_changed(queue: Grc<Self>) -> impl Observable<Message = bool> + Clone {
        queue.is_empty_changed_()
            .mapc(move |()| {
                Queue::is_empty(queue.clone())
            })
    }

    /// Notifies when the `is_empty` property changes.
    #[inline]
    pub fn is_empty_changed_(&self) -> impl Observable<Message = ()> + Clone {
        self.count_changed_()
    }

    /// Test whether the queue is full.
    #[inline]
    pub fn is_full(queue: Grc<Self>) -> impl Event<Item = bool> + Clone {
        cons_event(move |p| {
            Result::Ok(queue.count.read_at(p) == queue.max_count)
        })
    }

    /// Notifies when the `is_full` property changes.
    #[inline]
    pub fn is_full_changed(queue: Grc<Self>) -> impl Observable<Message = bool> + Clone {
        queue.is_full_changed_()
            .mapc(move |()| {
                Queue::is_full(queue.clone())
            })
    }

    /// Notifies when the `is_full` property changes.
    #[inline]
    pub fn is_full_changed_(&self) -> impl Observable<Message = ()> + Clone {
        self.count_changed_()
    }

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

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

    /// Notifies when the `count` property changes.
    #[inline]
    pub fn count_changed(queue: Grc<Self>) -> impl Observable<Message = isize> + Clone {
        queue.count_changed_()
            .mapc(move |()| {
                Queue::count(queue.clone())
            })
    }

    /// Notifies when the `count` property changes.
    #[inline]
    pub fn count_changed_(&self) -> impl Observable<Message = ()> + Clone {
        self.enqueue_stored().map(|_| {})
            .merge(self.dequeue_extracted().map(|_| {}))
    }

    /// Return the total number of enqueue operations, including those ones that have failed due to full capacity.
    #[inline]
    pub fn enqueue_count(queue: Grc<Self>) -> impl Event<Item = isize> + Clone {
        cons_event(move |p| {
            Result::Ok(queue.enqueue_count.read_at(p))
        })
    }

    /// Notifies when the `enqueue_count` property changes.
    #[inline]
    pub fn enqueue_count_changed(queue: Grc<Self>) -> impl Observable<Message = isize> + Clone {
        queue.enqueue_count_changed_()
            .mapc(move |()| {
                Queue::enqueue_count(queue.clone())
            })
    }

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

    /// Return the total number of items that could not be enqueued due to full capacity.
    #[inline]
    pub fn enqueue_lost_count(queue: Grc<Self>) -> impl Event<Item = isize> + Clone {
        cons_event(move |p| {
            Result::Ok(queue.enqueue_lost_count.read_at(p))
        })
    }

    /// Notifies when the `enqueue_lost_count` property changes.
    #[inline]
    pub fn enqueue_lost_count_changed(queue: Grc<Self>) -> impl Observable<Message = isize> + Clone {
        queue.enqueue_lost_count_changed_()
            .mapc(move |()| {
                Queue::enqueue_lost_count(queue.clone())
            })
    }

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

    /// Return the total number of input items that were stored.
    #[inline]
    pub fn enqueue_store_count(queue: Grc<Self>) -> impl Event<Item = isize> + Clone {
        cons_event(move |p| {
            Result::Ok(queue.enqueue_store_count.read_at(p))
        })
    }

    /// Notifies when the `enqueue_store_count` property changes.
    #[inline]
    pub fn enqueue_store_count_changed(queue: Grc<Self>) -> impl Observable<Message = isize> + Clone {
        queue.enqueue_store_count_changed_()
            .mapc(move |()| {
                Queue::enqueue_store_count(queue.clone())
            })
    }

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

    /// Return the total number of requests to dequeue the items.
    #[inline]
    pub fn dequeue_count(queue: Grc<Self>) -> impl Event<Item = isize> + Clone {
        cons_event(move |p| {
            Result::Ok(queue.dequeue_count.read_at(p))
        })
    }

    /// Notifies when the `dequeue_count` property changes.
    #[inline]
    pub fn dequeue_count_changed(queue: Grc<Self>) -> impl Observable<Message = isize> + Clone {
        queue.dequeue_count_changed_()
            .mapc(move |()| {
                Queue::dequeue_count(queue.clone())
            })
    }

    /// Notifies when the `dequeue_count` property changes.
    #[inline]
    pub fn dequeue_count_changed_(&self) -> impl Observable<Message = ()> + Clone {
        self.dequeue_requested()
    }

    /// Return the total number of items that were extracted from the queue with help of dequeue operations.
    #[inline]
    pub fn dequeue_extract_count(queue: Grc<Self>) -> impl Event<Item = isize> + Clone {
        cons_event(move |p| {
            Result::Ok(queue.dequeue_extract_count.read_at(p))
        })
    }

    /// Notifies when the `dequeue_extract_count` property changes.
    #[inline]
    pub fn dequeue_extract_count_changed(queue: Grc<Self>) -> impl Observable<Message = isize> + Clone {
        queue.dequeue_extract_count_changed_()
            .mapc(move |()| {
                Queue::dequeue_extract_count(queue.clone())
            })
    }

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

    /// Return the load factor: the queue size divided by its capacity, i.e. maximum size.
    #[inline]
    pub fn load_factor(queue: Grc<Self>) -> impl Event<Item = f64> + Clone {
        cons_event(move |p| {
            Result::Ok({
                let x = queue.count.read_at(p);
                let y = queue.max_count;
                (x as f64) / (y as f64)
            })
        })
    }

    /// Notifies when the `load_factor` property changes.
    #[inline]
    pub fn load_factor_changed(queue: Grc<Self>) -> impl Observable<Message = f64> + Clone {
        queue.load_factor_changed_()
            .mapc(move |()| {
                Queue::load_factor(queue.clone())
            })
    }

    /// Notifies when the `load_factor` property changes.
    #[inline]
    pub fn load_factor_changed_(&self) -> impl Observable<Message = ()> + Clone {
        self.count_changed_()
    }

    /// Return the rate of input items that were enqueued: how many items per time.
    #[inline]
    pub fn enqueue_rate(queue: Grc<Self>) -> impl Event<Item = f64> + Clone {
        cons_event(move |p| {
            Result::Ok({
                let x  = queue.enqueue_count.read_at(p);
                let t0 = p.run.specs.start_time;
                let t  = p.time;
                (x as f64) / (t - t0)
            })
        })
    }

    /// Return the rate of input items that were stored: how many items per time.
    #[inline]
    pub fn store_rate(queue: Grc<Self>) -> impl Event<Item = f64> + Clone {
        cons_event(move |p| {
            Result::Ok({
                let x  = queue.enqueue_store_count.read_at(p);
                let t0 = p.run.specs.start_time;
                let t  = p.time;
                (x as f64) / (t - t0)
            })
        })
    }

    /// Return the rate of requests for dequeueing the items: how many items per time.
    /// It does not include the failed attempts to dequeue immediately without suspension.
    #[inline]
    pub fn dequeue_rate(queue: Grc<Self>) -> impl Event<Item = f64> + Clone {
        cons_event(move |p| {
            Result::Ok({
                let x  = queue.dequeue_count.read_at(p);
                let t0 = p.run.specs.start_time;
                let t  = p.time;
                (x as f64) / (t - t0)
            })
        })
    }

    /// Return the rate of output items that were actually extracted from the queue: how many items per time.
    #[inline]
    pub fn dequeue_extract_rate(queue: Grc<Self>) -> impl Event<Item = f64> + Clone {
        cons_event(move |p| {
            Result::Ok({
                let x  = queue.dequeue_extract_count.read_at(p);
                let t0 = p.run.specs.start_time;
                let t  = p.time;
                (x as f64) / (t - t0)
            })
        })
    }

    /// Return the wait time from the time at which the item was stored in the queue to
    /// the time at which it was dequeued.
    #[inline]
    pub fn wait_time(queue: Grc<Self>) -> impl Event<Item = SamplingStats<f64>> + Clone {
        cons_event(move |p| {
            Result::Ok(queue.wait_time.read_at(p))
        })
    }

    /// Notifies when the `wait_time` property changes.
    #[inline]
    pub fn wait_time_changed(queue: Grc<Self>) -> impl Observable<Message = SamplingStats<f64>> + Clone {
        queue.wait_time_changed_()
            .mapc(move |()| {
                Queue::wait_time(queue.clone())
            })
    }

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

    /// Return the total wait time from the time at which the enqueue operation
    /// was initiated to the time at which the item was dequeued.
    ///
    /// In some sense, `total_wait_time` == `enqueue_wait_time` + `wait_time`.
    #[inline]
    pub fn total_wait_time(queue: Grc<Self>) -> impl Event<Item = SamplingStats<f64>> + Clone {
        cons_event(move |p| {
            Result::Ok(queue.total_wait_time.read_at(p))
        })
    }

    /// Notifies when the `total_wait_time` property changes.
    #[inline]
    pub fn total_wait_time_changed(queue: Grc<Self>) -> impl Observable<Message = SamplingStats<f64>> + Clone {
        queue.total_wait_time_changed_()
            .mapc(move |()| {
                Queue::total_wait_time(queue.clone())
            })
    }

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

    /// Return the enqueue wait time from the time at which the enqueue operation
    /// was initiated to the time at which the item was stored in the queue.
    #[inline]
    pub fn enqueue_wait_time(queue: Grc<Self>) -> impl Event<Item = SamplingStats<f64>> + Clone {
        cons_event(move |p| {
            Result::Ok(queue.enqueue_wait_time.read_at(p))
        })
    }

    /// Notifies when the `enqueue_wait_time` property changes.
    #[inline]
    pub fn enqueue_wait_time_changed(queue: Grc<Self>) -> impl Observable<Message = SamplingStats<f64>> + Clone {
        queue.enqueue_wait_time_changed_()
            .mapc(move |()| {
                Queue::enqueue_wait_time(queue.clone())
            })
    }

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

    /// Return the dequeue wait time from the time at which the dequeue request was made
    /// to the time at which the corresponding item was actually dequeued.
    #[inline]
    pub fn dequeue_wait_time(queue: Grc<Self>) -> impl Event<Item = SamplingStats<f64>> + Clone {
        cons_event(move |p| {
            Result::Ok(queue.dequeue_wait_time.read_at(p))
        })
    }

    /// Notifies when the `dequeue_wait_time` property changes.
    #[inline]
    pub fn dequeue_wait_time_changed(queue: Grc<Self>) -> impl Observable<Message = SamplingStats<f64>> + Clone {
        queue.dequeue_wait_time_changed_()
            .mapc(move |()| {
                Queue::dequeue_wait_time(queue.clone())
            })
    }

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

    /// Return a long-term average queue rate calculated as
    /// the average queue size divided by the average wait time.
    ///
    /// This value may be less than the actual arrival rate as the queue is
    /// bounded and new arrivals may be blocked while the queue remains full.
    #[inline]
    pub fn rate(queue: Grc<Self>) -> impl Event<Item = f64> + Clone {
        cons_event(move |p| {
            Result::Ok({
                let x = queue.count_stats.read_at(p);
                let y = queue.wait_time.read_at(p);
                x.mean() / y.mean
            })
        })
    }

    /// Notifies when the `rate` property changes.
    #[inline]
    pub fn rate_changed(queue: Grc<Self>) -> impl Observable<Message = f64> + Clone {
        queue.rate_changed_()
            .mapc(move |()| {
                Queue::rate(queue.clone())
            })
    }

    /// Notifies when the `rate` property changes.
    #[inline]
    pub fn rate_changed_(&self) -> impl Observable<Message = ()> + Clone {
        self.enqueue_stored().map(|_| {})
            .merge(self.dequeue_extracted().map(|_| {}))
    }

    /// Dequeue by suspending the process if the queue is empty.
    pub fn dequeue(queue: Grc<Self>) -> impl Process<Item = T> {
        cons_event({
            let queue = queue.clone();
            move |p| {
                queue.dequeue_request(p)
            }
        })
        .into_process()
        .and_then(move |t| {
            request_resource(queue.dequeue_resource.clone())
                .and_then(move |()| {
                    cons_event(move |p| {
                        queue.dequeue_extract(t, p)
                    })
                    .into_process()
                })
        })
    }

    /// Dequeue with output prioerity by suspending the process if the queue is empty.
    pub fn dequeue_with_output_priority(queue: Grc<Self>, po: SO::Priority) -> impl Process<Item = T>
        where SO::Priority: Clone
    {
        cons_event({
            let queue = queue.clone();
            move |p| {
                queue.dequeue_request(p)
            }
        })
        .into_process()
        .and_then(move |t| {
            request_resource_with_priority(queue.dequeue_resource.clone(), po)
                .and_then(move |()| {
                    cons_event(move |p| {
                        queue.dequeue_extract(t, p)
                    })
                    .into_process()
                })
        })
    }

    /// Try to dequeue immediately.
    pub fn try_dequeue(queue: Grc<Self>) -> impl Event<Item = Option<T>> {
        try_request_resource_within_event(queue.dequeue_resource.clone())
            .and_then(move |f| {
                if f {
                    cons_event(move |p| {
                        let t = queue.dequeue_request(p)?;
                        let x = queue.dequeue_extract(t, p)?;
                        Result::Ok(Some(x))
                    }).into_boxed()
                } else {
                    return_event(None)
                        .into_boxed()
                }
            })
    }

    /// Remove the item from the queue and return a flag indicating
    /// whether the item was found and actually removed.
    pub fn delete(queue: Grc<Self>, item: T) -> impl Event<Item = bool>
        where T: PartialEq
    {
        let pred = move |x: &T| { *x == item };
        Queue::delete_by(queue, pred)
            .map(|x| { x.is_some() })
    }

    /// Remove the specified item from the queue.
    pub fn delete_(queue: Grc<Self>, item: T) -> impl Event<Item = ()>
        where T: PartialEq
    {
        let pred = move |x: &T| { *x == item };
        Queue::delete_by(queue, pred)
            .map(|_| ())
    }

    /// Remove an item satisfying the specified predicate and return the item if found.
    pub fn delete_by<F>(queue: Grc<Self>, pred: F) -> impl Event<Item = Option<T>>
        where F: Fn(&T) -> bool + 'static
    {
        try_request_resource_within_event(queue.dequeue_resource.clone())
            .and_then(move |f| {
                if f {
                    cons_event(move |p| {
                        let pred = move |x: &QueueItem<T>| { pred(&x.value) };
                        let pred = Box::new(pred);
                        match queue.queue_store.remove_boxed_by(pred, p) {
                            None => {
                                release_resource_within_event(queue.dequeue_resource.clone())
                                    .call_event(p)?;
                                Result::Ok(None)
                            },
                            Some(i) => {
                                let t = queue.dequeue_request(p)?;
                                let x = queue.dequeue_post_extract(t, i, p)?;
                                Result::Ok(Some(x))
                            }
                        }
                    }).into_boxed()
                } else {
                    return_event(None)
                        .into_boxed()
                }
            })
    }

    /// Test whether there is an item satisfying the specified predicate.
    pub fn exists<F>(queue: Grc<Self>, pred: F) -> impl Event<Item = bool>
        where F: Fn(&T) -> bool + 'static
    {
        cons_event(move |p| {
            let pred = move |x: &QueueItem<T>| { pred(&x.value) };
            let pred = Box::new(pred);
            Result::Ok(queue.queue_store.exists_boxed(pred, p))
        })
    }

    /// Find an item satisfying the specified predicate.
    pub fn find<F>(queue: Grc<Self>, pred: F) -> impl Event<Item = Option<T>>
        where F: Fn(&T) -> bool + 'static,
              T: Clone
    {
        cons_event(move |p| {
            let pred = move |x: &QueueItem<T>| { pred(&x.value) };
            let pred = Box::new(pred);
            Result::Ok(queue.queue_store.find_boxed(pred, p).map(|x| { x.value.clone() }))
        })
    }

    /// Clear the queue.
    pub fn clear(queue: Grc<Self>) -> impl Event<Item = ()> {
        cons_event(move |p| {
            loop {
                let x = Queue::try_dequeue(queue.clone()).call_event(p)?;
                match x {
                    None => return Result::Ok(()),
                    Some(_) => {}
                }
            }
        })
    }

    /// Enqueue the item by suspending the process if the queue is full.
    pub fn enqueue(queue: Grc<Self>, item: T) -> impl Process<Item = ()> {
        cons_event({
            let queue = queue.clone();
            move |p| {
                queue.enqueue_initiate(item, p)
            }
        })
        .into_process()
        .and_then(move |i| {
            request_resource(queue.enqueue_resource.clone())
                .and_then(move |()| {
                    cons_event(move |p| {
                        queue.enqueue_store(i, p)
                    })
                    .into_process()
                })
        })
    }

    /// Enqueue the item with input priority by suspending the process
    /// if the queue is full.
    pub fn enqueue_with_input_priority(queue: Grc<Self>, pi: SI::Priority, item: T) -> impl Process<Item = ()>
        where SI::Priority: Clone
    {
        cons_event({
            let queue = queue.clone();
            move |p| {
                queue.enqueue_initiate(item, p)
            }
        })
        .into_process()
        .and_then(move |i| {
            request_resource_with_priority(queue.enqueue_resource.clone(), pi)
                .and_then(move |()| {
                    cons_event(move |p| {
                        queue.enqueue_store(i, p)
                    })
                    .into_process()
                })
        })
    }

    /// Enqueue the item with storing priority by suspending the process
    /// if the queue is full.
    pub fn enqueue_with_storing_priority(queue: Grc<Self>, pm: SM::Priority, item: T) -> impl Process<Item = ()>
        where SM::Priority: Clone
    {
        cons_event({
            let queue = queue.clone();
            move |p| {
                queue.enqueue_initiate(item, p)
            }
        })
        .into_process()
        .and_then(move |i| {
            request_resource(queue.enqueue_resource.clone())
                .and_then(move |()| {
                    cons_event(move |p| {
                        queue.enqueue_store_with_priority(pm, i, p)
                    })
                    .into_process()
                })
        })
    }

    /// Enqueue the item with input and storing priorities by suspending the process
    /// if the queue is full.
    pub fn enqueue_with_input_and_storing_priorities(queue: Grc<Self>, pi: SI::Priority, pm: SM::Priority, item: T) -> impl Process<Item = ()>
        where SI::Priority: Clone,
              SM::Priority: Clone
    {
        cons_event({
            let queue = queue.clone();
            move |p| {
                queue.enqueue_initiate(item, p)
            }
        })
        .into_process()
        .and_then(move |i| {
            request_resource_with_priority(queue.enqueue_resource.clone(), pi)
                .and_then(move |()| {
                    cons_event(move |p| {
                        queue.enqueue_store_with_priority(pm, i, p)
                    })
                    .into_process()
                })
        })
    }

    /// Try to enqueue the item. Return `false` within `Event` computation if the queue is full.
    pub fn try_enqueue(queue: Grc<Self>, item: T) -> impl Event<Item = bool> {
        cons_event(move |p| {
            let x = {
                try_request_resource_within_event(queue.enqueue_resource.clone())
                    .call_event(p)
            }?;
            if x {
                let i = queue.enqueue_initiate(item, p)?;
                queue.enqueue_store(i, p)?;
                Result::Ok(true)
            } else {
                Result::Ok(false)
            }
        })
    }

    /// Try to enqueue the item with storing priority. Return `false`
    /// within `Event` computation if the queue is full.
    pub fn try_enqueue_with_storing_priority(queue: Grc<Self>, pm: SM::Priority, item: T) -> impl Event<Item = bool> {
        cons_event(move |p| {
            let x = {
                try_request_resource_within_event(queue.enqueue_resource.clone())
                    .call_event(p)
            }?;
            if x {
                let i = queue.enqueue_initiate(item, p)?;
                queue.enqueue_store_with_priority(pm, i, p)?;
                Result::Ok(true)
            } else {
                Result::Ok(false)
            }
        })
    }

    /// Try to enqueue the item. If the queue is full then
    /// the item will be counted as lost and `false` will be returned.
    pub fn enqueue_or_lose(queue: Grc<Self>, item: T) -> impl Event<Item = bool> {
        cons_event(move |p| {
            let x = {
                try_request_resource_within_event(queue.enqueue_resource.clone())
                    .call_event(p)
            }?;
            if x {
                let i = queue.enqueue_initiate(item, p)?;
                queue.enqueue_store(i, p)?;
                Result::Ok(true)
            } else {
                queue.enqueue_deny(item, p)?;
                Result::Ok(false)
            }
        })
    }

    /// Try to enqueue the item with storing piority. If the queue is full then
    /// the item will be counted as lost and `false` will be returned.
    pub fn enqueue_with_storing_priority_or_lose(queue: Grc<Self>, pm: SM::Priority, item: T) -> impl Event<Item = bool> {
        cons_event(move |p| {
            let x = {
                try_request_resource_within_event(queue.enqueue_resource.clone())
                    .call_event(p)
            }?;
            if x {
                let i = queue.enqueue_initiate(item, p)?;
                queue.enqueue_store_with_priority(pm, i, p)?;
                Result::Ok(true)
            } else {
                queue.enqueue_deny(item, p)?;
                Result::Ok(false)
            }
        })
    }

    /// Try to enqueue the item. If the queue is full then the item will be counted as lost.
    pub fn enqueue_or_lose_(queue: Grc<Self>, item: T) -> impl Event<Item = ()> {
        Queue::enqueue_or_lose(queue, item)
            .map(|_| {})
    }

    /// Try to enqueue the item with storing priority. If the queue is full then
    /// the item will be counted as lost.
    pub fn enqueue_with_storing_priority_or_lose_(queue: Grc<Self>, pm: SM::Priority, item: T) -> impl Event<Item = ()> {
        Queue::enqueue_with_storing_priority_or_lose(queue, pm, item)
            .map(|_| {})
    }

    /// Notifies when the enqueue operation is initiated.
    #[inline]
    pub fn enqueue_initiated(&self) -> impl Observable<Message = T> + Clone {
        self.enqueue_initiated_source.publish()
    }

    /// Notifies when the item to be enqueued is stored.
    #[inline]
    pub fn enqueue_stored(&self) -> impl Observable<Message = T> + Clone {
        self.enqueue_stored_source.publish()
    }

    /// Notifies when the item that would have to be enqueued is lost.
    #[inline]
    pub fn enqueue_lost(&self) -> impl Observable<Message = T> + Clone {
        self.enqueue_lost_source.publish()
    }

    /// Notifies when the dequeue operation is requested for.
    #[inline]
    pub fn dequeue_requested(&self) -> impl Observable<Message = ()> + Clone {
        self.dequeue_requested_source.publish()
    }

    /// Notifies when the item is dequeued.
    #[inline]
    pub fn dequeue_extracted(&self) -> impl Observable<Message = T> + Clone {
        self.dequeue_extracted_source.publish()
    }

    /// Notifies whenever any property changes.
    #[inline]
    pub fn changed_(&self) -> impl Observable<Message = ()> + Clone {
        self.enqueue_initiated().map(|_| {})
            .merge(self.enqueue_stored().map(|_| {}))
            .merge(self.enqueue_lost().map(|_| {}))
            .merge(self.dequeue_requested())
            .merge(self.dequeue_extracted().map(|_| {}))
    }

    /// Accept the dequeue request and return the current simulation time.
    fn dequeue_request(&self, p: &Point) -> simulation::Result<f64> {
        let c  = self.dequeue_count.read_at(p);
        let c2 = c + 1;
        self.dequeue_count.write_at(c2, p);
        self.dequeue_requested_source.trigger_at(&(), p)?;
        Result::Ok(p.time)
    }

    /// Extract an item by the dequeue request.
    fn dequeue_extract(&self, t_r: f64, p: &Point) -> simulation::Result<T> {
        let i = self.queue_store.pop(p).unwrap();
        self.dequeue_post_extract(t_r, i, p)
    }

    /// A post action after extracting the item by the dequeue request.
    fn dequeue_post_extract(&self, t_r: f64, i: QueueItem<T>, p: &Point) -> simulation::Result<T> {
        let t  = p.time;
        let c  = self.count.read_at(p);
        let c2 = c - 1;
        let stats  = self.count_stats.read_at(p);
        let stats2 = stats.add(t, c2);
        let ec  = self.dequeue_extract_count.read_at(p);
        let ec2 = ec + 1;
        self.count.write_at(c2, p);
        self.count_stats.write_at(stats2, p);
        self.dequeue_extract_count.write_at(ec2, p);
        self.dequeue_stat(t_r, &i, p);
        release_resource_within_event(self.enqueue_resource.clone())
            .call_event(p)?;
        self.dequeue_extracted_source
            .trigger_at(&i.value, p)?;
        Result::Ok(i.value)
    }

    /// Update the statistics for the output wait time of the dequeue operation
    /// and the wait time of storing in the queue.
    fn dequeue_stat(&self, t_r: f64, i: &QueueItem<T>, p: &Point) {
        let t0 = i.input_time;
        let t1 = i.storing_time;
        let t  = p.time;
        let stats  = self.dequeue_wait_time.read_at(p);
        let stats2 = stats.add(t - t_r);
        self.dequeue_wait_time.write_at(stats2, p);
        let stats  = self.total_wait_time.read_at(p);
        let stats2 = stats.add(t - t0);
        self.total_wait_time.write_at(stats2, p);
        let stats  = self.wait_time.read_at(p);
        let stats2 = stats.add(t - t1);
        self.wait_time.write_at(stats2, p);
    }

    /// Initiate the process of enqueueing the item.
    fn enqueue_initiate(&self, item: T, p: &Point) -> simulation::Result<QueueItem<T>> {
        let t = p.time;
        let c = self.enqueue_count.read_at(p);
        self.enqueue_count.write_at(c + 1, p);
        self.enqueue_initiated_source
            .trigger_at(&item, p)?;
        Result::Ok(QueueItem {
            value: item,
            input_time: t,
            storing_time: t
        })
    }

    /// Store the item.
    fn enqueue_store(&self, item: QueueItem<T>, p: &Point) -> simulation::Result<()> {
        let t  = p.time;
        let i2 = QueueItem {
            value: item.value,
            input_time: item.input_time,
            storing_time: t
        };
        self.queue_store.push(i2.clone(), p);
        let c  = self.count.read_at(p);
        let c2 = c + 1;
        self.count.write_at(c2, p);
        let stats  = self.count_stats.read_at(p);
        let stats2 = stats.add(t, c2);
        self.count_stats.write_at(stats2, p);
        let sc  = self.enqueue_store_count.read_at(p);
        let sc2 = sc + 1;
        self.enqueue_store_count.write_at(sc2, p);
        self.enqueue_stat(&i2, p);
        release_resource_within_event(self.dequeue_resource.clone())
            .call_event(p)?;
        self.enqueue_stored_source
            .trigger_at(&i2.value, p)
    }

    /// Store the item with priority.
    fn enqueue_store_with_priority(&self, pm: SM::Priority, item: QueueItem<T>, p: &Point) -> simulation::Result<()> {
        let t  = p.time;
        let i2 = QueueItem {
            value: item.value,
            input_time: item.input_time,
            storing_time: t
        };
        self.queue_store.push_with_priority(pm, i2.clone(), p);
        let c  = self.count.read_at(p);
        let c2 = c + 1;
        self.count.write_at(c2, p);
        let stats  = self.count_stats.read_at(p);
        let stats2 = stats.add(t, c2);
        self.count_stats.write_at(stats2, p);
        let sc  = self.enqueue_store_count.read_at(p);
        let sc2 = sc + 1;
        self.enqueue_store_count.write_at(sc2, p);
        self.enqueue_stat(&i2, p);
        release_resource_within_event(self.dequeue_resource.clone())
            .call_event(p)?;
        self.enqueue_stored_source
            .trigger_at(&i2.value, p)
    }

    /// Deny the enqueue operation.
    fn enqueue_deny(&self, item: T, p: &Point) -> simulation::Result<()> {
        let c  = self.enqueue_lost_count.read_at(p);
        let c2 = c + 1;
        self.enqueue_lost_count.write_at(c2, p);
        self.enqueue_lost_source
            .trigger_at(&item, p)
    }

    /// Update the statistics for the input wait time of the enqueue operation.
    fn enqueue_stat(&self, i: &QueueItem<T>, p: &Point) {
        let t0 = i.input_time;
        let t1 = i.storing_time;
        let stats  = self.enqueue_wait_time.read_at(p);
        let stats2 = stats.add(t1 - t0);
        self.enqueue_wait_time.write_at(stats2, p);
    }

    /// Reset the statistics.
    pub fn reset(queue: Grc<Self>) -> impl Event<Item = ()> + Clone {
        cons_event(move |p| {
            let t = p.time;
            let count = queue.count.read_at(p);
            queue.count_stats.write_at(TimingStats::from_sample(t, count), p);
            queue.enqueue_count.write_at(0, p);
            queue.enqueue_lost_count.write_at(0, p);
            queue.enqueue_store_count.write_at(0, p);
            queue.dequeue_count.write_at(0, p);
            queue.dequeue_extract_count.write_at(0, p);
            queue.wait_time.write_at(SamplingStats::empty(), p);
            queue.total_wait_time.write_at(SamplingStats::empty(), p);
            queue.enqueue_wait_time.write_at(SamplingStats::empty(), p);
            queue.dequeue_wait_time.write_at(SamplingStats::empty(), p);
            Result::Ok(())
        })
    }

    /// Wait while the queue is full.
    pub fn wait_while_full(queue: Grc<Self>) -> impl Process<Item = ()> {
        Queue::is_full(queue.clone())
            .into_process()
            .and_then(move |x| {
                if x {
                    process_await(queue.dequeue_extracted())
                        .and_then(move |_| {
                            Queue::wait_while_full(queue)
                        })
                        .into_boxed()
                } else {
                    return_process(())
                        .into_boxed()
                }
            })
    }
}

/// Computation that creates a new `Queue`.
#[derive(Clone)]
pub struct NewQueue<SI, SM, SO, T> {

    /// The enqueue strategy.
    enqueue_strategy: SI,

    /// The storing strategy.
    storing_strategy: SM,

    /// The output strategy.
    dequeue_strategy: SO,

    /// The capacity.
    max_count: isize,

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

impl<SI, SM, SO, T> Event for NewQueue<SI, SM, SO, T>
    where SI: QueueStrategy + 'static,
          SM: QueueStrategy,
          SO: QueueStrategy + 'static,
          T: 'static
{
    type Item = Queue<SI, SM, SO, T>;

    #[doc(hidden)]
    #[inline]
    fn call_event(self, p: &Point) -> simulation::Result<Self::Item> {
        let NewQueue { enqueue_strategy, storing_strategy, dequeue_strategy, max_count, _phantom } = self;
        if max_count < 0 {
            let msg = String::from("The queue capacity cannot be actually negative");
            let err = Error::retry(msg);
            Result::Err(err)
        } else {
            let t = p.time;
            let enqueue_resource = {
                Resource::<SI>::new_with_max_count(enqueue_strategy, max_count, Some(max_count))
                    .call_simulation(p.run)?
            };
            let queue_store = storing_strategy.new_storage();
            let dequeue_resource = {
                Resource::<SO>::new_with_max_count(dequeue_strategy, 0, Some(max_count))
                    .call_simulation(p.run)?
            };
            Result::Ok(Queue {
                max_count: max_count,
                enqueue_resource: Grc::new(enqueue_resource),
                queue_store: queue_store,
                dequeue_resource: Grc::new(dequeue_resource),
                count: RefComp::new(0),
                count_stats: RefComp::new(TimingStats::from_sample(t, 0)),
                enqueue_count: RefComp::new(0),
                enqueue_lost_count: RefComp::new(0),
                enqueue_store_count: RefComp::new(0),
                dequeue_count: RefComp::new(0),
                dequeue_extract_count: RefComp::new(0),
                wait_time: RefComp::new(SamplingStats::empty()),
                total_wait_time: RefComp::new(SamplingStats::empty()),
                enqueue_wait_time: RefComp::new(SamplingStats::empty()),
                dequeue_wait_time: RefComp::new(SamplingStats::empty()),
                enqueue_initiated_source: ObservableSource::new(),
                enqueue_lost_source: ObservableSource::new(),
                enqueue_stored_source: ObservableSource::new(),
                dequeue_requested_source: ObservableSource::new(),
                dequeue_extracted_source: ObservableSource::new()
            })
        }
    }
}