crysome/Crysome.Common/Crysome.Common.Network/RudpChannel.cs

using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Net;
using System.Net.Sockets;
using System.Threading;
using System.Threading.Tasks;

namespace Crysome.Common.Network;

public sealed class RudpChannel : IDisposable
{
	private sealed class Pending
	{
		public uint Seq;

		public byte[][] Datagrams;

		public DateTime SentAt;

		public DateTime LastSent;

		public int Retries;
	}

	private sealed class FragBuf
	{
		public byte[][] Frags;

		public int RecvCount;

		public DateTime Created = DateTime.UtcNow;

		public bool IsComplete
		{
			get
			{
				if (Frags != null)
				{
					return RecvCount >= Frags.Length;
				}
				return false;
			}
		}
	}

	public const int HeaderSize = 19;

	private const int WireMtuSafePayload = 1200;

	public const byte F_RELIABLE = 1;

	public const byte F_UNRELIABLE = 2;

	public const byte F_ACK = 4;

	public const byte F_SYN = 8;

	public const byte F_SYNACK = 16;

	public const byte F_FIN = 32;

	public const byte F_KEEPALIVE = 64;

	private const int InitRto = 500;

	private const int MaxRto = 16000;

	private const int MaxRetries = 15;

	private const int KeepaliveMs = 5000;

	private const int TimeoutMs = 30000;

	private const int MaxInFlight = 64;

	private readonly UdpClient _socket;

	private readonly bool _ownsSocket;

	private uint _sendSeq;

	private uint _recvSeq;

	private uint _unrelSendSeq;

	private uint _unrelLastSeq;

	private readonly object _seqLock = new object();

	private readonly ConcurrentDictionary<uint, Pending> _pending = new ConcurrentDictionary<uint, Pending>();

	private readonly SortedDictionary<uint, byte[]> _ooo = new SortedDictionary<uint, byte[]>();

	private readonly object _oooLock = new object();

	private readonly ConcurrentDictionary<uint, FragBuf> _fragBufs = new ConcurrentDictionary<uint, FragBuf>();

	private readonly ConcurrentDictionary<uint, FragBuf> _relFragBufs = new ConcurrentDictionary<uint, FragBuf>();

	private int _rtoMs = 500;

	private bool _rtoDoubledThisCycle;

	private DateTime _lastActivity = DateTime.UtcNow;

	private CancellationTokenSource _cts;

	private bool _disposed;

	private readonly SemaphoreSlim _sendWindow = new SemaphoreSlim(64, 64);

	private readonly object _statsLock = new object();

	private double _srtt = 100.0;

	private double _rttvar = 50.0;

	public uint SessionId { get; private set; }

	public IPEndPoint Remote { get; private set; }

	public bool IsAlive
	{
		get
		{
			if (!_disposed)
			{
				return (DateTime.UtcNow - _lastActivity).TotalMilliseconds < 30000.0;
			}
			return false;
		}
	}

	public double RttMs { get; private set; } = 100.0;

	public double LossRate { get; private set; }

	public long BytesSent { get; private set; }

	public long BytesRecv { get; private set; }

	public long PktsSent { get; private set; }

	public long PktsLost { get; private set; }

	public event Action<byte[]> ReliableReceived;

	public event Action<byte, byte[]> UnreliableFrameReceived;

	public event Action Disconnected;

	public event Action<double> CongestionChanged;

	public RudpChannel(uint sessionId, IPEndPoint remote, int localPort = 0)
	{
		SessionId = sessionId;
		Remote = remote;
		_socket = new UdpClient(localPort);
		_socket.Client.ReceiveBufferSize = 8388608;
		_socket.Client.SendBufferSize = 8388608;
		_ownsSocket = true;
	}

	public RudpChannel(uint sessionId, IPEndPoint remote, UdpClient sharedSocket)
	{
		SessionId = sessionId;
		Remote = remote;
		_socket = sharedSocket;
		_ownsSocket = false;
	}

