//! Unix domain sockets (stage 18): AF_UNIX, sockets between programs on
//! the same machine.
//!
//! A network socket (net/socket.rs) talks to another computer through the
//! network card. A Unix socket talks to another *program on this
//! computer*, and the kernel just moves the bytes from one socket's queue
//! to the other's. Wayland, X11, D-Bus, systemd, Docker and most local
//! servers use them. Why not a pipe? A Unix socket
//!
//!   * goes both ways (a pipe is one-way),
//!   * has a *name* other programs can find: a file in the file system
//!     ("/tmp/.X11-unix/X0", made by `bind`), or an "abstract" name that
//!     lives only in the kernel, or none at all (`socketpair`),
//!   * lets one server `accept` many clients, like TCP,
//!   * can carry *open files* from one program to another (SCM_RIGHTS):
//!     the receiver gets new fds for the same open files. Wayland sends
//!     shared-memory buffers that way.
//!   * says who is on the other end (SO_PEERCRED: pid, uid, gid).
//!
//! The three kinds, picked by socket()'s type:
//!
//!   SOCK_STREAM     a byte stream, like TCP or a pipe: message borders
//!                   are not kept (two writes may come out as one read)
//!   SOCK_DGRAM      separate messages ("datagrams"), like UDP, but
//!                   reliable and in order; no connection needed
//!   SOCK_SEQPACKET  connected like STREAM, but keeps message borders
//!
//! How it's built. Every socket has an `Endpoint`: a queue of messages
//! that were sent *to* it, plus its name, its peer and flags. Sending
//! means "lock the peer's endpoint, append to its queue, wake its
//! readers". Each queue has a size limit (SO_RCVBUF, and the sender's
//! SO_SNDBUF); a sender that finds the queue full waits on the peer's
//! `writers` queue, or gets EAGAIN when non-blocking. That waiting is
//! called *backpressure*: a fast writer can't use up all the memory.
//!
//! Connecting: `connect` makes a brand-new endpoint for the server's side
//! and puts it in the listening socket's queue of pending connections;
//! `accept` takes it out and gives it an fd. (Linux calls the not yet
//! accepted one an "embryo".)
//!
//! Lock rule: an endpoint's `state` is locked with interrupts off (it is
//! read in wait conditions, see thread.rs), and we never hold two
//! endpoints' locks at once, so two sockets sending to each other can't
//! deadlock.
//!
//! Known gap: a socket that is sent through *itself* (its own fd, in its
//! own queue) keeps itself alive forever. Linux has a garbage collector
//! for such cycles; we don't (yet).

use crate::errno::Errno;
use crate::fs::{self, file::File, file::Stat, file::KIND_SOCKET, Inode, InodeKind};
use crate::poll::{self, POLLERR, POLLHUP, POLLIN, POLLOUT, POLLRDHUP};
use crate::process::Process;
use crate::signal;
use crate::syscall::{copy_to_user, current, read_iovecs, user_slice, user_slice_mut};
use crate::thread::WaitQueue;
use alloc::{
    collections::VecDeque,
    sync::{Arc, Weak},
    vec::Vec,
};
use core::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
use spin::Mutex;
use x86_64::instructions::interrupts::without_interrupts;

pub const AF_UNSPEC: u64 = 0;
pub const AF_UNIX: u64 = 1;
const SOCK_STREAM: u64 = 1;
const SOCK_DGRAM: u64 = 2;
const SOCK_RAW: u64 = 3;
const SOCK_SEQPACKET: u64 = 5;
const SOCK_NONBLOCK: u64 = 0o4000;
const SOCK_CLOEXEC: u64 = 0o2000000;

// send/recv flags (Linux's values).
const MSG_OOB: u64 = 0x1;
const MSG_PEEK: u64 = 0x2;
const MSG_CTRUNC: u32 = 0x8;
const MSG_TRUNC: u64 = 0x20;
const MSG_DONTWAIT: u64 = 0x40;
const MSG_WAITALL: u64 = 0x100;
const MSG_NOSIGNAL: u64 = 0x4000;
const MSG_CMSG_CLOEXEC: u64 = 0x4000_0000;

// Socket options.
const SOL_SOCKET: u64 = 1;
const SO_REUSEADDR: u64 = 2;
const SO_TYPE: u64 = 3;
const SO_ERROR: u64 = 4;
const SO_SNDBUF: u64 = 7;
const SO_RCVBUF: u64 = 8;
const SO_KEEPALIVE: u64 = 9;
const SO_PASSCRED: u64 = 16;
const SO_PEERCRED: u64 = 17;
const SO_RCVTIMEO: u64 = 20;
const SO_SNDTIMEO: u64 = 21;
const SO_ACCEPTCONN: u64 = 30;
const SO_SNDBUFFORCE: u64 = 32;
const SO_RCVBUFFORCE: u64 = 33;
const SO_PROTOCOL: u64 = 38;
const SO_DOMAIN: u64 = 39;

// Control messages ("ancillary data") in sendmsg/recvmsg.
const SCM_RIGHTS: u32 = 1;
const SCM_CREDENTIALS: u32 = 2;
/// At most this many fds in one message (Linux's SCM_MAX_FD).
const SCM_MAX_FD: usize = 253;

/// Buffer sizes, as on a default Linux: 208 KiB each way; setsockopt
/// doubles what it's given (Linux keeps room for bookkeeping) within
/// these limits.
const DEFAULT_BUF: usize = 212_992;
const MAX_BUF: usize = 2 * 212_992;
const MIN_SNDBUF: usize = 4608;
const MIN_RCVBUF: usize = 2304;
/// What a datagram costs on top of its bytes (Linux's sk_buff header is
/// about that big): so a queue of empty messages can fill up too.
const MSG_OVERHEAD: usize = 256;
/// The most sendmsg/recvmsg copies in one call.
const MAX_IO: usize = 4 << 20;
/// Longest sockaddr_un: 2 bytes family + 108 path bytes.
const SUN_LEN: usize = 110;

#[derive(Clone, Copy, PartialEq, Eq, Debug)]
pub enum Type {
    Stream,
    Dgram,
    SeqPacket,
}

impl Type {
    fn number(self) -> u32 {
        match self {
            Type::Stream => SOCK_STREAM as u32,
            Type::Dgram => SOCK_DGRAM as u32,
            Type::SeqPacket => SOCK_SEQPACKET as u32,
        }
    }
}

/// A socket's address.
#[derive(Clone, PartialEq, Eq, Debug)]
pub enum Name {
    /// Never bound (socketpair, most clients).
    Unnamed,
    /// A file system name, as given to bind (getsockname shows it as is).
    Path(Vec<u8>),
    /// Linux's abstract names: sun_path starts with a 0 byte; the name
    /// is the rest and exists only in the kernel (no file, gone when the
    /// socket closes).
    Abstract(Vec<u8>),
}

/// struct ucred: who sent a message / is on the other end.
#[repr(C)]
#[derive(Clone, Copy, PartialEq, Eq, Debug, Default)]
pub struct Ucred {
    pub pid: u32,
    pub uid: u32,
    pub gid: u32,
}

/// The calling process: (the credentials a message carries by default:
/// pid + *real* ids, the ones SO_PEERCRED shows: pid + *effective* ids),
/// as Linux.
fn my_creds(p: &Process) -> (Ucred, Ucred) {
    let c = *p.creds.lock();
    let pid = p.pid as u32;
    (Ucred { pid, uid: c.ruid, gid: c.rgid }, Ucred { pid, uid: c.euid, gid: c.egid })
}

/// One message in a socket's queue. On a stream socket, chunks without
/// fds from the same sender are merged into the last one.
struct Msg {
    data: Vec<u8>,
    /// How much of `data` has been read (stream sockets read in pieces).
    start: usize,
    /// SCM_RIGHTS: open files on their way to the receiver ("in flight").
    /// Dropping the message closes them.
    fds: Vec<Arc<dyn File>>,
    cred: Ucred,
    from: Name,
}

