//! Waiting for several files at once: poll, select and epoll.
//!
//! A server talking to many clients can't call `read` on one of them and
//! block -- the others would wait. Instead it asks the kernel "tell me
//! which of these fds are ready" and then only reads/writes those, which
//! never blocks. That loop ("wait for events, handle each, repeat") is an
//! **event loop**; every GUI toolkit, web server and terminal emulator is
//! built around one. Three interfaces for the same idea:
//!
//!   * `poll(fds, n, timeout)`: an array of {fd, events wanted, events
//!     returned}. Simple; the kernel looks at every fd on every call.
//!   * `select(n, readfds, writefds, exceptfds, timeout)`: the 1983 BSD
//!     original, with bitmaps of fds instead of an array.
//!   * `epoll`: register the fds *once* in an epoll instance (itself an
//!     fd) with epoll_ctl, then epoll_wait returns just the ready ones.
//!     Linux's answer to servers with 10 000 connections.
//!
//! **How our kernel does it.** Every file can say what's ready right now
//! (`File::poll`). Anything that might make a file ready -- data into a
//! pipe, a key typed, a packet for a socket -- calls `notify()`, which
//! bumps a global counter and wakes every poller. A poller checks all its
//! fds; if none is ready it sleeps until the counter changes (or its
//! timeout), then checks again. Simple, and never misses an event; a big
//! kernel instead keeps per-file wait lists so one packet only wakes the
//! pollers of that socket.
//!
//! Simplification: epoll is *level-triggered* only (EPOLLET is accepted
//! but means the same thing: "ready" is reported as long as it lasts).
//! Regular files are always ready (Linux too), and epoll refuses them
//! with EPERM like Linux does.

use crate::errno::Errno;
use crate::fs::file::{File, KIND_DIR, KIND_FILE, Stat};
use crate::syscall::{read_timespec_us, user_slice, user_slice_mut};
use crate::thread::WaitQueue;
use alloc::sync::{Arc, Weak};
use alloc::vec::Vec;
use core::sync::atomic::{AtomicU64, Ordering};
use spin::Mutex;

// poll(2) event bits (Linux's values). epoll uses the same numbers.
pub const POLLIN: u16 = 0x1; // data to read (or EOF / a connection to accept)
pub const POLLPRI: u16 = 0x2; // urgent data
pub const POLLOUT: u16 = 0x4; // room to write
pub const POLLERR: u16 = 0x8; // error (e.g. pipe without readers)
pub const POLLHUP: u16 = 0x10; // other side hung up
pub const POLLNVAL: u16 = 0x20; // not an open fd
pub const POLLRDNORM: u16 = 0x40;
pub const POLLWRNORM: u16 = 0x100;
pub const POLLRDHUP: u16 = 0x2000;

/// Bumped on every possible readiness change (see the top).
static GENERATION: AtomicU64 = AtomicU64::new(0);
/// Pollers sleep here.
static POLLERS: WaitQueue = WaitQueue::new();

/// Something may have become ready: wake all pollers to look again.
pub fn notify() {
    GENERATION.fetch_add(1, Ordering::SeqCst);
    POLLERS.wake_all();
}

/// Add the "normal data" aliases a program may ask for instead.
fn with_aliases(events: u16) -> u16 {
    let mut e = events;
    if events & POLLIN != 0 {
        e |= POLLRDNORM;
    }
    if events & POLLOUT != 0 {
        e |= POLLWRNORM;
    }
    e
}

/// The poll loop: call `check` (with interrupts on) until it returns
/// Some, the deadline (PIT tick; Some(0) = don't wait) passes -> Ok(None),
/// or a signal arrives -> Err.
pub fn wait_ready<T>(deadline: Option<u64>, mut check: impl FnMut() -> Option<T>) -> Result<Option<T>, Errno> {
    loop {
        let generation = GENERATION.load(Ordering::SeqCst);
        if let Some(result) = check() {
            return Ok(Some(result));
        }
        if deadline.is_some_and(|d| crate::interrupts::ticks() >= d) {
            return Ok(None);
        }
        let changed = || (GENERATION.load(Ordering::SeqCst) != generation).then_some(());
        match deadline {
            None => POLLERS.wait_until_intr(changed)?,
            Some(d) => {
                POLLERS.wait_until_deadline_intr(d, changed)?;
            }
        }
    }
}

