fix(pipe): correct waitqueue wakeup semantics (#2148)

* fix(pipe): correct waitqueue wakeup semantics

Track blocked writer intents while holding the pipe state lock and choose single or broadcast wakeups according to whether heterogeneous write predicates may be waiting. Reuse the lock guard returned by the wait path so predicate publication, sleep, and revalidation remain ordered.

Move waitqueue, epoll, async I/O, and SIGPIPE notifications outside pipe spinlocks. Preserve partial-write results and observer notifications when readers close, and align zero-length, nonblocking EPIPE, resize, splice, and tee behavior with Linux semantics.

Prevent splice-held data from being consumed twice and keep tee source data intact while preserving partial progress. Add deterministic dunitest coverage for endpoint side effects, writer eligibility, partial-write notifications, splice and tee wakeups, reader baton passing, and resize threshold changes.

Signed-off-by: longjin <longjin@dragonos.org>

* fix(pipe): select eligible writer waiters

Replace unconditional multi-writer broadcasts with predicate-tagged waits. Track the free-space requirement of each blocked writer, choose one eligible class in round-robin order, and wake only that class to avoid thundering-herd retries.

Preserve progress with bounded fallback across the register-before-enqueue window. Pass the writer baton only after the selected write or splice operation completes or aborts, and pass the reader baton when an interrupted reader leaves consumable data behind.

Keep tagged waitqueue lookup and cleanup bounded, reserve the ordinary waiter tag, and move wake-all destruction outside the IRQ-disabled critical section. Add homogeneous and heterogeneous writer, splice, tee, signal, and resize regression coverage.

Tests: make fmt

Tests: make kernel

Tests: pipe_waitqueue_wakeup_test (11/11)

Tests: pipe_release_test (8/8)

Tests: splice_concurrent_io_test (7/7)

Tests: epoll_timeout_budget_test (1/1)

Tests: TCPResetDuringClose (12/12, twice)
Signed-off-by: longjin <longjin@dragonos.org>

* fix(pipe): serialize splice writer transactions

Add a per-pipe writer transaction that coordinates writes, output-side splice operations, endpoint close, and pipe resizing. Blocking writers publish their tagged wait predicate before releasing the transaction and pass an eligible writer baton only when another published waiter remains.

Keep file-to-pipe splice ownership across the input read and commit so competing writers cannot steal previously observed capacity. Validate readers before consuming stream input, including nonblocking EPIPE handling, and retain destination-only transaction locking for pipe-to-pipe splice and tee to avoid ABBA.

Add Dunitest regressions for no-reader stream preservation and competing writer serialization. Bound regression reads with poll deadlines so incorrect behavior fails deterministically instead of timing out.

Signed-off-by: longjin <longjin@dragonos.org>

---------

Signed-off-by: longjin <longjin@dragonos.org>

Author: fslongjin

kernel/src/filesystem/vfs/syscall/sys_splice.rs

@@ -3,7 +3,7 @@ use crate::filesystem::vfs::file::FileMode;
 use crate::filesystem::vfs::FileFlags;
 use crate::filesystem::vfs::{file::File, syscall::SpliceFlags, FileType};
 use crate::ipc::kill::send_signal_to_pid;
-use crate::ipc::pipe::{LockedPipeInode, PIPE_BUF};
+use crate::ipc::pipe::LockedPipeInode;
 use crate::process::resource::RLimitID;
 use crate::process::ProcessManager;
 use crate::syscall::table::Syscall;
@@ -256,28 +256,19 @@ fn splice_file_to_pipe(
 
     let limit = len.min(4096);
     let trusted_read_limit = splice_trusted_file_read_limit(file, offset, limit);
+    let mut transaction = pipe_inode.begin_writer_transaction();
+    pipe_inode.ensure_splice_readers(&transaction)?;
     if trusted_read_limit == Some(0) {
         return Ok(0);
     }
     let wanted = trusted_read_limit.unwrap_or(limit);
 
     let space = if flags.contains(SpliceFlags::SPLICE_F_NONBLOCK) {
-        let space = pipe_inode.writable_len();
-        if space == 0 && pipe_inode.has_readers() {
-            return Err(SystemError::EAGAIN_OR_EWOULDBLOCK);
-        }
-        if trusted_read_limit.is_some()
-            && wanted <= PIPE_BUF
-            && space < wanted
-            && pipe_inode.has_readers()
-        {
-            return Err(SystemError::EAGAIN_OR_EWOULDBLOCK);
-        }
-        space
+        pipe_inode.writable_for_splice_nonblock(&transaction, trusted_read_limit.map(|_| wanted))?
     } else if trusted_read_limit.is_some() {
-        pipe_inode.wait_writable_for_splice(wanted)?
+        pipe_inode.wait_writable_for_splice(&mut transaction, wanted)?
     } else {
-        pipe_inode.wait_writable_any_for_splice()?
+        pipe_inode.wait_writable_any_for_splice(&mut transaction)?
     };
 
     let buf_size = if trusted_read_limit.is_some() && space == 0 {
@@ -290,11 +281,14 @@ fn splice_file_to_pipe(
     // 从文件读取
     // 为了满足 Linux 语义：若后续写入 pipe 被信号中断且未写入任何字节，
     // 则不应推进输入文件的 file position。
-    let (read_len, advance_file_pos) = if let Some(off) = offset {
-        (file.pread(off, buf_size, &mut buffer)?, false)
+    let read_result = if let Some(off) = offset {
+        file.pread(off, buf_size, &mut buffer)
+            .map(|read_len| (read_len, false))
     } else {
-        (file.read_noadv(buf_size, &mut buffer)?, true)
+        file.read_noadv(buf_size, &mut buffer)
+            .map(|read_len| (read_len, true))
     };
+    let (read_len, advance_file_pos) = read_result?;
 
     if read_len == 0 {
         return Ok(0);
@@ -303,11 +297,7 @@ fn splice_file_to_pipe(
     buffer.truncate(read_len);
 
     // 写入 pipe
-    let written = if flags.contains(SpliceFlags::SPLICE_F_NONBLOCK) {
-        pipe_inode.write_from_splice_nonblock(&buffer)
-    } else {
-        pipe.write(buffer.len(), &buffer)
-    };
+    let written = pipe_inode.write_from_splice_locked(&mut transaction, &buffer);
 
     match written {
         Ok(write_len) => {