Mercurial > hg > ucis.core

comparison ThreadPool.cs @ 0:3ab940a0c7a0

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
Initial commit
author Ivo Smits <Ivo@UCIS.nl>
date Tue, 11 Sep 2012 16:28:53 +0200
parents
children 4b78cc5f116b
comparison
equal deleted inserted replaced
-1:000000000000 0:3ab940a0c7a0
1 using System;
2 using System.Threading;
3 using System.Collections.Generic;
4
5 namespace UCIS {
6 public class ThreadPool {
7 private static ThreadPool pManager = null;
8
9 public static ThreadPool DefaultPool {
10 get {
11 if (pManager == null) pManager = new ThreadPool();
12 return pManager;
13 }
14 }
15
16 //Starts a long-term background task
17 public static WorkItem RunTask(WaitCallback Callback, object State) {
18 return DefaultPool.QueueWorkItem(Callback, State);
19 }
20 //Starts a short-term background task
21 public static WorkItem RunCall(WaitCallback Callback, object State) {
22 return DefaultPool.QueueWorkItem(Callback, State);
23 }
24
25
26 public class WorkItem {
27 public WaitCallback Callback { get; internal set; }
28 public object State { get; internal set; }
29 public ThreadInfo Thread { get; internal set; }
30 }
31 public class ThreadInfo {
32 public Thread Thread { get; internal set; }
33 internal AutoResetEvent WaitHandle = new AutoResetEvent(false);
34 public WorkItem WorkItem { get; internal set; }
35 public bool Busy { get; internal set; }
36 public bool Abort { get; internal set; }
37 public DateTime LastActive { get; internal set; }
38 }
39
40 public class ExceptionEventArgs : EventArgs {
41 public ExceptionEventArgs(WorkItem Item, Exception Exception, bool ThrowError) {
42 this.Item = Item;
43 this.Exception = Exception;
44 this.ThrowError = ThrowError;
45 }
46 public WorkItem Item;
47 public Exception Exception;
48 public bool ThrowError;
49 }
50
51 private List<ThreadInfo> pThreads = new List<ThreadInfo>();
52 private List<ThreadInfo> pBusyThreads = new List<ThreadInfo>();
53 private Queue<ThreadInfo> pIdleThreads = new Queue<ThreadInfo>();
54 private int pThreadsMax;
55 private int pThreadsMinIdle;
56 private int pThreadsMaxIdle;
57
58 public event OnExceptionEventHandler OnException;
59 public delegate void OnExceptionEventHandler(ThreadPool sender, ExceptionEventArgs e);
60
61 public System.Collections.ObjectModel.ReadOnlyCollection<ThreadInfo> Threads {
62 get {
63 return new System.Collections.ObjectModel.ReadOnlyCollection<ThreadInfo>(pThreads);
64 }
65 }
66
67 public ThreadPool() : this(250, 1, 5) { }
68
69 public ThreadPool(int MaxThreads, int MinIdle, int MaxIdle) {
70 int I = 0;
71 if (MaxThreads < 0) {
72 throw new ArgumentOutOfRangeException("ThreadsMaxIdle", "ThreadsMaxIdle must greater than 0");
73 } else if (MaxThreads < MaxIdle) {
74 throw new ArgumentOutOfRangeException("ThreadsMax", "ThreadsMax must be greater than or equal to ThreadsMaxIdle");
75 } else if (MaxIdle < 0) {
76 throw new ArgumentOutOfRangeException("ThreadsMaxIdle", "ThreadsMaxIdle must greater than or equal to 0");
77 } else if (MinIdle < 0) {
78 throw new ArgumentOutOfRangeException("ThreadsMinIdle", "ThreadsMinIdle must greater than or equal to 0");
79 } else if (MinIdle > MaxIdle) {
80 throw new ArgumentOutOfRangeException("ThreadsMaxIdle", "ThreadsMaxIdle must be greater than or equal to ThreadsMinIdle");
81 }
82 pThreadsMax = MaxThreads;
83 pThreadsMinIdle = MinIdle;
84 pThreadsMaxIdle = MaxIdle;
85 for (I = 1; I <= pThreadsMinIdle; I++) {
86 StartThread(false);
87 }
88 }
89
90 public int ThreadsIdle {
91 get { return pIdleThreads.Count; }
92 }
93 public int ThreadsBusy {
94 get { return pBusyThreads.Count; }
95 }
96 public int ThreadsAlive {
97 get { return pThreads.Count; }
98 }
99 public int ThreadsMinIdle {
100 get { return pThreadsMinIdle; }
101 set {
102 if (value > pThreadsMaxIdle) {
103 throw new ArgumentOutOfRangeException("ThreadsMinIdle", "ThreadsMinIdle must be smaller than ThreadsMaxIdle");
104 } else if (value < 0) {
105 throw new ArgumentOutOfRangeException("ThreadsMinIdle", "ThreadsMinIdle must greater than or equal to 0");
106 } else {
107 int I = 0;
108 int C = 0;
109 C = pIdleThreads.Count;
110 if (value > C) {
111 for (I = C; I <= value - 1; I++) {
112 StartThread(false);
113 }
114 }
115 pThreadsMinIdle = value;
116 }
117 }
118 }
119 public int ThreadsMaxIdle {
120 get { return pThreadsMaxIdle; }
121 set {
122 if (pThreadsMinIdle > value) {
123 throw new ArgumentOutOfRangeException("ThreadsMaxIdle", "ThreadsMaxIdle must be greater than or equal to ThreadsMinIdle");
124 } else if (value < 0) {
125 throw new ArgumentOutOfRangeException("ThreadsMaxIdle", "ThreadsMaxIdle must greater than or equal to 0");
126 } else {
127 int I = 0;
128 int C = 0;
129 ThreadInfo T = default(ThreadInfo);
130 lock (pIdleThreads) {
131 C = pIdleThreads.Count;
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 }
143 }
144 public int ThreadsMax {
145 get { return pThreadsMax; }
146 set {
147 if (pThreadsMaxIdle > value) {
148 throw new ArgumentOutOfRangeException("ThreadsMax", "ThreadsMax must be greater than or equal to ThreadsMaxIdle");
149 } else if (value <= 0) {
150 throw new ArgumentOutOfRangeException("ThreadsMax", "ThreadsMax must greater than 0");
151 } else {
152 pThreadsMax = value;
153 }
154 }
155 }
156
157 public WorkItem QueueWorkItem(WaitCallback Callback, object State) {
158 WorkItem WorkItem = new WorkItem();
159 ThreadInfo Thread = null;
160 WorkItem.Callback = Callback;
161 WorkItem.State = State;
162 lock (pIdleThreads) {
163 while (pIdleThreads.Count > 0) {
164 Thread = pIdleThreads.Dequeue();
165 if (!Thread.Abort) {
166 break;
167 } else {
168 Thread = null;
169 }
170 }
171 }
172 if (Thread == null) {
173 if (pThreads.Count < pThreadsMax) {
174 Thread = StartThread(true);
175 } else {
176 throw new ThreadStateException("Thread limit exceeded");
177 }
178 }
179 Thread.LastActive = DateTime.Now;
180 WorkItem.Thread = Thread;
181 Thread.WorkItem = WorkItem;
182 Thread.WaitHandle.Set();
183 return WorkItem;
184 }
185
186 private ThreadInfo StartThread(bool Reserved) {
187 ThreadInfo Thread = new ThreadInfo();
188 Thread.Thread = new Thread(pWorker);
189 lock (pThreads) {
190 pThreads.Add(Thread);
191 if (!Reserved) pIdleThreads.Enqueue(Thread);
192 }
193 Thread.LastActive = DateTime.Now;
194 Thread.Thread.Start(Thread);
195 return Thread;
196 }
197
198 public void AbortAllThreads() {
199 lock (pIdleThreads) {
200 while (pIdleThreads.Count > 0) {
201 ThreadInfo Thread = pIdleThreads.Dequeue();
202 Thread.Abort = true;
203 Thread.WaitHandle.Set();
204 }
205 }
206 foreach (ThreadInfo Thread in pThreads.ToArray()) {
207 if (Thread != null && !Thread.Abort) {
208 Thread.Thread.Abort();
209 Thread.Abort = true;
210 Thread.WaitHandle.Set();
211 }
212 }
213 pIdleThreads.Clear();
214 }
215
216 //ToDo: add timer to kill old threads periodically
217 public void KillOldThreads() {
218 ThreadInfo Thread;
219 lock (pIdleThreads) {
220 if (pIdleThreads.Count == 0) return;
221 Thread = pIdleThreads.Dequeue();
222 }
223 if (DateTime.Now.Subtract(Thread.LastActive).TotalMinutes > 1) {
224 Thread.Abort = true;
225 Thread.WaitHandle.Set();
226 } else {
227 lock (pIdleThreads) pIdleThreads.Enqueue(Thread);
228 }
229 }
230
231 private void pWorker(object state) {
232 ThreadInfo Thread = (ThreadInfo)state;
233 if (Thread == null) throw new ArgumentNullException("state");
234 try {
235 while (true) {
236 if (Thread.WaitHandle == null) throw new ArgumentNullException("WaitHandle");
237 if (!Thread.WaitHandle.WaitOne(1000, false)) {
238 if (pBusyThreads.Count == 0) {
239 return;
240 } else {
241 continue;
242 }
243 }
244 if (Thread.Abort) break;
245
246 Thread.Busy = true;
247 lock (pBusyThreads) pBusyThreads.Add(Thread);
248 try {
249 if (Thread.WorkItem == null) throw new ArgumentNullException("WorkItem");
250 if (Thread.WorkItem.Callback == null) throw new ArgumentNullException("WorkItem.Callback");
251 Thread.WorkItem.Callback.Invoke(Thread.WorkItem.State);
252 } catch (ThreadAbortException ex) {
253 ExceptionEventArgs e = new ExceptionEventArgs(Thread.WorkItem, ex, false);
254 if (OnException != null) OnException(this, e);
255 if (e.ThrowError) Console.WriteLine("ThreadAbortException in ThreadPool thread: " + e.Exception.ToString());
256 return;
257 } catch (Exception ex) {
258 ExceptionEventArgs e = new ExceptionEventArgs(Thread.WorkItem, ex, true);
259 if (OnException != null) OnException(this, e);
260 if (e.ThrowError) {
261 Console.WriteLine("Exception in ThreadPool thread: " + e.Exception.ToString());
262 throw new Exception("Unhandled exception in work item", e.Exception);
263 }
264 } finally {
265 lock (pBusyThreads) pBusyThreads.Remove(Thread);
266 }
267 Thread.WorkItem.Thread = null;
268 Thread.WorkItem = null;
269 Thread.Busy = false;
270 lock (pIdleThreads) {
271 if (pIdleThreads.Count >= pThreadsMaxIdle) break;
272 pIdleThreads.Enqueue(Thread);
273 }
274 }
275 } finally {
276 Thread.Abort = true;
277 lock (pThreads) pThreads.Remove(Thread);
278 }
279 }
280 }
281 }