Patch pipe2 (#364)

kernel/src/ipc/pipe.rs

@@ -2,8 +2,8 @@ use crate::{
     arch::{sched::sched, CurrentIrqArch},
     exception::InterruptArch,
     filesystem::vfs::{
-        core::generate_inode_id, FilePrivateData, FileSystem, FileType, IndexNode, Metadata,
+        core::generate_inode_id, file::FileMode, FilePrivateData, FileSystem, FileType, IndexNode,
-        PollStatus,
+        Metadata, PollStatus,
     },
     include::bindings::bindings::PROC_INTERRUPTIBLE,
     libs::{spinlock::SpinLock, wait_queue::WaitQueue},
@@ -32,10 +32,11 @@ pub struct InnerPipeInode {
     data: [u8; PIPE_BUFF_SIZE],
     /// INode 元数据
     metadata: Metadata,
+    flags: FileMode,
 }
 
 impl LockedPipeInode {
-    pub fn new() -> Arc<Self> {
+    pub fn new(flags: FileMode) -> Arc<Self> {
         let inner = InnerPipeInode {
             self_ref: Weak::default(),
             valid_cnt: 0,
@@ -48,7 +49,7 @@ impl LockedPipeInode {
             metadata: Metadata {
                 dev_id: 0,
                 inode_id: generate_inode_id(),
-                size: 0,
+                size: PIPE_BUFF_SIZE as i64,
                 blk_size: 0,
                 blocks: 0,
                 atime: TimeSpec::default(),
@@ -61,6 +62,7 @@ impl LockedPipeInode {
                 gid: 0,
                 raw_dev: 0,
             },
+            flags,
         };
         let result = Arc::new(Self(SpinLock::new(inner)));
         let mut guard = result.0.lock();
@@ -85,11 +87,15 @@ impl IndexNode for LockedPipeInode {
         // 加锁
         let mut inode = self.0.lock();
 
-        //如果管道里面没有数据，则唤醒写端，
+        // 如果管道里面没有数据，则唤醒写端，
         while inode.valid_cnt == 0 {
             inode.write_wait_queue.wakeup(PROC_INTERRUPTIBLE.into());
-
+            // 如果为非阻塞管道，直接返回错误
-            // 在读等待队列中睡眠，并释放锁
+            if inode.flags.contains(FileMode::O_NONBLOCK) {
+                drop(inode);
+                return Err(SystemError::EAGAIN_OR_EWOULDBLOCK);
+            }
+            // 否则在读等待队列中睡眠，并释放锁
             unsafe {
                 let irq_guard = CurrentIrqArch::save_and_disable_irq();
                 inode.read_wait_queue.sleep_without_schedule();
@@ -170,6 +176,11 @@ impl IndexNode for LockedPipeInode {
         while len + inode.valid_cnt as usize > PIPE_BUFF_SIZE {
             // 唤醒读端
             inode.read_wait_queue.wakeup(PROC_INTERRUPTIBLE.into());
+            // 如果为非阻塞管道，直接返回错误
+            if inode.flags.contains(FileMode::O_NONBLOCK) {
+                drop(inode);
+                return Err(SystemError::ENOMEM);
+            }
             // 解锁并睡眠
             unsafe {
                 let irq_guard = CurrentIrqArch::save_and_disable_irq();

kernel/src/ipc/syscall.rs

@@ -8,7 +8,7 @@ use crate::{
     filesystem::vfs::file::{File, FileMode},
     include::bindings::bindings::{pid_t, verify_area, NULL},
     kwarn,
-    syscall::{Syscall, SystemError},
+    syscall::{user_access::UserBufferWriter, Syscall, SystemError},
 };
 
 use super::{
@@ -22,25 +22,37 @@ use super::{
 };
 
 impl Syscall {
-    /// # 创建匿名管道
+    /// # 创建带参数的匿名管道
     ///
     /// ## 参数
     ///
     /// - `fd`: 用于返回文件描述符的数组
-    pub fn pipe(fd: &mut [i32]) -> Result<usize, SystemError> {
+    /// - `flags`:设置管道的参数
-        let pipe_ptr = LockedPipeInode::new();
+    pub fn pipe2(fd: *mut i32, flags: FileMode) -> Result<usize, SystemError> {
-        let read_file = File::new(pipe_ptr.clone(), FileMode::O_RDONLY)?;
+        if flags.contains(FileMode::O_NONBLOCK)
-        let write_file = File::new(pipe_ptr.clone(), FileMode::O_WRONLY)?;
+            || flags.contains(FileMode::O_CLOEXEC)
+            || flags.contains(FileMode::O_RDONLY)
+        {
+            let mut user_buffer =
+                UserBufferWriter::new(fd, core::mem::size_of::<[c_int; 2]>(), true)?;
+            let fd = user_buffer.buffer::<i32>(0)?;
+            let pipe_ptr = LockedPipeInode::new(flags);
+            let mut read_file = File::new(pipe_ptr.clone(), FileMode::O_RDONLY)?;
+            let mut write_file = File::new(pipe_ptr.clone(), FileMode::O_WRONLY)?;
+            if flags.contains(FileMode::O_CLOEXEC) {
+                read_file.set_close_on_exec(true);
+                write_file.set_close_on_exec(true);
+            }
+            let read_fd = current_pcb().alloc_fd(read_file, None)?;
+            let write_fd = current_pcb().alloc_fd(write_file, None)?;
 
-        let read_fd = current_pcb().alloc_fd(read_file, None)?;
+            fd[0] = read_fd;
-        let write_fd = current_pcb().alloc_fd(write_file, None)?;
+            fd[1] = write_fd;
-
+            Ok(0)
-        fd[0] = read_fd;
+        } else {
-        fd[1] = write_fd;
+            Err(SystemError::EINVAL)
-
+        }
-        return Ok(0);
     }
-
     pub fn kill(pid: pid_t, sig: c_int) -> Result<usize, SystemError> {
         let sig = SignalNumber::from(sig);
         if sig == SignalNumber::INVALID {

kernel/src/syscall/mod.rs

@@ -652,13 +652,13 @@ impl Syscall {
 
             SYS_CLOCK => Self::clock(),
             SYS_PIPE => {
-                let pipefd = args[0] as *mut c_int;
+                let pipefd: *mut i32 = args[0] as *mut c_int;
-                match UserBufferWriter::new(pipefd, core::mem::size_of::<[c_int; 2]>(), from_user) {
+                let arg1 = args[1];
-                    Err(e) => Err(e),
+                if pipefd.is_null() {
-                    Ok(mut user_buffer) => match user_buffer.buffer::<i32>(0) {
+                    Err(SystemError::EFAULT)
-                        Err(e) => Err(e),
+                } else {
-                        Ok(pipefd) => Self::pipe(pipefd),
+                    let flags = FileMode::from_bits_truncate(arg1 as u32);
-                    },
+                    Self::pipe2(pipefd, flags)
                 }
             }
 

user/apps/about/about.c

@@ -41,6 +41,5 @@ int main()
 {
     print_ascii_logo();
     print_copyright();
-
     return 0;
 }

user/apps/shell/cmd.c

@@ -35,6 +35,7 @@ struct built_in_cmd_t shell_cmds[] = {
     {"free", shell_cmd_free},
     {"help", shell_help},
     {"pipe", shell_pipe_test},
+    {"pipe2", shell_pipe2_test},
     {"kill", shell_cmd_kill},
 
 };

user/apps/shell/cmd_test.c

@@ -5,6 +5,7 @@
 #include <string.h>
 #include <time.h>
 #include <unistd.h>
+#include <fcntl.h>
 
 #define buf_SIZE 256 // 定义消息的最大长度
 int shell_pipe_test(int argc, char **argv)
@@ -80,4 +81,86 @@ int shell_pipe_test(int argc, char **argv)
         wait(NULL);   // 等待子进程结束
     }
     return 0;
-}
+}
+int shell_pipe2_test(int argc, char **argv)
+{
+    int fd[2], i, n;
+
+    pid_t pid;
+    int ret = pipe2(fd, O_NONBLOCK); // 创建一个管道
+    if (ret < 0)
+    {
+        printf("pipe error\n");
+        exit(1);
+    }
+    pid = fork(); // 创建一个子进程
+    if (pid < 0)
+    {
+        printf("fork error\n");
+        exit(1);
+    }
+    if (pid == 0)
+    {                 // 子进程
+        close(fd[1]); // 关闭管道的写端
+        for (i = 0; i < 10; i++)
+        {
+            char buf[buf_SIZE] = {0};
+            n = read(fd[0], buf, buf_SIZE); // 从管道的读端读取一条消息
+            if (n > 0)
+            {
+
+                printf("Child process received message: %s\n", buf); // 打印收到的消息
+                if (strcmp(buf, "quit") == 0)
+                {                                     // 如果收到的消息是"quit"
+                    printf("Child process exits.\n"); // 打印退出信息
+                    break;                            // 跳出循环
+                }
+                else
+                {                                                    // 如果收到的消息不是"quit"
+                    printf("Child process is doing something...\n"); // 模拟子进程做一些操作
+                    // usleep(1000);
+                }
+            }
+            else
+            {
+                printf("read error,buf is empty\n");
+            }
+        }
+        close(fd[0]); // 关闭管道的读端
+        exit(0);
+    }
+    else
+    {                 // 父进程
+        close(fd[0]); // 关闭管道的读端
+        for (i = 0; i < 100; i++)
+        {
+            char *msg = "hello world";
+            if (i < 99 & i > 0)
+            {
+                msg = "how are you";
+                // usleep(1000);
+            }
+            if (i == 99)
+            {
+                msg = "quit";
+                // usleep(1000);
+            }
+            n = strlen(msg);
+            printf("Parent process send:%s\n", msg);
+
+            int r = write(fd[1], msg, n); // 向管道的写端写入一条消息
+            if (r < 0)
+            {
+                printf("write error,buf is full\n");
+            }
+            if (strcmp(msg, "quit") == 0)
+            {                                      // 如果发送的消息是"quit"
+                printf("Parent process exits.\n"); // 打印退出信息
+                break;                             // 跳出循环
+            }
+        }
+        close(fd[1]); // 关闭管道的写端
+        wait(NULL);   // 等待子进程结束
+    }
+    return 0;
+}

user/apps/shell/cmd_test.h

@@ -1,4 +1,5 @@
 #pragma once
 
 #include "cmd.h"
-int shell_pipe_test(int argc, char **argv);
+int shell_pipe_test(int argc, char **argv);
+int shell_pipe2_test(int argc, char **argv);

user/libs/libc/src/unistd.c

@@ -74,6 +74,15 @@ int pipe(int fd[2])
 {
     return (int)syscall_invoke(SYS_PIPE, fd, 0, 0, 0, 0, 0, 0, 0);
 }
+/**
+ * @brief 调用带参数的匿名管道
+ *
+ * @return int 如果失败返回负数
+ */
+int pipe2(int fd[2], int flags)
+{
+    return (int)syscall_invoke(SYS_PIPE, fd, flags, 0, 0, 0, 0, 0, 0);
+}
 /**
  * @brief fork当前进程，但是与父进程共享VM、flags、fd
  *