//! Native owned-message channel and shared handles. See os-handles/owned/CONTRACT.md.
//! A caller must bind `P` to a live kernel process ID before invoking this
//! table. `Object` IDs are kernel-owned names, never user pointers.

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct Handle(u64);

impl Handle {
    pub const fn from_raw(raw: u64) -> Self { Self(raw) }
    pub const fn raw(self) -> u64 { self.0 }
    pub const fn slot(self) -> u32 { self.0 as u32 }
    pub const fn generation(self) -> u32 { (self.0 >> 32) as u32 }
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum Object {
    File(u64),
    Channel(u64),
    Task(u64),
}

impl Object {
    const fn id(self) -> u64 {
        match self {
            Self::File(id) | Self::Channel(id) | Self::Task(id) => id,
        }
    }
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct Rights(u8);

impl Rights {
    pub const READ: Self = Self(1);
    pub const WRITE: Self = Self(2);
    pub const READ_WRITE: Self = Self(3);

    pub const fn contains(self, required: Self) -> bool {
        self.0 & required.0 == required.0
    }

    pub const fn is_nonempty_subset_of(self, original: Self) -> bool {
        self.0 != 0 && original.contains(self)
    }
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum Error {
    BadFd,
    Access,
    InvalidObject,
    Full,
    IdExhausted,
    MessageTooLarge,
    WouldBlock,
    Fault,
}

impl Error {
    pub const fn errno(self) -> u64 {
        match self {
            Self::BadFd => 9,
            Self::Access => 13,
            Self::InvalidObject => 22,
            Self::Full => 24,
            Self::IdExhausted => 75,
            Self::MessageTooLarge => 90,
            Self::WouldBlock => 11,
            Self::Fault => 14,
        }
    }
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
struct Entry<P> {
    handle: Handle,
    owner: P,
    object: Object,
    rights: Rights,
}

/// Complete scalar audit state. It does not expose a mutable table entry.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct Snapshot<P, const N: usize> {
    pub generations: [u32; N],
    pub entries: [Option<(u64, P, Object, Rights)>; N],
}

pub struct Table<P: Copy + Eq, const N: usize> {
    entries: [Option<Entry<P>>; N],
    generations: [u32; N],
}

impl<P: Copy + Eq, const N: usize> Table<P, N> {
    pub const fn new() -> Self {
        Self { entries: [None; N], generations: [0; N] }
    }

    /// Kernel-only deterministic seed for an isolated overflow probe. A user
    /// syscall adapter must never take the seed from caller input.
    pub const fn with_generation_seed(generations: [u32; N]) -> Self {
        Self { entries: [None; N], generations }
    }

    pub fn len(&self) -> usize {
        self.entries.iter().filter(|entry| entry.is_some()).count()
    }

    pub fn owned(&self, owner: P) -> usize {
        self.entries.iter().flatten().filter(|entry| entry.owner == owner).count()
    }

    /// Validate all fallible conditions before publishing the entry.
    pub fn open(&mut self, owner: P, object: Object) -> Result<Handle, Error> {
        let rights = match object {
            Object::File(_) => Rights::READ_WRITE,
            Object::Channel(_) => Rights::WRITE,
            Object::Task(_) => Rights::READ,
        };
        self.open_with_rights(owner, object, rights)
    }

    pub fn open_with_rights(&mut self, owner: P, object: Object,
        rights: Rights) -> Result<Handle, Error> {
        if object.id() == 0 { return Err(Error::InvalidObject); }
        let allowed = match object {
            Object::File(_) => Rights::READ_WRITE,
            Object::Channel(_) => Rights::WRITE,
            Object::Task(_) => Rights::READ,
        };
        if !rights.is_nonempty_subset_of(allowed) {
            return Err(Error::Access);
        }
        let slot = self.entries.iter().position(Option::is_none).ok_or(Error::Full)?;
        let slot_id = u32::try_from(slot).ok()
            .and_then(|value| value.checked_add(1)).ok_or(Error::IdExhausted)?;
        let generation = self.generations[slot]
            .checked_add(1).ok_or(Error::IdExhausted)?;
        let handle = Handle((u64::from(generation) << 32) | u64::from(slot_id));
        self.entries[slot] = Some(Entry { handle, owner, object, rights });
        self.generations[slot] = generation;
        Ok(handle)
    }

