dvcompute_branch/simulation/process/

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::rc::Rc;
use std::marker::PhantomData;
use std::ops::Deref;

use dvcompute_utils::collections::im::*;

use crate::simulation;
use crate::simulation::Point;
use crate::simulation::error::*;
use crate::simulation::event::*;
use crate::simulation::{Run as SimulationRun};
use crate::simulation::simulation::Simulation;
use crate::simulation::observable::*;
use crate::simulation::observable::observer::*;
use crate::simulation::observable::source::*;
use crate::simulation::observable::disposable::*;
use crate::simulation::ref_comp::RefComp;

use dvcompute_utils::grc::Grc;

/// Random processes.
pub mod random;

/// Additional operations.
pub mod ops;

/// Return a new `Process` computation by the specified pure value.
#[inline]
pub fn return_process<T>(val: T) -> Return<T> {
    Return { val: val }
}

/// Return a `Process` computation that panics with the specified error message.
#[inline]
pub fn panic_process<T>(msg: String) -> Panic<T> {
    Panic { msg: msg, _phantom: PhantomData }
}

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

/// Return the corresponding process identifier.
#[inline]
pub fn process_id() -> Id {
    Id {}
}

/// Cancel the current process.
#[inline]
pub fn cancel_process<T>() -> Cancel<T> {
    Cancel { _phantom: PhantomData }
}

/// Hold the process for the specified time interval.
#[inline]
pub fn hold_process(dt: f64) -> Hold {
     Hold { dt: dt }
}

/// Passivate the process.
#[inline]
pub fn passivate_process() -> Passivate {
    Passivate {}
}

/// Passivate the process before performing some action.
#[inline]
pub fn passivate_process_before<M>(comp: M) -> PassivateBefore<M>
    where M: Event<Item = ()>
{
    PassivateBefore { comp: comp }
}

/// Execute the `Process` computation in loop.
#[inline]
pub fn loop_process<F, M>(f: F) -> Loop<F, M>
    where F: Fn() -> M + Clone + 'static,
          M: Process<Item = ()> + 'static
{
    Loop { f: f, _phantom: PhantomData }
}

/// Spawn the child process. In case of cancelling one of the processes,
/// other process will be cancelled too.
#[inline]
pub fn spawn_process<M>(comp: M) -> Spawn<M>
    where M: Process<Item = ()> + Clone + 'static
{
    spawn_process_with(ProcessCancellation::CancelTogether, comp)
}

/// Spawn the child process specifying how the child and parent processes
/// should be cancelled in case of need.
#[inline]
pub fn spawn_process_using_id<M>(pid: Grc<ProcessId>, comp: M) -> SpawnUsingId<M>
    where M: Process<Item = ()> + Clone + 'static
{
    spawn_process_using_id_with(ProcessCancellation::CancelTogether, pid, comp)
}

/// Spawn the child process specifying how the child and parent processes
/// should be cancelled in case of need.
#[inline]
pub fn spawn_process_with<M>(cancellation: ProcessCancellation, comp: M) -> Spawn<M>
    where M: Process<Item = ()> + Clone + 'static
{
    Spawn { cancellation: cancellation, comp: comp }
}

/// Spawn the child process specifying how the child and parent processes
/// should be cancelled in case of need.
#[inline]
pub fn spawn_process_using_id_with<M>(cancellation: ProcessCancellation, pid: Grc<ProcessId>, comp: M) -> SpawnUsingId<M>
    where M: Process<Item = ()> + Clone + 'static
{
    SpawnUsingId { cancellation: cancellation, comp: comp, comp_id: pid }
}

/// Await a signal that should be emitted by the specified observable.
#[inline]
pub fn process_await<O>(observable: O) -> Await<O, O::Message>
    where O: Observable,
          O::Message: Clone + 'static
{
    Await { observable: observable, _phantom: PhantomData }
}

/// Await a cancellable signal that should be emitted by the specified observable.
#[inline]
pub fn process_await_result<O, M>(observable: O) -> AwaitResult<O, M>
    where O: Observable<Message = simulation::Result<M>>,
          M: Clone + 'static
{
    AwaitResult { observable: observable, _phantom: PhantomData }
}

/// Try to run the child process within the specified timeout.
#[inline]
pub fn timeout_process<M>(timeout: f64, comp: M) -> Timeout<M>
    where M: Process + Clone + 'static,
          M::Item: Clone
{
    Timeout { timeout: timeout, comp: comp }
}

/// Try to run the child process within the specified timeout by using the given identifier.
#[inline]
pub fn timeout_process_using_id<M>(timeout: f64, pid: Grc<ProcessId>, comp: M) -> TimeoutUsingId<M>
    where M: Process + Clone + 'static,
          M::Item: Clone
{
    TimeoutUsingId { timeout: timeout, comp: comp, comp_id: pid }
}

/// Like the GoTo statement it transfers the direction of computation.
#[inline]
pub fn transfer_process<T, M>(comp: M) -> Transfer<T, M>
    where M: Process<Item = ()>
{
    Transfer { comp: comp, _phantom: PhantomData }
}

/// A process that never returns a result.
#[inline]
pub fn never_process<T>() -> Never<T> {
    Never { _phantom: PhantomData }
}

/// Register a handler that will be invoked in case of cancelling the current process.
#[inline]
pub fn when_cancelling_process<M>(comp: M) -> WhenCancelling<M>
    where M: Observer<Message = (), Item = ()> + Clone + 'static
{
    WhenCancelling { comp: comp }
}

/// Embeds the result value within the current process, which may cancel the process, for example.
#[inline]
pub fn embed_result_process<T>(val: simulation::Result<T>) -> EmbedResult<T> {
    EmbedResult { val: val }
}

/// Create a sequence of computations.
#[inline]
pub fn process_sequence<M>(comps: List<M>) -> Sequence<M>
    where M: Process + Clone + 'static,
          M::Item: Clone
{
    Sequence { comps: comps, acc: List::Nil }
}

/// Create a sequence of computations, where the result is ignored.
#[inline]
pub fn process_sequence_<M>(comps: List<M>) -> Sequence_<M>
    where M: Process + Clone + 'static,
          M::Item: Clone
{
    Sequence_ { comps: comps }
}

/// Proceed with the process that would use the specified time point priority.
#[inline]
pub fn process_with_priority(priority: isize) -> ProcessWithPriority {
    ProcessWithPriority { priority: priority }
}

/// Wrap the computation so that it would restore its priority.
#[inline]
pub fn restore_process_priority<M>(priority: isize, comp: M) -> impl Process<Item = M::Item>
    where M: Process + Clone + 'static
{
    process_with_priority(priority).and_then(|()| { comp })
}

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

/// Check whether the `Process` computation should be cancelled.
#[doc(hidden)]
#[inline]
pub fn is_process_cancelled(pid: &ProcessId, p: &Point) -> bool {
    pid.is_cancel_activated(p)
}

/// Revoke the `Process` computation.
#[doc(hidden)]
pub fn revoke_process<T, C>(cont: C, pid: Grc<ProcessId>, p: &Point) -> simulation::Result<()>
    where C: FnOnce(simulation::Result<T>, Grc<ProcessId>, &Point) -> simulation::Result<()> + Clone
{
    pid.deactivate_cancel(p);
    cont(Result::Err(Error::Cancel), pid, p)
}

/// Revoke the `Process` computation using the boxed continuation.
#[doc(hidden)]
pub fn revoke_process_boxed<T>(cont: ProcessBoxCont<T>, pid: Grc<ProcessId>, p: &Point) -> simulation::Result<()> {
    pid.deactivate_cancel(p);
    cont.call_box((Result::Err(Error::Cancel), pid, p))
}

/// Cut the error `Process` computation.
#[doc(hidden)]
pub fn cut_error_process<T, C>(_cont: C, _pid: Grc<ProcessId>, err: Rc<OtherError>, _p: &Point) -> simulation::Result<()>
    where C: FnOnce(simulation::Result<T>, Grc<ProcessId>, &Point) -> simulation::Result<()>
{
    Result::Err(Error::Other(err))
}

/// Cut the error `Process` computation by using the boxed continuation.
#[doc(hidden)]
pub fn cut_error_process_boxed<T>(_cont: ProcessBoxCont<T>, _pid: Grc<ProcessId>, err: Rc<OtherError>, _p: &Point) -> simulation::Result<()> {
    Result::Err(Error::Other(err))
}

/// Propagate the error within `Process` computation.
#[doc(hidden)]
pub fn propagate_error_process<T, C>(cont: C, pid: Grc<ProcessId>, err: Rc<OtherError>, p: &Point) -> simulation::Result<()>
    where C: FnOnce(simulation::Result<T>, Grc<ProcessId>, &Point) -> simulation::Result<()>
{
    cont(Result::Err(Error::Other(err)), pid, p)
}

/// Propagate the error within `Process` computation.
#[doc(hidden)]
pub fn propagate_error_process_boxed<T>(cont: ProcessBoxCont<T>, pid: Grc<ProcessId>, err: Rc<OtherError>, p: &Point) -> simulation::Result<()> {
    cont.call_box((Result::Err(Error::Other(err)), pid, p))
}

/// Resume the `Process` computation.
#[doc(hidden)]
#[inline]
pub fn resume_process<T, C>(cont: C, pid: Grc<ProcessId>, t: T, p: &Point) -> simulation::Result<()>
    where C: FnOnce(simulation::Result<T>, Grc<ProcessId>, &Point) -> simulation::Result<()> + Clone
{
    if is_process_cancelled(&pid, p) {
        revoke_process(cont, pid, p)
    } else {
        cont(Result::Ok(t), pid, p)
    }
}

/// Resume the `Process` computation using the boxed continuation.
#[doc(hidden)]
#[inline]
pub fn resume_process_boxed<T>(cont: ProcessBoxCont<T>, pid: Grc<ProcessId>, t: T, p: &Point) -> simulation::Result<()> {
    if is_process_cancelled(&pid, p) {
        revoke_process_boxed(cont, pid, p)
    } else {
        cont.call_box((Result::Ok(t), pid, p))
    }
}

/// It defines how the parent and child computations should be cancelled.
#[derive(Clone, PartialEq, Eq, PartialOrd, Ord)]
pub enum ProcessCancellation {

    /// Cancel the both computations together.
    CancelTogether,

    /// Cancel the child if its parent is cancelled.
    CancelChildAfterParent,

    /// Cancel the parent if its child is cancelled.
    CancelParentAfterChild,

    /// Cancel the computations in isolation.
    CancelInIsolation
}

/// The computation identifier.
pub struct ProcessId {

    /// Whether the cancellation has been initiated.
    cancel_initiated: RefComp<bool>,

    /// Whether the cancellation has been activated.
    cancel_activated: RefComp<bool>,

    /// The count of preemption requests.
    preemption_count: RefComp<isize>,

    /// The source of events.
    source: ObservableSource<ProcessEvent>,

    /// Whether the process is already started.
    started: RefComp<bool>,

    /// Whether the process was passivated.
    react_flag: RefComp<Option<bool>>,

    /// The reactivation continuation.
    react_cont: RefComp<Option<ProcessBoxCont<()>>>,

    /// The priority that must be applied in case of reactivation.
    react_priority: RefComp<isize>,

