rudpstack.cs

rudp可靠保障得udp传输

//#define CONSOLE_TRACE
//#define CONSOLE_TRACE_MEMORY
//#define STATISTICS
//#define TEST_PACKETLOOSE

using System;
using System.Collections;
using System.Collections.Generic;
using System.Threading;
using System.Diagnostics;
using System.Net;
using System.Net.Sockets;
using System.Runtime.InteropServices;

using Helper.Threading.Collections;

namespace Helper.Net.RUDP
{

	#region Documentations

	/// <summary>
	/// Features :
	/// -------------
	/// 1 - Unreliable packets
	/// 2 - Reliable messaging (You are insured that the message arrives)
	/// 3 - Ordering messages delivery (Allow sending/receiving the messages in the same order)
	/// 4 - KeepAlive feature (to insure that there is no broken connection)
	/// 5 - Has a Tear down message (to close the connection gracefully)
	/// 6 - RendezVous mode and NAT traversal
	/// 7 - Packet fragmentation
	/// 8 - Sliding window (using slot technic)
	/// 9 - Piggy backing (ACK stored in packets)
	/// 10 - Multiples ACKs in a packet (Like SACK for TCP).
	/// 
	/// In progress :
	/// -------------
	/// 1 - Congestion Avoidance Problem :  http://www.cs.utk.edu/~dunigan/tcptour/javis/tcp_congavd.html
	///										http://www.ssfnet.org/Exchange/tcp/tcpTutorialNotes.html
	///										http://www.cs.cmu.edu/afs/cs/academic/class/15441-s06/lectures/L20-TCP-Congestion.pdf
	/// To do:
	/// ------
	/// 2 - Better errors management
	/// 3 - Bandwidth x Delay Product : http://www.cs.utk.edu/~dunigan/tcptour/javis/tcp_slidwin.html
	/// adapt the size of the congestion window. (And the Time out too)
	/// 4 - Slow start : http://www.cs.utk.edu/~dunigan/tcptour/javis/tcp_slowstart.html
	/// 5 - Fast Retransmit Algorithm : http://www.cs.utk.edu/~dunigan/tcptour/javis/tcp_fastrtx.html
	/// 6 - Fast-Recovery Algorithm : http://www.cs.utk.edu/~dunigan/tcptour/javis/tcp_fastrec.html
	/// 
	/// Improvement :
	/// -------------
	/// Optimization : http://gyroweb.inria.fr/~viennot/enseignement/projets/ltcodestcp/ltcodestcphevea.html
	/// Study UDT protocol to improve the used algorithms.
	/// Avoid SIN Attack : http://sewww.epfl.ch/SIC/SA/SPIP/Publications/IMG/pdf/9-3-page13.pdf
	/// http://www.microsoft.com/technet/technetmag/issues/2007/01/CableGuy/default.aspx?loc=fr
	/// Better NAT traversal : http://www.codeproject.com/useritems/STUN_client.asp
	/// 
	/// http://www.tcpipguide.com/free/index.htm
	///
	/// Documentation :
	/// ---------------
	// Test : http://www.pcausa.com/Utilities/pcattcp.htm
	///
	/// UDP server : http://clutch-inc.com/blog/?p=4
	///
	/// TCP: http://www.commentcamarche.net/internet/tcp.php3
	///		http://www.lri.fr/~colette/AnimationFlash/anim_TCP.swf
	///		http://iptables-tutorial.frozentux.net/fr/x1391.html
	///		http://en.wikipedia.org/wiki/Transmission_Control_Protocol
	///		http://www-lsr.imag.fr/users/Andrzej.Duda/PS/2-eme-annee/TCP.pdf
	///
	/// HSTCP: http://www.hep.ucl.ac.uk/~ytl/tcpip/highspeedtcp/hstcp/index.html
	///
	/// Based on TCP:
	/// ------------
	/// http://www.cs.utk.edu/~dunigan/tcptour/
	/// http://abcdrfc.free.fr/rfc-vf/rfc793.html (French)
	/// http://abcdrfc.free.fr/rfc-vf/pdf/rfc793.pdf
	///
	/// Some documentation at :
	/// -----------------------
	/// TCP :
	/// http://www.soi.wide.ad.jp/class/20020032/slides/11/index_32.html
	/// http://www.univ-orleans.fr/sciences/info/ressources/Modules/master2/cci_reseau/ccicours/ch3_couche_transport_1page.pdf
	/// 
	/// UDT :
	/// http://www.cs.uic.edu/~ygu/paper/udt-comnet-v3.pdf
	/// http://udt.sourceforge.net/doc/draft-gg-udt-01.txt
	/// http://www.cs.uic.edu/~ygu/paper/pfldnet04-udt-uic-anl.ppt
	/// </summary>