impl Msg {
    fn len(&self) -> usize {
        self.data.len() - self.start
    }
}

enum Role {
    /// Just made (or a datagram socket).
    Fresh,
    /// listen() was called: `pending` are connections not accepted yet.
    Listening { backlog: usize, pending: VecDeque<Arc<Endpoint>> },
    /// A stream/seqpacket socket with a peer (it may have gone since).
    Connected,
}

struct State {
    role: Role,
    /// The other end (stream: the connection; dgram: the default
    /// destination set by connect).
    peer: Option<Weak<Endpoint>>,
    name: Name,
    /// The registry key we're bound under (to remove it on close).
    key: Option<Key>,
    queue: VecDeque<Msg>,
    /// What the queue costs: bytes (stream), bytes + MSG_OVERHEAD per
    /// message (dgram, seqpacket). Senders wait while it's at the limit.
    queued: usize,
    rcv_shut: bool,
    snd_shut: bool,
    /// SO_PASSCRED: receive SCM_CREDENTIALS with every message.
    passcred: bool,
    /// SO_PEERCRED: who connected (server side) or listened (client).
    peercred: Option<Ucred>,
    /// A pending error for the next call (ECONNRESET: the peer closed
    /// with data we never read... or one it never read).
    error: Option<Errno>,
    closed: bool,
    rcvtimeo: Option<u64>,
    sndtimeo: Option<u64>,
}

pub struct Endpoint {
    /// Its number: fstat's st_ino and "socket:[N]" in /proc/self/fd.
    id: u64,
    ty: Type,
    state: Mutex<State>,
    /// Waiting for something to read in our queue (or a connection).
    readers: WaitQueue,
    /// Waiting for room in our queue (senders to us, connectors).
    writers: WaitQueue,
    sndbuf: AtomicUsize,
    rcvbuf: AtomicUsize,
}

static NEXT_ID: AtomicU64 = AtomicU64::new(1000);

impl Endpoint {
    fn new(ty: Type) -> Arc<Endpoint> {
        Arc::new(Endpoint {
            id: NEXT_ID.fetch_add(1, Ordering::Relaxed),
            ty,
            state: Mutex::new(State {
                role: Role::Fresh,
                peer: None,
                name: Name::Unnamed,
                key: None,
                queue: VecDeque::new(),
                queued: 0,
                rcv_shut: false,
                snd_shut: false,
                passcred: false,
                peercred: None,
                error: None,
                closed: false,
                rcvtimeo: None,
                sndtimeo: None,
            }),
            readers: WaitQueue::new(),
            writers: WaitQueue::new(),
            sndbuf: AtomicUsize::new(DEFAULT_BUF),
            rcvbuf: AtomicUsize::new(DEFAULT_BUF),
        })
    }

    /// Run `f` on our state, locked (with interrupts off: see the top).
    fn with<R>(&self, f: impl FnOnce(&mut State) -> R) -> R {
        without_interrupts(|| f(&mut self.state.lock()))
    }

    fn wake(&self) {
        self.readers.wake_all();
        self.writers.wake_all();
        poll::notify();
    }

    fn peer(&self) -> Option<Arc<Endpoint>> {
        self.with(|s| s.peer.as_ref().and_then(|w| w.upgrade()))
    }

    /// What a message of `len` bytes costs in a queue.
    fn cost(&self, len: usize) -> usize {
        if self.ty == Type::Stream { len } else { len + MSG_OVERHEAD }
    }

    /// The socket is closed (its last fd is gone), or a pending connection
    /// is thrown away (`embryo`). Tell the peer, empty the queue: the
    /// fds in flight in it are closed too, once our lock is released.
    fn close(self: &Arc<Self>, embryo: bool) {
        let (queue, pending, peer, key, unread) = self.with(|s| {
            s.closed = true;
            s.rcv_shut = true;
            s.snd_shut = true;
            let unread = !s.queue.is_empty();
            let pending = match core::mem::replace(&mut s.role, Role::Fresh) {
                Role::Listening { pending, .. } => pending,
                _ => VecDeque::new(),
            };
            s.queued = 0;
            (core::mem::take(&mut s.queue), pending, s.peer.take(), s.key.take(), unread)
        });
        if let Some(key) = key {
            unregister(&key, self);
        }
        if let Some(peer) = peer.and_then(|w| w.upgrade()) {
            if self.ty != Type::Dgram {
                // The connection is over: the peer reads EOF after its
                // data, and gets EPIPE writing. If we die with unread data
                // (or before being accepted), it gets ECONNRESET first.
                peer.with(|s| {
                    s.rcv_shut = true;
                    s.snd_shut = true;
                    if unread || embryo {
                        s.error = Some(Errno::ECONNRESET);
                    }
                });
            }
            peer.wake();
        }
        for p in pending {
            p.close(true);
        }
        self.wake();
        drop(queue); // closes the in-flight fds (maybe other sockets: no locks held)
    }
}

// --- names: the registry ----------------------------------------------------

/// How a bound socket is found: by the socket file's inode (path names:
/// whichever path leads to the file, even after a rename), or by its
/// abstract name.
#[derive(Clone, PartialEq, Eq, Debug)]
enum Key {
    Inode(u64, u64), // (st_dev, st_ino)
    Abstract(Vec<u8>),
}

struct Bound {
    key: Key,
    ep: Weak<Endpoint>,
    /// Keep the socket file's inode alive while bound (ramfs inode
    /// numbers are addresses: this way none can be reused meanwhile).
    _inode: Option<Arc<dyn Inode>>,
}

static REGISTRY: Mutex<Vec<Bound>> = Mutex::new(Vec::new());

fn registry<R>(f: impl FnOnce(&mut Vec<Bound>) -> R) -> R {
    without_interrupts(|| f(&mut REGISTRY.lock()))
}

fn unregister(key: &Key, ep: &Arc<Endpoint>) {
    registry(|r| r.retain(|b| !(b.key == *key && b.ep.as_ptr() == Arc::as_ptr(ep))));
}

fn find_key(key: &Key) -> Option<Arc<Endpoint>> {
    registry(|r| r.iter().rev().find(|b| b.key == *key).and_then(|b| b.ep.upgrade()))
}

/// The key of the socket file at `path` (and the inode), if it is one.
fn inode_key(path: &str) -> Result<(Key, Arc<dyn Inode>), Errno> {
    let (inode, real) = fs::lookup_resolved(path, true)?;
    if inode.kind() != InodeKind::Socket {
        return Err(Errno::ECONNREFUSED); // a file, but no socket's
    }
    let ino = inode.ino().unwrap_or_else(|| fs::file::ino_hash(real.as_bytes()));
    Ok((Key::Inode(fs::dev_of(&real), ino), inode))
}

/// A sockaddr_un from user space.
#[derive(Debug)]
enum Addr {
    /// AF_UNSPEC: connect() to "nothing" = disconnect (datagrams).
    Unspec,
    /// Just the family: bind picks an abstract name ("autobind").
    Auto,
    Path(Vec<u8>),
    Abstract(Vec<u8>),
}

fn read_addr(ptr: u64, len: u64) -> Result<Addr, Errno> {
    if len < 2 || len as usize > SUN_LEN {
        return Err(Errno::EINVAL);
    }
    let raw = user_slice(ptr, len)?;
    match u16::from_ne_bytes([raw[0], raw[1]]) as u64 {
        AF_UNIX => {}
        AF_UNSPEC => return Ok(Addr::Unspec),
        _ => return Err(Errno::EINVAL),
    }
    let path = &raw[2..];
    Ok(match path.first() {
        None => Addr::Auto,
        Some(0) => Addr::Abstract(path[1..].to_vec()),
        Some(_) => Addr::Path(path.iter().take_while(|&&b| b != 0).copied().collect()),
    })
}

/// The absolute path of a Path address (relative ones: from the cwd).
fn abs_path(raw: &[u8]) -> Result<alloc::string::String, Errno> {
    crate::syscall::path_at(crate::syscall::AT_FDCWD as u64, raw)
}

