aura-gui 0.3.5

//! Trabajo en segundo plano y entrega de resultados al hilo principal.
//!
//! Los hilos nunca tocan entidades: envían cierres `Send` al `Inbox`, que el hilo
//! principal aplica cuando el `Waker` despierta el bucle de eventos.

use crate::core::async_context::JobHandle;
use crate::core::entity::{EntityContext, EntityMap, Handle};
use crate::core::platform::lower_current_thread_priority;
use std::any::Any;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::mpsc::{self, Receiver, Sender};
use std::sync::{Arc, Mutex, OnceLock};
use std::thread;

/// Lo implementa la app para despertar su bucle (por ejemplo con `EventLoopProxy`).
pub trait Waker: Send + Sync {
    fn wake(&self);
}

struct NoopWaker;

impl Waker for NoopWaker {
    fn wake(&self) {}
}

pub type ForegroundJob = Box<dyn FnOnce(&mut EntityMap) + Send>;

/// Extremo clonable y `Send` para enviar trabajo al hilo principal.
#[derive(Clone)]
pub struct ForegroundSender {
    sender: Sender<ForegroundJob>,
    wake: Arc<InboxWake>,
}

struct InboxWake {
    pending: AtomicBool,
    waker: Mutex<Arc<dyn Waker>>,
}

impl InboxWake {
    fn wake(&self) {
        let waker = self.waker.lock().unwrap().clone();
        waker.wake();
    }
}

impl ForegroundSender {
    /// Devuelve `false` si la app ya se cerró y nadie recibirá el trabajo.
    pub fn send(&self, job: impl FnOnce(&mut EntityMap) + Send + 'static) -> bool {
        let delivered = self.sender.send(Box::new(job)).is_ok();
        if delivered && !self.wake.pending.swap(true, Ordering::AcqRel) {
            self.wake.wake();
        }
        delivered
    }

    pub fn update<T: Any>(
        &self,
        handle: Handle<T>,
        mutate: impl FnOnce(&mut T, &mut EntityContext<T>) + Send + 'static,
    ) -> bool {
        self.send(move |entities| {
            entities.update(handle, mutate);
        })
    }
}

pub struct Inbox {
    receiver: Receiver<ForegroundJob>,
    sender: ForegroundSender,
}

impl Inbox {
    pub fn new(waker: Arc<dyn Waker>) -> Self {
        let (sender, receiver) = mpsc::channel();
        let wake = Arc::new(InboxWake { pending: AtomicBool::new(false), waker: Mutex::new(waker) });
        Self { receiver, sender: ForegroundSender { sender, wake } }
    }

    pub fn sender(&self) -> ForegroundSender {
        self.sender.clone()
    }

    pub fn set_waker(&mut self, waker: Arc<dyn Waker>) {
        *self.sender.wake.waker.lock().unwrap() = waker;
        if self.sender.wake.pending.load(Ordering::Acquire) {
            self.sender.wake.wake();
        }
    }

    pub fn try_next(&self) -> Option<ForegroundJob> {
        if let Ok(job) = self.receiver.try_recv() {
            return Some(job);
        }
        self.sender.wake.pending.store(false, Ordering::Release);
        // Un productor pudo enviar mientras pending aún era true. Se comprueba
        // otra vez después de rearmar para no perder ese trabajo al dormir.
        self.receiver.try_recv().ok()
    }
}

impl Default for Inbox {
    fn default() -> Self {
        Self::new(Arc::new(NoopWaker))
    }
}

/// Señal de cancelación cooperativa para búsquedas y escaneos largos.
#[derive(Clone, Default)]
pub struct CancelToken(Arc<CancelState>);

#[derive(Default)]
struct CancelState {
    cancelled: AtomicBool,
    waker: Mutex<Option<std::task::Waker>>,
}

impl CancelToken {
    pub fn cancel(&self) {
        self.0.cancelled.store(true, Ordering::Release);
        let waker = self.0.waker.lock().unwrap().take();
        if let Some(waker) = waker {
            waker.wake();
        }
    }

    pub fn is_cancelled(&self) -> bool {
        self.0.cancelled.load(Ordering::Acquire)
    }

    /// Registra el executor antes de dormir, sin perder cancelaciones concurrentes.
    pub(crate) fn poll_cancelled(&self, waker: &std::task::Waker) -> bool {
        let mut slot = self.0.waker.lock().unwrap();
        if self.is_cancelled() {
            return true;
        }
        if slot.as_ref().is_none_or(|old| !old.will_wake(waker)) {
            *slot = Some(waker.clone());
        }
        false
    }

