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 std::ops::Deref;

use dvcompute::simulation;
use dvcompute::simulation::Point;
use dvcompute::simulation::error::*;
use dvcompute::simulation::process::*;

/// The `GeneratorBlock` computation that allows creating generators.
pub mod generator;

/// Defines basic blocks.
pub mod basic;

/// Additional operations.
pub mod ops;

/// Return a new `Block` computation by the specified pure function.
#[inline]
pub fn arr_block<I, O, F>(f: F) -> Arr<I, O, F>
    where F: FnOnce(I) -> O + 'static
{
    Arr { f: f, _phantom1: PhantomData, _phantom2: PhantomData }
}

/// Delay the `Block` computation.
#[inline]
pub fn delay_block<F, A>(f: F) -> Delay<F, A>
    where F: FnOnce() -> A + 'static,
          A: Block + 'static
{
    Delay { f: f, _phantom: PhantomData }
}

/// Construct a `Block` computation.
#[inline]
pub fn cons_block<I, F, M>(f: F) -> Cons<I, F, M>
    where F: FnOnce(I) -> M + 'static,
          M: Process + 'static
{
    Cons { f: f, _phantom1: PhantomData, _phantom2: PhantomData }
}

/// The termination block.
#[inline]
pub fn terminate_block<T>() -> Terminate<T> {
    Terminate { _phantom: PhantomData }
}

/// Transfer the control flow to another terminating block.
#[inline]
pub fn transfer_block<I, O, A>(comp: A) -> Transfer<I, O, A>
    where A: Block<Input = I, Output = ()>
{
    Transfer { comp: comp, _phantom1: PhantomData, _phantom2: PhantomData }
}

/// The simulation block.
pub trait Block {

    /// The input type of the computation.
    type Input;

    /// The output type of the computation.
    type Output;

    /// Call the `Block` computation.
    #[doc(hidden)]
    fn call_block<C>(self, a: Self::Input, cont: C, pid: Rc<ProcessId>, p: &Point) -> simulation::Result<()>
        where C: FnOnce(simulation::Result<Self::Output>, Rc<ProcessId>, &Point) -> simulation::Result<()> + 'static;

    /// Call the `Block` computation using the boxed continuation.
    #[doc(hidden)]
    fn call_block_boxed(self, a: Self::Input, cont: ProcessBoxCont<Self::Output>, pid: Rc<ProcessId>, p: &Point) -> simulation::Result<()>;

    /// Run the computation by the specified input.
    #[inline]
    fn run(self, a: Self::Input) -> Run<Self::Input, Self::Output, Self>
        where Self: Sized
    {
        Run { comp: self, input: a, _phantom: PhantomData }
    }

    /// Bind the current computation with its continuation within the resulting computation.
    #[inline]
    fn and_then<A>(self, other: A) -> AndThen<Self, A>
        where Self: Sized + 'static,
              A: Block<Input = Self::Output> + 'static,
    {
        AndThen { comp: self, other: other }
    }

    /// Convert into a boxed value.
    #[inline]
    fn into_boxed(self) -> BlockBox<Self::Input, Self::Output>
        where Self: Sized + 'static
    {
        BlockBox::new(move |a: Self::Input, cont: ProcessBoxCont<Self::Output>, pid: Rc<ProcessId>, p: &Point| {
            self.call_block_boxed(a, cont, pid, p)
        })
    }

    /// Trace the computation.
    #[inline]
    fn trace(self, msg: String) -> Trace<Self>
        where Self: Sized
    {
        Trace { comp: self, msg: msg }
    }
}

/// Allows converting to `Block` computations.
pub trait IntoBlock {

    /// The target computation.
    type Block: Block<Input = Self::Input, Output = Self::Output>;

    /// The input type of the computation.
    type Input;

    /// The output type of the computation.
    type Output;

    /// Convert to the `Block` computation.
    fn into_block(self) -> Self::Block;
}

impl<A: Block> IntoBlock for A {

    type Block = A;

    type Input = A::Input;

