//!
//! Supervisors enable users to supervise a subtree of children
//! or other supervisor trees under themselves.
use crate::broadcast::{Broadcast, Parent, Sender};
use crate::callbacks::Callbacks;
use crate::children::Children;
use crate::children_ref::ChildrenRef;
use crate::context::{BastionId, ContextState};
use crate::envelope::Envelope;
use crate::message::{BastionMessage, Deployment, Message};
use crate::path::{BastionPath, BastionPathElement};
use bastion_executor::pool;
use futures::prelude::*;
use futures::stream::FuturesOrdered;
use futures::{pending, poll};
use futures_timer::Delay;
use fxhash::FxHashMap;
use lightproc::prelude::*;
use std::cmp::{Eq, PartialEq};
use std::ops::Range;
use std::pin::Pin;
use std::sync::Arc;
use std::task::Poll;
use std::time::Duration;
use tracing::{debug, trace, warn};
#[derive(Debug)]
/// A supervisor that can supervise both [`Children`] and other
/// supervisors using a defined [`SupervisionStrategy`] (set
/// with [`with_strategy`] or [`SupervisionStrategy::OneForOne`]
/// by default).
///
/// When a supervised children group or supervisor faults, the
/// supervisor will restart it and eventually some of its other
/// supervised entities, depending on its supervision strategy.
///
/// Note that a supervisor, called the "system supervisor", is
/// created by the system at startup and is the supervisor
/// supervising children groups created via [`Bastion::children`].
///
/// # Example
///
/// ```rust
/// # use bastion::prelude::*;
/// #
/// # Bastion::init();
/// #
/// let sp_ref: SupervisorRef = Bastion::supervisor(|sp| {
/// // Configure the supervisor...
/// sp.with_strategy(SupervisionStrategy::OneForOne)
/// // ...and return it.
/// }).expect("Couldn't create the supervisor.");
/// #
/// # Bastion::start();
/// # Bastion::stop();
/// # Bastion::block_until_stopped();
/// ```
///
/// [`Children`]: children/struct.Children.html
/// [`SupervisionStrategy`]: supervisor/enum.SupervisionStrategy.html
/// [`with_strategy`]: #method.with_strategy
/// [`Bastion::children`]: struct.Bastion.html#method.children
pub struct Supervisor {
bcast: Broadcast,
// The order in which children and supervisors were added.
// It is only updated when at least one of those is resat.
order: Vec<BastionId>,
// Special wrapper around launched Children and Child that helps
// to figure out what and how to restart childs when it necessary.
// The key is the Children's BastionId and the values are
// represented as Child instances with the same executed closure.
tracked_groups: FxHashMap<BastionId, Vec<TrackedChildState>>,
// Hold the insertion order of the childs.
tracked_groups_order: FxHashMap<BastionId, usize>,
// The currently launched supervised children and supervisors.
// The last value is the amount of times a given actor has restarted.
launched: FxHashMap<BastionId, (usize, RecoverableHandle<Supervised>)>,
// Supervised children and supervisors that are stopped.
// This is used when resetting or recovering when the
// supervision strategy is not "one-for-one".
stopped: FxHashMap<BastionId, Supervised>,
// Supervised children and supervisors that were killed.
// This is used when resetting only.
killed: FxHashMap<BastionId, Supervised>,
strategy: SupervisionStrategy,
restart_strategy: RestartStrategy,
// The callbacks called at the supervisor's different
// lifecycle events.
callbacks: Callbacks,
// Whether this supervisor was started by the system (in
// which case, users shouldn't be able to get a reference
// to it).
is_system_supervisor: bool,
// Messages that were received before the supervisor was
// started. Those will be "replayed" once a start message
// is received.
pre_start_msgs: Vec<Envelope>,
started: bool,
// Stores amount of subtree restarts.
subtree_restarts: usize,
// Store the maximum acceptable restarts for the supervisor.
subtree_restarts_limit: usize,
}
#[derive(Debug, Clone)]
struct TrackedChildState {
id: BastionId,
state: Arc<Pin<Box<ContextState>>>,
restarts_counts: usize,
}
#[derive(Debug)]
enum RestartedElement {
Supervisor(BastionId),
Child { id: BastionId, parent_id: BastionId },
}
#[derive(Debug)]
enum ActorSearchMethod {
OneActor { id: BastionId, parent_id: BastionId },
FromActor { id: BastionId, parent_id: BastionId },
All,
}
#[derive(Debug, Clone)]
/// A "reference" to a [`Supervisor`], allowing to
/// communicate with it.
///
/// [`Supervisor`]: supervisor/struct.Supervisor.html
pub struct SupervisorRef {
id: BastionId,
sender: Sender,
path: Arc<BastionPath>,
}
#[derive(Debug, Clone)]
/// The strategy a supervisor should use when one of its
/// supervised children groups or supervisors dies (in
/// the case of a children group, it could be because one
/// of its elements panicked or returned an error).
///
/// The default strategy is `OneForOne`.
pub enum SupervisionStrategy {
/// When a children group dies (either because it got
/// killed, it panicked or returned an error), only
/// this group is restarted.
OneForOne,
/// When a children group dies (either because it got
/// killed, it panicked or returned an error), all the
/// children groups are restarted (even those which were
/// stopped) in the same order they were added to the
/// supervisor.
OneForAll,
/// When a children group dies (either because it got
/// killed, it panicked or returned an error), this
/// group and all the ones that were added to the
/// supervisor after it are restarted (even those which
/// were stopped) in the same order they were added to
/// the supervisor.
RestForOne,
}
#[derive(Debug)]
enum Supervised {
Supervisor(Supervisor),
Children(Children),
}
#[derive(Debug, Clone, Eq, PartialEq)]
/// The restart policy which is used during restoring failed
/// actors by the supervisor.
///
/// The default restart policy is `Always`.
pub enum RestartPolicy {
/// Restart the failed actor with unlimited amount of attempts.
Always,
/// Never restart the failed actor when it happens.
Never,
/// Restart the failed actor with the limited amount of attempts.
/// If the actor can't be run after N attempts, the failed actor
/// will be removed from the execution by the supervisor.
Tries(usize),
}
/// The strategy for a supervisor which is used for
/// restoring failed actors. It it fails after N attempts,
/// the supervisor will remove an actor.
///
/// The default strategy used is `ActorRestartStrategy::Immediate`
/// with the `RestartPolicy::Always` restart policy.
#[derive(Debug, Clone, PartialEq)]
pub struct RestartStrategy {
restart_policy: RestartPolicy,
strategy: ActorRestartStrategy,
}
#[derive(Debug, Clone, PartialEq)]
/// The strategy for restating an actor as far as it
/// returned an failure.
///
/// The default strategy is `Immediate`.
pub enum ActorRestartStrategy {
/// Restart an actor as soon as possible, since the moment
/// the actor finished with a failure.
Immediate,
/// Restart an actor after with the timeout. Each next restart
/// is increasing on the given duration.
LinearBackOff {
/// An initial delay before the restarting an actor.
timeout: Duration,
},
/// Restart an actor after with the timeout. Each next timeout
/// is increasing exponentially.
/// When passed a multiplier that equals to 1, the strategy works as the
/// linear back off strategy. Passing the multiplier that equals to 0 leads
/// to constant restart delays which is equal to the given timeout.
ExponentialBackOff {
/// An initial delay before the restarting an actor.
timeout: Duration,
/// Defines a multiplier how fast the timeout will be increasing.
multiplier: f64,
},
}
impl ActorRestartStrategy {
/// Calculate the expected restart delay for given strategy after n restarts.
pub fn calculate(&self, restarts_count: usize) -> Option<Duration> {
match *self {
ActorRestartStrategy::LinearBackOff { timeout } => {
let delay = timeout.mul_f64(restarts_count as f64);
Some(timeout + delay)
}
ActorRestartStrategy::ExponentialBackOff {
timeout,
multiplier,
} => {
let factor = multiplier * restarts_count as f64;
let delay = timeout.mul_f64(factor);
Some(timeout + delay)
}
_ => None,
}
}
}
impl Supervisor {
pub(crate) fn new(bcast: Broadcast) -> Self {
debug!("Supervisor({}): Initializing.", bcast.id());
let order = Vec::new();
let tracked_groups = FxHashMap::default();
let tracked_groups_order = FxHashMap::default();
let launched = FxHashMap::default();
let stopped = FxHashMap::default();
let killed = FxHashMap::default();
let strategy = SupervisionStrategy::default();
let restart_strategy = RestartStrategy::default();
let callbacks = Callbacks::new();
let is_system_supervisor = false;
let pre_start_msgs = Vec::new();
let started = false;
let subtree_restarts = 0;
let subtree_restarts_limit = 3;
Supervisor {
bcast,
order,
tracked_groups,
tracked_groups_order,
launched,
stopped,
killed,
strategy,
restart_strategy,
callbacks,
is_system_supervisor,
pre_start_msgs,
started,
subtree_restarts,
subtree_restarts_limit,
}
}
pub(crate) fn system(bcast: Broadcast) -> Self {
let mut supervisor = Supervisor::new(bcast);
supervisor.is_system_supervisor = true;
supervisor
}
fn stack(&self) -> ProcStack {
trace!("Supervisor({}): Creating ProcStack.", self.id());
// FIXME: with_pid
ProcStack::default()
}
pub(crate) async fn reset(&mut self, bcast: Option<Broadcast>) {
if let Some(bcast) = &bcast {
debug!(
"Supervisor({}): Resetting to Supervisor({}).",
self.id(),
bcast.id()
);
} else {
debug!(
"Supervisor({}): Resetting to Supervisor({}).",
self.id(),
self.id()
);
}
// TODO: stop or kill?
self.kill(0..self.order.len()).await;
if let Some(bcast) = bcast {
self.bcast = bcast;
} else {
self.bcast.clear_children();
}
debug!(
"Supervisor({}): Removing {} pre-start messages.",
self.id(),
self.pre_start_msgs.len()
);
self.pre_start_msgs.clear();
self.pre_start_msgs.shrink_to_fit();
let restarted_objects = self.search_restarted_objects(ActorSearchMethod::All);
self.restart(restarted_objects).await;
debug!(
"Supervisor({}): Removing {} stopped elements.",
self.id(),
self.stopped.len()
);
// TODO: should be empty
self.stopped.clear();
self.stopped.shrink_to_fit();
debug!(
"Supervisor({}): Removing {} killed elements.",
self.id(),
self.killed.len()
);
// TODO: should be empty
self.killed.clear();
self.killed.shrink_to_fit();
}
/// Returns this supervisor's identifier.
///
/// Note that the supervisor's identifier is reset when it is
/// restarted.
///
/// # Example
///
/// ```rust
/// # use bastion::prelude::*;
/// #
/// # Bastion::init();
/// #
/// Bastion::supervisor(|sp| {
/// let supervisor_id: &BastionId = sp.id();
/// // ...
/// # sp
/// }).expect("Couldn't create the supervisor.");
///
/// #
/// # Bastion::start();
/// # Bastion::stop();
/// # Bastion::block_until_stopped();
/// ```
pub fn id(&self) -> &BastionId {
&self.bcast.id()
}
pub(crate) fn bcast(&self) -> &Broadcast {
&self.bcast
}
pub(crate) fn callbacks(&self) -> &Callbacks {
&self.callbacks
}
pub(crate) fn as_ref(&self) -> SupervisorRef {
trace!(
"Supervisor({}): Creating new SupervisorRef({}).",
self.id(),
self.id()
);
// TODO: clone or ref?
let id = self.bcast.id().clone();
let sender = self.bcast.sender().clone();
let path = self.bcast.path().clone();
SupervisorRef::new(id, sender, path)
}
/// Creates a new supervisor, passes it through the specified
/// `init` closure and then starts supervising it.
///
/// If you don't need to chain calls to this `Supervisor`'s methods
/// and need to get a [`SupervisorRef`] referencing the newly
/// created supervisor, use the [`supervisor_ref`] method instead.
///
/// # Arguments
///
/// * `init` - The closure taking the new supervisor as an
/// argument and returning it once configured.
///
/// # Example
///
/// ```rust
/// # use bastion::prelude::*;
/// #
/// # Bastion::init();
/// #
/// # Bastion::supervisor(|parent| {
/// parent.supervisor(|sp| {
/// // Configure the supervisor...
/// sp.with_strategy(SupervisionStrategy::OneForOne)
/// // ...and return it.
/// })
/// # }).unwrap();
/// #
/// # Bastion::start();
/// # Bastion::stop();
/// # Bastion::block_until_stopped();
/// ```
///
/// [`SupervisorRef`]: ../struct.SupervisorRef.html
/// [`supervisor_ref`]: #method.supervisor_ref
pub fn supervisor<S>(self, init: S) -> Self
where
S: FnOnce(Supervisor) -> Supervisor,
{
debug!("Supervisor({}): Creating supervisor.", self.id());
let parent = Parent::supervisor(self.as_ref());
let bcast = Broadcast::new(parent, BastionPathElement::Supervisor(BastionId::new()));
debug!(
"Supervisor({}): Initializing Supervisor({}).",
self.id(),
bcast.id()
);
let supervisor = Supervisor::new(bcast);
let supervisor = init(supervisor);
debug!("Supervisor({}): Initialized.", supervisor.id());
debug!(
"Supervisor({}): Deploying Supervisor({}).",
self.id(),
supervisor.id()
);
let msg = BastionMessage::deploy_supervisor(supervisor);
let env = Envelope::new(msg, self.bcast.path().clone(), self.bcast.sender().clone());
self.bcast.send_self(env);
self
}
/// Creates a new `Supervisor`, passes it through the specified
/// `init` closure and then starts supervising it.
///
/// If you need to chain calls to this `Supervisor`'s methods and
/// don't need to get a [`SupervisorRef`] referencing the newly
/// created supervisor, use the [`supervisor`] method instead.
///
/// # Arguments
///
/// * `init` - The closure taking the new `Supervisor` as an
/// argument and returning it once configured.
///
/// # Example
///
/// ```rust
/// # use bastion::prelude::*;
/// #
/// # Bastion::init();
/// #
/// # Bastion::supervisor(|mut parent| {
/// let sp_ref: SupervisorRef = parent.supervisor_ref(|sp| {
/// // Configure the supervisor...
/// sp.with_strategy(SupervisionStrategy::OneForOne)
/// // ...and return it.
/// });
/// # parent
/// # }).unwrap();
/// #
/// # Bastion::start();
/// # Bastion::stop();
/// # Bastion::block_until_stopped();
/// ```
///
/// [`SupervisorRef`]: ../struct.SupervisorRef.html
/// [`supervisor`]: #method.supervisor
pub fn supervisor_ref<S>(&mut self, init: S) -> SupervisorRef
where
S: FnOnce(Supervisor) -> Supervisor,
{
debug!("Supervisor({}): Creating supervisor.", self.id());
let parent = Parent::supervisor(self.as_ref());
let bcast = Broadcast::new(parent, BastionPathElement::Supervisor(BastionId::new()));
debug!(
"Supervisor({}): Initializing Supervisor({}).",
self.id(),
bcast.id()
);
let supervisor = Supervisor::new(bcast);
let supervisor = init(supervisor);
debug!("Supervisor({}): Initialized.", supervisor.id());
let supervisor_ref = supervisor.as_ref();
debug!(
"Supervisor({}): Deploying Supervisor({}).",
self.id(),
supervisor.id()
);
let msg = BastionMessage::deploy_supervisor(supervisor);
let env = Envelope::new(msg, self.bcast.path().clone(), self.bcast.sender().clone());
self.bcast.send_self(env);
supervisor_ref
}
/// Creates a new [`Children`], passes it through the specified
/// `init` closure and then starts supervising it.
///
/// If you don't need to chain calls to this `Supervisor`'s methods
/// and need to get a [`ChildrenRef`] referencing the newly
/// created supervisor, use the [`children_ref`] method instead.
///
/// # Arguments
///
/// * `init` - The closure taking the new `Children` as an
/// argument and returning it once configured.
///
/// # Example
///
/// ```rust
/// # use bastion::prelude::*;
/// #
/// # Bastion::init();
/// #
/// # Bastion::supervisor(|sp| {
/// sp.children(|children| {
/// children.with_exec(|ctx: BastionContext| {
/// async move {
/// // Send and receive messages...
/// let opt_msg: Option<SignedMessage> = ctx.try_recv().await;
///
/// // ...and return `Ok(())` or `Err(())` when you are done...
/// Ok(())
/// // Note that if `Err(())` was returned, the supervisor would
/// // restart the children group.
/// }
/// })
/// })
/// # }).unwrap();
/// #
/// # Bastion::start();
/// # Bastion::stop();
/// # Bastion::block_until_stopped();
/// ```
///
/// [`Children`]: children/struct.Children.html
/// [`ChildrenRef`]: children/struct.ChildrenRef.html
/// [`children_ref`]: #method.children_ref
pub fn children<C>(self, init: C) -> Self
where
C: FnOnce(Children) -> Children,
{
debug!("Supervisor({}): Creating children group.", self.id());
let parent = Parent::supervisor(self.as_ref());
let bcast = Broadcast::new(parent, BastionPathElement::Children(BastionId::new()));
debug!(
"Supervisor({}): Initializing Children({}).",
self.id(),
bcast.id()
);
let children = Children::new(bcast);
let mut children = init(children);
debug!("Children({}): Initialized.", children.id());
// FIXME: children group elems launched without the group itself being launched
if let Err(e) = children.register_dispatchers() {
warn!("couldn't register all dispatchers into the registry: {}", e);
};
children.launch_elems();
debug!(
"Supervisor({}): Deploying Children({}).",
self.id(),
children.id()
);
let msg = BastionMessage::deploy_children(children);
let env = Envelope::new(msg, self.bcast.path().clone(), self.bcast.sender().clone());
self.bcast.send_self(env);
self
}
/// Creates a new [`Children`], passes it through the specified
/// `init` closure and then starts supervising it.
///
/// If you need to chain calls to this `Supervisor`'s methods and
/// don't need to get a [`ChildrenRef`] referencing the newly
/// created supervisor, use the [`children`] method instead.
///
/// # Arguments
///
/// * `init` - The closure taking the new `Children` as an
/// argument and returning it once configured.
///
/// # Example
///
/// ```rust
/// # use bastion::prelude::*;
/// #
/// # Bastion::init();
/// #
/// # Bastion::supervisor(|mut sp| {
/// let children_ref: ChildrenRef = sp.children_ref(|children| {
/// children.with_exec(|ctx: BastionContext| {
/// async move {
/// // Send and receive messages...
/// let opt_msg: Option<SignedMessage> = ctx.try_recv().await;
///
/// // ...and return `Ok(())` or `Err(())` when you are done...
/// Ok(())
/// // Note that if `Err(())` was returned, the supervisor would
/// // restart the children group.
/// }
/// })
/// });
/// # sp
/// # }).unwrap();
/// #
/// # Bastion::start();
/// # Bastion::stop();
/// # Bastion::block_until_stopped();
/// ```
///
/// [`Children`]: children/struct.Children.html
/// [`ChildrenRef`]: children/struct.ChildrenRef.html
/// [`children`]: #method.children
pub fn children_ref<C>(&self, init: C) -> ChildrenRef
where
C: FnOnce(Children) -> Children,
{
debug!("Supervisor({}): Creating children group.", self.id());
let parent = Parent::supervisor(self.as_ref());
let bcast = Broadcast::new(parent, BastionPathElement::Children(BastionId::new()));
debug!(
"Supervisor({}): Initializing Children({}).",
self.id(),
bcast.id()
);
let children = Children::new(bcast);
let mut children = init(children);
debug!("Children({}): Initialized.", children.id());
// FIXME: children group elems launched without the group itself being launched
children.launch_elems();
let children_ref = children.as_ref();
debug!(
"Supervisor({}): Deploying Children({}).",
self.id(),
children.id()
);
let msg = BastionMessage::deploy_children(children);
let env = Envelope::new(msg, self.bcast.path().clone(), self.bcast.sender().clone());
self.bcast.send_self(env);
children_ref
}
/// Sets the strategy the supervisor should use when one
/// of its supervised children groups or supervisors dies
/// (in the case of a children group, it could be because one
/// of its elements panicked or returned an error).
///
/// The default strategy is
/// [`SupervisionStrategy::OneForOne`].
///
/// # Arguments
///
/// * `strategy` - The strategy to use:
/// - [`SupervisionStrategy::OneForOne`] would only restart
/// the supervised children groups or supervisors that
/// fault.
/// - [`SupervisionStrategy::OneForAll`] would restart all
/// the supervised children groups or supervisors (even
/// those which were stopped) when one of them faults,
/// respecting the order in which they were added.
/// - [`SupervisionStrategy::RestForOne`] would restart the
/// supervised children groups or supervisors that fault
/// along with all the other supervised children groups
/// or supervisors that were added after them (even the
/// stopped ones), respecting the order in which they
/// were added.
///
/// # Example
///
/// ```rust
/// # use bastion::prelude::*;
/// #
/// # Bastion::init();
/// #
/// Bastion::supervisor(|sp| {
/// // Note that "one-for-one" is the default strategy.
/// sp.with_strategy(SupervisionStrategy::OneForOne)
/// }).expect("Couldn't create the supervisor");
/// #
/// # Bastion::start();
/// # Bastion::stop();
/// # Bastion::block_until_stopped();
/// ```
///
/// [`SupervisionStrategy::OneForOne`]: supervisor/enum.SupervisionStrategy.html#variant.OneForOne
/// [`SupervisionStrategy::OneForAll`]: supervisor/enum.SupervisionStrategy.html#variant.OneForAll
/// [`SupervisionStrategy::RestForOne`]: supervisor/enum.SupervisionStrategy.html#variant.RestForOne
pub fn with_strategy(mut self, strategy: SupervisionStrategy) -> Self {
trace!(
"Supervisor({}): Setting strategy: {:?}",
self.id(),
strategy
);
self.strategy = strategy;
self
}
/// Sets the actor restart strategy the supervisor should use
/// of its supervised children groups or supervisors dies to
/// restore in the correct state.
///
/// The default strategy is the [`ActorRestartStrategy::Immediate`] and
/// unlimited amount of retries.
///
/// # Example
///
/// ```rust
/// # use bastion::prelude::*;
/// # use std::time::Duration;
/// #
/// # Bastion::init();
/// # Bastion::supervisor(|sp| {
/// sp.with_restart_strategy(
/// RestartStrategy::default()
/// .with_restart_policy(RestartPolicy::Tries(5))
/// .with_actor_restart_strategy(
/// ActorRestartStrategy::ExponentialBackOff {
/// timeout: Duration::from_millis(5000),
/// multiplier: 3.0,
/// }
/// )
/// )
/// }).expect("Couldn't create the supervisor");
/// #
/// # Bastion::start();
/// # Bastion::stop();
/// # Bastion::block_until_stopped();
/// ```
pub fn with_restart_strategy(mut self, restart_strategy: RestartStrategy) -> Self {
trace!(
"Supervisor({}): Setting actor restart strategy: {:?}",
self.id(),
restart_strategy
);
self.restart_strategy = restart_strategy;
self
}
/// Sets the callbacks that will get called at this supervisor's
/// different lifecycle events.
///
/// See [`Callbacks`]'s documentation for more information about the
/// different callbacks available.
///
/// # Arguments
///
/// * `callbacks` - The callbacks that will get called for this
/// supervisor.
///
/// # Example
///
/// ```rust
/// # use bastion::prelude::*;
/// #
/// # Bastion::init();
/// #
/// Bastion::supervisor(|sp| {
/// let callbacks = Callbacks::new()
/// .with_before_start(|| println!("Supervisor started."))
/// .with_after_stop(|| println!("Supervisor stopped."));
///
/// sp.with_callbacks(callbacks)
/// }).expect("Couldn't create the supervisor.");
/// #
/// # Bastion::start();
/// # Bastion::stop();
/// # Bastion::block_until_stopped();
/// ```
///
/// [`Callbacks`]: struct.Callbacks.html
pub fn with_callbacks(mut self, callbacks: Callbacks) -> Self {
trace!(
"Supervisor({}): Setting callbacks: {:?}",
self.id(),
callbacks
);
self.callbacks = callbacks;
self
}
async fn restart(&mut self, objects: Vec<RestartedElement>) {
debug!(
"Supervisor({}): Restarting {:?} elements",
self.id(),
objects.len()
);
let mut restart_futures = FuturesOrdered::new();
for object in objects {
match object {
RestartedElement::Supervisor(supervisor_id) => {
let msg = BastionMessage::restart_subtree();
let env =
Envelope::new(msg, self.bcast.path().clone(), self.bcast.sender().clone());
self.bcast.send_child(&supervisor_id, env);
}
RestartedElement::Child { id, parent_id } => {
let index = match self.tracked_groups_order.get(&id) {
Some(index) => *index,
None => continue,
};
let childs = match self.tracked_groups.get_mut(&parent_id) {
Some(childs) => childs,
None => continue,
};
let tracked_state = match childs.get_mut(index) {
Some(tracked_state) => tracked_state,
None => continue,
};
let restarts_count = tracked_state.restarts_count();
let restart_required = match self.restart_strategy.restart_policy() {
RestartPolicy::Always => true,
RestartPolicy::Never => false,
RestartPolicy::Tries(max_retries) => restarts_count < max_retries,
};
let msg = match restart_required {
true => {
tracked_state.increase_restarts_counter();
let state = tracked_state.state();
BastionMessage::restore_child(id, state)
}
false => {
self.remove_child(&id.clone(), &parent_id.clone());
BastionMessage::drop_child(id)
}
};
let restart_strategy = self.restart_strategy.clone();
restart_futures.push(async move {
if restart_required {
restart_strategy.apply_strategy(restarts_count).await;
}
(parent_id, msg)
});
}
}
}
while let Some((receiver, msg)) = restart_futures.next().await {
let env = Envelope::new(msg, self.bcast.path().clone(), self.bcast.sender().clone());
self.bcast.send_child(&receiver, env);
}
}
fn remove_child(&mut self, id: &BastionId, parent_id: &BastionId) {
let index = match self.tracked_groups_order.get(id) {
Some(index) => *index,
None => return,
};
let childs = match self.tracked_groups.get_mut(parent_id) {
Some(childs) => childs,
None => return,
};
childs.remove(index);
for (new_index, state) in childs.iter().enumerate() {
let child_id = state.id.clone();
self.tracked_groups_order.insert(child_id, new_index);
}
}
async fn stop(&mut self, range: Range<usize>) {
debug!("Supervisor({}): Stopping range: {:?}", self.id(), range);
if range.start == 0 {
self.bcast.stop_children();
} else {
// FIXME: panics
for id in self.order.get(range.clone()).unwrap() {
trace!("Supervised({}): Stopping Supervised({}).", self.id(), id);
self.bcast.stop_child(id);
}
}
let mut supervised = FuturesOrdered::new();
// FIXME: panics?
for id in self.order.get(range.clone()).unwrap() {
// TODO: Err if None?
if let Some((_, launched)) = self.launched.remove(&id) {
// TODO: add a "stopped" list and poll from it instead of awaiting
supervised.push(launched);
}
}
while let Some(supervised) = supervised.next().await {
match supervised {
Some(supervised) => {
trace!(
"Supervisor({}): Supervised({}) stopped.",
self.id(),
supervised.id()
);
supervised.callbacks().after_stop();
let id = supervised.id().clone();
self.stopped.insert(id, supervised);
}
// FIXME
None => unimplemented!(),
}
}
}
async fn kill(&mut self, range: Range<usize>) {
debug!("Supervisor({}): Killing range: {:?}", self.id(), range);
if range.start == 0 {
self.bcast.kill_children();
} else {
// FIXME: panics
for id in self.order.get(range.clone()).unwrap() {
trace!("Supervised({}): Killing Supervised({}).", self.id(), id);
self.bcast.kill_child(id);
}
}
let mut supervised = FuturesOrdered::new();
// FIXME: panics?
for id in self.order.get(range.clone()).unwrap() {
// TODO: Err if None?
if let Some((_, launched)) = self.launched.remove(&id) {
// TODO: add a "stopped" list and poll from it instead of awaiting
supervised.push(launched);
}
}
while let Some(supervised) = supervised.next().await {
match supervised {
Some(supervised) => {
trace!(
"Supervisor({}): Supervised({}) stopped.",
self.id(),
supervised.id()
);
let id = supervised.id().clone();
self.killed.insert(id, supervised);
}
// FIXME
None => unimplemented!(),
}
}
}
fn stopped(&mut self) {
debug!("Supervisor({}): Stopped.", self.id());
self.bcast.stopped();
}
fn faulted(&mut self) {
debug!("Supervisor({}): Faulted.", self.id());
self.bcast.faulted();
}
async fn recover(&mut self, id: BastionId, parent_id: BastionId) -> Result<(), ()> {
debug!(
"Supervisor({}): Recovering using strategy: {:?}",
self.id(),
self.strategy
);
match self.strategy {
SupervisionStrategy::OneForOne => {
let search_method = ActorSearchMethod::OneActor { id, parent_id };
let objects = self.search_restarted_objects(search_method);
self.restart(objects).await;
}
SupervisionStrategy::OneForAll => {
let search_method = ActorSearchMethod::All;
let objects = self.search_restarted_objects(search_method);
self.restart(objects).await;
// TODO: should be empty
self.stopped.shrink_to_fit();
self.killed.shrink_to_fit();
}
SupervisionStrategy::RestForOne => {
let search_method = ActorSearchMethod::FromActor { id, parent_id };
let objects = self.search_restarted_objects(search_method);
self.restart(objects).await;
}
}
Ok(())
}
fn search_restarted_objects(&self, search_method: ActorSearchMethod) -> Vec<RestartedElement> {
let mut objects = Vec::new();
match search_method {
ActorSearchMethod::OneActor { id, parent_id } => {
let element = match self.tracked_groups.contains_key(&parent_id) {
true => RestartedElement::Child { id, parent_id },
false => RestartedElement::Supervisor(id),
};
objects.push(element)
}
ActorSearchMethod::FromActor { id, parent_id } => {
let childs = self.tracked_groups.get(&parent_id).unwrap();
let start_index = *self.tracked_groups_order.get(&id).unwrap();
// Adding all elements in the group from the given actor
childs.iter().skip(start_index).for_each(|tracked_state| {
let id = tracked_state.id();
let element = RestartedElement::Child {
id,
parent_id: parent_id.clone(),
};
objects.push(element)
});
// And then a rest after the failed group
let (rest_index, _) = self.launched.get(&parent_id).unwrap();
for index in *rest_index + 1..self.order.len() {
let element_id = &self.order[index];
match self.tracked_groups.get(element_id) {
Some(childs) => {
for tracked_state in childs {
let restarted_element = RestartedElement::Child {
id: tracked_state.id(),
parent_id: element_id.clone(),
};
objects.push(restarted_element);
}
}
None => {
let restarted_element = RestartedElement::Supervisor(id.clone());
objects.push(restarted_element);
}
}
}
}
ActorSearchMethod::All => {
for id in self.order.iter() {
match self.tracked_groups.get(&id) {
Some(childs) => {
for tracked_state in childs {
let restarted_element = RestartedElement::Child {
id: tracked_state.id(),
parent_id: id.clone(),
};
objects.push(restarted_element);
}
}
None => {
let restarted_element = RestartedElement::Supervisor(id.clone());
objects.push(restarted_element);
}
}
}
}
}
objects
}
async fn restart_subtree(&mut self) {
if self.subtree_restarts < self.subtree_restarts_limit {
self.subtree_restarts += 1;
let restarted_objects = self.search_restarted_objects(ActorSearchMethod::All);
self.restart(restarted_objects).await;
}
}
async fn deinit_with_stop(&mut self) {
self.stop(0..self.order.len()).await;
self.stopped();
}
async fn deinit_with_kill(&mut self) {
self.kill(0..self.order.len()).await;
self.stopped();
}
async fn deploy_supervised_object(&mut self, deployment: Box<Deployment>) {
let supervised = match *deployment {
Deployment::Supervisor(supervisor) => {
debug!(
"Supervisor({}): Deploying Supervisor({}).",
self.id(),
supervisor.id()
);
supervisor.callbacks().before_start();
Supervised::supervisor(supervisor)
}
Deployment::Children(children) => {
debug!(
"Supervisor({}): Deploying Children({}).",
self.id(),
children.id()
);
children.callbacks().before_start();
Supervised::children(children)
}
};
self.bcast.register(supervised.bcast());
if self.started {
let msg = BastionMessage::start();
let env = Envelope::new(msg, self.bcast.path().clone(), self.bcast.sender().clone());
self.bcast.send_child(supervised.id(), env);
}
debug!(
"Supervisor({}): Launching Supervised({}).",
self.id(),
supervised.id()
);
let id = supervised.id().clone();
let launched = supervised.launch();
self.launched
.insert(id.clone(), (self.order.len(), launched));
self.order.push(id);
}
async fn cleanup_supervised_object(&mut self, id: BastionId) {
// FIXME: Err if None?
if let Some((_, launched)) = self.launched.remove(&id) {
debug!("Supervisor({}): Supervised({}) stopped.", self.id(), id);
// TODO: add a "waiting" list an poll from it instead of awaiting
// FIXME: panics?
let supervised = launched.await.unwrap();
supervised.callbacks().after_stop();
self.bcast.unregister(&id);
self.stopped.insert(id.clone(), supervised);
}
}
async fn recover_supervised_object(
&mut self,
id: BastionId,
parent_id: BastionId,
) -> Result<(), ()> {
if self.launched.contains_key(&id) {
warn!("Supervisor({}): Supervised({}) faulted.", self.id(), id);
}
if self.recover(id, parent_id).await.is_err() {
// TODO: stop or kill?
self.kill(0..self.order.len()).await;
self.faulted();
return Err(());
}
Ok(())
}
async fn handle(&mut self, env: Envelope) -> Result<(), ()> {
match env {
Envelope {
msg: BastionMessage::Start,
..
} => unreachable!(),
Envelope {
msg: BastionMessage::Stop,
..
} => {
self.deinit_with_stop().await;
return Err(());
}
Envelope {
msg: BastionMessage::Kill,
..
} => {
self.deinit_with_kill().await;
return Err(());
}
Envelope {
msg: BastionMessage::Deploy(deployment),
..
} => self.deploy_supervised_object(deployment).await,
// FIXME
Envelope {
msg: BastionMessage::Prune { .. },
..
} => unimplemented!(),
Envelope {
msg: BastionMessage::SuperviseWith(strategy),
..
} => {
debug!(
"Supervisor({}): Setting strategy: {:?}",
self.id(),
strategy
);
self.strategy = strategy;
}
Envelope {
msg: BastionMessage::ApplyCallback { .. },
..
} => unreachable!(),
Envelope {
msg:
BastionMessage::InstantiatedChild {
parent_id,
child_id,
state,
},
..
} => {
let child_state = TrackedChildState::new(child_id.clone(), state);
match self.tracked_groups.get_mut(&parent_id) {
Some(childs) => {
childs.push(child_state);
self.tracked_groups_order.insert(child_id, childs.len() - 1);
}
None => {
self.tracked_groups.insert(parent_id, vec![child_state]);
self.tracked_groups_order.insert(child_id, 0);
}
}
}
Envelope {
msg: BastionMessage::Message(ref message),
..
} => {
debug!(
"Supervisor({}): Broadcasting a message: {:?}",
self.id(),
message
);
self.bcast.send_children(env);
}
Envelope {
msg: BastionMessage::RestartRequired { id, parent_id },
..
} => {
if self.recover_supervised_object(id, parent_id).await.is_err() {
return Err(());
}
}
Envelope {
msg: BastionMessage::FinishedChild { id, parent_id },
..
} => self.remove_child(&id, &parent_id),
Envelope {
msg: BastionMessage::RestartSubtree,
..
} => self.restart_subtree().await,
Envelope {
msg: BastionMessage::RestoreChild { .. },
..
} => unreachable!(),
Envelope {
msg: BastionMessage::DropChild { .. },
..
} => unreachable!(),
Envelope {
msg: BastionMessage::SetState { .. },
..
} => unreachable!(),
Envelope {
msg: BastionMessage::Stopped { id },
..
} => self.cleanup_supervised_object(id).await,
Envelope {
msg: BastionMessage::Faulted { id },
..
} => self.cleanup_supervised_object(id).await,
Envelope {
msg: BastionMessage::Heartbeat,
..
} => unreachable!(),
}
Ok(())
}
async fn initialize(&mut self) -> Result<(), ()> {
trace!(
"Supervisor({}): Received a new message (started=false): {:?}",
self.id(),
BastionMessage::Start
);
debug!("Supervisor({}): Starting.", self.id());
self.started = true;
let msg = BastionMessage::start();
let env = Envelope::new(msg, self.bcast.path().clone(), self.bcast.sender().clone());
self.bcast.send_children(env);
let msgs = self.pre_start_msgs.drain(..).collect::<Vec<_>>();
self.pre_start_msgs.shrink_to_fit();
debug!(
"Supervisor({}): Replaying messages received before starting.",
self.id()
);
for msg in msgs {
trace!("Supervisor({}): Replaying message: {:?}", self.id(), msg);
if self.handle(msg).await.is_err() {
return Err(());
}
}
Ok(())
}
async fn run(mut self) -> Self {
debug!("Supervisor({}): Launched.", self.id());
loop {
match poll!(&mut self.bcast.next()) {
// TODO: Err if started == true?
Poll::Ready(Some(Envelope {
msg: BastionMessage::Start,
..
})) => {
if self.initialize().await.is_err() {
return self;
}
}
Poll::Ready(Some(msg)) if !self.started => {
trace!(
"Supervisor({}): Received a new message (started=false): {:?}",
self.id(),
msg
);
self.pre_start_msgs.push(msg);
}
Poll::Ready(Some(msg)) => {
trace!(
"Supervisor({}): Received a new message (started=true): {:?}",
self.id(),
msg
);
if self.handle(msg).await.is_err() {
return self;
}
}
// NOTE: because `Broadcast` always holds both a `Sender` and
// `Receiver` of the same channel, this would only be
// possible if the channel was closed, which never happens.
Poll::Ready(None) => unreachable!(),
Poll::Pending => pending!(),
}
}
}
pub(crate) fn launch(self) -> RecoverableHandle<Self> {
debug!("Supervisor({}): Launching.", self.id());
let stack = self.stack();
pool::spawn(self.run(), stack)
}
}
impl SupervisorRef {
pub(crate) fn new(id: BastionId, sender: Sender, path: Arc<BastionPath>) -> Self {
SupervisorRef { id, sender, path }
}
/// Returns the identifier of the supervisor this `SupervisorRef`
/// is referencing.
///
/// Note that the supervisor's identifier is reset when it is
/// restarted.
///
/// # Example
///
/// ```rust
/// # use bastion::prelude::*;
/// #
/// # Bastion::init();
/// #
/// let supervisor_ref = Bastion::supervisor(|sp| {
/// // ...
/// # sp
/// }).expect("Couldn't create the supervisor.");
///
/// let supervisor_id: &BastionId = supervisor_ref.id();
/// #
/// # Bastion::start();
/// # Bastion::stop();
/// # Bastion::block_until_stopped();
/// ```
pub fn id(&self) -> &BastionId {
&self.id
}
/// Creates a new [`Supervisor`], passes it through the specified
/// `init` closure and then sends it to the supervisor this
/// `SupervisorRef` is referencing to supervise it.
///
/// This method returns a [`SupervisorRef`] referencing the newly
/// created supervisor if it succeeded, or `Err(())`
/// otherwise.
///
/// # Arguments
///
/// * `init` - The closure taking the new [`Supervisor`] as an
/// argument and returning it once configured.
///
/// # Example
///
/// ```
/// # use bastion::prelude::*;
/// #
/// # Bastion::init();
/// #
/// # let mut parent_ref = Bastion::supervisor(|sp| sp).unwrap();
/// let sp_ref: SupervisorRef = parent_ref.supervisor(|sp| {
/// // Configure the supervisor...
/// sp.with_strategy(SupervisionStrategy::OneForOne)
/// // ...and return it.
/// }).expect("Couldn't create the supervisor.");
/// #
/// # Bastion::start();
/// # Bastion::stop();
/// # Bastion::block_until_stopped();
/// ```
///
/// [`Supervisor`]: supervisor/struct.Supervisor.html
pub fn supervisor<S>(&self, init: S) -> Result<Self, ()>
where
S: FnOnce(Supervisor) -> Supervisor,
{
debug!("SupervisorRef({}): Creating supervisor.", self.id());
let parent = Parent::supervisor(self.clone());
let bcast = Broadcast::new(parent, BastionPathElement::Supervisor(BastionId::new()));
debug!(
"SupervisorRef({}): Initializing Supervisor({}).",
self.id(),
bcast.id()
);
let supervisor = Supervisor::new(bcast);
let supervisor = init(supervisor);
let supervisor_ref = supervisor.as_ref();
debug!("Supervisor({}): Initialized.", supervisor.id());
debug!(
"SupervisorRef({}): Deploying Supervisor({}).",
self.id(),
supervisor.id()
);
let msg = BastionMessage::deploy_supervisor(supervisor);
let env = Envelope::new(msg, self.path.clone(), self.sender.clone());
self.send(env).map_err(|_| ())?;
Ok(supervisor_ref)
}
/// Creates a new [`Children`], passes it through the specified
/// `init` closure and then sends it to the supervisor this
/// `SupervisorRef` is referencing to supervise it.
///
/// This methods returns a [`ChildrenRef`] referencing the newly
/// created children group it it succeeded, or `Err(())`
/// otherwise.
///
/// # Arguments
///
/// * `init` - The closure taking the new [`Children`] as an
/// argument and returning it once configured.
///
/// # Example
///
/// ```
/// # use bastion::prelude::*;
/// #
/// # Bastion::init();
/// #
/// # let sp_ref = Bastion::supervisor(|sp| sp).unwrap();
/// let children_ref: ChildrenRef = sp_ref.children(|children| {
/// children.with_exec(|ctx: BastionContext| {
/// async move {
/// // Send and receive messages...
/// let opt_msg: Option<SignedMessage> = ctx.try_recv().await;
///
/// // ...and return `Ok(())` or `Err(())` when you are done...
/// Ok(())
/// // Note that if `Err(())` was returned, the supervisor would
/// // restart the children group.
/// }
/// })
/// }).expect("Couldn't create the children group.");
/// #
/// # Bastion::start();
/// # Bastion::stop();
/// # Bastion::block_until_stopped();
/// ```
///
/// [`Children`]: children/struct.Children.html
/// [`ChildrenRef`]: children/struct.ChildrenRef.html
pub fn children<C>(&self, init: C) -> Result<ChildrenRef, ()>
where
C: FnOnce(Children) -> Children,
{
self.children_with_id(BastionId::new(), init)
}
pub(crate) fn children_with_id<C>(&self, id: BastionId, init: C) -> Result<ChildrenRef, ()>
where
C: FnOnce(Children) -> Children,
{
debug!("SupervisorRef({}): Creating children group.", self.id());
let parent = Parent::supervisor(self.clone());
let bcast = Broadcast::new(parent, BastionPathElement::Children(id));
debug!(
"SupervisorRef({}): Initializing Children({}).",
self.id(),
bcast.id()
);
let children = Children::new(bcast);
let mut children = init(children);
debug!("Children({}): Initialized.", children.id());
// FIXME: children group elems launched without the group itself being launched
children.launch_elems();
let children_ref = children.as_ref();
debug!(
"SupervisorRef({}): Deplying Children({}).",
self.id(),
children.id()
);
let msg = BastionMessage::deploy_children(children);
let env = Envelope::new(msg, self.path.clone(), self.sender.clone());
self.send(env).map_err(|_| ())?;
Ok(children_ref)
}
/// Sends to the supervisor this `SupervisorRef` is
/// referencing the strategy that it should start
/// using when one of its supervised children groups or
/// supervisors dies (in the case of a children group,
/// it could be because one of its elements panicked or
/// returned an error).
///
/// The default strategy `Supervisor` is
/// [`SupervisionStrategy::OneForOne`].
///
/// This method returns `()` if it succeeded, or `Err(())`
/// otherwise.
///
/// # Arguments
///
/// * `strategy` - The strategy to use:
/// - [`SupervisionStrategy::OneForOne`] would only restart
/// the supervised children groups or supervisors that
/// fault.
/// - [`SupervisionStrategy::OneForAll`] would restart all
/// the supervised children groups or supervisors (even
/// those which were stopped) when one of them faults,
/// respecting the order in which they were added.
/// - [`SupervisionStrategy::RestForOne`] would restart the
/// supervised children groups or supervisors that fault
/// along with all the other supervised children groups
/// or supervisors that were added after them (even the
/// stopped ones), respecting the order in which they
/// were added.
///
/// # Example
///
/// ```
/// # use bastion::prelude::*;
/// #
/// # Bastion::init();
/// #
/// # let sp_ref = Bastion::supervisor(|sp| sp).unwrap();
/// // Note that "one-for-one" is the default strategy.
/// sp_ref.strategy(SupervisionStrategy::OneForOne);
/// #
/// # Bastion::start();
/// # Bastion::stop();
/// # Bastion::block_until_stopped();
/// ```
///
/// [`SupervisionStrategy::OneForOne`]: supervisor/enum.SupervisionStrategy.html#variant.OneForOne
/// [`SupervisionStrategy::OneForAll`]: supervisor/enum.SupervisionStrategy.html#variant.OneForAll
/// [`SupervisionStrategy::RestForOne`]: supervisor/enum.SupervisionStrategy.html#variant.RestForOne
pub fn strategy(&self, strategy: SupervisionStrategy) -> Result<(), ()> {
debug!(
"SupervisorRef({}): Setting strategy: {:?}",
self.id(),
strategy
);
let msg = BastionMessage::supervise_with(strategy);
let env = Envelope::from_dead_letters(msg);
self.send(env).map_err(|_| ())
}
/// Sends a message to the supervisor this `SupervisorRef`
/// is referencing which will then send it to all of its
/// supervised children groups and supervisors.
///
/// This method returns `()` if it succeeded, or `Err(msg)`
/// otherwise.
///
/// # Arguments
///
/// * `msg` - The message to send.
///
/// # Example
///
/// ```
/// # use bastion::prelude::*;
/// #
/// # fn main() {
/// # Bastion::init();
/// #
/// # let sp_ref = Bastion::supervisor(|sp| sp).unwrap();
/// let msg = "A message containing data.";
/// sp_ref.broadcast(msg).expect("Couldn't send the message.");
///
/// # Bastion::children(|children| {
/// # children.with_exec(|ctx: BastionContext| {
/// # async move {
/// // And then in every future of the elements of the children
/// // groups that are supervised by this supervisor or one of
/// // its supervised supervisors (etc.)...
/// msg! { ctx.recv().await?,
/// ref msg: &'static str => {
/// assert_eq!(msg, &"A message containing data.");
/// };
/// // We are only broadcasting a `&'static str` in this
/// // example, so we know that this won't happen...
/// _: _ => ();
/// }
/// #
/// # Ok(())
/// # }
/// # })
/// # }).unwrap();
/// #
/// # Bastion::start();
/// # Bastion::stop();
/// # Bastion::block_until_stopped();
/// # }
/// ```
pub fn broadcast<M: Message>(&self, msg: M) -> Result<(), M> {
debug!(
"SupervisorRef({}): Broadcasting message: {:?}",
self.id(),
msg
);
let msg = BastionMessage::broadcast(msg);
let env = Envelope::from_dead_letters(msg);
// FIXME: panics?
self.send(env).map_err(|env| env.into_msg().unwrap())
}
/// Sends a message to the supervisor this `SupervisorRef`
/// is referencing to tell it to stop every running children
/// groups and supervisors that it is supervising.
///
/// This method returns `()` if it succeeded, or `Err(())`
/// otherwise.
///
/// # Example
///
/// ```
/// # use bastion::prelude::*;
/// #
/// # Bastion::init();
/// #
/// # let sp_ref = Bastion::supervisor(|sp| sp).unwrap();
/// sp_ref.stop().expect("Couldn't send the message.");
/// #
/// # Bastion::start();
/// # Bastion::stop();
/// # Bastion::block_until_stopped();
/// ```
pub fn stop(&self) -> Result<(), ()> {
debug!("SupervisorRef({}): Stopping.", self.id());
let msg = BastionMessage::stop();
let env = Envelope::from_dead_letters(msg);
self.send(env).map_err(|_| ())
}
/// Sends a message to the supervisor this `SupervisorRef`
/// is referencing to tell it to kill every running children
/// groups and supervisors that it is supervising.
///
/// This method returns `()` if it succeeded, or `Err(())`
/// otherwise.
///
/// # Example
///
/// ```
/// # use bastion::prelude::*;
/// #
/// # Bastion::init();
/// #
/// # let sp_ref = Bastion::supervisor(|sp| sp).unwrap();
/// sp_ref.kill().expect("Couldn't send the message.");
/// #
/// # Bastion::start();
/// # Bastion::stop();
/// # Bastion::block_until_stopped();
/// ```
pub fn kill(&self) -> Result<(), ()> {
debug!("SupervisorRef({}): Killing.", self.id());
let msg = BastionMessage::kill();
let env = Envelope::from_dead_letters(msg);
self.send(env).map_err(|_| ())
}
pub(crate) fn send(&self, env: Envelope) -> Result<(), Envelope> {
trace!("SupervisorRef({}): Sending message: {:?}", self.id(), env);
self.sender
.unbounded_send(env)
.map_err(|err| err.into_inner())
}
pub(crate) fn path(&self) -> &Arc<BastionPath> {
&self.path
}
}
impl TrackedChildState {
fn new(id: BastionId, state: Arc<Pin<Box<ContextState>>>) -> Self {
TrackedChildState {
id,
state,
restarts_counts: 0,
}
}
fn id(&self) -> BastionId {
self.id.clone()
}
fn state(&self) -> Arc<Pin<Box<ContextState>>> {
self.state.clone()
}
fn restarts_count(&self) -> usize {
self.restarts_counts
}
fn increase_restarts_counter(&mut self) {
self.restarts_counts += 1;
}
}
impl Supervised {
fn supervisor(supervisor: Supervisor) -> Self {
Supervised::Supervisor(supervisor)
}
fn children(children: Children) -> Self {
Supervised::Children(children)
}
fn stack(&self) -> ProcStack {
trace!("Supervised({}): Creating ProcStack.", self.id());
// FIXME: with_id
ProcStack::default()
}
fn id(&self) -> &BastionId {
match self {
Supervised::Supervisor(supervisor) => supervisor.id(),
Supervised::Children(children) => children.id(),
}
}
fn bcast(&self) -> &Broadcast {
match self {
Supervised::Supervisor(supervisor) => supervisor.bcast(),
Supervised::Children(children) => children.bcast(),
}
}
fn callbacks(&self) -> &Callbacks {
match self {
Supervised::Supervisor(supervisor) => supervisor.callbacks(),
Supervised::Children(children) => children.callbacks(),
}
}
fn launch(self) -> RecoverableHandle<Self> {
debug!("Supervised({}): Launching.", self.id());
let stack = self.stack();
match self {
Supervised::Supervisor(supervisor) => {
pool::spawn(
async {
// FIXME: panics?
let supervisor = supervisor.launch().await.unwrap();
Supervised::Supervisor(supervisor)
},
stack,
)
}
Supervised::Children(children) => {
pool::spawn(
async {
// FIXME: panics?
let children = children.launch().await.unwrap();
Supervised::Children(children)
},
stack,
)
}
}
}
}
impl RestartStrategy {
/// Creates a new instance of RestartStrategy.
///
/// # Arguments
///
/// * `restart_policy` - Defines a restart policy to use for failed actor:
/// - [`RestartStrategy::Always`] would restart the
/// failed actor each time as it fails.
/// - [`RestartStrategy::Never`] would not restart the
/// failed actor and remove it from tracking.
/// - [`RestartStrategy::Tries`] would restart the
/// failed actor a limited amount of times. If can't be started,
/// then will remove it from tracking.
///
/// * `strategy` - The strategy to use:
/// - [`ActorRestartStrategy::Immediate`] would restart the
/// failed actor as soon as possible.
/// - [`ActorRestartStrategy::LinearBackOff`] would restart the
/// failed actor with the delay increasing linearly.
/// - [`ActorRestartStrategy::ExponentialBackOff`] would restart the
/// failed actor with the delay, multiplied by given coefficient.
///
/// # Example
///
/// ```rust
/// # use std::time::Duration;
/// # use bastion::prelude::*;
/// #
/// let actor_restart_strategy = ActorRestartStrategy::LinearBackOff {
/// timeout: Duration::from_secs(5)
/// };
/// let restart_strategy = RestartStrategy::default()
/// .with_actor_restart_strategy(actor_restart_strategy);
/// ```
///
/// [`RestartStrategy::Always`]: enum.RestartPolicy.html#variant.Always
/// [`RestartStrategy::Never`]: enum.RestartPolicy.html#variant.Never
/// [`RestartStrategy::Tries`]: enum.RestartPolicy.html#variant.Tries
/// [`ActorRestartStrategy::Immediate`]: enum.ActorRestartStrategy.html#variant.Immediate
/// [`ActorRestartStrategy::LinearBackOff`]: enum.ActorRestartStrategy.html#variant.LinearBackOff
/// [`ActorRestartStrategy::ExponentialBackOff`]: enum.ActorRestartStrategy.html#variant.ExponentialBackOff
pub fn new(restart_policy: RestartPolicy, strategy: ActorRestartStrategy) -> Self {
RestartStrategy {
restart_policy,
strategy,
}
}
/// Returns the acceptable count of retries for the failed actor.
/// The `None` value means the amount of attempts is unlimited.
pub fn restart_policy(&self) -> RestartPolicy {
self.restart_policy.clone()
}
/// Returns the current strategy for restarting failed actors.
pub fn strategy(&self) -> ActorRestartStrategy {
self.strategy.clone()
}
/// Sets the limit of attempts for restoring failed actors.
pub fn with_restart_policy(mut self, restart_policy: RestartPolicy) -> Self {
self.restart_policy = restart_policy;
self
}
/// Sets the actor restart strategy the supervisor should use
/// of its supervised children groups or supervisors dies to
/// restore in the correct state.
pub fn with_actor_restart_strategy(mut self, strategy: ActorRestartStrategy) -> Self {
self.strategy = strategy;
self
}
pub(crate) async fn apply_strategy(&self, restarts_count: usize) {
if let Some(dur) = self.strategy.calculate(restarts_count) {
Delay::new(dur).await;
}
}
}
impl Default for SupervisionStrategy {
fn default() -> Self {
SupervisionStrategy::OneForOne
}
}
impl Default for RestartStrategy {
fn default() -> Self {
RestartStrategy {
restart_policy: RestartPolicy::Always,
strategy: ActorRestartStrategy::default(),
}
}
}
impl Default for ActorRestartStrategy {
fn default() -> Self {
ActorRestartStrategy::Immediate
}
}
impl PartialEq for SupervisorRef {
fn eq(&self, other: &Self) -> bool {
self.id == other.id
}
}
impl Eq for SupervisorRef {}