/// A timeout in milliseconds (negative = forever) as a deadline.
fn ms_deadline(ms: i64) -> Option<u64> {
    match ms {
        ms if ms < 0 => None,
        0 => Some(0),
        ms => Some(crate::time::deadline_after_us(ms as u64 * 1000)),
    }
}

/// A `struct timespec *` timeout (NULL = forever) as a deadline.
fn timespec_deadline(ptr: u64) -> Result<Option<u64>, Errno> {
    if ptr == 0 {
        return Ok(None);
    }
    Ok(Some(match read_timespec_us(ptr)? {
        0 => 0,
        us => crate::time::deadline_after_us(us),
    }))
}

/// ppoll / pselect6 / epoll_pwait: run with signal mask `mask_ptr` (if
/// given) while waiting. If a signal interrupts, the old mask comes back
/// only after its handler ran (signal.rs), like Linux -- otherwise the
/// signal we waited for could end up blocked again before delivery.
fn with_sigmask<T>(mask_ptr: u64, f: impl FnOnce() -> Result<T, Errno>) -> Result<T, Errno> {
    if mask_ptr == 0 {
        return f();
    }
    let mask = u64::from_le_bytes(user_slice(mask_ptr, 8)?.try_into().unwrap());
    crate::signal::set_temporary_mask(mask);
    let result = f();
    if !matches!(result, Err(Errno::EINTR | Errno::ERESTARTNOHAND)) {
        crate::signal::restore_temporary_mask();
    }
    result
}

// ---------------------------------------------------------------------------
// poll / ppoll
// ---------------------------------------------------------------------------

/// poll(fds, nfds, timeout_ms) (7) and ppoll(fds, nfds, *timespec, *sigmask,
/// 8) (271). `struct pollfd {int fd; short events; short revents;}`.
pub fn sys_poll(fds_ptr: u64, nfds: u64, deadline: Option<u64>) -> Result<u64, Errno> {
    if nfds > 1024 {
        return Err(Errno::EINVAL);
    }
    let p = crate::thread::current_process().ok_or(Errno::EINVAL)?;
    let raw = user_slice(fds_ptr, nfds * 8)?;
    // (fd, events wanted, the file: None = not open)
    let entries: Vec<(i32, u16, Option<Arc<dyn File>>)> = raw
        .chunks(8)
        .map(|c| {
            let fd = i32::from_le_bytes(c[0..4].try_into().unwrap());
            let events = u16::from_le_bytes(c[4..6].try_into().unwrap());
            (fd, events, if fd < 0 { None } else { p.file(fd as u64).ok() })
        })
        .collect();
    drop(p);
    let mut revents = alloc::vec![0u16; entries.len()];
    let ready = wait_ready(deadline, || {
        let mut count = 0;
        for (i, (fd, events, file)) in entries.iter().enumerate() {
            revents[i] = match file {
                _ if *fd < 0 => 0, // ignored entry
                None => POLLNVAL,
                Some(f) => f.poll() & (with_aliases(*events) | POLLERR | POLLHUP),
            };
            count += (revents[i] != 0) as u64;
        }
        (count > 0).then_some(count)
    })
    .map_err(|_| Errno::ERESTARTNOHAND)?
    .unwrap_or(0);
    let out = user_slice_mut(fds_ptr, nfds * 8)?;
    for (i, r) in revents.iter().enumerate() {
        out[8 * i + 6..8 * i + 8].copy_from_slice(&r.to_le_bytes());
    }
    Ok(ready)
}

pub fn sys_poll_ms(fds_ptr: u64, nfds: u64, timeout_ms: i64) -> Result<u64, Errno> {
    sys_poll(fds_ptr, nfds, ms_deadline(timeout_ms))
}

