use crate::Report;
use alloc::{boxed::Box, sync::Arc, vec::Vec};
use core::{fmt, num::NonZeroUsize};
use enough::{Stop, StopReason};
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
#[non_exhaustive]
pub enum Total {
Exact(u64),
Estimated(u64),
Unknown,
}
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
#[non_exhaustive]
pub enum Execution {
#[default]
Unspecified,
Sequence,
ForkJoin,
#[non_exhaustive]
WorkPool {
max_parallelism: NonZeroUsize,
},
}
impl Execution {
pub const fn work_pool(max_parallelism: NonZeroUsize) -> Self {
Self::WorkPool { max_parallelism }
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
#[non_exhaustive]
pub enum Outcome {
Succeeded,
Skipped,
Cancelled,
Failed,
Abandoned,
NotRun,
}
impl Outcome {
pub fn from_result<T, E>(result: &Result<T, E>, is_stop: impl FnOnce(&E) -> bool) -> Self {
match result {
Ok(_) => Self::Succeeded,
Err(error) if is_stop(error) => Self::Cancelled,
Err(_) => Self::Failed,
}
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
#[non_exhaustive]
pub enum PlanError {
EmptyOrZeroWeight,
Overflow,
AlreadyInUse,
NotALeaf,
Unsupported,
NotShareable,
Finished,
UnfinishedChildren,
NoMoreStages,
StageMismatch,
}
impl fmt::Display for PlanError {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.write_str(match self {
Self::EmptyOrZeroWeight => "a split needs at least one part with a positive weight",
Self::Overflow => "phase weights or identifiers overflowed",
Self::AlreadyInUse => "the phase already counted work or planned children",
Self::NotALeaf => "only a leaf phase has its own total",
Self::Unsupported => "this pulse cannot plan children",
Self::NotShareable => "this pulse cannot provide an owned view",
Self::Finished => "the phase already has an outcome",
Self::UnfinishedChildren => "every child must finish before its parent",
Self::NoMoreStages => "every declared stage has already run",
Self::StageMismatch => "stage reservation does not match the next declaration",
})
}
}
impl core::error::Error for PlanError {}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
#[non_exhaustive]
pub struct PhaseSpec<'a> {
pub name: &'a str,
pub weight: u64,
pub total: Total,
pub units: &'a str,
pub execution: Execution,
}
impl<'a> PhaseSpec<'a> {
pub const fn new(name: &'a str, weight: u64, total: Total) -> Self {
Self {
name,
weight,
total,
units: "items",
execution: Execution::Unspecified,
}
}
pub const fn units(mut self, units: &'a str) -> Self {
self.units = units;
self
}
pub const fn execution(mut self, execution: Execution) -> Self {
self.execution = execution;
self
}
pub fn validate_split(parts: &[PhaseSpec<'_>]) -> Result<u64, PlanError> {
if parts.is_empty() {
return Err(PlanError::EmptyOrZeroWeight);
}
let mut sum = 0_u64;
for part in parts {
if part.weight == 0 {
return Err(PlanError::EmptyOrZeroWeight);
}
sum = sum.checked_add(part.weight).ok_or(PlanError::Overflow)?;
}
Ok(sum)
}
}
pub trait Pulse: Stop + Report {
#[track_caller]
fn record_issue(&self, _error: PlanError) {}
fn split(
&self,
execution: Execution,
parts: &[PhaseSpec<'_>],
) -> Result<Vec<Child<'_>>, PlanError>;
fn start(&self) -> Result<(), PlanError> {
Ok(())
}
fn set_total(&self, _total: Total) -> Result<(), PlanError> {
Err(PlanError::Unsupported)
}
fn share(&self) -> Result<SharedPulse, PlanError> {
Err(PlanError::NotShareable)
}
}
macro_rules! forward_pulse {
($($wrapper:ty),*) => {$(
impl<P: Pulse + ?Sized> Pulse for $wrapper {
#[track_caller]
fn record_issue(&self, error: PlanError) {
(**self).record_issue(error);
}
#[inline]
fn split(
&self,
execution: Execution,
parts: &[PhaseSpec<'_>],
) -> Result<Vec<Child<'_>>, PlanError> {
(**self).split(execution, parts)
}
fn start(&self) -> Result<(), PlanError> {
(**self).start()
}
fn set_total(&self, total: Total) -> Result<(), PlanError> {
(**self).set_total(total)
}
fn share(&self) -> Result<SharedPulse, PlanError> {
(**self).share()
}
}
)*};
}
forward_pulse!(&P, &mut P, Box<P>, Arc<P>);
pub trait ChildPulse: Pulse {
fn complete_inferred(self: Box<Self>, outcome: Outcome) {
self.complete_as(outcome);
}
fn complete_as(self: Box<Self>, outcome: Outcome) {
let _ = self.finish(outcome);
}
fn finish(self: Box<Self>, outcome: Outcome) -> Result<(), PlanError>;
}
#[must_use = "finish the child after its work joins; dropping a tracked child abandons it"]
pub struct Child<'a> {
pulse: Option<Box<dyn ChildPulse + 'a>>,
}
impl<'a> Child<'a> {
pub fn complete_inferred(self, outcome: Outcome) {
if let Some(pulse) = self.pulse {
pulse.complete_inferred(outcome);
}
}
pub(crate) fn untracked(parent: &'a dyn Pulse) -> Self {
Self::new(Untracked { inner: parent })
}
#[track_caller]
pub fn plan(&self, execution: Execution, parts: &[PhaseSpec<'_>]) -> Vec<Child<'_>> {
self.pulse().plan(execution, parts)
}
pub fn new(pulse: impl ChildPulse + 'a) -> Self {
Self {
pulse: Some(Box::new(pulse)),
}
}
pub const fn inert() -> Self {
Self { pulse: None }
}
pub fn finish(self, outcome: Outcome) -> Result<(), PlanError> {
match self.pulse {
Some(pulse) => pulse.finish(outcome),
None => Ok(()),
}
}
#[inline]
pub(crate) fn pulse(&self) -> &(dyn Pulse + 'a) {
match &self.pulse {
Some(pulse) => &**pulse,
None => &NoPulse,
}
}
}
impl fmt::Debug for Child<'_> {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("Child").finish_non_exhaustive()
}
}
impl crate::Complete for Child<'_> {
fn complete_as(self, outcome: Outcome) {
if let Some(pulse) = self.pulse {
pulse.complete_as(outcome);
}
}
}
impl dyn Pulse + '_ {
#[track_caller]
pub fn plan(&self, execution: Execution, parts: &[PhaseSpec<'_>]) -> Vec<Child<'_>> {
match self.split(execution, parts) {
Ok(children) if children.len() == parts.len() => children,
Ok(_) => {
self.record_issue(PlanError::Unsupported);
untracked_children(self, parts.len())
}
Err(error) => {
self.record_issue(error);
untracked_children(self, parts.len())
}
}
}
}
pub(crate) fn untracked_children(parent: &dyn Pulse, count: usize) -> Vec<Child<'_>> {
let mut children = Vec::with_capacity(count);
for _ in 0..count {
children.push(Child::untracked(parent));
}
children
}
struct Untracked<P> {
inner: P,
}
impl<P: Pulse> Stop for Untracked<P> {
#[track_caller]
fn check(&self) -> Result<(), StopReason> {
self.inner.check()
}
fn may_stop(&self) -> bool {
self.inner.may_stop()
}
}
impl<P: Pulse> Report for Untracked<P> {
fn advance(&self, _: u64) {}
fn may_report(&self) -> bool {
false
}
}
impl<P: Pulse> Pulse for Untracked<P> {
fn split(&self, _: Execution, parts: &[PhaseSpec<'_>]) -> Result<Vec<Child<'_>>, PlanError> {
PhaseSpec::validate_split(parts)?;
Ok(untracked_children(&self.inner, parts.len()))
}
fn set_total(&self, _: Total) -> Result<(), PlanError> {
Ok(())
}
fn share(&self) -> Result<SharedPulse, PlanError> {
Ok(SharedPulse::new(Untracked {
inner: self.inner.share()?,
}))
}
#[track_caller]
fn record_issue(&self, error: PlanError) {
self.inner.record_issue(error);
}
}
impl<P: Pulse> ChildPulse for Untracked<P> {
fn finish(self: Box<Self>, _: Outcome) -> Result<(), PlanError> {
Ok(())
}
}
impl Stop for Child<'_> {
#[inline]
#[track_caller]
fn check(&self) -> Result<(), StopReason> {
match &self.pulse {
Some(pulse) => pulse.check(),
None => Ok(()),
}
}
#[inline]
fn may_stop(&self) -> bool {
matches!(&self.pulse, Some(pulse) if pulse.may_stop())
}
}
impl Report for Child<'_> {
#[inline]
#[track_caller]
fn advance(&self, completed: u64) {
if let Some(pulse) = &self.pulse {
pulse.advance(completed);
}
}
#[inline]
fn may_report(&self) -> bool {
matches!(&self.pulse, Some(pulse) if pulse.may_report())
}
}
impl Pulse for Child<'_> {
#[track_caller]
fn record_issue(&self, error: PlanError) {
self.pulse().record_issue(error);
}
fn split(
&self,
execution: Execution,
parts: &[PhaseSpec<'_>],
) -> Result<Vec<Child<'_>>, PlanError> {
self.pulse().split(execution, parts)
}
fn start(&self) -> Result<(), PlanError> {
self.pulse().start()
}
fn set_total(&self, total: Total) -> Result<(), PlanError> {
self.pulse().set_total(total)
}
fn share(&self) -> Result<SharedPulse, PlanError> {
self.pulse().share()
}
}
pub struct Inert {
_unique: u8,
}
#[allow(non_upper_case_globals)]
pub static NoPulse: Inert = Inert { _unique: 0 };
impl fmt::Debug for Inert {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.write_str("NoPulse")
}
}
impl Stop for Inert {
#[inline(always)]
fn check(&self) -> Result<(), StopReason> {
Ok(())
}
#[inline(always)]
fn may_stop(&self) -> bool {
false
}
}
impl Report for Inert {
#[inline(always)]
fn advance(&self, _: u64) {}
#[inline(always)]
fn may_report(&self) -> bool {
false
}
}
impl Pulse for Inert {
fn split(&self, _: Execution, parts: &[PhaseSpec<'_>]) -> Result<Vec<Child<'_>>, PlanError> {
PhaseSpec::validate_split(parts)?;
let mut children = Vec::with_capacity(parts.len());
for _ in parts {
children.push(Child::inert());
}
Ok(children)
}
fn set_total(&self, _: Total) -> Result<(), PlanError> {
Ok(())
}
fn share(&self) -> Result<SharedPulse, PlanError> {
Ok(SharedPulse::default())
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::ProgressExt;
use alloc::format;
#[test]
fn no_pulse_children_are_the_static_and_need_no_allocation() {
let parts = [
PhaseSpec::new("a", 1, Total::Exact(1)),
PhaseSpec::new("b", 1, Total::Exact(1)),
];
let children = NoPulse.split(Execution::Sequence, &parts).unwrap();
for child in children {
assert!(child.pulse.is_none(), "no boxed pulse");
assert!(core::ptr::addr_eq(child.pulse(), &NoPulse));
assert!(!child.may_stop() && !child.may_report());
assert!(child.live().is_none());
assert_eq!(child.step(5), Ok(()));
assert_eq!(
format!("{:?}", child.paced(1)),
"Paced { live: false, pending: 0, every: 1 }"
);
child.finish(Outcome::Succeeded).unwrap();
}
assert_eq!(
NoPulse.split(Execution::Sequence, &[]).err(),
Some(PlanError::EmptyOrZeroWeight)
);
}
}
#[derive(Clone, Default)]
pub struct SharedPulse(Option<Arc<dyn Pulse>>);
impl SharedPulse {
pub fn new(pulse: impl Pulse + 'static) -> Self {
Self(Some(Arc::new(pulse)))
}
pub fn as_pulse(&self) -> &dyn Pulse {
match &self.0 {
Some(p) => &**p,
None => &NoPulse,
}
}
#[track_caller]
pub fn plan(&self, execution: Execution, parts: &[PhaseSpec<'_>]) -> Vec<Child<'_>> {
self.as_pulse().plan(execution, parts)
}
}
impl fmt::Debug for SharedPulse {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("SharedPulse")
.field("live", &self.0.is_some())
.finish()
}
}
impl Stop for SharedPulse {
#[track_caller]
fn check(&self) -> Result<(), StopReason> {
self.as_pulse().check()
}
fn may_stop(&self) -> bool {
self.as_pulse().may_stop()
}
}
impl Report for SharedPulse {
#[track_caller]
fn advance(&self, n: u64) {
self.as_pulse().advance(n)
}
fn may_report(&self) -> bool {
self.as_pulse().may_report()
}
}
impl Pulse for SharedPulse {
#[track_caller]
fn record_issue(&self, error: PlanError) {
self.as_pulse().record_issue(error);
}
fn split(
&self,
execution: Execution,
parts: &[PhaseSpec<'_>],
) -> Result<Vec<Child<'_>>, PlanError> {
self.as_pulse().split(execution, parts)
}
fn start(&self) -> Result<(), PlanError> {
self.as_pulse().start()
}
fn set_total(&self, total: Total) -> Result<(), PlanError> {
self.as_pulse().set_total(total)
}
fn share(&self) -> Result<SharedPulse, PlanError> {
Ok(self.clone())
}
}