diff --git a/docs/fork-safety.md b/docs/fork-safety.md index c0a2a52..e585ae7 100644 --- a/docs/fork-safety.md +++ b/docs/fork-safety.md @@ -26,6 +26,42 @@ response, body sender, and cookie jar belongs to the process that created it. Using an inherited object raises `Wreq::ForkError`, even when the parent never started the runtime. wreq-ruby does not rebuild these objects. +## Concurrent prefork workers on Linux + +The [`multiprocess_client.rb`](../test/scripts/multiprocess_client.rb) regression +test covers a Linux prefork layout in which the parent loads wreq-ruby before it +starts the workers, but does not create a client or initialize the request +runtime. The master is clean and single-threaded at the fork boundary. The local +HTTP server is forked before `require "wreq"`, so the server process never +inherits the extension or any of its native state. + +The test uses two barriers. The first releases four workers to create a fresh +`Wreq::Client` in each process. After all four workers report that their client +exists, the second barrier releases them together to send requests. Each worker +uses the same fresh client for two requests. After every worker exits, the +parent creates its own fresh client and sends one final request. + +This works because the parent has not initialized the Tokio runtime when the +workers fork. Under copy-on-write process semantics, each child initializes a +separate runtime and its threads on its first request, in its own address space. +`ProcessLocal` rejects native-backed objects inherited from another process; it +does not stop separate children from creating their own objects and runtime +after the fork. + +The test completed 10 consecutive runs on WSL2 x86_64 with Ruby 3.4.8. Those +runs covered 40 worker processes, 80 worker requests, and 10 parent requests. +They produced no failures, deadlocks, or `Wreq::ForkError` exceptions. +The complete fork test file also passed with 4 runs and 100 assertions. + +The test server replies with `Connection: close`, so this result does not prove +connection pooling or connection reuse across requests. It verifies that +several prefork workers can create process-local clients and send requests when +the parent has only loaded the extension. If the parent initializes the runtime +before forking, the child remains unsupported even if it creates a new client. +The server waits on a release pipe after the final request so that its `SIGCHLD` +does not interrupt a no-GVL request that is still returning. This pipe is test +harness coordination, not an application requirement. + ## Forking after runtime initialization Once the parent starts an HTTP operation or otherwise uses the Tokio runtime, a diff --git a/test/fork_test.rb b/test/fork_test.rb index 7a772fb..ad84ac2 100644 --- a/test/fork_test.rb +++ b/test/fork_test.rb @@ -48,6 +48,21 @@ def test_loaded_extension_can_initialize_runtime_after_fork refute_match(/\[BUG\]|segmentation fault|panicked/i, stderr) end + def test_fresh_clients_can_request_from_concurrent_forked_workers + skip "this regression test is Linux-only" unless RbConfig::CONFIG.fetch("host_os").include?("linux") + skip "fork is not supported on this platform" unless Process.respond_to?(:fork) + + stdout, stderr, status = run_fork_script("multiprocess_client.rb", timeout: 60) + + assert status.success?, "subprocess failed with #{status.inspect}: #{stderr}" + assert_equal "ok\n", stdout + 4.times do |worker| + assert_match(/^worker_#{worker}=ok$/, stderr) + end + assert_match(/^parent_after_workers=ok$/, stderr) + refute_match(/Wreq::ForkError|\[BUG\]|segmentation fault|panicked/i, stderr) + end + def test_initialized_runtime_is_rejected_after_fork skip "fork is not supported on this platform" unless Process.respond_to?(:fork) @@ -67,7 +82,7 @@ def test_initialized_runtime_is_rejected_after_fork private - def run_fork_script(name) + def run_fork_script(name, timeout: 30) lib_dir = File.expand_path("../lib", __dir__) script = File.expand_path("scripts/#{name}", __dir__) @@ -82,25 +97,29 @@ def run_fork_script(name) err: stderr, pgroup: true ) - status = Timeout.timeout(30) { Process.wait2(pid).last } + status = Timeout.timeout(timeout) { Process.wait2(pid).last } + kill_process_group(pid) unless status.success? stdout.rewind stderr.rewind return [stdout.read, stderr.read, status] rescue Timeout::Error - begin - Process.kill("KILL", -pid) - rescue Errno::ESRCH - nil - end - - begin - Process.wait(pid) - rescue Errno::ECHILD - nil - end - + kill_process_group(pid) flunk "#{name} timed out" end end end + + def kill_process_group(pid) + begin + Process.kill("KILL", -pid) + rescue Errno::ESRCH + nil + end + + begin + Process.wait(pid) + rescue Errno::ECHILD + nil + end + end end diff --git a/test/scripts/multiprocess_client.rb b/test/scripts/multiprocess_client.rb new file mode 100644 index 0000000..5470aa3 --- /dev/null +++ b/test/scripts/multiprocess_client.rb @@ -0,0 +1,161 @@ +# frozen_string_literal: true + +require "rbconfig" +require "socket" +require "timeout" + +$stdout.sync = true +$stderr.sync = true + +WORKER_COUNT = 4 +REQUESTS_PER_WORKER = 2 +PARENT_REQUEST_PATH = "/parent/request/0" + +abort "multiprocess client test requires Linux" unless RbConfig::CONFIG.fetch("host_os").include?("linux") + +server = TCPServer.new("127.0.0.1", 0) +url = "http://127.0.0.1:#{server.addr[1]}" +server_release_reader, server_release_writer = IO.pipe +server_pid = fork do + server_release_writer.close + Timeout.timeout(55) do + observed_paths = Array.new((WORKER_COUNT * REQUESTS_PER_WORKER) + 1) do + socket = server.accept + begin + request_line = socket.gets + abort "server received an incomplete request" unless request_line + + while (line = socket.gets) + break if line == "\r\n" + end + + method, path, = request_line.split(" ", 3) + abort "server received an unexpected method: #{method.inspect}" unless method == "GET" + + socket.write( + "HTTP/1.1 200 OK\r\n" \ + "Content-Length: #{path.bytesize}\r\n" \ + "Connection: close\r\n\r\n" \ + "#{path}" + ) + path + ensure + socket.close + end + end + + expected_paths = WORKER_COUNT.times.flat_map do |worker| + REQUESTS_PER_WORKER.times.map { |request| "/worker/#{worker}/request/#{request}" } + end + expected_paths << PARENT_REQUEST_PATH + abort "server received unexpected paths: #{observed_paths.inspect}" unless observed_paths.sort == expected_paths.sort + abort "server did not receive its release signal" unless server_release_reader.read(1) == "." + end + exit! 0 +rescue => error + warn "server=unexpected #{error.class}: #{error.message}" + exit! 3 +ensure + server_release_reader.close + server.close +end +server.close +server_release_reader.close + +# The parent only loads the extension. It does not create a Client or initialize +# the request runtime before forking the workers. +require "wreq" + +client_start_reader, client_start_writer = IO.pipe +client_ready_reader, client_ready_writer = IO.pipe +request_start_reader, request_start_writer = IO.pipe +worker_pids = WORKER_COUNT.times.map do |worker| + fork do + server_release_writer.close + client_start_writer.close + client_ready_reader.close + request_start_writer.close + + abort "worker #{worker} did not receive its client signal" unless client_start_reader.read(1) == "." + client_start_reader.close + + client = Wreq::Client.new(no_proxy: true, http1_only: true, timeout: 5) + client_ready_writer.write(".") + client_ready_writer.close + + abort "worker #{worker} did not receive its request signal" unless request_start_reader.read(1) == "." + request_start_reader.close + + Timeout.timeout(30) do + REQUESTS_PER_WORKER.times do |request| + path = "/worker/#{worker}/request/#{request}" + response = client.get("#{url}#{path}") + abort "worker #{worker} request #{request} failed" unless response.bytes == path + end + end + + warn "worker_#{worker}=ok" + exit! 0 + rescue => error + warn "worker_#{worker}=unexpected #{error.class}: #{error.message}" + exit! 2 + ensure + client_start_reader.close unless client_start_reader.closed? + client_ready_writer.close unless client_ready_writer.closed? + request_start_reader.close unless request_start_reader.closed? + end +end + +client_start_reader.close +client_ready_writer.close +request_start_reader.close + +client_start_writer.write("." * WORKER_COUNT) +client_start_writer.close + +ready_workers = client_ready_reader.read(WORKER_COUNT) || "" +client_ready_reader.close +abort "only #{ready_workers.bytesize} workers created a Client" unless ready_workers.bytesize == WORKER_COUNT + +request_start_writer.write("." * WORKER_COUNT) +request_start_writer.close + +failed_workers = worker_pids.filter_map do |worker_pid| + pid, status = Process.wait2(worker_pid) + [pid, status] unless status.success? +end + +unless failed_workers.empty? + server_release_writer.close + begin + Process.kill("TERM", server_pid) + rescue Errno::ESRCH + nil + end + Process.wait(server_pid) + abort "workers failed: #{failed_workers.map { |pid, status| "#{pid}=#{status.inspect}" }.join(", ")}" +end + +begin + parent_client = Wreq::Client.new(no_proxy: true, http1_only: true, timeout: 5) + parent_response = parent_client.get("#{url}#{PARENT_REQUEST_PATH}") + raise "parent request after workers failed" unless parent_response.bytes == PARENT_REQUEST_PATH + warn "parent_after_workers=ok" +rescue => error + server_release_writer.close + begin + Process.kill("TERM", server_pid) + rescue Errno::ESRCH + nil + end + Process.wait(server_pid) + abort "parent_after_workers=unexpected #{error.class}: #{error.message}" +end + +server_release_writer.write(".") +server_release_writer.close + +_, server_status = Process.wait2(server_pid) +abort "server failed with #{server_status.inspect}" unless server_status.success? + +puts "ok"