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/src/client.rs b/src/client.rs index e42b488..1e25ecd 100644 --- a/src/client.rs +++ b/src/client.rs @@ -1,8 +1,8 @@ mod body; mod param; mod query; -mod req; -pub mod resp; +mod request; +pub mod response; use std::{net::IpAddr, time::Duration}; @@ -12,7 +12,7 @@ use wreq::Proxy; use crate::{ arch::{ProcessLocal, SUPPORTS_INTERFACE, SUPPORTS_TCP_USER_TIMEOUT}, - client::{req::execute_request, resp::Response}, + client::{request::execute_request, response::Response}, cookie::Jar, emulate::Emulation, error::wreq_error, @@ -123,6 +123,14 @@ struct Builder { #[magnus::wrap(class = "Wreq::Client", free_immediately, size)] pub struct Client(ProcessLocal); +/// Borrow the process-local wrapper without bypassing its ownership check. +impl AsRef> for Client { + #[inline] + fn as_ref(&self) -> &ProcessLocal { + &self.0 + } +} + // ===== impl Builder ===== impl Builder { @@ -204,8 +212,10 @@ impl Client { /// /// Returns `Wreq::BuilderError`, `Wreq::TlsError`, or another mapped native /// initialization error without unwinding through Ruby. - pub(crate) fn default_client(ruby: &Ruby) -> Result { + pub(crate) fn default(ruby: &Ruby) -> Result { Self::build(ruby, Builder::default()) + .map(ProcessLocal::new) + .map(Self) } /// Apply validated parameters and build the native client without the GVL. @@ -381,16 +391,6 @@ impl Client { // Ruby exceptions must be created after the GVL has been reacquired. result.map_err(|err| wreq_error(ruby, err)) } - - /// Clone the native client handle in the process that created it. - /// - /// # Errors - /// - /// Returns `Wreq::ForkError` when the client was inherited from a parent - /// process. - fn native_client(&self, ruby: &Ruby) -> Result { - self.0.get(ruby).cloned() - } } impl Client { @@ -398,138 +398,90 @@ impl Client { /// /// Request arguments are validated before the native client is built, so /// invalid options fail without initializing a connection pool. - pub(crate) fn request_with_default_client( + pub(crate) fn request_once_from_args( ruby: &Ruby, args: &[Value], ) -> Result { let ((method, url), request) = extract_request!(ruby, args, (Obj, String)); - let client = Self::default_client(ruby)?; - execute_request(ruby, client, *method, url, request) + let client = Self::default(ruby)?; + execute_request(ruby, &client, *method, url, request) } /// Send a request with `method` through a newly built default client. /// /// Request arguments are validated before the native client is built, so /// invalid options fail without initializing a connection pool. - pub(crate) fn execute_with_default_client( + pub(crate) fn request_once( ruby: &Ruby, method: Method, args: &[Value], ) -> Result { let ((url,), request) = extract_request!(ruby, args, (String,)); - let client = Self::default_client(ruby)?; - execute_request(ruby, client, method, url, request) + let client = Self::default(ruby)?; + execute_request(ruby, &client, method, url, request) } /// Send a HTTP request. #[inline] pub fn request(ruby: &Ruby, rb_self: &Self, args: &[Value]) -> Result { let ((method, url), request) = extract_request!(ruby, args, (Obj, String)); - execute_request(ruby, rb_self.native_client(ruby)?, *method, url, request) + execute_request(ruby, rb_self, *method, url, request) } /// Send a GET request. #[inline] pub fn get(ruby: &Ruby, rb_self: &Self, args: &[Value]) -> Result { let ((url,), request) = extract_request!(ruby, args, (String,)); - execute_request( - ruby, - rb_self.native_client(ruby)?, - Method::GET, - url, - request, - ) + execute_request(ruby, rb_self, Method::GET, url, request) } /// Send a POST request. #[inline] pub fn post(ruby: &Ruby, rb_self: &Self, args: &[Value]) -> Result { let ((url,), request) = extract_request!(ruby, args, (String,)); - execute_request( - ruby, - rb_self.native_client(ruby)?, - Method::POST, - url, - request, - ) + execute_request(ruby, rb_self, Method::POST, url, request) } /// Send a PUT request. #[inline] pub fn put(ruby: &Ruby, rb_self: &Self, args: &[Value]) -> Result { let ((url,), request) = extract_request!(ruby, args, (String,)); - execute_request( - ruby, - rb_self.native_client(ruby)?, - Method::PUT, - url, - request, - ) + execute_request(ruby, rb_self, Method::PUT, url, request) } /// Send a DELETE request. #[inline] pub fn delete(ruby: &Ruby, rb_self: &Self, args: &[Value]) -> Result { let ((url,), request) = extract_request!(ruby, args, (String,)); - execute_request( - ruby, - rb_self.native_client(ruby)?, - Method::DELETE, - url, - request, - ) + execute_request(ruby, rb_self, Method::DELETE, url, request) } /// Send a HEAD request. #[inline] pub fn head(ruby: &Ruby, rb_self: &Self, args: &[Value]) -> Result { let ((url,), request) = extract_request!(ruby, args, (String,)); - execute_request( - ruby, - rb_self.native_client(ruby)?, - Method::HEAD, - url, - request, - ) + execute_request(ruby, rb_self, Method::HEAD, url, request) } /// Send an OPTIONS request. #[inline] pub fn options(ruby: &Ruby, rb_self: &Self, args: &[Value]) -> Result { let ((url,), request) = extract_request!(ruby, args, (String,)); - execute_request( - ruby, - rb_self.native_client(ruby)?, - Method::OPTIONS, - url, - request, - ) + execute_request(ruby, rb_self, Method::OPTIONS, url, request) } /// Send a TRACE request. #[inline] pub fn trace(ruby: &Ruby, rb_self: &Self, args: &[Value]) -> Result { let ((url,), request) = extract_request!(ruby, args, (String,)); - execute_request( - ruby, - rb_self.native_client(ruby)?, - Method::TRACE, - url, - request, - ) + execute_request(ruby, rb_self, Method::TRACE, url, request) } /// Send a PATCH request. #[inline] pub fn patch(ruby: &Ruby, rb_self: &Self, args: &[Value]) -> Result { let ((url,), request) = extract_request!(ruby, args, (String,)); - execute_request( - ruby, - rb_self.native_client(ruby)?, - Method::PATCH, - url, - request, - ) + execute_request(ruby, rb_self, Method::PATCH, url, request) } } @@ -546,7 +498,7 @@ pub fn include(ruby: &Ruby, gem_module: &RModule) -> Result<(), magnus::Error> { client_class.define_method("trace", method!(Client::trace, -1))?; client_class.define_method("patch", method!(Client::patch, -1))?; - resp::include(ruby, gem_module)?; + response::include(ruby, gem_module)?; body::include(ruby, gem_module)?; Ok(()) } diff --git a/src/client/req.rs b/src/client/request.rs similarity index 88% rename from src/client/req.rs rename to src/client/request.rs index 64ce764..0a5f034 100644 --- a/src/client/req.rs +++ b/src/client/request.rs @@ -1,18 +1,21 @@ +//! Request option parsing, native request construction, and execution. + use std::{net::IpAddr, time::Duration}; use ::serde::Deserialize; use http::header; use magnus::{RHash, TryConvert, typed_data::Obj, value::ReprValue}; -use wreq::{Client, Proxy}; +use wreq::Proxy; use super::body::{Body, form::Form, json::Json}; use crate::{ - arch::SUPPORTS_INTERFACE, - client::{query::Query, resp::Response}, + arch::{ProcessLocal, SUPPORTS_INTERFACE}, + client::{query::Query, response::Response}, cookie::Cookies, emulate::Emulation, error::wreq_error, extractor::Extractor, + gvl, header::{Headers, OrigHeaders}, http::{Method, Version}, options::{NativeOption, Options}, @@ -21,7 +24,6 @@ use crate::{ /// The parameters for a request. #[derive(Default, Deserialize)] -#[non_exhaustive] pub struct Request { /// The emulation option for the request. #[serde(default)] @@ -177,14 +179,31 @@ impl Request { } } -pub fn execute_request>( +/// Build and execute one request with a process-local client. +/// +/// Request builder configuration is synchronous and runs without the GVL. +/// Only the completed request's network future is submitted to Tokio. +/// +/// # Errors +/// +/// Returns `Wreq::ForkError` for an inherited client, a mapped native error when +/// building or sending the request fails, or an interruption/runtime error from +/// [`rt::block_on`]. +pub fn execute_request( ruby: &magnus::Ruby, - client: Client, + client: &C, method: Method, url: U, mut request: Request, -) -> Result { - rt::block_on(ruby, async move { +) -> Result +where + U: AsRef, + C: AsRef>, +{ + // Process ownership failures construct a Ruby exception and require the GVL. + let client = client.as_ref().get(ruby)?; + + let request = gvl::nogvl(|| { let mut builder = client.request(method.into_ffi(), url.as_ref()); // Emulation options. @@ -304,8 +323,11 @@ pub fn execute_request>( builder = builder.body(wreq::Body::from(body)); } - // Send request. - builder.send().await.map(Response::new) - })? - .map_err(|err| wreq_error(ruby, err)) + builder.build() + }) + .map_err(|err| wreq_error(ruby, err))?; + + rt::block_on(ruby, client.execute(request))? + .map(Response::new) + .map_err(|err| wreq_error(ruby, err)) } diff --git a/src/client/resp.rs b/src/client/response.rs similarity index 99% rename from src/client/resp.rs rename to src/client/response.rs index 6dff164..7144e35 100644 --- a/src/client/resp.rs +++ b/src/client/response.rs @@ -1,3 +1,5 @@ +//! Ruby response bindings and response-body lifecycle management. + use std::{net::SocketAddr, sync::Arc}; use arc_swap::ArcSwapOption; diff --git a/src/lib.rs b/src/lib.rs index 0503557..dc56f96 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -20,7 +20,7 @@ mod tls; use magnus::{Error, Module, Ruby, Value}; use crate::{ - client::{Client, resp::Response}, + client::{Client, response::Response}, http::Method, }; @@ -30,55 +30,55 @@ const VERSION: &str = env!("CARGO_PKG_VERSION"); /// Send a HTTP request. #[inline] pub fn request(ruby: &Ruby, args: &[Value]) -> Result { - Client::request_with_default_client(ruby, args) + Client::request_once_from_args(ruby, args) } /// Send a GET request. #[inline] pub fn get(ruby: &Ruby, args: &[Value]) -> Result { - Client::execute_with_default_client(ruby, Method::GET, args) + Client::request_once(ruby, Method::GET, args) } /// Send a POST request. #[inline] pub fn post(ruby: &Ruby, args: &[Value]) -> Result { - Client::execute_with_default_client(ruby, Method::POST, args) + Client::request_once(ruby, Method::POST, args) } /// Send a PUT request. #[inline] pub fn put(ruby: &Ruby, args: &[Value]) -> Result { - Client::execute_with_default_client(ruby, Method::PUT, args) + Client::request_once(ruby, Method::PUT, args) } /// Send a DELETE request. #[inline] pub fn delete(ruby: &Ruby, args: &[Value]) -> Result { - Client::execute_with_default_client(ruby, Method::DELETE, args) + Client::request_once(ruby, Method::DELETE, args) } /// Send a HEAD request. #[inline] pub fn head(ruby: &Ruby, args: &[Value]) -> Result { - Client::execute_with_default_client(ruby, Method::HEAD, args) + Client::request_once(ruby, Method::HEAD, args) } /// Send an OPTIONS request. #[inline] pub fn options(ruby: &Ruby, args: &[Value]) -> Result { - Client::execute_with_default_client(ruby, Method::OPTIONS, args) + Client::request_once(ruby, Method::OPTIONS, args) } /// Send a TRACE request. #[inline] pub fn trace(ruby: &Ruby, args: &[Value]) -> Result { - Client::execute_with_default_client(ruby, Method::TRACE, args) + Client::request_once(ruby, Method::TRACE, args) } /// Send a PATCH request. #[inline] pub fn patch(ruby: &Ruby, args: &[Value]) -> Result { - Client::execute_with_default_client(ruby, Method::PATCH, args) + Client::request_once(ruby, Method::PATCH, args) } /// wreq ruby binding diff --git a/src/macros.rs b/src/macros.rs index 18d97c6..1288e09 100644 --- a/src/macros.rs +++ b/src/macros.rs @@ -151,7 +151,7 @@ macro_rules! extract_request { ($ruby:expr, $args:expr, $required:ty) => {{ let args = magnus::scan_args::scan_args::<$required, (), (), (), magnus::RHash, ()>($args)?; let required = args.required; - let request = crate::client::req::Request::new($ruby, args.keywords)?; + let request = crate::client::request::Request::new($ruby, args.keywords)?; (required, request) }}; } diff --git a/src/rt.rs b/src/rt.rs index 3cb8555..2f5cc4e 100644 --- a/src/rt.rs +++ b/src/rt.rs @@ -17,24 +17,6 @@ use crate::{ /// Initialize the global runtime lazily and preserve failures for Ruby. static RUNTIME: OnceLock> = OnceLock::new(); -/// Reject a child process that inherited an initialized native runtime. -/// -/// # Errors -/// -/// Returns `Wreq::ForkError` when the global runtime belongs to the parent -/// process. -fn ensure_runtime_owner(ruby: &Ruby) -> Result<(), magnus::Error> { - #[cfg(unix)] - if let Some((owner_pid, current_pid)) = arch::forked_process_ids() { - return Err(fork_error(ruby, owner_pid, current_pid)); - } - - #[cfg(not(unix))] - let _ = ruby; - - Ok(()) -} - /// Block on a future to completion on the current process's global Tokio runtime. /// /// The future runs without Ruby's GVL, so it must not construct Ruby objects or @@ -56,7 +38,12 @@ where #[cfg(unix)] arch::initialize_fork_tracking().map_err(|err| fork_handler_error(ruby, &err))?; - ensure_runtime_owner(ruby)?; + // A forked child must not read or use the parent's inherited Tokio state. + #[cfg(unix)] + if let Some((owner_pid, current_pid)) = arch::forked_process_ids() { + return Err(fork_error(ruby, owner_pid, current_pid)); + } + let runtime = RUNTIME .get_or_init(|| { let mut builder = Builder::new_multi_thread(); 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"