rudpstack.cs

rudp可靠保障得udp传输

					else
					{
						if (!fragment.rudp._controlWindow.CanSend(currentLength))
						{
							fragment.AsyncResult.ForceAsyncCall = false;
							fragment.IsInAsyncThread = true; // Next time, just wait
							ThreadPool.UnsafeRegisterWaitForSingleObject(fragment.rudp._controlWindow.WaitObject,
																		new WaitOrTimerCallback(SendNextFragments),
																		fragment,
																		-1,
																		true);
							return;
						}
					}

				//---- Send
				if (!PushPacketToSend(fragment.rudp,
									fragment.IsReliable,
									RUDPPacketChannel.UserPacket,
									fragment.Payload,
									fragment.Offset,
									currentLength))
				{
					fragment.Error = RUDPSocketError.SocketError;
					if (fragment.AsyncResult != null)
						OnSocketUnhandledError(fragment.rudp, fragment.Error, fragment.AsyncResult);

					ReleaseFragmentInformation(fragment);
					return;
				}

				fragment.Size -= currentLength;
				fragment.Offset += currentLength;
			}

			//---- End of send
			if (fragment.AsyncResult != null)
				fragment.rudp._physical.OnEndSend(fragment.rudp, fragment.AsyncResult);

			ReleaseFragmentInformation(fragment);
		}

		#endregion

		#region PushPacketToSend

		internal static bool PushPacketToSend(RUDPSocket rudp,
											bool reliablePacket,
											RUDPPacketChannel channel,
											byte[] payload,
											int offset,
											int payloadLength)
		{
			int packetId = -1;
			if (reliablePacket)
				packetId = Interlocked.Increment(ref rudp._ougoingPacketId);

			//---- Get the SACKs
			SACKSlot slot1 = null;
			SACKSlot slot2 = null;
			SACKSlot slot3 = null;
			SACKSlot slot4 = null;
			rudp._sackWindow.GetSLACKSlots(out slot1, out slot2, out slot3, out slot4);

			//---- Copy the payload to send
			byte[] rudpPayload = MakePacketPayload(rudp, packetId, channel, slot1, slot2, slot3, slot4, payload, offset, payloadLength);

			//---- Create a packet
			RUDPOutgoingPacket packet = NewOutgoingPacket(packetId, rudp._sequence, rudpPayload, channel);

			if (reliablePacket)
			{
				//---- Notify the control window
				rudp._controlWindow.OnSend(packetId, rudp._sequence, payloadLength);

				//---- Increment sequence number
				Interlocked.Exchange(ref rudp._sequence, rudp._sequence + payloadLength);
			}

			//---- In the "resend list"
			if (reliablePacket)
			{
				rudp._sendingPacketsLock.AcquireWriterLock(RUDPSocket.LockTimeOut);
				rudp._sendingPackets.Add(packet);
				rudp._sendingPacketsLock.ReleaseWriterLock();
			}

			//---- Send the packet
			packet.TSFirstSend = HiResTimer.MicroSeconds;
			if (!SocketSendPacket(rudp, packet, packet.Payload, packet.TSFirstSend))
			{
				rudp._sendingPacketsLock.AcquireWriterLock(RUDPSocket.LockTimeOut);
				rudp._sendingPackets.Remove(packet);
				rudp._sendingPacketsLock.ReleaseWriterLock();
				return false;
			}

			return true;
		}

		#endregion

		#region MakePacketPayload

		/// <summary>
		/// Create the RUDP packet : HEADER + payload
		/// </summary>
		internal static byte[] MakePacketPayload(RUDPSocket rudp,
												int packetId,
												RUDPPacketChannel channel,
												SACKSlot slot1, SACKSlot slot2, SACKSlot slot3, SACKSlot slot4,
												byte[] payload,
												int offset,
												int payloadLength)
		{
			//---- Here we make the payload while we will send:
			/**
			 * We create a header
			 * The format is:
			 * --------------
			 * - 1 byte : protocol version
			 * - 1 byte : header information
			 * - 1 byte : channel
			 * - 4 bytes : message ID
			 * - 4 bytes : advertized congestion window size
			 * - 4 bytes : payloadLength
			 * - 4 X (4 + 4) : 32 bytes for 4 SACK slots
			 * - the payload bytes
			 */
			int headerOffset = 0;
			byte[] packetPayload = PayloadManager.Allocate(channel, RUDPHeaderLength + payloadLength);

			//---- Protocol version
			packetPayload[headerOffset] = (byte)1;
			headerOffset++;

			//---- Header information
			// 3 bits : number of ACK slots

			// sack slot (3 bits)
			byte sacksSlotCount = 0;
			if (slot1 != null)
				sacksSlotCount++;
			if (slot2 != null)
				sacksSlotCount++;
			if (slot3 != null)
				sacksSlotCount++;
			if (slot4 != null)
				sacksSlotCount++;

			packetPayload[headerOffset] |= sacksSlotCount;
			headerOffset++;

			//---- Channel
			packetPayload[headerOffset] = (byte)channel;
			headerOffset++;

			//---- Message Id
			BinaryHelper.WriteInt(packetId, packetPayload, headerOffset);
			headerOffset += 4;

			//---- Control information : Advertised window
			if (rudp == null)
				BinaryHelper.WriteInt(-1, packetPayload, headerOffset);
			else
				BinaryHelper.WriteInt((int)rudp._controlWindow.AdvertisedWindow, packetPayload, headerOffset);
			headerOffset += 4;

			//---- Payload length
			BinaryHelper.WriteInt(payloadLength, packetPayload, headerOffset);
			headerOffset += 4;

			//---- SACK Slots
			if (slot1 != null)
			{
				BinaryHelper.WriteInt(slot1.StartPacketId, packetPayload, headerOffset);
				headerOffset += 4;
				BinaryHelper.WriteInt(slot1.EndPacketId, packetPayload, headerOffset);
				headerOffset += 4;
			}
			else headerOffset += 8;
			if (slot2 != null)
			{
				BinaryHelper.WriteInt(slot2.StartPacketId, packetPayload, headerOffset);
				headerOffset += 4;
				BinaryHelper.WriteInt(slot2.EndPacketId, packetPayload, headerOffset);
				headerOffset += 4;
			}
			else headerOffset += 8;
			if (slot3 != null)
			{
				BinaryHelper.WriteInt(slot3.StartPacketId, packetPayload, headerOffset);
				headerOffset += 4;
				BinaryHelper.WriteInt(slot3.EndPacketId, packetPayload, headerOffset);
				headerOffset += 4;
			}
			else headerOffset += 8;
			if (slot4 != null)
			{
				BinaryHelper.WriteInt(slot4.StartPacketId, packetPayload, headerOffset);
				headerOffset += 4;
				BinaryHelper.WriteInt(slot4.EndPacketId, packetPayload, headerOffset);
				headerOffset += 4;
			}
			else headerOffset += 8;

			if (payload == null)
				return packetPayload;

			//---- Payload
			Buffer.BlockCopy(payload, offset, packetPayload, headerOffset, payloadLength);

			return packetPayload;
		}

		#endregion

		#region UpdatePacketPayload

		static private void UpdatePacketPayload(byte[] packetPayload, SACKSlot slot1, SACKSlot slot2, SACKSlot slot3, SACKSlot slot4)
		{
			//---- Update header
			byte sacksSlotCount = 0;
			if (slot1 != null)
				sacksSlotCount++;
			if (slot2 != null)
				sacksSlotCount++;
			if (slot3 != null)
				sacksSlotCount++;
			if (slot4 != null)
				sacksSlotCount++;

			packetPayload[1] = sacksSlotCount;

			//---- Update slots
			int headerOffset = 15;
			if (slot1 != null)
			{
				BinaryHelper.WriteInt(slot1.StartPacketId, packetPayload, headerOffset);
				headerOffset += 4;
				BinaryHelper.WriteInt(slot1.EndPacketId, packetPayload, headerOffset);
				headerOffset += 4;
			}
			else headerOffset += 8;
			if (slot2 != null)
			{
				BinaryHelper.WriteInt(slot2.StartPacketId, packetPayload, headerOffset);
				headerOffset += 4;
				BinaryHelper.WriteInt(slot2.EndPacketId, packetPayload, headerOffset);
				headerOffset += 4;
			}
			else headerOffset += 8;
			if (slot3 != null)
			{
				BinaryHelper.WriteInt(slot3.StartPacketId, packetPayload, headerOffset);
				headerOffset += 4;
				BinaryHelper.WriteInt(slot3.EndPacketId, packetPayload, headerOffset);
				headerOffset += 4;
			}
			else headerOffset += 8;
			if (slot4 != null)
			{
				BinaryHelper.WriteInt(slot4.StartPacketId, packetPayload, headerOffset);
				headerOffset += 4;
				BinaryHelper.WriteInt(slot4.EndPacketId, packetPayload, headerOffset);
				headerOffset += 4;
			}
			//else headerOffset += 8;
		}

		#endregion

		#region SocketSendACK

		private static bool SocketSendACK(RUDPSocket rudp,
											PhysicalSocket physical,
											IPEndPoint remoteEndPoint,
											byte[] rudpPayload)
		{
			try
			{
				physical._socket.SendTo(rudpPayload, remoteEndPoint);
				//physical._socket.BeginSendTo(rudpPayload, 0, rudpPayload.Length, SocketFlags.None, rudp._remoteEndPoint, null, null);
			}
			catch (SocketException exception)
			{
				if (rudp != null)
					OnSocketUnhandledError(rudp, SocketErrorToRUDPSocketError(exception.SocketErrorCode), null);

				return false;
			}

			if (rudp != null)
				rudp._lastACKSendTS = HiResTimer.MicroSeconds;

			return true;
		}

		#endregion

		#region SocketSendPacket

		private static bool SocketSendPacket(RUDPSocket rudp, RUDPOutgoingPacket packet, byte[] rudpPayload, long now)
		{
			//---- Send the request
			try
			{
				rudp._physical._socket.SendTo(rudpPayload, rudp._remoteEndPoint);
				//rudp._physical._socket.BeginSendTo(rudpPayload, 0, rudpPayload.Length, SocketFlags.None, rudp._remoteEndPoint, null, null);
			}
			catch (SocketException exception)
			{
				if (exception.ErrorCode == (int)SocketError.MessageSize && packet.Channel == RUDPPacketChannel.MTUTuning)
				{
					// ICMP type 3 subtype 4
					// ICMP message, tell that this packet is too big
					rudp._pmtuDiscovery.OnICMPError(packet);
					return true;
				}

				OnSocketUnhandledError(rudp, SocketErrorToRUDPSocketError(exception.SocketErrorCode), null);
				return false;
			}

			rudp._lastSendTS = now;
			packet.TSLastSend = now;

			return true;
		}

		#endregion

		#region OnEndReceive

		/// <summary>
		/// Receive bytes from a socket
		/// </summary>
		private static void OnEndReceive(IAsyncResult result)
		{
			PhysicalSocket physical = (PhysicalSocket)result.AsyncState;

			EndPoint tempEndPoint = (EndPoint)physical._canReceiveFromEndPoint;

			//---- End receive
			EndPoint sender = new IPEndPoint(IPAddress.Any, 0);
			int size = -1;

			try
			{
				size = physical._socket.EndReceiveFrom(result, ref sender);
			}
			catch (SocketException socketException)
			{
				// The I/O operation has been aborted because of either 'a thread exit or an application request:
				// What you should really be doing is starting another BeginRead when you see a SocketException
				// with error code 995 (aborted) during EndRead.
				// What this means is that there wasn't any data on the socket to read, but that's fine if
				// you're just trying to read the next thing that comes off the socket.
				// NOTE that you should look at both a SocketException wrapped in an IOException and a straight SocketException. 

				// 995 :WSA_OPERATION_ABORTED
				// Overlapped operation aborted. This Win32 error indicates that an overlapped I/O operation
				// was canceled because of the closure of a socket. In addition, this error can occur when
				// executing the SIO_FLUSH ioctl command.
				if (socketException.ErrorCode == 995)
				{
					// Restart receiving
					physical._socket.BeginReceiveFrom(physical._receiveBuffer, 0, physical._receiveBuffer.Length, SocketFlags.None, ref tempEndPoint, new AsyncCallback(OnEndReceive), physical);
					return;
				}

				if (socketException.ErrorCode == 10054)
					physical._socket.BeginReceiveFrom(physical._receiveBuffer, 0, physical._receiveBuffer.Length, SocketFlags.None, ref tempEndPoint, new AsyncCallback(OnEndReceive), physical);
				else
				{
					OnSocketUnhandledError(physical, sender as IPEndPoint, socketException.SocketErrorCode);
					return;
				}
			}

			//----Simulate packet loss
#if TEST_PACKETLOOSE
			if (_looseRandom.NextDouble() < 0.1)
			{
				physical._socket.BeginReceiveFrom(physical._receiveBuffer, 0, physical._receiveBuffer.Length, SocketFlags.None, ref tempEndPoint, new AsyncCallback(OnEndReceive), physical);
				return;
			}
#endif

			//---- Handle the packet
			RUDPPacketChannel channel = RUDPPacketChannel.Undefined;
			int packetId = -2;
			int advertisedWindowSize = 0;
			SACKSlot slot1 = null;
			SACKSlot slot2 = null;
			SACKSlot slot3 = null;
			SACKSlot slot4 = null;
			byte[] payload = null;

			if (!_isStackRunning)
				return;

			//-- Decode payload
			HandlePayload(physical, physical._receiveBuffer, size, sender as IPEndPoint,
							out channel,
							out packetId,
							out advertisedWindowSize,
							out slot1,
							out slot2,
							out slot3,
							out slot4,
							out payload);

			//-- Restart receiving
			physical._socket.BeginReceiveFrom(physical._receiveBuffer, 0, physical._receiveBuffer.Length, SocketFlags.None, ref tempEndPoint, new AsyncCallback(OnEndReceive), physical);

			//-- Handle the packet
			HandlePacket(physical, sender as IPEndPoint, channel, packetId, advertisedWindowSize, slot1, slot2, slot3, slot4, payload);
		}

		#endregion

		#region HandlePayload

		/// <summary>
		/// Handle the bytes received from a socket.
		/// </summary>
		private static void HandlePayload(PhysicalSocket physical, byte[] payload, int length, IPEndPoint sender,
				out RUDPPacketChannel channel,
				out int packetId,
				out int advertisedWindowSize,
				out SACKSlot slot1,
				out SACKSlot slot2,
				out SACKSlot slot3,
				out SACKSlot slot4,
				out byte[] packetPayload)
		{
			int offset = 0;

			//-- Protocol version
			byte version = payload[offset];
			offset++;

			//-- Header information
			byte sacksSlotCount = payload[offset];
			offset++;

			//-- Channel
			byte channelByte = payload[offset];
			offset++;
			channel = (RUDPPacketChannel)Enum.ToObject(typeof(RUDPPacketChannel), channelByte);

			//-- Packet Id
			packetId = BinaryHelper.ReadInt(payload, offset);
			offset += 4;

			//-- Control information : Advertised window
			advertisedWindowSize = BinaryHelper.ReadInt(payload, offset);
			offset += 4;

			//-- Payload length
			int payloadLength = BinaryHelper.ReadInt(payload, offset);
			offset += 4;

			//---- SACK Slots
			slot1 = null;
			slot2 = null;
			slot3 = null;
			slot4 = null;
			int startPacketId;
			int endPacketId;
			if (sacksSlotCount > 0)
			{
				startPacketId = BinaryHelper.ReadInt(payload, offset);
				offset += 4;
				endPacketId = BinaryHelper.ReadInt(payload, offset);
				offset += 4;
				slot1 = new SACKSlot(startPacketId, endPacketId);
			}
			else offset += 8;