feat(wq): non interruptable and non blocking

Signed-off-by: Anhad Singh <[email protected]>

Author: Andy-Python-Programmer

src/aero_kernel/src/drivers/tty/vtty.rs

@@ -32,7 +32,7 @@ use crate::mem::paging::VirtAddr;
 use crate::userland::scheduler;
 use crate::userland::task::Task;
 use crate::userland::terminal::TerminalDevice;
-use crate::utils::sync::{Mutex, WaitQueue};
+use crate::utils::sync::{Mutex, WaitQueue, WaitQueueFlags};
 
 #[cfg(target_arch = "x86_64")]
 use crate::drivers::keyboard::KeyCode;
@@ -247,11 +247,12 @@ impl INodeInterface for Tty {
         self.connected.fetch_sub(1, Ordering::SeqCst);
     }
 
-    fn read_at(&self, _flags: OpenFlags, _offset: usize, buffer: &mut [u8]) -> fs::Result<usize> {
-        self.block_queue
-            .block_on(&self.stdin, |future| future.is_complete())?;
-
-        let mut stdin = self.stdin.lock_irq();
+    fn read_at(&self, flags: OpenFlags, _offset: usize, buffer: &mut [u8]) -> fs::Result<usize> {
+        let mut stdin = self.block_queue.wait(
+            WaitQueueFlags::from(flags) | WaitQueueFlags::DISABLE_IRQ,
+            &self.stdin,
+            |future| future.is_complete(),
+        )?;
 
         // record the back buffer size before swapping
         stdin.swap_buffer();

src/aero_kernel/src/fs/eventfd.rs

@@ -54,19 +54,14 @@ impl INodeInterface for EventFd {
         Ok(None)
     }
 
-    fn read_at(
-        &self,
-        _flags: OpenFlags,
-        _offset: usize,
-        buffer: &mut [u8],
-    ) -> super::Result<usize> {
+    fn read_at(&self, flags: OpenFlags, _offset: usize, buffer: &mut [u8]) -> super::Result<usize> {
         let size = core::mem::size_of::<u64>();
         assert!(buffer.len() >= size);
 
         // SAFETY: We have above verified that it is safe to dereference
         //         the value.
         let value = unsafe { &mut *(buffer.as_mut_ptr().cast::<u64>()) };
-        let mut count = self.wq.block_on(&self.count, |e| **e != 0)?;
+        let mut count = self.wq.wait(flags.into(), &self.count, |e| **e != 0)?;
 
         *value = *count;
         *count = 0; // reset the counter

src/aero_kernel/src/fs/pipe.rs

@@ -22,7 +22,7 @@ use alloc::sync::Arc;
 use spin::Once;
 
 use crate::utils::buffer::Buffer;
-use crate::utils::sync::{Mutex, WaitQueue};
+use crate::utils::sync::{Mutex, WaitQueue, WaitQueueFlags};
 
 use super::cache::DirCacheItem;
 use super::file_table::FileHandle;
@@ -84,7 +84,7 @@ impl INodeInterface for Pipe {
             return Err(FileSystemError::WouldBlock);
         }
 
-        let mut buffer = self.readers.block_on(&self.queue, |lock| {
+        let mut buffer = self.readers.wait(flags.into(), &self.queue, |lock| {
             lock.has_data() || self.active_writers() == 0
         })?;
 

src/aero_kernel/src/main.rs

@@ -45,7 +45,7 @@
     cfg_match, // https://github.com/rust-lang/rust/issues/115585
     associated_type_defaults,
     new_zeroed_alloc, // https://github.com/rust-lang/rust/issues/129396
-    sync_unsafe_cell
+    sync_unsafe_cell,
 )]
 // TODO(andypython): can we remove the dependency of "prelude_import" and "lang_items"?
 //     `lang_items`     => is currently used for the personality function (`rust_eh_personality`).
@@ -59,6 +59,7 @@
 #![reexport_test_harness_main = "test_main"]
 #![warn(clippy::needless_pass_by_value)]
 #![deny(clippy::ptr_as_ptr)]
+#![allow(binary_asm_labels)]
 
 #[macro_use]
 extern crate aero_proc;
@@ -214,7 +215,7 @@ fn kernel_dbg_thread() {
 
     use crate::drivers::uart::{self, LineStatus, COM_1};
     use crate::userland::task::TaskId;
-    use crate::utils::sync::WaitQueue;
+    use crate::utils::sync::{WaitQueue, WaitQueueFlags};
 
     uart::setup_interrupts();
 
@@ -229,7 +230,7 @@ fn kernel_dbg_thread() {
 
         loop {
             let mut com_1 = input_wq
-                .block_on(com_1, |com_1| {
+                .wait(WaitQueueFlags::empty(), com_1, |com_1| {
                     com_1.line_status().contains(LineStatus::INPUT_FULL)
                 })
                 .unwrap();

src/aero_kernel/src/socket/netlink.rs

@@ -221,7 +221,7 @@ impl INodeInterface for NetLinkSocket {
 
     fn recv(
         &self,
-        _fd_flags: OpenFlags,
+        fd_flags: OpenFlags,
         message_hdr: &mut MessageHeader,
         flags: socket::MessageFlags,
     ) -> fs::Result<usize> {
@@ -239,7 +239,7 @@ impl INodeInterface for NetLinkSocket {
 
         let mut queue = self
             .recv_wq
-            .block_on(&self.recv_queue, |queue| !queue.is_empty())?;
+            .wait(fd_flags.into(), &self.recv_queue, |queue| !queue.is_empty())?;
 
         let mut bytes_copied = 0;
         dbg!(message_hdr.iovecs_mut());

src/aero_kernel/src/socket/tcp.rs

@@ -32,7 +32,7 @@ use crate::fs::{self, FileSystemError};
 use crate::net;
 use crate::net::shim::PacketSend;
 use crate::net::{tcp, NetworkDevice};
-use crate::utils::sync::{Mutex, WaitQueue};
+use crate::utils::sync::{Mutex, WaitQueue, WaitQueueFlags};
 
 // ./aero.py -- -netdev user,id=mynet0 -device e1000,netdev=mynet0,id=ck_nic0 -object
 // filter-dump,id=mynet0,netdev=mynet0,file=qemulog.log
@@ -97,7 +97,7 @@ impl TcpSocket {
             Err(TcpError::WouldBlock) => {
                 drop(tcp);
 
-                let mut socket = self.wq.block_on(&self.tcp, |tcp| {
+                let mut socket = self.wq.wait(flags.into(), &self.tcp, |tcp| {
                     tcp.as_ref()
                         .is_none_or(|socket| !socket.recv_queue.is_empty())
                 })?;
@@ -145,7 +145,8 @@ impl INodeInterface for TcpSocket {
             *tcp = Some(socket);
         }
 
-        let _ = self.wq.block_on(&self.tcp, |x| {
+        // FIXME: connect() should pass the fd.
+        let _ = self.wq.wait(WaitQueueFlags::empty(), &self.tcp, |x| {
             x.as_ref().unwrap().state() == State::Established
         });
 

src/aero_kernel/src/socket/udp.rs

@@ -187,17 +187,16 @@ impl INodeInterface for UdpSocket {
 
     fn recv(
         &self,
-        _fd_flags: OpenFlags,
+        fd_flags: OpenFlags,
         message_hdr: &mut MessageHeader,
         _flags: MessageFlags,
     ) -> fs::Result<usize> {
         // assert!(flags.is_empty());
 
-        if self.inner.lock_irq().incoming.is_empty() && self.is_non_block() {
-            return Err(FileSystemError::WouldBlock);
-        }
+        let mut this = self
+            .wq
+            .wait(fd_flags.into(), &self.inner, |e| !e.incoming.is_empty())?;
 
-        let mut this = self.wq.block_on(&self.inner, |e| !e.incoming.is_empty())?;
         let packet = this.incoming.pop().expect("recv: someone was greedy");
 
         let mut data = packet.as_slice().to_vec();

src/aero_kernel/src/socket/unix.rs

@@ -32,7 +32,7 @@ use crate::fs::inode::{DirEntry, FileType, INodeInterface, Metadata, PollFlags,
 use crate::fs::{FileSystemError, Path};
 
 use crate::mem::paging::VirtAddr;
-use crate::utils::sync::{Mutex, WaitQueue};
+use crate::utils::sync::{Mutex, WaitQueue, WaitQueueFlags};
 
 use super::SocketAddrRef;
 
@@ -243,14 +243,11 @@ impl INodeInterface for UnixSocket {
         _offset: usize,
         user_buffer: &mut [u8],
     ) -> fs::Result<usize> {
-        if self.buffer.lock_irq().is_empty() && flags.is_nonblock() {
-            return Err(FileSystemError::WouldBlock);
-        }
-
-        let mut buffer = self.wq.block_on(&self.buffer, |e| !e.is_empty())?;
+        let mut buf = self
+            .wq
+            .wait(flags.into(), &self.buffer, |e| !e.is_empty())?;
 
-        let read = buffer.read(user_buffer);
-        Ok(read)
+        Ok(buf.read(user_buffer))
     }
 
     fn write_at(&self, _offset: usize, buffer: &[u8]) -> fs::Result<usize> {
@@ -325,12 +322,16 @@ impl INodeInterface for UnixSocket {
         target.wq.notify_all();
         core::mem::drop(itarget); // release the lock
 
-        let _ = self.wq.block_on(&self.inner, |e| e.state.is_connected())?;
+        // FIXME: connect() should pass fd.
+        let _ = self.wq.wait(WaitQueueFlags::empty(), &self.inner, |e| {
+            e.state.is_connected()
+        })?;
         Ok(())
     }
 
     fn accept(&self, address: Option<(VirtAddr, &mut u32)>) -> fs::Result<Arc<UnixSocket>> {
-        let mut inner = self.wq.block_on(&self.inner, |e| {
+        // TODO: accept
+        let mut inner = self.wq.wait(WaitQueueFlags::empty(), &self.inner, |e| {
             e.state.queue().is_some_and(|x| !x.is_empty())
         })?;
 
@@ -387,11 +388,9 @@ impl INodeInterface for UnixSocket {
             _ => return Err(FileSystemError::NotConnected),
         };
 
-        if self.buffer.lock_irq().is_empty() && fd_flags.is_nonblock() {
-            return Err(FileSystemError::WouldBlock);
-        }
-
-        let mut buffer = self.wq.block_on(&self.buffer, |e| !e.is_empty())?;
+        let mut buffer = self
+            .wq
+            .wait(fd_flags.into(), &self.buffer, |e| !e.is_empty())?;
 
         if let Some(addr) = header.name_mut::<SocketAddrUnix>() {
             *addr = peer.inner.lock_irq().address.as_ref().cloned().unwrap();

src/aero_kernel/src/syscall/fs.rs

@@ -56,7 +56,6 @@ impl fmt::Display for FileDescriptor {
             let path = file_handle.inode.absolute_path();
             write!(f, "{{ {} -> {} }}", self.0, path)
         } else {
-            // invalid file descriptor
             write!(f, "{{ {} -> INVALID }}", self.0)
         }
     }
@@ -164,10 +163,7 @@ pub fn getdents(fd: FileDescriptor, buffer: &mut [u8]) -> Result<usize, SyscallE
 
 #[syscall]
 pub fn close(fd: FileDescriptor) -> Result<usize, SyscallError> {
-    let res = scheduler::get_scheduler()
-        .current_task()
-        .file_table
-        .close_file(fd.into());
+    let res = scheduler::current_thread().file_table.close_file(fd.into());
 
     if res {
         Ok(0)
@@ -218,11 +214,10 @@ pub fn mkdirat(dfd: usize, path: &Path) -> Result<usize, SyscallError> {
         // pathname is interpreted relative to the current working directory of the
         // calling task.
         if dfd as isize == aero_syscall::AT_FDCWD {
-            let cwd = scheduler::get_scheduler().current_task().cwd_dirent();
+            let cwd = scheduler::current_thread().cwd_dirent();
             (cwd.inode(), path.as_str())
         } else {
-            let handle = scheduler::get_scheduler()
-                .current_task()
+            let handle = scheduler::current_thread()
                 .file_table
                 .get_handle(dfd)
                 .ok_or(SyscallError::EBADFD)?;

src/aero_kernel/src/syscall/ipc.rs

@@ -18,7 +18,7 @@
 use crate::userland::scheduler::get_scheduler;
 use crate::userland::task::TaskId;
 
-use crate::utils::sync::{Mutex, WaitQueue};
+use crate::utils::sync::{Mutex, WaitQueue, WaitQueueFlags};
 
 use aero_syscall::SyscallError;
 use alloc::collections::VecDeque;
@@ -103,7 +103,9 @@ pub fn recv(pid_ptr: &mut usize, output: &mut [u8], block: usize) -> Result<usiz
     let mq = &current.message_queue;
     let mut our_queue = mq
         .blockqueue
-        .block_on(&mq.queue, |msg| msg.front().is_some())
+        .wait(WaitQueueFlags::empty(), &mq.queue, |msg| {
+            msg.front().is_some()
+        })
         .unwrap();
 
     let msg = our_queue

src/aero_kernel/src/userland/task/mod.rs

@@ -36,7 +36,7 @@ use crate::arch::task::ArchTask;
 use crate::fs::file_table::FileTable;
 use crate::syscall::ipc::MessageQueue;
 use crate::syscall::ExecArgs;
-use crate::utils::sync::{Mutex, WaitQueue};
+use crate::utils::sync::{Mutex, WaitQueue, WaitQueueError, WaitQueueFlags};
 
 use crate::userland::signals::Signals;
 
@@ -142,10 +142,10 @@ impl Zombies {
         pids: &[usize],
         status: &mut u32,
         flags: WaitPidFlags,
-    ) -> SignalResult<usize> {
+    ) -> Result<usize, WaitQueueError> {
         let mut captured = None;
 
-        self.block.block_on(&self.list, |l| {
+        self.block.wait(WaitQueueFlags::empty(), &self.list, |l| {
             let mut cursor = l.front_mut();
 
             while let Some(t) = cursor.get() {
@@ -488,7 +488,7 @@ impl Task {
         pid: isize,
         status: &mut u32,
         flags: WaitPidFlags,
-    ) -> SignalResult<usize> {
+    ) -> Result<usize, WaitQueueError> {
         if pid == -1 {
             // wait for any child process if no specific process is requested.
             //

src/aero_kernel/src/userland/terminal.rs

@@ -22,8 +22,9 @@ use alloc::vec::Vec;
 
 use spin::RwLock;
 
+use crate::fs;
 use crate::fs::inode::INodeInterface;
-use crate::utils::sync::{Mutex, WaitQueue};
+use crate::utils::sync::{Mutex, WaitQueue, WaitQueueFlags};
 
 use super::signals::SignalError;
 use super::task::sessions::{Group, SESSIONS};
@@ -114,8 +115,10 @@ impl LineDiscipline {
         *self.termios.lock() = termios;
     }
 
-    pub fn read(&self, target: &mut [u8]) -> Result<usize, SignalError> {
-        let mut buffer = self.wq.block_on(&self.buffer, |buf| !buf.is_empty())?;
+    pub fn read(&self, target: &mut [u8]) -> fs::Result<usize> {
+        let mut buffer = self
+            .wq
+            .wait(WaitQueueFlags::empty(), &self.buffer, |buf| !buf.is_empty())?;
 
         let size = core::cmp::min(target.len(), buffer.len());
         target[..size].copy_from_slice(&buffer.drain(..size).collect::<Vec<_>>());