Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- #undef DEBUG
- using System;
- using System.Collections.Concurrent;
- using System.Diagnostics;
- using System.Threading;
- 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;
- var sw = PerformanceStopWatch.Start((duration) => {
- Debug.WriteLine("Take-> hash: {0} - duration: {1} ms, added: {2}, created: {3}",hash,duration.TotalMilliseconds,added,created);
- });
- #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();
- sw.Stop();
- #endif
- return item;
- }
- }
- }
Advertisement
Add Comment
Please, Sign In to add comment