/// Find the socket an address names (connect, sendto).
fn lookup(addr: &Addr) -> Result<Arc<Endpoint>, Errno> {
    let key = match addr {
        Addr::Path(raw) => inode_key(&abs_path(raw)?)?.0,
        Addr::Abstract(name) => Key::Abstract(name.clone()),
        Addr::Auto | Addr::Unspec => return Err(Errno::EINVAL),
    };
    find_key(&key).ok_or(Errno::ECONNREFUSED)
}

/// Encode a name as a sockaddr_un: (bytes, length).
fn encode(name: &Name) -> Vec<u8> {
    let mut out = (AF_UNIX as u16).to_ne_bytes().to_vec();
    match name {
        Name::Unnamed => {}
        Name::Path(p) => {
            out.extend_from_slice(p);
            out.push(0);
        }
        Name::Abstract(a) => {
            out.push(0);
            out.extend_from_slice(a);
        }
    }
    out
}

/// Write an address like Linux: *len_ptr says how much room there is;
/// we write at most that and set it to the real length.
fn write_addr(ptr: u64, len_ptr: u64, bytes: &[u8]) -> Result<(), Errno> {
    if ptr == 0 || len_ptr == 0 {
        return Ok(());
    }
    let room = i32::from_ne_bytes(user_slice(len_ptr, 4)?.try_into().unwrap());
    if room < 0 {
        return Err(Errno::EINVAL);
    }
    let n = bytes.len().min(room as usize);
    user_slice_mut(ptr, n as u64)?.copy_from_slice(&bytes[..n]);
    copy_to_user(len_ptr, &(bytes.len() as u32))
}

impl Endpoint {
    fn bind(self: &Arc<Self>, addr: Addr) -> Result<(), Errno> {
        if self.with(|s| s.name != Name::Unnamed || s.closed) {
            return Err(Errno::EINVAL); // already has a name
        }
        let (key, name, inode) = match addr {
            Addr::Unspec => return Err(Errno::EINVAL),
            Addr::Auto => return self.autobind(),
            Addr::Abstract(a) => (Key::Abstract(a.clone()), Name::Abstract(a), None),
            Addr::Path(raw) => {
                if raw.is_empty() {
                    return Err(Errno::EINVAL);
                }
                let path = abs_path(&raw)?;
                // The name is a new file of type "socket". If anything
                // is there already (even an old socket file nobody is
                // bound to), that's EADDRINUSE: servers unlink first.
                let inode = match fs::create(&path, InodeKind::Socket) {
                    Err(Errno::EEXIST) => return Err(Errno::EADDRINUSE),
                    Err(Errno::EOPNOTSUPP) => return Err(Errno::EPERM), // FAT, exFAT: no such files
                    other => other?,
                };
                crate::syscall::apply_create_mode(&*inode, 0o777); // srwxr-xr-x after the umask
                let (key, inode) = inode_key(&path)?;
                (key, Name::Path(raw), Some(inode))
            }
        };
        registry(|r| {
            r.retain(|b| b.ep.strong_count() > 0);
            if let Key::Abstract(_) = key {
                if r.iter().any(|b| b.key == key) {
                    return Err(Errno::EADDRINUSE);
                }
            }
            // (A path key can only repeat if the old file was deleted and
            // its inode number reused: the old entry is stale then.)
            r.retain(|b| b.key != key);
            r.push(Bound { key: key.clone(), ep: Arc::downgrade(self), _inode: inode });
            Ok(())
        })?;
        self.with(|s| {
            s.name = name;
            s.key = Some(key);
        });
        Ok(())
    }

    /// bind with just the family: a free abstract name of 5 hex digits,
    /// like Linux.
    fn autobind(self: &Arc<Self>) -> Result<(), Errno> {
        static NEXT: AtomicU64 = AtomicU64::new(0);
        for _ in 0..0x10_0000 {
            let n = NEXT.fetch_add(1, Ordering::Relaxed) & 0xf_ffff;
            let name = alloc::format!("{:05x}", n).into_bytes();
            match self.bind(Addr::Abstract(name)) {
                Err(Errno::EADDRINUSE) => continue,
                other => return other,
            }
        }
        Err(Errno::EADDRINUSE)
    }

    fn listen(&self, backlog: i64) -> Result<(), Errno> {
        if self.ty == Type::Dgram {
            return Err(Errno::EOPNOTSUPP);
        }
        let (_, peer_cred) = my_creds(&*current()?);
        let backlog = if backlog < 0 { 4096 } else { (backlog as usize).min(4096) };
        self.with(|s| {
            if s.name == Name::Unnamed {
                return Err(Errno::EINVAL); // nobody could find it
            }
            match &mut s.role {
                Role::Connected => Err(Errno::EINVAL),
                Role::Listening { backlog: b, .. } => {
                    *b = backlog;
                    Ok(())
                }
                Role::Fresh => {
                    s.role = Role::Listening { backlog, pending: VecDeque::new() };
                    s.peercred = Some(peer_cred); // what connecting clients will see
                    Ok(())
                }
            }
        })?;
        self.wake();
        Ok(())
    }

    fn connect(self: &Arc<Self>, addr: Addr, nonblock: bool) -> Result<(), Errno> {
        if self.ty == Type::Dgram {
            // Datagrams: just remember the default destination.
            let target = match addr {
                Addr::Unspec => None,
                addr => Some(lookup(&addr)?),
            };
            if let Some(t) = &target {
                if t.ty != self.ty {
                    return Err(Errno::EPROTOTYPE);
                }
                if t.with(|s| s.peer.as_ref().is_some_and(|w| w.as_ptr() != Arc::as_ptr(self))) {
                    return Err(Errno::EPERM); // it only talks to someone else
                }
            }
            self.with(|s| s.peer = target.as_ref().map(Arc::downgrade));
            self.wake();
            return Ok(());
        }
        self.with(|s| match s.role {
            Role::Fresh => Ok(()),
            Role::Connected => Err(Errno::EISCONN),
            Role::Listening { .. } => Err(Errno::EINVAL),
        })?;
        if let Addr::Unspec = addr {
            return Err(Errno::EAFNOSUPPORT);
        }
        let listener = lookup(&addr)?;
        if listener.ty != self.ty {
            return Err(Errno::EPROTOTYPE);
        }
        let (_, me) = my_creds(&*current()?);
        // The server's end of the new connection.
        let server = Endpoint::new(self.ty);
        let (lname, lcred, lpass) = listener.with(|s| (s.name.clone(), s.peercred, s.passcred));
        server.with(|s| {
            s.role = Role::Connected;
            s.peer = Some(Arc::downgrade(self));
            s.name = lname;
            s.peercred = Some(me);
            s.passcred = lpass;
        });
        server.sndbuf.store(listener.sndbuf.load(Ordering::Relaxed), Ordering::Relaxed);
        server.rcvbuf.store(listener.rcvbuf.load(Ordering::Relaxed), Ordering::Relaxed);
        let timeout = self.with(|s| s.sndtimeo);
        let mut item = Some(server.clone());
        // Into the listener's queue of pending connections; when that's
        // full (backlog), wait like Linux (or EAGAIN when non-blocking).
        wait_on(&listener.writers, timeout, || {
            let mut s = listener.state.lock();
            if s.closed {
                return Some(Err(Errno::ECONNREFUSED));
            }
            match &mut s.role {
                Role::Listening { backlog, pending } => {
                    if pending.len() <= *backlog {
                        pending.push_back(item.take().unwrap());
                        Some(Ok(()))
                    } else if nonblock {
                        Some(Err(Errno::EAGAIN))
                    } else {
                        None
                    }
                }
                _ => Some(Err(Errno::ECONNREFUSED)), // bound, but not listening
            }
        })??;
        self.with(|s| {
            s.role = Role::Connected;
            s.peer = Some(Arc::downgrade(&server));
            s.peercred = lcred;
        });
        listener.readers.wake_all();
        poll::notify();
        Ok(())
    }

