Resource Pool

using System;
using System.Linq;
using System.Collections.Concurrent;
using System.Diagnostics;
using System.Threading;
using Core.Threading;

namespace ConsoleApplication1
{
    class ResourcePoolExample
    {

        static void Main()
        {

            ExpensiveResource.InstanceCount = 0;

            int threadCount = 25;
            int maxInstances = 2;

            int min = threadCount- 2;

            ThreadPool.SetMinThreads(min, min);
            //create factory to create ExpensiveResource instances
            Func<ExpensiveResource> factory = () =>
            {
                return new ExpensiveResource();
            };

            ResourcePool<ExpensiveResource> pool = new ResourcePool<ExpensiveResource>(maxInstances, factory, 5000);

            //event to signal when all threads are started
            CountdownEvent allStarted = new CountdownEvent(threadCount);

            //event to signal when all threads are complete
            CountdownEvent allEnded = new CountdownEvent(threadCount);

            //event to signal all threads to go
            ManualResetEventSlim goevent = new ManualResetEventSlim(false);

            //Action that runs in every thread
            Action<int> testAction = (id) =>
            {

                string indent = "".PadLeft(id, '-');

                //signal this thread has started
                allStarted.Signal();
                Console.Write(".");
                //wait for every thread
                goevent.Wait();

                //delay each thread to simulate reality.
                Random rnd = new Random(Guid.NewGuid().ToString().GetHashCode());
                Thread.Sleep(rnd.Next(10, 100));

                ExpensiveResource resource = null;

                try
                {

                    //we grab our resource
                    resource = pool.Take();

                    Console.WriteLine(indent + "Action{0} Take -> Instance_{1}", id, resource.InstanceId);

                    //execute process
                    resource.ExpensiveProcess();

                }
                catch (Exception error)
                {
                    Console.WriteLine("ERROR ExpensiveProcess: {0}", error.Message);
                }
                finally
                {
                    //return our resource to the pool when finished
                    pool.Return(resource);
                    Console.WriteLine(indent + "Action{0} Return -> Instance_{1}", id, resource.InstanceId);

                    //signal when finished
                    allEnded.Signal();
                }
            };

            Console.WriteLine("Starting {0:N0} threads", threadCount);

            //create our worker threads
            for (int i = 0; i < threadCount; i++)
            {
                int workerId = i + 1;

                ThreadPool.QueueUserWorkItem((state) =>
                {
                    testAction((int)state);
                }, workerId);


            }
            Console.WriteLine();

            //wait for all threads  to startup
            allStarted.Wait();
            Console.WriteLine();
            Console.WriteLine("All {0:N0} threads have started",threadCount);
            //fire off all threads
            goevent.Set();

            //wait until all threads are completed
            allEnded.Wait();

            //uncomment to use 'Parallel lib'
            //Parallel.For(0,threadCount,testAction);

            Console.WriteLine();
            Console.WriteLine("threadCount {0}", threadCount);
            Console.WriteLine("pool.MaxInstances: {0}", pool.MaxInstances);
            Console.WriteLine("pool.InstancesCreated: {0}", pool.InstancesCreated);
            Console.WriteLine("ExpensiveResource.InstanceCount: {0}", ExpensiveResource.InstanceCount);
            Console.WriteLine("pool.Count: {0}", pool.Count);
            Console.WriteLine();
            Console.WriteLine("Test Finished");
        }
    }
    class ExpensiveResource
    {

        //global instances created
        public static int InstanceCount;

        //resource id
        public int InstanceId { get; private set; }

        public ExpensiveResource()
        {

            InstanceId = Interlocked.Increment(ref InstanceCount);
            Console.WriteLine("Created Instance #: {0}", InstanceId);
        }

        public void ExpensiveProcess()
        {
            //simulate long process
            Random rnd = new Random(Guid.NewGuid().ToString().GetHashCode());
            int st = rnd.Next(500, 1000);
            Thread.Sleep(st);
        }
    }
}


namespace Core.Threading {

