Mercurial > hg > ucis.core

comparison ThreadPool.cs @ 7:4b78cc5f116b

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
Fixes and improvements (some untested)
author Ivo Smits <Ivo@UCIS.nl>
date Sun, 13 Jan 2013 18:44:17 +0100
parents 3ab940a0c7a0
children 1474f92cf7e7
comparison
equal deleted inserted replaced
6:5ce7a138fdba 7:4b78cc5f116b
1 using System; 1 using System;
2 using System.Collections.Generic;
3 using System.Collections.ObjectModel;
2 using System.Threading; 4 using System.Threading;
3 using System.Collections.Generic;
4 5
5 namespace UCIS { 6 namespace UCIS {
6 public class ThreadPool { 7 public class ThreadPool {
7 private static ThreadPool pManager = null; 8 private static ThreadPool pManager = null;
8 9
47 public Exception Exception; 48 public Exception Exception;
48 public bool ThrowError; 49 public bool ThrowError;
49 } 50 }
50 51
51 private List<ThreadInfo> pThreads = new List<ThreadInfo>(); 52 private List<ThreadInfo> pThreads = new List<ThreadInfo>();
52 private List<ThreadInfo> pBusyThreads = new List<ThreadInfo>(); 53 private int pBusyThreads = 0;
53 private Queue<ThreadInfo> pIdleThreads = new Queue<ThreadInfo>(); 54 private Queue<ThreadInfo> pIdleThreads = new Queue<ThreadInfo>();
54 private int pThreadsMax; 55 private int pThreadsMax;
55 private int pThreadsMinIdle; 56 private int pThreadsMinIdle;
56 private int pThreadsMaxIdle; 57 private int pThreadsMaxIdle;
57 58
58 public event OnExceptionEventHandler OnException; 59 public event OnExceptionEventHandler OnException;
59 public delegate void OnExceptionEventHandler(ThreadPool sender, ExceptionEventArgs e); 60 public delegate void OnExceptionEventHandler(ThreadPool sender, ExceptionEventArgs e);
60 61
61 public System.Collections.ObjectModel.ReadOnlyCollection<ThreadInfo> Threads { 62 public ReadOnlyCollection<ThreadInfo> Threads { get { return pThreads.AsReadOnly(); } }
62 get { 63
63 return new System.Collections.ObjectModel.ReadOnlyCollection<ThreadInfo>(pThreads); 64 public ThreadPool() : this(250, 0, 5) { }
64 }
65 }
66
67 public ThreadPool() : this(250, 1, 5) { }
68 65
69 public ThreadPool(int MaxThreads, int MinIdle, int MaxIdle) { 66 public ThreadPool(int MaxThreads, int MinIdle, int MaxIdle) {
70 int I = 0;
71 if (MaxThreads < 0) { 67 if (MaxThreads < 0) {
72 throw new ArgumentOutOfRangeException("ThreadsMaxIdle", "ThreadsMaxIdle must greater than 0"); 68 throw new ArgumentOutOfRangeException("ThreadsMaxIdle", "ThreadsMaxIdle must greater than 0");
73 } else if (MaxThreads < MaxIdle) { 69 } else if (MaxThreads < MaxIdle) {
74 throw new ArgumentOutOfRangeException("ThreadsMax", "ThreadsMax must be greater than or equal to ThreadsMaxIdle"); 70 throw new ArgumentOutOfRangeException("ThreadsMax", "ThreadsMax must be greater than or equal to ThreadsMaxIdle");
75 } else if (MaxIdle < 0) { 71 } else if (MaxIdle < 0) {
80 throw new ArgumentOutOfRangeException("ThreadsMaxIdle", "ThreadsMaxIdle must be greater than or equal to ThreadsMinIdle"); 76 throw new ArgumentOutOfRangeException("ThreadsMaxIdle", "ThreadsMaxIdle must be greater than or equal to ThreadsMinIdle");
81 } 77 }
82 pThreadsMax = MaxThreads; 78 pThreadsMax = MaxThreads;
83 pThreadsMinIdle = MinIdle; 79 pThreadsMinIdle = MinIdle;
84 pThreadsMaxIdle = MaxIdle; 80 pThreadsMaxIdle = MaxIdle;
85 for (I = 1; I <= pThreadsMinIdle; I++) { 81 for (int I = 1; I <= pThreadsMinIdle; I++) {
86 StartThread(false); 82 StartThread(false);
87 } 83 }
88 } 84 }
89 85
90 public int ThreadsIdle { 86 public int ThreadsIdle { get { return pIdleThreads.Count; } }
91 get { return pIdleThreads.Count; } 87 public int ThreadsBusy { get { return pBusyThreads; } }
92 } 88 public int ThreadsAlive { get { return pThreads.Count; } }
93 public int ThreadsBusy {
94 get { return pBusyThreads.Count; }
95 }
96 public int ThreadsAlive {
97 get { return pThreads.Count; }
98 }
99 public int ThreadsMinIdle { 89 public int ThreadsMinIdle {
100 get { return pThreadsMinIdle; } 90 get { return pThreadsMinIdle; }
101 set { 91 set {
102 if (value > pThreadsMaxIdle) { 92 if (value > pThreadsMaxIdle) {
103 throw new ArgumentOutOfRangeException("ThreadsMinIdle", "ThreadsMinIdle must be smaller than ThreadsMaxIdle"); 93 throw new ArgumentOutOfRangeException("ThreadsMinIdle", "ThreadsMinIdle must be smaller than ThreadsMaxIdle");
117 } 107 }
118 } 108 }
119 public int ThreadsMaxIdle { 109 public int ThreadsMaxIdle {
120 get { return pThreadsMaxIdle; } 110 get { return pThreadsMaxIdle; }
121 set { 111 set {
122 if (pThreadsMinIdle > value) { 112 if (pThreadsMinIdle > value) throw new ArgumentOutOfRangeException("ThreadsMaxIdle", "ThreadsMaxIdle must be greater than or equal to ThreadsMinIdle");
123 throw new ArgumentOutOfRangeException("ThreadsMaxIdle", "ThreadsMaxIdle must be greater than or equal to ThreadsMinIdle"); 113 if (value < 0) throw new ArgumentOutOfRangeException("ThreadsMaxIdle", "ThreadsMaxIdle must greater than or equal to 0");
124 } else if (value < 0) { 114 lock (pIdleThreads) {
125 throw new ArgumentOutOfRangeException("ThreadsMaxIdle", "ThreadsMaxIdle must greater than or equal to 0"); 115 while (value > pIdleThreads.Count) {
126 } else { 116 ThreadInfo T = pIdleThreads.Dequeue();
127 int I = 0; 117 T.Abort = true;
128 int C = 0; 118 T.WaitHandle.Set();
129 ThreadInfo T = default(ThreadInfo); 119 }
130 lock (pIdleThreads) { 120 }
131 C = pIdleThreads.Count; 121 pThreadsMaxIdle = value;
132 if (value < C) {
133 for (I = value; I <= C - 1; I++) {
134 T = pIdleThreads.Dequeue();
135 T.Abort = true;
136 T.WaitHandle.Set();
137 }
138 }
139 }
140 pThreadsMaxIdle = value;
141 }
142 } 122 }
143 } 123 }
144 public int ThreadsMax { 124 public int ThreadsMax {
145 get { return pThreadsMax; } 125 get { return pThreadsMax; }
146 set { 126 set {
147 if (pThreadsMaxIdle > value) { 127 if (pThreadsMaxIdle > value) throw new ArgumentOutOfRangeException("ThreadsMax", "ThreadsMax must be greater than or equal to ThreadsMaxIdle");
148 throw new ArgumentOutOfRangeException("ThreadsMax", "ThreadsMax must be greater than or equal to ThreadsMaxIdle"); 128 if (value <= 0) throw new ArgumentOutOfRangeException("ThreadsMax", "ThreadsMax must greater than 0");
149 } else if (value <= 0) { 129 pThreadsMax = value;
150 throw new ArgumentOutOfRangeException("ThreadsMax", "ThreadsMax must greater than 0");
151 } else {
152 pThreadsMax = value;
153 }
154 } 130 }
155 } 131 }
156 132
157 public WorkItem QueueWorkItem(WaitCallback Callback, object State) { 133 public WorkItem QueueWorkItem(WaitCallback Callback, object State) {
158 WorkItem WorkItem = new WorkItem(); 134 WorkItem WorkItem = new WorkItem() { Callback = Callback, State = State };
159 ThreadInfo Thread = null; 135 ThreadInfo Thread = null;
160 WorkItem.Callback = Callback;
161 WorkItem.State = State;
162 lock (pIdleThreads) { 136 lock (pIdleThreads) {
163 while (pIdleThreads.Count > 0) { 137 while (Thread == null && pIdleThreads.Count > 0) {
164 Thread = pIdleThreads.Dequeue(); 138 Thread = pIdleThreads.Dequeue();
165 if (!Thread.Abort) { 139 if (Thread.Abort) Thread = null;
166 break;
167 } else {
168 Thread = null;
169 }
170 } 140 }
171 } 141 }
172 if (Thread == null) { 142 if (Thread == null) {
173 if (pThreads.Count < pThreadsMax) { 143 if (pThreads.Count >= pThreadsMax) throw new ThreadStateException("Thread limit exceeded");
174 Thread = StartThread(true); 144 Thread = StartThread(true);
175 } else {
176 throw new ThreadStateException("Thread limit exceeded");
177 }
178 } 145 }
179 Thread.LastActive = DateTime.Now; 146 Thread.LastActive = DateTime.Now;
180 WorkItem.Thread = Thread; 147 WorkItem.Thread = Thread;
181 Thread.WorkItem = WorkItem; 148 Thread.WorkItem = WorkItem;
182 Thread.WaitHandle.Set(); 149 Thread.WaitHandle.Set();
233 if (Thread == null) throw new ArgumentNullException("state"); 200 if (Thread == null) throw new ArgumentNullException("state");
234 try { 201 try {
235 while (true) { 202 while (true) {
236 if (Thread.WaitHandle == null) throw new ArgumentNullException("WaitHandle"); 203 if (Thread.WaitHandle == null) throw new ArgumentNullException("WaitHandle");
237 if (!Thread.WaitHandle.WaitOne(1000, false)) { 204 if (!Thread.WaitHandle.WaitOne(1000, false)) {
238 if (pBusyThreads.Count == 0) { 205 if (pBusyThreads <= 0) return;
239 return; 206 continue;
240 } else { 207 }
241 continue;
242 }
243 }
244 if (Thread.Abort) break; 208 if (Thread.Abort) break;
245 209
246 Thread.Busy = true; 210 Thread.Busy = true;
247 lock (pBusyThreads) pBusyThreads.Add(Thread); 211 Interlocked.Increment(ref pBusyThreads);
248 try { 212 try {
249 if (Thread.WorkItem == null) throw new ArgumentNullException("WorkItem"); 213 if (Thread.WorkItem == null) throw new ArgumentNullException("WorkItem");
250 if (Thread.WorkItem.Callback == null) throw new ArgumentNullException("WorkItem.Callback"); 214 if (Thread.WorkItem.Callback == null) throw new ArgumentNullException("WorkItem.Callback");
251 Thread.WorkItem.Callback.Invoke(Thread.WorkItem.State); 215 Thread.WorkItem.Callback.Invoke(Thread.WorkItem.State);
252 } catch (ThreadAbortException ex) { 216 } catch (ThreadAbortException ex) {
260 if (e.ThrowError) { 224 if (e.ThrowError) {
261 Console.WriteLine("Exception in ThreadPool thread: " + e.Exception.ToString()); 225 Console.WriteLine("Exception in ThreadPool thread: " + e.Exception.ToString());
262 throw new Exception("Unhandled exception in work item", e.Exception); 226 throw new Exception("Unhandled exception in work item", e.Exception);
263 } 227 }
264 } finally { 228 } finally {
265 lock (pBusyThreads) pBusyThreads.Remove(Thread); 229 Interlocked.Decrement(ref pBusyThreads);
266 } 230 }
267 Thread.WorkItem.Thread = null; 231 Thread.WorkItem.Thread = null;
268 Thread.WorkItem = null; 232 Thread.WorkItem = null;
269 Thread.Busy = false; 233 Thread.Busy = false;
270 lock (pIdleThreads) { 234 lock (pIdleThreads) {