    fn accept(&self, nonblock: bool) -> Result<Arc<Endpoint>, Errno> {
        let timeout = self.with(|s| s.rcvtimeo);
        let ep = wait_on(&self.readers, timeout, || {
            let mut s = self.state.lock();
            match &mut s.role {
                Role::Listening { pending, .. } => match pending.pop_front() {
                    Some(ep) => Some(Ok(ep)),
                    None if nonblock => Some(Err(Errno::EAGAIN)),
                    None => None,
                },
                _ => Some(Err(Errno::EINVAL)),
            }
        })??;
        self.writers.wake_all(); // room in the backlog
        poll::notify();
        Ok(ep)
    }

    fn shutdown(&self, how: u64) -> Result<(), Errno> {
        let (rd, wr) = match how {
            0 => (true, false),
            1 => (false, true),
            2 => (true, true),
            _ => return Err(Errno::EINVAL),
        };
        let peer = self.with(|s| {
            s.rcv_shut |= rd;
            s.snd_shut |= wr;
            s.peer.as_ref().and_then(|w| w.upgrade())
        });
        if let Some(peer) = peer.filter(|_| self.ty != Type::Dgram) {
            // The other side sees the mirror image: we stop writing, it
            // reads EOF; we stop reading, its writes fail (EPIPE).
            peer.with(|s| {
                s.rcv_shut |= wr;
                s.snd_shut |= rd;
            });
            peer.wake();
        }
        self.wake();
        Ok(())
    }
}

/// Wait on `q` until `f` says Some, with an optional timeout
/// (SO_RCVTIMEO/SO_SNDTIMEO: then EAGAIN, like Linux).
fn wait_on<T>(q: &WaitQueue, timeout_us: Option<u64>, f: impl FnMut() -> Option<T>) -> Result<T, Errno> {
    match timeout_us {
        None => q.wait_until_intr(f),
        Some(us) => q.wait_until_deadline_intr(crate::time::deadline_after_us(us), f)?.ok_or(Errno::EAGAIN),
    }
}

// --- sending and receiving --------------------------------------------------

/// What one recv found.
struct Received {
    data: Vec<u8>,
    /// The message's whole length (datagrams: more than `data` when it
    /// didn't fit, and the rest is thrown away: MSG_TRUNC).
    full: usize,
    fds: Vec<Arc<dyn File>>,
    cred: Option<Ucred>,
    from: Name,
}

impl Endpoint {
    /// Send `data` (+ `fds`, SCM_RIGHTS) to our peer, or to `to`.
    /// Returns how many bytes went; EPIPE is for the caller to turn into
    /// SIGPIPE.
    fn send(self: &Arc<Self>, data: &[u8], fds: Vec<Arc<dyn File>>, cred: Ucred, to: Option<Addr>, nonblock: bool) -> Result<usize, Errno> {
        let (name, timeout) = self.with(|s| (s.name.clone(), s.sndtimeo));
        if self.ty == Type::Dgram {
            return self.send_dgram(data, fds, cred, to, nonblock, name, timeout);
        }
        // Stream and seqpacket: only ever to the connected peer.
        let (connected, snd_shut) = self.with(|s| (matches!(s.role, Role::Connected), s.snd_shut));
        if to.is_some() {
            return Err(if connected { Errno::EISCONN } else { Errno::EOPNOTSUPP });
        }
        if !connected {
            return Err(Errno::ENOTCONN);
        }
        let peer = match self.peer() {
            Some(p) if !snd_shut => p,
            _ => return Err(Errno::EPIPE),
        };
        let limit = self.sndbuf.load(Ordering::Relaxed).min(peer.rcvbuf.load(Ordering::Relaxed));
        if self.ty == Type::SeqPacket && self.cost(data.len()) > limit {
            return Err(Errno::EMSGSIZE);
        }
        if data.is_empty() && self.ty == Type::Stream && fds.is_empty() {
            return Ok(0);
        }
        let mut fds = Some(fds);
        let mut sent = 0;
        loop {
            // Put as much as fits into the peer's queue; wait for room.
            let r = wait_on(&peer.writers, timeout, || {
                let mut s = peer.state.lock();
                if s.rcv_shut || s.closed {
                    return Some(Err(Errno::EPIPE));
                }
                let room = limit.saturating_sub(s.queued);
                let n = if self.ty == Type::Stream {
                    room.min(data.len() - sent)
                } else if s.queue.is_empty() || room >= self.cost(data.len()) {
                    data.len()
                } else {
                    0
                };
                if room == 0 || (n == 0 && data.len() > sent) {
                    return if nonblock { Some(Err(Errno::EAGAIN)) } else { None };
                }
                let chunk = &data[sent..sent + n];
                let new_fds = fds.take().unwrap_or_default();
                s.queued += self.cost(n);
                match s.queue.back_mut() {
                    // Stream: glue onto the last chunk if nothing tells them apart.
                    Some(last) if self.ty == Type::Stream && new_fds.is_empty() && last.fds.is_empty() && last.cred == cred && last.start == 0 => {
                        last.data.extend_from_slice(chunk)
                    }
                    _ => s.queue.push_back(Msg { data: chunk.to_vec(), start: 0, fds: new_fds, cred, from: name.clone() }),
                }
                drop(s);
                peer.readers.wake_all();
                poll::notify();
                Some(Ok(n))
            });
            match r {
                Ok(Ok(n)) => {
                    sent += n;
                    if sent >= data.len() {
                        return Ok(sent);
                    }
                }
                // Partly sent, then full (non-blocking), a signal or a
                // timeout: say what went.
                Ok(Err(Errno::EAGAIN)) | Err(_) if sent > 0 => return Ok(sent),
                Ok(Err(e)) | Err(e) => return Err(e),
            }
        }
    }

    #[allow(clippy::too_many_arguments)]
    fn send_dgram(self: &Arc<Self>, data: &[u8], fds: Vec<Arc<dyn File>>, cred: Ucred, to: Option<Addr>, nonblock: bool, name: Name, timeout: Option<u64>) -> Result<usize, Errno> {
        if self.with(|s| s.snd_shut) {
            return Err(Errno::EPIPE);
        }
        let target = match to {
            Some(addr) => lookup(&addr)?,
            None => match self.with(|s| s.peer.clone()) {
                None => return Err(Errno::ENOTCONN),
                Some(w) => w.upgrade().ok_or(Errno::ECONNREFUSED)?,
            },
        };
        if target.ty != Type::Dgram {
            return Err(Errno::EPROTOTYPE);
        }
        let limit = self.sndbuf.load(Ordering::Relaxed).min(target.rcvbuf.load(Ordering::Relaxed));
        if self.cost(data.len()) > limit {
            return Err(Errno::EMSGSIZE);
        }
        let me = Arc::as_ptr(self);
        let mut msg = Some(Msg { data: data.to_vec(), start: 0, fds, cred, from: name });
        wait_on(&target.writers, timeout, || {
            let mut s = target.state.lock();
            if s.closed {
                return Some(Err(Errno::ECONNREFUSED));
            }
            if s.peer.as_ref().is_some_and(|w| w.as_ptr() != me) {
                return Some(Err(Errno::EPERM)); // it only listens to its own peer
            }
            if s.rcv_shut {
                return Some(Err(Errno::EPIPE));
            }
            if !s.queue.is_empty() && s.queued + self.cost(data.len()) > limit {
                return if nonblock { Some(Err(Errno::EAGAIN)) } else { None };
            }
            s.queued += self.cost(data.len());
            s.queue.push_back(msg.take().unwrap());
            drop(s);
            target.readers.wake_all();
            poll::notify();
            Some(Ok(data.len()))
        })?
    }

