diff --git a/.github/workflows/cygwin.yml b/.github/workflows/cygwin.yml index aeb98b95eca222..fa00cba8c58104 100644 --- a/.github/workflows/cygwin.yml +++ b/.github/workflows/cygwin.yml @@ -58,7 +58,8 @@ jobs: - name: configure run: | - ./autogen.sh + # The autoreconf wrapper in autoconf-20260320-1 has a shell syntax error + AUTORECONF=autoreconf-2.73 ./autogen.sh ./configure --disable-install-doc shell: C:\cygwin\bin\bash.EXE --noprofile --norc -e -o igncr -o pipefail {0} diff --git a/array.c b/array.c index abaf5184e92cad..806ce4d231d7c6 100644 --- a/array.c +++ b/array.c @@ -5725,52 +5725,20 @@ rb_ary_cmp(VALUE ary1, VALUE ary2) return INT2FIX(-1); } -static VALUE -ary_add_hash(VALUE hash, VALUE ary) -{ - long i; - - for (i=0; i SMALL_ARRAY_LEN && RARRAY_LEN(argv[i]) > SMALL_ARRAY_LEN); - if (is_hash[i]) argv[i] = ary_make_hash(argv[i]); + is_set[i] = (length > SMALL_ARRAY_LEN && RARRAY_LEN(argv[i]) > SMALL_ARRAY_LEN); + if (is_set[i]) { + argv[i] = ary_to_set(argv[i]); + } } - for (i = 0; i < RARRAY_LEN(ary); i++) { + for (long i = 0; i < RARRAY_LEN(ary); i++) { int j; VALUE elt = rb_ary_elt(ary, i); for (j = 0; j < argc; j++) { - if (is_hash[j]) { - if (rb_hash_stlike_lookup(argv[j], elt, NULL)) + if (is_set[j]) { + if (rb_set_lookup(argv[j], elt)) break; } else { @@ -5903,17 +5867,13 @@ rb_ary_difference_multi(int argc, VALUE *argv, VALUE ary) static VALUE rb_ary_and(VALUE ary1, VALUE ary2) { - VALUE hash, ary3, v; - st_data_t vv; - long i; - ary2 = to_ary(ary2); - ary3 = rb_ary_new(); + VALUE ary3 = rb_ary_new(); if (RARRAY_LEN(ary1) == 0 || RARRAY_LEN(ary2) == 0) return ary3; if (RARRAY_LEN(ary1) <= SMALL_ARRAY_LEN && RARRAY_LEN(ary2) <= SMALL_ARRAY_LEN) { - for (i=0; i new_array @@ -6017,8 +5956,6 @@ rb_ary_union_hash(VALUE hash, VALUE ary2) static VALUE rb_ary_or(VALUE ary1, VALUE ary2) { - VALUE hash; - ary2 = to_ary(ary2); if (RARRAY_LEN(ary1) + RARRAY_LEN(ary2) <= SMALL_ARRAY_LEN) { VALUE ary3 = rb_ary_new(); @@ -6027,10 +5964,11 @@ rb_ary_or(VALUE ary1, VALUE ary2) return ary3; } - hash = ary_make_hash(ary1); - rb_ary_union_hash(hash, ary2); + VALUE set = rb_obj_hide(rb_set_new_capa(RARRAY_LEN(ary1) + RARRAY_LEN(ary2))); + rb_ary_union_set(set, ary1); + rb_ary_union_set(set, ary2); - return rb_hash_values(hash); + return rb_set_to_a(set); } /* @@ -6059,12 +5997,8 @@ rb_ary_or(VALUE ary1, VALUE ary2) static VALUE rb_ary_union_multi(int argc, VALUE *argv, VALUE ary) { - int i; - long sum; - VALUE hash; - - sum = RARRAY_LEN(ary); - for (i = 0; i < argc; i++) { + long sum = RARRAY_LEN(ary); + for (int i = 0; i < argc; i++) { argv[i] = to_ary(argv[i]); sum += RARRAY_LEN(argv[i]); } @@ -6073,15 +6007,16 @@ rb_ary_union_multi(int argc, VALUE *argv, VALUE ary) VALUE ary_union = rb_ary_new(); rb_ary_union(ary_union, ary); - for (i = 0; i < argc; i++) rb_ary_union(ary_union, argv[i]); + for (int i = 0; i < argc; i++) rb_ary_union(ary_union, argv[i]); return ary_union; } - hash = ary_make_hash(ary); - for (i = 0; i < argc; i++) rb_ary_union_hash(hash, argv[i]); + VALUE set = rb_obj_hide(rb_set_new_capa(sum)); + rb_ary_union_set(set, ary); + for (int i = 0; i < argc; i++) rb_ary_union_set(set, argv[i]); - return rb_hash_values(hash); + return rb_set_to_a(set); } /* @@ -6101,35 +6036,30 @@ rb_ary_union_multi(int argc, VALUE *argv, VALUE ary) static VALUE rb_ary_intersect_p(VALUE ary1, VALUE ary2) { - VALUE hash, v, result, shorter, longer; - st_data_t vv; - long i; - ary2 = to_ary(ary2); if (RARRAY_LEN(ary1) == 0 || RARRAY_LEN(ary2) == 0) return Qfalse; if (RARRAY_LEN(ary1) <= SMALL_ARRAY_LEN && RARRAY_LEN(ary2) <= SMALL_ARRAY_LEN) { - for (i=0; i RARRAY_LEN(ary2)) { longer = ary1; shorter = ary2; } - hash = ary_make_hash(shorter); - result = Qfalse; + VALUE set = ary_to_set(shorter); + VALUE result = Qfalse; - for (i=0; i>IO: thread_io_close_notify_all Note over ThreadB: rb_mutex_sleep - IO->>Scheduler: rb_fiber_scheduler_fiber_interrupt(Fiber1) - Scheduler->>Fiber1: fiber_interrupt with IOError + IO->>Scheduler: rb_fiber_scheduler_fiber_interrupt(Target1) + Scheduler->>Target1: raise IOError if alive? activate Fiber1 Note over IO: fiber_interrupt causes removal from blocking list Fiber1->>IO: rb_io_blocking_operation_exit() IO-->>ThreadB: Wakeup thread deactivate Fiber1 - IO->>Scheduler: rb_fiber_scheduler_fiber_interrupt(Fiber2) - Scheduler->>Fiber2: fiber_interrupt with IOError + IO->>Scheduler: rb_fiber_scheduler_fiber_interrupt(Target2) + Scheduler->>Target2: raise IOError if alive? activate Fiber2 Note over IO: fiber_interrupt causes removal from blocking list Fiber2->>IO: rb_io_blocking_operation_exit() diff --git a/enum.c b/enum.c index a2941dd7dd5b38..8e6c40639a10e5 100644 --- a/enum.c +++ b/enum.c @@ -20,6 +20,7 @@ #include "internal/proc.h" #include "internal/rational.h" #include "internal/re.h" +#include "internal/set.h" #include "ruby/util.h" #include "ruby_assert.h" #include "symbol.h" @@ -4870,18 +4871,26 @@ enum_sum(int argc, VALUE* argv, VALUE obj) } static VALUE -uniq_func(RB_BLOCK_CALL_FUNC_ARGLIST(i, hash)) +uniq_func(RB_BLOCK_CALL_FUNC_ARGLIST(i, set)) { ENUM_WANT_SVALUE(); - rb_hash_add_new_element(hash, i, i); + rb_set_add_no_check(set, i); return Qnil; } +struct uniq_iter_memo { + VALUE set; + VALUE ary; +}; + static VALUE -uniq_iter(RB_BLOCK_CALL_FUNC_ARGLIST(i, hash)) +uniq_iter(RB_BLOCK_CALL_FUNC_ARGLIST(i, memo_)) { + struct uniq_iter_memo *memo = (struct uniq_iter_memo *)memo_; ENUM_WANT_SVALUE(); - rb_hash_add_new_element(hash, rb_yield_values2(argc, argv), i); + if (rb_set_add_no_check(memo->set, rb_yield_values2(argc, argv))) { + rb_ary_push(memo->ary, i); + } return Qnil; } @@ -4909,15 +4918,18 @@ uniq_iter(RB_BLOCK_CALL_FUNC_ARGLIST(i, hash)) static VALUE enum_uniq(VALUE obj) { - VALUE hash, ret; - rb_block_call_func *const func = - rb_block_given_p() ? uniq_iter : uniq_func; - - hash = rb_obj_hide(rb_hash_new()); - rb_block_call(obj, id_each, 0, 0, func, hash); - ret = rb_hash_values(hash); - rb_hash_clear(hash); - return ret; + if (rb_block_given_p()) { + struct uniq_iter_memo memo; + memo.set = rb_obj_hide(rb_set_new()); + memo.ary = rb_ary_new(); + rb_block_call(obj, id_each, 0, 0, uniq_iter, (VALUE)&memo); + return memo.ary; + } + else { + VALUE set = rb_obj_hide(rb_set_new()); + rb_block_call(obj, id_each, 0, 0, uniq_func, set); + return rb_set_to_a(set); + } } static VALUE diff --git a/ext/socket/ipsocket.c b/ext/socket/ipsocket.c index be8ee53a427802..506a56a1865bc4 100644 --- a/ext/socket/ipsocket.c +++ b/ext/socket/ipsocket.c @@ -713,7 +713,7 @@ init_fast_fallback_inetsock_internal(VALUE v) VALUE test_delay_setting = rb_hash_aref(test_mode_settings, ID2SYM(rb_intern("delay"))); if (!NIL_P(test_delay_setting)) { VALUE rb_test_delay_ms = rb_hash_aref(test_delay_setting, ID2SYM(rb_intern(family_sym))); - long test_delay_ms = NIL_P(rb_test_delay_ms) ? 0 : rb_test_delay_ms; + long test_delay_ms = NIL_P(rb_test_delay_ms) ? 0 : NUM2LONG(rb_test_delay_ms); arg->getaddrinfo_entries[i]->test_sleep_ms = test_delay_ms; } diff --git a/file.c b/file.c index f7d0a46420d4f2..e3e12052e9e2de 100644 --- a/file.c +++ b/file.c @@ -3133,24 +3133,24 @@ lchmod_internal(const char *path, void *mode) * call-seq: * File.lchmod(mode, *paths) -> paths_count * - * Not supported on some platforms (raises NotImplementedError). + * Not supported on Linux or Windows (raises NotImplementedError). * - * When supported: like File::chmod, but does not follow symbolic links, + * When supported: like File::chmod, + * but does not follow [symbolic links](rdoc-ref:file/symbolic_links.md), * and therefore changes the mode of the entries given by `paths`; * returns the number of paths given: * * ```ruby * File.write('t.tmp', '') * File.symlink('t.tmp', 'link') - * File.stat('t.tmp').mode.to_s(8) # => "100664" - * File.stat('link').mode.to_s(8) # => "100664" + * File.lstat('t.tmp').mode.to_s(8) # => "100664" + * File.lstat('link').mode.to_s(8) # => "120755" * File.lchmod(0777, 'link') - * File.stat('t.tmp').mode.to_s(8) # => "100664" - * File.stat('link').mode.to_s(8) # => "100777" + * File.lstat('t.tmp').mode.to_s(8) # => "100664" + * File.lstat('link').mode.to_s(8) # => "120777" * File.delete('t.tmp') * File.delete('link') * ``` - * */ static VALUE @@ -3881,19 +3881,28 @@ unlink_internal(const char *path, void *arg) } /* + * :markup: markdown + * * call-seq: - * File.delete(*filepaths) -> integer - * File.unlink(*filepaths) -> integer + * File.delete(*paths) -> integer + * File.unlink(*paths) -> integer * - * Removes the file entry at each path in +filepaths+; - * returns the number of removed files. + * Removes the entry at each path in `paths`; + * returns the count of removed entries. * - * File.write('t.tmp', 'foo') - * File.write('u.tmp', 'bar') - * File.delete('t.tmp', 'u.tmp') # => 2 + * Does not follow [symbolic links](rdoc-ref:file/symbolic_links.md); + * if an entry is a symlink, the link itself is removed. + * + * ```ruby + * File.write('t.tmp', 'foo') + * File.write('u.tmp', 'bar') + * File.delete('t.tmp', 'u.tmp') # => 2 + * File.symlink('README.md', 'foo') + * File.unlink('foo') # => 1 + * ``` * * Raises an exception on any error; - * some files may have been deleted before the path causing the error. + * some entries may have been deleted before the path causing the error. */ static VALUE diff --git a/include/ruby/fiber/scheduler.h b/include/ruby/fiber/scheduler.h index 73a1ed40640be0..10eb1352777da8 100644 --- a/include/ruby/fiber/scheduler.h +++ b/include/ruby/fiber/scheduler.h @@ -25,7 +25,8 @@ RBIMPL_SYMBOL_EXPORT_BEGIN() // Version 3: Adds support for `fiber_interrupt`. // Version 4: IO hooks use single-transfer `(offset, length)` semantics. -#define RUBY_FIBER_SCHEDULER_VERSION 4 +// Version 5: `fiber_interrupt` may receive an operation-scoped proxy. +#define RUBY_FIBER_SCHEDULER_VERSION 5 struct timeval; struct rb_thread_struct; @@ -475,16 +476,22 @@ int rb_fiber_scheduler_blocking_operation_cancel(rb_fiber_scheduler_blocking_ope VALUE rb_fiber_scheduler_blocking_operation_wait(VALUE scheduler, void* (*function)(void *), void *data, rb_unblock_function_t *unblock_function, void *data2, int flags, struct rb_fiber_scheduler_blocking_operation_state *state); /** - * Interrupt a fiber by raising an exception. You can construct an exception using `rb_make_exception`. + * Interrupt a target by raising an exception. You can construct an exception using `rb_make_exception`. + * + * The target is usually a Fiber. For an IO operation, it may instead be an + * operation-scoped proxy which responds to `alive?`, `raise`, and `transfer`. + * Calling `transfer` raises the stored exception in the target Fiber, allowing + * the proxy to be queued directly. Schedulers should restrict their use of the + * target to those methods. * * This hook may be invoked by a different thread. * * @param[in] scheduler Target scheduler. - * @param[in] fiber The fiber to interrupt. + * @param[in] target The Fiber or operation-scoped proxy to interrupt. * @param[in] exception The exception to raise in the fiber. * @return What `scheduler.fiber_interrupt` returns. */ -VALUE rb_fiber_scheduler_fiber_interrupt(VALUE scheduler, VALUE fiber, VALUE exception); +VALUE rb_fiber_scheduler_fiber_interrupt(VALUE scheduler, VALUE target, VALUE exception); /** * Create and schedule a non-blocking fiber. diff --git a/internal/io.h b/internal/io.h index 2110f0b0876271..f47aba42d1e9af 100644 --- a/internal/io.h +++ b/internal/io.h @@ -27,6 +27,9 @@ struct rb_io_blocking_operation { // The execution context of the blocking operation. struct rb_execution_context_struct *ec; + + // An operation-scoped target passed to the Fiber scheduler if interrupted. + VALUE scheduler_interrupt_target; }; /** Ruby's IO, metadata and buffers. */ diff --git a/internal/scheduler.h b/internal/scheduler.h new file mode 100644 index 00000000000000..f79b5d02a6dbc9 --- /dev/null +++ b/internal/scheduler.h @@ -0,0 +1,7 @@ +#pragma once + +#include "ruby/ruby.h" + +VALUE rb_fiber_scheduler_interrupt_target_new(VALUE fiber, VALUE exception); +VALUE rb_fiber_scheduler_interrupt_target_exception(VALUE target); +void rb_fiber_scheduler_interrupt_target_invalidate(VALUE target); diff --git a/internal/set.h b/internal/set.h index 48c282c3d4e23d..38028ecb263709 100644 --- a/internal/set.h +++ b/internal/set.h @@ -12,5 +12,8 @@ #include "ruby/ruby.h" VALUE rb_ident_set_new(void); +bool rb_set_add_no_check(VALUE set, VALUE element); +bool rb_set_delete_no_check(VALUE set, VALUE element); +VALUE rb_set_to_a(VALUE set); #endif /* INTERNAL_SET_H */ diff --git a/pathname_builtin.rb b/pathname_builtin.rb index 8728c521c7f4db..2191beccd1137a 100644 --- a/pathname_builtin.rb +++ b/pathname_builtin.rb @@ -1543,23 +1543,25 @@ def chmod(mode) File.chmod(mode, @path) end # call-seq: # lchmod(mode) -> 1 # - # Not supported on some platforms (raises NotImplementedError). + # Not supported on Linux or Windows (raises NotImplementedError). # - # When supported: like Pathname::chmod, but does not follow symbolic links, + # When supported: like Pathname#chmod, + # but does not follow [symbolic links](rdoc-ref:file/symbolic_links.md), # and therefore changes the mode of the entry specified by `self`: # # ```ruby - # File.write('t.tmp', '') + # file = Pathname('t.tmp') + # file.write('') # File.symlink('t.tmp', 'link') - # File.stat('t.tmp').mode.to_s(8) # => "100664" - # File.stat('link').mode.to_s(8) # => "100664" - # Pathname('link').lchmod(0777) - # File.stat('t.tmp').mode.to_s(8) # => "100664" - # File.stat('link').mode.to_s(8) # => "100777" - # File.delete('t.tmp') - # File.delete('link') - # ``` - # + # symlink = Pathname('link') + # file.lstat.mode.to_s(8) # => "100644" + # symlink.lstat.mode.to_s(8) # => "120755" + # symlink.lchmod(0777) + # file.lstat.mode.to_s(8) # => "100644" + # symlink.lstat.mode.to_s(8) # => "120777" + # file.delete + # symlink.delete + # ```` def lchmod(mode) File.lchmod(mode, @path) end # :markup: markdown @@ -2975,11 +2977,18 @@ class Pathname # * mixed * # unlink -> 0 or 1 # # Removes the entry represented by `self`; - # returns `0` if a directory, `1` if a file: + # returns `0` if a directory, `1` otherwise. + # + # Does not follow [symbolic links](rdoc-ref:file/symbolic_links.md); + # if the entry is a symlink, the link itself is removed. # # ```ruby # Pathname(Pathname.mktmpdir).unlink # => 0 # Pathname(Tempfile.create).unlink # => 1 + # pn_target = Pathname('README.md') # => # + # pn_link = Pathname('foo') # => # + # pn_link.make_symlink(pn_target) + # pn_link.delete # ``` # def unlink() diff --git a/scheduler.c b/scheduler.c index d315b275fbad10..0c955a28747aaf 100644 --- a/scheduler.c +++ b/scheduler.c @@ -17,6 +17,7 @@ #include "ruby/thread.h" // For `ruby_thread_has_gvl_p`: +#include "internal/scheduler.h" #include "internal/thread.h" // For atomic operations: @@ -51,6 +52,118 @@ static ID id_fiber_schedule; // Our custom blocking operation class static VALUE rb_cFiberSchedulerBlockingOperation; +static VALUE rb_cFiberSchedulerInterruptTarget; + +struct rb_fiber_scheduler_interrupt_target { + VALUE fiber; + VALUE exception; + bool active; +}; + +static void +interrupt_target_mark(void *ptr) +{ + struct rb_fiber_scheduler_interrupt_target *target = ptr; + + rb_gc_mark(target->fiber); + rb_gc_mark(target->exception); +} + +static size_t +interrupt_target_memsize(const void *ptr) +{ + return sizeof(struct rb_fiber_scheduler_interrupt_target); +} + +static const rb_data_type_t interrupt_target_data_type = { + "Fiber::Scheduler::InterruptTarget", + { + interrupt_target_mark, + RUBY_DEFAULT_FREE, + interrupt_target_memsize, + }, + 0, 0, RUBY_TYPED_FREE_IMMEDIATELY +}; + +static struct rb_fiber_scheduler_interrupt_target * +get_interrupt_target(VALUE self) +{ + struct rb_fiber_scheduler_interrupt_target *target; + TypedData_Get_Struct(self, struct rb_fiber_scheduler_interrupt_target, &interrupt_target_data_type, target); + return target; +} + +static VALUE +interrupt_target_alive_p(VALUE self) +{ + struct rb_fiber_scheduler_interrupt_target *target = get_interrupt_target(self); + + if (!target->active) { + return Qfalse; + } + + return rb_fiber_alive_p(target->fiber); +} + +static VALUE +interrupt_target_raise(int argc, VALUE *argv, VALUE self) +{ + struct rb_fiber_scheduler_interrupt_target *target = get_interrupt_target(self); + + if (!target->active) { + return Qnil; + } + + return rb_fiber_raise(target->fiber, argc, argv); +} + +static VALUE +interrupt_target_transfer(VALUE self) +{ + struct rb_fiber_scheduler_interrupt_target *target = get_interrupt_target(self); + + if (!target->active || !RTEST(rb_fiber_alive_p(target->fiber))) { + return Qnil; + } + + VALUE exception = target->exception; + return rb_fiber_raise(target->fiber, 1, &exception); +} + +VALUE +rb_fiber_scheduler_interrupt_target_new(VALUE fiber, VALUE exception) +{ + struct rb_fiber_scheduler_interrupt_target *target; + VALUE self = TypedData_Make_Struct(rb_cFiberSchedulerInterruptTarget, struct rb_fiber_scheduler_interrupt_target, &interrupt_target_data_type, target); + + target->fiber = fiber; + target->exception = exception; + target->active = true; + + return self; +} + +VALUE +rb_fiber_scheduler_interrupt_target_exception(VALUE self) +{ + struct rb_fiber_scheduler_interrupt_target *target = get_interrupt_target(self); + + if (target->active) { + return target->exception; + } + + return Qnil; +} + +void +rb_fiber_scheduler_interrupt_target_invalidate(VALUE self) +{ + struct rb_fiber_scheduler_interrupt_target *target = get_interrupt_target(self); + + target->active = false; + target->fiber = Qnil; + target->exception = Qnil; +} /* * Custom blocking operation structure for blocking operations @@ -344,6 +457,15 @@ Init_Fiber_Scheduler(void) // Register the anonymous class as a GC root so it doesn't get collected rb_gc_register_mark_object(rb_cFiberSchedulerBlockingOperation); + // Define an anonymous interrupt target class. Instances are passed to + // fiber_interrupt and cannot be instantiated directly. + rb_cFiberSchedulerInterruptTarget = rb_class_new(rb_cObject); + rb_undef_alloc_func(rb_cFiberSchedulerInterruptTarget); + rb_define_method(rb_cFiberSchedulerInterruptTarget, "alive?", interrupt_target_alive_p, 0); + rb_define_method(rb_cFiberSchedulerInterruptTarget, "raise", interrupt_target_raise, -1); + rb_define_method(rb_cFiberSchedulerInterruptTarget, "transfer", interrupt_target_transfer, 0); + rb_gc_register_mark_object(rb_cFiberSchedulerInterruptTarget); + #if 0 /* for RDoc */ rb_cFiberScheduler = rb_define_class_under(rb_cFiber, "Scheduler", rb_cObject); rb_define_method(rb_cFiberScheduler, "close", rb_fiber_scheduler_close, 0); @@ -1154,17 +1276,21 @@ VALUE rb_fiber_scheduler_blocking_operation_wait(VALUE scheduler, void* (*functi /* * Document-method: Fiber::Scheduler#fiber_interrupt - * call-seq: fiber_interrupt(fiber, exception) + * call-seq: fiber_interrupt(target, exception) * - * Invoked by Ruby's core methods to notify the scheduler that the blocked fiber should be interrupted - * with an exception. For example, IO#close uses this method to interrupt fibers that are performing - * blocking IO operations. + * Invoked by Ruby's core methods to notify the scheduler that a blocked fiber + * should be interrupted with an exception. For IO operations, +target+ is an + * operation-scoped proxy which responds to #alive?, #raise, and #transfer. The + * scheduler should enqueue the target on its owning thread and use only those + * methods. #transfer raises the interruption's stored exception in the target. + * Once the operation completes, #alive? returns false and #raise has no effect, + * preventing a delayed exception from escaping into a later operation. * */ -VALUE rb_fiber_scheduler_fiber_interrupt(VALUE scheduler, VALUE fiber, VALUE exception) +VALUE rb_fiber_scheduler_fiber_interrupt(VALUE scheduler, VALUE target, VALUE exception) { VALUE arguments[] = { - fiber, exception + target, exception }; VALUE result; diff --git a/set.c b/set.c index 1c19c2ca81f057..95d9b815f9f757 100644 --- a/set.c +++ b/set.c @@ -2278,6 +2278,28 @@ rb_ident_set_new(void) return set_alloc_with_size_and_type(rb_cSet, 0, &identhash); } +bool +rb_set_add_no_check(VALUE set, VALUE element) +{ + if (set_insert(RSET_TABLE(set), (st_data_t)element) == 0) { + RB_OBJ_WRITTEN(set, Qundef, element); + return true; + } + return false; +} + +bool +rb_set_delete_no_check(VALUE set, VALUE element) +{ + return set_table_delete(RSET_TABLE(set), (st_data_t *)&element) != 0; +} + +VALUE +rb_set_to_a(VALUE set) +{ + return set_i_to_a(set); +} + /* C-API functions */ void diff --git a/test/fiber/scheduler.rb b/test/fiber/scheduler.rb index af66ba106002e1..2cb04812fa2d4f 100644 --- a/test/fiber/scheduler.rb +++ b/test/fiber/scheduler.rb @@ -313,24 +313,9 @@ def unblock(blocker, fiber) io.write_nonblock('.') end - class FiberInterrupt - def initialize(fiber, exception) - @fiber = fiber - @exception = exception - end - - def alive? - @fiber.alive? - end - - def transfer - @fiber.raise(@exception) - end - end - - def fiber_interrupt(fiber, exception) + def fiber_interrupt(target, _exception) @lock.synchronize do - @ready << FiberInterrupt.new(fiber, exception) + @ready << target end io = @urgent.last diff --git a/test/fiber/test_io_close.rb b/test/fiber/test_io_close.rb index 742b40841d90d6..14c6731acbb9e2 100644 --- a/test/fiber/test_io_close.rb +++ b/test/fiber/test_io_close.rb @@ -1,5 +1,6 @@ # frozen_string_literal: true require 'test/unit' +require 'io/wait' require_relative 'scheduler' class TestFiberIOClose < Test::Unit::TestCase @@ -44,6 +45,46 @@ def test_io_close_across_fibers end end + def test_io_close_interrupt_does_not_escape_blocking_operation + with_socket_pair do |source, source_peer| + with_socket_pair do |unrelated, unrelated_peer| + errors = [] + + thread = Thread.new do + scheduler = Scheduler.new + Fiber.set_scheduler scheduler + + 5.times do + Fiber.schedule do + begin + source.wait_readable(0.01) + source.close + rescue => error + errors << [:source, error] + end + + begin + unrelated.wait_readable(0.01) + rescue => error + errors << [:unrelated, error] + end + end + end + end + + thread.join + + assert_equal 4, errors.size + assert_equal [:source], errors.map(&:first).uniq + errors.each do |location, error| + assert_equal :source, location + assert_instance_of IOError, error + assert_match(/closed/, error.message) + end + end + end + end + def test_io_close_blocking_thread omit "Interrupting a io_wait read is not supported!" if RUBY_PLATFORM =~ /mswin|mingw/ @@ -80,9 +121,19 @@ def test_io_close_blocking_fiber with_socket_pair do |i, o| error = nil + scheduler = nil + + scheduler_class = Class.new(Scheduler) do + attr_reader :interrupt_target + + def fiber_interrupt(target, exception) + @interrupt_target = target + super + end + end thread = Thread.new do - scheduler = Scheduler.new + scheduler = scheduler_class.new Fiber.set_scheduler scheduler Fiber.schedule do @@ -102,6 +153,11 @@ def test_io_close_blocking_fiber assert_instance_of IOError, error assert_match(/closed/, error.message) + + interrupt_target = scheduler.interrupt_target + assert_not_predicate interrupt_target, :alive? + assert_nil interrupt_target.transfer + assert_nil interrupt_target.raise(IOError.new) end end end diff --git a/thread.c b/thread.c index 5de075cf73ad9a..4b64e8ed92d49c 100644 --- a/thread.c +++ b/thread.c @@ -87,6 +87,7 @@ #include "internal/object.h" #include "internal/proc.h" #include "ruby/fiber/scheduler.h" +#include "internal/scheduler.h" #include "internal/signal.h" #include "internal/thread.h" #include "internal/time.h" @@ -2023,6 +2024,10 @@ rb_io_blocking_operation_exit(struct rb_io *io, struct rb_io_blocking_operation // Indicate that the blocking operation is no longer active: blocking_operation->ec = NULL; + if (!NIL_P(blocking_operation->scheduler_interrupt_target)) { + rb_fiber_scheduler_interrupt_target_invalidate(blocking_operation->scheduler_interrupt_target); + } + if (RB_TEST(wakeup_mutex)) { struct io_blocking_operation_arguments arguments = { .io = io, @@ -2047,6 +2052,30 @@ rb_thread_io_blocking_operation_ensure(VALUE _argument) return Qnil; } +struct thread_io_blocking_operation_arguments { + VALUE (*function)(VALUE); + VALUE argument; + struct rb_io_blocking_operation *blocking_operation; +}; + +static VALUE +rb_thread_io_blocking_operation_body(VALUE _arguments) +{ + struct thread_io_blocking_operation_arguments *arguments = (void *)_arguments; + VALUE result = arguments->function(arguments->argument); + VALUE target = arguments->blocking_operation->scheduler_interrupt_target; + + if (!NIL_P(target)) { + VALUE exception = rb_fiber_scheduler_interrupt_target_exception(target); + + if (!NIL_P(exception)) { + rb_exc_raise(exception); + } + } + + return result; +} + /* * Executes a function that performs a blocking IO operation, while properly tracking * the operation in the IO's blocking_operations list. This ensures proper cleanup @@ -2069,6 +2098,7 @@ rb_thread_io_blocking_operation(VALUE self, VALUE(*function)(VALUE), VALUE argum rb_execution_context_t *ec = GET_EC(); struct rb_io_blocking_operation blocking_operation = { .ec = ec, + .scheduler_interrupt_target = Qnil, }; rb_io_blocking_operation_enter(io, &blocking_operation); @@ -2077,7 +2107,13 @@ rb_thread_io_blocking_operation(VALUE self, VALUE(*function)(VALUE), VALUE argum .blocking_operation = &blocking_operation }; - return rb_ensure(function, argument, rb_thread_io_blocking_operation_ensure, (VALUE)&io_blocking_operation_arguments); + struct thread_io_blocking_operation_arguments arguments = { + .function = function, + .argument = argument, + .blocking_operation = &blocking_operation, + }; + + return rb_ensure(rb_thread_io_blocking_operation_body, (VALUE)&arguments, rb_thread_io_blocking_operation_ensure, (VALUE)&io_blocking_operation_arguments); } static bool @@ -2187,6 +2223,7 @@ rb_thread_io_blocking_call(struct rb_io* io, rb_blocking_function_t *func, void struct rb_io_blocking_operation blocking_operation = { .ec = ec, + .scheduler_interrupt_target = Qnil, }; rb_io_blocking_operation_enter(io, &blocking_operation); @@ -3086,7 +3123,16 @@ thread_io_close_notify_all(VALUE _io) rb_thread_t *thread = ec->thread_ptr; if (thread->scheduler != Qnil) { - rb_fiber_scheduler_fiber_interrupt(thread->scheduler, rb_fiberptr_self(ec->fiber_ptr), error); + VALUE target = blocking_operation->scheduler_interrupt_target; + + if (NIL_P(target)) { + VALUE fiber = rb_fiberptr_self(ec->fiber_ptr); + target = rb_fiber_scheduler_interrupt_target_new(fiber, error); + blocking_operation->scheduler_interrupt_target = target; + } + + rb_fiber_scheduler_fiber_interrupt(thread->scheduler, target, error); + RB_GC_GUARD(target); } else { // If the thread is not the current thread, we need to enqueue an error: @@ -4884,6 +4930,7 @@ thread_io_wait(rb_thread_t *th, struct rb_io *io, int fd, int events, struct tim if (io) { blocking_operation.ec = ec; + blocking_operation.scheduler_interrupt_target = Qnil; COMPILER_WARNING_PUSH #if RBIMPL_COMPILER_SINCE(GCC, 12, 0, 0) COMPILER_WARNING_IGNORED(-Wdangling-pointer) @@ -4925,10 +4972,11 @@ COMPILER_WARNING_POP break; case io_wait_unhandled: EC_PUSH_TAG(ec); + struct timeval *volatile blocking_timeout = timeout; if ((state = EC_EXEC_TAG()) == TAG_NONE) { rb_hrtime_t *to, rel, end = 0; RUBY_VM_CHECK_INTS_BLOCKING(ec); - timeout_prepare(&to, &rel, &end, timeout); + timeout_prepare(&to, &rel, &end, blocking_timeout); do { nfds = numberof(fds); result = wait_for_single_fd_blocking_region(th, fds, nfds, to, &lerrno); @@ -5054,12 +5102,14 @@ thread_io_wait(rb_thread_t *th, struct rb_io *io, int fd, int events, struct tim if (io) { args.io = io; blocking_operation.ec = th->ec; + blocking_operation.scheduler_interrupt_target = Qnil; rb_io_blocking_operation_enter(io, &blocking_operation); args.blocking_operation = &blocking_operation; } else { args.io = NULL; blocking_operation.ec = NULL; + blocking_operation.scheduler_interrupt_target = Qnil; args.blocking_operation = NULL; }