    fn entry(&self, owner: P, handle: Handle) -> Result<Entry<P>, Error> {
        let slot = handle.slot().checked_sub(1)
            .and_then(|value| usize::try_from(value).ok()).ok_or(Error::BadFd)?;
        self.entries.get(slot).copied().flatten()
            .filter(|entry| entry.owner == owner && entry.handle == handle)
            .ok_or(Error::BadFd)
    }

    /// A File handle cannot be used through the Channel or Task API.
    pub fn file_read(&self, owner: P, handle: Handle) -> Result<u64, Error> {
        let entry = self.entry(owner, handle)?;
        match entry.object {
            Object::File(id) if entry.rights.contains(Rights::READ) => Ok(id),
            Object::File(_) => Err(Error::Access),
            _ => Err(Error::BadFd),
        }
    }

    pub fn file_write(&self, owner: P, handle: Handle) -> Result<u64, Error> {
        let entry = self.entry(owner, handle)?;
        match entry.object {
            Object::File(id) if entry.rights.contains(Rights::WRITE) => Ok(id),
            Object::File(_) => Err(Error::Access),
            _ => Err(Error::BadFd),
        }
    }

    pub fn channel_send(&self, owner: P, handle: Handle) -> Result<u64, Error> {
        let entry = self.entry(owner, handle)?;
        match entry.object {
            Object::Channel(id) if entry.rights.contains(Rights::WRITE) => Ok(id),
            Object::Channel(_) => Err(Error::Access),
            _ => Err(Error::BadFd),
        }
    }

    pub fn task_inspect(&self, owner: P, handle: Handle) -> Result<u64, Error> {
        let entry = self.entry(owner, handle)?;
        match entry.object {
            Object::Task(id) if entry.rights.contains(Rights::READ) => Ok(id),
            Object::Task(_) => Err(Error::Access),
            _ => Err(Error::BadFd),
        }
    }

    /// Atomically publish a second reference only with a rights subset.
    pub fn duplicate(&mut self, owner: P, source: Handle,
        requested: Rights) -> Result<Handle, Error> {
        let original = self.entry(owner, source)?;
        if !requested.is_nonempty_subset_of(original.rights) {
            return Err(Error::Access);
        }
        let slot = self.entries.iter().position(Option::is_none).ok_or(Error::Full)?;
        let slot_id = u32::try_from(slot).ok()
            .and_then(|value| value.checked_add(1)).ok_or(Error::IdExhausted)?;
        let generation = self.generations[slot]
            .checked_add(1).ok_or(Error::IdExhausted)?;
        let handle = Handle((u64::from(generation) << 32) | u64::from(slot_id));
        self.entries[slot] = Some(Entry {handle, owner,
            object: original.object, rights: requested});
        self.generations[slot] = generation;
        Ok(handle)
    }

    pub fn close(&mut self, owner: P, handle: Handle) -> Result<(), Error> {
        let slot = self.entries.iter().position(|entry| entry.is_some_and(
            |entry| entry.owner == owner && entry.handle == handle))
            .ok_or(Error::BadFd)?;
        self.entries[slot] = None;
        Ok(())
    }

    pub fn close_owner(&mut self, owner: P) -> usize {
        let mut closed = 0;
        for entry in &mut self.entries {
            if entry.is_some_and(|entry| entry.owner == owner) {
                *entry = None;
                closed += 1;
            }
        }
        closed
    }

