Click here to Skip to main content
Click here to Skip to main content
Articles » Languages » C# » General » Downloads
 
Add your own
alternative version

Lazy Alternatives - LazyAndWeak and BackgroundLoader

, 1 Dec 2011
This article will present two alternatives to Lazy.
LazyAlternatives.zip
LazyAlternatives
LazyAlternatives.suo
LazyAlternativesSample
Properties
Settings.settings
Pfz
Caching
Collections
DataTypes
DynamicObjects
Internal
Extensions
Factoring
Pfz.csproj.user
Pfz.Phone.csproj.user
Pfz.ruleset
Pfz.Silverlight.csproj.user
Pfz.snk
Pfz.suo
PhoneSpecific
Properties
Remoting
Instructions
Internal
Serializers
Udp
Serialization
BinaryBuiltIn
Threading
Contexts
Disposers
Unsafe
using System;
using System.Collections.Generic;
using System.Net;
using System.Net.Sockets;
using System.Threading;
using Pfz.Threading;

namespace Pfz.Remoting.Udp
{
	/// <summary>
	/// An UDP connection that supports guaranteed delivery. It is not ordered, so it does not replaces the Tcp/IP.
	/// </summary>
	public sealed class GuaranteedUdpConnection:
		ThreadSafeDisposable,
		IGuaranteedUdpConnection
	{
		private readonly Dictionary<long, _BytePair> _pendings = new Dictionary<long, _BytePair>();
		private readonly ManagedManualResetEvent _resendEvent1 = new ManagedManualResetEvent();
		private readonly ManagedManualResetEvent _resendEvent2 = new ManagedManualResetEvent();
		private readonly ManagedAutoResetEvent _confirmEvent1 = new ManagedAutoResetEvent();
		private readonly ManagedAutoResetEvent _confirmEvent2 = new ManagedAutoResetEvent();
		private readonly ManagedSemaphore _supportsGuaranteedSemaphore;
		private readonly List<long> _missings = new List<long>();
		private readonly int _confirmImmediatelyCount;
		private readonly int _pendingGuaranteedDeliveryCount;
		private UdpClient _udp;
		#if WINDOWS_PHONE
		private readonly EndPoint _remoteEndpoint;
		private int _idGenerator = 1;
		#else
		private readonly IPEndPoint _remoteEndpoint;
		private long _idGenerator = 1;
		#endif

		#if !WINDOWS_PHONE
		internal GuaranteedUdpConnection(GuaranteedUdpListener listener, IPEndPoint endpoint)
		{
			try
			{
				_listener = listener;
				_udp = listener._udp;
				_remoteEndpoint = endpoint;

				_pendingGuaranteedDeliveryCount = listener._pendingGuaranteedDeliveryCount;
				_supportsGuaranteedSemaphore = new ManagedSemaphore(listener._pendingGuaranteedDeliveryCount);
				_confirmImmediatelyCount = listener._confirmImmediatelyCount;

				UnlimitedThreadPool.Run(_Resender);
				UnlimitedThreadPool.Run(_Confirmer);
			}
			catch
			{
				Dispose();
				throw;
			}
		}
		#endif

		/// <summary>
		/// Creates a new instance of this class.
		/// </summary>
		#if WINDOWS_PHONE
			public GuaranteedUdpConnection(string hostname, int port, int pendingGuaranteedDeliveryCount=57):
				this(new DnsEndPoint(hostname, port), pendingGuaranteedDeliveryCount)
			{
			}
		#else
			public GuaranteedUdpConnection(string hostname, int port, int pendingGuaranteedDeliveryCount=57):
				this(new IPEndPoint(_GetHostAddress(hostname), port), pendingGuaranteedDeliveryCount)
			{
			}

			private static IPAddress _GetHostAddress(string hostname)
			{
				var result = Dns.GetHostAddresses(hostname);
				if (result.Length == 0)
					throw new ArgumentException("Can't find the host: " + hostname, "hostname");

				foreach(var item in result)
					if (item.AddressFamily == AddressFamily.InterNetwork)
						return item;

				throw new ArgumentException("Can't find an IP for the host: " + hostname, "hostname");
			}
		#endif

		private static readonly byte[] _emptyBytes = new byte[1];
		/// <summary>
		/// Creates a new instance of this class.
		/// </summary>
		#if WINDOWS_PHONE
		public GuaranteedUdpConnection(EndPoint remoteEndpoint, int pendingGuaranteedDeliveryCount=57)
		#else
		public GuaranteedUdpConnection(IPEndPoint remoteEndpoint, int pendingGuaranteedDeliveryCount=57)
		#endif
		{
			try
			{
				if (remoteEndpoint == null)
					throw new ArgumentNullException("remoteEndpoint");

				if (pendingGuaranteedDeliveryCount < 2)
					throw new ArgumentException("pendingGuaranteedDeliveryCount must be at least 2.", "pendingGuaranteedDeliveryCount");

				_pendingGuaranteedDeliveryCount = pendingGuaranteedDeliveryCount;
				_udp = new UdpClient(0);
				_remoteEndpoint = remoteEndpoint;
				_supportsGuaranteedSemaphore = new ManagedSemaphore(pendingGuaranteedDeliveryCount);
				_confirmImmediatelyCount = pendingGuaranteedDeliveryCount/2;

				UnlimitedThreadPool.Run(_Resender);
				GuaranteedSend(_emptyBytes, 0, 1);
				UnlimitedThreadPool.Run(_Confirmer);
				UnlimitedThreadPool.Run(_Receiver);
			}
			catch
			{
				Dispose();
				throw;
			}
		}

		/// <summary>
		/// Releases all resources used by this UDP connection.
		/// </summary>
		protected override void Dispose(bool disposing)
		{
			if (disposing)
			{
				if (_supportsGuaranteedSemaphore != null)
					_supportsGuaranteedSemaphore.Dispose();

				if (_resendEvent1 != null)
					_resendEvent1.Dispose();

				if (_resendEvent2 != null)
					_resendEvent2.Dispose();

				if (_confirmEvent1 != null)
					_confirmEvent1.Dispose();

				if (_confirmEvent2 != null)
					_confirmEvent2.Dispose();

				if (_receivesEvent != null)
					_receivesEvent.Dispose();

				#if WINDOWS_PHONE
					Disposer.Dispose(ref _udp);
				#else
					if (_listener == null)
						Disposer.Dispose(ref _udp);
					else
					{
						lock(_listener.DisposeLock)
							if (!_listener.WasDisposed)
								_listener._connections.Remove(_remoteEndpoint);
					}
				#endif
			}

			base.Dispose(disposing);
		}

		/// <summary>
		/// Gets information about the RemoteEndpoint.
		/// </summary>
		public EndPoint RemoteEndpoint
		{
			get
			{
				return _remoteEndpoint;
			}
		}

		private static readonly TimeSpan _200Milliseconds = TimeSpan.FromMilliseconds(200);
		private void _Resender()
		{
			try
			{
				while(true)
				{
					_resendEvent1.WaitOne();

					if (_resendEvent2.WaitOne(200))
						return;

					lock(DisposeLock)
					{
						if (WasDisposed)
							return;

						foreach(var pair in _pendings.Values)
						{
							var dateTime = pair._dateTime;
							var now = DateTime.Now;
							TimeSpan diff = now-dateTime;
							if (diff >= TimeSpan.Zero && diff < _200Milliseconds)
								continue;

							if (pair._count > (60000 / 200))
								return; // this is a timeout, so we return, disposing ourselves.

							var bytes = pair._bytes;
							_udp.Send(bytes, bytes.Length, _remoteEndpoint);
							pair._dateTime = DateTime.Now;
							pair._count++;
						}
					}
				}
			}
			catch
			{
			}
			finally
			{
				Dispose();
			}
		}

		private int _countReceivesWithoutConfirmation;
		private void _Confirmer()
		{
			try
			{
				while(true)
				{
					// we will keep sending packets from 5 to 5 seconds to avoid disconnect.
					_confirmEvent1.WaitOne(5000);
					_confirmEvent2.WaitOne(57);

					if (_confirmEvent2.WasDisposed)
						return;

					_countReceivesWithoutConfirmation = 0;

					byte[] buffer;
					int bufferLength;
					lock(DisposeLock)
					{
						if (WasDisposed)
							return;

						long lastReceivedId = -_lastReceivedId;

						int count = _missings.Count;
						if (count > 50)
						{
							_confirmEvent1.Set();
							count = 50;
							lastReceivedId = -1;
						}

						bufferLength = (count*8) + 10;
						buffer = new byte[bufferLength];
						var sizeBytes = BitConverter.GetBytes((ushort)bufferLength-10);
						buffer[0] = sizeBytes[0];
						buffer[1] = sizeBytes[1];
						var idBytes = BitConverter.GetBytes(lastReceivedId);
						Buffer.BlockCopy(idBytes, 0, buffer, 2, 8);
						int pos = 10;
						for(int i=0; i<count; i++)
						{
							long missingId = _missings[i];
							Buffer.BlockCopy(BitConverter.GetBytes(missingId), 0, buffer, pos, 8);

							pos += 8;
						}
					}

					_udp.Send(buffer, bufferLength, _remoteEndpoint);
				}
			}
			catch
			{
			}
			finally
			{
				Dispose();
			}
		}

		private long _lastReceivedId = 1;
		private bool _ProcessReceived(long id, byte[] received)
		{
			if (id < _lastReceivedId)
			{
				bool result = _missings.Remove(id);
				return result;
			}

			if (id == _lastReceivedId)
				return false;

			if (id == _lastReceivedId+1)
			{
				_lastReceivedId++;

				return true;
			}

			for(long i=_lastReceivedId+1; i<id; i++)
				_missings.Add(i);

			_lastReceivedId = id;

			return true;
		}

		private readonly HashSet<long> _toRemove = new HashSet<long>();
		private long _lastConfirmedId = 1;
		private void _ProcessConfirmation(long lastConfirmedId, byte[] received)
		{
			lock(DisposeLock)
			{
				if (lastConfirmedId != 1)
					_lastConfirmedId = lastConfirmedId;

				_toRemove.Clear();
				foreach(long id in _pendings.Keys)
					if (id <= lastConfirmedId)
						_toRemove.Add(id);

				int length = received.Length;
				for(int pos=10; pos < length; pos += 8)
				{
					long idStillMissing = BitConverter.ToInt64(received, pos);
					_toRemove.Remove(idStillMissing);
				}

				int toRemoveCount = _toRemove.Count;
				if (toRemoveCount > 0)
				{
					foreach(var id in _toRemove)
						_pendings.Remove(id);

					_supportsGuaranteedSemaphore.Exit(toRemoveCount);

					if (_pendings.Count == 0)
						_resendEvent1.Reset();
				}
			}
		}

		#if !WINDOWS_PHONE
		private readonly GuaranteedUdpListener _listener;
		#endif
		internal readonly ManagedAutoResetEvent _receivesEvent = new ManagedAutoResetEvent();
		internal readonly Queue<byte[]> _receivesQueue = new Queue<byte[]>();
		/// <summary>
		/// Receives an UDP packet. Note that it includes a header used by the framework, so
		/// check the UdpHeaderSize property.
		/// </summary>
		public byte[] UdpReceive()
		{
			while(true)
			{
				lock(DisposeLock)
				{
					if (WasDisposed)
						return null;

					if (_receivesQueue.Count > 0)
						return _receivesQueue.Dequeue();
				}

				_receivesEvent.WaitOne();
			}
		}

		private void _Receiver()
		{
			try
			{
				#if !WINDOWS_PHONE
					IPEndPoint endpoint = null;
				#endif

				while(!WasDisposed)
				{
					#if WINDOWS_PHONE
					byte[] received = _udp.Receive();
					#else
					byte[] received = _udp.Receive(ref endpoint);
					#endif

					if (received == null)
						return;

					#if !WINDOWS_PHONE
					if (!endpoint.Equals(_remoteEndpoint))
						continue;
					#endif

					_StoreReceived(received);
				}
			}
			catch
			{
			}
			finally
			{
				Dispose();
			}
		}
		internal void _StoreReceived(byte[] received)
		{
			lock(DisposeLock)
			{
				if (WasDisposed)
					return;

				if (received.Length < 2)
					return;

				int size = BitConverter.ToUInt16(received, 0);
				if (size + 10 != received.Length)
					return;

				long id = BitConverter.ToInt64(received, 2);
				if (id < 0)
				{
					_ProcessConfirmation(-id, received);
					return;
				}

				#if !WINDOWS_PHONE
				bool canEnqueue = id != 2 || _listener == null;
				if (canEnqueue)
				#endif
				{
					if (_receivesQueue.Count >= _pendingGuaranteedDeliveryCount)
						return;
				}

				bool firstTime = id == 0 || _ProcessReceived(id, received);
				_confirmEvent1.Set();

				if (!firstTime)
					return;

				#if !WINDOWS_PHONE
				if (canEnqueue)
				#endif
				{
					_receivesQueue.Enqueue(received);
					_receivesEvent.Set();
				}

				if (_countReceivesWithoutConfirmation < _confirmImmediatelyCount)
					_countReceivesWithoutConfirmation++;
				else
				{
					_countReceivesWithoutConfirmation = 0;
					_confirmEvent2.Set();
				}
			}
		}

		/// <summary>
		/// Sends a guaranteed to arrive packet. It is still unordered.
		/// </summary>
		public void GuaranteedSend(byte[] buffer, int offset, int count)
		{
			if (buffer == null)
				throw new ArgumentNullException("buffer");

			if (offset < 0)
				throw new ArgumentException("offset can't be negative.", "offset");

			if (offset > buffer.Length)
				throw new ArgumentOutOfRangeException("offset");

			if (count > 65535)
				throw new ArgumentOutOfRangeException("count", "Packets must have at maximum 65535 bytes.");

			if (offset+count > buffer.Length)
				throw new ArgumentException("offset+count is outside the buffer limits.");

			if (count == 0)
				return;

			int newLength = count + 10;
			var lengthBytes = BitConverter.GetBytes((ushort)count);
			byte[] newBuffer = new byte[newLength];
			newBuffer[0] = lengthBytes[0];
			newBuffer[1] = lengthBytes[1];
			Buffer.BlockCopy(buffer, offset, newBuffer, 10, count);

			long id = Interlocked.Increment(ref _idGenerator);
			Buffer.BlockCopy(BitConverter.GetBytes(id), 0, newBuffer, 2, 8);

			_supportsGuaranteedSemaphore.Enter();
			lock(DisposeLock)
			{
				CheckUndisposed();

				_pendings.Add(id, new _BytePair(newBuffer));
				_resendEvent1.Set();

				_udp.Send(newBuffer, newLength, _remoteEndpoint);
			}
		}

		/// <summary>
		/// Sends an udp packet. Note that if you are using guaranteed sends you should not use the udp object to send,
		/// you must send udp packets by this method.
		/// </summary>
		public void UdpSend(byte[] buffer, int offset, int count)
		{
			if (buffer == null)
				throw new ArgumentNullException("buffer");

			if (offset < 0)
				throw new ArgumentException("offset can't be negative.", "offset");

			if (offset > buffer.Length)
				throw new ArgumentOutOfRangeException("offset");

			if (count > 65535)
				throw new ArgumentOutOfRangeException("Packets must have at maximum 65535 bytes.");

			if (offset+count > buffer.Length)
				throw new ArgumentException("offset+count is outside the buffer limits.");

			if (count == 0)
				return;

			int newLength = count + 10;
			var lengthBytes = BitConverter.GetBytes((ushort)count);
			byte[] newBuffer = new byte[newLength];
			newBuffer[0] = lengthBytes[0];
			newBuffer[1] = lengthBytes[1];
			// the id remains 0 because it is not set.
			Buffer.BlockCopy(buffer, offset, newBuffer, 10, count);

			lock(DisposeLock)
			{
				CheckUndisposed();

				_udp.Send(newBuffer, newLength, _remoteEndpoint);
			}
		}

		/// <summary>
		/// Gets the size of the Header information put in all packets.
		/// </summary>
		public int UdpHeaderSize
		{
			get
			{
				return 10;
			}
		}

		/// <summary>
		/// Checks if a packet was sent as guaranteed (true) or not.
		/// </summary>
		public bool IsGuaranteedPacket(byte[] packet)
		{
			return BitConverter.ToInt64(packet, 2) != 0;
		}
	}
}

