From a610e32e4b07d860048d478e2f2c851c7c9bb4d6 Mon Sep 17 00:00:00 2001 From: Jan Tattermusch Date: Tue, 13 Sep 2016 16:53:05 +0200 Subject: [PATCH 1/3] throw correct exception failed writes --- .../Internal/AsyncCallServerTest.cs | 4 +- .../Grpc.Core.Tests/Internal/AsyncCallTest.cs | 74 ++++++++++++++++++- src/csharp/Grpc.Core/Internal/AsyncCall.cs | 22 +++++- .../Grpc.Core/Internal/AsyncCallBase.cs | 32 +++++++- .../Grpc.Core/Internal/AsyncCallServer.cs | 6 ++ 5 files changed, 129 insertions(+), 9 deletions(-) diff --git a/src/csharp/Grpc.Core.Tests/Internal/AsyncCallServerTest.cs b/src/csharp/Grpc.Core.Tests/Internal/AsyncCallServerTest.cs index c35aaf680f7..09790120d17 100644 --- a/src/csharp/Grpc.Core.Tests/Internal/AsyncCallServerTest.cs +++ b/src/csharp/Grpc.Core.Tests/Internal/AsyncCallServerTest.cs @@ -33,6 +33,7 @@ using System; using System.Collections.Generic; +using System.IO; using System.Runtime.InteropServices; using System.Threading.Tasks; @@ -149,8 +150,7 @@ namespace Grpc.Core.Internal.Tests var writeTask = responseStream.WriteAsync("request1"); fakeCall.SendCompletionHandler(false); - // TODO(jtattermusch): should we throw a different exception type instead? - Assert.ThrowsAsync(typeof(InvalidOperationException), async () => await writeTask); + Assert.ThrowsAsync(typeof(IOException), async () => await writeTask); fakeCall.ReceivedCloseOnServerHandler(true, cancelled: true); AssertFinished(asyncCallServer, fakeCall, finishedTask); diff --git a/src/csharp/Grpc.Core.Tests/Internal/AsyncCallTest.cs b/src/csharp/Grpc.Core.Tests/Internal/AsyncCallTest.cs index 98e27a17a1d..8ee4d184abc 100644 --- a/src/csharp/Grpc.Core.Tests/Internal/AsyncCallTest.cs +++ b/src/csharp/Grpc.Core.Tests/Internal/AsyncCallTest.cs @@ -180,21 +180,46 @@ namespace Grpc.Core.Internal.Tests } [Test] - public void ClientStreaming_WriteCompletionFailure() + public void ClientStreaming_WriteFailureThrowsRpcException() { var resultTask = asyncCall.ClientStreamingCallAsync(); var requestStream = new ClientRequestStream(asyncCall); var writeTask = requestStream.WriteAsync("request1"); fakeCall.SendCompletionHandler(false); - // TODO: maybe IOException or waiting for RPCException is more appropriate here. - Assert.ThrowsAsync(typeof(InvalidOperationException), async () => await writeTask); + + // The write will wait for call to finish to receive the status code. + Assert.IsFalse(writeTask.IsCompleted); + + fakeCall.UnaryResponseClientHandler(true, + CreateClientSideStatus(StatusCode.Internal), + null, + new Metadata()); + + var ex = Assert.ThrowsAsync(async () => await writeTask); + Assert.AreEqual(StatusCode.Internal, ex.Status.StatusCode); + + AssertUnaryResponseError(asyncCall, fakeCall, resultTask, StatusCode.Internal); + } + + [Test] + public void ClientStreaming_WriteFailureThrowsRpcException2() + { + var resultTask = asyncCall.ClientStreamingCallAsync(); + var requestStream = new ClientRequestStream(asyncCall); + + var writeTask = requestStream.WriteAsync("request1"); fakeCall.UnaryResponseClientHandler(true, CreateClientSideStatus(StatusCode.Internal), null, new Metadata()); + fakeCall.SendCompletionHandler(false); + + var ex = Assert.ThrowsAsync(async () => await writeTask); + Assert.AreEqual(StatusCode.Internal, ex.Status.StatusCode); + AssertUnaryResponseError(asyncCall, fakeCall, resultTask, StatusCode.Internal); } @@ -415,6 +440,49 @@ namespace Grpc.Core.Internal.Tests Assert.DoesNotThrowAsync(async () => await requestStream.CompleteAsync()); } + [Test] + public void DuplexStreaming_WriteFailureThrowsRpcException() + { + asyncCall.StartDuplexStreamingCall(); + var requestStream = new ClientRequestStream(asyncCall); + var responseStream = new ClientResponseStream(asyncCall); + + var writeTask = requestStream.WriteAsync("request1"); + fakeCall.SendCompletionHandler(false); + + // The write will wait for call to finish to receive the status code. + Assert.IsFalse(writeTask.IsCompleted); + + var readTask = responseStream.MoveNext(); + fakeCall.ReceivedMessageHandler(true, null); + fakeCall.ReceivedStatusOnClientHandler(true, CreateClientSideStatus(StatusCode.PermissionDenied)); + + var ex = Assert.ThrowsAsync(async () => await writeTask); + Assert.AreEqual(StatusCode.PermissionDenied, ex.Status.StatusCode); + + AssertStreamingResponseError(asyncCall, fakeCall, readTask, StatusCode.PermissionDenied); + } + + [Test] + public void DuplexStreaming_WriteFailureThrowsRpcException2() + { + asyncCall.StartDuplexStreamingCall(); + var requestStream = new ClientRequestStream(asyncCall); + var responseStream = new ClientResponseStream(asyncCall); + + var writeTask = requestStream.WriteAsync("request1"); + + var readTask = responseStream.MoveNext(); + fakeCall.ReceivedMessageHandler(true, null); + fakeCall.ReceivedStatusOnClientHandler(true, CreateClientSideStatus(StatusCode.PermissionDenied)); + fakeCall.SendCompletionHandler(false); + + var ex = Assert.ThrowsAsync(async () => await writeTask); + Assert.AreEqual(StatusCode.PermissionDenied, ex.Status.StatusCode); + + AssertStreamingResponseError(asyncCall, fakeCall, readTask, StatusCode.PermissionDenied); + } + [Test] public void DuplexStreaming_WriteAfterCancellationRequestThrowsTaskCanceledException() { diff --git a/src/csharp/Grpc.Core/Internal/AsyncCall.cs b/src/csharp/Grpc.Core/Internal/AsyncCall.cs index f549c528762..db40b553ce1 100644 --- a/src/csharp/Grpc.Core/Internal/AsyncCall.cs +++ b/src/csharp/Grpc.Core/Internal/AsyncCall.cs @@ -341,6 +341,11 @@ namespace Grpc.Core.Internal get { return true; } } + protected override Exception GetRpcExceptionClientOnly() + { + return new RpcException(finishedStatus.Value.Status); + } + protected override Task CheckSendAllowedOrEarlyResult() { var earlyResult = CheckSendPreconditionsClientSide(); @@ -452,6 +457,7 @@ namespace Grpc.Core.Internal using (Profilers.ForCurrentThread().NewScope("AsyncCall.HandleUnaryResponse")) { + TaskCompletionSource delayedTcs; TResponse msg = default(TResponse); var deserializeException = TryDeserialize(receivedMessage, out msg); @@ -464,14 +470,19 @@ namespace Grpc.Core.Internal receivedStatus = new ClientSideStatus(DeserializeResponseFailureStatus, receivedStatus.Trailers); } finishedStatus = receivedStatus; + delayedTcs = delayedStreamingWriteTcs; ReleaseResourcesIfPossible(); } responseHeadersTcs.SetResult(responseHeaders); - var status = receivedStatus.Status; + if (delayedTcs != null) + { + delayedTcs.SetException(GetRpcExceptionClientOnly()); + } + var status = receivedStatus.Status; if (status.StatusCode != StatusCode.OK) { unaryResponseTcs.SetException(new RpcException(status)); @@ -490,16 +501,23 @@ namespace Grpc.Core.Internal // NOTE: because this event is a result of batch containing GRPC_OP_RECV_STATUS_ON_CLIENT, // success will be always set to true. + TaskCompletionSource delayedTcs; + lock (myLock) { finished = true; finishedStatus = receivedStatus; + delayedTcs = delayedStreamingWriteTcs; ReleaseResourcesIfPossible(); } - var status = receivedStatus.Status; + if (delayedTcs != null) + { + delayedTcs.SetException(GetRpcExceptionClientOnly()); + } + var status = receivedStatus.Status; if (status.StatusCode != StatusCode.OK) { streamingCallFinishedTcs.SetException(new RpcException(status)); diff --git a/src/csharp/Grpc.Core/Internal/AsyncCallBase.cs b/src/csharp/Grpc.Core/Internal/AsyncCallBase.cs index eb9c3ea62d1..b27baba942f 100644 --- a/src/csharp/Grpc.Core/Internal/AsyncCallBase.cs +++ b/src/csharp/Grpc.Core/Internal/AsyncCallBase.cs @@ -68,6 +68,7 @@ namespace Grpc.Core.Internal protected TaskCompletionSource streamingReadTcs; // Completion of a pending streaming read if not null. protected TaskCompletionSource streamingWriteTcs; // Completion of a pending streaming write or send close from client if not null. + protected TaskCompletionSource delayedStreamingWriteTcs; protected TaskCompletionSource sendStatusFromServerTcs; protected bool readingDone; // True if last read (i.e. read with null payload) was already received. @@ -200,6 +201,12 @@ namespace Grpc.Core.Internal get; } + /// + /// Returns an exception to throw for a failed send operation. + /// It is only allowed to call this method for a call that has already finished. + /// + protected abstract Exception GetRpcExceptionClientOnly(); + private void ReleaseResources() { if (call != null) @@ -252,18 +259,39 @@ namespace Grpc.Core.Internal /// protected void HandleSendFinished(bool success) { + bool delayCompletion = false; TaskCompletionSource origTcs = null; lock (myLock) { origTcs = streamingWriteTcs; streamingWriteTcs = null; + if (!success && !finished && IsClient) + { + // We should be setting this only once per call, following writes will be short circuited. + GrpcPreconditions.CheckState (delayedStreamingWriteTcs == null); + delayedStreamingWriteTcs = origTcs; + delayCompletion = true; + } + ReleaseResourcesIfPossible(); } if (!success) { - origTcs.SetException(new InvalidOperationException("Send failed")); + if (!delayCompletion) + { + if (IsClient) + { + GrpcPreconditions.CheckState(finished); // implied by !success && !delayCompletion && IsClient + origTcs.SetException(GetRpcExceptionClientOnly()); + } + else + { + origTcs.SetException (new IOException("Error sending from server.")); + } + } + // if delayCompletion == true, postpone SetException until call finishes. } else { @@ -283,7 +311,7 @@ namespace Grpc.Core.Internal if (!success) { - sendStatusFromServerTcs.SetException(new InvalidOperationException("Error sending status from server.")); + sendStatusFromServerTcs.SetException(new IOException("Error sending status from server.")); } else { diff --git a/src/csharp/Grpc.Core/Internal/AsyncCallServer.cs b/src/csharp/Grpc.Core/Internal/AsyncCallServer.cs index 56c23ba3ef3..50fdfa9006a 100644 --- a/src/csharp/Grpc.Core/Internal/AsyncCallServer.cs +++ b/src/csharp/Grpc.Core/Internal/AsyncCallServer.cs @@ -33,6 +33,7 @@ using System; using System.Diagnostics; +using System.IO; using System.Runtime.CompilerServices; using System.Runtime.InteropServices; using System.Threading; @@ -193,6 +194,11 @@ namespace Grpc.Core.Internal get { return false; } } + protected override Exception GetRpcExceptionClientOnly() + { + throw new InvalidOperationException("Call be only called for client calls"); + } + protected override void OnAfterReleaseResources() { server.RemoveCallReference(this); From 7a73bec0e5cb11cf57b2f3ee77cee1249988ca4f Mon Sep 17 00:00:00 2001 From: Jan Tattermusch Date: Fri, 16 Sep 2016 16:55:17 +0200 Subject: [PATCH 2/3] dont allow new writes if theres a write with delayed completion --- .../Grpc.Core.Tests/Internal/AsyncCallTest.cs | 28 +++++++++++++++++++ src/csharp/Grpc.Core/Internal/AsyncCall.cs | 3 +- .../Grpc.Core/Internal/AsyncCallServer.cs | 2 +- 3 files changed, 31 insertions(+), 2 deletions(-) diff --git a/src/csharp/Grpc.Core.Tests/Internal/AsyncCallTest.cs b/src/csharp/Grpc.Core.Tests/Internal/AsyncCallTest.cs index 8ee4d184abc..616bc06d764 100644 --- a/src/csharp/Grpc.Core.Tests/Internal/AsyncCallTest.cs +++ b/src/csharp/Grpc.Core.Tests/Internal/AsyncCallTest.cs @@ -223,6 +223,34 @@ namespace Grpc.Core.Internal.Tests AssertUnaryResponseError(asyncCall, fakeCall, resultTask, StatusCode.Internal); } + [Test] + public void ClientStreaming_WriteFailureThrowsRpcException3() + { + var resultTask = asyncCall.ClientStreamingCallAsync(); + var requestStream = new ClientRequestStream(asyncCall); + + var writeTask = requestStream.WriteAsync("request1"); + fakeCall.SendCompletionHandler(false); + + // Until the delayed write completion has been triggered, + // we still act as if there was an active write. + Assert.Throws(typeof(InvalidOperationException), () => requestStream.WriteAsync("request2")); + + fakeCall.UnaryResponseClientHandler(true, + CreateClientSideStatus(StatusCode.Internal), + null, + new Metadata()); + + var ex = Assert.ThrowsAsync(async () => await writeTask); + Assert.AreEqual(StatusCode.Internal, ex.Status.StatusCode); + + // Following attempts to write keep delivering the same status + var ex2 = Assert.ThrowsAsync(async () => await requestStream.WriteAsync("after call has finished")); + Assert.AreEqual(StatusCode.Internal, ex2.Status.StatusCode); + + AssertUnaryResponseError(asyncCall, fakeCall, resultTask, StatusCode.Internal); + } + [Test] public void ClientStreaming_WriteAfterReceivingStatusThrowsRpcException() { diff --git a/src/csharp/Grpc.Core/Internal/AsyncCall.cs b/src/csharp/Grpc.Core/Internal/AsyncCall.cs index db40b553ce1..264c28ae7a7 100644 --- a/src/csharp/Grpc.Core/Internal/AsyncCall.cs +++ b/src/csharp/Grpc.Core/Internal/AsyncCall.cs @@ -372,7 +372,8 @@ namespace Grpc.Core.Internal private Task CheckSendPreconditionsClientSide() { GrpcPreconditions.CheckState(!halfcloseRequested, "Request stream has already been completed."); - GrpcPreconditions.CheckState(streamingWriteTcs == null, "Only one write can be pending at a time."); + // if there is a delayed streaming write, we will treat that as if the write was still in progress until the call finishes. + GrpcPreconditions.CheckState(streamingWriteTcs == null && (finished || delayedStreamingWriteTcs == null), "Only one write can be pending at a time."); if (cancelRequested) { diff --git a/src/csharp/Grpc.Core/Internal/AsyncCallServer.cs b/src/csharp/Grpc.Core/Internal/AsyncCallServer.cs index 50fdfa9006a..8d9f548d62e 100644 --- a/src/csharp/Grpc.Core/Internal/AsyncCallServer.cs +++ b/src/csharp/Grpc.Core/Internal/AsyncCallServer.cs @@ -208,7 +208,7 @@ namespace Grpc.Core.Internal { GrpcPreconditions.CheckState(!halfcloseRequested, "Response stream has already been completed."); GrpcPreconditions.CheckState(!finished, "Already finished."); - GrpcPreconditions.CheckState(streamingWriteTcs == null, "Only one write can be pending at a time"); + GrpcPreconditions.CheckState(streamingWriteTcs == null && delayedStreamingWriteTcs == null, "Only one write can be pending at a time"); GrpcPreconditions.CheckState(!disposed); return null; From 6eb987780a80d2ba83feafb3b0a98a9c60e0153a Mon Sep 17 00:00:00 2001 From: Jan Tattermusch Date: Fri, 16 Sep 2016 17:19:11 +0200 Subject: [PATCH 3/3] simplify delayed streaming write logic --- src/csharp/Grpc.Core/Internal/AsyncCall.cs | 28 ++++++++++++------- .../Grpc.Core/Internal/AsyncCallBase.cs | 20 +++++++------ .../Grpc.Core/Internal/AsyncCallServer.cs | 2 +- 3 files changed, 31 insertions(+), 19 deletions(-) diff --git a/src/csharp/Grpc.Core/Internal/AsyncCall.cs b/src/csharp/Grpc.Core/Internal/AsyncCall.cs index 264c28ae7a7..9abaf1120f0 100644 --- a/src/csharp/Grpc.Core/Internal/AsyncCall.cs +++ b/src/csharp/Grpc.Core/Internal/AsyncCall.cs @@ -372,8 +372,7 @@ namespace Grpc.Core.Internal private Task CheckSendPreconditionsClientSide() { GrpcPreconditions.CheckState(!halfcloseRequested, "Request stream has already been completed."); - // if there is a delayed streaming write, we will treat that as if the write was still in progress until the call finishes. - GrpcPreconditions.CheckState(streamingWriteTcs == null && (finished || delayedStreamingWriteTcs == null), "Only one write can be pending at a time."); + GrpcPreconditions.CheckState(streamingWriteTcs == null, "Only one write can be pending at a time."); if (cancelRequested) { @@ -458,7 +457,7 @@ namespace Grpc.Core.Internal using (Profilers.ForCurrentThread().NewScope("AsyncCall.HandleUnaryResponse")) { - TaskCompletionSource delayedTcs; + TaskCompletionSource delayedStreamingWriteTcs = null; TResponse msg = default(TResponse); var deserializeException = TryDeserialize(receivedMessage, out msg); @@ -471,16 +470,21 @@ namespace Grpc.Core.Internal receivedStatus = new ClientSideStatus(DeserializeResponseFailureStatus, receivedStatus.Trailers); } finishedStatus = receivedStatus; - delayedTcs = delayedStreamingWriteTcs; + + if (isStreamingWriteCompletionDelayed) + { + delayedStreamingWriteTcs = streamingWriteTcs; + streamingWriteTcs = null; + } ReleaseResourcesIfPossible(); } responseHeadersTcs.SetResult(responseHeaders); - if (delayedTcs != null) + if (delayedStreamingWriteTcs != null) { - delayedTcs.SetException(GetRpcExceptionClientOnly()); + delayedStreamingWriteTcs.SetException(GetRpcExceptionClientOnly()); } var status = receivedStatus.Status; @@ -502,20 +506,24 @@ namespace Grpc.Core.Internal // NOTE: because this event is a result of batch containing GRPC_OP_RECV_STATUS_ON_CLIENT, // success will be always set to true. - TaskCompletionSource delayedTcs; + TaskCompletionSource delayedStreamingWriteTcs = null; lock (myLock) { finished = true; finishedStatus = receivedStatus; - delayedTcs = delayedStreamingWriteTcs; + if (isStreamingWriteCompletionDelayed) + { + delayedStreamingWriteTcs = streamingWriteTcs; + streamingWriteTcs = null; + } ReleaseResourcesIfPossible(); } - if (delayedTcs != null) + if (delayedStreamingWriteTcs != null) { - delayedTcs.SetException(GetRpcExceptionClientOnly()); + delayedStreamingWriteTcs.SetException(GetRpcExceptionClientOnly()); } var status = receivedStatus.Status; diff --git a/src/csharp/Grpc.Core/Internal/AsyncCallBase.cs b/src/csharp/Grpc.Core/Internal/AsyncCallBase.cs index b27baba942f..9f9d260e7eb 100644 --- a/src/csharp/Grpc.Core/Internal/AsyncCallBase.cs +++ b/src/csharp/Grpc.Core/Internal/AsyncCallBase.cs @@ -68,8 +68,8 @@ namespace Grpc.Core.Internal protected TaskCompletionSource streamingReadTcs; // Completion of a pending streaming read if not null. protected TaskCompletionSource streamingWriteTcs; // Completion of a pending streaming write or send close from client if not null. - protected TaskCompletionSource delayedStreamingWriteTcs; protected TaskCompletionSource sendStatusFromServerTcs; + protected bool isStreamingWriteCompletionDelayed; // Only used for the client side. protected bool readingDone; // True if last read (i.e. read with null payload) was already received. protected bool halfcloseRequested; // True if send close have been initiated. @@ -263,16 +263,20 @@ namespace Grpc.Core.Internal TaskCompletionSource origTcs = null; lock (myLock) { - origTcs = streamingWriteTcs; - streamingWriteTcs = null; + if (!success && !finished && IsClient) { + // We should be setting this only once per call, following writes will be short circuited + // because they cannot start until the entire call finishes. + GrpcPreconditions.CheckState(!isStreamingWriteCompletionDelayed); - if (!success && !finished && IsClient) - { - // We should be setting this only once per call, following writes will be short circuited. - GrpcPreconditions.CheckState (delayedStreamingWriteTcs == null); - delayedStreamingWriteTcs = origTcs; + // leave streamingWriteTcs set, it will be completed once call finished. + isStreamingWriteCompletionDelayed = true; delayCompletion = true; } + else + { + origTcs = streamingWriteTcs; + streamingWriteTcs = null; + } ReleaseResourcesIfPossible(); } diff --git a/src/csharp/Grpc.Core/Internal/AsyncCallServer.cs b/src/csharp/Grpc.Core/Internal/AsyncCallServer.cs index 8d9f548d62e..50fdfa9006a 100644 --- a/src/csharp/Grpc.Core/Internal/AsyncCallServer.cs +++ b/src/csharp/Grpc.Core/Internal/AsyncCallServer.cs @@ -208,7 +208,7 @@ namespace Grpc.Core.Internal { GrpcPreconditions.CheckState(!halfcloseRequested, "Response stream has already been completed."); GrpcPreconditions.CheckState(!finished, "Already finished."); - GrpcPreconditions.CheckState(streamingWriteTcs == null && delayedStreamingWriteTcs == null, "Only one write can be pending at a time"); + GrpcPreconditions.CheckState(streamingWriteTcs == null, "Only one write can be pending at a time"); GrpcPreconditions.CheckState(!disposed); return null;