diff options
author | 2016-09-21 14:10:47 -0700 | |
---|---|---|
committer | 2016-09-21 14:10:47 -0700 | |
commit | 24129f010cb2b9ebccb44b4e0486d89aff7a74ae (patch) | |
tree | 6fe3d15f4b12ad79f96fa022aab6adccf0c16fea /src/csharp/Grpc.Core/Internal | |
parent | 2579bf394175637e44e9696959deac9d3c867c8e (diff) | |
parent | d53957cf4e3924d2abd201fa60ba1e2ae9ba2526 (diff) |
Merge remote-tracking branch 'upstream/master' into cli_list
Diffstat (limited to 'src/csharp/Grpc.Core/Internal')
-rw-r--r-- | src/csharp/Grpc.Core/Internal/AsyncCall.cs | 31 | ||||
-rw-r--r-- | src/csharp/Grpc.Core/Internal/AsyncCallBase.cs | 40 | ||||
-rw-r--r-- | src/csharp/Grpc.Core/Internal/AsyncCallServer.cs | 6 |
3 files changed, 71 insertions, 6 deletions
diff --git a/src/csharp/Grpc.Core/Internal/AsyncCall.cs b/src/csharp/Grpc.Core/Internal/AsyncCall.cs index f549c52876..9abaf1120f 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<object> delayedStreamingWriteTcs = null; TResponse msg = default(TResponse); var deserializeException = TryDeserialize(receivedMessage, out msg); @@ -465,13 +471,23 @@ namespace Grpc.Core.Internal } finishedStatus = receivedStatus; + if (isStreamingWriteCompletionDelayed) + { + delayedStreamingWriteTcs = streamingWriteTcs; + streamingWriteTcs = null; + } + ReleaseResourcesIfPossible(); } responseHeadersTcs.SetResult(responseHeaders); - var status = receivedStatus.Status; + if (delayedStreamingWriteTcs != null) + { + delayedStreamingWriteTcs.SetException(GetRpcExceptionClientOnly()); + } + var status = receivedStatus.Status; if (status.StatusCode != StatusCode.OK) { unaryResponseTcs.SetException(new RpcException(status)); @@ -490,16 +506,27 @@ 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<object> delayedStreamingWriteTcs = null; + lock (myLock) { finished = true; finishedStatus = receivedStatus; + if (isStreamingWriteCompletionDelayed) + { + delayedStreamingWriteTcs = streamingWriteTcs; + streamingWriteTcs = null; + } ReleaseResourcesIfPossible(); } - var status = receivedStatus.Status; + if (delayedStreamingWriteTcs != null) + { + delayedStreamingWriteTcs.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 eb9c3ea62d..9f9d260e7e 100644 --- a/src/csharp/Grpc.Core/Internal/AsyncCallBase.cs +++ b/src/csharp/Grpc.Core/Internal/AsyncCallBase.cs @@ -69,6 +69,7 @@ namespace Grpc.Core.Internal protected TaskCompletionSource<TRead> streamingReadTcs; // Completion of a pending streaming read if not null. protected TaskCompletionSource<object> streamingWriteTcs; // Completion of a pending streaming write or send close from client if not null. protected TaskCompletionSource<object> 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. @@ -200,6 +201,12 @@ namespace Grpc.Core.Internal get; } + /// <summary> + /// 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. + /// </summary> + protected abstract Exception GetRpcExceptionClientOnly(); + private void ReleaseResources() { if (call != null) @@ -252,18 +259,43 @@ namespace Grpc.Core.Internal /// </summary> protected void HandleSendFinished(bool success) { + bool delayCompletion = false; TaskCompletionSource<object> 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); + + // leave streamingWriteTcs set, it will be completed once call finished. + isStreamingWriteCompletionDelayed = true; + delayCompletion = true; + } + else + { + origTcs = streamingWriteTcs; + streamingWriteTcs = null; + } 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 +315,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 56c23ba3ef..50fdfa9006 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); |