use crate::{
Execution, Outcome, PhaseSpec, PlanError, Report, Total,
sync::{Counter, MetadataCell},
};
use alloc::{boxed::Box, string::String, sync::Arc, vec::Vec};
use core::{
fmt,
sync::atomic::{AtomicBool, AtomicU8, AtomicUsize, Ordering},
};
#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub struct NodeId(usize);
impl NodeId {
pub const ROOT: Self = Self(0);
pub const fn get(self) -> usize {
self.0
}
}
impl fmt::Display for NodeId {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
self.0.fmt(f)
}
}
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
#[non_exhaustive]
pub enum Status {
#[default]
NotStarted,
Running,
Finished(Outcome),
}
#[derive(Clone, Copy, Debug, PartialEq)]
#[non_exhaustive]
pub struct Summary {
pub fraction: Option<f64>,
pub unresolved_fraction: f64,
pub status: Status,
}
fn credited(status: Status) -> bool {
matches!(
status,
Status::Finished(Outcome::Succeeded | Outcome::Skipped | Outcome::NotRun)
)
}
fn leaf_fraction(status: Status, total: Total, completed: u64, overflowed: bool) -> Option<f64> {
match (status, total) {
(Status::Finished(Outcome::Succeeded | Outcome::Skipped), _) => Some(1.0),
(Status::Finished(Outcome::NotRun), _) => Some(0.0),
_ if overflowed => None,
(_, Total::Exact(0) | Total::Estimated(0)) => Some(0.0),
(_, Total::Exact(n) | Total::Estimated(n)) => Some((completed as f64 / n as f64).min(1.0)),
_ => None,
}
}
#[derive(Clone)]
struct Metadata {
units: Name,
total: Total,
revisions: Vec<Total>,
execution: Execution,
children: Vec<Arc<Node>>,
frozen: Option<Snapshot>,
}
const UNKNOWN: u8 = u8::MAX;
fn encode(outcome: Outcome) -> u8 {
match outcome {
Outcome::Succeeded => 0,
Outcome::Skipped => 1,
Outcome::Cancelled => 2,
Outcome::Failed => 3,
Outcome::Abandoned => 4,
Outcome::NotRun => 5,
_ => UNKNOWN,
}
}
fn decode(code: u8) -> Option<Outcome> {
match code {
0 => Some(Outcome::Succeeded),
1 => Some(Outcome::Skipped),
2 => Some(Outcome::Cancelled),
3 => Some(Outcome::Failed),
4 => Some(Outcome::Abandoned),
5 => Some(Outcome::NotRun),
_ => None,
}
}
const MAX_REVISIONS: usize = 16;
#[derive(Clone)]
enum Name {
Inline { len: u8, bytes: [u8; INLINE_NAME] },
Heap(Box<str>),
}
const INLINE_NAME: usize = 22;
impl Name {
fn new(name: &str) -> Self {
match name.len() {
len @ ..=INLINE_NAME => {
let mut bytes = [0; INLINE_NAME];
bytes[..len].copy_from_slice(name.as_bytes());
Self::Inline {
len: len as u8,
bytes,
}
}
_ => Self::Heap(name.into()),
}
}
fn as_str(&self) -> &str {
match self {
Self::Inline { len, bytes } => {
core::str::from_utf8(&bytes[..usize::from(*len)]).unwrap_or_default()
}
Self::Heap(name) => name,
}
}
}
struct Node {
id: NodeId,
parent: Option<NodeId>,
name: Name,
weight: u64,
initial_total: Total,
next_id: Arc<AtomicUsize>,
issued: AtomicBool,
branch: AtomicBool,
state: AtomicU8,
outcome: AtomicU8,
inferred: AtomicBool,
completed: Counter,
final_count: Counter,
meta: MetadataCell<Metadata>,
}
impl Node {
fn new(
id: NodeId,
parent: Option<NodeId>,
spec: &PhaseSpec<'_>,
next_id: Arc<AtomicUsize>,
) -> Self {
Self {
id,
parent,
name: Name::new(spec.name),
weight: spec.weight,
initial_total: spec.total,
next_id,
issued: AtomicBool::new(false),
branch: AtomicBool::new(false),
state: AtomicU8::new(0),
outcome: AtomicU8::new(UNKNOWN),
inferred: AtomicBool::new(false),
completed: Counter::new(),
final_count: Counter::new(),
meta: MetadataCell::new(Metadata {
units: Name::new(spec.units),
total: spec.total,
revisions: Vec::new(),
execution: spec.execution,
children: Vec::new(),
frozen: None,
}),
}
}
fn snapshot(&self) -> Snapshot {
self.snapshot_with(false)
.expect("blocking metadata read succeeds")
}
fn read(&self, nonblocking: bool) -> Option<(Status, &Counter, Arc<Metadata>)> {
let state = self.state.load(Ordering::Acquire);
let meta = if nonblocking {
self.meta.try_get()?
} else {
self.meta.get()
};
Some(match state {
0 => (Status::NotStarted, &self.completed, meta),
1 => (Status::Running, &self.completed, meta),
_ => {
let outcome = decode(self.outcome.load(Ordering::Relaxed));
let status = Status::Finished(outcome.unwrap_or(Outcome::Abandoned));
(status, &self.final_count, meta)
}
})
}
fn snapshot_with(&self, nonblocking: bool) -> Option<Snapshot> {
let (status, counter, meta) = self.read(nonblocking)?;
if let Some(frozen) = &meta.frozen {
return Some(frozen.clone());
}
let mut result = Snapshot {
id: self.id,
parent: self.parent,
name: self.name.as_str().into(),
weight: self.weight,
units: meta.units.as_str().into(),
total: meta.total,
initial_total: self.initial_total,
total_revisions: meta.revisions.clone(),
execution: meta.execution,
status,
completed: counter.get(),
overflowed: counter.overflowed(),
completion_inferred: self.inferred.load(Ordering::Relaxed),
children: Vec::with_capacity(meta.children.len()),
};
for child in &meta.children {
let child = child.snapshot_with(nonblocking)?;
if result.status == Status::NotStarted && child.status != Status::NotStarted {
result.status = Status::Running;
}
result.children.push(child);
}
Some(result)
}
fn summary(&self, nonblocking: bool) -> Option<Summary> {
let (status, counter, meta) = self.read(nonblocking)?;
if let Some(frozen) = &meta.frozen {
return Some(Summary {
fraction: frozen.fraction(),
unresolved_fraction: frozen.unresolved_fraction(),
status: frozen.status,
});
}
let fraction = leaf_fraction(status, meta.total, counter.get(), counter.overflowed());
let mut result = Summary {
fraction,
unresolved_fraction: if fraction.is_some() { 0.0 } else { 1.0 },
status,
};
if meta.children.is_empty() {
return Some(result);
}
let mut sum = 0.0;
for child in &meta.children {
sum += child.weight as f64;
}
let mut fraction = 0.0;
let mut unresolved = 0.0;
for child in &meta.children {
let observed = child.summary(nonblocking)?;
if result.status == Status::NotStarted && observed.status != Status::NotStarted {
result.status = Status::Running;
}
let weight = child.weight as f64 / sum;
fraction += weight * observed.fraction.unwrap_or(0.0);
unresolved += weight * observed.unresolved_fraction;
}
if !credited(status) {
result.fraction = Some(fraction.clamp(0.0, 1.0));
result.unresolved_fraction = unresolved;
}
Some(result)
}
fn finished(&self) -> bool {
self.state.load(Ordering::Acquire) == 2
}
}
#[must_use = "finish the phase after its work joins; dropping it unfinished records Abandoned"]
pub struct Phase {
node: Arc<Node>,
}
impl Phase {
pub fn new(name: impl Into<String>, total: Total) -> Self {
let name: String = name.into();
Self {
node: Arc::new(Node::new(
NodeId::ROOT,
None,
&PhaseSpec::new(&name, 1, total),
Arc::new(AtomicUsize::new(1)),
)),
}
}
pub fn reporter(&self) -> Reporter {
self.node.issued.store(true, Ordering::Relaxed);
self.deferred_reporter()
}
pub(crate) fn deferred_reporter(&self) -> Reporter {
Reporter {
node: Arc::clone(&self.node),
}
}
pub fn observer(&self) -> Observer {
Observer {
node: Arc::clone(&self.node),
}
}
pub fn id(&self) -> NodeId {
self.node.id
}
pub fn start(&mut self) -> Result<(), PlanError> {
self.ensure_live()?;
self.node.state.store(1, Ordering::Release);
Ok(())
}
pub fn set_units(&mut self, units: impl Into<String>) -> Result<(), PlanError> {
self.ensure_unused()?;
let units: String = units.into();
let mut meta = (*self.node.meta.get()).clone();
meta.units = Name::new(&units);
self.node.meta.publish(meta);
Ok(())
}
pub fn set_execution(&mut self, execution: Execution) -> Result<(), PlanError> {
self.ensure_unused()?;
let mut meta = (*self.node.meta.get()).clone();
meta.execution = execution;
self.node.meta.publish(meta);
Ok(())
}
pub fn set_total(&mut self, total: Total) -> Result<(), PlanError> {
self.ensure_live()?;
if self.node.branch.load(Ordering::Relaxed) {
return Err(PlanError::NotALeaf);
}
let mut meta = (*self.node.meta.get()).clone();
if meta.total != total {
meta.total = total;
if meta.revisions.len() == MAX_REVISIONS {
meta.revisions.remove(0);
}
meta.revisions.push(total);
self.node.meta.publish(meta);
}
Ok(())
}
pub fn split<const N: usize>(
&mut self,
execution: Execution,
parts: [PhaseSpec<'_>; N],
) -> Result<[Phase; N], PlanError> {
match self.split_vec(execution, &parts)?.try_into() {
Ok(children) => Ok(children),
Err(_) => unreachable!("one child per part"),
}
}
pub fn split_vec(
&mut self,
execution: Execution,
parts: &[PhaseSpec<'_>],
) -> Result<Vec<Phase>, PlanError> {
let mut children = Vec::with_capacity(parts.len());
self.split_each(execution, parts, &mut |child| children.push(child))?;
Ok(children)
}
#[allow(deprecated)] pub(crate) fn split_each(
&mut self,
execution: Execution,
parts: &[PhaseSpec<'_>],
each: &mut dyn FnMut(Phase),
) -> Result<(), PlanError> {
self.ensure_unused()?;
PhaseSpec::validate_split(parts)?;
let first = self
.node
.next_id
.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |n| {
n.checked_add(parts.len())
})
.map_err(|_| PlanError::Overflow)?;
let mut nodes = Vec::with_capacity(parts.len());
for (i, part) in parts.iter().enumerate() {
nodes.push(Arc::new(Node::new(
NodeId(first + i),
Some(self.node.id),
part,
Arc::clone(&self.node.next_id),
)));
}
let mut meta = (*self.node.meta.get()).clone();
meta.execution = execution;
meta.total = Total::Unknown;
meta.revisions.clear();
for node in &nodes {
each(Phase {
node: Arc::clone(node),
});
}
meta.children = nodes;
self.node.meta.publish(meta);
self.node.branch.store(true, Ordering::Release);
Ok(())
}
pub fn finish(&mut self) -> Result<(), PlanError> {
self.finish_with(Outcome::Succeeded)
}
pub fn finish_with(&mut self, outcome: Outcome) -> Result<(), PlanError> {
self.ensure_live()?;
for child in &self.node.meta.get().children {
if !child.finished() {
return Err(PlanError::UnfinishedChildren);
}
}
self.seal(outcome);
Ok(())
}
pub(crate) fn record_outcome(&mut self, outcome: Outcome) -> Result<(), PlanError> {
self.ensure_live()?;
self.seal(outcome);
Ok(())
}
pub(crate) fn complete_inferred(&mut self, outcome: Outcome) -> Result<(), PlanError> {
self.ensure_live()?;
self.node.inferred.store(true, Ordering::Relaxed);
let outcome = if self
.node
.summary(false)
.is_some_and(|s| s.status == Status::Running)
{
Outcome::Abandoned
} else {
outcome
};
self.record_outcome(outcome)
}
fn seal(&mut self, outcome: Outcome) {
let code = encode(outcome);
let mut running_child = false;
for child in &self.node.meta.get().children {
if !child.finished() {
running_child = true;
}
}
if code == UNKNOWN || running_child {
let mut snapshot = self.node.snapshot();
resolve_children(&mut snapshot, outcome);
snapshot.status = Status::Finished(outcome);
let mut meta = (*self.node.meta.get()).clone();
meta.frozen = Some(snapshot);
self.node.meta.publish(meta);
}
self.node.final_count.copy_from(&self.node.completed);
self.node.outcome.store(code, Ordering::Relaxed);
self.node.state.store(2, Ordering::Release);
}
fn ensure_live(&self) -> Result<(), PlanError> {
if self.node.finished() {
Err(PlanError::Finished)
} else {
Ok(())
}
}
fn ensure_unused(&self) -> Result<(), PlanError> {
self.ensure_live()?;
if self.node.issued.load(Ordering::Relaxed) || self.node.branch.load(Ordering::Relaxed) {
Err(PlanError::AlreadyInUse)
} else {
Ok(())
}
}
}
impl Drop for Phase {
fn drop(&mut self) {
if !self.node.finished() {
self.seal(Outcome::Abandoned);
}
}
}
impl how_far::Complete for Phase {
fn complete_as(mut self, outcome: Outcome) {
let _ = self.record_outcome(outcome);
}
}
fn resolve_children(snapshot: &mut Snapshot, parent: Outcome) {
for child in &mut snapshot.children {
let outcome = match child.status {
Status::Finished(_) => continue,
Status::NotStarted => match parent {
Outcome::Succeeded | Outcome::Skipped => Outcome::Skipped,
Outcome::Abandoned => Outcome::Abandoned,
_ => Outcome::NotRun,
},
Status::Running => Outcome::Abandoned,
};
resolve_children(child, outcome);
child.status = Status::Finished(outcome);
child.completion_inferred = true;
}
}
#[derive(Clone)]
pub struct Reporter {
node: Arc<Node>,
}
impl fmt::Debug for Reporter {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("Reporter")
.field("phase", &self.node.id)
.finish_non_exhaustive()
}
}
impl Report for Reporter {
#[inline]
#[track_caller]
#[allow(clippy::collapsible_match)] fn advance(&self, completed: u64) {
if completed == 0 || self.node.branch.load(Ordering::Relaxed) {
return;
}
if !self.node.issued.load(Ordering::Relaxed) {
self.node.issued.store(true, Ordering::Relaxed);
}
match self.node.state.load(Ordering::Acquire) {
2 => return,
0 => {
if self
.node
.state
.compare_exchange(0, 1, Ordering::Relaxed, Ordering::Relaxed)
== Err(2)
{
return;
}
}
_ => {}
}
self.node.completed.add(completed);
}
fn may_report(&self) -> bool {
!self.node.branch.load(Ordering::Relaxed)
}
}
#[derive(Clone)]
pub struct Observer {
node: Arc<Node>,
}
impl fmt::Debug for Observer {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("Observer")
.field("phase", &self.node.id)
.finish_non_exhaustive()
}
}
impl Observer {
pub fn id(&self) -> NodeId {
self.node.id
}
pub fn summary(&self) -> Summary {
self.node.summary(false).expect("blocking read")
}
pub fn try_summary(&self) -> Option<Summary> {
self.node.summary(true)
}
pub fn name(&self) -> &str {
self.node.name.as_str()
}
pub fn snapshot(&self) -> Snapshot {
self.node.snapshot()
}
pub fn try_snapshot(&self) -> Option<Snapshot> {
self.node.snapshot_with(true)
}
pub fn is_finished(&self) -> bool {
self.node.finished()
}
pub fn children(&self) -> Vec<Observer> {
let meta = self.node.meta.get();
let mut children = Vec::with_capacity(meta.children.len());
for node in &meta.children {
children.push(Observer {
node: Arc::clone(node),
});
}
children
}
}
#[derive(Clone, Debug, PartialEq)]
#[non_exhaustive]
pub struct Snapshot {
pub id: NodeId,
pub parent: Option<NodeId>,
pub name: String,
pub weight: u64,
pub units: String,
pub total: Total,
pub initial_total: Total,
pub total_revisions: Vec<Total>,
pub execution: Execution,
pub status: Status,
pub completed: u64,
pub overflowed: bool,
pub completion_inferred: bool,
pub children: Vec<Snapshot>,
}
impl Snapshot {
pub fn write_json(&self, out: &mut impl fmt::Write) -> fmt::Result {
out.write_str("{\"schema_version\":2,\"root\":")?;
self.write_node_json(out)?;
out.write_char('}')
}
fn write_node_json(&self, out: &mut impl fmt::Write) -> fmt::Result {
use crate::json::{outcome_name, quote};
write!(out, "{{\"id\":{},\"parent\":", self.id)?;
match self.parent {
Some(id) => write!(out, "{id}")?,
None => out.write_str("null")?,
}
out.write_str(",\"name\":")?;
quote(out, &self.name)?;
write!(out, ",\"weight\":\"{}\",\"units\":", self.weight)?;
quote(out, &self.units)?;
out.write_str(",\"total\":")?;
write_total(out, self.total)?;
out.write_str(",\"initial_total\":")?;
write_total(out, self.initial_total)?;
out.write_str(",\"total_revisions\":[")?;
for (i, total) in self.total_revisions.iter().enumerate() {
if i > 0 {
out.write_char(',')?;
}
write_total(out, *total)?;
}
let (execution, capacity) = match self.execution {
Execution::Unspecified => ("Unspecified", None),
Execution::Sequence => ("Sequence", None),
Execution::ForkJoin => ("ForkJoin", None),
Execution::WorkPool {
max_parallelism, ..
} => ("WorkPool", Some(max_parallelism)),
_ => ("Other", None),
};
write!(out, "],\"execution\":\"{execution}\",\"max_parallelism\":")?;
match capacity {
Some(n) => write!(out, "{n}")?,
None => out.write_str("null")?,
}
let (status, outcome) = match self.status {
Status::NotStarted => ("NotStarted", None),
Status::Running => ("Running", None),
Status::Finished(outcome) => ("Finished", Some(outcome)),
};
write!(out, ",\"status\":\"{status}\",\"outcome\":")?;
match outcome {
Some(outcome) => write!(out, "\"{}\"", outcome_name(outcome))?,
None => out.write_str("null")?,
}
write!(out, ",\"completion_inferred\":{}", self.completion_inferred)?;
write!(
out,
",\"completed\":\"{}\",\"overflowed\":{},\"overrun\":{},\"fraction\":",
self.completed,
self.overflowed,
self.overrun()
)?;
match self.fraction() {
Some(fraction) => write!(out, "{fraction}")?,
None => out.write_str("null")?,
}
write!(
out,
",\"unresolved_fraction\":{},\"children\":[",
self.unresolved_fraction()
)?;
for (i, child) in self.children.iter().enumerate() {
if i > 0 {
out.write_char(',')?;
}
child.write_node_json(out)?;
}
out.write_str("]}")
}
pub const fn counter_max() -> u64 {
crate::sync::COUNT_MAX
}
pub fn fraction(&self) -> Option<f64> {
if self.children.is_empty() || credited(self.status) {
return leaf_fraction(self.status, self.total, self.completed, self.overflowed);
}
let sum = self.child_weight_sum();
let mut result = 0.0;
for child in &self.children {
result += child.weight as f64 / sum * child.fraction().unwrap_or(0.0);
}
Some(result.clamp(0.0, 1.0))
}
pub fn unresolved_fraction(&self) -> f64 {
if self.children.is_empty() {
return if self.fraction().is_some() { 0.0 } else { 1.0 };
}
if credited(self.status) {
return 0.0;
}
let sum = self.child_weight_sum();
let mut unresolved = 0.0;
for child in &self.children {
unresolved += child.weight as f64 / sum * child.unresolved_fraction();
}
unresolved
}
fn child_weight_sum(&self) -> f64 {
let mut sum = 0.0;
for child in &self.children {
sum += child.weight as f64;
}
sum
}
pub fn overrun(&self) -> bool {
matches!(self.total, Total::Exact(n) if self.completed > n) || self.overflowed
}
}
fn write_total(out: &mut impl fmt::Write, total: Total) -> fmt::Result {
match total {
Total::Unknown => out.write_str("{\"kind\":\"Unknown\"}"),
Total::Exact(n) => write!(out, "{{\"kind\":\"Exact\",\"count\":\"{n}\"}}"),
Total::Estimated(n) => write!(out, "{{\"kind\":\"Estimated\",\"count\":\"{n}\"}}"),
_ => out.write_str("{\"kind\":\"Other\"}"),
}
}
#[cfg(all(test, feature = "std"))]
mod tests {
use super::*;
use crate::Stop;
use crate::poll::LocalPoller;
use almost_enough::Stopper;
#[test]
fn a_report_racing_with_the_finish_is_not_shown() {
let mut leaf = Phase::new("leaf", Total::Exact(3));
leaf.reporter().advance(2);
leaf.finish().unwrap();
leaf.node.completed.add(5);
let snapshot = leaf.observer().snapshot();
assert_eq!(snapshot.completed, 2);
assert_eq!(snapshot.status, Status::Finished(Outcome::Succeeded));
}
#[test]
fn a_busy_child_skips_ui_snapshots_while_reports_callbacks_and_cancellation_continue() {
let mut root = Phase::new("root", Total::Unknown);
let [child] = root
.split(
Execution::Sequence,
[PhaseSpec::new("child", 1, Total::Exact(3))],
)
.unwrap();
let observer = root.observer();
let stop = Stopper::new();
let cancel = stop.clone();
let mut poller = LocalPoller::new(observer.clone());
poller.subscribe(move |event| {
assert!(event.try_snapshot().is_none());
assert!(!event.snapshot_materialized());
cancel.cancel();
});
child.node.meta.with_lock_for_test(|| {
child.reporter().advance(1);
assert!(observer.try_snapshot().is_none());
poller.poll();
assert!(stop.check().is_err());
});
assert_eq!(observer.try_snapshot().unwrap().children[0].completed, 1);
}
}