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

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