优化 线程池 队列处理。 加入线程池为空时超时等待。

Author: maikebing

launchSettings.json

@@ -0,0 +1,14 @@
+{
+  "profiles": {
+    "Docker Compose": {
+      "commandName": "DockerCompose",
+      "commandVersion": "1.0",
+      "composeLaunchAction": "None",
+      "composeLaunchServiceName": "apache.iotdb.samples",
+      "serviceActions": {
+        "apache.iotdb.samples": "StartDebugging",
+        "iotdb": "StartWithoutDebugging"
+      }
+    }
+  }
+}

samples/Apache.IoTDB.Samples/SessionPoolTest.cs

@@ -276,6 +276,7 @@ public async Task TestNonSqlBy_ADO()
         {
             var cnts = new IoTDB.Data.IoTDBConnectionStringBuilder();
             cnts.DataSource = host;
+            cnts.TimeOut =(int) TimeSpan.FromSeconds(20).TotalMilliseconds;
             var cnt = new IoTDB.Data.IoTDBConnection(cnts.ConnectionString);
             await cnt.OpenAsync();
             var session_pool = cnt.SessionPool;

src/Apache.IoTDB.Data/DataReaderExtensions.cs

@@ -14,7 +14,7 @@ public static class DataReaderExtensions
     {
         public static SessionPool CreateSession(this IoTDBConnectionStringBuilder db)
         {
-            return new SessionPool(db.DataSource, db.Port, db.Username, db.Password, db.FetchSize, db.ZoneId, db.PoolSize,db.Compression);
+            return new SessionPool(db.DataSource, db.Port, db.Username, db.Password, db.FetchSize, db.ZoneId, db.PoolSize,db.Compression,db.TimeOut);
         }
 
         public static List<T> ToObject<T>(this IDataReader dataReader)

src/Apache.IoTDB.Data/IoTDBConnectionStringBuilder.cs

@@ -6,6 +6,7 @@
 using System.Diagnostics;
 using System.Globalization;
 using System.Reflection;
+using System.Threading;
 
 namespace Apache.IoTDB.Data
 {
@@ -23,7 +24,8 @@ public class IoTDBConnectionStringBuilder : DbConnectionStringBuilder
         private const string CompressionKeyword = "Compression";
         private const string PoolSizeKeyword = "PoolSize";
         private const string ZoneIdKeyword = "ZoneId";
- 
+        private const string TimeOutKeyword = "TimeOut";
+
         private enum Keywords
         {
             DataSource,
@@ -33,7 +35,8 @@ private enum Keywords
             FetchSize,
             Compression,
             PoolSize,
-            ZoneId
+            ZoneId,
+            TimeOut
         }
 
         private static readonly IReadOnlyList<string> _validKeywords;
@@ -47,9 +50,11 @@ private enum Keywords
         private string _zoneId = "UTC+08:00";
         private int  _port = 6667;
       private   int  _poolSize =8;
+        private int _timeOut=10000;
+
         static IoTDBConnectionStringBuilder()
         {
-            var validKeywords = new string[8];
+            var validKeywords = new string[9];
             validKeywords[(int)Keywords.DataSource] = DataSourceKeyword;
              validKeywords[(int)Keywords.Username] = UserNameKeyword;
             validKeywords[(int)Keywords.Password] = PasswordKeyword;
@@ -58,9 +63,10 @@ static IoTDBConnectionStringBuilder()
             validKeywords[(int)Keywords.Compression] = CompressionKeyword;
             validKeywords[(int)Keywords.PoolSize] = PoolSizeKeyword;
             validKeywords[(int)Keywords.ZoneId] = ZoneIdKeyword;
+            validKeywords[(int)Keywords.TimeOut] =TimeOutKeyword;
             _validKeywords = validKeywords;
 
-            _keywords = new Dictionary<string, Keywords>(6, StringComparer.OrdinalIgnoreCase)
+            _keywords = new Dictionary<string, Keywords>(9, StringComparer.OrdinalIgnoreCase)
             {
                 [DataSourceKeyword] = Keywords.DataSource,
                 [UserNameKeyword] = Keywords.Username,
@@ -69,7 +75,8 @@ static IoTDBConnectionStringBuilder()
                 [FetchSizeKeyword] = Keywords.FetchSize,
                 [CompressionKeyword] = Keywords.Compression,
                 [PoolSizeKeyword] = Keywords.PoolSize,
-                [ZoneIdKeyword] = Keywords.ZoneId
+                [ZoneIdKeyword] = Keywords.ZoneId,
+                [TimeOutKeyword] = Keywords.TimeOut
             };
         }
 
@@ -135,8 +142,12 @@ public virtual string ZoneId
             get => _zoneId;
             set => base[ZoneIdKeyword] = _zoneId = value;
         }
- 
-        
+
+        public virtual int TimeOut
+    {
+            get => _timeOut;
+            set => base[PoolSizeKeyword] = _timeOut = value;
+        }
 
         /// <summary>
         ///     Gets a collection containing the keys used by the connection string.
@@ -162,8 +173,10 @@ public override ICollection Values
                 return new ReadOnlyCollection<object>(values);
             }
         }
