2021-08-13 19:51:11 +00:00
|
|
|
#![doc = concat!(
|
|
|
|
include_str!("../doc/actor.md"),
|
|
|
|
include_str!("../doc/what-is-an-actor.md"),
|
|
|
|
include_str!("../doc/flow-control.md"),
|
2021-08-14 00:12:11 +00:00
|
|
|
include_str!("../doc/linked-tasks.md"),
|
2021-08-13 19:51:11 +00:00
|
|
|
)]
|
|
|
|
|
2021-09-23 19:43:32 +00:00
|
|
|
use super::dataflow::Graph;
|
2021-07-03 07:03:52 +00:00
|
|
|
use super::error::Error;
|
2021-07-24 21:22:01 +00:00
|
|
|
use super::error::encode_error;
|
2021-07-08 22:04:11 +00:00
|
|
|
use super::error::error;
|
2021-07-15 07:13:31 +00:00
|
|
|
use super::rewrite::CaveatError;
|
|
|
|
use super::rewrite::CheckedCaveat;
|
2021-09-23 19:43:32 +00:00
|
|
|
use super::schemas::sturdy;
|
2021-07-03 07:03:52 +00:00
|
|
|
|
2021-09-24 11:04:15 +00:00
|
|
|
use preserves::value::ArcValue;
|
2021-07-03 07:03:52 +00:00
|
|
|
use preserves::value::Domain;
|
|
|
|
use preserves::value::IOValue;
|
|
|
|
use preserves::value::Map;
|
|
|
|
use preserves::value::NestedValue;
|
2021-08-27 13:31:18 +00:00
|
|
|
use preserves::value::Set;
|
2021-09-19 14:53:37 +00:00
|
|
|
use preserves_schema::ParseError;
|
|
|
|
use preserves_schema::support::Parse;
|
|
|
|
use preserves_schema::support::Unparse;
|
2021-07-03 07:03:52 +00:00
|
|
|
|
2021-09-23 19:43:32 +00:00
|
|
|
use std::any::Any;
|
2021-07-03 07:03:52 +00:00
|
|
|
use std::boxed::Box;
|
|
|
|
use std::collections::hash_map::HashMap;
|
2021-07-22 14:53:56 +00:00
|
|
|
use std::convert::TryFrom;
|
2021-07-15 11:13:22 +00:00
|
|
|
use std::convert::TryInto;
|
2021-09-23 19:43:32 +00:00
|
|
|
use std::marker::PhantomData;
|
2021-08-27 13:31:18 +00:00
|
|
|
use std::num::NonZeroU64;
|
2021-07-03 07:03:52 +00:00
|
|
|
use std::sync::Arc;
|
2021-07-25 21:12:07 +00:00
|
|
|
use std::sync::Mutex;
|
2021-07-15 11:13:22 +00:00
|
|
|
use std::sync::RwLock;
|
2021-07-22 14:53:56 +00:00
|
|
|
use std::sync::Weak;
|
|
|
|
use std::sync::atomic::{AtomicI64, AtomicU64, Ordering};
|
2021-08-30 21:41:51 +00:00
|
|
|
use std::time;
|
2021-07-03 07:03:52 +00:00
|
|
|
|
|
|
|
use tokio::select;
|
2021-07-27 14:30:42 +00:00
|
|
|
use tokio::sync::Notify;
|
2021-07-03 07:03:52 +00:00
|
|
|
use tokio::sync::mpsc::{unbounded_channel, UnboundedSender, UnboundedReceiver};
|
|
|
|
use tokio_util::sync::CancellationToken;
|
|
|
|
|
2021-07-08 22:04:11 +00:00
|
|
|
use tracing::Instrument;
|
|
|
|
|
2021-08-13 19:51:11 +00:00
|
|
|
/// The type of messages and assertions that can be exchanged among
|
|
|
|
/// distributed objects, including via [dataspace][crate::dataspace].
|
|
|
|
///
|
|
|
|
/// A Preserves value where embedded references are instances of
|
|
|
|
/// [`Cap`].
|
|
|
|
///
|
|
|
|
/// While [`Ref<M>`] can be used within a process, where arbitrary
|
|
|
|
/// `M`-values can be exchanged among objects, for distributed or
|
|
|
|
/// polyglot systems a *lingua franca* has to be chosen. `AnyValue` is
|
|
|
|
/// that language.
|
2021-09-24 11:04:15 +00:00
|
|
|
pub type AnyValue = ArcValue<Arc<Cap>>;
|
2021-08-11 21:16:01 +00:00
|
|
|
|
2021-08-13 19:51:11 +00:00
|
|
|
/// The type of process-unique actor IDs.
|
2021-08-11 21:40:48 +00:00
|
|
|
pub type ActorId = u64;
|
2021-08-13 19:51:11 +00:00
|
|
|
|
2021-08-27 13:31:18 +00:00
|
|
|
/// The type of process-unique facet IDs.
|
|
|
|
pub type FacetId = NonZeroU64;
|
|
|
|
|
2021-08-13 19:51:11 +00:00
|
|
|
/// The type of process-unique assertion handles.
|
|
|
|
///
|
|
|
|
/// Used both as a reference to [retract][Entity::retract]
|
|
|
|
/// previously-asserted assertions and as an indexing key to associate
|
|
|
|
/// local state with some incoming assertion in an entity.
|
2021-07-22 08:07:49 +00:00
|
|
|
pub type Handle = u64;
|
2021-07-03 07:03:52 +00:00
|
|
|
|
2021-09-23 19:43:32 +00:00
|
|
|
/// The type of process-unique field instance IDs.
|
|
|
|
pub type FieldId = NonZeroU64;
|
|
|
|
|
|
|
|
/// The type of process-unique field observer block IDs.
|
|
|
|
pub type BlockId = NonZeroU64;
|
|
|
|
|
2021-08-13 19:51:11 +00:00
|
|
|
/// Responses to events must have type `ActorResult`.
|
2021-07-03 07:03:52 +00:00
|
|
|
pub type ActorResult = Result<(), Error>;
|
2021-08-13 19:51:11 +00:00
|
|
|
|
2021-08-27 13:31:18 +00:00
|
|
|
/// The [`Actor::boot`] method returns an `ActorHandle`, representing
|
|
|
|
/// the actor's mainloop task.
|
2021-07-03 07:03:52 +00:00
|
|
|
pub type ActorHandle = tokio::task::JoinHandle<ActorResult>;
|
|
|
|
|
2021-08-30 21:41:51 +00:00
|
|
|
/// The type of the "disarm" function returned from [`Activation::prevent_inert_check`].
|
|
|
|
pub type DisarmFn = Box<dyn Send + FnOnce()>;
|
|
|
|
|
2021-08-13 19:51:11 +00:00
|
|
|
/// A small protocol for indicating successful synchronisation with
|
|
|
|
/// some peer; see [Entity::sync].
|
2021-08-30 21:41:51 +00:00
|
|
|
#[derive(Debug)]
|
2021-07-22 14:53:56 +00:00
|
|
|
pub struct Synced;
|
2021-07-21 23:05:08 +00:00
|
|
|
|
2021-08-13 19:51:11 +00:00
|
|
|
/// The core metaprotocol implemented by every object.
|
|
|
|
///
|
|
|
|
/// Entities communicate with each other by asserting and retracting
|
|
|
|
/// values and by sending messages (which can be understood very
|
|
|
|
/// approximately as "infinitesimally brief" assertions of the message
|
|
|
|
/// body).
|
|
|
|
///
|
|
|
|
/// Every assertion placed at a receiving entity *R* from some sending
|
|
|
|
/// entity *S* lives so long as *S*'s actor survives, *R*'s actor
|
|
|
|
/// survives, and *S* does not retract the assertion. Messages, by
|
|
|
|
/// contrast, are transient.
|
|
|
|
///
|
|
|
|
/// Implementors of [`Entity`] accept assertions from peers in method
|
|
|
|
/// [`assert`][Entity::assert]; notification of retraction of a
|
|
|
|
/// previously-asserted value happens in method
|
|
|
|
/// [`retract`][Entity::retract]; and notification of a message in
|
|
|
|
/// method [`message`][Entity::message].
|
|
|
|
///
|
|
|
|
/// In addition, entities may *synchronise* with each other: the
|
|
|
|
/// [`sync`][Entity::sync] method responds to a synchronisation
|
|
|
|
/// request.
|
|
|
|
///
|
|
|
|
/// Finally, the Rust implementation of the Syndicated Actor model
|
|
|
|
/// offers a hook for running some code at the end of an Entity's
|
|
|
|
/// containing [`Actor`]'s lifetime
|
|
|
|
/// ([`exit_hook`][Entity::exit_hook]).
|
|
|
|
///
|
|
|
|
/// # What to implement
|
|
|
|
///
|
|
|
|
/// The default implementations of the methods here generally do
|
|
|
|
/// nothing; override them to add actual behaviour to your entity.
|
|
|
|
///
|
|
|
|
#[allow(unused_variables)]
|
2021-07-26 08:53:56 +00:00
|
|
|
pub trait Entity<M>: Send {
|
2021-08-13 19:51:11 +00:00
|
|
|
/// Receive notification of a new assertion from a peer.
|
|
|
|
///
|
|
|
|
/// The `turn` parameter represents the current
|
|
|
|
/// [activation][Activation]; `assertion` is the value (of type
|
|
|
|
/// `M`) asserted; and `handle` is the process-unique name for
|
|
|
|
/// this particular assertion instance that will be used later
|
|
|
|
/// when it is [retracted][Entity::retract].
|
|
|
|
///
|
|
|
|
/// The default implementation does nothing.
|
|
|
|
fn assert(&mut self, turn: &mut Activation, assertion: M, handle: Handle) -> ActorResult {
|
2021-07-03 07:03:52 +00:00
|
|
|
Ok(())
|
|
|
|
}
|
2021-08-13 19:51:11 +00:00
|
|
|
|
|
|
|
/// Receive notification of retraction of a previous assertion from a peer.
|
|
|
|
///
|
|
|
|
/// This happens either when the asserting peer explicitly
|
|
|
|
/// retracts an assertion, or when its animating [`Actor`]
|
|
|
|
/// terminates.
|
|
|
|
///
|
|
|
|
/// The `turn` parameter represents the current
|
|
|
|
/// [activation][Activation], and `handle` is the process-unique
|
|
|
|
/// name for this particular assertion instance being retracted.
|
|
|
|
///
|
|
|
|
/// Note that no `assertion` value is provided: entities needing
|
|
|
|
/// to know the value that was previously asserted must remember
|
|
|
|
/// it themselves (perhaps in a [`Map`] keyed by `handle`).
|
|
|
|
///
|
|
|
|
/// The default implementation does nothing.
|
|
|
|
fn retract(&mut self, turn: &mut Activation, handle: Handle) -> ActorResult {
|
2021-07-03 07:03:52 +00:00
|
|
|
Ok(())
|
|
|
|
}
|
2021-08-13 19:51:11 +00:00
|
|
|
|
|
|
|
/// Receive notification of a message from a peer.
|
|
|
|
///
|
|
|
|
/// The `turn` parameter represents the current
|
|
|
|
/// [activation][Activation], and `message` is the body of the
|
|
|
|
/// message sent.
|
|
|
|
///
|
|
|
|
/// The default implementation does nothing.
|
|
|
|
fn message(&mut self, turn: &mut Activation, message: M) -> ActorResult {
|
2021-07-03 07:03:52 +00:00
|
|
|
Ok(())
|
|
|
|
}
|
2021-08-13 19:51:11 +00:00
|
|
|
|
|
|
|
/// Respond to a synchronisation request from a peer.
|
|
|
|
///
|
|
|
|
/// Implementors of [`Entity`] will seldom override this. The
|
|
|
|
/// default implementation fulfils the synchronisation protocol by
|
|
|
|
/// responding to `peer` with a `Synced` message.
|
|
|
|
///
|
|
|
|
/// In special cases, for example when an entity is a proxy for
|
|
|
|
/// some remote entity, the right thing to do is to forward the
|
|
|
|
/// synchronisation request on to another entity; in those cases,
|
|
|
|
/// overriding the default behaviour is appropriate.
|
|
|
|
fn sync(&mut self, turn: &mut Activation, peer: Arc<Ref<Synced>>) -> ActorResult {
|
|
|
|
turn.message(&peer, Synced);
|
2021-07-06 18:56:36 +00:00
|
|
|
Ok(())
|
|
|
|
}
|
2021-08-13 19:51:11 +00:00
|
|
|
|
2021-08-30 12:17:40 +00:00
|
|
|
/// Optional callback for running actions when the entity's owning [Facet] terminates
|
|
|
|
/// cleanly. Will not be called in case of abnormal shutdown (crash) of an actor.
|
|
|
|
///
|
2021-09-24 14:14:55 +00:00
|
|
|
/// Programs register an entity's stop hook with [Activation::on_stop_notify].
|
2021-08-30 12:17:40 +00:00
|
|
|
///
|
|
|
|
/// The default implementation does nothing.
|
|
|
|
fn stop(&mut self, turn: &mut Activation) -> ActorResult {
|
|
|
|
Ok(())
|
|
|
|
}
|
|
|
|
|
2021-08-13 19:51:11 +00:00
|
|
|
/// Optional callback for running cleanup actions when the
|
|
|
|
/// entity's animating [Actor] terminates.
|
|
|
|
///
|
|
|
|
/// Programs register an entity's exit hook with
|
|
|
|
/// [RunningActor::add_exit_hook].
|
|
|
|
///
|
|
|
|
/// The default implementation does nothing.
|
|
|
|
fn exit_hook(&mut self, turn: &mut Activation, exit_status: &Arc<ActorResult>) -> ActorResult {
|
2021-07-21 23:05:08 +00:00
|
|
|
Ok(())
|
2021-07-06 18:56:36 +00:00
|
|
|
}
|
2021-07-03 07:03:52 +00:00
|
|
|
}
|
|
|
|
|
2021-08-13 19:51:11 +00:00
|
|
|
/// An "inert" entity, that does nothing in response to any event delivered to it.
|
|
|
|
///
|
|
|
|
/// Useful as a placeholder or dummy in various situations.
|
2021-07-24 21:22:01 +00:00
|
|
|
pub struct InertEntity;
|
|
|
|
impl<M> Entity<M> for InertEntity {}
|
|
|
|
|
2021-07-23 06:11:48 +00:00
|
|
|
enum CleanupAction {
|
2021-07-24 21:22:01 +00:00
|
|
|
ForMyself(Action),
|
|
|
|
ForAnother(Arc<Mailbox>, Action),
|
2021-07-12 15:41:12 +00:00
|
|
|
}
|
|
|
|
|
2021-07-23 06:11:48 +00:00
|
|
|
type CleanupActions = Map<Handle, CleanupAction>;
|
2021-07-03 07:03:52 +00:00
|
|
|
|
2021-08-13 19:51:11 +00:00
|
|
|
type Action = Box<dyn Send + FnOnce(&mut Activation) -> ActorResult>;
|
2021-09-23 19:43:32 +00:00
|
|
|
type Block = Box<dyn Send + FnMut(&mut Activation) -> ActorResult>;
|
2021-08-14 00:12:11 +00:00
|
|
|
|
|
|
|
#[doc(hidden)]
|
|
|
|
pub type PendingEventQueue = Vec<Action>;
|
2021-08-13 19:51:11 +00:00
|
|
|
|
|
|
|
/// The main API for programming Syndicated Actor objects.
|
|
|
|
///
|
|
|
|
/// Through `Activation`s, programs can access the state of their
|
2021-08-27 13:31:18 +00:00
|
|
|
/// animating [`RunningActor`] and their active [`Facet`].
|
|
|
|
///
|
|
|
|
/// Usually, an `Activation` will be supplied to code that needs one; but when non-Actor code
|
|
|
|
/// (such as a [linked task][crate::actor#linked-tasks]) needs to enter an Actor's execution
|
|
|
|
/// context, use [`FacetRef::activate`] to construct one.
|
2021-08-13 19:51:11 +00:00
|
|
|
///
|
|
|
|
/// Many actions that an entity can perform are methods directly on
|
2021-08-27 13:31:18 +00:00
|
|
|
/// `Activation`, but methods on the [`RunningActor`] and [`FacetRef`]
|
2021-08-13 19:51:11 +00:00
|
|
|
/// values contained in an `Activation` are also sometimes useful.
|
|
|
|
///
|
|
|
|
/// This is what other implementations call a "Turn", renamed here to
|
2021-09-24 11:04:15 +00:00
|
|
|
/// avoid conflicts with [`crate::schemas::protocol::Turn`].
|
2021-07-03 07:03:52 +00:00
|
|
|
pub struct Activation<'activation> {
|
2021-08-27 13:31:18 +00:00
|
|
|
/// A reference to the currently active [`Facet`] and the implementation-side state of its
|
|
|
|
/// [`Actor`].
|
|
|
|
pub facet: FacetRef,
|
2021-08-13 19:51:11 +00:00
|
|
|
/// A reference to the current state of the active [`Actor`].
|
2021-07-24 21:22:01 +00:00
|
|
|
pub state: &'activation mut RunningActor,
|
2021-09-23 19:43:32 +00:00
|
|
|
active_block: Option<BlockId>,
|
2021-07-24 21:22:01 +00:00
|
|
|
pending: EventBuffer,
|
|
|
|
}
|
|
|
|
|
|
|
|
struct EventBuffer {
|
2021-08-13 19:51:11 +00:00
|
|
|
pub account: Arc<Account>,
|
2021-07-22 07:56:21 +00:00
|
|
|
queues: HashMap<ActorId, (UnboundedSender<SystemMessage>, PendingEventQueue)>,
|
2021-07-24 21:22:01 +00:00
|
|
|
for_myself: PendingEventQueue,
|
2021-07-03 07:03:52 +00:00
|
|
|
}
|
|
|
|
|
2021-08-13 19:51:11 +00:00
|
|
|
/// An `Account` records a "debt" in terms of outstanding work items.
|
|
|
|
///
|
|
|
|
/// It is part of the flow control mechanism - see [the module-level
|
|
|
|
/// documentation][crate::actor#flow-control] for more.
|
2021-07-15 11:13:22 +00:00
|
|
|
#[derive(Debug)]
|
2021-08-13 19:51:11 +00:00
|
|
|
pub struct Account {
|
2021-07-15 11:13:22 +00:00
|
|
|
id: u64,
|
|
|
|
debt: Arc<AtomicI64>,
|
2021-07-27 14:30:42 +00:00
|
|
|
notify: Notify,
|
2021-07-15 11:13:22 +00:00
|
|
|
}
|
|
|
|
|
2021-08-13 19:51:11 +00:00
|
|
|
/// A `LoanedItem<T>` is a `T` with an associated `cost` recorded
|
2021-08-14 00:12:11 +00:00
|
|
|
/// against it in the ledger of a given [`Account`]. The cost is
|
|
|
|
/// repaid automatically when the `LoanedItem<T>` is `Drop`ped.
|
2021-08-13 19:51:11 +00:00
|
|
|
///
|
2021-08-14 00:12:11 +00:00
|
|
|
/// `LoanedItem`s are part of the flow control mechanism - see [the
|
|
|
|
/// module-level documentation][crate::actor#flow-control] for more.
|
2021-07-15 11:13:22 +00:00
|
|
|
#[derive(Debug)]
|
|
|
|
pub struct LoanedItem<T> {
|
2021-08-13 19:51:11 +00:00
|
|
|
/// The account against which this loan is recorded.
|
|
|
|
pub account: Arc<Account>,
|
|
|
|
/// The cost of this particular `T`.
|
2021-07-15 11:13:22 +00:00
|
|
|
pub cost: usize,
|
2021-08-13 19:51:11 +00:00
|
|
|
/// The underlying item itself.
|
2021-07-15 11:13:22 +00:00
|
|
|
pub item: T,
|
|
|
|
}
|
|
|
|
|
2021-07-03 07:03:52 +00:00
|
|
|
enum SystemMessage {
|
|
|
|
Release,
|
2021-09-23 19:43:32 +00:00
|
|
|
ReleaseField(FieldId),
|
2021-07-21 21:53:55 +00:00
|
|
|
Turn(LoanedItem<PendingEventQueue>),
|
2021-07-03 07:03:52 +00:00
|
|
|
Crash(Error),
|
|
|
|
}
|
|
|
|
|
2021-08-13 19:51:11 +00:00
|
|
|
/// The mechanism by which events are delivered to a given [`Actor`].
|
2021-07-03 07:03:52 +00:00
|
|
|
pub struct Mailbox {
|
2021-08-13 19:51:11 +00:00
|
|
|
/// The ID of the actor this mailbox corresponds to.
|
2021-07-03 07:03:52 +00:00
|
|
|
pub actor_id: ActorId,
|
|
|
|
tx: UnboundedSender<SystemMessage>,
|
|
|
|
}
|
|
|
|
|
2021-08-13 19:51:11 +00:00
|
|
|
/// Each actor owns an instance of this structure.
|
|
|
|
///
|
|
|
|
/// It holds the receive-half of the actor's mailbox, plus a reference
|
|
|
|
/// to the actor's private state.
|
2021-07-03 07:03:52 +00:00
|
|
|
pub struct Actor {
|
2021-07-23 06:10:09 +00:00
|
|
|
rx: UnboundedReceiver<SystemMessage>,
|
2021-07-24 21:22:01 +00:00
|
|
|
ac_ref: ActorRef,
|
|
|
|
}
|
|
|
|
|
2021-08-13 19:51:11 +00:00
|
|
|
/// A reference to an actor's private [`ActorState`].
|
2021-07-24 21:22:01 +00:00
|
|
|
#[derive(Clone)]
|
|
|
|
pub struct ActorRef {
|
2021-08-13 19:51:11 +00:00
|
|
|
/// The ID of the referenced actor.
|
2021-07-24 21:22:01 +00:00
|
|
|
pub actor_id: ActorId,
|
2021-07-25 21:12:07 +00:00
|
|
|
state: Arc<Mutex<ActorState>>,
|
2021-07-24 21:22:01 +00:00
|
|
|
}
|
|
|
|
|
2021-08-27 13:31:18 +00:00
|
|
|
/// A combination of an [`ActorRef`] with a [`FacetId`], acting as a capability to enter the
|
|
|
|
/// execution context of a facet from a linked task.
|
|
|
|
#[derive(Clone)]
|
|
|
|
pub struct FacetRef {
|
|
|
|
pub actor: ActorRef,
|
|
|
|
pub facet_id: FacetId,
|
|
|
|
}
|
|
|
|
|
2021-08-13 19:51:11 +00:00
|
|
|
/// The state of an actor: either `Running` or `Terminated`.
|
2021-07-24 21:22:01 +00:00
|
|
|
pub enum ActorState {
|
2021-08-13 19:51:11 +00:00
|
|
|
/// A non-terminated actor has an associated [`RunningActor`] state record.
|
2021-07-24 21:22:01 +00:00
|
|
|
Running(RunningActor),
|
2021-08-13 19:51:11 +00:00
|
|
|
/// A terminated actor has an [`ActorResult`] as its `exit_status`.
|
2021-07-24 21:22:01 +00:00
|
|
|
Terminated {
|
2021-08-13 19:51:11 +00:00
|
|
|
/// The exit status of the actor: `Ok(())` for normal
|
|
|
|
/// termination, `Err(_)` for abnormal termination.
|
2021-07-24 21:22:01 +00:00
|
|
|
exit_status: Arc<ActorResult>,
|
|
|
|
},
|
|
|
|
}
|
|
|
|
|
2021-08-14 00:12:11 +00:00
|
|
|
/// State associated with each non-terminated [`Actor`].
|
2021-07-24 21:22:01 +00:00
|
|
|
pub struct RunningActor {
|
2021-08-14 00:12:11 +00:00
|
|
|
/// The ID of the actor this state belongs to.
|
2021-07-24 21:22:01 +00:00
|
|
|
pub actor_id: ActorId,
|
|
|
|
tx: UnboundedSender<SystemMessage>,
|
2021-07-22 14:53:56 +00:00
|
|
|
mailbox: Weak<Mailbox>,
|
2021-09-23 19:43:32 +00:00
|
|
|
dataflow: Graph<FieldId, BlockId>,
|
|
|
|
fields: HashMap<FieldId, Box<dyn Any + Send>>,
|
|
|
|
blocks: HashMap<BlockId, (FacetId, Block)>,
|
2021-08-27 13:31:18 +00:00
|
|
|
exit_hooks: Vec<Box<dyn Send + FnOnce(&mut Activation, &Arc<ActorResult>) -> ActorResult>>,
|
2021-09-07 17:12:32 +00:00
|
|
|
cleanup_actions: CleanupActions,
|
2021-08-27 13:31:18 +00:00
|
|
|
facet_nodes: Map<FacetId, Facet>,
|
|
|
|
facet_children: Map<FacetId, Set<FacetId>>,
|
|
|
|
root: FacetId,
|
|
|
|
}
|
|
|
|
|
2021-09-23 19:43:32 +00:00
|
|
|
/// Handle to a shared, mutable field (i.e. a *dataflow variable*) within a [`RunningActor`].
|
|
|
|
///
|
|
|
|
/// Use [`Activation::field`] to create fields, and use [`Activation::get`],
|
|
|
|
/// [`::get_mut`][Activation::get_mut], and [`::set`][Activation::set] to read and write field
|
|
|
|
/// values. Use [`Activation::dataflow`] to create a reactive block within a facet that will be
|
|
|
|
/// (re-)executed whenever some dependent field changes value.
|
|
|
|
///
|
|
|
|
#[derive(Debug)]
|
|
|
|
pub struct Field<T: Any + Send> {
|
|
|
|
pub field_id: FieldId,
|
|
|
|
tx: UnboundedSender<SystemMessage>,
|
|
|
|
phantom: PhantomData<T>,
|
|
|
|
}
|
|
|
|
|
2021-08-27 13:31:18 +00:00
|
|
|
/// State associated with each facet in an [`Actor`]'s facet tree.
|
|
|
|
///
|
|
|
|
/// # Inert facets
|
|
|
|
///
|
|
|
|
/// A facet is considered *inert* if:
|
|
|
|
///
|
|
|
|
/// 1. it has no child facets;
|
|
|
|
/// 2. it has no cleanup actions (that is, no assertions placed by any of its entities);
|
|
|
|
/// 3. it has no linked tasks; and
|
|
|
|
/// 4. it has no "inert check preventers" (see [Activation::prevent_inert_check]).
|
|
|
|
///
|
|
|
|
/// If a facet is created and is inert at the moment that its `boot` function returns, it is
|
|
|
|
/// automatically terminated.
|
|
|
|
///
|
|
|
|
/// When a facet is terminated, if its parent facet is inert, the parent is terminated.
|
|
|
|
///
|
|
|
|
/// If the root facet in an actor is terminated, the entire actor is terminated (with exit
|
|
|
|
/// status `Ok(())`).
|
|
|
|
///
|
|
|
|
pub struct Facet {
|
|
|
|
/// The ID of the facet.
|
|
|
|
pub facet_id: FacetId,
|
|
|
|
/// The ID of the facet's parent facet, if any; if None, this facet is the `Actor`'s root facet.
|
|
|
|
pub parent_facet_id: Option<FacetId>,
|
2021-09-07 17:12:32 +00:00
|
|
|
outbound_handles: Set<Handle>,
|
2021-08-27 13:31:18 +00:00
|
|
|
stop_actions: Vec<Action>,
|
2021-07-06 18:56:36 +00:00
|
|
|
linked_tasks: Map<u64, CancellationToken>,
|
2021-08-27 13:31:18 +00:00
|
|
|
inert_check_preventers: Arc<AtomicU64>,
|
2021-07-15 07:13:31 +00:00
|
|
|
}
|
|
|
|
|
2021-08-13 19:51:11 +00:00
|
|
|
/// A reference to an object that expects messages/assertions of type
|
|
|
|
/// `M`.
|
|
|
|
///
|
|
|
|
/// The object can be in the same actor, in a different local
|
|
|
|
/// (in-process) actor, or accessible across a network link.
|
2021-07-22 14:53:56 +00:00
|
|
|
pub struct Ref<M> {
|
2021-08-14 00:12:11 +00:00
|
|
|
/// Mailbox of the actor owning the referenced entity.
|
2021-07-22 14:53:56 +00:00
|
|
|
pub mailbox: Arc<Mailbox>,
|
2021-08-27 13:31:18 +00:00
|
|
|
/// ID of the facet (within the actor) owning the referenced entity.
|
|
|
|
pub facet_id: FacetId,
|
2021-08-14 00:12:11 +00:00
|
|
|
/// Mutex owning and guarding the state backing the referenced entity.
|
2021-07-25 21:12:07 +00:00
|
|
|
pub target: Mutex<Option<Box<dyn Entity<M>>>>,
|
2021-07-21 23:05:08 +00:00
|
|
|
}
|
|
|
|
|
2021-08-13 19:51:11 +00:00
|
|
|
/// Specialization of `Ref<M>` for messages/assertions of type
|
|
|
|
/// [`AnyValue`].
|
|
|
|
///
|
|
|
|
/// All polyglot and network communication is done in terms of `Cap`s.
|
|
|
|
///
|
|
|
|
/// `Cap`s can also be *attenuated* ([Hardy 2017]; [Miller 2006]) to
|
|
|
|
/// reduce (or otherwise transform) the range of assertions and
|
|
|
|
/// messages they can be used to send to their referent. The
|
|
|
|
/// Syndicated Actor model uses
|
|
|
|
/// [Macaroon](https://syndicate-lang.org/doc/capabilities/)-style
|
|
|
|
/// capability attenuation.
|
|
|
|
///
|
|
|
|
/// [Hardy 2017]: http://cap-lore.com/CapTheory/Patterns/Attenuation.html
|
|
|
|
/// [Miller 2006]: http://www.erights.org/talks/thesis/markm-thesis.pdf
|
2021-07-22 14:53:56 +00:00
|
|
|
#[derive(Clone, PartialEq, Eq, Hash, PartialOrd, Ord)]
|
|
|
|
pub struct Cap {
|
2021-08-14 00:12:11 +00:00
|
|
|
#[doc(hidden)]
|
2021-08-11 21:16:01 +00:00
|
|
|
pub underlying: Arc<Ref<AnyValue>>,
|
2021-08-14 00:12:11 +00:00
|
|
|
#[doc(hidden)]
|
2021-07-22 14:53:56 +00:00
|
|
|
pub attenuation: Vec<CheckedCaveat>,
|
2021-07-03 07:03:52 +00:00
|
|
|
}
|
|
|
|
|
2021-08-13 19:51:11 +00:00
|
|
|
/// Adapter for converting an underlying [`Ref<M>`] to a [`Cap`].
|
|
|
|
///
|
|
|
|
/// The [`Entity`] implementation for `Guard` decodes `AnyValue`
|
|
|
|
/// assertions/messages to type `M` before passing them on to the
|
|
|
|
/// underlying entity.
|
2021-09-19 14:53:37 +00:00
|
|
|
pub struct Guard<L, M>
|
2021-07-22 14:53:56 +00:00
|
|
|
where
|
2021-09-19 14:53:37 +00:00
|
|
|
M: for<'a> Unparse<&'a L, AnyValue>,
|
|
|
|
M: for<'a> Parse<&'a L, AnyValue>,
|
2021-07-22 14:53:56 +00:00
|
|
|
{
|
2021-09-19 14:53:37 +00:00
|
|
|
underlying: Arc<Ref<M>>,
|
|
|
|
literals: Arc<L>,
|
2021-07-03 07:03:52 +00:00
|
|
|
}
|
|
|
|
|
2021-08-28 12:39:00 +00:00
|
|
|
/// Simple entity that stops its containing facet when any assertion it receives is
|
|
|
|
/// subsequently retracted.
|
|
|
|
pub struct StopOnRetract;
|
|
|
|
|
2021-09-23 19:43:32 +00:00
|
|
|
/// Returned from the function given to [`FacetRef::activate_exit`] to indicate how the actor
|
|
|
|
/// should proceed.
|
|
|
|
pub enum RunDisposition {
|
|
|
|
Continue,
|
|
|
|
Terminate(ActorResult),
|
|
|
|
}
|
|
|
|
|
2021-07-03 07:03:52 +00:00
|
|
|
//---------------------------------------------------------------------------
|
|
|
|
|
2021-08-11 21:40:48 +00:00
|
|
|
const BUMP_AMOUNT: u8 = 10;
|
|
|
|
|
|
|
|
static NEXT_ACTOR_ID: AtomicU64 = AtomicU64::new(1);
|
2021-08-14 00:12:11 +00:00
|
|
|
#[doc(hidden)]
|
2021-08-13 01:40:48 +00:00
|
|
|
pub fn next_actor_id() -> ActorId {
|
2021-08-11 21:40:48 +00:00
|
|
|
NEXT_ACTOR_ID.fetch_add(BUMP_AMOUNT.into(), Ordering::Relaxed)
|
|
|
|
}
|
|
|
|
|
2021-08-27 13:31:18 +00:00
|
|
|
static NEXT_FACET_ID: AtomicU64 = AtomicU64::new(2);
|
|
|
|
#[doc(hidden)]
|
|
|
|
pub fn next_facet_id() -> FacetId {
|
|
|
|
FacetId::new(NEXT_FACET_ID.fetch_add(BUMP_AMOUNT.into(), Ordering::Relaxed))
|
|
|
|
.expect("Internal error: Attempt to allocate FacetId of zero. Too many FacetIds allocated. Restart the process.")
|
|
|
|
}
|
|
|
|
|
|
|
|
static NEXT_HANDLE: AtomicU64 = AtomicU64::new(3);
|
2021-08-14 00:12:11 +00:00
|
|
|
/// Allocate a process-unique `Handle`.
|
2021-08-13 01:40:48 +00:00
|
|
|
pub fn next_handle() -> Handle {
|
2021-08-11 21:40:48 +00:00
|
|
|
NEXT_HANDLE.fetch_add(BUMP_AMOUNT.into(), Ordering::Relaxed)
|
|
|
|
}
|
|
|
|
|
2021-08-27 13:31:18 +00:00
|
|
|
static NEXT_ACCOUNT_ID: AtomicU64 = AtomicU64::new(4);
|
2021-08-26 10:39:08 +00:00
|
|
|
|
2021-08-27 13:31:18 +00:00
|
|
|
static NEXT_TASK_ID: AtomicU64 = AtomicU64::new(5);
|
2021-07-15 11:13:22 +00:00
|
|
|
|
2021-09-23 19:43:32 +00:00
|
|
|
static NEXT_FIELD_ID: AtomicU64 = AtomicU64::new(6);
|
|
|
|
|
|
|
|
static NEXT_BLOCK_ID: AtomicU64 = AtomicU64::new(7);
|
|
|
|
|
2021-07-08 22:04:11 +00:00
|
|
|
preserves_schema::support::lazy_static! {
|
2021-08-14 00:12:11 +00:00
|
|
|
#[doc(hidden)]
|
2021-07-15 11:13:22 +00:00
|
|
|
pub static ref SYNDICATE_CREDIT: i64 = {
|
|
|
|
let credit =
|
|
|
|
std::env::var("SYNDICATE_CREDIT").unwrap_or("100".to_owned())
|
|
|
|
.parse::<i64>().expect("Valid SYNDICATE_CREDIT environment variable");
|
2021-08-26 08:06:05 +00:00
|
|
|
tracing::debug!("Configured SYNDICATE_CREDIT = {}", credit);
|
2021-07-15 11:13:22 +00:00
|
|
|
credit
|
|
|
|
};
|
|
|
|
|
2021-08-14 00:12:11 +00:00
|
|
|
#[doc(hidden)]
|
2021-08-13 19:51:11 +00:00
|
|
|
pub static ref ACCOUNTS: RwLock<Map<u64, (tracing::Span, Arc<AtomicI64>)>> =
|
2021-07-15 11:13:22 +00:00
|
|
|
RwLock::new(Map::new());
|
|
|
|
}
|
|
|
|
|
2021-08-11 21:16:01 +00:00
|
|
|
impl TryFrom<&AnyValue> for Synced {
|
2021-07-22 14:53:56 +00:00
|
|
|
type Error = ParseError;
|
2021-08-11 21:16:01 +00:00
|
|
|
fn try_from(value: &AnyValue) -> Result<Self, Self::Error> {
|
2021-07-22 14:53:56 +00:00
|
|
|
if let Some(true) = value.value().as_boolean() {
|
|
|
|
Ok(Synced)
|
|
|
|
} else {
|
|
|
|
Err(ParseError::conformance_error("Synced"))
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2021-08-11 21:16:01 +00:00
|
|
|
impl From<&Synced> for AnyValue {
|
2021-07-22 14:53:56 +00:00
|
|
|
fn from(_value: &Synced) -> Self {
|
2021-08-11 21:16:01 +00:00
|
|
|
AnyValue::new(true)
|
2021-07-22 14:53:56 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2021-09-19 14:53:37 +00:00
|
|
|
impl<'a> Parse<&'a (), AnyValue> for Synced {
|
|
|
|
fn parse(_language: &'a (), value: &AnyValue) -> Result<Self, ParseError> {
|
|
|
|
Synced::try_from(value)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
impl<'a> Unparse<&'a (), AnyValue> for Synced {
|
|
|
|
fn unparse(&self, _language: &'a ()) -> AnyValue {
|
|
|
|
self.into()
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2021-09-23 19:43:32 +00:00
|
|
|
impl From<ActorResult> for RunDisposition {
|
|
|
|
fn from(v: ActorResult) -> Self {
|
|
|
|
match v {
|
|
|
|
Ok(()) => RunDisposition::Continue,
|
|
|
|
Err(e) => RunDisposition::Terminate(Err(e)),
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2021-08-27 13:31:18 +00:00
|
|
|
impl FacetRef {
|
|
|
|
/// Executes `f` in a new "[turn][Activation]" for `actor`. If `f` returns `Ok(())`,
|
|
|
|
/// [commits the turn][Activation::deliver] and performs the buffered actions; otherwise,
|
|
|
|
/// [abandons the turn][Activation::clear] and discards the buffered actions.
|
2021-08-14 00:12:11 +00:00
|
|
|
///
|
|
|
|
/// Bills any activity to `account`.
|
2021-08-27 13:31:18 +00:00
|
|
|
pub fn activate<F>(
|
|
|
|
&self,
|
2021-08-13 19:51:11 +00:00
|
|
|
account: Arc<Account>,
|
2021-07-24 21:22:01 +00:00
|
|
|
f: F,
|
|
|
|
) -> ActorResult where
|
|
|
|
F: FnOnce(&mut Activation) -> ActorResult,
|
|
|
|
{
|
2021-09-23 19:43:32 +00:00
|
|
|
self.activate_exit(account, |t| f(t).into())
|
2021-07-24 21:22:01 +00:00
|
|
|
}
|
|
|
|
|
2021-08-27 13:31:18 +00:00
|
|
|
/// Executes `f` in a new "[turn][Activation]" for `actor`. If `f` returns
|
|
|
|
/// `Some(exit_status)`, terminates `actor` with that `exit_status`. Otherwise, if `f`
|
|
|
|
/// returns `None`, leaves `actor` in runnable state. [Commits buffered
|
|
|
|
/// actions][Activation::deliver] unless `actor` terminates with an `Err` status.
|
2021-08-14 00:12:11 +00:00
|
|
|
///
|
|
|
|
/// Bills any activity to `account`.
|
2021-08-27 13:31:18 +00:00
|
|
|
pub fn activate_exit<F>(
|
|
|
|
&self,
|
2021-08-13 19:51:11 +00:00
|
|
|
account: Arc<Account>,
|
2021-07-24 21:22:01 +00:00
|
|
|
f: F,
|
2021-09-23 19:43:32 +00:00
|
|
|
) -> ActorResult where
|
|
|
|
F: FnOnce(&mut Activation) -> RunDisposition,
|
2021-07-24 21:22:01 +00:00
|
|
|
{
|
2021-08-27 13:31:18 +00:00
|
|
|
match self.actor.state.lock() {
|
2021-07-24 21:22:01 +00:00
|
|
|
Err(_) => panicked_err(),
|
|
|
|
Ok(mut g) => match &mut *g {
|
|
|
|
ActorState::Terminated { exit_status } =>
|
2021-09-23 19:43:32 +00:00
|
|
|
Err(error("Could not activate terminated actor",
|
|
|
|
encode_error((**exit_status).clone()))),
|
2021-08-14 00:12:11 +00:00
|
|
|
ActorState::Running(state) => {
|
2021-08-30 10:03:46 +00:00
|
|
|
tracing::trace!(actor_id=?self.actor.actor_id, "activate");
|
2021-08-27 13:31:18 +00:00
|
|
|
let mut activation = Activation::make(self, account, state);
|
2021-09-23 19:43:32 +00:00
|
|
|
let f_result = f(&mut activation);
|
|
|
|
let result = match activation.restore_invariants(f_result) {
|
|
|
|
RunDisposition::Continue => Ok(()),
|
|
|
|
RunDisposition::Terminate(exit_status) => {
|
2021-08-14 00:12:11 +00:00
|
|
|
if exit_status.is_err() {
|
|
|
|
activation.clear();
|
|
|
|
}
|
|
|
|
drop(activation);
|
2021-07-24 21:22:01 +00:00
|
|
|
let exit_status = Arc::new(exit_status);
|
2021-08-27 13:31:18 +00:00
|
|
|
let mut t = Activation::make(&self.actor.facet_ref(state.root),
|
|
|
|
Account::new(crate::name!("shutdown")),
|
|
|
|
state);
|
2021-08-30 12:17:40 +00:00
|
|
|
if let Err(err) = t._terminate_facet(t.state.root, exit_status.is_ok()) {
|
|
|
|
// This can only occur as the result of an internal error in this file's code.
|
|
|
|
tracing::error!(?err, "unexpected error from terminate_facet");
|
|
|
|
panic!("Unexpected error result from terminate_facet");
|
|
|
|
}
|
2021-09-07 15:28:53 +00:00
|
|
|
// TODO: The linked_tasks are being cancelled above ^ when their Facets drop.
|
|
|
|
// TODO: We don't want that: we want (? do we?) exit hooks to run before linked_tasks are cancelled.
|
|
|
|
// TODO: Example: send an error message in an exit_hook that is processed and delivered by a linked_task.
|
2021-07-24 21:22:01 +00:00
|
|
|
for action in std::mem::take(&mut t.state.exit_hooks) {
|
|
|
|
if let Err(err) = action(&mut t, &exit_status) {
|
2021-08-30 10:01:47 +00:00
|
|
|
tracing::error!(?err, "error in exit hook");
|
2021-07-24 21:22:01 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
*g = ActorState::Terminated {
|
|
|
|
exit_status: Arc::clone(&exit_status),
|
|
|
|
};
|
2021-09-23 19:43:32 +00:00
|
|
|
(*exit_status).clone()
|
2021-07-24 21:22:01 +00:00
|
|
|
}
|
2021-08-30 10:03:46 +00:00
|
|
|
};
|
|
|
|
tracing::trace!(actor_id=?self.actor.actor_id, "deactivate");
|
|
|
|
result
|
2021-08-14 00:12:11 +00:00
|
|
|
}
|
2021-07-24 21:22:01 +00:00
|
|
|
}
|
2021-07-03 07:03:52 +00:00
|
|
|
}
|
|
|
|
}
|
2021-08-27 13:31:18 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
impl<'activation> Activation<'activation> {
|
|
|
|
fn make(
|
|
|
|
facet: &FacetRef,
|
|
|
|
account: Arc<Account>,
|
|
|
|
state: &'activation mut RunningActor,
|
|
|
|
) -> Self {
|
|
|
|
Activation {
|
|
|
|
facet: facet.clone(),
|
|
|
|
state,
|
2021-09-23 19:43:32 +00:00
|
|
|
active_block: None,
|
2021-08-27 13:31:18 +00:00
|
|
|
pending: EventBuffer::new(account),
|
|
|
|
}
|
|
|
|
}
|
2021-07-03 07:03:52 +00:00
|
|
|
|
2021-07-22 14:53:56 +00:00
|
|
|
fn immediate_oid<M>(&self, r: &Arc<Ref<M>>) {
|
2021-08-27 13:31:18 +00:00
|
|
|
if r.mailbox.actor_id != self.facet.actor.actor_id {
|
2021-07-24 21:22:01 +00:00
|
|
|
panic!("Cannot use for_myself to send to remote peers");
|
2021-07-12 15:41:12 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2021-08-27 13:31:18 +00:00
|
|
|
fn with_facet<F>(&mut self, check_existence: bool, facet_id: FacetId, f: F) -> ActorResult
|
|
|
|
where
|
|
|
|
F: FnOnce(&mut Activation) -> ActorResult,
|
|
|
|
{
|
|
|
|
if !check_existence || self.state.facet_nodes.contains_key(&facet_id) {
|
2021-08-30 10:03:46 +00:00
|
|
|
tracing::trace!(check_existence, facet_id, "is_alive");
|
2021-08-27 13:31:18 +00:00
|
|
|
let old_facet_id = self.facet.facet_id;
|
|
|
|
self.facet.facet_id = facet_id;
|
|
|
|
let result = f(self);
|
|
|
|
self.facet.facet_id = old_facet_id;
|
|
|
|
result
|
|
|
|
} else {
|
2021-08-30 10:03:46 +00:00
|
|
|
tracing::trace!(facet_id, "not_alive");
|
2021-08-27 13:31:18 +00:00
|
|
|
Ok(())
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
#[doc(hidden)]
|
|
|
|
pub fn with_entity<M, F>(&mut self, r: &Arc<Ref<M>>, f: F) -> ActorResult where
|
|
|
|
F: FnOnce(&mut Activation, &mut dyn Entity<M>) -> ActorResult
|
|
|
|
{
|
|
|
|
self.with_facet(true, r.facet_id, |t| r.internal_with_entity(|e| f(t, e)))
|
|
|
|
}
|
|
|
|
|
|
|
|
fn active_facet<'a>(&'a mut self) -> Option<&'a mut Facet> {
|
|
|
|
self.state.get_facet(self.facet.facet_id)
|
|
|
|
}
|
|
|
|
|
2021-08-14 00:12:11 +00:00
|
|
|
/// Core API: assert `a` at recipient `r`.
|
|
|
|
///
|
|
|
|
/// Returns the [`Handle`] for the new assertion.
|
2021-08-30 21:41:51 +00:00
|
|
|
pub fn assert<M: 'static + Send + std::fmt::Debug>(&mut self, r: &Arc<Ref<M>>, a: M) -> Handle {
|
2021-08-11 21:40:48 +00:00
|
|
|
let handle = next_handle();
|
2021-09-07 17:12:32 +00:00
|
|
|
if let Some(f) = self.state.get_facet(self.facet.facet_id) {
|
2021-08-30 21:41:51 +00:00
|
|
|
tracing::trace!(?r, ?handle, ?a, "assert");
|
2021-09-07 17:12:32 +00:00
|
|
|
f.outbound_handles.insert(handle);
|
2021-08-27 13:31:18 +00:00
|
|
|
drop(f);
|
2021-09-07 17:12:32 +00:00
|
|
|
self.state.insert_retract_cleanup_action(&r, handle);
|
2021-08-27 13:31:18 +00:00
|
|
|
{
|
|
|
|
let r = Arc::clone(r);
|
|
|
|
self.pending.queue_for(&r).push(Box::new(
|
2021-08-30 21:41:51 +00:00
|
|
|
move |t| t.with_entity(&r, |t, e| {
|
|
|
|
tracing::trace!(?handle, ?a, "asserted");
|
|
|
|
e.assert(t, a, handle)
|
|
|
|
})));
|
2021-08-27 13:31:18 +00:00
|
|
|
}
|
2021-07-15 07:13:31 +00:00
|
|
|
}
|
2021-07-12 15:41:12 +00:00
|
|
|
handle
|
|
|
|
}
|
|
|
|
|
2021-08-14 00:12:11 +00:00
|
|
|
/// Core API: assert `a` at `r`, which must be a `Ref<M>` within the active actor.
|
|
|
|
///
|
|
|
|
/// It's perfectly OK to use method [`assert`][Self::assert] even
|
|
|
|
/// for `Ref`s that are part of the active actor. The difference
|
|
|
|
/// between `assert` and `assert_for_myself` is that `r`'s handler
|
|
|
|
/// for `assert` runs in a separate, later [`Activation`], while
|
|
|
|
/// `r`'s handler for `assert_for_myself` runs in *this*
|
|
|
|
/// [`Activation`], before it commits.
|
|
|
|
///
|
|
|
|
/// Returns the [`Handle`] for the new assertion.
|
|
|
|
///
|
|
|
|
/// # Panics
|
|
|
|
///
|
|
|
|
/// Panics if `r` is not part of the active actor.
|
2021-08-30 21:41:51 +00:00
|
|
|
pub fn assert_for_myself<M: 'static + Send + std::fmt::Debug>(&mut self, r: &Arc<Ref<M>>, a: M) -> Handle {
|
2021-07-21 21:53:55 +00:00
|
|
|
self.immediate_oid(r);
|
2021-08-11 21:40:48 +00:00
|
|
|
let handle = next_handle();
|
2021-08-27 13:31:18 +00:00
|
|
|
if let Some(f) = self.active_facet() {
|
2021-08-30 21:41:51 +00:00
|
|
|
tracing::trace!(?r, ?handle, ?a, "assert_for_myself");
|
2021-09-07 17:12:32 +00:00
|
|
|
f.outbound_handles.insert(handle);
|
|
|
|
drop(f);
|
2021-08-27 13:31:18 +00:00
|
|
|
{
|
|
|
|
let r = Arc::clone(r);
|
2021-09-07 17:12:32 +00:00
|
|
|
self.state.cleanup_actions.insert(
|
2021-08-27 13:31:18 +00:00
|
|
|
handle,
|
|
|
|
CleanupAction::ForMyself(Box::new(
|
2021-08-30 21:41:51 +00:00
|
|
|
move |t| t.with_entity(&r, |t, e| {
|
|
|
|
tracing::trace!(?handle, "retracted");
|
2021-09-07 17:12:32 +00:00
|
|
|
if let Some(f) = t.active_facet() {
|
|
|
|
f.outbound_handles.remove(&handle);
|
|
|
|
}
|
2021-08-30 21:41:51 +00:00
|
|
|
e.retract(t, handle)
|
|
|
|
}))));
|
2021-08-27 13:31:18 +00:00
|
|
|
}
|
|
|
|
{
|
|
|
|
let r = Arc::clone(r);
|
|
|
|
self.pending.for_myself.push(Box::new(
|
2021-08-30 21:41:51 +00:00
|
|
|
move |t| t.with_entity(&r, |t, e| {
|
|
|
|
tracing::trace!(?handle, ?a, "asserted");
|
|
|
|
e.assert(t, a, handle)
|
|
|
|
})));
|
2021-08-27 13:31:18 +00:00
|
|
|
}
|
2021-07-15 07:13:31 +00:00
|
|
|
}
|
2021-07-03 07:03:52 +00:00
|
|
|
handle
|
|
|
|
}
|
|
|
|
|
2021-08-28 12:39:00 +00:00
|
|
|
fn half_link(&mut self, t_other: &mut Activation) {
|
|
|
|
let entity_ref = t_other.create::<AnyValue, _>(StopOnRetract);
|
|
|
|
let handle = next_handle();
|
2021-08-30 21:41:51 +00:00
|
|
|
tracing::trace!(?handle, ?entity_ref, "half_link");
|
2021-09-07 17:12:32 +00:00
|
|
|
self.state.insert_retract_cleanup_action(&entity_ref, handle);
|
|
|
|
self.active_facet().unwrap().outbound_handles.insert(handle);
|
2021-08-28 12:39:00 +00:00
|
|
|
t_other.with_entity(&entity_ref, |t, e| e.assert(t, AnyValue::new(true), handle)).unwrap();
|
|
|
|
}
|
|
|
|
|
2021-08-14 00:12:11 +00:00
|
|
|
/// Core API: retract a previously-established assertion.
|
2021-07-03 07:03:52 +00:00
|
|
|
pub fn retract(&mut self, handle: Handle) {
|
2021-09-07 17:12:32 +00:00
|
|
|
if let Some(d) = self.state.cleanup_actions.remove(&handle) {
|
|
|
|
self.pending.execute_cleanup_action(d)
|
2021-07-12 15:41:12 +00:00
|
|
|
}
|
2021-07-03 07:03:52 +00:00
|
|
|
}
|
|
|
|
|
2021-09-23 19:43:32 +00:00
|
|
|
/// Core API: assert, retract, or replace an assertion.
|
|
|
|
pub fn update<M: 'static + Send + std::fmt::Debug>(
|
|
|
|
&mut self,
|
|
|
|
handle: &mut Option<Handle>,
|
|
|
|
r: &Arc<Ref<M>>,
|
|
|
|
a: Option<M>,
|
|
|
|
) {
|
|
|
|
let saved = handle.take();
|
|
|
|
if let Some(a) = a {
|
|
|
|
*handle = Some(self.assert(r, a));
|
|
|
|
}
|
|
|
|
if let Some(h) = saved {
|
|
|
|
self.retract(h);
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2021-08-14 00:12:11 +00:00
|
|
|
/// Core API: send message `m` to recipient `r`.
|
2021-08-30 21:41:51 +00:00
|
|
|
pub fn message<M: 'static + Send + std::fmt::Debug>(&mut self, r: &Arc<Ref<M>>, m: M) {
|
|
|
|
tracing::trace!(?r, ?m, "message");
|
2021-07-22 14:53:56 +00:00
|
|
|
let r = Arc::clone(r);
|
2021-07-24 21:22:01 +00:00
|
|
|
self.pending.queue_for(&r).push(Box::new(
|
2021-08-30 21:41:51 +00:00
|
|
|
move |t| t.with_entity(&r, |t, e| {
|
|
|
|
tracing::trace!(?m, "delivered");
|
|
|
|
e.message(t, m)
|
|
|
|
})))
|
2021-07-03 07:03:52 +00:00
|
|
|
}
|
|
|
|
|
2021-08-14 00:12:11 +00:00
|
|
|
/// Core API: send message `m` to recipient `r`, which must be a
|
|
|
|
/// `Ref<M>` within the active actor.
|
|
|
|
///
|
|
|
|
/// Method `message_for_myself` is to [`message`][Self::message]
|
|
|
|
/// as [`assert_for_myself`][Self::assert_for_myself] is to
|
|
|
|
/// [`assert`][Self::assert].
|
|
|
|
///
|
|
|
|
/// # Panics
|
|
|
|
///
|
|
|
|
/// Panics if `r` is not part of the active actor.
|
2021-07-26 08:53:56 +00:00
|
|
|
pub fn message_for_myself<M: 'static + Send>(&mut self, r: &Arc<Ref<M>>, m: M) {
|
2021-07-21 21:53:55 +00:00
|
|
|
self.immediate_oid(r);
|
2021-07-22 14:53:56 +00:00
|
|
|
let r = Arc::clone(r);
|
2021-07-24 21:22:01 +00:00
|
|
|
self.pending.for_myself.push(Box::new(
|
2021-08-27 13:31:18 +00:00
|
|
|
move |t| t.with_entity(&r, |t, e| e.message(t, m))))
|
2021-07-08 22:04:11 +00:00
|
|
|
}
|
|
|
|
|
2021-08-14 00:12:11 +00:00
|
|
|
/// Core API: begins a synchronisation with `r`.
|
|
|
|
///
|
|
|
|
/// Once the synchronisation request reaches `r`'s actor, it will
|
|
|
|
/// send a response to `peer`, which acts as a continuation for
|
|
|
|
/// the synchronisation request.
|
2021-07-26 08:53:56 +00:00
|
|
|
pub fn sync<M: 'static + Send>(&mut self, r: &Arc<Ref<M>>, peer: Arc<Ref<Synced>>) {
|
2021-07-22 07:56:21 +00:00
|
|
|
let r = Arc::clone(r);
|
2021-07-24 21:22:01 +00:00
|
|
|
self.pending.queue_for(&r).push(Box::new(
|
2021-08-27 13:31:18 +00:00
|
|
|
move |t| t.with_entity(&r, |t, e| e.sync(t, peer))))
|
2021-07-06 18:56:36 +00:00
|
|
|
}
|
|
|
|
|
2021-08-30 12:17:40 +00:00
|
|
|
/// Registers the entity `r` in the list of stop actions for the active facet. If the facet
|
|
|
|
/// terminates cleanly, `r`'s [`stop`][Entity::stop] will be called.
|
|
|
|
///
|
2021-09-24 14:14:55 +00:00
|
|
|
/// **Note.** If the actor crashes, stop actions will *not* be called.
|
2021-08-30 12:17:40 +00:00
|
|
|
///
|
|
|
|
/// Use [`RunningActor::add_exit_hook`] to install a callback that will be called at the
|
|
|
|
/// end of the lifetime of the *actor* rather than the facet. (Also, exit hooks are called
|
|
|
|
/// no matter whether actor termination was normal or abnormal.)
|
2021-09-24 14:14:55 +00:00
|
|
|
pub fn on_stop_notify<M: 'static + Send>(&mut self, r: &Arc<Ref<M>>) {
|
2021-08-30 12:17:40 +00:00
|
|
|
if let Some(f) = self.active_facet() {
|
|
|
|
let r = Arc::clone(r);
|
|
|
|
f.stop_actions.push(Box::new(move |t| r.internal_with_entity(|e| e.stop(t))));
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2021-09-24 14:14:55 +00:00
|
|
|
/// Registers `action` in the list of stop actions for the active facet. If the facet
|
|
|
|
/// terminates cleanly, `action` will be called. See also notes against
|
|
|
|
/// [`on_stop_notify`][Activation::on_stop_notify].
|
|
|
|
pub fn on_stop<F: 'static + Send + FnOnce(&mut Activation) -> ActorResult>(&mut self, action: F) {
|
|
|
|
if let Some(f) = self.active_facet() {
|
|
|
|
f.stop_actions.push(Box::new(action));
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2021-08-14 00:12:11 +00:00
|
|
|
/// Retrieve the [`Account`] against which actions are recorded.
|
2021-08-13 19:51:11 +00:00
|
|
|
pub fn account(&self) -> &Arc<Account> {
|
|
|
|
&self.pending.account
|
2021-07-24 21:22:01 +00:00
|
|
|
}
|
|
|
|
|
2021-08-14 00:12:11 +00:00
|
|
|
/// Discards all pending actions in this activation.
|
|
|
|
pub fn clear(&mut self) {
|
|
|
|
self.pending.clear();
|
|
|
|
}
|
|
|
|
|
|
|
|
/// Delivers all pending actions in this activation.
|
|
|
|
///
|
|
|
|
/// This is called automatically when an `Activation` is
|
|
|
|
/// `Drop`ped.
|
|
|
|
///
|
|
|
|
/// # Panics
|
|
|
|
///
|
|
|
|
/// Panics if any pending actions "`for_myself`" (resulting from
|
|
|
|
/// [`assert_for_myself`][Self::assert_for_myself] or
|
|
|
|
/// [`message_for_myself`][Self::message_for_myself]) are
|
|
|
|
/// outstanding at the time of the call.
|
2021-07-24 21:22:01 +00:00
|
|
|
pub fn deliver(&mut self) {
|
|
|
|
self.pending.deliver();
|
|
|
|
}
|
2021-08-27 13:31:18 +00:00
|
|
|
|
|
|
|
/// Construct an entity with behaviour [`InertEntity`] within the active facet.
|
|
|
|
pub fn inert_entity<M>(&mut self) -> Arc<Ref<M>> {
|
|
|
|
self.create(InertEntity)
|
|
|
|
}
|
|
|
|
|
|
|
|
/// Construct an entity with behaviour `e` within the active facet.
|
|
|
|
pub fn create<M, E: Entity<M> + Send + 'static>(&mut self, e: E) -> Arc<Ref<M>> {
|
|
|
|
let r = self.create_inert();
|
|
|
|
r.become_entity(e);
|
|
|
|
r
|
|
|
|
}
|
|
|
|
|
|
|
|
/// Construct an entity (within the active facet) whose behaviour will be specified later
|
|
|
|
/// via [`become_entity`][Ref::become_entity].
|
|
|
|
pub fn create_inert<M>(&mut self) -> Arc<Ref<M>> {
|
|
|
|
Arc::new(Ref {
|
|
|
|
mailbox: self.state.mailbox(),
|
|
|
|
facet_id: self.facet.facet_id,
|
|
|
|
target: Mutex::new(None),
|
|
|
|
})
|
|
|
|
}
|
|
|
|
|
|
|
|
/// Start a new [linked task][crate::actor#linked-tasks] attached to the active facet. The
|
|
|
|
/// task will execute the future "`boot`" to completion unless it is cancelled first (by
|
|
|
|
/// e.g. termination of the owning facet or crashing of the owning actor). Uses `name` for
|
|
|
|
/// log messages emitted by the task.
|
|
|
|
pub fn linked_task<F: 'static + Send + futures::Future<Output = ActorResult>>(
|
|
|
|
&mut self,
|
|
|
|
name: tracing::Span,
|
|
|
|
boot: F,
|
|
|
|
) {
|
|
|
|
let mailbox = self.state.mailbox();
|
2021-08-30 21:41:51 +00:00
|
|
|
let facet = self.facet.clone();
|
2021-08-27 13:31:18 +00:00
|
|
|
if let Some(f) = self.active_facet() {
|
|
|
|
let token = CancellationToken::new();
|
2021-08-28 16:50:55 +00:00
|
|
|
let task_id = NEXT_TASK_ID.fetch_add(BUMP_AMOUNT.into(), Ordering::Relaxed);
|
2021-08-27 13:31:18 +00:00
|
|
|
name.record("task_id", &task_id);
|
|
|
|
{
|
|
|
|
let token = token.clone();
|
|
|
|
tokio::spawn(async move {
|
|
|
|
tracing::trace!(task_id, "linked task start");
|
2021-08-30 21:41:51 +00:00
|
|
|
let result = select! {
|
2021-08-27 13:31:18 +00:00
|
|
|
_ = token.cancelled() => {
|
|
|
|
tracing::trace!(task_id, "linked task cancelled");
|
|
|
|
Ok(())
|
|
|
|
}
|
|
|
|
result = boot => {
|
|
|
|
match &result {
|
|
|
|
Ok(()) => {
|
|
|
|
tracing::trace!(task_id, "linked task normal stop");
|
|
|
|
()
|
|
|
|
}
|
|
|
|
Err(e) => {
|
|
|
|
tracing::error!(task_id, "linked task error: {}", e);
|
|
|
|
let _ = mailbox.tx.send(SystemMessage::Crash(e.clone()));
|
|
|
|
()
|
|
|
|
}
|
|
|
|
}
|
|
|
|
result
|
|
|
|
}
|
2021-08-30 21:41:51 +00:00
|
|
|
};
|
|
|
|
let _ = facet.activate(
|
|
|
|
Account::new(crate::name!("release_linked_task")), |t| {
|
|
|
|
if let Some(f) = t.active_facet() {
|
|
|
|
tracing::trace!(task_id, "cancellation token removed");
|
|
|
|
f.linked_tasks.remove(&task_id);
|
|
|
|
}
|
|
|
|
Ok(())
|
|
|
|
});
|
|
|
|
result
|
2021-08-27 13:31:18 +00:00
|
|
|
}.instrument(name));
|
|
|
|
}
|
|
|
|
f.linked_tasks.insert(task_id, token);
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2021-08-30 21:41:51 +00:00
|
|
|
/// Executes the given action after the given duration has elapsed (so long as the active
|
|
|
|
/// facet still exists at that time).
|
|
|
|
pub fn after(&mut self, duration: time::Duration, a: Action) {
|
|
|
|
self.at(time::Instant::now() + duration, a)
|
|
|
|
}
|
|
|
|
|
|
|
|
/// Executes the given action at the given instant (so long as the active facet still
|
|
|
|
/// exists at that time).
|
|
|
|
pub fn at<I: Into<tokio::time::Instant>>(&mut self, instant: I, a: Action) {
|
|
|
|
let facet = self.facet.clone();
|
|
|
|
let account = Arc::clone(self.account());
|
|
|
|
let instant = instant.into();
|
|
|
|
self.linked_task(crate::name!("Activation::at"), async move {
|
|
|
|
tokio::time::sleep_until(instant.into()).await;
|
|
|
|
facet.activate(account, a)
|
|
|
|
});
|
|
|
|
}
|
|
|
|
|
2021-08-27 14:19:14 +00:00
|
|
|
fn enqueue_for_myself_at_commit(&mut self, action: Action) {
|
|
|
|
let mailbox = self.state.mailbox();
|
|
|
|
self.pending.queue_for_mailbox(&mailbox).push(action);
|
|
|
|
}
|
|
|
|
|
|
|
|
/// Schedule the creation of a new actor when the Activation commits.
|
|
|
|
pub fn spawn<F: 'static + Send + FnOnce(&mut Activation) -> ActorResult>(
|
|
|
|
&mut self,
|
|
|
|
name: tracing::Span,
|
|
|
|
boot: F,
|
2021-09-01 15:31:01 +00:00
|
|
|
) -> ActorRef {
|
|
|
|
let ac = Actor::new();
|
|
|
|
let ac_ref = ac.ac_ref.clone();
|
2021-08-27 14:19:14 +00:00
|
|
|
self.enqueue_for_myself_at_commit(Box::new(move |_| {
|
2021-09-01 15:31:01 +00:00
|
|
|
ac.boot(name, boot);
|
2021-08-27 14:19:14 +00:00
|
|
|
Ok(())
|
|
|
|
}));
|
2021-09-01 15:31:01 +00:00
|
|
|
ac_ref
|
2021-08-27 14:19:14 +00:00
|
|
|
}
|
|
|
|
|
2021-08-28 12:39:00 +00:00
|
|
|
/// Schedule the creation of a new actor when the Activation commits.
|
|
|
|
///
|
|
|
|
/// The new actor will be "linked" to the active facet: if the new actor terminates, the
|
|
|
|
/// active facet is stopped, and if the active facet stops, the new actor's root facet is
|
|
|
|
/// stopped.
|
|
|
|
pub fn spawn_link<F: 'static + Send + FnOnce(&mut Activation) -> ActorResult>(
|
|
|
|
&mut self,
|
|
|
|
name: tracing::Span,
|
|
|
|
boot: F,
|
2021-09-01 15:31:01 +00:00
|
|
|
) -> ActorRef {
|
|
|
|
let ac = Actor::new();
|
|
|
|
let ac_ref = ac.ac_ref.clone();
|
2021-08-28 12:39:00 +00:00
|
|
|
let facet_id = self.facet.facet_id;
|
|
|
|
self.enqueue_for_myself_at_commit(Box::new(move |t| {
|
|
|
|
t.with_facet(true, facet_id, move |t| {
|
2021-09-01 15:31:01 +00:00
|
|
|
ac.link(t).boot(name, boot);
|
2021-08-28 12:39:00 +00:00
|
|
|
Ok(())
|
|
|
|
})
|
|
|
|
}));
|
2021-09-01 15:31:01 +00:00
|
|
|
ac_ref
|
2021-08-28 12:39:00 +00:00
|
|
|
}
|
|
|
|
|
2021-08-27 13:31:18 +00:00
|
|
|
/// Create a new subfacet of the currently-active facet. Runs `boot` in the new facet's
|
|
|
|
/// context. If `boot` returns leaving the new facet [inert][Facet#inert-facets],
|
2021-08-28 12:39:00 +00:00
|
|
|
pub fn facet<F: FnOnce(&mut Activation) -> ActorResult>(
|
2021-08-27 13:31:18 +00:00
|
|
|
&mut self,
|
|
|
|
boot: F,
|
|
|
|
) -> Result<FacetId, Error> {
|
|
|
|
let f = Facet::new(Some(self.facet.facet_id));
|
|
|
|
let facet_id = f.facet_id;
|
|
|
|
self.state.facet_nodes.insert(facet_id, f);
|
|
|
|
self.state.facet_children.entry(self.facet.facet_id).or_default().insert(facet_id);
|
|
|
|
self.with_facet(true /* TODO: tiny optimisation: "false" would be safe here */, facet_id, move |t| {
|
|
|
|
boot(t)?;
|
|
|
|
t.stop_if_inert();
|
|
|
|
Ok(())
|
|
|
|
})?;
|
|
|
|
Ok(facet_id)
|
|
|
|
}
|
|
|
|
|
|
|
|
/// Useful during facet (and actor) startup, in some situations: when a facet `boot`
|
|
|
|
/// procedure would return while the facet is inert, but the facet should survive until
|
|
|
|
/// some subsequent time, call `prevent_inert_check` to increment a counter that prevents
|
|
|
|
/// inertness-checks from succeeding on the active facet.
|
|
|
|
///
|
|
|
|
/// The result of `prevent_inert_check` is a function which, when called, decrements the
|
|
|
|
/// counter again. After the counter has been decremented, any subsequent inertness checks
|
|
|
|
/// will no longer be artificially forced to fail.
|
|
|
|
///
|
|
|
|
/// An example of when you might want this: creating an actor having only a single
|
|
|
|
/// Dataspace entity within it, then using the Dataspace from other actors. At the start of
|
|
|
|
/// its life, the Dataspace actor will have no outbound assertions, no child facets, and no
|
|
|
|
/// linked tasks, so the only way to prevent it from being prematurely garbage collected is
|
|
|
|
/// to use `prevent_inert_check` in its boot function.
|
2021-08-30 21:41:51 +00:00
|
|
|
pub fn prevent_inert_check(&mut self) -> DisarmFn {
|
2021-08-27 13:31:18 +00:00
|
|
|
if let Some(f) = self.active_facet() {
|
|
|
|
Box::new(f.prevent_inert_check())
|
|
|
|
} else {
|
|
|
|
Box::new(|| ())
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
/// Arranges for the [`Facet`] named by `facet_id` to be stopped cleanly when `self`
|
|
|
|
/// commits. If `continuation` is supplied, the facet to be stopped hasn't been stopped
|
|
|
|
/// yet, none of the shutdown handlers yields an error, and the facet's parent facet is
|
|
|
|
/// alive, executes `continuation` in the parent facet's context.
|
|
|
|
pub fn stop_facet(&mut self, facet_id: FacetId, continuation: Option<Action>) {
|
|
|
|
let maybe_parent_id = self.active_facet().and_then(|f| f.parent_facet_id);
|
2021-08-27 14:19:14 +00:00
|
|
|
self.enqueue_for_myself_at_commit(Box::new(move |t| {
|
|
|
|
t._terminate_facet(facet_id, true)?;
|
|
|
|
if let Some(k) = continuation {
|
|
|
|
if let Some(parent_id) = maybe_parent_id {
|
|
|
|
t.with_facet(true, parent_id, k)?;
|
2021-08-27 13:31:18 +00:00
|
|
|
}
|
2021-08-27 14:19:14 +00:00
|
|
|
}
|
|
|
|
Ok(())
|
|
|
|
}));
|
2021-08-27 13:31:18 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
/// Arranges for the active facet to be stopped cleanly when `self` commits.
|
|
|
|
///
|
2021-09-23 19:43:32 +00:00
|
|
|
/// Equivalent to `self.stop_facet(self.facet.facet_id, None)`.
|
2021-08-27 13:31:18 +00:00
|
|
|
pub fn stop(&mut self) {
|
|
|
|
self.stop_facet(self.facet.facet_id, None)
|
|
|
|
}
|
|
|
|
|
|
|
|
fn stop_if_inert(&mut self) {
|
2021-08-28 12:39:00 +00:00
|
|
|
let facet_id = self.facet.facet_id;
|
|
|
|
self.enqueue_for_myself_at_commit(Box::new(move |t| {
|
2021-08-30 10:03:46 +00:00
|
|
|
tracing::trace!("Checking inertness of facet {} from facet {}", facet_id, t.facet.facet_id);
|
2021-08-28 12:39:00 +00:00
|
|
|
if t.state.facet_exists_and_is_inert(facet_id) {
|
2021-08-30 10:03:46 +00:00
|
|
|
tracing::trace!(" - facet {} is inert, stopping it", facet_id);
|
2021-08-28 12:39:00 +00:00
|
|
|
t.stop_facet(facet_id, None);
|
2021-08-30 10:03:46 +00:00
|
|
|
} else {
|
|
|
|
tracing::trace!(" - facet {} is not inert", facet_id);
|
2021-08-28 12:39:00 +00:00
|
|
|
}
|
|
|
|
Ok(())
|
|
|
|
}))
|
2021-08-27 13:31:18 +00:00
|
|
|
}
|
|
|
|
|
2021-08-30 10:03:46 +00:00
|
|
|
fn _terminate_facet(&mut self, facet_id: FacetId, alive: bool) -> ActorResult {
|
2021-08-27 13:31:18 +00:00
|
|
|
if let Some(mut f) = self.state.facet_nodes.remove(&facet_id) {
|
2021-08-30 10:03:46 +00:00
|
|
|
tracing::debug!("{} termination of {:?}",
|
|
|
|
if alive { "living" } else { "post-exit" },
|
2021-08-27 13:31:18 +00:00
|
|
|
facet_id);
|
|
|
|
if let Some(p) = f.parent_facet_id {
|
|
|
|
self.state.facet_children.get_mut(&p).map(|children| children.remove(&facet_id));
|
|
|
|
}
|
|
|
|
self.with_facet(false, facet_id, |t| {
|
|
|
|
if let Some(children) = t.state.facet_children.remove(&facet_id) {
|
|
|
|
for child_id in children.into_iter() {
|
2021-08-30 10:03:46 +00:00
|
|
|
t._terminate_facet(child_id, alive)?;
|
2021-08-27 13:31:18 +00:00
|
|
|
}
|
|
|
|
}
|
2021-08-30 10:03:46 +00:00
|
|
|
if alive {
|
2021-08-27 13:31:18 +00:00
|
|
|
for action in std::mem::take(&mut f.stop_actions) {
|
|
|
|
action(t)?;
|
|
|
|
}
|
|
|
|
let parent_facet_id = f.parent_facet_id;
|
2021-09-07 17:12:32 +00:00
|
|
|
f.retract_outbound(t);
|
|
|
|
// ^ we need retraction to happen right here so that child-facet
|
|
|
|
// cleanup-actions are performed before parent-facet cleanup-actions.
|
2021-08-27 13:31:18 +00:00
|
|
|
if let Some(p) = parent_facet_id {
|
|
|
|
if t.state.facet_exists_and_is_inert(p) {
|
|
|
|
t._terminate_facet(p, true)?;
|
|
|
|
}
|
|
|
|
} else {
|
|
|
|
t.state.shutdown();
|
|
|
|
}
|
2021-09-07 17:12:32 +00:00
|
|
|
} else {
|
|
|
|
f.retract_outbound(t);
|
2021-08-27 13:31:18 +00:00
|
|
|
}
|
|
|
|
Ok(())
|
|
|
|
})
|
|
|
|
} else {
|
|
|
|
Ok(())
|
|
|
|
}
|
|
|
|
}
|
2021-09-23 19:43:32 +00:00
|
|
|
|
|
|
|
/// Create a new dataflow variable (field) within the active [`Actor`].
|
|
|
|
pub fn field<T: Any + Send>(&mut self, initial_value: T) -> Arc<Field<T>> {
|
|
|
|
let field_id = FieldId::new(NEXT_FIELD_ID.fetch_add(BUMP_AMOUNT.into(), Ordering::Relaxed))
|
|
|
|
.expect("Internal error: Attempt to allocate FieldId of zero. Too many FieldIds allocated. Restart the process.");
|
|
|
|
self.state.fields.insert(field_id, Box::new(initial_value));
|
|
|
|
Arc::new(Field {
|
|
|
|
field_id,
|
|
|
|
tx: self.state.tx.clone(),
|
|
|
|
phantom: PhantomData,
|
|
|
|
})
|
|
|
|
}
|
|
|
|
|
|
|
|
/// Retrieve a reference to the current value of a dataflow variable (field); if execution
|
|
|
|
/// is currently within a [dataflow block][Activation::dataflow], marks the block as
|
|
|
|
/// *depending upon* the field.
|
|
|
|
///
|
|
|
|
pub fn get<T: Any + Send>(&mut self, field: &Field<T>) -> &T {
|
|
|
|
tracing::trace!(field = ?field.field_id, block = ?self.active_block, "get");
|
|
|
|
if let Some(block) = self.active_block {
|
|
|
|
self.state.dataflow.record_observation(block, field.field_id);
|
|
|
|
}
|
|
|
|
let any = self.state.fields.get(&field.field_id)
|
|
|
|
.expect("Attempt to get() missing field: wrong actor?");
|
|
|
|
any.downcast_ref().expect("Attempt to access field at incorrect type")
|
|
|
|
}
|
|
|
|
|
|
|
|
/// Retrieve a mutable reference to the contents of a dataflow variable (field). As for
|
|
|
|
/// [`get`][Activation::get], if used within a dataflow block, marks the block as
|
|
|
|
/// *depending upon* the field. In addition, because the caller may mutate the field, this
|
|
|
|
/// function (pessimistically) marks the field as dirty, which will lead to later
|
|
|
|
/// reevaluation of dependent blocks.
|
|
|
|
///
|
|
|
|
pub fn get_mut<T: Any + Send>(&mut self, field: &Field<T>) -> &mut T {
|
|
|
|
tracing::trace!(field = ?field.field_id, block = ?self.active_block, "get_mut");
|
|
|
|
{
|
|
|
|
// Overapproximation.
|
|
|
|
if let Some(block) = self.active_block {
|
|
|
|
self.state.dataflow.record_observation(block, field.field_id);
|
|
|
|
}
|
|
|
|
self.state.dataflow.record_damage(field.field_id);
|
|
|
|
}
|
|
|
|
let any = self.state.fields.get_mut(&field.field_id)
|
|
|
|
.expect("Attempt to get_mut() missing field: wrong actor?");
|
|
|
|
any.downcast_mut().expect("Attempt to access field at incorrect type")
|
|
|
|
}
|
|
|
|
|
|
|
|
/// Overwrite the value of a dataflow variable (field). Marks the field as dirty, even if
|
|
|
|
/// the new value is [`eq`][std::cmp::PartialEq::eq] to the value being overwritten.
|
|
|
|
///
|
|
|
|
pub fn set<T: Any + Send>(&mut self, field: &Field<T>, value: T) {
|
|
|
|
tracing::trace!(field = ?field.field_id, block = ?self.active_block, "set");
|
|
|
|
// Overapproximation in many cases, since the new value may not produce an
|
|
|
|
// observable difference (may be equal to the current value).
|
|
|
|
self.state.dataflow.record_damage(field.field_id);
|
|
|
|
let any = self.state.fields.get_mut(&field.field_id)
|
|
|
|
.expect("Attempt to set() missing field: wrong actor?");
|
|
|
|
*any = Box::new(value);
|
|
|
|
}
|
|
|
|
|
|
|
|
fn with_block(&mut self, block_id: BlockId, block: &mut Block) -> ActorResult {
|
|
|
|
let saved = self.active_block.replace(block_id);
|
|
|
|
let result = block(self);
|
|
|
|
self.active_block = saved;
|
|
|
|
result
|
|
|
|
}
|
|
|
|
|
|
|
|
/// Creates (and immediately executes) a new *dataflow block* that will be reexecuted if
|
|
|
|
/// any of its *dependent fields* (accessed via e.g. [`get`][Activation::get] or
|
|
|
|
/// [`get_mut`][Activation::get_mut]) are mutated.
|
|
|
|
///
|
|
|
|
pub fn dataflow<F: 'static + Send + FnMut(&mut Activation) -> ActorResult>(&mut self, block: F) -> ActorResult {
|
|
|
|
let block_id = BlockId::new(NEXT_BLOCK_ID.fetch_add(BUMP_AMOUNT.into(), Ordering::Relaxed))
|
|
|
|
.expect("Internal error: Attempt to allocate BlockId of zero. Too many BlockIds allocated. Restart the process.");
|
|
|
|
let mut block: Block = Box::new(block);
|
|
|
|
self.with_block(block_id, &mut block)?;
|
|
|
|
self.state.blocks.insert(block_id, (self.facet.facet_id, block));
|
|
|
|
Ok(())
|
|
|
|
}
|
|
|
|
|
|
|
|
fn repair_dataflow(&mut self) -> Result<bool, Error> {
|
|
|
|
let mut pass_number = 0;
|
|
|
|
while !self.state.dataflow.is_clean() {
|
|
|
|
pass_number += 1;
|
|
|
|
tracing::trace!(?pass_number, "repair_dataflow");
|
|
|
|
let damaged_field_ids = self.state.dataflow.take_damaged_nodes();
|
|
|
|
for field_id in damaged_field_ids.into_iter() {
|
|
|
|
let block_ids = self.state.dataflow.take_observers_of(&field_id);
|
|
|
|
for block_id in block_ids.into_iter() {
|
|
|
|
if let Some((facet_id, mut block)) = self.state.blocks.remove(&block_id) {
|
|
|
|
let result = self.with_facet(
|
|
|
|
true, facet_id, |t| t.with_block(block_id, &mut block));
|
|
|
|
self.state.blocks.insert(block_id, (facet_id, block));
|
|
|
|
result?;
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
tracing::trace!(passes = ?pass_number, "repair_dataflow complete");
|
|
|
|
Ok(pass_number > 0)
|
|
|
|
}
|
|
|
|
|
|
|
|
fn _restore_invariants(&mut self) -> ActorResult {
|
|
|
|
loop {
|
|
|
|
loop {
|
|
|
|
let actions = std::mem::take(&mut self.pending.for_myself);
|
|
|
|
if actions.is_empty() { break; }
|
|
|
|
for action in actions.into_iter() { action(self)? }
|
|
|
|
}
|
|
|
|
if !self.repair_dataflow()? {
|
|
|
|
break;
|
|
|
|
}
|
|
|
|
}
|
|
|
|
Ok(())
|
|
|
|
}
|
|
|
|
|
|
|
|
fn restore_invariants(&mut self, d: RunDisposition) -> RunDisposition {
|
|
|
|
match d {
|
|
|
|
RunDisposition::Continue =>
|
|
|
|
self._restore_invariants().into(),
|
|
|
|
RunDisposition::Terminate(Ok(())) =>
|
|
|
|
RunDisposition::Terminate(self._restore_invariants()),
|
|
|
|
RunDisposition::Terminate(Err(_)) =>
|
|
|
|
d,
|
|
|
|
}
|
|
|
|
}
|
2021-07-24 21:22:01 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
impl EventBuffer {
|
2021-08-13 19:51:11 +00:00
|
|
|
fn new(account: Arc<Account>) -> Self {
|
2021-07-24 21:22:01 +00:00
|
|
|
EventBuffer {
|
2021-08-13 19:51:11 +00:00
|
|
|
account,
|
2021-07-24 21:22:01 +00:00
|
|
|
queues: HashMap::new(),
|
|
|
|
for_myself: Vec::new(),
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
fn execute_cleanup_action(&mut self, d: CleanupAction) {
|
|
|
|
match d {
|
|
|
|
CleanupAction::ForAnother(mailbox, action) =>
|
|
|
|
self.queue_for_mailbox(&mailbox).push(action),
|
|
|
|
CleanupAction::ForMyself(action) =>
|
|
|
|
self.for_myself.push(action),
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2021-07-22 14:53:56 +00:00
|
|
|
fn queue_for<M>(&mut self, r: &Arc<Ref<M>>) -> &mut PendingEventQueue {
|
|
|
|
self.queue_for_mailbox(&r.mailbox)
|
|
|
|
}
|
|
|
|
|
|
|
|
fn queue_for_mailbox(&mut self, mailbox: &Arc<Mailbox>) -> &mut PendingEventQueue {
|
|
|
|
&mut self.queues.entry(mailbox.actor_id)
|
|
|
|
.or_insert((mailbox.tx.clone(), Vec::new())).1
|
2021-07-03 07:03:52 +00:00
|
|
|
}
|
|
|
|
|
2021-08-14 00:12:11 +00:00
|
|
|
fn clear(&mut self) {
|
|
|
|
self.queues = HashMap::new();
|
|
|
|
self.for_myself = PendingEventQueue::new();
|
|
|
|
}
|
|
|
|
|
2021-07-06 18:56:36 +00:00
|
|
|
fn deliver(&mut self) {
|
2021-08-30 21:41:51 +00:00
|
|
|
tracing::trace!("EventBuffer::deliver");
|
2021-07-24 21:22:01 +00:00
|
|
|
if !self.for_myself.is_empty() {
|
|
|
|
panic!("Unprocessed for_myself events remain at deliver() time");
|
2021-07-12 15:41:12 +00:00
|
|
|
}
|
2021-07-22 07:56:21 +00:00
|
|
|
for (_actor_id, (tx, turn)) in std::mem::take(&mut self.queues).into_iter() {
|
2021-08-30 21:49:08 +00:00
|
|
|
// Deliberately ignore send errors here: they indicate that the recipient is no
|
|
|
|
// longer alive. But we don't care about that case, since we have to be robust to
|
|
|
|
// crashes anyway. (When we were printing errors we saw here, an example of the
|
|
|
|
// problems it caused was a relay output_loop that received EPIPE causing the relay
|
|
|
|
// to crash, just as it was receiving thousands of messages a second, leading to
|
|
|
|
// many, many log reports of failed send_actions from the following line.)
|
|
|
|
let _ = send_actions(&tx, &self.account, turn);
|
2021-07-03 07:03:52 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2021-07-24 21:22:01 +00:00
|
|
|
impl Drop for EventBuffer {
|
2021-07-03 07:03:52 +00:00
|
|
|
fn drop(&mut self) {
|
|
|
|
self.deliver()
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2021-08-13 19:51:11 +00:00
|
|
|
impl Account {
|
2021-08-14 00:12:11 +00:00
|
|
|
/// Construct a new `Account`, storing `name` within it for
|
|
|
|
/// debugging use.
|
2021-07-15 11:13:22 +00:00
|
|
|
pub fn new(name: tracing::Span) -> Arc<Self> {
|
2021-08-13 19:51:11 +00:00
|
|
|
let id = NEXT_ACCOUNT_ID.fetch_add(1, Ordering::Relaxed);
|
2021-07-15 11:13:22 +00:00
|
|
|
let debt = Arc::new(AtomicI64::new(0));
|
2021-08-13 19:51:11 +00:00
|
|
|
ACCOUNTS.write().unwrap().insert(id, (name, Arc::clone(&debt)));
|
|
|
|
Arc::new(Account {
|
2021-07-15 11:13:22 +00:00
|
|
|
id,
|
|
|
|
debt,
|
2021-07-27 14:30:42 +00:00
|
|
|
notify: Notify::new(),
|
2021-07-15 11:13:22 +00:00
|
|
|
})
|
|
|
|
}
|
|
|
|
|
2021-08-14 00:12:11 +00:00
|
|
|
/// Retrieve the current account balance: the number of
|
|
|
|
/// currently-outstanding work items.
|
2021-07-15 11:13:22 +00:00
|
|
|
pub fn balance(&self) -> i64 {
|
|
|
|
self.debt.load(Ordering::Relaxed)
|
|
|
|
}
|
|
|
|
|
2021-08-14 00:12:11 +00:00
|
|
|
/// Borrow `token_count` work items against this account.
|
2021-07-15 11:13:22 +00:00
|
|
|
pub fn borrow(&self, token_count: usize) {
|
|
|
|
let token_count: i64 = token_count.try_into().expect("manageable token count");
|
|
|
|
self.debt.fetch_add(token_count, Ordering::Relaxed);
|
|
|
|
}
|
|
|
|
|
2021-08-14 00:12:11 +00:00
|
|
|
/// Repay `token_count` work items previously borrowed against this account.
|
2021-07-15 11:13:22 +00:00
|
|
|
pub fn repay(&self, token_count: usize) {
|
|
|
|
let token_count: i64 = token_count.try_into().expect("manageable token count");
|
|
|
|
let _old_debt = self.debt.fetch_sub(token_count, Ordering::Relaxed);
|
2021-07-27 14:30:42 +00:00
|
|
|
if _old_debt - token_count <= *SYNDICATE_CREDIT {
|
|
|
|
self.notify.notify_one();
|
|
|
|
}
|
2021-07-15 11:13:22 +00:00
|
|
|
}
|
|
|
|
|
2021-08-14 00:12:11 +00:00
|
|
|
/// Suspend execution until enough "clear funds" exist in this
|
|
|
|
/// account for some subsequent activity to be permissible.
|
2021-07-15 11:13:22 +00:00
|
|
|
pub async fn ensure_clear_funds(&self) {
|
|
|
|
let limit = *SYNDICATE_CREDIT;
|
2021-07-27 14:30:42 +00:00
|
|
|
// tokio::task::yield_now().await;
|
2021-07-15 11:13:22 +00:00
|
|
|
while self.balance() > limit {
|
2021-07-27 14:30:42 +00:00
|
|
|
// tokio::task::yield_now().await;
|
|
|
|
self.notify.notified().await;
|
2021-07-15 07:13:31 +00:00
|
|
|
}
|
|
|
|
}
|
2021-07-15 11:13:22 +00:00
|
|
|
}
|
2021-07-15 07:13:31 +00:00
|
|
|
|
2021-08-13 19:51:11 +00:00
|
|
|
impl Drop for Account {
|
2021-07-15 11:13:22 +00:00
|
|
|
fn drop(&mut self) {
|
2021-08-13 19:51:11 +00:00
|
|
|
ACCOUNTS.write().unwrap().remove(&self.id);
|
2021-07-15 11:13:22 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
impl<T> LoanedItem<T> {
|
2021-08-14 00:12:11 +00:00
|
|
|
/// Construct a new `LoanedItem<T>` containing `item`, recording
|
|
|
|
/// `cost` work items against `account`.
|
2021-08-13 19:51:11 +00:00
|
|
|
pub fn new(account: &Arc<Account>, cost: usize, item: T) -> Self {
|
|
|
|
account.borrow(cost);
|
|
|
|
LoanedItem { account: Arc::clone(account), cost, item }
|
2021-07-15 11:13:22 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
impl<T> Drop for LoanedItem<T> {
|
|
|
|
fn drop(&mut self) {
|
2021-08-13 19:51:11 +00:00
|
|
|
self.account.repay(self.cost);
|
2021-07-15 11:13:22 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2021-07-22 07:56:21 +00:00
|
|
|
#[must_use]
|
|
|
|
fn send_actions(
|
|
|
|
tx: &UnboundedSender<SystemMessage>,
|
2021-08-13 19:51:11 +00:00
|
|
|
account: &Arc<Account>,
|
2021-07-22 07:56:21 +00:00
|
|
|
t: PendingEventQueue,
|
|
|
|
) -> ActorResult {
|
|
|
|
let token_count = t.len();
|
2021-08-13 19:51:11 +00:00
|
|
|
tx.send(SystemMessage::Turn(LoanedItem::new(account, token_count, t)))
|
2021-08-11 21:16:01 +00:00
|
|
|
.map_err(|_| error("Target actor not running", AnyValue::new(false)))
|
2021-07-03 07:03:52 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
impl std::fmt::Debug for Mailbox {
|
|
|
|
fn fmt(&self, f: &mut std::fmt::Formatter) -> Result<(), std::fmt::Error> {
|
2021-07-22 14:53:56 +00:00
|
|
|
write!(f, "#<Mailbox {}>", self.actor_id)
|
2021-07-03 07:03:52 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
impl std::hash::Hash for Mailbox {
|
|
|
|
fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
|
2021-07-22 14:53:56 +00:00
|
|
|
self.actor_id.hash(state)
|
2021-07-03 07:03:52 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
impl Eq for Mailbox {}
|
|
|
|
impl PartialEq for Mailbox {
|
|
|
|
fn eq(&self, other: &Mailbox) -> bool {
|
2021-07-22 14:53:56 +00:00
|
|
|
self.actor_id == other.actor_id
|
2021-07-03 07:03:52 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
impl Ord for Mailbox {
|
|
|
|
fn cmp(&self, other: &Mailbox) -> std::cmp::Ordering {
|
2021-07-22 14:53:56 +00:00
|
|
|
return self.actor_id.cmp(&other.actor_id)
|
2021-07-03 07:03:52 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
impl PartialOrd for Mailbox {
|
|
|
|
fn partial_cmp(&self, other: &Mailbox) -> Option<std::cmp::Ordering> {
|
|
|
|
return Some(self.cmp(&other))
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
impl Drop for Mailbox {
|
|
|
|
fn drop(&mut self) {
|
2021-07-22 14:53:56 +00:00
|
|
|
let _ = self.tx.send(SystemMessage::Release);
|
|
|
|
()
|
2021-07-03 07:03:52 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
impl Actor {
|
2021-08-27 13:31:18 +00:00
|
|
|
/// Create a new actor. It still needs to be [`boot`ed][Self::boot].
|
2021-07-03 07:03:52 +00:00
|
|
|
pub fn new() -> Self {
|
|
|
|
let (tx, rx) = unbounded_channel();
|
2021-08-11 21:40:48 +00:00
|
|
|
let actor_id = next_actor_id();
|
2021-08-27 13:31:18 +00:00
|
|
|
let root = Facet::new(None);
|
2021-07-08 22:04:11 +00:00
|
|
|
// tracing::trace!(id = actor_id, "Actor::new");
|
2021-08-27 13:31:18 +00:00
|
|
|
let mut st = RunningActor {
|
|
|
|
actor_id,
|
|
|
|
tx,
|
|
|
|
mailbox: Weak::new(),
|
2021-09-23 19:43:32 +00:00
|
|
|
dataflow: Graph::new(),
|
|
|
|
fields: HashMap::new(),
|
|
|
|
blocks: HashMap::new(),
|
2021-08-27 13:31:18 +00:00
|
|
|
exit_hooks: Vec::new(),
|
2021-09-07 17:12:32 +00:00
|
|
|
cleanup_actions: Map::new(),
|
2021-08-27 13:31:18 +00:00
|
|
|
facet_nodes: Map::new(),
|
|
|
|
facet_children: Map::new(),
|
|
|
|
root: root.facet_id,
|
|
|
|
};
|
|
|
|
st.facet_nodes.insert(root.facet_id, root);
|
2021-07-03 07:03:52 +00:00
|
|
|
Actor {
|
2021-07-23 06:10:09 +00:00
|
|
|
rx,
|
2021-07-24 21:22:01 +00:00
|
|
|
ac_ref: ActorRef {
|
|
|
|
actor_id,
|
2021-08-27 13:31:18 +00:00
|
|
|
state: Arc::new(Mutex::new(ActorState::Running(st))),
|
2021-07-24 21:22:01 +00:00
|
|
|
},
|
2021-07-03 07:03:52 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2021-08-28 12:39:00 +00:00
|
|
|
fn link(self, t_parent: &mut Activation) -> Self {
|
|
|
|
if t_parent.active_facet().is_none() {
|
|
|
|
panic!("No active facet when calling spawn_link");
|
|
|
|
}
|
|
|
|
self.ac_ref.root_facet_ref().activate(Account::new(crate::name!("link")), |t_child| {
|
|
|
|
t_parent.half_link(t_child);
|
|
|
|
t_child.half_link(t_parent);
|
|
|
|
Ok(())
|
|
|
|
}).expect("Failed during link");
|
|
|
|
self
|
|
|
|
}
|
|
|
|
|
2021-08-14 00:12:11 +00:00
|
|
|
/// Start the actor's mainloop. Takes ownership of `self`. The
|
|
|
|
/// `name` is used as context for any log messages emitted by the
|
|
|
|
/// actor. The `boot` function is called in the actor's context,
|
|
|
|
/// and then the mainloop is entered.
|
2021-07-24 21:22:01 +00:00
|
|
|
pub fn boot<F: 'static + Send + FnOnce(&mut Activation) -> ActorResult>(
|
|
|
|
mut self,
|
|
|
|
name: tracing::Span,
|
|
|
|
boot: F,
|
|
|
|
) -> ActorHandle {
|
|
|
|
name.record("actor_id", &self.ac_ref.actor_id);
|
|
|
|
tokio::spawn(async move {
|
|
|
|
tracing::trace!("start");
|
2021-08-27 13:31:18 +00:00
|
|
|
self.run(|t| {
|
2021-08-27 21:38:51 +00:00
|
|
|
t.facet(boot)?;
|
2021-08-27 13:31:18 +00:00
|
|
|
Ok(())
|
|
|
|
}).await;
|
2021-07-24 21:22:01 +00:00
|
|
|
let result = self.ac_ref.exit_status().expect("terminated");
|
|
|
|
match &result {
|
|
|
|
Ok(()) => tracing::trace!("normal stop"),
|
|
|
|
Err(e) => tracing::error!("error stop: {}", e),
|
|
|
|
}
|
|
|
|
result
|
|
|
|
}.instrument(name))
|
|
|
|
}
|
|
|
|
|
|
|
|
async fn run<F: 'static + Send + FnOnce(&mut Activation) -> ActorResult>(
|
|
|
|
&mut self,
|
|
|
|
boot: F,
|
|
|
|
) -> () {
|
2021-08-28 12:39:00 +00:00
|
|
|
let root_facet_ref = self.ac_ref.root_facet_ref();
|
2021-08-27 13:31:18 +00:00
|
|
|
|
|
|
|
let terminate = |result: ActorResult| {
|
|
|
|
let _ = root_facet_ref.activate_exit(Account::new(crate::name!("shutdown")),
|
2021-09-23 19:43:32 +00:00
|
|
|
|_| RunDisposition::Terminate(result));
|
2021-08-27 13:31:18 +00:00
|
|
|
};
|
|
|
|
|
|
|
|
if root_facet_ref.activate(Account::new(crate::name!("boot")), boot).is_err() {
|
2021-07-24 21:22:01 +00:00
|
|
|
return;
|
|
|
|
}
|
|
|
|
|
|
|
|
loop {
|
2021-08-30 10:03:46 +00:00
|
|
|
tracing::trace!(actor_id = ?self.ac_ref.actor_id, "mainloop top");
|
2021-07-24 21:22:01 +00:00
|
|
|
match self.rx.recv().await {
|
|
|
|
None => {
|
2021-08-27 13:31:18 +00:00
|
|
|
return terminate(Err(error("Unexpected channel close", AnyValue::new(false))));
|
2021-07-24 21:22:01 +00:00
|
|
|
}
|
|
|
|
Some(m) => match m {
|
|
|
|
SystemMessage::Release => {
|
2021-08-30 10:03:46 +00:00
|
|
|
tracing::trace!(actor_id = ?self.ac_ref.actor_id, "SystemMessage::Release");
|
2021-08-27 13:31:18 +00:00
|
|
|
return terminate(Ok(()));
|
2021-07-24 21:22:01 +00:00
|
|
|
}
|
2021-09-23 19:43:32 +00:00
|
|
|
SystemMessage::ReleaseField(field_id) => {
|
|
|
|
tracing::trace!(actor_id = ?self.ac_ref.actor_id,
|
|
|
|
"SystemMessage::ReleaseField({})", field_id);
|
|
|
|
self.ac_ref.access(|s| if let ActorState::Running(ra) = s.unwrap() {
|
|
|
|
ra.fields.remove(&field_id);
|
|
|
|
})
|
|
|
|
}
|
2021-07-24 21:22:01 +00:00
|
|
|
SystemMessage::Turn(mut loaned_item) => {
|
2021-08-30 10:03:46 +00:00
|
|
|
tracing::trace!(actor_id = ?self.ac_ref.actor_id, "SystemMessage::Turn");
|
2021-09-23 19:43:32 +00:00
|
|
|
let actions = std::mem::take(&mut loaned_item.item);
|
2021-08-27 13:31:18 +00:00
|
|
|
let r = root_facet_ref.activate(Arc::clone(&loaned_item.account), |t| {
|
2021-09-23 19:43:32 +00:00
|
|
|
for action in actions.into_iter() { action(t)? }
|
2021-08-27 13:31:18 +00:00
|
|
|
Ok(())
|
|
|
|
});
|
2021-07-24 21:22:01 +00:00
|
|
|
if r.is_err() { return; }
|
|
|
|
}
|
|
|
|
SystemMessage::Crash(e) => {
|
2021-08-30 21:41:51 +00:00
|
|
|
tracing::trace!(actor_id = ?self.ac_ref.actor_id,
|
|
|
|
"SystemMessage::Crash({:?})", &e);
|
2021-08-27 13:31:18 +00:00
|
|
|
return terminate(Err(e));
|
2021-07-24 21:22:01 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2021-08-27 13:31:18 +00:00
|
|
|
impl Facet {
|
|
|
|
fn new(parent_facet_id: Option<FacetId>) -> Self {
|
|
|
|
Facet {
|
|
|
|
facet_id: next_facet_id(),
|
|
|
|
parent_facet_id,
|
2021-09-07 17:12:32 +00:00
|
|
|
outbound_handles: Set::new(),
|
2021-08-27 13:31:18 +00:00
|
|
|
stop_actions: Vec::new(),
|
|
|
|
linked_tasks: Map::new(),
|
|
|
|
inert_check_preventers: Arc::new(AtomicU64::new(0)),
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
fn prevent_inert_check(&mut self) -> impl FnOnce() {
|
|
|
|
let inert_check_preventers = Arc::clone(&self.inert_check_preventers);
|
|
|
|
let armed = AtomicU64::new(1);
|
|
|
|
inert_check_preventers.fetch_add(1, Ordering::Relaxed);
|
|
|
|
move || {
|
|
|
|
match armed.compare_exchange(1, 0, Ordering::SeqCst, Ordering::SeqCst) {
|
|
|
|
Ok(_) => {
|
|
|
|
inert_check_preventers.fetch_sub(1, Ordering::Relaxed);
|
|
|
|
()
|
|
|
|
}
|
|
|
|
Err(_) => (),
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
2021-08-28 12:39:00 +00:00
|
|
|
|
2021-09-07 17:12:32 +00:00
|
|
|
fn retract_outbound(&mut self, t: &mut Activation) {
|
|
|
|
for handle in std::mem::take(&mut self.outbound_handles).into_iter() {
|
|
|
|
tracing::trace!(h = ?handle, "retract on termination");
|
|
|
|
t.retract(handle);
|
|
|
|
}
|
2021-08-28 12:39:00 +00:00
|
|
|
}
|
2021-08-27 13:31:18 +00:00
|
|
|
}
|
|
|
|
|
2021-09-23 19:43:32 +00:00
|
|
|
fn panicked_err() -> ActorResult {
|
|
|
|
Err(error("Actor panicked", AnyValue::new(false)))
|
2021-07-24 21:22:01 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
impl ActorRef {
|
2021-08-14 00:12:11 +00:00
|
|
|
/// Uses an internal mutex to access the internal state: takes the
|
|
|
|
/// lock, calls `f` with the internal state, releases the lock,
|
|
|
|
/// and returns the result of `f`.
|
2021-07-25 21:12:07 +00:00
|
|
|
pub fn access<R, F: FnOnce(Option<&mut ActorState>) -> R>(&self, f: F) -> R {
|
|
|
|
match self.state.lock() {
|
2021-07-24 21:22:01 +00:00
|
|
|
Err(_) => f(None),
|
|
|
|
Ok(mut g) => f(Some(&mut *g)),
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2021-08-14 00:12:11 +00:00
|
|
|
/// Retrieves the exit status of the denoted actor. If it is still
|
|
|
|
/// running, yields `None`; otherwise, yields `Some(Ok(()))` if it
|
|
|
|
/// exited normally, or `Some(Err(_))` if it terminated
|
|
|
|
/// abnormally.
|
2021-07-24 21:22:01 +00:00
|
|
|
pub fn exit_status(&self) -> Option<ActorResult> {
|
2021-07-25 21:12:07 +00:00
|
|
|
self.access(|s| s.map_or_else(
|
2021-09-23 19:43:32 +00:00
|
|
|
|| Some(panicked_err()),
|
2021-07-24 21:22:01 +00:00
|
|
|
|state| match state {
|
|
|
|
ActorState::Running(_) => None,
|
|
|
|
ActorState::Terminated { exit_status } => Some((**exit_status).clone()),
|
|
|
|
}))
|
|
|
|
}
|
|
|
|
|
2021-08-27 13:31:18 +00:00
|
|
|
fn facet_ref(&self, facet_id: FacetId) -> FacetRef {
|
|
|
|
FacetRef {
|
|
|
|
actor: self.clone(),
|
|
|
|
facet_id,
|
2021-07-24 21:22:01 +00:00
|
|
|
}
|
|
|
|
}
|
2021-08-28 12:39:00 +00:00
|
|
|
|
|
|
|
fn root_facet_id(&self) -> FacetId {
|
|
|
|
self.access(|s| match s.expect("Actor missing its state") {
|
|
|
|
ActorState::Terminated { .. } => panic!("Actor unexpectedly in terminated state"),
|
|
|
|
ActorState::Running(ra) => ra.root, // what a lot of work to get this one number
|
|
|
|
})
|
|
|
|
}
|
|
|
|
|
|
|
|
fn root_facet_ref(&self) -> FacetRef {
|
|
|
|
self.facet_ref(self.root_facet_id())
|
|
|
|
}
|
2021-07-24 21:22:01 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
impl RunningActor {
|
2021-08-14 00:12:11 +00:00
|
|
|
/// Requests a shutdown of the actor. The shutdown request is
|
|
|
|
/// handled by the actor's main loop, causing it to terminate with
|
|
|
|
/// exit status `Ok(())`.
|
2021-07-24 21:22:01 +00:00
|
|
|
pub fn shutdown(&self) {
|
|
|
|
let _ = self.tx.send(SystemMessage::Release);
|
2021-07-06 18:56:36 +00:00
|
|
|
}
|
|
|
|
|
2021-07-22 14:53:56 +00:00
|
|
|
fn mailbox(&mut self) -> Arc<Mailbox> {
|
|
|
|
match self.mailbox.upgrade() {
|
|
|
|
None => {
|
|
|
|
let new_mailbox = Arc::new(Mailbox {
|
|
|
|
actor_id: self.actor_id,
|
|
|
|
tx: self.tx.clone(),
|
|
|
|
});
|
|
|
|
self.mailbox = Arc::downgrade(&new_mailbox);
|
|
|
|
new_mailbox
|
|
|
|
}
|
|
|
|
Some(m) => m
|
|
|
|
}
|
2021-07-06 18:56:36 +00:00
|
|
|
}
|
|
|
|
|
2021-08-14 00:12:11 +00:00
|
|
|
/// Registers the entity `r` in the list of exit hooks for this
|
|
|
|
/// actor. When the actor terminates, `r`'s
|
|
|
|
/// [`exit_hook`][Entity::exit_hook] will be called.
|
2021-07-26 08:53:56 +00:00
|
|
|
pub fn add_exit_hook<M: 'static + Send>(&mut self, r: &Arc<Ref<M>>) {
|
2021-07-22 14:53:56 +00:00
|
|
|
let r = Arc::clone(r);
|
2021-08-27 13:31:18 +00:00
|
|
|
self.exit_hooks.push(Box::new(
|
|
|
|
move |t, exit_status| r.internal_with_entity(|e| e.exit_hook(t, &exit_status))))
|
2021-07-06 18:56:36 +00:00
|
|
|
}
|
|
|
|
|
2021-08-27 13:31:18 +00:00
|
|
|
fn get_facet(&mut self, facet_id: FacetId) -> Option<&mut Facet> {
|
|
|
|
self.facet_nodes.get_mut(&facet_id)
|
|
|
|
}
|
|
|
|
|
|
|
|
/// See the definition of an [inert facet][Facet#inert-facets].
|
|
|
|
fn facet_exists_and_is_inert(&mut self, facet_id: FacetId) -> bool {
|
|
|
|
let no_kids = self.facet_children.get(&facet_id).map(|cs| cs.is_empty()).unwrap_or(true);
|
|
|
|
if let Some(f) = self.get_facet(facet_id) {
|
|
|
|
no_kids &&
|
2021-09-07 17:12:32 +00:00
|
|
|
f.outbound_handles.is_empty() &&
|
2021-08-27 13:31:18 +00:00
|
|
|
f.linked_tasks.is_empty() &&
|
|
|
|
f.inert_check_preventers.load(Ordering::Relaxed) == 0
|
|
|
|
} else {
|
|
|
|
false
|
2021-07-08 22:04:11 +00:00
|
|
|
}
|
2021-07-03 07:03:52 +00:00
|
|
|
}
|
2021-09-07 17:12:32 +00:00
|
|
|
|
|
|
|
fn insert_retract_cleanup_action<M: 'static + Send>(
|
|
|
|
&mut self,
|
|
|
|
r: &Arc<Ref<M>>,
|
|
|
|
handle: Handle,
|
|
|
|
) {
|
|
|
|
let r = Arc::clone(r);
|
|
|
|
self.cleanup_actions.insert(
|
|
|
|
handle,
|
|
|
|
CleanupAction::ForAnother(Arc::clone(&r.mailbox), Box::new(
|
|
|
|
move |t| t.with_entity(&r, |t, e| {
|
|
|
|
tracing::trace!(?handle, "retracted");
|
|
|
|
if let Some(f) = t.active_facet() {
|
|
|
|
f.outbound_handles.remove(&handle);
|
|
|
|
}
|
|
|
|
e.retract(t, handle)
|
|
|
|
}))));
|
|
|
|
}
|
2021-07-03 07:03:52 +00:00
|
|
|
}
|
|
|
|
|
2021-09-23 19:43:32 +00:00
|
|
|
impl<T: Any + Send> Eq for Field<T> {}
|
|
|
|
impl<T: Any + Send> PartialEq for Field<T> {
|
|
|
|
fn eq(&self, other: &Field<T>) -> bool {
|
|
|
|
self.field_id == other.field_id
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
impl<T: Any + Send> Drop for Field<T> {
|
|
|
|
fn drop(&mut self) {
|
|
|
|
let _ = self.tx.send(SystemMessage::ReleaseField(self.field_id));
|
|
|
|
()
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2021-07-03 07:03:52 +00:00
|
|
|
impl Drop for Actor {
|
|
|
|
fn drop(&mut self) {
|
2021-07-23 06:10:09 +00:00
|
|
|
self.rx.close();
|
2021-08-30 10:03:46 +00:00
|
|
|
tracing::trace!(actor_id = ?self.ac_ref.actor_id, "Actor::drop");
|
2021-07-24 21:22:01 +00:00
|
|
|
}
|
|
|
|
}
|
2021-07-15 07:13:31 +00:00
|
|
|
|
2021-08-27 13:31:18 +00:00
|
|
|
impl Drop for Facet {
|
2021-07-24 21:22:01 +00:00
|
|
|
fn drop(&mut self) {
|
2021-07-03 07:03:52 +00:00
|
|
|
for (_task_id, token) in std::mem::take(&mut self.linked_tasks).into_iter() {
|
|
|
|
token.cancel();
|
|
|
|
}
|
|
|
|
|
2021-09-07 17:12:32 +00:00
|
|
|
if !self.outbound_handles.is_empty() {
|
|
|
|
panic!("Internal error: outbound_handles not empty at drop time");
|
2021-07-03 07:03:52 +00:00
|
|
|
}
|
2021-07-15 07:13:31 +00:00
|
|
|
|
2021-08-30 10:01:47 +00:00
|
|
|
tracing::trace!(facet_id = ?self.facet_id, "Facet::drop");
|
2021-07-03 07:03:52 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2021-08-14 00:12:11 +00:00
|
|
|
/// Directly injects `action` into `mailbox`, billing subsequent activity against `account`.
|
|
|
|
///
|
2021-08-27 13:31:18 +00:00
|
|
|
/// Primarily for use by [linked tasks][Activation::linked_task].
|
2021-07-21 21:53:55 +00:00
|
|
|
#[must_use]
|
2021-08-13 19:51:11 +00:00
|
|
|
pub fn external_event(mailbox: &Arc<Mailbox>, account: &Arc<Account>, action: Action) -> ActorResult {
|
|
|
|
send_actions(&mailbox.tx, account, vec![action])
|
2021-07-21 21:53:55 +00:00
|
|
|
}
|
2021-07-15 07:13:31 +00:00
|
|
|
|
2021-08-14 00:12:11 +00:00
|
|
|
/// Directly injects `actions` into `mailbox`, billing subsequent activity against `account`.
|
|
|
|
///
|
2021-08-27 13:31:18 +00:00
|
|
|
/// Primarily for use by [linked tasks][Activation::linked_task].
|
2021-07-21 21:53:55 +00:00
|
|
|
#[must_use]
|
2021-08-14 00:12:11 +00:00
|
|
|
pub fn external_events(mailbox: &Arc<Mailbox>, account: &Arc<Account>, actions: PendingEventQueue) -> ActorResult {
|
|
|
|
send_actions(&mailbox.tx, account, actions)
|
2021-07-22 14:53:56 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
impl<M> Ref<M> {
|
2021-08-14 00:12:11 +00:00
|
|
|
/// Supplies the behaviour (`e`) for a `Ref` created via
|
2021-08-27 13:31:18 +00:00
|
|
|
/// [`create_inert`][Activation::create_inert].
|
2021-08-14 00:12:11 +00:00
|
|
|
///
|
|
|
|
/// # Panics
|
|
|
|
///
|
|
|
|
/// Panics if this `Ref` has already been given a behaviour.
|
2021-07-22 14:53:56 +00:00
|
|
|
pub fn become_entity<E: 'static + Entity<M>>(&self, e: E) {
|
2021-07-25 21:12:07 +00:00
|
|
|
let mut g = self.target.lock().expect("unpoisoned");
|
2021-07-22 14:53:56 +00:00
|
|
|
if g.is_some() {
|
|
|
|
panic!("Double initialization of Ref");
|
|
|
|
}
|
|
|
|
*g = Some(Box::new(e));
|
|
|
|
}
|
|
|
|
|
2021-08-27 13:31:18 +00:00
|
|
|
fn internal_with_entity<R, F: FnOnce(&mut dyn Entity<M>) -> R>(&self, f: F) -> R {
|
2021-07-25 21:12:07 +00:00
|
|
|
let mut g = self.target.lock().expect("unpoisoned");
|
2021-07-22 14:53:56 +00:00
|
|
|
f(g.as_mut().expect("initialized").as_mut())
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
impl<M> Ref<M> {
|
2021-08-14 00:12:11 +00:00
|
|
|
/// Retrieves a process-unique identifier for the `Ref`; `Ref`s
|
|
|
|
/// are compared by this identifier.
|
2021-07-22 14:53:56 +00:00
|
|
|
pub fn oid(&self) -> usize {
|
|
|
|
std::ptr::addr_of!(*self) as usize
|
|
|
|
}
|
2021-07-21 21:53:55 +00:00
|
|
|
}
|
2021-07-15 07:13:31 +00:00
|
|
|
|
2021-07-22 14:53:56 +00:00
|
|
|
impl<M> PartialEq for Ref<M> {
|
|
|
|
fn eq(&self, other: &Self) -> bool {
|
|
|
|
self.oid() == other.oid()
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
impl<M> Eq for Ref<M> {}
|
|
|
|
|
|
|
|
impl<M> std::hash::Hash for Ref<M> {
|
|
|
|
fn hash<H>(&self, hash: &mut H) where H: std::hash::Hasher {
|
|
|
|
self.oid().hash(hash)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
impl<M> PartialOrd for Ref<M> {
|
|
|
|
fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
|
|
|
|
Some(self.cmp(other))
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
impl<M> Ord for Ref<M> {
|
|
|
|
fn cmp(&self, other: &Self) -> std::cmp::Ordering {
|
|
|
|
self.oid().cmp(&other.oid())
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
impl<M> std::fmt::Debug for Ref<M> {
|
|
|
|
fn fmt(&self, f: &mut std::fmt::Formatter) -> Result<(), std::fmt::Error> {
|
2021-08-30 10:03:46 +00:00
|
|
|
write!(f, "⌜{}/{}:{:016x}⌝", self.mailbox.actor_id, self.facet_id, self.oid())
|
2021-07-22 14:53:56 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
impl Cap {
|
2021-08-14 00:12:11 +00:00
|
|
|
/// Given a `Ref<M>`, where `M` is interconvertible with
|
|
|
|
/// `AnyValue`, yields a `Cap` for the referenced entity. The
|
|
|
|
/// `Cap` automatically decodes presented `AnyValue`s into
|
|
|
|
/// instances of `M`.
|
2021-09-19 14:53:37 +00:00
|
|
|
pub fn guard<L: 'static + Sync + Send, M: 'static + Send>(
|
|
|
|
literals: Arc<L>,
|
|
|
|
underlying: Arc<Ref<M>>,
|
|
|
|
) -> Arc<Self>
|
2021-07-22 14:53:56 +00:00
|
|
|
where
|
2021-09-19 14:53:37 +00:00
|
|
|
M: for<'a> Unparse<&'a L, AnyValue>,
|
|
|
|
M: for<'a> Parse<&'a L, AnyValue>,
|
2021-07-22 14:53:56 +00:00
|
|
|
{
|
|
|
|
Self::new(&Arc::new(Ref {
|
|
|
|
mailbox: Arc::clone(&underlying.mailbox),
|
2021-08-27 13:31:18 +00:00
|
|
|
facet_id: underlying.facet_id,
|
2021-09-19 14:53:37 +00:00
|
|
|
target: Mutex::new(Some(Box::new(Guard { underlying: underlying, literals }))),
|
2021-07-22 14:53:56 +00:00
|
|
|
}))
|
|
|
|
}
|
|
|
|
|
2021-08-14 00:12:11 +00:00
|
|
|
/// Directly constructs a `Cap` for `underlying`.
|
2021-08-11 21:16:01 +00:00
|
|
|
pub fn new(underlying: &Arc<Ref<AnyValue>>) -> Arc<Self> {
|
2021-07-22 14:53:56 +00:00
|
|
|
Arc::new(Cap {
|
|
|
|
underlying: Arc::clone(underlying),
|
|
|
|
attenuation: Vec::new(),
|
|
|
|
})
|
2021-07-22 07:56:21 +00:00
|
|
|
}
|
|
|
|
|
2021-08-14 00:12:11 +00:00
|
|
|
/// Yields a fresh `Cap` for `self`'s `underlying`, copying the
|
|
|
|
/// existing attenuation of `self` to the new `Cap` and adding
|
|
|
|
/// `attenuation` to it.
|
2021-07-15 07:13:31 +00:00
|
|
|
pub fn attenuate(&self, attenuation: &sturdy::Attenuation) -> Result<Arc<Self>, CaveatError> {
|
2021-07-22 14:53:56 +00:00
|
|
|
let mut r = Cap { attenuation: self.attenuation.clone(), .. self.clone() };
|
2021-07-15 07:13:31 +00:00
|
|
|
r.attenuation.extend(attenuation.check()?);
|
|
|
|
Ok(Arc::new(r))
|
|
|
|
}
|
|
|
|
|
2021-08-14 00:12:11 +00:00
|
|
|
/// Applies the contained attenuation to `a`, yielding `None` if
|
|
|
|
/// `a` is filtered out, or `Some(_)` if it is accepted (and
|
|
|
|
/// possibly transformed).
|
2021-08-11 21:16:01 +00:00
|
|
|
pub fn rewrite(&self, mut a: AnyValue) -> Option<AnyValue> {
|
2021-07-15 07:13:31 +00:00
|
|
|
for c in &self.attenuation {
|
|
|
|
match c.rewrite(&a) {
|
|
|
|
Some(v) => a = v,
|
|
|
|
None => return None,
|
|
|
|
}
|
|
|
|
}
|
2021-07-22 07:56:21 +00:00
|
|
|
Some(a)
|
2021-07-06 18:56:36 +00:00
|
|
|
}
|
2021-07-22 14:53:56 +00:00
|
|
|
|
2021-08-14 00:12:11 +00:00
|
|
|
/// Translates `m` into an `AnyValue`, passes it through
|
|
|
|
/// [`rewrite`][Self::rewrite], and then
|
|
|
|
/// [`assert`s][Activation::assert] it using the activation `t`.
|
2021-09-19 14:53:37 +00:00
|
|
|
pub fn assert<L, M: Unparse<L, AnyValue>>(&self, t: &mut Activation, literals: L, m: &M) -> Option<Handle>
|
|
|
|
{
|
|
|
|
self.rewrite(m.unparse(literals)).map(|m| t.assert(&self.underlying, m))
|
2021-07-22 14:53:56 +00:00
|
|
|
}
|
|
|
|
|
2021-09-23 19:43:32 +00:00
|
|
|
/// `update` is to [`assert`] as [`Activation::update`] is to [`Activation::assert`].
|
|
|
|
pub fn update<L, M: Unparse<L, AnyValue>>(
|
|
|
|
&self,
|
|
|
|
t: &mut Activation,
|
|
|
|
handle: &mut Option<Handle>,
|
|
|
|
literals: L,
|
|
|
|
m: Option<&M>,
|
|
|
|
) {
|
|
|
|
t.update(handle, &self.underlying, m.and_then(|m| self.rewrite(m.unparse(literals))))
|
|
|
|
}
|
|
|
|
|
2021-08-14 00:12:11 +00:00
|
|
|
/// Translates `m` into an `AnyValue`, passes it through
|
|
|
|
/// [`rewrite`][Self::rewrite], and then sends it via method
|
|
|
|
/// [`message`][Activation::message] on the activation `t`.
|
2021-09-19 14:53:37 +00:00
|
|
|
pub fn message<L, M: Unparse<L, AnyValue>>(&self, t: &mut Activation, literals: L, m: &M)
|
|
|
|
{
|
|
|
|
if let Some(m) = self.rewrite(m.unparse(literals)) {
|
2021-07-22 14:53:56 +00:00
|
|
|
t.message(&self.underlying, m)
|
|
|
|
}
|
|
|
|
}
|
2021-07-06 18:56:36 +00:00
|
|
|
}
|
|
|
|
|
2021-07-22 14:53:56 +00:00
|
|
|
impl std::fmt::Debug for Cap {
|
2021-07-08 22:04:11 +00:00
|
|
|
fn fmt(&self, f: &mut std::fmt::Formatter) -> Result<(), std::fmt::Error> {
|
2021-07-15 07:13:31 +00:00
|
|
|
if self.attenuation.is_empty() {
|
2021-07-22 14:53:56 +00:00
|
|
|
self.underlying.fmt(f)
|
2021-07-15 07:13:31 +00:00
|
|
|
} else {
|
2021-08-30 10:03:46 +00:00
|
|
|
write!(f, "⌜{}/{}:{:016x}\\{:?}⌝",
|
2021-07-22 14:53:56 +00:00
|
|
|
self.underlying.mailbox.actor_id,
|
2021-08-30 10:03:46 +00:00
|
|
|
self.underlying.facet_id,
|
2021-07-22 14:53:56 +00:00
|
|
|
self.underlying.oid(),
|
|
|
|
self.attenuation)
|
2021-07-15 07:13:31 +00:00
|
|
|
}
|
2021-07-08 22:04:11 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2021-07-22 14:53:56 +00:00
|
|
|
impl Domain for Cap {}
|
2021-07-06 18:56:36 +00:00
|
|
|
|
2021-07-22 14:53:56 +00:00
|
|
|
impl std::convert::TryFrom<&IOValue> for Cap {
|
2021-07-06 18:56:36 +00:00
|
|
|
type Error = preserves_schema::support::ParseError;
|
|
|
|
fn try_from(_v: &IOValue) -> Result<Self, Self::Error> {
|
2021-07-22 14:53:56 +00:00
|
|
|
panic!("Attempted to serialize Cap via IOValue");
|
2021-07-03 07:03:52 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2021-07-22 14:53:56 +00:00
|
|
|
impl std::convert::From<&Cap> for IOValue {
|
|
|
|
fn from(_v: &Cap) -> IOValue {
|
2021-07-06 18:56:36 +00:00
|
|
|
panic!("Attempted to deserialize Ref via IOValue");
|
2021-07-03 07:03:52 +00:00
|
|
|
}
|
|
|
|
}
|
2021-07-08 22:04:11 +00:00
|
|
|
|
2021-09-19 14:53:37 +00:00
|
|
|
impl<L: Sync + Send, M> Entity<AnyValue> for Guard<L, M>
|
2021-07-22 14:53:56 +00:00
|
|
|
where
|
2021-09-19 14:53:37 +00:00
|
|
|
M: for<'a> Unparse<&'a L, AnyValue>,
|
|
|
|
M: for<'a> Parse<&'a L, AnyValue>,
|
2021-07-22 14:53:56 +00:00
|
|
|
{
|
2021-08-11 21:16:01 +00:00
|
|
|
fn assert(&mut self, t: &mut Activation, a: AnyValue, h: Handle) -> ActorResult {
|
2021-09-19 14:53:37 +00:00
|
|
|
match M::parse(&*self.literals, &a) {
|
2021-08-27 13:31:18 +00:00
|
|
|
Ok(a) => t.with_entity(&self.underlying, |t, e| e.assert(t, a, h)),
|
2021-07-22 14:53:56 +00:00
|
|
|
Err(_) => Ok(()),
|
|
|
|
}
|
|
|
|
}
|
|
|
|
fn retract(&mut self, t: &mut Activation, h: Handle) -> ActorResult {
|
2021-08-27 13:31:18 +00:00
|
|
|
t.with_entity(&self.underlying, |t, e| e.retract(t, h))
|
2021-07-22 14:53:56 +00:00
|
|
|
}
|
2021-08-11 21:16:01 +00:00
|
|
|
fn message(&mut self, t: &mut Activation, m: AnyValue) -> ActorResult {
|
2021-09-19 14:53:37 +00:00
|
|
|
match M::parse(&*self.literals, &m) {
|
2021-08-27 13:31:18 +00:00
|
|
|
Ok(m) => t.with_entity(&self.underlying, |t, e| e.message(t, m)),
|
2021-07-22 14:53:56 +00:00
|
|
|
Err(_) => Ok(()),
|
|
|
|
}
|
|
|
|
}
|
|
|
|
fn sync(&mut self, t: &mut Activation, peer: Arc<Ref<Synced>>) -> ActorResult {
|
2021-08-27 13:31:18 +00:00
|
|
|
t.with_entity(&self.underlying, |t, e| e.sync(t, peer))
|
2021-07-22 14:53:56 +00:00
|
|
|
}
|
2021-08-30 12:17:40 +00:00
|
|
|
fn stop(&mut self, t: &mut Activation) -> ActorResult {
|
|
|
|
t.with_entity(&self.underlying, |t, e| e.stop(t))
|
|
|
|
}
|
2021-07-22 14:53:56 +00:00
|
|
|
fn exit_hook(&mut self, t: &mut Activation, exit_status: &Arc<ActorResult>) -> ActorResult {
|
2021-08-27 13:31:18 +00:00
|
|
|
self.underlying.internal_with_entity(|e| e.exit_hook(t, exit_status))
|
2021-07-22 14:53:56 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2021-08-28 12:39:00 +00:00
|
|
|
impl<M> Entity<M> for StopOnRetract {
|
|
|
|
fn retract(&mut self, t: &mut Activation, _h: Handle) -> ActorResult {
|
|
|
|
t.stop();
|
|
|
|
Ok(())
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2021-09-24 14:14:55 +00:00
|
|
|
impl<F: Send + FnMut(&mut Activation) -> ActorResult> Entity<Synced> for F {
|
|
|
|
fn message(&mut self, t: &mut Activation, _m: Synced) -> ActorResult {
|
|
|
|
self(t)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2021-08-11 22:03:50 +00:00
|
|
|
/// A convenient Syndicate-enhanced variation on
|
|
|
|
/// [`tracing::info_span`].
|
|
|
|
///
|
|
|
|
/// Includes fields `actor_id`, `task_id` and `oid`, so that they show
|
|
|
|
/// up in those circumstances where they happen to be defined as part
|
|
|
|
/// of the operation of the [`crate::actor`] module.
|
2021-07-08 22:04:11 +00:00
|
|
|
#[macro_export]
|
|
|
|
macro_rules! name {
|
|
|
|
() => {tracing::info_span!(actor_id = tracing::field::Empty,
|
|
|
|
task_id = tracing::field::Empty,
|
|
|
|
oid = tracing::field::Empty)};
|
|
|
|
($($item:tt)*) => {tracing::info_span!($($item)*,
|
|
|
|
actor_id = tracing::field::Empty,
|
|
|
|
task_id = tracing::field::Empty,
|
|
|
|
oid = tracing::field::Empty)}
|
|
|
|
}
|
2021-09-23 19:43:32 +00:00
|
|
|
|
|
|
|
/// A convenient way of cloning a bunch of state shared among [entities][Entity], actions,
|
|
|
|
/// linked tasks, etc.
|
|
|
|
///
|
|
|
|
/// Directly drawn from the discussion [here](https://github.com/rust-lang/rfcs/issues/2407).
|
|
|
|
///
|
|
|
|
#[macro_export]
|
|
|
|
macro_rules! enclose {
|
|
|
|
( ( $($name:ident),* ) $closure:expr ) => {
|
|
|
|
{
|
|
|
|
$(let $name = $name.clone();)*
|
|
|
|
$closure
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|