    type Output = A::Output;

    #[inline]
    fn into_block(self) -> Self::Block {
        self
    }
}

/// The `Block` computation box.
#[must_use = "computations are lazy and do nothing unless to be run"]
pub struct BlockBox<I, O> {
    f: Box<dyn BlockFnBox<I, O>>
}

#[doc(hidden)]
impl<I, O> BlockBox<I, O> {

    #[inline]
    pub fn new<F>(f: F) -> Self
        where F: FnOnce(I, ProcessBoxCont<O>, Rc<ProcessId>, &Point) -> simulation::Result<()> + 'static
    {
        BlockBox { f: Box::new(f) }
    }

    #[inline]
    pub fn call_box(self, args: (I, ProcessBoxCont<O>, Rc<ProcessId>, &Point)) -> simulation::Result<()> {
        self.f.call_box(args)
    }
}

impl<I, O> Block for BlockBox<I, O> {

    type Input  = I;
    type Output = O;

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

    #[doc(hidden)]
    #[inline]
    fn call_block_boxed(self, a: Self::Input, cont: ProcessBoxCont<Self::Output>, pid: Rc<ProcessId>, p: &Point) -> simulation::Result<()> {
        self.call_box((a, cont, pid, p))
    }

    #[inline]
    fn into_boxed(self) -> BlockBox<Self::Input, Self::Output>
        where Self: Sized + 'static
    {
        self
    }
}

/// A trait to support the stable version of Rust, where there is no `FnBox`.
trait BlockFnBox<I, O> {

    /// Call the corresponding function.
    fn call_box(self: Box<Self>, args: (I, ProcessBoxCont<O>, Rc<ProcessId>, &Point)) -> simulation::Result<()>;
}

impl<I, O, F> BlockFnBox<I, O> for F
    where F: for<'a> FnOnce(I, ProcessBoxCont<O>, Rc<ProcessId>, &'a Point) -> simulation::Result<()>
{
    fn call_box(self: Box<Self>, args: (I, ProcessBoxCont<O>, Rc<ProcessId>, &Point)) -> simulation::Result<()> {
        let this: Self = *self;
        this(args.0, args.1, args.2, args.3)
    }
}

/// Allows creating the `Block` computation by a pure function.
#[must_use = "computations are lazy and do nothing unless to be run"]
#[derive(Clone)]
pub struct Arr<I, O, F> {

    /// Return a pure function, which is then transformed to the computation.
    f: F,

    /// To keep the type parameter.
    _phantom1: PhantomData<I>,

    /// To keep the type parameter.
    _phantom2: PhantomData<O>
}

impl<I, O, F> Block for Arr<I, O, F>
    where F: FnOnce(I) -> O + 'static
{
    type Input  = I;
    type Output = O;

    #[doc(hidden)]
    #[inline]
    fn call_block<C>(self, a: Self::Input, cont: C, pid: Rc<ProcessId>, p: &Point) -> simulation::Result<()>
        where C: FnOnce(simulation::Result<Self::Output>, Rc<ProcessId>, &Point) -> simulation::Result<()> + 'static
    {
        if is_process_cancelled(&pid, p) {
            revoke_process(cont, pid, p)
        } else {
            let Arr { f, _phantom1, _phantom2 } = self;
            cont(Result::Ok(f(a)), pid, p)
        }
    }

    #[doc(hidden)]
    #[inline]
    fn call_block_boxed(self, a: Self::Input, cont: ProcessBoxCont<Self::Output>, pid: Rc<ProcessId>, p: &Point) -> simulation::Result<()> {
        if is_process_cancelled(&pid, p) {
            revoke_process_boxed(cont, pid, p)
        } else {
            let Arr { f, _phantom1, _phantom2 } = self;
            cont.call_box((Result::Ok(f(a)), pid, p))
        }
    }
}

/// Allows delaying the `Block` computation by the specified function.
#[must_use = "computations are lazy and do nothing unless to be run"]
#[derive(Clone)]
pub struct Delay<F, A> {