/// ppoll(fds, nfds, *timespec, *sigmask, sigsetsize). Stage 17: like
/// Linux, a mask whose size isn't 8 bytes (64 signals) is EINVAL.
pub fn sys_ppoll(fds_ptr: u64, nfds: u64, ts_ptr: u64, mask_ptr: u64, setsize: u64) -> Result<u64, Errno> {
    check_setsize(mask_ptr, setsize)?;
    let deadline = timespec_deadline(ts_ptr)?;
    with_sigmask(mask_ptr, || sys_poll(fds_ptr, nfds, deadline))
}

// ---------------------------------------------------------------------------
// select / pselect6
// ---------------------------------------------------------------------------

/// select(nfds, *readfds, *writefds, *exceptfds, *timeval) (23). Each set
/// is a bitmap: bit n = fd n. On return the sets hold only the ready fds.
pub fn sys_select(nfds: u64, sets: [u64; 3], deadline: Option<u64>) -> Result<u64, Errno> {
    if nfds > 1024 {
        return Err(Errno::EINVAL);
    }
    let words = nfds.div_ceil(64) as usize;
    // Read the three bitmaps (a null pointer = empty set).
    let mut wanted = [[0u64; 16]; 3];
    for (k, &ptr) in sets.iter().enumerate() {
        if ptr != 0 {
            let bytes = user_slice(ptr, words as u64 * 8)?;
            for w in 0..words {
                wanted[k][w] = u64::from_le_bytes(bytes[8 * w..8 * w + 8].try_into().unwrap());
            }
        }
    }
    let p = crate::thread::current_process().ok_or(Errno::EINVAL)?;
    let mut files: Vec<(usize, Arc<dyn File>)> = Vec::new();
    for fd in 0..nfds as usize {
        if (0..3).any(|k| wanted[k][fd / 64] & (1 << (fd % 64)) != 0) {
            files.push((fd, p.file(fd as u64)?)); // EBADF for a closed fd
        }
    }
    drop(p);
    let mut result = [[0u64; 16]; 3];
    let ready = wait_ready(deadline, || {
        result = [[0u64; 16]; 3];
        let mut count = 0;
        for (fd, file) in &files {
            let events = file.poll();
            let bit = 1u64 << (fd % 64);
            // readable: data, EOF/hangup or an error (the read won't block)
            let tests = [POLLIN | POLLHUP | POLLERR, POLLOUT | POLLERR, POLLPRI];
            for k in 0..3 {
                if wanted[k][fd / 64] & bit != 0 && events & tests[k] != 0 {
                    result[k][fd / 64] |= bit;
                    count += 1;
                }
            }
        }
        (count > 0).then_some(count)
    })
    .map_err(|_| Errno::ERESTARTNOHAND)?
    .unwrap_or(0);
    for (k, &ptr) in sets.iter().enumerate() {
        if ptr != 0 {
            let out = user_slice_mut(ptr, words as u64 * 8)?;
            for w in 0..words {
                out[8 * w..8 * w + 8].copy_from_slice(&result[k][w].to_le_bytes());
            }
        }
    }
    Ok(ready)
}

/// select's `struct timeval {seconds, microseconds}` (NULL = forever).
pub fn sys_select_tv(nfds: u64, r: u64, w: u64, e: u64, tv_ptr: u64) -> Result<u64, Errno> {
    let deadline = if tv_ptr == 0 {
        None
    } else {
        let tv = user_slice(tv_ptr, 16)?;
        let secs = i64::from_le_bytes(tv[..8].try_into().unwrap());
        let usecs = i64::from_le_bytes(tv[8..].try_into().unwrap());
        if secs < 0 || !(0..1_000_000).contains(&usecs) {
            return Err(Errno::EINVAL);
        }
        let us = secs as u64 * 1_000_000 + usecs as u64;
        Some(if us == 0 { 0 } else { crate::time::deadline_after_us(us) })
    };
    sys_select(nfds, [r, w, e], deadline)
}

