Merge pull request #116 from ruccho/feature/backpressure

mayuki · web-flow · commit fd748dae5b39 · 2025-02-03T16:07:36.000+09:00
Add backpressure control and allow asynchronous flushing
diff --git a/README.md b/README.md
@@ -231,6 +231,7 @@ Once the handler sends a request, these settings become immutable and cannot be
 |Http2MaxSendBufferSize|Gets or sets the maximum write buffer size for each HTTP/2 stream. Default is currently 1MB, but may change.|
 |Http2InitialMaxSendStreams|Gets or sets the initial maximum of locally initiated (send) streams. This value will be overwritten by the value included in the initial SETTINGS frame received from the peer as part of a connection preface.|
 |UnixDomainSocketPath|Gets or sets the path to a Unix Domain Socket to be used as HTTP communication channel instead of the default TCP.|
+|ResponsePipeOptions|Gets or sets the options for the pipe used to receive the response body.|
 
 Most of them expose [hyper client settings](https://docs.rs/hyper-util/latest/hyper_util/client/legacy/struct.Builder.html), so please check those as well.
 
diff --git a/native/yaha_native/src/binding.rs b/native/yaha_native/src/binding.rs
@@ -9,7 +9,7 @@ use hyper::{
     Request, StatusCode, Uri, Version,
 };
 use rustls::pki_types::{CertificateDer, PrivateKeyDer};
-use tokio::select;
+use tokio::{select, sync::oneshot};
 use tokio_util::sync::CancellationToken;
 
 use crate::interop::{ByteBuffer, StringBuffer};
@@ -65,7 +65,7 @@ pub extern "C" fn yaha_init_context(
         status_code: i32,
         version: YahaHttpVersion,
     ),
-    on_receive: extern "C" fn(req_seq: i32, state: NonZeroIsize, length: usize, buf: *const u8),
+    on_receive: extern "C" fn(req_seq: i32, state: NonZeroIsize, length: usize, buf: *const u8, task_handle: usize),
     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);
@@ -86,7 +86,7 @@ pub extern "C" fn yaha_dispose_context(ctx: *mut YahaNativeContext) {
     ctx.on_status_code_and_headers_receive = _sentinel_on_status_code_and_headers_receive;
 }
 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) { panic!("The context has already disposed: on_receive"); }
+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]
@@ -347,7 +347,6 @@ pub extern "C" fn yaha_client_config_unix_domain_socket_path(
     ctx.uds_socket_path.get_or_insert(uds_socket_path.into());
 }
 
