Tests | Fix Tests regarding Concurrency both Sync and Async versions. (#2279)

Author: H-Yeo

src/Microsoft.Data.SqlClient/tests/ManualTests/SQL/RetryLogic/SqlCommandReliabilityTest.cs

@@ -3,6 +3,7 @@
 // See the LICENSE file in the project root for more information.
 
 using System;
+using System.Collections.Generic;
 using System.Data;
 using System.Threading;
 using System.Threading.Tasks;
@@ -11,10 +12,37 @@
 
 namespace Microsoft.Data.SqlClient.ManualTesting.Tests
 {
-    public class SqlCommandReliabilityTest
+    public class SqlCommandReliabilityTest : IDisposable
     {
+        private const int ConcurrentParallelExecutions = 3;
+        private const string UnexpectedSqlConnectionExceptionMessage = "An unexpected SQL Connection Error occured and is not part of the test.";
         private readonly string _exceedErrMsgPattern = RetryLogicTestHelper.s_exceedErrMsgPattern;
         private readonly string _cancelErrMsgPattern = RetryLogicTestHelper.s_cancelErrMsgPattern;
+        private readonly int _originalWorkerThreads;
+        private readonly int _originalCompletionPortThreads;
+        private bool _disposed = false;
+
+        public SqlCommandReliabilityTest()
+        {
+            ThreadPool.GetMinThreads(out _originalWorkerThreads, out _originalCompletionPortThreads);
+            ThreadPool.SetMinThreads(ConcurrentParallelExecutions * 2, ConcurrentParallelExecutions * 2); // Multiply by 2 for possible additional threads needed in functions
+        }
+
+        // Dispose Pattern
+        public void Dispose()
+        {
+            Dispose(true);
+            GC.SuppressFinalize(this);
+        }
+
+        protected virtual void Dispose(bool disposing)
+        {
+            if (!_disposed)
+            {
+                ThreadPool.SetMinThreads(_originalWorkerThreads, _originalCompletionPortThreads);
+                _disposed = true;
+            }
+        }
 
         #region Sync
         [ConditionalTheory(typeof(DataTestUtility), nameof(DataTestUtility.AreConnStringsSetup))]
@@ -524,160 +552,119 @@ public async void RetryExecuteAsyncCancel(string cnnString, SqlRetryLogicBasePro
         [MemberData(nameof(RetryLogicTestHelper.GetConnectionAndRetryStrategyInvalidCommand), parameters: new object[] { 2 }, MemberType = typeof(RetryLogicTestHelper), DisableDiscoveryEnumeration = true)]
         public void ConcurrentExecution(string cnnString, SqlRetryLogicBaseProvider provider)
         {
-            int numberOfTries = provider.RetryLogic.NumberOfTries;
             string query = "SELECT bad command";
-            int retriesCount = 0;
-            int concurrentExecution = 3;
-            provider.Retrying += (s, e) => Interlocked.Increment(ref retriesCount);
+            ProcessDataInParallel(cnnString, provider, query, cmd => cmd.ExecuteScalar());
+            ProcessDataInParallel(cnnString, provider, query, cmd => cmd.ExecuteNonQuery());
+            ProcessDataInParallel(cnnString, provider, query, cmd => cmd.ExecuteReader());
+            ProcessDataInParallel(cnnString, provider, query, cmd => cmd.ExecuteXmlReader());
 
-            Parallel.For(0, concurrentExecution,
-            i =>
+            if (DataTestUtility.IsNotAzureSynapse())
             {
-                using (SqlConnection cnn = new SqlConnection(cnnString))
-                using (SqlCommand cmd = cnn.CreateCommand())
-                {
-                    cnn.Open();
-                    cmd.RetryLogicProvider = provider;
-                    cmd.CommandText = query;
-                    Assert.Throws<AggregateException>(() => cmd.ExecuteScalar());
-                }
-            });
-            Assert.Equal(numberOfTries * concurrentExecution, retriesCount + concurrentExecution);
+                query += " FOR XML AUTO";
+                ProcessDataInParallel(cnnString, provider, query, cmd => cmd.ExecuteXmlReader());
+            }
+        }
 
-            retriesCount = 0;
-            Parallel.For(0, concurrentExecution,
-            i =>
+        [ConditionalTheory(typeof(DataTestUtility), nameof(DataTestUtility.AreConnStringsSetup))]
+        [MemberData(nameof(RetryLogicTestHelper.GetConnectionAndRetryStrategyInvalidCommand), parameters: new object[] { 2 }, MemberType = typeof(RetryLogicTestHelper), DisableDiscoveryEnumeration = true)]
+        public async void ConcurrentExecutionAsync(string cnnString, SqlRetryLogicBaseProvider provider)
+        {
+            string query = "SELECT bad command";
+            await ProcessDataAsAsync(cnnString, provider, query, cmd => cmd.ExecuteScalarAsync());
+            await ProcessDataAsAsync(cnnString, provider, query, cmd => cmd.ExecuteNonQueryAsync());
+            await ProcessDataAsAsync(cnnString, provider, query, cmd => cmd.ExecuteReaderAsync());
+            await ProcessDataAsAsync(cnnString, provider, query, cmd => cmd.ExecuteXmlReaderAsync());
+
+            if (DataTestUtility.IsNotAzureSynapse())
             {
-                using (SqlConnection cnn = new SqlConnection(cnnString))
-                using (SqlCommand cmd = cnn.CreateCommand())
+                query += " FOR XML AUTO";
+                await ProcessDataAsAsync(cnnString, provider, query, cmd => cmd.ExecuteXmlReaderAsync());
+            }
+        }
+        #endregion
+
+        #region private members
+        private SqlCommand CreateCommand(SqlConnection cnn, SqlRetryLogicBaseProvider provider, int cancelAfterRetries)
+        {
+            cnn.Open();
+            SqlCommand cmd = cnn.CreateCommand();
+            cmd.RetryLogicProvider = provider;
+            cmd.RetryLogicProvider.Retrying += (object s, SqlRetryingEventArgs e) =>
+            {
+                Assert.Equal(e.RetryCount, e.Exceptions.Count);
+                Assert.NotEqual(TimeSpan.Zero, e.Delay);
+
+                if (e.RetryCount >= cancelAfterRetries)
                 {
-                    cnn.Open();
-                    cmd.RetryLogicProvider = provider;
-                    cmd.CommandText = query;
-                    Assert.Throws<AggregateException>(() => cmd.ExecuteNonQuery());
+                    e.Cancel = true;
                 }
-            });
-            Assert.Equal(numberOfTries * concurrentExecution, retriesCount + concurrentExecution);
+            };
+            return cmd;
+        }
 
-            retriesCount = 0;
-            Parallel.For(0, concurrentExecution,
+        private static void ProcessDataInParallel(string cnnString, SqlRetryLogicBaseProvider provider,
+                                                  string query, Action<SqlCommand> commandAction)
+        {
+            int numberOfTries = provider.RetryLogic.NumberOfTries;
+            int retryExceptionCount = 0;
+
+            Parallel.For(0, ConcurrentParallelExecutions,
             i =>
             {
                 using (SqlConnection cnn = new SqlConnection(cnnString))
-                using (SqlCommand cmd = cnn.CreateCommand())
                 {
                     cnn.Open();
-                    cmd.RetryLogicProvider = provider;
-                    cmd.CommandText = query;
-                    Assert.Throws<AggregateException>(() => cmd.ExecuteReader());
-                }
-            });
-            Assert.Equal(numberOfTries * concurrentExecution, retriesCount + concurrentExecution);
+                    Assert.True(cnn.State == ConnectionState.Open, UnexpectedSqlConnectionExceptionMessage);
 
-            if(DataTestUtility.IsNotAzureSynapse())
-            {
-                retriesCount = 0;
-                Parallel.For(0, concurrentExecution,
-                i =>
-                {
-                    using (SqlConnection cnn = new SqlConnection(cnnString))
                     using (SqlCommand cmd = cnn.CreateCommand())
                     {
-                        cnn.Open();
                         cmd.RetryLogicProvider = provider;
-                        cmd.CommandText = query + " FOR XML AUTO";
-                        Assert.Throws<AggregateException>(() => cmd.ExecuteXmlReader());
+                        cmd.CommandText = query;
+
+                        AggregateException retryAggregateException = Assert.Throws<AggregateException>(() => commandAction(cmd));
+                        Interlocked.Add(ref retryExceptionCount, retryAggregateException.InnerExceptions?.Count ?? 0);
                     }
-                });
-                Assert.Equal(numberOfTries * concurrentExecution, retriesCount + concurrentExecution);
-            }
-            
-            retriesCount = 0;
-            Parallel.For(0, concurrentExecution,
-            i =>
-            {
-                using (SqlConnection cnn = new SqlConnection(cnnString))
-                using (SqlCommand cmd = cnn.CreateCommand())
-                {
-                    cnn.Open();
-                    cmd.RetryLogicProvider = provider;
-                    cmd.CommandText = query;
-                    Assert.ThrowsAsync<AggregateException>(() => cmd.ExecuteScalarAsync()).Wait();
                 }
             });
-            Assert.Equal(numberOfTries * concurrentExecution, retriesCount + concurrentExecution);
 
-            retriesCount = 0;
-            Parallel.For(0, concurrentExecution,
-            i =>
-            {
-                using (SqlConnection cnn = new SqlConnection(cnnString))
-                using (SqlCommand cmd = cnn.CreateCommand())
-                {
-                    cnn.Open();
-                    cmd.RetryLogicProvider = provider;
-                    cmd.CommandText = query;
-                    Assert.ThrowsAsync<AggregateException>(() => cmd.ExecuteNonQueryAsync()).Wait();
-                }
-            });
-            Assert.Equal(numberOfTries * concurrentExecution, retriesCount + concurrentExecution);
+            // Assertion for Retries
+            int expectedTotalNumberOfRetries = numberOfTries * ConcurrentParallelExecutions;
+            Assert.Equal(expectedTotalNumberOfRetries, retryExceptionCount);
+        }
 
-            retriesCount = 0;
-            Parallel.For(0, concurrentExecution,
-            i =>
-            {
-                using (SqlConnection cnn = new SqlConnection(cnnString))
-                using (SqlCommand cmd = cnn.CreateCommand())
-                {
-                    cnn.Open();
-                    cmd.RetryLogicProvider = provider;
-                    cmd.CommandText = query;
-                    Assert.ThrowsAsync<AggregateException>(() => cmd.ExecuteReaderAsync()).Wait();
-                }
-            });
-            Assert.Equal(numberOfTries * concurrentExecution, retriesCount + concurrentExecution);
+        private static async Task ProcessDataAsAsync(string cnnString, SqlRetryLogicBaseProvider provider,
+                                                     string query, Func<SqlCommand, Task> commandAction)
+        {
+            int numberOfTries = provider.RetryLogic.NumberOfTries;
+            int retryExceptionCount = 0;
 
-            // TODO: there is a known issue by ExecuteXmlReaderAsync that should be solved first- issue #44
-            /*
-            
-            if(DataTestUtility.IsNotAzureSynapse())
+            List<Task> tasks = new List<Task>();
+            for (int i = 0; i < ConcurrentParallelExecutions; i++)
             {
-                retriesCount = 0;
-                Parallel.For(0, concurrentExecution,
-                i =>
+                tasks.Add(Task.Run(async () =>
                 {
                     using (SqlConnection cnn = new SqlConnection(cnnString))
-                    using (SqlCommand cmd = cnn.CreateCommand())
                     {
-                        cnn.Open();
-                        cmd.RetryLogicProvider = provider;
-                        cmd.CommandText = query + " FOR XML AUTO";
-                        Assert.ThrowsAsync<AggregateException>(() => cmd.ExecuteXmlReaderAsync()).Wait();
+                        await cnn.OpenAsync();
+                        Assert.True(cnn.State == ConnectionState.Open, UnexpectedSqlConnectionExceptionMessage);
+
+                        using (SqlCommand cmd = cnn.CreateCommand())
+                        {
+                            cmd.RetryLogicProvider = provider;
+                            cmd.CommandText = query;
+
+                            AggregateException retryAggregateException = await Assert.ThrowsAsync<AggregateException>(async () => await commandAction(cmd));
+                            Interlocked.Add(ref retryExceptionCount, retryAggregateException.InnerExceptions?.Count ?? 0);
+                        }
                     }
-                });
-                Assert.Equal(numberOfTries * concurrentExecution, retriesCount + concurrentExecution);
+                }));
             }
-            */
-        }
-        #endregion
 
-        #region private members
-        private SqlCommand CreateCommand(SqlConnection cnn, SqlRetryLogicBaseProvider provider, int cancelAfterRetries)
-        {
-            cnn.Open();
-            SqlCommand cmd = cnn.CreateCommand();
-            cmd.RetryLogicProvider = provider;
-            cmd.RetryLogicProvider.Retrying += (object s, SqlRetryingEventArgs e) =>
-            {
-                Assert.Equal(e.RetryCount, e.Exceptions.Count);
-                Assert.NotEqual(TimeSpan.Zero, e.Delay);
+            await Task.WhenAll(tasks);
 
-                if (e.RetryCount >= cancelAfterRetries)
-                {
-                    e.Cancel = true;
-                }
-            };
-            return cmd;
+            // Assertion for Retries
+            int expectedTotalNumberOfRetries = numberOfTries * ConcurrentParallelExecutions;
+            Assert.Equal(expectedTotalNumberOfRetries, retryExceptionCount);
         }
         #endregion
     }