sketchbook 0.0.2

use core::any::Any;
use std::sync::mpsc::{self, SendError};
use std::sync::mpsc::{Receiver, Sender};
use std::thread::spawn;
use std::{
    marker::PhantomData,
    sync::{atomic::AtomicBool, Arc, Weak},
    thread::JoinHandle,
    time::Duration,
};

use crate::book::*;
use crate::env::*;
use crate::ream::*;
use crate::*;

/// Value that can be stored and then made into the sketch type later.
struct ParallelCreate<S, I> {
    input: I,
    marker: PhantomData<fn() -> S>,
}

/// Type erased creation of boxed sketches.
trait CreateItem<E>
where
    E: Environment,
    Self: Send,
{
    fn create(self, mill: &mut MillOf<E>) -> ParallelItem;
    fn create_from_box(self: Box<Self>, mill: &mut MillOf<E>) -> ParallelItem;
    fn into_input(self: Box<Self>) -> Box<dyn Any>;
}

impl<S, I> CreateItem<S::Env> for ParallelCreate<S, I>
where
    S: Sketch + Setup<I> + 'static,
    I: Send + 'static,
    MillOf<S::Env>: Clone + Send,
{
    fn create(self, mill: &mut MillOf<S::Env>) -> ParallelItem {
        let input = self.input;

        // we need to clone it to allow sending it to the thread
        let mill = mill.clone();

        // thread stop detection
        let full_alive_arc = Arc::new(AtomicBool::new(true));
        let alive_arc = Arc::downgrade(&full_alive_arc);

        // spawn thread for sketch to run on
        let handle = spawn(move || parallel_run::<S, I>(mill, input, full_alive_arc));

        ParallelItem { alive_arc, handle }
    }

    fn create_from_box(self: Box<Self>, mill: &mut MillOf<S::Env>) -> ParallelItem {
        self.create(mill)
    }

    fn into_input(self: Box<Self>) -> Box<dyn Any> {
        Box::new(self.input)
    }
}

/// Stop detection flag and thread handle to running sketch.
struct ParallelItem {
    alive_arc: Weak<AtomicBool>,
    handle: JoinHandle<()>,
}

/// Parallel ream.
///
/// Sketches in this ream are run in parallel.
/// This requires the use of threads. Each sketch will run on an independant thread.
///
/// ```
/// use sketchbook::*;
/// use sketchbook::ream::*;
///
/// #[apply(derive_sketch)]
/// #[sketch(env=minimal)]
/// struct App1 {
///     #[page]
///     page: minimal::Page,
/// }
///
/// impl Setup for App1 {
///     fn setup(page: minimal::Page, _: ()) -> Self {
///         Self { page }
///     }
/// }
///
/// #[apply(derive_sketch)]
/// #[sketch(env=minimal)]
/// struct App2 {
///     #[page]
///     page: minimal::Page,
/// }
///
/// impl Setup<ParallelProxy<minimal::Env>> for App2 {
///     fn setup(page: minimal::Page, proxy: ParallelProxy<minimal::Env>) -> Self {
///         // add an extra page after this sketch gets created
///         proxy.add_page::<App1>();
///     
///         Self { page }
///     }
/// }
///
/// // create book and add pages
/// let mut pages = Book::<Parallel<_>>::pages();
/// pages.add_page::<App1>();
/// pages.add_page_with::<App2, _>(pages.ream_proxy());
/// pages.bind();
/// ```
#[cfg_attr(docsrs, doc(cfg(feature = "ream-parallel")))]
pub struct Parallel<E> {
    to_create_send: Sender<Box<dyn CreateItem<E>>>,
    to_create_recv: Receiver<Box<dyn CreateItem<E>>>,
    to_creates: Vec<Box<dyn CreateItem<E>>>,
    items: Vec<ParallelItem>,
    finished: Vec<usize>,
}

impl<E> core::fmt::Debug for Parallel<E> {
    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
        f.debug_struct("Parallel").finish()
    }
}

impl<E> Default for Parallel<E>
where
    E: Environment,
{
    fn default() -> Self {
        let (to_create_send, to_create_recv) = mpsc::channel();
        Self {
            to_create_send,
            to_create_recv,
            items: Vec::new(),
            to_creates: Vec::new(),
            finished: Vec::new(),
        }
    }
}

impl<E> Ream for Parallel<E>
where
    E: Environment,
{
    type Env = E;

    type Error = Box<dyn Any + Send + 'static>;

    fn update(&mut self, mill: &mut MillOf<E>) -> Result<Status, Self::Error> {
        // create sketches that have been requested
        if !self.to_creates.is_empty() {
            for to_create in self.to_creates.drain(..) {
                self.items.push(to_create.create_from_box(mill));
            }
        }

        // create sketches from proxies
        use std::sync::mpsc::TryRecvError::*;
        loop {
            match self.to_create_recv.try_recv() {
                Ok(to_create) => {
                    self.items.push(to_create.create_from_box(mill));
                }
                Err(Empty) => {
                    break;
                }
                Err(Disconnected) => {
                    panic!("something happend to channel")
                }
            }
        }

        // check for threads that have stopped
        for (i, item) in self.items.iter().enumerate() {
            if item.alive_arc.upgrade().is_none() {
                self.finished.push(i);
            }
        }

        // remove the finished sketches
        for i in self.finished.drain(..) {
            let item = self.items.swap_remove(i);
            item.handle.join()?;
        }

        if self.items.is_empty() {
            Ok(Status::Stop)
        } else {
            Ok(Status::Continue)
        }
    }
}

