dvcompute_dist/simulation/resource/

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 crate::simulation;
use crate::simulation::error::*;
use crate::simulation::Run;
use crate::simulation::Point;
use crate::simulation::ref_comp::RefComp;
use crate::simulation::simulation::*;
use crate::simulation::event::*;
use crate::simulation::process::*;
use crate::simulation::strategy::*;

use dvcompute_utils::grc::Grc;

/// The resources that gather its statistics.
pub mod stats;

/// Request for the resource within `Process` computation.
#[inline]
pub fn request_resource<S>(resource: Grc<Resource<S>>) -> Request<S>
    where S: QueueStrategy + Clone + 'static
{
    Request { resource: resource }
}

/// Request for the resource within `Process` computation with the specified priority.
#[inline]
pub fn request_resource_with_priority<S>(resource: Grc<Resource<S>>, priority: S::Priority) -> RequestWithPriority<S>
    where S: QueueStrategy + Clone + 'static,
          S::Priority: Clone
{
    RequestWithPriority { resource: resource, priority: priority }
}

/// Release the resource within `Process` computation.
#[inline]
pub fn release_resource<S>(resource: Grc<Resource<S>>) -> Release<S>
    where S: QueueStrategy
{
    Release { resource: resource }
}

/// Release the resource within `Event` computation.
#[inline]
pub fn release_resource_within_event<S>(resource: Grc<Resource<S>>) -> ReleaseWithinEvent<S>
    where S: QueueStrategy
{
    ReleaseWithinEvent { resource: resource }
}

/// Try to request for the resource immediately and return a flag
/// indicating whether the resource was aquired.
#[inline]
pub fn try_request_resource_within_event<S>(resource: Grc<Resource<S>>) -> TryRequestWithinEvent<S>
    where S: QueueStrategy
{
    TryRequestWithinEvent { resource: resource }
}

/// Create a new FCFS (a.k.a FIFO) resource by the specified initial count that becomes the capacity as well.
#[inline]
pub fn new_fcfs_resource(count: isize) -> NewResource<FCFSStrategy> {
    NewResource { strategy: FCFSStrategy::Instance, count: count, max_count: Some(count) }
}

/// Create a new FCFS (a.k.a FIFO) resource by the specified initial count and optional maximum count, i.e. capacity.
#[inline]
pub fn new_fcfs_resource_with_max_count(count: isize, max_count: Option<isize>) -> NewResource<FCFSStrategy> {
    NewResource { strategy: FCFSStrategy::Instance, count: count, max_count: max_count }
}

/// Create a new LCFS (a.k.a LIFO) resource by the specified initial count that becomes the capacity as well.
#[inline]
pub fn new_lcfs_resource(count: isize) -> NewResource<LCFSStrategy> {
    NewResource { strategy: LCFSStrategy::Instance, count: count, max_count: Some(count) }
}

/// Create a new LCFS (a.k.a LIFO) resource by the specified initial count and optional maximum count, i.e. capacity.
#[inline]
pub fn new_lcfs_resource_with_max_count(count: isize, max_count: Option<isize>) -> NewResource<LCFSStrategy> {
    NewResource { strategy: LCFSStrategy::Instance, count: count, max_count: max_count }
}

/// The `Resource` based on using the FCFS (a.k.a. FIFO) strategy (First Come - First Served).
pub type FCFSResource = Resource<FCFSStrategy>;

/// The `Resource` based on using the LCFS (a.k.a. LIFO) strategy (Last Come - First Served).
pub type LCFSResource = Resource<LCFSStrategy>;

/// Represents a simple resource that gathers its statistics.
pub struct Resource<S> where S: QueueStrategy {

    /// The maximum count of the resource, where `None` means that there is no upper bound.
    max_count: Option<isize>,

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

    /// The Wait list.
    wait_list: QueueStorageBox<ResourceItem, S::Priority>
}

impl<S> PartialEq for Resource<S> where S: QueueStrategy {

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

impl<S> Eq for Resource<S> where S: QueueStrategy {}

/// Identifies the resource item.
#[derive(Clone)]
struct ResourceItem {

    /// The process identifier.
    pid: Grc<ProcessId>,

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

impl<S> Resource<S> where S: QueueStrategy {

    /// Create a new resource by the specified queue storage and initial count,
    /// where the latter becomes the capacity as well.
    #[inline]
    pub fn new(strategy: S, count: isize) -> NewResource<S> {
        NewResource { strategy: strategy, count: count, max_count: Some(count) }
    }

