引入 KCP,进行基本迁移

This commit is contained in:
2026-03-26 11:25:58 +08:00
parent aeca7a346c
commit 815d5f1f24
24 changed files with 1466 additions and 483 deletions
@@ -1,193 +0,0 @@
using System;
using System.Collections.Generic;
using System.Net;
using System.Net.Sockets;
namespace Network.NetworkTransport
{
public class ClientSession
{
private ITransport _transport;
private IPEndPoint _remote;
public long LastActivityTs { get; private set; }
public uint SendSequenceNumber { get; private set; } = 0;
private int _currentTicks = 0;
private int _nextSendTicks = 0;
private int _sendInterval = 10;
// 重传时间 5s
private long _retransmitTicks = 5000;
// 上层交付
private readonly LinkedList<Packet> _sendQueue = new LinkedList<Packet>();
// 已发送但未确认
private readonly LinkedList<Packet> _sendBuffer = new LinkedList<Packet>();
// 已收到但乱序
private readonly LinkedList<Packet> _receiveBuffer = new LinkedList<Packet>();
// 已收到可交付
private readonly LinkedList<Packet> _receiveQueue = new LinkedList<Packet>();
private bool _hasReceived = false;
private uint _expectedAck = 0;
private readonly object _lockObj = new object();
public ClientSession(ITransport transport, IPEndPoint remote)
{
_transport = transport;
_remote = remote;
LastActivityTs = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds();
}
public uint GetExpectedAck() => _expectedAck;
public void SetSendInterval(int interval) => _sendInterval = interval;
private uint GetNextSendSequence()
{
lock (_lockObj)
{
return SendSequenceNumber++;
}
}
public void Tick(int currentTicks)
{
_currentTicks = currentTicks;
if (_currentTicks >= _nextSendTicks)
{
_nextSendTicks = currentTicks + _sendInterval;
SendPacketInternal();
}
}
public void SendPacket(byte[] data)
{
_sendQueue.AddLast(Packet.CreateDataPacket(GetNextSendSequence(), data));
}
public List<Packet> ReceivePackets()
{
var list = new List<Packet>();
lock (_lockObj)
{
while (_receiveQueue.Count > 0)
{
var packet = _receiveQueue.First.Value;
list.Add(packet);
_receiveQueue.RemoveFirst();
}
}
return list;
}
private void SendPacketInternal()
{
if (_hasReceived)
{
var packet = Packet.CreateAckPacket(_expectedAck);
_sendBuffer.AddLast(packet);
var bytes = packet.ToBytes();
_transport.SendTo(bytes, _remote);
_hasReceived = false;
}
foreach (var packet in _receiveBuffer)
{
if (_currentTicks - packet.Timestamp > _retransmitTicks)
{
var bytes = packet.ToBytes();
_transport.SendTo(bytes, _remote);
}
else break;
}
while (_sendQueue.Count > 0)
{
var packet = _sendQueue.First.Value;
_sendBuffer.AddLast(packet);
var bytes = packet.ToBytes();
_transport.SendTo(bytes, _remote);
}
}
public void ReceivePacketsInternal(Packet packet)
{
uint seq = packet.SequenceNumber;
// 是否是按序到达的包
if (seq == _expectedAck)
{
_receiveQueue.AddLast(packet);
while (_receiveBuffer.Count > 0)
{
var pendingPacket = _receiveBuffer.First.Value;
if (seq != pendingPacket.SequenceNumber) break;
seq++;
_receiveQueue.AddLast(pendingPacket);
_receiveBuffer.RemoveFirst();
}
_expectedAck = seq + 1;
_hasReceived = true;
}
// 将包按顺序追加在 receivingPackets 后面
else
{
var firstNode = _receiveBuffer.First;
while (firstNode.Next != null)
{
if (firstNode.Value.SequenceNumber > seq)
{
var node = new LinkedListNode<Packet>(packet);
_receiveBuffer.AddBefore(firstNode, node);
break;
}
firstNode = firstNode.Next;
}
if (firstNode == null) _receiveBuffer.AddLast(packet);
}
}
public bool TryProcessReceiveSequence(uint sequenceNumber, out bool shouldDeliver)
{
lock (_lockObj)
{
LastActivityTs = DateTime.Now;
if (sequenceNumber == _expectedAck)
{
_expectedAck++;
_receivedSequences.Add(sequenceNumber);
shouldDeliver = true;
return true;
}
else if (sequenceNumber < _expectedAck)
{
shouldDeliver = false;
return _receivedSequences.Contains(sequenceNumber);
}
else
{
shouldDeliver = false;
return false;
}
}
}
}
}
@@ -1,11 +0,0 @@
fileFormatVersion: 2
guid: ec6c25bc42967db499742dfa355380b7
MonoImporter:
externalObjects: {}
serializedVersion: 2
defaultReferences: []
executionOrder: 0
icon: {instanceID: 0}
userData:
assetBundleName:
assetBundleVariant:
@@ -6,6 +6,7 @@ namespace Network.NetworkTransport
{
public interface ITransport
{
void Send(byte[] data);
void SendTo(byte[] data, IPEndPoint target);
void SendToAll(byte[] data);
event Action<byte[], IPEndPoint> OnReceive;
@@ -1,63 +0,0 @@
using System;
using System.Linq;
using UnityEngine;
namespace Network.NetworkTransport
{
public enum PacketType : byte
{
Data = 1,
Ack = 2,
}
public struct Packet
{
public PacketType Type;
public uint SequenceNumber;
public byte[] Data;
public long Timestamp;
public byte[] ToBytes()
{
var result = new byte[1 + 4 + 8 + Data.Length];
result[0] = (byte)Type;
BitConverter.GetBytes(SequenceNumber).CopyTo(result, 1);
BitConverter.GetBytes(Timestamp).CopyTo(result, 5);
Data.CopyTo(result, 13);
return result;
}
public static Packet FromBytes(byte[] data)
{
return new Packet
{
Type = (PacketType)data[0],
SequenceNumber = BitConverter.ToUInt32(data, 1),
Timestamp = BitConverter.ToInt64(data, 5),
Data = new ArraySegment<byte>(data, 5, data.Length - 5).ToArray()
};
}
public static Packet CreateDataPacket(uint seqNum, byte[] data)
{
return new Packet
{
Type = PacketType.Data,
SequenceNumber = seqNum,
Data = data,
Timestamp = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds()
};
}
public static Packet CreateAckPacket(uint seqNum)
{
return new Packet
{
Type = PacketType.Ack,
SequenceNumber = seqNum,
Data = Array.Empty<byte>(),
Timestamp = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds()
};
}
}
}
@@ -1,11 +0,0 @@
fileFormatVersion: 2
guid: b84a2cb7ffe3cd14180358559e526dbe
MonoImporter:
externalObjects: {}
serializedVersion: 2
defaultReferences: []
executionOrder: 0
icon: {instanceID: 0}
userData:
assetBundleName:
assetBundleVariant:
@@ -1,8 +1,7 @@
using System;
using System.Collections.Generic;
using System.Collections.Concurrent;
using System.Net;
using System.Net.Sockets;
using System.Threading;
using System.Threading.Tasks;
namespace Network.NetworkTransport
@@ -13,93 +12,103 @@ namespace Network.NetworkTransport
private readonly IPEndPoint _defaultRemoteEndPoint;
private readonly bool _isServer;
private readonly List<ClientSession> _sessions = new();
// Stage one keeps this class name for compatibility while collapsing it to plain UDP.
private readonly ConcurrentDictionary<string, IPEndPoint> _knownRemoteEndPoints = new();
private readonly Timer _retransmitTimer;
private readonly Timer _cleanupTimer;
//TODO: volatile 关键字
private volatile bool _isRunning;
// 配置参数
private const int RetransmitTimeoutMs = 1000;
private const int SessionTimeoutMs = 30000;
private const int MaxRetransmitAttempts = 5;
public event Action<byte[], IPEndPoint> OnReceive;
private Task _receiveTask;
private Task _receiveTask = Task.CompletedTask;
// 构造函数——服务端模式
public ReliableUdpTransport(int listenPort)
{
_client = new UdpClient(listenPort);
_isServer = true;
_retransmitTimer = new Timer(CheckRetransmit, null, 100, 100);
_cleanupTimer = new Timer(CleanupSessions, null, 5000, 5000);
Console.WriteLine($"[Transport] 服务端模式,监听端口: {listenPort}");
}
// 构造函数——客户端模式
public ReliableUdpTransport(string serverIP, int serverPort)
{
_client = new UdpClient(0);
_defaultRemoteEndPoint = new IPEndPoint(IPAddress.Parse(serverIP), serverPort);
_isServer = false;
_retransmitTimer = new Timer(CheckRetransmit, null, 100, 100);
_cleanupTimer = new Timer(CleanupSessions, null, 5000, 5000);
Console.WriteLine($"[Transport] 客户端模式,目标: {_defaultRemoteEndPoint}");
}
public async Task StartAsync()
{
_sessions.Clear();
if (_isRunning)
{
return;
}
_knownRemoteEndPoints.Clear();
_isRunning = true;
Console.WriteLine("[Transport] 传输层启动");
// 开始接收数据
_receiveTask = ReceiveLoop();
await Task.Delay(100); // 给接收循环一点启动时间
}
public void Tick()
{
foreach (var session in _sessions)
{
session.Tick(DateTime.UtcNow.Millisecond);
}
await Task.Yield();
}
public void Stop()
{
_isRunning = false;
_retransmitTimer.Dispose();
_cleanupTimer.Dispose();
_client.Close();
_sessions.Clear();
Console.WriteLine("[Transport] 传输层停止");
}
public async void SendTo(Packet packet, IPEndPoint target)
{
if (!_isRunning)
{
return;
}
var bytes = packet.ToBytes();
await _client.SendAsync(bytes, bytes.Length, target);
Console.WriteLine($"[Transport] 发送数据包到 {target}");
_isRunning = false;
_client.Close();
_knownRemoteEndPoints.Clear();
Console.WriteLine("[Transport] 传输层停止");
}
public void Send(byte[] data)
{
if (_defaultRemoteEndPoint == null)
{
throw new InvalidOperationException("Default remote endpoint is not configured.");
}
SendTo(data, _defaultRemoteEndPoint);
}
public void SendTo(byte[] data, IPEndPoint target)
{
if (data == null)
{
throw new ArgumentNullException(nameof(data));
}
if (target == null)
{
throw new ArgumentNullException(nameof(target));
}
EnsureRunning();
RememberRemote(target);
_client.Send(data, data.Length, target);
Console.WriteLine($"[Transport] 发送数据到 {target}");
}
public void SendToAll(byte[] data)
{
foreach (var session in _sessions)
if (data == null)
{
session.SendPacket(data);
throw new ArgumentNullException(nameof(data));
}
EnsureRunning();
if (!_isServer)
{
throw new InvalidOperationException("SendToAll is only supported in server mode.");
}
foreach (var remoteEndPoint in _knownRemoteEndPoints.Values)
{
_client.Send(data, data.Length, remoteEndPoint);
Console.WriteLine($"[Transport] 广播数据到 {remoteEndPoint}");
}
}
@@ -110,20 +119,16 @@ namespace Network.NetworkTransport
try
{
var result = await _client.ReceiveAsync();
var packet = Packet.FromBytes(result.Buffer);
if (packet.Type == PacketType.Data)
{
HandleDataPacket(packet, result.RemoteEndPoint);
}
else if (packet.Type == PacketType.Ack)
{
HandleAckPacket(packet, result.RemoteEndPoint);
}
RememberRemote(result.RemoteEndPoint);
OnReceive?.Invoke(result.Buffer, result.RemoteEndPoint);
}
catch (ObjectDisposedException)
catch (ObjectDisposedException) when (!_isRunning)
{
return; // 正常关闭
return;
}
catch (SocketException) when (!_isRunning)
{
return;
}
catch (Exception e)
{
@@ -132,154 +137,22 @@ namespace Network.NetworkTransport
}
}
private void HandleDataPacket(Packet packet, IPEndPoint senderEndPoint)
{
var session = GetOrCreateSession(senderEndPoint);
Console.WriteLine(
$"[Transport] 收到数据包从{senderEndPoint} SeqNum={packet.SequenceNumber}, DataLen={packet.Data.Length}");
// 发送ACK
var ackPacket = Packet.CreateAckPacket(packet.SequenceNumber);
SendPacketTo(ackPacket, senderEndPoint);
Console.WriteLine($"[Transport] 发送ACK 到 {senderEndPoint} SeqNum={packet.SequenceNumber}");
// 检查是否应该交付
if (session.TryProcessReceiveSequence(packet.SequenceNumber, out bool shouldDeliver))
{
if (shouldDeliver)
{
OnReceive?.Invoke(packet.Data, senderEndPoint);
Console.WriteLine($"[Transport] 交付数据包从 {senderEndPoint} SeqNum={packet.SequenceNumber}");
}
else
{
Console.WriteLine($"[Transport] 重复包从 {senderEndPoint} SeqNum={packet.SequenceNumber},忽略");
}
}
else
{
// 乱序到达,暂存(简化处理:直接丢弃,依赖重传)
Console.WriteLine($"[Transport] 乱序包从 {senderEndPoint} SeqNum={packet.SequenceNumber},丢弃");
}
}
private void HandleAckPacket(Packet packet, IPEndPoint senderEndPoint)
{
var session = GetOrCreateSession(senderEndPoint);
Console.WriteLine($"[Transport] 收到ACK从 {senderEndPoint} SeqNum={packet.SequenceNumber}");
if (session.PendingAcks.TryRemove(packet.SequenceNumber, out _))
{
Console.WriteLine($"[Transport] 确认包到 {senderEndPoint} SeqNum={packet.SequenceNumber}");
}
}
private ClientSession GetOrCreateSession(IPEndPoint endPoint)
{
string key = endPoint.ToString();
return _sessions.GetOrAdd(key, _ =>
{
var session = new ClientSession(endPoint);
Console.WriteLine($"创建新会话:{endPoint}");
return session;
});
}
private void CheckRetransmit(object state)
private void EnsureRunning()
{
if (!_isRunning)
{
throw new InvalidOperationException("Transport has not been started.");
}
}
private void RememberRemote(IPEndPoint remoteEndPoint)
{
if (remoteEndPoint == null)
{
return;
}
var now = DateTime.Now;
var toRetransmit = new List<(IPEndPoint target, uint seqNum, Packet packet)>();
foreach (var sessionKvp in _sessions)
{
var session = sessionKvp.Value;
foreach (var ackKvp in session.PendingAcks)
{
var timeSinceLastSend = now - ackKvp.Value.sendTime;
if (timeSinceLastSend.TotalMilliseconds > RetransmitTimeoutMs)
{
toRetransmit.Add((session.EndPoint, ackKvp.Key, ackKvp.Value.packet));
}
}
}
foreach (var (target, seqNum, packet) in toRetransmit)
{
var session = GetOrCreateSession(target);
if (session.PendingAcks.ContainsKey(seqNum))
{
// 更新发送时间
session.PendingAcks[seqNum] = (packet, now);
SendPacketTo(packet, target);
Console.WriteLine($"[Transport] 重传包到 {target} SeqNum={seqNum}");
}
}
}
private void CleanupSessions(object state)
{
if (!_isRunning)
{
return;
}
var now = DateTime.Now;
var toRemove = new List<string>();
foreach (var sessionKvp in _sessions)
{
var session = sessionKvp.Value;
var timeSinceLastActivity = now - session.LastActivity;
if (timeSinceLastActivity.TotalMilliseconds > SessionTimeoutMs)
{
toRemove.Add(sessionKvp.Key);
}
}
foreach (string key in toRemove)
{
if (_sessions.TryRemove(key, out var session))
{
Console.WriteLine($"[Transport] 清理超时会话:{session.EndPoint}");
}
}
if (_isServer)
{
PrintSessionInfo();
}
}
private async void SendPacketTo(Packet packet, IPEndPoint endPoint)
{
try
{
var data = packet.ToBytes();
await _client.SendAsync(data, data.Length, endPoint);
}
catch (Exception e)
{
Console.WriteLine($"[Transport] 发送错误:{e.Message}");
}
}
private void PrintSessionInfo()
{
Console.WriteLine($"当前活跃会话数:{_sessions.Count}");
foreach (var sessionKvp in _sessions)
{
var session = sessionKvp.Value;
Console.WriteLine(
$" 会话:{session.EndPoint},发送SeqNum:{session.SendSequenceNumber},期望接收:{session.GetExpectedAck()},待确认: {session.PendingAcks.Count}");
}
_knownRemoteEndPoints[remoteEndPoint.ToString()] = remoteEndPoint;
}
}
}
}