strat9_kernel/vfs/

pipe.rs

//! Kernel pipe implementation.
//!
//! A pipe is a unidirectional byte stream between two file descriptors.
//! The read end blocks when empty; the write end returns EPIPE when
//! the read end is closed.

use crate::{
    sync::{waitqueue::WaitQueue, SpinLock},
    syscall::error::SyscallError,
};
use alloc::sync::Arc;

const PIPE_BUF_SIZE: usize = 4096;

/// Shared state for one pipe instance.
struct PipeInner {
    buf: [u8; PIPE_BUF_SIZE],
    read_pos: usize,
    write_pos: usize,
    /// Number of bytes currently buffered.
    len: usize,
    read_closed: bool,
    write_closed: bool,
    /// Number of open file-descriptions referencing the read end.
    /// The end is marked closed only when this reaches zero.
    read_refs: usize,
    /// Number of open file-descriptions referencing the write end.
    write_refs: usize,
}

impl PipeInner {
    /// Creates a new instance.
    fn new() -> Self {
        PipeInner {
            buf: [0u8; PIPE_BUF_SIZE],
            read_pos: 0,
            write_pos: 0,
            len: 0,
            read_closed: false,
            write_closed: false,
            read_refs: 1,
            write_refs: 1,
        }
    }

    /// Returns whether empty.
    #[allow(dead_code)]
    fn is_empty(&self) -> bool {
        self.len == 0
    }

    /// Returns whether full.
    #[allow(dead_code)]
    fn is_full(&self) -> bool {
        self.len >= PIPE_BUF_SIZE
    }

    /// Performs the available read operation.
    fn available_read(&self) -> usize {
        self.len
    }

    /// Performs the available write operation.
    fn available_write(&self) -> usize {
        PIPE_BUF_SIZE - self.len
    }
}

/// Shared pipe handle.
pub struct Pipe {
    inner: SpinLock<PipeInner>,
    /// Woken when data becomes available (or write end is closed).
    readers: WaitQueue,
    /// Woken when space becomes available (or read end is closed).
    writers: WaitQueue,
}

impl Pipe {
    /// Creates a new instance.
    pub fn new() -> Arc<Self> {
        Arc::new(Pipe {
            inner: SpinLock::new(PipeInner::new()),
            readers: WaitQueue::new(),
            writers: WaitQueue::new(),
        })
    }

    /// Read from the pipe. Returns 0 on EOF (write end closed + empty).
    /// If `non_block` is true and no data is available, returns EAGAIN.
    pub fn read(&self, buf: &mut [u8], non_block: bool) -> Result<usize, SyscallError> {
        loop {
            // Phase 1: try to consume data without blocking.
            {
                let mut inner = self.inner.lock();
                if inner.available_read() > 0 {
                    let to_read = core::cmp::min(buf.len(), inner.available_read());
                    for i in 0..to_read {
                        buf[i] = inner.buf[inner.read_pos];
                        inner.read_pos = (inner.read_pos + 1) % PIPE_BUF_SIZE;
                    }
                    inner.len -= to_read;
                    self.writers.wake_one();
                    return Ok(to_read);
                }
                if inner.write_closed {
                    return Ok(0); // EOF
                }
                if non_block {
                    return Err(SyscallError::Again);
                }
            }

            // Phase 2: wait for data or close.
            let result = self.readers.wait_until(|| {
                let inner = self.inner.lock();
                if inner.available_read() > 0 {
                    return Some(true);
                }
                if inner.write_closed {
                    return Some(false);
                }
                None
            });

            match result {
                true => continue,
                false => return Ok(0),
            }
        }
    }

