Mercurial > pub > ImplabNet
annotate Implab/Parallels/DispatchPool.cs @ 89:ce0171cacec4 v2
improved performance of a chained map operation
| author | cin |
|---|---|
| date | Wed, 08 Oct 2014 02:19:45 +0400 |
| parents | 2c5631b43c7d |
| children | 4c0e5ef99986 |
| rev | line source |
|---|---|
| 15 | 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 abstract class DispatchPool<TUnit> : IDisposable { | |
| 81 | 10 readonly int m_minThreadsLimit; |
| 11 readonly int m_maxThreadsLimit; | |
| 12 readonly int m_releaseTimeout = 1000; // the timeout while the working thread will wait for the new tasks before exit | |
|
21
6a56df4ec59e
DispatchPool works again, but performance is poor in some cases
cin
parents:
20
diff
changeset
|
13 |
| 81 | 14 int m_threads = 0; // the current size of the pool |
|
21
6a56df4ec59e
DispatchPool works again, but performance is poor in some cases
cin
parents:
20
diff
changeset
|
15 int m_maxRunningThreads = 0; // the meximum reached size of the pool |
| 81 | 16 int m_exit = 0; // the pool is going to shutdown, all unused workers are released |
| 80 | 17 |
| 81 | 18 readonly object m_signal = new object(); // used to pulse waiting threads |
| 15 | 19 |
| 20 protected DispatchPool(int min, int max) { | |
| 21 if (min < 0) | |
| 22 throw new ArgumentOutOfRangeException("min"); | |
| 23 if (max <= 0) | |
| 24 throw new ArgumentOutOfRangeException("max"); | |
| 25 | |
| 26 if (min > max) | |
| 27 min = max; | |
| 81 | 28 m_minThreadsLimit = min; |
| 29 m_maxThreadsLimit = max; | |
| 15 | 30 } |
| 31 | |
| 32 protected DispatchPool(int threads) | |
| 33 : this(threads, threads) { | |
| 34 } | |
| 35 | |
| 36 protected DispatchPool() { | |
| 37 int maxThreads, maxCP; | |
| 38 ThreadPool.GetMaxThreads(out maxThreads, out maxCP); | |
| 39 | |
| 81 | 40 m_minThreadsLimit = 0; |
| 41 m_maxThreadsLimit = maxThreads; | |
| 15 | 42 } |
| 43 | |
| 44 protected void InitPool() { | |
| 81 | 45 for (int i = 0; i < m_minThreadsLimit; i++) |
| 15 | 46 StartWorker(); |
| 47 } | |
| 48 | |
| 20 | 49 public int PoolSize { |
| 15 | 50 get { |
| 80 | 51 Thread.MemoryBarrier(); |
| 81 | 52 return m_threads; |
| 20 | 53 } |
| 54 } | |
| 81 | 55 |
| 15 | 56 public int MaxRunningThreads { |
| 57 get { | |
| 80 | 58 Thread.MemoryBarrier(); |
| 15 | 59 return m_maxRunningThreads; |
| 60 } | |
| 61 } | |
| 62 | |
| 63 protected bool IsDisposed { | |
| 64 get { | |
| 80 | 65 Thread.MemoryBarrier(); |
| 81 | 66 return m_exit == 1; |
| 15 | 67 } |
| 68 } | |
| 69 | |
| 17 | 70 protected abstract bool TryDequeue(out TUnit unit); |
| 71 | |
| 89 | 72 bool Dequeue(out TUnit unit, int timeout) { |
| 81 | 73 int ts = Environment.TickCount; |
| 74 if (TryDequeue(out unit)) | |
| 75 return true; | |
| 76 lock (m_signal) { | |
| 77 while (!TryDequeue(out unit) && m_exit == 0) | |
| 78 if(!Monitor.Wait(m_signal, Math.Max(0, ts + timeout - Environment.TickCount))) { | |
| 79 // timeout | |
| 80 return false; | |
| 80 | 81 } |
| 81 | 82 // queue item or terminate |
| 83 Monitor.Pulse(m_signal); | |
| 84 if (m_exit == 1) | |
| 85 return false; | |
| 80 | 86 } |
| 81 | 87 return true; |
| 22 | 88 } |
| 89 | |
| 81 | 90 protected void SignalThread() { |
| 91 lock (m_signal) { | |
| 92 Monitor.Pulse(m_signal); | |
| 30 | 93 } |
| 94 } | |
| 95 | |
| 17 | 96 #region thread slots traits |
| 97 | |
| 98 bool AllocateThreadSlot() { | |
| 16 | 99 int current; |
| 15 | 100 // use spins to allocate slot for the new thread |
| 101 do { | |
| 81 | 102 current = m_threads; |
| 103 if (current >= m_maxThreadsLimit || m_exit == 1) | |
| 15 | 104 // no more slots left or the pool has been disposed |
| 105 return false; | |
| 81 | 106 } while (current != Interlocked.CompareExchange(ref m_threads, current + 1, current)); |
| 15 | 107 |
| 17 | 108 UpdateMaxThreads(current + 1); |
| 109 | |
| 110 return true; | |
| 111 } | |
| 15 | 112 |
| 17 | 113 bool AllocateThreadSlot(int desired) { |
| 81 | 114 if (desired - 1 != Interlocked.CompareExchange(ref m_threads, desired, desired - 1)) |
| 17 | 115 return false; |
| 116 | |
| 117 UpdateMaxThreads(desired); | |
| 15 | 118 |
| 17 | 119 return true; |
| 120 } | |
| 121 | |
| 122 bool ReleaseThreadSlot(out bool last) { | |
| 123 last = false; | |
| 124 int current; | |
| 125 // use spins to release slot for the new thread | |
| 80 | 126 Thread.MemoryBarrier(); |
| 17 | 127 do { |
| 81 | 128 current = m_threads; |
| 129 if (current <= m_minThreadsLimit && m_exit == 0) | |
| 17 | 130 // the thread is reserved |
| 131 return false; | |
| 81 | 132 } while (current != Interlocked.CompareExchange(ref m_threads, current - 1, current)); |
| 17 | 133 |
| 134 last = (current == 1); | |
| 15 | 135 |
| 136 return true; | |
| 137 } | |
| 138 | |
| 17 | 139 void UpdateMaxThreads(int count) { |
| 140 int max; | |
| 16 | 141 do { |
| 17 | 142 max = m_maxRunningThreads; |
| 143 if (max >= count) | |
| 144 break; | |
| 145 } while(max != Interlocked.CompareExchange(ref m_maxRunningThreads, count, max)); | |
| 16 | 146 } |
| 147 | |
| 17 | 148 #endregion |
| 149 | |
| 81 | 150 protected bool StartWorker() { |
| 17 | 151 if (AllocateThreadSlot()) { |
| 152 // slot successfully allocated | |
| 153 var worker = new Thread(this.Worker); | |
| 154 worker.IsBackground = true; | |
| 155 worker.Start(); | |
| 156 | |
| 157 return true; | |
| 158 } else { | |
| 159 return false; | |
| 160 } | |
| 16 | 161 } |
| 162 | |
|
21
6a56df4ec59e
DispatchPool works again, but performance is poor in some cases
cin
parents:
20
diff
changeset
|
163 protected abstract void InvokeUnit(TUnit unit); |
|
6a56df4ec59e
DispatchPool works again, but performance is poor in some cases
cin
parents:
20
diff
changeset
|
164 |
| 41 | 165 protected virtual void Worker() { |
|
21
6a56df4ec59e
DispatchPool works again, but performance is poor in some cases
cin
parents:
20
diff
changeset
|
166 TUnit unit; |
| 81 | 167 bool last; |
| 15 | 168 do { |
| 81 | 169 while (Dequeue(out unit, m_releaseTimeout)) { |
|
21
6a56df4ec59e
DispatchPool works again, but performance is poor in some cases
cin
parents:
20
diff
changeset
|
170 InvokeUnit(unit); |
| 81 | 171 } |
| 172 if(!ReleaseThreadSlot(out last)) | |
|
21
6a56df4ec59e
DispatchPool works again, but performance is poor in some cases
cin
parents:
20
diff
changeset
|
173 continue; |
| 81 | 174 // queue may be not empty |
| 175 if (last && TryDequeue(out unit)) { | |
| 176 InvokeUnit(unit); | |
| 177 if (AllocateThreadSlot(1)) | |
| 178 continue; | |
| 179 // we can safely exit since pool is alive | |
| 15 | 180 } |
| 81 | 181 break; |
| 182 } while(true); | |
| 183 } | |
| 15 | 184 |
| 185 | |
| 186 protected virtual void Dispose(bool disposing) { | |
| 187 if (disposing) { | |
| 81 | 188 if (0 == Interlocked.CompareExchange(ref m_exit, 1, 0)) { // implies memory barrier |
| 15 | 189 // wake sleeping threads |
| 81 | 190 SignalThread(); |
| 15 | 191 GC.SuppressFinalize(this); |
| 192 } | |
| 193 } | |
| 194 } | |
| 195 | |
| 196 public void Dispose() { | |
| 197 Dispose(true); | |
| 198 } | |
| 199 | |
| 200 ~DispatchPool() { | |
| 201 Dispose(false); | |
| 202 } | |
| 203 } | |
| 204 } |