    /// Return the computation.
    f: F,

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

impl<F, A> Block for Delay<F, A>
    where F: FnOnce() -> A + 'static,
          A: Block + 'static
{
    type Input  = A::Input;
    type Output = A::Output;

    #[doc(hidden)]
    #[inline]
    fn call_block<C>(self, a: Self::Input, cont: C, pid: Rc<ProcessId>, p: &Point) -> simulation::Result<()>
        where C: FnOnce(simulation::Result<Self::Output>, Rc<ProcessId>, &Point) -> simulation::Result<()> + 'static
    {
        let Delay { f, _phantom } = self;
        let comp = f();
        comp.call_block(a, cont, pid, p)
    }

    #[doc(hidden)]
    #[inline]
    fn call_block_boxed(self, a: Self::Input, cont: ProcessBoxCont<Self::Output>, pid: Rc<ProcessId>, p: &Point) -> simulation::Result<()> {
        let Delay { f, _phantom } = self;
        let comp = f();
        comp.call_block_boxed(a, cont, pid, p)
    }
}

/// A composition of the `Block` computations.
#[must_use = "computations are lazy and do nothing unless to be run"]
#[derive(Clone)]
pub struct AndThen<A, U>
    where A: Block,
          U: Block<Input = A::Output>
{
    /// The current computation.
    comp: A,

    /// The next computation.
    other: U
}

impl<A, U> Block for AndThen<A, U>
    where A: Block + 'static,
          U: Block<Input = A::Output> + 'static
{
    type Input  = A::Input;
    type Output = U::Output;

    #[doc(hidden)]
    #[inline]
    fn call_block<C>(self, a: Self::Input, cont: C, pid: Rc<ProcessId>, p: &Point) -> simulation::Result<()>
        where C: FnOnce(simulation::Result<Self::Output>, Rc<ProcessId>, &Point) -> simulation::Result<()> + 'static
    {
        if is_process_cancelled(&pid, p) {
            revoke_process(cont, pid, p)
        } else {
            let AndThen { comp, other } = self;
            let cont = move |b: simulation::Result<A::Output>, pid: Rc<ProcessId>, p: &Point| {
                match b {
                    Result::Ok(b) => {
                        other.call_block(b, cont, pid, p)
                    },
                    Result::Err(Error::Cancel) => {
                        revoke_process(cont, pid, p)
                    },
                    Result::Err(Error::Other(e)) => {
                        match e.deref() {
                            &OtherError::Retry(_) => {
                                cut_error_process(cont, pid, e, p)
                            },
                            &OtherError::Panic(_) => {
                                cut_error_process(cont, pid, e, p)
                            },
                            &OtherError::IO(_) => {
                                propagate_error_process(cont, pid, e, p)
                            }
                        }
                    }
                }
            };
            comp.call_block(a, cont, pid, p)
        }
    }

    #[doc(hidden)]
    #[inline]
    fn call_block_boxed(self, a: Self::Input, cont: ProcessBoxCont<Self::Output>, pid: Rc<ProcessId>, p: &Point) -> simulation::Result<()> {
        if is_process_cancelled(&pid, p) {
            revoke_process_boxed(cont, pid, p)
        } else {
            let AndThen { comp, other } = self;
            let cont = move |b: simulation::Result<A::Output>, pid: Rc<ProcessId>, p: &Point| {
                match b {
                    Result::Ok(b) => {
                        other.call_block_boxed(b, cont, pid, p)
                    },
                    Result::Err(Error::Cancel) => {
                        revoke_process_boxed(cont, pid, p)
                    },
                    Result::Err(Error::Other(e)) => {
                        match e.deref() {
                            &OtherError::Retry(_) => {
                                cut_error_process_boxed(cont, pid, e, p)
                            },
                            &OtherError::Panic(_) => {
                                cut_error_process_boxed(cont, pid, e, p)
                            },
                            &OtherError::IO(_) => {
                                propagate_error_process_boxed(cont, pid, e, p)
                            }
                        }
                    }
                }
            };
            comp.call_block(a, cont, pid, p)
        }
    }
}

/// Allows constructing a `Block` computation by the specified function.
#[must_use = "computations are lazy and do nothing unless to be run"]
#[derive(Clone)]
pub struct Cons<I, F, M> {