By viewing downloads associated with this article you agree to the Terms of Service and the article's licence.

If a file you wish to view isn't highlighted, and is a text file (not binary), please let us know and we'll add colourisation support for it.

License

This article, along with any associated source code and files, is licensed under The Code Project Open License (CPOL)

About the Author

Paulo Zemek
Architect
Canada Canada
I started to program computers when I was 11 years old, as a hobbist, programming in AMOS Basic and Blitz Basic for Amiga.
At 12 I had my first try with assembler, but it was too difficult at the time. Then, in the same year, I learned C and, after learning C, I was finally able to learn assembler (for Motorola 680x0).
Not sure, but probably between 12 and 13, I started to learn C++. I always programmed "in an object oriented way", but using function pointers instead of virtual methods.
 
At 15 I started to learn Pascal at school and to use Delphi. At 16 I started my first internship (using Delphi). At 18 I started to work professionally using C++ and since then I've developed my programming skills as a professional developer in C++ and C#, generally creating libraries that help other developers do they work easier, faster and with less errors.
 
Want more info or simply want to contact me?
Take a look at: http://paulozemek.azurewebsites.net/
Or e-mail me at: paulozemek@outlook.com
 
Codeproject MVP 2012
Microsoft MVP 2013

| Advertise | Privacy | Mobile
Web03 | 2.8.140721.1 | Last Updated 1 Dec 2011
Article Copyright 2011 by Paulo Zemek
Everything else Copyright © CodeProject, 1999-2014
Terms of Service
Layout: fixed | fluid