    /// Receive up to `want` bytes.
    fn recv(&self, want: usize, flags: u64, nonblock: bool) -> Result<Received, Errno> {
        let peek = flags & MSG_PEEK != 0;
        let timeout = self.with(|s| s.rcvtimeo);
        let r = wait_on(&self.readers, timeout, || {
            let mut s = self.state.lock();
            match (&s.role, self.ty) {
                (Role::Listening { .. }, _) | (Role::Fresh, Type::Stream) => return Some(Err(Errno::EINVAL)),
                (Role::Fresh, Type::SeqPacket) => return Some(Err(Errno::ENOTCONN)),
                _ => {}
            }
            if s.queue.is_empty() {
                if let Some(e) = s.error.take() {
                    return Some(Err(e));
                }
                if s.rcv_shut {
                    return Some(Ok(Received { data: Vec::new(), full: 0, fds: Vec::new(), cred: None, from: Name::Unnamed }));
                }
                return if nonblock { Some(Err(Errno::EAGAIN)) } else { None };
            }
            let passcred = s.passcred;
            let st = &mut *s;
            let first = st.queue.front_mut().unwrap();
            let cred = first.cred;
            let from = first.from.clone();
            let fds = if peek { first.fds.clone() } else { core::mem::take(&mut first.fds) };
            let mut data = Vec::new();
            let full;
            if self.ty == Type::Stream {
                // Gather from as many chunks as wanted, but not past one
                // carrying fds (they belong with its first byte) or from
                // another sender when credentials are passed.
                let had_fds = !fds.is_empty();
                let mut used = 0;
                let mut i = 0;
                while data.len() < want && i < st.queue.len() {
                    let m = &mut st.queue[i];
                    if i > 0 && (had_fds || !m.fds.is_empty() || (passcred && m.cred != cred)) {
                        break;
                    }
                    let n = m.len().min(want - data.len());
                    data.extend_from_slice(&m.data[m.start..m.start + n]);
                    if !peek {
                        m.start += n;
                        used += n;
                    }
                    i += 1;
                }
                full = data.len();
                if !peek {
                    st.queued -= used;
                    while st.queue.front().is_some_and(|m| m.len() == 0 && m.fds.is_empty()) {
                        st.queue.pop_front();
                    }
                }
            } else {
                // A whole message; what doesn't fit is lost.
                full = first.len();
                let n = full.min(want);
                data.extend_from_slice(&first.data[first.start..first.start + n]);
                if !peek {
                    st.queued -= self.cost(full);
                    st.queue.pop_front(); // (its fds were taken above)
                }
            }
            Some(Ok(Received { data, full, fds, cred: passcred.then_some(cred), from }))
        })??;
        // Room again in our queue: wake senders waiting for it.
        if !peek {
            self.writers.wake_all();
            poll::notify();
        }
        Ok(r)
    }

    fn poll(&self) -> u16 {
        let (mut ready, peer) = self.with(|s| {
            let mut ready = 0;
            if s.error.is_some() {
                ready |= POLLERR;
            }
            if let Role::Listening { pending, .. } = &s.role {
                return (if pending.is_empty() { 0 } else { POLLIN }, None);
            }
            if self.ty == Type::Stream && matches!(s.role, Role::Fresh) {
                return (POLLOUT | POLLHUP, None);
            }
            if s.rcv_shut && s.snd_shut {
                ready |= POLLHUP;
            }
            if s.rcv_shut {
                ready |= POLLIN | POLLRDHUP;
            }
            if !s.queue.is_empty() {
                ready |= POLLIN;
            }
            let peer = s.peer.as_ref().map(|w| w.upgrade());
            if s.snd_shut {
                ready |= POLLOUT; // a write fails right away (EPIPE): not blocked
            }
            (ready, peer)
        });
        if ready & POLLOUT == 0 {
            match peer {
                // Unconnected datagram socket: sendto never waits on us.
                None => ready |= POLLOUT,
                Some(None) => ready |= POLLOUT, // peer gone: writing fails at once
                Some(Some(p)) => {
                    let limit = self.sndbuf.load(Ordering::Relaxed).min(p.rcvbuf.load(Ordering::Relaxed));
                    if p.with(|s| s.closed || s.rcv_shut || s.queued < limit) {
                        ready |= POLLOUT;
                    }
                }
            }
        }
        ready
    }
}

// --- the open file ----------------------------------------------------------

/// An AF_UNIX socket as an open file (what an fd points to). When the
/// last fd (or in-flight copy) is gone, it's dropped and the socket
/// closes.
pub struct UnixSocket {
    ep: Arc<Endpoint>,
    nonblock: AtomicBool,
}

impl UnixSocket {
    fn new(ep: Arc<Endpoint>, nonblock: bool) -> Arc<dyn File> {
        Arc::new(UnixSocket { ep, nonblock: AtomicBool::new(nonblock) })
    }
    fn nb(&self, flags: u64) -> bool {
        self.nonblock.load(Ordering::Relaxed) || flags & MSG_DONTWAIT != 0
    }
    pub fn id(&self) -> u64 {
        self.ep.id
    }
}

impl Drop for UnixSocket {
    fn drop(&mut self) {
        self.ep.close(false);
    }
}

impl File for UnixSocket {
    fn read(&self, buf: &mut [u8]) -> Result<usize, Errno> {
        let r = self.ep.recv(buf.len(), 0, self.nb(0))?;
        buf[..r.data.len()].copy_from_slice(&r.data);
        Ok(r.data.len()) // (fds sent along are closed: read can't take them)
    }
    fn write(&self, buf: &[u8]) -> Result<usize, Errno> {
        let (cred, _) = my_creds(&*current()?);
        self.ep.send(buf, Vec::new(), cred, None, self.nb(0))
    }
    fn stat(&self) -> Result<Stat, Errno> {
        Ok(Stat { kind: KIND_SOCKET, size: 0, mode: 0o777 })
    }
    fn as_unix(&self) -> Option<&UnixSocket> {
        Some(self)
    }
    fn anon_ino(&self) -> Option<u64> {
        Some(self.ep.id)
    }
    fn set_nonblocking(&self, on: bool) {
        self.nonblock.store(on, Ordering::Relaxed);
    }
    fn nonblocking(&self) -> bool {
        self.nonblock.load(Ordering::Relaxed)
    }
    fn poll(&self) -> u16 {
        self.ep.poll()
    }
    fn ioctl(&self, cmd: u64, arg: u64) -> Result<u64, Errno> {
        const FIONREAD: u64 = 0x541B;
        if cmd != FIONREAD {
            return Err(Errno::ENOTTY);
        }
        // Bytes waiting (stream) or the next message's size.
        let n = self.ep.with(|s| match self.ep.ty {
            Type::Stream => s.queue.iter().map(|m| m.len()).sum::<usize>(),
            _ => s.queue.front().map_or(0, |m| m.len()),
        });
        copy_to_user(arg, &(n as i32))?;
        Ok(0)
    }
}

// --- the system calls -------------------------------------------------------

const SYS_SOCKET: u64 = 41;
const SYS_CONNECT: u64 = 42;
const SYS_ACCEPT: u64 = 43;
const SYS_SENDTO: u64 = 44;
const SYS_RECVFROM: u64 = 45;
const SYS_SENDMSG: u64 = 46;
const SYS_RECVMSG: u64 = 47;
const SYS_SHUTDOWN: u64 = 48;
const SYS_BIND: u64 = 49;
const SYS_LISTEN: u64 = 50;
const SYS_GETSOCKNAME: u64 = 51;
const SYS_GETPEERNAME: u64 = 52;
const SYS_SOCKETPAIR: u64 = 53;
const SYS_SETSOCKOPT: u64 = 54;
const SYS_GETSOCKOPT: u64 = 55;
const SYS_ACCEPT4: u64 = 288;