-
 #[no_mangle]
 pub extern "C" fn yaha_build_client(ctx: *mut YahaNativeContext) {
     let ctx = YahaNativeContextInternal::from_raw_context(ctx);
@@ -582,7 +581,30 @@ pub extern "C" fn yaha_request_begin(
                                     Ok(frame) => {
                                         if frame.is_data() {
                                             let data = frame.into_data().unwrap();
-                                            (ctx.on_receive)(seq, state, data.len(), data.as_ptr());
+                                            let (tx, rx) = oneshot::channel::<Result<(), String>>();
+                                            let tx = Box::into_raw(Box::new(tx)) as usize;
+
+                                            (ctx.on_receive)(seq, state, data.len(), data.as_ptr(), tx);
+                                            match rx.await {
+                                                Ok(result) => {
+                                                    if let Err(err) = result {
+                                                        // the sender reports an error
+                                                        LAST_ERROR.with(|v| {
+                                                            *v.borrow_mut() = Some(err);
+                                                        });
+                                                        (ctx.on_complete)(seq, state, CompletionReason::Error, 0);
+                                                        return;
+                                                    }
+                                                },
+                                                Err(e) => {
+                                                    // the sender is dropped without sending
+                                                    LAST_ERROR.with(|v| {
+                                                        *v.borrow_mut() = Some("on_receive() has not completed correctly.".to_string());
+                                                    });
+                                                    (ctx.on_complete)(seq, state, CompletionReason::Error, 0);
+                                                    return;
+                                                }
+                                            }
                                         } else if frame.is_trailers() {
 
                                             trailer_received = true;
@@ -815,3 +837,14 @@ pub extern "C" fn yaha_request_destroy(
     let req_ctx = crate::context::to_internal_arc(req_ctx);
     true
 }
+
+#[no_mangle]
+pub extern "C" fn yaha_complete_task(task_handle: usize, error: *const StringBuffer) {
+    let tx = unsafe { Box::from_raw(task_handle as *mut oneshot::Sender<Result<(), String>>) };
+    if error.is_null() {
+        tx.send(Ok(())).unwrap();
+    } else {
+        let error = unsafe { (*error).to_str().to_string() };
+        tx.send(Err(error)).unwrap();
+    }
+}
diff --git a/native/yaha_native/src/context.rs b/native/yaha_native/src/context.rs
@@ -33,7 +33,7 @@ use crate::{primitives::{CompletionReason, YahaHttpVersion}};
 
 type OnStatusCodeAndHeadersReceive =
     extern "C" fn(req_seq: i32, state: NonZeroIsize, status_code: i32, version: YahaHttpVersion);
-type OnReceive = extern "C" fn(req_seq: i32, state: NonZeroIsize, length: usize, buf: *const u8);
+type OnReceive = extern "C" fn(req_seq: i32, state: NonZeroIsize, length: usize, buf: *const u8, task_handle: usize);
 type OnComplete = extern "C" fn(req_seq: i32, state: NonZeroIsize, reason: CompletionReason, h2_error_code: u32);
 type OnServerCertificateVerificationHandler = extern "C" fn(callback_state: NonZeroIsize, server_name: *const u8, server_name_len: usize, certificate_der: *const u8, certificate_der_len: usize, now: u64) -> bool;
 
diff --git a/src/YetAnotherHttpHandler/NativeHttpHandlerCore.cs b/src/YetAnotherHttpHandler/NativeHttpHandlerCore.cs
@@ -23,6 +23,7 @@ internal class NativeHttpHandlerCore : IDisposable
         private readonly YahaContextSafeHandle _handle;
         private GCHandle? _onVerifyServerCertificateHandle; // The handle must be released in Dispose if it is allocated.
         private bool _disposed = false;
+        private PipeOptions? _responsePipeOptions;
 
         // 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.
@@ -199,6 +200,12 @@ private unsafe void Initialize(YahaNativeContext* ctx, NativeClientSettings sett
                 }
             }
 
+            if (settings.ResponsePipeOptions is { } responsePipeOptions)
+            {
+                if (YahaEventSource.Log.IsEnabled()) YahaEventSource.Log.Info($"Option '{nameof(settings.ResponsePipeOptions)}' = {responsePipeOptions}");
+                _responsePipeOptions = responsePipeOptions;
+            }
+
             NativeMethods.yaha_build_client(ctx);
 
             if (YahaEventSource.Log.IsEnabled()) YahaEventSource.Log.Info($"{nameof(NativeHttpHandlerCore)} created");
@@ -327,7 +334,7 @@ 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, cancellationToken);
+            var requestContextManaged = new RequestContext(_handle, reqCtxHandle, request, requestSequence, _responsePipeOptions, cancellationToken);
             requestContextManaged.Allocate();
             if (cancellationToken.IsCancellationRequested)
             {
@@ -450,13 +457,63 @@ 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)
+        private static unsafe void OnReceive(int reqSeq, IntPtr state, UIntPtr length, byte* buf, nuint taskHandle)
         {
-            if (YahaEventSource.Log.IsEnabled()) YahaEventSource.Log.Trace($"[ReqSeq:{reqSeq}:State:0x{state:X}] Response data received: Length={length}");
+            try
+            {
+                if (YahaEventSource.Log.IsEnabled()) YahaEventSource.Log.Trace($"[ReqSeq:{reqSeq}:State:0x{state:X}] Response data received: Length={length}");
 
-            var bufSpan = new Span<byte>(buf, (int)length);
-            var requestContext = RequestContext.FromHandle(state);
-            requestContext.Response.Write(bufSpan);
+                var bufSpan = new Span<byte>(buf, (int)length);
+                var requestContext = RequestContext.FromHandle(state);
+                var write = requestContext.Response.WriteAsync(bufSpan);
+
+                if (write.IsCompleted)
+                {
+                    write.GetAwaiter().GetResult();
+                    CompleteTask(taskHandle);
+                }
+                else
+                {
+                    // backpressure is occurred
+                    var awaiter = write.GetAwaiter();
+                    awaiter.UnsafeOnCompleted(() =>
+                    {
+                        try
+                        {
+                            awaiter.GetResult();
+                        }
+                        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());
+                            return;
+                        }
+
+                        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());
+            }
+        }
+
+        private static unsafe void CompleteTask(nuint taskHandle, string? error = null)
+        {
+            if (error is null)
+            {
+                NativeMethods.yaha_complete_task(taskHandle, (StringBuffer*)0);
+                return;
+            }
+
+            using var messageUtf8 = new TempUtf8String(error);
+            fixed (byte* messagePtr = messageUtf8.Span)
+            {
+                var sb = new StringBuffer(messagePtr, messageUtf8.Span.Length);
+                NativeMethods.yaha_complete_task((nuint)(nint)taskHandle, &sb);
+            }
         }
 
         [MonoPInvokeCallback(typeof(NativeMethods.yaha_init_context_on_complete_delegate))]
diff --git a/src/YetAnotherHttpHandler/NativeMethods.Uwp.g.cs b/src/YetAnotherHttpHandler/NativeMethods.Uwp.g.cs
@@ -33,7 +33,7 @@ internal static unsafe partial class NativeMethods
         public delegate void yaha_init_context_on_status_code_and_headers_receive_delegate(int req_seq, nint state, int status_code, YahaHttpVersion version);
 
         [UnmanagedFunctionPointer(CallingConvention.Cdecl)]
-        public delegate void yaha_init_context_on_receive_delegate(int req_seq, nint state, nuint length, byte* buf);
+        public delegate void yaha_init_context_on_receive_delegate(int req_seq, nint state, nuint length, byte* buf, nuint task_handle);
 
         [UnmanagedFunctionPointer(CallingConvention.Cdecl)]
         public delegate void yaha_init_context_on_complete_delegate(int req_seq, nint state, CompletionReason reason, uint h2_error_code);
@@ -172,6 +172,9 @@ internal static unsafe partial class NativeMethods
         [return: MarshalAs(UnmanagedType.U1)]
         public static extern bool yaha_request_destroy(YahaNativeContext* ctx, YahaNativeRequestContext* req_ctx);
 
+        [DllImport(__DllName, EntryPoint = "yaha_complete_task", CallingConvention = CallingConvention.Cdecl, ExactSpelling = true)]
+        public static extern void yaha_complete_task(nuint task_handle, StringBuffer* error);
+
 
     }
 
diff --git a/src/YetAnotherHttpHandler/NativeMethods.g.cs b/src/YetAnotherHttpHandler/NativeMethods.g.cs
@@ -38,7 +38,7 @@ internal static unsafe partial class NativeMethods
         public delegate void yaha_init_context_on_status_code_and_headers_receive_delegate(int req_seq, nint state, int status_code, YahaHttpVersion version);
 
         [UnmanagedFunctionPointer(CallingConvention.Cdecl)]
-        public delegate void yaha_init_context_on_receive_delegate(int req_seq, nint state, nuint length, byte* buf);
+        public delegate void yaha_init_context_on_receive_delegate(int req_seq, nint state, nuint length, byte* buf, nuint task_handle);
 
         [UnmanagedFunctionPointer(CallingConvention.Cdecl)]
         public delegate void yaha_init_context_on_complete_delegate(int req_seq, nint state, CompletionReason reason, uint h2_error_code);
@@ -177,6 +177,9 @@ internal static unsafe partial class NativeMethods
         [return: MarshalAs(UnmanagedType.U1)]
         public static extern bool yaha_request_destroy(YahaNativeContext* ctx, YahaNativeRequestContext* req_ctx);
 
+        [DllImport(__DllName, EntryPoint = "yaha_complete_task", CallingConvention = CallingConvention.Cdecl, ExactSpelling = true)]
+        public static extern void yaha_complete_task(nuint task_handle, StringBuffer* error);
+
 
     }
 