	public void Start()
	{
		_cts = new CancellationTokenSource();
		if (_ownsSocket)
		{
			Task.Run(delegate
			{
				OwnedReceiveLoop(_cts.Token);
			});
		}
		Task.Run(delegate
		{
			RetransmitLoop(_cts.Token);
		});
		Task.Run(delegate
		{
			KeepaliveLoop(_cts.Token);
		});
	}

	public void Dispose()
	{
		if (_disposed)
		{
			return;
		}
		_disposed = true;
		try
		{
			_cts?.Cancel();
		}
		catch
		{
		}
		if (_ownsSocket)
		{
			try
			{
				_socket?.Close();
			}
			catch
			{
			}
		}
		int count = _pending.Count;
		_pending.Clear();
		for (int i = 0; i < count + 64; i++)
		{
			try
			{
				_sendWindow.Release();
			}
			catch
			{
			}
		}
	}

	public void SendReliable(byte[] payload)
	{
		if (_disposed)
		{
			return;
		}
		_sendWindow.Wait();
		if (_disposed)
		{
			_sendWindow.Release();
			return;
		}
		int num = Math.Max(1, (payload.Length + 1200 - 1) / 1200);
		uint num2;
		lock (_seqLock)
		{
			num2 = ++_sendSeq;
		}
		byte[][] array = new byte[num][];
		for (ushort num3 = 0; num3 < num; num3++)
		{
			int num4 = num3 * 1200;
			int payLen = Math.Min(1200, payload.Length - num4);
			array[num3] = BuildDatagram(1, num2, 0u, num3, (ushort)num, 0, payload, num4, payLen);
		}
		Pending value = new Pending
		{
			Seq = num2,
			Datagrams = array,
			SentAt = DateTime.UtcNow,
			LastSent = DateTime.UtcNow
		};
		_pending[num2] = value;
		bool flag = num > 4;
		byte[][] array2 = array;
		foreach (byte[] dg in array2)
		{
			RawSend(dg);
			if (flag)
			{
				Thread.Sleep(2);
			}
		}
		PktsSent++;
	}

	public void SendUnreliable(byte packetTypeId, byte[] payload)
	{
		if (_disposed)
		{
			return;
		}
		uint seqNum;
		lock (_seqLock)
		{
			seqNum = ++_unrelSendSeq;
		}
		int num = Math.Max(1, (payload.Length + 1200 - 1) / 1200);
		for (ushort num2 = 0; num2 < num; num2++)
		{
			int num3 = num2 * 1200;
			int payLen = Math.Min(1200, payload.Length - num3);
			byte[] dg = BuildDatagram(2, seqNum, 0u, num2, (ushort)num, packetTypeId, payload, num3, payLen);
			RawSend(dg);
			if (num > 16 && (num2 & 0xF) == 15 && num2 < num - 1)
			{
				Thread.Sleep(1);
			}
		}
	}

	public void SendSyn()
	{
		byte[] dg = BuildDatagram(8, 0u, 0u, 0, 1, 0, new byte[0], 0, 0);
		RawSend(dg);
	}

	public void SendSynAck()
	{
		byte[] dg = BuildDatagram(16, SessionId, 0u, 0, 1, 0, new byte[0], 0, 0);
		RawSend(dg);
	}

	public void SendFin()
	{
		byte[] dg = BuildDatagram(32, 0u, 0u, 0, 1, 0, new byte[0], 0, 0);
		try
		{
			RawSend(dg);
		}
		catch
		{
		}
	}

	public void FeedDatagram(byte[] datagram)
	{
		ProcessDatagram(datagram);
	}

	private async void OwnedReceiveLoop(CancellationToken ct)
	{
		while (!ct.IsCancellationRequested && !_disposed)
		{
			try
			{
				ProcessDatagram((await _socket.ReceiveAsync()).Buffer);
			}
			catch (ObjectDisposedException)
			{
				break;
			}
			catch
			{
			}
		}
	}

