dvcompute_dist/simulation/observable/

observer.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;

#[cfg(feature="dist_mode")]
use std::sync::Arc;

#[cfg(feature="dist_mode")]
use std::slice;

#[cfg(feature="dist_mode")]
use std::mem;

#[cfg(feature="dist_mode")]
use std::ptr;

#[cfg(feature="dist_mode")]
use libc::*;

use crate::simulation;
use crate::simulation::point::Point;

#[cfg(feature="dist_mode")]
use crate::simulation::error::*;

#[cfg(feature="dist_mode")]
use crate::simulation::utils::byte_vec::*;

use dvcompute_utils::grc::Grc;

/// Return a new `Observer` computation by the specified pure value.
#[inline]
pub fn return_observer<M, T>(val: T) -> Return<M, T>
    where T: Clone
{
    Return { val: val, _phantom: PhantomData }
}

/// Delay the `Observer` computation.
#[inline]
pub fn delay_observer<F, O>(f: F) -> Delay<F, O>
    where F: Fn() -> O,
          O: Observer
{
    Delay { f: f, _phantom: PhantomData }
}

/// Construct a new `Observer` computation by the specified function.
#[inline]
pub fn cons_observer<F, M, T>(f: F) -> Cons<F, M, T>
    where F: Fn(&M, &Point) -> simulation::Result<T>
{
     Cons { f: f, _phantom1: PhantomData, _phantom2: PhantomData }
}

/// Return a new `Observer` computation that returns the outer message which it should react to.
#[inline]
pub fn message_observer<M>() -> Message<M>
    where M: Clone
{
    Message { _phantom: PhantomData }
}

/// Trace the computation.
#[inline]
pub fn trace_observer<O>(msg: String, comp: O) -> Trace<O>
    where O: Observer
{
    Trace { comp: comp, msg: msg}
}

/// The computation that receives notifications.
pub trait Observer {

    /// The outer message that the observer reacts to.
    type Message;

    /// The type of the item that the computation returns.
    type Item;

    /// Call the computation in the current modeling time to react to the specified message.
    #[doc(hidden)]
    fn call_observer(&self, m: &Self::Message, p: &Point) -> simulation::Result<Self::Item>;

    /// Bind the current computation with its continuation within the resulting computation.
    #[inline]
    fn and_then<U, F>(self, f: F) -> AndThen<Self, U, F>
        where Self: Sized,
              U: Observer<Message = Self::Message>,
              F: Fn(Self::Item) -> U,
    {
        AndThen { comp: self, f: f, _phantom: PhantomData }
    }

    /// Map the current computation using the specified transform.
    #[inline]
    fn map<B, F>(self, f: F) -> Map<Self, B, F>
        where Self: Sized,
              F: Fn(Self::Item) -> B,
    {
        Map { comp: self, f: f, _phantom: PhantomData }
    }

    /// Zip the current computation with another one within the resulting computation.
    #[inline]
    fn zip<U>(self, other: U) -> Zip<Self, U>
        where Self: Sized,
              U: Observer<Message = Self::Message>
    {
        Zip { comp: self, other: other }
    }

    /// The function application.
    #[inline]
    fn ap<U, B>(self, other: U) -> Ap<Self, U, B>
        where Self: Sized,
              Self::Item: Fn(U::Item) -> B,
              U: Observer<Message = Self::Message>
    {
        Ap { comp: self, other: other, _phantom: PhantomData }
    }

    /// Convert into a boxed value.
    #[inline]
    fn into_boxed(self) -> ObserverBox<Self::Message, Self::Item>
        where Self: Sized + Clone + 'static
    {
        ObserverBox::new(move |m: &Self::Message, p: &Point| {
            self.call_observer(m, p)
        })
    }
}

/// Allows converting to `Observer` computations.
pub trait IntoObserver {

    /// The target computation.
    type Observer: Observer<Message = Self::Message, Item = Self::Item>;

    /// The type of messages that are processed by the observer.
    type Message;

    /// The type of items that the observer returns.
    type Item;

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

impl<M: Observer> IntoObserver for M {