diff --git a/src/YetAnotherHttpHandler/RequestContext.cs b/src/YetAnotherHttpHandler/RequestContext.cs
@@ -33,11 +33,11 @@ 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, CancellationToken cancellationToken)
+        internal RequestContext(YahaContextSafeHandle ctx, YahaRequestContextSafeHandle requestContext, HttpRequestMessage requestMessage, int requestSequence, PipeOptions? responsePipeOptions,  CancellationToken cancellationToken)
         {
             _ctxHandle = ctx;
             _requestContextHandle = requestContext;
-            _response = new ResponseContext(requestMessage, this, cancellationToken);
+            _response = new ResponseContext(requestMessage, this, responsePipeOptions, cancellationToken);
             _readRequestTask = default;
             _requestSequence = requestSequence;
             _cancellationTokenSource = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
diff --git a/src/YetAnotherHttpHandler/ResponseContext.cs b/src/YetAnotherHttpHandler/ResponseContext.cs
@@ -13,16 +13,18 @@ namespace Cysharp.Net.Http
     internal class ResponseContext
     {
         private readonly RequestContext _requestContext;
-        private readonly Pipe _pipe = new Pipe(System.IO.Pipelines.PipeOptions.Default);
+        private readonly Pipe _pipe;
         private readonly TaskCompletionSource<HttpResponseMessage> _responseTask;
         private readonly HttpResponseMessage _message;
         private readonly CancellationToken _cancellationToken;
         private readonly CancellationTokenRegistration _tokenRegistration;
         private readonly object _writeLock = new object();
         private bool _completed = false;
+        private Task<FlushResult>? _latestFlushTask;
 
-        internal ResponseContext(HttpRequestMessage requestMessage, RequestContext requestContext, CancellationToken cancellationToken)
+        internal ResponseContext(HttpRequestMessage requestMessage, RequestContext requestContext, PipeOptions? pipeOptions, CancellationToken cancellationToken)
         {
+            _pipe = new Pipe(pipeOptions ?? PipeOptions.Default);
             _requestContext = requestContext;
             _responseTask = new TaskCompletionSource<HttpResponseMessage>(TaskCreationOptions.RunContinuationsAsynchronously);
             _cancellationToken = cancellationToken;
@@ -42,22 +44,27 @@ internal ResponseContext(HttpRequestMessage requestMessage, RequestContext reque
 #endif
         }
 
-        public void Write(ReadOnlySpan<byte> data)
+        public ValueTask<FlushResult> WriteAsync(ReadOnlySpan<byte> data)
         {
-            // NOTE: Currently, this method is called from the rust-side thread (tokio-worker-thread),
-            //        so care must be taken because throwing a managed exception will cause a crash.
             lock (_writeLock)
             {
-                if (_completed) return;
+                if (_completed) return default;
+
+                WaitForLatestFlush();
 
                 var buffer = _pipe.Writer.GetSpan(data.Length);
                 data.CopyTo(buffer);
                 _pipe.Writer.Advance(data.Length);
-                var t = _pipe.Writer.FlushAsync();
-                if (!t.IsCompleted)
+
+                var flush = _pipe.Writer.FlushAsync(_cancellationToken);
+                if (flush.IsCompleted)
                 {
-                    t.AsTask().GetAwaiter().GetResult();
+                    _latestFlushTask = null;
+                    return flush;
                 }
+
+                _latestFlushTask = flush.AsTask();
+                return new ValueTask<FlushResult>(_latestFlushTask);
             }
         }
 
@@ -105,6 +112,7 @@ public void Complete()
             if (YahaEventSource.Log.IsEnabled()) YahaEventSource.Log.Trace($"[ReqSeq:{_requestContext.RequestSequence}] Response completed. (_completed={_completed})");
             lock (_writeLock)
             {
+                WaitForLatestFlush();
                 _pipe.Writer.Complete();
                 _completed = true;
             }
@@ -139,6 +147,7 @@ public void CompleteAsFailed(string errorMessage, uint h2ErrorCode)
                 }
 #endif
                 _responseTask.TrySetException(ex);
+                WaitForLatestFlush();
                 _pipe.Writer.Complete(ex);
                 _completed = true;
             }
@@ -152,6 +161,7 @@ public void Cancel()
             {
                 _requestContext.TryAbort();
                 _responseTask.TrySetCanceled(_cancellationToken);
+                WaitForLatestFlush();
                 _pipe.Writer.Complete(new OperationCanceledException(_cancellationToken));
                 _completed = true;
             }
@@ -177,6 +187,23 @@ public async Task<HttpResponseMessage> GetResponseAsync()
             }
         }
 
