调整线程池处理

maikebing · maikebing · commit 7961a24c675a · 2022-08-11T15:29:46.000+08:00
diff --git a/src/Example/Program.cs b/src/Example/Program.cs
@@ -85,19 +85,27 @@ static void Main(string[] args)
                 connection.CreateCommand("CREATE TABLE IF NOT EXISTS telemetrydata  (ts timestamp,value_type  tinyint, value_boolean bool, value_string binary(10240), value_long bigint,value_datetime timestamp,value_double double)   TAGS (deviceid binary(32),keyname binary(64));").ExecuteNonQuery();
                 var devid1 = $"{Guid.NewGuid():N}";
                 var devid2 = $"{Guid.NewGuid():N}";
+                var devid3 = $"{Guid.NewGuid():N}";
+                var devid4 = $"{Guid.NewGuid():N}";
                 DateTime dt = DateTime.Now;
-                UploadTelemetryData(connection, devid1, "1#air-compressor-two-level-discharge-temperature", 4000);
-                UploadTelemetryData(connection, devid2, "1#air-compressor-load-rate", 4000);
-
+                UploadTelemetryData(connection, devid1, "1#air-compressor-two-level-discharge-temperature", 5000);
+                UploadTelemetryData(connection, devid2, "1#air-compressor-load-rate", 5000);
+                Console.WriteLine("");
+                Console.WriteLine("");
+                Console.WriteLine("");
                 DateTime dt2 = DateTime.Now;
-                UploadTelemetryDataPool(connection, devid1, "1#air-compressor-two-level-discharge-temperature1", 4000);
-                UploadTelemetryDataPool(connection, devid2, "1#air-compressor-load-rate1", 4000);
-                Console.WriteLine($"UploadTelemetryData 耗时:{DateTime.Now.Subtract(dt).TotalSeconds}");
-                Console.WriteLine($"UploadTelemetryDataPool 耗时:{DateTime.Now.Subtract(dt2).TotalSeconds}");
+                UploadTelemetryDataPool(connection, devid3, "1#air-compressor-two-level-discharge-temperature1", 5000);
+                UploadTelemetryDataPool(connection, devid4, "1#air-compressor-load-rate1", 5000);
+                var t1 = DateTime.Now.Subtract(dt).TotalSeconds;
+                var t2 = DateTime.Now.Subtract(dt2).TotalSeconds;
+                Console.WriteLine("Done");
+                Thread.Sleep(TimeSpan.FromSeconds(1));
+                Console.WriteLine($"UploadTelemetryData 耗时:{t1}");
+                Console.WriteLine($"UploadTelemetryDataPool 耗时:{t2}");
+                Thread.Sleep(TimeSpan.FromSeconds(2));
                 var reader2 = connection.CreateCommand("select last_row(*) from telemetrydata group by deviceid,keyname ;").ExecuteReader();
                 ConsoleTableBuilder.From(reader2.ToDataTable()).WithFormat(ConsoleTableBuilderFormat.Default).ExportAndWriteLine();
                 var reader3 = connection.CreateCommand("select * from telemetrydata").ExecuteReader();
-
                 List<string> list = new List<string>();
                 while (reader3.Read())
                 {
@@ -258,7 +266,15 @@ static void UploadTelemetryDataPool(TaosConnection connection, string devid, str
         {
             Parallel.For(0, count,new ParallelOptions() { MaxDegreeOfParallelism=connection.PoolSize }, i =>
             {
-                connection.CreateCommand($"INSERT INTO device_{devid} USING telemetrydata TAGS(\"{devid}\",\"{keyname}\") values (now,2,true,'{i}',{i},now,{i});").ExecuteNonQuery();
+                try
+                {
+                    connection.CreateCommand($"INSERT INTO device_{devid} USING telemetrydata TAGS(\"{devid}\",\"{keyname}\") values (now,2,true,'{i}',{i},now,{i});").ExecuteNonQuery();
+                    Console.WriteLine($"线程:{Thread.CurrentThread.ManagedThreadId} 第{i}条数据, OK");
+                }
+                catch (Exception ex)
+                {
+                    Console.WriteLine($"线程:{Thread.CurrentThread.ManagedThreadId} 第{i}条数据, {ex.Message}");
+                }
             });
         }
     }
diff --git a/src/IoTSharp.Data.Taos/ConcurrentTaosQueue.cs b/src/IoTSharp.Data.Taos/ConcurrentTaosQueue.cs
@@ -1,6 +1,7 @@
 ﻿using System;
 using System.Collections.Concurrent;
 using System.Collections.Generic;
+using System.Diagnostics;
 using System.Linq;
 using System.Text;
 using System.Threading;
@@ -26,10 +27,10 @@ public void Return(IntPtr client)
         {
             Monitor.Enter(TaosQueue);
             TaosQueue.Enqueue(client);
-            System.Diagnostics.Debug.WriteLine($"TaosQueue Return:{client}");
+            Debug.WriteLine($"线程{Thread.CurrentThread.ManagedThreadId} 归还 {client}");
             Monitor.Pulse(TaosQueue);
             Monitor.Exit(TaosQueue);
-           
+            Thread.Sleep(0);
         }
         int _ref = 0;
         public void AddRef()
@@ -50,16 +51,29 @@ public void RemoveRef()
                 _ref--;
             }
         }
