Click here to Skip to main content
11,629,535 members (63,969 online)
Click here to Skip to main content
Articles » Languages » C# » General » Downloads
Add your own
alternative version

Tagged as

Lazy Alternatives - LazyAndWeak and BackgroundLoader

, 1 Dec 2011 CPOL 10.8K 305 25
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.Sockets;
using System.Reflection;
using System.Threading;
using Pfz.Caching;
using Pfz.DynamicObjects;
using Pfz.DynamicObjects.Internal;
using Pfz.Remoting.Instructions;
using Pfz.Remoting.Internal;
using Pfz.Threading;

namespace Pfz.Remoting
{
	/// <summary>
	/// Class used to access objects remotelly.
	/// </summary>
	public class RemotingClient:
		RemotingCommon,
		IGarbageCollectionAware
	{
		#region Fields
			private IChanneller _channeller;
			internal long _id;

			internal _BidirectionalDictionary _objectsUsedByTheOtherSide;
			internal Dictionary<long, _RemotingProxy> _wrappers = new Dictionary<long, _RemotingProxy>();
			internal object _wrappersLock = new object();
			internal readonly object _collectWrappersEvent = new object();
			internal bool _weCausedActualCollection;
			internal readonly object _registeredEventsLock = new object();
			internal Dictionary<EventInfo, Dictionary<Delegate, WeakList<object>>> _registeredEvents;

			private BaseDisposableThreadLocal _threadDatas = new BaseDisposableThreadLocal();
		#endregion

		#region Constructors
			/// <summary>
			/// Creates a new RemotingClient.
			/// </summary>
			public RemotingClient()
			{
				GCUtils.RegisterForCollectedNotification(this);
				_objectsUsedByTheOtherSide = new _BidirectionalDictionary();
				_parameters._allowSyncCalls = true;
			}

			/// <summary>
			/// Creates a new RemotingClient with the given parameters
			/// </summary>
			public RemotingClient(RemotingParameters parameters):
				base(parameters)
			{
				GCUtils.RegisterForCollectedNotification(this);
				_objectsUsedByTheOtherSide = new _BidirectionalDictionary();
			}
		#endregion
		#region Dispose
			/// <summary>
			/// Frees the unmanaged resources of this RemotingClient.
			/// </summary>
			protected override void Dispose(bool disposing)
			{
				if (disposing)
				{
					GCUtils.UnregisterFromCollectedNotification(this);

					if (_collectWrappersEvent != null)
						lock(_collectWrappersEvent)
							Monitor.Pulse(_collectWrappersEvent);

					Disposer.Dispose(ref _channeller);
					Disposer.Dispose(ref _objectsUsedByTheOtherSide);
					Disposer.Dispose(ref _threadDatas);

					var registeredEventsLock = _registeredEventsLock;
					if (registeredEventsLock != null)
					{
						lock(registeredEventsLock)
						{
							var registeredEvents = _registeredEvents;
							if (registeredEvents != null)
							{
								foreach(var pair in registeredEvents)
								{
									var eventInfo = pair.Key;
									var delegateDictionary = pair.Value;

									foreach(var pair2 in delegateDictionary)
									{
										var handler = pair2.Key;
										var weakList = pair2.Value;

										foreach(object obj in weakList)
											eventInfo.RemoveEventHandler(obj, handler);
									}
								}
							}
						}
					}

					Disposer.Dispose(ref _runnableRunner);

					var disposed = Disposed;
					if (disposed != null)
						disposed(this, null);
				}

				base.Dispose(disposing);
			}
		#endregion
		#region _Collected
			void IGarbageCollectionAware.OnCollected()
			{
				if (!_weCausedActualCollection)
				{
					if (_collectWrappersEvent != null)
						lock(_collectWrappersEvent)
							Monitor.Pulse(_collectWrappersEvent);
				}

				try
				{
					lock(_registeredEventsLock)
					{
						var oldRegisteredEvents = _registeredEvents;

						if (oldRegisteredEvents != null)
						{
							var newRegisteredEvents = new Dictionary<EventInfo, Dictionary<Delegate, WeakList<object>>>();
							foreach(var pair in oldRegisteredEvents)
							{
								var delegateDictionary = pair.Value;
								var newDelegateDictionary = new Dictionary<Delegate, WeakList<object>>();

								foreach(var pair2 in delegateDictionary)
								{
									if (pair2.Value.Count > 0)
										newDelegateDictionary.Add(pair2.Key, pair2.Value);
								}

								if (newDelegateDictionary.Count > 0)
									newRegisteredEvents.Add(pair.Key, newDelegateDictionary);
							}

							if (newRegisteredEvents.Count > 0)
								_registeredEvents = newRegisteredEvents;
							else
								_registeredEvents = null;
						}
					}
				}
				catch
				{
					var channeller = _channeller;
					if (!WasDisposed && channeller != null && !channeller.WasDisposed)
						throw;
				}
			}
		#endregion

		#region Properties
			#region ExecutingClient
				[ThreadStatic]
				internal static RemotingClient _executingClient;

				/// <summary>
				/// Gets the RemotingClient that invoked the actual method, directly or indirectly.
				/// Will return null if the actual method was not invoked by a remote call.
				/// </summary>
				public static RemotingClient ExecutingClient
				{
					get
					{
						return _executingClient;
					}
				}
			#endregion
			#region Channeller
				/// <summary>
				/// Gets the channeller used by this remoting client.
				/// </summary>
				public IChanneller Channeller
				{
					get
					{
						return _channeller;
					}
				}
			#endregion

			#region IsStarted
				private bool _isStarted;
				/// <summary>
				/// Gets a value indicating if Start was already called or not.
				/// </summary>
				public bool IsStarted
				{
					get
					{
						return _isStarted;
					}
				}
			#endregion
			#region RunnableQueue
				private RunnableRunner _runnableRunner;

				/// <summary>
				/// Gets a runnable runner (dispatcher, if you prefer) to send messages to this client.
				/// Use it when executing asynchronous calls, so a slow client will not hang the server.
				/// Do not use ThreadPool threads or Tasks to send async messages to the clients, as 
				/// eventually all threads will be occupied to respond to the slowest client.
				/// </summary>
				public RunnableRunner RunnableRunner
				{
					get
					{
						var result = _runnableRunner;
						if (result == null)
						{
							lock(DisposeLock)
							{
								CheckUndisposed();

								result = _runnableRunner;
								if (result == null)
								{
									result = new RunnableRunner(_RemotingCleaner.Instance);
									_runnableRunner = result;
									_RemotingSetter setter = new _RemotingSetter(this);
									result.Run(setter);
								}
							}
						}

						return result;
					}
				}
			#endregion
		#endregion
		#region Methods
			#region GetFromRemoteObject
				/// <summary>
				/// Gets the RemotingClient that created the given remote object.
				/// Returns null if the object is not remote.
				/// </summary>
				public static RemotingClient GetFromRemoteObject(object obj)
				{
					BaseImplementedProxy baseImplementedProxy = obj as BaseImplementedProxy;
					if (baseImplementedProxy == null)
						return null;

					object fieldValue = baseImplementedProxy._proxyObject;
					_RemotingProxy remotingProxy = fieldValue as _RemotingProxy;
					if (remotingProxy == null)
						return null;

					return remotingProxy.RemotingClient;
				}
			#endregion

			#region Start
				/// <summary>
				/// Starts this remoting client.
				/// Parameters will not accept changes anymore.
				/// </summary>
				public void Start(IChanneller channeller)
				{
					if (channeller == null)
						throw new ArgumentException("channeller can't be null.");

					CheckThread();

					if (_isStarted)
						throw new RemotingException("This RemotingClient is already running.");

					_parameters._isReadOnly = true;
					_isStarted = true;

					_channeller = channeller;
					channeller.ChannelCreated += _RunAsServer;
					channeller.Start();

					UnlimitedThreadPool.Run(_CollectWrappers);
				}

				/// <summary>
				/// Starts this RemotingClient.
				/// </summary>
				public void Start(string hostname, int port)
				{
					//Start(HybridChanneller.Connect(hostname, port, DefaultBufferSizePerChannel));

					var client = new TcpClient(hostname, port);
					var channeller = ChannellerConnection.Create(client, DefaultBufferSizePerChannel);
					Start(channeller);
				}
			#endregion
			#region Disconnect
				/// <summary>
				/// Disconnects the active connection.
				/// This may end-up disposing the RemotingClient if it does not supports reconnections.
				/// </summary>
				public void Disconnect()
				{
					var channeller = _channeller;
					if (channeller != null)
						channeller.Dispose();
				}
			#endregion

			#region _CreateConnectionIfNeeded
				private object _connectLock = new object();
				internal void _CreateConnectionIfNeeded()
				{
					CheckUndisposed();
			
					if (!_isStarted)
						throw new RemotingException("You must call Start().");
				}
			#endregion

			#region _GetWrappedDelegate
				internal object _GetWrappedDelegate(_WrappedDelegate wrappedDelegate)
				{
					long id = wrappedDelegate.Id;

					_RemotingProxyDelegate proxy = new _RemotingProxyDelegate(this, id);
					var result = DelegateProxier.Proxy(proxy, wrappedDelegate.DelegateType);
					proxy.ImplementedDelegate = result;

					lock(_wrappersLock)
						_wrappers.Add(id, proxy);

					return result;
				}
			#endregion
			#region _GetWrappedObject
				internal object _GetWrappedObject(_Wrapped wrappedObject)
				{
					_RemotingProxyObject proxy = new _RemotingProxyObject(this, wrappedObject);
					var result = InterfaceProxier.Proxy(proxy, wrappedObject.InterfaceTypes);
					proxy.ImplementedObject = result;

					lock(_wrappersLock)
						_wrappers.Add(wrappedObject.Id, proxy);

					return result;
				}
			#endregion
			#region _GetReferencedObject
				internal object _GetReferencedObject(_Reference reference)
				{
					_RemotingProxy result;
					if (!_wrappers.TryGetValue(reference.Id, out result))
						throw new RemotingException("Got a reference to an inexisting wrapper.");

					_RemotingProxyObject proxyObj = result as _RemotingProxyObject;
					if (proxyObj != null)
						return proxyObj.ImplementedObject;

					_RemotingProxyDelegate delegateProxy = (_RemotingProxyDelegate)result;
					return delegateProxy.ImplementedDelegate;
				}
			#endregion
			#region _GetThreadData
				private _ThreadData _GetThreadData()
				{
					var thread = Thread.CurrentThread;

					if (thread.IsThreadPoolThread)
						throw new InvalidOperationException("Can't use Pfz.Remoting from .Net default ThreadPool. Such threads are revived with all their variables, causing a lot of problems.");

					var threadDatas = _threadDatas;

					if (threadDatas == null)
						throw new ObjectDisposedException("RemotingClient");

					lock(threadDatas._lock)
					{
						DisposeAssurer disposeCaller;
						if (threadDatas._dictionary.TryGetValue(thread, out disposeCaller))
						{
							_ThreadData result = (_ThreadData)disposeCaller.Value;
							if (!result._channel.WasDisposed)
								return result;
						}

						var threadData = new _ThreadData(_channeller.CreateChannel(), this);
						disposeCaller = DisposeAssurer.Create((IDisposable)threadData);
						threadDatas._dictionary[thread] = disposeCaller;
						return threadData;
					}
				}
			#endregion
			#region _Invoke
				internal object _Invoke(Instruction instruction)
				{
					bool allowSyncCalls = AllowSyncRemoting.Value ?? AllowSyncCalls;
					if (!allowSyncCalls && ExecutingClient != this)
						throw new RemotingSyncException("This client does not support synchronous calls. Use the RunnableRunner or set AllowSyncCalls/AllowSyncRemoting.Value to true.");

					var threadData = _GetThreadData();

					threadData.Serialize(instruction);

					object resultObject;
					while(true)
					{
						try
						{
							resultObject = threadData.Deserialize();
						}
						catch(Exception exception)
						{
							Dispose(exception);

							throw;
						}

						instruction = resultObject as Instruction;
						if (instruction == null)
							break;

						instruction.Run(this, threadData);
					}

					RemotingResult result = (RemotingResult)resultObject;
					var exception2 = result.Exception;
					if (exception2 != null)
						throw exception2;

					var resultValue = result.Value;
					return resultValue;
				}
				internal object _Invoke(Instruction instruction, MethodInfo methodInfo, object[] outParameters)
				{
					bool allowSyncCalls = AllowSyncRemoting.Value ?? AllowSyncCalls;
					if (!allowSyncCalls && ExecutingClient != this)
						throw new RemotingSyncException("This client does not support synchronous calls. Use the RunnableRunner or set AllowSyncCalls/AllowSyncRemoting.Value to true.");

					var threadData = _GetThreadData();

					threadData.Serialize(instruction);

					object resultObject;
					while(true)
					{
						try
						{
							resultObject = threadData.Deserialize();
						}
						catch(Exception exception)
						{
							Dispose(exception);

							throw;
						}

						instruction = resultObject as Instruction;
						if (instruction == null)
							break;

						instruction.Run(this, threadData);
					}

					RemotingResult result = (RemotingResult)resultObject;
					var exception2 = result.Exception;
					if (exception2 != null)
						throw exception2;

					_ProcessOut(methodInfo, result.OutValues, outParameters);

					var resultValue = result.Value;
					return resultValue;
				}
			#endregion

			#region _CollectWrappers
				private void _CollectWrappers()
				{
					try
					{
						_ThreadData threadData = null;
						while(true)
						{
							lock(_collectWrappersEvent)
							{
								if (WasDisposed)
									return;

								Monitor.Wait(_collectWrappersEvent);
							}

							if (WasDisposed)
								return;

							if (threadData == null)
								threadData = _GetThreadData();

							lock(_objectsUsedByTheOtherSide.DisposeLock)
							{
								if (WasDisposed)
									return;

								if (_objectsUsedByTheOtherSide._dictionary1.Count < 2)
									continue;

								threadData.Serialize(InstructionCollect.Instance);
								long[] collectedIds = (long[])threadData.Deserialize();
								_objectsUsedByTheOtherSide.RemoveIds(collectedIds);
							}
						}
					}
					catch(Exception exception)
					{
						Dispose(exception);
					}
				}
			#endregion

			#region _RunAsServer
				private void _RunAsServer(object sender, ChannelCreatedEventArgs args)
				{
					var data = args.Data;
					if (data != null)
					{
						if (data == _FakeNull.Instance)
							args.Data = null;

						OnUserChannelCreated(args);

						return;
					}

					try
					{
						var channel = args.Channel;

						var thread = Thread.CurrentThread;
						using(var threadData = new _ThreadData(channel, this))
						{
							try
							{
								_threadDatas._Value = threadData;

								while(true)
								{
									object instructionObject;

									instructionObject = threadData.Deserialize();
									var instruction = (Instruction)instructionObject;

									var oldExecutingClient = _executingClient;
									try
									{
										_executingClient = this;
										instruction.Run(this, threadData);
									}
									finally
									{
										_executingClient = oldExecutingClient;
									}
								}
							}
							finally
							{
								_threadDatas._Value = null;
							}
						}
					}
					catch(Exception exception)
					{
						// TODO remove this when the bug is found.
						/*try
						{
							StringBuilder message = new StringBuilder();
							var actualException = exception;
							while(actualException != null)
							{
								message.Append(actualException.GetType().FullName);
								message.Append("\r\n");
								message.Append(actualException.Message);
								message.Append("\r\n");
								message.Append(actualException.StackTrace);

								actualException = actualException.InnerException;
							}

							message.Append("\r\n\r\n\r\n");
							File.AppendAllText("RemotingClient.Exception.txt", message.ToString());
						}
						catch
						{
						}*/

						if (exception.GetBaseException() is RemotingSyncException)
							throw;

						var channeller = _channeller;
						if (!WasDisposed && channeller != null && channeller.WasDisposed)
							Dispose(exception);
					}
				}
			#endregion

			#region CreateUserChannel
				/// <summary>
				/// Creates an stream to communicate to the other side, without opening a new tcp ip port.
				/// </summary>
				public IChannel CreateUserChannel(object serializableData = null)
				{
					_CreateConnectionIfNeeded();

					if (serializableData == null)
						serializableData = _FakeNull.Instance;

					var result = _channeller.CreateChannel(serializableData);
					return result;
				}
			#endregion

			#region InvokeStaticMethod
				/// <summary>
				/// Invokes a registered static method on the other side.
				/// </summary>
				public object InvokeStaticMethod(string methodName, params object[] parameters)
				{
					_CreateConnectionIfNeeded();
					var instruction = new InstructionInvokeStaticMethod();

					instruction.MethodName = methodName;
					instruction.Parameters = parameters;

					return _Invoke(instruction);
				}
			#endregion
			#region Create
				/// <summary>
				/// Creates a registered object on the other side.
				/// </summary>
				public object Create(string name, params object[] parameters)
				{
					_CreateConnectionIfNeeded();
					var instruction = new InstructionCreateObject();

					instruction.Name = name;
					instruction.Parameters = parameters;

					return _Invoke(instruction);
				}

				/// <summary>
				/// Creates an interface registered on the other side, using its default name and constructor.
				/// </summary>
				public T Create<T>()
				{
					return (T)Create(typeof(T).FullName);
				}
			#endregion

			#region Out Values in General
				private static readonly Dictionary<MethodInfo, int[]> _outIndexesDictionary = new Dictionary<MethodInfo, int[]>();
				private static readonly YieldReaderWriterLock _outIndexesDictionaryLock = new YieldReaderWriterLock();
				internal static int[] _GetOutIndexes(MethodInfo methodInfo)
				{
					bool result;
					int[] outIndexes;
					using(_outIndexesDictionaryLock.ReadLock())
						result = _outIndexesDictionary.TryGetValue(methodInfo, out outIndexes);

					if (!result)
					{
						using(var upgradeableLock = _outIndexesDictionaryLock.UpgradeableLock())
						{
							result = _outIndexesDictionary.TryGetValue(methodInfo, out outIndexes);
							if (result)
								return outIndexes;

							List<int> list = new List<int>();
							int parameterIndex = -1;
							foreach(var parameter in methodInfo.GetParameters())
							{
								parameterIndex++;

								if (parameter.ParameterType.IsByRef)
									list.Add(parameterIndex);
							}

							if (list.Count > 0)
								outIndexes = list.ToArray();

							upgradeableLock.Upgrade();
							_outIndexesDictionary.Add(methodInfo, outIndexes);
						}
					}

					return outIndexes;
				}
				private static void _ProcessOut(MethodInfo methodInfo, object[] resultOutParameters, object[] outParameters)
				{
					if (resultOutParameters == null)
						return;

					var outIndexes = _GetOutIndexes(methodInfo);
					int resultIndex = -1;
					foreach(int index in outIndexes)
					{
						resultIndex++;

						outParameters[index] = resultOutParameters[resultIndex];
					}
				}
				internal static object[] _GetOutValues(MethodInfo methodInfo, object[] outParameters)
				{
					if (outParameters == null)
						return null;

					var outIndexes = _GetOutIndexes(methodInfo);
					if (outIndexes == null)
						return null;

					int count = outIndexes.Length;
					object[] result = new object[count];
					for(int i=0; i<count; i++)
					{
						int index = outIndexes[i];
						result[i] = outParameters[index];
					}
					return result;
				}
			#endregion
		#endregion
		#region Events
			#region Disposed
				/// <summary>
				/// Event invoked when this RemotingClient is disposed.
				/// To guarantee that it will be invoked, set this event in the RemotingClientParameters before
				/// creating it.
				/// </summary>
				public event EventHandler Disposed;
			#endregion
		#endregion
	}
}

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)

Share

About the Author

Paulo Zemek
Engineer Microsoft Corporation
United States United States
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.

Now I just started working as a Senior Software Engineer at Microsoft.

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, 2015
Microsoft MVP 2013-2014

You may also be interested in...

| Advertise | Privacy | Terms of Use | Mobile
Web04 | 2.8.150723.1 | Last Updated 1 Dec 2011
Article Copyright 2011 by Paulo Zemek
Everything else Copyright © CodeProject, 1999-2015
Layout: fixed | fluid