    /// <summary>
    /// Resource pool that uses a <see cref="System.Collections.Concurrent.BlockingCollection<T>"/> that limits the use of specified resource.
    /// If an item in the pool is requested and the instance count is under the maximum a new instance of the resource is created via <see cref="Func<T>"/> parameter.
    /// </summary>
    /// <example>
    /*
         class ResourcePoolExample {

        static void Main() {

            ExpensiveResource.InstanceCount = 0;

            int threadCount = 8;
            int maxInstances = 2;

            //create factory to create ExpensiveResource instances
            Func<ExpensiveResource> factory = () => {
                return new ExpensiveResource();
            };

            ResourcePool<ExpensiveResource> pool = new ResourcePool<ExpensiveResource>(maxInstances,factory,5000);

            //event to signal when all threads are complete
            CountdownEvent cde = new CountdownEvent(threadCount);

            //event to signal all threads to go
            ManualResetEventSlim goevent = new ManualResetEventSlim(false);

            //Action that runs in every thread
            Action<int> testAction = (id) => {

                string indent = "".PadLeft(id,'-');

                //wait for every thread
                goevent.Wait();

                //delay each thread to simulate reality.
                Random rnd = new Random(Guid.NewGuid().ToString().GetHashCode());
                Thread.Sleep(rnd.Next(10,100));

                ExpensiveResource resource = null;

                try {

                    //we grab our resource
                    resource = pool.Take();

                    Console.WriteLine(indent + "Action{0} Take -> Instance_{1}",id,resource.InstanceId);

                    //execute process
                    resource.ExpensiveProcess();

                }
                catch(Exception error) {
                    Console.WriteLine("ERROR ExpensiveProcess: {0}",error.Message);
                }
                finally {
                    //return our resource to the pool when finished
                    pool.Return(resource);
                    Console.WriteLine(indent + "Action{0} Return -> Instance_{1}",id,resource.InstanceId);

                    //signal when finished
                    cde.Signal();
                }
            };

            //create our worker threads
            for(int i = 0;i < threadCount;i++) {
                int workerId = i + 1;

                ThreadPool.QueueUserWorkItem((state) => {
                    testAction((int)state);
                },workerId);
            }

            Thread.Sleep(10);

            //fire off all threads
            goevent.Set();

            //wait until all threads are completed
            cde.Wait();

            //uncomment to use 'Parallel lib'
            //Parallel.For(0,threadCount,testAction);

            Console.WriteLine();
            Console.WriteLine("threadCount {0}",threadCount);
            Console.WriteLine("pool.MaxInstances: {0}",pool.MaxInstances);
            Console.WriteLine("pool.InstancesCreated: {0}",pool.InstancesCreated);
            Console.WriteLine("ExpensiveResource.InstanceCount: {0}",ExpensiveResource.InstanceCount);
            Console.WriteLine("pool.Count: {0}",pool.Count);
            Console.WriteLine();
            Console.WriteLine("Test Finished");
        }
    }
    class ExpensiveResource {

        //global instances created
        public static int InstanceCount;

        //resource id
        public int InstanceId { get; private set; }

        public ExpensiveResource() {

            InstanceId = Interlocked.Increment(ref InstanceCount);
            Console.WriteLine("Created Instance #: {0}",InstanceId);
        }

        public void ExpensiveProcess() {
            //simulate long process
            Random rnd = new Random(Guid.NewGuid().ToString().GetHashCode());
            int st = rnd.Next(500,1000);
            Thread.Sleep(st);
        }
    }
     */
    /// </example>
    /// <typeparam name="T">class</typeparam>
    public class ResourcePool<T> where T:class {

        protected readonly object _syncroot = new object();
        protected readonly BlockingCollection<T> _items;
        protected readonly Func<T> _factory;
        readonly int _timeout;
        volatile int _maxInstances;
        volatile int _created;

        /// <summary>Instances that were created</summary>
        public int InstancesCreated { get { return _created; } }

        /// <summary>Maximum instances that could be created</summary>
        public int MaxInstances { get { return _maxInstances; } }

        /// <summary>Gets the number of current objects available in the pool</summary>
        public int Count { get { return _items.Count; } }

        ///<summary>Gets the wait time if all resources are taken and no more can be created, default is -1 (infinity)</summary>
        public int Timeout { get { return _timeout; } }

        /// <summary>
        /// ctr
        /// </summary>
        /// <param name="maxInstances">Maximum number of instances that are to be created</param>
        /// <param name="factory">Factory function to create an instance of the specified Type </param>
        /// <param name="timeout">The wait time if all resources are taken and no more can be created, default is -1 (infinity)</param>
        public ResourcePool(int maxInstances,Func<T> factory,int timeout = -1) {

            if(factory == null)
                throw new ArgumentNullException("factory");

            this._maxInstances = Math.Max(1,maxInstances);
            this._items = new BlockingCollection<T>(_maxInstances);
            this._factory = factory;
            this._timeout = timeout;
        }

        /// <summary>Increments the MaxInstances value</summary>
        public void IncrementMaxInstances(int count) {
            _maxInstances += count;
        }

        protected virtual void OnCreated(T instance) {
            return;
        }
        protected virtual bool CanReturn(T instance) {
            return true;
        }

        /// <summary>
        /// Returns an object back to the resource pool to be available the next time <see cref="Take"/> is called
        /// </summary>
        /// <param name="item">The object to be placed back into the pool</param>
        /// <returns>True if object is added back to the pool, otherwise false</returns>
        public virtual bool Return(T item) {
            if(item == null)
                return false;

            bool added = false;
            if(CanReturn(item)) {
                added = _items.TryAdd(item);
            }
#if DEBUG
            Debug.WriteLine("Return-> hash: {0} - added: {1}",item.GetHashCode(),added);
#endif
            return added;
        }

        /// <summary>
        /// Either takes an existing object from the pool, or it creates a new instance if the <see cref="InstancesCreated"/> have not exceed the <see cref="MaxInstances"/> value.
        /// </summary>
        /// <returns>An existing object from the pool or a new instnace</returns>
        public virtual T Take() {

#if DEBUG
            int hash = -1;
            bool added = false;

#endif
            bool created = false;
            T item = null;

            //take out the resource immediately
            if(_items.TryTake(out item,0))
                return item;

            lock(_syncroot) {

                //create resource if we can
                if(_created < _maxInstances) {

                    item = _factory();

                    _created++;
                    _items.Add(item);
                    created = true;
                }
            }

            if(created) {
                OnCreated(item);
                return item;
            }

            if(_created >= _maxInstances) {

                //all resources are created, wait until there is one available
                if(!_items.TryTake(out item,_timeout))
                    //we timed out
                    throw new TimeoutException();
            }

#if DEBUG
            if(item != null)
                hash = item.GetHashCode();

#endif
            return item;
        }
    }
}