aura-gui 0.2.2

//! Futuros en el hilo principal: `cx.spawn(|this, cx| async move { … })`.
//!
//! Como en GPUI, una secuencia «esperar, calcular en segundo plano, actualizar»
//! se escribe de corrido con `.await` en vez de anidar callbacks. Los futuros no
//! tocan entidades directamente: `AsyncContext::update` encola el cambio en el
//! inbox y devuelve una `Task` con el resultado. No hace falta tokio: el
//! despertador de cada futuro reutiliza el `Waker` del bucle de eventos.

use crate::core::entity::{EntityContext, Handle};
use crate::core::task::Task;
use crate::core::tasks::{BackgroundExecutor, CancelToken, ForegroundSender};
use slotmap::{SlotMap, new_key_type};
use std::any::Any;
use std::future::Future;
use std::pin::Pin;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};
use std::task::{Context, Wake, Waker};
use std::time::Duration;

#[cfg(test)]
#[path = "executor_tests.rs"]
mod executor_tests;

type LocalFuture = Pin<Box<dyn Future<Output = ()>>>;
type ReadyList = Arc<Mutex<Vec<FutureId>>>;

new_key_type! { struct FutureId; }

struct LocalTask {
    future: LocalFuture,
    wake: Arc<FutureWaker>,
}

/// Futuros vivos del hilo principal; se sondean solo cuando alguien los despierta.
#[derive(Default)]
pub struct LocalExecutor {
    futures: SlotMap<FutureId, LocalTask>,
    ready: ReadyList,
}

struct FutureWaker {
    id: FutureId,
    ready: ReadyList,
    sender: ForegroundSender,
    queued: AtomicBool,
    active: AtomicBool,
}

impl Wake for FutureWaker {
    fn wake(self: Arc<Self>) {
        self.wake_by_ref();
    }

    fn wake_by_ref(self: &Arc<Self>) {
        if !self.active.load(Ordering::Acquire) || self.queued.swap(true, Ordering::AcqRel) {
            return;
        }
        self.ready.lock().unwrap().push(self.id);
        // Un solo trabajo por futuro listo, aunque reciba muchos wakes.
        self.sender.send(|_| {});
    }
}

impl LocalExecutor {
    pub fn spawn(&mut self, future: impl Future<Output = ()> + 'static, sender: &ForegroundSender) {
        let id = self.futures.insert_with_key(|id| LocalTask {
            future: Box::pin(future),
            wake: Arc::new(FutureWaker {
                id,
                ready: self.ready.clone(),
                sender: sender.clone(),
                queued: AtomicBool::new(false),
                active: AtomicBool::new(true),
            }),
        });
        self.futures[id].wake.wake_by_ref();
    }

    /// Sondea los futuros despertados; `true` si sondeó alguno.
    pub fn poll_ready(&mut self, _sender: &ForegroundSender) -> bool {
        let ids = self.ready.lock().map(|mut ready| std::mem::take(&mut *ready)).unwrap_or_default();
        let mut polled = false;
        for id in ids {
            let Some(task) = self.futures.get_mut(id) else { continue };
            task.wake.queued.store(false, Ordering::Release);
            let waker = Waker::from(task.wake.clone());
            polled = true;
            if task.future.as_mut().poll(&mut Context::from_waker(&waker)).is_ready() {
                task.wake.active.store(false, Ordering::Release);
                self.futures.remove(id);
            }
        }
        polled
    }

    pub fn live_futures(&self) -> usize {
        self.futures.len()
    }
}

impl Drop for LocalExecutor {
    fn drop(&mut self) {
        for task in self.futures.values() {
            task.wake.active.store(false, Ordering::Release);
        }
    }
}

/// Asa de un trabajo lanzado con [`EntityContext::job`]. Soltarlo no lo cancela; `cancel()` avisa al trabajo, que debe
/// mirar su `CancelToken`, y evita que su resultado llegue a la vista.
#[derive(Clone, Default)]
pub struct JobHandle {
    cancel: CancelToken,
}

impl JobHandle {
    pub fn cancel(&self) {
        self.cancel.cancel();
    }

    pub fn is_cancelled(&self) -> bool {
        self.cancel.is_cancelled()
    }
}

