feat: 实现udp并发接受数据？？

server/src/Peripherals/CameraClient.cs

@@ -237,6 +237,33 @@ class Camera
         return result.Value;
     }
 
+    /// <summary>
+    /// 读取摄像头寄存器
+    /// </summary>
+    /// <param name="registerAddr">寄存器地址</param>
+    /// <returns>读取到的寄存器值</returns>
+    private async ValueTask<Result<byte>> ReadRegister(UInt16 registerAddr)
+    {
+        var i2c = new Peripherals.I2cClient.I2c(this.address, this.port, this.taskID, this.timeout);
+        
+        // 地址高低字节
+        var addrBytes = new byte[2];
+        addrBytes[0] = (byte)(registerAddr >> 8);
+        addrBytes[1] = (byte)(registerAddr & 0xFF);
+
+        // 先写寄存器地址
+        var writeResult = await i2c.WriteData(CAM_I2C_ADDR, addrBytes, CAM_PROTO);
+        if (!writeResult.IsSuccessful)
+            return new(writeResult.Error);
+
+        // 再读一个字节
+        var readResult = await i2c.ReadData(CAM_I2C_ADDR, 1, CAM_PROTO);
+        if (!readResult.IsSuccessful)
+            return new(readResult.Error);
+
+        return readResult.Value[0];
+    }
+
     /// <summary>
     /// 批量配置I2C寄存器
     /// </summary>

server/src/UdpServer.cs

@@ -1,9 +1,9 @@
+using System.Collections.Concurrent;
 using System.Net;
 using System.Net.Sockets;
 using System.Runtime.CompilerServices;
 using System.Text;
 using DotNext;
-using DotNext.Threading;
 using Newtonsoft.Json;
 using WebProtocol;
 
