diff --git a/compiler/cpp/src/thrift/generate/t_netstd_generator.cc b/compiler/cpp/src/thrift/generate/t_netstd_generator.cc index 0e1f25caced..373623cf731 100644 --- a/compiler/cpp/src/thrift/generate/t_netstd_generator.cc +++ b/compiler/cpp/src/thrift/generate/t_netstd_generator.cc @@ -2439,6 +2439,20 @@ void t_netstd_generator::generate_service_client(ostream& out, t_service* tservi << indent() << "public Client(TProtocol inputProtocol, TProtocol outputProtocol) : base(inputProtocol, outputProtocol)" << '\n' << indent() << "{" << '\n' << indent() << "}" << '\n' + << '\n' + << indent() << "public Client(TTransport transport, TProtocolFactory protocolFactory) : this(transport, protocolFactory, protocolFactory)" << '\n' + << indent() << "{" << '\n' + << indent() << "}" << '\n' + << '\n' + << indent() << "public Client(TTransport transport, TProtocolFactory inputProtocolFactory, TProtocolFactory outputProtocolFactory)" << '\n' + << indent() << " : base(transport, inputProtocolFactory, outputProtocolFactory)" << '\n' + << indent() << "{" << '\n' + << indent() << "}" << '\n' + << '\n' + << indent() << "public Client(TProtocol inputProtocol, TProtocol outputProtocol, TTransport transport, TProtocolFactory inputProtocolFactory, TProtocolFactory outputProtocolFactory)" << '\n' + << indent() << " : base(inputProtocol, outputProtocol, transport, inputProtocolFactory, outputProtocolFactory)" << '\n' + << indent() << "{" << '\n' + << indent() << "}" << '\n' << '\n'; vector functions = tservice->get_functions(); @@ -2454,6 +2468,13 @@ void t_netstd_generator::generate_service_client(ostream& out, t_service* tservi out << indent() << "public async " << function_signature_async(*functions_iterator, "") << '\n' << indent() << "{" << '\n'; indent_up(); + out << indent() + << (((!(*functions_iterator)->is_oneway()) && !(*functions_iterator)->get_returntype()->is_void()) + ? "return await " + : "await ") + << "ExecutePerCallAsync(async () =>" << '\n' + << indent() << "{" << '\n'; + indent_up(); out << indent() << "await send_" << function_name << "("; string call_args = argument_list((*functions_iterator)->get_arglist(),false); if(! call_args.empty()) { @@ -2465,6 +2486,8 @@ void t_netstd_generator::generate_service_client(ostream& out, t_service* tservi << "await recv_" << function_name << "(" << CANCELLATION_TOKEN_NAME << ");" << '\n'; } indent_down(); + out << indent() << "}, " << CANCELLATION_TOKEN_NAME << ");" << '\n'; + indent_down(); out << indent() << "}" << '\n' << '\n'; // async send diff --git a/lib/netstd/Tests/Thrift.Tests/Transports/THttpTransportTests.cs b/lib/netstd/Tests/Thrift.Tests/Transports/THttpTransportTests.cs index 2a2d884203a..71b714c36b9 100644 --- a/lib/netstd/Tests/Thrift.Tests/Transports/THttpTransportTests.cs +++ b/lib/netstd/Tests/Thrift.Tests/Transports/THttpTransportTests.cs @@ -15,8 +15,19 @@ // specific language governing permissions and limitations // under the License. +using System; +using System.Collections.Concurrent; +using System.Collections.Generic; +using System.IO; +using System.Linq; +using System.Net; using System.Net.Http; +using System.Threading; +using System.Threading.Tasks; using Microsoft.VisualStudio.TestTools.UnitTesting; +using Thrift.Protocol; +using Thrift.Protocol.Entities; +using Thrift.Transport; using Thrift.Transport.Client; namespace Thrift.Tests.Transports @@ -36,5 +47,752 @@ public void THttpTransport_Uses_Configured_ConnectionTimeout_Test() Assert.IsTrue(client.Timeout.TotalMilliseconds == 5000); Assert.IsTrue(httpClientTransport.ConnectTimeout == 5000); } + + [TestMethod] + public void ClosedTHttpTransportReportsNotOpen() + { + using var transport = CreateTransport(new TrackingHandler(_ => CreateResponse("ok"))); + + transport.Close(); + + Assert.IsFalse(transport.IsOpen); + } + + [TestMethod] + public async Task ClosedTHttpTransportRejectsWrites() + { + using var transport = CreateTransport(new TrackingHandler(_ => CreateResponse("ok"))); + + transport.Close(); + + await Assert.ThrowsAsync( + () => transport.WriteAsync(new byte[] { 1 }, CancellationToken.None)); + } + + [TestMethod] + public async Task ClosedTransportsRejectOpenAsync() + { + using var parent = CreateTransport(new TrackingHandler(_ => CreateResponse("ok"))); + var call = await parent.CreatePerCallTransportAsync(); + call.Dispose(); + parent.Close(); + + await Assert.ThrowsAsync( + () => parent.OpenAsync(CancellationToken.None)); + await Assert.ThrowsAsync( + () => call.OpenAsync(CancellationToken.None)); + } + + [TestMethod] + public async Task ConnectTimeout_StopsBlockedResponseBodyRead() + { + using var parent = CreateTransport(new BlockingResponseHandler()); + parent.ConnectTimeout = 50; + await parent.WriteAsync(new[] { (byte)1 }, CancellationToken.None); + await parent.FlushAsync(CancellationToken.None); + + var buffer = new byte[1]; + var ex = await Assert.ThrowsAsync(() => parent.ReadAsync(buffer, 0, 1, CancellationToken.None).AsTask()); + Assert.AreEqual(TTransportException.ExceptionType.Interrupted, ex.Type); + } + + [TestMethod] + public async Task DisposedHttpClientDuringFlushIsReportedAsNotOpen() + { + using var parent = CreateTransport(new DisposedHttpClientHandler()); + await parent.WriteAsync(new byte[] { 1 }, CancellationToken.None); + + var ex = await Assert.ThrowsAsync( + () => parent.FlushAsync(CancellationToken.None)); + + Assert.AreEqual(TTransportException.ExceptionType.NotOpen, ex.Type); + } + + [TestMethod] + public async Task PerCallTransport_ConnectTimeout_StopsBlockedResponseBodyRead() + { + using var parent = CreateTransport(new BlockingResponseHandler()); + parent.ConnectTimeout = 50; + var call = await parent.CreatePerCallTransportAsync(); + await call.WriteAsync(new[] { (byte)1 }, CancellationToken.None); + await call.FlushAsync(CancellationToken.None); + + var buffer = new byte[1]; + var ex = await Assert.ThrowsAsync(() => call.ReadAsync(buffer, 0, 1, CancellationToken.None).AsTask()); + Assert.AreEqual(TTransportException.ExceptionType.Interrupted, ex.Type); + call.Dispose(); + } + + [TestMethod] + public async Task CallerCancellationDuringResponseReadIsNotTranslatedToTransportTimeout() + { + using (var parent = CreateTransport(new BlockingResponseHandler())) + { + await parent.WriteAsync(new byte[] { 1 }, CancellationToken.None); + await parent.FlushAsync(CancellationToken.None); + + using var cancellation = new CancellationTokenSource(); + var readTask = parent.ReadAsync(new byte[1], 0, 1, cancellation.Token).AsTask(); + cancellation.Cancel(); + await Assert.ThrowsAsync(() => readTask); + } + + using (var parent = CreateTransport(new BlockingResponseHandler())) + { + var call = await parent.CreatePerCallTransportAsync(); + await call.WriteAsync(new byte[] { 1 }, CancellationToken.None); + await call.FlushAsync(CancellationToken.None); + + using var cancellation = new CancellationTokenSource(); + var readTask = call.ReadAsync(new byte[1], 0, 1, cancellation.Token).AsTask(); + cancellation.Cancel(); + await Assert.ThrowsAsync(() => readTask); + call.Dispose(); + } + } + + [TestMethod] + public async Task ClosingTransportCancelsActiveResponseReads() + { + using (var parent = CreateTransport(new BlockingResponseHandler())) + { + await parent.WriteAsync(new byte[] { 1 }, CancellationToken.None); + await parent.FlushAsync(CancellationToken.None); + var readTask = parent.ReadAsync(new byte[1], 0, 1, CancellationToken.None).AsTask(); + + parent.Close(); + + var ex = await Assert.ThrowsAsync(() => readTask); + Assert.AreEqual(TTransportException.ExceptionType.Interrupted, ex.Type); + } + + using (var parent = CreateTransport(new BlockingResponseHandler())) + { + var call = await parent.CreatePerCallTransportAsync(); + await call.WriteAsync(new byte[] { 1 }, CancellationToken.None); + await call.FlushAsync(CancellationToken.None); + var readTask = call.ReadAsync(new byte[1], 0, 1, CancellationToken.None).AsTask(); + + call.Dispose(); + + var ex = await Assert.ThrowsAsync(() => readTask); + Assert.AreEqual(TTransportException.ExceptionType.Interrupted, ex.Type); + } + } + + [TestMethod] + public async Task CreatePerCallTransportAsync_CreatesDistinctTransports() + { + using var parent = CreateTransport(new TrackingHandler(_ => CreateResponse("ok"))); + var first = await parent.CreatePerCallTransportAsync(); + var second = await parent.CreatePerCallTransportAsync(); + + Assert.AreNotSame(first, second); + first.Dispose(); + second.Dispose(); + } + + [TestMethod] + public async Task PerCallTransport_SendsReadsAndDisposesResponseAndRequestResources() + { + var responseContent = new TrackingContent("response"); + Stream? requestStream = null; + var handler = new TrackingHandler(req => + { + // Capture the request stream while the HttpRequestMessage/StreamContent are + // still alive, i.e. during the send itself. + requestStream = req.Content!.ReadAsStreamAsync().GetAwaiter().GetResult(); + return new HttpResponseMessage(HttpStatusCode.OK) { Content = responseContent }; + }); + using var parent = CreateTransport(handler); + var call = await parent.CreatePerCallTransportAsync(); + + await call.WriteAsync(new byte[] { 1, 2, 3 }, CancellationToken.None); + await call.FlushAsync(CancellationToken.None); + + // The HttpRequestMessage/StreamContent used to send the request are disposed + // deterministically as soon as the request has been sent, rather than being left + // for the GC. + await Assert.ThrowsAsync(() => handler.Requests[0].Content!.ReadAsStreamAsync()); + + // The underlying request stream is owned by the per-call transport (not the + // now-disposed HttpRequestMessage/StreamContent) and remains usable until the call + // itself is disposed. + Assert.AreEqual(1, requestStream!.ReadByte()); + + var buffer = new byte[8]; + var count = await call.ReadAsync(buffer, 0, buffer.Length, CancellationToken.None); + Assert.AreEqual(8, count); + Assert.AreEqual("response", System.Text.Encoding.UTF8.GetString(buffer)); + + call.Dispose(); + + Assert.IsTrue(responseContent.IsDisposed); + Assert.ThrowsExactly(() => requestStream!.ReadByte()); + } + + [TestMethod] + public async Task PerCallTransport_FailedResponse_DisposesResponse() + { + var responseContent = new TrackingContent("failure"); + using var parent = CreateTransport(new TrackingHandler(_ => new HttpResponseMessage(HttpStatusCode.InternalServerError) { Content = responseContent })); + var call = await parent.CreatePerCallTransportAsync(); + await call.WriteAsync(new byte[] { 1 }, CancellationToken.None); + + await Assert.ThrowsAsync(() => call.FlushAsync(CancellationToken.None)); + Assert.IsTrue(responseContent.IsDisposed); + call.Dispose(); + } + + [TestMethod] + public async Task PerCallTransport_RewindsSeekableResponseStream() + { + var responseStream = new MemoryStream(System.Text.Encoding.UTF8.GetBytes("response"), writable: false); + responseStream.Position = 2; + var responseContent = new TrackingContent(responseStream); + using var parent = CreateTransport(new TrackingHandler(_ => + new HttpResponseMessage(HttpStatusCode.OK) { Content = responseContent })); + var call = await parent.CreatePerCallTransportAsync(); + await call.WriteAsync(new byte[] { 1 }, CancellationToken.None); + await call.FlushAsync(CancellationToken.None); + + var buffer = new byte[8]; + var count = await call.ReadAsync(buffer, 0, buffer.Length, CancellationToken.None); + + Assert.AreEqual(8, count); + Assert.AreEqual("response", System.Text.Encoding.UTF8.GetString(buffer)); + call.Dispose(); + } + + [TestMethod] + public async Task PerCallTransport_Cancellation_DoesNotLeakResponse() + { + var handler = new TrackingHandler(_ => CreateResponse("response")); + using var parent = CreateTransport(handler); + var call = await parent.CreatePerCallTransportAsync(); + await call.WriteAsync(new byte[] { 1 }, CancellationToken.None); + using var cancellation = new CancellationTokenSource(); + cancellation.Cancel(); + + await Assert.ThrowsAsync(() => call.FlushAsync(cancellation.Token)); + Assert.AreEqual(0, handler.Requests.Count); + call.Dispose(); + } + + [TestMethod] + public async Task DisposingPerCallTransport_DoesNotCloseParentHttpClient() + { + var handler = new TrackingHandler(_ => CreateResponse("ok")); + using var parent = CreateTransport(handler); + var first = await parent.CreatePerCallTransportAsync(); + await first.WriteAsync(new byte[] { 1 }, CancellationToken.None); + await first.FlushAsync(CancellationToken.None); + first.Dispose(); + + var second = await parent.CreatePerCallTransportAsync(); + await second.WriteAsync(new byte[] { 2 }, CancellationToken.None); + await second.FlushAsync(CancellationToken.None); + second.Dispose(); + + Assert.AreEqual(2, handler.Requests.Count); + Assert.IsTrue(parent.IsOpen); + } + + [TestMethod] + public async Task SharedTransport_RemainsUsableAcrossMultipleCalls() + { + var handler = new TrackingHandler(_ => CreateResponse("ok")); + using var parent = CreateTransport(handler); + await parent.WriteAsync(new byte[] { 1 }, CancellationToken.None); + await parent.FlushAsync(CancellationToken.None); + await parent.WriteAsync(new byte[] { 2 }, CancellationToken.None); + await parent.FlushAsync(CancellationToken.None); + Assert.AreEqual(2, handler.Requests.Count); + Assert.IsTrue(parent.IsOpen); + } + + [TestMethod] + public async Task PerCallTransport_Disposal_DoesNotDisposeSharedTransport() + { + var handler = new TrackingHandler(_ => CreateResponse("ok")); + var parent = CreateTransport(handler); + var call = await parent.CreatePerCallTransportAsync(); + + call.Dispose(); + + Assert.IsFalse(handler.IsDisposed); + Assert.IsTrue(parent.IsOpen); + parent.Dispose(); + Assert.IsTrue(handler.IsDisposed); + } + + [TestMethod] + public async Task PerCallTransport_ReadFailure_IsDisposedByCallOwner() + { + var responseContent = new TrackingContent(new ThrowingReadStream()); + using var parent = CreateTransport(new TrackingHandler(_ => + new HttpResponseMessage(HttpStatusCode.OK) { Content = responseContent })); + var call = await parent.CreatePerCallTransportAsync(); + await call.WriteAsync(new byte[] { 1 }, CancellationToken.None); + await call.FlushAsync(CancellationToken.None); + var ex = await Assert.ThrowsAsync(() => call.ReadAsync(new byte[1], 0, 1, CancellationToken.None).AsTask()); + Assert.AreEqual(TTransportException.ExceptionType.Unknown, ex.Type); + Assert.IsInstanceOfType(ex.InnerException); + call.Dispose(); + Assert.IsFalse(call.IsOpen); + Assert.IsTrue(responseContent.IsDisposed); + Assert.IsTrue(parent.IsOpen); + } + + [TestMethod] + public async Task PerCallTransport_ReadFailureAfterFlush_DisposesOwnedResponse() + { + var responseContent = new TrackingContent(new ThrowingReadStream()); + using var parent = CreateTransport(new TrackingHandler(_ => + new HttpResponseMessage(HttpStatusCode.OK) { Content = responseContent })); + var call = await parent.CreatePerCallTransportAsync(); + await call.WriteAsync(new byte[] { 1 }, CancellationToken.None); + + await call.FlushAsync(CancellationToken.None); + var ex = await Assert.ThrowsAsync(() => call.ReadAsync(new byte[1], 0, 1, CancellationToken.None).AsTask()); + Assert.AreEqual(TTransportException.ExceptionType.Unknown, ex.Type); + Assert.IsInstanceOfType(ex.InnerException); + call.Dispose(); + + Assert.IsFalse(call.IsOpen); + Assert.IsTrue(responseContent.IsDisposed); + } + + [TestMethod] + public async Task PerCallTransport_StreamInitializationFailure_PreventsRetryOnSameCall() + { + var failingContent = new ThrowingStreamInitContent(); + var handler = new TrackingHandler(_ => + new HttpResponseMessage(HttpStatusCode.OK) { Content = failingContent }); + using var parent = CreateTransport(handler); + var call = await parent.CreatePerCallTransportAsync(); + await call.WriteAsync(new byte[] { 1 }, CancellationToken.None); + + var ex = await Assert.ThrowsAsync(() => call.FlushAsync(CancellationToken.None)); + Assert.AreEqual(TTransportException.ExceptionType.Unknown, ex.Type); + Assert.IsInstanceOfType(ex.InnerException); + Assert.IsTrue(failingContent.IsDisposed); + + await Assert.ThrowsAsync(() => call.FlushAsync(CancellationToken.None)); + Assert.AreEqual(1, handler.Requests.Count); + call.Dispose(); + } + + [TestMethod] + public async Task ConcurrentPerCallTransports_CompleteIndependently() + { + var handler = new IndependentResponseHandler(); + using var parent = CreateTransport(handler); + var first = await parent.CreatePerCallTransportAsync(); + var second = await parent.CreatePerCallTransportAsync(); + var firstTask = SendAndReadAsync(first, 1); + var secondTask = SendAndReadAsync(second, 2); + await handler.RequestsStarted.Task; + handler.Release(0); + Assert.AreEqual("response-0", await firstTask); + Assert.IsFalse(secondTask.IsCompleted); + handler.Release(1); + Assert.AreEqual("response-1", await secondTask); + first.Dispose(); + second.Dispose(); + } + + [TestMethod] + public async Task GeneratedClientUsesIndependentPerCallTransports() + { + var handler = new GeneratedClientHandler(); + using var transport = CreateTransport(handler); + var protocolFactory = new TrackingBinaryProtocolFactory(); + using var client = new ThriftTest.ThriftTest.Client(transport, protocolFactory); + + var firstCall = client.testString("first", CancellationToken.None); + var secondCall = client.testString("second", CancellationToken.None); + await handler.BothRequestsStarted.Task; + handler.ReleaseResponses(); + + var responses = await Task.WhenAll(firstCall, secondCall); + var requestBytes = handler.Requests.ToArray(); + Assert.AreEqual(2, requestBytes.Length); + Assert.IsFalse(requestBytes[0].SequenceEqual(requestBytes[1])); + Assert.AreEqual(6, protocolFactory.CreatedCount); + CollectionAssert.AreEquivalent(new[] { "response-0", "response-1" }, responses); + } + + [TestMethod] + public async Task ProtocolFactory_IsInvokedPerCallWithPerCallTransport() + { + using var parent = CreateTransport(new TrackingHandler(_ => CreateResponse("ok"))); + var factory = new TrackingProtocolFactory(); + var firstTransport = await parent.CreatePerCallTransportAsync(); + using var firstProtocol = factory.GetProtocol(firstTransport); + var secondTransport = await parent.CreatePerCallTransportAsync(); + using var secondProtocol = factory.GetProtocol(secondTransport); + Assert.AreEqual(2, factory.Transports.Count); + Assert.AreSame(firstTransport, factory.Transports[0]); + Assert.AreSame(secondTransport, factory.Transports[1]); + } + + [TestMethod] + public void BaseClient_DisposesInputProtocolWhenOutputFactoryFails() + { + var transport = new TrackingMemoryBufferTransport(); + + try + { + new TestBaseClient( + transport, + new TrackingBinaryProtocolFactory(), + new ThrowingProtocolFactory()); + Assert.Fail("The output protocol factory should have failed."); + } + catch (InvalidOperationException) + { + } + + Assert.IsTrue(transport.IsDisposed); + } + + [TestMethod] + public async Task ClosingParentDuringPerCallSendRejectsLateResponse() + { + var handler = new ReleaseResponseHandler(); + using var parent = CreateTransport(handler); + var call = await parent.CreatePerCallTransportAsync(); + await call.WriteAsync(new byte[] { 1 }, CancellationToken.None); + var flushTask = call.FlushAsync(CancellationToken.None); + await handler.RequestStarted.Task; + + parent.Close(); + handler.Release(); + + var ex = await Assert.ThrowsAsync(() => flushTask); + Assert.AreEqual(TTransportException.ExceptionType.NotOpen, ex.Type); + Assert.IsFalse(call.IsOpen); + } + + [TestMethod] + public async Task DisposingPerCallTransportCancelsInFlightSend() + { + var handler = new CancellationHandler(); + using var parent = CreateTransport(handler); + var call = await parent.CreatePerCallTransportAsync(); + await call.WriteAsync(new byte[] { 1 }, CancellationToken.None); + var flushTask = call.FlushAsync(CancellationToken.None); + await handler.RequestStarted.Task; + + call.Dispose(); + + var ex = await Assert.ThrowsAsync(() => flushTask); + Assert.AreEqual(TTransportException.ExceptionType.Interrupted, ex.Type); + await handler.CancellationObserved.Task; + } + + [TestMethod] + public async Task CancellationWhileRequestIsBlocked_ReleasesResponseAndParentRemainsUsable() + { + var handler = new CancellationHandler(); + using var parent = CreateTransport(handler); + var call = await parent.CreatePerCallTransportAsync(); + await call.WriteAsync(new byte[] { 1 }, CancellationToken.None); + using var cancellation = new CancellationTokenSource(); + var flushTask = call.FlushAsync(cancellation.Token); + await handler.RequestStarted.Task; + cancellation.Cancel(); + await Assert.ThrowsAsync(() => flushTask); + await handler.CancellationObserved.Task; + call.Dispose(); + Assert.IsNull(handler.BlockedResponse); + var followUp = await parent.CreatePerCallTransportAsync(); + await followUp.WriteAsync(new byte[] { 2 }, CancellationToken.None); + await followUp.FlushAsync(CancellationToken.None); + followUp.Dispose(); + Assert.IsTrue(parent.IsOpen); + } + + private static async Task SendAndReadAsync(TTransport transport, byte value) + { + await transport.WriteAsync(new[] { value }, CancellationToken.None); + await transport.FlushAsync(CancellationToken.None); + var buffer = new byte[10]; + var count = await transport.ReadAsync(buffer, 0, buffer.Length, CancellationToken.None); + return System.Text.Encoding.UTF8.GetString(buffer, 0, count); + } + + private static THttpTransport CreateTransport(HttpMessageHandler handler) + { + return new THttpTransport(new HttpClient(handler), null, new Uri("http://localhost/test")); + } + + private static HttpResponseMessage CreateResponse(string value) + { + return new HttpResponseMessage(HttpStatusCode.OK) { Content = new StringContent(value) }; + } + + private sealed class TrackingBinaryProtocolFactory : TProtocolFactory + { + private int _createdCount; + public int CreatedCount => _createdCount; + + public override TProtocol GetProtocol(TTransport trans) + { + Interlocked.Increment(ref _createdCount); + return new TBinaryProtocol(trans); + } + } + + private sealed class ThrowingProtocolFactory : TProtocolFactory + { + public override TProtocol GetProtocol(TTransport trans) + { + throw new InvalidOperationException("Synthetic protocol factory failure."); + } + } + + private sealed class TestBaseClient : TBaseClient + { + public TestBaseClient( + TTransport transport, + TProtocolFactory inputProtocolFactory, + TProtocolFactory outputProtocolFactory) + : base(transport, inputProtocolFactory, outputProtocolFactory) + { + } + } + + private sealed class TrackingMemoryBufferTransport : TMemoryBufferTransport + { + public TrackingMemoryBufferTransport() + : base(new TConfiguration()) + { + } + + public bool IsDisposed { get; private set; } + + protected override void Dispose(bool disposing) + { + IsDisposed = true; + base.Dispose(disposing); + } + } + + private sealed class GeneratedClientHandler : HttpMessageHandler + { + private readonly TaskCompletionSource _release = NewSignal(); + private int _requestCount; + public ConcurrentQueue Requests { get; } = new ConcurrentQueue(); + public TaskCompletionSource BothRequestsStarted { get; } = NewSignal(); + + protected override async Task SendAsync(HttpRequestMessage request, CancellationToken cancellationToken) + { + var requestBytes = await request.Content!.ReadAsByteArrayAsync(cancellationToken); + Requests.Enqueue(requestBytes); + var index = Interlocked.Increment(ref _requestCount) - 1; + if (index == 1) + { + BothRequestsStarted.TrySetResult(true); + } + await _release.Task.WaitAsync(cancellationToken); + return new HttpResponseMessage(HttpStatusCode.OK) + { + Content = new ByteArrayContent(CreateStringReply("response-" + index)) + }; + } + + public void ReleaseResponses() => _release.TrySetResult(true); + + private static byte[] CreateStringReply(string value) + { + var transport = new TMemoryBufferTransport(new TConfiguration()); + var protocol = new TBinaryProtocol(transport); + protocol.WriteMessageBeginAsync(new TMessage("testString", TMessageType.Reply, 1), CancellationToken.None).GetAwaiter().GetResult(); + protocol.WriteStructBeginAsync(new TStruct("testString_result"), CancellationToken.None).GetAwaiter().GetResult(); + protocol.WriteFieldBeginAsync(new TField("success", TType.String, 0), CancellationToken.None).GetAwaiter().GetResult(); + protocol.WriteStringAsync(value, CancellationToken.None).GetAwaiter().GetResult(); + protocol.WriteFieldEndAsync(CancellationToken.None).GetAwaiter().GetResult(); + protocol.WriteFieldStopAsync(CancellationToken.None).GetAwaiter().GetResult(); + protocol.WriteStructEndAsync(CancellationToken.None).GetAwaiter().GetResult(); + protocol.WriteMessageEndAsync(CancellationToken.None).GetAwaiter().GetResult(); + var bytes = transport.GetBuffer(); + protocol.Dispose(); + return bytes; + } + } + + private sealed class TrackingHandler : HttpMessageHandler + { + private readonly Func _responseFactory; + public TrackingHandler(Func responseFactory) => _responseFactory = responseFactory; + public List Requests { get; } = new List(); + public bool IsDisposed { get; private set; } + protected override Task SendAsync(HttpRequestMessage request, CancellationToken cancellationToken) + { + cancellationToken.ThrowIfCancellationRequested(); + Requests.Add(request); + return Task.FromResult(_responseFactory(request)); + } + protected override void Dispose(bool disposing) + { + IsDisposed = true; + base.Dispose(disposing); + } + } + + private class TrackingContent : HttpContent + { + private readonly byte[]? _bytes; + private readonly Stream? _stream; + public TrackingContent(string value) => _bytes = System.Text.Encoding.UTF8.GetBytes(value); + public TrackingContent(Stream stream) => _stream = stream; + public bool IsDisposed { get; private set; } + protected override Task SerializeToStreamAsync(Stream stream, TransportContext? context) => stream.WriteAsync(_bytes ?? Array.Empty(), 0, _bytes?.Length ?? 0); + protected override bool TryComputeLength(out long length) { length = _bytes?.Length ?? 0; return true; } + protected override Task CreateContentReadStreamAsync() => Task.FromResult(_stream ?? (Stream)new MemoryStream(_bytes ?? Array.Empty(), writable: false)); + protected override void Dispose(bool disposing) { IsDisposed = true; _stream?.Dispose(); base.Dispose(disposing); } + } + + private sealed class ThrowingReadStream : MemoryStream + { + public override Task ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken) => + Task.FromException(new IOException("Synthetic response read failure.")); + } + + private sealed class ThrowingStreamInitContent : HttpContent + { + public bool IsDisposed { get; private set; } + protected override Task SerializeToStreamAsync(Stream stream, TransportContext? context) => Task.CompletedTask; + protected override bool TryComputeLength(out long length) { length = 0; return true; } + protected override Task CreateContentReadStreamAsync() => + Task.FromException(new IOException("Synthetic stream initialization failure.")); + protected override void Dispose(bool disposing) { IsDisposed = true; base.Dispose(disposing); } + } + + private sealed class TrackingProtocolFactory : TProtocolFactory + { + public List Transports { get; } = new List(); + public override TProtocol GetProtocol(TTransport trans) + { + Transports.Add(trans); + return new TBinaryProtocol(trans); + } + } + + private sealed class IndependentResponseHandler : HttpMessageHandler + { + private readonly TaskCompletionSource[] _releases = { NewSignal(), NewSignal() }; + private int _requestCount; + public TaskCompletionSource RequestsStarted { get; } = NewSignal(); + protected override async Task SendAsync(HttpRequestMessage request, CancellationToken cancellationToken) + { + var index = Interlocked.Increment(ref _requestCount) - 1; + if (index == 1) RequestsStarted.TrySetResult(true); + await _releases[index].Task.WaitAsync(cancellationToken); + return CreateResponse("response-" + index); + } + public void Release(int index) => _releases[index].TrySetResult(true); + } + + private sealed class BlockingResponseHandler : HttpMessageHandler + { + protected override Task SendAsync(HttpRequestMessage request, CancellationToken cancellationToken) + { + return Task.FromResult(new HttpResponseMessage(HttpStatusCode.OK) + { + Content = new BlockingReadContent() + }); + } + } + + private sealed class DisposedHttpClientHandler : HttpMessageHandler + { + protected override Task SendAsync( + HttpRequestMessage request, + CancellationToken cancellationToken) + { + throw new ObjectDisposedException("HttpClient"); + } + } + + private sealed class BlockingReadContent : HttpContent + { + protected override Task SerializeToStreamAsync(Stream stream, TransportContext? context) => Task.CompletedTask; + protected override bool TryComputeLength(out long length) + { + length = 0; + return false; + } + protected override Task CreateContentReadStreamAsync() => Task.FromResult(new BlockingReadStream()); + } + + private sealed class BlockingReadStream : Stream + { + public override bool CanRead => true; + public override bool CanSeek => false; + public override bool CanWrite => false; + public override long Length => throw new NotSupportedException(); + public override long Position { get => throw new NotSupportedException(); set => throw new NotSupportedException(); } + public override void Flush() { } + public override int Read(byte[] buffer, int offset, int count) => throw new NotSupportedException(); + public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException(); + public override void SetLength(long value) => throw new NotSupportedException(); + public override void Write(byte[] buffer, int offset, int count) => throw new NotSupportedException(); + public override async Task ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken) + { + await Task.Delay(Timeout.Infinite, cancellationToken); + return 0; + } + public override ValueTask ReadAsync(Memory buffer, CancellationToken cancellationToken = default) + { + return new ValueTask(Task.Delay(Timeout.Infinite, cancellationToken).ContinueWith(_ => 0, cancellationToken, TaskContinuationOptions.None, TaskScheduler.Default)); + } + } + + private sealed class ReleaseResponseHandler : HttpMessageHandler + { + private readonly TaskCompletionSource _release = NewSignal(); + public TaskCompletionSource RequestStarted { get; } = NewSignal(); + + protected override async Task SendAsync(HttpRequestMessage request, CancellationToken cancellationToken) + { + RequestStarted.TrySetResult(true); + await _release.Task; + return CreateResponse("late"); + } + + public void Release() => _release.TrySetResult(true); + } + + private sealed class CancellationHandler : HttpMessageHandler + { + private readonly TaskCompletionSource _blocked = NewSignal(); + private int _requestCount; + public TaskCompletionSource RequestStarted { get; } = NewSignal(); + public TaskCompletionSource CancellationObserved { get; } = NewSignal(); + public HttpResponseMessage? BlockedResponse { get; private set; } + protected override async Task SendAsync(HttpRequestMessage request, CancellationToken cancellationToken) + { + if (Interlocked.Increment(ref _requestCount) > 1) + return CreateResponse("follow-up"); + RequestStarted.TrySetResult(true); + try + { + await _blocked.Task.WaitAsync(cancellationToken); + BlockedResponse = CreateResponse("unexpected"); + return BlockedResponse; + } + catch (OperationCanceledException) + { + CancellationObserved.TrySetResult(true); + throw; + } + } + } + + private static TaskCompletionSource NewSignal() => + new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); } } diff --git a/lib/netstd/Thrift/TBaseClient.cs b/lib/netstd/Thrift/TBaseClient.cs index aefb54dd5a1..539d5429ce0 100644 --- a/lib/netstd/Thrift/TBaseClient.cs +++ b/lib/netstd/Thrift/TBaseClient.cs @@ -19,6 +19,7 @@ using System.Threading; using System.Threading.Tasks; using Thrift.Protocol; +using Thrift.Transport; namespace Thrift { @@ -32,6 +33,10 @@ public abstract class TBaseClient { private readonly TProtocol _inputProtocol; private readonly TProtocol _outputProtocol; + private readonly ITPerCallTransportProvider _perCallTransportProvider; + private readonly TProtocolFactory _inputProtocolFactory; + private readonly TProtocolFactory _outputProtocolFactory; + private readonly AsyncLocal _perCallProtocols = new AsyncLocal(); private bool _isDisposed; private int _seqId; public readonly Guid ClientId = Guid.NewGuid(); @@ -42,13 +47,182 @@ protected TBaseClient(TProtocol inputProtocol, TProtocol outputProtocol) _outputProtocol = outputProtocol ?? throw new ArgumentNullException(nameof(outputProtocol)); } - public TProtocol InputProtocol => _inputProtocol; + protected TBaseClient( + TProtocol inputProtocol, + TProtocol outputProtocol, + TTransport transport, + TProtocolFactory inputProtocolFactory, + TProtocolFactory outputProtocolFactory) + : this(inputProtocol, outputProtocol) + { + _perCallTransportProvider = transport as ITPerCallTransportProvider; + if (_perCallTransportProvider != null) + { + _inputProtocolFactory = inputProtocolFactory ?? throw new ArgumentNullException(nameof(inputProtocolFactory)); + _outputProtocolFactory = outputProtocolFactory ?? throw new ArgumentNullException(nameof(outputProtocolFactory)); + } + } + + protected TBaseClient( + TTransport transport, + TProtocolFactory inputProtocolFactory, + TProtocolFactory outputProtocolFactory) + { + if (transport == null) throw new ArgumentNullException(nameof(transport)); + if (inputProtocolFactory == null) throw new ArgumentNullException(nameof(inputProtocolFactory)); + if (outputProtocolFactory == null) throw new ArgumentNullException(nameof(outputProtocolFactory)); + + var protocols = CreateProtocols(transport, inputProtocolFactory, outputProtocolFactory); + _inputProtocol = protocols.InputProtocol; + _outputProtocol = protocols.OutputProtocol; + _perCallTransportProvider = transport as ITPerCallTransportProvider; + if (_perCallTransportProvider != null) + { + _inputProtocolFactory = inputProtocolFactory; + _outputProtocolFactory = outputProtocolFactory; + } + } + + private static ProtocolPair CreateProtocols( + TTransport transport, + TProtocolFactory inputProtocolFactory, + TProtocolFactory outputProtocolFactory) + { + TProtocol inputProtocol = null; + TProtocol outputProtocol = null; + try + { + inputProtocol = inputProtocolFactory.GetProtocol(transport) + ?? throw new InvalidOperationException("The input protocol factory returned null."); + outputProtocol = outputProtocolFactory.GetProtocol(transport) + ?? throw new InvalidOperationException("The output protocol factory returned null."); + return new ProtocolPair(inputProtocol, outputProtocol); + } + catch + { + outputProtocol?.Dispose(); + if (inputProtocol != null && !ReferenceEquals(inputProtocol, outputProtocol)) + { + inputProtocol.Dispose(); + } + if (inputProtocol == null && outputProtocol == null) + { + transport.Dispose(); + } + throw; + } + } - public TProtocol OutputProtocol => _outputProtocol; + private sealed class ProtocolPair + { + public ProtocolPair(TProtocol inputProtocol, TProtocol outputProtocol) + { + InputProtocol = inputProtocol; + OutputProtocol = outputProtocol; + } + + public TProtocol InputProtocol { get; } + public TProtocol OutputProtocol { get; } + } + + public TProtocol InputProtocol => _perCallProtocols.Value?.InputProtocol ?? _inputProtocol; + + public TProtocol OutputProtocol => _perCallProtocols.Value?.OutputProtocol ?? _outputProtocol; public int SeqId { - get { return ++_seqId; } + get { return Interlocked.Increment(ref _seqId); } + } + + protected async Task ExecutePerCallAsync(Func operation, CancellationToken cancellationToken) + { + await ExecutePerCallCoreAsync(async () => + { + await operation(); + return true; + }, cancellationToken); + } + + protected Task ExecutePerCallAsync( + Func> operation, + CancellationToken cancellationToken) + { + return ExecutePerCallCoreAsync(operation, cancellationToken); + } + + private async Task ExecutePerCallCoreAsync( + Func> operation, + CancellationToken cancellationToken) + { + if (_perCallTransportProvider == null) + return await operation(); + + var transport = await _perCallTransportProvider.CreatePerCallTransportAsync(cancellationToken); + TProtocol inputProtocol = null; + TProtocol outputProtocol = null; + try + { + inputProtocol = _inputProtocolFactory.GetProtocol(transport) + ?? throw new InvalidOperationException("The input protocol factory returned null."); + outputProtocol = _outputProtocolFactory.GetProtocol(transport) + ?? throw new InvalidOperationException("The output protocol factory returned null."); + } + catch + { + if (outputProtocol != null) + { + outputProtocol.Dispose(); + } + if (inputProtocol != null && !ReferenceEquals(inputProtocol, outputProtocol)) + { + inputProtocol.Dispose(); + } + if (inputProtocol == null && outputProtocol == null) + { + transport.Dispose(); + } + throw; + } + + var protocols = new PerCallProtocols(inputProtocol, outputProtocol); + var previousProtocols = _perCallProtocols.Value; + _perCallProtocols.Value = protocols; + try + { + return await operation(); + } + finally + { + _perCallProtocols.Value = previousProtocols; + protocols.Dispose(); + } + } + + private sealed class PerCallProtocols : IDisposable + { + public PerCallProtocols(TProtocol inputProtocol, TProtocol outputProtocol) + { + InputProtocol = inputProtocol; + OutputProtocol = outputProtocol; + } + + public TProtocol InputProtocol { get; } + public TProtocol OutputProtocol { get; } + + public void Dispose() + { + try + { + OutputProtocol.Dispose(); + } + finally + { + if (!ReferenceEquals(InputProtocol, OutputProtocol)) + { + InputProtocol.Dispose(); + } + } + } } public virtual async Task OpenTransportAsync() diff --git a/lib/netstd/Thrift/Transport/Client/THttpPerCallTransport.cs b/lib/netstd/Thrift/Transport/Client/THttpPerCallTransport.cs new file mode 100644 index 00000000000..4630fdcf35f --- /dev/null +++ b/lib/netstd/Thrift/Transport/Client/THttpPerCallTransport.cs @@ -0,0 +1,306 @@ +// 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.Diagnostics; +using System.IO; +using System.Net.Http; +using System.Threading; +using System.Threading.Tasks; + +namespace Thrift.Transport.Client +{ + /// + /// Implementation of a per-call transport for THttpTransport. + /// Each call to the server will create a new HTTP request and response. + /// + internal class THttpPerCallTransport : TEndpointTransport + { + private readonly THttpTransport _parent; + private readonly MemoryStream _request = new MemoryStream(); + private readonly CancellationTokenSource _lifecycleCancellation = new CancellationTokenSource(); + + 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 + { + get + { + lock (_parent.LifecycleLock) + { + return !_disposed && _parent.IsLifecycleOpen; + } + } + } + + public override Task WriteAsync( + byte[] buffer, + int offset, + int length, + CancellationToken cancellationToken) + { + lock (_parent.LifecycleLock) + { + if (_disposed || !_parent.IsLifecycleOpen) + { + throw new TTransportException(TTransportException.ExceptionType.NotOpen, + "The transport has been closed."); + } + 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) + { + int connectTimeout; + var responseDeadline = Stopwatch.StartNew(); + CancellationTokenSource operationCts; + lock (_parent.LifecycleLock) + { + if (_disposed || !_parent.IsLifecycleOpen) + { + throw new TTransportException(TTransportException.ExceptionType.NotOpen, + "The transport has been closed."); + } + if (_sent) + { + throw new InvalidOperationException("The request has already been sent."); + } + + // Once a send is attempted the server may have processed the RPC, even if + // response acquisition fails. Require a fresh per-call transport to retry. + _sent = true; + _request.Position = 0; + connectTimeout = _parent.ConnectTimeout; + operationCts = CancellationTokenSource.CreateLinkedTokenSource( + cancellationToken, + _lifecycleCancellation.Token); + } + + HttpResponseMessage response = null; + CancellationTokenSource responseReadTimeout = null; + Stream responseStream = null; + try + { + response = await _parent.SendAsync(_request, operationCts.Token); + lock (_parent.LifecycleLock) + { + if (_disposed || !_parent.IsLifecycleOpen) + { + throw new TTransportException(TTransportException.ExceptionType.NotOpen, + "The transport has been closed."); + } + + responseReadTimeout = new CancellationTokenSource(); + if (connectTimeout > 0) + { + var remainingTimeout = connectTimeout - (int)responseDeadline.ElapsedMilliseconds; + responseReadTimeout.CancelAfter(Math.Max(0, remainingTimeout)); + } + } + using (var acquisitionCts = CancellationTokenSource.CreateLinkedTokenSource( + operationCts.Token, responseReadTimeout.Token)) + { + responseStream = await THttpTransport.ReadAsStreamAsync( + response.Content, acquisitionCts.Token); + } + if (responseStream.CanSeek) + { + responseStream.Seek(0, SeekOrigin.Begin); + } + + // Only take ownership of the new response/timeout/stream once stream acquisition + // and normalization have succeeded, so a failed FlushAsync leaves no dangling + // resources behind and a subsequent retry does not leak the previous response. + lock (_parent.LifecycleLock) + { + if (_disposed || !_parent.IsLifecycleOpen) + { + throw new TTransportException(TTransportException.ExceptionType.NotOpen, + "The transport has been closed."); + } + + _response?.Dispose(); + _response = response; + response = null; + _responseReadTimeout?.Dispose(); + _responseReadTimeout = responseReadTimeout; + responseReadTimeout = null; + _responseStream = responseStream; + responseStream = null; + } + } + catch (TTransportException) + { + throw; + } + catch (IOException iox) + { + throw new TTransportException(TTransportException.ExceptionType.Unknown, iox.ToString(), iox); + } + catch (HttpRequestException wx) + { + throw new TTransportException(TTransportException.ExceptionType.Unknown, + "Couldn't connect to server: " + wx, wx); + } + catch (OperationCanceledException ocx) + { + throw new TTransportException(TTransportException.ExceptionType.Interrupted, ocx.Message, ocx); + } + catch (Exception ex) + { + throw new TTransportException(TTransportException.ExceptionType.Unknown, ex.Message, ex); + } + finally + { + responseStream?.Dispose(); + response?.Dispose(); + responseReadTimeout?.Dispose(); + operationCts.Dispose(); + } + } + + public override async ValueTask 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; + Stream responseStream; + bool hasResponseReadTimeout; + lock (_parent.LifecycleLock) + { + if (_disposed || !_parent.IsLifecycleOpen || _responseStream == null) + { + throw new TTransportException(TTransportException.ExceptionType.NotOpen, + "The transport has been closed."); + } + + responseStream = _responseStream; + hasResponseReadTimeout = _responseReadTimeout != null; + readCts = CancellationTokenSource.CreateLinkedTokenSource( + cancellationToken, + _lifecycleCancellation.Token, + _responseReadTimeout?.Token ?? CancellationToken.None); + } + + var readToken = readCts?.Token ?? cancellationToken; + try + { + var count = await responseStream.ReadAsync( + buffer, + offset, + length, + readToken); + + CountConsumedMessageBytes(count); + return count; + } + catch (OperationCanceledException ocx) when (hasResponseReadTimeout && !cancellationToken.IsCancellationRequested) + { + throw new TTransportException(TTransportException.ExceptionType.Interrupted, ocx.Message, ocx); + } + catch (ObjectDisposedException odx) + { + throw new TTransportException(TTransportException.ExceptionType.NotOpen, + "The transport has been closed.", odx); + } + catch (IOException iox) + { + throw new TTransportException(TTransportException.ExceptionType.Unknown, iox.ToString(), iox); + } + finally + { + readCts?.Dispose(); + } + } + + protected override void Dispose(bool disposing) + { + bool shouldCancel = false; + lock (_parent.LifecycleLock) + { + if (_disposed) + { + return; + } + + if (disposing) + { + shouldCancel = true; + _response?.Dispose(); + _response = null; + _responseReadTimeout?.Dispose(); + _responseReadTimeout = null; + _request.Dispose(); + } + + _disposed = true; + _parent.RemovePerCallTransport(this); + } + + if (shouldCancel) + { + _lifecycleCancellation.Cancel(); + _lifecycleCancellation.Dispose(); + } + } + + public override Task OpenAsync(CancellationToken cancellationToken = default) + { + cancellationToken.ThrowIfCancellationRequested(); + lock (_parent.LifecycleLock) + { + if (_disposed || !_parent.IsLifecycleOpen) + { + throw new TTransportException(TTransportException.ExceptionType.NotOpen, + "The transport has been closed."); + } + } + return Task.CompletedTask; + } + + public override void Close() + { + // Do not dispose the parent transport, as it is shared across multiple per-call transports. + this.Dispose(); + } + } +} diff --git a/lib/netstd/Thrift/Transport/Client/THttpTransport.cs b/lib/netstd/Thrift/Transport/Client/THttpTransport.cs index 439777eb390..29ac8c0098e 100644 --- a/lib/netstd/Thrift/Transport/Client/THttpTransport.cs +++ b/lib/netstd/Thrift/Transport/Client/THttpTransport.cs @@ -17,6 +17,7 @@ using System; using System.Collections.Generic; +using System.Diagnostics; using System.IO; using System.Linq; using System.Net.Http; @@ -32,15 +33,22 @@ namespace Thrift.Transport.Client { // ReSharper disable once InconsistentNaming - public class THttpTransport : TEndpointTransport + public class THttpTransport : TEndpointTransport, ITPerCallTransportProvider { private readonly X509Certificate[] _certificates; private readonly Uri _uri; + private readonly object _lifecycleLock = new object(); + private readonly HashSet _perCallTransports = new HashSet(); + private CancellationTokenSource _lifecycleCancellation = new CancellationTokenSource(); + private bool _lifecycleOpen = true; + private int _flushInProgress; private int _connectTimeout = 30000; // Timeouts in milliseconds private HttpClient _httpClient; private Stream _inputStream; private MemoryStream _outputStream = new MemoryStream(); + private HttpResponseMessage _response; + private CancellationTokenSource _responseReadTimeout; private bool _isDisposed; public THttpTransport(Uri uri, TConfiguration config, IDictionary customRequestHeaders = null, string userAgent = null) @@ -109,7 +117,16 @@ public int ConnectTimeout } } - public override bool IsOpen => true; + public override bool IsOpen + { + get + { + lock (_lifecycleLock) + { + return _lifecycleOpen && _httpClient != null; + } + } + } public HttpRequestHeaders RequestHeaders => _httpClient.DefaultRequestHeaders; @@ -118,27 +135,50 @@ public int ConnectTimeout public override Task OpenAsync(CancellationToken cancellationToken) { cancellationToken.ThrowIfCancellationRequested(); + lock (_lifecycleLock) + { + if (!_lifecycleOpen || _httpClient == null) + { + throw new TTransportException(TTransportException.ExceptionType.NotOpen, + "The transport has been closed."); + } + } return Task.CompletedTask; } public override void Close() { - if (_inputStream != null) + List transports; + CancellationTokenSource lifecycleCancellation; + lock (_lifecycleLock) { - _inputStream.Dispose(); - _inputStream = null; - } + _lifecycleOpen = false; + transports = new List(_perCallTransports); + _perCallTransports.Clear(); + lifecycleCancellation = _lifecycleCancellation; + _lifecycleCancellation = null; - if (_outputStream != null) - { - _outputStream.Dispose(); + _inputStream?.Dispose(); + _inputStream = null; + var outputStream = _outputStream; _outputStream = null; + if (_flushInProgress == 0) + { + outputStream?.Dispose(); + } + _response?.Dispose(); + _response = null; + _responseReadTimeout?.Dispose(); + _responseReadTimeout = null; + _httpClient?.Dispose(); + _httpClient = null; } - if (_httpClient != null) + lifecycleCancellation?.Cancel(); + lifecycleCancellation?.Dispose(); + foreach (var transport in transports) { - _httpClient.Dispose(); - _httpClient = null; + transport.Dispose(); } } @@ -151,12 +191,33 @@ public override async ValueTask ReadAsync(byte[] buffer, int offset, int le CheckReadBytesAvailable(length); + CancellationTokenSource readCts = null; + Stream inputStream; + bool hasResponseReadTimeout; + lock (_lifecycleLock) + { + if (!_lifecycleOpen || _inputStream == null) + { + throw new TTransportException(TTransportException.ExceptionType.NotOpen, + "The transport has been closed."); + } + + inputStream = _inputStream; + hasResponseReadTimeout = _responseReadTimeout != null; + readCts = CancellationTokenSource.CreateLinkedTokenSource( + cancellationToken, + _lifecycleCancellation.Token, + _responseReadTimeout?.Token ?? CancellationToken.None); + } + try { + var readToken = readCts?.Token ?? cancellationToken; + int ret; #if NET5_0_OR_GREATER - var ret = await _inputStream.ReadAsync(new Memory(buffer, offset, length), cancellationToken); + ret = await inputStream.ReadAsync(new Memory(buffer, offset, length), readToken); #else - var ret = await _inputStream.ReadAsync(buffer, offset, length, cancellationToken); + ret = await inputStream.ReadAsync(buffer, offset, length, readToken); #endif if (ret == -1) { @@ -166,20 +227,45 @@ public override async ValueTask ReadAsync(byte[] buffer, int offset, int le CountConsumedMessageBytes(ret); return ret; } + catch (OperationCanceledException ocx) when (hasResponseReadTimeout && !cancellationToken.IsCancellationRequested) + { + throw new TTransportException(TTransportException.ExceptionType.Interrupted, ocx.Message, ocx); + } + catch (ObjectDisposedException odx) + { + throw new TTransportException(TTransportException.ExceptionType.NotOpen, + "The transport has been closed.", odx); + } catch (IOException iox) { throw new TTransportException(TTransportException.ExceptionType.Unknown, iox.ToString(), iox); } + finally + { + readCts?.Dispose(); + } } public override async Task WriteAsync(byte[] buffer, int offset, int length, CancellationToken cancellationToken) { cancellationToken.ThrowIfCancellationRequested(); + Stream outputStream; + lock (_lifecycleLock) + { + if (!_lifecycleOpen || _outputStream == null) + { + throw new TTransportException(TTransportException.ExceptionType.NotOpen, + "The transport has been closed."); + } + + outputStream = _outputStream; + } + #if NET5_0_OR_GREATER - await _outputStream.WriteAsync(buffer.AsMemory(offset, length), cancellationToken); + await outputStream.WriteAsync(buffer.AsMemory(offset, length), cancellationToken); #else - await _outputStream.WriteAsync(buffer, offset, length, cancellationToken); + await outputStream.WriteAsync(buffer, offset, length, cancellationToken); #endif } @@ -237,28 +323,92 @@ private void ConfigureClient(HttpClient httpClient) public override async Task FlushAsync(CancellationToken cancellationToken) { + HttpClient httpClient; + Stream outputStream; + int connectTimeout; + CancellationTokenSource operationCts; + lock (_lifecycleLock) + { + if (!_lifecycleOpen || _httpClient == null || _outputStream == null) + { + throw new TTransportException(TTransportException.ExceptionType.NotOpen, + "The transport has been closed."); + } + httpClient = _httpClient; + outputStream = _outputStream; + connectTimeout = _connectTimeout; + operationCts = CancellationTokenSource.CreateLinkedTokenSource( + cancellationToken, + _lifecycleCancellation.Token); + _flushInProgress++; + } + + StreamContent contentStream = null; + HttpRequestMessage request = null; + HttpResponseMessage response = null; + CancellationTokenSource responseReadTimeout = null; + Stream inputStream = null; + var responseDeadline = Stopwatch.StartNew(); try { - _outputStream.Seek(0, SeekOrigin.Begin); + 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, + operationCts.Token); + response.EnsureSuccessStatusCode(); + + responseReadTimeout = new CancellationTokenSource(); + var remainingTimeout = connectTimeout - (int)responseDeadline.ElapsedMilliseconds; + if (connectTimeout > 0) + { + responseReadTimeout.CancelAfter(Math.Max(0, remainingTimeout)); + } + + using (var acquisitionCts = CancellationTokenSource.CreateLinkedTokenSource( + operationCts.Token, responseReadTimeout.Token)) + { + inputStream = await ReadAsStreamAsync(response.Content, acquisitionCts.Token); + } + if (inputStream.CanSeek) + { + inputStream.Seek(0, SeekOrigin.Begin); + } - _inputStream?.Dispose(); -#if NET5_0_OR_GREATER - _inputStream = await response.Content.ReadAsStreamAsync(cancellationToken); -#else - _inputStream = await response.Content.ReadAsStreamAsync(); -#endif - if (_inputStream.CanSeek) + // Only take ownership of the new response/timeout/stream once stream acquisition + // has succeeded, so a failed FlushAsync leaves no dangling resources behind. + lock (_lifecycleLock) + { + if (!_lifecycleOpen || !ReferenceEquals(_httpClient, httpClient)) { - _inputStream.Seek(0, SeekOrigin.Begin); + throw new TTransportException(TTransportException.ExceptionType.NotOpen, + "The transport has been closed."); } + + _inputStream?.Dispose(); + _inputStream = inputStream; + inputStream = null; + _response?.Dispose(); + _response = response; + response = null; + _responseReadTimeout?.Dispose(); + _responseReadTimeout = responseReadTimeout; + responseReadTimeout = null; } } + catch (TTransportException) + { + throw; + } catch (IOException iox) { throw new TTransportException(TTransportException.ExceptionType.Unknown, iox.ToString(), iox); @@ -268,6 +418,11 @@ public override async Task FlushAsync(CancellationToken cancellationToken) throw new TTransportException(TTransportException.ExceptionType.Unknown, "Couldn't connect to server: " + wx, wx); } + catch (ObjectDisposedException odx) + { + throw new TTransportException(TTransportException.ExceptionType.NotOpen, + "The transport has been closed.", odx); + } catch (OperationCanceledException ocx) { throw new TTransportException(TTransportException.ExceptionType.Interrupted, ocx.Message, ocx); @@ -278,8 +433,30 @@ public override async Task FlushAsync(CancellationToken cancellationToken) } finally { - _outputStream = new MemoryStream(); - ResetMessageSizeAndConsumedBytes(); + inputStream?.Dispose(); + response?.Dispose(); + responseReadTimeout?.Dispose(); + operationCts?.Dispose(); + request?.Dispose(); + if (request == null) + { + contentStream?.Dispose(); + } + + lock (_lifecycleLock) + { + _flushInProgress--; + if (_lifecycleOpen) + { + _outputStream = new MemoryStream(); + } + else + { + outputStream.Dispose(); + _outputStream = null; + } + ResetMessageSizeAndConsumedBytes(); + } } } @@ -287,16 +464,217 @@ public override async Task FlushAsync(CancellationToken cancellationToken) // IDisposable protected override void Dispose(bool disposing) { - if (!_isDisposed) + if (_isDisposed) + { + return; + } + + if (disposing) + { + Close(); + } + _isDisposed = true; + } + + public ValueTask CreatePerCallTransportAsync(CancellationToken cancellationToken = default) + { + cancellationToken.ThrowIfCancellationRequested(); + lock (_lifecycleLock) { - if (disposing) + if (!_lifecycleOpen || _httpClient == null) { - _inputStream?.Dispose(); - _outputStream?.Dispose(); - _httpClient?.Dispose(); + throw new TTransportException(TTransportException.ExceptionType.NotOpen, + "The transport has been closed."); } + + var transport = new THttpPerCallTransport(this); + _perCallTransports.Add(transport); + return new ValueTask(transport); + } + } + + internal object LifecycleLock => _lifecycleLock; + + internal bool IsLifecycleOpen + { + get + { + lock (_lifecycleLock) + { + return _lifecycleOpen && _httpClient != null; + } + } + } + + internal void RemovePerCallTransport(THttpPerCallTransport transport) + { + lock (_lifecycleLock) + { + _perCallTransports.Remove(transport); + } + } + + internal static async Task ReadAsStreamAsync(HttpContent content, CancellationToken cancellationToken) + { +#if NET5_0_OR_GREATER + return await content.ReadAsStreamAsync(cancellationToken); +#else + var readTask = content.ReadAsStreamAsync(); + if (readTask.IsCompleted || !cancellationToken.CanBeCanceled) + { + return await readTask; + } + + var cancellationTask = new TaskCompletionSource(); + using (cancellationToken.Register(() => cancellationTask.TrySetResult(true))) + { + if (await Task.WhenAny(readTask, cancellationTask.Task) != readTask) + { + // The legacy API cannot cancel acquisition. If it later succeeds, dispose + // the stream because cancellation has already abandoned the acquisition. + _ = readTask.ContinueWith( + task => + { + if (task.Status == TaskStatus.RanToCompletion) + { + task.Result.Dispose(); + } + else if (task.IsFaulted) + { + var ignored = task.Exception; + } + }, + CancellationToken.None, + TaskContinuationOptions.ExecuteSynchronously, + TaskScheduler.Default); + throw new OperationCanceledException(cancellationToken); + } + } + + return await readTask; +#endif + } + + /// + /// Sends the given request stream to the server and returns the HTTP response. + /// + /// The request stream to send. + /// The cancellation token to cancel the operation. + /// The HTTP response message. + /// Thrown if an error occurs during the request. + internal async Task SendAsync(Stream request, CancellationToken cancellationToken) + { + // Snapshot the client under the lifecycle lock so Close()/Dispose() cannot race + // the state check and publish a response after the parent has closed. + HttpClient httpClient; + lock (_lifecycleLock) + { + if (!_lifecycleOpen || _httpClient == null) + { + throw new TTransportException(TTransportException.ExceptionType.NotOpen, + "The transport has been closed."); + } + httpClient = _httpClient; + } + + // Wrap the caller-owned request stream so that disposing the HttpRequestMessage/ + // StreamContent below (which we do deterministically once the request has been sent) + // does not dispose the stream itself; that remains owned and disposed by the caller. + var content = new StreamContent(new NonDisposingStream(request)); + content.Headers.ContentType = ContentType ?? new MediaTypeHeaderValue("application/x-thrift"); + + using (var message = new HttpRequestMessage(HttpMethod.Post, _uri) { Content = content }) + { + HttpResponseMessage response = null; + try + { + response = await httpClient.SendAsync( + message, + HttpCompletionOption.ResponseHeadersRead, + cancellationToken); + + response.EnsureSuccessStatusCode(); + lock (_lifecycleLock) + { + if (!_lifecycleOpen || !ReferenceEquals(_httpClient, httpClient)) + { + throw new TTransportException(TTransportException.ExceptionType.NotOpen, + "The transport has been closed."); + } + } + var result = response; + response = null; + return result; + } + catch (TTransportException) + { + throw; + } + catch (IOException iox) + { + throw new TTransportException(TTransportException.ExceptionType.Unknown, iox.ToString(), iox); + } + catch (HttpRequestException wx) + { + throw new TTransportException(TTransportException.ExceptionType.Unknown, + "Couldn't connect to server: " + wx, wx); + } + catch (ObjectDisposedException odx) + { + throw new TTransportException(TTransportException.ExceptionType.NotOpen, + "The transport has been closed.", odx); + } + catch (OperationCanceledException ocx) + { + throw new TTransportException(TTransportException.ExceptionType.Interrupted, ocx.Message, ocx); + } + catch (Exception ex) + { + throw new TTransportException(TTransportException.ExceptionType.Unknown, ex.Message, ex); + } + finally + { + response?.Dispose(); + } + } + } + + /// + /// A wrapper whose Dispose is a no-op, used to let an + /// / pair be disposed + /// deterministically without disposing a caller-owned inner stream. + /// + private sealed class NonDisposingStream : Stream + { + private readonly Stream _inner; + + public NonDisposingStream(Stream inner) => _inner = inner; + + public override bool CanRead => _inner.CanRead; + public override bool CanSeek => _inner.CanSeek; + public override bool CanWrite => _inner.CanWrite; + public override long Length => _inner.Length; + public override long Position { get => _inner.Position; set => _inner.Position = value; } + + public override void Flush() => _inner.Flush(); + public override int Read(byte[] buffer, int offset, int count) => _inner.Read(buffer, offset, count); + public override long Seek(long offset, SeekOrigin origin) => _inner.Seek(offset, origin); + public override void SetLength(long value) => _inner.SetLength(value); + public override void Write(byte[] buffer, int offset, int count) => _inner.Write(buffer, offset, count); + + public override Task ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken) => + _inner.ReadAsync(buffer, offset, count, cancellationToken); + +#if NET5_0_OR_GREATER + public override ValueTask ReadAsync(Memory buffer, CancellationToken cancellationToken = default) => + _inner.ReadAsync(buffer, cancellationToken); +#endif + + protected override void Dispose(bool disposing) + { + // Intentionally do not dispose the inner stream; it is owned by the caller. + base.Dispose(disposing); } - _isDisposed = true; } } } diff --git a/lib/netstd/Thrift/Transport/ITPerCallTransportProvider.cs b/lib/netstd/Thrift/Transport/ITPerCallTransportProvider.cs new file mode 100644 index 00000000000..498cbd43716 --- /dev/null +++ b/lib/netstd/Thrift/Transport/ITPerCallTransportProvider.cs @@ -0,0 +1,36 @@ +// 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.Threading; +using System.Threading.Tasks; + +namespace Thrift.Transport +{ + /// + /// Provides a mechanism to create per-call transports for each request. + /// The caller owns each returned transport and must dispose it when the request is complete. + /// + public interface ITPerCallTransportProvider + { + /// + /// Creates a new call transport for the current request. + /// + /// Token used to cancel transport creation. + /// A new transport owned by the caller. + ValueTask CreatePerCallTransportAsync(CancellationToken cancellationToken = default); + } +}