- 
- 
+
+
+
+
 
 
 
@@ -211,6 +224,9 @@ public override object this[string keyword]
                     case Keywords.ZoneId:
                         ZoneId = Convert.ToString(value, CultureInfo.InvariantCulture);
                         return;
+                    case Keywords.TimeOut:
+                        TimeOut = Convert.ToInt32(value, CultureInfo.InvariantCulture);
+                        return;
                     default:
                         Debug.WriteLine(false, "Unexpected keyword: " + keyword);
                         return;
@@ -339,6 +355,8 @@ private object GetAt(Keywords index)
                     return PoolSize;
                 case Keywords.ZoneId:
                     return ZoneId;
+                case Keywords.TimeOut:
+                    return TimeOut;
                 default:
                     Debug.Assert(false, "Unexpected keyword: " + index);
                     return null;
@@ -378,6 +396,9 @@ private void Reset(Keywords index)
                 case Keywords.ZoneId:
                     _zoneId = "UTC+08:00";
                     return;
+                case Keywords.TimeOut:
+                    _timeOut = 10000;//10sec.
+                    return;
                 default:
                     Debug.Assert(false, "Unexpected keyword: " + index);
                     return;

src/Apache.IoTDB/ConcurrentClientQueue.cs

@@ -1,43 +1,86 @@
+using System;
 using System.Collections.Concurrent;
 using System.Collections.Generic;
+using System.Diagnostics;
 using System.Threading;
 
 namespace Apache.IoTDB
 {
-    public class ConcurrentClientQueue
-    {
-        public ConcurrentQueue<Client> ClientQueue { get; }
-
-        public ConcurrentClientQueue(List<Client> clients)
-        {
-            ClientQueue = new ConcurrentQueue<Client>(clients);
-        }
-
-        public ConcurrentClientQueue()
-        {
-            ClientQueue = new ConcurrentQueue<Client>();
-        }
-
-        public void Add(Client client)
-        {
-            Monitor.Enter(ClientQueue);
-            ClientQueue.Enqueue(client);
-            Monitor.Pulse(ClientQueue);
-            Monitor.Exit(ClientQueue);
-        }
-
-        public Client Take()
-        {
-            Monitor.Enter(ClientQueue);
-            
-            if (ClientQueue.IsEmpty)
-            {
-                Monitor.Wait(ClientQueue);
-            }
-
-            ClientQueue.TryDequeue(out var client);
-            Monitor.Exit(ClientQueue);
-            return client;
-        }
+    public class ConcurrentClientQueue
+    {
+        public ConcurrentQueue<Client> ClientQueue { get; }
+
+        public ConcurrentClientQueue(List<Client> clients)
+        {
+            ClientQueue = new ConcurrentQueue<Client>(clients);
+        }
+        public ConcurrentClientQueue()
+        {
+            ClientQueue = new ConcurrentQueue<Client>();
+        }
+        public void Add(Client client) => Return(client);
+
+        public void Return(Client client)
+        {
+            Monitor.Enter(ClientQueue);
+            ClientQueue.Enqueue(client);
+#if DEBUG
+            Console.WriteLine($"线程{Thread.CurrentThread.ManagedThreadId} 归还 {client}");
+#endif
+            Monitor.Pulse(ClientQueue);
+            Monitor.Exit(ClientQueue);
+            Thread.Sleep(0);
+        }
+        int _ref = 0;
+        public void AddRef()
+        {
+            lock (this)
+            {
+                _ref++;
+            }
+        }
+        public int GetRef()
+        {
+            return _ref;
+        }
+        public void RemoveRef()
+        {
+            lock (this)
+            {
+                _ref--;
+            }
+        }
+        public int Timeout { get; set; } = 10;
+        public Client Take()
+        {
+            Client client = null;
+            Monitor.Enter(ClientQueue);
+            if (ClientQueue.IsEmpty)
+            {
+#if DEBUG
+                Console.WriteLine($"线程{Thread.CurrentThread.ManagedThreadId} 连接池已空,请等待 超时时长:{Timeout}");
+#endif
+                Monitor.Wait(ClientQueue, TimeSpan.FromSeconds(Timeout));
+            }
+            if (!ClientQueue.TryDequeue(out client))
+            {
+#if DEBUG
+                Console.WriteLine($"线程{Thread.CurrentThread.ManagedThreadId} 从连接池获取连接失败，等待并重试");
+#endif
+            }
+            else
+            {
+
+#if DEBUG
+                Console.WriteLine($"线程{Thread.CurrentThread.ManagedThreadId} 拿走 {client}");
+#endif
+            }
+            Monitor.Exit(ClientQueue);
+            if (client == null)
+            {
+                throw new TimeoutException($"Connection pool is empty and wait time out({Timeout}s)!");
+            }
+            return client;
+        }
     }
 }