    type Observer = M;

    type Message = M::Message;

    type Item = M::Item;

    #[inline]
    fn into_observer(self) -> Self::Observer {
        self
    }
}

/// It represents the boxed `Observer` computation.
#[must_use = "computations are lazy and do nothing unless to be run"]
pub struct ObserverBox<M, T> where M: 'static, T: 'static {
    f: Grc<Box<dyn Fn(&M, &Point) -> simulation::Result<T>>>
}

impl<M, T> ObserverBox<M, T> {

    /// Create a new boxed computation.
    #[doc(hidden)]
    #[inline]
    fn new<F>(f: F) -> Self
        where F: Fn(&M, &Point) -> simulation::Result<T> + Clone + 'static
    {
        ObserverBox {
            f: Grc::new(Box::new(f))
        }
    }

    /// Call the boxed function.
    #[doc(hidden)]
    #[inline]
    pub fn call(&self, m: &M, p: &Point) -> simulation::Result<T> {
        let ObserverBox { f } = self;
        f(m, p)
    }
}

impl<M, T> Clone for ObserverBox<M, T> {

    #[inline]
    fn clone(&self) -> Self {
        let f = &self.f;
        ObserverBox {
            f: f.clone()
        }
    }
}

impl<M, T> Observer for ObserverBox<M, T> {

    type Message = M;
    type Item = T;

    #[doc(hidden)]
    #[inline]
    fn call_observer(&self, m: &M, p: &Point) -> simulation::Result<T> {
        self.call(m, p)
    }

    #[inline]
    fn into_boxed(self) -> ObserverBox<Self::Message, Self::Item>
        where Self: Sized + Clone + 'static
    {
        self
    }
}

/// It represents a raw trait object.
#[cfg(feature="dist_mode")]
#[repr(C)]
#[derive(Copy, Clone)]
struct ObserverTraitObject {

    field1: *mut c_void,
    field2: *mut c_void
}

/// A C-friendly representaton of the `Observer` computation.
#[cfg(feature="dist_mode")]
#[repr(C)]
pub struct ObserverRepr {

    /// Delete the object.
    delete: unsafe extern "C" fn(obj: *mut ObserverTraitObject),

    /// Clone the computation.
    clone: unsafe extern "C" fn(obj: *const ObserverTraitObject) -> ObserverRepr,

    /// The callback.
    callback: unsafe extern "C" fn(obj: *const ObserverTraitObject, message: *const u8, count: usize, p: *const Point) -> *mut ErrorRepr,

    /// The trait object.
    trait_object: ObserverTraitObject
}

#[cfg(feature="dist_mode")]
impl Drop for ObserverRepr {

    fn drop(&mut self) {
        unsafe {
            (self.delete)(&mut self.trait_object);
        }
    }
}

#[cfg(feature="dist_mode")]
impl Clone for ObserverRepr {

    fn clone(&self) -> Self {
        unsafe {
            (self.clone)(&self.trait_object)
        }
    }
}

#[cfg(feature="dist_mode")]
impl ObserverRepr {

    /// Convert to a C-friendly representation.
    #[inline]
    pub fn into_repr(comp: ObserverBox<&[u8], ()>) -> ObserverRepr {
        unsafe {
            ObserverRepr {
                delete: delete_observer_repr,
                clone: clone_observer_repr,
                callback: call_observer_repr,
                trait_object: ObserverTraitObject {
                    field1: mem::transmute(comp),
                    field2: ptr::null_mut()
                }
            }
        }
    }

