##// END OF EJS Templates
implab » ImplabNet
RSS
added memory barriers
cin -
r80:4f20870d0816 v2
parent child
Show More
Show file before | Show file after | Hide comments
@@ -0,0 +1,85
1 using System;
2 using Implab.Parallels;
3 using System.Threading;
4
5 namespace Implab {
6 /*public class SyncPool<T> : IDisposable {
7 readonly Func<T> m_factory;
8 readonly Action<T> m_cleanup;
9 readonly int m_size;
10 readonly MTQueue<T> m_queue = new MTQueue<T>();
11
12 volatile bool m_disposed;
13
14 volatile int m_count;
15
16 public SyncPool(Func<T> factory, Action<T> cleanup, int size) {
17 Safe.ArgumentNotNull(factory, "factory");
18 Safe.ArgumentInRange(size, 1, size, "size");
19
20 m_factory = factory;
21 m_cleanup = cleanup;
22 m_size = size;
23 }
24
25 public SyncPool(Func<T> factory, Action<T> cleanup) : this(factory,cleanup,Environment.ProcessorCount+1) {
26 }
27
28 public SyncPool(Func<T> factory) : this(factory,null,Environment.ProcessorCount+1) {
29 }
30
31 public SyncPoolWrapper<T> Allocate() {
32 if (m_disposed)
33 throw new ObjectDisposedException(this.ToString());
34
35 T instance;
36 if (m_queue.TryDequeue(out instance)) {
37 Interlocked.Decrement(ref m_count);
38 return instance;
39 } else {
40 instance = m_factory();
41 }
42 return new SyncPoolWrapper<T>(instance, this);
43 }
44
45 public void Release(T instance) {
46 if (m_count < m_size && !m_disposed) {
47 Interlocked.Increment(ref m_count);
48
49 if (m_cleanup != null)
50 m_cleanup(instance);
51
52 m_queue.Enqueue(instance);
53
54 // пока элемент возвращался в кеш, была начата операция освобождения всего кеша
55 // и возможно уже законцена, в таком случае следует извлечь элемент обратно и
56 // освободить его. Если операция освобождения кеша еще не заврешилась, то будет
57 // изъят и освобожден произвольный элемен, что не повлияет на ход всего процесса.
58 if (m_disposed && m_queue.TryDequeue(out instance))
59 Safe.Dispose(instance);
60
61 } else {
62 Safe.Dispose(instance);
63 }
64 }
65
66 protected virtual void Dispose(bool disposing) {
67 if (disposing) {
68 m_disposed = true;
69 T instance;
70 while (m_queue.TryDequeue(out instance))
71 Safe.Dispose(instance);
72 }
73 }
74
75 #region IDisposable implementation
76
77 public void Dispose() {
78 Dispose(true);
79 GC.SuppressFinalize(this);
80 }
81
82 #endregion
83 }*/
84 }
85
Show file before | Show file after | Hide comments
@@ -0,0 +1,24
1 using System;
2
3 namespace Implab {
4 /*public struct SyncPoolWrapper<T> : IDisposable {
5 readonly T m_value;
6 readonly SyncPool<T> m_pool;
7
8 internal SyncPoolWrapper(T value, SyncPool<T> pool) {
9 m_value = value;
10 m_pool = pool;
11 }
12
13 public T Value {
14 get { return m_value; }
15 }
16
17 #region IDisposable implementation
18 public void Dispose() {
19 m_pool.Release(m_value);
20 }
21 #endregion
22 }*/
23 }
24
Show file before | Show file after | Hide comments
@@ -291,7 +291,7 namespace Implab.Test {
291 [TestMethod]
291 [TestMethod]
292 public void ChainedMapTest() {
292 public void ChainedMapTest() {
293
293
294 using (var pool = new WorkerPool(0,100,100)) {
294 using (var pool = new WorkerPool(0,10,100)) {
295 const int count = 10000;
295 const int count = 10000;
296
296
297 var args = new double[count];
297 var args = new double[count];
Show file before | Show file after | Hide comments
@@ -16,8 +16,6 Project("{2150E333-8FDC-42A3-9474-1A3956
16 EndProject
16 EndProject
17 Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Implab.Test.mono", "Implab.Test\Implab.Test.mono.csproj", "{2BD05F84-E067-4B87-9477-FDC2676A21C6}"
17 Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Implab.Test.mono", "Implab.Test\Implab.Test.mono.csproj", "{2BD05F84-E067-4B87-9477-FDC2676A21C6}"
18 EndProject
18 EndProject
19 Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Implab.Fx.Test.mono", "Implab.Fx.Test\Implab.Fx.Test.mono.csproj", "{2BD05F84-E067-4B87-9477-FDC2676A21C6}"
20 EndProject
21 Global
19 Global
22 GlobalSection(SolutionConfigurationPlatforms) = preSolution
20 GlobalSection(SolutionConfigurationPlatforms) = preSolution
23 Debug|Any CPU = Debug|Any CPU
21 Debug|Any CPU = Debug|Any CPU
@@ -77,7 +75,6 Global
77 EndGlobalSection
75 EndGlobalSection
78 GlobalSection(NestedProjects) = preSolution
76 GlobalSection(NestedProjects) = preSolution
79 {2BD05F84-E067-4B87-9477-FDC2676A21C6} = {BCA337C3-BFDC-4825-BBDB-E6D467E4E452}
77 {2BD05F84-E067-4B87-9477-FDC2676A21C6} = {BCA337C3-BFDC-4825-BBDB-E6D467E4E452}
80 {2BD05F84-E067-4B87-9477-FDC2676A21C6} = {BCA337C3-BFDC-4825-BBDB-E6D467E4E452}
81 EndGlobalSection
78 EndGlobalSection
82 GlobalSection(MonoDevelopProperties) = preSolution
79 GlobalSection(MonoDevelopProperties) = preSolution
83 StartupItem = Implab\Implab.csproj
80 StartupItem = Implab\Implab.csproj
Show file before | Show file after | Hide comments
@@ -7,6 +7,8
7 <OutputType>Library</OutputType>
7 <OutputType>Library</OutputType>
8 <RootNamespace>Implab</RootNamespace>
8 <RootNamespace>Implab</RootNamespace>
9 <AssemblyName>Implab</AssemblyName>
9 <AssemblyName>Implab</AssemblyName>
10 <ProductVersion>8.0.30703</ProductVersion>
11 <SchemaVersion>2.0</SchemaVersion>
10 </PropertyGroup>
12 </PropertyGroup>
11 <PropertyGroup Condition=" '$(Configuration)|$(Platform)' == 'Debug|AnyCPU' ">
13 <PropertyGroup Condition=" '$(Configuration)|$(Platform)' == 'Debug|AnyCPU' ">
12 <DebugSymbols>true</DebugSymbols>
14 <DebugSymbols>true</DebugSymbols>
@@ -121,6 +123,8
121 <Compile Include="PromiseExtensions.cs" />
123 <Compile Include="PromiseExtensions.cs" />
122 <Compile Include="TransientPromiseException.cs" />
124 <Compile Include="TransientPromiseException.cs" />
123 <Compile Include="SyncContextPromise.cs" />
125 <Compile Include="SyncContextPromise.cs" />
126 <Compile Include="SyncPool.cs" />
127 <Compile Include="SyncPoolWrapper.cs" />
124 </ItemGroup>
128 </ItemGroup>
125 <Import Project="$(MSBuildBinPath)\Microsoft.CSharp.targets" />
129 <Import Project="$(MSBuildBinPath)\Microsoft.CSharp.targets" />
126 <ItemGroup />
130 <ItemGroup />
Show file before | Show file after | Hide comments
@@ -19,7 +19,9 namespace Implab.JSON {
19 }
19 }
20
20
21 /// <summary>
21 /// <summary>
22 /// Интерпретировать массивы как множественные элементы (убирает один уровень вложенности)
22 /// Интерпретировать массивы как множественные элементы (убирает один уровень вложенности), иначе массив
23 /// представляется в виде узла, дочерними элементами которого являются элементы массива, имена дочерних элементов
24 /// определяются свойством <see cref="ArrayItemName"/>. По умолчанию <c>false</c>.
23 /// </summary>
25 /// </summary>
24 public bool FlattenArrays {
26 public bool FlattenArrays {
25 get;
27 get;
@@ -44,6 +46,7 namespace Implab.JSON {
44
46
45 /// <summary>
47 /// <summary>
46 /// Имя элемента для массивов, если не включена опция <see cref="FlattenArrays"/>.
48 /// Имя элемента для массивов, если не включена опция <see cref="FlattenArrays"/>.
49 /// По умолчанию <c>item</c>.
47 /// </summary>
50 /// </summary>
48 public string ArrayItemName {
51 public string ArrayItemName {
49 get;
52 get;
Show file before | Show file after | Hide comments
@@ -153,7 +153,8 namespace Implab.Parallels {
153 var res = new TDst[source.Length];
153 var res = new TDst[source.Length];
154 var pending = source.Length;
154 var pending = source.Length;
155
155
156 var semaphore = new Semaphore(threads, threads);
156 object locker = new object();
157 int slots = threads;
157
158
158 // Analysis disable AccessToDisposedClosure
159 // Analysis disable AccessToDisposedClosure
159 AsyncPool.InvokeNewThread(() => {
160 AsyncPool.InvokeNewThread(() => {
@@ -162,22 +163,28 namespace Implab.Parallels {
162 break; // stop processing in case of error or cancellation
163 break; // stop processing in case of error or cancellation
163 var idx = i;
164 var idx = i;
164
165
165 semaphore.WaitOne();
166 lock(locker) {
167 while(slots == 0)
168 Monitor.Wait(locker);
169 slots--;
170 }
166 try {
171 try {
167 var p1 = transform(source[i]);
172 transform(source[i])
168 p1.Anyway(() => semaphore.Release());
173 .Anyway(() => {
169 p1.Then(
174 lock(locker) {
170 x => {
175 slots ++;
171 res[idx] = x;
176 Monitor.Pulse(locker);
172 var left = Interlocked.Decrement(ref pending);
177 }
173 if (left == 0)
178 })
174 promise.Resolve(res);
179 .Last(
175 },
180 x => {
176 e => {
181 res[idx] = x;
177 promise.Reject(e);
182 var left = Interlocked.Decrement(ref pending);
178 throw new TransientPromiseException(e);
183 if (left == 0)
179 }
184 promise.Resolve(res);
180 );
185 },
186 e => promise.Reject(e)
187 );
181
188
182 } catch (Exception e) {
189 } catch (Exception e) {
183 promise.Reject(e);
190 promise.Reject(e);
@@ -186,7 +193,7 namespace Implab.Parallels {
186 return 0;
193 return 0;
187 });
194 });
188
195
189 return promise.Anyway(semaphore.Dispose);
196 return promise;
190 }
197 }
191
198
192 }
199 }
Show file before | Show file after | Hide comments
@@ -9,16 +9,17 namespace Implab.Parallels {
9 public abstract class DispatchPool<TUnit> : IDisposable {
9 public abstract class DispatchPool<TUnit> : IDisposable {
10 readonly int m_minThreads;
10 readonly int m_minThreads;
11 readonly int m_maxThreads;
11 readonly int m_maxThreads;
12 readonly int m_releaseTimeout = 100; // the timeout while the working thread will wait for the new tasks before exit
12
13
13 int m_createdThreads = 0; // the current size of the pool
14 int m_createdThreads = 0; // the current size of the pool
14 int m_activeThreads = 0; // the count of threads which are active
15 int m_activeThreads = 0; // the count of threads which are active
15 int m_sleepingThreads = 0; // the count of currently inactive threads
16 int m_sleepingThreads = 0; // the count of currently inactive threads
16 int m_maxRunningThreads = 0; // the meximum reached size of the pool
17 int m_maxRunningThreads = 0; // the meximum reached size of the pool
17 int m_exitRequired = 0; // the pool is going to shutdown, all unused workers are released
18 int m_exitRequired = 0; // the pool is going to shutdown, all unused workers are released
18 int m_releaseTimeout = 100; // the timeout while the working thread will wait for the new tasks before exit
19
19 int m_wakeEvents = 0; // the count of wake events
20 int m_wakeEvents = 0; // the count of wake events
20
21
21 AutoResetEvent m_hasTasks = new AutoResetEvent(false);
22 readonly object m_signalLocker = new object();
22
23
23 protected DispatchPool(int min, int max) {
24 protected DispatchPool(int min, int max) {
24 if (min < 0)
25 if (min < 0)
@@ -51,68 +52,76 namespace Implab.Parallels {
51
52
52 public int PoolSize {
53 public int PoolSize {
53 get {
54 get {
55 Thread.MemoryBarrier();
54 return m_createdThreads;
56 return m_createdThreads;
55 }
57 }
56 }
58 }
57
59
58 public int ActiveThreads {
60 public int ActiveThreads {
59 get {
61 get {
62 Thread.MemoryBarrier();
60 return m_activeThreads;
63 return m_activeThreads;
61 }
64 }
62 }
65 }
63
66
64 public int MaxRunningThreads {
67 public int MaxRunningThreads {
65 get {
68 get {
69 Thread.MemoryBarrier();
66 return m_maxRunningThreads;
70 return m_maxRunningThreads;
67 }
71 }
68 }
72 }
69
73
70 protected bool IsDisposed {
74 protected bool IsDisposed {
71 get {
75 get {
72 return m_exitRequired != 0;
76 Thread.MemoryBarrier();
77 return m_exitRequired == 1;
73 }
78 }
74 }
79 }
75
80
76 protected abstract bool TryDequeue(out TUnit unit);
81 protected abstract bool TryDequeue(out TUnit unit);
77
82
78 #region thread execution traits
83 #region thread signaling traits
79 int SignalThread() {
84 int SignalThread() {
80 var signals = Interlocked.Increment(ref m_wakeEvents);
85 var signals = Interlocked.Increment(ref m_wakeEvents);
81 if(signals == 1)
86 if(signals == 1)
82 m_hasTasks.Set();
87 lock(m_signalLocker)
88 Monitor.Pulse(m_signalLocker);
83 return signals;
89 return signals;
84 }
90 }
85
91
86 bool FetchSignalOrWait(int timeout) {
92 bool FetchSignalOrWait(int timeout) {
87 var start = Environment.TickCount;
93 var start = Environment.TickCount;
88
94 int signals;
89 // означает, что поток владеет блокировкой и при успешном получении сигнала должен
95 Thread.MemoryBarrier(); // m_wakeEvents volatile first read
90 // ее вернуть, чтобы другой ожидающий поток смог
91 bool hasLock = false;
92 do {
96 do {
93 int signals;
97 signals = m_wakeEvents;
94 do {
98 if (signals == 0)
95 signals = m_wakeEvents;
99 break;
96 if (signals == 0)
100 } while (Interlocked.CompareExchange(ref m_wakeEvents, signals - 1, signals) != signals);
97 break;
98 } while (Interlocked.CompareExchange(ref m_wakeEvents, signals - 1, signals) != signals);
99
101
100 if (signals >= 1) {
102 if (signals == 0) {
101 if (signals > 1 && hasLock)
103 // no signal is fetched
102 m_hasTasks.Set();
104 lock(m_signalLocker) {
105 while(m_wakeEvents == 0) {
106 if (timeout != -1)
107 timeout = Math.Max(0, timeout - (Environment.TickCount - start));
108 if(!Monitor.Wait(m_signalLocker,timeout))
109 return false; // timeout
110 }
111 // m_wakeEvents > 0
112 if (Interlocked.Decrement(ref m_wakeEvents) > 0) //syncronized
113 Monitor.Pulse(m_signalLocker);
114
115 // signal fetched
103 return true;
116 return true;
104 }
117 }
105
118
106 if (timeout != -1)
119 } else {
107 timeout = Math.Max(0, timeout - (Environment.TickCount - start));
120 // signal fetched
121 return true;
122 }
108
123
109 // если сигналов больше не осталось, то первый поток, который дошел сюда сбросит событие
110 // и уйдет на пустой цикл, после чего заблокируется
111
124
112 hasLock = true;
113 } while (m_hasTasks.WaitOne(timeout));
114
115 return false;
116 }
125 }
117
126
118 bool Sleep(int timeout) {
127 bool Sleep(int timeout) {
@@ -131,7 +140,8 namespace Implab.Parallels {
131 /// Запускает либо новый поток, если раньше не было ни одного потока, либо устанавливает событие пробуждение одного спящего потока
140 /// Запускает либо новый поток, если раньше не было ни одного потока, либо устанавливает событие пробуждение одного спящего потока
132 /// </summary>
141 /// </summary>
133 protected void GrowPool() {
142 protected void GrowPool() {
134 if (m_exitRequired != 0)
143 Thread.MemoryBarrier();
144 if (m_exitRequired == 1)
135 return;
145 return;
136 if (m_sleepingThreads > m_wakeEvents) {
146 if (m_sleepingThreads > m_wakeEvents) {
137 //Console.WriteLine("Waking threads (sleeps {0}, pending {1})", m_sleepingThreads, m_wakeEvents);
147 //Console.WriteLine("Waking threads (sleeps {0}, pending {1})", m_sleepingThreads, m_wakeEvents);
@@ -204,7 +214,7 namespace Implab.Parallels {
204 // use spins to allocate slot for the new thread
214 // use spins to allocate slot for the new thread
205 do {
215 do {
206 current = m_createdThreads;
216 current = m_createdThreads;
207 if (current >= m_maxThreads || m_exitRequired != 0)
217 if (current >= m_maxThreads || m_exitRequired == 1)
208 // no more slots left or the pool has been disposed
218 // no more slots left or the pool has been disposed
209 return false;
219 return false;
210 } while (current != Interlocked.CompareExchange(ref m_createdThreads, current + 1, current));
220 } while (current != Interlocked.CompareExchange(ref m_createdThreads, current + 1, current));
@@ -227,6 +237,7 namespace Implab.Parallels {
227 last = false;
237 last = false;
228 int current;
238 int current;
229 // use spins to release slot for the new thread
239 // use spins to release slot for the new thread
240 Thread.MemoryBarrier();
230 do {
241 do {
231 current = m_createdThreads;
242 current = m_createdThreads;
232 if (current <= m_minThreads && m_exitRequired == 0)
243 if (current <= m_minThreads && m_exitRequired == 0)
@@ -264,6 +275,7 namespace Implab.Parallels {
264 // slot successfully allocated
275 // slot successfully allocated
265 var worker = new Thread(this.Worker);
276 var worker = new Thread(this.Worker);
266 worker.IsBackground = true;
277 worker.IsBackground = true;
278 Interlocked.Increment(ref m_activeThreads);
267 worker.Start();
279 worker.Start();
268
280
269 return true;
281 return true;
@@ -277,15 +289,14 namespace Implab.Parallels {
277 protected virtual void Worker() {
289 protected virtual void Worker() {
278 TUnit unit;
290 TUnit unit;
279 //Console.WriteLine("{0}: Active", Thread.CurrentThread.ManagedThreadId);
291 //Console.WriteLine("{0}: Active", Thread.CurrentThread.ManagedThreadId);
280 Interlocked.Increment(ref m_activeThreads);
292 int count = 0;;
293 Thread.MemoryBarrier();
281 do {
294 do {
282 // exit if requested
295 // exit if requested
283 if (m_exitRequired != 0) {
296 if (m_exitRequired == 1) {
284 // release the thread slot
297 // release the thread slot
285 Interlocked.Decrement(ref m_activeThreads);
298 Interlocked.Decrement(ref m_activeThreads);
286 if (ReleaseThreadSlotAnyway()) // it was the last worker
299 if (!ReleaseThreadSlotAnyway()) // it was the last worker
287 m_hasTasks.Dispose();
288 else
289 SignalThread(); // wake next worker
300 SignalThread(); // wake next worker
290 break;
301 break;
291 }
302 }
@@ -293,14 +304,17 namespace Implab.Parallels {
293 // fetch task
304 // fetch task
294 if (TryDequeue(out unit)) {
305 if (TryDequeue(out unit)) {
295 InvokeUnit(unit);
306 InvokeUnit(unit);
307 count ++;
296 continue;
308 continue;
297 }
309 }
298 Interlocked.Decrement(ref m_activeThreads);
310 Interlocked.Decrement(ref m_activeThreads);
299
311
312 Console.WriteLine("{0}: Suspend processed({1})", Thread.CurrentThread.ManagedThreadId,count);
300 // entering suspend state
313 // entering suspend state
301 // keep this thread and wait
314 // keep this thread and wait
302 if (!Suspend())
315 if (!Suspend())
303 break;
316 break;
317 count = 0;
304 //Console.WriteLine("{0}: Awake", Thread.CurrentThread.ManagedThreadId);
318 //Console.WriteLine("{0}: Awake", Thread.CurrentThread.ManagedThreadId);
305 Interlocked.Increment(ref m_activeThreads);
319 Interlocked.Increment(ref m_activeThreads);
306 } while (true);
320 } while (true);
@@ -309,15 +323,10 namespace Implab.Parallels {
309
323
310 protected virtual void Dispose(bool disposing) {
324 protected virtual void Dispose(bool disposing) {
311 if (disposing) {
325 if (disposing) {
312 if (m_exitRequired == 0) {
326 if (0 == Interlocked.CompareExchange(ref m_exitRequired, 1, 0)) { // implies memory barrier
313 if (Interlocked.CompareExchange(ref m_exitRequired, 1, 0) != 0)
314 return;
315
316 // wake sleeping threads
327 // wake sleeping threads
317 if (m_createdThreads > 0)
328 if (m_createdThreads > 0)
318 SignalThread();
329 SignalThread();
319 else
320 m_hasTasks.Dispose();
321 GC.SuppressFinalize(this);
330 GC.SuppressFinalize(this);
322 }
331 }
323 }
332 }
Show file before | Show file after | Hide comments
@@ -18,6 +18,8 namespace Implab.Parallels {
18 Node m_last;
18 Node m_last;
19
19
20 public void Enqueue(T value) {
20 public void Enqueue(T value) {
21 Thread.MemoryBarrier();
22
21 var last = m_last;
23 var last = m_last;
22 var next = new Node(value);
24 var next = new Node(value);
23
25
@@ -35,6 +37,7 namespace Implab.Parallels {
35 Node next = null;
37 Node next = null;
36 value = default(T);
38 value = default(T);
37
39
40 Thread.MemoryBarrier();
38 do {
41 do {
39 first = m_first;
42 first = m_first;
40 if (first == null)
43 if (first == null)
Show file before | Show file after | Hide comments
@@ -12,20 +12,24 namespace Implab.Parallels {
12 MTQueue<Action> m_queue = new MTQueue<Action>();
12 MTQueue<Action> m_queue = new MTQueue<Action>();
13 int m_queueLength = 0;
13 int m_queueLength = 0;
14 readonly int m_threshold = 1;
14 readonly int m_threshold = 1;
15 int m_workers = 0;
15
16
16 public WorkerPool(int minThreads, int maxThreads, int threshold)
17 public WorkerPool(int minThreads, int maxThreads, int threshold)
17 : base(minThreads, maxThreads) {
18 : base(minThreads, maxThreads) {
18 m_threshold = threshold;
19 m_threshold = threshold;
20 m_workers = minThreads;
19 InitPool();
21 InitPool();
20 }
22 }
21
23
22 public WorkerPool(int minThreads, int maxThreads) :
24 public WorkerPool(int minThreads, int maxThreads) :
23 base(minThreads, maxThreads) {
25 base(minThreads, maxThreads) {
26 m_workers = minThreads;
24 InitPool();
27 InitPool();
25 }
28 }
26
29
27 public WorkerPool(int threads)
30 public WorkerPool(int threads)
28 : base(threads) {
31 : base(threads) {
32 m_workers = threads;
29 InitPool();
33 InitPool();
30 }
34 }
31
35
@@ -62,8 +66,10 namespace Implab.Parallels {
62 var len = Interlocked.Increment(ref m_queueLength);
66 var len = Interlocked.Increment(ref m_queueLength);
63 m_queue.Enqueue(unit);
67 m_queue.Enqueue(unit);
64
68
65 if (len > m_threshold*ActiveThreads)
69 if (len > m_threshold * m_workers) {
70 Interlocked.Increment(ref m_workers);
66 GrowPool();
71 GrowPool();
72 }
67 }
73 }
68
74
69 protected override bool TryDequeue(out Action unit) {
75 protected override bool TryDequeue(out Action unit) {
@@ -85,8 +91,10 namespace Implab.Parallels {
85 // Suspend
91 // Suspend
86 // queueLength > 0
92 // queueLength > 0
87 // continue
93 // continue
94 Thread.MemoryBarrier();
88 if (m_queueLength > 0)
95 if (m_queueLength > 0)
89 return true;
96 return true;
97 Interlocked.Decrement(ref m_workers);
90 return base.Suspend();
98 return base.Suspend();
91 }
99 }
92
100
Show file before | Show file after | Hide comments
@@ -142,18 +142,20 namespace Implab {
142
142
143 void WaitTransition() {
143 void WaitTransition() {
144 while (m_state == TRANSITIONAL_STATE) {
144 while (m_state == TRANSITIONAL_STATE) {
145 /* noop */
145 Thread.MemoryBarrier();
146 }
146 }
147 }
147 }
148
148
149 public bool IsResolved {
149 public bool IsResolved {
150 get {
150 get {
151 Thread.MemoryBarrier();
151 return m_state > 1;
152 return m_state > 1;
152 }
153 }
153 }
154 }
154
155
155 public bool IsCancelled {
156 public bool IsCancelled {
156 get {
157 get {
158 Thread.MemoryBarrier();
157 return m_state == CANCELLED_STATE;
159 return m_state == CANCELLED_STATE;
158 }
160 }
159 }
161 }
General Comments 0
You need to be logged in to leave comments. Login now