From 1053d71d29031d1de42654cce62c12b38b95180a Mon Sep 17 00:00:00 2001 From: SikongJueluo Date: Thu, 17 Jul 2025 14:11:24 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E7=94=B1=E4=BA=8E=E8=A7=A3=E6=9E=90?= =?UTF-8?q?=E9=94=99=E8=AF=AF=E5=AF=BC=E8=87=B4=E7=9A=84=E6=97=A0=E6=B3=95?= =?UTF-8?q?=E9=80=9A=E4=BF=A1=E7=9A=84=E9=97=AE=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- server/server.csproj | 4 +- server/src/UdpClientPool.cs | 6 +- server/src/UdpServer.cs | 176 +++++++++++++++--------------------- server/src/WebProtocol.cs | 20 ++++ 4 files changed, 99 insertions(+), 107 deletions(-) diff --git a/server/server.csproj b/server/server.csproj index 42d0ef1..59e8975 100644 --- a/server/server.csproj +++ b/server/server.csproj @@ -14,8 +14,8 @@ - - + + diff --git a/server/src/UdpClientPool.cs b/server/src/UdpClientPool.cs index 06e7a30..bd2a3c7 100644 --- a/server/src/UdpClientPool.cs +++ b/server/src/UdpClientPool.cs @@ -525,9 +525,9 @@ public class UDPClientPool { var bytes = udpDatas[i].Data; var expectedLen = ((pkgList[i].Options.BurstLength + 1) * 4); - if ((bytes.Length - 4) != expectedLen) - return new(new Exception($"Expected {expectedLen} bytes but received {bytes.Length - 4} bytes at segment {i}")); - resultData.AddRange(bytes[4..]); + if ((bytes.Length - 8) != expectedLen) + return new(new Exception($"Expected {expectedLen} bytes but received {bytes.Length - 8} bytes at segment {i}")); + resultData.AddRange(bytes[8..]); } // Validate total data length diff --git a/server/src/UdpServer.cs b/server/src/UdpServer.cs index 9420b1f..4705f14 100644 --- a/server/src/UdpServer.cs +++ b/server/src/UdpServer.cs @@ -6,6 +6,7 @@ using System.Runtime.CompilerServices; using System.Text; using Common; using DotNext; +using DotNext.Threading; using Newtonsoft.Json; using WebProtocol; @@ -81,7 +82,7 @@ public class UDPServer private ConcurrentDictionary> udpData = new ConcurrentDictionary>(); - private readonly ReaderWriterLockSlim udpDataLock = new ReaderWriterLockSlim(); + private readonly AsyncReaderWriterLock udpDataLock = new AsyncReaderWriterLock(); private int listenPort; private List listeners = new List(); @@ -189,45 +190,36 @@ public class UDPServer ) { UDPData? data = null; + var key = $"{ipAddr}-{taskID}"; - var startTime = DateTime.Now; - var isTimeout = false; - while (!isTimeout) + try { - var elapsed = DateTime.Now - startTime; - isTimeout = elapsed >= TimeSpan.FromMilliseconds(timeout); - if (isTimeout) break; - - udpDataLock.EnterWriteLock(); - try + using (await udpDataLock.AcquireWriteLockAsync(TimeSpan.FromMilliseconds(timeout))) { - var key = $"{ipAddr}-{taskID}"; if (udpData.TryGetValue(key, out var sortedList) && sortedList.Count > 0) { // 获取最早的数据（第一个元素） var firstKey = sortedList.Keys[0]; data = sortedList[firstKey]; sortedList.RemoveAt(0); - break; } } - finally + + if (data is null) { - udpDataLock.ExitWriteLock(); + logger.Trace("Get nothing even after time out"); + return new(null); + } + else + { + return new(data.DeepClone()); } - - await Task.Delay(cycle); } - - if (data is null) + catch { logger.Trace("Get nothing even after time out"); return new(null); } - else - { - return new(data.DeepClone()); - } } /// @@ -240,43 +232,38 @@ public class UDPServer public async ValueTask>> FindDataArrayAsync(string ipAddr, int taskID, int timeout = 1000) { List? data = null; + var key = $"{ipAddr}-{taskID}"; - var startTime = DateTime.Now; - var isTimeout = false; - while (!isTimeout) + try { - var elapsed = DateTime.Now - startTime; - isTimeout = elapsed >= TimeSpan.FromMilliseconds(timeout); - if (isTimeout) break; - - udpDataLock.EnterWriteLock(); - try + using (await udpDataLock.AcquireWriteLockAsync(TimeSpan.FromMilliseconds(timeout))) { - var key = $"{ipAddr}-{taskID}"; if (udpData.TryGetValue(key, out var sortedList) && sortedList.Count > 0) { data = new List(sortedList.Values); // 输出数据 // PrintDataArray(data); sortedList.Clear(); - break; } + } - finally + + if (data is null) { - udpDataLock.ExitWriteLock(); + logger.Trace("Get nothing even after time out"); + return new(null); + } + else + { + return new(data); } } - - if (data is null) + catch { logger.Trace("Get nothing even after time out"); return new(null); } - else - { - return new(data); - } + } /// @@ -290,28 +277,21 @@ public class UDPServer { List? data = null; - var startTime = DateTime.Now; - var isTimeout = false; - while (!isTimeout) + try { - var elapsed = DateTime.Now - startTime; - isTimeout = elapsed >= TimeSpan.FromMilliseconds(timeout); - if (isTimeout) break; - - udpDataLock.EnterReadLock(); - try + using (await udpDataLock.AcquireReadLockAsync(TimeSpan.FromMilliseconds(timeout))) { var key = $"{ipAddr}-{taskID}"; if (udpData.TryGetValue(key, out var sortedList) && sortedList.Count > 0) { data = new List(sortedList.Values); - break; } } - finally - { - udpDataLock.ExitReadLock(); - } + } + catch (TimeoutException) + { + logger.Trace("Failed to acquire read lock within timeout"); + return new(null); } if (data is null) @@ -336,28 +316,21 @@ public class UDPServer { int? count = null; - var startTime = DateTime.Now; - var isTimeout = false; - while (!isTimeout) + try { - var elapsed = DateTime.Now - startTime; - isTimeout = elapsed >= TimeSpan.FromMilliseconds(timeout); - if (isTimeout) break; - - udpDataLock.EnterReadLock(); - try + using (await udpDataLock.AcquireReadLockAsync(TimeSpan.FromMilliseconds(timeout))) { var key = $"{ipAddr}-{taskID}"; if (udpData.TryGetValue(key, out var sortedList)) { count = sortedList.Count; - break; } } - finally - { - udpDataLock.ExitReadLock(); - } + } + catch (TimeoutException) + { + logger.Trace("Failed to acquire read lock within timeout"); + return Optional.None; } if (count is null) @@ -426,7 +399,7 @@ public class UDPServer private async Task ReceiveHandler(byte[] data, IPEndPoint endPoint, DateTime time) { // 异步锁保护 udpData - await Task.Run(() => + await Task.Run(async () => { try { @@ -438,7 +411,7 @@ public class UDPServer return; } - var udpDataObj = RecordUDPData(data, endPoint, time, Convert.ToInt32(data[1])); + var udpDataObj = await RecordUDPData(data, endPoint, time, Convert.ToInt32(data[1 + 4])); // PrintData(udpDataObj); } catch (Exception e) @@ -448,7 +421,7 @@ public class UDPServer }); } - private UDPData RecordUDPData(byte[] bytes, IPEndPoint remoteEP, DateTime time, int taskID) + private async Task RecordUDPData(byte[] bytes, IPEndPoint remoteEP, DateTime time, int taskID) { var remoteAddress = remoteEP.Address.ToString(); var remotePort = remoteEP.Port; @@ -465,29 +438,30 @@ public class UDPServer var key = $"{remoteAddress}-{taskID}"; - udpDataLock.EnterWriteLock(); try { - var sortedList = udpData.GetOrAdd(key, _ => new SortedList()); - - // 处理相同时间戳的情况，添加微小的时间差 - var uniqueTime = data.Timestamp; - while (sortedList.ContainsKey(uniqueTime)) + using (await udpDataLock.AcquireWriteLockAsync(TimeSpan.FromMilliseconds(5000))) { - logger.Warn( - $"Duplicate timestamp detected for {remoteAddress}:{remotePort} at {uniqueTime}."); - uniqueTime += 1; - } + var sortedList = udpData.GetOrAdd(key, _ => new SortedList()); - sortedList.Add(uniqueTime, data); - // 输出单个数据 - PrintData(data); - // 输出全部数据 - // PrintAllData(); + // 处理相同时间戳的情况，添加微小的时间差 + var uniqueTime = data.Timestamp; + while (sortedList.ContainsKey(uniqueTime)) + { + logger.Warn( + $"Duplicate timestamp detected for {remoteAddress}:{remotePort} at {uniqueTime}."); + uniqueTime += 1; + } + + sortedList.Add(uniqueTime, data); + // 输出单个数据 + PrintData(data); + } } - finally + catch (TimeoutException) { - udpDataLock.ExitWriteLock(); + logger.Error($"Failed to acquire write lock for recording UDP data from {remoteAddress}:{remotePort}"); + throw; } return data; @@ -550,24 +524,26 @@ public class UDPServer /// 将所有数据输出到log中 /// /// void - public void PrintAllData() + public async Task PrintAllDataAsync() { logger.Debug("Ready Data:"); - udpDataLock.EnterReadLock(); try { - foreach (var kvp in udpData) + using (await udpDataLock.AcquireReadLockAsync(TimeSpan.FromMilliseconds(5000))) { - foreach (var data in kvp.Value.Values) + foreach (var kvp in udpData) { - logger.Debug(PrintData(data)); + foreach (var data in kvp.Value.Values) + { + logger.Debug(PrintData(data)); + } } } } - finally + catch (TimeoutException) { - udpDataLock.ExitReadLock(); + logger.Error("Failed to acquire read lock for printing all data"); } } @@ -580,18 +556,14 @@ public class UDPServer public void ClearUDPData(string ipAddr, int taskID) { var key = $"{ipAddr}-{taskID}"; - udpDataLock.EnterWriteLock(); - try + + using (udpDataLock.AcquireWriteLock()) { if (udpData.TryGetValue(key, out var sortedList)) { sortedList.Clear(); } } - finally - { - udpDataLock.ExitWriteLock(); - } // 强制进行ARP刷新，防止后续传输时造成影响 // FlushArpEntry(ipAddr); diff --git a/server/src/WebProtocol.cs b/server/src/WebProtocol.cs index c3aa4cc..55467b1 100644 --- a/server/src/WebProtocol.cs +++ b/server/src/WebProtocol.cs @@ -393,6 +393,16 @@ namespace WebProtocol bytes[8..]); } + /// + /// [TODO:description] + /// + /// [TODO:parameter] + /// [TODO:return] + public static bool IsRecvDataPackage(byte[] bytes) + { + return bytes[4] == (byte)PackSign.RecvData; + } + /// /// 将数据包转化为字节数组 /// @@ -496,6 +506,16 @@ namespace WebProtocol return new RecvRespPackage(timestamp, bytes[5], bytes[6]); } + /// + /// [TODO:description] + /// + /// [TODO:parameter] + /// [TODO:return] + public static bool IsRecvRespPackage(byte[] bytes) + { + return bytes[4] == (byte)PackSign.RecvResp; + } + /// /// 将数据包转化为字节数组 ///