    /// Call the representation.
    #[inline]
    fn call_repr(&self, m: &[u8], p: &Point) -> *mut ErrorRepr {
        unsafe {
            (self.callback)(&self.trait_object, m.as_ptr(), m.len(), p)
        }
    }
}

/// Call the `ObserverBox` representation.
#[cfg(feature="dist_mode")]
unsafe extern "C" fn call_observer_repr(comp: *const ObserverTraitObject, m: *const u8, count: usize, p: *const Point) -> *mut ErrorRepr {
    let comp: ObserverBox<&[u8], ()> = mem::transmute((*comp).field1);
    let m = slice::from_raw_parts(m, count);
    match comp.call_observer(&m, &*p) {
        Result::Ok(()) => {
            mem::forget(comp);
            ptr::null_mut()
        },
        Result::Err(e) => {
            mem::forget(comp);
            let e = ErrorRepr::new(e);
            Box::into_raw(Box::new(e))
        }
    }
}

/// Clone the `ObserverBox` representation.
#[cfg(feature="dist_mode")]
unsafe extern "C" fn clone_observer_repr(comp: *const ObserverTraitObject) -> ObserverRepr {
    let comp: ObserverBox<&[u8], ()> = mem::transmute((*comp).field1);
    let x = ObserverRepr {
        delete: delete_observer_repr,
        clone: clone_observer_repr,
        callback: call_observer_repr,
        trait_object: ObserverTraitObject {
            field1: mem::transmute(comp.clone()),
            field2: ptr::null_mut()
        }
    };
    mem::forget(comp);
    x
}

/// Delete the `ObserverBox` representation.
#[cfg(feature="dist_mode")]
unsafe extern "C" fn delete_observer_repr(comp: *mut ObserverTraitObject) {
    let _: ObserverBox<&[u8], ()> = mem::transmute((*comp).field1);
}

#[cfg(feature="dist_mode")]
impl Observer for ObserverRepr {

    type Message = Arc<ByteVecRepr>;

    type Item = ();

    #[doc(hidden)]
    #[inline]
    fn call_observer(&self, m: &Self::Message, p: &Point) -> simulation::Result<Self::Item> {
        unsafe {
            let m = m.slice();
            let e = self.call_repr(m, p);
            if e == ptr::null_mut() {
                Result::Ok(())
            } else {
                let e = ffi_error_repr_into_error(e);
                Result::Err(e)
            }
        }
    }
}

/// Allows creating the `Observer` computation from a pure value.
#[must_use = "computations are lazy and do nothing unless to be run"]
#[derive(Clone)]
pub struct Return<M, T> {

    /// Return a pure value, which is then transformed to the computation.
    val: T,

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

impl<M, T> Observer for Return<M, T>
    where T: Clone
{
    type Message = M;
    type Item = T;

    #[doc(hidden)]
    #[inline]
    fn call_observer(&self, _: &M, _: &Point) -> simulation::Result<T> {
        Result::Ok(self.val.clone())
    }
}

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

    /// Return the computation.
    f: F,

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

impl<F, O> Observer for Delay<F, O>
    where F: Fn() -> O,
          O: Observer
{
    type Message = O::Message;
    type Item = O::Item;

    #[doc(hidden)]
    #[inline]
    fn call_observer(&self, m: &O::Message, p: &Point) -> simulation::Result<O::Item> {
        let Delay { f, _phantom } = self;
        f().call_observer(m, p)
    }
}

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

    /// The function of time point.
    f: F,

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

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

impl<F, M, T> Observer for Cons<F, M, T>
    where F: Fn(&M, &Point) -> simulation::Result<T>
{
    type Message = M;
    type Item = T;

    #[doc(hidden)]
    #[inline]
    fn call_observer(&self, m: &M, p: &Point) -> simulation::Result<T> {
        let Cons { f, _phantom1, _phantom2 } = self;
        f(m, p)
    }
}

/// Allows creating the `Observer` computation that returns the outer message.
#[must_use = "computations are lazy and do nothing unless to be run"]
#[derive(Clone)]
pub struct Message<M> {

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

impl<M> Observer for Message<M>
    where M: Clone
{
    type Message = M;
    type Item = M;

    #[doc(hidden)]
    #[inline]
    fn call_observer(&self, m: &Self::Message, _: &Point) -> simulation::Result<Self::Item> {
        Result::Ok(m.clone())
    }
}

/// The monadic bind for the `Observer` computation.
#[must_use = "computations are lazy and do nothing unless to be run"]
#[derive(Clone)]
pub struct AndThen<O, U, F> {