    pub fn snapshot(&self) -> Snapshot<P, N> {
        Snapshot { generations: self.generations,
            entries: self.entries.map(|entry| entry.map(
            |entry| (entry.handle.raw(), entry.owner, entry.object, entry.rights))) }
    }
}

impl<P: Copy + Eq, const N: usize> Default for Table<P, N> {
    fn default() -> Self { Self::new() }
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct WaitToken(u64);

impl WaitToken {
    pub const fn from_raw(raw: u64) -> Self { Self(raw) }
    pub const fn raw(self) -> u64 { self.0 }
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum WaitResult { Pending, Closed }

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum CloseResult { Remaining(u32), LastClosed }

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
struct SharedEntry<P> {
    handle: Handle,
    owner: P,
    resource: usize,
    rights: Rights,
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
struct ResourceRecord {
    object: Object,
    refs: u32,
    closed: bool,
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
struct Watcher<P> {
    token: WaitToken,
    owner: P,
    resource: usize,
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct SharedSnapshot<P, const N: usize> {
    pub generations: [u32; N],
    pub entries: [Option<(u64, P, usize, Rights)>; N],
    pub resources: [Option<(Object, u32, bool)>; N],
    pub watchers: [Option<(u64, P, usize)>; N],
    pub next_wait: u64,
}

pub struct SharedObjects<P: Copy + Eq, const N: usize> {
    entries: [Option<SharedEntry<P>>; N],
    generations: [u32; N],
    resources: [Option<ResourceRecord>; N],
    watchers: [Option<Watcher<P>>; N],
    next_wait: u64,
}

impl<P: Copy + Eq, const N: usize> SharedObjects<P, N> {
    pub const fn new() -> Self {
        Self {entries: [None; N], generations: [0; N],
            resources: [None; N], watchers: [None; N], next_wait: 1}
    }

    pub fn handles(&self) -> usize {
        self.entries.iter().flatten().count()
    }

    pub fn owned(&self, owner: P) -> usize {
        self.entries.iter().flatten().filter(|entry| entry.owner == owner).count()
    }

    pub fn live_resources(&self) -> usize {
        self.resources.iter().flatten().filter(|item| item.refs > 0).count()
    }

    fn next_handle(&self, slot: usize) -> Result<(Handle, u32), Error> {
        let slot_id = u32::try_from(slot).ok()
            .and_then(|value| value.checked_add(1)).ok_or(Error::IdExhausted)?;
        let generation = self.generations[slot]
            .checked_add(1).ok_or(Error::IdExhausted)?;
        Ok((Handle((u64::from(generation) << 32) | u64::from(slot_id)),
            generation))
    }

    fn lookup(&self, owner: P, handle: Handle) -> Result<SharedEntry<P>, Error> {
        let slot = handle.slot().checked_sub(1)
            .and_then(|value| usize::try_from(value).ok()).ok_or(Error::BadFd)?;
        self.entries.get(slot).copied().flatten()
            .filter(|entry| entry.owner == owner && entry.handle == handle)
            .ok_or(Error::BadFd)
    }

    /// The caller must obtain the new object and owner from kernel authority.
    pub fn open_kernel(&mut self, owner: P, object: Object,
        rights: Rights) -> Result<Handle, Error> {
        if object.id() == 0 { return Err(Error::InvalidObject); }
        let allowed = match object {
            Object::File(_) => Rights::READ_WRITE,
            Object::Channel(_) => Rights::READ_WRITE,
            Object::Task(_) => Rights::READ,
        };
        if !rights.is_nonempty_subset_of(allowed) { return Err(Error::Access); }
        if self.resources.iter().flatten().any(|item| item.object == object) {
            return Err(Error::InvalidObject);
        }
        let handle_slot = self.entries.iter().position(Option::is_none)
            .ok_or(Error::Full)?;
        let resource_slot = self.resources.iter().position(Option::is_none)
            .ok_or(Error::Full)?;
        let (handle, generation) = self.next_handle(handle_slot)?;
        self.resources[resource_slot] = Some(ResourceRecord
            {object, refs: 1, closed: false});
        self.entries[handle_slot] = Some(SharedEntry
            {handle, owner, resource: resource_slot, rights});
        self.generations[handle_slot] = generation;
        Ok(handle)
    }

    /// Kernel-mediated second reference for another live process generation.
    pub fn grant_kernel(&mut self, owner: P, recipient: P, source: Handle,
        requested: Rights) -> Result<Handle, Error> {
        let source = self.lookup(owner, source)?;
        if recipient == owner || !requested.is_nonempty_subset_of(source.rights) {
            return Err(Error::Access);
        }
        let resource = self.resources[source.resource].ok_or(Error::BadFd)?;
        if resource.closed || resource.refs == 0 { return Err(Error::BadFd); }
        let following = resource.refs.checked_add(1).ok_or(Error::IdExhausted)?;
        let slot = self.entries.iter().position(Option::is_none).ok_or(Error::Full)?;
        let (handle, generation) = self.next_handle(slot)?;
        self.entries[slot] = Some(SharedEntry
            {handle, owner: recipient, resource: source.resource, rights: requested});
        self.resources[source.resource] = Some(ResourceRecord {refs: following, ..resource});
        self.generations[slot] = generation;
        Ok(handle)
    }

