diff --git a/compiler/cpp/src/thrift/generate/t_netstd_generator.cc b/compiler/cpp/src/thrift/generate/t_netstd_generator.cc index 09a999d9885..edf56ace61f 100644 --- a/compiler/cpp/src/thrift/generate/t_netstd_generator.cc +++ b/compiler/cpp/src/thrift/generate/t_netstd_generator.cc @@ -2432,13 +2432,46 @@ void t_netstd_generator::generate_service_client(ostream& out, t_service* tservi << indent() << "{" << '\n'; indent_up(); - out << indent() << "public Client(TProtocol protocol) : this(protocol, protocol)" << '\n' + out << indent() << "/// " << '\n' + << indent() << "/// Initializes a client that reuses a single protocol instance for both directions." << '\n' + << indent() << "/// " << '\n' + << indent() << "/// Use this constructor when you already have a single protocol instance and want the client to share it for the lifetime of the client. This is the shared-protocol path; use the transport/factory overloads when you want a fresh protocol pair per call." << '\n' + << indent() << "/// The shared protocol used for both input and output." << '\n' + << indent() << "public Client(TProtocol protocol) : this(protocol, protocol)" << '\n' << indent() << "{" << '\n' << indent() << "}" << '\n' << '\n' + << indent() << "/// " << '\n' + << indent() << "/// Initializes a client with separate shared input and output protocols." << '\n' + << indent() << "/// " << '\n' + << indent() << "/// Use this constructor when you want a single, long-lived protocol pair for the client. This is the shared-protocol path; the per-call overloads are intended for transports such as that can create a fresh protocol pair for each request." << '\n' + << indent() << "/// The shared input protocol used for the lifetime of the client." << '\n' + << indent() << "/// The shared output protocol used for the lifetime of the client." << '\n' << indent() << "public Client(TProtocol inputProtocol, TProtocol outputProtocol) : base(inputProtocol, outputProtocol)" << '\n' << indent() << "{" << '\n' << indent() << "}" << '\n' + << '\n' + << indent() << "/// " << '\n' + << indent() << "/// Initializes a client that reuses a single protocol pair for the shared client and, when the transport supports it, creates a fresh protocol pair for each call." << '\n' + << indent() << "/// " << '\n' + << indent() << "/// Use this overload for transports such as when you want a per-call protocol pair (for example, to issue overlapping HTTP requests in parallel). If the transport does not support per-call mode, the client continues to use the shared protocol instance." << '\n' + << indent() << "/// The shared transport. The client owns protocols created for this transport." << '\n' + << indent() << "/// The factory used to create input and output protocols for the shared client and, when supported, each call." << '\n' + << indent() << "public Client(TTransport transport, TProtocolFactory protocolFactory) : this(transport, protocolFactory, protocolFactory)" << '\n' + << indent() << "{" << '\n' + << indent() << "}" << '\n' + << '\n' + << indent() << "/// " << '\n' + << indent() << "/// Initializes a client with separate factories for the shared client and, when the transport supports it, a fresh protocol pair per call." << '\n' + << indent() << "/// " << '\n' + << indent() << "/// Use this overload when the transport supports a per-call protocol path, such as concurrent HTTP calls through . The transport may still choose to reuse shared protocols if it does not implement ." << '\n' + << indent() << "/// The shared transport. The client owns protocols created for this transport." << '\n' + << indent() << "/// The factory used to create the shared input protocol and per-call input protocols when supported." << '\n' + << indent() << "/// The factory used to create the shared output protocol and per-call output protocols when supported." << '\n' + << indent() << "public Client(TTransport transport, TProtocolFactory inputProtocolFactory, TProtocolFactory outputProtocolFactory)" << '\n' + << indent() << " : base(transport, inputProtocolFactory, outputProtocolFactory)" << '\n' + << indent() << "{" << '\n' + << indent() << "}" << '\n' << '\n'; vector functions = tservice->get_functions(); @@ -2454,6 +2487,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 +2505,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/README.md b/lib/netstd/README.md index a3623e1c599..273304db118 100644 --- a/lib/netstd/README.md +++ b/lib/netstd/README.md @@ -27,6 +27,58 @@ or - Ensure you have a suitable .NET Core SDK installed, or use the [Ubuntu docker image](../../build/docker/README.md) - Follow common automake build practice: `./ bootstrap && ./ configure && make` +## Per-call HTTP transport mode + +`THttpTransport` supports two modes: + +- shared mode: a single protocol pair is reused for the lifetime of the client. +- per-call mode: a fresh protocol pair is created for each request when the client is constructed with a transport and protocol factory. + +The shared mode is the default for code such as: + +```csharp +var http = new THttpTransport(new Uri("http://localhost:9090")); +var client = new Calculator.Client(new TBinaryProtocol(http)); +``` + +In this form, the same `HttpClient`, default headers, and `ConnectTimeout` remain in effect for the whole client, and the request/response buffers are also reused by the shared protocol instance. + +To opt into the per-call path, construct the client with the transport and a protocol factory instead of an already-created protocol instance: + +```csharp +var http = new THttpTransport(new Uri("http://localhost:9090")); +var client = new Calculator.Client(http, new TBinaryProtocol.Factory()); +``` + +This constructor enables `ITPerCallTransportProvider` behavior for the generated RPC methods such as `add()`: each method scopes its send and receive operations to a fresh protocol pair for that call. Calling generated `send_add()` and `recv_add()` separately does not use that per-call scope; those methods use the client's shared protocols. Use the regular RPC method when you need per-call behavior. + +The per-call path is useful when you want multiple HTTP requests to be in flight at once. Each call gets its own request stream, response stream, and buffered protocol state, while the underlying `HttpClient`, default headers, and `ConnectTimeout` remain shared. In other words: + +- shared across calls: `HttpClient`, default headers, `ConnectTimeout`, connection-level settings +- private to each call: request payload, response payload, protocol buffers, per-call `TTransport` state + +The same pattern also works when you need a protocol wrapper such as `TMultiplexedProtocol`. Netstd does not provide a built-in `TMultiplexedProtocol.Factory`, so you create a small protocol factory subclass that creates the wrapper for each per-call transport: + +```csharp +var http = new THttpTransport(new Uri("http://localhost:9090")); +var client = new Calculator.Client(http, new MultiplexedBinaryProtocolFactory("Calculator")); + +sealed class MultiplexedBinaryProtocolFactory : TProtocolFactory +{ + private readonly string _serviceName; + + public MultiplexedBinaryProtocolFactory(string serviceName) + { + _serviceName = serviceName; + } + + public override TProtocol GetProtocol(TTransport trans) + => new TMultiplexedProtocol(new TBinaryProtocol(trans), _serviceName); +} +``` + +Socket transports such as `TSocketTransport` are not affected. They do not implement `ITPerCallTransportProvider`, so they continue to use a single shared transport/protocol pair. Wrappers such as `TBufferedTransport` or `TFramedTransport` also disable the per-call path because they are not themselves transport providers. + ## Known issues - In trace logging mode you can see some not important internal exceptions diff --git a/lib/netstd/Tests/Thrift.Tests/Transports/THttpTransportTests.cs b/lib/netstd/Tests/Thrift.Tests/Transports/THttpTransportTests.cs index 2a2d884203a..2608a7b069c 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,967 @@ 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 BufferedTransport_PreservesPerCallTransportSupport() + { + using var parent = CreateTransport(new TrackingHandler(_ => CreateResponse("ok"))); + var buffered = new TBufferedTransport(parent); + + var provider = buffered as ITPerCallTransportProvider; + Assert.IsNotNull(provider); + + using var call = await provider.CreatePerCallTransportAsync(); + Assert.IsInstanceOfType(call, typeof(TBufferedTransport)); + Assert.AreNotSame(parent, ((TBufferedTransport)call).UnderlyingTransport); + } + + [TestMethod] + public async Task FramedTransport_PreservesPerCallTransportSupport() + { + using var parent = CreateTransport(new TrackingHandler(_ => CreateResponse("ok"))); + var framed = new TFramedTransport(parent); + + var provider = framed as ITPerCallTransportProvider; + Assert.IsNotNull(provider); + + using var call = await provider.CreatePerCallTransportAsync(); + Assert.IsInstanceOfType(call, typeof(TFramedTransport)); + Assert.AreNotSame(parent, ((TFramedTransport)call).InnerTransport); + } + + [TestMethod] + public async Task LayeredTransport_WithoutPerCallSupport_Throws() + { + var buffered = new TBufferedTransport(new TMemoryBufferTransport(new TConfiguration())); + var framed = new TFramedTransport(new TMemoryBufferTransport(new TConfiguration())); + + var bufferedException = await Assert.ThrowsAsync( + () => buffered.CreatePerCallTransportAsync().AsTask()); + StringAssert.Contains(bufferedException.Message, nameof(ITPerCallTransportProvider)); + + var framedException = await Assert.ThrowsAsync( + () => framed.CreatePerCallTransportAsync().AsTask()); + StringAssert.Contains(framedException.Message, nameof(ITPerCallTransportProvider)); + } + + [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 LegacyResponseReadCancellationDisposesResponseStream() + { + var sharedResponseStream = new BlockingReadStream(); + using (var parent = CreateTransport(new BlockingResponseHandler(sharedResponseStream))) + { + 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); + Assert.IsTrue(sharedResponseStream.IsDisposed); + } + + var perCallResponseStream = new BlockingReadStream(); + using (var parent = CreateTransport(new BlockingResponseHandler(perCallResponseStream))) + { + 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); + Assert.IsTrue(perCallResponseStream.IsDisposed); + call.Dispose(); + } + } + + [TestMethod] + public async Task CancellationAfterCompletedSharedReadDoesNotDisposeResponseStream() + { + using var cancellation = new CancellationTokenSource(); + var responseStream = new CancellationAfterReadStream(cancellation.Cancel); + using var parent = CreateTransport(new TrackingHandler(_ => + new HttpResponseMessage(HttpStatusCode.OK) { Content = new TrackingContent(responseStream) })); + await parent.WriteAsync(new byte[] { 1 }, CancellationToken.None); + await parent.FlushAsync(CancellationToken.None); + + var count = await parent.ReadAsync(new byte[1], 0, 1, cancellation.Token); + + Assert.AreEqual(1, count); + Assert.IsFalse(responseStream.IsDisposed); + parent.Close(); + Assert.IsTrue(responseStream.IsDisposed); + } + + [TestMethod] + public async Task CancellationAfterCompletedPerCallReadDoesNotDisposeResponseStream() + { + using var cancellation = new CancellationTokenSource(); + var responseStream = new CancellationAfterReadStream(cancellation.Cancel); + using var parent = CreateTransport(new TrackingHandler(_ => + new HttpResponseMessage(HttpStatusCode.OK) { Content = new TrackingContent(responseStream) })); + var call = await parent.CreatePerCallTransportAsync(); + await call.WriteAsync(new byte[] { 1 }, CancellationToken.None); + await call.FlushAsync(CancellationToken.None); + + var count = await call.ReadAsync(new byte[1], 0, 1, cancellation.Token); + + Assert.AreEqual(1, count); + Assert.IsFalse(responseStream.IsDisposed); + call.Dispose(); + Assert.IsTrue(responseStream.IsDisposed); + } + + [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 BaseClient_UsesIndependentPerCallProtocols() + { + using var transport = CreateTransport(new TrackingHandler(_ => CreateResponse("ok"))); + var protocolFactory = new TrackingBinaryProtocolFactory(); + var client = new PerCallTrackingClient(transport, protocolFactory, protocolFactory); + + try + { + var firstCall = client.RunPerCallAsync(); + var secondCall = client.RunPerCallAsync(); + + var responses = await Task.WhenAll(firstCall, secondCall); + Assert.AreEqual(2, responses.Length); + CollectionAssert.AreEquivalent(new[] { "ok", "ok" }, responses); + Assert.AreEqual(6, protocolFactory.CreatedCount); + } + finally + { + client.Dispose(); + } + } + + [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 PerCallTrackingClient : TBaseClient + { + public PerCallTrackingClient( + TTransport transport, + TProtocolFactory inputProtocolFactory, + TProtocolFactory outputProtocolFactory) + : base(transport, inputProtocolFactory, outputProtocolFactory) + { + } + + public Task RunPerCallAsync() + { + return ExecutePerCallAsync(() => Task.FromResult("ok"), CancellationToken.None); + } + } + + 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 + { + private readonly BlockingReadStream _responseStream; + + public BlockingResponseHandler() + : this(new BlockingReadStream()) + { + } + + public BlockingResponseHandler(BlockingReadStream responseStream) + { + _responseStream = responseStream; + } + + protected override Task SendAsync(HttpRequestMessage request, CancellationToken cancellationToken) + { + return Task.FromResult(new HttpResponseMessage(HttpStatusCode.OK) + { + Content = new BlockingReadContent(_responseStream) + }); + } + } + + private sealed class DisposedHttpClientHandler : HttpMessageHandler + { + protected override Task SendAsync( + HttpRequestMessage request, + CancellationToken cancellationToken) + { + throw new ObjectDisposedException("HttpClient"); + } + } + + private sealed class BlockingReadContent : HttpContent + { + private readonly BlockingReadStream _responseStream; + + public BlockingReadContent(BlockingReadStream responseStream) + { + _responseStream = responseStream; + } + + 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(_responseStream); + } + + private sealed class BlockingReadStream : Stream + { + private readonly TaskCompletionSource _pendingRead = + new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + private bool _disposed; + + public bool IsDisposed => _disposed; + public override bool CanRead => !_disposed; + 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 Task ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken) => + _pendingRead.Task; + public override ValueTask ReadAsync(Memory buffer, CancellationToken cancellationToken = default) + { + return new ValueTask(_pendingRead.Task); + } + protected override void Dispose(bool disposing) + { + if (disposing && !_disposed) + { + _disposed = true; + _pendingRead.TrySetException(new ObjectDisposedException(nameof(BlockingReadStream))); + } + base.Dispose(disposing); + } + } + + private sealed class CancellationAfterReadStream : Stream + { + private readonly Action _cancel; + private int _readCount; + + public CancellationAfterReadStream(Action cancel) => _cancel = cancel; + public bool IsDisposed { get; private set; } + public override bool CanRead => !IsDisposed; + 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 Task ReadAsync( + byte[] buffer, + int offset, + int count, + CancellationToken cancellationToken) + { + var completedRead = Task.FromResult(1); + if (Interlocked.Increment(ref _readCount) == 1) + { + _cancel(); + } + return completedRead; + } + + public override ValueTask ReadAsync( + Memory buffer, + CancellationToken cancellationToken = default) + { + var completedRead = new ValueTask(1); + if (Interlocked.Increment(ref _readCount) == 1) + { + _cancel(); + } + return completedRead; + } + + protected override void Dispose(bool disposing) + { + if (disposing) + { + IsDisposed = true; + } + base.Dispose(disposing); + } + } + + 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..4e7025f1012 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,209 @@ protected TBaseClient(TProtocol inputProtocol, TProtocol outputProtocol) _outputProtocol = outputProtocol ?? throw new ArgumentNullException(nameof(outputProtocol)); } - public TProtocol InputProtocol => _inputProtocol; + /// + /// Initializes a client with protocols created from the supplied transport and protocol factories. + /// + /// + /// The shared transport. When it implements , + /// each client call uses protocols created for a new per-call transport. + /// + /// The factory used to create input protocols. + /// The factory used to create output protocols. + /// + /// Thrown when , , + /// or is . + /// + /// + /// Thrown when a protocol factory returns . + /// + 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; + } + } - public TProtocol OutputProtocol => _outputProtocol; + 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; + } + } + private sealed class ProtocolPair + { + public ProtocolPair(TProtocol inputProtocol, TProtocol outputProtocol) + { + InputProtocol = inputProtocol; + OutputProtocol = outputProtocol; + } + + public TProtocol InputProtocol { get; } + public TProtocol OutputProtocol { get; } + } + + /// + /// Gets the input protocol for the current call, or the shared input protocol when no + /// per-call transport is active. + /// + public TProtocol InputProtocol => _perCallProtocols.Value?.InputProtocol ?? _inputProtocol; + + /// + /// Gets the output protocol for the current call, or the shared output protocol when no + /// per-call transport is active. + /// + public TProtocol OutputProtocol => _perCallProtocols.Value?.OutputProtocol ?? _outputProtocol; + + /// + /// Gets the next client sequence identifier. + /// public int SeqId { - get { return ++_seqId; } + get { return Interlocked.Increment(ref _seqId); } + } + + /// + /// Executes an operation using protocols scoped to a new per-call transport when the + /// shared transport supports per-call transports. + /// + /// The operation to execute. + /// The token used to cancel per-call transport creation. + /// A task that represents the asynchronous operation. + protected async Task ExecutePerCallAsync(Func operation, CancellationToken cancellationToken) + { + await ExecutePerCallCoreAsync(async () => + { + await operation(); + return true; + }, cancellationToken); + } + + /// + /// Executes an operation using protocols scoped to a new per-call transport when the + /// shared transport supports per-call transports. + /// + /// The type of result returned by the operation. + /// The operation to execute. + /// The token used to cancel per-call transport creation. + /// A task that contains the operation result. + 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) + ?? throw new InvalidOperationException("The per-call transport provider returned null."); + 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..39466da6752 --- /dev/null +++ b/lib/netstd/Thrift/Transport/Client/THttpPerCallTransport.cs @@ -0,0 +1,328 @@ +// 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) + { + _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; + CancellationToken lifecycleToken; + CancellationToken responseTimeoutToken; + lock (_parent.LifecycleLock) + { + if (_disposed || !_parent.IsLifecycleOpen || _responseStream == null) + { + throw new TTransportException(TTransportException.ExceptionType.NotOpen, + "The transport has been closed."); + } + + responseStream = _responseStream; + lifecycleToken = _lifecycleCancellation.Token; + responseTimeoutToken = _responseReadTimeout?.Token ?? CancellationToken.None; + readCts = CancellationTokenSource.CreateLinkedTokenSource( + cancellationToken, + lifecycleToken, + responseTimeoutToken); + } + + var readToken = readCts?.Token ?? cancellationToken; + try + { + var count = await THttpTransport.ReadResponseAsync( + responseStream, + buffer, + offset, + length, + readToken); + + CountConsumedMessageBytes(count); + return count; + } + catch (OperationCanceledException ocx) when (!cancellationToken.IsCancellationRequested && + (responseTimeoutToken.IsCancellationRequested || lifecycleToken.IsCancellationRequested || !IsOpen)) + { + throw new TTransportException(TTransportException.ExceptionType.Interrupted, ocx.Message, ocx); + } + catch (ObjectDisposedException odx) when (cancellationToken.IsCancellationRequested) + { + throw new OperationCanceledException(odx.Message, odx, cancellationToken); + } + catch (ObjectDisposedException odx) when (responseTimeoutToken.IsCancellationRequested || + lifecycleToken.IsCancellationRequested || !IsOpen) + { + throw new TTransportException(TTransportException.ExceptionType.Interrupted, odx.Message, odx); + } + catch (ObjectDisposedException odx) + { + throw new TTransportException(TTransportException.ExceptionType.NotOpen, + "The transport has been closed.", odx); + } + catch (IOException iox) when (cancellationToken.IsCancellationRequested) + { + throw new OperationCanceledException(iox.Message, iox, cancellationToken); + } + catch (IOException iox) when (responseTimeoutToken.IsCancellationRequested || + lifecycleToken.IsCancellationRequested || !IsOpen) + { + throw new TTransportException(TTransportException.ExceptionType.Interrupted, iox.Message, iox); + } + 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. + Dispose(); + } + } +} diff --git a/lib/netstd/Thrift/Transport/Client/THttpTransport.cs b/lib/netstd/Thrift/Transport/Client/THttpTransport.cs index 439777eb390..e169a23fccc 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,38 @@ public override async ValueTask ReadAsync(byte[] buffer, int offset, int le CheckReadBytesAvailable(length); + CancellationTokenSource readCts = null; + Stream inputStream; + CancellationToken lifecycleToken; + CancellationToken responseTimeoutToken; + lock (_lifecycleLock) + { + if (!_lifecycleOpen || _inputStream == null) + { + throw new TTransportException(TTransportException.ExceptionType.NotOpen, + "The transport has been closed."); + } + + inputStream = _inputStream; + lifecycleToken = _lifecycleCancellation.Token; + responseTimeoutToken = _responseReadTimeout?.Token ?? CancellationToken.None; + readCts = CancellationTokenSource.CreateLinkedTokenSource( + cancellationToken, + lifecycleToken, + responseTimeoutToken); + } + 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 ReadResponseAsync( + inputStream, + new Memory(buffer, offset, length), + readToken); #else - var ret = await _inputStream.ReadAsync(buffer, offset, length, cancellationToken); + ret = await ReadResponseAsync(inputStream, buffer, offset, length, readToken); #endif if (ret == -1) { @@ -166,20 +232,65 @@ public override async ValueTask ReadAsync(byte[] buffer, int offset, int le CountConsumedMessageBytes(ret); return ret; } + catch (OperationCanceledException ocx) when (!cancellationToken.IsCancellationRequested && + (responseTimeoutToken.IsCancellationRequested || lifecycleToken.IsCancellationRequested || + !IsLifecycleOpen)) + { + throw new TTransportException(TTransportException.ExceptionType.Interrupted, ocx.Message, ocx); + } + catch (ObjectDisposedException odx) when (cancellationToken.IsCancellationRequested) + { + throw new OperationCanceledException(odx.Message, odx, cancellationToken); + } + catch (ObjectDisposedException odx) when (responseTimeoutToken.IsCancellationRequested || + lifecycleToken.IsCancellationRequested || !IsLifecycleOpen) + { + throw new TTransportException(TTransportException.ExceptionType.Interrupted, odx.Message, odx); + } + catch (ObjectDisposedException odx) + { + throw new TTransportException(TTransportException.ExceptionType.NotOpen, + "The transport has been closed.", odx); + } + catch (IOException iox) when (cancellationToken.IsCancellationRequested) + { + throw new OperationCanceledException(iox.Message, iox, cancellationToken); + } + catch (IOException iox) when (responseTimeoutToken.IsCancellationRequested || + lifecycleToken.IsCancellationRequested || !IsLifecycleOpen) + { + throw new TTransportException(TTransportException.ExceptionType.Interrupted, iox.Message, iox); + } 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 +348,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); + + contentStream = new StreamContent(outputStream); + contentStream.Headers.ContentType = ContentType ?? new MediaTypeHeaderValue(@"application/x-thrift"); - using (var contentStream = new StreamContent(_outputStream)) + 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) { - contentStream.Headers.ContentType = ContentType ?? new MediaTypeHeaderValue(@"application/x-thrift"); + responseReadTimeout.CancelAfter(Math.Max(0, remainingTimeout)); + } - var response = (await _httpClient.PostAsync(_uri, contentStream, cancellationToken)).EnsureSuccessStatusCode(); + 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 +443,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 +458,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 +489,319 @@ 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 + } + + internal static async Task ReadResponseAsync( + Stream stream, + byte[] buffer, + int offset, + int length, + CancellationToken cancellationToken) + { + var readTask = stream.ReadAsync(buffer, offset, length, cancellationToken); + return await AwaitResponseReadAsync(stream, readTask, cancellationToken); + } + +#if NET5_0_OR_GREATER + internal static async ValueTask ReadResponseAsync( + Stream stream, + Memory buffer, + CancellationToken cancellationToken) + { + var readTask = stream.ReadAsync(buffer, cancellationToken).AsTask(); + return await AwaitResponseReadAsync(stream, readTask, cancellationToken); + } +#endif + + private static async Task AwaitResponseReadAsync( + Stream stream, + Task readTask, + CancellationToken cancellationToken) + { + var pendingRead = new PendingResponseRead(stream, readTask); + using (cancellationToken.Register(CancelPendingResponseRead, pendingRead)) + { + try + { + var count = await readTask; + pendingRead.ThrowIfCancellationWon(cancellationToken); + return count; + } + catch (Exception ex) when (pendingRead.CancellationWon && + (ex is ObjectDisposedException || ex is IOException)) + { + throw new OperationCanceledException("The response read was canceled.", ex, cancellationToken); + } + } + } + + private static void CancelPendingResponseRead(object state) + { + ((PendingResponseRead)state).Cancel(); + } + + private sealed class PendingResponseRead + { + private readonly object _lock = new object(); + private readonly Stream _stream; + private readonly Task _readTask; + private bool _cancellationWon; + + public PendingResponseRead(Stream stream, Task readTask) + { + _stream = stream; + _readTask = readTask; + } + + public bool CancellationWon + { + get + { + lock (_lock) + { + return _cancellationWon; + } + } + } + + public void Cancel() + { + lock (_lock) + { + if (_readTask.IsCompleted) + { + return; + } + + _cancellationWon = true; + } + + _stream.Dispose(); + } + + public void ThrowIfCancellationWon(CancellationToken cancellationToken) + { + lock (_lock) + { + if (_cancellationWon) + { + throw new OperationCanceledException( + "The response read was canceled.", cancellationToken); + } + } + } + } + + /// + /// 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); + } +} diff --git a/lib/netstd/Thrift/Transport/Layered/TBufferedTransport.cs b/lib/netstd/Thrift/Transport/Layered/TBufferedTransport.cs index 977dcbfd119..246f9d45b0c 100644 --- a/lib/netstd/Thrift/Transport/Layered/TBufferedTransport.cs +++ b/lib/netstd/Thrift/Transport/Layered/TBufferedTransport.cs @@ -67,6 +67,11 @@ public TTransport UnderlyingTransport } } + protected override TTransport CreatePerCallWrapper(TTransport innerTransport) + { + return new TBufferedTransport(innerTransport, DesiredBufferSize); + } + public override bool IsOpen => !IsDisposed && InnerTransport.IsOpen; public override async Task OpenAsync(CancellationToken cancellationToken) diff --git a/lib/netstd/Thrift/Transport/Layered/TFramedTransport.cs b/lib/netstd/Thrift/Transport/Layered/TFramedTransport.cs index faa3fa6e90f..8525c3ced9b 100644 --- a/lib/netstd/Thrift/Transport/Layered/TFramedTransport.cs +++ b/lib/netstd/Thrift/Transport/Layered/TFramedTransport.cs @@ -53,6 +53,11 @@ public TFramedTransport(TTransport transport) InitWriteBuffer(); } + protected override TTransport CreatePerCallWrapper(TTransport innerTransport) + { + return new TFramedTransport(innerTransport); + } + public override bool IsOpen => !IsDisposed && InnerTransport.IsOpen; public override async Task OpenAsync(CancellationToken cancellationToken) diff --git a/lib/netstd/Thrift/Transport/Layered/TLayeredTransport.cs b/lib/netstd/Thrift/Transport/Layered/TLayeredTransport.cs index f9a4ce477af..80e8293f318 100644 --- a/lib/netstd/Thrift/Transport/Layered/TLayeredTransport.cs +++ b/lib/netstd/Thrift/Transport/Layered/TLayeredTransport.cs @@ -16,13 +16,15 @@ // under the License. using System; +using System.Threading; +using System.Threading.Tasks; #pragma warning disable IDE0079 // net20 - unneeded suppression #pragma warning disable IDE0290 // net8 - primary CTOR namespace Thrift.Transport { - public abstract class TLayeredTransport : TTransport + public abstract class TLayeredTransport : TTransport, ITPerCallTransportProvider { public readonly TTransport InnerTransport; @@ -33,6 +35,20 @@ public TLayeredTransport(TTransport transport) InnerTransport = transport ?? throw new ArgumentNullException(nameof(transport)); } + public virtual async ValueTask CreatePerCallTransportAsync(CancellationToken cancellationToken = default) + { + if (!(InnerTransport is ITPerCallTransportProvider provider)) + { + throw new InvalidOperationException( + $"{GetType().Name} cannot provide a per-call transport because its inner transport does not implement {nameof(ITPerCallTransportProvider)}."); + } + + var perCallTransport = await provider.CreatePerCallTransportAsync(cancellationToken); + return CreatePerCallWrapper(perCallTransport); + } + + protected abstract TTransport CreatePerCallWrapper(TTransport innerTransport); + public override void UpdateKnownMessageSize(long size) { InnerTransport.UpdateKnownMessageSize(size); diff --git a/tutorial/netstd/Client/Program.cs b/tutorial/netstd/Client/Program.cs index 29e3f3c3597..a24cf83e14c 100644 --- a/tutorial/netstd/Client/Program.cs +++ b/tutorial/netstd/Client/Program.cs @@ -66,6 +66,9 @@ will diplay help information Client -tr: -bf: -pr: [-mc:] [-multiplex] will run client with specified arguments (tcp transport and binary protocol by default) and with 1 client + Client -tr:http -parallel + will issue concurrent HTTP calls through one per-call client + Options: -tr (transport): tcp - (default) tcp transport (localhost:9090) @@ -88,6 +91,9 @@ will run client with specified arguments (tcp transport and binary protocol by d -mc (multiple clients): - number of multiple clients to connect to server (max 100, default 1) + -parallel (HTTP only): + issue three concurrent add() calls through one client using per-call transports + Sample: Client -tr:tcp -pr:binary "); @@ -114,6 +120,31 @@ public static async Task Main(string[] args) private static async Task RunAsync(string[] args, CancellationToken cancellationToken) { + if (args.Any(x => x.Equals("-parallel", StringComparison.OrdinalIgnoreCase))) + { + if (GetTransport(args) != Transport.Http) + { + throw new ArgumentException("The -parallel option requires -tr:http."); + } + + if (GetMultiplex(args)) + { + throw new ArgumentException("The -parallel option does not support -multiplex."); + } + + if (GetBuffering(args) != Buffering.None) + { + throw new ArgumentException("The -parallel option requires -bf:none."); + } + + if (GetProtocol(args) != Protocol.Binary) + { + throw new ArgumentException("The tutorial HTTP server uses the binary protocol with -parallel."); + } + + return await RunConcurrentHttpCallsAsync(args, cancellationToken); + } + var numClients = GetNumberOfClients(args); if (Logger.IsEnabled(LogLevel.Information)) @@ -143,6 +174,36 @@ private static async Task RunAsync(string[] args, CancellationToken cancel return tasks.All(task => task.Result); } + private static async Task RunConcurrentHttpCallsAsync( + string[] args, + CancellationToken cancellationToken) + { + var transport = MakeTransport(args); + var client = new Calculator.Client(transport, new TBinaryProtocol.Factory()); + try + { + await client.OpenTransportAsync(cancellationToken); + var sums = await Task.WhenAll( + client.add(1, 1, cancellationToken), + client.add(2, 3, cancellationToken), + client.add(10, 20, cancellationToken)); + + var expected = new[] { 2, 5, 30 }; + if (!sums.SequenceEqual(expected)) + { + throw new InvalidOperationException( + $"Concurrent HTTP calls returned unexpected results: {string.Join(", ", sums)}."); + } + + Logger.LogInformation("Concurrent HTTP add() results: {results}", string.Join(", ", sums)); + return true; + } + finally + { + client.Dispose(); + } + } + private static bool GetMultiplex(string[] args) { var mplex = args.FirstOrDefault(x => x.StartsWith("-multiplex")); diff --git a/tutorial/netstd/README.md b/tutorial/netstd/README.md index 8301e2702ed..c4577d13113 100644 --- a/tutorial/netstd/README.md +++ b/tutorial/netstd/README.md @@ -109,6 +109,20 @@ Remarks: with binaries in case of command line usage (or at project level in case of debugging from IDE). Password for certificate - "ThriftTest". +## Concurrent HTTP calls + +The client tutorial can issue several requests concurrently through one `Calculator.Client`. Its transport-and-protocol-factory constructor enables the generated methods to use an independent protocol and HTTP request/response state for each call. + +Build the tutorial solution, start the HTTP server in one terminal, then run the client in another (from `tutorial/netstd`): + +```sh +dotnet build Tutorial.slnx +dotnet run --project Server/Server.csproj -- -tr:http +dotnet run --project Client/Client.csproj -- -tr:http -parallel +``` + +The client checks the expected results (`2, 5, 30`) and disposes the client after all calls complete. `-mc:` remains a separate option for running multiple clients, each with its own transport. + # How to test communication between NetCore and Python * Generate code with the latest **thrift** utility