Opt-in for RFC 6724-compliant host resolution

lib/async/scheduler.rb

@@ -11,6 +11,7 @@
 require_relative "task"
 require_relative "timeout"
 require_relative "fork_handler"
+require_relative "semaphore"
 
 require "io/event"
 
@@ -32,6 +33,12 @@ class Scheduler < Node
 			value == "true" ? true : nil
 		end
 
+		# Whether schedulers resolve hostnames with the platform resolver by default. Set `ASYNC_SCHEDULER_PLATFORM_RESOLVER=true` to enable.
+		PLATFORM_RESOLVER = ENV.fetch("ASYNC_SCHEDULER_PLATFORM_RESOLVER", nil) == "true"
+
+		# The default maximum number of callers waiting on platform resolver lookups at once, per scheduler. Each lookup runs on its own thread.
+		PLATFORM_RESOLVER_LIMIT = 16
+
 		# Raised when an operation is attempted on a closed scheduler.
 		class ClosedError < RuntimeError
 			# Create a new error.
@@ -75,7 +82,16 @@ def blocking_operation_wait(work)
 		# @public Since *Async v1*.
 		# @parameter parent [Node | Nil] The parent node to use for task hierarchy.
 		# @parameter selector [IO::Event::Selector] The selector to use for event handling.
-		def initialize(parent = nil, selector: nil, profiler: Profiler&.default, worker_pool: WORKER_POOL)
+		# @parameter platform_resolver [Boolean | Integer] Whether to resolve hostnames with the platform resolver (`getaddrinfo`) instead of `Resolv`. An integer enables it and sets the maximum number of callers waiting on lookups at once (default {PLATFORM_RESOLVER_LIMIT}).
+		def initialize(parent = nil, selector: nil, profiler: Profiler&.default, worker_pool: WORKER_POOL, platform_resolver: PLATFORM_RESOLVER)
+			# Validate the limit before allocating the selector or worker pool, so that an invalid argument doesn't leak them:
+			if platform_resolver
+				limit = platform_resolver == true ? PLATFORM_RESOLVER_LIMIT : Integer(platform_resolver)
+				Kernel.raise ArgumentError, "platform_resolver limit must be positive!" unless limit.positive?
+
+				address_resolve_semaphore = Semaphore.new(limit)
+			end
+
 			super(parent)
 
 			@selector = selector || ::IO::Event::Selector.new(Fiber.current)