    /// Whether the process was interrupted prematurely.
    interrupt_flag: RefComp<Option<bool>>,

    /// The interruption continuation.
    interrupt_cont: RefComp<Option<ProcessBoxCont<()>>>,

    /// The time of interruption.
    interrupt_time: RefComp<f64>,

    /// The version of interruption.
    interrupt_ver: RefComp<i64>,

    /// The priority that must be applied in case of interruption.
    interrupt_priority: RefComp<isize>
}

impl ProcessId {

    /// Create a new computation identifier.
    pub fn new() -> NewProcessId {
        NewProcessId {}
    }

    /// Prepare the process identifier.
    fn prepare(pid: Grc<ProcessId>, p: &Point) -> simulation::Result<()> {
        let y = pid.started.read_at(p);
        if y {
            panic!("Another process with the specified identifier has been started already")
        } else {
            pid.started.write_at(true, p)
        }
        let disposable = {
            let observable = pid.observable();
            let pid = Grc::downgrade(&pid);
            observable
                .subscribe(cons_observer(move |e, p| {
                    match pid.upgrade() {
                        None => {
                            let msg = String::from("It should not happen");
                            let err = Error::retry(msg);
                            Result::Err(err)
                        },
                        Some(pid) => {
                            match e {
                                ProcessEvent::CancelInitiating => {
                                    if pid.is_cancel_activated(p) {
                                        match ProcessId::interrupt(pid.clone()).call_event(p) {
                                            Result::Ok(()) => ProcessId::reactivate(pid).call_event(p),
                                            Result::Err(e) => Result::Err(e)
                                        }
                                    } else {
                                        Result::Ok(())
                                    }
                                },
                                ProcessEvent::PreemptionInitiating => {
                                    ProcessId::on_preempted_at(pid, p)
                                },
                                ProcessEvent::PreemptionEnding => {
                                    Result::Ok(())
                                }
                            }
                        }
                    }
                })).call_event(p)
        };
        match disposable {
            Result::Ok(_) => Result::Ok(()),
            Result::Err(e) => Result::Err(e)
        }
    }

    /// Returns the observable of computation events.
    #[inline]
    pub fn observable(&self) -> Publish<ProcessEvent> {
        self.source.publish()
    }

    /// Signal when the cancellation is initiated.
    #[inline]
    pub fn cancel_initiating(&self) -> impl Observable<Message = ProcessEvent> + Clone {
        self.observable().filter(|m| { *m == ProcessEvent::CancelInitiating })
    }

    /// Whether the cancellation was initiated.
    #[doc(hidden)]
    #[inline]
    pub fn is_cancel_initiated_at(&self, p: &Point) -> bool {
        self.cancel_initiated.read_at(p)
    }

    /// Whether the cancellation was initiated.
    #[inline]
    pub fn is_cancel_initiated(pid: Grc<ProcessId>) -> impl Event<Item = bool> + Clone {
        cons_event(move |p| Result::Ok(pid.is_cancel_initiated_at(p)))
    }

    /// Whether the cancellation was activated.
    #[inline]
    fn is_cancel_activated(&self, p: &Point) -> bool {
        self.cancel_activated.read_at(p)
    }

    /// Deactivate the cancellation.
    #[inline]
    fn deactivate_cancel(&self, p: &Point) {
        self.cancel_activated.write_at(false, p);
    }

    /// Cancel a process with the specified identifier.
    #[inline]
    pub fn cancel(pid: Grc<ProcessId>) -> CancelById {
        CancelById { pid: pid }
    }

    /// If the main computation is cancelled then all nested ones will be cancelled too.
    fn _bind_cancel(this: Grc<ProcessId>, others: Vec<Grc<ProcessId>>, p: &Point) -> simulation::Result<DisposableBox> {
        let this2 = this.clone();
        let this1 = this;
        let others2 = {
            others.iter().map(|pid| pid.clone()).collect::<Vec<_>>()
        };
        let others1 = others;
        let hs1 = {
            others1.iter().map(|pid| {
                let this1 = Grc::downgrade(&this1);
                pid.cancel_initiating()
                    .subscribe(cons_observer(move |_, p| {
                        match this1.upgrade() {
                            None => Result::Ok(()),
                            Some(this1) => this1.initiate_cancel_at(p)
                        }
                    }))
            })
        };
        let hs1 = event_sequence(hs1).call_event(p);
        match hs1 {
            Result::Err(e) => Result::Err(e),
            Result::Ok(hs1) => {
                let hs2 = {
                    others2.iter().map(|pid| {
                        let pid = Grc::downgrade(&pid);
                        this2.cancel_initiating()
                            .subscribe(cons_observer(move |_, p| {
                                match pid.upgrade() {
                                    None => Result::Ok(()),
                                    Some(pid) => pid.initiate_cancel_at(p)
                                }
                            }))
                    })
                };
                let hs2 = event_sequence(hs2).call_event(p);
                match hs2 {
                    Result::Err(e) => Result::Err(e),
                    Result::Ok(hs2) => {
                        let hs1 = concat_disposables(hs1.into_iter());
                        let hs2 = concat_disposables(hs2.into_iter());

                        Result::Ok(hs1.merge(hs2).into_boxed())
                    }
                }
            }
        }
    }

    /// Connect the parent computation to the child one.
    fn connect_cancel(parent: Grc<ProcessId>, cancellation: ProcessCancellation, child: Grc<ProcessId>, p: &Point) -> simulation::Result<DisposableBox> {
        let h1 = match cancellation {
            ProcessCancellation::CancelTogether |
            ProcessCancellation::CancelChildAfterParent => {
                let child = Grc::downgrade(&child);
                parent.cancel_initiating()
                    .subscribe(cons_observer(move |_, p| {
                        match child.upgrade() {
                            None => Result::Ok(()),
                            Some(child) => child.initiate_cancel_at(p)
                        }
                    }))
                    .call_event(p)
            },
            ProcessCancellation::CancelParentAfterChild |
            ProcessCancellation::CancelInIsolation => {
                Result::Ok(empty_disposable().into_boxed())
            }
        };
        match h1 {
            Result::Err(e) => Result::Err(e),
            Result::Ok(h1) => {
                let h2 = match cancellation {
                    ProcessCancellation::CancelTogether |
                    ProcessCancellation::CancelParentAfterChild => {
                        let parent = Grc::downgrade(&parent);
                        child.cancel_initiating()
                            .subscribe(cons_observer(move |_, p| {
                                match parent.upgrade() {
                                    None => Result::Ok(()),
                                    Some(parent) => parent.initiate_cancel_at(p)
                                }
                            }))
                            .call_event(p)
                    },
                    ProcessCancellation::CancelChildAfterParent |
                    ProcessCancellation::CancelInIsolation => {
                        Result::Ok(empty_disposable().into_boxed())
                    }
                };
                match h2 {
                    Result::Err(e) => {
                        h1.call_box((p,))?;
                        Result::Err(e)
                    },
                    Result::Ok(h2) => Result::Ok(h1.merge(h2).into_boxed())
                }
            }
        }
    }

    /// Initiate the cancellation.
    #[doc(hidden)]
    pub fn initiate_cancel_at(&self, p: &Point) -> simulation::Result<()> {
        let f = self.cancel_initiated.read_at(p);
        if !f {
            self.cancel_initiated.write_at(true, p);
            self.cancel_activated.write_at(true, p);
            self.source.trigger_at(&ProcessEvent::CancelInitiating, p)
        } else {
            Result::Ok(())
        }
    }

    /// Preempt the computation.
    #[doc(hidden)]
    pub fn begin_preemption_at(&self, p: &Point) -> simulation::Result<()> {
        let f = self.cancel_initiated.read_at(p);
        if !f {
            let n = self.preemption_count.read_at(p);
            self.preemption_count.write_at(n + 1, p);
            if n == 0 {
                self.source.trigger_at(&ProcessEvent::PreemptionInitiating, p)
            } else {
                Result::Ok(())
            }
        } else {
            Result::Ok(())
        }
    }

    /// Proceed with the computation after it was preempted earlier.
    #[doc(hidden)]
    pub fn end_preemption_at(&self, p: &Point) -> simulation::Result<()> {
        let f = self.cancel_initiated.read_at(p);
        if !f {
            let n = self.preemption_count.read_at(p);
            self.preemption_count.write_at(n - 1, p);
            if n - 1 == 0 {
                self.source.trigger_at(&ProcessEvent::PreemptionEnding, p)
            } else {
                Result::Ok(())
            }
        } else {
            Result::Ok(())
        }
    }

    /// Preempt the computation.
    #[inline]
    pub fn begin_preemption(pid: Grc<ProcessId>) -> impl Event<Item = ()> + Clone {
        cons_event(move |p| pid.begin_preemption_at(p))
    }

    /// Proceed with the computation after it was preempted earlier.
    #[inline]
    pub fn end_preemption(pid: Grc<ProcessId>) -> impl Event<Item = ()> + Clone {
        cons_event(move |p| pid.end_preemption_at(p))
    }

    /// Signal when the computation is preempted.
    #[inline]
    pub fn preemption_initiating(&self) -> impl Observable<Message = ProcessEvent> + Clone {
        self.observable().filter(|m| { *m == ProcessEvent::PreemptionInitiating })
    }

    /// Signal when the computation is proceeded after it was preempted.
    #[inline]
    pub fn preemption_ending(&self) -> impl Observable<Message = ProcessEvent> + Clone {
        self.observable().filter(|m| { *m == ProcessEvent::PreemptionEnding })
    }

    /// Whether the computation was preempted.
    #[inline]
    fn is_preempted(&self, p: &Point) -> bool {
        self.preemption_count.read_at(p) > 0
    }

    /// Interrupt a process with the specified identifier if the process
    /// is held by computation `hold_process`. Returns an `Event` computation.
    #[inline]
    pub fn interrupt(pid: Grc<ProcessId>) -> Interrupt {
        Interrupt { pid: pid }
    }

    /// Test whether the process with the specified identifier was interrupted.
    /// Returns an `Event` computation.
    #[inline]
    pub fn is_interrupted(pid: Grc<ProcessId>) -> IsInterrupted {
        IsInterrupted { pid: pid }
    }

    /// Return the expected interruption time after finishing the `hold_process` computation,
    /// which value may change if the corresponding process is preempted.
    /// Returns an `Event` computation.
    #[inline]
    pub fn interruption_time(pid: Grc<ProcessId>) -> InterruptionTime {
        InterruptionTime { pid: pid }
    }

    /// Test whether the process with the specified identifier was passivated.
    /// Returns an `Event` computation.
    #[inline]
    pub fn is_passivated(pid: Grc<ProcessId>) -> IsPassivated {
        IsPassivated { pid: pid }
    }

    /// Reactivate a process with the specified identifier.
    #[inline]
    pub fn reactivate(pid: Grc<ProcessId>) -> Reactivate {
        Reactivate { pid: pid }
    }

    /// Reactivate immediately a process with the specified identifier.
    #[inline]
    pub fn reactivate_immediately(pid: Grc<ProcessId>) -> ReactivateImmediately {
        ReactivateImmediately { pid: pid }
    }

    /// Reactivate immediately the processes with specfified identifies.
    #[inline]
    pub fn reactivate_many_immediately(pids: List<Grc<ProcessId>>) -> ReactivateManyImmediately {
        ReactivateManyImmediately { pids: pids }
    }

