From 2fa984488a8c334c3811cb794ed239d6af4787e1 Mon Sep 17 00:00:00 2001 From: Samuel Williams Date: Fri, 21 Aug 2026 22:15:14 +1200 Subject: [PATCH 01/10] Add io_uring futex support Signed-off-by: Samuel Williams --- .github/workflows/test-futex.yaml | 38 +++++ ext/extconf.rb | 1 + ext/io/event/event.c | 2 +- ext/io/event/selector/uring.c | 253 +++++++++++++++++++++++++++- ext/io/event/selector/uring.h | 2 +- fixtures/io/event/test_scheduler.rb | 5 + lib/io/event.rb | 4 + lib/io/event/debug/selector.rb | 6 + lib/io/event/futex.rb | 30 ++++ releases.md | 5 +- test/io/event/futex.rb | 100 +++++++++++ 11 files changed, 442 insertions(+), 4 deletions(-) create mode 100644 .github/workflows/test-futex.yaml create mode 100644 lib/io/event/futex.rb create mode 100644 test/io/event/futex.rb diff --git a/.github/workflows/test-futex.yaml b/.github/workflows/test-futex.yaml new file mode 100644 index 00000000..7f0baf53 --- /dev/null +++ b/.github/workflows/test-futex.yaml @@ -0,0 +1,38 @@ +name: Test Futex + +on: [push, pull_request] + +permissions: + contents: read + +jobs: + test: + runs-on: ubuntu-24.04 + + steps: + - uses: actions/checkout@v6 + + - name: Install liburing with futex support + run: | + git clone --branch liburing-2.6 --depth 1 https://github.com/axboe/liburing.git /tmp/liburing + make -C /tmp/liburing -j2 + sudo make -C /tmp/liburing install + echo "/usr/lib" | sudo tee /etc/ld.so.conf.d/liburing.conf + sudo ldconfig + + - 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 is unavailable!" unless defined?(IO::Event::Futex)' + bundle exec sus test/io/event/futex.rb diff --git a/ext/extconf.rb b/ext/extconf.rb index f9c43d80..3a1f4ad0 100755 --- a/ext/extconf.rb +++ b/ext/extconf.rb @@ -34,6 +34,7 @@ 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") $srcs << "io/event/selector/uring.c" end diff --git a/ext/io/event/event.c b/ext/io/event/event.c index 59492544..506b9cc8 100644 --- a/ext/io/event/event.c +++ b/ext/io/event/event.c @@ -23,7 +23,7 @@ void Init_IO_Event(void) Init_IO_Event_Selector(IO_Event_Selector); #ifdef IO_EVENT_SELECTOR_URING - Init_IO_Event_Selector_URing(IO_Event_Selector); + Init_IO_Event_Selector_URing(IO_Event, IO_Event_Selector); #endif #ifdef IO_EVENT_SELECTOR_EPOLL diff --git a/ext/io/event/selector/uring.c b/ext/io/event/selector/uring.c index 9b3b959a..46deee0e 100644 --- a/ext/io/event/selector/uring.c +++ b/ext/io/event/selector/uring.c @@ -17,6 +17,15 @@ #include +#if defined(HAVE_IO_URING_PREP_FUTEX_WAIT) && defined(HAVE_RUBY_IO_BUFFER_H) +#include +#include + +#ifndef FUTEX2_SIZE_U32 +#define FUTEX2_SIZE_U32 2 +#endif +#endif + // `io_uring` support for `IORING_OP_WAITID` was introduced in Linux 6.7. When available, we use it to wait for process exit directly in the ring, instead of polling on a pidfd. #if defined(HAVE_IO_URING_PREP_WAITID) && (LINUX_VERSION_CODE >= KERNEL_VERSION(6,7,0)) #define IO_EVENT_SELECTOR_URING_USE_WAITID @@ -34,6 +43,150 @@ enum { enum {URING_ENTRIES = 64}; +#if defined(HAVE_IO_URING_PREP_FUTEX_WAIT) && defined(HAVE_RUBY_IO_BUFFER_H) + +#pragma mark - Futex Data Type + +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) { + struct IO_Event_Futex *futex = _futex; + if (!NIL_P(futex->buffer)) rb_io_buffer_try_unlock(futex->buffer); + 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 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); + if (!NIL_P(futex->buffer)) rb_io_buffer_unlock(futex->buffer); + rb_io_buffer_lock(buffer); + 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_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!"); + + struct IO_Event_Futex *futex = NULL; + TypedData_Get_Struct(self, struct IO_Event_Futex, &IO_Event_Futex_Type, futex); + int result = syscall(SYS_futex, futex->address, 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; +} + +static 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; +} + +#endif + #pragma mark - Data Type struct IO_Event_Selector_URing @@ -804,6 +957,77 @@ 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(HAVE_IO_URING_PREP_FUTEX_WAIT) && defined(HAVE_RUBY_IO_BUFFER_H) + +#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); + + return rb_ensure(futex_wait_transfer, (VALUE)&arguments, futex_wait_ensure, (VALUE)&arguments); +} + +#endif + #pragma mark - IO#wait static inline @@ -1724,6 +1948,8 @@ 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_supported_p(void) { struct io_uring ring; @@ -1754,13 +1980,21 @@ static int IO_Event_Selector_URing_supported_p(void) { return 0; } + +#if defined(HAVE_IO_URING_PREP_FUTEX_WAIT) && defined(HAVE_RUBY_IO_BUFFER_H) + 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); + io_uring_free_probe(probe); + } +#endif io_uring_queue_exit(&ring); return 1; } -void Init_IO_Event_Selector_URing(VALUE IO_Event_Selector) { +void Init_IO_Event_Selector_URing(VALUE IO_Event, VALUE IO_Event_Selector) { if (!IO_Event_Selector_URing_supported_p()) { return; } @@ -1803,4 +2037,21 @@ 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(HAVE_IO_URING_PREP_FUTEX_WAIT) && defined(HAVE_RUBY_IO_BUFFER_H) + if (IO_Event_Selector_URing_futex_supported) { + 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, "wake", IO_Event_Futex_wake, -1); + rb_define_method(IO_Event_Futex, "signal", IO_Event_Futex_signal, -1); + + id_offset = rb_intern("offset"); + + rb_define_method(IO_Event_Selector_URing, "futex_wait", IO_Event_Selector_URing_futex_wait, 3); + } +#endif } diff --git a/ext/io/event/selector/uring.h b/ext/io/event/selector/uring.h index f3ad316d..5f8f5f87 100644 --- a/ext/io/event/selector/uring.h +++ b/ext/io/event/selector/uring.h @@ -7,4 +7,4 @@ #define IO_EVENT_SELECTOR_URING -void Init_IO_Event_Selector_URing(VALUE IO_Event_Selector); +void Init_IO_Event_Selector_URing(VALUE IO_Event, VALUE IO_Event_Selector); diff --git a/fixtures/io/event/test_scheduler.rb b/fixtures/io/event/test_scheduler.rb index 6c71315d..b9ea8a40 100644 --- a/fixtures/io/event/test_scheduler.rb +++ b/fixtures/io/event/test_scheduler.rb @@ -37,6 +37,11 @@ module Forwarders def io_close(descriptor) @selector.io_close(descriptor) end + + # Wait while the futex contains the expected value. + def futex_wait(futex, expected) + @selector.futex_wait(Fiber.current, futex, expected) + end end def initialize(selector: nil, worker_pool: nil, maximum_worker_count: nil) diff --git a/lib/io/event.rb b/lib/io/event.rb index 8d953a72..42ee7e3e 100644 --- a/lib/io/event.rb +++ b/lib/io/event.rb @@ -8,3 +8,7 @@ require_relative "event/selector" require_relative "event/timers" require_relative "event/native" + +if defined?(IO::Event::Futex) + require_relative "event/futex" +end diff --git a/lib/io/event/debug/selector.rb b/lib/io/event/debug/selector.rb index a5eca436..1587b8b4 100644 --- a/lib/io/event/debug/selector.rb +++ b/lib/io/event/debug/selector.rb @@ -19,6 +19,12 @@ 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 end # Wrap the given selector with debugging. diff --git a/lib/io/event/futex.rb b/lib/io/event/futex.rb new file mode 100644 index 00000000..6f057bc9 --- /dev/null +++ b/lib/io/event/futex.rb @@ -0,0 +1,30 @@ +# frozen_string_literal: true + +# Released under the MIT License. +# Copyright, 2026, by Samuel Williams. + +class IO + module Event + # Represents a Linux futex backed by an aligned 32-bit word in an {IO::Buffer}. + # + # This class is only defined when the native io_uring selector supports + # `IORING_OP_FUTEX_WAIT`. + class Futex + # Wait while the futex contains the expected value. + # + # @asynchronous + # @parameter expected [Integer] The value that must still be present before waiting. + # @returns [Boolean] `true` when woken, or `false` when the value had already changed. + # @raises [NotImplementedError] If the current fiber scheduler does not support futex waits. + def wait(expected = value) + scheduler = Fiber.scheduler + + unless scheduler&.respond_to?(:futex_wait) + raise NotImplementedError, "The current fiber scheduler does not support futex waits!" + end + + scheduler.futex_wait(self, expected) + end + end + end +end diff --git a/releases.md b/releases.md index aa0357e5..33516bcc 100644 --- a/releases.md +++ b/releases.md @@ -1,5 +1,9 @@ # Releases +## Unreleased + + - Add `IO::Event::Futex` on Linux systems with io_uring futex support, providing an asynchronous shared-memory notification primitive. + ## 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/futex.rb b/test/io/event/futex.rb new file mode 100644 index 00000000..a1d1900a --- /dev/null +++ b/test/io/event/futex.rb @@ -0,0 +1,100 @@ +# frozen_string_literal: true + +# Released under the MIT License. +# Copyright, 2026, by Samuel Williams. + +require "io/event" + +return unless defined?(IO::Event::Futex) + +describe IO::Event::Futex do + let(:buffer) {IO::Buffer.new(8)} + let(:futex) {subject.new(buffer)} + + 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 "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 "requires scheduler support" do + expect do + futex.wait(0) + end.to raise_exception(NotImplementedError) + end + + it "waits asynchronously for a signal" do + selector = IO::Event::Selector::URing.new(Fiber.current) + 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 "does not wait when the value has changed" do + selector = IO::Event::Selector::URing.new(Fiber.current) + 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 +end From c130e79647ca34872b210777dce4c0a9b4216a61 Mon Sep 17 00:00:00 2001 From: Samuel Williams Date: Fri, 21 Aug 2026 22:17:22 +1200 Subject: [PATCH 02/10] Allow multiple futexes per buffer Signed-off-by: Samuel Williams --- ext/io/event/selector/uring.c | 6 +----- lib/io/event/futex.rb | 3 +++ 2 files changed, 4 insertions(+), 5 deletions(-) diff --git a/ext/io/event/selector/uring.c b/ext/io/event/selector/uring.c index 46deee0e..36238ee5 100644 --- a/ext/io/event/selector/uring.c +++ b/ext/io/event/selector/uring.c @@ -65,9 +65,7 @@ static void IO_Event_Futex_compact(void *_futex) { } static void IO_Event_Futex_free(void *_futex) { - struct IO_Event_Futex *futex = _futex; - if (!NIL_P(futex->buffer)) rb_io_buffer_try_unlock(futex->buffer); - xfree(futex); + xfree(_futex); } static size_t IO_Event_Futex_size(const void *_futex) { @@ -123,8 +121,6 @@ static VALUE IO_Event_Futex_initialize(int argc, VALUE *argv, VALUE self) { struct IO_Event_Futex *futex = NULL; TypedData_Get_Struct(self, struct IO_Event_Futex, &IO_Event_Futex_Type, futex); - if (!NIL_P(futex->buffer)) rb_io_buffer_unlock(futex->buffer); - rb_io_buffer_lock(buffer); RB_OBJ_WRITE(self, &futex->buffer, buffer); futex->address = address; diff --git a/lib/io/event/futex.rb b/lib/io/event/futex.rb index 6f057bc9..16565605 100644 --- a/lib/io/event/futex.rb +++ b/lib/io/event/futex.rb @@ -9,6 +9,9 @@ module Event # # This class is only defined when the native io_uring selector supports # `IORING_OP_FUTEX_WAIT`. + # + # The buffer must not be explicitly freed or resized while a futex refers + # to it. The futex retains the buffer so it cannot be garbage collected. class Futex # Wait while the futex contains the expected value. # From f7833bfd46bab03501fa19e5177ba030930ff114 Mon Sep 17 00:00:00 2001 From: Samuel Williams Date: Fri, 21 Aug 2026 22:33:32 +1200 Subject: [PATCH 03/10] Add vector and blocking futex waits --- .github/workflows/test-futex.yaml | 2 +- ext/extconf.rb | 1 + ext/io/event/selector/uring.c | 230 +++++++++++++++++++++++++++- fixtures/io/event/test_scheduler.rb | 11 ++ lib/io/event/debug/selector.rb | 6 + lib/io/event/futex.rb | 15 -- releases.md | 2 +- test/io/event/futex.rb | 147 +++++++++++++++++- 8 files changed, 388 insertions(+), 26 deletions(-) diff --git a/.github/workflows/test-futex.yaml b/.github/workflows/test-futex.yaml index 7f0baf53..fcc03a68 100644 --- a/.github/workflows/test-futex.yaml +++ b/.github/workflows/test-futex.yaml @@ -34,5 +34,5 @@ jobs: bundle exec ruby extconf.rb make cd .. - bundle exec ruby -e 'require "io/event"; abort "IO::Event::Futex is unavailable!" unless defined?(IO::Event::Futex)' + bundle exec ruby -e 'require "io/event"; abort "IO::Event::Futex is unavailable!" unless defined?(IO::Event::Futex); 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 3a1f4ad0..94b79913 100755 --- a/ext/extconf.rb +++ b/ext/extconf.rb @@ -35,6 +35,7 @@ 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 diff --git a/ext/io/event/selector/uring.c b/ext/io/event/selector/uring.c index 36238ee5..bc1371cd 100644 --- a/ext/io/event/selector/uring.c +++ b/ext/io/event/selector/uring.c @@ -17,13 +17,21 @@ #include -#if defined(HAVE_IO_URING_PREP_FUTEX_WAIT) && defined(HAVE_RUBY_IO_BUFFER_H) +#if defined(__linux__) && defined(HAVE_IO_URING_PREP_FUTEX_WAIT) && defined(HAVE_RUBY_IO_BUFFER_H) #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 #endif // `io_uring` support for `IORING_OP_WAITID` was introduced in Linux 6.7. When available, we use it to wait for process exit directly in the ring, instead of polling on a pidfd. @@ -43,7 +51,7 @@ enum { enum {URING_ENTRIES = 64}; -#if defined(HAVE_IO_URING_PREP_FUTEX_WAIT) && defined(HAVE_RUBY_IO_BUFFER_H) +#if defined(__linux__) && defined(HAVE_IO_URING_PREP_FUTEX_WAIT) && defined(HAVE_RUBY_IO_BUFFER_H) #pragma mark - Futex Data Type @@ -181,6 +189,138 @@ static uint32_t *IO_Event_Futex_address(VALUE self) { return futex->address; } +static ID id_futex_wait; +static ID id_futex_waitv; + +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 klass, VALUE entries) { + (void)klass; + 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) { + VALUE scheduler = rb_fiber_scheduler_current(); + if (NIL_P(scheduler)) { + return IO_Event_Futex_blocking_waitv(klass, 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 + #endif #pragma mark - Data Type @@ -953,7 +1093,7 @@ 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(HAVE_IO_URING_PREP_FUTEX_WAIT) && defined(HAVE_RUBY_IO_BUFFER_H) +#if defined(__linux__) && defined(HAVE_IO_URING_PREP_FUTEX_WAIT) && defined(HAVE_RUBY_IO_BUFFER_H) #pragma mark - Futex Wait @@ -1022,6 +1162,69 @@ static VALUE IO_Event_Selector_URing_futex_wait(VALUE self, VALUE fiber, VALUE f return rb_ensure(futex_wait_transfer, (VALUE)&arguments, futex_wait_ensure, (VALUE)&arguments); } +#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); + + return rb_ensure(futex_waitv_transfer, (VALUE)&arguments, futex_wait_ensure, (VALUE)&arguments); +} + +#endif + #endif #pragma mark - IO#wait @@ -1945,6 +2148,7 @@ 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; @@ -1977,10 +2181,13 @@ static int IO_Event_Selector_URing_supported_p(void) { return 0; } -#if defined(HAVE_IO_URING_PREP_FUTEX_WAIT) && defined(HAVE_RUBY_IO_BUFFER_H) +#if defined(__linux__) && defined(HAVE_IO_URING_PREP_FUTEX_WAIT) && defined(HAVE_RUBY_IO_BUFFER_H) 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 @@ -2034,7 +2241,7 @@ void Init_IO_Event_Selector_URing(VALUE IO_Event, VALUE IO_Event_Selector) { rb_define_method(IO_Event_Selector_URing, "process_wait", IO_Event_Selector_URing_process_wait, 3); -#if defined(HAVE_IO_URING_PREP_FUTEX_WAIT) && defined(HAVE_RUBY_IO_BUFFER_H) +#if defined(__linux__) && defined(HAVE_IO_URING_PREP_FUTEX_WAIT) && defined(HAVE_RUBY_IO_BUFFER_H) if (IO_Event_Selector_URing_futex_supported) { VALUE IO_Event_Futex = rb_define_class_under(IO_Event, "Futex", rb_cObject); rb_define_alloc_func(IO_Event_Futex, IO_Event_Futex_allocate); @@ -2044,10 +2251,23 @@ void Init_IO_Event_Selector_URing(VALUE IO_Event, VALUE IO_Event_Selector) { rb_define_method(IO_Event_Futex, "increment", IO_Event_Futex_increment, -1); 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_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"); 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, 3); + } +#endif } #endif } diff --git a/fixtures/io/event/test_scheduler.rb b/fixtures/io/event/test_scheduler.rb index b9ea8a40..e372ea80 100644 --- a/fixtures/io/event/test_scheduler.rb +++ b/fixtures/io/event/test_scheduler.rb @@ -40,7 +40,18 @@ def io_close(descriptor) # 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 diff --git a/lib/io/event/debug/selector.rb b/lib/io/event/debug/selector.rb index 1587b8b4..5bcda250 100644 --- a/lib/io/event/debug/selector.rb +++ b/lib/io/event/debug/selector.rb @@ -25,6 +25,12 @@ 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/lib/io/event/futex.rb b/lib/io/event/futex.rb index 16565605..dc091a19 100644 --- a/lib/io/event/futex.rb +++ b/lib/io/event/futex.rb @@ -13,21 +13,6 @@ module Event # The buffer must not be explicitly freed or resized while a futex refers # to it. The futex retains the buffer so it cannot be garbage collected. class Futex - # Wait while the futex contains the expected value. - # - # @asynchronous - # @parameter expected [Integer] The value that must still be present before waiting. - # @returns [Boolean] `true` when woken, or `false` when the value had already changed. - # @raises [NotImplementedError] If the current fiber scheduler does not support futex waits. - def wait(expected = value) - scheduler = Fiber.scheduler - - unless scheduler&.respond_to?(:futex_wait) - raise NotImplementedError, "The current fiber scheduler does not support futex waits!" - end - - scheduler.futex_wait(self, expected) - end end end end diff --git a/releases.md b/releases.md index 33516bcc..27e24393 100644 --- a/releases.md +++ b/releases.md @@ -2,7 +2,7 @@ ## Unreleased - - Add `IO::Event::Futex` on Linux systems with io_uring futex support, providing an asynchronous shared-memory notification primitive. + - Add `IO::Event::Futex` on Linux systems with io_uring futex support, including asynchronous single and vector waits over shared memory. ## v1.21.1 diff --git a/test/io/event/futex.rb b/test/io/event/futex.rb index a1d1900a..5e2ceb06 100644 --- a/test/io/event/futex.rb +++ b/test/io/event/futex.rb @@ -4,6 +4,7 @@ # Copyright, 2026, by Samuel Williams. require "io/event" +require "io/event/test_scheduler" return unless defined?(IO::Event::Futex) @@ -51,10 +52,21 @@ end with "#wait" do - it "requires scheduler support" do - expect do - futex.wait(0) - end.to raise_exception(NotImplementedError) + 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 @@ -81,6 +93,29 @@ thread&.join end + it "uses the current scheduler" do + selector = IO::Event::Selector::URing.new(Fiber.current) + 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 = IO::Event::Selector::URing.new(Fiber.current) futex.value = 1 @@ -97,4 +132,108 @@ selector&.close end end + + with ".wait_any" do + 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 = IO::Event::Selector::URing.new(Fiber.current) + unless selector.respond_to?(:futex_waitv) + skip "io_uring futex waitv operations are not available" + end + + 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 = IO::Event::Selector::URing.new(Fiber.current) + unless selector.respond_to?(:futex_waitv) + skip "io_uring futex waitv operations are not available" + end + + 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 = IO::Event::Selector::URing.new(Fiber.current) + unless selector.respond_to?(:futex_waitv) + skip "io_uring futex waitv operations are not available" + end + + 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 From ab74cc0bd476df0d5b69d593eeefe789559284ba Mon Sep 17 00:00:00 2001 From: Samuel Williams Date: Fri, 21 Aug 2026 22:35:26 +1200 Subject: [PATCH 04/10] Fix futex waitv method arity --- ext/io/event/selector/uring.c | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/ext/io/event/selector/uring.c b/ext/io/event/selector/uring.c index bc1371cd..663823eb 100644 --- a/ext/io/event/selector/uring.c +++ b/ext/io/event/selector/uring.c @@ -2265,7 +2265,7 @@ void Init_IO_Event_Selector_URing(VALUE IO_Event, VALUE IO_Event_Selector) { #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, 3); + rb_define_method(IO_Event_Selector_URing, "futex_waitv", IO_Event_Selector_URing_futex_waitv, 2); } #endif } From d3b1ffe7f401b856c718750e7b5216e5a9f0aa9c Mon Sep 17 00:00:00 2001 From: Samuel Williams Date: Fri, 21 Aug 2026 23:02:12 +1200 Subject: [PATCH 05/10] Separate the Linux futex implementation --- .github/workflows/test-futex.yaml | 2 +- ext/extconf.rb | 7 +- ext/io/event/event.c | 7 +- ext/io/event/futex.c | 301 ++++++++++++++++++++++++++++ ext/io/event/futex.h | 30 +++ ext/io/event/selector/uring.c | 316 +----------------------------- ext/io/event/selector/uring.h | 2 +- lib/io/event.rb | 4 - lib/io/event/futex.rb | 18 -- releases.md | 2 +- test/io/event/futex.rb | 209 +++++++++++--------- 11 files changed, 468 insertions(+), 430 deletions(-) create mode 100644 ext/io/event/futex.c create mode 100644 ext/io/event/futex.h delete mode 100644 lib/io/event/futex.rb diff --git a/.github/workflows/test-futex.yaml b/.github/workflows/test-futex.yaml index fcc03a68..c72d66ca 100644 --- a/.github/workflows/test-futex.yaml +++ b/.github/workflows/test-futex.yaml @@ -34,5 +34,5 @@ jobs: bundle exec ruby extconf.rb make cd .. - bundle exec ruby -e 'require "io/event"; abort "IO::Event::Futex is unavailable!" unless defined?(IO::Event::Futex); selector = IO::Event::Selector::URing.new(Fiber.current); abort "futex_waitv is unavailable!" unless selector.respond_to?(:futex_waitv)' + 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 94b79913..2a87886b 100755 --- a/ext/extconf.rb +++ b/ext/extconf.rb @@ -31,6 +31,11 @@ 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") @@ -59,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 506b9cc8..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 @@ -23,7 +28,7 @@ void Init_IO_Event(void) Init_IO_Event_Selector(IO_Event_Selector); #ifdef IO_EVENT_SELECTOR_URING - Init_IO_Event_Selector_URing(IO_Event, IO_Event_Selector); + Init_IO_Event_Selector_URing(IO_Event_Selector); #endif #ifdef IO_EVENT_SELECTOR_EPOLL diff --git a/ext/io/event/futex.c b/ext/io/event/futex.c new file mode 100644 index 00000000..3201a475 --- /dev/null +++ b/ext/io/event/futex.c @@ -0,0 +1,301 @@ +// Released under the MIT License. +// Copyright, 2026, by Samuel Williams. + +#include "futex.h" + +#ifdef IO_EVENT_FUTEX + +#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_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, "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 663823eb..1fabb813 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" @@ -17,23 +18,6 @@ #include -#if defined(__linux__) && defined(HAVE_IO_URING_PREP_FUTEX_WAIT) && defined(HAVE_RUBY_IO_BUFFER_H) -#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 -#endif - // `io_uring` support for `IORING_OP_WAITID` was introduced in Linux 6.7. When available, we use it to wait for process exit directly in the ring, instead of polling on a pidfd. #if defined(HAVE_IO_URING_PREP_WAITID) && (LINUX_VERSION_CODE >= KERNEL_VERSION(6,7,0)) #define IO_EVENT_SELECTOR_URING_USE_WAITID @@ -51,278 +35,6 @@ enum { enum {URING_ENTRIES = 64}; -#if defined(__linux__) && defined(HAVE_IO_URING_PREP_FUTEX_WAIT) && defined(HAVE_RUBY_IO_BUFFER_H) - -#pragma mark - Futex Data Type - -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 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_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!"); - - struct IO_Event_Futex *futex = NULL; - TypedData_Get_Struct(self, struct IO_Event_Futex, &IO_Event_Futex_Type, futex); - int result = syscall(SYS_futex, futex->address, 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; -} - -static 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; -} - -static ID id_futex_wait; -static ID id_futex_waitv; - -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 klass, VALUE entries) { - (void)klass; - 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) { - VALUE scheduler = rb_fiber_scheduler_current(); - if (NIL_P(scheduler)) { - return IO_Event_Futex_blocking_waitv(klass, 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 - -#endif - #pragma mark - Data Type struct IO_Event_Selector_URing @@ -1093,7 +805,7 @@ 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(__linux__) && defined(HAVE_IO_URING_PREP_FUTEX_WAIT) && defined(HAVE_RUBY_IO_BUFFER_H) +#if defined(IO_EVENT_FUTEX) && defined(HAVE_IO_URING_PREP_FUTEX_WAIT) #pragma mark - Futex Wait @@ -2181,7 +1893,7 @@ static int IO_Event_Selector_URing_supported_p(void) { return 0; } -#if defined(__linux__) && defined(HAVE_IO_URING_PREP_FUTEX_WAIT) && defined(HAVE_RUBY_IO_BUFFER_H) +#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); @@ -2197,7 +1909,7 @@ static int IO_Event_Selector_URing_supported_p(void) { return 1; } -void Init_IO_Event_Selector_URing(VALUE IO_Event, VALUE IO_Event_Selector) { +void Init_IO_Event_Selector_URing(VALUE IO_Event_Selector) { if (!IO_Event_Selector_URing_supported_p()) { return; } @@ -2241,26 +1953,8 @@ void Init_IO_Event_Selector_URing(VALUE IO_Event, VALUE IO_Event_Selector) { rb_define_method(IO_Event_Selector_URing, "process_wait", IO_Event_Selector_URing_process_wait, 3); -#if defined(__linux__) && defined(HAVE_IO_URING_PREP_FUTEX_WAIT) && defined(HAVE_RUBY_IO_BUFFER_H) +#if defined(IO_EVENT_FUTEX) && defined(HAVE_IO_URING_PREP_FUTEX_WAIT) if (IO_Event_Selector_URing_futex_supported) { - 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, "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_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"); - rb_define_method(IO_Event_Selector_URing, "futex_wait", IO_Event_Selector_URing_futex_wait, 3); #ifdef HAVE_IO_URING_PREP_FUTEX_WAITV diff --git a/ext/io/event/selector/uring.h b/ext/io/event/selector/uring.h index 5f8f5f87..f3ad316d 100644 --- a/ext/io/event/selector/uring.h +++ b/ext/io/event/selector/uring.h @@ -7,4 +7,4 @@ #define IO_EVENT_SELECTOR_URING -void Init_IO_Event_Selector_URing(VALUE IO_Event, VALUE IO_Event_Selector); +void Init_IO_Event_Selector_URing(VALUE IO_Event_Selector); diff --git a/lib/io/event.rb b/lib/io/event.rb index 42ee7e3e..8d953a72 100644 --- a/lib/io/event.rb +++ b/lib/io/event.rb @@ -8,7 +8,3 @@ require_relative "event/selector" require_relative "event/timers" require_relative "event/native" - -if defined?(IO::Event::Futex) - require_relative "event/futex" -end diff --git a/lib/io/event/futex.rb b/lib/io/event/futex.rb deleted file mode 100644 index dc091a19..00000000 --- a/lib/io/event/futex.rb +++ /dev/null @@ -1,18 +0,0 @@ -# frozen_string_literal: true - -# Released under the MIT License. -# Copyright, 2026, by Samuel Williams. - -class IO - module Event - # Represents a Linux futex backed by an aligned 32-bit word in an {IO::Buffer}. - # - # This class is only defined when the native io_uring selector supports - # `IORING_OP_FUTEX_WAIT`. - # - # The buffer must not be explicitly freed or resized while a futex refers - # to it. The futex retains the buffer so it cannot be garbage collected. - class Futex - end - end -end diff --git a/releases.md b/releases.md index 27e24393..8f5a70a0 100644 --- a/releases.md +++ b/releases.md @@ -2,7 +2,7 @@ ## Unreleased - - Add `IO::Event::Futex` on Linux systems with io_uring futex support, including asynchronous single and vector waits over shared memory. + - Add `IO::Event::Futex` on Linux systems, including blocking and scheduler-aware single and vector waits over shared memory. ## v1.21.1 diff --git a/test/io/event/futex.rb b/test/io/event/futex.rb index 5e2ceb06..d6915062 100644 --- a/test/io/event/futex.rb +++ b/test/io/event/futex.rb @@ -11,6 +11,28 @@ 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 @@ -70,7 +92,7 @@ end it "waits asynchronously for a signal" do - selector = IO::Event::Selector::URing.new(Fiber.current) + selector = uring_selector result = nil fiber = Fiber.new do @@ -94,7 +116,7 @@ end it "uses the current scheduler" do - selector = IO::Event::Selector::URing.new(Fiber.current) + selector = uring_selector scheduler = IO::Event::TestScheduler.new(selector: selector) result = nil @@ -117,7 +139,7 @@ end it "does not wait when the value has changed" do - selector = IO::Event::Selector::URing.new(Fiber.current) + selector = uring_selector futex.value = 1 result = nil @@ -133,107 +155,112 @@ end end - with ".wait_any" do - 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 + 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 - 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 = IO::Event::Selector::URing.new(Fiber.current) - unless selector.respond_to?(:futex_waitv) - skip "io_uring futex waitv operations are not available" + 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 - 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]]) + 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 - fiber.transfer - thread = Thread.new do - sleep 0.01 - second.signal + 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 - selector.select(1) - thread.join - - expect(result).to be == 1 - ensure - selector&.close - thread&.join - end - - it "uses the current scheduler" do - selector = IO::Event::Selector::URing.new(Fiber.current) - unless selector.respond_to?(:futex_waitv) - skip "io_uring futex waitv operations are not available" - end - - 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 + 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 - 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 = IO::Event::Selector::URing.new(Fiber.current) - unless selector.respond_to?(:futex_waitv) - skip "io_uring futex waitv operations are not available" + 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 - 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]]) + 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 - fiber.transfer - selector.select(1) - - expect(result).to be_nil - ensure - selector&.close end end end From eb88752f177c88b3f92884b76751860ffbbb395a Mon Sep 17 00:00:00 2001 From: Samuel Williams Date: Fri, 21 Aug 2026 23:14:43 +1200 Subject: [PATCH 06/10] Retain futex values and cover debug forwarding --- ext/io/event/selector/uring.c | 8 ++++++-- test/io/event/debug/selector.rb | 12 ++++++++++++ 2 files changed, 18 insertions(+), 2 deletions(-) diff --git a/ext/io/event/selector/uring.c b/ext/io/event/selector/uring.c index 1fabb813..1e2d0f9a 100644 --- a/ext/io/event/selector/uring.c +++ b/ext/io/event/selector/uring.c @@ -871,7 +871,9 @@ static VALUE IO_Event_Selector_URing_futex_wait(VALUE self, VALUE fiber, VALUE f io_uring_sqe_set_data(sqe, completion); io_uring_submit_pending(selector); - return rb_ensure(futex_wait_transfer, (VALUE)&arguments, futex_wait_ensure, (VALUE)&arguments); + 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 @@ -932,7 +934,9 @@ static VALUE IO_Event_Selector_URing_futex_waitv(VALUE self, VALUE fiber, VALUE io_uring_sqe_set_data(sqe, completion); io_uring_submit_pending(selector); - return rb_ensure(futex_waitv_transfer, (VALUE)&arguments, futex_wait_ensure, (VALUE)&arguments); + VALUE result = rb_ensure(futex_waitv_transfer, (VALUE)&arguments, futex_wait_ensure, (VALUE)&arguments); + RB_GC_GUARD(entries); + return result; } #endif 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 From 431ea553d683c6a6c02d13dca4990c7613563016 Mon Sep 17 00:00:00 2001 From: Samuel Williams Date: Fri, 21 Aug 2026 23:44:33 +1200 Subject: [PATCH 07/10] Use Ubuntu 26.04 for futex tests --- .github/workflows/test-futex.yaml | 9 ++------- 1 file changed, 2 insertions(+), 7 deletions(-) diff --git a/.github/workflows/test-futex.yaml b/.github/workflows/test-futex.yaml index c72d66ca..210951ee 100644 --- a/.github/workflows/test-futex.yaml +++ b/.github/workflows/test-futex.yaml @@ -7,18 +7,13 @@ permissions: jobs: test: - runs-on: ubuntu-24.04 + runs-on: ubuntu-26.04 steps: - uses: actions/checkout@v6 - name: Install liburing with futex support - run: | - git clone --branch liburing-2.6 --depth 1 https://github.com/axboe/liburing.git /tmp/liburing - make -C /tmp/liburing -j2 - sudo make -C /tmp/liburing install - echo "/usr/lib" | sudo tee /etc/ld.so.conf.d/liburing.conf - sudo ldconfig + run: sudo apt-get update && sudo apt-get install -y liburing-dev - uses: ruby/setup-ruby@v1 with: From a83b9ed28312581f48c6685116d588bd7d781b9d Mon Sep 17 00:00:00 2001 From: Samuel Williams Date: Fri, 21 Aug 2026 23:50:51 +1200 Subject: [PATCH 08/10] Add futex atomic operations --- ext/io/event/futex.c | 32 ++++++++++++++++++++++++++++++++ releases.md | 2 +- test/io/event/futex.rb | 25 +++++++++++++++++++++++++ 3 files changed, 58 insertions(+), 1 deletion(-) diff --git a/ext/io/event/futex.c b/ext/io/event/futex.c index 3201a475..c0de48bd 100644 --- a/ext/io/event/futex.c +++ b/ext/io/event/futex.c @@ -6,6 +6,7 @@ #ifdef IO_EVENT_FUTEX #include +#include #include #include #include @@ -121,6 +122,35 @@ static VALUE IO_Event_Futex_increment(int argc, VALUE *argv, VALUE self) { 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); @@ -284,6 +314,8 @@ void Init_IO_Event_Futex(VALUE IO_Event) { 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); diff --git a/releases.md b/releases.md index 8f5a70a0..40ce2e68 100644 --- a/releases.md +++ b/releases.md @@ -2,7 +2,7 @@ ## Unreleased - - Add `IO::Event::Futex` on Linux systems, including blocking and scheduler-aware single and vector waits over shared memory. + - 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 diff --git a/test/io/event/futex.rb b/test/io/event/futex.rb index d6915062..bc4260ec 100644 --- a/test/io/event/futex.rb +++ b/test/io/event/futex.rb @@ -47,6 +47,31 @@ 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 From 936d3c1199258e1a16c9502b60bde018dbd95b9d Mon Sep 17 00:00:00 2001 From: Samuel Williams Date: Sat, 22 Aug 2026 11:51:38 +1200 Subject: [PATCH 09/10] Update .github/workflows/test-futex.yaml Co-authored-by: Olle Jonsson --- .github/workflows/test-futex.yaml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/test-futex.yaml b/.github/workflows/test-futex.yaml index 210951ee..d8089e93 100644 --- a/.github/workflows/test-futex.yaml +++ b/.github/workflows/test-futex.yaml @@ -10,7 +10,7 @@ jobs: runs-on: ubuntu-26.04 steps: - - uses: actions/checkout@v6 + - uses: actions/checkout@v7 - name: Install liburing with futex support run: sudo apt-get update && sudo apt-get install -y liburing-dev From 862dd5a8dcd8112496a7c113e15875d1fd50886f Mon Sep 17 00:00:00 2001 From: Samuel Williams Date: Sun, 30 Aug 2026 12:25:10 +1200 Subject: [PATCH 10/10] Fix Futex test formatting Assisted-By: devx/065220da-7d48-4279-a513-aedd3bbfa4da --- test/io/event/futex.rb | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/test/io/event/futex.rb b/test/io/event/futex.rb index bc4260ec..dfb38330 100644 --- a/test/io/event/futex.rb +++ b/test/io/event/futex.rb @@ -47,27 +47,27 @@ 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