@@ -99,6 +115,9 @@ def initialize(parent = nil, selector: nil, profiler: Profiler&.default, worker_
 			if @worker_pool
 				self.singleton_class.prepend(BlockingOperationWait)
 			end
+
+			# If set, hostnames are resolved with the platform resolver, and this limits how many callers wait on lookups at once:
+			@address_resolve_semaphore = address_resolve_semaphore
 		end
 
 		# Compute the scheduler load according to the busy and idle times that are updated by the run loop.
@@ -288,15 +307,44 @@ def kernel_sleep(duration = nil)
 
 		# Resolve the address of the given hostname.
 		#
+		# By default this uses `Resolv`, which only reads `/etc/hosts` and `/etc/resolv.conf`. With `platform_resolver:` (or `ASYNC_SCHEDULER_PLATFORM_RESOLVER=true`), it uses the platform resolver (`getaddrinfo`) on a short-lived thread instead, which honours the same configuration as code running outside the scheduler (e.g. `/etc/resolver` on macOS, `nsswitch.conf`, RFC 6724 address ordering), without blocking the event loop. At most {PLATFORM_RESOLVER_LIMIT} callers (or the limit given to `platform_resolver:`) wait on platform lookups at once per scheduler; further callers wait for a free slot. A caller that is stopped or times out frees its slot immediately, but `getaddrinfo` can't be cancelled, so the abandoned lookup may keep running until the system resolver gives up, just as it would outside the scheduler.
+		#
 		# @public Since *Async v2*.
 		# @asynchronous May be non-blocking.
 		#
 		# @parameter hostname [String] The hostname to resolve.
+		# @returns [Array(String)] The resolved addresses.
 		def address_resolve(hostname)
 			# On some platforms, hostnames may contain a device-specific suffix (e.g. %en0). We need to strip this before resolving.
 			# See <https://github.com/socketry/async/issues/180> for more details.
 			hostname = hostname.split("%", 2).first
-			::Resolv.getaddresses(hostname)
+
+			if semaphore = @address_resolve_semaphore
+				platform_address_resolve(hostname, semaphore)
+			else
+				::Resolv.getaddresses(hostname)
+			end
+		end
+
+		# Resolve the given hostname with `getaddrinfo` on its own thread, holding a slot in the given semaphore while the caller waits.
+		private def platform_address_resolve(hostname, semaphore)
+			semaphore.acquire do
+				thread = ::Thread.new do
+					::Addrinfo.getaddrinfo(hostname, nil, nil, :STREAM).map(&:ip_address).uniq
+				rescue => error
+					# Return the error rather than raising it, otherwise `Thread.abort_on_exception` would re-raise it on the main thread:
+					error
+				end
+
+				result = thread.value
+
+				::Kernel.raise result if result.is_a?(::Exception)
+
+				result
+			ensure
+				# If the caller is stopped or times out, don't leave the Ruby thread waiting. This interrupts the wait, but the native `getaddrinfo` call can't be cancelled and runs until the system resolver gives up:
+				thread&.kill
+			end
 		end
 
 		# Wait for the specified IO to become ready for the specified events.

releases.md

@@ -1,5 +1,9 @@
 # Releases
 
+## Unreleased
+
+  - Add `ASYNC_SCHEDULER_PLATFORM_RESOLVER=true` (or `Scheduler.new(platform_resolver: true)`) to resolve hostnames with the platform resolver (`getaddrinfo`) on a short-lived thread instead of `Resolv`. Lookups inside the scheduler then honour the same configuration as lookups outside it, e.g. `/etc/resolver` on macOS, `nsswitch.conf`, and RFC 6724 address ordering, without blocking the event loop. At most 16 callers wait on lookups at once per scheduler; pass an integer (e.g. `platform_resolver: 4`) to change the limit. `getaddrinfo` can't be cancelled, so a lookup whose caller is stopped or times out may keep running until the system resolver gives up, as it would outside the scheduler.
+
 ## v2.46.0
 
   - Fixed `Thread#kill` raising a `TypeError` while shutting down a thread with an active `Async::Reactor`.

test/async/scheduler/address.rb

@@ -33,5 +33,193 @@
 
 			expect(address.ipv6?).to be == true
 		end
+
+		it "can resolve hostnames with device suffix" do
+			addresses = Addrinfo.getaddrinfo("localhost%lo0", "80")
+
+			expect(addresses).not.to be(:empty?)
+		end
+
+		it "raises an error for unknown hostnames" do
+			expect do
+				Addrinfo.getaddrinfo("async.invalid", "80")
+			end.to raise_exception(SocketError)
+		end
+	end
+
+	with "platform_resolver: false" do
+		it "resolves with Resolv" do
+			mock(::Resolv) do |mock|
+				mock.replace(:getaddresses) do |hostname|
+					["192.0.2.1"]
+				end
+			end
+
+			default = Async::Scheduler.new(platform_resolver: false)
+			expect(default.address_resolve("example.invalid")).to be == ["192.0.2.1"]
+		ensure
+			default&.close
+		end
+	end
+
+	with "platform_resolver: true" do
+		def resolve(hostname, platform_resolver: true)
+			platform = Async::Scheduler.new(platform_resolver: platform_resolver)
+			platform.address_resolve(hostname)
+		ensure
+			platform&.close
+		end
+
+		# Mocks are scoped to the current thread, but the lookup runs on its own thread, so replace `Addrinfo.getaddrinfo` globally. The block only sees lookups made without a fiber scheduler, i.e. on the resolver thread:
+		def stub_getaddrinfo(stub)
+			Addrinfo.singleton_class.alias_method(:original_getaddrinfo, :getaddrinfo)
+			Addrinfo.define_singleton_method(:getaddrinfo) do |hostname, *arguments, **options|
+				if Fiber.scheduler
+					original_getaddrinfo(hostname, *arguments, **options)
+				else
+					stub.call(hostname, *arguments, **options)
+				end
+			end
+
+			yield
+		ensure
+			Addrinfo.singleton_class.alias_method(:getaddrinfo, :original_getaddrinfo)
+			Addrinfo.singleton_class.remove_method(:original_getaddrinfo)
+		end
+
+		it "resolves hostnames" do
+			expect(resolve("localhost")).not.to be(:empty?)
+		end
+
+		it "strips a device suffix" do
+			expect(resolve("localhost%lo0")).not.to be(:empty?)
+		end
+
+		it "raises an error for unknown hostnames" do
+			expect do
+				resolve("async.invalid")
+			end.to raise_exception(SocketError)
+		end
+
+		it "resolves through Ruby's address_resolve hook" do
+			lookups = Thread::Queue.new
+
+			stub_getaddrinfo(->(hostname, *arguments, **options) do
+				lookups << hostname
+				Addrinfo.original_getaddrinfo(hostname, *arguments, **options)
+			end) do
+				thread = Thread.new do
+					scheduler = Async::Scheduler.new(platform_resolver: true)
+					Fiber.set_scheduler(scheduler)
+
+					scheduler.run do
+						Addrinfo.getaddrinfo("localhost", 80, nil, :STREAM)
+					end.wait
+				ensure
+					Fiber.set_scheduler(nil)
+				end
+
+				addresses = thread.value
+
+				expect(addresses).not.to be(:empty?)
+				expect(addresses.map(&:ip_port).uniq).to be == [80]
+				expect(lookups.size).to be == 1
+				expect(lookups.pop).to be == "localhost"
+			end
+		end
+
+		it "does not re-raise errors on the main thread when threads abort on exception" do
+			abort_on_exception = Thread.abort_on_exception
+			Thread.abort_on_exception = true
+
+			expect do
+				resolve("async.invalid")
+			end.to raise_exception(SocketError)
+
+			# An aborting thread raises asynchronously in the main thread; give it a chance to arrive:
+			sleep(0.05)
+		ensure
+			Thread.abort_on_exception = abort_on_exception
+		end
+
+		it "exits the lookup's Ruby thread when the caller is stopped" do
+			started = Thread::Queue.new
+
+			platform = Async::Scheduler.new(platform_resolver: true)
+
+			stub_getaddrinfo(->(hostname, *arguments, **options) do
+				started << Thread.current
+				sleep(10)
+			end) do
+				task = reactor.async do
+					platform.address_resolve("slow.invalid")
+				end
+
+				lookup = started.pop
+				task.stop
+
+				# This only shows the Ruby thread exits; a native `getaddrinfo` call can't be interrupted and keeps running:
+				expect(lookup.join(1)).to be == lookup
+			end
+		ensure
+			platform&.close
+		end
+
+		it "limits the number of concurrent lookups" do
+			mutex = Thread::Mutex.new
+			active = 0
+			maximum = 0
+
+			platform = Async::Scheduler.new(platform_resolver: 2)
+
+			stub_getaddrinfo(->(hostname, *arguments, **options) do
+				mutex.synchronize{maximum = [maximum, active += 1].max}
+				sleep(0.05)
+				mutex.synchronize{active -= 1}
+				Addrinfo.original_getaddrinfo("localhost", *arguments, **options)
+			end) do
+				tasks = 5.times.map do
+					reactor.async{platform.address_resolve("slow.invalid")}
+				end
+
+				tasks.each{|task| expect(task.wait).not.to be(:empty?)}
+			end
+
+			expect(maximum).to be == 2
+		ensure
+			platform&.close
+		end
+
+		it "rejects a non-positive limit before allocating a selector" do
+			mock(::IO::Event::Selector) do |mock|
+				mock.replace(:new) do |*arguments|
+					Kernel.raise "Allocated a selector for an invalid scheduler!"
+				end
+			end
+
+			expect do
+				Async::Scheduler.new(platform_resolver: 0)
+			end.to raise_exception(ArgumentError)
+		end
+
+		it "does not block the event loop while resolving" do
+			stub_getaddrinfo(->(hostname, *arguments, **options) do
+				sleep(0.1)
+				Addrinfo.original_getaddrinfo("localhost", *arguments, **options)
+			end) do
+				ticks = 0
+				ticker = reactor.async do
+					loop do
+						ticks += 1
+						sleep(0.01)
+					end
+				end
+
+				expect(resolve("slow.invalid")).not.to be(:empty?)
+				expect(ticks).to be >= 5
+			ensure
+				ticker&.stop
+			end
+		end
 	end
 end