From 09e5459e595d93a343761291e47829e473ffba2b Mon Sep 17 00:00:00 2001 From: Weiteng Chen Date: Wed, 26 Aug 2026 15:40:59 -0700 Subject: [PATCH 1/4] Expand Windows loader compatibility --- litebox_shim_windows/src/syscalls/file.rs | 4 +++- litebox_shim_windows/src/syscalls/lpc.rs | 20 ++++++++++++++++++++ litebox_shim_windows/src/syscalls/section.rs | 16 ++++++++++++++-- 3 files changed, 37 insertions(+), 3 deletions(-) diff --git a/litebox_shim_windows/src/syscalls/file.rs b/litebox_shim_windows/src/syscalls/file.rs index 09ffe5c72..5532191d7 100644 --- a/litebox_shim_windows/src/syscalls/file.rs +++ b/litebox_shim_windows/src/syscalls/file.rs @@ -981,11 +981,13 @@ impl Task { Ok(status) => status, Err(FileStatusError::PathError(PathError::NoSuchFileOrDirectory)) => { let parent = parent_directory_path(&path); - return if self.fs.file_status(parent).is_ok() { + let status = if self.fs.file_status(parent).is_ok() { NtStatus::OBJECT_NAME_NOT_FOUND } else { NtStatus::OBJECT_PATH_NOT_FOUND }; + litebox_util_log::debug!(path:% = path, status:? = status; "NtQueryAttributesFile path not found"); + return status; } Err(error) => return map_file_status_error(error), }; diff --git a/litebox_shim_windows/src/syscalls/lpc.rs b/litebox_shim_windows/src/syscalls/lpc.rs index 8202b4cbb..8926a370d 100644 --- a/litebox_shim_windows/src/syscalls/lpc.rs +++ b/litebox_shim_windows/src/syscalls/lpc.rs @@ -781,6 +781,26 @@ pub(crate) struct ConnectPortParameters { } impl Task { + pub(crate) fn sys_nt_alpc_connect_port( + port_handle: MutPtr, + port_name: ConstPtr, + ) -> NtStatus { + if port_handle.write_at_offset(0, Handle::default()).is_none() { + return NtStatus::ACCESS_VIOLATION; + } + let port_name = match port_name + .read_at_offset(0) + .ok_or(NtStatus::ACCESS_VIOLATION) + .and_then(UnicodeString::read_string::) + { + Ok(name) => name, + Err(status) => return status, + }; + + litebox_util_log::debug!(port_name:% = port_name; "ALPC port is unavailable"); + NtStatus::OBJECT_NAME_NOT_FOUND + } + pub(crate) fn sys_nt_connect_port(&self, params: ConnectPortParameters) -> NtStatus { if params.security_qos.read_at_offset(0).is_none() { return NtStatus::ACCESS_VIOLATION; diff --git a/litebox_shim_windows/src/syscalls/section.rs b/litebox_shim_windows/src/syscalls/section.rs index 798280645..ad95c30a6 100644 --- a/litebox_shim_windows/src/syscalls/section.rs +++ b/litebox_shim_windows/src/syscalls/section.rs @@ -766,12 +766,20 @@ impl Task { // anonymous backing that avoids kernel-side content storage. return Err(NtStatus::NOT_SUPPORTED); } + let is_reserved = section + .attributes + .contains(SectionAllocationAttributes::SEC_RESERVE); + let initial_permissions = if is_reserved { + MemoryRegionPermissions::empty() + } else { + permissions + }; let mapping = create_pages( &self.global.page_manager, None, length, CreatePagesFlags::empty(), - permissions, + initial_permissions, |_| Ok(0), ) .map_err(|_| { @@ -794,7 +802,11 @@ impl Task { size: mapped_size, allocation_protect: section.protection, type_: MemoryType::MEM_MAPPED, - pages: committed_pages(base, mapped_size, page_protection), + pages: if is_reserved { + RangeMap::new() + } else { + committed_pages(base, mapped_size, page_protection) + }, }, ); Ok(MappedPagefileSectionView { base, view_size }) From a953a860daf6d1f8479f5488df012b61695926f2 Mon Sep 17 00:00:00 2001 From: Weiteng Chen Date: Wed, 26 Aug 2026 15:41:48 -0700 Subject: [PATCH 2/4] Implement I/O completion port queues --- litebox_shim_windows/src/syscalls/iocp.rs | 166 +++++++++++++++++++++- litebox_shim_windows/src/syscalls/wait.rs | 2 +- 2 files changed, 162 insertions(+), 6 deletions(-) diff --git a/litebox_shim_windows/src/syscalls/iocp.rs b/litebox_shim_windows/src/syscalls/iocp.rs index c4b8a008a..1ea9c83cf 100644 --- a/litebox_shim_windows/src/syscalls/iocp.rs +++ b/litebox_shim_windows/src/syscalls/iocp.rs @@ -3,16 +3,25 @@ //! Windows NT I/O completion port syscalls. -use alloc::sync::Arc; +use alloc::collections::VecDeque; +use alloc::sync::{Arc, Weak}; use core::marker::PhantomData; +use litebox::event::observer::Observer; +use litebox::event::polling::{Pollee, TryOpError}; +use litebox::event::wait::WaitError; +use litebox::event::{Events, IOPollable}; use litebox::fd::{FdEnabledSubsystem, FdEnabledSubsystemEntry}; -use litebox::platform::{RawMutPointer as _, RawPointerProvider}; +use litebox::platform::{RawConstPointer as _, RawMutPointer as _, RawPointerProvider}; +use litebox::sync::Mutex; use litebox_common_windows::nt_status::NtStatus; +use zerocopy::{FromBytes, Immutable, IntoBytes}; use crate::nt_types::{AccessMask, ObjectAttributes, read_object_attributes}; use crate::syscalls::Handle; -use crate::{ConstPtr, MutPtr, ShimFS, Task, probe_guest_output_preserving_value}; +use crate::{ + ConstPtr, MutPtr, ShimFS, Task, probe_guest_output_buffer, probe_guest_output_preserving_value, +}; bitflags::bitflags! { #[derive(Clone, Copy, Debug, Eq, PartialEq)] @@ -65,14 +74,50 @@ pub(crate) struct IoCompletionHandleObject { pub(crate) struct IoCompletionObject { _number_of_concurrent_threads: u32, - _not_send_without_platform: PhantomData, + packets: Mutex>, + pollee: Pollee, +} + +#[repr(C)] +#[derive(Clone, Copy, FromBytes, Immutable, IntoBytes)] +pub(crate) struct IoCompletionPacket { + completion_key: usize, + completion_value: usize, + status: usize, + information: usize, } impl IoCompletionObject { fn new(number_of_concurrent_threads: u32) -> Self { Self { _number_of_concurrent_threads: number_of_concurrent_threads, - _not_send_without_platform: PhantomData, + packets: Mutex::new(VecDeque::new()), + pollee: Pollee::new(), + } + } + + fn post(&self, packet: IoCompletionPacket) { + self.packets.lock().push_back(packet); + self.pollee.notify_observers(Events::IN); + } + + fn remove(&self, count: usize) -> VecDeque { + let mut packets = self.packets.lock(); + let remove_count = count.min(packets.len()); + packets.drain(..remove_count).collect() + } +} + +impl IOPollable for IoCompletionObject { + fn register_observer(&self, observer: Weak>, mask: Events) { + self.pollee.register_observer(observer, mask); + } + + fn check_io_events(&self) -> Events { + if self.packets.lock().is_empty() { + Events::empty() + } else { + Events::IN } } } @@ -148,6 +193,117 @@ impl Task { } NtStatus::SUCCESS } + + pub(crate) fn sys_nt_set_io_completion( + &self, + io_completion_handle: Handle, + completion_key: usize, + completion_value: usize, + status: i32, + information: usize, + ) -> NtStatus { + let entry = match self.typed_handle_entry_with_access::>( + io_completion_handle, + IoCompletionAccess::MODIFY_STATE.bits(), + ) { + Ok(entry) => entry, + Err(status) => return status, + }; + entry.with_entry(|entry| { + entry.port.post(IoCompletionPacket { + completion_key, + completion_value, + status: status.cast_unsigned() as usize, + information, + }); + }); + NtStatus::SUCCESS + } + + pub(crate) fn sys_nt_remove_io_completion_ex( + &self, + io_completion_handle: Handle, + io_completion_information: MutPtr, + count: u32, + num_entries_removed: MutPtr, + timeout: Option>, + alertable: bool, + ) -> NtStatus { + if count == 0 { + return NtStatus::INVALID_PARAMETER; + } + if let Err(status) = probe_guest_output_buffer::( + MutPtr::::from_usize(io_completion_information.as_usize()), + (count as usize).saturating_mul(core::mem::size_of::()), + ) { + return status; + } + if let Err(status) = probe_guest_output_preserving_value::(num_entries_removed) + { + return status; + } + let timeout = match timeout { + Some(timeout) => match timeout.read_at_offset(0) { + Some(timeout) => Some(self.wait_timeout_duration(timeout)), + None => return NtStatus::ACCESS_VIOLATION, + }, + None => None, + }; + let entry = match self.typed_handle_entry_with_access::>( + io_completion_handle, + IoCompletionAccess::MODIFY_STATE.bits(), + ) { + Ok(entry) => entry, + Err(status) => return status, + }; + let port = entry.with_entry(IoCompletionHandleObject::port); + if alertable { + litebox_util_log::debug!("Treating alertable IOCP wait as non-alertable"); + } + + let wait_cx = self.wait_cx().with_timeout(timeout); + match wait_cx.wait_on_events( + false, + Events::IN, + |observer, mask| { + port.register_observer(observer, mask); + Ok(()) + }, + || { + let packets = port.remove(count as usize); + if packets.is_empty() { + return Err(TryOpError::::TryAgain); + } + for (index, packet) in packets.iter().copied().enumerate() { + if io_completion_information + .write_at_offset( + index.try_into().expect("packet count fits in isize"), + packet, + ) + .is_none() + { + return Err(TryOpError::Other(NtStatus::ACCESS_VIOLATION)); + } + } + Ok(packets + .len() + .try_into() + .expect("packet count is bounded by the u32 input count")) + }, + ) { + Ok(removed) => { + if num_entries_removed.write_at_offset(0, removed).is_none() { + NtStatus::ACCESS_VIOLATION + } else { + NtStatus::SUCCESS + } + } + Err(TryOpError::WaitError(WaitError::TimedOut)) => NtStatus::TIMEOUT, + Err(TryOpError::WaitError(WaitError::Interrupted)) => NtStatus::ALERTED, + Err(TryOpError::TryAgain) => unreachable!("blocking wait cannot return TryAgain"), + Err(TryOpError::Other(status)) => status, + } + } } #[cfg(test)] diff --git a/litebox_shim_windows/src/syscalls/wait.rs b/litebox_shim_windows/src/syscalls/wait.rs index 28f5e2edd..c2c312413 100644 --- a/litebox_shim_windows/src/syscalls/wait.rs +++ b/litebox_shim_windows/src/syscalls/wait.rs @@ -157,7 +157,7 @@ impl Task { ) } - fn wait_timeout_duration(&self, timeout: i64) -> core::time::Duration { + pub(crate) fn wait_timeout_duration(&self, timeout: i64) -> core::time::Duration { const WINDOWS_TO_UNIX_EPOCH_SECONDS: u64 = 11_644_473_600; if timeout <= 0 { From 35add9daab9b3bf3a56b65656e13d47f4ce73190 Mon Sep 17 00:00:00 2001 From: Weiteng Chen Date: Sun, 30 Aug 2026 13:55:41 -0700 Subject: [PATCH 3/4] Expand Windows I/O completion port support --- litebox_shim_windows/src/lib.rs | 44 ++ litebox_shim_windows/src/syscalls/file.rs | 2 + litebox_shim_windows/src/syscalls/iocp.rs | 613 ++++++++++++++++-- litebox_shim_windows/src/syscalls/mod.rs | 120 ++++ .../src/syscalls/object_manager.rs | 34 + litebox_shim_windows/src/syscalls/registry.rs | 41 +- litebox_shim_windows/src/syscalls/thread.rs | 1 + litebox_shim_windows/src/syscalls/wait.rs | 4 +- litebox_shim_windows/src/tests.rs | 6 + litebox_shim_windows/src/wait.rs | 36 +- 10 files changed, 822 insertions(+), 79 deletions(-) diff --git a/litebox_shim_windows/src/lib.rs b/litebox_shim_windows/src/lib.rs index 3aae71ebf..9d64ad329 100644 --- a/litebox_shim_windows/src/lib.rs +++ b/litebox_shim_windows/src/lib.rs @@ -555,6 +555,7 @@ impl WindowsShim { process: process.clone(), fs, wait_state: wait::WaitState::new(self.0.platform), + io_completion_worker: Mutex::new(syscalls::iocp::IoCompletionWorkerState::new()), entry_point: load_info.entry_point, stack_top: load_info.stack_top, context: load_info.environment.context, @@ -737,6 +738,7 @@ struct Task { process: Arc>, fs: Arc, wait_state: wait::WaitState, + io_completion_worker: Mutex>, entry_point: usize, stack_top: usize, context: usize, @@ -748,6 +750,7 @@ impl Task { fn complete_current_thread(&self) { let thread_id = self.thread_object.thread_id(); self.thread_object.complete(|| { + self.release_io_completion_worker(); self.thread_object.abandon_owned_mutants(thread_id); self.process.detach_thread(thread_id); }); @@ -1174,6 +1177,47 @@ impl Task { object_attributes, number_of_concurrent_threads, ), + SyscallRequest::NtOpenIoCompletion { + io_completion_handle, + desired_access, + object_attributes, + } => self.sys_nt_open_io_completion( + io_completion_handle, + desired_access, + object_attributes, + ), + SyscallRequest::NtSetIoCompletion { + io_completion_handle, + completion_key, + completion_value, + status, + information, + } => self.sys_nt_set_io_completion( + io_completion_handle, + completion_key, + completion_value, + status, + information, + ), + SyscallRequest::NtRemoveIoCompletionEx { + io_completion_handle, + io_completion_information, + count, + num_entries_removed, + timeout, + alertable, + } => self.sys_nt_remove_io_completion_ex( + io_completion_handle, + io_completion_information, + count, + num_entries_removed, + timeout, + alertable, + ), + SyscallRequest::NtAlpcConnectPort { + port_handle, + port_name, + } => Self::sys_nt_alpc_connect_port(port_handle, port_name), SyscallRequest::NtConnectPort { port_handle, port_name, diff --git a/litebox_shim_windows/src/syscalls/file.rs b/litebox_shim_windows/src/syscalls/file.rs index 5532191d7..de5c26065 100644 --- a/litebox_shim_windows/src/syscalls/file.rs +++ b/litebox_shim_windows/src/syscalls/file.rs @@ -1288,6 +1288,8 @@ impl Task { length: u32, file_information_class: u32, ) -> NtStatus { + // TODO(windows-file-completion): Support FileCompletionInformation to associate a file + // handle with an IO completion port and its completion key. if FileHandleInformationClass::try_from(file_information_class) != Ok(FileHandleInformationClass::FilePositionInformation) { diff --git a/litebox_shim_windows/src/syscalls/iocp.rs b/litebox_shim_windows/src/syscalls/iocp.rs index 1ea9c83cf..7f5887dcf 100644 --- a/litebox_shim_windows/src/syscalls/iocp.rs +++ b/litebox_shim_windows/src/syscalls/iocp.rs @@ -3,7 +3,7 @@ //! Windows NT I/O completion port syscalls. -use alloc::collections::VecDeque; +use alloc::collections::{BTreeSet, VecDeque}; use alloc::sync::{Arc, Weak}; use core::marker::PhantomData; @@ -12,12 +12,12 @@ use litebox::event::polling::{Pollee, TryOpError}; use litebox::event::wait::WaitError; use litebox::event::{Events, IOPollable}; use litebox::fd::{FdEnabledSubsystem, FdEnabledSubsystemEntry}; -use litebox::platform::{RawConstPointer as _, RawMutPointer as _, RawPointerProvider}; +use litebox::platform::{RawConstPointer as _, RawMutPointer as _}; use litebox::sync::Mutex; use litebox_common_windows::nt_status::NtStatus; use zerocopy::{FromBytes, Immutable, IntoBytes}; -use crate::nt_types::{AccessMask, ObjectAttributes, read_object_attributes}; +use crate::nt_types::{AccessMask, ObjectAttributes, ObjectAttributesFlags}; use crate::syscalls::Handle; use crate::{ ConstPtr, MutPtr, ShimFS, Task, probe_guest_output_buffer, probe_guest_output_preserving_value, @@ -73,11 +73,17 @@ pub(crate) struct IoCompletionHandleObject { } pub(crate) struct IoCompletionObject { - _number_of_concurrent_threads: u32, + concurrency_limit: usize, + active_threads: Mutex>, packets: Mutex>, pollee: Pollee, } +pub(crate) struct IoCompletionWorkerState { + port: Option>>, + suspended: bool, +} + #[repr(C)] #[derive(Clone, Copy, FromBytes, Immutable, IntoBytes)] pub(crate) struct IoCompletionPacket { @@ -90,21 +96,81 @@ pub(crate) struct IoCompletionPacket { impl IoCompletionObject { fn new(number_of_concurrent_threads: u32) -> Self { Self { - _number_of_concurrent_threads: number_of_concurrent_threads, + concurrency_limit: number_of_concurrent_threads.max(1) as usize, + active_threads: Mutex::new(BTreeSet::new()), packets: Mutex::new(VecDeque::new()), pollee: Pollee::new(), } } fn post(&self, packet: IoCompletionPacket) { + // TODO(windows-iocp-file-posting): Post packets when associated asynchronous file I/O + // completes, rather than relying exclusively on NtSetIoCompletion. self.packets.lock().push_back(packet); self.pollee.notify_observers(Events::IN); } - fn remove(&self, count: usize) -> VecDeque { + fn remove_with( + &self, + count: usize, + mut write_packet: impl FnMut(usize, IoCompletionPacket) -> Result<(), NtStatus>, + mut write_count: impl FnMut(u32) -> Result<(), NtStatus>, + ) -> Result { let mut packets = self.packets.lock(); let remove_count = count.min(packets.len()); - packets.drain(..remove_count).collect() + if remove_count == 0 { + return Ok(false); + } + for (index, packet) in packets.iter().take(remove_count).copied().enumerate() { + write_packet(index, packet)?; + } + write_count( + remove_count + .try_into() + .expect("packet count is bounded by the u32 input count"), + )?; + packets.drain(..remove_count); + Ok(true) + } + + fn remove_with_capacity( + &self, + thread_id: usize, + count: usize, + write_packet: impl FnMut(usize, IoCompletionPacket) -> Result<(), NtStatus>, + write_count: impl FnMut(u32) -> Result<(), NtStatus>, + ) -> Result { + let mut active_threads = self.active_threads.lock(); + let already_active = active_threads.contains(&thread_id); + if !already_active && active_threads.len() >= self.concurrency_limit { + return Ok(false); + } + let removed = self.remove_with(count, write_packet, write_count)?; + if removed && !already_active { + active_threads.insert(thread_id); + } + Ok(removed) + } + + fn deactivate(&self, thread_id: usize) { + if self.active_threads.lock().remove(&thread_id) { + self.pollee.notify_observers(Events::IN); + } + } + + fn reactivate(&self, thread_id: usize) { + // Windows permits a previously associated worker that blocked elsewhere to resume even + // when this temporarily exceeds the port's concurrency limit. + self.active_threads.lock().insert(thread_id); + } +} + +impl IoCompletionWorkerState { + pub(crate) fn new() -> Self { + Self { + port: None, + suspended: false, + } } } @@ -128,20 +194,65 @@ impl IoCompletionHandleObject { } } -fn validate_io_completion_object_attributes( - object_attributes: Option>, -) -> Result<(), NtStatus> { - let Some(object_attributes) = object_attributes else { - return Ok(()); - }; - let object_attributes = read_object_attributes::(object_attributes)?; - if object_attributes.object_name == 0 && !object_attributes.root_directory.is_null() { - return Err(NtStatus::OBJECT_NAME_INVALID); +impl Task { + pub(crate) fn suspend_io_completion_worker(&self) { + let mut worker = self.io_completion_worker.lock(); + debug_assert!(!worker.suspended); + if let Some(port) = worker.port.as_ref().and_then(Weak::upgrade) { + port.deactivate(self.thread_object.thread_id()); + worker.suspended = true; + } else { + worker.port = None; + } + } + + pub(crate) fn resume_io_completion_worker(&self) { + let mut worker = self.io_completion_worker.lock(); + if worker.suspended { + if let Some(port) = worker.port.as_ref().and_then(Weak::upgrade) { + port.reactivate(self.thread_object.thread_id()); + } else { + worker.port = None; + } + worker.suspended = false; + } + } + + pub(crate) fn release_io_completion_worker(&self) { + let mut worker = self.io_completion_worker.lock(); + if let Some(port) = worker.port.take() + && !worker.suspended + && let Some(port) = port.upgrade() + { + port.deactivate(self.thread_object.thread_id()); + } + worker.suspended = false; + } + + /// Check if the task is associated with the given IO completion port. If the task is + /// associated with a different port, it will be deactivated from that port. + fn prepare_io_completion_wait(&self, port: &Arc>) -> bool { + let mut worker = self.io_completion_worker.lock(); + debug_assert!(!worker.suspended); + if worker + .port + .as_ref() + .is_some_and(|active_port| active_port.as_ptr() == Arc::as_ptr(port)) + { + return true; + } + if let Some(active_port) = worker.port.take().and_then(|port| port.upgrade()) { + active_port.deactivate(self.thread_object.thread_id()); + } + false + } + + fn associate_io_completion_worker(&self, port: &Arc>) { + let mut worker = self.io_completion_worker.lock(); + debug_assert!(worker.port.is_none()); + worker.port = Some(Arc::downgrade(port)); } - Ok(()) -} -impl Task { fn insert_io_completion_handle( &self, port: Arc>, @@ -174,16 +285,60 @@ impl Task { { return status; } - if let Err(status) = validate_io_completion_object_attributes::(object_attributes) - { - return status; + let (object_attributes, io_completion_name) = + match self.read_dispatcher_object_attributes(object_attributes, false) { + Ok(value) => value, + Err(status) => return status, + }; + let granted_access = IoCompletionAccess::from_desired_access(desired_access); + if let Some(io_completion_name) = io_completion_name { + let port = Arc::new(IoCompletionObject::new(number_of_concurrent_threads)); + return self.process.object_manager.create_io_completion( + &io_completion_name, + &port, + |port| { + let Some(object_attributes) = object_attributes else { + return NtStatus::INVALID_PARAMETER; + }; + if !ObjectAttributesFlags::from_bits_retain(object_attributes.attributes) + .contains(ObjectAttributesFlags::OPENIF) + { + return NtStatus::OBJECT_NAME_COLLISION; + } + self.publish_io_completion_handle( + io_completion_handle, + port, + granted_access, + NtStatus::OBJECT_NAME_EXISTS, + ) + }, + || { + self.publish_io_completion_handle( + io_completion_handle, + Arc::clone(&port), + granted_access, + NtStatus::SUCCESS, + ) + }, + ); } - // TODO: model the IOCP packet queue, concurrency accounting, named-object lookup, - // and file-handle association once completion posting/removal and file completion - // context syscalls are implemented. let port = Arc::new(IoCompletionObject::new(number_of_concurrent_threads)); - let granted_access = IoCompletionAccess::from_desired_access(desired_access); + self.publish_io_completion_handle( + io_completion_handle, + port, + granted_access, + NtStatus::SUCCESS, + ) + } + + fn publish_io_completion_handle( + &self, + io_completion_handle: MutPtr, + port: Arc>, + granted_access: IoCompletionAccess, + success_status: NtStatus, + ) -> NtStatus { let Ok(handle) = self.insert_io_completion_handle(port, granted_access) else { return NtStatus::QUOTA_EXCEEDED; }; @@ -191,7 +346,40 @@ impl Task { self.close_io_completion_handle(handle); return NtStatus::ACCESS_VIOLATION; } - NtStatus::SUCCESS + success_status + } + + pub(crate) fn sys_nt_open_io_completion( + &self, + io_completion_handle: MutPtr, + desired_access: u32, + object_attributes: Option>, + ) -> NtStatus { + if let Err(status) = + probe_guest_output_preserving_value::(io_completion_handle) + { + return status; + } + let io_completion_name = + match self.read_dispatcher_object_attributes(object_attributes, true) { + Ok((_, Some(name))) => name, + Ok((_, None)) => return NtStatus::OBJECT_NAME_INVALID, + Err(status) => return status, + }; + let port = match self + .process + .object_manager + .resolve_io_completion(&io_completion_name) + { + Ok(port) => port, + Err(status) => return status, + }; + self.publish_io_completion_handle( + io_completion_handle, + port, + IoCompletionAccess::from_desired_access(desired_access), + NtStatus::SUCCESS, + ) } pub(crate) fn sys_nt_set_io_completion( @@ -257,47 +445,45 @@ impl Task { Err(status) => return status, }; let port = entry.with_entry(IoCompletionHandleObject::port); + let already_associated = self.prepare_io_completion_wait(&port); if alertable { + // TODO(windows-apc): Interrupt alertable IOCP waits to deliver queued user APCs. litebox_util_log::debug!("Treating alertable IOCP wait as non-alertable"); } - let wait_cx = self.wait_cx().with_timeout(timeout); - match wait_cx.wait_on_events( + match self.wait_on_events( false, + timeout, Events::IN, |observer, mask| { port.register_observer(observer, mask); Ok(()) }, - || { - let packets = port.remove(count as usize); - if packets.is_empty() { - return Err(TryOpError::::TryAgain); - } - for (index, packet) in packets.iter().copied().enumerate() { - if io_completion_information - .write_at_offset( - index.try_into().expect("packet count fits in isize"), - packet, - ) - .is_none() - { - return Err(TryOpError::Other(NtStatus::ACCESS_VIOLATION)); + || match port.remove_with_capacity( + self.thread_object.thread_id(), + count as usize, + |index, packet| { + io_completion_information + .write_at_offset(index.cast_signed(), packet) + .ok_or(NtStatus::ACCESS_VIOLATION) + }, + |removed| { + num_entries_removed + .write_at_offset(0, removed) + .ok_or(NtStatus::ACCESS_VIOLATION) + }, + ) { + Ok(true) => { + if !already_associated { + self.associate_io_completion_worker(&port); } + Ok(()) } - Ok(packets - .len() - .try_into() - .expect("packet count is bounded by the u32 input count")) + Ok(false) => Err(TryOpError::TryAgain), + Err(status) => Err(TryOpError::Other(status)), }, ) { - Ok(removed) => { - if num_entries_removed.write_at_offset(0, removed).is_none() { - NtStatus::ACCESS_VIOLATION - } else { - NtStatus::SUCCESS - } - } + Ok(()) => NtStatus::SUCCESS, Err(TryOpError::WaitError(WaitError::TimedOut)) => NtStatus::TIMEOUT, Err(TryOpError::WaitError(WaitError::Interrupted)) => NtStatus::ALERTED, Err(TryOpError::TryAgain) => unreachable!("blocking wait cannot return TryAgain"), @@ -314,8 +500,11 @@ mod tests { use litebox_common_windows::nt_status::NtStatus; use super::*; - use crate::nt_types::ObjectAttributes; - use crate::tests::{const_ptr, mut_ptr, test_task}; + use crate::nt_types::{ObjectAttributes, ObjectAttributesFlags}; + use crate::syscalls::event::EventType; + use crate::tests::{ + const_ptr, mut_ptr, object_attributes, test_task, unicode_string, utf16_units, + }; const IO_COMPLETION_ALL_ACCESS: u32 = 0x001f_0003; @@ -323,6 +512,318 @@ mod tests { size_of::().trunc() } + fn packet(completion_key: usize) -> IoCompletionPacket { + IoCompletionPacket { + completion_key, + completion_value: 0, + status: 0, + information: 0, + } + } + + fn remove_one( + task: &Task, + port: &Arc>, + ) -> Result, NtStatus> { + let already_associated = task.prepare_io_completion_wait(port); + let mut packet = None; + let removed = port.remove_with_capacity( + task.thread_object.thread_id(), + 1, + |_, value| { + packet = Some(value); + Ok(()) + }, + |_| Ok(()), + )?; + if removed && !already_associated { + task.associate_io_completion_worker(port); + } + Ok(packet) + } + + #[test] + fn concurrency_limit_throttles_other_workers() { + let first = test_task(); + let second = first.clone_for_test().unwrap(); + let port = Arc::new(IoCompletionObject::new(1)); + port.post(packet(1)); + port.post(packet(2)); + + assert_eq!( + remove_one(&first, &port).unwrap().unwrap().completion_key, + 1 + ); + assert!(remove_one(&second, &port).unwrap().is_none()); + assert_eq!(port.packets.lock().len(), 1); + + assert_eq!( + remove_one(&first, &port).unwrap().unwrap().completion_key, + 2 + ); + first.release_io_completion_worker(); + second.release_io_completion_worker(); + } + + #[test] + fn blocking_worker_releases_concurrency_capacity() { + let first = test_task(); + let second = first.clone_for_test().unwrap(); + let port = Arc::new(IoCompletionObject::new(1)); + port.post(packet(1)); + port.post(packet(2)); + + assert!(remove_one(&first, &port).unwrap().is_some()); + first.suspend_io_completion_worker(); + assert_eq!( + remove_one(&second, &port).unwrap().unwrap().completion_key, + 2 + ); + + port.post(packet(3)); + assert!( + !port + .remove_with_capacity( + first.thread_object.thread_id(), + 1, + |_, _| unreachable!("a suspended worker must not dequeue at capacity"), + |_| unreachable!("a suspended worker must not dequeue at capacity"), + ) + .unwrap() + ); + assert_eq!(port.packets.lock().len(), 1); + + first.resume_io_completion_worker(); + assert_eq!(port.active_threads.lock().len(), 2); + first.release_io_completion_worker(); + second.release_io_completion_worker(); + assert!(port.active_threads.lock().is_empty()); + } + + #[test] + fn timed_out_wait_restores_worker_capacity() { + let task = test_task(); + let port = Arc::new(IoCompletionObject::new(1)); + port.post(packet(1)); + assert!(remove_one(&task, &port).unwrap().is_some()); + + let nonblocking_result: Result<(), TryOpError> = task.wait_on_events( + true, + None, + Events::IN, + |_, _| unreachable!("a nonblocking wait must not register an observer"), + || Err(TryOpError::TryAgain), + ); + assert!(matches!(nonblocking_result, Err(TryOpError::TryAgain))); + assert!( + port.active_threads + .lock() + .contains(&task.thread_object.thread_id()) + ); + assert!(!task.io_completion_worker.lock().suspended); + + let result: Result<(), TryOpError> = task.wait_on_events( + false, + Some(core::time::Duration::ZERO), + Events::IN, + |_, _| Ok::<(), NtStatus>(()), + || Err(TryOpError::TryAgain), + ); + + assert!(matches!( + result, + Err(TryOpError::WaitError(WaitError::TimedOut)) + )); + assert!( + port.active_threads + .lock() + .contains(&task.thread_object.thread_id()) + ); + assert!(!task.io_completion_worker.lock().suspended); + task.release_io_completion_worker(); + } + + #[test] + fn remove_write_failure_preserves_packets() { + let port = IoCompletionObject::::new(0); + for completion_key in [1, 2] { + port.post(IoCompletionPacket { + completion_key, + completion_value: 0, + status: 0, + information: 0, + }); + } + + assert_eq!( + port.remove_with( + 2, + |index, _| { + if index == 1 { + Err(NtStatus::ACCESS_VIOLATION) + } else { + Ok(()) + } + }, + |_| unreachable!("the count is not written after a packet write failure"), + ), + Err(NtStatus::ACCESS_VIOLATION) + ); + assert_eq!(port.packets.lock().len(), 2); + + let mut completion_keys = alloc::vec::Vec::new(); + let mut removed = 0; + assert_eq!( + port.remove_with( + 2, + |_, packet| { + completion_keys.push(packet.completion_key); + Ok(()) + }, + |count| { + removed = count; + Ok(()) + }, + ), + Ok(true) + ); + assert_eq!(completion_keys, [1, 2]); + assert_eq!(removed, 2); + assert!(port.packets.lock().is_empty()); + } + + #[test] + fn named_ports_open_and_share_packets() { + let task = test_task(); + let name_units = utf16_units("\\BaseNamedObjects\\LiteBoxIoCompletion"); + let name = unicode_string(&name_units); + let attributes = object_attributes(&name, ObjectAttributesFlags::CASE_INSENSITIVE.bits()); + let openif_attributes = ObjectAttributes { + attributes: (ObjectAttributesFlags::CASE_INSENSITIVE | ObjectAttributesFlags::OPENIF) + .bits(), + ..attributes + }; + + let mut created = Handle::default(); + assert_eq!( + task.sys_nt_create_io_completion( + mut_ptr(&mut created), + IO_COMPLETION_ALL_ACCESS, + Some(const_ptr(&attributes)), + 1, + ), + NtStatus::SUCCESS + ); + let mut collision = Handle::default(); + assert_eq!( + task.sys_nt_create_io_completion( + mut_ptr(&mut collision), + IO_COMPLETION_ALL_ACCESS, + Some(const_ptr(&attributes)), + 2, + ), + NtStatus::OBJECT_NAME_COLLISION + ); + assert_eq!( + task.sys_nt_create_io_completion( + mut_ptr(&mut collision), + IO_COMPLETION_ALL_ACCESS, + Some(const_ptr(&openif_attributes)), + 2, + ), + NtStatus::OBJECT_NAME_EXISTS + ); + + let mut opened = Handle::default(); + assert_eq!( + task.sys_nt_open_io_completion( + mut_ptr(&mut opened), + IO_COMPLETION_ALL_ACCESS, + Some(const_ptr(&attributes)), + ), + NtStatus::SUCCESS + ); + assert_eq!( + task.sys_nt_set_io_completion(created, 0x1234, 0x5678, 0, 9), + NtStatus::SUCCESS + ); + let mut packet = IoCompletionPacket { + completion_key: 0, + completion_value: 0, + status: 0, + information: 0, + }; + let mut removed = 0; + let timeout = 0; + assert_eq!( + task.sys_nt_remove_io_completion_ex( + opened, + mut_ptr(&mut packet), + 1, + mut_ptr(&mut removed), + Some(const_ptr(&timeout)), + false, + ), + NtStatus::SUCCESS + ); + assert_eq!(removed, 1); + assert_eq!(packet.completion_key, 0x1234); + assert_eq!(packet.completion_value, 0x5678); + assert_eq!(packet.information, 9); + + for handle in [created, collision, opened] { + assert_eq!(task.sys_nt_close(handle), NtStatus::SUCCESS); + } + let mut recreated = Handle::default(); + assert_eq!( + task.sys_nt_create_io_completion( + mut_ptr(&mut recreated), + IO_COMPLETION_ALL_ACCESS, + Some(const_ptr(&attributes)), + 1, + ), + NtStatus::SUCCESS + ); + } + + #[test] + fn named_port_rejects_existing_object_of_another_type() { + let task = test_task(); + let name_units = utf16_units("\\BaseNamedObjects\\LiteBoxIoCompletionTypeMismatch"); + let name = unicode_string(&name_units); + let attributes = object_attributes(&name, ObjectAttributesFlags::CASE_INSENSITIVE.bits()); + let mut event = Handle::default(); + assert_eq!( + task.sys_nt_create_event( + mut_ptr(&mut event), + IO_COMPLETION_ALL_ACCESS, + Some(const_ptr(&attributes)), + EventType::Notification as u32, + 0, + ), + NtStatus::SUCCESS + ); + + let mut io_completion = Handle::default(); + assert_eq!( + task.sys_nt_create_io_completion( + mut_ptr(&mut io_completion), + IO_COMPLETION_ALL_ACCESS, + Some(const_ptr(&attributes)), + 1, + ), + NtStatus::OBJECT_TYPE_MISMATCH + ); + assert_eq!( + task.sys_nt_open_io_completion( + mut_ptr(&mut io_completion), + IO_COMPLETION_ALL_ACCESS, + Some(const_ptr(&attributes)), + ), + NtStatus::OBJECT_TYPE_MISMATCH + ); + } + #[test] fn create_validates_object_attributes_without_clobbering_output() { let task = test_task(); diff --git a/litebox_shim_windows/src/syscalls/mod.rs b/litebox_shim_windows/src/syscalls/mod.rs index b16089a94..facd769f8 100644 --- a/litebox_shim_windows/src/syscalls/mod.rs +++ b/litebox_shim_windows/src/syscalls/mod.rs @@ -230,6 +230,30 @@ pub(crate) enum SyscallRequest { object_attributes: Option>, number_of_concurrent_threads: u32, }, + NtOpenIoCompletion { + io_completion_handle: Platform::RawMutPointer, + desired_access: u32, + object_attributes: Option>, + }, + NtSetIoCompletion { + io_completion_handle: Handle, + completion_key: usize, + completion_value: usize, + status: i32, + information: usize, + }, + NtRemoveIoCompletionEx { + io_completion_handle: Handle, + io_completion_information: Platform::RawMutPointer, + count: u32, + num_entries_removed: Platform::RawMutPointer, + timeout: Option>, + alertable: bool, + }, + NtAlpcConnectPort { + port_handle: Platform::RawMutPointer, + port_name: Platform::RawConstPointer, + }, NtConnectPort { port_handle: Platform::RawMutPointer, port_name: Platform::RawConstPointer, @@ -1015,6 +1039,30 @@ impl SyscallRequest { object_attributes:*, number_of_concurrent_threads, })), + NtSysno::NtOpenIoCompletion => Some(sys_req!(NtOpenIoCompletion { + io_completion_handle:*, + desired_access, + object_attributes:*, + })), + NtSysno::NtSetIoCompletion => Some(sys_req!(NtSetIoCompletion { + io_completion_handle: { Handle::from_raw }, + completion_key, + completion_value, + status, + information, + })), + NtSysno::NtRemoveIoCompletionEx => Some(sys_req!(NtRemoveIoCompletionEx { + io_completion_handle:{ Handle::from_raw }, + io_completion_information:*, + count, + num_entries_removed:*, + timeout:*, + alertable:{ |value: u8| value != 0 }, + })), + NtSysno::NtAlpcConnectPort => Some(sys_req!(NtAlpcConnectPort { + port_handle:*, + port_name:*, + })), NtSysno::NtConnectPort => Some(sys_req!(NtConnectPort { port_handle:*, port_name:*, @@ -1836,6 +1884,78 @@ impl> mod tests { use super::*; + #[cfg(target_arch = "x86_64")] + #[test] + fn open_io_completion_decodes_arguments() { + let registers = litebox_common_linux::PtRegs { + orig_rax: NtSysno::NtOpenIoCompletion.as_raw() as usize, + r10: 0x1000, + rdx: 0x001f_0003, + r8: 0x2000, + ..Default::default() + }; + + let Some(SyscallRequest::NtOpenIoCompletion { + io_completion_handle, + desired_access, + object_attributes, + }) = SyscallRequest::::try_from_raw(®isters) + else { + panic!("NtOpenIoCompletion should decode"); + }; + + assert_eq!(io_completion_handle.as_usize(), 0x1000); + assert_eq!(desired_access, 0x001f_0003); + assert_eq!( + object_attributes + .expect("object attributes should be present") + .as_usize(), + 0x2000 + ); + } + + #[cfg(target_arch = "x86_64")] + #[test] + fn remove_io_completion_ex_decodes_register_and_stack_arguments() { + let timeout = -10i64; + let mut stack = [0usize; 7]; + stack[5] = (&raw const timeout).addr(); + stack[6] = 1; + let registers = litebox_common_linux::PtRegs { + orig_rax: NtSysno::NtRemoveIoCompletionEx.as_raw() as usize, + r10: 0x40, + rdx: 0x1000, + r8: 3, + r9: 0x2000, + rsp: stack.as_ptr().addr(), + ..Default::default() + }; + + let Some(SyscallRequest::NtRemoveIoCompletionEx { + io_completion_handle, + io_completion_information, + count, + num_entries_removed, + timeout: decoded_timeout, + alertable, + }) = SyscallRequest::::try_from_raw(®isters) + else { + panic!("NtRemoveIoCompletionEx should decode"); + }; + + assert_eq!(io_completion_handle, Handle::from_raw(0x40)); + assert_eq!(io_completion_information.as_usize(), 0x1000); + assert_eq!(count, 3); + assert_eq!(num_entries_removed.as_usize(), 0x2000); + assert_eq!( + decoded_timeout + .expect("timeout should be present") + .as_usize(), + (&raw const timeout).addr() + ); + assert!(alertable); + } + #[test] fn handle_encodes_raw_fds_and_rejects_invalid_values() { let first_handle = Handle::from_raw_fd(0).expect("raw fd 0 should encode"); diff --git a/litebox_shim_windows/src/syscalls/object_manager.rs b/litebox_shim_windows/src/syscalls/object_manager.rs index fdc0b4628..20d7223aa 100644 --- a/litebox_shim_windows/src/syscalls/object_manager.rs +++ b/litebox_shim_windows/src/syscalls/object_manager.rs @@ -24,6 +24,7 @@ use crate::nt_types::{ }; use crate::syscalls::Handle; use crate::syscalls::event::EventObject; +use crate::syscalls::iocp::IoCompletionObject; use crate::syscalls::mutant::MutantObject; use crate::syscalls::section::{ SectionObject, WINDOWS_SESSION_SHARED_SECTION_OBJECT, WINDOWS_SHARED_SECTION_OBJECT, @@ -211,6 +212,9 @@ enum NamedObject { Event { event: Weak>, }, + IoCompletion { + io_completion: Weak>, + }, Mutant { mutant: Weak>, }, @@ -391,6 +395,7 @@ impl ObjectNode { new_directory() => NamedObject::Directory { children: BTreeMap::new() }; new_symlink(target: String) => NamedObject::Symlink { target }; new_event(event: Weak>) => NamedObject::Event { event }; + new_io_completion(io_completion: Weak>) => NamedObject::IoCompletion { io_completion }; new_mutant(mutant: Weak>) => NamedObject::Mutant { mutant }; new_semaphore(semaphore: Weak>) => NamedObject::Semaphore { semaphore }; new_section(section: Weak>) => NamedObject::Section { section }; @@ -438,6 +443,7 @@ impl ObjectNode { directory_object, ObjectLeafLookup<()>, NamedObject::Directory { .. } => ObjectLeafLookup::Live(()); pub(super) symlink_target, ObjectLeafLookup, NamedObject::Symlink { target } => ObjectLeafLookup::Live(target.clone()); event_object, ObjectLeafLookup>>, NamedObject::Event { event } => ObjectLeafLookup::from_weak(event); + io_completion_object, ObjectLeafLookup>>, NamedObject::IoCompletion { io_completion } => ObjectLeafLookup::from_weak(io_completion); mutant_object, ObjectLeafLookup>>, NamedObject::Mutant { mutant } => ObjectLeafLookup::from_weak(mutant); semaphore_object, ObjectLeafLookup>>, NamedObject::Semaphore { semaphore } => ObjectLeafLookup::from_weak(semaphore); section_object, ObjectLeafLookup>>, NamedObject::Section { section } => ObjectLeafLookup::from_weak(section); @@ -450,6 +456,9 @@ impl ObjectNode { NamedObject::Directory { .. } => Some("Directory"), NamedObject::Symlink { .. } => Some("SymbolicLink"), NamedObject::Event { event } => event.upgrade().map(|_| "Event"), + NamedObject::IoCompletion { io_completion } => { + io_completion.upgrade().map(|_| "IoCompletion") + } NamedObject::Mutant { mutant } => mutant.upgrade().map(|_| "Mutant"), NamedObject::Semaphore { semaphore } => semaphore.upgrade().map(|_| "Semaphore"), NamedObject::Section { section } => section.upgrade().map(|_| "Section"), @@ -549,6 +558,24 @@ impl ObjectManager { ) } + pub(super) fn create_io_completion( + &self, + path: &str, + io_completion: &Arc>, + on_exists: impl FnOnce(Arc>) -> NtStatus, + on_created: impl FnOnce() -> NtStatus, + ) -> NtStatus { + let io_completion = Arc::downgrade(io_completion); + self.create_child( + path, + |node| node.io_completion_object(), + |path, parent, name| ObjectNode::new_io_completion(path, parent, name, io_completion), + NtStatus::OBJECT_TYPE_MISMATCH, + on_exists, + |_| on_created(), + ) + } + pub(super) fn create_mutant( &self, path: &str, @@ -714,6 +741,13 @@ impl ObjectManager { self.resolve_object_leaf(path, false, |node| node.event_object()) } + pub(super) fn resolve_io_completion( + &self, + path: &str, + ) -> Result>, NtStatus> { + self.resolve_object_leaf(path, false, |node| node.io_completion_object()) + } + pub(super) fn resolve_mutant( &self, path: &str, diff --git a/litebox_shim_windows/src/syscalls/registry.rs b/litebox_shim_windows/src/syscalls/registry.rs index b68e8df29..2dc573083 100644 --- a/litebox_shim_windows/src/syscalls/registry.rs +++ b/litebox_shim_windows/src/syscalls/registry.rs @@ -1187,23 +1187,30 @@ impl Task { completion_filter, params.watch_tree, ); - let wait_cx = self.wait_cx(); - let wait_result = - self.global - .registry - .notification_pollee - .wait(&wait_cx, false, Events::IN, || { - if self.global.registry.notification_generation( - &path, - completion_filter, - params.watch_tree, - ) == generation - { - Err(TryOpError::::TryAgain) - } else { - Ok(()) - } - }); + let wait_result = self.wait_on_events( + false, + None, + Events::IN, + |observer, mask| { + self.global + .registry + .notification_pollee + .register_observer(observer, mask); + Ok(()) + }, + || { + if self.global.registry.notification_generation( + &path, + completion_filter, + params.watch_tree, + ) == generation + { + Err(TryOpError::::TryAgain) + } else { + Ok(()) + } + }, + ); let status = match wait_result { Ok(()) => NtStatus::NOTIFY_ENUM_DIR, Err(TryOpError::WaitError(litebox::event::wait::WaitError::Interrupted)) => { diff --git a/litebox_shim_windows/src/syscalls/thread.rs b/litebox_shim_windows/src/syscalls/thread.rs index 847382d46..72c2e00f0 100644 --- a/litebox_shim_windows/src/syscalls/thread.rs +++ b/litebox_shim_windows/src/syscalls/thread.rs @@ -413,6 +413,7 @@ impl Task { process: self.process.clone(), fs: self.fs.clone(), wait_state: crate::wait::WaitState::new(self.global.platform), + io_completion_worker: Mutex::new(super::iocp::IoCompletionWorkerState::new()), entry_point: ntdll.ldr_initialize_thunk, stack_top: environment.stack_top, context: environment.context, diff --git a/litebox_shim_windows/src/syscalls/wait.rs b/litebox_shim_windows/src/syscalls/wait.rs index c2c312413..013a77b14 100644 --- a/litebox_shim_windows/src/syscalls/wait.rs +++ b/litebox_shim_windows/src/syscalls/wait.rs @@ -121,9 +121,9 @@ impl Task { litebox_util_log::debug!("Treating alertable wait as non-alertable"); } - let wait_cx = self.wait_cx().with_timeout(timeout); - match wait_cx.wait_on_events( + match self.wait_on_events( false, + timeout, Events::IN, |observer, mask| { object.register_observer(observer, mask); diff --git a/litebox_shim_windows/src/tests.rs b/litebox_shim_windows/src/tests.rs index c1d25f4bf..6f404cf3f 100644 --- a/litebox_shim_windows/src/tests.rs +++ b/litebox_shim_windows/src/tests.rs @@ -174,6 +174,9 @@ pub(crate) fn test_task_with_nls_files(nls_files: &[(&str, &[u8])]) -> Task Task { process: self.process.clone(), fs: self.fs.clone(), wait_state: crate::wait::WaitState::new(self.global.platform), + io_completion_worker: litebox::sync::Mutex::new( + crate::syscalls::iocp::IoCompletionWorkerState::new(), + ), entry_point: 0, stack_top: 0, context: 0, diff --git a/litebox_shim_windows/src/wait.rs b/litebox_shim_windows/src/wait.rs index 3704e81b5..96f9b0775 100644 --- a/litebox_shim_windows/src/wait.rs +++ b/litebox_shim_windows/src/wait.rs @@ -4,7 +4,13 @@ //! Wait state management. //! //! Use a dedicated module to prevent code from accidentally accessing -//! `wait_state` without going through `wait_cx()`. +//! `wait_state` without going through `wait_on_events()`. + +use alloc::sync::Weak; + +use litebox::event::Events; +use litebox::event::observer::Observer; +use litebox::event::polling::TryOpError; use crate::{ShimFS, ShimPlatform, Task}; @@ -17,9 +23,31 @@ impl WaitState { } impl Task { - /// Returns a wait context to use to perform interruptible waits. - pub(crate) fn wait_cx(&self) -> litebox::event::wait::WaitContext<'_, Platform> { - self.wait_state.0.context().with_check_for_interrupt(self) + pub(crate) fn wait_on_events( + &self, + nonblock: bool, + timeout: Option, + events: Events, + register_observer: impl FnOnce(Weak>, Events) -> Result<(), E>, + mut try_op: impl FnMut() -> Result>, + ) -> Result> { + match try_op() { + Err(TryOpError::TryAgain) if !nonblock => {} + ret => return ret, + } + + // The core helper probes again before blocking, so readiness racing with this point can + // briefly release the IOCP slot without a host block. Exact accounting requires a hook + // around WaitContext's platform block; keep this approximation local to the Windows shim. + self.suspend_io_completion_worker(); + let _resume_worker = litebox::utils::defer(|| self.resume_io_completion_worker()); + let wait_context = self + .wait_state + .0 + .context() + .with_check_for_interrupt(self) + .with_timeout(timeout); + wait_context.wait_on_events(false, events, register_observer, try_op) } /// Publishes the handle used by other threads to interrupt this one. From 9931d2623cfe81f95db773d0bf4fb7e0140726fe Mon Sep 17 00:00:00 2001 From: Weiteng Chen Date: Sun, 30 Aug 2026 14:55:02 -0700 Subject: [PATCH 4/4] clean up code --- litebox_shim_windows/src/syscalls/file.rs | 4 +- litebox_shim_windows/src/syscalls/iocp.rs | 195 +--------------------- litebox_shim_windows/src/syscalls/mod.rs | 72 -------- 3 files changed, 9 insertions(+), 262 deletions(-) diff --git a/litebox_shim_windows/src/syscalls/file.rs b/litebox_shim_windows/src/syscalls/file.rs index de5c26065..ceded0262 100644 --- a/litebox_shim_windows/src/syscalls/file.rs +++ b/litebox_shim_windows/src/syscalls/file.rs @@ -981,13 +981,11 @@ impl Task { Ok(status) => status, Err(FileStatusError::PathError(PathError::NoSuchFileOrDirectory)) => { let parent = parent_directory_path(&path); - let status = if self.fs.file_status(parent).is_ok() { + return if self.fs.file_status(parent).is_ok() { NtStatus::OBJECT_NAME_NOT_FOUND } else { NtStatus::OBJECT_PATH_NOT_FOUND }; - litebox_util_log::debug!(path:% = path, status:? = status; "NtQueryAttributesFile path not found"); - return status; } Err(error) => return map_file_status_error(error), }; diff --git a/litebox_shim_windows/src/syscalls/iocp.rs b/litebox_shim_windows/src/syscalls/iocp.rs index 7f5887dcf..79379cde5 100644 --- a/litebox_shim_windows/src/syscalls/iocp.rs +++ b/litebox_shim_windows/src/syscalls/iocp.rs @@ -420,9 +420,14 @@ impl Task { if count == 0 { return NtStatus::INVALID_PARAMETER; } + let Some(buffer_length) = + (count as usize).checked_mul(core::mem::size_of::()) + else { + return NtStatus::INVALID_PARAMETER; + }; if let Err(status) = probe_guest_output_buffer::( MutPtr::::from_usize(io_completion_information.as_usize()), - (count as usize).saturating_mul(core::mem::size_of::()), + buffer_length, ) { return status; } @@ -500,11 +505,8 @@ mod tests { use litebox_common_windows::nt_status::NtStatus; use super::*; - use crate::nt_types::{ObjectAttributes, ObjectAttributesFlags}; - use crate::syscalls::event::EventType; - use crate::tests::{ - const_ptr, mut_ptr, object_attributes, test_task, unicode_string, utf16_units, - }; + use crate::nt_types::ObjectAttributes; + use crate::tests::{const_ptr, mut_ptr, test_task}; const IO_COMPLETION_ALL_ACCESS: u32 = 0x001f_0003; @@ -643,187 +645,6 @@ mod tests { task.release_io_completion_worker(); } - #[test] - fn remove_write_failure_preserves_packets() { - let port = IoCompletionObject::::new(0); - for completion_key in [1, 2] { - port.post(IoCompletionPacket { - completion_key, - completion_value: 0, - status: 0, - information: 0, - }); - } - - assert_eq!( - port.remove_with( - 2, - |index, _| { - if index == 1 { - Err(NtStatus::ACCESS_VIOLATION) - } else { - Ok(()) - } - }, - |_| unreachable!("the count is not written after a packet write failure"), - ), - Err(NtStatus::ACCESS_VIOLATION) - ); - assert_eq!(port.packets.lock().len(), 2); - - let mut completion_keys = alloc::vec::Vec::new(); - let mut removed = 0; - assert_eq!( - port.remove_with( - 2, - |_, packet| { - completion_keys.push(packet.completion_key); - Ok(()) - }, - |count| { - removed = count; - Ok(()) - }, - ), - Ok(true) - ); - assert_eq!(completion_keys, [1, 2]); - assert_eq!(removed, 2); - assert!(port.packets.lock().is_empty()); - } - - #[test] - fn named_ports_open_and_share_packets() { - let task = test_task(); - let name_units = utf16_units("\\BaseNamedObjects\\LiteBoxIoCompletion"); - let name = unicode_string(&name_units); - let attributes = object_attributes(&name, ObjectAttributesFlags::CASE_INSENSITIVE.bits()); - let openif_attributes = ObjectAttributes { - attributes: (ObjectAttributesFlags::CASE_INSENSITIVE | ObjectAttributesFlags::OPENIF) - .bits(), - ..attributes - }; - - let mut created = Handle::default(); - assert_eq!( - task.sys_nt_create_io_completion( - mut_ptr(&mut created), - IO_COMPLETION_ALL_ACCESS, - Some(const_ptr(&attributes)), - 1, - ), - NtStatus::SUCCESS - ); - let mut collision = Handle::default(); - assert_eq!( - task.sys_nt_create_io_completion( - mut_ptr(&mut collision), - IO_COMPLETION_ALL_ACCESS, - Some(const_ptr(&attributes)), - 2, - ), - NtStatus::OBJECT_NAME_COLLISION - ); - assert_eq!( - task.sys_nt_create_io_completion( - mut_ptr(&mut collision), - IO_COMPLETION_ALL_ACCESS, - Some(const_ptr(&openif_attributes)), - 2, - ), - NtStatus::OBJECT_NAME_EXISTS - ); - - let mut opened = Handle::default(); - assert_eq!( - task.sys_nt_open_io_completion( - mut_ptr(&mut opened), - IO_COMPLETION_ALL_ACCESS, - Some(const_ptr(&attributes)), - ), - NtStatus::SUCCESS - ); - assert_eq!( - task.sys_nt_set_io_completion(created, 0x1234, 0x5678, 0, 9), - NtStatus::SUCCESS - ); - let mut packet = IoCompletionPacket { - completion_key: 0, - completion_value: 0, - status: 0, - information: 0, - }; - let mut removed = 0; - let timeout = 0; - assert_eq!( - task.sys_nt_remove_io_completion_ex( - opened, - mut_ptr(&mut packet), - 1, - mut_ptr(&mut removed), - Some(const_ptr(&timeout)), - false, - ), - NtStatus::SUCCESS - ); - assert_eq!(removed, 1); - assert_eq!(packet.completion_key, 0x1234); - assert_eq!(packet.completion_value, 0x5678); - assert_eq!(packet.information, 9); - - for handle in [created, collision, opened] { - assert_eq!(task.sys_nt_close(handle), NtStatus::SUCCESS); - } - let mut recreated = Handle::default(); - assert_eq!( - task.sys_nt_create_io_completion( - mut_ptr(&mut recreated), - IO_COMPLETION_ALL_ACCESS, - Some(const_ptr(&attributes)), - 1, - ), - NtStatus::SUCCESS - ); - } - - #[test] - fn named_port_rejects_existing_object_of_another_type() { - let task = test_task(); - let name_units = utf16_units("\\BaseNamedObjects\\LiteBoxIoCompletionTypeMismatch"); - let name = unicode_string(&name_units); - let attributes = object_attributes(&name, ObjectAttributesFlags::CASE_INSENSITIVE.bits()); - let mut event = Handle::default(); - assert_eq!( - task.sys_nt_create_event( - mut_ptr(&mut event), - IO_COMPLETION_ALL_ACCESS, - Some(const_ptr(&attributes)), - EventType::Notification as u32, - 0, - ), - NtStatus::SUCCESS - ); - - let mut io_completion = Handle::default(); - assert_eq!( - task.sys_nt_create_io_completion( - mut_ptr(&mut io_completion), - IO_COMPLETION_ALL_ACCESS, - Some(const_ptr(&attributes)), - 1, - ), - NtStatus::OBJECT_TYPE_MISMATCH - ); - assert_eq!( - task.sys_nt_open_io_completion( - mut_ptr(&mut io_completion), - IO_COMPLETION_ALL_ACCESS, - Some(const_ptr(&attributes)), - ), - NtStatus::OBJECT_TYPE_MISMATCH - ); - } - #[test] fn create_validates_object_attributes_without_clobbering_output() { let task = test_task(); diff --git a/litebox_shim_windows/src/syscalls/mod.rs b/litebox_shim_windows/src/syscalls/mod.rs index facd769f8..9ab2747b5 100644 --- a/litebox_shim_windows/src/syscalls/mod.rs +++ b/litebox_shim_windows/src/syscalls/mod.rs @@ -1884,78 +1884,6 @@ impl> mod tests { use super::*; - #[cfg(target_arch = "x86_64")] - #[test] - fn open_io_completion_decodes_arguments() { - let registers = litebox_common_linux::PtRegs { - orig_rax: NtSysno::NtOpenIoCompletion.as_raw() as usize, - r10: 0x1000, - rdx: 0x001f_0003, - r8: 0x2000, - ..Default::default() - }; - - let Some(SyscallRequest::NtOpenIoCompletion { - io_completion_handle, - desired_access, - object_attributes, - }) = SyscallRequest::::try_from_raw(®isters) - else { - panic!("NtOpenIoCompletion should decode"); - }; - - assert_eq!(io_completion_handle.as_usize(), 0x1000); - assert_eq!(desired_access, 0x001f_0003); - assert_eq!( - object_attributes - .expect("object attributes should be present") - .as_usize(), - 0x2000 - ); - } - - #[cfg(target_arch = "x86_64")] - #[test] - fn remove_io_completion_ex_decodes_register_and_stack_arguments() { - let timeout = -10i64; - let mut stack = [0usize; 7]; - stack[5] = (&raw const timeout).addr(); - stack[6] = 1; - let registers = litebox_common_linux::PtRegs { - orig_rax: NtSysno::NtRemoveIoCompletionEx.as_raw() as usize, - r10: 0x40, - rdx: 0x1000, - r8: 3, - r9: 0x2000, - rsp: stack.as_ptr().addr(), - ..Default::default() - }; - - let Some(SyscallRequest::NtRemoveIoCompletionEx { - io_completion_handle, - io_completion_information, - count, - num_entries_removed, - timeout: decoded_timeout, - alertable, - }) = SyscallRequest::::try_from_raw(®isters) - else { - panic!("NtRemoveIoCompletionEx should decode"); - }; - - assert_eq!(io_completion_handle, Handle::from_raw(0x40)); - assert_eq!(io_completion_information.as_usize(), 0x1000); - assert_eq!(count, 3); - assert_eq!(num_entries_removed.as_usize(), 0x2000); - assert_eq!( - decoded_timeout - .expect("timeout should be present") - .as_usize(), - (&raw const timeout).addr() - ); - assert!(alertable); - } - #[test] fn handle_encodes_raw_fds_and_rejects_invalid_values() { let first_handle = Handle::from_raw_fd(0).expect("raw fd 0 should encode");