/// Called before the normal socket calls: handles AF_UNIX ones, and
/// sendmsg/recvmsg for every kind of socket. None = not ours.
pub fn dispatch(nr: u64, a: [u64; 6]) -> Option<Result<u64, Errno>> {
    match nr {
        SYS_SOCKET if a[0] == AF_UNIX => return Some(sys_socket(a[1], a[2])),
        SYS_SOCKETPAIR => return Some(sys_socketpair(a[0], a[1], a[2], a[3])),
        SYS_SENDMSG => return Some(sys_sendmsg(a[0], a[1], a[2])),
        SYS_RECVMSG => return Some(sys_recvmsg(a[0], a[1], a[2])),
        SYS_CONNECT | SYS_ACCEPT | SYS_ACCEPT4 | SYS_SENDTO | SYS_RECVFROM | SYS_SHUTDOWN | SYS_BIND | SYS_LISTEN
        | SYS_GETSOCKNAME | SYS_GETPEERNAME | SYS_SETSOCKOPT | SYS_GETSOCKOPT => {}
        _ => return None,
    }
    // The rest: only if the fd is a Unix socket.
    let file = current().ok()?.file(a[0]).ok()?;
    let sock = file.as_unix()?;
    let ep = &sock.ep;
    Some(match nr {
        SYS_BIND => read_addr(a[1], a[2]).and_then(|addr| ep.bind(addr)).map(|_| 0),
        SYS_LISTEN => ep.listen(a[1] as i32 as i64).map(|_| 0),
        SYS_CONNECT => read_addr(a[1], a[2]).and_then(|addr| ep.connect(addr, sock.nb(0))).map(|_| 0),
        SYS_ACCEPT => sys_accept(sock, a[1], a[2], 0),
        SYS_ACCEPT4 => sys_accept(sock, a[1], a[2], a[3]),
        SYS_SENDTO => {
            let to = if a[4] == 0 { Ok(None) } else { read_addr(a[4], a[5]).map(Some) };
            to.and_then(|to| user_slice(a[1], a[2]).and_then(|data| send(sock, data, Vec::new(), None, to, a[3])))
        }
        SYS_RECVFROM => sys_recvfrom(sock, a[1], a[2], a[3], a[4], a[5]),
        SYS_SHUTDOWN => ep.shutdown(a[1]).map(|_| 0),
        SYS_GETSOCKNAME => write_addr(a[1], a[2], &encode(&ep.with(|s| s.name.clone()))).map(|_| 0),
        SYS_GETPEERNAME => match ep.with(|s| (matches!(s.role, Role::Connected) || ep.ty == Type::Dgram, s.peer.clone())) {
            (true, Some(peer)) => {
                let name = peer.upgrade().map_or(Name::Unnamed, |p| p.with(|s| s.name.clone()));
                write_addr(a[1], a[2], &encode(&name)).map(|_| 0)
            }
            _ => Err(Errno::ENOTCONN),
        },
        SYS_SETSOCKOPT => setsockopt(ep, a[1], a[2], a[3], a[4]),
        SYS_GETSOCKOPT => getsockopt(sock, a[1], a[2], a[3], a[4]),
        _ => unreachable!(),
    })
}

fn sys_socket(ty: u64, protocol: u64) -> Result<u64, Errno> {
    if ty & !(0xf | SOCK_NONBLOCK | SOCK_CLOEXEC) != 0 {
        return Err(Errno::EINVAL);
    }
    let kind = match ty & 0xf {
        SOCK_STREAM => Type::Stream,
        SOCK_DGRAM | SOCK_RAW => Type::Dgram, // (Linux: raw = dgram here)
        SOCK_SEQPACKET => Type::SeqPacket,
        _ => return Err(Errno::ESOCKTNOSUPPORT),
    };
    if protocol > 1 {
        return Err(Errno::EPROTONOSUPPORT);
    }
    let p = current()?;
    p.add_file_with(UnixSocket::new(Endpoint::new(kind), ty & SOCK_NONBLOCK != 0), 0, ty & SOCK_CLOEXEC != 0)
}

/// socketpair: two sockets connected to each other, nameless. The
/// simplest way to get a two-way channel to a child process.
fn sys_socketpair(domain: u64, ty: u64, protocol: u64, out: u64) -> Result<u64, Errno> {
    match domain {
        AF_UNIX => {}
        2 => return Err(Errno::EOPNOTSUPP), // AF_INET: Linux says this too
        _ => return Err(Errno::EAFNOSUPPORT),
    }
    if ty & !(0xf | SOCK_NONBLOCK | SOCK_CLOEXEC) != 0 {
        return Err(Errno::EINVAL);
    }
    let kind = match ty & 0xf {
        SOCK_STREAM => Type::Stream,
        SOCK_DGRAM | SOCK_RAW => Type::Dgram,
        SOCK_SEQPACKET => Type::SeqPacket,
        _ => return Err(Errno::ESOCKTNOSUPPORT),
    };
    if protocol > 1 {
        return Err(Errno::EPROTONOSUPPORT);
    }
    user_slice_mut(out, 8)?; // check before making anything
    let p = current()?;
    let (_, me) = my_creds(&p);
    let (a, b) = (Endpoint::new(kind), Endpoint::new(kind));
    for (x, y) in [(&a, &b), (&b, &a)] {
        x.with(|s| {
            if kind != Type::Dgram {
                s.role = Role::Connected;
            }
            s.peer = Some(Arc::downgrade(y));
            s.peercred = Some(me);
        });
    }
    let nb = ty & SOCK_NONBLOCK != 0;
    let cloexec = ty & SOCK_CLOEXEC != 0;
    let fd0 = p.add_file_with(UnixSocket::new(a, nb), 0, cloexec)?;
    let fd1 = match p.add_file_with(UnixSocket::new(b, nb), 0, cloexec) {
        Ok(fd) => fd,
        Err(e) => {
            let _ = p.close(fd0);
            return Err(e);
        }
    };
    copy_to_user(out, &[fd0 as u32, fd1 as u32])?;
    Ok(0)
}

fn sys_accept(sock: &UnixSocket, addr: u64, len_ptr: u64, flags: u64) -> Result<u64, Errno> {
    if flags & !(SOCK_NONBLOCK | SOCK_CLOEXEC) != 0 {
        return Err(Errno::EINVAL);
    }
    if sock.ep.ty == Type::Dgram {
        return Err(Errno::EOPNOTSUPP);
    }
    let ep = sock.ep.accept(sock.nb(0))?;
    // The address of who connected (usually unnamed).
    let name = ep.peer().map_or(Name::Unnamed, |c| c.with(|s| s.name.clone()));
    let fd = current()?.add_file_with(UnixSocket::new(ep, flags & SOCK_NONBLOCK != 0), 0, flags & SOCK_CLOEXEC != 0)?;
    write_addr(addr, len_ptr, &encode(&name))?;
    Ok(fd)
}

/// Send, with SIGPIPE for EPIPE unless MSG_NOSIGNAL.
fn send(sock: &UnixSocket, data: &[u8], fds: Vec<Arc<dyn File>>, cred: Option<Ucred>, to: Option<Addr>, flags: u64) -> Result<u64, Errno> {
    if flags & MSG_OOB != 0 {
        return Err(Errno::EOPNOTSUPP);
    }
    let p = current()?;
    let cred = cred.unwrap_or_else(|| my_creds(&p).0);
    let r = sock.ep.send(data, fds, cred, to, sock.nb(flags));
    if r == Err(Errno::EPIPE) && flags & MSG_NOSIGNAL == 0 {
        signal::send(&p, signal::SIGPIPE, signal::SigInfo::from_pid(signal::SIGPIPE, p.pid));
    }
    r.map(|n| n as u64)
}

