Functional tests of parallel streaming and cancellation (#473)

Author: JamesNK

test/FunctionalTests/Client/CancellationTests.cs

@@ -0,0 +1,124 @@
+#region Copyright notice and license
+
+// Copyright 2019 The gRPC Authors
+//
+// Licensed 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.
+
+#endregion
+
+using System;
+using System.Diagnostics;
+using System.IO;
+using System.Threading;
+using System.Threading.Tasks;
+using Google.Protobuf;
+using Grpc.Core;
+using Grpc.Tests.Shared;
+using NUnit.Framework;
+using Streaming;
+
+namespace Grpc.AspNetCore.FunctionalTests.Client
+{
+    [TestFixture]
+    public class CancellationTests : FunctionalTestBase
+    {
+        [TestCase(1)]
+        [TestCase(5)]
+        [TestCase(20)]
+        public async Task DuplexStreaming_CancelAfterHeadersInParallel_Success(int tasks)
+        {
+            await CancelInParallel(tasks, waitForHeaders: true, interations: 10);
+        }
+
+        [TestCase(1)]
+        [TestCase(5)]
+        [TestCase(20)]
+        public async Task DuplexStreaming_CancelWithoutHeadersInParallel_Success(int tasks)
+        {
+            await CancelInParallel(tasks, waitForHeaders: false, interations: 10);
+        }
+
+        private async Task CancelInParallel(int tasks, bool waitForHeaders, int interations)
+        {
+            SetExpectedErrorsFilter(writeContext =>
+            {
+                if (writeContext.LoggerName == "SERVER FunctionalTestsWebsite.Services.StreamService")
+                {
+                    // Kestrel cancellation error message
+                    if (writeContext.Exception is IOException &&
+                        writeContext.Exception.Message == "The client reset the request stream.")
+                    {
+                        return true;
+                    }
+
+                    // Cancellation when service is receiving message
+                    if (writeContext.Exception is InvalidOperationException &&
+                        writeContext.Exception.Message == "Cannot write message after request is complete.")
+                    {
+                        return true;
+                    }
+
+                    // Cancellation before service writes message
+                    if (writeContext.Exception is TaskCanceledException &&
+                        writeContext.Exception.Message == "A task was canceled.")
+                    {
+                        return true;
+                    }
+                }
+
+                if (writeContext.LoggerName == "Grpc.Net.Client.Internal.GrpcCall")
+                {
+                    // Cancellation when call hasn't returned headers
+                    if (writeContext.EventId.Name == "ErrorStartingCall" &&
+                        writeContext.Exception is TaskCanceledException)
+                    {
+                        return true;
+                    }
+                }
+
+                return false;
+            });
+
+            // Arrange
+            var data = new byte[1024 * 64];
+
+            var client = new StreamService.StreamServiceClient(Channel);
+
+            await TestHelpers.RunParallel(tasks, async () =>
+            {
+                for (int i = 0; i < interations; i++)
+                {
+                    var cts = new CancellationTokenSource();
+                    var headers = new Metadata();
+                    if (waitForHeaders)
+                    {
+                        headers.Add("flush-headers", bool.TrueString);
+                    }
+                    var call = client.EchoAllData(cancellationToken: cts.Token, headers: headers);
+
+                    if (waitForHeaders)
+                    {
+                        await call.ResponseHeadersAsync.DefaultTimeout();
+                    }
+
+                    await call.RequestStream.WriteAsync(new DataMessage
+                    {
+                        Data = ByteString.CopyFrom(data)
+                    }).DefaultTimeout();
+
+                    cts.Cancel();
+                }
+            });
+        }
+    }
+}

test/FunctionalTests/Client/StreamingTests.cs

@@ -22,6 +22,7 @@
 using System.Threading;
 using System.Threading.Tasks;
 using Google.Protobuf;
+using Grpc.AspNetCore.FunctionalTests.Infrastructure;
 using Grpc.Core;
 using Grpc.Tests.Shared;
 using Microsoft.Extensions.Logging;
@@ -47,7 +48,7 @@ public async Task DuplexStream_SendLargeFileBatchedAndRecieveLargeFileBatched_Su
             var client = new StreamService.StreamServiceClient(Channel);
 
             // Act
-            var call = client.DuplexData();
+            var call = client.BufferAllData();
 
             var sent = 0;
             while (sent < data.Length)
@@ -81,7 +82,7 @@ await call.RequestStream.WriteAsync(new DataMessage
         }
 
         [Test]
-        public async Task ClientStream_SendLargeFileBatchedAndRecieveLargeFileBatched_Success()
+        public async Task ClientStream_SendLargeFileBatched_Success()
         {
             // Arrange
             var total = 1024 * 1024 * 64; // 64 MB
@@ -238,5 +239,88 @@ public async Task DuplexStream_SendToUnimplementedMethodAfterResponseReceived_Mo
                 Assert.AreEqual(StatusCode.Unimplemented, status.StatusCode);
             }
         }