	#endregion

	#region Articles to improve this implementation

	// On Making TCP More Robust to Packet Reordering
	// http://216.239.59.104/search?q=cache:yL3n-wdnB4AJ:www.icir.org/mallman/papers/tcp-reorder-ccr.ps+tcp+several+ack+packet&hl=fr&ct=clnk&cd=7&gl=be
	// http://www.icir.org/mallman/papers/

	// UDT
	// http://udt.sf.net

	// Hyper Threading
	// http://www.intel.com/cd/ids/developer/asmo-na/eng/194566.htm?prn=Y

	// http://msdn.microsoft.com/msdnmag/issues/05/08/Concurrency/

	#endregion

	static public class RUDPStack
	{

		#region Variables

#if TEST_PACKETLOOSE
		static private Random _looseRandom = new Random();
#endif

		static private readonly bool IsSingleCpuMachine = (Environment.ProcessorCount == 1);

		static internal int UDPHeaderLength = 20 + 8; // 20 bytes for IP , 8 bytes for UDP 
		static internal int RUDPHeaderLength = 47; // 1 + 1 + 1 + 4 + 4 + 4 + 4 * 8

		// The time we wait before sending a keep alive message
		static internal long KeepAliveInterval = 9000000; // 9 seconds

		// The time we wait before a bandwith test
		static internal long BandwidthInterval = 1000000; // 1 second

		// Send ACK every 1 ms
		private const int DelayACKTime = 200000;

		//---- Stack Management
		static private volatile bool _isStackRunning;

		//---- The list of physical sockets, to manage mapping between RUDP and Physical
		static private Dictionary<IPEndPoint, PhysicalSocket> _physicals = new Dictionary<IPEndPoint, PhysicalSocket>(1000);

		//---- The list of logical sockets
		static private List<RUDPSocket> _rudpSockets = new List<RUDPSocket>();
		static private ReaderWriterLock _rudpSocketsLock = new ReaderWriterLock();

		//---- Timers
		static private Thread _protocolControlThread;

		#endregion

		#region Constructor

		static RUDPStack()
		{
			AppDomain.CurrentDomain.ProcessExit += new EventHandler(CurrentDomain_ProcessExit);

			//---- Timers
			_protocolControlThread = new Thread(new ThreadStart(HeartBeat));
			_isStackRunning = true;
			_protocolControlThread.Name = "RUDP Stack Control";
			_protocolControlThread.IsBackground = true;
			_protocolControlThread.Priority = ThreadPriority.Normal;

			// Just below normal to allow UDP stack to run
			_protocolControlThread.Start();
		}

		#endregion

		#region StopStack

		static internal void StopStack()
		{
			//---- Close all the connections
			_rudpSocketsLock.AcquireWriterLock(RUDPSocket.LockTimeOut);

			for (int index = _rudpSockets.Count - 1; index > -1; index--)
				Close(_rudpSockets[index]);

			_rudpSocketsLock.ReleaseWriterLock();

			//---- Stop everything
			_isStackRunning = false;
		}

		#endregion

		#region Instance Factory