/// Receive into one buffer of `want` bytes: MSG_WAITALL (stream: keep
/// going until it's full) and MSG_TRUNC (datagrams: say the real size).
fn recv(sock: &UnixSocket, want: usize, flags: u64) -> Result<(Received, usize), Errno> {
    if flags & MSG_OOB != 0 {
        return Err(Errno::EOPNOTSUPP);
    }
    let nb = sock.nb(flags);
    let mut r = sock.ep.recv(want, flags, nb)?;
    if flags & MSG_WAITALL != 0 && flags & MSG_PEEK == 0 && sock.ep.ty == Type::Stream && !nb {
        while r.data.len() < want && r.fds.is_empty() {
            match sock.ep.recv(want - r.data.len(), flags, nb) {
                Ok(more) if !more.data.is_empty() && more.fds.is_empty() && more.cred == r.cred => {
                    r.data.extend_from_slice(&more.data);
                }
                Ok(more) if more.data.is_empty() => break, // EOF
                Ok(_) => break, // (can't happen: fds stop at a message border)
                Err(_) => break, // a signal: return what we have
            }
        }
        r.full = r.data.len();
    }
    let ret = if flags & MSG_TRUNC != 0 && sock.ep.ty != Type::Stream { r.full } else { r.data.len() };
    Ok((r, ret))
}

fn sys_recvfrom(sock: &UnixSocket, buf: u64, len: u64, flags: u64, addr: u64, len_ptr: u64) -> Result<u64, Errno> {
    let out = user_slice_mut(buf, len)?;
    let (r, ret) = recv(sock, out.len(), flags)?;
    out[..r.data.len()].copy_from_slice(&r.data);
    if addr != 0 {
        if r.from == Name::Unnamed {
            copy_to_user(len_ptr, &0u32)?;
        } else {
            write_addr(addr, len_ptr, &encode(&r.from))?;
        }
    }
    Ok(ret as u64) // (fds can't be taken with recvfrom: they're closed)
}

// --- sendmsg / recvmsg ------------------------------------------------------

/// struct msghdr (56 bytes on x86_64).
#[repr(C)]
#[derive(Clone, Copy)]
struct MsgHdr {
    name: u64,
    namelen: u32,
    _pad: u32,
    iov: u64,
    iovlen: u64,
    control: u64,
    controllen: u64,
    flags: i32,
    _pad2: u32,
}

const CMSG_HDR: usize = 16; // struct cmsghdr: len u64, level i32, type i32

fn cmsg_align(n: usize) -> usize {
    (n + 7) & !7
}

fn read_msghdr(ptr: u64) -> Result<MsgHdr, Errno> {
    let raw = user_slice(ptr, 56)?;
    Ok(unsafe { core::ptr::read_unaligned(raw.as_ptr() as *const MsgHdr) })
}

/// Gather the iovecs into one buffer (at most MAX_IO bytes).
fn gather(h: &MsgHdr) -> Result<Vec<u8>, Errno> {
    let mut data = Vec::new();
    for (base, len) in read_iovecs(h.iov, h.iovlen)? {
        let room = MAX_IO - data.len();
        data.extend_from_slice(user_slice(base, len.min(room as u64))?);
    }
    Ok(data)
}

/// Spread `data` over the iovecs; the room there.
fn iov_room(h: &MsgHdr) -> Result<usize, Errno> {
    let iov = read_iovecs(h.iov, h.iovlen)?;
    Ok(iov.iter().map(|&(_, l)| l as usize).sum::<usize>().min(MAX_IO))
}

fn scatter(h: &MsgHdr, mut data: &[u8]) -> Result<(), Errno> {
    for (base, len) in read_iovecs(h.iov, h.iovlen)? {
        if data.is_empty() {
            break;
        }
        let n = (len as usize).min(data.len());
        user_slice_mut(base, n as u64)?.copy_from_slice(&data[..n]);
        data = &data[n..];
    }
    Ok(())
}

/// The control messages of a sendmsg: fds (SCM_RIGHTS) and credentials.
fn read_control(h: &MsgHdr, p: &Process) -> Result<(Vec<Arc<dyn File>>, Option<Ucred>), Errno> {
    let mut fds = Vec::new();
    let mut cred = None;
    if h.control == 0 || h.controllen == 0 {
        return Ok((fds, cred));
    }
    let ctl = user_slice(h.control, h.controllen)?;
    let mut off = 0;
    while off + CMSG_HDR <= ctl.len() {
        let len = u64::from_ne_bytes(ctl[off..off + 8].try_into().unwrap()) as usize;
        let level = u32::from_ne_bytes(ctl[off + 8..off + 12].try_into().unwrap());
        let ty = u32::from_ne_bytes(ctl[off + 12..off + 16].try_into().unwrap());
        if len < CMSG_HDR || off + len > ctl.len() {
            return Err(Errno::EINVAL);
        }
        let body = &ctl[off + CMSG_HDR..off + len];
        match (level as u64, ty) {
            (SOL_SOCKET, SCM_RIGHTS) => {
                if body.len() % 4 != 0 || fds.len() + body.len() / 4 > SCM_MAX_FD {
                    return Err(Errno::EINVAL);
                }
                for fd in body.chunks(4) {
                    let fd = i32::from_ne_bytes(fd.try_into().unwrap());
                    if fd < 0 {
                        return Err(Errno::EBADF);
                    }
                    fds.push(p.file(fd as u64)?);
                }
            }
            (SOL_SOCKET, SCM_CREDENTIALS) => {
                if body.len() < 12 {
                    return Err(Errno::EINVAL);
                }
                let c = Ucred {
                    pid: u32::from_ne_bytes(body[0..4].try_into().unwrap()),
                    uid: u32::from_ne_bytes(body[4..8].try_into().unwrap()),
                    gid: u32::from_ne_bytes(body[8..12].try_into().unwrap()),
                };
                // Only root may claim to be someone else.
                let me = *p.creds.lock();
                let root = me.euid == 0;
                let uid_ok = root || [me.ruid, me.euid, me.suid].contains(&c.uid);
                let gid_ok = root || [me.rgid, me.egid, me.sgid].contains(&c.gid);
                if !(root || c.pid as u64 == p.pid) || !uid_ok || !gid_ok {
                    return Err(Errno::EPERM);
                }
                cred = Some(c);
            }
            _ => return Err(Errno::EINVAL),
        }
        off += cmsg_align(len);
    }
    Ok((fds, cred))
}

fn sys_sendmsg(fd: u64, msg: u64, flags: u64) -> Result<u64, Errno> {
    let p = current()?;
    let file = p.file(fd)?;
    let h = read_msghdr(msg)?;
    let data = gather(&h)?;
    if let Some(s) = file.as_socket() {
        // A network socket: no control messages, just the data.
        let to = if h.name == 0 { None } else { Some(crate::syscall::read_sockaddr(h.name)?) };
        return s.send(&data, to).map(|n| n as u64);
    }
    let sock = file.as_unix().ok_or(Errno::ENOTSOCK)?;
    let (fds, cred) = read_control(&h, &p)?;
    let to = if h.name == 0 || h.namelen == 0 { None } else { Some(read_addr(h.name, h.namelen as u64)?) };
    drop(p);
    send(sock, &data, fds, cred, to, flags)
}