    pub fn query(&self, owner: P, handle: Handle) -> Result<Object, Error> {
        let entry = self.lookup(owner, handle)?;
        let resource = self.resources[entry.resource].ok_or(Error::BadFd)?;
        if resource.closed || resource.refs == 0 { return Err(Error::BadFd); }
        Ok(resource.object)
    }

    pub fn channel_access(&self, owner: P, handle: Handle,
        required: Rights) -> Result<u64, Error> {
        let entry = self.lookup(owner, handle)?;
        let resource = self.resources[entry.resource].ok_or(Error::BadFd)?;
        if resource.closed || resource.refs == 0 { return Err(Error::BadFd); }
        let Object::Channel(id) = resource.object else { return Err(Error::BadFd); };
        if !entry.rights.contains(required) { return Err(Error::Access); }
        Ok(id)
    }

    pub fn watch_close(&mut self, owner: P, handle: Handle)
        -> Result<WaitToken, Error> {
        let entry = self.lookup(owner, handle)?;
        let slot = self.watchers.iter().position(Option::is_none).ok_or(Error::Full)?;
        let following = self.next_wait.checked_add(1).ok_or(Error::IdExhausted)?;
        let token = WaitToken(self.next_wait);
        self.watchers[slot] = Some(Watcher
            {token, owner, resource: entry.resource});
        self.next_wait = following;
        Ok(token)
    }

    pub fn poll_close(&mut self, owner: P, token: WaitToken)
        -> Result<WaitResult, Error> {
        let slot = self.watchers.iter().position(|item| item.is_some_and(
            |item| item.owner == owner && item.token == token))
            .ok_or(Error::BadFd)?;
        let watcher = self.watchers[slot].ok_or(Error::BadFd)?;
        let resource = self.resources[watcher.resource].ok_or(Error::BadFd)?;
        if !resource.closed { return Ok(WaitResult::Pending); }
        if resource.refs != 0 { return Err(Error::BadFd); }
        self.watchers[slot] = None;
        if !self.watchers.iter().flatten()
            .any(|item| item.resource == watcher.resource) {
            self.resources[watcher.resource] = None;
        }
        Ok(WaitResult::Closed)
    }

    pub fn close(&mut self, owner: P, handle: Handle)
        -> Result<CloseResult, Error> {
        let entry = self.lookup(owner, handle)?;
        let slot = usize::try_from(handle.slot() - 1).map_err(|_| Error::BadFd)?;
        let resource = self.resources[entry.resource].ok_or(Error::BadFd)?;
        if resource.closed || resource.refs == 0 { return Err(Error::BadFd); }
        self.entries[slot] = None;
        if resource.refs > 1 {
            let remaining = resource.refs - 1;
            self.resources[entry.resource] = Some(ResourceRecord
                {refs: remaining, ..resource});
            Ok(CloseResult::Remaining(remaining))
        } else {
            if self.watchers.iter().flatten()
                .any(|item| item.resource == entry.resource) {
                self.resources[entry.resource] = Some(ResourceRecord
                    {refs: 0, closed: true, ..resource});
            } else {
                self.resources[entry.resource] = None;
            }
            Ok(CloseResult::LastClosed)
        }
    }

    pub fn close_owner(&mut self, owner: P) -> usize {
        let mut closed = 0;
        for slot in 0..N {
            if let Some(entry) = self.entries[slot].filter(|entry| entry.owner == owner) {
                if self.close(owner, entry.handle).is_ok() { closed += 1; }
            }
        }
        closed
    }