    /// Define a reaction when the process with the specified identifier is preempted.
    fn on_preempted_at(pid: Grc<ProcessId>, p: &Point) -> simulation::Result<()> {
        match pid.interrupt_cont.swap_at(None, p) {
            Some(cont) => {
                let t  = pid.interrupt_time.read_at(p);
                let dt = t - p.time;
                let v  = pid.interrupt_ver.read_at(p);
                let priority = pid.interrupt_priority.read_at(p);
                pid.interrupt_flag.write_at(Some(true), p);
                pid.interrupt_ver.write_at(v.wrapping_add(1), p);
                let cont = substitute_process_boxed(cont, move |cont, pid, (), p| {
                    restore_process_priority(priority, hold_process(dt))
                        .call_process_boxed(cont, pid, p)
                });
                reenter_process(cont, pid, (), p)
            },
            None => {
                match pid.react_cont.swap_at(None, p) {
                    None => Result::Ok(()),
                    Some(cont) => {
                        let cont = substitute_process_boxed(cont, move |cont, pid, (), p| {
                            reenter_process(cont, pid, (), p)
                        });
                        pid.react_cont.write_at(Some(cont), p);
                        Result::Ok(())
                    }
                }
            }
        }
    }
}

impl PartialEq for ProcessId {

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

impl Eq for ProcessId {}

/// A computation that creates a `ProcessId` value.
#[must_use = "computations are lazy and do nothing unless to be run"]
#[derive(Clone)]
pub struct NewProcessId {}

impl Simulation for NewProcessId {

    type Item = ProcessId;

    #[doc(hidden)]
    fn call_simulation(self, r: &SimulationRun) -> simulation::Result<Self::Item> {
        let specs = &r.specs;
        Result::Ok(ProcessId {
            cancel_initiated: RefComp::new(false),
            cancel_activated: RefComp::new(false),
            preemption_count: RefComp::new(0),
            source: ObservableSource::new(),
            started: RefComp::new(false),
            react_flag: RefComp::new(None),
            react_cont: RefComp::new(None),
            react_priority: RefComp::new(0),
            interrupt_flag: RefComp::new(None),
            interrupt_cont: RefComp::new(None),
            interrupt_time: RefComp::new(specs.start_time),
            interrupt_ver: RefComp::new(0),
            interrupt_priority: RefComp::new(0)
        })
    }
}

/// The event that occurs whithin the `Process` computation.
#[derive(Debug, Clone, PartialEq, PartialOrd, Eq, Ord)]
pub enum ProcessEvent {

    /// Cancel the computation.
    CancelInitiating,

    /// Preempt the computation.
    PreemptionInitiating,

    /// Proceed with the computation after it was preempted.
    PreemptionEnding
}

/// The computation based on continuations.
pub trait Process {

    /// The type of the item that is used within the computation.
    type Item;

    /// Call the `Process` computation.
    #[doc(hidden)]
    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;

    /// Call the `Process` computation using the boxed continuation.
    #[doc(hidden)]
    fn call_process_boxed(self, cont: ProcessBoxCont<Self::Item>, pid: Grc<ProcessId>, p: &Point) -> simulation::Result<()>;

    /// 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 + 'static,
              U: Process + 'static,
              F: FnOnce(Self::Item) -> U + Clone + 'static,
    {
        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 + 'static,
              F: FnOnce(Self::Item) -> B + Clone + 'static,
    {
        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 + 'static,
              Self::Item: Clone,
              U: Process + Clone + 'static
    {
        Zip { comp: self, other: other }
    }

    /// The function application.
    #[inline]
    fn ap<U, B>(self, other: U) -> Ap<Self, U, B>
        where Self: Sized + 'static,
              Self::Item: FnOnce(U::Item) -> B + Clone,
              U: Process + Clone + 'static,
              B: 'static
    {
        Ap { comp: self, other: other, _phantom: PhantomData }
    }

    /// Finalize the current computation regardless of canceling it or not.
    #[inline]
    fn finally<U>(self, finalization: U) -> Finally<Self, U>
        where Self: Sized + 'static,
              Self::Item: Clone,
              U: Process<Item = ()> + Clone + 'static
    {
        Finally { comp: self, finalization: finalization }
    }

    /// Run the `Process` computation.
    #[inline]
    fn run(self) -> Run<Self>
        where Self: Process<Item = ()> + Sized
    {
        Run { comp: self, pid: None }
    }

    /// Run the `Process` computation using the specified process identifier.
    #[inline]
    fn run_using_id(self, pid: Grc<ProcessId>) -> Run<Self>
        where Self: Process<Item = ()> + Sized
    {
        Run { comp: self, pid: Some(pid) }
    }

    /// Convert into a boxed value.
    #[inline]
    fn into_boxed(self) -> ProcessBox<Self::Item>
        where Self: Sized + Clone + 'static
    {
        ProcessBox::new(move |cont: ProcessBoxCont<Self::Item>, pid: Grc<ProcessId>, p: &Point| {
            self.call_process_boxed(cont, pid, p)
        })
    }
}

/// Allows converting to `Process` computations.
pub trait IntoProcess {

    /// The target computation.
    type Process: Process<Item = Self::Item>;

    /// The type of item that is returned by the computation.
    type Item;

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

impl<M: Process> IntoProcess for M {

    type Process = M;

    type Item = M::Item;

    #[inline]
    fn into_process(self) -> Self::Process {
        self
    }
}

/// The continuation of the boxed `Process` computation.
#[doc(hidden)]
pub struct ProcessBoxCont<T> {
    f: Box<dyn ProcessContFnBoxClone<T>>
}

impl<T> ProcessBoxCont<T> {

    /// Create a new continuation box.
    #[inline]
    pub fn new<C>(cont: C) -> Self
        where C: FnOnce(simulation::Result<T>, Grc<ProcessId>, &Point) -> simulation::Result<()> + Clone + 'static
    {
        ProcessBoxCont {
            f: Box::new(cont)
        }
    }

    /// Call the boxed function.
    #[inline]
    pub fn call_box(self, arg: (simulation::Result<T>, Grc<ProcessId>, &Point)) -> simulation::Result<()> {
        let ProcessBoxCont { f } = self;
        f.call_box(arg)
    }
}

impl<T> Clone for ProcessBoxCont<T> {

    #[inline]
    fn clone(&self) -> Self {
        ProcessBoxCont {
            f: self.f.call_clone()
        }
    }
}

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

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

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

/// A trait to implement a cloneable `FnBox`.
trait ProcessContFnBoxClone<T>: ProcessContFnBox<T> {

    /// Clone the function.
    fn call_clone(&self) -> Box<dyn ProcessContFnBoxClone<T>>;
}

impl<T, F> ProcessContFnBoxClone<T> for F
    where F: for<'a> FnOnce(simulation::Result<T>, Grc<ProcessId>, &'a Point) -> simulation::Result<()> + Clone + 'static
{
    fn call_clone(&self) -> Box<dyn ProcessContFnBoxClone<T>> {
        Box::new(self.clone())
    }
}

/// The `Process` computation box.
#[must_use = "computations are lazy and do nothing unless to be run"]
pub struct ProcessBox<T> {
    f: Box<dyn ProcessFnBoxClone<T>>
}

#[doc(hidden)]
impl<T> ProcessBox<T> {

    #[inline]
    pub fn new<F>(f: F) -> Self
        where F: FnOnce(ProcessBoxCont<T>, Grc<ProcessId>, &Point) -> simulation::Result<()> + Clone + 'static
    {
        ProcessBox {
            f: Box::new(f)
        }
    }

    #[inline]
    pub fn call_box(self, args: (ProcessBoxCont<T>, Grc<ProcessId>, &Point)) -> simulation::Result<()> {
        let ProcessBox { f } = self;
        f.call_box(args)
    }
}

impl<T> Clone for ProcessBox<T> {

    #[inline]
    fn clone(&self) -> Self {
        ProcessBox {
            f: self.f.call_clone()
        }
    }
}

impl<T> Process for ProcessBox<T> {

    type Item = T;

    #[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 cont = ProcessBoxCont::new(cont);
        self.call_box((cont, pid, p))
    }

    #[doc(hidden)]
    #[inline]
    fn call_process_boxed(self, cont: ProcessBoxCont<Self::Item>, pid: Grc<ProcessId>, p: &Point) -> simulation::Result<()> {
        self.call_box((cont, pid, p))
    }

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

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

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

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

/// A trait to implement a cloneable `FnBox`.
trait ProcessFnBoxClone<T>: ProcessFnBox<T> {

    /// Clone the function.
    fn call_clone(&self) -> Box<dyn ProcessFnBoxClone<T>>;
}

impl<T, F> ProcessFnBoxClone<T> for F
    where F: for<'a> FnOnce(ProcessBoxCont<T>, Grc<ProcessId>, &'a Point) -> simulation::Result<()> + Clone + 'static
{
    fn call_clone(&self) -> Box<dyn ProcessFnBoxClone<T>> {
        Box::new(self.clone())
    }
}

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

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

impl<T> Process for Return<T> {

    type Item = T;

    #[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
    {
        if is_process_cancelled(&pid, p) {
            revoke_process(cont, pid, p)
        } else {
            let Return { val } = self;
            cont(Result::Ok(val), pid, p)
        }
    }

    #[doc(hidden)]
    #[inline]
    fn call_process_boxed(self, cont: ProcessBoxCont<Self::Item>, pid: Grc<ProcessId>, p: &Point) -> simulation::Result<()> {
        if is_process_cancelled(&pid, p) {
            revoke_process_boxed(cont, pid, p)
        } else {
            let Return { val } = self;
            cont.call_box((Result::Ok(val), pid, p))
        }
    }
}

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

    /// Return the computation.
    f: F,

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

impl<F, M> Process for Delay<F, M>
    where F: FnOnce() -> M + Clone + 'static,
          M: Process + 'static
{
    type Item = M::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 Delay { f, _phantom } = self;
        return_process(()).and_then(move |_| {
            f()
        }).call_process(cont, pid, p)
    }

    #[doc(hidden)]
    #[inline]
    fn call_process_boxed(self, cont: ProcessBoxCont<Self::Item>, pid: Grc<ProcessId>, p: &Point) -> simulation::Result<()> {
        let Delay { f, _phantom } = self;
        return_process(()).and_then(move |_| {
            f()
        }).call_process_boxed(cont, pid, p)
    }
}

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

