fix: pipe 读取/写入阻塞时,无法kill进程的问题 (#889)

kernel/Cargo.toml

@@ -37,6 +37,7 @@ bitmap = { path = "crates/bitmap" }
 driver_base_macros = { "path" = "crates/driver_base_macros" }
 # 一个no_std的hashmap、hashset
 elf = { version = "=0.7.2", default-features = false }
+fdt = { git = "https://git.mirrors.dragonos.org.cn/DragonOS-Community/fdt", rev = "9862813020" }
 hashbrown = "=0.13.2"
 ida = { path = "src/libs/ida" }
 intertrait = { path = "crates/intertrait" }
@@ -48,11 +49,11 @@ num-derive = "=0.3"
 num-traits = { git = "https://git.mirrors.dragonos.org.cn/DragonOS-Community/num-traits.git", rev="1597c1c", default-features = false }
 smoltcp = { version = "=0.11.0", default-features = false, features = ["log", "alloc",  "socket-raw", "socket-udp", "socket-tcp", "socket-icmp", "socket-dhcpv4", "socket-dns", "proto-ipv4", "proto-ipv6"]}
 system_error = { path = "crates/system_error" }
-unified-init = { path = "crates/unified-init" }
-virtio-drivers = { git = "https://git.mirrors.dragonos.org.cn/DragonOS-Community/virtio-drivers", rev = "f91c807965" }
-fdt = { git = "https://git.mirrors.dragonos.org.cn/DragonOS-Community/fdt", rev = "9862813020" }
 uefi = { version = "=0.26.0", features = ["alloc"] }
 uefi-raw = "=0.5.0"
+unified-init = { path = "crates/unified-init" }
+virtio-drivers = { git = "https://git.mirrors.dragonos.org.cn/DragonOS-Community/virtio-drivers", rev = "f91c807965" }
+wait_queue_macros = { path = "crates/wait_queue_macros" }
 paste = "=1.0.14"
 slabmalloc = { path = "crates/rust-slabmalloc" }
 log = "0.4.21"

kernel/crates/wait_queue_macros/Cargo.toml

@@ -0,0 +1,7 @@
+[package]
+name = "wait_queue_macros"
+version = "0.1.0"
+edition = "2021"
+authors = ["longjin <longjin@dragonos.org>"]
+
+[dependencies]

kernel/crates/wait_queue_macros/src/lib.rs

@@ -0,0 +1,60 @@
+#![no_std]
+
+/// Wait for a condition to become true.
+///
+/// This macro will wait for a condition to become true.
+///
+/// ## Parameters
+///
+/// - `$wq`: The wait queue to wait on.
+/// - `$condition`: The condition to wait for. (you can pass a function or a boolean expression)
+/// - `$cmd`: The command to execute while waiting.
+#[macro_export]
+macro_rules! wq_wait_event_interruptible {
+    ($wq:expr, $condition: expr, $cmd: expr) => {{
+        let mut retval = Ok(());
+        if !$condition {
+            retval = wait_queue_macros::_wq_wait_event_interruptible!($wq, $condition, $cmd);
+        }
+
+        retval
+    }};
+}
+
+#[macro_export]
+#[allow(clippy::crate_in_macro_def)]
+macro_rules! _wq_wait_event_interruptible {
+    ($wq:expr, $condition: expr, $cmd: expr) => {{
+        wait_queue_macros::__wq_wait_event!($wq, $condition, true, Ok(()), {
+            $cmd;
+            crate::sched::schedule(SchedMode::SM_NONE)
+        })
+    }};
+}
+
+#[macro_export]
+macro_rules! __wq_wait_event(
+    ($wq:expr, $condition: expr, $interruptible: expr, $ret: expr, $cmd:expr) => {{
+        let mut retval = $ret;
+        let mut exec_finish_wait = true;
+        loop {
+            let x = $wq.prepare_to_wait_event($interruptible);
+            if $condition {
+                break;
+            }
+
+            if $interruptible && !x.is_ok() {
+                retval = x;
+                exec_finish_wait = false;
+                break;
+            }
+
+            $cmd;
+        }
+        if exec_finish_wait {
+            $wq.finish_wait();
+        }
+
+        retval
+    }};
+);

kernel/src/arch/x86_64/interrupt/mod.rs

@@ -132,6 +132,12 @@ pub struct TrapFrame {
     pub ss: ::core::ffi::c_ulong,
 }
 
+impl Default for TrapFrame {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
 impl TrapFrame {
     pub fn new() -> Self {
         Self {

kernel/src/filesystem/vfs/file.rs

@@ -236,7 +236,14 @@ impl File {
 
         let len = self
             .inode
-            .read_at(offset, len, buf, self.private_data.lock())?;
+            .read_at(offset, len, buf, self.private_data.lock())
+            .map_err(|e| {
+                if e == SystemError::ERESTARTSYS {
+                    SystemError::EINTR
+                } else {
+                    e
+                }
+            })?;
 
         if update_offset {
             self.offset
@@ -261,11 +268,24 @@ impl File {
 
         // 如果文件指针已经超过了文件大小，则需要扩展文件大小
         if offset > self.inode.metadata()?.size as usize {
-            self.inode.resize(offset)?;
+            self.inode.resize(offset).map_err(|e| {
+                if e == SystemError::ERESTARTSYS {
+                    SystemError::EINTR
+                } else {
+                    e
+                }
+            })?;
         }
         let len = self
             .inode
-            .write_at(offset, len, buf, self.private_data.lock())?;
+            .write_at(offset, len, buf, self.private_data.lock())
+            .map_err(|e| {
+                if e == SystemError::ERESTARTSYS {
+                    SystemError::EINTR
+                } else {
+                    e
+                }
+            })?;
 
         if update_offset {
             self.offset
@@ -555,7 +575,11 @@ pub struct FileDescriptorVec {
     /// 当前进程打开的文件描述符
     fds: Vec<Option<Arc<File>>>,
 }
-
+impl Default for FileDescriptorVec {
+    fn default() -> Self {
+        Self::new()
+    }
+}
 impl FileDescriptorVec {
     pub const PROCESS_MAX_FD: usize = 1024;
 

kernel/src/ipc/pipe.rs

@@ -1,6 +1,4 @@
 use crate::{
-    arch::CurrentIrqArch,
-    exception::InterruptArch,
     filesystem::vfs::{
         core::generate_inode_id, file::FileMode, syscall::ModeType, FilePrivateData, FileSystem,
         FileType, IndexNode, Metadata,
@@ -11,7 +9,7 @@ use crate::{
     },
     net::event_poll::{EPollEventType, EPollItem, EventPoll},
     process::ProcessState,
-    sched::{schedule, SchedMode},
+    sched::SchedMode,
     time::PosixTimeSpec,
 };
 
@@ -106,6 +104,10 @@ impl InnerPipeInode {
         Ok(())
     }
 
+    fn buf_full(&self) -> bool {
+        return self.valid_cnt as usize == PIPE_BUFF_SIZE;
+    }
+
     pub fn remove_epoll(&self, epoll: &Weak<SpinLock<EventPoll>>) -> Result<(), SystemError> {
         let is_remove = !self
             .epitems
@@ -166,6 +168,16 @@ impl LockedPipeInode {
     pub fn inner(&self) -> &SpinLock<InnerPipeInode> {
         &self.inner
     }
+
+    fn readable(&self) -> bool {
+        let inode = self.inner.lock();
+        return inode.valid_cnt > 0 || inode.writer == 0;
+    }
+
+    fn writeable(&self) -> bool {
+        let inode = self.inner.lock();
+        return !inode.buf_full() || inode.reader == 0;
+    }
 }
 
 impl IndexNode for LockedPipeInode {
@@ -189,6 +201,7 @@ impl IndexNode for LockedPipeInode {
         if buf.len() < len {
             return Err(SystemError::EINVAL);
         }
+        // log::debug!("pipe mode: {:?}", mode);
         // 加锁
         let mut inode = self.inner.lock();
 
@@ -209,14 +222,12 @@ impl IndexNode for LockedPipeInode {
             }
 
             // 否则在读等待队列中睡眠，并释放锁
-            unsafe {
-                let irq_guard = CurrentIrqArch::save_and_disable_irq();
-
             drop(inode);
-                self.read_wait_queue.sleep_without_schedule();
-                drop(irq_guard);
+            let r = wq_wait_event_interruptible!(self.read_wait_queue, self.readable(), {});
+            if r.is_err() {
+                return Err(SystemError::ERESTARTSYS);
             }
-            schedule(SchedMode::SM_NONE);
+
             inode = self.inner.lock();
         }
 
@@ -367,13 +378,11 @@ impl IndexNode for LockedPipeInode {
             }
 
             // 解锁并睡眠
-            unsafe {
-                let irq_guard = CurrentIrqArch::save_and_disable_irq();
             drop(inode);
-                self.write_wait_queue.sleep_without_schedule();
-                drop(irq_guard);
+            let r = wq_wait_event_interruptible!(self.write_wait_queue, self.writeable(), {});
+            if r.is_err() {
+                return Err(SystemError::ERESTARTSYS);
             }
-            schedule(SchedMode::SM_NONE);
             inode = self.inner.lock();
         }
 

kernel/src/ipc/signal.rs

@@ -16,6 +16,35 @@ use super::signal_types::{
 };
 
 impl Signal {
+    pub fn signal_pending_state(
+        interruptible: bool,
+        task_wake_kill: bool,
+        pcb: &Arc<ProcessControlBlock>,
+    ) -> bool {
+        if !interruptible && !task_wake_kill {
+            return false;
+        }
+
+        if !pcb.has_pending_signal() {
+            return false;
+        }
+
+        return interruptible || Self::fatal_signal_pending(pcb);
+    }
+
+    /// 判断当前进程是否收到了SIGKILL信号
+    pub fn fatal_signal_pending(pcb: &Arc<ProcessControlBlock>) -> bool {
+        let guard = pcb.sig_info_irqsave();
+        if guard
+            .sig_pending()
+            .signal()
+            .contains(Signal::SIGKILL.into())
+        {
+            return true;
+        }
+
+        return false;
+    }
     /// 向目标进程发送信号
     ///
     /// ## 参数

kernel/src/ipc/signal_types.rs

@@ -81,6 +81,12 @@ impl SignalStruct {
     }
 }
 
+impl Default for SignalStruct {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
 impl Deref for SignalStruct {
     type Target = InnerSignalStruct;
 

kernel/src/lib.rs

@@ -12,6 +12,7 @@
 #![feature(c_void_variant)]
 #![feature(extract_if)]
 #![feature(fn_align)]
+#![feature(linked_list_retain)]
 #![feature(naked_functions)]
 #![feature(new_uninit)]
 #![feature(ptr_internals)]
@@ -83,6 +84,8 @@ extern crate x86;
 extern crate klog_types;
 extern crate uefi;
 extern crate uefi_raw;
+#[macro_use]
+extern crate wait_queue_macros;
 
 use crate::mm::allocator::kernel_allocator::KernelAllocator;
 

kernel/src/libs/wait_queue.rs

@@ -3,9 +3,10 @@ use core::intrinsics::unlikely;
 
 use alloc::{collections::LinkedList, sync::Arc, vec::Vec};
 use log::{error, warn};
+use system_error::SystemError;
 
 use crate::{
-    arch::CurrentIrqArch,
+    arch::{ipc::signal::Signal, CurrentIrqArch},
     exception::InterruptArch,
     process::{ProcessControlBlock, ProcessManager, ProcessState},
     sched::{schedule, SchedMode},
@@ -32,6 +33,34 @@ impl WaitQueue {
         WaitQueue(SpinLock::new(InnerWaitQueue::INIT))
     }
 
+    pub fn prepare_to_wait_event(&self, interruptible: bool) -> Result<(), SystemError> {
+        let mut guard: SpinLockGuard<InnerWaitQueue> = self.0.lock_irqsave();
+        let pcb = ProcessManager::current_pcb();
+        if Signal::signal_pending_state(interruptible, false, &pcb) {
+            return Err(SystemError::ERESTARTSYS);
+        } else {
+            ProcessManager::mark_sleep(interruptible).unwrap_or_else(|e| {
+                panic!("sleep error: {:?}", e);
+            });
+            guard.wait_list.push_back(ProcessManager::current_pcb());
+            drop(guard);
+        }
+        Ok(())
+    }
+
+    pub fn finish_wait(&self) {
+        let pcb = ProcessManager::current_pcb();
+        let mut writer = pcb.sched_info().inner_lock_write_irqsave();
+        let mut guard: SpinLockGuard<InnerWaitQueue> = self.0.lock_irqsave();
+
+        writer.set_state(ProcessState::Runnable);
+        writer.set_wakeup();
+
+        guard.wait_list.retain(|x| !Arc::ptr_eq(x, &pcb));
+        drop(guard);
+        drop(writer);
+    }
+
     /// @brief 让当前进程在等待队列上进行等待，并且，允许被信号打断
     pub fn sleep(&self) {
         before_sleep_check(0);
@@ -50,7 +79,7 @@ impl WaitQueue {
         F: FnOnce(),
     {
         before_sleep_check(0);
-        let mut guard: SpinLockGuard<InnerWaitQueue> = self.0.lock();
+        let mut guard: SpinLockGuard<InnerWaitQueue> = self.0.lock_irqsave();
         let irq_guard = unsafe { CurrentIrqArch::save_and_disable_irq() };
         ProcessManager::mark_sleep(true).unwrap_or_else(|e| {
             panic!("sleep error: {:?}", e);

kernel/src/mm/page.rs

@@ -458,6 +458,12 @@ pub struct PageFlags<Arch> {
     phantom: PhantomData<Arch>,
 }
 
+impl<Arch: MemoryManagementArch> Default for PageFlags<Arch> {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
 #[allow(dead_code)]
 impl<Arch: MemoryManagementArch> PageFlags<Arch> {
     #[inline(always)]

kernel/src/process/kthread.rs

@@ -80,6 +80,12 @@ impl KernelThreadPcbPrivate {
     }
 }
 
+impl Default for KernelThreadPcbPrivate {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
 /// 内核线程的闭包，参数必须与闭包的参数一致，返回值必须是i32
 ///
 /// 元组的第一个元素是闭包，第二个元素是闭包的参数对象

kernel/src/process/mod.rs

@@ -975,6 +975,14 @@ impl ProcessControlBlock {
         return None;
     }
 
+    /// 判断当前进程是否有未处理的信号
+    pub fn has_pending_signal(&self) -> bool {
+        let sig_info = self.sig_info_irqsave();
+        let has_pending = sig_info.sig_pending().has_pending();
+        drop(sig_info);
+        return has_pending;
+    }
+
     pub fn sig_struct(&self) -> SpinLockGuard<SignalStruct> {
         self.sig_struct.lock_irqsave()
     }

kernel/src/sched/fair.rs

@@ -1373,6 +1373,11 @@ impl CfsRunQueue {
     }
 }
 
+impl Default for CfsRunQueue {
+    fn default() -> Self {
+        Self::new()
+    }
+}
 pub struct CompletelyFairScheduler;
 
 impl CompletelyFairScheduler {