diff --git a/README.md b/README.md index e5c4c1b..1ef044d 100644 --- a/README.md +++ b/README.md @@ -148,6 +148,14 @@ With these steps, your HttpClient is now compatible with HTTP/2.✨ However, since it does not have features such as connection control by the number of streams, it is necessary to separate handler instances when explicitly creating different connections. +### Disposing a handler + +`YetAnotherHttpHandler.Dispose()` rejects new requests and closes a native callback gate shared by response headers, response bodies, completion notifications and custom TLS certificate verification. It cancels pending requests and waits for admitted callbacks and outstanding body-flush acknowledgements before releasing their managed state. Repeated and concurrent disposal waits for the same cleanup. + +After `Dispose()` returns, this handler cannot invoke another managed callback. Native connection cleanup and OS DNS work may finish later; their native state remains alive independently. Other handlers sharing the runtime continue running. Disposal does not wait for all native work or arbitrary application continuations and `HttpContent` producers; it does stop the package's own upload loop. + +Dispose handlers **before** Unity/IL2CPP runtime cleanup or unload begins. The package does not automatically subscribe to application lifecycle events. Disposal can still block on an executing callback. Calling `Dispose()` synchronously inside a native callback, including `OnVerifyServerCertificate`, throws `InvalidOperationException` to prevent waiting for itself. A callback must also not wait for another thread to dispose the handler. + ### Using gRPC (grpc-dotnet) library To use grpc-dotnet (Grpc.Net.Client), add the following additional libraries: @@ -351,6 +359,8 @@ When creating a package, The following artifacts directory is used. - native/artifacts/{.NET RID}/{lib}yaha_native.{dll,so} +Managed sources and native binaries must come from the same build. The callback gate requires `yaha_context_disable_callbacks`, `yaha_context_wait_callbacks` and a matching response-completion handle layout. Before distributing a Unity package, update the prebuilt libraries for every platform through the `Build Native Libraries` workflow with `update-unity-native` enabled. + ```bash # Generate THIRD-PARTY-NOTICES using cargo-about cargo about generate about.hbs > ../THIRD-PARTY-NOTICES diff --git a/native/yaha_native/src/binding.rs b/native/yaha_native/src/binding.rs index ae3845f..7fcbe9a 100644 --- a/native/yaha_native/src/binding.rs +++ b/native/yaha_native/src/binding.rs @@ -23,6 +23,12 @@ use crate::{ primitives::WriteResult, }; use futures_util::StreamExt; +use crate::callback_gate::CallbackGuard; + +struct ResponseCompletion { + sender: oneshot::Sender>, + _guard: CallbackGuard, +} #[no_mangle] @@ -57,6 +63,9 @@ pub extern "C" fn yaha_init_runtime(worker_threads: i32) -> *mut YahaNativeRunti #[no_mangle] pub extern "C" fn yaha_dispose_runtime(ctx: *mut YahaNativeRuntimeContext) { let ctx = unsafe { Box::from_raw(ctx as *mut YahaNativeRuntimeContextInternal) }; + // Callback gates are already closed and drained. Native-only work (such + // as OS DNS resolution) can finish without blocking managed shutdown. + ctx.runtime.shutdown_background(); } #[no_mangle] @@ -72,25 +81,47 @@ pub extern "C" fn yaha_init_context( on_complete: extern "C" fn(req_seq: i32, state: NonZeroIsize, reason: CompletionReason, h2_error_code: u32), ) -> *mut YahaNativeContext { let runtime_ctx = YahaNativeRuntimeContextInternal::from_raw_context(runtime_ctx); - let ctx = Box::new(YahaNativeContextInternal::new( + let ctx = Arc::new(YahaNativeContextInternal::new( runtime_ctx.runtime.handle().clone(), on_status_code_and_headers_receive, on_receive, on_complete, )); - Box::into_raw(ctx) as *mut YahaNativeContext + Arc::into_raw(ctx) as *mut YahaNativeContext +} + +/// Clones an owning reference that can outlive the managed SafeHandle. +/// +/// # Safety +/// `ctx` must come from `yaha_init_context`, with a strong reference kept alive +/// for this call. Configuration must not be mutated while the clone is alive. +unsafe fn clone_context(ctx: *const YahaNativeContext) -> Arc> { + let ptr = ctx as *const YahaNativeContextInternal; + unsafe { + Arc::increment_strong_count(ptr); + Arc::from_raw(ptr) + } } #[no_mangle] pub extern "C" fn yaha_dispose_context(ctx: *mut YahaNativeContext) { - let mut ctx = unsafe { Box::from_raw(ctx as *mut YahaNativeContextInternal) }; - ctx.on_complete = _sentinel_on_complete; - ctx.on_receive = _sentinel_on_receive; - ctx.on_status_code_and_headers_receive = _sentinel_on_status_code_and_headers_receive; + // Reclaim the managed owner's reference, releasing it when this call returns. + let ctx = unsafe { Arc::from_raw(ctx as *const YahaNativeContextInternal) }; + ctx.callbacks.close(); + ctx.callbacks.wait(); +} + +#[no_mangle] +pub extern "C" fn yaha_context_disable_callbacks(ctx: *const YahaNativeContext) { + let ctx = unsafe { &*(ctx as *const YahaNativeContextInternal) }; + ctx.callbacks.close(); +} + +#[no_mangle] +pub extern "C" fn yaha_context_wait_callbacks(ctx: *const YahaNativeContext) { + let ctx = unsafe { &*(ctx as *const YahaNativeContextInternal) }; + ctx.callbacks.wait(); } -extern "C" fn _sentinel_on_complete(_: i32, _: NonZeroIsize, _: CompletionReason, _: u32) { panic!("The context has already disposed: on_complete"); } -extern "C" fn _sentinel_on_receive(_: i32, _: NonZeroIsize, _: usize, _: *const u8, _: usize) { panic!("The context has already disposed: on_receive"); } -extern "C" fn _sentinel_on_status_code_and_headers_receive(_: i32, _: NonZeroIsize, _: i32, _: YahaHttpVersion) { panic!("The context has already disposed: on_status_code_and_headers_receive"); } #[no_mangle] pub extern "C" fn yaha_client_config_add_root_certificates( @@ -486,7 +517,7 @@ pub extern "C" fn yaha_request_begin( req_ctx: *const YahaNativeRequestContext, state: NonZeroIsize ) -> bool { - let ctx = YahaNativeContextInternal::from_raw_context(ctx); + let ctx = unsafe { clone_context(ctx) }; // Begin request on async runtime. let body; @@ -508,6 +539,7 @@ pub extern "C" fn yaha_request_begin( { let req_ctx = req_ctx.clone(); ctx.runtime.clone().spawn(async move { + let ctx = &*ctx; let cancellation_token = { let req_ctx = req_ctx.lock().unwrap(); req_ctx.cancellation_token.clone() @@ -532,14 +564,14 @@ pub extern "C" fn yaha_request_begin( let mut req_ctx = req_ctx.lock().unwrap(); req_ctx.last_error = Some("The client has not been built. You need to build it before sending the request.".to_string()); } - (ctx.on_complete)(seq, state, CompletionReason::Error, 0); + ctx.notify_complete(seq, state, CompletionReason::Error, 0); return; } // Send a request and wait for response status and headers. let res = select! { _ = cancellation_token.cancelled() => { - (ctx.on_complete)(seq, state, CompletionReason::Aborted, 0); + ctx.notify_complete(seq, state, CompletionReason::Aborted, 0); return; } res = ctx.request(req) => { @@ -570,7 +602,7 @@ pub extern "C" fn yaha_request_begin( req_ctx.response_status = res.status(); req_ctx.response_version = YahaHttpVersion::from(res.version()); } - (ctx.on_status_code_and_headers_receive)( + ctx.notify_headers( seq, state, res.status().as_u16() as i32, @@ -585,7 +617,7 @@ pub extern "C" fn yaha_request_begin( while !body.is_end_stream() { select! { _ = cancellation_token.cancelled() => { - (ctx.on_complete)(seq, state, CompletionReason::Aborted, 0); + ctx.notify_complete(seq, state, CompletionReason::Aborted, 0); return; } received = body.frame() => { @@ -595,10 +627,17 @@ pub extern "C" fn yaha_request_begin( Ok(frame) => { if frame.is_data() { let data = frame.into_data().unwrap(); + let Some(guard) = ctx.callbacks.enter() else { + return; + }; let (tx, rx) = oneshot::channel::>(); - let tx = Box::into_raw(Box::new(tx)) as usize; + let tx = Box::into_raw(Box::new(ResponseCompletion { + sender: tx, + _guard: guard.retain_for_deferred_ack(), + })) as usize; (ctx.on_receive)(seq, state, data.len(), data.as_ptr(), tx); + drop(guard); match rx.await { Ok(result) => { if let Err(err) = result { @@ -607,7 +646,7 @@ pub extern "C" fn yaha_request_begin( let mut req_ctx = req_ctx.lock().unwrap(); req_ctx.last_error = Some(err); } - (ctx.on_complete)(seq, state, CompletionReason::Error, 0); + ctx.notify_complete(seq, state, CompletionReason::Error, 0); return; } }, @@ -617,7 +656,7 @@ pub extern "C" fn yaha_request_begin( let mut req_ctx = req_ctx.lock().unwrap(); req_ctx.last_error = Some("on_receive() has not completed correctly.".to_string()); } - (ctx.on_complete)(seq, state, CompletionReason::Error, 0); + ctx.notify_complete(seq, state, CompletionReason::Error, 0); return; } } @@ -659,7 +698,7 @@ pub extern "C" fn yaha_request_begin( let rc = reason.map(|r| u32::from(r)); - (ctx.on_complete)(seq, state, CompletionReason::Error, rc.unwrap_or_default()); + ctx.notify_complete(seq, state, CompletionReason::Error, rc.unwrap_or_default()); return; } } @@ -678,7 +717,7 @@ pub extern "C" fn yaha_request_begin( req_ctx.try_complete(); } - (ctx.on_complete)(seq, state, CompletionReason::Success, 0); + ctx.notify_complete(seq, state, CompletionReason::Success, 0); { let mut req_ctx = req_ctx.lock().unwrap(); @@ -691,7 +730,7 @@ pub extern "C" fn yaha_request_begin( true } -fn complete_with_error(ctx: &mut YahaNativeContextInternal, req_ctx: Arc>, seq: i32, state: NonZeroIsize, err: hyper_util::client::legacy::Error) { +fn complete_with_error(ctx: &YahaNativeContextInternal, req_ctx: Arc>, seq: i32, state: NonZeroIsize, err: hyper_util::client::legacy::Error) { let mut h2_error_code = None; { @@ -710,7 +749,7 @@ fn complete_with_error(ctx: &mut YahaNativeContextInternal, req_ctx: Arc>) }; + let completion = unsafe { Box::from_raw(task_handle as *mut ResponseCompletion) }; + let ResponseCompletion { sender: tx, _guard } = *completion; if error.is_null() { - tx.send(Ok(())).unwrap(); + let _ = tx.send(Ok(())); } else { let error = unsafe { (*error).to_str().to_string() }; - tx.send(Err(error)).unwrap(); + let _ = tx.send(Err(error)); } } #[cfg(test)] mod tests { + use super::*; use std::ptr::null; + use std::sync::atomic::{AtomicUsize, Ordering}; + use std::sync::mpsc; - use super::*; + extern "C" fn headers(_: i32, state: NonZeroIsize, _: i32, _: YahaHttpVersion) { + unsafe { &*(state.get() as *const AtomicUsize) }.fetch_add(1, Ordering::SeqCst); + } + + extern "C" fn complete(_: i32, state: NonZeroIsize, _: CompletionReason, _: u32) { + unsafe { &*(state.get() as *const AtomicUsize) }.fetch_add(1, Ordering::SeqCst); + } + + extern "C" fn receive(_: i32, _: NonZeroIsize, _: usize, _: *const u8, _: usize) {} + + #[test] + fn retained_context_suppresses_late_callbacks_after_managed_owner_is_released() { + let calls = AtomicUsize::new(0); + let state = NonZeroIsize::new(&calls as *const _ as isize).unwrap(); + let runtime = yaha_init_runtime(1); + let ctx = yaha_init_context(runtime, headers, receive, complete); + let retained = unsafe { clone_context(ctx) }; + retained.notify_headers(1, state, 200, YahaHttpVersion::Http11); + retained.notify_complete(1, state, CompletionReason::Success, 0); + assert_eq!(calls.load(Ordering::SeqCst), 2); + + yaha_context_disable_callbacks(ctx); + yaha_context_wait_callbacks(ctx); + yaha_dispose_context(ctx); + yaha_dispose_runtime(runtime); + + // Simulate a request which still owns the context after Dispose. + retained.notify_headers(1, state, 200, YahaHttpVersion::Http11); + retained.notify_complete(1, state, CompletionReason::Aborted, 0); + assert_eq!(calls.load(Ordering::SeqCst), 2); + } + + #[test] + fn runtime_release_does_not_join_native_blocking_work() { + let runtime = yaha_init_runtime(1); + let handle = YahaNativeRuntimeContextInternal::from_raw_context(runtime).runtime.handle().clone(); + let (entered_tx, entered_rx) = mpsc::channel(); + let (release_tx, release_rx) = mpsc::channel(); + let (done_tx, done_rx) = mpsc::channel(); + handle.spawn_blocking(move || { + entered_tx.send(()).unwrap(); + let released = release_rx.recv_timeout(Duration::from_secs(5)).is_ok(); + done_tx.send(released).unwrap(); + }); + entered_rx.recv_timeout(Duration::from_secs(5)).unwrap(); + yaha_dispose_runtime(runtime); + // A blocking Runtime drop would wait for the timeout and fail this test. + assert!(matches!(done_rx.try_recv(), Err(mpsc::TryRecvError::Empty))); + release_tx.send(()).unwrap(); + assert!(done_rx.recv_timeout(Duration::from_secs(5)).unwrap()); + } unsafe fn set_header(req_ctx: *const YahaNativeRequestContext, name: &[u8], value: &[u8]) -> bool { let name = StringBuffer::new(name.as_ptr(), name.len() as i32); diff --git a/native/yaha_native/src/callback_gate.rs b/native/yaha_native/src/callback_gate.rs new file mode 100644 index 0000000..e95dd9a --- /dev/null +++ b/native/yaha_native/src/callback_gate.rs @@ -0,0 +1,133 @@ +use std::sync::{Arc, Condvar, Mutex}; + +#[derive(Clone, Debug, Default)] +pub struct CallbackGate(Arc); + +#[derive(Debug, Default)] +struct Inner { + state: Mutex, + idle: Condvar, +} + +#[derive(Debug, Default)] +struct State { + closed: bool, + active: usize, +} + +pub struct CallbackGuard(Arc); + +impl CallbackGate { + pub fn enter(&self) -> Option { + let mut state = self.0.state.lock().unwrap(); + if state.closed { + return None; + } + state.active += 1; + Some(CallbackGuard(self.0.clone())) + } + + pub fn close(&self) { + self.0.state.lock().unwrap().closed = true; + } + + pub fn wait(&self) { + let mut state = self.0.state.lock().unwrap(); + assert!(state.closed); + while state.active != 0 { + state = self.0.idle.wait(state).unwrap(); + } + } +} + +impl CallbackGuard { + pub fn retain_for_deferred_ack(&self) -> Self { + self.0.state.lock().unwrap().active += 1; + Self(self.0.clone()) + } +} + +impl Drop for CallbackGuard { + fn drop(&mut self) { + let mut state = self.0.state.lock().unwrap(); + state.active -= 1; + if state.active == 0 { + self.0.idle.notify_all(); + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::{sync::mpsc, thread, time::Duration}; + + const TIMEOUT: Duration = Duration::from_secs(5); + + #[test] + fn close_rejects_new_callbacks_and_waits_for_an_admitted_callback() { + let gate = CallbackGate::default(); + let guard = gate.enter().unwrap(); + gate.close(); + assert!(gate.enter().is_none()); + let stopping = gate.clone(); + let (done_tx, done_rx) = mpsc::channel(); + let waiter = thread::spawn(move || { + stopping.wait(); + done_tx.send(()).unwrap(); + }); + assert!(matches!( + done_rx.recv_timeout(Duration::from_millis(50)), + Err(mpsc::RecvTimeoutError::Timeout) + )); + drop(guard); + done_rx.recv_timeout(TIMEOUT).unwrap(); + waiter.join().unwrap(); + gate.close(); + gate.wait(); + } + + #[test] + fn deferred_acknowledgement_outlives_callback_even_when_close_races_it() { + let gate = CallbackGate::default(); + let guard = gate.enter().unwrap(); + gate.close(); + let acknowledgement = guard.retain_for_deferred_ack(); + drop(guard); + let stopping = gate.clone(); + let (done_tx, done_rx) = mpsc::channel(); + let waiter = thread::spawn(move || { + stopping.wait(); + done_tx.send(()).unwrap(); + }); + assert!(matches!( + done_rx.recv_timeout(Duration::from_millis(50)), + Err(mpsc::RecvTimeoutError::Timeout) + )); + drop(acknowledgement); + done_rx.recv_timeout(TIMEOUT).unwrap(); + waiter.join().unwrap(); + } + + #[test] + fn closing_one_handler_does_not_close_another_or_join_native_work() { + let runtime = tokio::runtime::Runtime::new().unwrap(); + let gate = CallbackGate::default(); + let other = CallbackGate::default(); + let late_callback = gate.clone(); + let (entered_tx, entered_rx) = mpsc::channel(); + let (release_tx, release_rx) = mpsc::channel(); + let work = runtime.spawn_blocking(move || { + entered_tx.send(()).unwrap(); + release_rx.recv_timeout(TIMEOUT).unwrap(); + assert!(late_callback.enter().is_none()); + }); + entered_rx.recv_timeout(TIMEOUT).unwrap(); + gate.close(); + gate.wait(); + assert!(!work.is_finished()); + assert!(other.enter().is_some()); + release_tx.send(()).unwrap(); + runtime.block_on(work).unwrap(); + } +} diff --git a/native/yaha_native/src/context.rs b/native/yaha_native/src/context.rs index 446e689..afab150 100644 --- a/native/yaha_native/src/context.rs +++ b/native/yaha_native/src/context.rs @@ -29,6 +29,8 @@ use hyperlocal::UnixConnector; use rustls::pki_types::{CertificateDer, PrivateKeyDer}; use tokio_util::sync::CancellationToken; +use crate::callback_gate::CallbackGate; + use crate::{primitives::{CompletionReason, YahaHttpVersion}}; type OnStatusCodeAndHeadersReceive = @@ -64,6 +66,7 @@ impl YahaNativeRuntimeContextInternal { pub struct YahaNativeContext; pub struct YahaNativeContextInternal<'a> { pub runtime: tokio::runtime::Handle, + pub callbacks: CallbackGate, pub client_builder: Option, pub skip_certificate_verification: Option, pub server_certificate_verification_handler: Option<(OnServerCertificateVerificationHandler, NonZeroIsize)>, @@ -96,6 +99,7 @@ impl YahaNativeContextInternal<'_> { ) -> Self { YahaNativeContextInternal { runtime: runtime_handle, + callbacks: CallbackGate::default(), tcp_client: None, client_builder: Some(Client::builder(TokioExecutor::new())), skip_certificate_verification: None, @@ -115,6 +119,18 @@ impl YahaNativeContextInternal<'_> { } } + pub fn notify_headers(&self, seq: i32, state: NonZeroIsize, status: i32, version: YahaHttpVersion) { + if let Some(_guard) = self.callbacks.enter() { + (self.on_status_code_and_headers_receive)(seq, state, status, version); + } + } + + pub fn notify_complete(&self, seq: i32, state: NonZeroIsize, reason: CompletionReason, h2_error_code: u32) { + if let Some(_guard) = self.callbacks.enter() { + (self.on_complete)(seq, state, reason, h2_error_code); + } + } + pub fn build_client(&mut self) { let mut builder = self.client_builder.take().unwrap(); builder.timer(TokioTimer::new()); @@ -151,6 +167,7 @@ impl YahaNativeContextInternal<'_> { danger::CustomCerficateVerification::new( server_certificate_verification_handler, signature_algorithms, + self.callbacks.clone(), ), )) .with_no_client_auth(); @@ -256,6 +273,7 @@ mod danger { use rustls::pki_types::{CertificateDer, ServerName, UnixTime}; use super::OnServerCertificateVerificationHandler; + use crate::callback_gate::CallbackGate; #[derive(Debug)] pub struct NoCertificateVerification {} @@ -263,16 +281,19 @@ mod danger { pub struct CustomCerficateVerification { handler: (OnServerCertificateVerificationHandler, NonZeroIsize), signature_algorithms: WebPkiSupportedAlgorithms, + callbacks: CallbackGate, } impl CustomCerficateVerification { pub fn new( handler: (OnServerCertificateVerificationHandler, NonZeroIsize), signature_algorithms: WebPkiSupportedAlgorithms, + callbacks: CallbackGate, ) -> Self { Self { handler, signature_algorithms, + callbacks, } } } @@ -308,6 +329,9 @@ mod danger { _ocsp_response: &[u8], now: UnixTime, ) -> Result { + let Some(_guard) = self.callbacks.enter() else { + return Err(Error::InvalidCertificate(rustls::CertificateError::ApplicationVerificationFailure)); + }; let server_name = server_name.to_str(); let server_name = server_name.as_bytes(); let cetificate_der = end_entity.as_ref(); @@ -434,17 +458,22 @@ mod tests { io::{self, Cursor}, net::{TcpListener, TcpStream}, num::NonZeroIsize, - sync::Arc, + sync::{ + atomic::{AtomicUsize, Ordering}, + Arc, + }, thread, time::Duration, }; use rustls::{ - pki_types::{CertificateDer, PrivateKeyDer, ServerName}, + client::danger::ServerCertVerifier, + pki_types::{CertificateDer, PrivateKeyDer, ServerName, UnixTime}, ClientConfig, ClientConnection, ServerConfig, ServerConnection, SupportedProtocolVersion, }; use super::danger::CustomCerficateVerification; + use crate::callback_gate::CallbackGate; const CERTIFICATE_A: &[u8] = include_bytes!("../testdata/certificate-verify/cert-a.pem"); const PRIVATE_KEY_A: &[u8] = include_bytes!("../testdata/certificate-verify/key-a.pem"); @@ -515,6 +544,7 @@ mod tests { NonZeroIsize::new(1).expect("the callback state must be non-zero"), ), rustls::crypto::ring::default_provider().signature_verification_algorithms, + CallbackGate::default(), ); let client_config = ClientConfig::builder_with_protocol_versions(&[protocol_version]) .dangerous() @@ -566,4 +596,28 @@ mod tests { complete_handshake(&rustls::version::TLS13, PRIVATE_KEY_B) .expect_err("a TLS 1.3 signature made by an unrelated key must be rejected"); } + + extern "C" fn verify(state: NonZeroIsize, _: *const u8, _: usize, _: *const u8, _: usize, _: u64) -> bool { + unsafe { &*(state.get() as *const AtomicUsize) }.fetch_add(1, Ordering::SeqCst); + true + } + + #[test] + fn previously_created_tls_verifier_observes_closed_gate() { + let calls = AtomicUsize::new(0); + let callbacks = CallbackGate::default(); + let verifier = CustomCerficateVerification::new( + (verify, NonZeroIsize::new(&calls as *const _ as isize).unwrap()), + rustls::crypto::ring::default_provider().signature_verification_algorithms, + callbacks.clone(), + ); + let certificate = CertificateDer::from(&b"test certificate"[..]); + let name = ServerName::try_from("localhost").unwrap(); + let now = UnixTime::since_unix_epoch(Duration::from_secs(0)); + assert!(verifier.verify_server_cert(&certificate, &[], &name, &[], now).is_ok()); + callbacks.close(); + callbacks.wait(); + assert!(verifier.verify_server_cert(&certificate, &[], &name, &[], now).is_err()); + assert_eq!(calls.load(Ordering::SeqCst), 1); + } } diff --git a/native/yaha_native/src/lib.rs b/native/yaha_native/src/lib.rs index 29a702f..b6a3f53 100644 --- a/native/yaha_native/src/lib.rs +++ b/native/yaha_native/src/lib.rs @@ -1,4 +1,5 @@ mod context; +mod callback_gate; mod interop; mod primitives; diff --git a/src/YetAnotherHttpHandler/NativeHttpHandlerCore.cs b/src/YetAnotherHttpHandler/NativeHttpHandlerCore.cs index c32c04f..b49cf4f 100644 --- a/src/YetAnotherHttpHandler/NativeHttpHandlerCore.cs +++ b/src/YetAnotherHttpHandler/NativeHttpHandlerCore.cs @@ -1,4 +1,6 @@ using System; +using System.Collections.Concurrent; +using System.Collections.Generic; using System.Diagnostics; using System.Text; using System.IO.Pipelines; @@ -22,8 +24,36 @@ internal class NativeHttpHandlerCore : IDisposable //private unsafe YahaNativeContext* _ctx; private readonly YahaContextSafeHandle _handle; private bool _disposed = false; + private readonly object _disposeLock = new object(); + private readonly ConcurrentDictionary _requests = new ConcurrentDictionary(); private PipeOptions? _responsePipeOptions; + [ThreadStatic] + private static int _callbackDepth; + + private readonly struct CallbackScope : IDisposable + { + public static CallbackScope Enter() + { + _callbackDepth++; + return default; + } + + public void Dispose() => _callbackDepth--; + } + + internal static void ThrowIfInCallback() + { + if (_callbackDepth != 0) + { + throw new InvalidOperationException("Cannot synchronously dispose a handler from a native callback. Dispose it after the callback returns."); + } + } + + internal void RemoveRequest(RequestContext request) + { + _requests.TryRemove(request.RequestSequence, out _); + } // NOTE: We need to keep the callback delegates in advance. // The delegates are kept on the Rust side, so it will crash if they are garbage collected. private static readonly unsafe NativeMethods.yaha_init_context_on_status_code_and_headers_receive_delegate OnStatusCodeAndHeaderReceiveCallback = OnStatusCodeAndHeaderReceive; @@ -50,7 +80,7 @@ public unsafe NativeHttpHandlerCore(NativeClientSettings settings) catch { // NOTE: If the initialization fails, we need to release the runtime. - NativeRuntime.Instance.Release(); + Dispose(); throw; } finally @@ -237,8 +267,19 @@ public Task SendAsync(HttpRequestMessage request, Cancellat static async Task SendBodyAsync(HttpContent requestContent, PipeWriter writer, CancellationToken cancellationToken) { - await requestContent.CopyToAsync(writer.AsStream()).ConfigureAwait(false); // TODO: cancel - await writer.CompleteAsync().ConfigureAwait(false); + Exception? error = null; + try + { + await requestContent.CopyToAsync(writer.AsStream()).ConfigureAwait(false); // TODO: cancel + } + catch (Exception ex) + { + error = ex; + } + finally + { + await writer.CompleteAsync(error).ConfigureAwait(false); + } } } @@ -271,6 +312,14 @@ private unsafe RequestContext Send(HttpRequestMessage request, CancellationToken return UnsafeSend(_handle, reqCtxHandle, requestSequence, request, cancellationToken); } + catch + { + // Header/URI validation can fail before a managed request + // is registered. Do not leave its parent/runtime references + // waiting for a SafeHandle finalizer after handler disposal. + reqCtxHandle.Dispose(); + throw; + } finally { if (addRefReqContext) @@ -340,8 +389,9 @@ private unsafe RequestContext UnsafeSend(YahaContextSafeHandle ctxHandle, YahaRe NativeMethods.yaha_request_set_has_body(ctx, reqCtx, request.Content != null); // Prepare a request context - var requestContextManaged = new RequestContext(_handle, reqCtxHandle, request, requestSequence, _responsePipeOptions, cancellationToken); + var requestContextManaged = new RequestContext(this, _handle, reqCtxHandle, request, requestSequence, _responsePipeOptions, cancellationToken); requestContextManaged.Allocate(); + _requests.TryAdd(requestSequence, requestContextManaged); if (cancellationToken.IsCancellationRequested) { // Dispose the request context immediately. @@ -386,6 +436,8 @@ private static unsafe void SetRequestHeaders(HttpHeaders headers, YahaNativeCont [MonoPInvokeCallback(typeof(NativeMethods.yaha_init_context_on_status_code_and_headers_receive_delegate))] private static unsafe void OnStatusCodeAndHeaderReceive(int reqSeq, IntPtr state, int statusCode, YahaHttpVersion version) { + using var _ = CallbackScope.Enter(); + if (YahaEventSource.Log.IsEnabled()) YahaEventSource.Log.Info($"[ReqSeq:{reqSeq}:State:0x{state:X}] Status code and headers received: StatusCode={statusCode}; Version={version}"); var requestContext = RequestContext.FromHandle(state); @@ -438,6 +490,8 @@ private static unsafe void OnStatusCodeAndHeaderReceive(int reqSeq, IntPtr state [MonoPInvokeCallback(typeof(NativeMethods.yaha_client_config_set_server_certificate_verification_handler_handler_delegate))] private static unsafe bool OnServerCertificateVerification(IntPtr callbackState, byte* serverNamePtr, UIntPtr /*nuint*/ serverNameLength, byte* certificateDerPtr, UIntPtr /*nuint*/ certificateDerLength, ulong now) { + using var _ = CallbackScope.Enter(); + var serverName = UnsafeUtilities.GetStringFromUtf8Bytes(new ReadOnlySpan(serverNamePtr, (int)serverNameLength)); var certificateDer = new ReadOnlySpan(certificateDerPtr, (int)certificateDerLength); if (YahaEventSource.Log.IsEnabled()) YahaEventSource.Log.Trace($"OnServerCertificateVerification: State=0x{callbackState:X}; ServerName={serverName}; CertificateDer.Length={certificateDer.Length}; Now={now}"); @@ -465,15 +519,22 @@ private static unsafe bool OnServerCertificateVerification(IntPtr callbackState, [MonoPInvokeCallback(typeof(NativeMethods.yaha_init_context_on_receive_delegate))] private static unsafe void OnReceive(int reqSeq, IntPtr state, UIntPtr length, byte* buf, nuint taskHandle) { + using var _ = CallbackScope.Enter(); + + ResponseContext? response = null; + var ackDeferred = false; try { if (YahaEventSource.Log.IsEnabled()) YahaEventSource.Log.Trace($"[ReqSeq:{reqSeq}:State:0x{state:X}] Response data received: Length={length}"); var bufSpan = new Span(buf, (int)length); var requestContext = RequestContext.FromHandle(state); - var write = requestContext.Response.WriteAsync(bufSpan); + response = requestContext.Response; + var write = response.WriteAsync(bufSpan, taskHandle, out ackDeferred); - if (write.IsCompleted) + // A deferred flush may already be completed by Cancel, which also acknowledges + // the task, so branch on the deferral rather than write.IsCompleted. + if (!ackDeferred) { write.GetAwaiter().GetResult(); CompleteTask(taskHandle); @@ -491,22 +552,32 @@ private static unsafe void OnReceive(int reqSeq, IntPtr state, UIntPtr length, b catch (Exception ex) { if (YahaEventSource.Log.IsEnabled()) YahaEventSource.Log.Error($"[ReqSeq:{reqSeq}:State:0x{state:X}] Failed to flush response data: {ex}"); - CompleteTask(taskHandle, ex.ToString()); + if (response.TryTakeDeferredAck(taskHandle)) + { + CompleteTask(taskHandle, ex.ToString()); + } return; } - CompleteTask(taskHandle); + // Cancel may already have acknowledged the task. + if (response.TryTakeDeferredAck(taskHandle)) + { + CompleteTask(taskHandle); + } }); } } catch (Exception ex) { if (YahaEventSource.Log.IsEnabled()) YahaEventSource.Log.Error($"[ReqSeq:{reqSeq}:State:0x{state:X}] Failed to flush response data: {ex}"); - CompleteTask(taskHandle, ex.ToString()); + if (!ackDeferred || response!.TryTakeDeferredAck(taskHandle)) + { + CompleteTask(taskHandle, ex.ToString()); + } } } - private static unsafe void CompleteTask(nuint taskHandle, string? error = null) + internal static unsafe void CompleteTask(nuint taskHandle, string? error = null) { if (error is null) { @@ -525,6 +596,8 @@ private static unsafe void CompleteTask(nuint taskHandle, string? error = null) [MonoPInvokeCallback(typeof(NativeMethods.yaha_init_context_on_complete_delegate))] private static unsafe void OnComplete(int reqSeq, IntPtr state, CompletionReason reason, uint h2ErrorCode) { + using var _ = CallbackScope.Enter(); + if (YahaEventSource.Log.IsEnabled()) YahaEventSource.Log.Info($"[ReqSeq:{reqSeq}:State:0x{state:X}] Response completed: Reason={reason}; H2ErrorCode=0x{h2ErrorCode:x}"); var requestContext = RequestContext.FromHandle(state); @@ -632,21 +705,46 @@ public void Dispose() private void Dispose(bool disposing) { - if (_disposed) + ThrowIfInCallback(); + lock (_disposeLock) { - return; + if (_disposed) + { + return; + } + + var requests = _DisableAndDrainCallbacks(); + _ReleaseDrainedRequests(requests); + _handle.Dispose(); + NativeRuntime.Instance.Release(); + _disposed = true; } + } - if (YahaEventSource.Log.IsEnabled()) YahaEventSource.Log.Info($"Dispose {nameof(NativeHttpHandlerCore)}; disposing={disposing}"); + private unsafe ICollection _DisableAndDrainCallbacks() + { + NativeMethods.yaha_context_disable_callbacks(_handle.DangerousGet()); + // The owner's admission coordination guarantees no request can + // register after this snapshot and escape cancellation and release. + var requests = _requests.Values; + foreach (var request in requests) + { + request.Response.Cancel(); + } - NativeRuntime.Instance.Release(); // We always need to release runtime. + NativeMethods.yaha_context_wait_callbacks(_handle.DangerousGet()); + return requests; + } - if (disposing) + private static void _ReleaseDrainedRequests(ICollection requests) + { + // Disabled callbacks no longer deliver OnComplete, and runtime + // shutdown cannot rely on queued ThreadPool work for cleanup. + foreach (var request in requests) { - _handle.Dispose(); + request.Release(); + request.Dispose(); } - - _disposed = true; } } } diff --git a/src/YetAnotherHttpHandler/NativeMethods.Uwp.g.cs b/src/YetAnotherHttpHandler/NativeMethods.Uwp.g.cs index f6e1a33..cbdd2d1 100644 --- a/src/YetAnotherHttpHandler/NativeMethods.Uwp.g.cs +++ b/src/YetAnotherHttpHandler/NativeMethods.Uwp.g.cs @@ -44,6 +44,12 @@ internal static unsafe partial class NativeMethods [DllImport(__DllName, EntryPoint = "yaha_dispose_context", CallingConvention = CallingConvention.Cdecl, ExactSpelling = true)] public static extern void yaha_dispose_context(YahaNativeContext* ctx); + [DllImport(__DllName, EntryPoint = "yaha_context_disable_callbacks", CallingConvention = CallingConvention.Cdecl, ExactSpelling = true)] + public static extern void yaha_context_disable_callbacks(YahaNativeContext* ctx); + + [DllImport(__DllName, EntryPoint = "yaha_context_wait_callbacks", CallingConvention = CallingConvention.Cdecl, ExactSpelling = true)] + public static extern void yaha_context_wait_callbacks(YahaNativeContext* ctx); + [DllImport(__DllName, EntryPoint = "yaha_client_config_add_root_certificates", CallingConvention = CallingConvention.Cdecl, ExactSpelling = true)] public static extern nuint yaha_client_config_add_root_certificates(YahaNativeContext* ctx, StringBuffer* root_certs); diff --git a/src/YetAnotherHttpHandler/NativeMethods.g.cs b/src/YetAnotherHttpHandler/NativeMethods.g.cs index b060637..bbf24c9 100644 --- a/src/YetAnotherHttpHandler/NativeMethods.g.cs +++ b/src/YetAnotherHttpHandler/NativeMethods.g.cs @@ -49,6 +49,12 @@ internal static unsafe partial class NativeMethods [DllImport(__DllName, EntryPoint = "yaha_dispose_context", CallingConvention = CallingConvention.Cdecl, ExactSpelling = true)] public static extern void yaha_dispose_context(YahaNativeContext* ctx); + [DllImport(__DllName, EntryPoint = "yaha_context_disable_callbacks", CallingConvention = CallingConvention.Cdecl, ExactSpelling = true)] + public static extern void yaha_context_disable_callbacks(YahaNativeContext* ctx); + + [DllImport(__DllName, EntryPoint = "yaha_context_wait_callbacks", CallingConvention = CallingConvention.Cdecl, ExactSpelling = true)] + public static extern void yaha_context_wait_callbacks(YahaNativeContext* ctx); + [DllImport(__DllName, EntryPoint = "yaha_client_config_add_root_certificates", CallingConvention = CallingConvention.Cdecl, ExactSpelling = true)] public static extern nuint yaha_client_config_add_root_certificates(YahaNativeContext* ctx, StringBuffer* root_certs); diff --git a/src/YetAnotherHttpHandler/RequestContext.cs b/src/YetAnotherHttpHandler/RequestContext.cs index e2292be..eec73fd 100644 --- a/src/YetAnotherHttpHandler/RequestContext.cs +++ b/src/YetAnotherHttpHandler/RequestContext.cs @@ -13,6 +13,9 @@ namespace Cysharp.Net.Http { internal class RequestContext : IDisposable { + private readonly NativeHttpHandlerCore _owner; + private readonly object _disposeLock = new object(); + private bool _disposed; private readonly Pipe _pipe = new(PipeOptions.Default); private readonly CancellationTokenSource _cancellationTokenSource; private readonly int _requestSequence; @@ -34,8 +37,9 @@ internal class RequestContext : IDisposable public PipeWriter Writer => _pipe.Writer; public IntPtr Handle => GCHandle.ToIntPtr(_handle); - internal RequestContext(YahaContextSafeHandle ctx, YahaRequestContextSafeHandle requestContext, HttpRequestMessage requestMessage, int requestSequence, PipeOptions? responsePipeOptions, CancellationToken cancellationToken) + internal RequestContext(NativeHttpHandlerCore owner, YahaContextSafeHandle ctx, YahaRequestContextSafeHandle requestContext, HttpRequestMessage requestMessage, int requestSequence, PipeOptions? responsePipeOptions, CancellationToken cancellationToken) { + _owner = owner; _ctxHandle = ctx; _requestContextHandle = requestContext; _response = new ResponseContext(requestMessage, this, responsePipeOptions, cancellationToken); @@ -49,14 +53,21 @@ internal RequestContext(YahaContextSafeHandle ctx, YahaRequestContextSafeHandle internal void Start(bool hasBody) { - if (hasBody) + lock (_disposeLock) { - _readRequestTask = RunReadRequestLoopAsync(_cancellationTokenSource.Token); - } - else - { - if (YahaEventSource.Log.IsEnabled()) YahaEventSource.Log.Trace($"[ReqSeq:{_requestSequence}:State:0x{Handle:X}] The request has no body. Complete immediately."); - TryCompleteBody(); + // A very fast response can finish before request_begin returns. + if (_disposed) + { + return; + } + if (hasBody) + { + _readRequestTask = RunReadRequestLoopAsync(_cancellationTokenSource.Token); + } + else + { + TryCompleteBody(); + } } } @@ -75,9 +86,12 @@ public void Allocate() /// public void Release() { - Debug.Assert(_handle.IsAllocated); lock (_handleLock) { + if (!_handle.IsAllocated) + { + return; + } if (YahaEventSource.Log.IsEnabled()) YahaEventSource.Log.Trace($"[ReqSeq:{_requestSequence}:State:0x{Handle:X}] Releasing state"); _handle.Free(); _handle = default; @@ -141,7 +155,7 @@ private void WriteBody(Span data) { // TODO: retryAfter += retryInterval; - Thread.Sleep(Math.Min(1000, retryAfter)); + _cancellationTokenSource.Token.WaitHandle.WaitOne(Math.Min(1000, retryAfter)); } } @@ -352,27 +366,38 @@ public void Dispose() private void Dispose(bool disposing) { - if (YahaEventSource.Log.IsEnabled()) YahaEventSource.Log.Info($"[ReqSeq:{_requestSequence}:State:0x{Handle:X}] Dispose RequestContext: disposing={disposing}"); + lock (_disposeLock) + { + if (_disposed) + { + return; + } - // Abort the request and dispose the request context handle whether called from manual Dispose or the finalizer. - TryAbort(); + TryAbort(); + _fullyCompleted.Wait(); + _cancellationTokenSource.Cancel(); + + _WaitForUploadLoopToStop(); + TryReleaseNativeHandles(); + if (disposing) + { + _cancellationTokenSource.Dispose(); + } + _disposed = true; + _owner.RemoveRequest(this); + } + } - if (disposing) + private void _WaitForUploadLoopToStop() + { + if (_readRequestTask != null) { - _cancellationTokenSource.Cancel(); - _cancellationTokenSource.Dispose(); - // DO NOT Dispose `_ctx` here. + _readRequestTask.GetAwaiter().GetResult(); } else { - // Executing within the finalizer thread. - // NOTE: Waits by blocking until the request is completed on the native side. - // If not waited here, issues such as crashes may occur when callbacks are invoked after the .NET side is destroyed by Unity's Domain Reload. - // However, caution is needed with the invocation order and timing of callbacks, as well as the handling of locks, since the finalizer thread may become blocked. - _fullyCompleted.Wait(); + TryCompleteBody(); } - - TryReleaseNativeHandles(); } } -} \ No newline at end of file +} diff --git a/src/YetAnotherHttpHandler/ResponseContext.cs b/src/YetAnotherHttpHandler/ResponseContext.cs index c0349e3..419485d 100644 --- a/src/YetAnotherHttpHandler/ResponseContext.cs +++ b/src/YetAnotherHttpHandler/ResponseContext.cs @@ -21,6 +21,7 @@ internal class ResponseContext private readonly object _writeLock = new object(); private bool _completed = false; private Task? _latestFlushTask; + private IntPtr _deferredAck; internal ResponseContext(HttpRequestMessage requestMessage, RequestContext requestContext, PipeOptions? pipeOptions, CancellationToken cancellationToken) { @@ -44,8 +45,15 @@ internal ResponseContext(HttpRequestMessage requestMessage, RequestContext reque #endif } - public ValueTask WriteAsync(ReadOnlySpan data) + /// + /// Writes response data. When the flush does not complete synchronously, + /// is held as a deferred acknowledgement ( is true) and must be + /// claimed with before completing it, even if the returned task has + /// already completed; may claim and complete it first. + /// + public ValueTask WriteAsync(ReadOnlySpan data, nuint ackHandle, out bool ackDeferred) { + ackDeferred = false; lock (_writeLock) { if (_completed) return default; @@ -64,10 +72,18 @@ public ValueTask WriteAsync(ReadOnlySpan data) } _latestFlushTask = flush.AsTask(); + _deferredAck = (IntPtr)(nint)ackHandle; + ackDeferred = true; return new ValueTask(_latestFlushTask); } } + public bool TryTakeDeferredAck(nuint ackHandle) + { + var handle = (IntPtr)(nint)ackHandle; + return Interlocked.CompareExchange(ref _deferredAck, IntPtr.Zero, handle) == handle; + } + public void SetHeader(ReadOnlySpan nameBytes, ReadOnlySpan valueBytes) { var (name, isHttpContentHeader) = Utf8Strings.HttpHeaders.FromSpan(nameBytes); @@ -164,17 +180,35 @@ public void Cancel() { if (YahaEventSource.Log.IsEnabled()) YahaEventSource.Log.Trace($"[ReqSeq:{_requestContext.RequestSequence}] Response was cancelled"); + // Unblock a full pipe before taking _writeLock: a callback thread can be + // blocked in WaitForLatestFlush() while holding it. + // Covered by Dispose_UnblocksUnreadResponseBody. + _pipe.Writer.CancelPendingFlush(); lock (_writeLock) { if (_completed) return; _requestContext.TryAbort(); _responseTask.TrySetCanceled(_cancellationToken); - WaitForLatestFlush(); + // Do not wait for the latest flush task: its continuation runs on + // the PipeOptions writer scheduler, which may be the thread calling + // Cancel (e.g. a main-thread scheduler during Dispose). No write can + // start while _writeLock is held, so cancelling here leaves no + // pending flush when the writer is completed. + _pipe.Writer.CancelPendingFlush(); _pipe.Writer.Complete(new OperationCanceledException(_cancellationToken)); _completed = true; } + // Acknowledge a deferred OnReceive without waiting for its flush + // continuation; Dispose waits for every acknowledgement to be released. + // Covered by Dispose_DoesNotWaitForFlushContinuationOnDisposingThread. + var deferredAck = Interlocked.Exchange(ref _deferredAck, IntPtr.Zero); + if (deferredAck != IntPtr.Zero) + { + NativeHttpHandlerCore.CompleteTask((nuint)(nint)deferredAck, "The response was cancelled."); + } + _tokenRegistration.Dispose(); } diff --git a/src/YetAnotherHttpHandler/YahaSafeHandles.cs b/src/YetAnotherHttpHandler/YahaSafeHandles.cs index e6df699..dedc378 100644 --- a/src/YetAnotherHttpHandler/YahaSafeHandles.cs +++ b/src/YetAnotherHttpHandler/YahaSafeHandles.cs @@ -63,6 +63,8 @@ protected override bool ReleaseHandle() NativeMethods.yaha_dispose_context((YahaNativeContext*)handle); _parent?.DangerousRelease(); + // Native owners may still retain the context, but yaha_dispose_context has + // closed and drained its callback gate, so they never use this handle again. _onVerifyServerCertificateHandle?.Free(); return true; diff --git a/src/YetAnotherHttpHandler/YetAnotherHttpHandler.cs b/src/YetAnotherHttpHandler/YetAnotherHttpHandler.cs index 42f60b7..24339fb 100644 --- a/src/YetAnotherHttpHandler/YetAnotherHttpHandler.cs +++ b/src/YetAnotherHttpHandler/YetAnotherHttpHandler.cs @@ -13,6 +13,7 @@ namespace Cysharp.Net.Http public class YetAnotherHttpHandler : HttpMessageHandler { private readonly NativeClientSettings _settings = new NativeClientSettings(); + private readonly object _lifetimeLock = new object(); private bool _disposed; private NativeHttpHandlerCore? _handler; @@ -207,16 +208,29 @@ private NativeHttpHandlerCore SetupHandler() protected override Task SendAsync(HttpRequestMessage request, CancellationToken cancellationToken) { - ThrowIfDisposed(); - var handler = _handler ?? SetupHandler(); - return handler.SendAsync(request, cancellationToken); + lock (_lifetimeLock) + { + ThrowIfDisposed(); + var handler = _handler ?? SetupHandler(); + return handler.SendAsync(request, cancellationToken); + } } protected override void Dispose(bool disposing) { - _handler?.Dispose(); + NativeHttpHandlerCore.ThrowIfInCallback(); + NativeHttpHandlerCore? handler; + lock (_lifetimeLock) + { + _disposed = true; + handler = _handler; + } + + // Do not hold the admission lock while joining callbacks: user + // certificate verification may try to send another request, which + // must observe ObjectDisposedException instead of deadlocking. + handler?.Dispose(); _handler = null; - _disposed = true; } private void ThrowIfDisposed() diff --git a/test/YetAnotherHttpHandler.Test/CertificateVerificationCallbackLifetimeTest.cs b/test/YetAnotherHttpHandler.Test/CertificateVerificationCallbackLifetimeTest.cs index 8605103..e77f317 100644 --- a/test/YetAnotherHttpHandler.Test/CertificateVerificationCallbackLifetimeTest.cs +++ b/test/YetAnotherHttpHandler.Test/CertificateVerificationCallbackLifetimeTest.cs @@ -90,7 +90,9 @@ private static async Task RunHandshakeScenarioAsync() await accepted.Task.WaitAsync(timeout.Token); // Capture the GCHandle owned by the native context before disposing the handler. - // The active request must keep this exact callback alive until TLS verification finishes. + // Dispose drains the request, so the handle may be freed even though the native + // TLS task still retains the context; its closed callback gate must then reject + // verification instead of invoking whatever now occupies the freed slot. var previousCallbackState = GetCallbackState(handler); handler.Dispose(); var callbackRetainedAfterDispose = IsCallbackTarget(previousCallbackState, strictVerifier); @@ -114,7 +116,6 @@ private static async Task RunHandshakeScenarioAsync() { await serverTask.WaitAsync(timeout.Token); Assert.Equal(0, Volatile.Read(ref replacementCalls)); - Assert.True(callbackRetainedAfterDispose, "The certificate verifier was released while its TLS request was still active."); Assert.Null(response); } } diff --git a/test/YetAnotherHttpHandler.Test/DisposeTest.cs b/test/YetAnotherHttpHandler.Test/DisposeTest.cs new file mode 100644 index 0000000..0bf0919 --- /dev/null +++ b/test/YetAnotherHttpHandler.Test/DisposeTest.cs @@ -0,0 +1,295 @@ +using System.IO.Pipelines; +using Cysharp.Net.Http; + +namespace _YetAnotherHttpHandler.Test; + +[Collection(nameof(YetAnotherHttpHandlerTest))] +public class DisposeTest(ITestOutputHelper testOutputHelper) : UseTestServerTestBase(testOutputHelper) +{ + protected override TimeSpan UnexpectedTimeout => TimeSpan.FromSeconds(30); + + [Fact] + public async Task Dispose_ReleasesRequestsThatFailBeforeNativeWorkStarts() + { + using var handler = new NativeHttpHandlerCore(new NativeClientSettings()); + using var request = new HttpRequestMessage(HttpMethod.Get, new Uri("relative", UriKind.Relative)); + // Call the core directly so URI conversion fails after native request + // allocation, instead of HttpClient rejecting it before allocation. + await Assert.ThrowsAsync(() => handler.SendAsync(request, CancellationToken.None)); + + var runtime = NativeRuntime.Instance.Acquire(); + NativeRuntime.Instance.Release(); + handler.Dispose(); + + // A failed, unregistered request must not keep the runtime alive until + // GC finalizes its SafeHandle. No GC.Collect is needed for disposal. + Assert.True(runtime.IsClosed); + } + + [Fact] + public async Task Dispose_CancelsPendingHeaders_WhileAnotherHandlerKeepsWorking() + { + await using var server = await LaunchServerAsync(TestServerListenMode.InsecureHttp1Only); + using var handler = new YetAnotherHttpHandler(); + using var client = new HttpClient(handler, disposeHandler: false); + using var otherHandler = new YetAnotherHttpHandler(); + using var otherClient = new HttpClient(otherHandler, disposeHandler: false); + Assert.Equal("__OK__", await otherClient.GetStringAsync(server.BaseUri, TimeoutToken)); + + var response = client.GetAsync($"{server.BaseUri}/slow-response-headers", TimeoutToken); + await _DisposeOnDedicatedThread(handler).WaitAsync(TimeoutToken); + + // HttpClient's managed continuations can run after native work stops. + await _AssertRequestStoppedAsync(response); + Assert.Equal("__OK__", await otherClient.GetStringAsync(server.BaseUri, TimeoutToken)); + await Assert.ThrowsAsync(() => client.GetAsync(server.BaseUri, TimeoutToken)); + handler.Dispose(); + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task Dispose_UnblocksUnreadResponseBody(bool http2) + { + await using var server = await LaunchServerAsync(http2 + ? TestServerListenMode.InsecureHttp2Only : TestServerListenMode.InsecureHttp1Only); + using var handler = new YetAnotherHttpHandler + { + Http2Only = http2, + ResponsePipeOptions = new PipeOptions(pauseWriterThreshold: 1, resumeWriterThreshold: 1, useSynchronizationContext: false), + }; + using var client = new HttpClient(handler, disposeHandler: false); + // No request cancellation token: Dispose itself must unblock the flush. + using var response = await client.GetAsync($"{server.BaseUri}/random?size=1048576", HttpCompletionOption.ResponseHeadersRead).WaitAsync(TimeoutToken); + using var body = await response.Content.ReadAsStreamAsync(TimeoutToken); + Assert.Equal(1, await body.ReadAsync(new byte[1], TimeoutToken)); + + await _DisposeOnDedicatedThread(handler).WaitAsync(TimeoutToken); + + await _AssertRequestStoppedAsync(body.CopyToAsync(Stream.Null, TimeoutToken)); + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task Dispose_DoesNotWaitForFlushContinuationOnDisposingThread(bool useWriterScheduler) + { + await using var server = await LaunchServerAsync(TestServerListenMode.InsecureHttp1Only); + var disposingThread = new DisposingThreadQueue(); + using var handler = new YetAnotherHttpHandler + { + ResponsePipeOptions = useWriterScheduler + ? new PipeOptions(writerScheduler: disposingThread, pauseWriterThreshold: 1, resumeWriterThreshold: 1, useSynchronizationContext: false) + : new PipeOptions(pauseWriterThreshold: 1, resumeWriterThreshold: 1, useSynchronizationContext: true), + }; + using var client = new HttpClient(handler, disposeHandler: false); + var previousContext = SynchronizationContext.Current; + SynchronizationContext.SetSynchronizationContext(disposingThread); + HttpResponseMessage response; + try + { + response = await client.GetAsync($"{server.BaseUri}/random?size=1048576", HttpCompletionOption.ResponseHeadersRead).WaitAsync(TimeoutToken).ConfigureAwait(false); + } + finally + { + SynchronizationContext.SetSynchronizationContext(previousContext); + } + using var _ = response; + using var body = await response.Content.ReadAsStreamAsync(TimeoutToken); + Assert.Equal(1, await body.ReadAsync(new byte[1], TimeoutToken)); + // Let OnReceive hit backpressure and defer its acknowledgement. + await Task.Delay(200, TimeoutToken); + + // The flush continuation is posted to the disposing thread, which is + // blocked in Dispose and cannot run it until Dispose returns. + var dispose = Task.Factory.StartNew(() => + { + SynchronizationContext.SetSynchronizationContext(disposingThread); + handler.Dispose(); + }, CancellationToken.None, TaskCreationOptions.LongRunning, TaskScheduler.Default); + try + { + await dispose.WaitAsync(TimeSpan.FromSeconds(10)); + } + finally + { + // Unblock a deadlocked Dispose so a failure does not hang the run. + disposingThread.DrainOnThreadPool(); + } + + await _AssertRequestStoppedAsync(body.CopyToAsync(Stream.Null, TimeoutToken)); + } + + private sealed class DisposingThreadQueue : PipeScheduler + { + private readonly object _lock = new(); + private readonly List<(Action Action, object? State)> _queue = new(); + private bool _drained; + + public override void Schedule(Action action, object? state) + { + lock (_lock) + { + if (!_drained) + { + _queue.Add((action, state)); + return; + } + } + System.Threading.ThreadPool.UnsafeQueueUserWorkItem(action, state, preferLocal: false); + } + + public void DrainOnThreadPool() + { + lock (_lock) + { + _drained = true; + foreach (var (action, state) in _queue) + { + System.Threading.ThreadPool.UnsafeQueueUserWorkItem(action, state, preferLocal: false); + } + _queue.Clear(); + } + } + + public static implicit operator SynchronizationContext(DisposingThreadQueue queue) => new QueueSynchronizationContext(queue); + } + + private sealed class QueueSynchronizationContext(DisposingThreadQueue queue) : SynchronizationContext + { + public override void Post(SendOrPostCallback d, object? state) => queue.Schedule(s => d(s), state); + public override SynchronizationContext CreateCopy() => this; + } + + [Fact] + public async Task Dispose_UnblocksUploadToServerThatDoesNotRead() + { + await using var server = await LaunchServerAsync(TestServerListenMode.InsecureHttp2Only); + using var handler = new YetAnotherHttpHandler + { + Http2Only = true, + }; + using var client = new HttpClient(handler, disposeHandler: false); + using var request = new HttpRequestMessage(HttpMethod.Post, $"{server.BaseUri}/post-never-read") + { + Content = new ByteArrayContent(new byte[8 * 1024 * 1024]), + }; + using var response = await client.SendAsync(request, HttpCompletionOption.ResponseHeadersRead).WaitAsync(TimeoutToken); + + await _DisposeOnDedicatedThread(handler).WaitAsync(TimeoutToken); + + await _AssertRequestStoppedAsync(response.Content.CopyToAsync(Stream.Null, TimeoutToken)); + } + + [Fact] + public async Task Dispose_WaitsForCertificateCallback_AndConcurrentDispose() + { + await using var server = await LaunchServerAsync(TestServerListenMode.SecureHttp2Only); + using var callbackRelease = new ManualResetEventSlim(); + var callbackEntered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var callbackReturned = false; + using var handler = new YetAnotherHttpHandler + { + OnVerifyServerCertificate = (_, _, _) => + { + callbackEntered.TrySetResult(); + callbackRelease.Wait(TimeoutToken); + callbackReturned = true; + return true; + }, + }; + using var client = new HttpClient(handler, disposeHandler: false); + var response = client.GetAsync(server.BaseUri); + Task? first = null; + Task? second = null; + try + { + await callbackEntered.Task.WaitAsync(TimeoutToken); + first = _DisposeOnDedicatedThread(handler); + // Cancellation of the pending response proves Dispose has started. + await _AssertRequestStoppedAsync(response); + Assert.False(first.IsCompleted); + second = _DisposeOnDedicatedThread(handler); + // The first Dispose is held inside the verifier. A second caller + // must not return early just because the public handler is closed. + Assert.NotSame(second, await Task.WhenAny(second, Task.Delay(100, TimeoutToken))); + await Assert.ThrowsAsync(() => client.GetAsync(server.BaseUri)); + } + finally + { + callbackRelease.Set(); + if (first != null) + { + await first.WaitAsync(TimeoutToken); + } + if (second != null) + { + await second.WaitAsync(TimeoutToken); + } + } + Assert.True(callbackReturned); + } + + [Fact] + public async Task Dispose_FromCertificateCallback_IsRejectedWithoutDisposingHandler() + { + await using var server = await LaunchServerAsync(TestServerListenMode.SecureHttp2Only); + using var handler = new YetAnotherHttpHandler(); + Exception? disposalError = null; + handler.OnVerifyServerCertificate = (_, _, _) => + { + disposalError = Record.Exception(handler.Dispose); + return true; + }; + using var client = new HttpClient(handler, disposeHandler: false); + + Assert.Equal("__OK__", await client.GetStringAsync(server.BaseUri, TimeoutToken)); + Assert.IsType(disposalError); + Assert.Equal("__OK__", await client.GetStringAsync(server.BaseUri, TimeoutToken)); + await _DisposeOnDedicatedThread(handler).WaitAsync(TimeoutToken); + } + + [Fact] + public async Task Dispose_RacesWithRequestAdmission() + { + await using var server = await LaunchServerAsync(TestServerListenMode.InsecureHttp1Only); + for (var iteration = 0; iteration < 30; iteration++) + { + using var handler = new YetAnotherHttpHandler(); + using var client = new HttpClient(handler, disposeHandler: false); + using var start = new ManualResetEventSlim(); + var sends = Enumerable.Range(0, 4).Select(_ => Task.Run(async () => + { + start.Wait(TimeoutToken); + try + { + using var response = await client.GetAsync(server.BaseUri, TimeoutToken); + response.EnsureSuccessStatusCode(); + } + catch (Exception ex) when (ex is ObjectDisposedException or HttpRequestException or OperationCanceledException) + { + // Either admitted and cancelled, or rejected at admission. + } + })).ToArray(); + var dispose = Task.Run(() => + { + start.Wait(TimeoutToken); + handler.Dispose(); + }); + start.Set(); + await Task.WhenAll(sends.Append(dispose)).WaitAsync(TimeoutToken); + } + } + + private static Task _DisposeOnDedicatedThread(YetAnotherHttpHandler handler) + { + return Task.Factory.StartNew(handler.Dispose, CancellationToken.None, TaskCreationOptions.LongRunning, TaskScheduler.Default); + } + + private static async Task _AssertRequestStoppedAsync(Task request) + { + var error = await Record.ExceptionAsync(() => request.WaitAsync(TimeSpan.FromSeconds(5))); + Assert.True(error is HttpRequestException or OperationCanceledException or IOException, + $"Expected request cancellation/failure, got {error}"); + } +}