Use asynchronous data access methods in appMpower.Orm (#9281)

LLT21 · web-flow · commit d4b3ac80417a · 2024-09-30T10:54:17.000-07:00
* Use asynchronous data access methods in  appMpower.Orm

* Improved connection pool handling

* Using tuple with base connection elements instead of "this" for pooling

* Improved caching

* Better separation of keyed and unkeyed command caching

---------

Co-authored-by: LLT21 &lt;&gt;
diff --git a/frameworks/CSharp/appmpower/src/appMpower.Orm/Data/DbCommand.cs b/frameworks/CSharp/appmpower/src/appMpower.Orm/Data/DbCommand.cs
@@ -20,12 +20,6 @@ public DbCommand(string commandText, DbConnection dbConnection)
          _dbConnection = dbConnection;
       }
 
-      public DbCommand(string commandText, DbConnection dbConnection, bool keyed)
-      {
-         _odbcCommand = dbConnection.GetCommand(commandText, CommandType.Text, keyed);
-         _dbConnection = dbConnection;
-      }
-
       public DbCommand(string commandText, CommandType commandType, DbConnection dbConnection)
       {
          _odbcCommand = dbConnection.GetCommand(commandText, commandType);
@@ -175,14 +169,19 @@ public IDataReader ExecuteReader()
 
       public async Task<int> ExecuteNonQueryAsync()
       {
-         return await (_odbcCommand as System.Data.Common.DbCommand).ExecuteNonQueryAsync();
+         return await _odbcCommand.ExecuteNonQueryAsync();
       }
 
       public IDataReader ExecuteReader(CommandBehavior behavior)
       {
          return _odbcCommand.ExecuteReader(behavior);
       }
 
+      public async Task<System.Data.Common.DbDataReader> ExecuteReaderAsync(CommandBehavior behavior)
+      {
+         return await _odbcCommand.ExecuteReaderAsync(behavior);
+      }
+
 #nullable enable
       public object? ExecuteScalar()
       {
@@ -197,8 +196,7 @@ public void Prepare()
 
       public void Dispose()
       {
-         if (_dbConnection._keyed) _dbConnection._keyedOdbcCommands.TryAdd(_odbcCommand.CommandText, _odbcCommand);
-         else _dbConnection._odbcCommands.Push(_odbcCommand);
+         _dbConnection.Release(_odbcCommand);
       }
    }
 }
diff --git a/frameworks/CSharp/appmpower/src/appMpower.Orm/Data/DbConnection.cs b/frameworks/CSharp/appmpower/src/appMpower.Orm/Data/DbConnection.cs
@@ -1,26 +1,27 @@
 using System.Collections.Concurrent;
 using System.Data;
-using System.Data.Odbc; 
+using System.Data.Odbc;
 
 namespace appMpower.Orm.Data
 {
    public class DbConnection : IDbConnection
    {
       private string _connectionString;
-      internal bool _keyed = false; 
-      internal int _number; 
-      internal OdbcConnection _odbcConnection;
-      internal ConcurrentStack<OdbcCommand> _odbcCommands = new();
-      internal Dictionary<string, OdbcCommand> _keyedOdbcCommands;
+      private bool _keyed = false;
+      private int _number;
+      private OdbcConnection _odbcConnection;
+      private ConcurrentStack<OdbcCommand> _odbcCommands = new();
+      private Dictionary<string, OdbcCommand> _keyedOdbcCommands;
 
       public DbConnection()
       {
-         _connectionString = DbProviderFactory.ConnectionString;
       }
 
-      public DbConnection(string connectionString)
+      public DbConnection(string connectionString, bool keyed = false)
       {
-         _connectionString = connectionString;
+         _keyed = keyed;
+         _connectionString = connectionString; 
+         GetConnection();
       }
 
       public IDbConnection Connection
@@ -43,7 +44,22 @@ public string ConnectionString
          }
          set
          {
-            _odbcConnection.ConnectionString = value;
+            _connectionString = value; 
+            GetConnection();
+         }
+      }
+
+      private void GetConnection()
+      {
+         if (_keyed)
+         {
+            (_number, _odbcConnection, _keyedOdbcCommands) = 
+               DbConnectionsKeyed.GetConnectionBase(_connectionString).GetAwaiter().GetResult();
+         }
+         else
+         {
+            (_number, _odbcConnection, _odbcCommands) = 
+               DbConnections.GetConnectionBase(_connectionString).GetAwaiter().GetResult();
          }
       }
 
