Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
25 changes: 2 additions & 23 deletions ext/io/event/selector/epoll.c
Original file line number Diff line number Diff line change
Expand Up @@ -41,11 +41,6 @@ struct IO_Event_Selector_EPoll
int descriptor;
pid_t owner;

// Flag indicating whether the selector is currently blocked in a system call.
// Set to 1 when blocked in epoll_wait() without GVL, 0 otherwise.
// Used by wakeup() to determine if an interrupt signal is needed.
int blocked;

struct timespec idle_duration;

struct IO_Event_Interrupt interrupt;
Expand Down Expand Up @@ -320,8 +315,6 @@ VALUE IO_Event_Selector_EPoll_allocate(VALUE self) {
IO_Event_Selector_initialize(&selector->backend, self, Qnil);
selector->descriptor = -1;
selector->owner = 0;
selector->blocked = 0;

selector->descriptors.element_initialize = IO_Event_Selector_EPoll_Descriptor_initialize;
selector->descriptors.element_free = IO_Event_Selector_EPoll_Descriptor_free;
IO_Event_Array_initialize(&selector->descriptors, IO_EVENT_ARRAY_DEFAULT_COUNT, sizeof(struct IO_Event_Selector_EPoll_Descriptor));
Expand Down Expand Up @@ -846,14 +839,6 @@ int select_blocking_allowed(struct timespec * timespec) {
if (timeout_is_nonblocking(timespec)) {
return 0;
}

#ifndef RB_NOGVL_PENDING_INTR_FAIL
// On Rubies without `RB_NOGVL_PENDING_INTR_FAIL`, `rb_thread_call_without_gvl2` can enter an indefinite native wait even if a masked interrupt is already pending for this thread. This is the last safe point to avoid that wait: we still hold the GVL, so the pending interrupt queue cannot change concurrently before we decide whether to enter the blocking path.
if (IO_Event_Selector_pending_interrupt()) {
return 0;
}
#endif

return 1;
}

Expand Down Expand Up @@ -918,13 +903,7 @@ void * select_internal(void *_arguments) {
static
int select_internal_without_gvl(struct select_arguments *arguments) {
arguments->result = -1;
arguments->selector->blocked = 1;
#ifdef RB_NOGVL_PENDING_INTR_FAIL
rb_nogvl(select_internal, (void *)arguments, RUBY_UBF_IO, 0, RB_NOGVL_INTR_FAIL | RB_NOGVL_PENDING_INTR_FAIL);
#else
rb_thread_call_without_gvl2(select_internal, (void *)arguments, RUBY_UBF_IO, 0);
#endif
arguments->selector->blocked = 0;
IO_Event_Selector_blocking_operation(&arguments->selector->backend, select_internal, (void *)arguments, RUBY_UBF_IO, 0);

if (arguments->result == -1) {
// If Ruby skips the native callback, the result sentinel can remain `-1`; `errno` may be `0` or `EINTR` depending on the Ruby implementation. Both cases mean the blocking wait did not produce any events.
Expand Down Expand Up @@ -1089,7 +1068,7 @@ VALUE IO_Event_Selector_EPoll_wakeup(VALUE self) {
TypedData_Get_Struct(self, struct IO_Event_Selector_EPoll, &IO_Event_Selector_EPoll_Type, selector);

// If we are blocking, we can schedule a nop event to wake up the selector:
if (selector->blocked) {
if (selector->backend.blocked) {
IO_Event_Interrupt_signal(&selector->interrupt);

return Qtrue;
Expand Down
26 changes: 2 additions & 24 deletions ext/io/event/selector/kqueue.c
Original file line number Diff line number Diff line change
Expand Up @@ -50,11 +50,6 @@ struct IO_Event_Selector_KQueue
int descriptor;
pid_t owner;

// Flag indicating whether the selector is currently blocked in a system call.
// Set to 1 when blocked in kevent() without GVL, 0 otherwise.
// Used by wakeup() to determine if an interrupt signal is needed.
int blocked;

struct timespec idle_duration;

#ifdef IO_EVENT_SELECTOR_KQUEUE_USE_INTERRUPT
Expand Down Expand Up @@ -299,8 +294,6 @@ VALUE IO_Event_Selector_KQueue_allocate(VALUE self) {
IO_Event_Selector_initialize(&selector->backend, self, Qnil);
selector->descriptor = -1;
selector->owner = 0;
selector->blocked = 0;

selector->descriptors.element_initialize = IO_Event_Selector_KQueue_Descriptor_initialize;
selector->descriptors.element_free = IO_Event_Selector_KQueue_Descriptor_free;

Expand Down Expand Up @@ -850,14 +843,6 @@ int select_blocking_allowed(struct timespec * timespec) {
if (timespec && timespec->tv_sec == 0 && timespec->tv_nsec == 0) {
return 0;
}

#ifndef RB_NOGVL_PENDING_INTR_FAIL
// On Rubies without `RB_NOGVL_PENDING_INTR_FAIL`, `rb_thread_call_without_gvl2` can enter an indefinite native wait even if a masked interrupt is already pending for this thread. This is the last safe point to avoid that wait: we still hold the GVL, so the pending interrupt queue cannot change concurrently before we decide whether to enter the blocking path.
if (IO_Event_Selector_pending_interrupt()) {
return 0;
}
#endif

return 1;
}

Expand Down Expand Up @@ -886,14 +871,7 @@ void * select_internal(void *_arguments) {
static
int select_internal_without_gvl(struct select_arguments *arguments) {
arguments->result = -1;
arguments->selector->blocked = 1;

#ifdef RB_NOGVL_PENDING_INTR_FAIL
rb_nogvl(select_internal, (void *)arguments, RUBY_UBF_IO, 0, RB_NOGVL_INTR_FAIL | RB_NOGVL_PENDING_INTR_FAIL);
#else
rb_thread_call_without_gvl2(select_internal, (void *)arguments, RUBY_UBF_IO, 0);
#endif
arguments->selector->blocked = 0;
IO_Event_Selector_blocking_operation(&arguments->selector->backend, select_internal, (void *)arguments, RUBY_UBF_IO, 0);

if (arguments->result == -1) {
// If Ruby skips the native callback, the result sentinel can remain `-1`; `errno` may be `0` or `EINTR` depending on the Ruby implementation. Both cases mean the blocking wait did not produce any events.
Expand Down Expand Up @@ -1069,7 +1047,7 @@ VALUE IO_Event_Selector_KQueue_wakeup(VALUE self) {
struct IO_Event_Selector_KQueue *selector = NULL;
TypedData_Get_Struct(self, struct IO_Event_Selector_KQueue, &IO_Event_Selector_KQueue_Type, selector);

if (selector->blocked) {
if (selector->backend.blocked) {
#ifdef IO_EVENT_SELECTOR_KQUEUE_USE_INTERRUPT
IO_Event_Interrupt_signal(&selector->interrupt);
#else
Expand Down
64 changes: 62 additions & 2 deletions ext/io/event/selector/selector.c
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,59 @@

static const int DEBUG = 0;

ID IO_Event_Selector_pending_interrupt_p_id;
#ifndef RB_NOGVL_PENDING_INTR_FAIL
static ID handle_interrupt_id;
static VALUE signal_exception_never = Qnil;

struct IO_Event_Selector_blocking_operation_arguments {
void *(*function)(void *);
void *data;
rb_unblock_function_t *unblock_function;
void *unblock_data;
};

static VALUE IO_Event_Selector_blocking_operation_yield(RB_BLOCK_CALL_FUNC_ARGLIST(yielded_argument, callback_argument)) {
struct IO_Event_Selector_blocking_operation_arguments *arguments = (struct IO_Event_Selector_blocking_operation_arguments *)callback_argument;

rb_thread_call_without_gvl2(arguments->function, arguments->data, arguments->unblock_function, arguments->unblock_data);

return Qnil;
}

static VALUE IO_Event_Selector_blocking_operation_fallback(VALUE callback_argument) {
return rb_block_call(rb_cThread, handle_interrupt_id, 1, &signal_exception_never, IO_Event_Selector_blocking_operation_yield, callback_argument);
}
#endif

void IO_Event_Selector_blocking_operation(struct IO_Event_Selector *selector, void *(*function)(void *), void *data, rb_unblock_function_t *unblock_function, void *unblock_data) {
int state = 0;
selector->blocked = 1;

#ifdef RB_NOGVL_PENDING_INTR_FAIL
rb_nogvl(function, data, unblock_function, unblock_data, RB_NOGVL_INTR_FAIL | RB_NOGVL_PENDING_INTR_FAIL);
#else
// `Thread.handle_interrupt` resets `pending_interrupt_queue_checked` and
// raises the VM interrupt flag when its mask is pushed with a non-empty
// pending queue. Its C block calls `rb_thread_call_without_gvl2` directly,
// without a Ruby safepoint between refreshing the flag and entering
// `unblock_function_set`. That function either observes the interrupt or
// installs the UBF while holding Ruby's interrupt lock.
//
// Keep signal exceptions deferred here: this fallback should prevent a
// stranded native wait without changing the caller's delivery timing.
struct IO_Event_Selector_blocking_operation_arguments arguments = {
.function = function,
.data = data,
.unblock_function = unblock_function,
.unblock_data = unblock_data,
};
rb_protect(IO_Event_Selector_blocking_operation_fallback, (VALUE)&arguments, &state);
#endif

selector->blocked = 0;

if (state) rb_jump_tag(state);
}

#ifndef HAVE_RB_IO_DESCRIPTOR
static ID id_fileno;
Expand Down Expand Up @@ -89,7 +141,14 @@ VALUE IO_Event_Selector_process_wait(rb_pid_t pid, int flags) {
}

void Init_IO_Event_Selector(VALUE IO_Event_Selector) {
IO_Event_Selector_pending_interrupt_p_id = rb_intern("pending_interrupt?");
#ifndef RB_NOGVL_PENDING_INTR_FAIL
handle_interrupt_id = rb_intern("handle_interrupt");
signal_exception_never = rb_hash_new();
rb_hash_aset(signal_exception_never, rb_eSignal, ID2SYM(rb_intern("never")));
rb_funcall(signal_exception_never, rb_intern("compare_by_identity"), 0);
rb_obj_freeze(signal_exception_never);
rb_gc_register_mark_object(signal_exception_never);
#endif

rb_IO_Event_Selector = IO_Event_Selector;
rb_gc_register_mark_object(rb_IO_Event_Selector);
Expand All @@ -114,6 +173,7 @@ void IO_Event_Selector_initialize(struct IO_Event_Selector *backend, VALUE self,

backend->waiting = NULL;
backend->ready = NULL;
backend->blocked = 0;
}

VALUE IO_Event_Selector_loop_resume(struct IO_Event_Selector *backend, VALUE fiber, int argc, VALUE *argv) {
Expand Down
14 changes: 9 additions & 5 deletions ext/io/event/selector/selector.h
Original file line number Diff line number Diff line change
Expand Up @@ -40,11 +40,11 @@ static inline int IO_Event_try_again(int error) {
return error == EAGAIN || error == EWOULDBLOCK;
}

extern ID IO_Event_Selector_pending_interrupt_p_id;

static inline int IO_Event_Selector_pending_interrupt(void) {
return RTEST(rb_funcall(rb_cThread, IO_Event_Selector_pending_interrupt_p_id, 0));
}
// Enter a blocking region without the GVL. On Rubies without
// `RB_NOGVL_PENDING_INTR_FAIL`, refresh pending-interrupt state immediately
// before releasing the GVL so a signal cannot be stranded in the queue.
struct IO_Event_Selector;
void IO_Event_Selector_blocking_operation(struct IO_Event_Selector *selector, void *(*function)(void *), void *data, rb_unblock_function_t *unblock_function, void *unblock_data);

#ifdef HAVE_RB_IO_DESCRIPTOR
#define IO_Event_Selector_io_descriptor(io) rb_io_descriptor(io)
Expand Down Expand Up @@ -90,6 +90,10 @@ struct IO_Event_Selector {
VALUE self;
VALUE loop;

// Whether the selector is currently blocked in a system call without the GVL.
// Used by wakeup() to determine if an interrupt signal is needed.
int blocked;

// The ready queue is a list of fibers that are ready to be resumed from the event loop fiber.
// Append to waiting (front/head of queue).
struct IO_Event_Selector_Queue *waiting;
Expand Down
23 changes: 2 additions & 21 deletions ext/io/event/selector/uring.c
Original file line number Diff line number Diff line change
Expand Up @@ -42,10 +42,6 @@ struct IO_Event_Selector_URing
struct io_uring ring;
pid_t owner;

// Flag indicating whether the selector is currently blocked in a system call.
// Set to 1 when blocked in io_uring_wait_cqe_timeout() without GVL, 0 otherwise.
int blocked;

// Interrupt used to wake the selector from another thread without touching the ring's SQ.
// This allows IORING_SETUP_SINGLE_ISSUER: only the owner thread ever submits SQEs.
// Uses eventfd on Linux, pipe fallback elsewhere.
Expand Down Expand Up @@ -254,7 +250,6 @@ VALUE IO_Event_Selector_URing_allocate(VALUE self) {
selector->ring.ring_fd = -1;
selector->owner = 0;

selector->blocked = 0;
selector->interrupt.descriptor = -1;
selector->wakeup_registered = 0;

Expand Down Expand Up @@ -1247,14 +1242,6 @@ int select_blocking_allowed(struct __kernel_timespec *timespec) {
if (timespec && timespec->tv_sec == 0 && timespec->tv_nsec == 0) {
return 0;
}

#ifndef RB_NOGVL_PENDING_INTR_FAIL
// On Rubies without `RB_NOGVL_PENDING_INTR_FAIL`, `rb_thread_call_without_gvl2` can enter an indefinite native wait even if a masked interrupt is already pending for this thread. This is the last safe point to avoid that wait: we still hold the GVL, so the pending interrupt queue cannot change concurrently before we decide whether to enter the blocking path.
if (IO_Event_Selector_pending_interrupt()) {
return 0;
}
#endif

return 1;
}

Expand Down Expand Up @@ -1295,13 +1282,7 @@ int select_internal_without_gvl(struct select_arguments *arguments) {
io_uring_submit_flush(selector);

arguments->result = -EINTR;
selector->blocked = 1;
#ifdef RB_NOGVL_PENDING_INTR_FAIL
rb_nogvl(select_internal, (void *)arguments, RUBY_UBF_IO, 0, RB_NOGVL_INTR_FAIL | RB_NOGVL_PENDING_INTR_FAIL);
#else
rb_thread_call_without_gvl2(select_internal, (void *)arguments, RUBY_UBF_IO, 0);
#endif
selector->blocked = 0;
IO_Event_Selector_blocking_operation(&selector->backend, select_internal, (void *)arguments, RUBY_UBF_IO, 0);

if (arguments->result == -ETIME) {
return 0;
Expand Down Expand Up @@ -1455,7 +1436,7 @@ VALUE IO_Event_Selector_URing_wakeup(VALUE self) {

// Wake the selector by signalling the interrupt. This is safe from any thread
// and never touches the ring's SQ, which is required for IORING_SETUP_SINGLE_ISSUER.
if (selector->blocked) {
if (selector->backend.blocked) {
IO_Event_Interrupt_signal(&selector->interrupt);
return Qtrue;
}
Expand Down
4 changes: 4 additions & 0 deletions releases.md
Original file line number Diff line number Diff line change
@@ -1,5 +1,9 @@
# Releases

## Unreleased

- Prevent `URing`, `EPoll`, and `KQueue` from entering a native wait when a signal exception becomes pending during the GVL transition. On Ruby versions without `RB_NOGVL_PENDING_INTR_FAIL`, the fallback now refreshes pending-interrupt state immediately before releasing the GVL while preserving the caller's exception-delivery timing.

## v1.19.2

- Use `rb_process_status_for` when available to construct `URing` `process_wait` results directly from `waitid`, avoiding the extra reap syscall previously needed to build a `Process::Status`.
Expand Down
Loading