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::hash::{Hash, Hasher};

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

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

/// Assemble the transact by the specified number within `Process` computation.
#[inline]
pub fn assemble_transact<T>(transact: Rc<Transact<T>>, count: isize) -> AssembleTransact<T> {
    AssembleTransact { transact: transact, count: count }
}

/// Gather the transacts by the specified number within `Process` computation.
#[inline]
pub fn gather_transacts<T>(transact: Rc<Transact<T>>, count: isize) -> GatherTransacts<T> {
    GatherTransacts { transact: transact, count: count }
}

/// Test within `Event` computation whether another transact is being assembled for the corresponding assembly set.
#[inline]
pub fn is_transact_assembling<T>(transact: Rc<Transact<T>>) -> IsTransactAssembling<T> {
    IsTransactAssembling { transact: transact }
}

/// Test within `Event` computation whether the transacts are being gathered for the corresponding assembly set.
#[inline]
pub fn is_transact_gathering<T>(transact: Rc<Transact<T>>) -> IsTransactGathering<T> {
    IsTransactGathering { transact: transact }
}

/// Defines an assembly set.
pub struct AssemblySet {

    /// The sequence number.
    sequence_no: u64,

    /// The assembling transact.
    assembling_transact: RefComp<Option<Rc<ProcessId>>>,

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

    /// The gathering transacts.
    gathering_transacts: FCFSStorage<Rc<ProcessId>>,

    /// The gathering count.
    gathering_count: RefComp<isize>
}

impl PartialEq for AssemblySet {

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

impl Eq for AssemblySet {}

impl Hash for AssemblySet {

    fn hash<H: Hasher>(&self, state: &mut H) {
        self.sequence_no.hash(state)
    }
}

impl AssemblySet {

    /// Create a new assembly set within `Event` computation.
    #[inline]
    pub fn new() -> NewAssemblySet {
        NewAssemblySet {}
    }
}

/// Computation that creates an assembly set.
#[derive(Clone)]
pub struct NewAssemblySet {}

impl Simulation for NewAssemblySet {

    type Item = AssemblySet;

    #[doc(hidden)]
    fn call_simulation(self, r: &Run) -> simulation::Result<Self::Item> {
        let gen = &r.generator;
        let sequence_no = gen.random_sequence_no();
        Result::Ok(AssemblySet {
            sequence_no: sequence_no,
            assembling_transact: RefComp::new(None),
            assembling_count: RefComp::new(0),
            gathering_transacts: FCFSStorage::new(),
            gathering_count: RefComp::new(0)
        })
    }
}

/// Assemble the transact by the specified number.
#[derive(Clone)]
pub struct AssembleTransact<T> {

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

    /// The count.
    count: isize
}

impl<T> Process for AssembleTransact<T> {

    type Item = ();

    #[doc(hidden)]
    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
    {
        if is_process_cancelled(&pid, p) {
            revoke_process(cont, pid, p)
        } else {
            let AssembleTransact { transact, count } = self;
            match transact.assembly_set(p) {
                Result::Err(e) => Result::Err(e),
                Result::Ok(s) => {
                    let a = s.assembling_count.read_at(p);
                    if a == 0 {
                        let count = count - 1;
                        if count < 0 {
                            let msg = String::from("The number of transacts must be positive");
                            let err = OtherError::retry(msg);
                            cut_error_process(cont, pid, err, p)
                        } else if count == 0 {
                            Result::Ok(())
                        } else {
                            match transact.require_process_id(p) {
                                Result::Err(e) => Result::Err(e),
                                Result::Ok(pid0) => {
                                    s.assembling_transact.write_at(Some(pid0), p);
                                    s.assembling_count.write_at(count, p);
                                    passivate_process().call_process(cont, pid, p)
                                }
                            }
                        }
                    } else {
                        let a = a - 1;
                        if a == 0 {
                            let pid0 = s.assembling_transact.swap_at(None, p).unwrap();
                            s.assembling_count.write_at(a, p);
                            match ProcessId::reactivate_immediately(pid0).call_event(p) {
                                Result::Err(e) => Result::Err(e),
                                Result::Ok(()) => {
                                    cancel_process().call_process(cont, pid, p)
                                }
                            }
                        } else {
                            s.assembling_count.write_at(a, p);
                            cancel_process().call_process(cont, pid, p)
                        }
                    }
                }
            }
        }
    }