@@ -72,9 +72,11 @@ public class UDPServer
 {
     private static NLog.Logger logger = NLog.LogManager.GetCurrentClassLogger();
 
-    private Dictionary<string, Queue<UDPData>> udpData = new Dictionary<string, Queue<UDPData>>();
+    private ConcurrentDictionary<string, ConcurrentQueue<UDPData>> udpData = new ConcurrentDictionary<string, ConcurrentQueue<UDPData>>();
 
-    private Semaphore taskPool = new Semaphore(3, 3);
+    const int parallelTaskNum = 5;
+    Task[] ReceiveTasks = new Task[parallelTaskNum];
+    private int runningTaskNum = 0;
 
     private int listenPort;
     private UdpClient listener;
@@ -145,28 +147,19 @@ public class UDPServer
     {
         UDPData? data = null;
 
-        // logger.Debug($"Caller \"{callerName}|{callerLineNum}\": Try to find {ipAddr}-{taskID} UDP Data");
-
         var startTime = DateTime.Now;
         var isTimeout = false;
-        var timeleft = TimeSpan.FromMilliseconds(timeout);
         while (!isTimeout)
         {
             var elapsed = DateTime.Now - startTime;
             isTimeout = elapsed >= TimeSpan.FromMilliseconds(timeout);
             if (isTimeout) break;
 
-            timeleft = TimeSpan.FromMilliseconds(timeout) - elapsed;
+            if (udpData.TryGetValue($"{ipAddr}-{taskID}", out var dataQueue) &&
-            using (await udpData.AcquireWriteLockAsync(timeleft))
+                dataQueue.TryDequeue(out data))
             {
-                if (udpData.ContainsKey($"{ipAddr}-{taskID}") &&
+                // logger.Debug($"Find UDP Data: {data.ToString()}");
-                    udpData.TryGetValue($"{ipAddr}-{taskID}", out var dataQueue) &&
+                break;
-                    dataQueue.Count > 0)
-                {
-                    data = dataQueue.Dequeue();
-                    // logger.Debug($"Find UDP Data: {data.ToString()}");
-                    break;
-                }
             }
 
             await Task.Delay(cycle);
@@ -196,24 +189,21 @@ public class UDPServer
 
         var startTime = DateTime.Now;
         var isTimeout = false;
-        var timeleft = TimeSpan.FromMilliseconds(timeout);
         while (!isTimeout)
         {
             var elapsed = DateTime.Now - startTime;
             isTimeout = elapsed >= TimeSpan.FromMilliseconds(timeout);
             if (isTimeout) break;
 
-            timeleft = TimeSpan.FromMilliseconds(timeout) - elapsed;
+            if (udpData.TryGetValue($"{ipAddr}-{taskID}", out var dataQueue) &&
-            using (await udpData.AcquireWriteLockAsync(timeleft))
+                !dataQueue.IsEmpty)
             {
-                if (udpData.ContainsKey($"{ipAddr}-{taskID}") &&
+                data = new List<UDPData>();
-                    udpData.TryGetValue($"{ipAddr}-{taskID}", out var dataQueue) &&
+                while (dataQueue.TryDequeue(out var item))
-                    dataQueue.Count > 0)
                 {
-                    data = dataQueue.ToList();
+                    data.Add(item);
-                    dataQueue.Clear();
-                    break;
                 }
+                break;
             }
         }
 
@@ -241,24 +231,18 @@ public class UDPServer
 
         var startTime = DateTime.Now;
         var isTimeout = false;
-        var timeleft = TimeSpan.FromMilliseconds(timeout);
         while (!isTimeout)
         {
             var elapsed = DateTime.Now - startTime;
             isTimeout = elapsed >= TimeSpan.FromMilliseconds(timeout);
             if (isTimeout) break;
 
-            timeleft = TimeSpan.FromMilliseconds(timeout) - elapsed;
+            if (udpData.TryGetValue($"{ipAddr}-{taskID}", out var dataQueue) &&
-            using (await udpData.AcquireReadLockAsync(timeleft))
+                !dataQueue.IsEmpty)
             {
-                if (udpData.ContainsKey($"{ipAddr}-{taskID}") &&
+                data = dataQueue.ToArray().ToList();
-                    udpData.TryGetValue($"{ipAddr}-{taskID}", out var dataQueue) &&
+                // logger.Debug($"Find UDP Data Array: {JsonConvert.SerializeObject(data)}");
-                    dataQueue.Count > 0)
+                break;
-                {
-                    data = dataQueue.ToList();
-                    // logger.Debug($"Find UDP Data Array: {JsonConvert.SerializeObject(data)}");
-                    break;
-                }
             }
         }
 
@@ -286,22 +270,16 @@ public class UDPServer
 
         var startTime = DateTime.Now;
         var isTimeout = false;
-        var timeleft = TimeSpan.FromMilliseconds(timeout);
         while (!isTimeout)
         {
             var elapsed = DateTime.Now - startTime;
             isTimeout = elapsed >= TimeSpan.FromMilliseconds(timeout);
             if (isTimeout) break;
 
-            timeleft = TimeSpan.FromMilliseconds(timeout) - elapsed;
+            if (udpData.TryGetValue($"{ipAddr}-{taskID}", out var dataQueue))
-            using (await udpData.AcquireReadLockAsync(timeleft))
             {
-                if (udpData.ContainsKey($"{ipAddr}-{taskID}") &&
+                count = dataQueue.Count;
-                    udpData.TryGetValue($"{ipAddr}-{taskID}", out var dataQueue))
+                break;
-                {
-                    count = dataQueue.Count;
-                    break;
-                }
             }
         }
 
@@ -367,17 +345,21 @@ public class UDPServer
 
         return retPack.Value;
     }
-    static int ReceiveHandleCcount = 0;
 
     private void ReceiveHandler(IAsyncResult res)
     {
         var remoteEP = new IPEndPoint(IPAddress.Any, listenPort);
         byte[] bytes = listener.EndReceive(res, ref remoteEP);
 
-        // 提前开始接收下一个包
+        // 继续异步接收
-        listener.BeginReceive(new AsyncCallback(ReceiveHandler), null);
+        if (isRunning)
-        logger.Debug($"Test ReceiveHandler Count = {ReceiveHandleCcount}");
+        {
-        ReceiveHandleCcount++;
+            listener.BeginReceive(new AsyncCallback(ReceiveHandler), null);
+        }
+        else
+        {
+            runningTaskNum--;
+        }
 
         // Handle RemoteEP
         if (remoteEP is null)
@@ -426,23 +408,11 @@ public class UDPServer
             HasRead = false,
         };
 
-        using (udpData.AcquireWriteLock())
+        var key = $"{remoteAddress}-{taskID}";
-        {
+        var dataQueue = udpData.GetOrAdd(key, _ => new ConcurrentQueue<UDPData>());
-            // Record UDP Receive Data
+        dataQueue.Enqueue(data);
-            if (udpData.ContainsKey($"{remoteAddress}-{taskID}") &&
+
-                udpData.TryGetValue($"{remoteAddress}-{taskID}", out var dataQueue))
+        logger.Debug($"Test dataQueue.Count = {dataQueue.Count}");
-            {
-                dataQueue.Enqueue(data);
-                logger.Debug($"Test Lock dataQueue.Count = {dataQueue.Count}");
-            }
-            else
-            {
-                var queue = new Queue<UDPData>();
-                queue.Enqueue(data);
-                udpData.Add($"{remoteAddress}-{taskID}", queue);
-                logger.Trace("Receive data from new client");
-            }
-        }
 
         return data;
     }
@@ -497,16 +467,13 @@ public class UDPServer
     /// <returns> void </returns>
     public void PrintAllData()
     {
-        using (udpData.AcquireReadLock())
+        logger.Debug("Ready Data:");
-        {
-            // logger.Debug("Ready Data:");
 
-            foreach (var ip in udpData)
+        foreach (var kvp in udpData)
+        {
+            foreach (var data in kvp.Value)
             {
-                foreach (var data in ip.Value)
+                logger.Debug(data.ToString());
-                {
-                    // logger.Debug(data.ToString());
-                }
             }
         }
     }
@@ -519,35 +486,58 @@ public class UDPServer
     /// <returns>无</returns>
     public async Task ClearUDPData(string ipAddr, int taskID)
     {
-        using (await udpData.AcquireWriteLockAsync())
+        var key = $"{ipAddr}-{taskID}";
+        if (udpData.TryGetValue(key, out var dataQueue))
         {
-            if (udpData.ContainsKey($"{ipAddr}-{taskID}") &&
+            // 清空队列的最有效方式是替换为新的队列
-                udpData.TryGetValue($"{ipAddr}-{taskID}", out var dataQueue) &&
+            udpData.TryUpdate(key, new ConcurrentQueue<UDPData>(), dataQueue);
-                dataQueue.Count > 0)
-            {
-                dataQueue.Clear();
-            }
         }
     }
 
-
     /// <summary>
     /// Start UDP Server
     /// </summary>
     /// <returns>None</returns>
     public void Start()
     {
+        this.isRunning = true;
         try
         {
-            this.listener.BeginReceive(new AsyncCallback(ReceiveHandler), null);
+            Task.Run(() =>
+            {
+                while (isRunning)
+                {
+                    if (runningTaskNum < parallelTaskNum)
+                    {
+                        StartReceive();
+                        logger.Debug($"Begin Receive Task, Now Running Num: {runningTaskNum + 1}");
+                        runningTaskNum++;
+                    }
+                    Task.Delay(100).Wait();
+                }
+            });
         }
         catch (Exception e)
         {
             Console.WriteLine(e.ToString());
+            this.isRunning = false;
         }
-        finally
+    }
+
+    private void StartReceive(int times = 5)
+    {
+
+        for (int i = 0; i < times; i++)
         {
-            this.isRunning = true;
+            try
+            {
+                this.listener.BeginReceive(new AsyncCallback(ReceiveHandler), null);
+                break; // BeginReceive is async, break after scheduling
+            }
+            catch (Exception ex)
+            {
+                Console.WriteLine(ex.ToString());
+            }
         }
     }