+        public int Timeout { get; set; }
         public IntPtr Take()
         {
+            IntPtr client = IntPtr.Zero;
             Monitor.Enter(TaosQueue);
             if (TaosQueue.IsEmpty)
             {
-                Monitor.Wait(TaosQueue);
+                Debug.WriteLine($"线程{Thread.CurrentThread.ManagedThreadId} 连接池已空,请等待 超时时长:{Timeout}");
+                Monitor.Wait(TaosQueue, TimeSpan.FromSeconds(Timeout));
+            }
+            if (!TaosQueue.TryDequeue(out client))
+            {
+                Debug.WriteLine($"线程{Thread.CurrentThread.ManagedThreadId} 从连接池获取连接失败，等待并重试");
+            }
+            else
+            {
+                Debug.WriteLine($"线程{Thread.CurrentThread.ManagedThreadId} 拿走 {client}");
             }
-            TaosQueue.TryDequeue(out var client);
-            System.Diagnostics.Debug.WriteLine($"TaosQueue Take:{client}");
             Monitor.Exit(TaosQueue);
+            if (client == IntPtr.Zero)
+            {
+                throw new TimeoutException($"Connection pool is empty and wait time out({Timeout}s)!");
+            }
             return client;
         }
     }
diff --git a/src/IoTSharp.Data.Taos/TaosCommand.cs b/src/IoTSharp.Data.Taos/TaosCommand.cs
@@ -289,6 +289,14 @@ internal long GetDateTimeFrom(DateTime dt, IntPtr _taos)
         /// <exception cref="TaosException">A Taos error occurs during execution.</exception>
         public new virtual TaosDataReader ExecuteReader(CommandBehavior behavior)
         {
+            var _taos = _connection.TakeClient();
+            var dr= ExecuteReader(behavior, _taos);
+            dr.OnDispose += (object sender, EventArgs e)=>_connection.ReturnClient(_taos);
+            return dr;
+        }
+        
+        private TaosDataReader ExecuteReader(CommandBehavior behavior,IntPtr _taos)
+        { 
             if ((behavior & ~(CommandBehavior.Default | CommandBehavior.SequentialAccess | CommandBehavior.SingleResult
                               | CommandBehavior.SingleRow | CommandBehavior.CloseConnection)) != 0)
             {
@@ -322,7 +330,7 @@ internal long GetDateTimeFrom(DateTime dt, IntPtr _taos)
             var closeConnection = (behavior & CommandBehavior.CloseConnection) != 0;
             try
             {
-                var _taos = _connection.TakeClient();
+     
 #if DEBUG
                 Debug.WriteLine($"_commandText:{_commandText}");
 #endif
@@ -605,10 +613,14 @@ public override int ExecuteNonQuery()
             {
                 throw new InvalidOperationException($"CallRequiresSetCommandText{nameof(ExecuteNonQuery)}");
             }
-            using (var reader = ExecuteReader())
+            var _taos= _connection.TakeClient();
+            int result = -1;
+            using (var reader = ExecuteReader( CommandBehavior.Default,_taos))
             {
-                return reader.RecordsAffected;
+                result= reader.RecordsAffected;
             }
+            _connection.ReturnClient(_taos);
+            return result;
         }
 
         /// <summary>
@@ -626,13 +638,16 @@ public override object ExecuteScalar()
             {
                 throw new InvalidOperationException($"CallRequiresSetCommandText{nameof(ExecuteScalar)}");
             }
-
+            var _taos = _connection.TakeClient();
+            object result =null;
             using (var reader = ExecuteReader())
             {
-                return reader.Read()
+                result= reader.Read()
                     ? reader.GetValue(0)
                     : null;
             }
+            _connection.ReturnClient(_taos);
+            return result;
         }
       
         /// <summary>
diff --git a/src/IoTSharp.Data.Taos/TaosConnection.cs b/src/IoTSharp.Data.Taos/TaosConnection.cs
@@ -101,6 +101,8 @@ public override string ConnectionString
 
         internal TaosConnectionStringBuilder ConnectionStringBuilder { get; set; }
 
+        public override int ConnectionTimeout => ConnectionStringBuilder.ConnectionTimeout;
+
 
         /// <summary>
         ///     Gets the path to the database file. Will be absolute for open connections.
@@ -207,18 +209,18 @@ public override void Open()
             }
             if (!g_pool.ContainsKey(_connectionString))
             {
-                g_pool.Add(_connectionString, new ConcurrentTaosQueue());
+                g_pool.Add(_connectionString, new ConcurrentTaosQueue() {  Timeout= ConnectionTimeout});
             }
             _queue = g_pool[_connectionString];
             _queue.AddRef();
             if (ConnectionString == null)
             {
                 throw new InvalidOperationException("Open Requires Set ConnectionString");
             }