impl<T: Any> EntityContext<'_, T> {
    /// Ejecuta `work` en el pool de la app y, al acabar, `done` en el hilo de la UI con el resultado; luego la
    /// vista se redibuja. Sustituye al hilo + canal + `sender.update` escritos a mano. Si el trabajo se canceló,
    /// `done` no se llama.
    pub fn job<R: Send + 'static>(
        &mut self,
        work: impl FnOnce(&CancelToken) -> R + Send + 'static,
        done: impl FnOnce(&mut T, R, &mut EntityContext<T>) + Send + 'static,
    ) -> JobHandle {
        let pool = self.background.clone();
        launch(self, move |task| pool.spawn(task), work, done)
    }

    /// Como [`EntityContext::job`] pero en un hilo propio con nombre: para lo que bloquea mucho rato (esperar al
    /// navegador, un `ffmpeg`) y no debe ocupar un hilo del pool.
    pub fn job_blocking<R: Send + 'static>(
        &mut self,
        name: &str,
        work: impl FnOnce(&CancelToken) -> R + Send + 'static,
        done: impl FnOnce(&mut T, R, &mut EntityContext<T>) + Send + 'static,
    ) -> JobHandle {
        let name = name.to_owned();
        launch(self, move |task| BackgroundExecutor::spawn_dedicated(&name, task), work, done)
    }
}

fn launch<T: Any, R: Send + 'static>(
    cx: &mut EntityContext<T>,
    run: impl FnOnce(Box<dyn FnOnce() + Send>),
    work: impl FnOnce(&CancelToken) -> R + Send + 'static,
    done: impl FnOnce(&mut T, R, &mut EntityContext<T>) + Send + 'static,
) -> JobHandle {
    let (sender, handle) = (cx.foreground(), cx.handle());
    let job = JobHandle::default();
    let token = job.cancel.clone();
    run(Box::new(move || {
        let result = work(&token);
        if !token.is_cancelled() {
            sender.update(handle, move |this, cx| {
                done(this, result, cx);
                cx.notify();
            });
        }
    }));
    job
}

/// Lo que un futuro del hilo principal puede pedir sin tener prestado el `EntityMap`.
#[derive(Clone)]
pub struct AsyncContext {
    pub(crate) sender: ForegroundSender,
    pub(crate) background: BackgroundExecutor,
}

impl AsyncContext {
    /// Actualiza una entidad en el hilo principal; la tarea da `None` si ya no existe.
    pub fn update<T: Any, R: Send + 'static>(
        &self,
        handle: Handle<T>,
        mutate: impl FnOnce(&mut T, &mut EntityContext<T>) -> R + Send + 'static,
    ) -> Task<Option<R>> {
        let (task, completer) = Task::pending();
        if self.sender.send(move |entities| completer.complete(entities.update(handle, mutate))) {
            task
        } else {
            Task::ready(None)
        }
    }

    pub fn background_spawn<R: Send + 'static>(&self, job: impl FnOnce(&CancelToken) -> R + Send + 'static) -> Task<R> {
        self.background.spawn_task(job)
    }

    /// Se completa tras `duration`, usando la cola de temporizadores de la app.
    pub fn timer(&self, duration: Duration) -> Task<()> {
        let (task, completer) = Task::pending();
        let cancel = task.cancel_token();
        self.sender.send(move |entities| {
            if !cancel.is_cancelled() {
                entities.timers.schedule_cancellable(duration, cancel, move |_| completer.complete(()));
            }
        });
        task
    }
}

#[cfg(test)]
mod tests {
    use crate::core::entity::EntityMap;
    use std::time::{Duration, Instant};

    #[derive(Default)]
    struct Counter {
        value: u32,
        task: Option<crate::core::task::Task<()>>,
    }

    fn run_until(entities: &mut EntityMap, mut done: impl FnMut(&EntityMap) -> bool) {
        let deadline = Instant::now() + Duration::from_secs(5);
        while !done(entities) {
            assert!(Instant::now() < deadline, "el futuro no terminó");
            entities.run_pending(Instant::now());
            std::thread::sleep(Duration::from_millis(1));
        }
    }

    #[test]
    fn spawned_futures_await_background_work_and_update_their_entity() {
        let mut entities = EntityMap::new();
        let counter = entities.insert(Counter::default());
        entities.update(counter, |counter, cx| {
            counter.task = Some(cx.spawn(|this, cx| async move {
                cx.timer(Duration::from_millis(5)).await;
                let value = cx.background_spawn(|_| 40 + 2).await;
                cx.update(this, move |counter, _| counter.value = value).await;
            }));
        });
        run_until(&mut entities, |entities| entities.read(counter).is_some_and(|counter| counter.value == 42));
        assert_eq!(entities.local.live_futures(), 0, "el futuro terminado se libera");
    }

    #[test]
    fn replacing_the_task_cancels_the_previous_future() {
        let mut entities = EntityMap::new();
        let counter = entities.insert(Counter::default());
        for value in [1, 2] {
            entities.update(counter, |counter, cx| {
                counter.task = Some(cx.spawn(move |this, cx| async move {
                    cx.timer(Duration::from_millis(20)).await;
                    cx.update(this, move |counter, _| counter.value += value).await;
                }));
            });
        }
        run_until(&mut entities, |entities| entities.local.live_futures() == 0);
        assert_eq!(entities.read(counter).map(|counter| counter.value), Some(2), "solo corre el último (debounce)");
    }
}