	private void ProcessDatagram(byte[] dg)
	{
		if (dg == null || dg.Length < 19)
		{
			return;
		}
		_lastActivity = DateTime.UtcNow;
		BytesRecv += dg.Length;
		BitConverter.ToUInt32(dg, 0);
		byte b = dg[4];
		uint seq = BitConverter.ToUInt32(dg, 5);
		uint ackSeq = BitConverter.ToUInt32(dg, 9);
		ushort fragIdx = BitConverter.ToUInt16(dg, 13);
		ushort fragCount = BitConverter.ToUInt16(dg, 15);
		byte typeId = dg[17];
		int payLen = dg.Length - 19;
		if ((b & 4) != 0)
		{
			HandleAck(ackSeq);
		}
		else if ((b & 0x40) == 0)
		{
			if ((b & 0x20) != 0)
			{
				TriggerDisconnect();
			}
			else if ((b & 1) != 0)
			{
				HandleReliableData(seq, fragIdx, fragCount, dg, payLen);
			}
			else if ((b & 2) != 0)
			{
				HandleUnreliableData(seq, typeId, fragIdx, fragCount, dg, payLen);
			}
		}
	}

	private void HandleAck(uint ackSeq)
	{
		if (_pending.TryRemove(ackSeq, out var value))
		{
			double totalMilliseconds = (DateTime.UtcNow - value.SentAt).TotalMilliseconds;
			UpdateRtt(totalMilliseconds);
			try
			{
				_sendWindow.Release();
			}
			catch
			{
			}
		}
	}

	private void HandleReliableData(uint seq, ushort fragIdx, ushort fragCount, byte[] dg, int payLen)
	{
		int num = 19;
		if (payLen < 0 || num + payLen > dg.Length)
		{
			return;
		}
		byte[] array = new byte[payLen];
		if (payLen > 0)
		{
			Buffer.BlockCopy(dg, num, array, 0, payLen);
		}
		if (fragCount > 1)
		{
			FragBuf orAdd = _relFragBufs.GetOrAdd(seq, (uint _) => new FragBuf
			{
				Frags = new byte[fragCount][]
			});
			if (fragIdx >= orAdd.Frags.Length || orAdd.Frags[fragIdx] != null)
			{
				return;
			}
			orAdd.Frags[fragIdx] = array;
			Interlocked.Increment(ref orAdd.RecvCount);
			if (!orAdd.IsComplete)
			{
				return;
			}
			_relFragBufs.TryRemove(seq, out var _);
			SendAck(seq);
			int num2 = 0;
			byte[][] frags = orAdd.Frags;
			foreach (byte[] array2 in frags)
			{
				if (array2 != null)
				{
					num2 += array2.Length;
				}
			}
			byte[] array3 = new byte[num2];
			int num4 = 0;
			frags = orAdd.Frags;
			foreach (byte[] array4 in frags)
			{
				if (array4 != null)
				{
					Buffer.BlockCopy(array4, 0, array3, num4, array4.Length);
					num4 += array4.Length;
				}
			}
			array = array3;
		}
		else
		{
			SendAck(seq);
		}
		lock (_oooLock)
		{
			uint num5 = _recvSeq + 1;
			if (seq == num5 || _recvSeq == 0)
			{
				_recvSeq = seq;
				DispatchReliable(array);
				while (true)
				{
					uint num6 = _recvSeq + 1;
					if (_ooo.TryGetValue(num6, out var value2))
					{
						_ooo.Remove(num6);
						_recvSeq = num6;
						DispatchReliable(value2);
						continue;
					}
					break;
				}
			}
			else if ((int)(seq - num5) > 0 && !_ooo.ContainsKey(seq))
			{
				_ooo[seq] = array;
			}
		}
	}