    /// Create a new resource by the specified queue storage, initial count
    /// and optional maximum count, i.e. capacity.
    #[inline]
    pub fn new_with_max_count(strategy: S, count: isize, max_count: Option<isize>) -> NewResource<S> {
        NewResource { strategy: strategy, count: count, max_count: max_count }
    }

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

/// Computation that creates a new `Resource`.
#[derive(Clone)]
pub struct NewResource<S> {

    /// The queue strategy.
    strategy: S,

    /// The initial count.
    count: isize,

    /// The optional maximum count.
    max_count: Option<isize>
}

impl<S> Simulation for NewResource<S> where S: QueueStrategy {

    type Item = Resource<S>;

    #[doc(hidden)]
    #[inline]
    fn call_simulation(self, _r: &Run) -> simulation::Result<Self::Item> {
        let NewResource { strategy, count, max_count } = self;
        if count < 0 {
            let msg = String::from("The resource count cannot be actually negative");
            let err = Error::retry(msg);
            Result::Err(err)
        } else if count > max_count.unwrap_or(count) {
            let msg = String::from("The resource count cannot be greater than its upper bound");
            let err = Error::retry(msg);
            Result::Err(err)
        } else {
            let wait_list = strategy.new_storage();
            Result::Ok(Resource {
                max_count: max_count,
                count: RefComp::new(count),
                wait_list: wait_list
            })
        }
    }
}

/// Request for the resource.
#[must_use = "computations are lazy and do nothing unless to be run"]
#[derive(Clone)]
pub struct Request<S>
    where S: QueueStrategy + Clone + 'static
{
    /// The resource.
    resource: Grc<Resource<S>>
}

impl<S> Process for Request<S>
    where S: QueueStrategy + Clone + 'static
{
    type Item = ();

    #[doc(hidden)]
    #[inline]
    fn call_process<C>(self, cont: C, pid: Grc<ProcessId>, p: &Point) -> simulation::Result<()>
        where C: FnOnce(simulation::Result<Self::Item>, Grc<ProcessId>, &Point) -> simulation::Result<()> + Clone + 'static
    {
        let Request { resource } = self;
        let a = resource.count.read_at(p);
        if a > 0 {
            resource.count.write_at(a - 1, p);
            resume_process(cont, pid, (), p)
        } else {
            let comp = Request { resource: resource.clone() };
            let cont = ProcessBoxCont::new(cont);
            let cont = FrozenProcess::with_reentering(cont, pid.clone(), (), comp, p)?;
            let item = ResourceItem {
                pid: pid,
                cont: cont
            };
            resource.wait_list.push(item, p);
            Result::Ok(())
        }
    }

    #[doc(hidden)]
    fn call_process_boxed(self, cont: ProcessBoxCont<Self::Item>, pid: Grc<ProcessId>, p: &Point) -> simulation::Result<()> {
        let Request { resource } = self;
        let a = resource.count.read_at(p);
        if a > 0 {
            resource.count.write_at(a - 1, p);
            resume_process_boxed(cont, pid, (), p)
        } else {
            let comp = Request { resource: resource.clone() };
            let cont = FrozenProcess::with_reentering(cont, pid.clone(), (), comp, p)?;
            let item = ResourceItem {
                pid: pid,
                cont: cont
            };
            resource.wait_list.push(item, p);
            Result::Ok(())
        }
    }
}

/// Request for the resource with priority.
#[must_use = "computations are lazy and do nothing unless to be run"]
#[derive(Clone)]
pub struct RequestWithPriority<S>
    where S: QueueStrategy + Clone + 'static
{
    /// The resource.
    resource: Grc<Resource<S>>,

    /// The priority of the request.
    priority: S::Priority
}

impl<S> Process for RequestWithPriority<S>
    where S: QueueStrategy + Clone + 'static,
          S::Priority: Clone
{
    type Item = ();

    #[doc(hidden)]
    #[inline]
    fn call_process<C>(self, cont: C, pid: Grc<ProcessId>, p: &Point) -> simulation::Result<()>
        where C: FnOnce(simulation::Result<Self::Item>, Grc<ProcessId>, &Point) -> simulation::Result<()> + Clone + 'static
    {
        let RequestWithPriority { resource, priority } = self;
        let a = resource.count.read_at(p);
        if a > 0 {
            resource.count.write_at(a - 1, p);
            resume_process(cont, pid, (), p)
        } else {
            let comp = RequestWithPriority { resource: resource.clone(), priority: priority.clone() };
            let cont = ProcessBoxCont::new(cont);
            let cont = FrozenProcess::with_reentering(cont, pid.clone(), (), comp, p)?;
            let item = ResourceItem {
                pid: pid,
                cont: cont
            };
            resource.wait_list.push_with_priority(priority, item, p);
            Result::Ok(())
        }
    }

    #[doc(hidden)]
    fn call_process_boxed(self, cont: ProcessBoxCont<Self::Item>, pid: Grc<ProcessId>, p: &Point) -> simulation::Result<()> {
        let RequestWithPriority { resource, priority } = self;
        let a = resource.count.read_at(p);
        if a > 0 {
            resource.count.write_at(a - 1, p);
            resume_process_boxed(cont, pid, (), p)
        } else {
            let comp = RequestWithPriority { resource: resource.clone(), priority: priority.clone() };
            let cont = FrozenProcess::with_reentering(cont, pid.clone(), (), comp, p)?;
            let item = ResourceItem {
                pid: pid,
                cont: cont
            };
            resource.wait_list.push_with_priority(priority, item, p);
            Result::Ok(())
        }
    }
}

/// Release the resource.
#[must_use = "computations are lazy and do nothing unless to be run"]
#[derive(Clone)]
pub struct Release<S> where S: QueueStrategy + 'static {