impl<E> Parallel<E>
where
    E: Environment,
{
    /// Create proxy for parallel ream.
    pub fn create_proxy(&self) -> ParallelProxy<E> {
        ParallelProxy {
            to_create_send: self.to_create_send.clone(),
        }
    }
}

/// Proxy for a parallel ream.
///
/// This type allows for adding pages to a parallel ream from other threads.
pub struct ParallelProxy<E> {
    to_create_send: Sender<Box<dyn CreateItem<E>>>,
}

impl<E> core::fmt::Debug for ParallelProxy<E> {
    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
        f.debug_struct("ParallelProxy").finish()
    }
}

impl<E> Clone for ParallelProxy<E> {
    fn clone(&self) -> Self {
        Self {
            to_create_send: self.to_create_send.clone(),
        }
    }
}

impl<E> ParallelProxy<E>
where
    E: Environment,
    MillOf<E>: Clone + Send,
{
    /// Add page.
    ///
    /// If the ream has been dropped then the page will not be created.
    /// The returned flag indicates if the page was actually created.
    pub fn add_page<S>(&self) -> bool
    where
        S: Sketch<Env = E> + Setup<()> + 'static,
    {
        self.add_page_with::<S, ()>(()).is_none()
    }

    /// Add page with input data.
    ///
    /// If the ream has been dropped then the page will not be created.
    /// The input value will be returned if this happens.
    pub fn add_page_with<S, I>(&self, input: I) -> Option<I>
    where
        S: Sketch<Env = E> + Setup<I> + 'static,
        I: Send + 'static,
    {
        match self.to_create_send
            .send(Box::new(ParallelCreate::<S, I> {
                marker: PhantomData,
                input,
            })) {
            Ok(_) => None,
            Err(SendError(create_item)) => {
                match create_item.into_input().downcast::<I>() {
                    Ok(boxed) => Some(*boxed),
                    Err(_) => unreachable!("Unable to downcast input back into it's original type. This should not be possible."),
                }
            }
        }
    }
}

/// Extension trait for parallel ream.
pub trait ParallelPagesExt<E>
where
    E: Environment,
{
    /// Get a proxy to the ream.
    ///
    /// This allows adding pages from inside sketches.
    fn ream_proxy(&self) -> ParallelProxy<E>;

    /// Add a page.
    fn add_page<S>(&mut self)
    where
        S: Sketch<Env = E> + Setup<()> + 'static;

    /// Add a page with input data.
    fn add_page_with<S, I>(&mut self, input: I)
    where
        S: Sketch<Env = E> + Setup<I> + 'static,
        I: Send + 'static;
}

// runner that is the thread main
fn parallel_run<S, I>(mut mill: MillOf<S::Env>, input: I, alive_arc: Arc<AtomicBool>)
where
    S: Setup<I>,
    MillOf<S::Env>: Mill,
{
    // create sketch local to thread
    // this makes it possible for the sketch to not be send
    let mut sketch = S::setup(mill.new_page(), input);

    let mut group_count: usize = 0;

    while let Status::Continue = sketch.get_page_mut().status() {
        group_count += 1;

        // process group of events
        sketch.get_page_mut().start_group();
        while let Some(event) = sketch.get_page_mut().next_event_in_group() {
            sketch.get_page_mut().before_event(&event);
            sketch.handle_environment_event(&event);
            sketch.get_page_mut().finish_event(event);
        }
        sketch.get_page_mut().end_group();

        if group_count > 100 {
            // force yield to stop 100% CPU usage
            std::thread::sleep(Duration::from_millis(1));
        } else {
            // soft yield as now is a good time to switch CPU tasks
            std::thread::yield_now();
        }
    }

    drop(alive_arc);
}

impl<E> ParallelPagesExt<E> for Pages<Parallel<E>>
where
    E: Environment,
    MillOf<E>: Clone + Send,
{
    fn ream_proxy(&self) -> ParallelProxy<E> {
        self.ream.create_proxy()
    }

    fn add_page<S>(&mut self)
    where
        S: Sketch<Env = E> + Setup<()> + 'static,
    {
        self.add_page_with::<S, ()>(())
    }

    fn add_page_with<S, I>(&mut self, input: I)
    where
        S: Sketch<Env = E> + Setup<I> + 'static,
        I: Send + 'static,
    {
        self.ream.to_creates.push(Box::new(ParallelCreate::<S, I> {
            marker: PhantomData,
            input,
        }));
    }
}