pipe: return EOF when no active writers

Signed-off-by: Andy-Python-Programmer <[email protected]>

Author: Andy-Python-Programmer

src/aero_kernel/src/fs/inode.rs

@@ -227,6 +227,10 @@ pub trait INodeInterface: Send + Sync {
         Err(FileSystemError::NotSocket)
     }
 
+    fn accept(&self, _address: SocketAddr) -> Result<()> {
+        Err(FileSystemError::NotSocket)
+    }
+
     /// Returns the inner UNIX socket inode if bound to one.
     fn as_unix_socket(&self) -> Result<Arc<dyn INodeInterface>> {
         Err(FileSystemError::NotSocket)

src/aero_kernel/src/fs/pipe.rs

@@ -1,3 +1,6 @@
+use core::sync::atomic::{AtomicUsize, Ordering};
+
+use aero_syscall::OpenFlags;
 use alloc::{sync::Arc, vec::Vec};
 
 use crate::utils::sync::{BlockQueue, Mutex};
@@ -38,41 +41,59 @@ impl Buffer {
     }
 }
 
-struct BufferQueue {
-    buffer: Mutex<Buffer>,
-}
-
-impl BufferQueue {
-    fn new() -> Self {
-        Self {
-            buffer: Mutex::new(Buffer::new()),
-        }
-    }
-}
-
 pub struct Pipe {
-    queue: BufferQueue,
+    queue: Mutex<Buffer>,
 
     readers: BlockQueue,
     writers: BlockQueue,
+
+    /// The number of writers currently connected to the pipe.
+    num_writers: AtomicUsize,
 }
 
 impl Pipe {
     pub fn new() -> Arc<Self> {
         Arc::new(Self {
-            queue: BufferQueue::new(),
+            queue: Mutex::new(Buffer::new()),
 
             readers: BlockQueue::new(),
             writers: BlockQueue::new(),
+
+            num_writers: AtomicUsize::new(0),
         })
     }
+
+    /// Returns whether the pipe has active writers.
+    pub fn has_active_writers(&self) -> usize {
+        self.num_writers.load(Ordering::SeqCst)
+    }
 }
 
 impl INodeInterface for Pipe {
+    fn open(&self, flags: OpenFlags) -> super::Result<()> {
+        // Write end of the pipe:
+        if flags.contains(OpenFlags::O_WRONLY) {
+            self.num_writers.fetch_add(1, Ordering::SeqCst);
+        }
+
+        Ok(())
+    }
+
+    fn close(&self, flags: OpenFlags) {
+        // Write end of the pipe:
+        if flags.contains(OpenFlags::O_WRONLY) {
+            let active_writers = (self.num_writers.fetch_sub(1, Ordering::SeqCst) - 1) == 0;
+            // There are no active writers and no data to read (reached EOF).
+            if active_writers {
+                self.readers.notify_complete();
+            }
+        }
+    }
+
     fn read_at(&self, _offset: usize, buf: &mut [u8]) -> super::Result<usize> {
-        let mut buffer = self
-            .readers
-            .block_on(&self.queue.buffer, |lock| lock.has_data())?;
+        let mut buffer = self.readers.block_on(&self.queue, |lock| {
+            lock.has_data() || !self.has_active_writers() == 0
+        })?;
 
         let read = buffer.read_data(buf);
 
@@ -85,7 +106,7 @@ impl INodeInterface for Pipe {
     }
 
     fn write_at(&self, offset: usize, buf: &[u8]) -> super::Result<usize> {
-        let res = offset + self.queue.buffer.lock().write_data(buf);
+        let res = offset + self.queue.lock().write_data(buf);
         self.readers.notify_complete();
 
         Ok(res)

src/aero_kernel/src/socket/unix.rs

@@ -163,6 +163,10 @@ impl INodeInterface for UnixSocket {
         Ok(())
     }
 
+    fn accept(&self, _address: SocketAddr) -> Result<()> {
+        unimplemented!()
+    }
+
     fn poll(&self, table: Option<&mut PollTable>) -> Result<PollFlags> {
         table.map(|e| e.insert(&self.wq));
 

src/aero_kernel/src/syscall/mod.rs

@@ -218,6 +218,7 @@ pub fn generic_do_syscall(
         SYS_BIND => net::bind(b, c, d),
         SYS_CONNECT => net::connect(b, c, d),
         SYS_LISTEN => net::listen(b, c),
+        SYS_ACCEPT => net::accept(b, c, d),
 
         SYS_GETTIME => time::gettime(b, c),
         SYS_SLEEP => time::sleep(b),

src/aero_kernel/src/syscall/net.rs

@@ -33,6 +33,20 @@ pub fn connect(fd: usize, address: usize, length: usize) -> Result<usize, Syscal
     Ok(0)
 }
 
+/// Accept a connection on a socket.
+#[syscall]
+pub fn accept(fd: usize, address: usize, _length: &mut u32) -> Result<usize, SyscallError> {
+    let address = socket_addr_from_addr(VirtAddr::new(address as u64))?;
+    let file = scheduler::get_scheduler()
+        .current_task()
+        .file_table
+        .get_handle(fd)
+        .ok_or(SyscallError::EINVAL)?;
+
+    file.inode().accept(address)?;
+    Ok(0)
+}
+
 /// Marks the socket as a passive socket (i.e. as a socket that will be used to accept incoming
 /// connection requests).
 #[syscall]

src/aero_kernel/src/userland/task.rs

@@ -450,6 +450,16 @@ impl Task {
     }
 
     pub(super) fn update_state(&self, state: TaskState) {
+        if state != TaskState::Runnable {
+            log::warn!(
+                "Task::update_state() updated the task state to {state:?}! (pid={:?}, tid={:?})",
+                self.pid,
+                self.tid
+            );
+
+            crate::unwind::unwind_stack_trace();
+        }
+
         self.state.store(state as _, Ordering::SeqCst);
     }