    /// The resource.
    resource: Grc<Resource<S>>
}

impl<S> Process for Release<S>
    where S: QueueStrategy
{
    type Item = ();

    #[doc(hidden)]
    #[inline]
    fn call_process<C>(self, cont: C, pid: Grc<ProcessId>, p: &Point) -> simulation::Result<()>
        where C: FnOnce(simulation::Result<Self::Item>, Grc<ProcessId>, &Point) -> simulation::Result<()> + Clone + 'static
    {
        let Release { resource } = self;
        let comp = ReleaseWithinEvent { resource: resource };
        comp.call_event(p)?;
        resume_process(cont, pid, (), p)
    }

    #[doc(hidden)]
    fn call_process_boxed(self, cont: ProcessBoxCont<Self::Item>, pid: Grc<ProcessId>, p: &Point) -> simulation::Result<()> {
        let Release { resource } = self;
        let comp = ReleaseWithinEvent { resource: resource };
        comp.call_event(p)?;
        resume_process_boxed(cont, pid, (), p)
    }
}

/// Release the resource.
#[must_use = "computations are lazy and do nothing unless to be run"]
#[derive(Clone)]
pub struct ReleaseWithinEvent<S> where S: QueueStrategy + 'static {

    /// The resource.
    resource: Grc<Resource<S>>
}

impl<S> Event for ReleaseWithinEvent<S>
    where S: QueueStrategy
{
    type Item = ();

    #[doc(hidden)]
    fn call_event(self, p: &Point) -> simulation::Result<()> {
        let ReleaseWithinEvent { resource } = self;
        let a  = resource.count.read_at(p);
        let a2 = a + 1;
        if a2 > resource.max_count.unwrap_or(a2) {
            let msg = String::from("The resource count cannot be greater than its upper bound");
            let err = Error::retry(msg);
            Result::Err(err)
        } else {
            let t = p.time;
            loop {
                if resource.wait_list.is_empty(p) {
                    resource.count.write_at(a2, p);
                    break;
                } else {
                    let ResourceItem { pid, cont: cont0 } = {
                        resource.wait_list.pop(p).unwrap()
                    };
                    match cont0.unfreeze(p)? {
                        None => continue,
                        Some(cont) => {
                            enqueue_event(t, {
                                cons_event(move |p| {
                                    resume_process_boxed(cont, pid, (), p)
                                }).into_boxed()
                            }).call_event(p)?;
                            break;
                        }
                    }
                }
            }
            Result::Ok(())
        }
    }
}

/// Try to request for the resource immediately and return a flag
/// indicating whether the resource was aquired.
#[must_use = "computations are lazy and do nothing unless to be run"]
#[derive(Clone)]
pub struct TryRequestWithinEvent<S>
    where S: QueueStrategy + 'static
{
    /// The resource.
    resource: Grc<Resource<S>>
}

impl<S> Event for TryRequestWithinEvent<S>
    where S: QueueStrategy
{
    type Item = bool;

    #[doc(hidden)]
    #[inline]
    fn call_event(self, p: &Point) -> simulation::Result<Self::Item> {
        let TryRequestWithinEvent { resource } = self;
        let a = resource.count.read_at(p);
        if a > 0 {
            resource.count.write_at(a - 1, p);
            Result::Ok(true)
        } else {
            Result::Ok(false)
        }
    }
}