aboutsummaryrefslogtreecommitdiffhomepage
path: root/src/csharp/Grpc.Core/Internal
diff options
context:
space:
mode:
Diffstat (limited to 'src/csharp/Grpc.Core/Internal')
-rw-r--r--src/csharp/Grpc.Core/Internal/AsyncCall.cs31
-rw-r--r--src/csharp/Grpc.Core/Internal/AsyncCallBase.cs40
-rw-r--r--src/csharp/Grpc.Core/Internal/AsyncCallServer.cs6
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);