/// pselect6(nfds, r, w, e, *timespec, *{sigmask ptr, size}) (270).
pub fn sys_pselect6(nfds: u64, r: u64, w: u64, e: u64, ts_ptr: u64, mask_arg: u64) -> Result<u64, Errno> {
    let deadline = timespec_deadline(ts_ptr)?;
    let (mask_ptr, setsize) = if mask_arg == 0 {
        (0, 8)
    } else {
        let raw = user_slice(mask_arg, 16)?;
        (u64::from_le_bytes(raw[..8].try_into().unwrap()), u64::from_le_bytes(raw[8..].try_into().unwrap()))
    };
    check_setsize(mask_ptr, setsize)?;
    with_sigmask(mask_ptr, || sys_select(nfds, [r, w, e], deadline))
}

// ---------------------------------------------------------------------------
// epoll
// ---------------------------------------------------------------------------

pub const EPOLL_CTL_ADD: u64 = 1;
pub const EPOLL_CTL_DEL: u64 = 2;
pub const EPOLL_CTL_MOD: u64 = 3;
pub const EPOLLET: u32 = 1 << 31;
pub const EPOLLONESHOT: u32 = 1 << 30;
pub const EPOLL_CLOEXEC: u64 = 0o2000000;

/// One watched fd.
struct Interest {
    fd: u64,
    /// Weak: closing the file (its last fd) makes the entry go away, like
    /// Linux; the epoll instance doesn't keep files open.
    file: Weak<dyn File>,
    events: u32,
    data: u64,
    /// EPOLLONESHOT: reported once, then off until EPOLL_CTL_MOD.
    disabled: bool,
}

/// An epoll instance: the "interest list".
pub struct Epoll {
    interest: Mutex<Vec<Interest>>,
}

impl Epoll {
    /// The ready entries as (events, data), at most `max`.
    fn ready(&self, max: usize) -> Vec<(u32, u64)> {
        let mut out = Vec::new();
        let mut list = self.interest.lock();
        list.retain(|i| i.file.strong_count() > 0); // closed files drop out
        for i in list.iter_mut().filter(|i| !i.disabled) {
            let Some(file) = i.file.upgrade() else { continue };
            let events = file.poll() as u32 & (i.events | (POLLERR | POLLHUP) as u32);
            if events != 0 {
                out.push((events, i.data));
                if i.events & EPOLLONESHOT != 0 {
                    i.disabled = true;
                }
                if out.len() == max {
                    break;
                }
            }
        }
        out
    }
}

impl File for Epoll {
    fn read(&self, _buf: &mut [u8]) -> Result<usize, Errno> {
        Err(Errno::EINVAL)
    }
    fn write(&self, _buf: &[u8]) -> Result<usize, Errno> {
        Err(Errno::EINVAL)
    }
    fn stat(&self) -> Result<Stat, Errno> {
        Ok(Stat { kind: crate::fs::file::KIND_DEVICE, size: 0, mode: 0o600 })
    }
    /// An epoll fd is readable when one of its files is ready (so epoll
    /// instances can be polled themselves). Doesn't consume ONESHOTs.
    fn poll(&self) -> u16 {
        let list = self.interest.lock();
        let ready = list.iter().any(|i| {
            !i.disabled && i.file.upgrade().is_some_and(|f| f.poll() as u32 & (i.events | (POLLERR | POLLHUP) as u32) != 0)
        });
        if ready { POLLIN } else { 0 }
    }
    fn as_epoll(&self) -> Option<&Epoll> {
        Some(self)
    }
}

/// epoll_create1(flags) (291); epoll_create(size) (213) is the same
/// without flags.
pub fn sys_epoll_create1(flags: u64) -> Result<u64, Errno> {
    if flags & !EPOLL_CLOEXEC != 0 {
        return Err(Errno::EINVAL);
    }
    let p = crate::thread::current_process().ok_or(Errno::EINVAL)?;
    p.add_file_with(Arc::new(Epoll { interest: Mutex::new(Vec::new()) }), 0, flags & EPOLL_CLOEXEC != 0)
}