    #[doc(hidden)]
    fn call_process_boxed(self, cont: ProcessBoxCont<Self::Item>, pid: Rc<ProcessId>, p: &Point) -> simulation::Result<()> {
        if is_process_cancelled(&pid, p) {
            revoke_process_boxed(cont, pid, p)
        } else {
            let AssembleTransact { transact, count } = self;
            match transact.assembly_set(p) {
                Result::Err(e) => Result::Err(e),
                Result::Ok(s) => {
                    let a = s.assembling_count.read_at(p);
                    if a == 0 {
                        let count = count - 1;
                        if count < 0 {
                            let msg = String::from("The number of transacts must be positive");
                            let err = OtherError::retry(msg);
                            cut_error_process_boxed(cont, pid, err, p)
                        } else if count == 0 {
                            Result::Ok(())
                        } else {
                            match transact.require_process_id(p) {
                                Result::Err(e) => Result::Err(e),
                                Result::Ok(pid0) => {
                                    s.assembling_transact.write_at(Some(pid0), p);
                                    s.assembling_count.write_at(count, p);
                                    passivate_process().call_process_boxed(cont, pid, p)
                                }
                            }
                        }
                    } else {
                        let a = a - 1;
                        if a == 0 {
                            let pid0 = s.assembling_transact.swap_at(None, p).unwrap();
                            s.assembling_count.write_at(a, p);
                            match ProcessId::reactivate_immediately(pid0).call_event(p) {
                                Result::Err(e) => Result::Err(e),
                                Result::Ok(()) => {
                                    cancel_process().call_process_boxed(cont, pid, p)
                                }
                            }
                        } else {
                            s.assembling_count.write_at(a, p);
                            cancel_process().call_process_boxed(cont, pid, p)
                        }
                    }
                }
            }
        }
    }
}

/// Gather the transacts by the specified number.
#[derive(Clone)]
pub struct GatherTransacts<T> {

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

    /// The count.
    count: isize
}

impl<T> Process for GatherTransacts<T> {

    type Item = ();

    #[doc(hidden)]
    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
    {
        if is_process_cancelled(&pid, p) {
            revoke_process(cont, pid, p)
        } else {
            let GatherTransacts { transact, count } = self;
            match transact.assembly_set(p) {
                Result::Err(e) => Result::Err(e),
                Result::Ok(s) => {
                    let a = s.gathering_count.read_at(p);
                    if a == 0 {
                        let count = count - 1;
                        if count < 0 {
                            let msg = String::from("The number of transacts must be positive");
                            let err = OtherError::retry(msg);
                            cut_error_process(cont, pid, err, p)
                        } else if count == 0 {
                            Result::Ok(())
                        } else {
                            match transact.require_process_id(p) {
                                Result::Err(e) => Result::Err(e),
                                Result::Ok(pid0) => {
                                    let priority = Transact::priority_at(&transact, p);
                                    s.gathering_transacts.push_with_priority(priority, pid0, p);
                                    s.gathering_count.write_at(count, p);
                                    passivate_process().call_process(cont, pid, p)
                                }
                            }
                        }
                    } else {
                        match transact.require_process_id(p) {
                            Result::Err(e) => Result::Err(e),
                            Result::Ok(pid0) => {
                                let a = a - 1;
                                let priority = Transact::priority_at(&transact, p);
                                s.gathering_transacts.push_with_priority(priority, pid0, p);
                                s.gathering_count.write_at(a, p);
                                if a == 0 {
                                    let comp = cons_event(move |p| {
                                        let capacity = if count > 0 { count as usize } else { 1 };
                                        let mut pids = Vec::with_capacity(capacity);
                                        while let Some(pid0) = s.gathering_transacts.pop(p) {
                                            pids.push(pid0);
                                        }
                                        let pids = pids.into_iter();
                                        ProcessId::reactivate_many_immediately(pids).call_event(p)
                                    });
                                    passivate_process_before(comp).call_process(cont, pid, p)
                                } else {
                                    passivate_process().call_process(cont, pid, p)
                                }
                            }
                        }
                    }
                }
            }
        }
    }