		/// <summary>
		/// Static method to get/create a PhysicalSocket
		/// </summary>
		/// <param name="endPoint">The binded end point</param>
		internal static PhysicalSocket GetInstance(IPEndPoint endPoint)
		{
			lock (_physicals)
			{
				PhysicalSocket physical;

				bool isAnyEndPoint = endPoint.Equals(new IPEndPoint(IPAddress.Any, 0));

				// Check if there is already an existing instance
				if (!isAnyEndPoint)
				{
					if (_physicals.TryGetValue(endPoint, out physical))
						return physical;

					// In case no instance exists, create one
					physical = new PhysicalSocket();
					physical.Bind(endPoint);
					RegisterPhysicalSocket(physical);
					_physicals.Add(endPoint, physical);
				}
				else
				{
					physical = new PhysicalSocket();

					while (true)
					{
						int port = new Random().Next(Int16.MaxValue);
						IPAddress localAddress = LocalIPAddress(ProtocolType.IPv4);
						endPoint = new IPEndPoint(localAddress, port);

						// In case no instance exists, create one
						try
						{
							physical.Bind(endPoint);
							break;
						}
						catch { }
					}

					RegisterPhysicalSocket(physical);
					_physicals.Add(endPoint, physical);
				}

				return physical;
			}
		}

		/// <summary>
		/// Release a PhysicalSocket instance and all its resources.
		/// </summary>
		/// <param name="physical">The socket to release</param>
		internal static void ReleaseInstance(PhysicalSocket physical)
		{
			lock (_physicals)
			{
				physical.Dispose();
				_physicals.Remove((IPEndPoint)physical._socket.LocalEndPoint);
			}
		}

		#endregion

		#region RegisterPhysicalSocket

		static private void RegisterPhysicalSocket(PhysicalSocket physical)
		{
			EndPoint tempEndPoint = (EndPoint)physical._canReceiveFromEndPoint;
			physical._socket.BeginReceiveFrom(physical._receiveBuffer, 0, physical._receiveBuffer.Length, SocketFlags.None, ref tempEndPoint, new AsyncCallback(OnEndReceive), physical);
		}

		#endregion

		#region RegisterConnectedSocket

		internal static void RegisterRUDPSocket(RUDPSocket rudp)
		{
			_rudpSocketsLock.AcquireWriterLock(RUDPSocket.LockTimeOut);
			if (!_rudpSockets.Contains(rudp))
				_rudpSockets.Add(rudp);
			_rudpSocketsLock.ReleaseWriterLock();
		}

		internal static void UnregisterRUDPSocket(RUDPSocket rudp)
		{
			_rudpSocketsLock.AcquireWriterLock(RUDPSocket.LockTimeOut);
			_rudpSockets.Remove(rudp);
			_rudpSocketsLock.ReleaseWriterLock();
		}

		#endregion

		#region BeginConnect

		internal static RUDPSocketError BeginConnect(RUDPSocket rudp, int timeOut)
		{
			Trace("Connecting to :" + rudp._remoteEndPoint);

			if (rudp._status == RUDPSocketStatus.Connected)
				return RUDPSocketError.IsConnected;

			if (rudp._status == RUDPSocketStatus.Connecting)
				return RUDPSocketError.AlreadyInProgress;

			//---- Set the status
			rudp.Reset(RUDPSocketStatus.Connecting);

			//---- Register for the stack
			RUDPStack.RegisterRUDPSocket(rudp);

			//---- Send a ping
			if (rudp.IsRendezVousMode)
				PushPacketToSend(rudp, true, RUDPPacketChannel.PingRendezVous, null, 0, 0);
			else
				PushPacketToSend(rudp, true, RUDPPacketChannel.Ping, null, 0, 0);

			return RUDPSocketError.Success;
		}

		#endregion

		#region BeginAccept

		internal static bool BeginAccept(RUDPSocket rudp)
		{
			Trace("Accepting at :" + rudp._remoteEndPoint);

			if (rudp._status != RUDPSocketStatus.Accepting)
				return false;

			//---- Register for the stack
			RUDPStack.RegisterRUDPSocket(rudp);

			return true;
		}

		#endregion

		#region CurrentDomain_ProcessExit

		private static void CurrentDomain_ProcessExit(object sender, EventArgs e)
		{
			// Try to shutdow gracefully, but it is not sure because the process is killed
			_rudpSocketsLock.AcquireWriterLock(RUDPSocket.LockTimeOut);

			for (int index = _rudpSockets.Count - 1; index > -1; index--)
				Shutdown(_rudpSockets[index]);

			_rudpSocketsLock.ReleaseWriterLock();
		}

		#endregion