	private void HandleUnreliableData(uint seq, byte typeId, ushort fragIdx, ushort fragCount, byte[] dg, int payLen)
	{
		lock (_seqLock)
		{
			if (_unrelLastSeq != 0 && (int)(seq - _unrelLastSeq) <= 0)
			{
				return;
			}
		}
		int num = 19;
		if (payLen < 0 || num + payLen > dg.Length)
		{
			return;
		}
		byte[] array = new byte[payLen];
		if (payLen > 0)
		{
			Buffer.BlockCopy(dg, num, array, 0, payLen);
		}
		if (fragCount <= 1)
		{
			lock (_seqLock)
			{
				_unrelLastSeq = seq;
			}
			DispatchUnreliable(typeId, array);
			return;
		}
		FragBuf orAdd = _fragBufs.GetOrAdd(seq, (uint _) => new FragBuf
		{
			Frags = new byte[fragCount][]
		});
		if (fragIdx >= orAdd.Frags.Length || orAdd.Frags[fragIdx] != null)
		{
			return;
		}
		orAdd.Frags[fragIdx] = array;
		Interlocked.Increment(ref orAdd.RecvCount);
		if (!orAdd.IsComplete)
		{
			return;
		}
		_fragBufs.TryRemove(seq, out var _);
		lock (_seqLock)
		{
			if (_unrelLastSeq != 0 && (int)(seq - _unrelLastSeq) <= 0)
			{
				return;
			}
			_unrelLastSeq = seq;
		}
		int num2 = 0;
		byte[][] frags = orAdd.Frags;
		foreach (byte[] array2 in frags)
		{
			if (array2 != null)
			{
				num2 += array2.Length;
			}
		}
		byte[] array3 = new byte[num2];
		int num4 = 0;
		frags = orAdd.Frags;
		foreach (byte[] array4 in frags)
		{
			if (array4 != null)
			{
				Buffer.BlockCopy(array4, 0, array3, num4, array4.Length);
				num4 += array4.Length;
			}
		}
		DispatchUnreliable(typeId, array3);
	}

	private void DispatchReliable(byte[] data)
	{
		Task.Run(delegate
		{
			try
			{
				this.ReliableReceived?.Invoke(data);
			}
			catch
			{
			}
		});
	}

	private void DispatchUnreliable(byte typeId, byte[] data)
	{
		Task.Run(delegate
		{
			try
			{
				this.UnreliableFrameReceived?.Invoke(typeId, data);
			}
			catch
			{
			}
		});
	}

	private async void RetransmitLoop(CancellationToken ct)
	{
		while (!ct.IsCancellationRequested && !_disposed)
		{
			await Task.Delay(50, ct).ContinueWith(delegate
			{
			});
			DateTime utcNow = DateTime.UtcNow;
			FragBuf value;
			foreach (KeyValuePair<uint, FragBuf> fragBuf in _fragBufs)
			{
				if ((utcNow - fragBuf.Value.Created).TotalSeconds > 10.0)
				{
					_fragBufs.TryRemove(fragBuf.Key, out value);
				}
			}
			foreach (KeyValuePair<uint, FragBuf> relFragBuf in _relFragBufs)
			{
				if ((utcNow - relFragBuf.Value.Created).TotalSeconds > 60.0)
				{
					_relFragBufs.TryRemove(relFragBuf.Key, out value);
				}
			}
			_rtoDoubledThisCycle = false;
			foreach (KeyValuePair<uint, Pending> item in _pending)
			{
				Pending value2 = item.Value;
				if ((utcNow - value2.LastSent).TotalMilliseconds < (double)_rtoMs)
				{
					continue;
				}
				if (value2.Retries >= 15)
				{
					if (_pending.TryRemove(value2.Seq, out var _))
					{
						PktsLost++;
						UpdateLossRate();
						try
						{
							_sendWindow.Release();
						}
						catch
						{
						}
					}
					continue;
				}
				value2.Retries++;
				value2.LastSent = utcNow;
				if (!_rtoDoubledThisCycle)
				{
					_rtoMs = Math.Min(_rtoMs * 2, 16000);
					_rtoDoubledThisCycle = true;
				}
				bool flag = value2.Datagrams.Length > 4;
				byte[][] datagrams = value2.Datagrams;
				foreach (byte[] dg in datagrams)
				{
					RawSend(dg);
					if (flag)
					{
						Thread.Sleep(1);
					}
				}
			}
			UpdateCongestionMetric();
			if ((utcNow - _lastActivity).TotalMilliseconds > 30000.0)
			{
				TriggerDisconnect();
			}
		}
	}