    #[doc(hidden)]
    fn call_process_boxed(self, cont: ProcessBoxCont<Self::Item>, pid: Rc<ProcessId>, p: &Point) -> simulation::Result<()> {
        if is_process_cancelled(&pid, p) {
            revoke_process_boxed(cont, pid, p)
        } else {
            let GatherTransacts { transact, count } = self;
            match transact.assembly_set(p) {
                Result::Err(e) => Result::Err(e),
                Result::Ok(s) => {
                    let a = s.gathering_count.read_at(p);
                    if a == 0 {
                        let count = count - 1;
                        if count < 0 {
                            let msg = String::from("The number of transacts must be positive");
                            let err = OtherError::retry(msg);
                            cut_error_process_boxed(cont, pid, err, p)
                        } else if count == 0 {
                            Result::Ok(())
                        } else {
                            match transact.require_process_id(p) {
                                Result::Err(e) => Result::Err(e),
                                Result::Ok(pid0) => {
                                    let priority = Transact::priority_at(&transact, p);
                                    s.gathering_transacts.push_with_priority(priority, pid0, p);
                                    s.gathering_count.write_at(count, p);
                                    passivate_process().call_process_boxed(cont, pid, p)
                                }
                            }
                        }
                    } else {
                        match transact.require_process_id(p) {
                            Result::Err(e) => Result::Err(e),
                            Result::Ok(pid0) => {
                                let a = a - 1;
                                let priority = Transact::priority_at(&transact, p);
                                s.gathering_transacts.push_with_priority(priority, pid0, p);
                                s.gathering_count.write_at(a, p);
                                if a == 0 {
                                    let comp = cons_event(move |p| {
                                        let capacity = if count > 0 { count as usize } else { 1 };
                                        let mut pids = Vec::with_capacity(capacity);
                                        while let Some(pid0) = s.gathering_transacts.pop(p) {
                                            pids.push(pid0);
                                        }
                                        let pids = pids.into_iter();
                                        ProcessId::reactivate_many_immediately(pids).call_event(p)
                                    });
                                    passivate_process_before(comp).call_process_boxed(cont, pid, p)
                                } else {
                                    passivate_process().call_process_boxed(cont, pid, p)
                                }
                            }
                        }
                    }
                }
            }
        }
    }
}

/// Test whether another transact is assembled for the corresponding assembly set.
#[derive(Clone)]
pub struct IsTransactAssembling<T> {

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

impl<T> Event for IsTransactAssembling<T> {

    type Item = bool;

    #[doc(hidden)]
    #[inline]
    fn call_event(self, p: &Point) -> simulation::Result<Self::Item> {
        let IsTransactAssembling { transact } = self;
        match transact.assembly_set(p) {
            Result::Err(e) => Result::Err(e),
            Result::Ok(s) => {
                let a = s.assembling_count.read_at(p);
                Result::Ok(a > 0)
            }
        }
    }
}

/// Test whether the transacts are gathered for the corresponding assembly set.
#[derive(Clone)]
pub struct IsTransactGathering<T> {

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

impl<T> Event for IsTransactGathering<T> {

    type Item = bool;

    #[doc(hidden)]
    #[inline]
    fn call_event(self, p: &Point) -> simulation::Result<Self::Item> {
        let IsTransactGathering { transact } = self;
        match transact.assembly_set(p) {
            Result::Err(e) => Result::Err(e),
            Result::Ok(s) => {
                let a = s.gathering_count.read_at(p);
                Result::Ok(a > 0)
            }
        }
    }
}