    /// Return the `Process` computation.
    f: F,

    /// To keep the type parameter.
    _phantom1: PhantomData<I>,

    /// To keep the type parameter.
    _phantom2: PhantomData<M>
}

impl<I, F, M> Block for Cons<I, F, M>
    where F: FnOnce(I) -> M + 'static,
          M: Process + 'static
{
    type Input  = I;
    type Output = M::Item;

    #[doc(hidden)]
    #[inline]
    fn call_block<C>(self, a: Self::Input, cont: C, pid: Rc<ProcessId>, p: &Point) -> simulation::Result<()>
        where C: FnOnce(simulation::Result<Self::Output>, Rc<ProcessId>, &Point) -> simulation::Result<()> + 'static
    {
        let Cons { f, _phantom1, _phantom2 } = self;
        let comp = f(a);
        comp.call_process(cont, pid, p)
    }

    #[doc(hidden)]
    #[inline]
    fn call_block_boxed(self, a: Self::Input, cont: ProcessBoxCont<Self::Output>, pid: Rc<ProcessId>, p: &Point) -> simulation::Result<()> {
        let Cons { f, _phantom1, _phantom2 } = self;
        let comp = f(a);
        comp.call_process_boxed(cont, pid, p)
    }
}

/// Defines a termination block.
#[must_use = "computations are lazy and do nothing unless to be run"]
#[derive(Clone)]
pub struct Terminate<T> {

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

impl<T> Block for Terminate<T> {

    type Input  = T;
    type Output = ();

    #[doc(hidden)]
    #[inline]
    fn call_block<C>(self, _a: Self::Input, cont: C, pid: Rc<ProcessId>, p: &Point) -> simulation::Result<()>
        where C: FnOnce(simulation::Result<Self::Output>, Rc<ProcessId>, &Point) -> simulation::Result<()> + 'static
    {
        if is_process_cancelled(&pid, p) {
            revoke_process(cont, pid, p)
        } else {
            let Terminate { _phantom } = self;
            cont(Result::Ok(()), pid, p)
        }
    }

    #[doc(hidden)]
    #[inline]
    fn call_block_boxed(self, _a: Self::Input, cont: ProcessBoxCont<Self::Output>, pid: Rc<ProcessId>, p: &Point) -> simulation::Result<()> {
        if is_process_cancelled(&pid, p) {
            revoke_process_boxed(cont, pid, p)
        } else {
            let Terminate { _phantom } = self;
            cont.call_box((Result::Ok(()), pid, p))
        }
    }
}

/// Defines a block that transfers the control flow to another terminating block.
#[must_use = "computations are lazy and do nothing unless to be run"]
#[derive(Clone)]
pub struct Transfer<I, O, A> {

    /// Another terminating block computation.
    comp: A,

    /// To keep the type parameter.
    _phantom1: PhantomData<I>,