    pub(crate) fn clear_waker(&self) {
        self.0.waker.lock().unwrap().take();
    }
}

/// «Solo vale la última»: al lanzar un trabajo nuevo se cancela el anterior y, aunque este ya hubiera terminado,
/// su resultado se descarta. Para búsquedas al escribir, miniaturas y listas que dependen de una selección.
#[derive(Default)]
pub struct Latest {
    generation: Arc<AtomicU64>,
    current: Option<JobHandle>,
}

impl Latest {
    pub fn run<T: Any, R: Send + 'static>(
        &mut self,
        cx: &mut EntityContext<T>,
        work: impl FnOnce(&CancelToken) -> R + Send + 'static,
        done: impl FnOnce(&mut T, R, &mut EntityContext<T>) + Send + 'static,
    ) {
        self.cancel();
        let ticket = self.generation.load(Ordering::Acquire);
        let generation = self.generation.clone();
        self.current = Some(cx.job(work, move |this, result, cx| {
            if generation.load(Ordering::Acquire) == ticket {
                done(this, result, cx);
            }
        }));
    }

    /// Descarta lo que esté en vuelo (por ejemplo al cerrar el panel).
    pub fn cancel(&mut self) {
        self.generation.fetch_add(1, Ordering::AcqRel);
        if let Some(job) = self.current.take() {
            job.cancel();
        }
    }
}

type Job = Box<dyn FnOnce() + Send>;

struct LazyPool {
    workers: usize,
    sender: OnceLock<Sender<Job>>,
}

/// Pool perezoso y adaptativo de hilos de baja prioridad (QoS `Utility` / `nice 10`).
/// No crea ningún hilo del sistema hasta el primer [`Self::spawn`]; clonarlo comparte el mismo pool.
#[derive(Clone)]
pub struct BackgroundExecutor {
    pool: Arc<LazyPool>,
}

impl BackgroundExecutor {
    /// Elige el número de hilos dejando al menos 1 núcleo libre para el hilo de UI y la GPU.
    pub fn recommended_workers() -> usize {
        let cores = thread::available_parallelism().map(|n| n.get()).unwrap_or(2);
        cores.saturating_sub(1).clamp(1, 4)
    }

    pub fn new(workers: usize) -> Self {
        Self { pool: Arc::new(LazyPool { workers: workers.max(1), sender: OnceLock::new() }) }
    }

    fn sender(&self) -> &Sender<Job> {
        self.pool.sender.get_or_init(|| {
            let (sender, receiver) = mpsc::channel::<Job>();
            let receiver = Arc::new(Mutex::new(receiver));
            for index in 0..self.pool.workers {
                let receiver = receiver.clone();
                let spawned = thread::Builder::new().name(format!("aura-worker-{index}")).spawn(move || {
                    lower_current_thread_priority();
                    loop {
                        let job = match receiver.lock() {
                            Ok(guard) => guard.recv(),
                            Err(_) => return,
                        };
                        match job {
                            Ok(job) => job(),
                            Err(_) => return,
                        }
                    }
                });
                if let Err(error) = spawned {
                    eprintln!("no se pudo crear un hilo de trabajo: {error}");
                }
            }
            sender
        })
    }

    /// `true` si el pool ya arrancó sus hilos de sistema.
    pub fn is_started(&self) -> bool {
        self.pool.sender.get().is_some()
    }

    pub fn spawn(&self, job: impl FnOnce() + Send + 'static) {
        if self.sender().send(Box::new(job)).is_err() {
            eprintln!("el pool de trabajo ya terminó");
        }
    }

    /// Hilo propio de prioridad rebajada para tareas que viven mucho (lectores de LSP, pty, watcher).
    pub fn spawn_dedicated(name: &str, job: impl FnOnce() + Send + 'static) {
        let spawned = thread::Builder::new().name(name.to_owned()).spawn(move || {
            lower_current_thread_priority();
            job();
        });
        if let Err(error) = spawned {
            eprintln!("no se pudo crear el hilo {name}: {error}");
        }
    }
}

impl Default for BackgroundExecutor {
    fn default() -> Self {
        Self::new(Self::recommended_workers())
    }
}