@@ -99,23 +115,33 @@ public IDbCommand CreateCommand()
 
       public void Open()
       {
-         if (_odbcConnection is null)
+         if (_odbcConnection.State == ConnectionState.Closed)
          {
-            DbConnections.GetConnection(_connectionString, this);
+            _odbcConnection.Open();
          }
+      }
 
+      public async Task OpenAsync()
+      {
          if (_odbcConnection.State == ConnectionState.Closed)
          {
-            _odbcConnection.Open();
+            await _odbcConnection.OpenAsync();
          }
       }
 
       public void Dispose()
       {
-         DbConnections.Release(this);
+         if (_keyed)
+         {
+            DbConnectionsKeyed.Release((Number: _number, OdbcConnection: _odbcConnection, KeyedOdbcCommands: _keyedOdbcCommands));
+         }
+         else
+         {
+            DbConnections.Release((Number: _number, OdbcConnection: _odbcConnection, OdbcCommands: _odbcCommands));
+         }
       }
 
-      internal OdbcCommand GetCommand(string commandText, CommandType commandType, bool keyed = false)
+      internal OdbcCommand GetCommand(string commandText, CommandType commandType)
       {
          OdbcCommand odbcCommand;
 
@@ -129,25 +155,20 @@ internal OdbcCommand GetCommand(string commandText, CommandType commandType, boo
 
             return odbcCommand; 
          }
-         else if (_keyed && _keyedOdbcCommands.TryGetValue(commandText, out odbcCommand))
-         {
-            return odbcCommand; 
-         }
-         else
-         {
-            if (!_keyed && keyed) 
-            {
-               _keyedOdbcCommands = new();
-               _keyed = keyed; 
-            }
-            
-            odbcCommand = _odbcConnection.CreateCommand();
-            odbcCommand.CommandText = commandText;
-            odbcCommand.CommandType = commandType;
-            odbcCommand.Prepare();
+         else if (_keyed && _keyedOdbcCommands.TryGetValue(commandText, out odbcCommand)) return odbcCommand; 
 
-            return odbcCommand;
-         }
+         odbcCommand = _odbcConnection.CreateCommand();
+         odbcCommand.CommandText = commandText;
+         odbcCommand.CommandType = commandType;
+         odbcCommand.Prepare();
+
+         return odbcCommand;
+      }
+
+      internal void Release(OdbcCommand odbcCommand)
+      {
+         if (_keyed) _keyedOdbcCommands.TryAdd(odbcCommand.CommandText, odbcCommand);
+         else _odbcCommands.Push(odbcCommand);
       }
    }
 }
diff --git a/frameworks/CSharp/appmpower/src/appMpower.Orm/Data/DbConnections.cs b/frameworks/CSharp/appmpower/src/appMpower.Orm/Data/DbConnections.cs
@@ -1,61 +1,61 @@
 using System.Collections.Concurrent;
