|
12
|
1 using System;
|
|
|
2 using System.Collections.Generic;
|
|
|
3 using System.Linq;
|
|
|
4 using System.Text;
|
|
|
5 using System.Threading;
|
|
|
6 using System.Diagnostics;
|
|
|
7
|
|
|
8 namespace Implab.Parallels {
|
|
|
9 public class WorkerPool : IDisposable {
|
|
|
10 readonly int m_minThreads;
|
|
|
11 readonly int m_maxThreads;
|
|
|
12 int m_runningThreads;
|
|
|
13 object m_lock = new object();
|
|
|
14
|
|
|
15 bool m_disposed = false;
|
|
13
|
16
|
|
|
17 // this event will signal that workers can try to fetch a task from queue or the pool has been disposed
|
|
12
|
18 ManualResetEvent m_hasTasks = new ManualResetEvent(false);
|
|
|
19 Queue<Action> m_queue = new Queue<Action>();
|
|
|
20
|
|
|
21 public WorkerPool(int min, int max) {
|
|
|
22 if (min < 0)
|
|
|
23 throw new ArgumentOutOfRangeException("min");
|
|
13
|
24 if (max <= 0)
|
|
|
25 throw new ArgumentOutOfRangeException("max");
|
|
|
26
|
|
12
|
27 if (min > max)
|
|
|
28 min = max;
|
|
|
29 m_minThreads = min;
|
|
|
30 m_maxThreads = max;
|
|
|
31
|
|
13
|
32 InitPool();
|
|
|
33 }
|
|
|
34
|
|
|
35 public WorkerPool(int max)
|
|
|
36 : this(0, max) {
|
|
|
37 }
|
|
|
38
|
|
|
39 public WorkerPool() {
|
|
|
40 int maxThreads, maxCP;
|
|
|
41 ThreadPool.GetMaxThreads(out maxThreads, out maxCP);
|
|
|
42
|
|
|
43 m_minThreads = 0;
|
|
|
44 m_maxThreads = maxThreads;
|
|
|
45
|
|
|
46 InitPool();
|
|
|
47 }
|
|
|
48
|
|
|
49 void InitPool() {
|
|
12
|
50 for (int i = 0; i < m_minThreads; i++)
|
|
|
51 StartWorker();
|
|
|
52 }
|
|
|
53
|
|
13
|
54 public int ThreadCount {
|
|
|
55 get {
|
|
|
56 return m_runningThreads;
|
|
|
57 }
|
|
|
58 }
|
|
|
59
|
|
12
|
60 public Promise<T> Invoke<T>(Func<T> task) {
|
|
|
61 if (m_disposed)
|
|
|
62 throw new ObjectDisposedException(ToString());
|
|
|
63 if (task == null)
|
|
|
64 throw new ArgumentNullException("task");
|
|
|
65
|
|
|
66 var promise = new Promise<T>();
|
|
|
67
|
|
13
|
68 var queueLen = EnqueueTask(delegate() {
|
|
|
69 try {
|
|
|
70 promise.Resolve(task());
|
|
|
71 } catch (Exception e) {
|
|
|
72 promise.Reject(e);
|
|
|
73 }
|
|
|
74 });
|
|
12
|
75
|
|
13
|
76 if (queueLen > 1)
|
|
|
77 StartWorker();
|
|
12
|
78
|
|
|
79 return promise;
|
|
|
80 }
|
|
|
81
|
|
|
82 bool StartWorker() {
|
|
|
83 var current = m_runningThreads;
|
|
|
84 // use spins to allocate slot for the new thread
|
|
|
85 do {
|
|
|
86 if (current >= m_maxThreads)
|
|
|
87 // no more slots left
|
|
|
88 return false;
|
|
|
89 } while (current != Interlocked.CompareExchange(ref m_runningThreads, current + 1, current));
|
|
|
90
|
|
|
91 // slot successfully allocated
|
|
|
92
|
|
|
93 var worker = new Thread(this.Worker);
|
|
13
|
94 worker.IsBackground = true;
|
|
12
|
95 worker.Start();
|
|
|
96
|
|
|
97 return true;
|
|
|
98 }
|
|
|
99
|
|
13
|
100 int EnqueueTask(Action task) {
|
|
12
|
101 Debug.Assert(task != null);
|
|
|
102 lock (m_queue) {
|
|
|
103 m_queue.Enqueue(task);
|
|
|
104 m_hasTasks.Set();
|
|
13
|
105 return m_queue.Count;
|
|
12
|
106 }
|
|
|
107 }
|
|
|
108
|
|
|
109 bool FetchTask(out Action task) {
|
|
|
110 task = null;
|
|
|
111
|
|
|
112 while (true) {
|
|
|
113
|
|
|
114 m_hasTasks.WaitOne();
|
|
|
115
|
|
|
116 if (m_disposed)
|
|
|
117 return false;
|
|
|
118
|
|
|
119 lock (m_queue) {
|
|
|
120 if (m_queue.Count > 0) {
|
|
|
121 task = m_queue.Dequeue();
|
|
|
122 return true;
|
|
|
123 }
|
|
|
124
|
|
|
125 // no tasks left
|
|
13
|
126 // signal that no more tasks left, current lock ensures that this event won't suppress newly added task
|
|
12
|
127 m_hasTasks.Reset();
|
|
|
128 }
|
|
|
129
|
|
|
130 bool exit = true;
|
|
|
131
|
|
|
132 var current = m_runningThreads;
|
|
|
133 do {
|
|
|
134 if (current <= m_minThreads) {
|
|
|
135 exit = false; // this thread should return and wait for the new events
|
|
|
136 break;
|
|
|
137 }
|
|
|
138 } while (current != Interlocked.CompareExchange(ref m_runningThreads, current - 1, current));
|
|
|
139
|
|
|
140 if (exit)
|
|
|
141 return false;
|
|
|
142 }
|
|
|
143 }
|
|
|
144
|
|
|
145 void Worker() {
|
|
|
146 Action task;
|
|
|
147 while (FetchTask(out task))
|
|
|
148 task();
|
|
|
149 }
|
|
|
150
|
|
|
151 protected virtual void Dispose(bool disposing) {
|
|
|
152 if (disposing) {
|
|
|
153 lock (m_lock) {
|
|
|
154 if (m_disposed)
|
|
|
155 return;
|
|
|
156 m_disposed = true;
|
|
|
157 }
|
|
|
158 m_hasTasks.Set();
|
|
|
159 GC.SuppressFinalize(this);
|
|
|
160 }
|
|
|
161 }
|
|
|
162
|
|
|
163 public void Dispose() {
|
|
|
164 Dispose(true);
|
|
|
165 }
|
|
|
166
|
|
|
167 ~WorkerPool() {
|
|
|
168 Dispose(false);
|
|
|
169 }
|
|
|
170 }
|
|
|
171 }
|