/// `struct epoll_event {u32 events; u64 data;}` -- packed: 12 bytes.
fn read_event(ptr: u64) -> Result<(u32, u64), Errno> {
    let b = user_slice(ptr, 12)?;
    Ok((u32::from_le_bytes(b[0..4].try_into().unwrap()), u64::from_le_bytes(b[4..12].try_into().unwrap())))
}

/// epoll_ctl(epfd, op, fd, *event) (233).
pub fn sys_epoll_ctl(epfd: u64, op: u64, fd: u64, event_ptr: u64) -> Result<u64, Errno> {
    let p = crate::thread::current_process().ok_or(Errno::EINVAL)?;
    let ep_file = p.file(epfd)?;
    let file = p.file(fd)?;
    let ep = ep_file.as_epoll().ok_or(Errno::EINVAL)?;
    if Arc::ptr_eq(&ep_file, &file) {
        return Err(Errno::EINVAL); // can't watch itself
    }
    if matches!(file.stat()?.kind, KIND_FILE | KIND_DIR) {
        return Err(Errno::EPERM); // always ready: pointless (Linux says EPERM too)
    }
    let mut list = ep.interest.lock();
    list.retain(|i| i.file.strong_count() > 0);
    let pos = list.iter().position(|i| i.fd == fd && i.file.upgrade().is_some_and(|f| Arc::ptr_eq(&f, &file)));
    match op {
        EPOLL_CTL_ADD | EPOLL_CTL_MOD => {
            let (events, data) = read_event(event_ptr)?;
            match (op, pos) {
                (EPOLL_CTL_ADD, Some(_)) => return Err(Errno::EEXIST),
                (EPOLL_CTL_ADD, None) => list.push(Interest { fd, file: Arc::downgrade(&file), events, data, disabled: false }),
                (_, None) => return Err(Errno::ENOENT),
                (_, Some(i)) => list[i] = Interest { fd, file: Arc::downgrade(&file), events, data, disabled: false },
            }
        }
        EPOLL_CTL_DEL => {
            list.remove(pos.ok_or(Errno::ENOENT)?);
        }
        _ => return Err(Errno::EINVAL),
    }
    drop(list);
    notify(); // a waiter may care about the new entry
    Ok(0)
}

/// epoll_wait(epfd, *events, max, timeout_ms) (232) and epoll_pwait(...,
/// *sigmask, 8) (281).
/// The sigset size check of ppoll / pselect6 / epoll_pwait (stage 17).
fn check_setsize(mask_ptr: u64, setsize: u64) -> Result<(), Errno> {
    if mask_ptr != 0 && setsize != 8 { Err(Errno::EINVAL) } else { Ok(()) }
}

pub fn sys_epoll_pwait(epfd: u64, events_ptr: u64, max: i64, timeout_ms: i64, mask_ptr: u64, setsize: u64) -> Result<u64, Errno> {
    check_setsize(mask_ptr, setsize)?;
    if max <= 0 || max > 1024 {
        return Err(Errno::EINVAL);
    }
    let file = crate::thread::current_process().ok_or(Errno::EINVAL)?.file(epfd)?;
    let ep = file.as_epoll().ok_or(Errno::EINVAL)?;
    user_slice_mut(events_ptr, max as u64 * 12)?; // check first
    let ready = with_sigmask(mask_ptr, || {
        wait_ready(ms_deadline(timeout_ms), || {
            let r = ep.ready(max as usize);
            (!r.is_empty()).then_some(r)
        })
        .map_err(|_| Errno::EINTR) // epoll_wait is never restarted (Linux)
    })?
    .unwrap_or_default();
    let out = user_slice_mut(events_ptr, ready.len() as u64 * 12)?;
    for (i, (events, data)) in ready.iter().enumerate() {
        out[12 * i..12 * i + 4].copy_from_slice(&events.to_le_bytes());
        out[12 * i + 4..12 * i + 12].copy_from_slice(&data.to_le_bytes());
    }
    Ok(ready.len() as u64)
}