    /// Write to the pipe. Returns EPIPE if read end is closed.
    /// If `non_block` is true and buffer is full, returns EAGAIN.
    pub fn write(&self, buf: &[u8], non_block: bool) -> Result<usize, SyscallError> {
        if buf.is_empty() {
            return Ok(0);
        }

        let mut total = 0;
        while total < buf.len() {
            let non_block = non_block;
            let wrote_some = self.writers.wait_until(|| {
                let mut inner = self.inner.lock();

                if inner.read_closed {
                    if total > 0 {
                        return Some(Ok(0));
                    }
                    return Some(Err(SyscallError::Pipe));
                }

                if inner.available_write() > 0 {
                    let to_write = core::cmp::min(buf.len() - total, inner.available_write());
                    for i in 0..to_write {
                        let wp = inner.write_pos;
                        inner.buf[wp] = buf[total + i];
                        inner.write_pos = (inner.write_pos + 1) % PIPE_BUF_SIZE;
                    }
                    inner.len += to_write;
                    total += to_write;
                    self.readers.wake_one();
                    return Some(Ok(0));
                }

                if non_block {
                    return Some(Err(SyscallError::Again));
                }

                None
            });

            match wrote_some {
                Ok(_) => {
                    if total >= buf.len() {
                        break;
                    }
                }
                // Transient: buffer full in non-blocking mode : surface EAGAIN.
                Err(SyscallError::Again) if non_block => return Err(SyscallError::Again),
                // F13 FIX (L2 host harness, testing-findings.md): EPIPE from
                // a closed read end was silently swallowed by the previous
                // catch-all arm, spinning the writing task forever at 100%
                // CPU instead of failing with EPIPE. Propagate it.
                Err(e) => return Err(e),
            }
        }
        Ok(total)
    }

    /// Increment the read-end refcount (called on dup/fork).
    pub fn dup_read(&self) {
        self.inner.lock().read_refs += 1;
    }

    /// Increment the write-end refcount (called on dup/fork).
    pub fn dup_write(&self) {
        self.inner.lock().write_refs += 1;
    }

    /// Decrement the read-end refcount; marks the end closed only when it
    /// reaches zero.  Returns true if the end was actually closed.
    pub fn close_read(&self) -> bool {
        let mut inner = self.inner.lock();
        if inner.read_refs == 0 {
            return false;
        }
        inner.read_refs -= 1;
        if inner.read_refs == 0 {
            inner.read_closed = true;
            drop(inner);
            self.writers.wake_all();
            true
        } else {
            false
        }
    }

    /// Decrement the write-end refcount; marks the end closed only when it
    /// reaches zero.  Returns true if the end was actually closed.
    pub fn close_write(&self) -> bool {
        let mut inner = self.inner.lock();
        if inner.write_refs == 0 {
            return false;
        }
        inner.write_refs -= 1;
        if inner.write_refs == 0 {
            inner.write_closed = true;
            drop(inner);
            self.readers.wake_all();
            true
        } else {
            false
        }
    }
}

// ============================================================================
// Pipe as a VFS Scheme
// ============================================================================

use super::scheme::{
    finalize_pseudo_stat, DirEntry, FileStat, OpenFlags, OpenResult, Scheme, DEV_PIPEFS,
};
use alloc::{collections::BTreeMap, vec::Vec};
use core::sync::atomic::{AtomicU64, Ordering};

/// A scheme that manages kernel pipes.
///
/// Each pipe gets two file_ids: even = read end, odd = write end.
pub struct PipeScheme {
    pipes: SpinLock<BTreeMap<u64, Arc<Pipe>>>,
    /// Per-fd open flags (for O_NONBLOCK).
    flags: SpinLock<BTreeMap<u64, OpenFlags>>,
}

static NEXT_PIPE_ID: AtomicU64 = AtomicU64::new(2); // Start at 2 (even numbers)

impl PipeScheme {
    /// Creates a new instance.
    pub fn new() -> Self {
        PipeScheme {
            pipes: SpinLock::new(BTreeMap::new()),
            flags: SpinLock::new(BTreeMap::new()),
        }
    }