+        private void WaitForLatestFlush()
+        {
+            // PipeWriter is not thread-safe, so we need to wait for the latest flush task to complete before writing to the pipe.
+
+            if (_latestFlushTask is { IsCompleted: false } latestFlushTask)
+            {
+                try
+                {
+                    latestFlushTask.Wait();
+                }
+                catch (Exception)
+                {
+                    // It is safe to ignore an exception thrown by the latest flush task because it will be caught by NativeHttpHandlerCore.OnReceive().
+                }
+            }
+        }
+
         internal static class Http2ErrorCode
         {
             // https://github.com/dotnet/aspnetcore/blob/release/8.0/src/Shared/ServerInfrastructure/Http2/Http2ErrorCode.cs
diff --git a/src/YetAnotherHttpHandler/YetAnotherHttpHandler.cs b/src/YetAnotherHttpHandler/YetAnotherHttpHandler.cs
diff --git a/test/YetAnotherHttpHandler.Test/BackPressureTest.cs b/test/YetAnotherHttpHandler.Test/BackPressureTest.cs
diff --git a/test/YetAnotherHttpHandler.Test/TestServerForHttp1AndHttp2.cs b/test/YetAnotherHttpHandler.Test/TestServerForHttp1AndHttp2.cs

Original file line number	Diff line number	Diff line change
`@@ -13,16 +13,18 @@ namespace Cysharp.Net.Http`
`13`	`13`	`internal class ResponseContext`
`14`	`14`	`{`
`15`	`15`	`private readonly RequestContext _requestContext;`
`16`		`- private readonly Pipe _pipe = new Pipe(System.IO.Pipelines.PipeOptions.Default);`
	`16`	`+ private readonly Pipe _pipe;`
`17`	`17`	`private readonly TaskCompletionSource<HttpResponseMessage> _responseTask;`
`18`	`18`	`private readonly HttpResponseMessage _message;`
`19`	`19`	`private readonly CancellationToken _cancellationToken;`
`20`	`20`	`private readonly CancellationTokenRegistration _tokenRegistration;`
`21`	`21`	`private readonly object _writeLock = new object();`
`22`	`22`	`private bool _completed = false;`
	`23`	`+ private Task<FlushResult>? _latestFlushTask;`
`23`	`24`
`24`		`- internal ResponseContext(HttpRequestMessage requestMessage, RequestContext requestContext, CancellationToken cancellationToken)`
	`25`	`+ internal ResponseContext(HttpRequestMessage requestMessage, RequestContext requestContext, PipeOptions? pipeOptions, CancellationToken cancellationToken)`
`25`	`26`	`{`
	`27`	`+ _pipe = new Pipe(pipeOptions ?? PipeOptions.Default);`
`26`	`28`	`_requestContext = requestContext;`
`27`	`29`	`_responseTask = new TaskCompletionSource<HttpResponseMessage>(TaskCreationOptions.RunContinuationsAsynchronously);`
`28`	`30`	`_cancellationToken = cancellationToken;`
`@@ -42,22 +44,27 @@ internal ResponseContext(HttpRequestMessage requestMessage, RequestContext reque`
`42`	`44`	`#endif`
`43`	`45`	`}`
`44`	`46`
`45`		`- public void Write(ReadOnlySpan<byte> data)`
	`47`	`+ public ValueTask<FlushResult> WriteAsync(ReadOnlySpan<byte> data)`
`46`	`48`	`{`
`47`		`- // NOTE: Currently, this method is called from the rust-side thread (tokio-worker-thread),`
`48`		`- // so care must be taken because throwing a managed exception will cause a crash.`
`49`	`49`	`lock (_writeLock)`
`50`	`50`	`{`
`51`		`- if (_completed) return;`
	`51`	`+ if (_completed) return default;`
	`52`	`+`
	`53`	`+ WaitForLatestFlush();`
`52`	`54`
`53`	`55`	`var buffer = _pipe.Writer.GetSpan(data.Length);`
`54`	`56`	`data.CopyTo(buffer);`
`55`	`57`	`_pipe.Writer.Advance(data.Length);`
`56`		`- var t = _pipe.Writer.FlushAsync();`
`57`		`- if (!t.IsCompleted)`
	`58`	`+`
	`59`	`+ var flush = _pipe.Writer.FlushAsync(_cancellationToken);`
	`60`	`+ if (flush.IsCompleted)`
`58`	`61`	`{`
`59`		`- t.AsTask().GetAwaiter().GetResult();`
	`62`	`+ _latestFlushTask = null;`
	`63`	`+ return flush;`
`60`	`64`	`}`
	`65`	`+`
	`66`	`+ _latestFlushTask = flush.AsTask();`
	`67`	`+ return new ValueTask<FlushResult>(_latestFlushTask);`
`61`	`68`	`}`
`62`	`69`	`}`
`63`	`70`
`@@ -105,6 +112,7 @@ public void Complete()`
`105`	`112`	`if (YahaEventSource.Log.IsEnabled()) YahaEventSource.Log.Trace($"[ReqSeq:{_requestContext.RequestSequence}] Response completed. (_completed={_completed})");`
`106`	`113`	`lock (_writeLock)`
`107`	`114`	`{`
	`115`	`+ WaitForLatestFlush();`
`108`	`116`	`_pipe.Writer.Complete();`
`109`	`117`	`_completed = true;`
`110`	`118`	`}`
`@@ -139,6 +147,7 @@ public void CompleteAsFailed(string errorMessage, uint h2ErrorCode)`
`139`	`147`	`}`
`140`	`148`	`#endif`
`141`	`149`	`_responseTask.TrySetException(ex);`
	`150`	`+ WaitForLatestFlush();`
`142`	`151`	`_pipe.Writer.Complete(ex);`
`143`	`152`	`_completed = true;`
`144`	`153`	`}`
`@@ -152,6 +161,7 @@ public void Cancel()`
`152`	`161`	`{`
`153`	`162`	`_requestContext.TryAbort();`
`154`	`163`	`_responseTask.TrySetCanceled(_cancellationToken);`
	`164`	`+ WaitForLatestFlush();`
`155`	`165`	`_pipe.Writer.Complete(new OperationCanceledException(_cancellationToken));`
`156`	`166`	`_completed = true;`
`157`	`167`	`}`
`@@ -177,6 +187,23 @@ public async Task<HttpResponseMessage> GetResponseAsync()`
`177`	`187`	`}`
`178`	`188`	`}`
`179`	`189`
	`190`	`+ private void WaitForLatestFlush()`
	`191`	`+ {`
	`192`	`+ // PipeWriter is not thread-safe, so we need to wait for the latest flush task to complete before writing to the pipe.`
	`193`	`+`
	`194`	`+ if (_latestFlushTask is { IsCompleted: false } latestFlushTask)`
	`195`	`+ {`
	`196`	`+ try`
	`197`	`+ {`
	`198`	`+ latestFlushTask.Wait();`
	`199`	`+ }`
	`200`	`+ catch (Exception)`
	`201`	`+ {`
	`202`	`+ // It is safe to ignore an exception thrown by the latest flush task because it will be caught by NativeHttpHandlerCore.OnReceive().`
	`203`	`+ }`
	`204`	`+ }`
	`205`	`+ }`
	`206`	`+`
`180`	`207`	`internal static class Http2ErrorCode`
`181`	`208`	`{`
`182`	`209`	`// https://github.com/dotnet/aspnetcore/blob/release/8.0/src/Shared/ServerInfrastructure/Http2/Http2ErrorCode.cs`