    /// The current computation.
    comp: O,

    /// The continuation of the current computation.
    f: F,

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

impl<O, U, F> Observer for AndThen<O, U, F>
    where O: Observer,
          U: Observer<Message = O::Message>,
          F: Fn(O::Item) -> U,
{
    type Message = U::Message;
    type Item = U::Item;

    #[doc(hidden)]
    #[inline]
    fn call_observer(&self, m: &Self::Message, p: &Point) -> simulation::Result<Self::Item> {
        let AndThen { comp, f, _phantom } = self;
        match comp.call_observer(m, p) {
            Result::Ok(a) => {
                let u = f(a);
                u.call_observer(m, p)
            },
            Result::Err(e) => {
                Result::Err(e)
            }
        }
    }
}

/// The functor for the `Observer` computation.
#[must_use = "computations are lazy and do nothing unless to be run"]
#[derive(Clone)]
pub struct Map<O, B, F> {

    /// The current computation.
    comp: O,

    /// The transform.
    f: F,

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

impl<O, B, F> Observer for Map<O, B, F>
    where O: Observer,
          F: Fn(O::Item) -> B,
{
    type Message = O::Message;
    type Item = B;

    #[doc(hidden)]
    #[inline]
    fn call_observer(&self, m: &Self::Message, p: &Point) -> simulation::Result<Self::Item> {
        let Map { comp, f, _phantom } = self;
        match comp.call_observer(m, p) {
            Result::Ok(a) => Result::Ok(f(a)),
            Result::Err(e) => Result::Err(e)
        }
    }
}

/// The zip of two `Observer` computations.
#[must_use = "computations are lazy and do nothing unless to be run"]
#[derive(Clone)]
pub struct Zip<O, U> {

    /// The current computation.
    comp: O,

    /// Another computation.
    other: U,
}

impl<O, U> Observer for Zip<O, U>
    where O: Observer,
          U: Observer<Message = O::Message>
{
    type Message = O::Message;
    type Item = (O::Item, U::Item);

    #[doc(hidden)]
    #[inline]
    fn call_observer(&self, m: &Self::Message, p: &Point) -> simulation::Result<Self::Item> {
        let Zip { comp, other } = self;
        match comp.call_observer(m, p) {
            Result::Ok(a) => {
                match other.call_observer(m, p) {
                    Result::Ok(b) => Result::Ok((a, b)),
                    Result::Err(e) => Result::Err(e)
                }
            },
            Result::Err(e) => Result::Err(e)
        }
    }
}

/// The function application for the `Observer` computation.
#[must_use = "computations are lazy and do nothing unless to be run"]
#[derive(Clone)]
pub struct Ap<O, U, B> {

    /// The current computation.
    comp: O,

    /// The continuation of the current computation.
    other: U,

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

impl<O, U, B> Observer for Ap<O, U, B>
    where O: Observer,
          U: Observer<Message = O::Message>,
          O::Item: Fn(U::Item) -> B,
{
    type Message = O::Message;
    type Item = B;

    #[doc(hidden)]
    #[inline]
    fn call_observer(&self, m: &Self::Message, p: &Point) -> simulation::Result<Self::Item> {
        let Ap { comp, other, _phantom } = self;
        match comp.call_observer(m, p) {
            Result::Ok(f) => {
                match other.call_observer(m, p) {
                    Result::Ok(a) => Result::Ok(f(a)),
                    Result::Err(e) => Result::Err(e)
                }
            },
            Result::Err(e) => Result::Err(e)
        }
    }
}

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

    /// The computation.
    comp: O,

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

impl<O> Observer for Trace<O>
    where O: Observer
{
    type Message = O::Message;
    type Item = O::Item;

    #[doc(hidden)]
    #[inline]
    fn call_observer(&self, m: &Self::Message, p: &Point) -> simulation::Result<Self::Item> {
        let Trace { comp, msg } = self;
        p.trace(&msg);
        comp.call_observer(m, p)
    }
}