diff --git a/Implab.Test/AsyncTests.cs b/Implab.Test/AsyncTests.cs --- a/Implab.Test/AsyncTests.cs +++ b/Implab.Test/AsyncTests.cs @@ -291,7 +291,7 @@ namespace Implab.Test { [TestMethod] public void ChainedMapTest() { - using (var pool = new WorkerPool(0,100,100)) { + using (var pool = new WorkerPool(0,10,100)) { const int count = 10000; var args = new double[count]; diff --git a/Implab.mono.sln b/Implab.mono.sln --- a/Implab.mono.sln +++ b/Implab.mono.sln @@ -16,8 +16,6 @@ Project("{2150E333-8FDC-42A3-9474-1A3956 EndProject Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Implab.Test.mono", "Implab.Test\Implab.Test.mono.csproj", "{2BD05F84-E067-4B87-9477-FDC2676A21C6}" EndProject -Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Implab.Fx.Test.mono", "Implab.Fx.Test\Implab.Fx.Test.mono.csproj", "{2BD05F84-E067-4B87-9477-FDC2676A21C6}" -EndProject Global GlobalSection(SolutionConfigurationPlatforms) = preSolution Debug|Any CPU = Debug|Any CPU @@ -77,7 +75,6 @@ Global EndGlobalSection GlobalSection(NestedProjects) = preSolution {2BD05F84-E067-4B87-9477-FDC2676A21C6} = {BCA337C3-BFDC-4825-BBDB-E6D467E4E452} - {2BD05F84-E067-4B87-9477-FDC2676A21C6} = {BCA337C3-BFDC-4825-BBDB-E6D467E4E452} EndGlobalSection GlobalSection(MonoDevelopProperties) = preSolution StartupItem = Implab\Implab.csproj diff --git a/Implab/Implab.csproj b/Implab/Implab.csproj --- a/Implab/Implab.csproj +++ b/Implab/Implab.csproj @@ -7,6 +7,8 @@ Library Implab Implab + 8.0.30703 + 2.0 true @@ -121,6 +123,8 @@ + + diff --git a/Implab/JSON/JSONXmlReaderOptions.cs b/Implab/JSON/JSONXmlReaderOptions.cs --- a/Implab/JSON/JSONXmlReaderOptions.cs +++ b/Implab/JSON/JSONXmlReaderOptions.cs @@ -19,7 +19,9 @@ namespace Implab.JSON { } /// - /// Интерпретировать массивы как множественные элементы (убирает один уровень вложенности) + /// Интерпретировать массивы как множественные элементы (убирает один уровень вложенности), иначе массив + /// представляется в виде узла, дочерними элементами которого являются элементы массива, имена дочерних элементов + /// определяются свойством . По умолчанию false. /// public bool FlattenArrays { get; @@ -44,6 +46,7 @@ namespace Implab.JSON { /// /// Имя элемента для массивов, если не включена опция . + /// По умолчанию item. /// public string ArrayItemName { get; diff --git a/Implab/Parallels/ArrayTraits.cs b/Implab/Parallels/ArrayTraits.cs --- a/Implab/Parallels/ArrayTraits.cs +++ b/Implab/Parallels/ArrayTraits.cs @@ -153,7 +153,8 @@ namespace Implab.Parallels { var res = new TDst[source.Length]; var pending = source.Length; - var semaphore = new Semaphore(threads, threads); + object locker = new object(); + int slots = threads; // Analysis disable AccessToDisposedClosure AsyncPool.InvokeNewThread(() => { @@ -162,22 +163,28 @@ namespace Implab.Parallels { break; // stop processing in case of error or cancellation var idx = i; - semaphore.WaitOne(); + lock(locker) { + while(slots == 0) + Monitor.Wait(locker); + slots--; + } try { - var p1 = transform(source[i]); - p1.Anyway(() => semaphore.Release()); - p1.Then( - x => { - res[idx] = x; - var left = Interlocked.Decrement(ref pending); - if (left == 0) - promise.Resolve(res); - }, - e => { - promise.Reject(e); - throw new TransientPromiseException(e); - } - ); + transform(source[i]) + .Anyway(() => { + lock(locker) { + slots ++; + Monitor.Pulse(locker); + } + }) + .Last( + x => { + res[idx] = x; + var left = Interlocked.Decrement(ref pending); + if (left == 0) + promise.Resolve(res); + }, + e => promise.Reject(e) + ); } catch (Exception e) { promise.Reject(e); @@ -186,7 +193,7 @@ namespace Implab.Parallels { return 0; }); - return promise.Anyway(semaphore.Dispose); + return promise; } } diff --git a/Implab/Parallels/DispatchPool.cs b/Implab/Parallels/DispatchPool.cs --- a/Implab/Parallels/DispatchPool.cs +++ b/Implab/Parallels/DispatchPool.cs @@ -9,16 +9,17 @@ namespace Implab.Parallels { public abstract class DispatchPool : IDisposable { readonly int m_minThreads; readonly int m_maxThreads; + readonly int m_releaseTimeout = 100; // the timeout while the working thread will wait for the new tasks before exit int m_createdThreads = 0; // the current size of the pool int m_activeThreads = 0; // the count of threads which are active int m_sleepingThreads = 0; // the count of currently inactive threads int m_maxRunningThreads = 0; // the meximum reached size of the pool int m_exitRequired = 0; // the pool is going to shutdown, all unused workers are released - int m_releaseTimeout = 100; // the timeout while the working thread will wait for the new tasks before exit + int m_wakeEvents = 0; // the count of wake events - AutoResetEvent m_hasTasks = new AutoResetEvent(false); + readonly object m_signalLocker = new object(); protected DispatchPool(int min, int max) { if (min < 0) @@ -51,68 +52,76 @@ namespace Implab.Parallels { public int PoolSize { get { + Thread.MemoryBarrier(); return m_createdThreads; } } public int ActiveThreads { get { + Thread.MemoryBarrier(); return m_activeThreads; } } public int MaxRunningThreads { get { + Thread.MemoryBarrier(); return m_maxRunningThreads; } } protected bool IsDisposed { get { - return m_exitRequired != 0; + Thread.MemoryBarrier(); + return m_exitRequired == 1; } } protected abstract bool TryDequeue(out TUnit unit); - #region thread execution traits + #region thread signaling traits int SignalThread() { var signals = Interlocked.Increment(ref m_wakeEvents); if(signals == 1) - m_hasTasks.Set(); + lock(m_signalLocker) + Monitor.Pulse(m_signalLocker); return signals; } bool FetchSignalOrWait(int timeout) { var start = Environment.TickCount; - - // означает, что поток владеет блокировкой и при успешном получении сигнала должен - // ее вернуть, чтобы другой ожидающий поток смог - bool hasLock = false; + int signals; + Thread.MemoryBarrier(); // m_wakeEvents volatile first read do { - int signals; - do { - signals = m_wakeEvents; - if (signals == 0) - break; - } while (Interlocked.CompareExchange(ref m_wakeEvents, signals - 1, signals) != signals); + signals = m_wakeEvents; + if (signals == 0) + break; + } while (Interlocked.CompareExchange(ref m_wakeEvents, signals - 1, signals) != signals); - if (signals >= 1) { - if (signals > 1 && hasLock) - m_hasTasks.Set(); + if (signals == 0) { + // no signal is fetched + lock(m_signalLocker) { + while(m_wakeEvents == 0) { + if (timeout != -1) + timeout = Math.Max(0, timeout - (Environment.TickCount - start)); + if(!Monitor.Wait(m_signalLocker,timeout)) + return false; // timeout + } + // m_wakeEvents > 0 + if (Interlocked.Decrement(ref m_wakeEvents) > 0) //syncronized + Monitor.Pulse(m_signalLocker); + + // signal fetched return true; } - if (timeout != -1) - timeout = Math.Max(0, timeout - (Environment.TickCount - start)); + } else { + // signal fetched + return true; + } - // если сигналов больше не осталось, то первый поток, который дошел сюда сбросит событие - // и уйдет на пустой цикл, после чего заблокируется - hasLock = true; - } while (m_hasTasks.WaitOne(timeout)); - - return false; } bool Sleep(int timeout) { @@ -131,7 +140,8 @@ namespace Implab.Parallels { /// Запускает либо новый поток, если раньше не было ни одного потока, либо устанавливает событие пробуждение одного спящего потока /// protected void GrowPool() { - if (m_exitRequired != 0) + Thread.MemoryBarrier(); + if (m_exitRequired == 1) return; if (m_sleepingThreads > m_wakeEvents) { //Console.WriteLine("Waking threads (sleeps {0}, pending {1})", m_sleepingThreads, m_wakeEvents); @@ -204,7 +214,7 @@ namespace Implab.Parallels { // use spins to allocate slot for the new thread do { current = m_createdThreads; - if (current >= m_maxThreads || m_exitRequired != 0) + if (current >= m_maxThreads || m_exitRequired == 1) // no more slots left or the pool has been disposed return false; } while (current != Interlocked.CompareExchange(ref m_createdThreads, current + 1, current)); @@ -227,6 +237,7 @@ namespace Implab.Parallels { last = false; int current; // use spins to release slot for the new thread + Thread.MemoryBarrier(); do { current = m_createdThreads; if (current <= m_minThreads && m_exitRequired == 0) @@ -264,6 +275,7 @@ namespace Implab.Parallels { // slot successfully allocated var worker = new Thread(this.Worker); worker.IsBackground = true; + Interlocked.Increment(ref m_activeThreads); worker.Start(); return true; @@ -277,15 +289,14 @@ namespace Implab.Parallels { protected virtual void Worker() { TUnit unit; //Console.WriteLine("{0}: Active", Thread.CurrentThread.ManagedThreadId); - Interlocked.Increment(ref m_activeThreads); + int count = 0;; + Thread.MemoryBarrier(); do { // exit if requested - if (m_exitRequired != 0) { + if (m_exitRequired == 1) { // release the thread slot Interlocked.Decrement(ref m_activeThreads); - if (ReleaseThreadSlotAnyway()) // it was the last worker - m_hasTasks.Dispose(); - else + if (!ReleaseThreadSlotAnyway()) // it was the last worker SignalThread(); // wake next worker break; } @@ -293,14 +304,17 @@ namespace Implab.Parallels { // fetch task if (TryDequeue(out unit)) { InvokeUnit(unit); + count ++; continue; } Interlocked.Decrement(ref m_activeThreads); + Console.WriteLine("{0}: Suspend processed({1})", Thread.CurrentThread.ManagedThreadId,count); // entering suspend state // keep this thread and wait if (!Suspend()) break; + count = 0; //Console.WriteLine("{0}: Awake", Thread.CurrentThread.ManagedThreadId); Interlocked.Increment(ref m_activeThreads); } while (true); @@ -309,15 +323,10 @@ namespace Implab.Parallels { protected virtual void Dispose(bool disposing) { if (disposing) { - if (m_exitRequired == 0) { - if (Interlocked.CompareExchange(ref m_exitRequired, 1, 0) != 0) - return; - + if (0 == Interlocked.CompareExchange(ref m_exitRequired, 1, 0)) { // implies memory barrier // wake sleeping threads if (m_createdThreads > 0) SignalThread(); - else - m_hasTasks.Dispose(); GC.SuppressFinalize(this); } } diff --git a/Implab/Parallels/MTQueue.cs b/Implab/Parallels/MTQueue.cs --- a/Implab/Parallels/MTQueue.cs +++ b/Implab/Parallels/MTQueue.cs @@ -18,6 +18,8 @@ namespace Implab.Parallels { Node m_last; public void Enqueue(T value) { + Thread.MemoryBarrier(); + var last = m_last; var next = new Node(value); @@ -35,6 +37,7 @@ namespace Implab.Parallels { Node next = null; value = default(T); + Thread.MemoryBarrier(); do { first = m_first; if (first == null) diff --git a/Implab/Parallels/WorkerPool.cs b/Implab/Parallels/WorkerPool.cs --- a/Implab/Parallels/WorkerPool.cs +++ b/Implab/Parallels/WorkerPool.cs @@ -12,20 +12,24 @@ namespace Implab.Parallels { MTQueue m_queue = new MTQueue(); int m_queueLength = 0; readonly int m_threshold = 1; + int m_workers = 0; public WorkerPool(int minThreads, int maxThreads, int threshold) : base(minThreads, maxThreads) { m_threshold = threshold; + m_workers = minThreads; InitPool(); } public WorkerPool(int minThreads, int maxThreads) : base(minThreads, maxThreads) { + m_workers = minThreads; InitPool(); } public WorkerPool(int threads) : base(threads) { + m_workers = threads; InitPool(); } @@ -62,8 +66,10 @@ namespace Implab.Parallels { var len = Interlocked.Increment(ref m_queueLength); m_queue.Enqueue(unit); - if (len > m_threshold*ActiveThreads) + if (len > m_threshold * m_workers) { + Interlocked.Increment(ref m_workers); GrowPool(); + } } protected override bool TryDequeue(out Action unit) { @@ -85,8 +91,10 @@ namespace Implab.Parallels { // Suspend // queueLength > 0 // continue + Thread.MemoryBarrier(); if (m_queueLength > 0) return true; + Interlocked.Decrement(ref m_workers); return base.Suspend(); } diff --git a/Implab/Promise.cs b/Implab/Promise.cs --- a/Implab/Promise.cs +++ b/Implab/Promise.cs @@ -142,18 +142,20 @@ namespace Implab { void WaitTransition() { while (m_state == TRANSITIONAL_STATE) { - /* noop */ + Thread.MemoryBarrier(); } } public bool IsResolved { get { + Thread.MemoryBarrier(); return m_state > 1; } } public bool IsCancelled { get { + Thread.MemoryBarrier(); return m_state == CANCELLED_STATE; } } diff --git a/Implab/SyncPool.cs b/Implab/SyncPool.cs new file mode 100644 --- /dev/null +++ b/Implab/SyncPool.cs @@ -0,0 +1,85 @@ +using System; +using Implab.Parallels; +using System.Threading; + +namespace Implab { + /*public class SyncPool : IDisposable { + readonly Func m_factory; + readonly Action m_cleanup; + readonly int m_size; + readonly MTQueue m_queue = new MTQueue(); + + volatile bool m_disposed; + + volatile int m_count; + + public SyncPool(Func factory, Action cleanup, int size) { + Safe.ArgumentNotNull(factory, "factory"); + Safe.ArgumentInRange(size, 1, size, "size"); + + m_factory = factory; + m_cleanup = cleanup; + m_size = size; + } + + public SyncPool(Func factory, Action cleanup) : this(factory,cleanup,Environment.ProcessorCount+1) { + } + + public SyncPool(Func factory) : this(factory,null,Environment.ProcessorCount+1) { + } + + public SyncPoolWrapper Allocate() { + if (m_disposed) + throw new ObjectDisposedException(this.ToString()); + + T instance; + if (m_queue.TryDequeue(out instance)) { + Interlocked.Decrement(ref m_count); + return instance; + } else { + instance = m_factory(); + } + return new SyncPoolWrapper(instance, this); + } + + public void Release(T instance) { + if (m_count < m_size && !m_disposed) { + Interlocked.Increment(ref m_count); + + if (m_cleanup != null) + m_cleanup(instance); + + m_queue.Enqueue(instance); + + // пока элемент возвращался в кеш, была начата операция освобождения всего кеша + // и возможно уже законцена, в таком случае следует извлечь элемент обратно и + // освободить его. Если операция освобождения кеша еще не заврешилась, то будет + // изъят и освобожден произвольный элемен, что не повлияет на ход всего процесса. + if (m_disposed && m_queue.TryDequeue(out instance)) + Safe.Dispose(instance); + + } else { + Safe.Dispose(instance); + } + } + + protected virtual void Dispose(bool disposing) { + if (disposing) { + m_disposed = true; + T instance; + while (m_queue.TryDequeue(out instance)) + Safe.Dispose(instance); + } + } + + #region IDisposable implementation + + public void Dispose() { + Dispose(true); + GC.SuppressFinalize(this); + } + + #endregion + }*/ +} + diff --git a/Implab/SyncPoolWrapper.cs b/Implab/SyncPoolWrapper.cs new file mode 100644 --- /dev/null +++ b/Implab/SyncPoolWrapper.cs @@ -0,0 +1,24 @@ +using System; + +namespace Implab { + /*public struct SyncPoolWrapper : IDisposable { + readonly T m_value; + readonly SyncPool m_pool; + + internal SyncPoolWrapper(T value, SyncPool pool) { + m_value = value; + m_pool = pool; + } + + public T Value { + get { return m_value; } + } + + #region IDisposable implementation + public void Dispose() { + m_pool.Release(m_value); + } + #endregion + }*/ +} +