    /// Create a new pipe pair. Returns (read_file_id, write_file_id).
    pub fn create_pipe(&self) -> (u64, Arc<Pipe>) {
        let base_id = NEXT_PIPE_ID.fetch_add(2, Ordering::SeqCst);
        let pipe = Pipe::new();
        self.pipes.lock().insert(base_id, pipe.clone());
        (base_id, pipe)
    }

    /// Store open flags for a pipe fd.
    pub fn set_flags(&self, file_id: u64, flags: OpenFlags) {
        self.flags.lock().insert(file_id, flags);
    }

    /// Returns pipe.
    fn get_pipe(&self, file_id: u64) -> Result<Arc<Pipe>, SyscallError> {
        let base = file_id & !1; // Even = base
        self.pipes
            .lock()
            .get(&base)
            .cloned()
            .ok_or(SyscallError::BadHandle)
    }

    /// Returns whether read end.
    fn is_read_end(file_id: u64) -> bool {
        file_id & 1 == 0
    }
}

impl Scheme for PipeScheme {
    /// Performs the open operation.
    fn open(&self, _path: &str, _flags: OpenFlags) -> Result<OpenResult, SyscallError> {
        Err(SyscallError::NotSupported) // Pipes are created via sys_pipe, not open()
    }

    /// Performs the read operation.
    fn read(&self, file_id: u64, _offset: u64, buf: &mut [u8]) -> Result<usize, SyscallError> {
        if !Self::is_read_end(file_id) {
            return Err(SyscallError::PermissionDenied);
        }
        let pipe = self.get_pipe(file_id)?;
        let non_block = self
            .flags
            .lock()
            .get(&file_id)
            .map(|f| f.contains(OpenFlags::NONBLOCK))
            .unwrap_or(false);
        pipe.read(buf, non_block)
    }

    /// Performs the write operation.
    fn write(&self, file_id: u64, _offset: u64, buf: &[u8]) -> Result<usize, SyscallError> {
        if Self::is_read_end(file_id) {
            return Err(SyscallError::PermissionDenied);
        }
        let pipe = self.get_pipe(file_id)?;
        let non_block = self
            .flags
            .lock()
            .get(&file_id)
            .map(|f| f.contains(OpenFlags::NONBLOCK))
            .unwrap_or(false);
        pipe.write(buf, non_block)
    }

    /// Performs the close operation.
    fn close(&self, file_id: u64) -> Result<(), SyscallError> {
        let pipe = self.get_pipe(file_id)?;
        if Self::is_read_end(file_id) {
            pipe.close_read();
        } else {
            pipe.close_write();
        }

        // Remove the shared Pipe entry only when both ends are fully closed
        // (both refcounts have reached zero).
        let base = file_id & !1;
        let inner = pipe.inner.lock();
        if inner.read_closed && inner.write_closed {
            drop(inner);
            self.pipes.lock().remove(&base);
        }
        Ok(())
    }

    /// Performs the stat operation.
    fn stat(&self, file_id: u64) -> Result<FileStat, SyscallError> {
        let pipe = self.get_pipe(file_id)?;
        let inner = pipe.inner.lock();
        Ok(finalize_pseudo_stat(
            FileStat {
                st_ino: file_id,
                st_mode: 0o010600, // S_IFIFO | rw-------
                st_nlink: 1,
                st_size: inner.len as u64,
                st_blksize: PIPE_BUF_SIZE as u64,
                st_blocks: 0,
                ..FileStat::zeroed()
            },
            DEV_PIPEFS,
            0,
        ))
    }

    /// Performs the readdir operation.
    fn readdir(&self, _file_id: u64) -> Result<Vec<DirEntry>, SyscallError> {
        Err(SyscallError::InvalidArgument)
    }

    /// Performs the size operation.
    fn size(&self, file_id: u64) -> Result<u64, SyscallError> {
        let pipe = self.get_pipe(file_id)?;
        let len = pipe.inner.lock().len;
        Ok(len as u64)
    }
}