palladium/async_runtime/

mod.rs

// Async runtime for Palladium
// "Orchestrating concurrent legends"

use std::collections::VecDeque;
use std::sync::{Arc, Mutex};
use std::thread;
use std::time::Duration;

/// Future trait for asynchronous computations
pub trait Future {
    type Output;

    /// Poll the future to check if it's ready
    fn poll(&mut self) -> Poll<Self::Output>;
}

/// Result of polling a future
pub enum Poll<T> {
    /// Future is ready with a value
    Ready(T),
    /// Future is not ready, should be polled again later
    Pending,
}

/// Task represents an asynchronous computation
pub struct Task {
    id: usize,
    poll_fn: Box<dyn FnMut() -> Poll<()> + Send>,
}

/// Runtime for executing async tasks
pub struct AsyncRuntime {
    /// Queue of tasks ready to be polled
    ready_queue: Arc<Mutex<VecDeque<Task>>>,
    /// Number of worker threads
    num_workers: usize,
    /// Whether the runtime is running
    running: Arc<Mutex<bool>>,
}

impl AsyncRuntime {
    /// Create a new async runtime
    pub fn new(num_workers: usize) -> Self {
        Self {
            ready_queue: Arc::new(Mutex::new(VecDeque::new())),
            num_workers,
            running: Arc::new(Mutex::new(false)),
        }
    }

    /// Spawn a new async task
    pub fn spawn<F>(&self, mut future: F) -> TaskHandle
    where
        F: Future<Output = ()> + Send + 'static,
    {
        static TASK_ID_COUNTER: std::sync::atomic::AtomicUsize =
            std::sync::atomic::AtomicUsize::new(0);
        let id = TASK_ID_COUNTER.fetch_add(1, std::sync::atomic::Ordering::SeqCst);

        let task = Task {
            id,
            poll_fn: Box::new(move || future.poll()),
        };

        self.ready_queue.lock().unwrap().push_back(task);

        TaskHandle { id }
    }

    /// Run the async runtime
    pub fn run(&self) {
        *self.running.lock().unwrap() = true;

        let mut workers = Vec::new();

        for worker_id in 0..self.num_workers {
            let ready_queue = Arc::clone(&self.ready_queue);
            let running = Arc::clone(&self.running);

            let handle = thread::spawn(move || {
                worker_loop(worker_id, ready_queue, running);
            });

            workers.push(handle);
        }

        // Wait for all workers to finish
        for worker in workers {
            worker.join().unwrap();
        }
    }

    /// Stop the runtime
    pub fn stop(&self) {
        *self.running.lock().unwrap() = false;
    }
}

/// Worker loop for processing tasks
fn worker_loop(
    worker_id: usize,
    ready_queue: Arc<Mutex<VecDeque<Task>>>,
    running: Arc<Mutex<bool>>,
) {
    loop {
        // Check if we should stop
        if !*running.lock().unwrap() {
            break;
        }

        // Try to get a task from the queue
        let task = {
            let mut queue = ready_queue.lock().unwrap();
            queue.pop_front()
        };

        if let Some(mut task) = task {
            // Poll the task
            match (task.poll_fn)() {
                Poll::Ready(()) => {
                    // Task completed
                    println!("Worker {}: Task {} completed", worker_id, task.id);
                }
                Poll::Pending => {
                    // Task not ready, put it back in the queue
                    ready_queue.lock().unwrap().push_back(task);
                }
            }
        } else {
            // No tasks available, sleep briefly
            thread::sleep(Duration::from_millis(10));
        }
    }
}

/// Handle to a spawned task
pub struct TaskHandle {
    #[allow(dead_code)]
    id: usize,
}

/// Simple implementation of an async sleep
pub struct Sleep {
    deadline: std::time::Instant,
}

impl Sleep {
    pub fn new(duration: Duration) -> Self {
        Self {
            deadline: std::time::Instant::now() + duration,
        }
    }
}

impl Future for Sleep {
    type Output = ();

    fn poll(&mut self) -> Poll<Self::Output> {
        if std::time::Instant::now() >= self.deadline {
            Poll::Ready(())
        } else {
            Poll::Pending
        }
    }
}