+using System.Data.Odbc;
 
 namespace appMpower.Orm.Data
 {
-   public static class DbConnections
+   internal static class DbConnections
    {
+      private static bool _maxConnectionsCreated = false;
       private static short _createdConnections = 0;
-      private static ConcurrentStack<DbConnection> _connectionsStack = new();
+      private static short _maxConnections = 500;
 
-      public static DbConnection GetConnection(string connectionString)
+      private static ConcurrentStack<(int Number, OdbcConnection OdbcConnection, ConcurrentStack<OdbcCommand> OdbcCommands)> _connectionsStack = new();
+      private static ConcurrentQueue<TaskCompletionSource<(int Number, OdbcConnection OdbcConnection, ConcurrentStack<OdbcCommand> OdbcCommands)>> _waitingQueue = new();
+      
+      internal static async Task<(int Number, OdbcConnection OdbcConnection, ConcurrentStack<OdbcCommand> OdbcCommands)> GetConnectionBase(string connectionString)
       {
-         DbConnection popDbConnection; 
+         (int Number, OdbcConnection OdbcConnection, ConcurrentStack<OdbcCommand> OdbcCommands) dbConnectionBase;
 
-         if (!_connectionsStack.TryPop(out popDbConnection))
+         if (!_connectionsStack.TryPop(out dbConnectionBase))
          {
-            popDbConnection = new DbConnection();
-            popDbConnection._odbcConnection = new System.Data.Odbc.OdbcConnection(connectionString);
+            if (_maxConnectionsCreated)
+            {
+               dbConnectionBase = await GetDbConnectionBaseAsync();
+            }
+            else
+            {
+               _createdConnections++;
+               dbConnectionBase = (Number: _maxConnections, OdbcConnection: new OdbcConnection(connectionString), OdbcCommands: new ConcurrentStack<OdbcCommand>());
 
-            _createdConnections++;
-            popDbConnection._number = _createdConnections;
+               if (_createdConnections == _maxConnections) _maxConnectionsCreated = true;
 
-            if (_createdConnections % 25 == 0)
-            {
-               Console.WriteLine("Pooled connections created: " + _createdConnections.ToString());
+               //Console.WriteLine("opened connection number: " + dbConnectionBase._number);
             }
          }
 
-         return popDbConnection; 
+         return dbConnectionBase;
       }
 
-
-      public static void GetConnection(string connectionString, DbConnection dbConnection)
+      internal static void Release((int Number, OdbcConnection OdbcConnection, ConcurrentStack<OdbcCommand> OdbcCommands) dbConnectionBase)
       {
-         DbConnection popDbConnection = null;
+         TaskCompletionSource<(int Number, OdbcConnection OdbcConnection, ConcurrentStack<OdbcCommand> OdbcCommands)> taskCompletionSource;
 
-         if (_connectionsStack.TryPop(out popDbConnection))
+         if (_waitingQueue.TryDequeue(out taskCompletionSource))
          {
-            dbConnection._odbcConnection = popDbConnection._odbcConnection; 
-            dbConnection._odbcCommands = popDbConnection._odbcCommands;
-            dbConnection._number = popDbConnection._number; 
+            taskCompletionSource.SetResult(dbConnectionBase);
          }
          else
          {
-            dbConnection._odbcConnection = new System.Data.Odbc.OdbcConnection(connectionString);
-
-            _createdConnections++;
-            dbConnection._number = _createdConnections;
-
-           if (_createdConnections % 25 == 0)
-            {
-               Console.WriteLine("Pooled connections created: " + _createdConnections.ToString());
-            }
-          }
+            _connectionsStack.Push(dbConnectionBase);
+         }
       }
 
-      public static void Release(DbConnection dbConnection)
+      private static Task<(int Number, OdbcConnection OdbcConnection, ConcurrentStack<OdbcCommand> OdbcCommands)> GetDbConnectionBaseAsync()
       {
-         _connectionsStack.Push(dbConnection);
+         var taskCompletionSource = new TaskCompletionSource<(int Number, OdbcConnection OdbcConnection, ConcurrentStack<OdbcCommand> OdbcCommands)>(TaskCreationOptions.RunContinuationsAsynchronously);
+
+         _waitingQueue.Enqueue(taskCompletionSource);
+         return taskCompletionSource.Task;
       }
    }
 }
