mcs/class/corlib/System.Threading/ThreadPool.cs

   1 //
   2 // System.Threading.ThreadPool
   3 //
   4 // Author:
   5 //   Patrik Torstensson (patrik.torstensson@labs2.com)
   6 //   Dick Porter (dick@ximian.com)
   7 //
   8 // (C) Ximian, Inc.  http://www.ximian.com
   9 // (C) Patrik Torstensson
  10 //
  11 using System;
  12 using System.Collections;
  13
  14 namespace System.Threading
  15 {
  16    /// <summary> (Patrik T notes)
  17    /// This threadpool is focused on saving resources not giving max performance.
  18    ///
  19    /// Note, this class is not perfect but it works. ;-) Should also replace
  20    /// the queue with an internal one (performance)
  21    ///
  22    /// This class should also use a specialized queue to increase performance..
  23    /// </summary
  24    ///
  25    public sealed class ThreadPool {
  26       internal struct ThreadPoolWorkItem {
  27          public WaitCallback _CallBack;
  28          public object _Context;
  29       }
  30
  31       private int _ThreadTimeout;
  32
  33       private long _MaxThreads;
  34       private long _CurrentThreads;
  35       private long _ThreadsInUse;
  36       private long _RequestInQueue;
  37       private long _ThreadCreateTriggerRequests;
  38
  39       private Thread _MonitorThread;
  40       private Queue _RequestQueue;
  41
  42       private ArrayList _Threads;
  43       private ManualResetEvent _DataInQueue;
  44
  45       static ThreadPool _Threadpool;
  46
  47       static ThreadPool() {
  48          _Threadpool = new ThreadPool();
  49       }
  50
  51       private ThreadPool() {
  52          // 30 sec timeout default
  53          _ThreadTimeout = 30 * 1000;
  54
  55          // Used to signal that there is data in the queue
  56          _DataInQueue = new ManualResetEvent(false);
  57
  58          _Threads = ArrayList.Synchronized(new ArrayList());
  59
  60          // Holds requests..
  61          _RequestQueue = Queue.Synchronized(new Queue(128));
  62
  63          _MaxThreads = 64;
  64          _CurrentThreads = 0;
  65          _RequestInQueue = 0;
  66          _ThreadsInUse = 0;
  67          _ThreadCreateTriggerRequests = 5;
  68
  69          // Keeps track of requests in the queue and inreases the number of threads if neededs
  70          _MonitorThread = new Thread(new ThreadStart(MonitorThread));
  71          _MonitorThread.Start();
  72       }
  73
  74       internal void RemoveThread() {
  75          Interlocked.Decrement(ref _CurrentThreads);
  76          _Threads.Remove(Thread.CurrentThread);
  77       }
  78
  79       internal void CheckIfStartThread() {
  80          bool bCreateThread = false;
  81
  82          if (_CurrentThreads == 0) {
  83             bCreateThread = true;
  84          }
  85
  86          if ((_MaxThreads == -1 || _CurrentThreads < _MaxThreads) && _ThreadsInUse > 0 && _RequestInQueue > _ThreadCreateTriggerRequests) {
  87             bCreateThread = true;
  88          }
  89
  90          if (bCreateThread) {
  91             Interlocked.Increment(ref _CurrentThreads);
  92
  93             Thread Start = new Thread(new ThreadStart(WorkerThread));
  94             Start.Start();
  95             Start.IsThreadPoolThreadInternal = true;
  96
  97             _Threads.Add(Start);
  98          }
  99       }
 100
 101       internal void AddItem(ref ThreadPoolWorkItem Item) {
 102          CheckIfStartThread();
 103
 104          if (Interlocked.Increment(ref _RequestInQueue) == 1) {
 105             _DataInQueue.Set();
 106          }
 107
 108          _RequestQueue.Enqueue(Item);
 109       }
 110
 111       // Work Thread main function
 112       internal void WorkerThread() {
 113          bool bWaitForData = true;
 114
 115          while (true) {
 116             if (bWaitForData) {
 117                if (!_DataInQueue.WaitOne(_ThreadTimeout, false)) {
 118                   // timeout
 119                   RemoveThread();
 120                   return;
 121                }
 122             }
 123
 124             Interlocked.Increment(ref _ThreadsInUse);
 125
 126             try {
 127                ThreadPoolWorkItem oItem = (ThreadPoolWorkItem) _RequestQueue.Dequeue();
 128
 129                if (Interlocked.Decrement(ref _RequestInQueue) == 0) {
 130                   _DataInQueue.Reset();
 131                }
 132
 133                oItem._CallBack(oItem._Context);
 134             }
 135             catch (InvalidOperationException) {
 136                // Queue empty
 137                bWaitForData = true;
 138             }
 139             catch (ThreadAbortException) {
 140                // We will leave here.. (thread abort can't be handled)
 141                RemoveThread();
 142             }
 143             finally {
 144                Interlocked.Decrement(ref _ThreadsInUse);
 145             }
 146          }
 147       }
 148
 149       internal void MonitorThread() {
 150          while (true) {
 151             Thread.Sleep(500);
 152
 153             CheckIfStartThread();
 154          }
 155       }
 156
 157       internal bool QueueUserWorkItemInternal(WaitCallback callback) {
 158          return QueueUserWorkItem(callback, null);
 159       }
 160
 161       internal bool QueueUserWorkItemInternal(WaitCallback callback, object context) {
 162          ThreadPoolWorkItem Item = new ThreadPoolWorkItem();
 163
 164          Item._CallBack = callback;
 165          Item._Context = context;
 166
 167          AddItem(ref Item);
 168
 169          // LAMESPEC: Return value? should use exception here if anything goes wrong
 170          return true;
 171       }
 172
 173       public static bool BindHandle(IntPtr osHandle) {
 174          throw new NotSupportedException("This is a win32 specific method, not supported Mono");
 175                 }
 176
 177                 public static bool QueueUserWorkItem(WaitCallback callback) {
 178          return _Threadpool.QueueUserWorkItemInternal(callback);
 179                 }
 180
 181                 public static bool QueueUserWorkItem(WaitCallback callback, object state) {
 182          return _Threadpool.QueueUserWorkItemInternal(callback, state);
 183                 }
 184
 185       public static bool UnsafeQueueUserWorkItem(WaitCallback callback, object state) {
 186          return _Threadpool.QueueUserWorkItemInternal(callback, state);
 187       }
 188
 189       [MonoTODO]
 190                 public static RegisteredWaitHandle RegisterWaitForSingleObject(WaitHandle waitObject, WaitOrTimerCallback callback, object state, int millisecondsTimeOutInterval, bool executeOnlyOnce) {
 191                         if (millisecondsTimeOutInterval < -1) {
 192                                 throw new ArgumentOutOfRangeException("timeout < -1");
 193                         }
 194
 195          throw new NotImplementedException();
 196       }
 197
 198                 [MonoTODO]
 199                 public static RegisteredWaitHandle RegisterWaitForSingleObject(WaitHandle waitObject, WaitOrTimerCallback callback, object state, long millisecondsTimeOutInterval, bool executeOnlyOnce) {
 200                         if (millisecondsTimeOutInterval < -1) {
 201                                 throw new ArgumentOutOfRangeException("timeout < -1");
 202                         }
 203
 204          throw new NotImplementedException();
 205       }
 206
 207                 [MonoTODO]
 208                 public static RegisteredWaitHandle RegisterWaitForSingleObject(WaitHandle waitObject, WaitOrTimerCallback callback, object state, TimeSpan timeout, bool executeOnlyOnce) {
 209                         // LAMESPEC: I assume it means "timeout" when it says "millisecondsTimeOutInterval"
 210                         if (timeout.Milliseconds < -1) {
 211                                 throw new ArgumentOutOfRangeException("timeout < -1");
 212                         }
 213                         if (timeout.Milliseconds > Int32.MaxValue) {
 214                                 throw new NotSupportedException("timeout too large");
 215                         }
 216
 217          throw new NotImplementedException();
 218       }
 219
 220       [CLSCompliant(false)][MonoTODO]
 221                 public static RegisteredWaitHandle RegisterWaitForSingleObject(WaitHandle waitObject, WaitOrTimerCallback callback, object state, uint millisecondsTimeOutInterval, bool executeOnlyOnce) {
 222          throw new NotImplementedException();
 223       }
 224
 225                 [MonoTODO]
 226                 public static RegisteredWaitHandle UnsafeRegisterWaitForSingleObject(WaitHandle waitObject, WaitOrTimerCallback callback, object state, int millisecondsTimeOutInterval, bool executeOnlyOnce) {
 227          throw new NotImplementedException();
 228       }
 229
 230                 [MonoTODO]
 231                 public static RegisteredWaitHandle UnsafeRegisterWaitForSingleObject(WaitHandle waitObject, WaitOrTimerCallback callback, object state, long millisecondsTimeOutInterval, bool executeOnlyOnce) {
 232          throw new NotImplementedException();
 233                 }
 234
 235                 [MonoTODO]
 236                 public static RegisteredWaitHandle UnsafeRegisterWaitForSingleObject(WaitHandle waitObject, WaitOrTimerCallback callback, object state, TimeSpan timeout, bool executeOnlyOnce) {
 237          throw new NotImplementedException();
 238       }
 239
 240                 [CLSCompliant(false)][MonoTODO]
 241                 public static RegisteredWaitHandle UnsafeRegisterWaitForSingleObject(WaitHandle waitObject, WaitOrTimerCallback callback, object state, uint millisecondsTimeOutInterval, bool executeOnlyOnce) {
 242          throw new NotImplementedException();
 243       }
 244         }
 245 }