aura-gui 0.3.5

//! `Task<T>`: el resultado futuro de un trabajo, como `Task` de GPUI.
//!
//! Soltar la tarea cancela el trabajo (el pool lo salta si aún no empezó y el
//! trabajo puede consultar el `CancelToken`). Guardar la tarea en un campo y
//! reemplazarla cancela la anterior, que es la forma natural de hacer debounce.
//! `.detach()` la deja correr sin dueño.

use crate::core::tasks::{BackgroundExecutor, CancelToken};
use std::future::Future;
use std::pin::Pin;
use std::sync::{Arc, Mutex};
use std::task::{Context, Poll, Waker};

struct Slot<T> {
    value: Option<T>,
    waker: Option<Waker>,
}

/// Mitad que entrega el valor de una `Task`.
pub struct Completer<T> {
    slot: Arc<Mutex<Slot<T>>>,
}

impl<T> Completer<T> {
    pub fn complete(self, value: T) {
        let waker = match self.slot.lock() {
            Ok(mut slot) => {
                slot.value = Some(value);
                slot.waker.take()
            }
            Err(_) => None,
        };
        if let Some(waker) = waker {
            waker.wake();
        }
    }
}

#[must_use = "soltar una Task la cancela; usa `.detach()` para dejarla correr"]
pub struct Task<T> {
    slot: Arc<Mutex<Slot<T>>>,
    cancel: CancelToken,
    detached: bool,
}

impl<T> Task<T> {
    /// Tarea pendiente y la mitad que la completa (para puentes con otros mecanismos).
    pub fn pending() -> (Task<T>, Completer<T>) {
        let slot = Arc::new(Mutex::new(Slot { value: None, waker: None }));
        let task = Task { slot: slot.clone(), cancel: CancelToken::default(), detached: false };
        (task, Completer { slot })
    }

    pub fn ready(value: T) -> Task<T> {
        let (task, completer) = Task::pending();
        completer.complete(value);
        task
    }

    pub(crate) fn cancel_token(&self) -> CancelToken {
        self.cancel.clone()
    }

    pub fn detach(mut self) {
        self.detached = true;
    }

    pub fn is_finished(&self) -> bool {
        self.slot.lock().is_ok_and(|slot| slot.value.is_some())
    }

    /// Toma el valor si ya está listo, sin esperar.
    pub fn try_take(&mut self) -> Option<T> {
        self.slot.lock().ok()?.value.take()
    }
}

impl<T> Future for Task<T> {
    type Output = T;

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<T> {
        let Ok(mut slot) = self.slot.lock() else { return Poll::Pending };
        match slot.value.take() {
            Some(value) => Poll::Ready(value),
            None => {
                slot.waker = Some(cx.waker().clone());
                Poll::Pending
            }
        }
    }
}

impl<T> Drop for Task<T> {
    fn drop(&mut self) {
        if !self.detached {
            self.cancel.cancel();
        }
    }
}

/// Envuelve un futuro para que termine (sin valor) en cuanto se cancele su `Task`.
pub(crate) struct Cancellable<F> {
    pub(crate) future: Pin<Box<F>>,
    pub(crate) cancel: CancelToken,
}

impl<F: Future> Future for Cancellable<F> {
    type Output = Option<F::Output>;

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        if self.cancel.poll_cancelled(cx.waker()) {
            return Poll::Ready(None);
        }
        let result = self.future.as_mut().poll(cx).map(Some);
        if result.is_ready() {
            self.cancel.clear_waker();
        }
        result
    }
}

impl BackgroundExecutor {
    /// Ejecuta `job` en el pool y devuelve su resultado como `Task`. El trabajo
    /// recibe el `CancelToken` para cortar a mitad si la tarea se soltó.
    pub fn spawn_task<T: Send + 'static>(&self, job: impl FnOnce(&CancelToken) -> T + Send + 'static) -> Task<T> {
        let (mut task, completer) = Task::pending();
        let cancel = CancelToken::default();
        task.cancel = cancel.clone();
        self.spawn(move || {
            if !cancel.is_cancelled() {
                completer.complete(job(&cancel));
            }
        });
        task
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::atomic::{AtomicBool, Ordering};
    use std::time::Duration;

    /// Espera activa mínima para las pruebas: sondea hasta que la tarea termine.
    fn wait<T>(mut task: Task<T>) -> T {
        for _ in 0..5_000 {
            if let Some(value) = task.try_take() {
                return value;
            }
            std::thread::sleep(Duration::from_millis(1));
        }
        panic!("la tarea no terminó");
    }

    #[test]
    fn background_tasks_deliver_their_result() {
        let executor = BackgroundExecutor::new(1);
        assert_eq!(wait(executor.spawn_task(|_| 6 * 7)), 42);
    }

    #[test]
    fn dropping_a_task_cancels_work_that_has_not_started() {
        let executor = BackgroundExecutor::new(1);
        let (release, gate) = std::sync::mpsc::channel::<()>();
        // Ocupa el único hilo para que la segunda tarea quede en cola.
        executor.spawn(move || {
            gate.recv().ok();
        });
        let ran = Arc::new(AtomicBool::new(false));
        let flag = ran.clone();
        drop(executor.spawn_task(move |_| flag.store(true, Ordering::SeqCst)));
        release.send(()).ok();
        let probe = executor.spawn_task(|_| ());
        wait(probe);
        assert!(!ran.load(Ordering::SeqCst), "la tarea soltada no debe ejecutarse");
    }
}