	private async void KeepaliveLoop(CancellationToken ct)
	{
		while (!ct.IsCancellationRequested && !_disposed)
		{
			await Task.Delay(5000, ct).ContinueWith(delegate
			{
			});
			if (!_disposed)
			{
				try
				{
					byte[] dg = BuildDatagram(64, 0u, 0u, 0, 1, 0, new byte[0], 0, 0);
					RawSend(dg);
				}
				catch
				{
				}
				continue;
			}
			break;
		}
	}

	private void SendAck(uint seq)
	{
		byte[] dg = BuildDatagram(4, 0u, seq, 0, 1, 0, new byte[0], 0, 0);
		try
		{
			RawSend(dg);
		}
		catch
		{
		}
	}

	private void UpdateRtt(double sample)
	{
		lock (_statsLock)
		{
			_rttvar = 0.75 * _rttvar + 0.25 * Math.Abs(_srtt - sample);
			_srtt = 0.875 * _srtt + 0.125 * sample;
			RttMs = _srtt;
			_rtoMs = Math.Max(200, Math.Min(16000, (int)(_srtt + 4.0 * _rttvar)));
		}
	}

	private void UpdateLossRate()
	{
		double val = ((PktsSent > 0) ? ((double)PktsLost / (double)PktsSent) : 0.0);
		LossRate = Math.Min(1.0, val);
	}

	private void UpdateCongestionMetric()
	{
		UpdateLossRate();
		double num = Math.Min(1.0, RttMs / 2000.0);
		double lossRate = LossRate;
		double obj = Math.Min(1.0, num * 0.5 + lossRate * 0.5);
		try
		{
			this.CongestionChanged?.Invoke(obj);
		}
		catch
		{
		}
	}

	private void TriggerDisconnect()
	{
		if (_disposed)
		{
			return;
		}
		Dispose();
		try
		{
			this.Disconnected?.Invoke();
		}
		catch
		{
		}
	}

	private byte[] BuildDatagram(byte flags, uint seqNum, uint ackNum, ushort fragIdx, ushort fragCount, byte packetTypeId, byte[] payload, int payOffset, int payLen)
	{
		byte[] array = new byte[19 + payLen];
		Buffer.BlockCopy(BitConverter.GetBytes(SessionId), 0, array, 0, 4);
		array[4] = flags;
		Buffer.BlockCopy(BitConverter.GetBytes(seqNum), 0, array, 5, 4);
		Buffer.BlockCopy(BitConverter.GetBytes(ackNum), 0, array, 9, 4);
		Buffer.BlockCopy(BitConverter.GetBytes(fragIdx), 0, array, 13, 2);
		Buffer.BlockCopy(BitConverter.GetBytes(fragCount), 0, array, 15, 2);
		array[17] = packetTypeId;
		array[18] = 0;
		if (payLen > 0)
		{
			Buffer.BlockCopy(payload, payOffset, array, 19, payLen);
		}
		return array;
	}

	private void RawSend(byte[] dg)
	{
		if (_disposed || dg == null)
		{
			return;
		}
		try
		{
			_socket.Send(dg, dg.Length, Remote);
			BytesSent += dg.Length;
		}
		catch
		{
		}
	}

	public static bool ParseHeader(byte[] dg, int len, out uint sessionId, out byte flags, out uint seqNum, out uint ackNum, out ushort fragIdx, out ushort fragCount, out byte packetTypeId)
	{
		sessionId = 0u;
		flags = 0;
		seqNum = 0u;
		ackNum = 0u;
		fragIdx = 0;
		fragCount = 0;
		packetTypeId = 0;
		if (dg == null || len < 19)
		{
			return false;
		}
		sessionId = BitConverter.ToUInt32(dg, 0);
		flags = dg[4];
		seqNum = BitConverter.ToUInt32(dg, 5);
		ackNum = BitConverter.ToUInt32(dg, 9);
		fragIdx = BitConverter.ToUInt16(dg, 13);
		fragCount = BitConverter.ToUInt16(dg, 15);
		packetTypeId = dg[17];
		return true;
	}
}