From 046c8ced0da1fffa89003ba47d0b7389af1a4b7e Mon Sep 17 00:00:00 2001 From: Wildan M Date: Wed, 11 Mar 2026 08:12:07 +0700 Subject: [PATCH] Add lock token to pipe queue without solving borrow checker --- src/context/context.rs | 4 ++-- src/scheme/pipe.rs | 44 +++++++++++++++++++++++------------------- src/scheme/user.rs | 10 ++++++---- 3 files changed, 32 insertions(+), 26 deletions(-) diff --git a/src/context/context.rs b/src/context/context.rs index 94378851b0..0be56c7508 100644 --- a/src/context/context.rs +++ b/src/context/context.rs @@ -986,7 +986,7 @@ pub fn bulk_add_fds( descriptions: Vec>, payload: UserSliceRw, cloexec: bool, - token: &mut CleanLockToken, + token: &mut LockToken, ) -> Result { let cnt = descriptions.len(); if payload.len() != cnt * size_of::() { @@ -1020,7 +1020,7 @@ pub fn bulk_insert_fds( descriptions: Vec>, payload: UserSliceRw, cloexec: bool, - token: &mut CleanLockToken, + token: &mut LockToken, ) -> Result { let cnt = descriptions.len(); if payload.len() != cnt * size_of::() { diff --git a/src/scheme/pipe.rs b/src/scheme/pipe.rs index 1c0bd4df47..8503af2732 100644 --- a/src/scheme/pipe.rs +++ b/src/scheme/pipe.rs @@ -4,7 +4,6 @@ use core::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; use syscall::{data::GlobalSchemes, CallFlags}; use hashbrown::{hash_map::DefaultHashBuilder, HashMap}; -use spin::Mutex; use crate::{ context::{ @@ -12,7 +11,7 @@ use crate::{ file::{FileDescription, InternalFlags, LockedFileDescription}, }, event, - sync::{CleanLockToken, RwLock, WaitCondition, L1}, + sync::{CleanLockToken, Mutex, RwLock, WaitCondition, L1}, syscall::{ data::Stat, error::{Error, Result, EAGAIN, EBADF, EINTR, EINVAL, ENOENT, EPIPE}, @@ -100,14 +99,15 @@ impl KernelScheme for PipeScheme { if is_writer_not_reader && flags.contains(EVENT_WRITE) - && (pipe.queue.lock().len() <= MAX_QUEUE_SIZE + && (pipe.queue.lock(token.token()).len() <= MAX_QUEUE_SIZE || !pipe.reader_is_alive.load(Ordering::Acquire)) { ready |= EventFlags::EVENT_WRITE; } if !is_writer_not_reader && flags.contains(EVENT_READ) - && (!pipe.queue.lock().is_empty() || !pipe.writer_is_alive.load(Ordering::Acquire)) + && (!pipe.queue.lock(token.token()).is_empty() + || !pipe.writer_is_alive.load(Ordering::Acquire)) { ready |= EventFlags::EVENT_READ; } @@ -254,7 +254,8 @@ impl KernelScheme for PipeScheme { let pipe = Self::get_pipe(key, token)?; loop { - let mut vec = pipe.queue.lock(); + let mut vec = pipe.queue.lock(token.token()); + let (vec, mut token) = vec.token_split(); let (s1, s2) = vec.as_slices(); let s1_count = core::cmp::min(user_buf.len(), s1.len()); @@ -274,13 +275,13 @@ impl KernelScheme for PipeScheme { let _ = vec.drain(..bytes_read); if bytes_read > 0 { - event::trigger( + event::trigger_locked( GlobalSchemes::Pipe.scheme_id(), key | WRITE_NOT_READ_BIT, EVENT_WRITE, token, ); - pipe.write_condition.notify(token); + pipe.write_condition.notify_locked(token); return Ok(bytes_read); } else if user_buf.is_empty() { @@ -312,7 +313,8 @@ impl KernelScheme for PipeScheme { let pipe = Self::get_pipe(key, token)?; loop { - let mut vec = pipe.queue.lock(); + let mut vec = pipe.queue.lock(token.token()); + let (vec, mut token) = vec.token_split(); if !pipe.reader_is_alive.load(Ordering::Relaxed) { return Err(Error::new(EPIPE)); @@ -341,8 +343,8 @@ impl KernelScheme for PipeScheme { } if bytes_written > 0 { - event::trigger(GlobalSchemes::Pipe.scheme_id(), key, EVENT_READ, token); - pipe.read_condition.notify(token); + event::trigger_locked(GlobalSchemes::Pipe.scheme_id(), key, EVENT_READ, token); + pipe.read_condition.notify_locked(token); return Ok(bytes_written); } else if user_buf.is_empty() { @@ -390,7 +392,8 @@ impl KernelScheme for PipeScheme { }; loop { - let mut vec = pipe.fd_queue.lock(); + let mut vec = pipe.fd_queue.lock(token.token()); + let (vec, mut token) = vec.token_split(); if !pipe.reader_is_alive.load(Ordering::Relaxed) { return Err(Error::new(EPIPE)); @@ -412,8 +415,8 @@ impl KernelScheme for PipeScheme { let fds_written = vec.len() - before_len; if fds_written > 0 { - event::trigger(GlobalSchemes::Pipe.scheme_id(), key, EVENT_READ, token); - pipe.read_condition.notify(token); + event::trigger_locked(GlobalSchemes::Pipe.scheme_id(), key, EVENT_READ, token); + pipe.read_condition.notify_locked(token); return Ok(fds_written); } @@ -448,7 +451,8 @@ impl KernelScheme for PipeScheme { } loop { - let mut vec = pipe.fd_queue.lock(); + let mut vec = pipe.fd_queue.lock(token.token()); + let (vec, mut token) = vec.token_split(); let fds_available = vec.len(); let max_fds_read = payload.len() / core::mem::size_of::(); @@ -461,24 +465,24 @@ impl KernelScheme for PipeScheme { fds_to_transfer, payload, flags.contains(CallFlags::FD_CLOEXEC), - token, + &mut token, )?; } else { bulk_add_fds( fds_to_transfer, payload, flags.contains(CallFlags::FD_CLOEXEC), - token, + &mut token, )?; } - event::trigger( + event::trigger_locked( GlobalSchemes::Pipe.scheme_id(), key | WRITE_NOT_READ_BIT, EVENT_WRITE, token, ); - pipe.write_condition.notify(token); + pipe.write_condition.notify_locked(token); return Ok(fds_to_read); } @@ -495,9 +499,9 @@ impl KernelScheme for PipeScheme { pub struct Pipe { read_condition: WaitCondition, // signals whether there are available bytes to read write_condition: WaitCondition, // signals whether there is room for additional bytes - queue: Mutex>, + queue: Mutex>, reader_is_alive: AtomicBool, // starts set, unset when reader closes writer_is_alive: AtomicBool, // starts set, unset when writer closes has_run_dup: AtomicBool, - fd_queue: Mutex>>, + fd_queue: Mutex>>, } diff --git a/src/scheme/user.rs b/src/scheme/user.rs index b73222d23a..f39d888209 100644 --- a/src/scheme/user.rs +++ b/src/scheme/user.rs @@ -1236,19 +1236,20 @@ impl UserInner { _ => return Err(Error::new(ENOENT)), }; + let mut token = token.downgrade(); let num_fds = if flags.contains(FobtainFdFlags::UPPER_TBL) { bulk_insert_fds( descriptions, payload, flags.contains(FobtainFdFlags::CLOEXEC), - token, + &mut token.token(), )? } else { bulk_add_fds( descriptions, payload, flags.contains(FobtainFdFlags::CLOEXEC), - token, + &mut token.token(), )? }; @@ -2056,20 +2057,21 @@ impl KernelScheme for UserScheme { Response::MultipleFds(fds) => fds, }; + let mut token = token.downgrade(); let num_fds = if let Some(descriptions) = descriptions_opt { if recvfd_flags.contains(RecvFdFlags::UPPER_TBL) { bulk_insert_fds( descriptions, payload, recvfd_flags.contains(RecvFdFlags::CLOEXEC), - token, + &mut token, )? } else { bulk_add_fds( descriptions, payload, recvfd_flags.contains(RecvFdFlags::CLOEXEC), - token, + &mut token, )? } } else {