    /// The current computation.
    comp: M,

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

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

impl<M, U, F> Process for AndThen<M, U, F>
    where F: FnOnce(M::Item) -> U + Clone + 'static,
          M: Process + 'static,
          U: Process + 'static
{
    type Item = U::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
    {
        // m >>= k = \c -> m (\a -> k a c)

        if is_process_cancelled(&pid, p) {
            revoke_process(cont, pid, p)
        } else {
            let AndThen { comp, f, _phantom } = self;
            let cont = move |a: simulation::Result<M::Item>, pid: Grc<ProcessId>, p: &Point| {
                match a {
                    Result::Ok(a) => {
                        f(a).call_process(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_process(cont, pid, p)
        }
    }

    #[doc(hidden)]
    #[inline]
    fn call_process_boxed(self, cont: ProcessBoxCont<Self::Item>, pid: Grc<ProcessId>, p: &Point) -> simulation::Result<()> {
        // m >>= k = \c -> m (\a -> k a c)

        if is_process_cancelled(&pid, p) {
            revoke_process_boxed(cont, pid, p)
        } else {
            let AndThen { comp, f, _phantom } = self;
            let cont = move |a: simulation::Result<M::Item>, pid: Grc<ProcessId>, p: &Point| {
                match a {
                    Result::Ok(a) => {
                        f(a).call_process_boxed(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_process(cont, pid, p)
        }
    }
}

/// The finally block for the `Process` computation.
#[must_use = "computations are lazy and do nothing unless to be run"]
#[derive(Clone)]
pub struct Finally<M, U> {

    /// The current computation.
    comp: M,

    /// The finalization computation.
    finalization: U
}

impl<M, U> Process for Finally<M, U>
    where M: Process + 'static,
          M::Item: Clone,
          U: Process<Item = ()> + Clone + 'static
{
    type Item = M::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
    {
        if is_process_cancelled(&pid, p) {
            revoke_process(cont, pid, p)
        } else {
            let Finally { comp, finalization } = self;
            let cont = move |a: simulation::Result<M::Item>, pid: Grc<ProcessId>, p: &Point| {
                let cont = move |a0: simulation::Result<()>, pid: Grc<ProcessId>, p: &Point| {
                    match a0 {
                        Result::Ok(()) => cont(a, pid, p),
                        Result::Err(e0) => {
                            match a {
                                Result::Ok(_)  => cont(Result::Err(e0), pid, p),
                                Result::Err(e) => cont(Result::Err(e.merge(&e0)), pid, p)
                            }
                        }
                    }
                };
                finalization.call_process(cont, pid, p)
            };
            comp.call_process(cont, pid, p)
        }
    }

    #[doc(hidden)]
    #[inline]
    fn call_process_boxed(self, cont: ProcessBoxCont<Self::Item>, pid: Grc<ProcessId>, p: &Point) -> simulation::Result<()> {
        if is_process_cancelled(&pid, p) {
            revoke_process_boxed(cont, pid, p)
        } else {
            let Finally { comp, finalization } = self;
            let cont = move |a: simulation::Result<M::Item>, pid: Grc<ProcessId>, p: &Point| {
                let cont = move |a0: simulation::Result<()>, pid: Grc<ProcessId>, p: &Point| {
                    match a0 {
                        Result::Ok(()) => cont.call_box((a, pid, p)),
                        Result::Err(e0) => {
                            match a {
                                Result::Ok(_)  => cont.call_box((Result::Err(e0), pid, p)),
                                Result::Err(e) => cont.call_box((Result::Err(e.merge(&e0)), pid, p))
                            }
                        }
                    }
                };
                finalization.call_process(cont, pid, p)
            };
            comp.call_process(cont, pid, p)
        }
    }
}

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

    /// The current computation.
    comp: M,

    /// The transform.
    f: F,

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

impl<M, B, F> Process for Map<M, B, F>
    where F: FnOnce(M::Item) -> B + Clone + 'static,
          M: Process + 'static,
          B: 'static
{
    type Item = B;

    #[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 Map { comp, f, _phantom } = self;
        comp.and_then(move |a| {
            return_process(f(a))
        }).call_process(cont, pid, p)
    }

    #[doc(hidden)]
    #[inline]
    fn call_process_boxed(self, cont: ProcessBoxCont<Self::Item>, pid: Grc<ProcessId>, p: &Point) -> simulation::Result<()> {
        let Map { comp, f, _phantom } = self;
        comp.and_then(move |a| {
            return_process(f(a))
        }).call_process_boxed(cont, pid, p)
    }
}

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

    /// The current computation.
    comp: M,

    /// Another computation.
    other: U,
}

impl<M, U> Process for Zip<M, U>
    where M: Process + 'static,
          M::Item: Clone,
          U: Process + Clone + 'static
{
    type Item = (M::Item, U::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 Zip { comp, other } = self;
        comp.and_then(move |a| {
            other.and_then(move |b| {
                return_process((a, b))
            })
        }).call_process(cont, pid, p)
    }

    #[doc(hidden)]
    #[inline]
    fn call_process_boxed(self, cont: ProcessBoxCont<Self::Item>, pid: Grc<ProcessId>, p: &Point) -> simulation::Result<()> {
        let Zip { comp, other } = self;
        comp.and_then(move |a| {
            other.and_then(move |b| {
                return_process((a, b))
            })
        }).call_process_boxed(cont, pid, p)
    }
}

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

    /// The current computation.
    comp: M,

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

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

impl<M, U, B> Process for Ap<M, U, B>
    where M: Process + 'static,
          U: Process + Clone + 'static,
          M::Item: FnOnce(U::Item) -> B + Clone,
          B: 'static
{
    type Item = B;

    #[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 Ap { comp, other, _phantom } = self;
        comp.and_then(move |f| {
            other.and_then(move |a| {
                return_process(f(a))
            })
        }).call_process(cont, pid, p)
    }

    #[doc(hidden)]
    #[inline]
    fn call_process_boxed(self, cont: ProcessBoxCont<Self::Item>, pid: Grc<ProcessId>, p: &Point) -> simulation::Result<()> {
        let Ap { comp, other, _phantom } = self;
        comp.and_then(move |f| {
            other.and_then(move |a| {
                return_process(f(a))
            })
        }).call_process_boxed(cont, pid, p)
    }
}

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

    /// The current computation.
    comp: M,

    /// The process identifier.
    pid: Option<Grc<ProcessId>>
}

impl<M> Event for Run<M>
    where M: Process<Item = ()>
{
    type Item = M::Item;

    #[doc(hidden)]
    #[inline]
    fn call_event(self, p: &Point) -> simulation::Result<M::Item> {
        let Run { comp, pid } = self;
        let pid = {
            match pid {
                Some(pid) => pid,
                None => {
                    match ProcessId::new().into_event().call_event(p) {
                        Result::Ok(pid) => Grc::new(pid),
                        Result::Err(e) => return Result::Err(e)
                    }
                }
            }
        };
        let cont = &initial_cont;
        match ProcessId::prepare(pid.clone(), p) {
            Result::Err(e) => Result::Err(e),
            Result::Ok(()) => comp.call_process(cont, pid, p)
        }
    }
}

/// The initial continuation.
fn initial_cont(_a: simulation::Result<()>, _pid: Grc<ProcessId>, _p: &Point) -> simulation::Result<()> {
    Result::Ok(())
}

/// Return the corresponding process identifier.
#[must_use = "computations are lazy and do nothing unless to be run"]
#[derive(Clone)]
pub struct Id {}

impl Process for Id {

    type Item = Grc<ProcessId>;

    #[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
    {
        cont(Result::Ok(pid.clone()), pid, p)
    }

    #[doc(hidden)]
    #[inline]
    fn call_process_boxed(self, cont: ProcessBoxCont<Self::Item>, pid: Grc<ProcessId>, p: &Point) -> simulation::Result<()> {
        cont.call_box((Result::Ok(pid.clone()), pid, p))
    }
}

/// Cancel a process by the specified process identifier.
#[must_use = "computations are lazy and do nothing unless to be run"]
#[derive(Clone)]
pub struct CancelById {

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

impl Event for CancelById {

    type Item = ();

    #[doc(hidden)]
    #[inline]
    fn call_event(self, p: &Point) -> simulation::Result<()> {
        let CancelById { pid } = self;
        pid.initiate_cancel_at(p)
    }
}

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

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

impl<T> Process for Cancel<T> {

    type Item = T;

    #[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
    {
        match pid.initiate_cancel_at(p) {
            Result::Ok(()) => {
                if is_process_cancelled(&pid, p) {
                    revoke_process(cont, pid, p)
                } else {
                    Result::Ok(())
                }
            },
            Result::Err(e) => Result::Err(e)
        }
    }

    #[doc(hidden)]
    #[inline]
    fn call_process_boxed(self, cont: ProcessBoxCont<Self::Item>, pid: Grc<ProcessId>, p: &Point) -> simulation::Result<()> {
        match pid.initiate_cancel_at(p) {
            Result::Ok(()) => {
                if is_process_cancelled(&pid, p) {
                    revoke_process_boxed(cont, pid, p)
                } else {
                    Result::Ok(())
                }
            },
            Result::Err(e) => Result::Err(e)
        }
    }
}

/// The hold block for the `Process` computation.
#[must_use = "computations are lazy and do nothing unless to be run"]
#[derive(Clone)]
pub struct Hold {

    /// The time interval of holding the process.
    dt: f64
}

impl Process for Hold {

    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 cont = ProcessBoxCont::new(cont);
        self.call_process_boxed(cont, pid, p)
    }

    #[doc(hidden)]
    #[inline]
    fn call_process_boxed(self, cont: ProcessBoxCont<Self::Item>, pid: Grc<ProcessId>, p: &Point) -> simulation::Result<()> {
        let Hold { dt } = self;
        let t = p.time + dt;
        pid.interrupt_cont.write_at(Some(cont), p);
        pid.interrupt_flag.write_at(Some(false), p);
        pid.interrupt_time.write_at(t, p);
        pid.interrupt_priority.write_at(p.priority, p);
        let v = pid.interrupt_ver.read_at(p);
        enqueue_event(t, {
            cons_event(move |p| {
                if p.priority >= p.minimal_priority {
                    let v2 = pid.interrupt_ver.read_at(p);
                    if v == v2 {
                        pid.interrupt_flag.write_at(None, p);
                        let cont = pid.interrupt_cont.swap_at(None, p).unwrap();
                        resume_process_boxed(cont, pid, (), p)
                    } else {
                        Result::Ok(())
                    }

                } else {
                    pid.initiate_cancel_at(p)?;
                    if is_process_cancelled(&pid, p) {
                        let v2 = pid.interrupt_ver.read_at(p);
                        if v == v2 {
                            pid.interrupt_flag.write_at(None, p);
                            let cont = pid.interrupt_cont.swap_at(None, p).unwrap();
                            revoke_process_boxed(cont, pid, p)
                        } else {
                            Result::Ok(())
                        }
                    } else {
                        Result::Ok(())
                    }
                }
            }).into_boxed()
        }).call_event(p)
    }
}

/// Interrupt a process with the specified identifier if the process
/// is held by computation `hold_process`.
#[must_use = "computations are lazy and do nothing unless to be run"]
#[derive(Clone)]
pub struct Interrupt {

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

impl Event for Interrupt {

    type Item = ();

    #[doc(hidden)]
    #[inline]
    fn call_event(self, p: &Point) -> simulation::Result<Self::Item> {
        let Interrupt { pid } = self;
        match pid.interrupt_cont.swap_at(None, p) {
            None => Result::Ok(()),
            Some(cont) => {
                let v = pid.interrupt_ver.read_at(p);
                let priority = pid.interrupt_priority.read_at(p);
                pid.interrupt_ver.write_at(v.wrapping_add(1), p);
                pid.interrupt_flag.write_at(Some(true), p);
                enqueue_event_with_priority(p.time, priority, {
                    cons_event(move |p| {
                        resume_process_boxed(cont, pid, (), p)
                    }).into_boxed()
                }).call_event(p)
            }
        }
    }
}

/// Test whether the process with the specified identifier was interrupted.
#[must_use = "computations are lazy and do nothing unless to be run"]
#[derive(Clone)]
pub struct IsInterrupted {

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

impl Event for IsInterrupted {

    type Item = bool;

    #[doc(hidden)]
    #[inline]
    fn call_event(self, p: &Point) -> simulation::Result<Self::Item> {
        let IsInterrupted { pid } = self;
        match pid.interrupt_flag.read_at(p) {
            Some(true) => Result::Ok(true),
            _ => Result::Ok(false)
        }
    }
}

/// Return the expected interruption time after finishing the `hold_process` computation,
/// which value may change if the corresponding process is preempted.
#[must_use = "computations are lazy and do nothing unless to be run"]
#[derive(Clone)]
pub struct InterruptionTime {

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

impl Event for InterruptionTime {

    type Item = Option<f64>;

    #[doc(hidden)]
    #[inline]
    fn call_event(self, p: &Point) -> simulation::Result<Self::Item> {
        let InterruptionTime { pid } = self;
        match pid.interrupt_flag.read_at(p) {
            Some(false) => Result::Ok(Some(pid.interrupt_time.read_at(p))),
            _ => Result::Ok(None)
        }
    }
}

/// Passivate the `Process` computation.
#[must_use = "computations are lazy and do nothing unless to be run"]
#[derive(Clone)]
pub struct Passivate {}

impl Process for Passivate {

    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 cont = ProcessBoxCont::new(cont);
        self.call_process_boxed(cont, pid, p)
    }

    #[doc(hidden)]
    #[inline]
    fn call_process_boxed(self, cont: ProcessBoxCont<Self::Item>, pid: Grc<ProcessId>, p: &Point) -> simulation::Result<()> {
        match pid.react_flag.read_at(p) {
            Some(true) => {
                let msg = String::from("Cannot passivate the process twice");
                let err = OtherError::retry(msg);
                cut_error_process_boxed(cont, pid, err, p)
            },
            _ => {
                match pid.react_cont.swap_at(Some(cont), p) {
                    Some(_) => panic!("The reactivation continuation has diverged"),
                    None => {
                        pid.react_priority.write_at(p.priority, p);
                        pid.react_flag.write_at(Some(true), p);
                        Result::Ok(())
                    }
                }
            }
        }
    }
}

/// Passivate the `Process` computation before performing some action.
#[must_use = "computations are lazy and do nothing unless to be run"]
#[derive(Clone)]
pub struct PassivateBefore<M> {

    /// The action to perform after passivating the process.
    comp: M
}

impl<M> Process for PassivateBefore<M>
    where M: Event<Item = ()>
{
    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 cont = ProcessBoxCont::new(cont);
        self.call_process_boxed(cont, pid, p)
    }

    #[doc(hidden)]
    #[inline]
    fn call_process_boxed(self, cont: ProcessBoxCont<Self::Item>, pid: Grc<ProcessId>, p: &Point) -> simulation::Result<()> {
        match pid.react_flag.read_at(p) {
            Some(true) => {
                let msg = String::from("Cannot passivate the process twice");
                let err = OtherError::retry(msg);
                cut_error_process_boxed(cont, pid, err, p)
            },
            _ => {
                match pid.react_cont.swap_at(Some(cont), p) {
                    Some(_) => panic!("The reactivation continuation has diverged"),
                    None => {
                        let PassivateBefore { comp } = self;
                        pid.react_priority.write_at(p.priority, p);
                        pid.react_flag.write_at(Some(true), p);
                        comp.call_event(p)
                    }
                }
            }
        }
    }
}

/// Test whether the process with the specified identifier was passivated.
#[must_use = "computations are lazy and do nothing unless to be run"]
#[derive(Clone)]
pub struct IsPassivated {

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

impl Event for IsPassivated {

    type Item = bool;

    #[doc(hidden)]
    #[inline]
    fn call_event(self, p: &Point) -> simulation::Result<Self::Item> {
        let IsPassivated { pid } = self;
        match pid.react_flag.read_at(p) {
            Some(true) => Result::Ok(true),
            _ => Result::Ok(false)
        }
    }
}

/// Reactivate the process with the specified identifier.
#[must_use = "computations are lazy and do nothing unless to be run"]
#[derive(Clone)]
pub struct Reactivate {

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

impl Event for Reactivate {

    type Item = ();

    #[doc(hidden)]
    #[inline]
    fn call_event(self, p: &Point) -> simulation::Result<Self::Item> {
        let Reactivate { pid } = self;
        match pid.react_cont.swap_at(None, p) {
            Some(cont) => {
                match pid.react_flag.swap_at(None, p) {
                    Some(true) => {
                        let priority = pid.react_priority.read_at(p);
                        enqueue_event_with_priority(p.time, priority, {
                            cons_event(move |p| {
                                resume_process_boxed(cont, pid, (), p)
                            }).into_boxed()
                        }).call_event(p)
                    },
                    _ => {
                        panic!("The reactivation continuation has diverged")
                    }
                }
            },
            None => {
                Result::Ok(())
            }
        }
    }
}

/// Reactivate immediately the process with the specified identifier.
#[must_use = "computations are lazy and do nothing unless to be run"]
#[derive(Clone)]
pub struct ReactivateImmediately {

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

impl Event for ReactivateImmediately {

    type Item = ();

    #[doc(hidden)]
    #[inline]
    fn call_event(self, p: &Point) -> simulation::Result<Self::Item> {
        let ReactivateImmediately { pid } = self;
        match pid.react_cont.swap_at(None, p) {
            Some(cont) => {
                match pid.react_flag.swap_at(None, p) {
                    Some(true) => {
                        let priority = pid.react_priority.read_at(p);
                        if p.priority == priority {
                            resume_process_boxed(cont, pid, (), p)
                        } else {
                            enqueue_event_with_priority(p.time, priority, {
                                cons_event(move |p| {
                                    resume_process_boxed(cont, pid, (), p)
                                }).into_boxed()
                            }).call_event(p)
                        }
                    },
                    _ => {
                        panic!("The reactivation continuation has diverged")
                    }
                }
            },
            None => {
                Result::Ok(())
            }
        }
    }
}

/// Reactivate immediately the processes with specified identifiers.
#[must_use = "computations are lazy and do nothing unless to be run"]
#[derive(Clone)]
pub struct ReactivateManyImmediately {

    /// The process identifiers.
    pids: List<Grc<ProcessId>>
}

impl Event for ReactivateManyImmediately {

    type Item = ();

    #[doc(hidden)]
    #[inline]
    fn call_event(self, p: &Point) -> simulation::Result<Self::Item> {
        let ReactivateManyImmediately { pids } = self;
        match pids {
            List::Nil => Result::Ok(()),
            List::Cons(pid, pids) => {
                match ProcessId::reactivate_immediately(pid).call_event(p) {
                    Result::Err(e) => Result::Err(e),
                    Result::Ok(()) => {
                        let pids = pids.as_list();
                        let comp = ReactivateManyImmediately { pids: pids };
                        yield_event(comp).call_event(p)
                    }
                }
            }
        }
    }
}

/// Executes the `Process` computation in loop.
#[must_use = "computations are lazy and do nothing unless to be run"]
#[derive(Clone)]
pub struct Loop<F, M> {

    /// The computation generator.
    f: F,

    /// The type keeper.
    _phantom: PhantomData<M>
}

impl<F, M> Process for Loop<F, M>
    where F: Fn() -> M + Clone + 'static,
          M: Process<Item = ()> + 'static
{
    type Item = M::Item;

    #[doc(hidden)]
    #[inline(never)]
    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 Loop { f, _phantom } = self;
        let comp = f();
        comp.and_then(move |()| {
            Loop { f: f, _phantom: PhantomData }
        }).call_process(cont, pid, p)
    }

    #[doc(hidden)]
    #[inline(never)]
    fn call_process_boxed(self, cont: ProcessBoxCont<Self::Item>, pid: Grc<ProcessId>, p: &Point) -> simulation::Result<()> {
        let Loop { f, _phantom } = self;
        let comp = f();
        comp.and_then(move |()| {
            Loop { f: f, _phantom: PhantomData }
        }).call_process_boxed(cont, pid, p)
    }
}

/// The sequence of computations.
#[must_use = "computations are lazy and do nothing unless to be run"]
#[derive(Clone)]
pub struct Sequence<M> where M: Process {

    /// The computations.
    comps: List<M>,

    /// The accumulator of results.
    acc: List<M::Item>
}

impl<M> Process for Sequence<M>
    where M: Process + Clone + 'static,
          M::Item: Clone
{
    type Item = List<M::Item>;

    #[doc(hidden)]
    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 Sequence { comps, acc } = self;
        match comps {
            List::Nil => cont(Result::Ok(acc.reversed()), pid, p),
            List::Cons(comp, comps) => {
                comp.and_then(move |a| {
                    let comps = comps.as_list();
                    let acc = List::Cons(a, Rc::new(acc));
                    Sequence { comps: comps, acc: acc }
                }).call_process(cont, pid, p)
            }
        }
    }

    #[doc(hidden)]
    fn call_process_boxed(self, cont: ProcessBoxCont<Self::Item>, pid: Grc<ProcessId>, p: &Point) -> simulation::Result<()> {
        let Sequence { comps, acc } = self;
        match comps {
            List::Nil => cont.call_box((Result::Ok(acc.reversed()), pid, p)),
            List::Cons(comp, comps) => {
                comp.and_then(move |a| {
                    let comps = comps.as_list();
                    let acc = List::Cons(a, Rc::new(acc));
                    Sequence { comps: comps, acc: acc }
                }).call_process_boxed(cont, pid, p)
            }
        }
    }
}

/// The sequence of computations with ignored result.
#[must_use = "computations are lazy and do nothing unless to be run"]
#[derive(Clone)]
pub struct Sequence_<M>
{
    /// The computations.
    comps: List<M>
}

impl<M> Process for Sequence_<M>
    where M: Process + Clone + 'static,
          M::Item: Clone
{
    type Item = ();

    #[doc(hidden)]
    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 Sequence_ { comps } = self;
        match comps {
            List::Nil => cont(Result::Ok(()), pid, p),
            List::Cons(comp, comps) => {
                comp.and_then(move |_| {
                    let comps = comps.as_list();
                    Sequence_ { comps: comps }
                }).call_process(cont, pid, p)
            }
        }
    }

    #[doc(hidden)]
    fn call_process_boxed(self, cont: ProcessBoxCont<Self::Item>, pid: Grc<ProcessId>, p: &Point) -> simulation::Result<()> {
        let Sequence_ { comps } = self;
        match comps {
            List::Nil => cont.call_box((Result::Ok(()), pid, p)),
            List::Cons(comp, comps) => {
                comp.and_then(move |_| {
                    let comps = comps.as_list();
                    Sequence_ { comps: comps }
                }).call_process_boxed(cont, pid, p)
            }
        }
    }
}

/// Allows spawning another process bound with the current one.
#[must_use = "computations are lazy and do nothing unless to be run"]
#[derive(Clone)]
pub struct SpawnUsingId<M>
{
    /// How to cancel the processes.
    cancellation: ProcessCancellation,

    /// The computation.
    comp: M,

    /// The child process identifier.
    comp_id: Grc<ProcessId>
}

impl<M> Process for SpawnUsingId<M>
    where M: Process<Item = ()> + Clone + 'static,
{
    type Item = ();

    #[doc(hidden)]
    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
    {
        if is_process_cancelled(&pid, p) {
            revoke_process(cont, pid, p)
        } else {
            let SpawnUsingId { cancellation, comp, comp_id } = self;
            match ProcessId::prepare(comp_id.clone(), p) {
                Result::Err(e) => Result::Err(e),
                Result::Ok(()) => {
                    let hs = ProcessId::connect_cancel(pid.clone(), cancellation, comp_id.clone(), p);
                    match hs {
                        Result::Err(e) => Result::Err(e),
                        Result::Ok(hs) => {
                            enqueue_event(p.time, {
                                cons_event(move |p| {
                                    let cont = move |a: simulation::Result<()>, _pid: Grc<ProcessId>, p: &Point| {
                                        hs.dispose(p)?;
                                        match a {
                                            Result::Ok(()) => Result::Ok(()),
                                            Result::Err(Error::Cancel) => Result::Ok(()),
                                            Result::Err(Error::Other(e)) => Result::Err(Error::Other(e))
                                        }
                                    };
                                    comp.call_process(cont, comp_id, p)
                                }).into_boxed()
                            }).and_then(move |()| {
                                cons_event(move |p| {
                                    resume_process(cont, pid, (), p)
                                })
                            }).call_event(p)
                        }
                    }
                }
            }
        }
    }

    #[doc(hidden)]
    fn call_process_boxed(self, cont: ProcessBoxCont<Self::Item>, pid: Grc<ProcessId>, p: &Point) -> simulation::Result<()> {
        if is_process_cancelled(&pid, p) {
            revoke_process_boxed(cont, pid, p)
        } else {
            let SpawnUsingId { cancellation, comp, comp_id } = self;
            match ProcessId::prepare(comp_id.clone(), p) {
                Result::Err(e) => Result::Err(e),
                Result::Ok(()) => {
                    let hs = ProcessId::connect_cancel(pid.clone(), cancellation, comp_id.clone(), p);
                    match hs {
                        Result::Err(e) => Result::Err(e),
                        Result::Ok(hs) => {
                            enqueue_event(p.time, {
                                cons_event(move |p| {
                                    let cont = move |a: simulation::Result<()>, _pid: Grc<ProcessId>, p: &Point| {
                                        hs.dispose(p)?;
                                        match a {
                                            Result::Ok(()) => Result::Ok(()),
                                            Result::Err(Error::Cancel) => Result::Ok(()),
                                            Result::Err(Error::Other(e)) => Result::Err(Error::Other(e))
                                        }
                                    };
                                    comp.call_process(cont, comp_id, p)
                                }).into_boxed()
                            }).and_then(move |()| {
                                cons_event(move |p| {
                                    resume_process_boxed(cont, pid, (), p)
                                })
                            }).call_event(p)
                        }
                    }
                }
            }
        }
    }
}

/// Allows spawning another process bound with the current one.
#[must_use = "computations are lazy and do nothing unless to be run"]
#[derive(Clone)]
pub struct Spawn<M>
{
    /// How to cancel the processes.
    cancellation: ProcessCancellation,

    /// The computation.
    comp: M
}

impl<M> Process for Spawn<M>
    where M: Process<Item = ()> + Clone + 'static
{
    type Item = ();

    #[doc(hidden)]
    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
    {
        if is_process_cancelled(&pid, p) {
            revoke_process(cont, pid, p)
        } else {
            let Spawn { cancellation, comp } = self;
            ProcessId::new()
                .into_process()
                .and_then(move |pid| {
                    spawn_process_using_id_with(cancellation, Grc::new(pid), comp)
                })
                .call_process(cont, pid, p)
        }
    }

    #[doc(hidden)]
    fn call_process_boxed(self, cont: ProcessBoxCont<Self::Item>, pid: Grc<ProcessId>, p: &Point) -> simulation::Result<()> {
        if is_process_cancelled(&pid, p) {
            revoke_process_boxed(cont, pid, p)
        } else {
            let Spawn { cancellation, comp } = self;
            ProcessId::new()
                .into_process()
                .and_then(move |pid| {
                    spawn_process_using_id_with(cancellation, Grc::new(pid), comp)
                })
                .call_process_boxed(cont, pid, p)
        }
    }
}

/// Allows running another process within the specified timeout by using the given identifier.
#[must_use = "computations are lazy and do nothing unless to be run"]
#[derive(Clone)]
pub struct TimeoutUsingId<M>
{
    /// Timeout.
    timeout: f64,

    /// The computation.
    comp: M,

    /// The child process identifier.
    comp_id: Grc<ProcessId>
}

impl<M> Process for TimeoutUsingId<M>
    where M: Process + Clone + 'static,
          M::Item: Clone
{
    type Item = Option<M::Item>;

    #[doc(hidden)]
    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
    {
        if is_process_cancelled(&pid, p) {
            revoke_process(cont, pid, p)
        } else {
            let TimeoutUsingId { timeout, comp, comp_id } = self;
            let comp_id_clone = comp_id.clone();
            let s = Grc::new(ObservableSource::new());
            let s_clone = s.clone();
            ProcessId::new()
                .into_process()
                .and_then(move |timeout_id| {
                    let timeout_id = Grc::new(timeout_id);
                    spawn_process_using_id_with(ProcessCancellation::CancelChildAfterParent, timeout_id.clone(), {
                        hold_process(timeout)
                            .and_then(move |()| {
                                ProcessId::cancel(comp_id_clone)
                                    .into_process()
                            })
                    })
                    .and_then(move |()| {
                        let r = Grc::new(RefComp::new(None));
                        spawn_process_using_id_with(ProcessCancellation::CancelChildAfterParent, comp_id, {
                            let r_clone = r.clone();
                            comp.and_then(move |item| {
                                    RefComp::write(r_clone, Some(item))
                                        .into_process()
                                })
                                .finally({
                                    ProcessId::cancel(timeout_id)
                                        .and_then(move |()| {
                                            RefComp::read(r)
                                                .and_then(move |item| {
                                                    s_clone.trigger(item)
                                                })
                                        })
                                        .into_process()
                                })
                        })
                    })
                })
                .and_then(move |()| {
                    process_await(s.publish())
                })
                .call_process(cont, pid, p)
        }
    }

    #[doc(hidden)]
    fn call_process_boxed(self, cont: ProcessBoxCont<Self::Item>, pid: Grc<ProcessId>, p: &Point) -> simulation::Result<()> {
        if is_process_cancelled(&pid, p) {
            revoke_process_boxed(cont, pid, p)
        } else {
            let TimeoutUsingId { timeout, comp, comp_id } = self;
            let comp_id_clone = comp_id.clone();
            let s = Grc::new(ObservableSource::new());
            let s_clone = s.clone();
            ProcessId::new()
                .into_process()
                .and_then(move |timeout_id| {
                    let timeout_id = Grc::new(timeout_id);
                    spawn_process_using_id_with(ProcessCancellation::CancelChildAfterParent, timeout_id.clone(), {
                        hold_process(timeout)
                            .and_then(move |()| {
                                ProcessId::cancel(comp_id_clone)
                                    .into_process()
                            })
                    })
                    .and_then(move |()| {
                        let r = Grc::new(RefComp::new(None));
                        spawn_process_using_id_with(ProcessCancellation::CancelChildAfterParent, comp_id, {
                            let r_clone = r.clone();
                            comp.and_then(move |item| {
                                    RefComp::write(r_clone, Some(item))
                                        .into_process()
                                })
                                .finally({
                                    ProcessId::cancel(timeout_id)
                                        .and_then(move |()| {
                                            RefComp::read(r)
                                                .and_then(move |item| {
                                                    s_clone.trigger(item)
                                                })
                                        })
                                        .into_process()
                                })
                        })
                    })
                })
                .and_then(move |()| {
                    process_await(s.publish())
                })
                .call_process_boxed(cont, pid, p)
        }
    }
}

/// Allows running another process within the specified timeout.
#[must_use = "computations are lazy and do nothing unless to be run"]
#[derive(Clone)]
pub struct Timeout<M>
{
    /// Timeout.
    timeout: f64,

    /// The computation.
    comp: M
}

impl<M> Process for Timeout<M>
    where M: Process + Clone + 'static,
          M::Item: Clone
{
    type Item = Option<M::Item>;

    #[doc(hidden)]
    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
    {
        if is_process_cancelled(&pid, p) {
            revoke_process(cont, pid, p)
        } else {
            let Timeout { timeout, comp } = self;
            ProcessId::new()
                .into_process()
                .and_then(move |comp_id| {
                    timeout_process_using_id(timeout, Grc::new(comp_id), comp)
                })
                .call_process(cont, pid, p)
        }
    }

    #[doc(hidden)]
    fn call_process_boxed(self, cont: ProcessBoxCont<Self::Item>, pid: Grc<ProcessId>, p: &Point) -> simulation::Result<()> {
        if is_process_cancelled(&pid, p) {
            revoke_process_boxed(cont, pid, p)
        } else {
            let Timeout { timeout, comp } = self;
            ProcessId::new()
                .into_process()
                .and_then(move |comp_id| {
                    timeout_process_using_id(timeout, Grc::new(comp_id), comp)
                })
                .call_process_boxed(cont, pid, p)
        }
    }
}

/// Represents a temporarily frozen computation.
#[doc(hidden)]
#[derive(Clone)]
pub struct FrozenProcess<T> {

    /// The frozen continuation.
    comp: EventBox<Option<ProcessBoxCont<T>>>
}

impl<T> FrozenProcess<T>
    where T: 'static
{
    /// Unfreeze the process computation.
    #[doc(hidden)]
    #[inline]
    pub fn unfreeze(self, p: &Point) -> simulation::Result<Option<ProcessBoxCont<T>>> {
        let FrozenProcess { comp } = self;
        comp.call_box((p,))
    }

    /// Freeze the process computation.
    #[doc(hidden)]
    pub fn new(cont: ProcessBoxCont<T>, pid: Grc<ProcessId>, p: &Point) -> simulation::Result<FrozenProcess<T>>
        where T: Clone
    {
        let cont = substitute_process_priority_boxed(p.priority, cont);
        let rh: Grc<RefComp<Option<DisposableBox>>> = Grc::new(RefComp::new(None));
        let rc = Grc::new(RefComp::new(Some(cont)));
        let h  = {
            let rh = rh.clone();
            let rc = rc.clone();
            let weak_pid = Grc::downgrade(&pid);
            pid.cancel_initiating()
                .subscribe(cons_observer(move |_, p| {
                    let pid = weak_pid.upgrade().expect("The process identifier cannot be removed");
                    let h = rh.swap_at(None, p).unwrap();
                    h.dispose(p)?;
                    match rc.swap_at(None, p) {
                        None => Result::Ok(()),
                        Some(cont) => {
                            let pid = pid.clone();
                            enqueue_event(p.time, {
                                cons_event(move |p| {
                                    let z = is_process_cancelled(&pid, p);
                                    if z {
                                        revoke_process_boxed(cont, pid, p)
                                    } else {
                                        Result::Ok(())
                                    }
                                }).into_boxed()
                            }).call_event(p)
                        }
                    }
                }))
                .call_event(p)
        };
        match h {
            Result::Err(e) => Result::Err(e),
            Result::Ok(h) => {
                rh.write_at(Some(h), p);
                let comp = {
                    cons_event(move |p| {
                        let h = rh.swap_at(None, p).unwrap();
                        h.dispose(p)?;
                        let cont = rc.swap_at(None, p);
                        Result::Ok(cont)
                    }).into_boxed()
                };
                Result::Ok(FrozenProcess { comp: comp })
            }
        }
    }

    /// Freeze the computation parameters specifying what should be done when reentering the computation.
    #[doc(hidden)]
    pub fn with_reentering<M>(cont: ProcessBoxCont<T>, pid: Grc<ProcessId>, val: T, comp: M, p: &Point) -> simulation::Result<FrozenProcess<T>>
        where M: Process<Item = T> + Clone + 'static,
              T: Clone
    {
        let cont = substitute_process_priority_boxed(p.priority, cont);
        let rh: Grc<RefComp<Option<DisposableBox>>> = Grc::new(RefComp::new(None));
        let rc = Grc::new(RefComp::new(Some(cont)));
        let h  = {
            let rh = rh.clone();
            let rc = rc.clone();
            let weak_pid = Grc::downgrade(&pid);
            pid.cancel_initiating()
                .subscribe(cons_observer(move |_, p| {
                    let pid = weak_pid.upgrade().expect("The process identifier cannot be removed");
                    let h = rh.swap_at(None, p).unwrap();
                    h.dispose(p)?;
                    match rc.swap_at(None, p) {
                        None => Result::Ok(()),
                        Some(cont) => {
                            let pid = pid.clone();
                            enqueue_event(p.time, {
                                cons_event(move |p| {
                                    let z = is_process_cancelled(&pid, p);
                                    if z {
                                        revoke_process_boxed(cont, pid, p)
                                    } else {
                                        Result::Ok(())
                                    }
                                }).into_boxed()
                            }).call_event(p)
                        }
                    }
                }))
                .call_event(p)
        };
        match h {
            Result::Err(e) => Result::Err(e),
            Result::Ok(h) => {
                rh.write_at(Some(h), p);
                let comp = {
                    cons_event(move |p| {
                        let h = rh.swap_at(None, p).unwrap();
                        h.dispose(p)?;
                        match rc.swap_at(None, p) {
                            None => Result::Ok(None),
                            Some(cont) => {
                                let f = pid.is_preempted(p);
                                if !f  {
                                    Result::Ok(Some(cont))
                                } else {
                                    let cont = ProcessBoxCont::new(move |a, pid, p| {
                                        match a {
                                            Result::Ok(_) => comp.call_process_boxed(cont, pid, p),
                                            Result::Err(e) => cont.call_box((Result::Err(e), pid, p))
                                        }
                                    });
                                    match sleep_process(cont, pid, val, p) {
                                        Result::Ok(()) => Result::Ok(None),
                                        Result::Err(e) => Result::Err(e)
                                    }
                                }
                            }
                        }
                    }).into_boxed()
                };
                Result::Ok(FrozenProcess { comp: comp })
            }
        }
    }
}

/// Reenter the process computation.
#[doc(hidden)]
pub fn reenter_process<T>(cont: ProcessBoxCont<T>, pid: Grc<ProcessId>, val: T, p: &Point) -> simulation::Result<()>
    where T: Clone + 'static
{
    let f = pid.is_preempted(p);
    if !f {
        enqueue_event(p.time, {
            cons_event(move |p| {
                let f = pid.is_preempted(p);
                if !f {
                    resume_process_boxed(cont, pid, val, p)
                } else {
                    sleep_process(cont, pid, val, p)
                }
            }).into_boxed()
        }).call_event(p)
    } else {
        sleep_process(cont, pid, val, p)
    }
}

/// Sleep until the preempted computation will be reentered.
#[doc(hidden)]
pub fn sleep_process<T>(cont: ProcessBoxCont<T>, pid: Grc<ProcessId>, val: T, p: &Point) -> simulation::Result<()>
    where T: Clone + 'static
{
    let rh: Grc<RefComp<Option<DisposableBox>>> = Grc::new(RefComp::new(None));
    let rc = Grc::new(RefComp::new(Some(cont)));
    let rv = Grc::new(RefComp::new(Some(val)));
    let h  = {
        let rh = rh.clone();
        pid.observable()
            .subscribe(cons_observer(move |e, p| {
                let h = rh.swap_at(None, p).unwrap();
                h.dispose(p)?;
                match e {
                    &ProcessEvent::CancelInitiating => {
                        let pid = pid.clone();
                        let rc  = rc.clone();
                        enqueue_event(p.time, {
                            cons_event(move |p| {
                                let z = is_process_cancelled(&pid, p);
                                if z {
                                    let cont = rc.swap_at(None, p).unwrap();
                                    revoke_process_boxed(cont, pid, p)
                                } else {
                                    Result::Ok(())
                                }
                            }).into_boxed()
                        }).call_event(p)
                    },
                    &ProcessEvent::PreemptionEnding => {
                        let pid = pid.clone();
                        let rc  = rc.clone();
                        let rv  = rv.clone();
                        enqueue_event(p.time, {
                            cons_event(move |p| {
                                let cont = rc.swap_at(None, p).unwrap();
                                let val  = rv.swap_at(None, p).unwrap();
                                reenter_process(cont, pid, val, p)
                            }).into_boxed()
                        }).call_event(p)
                    },
                    &ProcessEvent::PreemptionInitiating => {
                        panic!("The computation was already preempted")
                    }
                }
            }))
            .call_event(p)
    };
    match h {
        Result::Err(e) => Result::Err(e),
        Result::Ok(h) => {
            rh.write_at(Some(h), p);
            Result::Ok(())
        }
    }
}

/// Substitute the process computation.
#[doc(hidden)]
pub fn substitute_process<C, T, F>(cont: C, f: F) -> ProcessBoxCont<T>
    where C: FnOnce(simulation::Result<T>, Grc<ProcessId>, &Point) -> simulation::Result<()> + Clone + 'static,
          T: 'static,
          F: FnOnce(C, Grc<ProcessId>, T, &Point) -> simulation::Result<()> + Clone + 'static,
{
    ProcessBoxCont::new(move |a, pid, p| {
        match a {
            Result::Ok(a) => f(cont, pid, a, p),
            Result::Err(e) => cont(Result::Err(e), pid, p)
        }
    })
}

/// Substitute the process computation.
#[doc(hidden)]
pub fn substitute_process_boxed<T, F>(cont: ProcessBoxCont<T>, f: F) -> ProcessBoxCont<T>
    where T: 'static,
          F: FnOnce(ProcessBoxCont<T>, Grc<ProcessId>, T, &Point) -> simulation::Result<()> + Clone + 'static
{
    ProcessBoxCont::new(move |a, pid, p| {
        match a {
            Result::Ok(a) => f(cont, pid, a, p),
            Result::Err(e) => cont.call_box((Result::Err(e), pid, p))
        }
    })
}

/// Substitute the process priority.
#[doc(hidden)]
pub fn substitute_process_priority<C, T>(priority: isize, cont: C) -> ProcessBoxCont<T>
    where C: FnOnce(simulation::Result<T>, Grc<ProcessId>, &Point) -> simulation::Result<()> + Clone + 'static,
          T: Clone + 'static
{
    ProcessBoxCont::new(move |a, pid, p| {
        match a {
            Result::Ok(a) => {
                if p.priority == priority {
                    cont(Result::Ok(a), pid, p)
                } else {
                    enqueue_event_with_priority(p.time, priority, {
                        cons_event(move |p| {
                            resume_process(cont, pid, a, p)
                        }).into_boxed()
                    }).call_event(p)
                }
            },
            Result::Err(e) => {
                cont(Result::Err(e), pid, p)
            }
        }
    })
}

/// Substitute the process priority.
#[doc(hidden)]
pub fn substitute_process_priority_boxed<T>(priority: isize, cont: ProcessBoxCont<T>) -> ProcessBoxCont<T>
    where T: Clone + 'static
{
    ProcessBoxCont::new(move |a, pid, p| {
        match a {
            Result::Ok(a) => {
                if p.priority == priority {
                    cont.call_box((Result::Ok(a), pid, p))
                } else {
                    enqueue_event_with_priority(p.time, priority, {
                        cons_event(move |p| {
                            resume_process_boxed(cont, pid, a, p)
                        }).into_boxed()
                    }).call_event(p)
                }
            },
            Result::Err(e) => {
                cont.call_box((Result::Err(e), pid, p))
            }
        }
    })
}

/// Await a signal that should be emitted by the specified observable.
#[must_use = "computations are lazy and do nothing unless to be run"]
#[derive(Clone)]
pub struct Await<O, M> {

    /// The observable.
    observable: O,

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

impl<O, M> Process for Await<O, M>
    where O: Observable<Message = M>,
          M: Clone + 'static
{
    type Item = M;

    #[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
    {
        self.call_process_boxed(ProcessBoxCont::new(cont), pid, p)
    }

    #[doc(hidden)]
    fn call_process_boxed(self, cont: ProcessBoxCont<Self::Item>, pid: Grc<ProcessId>, p: &Point) -> simulation::Result<()> {
        match FrozenProcess::new(cont, pid.clone(), p) {
            Result::Err(e) => Result::Err(e),
            Result::Ok(cont) => {
                let rh: Grc<RefComp<Option<DisposableBox>>> = Grc::new(RefComp::new(None));
                let rh2: Grc<RefComp<Option<DisposableBox>>> = Grc::new(RefComp::new(None));
                let rc = RefComp::new(Some(cont));
                let h  = {
                    let rh = rh.clone();
                    let rh2 = rh2.clone();
                    let pid = pid.clone();
                    let Await { observable, _phantom } = self;
                    observable
                        .subscribe(cons_observer(move |x: &M, p| {
                            if let Some(h) = rh.swap_at(None, p) {
                                h.dispose(p)?;
                            }                            
                            if let Some(h2) = rh2.swap_at(None, p) {
                                h2.dispose(p)?;
                            }                            
                            let cont = rc.swap_at(None, p).unwrap();
                            match cont.unfreeze(p) {
                                Result::Err(e) => Result::Err(e),
                                Result::Ok(None) => Result::Ok(()),
                                Result::Ok(Some(cont)) => {
                                    let pid = pid.clone();
                                    reenter_process(cont, pid, x.clone(), p)
                                }
                            }
                        }))
                        .call_event(p)
                };
                match h {
                    Result::Err(e) => return Result::Err(e),
                    Result::Ok(h) => {
                        let h2 = {
                            let rh = rh.clone();
                            let rh2 = rh2.clone();
                            pid.cancel_initiating()
                                .subscribe(cons_observer(move |_, p| {
                                    if let Some(h) = rh.swap_at(None, p) {
                                        h.dispose(p)?;
                                    }                            
                                    if let Some(h2) = rh2.swap_at(None, p) {
                                        h2.dispose(p)?;
                                    }
                                    Result::Ok(())
                                }))
                                .call_event(p)
                        };
                        match h2 {
                            Result::Err(e) => {
                                h.dispose(p)?;
                                return Result::Err(e)
                            },
                            Result::Ok(h2) => {
                                rh.write_at(Some(h), p);
                                rh2.write_at(Some(h2), p);
                                Result::Ok(())
                            }
                        }
                    }
                }
            }
        }
    }
}

/// Await a cancellable signal that should be emitted by the specified observable.
#[must_use = "computations are lazy and do nothing unless to be run"]
#[derive(Clone)]
pub struct AwaitResult<O, M> {

    /// The observable.
    observable: O,

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

impl<O, M> Process for AwaitResult<O, M>
    where O: Observable<Message = simulation::Result<M>>,
          M: Clone + 'static
{
    type Item = M;

    #[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
    {
        self.call_process_boxed(ProcessBoxCont::new(cont), pid, p)
    }

    #[doc(hidden)]
    fn call_process_boxed(self, cont: ProcessBoxCont<Self::Item>, pid: Grc<ProcessId>, p: &Point) -> simulation::Result<()> {
        match FrozenProcess::new(cont, pid.clone(), p) {
            Result::Err(e) => Result::Err(e),
            Result::Ok(cont) => {
                let rh: Grc<RefComp<Option<DisposableBox>>> = Grc::new(RefComp::new(None));
                let rh2: Grc<RefComp<Option<DisposableBox>>> = Grc::new(RefComp::new(None));
                let rc = RefComp::new(Some(cont));
                let h  = {
                    let rh = rh.clone();
                    let rh2 = rh2.clone();
                    let pid = pid.clone();
                    let AwaitResult { observable, _phantom } = self;
                    observable
                        .subscribe(cons_observer(move |x: &simulation::Result<M>, p| {
                            if let Some(h) = rh.swap_at(None, p) {
                                h.dispose(p)?;
                            }                            
                            if let Some(h2) = rh2.swap_at(None, p) {
                                h2.dispose(p)?;
                            }                            
                            let cont = rc.swap_at(None, p).unwrap();
                            match cont.unfreeze(p) {
                                Result::Err(e) => Result::Err(e),
                                Result::Ok(None) => Result::Ok(()),
                                Result::Ok(Some(cont)) => {
                                    let pid = pid.clone();
                                    match x {
                                        Result::Ok(x) => {
                                            reenter_process(cont, pid, x.clone(), 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.clone(), p)
                                                },
                                                &OtherError::Panic(_) => {
                                                    cut_error_process_boxed(cont, pid, e.clone(), p)
                                                },
                                                &OtherError::IO(_) => {
                                                    propagate_error_process_boxed(cont, pid, e.clone(), p)
                                                }
                                            }
                                        }
                                    }
                                }
                            }
                        }))
                        .call_event(p)
                };
                match h {
                    Result::Err(e) => return Result::Err(e),
                    Result::Ok(h) => {
                        let h2 = {
                            let rh = rh.clone();
                            let rh2 = rh2.clone();
                            pid.cancel_initiating()
                                .subscribe(cons_observer(move |_, p| {
                                    if let Some(h) = rh.swap_at(None, p) {
                                        h.dispose(p)?;
                                    }                            
                                    if let Some(h2) = rh2.swap_at(None, p) {
                                        h2.dispose(p)?;
                                    }
                                    Result::Ok(())
                                }))
                                .call_event(p)
                        };
                        match h2 {
                            Result::Err(e) => {
                                h.dispose(p)?;
                                return Result::Err(e)
                            },
                            Result::Ok(h2) => {
                                rh.write_at(Some(h), p);
                                rh2.write_at(Some(h2), p);
                                Result::Ok(())
                            }
                        }
                    }
                }
            }
        }
    }
}

/// Like the GoTo statement it transfers the direction of computation.
#[must_use = "computations are lazy and do nothing unless to be run"]
#[derive(Clone)]
pub struct Transfer<T, M> {

    /// The computation to transfer the flow control to.
    comp: M,

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

impl<T, M> Process for Transfer<T, M>
    where M: Process<Item = ()>
{
    type Item = T;

    #[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
    {
        if is_process_cancelled(&pid, p) {
            revoke_process(cont, pid, p)
        } else {
            let Transfer { comp, _phantom } = self;
            fn cont(a: simulation::Result<()>, _pid: Grc<ProcessId>, _p: &Point) -> simulation::Result<()> {
                a
            }
            comp.call_process(cont, pid, p)
        }
    }

    #[doc(hidden)]
    #[inline]
    fn call_process_boxed(self, cont: ProcessBoxCont<Self::Item>, pid: Grc<ProcessId>, p: &Point) -> simulation::Result<()> {
        if is_process_cancelled(&pid, p) {
            revoke_process_boxed(cont, pid, p)
        } else {
            let Transfer { comp, _phantom } = self;
            fn cont(a: simulation::Result<()>, _pid: Grc<ProcessId>, _p: &Point) -> simulation::Result<()> {
                a
            }
            comp.call_process(cont, pid, p)
        }
    }
}

/// The process that never returns a result.
#[must_use = "computations are lazy and do nothing unless to be run"]
#[derive(Clone)]
pub struct Never<T> {

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

impl<T> Process for Never<T> {

    type Item = T;

    #[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
    {
        Result::Ok(())
    }

    #[doc(hidden)]
    #[inline]
    fn call_process_boxed(self, _cont: ProcessBoxCont<Self::Item>, _pid: Grc<ProcessId>, _p: &Point) -> simulation::Result<()> {
        Result::Ok(())
    }
}

/// Register a handler that will be invoked in case of cancelling the current process.
#[must_use = "computations are lazy and do nothing unless to be run"]
#[derive(Clone)]
pub struct WhenCancelling<M> {

    /// The computation.
    comp: M
}

impl<M> Process for WhenCancelling<M>
    where M: Observer<Message = (), Item = ()> + 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
    {
        if is_process_cancelled(&pid, p) {
            revoke_process(cont, pid, p)
        } else {
            let WhenCancelling { comp } = self;
            pid.cancel_initiating()
                .subscribe(cons_observer(move |_, p| {
                    comp.call_observer(&(), p)
                }))
                .call_event(p)?;
            resume_process(cont, pid, (), p)
        }
    }

    #[doc(hidden)]
    #[inline]
    fn call_process_boxed(self, cont: ProcessBoxCont<Self::Item>, pid: Grc<ProcessId>, p: &Point) -> simulation::Result<()> {
        if is_process_cancelled(&pid, p) {
            revoke_process_boxed(cont, pid, p)
        } else {
            let WhenCancelling { comp } = self;
            pid.cancel_initiating()
                .subscribe(cons_observer(move |_, p| {
                    comp.call_observer(&(), p)
                }))
                .call_event(p)?;
            resume_process_boxed(cont, pid, (), p)
        }
    }
}

/// Allows creating the `Process` computation that panics with the specified message.
#[must_use = "computations are lazy and do nothing unless to be run"]
#[derive(Clone)]
pub struct Panic<T> {

    /// The panic message.
    msg: String,

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

impl<T> Process for Panic<T> {

    type Item = T;

    #[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
    {
        if is_process_cancelled(&pid, p) {
            revoke_process(cont, pid, p)
        } else {
            let Panic { msg, _phantom } = self;
            let err = Rc::new(OtherError::Panic(msg));
            cut_error_process(cont, pid, err, p)
        }
    }

    #[doc(hidden)]
    #[inline]
    fn call_process_boxed(self, cont: ProcessBoxCont<Self::Item>, pid: Grc<ProcessId>, p: &Point) -> simulation::Result<()> {
        if is_process_cancelled(&pid, p) {
            revoke_process_boxed(cont, pid, p)
        } else {
            let Panic { msg, _phantom } = self;
            let err = Rc::new(OtherError::Panic(msg));
            cut_error_process_boxed(cont, pid, err, p)
        }
    }
}

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

    /// The computation.
    comp: M,

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

impl<M> Process for Trace<M>
    where M: Process
{
    type Item = M::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
    {
        if is_process_cancelled(&pid, p) {
            revoke_process(cont, pid, p)
        } else {
            let Trace { comp, msg } = self;
            p.trace(&msg);
            comp.call_process(cont, pid, p)
        }
    }

    #[doc(hidden)]
    #[inline]
    fn call_process_boxed(self, cont: ProcessBoxCont<Self::Item>, pid: Grc<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_process_boxed(cont, pid, p)
        }
    }
}

#[must_use = "computations are lazy and do nothing unless to be run"]
#[derive(Clone)]
pub struct ProcessWithPriority {

    /// The time point priority.
    priority: isize
}

impl Process for ProcessWithPriority {

    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
    {
        if is_process_cancelled(&pid, p) {
            revoke_process(cont, pid, p)
        } else {
            let ProcessWithPriority { priority } = self;
            if priority == p.priority {
                cont(Result::Ok(()), pid, p)
            } else {
                let comp = cons_event(move |p| {
                    if is_process_cancelled(&pid, p) {
                        revoke_process(cont, pid, p)
                    } else {
                        cont(Result::Ok(()), pid, p)
                    }
                });
                enqueue_event_with_priority(p.time, priority, comp.into_boxed())
                    .call_event(p)
            }
        }
    }

    #[doc(hidden)]
    #[inline]
    fn call_process_boxed(self, cont: ProcessBoxCont<Self::Item>, pid: Grc<ProcessId>, p: &Point) -> simulation::Result<()> {
        if is_process_cancelled(&pid, p) {
            revoke_process_boxed(cont, pid, p)
        } else {
            let ProcessWithPriority { priority } = self;
            if priority == p.priority {
                cont.call_box((Result::Ok(()), pid, p))
            } else {
                let comp = cons_event(move |p| {
                    if is_process_cancelled(&pid, p) {
                        revoke_process_boxed(cont, pid, p)
                    } else {
                        cont.call_box((Result::Ok(()), pid, p))
                    }
                });
                enqueue_event_with_priority(p.time, priority, comp.into_boxed())
                    .call_event(p)
            }
        }
    }
}

/// Embeds the result value within the `Process` computation.
#[must_use = "computations are lazy and do nothing unless to be run"]
#[derive(Clone)]
pub struct EmbedResult<T> {

    /// The value to be embedded.
    val: simulation::Result<T>
}

impl<T> Process for EmbedResult<T> {

    type Item = T;

    #[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
    {
        if is_process_cancelled(&pid, p) {
            revoke_process(cont, pid, p)
        } else {
            let EmbedResult { val } = self;
            match val {
                Result::Ok(a) => {
                    cont(Result::Ok(a), pid, p)
                },
                Result::Err(Error::Cancel) => {
                    let comp = cancel_process();
                    comp.call_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)
                        }
                    }
                }
            }
        }
    }

    #[doc(hidden)]
    #[inline]
    fn call_process_boxed(self, cont: ProcessBoxCont<Self::Item>, pid: Grc<ProcessId>, p: &Point) -> simulation::Result<()> {
        if is_process_cancelled(&pid, p) {
            revoke_process_boxed(cont, pid, p)
        } else {
            let EmbedResult { val } = self;
            match val {
                Result::Ok(a) => {
                    cont.call_box((Result::Ok(a), pid, p))
                },
                Result::Err(Error::Cancel) => {
                    let comp = cancel_process();
                    comp.call_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)
                        }
                    }
                }
            }
        }
    }
}