fn sys_recvmsg(fd: u64, msg: u64, flags: u64) -> Result<u64, Errno> {
    let p = current()?;
    let file = p.file(fd)?;
    let h = read_msghdr(msg)?;
    let flags_ptr = msg + 48;
    if let Some(s) = file.as_socket() {
        let mut buf = alloc::vec![0u8; iov_room(&h)?];
        let (n, from) = s.recv(&mut buf)?;
        scatter(&h, &buf[..n])?;
        if let (Some(from), true) = (from, h.name != 0) {
            crate::syscall::write_sockaddr(h.name, msg + 8, from)?;
        } else {
            copy_to_user(msg + 8, &0u32)?;
        }
        copy_to_user(msg + 40, &0u64)?; // no control data
        copy_to_user(flags_ptr, &0i32)?;
        return Ok(n as u64);
    }
    let sock = file.as_unix().ok_or(Errno::ENOTSOCK)?;
    let (r, ret) = recv(sock, iov_room(&h)?, flags)?;
    scatter(&h, &r.data)?;
    let mut out_flags = 0u32;
    if r.full > r.data.len() {
        out_flags |= MSG_TRUNC as u32;
    }
    // The sender's address.
    if h.name != 0 {
        if r.from == Name::Unnamed {
            copy_to_user(msg + 8, &0u32)?;
        } else {
            let bytes = encode(&r.from);
            let n = bytes.len().min(h.namelen as usize);
            user_slice_mut(h.name, n as u64)?.copy_from_slice(&bytes[..n]);
            copy_to_user(msg + 8, &(bytes.len() as u32))?;
        }
    } else {
        copy_to_user(msg + 8, &0u32)?;
    }
    // Control messages: credentials, then as many fds as fit (the rest
    // are closed, and MSG_CTRUNC says so).
    let mut ctl: Vec<u8> = Vec::new();
    let room = if h.control == 0 { 0 } else { h.controllen as usize };
    if let Some(c) = r.cred {
        let len = CMSG_HDR + 12;
        if room >= len {
            ctl.extend_from_slice(&(len as u64).to_ne_bytes());
            ctl.extend_from_slice(&(SOL_SOCKET as u32).to_ne_bytes());
            ctl.extend_from_slice(&SCM_CREDENTIALS.to_ne_bytes());
            for v in [c.pid, c.uid, c.gid] {
                ctl.extend_from_slice(&v.to_ne_bytes());
            }
            ctl.resize(cmsg_align(len).min(room), 0);
        } else {
            out_flags |= MSG_CTRUNC;
        }
    }
    if !r.fds.is_empty() {
        let left = room.saturating_sub(ctl.len());
        let fit = if left < CMSG_HDR + 4 { 0 } else { ((left - CMSG_HDR) / 4).min(r.fds.len()) };
        let cloexec = flags & MSG_CMSG_CLOEXEC != 0;
        let mut installed = Vec::new();
        for f in r.fds.iter().take(fit) {
            match p.add_file_with(f.clone(), 0, cloexec) {
                Ok(fd) => installed.push(fd as i32),
                Err(_) => break, // out of fds (EMFILE): the rest are lost
            }
        }
        if installed.len() < r.fds.len() {
            out_flags |= MSG_CTRUNC;
        }
        if !installed.is_empty() {
            let len = CMSG_HDR + 4 * installed.len();
            ctl.extend_from_slice(&(len as u64).to_ne_bytes());
            ctl.extend_from_slice(&(SOL_SOCKET as u32).to_ne_bytes());
            ctl.extend_from_slice(&SCM_RIGHTS.to_ne_bytes());
            for fd in &installed {
                ctl.extend_from_slice(&fd.to_ne_bytes());
            }
            let end = ctl.len() - len + cmsg_align(len);
            ctl.resize(end.min(room), 0);
        }
    }
    drop(r.fds); // fds that didn't fit are closed here
    if !ctl.is_empty() {
        user_slice_mut(h.control, ctl.len() as u64)?.copy_from_slice(&ctl);
    }
    copy_to_user(msg + 40, &(ctl.len() as u64))?;
    copy_to_user(flags_ptr, &(out_flags as i32))?;
    Ok(ret as u64)
}

// --- socket options -----------------------------------------------------------

fn read_int(ptr: u64, len: u64) -> Result<i32, Errno> {
    if len < 4 {
        return Err(Errno::EINVAL);
    }
    Ok(i32::from_ne_bytes(user_slice(ptr, 4)?.try_into().unwrap()))
}

/// struct timeval -> microseconds (None = 0 = no timeout).
fn read_timeval(ptr: u64, len: u64) -> Result<Option<u64>, Errno> {
    if len < 16 {
        return Err(Errno::EINVAL);
    }
    let raw = user_slice(ptr, 16)?;
    let sec = i64::from_ne_bytes(raw[0..8].try_into().unwrap());
    let usec = i64::from_ne_bytes(raw[8..16].try_into().unwrap());
    if sec < 0 || !(0..1_000_000).contains(&usec) {
        return Err(Errno::EINVAL);
    }
    let us = sec as u64 * 1_000_000 + usec as u64;
    Ok((us != 0).then_some(us))
}

fn setsockopt(ep: &Endpoint, level: u64, name: u64, val: u64, len: u64) -> Result<u64, Errno> {
    if level != SOL_SOCKET {
        return Err(Errno::ENOPROTOOPT);
    }
    match name {
        SO_SNDBUF | SO_SNDBUFFORCE => {
            let v = (read_int(val, len)?.max(0) as usize).saturating_mul(2).clamp(MIN_SNDBUF, MAX_BUF);
            ep.sndbuf.store(v, Ordering::Relaxed);
            ep.wake();
        }
        SO_RCVBUF | SO_RCVBUFFORCE => {
            let v = (read_int(val, len)?.max(0) as usize).saturating_mul(2).clamp(MIN_RCVBUF, MAX_BUF);
            ep.rcvbuf.store(v, Ordering::Relaxed);
            ep.wake();
        }
        SO_PASSCRED => {
            let on = read_int(val, len)? != 0;
            ep.with(|s| s.passcred = on);
        }
        SO_RCVTIMEO => {
            let t = read_timeval(val, len)?;
            ep.with(|s| s.rcvtimeo = t);
        }
        SO_SNDTIMEO => {
            let t = read_timeval(val, len)?;
            ep.with(|s| s.sndtimeo = t);
        }
        SO_REUSEADDR | SO_KEEPALIVE => {
            read_int(val, len)?; // accepted, means nothing here
        }
        _ => return Err(Errno::ENOPROTOOPT),
    }
    Ok(0)
}

fn getsockopt(sock: &UnixSocket, level: u64, name: u64, val: u64, len_ptr: u64) -> Result<u64, Errno> {
    if level != SOL_SOCKET {
        return Err(Errno::ENOPROTOOPT);
    }
    let ep = &sock.ep;
    let room = i32::from_ne_bytes(user_slice(len_ptr, 4)?.try_into().unwrap());
    if room < 0 {
        return Err(Errno::EINVAL);
    }
    let int = |v: i64| (v as i32).to_ne_bytes().to_vec();
    let timeval = |t: Option<u64>| {
        let us = t.unwrap_or(0);
        let mut v = ((us / 1_000_000) as i64).to_ne_bytes().to_vec();
        v.extend_from_slice(&((us % 1_000_000) as i64).to_ne_bytes());
        v
    };
    let bytes = match name {
        SO_TYPE => int(ep.ty.number() as i64),
        SO_ERROR => int(ep.with(|s| s.error.take()).map_or(0, |e| e as i64)),
        SO_SNDBUF => int(ep.sndbuf.load(Ordering::Relaxed) as i64),
        SO_RCVBUF => int(ep.rcvbuf.load(Ordering::Relaxed) as i64),
        SO_PASSCRED => int(ep.with(|s| s.passcred) as i64),
        SO_ACCEPTCONN => int(ep.with(|s| matches!(s.role, Role::Listening { .. })) as i64),
        SO_DOMAIN => int(AF_UNIX as i64),
        SO_PROTOCOL => int(0),
        SO_REUSEADDR | SO_KEEPALIVE => int(0),
        SO_RCVTIMEO => timeval(ep.with(|s| s.rcvtimeo)),
        SO_SNDTIMEO => timeval(ep.with(|s| s.sndtimeo)),
        SO_PEERCRED => {
            // Nobody there: pid 0, uid/gid -1 (Linux's "overflow" ids).
            let c = ep.with(|s| s.peercred).unwrap_or(Ucred { pid: 0, uid: u32::MAX, gid: u32::MAX });
            let mut v = c.pid.to_ne_bytes().to_vec();
            v.extend_from_slice(&c.uid.to_ne_bytes());
            v.extend_from_slice(&c.gid.to_ne_bytes());
            v
        }
        _ => return Err(Errno::ENOPROTOOPT),
    };
    let n = bytes.len().min(room as usize);
    user_slice_mut(val, n as u64)?.copy_from_slice(&bytes[..n]);
    copy_to_user(len_ptr, &(n as u32))?;
    Ok(0)
}

/// How many sockets have a name right now (tests: nothing left bound).
pub fn bound_count() -> usize {
    registry(|r| r.iter().filter(|b| b.ep.strong_count() > 0).count())
}