Copilot commented on code in PR #3930: URL: https://github.com/apache/thrift/pull/3930#discussion_r4098418028
########## lib/netstd/Thrift/Transport/Client/THttpPerCallTransport.cs: ########## @@ -0,0 +1,179 @@ +// Licensed to the Apache Software Foundation(ASF) under one +// or more contributor license agreements.See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership.The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +using System; +using System.IO; +using System.Net.Http; +using System.Threading; +using System.Threading.Tasks; + +namespace Thrift.Transport.Client +{ + /// <summary> + /// Implementation of a per-call transport for THttpTransport. + /// Each call to the server will create a new HTTP request and response. + /// </summary> + internal class THttpPerCallTransport : TEndpointTransport + { + private readonly THttpTransport _parent; + private readonly MemoryStream _request = new MemoryStream(); + + private HttpResponseMessage _response; + private Stream _responseStream; + private CancellationTokenSource _responseReadTimeout; + private bool _sent; + private bool _disposed; + + public THttpPerCallTransport(THttpTransport parent) : base(parent.Configuration) + { + this._parent = parent; + } + + public override bool IsOpen => !_disposed; + + public override Task WriteAsync( + byte[] buffer, + int offset, + int length, + CancellationToken cancellationToken) + { + if (_sent) + { + throw new InvalidOperationException("The request has already been sent."); + } + + return _request.WriteAsync( + buffer, + offset, + length, + cancellationToken); + } + + public override async Task FlushAsync( + CancellationToken cancellationToken) + { + if (_sent) + { + throw new InvalidOperationException("The request has already been sent."); + } + + _request.Position = 0; + var response = await _parent.SendAsync(_request, cancellationToken); + + var responseReadTimeout = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); + if (_parent.ConnectTimeout > 0) + { + responseReadTimeout.CancelAfter(_parent.ConnectTimeout); + } + + Stream responseStream; + try + { +#if NET5_0_OR_GREATER + responseStream = await response.Content.ReadAsStreamAsync(cancellationToken); +#else + responseStream = await response.Content.ReadAsStreamAsync(); +#endif + } + catch + { + response.Dispose(); + responseReadTimeout.Dispose(); + throw; + } + + // Only take ownership of the new response/timeout once stream acquisition has + // succeeded, so a failed FlushAsync leaves no dangling resources behind and a + // subsequent retry does not leak the previous response. + _response?.Dispose(); + _response = response; + _responseReadTimeout?.Dispose(); + _responseReadTimeout = responseReadTimeout; + _responseStream = responseStream; + _sent = true; + } + + public override async ValueTask<int> ReadAsync( + byte[] buffer, + int offset, + int length, + CancellationToken cancellationToken) + { + if (!_sent) + { + throw new TTransportException( TTransportException.ExceptionType.NotOpen, "The request has not been sent."); + } + + CheckReadBytesAvailable(length); + + CancellationTokenSource readCts = null; + if (_responseReadTimeout != null) + { + readCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, _responseReadTimeout.Token); + } + var readToken = readCts?.Token ?? cancellationToken; + try + { + var count = await _responseStream.ReadAsync( + buffer, + offset, + length, + readToken); + + CountConsumedMessageBytes(count); + return count; + } + catch (OperationCanceledException ocx) when (_responseReadTimeout != null) + { + throw new TTransportException(TTransportException.ExceptionType.Interrupted, ocx.Message, ocx); + } + finally + { + readCts?.Dispose(); + } Review Comment: The per-call path lets an `IOException` from the response stream escape unchanged, whereas `THttpTransport.ReadAsync` wraps the same failure in `TTransportException` (`THttpTransport.cs:192-195`). Clients using the new provider therefore get a different exception contract for the same HTTP transport failure; catch and wrap `IOException` here as the parent implementation does. ########## lib/netstd/Thrift/Transport/Client/THttpTransport.cs: ########## @@ -293,10 +348,83 @@ protected override void Dispose(bool disposing) { _inputStream?.Dispose(); _outputStream?.Dispose(); + _response?.Dispose(); + _response = null; + _responseReadTimeout?.Dispose(); + _responseReadTimeout = null; _httpClient?.Dispose(); } } _isDisposed = true; } + + public ValueTask<TTransport> CreatePerCallTransportAsync(CancellationToken cancellationToken = default) + { + return new ValueTask<TTransport>(new THttpPerCallTransport(this)); + } + + /// <summary> + /// Sends the given request stream to the server and returns the HTTP response. + /// </summary> + /// <param name="request">The request stream to send.</param> + /// <param name="cancellationToken">The cancellation token to cancel the operation.</param> + /// <returns>The HTTP response message.</returns> + /// <exception cref="TTransportException">Thrown if an error occurs during the request.</exception> + internal async Task<HttpResponseMessage> SendAsync(Stream request, CancellationToken cancellationToken) + { + // Snapshot the field locally: Close()/Dispose() running concurrently on the parent + // transport can null out or dispose _httpClient while a per-call transport is in flight. + var httpClient = _httpClient ?? throw new TTransportException(TTransportException.ExceptionType.NotOpen, + "The transport has been closed."); + var content = new StreamContent(request); + content.Headers.ContentType = ContentType ?? new MediaTypeHeaderValue("application/x-thrift"); + + var message = new HttpRequestMessage(HttpMethod.Post, _uri) + { + Content = content + }; + + // Note: message/content are intentionally not disposed here. Disposing StreamContent + // would also dispose the caller-owned "request" stream, which is owned and disposed by + // THttpPerCallTransport (or the parent's own _outputStream) instead. Review Comment: Every per-call request leaves its HttpRequestMessage and StreamContent undisposed. Because the content wraps the caller-owned stream, disposing them requires either a non-owning content wrapper or transferring stream ownership only after SendAsync completes, but leaving them for GC delays cleanup of the request/content graph on every call and defeats deterministic resource disposal promised by the per-call transport. Dispose the message/content on the success path and use an ownership-safe path for failures/retries. ########## lib/netstd/Thrift/Transport/Client/THttpTransport.cs: ########## @@ -237,27 +264,48 @@ private void ConfigureClient(HttpClient httpClient) public override async Task FlushAsync(CancellationToken cancellationToken) { + StreamContent contentStream = null; + HttpRequestMessage request = null; + HttpResponseMessage response = null; try { _outputStream.Seek(0, SeekOrigin.Begin); - using (var contentStream = new StreamContent(_outputStream)) - { - contentStream.Headers.ContentType = ContentType ?? new MediaTypeHeaderValue(@"application/x-thrift"); + contentStream = new StreamContent(_outputStream); + contentStream.Headers.ContentType = ContentType ?? new MediaTypeHeaderValue(@"application/x-thrift"); - var response = (await _httpClient.PostAsync(_uri, contentStream, cancellationToken)).EnsureSuccessStatusCode(); + request = new HttpRequestMessage(HttpMethod.Post, _uri) + { + Content = contentStream + }; + + response = await _httpClient.SendAsync( + request, + HttpCompletionOption.ResponseHeadersRead, + cancellationToken); + response.EnsureSuccessStatusCode(); + + _responseReadTimeout?.Dispose(); + _responseReadTimeout = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); + if (_connectTimeout > 0) + { + _responseReadTimeout.CancelAfter(_connectTimeout); + } Review Comment: The response-read timeout source is assigned to the transport before `ReadAsStreamAsync` succeeds. When stream initialization throws, the `finally` block disposes the response but leaves this linked `CancellationTokenSource` (and its timer/registration) stored and live until a later flush or transport disposal. Create it locally and install it only after stream acquisition succeeds, or dispose and clear it on this failure path. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
