diff options
Diffstat (limited to 'test/fiber')
-rw-r--r-- | test/fiber/autoload.rb | 3 | ||||
-rw-r--r-- | test/fiber/scheduler.rb | 307 | ||||
-rw-r--r-- | test/fiber/test_address_resolve.rb | 278 | ||||
-rw-r--r-- | test/fiber/test_enumerator.rb | 16 | ||||
-rw-r--r-- | test/fiber/test_io.rb | 165 | ||||
-rw-r--r-- | test/fiber/test_io_buffer.rb | 199 | ||||
-rw-r--r-- | test/fiber/test_mutex.rb | 42 | ||||
-rw-r--r-- | test/fiber/test_process.rb | 42 | ||||
-rw-r--r-- | test/fiber/test_queue.rb | 54 | ||||
-rw-r--r-- | test/fiber/test_ractor.rb | 2 | ||||
-rw-r--r-- | test/fiber/test_scheduler.rb | 97 | ||||
-rw-r--r-- | test/fiber/test_sleep.rb | 25 | ||||
-rw-r--r-- | test/fiber/test_storage.rb | 115 | ||||
-rw-r--r-- | test/fiber/test_thread.rb | 85 |
14 files changed, 1363 insertions, 67 deletions
diff --git a/test/fiber/autoload.rb b/test/fiber/autoload.rb new file mode 100644 index 0000000000..dcb27164a7 --- /dev/null +++ b/test/fiber/autoload.rb @@ -0,0 +1,3 @@ +sleep 0.01 +module TestFiberSchedulerAutoload +end diff --git a/test/fiber/scheduler.rb b/test/fiber/scheduler.rb index 0461cfbe3f..3926226ca3 100644 --- a/test/fiber/scheduler.rb +++ b/test/fiber/scheduler.rb @@ -1,8 +1,13 @@ # frozen_string_literal: true # This is an example and simplified scheduler for test purposes. -# It is not efficient for a large number of file descriptors as it uses IO.select(). -# Production Fiber schedulers should use epoll/kqueue/etc. +# - It is not efficient for a large number of file descriptors as it uses +# IO.select(). +# - It does not correctly handle multiple calls to `wait` with the same file +# descriptor and overlapping events. +# - Production fiber schedulers should use epoll/kqueue/etc. Consider using the +# [`io-event`](https://github.com/socketry/io-event) gem instead of this +# scheduler if you want something simple to build on. require 'fiber' require 'socket' @@ -14,15 +19,25 @@ rescue LoadError end class Scheduler - def initialize + experimental = Warning[:experimental] + begin + Warning[:experimental] = false + IO::Buffer.new(0) + ensure + Warning[:experimental] = experimental + end + + def initialize(fiber = Fiber.current) + @fiber = fiber + @readable = {} @writable = {} @waiting = {} @closed = false - @lock = Mutex.new - @blocking = 0 + @lock = Thread::Mutex.new + @blocking = Hash.new.compare_by_identity @ready = [] @urgent = IO.pipe @@ -32,6 +47,10 @@ class Scheduler attr :writable attr :waiting + def transfer + @fiber.transfer + end + def next_timeout _fiber, timeout = @waiting.min_by{|key, value| value} @@ -49,8 +68,8 @@ class Scheduler def run # $stderr.puts [__method__, Fiber.current].inspect - while @readable.any? or @writable.any? or @waiting.any? or @blocking.positive? - # Can only handle file descriptors up to 1024... + while @readable.any? or @writable.any? or @waiting.any? or @blocking.any? + # May only handle file descriptors up to 1024... readable, writable = IO.select(@readable.keys + [@urgent.first], @writable.keys, [], next_timeout) # puts "readable: #{readable}" if readable&.any? @@ -60,6 +79,7 @@ class Scheduler readable&.each do |io| if fiber = @readable.delete(io) + @writable.delete(io) if @writable[io] == fiber selected[fiber] = IO::READABLE elsif io == @urgent.first @urgent.first.read_nonblock(1024) @@ -68,12 +88,13 @@ class Scheduler writable&.each do |io| if fiber = @writable.delete(io) - selected[fiber] |= IO::WRITABLE + @readable.delete(io) if @readable[io] == fiber + selected[fiber] = selected.fetch(fiber, 0) | IO::WRITABLE end end selected.each do |fiber, events| - fiber.resume(events) + fiber.transfer(events) end if @waiting.any? @@ -83,7 +104,7 @@ class Scheduler waiting.each do |fiber, timeout| if fiber.alive? if timeout <= time - fiber.resume + fiber.transfer else @waiting[fiber] = timeout end @@ -99,23 +120,43 @@ class Scheduler end ready.each do |fiber| - fiber.resume + fiber.transfer end end end end - def close + # A fiber scheduler hook, invoked when the scheduler goes out of scope. + def scheduler_close + close(true) + end + + # If the `scheduler_close` hook does not exist, this method `close` will be + # invoked instead when the fiber scheduler goes out of scope. This is legacy + # behaviour, you should almost certainly use `scheduler_close`. The reason for + # this, is `scheduler_close` is called when the scheduler goes out of scope, + # while `close` may be called by the user. + def close(internal = false) # $stderr.puts [__method__, Fiber.current].inspect - raise "Scheduler already closed!" if @closed + unless internal + if Fiber.scheduler == self + return Fiber.set_scheduler(nil) + end + end + + if @closed + raise "Scheduler already closed!" + end self.run ensure - @urgent.each(&:close) - @urgent = nil + if @urgent + @urgent.each(&:close) + @urgent = nil + end - @closed = true + @closed ||= true # We freeze to detect any unintended modifications after the scheduler is closed: self.freeze @@ -129,6 +170,7 @@ class Scheduler Process.clock_gettime(Process::CLOCK_MONOTONIC) end + # This hook is invoked by `Timeout.timeout` and related code. def timeout_after(duration, klass, message, &block) fiber = Fiber.current @@ -147,6 +189,7 @@ class Scheduler end end + # This hook is invoked by `Process.wait`, `system`, and backticks. def process_wait(pid, flags) # $stderr.puts [__method__, pid, flags, Fiber.current].inspect @@ -156,21 +199,47 @@ class Scheduler end.value end + # This hook is invoked by `IO#read` and `IO#write` in the case that `io_read` + # and `io_write` hooks are not available. This implementation is not + # completely general, in the sense that calling `io_wait` multiple times with + # the same `io` and `events` will not work, which is okay for tests but not + # for real code. Correct fiber schedulers should not have this limitation. def io_wait(io, events, duration) # $stderr.puts [__method__, io, events, duration, Fiber.current].inspect + fiber = Fiber.current + unless (events & IO::READABLE).zero? - @readable[io] = Fiber.current + @readable[io] = fiber + readable = true end unless (events & IO::WRITABLE).zero? - @writable[io] = Fiber.current + @writable[io] = fiber + writable = true end - Fiber.yield + if duration + @waiting[fiber] = current_time + duration + end + + @fiber.transfer + ensure + @waiting.delete(fiber) if duration + @readable.delete(io) if readable + @writable.delete(io) if writable end - # Used for Kernel#sleep and Mutex#sleep + # This hook is invoked by `IO.select`. Using a thread ensures that the + # operation does not block the fiber scheduler. + def io_select(...) + # Emulate the operation using a non-blocking thread: + Thread.new do + IO.select(...) + end.value + end + + # This hook is invoked by `Kernel#sleep` and `Thread::Mutex#sleep`. def kernel_sleep(duration = nil) # $stderr.puts [__method__, duration, Fiber.current].inspect @@ -179,30 +248,35 @@ class Scheduler return true end - # Used when blocking on synchronization (Mutex#lock, Queue#pop, SizedQueue#push, ...) + # This hook is invoked by blocking options such as `Thread::Mutex#lock`, + # `Thread::Queue#pop` and `Thread::SizedQueue#push`, which are unblocked by + # other threads/fibers. To unblock a blocked fiber, you should call `unblock` + # with the same `blocker` and `fiber` arguments. def block(blocker, timeout = nil) # $stderr.puts [__method__, blocker, timeout].inspect + fiber = Fiber.current + if timeout - @waiting[Fiber.current] = current_time + timeout + @waiting[fiber] = current_time + timeout begin - Fiber.yield + @fiber.transfer ensure # Remove from @waiting in the case #unblock was called before the timeout expired: - @waiting.delete(Fiber.current) + @waiting.delete(fiber) end else - @blocking += 1 + @blocking[fiber] = true begin - Fiber.yield + @fiber.transfer ensure - @blocking -= 1 + @blocking.delete(fiber) end end end - # Used when synchronization wakes up a previously-blocked fiber (Mutex#unlock, Queue#push, ...). - # This might be called from another thread. + # This method is invoked from a thread or fiber to unblock a fiber that is + # blocked by `block`. It is expected to be thread safe. def unblock(blocker, fiber) # $stderr.puts [__method__, blocker, fiber].inspect # $stderr.puts blocker.backtrace.inspect @@ -216,11 +290,184 @@ class Scheduler io.write_nonblock('.') end + # This hook is invoked by `Fiber.schedule`. Strictly speaking, you should use + # it to create scheduled fibers, but it is not required in practice; + # `Fiber.new` is usually sufficient. def fiber(&block) fiber = Fiber.new(blocking: false, &block) - fiber.resume + fiber.transfer return fiber end + + # This hook is invoked by `Addrinfo.getaddrinfo`. Using a thread ensures that + # the operation does not block the fiber scheduler, since `getaddrinfo` is + # usually provided by `libc` and is blocking. + def address_resolve(hostname) + Thread.new do + Addrinfo.getaddrinfo(hostname, nil).map(&:ip_address).uniq + end.value + end +end + +# This scheduler class implements `io_read` and `io_write` hooks which require +# `IO::Buffer`. +class IOBufferScheduler < Scheduler + EAGAIN = -Errno::EAGAIN::Errno + + def io_read(io, buffer, length, offset) + total = 0 + io.nonblock = true + + while true + maximum_size = buffer.size - offset + result = blocking{buffer.read(io, maximum_size, offset)} + + if result > 0 + total += result + offset += result + break if total >= length + elsif result == 0 + break + elsif result == EAGAIN + if length > 0 + self.io_wait(io, IO::READABLE, nil) + else + return result + end + elsif result < 0 + return result + end + end + + return total + end + + def io_write(io, buffer, length, offset) + total = 0 + io.nonblock = true + + while true + maximum_size = buffer.size - offset + result = blocking{buffer.write(io, maximum_size, offset)} + + if result > 0 + total += result + offset += result + break if total >= length + elsif result == 0 + break + elsif result == EAGAIN + if length > 0 + self.io_wait(io, IO::WRITABLE, nil) + else + return result + end + elsif result < 0 + return result + end + end + + return total + end + + def io_pread(io, buffer, from, length, offset) + total = 0 + io.nonblock = true + + while true + maximum_size = buffer.size - offset + result = blocking{buffer.pread(io, from, maximum_size, offset)} + + if result > 0 + total += result + offset += result + from += result + break if total >= length + elsif result == 0 + break + elsif result == EAGAIN + if length > 0 + self.io_wait(io, IO::READABLE, nil) + else + return result + end + elsif result < 0 + return result + end + end + + return total + end + + def io_pwrite(io, buffer, from, length, offset) + total = 0 + io.nonblock = true + + while true + maximum_size = buffer.size - offset + result = blocking{buffer.pwrite(io, from, maximum_size, offset)} + + if result > 0 + total += result + offset += result + from += result + break if total >= length + elsif result == 0 + break + elsif result == EAGAIN + if length > 0 + self.io_wait(io, IO::WRITABLE, nil) + else + return result + end + elsif result < 0 + return result + end + end + + return total + end + + def blocking(&block) + Fiber.blocking(&block) + end +end + +# This scheduler has a broken implementation of `unblock`` in the sense that it +# raises an exception. This is used to test the behavior of the scheduler when +# unblock raises an exception. +class BrokenUnblockScheduler < Scheduler + def unblock(blocker, fiber) + super + + raise "Broken unblock!" + end +end + +# This scheduler has a broken implementation of `unblock` in the sense that it +# sleeps. This is used to test the behavior of the scheduler when unblock +# messes with the internal thread state in an unexpected way. +class SleepingUnblockScheduler < Scheduler + # This method is invoked when the thread is exiting. + def unblock(blocker, fiber) + super + + # This changes the current thread state to `THREAD_RUNNING` which causes `thread_join_sleep` to hang. + sleep(0.1) + end +end + +# This scheduler has a broken implementation of `kernel_sleep` in the sense that +# it invokes a blocking sleep which can cause a deadlock in some cases. +class SleepingBlockingScheduler < Scheduler + def kernel_sleep(duration = nil) + # Deliberaly sleep in a blocking state which can trigger a deadlock if the implementation is not correct. + Fiber.blocking{sleep 0.0001} + + self.block(:sleep, duration) + + return true + end end diff --git a/test/fiber/test_address_resolve.rb b/test/fiber/test_address_resolve.rb new file mode 100644 index 0000000000..09c8db6049 --- /dev/null +++ b/test/fiber/test_address_resolve.rb @@ -0,0 +1,278 @@ +# frozen_string_literal: true +require 'test/unit' +require_relative 'scheduler' + +class TestAddressResolve < Test::Unit::TestCase + class NullScheduler < Scheduler + def address_resolve(*) + end + end + + class StubScheduler < Scheduler + def address_resolve(hostname) + ["1.2.3.4", "1234::1"] + end + end + + def test_addrinfo_getaddrinfo_ipv4_domain_blocking + Thread.new do + scheduler = StubScheduler.new + Fiber.set_scheduler scheduler + + Fiber.schedule do + result = Addrinfo.getaddrinfo("example.com", 80, :AF_INET, :STREAM) + assert_equal(1, result.count) + + ai = result.first + assert_equal("1.2.3.4", ai.ip_address) + assert_equal(80, ai.ip_port) + assert_equal(Socket::AF_INET, ai.afamily) + assert_equal(Socket::SOCK_STREAM, ai.socktype) + end + end.join + end + + def test_addrinfo_getaddrinfo_ipv6_domain_blocking + Thread.new do + scheduler = StubScheduler.new + Fiber.set_scheduler scheduler + + Fiber.schedule do + result = Addrinfo.getaddrinfo("example.com", 80, :AF_INET6, :STREAM) + assert_equal(1, result.count) + + ai = result.first + assert_equal("1234::1", ai.ip_address) + assert_equal(80, ai.ip_port) + assert_equal(Socket::AF_INET6, ai.afamily) + assert_equal(Socket::SOCK_STREAM, ai.socktype) + end + end.join + end + + def test_addrinfo_getaddrinfo_pf_unspec_domain_blocking + Thread.new do + scheduler = StubScheduler.new + Fiber.set_scheduler scheduler + + Fiber.schedule do + results = Addrinfo.getaddrinfo("example.com", 80, :PF_UNSPEC, :STREAM) + assert_equal(2, results.count) + + ai_ipv4 = results.first + assert_equal("1.2.3.4", ai_ipv4.ip_address) + assert_equal(80, ai_ipv4.ip_port) + assert_equal(Socket::AF_INET, ai_ipv4.afamily) + assert_equal(Socket::SOCK_STREAM, ai_ipv4.socktype) + + ai_ipv6 = results.last + assert_equal("1234::1", ai_ipv6.ip_address) + assert_equal(80, ai_ipv6.ip_port) + assert_equal(Socket::AF_INET6, ai_ipv6.afamily) + assert_equal(Socket::SOCK_STREAM, ai_ipv6.socktype) + end + end.join + end + + def test_addrinfo_getaddrinfo_full_domain_blocking + Thread.new do + scheduler = StubScheduler.new + Fiber.set_scheduler scheduler + + Fiber.schedule do + results = Addrinfo.getaddrinfo("example.com", 80) + assert_equal(6, results.count) + + ai_ipv4_tcp = results[0] + assert_equal("1.2.3.4", ai_ipv4_tcp.ip_address) + assert_equal(80, ai_ipv4_tcp.ip_port) + assert_equal(Socket::AF_INET, ai_ipv4_tcp.afamily) + assert_equal(Socket::SOCK_STREAM, ai_ipv4_tcp.socktype) + + ai_ipv4_udp = results[1] + assert_equal("1.2.3.4", ai_ipv4_udp.ip_address) + assert_equal(80, ai_ipv4_udp.ip_port) + assert_equal(Socket::AF_INET, ai_ipv4_udp.afamily) + assert_equal(Socket::SOCK_DGRAM, ai_ipv4_udp.socktype) + + ai_ipv4_sock_raw = results[2] + assert_equal("1.2.3.4", ai_ipv4_sock_raw.ip_address) + assert_equal(80, ai_ipv4_sock_raw.ip_port) + assert_equal(Socket::AF_INET, ai_ipv4_sock_raw.afamily) + assert_equal(Socket::SOCK_RAW, ai_ipv4_sock_raw.socktype) + + ai_ipv6_tcp = results[3] + assert_equal("1234::1", ai_ipv6_tcp.ip_address) + assert_equal(80, ai_ipv6_tcp.ip_port) + assert_equal(Socket::AF_INET6, ai_ipv6_tcp.afamily) + assert_equal(Socket::SOCK_STREAM, ai_ipv6_tcp.socktype) + + ai_ipv6_udp = results[4] + assert_equal("1234::1", ai_ipv6_udp.ip_address) + assert_equal(80, ai_ipv6_udp.ip_port) + assert_equal(Socket::AF_INET6, ai_ipv6_udp.afamily) + assert_equal(Socket::SOCK_DGRAM, ai_ipv6_udp.socktype) + + ai_ipv6_sock_raw = results[5] + assert_equal("1234::1", ai_ipv6_sock_raw.ip_address) + assert_equal(80, ai_ipv6_sock_raw.ip_port) + assert_equal(Socket::AF_INET6, ai_ipv6_sock_raw.afamily) + assert_equal(Socket::SOCK_RAW, ai_ipv6_sock_raw.socktype) + end + end.join + end + + def test_addrinfo_getaddrinfo_numeric_non_blocking + Thread.new do + scheduler = NullScheduler.new # scheduler hook not invoked + Fiber.set_scheduler scheduler + + Fiber.schedule do + result = Addrinfo.getaddrinfo("1.2.3.4", 80, :AF_INET, :STREAM) + assert_equal(1, result.count) + + ai = result.first + assert_equal("1.2.3.4", ai.ip_address) + assert_equal(80, ai.ip_port) + assert_equal(Socket::AF_INET, ai.afamily) + assert_equal(Socket::SOCK_STREAM, ai.socktype) + end + end.join + end + + def test_addrinfo_getaddrinfo_any_non_blocking + Thread.new do + scheduler = NullScheduler.new # scheduler hook not invoked + Fiber.set_scheduler scheduler + + Fiber.schedule do + result = Addrinfo.getaddrinfo("<any>", 80, :AF_INET, :STREAM) + assert_equal(1, result.count) + + ai = result.first + assert_equal("0.0.0.0", ai.ip_address) + assert_equal(80, ai.ip_port) + assert_equal(Socket::AF_INET, ai.afamily) + assert_equal(Socket::SOCK_STREAM, ai.socktype) + end + end.join + end + + def test_addrinfo_getaddrinfo_localhost + Thread.new do + scheduler = StubScheduler.new + Fiber.set_scheduler scheduler + + Fiber.schedule do + results = Addrinfo.getaddrinfo("localhost", 80, :AF_INET, :STREAM) + assert_equal(1, results.count) + + ai = results.first + assert_equal("1.2.3.4", ai.ip_address) + end + end.join + end + + def test_addrinfo_getaddrinfo_non_existing_domain_blocking + Thread.new do + scheduler = NullScheduler.new # invoked, returns nil + Fiber.set_scheduler scheduler + + Fiber.schedule do + assert_raise(Socket::ResolutionError) { + Addrinfo.getaddrinfo("non-existing-domain.abc", nil) + } + end + end.join + end + + def test_addrinfo_getaddrinfo_no_host_non_blocking + Thread.new do + scheduler = NullScheduler.new # scheduler hook not invoked + Fiber.set_scheduler scheduler + + Fiber.schedule do + result = Addrinfo.getaddrinfo(nil, 80, :AF_INET, :STREAM) + assert_equal(1, result.count) + + ai = result.first + assert_equal("127.0.0.1", ai.ip_address) + assert_equal(80, ai.ip_port) + assert_equal(Socket::AF_INET, ai.afamily) + assert_equal(Socket::SOCK_STREAM, ai.socktype) + end + end.join + end + + def test_addrinfo_ip_domain_blocking + Thread.new do + scheduler = StubScheduler.new + Fiber.set_scheduler scheduler + + Fiber.schedule do + ai = Addrinfo.ip("example.com") + + assert_equal("1.2.3.4", ai.ip_address) + end + end.join + end + + def test_addrinfo_tcp_domain_blocking + Thread.new do + scheduler = StubScheduler.new + Fiber.set_scheduler scheduler + + Fiber.schedule do + ai = Addrinfo.tcp("example.com", 80) + + assert_equal("1.2.3.4", ai.ip_address) + assert_equal(80, ai.ip_port) + assert_equal(Socket::AF_INET, ai.afamily) + assert_equal(Socket::SOCK_STREAM, ai.socktype) + end + end.join + end + + def test_addrinfo_udp_domain_blocking + Thread.new do + scheduler = StubScheduler.new + Fiber.set_scheduler scheduler + + Fiber.schedule do + ai = Addrinfo.udp("example.com", 80) + + assert_equal("1.2.3.4", ai.ip_address) + assert_equal(80, ai.ip_port) + assert_equal(Socket::AF_INET, ai.afamily) + assert_equal(Socket::SOCK_DGRAM, ai.socktype) + end + end.join + end + + def test_ip_socket_getaddress_domain_blocking + Thread.new do + scheduler = StubScheduler.new + Fiber.set_scheduler scheduler + + Fiber.schedule do + ip = IPSocket.getaddress("example.com") + + assert_equal("1.2.3.4", ip) + end + end.join + end + + # This test "hits deep" into C function call chain. + def test_socket_getnameinfo_domain_blocking + Thread.new do + scheduler = StubScheduler.new + Fiber.set_scheduler scheduler + + Fiber.schedule do + result = Socket.getnameinfo(["AF_INET", 80, "example.com"], Socket::NI_NUMERICSERV | Socket::NI_NUMERICHOST) + + assert_equal(["1.2.3.4", "80"], result) + end + end.join + end +end diff --git a/test/fiber/test_enumerator.rb b/test/fiber/test_enumerator.rb index cd4ccd1de5..e9410f925c 100644 --- a/test/fiber/test_enumerator.rb +++ b/test/fiber/test_enumerator.rb @@ -6,16 +6,10 @@ class TestFiberEnumerator < Test::Unit::TestCase MESSAGE = "Hello World" def test_read_characters - skip "UNIXSocket is not defined!" unless defined?(UNIXSocket) + omit "UNIXSocket is not defined!" unless defined?(UNIXSocket) i, o = UNIXSocket.pair - unless i.nonblock? && o.nonblock? - i.close - o.close - skip "I/O is not non-blocking!" - end - message = String.new thread = Thread.new do @@ -48,4 +42,12 @@ class TestFiberEnumerator < Test::Unit::TestCase assert_predicate(i, :closed?) assert_predicate(o, :closed?) end + + def enumerator_fiber_is_nonblocking + enumerator = Enumerator.new do |yielder| + yielder << Fiber.current.blocking? + end + + assert_equal(false, enumerator.next) + end end diff --git a/test/fiber/test_io.rb b/test/fiber/test_io.rb index fafda7c310..0e3e086d5a 100644 --- a/test/fiber/test_io.rb +++ b/test/fiber/test_io.rb @@ -6,14 +6,12 @@ class TestFiberIO < Test::Unit::TestCase MESSAGE = "Hello World" def test_read - skip "UNIXSocket is not defined!" unless defined?(UNIXSocket) + omit unless defined?(UNIXSocket) i, o = UNIXSocket.pair - - unless i.nonblock? && o.nonblock? - i.close - o.close - skip "I/O is not non-blocking!" + if RUBY_PLATFORM=~/mswin|mingw/ + i.nonblock = true + o.nonblock = true end message = nil @@ -41,11 +39,15 @@ class TestFiberIO < Test::Unit::TestCase end def test_heavy_read - skip unless defined?(UNIXSocket) + omit unless defined?(UNIXSocket) 16.times.map do Thread.new do i, o = UNIXSocket.pair + if RUBY_PLATFORM=~/mswin|mingw/ + i.nonblock = true + o.nonblock = true + end scheduler = Scheduler.new Fiber.set_scheduler scheduler @@ -64,16 +66,11 @@ class TestFiberIO < Test::Unit::TestCase end def test_epipe_on_read - skip "UNIXSocket is not defined!" unless defined?(UNIXSocket) + omit unless defined?(UNIXSocket) + omit "nonblock=true isn't properly supported on Windows" if RUBY_PLATFORM=~/mswin|mingw/ i, o = UNIXSocket.pair - unless i.nonblock? && o.nonblock? - i.close - o.close - skip "I/O is not non-blocking!" - end - error = nil thread = Thread.new do @@ -97,4 +94,144 @@ class TestFiberIO < Test::Unit::TestCase assert_kind_of Errno::EPIPE, error end + + def test_tcp_accept + server = TCPServer.new('localhost', 0) + + th = Thread.new do + Fiber.set_scheduler(Scheduler.new) + + Fiber.schedule do + sender = server.accept + sender.wait_writable + sender.write "hello" + sender.close + end + end + + recver = TCPSocket.new('localhost', server.local_address.ip_port) + assert "hello", recver.read + + recver.close + server.close + th.join + end + + def test_tcp_connect + server = TCPServer.new('localhost', 0) + + th = Thread.new do + Fiber.set_scheduler(Scheduler.new) + + Fiber.schedule do + sender = TCPSocket.new('localhost', server.local_address.ip_port) + sender.write "hello" + sender.close + end + end + + recver = server.accept + assert "hello", recver.read + + recver.close + server.close + th.join + end + + def test_read_write_blocking + omit "UNIXSocket is not defined!" unless defined?(UNIXSocket) + + i, o = UNIXSocket.pair + i.nonblock = false + o.nonblock = false + + message = nil + + thread = Thread.new do + # This scheduler provides non-blocking `io_read`/`io_write`: + scheduler = IOBufferScheduler.new + Fiber.set_scheduler scheduler + + Fiber.schedule do + message = i.read(20) + i.close + end + + Fiber.schedule do + o.write("Hello World") + o.close + end + end + + thread.join + + assert_equal MESSAGE, message + assert_predicate(i, :closed?) + assert_predicate(o, :closed?) + end + + def test_puts_empty + omit "UNIXSocket is not defined!" unless defined?(UNIXSocket) + + i, o = UNIXSocket.pair + i.nonblock = false + o.nonblock = false + + thread = Thread.new do + # This scheduler provides non-blocking `io_read`/`io_write`: + scheduler = IOBufferScheduler.new + Fiber.set_scheduler scheduler + + Fiber.schedule do + # This was causing a segfault on older Ruby. + o.puts "" + o.puts nil + o.close + end + end + + thread.join + + message = i.read + i.close + + assert_equal $/*2, message + end + + def test_io_select + omit "UNIXSocket is not defined!" unless defined?(UNIXSocket) + + UNIXSocket.pair do |r, w| + result = nil + + thread = Thread.new do + scheduler = Scheduler.new + Fiber.set_scheduler scheduler + + Fiber.schedule do + w.write("Hello World") + result = IO.select([r], [w]) + end + end + + thread.join + + assert_equal [[r], [w], []], result + end + end + + def test_backquote + result = nil + + thread = Thread.new do + scheduler = Scheduler.new + Fiber.set_scheduler scheduler + Fiber.schedule do + result = `#{EnvUtil.rubybin} -e "sleep 0.1;puts %[ok]"` + end + end + thread.join + + assert_equal "ok\n", result + end end diff --git a/test/fiber/test_io_buffer.rb b/test/fiber/test_io_buffer.rb new file mode 100644 index 0000000000..a08b1ce1a9 --- /dev/null +++ b/test/fiber/test_io_buffer.rb @@ -0,0 +1,199 @@ +# frozen_string_literal: true +require 'test/unit' +require_relative 'scheduler' + +require 'timeout' + +class TestFiberIOBuffer < Test::Unit::TestCase + MESSAGE = "Hello World" + + def test_read_write_blocking + omit "UNIXSocket is not defined!" unless defined?(UNIXSocket) + + i, o = UNIXSocket.pair + i.nonblock = false + o.nonblock = false + + message = nil + + thread = Thread.new do + scheduler = IOBufferScheduler.new + Fiber.set_scheduler scheduler + + Fiber.schedule do + message = i.read(20) + i.close + end + + Fiber.schedule do + o.write(MESSAGE) + o.close + end + end + + thread.join + + assert_equal MESSAGE, message + assert_predicate(i, :closed?) + assert_predicate(o, :closed?) + ensure + i&.close + o&.close + end + + def test_timeout_after + omit "UNIXSocket is not defined!" unless defined?(UNIXSocket) + + i, o = UNIXSocket.pair + i.nonblock = false + o.nonblock = false + + message = nil + error = nil + + thread = Thread.new do + scheduler = IOBufferScheduler.new + Fiber.set_scheduler scheduler + + Fiber.schedule do + Timeout.timeout(0.1) do + message = i.read(20) + end + rescue Timeout::Error => error + # Assertions below. + ensure + i.close + end + end + + thread.join + + assert_nil message + assert_kind_of Timeout::Error, error + ensure + i&.close + o&.close + end + + def test_read_nonblock + omit "UNIXSocket is not defined!" unless defined?(UNIXSocket) + + i, o = UNIXSocket.pair + + message = nil + + thread = Thread.new do + scheduler = IOBufferScheduler.new + Fiber.set_scheduler scheduler + + Fiber.schedule do + message = i.read_nonblock(20, exception: false) + i.close + end + end + + thread.join + + assert_equal :wait_readable, message + ensure + i&.close + o&.close + end + + def test_write_nonblock + omit "UNIXSocket is not defined!" unless defined?(UNIXSocket) + + i, o = UNIXSocket.pair + + thread = Thread.new do + scheduler = IOBufferScheduler.new + Fiber.set_scheduler scheduler + + Fiber.schedule do + o.write_nonblock(MESSAGE, exception: false) + o.close + end + end + + thread.join + + assert_equal MESSAGE, i.read + ensure + i&.close + o&.close + end + + def test_io_buffer_read_write + omit "UNIXSocket is not defined!" unless defined?(UNIXSocket) + + i, o = UNIXSocket.pair + source_buffer = IO::Buffer.for("Hello World!") + destination_buffer = IO::Buffer.new(source_buffer.size) + + # Test non-scheduler code path: + source_buffer.write(o, source_buffer.size) + destination_buffer.read(i, source_buffer.size) + assert_equal source_buffer, destination_buffer + + # Test scheduler code path: + destination_buffer.clear + + thread = Thread.new do + scheduler = IOBufferScheduler.new + Fiber.set_scheduler scheduler + + Fiber.schedule do + source_buffer.write(o, source_buffer.size) + destination_buffer.read(i, source_buffer.size) + end + end + + thread.join + + assert_equal source_buffer, destination_buffer + ensure + i&.close + o&.close + end + + def nonblockable?(io) + io.nonblock{} + true + rescue + false + end + + def test_io_buffer_pread_pwrite + file = Tempfile.new("test_io_buffer_pread_pwrite") + + omit "Non-blocking file IO is not supported" unless nonblockable?(file) + + source_buffer = IO::Buffer.for("Hello World!") + destination_buffer = IO::Buffer.new(source_buffer.size) + + # Test non-scheduler code path: + source_buffer.pwrite(file, 1, source_buffer.size) + destination_buffer.pread(file, 1, source_buffer.size) + assert_equal source_buffer, destination_buffer + + # Test scheduler code path: + destination_buffer.clear + file.truncate(0) + + thread = Thread.new do + scheduler = IOBufferScheduler.new + Fiber.set_scheduler scheduler + + Fiber.schedule do + source_buffer.pwrite(file, 1, source_buffer.size) + destination_buffer.pread(file, 1, source_buffer.size) + end + end + + thread.join + + assert_equal source_buffer, destination_buffer + ensure + file&.close! + end +end diff --git a/test/fiber/test_mutex.rb b/test/fiber/test_mutex.rb index 0842427760..2cee2cc235 100644 --- a/test/fiber/test_mutex.rb +++ b/test/fiber/test_mutex.rb @@ -4,7 +4,7 @@ require_relative 'scheduler' class TestFiberMutex < Test::Unit::TestCase def test_mutex_synchronize - mutex = Mutex.new + mutex = Thread::Mutex.new thread = Thread.new do scheduler = Scheduler.new @@ -23,7 +23,7 @@ class TestFiberMutex < Test::Unit::TestCase end def test_mutex_interleaved_locking - mutex = Mutex.new + mutex = Thread::Mutex.new thread = Thread.new do scheduler = Scheduler.new @@ -48,7 +48,7 @@ class TestFiberMutex < Test::Unit::TestCase end def test_mutex_thread - mutex = Mutex.new + mutex = Thread::Mutex.new mutex.lock thread = Thread.new do @@ -71,7 +71,7 @@ class TestFiberMutex < Test::Unit::TestCase end def test_mutex_fiber_raise - mutex = Mutex.new + mutex = Thread::Mutex.new ran = false main = Thread.new do @@ -103,8 +103,8 @@ class TestFiberMutex < Test::Unit::TestCase end def test_condition_variable - mutex = Mutex.new - condition = ConditionVariable.new + mutex = Thread::Mutex.new + condition = Thread::ConditionVariable.new signalled = 0 @@ -138,7 +138,7 @@ class TestFiberMutex < Test::Unit::TestCase end def test_queue - queue = Queue.new + queue = Thread::Queue.new processed = 0 thread = Thread.new do @@ -169,7 +169,7 @@ class TestFiberMutex < Test::Unit::TestCase end def test_queue_pop_waits - queue = Queue.new + queue = Thread::Queue.new running = false thread = Thread.new do @@ -194,11 +194,11 @@ class TestFiberMutex < Test::Unit::TestCase end def test_mutex_deadlock - error_pattern = /No live threads left. Deadlock\?/ + error_pattern = /lock already owned by another fiber/ assert_in_out_err %W[-I#{__dir__} -], <<-RUBY, ['in synchronize'], error_pattern, success: false require 'scheduler' - mutex = Mutex.new + mutex = Thread::Mutex.new thread = Thread.new do scheduler = Scheduler.new @@ -207,7 +207,7 @@ class TestFiberMutex < Test::Unit::TestCase Fiber.schedule do mutex.synchronize do puts 'in synchronize' - Fiber.yield + scheduler.transfer end end @@ -217,4 +217,24 @@ class TestFiberMutex < Test::Unit::TestCase thread.join RUBY end + + def test_mutex_fiber_deadlock_no_scheduler + thr = Thread.new do + loop do + sleep 1 + end + end + + mutex = Mutex.new + mutex.synchronize do + error = assert_raise ThreadError do + Fiber.new do + mutex.lock + end.resume + end + assert_includes error.message, "deadlock; lock already owned by another fiber belonging to the same thread" + end + ensure + thr&.kill&.join + end end diff --git a/test/fiber/test_process.rb b/test/fiber/test_process.rb index c6583cac9b..a09b070c0a 100644 --- a/test/fiber/test_process.rb +++ b/test/fiber/test_process.rb @@ -3,13 +3,15 @@ require 'test/unit' require_relative 'scheduler' class TestFiberProcess < Test::Unit::TestCase + TRUE_CMD = RUBY_PLATFORM =~ /mswin|mingw/ ? "exit 0" : "true" + def test_process_wait Thread.new do scheduler = Scheduler.new Fiber.set_scheduler scheduler Fiber.schedule do - pid = Process.spawn("true") + pid = Process.spawn(TRUE_CMD) Process.wait(pid) # TODO test that scheduler was invoked. @@ -25,7 +27,7 @@ class TestFiberProcess < Test::Unit::TestCase Fiber.set_scheduler scheduler Fiber.schedule do - system("true") + system(TRUE_CMD) # TODO test that scheduler was invoked (currently it's not). @@ -33,4 +35,40 @@ class TestFiberProcess < Test::Unit::TestCase end end.join end + + def test_system_faulty_process_wait + Thread.new do + scheduler = Scheduler.new + + def scheduler.process_wait(pid, flags) + Fiber.blocking{Process.wait(pid, flags)} + + # Don't return `Process::Status` instance. + return false + end + + Fiber.set_scheduler scheduler + + Fiber.schedule do + assert_raise TypeError do + system(TRUE_CMD) + end + end + end.join + end + + def test_fork + omit 'fork not supported' unless Process.respond_to?(:fork) + Thread.new do + scheduler = Scheduler.new + Fiber.set_scheduler scheduler + + Fiber.schedule do + pid = Process.fork {} + Process.wait(pid) + + assert_predicate $?, :success? + end + end.join + end end diff --git a/test/fiber/test_queue.rb b/test/fiber/test_queue.rb new file mode 100644 index 0000000000..d78b026f11 --- /dev/null +++ b/test/fiber/test_queue.rb @@ -0,0 +1,54 @@ +# frozen_string_literal: true +require 'test/unit' +require_relative 'scheduler' + +class TestFiberQueue < Test::Unit::TestCase + def test_pop_with_timeout + queue = Thread::Queue.new + kill = false + result = :unspecified + + thread = Thread.new do + scheduler = Scheduler.new + Fiber.set_scheduler(scheduler) + + Fiber.schedule do + result = queue.pop(timeout: 0.0001) + end + + scheduler.run + end + until thread.join(2) + kill = true + thread.kill + end + + assert_false(kill, 'Getting stuck due to a possible compiler bug.') + assert_nil result + end + + def test_pop_with_timeout_and_value + queue = Thread::Queue.new + queue.push(:something) + kill = false + result = :unspecified + + thread = Thread.new do + scheduler = Scheduler.new + Fiber.set_scheduler(scheduler) + + Fiber.schedule do + result = queue.pop(timeout: 0.0001) + end + + scheduler.run + end + until thread.join(2) + kill = true + thread.kill + end + + assert_false(kill, 'Getting stuck due to a possible compiler bug.') + assert_equal :something, result + end +end diff --git a/test/fiber/test_ractor.rb b/test/fiber/test_ractor.rb index d03455a9f7..3c4ccbd8e5 100644 --- a/test/fiber/test_ractor.rb +++ b/test/fiber/test_ractor.rb @@ -4,7 +4,7 @@ require "fiber" class TestFiberCurrentRactor < Test::Unit::TestCase def setup - skip unless defined? Ractor + omit unless defined? Ractor end def test_ractor_shareable diff --git a/test/fiber/test_scheduler.rb b/test/fiber/test_scheduler.rb index eeb0d67ec5..34effad816 100644 --- a/test/fiber/test_scheduler.rb +++ b/test/fiber/test_scheduler.rb @@ -27,6 +27,18 @@ class TestFiberScheduler < Test::Unit::TestCase refute f.blocking? end + def test_fiber_blocking + f = Fiber.new(blocking: false) do + fiber = Fiber.current + refute fiber.blocking? + Fiber.blocking do |_fiber| + assert_equal fiber, _fiber + assert fiber.blocking? + end + end + f.resume + end + def test_closed_at_thread_exit scheduler = Scheduler.new @@ -55,6 +67,7 @@ class TestFiberScheduler < Test::Unit::TestCase def test_close_at_exit assert_in_out_err %W[-I#{__dir__} -], <<-RUBY, ['Running Fiber'], [], success: true require 'scheduler' + Warning[:experimental] = false scheduler = Scheduler.new Fiber.set_scheduler scheduler @@ -66,9 +79,23 @@ class TestFiberScheduler < Test::Unit::TestCase RUBY end - def test_optional_close + def test_minimal_interface + scheduler = Object.new + + def scheduler.block + end + + def scheduler.unblock + end + + def scheduler.io_wait + end + + def scheduler.kernel_sleep + end + thread = Thread.new do - Fiber.set_scheduler Object.new + Fiber.set_scheduler scheduler end thread.join @@ -89,4 +116,70 @@ class TestFiberScheduler < Test::Unit::TestCase thread.join end + + def test_autoload + 10.times do + Object.autoload(:TestFiberSchedulerAutoload, File.expand_path("autoload.rb", __dir__)) + + thread = Thread.new do + scheduler = Scheduler.new + Fiber.set_scheduler scheduler + + 10.times do + Fiber.schedule do + Object.const_get(:TestFiberSchedulerAutoload) + end + end + end + + thread.join + ensure + $LOADED_FEATURES.delete(File.expand_path("autoload.rb", __dir__)) + Object.send(:remove_const, :TestFiberSchedulerAutoload) + end + end + + def test_deadlock + mutex = Thread::Mutex.new + condition = Thread::ConditionVariable.new + q = 0.0001 + + signaller = Thread.new do + loop do + mutex.synchronize do + condition.signal + end + sleep q + end + end + + i = 0 + + thread = Thread.new do + scheduler = SleepingBlockingScheduler.new + Fiber.set_scheduler scheduler + + Fiber.schedule do + 10.times do + mutex.synchronize do + condition.wait(mutex) + sleep q + i += 1 + end + end + end + end + + # Wait for 10 seconds at most... if it doesn't finish, it's deadlocked. + thread.join(10) + + # If it's deadlocked, it will never finish, so this will be 0. + assert_equal 10, i + ensure + # Make sure the threads are dead... + thread.kill + signaller.kill + thread.join + signaller.join + end end diff --git a/test/fiber/test_sleep.rb b/test/fiber/test_sleep.rb index e882766345..a7e88c0367 100644 --- a/test/fiber/test_sleep.rb +++ b/test/fiber/test_sleep.rb @@ -43,4 +43,29 @@ class TestFiberSleep < Test::Unit::TestCase assert_operator seconds, :>=, 2, "actual: %p" % seconds end + + def test_broken_sleep + thread = Thread.new do + Thread.current.report_on_exception = false + + scheduler = Scheduler.new + + def scheduler.kernel_sleep(duration = nil) + raise "Broken sleep!" + end + + Fiber.set_scheduler scheduler + + Fiber.schedule do + sleep 0 + end + + ensure + scheduler.close + end + + assert_raise(RuntimeError) do + thread.join + end + end end diff --git a/test/fiber/test_storage.rb b/test/fiber/test_storage.rb new file mode 100644 index 0000000000..3726decbdb --- /dev/null +++ b/test/fiber/test_storage.rb @@ -0,0 +1,115 @@ +# frozen_string_literal: true +require 'test/unit' + +class TestFiberStorage < Test::Unit::TestCase + def test_storage + Fiber.new do + Fiber[:x] = 10 + assert_kind_of Hash, Fiber.current.storage + assert_predicate Fiber.current.storage, :any? + end.resume + end + + def test_storage_inherited + Fiber.new do + Fiber[:foo] = :bar + + Fiber.new do + assert_equal :bar, Fiber[:foo] + Fiber[:bar] = :baz + end.resume + + assert_nil Fiber[:bar] + end.resume + end + + def test_variable_assignment + Fiber.new do + Fiber[:foo] = :bar + assert_equal :bar, Fiber[:foo] + end.resume + end + + def test_storage_assignment + old, Warning[:experimental] = Warning[:experimental], false + + Fiber.new do + Fiber.current.storage = {foo: :bar} + assert_equal :bar, Fiber[:foo] + end.resume + ensure + Warning[:experimental] = old + end + + def test_storage_only_allow_access_from_same_fiber + old, Warning[:experimental] = Warning[:experimental], false + + f = Fiber.new do + Fiber[:a] = 1 + end + assert_raise(ArgumentError) { f.storage } + assert_raise(ArgumentError) { f.storage = {} } + ensure + Warning[:experimental] = old + end + + def test_inherited_storage + Fiber.new(storage: {foo: :bar}) do + f = Fiber.new do + assert_equal :bar, Fiber[:foo] + end + f.resume + end.resume + end + + def test_enumerator_inherited_storage + Fiber.new do + Fiber[:item] = "Hello World" + + enumerator = Enumerator.new do |out| + out << Fiber.current + out << Fiber[:item] + end + + # The fiber within the enumerator is not equal to the current... + assert_not_equal Fiber.current, enumerator.next + + # But it inherited the storage from the current fiber: + assert_equal "Hello World", enumerator.next + end.resume + end + + def test_thread_inherited_storage + Fiber.new do + Fiber[:x] = 10 + + x = Thread.new do + Fiber[:y] = 20 + Fiber[:x] + end.value + + assert_equal 10, x + assert_equal nil, Fiber[:y] + end.resume + end + + def test_enumerator_count + Fiber.new do + Fiber[:count] = 0 + + enumerator = Enumerator.new do |y| + Fiber[:count] += 1 + y << Fiber[:count] + end + + assert_equal 1, enumerator.next + assert_equal 0, Fiber[:count] + end.resume + end + + def test_storage_assignment_type_error + assert_raise(TypeError) do + Fiber.new(storage: {Object.new => "bar"}) {} + end + end +end diff --git a/test/fiber/test_thread.rb b/test/fiber/test_thread.rb index 5fc80f0e6c..5e3cc6d0e1 100644 --- a/test/fiber/test_thread.rb +++ b/test/fiber/test_thread.rb @@ -20,6 +20,53 @@ class TestFiberThread < Test::Unit::TestCase assert_equal :done, thread.value end + def test_thread_join_timeout + sleeper = nil + + thread = Thread.new do + scheduler = Scheduler.new + Fiber.set_scheduler scheduler + + Fiber.schedule do + sleeper = Thread.new{sleep} + sleeper.join(0.1) + end + + scheduler.run + end + + thread.join + + assert_predicate sleeper, :alive? + ensure + sleeper&.kill&.join + end + + def test_thread_join_implicit + sleeping = false + finished = false + + thread = Thread.new do + scheduler = Scheduler.new + Fiber.set_scheduler scheduler + + Fiber.schedule do + sleeping = true + sleep(0.1) + finished = true + end + + :done + end + + Thread.pass until sleeping + + thread.join + + assert_equal :done, thread.value + assert finished, "Scheduler thread's task should be finished!" + end + def test_thread_join_blocking thread = Thread.new do scheduler = Scheduler.new @@ -42,4 +89,42 @@ class TestFiberThread < Test::Unit::TestCase assert_equal :done, thread.value end + + def test_broken_unblock + thread = Thread.new do + Thread.current.report_on_exception = false + + scheduler = BrokenUnblockScheduler.new + + Fiber.set_scheduler scheduler + + Fiber.schedule do + Thread.new{ + Thread.current.report_on_exception = false + }.join + end + + scheduler.run + ensure + scheduler.close + end + + assert_raise(RuntimeError) do + thread.join + end + end + + def test_thread_join_hang + thread = Thread.new do + scheduler = SleepingUnblockScheduler.new + + Fiber.set_scheduler scheduler + + Fiber.schedule do + Thread.new{sleep(0.01)}.value + end + end + + thread.join + end end |