		#region Close

		/// <summary>
		/// Close the socket. Send the tear down message.
		/// </summary>
		internal static void Close(RUDPSocket rudp)
		{
			if (rudp._status == RUDPSocketStatus.Closed)
				return;

			if (rudp._status == RUDPSocketStatus.Accepting)
			{
				rudp.Reset(RUDPSocketStatus.Closed);
				return;
			}

			AsyncShutdown(rudp);
		}

		private static void AsyncShutdown(RUDPSocket rudp)
		{
			ThreadPool.QueueUserWorkItem(new WaitCallback(AsyncShutdownCB), rudp);
		}

		private static void AsyncShutdownCB(object state)
		{
			Shutdown((RUDPSocket)state);
		}

		#endregion

		#region Shutdown

		internal static void Shutdown(RUDPSocket rudp)
		{
			if (rudp._status == RUDPSocketStatus.Accepting)
			{
				rudp.Reset(RUDPSocketStatus.Closed);
				return;
			}

			if (rudp._status == RUDPSocketStatus.Closed ||
				rudp._isShutingDown)
				return;

			if (rudp._status == RUDPSocketStatus.Closing ||
				rudp._status == RUDPSocketStatus.ClosingACKed)
				return;

			if (rudp._status == RUDPSocketStatus.Connecting)
			{
				rudp.Reset(RUDPSocketStatus.Closed);
				return;
			}

			//---- Send the tear down message

			//-- Update the status
			rudp._isShutingDown = true;
			rudp._status = RUDPSocketStatus.Closing;

			//-- Wait for sending
			while (!rudp._controlWindow.CanSend(0))
			{
				if (rudp._status != RUDPSocketStatus.Closing)
					return;
				Thread.Sleep(100);
			}

			//-- Send the tear down message
			PushPacketToSend(rudp, true, RUDPPacketChannel.TearDown, null, 0, 0);

			//---- Currently closing the connection, wait for the end of the operation
			long startTime = HiResTimer.MicroSeconds;

			//-- Wait until closed
			// Wait until "ClosingACKed"
			// Wait until we have receive the "TearDown" message too and send the ACK
			// Wait until "Time out"
			while (rudp._status == RUDPSocketStatus.Closing &&
					rudp._sendingPackets.Count > 0 &&
					(HiResTimer.MicroSeconds - startTime) < rudp._sto)
			{
				Thread.Sleep(100);
			}

			//---- Set the status as closed
			rudp.Reset(RUDPSocketStatus.Closed);

			//---- Notify
			rudp._physical.OnDisconnected(rudp, RUDPSocketError.Shutdown);
		}

		#endregion

		#region SendPayload

		internal static RUDPSocketError SendPayload(RUDPSocket rudp, byte[] payload, int offset, int payloadLength, bool reliable, RUDPSendIAsyncResult asyncResult)
		{
			// We are no longer active
			if (!_isStackRunning)
				return RUDPSocketError.SystemNotReady;

			//---- Only when connected
			if (rudp._status != RUDPSocketStatus.Connected)
				return RUDPSocketError.NotConnected;

			//---- Fragmentation
			// Full header (IP + UDP + RUDP)
			int MSS = rudp._mtu - (RUDPStack.UDPHeaderLength + RUDPStack.RUDPHeaderLength);

			//---- Fragmentation
			asyncResult.ForceAsyncCall = true;
			FragmentInformation fragments = NewFragmentInformation(rudp, reliable, MSS, payload, offset, payloadLength, asyncResult);
			SendNextFragments(fragments, false);

			return fragments.Error;
		}

		#endregion

		#region SendNextFragments

		static private void SendNextFragments(object state, bool timeOut)
		{
			FragmentInformation fragment = state as FragmentInformation;

			while (fragment.Size > 0)
			{
				int currentLength = Math.Min(fragment.MTU, fragment.Size);

				//---- If cannot send, will send later
				if (fragment.IsReliable)
					if (/*fragment.IsInAsyncThread ||*/ fragment.AsyncResult == null)
					{
						while (!fragment.rudp._controlWindow.CanSend(currentLength))
							fragment.rudp._controlWindow.WaitObject.WaitOne();
					}