andrew4582

ResourcePool<T>

Jul 24th, 2011
263
0
Never
Not a member of Pastebin yet? Sign Up, it unlocks many cool features!
C# 8.80 KB | None | 0 0
  1. #undef DEBUG
  2.  
  3. using System;
  4. using System.Collections.Concurrent;
  5. using System.Diagnostics;
  6. using System.Threading;
  7.  
  8. namespace Core.Threading {
  9.  
  10.     /// <summary>
  11.     /// Resource pool that uses a <see cref="System.Collections.Concurrent.BlockingCollection<T>"/> that limits the use of specified resource.
  12.     /// 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.
  13.     /// </summary>
  14.     /// <example>
  15.     /*
  16.          class ResourcePoolExample {
  17.          
  18.         static void Main() {
  19.  
  20.             ExpensiveResource.InstanceCount = 0;
  21.  
  22.             int threadCount = 8;
  23.             int maxInstances = 2;
  24.  
  25.             //create factory to create ExpensiveResource instances
  26.             Func<ExpensiveResource> factory = () => {
  27.                 return new ExpensiveResource();
  28.             };
  29.  
  30.             ResourcePool<ExpensiveResource> pool = new ResourcePool<ExpensiveResource>(maxInstances,factory,5000);
  31.            
  32.             //event to signal when all threads are complete
  33.             CountdownEvent cde = new CountdownEvent(threadCount);
  34.  
  35.             //event to signal all threads to go
  36.             ManualResetEventSlim goevent = new ManualResetEventSlim(false);
  37.  
  38.             //Action that runs in every thread
  39.             Action<int> testAction = (id) => {
  40.  
  41.                 string indent = "".PadLeft(id,'-');
  42.                
  43.                 //wait for every thread
  44.                 goevent.Wait();
  45.  
  46.                 //delay each thread to simulate reality.
  47.                 Random rnd = new Random(Guid.NewGuid().ToString().GetHashCode());
  48.                 Thread.Sleep(rnd.Next(10,100));
  49.  
  50.                 ExpensiveResource resource = null;
  51.  
  52.                 try {
  53.  
  54.                     //we grab our resource
  55.                     resource = pool.Take();
  56.  
  57.                     Console.WriteLine(indent + "Action{0} Take -> Instance_{1}",id,resource.InstanceId);
  58.  
  59.                     //execute process
  60.                     resource.ExpensiveProcess();
  61.  
  62.                 }
  63.                 catch(Exception error) {
  64.                     Console.WriteLine("ERROR ExpensiveProcess: {0}",error.Message);
  65.                 }
  66.                 finally {
  67.                     //return our resource to the pool when finished
  68.                     pool.Return(resource);
  69.                     Console.WriteLine(indent + "Action{0} Return -> Instance_{1}",id,resource.InstanceId);
  70.  
  71.                     //signal when finished
  72.                     cde.Signal();
  73.                 }
  74.             };
  75.  
  76.             //create our worker threads
  77.             for(int i = 0;i < threadCount;i++) {
  78.                 int workerId = i + 1;
  79.  
  80.                 ThreadPool.QueueUserWorkItem((state) => {
  81.                     testAction((int)state);
  82.                 },workerId);
  83.             }
  84.  
  85.             Thread.Sleep(10);
  86.      
  87.             //fire off all threads
  88.             goevent.Set();
  89.  
  90.             //wait until all threads are completed
  91.             cde.Wait();
  92.  
  93.             //uncomment to use 'Parallel lib'
  94.             //Parallel.For(0,threadCount,testAction);
  95.  
  96.             Console.WriteLine();
  97.             Console.WriteLine("threadCount {0}",threadCount);
  98.             Console.WriteLine("pool.MaxInstances: {0}",pool.MaxInstances);
  99.             Console.WriteLine("pool.InstancesCreated: {0}",pool.InstancesCreated);
  100.             Console.WriteLine("ExpensiveResource.InstanceCount: {0}",ExpensiveResource.InstanceCount);
  101.             Console.WriteLine("pool.Count: {0}",pool.Count);
  102.             Console.WriteLine();
  103.             Console.WriteLine("Test Finished");
  104.         }
  105.     }
  106.     class ExpensiveResource {
  107.  
  108.         //global instances created
  109.         public static int InstanceCount;
  110.  
  111.         //resource id
  112.         public int InstanceId { get; private set; }
  113.  
  114.         public ExpensiveResource() {
  115.            
  116.             InstanceId = Interlocked.Increment(ref InstanceCount);
  117.             Console.WriteLine("Created Instance #: {0}",InstanceId);
  118.         }
  119.  
  120.         public void ExpensiveProcess() {
  121.             //simulate long process
  122.             Random rnd = new Random(Guid.NewGuid().ToString().GetHashCode());
  123.             int st = rnd.Next(500,1000);
  124.             Thread.Sleep(st);
  125.         }
  126.     }
  127.      */
  128.     /// </example>
  129.     /// <typeparam name="T">class</typeparam>
  130.     public class ResourcePool<T> where T:class {
  131.  
  132.         protected readonly object _syncroot = new object();
  133.         protected readonly BlockingCollection<T> _items;
  134.         protected readonly Func<T> _factory;
  135.         readonly int _timeout;
  136.         volatile int _maxInstances;
  137.         volatile int _created;
  138.  
  139.         /// <summary>Instances that were created</summary>
  140.         public int InstancesCreated { get { return _created; } }
  141.  
  142.         /// <summary>Maximum instances that could be created</summary>
  143.         public int MaxInstances { get { return _maxInstances; } }
  144.  
  145.         /// <summary>Gets the number of current objects available in the pool</summary>
  146.         public int Count { get { return _items.Count; } }
  147.  
  148.         ///<summary>Gets the wait time if all resources are taken and no more can be created, default is -1 (infinity)</summary>
  149.         public int Timeout { get { return _timeout; } }
  150.  
  151.         /// <summary>
  152.         /// ctr
  153.         /// </summary>
  154.         /// <param name="maxInstances">Maximum number of instances that are to be created</param>
  155.         /// <param name="factory">Factory function to create an instance of the specified Type </param>
  156.         /// <param name="timeout">The wait time if all resources are taken and no more can be created, default is -1 (infinity)</param>
  157.         public ResourcePool(int maxInstances,Func<T> factory,int timeout = -1) {
  158.  
  159.             if(factory == null)
  160.                 throw new ArgumentNullException("factory");
  161.  
  162.             this._maxInstances = Math.Max(1,maxInstances);
  163.             this._items = new BlockingCollection<T>(_maxInstances);
  164.             this._factory = factory;
  165.             this._timeout = timeout;
  166.         }
  167.  
  168.         /// <summary>Increments the MaxInstances value</summary>
  169.         public void IncrementMaxInstances(int count) {
  170.             _maxInstances += count;
  171.         }
  172.  
  173.         protected virtual void OnCreated(T instance) {
  174.             return;
  175.         }
  176.         protected virtual bool CanReturn(T instance) {
  177.             return true;
  178.         }
  179.  
  180.         /// <summary>
  181.         /// Returns an object back to the resource pool to be available the next time <see cref="Take"/> is called
  182.         /// </summary>
  183.         /// <param name="item">The object to be placed back into the pool</param>
  184.         /// <returns>True if object is added back to the pool, otherwise false</returns>
  185.         public virtual bool Return(T item) {
  186.             if(item == null)
  187.                 return false;
  188.  
  189.             bool added = false;
  190.             if(CanReturn(item)) {
  191.                 added = _items.TryAdd(item);
  192.             }
  193. #if DEBUG
  194.             Debug.WriteLine("Return-> hash: {0} - added: {1}",item.GetHashCode(),added);
  195. #endif
  196.             return added;
  197.         }
  198.  
  199.         /// <summary>
  200.         /// 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.
  201.         /// </summary>
  202.         /// <returns>An existing object from the pool or a new instnace</returns>
  203.         public virtual T Take() {
  204.  
  205. #if DEBUG
  206.             int hash = -1;
  207.             bool added = false;
  208.            
  209.             var sw = PerformanceStopWatch.Start((duration) => {
  210.                 Debug.WriteLine("Take-> hash: {0} - duration: {1} ms, added: {2}, created: {3}",hash,duration.TotalMilliseconds,added,created);
  211.             });
  212. #endif
  213.             bool created = false;
  214.             T item = null;
  215.  
  216.             //take out the resource immediately
  217.             if(_items.TryTake(out item,0))
  218.                 return item;
  219.  
  220.             lock(_syncroot) {
  221.  
  222.                 //create resource if we can
  223.                 if(_created < _maxInstances) {
  224.  
  225.                     item = _factory();
  226.  
  227.                     _created++;
  228.                     _items.Add(item);
  229.                     created = true;
  230.                 }
  231.             }
  232.  
  233.             if(created) {
  234.                 OnCreated(item);
  235.                 return item;
  236.             }
  237.  
  238.             if(_created >= _maxInstances) {
  239.  
  240.                 //all resources are created, wait until there is one available
  241.                 if(!_items.TryTake(out item,_timeout))
  242.                     //we timed out
  243.                     throw new TimeoutException();
  244.             }
  245.  
  246. #if DEBUG
  247.             if(item != null)
  248.                 hash = item.GetHashCode();
  249.  
  250.             sw.Stop();
  251. #endif
  252.             return item;
  253.         }
  254.     }
  255. }
Advertisement
Add Comment
Please, Sign In to add comment