photon-ring 3.0.0

// Copyright 2026 Photon Ring Contributors
// SPDX-License-Identifier: MIT OR Apache-2.0

extern crate std;

use alloc::sync::Arc;
use alloc::vec::Vec;
use core::sync::atomic::{AtomicBool, AtomicU8, Ordering};
use std::thread::JoinHandle;

use super::{STAGE_PANICKED, STAGE_RUNNING};

/// A multi-stage processing pipeline.
///
/// Each stage reads from one Photon Ring channel, applies a transformation,
/// and publishes to the next channel. Stages run on dedicated threads.
///
/// Created via [`Pipeline::builder`] ->
/// [`PipelineBuilder::input`](super::builder::PipelineBuilder::input) ->
/// [`StageBuilder::then`](super::builder::StageBuilder::then) /
/// [`StageBuilder::fan_out`](super::builder::StageBuilder::fan_out) ->
/// [`StageBuilder::build`](super::builder::StageBuilder::build).
pub struct Pipeline {
    pub(super) handles: Vec<JoinHandle<()>>,
    pub(super) shutdown: Arc<AtomicBool>,
    pub(super) statuses: Vec<Arc<AtomicU8>>,
}

impl Pipeline {
    /// Create a new pipeline builder.
    pub fn builder() -> super::builder::PipelineBuilder {
        super::builder::PipelineBuilder::new()
    }

    /// Signal all stages to shut down gracefully.
    ///
    /// Stages will finish processing their current item and then exit.
    /// Call [`join`](Pipeline::join) to wait for them to complete.
    pub fn shutdown(&self) {
        self.shutdown.store(true, Ordering::Release);
    }

    /// Wait for all stage threads to finish.
    ///
    /// This consumes the pipeline. Call [`shutdown`](Pipeline::shutdown)
    /// first, or the threads may run indefinitely.
    ///
    /// Stage panics are captured at the thread boundary (a panicking stage
    /// records its status and exits), so joining never fails. Detect failures
    /// with [`panicked_stages`](Pipeline::panicked_stages) or
    /// [`is_healthy`](Pipeline::is_healthy), or use
    /// [`try_join`](Pipeline::try_join) to get the panicked indices back.
    pub fn join(mut self) {
        for h in core::mem::take(&mut self.handles) {
            let _ = h.join();
        }
    }

    /// Wait for all stage threads to finish, reporting which stages panicked.
    ///
    /// Returns `Ok(())` if every stage completed cleanly, or `Err(payload)`
    /// where `payload` downcasts to a `Vec<usize>` of the panicked stage
    /// indices (ascending). Because panics are captured at the thread
    /// boundary, the original panic payload is not recoverable — the indices
    /// identify which stage closures failed.
    pub fn try_join(mut self) -> Result<(), alloc::boxed::Box<dyn core::any::Any + Send>> {
        for h in core::mem::take(&mut self.handles) {
            let _ = h.join();
        }
        let panicked = self.panicked_stages();
        if panicked.is_empty() {
            Ok(())
        } else {
            Err(alloc::boxed::Box::new(panicked))
        }
    }

    /// Return the indices of stages that have panicked.
    ///
    /// Returns an empty vec if all stages are healthy.
    pub fn panicked_stages(&self) -> Vec<usize> {
        self.statuses
            .iter()
            .enumerate()
            .filter(|(_, s)| s.load(Ordering::Acquire) == STAGE_PANICKED)
            .map(|(i, _)| i)
            .collect()
    }

    /// Returns `true` if all stages are still running (none have panicked
    /// or completed).
    pub fn is_healthy(&self) -> bool {
        self.statuses
            .iter()
            .all(|s| s.load(Ordering::Acquire) == STAGE_RUNNING)
    }

    /// Number of stages in this pipeline.
    pub fn stage_count(&self) -> usize {
        self.statuses.len()
    }
}

/// Signals shutdown on drop. Call `join()` before dropping for a clean shutdown.
impl Drop for Pipeline {
    fn drop(&mut self) {
        self.shutdown.store(true, Ordering::Release);
    }
}