/// Channel for async communication
pub struct Channel<T> {
    queue: Arc<Mutex<VecDeque<T>>>,
}

impl<T> Default for Channel<T> {
    fn default() -> Self {
        Self {
            queue: Arc::new(Mutex::new(VecDeque::new())),
        }
    }
}

impl<T> Channel<T> {
    pub fn new() -> Self {
        Self::default()
    }

    pub fn sender(&self) -> Sender<T> {
        Sender {
            queue: Arc::clone(&self.queue),
        }
    }

    pub fn receiver(&self) -> Receiver<T> {
        Receiver {
            queue: Arc::clone(&self.queue),
        }
    }
}

/// Sender end of a channel
pub struct Sender<T> {
    queue: Arc<Mutex<VecDeque<T>>>,
}

impl<T> Sender<T> {
    pub fn send(&self, value: T) {
        self.queue.lock().unwrap().push_back(value);
    }
}

/// Receiver end of a channel
pub struct Receiver<T> {
    queue: Arc<Mutex<VecDeque<T>>>,
}

impl<T> Receiver<T> {
    pub fn try_recv(&self) -> Option<T> {
        self.queue.lock().unwrap().pop_front()
    }
}

/// Future for receiving from a channel
pub struct RecvFuture<T> {
    receiver: Receiver<T>,
}

impl<T> RecvFuture<T> {
    pub fn new(receiver: Receiver<T>) -> Self {
        Self { receiver }
    }
}

impl<T> Future for RecvFuture<T> {
    type Output = Option<T>;

    fn poll(&mut self) -> Poll<Self::Output> {
        if let Some(value) = self.receiver.try_recv() {
            Poll::Ready(Some(value))
        } else {
            Poll::Pending
        }
    }
}

/// Async I/O operations
pub mod io {
    use super::*;
    use std::fs;
    use std::io;
    use std::path::Path;

    /// Async file read
    pub struct ReadFile {
        path: String,
        state: ReadFileState,
    }

    enum ReadFileState {
        NotStarted,
        Reading,
        Done(io::Result<String>),
    }

    impl ReadFile {
        pub fn new(path: impl AsRef<Path>) -> Self {
            Self {
                path: path.as_ref().to_str().unwrap().to_string(),
                state: ReadFileState::NotStarted,
            }
        }
    }

    impl Future for ReadFile {
        type Output = io::Result<String>;

        fn poll(&mut self) -> Poll<Self::Output> {
            match &mut self.state {
                ReadFileState::NotStarted => {
                    // Start reading in a background thread
                    let path = self.path.clone();
                    thread::spawn(move || fs::read_to_string(path));
                    self.state = ReadFileState::Reading;
                    Poll::Pending
                }
                ReadFileState::Reading => {
                    // In a real implementation, we'd check if the thread is done
                    // For now, we'll do a blocking read
                    let result = fs::read_to_string(&self.path);
                    self.state = ReadFileState::Done(result);
                    Poll::Pending
                }
                ReadFileState::Done(result) => {
                    // Clone the result to return it
                    match result {
                        Ok(content) => Poll::Ready(Ok(content.clone())),
                        Err(e) => Poll::Ready(Err(io::Error::new(e.kind(), e.to_string()))),
                    }
                }
            }
        }
    }

    /// Async file write
    pub struct WriteFile {
        path: String,
        content: String,
        state: WriteFileState,
    }

    enum WriteFileState {
        NotStarted,
        #[allow(dead_code)]
        Writing,
        Done(io::Result<()>),
    }

    impl WriteFile {
        pub fn new(path: impl AsRef<Path>, content: String) -> Self {
            Self {
                path: path.as_ref().to_str().unwrap().to_string(),
                content,
                state: WriteFileState::NotStarted,
            }
        }
    }

    impl Future for WriteFile {
        type Output = io::Result<()>;

