diff --git a/components/axpoll/src/axtest.rs b/components/axpoll/src/axtest.rs index bfb60ebe9e..226ec90a8d 100644 --- a/components/axpoll/src/axtest.rs +++ b/components/axpoll/src/axtest.rs @@ -77,6 +77,27 @@ fn axpoll_wakes_only_matching_interests() { ax_assert_eq!(unsafe { poll_set.wake(IoEvents::IN | IoEvents::OUT) }, 0); } +#[axtest] +fn axpoll_exclusive_wake_keeps_other_matching_waiters() { + let poll_set = PollSet::new(); + let first_counter = WakeCounter::new(); + let second_counter = WakeCounter::new(); + let first_waker = counter_waker(&first_counter); + let second_waker = counter_waker(&second_counter); + + unsafe { + poll_set.register(&first_waker, IoEvents::IN); + poll_set.register(&second_waker, IoEvents::IN); + } + + ax_assert_eq!(unsafe { poll_set.wake_one(IoEvents::IN) }, 1); + ax_assert_eq!(first_counter.count() + second_counter.count(), 1); + ax_assert_eq!(unsafe { poll_set.wake_one(IoEvents::IN) }, 1); + ax_assert_eq!(first_counter.count(), 1); + ax_assert_eq!(second_counter.count(), 1); + ax_assert_eq!(unsafe { poll_set.wake_one(IoEvents::IN) }, 0); +} + #[axtest] fn axpoll_capacity_overwrite_and_drop_rules_hold() { let poll_set = PollSet::new(); diff --git a/components/axpoll/src/lib.rs b/components/axpoll/src/lib.rs index b1d19f10e6..81d0dda56a 100644 --- a/components/axpoll/src/lib.rs +++ b/components/axpoll/src/lib.rs @@ -145,6 +145,24 @@ impl Inner { } old.cursor = 0; } + + fn take_one_ready(&mut self, ready: IoEvents) -> Option { + let len = self.len(); + let mut selected = None; + let mut keep_len = 0; + + for index in 0..len { + let entry = unsafe { self.entries[index].assume_init_read() }; + if selected.is_none() && entry.interests.intersects(ready) { + selected = Some(entry); + } else { + self.entries[keep_len].write(entry); + keep_len += 1; + } + } + self.cursor = keep_len; + selected + } } impl Drop for Inner { @@ -213,6 +231,31 @@ impl PollSet { woke } + /// Wakes one registered waker whose interests intersect `ready`. + /// + /// Matching wakers that are not selected remain registered. This is used + /// by wait queues with Linux-style exclusive wakeup semantics, where one + /// readiness transition should give one waiter a chance to consume work. + /// + /// # Safety + /// + /// This method is task/deferred-context only. Callers must not invoke it + /// from hard IRQ, NMI, or trap callbacks. The readiness state represented + /// by `ready` must be published before this method is called, and callers + /// must not hold locks that may be re-entered by waker execution or poll + /// wakeup paths. + pub unsafe fn wake_one(&self, ready: IoEvents) -> usize { + let Some(inner) = self.0.get() else { + return 0; + }; + let ready_entry = inner.lock().take_one_ready(ready); + let Some(entry) = ready_entry else { + return 0; + }; + entry.wake(); + 1 + } + /// Wakes up registered wakers whose interests intersect `ready` from IRQ context. /// /// Unlike [`wake`](Self::wake), this does not allocate a replacement diff --git a/components/axpoll/tests/tests.rs b/components/axpoll/tests/tests.rs index 4ec16e3b14..01dc2ba0e7 100644 --- a/components/axpoll/tests/tests.rs +++ b/components/axpoll/tests/tests.rs @@ -159,6 +159,27 @@ fn wake_only_matching_interests() { assert_eq!(write_counter.count(), 1); } +#[test] +fn wake_one_keeps_remaining_matching_waiters_registered() { + let ps = PollSet::new(); + let first_counter = Counter::new(); + let second_counter = Counter::new(); + let first_waker = Waker::from(first_counter.clone()); + let second_waker = Waker::from(second_counter.clone()); + + unsafe { + ps.register(&first_waker, IoEvents::IN); + ps.register(&second_waker, IoEvents::IN); + } + + assert_eq!(unsafe { ps.wake_one(IoEvents::IN) }, 1); + assert_eq!(first_counter.count() + second_counter.count(), 1); + assert_eq!(unsafe { ps.wake_one(IoEvents::IN) }, 1); + assert_eq!(first_counter.count(), 1); + assert_eq!(second_counter.count(), 1); + assert_eq!(unsafe { ps.wake_one(IoEvents::IN) }, 0); +} + #[test] fn concurrent_registers_preserve_interests() { const NUM_WAITERS: usize = 64; diff --git a/os/StarryOS/kernel/src/axtest_exports.rs b/os/StarryOS/kernel/src/axtest_exports.rs index 0c3ff5260b..1460d7e9e3 100644 --- a/os/StarryOS/kernel/src/axtest_exports.rs +++ b/os/StarryOS/kernel/src/axtest_exports.rs @@ -42,6 +42,14 @@ pub fn pipe_resize_rejects_oversized_pipe() -> bool { super::file::resize_rejects_oversized_pipe_for_test() } +pub fn pipe_linux_io_semantics_hold() -> bool { + super::file::pipe_linux_io_semantics_hold_for_test() +} + +pub fn interrupted_pipe_write_preserves_partial_progress() -> bool { + super::file::interrupted_pipe_write_preserves_partial_progress_for_test() +} + pub fn fcntl_setpipe_size_returns_capacity() -> bool { super::syscall::fcntl_setpipe_size_returns_capacity_for_test() } @@ -58,6 +66,22 @@ pub fn concurrent_epoll_reverse_add_is_serialized() -> bool { super::file::concurrent_reverse_add_is_serialized_for_test() } +pub fn epoll_level_aliases_rotate_in_linux_callback_order() -> bool { + super::file::level_aliases_rotate_in_linux_callback_order_for_test() +} + +pub fn epoll_edge_readiness_requires_a_new_notification() -> bool { + super::file::edge_readiness_requires_a_new_notification_for_test() +} + +pub fn epoll_edge_callback_does_not_reenter_target() -> bool { + super::file::edge_callback_does_not_reenter_target_for_test() +} + +pub fn epoll_hup_does_not_synthesize_readable() -> bool { + super::file::epoll_hup_does_not_synthesize_readable_for_test() +} + pub fn process_mem_stats_formats_linux_fields() -> bool { use super::mm::ProcessMemStats; diff --git a/os/StarryOS/kernel/src/file/epoll.rs b/os/StarryOS/kernel/src/file/epoll.rs index 3e5ad6230b..1349f371db 100644 --- a/os/StarryOS/kernel/src/file/epoll.rs +++ b/os/StarryOS/kernel/src/file/epoll.rs @@ -14,7 +14,7 @@ use alloc::{ }; use core::{ hash::{Hash, Hasher}, - sync::atomic::{AtomicBool, Ordering}, + sync::atomic::{AtomicBool, AtomicUsize, Ordering}, task::{Context, Waker}, }; @@ -113,17 +113,7 @@ enum ConsumeResult { } fn match_ready_events(current: IoEvents, interested: IoEvents) -> IoEvents { - let mut matched = (current & interested) | (current & IoEvents::ALWAYS_POLL); - // When the fd is hung up, also force IN so that epoll callers who only - // inspect EPOLLIN (a common pattern for pipes/sockets) can detect EOF. - // This is safe because a hung-up fd is always readable (read() returns 0 - // immediately). Linux epoll reports EPOLLHUP regardless of interest, but - // applications that mask on EPOLLIN alone still need to see the event. - // Calling `poll(2)` directly is unaffected by this epoll-only convention. - if matched.contains(IoEvents::HUP) { - matched |= IoEvents::IN; - } - matched + (current & interested) | (current & IoEvents::ALWAYS_POLL) } fn register_events(interested: IoEvents) -> IoEvents { @@ -180,9 +170,11 @@ struct EpollInterest { // A weak owner preserves same-process waiter refreshes without extending // the originating process lifetime. signalfd_registration_owner: Option>, + registration_order: usize, mode: SpinNoIrq, exclusive: bool, in_ready_queue: AtomicBool, + owner_repoll_pending: AtomicBool, } impl EpollInterest { @@ -191,6 +183,7 @@ impl EpollInterest { event: EpollEvent, flags: EpollFlags, nested_link: Option, + registration_order: usize, ) -> Self { Self { signalfd_registration_owner: key @@ -200,9 +193,11 @@ impl EpollInterest { key, event, nested_link, + registration_order, mode: SpinNoIrq::new(TriggerMode::from_flags(flags)), exclusive: flags.contains(EpollFlags::EXCLUSIVE), in_ready_queue: AtomicBool::new(false), + owner_repoll_pending: AtomicBool::new(false), } } @@ -216,6 +211,16 @@ impl EpollInterest { self.mode.lock().is_enabled() } + #[inline] + fn is_edge_triggered(&self) -> bool { + matches!(*self.mode.lock(), TriggerMode::Edge) + } + + #[inline] + fn is_level_triggered(&self) -> bool { + matches!(*self.mode.lock(), TriggerMode::Level) + } + #[inline] fn is_in_queue(&self) -> bool { self.in_ready_queue.load(Ordering::Acquire) @@ -276,11 +281,28 @@ impl EpollInterest { self.signalfd_registration_owner .as_ref() .is_none_or(|owner| { + let current_task = current(); + let Some(thread) = current_task.try_as_thread() else { + return false; + }; owner .upgrade() - .is_some_and(|owner| Arc::ptr_eq(&owner, ¤t().as_thread().proc_data)) + .is_some_and(|owner| Arc::ptr_eq(&owner, &thread.proc_data)) }) } + + fn request_owner_repoll(&self) { + self.owner_repoll_pending.store(true, Ordering::Release); + } + + fn requires_owner_repoll(&self) -> bool { + self.is_edge_triggered() && self.signalfd_registration_owner.is_some() + } + + fn take_owner_repoll_request(&self) -> bool { + self.can_refresh_waker_from_current_process() + && self.owner_repoll_pending.swap(false, Ordering::AcqRel) + } } struct InterestWaker { @@ -302,13 +324,33 @@ impl Wake for InterestWaker { return; }; - if interest.try_mark_in_queue() { - epoll.enqueue_marked_ready(&interest); - trace!( - "Epoll: fd={} added to ready queue, events={:?} wake up poller", - interest.key.fd, interest.event.events - ); + // signalfd readiness includes the calling thread's pending signals, so + // even a callback running in the same process cannot safely poll or + // re-register on behalf of the epoll waiter. A child after fork is an + // additional case where doing so would steal the parent's registration. + // Wake the original waiter and let it refresh exactly once in context. + if interest.requires_owner_repoll() { + interest.request_owner_repoll(); + epoll.wake_ready_waiters(1); + return; + } + + if interest.is_edge_triggered() { + // A target may invoke its waker while holding an internal lock. + // The callback must therefore only publish epoll-owned state; in + // particular, calling file.poll() or file.register() here could + // re-enter that target lock on the same thread. The epoll waiter + // rearms the consumed PollSet entry from task context. + if interest.is_enabled() && interest.try_mark_in_queue() { + epoll.enqueue_marked_ready(&interest); + trace!( + "Epoll: fd={} added to ready queue, events={:?}", + interest.key.fd, interest.event.events + ); + } + return; } + epoll.publish_ready_for_file(&interest); } } @@ -318,6 +360,7 @@ pub(super) struct EpollInner { ready_queue: SpinNoIrq>>, overflow_ready: AtomicBool, poll_ready: PollSet, + next_registration_order: AtomicUsize, } impl Default for EpollInner { @@ -328,6 +371,7 @@ impl Default for EpollInner { ready_queue: SpinNoIrq::new(VecDeque::new()), overflow_ready: AtomicBool::new(false), poll_ready: PollSet::new(), + next_registration_order: AtomicUsize::new(0), } } } @@ -342,6 +386,23 @@ impl EpollInner { unsafe { self.poll_ready.register(context.waker(), IoEvents::IN) }; } + fn register_waker_only(self: &Arc, interest: &Arc) { + let Some(file) = interest.key.get_file() else { + return; + }; + + if !interest.is_enabled() { + return; + } + + let waker = Waker::from(Arc::new(InterestWaker { + epoll: Arc::downgrade(self), + interest: Arc::downgrade(interest), + })); + let mut context = Context::from_waker(&waker); + file.register(&mut context, register_events(interest.event.events)); + } + /// Remove an interest while the global topology mutex is held. fn remove_interest_locked(&self, key: &EntryKey) -> Option> { let interest = self.interests.lock().remove(key)?; @@ -390,7 +451,7 @@ impl EpollInner { } } - fn enqueue_marked_ready(&self, interest: &Arc) { + fn enqueue_marked_ready_without_wake(&self, interest: &Arc) { let queued = { let mut queue = self.ready_queue.lock(); if queue.len() == queue.capacity() { @@ -408,8 +469,79 @@ impl EpollInner { interest.mark_not_in_queue(); self.overflow_ready.store(true, Ordering::Release); } - // Ready queue or overflow state is published before waking epoll waiters. - unsafe { self.poll_ready.wake(IoEvents::IN) }; + } + + fn wake_ready_waiters(&self, published: usize) { + for _ in 0..published { + // Each registered epoll waiter is exclusive. Stop once no waiter + // remains instead of needlessly walking an empty poll set. + if unsafe { self.poll_ready.wake_one(IoEvents::IN) } == 0 { + break; + } + } + } + + fn enqueue_marked_ready(&self, interest: &Arc) { + self.enqueue_marked_ready_without_wake(interest); + // Ready queue or overflow state is published before giving one + // exclusive epoll waiter a chance to consume it. Linux registers + // epoll_wait callers as exclusive waiters so one callback cannot make + // multiple callers race over the same level-triggered ready entry. + self.wake_ready_waiters(1); + } + + fn publish_ready_for_file(&self, source: &Arc) { + let interests = match self.snapshot_interests() { + Ok(interests) => interests, + Err(_) => { + // Allocation failure must not lose the callback that reached + // us. The overflow path will rediscover other ready aliases. + self.overflow_ready.store(true, Ordering::Release); + if source.is_enabled() && source.try_mark_in_queue() { + self.enqueue_marked_ready(source); + } else { + self.wake_ready_waiters(1); + } + return; + } + }; + + // One file readiness transition can invoke multiple registered + // callbacks for dup aliases. Publish all matching interests before + // waking epoll_wait callers so a re-entrant waiter cannot consume and + // requeue the first LT item ahead of an alias that is also ready. + let mut published = 0; + let mut interests = interests; + // Linux's non-exclusive poll callbacks are linked at the wait-queue + // head, so the most recently registered alias callback runs first. + // Preserve that ordering instead of exposing HashMap iteration order. + interests.sort_unstable_by_key(|interest| core::cmp::Reverse(interest.registration_order)); + for interest in interests { + let same_callback_batch = source.is_level_triggered() + && interest.is_level_triggered() + && Weak::ptr_eq(&interest.key.file, &source.key.file); + if (!same_callback_batch && !Arc::ptr_eq(&interest, source)) + || !interest.is_enabled() + || interest.is_in_queue() + { + continue; + } + let Some(file) = interest.key.get_file() else { + self.remove_invalid_interest(&interest); + continue; + }; + if !match_ready_events(file.poll(), interest.event.events).is_empty() + && interest.try_mark_in_queue() + { + self.enqueue_marked_ready_without_wake(&interest); + published += 1; + trace!( + "Epoll: fd={} added to ready queue, events={:?}", + interest.key.fd, interest.event.events + ); + } + } + self.wake_ready_waiters(published); } fn remove_ready_entries_for(&self, target: &Weak) { @@ -478,8 +610,8 @@ impl EpollInner { })(); if result.is_err() { self.overflow_ready.store(true, Ordering::Release); - // Overflow state is published before waking epoll waiters. - unsafe { self.poll_ready.wake(IoEvents::IN) }; + // Overflow state is published before waking one exclusive waiter. + unsafe { self.poll_ready.wake_one(IoEvents::IN) }; } result } @@ -501,28 +633,22 @@ impl Epoll { return; } - let Some(file) = interest.key.get_file() else { - return; - }; - - if !interest.is_enabled() { - return; - } - - let waker = Waker::from(Arc::new(InterestWaker { - epoll: Arc::downgrade(&self.inner), - interest: Arc::downgrade(interest), - })); - - let mut context = Context::from_waker(&waker); - file.register(&mut context, register_events(interest.event.events)); + self.inner.register_waker_only(interest); } /// Registers enabled interests with the thread currently waiting in epoll. pub fn register_waiter_wakers(&self) -> AxResult { let interests = self.inner.snapshot_interests()?; for interest in &interests { - self.register_waker_only(interest); + if interest.take_owner_repoll_request() { + // A callback consumed outside owner context cannot safely poll + // signalfd readiness there. Recheck exactly once in the owner + // waiter without turning ordinary EPOLLET waits into LT polls. + self.inner.register_waker_only(interest); + self.inner.publish_ready_for_file(interest); + } else { + self.register_waker_only(interest); + } } Ok(()) } @@ -601,6 +727,9 @@ impl Epoll { event, flags, nested_link.clone(), + self.inner + .next_registration_order + .fetch_add(1, Ordering::Relaxed), )); self.inner .interests @@ -632,6 +761,24 @@ impl Epoll { ) } + #[cfg(axtest)] + pub(super) fn add_file_for_test( + &self, + fd: i32, + target: Arc, + user_data: u64, + flags: EpollFlags, + ) -> AxResult<()> { + self.add_interest( + EntryKey::for_test(fd, &target), + EpollEvent { + events: IoEvents::IN, + user_data, + }, + flags, + ) + } + pub fn modify(&self, fd: i32, event: EpollEvent, flags: EpollFlags) -> AxResult<()> { let key = EntryKey::new(fd)?; @@ -647,6 +794,7 @@ impl Epoll { event, flags, old.nested_link.clone(), + old.registration_order, )); // Preserve ready-queue membership across the swap. The ready_queue @@ -709,13 +857,12 @@ impl Epoll { // into the loop and filling out[] with duplicates of one ready fd. let mut txlist = self.inner.drain_ready_queue()?; let mut count = 0; - let mut keep: VecDeque> = VecDeque::new(); + let mut level_ready: VecDeque> = VecDeque::new(); - while let Some(weak_interest) = txlist.pop_front() { - if count >= max_events { - keep.push_back(weak_interest); - continue; - } + while count < max_events { + let Some(weak_interest) = txlist.pop_front() else { + break; + }; let Some(interest) = weak_interest.upgrade() else { continue; // interest already removed @@ -747,32 +894,24 @@ impl Epoll { if let Err(err) = put_event(count, event) { interest.restore_mode(old_mode); interest.in_ready_queue.store(true, Ordering::Release); - self.inner.enqueue_marked_ready(&interest); - for entry in txlist.into_iter().chain(keep) { + self.inner.enqueue_marked_ready_without_wake(&interest); + let mut published = 1; + for entry in txlist.into_iter().chain(level_ready) { if let Some(interest) = entry.upgrade() && interest.is_in_queue() { - self.inner.enqueue_marked_ready(&interest); + self.inner.enqueue_marked_ready_without_wake(&interest); + published += 1; } } + self.inner.wake_ready_waiters(published); return if count == 0 { Err(err) } else { Ok(count) }; } count += 1; if keep_ready { - keep.push_back(Arc::downgrade(&interest)); + level_ready.push_back(Arc::downgrade(&interest)); } else { - // EPOLLET edge-triggered: after reporting the fd once, - // it must NOT be reported again until a *new* edge - // (a fresh wakeup) arrives — even if the fd is still - // readable because the caller left data unconsumed - // (man 7 epoll; Linux ep_send_events does not re-add an - // edge-triggered epi to the ready list). Re-arm a fresh - // waker so the next edge transition re-queues the - // interest via InterestWaker::wake_by_ref; do NOT - // re-enqueue based on the current (possibly residual) - // readability, which would degrade EPOLLET into - // level-triggered behavior. interest.mark_not_in_queue(); self.register_waker_only(&interest); } @@ -794,15 +933,19 @@ impl Epoll { } } - if !keep.is_empty() { - for entry in keep { - if let Some(interest) = entry.upgrade() - && interest.is_in_queue() - { - self.inner.enqueue_marked_ready(&interest); - } + // Linux puts entries not visited because of maxevents before LT + // entries returned by this scan. That rotation lets successive + // epoll_wait callers make progress across the ready list. + let mut published = 0; + for entry in txlist.into_iter().chain(level_ready) { + if let Some(interest) = entry.upgrade() + && interest.is_in_queue() + { + self.inner.enqueue_marked_ready_without_wake(&interest); + published += 1; } } + self.inner.wake_ready_waiters(published); if count == 0 { Err(AxError::WouldBlock) @@ -824,10 +967,10 @@ pub(crate) fn epoll_event_matching_rules_hold_for_test() -> bool { // the caller's interest mask. && match_ready_events(E::HUP, E::OUT).contains(E::HUP) && match_ready_events(E::ERR, E::empty()).contains(E::ERR) - // HUP forces IN even when the caller is not interested in IN, so that - // pipes report EOF on EPOLLHUP-only subscriptions. - && (match_ready_events(E::HUP, E::OUT).contains(E::IN)) - // HUP combining with interested IN yields both IN and HUP. + // HUP alone does not synthesize IN. Linux still forwards HUP even if + // the caller only subscribed to another readiness class. + && !match_ready_events(E::HUP, E::OUT).contains(E::IN) + // A source that explicitly reports both HUP and IN preserves both. && { let m = match_ready_events(E::HUP | E::IN, E::IN); m.contains(E::IN) && m.contains(E::HUP) @@ -859,3 +1002,10 @@ pub(crate) fn epoll_event_matching_rules_hold_for_test() -> bool { && TriggerMode::OneShot { fired: false }.is_enabled() && !TriggerMode::OneShot { fired: true }.is_enabled() } + +#[cfg(axtest)] +pub(crate) fn epoll_hup_does_not_synthesize_readable_for_test() -> bool { + let matched = match_ready_events(IoEvents::HUP, IoEvents::IN); + + matched.bits() == IoEvents::HUP.bits() +} diff --git a/os/StarryOS/kernel/src/file/epoll_axtest.rs b/os/StarryOS/kernel/src/file/epoll_axtest.rs index 18b1861b25..a47a6ae93f 100644 --- a/os/StarryOS/kernel/src/file/epoll_axtest.rs +++ b/os/StarryOS/kernel/src/file/epoll_axtest.rs @@ -1,12 +1,19 @@ //! Deterministic concurrency hooks for epoll kernel tests. -use alloc::sync::Arc; -use core::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; +use alloc::{borrow::Cow, sync::Arc, task::Wake}; +use core::{ + sync::atomic::{AtomicBool, AtomicUsize, Ordering}, + task::{Context, Waker}, +}; use ax_errno::AxError; use ax_kspin::SpinNoIrq; +use axpoll::{IoEvents, PollSet, Pollable}; -use super::epoll::Epoll; +use super::{ + FileLike, + epoll::{Epoll, EpollFlags}, +}; static EPOLL_ADD_TEST_BARRIER_ENABLED: AtomicBool = AtomicBool::new(false); static EPOLL_ADD_TEST_BARRIER_ARRIVALS: AtomicUsize = AtomicUsize::new(0); @@ -57,3 +64,199 @@ pub(crate) fn concurrent_reverse_add_is_serialized_for_test() -> bool { [None, Some(AxError::FilesystemLoop)] | [Some(AxError::FilesystemLoop), None] ) } + +struct ReadyFile { + ready: AtomicBool, + poll_waiters: PollSet, +} + +impl ReadyFile { + fn new() -> Arc { + Arc::new(Self { + ready: AtomicBool::new(false), + poll_waiters: PollSet::new(), + }) + } + + fn make_ready(&self) { + self.ready.store(true, Ordering::Release); + unsafe { self.poll_waiters.wake(IoEvents::IN) }; + } +} + +impl FileLike for ReadyFile { + fn path(&self) -> Cow<'_, str> { + "axtest:[epoll-ready-file]".into() + } +} + +impl Pollable for ReadyFile { + fn poll(&self) -> IoEvents { + if self.ready.load(Ordering::Acquire) { + IoEvents::IN + } else { + IoEvents::empty() + } + } + + fn register(&self, context: &mut Context<'_>, events: IoEvents) { + unsafe { self.poll_waiters.register(context.waker(), events) }; + } +} + +struct CallbackBoundaryFile { + ready: AtomicBool, + waking: AtomicBool, + callback_reentered_file: AtomicBool, + poll_waiters: PollSet, +} + +impl CallbackBoundaryFile { + fn new() -> Arc { + Arc::new(Self { + ready: AtomicBool::new(false), + waking: AtomicBool::new(false), + callback_reentered_file: AtomicBool::new(false), + poll_waiters: PollSet::new(), + }) + } + + fn make_ready(&self) { + self.ready.store(true, Ordering::Release); + self.waking.store(true, Ordering::Release); + unsafe { self.poll_waiters.wake(IoEvents::IN) }; + self.waking.store(false, Ordering::Release); + } + + fn callback_reentered_file(&self) -> bool { + self.callback_reentered_file.load(Ordering::Acquire) + } + + fn record_callback_reentry(&self) { + if self.waking.load(Ordering::Acquire) { + self.callback_reentered_file.store(true, Ordering::Release); + } + } +} + +impl FileLike for CallbackBoundaryFile { + fn path(&self) -> Cow<'_, str> { + "axtest:[epoll-callback-boundary-file]".into() + } +} + +impl Pollable for CallbackBoundaryFile { + fn poll(&self) -> IoEvents { + self.record_callback_reentry(); + if self.ready.load(Ordering::Acquire) { + IoEvents::IN + } else { + IoEvents::empty() + } + } + + fn register(&self, context: &mut Context<'_>, events: IoEvents) { + self.record_callback_reentry(); + unsafe { self.poll_waiters.register(context.waker(), events) }; + } +} + +struct EpollWaiter { + epoll: Arc, + result_index: usize, + results: Arc; 2]>>, +} + +impl EpollWaiter { + fn collect_one(&self) { + let mut user_data = None; + let result = self.epoll.poll_events_with(1, |_index, event| { + user_data = Some(event.data); + Ok(()) + }); + if matches!(result, Ok(1)) { + self.results.lock()[self.result_index] = user_data; + } + } +} + +impl Wake for EpollWaiter { + fn wake(self: Arc) { + self.collect_one(); + } + + fn wake_by_ref(self: &Arc) { + self.collect_one(); + } +} + +pub(crate) fn level_aliases_rotate_in_linux_callback_order_for_test() -> bool { + let epoll = Arc::new(Epoll::new()); + let target = ReadyFile::new(); + let target_file: Arc = target.clone(); + let results = Arc::new(SpinNoIrq::new([None, None])); + + epoll + .add_file_for_test(1, target_file.clone(), 0x11, EpollFlags::empty()) + .expect("first test interest must be added"); + epoll + .add_file_for_test(2, target_file, 0x22, EpollFlags::empty()) + .expect("second test interest must be added"); + + for result_index in 0..2 { + let waiter = Arc::new(EpollWaiter { + epoll: epoll.clone(), + result_index, + results: results.clone(), + }); + let waker = Waker::from(waiter); + let mut context = Context::from_waker(&waker); + epoll.register(&mut context, IoEvents::IN); + } + + target.make_ready(); + results.lock().as_slice() == [Some(0x22), Some(0x11)] +} + +pub(crate) fn edge_readiness_requires_a_new_notification_for_test() -> bool { + let epoll = Epoll::new(); + let target = ReadyFile::new(); + let target_file: Arc = target.clone(); + + epoll + .add_file_for_test(1, target_file, 0x33, EpollFlags::EDGE_TRIGGER) + .expect("edge-triggered test interest must be added"); + + target.make_ready(); + let first = collect_one_event(&epoll); + let without_new_notification = collect_one_event(&epoll); + target.make_ready(); + let after_new_notification = collect_one_event(&epoll); + + first == Ok((1, Some(0x33))) + && without_new_notification == Err(AxError::WouldBlock) + && after_new_notification == Ok((1, Some(0x33))) +} + +pub(crate) fn edge_callback_does_not_reenter_target_for_test() -> bool { + let epoll = Epoll::new(); + let target = CallbackBoundaryFile::new(); + let target_file: Arc = target.clone(); + + epoll + .add_file_for_test(1, target_file, 0x44, EpollFlags::EDGE_TRIGGER) + .expect("edge-triggered test interest must be added"); + + target.make_ready(); + + !target.callback_reentered_file() +} + +fn collect_one_event(epoll: &Epoll) -> Result<(usize, Option), AxError> { + let mut user_data = None; + let count = epoll.poll_events_with(1, |_index, event| { + user_data = Some(event.data); + Ok(()) + })?; + Ok((count, user_data)) +} diff --git a/os/StarryOS/kernel/src/file/event.rs b/os/StarryOS/kernel/src/file/event.rs index bd8fa9542b..3a796b24da 100644 --- a/os/StarryOS/kernel/src/file/event.rs +++ b/os/StarryOS/kernel/src/file/event.rs @@ -33,6 +33,13 @@ impl EventFd { } impl FileLike for EventFd { + fn validate_write_len(&self, len: usize) -> ax_io::Result { + if len != size_of::() { + return Err(AxError::InvalidInput); + } + Ok(()) + } + fn read(&self, dst: &mut IoDst) -> ax_io::Result { if dst.remaining_mut() < size_of::() { return Err(AxError::InvalidInput); @@ -63,7 +70,7 @@ impl FileLike for EventFd { } fn write(&self, src: &mut IoSrc) -> ax_io::Result { - if src.remaining() != size_of::() { + if src.remaining() < size_of::() { return Err(AxError::InvalidInput); } diff --git a/os/StarryOS/kernel/src/file/mod.rs b/os/StarryOS/kernel/src/file/mod.rs index ca298330f1..67638bce9b 100644 --- a/os/StarryOS/kernel/src/file/mod.rs +++ b/os/StarryOS/kernel/src/file/mod.rs @@ -46,7 +46,13 @@ use starry_process::Pid; #[cfg(axtest)] pub(crate) use self::epoll::epoll_event_matching_rules_hold_for_test; #[cfg(axtest)] -pub(crate) use self::epoll_axtest::concurrent_reverse_add_is_serialized_for_test; +pub(crate) use self::epoll::epoll_hup_does_not_synthesize_readable_for_test; +#[cfg(axtest)] +pub(crate) use self::epoll_axtest::{ + concurrent_reverse_add_is_serialized_for_test, edge_callback_does_not_reenter_target_for_test, + edge_readiness_requires_a_new_notification_for_test, + level_aliases_rotate_in_linux_callback_order_for_test, +}; #[cfg(axtest)] pub(crate) use self::epoll_topology::epoll_arc_operations_hold_for_test; #[cfg(axtest)] @@ -70,7 +76,8 @@ pub(crate) use self::fs::metadata_to_kstat_conversion_rules_hold_for_test; pub(crate) use self::mount_table::{MountTableFile, notify_mount_namespace_changed}; #[cfg(axtest)] pub(crate) use self::pipe::{ - peer_close_with_multiple_readers_is_visible_for_test, + interrupted_pipe_write_preserves_partial_progress_for_test, + peer_close_with_multiple_readers_is_visible_for_test, pipe_linux_io_semantics_hold_for_test, pipe_resize_rounding_and_state_rules_hold_for_test, resize_rejects_oversized_pipe_for_test, }; #[cfg(axtest)] @@ -199,6 +206,15 @@ pub type IoSrc<'a> = dyn ReadBuf + 'a; #[allow(dead_code)] pub trait FileLike: Pollable + DowncastSync { + /// Validate a scalar write length before importing the user buffer. + /// + /// File types with count errors that take precedence over `EFAULT` can + /// override this hook. The full write operation must repeat any invariant + /// needed to remain correct for non-scalar callers. + fn validate_write_len(&self, _len: usize) -> AxResult { + Ok(()) + } + fn read(&self, _dst: &mut IoDst) -> AxResult { Err(AxError::InvalidInput) } diff --git a/os/StarryOS/kernel/src/file/pipe.rs b/os/StarryOS/kernel/src/file/pipe.rs index 4214ac3f6e..792ef08d09 100644 --- a/os/StarryOS/kernel/src/file/pipe.rs +++ b/os/StarryOS/kernel/src/file/pipe.rs @@ -1,4 +1,4 @@ -use alloc::{borrow::Cow, format, sync::Arc}; +use alloc::{borrow::Cow, collections::VecDeque, format, sync::Arc}; use core::{ mem, sync::atomic::{AtomicBool, Ordering}, @@ -32,6 +32,7 @@ use crate::{ const RING_BUFFER_INIT_SIZE: usize = 65536; // 64 KiB const RING_BUFFER_MAX_SIZE: usize = 1024 * 1024; // 1 MiB +const PIPE_BUF: usize = PAGE_SIZE_4K; struct Shared { state: Mutex, @@ -41,10 +42,77 @@ struct Shared { struct PipeState { buffer: HeapRb, + buffers: VecDeque, readers: usize, writers: usize, } +impl PipeState { + fn has_free_buffer(&self) -> bool { + self.buffers.len() < self.buffer.capacity().get() / PIPE_BUF + } + + fn can_merge(&self, bytes: usize) -> bool { + self.buffers + .back() + .is_some_and(|length| length + bytes <= PIPE_BUF) + } + + fn copy_from(&mut self, src: &mut IoSrc, limit: usize) -> AxResult { + let (left, right) = self.buffer.vacant_slices_mut(); + let left_limit = left.len().min(limit); + // `left` covers vacant ring storage and the following `read` initializes + // exactly the returned prefix before the write index is advanced. + let left = unsafe { left.assume_init_mut() }; + let mut copied = src.read(&mut left[..left_limit])?; + if copied == left_limit && copied < limit { + let right_limit = right.len().min(limit - copied); + // The same vacant-storage contract applies to the wrapped slice. + let right = unsafe { right.assume_init_mut() }; + copied += src.read(&mut right[..right_limit])?; + } + // Both reads initialized the first `copied` bytes across the two vacant + // slices, and neither slice aliases occupied ring contents. + unsafe { self.buffer.advance_write_index(copied) }; + Ok(copied) + } + + fn merge_from(&mut self, src: &mut IoSrc, bytes: usize) -> AxResult { + debug_assert!(self.can_merge(bytes)); + let copied = self.copy_from(src, bytes)?; + *self + .buffers + .back_mut() + .expect("merge requires an existing pipe buffer") += copied; + Ok(copied) + } + + fn append_from(&mut self, src: &mut IoSrc) -> AxResult { + debug_assert!(self.has_free_buffer()); + let limit = src.remaining().min(PIPE_BUF); + let copied = self.copy_from(src, limit)?; + if copied > 0 { + self.buffers.push_back(copied); + } + Ok(copied) + } + + fn consume(&mut self, mut bytes: usize) { + while bytes > 0 { + let front = self + .buffers + .front_mut() + .expect("pipe bytes require a pipe buffer"); + let consumed = bytes.min(*front); + *front -= consumed; + bytes -= consumed; + if *front == 0 { + self.buffers.pop_front(); + } + } + } +} + pub struct Pipe { read_side: bool, shared: Arc, @@ -84,6 +152,7 @@ impl Pipe { let shared = Arc::new(Shared { state: Mutex::new(PipeState { buffer: HeapRb::new(RING_BUFFER_INIT_SIZE), + buffers: VecDeque::new(), readers: 1, writers: 1, }), @@ -124,7 +193,7 @@ impl Pipe { if new_size == old_size { return Ok(()); } - if new_size < state.buffer.occupied_len() { + if new_size / PIPE_BUF < state.buffers.len() { return Err(AxError::ResourceBusy); } let old_buffer = mem::replace( @@ -144,14 +213,118 @@ impl Pipe { Ok(()) } + fn write_with_broken_pipe_handler( + &self, + src: &mut IoSrc, + on_broken_pipe: impl Fn(), + ) -> AxResult { + if !self.is_write() { + return Err(AxError::BadFileDescriptor); + } + let size = src.remaining(); + if size == 0 { + return Ok(0); + } + + let mut total_written = 0; + let mut merge_pending = true; + let merge_bytes = size % PIPE_BUF; + + let result = block_on(poll_io(self, IoEvents::OUT, self.nonblocking(), || { + enum WriteStep { + Closed, + WouldBlock, + Wrote(usize), + } + + let step = { + let mut state = self.shared.state.lock(); + // Linux makes writes no larger than PIPE_BUF commit atomically; + // nonblocking callers get EAGAIN until the whole record fits. + if state.readers == 0 { + WriteStep::Closed + } else { + let mut written = 0; + if merge_pending { + merge_pending = false; + if merge_bytes > 0 && state.can_merge(merge_bytes) { + written += state.merge_from(src, merge_bytes)?; + } + } + while src.remaining() > 0 && state.has_free_buffer() { + let appended = state.append_from(src)?; + written += appended; + if appended == 0 { + break; + } + } + if written == 0 { + WriteStep::WouldBlock + } else { + WriteStep::Wrote(written) + } + } + }; + + let written = match step { + WriteStep::Closed => { + if total_written > 0 { + return Ok(total_written); + } + on_broken_pipe(); + return Err(AxError::BrokenPipe); + } + WriteStep::WouldBlock => return Err(AxError::WouldBlock), + WriteStep::Wrote(written) => written, + }; + + if written > 0 { + // Pipe bytes were committed before waking readers. + unsafe { self.shared.poll_rx.wake(IoEvents::IN) }; + total_written += written; + if total_written == size || self.nonblocking() { + return Ok(total_written); + } + } + Err(AxError::WouldBlock) + })); + + // Linux returns committed bytes instead of EINTR once a pipe write + // has made progress. This also prevents SA_RESTART from replaying the + // whole userspace buffer after the prefix is already visible. + match result { + Err(AxError::Interrupted) if total_written > 0 => Ok(total_written), + result => result, + } + } + #[cfg(axtest)] - fn duplicate_read_end_for_test(&self) -> Pipe { + fn write_without_sigpipe_for_test(&self, src: &mut IoSrc) -> AxResult { + // Axtests run in a kernel task without Starry process signal state. The + // write transition is identical, but SIGPIPE delivery is outside this + // direct pipe test and cannot be requested from that task. + self.write_with_broken_pipe_handler(src, || {}) + } + + #[cfg(axtest)] + pub(crate) fn duplicate_read_end_for_test(&self) -> Pipe { assert!(self.is_read()); self.shared.state.lock().readers += 1; Pipe { read_side: true, shared: self.shared.clone(), - non_blocking: AtomicBool::new(false), + non_blocking: AtomicBool::new(self.nonblocking()), + } + } + + #[cfg(axtest)] + pub(crate) fn duplicate_write_end_for_test(&self) -> Pipe { + assert!(self.is_write()); + self.shared.state.lock().writers += 1; + Pipe { + read_side: false, + shared: self.shared.clone(), + non_blocking: AtomicBool::new(self.nonblocking()), } } } @@ -186,6 +359,153 @@ pub(crate) fn resize_rejects_oversized_pipe_for_test() -> bool { read_end.resize(1024 * 1024 + 1).is_err() } +#[cfg(axtest)] +pub(crate) fn pipe_linux_io_semantics_hold_for_test() -> bool { + let null_io_matches = { + let (read_end, write_end) = Pipe::new(); + read_end.set_nonblocking(true).ok(); + write_end.set_nonblocking(true).ok(); + + let mut empty_dst: &mut [u8] = &mut []; + let null_read = read_end.read(&mut empty_dst as &mut dyn super::WriteBuf); + drop(read_end); + let mut empty_src: &[u8] = &[]; + let null_write = + write_end.write_without_sigpipe_for_test(&mut empty_src as &mut dyn super::ReadBuf); + + null_read == Ok(0) && null_write == Ok(0) + }; + + let atomic_write_and_poll_match = { + let (read_end, write_end) = Pipe::new(); + write_end.set_nonblocking(true).ok(); + let resized = write_end.resize(PIPE_BUF).is_ok(); + let initial = [b'a'; 4000]; + let mut initial_src: &[u8] = &initial; + let initial_write = + write_end.write_without_sigpipe_for_test(&mut initial_src as &mut dyn super::ReadBuf); + let atomic = [b'b'; 200]; + let mut atomic_src: &[u8] = &atomic; + let atomic_write = + write_end.write_without_sigpipe_for_test(&mut atomic_src as &mut dyn super::ReadBuf); + let queued = read_end.shared.state.lock().buffer.occupied_len(); + + resized + && initial_write == Ok(initial.len()) + && atomic_write == Err(AxError::WouldBlock) + && queued == initial.len() + && !write_end.poll().contains(IoEvents::OUT) + }; + + let closed_reader_poll_matches = { + let (read_end, write_end) = Pipe::new(); + drop(read_end); + let events = write_end.poll(); + events.contains(IoEvents::OUT | IoEvents::ERR) + }; + + let duplicates_preserve_nonblocking = { + let (read_end, write_end) = Pipe::new(); + read_end.set_nonblocking(true).ok(); + write_end.set_nonblocking(true).ok(); + read_end.duplicate_read_end_for_test().nonblocking() + && write_end.duplicate_write_end_for_test().nonblocking() + }; + + let page_slot_fragmentation_matches = { + let (read_end, write_end) = Pipe::new(); + write_end.set_nonblocking(true).ok(); + let resized = write_end.resize(2 * PIPE_BUF).is_ok(); + let initial = [b'a'; 5000]; + let mut initial_src: &[u8] = &initial; + let initial_write = + write_end.write_without_sigpipe_for_test(&mut initial_src as &mut dyn super::ReadBuf); + let mut consumed = [0u8; 1000]; + let mut consumed_dst: &mut [u8] = &mut consumed; + let initial_read = read_end.read(&mut consumed_dst as &mut dyn super::WriteBuf); + let shrink = write_end.resize(PIPE_BUF); + let atomic = [b'b'; 4000]; + let mut atomic_src: &[u8] = &atomic; + let atomic_write = + write_end.write_without_sigpipe_for_test(&mut atomic_src as &mut dyn super::ReadBuf); + + resized + && initial_write == Ok(initial.len()) + && initial_read == Ok(consumed.len()) + && !write_end.poll().contains(IoEvents::OUT) + && shrink == Err(AxError::ResourceBusy) + && atomic_write == Err(AxError::WouldBlock) + }; + + null_io_matches + && atomic_write_and_poll_match + && closed_reader_poll_matches + && duplicates_preserve_nonblocking + && page_slot_fragmentation_matches +} + +#[cfg(axtest)] +pub(crate) fn interrupted_pipe_write_preserves_partial_progress_for_test() -> bool { + use ax_task::TaskState; + + let (read_end, write_end) = Pipe::new(); + if write_end.resize(PIPE_BUF).is_err() { + return false; + } + + let initial = [b'a'; PIPE_BUF]; + let mut initial_src: &[u8] = &initial; + if write_end.write_without_sigpipe_for_test(&mut initial_src) != Ok(PIPE_BUF) { + return false; + } + + let write_end = Arc::new(write_end); + let result = Arc::new(Mutex::new(None)); + let writer_task = { + let write_end = Arc::clone(&write_end); + let result = Arc::clone(&result); + ax_task::spawn(move || { + let bytes = [b'b'; 2 * PIPE_BUF]; + let mut src: &[u8] = &bytes; + *result.lock() = Some(write_end.write_without_sigpipe_for_test(&mut src)); + }) + }; + + if !wait_for_pipe_test_condition(|| writer_task.state() == TaskState::Blocked) { + writer_task.interrupt(); + writer_task.join(); + return false; + } + + let mut consumed = [0u8; PIPE_BUF]; + let mut dst: &mut [u8] = &mut consumed; + if read_end.read(&mut dst) != Ok(PIPE_BUF) { + writer_task.interrupt(); + writer_task.join(); + return false; + } + + let refilled_and_blocked = wait_for_pipe_test_condition(|| { + read_end.shared.state.lock().buffer.occupied_len() == PIPE_BUF + && writer_task.state() == TaskState::Blocked + }); + writer_task.interrupt(); + writer_task.join(); + + refilled_and_blocked && *result.lock() == Some(Ok(PIPE_BUF)) +} + +#[cfg(axtest)] +fn wait_for_pipe_test_condition(mut condition: impl FnMut() -> bool) -> bool { + for _ in 0..10_000 { + if condition() { + return true; + } + ax_task::yield_now(); + } + false +} + fn raise_pipe() { let curr = current(); send_signal_to_process( @@ -206,13 +526,14 @@ impl FileLike for Pipe { block_on(poll_io(self, IoEvents::IN, self.nonblocking(), || { let (read, writers) = { - let state = self.shared.state.lock(); + let mut state = self.shared.state.lock(); let (left, right) = state.buffer.as_slices(); let mut count = dst.write(left)?; if count >= left.len() { count += dst.write(right)?; } unsafe { state.buffer.advance_read_index(count) }; + state.consume(count); (count, state.writers) }; if read > 0 { @@ -228,55 +549,7 @@ impl FileLike for Pipe { } fn write(&self, src: &mut IoSrc) -> AxResult { - if !self.is_write() { - return Err(AxError::BadFileDescriptor); - } - let size = src.remaining(); - if size == 0 { - return Ok(0); - } - - let mut total_written = 0; - - block_on(poll_io(self, IoEvents::OUT, self.nonblocking(), || { - enum WriteStep { - Closed, - Wrote(usize), - } - - let step = { - let mut state = self.shared.state.lock(); - if state.readers == 0 { - WriteStep::Closed - } else { - let (left, right) = state.buffer.vacant_slices_mut(); - let mut count = src.read(unsafe { left.assume_init_mut() })?; - if count >= left.len() { - count += src.read(unsafe { right.assume_init_mut() })?; - } - unsafe { state.buffer.advance_write_index(count) }; - WriteStep::Wrote(count) - } - }; - - let WriteStep::Wrote(written) = step else { - if total_written > 0 { - return Ok(total_written); - } - raise_pipe(); - return Err(AxError::BrokenPipe); - }; - - if written > 0 { - // Pipe bytes were committed before waking readers. - unsafe { self.shared.poll_rx.wake(IoEvents::IN) }; - total_written += written; - if total_written == size || self.nonblocking() { - return Ok(total_written); - } - } - Err(AxError::WouldBlock) - })) + self.write_with_broken_pipe_handler(src, raise_pipe) } fn stat(&self) -> AxResult { @@ -319,28 +592,34 @@ impl Pollable for Pipe { let mut events = IoEvents::empty(); let state = self.shared.state.lock(); if self.read_side { - events.set(IoEvents::IN, state.buffer.occupied_len() > 0); + events.set( + IoEvents::IN | IoEvents::RDNORM, + state.buffer.occupied_len() > 0, + ); events.set(IoEvents::HUP, state.writers == 0); } else { events.set(IoEvents::ERR, state.readers == 0); - events.set( - IoEvents::OUT, - state.readers > 0 && state.buffer.vacant_len() > 0, - ); + // Linux reports POLLOUT when the pipe has a free PIPE_BUF-sized + // slot, independently of whether the reader has already closed. + events.set(IoEvents::OUT | IoEvents::WRNORM, state.has_free_buffer()); } events } fn register(&self, context: &mut Context<'_>, events: IoEvents) { + let read_ready = events.intersects(IoEvents::IN | IoEvents::RDNORM); + let write_ready = events.intersects(IoEvents::OUT | IoEvents::WRNORM); let mut interests = if self.read_side { - events & (IoEvents::IN | IoEvents::HUP) + events & IoEvents::HUP } else { - events & (IoEvents::OUT | IoEvents::ERR) + events & IoEvents::ERR }; - if self.read_side && events.contains(IoEvents::IN) { + if self.read_side && read_ready { + interests.insert(IoEvents::IN); interests.insert(IoEvents::HUP); } - if !self.read_side && events.contains(IoEvents::OUT) { + if !self.read_side && write_ready { + interests.insert(IoEvents::OUT); interests.insert(IoEvents::ERR); } if interests.is_empty() { diff --git a/os/StarryOS/kernel/src/syscall/fs/io.rs b/os/StarryOS/kernel/src/syscall/fs/io.rs index 1b4a045f8e..b44d910eca 100644 --- a/os/StarryOS/kernel/src/syscall/fs/io.rs +++ b/os/StarryOS/kernel/src/syscall/fs/io.rs @@ -138,6 +138,7 @@ pub fn sys_readv(fd: i32, iov: *const IoVec, iovcnt: usize) -> AxResult { pub fn sys_write(fd: i32, buf: *mut u8, len: usize) -> AxResult { debug!("sys_write <= fd: {fd}, buf: {buf:p}, len: {len}"); let file_like = get_file_like(fd)?; + file_like.validate_write_len(len)?; validate_user_read_buf(buf.cast_const(), len)?; memfd_checks_before_stream_write(&file_like, len as u64)?; let data = copy_user_read_buf(buf.cast_const(), len)?; @@ -146,8 +147,12 @@ pub fn sys_write(fd: i32, buf: *mut u8, len: usize) -> AxResult { pub fn sys_writev(fd: i32, iov: *const IoVec, iovcnt: usize) -> AxResult { debug!("sys_writev <= fd: {fd}, iovcnt: {iovcnt}"); - let total = validate_user_iov_buf_regions(iov, iovcnt)?; let file_like = get_file_like(fd)?; + // Check length invariants (e.g. eventfd count) before importing segment + // data, so a count error (EINVAL) takes precedence over a bad segment + // pointer (EFAULT), matching Linux vfs_writev / eventfd_write ordering. + file_like.validate_write_len(iov_total_len(iov, iovcnt)?)?; + let total = validate_user_iov_buf_regions(iov, iovcnt)?; memfd_checks_before_stream_write(&file_like, total as u64)?; let data = copy_user_iov_read_buf(iov, iovcnt)?; file_like.write(&mut data.as_slice()).map(|n| n as _) @@ -551,8 +556,9 @@ pub fn sys_pwritev2( } if offset == -1 { // offset == -1: use current file position (like writev) - let total = validate_user_iov_buf_regions(iov, iovcnt)?; let file_like = get_file_like(fd)?; + file_like.validate_write_len(iov_total_len(iov, iovcnt)?)?; + let total = validate_user_iov_buf_regions(iov, iovcnt)?; memfd_checks_before_stream_write(&file_like, total as u64)?; let data = copy_user_iov_read_buf(iov, iovcnt)?; file_like.write(&mut data.as_slice()).map(|n| n as _) @@ -591,6 +597,27 @@ fn validate_user_read_buf(buf: *const u8, len: usize) -> AxResult<()> { Ok(()) } +/// Sum of `iov_len` across the iovec array. Reads the iovec *struct* (so a bad +/// array pointer still yields `EFAULT`) but does not touch `iov_base`, letting +/// callers enforce length invariants (e.g. eventfd's 8-byte count) before any +/// segment payload is imported. Same overflow cap as [`IoVectorBuf`]. +fn iov_total_len(iov: *const IoVec, iovcnt: usize) -> AxResult { + if iovcnt > 1024 { + return Err(AxError::InvalidInput); + } + let mut total = 0usize; + for i in 0..iovcnt { + let entry = iov.wrapping_add(i).vm_read()?; + if entry.iov_len < 0 { + return Err(AxError::InvalidInput); + } + total = total + .checked_add(entry.iov_len as usize) + .ok_or(AxError::InvalidInput)?; + } + Ok(total) +} + /// Validate each `iovec` segment is readable; returns total length (same cap as [`IoVectorBuf`]). fn validate_user_iov_buf_regions(iov: *const IoVec, iovcnt: usize) -> AxResult { if iovcnt > 1024 { diff --git a/os/StarryOS/kernel/src/syscall/io_mpx/poll.rs b/os/StarryOS/kernel/src/syscall/io_mpx/poll.rs index 7187f2141d..9582a0ac3c 100644 --- a/os/StarryOS/kernel/src/syscall/io_mpx/poll.rs +++ b/os/StarryOS/kernel/src/syscall/io_mpx/poll.rs @@ -70,12 +70,6 @@ fn collect_ready_poll_events( let mut res = 0usize; for ((fd, events), revent_index) in fds.0.iter().zip(revent_indices.iter()) { let mut result = fd.poll(); - if result.contains(IoEvents::IN) { - result |= IoEvents::RDNORM; - } - if result.contains(IoEvents::OUT) { - result |= IoEvents::WRNORM; - } // POSIX: POLLHUP and POLLERR are always reported in revents, // even if not requested in events. They must NOT be masked out. let always_report = @@ -104,15 +98,15 @@ fn do_poll( let mut revent_indices = Vec::with_capacity(poll_fds.len()); for (index, fd) in poll_fds.iter_mut().enumerate() { fd.revents = 0; - if fd.fd == -1 { - // Skip -1 + if fd.fd < 0 { + // Linux ignores every negative descriptor and returns zero revents. continue; } match get_file_like(fd.fd) { Ok(f) => { fds.push(( f, - IoEvents::from_bits(fd.events as _).ok_or(AxError::InvalidInput)? + IoEvents::from_bits_truncate(u32::from(fd.events as u16)) | IoEvents::ALWAYS_POLL, )); revent_indices.push(index); @@ -162,11 +156,13 @@ pub fn sys_poll(fds: UserPtr, nfds: u32, timeout: i32) -> AxResult 0 { write_poll_revents(fds, &poll_fds)?; } - Ok(res) + res } pub fn sys_ppoll( @@ -188,11 +184,13 @@ pub fn sys_ppoll( &mut poll_fds, timeout, nullable!(sigmask.get_as_ref())?.copied(), - )?; + ); + // Match poll(2): interruption does not leave the caller's old revents + // values visible, and a failed writeback is reported as EFAULT. if nfds > 0 { write_poll_revents(fds, &poll_fds)?; } - Ok(res) + res } #[cfg(axtest)] diff --git a/os/StarryOS/kernel/src/syscall/mod.rs b/os/StarryOS/kernel/src/syscall/mod.rs index 9189df9a6e..32eab3c79a 100644 --- a/os/StarryOS/kernel/src/syscall/mod.rs +++ b/os/StarryOS/kernel/src/syscall/mod.rs @@ -24,12 +24,34 @@ pub use self::{ use crate::task::{AsThread, SeccompDecision, do_exit, seccomp_errno}; pub fn syscall_allows_signal_restart(sysno: usize) -> bool { - // Per signal(7), only the System V message-queue blocking calls (msgsnd / - // msgrcv) are never restarted even with SA_RESTART. The POSIX message-queue - // calls (mq_send / mq_receive / mq_timedsend / mq_timedreceive) ARE in the - // SA_RESTART-restartable set, so they must not be listed here or a handler - // installed with SA_RESTART would wrongly see EINTR. - !matches!(Sysno::new(sysno), Some(Sysno::msgsnd | Sysno::msgrcv)) + // Linux never restarts fd-multiplexing waits or System V message-queue + // blocking calls, even when the delivered handler uses SA_RESTART. Keep + // the classification here because signal delivery only sees the syscall + // number and the interrupted -EINTR result. + let Some(sysno) = Sysno::new(sysno) else { + return true; + }; + + if matches!( + sysno, + Sysno::ppoll + | Sysno::pselect6 + | Sysno::epoll_pwait + | Sysno::epoll_pwait2 + | Sysno::msgsnd + | Sysno::msgrcv + ) { + return false; + } + + // The legacy multiplexing entry points exist in the x86_64 syscall table + // but not in the generic tables used by riscv64, aarch64, and loongarch64. + #[cfg(target_arch = "x86_64")] + if matches!(sysno, Sysno::poll | Sysno::select | Sysno::epoll_wait) { + return false; + } + + true } // `#[inline(never)]` keeps `sysno` reachable as a real call target so a kprobe @@ -1051,13 +1073,24 @@ pub(crate) fn membarrier_validation_rules_hold_for_test() -> bool { #[cfg(axtest)] pub(crate) fn syscall_signal_restart_rules_hold_for_test() -> bool { - // syscall_allows_signal_restart: returns false only for msgsnd and msgrcv. use syscalls::Sysno; - assert!(syscall_allows_signal_restart(0)); // invalid syscall → true - assert!(syscall_allows_signal_restart(Sysno::read as usize)); // read → true - assert!(syscall_allows_signal_restart(Sysno::write as usize)); // write → true - assert!(!syscall_allows_signal_restart(Sysno::msgsnd as usize)); // msgsnd → false - assert!(!syscall_allows_signal_restart(Sysno::msgrcv as usize)); // msgrcv → false + + assert!(syscall_allows_signal_restart(Sysno::read as usize)); + assert!(syscall_allows_signal_restart(Sysno::write as usize)); + for sysno in [ + Sysno::ppoll, + Sysno::pselect6, + Sysno::epoll_pwait, + Sysno::epoll_pwait2, + Sysno::msgsnd, + Sysno::msgrcv, + ] { + assert!(!syscall_allows_signal_restart(sysno as usize)); + } + #[cfg(target_arch = "x86_64")] + for sysno in [Sysno::poll, Sysno::select, Sysno::epoll_wait] { + assert!(!syscall_allows_signal_restart(sysno as usize)); + } true } diff --git a/os/StarryOS/kernel/tests/cases/axtest_fs.rs b/os/StarryOS/kernel/tests/cases/axtest_fs.rs index bfbe259bbf..2a3a3bd170 100644 --- a/os/StarryOS/kernel/tests/cases/axtest_fs.rs +++ b/os/StarryOS/kernel/tests/cases/axtest_fs.rs @@ -11,6 +11,16 @@ fn pipe_resize_rejects_oversized_pipe() { ax_assert!(axtest_exports::pipe_resize_rejects_oversized_pipe()); } +#[axtest] +fn pipe_linux_io_semantics_hold() { + ax_assert!(axtest_exports::pipe_linux_io_semantics_hold()); +} + +#[axtest] +fn interrupted_pipe_write_preserves_partial_progress() { + ax_assert!(axtest_exports::interrupted_pipe_write_preserves_partial_progress()); +} + #[axtest] fn fcntl_setpipe_size_returns_capacity() { ax_assert!(axtest_exports::fcntl_setpipe_size_returns_capacity()); @@ -26,6 +36,26 @@ fn concurrent_epoll_reverse_add_is_serialized() { ax_assert!(axtest_exports::concurrent_epoll_reverse_add_is_serialized()); } +#[axtest] +fn epoll_level_aliases_rotate_in_linux_callback_order() { + ax_assert!(axtest_exports::epoll_level_aliases_rotate_in_linux_callback_order()); +} + +#[axtest] +fn epoll_edge_readiness_requires_a_new_notification() { + ax_assert!(axtest_exports::epoll_edge_readiness_requires_a_new_notification()); +} + +#[axtest] +fn epoll_edge_callback_does_not_reenter_target() { + ax_assert!(axtest_exports::epoll_edge_callback_does_not_reenter_target()); +} + +#[axtest] +fn epoll_hup_does_not_synthesize_readable() { + ax_assert!(axtest_exports::epoll_hup_does_not_synthesize_readable()); +} + #[axtest] fn pipe_resize_rounding_and_state_rules_hold() { ax_assert!(axtest_exports::pipe_resize_rounding_and_state_rules_hold()); diff --git a/test-suit/starryos/qemu/system/bugfix-signalfd-epoll-wakeup/src/main.c b/test-suit/starryos/qemu/system/bugfix-signalfd-epoll-wakeup/src/main.c index e0f5909779..61bb3f2895 100644 --- a/test-suit/starryos/qemu/system/bugfix-signalfd-epoll-wakeup/src/main.c +++ b/test-suit/starryos/qemu/system/bugfix-signalfd-epoll-wakeup/src/main.c @@ -17,6 +17,7 @@ static int passed; static int failed; static _Atomic int waiter_entered; +static _Atomic int waiter_finished; struct waiter_context { int epoll_fd; @@ -49,6 +50,7 @@ static void *wait_for_signalfd_event(void *opaque) context->wait_result = epoll_wait(context->epoll_fd, &context->event, 1, 2000); context->wait_errno = errno; if (context->wait_result != 1) { + atomic_store_explicit(&waiter_finished, 1, memory_order_release); return NULL; } @@ -59,6 +61,7 @@ static void *wait_for_signalfd_event(void *opaque) sizeof(context->signal_info) ); context->read_errno = errno; + atomic_store_explicit(&waiter_finished, 1, memory_order_release); return NULL; } @@ -76,23 +79,29 @@ static int wait_for_epoll_waiter(void) return nanosleep(&settle, NULL); } -static int check_inherited_signalfd_epoll(int epoll_fd, int signal_fd) +static int update_and_consume_child_signal(int signal_fd) { - struct epoll_event event; + sigset_t mask; struct signalfd_siginfo signal_info; - errno = 0; - int wait_result = epoll_wait(epoll_fd, &event, 1, 500); - if (wait_result != 0) { - printf("FAIL: inherited epoll reported signalfd in child: result=%d errno=%d (%s)\n", - wait_result, errno, strerror(errno)); + sigemptyset(&mask); + sigaddset(&mask, SIGUSR1); + if (signalfd(signal_fd, &mask, SFD_CLOEXEC | SFD_NONBLOCK) != signal_fd) { + printf("FAIL: child could not update inherited signalfd: errno=%d (%s)\n", + errno, strerror(errno)); + return EXIT_FAILURE; + } + + if (kill(getpid(), SIGUSR1) != 0) { + printf("FAIL: child could not queue SIGUSR1: errno=%d (%s)\n", + errno, strerror(errno)); return EXIT_FAILURE; } errno = 0; ssize_t read_length = read(signal_fd, &signal_info, sizeof(signal_info)); if (read_length != (ssize_t)sizeof(signal_info)) { - printf("FAIL: inherited signalfd did not read child SIGUSR1: result=%zd errno=%d (%s)\n", + printf("FAIL: inherited signalfd did not read self-sent SIGUSR1: result=%zd errno=%d (%s)\n", read_length, errno, strerror(errno)); return EXIT_FAILURE; } @@ -102,7 +111,7 @@ static int check_inherited_signalfd_epoll(int epoll_fd, int signal_fd) return EXIT_FAILURE; } - printf("PASS: inherited signalfd reads child SIGUSR1 without epoll readiness\n"); + printf("PASS: inherited signalfd reads the child's self-sent SIGUSR1\n"); return EXIT_SUCCESS; } @@ -113,9 +122,31 @@ static void test_forked_child_signalfd_epoll_isolation(int epoll_fd, int signal_ .tv_nsec = 100 * 1000 * 1000, }; int ready_pipe[2] = {-1, -1}; + struct waiter_context context = { + .epoll_fd = epoll_fd, + .signal_fd = signal_fd, + .wait_result = -1, + .wait_errno = 0, + .read_length = -1, + .read_errno = 0, + }; + pthread_t waiter; + + atomic_store_explicit(&waiter_entered, 0, memory_order_release); + atomic_store_explicit(&waiter_finished, 0, memory_order_release); + int waiter_started = pthread_create(&waiter, NULL, wait_for_signalfd_event, + &context) == 0; + expect_true(waiter_started, "start parent EPOLLET waiter before fork"); + if (!waiter_started) { + return; + } + expect_true(wait_for_epoll_waiter() == 0, + "wait for parent EPOLLET waiter to block"); expect_true(pipe(ready_pipe) == 0, "create fork readiness pipe"); if (ready_pipe[0] < 0 || ready_pipe[1] < 0) { + pthread_kill(waiter, SIGUSR1); + pthread_join(waiter, NULL); return; } @@ -124,13 +155,15 @@ static void test_forked_child_signalfd_epoll_isolation(int epoll_fd, int signal_ expect_true(child >= 0, "fork inherited signalfd and epoll"); if (child == 0) { const char ready = 'R'; + int child_result; close(ready_pipe[0]); + child_result = update_and_consume_child_signal(signal_fd); if (write(ready_pipe[1], &ready, sizeof(ready)) != (ssize_t)sizeof(ready)) { _exit(EXIT_FAILURE); } close(ready_pipe[1]); - _exit(check_inherited_signalfd_epoll(epoll_fd, signal_fd)); + _exit(child_result); } close(ready_pipe[1]); @@ -138,16 +171,29 @@ static void test_forked_child_signalfd_epoll_isolation(int epoll_fd, int signal_ char ready = '\0'; expect_true(read(ready_pipe[0], &ready, sizeof(ready)) == (ssize_t)sizeof(ready) && ready == 'R', - "wait for child epoll_wait setup"); - expect_true(nanosleep(&settle, NULL) == 0, "let child enter epoll_wait"); - expect_true(kill(child, SIGUSR1) == 0, "send SIGUSR1 to child process"); + "wait for child signalfd activity"); int status = 0; expect_true(waitpid(child, &status, 0) == child, "wait for child process"); expect_true(WIFEXITED(status) && WEXITSTATUS(status) == EXIT_SUCCESS, - "inherited epoll ignores child signalfd readiness"); + "child consumes its own signal through inherited signalfd"); + expect_true(nanosleep(&settle, NULL) == 0, + "let the parent waiter refresh its signalfd registration"); + expect_true(atomic_load_explicit(&waiter_finished, memory_order_acquire) == 0, + "child signalfd activity does not publish parent epoll readiness"); + expect_true(pthread_kill(waiter, SIGUSR1) == 0, + "send SIGUSR1 to the blocked parent waiter after child activity"); } close(ready_pipe[0]); + + expect_true(pthread_join(waiter, NULL) == 0, + "join parent EPOLLET waiter after fork"); + expect_true(context.wait_result == 1 && context.event.data.fd == signal_fd && + (context.event.events & EPOLLIN) != 0, + "parent EPOLLET waiter remains registered after child activity"); + expect_true(context.read_length == (ssize_t)sizeof(context.signal_info) && + context.signal_info.ssi_signo == SIGUSR1, + "parent waiter reads its post-fork SIGUSR1"); } int main(void) @@ -166,7 +212,7 @@ int main(void) expect_true(epoll_fd >= 0, "create epoll"); struct epoll_event interest = { - .events = EPOLLIN, + .events = EPOLLIN | EPOLLET, .data.fd = signal_fd, }; expect_true(signal_fd >= 0 && epoll_fd >= 0 && @@ -183,6 +229,8 @@ int main(void) .read_errno = 0, }; pthread_t waiter; + atomic_store_explicit(&waiter_entered, 0, memory_order_release); + atomic_store_explicit(&waiter_finished, 0, memory_order_release); int waiter_started = signal_fd >= 0 && epoll_fd >= 0 && pthread_create(&waiter, NULL, wait_for_signalfd_event, &context) == 0; expect_true(waiter_started, "start epoll_wait thread"); diff --git a/test-suit/starryos/qemu/system/syscall-test-eventfd2/src/main.c b/test-suit/starryos/qemu/system/syscall-test-eventfd2/src/main.c index bf296db574..3f56a03237 100644 --- a/test-suit/starryos/qemu/system/syscall-test-eventfd2/src/main.c +++ b/test-suit/starryos/qemu/system/syscall-test-eventfd2/src/main.c @@ -8,12 +8,13 @@ * 4. 信号量模式：read 每次返回 1 并递减 * 5. 多次写入累积 * 6. 写 UINT64_MAX → EINVAL - * 7. 读写缓冲区大小校验（< 8 字节 → EINVAL） + * 7. 读写缓冲区大小校验（eventfd 写入必须恰好为 8 字节） * 8. 非阻塞模式：空 eventfd 读 → EAGAIN，满 eventfd 写 → EAGAIN * 9. 写 0 边界情况 * 10. 计数器溢出保护：写会使计数超过 UINT64_MAX-1 → EAGAIN * 11. 阻塞读：子进程写入后父进程阻塞读被唤醒 * 12. fork 继承：子进程可以读写父进程创建的 eventfd + * 13. poll 返回精确的 Linux readiness mask，并清空负 fd 的 revents */ #ifndef _GNU_SOURCE #define _GNU_SOURCE @@ -21,11 +22,15 @@ #include "test_framework.h" #include +#include #include #include #include #include +#include +#include #include +#include #include /* 标准 eventfd 读写辅助函数 */ @@ -37,6 +42,15 @@ static int do_read(int fd, uint64_t *val) { return (int)read(fd, val, sizeof(*val)); } +static long raw_ppoll(struct pollfd *fds, nfds_t nfds) { + const struct timespec timeout = { + .tv_sec = 0, + .tv_nsec = 0, + }; + + return syscall(SYS_ppoll, fds, nfds, &timeout, NULL, 0); +} + /* ─── 1. 基本创建与 flags ─────────────────────────────────── */ static void test_create_with_flags(void) { @@ -206,6 +220,55 @@ static void test_buffer_size_validation(void) { CHECK_RET(do_read(fd, &val64), (ssize_t)sizeof(val64), "read with 8-byte buffer succeeds"); CHECK(val64 == 1, "read with 8-byte buffer returns initval 1"); + /* eventfd checks count before touching the user pointer. */ + errno = 0; + ret = syscall(SYS_write, fd, (const void *)(uintptr_t)1, 7); + CHECK(ret == -1 && errno == EINVAL, + "short raw write with invalid pointer returns EINVAL before EFAULT"); + + errno = 0; + ret = syscall(SYS_write, fd, (const void *)(uintptr_t)1, 9); + CHECK(ret == -1 && errno == EINVAL, + "oversized raw write with invalid pointer returns EINVAL before EFAULT"); + + close(fd); + } + + { + int fd = eventfd(0, EFD_NONBLOCK); + uint8_t oversized[sizeof(uint64_t) + 1]; + uint64_t written = 17; + uint64_t observed = 0; + + CHECK(fd >= 0, "fd for oversized raw write test"); + memcpy(oversized, &written, sizeof(written)); + oversized[sizeof(written)] = 0xa5; + + errno = 0; + long ret = syscall(SYS_write, fd, oversized, sizeof(oversized)); + CHECK(ret == -1 && errno == EINVAL, + "raw write with 9-byte buffer returns EINVAL"); + CHECK_ERR(do_read(fd, &observed), EAGAIN, + "rejected oversized raw write leaves the counter unchanged"); + + struct iovec oversized_iov = { + .iov_base = oversized, + .iov_len = sizeof(oversized), + }; + errno = 0; + ret = syscall(SYS_writev, fd, &oversized_iov, 1); + CHECK(ret == -1 && errno == EINVAL, + "raw writev with one 9-byte segment returns EINVAL"); + + struct iovec invalid_oversized_iov = { + .iov_base = (void *)(uintptr_t)1, + .iov_len = sizeof(oversized), + }; + errno = 0; + ret = syscall(SYS_writev, fd, &invalid_oversized_iov, 1); + CHECK(ret == -1 && errno == EINVAL, + "oversized raw writev validates total length before iov_base"); + close(fd); } } @@ -375,6 +438,27 @@ static void test_fork_inheritance(void) { close(fd); } +/* ─── 13. poll readiness mask ───────────────────────────── */ + +static void test_poll_readiness_mask(void) { + int fd = eventfd(1, EFD_NONBLOCK); + struct pollfd fds[] = { + {.fd = fd, .events = INT16_MAX, .revents = (short)0x5a5a}, + {.fd = -2, .events = INT16_MAX, .revents = (short)0x5a5a}, + {.fd = -2, .events = POLLIN | POLLOUT, .revents = (short)0x5a5a}, + {.fd = -2, .events = POLLERR, .revents = (short)0x5a5a}, + }; + + CHECK(fd >= 0, "fd for poll readiness mask test"); + CHECK_RET(raw_ppoll(fds, sizeof(fds) / sizeof(fds[0])), 1, + "poll reports exactly one ready eventfd"); + CHECK(fds[0].revents == (POLLIN | POLLOUT), + "eventfd poll reports only Linux POLLIN|POLLOUT readiness"); + CHECK(fds[1].revents == 0 && fds[2].revents == 0 && fds[3].revents == 0, + "poll ignores negative fds and clears their revents"); + close(fd); +} + /* ─── main ────────────────────────────────────────────────── */ int main(void) { @@ -416,5 +500,8 @@ int main(void) { printf("\n--- 12. fork inheritance ---\n"); test_fork_inheritance(); + printf("\n--- 13. poll readiness mask ---\n"); + test_poll_readiness_mask(); + TEST_DONE(); } diff --git a/test-suit/starryos/qemu/system/syscall-test-select-poll-family/CMakeLists.txt b/test-suit/starryos/qemu/system/syscall-test-select-poll-family/CMakeLists.txt index e1aa390e47..729a20f66f 100644 --- a/test-suit/starryos/qemu/system/syscall-test-select-poll-family/CMakeLists.txt +++ b/test-suit/starryos/qemu/system/syscall-test-select-poll-family/CMakeLists.txt @@ -35,6 +35,7 @@ set(SOURCES src/poll_regular_file.c src/poll_closed_fd.c src/poll_pipe_hup.c + src/poll_unknown_events.c src/pselect_sigmask_block.c src/pselect_sigmask_restore.c src/pselect_no_sigmask.c diff --git a/test-suit/starryos/qemu/system/syscall-test-select-poll-family/src/main.c b/test-suit/starryos/qemu/system/syscall-test-select-poll-family/src/main.c index 80da5ac3a9..b01ddc695f 100644 --- a/test-suit/starryos/qemu/system/syscall-test-select-poll-family/src/main.c +++ b/test-suit/starryos/qemu/system/syscall-test-select-poll-family/src/main.c @@ -30,6 +30,7 @@ extern int run_poll_multiple_fds(void); extern int run_poll_regular_file(void); extern int run_poll_closed_fd(void); extern int run_poll_pipe_hup(void); +extern int run_poll_unknown_events(void); extern int run_pselect_sigmask_block(void); extern int run_pselect_sigmask_restore(void); @@ -67,7 +68,7 @@ static void run_module(const char *name, int (*fn)(void)) { int main(void) { printf("================================================\n"); printf(" TEST: select/poll/pselect6/ppoll deep suite\n"); - printf(" 43 modules, ~300+ checkpoints\n"); + printf(" 44 modules, ~300+ checkpoints\n"); printf("================================================\n"); printf("\n=== Phase 1: select basics ===\n"); @@ -100,6 +101,7 @@ int main(void) { run_module("poll_regular_file", run_poll_regular_file); run_module("poll_closed_fd", run_poll_closed_fd); run_module("poll_pipe_hup", run_poll_pipe_hup); + run_module("poll_unknown_events", run_poll_unknown_events); printf("\n=== Phase 3: signal interaction ===\n"); run_module("pselect_sigmask_block", run_pselect_sigmask_block); diff --git a/test-suit/starryos/qemu/system/syscall-test-select-poll-family/src/poll_unknown_events.c b/test-suit/starryos/qemu/system/syscall-test-select-poll-family/src/poll_unknown_events.c new file mode 100644 index 0000000000..250c96bb9e --- /dev/null +++ b/test-suit/starryos/qemu/system/syscall-test-select-poll-family/src/poll_unknown_events.c @@ -0,0 +1,84 @@ +#include "test_framework.h" +#include "helpers.h" + +#include +#include +#include +#include +#include + +#define UNKNOWN_POLL_EVENT ((short)0x0800) +#define KERNEL_SIGSET_SIZE (sizeof(unsigned long)) + +int run_poll_unknown_events(void) { + MODULE_START("poll_unknown_events"); + long ret; + +#ifdef SYS_poll + int poll_fds[2]; + CHECK_RET(create_pipe(poll_fds), 0, "poll pipe created"); + + struct pollfd poll_fd = { + .fd = poll_fds[0], + .events = UNKNOWN_POLL_EVENT, + .revents = 0, + }; + errno = 0; + ret = syscall(SYS_poll, &poll_fd, 1, 0); + CHECK(ret == 0 && errno == 0, + "poll ignores an unknown event while the writer is open"); + CHECK(poll_fd.revents == 0, + "poll does not echo an unknown event while the writer is open"); + + CHECK_RET(close(poll_fds[1]), 0, "poll writer closed"); + poll_fd.revents = 0; + errno = 0; + ret = syscall(SYS_poll, &poll_fd, 1, 0); + CHECK(ret == 1 && errno == 0, + "poll reports one ready fd instead of EINVAL after writer close"); + CHECK((poll_fd.revents & POLLHUP) != 0, + "poll reports POLLHUP even when only an unknown event was requested"); + CHECK((poll_fd.revents & UNKNOWN_POLL_EVENT) == 0, + "poll does not echo the unknown event after writer close"); + + CHECK_RET(close(poll_fds[0]), 0, "poll reader closed"); +#else + printf(" SKIP | raw SYS_poll is unavailable on this architecture\n"); +#endif + + int ppoll_fds[2]; + CHECK_RET(create_pipe(ppoll_fds), 0, "ppoll pipe created"); + + struct pollfd ppoll_fd = { + .fd = ppoll_fds[0], + .events = UNKNOWN_POLL_EVENT, + .revents = 0, + }; + struct timespec timeout = { .tv_sec = 0, .tv_nsec = 0 }; + errno = 0; + ret = syscall(SYS_ppoll, &ppoll_fd, 1, &timeout, NULL, + KERNEL_SIGSET_SIZE); + CHECK(ret == 0 && errno == 0, + "ppoll ignores an unknown event while the writer is open"); + CHECK(ppoll_fd.revents == 0, + "ppoll does not echo an unknown event while the writer is open"); + + CHECK_RET(close(ppoll_fds[1]), 0, "ppoll writer closed"); + ppoll_fd.revents = 0; + timeout.tv_sec = 0; + timeout.tv_nsec = 0; + errno = 0; + ret = syscall(SYS_ppoll, &ppoll_fd, 1, &timeout, NULL, + KERNEL_SIGSET_SIZE); + CHECK(ret == 1 && errno == 0, + "ppoll reports one ready fd instead of EINVAL after writer close"); + CHECK((ppoll_fd.revents & POLLHUP) != 0, + "ppoll reports POLLHUP even when only an unknown event was requested"); + CHECK((ppoll_fd.revents & UNKNOWN_POLL_EVENT) == 0, + "ppoll does not echo the unknown event after writer close"); + + CHECK_RET(close(ppoll_fds[0]), 0, "ppoll reader closed"); + + MODULE_SUMMARY("poll_unknown_events"); + MODULE_RETURN(); +}