    /// To keep the type parameter.
    _phantom2: PhantomData<O>
}

impl<I, O, A> Block for Transfer<I, O, A>
    where A: Block<Input = I, Output = ()>
{
    type Input  = I;
    type Output = O;

    #[doc(hidden)]
    #[inline]
    fn call_block<C>(self, a: Self::Input, cont: C, pid: Rc<ProcessId>, p: &Point) -> simulation::Result<()>
        where C: FnOnce(simulation::Result<Self::Output>, Rc<ProcessId>, &Point) -> simulation::Result<()> + 'static
    {
        if is_process_cancelled(&pid, p) {
            revoke_process(cont, pid, p)
        } else {
            let Transfer { comp, _phantom1, _phantom2 } = self;
            fn cont(b: simulation::Result<()>, _pid: Rc<ProcessId>, _p: &Point) -> simulation::Result<()> {
                b
            }
            comp.call_block(a, cont, pid, p)
        }
    }

    #[doc(hidden)]
    #[inline]
    fn call_block_boxed(self, a: Self::Input, cont: ProcessBoxCont<Self::Output>, pid: Rc<ProcessId>, p: &Point) -> simulation::Result<()> {
        if is_process_cancelled(&pid, p) {
            revoke_process_boxed(cont, pid, p)
        } else {
            let Transfer { comp, _phantom1, _phantom2 } = self;
            fn cont(b: simulation::Result<()>, _pid: Rc<ProcessId>, _p: &Point) -> simulation::Result<()> {
                b
            }
            comp.call_block(a, cont, pid, p)
        }
    }
}

/// Run the `Block` computation by the specified input.
#[must_use = "computations are lazy and do nothing unless to be run"]
#[derive(Clone)]
pub struct Run<I, O, A> {

    /// Return a pure function, which is then transformed to the computation.
    comp: A,

    /// The input argument.
    input: I,

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

impl<I, O, A> Process for Run<I, O, A>
    where A: Block<Input = I, Output = O> + 'static
{
    type Item = O;

    #[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
    {
        if is_process_cancelled(&pid, p) {
            revoke_process(cont, pid, p)
        } else {
            let Run { comp, input, _phantom } = self;
            comp.call_block(input, cont, pid, p)
        }
    }

    #[doc(hidden)]
    #[inline]
    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 Run { comp, input, _phantom } = self;
            comp.call_block_boxed(input, cont, pid, p)
        }
    }
}

impl<I, T, A> Block for I
    where I: Iterator<Item = A> + Sized + 'static,
          A: Block<Input = T, Output = T> + 'static
{
    type Input  = T;
    type Output = ();

    #[doc(hidden)]
    #[inline(never)]
    fn call_block<C>(self, a: Self::Input, cont: C, pid: Rc<ProcessId>, p: &Point) -> simulation::Result<()>
        where C: FnOnce(simulation::Result<Self::Output>, Rc<ProcessId>, &Point) -> simulation::Result<()> + 'static
    {
        let mut x = self;
        match x.next() {
            None => Result::Ok(()),
            Some(comp) => {
                comp.and_then(x)
                    .call_block(a, cont, pid, p)
            }
        }
    }

    #[doc(hidden)]
    #[inline(never)]
    fn call_block_boxed(self, a: Self::Input, cont: ProcessBoxCont<Self::Output>, pid: Rc<ProcessId>, p: &Point) -> simulation::Result<()> {
        let mut x = self;
        match x.next() {
            None => Result::Ok(()),
            Some(comp) => {
                comp.and_then(x)
                    .call_block_boxed(a, cont, pid, p)
            }
        }
    }
}

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

    /// The computation.
    comp: A,

    /// The message to print.
    msg: String
}

impl<A> Block for Trace<A>
    where A: Block
{
    type Input = A::Input;
    type Output = A::Output;

    #[doc(hidden)]
    #[inline]
    fn call_block<C>(self, a: Self::Input, cont: C, pid: Rc<ProcessId>, p: &Point) -> simulation::Result<()>
        where C: FnOnce(simulation::Result<Self::Output>, Rc<ProcessId>, &Point) -> simulation::Result<()> + 'static
    {
        if is_process_cancelled(&pid, p) {
            revoke_process(cont, pid, p)
        } else {
            let Trace { comp, msg } = self;
            p.trace(&msg);
            comp.call_block(a, cont, pid, p)
        }
    }

    #[doc(hidden)]
    #[inline]
    fn call_block_boxed(self, a: Self::Input, cont: ProcessBoxCont<Self::Output>, pid: Rc<ProcessId>, p: &Point) -> simulation::Result<()> {
        if is_process_cancelled(&pid, p) {
            revoke_process_boxed(cont, pid, p)
        } else {
            let Trace { comp, msg } = self;
            p.trace(&msg);
            comp.call_block_boxed(a, cont, pid, p)
        }
    }
}