Skip to content

Keep SemLock OS calls behind SemHandle methods #8740

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
youknowone merged 1 commit into from
Sep 19, 2026
Merged

Keep SemLock OS calls behind SemHandle methods #8740

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
80 changes: 69 additions & 11 deletions crates/host_env/src/multiprocessing.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,7 @@ use alloc::ffi::CString;
use std::io;

#[cfg(unix)]
pub use libc::{sem_t, timespec};
use libc::sem_t;
#[cfg(unix)]
use nix::errno::Errno;

Expand Down Expand Up @@ -92,6 +92,12 @@ pub enum WaitStatus {
Error(SemError),
}

#[cfg(unix)]
#[derive(Copy, Clone, Debug)]
pub struct Deadline {
spec: libc::timespec,
}

#[cfg(windows)]
use windows_sys::Win32::{
Foundation::{
Expand Down Expand Up @@ -161,8 +167,40 @@ impl SemHandle {
}

#[inline]
pub fn as_ptr(&self) -> *mut sem_t {
self.raw
pub fn as_handle_int(&self) -> isize {
self.raw as isize
}

#[inline]
pub fn trywait(&self) -> TryAcquireStatus {
sem_trywait_status(self.raw)
}

#[inline]
pub fn post(&self) -> Result<(), SemError> {
sem_post(self.raw)
}

#[inline]
pub fn wait(&self, deadline: Option<&Deadline>) -> WaitStatus {
sem_wait_status(self.raw, deadline.map(|deadline| &deadline.spec))
}

#[cfg(not(target_vendor = "apple"))]
#[inline]
pub fn value(&self) -> Result<i32, SemError> {
// Safety: `self.raw` is a live `sem_open` handle owned by this object.
unsafe { get_semaphore_value(self.raw) }
}

#[cfg(target_vendor = "apple")]
#[inline]
pub fn poll_wait_step(
&self,
deadline: &Deadline,
delay: u64,
) -> Result<PollWaitStep, SemError> {
sem_timedwait_poll_step(self.raw, &deadline.spec, delay)
}
}

Expand All @@ -181,10 +219,30 @@ impl SemHandle {
Self { raw }
}

#[inline]
pub fn as_handle_int(&self) -> isize {
self.raw as isize
}

#[inline]
pub fn as_raw(&self) -> HANDLE {
self.raw
}

#[inline]
pub fn wait(&self, timeout_ms: u32) -> u32 {
wait_for_single_object(self.raw, timeout_ms)
}

#[inline]
pub fn release(&self) -> Result<(), u32> {
release_semaphore(self.raw)
}

#[inline]
pub fn value(&self) -> Result<i32, ()> {
get_semaphore_value(self.raw)
}
}

#[cfg(unix)]
Expand Down Expand Up @@ -340,7 +398,7 @@ pub fn sem_unlink(name: &str) -> Result<(), SemError> {
///
/// `handle` must point to a valid `sem_t` that remains alive for the duration
/// of this call and is valid to pass to `sem_getvalue`.
pub unsafe fn get_semaphore_value(handle: *mut sem_t) -> Result<i32, SemError> {
unsafe fn get_semaphore_value(handle: *mut sem_t) -> Result<i32, SemError> {
let mut sval: libc::c_int = 0;
let res = unsafe { libc::sem_getvalue(handle, &mut sval) };
if res < 0 {
Expand All @@ -352,7 +410,7 @@ pub unsafe fn get_semaphore_value(handle: *mut sem_t) -> Result<i32, SemError> {

#[cfg(unix)]
#[allow(clippy::not_unsafe_ptr_arg_deref)]
pub fn sem_trywait_status(handle: *mut sem_t) -> TryAcquireStatus {
fn sem_trywait_status(handle: *mut sem_t) -> TryAcquireStatus {
if unsafe { libc::sem_trywait(handle) } == 0 {
TryAcquireStatus::Acquired
} else {
Expand All @@ -366,7 +424,7 @@ pub fn sem_trywait_status(handle: *mut sem_t) -> TryAcquireStatus {

#[cfg(unix)]
#[allow(clippy::not_unsafe_ptr_arg_deref)]
pub fn sem_post(handle: *mut sem_t) -> Result<(), SemError> {
fn sem_post(handle: *mut sem_t) -> Result<(), SemError> {
if unsafe { libc::sem_post(handle) } < 0 {
Err(SemError::from_errno(Errno::last()))
} else {
Expand All @@ -385,7 +443,7 @@ pub fn sem_value_max() -> i32 {
}

#[cfg(unix)]
pub fn gettimeofday() -> Result<libc::timeval, SemError> {
fn gettimeofday() -> Result<libc::timeval, SemError> {
let mut tv = libc::timeval {
tv_sec: 0,
tv_usec: 0,
Expand All @@ -398,7 +456,7 @@ pub fn gettimeofday() -> Result<libc::timeval, SemError> {
}

#[cfg(unix)]
pub fn deadline_from_timeout(timeout: f64) -> Result<libc::timespec, SemError> {
pub fn deadline_from_timeout(timeout: f64) -> Result<Deadline, SemError> {
let timeout = if timeout < 0.0 { 0.0 } else { timeout };
if !timeout.is_finite() {
return Err(SemError::InvalidInput);
Expand All @@ -423,12 +481,12 @@ pub fn deadline_from_timeout(timeout: f64) -> Result<libc::timespec, SemError> {
.checked_add((deadline.tv_nsec / 1_000_000_000) as libc::time_t)
.ok_or(SemError::InvalidInput)?;
deadline.tv_nsec %= 1_000_000_000;
Ok(deadline)
Ok(Deadline { spec: deadline })
}

#[cfg(unix)]
#[allow(clippy::not_unsafe_ptr_arg_deref)]
pub fn sem_wait_status(handle: *mut sem_t, deadline: Option<&libc::timespec>) -> WaitStatus {
fn sem_wait_status(handle: *mut sem_t, deadline: Option<&libc::timespec>) -> WaitStatus {
#[cfg(not(target_vendor = "apple"))]
if let Some(deadline) = deadline {
if unsafe { libc::sem_timedwait(handle, deadline) } == 0 {
Expand Down Expand Up @@ -474,7 +532,7 @@ pub enum PollWaitStep {

#[cfg(target_vendor = "apple")]
#[allow(clippy::not_unsafe_ptr_arg_deref)]
pub fn sem_timedwait_poll_step(
fn sem_timedwait_poll_step(
handle: *mut sem_t,
deadline: &libc::timespec,
delay: u64,
Expand Down
67 changes: 24 additions & 43 deletions crates/stdlib/src/multiprocessing.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -60,7 +60,7 @@ mod _multiprocessing {
impl SemLock {
#[pygetset]
fn handle(&self) -> isize {
self.handle.as_raw() as isize
self.handle.as_handle_int()
}

#[pygetset]
Expand Down Expand Up @@ -120,7 +120,7 @@ mod _multiprocessing {
}

// Check whether we can acquire without blocking
match host_multiprocessing::wait_for_single_object(self.handle.as_raw(), 0) {
match self.handle.wait(0) {
x if x == host_multiprocessing::wait_object_0() => {
self.last_tid
.store(host_multiprocessing::current_thread_id(), Ordering::Release);
Expand Down Expand Up @@ -149,10 +149,8 @@ mod _multiprocessing {
remaining.min(poll_ms)
};

let handle = self.handle.as_raw();
let res = vm.allow_threads(|| {
host_multiprocessing::wait_for_single_object(handle, wait_ms)
});
let handle = &self.handle;
let res = vm.allow_threads(|| handle.wait(wait_ms));

match res {
x if x == host_multiprocessing::wait_object_0() => {
Expand Down Expand Up @@ -195,7 +193,7 @@ mod _multiprocessing {
}
}

if let Err(err) = host_multiprocessing::release_semaphore(self.handle.as_raw()) {
if let Err(err) = self.handle.release() {
if host_multiprocessing::is_too_many_posts(err) {
return Err(vm.new_value_error("semaphore or lock released too many times"));
}
Expand Down Expand Up @@ -269,14 +267,12 @@ mod _multiprocessing {

#[pymethod]
fn _get_value(&self, vm: &VirtualMachine) -> PyResult<i32> {
host_multiprocessing::get_semaphore_value(self.handle.as_raw())
.map_err(|_| vm.new_last_os_error())
self.handle.value().map_err(|_| vm.new_last_os_error())
}

#[pymethod]
fn _is_zero(&self, vm: &VirtualMachine) -> PyResult<bool> {
let val = host_multiprocessing::get_semaphore_value(self.handle.as_raw())
.map_err(|_| vm.new_last_os_error())?;
let val = self.handle.value().map_err(|_| vm.new_last_os_error())?;
Ok(val == 0)
}

Expand Down Expand Up @@ -369,8 +365,6 @@ mod _multiprocessing {
};
use core::sync::atomic::{AtomicI32, AtomicU64, Ordering};
use rustpython_common::lock::PyMutex;
#[cfg(target_vendor = "apple")]
use rustpython_host_env::multiprocessing::sem_t;
use rustpython_host_env::multiprocessing::{
self as host_multiprocessing, SemError, TryAcquireStatus, WaitStatus,
};
Expand All @@ -385,19 +379,16 @@ mod _multiprocessing {
}

/// macOS fallback for sem_timedwait using select + sem_trywait polling
/// Matches sem_timedwait_save in semaphore.c
#[cfg(target_vendor = "apple")]
fn sem_timedwait_polled(
sem: *mut sem_t,
deadline: &host_multiprocessing::timespec,
handle: &SemHandle,
deadline: &host_multiprocessing::Deadline,
vm: &VirtualMachine,
) -> Result<(), SemWaitError> {
let mut delay: u64 = 0;

loop {
match vm.allow_threads(|| {
host_multiprocessing::sem_timedwait_poll_step(sem, deadline, delay)
}) {
match vm.allow_threads(|| handle.poll_wait_step(deadline, delay)) {
Ok(host_multiprocessing::PollWaitStep::Acquired) => return Ok(()),
Ok(host_multiprocessing::PollWaitStep::Timeout) => {
return Err(SemWaitError::Timeout);
Expand Down Expand Up @@ -462,7 +453,7 @@ mod _multiprocessing {
impl SemLock {
#[pygetset]
fn handle(&self) -> isize {
self.handle.as_ptr() as isize
self.handle.as_handle_int()
}

#[pygetset]
Expand Down Expand Up @@ -523,7 +514,7 @@ mod _multiprocessing {

// Check whether we can acquire without releasing the GIL and blocking
let try_status = loop {
match host_multiprocessing::sem_trywait_status(self.handle.as_ptr()) {
match self.handle.trywait() {
TryAcquireStatus::Interrupted => {
vm.check_signals()?;
}
Expand All @@ -539,11 +530,8 @@ mod _multiprocessing {
#[cfg(not(target_vendor = "apple"))]
{
loop {
let sem_ptr = self.handle.as_ptr();
// Py_BEGIN_ALLOW_THREADS / Py_END_ALLOW_THREADS
match vm.allow_threads(|| {
host_multiprocessing::sem_wait_status(sem_ptr, deadline.as_ref())
}) {
let handle = &self.handle;
match vm.allow_threads(|| handle.wait(deadline.as_ref())) {
WaitStatus::Acquired => break,
WaitStatus::Interrupted => {
vm.check_signals()?;
Expand All @@ -558,7 +546,7 @@ mod _multiprocessing {
{
// macOS: use polled fallback since sem_timedwait is not available
if let Some(ref dl) = deadline {
match sem_timedwait_polled(self.handle.as_ptr(), dl, vm) {
match sem_timedwait_polled(&self.handle, dl, vm) {
Ok(()) => {}
Err(SemWaitError::Timeout) => {
return Ok(false);
Expand All @@ -573,10 +561,8 @@ mod _multiprocessing {
} else {
// No timeout: use sem_wait (available on macOS)
loop {
let sem_ptr = self.handle.as_ptr();
match vm.allow_threads(|| {
host_multiprocessing::sem_wait_status(sem_ptr, None)
}) {
let handle = &self.handle;
match vm.allow_threads(|| handle.wait(None)) {
WaitStatus::Acquired => break,
WaitStatus::Interrupted => {
vm.check_signals()?;
Expand Down Expand Up @@ -634,9 +620,7 @@ mod _multiprocessing {
#[cfg(not(target_vendor = "apple"))]
{
// Linux: use sem_getvalue
let sval =
unsafe { host_multiprocessing::get_semaphore_value(self.handle.as_ptr()) }
.map_err(|err| os_error(vm, err))?;
let sval = self.handle.value().map_err(|err| os_error(vm, err))?;
if sval >= self.maxvalue {
return Err(vm.new_value_error("semaphore or lock released too many times"));
}
Expand All @@ -647,12 +631,10 @@ mod _multiprocessing {
// We will only check properly the maxvalue == 1 case
if self.maxvalue == 1 {
// make sure that already locked
match host_multiprocessing::sem_trywait_status(self.handle.as_ptr()) {
match self.handle.trywait() {
TryAcquireStatus::WouldBlock => {}
TryAcquireStatus::Acquired => {
if let Err(err) =
host_multiprocessing::sem_post(self.handle.as_ptr())
{
if let Err(err) = self.handle.post() {
return Err(os_error(vm, err));
}
return Err(
Expand All @@ -668,7 +650,7 @@ mod _multiprocessing {
}
}

if let Err(err) = host_multiprocessing::sem_post(self.handle.as_ptr()) {
if let Err(err) = self.handle.post() {
return Err(os_error(vm, err));
}

Expand Down Expand Up @@ -764,8 +746,7 @@ mod _multiprocessing {
#[cfg(not(target_vendor = "apple"))]
{
// Linux: use sem_getvalue
unsafe { host_multiprocessing::get_semaphore_value(self.handle.as_ptr()) }
.map_err(|err| os_error(vm, err))
self.handle.value().map_err(|err| os_error(vm, err))
}
#[cfg(target_vendor = "apple")]
{
Expand All @@ -786,7 +767,7 @@ mod _multiprocessing {
{
// macOS: HAVE_BROKEN_SEM_GETVALUE
// Try to acquire - if EAGAIN, value is 0
match host_multiprocessing::sem_trywait_status(self.handle.as_ptr()) {
match self.handle.trywait() {
TryAcquireStatus::WouldBlock => return Ok(true),
TryAcquireStatus::Interrupted => {
return Err(os_error(vm, SemError::Interrupted));
Expand All @@ -795,7 +776,7 @@ mod _multiprocessing {
TryAcquireStatus::Acquired => {}
}
// Successfully acquired - undo and return false
if let Err(err) = host_multiprocessing::sem_post(self.handle.as_ptr()) {
if let Err(err) = self.handle.post() {
return Err(os_error(vm, err));
}
Ok(false)
Expand Down
Loading