-            for (int i = 0; i < ConnectionStringBuilder.PoolSize; i++)
+            for (int i = 0; i < ConnectionStringBuilder.PoolSize+1; i++)
             {
                 var c = TDengine.Connect(this.DataSource, ConnectionStringBuilder.Username, ConnectionStringBuilder.Password, "", (short)ConnectionStringBuilder.Port);
-                if (c != null && c!=IntPtr.Zero)
+                if (c!=IntPtr.Zero)
                 {
                     _queue.Return(c);
                 }
diff --git a/src/IoTSharp.Data.Taos/TaosConnectionStringBuilder.cs b/src/IoTSharp.Data.Taos/TaosConnectionStringBuilder.cs
@@ -26,7 +26,7 @@ public class TaosConnectionStringBuilder : DbConnectionStringBuilder
         private const string DataBaseKeyword = "DataBase";
         private const string PortKeyword = "Port";
         private const string PoolSizeKeyword = "PoolSize";
-        
+        private const string TimeOutKeyword = "TimeOut";
         private enum Keywords
         {
             DataSource,
@@ -35,7 +35,9 @@ private enum Keywords
             Password,
             Port,
             Charset,
-            PoolSize
+            PoolSize,
+            TimeOut
+
 
         }
 
@@ -48,21 +50,23 @@ private enum Keywords
         private string _charset = System.Text.Encoding.UTF8.EncodingName;
         private string _password = string.Empty;
         private int  _port =6030;
-        private int _PoolSize=8;
+        private int _PoolSize=Environment.ProcessorCount;
+        private int _timeout = 30;
 
         static TaosConnectionStringBuilder()
         {
-            var validKeywords = new string[7];
+            var validKeywords = new string[8];
             validKeywords[(int)Keywords.DataSource] = DataSourceKeyword;
             validKeywords[(int)Keywords.DataBase] = DataBaseKeyword;
             validKeywords[(int)Keywords.Username] = UserNameKeyword;
             validKeywords[(int)Keywords.Password] = PasswordKeyword;
             validKeywords[(int)Keywords.Charset] =CharsetKeyword;
             validKeywords[(int)Keywords.Port] = PortKeyword;
             validKeywords[(int)Keywords.PoolSize] = PoolSizeKeyword;
+            validKeywords[(int)Keywords.TimeOut] = TimeOutKeyword;
             _validKeywords = validKeywords;
 
-            _keywords = new Dictionary<string, Keywords>(7, StringComparer.OrdinalIgnoreCase)
+            _keywords = new Dictionary<string, Keywords>(8, StringComparer.OrdinalIgnoreCase)
             {
                 [DataSourceKeyword] = Keywords.DataSource,
                 [UserNameKeyword] = Keywords.Username,
@@ -71,7 +75,8 @@ static TaosConnectionStringBuilder()
                 [CharsetKeyword] = Keywords.Charset,
                 [DataSourceNoSpaceKeyword] = Keywords.DataSource,
                 [PortKeyword] = Keywords.Port,
-                [PoolSizeKeyword] = Keywords.PoolSize
+                [PoolSizeKeyword] = Keywords.PoolSize,
+                [TimeOutKeyword] = Keywords.TimeOut
             };
         }
 
@@ -100,6 +105,8 @@ public virtual string DataSource
             get => _dataSource;
             set => base[DataSourceKeyword] = _dataSource = value;
         }
+
+     
         public virtual string Username
         {
             get => _userName;
@@ -164,7 +171,13 @@ public virtual string DataBase
         }
         internal bool ForceDatabaseName { get; set; } = false;
 
