diff --git a/.github/workflows/test-futex.yaml b/.github/workflows/test-futex.yaml new file mode 100644 index 00000000..d8089e93 --- /dev/null +++ b/.github/workflows/test-futex.yaml @@ -0,0 +1,33 @@ +name: Test Futex + +on: [push, pull_request] + +permissions: + contents: read + +jobs: + test: + runs-on: ubuntu-26.04 + + steps: + - uses: actions/checkout@v7 + + - name: Install liburing with futex support + run: sudo apt-get update && sudo apt-get install -y liburing-dev + + - uses: ruby/setup-ruby@v1 + with: + ruby-version: ruby + bundler-cache: true + + - name: Run futex tests + timeout-minutes: 5 + env: + RUBYLIB: lib:ext + run: | + cd ext + bundle exec ruby extconf.rb + make + cd .. + bundle exec ruby -e 'require "io/event"; abort "IO::Event::Futex.wait_any is unavailable!" unless defined?(IO::Event::Futex) && IO::Event::Futex.respond_to?(:wait_any); selector = IO::Event::Selector::URing.new(Fiber.current); abort "futex_waitv is unavailable!" unless selector.respond_to?(:futex_waitv)' + bundle exec sus test/io/event/futex.rb diff --git a/ext/extconf.rb b/ext/extconf.rb index f9c43d80..2a87886b 100755 --- a/ext/extconf.rb +++ b/ext/extconf.rb @@ -31,9 +31,16 @@ have_func("rb_ext_ractor_safe") have_func("&rb_fiber_transfer") +have_io_buffer = have_header("ruby/io/buffer.h") + +if RUBY_PLATFORM.include?("linux") && have_io_buffer && have_header("linux/futex.h") && have_header("sys/syscall.h") + $srcs << "io/event/futex.c" +end if have_library("uring") and have_header("liburing.h") have_func("io_uring_prep_waitid", "liburing.h") + have_func("io_uring_prep_futex_wait", "liburing.h") + have_func("io_uring_prep_futex_waitv", "liburing.h") $srcs << "io/event/selector/uring.c" end @@ -57,8 +64,6 @@ have_func("&rb_fiber_raise") have_func("epoll_pwait2(0, 0, 0, 0, 0)", "sys/epoll.h") if enable_config("epoll_pwait2", true) -have_header("ruby/io/buffer.h") - # Feature detection for blocking operation support if have_func("rb_fiber_scheduler_blocking_operation_extract") # Feature detection for pthread support (needed for WorkerPool) diff --git a/ext/io/event/event.c b/ext/io/event/event.c index 59492544..eb49c059 100644 --- a/ext/io/event/event.c +++ b/ext/io/event/event.c @@ -3,6 +3,7 @@ #include "event.h" #include "fiber.h" +#include "futex.h" #include "selector/selector.h" void Init_IO_Event(void) @@ -15,6 +16,10 @@ void Init_IO_Event(void) Init_IO_Event_Fiber(IO_Event); + #ifdef IO_EVENT_FUTEX + Init_IO_Event_Futex(IO_Event); + #endif + #ifdef HAVE_IO_EVENT_WORKER_POOL Init_IO_Event_WorkerPool(IO_Event); #endif diff --git a/ext/io/event/futex.c b/ext/io/event/futex.c new file mode 100644 index 00000000..c0de48bd --- /dev/null +++ b/ext/io/event/futex.c @@ -0,0 +1,333 @@ +// Released under the MIT License. +// Copyright, 2026, by Samuel Williams. + +#include "futex.h" + +#ifdef IO_EVENT_FUTEX + +#include +#include +#include +#include +#include + +#include +#include +#include + +struct IO_Event_Futex { + VALUE buffer; + uint32_t *address; +}; + +static const rb_data_type_t IO_Event_Futex_Type; + +static void IO_Event_Futex_mark(void *_futex) { + struct IO_Event_Futex *futex = _futex; + rb_gc_mark_movable(futex->buffer); +} + +static void IO_Event_Futex_compact(void *_futex) { + struct IO_Event_Futex *futex = _futex; + futex->buffer = rb_gc_location(futex->buffer); +} + +static void IO_Event_Futex_free(void *_futex) { + xfree(_futex); +} + +static size_t IO_Event_Futex_size(const void *_futex) { + return sizeof(struct IO_Event_Futex); +} + +static const rb_data_type_t IO_Event_Futex_Type = { + .wrap_struct_name = "IO::Event::Futex", + .function = { + .dmark = IO_Event_Futex_mark, + .dcompact = IO_Event_Futex_compact, + .dfree = IO_Event_Futex_free, + .dsize = IO_Event_Futex_size, + }, + .flags = RUBY_TYPED_FREE_IMMEDIATELY | RUBY_TYPED_WB_PROTECTED, +}; + +static VALUE IO_Event_Futex_allocate(VALUE klass) { + struct IO_Event_Futex *futex = NULL; + VALUE instance = TypedData_Make_Struct(klass, struct IO_Event_Futex, &IO_Event_Futex_Type, futex); + futex->buffer = Qnil; + futex->address = NULL; + return instance; +} + +static ID id_offset; +static ID id_futex_wait; +static ID id_futex_waitv; + +static VALUE IO_Event_Futex_initialize(int argc, VALUE *argv, VALUE self) { + VALUE buffer, options; + rb_scan_args(argc, argv, "1:", &buffer, &options); + + VALUE offset_value = Qundef; + if (!NIL_P(options)) { + ID keys[] = {id_offset}; + VALUE values[1]; + rb_get_kwargs(options, keys, 0, 1, values); + offset_value = values[0]; + } + + size_t offset = offset_value == Qundef ? 0 : NUM2SIZET(offset_value); + void *base = NULL; + size_t size = 0; + rb_io_buffer_get_bytes_for_writing(buffer, &base, &size); + + if (offset > size || size - offset < sizeof(uint32_t)) { + rb_raise(rb_eRangeError, "Futex offset exceeds the buffer size!"); + } + + uint32_t *address = (uint32_t *)((char *)base + offset); + if ((uintptr_t)address % sizeof(uint32_t) != 0) { + rb_raise(rb_eArgError, "Futex address must be aligned to 4 bytes!"); + } + + struct IO_Event_Futex *futex = NULL; + TypedData_Get_Struct(self, struct IO_Event_Futex, &IO_Event_Futex_Type, futex); + RB_OBJ_WRITE(self, &futex->buffer, buffer); + futex->address = address; + + return self; +} + +static VALUE IO_Event_Futex_value(VALUE self) { + struct IO_Event_Futex *futex = NULL; + TypedData_Get_Struct(self, struct IO_Event_Futex, &IO_Event_Futex_Type, futex); + return UINT2NUM(__atomic_load_n(futex->address, __ATOMIC_ACQUIRE)); +} + +static VALUE IO_Event_Futex_set_value(VALUE self, VALUE value) { + struct IO_Event_Futex *futex = NULL; + TypedData_Get_Struct(self, struct IO_Event_Futex, &IO_Event_Futex_Type, futex); + uint32_t converted = NUM2UINT(value); + __atomic_store_n(futex->address, converted, __ATOMIC_RELEASE); + return value; +} + +static VALUE IO_Event_Futex_increment(int argc, VALUE *argv, VALUE self) { + VALUE amount_value; + rb_scan_args(argc, argv, "01", &amount_value); + uint32_t amount = NIL_P(amount_value) ? 1 : NUM2UINT(amount_value); + + struct IO_Event_Futex *futex = NULL; + TypedData_Get_Struct(self, struct IO_Event_Futex, &IO_Event_Futex_Type, futex); + uint32_t value = __atomic_add_fetch(futex->address, amount, __ATOMIC_ACQ_REL); + return UINT2NUM(value); +} + +static VALUE IO_Event_Futex_decrement(int argc, VALUE *argv, VALUE self) { + VALUE amount_value; + rb_scan_args(argc, argv, "01", &amount_value); + uint32_t amount = NIL_P(amount_value) ? 1 : NUM2UINT(amount_value); + + struct IO_Event_Futex *futex = NULL; + TypedData_Get_Struct(self, struct IO_Event_Futex, &IO_Event_Futex_Type, futex); + uint32_t value = __atomic_sub_fetch(futex->address, amount, __ATOMIC_ACQ_REL); + return UINT2NUM(value); +} + +static VALUE IO_Event_Futex_compare_exchange(VALUE self, VALUE expected_value, VALUE desired_value) { + uint32_t expected = NUM2UINT(expected_value); + uint32_t desired = NUM2UINT(desired_value); + + struct IO_Event_Futex *futex = NULL; + TypedData_Get_Struct(self, struct IO_Event_Futex, &IO_Event_Futex_Type, futex); + bool exchanged = __atomic_compare_exchange_n( + futex->address, + &expected, + desired, + false, + __ATOMIC_ACQ_REL, + __ATOMIC_ACQUIRE + ); + + return exchanged ? Qtrue : Qfalse; +} + +static VALUE IO_Event_Futex_wake(int argc, VALUE *argv, VALUE self) { + VALUE count_value; + rb_scan_args(argc, argv, "01", &count_value); + int count = NIL_P(count_value) ? 1 : NUM2INT(count_value); + if (count < 0) rb_raise(rb_eArgError, "Wake count must be non-negative!"); + + int result = syscall(SYS_futex, IO_Event_Futex_address(self), FUTEX_WAKE, count, NULL, NULL, 0); + if (result < 0) rb_sys_fail("IO_Event_Futex_wake:futex"); + return INT2NUM(result); +} + +static VALUE IO_Event_Futex_signal(int argc, VALUE *argv, VALUE self) { + VALUE count_value; + rb_scan_args(argc, argv, "01", &count_value); + + VALUE value = IO_Event_Futex_increment(0, NULL, self); + VALUE arguments[] = {NIL_P(count_value) ? INT2NUM(1) : count_value}; + IO_Event_Futex_wake(1, arguments, self); + return value; +} + +uint32_t *IO_Event_Futex_address(VALUE self) { + struct IO_Event_Futex *futex = NULL; + TypedData_Get_Struct(self, struct IO_Event_Futex, &IO_Event_Futex_Type, futex); + return futex->address; +} + +struct IO_Event_Futex_BlockingWait { + uint32_t *address; + uint32_t expected; + int result; + int error; +}; + +static void *IO_Event_Futex_blocking_wait_without_gvl(void *_arguments) { + struct IO_Event_Futex_BlockingWait *arguments = _arguments; + arguments->result = syscall(SYS_futex, arguments->address, FUTEX_WAIT, arguments->expected, NULL, NULL, 0); + arguments->error = arguments->result < 0 ? errno : 0; + return NULL; +} + +static VALUE IO_Event_Futex_blocking_wait(VALUE self, VALUE expected_value) { + struct IO_Event_Futex_BlockingWait arguments = { + .address = IO_Event_Futex_address(self), + .expected = NUM2UINT(expected_value), + }; + + rb_thread_call_without_gvl(IO_Event_Futex_blocking_wait_without_gvl, &arguments, RUBY_UBF_IO, 0); + + if (arguments.result == 0) { + return Qtrue; + } else if (arguments.error == EAGAIN) { + return Qfalse; + } else { + rb_syserr_fail(arguments.error, "IO_Event_Futex_blocking_wait:futex"); + } + + return Qfalse; +} + +#ifdef SYS_futex_waitv + +struct IO_Event_Futex_BlockingWaitV { + struct futex_waitv *vector; + size_t count; + int result; + int error; +}; + +static void *IO_Event_Futex_blocking_waitv_without_gvl(void *_arguments) { + struct IO_Event_Futex_BlockingWaitV *arguments = _arguments; + arguments->result = syscall(SYS_futex_waitv, arguments->vector, arguments->count, 0, NULL, CLOCK_MONOTONIC); + arguments->error = arguments->result < 0 ? errno : 0; + return NULL; +} + +static VALUE IO_Event_Futex_blocking_waitv(VALUE entries) { + entries = rb_Array(entries); + long count = RARRAY_LEN(entries); + if (count < 1 || count > FUTEX_WAITV_MAX) { + rb_raise(rb_eArgError, "Futex vector must contain between 1 and %d entries!", FUTEX_WAITV_MAX); + } + + struct futex_waitv *vector = ALLOCA_N(struct futex_waitv, count); + for (long index = 0; index < count; index++) { + VALUE entry = rb_Array(RARRAY_AREF(entries, index)); + if (RARRAY_LEN(entry) != 2) { + rb_raise(rb_eArgError, "Each futex vector entry must contain a futex and its expected value!"); + } + + VALUE futex = RARRAY_AREF(entry, 0); + vector[index].val = NUM2UINT(RARRAY_AREF(entry, 1)); + vector[index].uaddr = (uintptr_t)IO_Event_Futex_address(futex); + vector[index].flags = FUTEX_32; + vector[index].__reserved = 0; + } + + struct IO_Event_Futex_BlockingWaitV arguments = { + .vector = vector, + .count = count, + }; + + rb_thread_call_without_gvl(IO_Event_Futex_blocking_waitv_without_gvl, &arguments, RUBY_UBF_IO, 0); + RB_GC_GUARD(entries); + + if (arguments.result >= 0) { + return INT2NUM(arguments.result); + } else if (arguments.error == EAGAIN) { + return Qnil; + } else { + rb_syserr_fail(arguments.error, "IO_Event_Futex_blocking_waitv:futex_waitv"); + } + + return Qnil; +} + +#endif + +static VALUE IO_Event_Futex_wait(int argc, VALUE *argv, VALUE self) { + VALUE expected_value; + rb_scan_args(argc, argv, "01", &expected_value); + + if (argc == 0) { + expected_value = IO_Event_Futex_value(self); + } + + VALUE scheduler = rb_fiber_scheduler_current(); + if (NIL_P(scheduler)) { + return IO_Event_Futex_blocking_wait(self, expected_value); + } + + if (!rb_respond_to(scheduler, id_futex_wait)) { + rb_raise(rb_eNotImpError, "The current fiber scheduler does not support futex waits!"); + } + + return rb_funcall(scheduler, id_futex_wait, 2, self, expected_value); +} + +#ifdef SYS_futex_waitv + +static VALUE IO_Event_Futex_wait_any(VALUE klass, VALUE entries) { + (void)klass; + VALUE scheduler = rb_fiber_scheduler_current(); + if (NIL_P(scheduler)) { + return IO_Event_Futex_blocking_waitv(entries); + } + + if (!rb_respond_to(scheduler, id_futex_waitv)) { + rb_raise(rb_eNotImpError, "The current fiber scheduler does not support vector futex waits!"); + } + + return rb_funcall(scheduler, id_futex_waitv, 1, entries); +} + +#endif + +void Init_IO_Event_Futex(VALUE IO_Event) { + VALUE IO_Event_Futex = rb_define_class_under(IO_Event, "Futex", rb_cObject); + rb_define_alloc_func(IO_Event_Futex, IO_Event_Futex_allocate); + rb_define_method(IO_Event_Futex, "initialize", IO_Event_Futex_initialize, -1); + rb_define_method(IO_Event_Futex, "value", IO_Event_Futex_value, 0); + rb_define_method(IO_Event_Futex, "value=", IO_Event_Futex_set_value, 1); + rb_define_method(IO_Event_Futex, "increment", IO_Event_Futex_increment, -1); + rb_define_method(IO_Event_Futex, "decrement", IO_Event_Futex_decrement, -1); + rb_define_method(IO_Event_Futex, "compare_exchange", IO_Event_Futex_compare_exchange, 2); + rb_define_method(IO_Event_Futex, "wake", IO_Event_Futex_wake, -1); + rb_define_method(IO_Event_Futex, "signal", IO_Event_Futex_signal, -1); + rb_define_method(IO_Event_Futex, "wait", IO_Event_Futex_wait, -1); + +#ifdef SYS_futex_waitv + rb_define_const(IO_Event_Futex, "WAITV_LIMIT", INT2NUM(FUTEX_WAITV_MAX)); + rb_define_singleton_method(IO_Event_Futex, "wait_any", IO_Event_Futex_wait_any, 1); +#endif + + id_offset = rb_intern("offset"); + id_futex_wait = rb_intern("futex_wait"); + id_futex_waitv = rb_intern("futex_waitv"); +} + +#endif diff --git a/ext/io/event/futex.h b/ext/io/event/futex.h new file mode 100644 index 00000000..efd43b54 --- /dev/null +++ b/ext/io/event/futex.h @@ -0,0 +1,30 @@ +// Released under the MIT License. +// Copyright, 2026, by Samuel Williams. + +#pragma once + +#include + +#if defined(__linux__) && defined(HAVE_RUBY_IO_BUFFER_H) && defined(HAVE_LINUX_FUTEX_H) && defined(HAVE_SYS_SYSCALL_H) + +#define IO_EVENT_FUTEX + +#include +#include + +#ifndef FUTEX2_SIZE_U32 +#define FUTEX2_SIZE_U32 2 +#endif + +#ifndef FUTEX_32 +#define FUTEX_32 2 +#endif + +#ifndef FUTEX_WAITV_MAX +#define FUTEX_WAITV_MAX 128 +#endif + +uint32_t *IO_Event_Futex_address(VALUE self); +void Init_IO_Event_Futex(VALUE IO_Event); + +#endif diff --git a/ext/io/event/selector/uring.c b/ext/io/event/selector/uring.c index 9b3b959a..1e2d0f9a 100644 --- a/ext/io/event/selector/uring.c +++ b/ext/io/event/selector/uring.c @@ -3,6 +3,7 @@ #include "uring.h" #include "selector.h" +#include "../futex.h" #include "../list.h" #include "../array.h" @@ -804,6 +805,144 @@ VALUE IO_Event_Selector_URing_process_wait(VALUE self, VALUE fiber, VALUE _pid, return rb_ensure(process_wait_transfer, (VALUE)&process_wait_arguments, process_wait_ensure, (VALUE)&process_wait_arguments); } +#if defined(IO_EVENT_FUTEX) && defined(HAVE_IO_URING_PREP_FUTEX_WAIT) + +#pragma mark - Futex Wait + +struct futex_wait_arguments { + struct IO_Event_Selector_URing *selector; + struct IO_Event_Selector_URing_Waiting *waiting; +}; + +static VALUE futex_wait_ensure(VALUE _arguments) { + struct futex_wait_arguments *arguments = (struct futex_wait_arguments *)_arguments; + + if (arguments->waiting->completion) { + struct io_uring_sqe *sqe = io_get_sqe(arguments->selector); + io_uring_prep_cancel(sqe, (void *)arguments->waiting->completion, 0); + io_uring_sqe_set_data(sqe, NULL); + io_uring_submit_now(arguments->selector); + } + + IO_Event_Selector_URing_Waiting_cancel(arguments->waiting); + return Qnil; +} + +static VALUE futex_wait_transfer(VALUE _arguments) { + struct futex_wait_arguments *arguments = (struct futex_wait_arguments *)_arguments; + IO_Event_Selector_loop_yield(&arguments->selector->backend); + + int32_t result = arguments->waiting->result; + if (result == 0) { + return Qtrue; + } else if (result == -EAGAIN) { + return Qfalse; + } else if (result < 0) { + rb_syserr_fail(-result, "futex_wait_transfer:io_uring_futex_wait"); + } + + return Qfalse; +} + +static VALUE IO_Event_Selector_URing_futex_wait(VALUE self, VALUE fiber, VALUE futex, VALUE expected_value) { + struct IO_Event_Selector_URing *selector = NULL; + TypedData_Get_Struct(self, struct IO_Event_Selector_URing, &IO_Event_Selector_URing_Type, selector); + + struct IO_Event_Selector_URing_Waiting waiting = { + .fiber = fiber, + }; + RB_OBJ_WRITTEN(self, Qundef, fiber); + + struct IO_Event_Selector_URing_Completion *completion = IO_Event_Selector_URing_Completion_acquire(selector, &waiting); + struct futex_wait_arguments arguments = { + .selector = selector, + .waiting = &waiting, + }; + + struct io_uring_sqe *sqe = io_get_sqe(selector); + io_uring_prep_futex_wait( + sqe, + IO_Event_Futex_address(futex), + NUM2UINT(expected_value), + FUTEX_BITSET_MATCH_ANY, + FUTEX2_SIZE_U32, + 0 + ); + io_uring_sqe_set_data(sqe, completion); + io_uring_submit_pending(selector); + + VALUE result = rb_ensure(futex_wait_transfer, (VALUE)&arguments, futex_wait_ensure, (VALUE)&arguments); + RB_GC_GUARD(futex); + return result; +} + +#ifdef HAVE_IO_URING_PREP_FUTEX_WAITV + +static VALUE futex_waitv_transfer(VALUE _arguments) { + struct futex_wait_arguments *arguments = (struct futex_wait_arguments *)_arguments; + IO_Event_Selector_loop_yield(&arguments->selector->backend); + + int32_t result = arguments->waiting->result; + if (result >= 0) { + return INT2NUM(result); + } else if (result == -EAGAIN) { + return Qnil; + } else { + rb_syserr_fail(-result, "futex_waitv_transfer:io_uring_futex_waitv"); + } + + return Qnil; +} + +static VALUE IO_Event_Selector_URing_futex_waitv(VALUE self, VALUE fiber, VALUE entries) { + struct IO_Event_Selector_URing *selector = NULL; + TypedData_Get_Struct(self, struct IO_Event_Selector_URing, &IO_Event_Selector_URing_Type, selector); + + entries = rb_Array(entries); + long count = RARRAY_LEN(entries); + if (count < 1 || count > FUTEX_WAITV_MAX) { + rb_raise(rb_eArgError, "Futex vector must contain between 1 and %d entries!", FUTEX_WAITV_MAX); + } + + struct futex_waitv *vector = ALLOCA_N(struct futex_waitv, count); + for (long index = 0; index < count; index++) { + VALUE entry = rb_Array(RARRAY_AREF(entries, index)); + if (RARRAY_LEN(entry) != 2) { + rb_raise(rb_eArgError, "Each futex vector entry must contain a futex and its expected value!"); + } + + VALUE futex = RARRAY_AREF(entry, 0); + vector[index].val = NUM2UINT(RARRAY_AREF(entry, 1)); + vector[index].uaddr = (uintptr_t)IO_Event_Futex_address(futex); + vector[index].flags = FUTEX_32; + vector[index].__reserved = 0; + } + + struct IO_Event_Selector_URing_Waiting waiting = { + .fiber = fiber, + }; + RB_OBJ_WRITTEN(self, Qundef, fiber); + + struct IO_Event_Selector_URing_Completion *completion = IO_Event_Selector_URing_Completion_acquire(selector, &waiting); + struct futex_wait_arguments arguments = { + .selector = selector, + .waiting = &waiting, + }; + + struct io_uring_sqe *sqe = io_get_sqe(selector); + io_uring_prep_futex_waitv(sqe, vector, count, 0); + io_uring_sqe_set_data(sqe, completion); + io_uring_submit_pending(selector); + + VALUE result = rb_ensure(futex_waitv_transfer, (VALUE)&arguments, futex_wait_ensure, (VALUE)&arguments); + RB_GC_GUARD(entries); + return result; +} + +#endif + +#endif + #pragma mark - IO#wait static inline @@ -1724,6 +1863,9 @@ VALUE IO_Event_Selector_URing_wakeup(VALUE self) { #pragma mark - Native Methods +static int IO_Event_Selector_URing_futex_supported = 0; +static int IO_Event_Selector_URing_futex_waitv_supported = 0; + static int IO_Event_Selector_URing_supported_p(void) { struct io_uring ring; @@ -1754,6 +1896,17 @@ static int IO_Event_Selector_URing_supported_p(void) { return 0; } + +#if defined(IO_EVENT_FUTEX) && defined(HAVE_IO_URING_PREP_FUTEX_WAIT) + struct io_uring_probe *probe = io_uring_get_probe_ring(&ring); + if (probe) { + IO_Event_Selector_URing_futex_supported = io_uring_opcode_supported(probe, IORING_OP_FUTEX_WAIT); +#ifdef HAVE_IO_URING_PREP_FUTEX_WAITV + IO_Event_Selector_URing_futex_waitv_supported = io_uring_opcode_supported(probe, IORING_OP_FUTEX_WAITV); +#endif + io_uring_free_probe(probe); + } +#endif io_uring_queue_exit(&ring); @@ -1803,4 +1956,16 @@ void Init_IO_Event_Selector_URing(VALUE IO_Event_Selector) { rb_define_method(IO_Event_Selector_URing, "io_close", IO_Event_Selector_URing_io_close, 1); rb_define_method(IO_Event_Selector_URing, "process_wait", IO_Event_Selector_URing_process_wait, 3); + +#if defined(IO_EVENT_FUTEX) && defined(HAVE_IO_URING_PREP_FUTEX_WAIT) + if (IO_Event_Selector_URing_futex_supported) { + rb_define_method(IO_Event_Selector_URing, "futex_wait", IO_Event_Selector_URing_futex_wait, 3); + +#ifdef HAVE_IO_URING_PREP_FUTEX_WAITV + if (IO_Event_Selector_URing_futex_waitv_supported) { + rb_define_method(IO_Event_Selector_URing, "futex_waitv", IO_Event_Selector_URing_futex_waitv, 2); + } +#endif + } +#endif } diff --git a/fixtures/io/event/test_scheduler.rb b/fixtures/io/event/test_scheduler.rb index 6c71315d..e372ea80 100644 --- a/fixtures/io/event/test_scheduler.rb +++ b/fixtures/io/event/test_scheduler.rb @@ -37,6 +37,22 @@ module Forwarders def io_close(descriptor) @selector.io_close(descriptor) end + + # Wait while the futex contains the expected value. + def futex_wait(futex, expected) + @blocked += 1 + @selector.futex_wait(Fiber.current, futex, expected) + ensure + @blocked -= 1 + end + + # Wait until any futex value changes. + def futex_waitv(entries) + @blocked += 1 + @selector.futex_waitv(Fiber.current, entries) + ensure + @blocked -= 1 + end end def initialize(selector: nil, worker_pool: nil, maximum_worker_count: nil) diff --git a/lib/io/event/debug/selector.rb b/lib/io/event/debug/selector.rb index a5eca436..5bcda250 100644 --- a/lib/io/event/debug/selector.rb +++ b/lib/io/event/debug/selector.rb @@ -19,6 +19,18 @@ def io_close(descriptor) log("Closing file descriptor #{descriptor}") @selector.io_close(descriptor) end + + # Wait for a futex value to change, forwarded to the underlying selector. + def futex_wait(fiber, futex, expected) + log("Waiting for futex #{futex.inspect} with value #{expected}") + @selector.futex_wait(fiber, futex, expected) + end + + # Wait for any futex value to change, forwarded to the underlying selector. + def futex_waitv(fiber, entries) + log("Waiting for futex vector #{entries.inspect}") + @selector.futex_waitv(fiber, entries) + end end # Wrap the given selector with debugging. diff --git a/releases.md b/releases.md index aa0357e5..40ce2e68 100644 --- a/releases.md +++ b/releases.md @@ -1,5 +1,9 @@ # Releases +## Unreleased + + - Add `IO::Event::Futex` on Linux systems, including atomic value operations and blocking and scheduler-aware single and vector waits over shared memory. + ## v1.21.1 - Fix the `URing` completion free-list empty check so its sole entry can be reused instead of unnecessarily allocating a new completion. @@ -8,7 +12,6 @@ ## v1.20.0 - Add compatibility with Ruby 4.1's fiber scheduler interface version 4. Buffered IO operations now use `(offset, length)`, perform a single transfer of at most `length` bytes, return short transfers directly, and report `-EAGAIN` without waiting. Earlier Ruby versions retain the existing minimum-progress behavior. - ## v1.19.5 - Preserve the original exception or non-local control flow when `IO::Event::WorkerPool` cancellation interrupts a blocked fiber, while still cancelling and draining the in-flight blocking operation before returning control to Ruby. diff --git a/test/io/event/debug/selector.rb b/test/io/event/debug/selector.rb index 5b30badb..5e11eb4a 100644 --- a/test/io/event/debug/selector.rb +++ b/test/io/event/debug/selector.rb @@ -93,6 +93,16 @@ def io_close(descriptor) :calls_io_close end + def futex_wait(fiber, futex, expected) + @calls << [:futex_wait, fiber, futex, expected] + :calls_futex_wait + end + + def futex_waitv(fiber, entries) + @calls << [:futex_waitv, fiber, entries] + :calls_futex_waitv + end + def select(duration = nil) @calls << [:select, duration] :calls_select @@ -153,6 +163,8 @@ def select(duration = nil) end expect(selector.io_close(input.fileno)).to be == :calls_io_close expect(selector.respond_to?(:io_close)).to be == true + expect(selector.futex_wait(fiber, :futex, 1)).to be == :calls_futex_wait + expect(selector.futex_waitv(fiber, [[:futex, 1]])).to be == :calls_futex_waitv expect(selector.select(0)).to be == :calls_select ensure input&.close diff --git a/test/io/event/futex.rb b/test/io/event/futex.rb new file mode 100644 index 00000000..dfb38330 --- /dev/null +++ b/test/io/event/futex.rb @@ -0,0 +1,291 @@ +# frozen_string_literal: true + +# Released under the MIT License. +# Copyright, 2026, by Samuel Williams. + +require "io/event" +require "io/event/test_scheduler" + +return unless defined?(IO::Event::Futex) + +describe IO::Event::Futex do + let(:buffer) {IO::Buffer.new(8)} + let(:futex) {subject.new(buffer)} + let(:uring_selector) do + unless defined?(IO::Event::Selector::URing) + skip "io_uring is not available" + end + + selector = IO::Event::Selector::URing.new(Fiber.current) + unless selector.respond_to?(:futex_wait) + selector.close + skip "io_uring futex operations are not available" + end + + selector + end + let(:waitv_selector) do + selector = uring_selector + unless selector.respond_to?(:futex_waitv) + selector.close + skip "io_uring futex waitv operations are not available" + end + + selector + end + + with "#value" do + it "stores and loads the value atomically" do + futex.value = 42 + expect(futex.value).to be == 42 + end + end + + with "#increment" do + it "increments the value" do + expect(futex.increment).to be == 1 + expect(futex.increment(2)).to be == 3 + end + end + + with "#decrement" do + it "decrements the value" do + futex.value = 3 + + expect(futex.decrement).to be == 2 + expect(futex.decrement(2)).to be == 0 + end + end + + with "#compare_exchange" do + it "exchanges a matching value" do + futex.value = 2 + + expect(futex.compare_exchange(2, 1)).to be == true + expect(futex.value).to be == 1 + end + + it "does not exchange a different value" do + futex.value = 2 + + expect(futex.compare_exchange(1, 0)).to be == false + expect(futex.value).to be == 2 + end + end + + with "offset:" do + it "can address independent words in one buffer" do + first = subject.new(buffer, offset: 0) + second = subject.new(buffer, offset: 4) + + first.value = 1 + second.value = 2 + + expect(first.value).to be == 1 + expect(second.value).to be == 2 + end + + it "rejects unaligned offsets" do + expect do + subject.new(buffer, offset: 1) + end.to raise_exception(ArgumentError) + end + + it "rejects offsets outside the buffer" do + expect do + subject.new(buffer, offset: 8) + end.to raise_exception(RangeError) + end + end + + with "#wait" do + it "waits without blocking other Ruby threads when no scheduler is installed" do + thread = Thread.new do + sleep 0.01 + futex.signal + end + + expect(futex.wait(0)).to be == true + expect(futex.value).to be == 1 + ensure + thread&.join + end + + it "does not wait without a scheduler when the value has changed" do + futex.value = 1 + expect(futex.wait(0)).to be == false + end + + it "waits asynchronously for a signal" do + selector = uring_selector + result = nil + + fiber = Fiber.new do + result = selector.futex_wait(Fiber.current, futex, 0) + end + fiber.transfer + + thread = Thread.new do + sleep 0.01 + futex.signal + end + + selector.select(1) + thread.join + + expect(result).to be == true + expect(futex.value).to be == 1 + ensure + selector&.close + thread&.join + end + + it "uses the current scheduler" do + selector = uring_selector + scheduler = IO::Event::TestScheduler.new(selector: selector) + result = nil + + Fiber.set_scheduler(scheduler) + Fiber.schedule do + result = futex.wait(0) + end + + thread = Thread.new do + sleep 0.01 + futex.signal + end + + scheduler.run + + expect(result).to be == true + ensure + Fiber.set_scheduler(nil) + thread&.join + end + + it "does not wait when the value has changed" do + selector = uring_selector + futex.value = 1 + result = nil + + fiber = Fiber.new do + result = selector.futex_wait(Fiber.current, futex, 0) + end + fiber.transfer + selector.select(1) + + expect(result).to be == false + ensure + selector&.close + end + end + + if IO::Event::Futex.respond_to?(:wait_any) + with ".wait_any" do + it "exposes the maximum number of wait entries" do + expect(subject::WAITV_LIMIT).to be == 128 + end + + it "rejects more than the maximum number of wait entries" do + entries = Array.new(subject::WAITV_LIMIT + 1){[futex, 0]} + + expect do + subject.wait_any(entries) + end.to raise_exception(ArgumentError) + end + + it "waits without blocking other Ruby threads when no scheduler is installed" do + first = subject.new(buffer, offset: 0) + second = subject.new(buffer, offset: 4) + + thread = Thread.new do + sleep 0.01 + second.signal + end + + expect(subject.wait_any([[first, 0], [second, 0]])).to be == 1 + ensure + thread&.join + end + + it "does not wait without a scheduler when a value has changed" do + first = subject.new(buffer, offset: 0) + second = subject.new(buffer, offset: 4) + second.value = 1 + + expect(subject.wait_any([[first, 0], [second, 0]])).to be_nil + end + + it "waits asynchronously for any futex to be signalled" do + selector = waitv_selector + + first = subject.new(buffer, offset: 0) + second = subject.new(buffer, offset: 4) + result = nil + + fiber = Fiber.new do + result = selector.futex_waitv(Fiber.current, [[first, 0], [second, 0]]) + end + fiber.transfer + + thread = Thread.new do + sleep 0.01 + second.signal + end + + selector.select(1) + thread.join + + expect(result).to be == 1 + ensure + selector&.close + thread&.join + end + + it "uses the current scheduler" do + selector = waitv_selector + + scheduler = IO::Event::TestScheduler.new(selector: selector) + first = subject.new(buffer, offset: 0) + second = subject.new(buffer, offset: 4) + result = nil + + Fiber.set_scheduler(scheduler) + Fiber.schedule do + result = subject.wait_any([[first, 0], [second, 0]]) + end + + thread = Thread.new do + sleep 0.01 + second.signal + end + + scheduler.run + + expect(result).to be == 1 + ensure + Fiber.set_scheduler(nil) + thread&.join + end + + it "returns nil when a value has changed" do + selector = waitv_selector + + first = subject.new(buffer, offset: 0) + second = subject.new(buffer, offset: 4) + second.value = 1 + result = :waiting + + fiber = Fiber.new do + result = selector.futex_waitv(Fiber.current, [[first, 0], [second, 0]]) + end + fiber.transfer + selector.select(1) + + expect(result).to be_nil + ensure + selector&.close + end + end + end +end