    pub fn snapshot(&self) -> SharedSnapshot<P, N> {
        SharedSnapshot {generations: self.generations,
            entries: self.entries.map(|item| item.map(|item|
                (item.handle.raw(), item.owner, item.resource, item.rights))),
            resources: self.resources.map(|item| item.map(|item|
                (item.object, item.refs, item.closed))),
            watchers: self.watchers.map(|item| item.map(|item|
                (item.token.raw(), item.owner, item.resource))),
            next_wait: self.next_wait}
    }
}

impl<P: Copy + Eq, const N: usize> Default for SharedObjects<P, N> {
    fn default() -> Self { Self::new() }
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct Message<const MAX: usize> {
    bytes: [u8; MAX],
    len: usize,
}

impl<const MAX: usize> Message<MAX> {
    pub fn as_slice(&self) -> &[u8] { &self.bytes[..self.len] }
    pub fn len(&self) -> usize { self.len }
    pub fn is_empty(&self) -> bool { self.len == 0 }
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct QueueSnapshot<P, const HANDLES: usize, const DEPTH: usize,
    const MAX: usize> where P: Copy + Eq {
    pub objects: SharedSnapshot<P, HANDLES>,
    pub slots: [Option<Message<MAX>>; DEPTH],
    pub head: usize,
    pub len: usize,
}

pub struct BoundedChannel<P: Copy + Eq, const HANDLES: usize,
    const DEPTH: usize, const MAX: usize> {
    objects: SharedObjects<P, HANDLES>,
    channel_id: u64,
    slots: [Option<Message<MAX>>; DEPTH],
    head: usize,
    len: usize,
}

impl<P: Copy + Eq, const HANDLES: usize, const DEPTH: usize,
    const MAX: usize> BoundedChannel<P, HANDLES, DEPTH, MAX> {
    pub const fn new(channel_id: u64) -> Self {
        Self {objects: SharedObjects::new(), channel_id,
            slots: [None; DEPTH], head: 0, len: 0}
    }

    pub fn open_kernel(&mut self, owner: P) -> Result<Handle, Error> {
        self.objects.open_kernel(owner, Object::Channel(self.channel_id),
            Rights::READ_WRITE)
    }

    pub fn grant_kernel(&mut self, owner: P, recipient: P, source: Handle,
        requested: Rights) -> Result<Handle, Error> {
        self.objects.grant_kernel(owner, recipient, source, requested)
    }

    fn access(&self, owner: P, handle: Handle,
        required: Rights) -> Result<(), Error> {
        if self.objects.channel_access(owner, handle, required)?
            != self.channel_id { return Err(Error::BadFd); }
        Ok(())
    }

    pub fn send(&mut self, owner: P, handle: Handle,
        bytes: &[u8]) -> Result<(), Error> {
        self.send_from(owner, handle, bytes.len(),
            |destination| { destination.copy_from_slice(bytes); Ok(()) })
    }

    /// Copy user data into a private scratch buffer before publishing a slot.
    /// A partial copy followed by Fault leaves the queue unchanged.
    pub fn send_from<F>(&mut self, owner: P, handle: Handle,
        length: usize, copy: F) -> Result<(), Error>
        where F: FnOnce(&mut [u8]) -> Result<(), Error> {
        self.access(owner, handle, Rights::WRITE)?;
        if length > MAX { return Err(Error::MessageTooLarge); }
        if self.len == DEPTH { return Err(Error::WouldBlock); }
        let mut bytes = [0u8; MAX];
        copy(&mut bytes[..length])?;
        let slot = (self.head + self.len) % DEPTH;
        debug_assert!(self.slots[slot].is_none());
        self.slots[slot] = Some(Message {bytes, len: length});
        self.len += 1;
        Ok(())
    }

    pub fn receive(&mut self, owner: P, handle: Handle)
        -> Result<Option<Message<MAX>>, Error> {
        self.access(owner, handle, Rights::READ)?;
        if self.len == 0 { return Ok(None); }
        let message = self.slots[self.head].take().ok_or(Error::BadFd)?;
        self.head = (self.head + 1) % DEPTH;
        self.len -= 1;
        Ok(Some(message))
    }

    pub fn peek(&self, owner: P, handle: Handle)
        -> Result<Option<Message<MAX>>, Error> {
        self.access(owner, handle, Rights::READ)?;
        if self.len == 0 { return Ok(None); }
        self.slots[self.head].ok_or(Error::BadFd).map(Some)
    }

    pub fn close(&mut self, owner: P, handle: Handle)
        -> Result<CloseResult, Error> {
        let result = self.objects.close(owner, handle)?;
        if result == CloseResult::LastClosed {
            self.slots = [None; DEPTH];
            self.head = 0;
            self.len = 0;
        }
        Ok(result)
    }

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

    pub fn snapshot(&self) -> QueueSnapshot<P, HANDLES, DEPTH, MAX> {
        QueueSnapshot {objects: self.objects.snapshot(), slots: self.slots,
            head: self.head, len: self.len}
    }
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct OwnedMessage<P: Copy + Eq, const MAX: usize> {
    pub sender: P,
    pub payload: Message<MAX>,
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct OwnedSnapshot<P: Copy + Eq, const HANDLES: usize,
    const DEPTH: usize, const MAX: usize> {
    pub queue: QueueSnapshot<P, HANDLES, DEPTH, MAX>,
    pub senders: [Option<P>; DEPTH],
}

pub struct OwnedChannel<P: Copy + Eq, const HANDLES: usize,
    const DEPTH: usize, const MAX: usize> {
    base: BoundedChannel<P, HANDLES, DEPTH, MAX>,
    senders: [Option<P>; DEPTH],
}

impl<P: Copy + Eq, const HANDLES: usize, const DEPTH: usize,
    const MAX: usize> OwnedChannel<P, HANDLES, DEPTH, MAX> {
    pub const fn new(channel_id: u64) -> Self {
        Self {base: BoundedChannel::new(channel_id), senders: [None; DEPTH]}
    }

    pub fn open_kernel(&mut self, owner: P) -> Result<Handle, Error> {
        self.base.open_kernel(owner)
    }

    pub fn grant_kernel(&mut self, owner: P, recipient: P, source: Handle,
        requested: Rights) -> Result<Handle, Error> {
        self.base.grant_kernel(owner, recipient, source, requested)
    }

    pub fn send(&mut self, owner: P, handle: Handle,
        bytes: &[u8]) -> Result<(), Error> {
        self.send_from(owner, handle, bytes.len(),
            |destination| { destination.copy_from_slice(bytes); Ok(()) })
    }

    pub fn send_from<F>(&mut self, owner: P, handle: Handle,
        length: usize, copy: F) -> Result<(), Error>
        where F: FnOnce(&mut [u8]) -> Result<(), Error> {
        // The base queue checks access, length and capacity and completes the
        // private copy before it publishes a slot. Provenance follows commit.
        let slot = if DEPTH == 0 { 0 } else {
            (self.base.head + self.base.len) % DEPTH
        };
        self.base.send_from(owner, handle, length, copy)?;
        self.senders[slot] = Some(owner);
        Ok(())
    }

    pub fn peek(&self, owner: P, handle: Handle)
        -> Result<Option<OwnedMessage<P, MAX>>, Error> {
        let Some(payload) = self.base.peek(owner, handle)? else {
            return Ok(None);
        };
        let sender = self.senders[self.base.head].ok_or(Error::BadFd)?;
        Ok(Some(OwnedMessage {sender, payload}))
    }

    pub fn receive(&mut self, owner: P, handle: Handle)
        -> Result<Option<OwnedMessage<P, MAX>>, Error> {
        let Some(front) = self.peek(owner, handle)? else { return Ok(None); };
        let slot = self.base.head;
        if self.base.receive(owner, handle)? != Some(front.payload) {
            return Err(Error::BadFd);
        }
        self.senders[slot] = None;
        Ok(Some(front))
    }

    pub fn close(&mut self, owner: P, handle: Handle)
        -> Result<CloseResult, Error> {
        let result = self.base.close(owner, handle)?;
        if result == CloseResult::LastClosed {
            self.senders = [None; DEPTH];
        }
        Ok(result)
    }

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

    pub fn snapshot(&self) -> OwnedSnapshot<P, HANDLES, DEPTH, MAX> {
        OwnedSnapshot {queue: self.base.snapshot(), senders: self.senders}
    }
}