##// END OF EJS Templates
implab » ImplabNet
RSS
Promise is rewritten to use interlocked operations instead of locks
Promise is rewritten to use interlocked operations instead of locks
cin - - Load All Authors

File last commit:

r17:7cd4a843b4e4 promises
r19:e3935fdf59a2 promises
Show More
DispatchPool.cs
238 lines | 7.3 KiB | text/x-csharp | CSharpLexer
/ Implab / Parallels / DispatchPool.cs
History | Source | Raw |Copy content |Copy permalink
cin
implemented parallel map and foreach for arrays...
 r15 using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using System.Diagnostics;
namespace Implab.Parallels {
public abstract class DispatchPool<TUnit> : IDisposable {
readonly int m_minThreads;
readonly int m_maxThreads;
int m_runningThreads = 0;
int m_maxRunningThreads = 0;
int m_suspended = 0;
int m_exitRequired = 0;
AutoResetEvent m_hasTasks = new AutoResetEvent(false);
protected DispatchPool(int min, int max) {
if (min < 0)
throw new ArgumentOutOfRangeException("min");
if (max <= 0)
throw new ArgumentOutOfRangeException("max");
if (min > max)
min = max;
m_minThreads = min;
m_maxThreads = max;
}
protected DispatchPool(int threads)
: this(threads, threads) {
}
protected DispatchPool() {
int maxThreads, maxCP;
ThreadPool.GetMaxThreads(out maxThreads, out maxCP);
m_minThreads = 0;
m_maxThreads = maxThreads;
}
protected void InitPool() {
for (int i = 0; i < m_minThreads; i++)
StartWorker();
}
public int ThreadCount {
get {
return m_runningThreads;
}
}
public int MaxRunningThreads {
get {
return m_maxRunningThreads;
}
}
protected bool IsDisposed {
get {
return m_exitRequired != 0;
}
}
cin
Improved worker pool
 r17 protected abstract bool TryDequeue(out TUnit unit);
protected virtual bool ExtendPool() {
if (m_suspended > 0) {
m_hasTasks.Set();
return true;
} else
return StartWorker();
}
/// <summary>
/// Запускает либо новый поток, если раньше не было ни одного потока, либо устанавливает событие пробуждение одного спящего потока
/// </summary>
protected void WakePool() {
m_hasTasks.Set(); // wake sleeping thread;
if (AllocateThreadSlot(1)) {
var worker = new Thread(this.Worker);
worker.IsBackground = true;
worker.Start();
}
}
protected virtual void Suspend() {
m_hasTasks.WaitOne();
}
#region thread slots traits
bool AllocateThreadSlot() {
cin
sync
 r16 int current;
cin
implemented parallel map and foreach for arrays...
 r15 // use spins to allocate slot for the new thread
do {
cin
sync
 r16 current = m_runningThreads;
cin
implemented parallel map and foreach for arrays...
 r15 if (current >= m_maxThreads || m_exitRequired != 0)
// no more slots left or the pool has been disposed
return false;
} while (current != Interlocked.CompareExchange(ref m_runningThreads, current + 1, current));
cin
Improved worker pool
 r17 UpdateMaxThreads(current + 1);
return true;
}
cin
implemented parallel map and foreach for arrays...
 r15
cin
Improved worker pool
 r17 bool AllocateThreadSlot(int desired) {
if (desired - 1 != Interlocked.CompareExchange(ref m_runningThreads, desired, desired - 1))
return false;
UpdateMaxThreads(desired);
cin
implemented parallel map and foreach for arrays...
 r15
cin
Improved worker pool
 r17 return true;
}
bool ReleaseThreadSlot(out bool last) {
last = false;
int current;
// use spins to release slot for the new thread
do {
current = m_runningThreads;
if (current <= m_minThreads && m_exitRequired == 0)
// the thread is reserved
return false;
} while (current != Interlocked.CompareExchange(ref m_runningThreads, current - 1, current));
last = (current == 1);
cin
implemented parallel map and foreach for arrays...
 r15
return true;
}
cin
Improved worker pool
 r17 /// <summary>
/// releases thread slot unconditionally, used during cleanup
/// </summary>
/// <returns>true - no more threads left</returns>
bool ReleaseThreadSlotAnyway() {
var left = Interlocked.Decrement(ref m_runningThreads);
return left == 0;
cin
implemented parallel map and foreach for arrays...
 r15 }
cin
Improved worker pool
 r17 void UpdateMaxThreads(int count) {
int max;
cin
sync
 r16 do {
cin
Improved worker pool
 r17 max = m_maxRunningThreads;
if (max >= count)
break;
} while(max != Interlocked.CompareExchange(ref m_maxRunningThreads, count, max));
cin
sync
 r16 }
cin
Improved worker pool
 r17 #endregion
bool StartWorker() {
if (AllocateThreadSlot()) {
// slot successfully allocated
var worker = new Thread(this.Worker);
worker.IsBackground = true;
worker.Start();
return true;
} else {
return false;
}
cin
sync
 r16 }
cin
implemented parallel map and foreach for arrays...
 r15 bool FetchTask(out TUnit unit) {
do {
// exit if requested
if (m_exitRequired != 0) {
// release the thread slot
cin
Improved worker pool
 r17 if (ReleaseThreadSlotAnyway()) // it was the last worker
cin
implemented parallel map and foreach for arrays...
 r15 m_hasTasks.Dispose();
else
cin
Improved worker pool
 r17 m_hasTasks.Set(); // wake next worker
cin
implemented parallel map and foreach for arrays...
 r15 unit = default(TUnit);
return false;
}
// fetch task
if (TryDequeue(out unit)) {
cin
Improved worker pool
 r17 ExtendPool();
cin
implemented parallel map and foreach for arrays...
 r15 return true;
}
//no tasks left, exit if the thread is no longer needed
cin
Improved worker pool
 r17 bool last;
if (ReleaseThreadSlot(out last)) {
if (last && m_hasTasks.WaitOne(0)) {
if (AllocateThreadSlot(1))
continue; // spin again...
else
// we failed to reallocate slot for this thread
// therefore we need to release the event
m_hasTasks.Set();
}
cin
sync
 r16
cin
implemented parallel map and foreach for arrays...
 r15 return false;
}
cin
sync
 r16 // entering suspend state
cin
implemented parallel map and foreach for arrays...
 r15 Interlocked.Increment(ref m_suspended);
cin
sync
 r16 // keep this thread and wait
Suspend();
cin
implemented parallel map and foreach for arrays...
 r15 Interlocked.Decrement(ref m_suspended);
} while (true);
}
protected abstract void InvokeUnit(TUnit unit);
void Worker() {
TUnit unit;
while (FetchTask(out unit))
InvokeUnit(unit);
}
protected virtual void Dispose(bool disposing) {
if (disposing) {
if (m_exitRequired == 0) {
if (Interlocked.CompareExchange(ref m_exitRequired, 1, 0) != 0)
return;
// wake sleeping threads
m_hasTasks.Set();
GC.SuppressFinalize(this);
}
}
}
public void Dispose() {
Dispose(true);
}
~DispatchPool() {
Dispose(false);
}
}
}