        fn poll(&mut self) -> Poll<Self::Output> {
            match &mut self.state {
                WriteFileState::NotStarted => {
                    // In a real implementation, this would be async
                    let result = fs::write(&self.path, &self.content);
                    self.state = WriteFileState::Done(result);
                    Poll::Pending
                }
                WriteFileState::Writing => Poll::Pending,
                WriteFileState::Done(result) => match result {
                    Ok(()) => Poll::Ready(Ok(())),
                    Err(e) => Poll::Ready(Err(io::Error::new(e.kind(), e.to_string()))),
                },
            }
        }
    }
}

/// Combinators for futures
pub mod combinators {
    use super::*;

    /// Map combinator
    pub struct Map<F, U, G> {
        future: F,
        mapper: Option<G>,
        _phantom: std::marker::PhantomData<U>,
    }

    impl<F, U, G> Map<F, U, G>
    where
        F: Future,
        G: FnOnce(F::Output) -> U,
    {
        pub fn new(future: F, mapper: G) -> Self {
            Self {
                future,
                mapper: Some(mapper),
                _phantom: std::marker::PhantomData,
            }
        }
    }

    impl<F, U, G> Future for Map<F, U, G>
    where
        F: Future,
        G: FnOnce(F::Output) -> U,
    {
        type Output = U;

        fn poll(&mut self) -> Poll<Self::Output> {
            match self.future.poll() {
                Poll::Ready(value) => {
                    let mapper = self.mapper.take().expect("Map polled after completion");
                    Poll::Ready(mapper(value))
                }
                Poll::Pending => Poll::Pending,
            }
        }
    }

    /// Join two futures
    pub struct Join<F1, F2, O1, O2>
    where
        F1: Future<Output = O1>,
        F2: Future<Output = O2>,
    {
        future1: Option<F1>,
        future2: Option<F2>,
        output1: Option<O1>,
        output2: Option<O2>,
    }

    impl<F1, F2, O1, O2> Join<F1, F2, O1, O2>
    where
        F1: Future<Output = O1>,
        F2: Future<Output = O2>,
    {
        pub fn new(future1: F1, future2: F2) -> Self {
            Self {
                future1: Some(future1),
                future2: Some(future2),
                output1: None,
                output2: None,
            }
        }
    }

    impl<F1, F2, O1, O2> Future for Join<F1, F2, O1, O2>
    where
        F1: Future<Output = O1>,
        F2: Future<Output = O2>,
    {
        type Output = (O1, O2);

        fn poll(&mut self) -> Poll<Self::Output> {
            // Poll first future if not complete
            if self.output1.is_none() {
                if let Some(ref mut f1) = self.future1 {
                    if let Poll::Ready(value) = f1.poll() {
                        self.output1 = Some(value);
                        self.future1 = None;
                    }
                }
            }

            // Poll second future if not complete
            if self.output2.is_none() {
                if let Some(ref mut f2) = self.future2 {
                    if let Poll::Ready(value) = f2.poll() {
                        self.output2 = Some(value);
                        self.future2 = None;
                    }
                }
            }

            // Check if both are complete
            if let (Some(v1), Some(v2)) = (self.output1.take(), self.output2.take()) {
                Poll::Ready((v1, v2))
            } else {
                // Restore values if we took them
                if let Some(v1) = self.output1.take() {
                    self.output1 = Some(v1);
                }
                if let Some(v2) = self.output2.take() {
                    self.output2 = Some(v2);
                }
                Poll::Pending
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_sleep_future() {
        let mut sleep = Sleep::new(Duration::from_millis(100));

        // Should be pending initially
        match sleep.poll() {
            Poll::Pending => {}
            Poll::Ready(()) => panic!("Sleep should not be ready immediately"),
        }

        // Wait and poll again
        thread::sleep(Duration::from_millis(150));
        match sleep.poll() {
            Poll::Ready(()) => {}
            Poll::Pending => panic!("Sleep should be ready after deadline"),
        }
    }

    #[test]
    fn test_channel() {
        let channel = Channel::<i32>::new();
        let sender = channel.sender();
        let receiver = channel.receiver();

        // Send some values
        sender.send(42);
        sender.send(100);

        // Receive values
        assert_eq!(receiver.try_recv(), Some(42));
        assert_eq!(receiver.try_recv(), Some(100));
        assert_eq!(receiver.try_recv(), None);
    }
}