+
+        const int Size64MB = 1024 * 1024 * 64;
+        const int Size1MB = 1024 * 1024 * 1;
+        const int Size64KB = 1024 * 64;
+
+        [TestCase(0, 0)]
+        [TestCase(1, 1)]
+        [TestCase(2, 1)]
+        [TestCase(3, 2)]
+        [TestCase(Size64MB, Size64KB)]
+        [TestCase(Size64MB, Size1MB)]
+        public async Task DuplexStreaming_SimultaniousSendAndReceive_Success(int total, int batchSize)
+        {
+            // Arrange
+            var data = new byte[batchSize];
+
+            var client = new StreamService.StreamServiceClient(Channel);
+
+            var (sent, received) = await EchoData(total, data, client);
+
+            // Assert
+            Assert.AreEqual(sent, total);
+            Assert.AreEqual(received, total);
+        }
+
+        private async Task<(int sent, int received)> EchoData(int total, byte[] data, StreamService.StreamServiceClient client)
+        {
+            var sent = 0;
+            var received = 0;
+            var call = client.EchoAllData();
+
+            var readTask = Task.Run(async () =>
+            {
+                await foreach (var message in call.ResponseStream.ReadAllAsync())
+                {
+                    received += message.Data.Length;
+
+                    Logger.LogInformation($"Received {sent} bytes");
+                }
+            });
+
+            while (sent < total)
+            {
+                var writeCount = Math.Min(total - sent, data.Length);
+
+                await call.RequestStream.WriteAsync(new DataMessage
+                {
+                    Data = ByteString.CopyFrom(data, 0, writeCount)
+                }).DefaultTimeout();
+
+                sent += writeCount;
+
+                Logger.LogInformation($"Sent {sent} bytes");
+            }
+
+            await call.RequestStream.CompleteAsync().DefaultTimeout();
+            await readTask;
+
+            return (sent, received);
+        }
+
+        [TestCase(1)]
+        [TestCase(5)]
+        [TestCase(20)]
+        public async Task DuplexStreaming_SimultaniousSendAndReceiveInParallel_Success(int tasks)
+        {
+            // Arrange
+            const int total = 1024 * 1024 * 1;
+            const int batchSize = 1024 * 64;
+
+            var data = new byte[batchSize];
+
+            var client = new StreamService.StreamServiceClient(Channel);
+
+            await TestHelpers.RunParallel(tasks, async () =>
+            {
+                var (sent, received) = await EchoData(total, data, client);
+
+                // Assert
+                Assert.AreEqual(sent, total);
+                Assert.AreEqual(received, total);
+            });
+        }
     }
 }

test/Shared/TestHelpers.cs

@@ -50,5 +50,16 @@ public static async Task AssertIsTrueRetryAsync(Func<bool> assert, string messag
 
             throw new Exception($"Assert failed after {Retrys} retries: {message}");
         }
+
+        public static Task RunParallel(int count, Func<Task> action)
+        {
+            var actionTasks = new Task[count];
+            for (int i = 0; i < actionTasks.Length; i++)
+            {
+                actionTasks[i] = action();
+            }
+
+            return Task.WhenAll(actionTasks);
+        }
     }
 }

testassets/FunctionalTestsWebsite/Services/StreamService.cs

@@ -18,6 +18,7 @@
 
 using System;
 using System.IO;
+using System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
 using Google.Protobuf;
@@ -36,7 +37,7 @@ public StreamService(ILoggerFactory loggerFactory)
             _logger = loggerFactory.CreateLogger<StreamService>();
         }
 
-        public override async Task DuplexData(
+        public override async Task BufferAllData(
             IAsyncStreamReader<DataMessage> requestStream,
             IServerStreamWriter<DataMessage> responseStream,
             ServerCallContext context)
@@ -67,6 +68,27 @@ await responseStream.WriteAsync(new DataMessage
             }
         }
 
+
+        public override async Task EchoAllData(
+            IAsyncStreamReader<DataMessage> requestStream,
+            IServerStreamWriter<DataMessage> responseStream,
+            ServerCallContext context)
+        {
+            var flushHeaders = context.RequestHeaders.Any(x => x.Key == "flush-headers");
+            if (flushHeaders)
+            {
+                await context.WriteResponseHeadersAsync(new Metadata());
+            }
+
+            await foreach (var message in requestStream.ReadAllAsync())
+            {
+                await responseStream.WriteAsync(new DataMessage
+                {
+                    Data = message.Data
+                });
+            }
+        }
+
         public override async Task<DataComplete> ClientStreamedData(
             IAsyncStreamReader<DataMessage> requestStream,
             ServerCallContext context)

testassets/Proto/streaming.proto

@@ -17,7 +17,8 @@ syntax = "proto3";
 package Streaming;
 
 service StreamService {
-  rpc DuplexData (stream DataMessage) returns (stream DataMessage);
+  rpc BufferAllData (stream DataMessage) returns (stream DataMessage);
+  rpc EchoAllData (stream DataMessage) returns (stream DataMessage);
   rpc ClientStreamedData (stream DataMessage) returns (DataComplete);
 }