+      
 
+        public int ConnectionTimeout
+        {
+            get => _timeout;
+            set => base[TimeOutKeyword] = _timeout = value;
+        }
 
         /// <summary>
         ///     Gets or sets the value associated with the specified key.
@@ -206,6 +219,9 @@ public override object this[string keyword]
                     case Keywords.PoolSize:
                         PoolSize = Convert.ToInt32(value, CultureInfo.InvariantCulture);
                         return;
+                    case Keywords.TimeOut:
+                        ConnectionTimeout = Convert.ToInt32(value, CultureInfo.InvariantCulture);
+                        return;
                     default:
                         Debug.Assert(false, "Unexpected keyword: " + keyword);
                         return;
@@ -330,6 +346,8 @@ private object GetAt(Keywords index)
                     return Charset;
                 case Keywords.PoolSize:
                     return PoolSize;
+                case Keywords.TimeOut:
+                    return ConnectionTimeout;
                 default:
                     Debug.Assert(false, "Unexpected keyword: " + index);
                     return null;
@@ -358,13 +376,16 @@ private void Reset(Keywords index)
                     _dataBase = string.Empty;
                     return;
                 case Keywords.Port:
-                    _port=6060;
+                    _port=6030;
                     return;
                 case Keywords.Charset:
                     _charset = System.Text.Encoding.UTF8.EncodingName;
                     return;
                 case Keywords.PoolSize:
-                    _PoolSize = 8;
+                    _PoolSize = Environment.ProcessorCount;
+                    return;
+                case Keywords.TimeOut   :
+                    _timeout = 30;
                     return;
                 default:
                     Debug.Assert(false, "Unexpected keyword: " + index);
diff --git a/src/IoTSharp.Data.Taos/TaosDataReader.cs b/src/IoTSharp.Data.Taos/TaosDataReader.cs
@@ -24,7 +24,6 @@ public class TaosDataReader : DbDataReader
         private readonly int _recordsAffected;
         private bool _closed;
         private bool _closeConnection;
-        private readonly IntPtr _taos;
         private IntPtr _taosResult;
         private int _fieldCount;
         IntPtr rowdata=IntPtr.Zero;
@@ -33,11 +32,12 @@ public class TaosDataReader : DbDataReader
         private DateTime _dt1970;
         private TAOS_BIND[] _binds;
 