diff --git a/frameworks/CSharp/appmpower/src/appMpower.Orm/Data/DbConnectionsKeyed.cs b/frameworks/CSharp/appmpower/src/appMpower.Orm/Data/DbConnectionsKeyed.cs
@@ -0,0 +1,61 @@
+using System.Collections.Concurrent;
+using System.Data.Odbc;
+
+namespace appMpower.Orm.Data
+{
+   internal static class DbConnectionsKeyed
+   {
+      private static bool _maxConnectionsCreated = false;
+      private static short _createdConnections = 0;
+      private static short _maxConnections = 500;
+
+      private static ConcurrentStack<(int Number, OdbcConnection OdbcConnection, Dictionary<string, OdbcCommand>)> _connectionsStack = new();
+      private static ConcurrentQueue<TaskCompletionSource<(int Number, OdbcConnection OdbcConnection, Dictionary<string, OdbcCommand>)>> _waitingQueue = new();
+
+      internal static async Task<(int Number, OdbcConnection OdbcConnection, Dictionary<string, OdbcCommand> KeyedOdbcCommands)> GetConnectionBase(string connectionString)
+      {
+         (int Number, OdbcConnection OdbcConnection, Dictionary<string, OdbcCommand> KeyedOdbcCommands) dbConnectionBase;
+
+         if (!_connectionsStack.TryPop(out dbConnectionBase))
+         {
+            if (_maxConnectionsCreated)
+            {
+               dbConnectionBase = await GetDbConnectionBaseAsync();
+            }
+            else
+            {
+               _createdConnections++;
+               dbConnectionBase = (Number: _maxConnections, OdbcConnection: new OdbcConnection(connectionString), KeyedOdbcCommands: new Dictionary<string, OdbcCommand>());
+
+               if (_createdConnections == _maxConnections) _maxConnectionsCreated = true;
+
+               //Console.WriteLine("opened connection number: " + dbConnectionBase._number);
+            }
+         }
+
+         return dbConnectionBase;
+      }
+
+      internal static void Release((int Number, OdbcConnection OdbcConnection, Dictionary<string, OdbcCommand> KeyedOdbcCommands) dbConnectionBase)
+      {
+         TaskCompletionSource<(int Number, OdbcConnection OdbcConnection, Dictionary<string, OdbcCommand>)> taskCompletionSource;
+
+         if (_waitingQueue.TryDequeue(out taskCompletionSource))
+         {
+            taskCompletionSource.SetResult(dbConnectionBase);
+         }
+         else
+         {
+            _connectionsStack.Push(dbConnectionBase);
+         }
+      }
+
+      private static Task<(int Number, OdbcConnection OdbcConnection, Dictionary<string, OdbcCommand>)> GetDbConnectionBaseAsync()
+      {
+         var taskCompletionSource = new TaskCompletionSource<(int Number, OdbcConnection OdbcConnection, Dictionary<string, OdbcCommand>)>(TaskCreationOptions.RunContinuationsAsynchronously);
+
+         _waitingQueue.Enqueue(taskCompletionSource);
+         return taskCompletionSource.Task;
+      }
+   }
+}
diff --git a/frameworks/CSharp/appmpower/src/appMpower.Orm/DotnetMethods.cs b/frameworks/CSharp/appmpower/src/appMpower.Orm/DotnetMethods.cs
@@ -21,7 +21,7 @@ public static class DotnetMethods
 
     public static byte[] Db()
     {
-        var world = RawDb.LoadSingleQueryRow();
+        var world = RawDb.LoadSingleQueryRow().GetAwaiter().GetResult();
 
         var memoryStream = new MemoryStream();
         using var utf8JsonWriter = new Utf8JsonWriter(memoryStream, _jsonWriterOptions);
@@ -33,7 +33,7 @@ public static byte[] Db()
 
     public static byte[] Query(int queries)
     {
-        World[] worlds = RawDb.ReadMultipleRows(queries);
+        World[] worlds = RawDb.ReadMultipleRows(queries).GetAwaiter().GetResult();
 
         var memoryStream = new MemoryStream();
         using var utf8JsonWriter = new Utf8JsonWriter(memoryStream, _jsonWriterOptions);
@@ -45,7 +45,7 @@ public static byte[] Query(int queries)
 
     public static byte[] Updates(int count)
     {
-        World[] worlds = RawDb.LoadMultipleUpdatesRows(count);
+        World[] worlds = RawDb.LoadMultipleUpdatesRows(count).GetAwaiter().GetResult();
 
         var memoryStream = new MemoryStream();
         using var utf8JsonWriter = new Utf8JsonWriter(memoryStream, _jsonWriterOptions);
@@ -57,7 +57,7 @@ public static byte[] Updates(int count)
 
     public static byte[] Fortunes()
     {
-        List<Fortune> fortunes = RawDb.LoadFortunesRows(); 
+        List<Fortune> fortunes = RawDb.LoadFortunesRows().GetAwaiter().GetResult(); 
         string fortunesView = FortunesView.Render(fortunes);
         byte[] byteArray = Encoding.UTF8.GetBytes(fortunesView);
 
diff --git a/frameworks/CSharp/appmpower/src/appMpower.Orm/NativeMethods.cs b/frameworks/CSharp/appmpower/src/appMpower.Orm/NativeMethods.cs
diff --git a/frameworks/CSharp/appmpower/src/appMpower.Orm/RawDb.cs b/frameworks/CSharp/appmpower/src/appMpower.Orm/RawDb.cs

Original file line number	Diff line number	Diff line change
`@@ -20,12 +20,6 @@ public DbCommand(string commandText, DbConnection dbConnection)`
`20`	`20`	`_dbConnection = dbConnection;`
`21`	`21`	`}`
`22`	`22`
`23`		`- public DbCommand(string commandText, DbConnection dbConnection, bool keyed)`
`24`		`- {`
`25`		`- _odbcCommand = dbConnection.GetCommand(commandText, CommandType.Text, keyed);`
`26`		`- _dbConnection = dbConnection;`
`27`		`- }`
`28`		`-`
`29`	`23`	`public DbCommand(string commandText, CommandType commandType, DbConnection dbConnection)`
`30`	`24`	`{`
`31`	`25`	`_odbcCommand = dbConnection.GetCommand(commandText, commandType);`
`@@ -175,14 +169,19 @@ public IDataReader ExecuteReader()`
`175`	`169`
`176`	`170`	`public async Task<int> ExecuteNonQueryAsync()`
`177`	`171`	`{`
`178`		`- return await (_odbcCommand as System.Data.Common.DbCommand).ExecuteNonQueryAsync();`
	`172`	`+ return await _odbcCommand.ExecuteNonQueryAsync();`
`179`	`173`	`}`
`180`	`174`
`181`	`175`	`public IDataReader ExecuteReader(CommandBehavior behavior)`
`182`	`176`	`{`
`183`	`177`	`return _odbcCommand.ExecuteReader(behavior);`
`184`	`178`	`}`
`185`	`179`
	`180`	`+ public async Task<System.Data.Common.DbDataReader> ExecuteReaderAsync(CommandBehavior behavior)`
	`181`	`+ {`
	`182`	`+ return await _odbcCommand.ExecuteReaderAsync(behavior);`
	`183`	`+ }`
	`184`	`+`
`186`	`185`	`#nullable enable`
`187`	`186`	`public object? ExecuteScalar()`
`188`	`187`	`{`
`@@ -197,8 +196,7 @@ public void Prepare()`
`197`	`196`
`198`	`197`	`public void Dispose()`
`199`	`198`	`{`
`200`		`- if (_dbConnection._keyed) _dbConnection._keyedOdbcCommands.TryAdd(_odbcCommand.CommandText, _odbcCommand);`
`201`		`- else _dbConnection._odbcCommands.Push(_odbcCommand);`
	`199`	`+ _dbConnection.Release(_odbcCommand);`
`202`	`200`	`}`
`203`	`201`	`}`
`204`	`202`	`}`