+
+
         internal TaosDataReader(TaosCommand taosCommand, taosField[] metas, bool closeConnection, IntPtr taos, IntPtr res, int recordsAffected, int fieldcount, TAOS_BIND[] binds)
         {
             _command = taosCommand;
             _closeConnection = closeConnection;
-           _taos = taos;
             _fieldCount = fieldcount;
             _hasRows = recordsAffected > 0;
             _recordsAffected = recordsAffected;
@@ -145,6 +145,8 @@ public override bool NextResult()
         public override void Close()
             => Dispose(true);
 
+
+        internal event EventHandler OnDispose;
         /// <summary>
         ///     Releases any resources used by the data reader and closes it.
         /// </summary>
@@ -160,8 +162,7 @@ protected override void Dispose(bool disposing)
             _command.DataReader = null;
        
             _closed = true;
-         
-           
+      
             TDengine.FreeResult(_taosResult);
             if (_binds != null)
             {
@@ -172,7 +173,7 @@ protected override void Dispose(bool disposing)
             {
                 TDengine.FreeResult(rowdata);
             }
-            _command.Connection.ReturnClient(_taos);
+            OnDispose?.Invoke(this, EventArgs.Empty);
         }
 
         /// <summary>

Original file line number	Diff line number	Diff line change
`@@ -289,6 +289,14 @@ internal long GetDateTimeFrom(DateTime dt, IntPtr _taos)`
`289`	`289`	`/// <exception cref="TaosException">A Taos error occurs during execution.</exception>`
`290`	`290`	`public new virtual TaosDataReader ExecuteReader(CommandBehavior behavior)`
`291`	`291`	`{`
	`292`	`+ var _taos = _connection.TakeClient();`
	`293`	`+ var dr= ExecuteReader(behavior, _taos);`
	`294`	`+ dr.OnDispose += (object sender, EventArgs e)=>_connection.ReturnClient(_taos);`
	`295`	`+ return dr;`
	`296`	`+ }`
	`297`	`+`
	`298`	`+ private TaosDataReader ExecuteReader(CommandBehavior behavior,IntPtr _taos)`
	`299`	`+ {`
`292`	`300`	`if ((behavior & ~(CommandBehavior.Default \| CommandBehavior.SequentialAccess \| CommandBehavior.SingleResult`
`293`	`301`	`\| CommandBehavior.SingleRow \| CommandBehavior.CloseConnection)) != 0)`
`294`	`302`	`{`
`@@ -322,7 +330,7 @@ internal long GetDateTimeFrom(DateTime dt, IntPtr _taos)`
`322`	`330`	`var closeConnection = (behavior & CommandBehavior.CloseConnection) != 0;`
`323`	`331`	`try`
`324`	`332`	`{`
`325`		`- var _taos = _connection.TakeClient();`
	`333`	`+`
`326`	`334`	`#if DEBUG`
`327`	`335`	`Debug.WriteLine($"_commandText:{_commandText}");`
`328`	`336`	`#endif`
`@@ -605,10 +613,14 @@ public override int ExecuteNonQuery()`
`605`	`613`	`{`
`606`	`614`	`throw new InvalidOperationException($"CallRequiresSetCommandText{nameof(ExecuteNonQuery)}");`
`607`	`615`	`}`
`608`		`- using (var reader = ExecuteReader())`
	`616`	`+ var _taos= _connection.TakeClient();`
	`617`	`+ int result = -1;`
	`618`	`+ using (var reader = ExecuteReader( CommandBehavior.Default,_taos))`
`609`	`619`	`{`
`610`		`- return reader.RecordsAffected;`
	`620`	`+ result= reader.RecordsAffected;`
`611`	`621`	`}`
	`622`	`+ _connection.ReturnClient(_taos);`
	`623`	`+ return result;`
`612`	`624`	`}`
`613`	`625`
`614`	`626`	`/// <summary>`
`@@ -626,13 +638,16 @@ public override object ExecuteScalar()`
`626`	`638`	`{`
`627`	`639`	`throw new InvalidOperationException($"CallRequiresSetCommandText{nameof(ExecuteScalar)}");`
`628`	`640`	`}`
`629`		`-`
	`641`	`+ var _taos = _connection.TakeClient();`
	`642`	`+ object result =null;`
`630`	`643`	`using (var reader = ExecuteReader())`
`631`	`644`	`{`
`632`		`- return reader.Read()`
	`645`	`+ result= reader.Read()`
`633`	`646`	`? reader.GetValue(0)`
`634`	`647`	`: null;`
`635`	`648`	`}`
	`649`	`+ _connection.ReturnClient(_taos);`
	`650`	`+ return result;`
`636`	`651`	`}`
`637`	`652`
`638`	`653`	`/// <summary>`
Original file line number	Diff line number	Diff line change
`@@ -101,6 +101,8 @@ public override string ConnectionString`
`101`	`101`
`102`	`102`	`internal TaosConnectionStringBuilder ConnectionStringBuilder { get; set; }`
`103`	`103`
	`104`	`+ public override int ConnectionTimeout => ConnectionStringBuilder.ConnectionTimeout;`
	`105`	`+`
`104`	`106`
`105`	`107`	`/// <summary>`
`106`	`108`	`/// Gets the path to the database file. Will be absolute for open connections.`
`@@ -207,18 +209,18 @@ public override void Open()`
`207`	`209`	`}`
`208`	`210`	`if (!g_pool.ContainsKey(_connectionString))`
`209`	`211`	`{`
`210`		`- g_pool.Add(_connectionString, new ConcurrentTaosQueue());`
	`212`	`+ g_pool.Add(_connectionString, new ConcurrentTaosQueue() { Timeout= ConnectionTimeout});`
`211`	`213`	`}`
`212`	`214`	`_queue = g_pool[_connectionString];`
`213`	`215`	`_queue.AddRef();`
`214`	`216`	`if (ConnectionString == null)`
`215`	`217`	`{`
`216`	`218`	`throw new InvalidOperationException("Open Requires Set ConnectionString");`
`217`	`219`	`}`
`218`		`- for (int i = 0; i < ConnectionStringBuilder.PoolSize; i++)`
	`220`	`+ for (int i = 0; i < ConnectionStringBuilder.PoolSize+1; i++)`
`219`	`221`	`{`
`220`	`222`	`var c = TDengine.Connect(this.DataSource, ConnectionStringBuilder.Username, ConnectionStringBuilder.Password, "", (short)ConnectionStringBuilder.Port);`
`221`		`- if (c != null && c!=IntPtr.Zero)`
	`223`	`+ if (c!=IntPtr.Zero)`
`222`	`224`	`{`
`223`	`225`	`_queue.Return(c);`
`224`	`226`	`}`