Mercurial > pub > ImplabNet
view Implab/Parallels/ArrayTraits.cs @ 34:dabf79fde388
fixed race condition in DispatchPool
author | cin |
---|---|
date | Thu, 10 Apr 2014 04:20:25 +0400 |
parents | 8eca2652d2ff |
children | 2fc0fbe7d58b |
line wrap: on
line source
using System; using System.Collections.Generic; using System.Diagnostics; using System.Linq; using System.Text; using System.Threading; namespace Implab.Parallels { public static class ArrayTraits { class ArrayIterator<TSrc> : DispatchPool<int> { readonly Action<TSrc> m_action; readonly TSrc[] m_source; readonly Promise<int> m_promise = new Promise<int>(); int m_pending; int m_next; public ArrayIterator(TSrc[] source, Action<TSrc> action, int threads) : base(threads) { Debug.Assert(source != null); Debug.Assert(action != null); m_next = 0; m_source = source; m_pending = source.Length; m_action = action; m_promise.Anyway(() => Dispose()); m_promise.Cancelled(() => Dispose()); InitPool(); } public Promise<int> Promise { get { return m_promise; } } protected override bool TryDequeue(out int unit) { unit = Interlocked.Increment(ref m_next) - 1; return unit >= m_source.Length ? false : true; } protected override void InvokeUnit(int unit) { try { m_action(m_source[unit]); var pending = Interlocked.Decrement(ref m_pending); if (pending == 0) m_promise.Resolve(m_source.Length); } catch (Exception e) { m_promise.Reject(e); } } } class ArrayMapper<TSrc, TDst>: DispatchPool<int> { readonly Func<TSrc, TDst> m_transform; readonly TSrc[] m_source; readonly TDst[] m_dest; readonly Promise<TDst[]> m_promise = new Promise<TDst[]>(); int m_pending; int m_next; public ArrayMapper(TSrc[] source, Func<TSrc, TDst> transform, int threads) : base(threads) { Debug.Assert (source != null); Debug.Assert( transform != null); m_next = 0; m_source = source; m_dest = new TDst[source.Length]; m_pending = source.Length; m_transform = transform; m_promise.Anyway(() => Dispose()); m_promise.Cancelled(() => Dispose()); InitPool(); } public Promise<TDst[]> Promise { get { return m_promise; } } protected override bool TryDequeue(out int unit) { unit = Interlocked.Increment(ref m_next) - 1; return unit >= m_source.Length ? false : true; } protected override void InvokeUnit(int unit) { try { m_dest[unit] = m_transform(m_source[unit]); var pending = Interlocked.Decrement(ref m_pending); if (pending == 0) m_promise.Resolve(m_dest); } catch (Exception e) { m_promise.Reject(e); } } } public static IPromise<TDst[]> ParallelMap<TSrc, TDst> (this TSrc[] source, Func<TSrc,TDst> transform, int threads) { if (source == null) throw new ArgumentNullException("source"); if (transform == null) throw new ArgumentNullException("transform"); var mapper = new ArrayMapper<TSrc, TDst>(source, transform, threads); return mapper.Promise; } public static IPromise<int> ParallelForEach<TSrc>(this TSrc[] source, Action<TSrc> action, int threads) { if (source == null) throw new ArgumentNullException("source"); if (action == null) throw new ArgumentNullException("action"); var iter = new ArrayIterator<TSrc>(source, action, threads); return iter.Promise; } public static IPromise<TDst[]> ChainedMap<TSrc, TDst>(this TSrc[] source, ChainedOperation<TSrc, TDst> transform, int threads) { if (source == null) throw new ArgumentNullException("source"); if (transform == null) throw new ArgumentNullException("transform"); if (threads <= 0) throw new ArgumentOutOfRangeException("Threads number must be greater then zero"); if (source.Length == 0) return Promise<TDst[]>.ResultToPromise(new TDst[0]); var promise = new Promise<TDst[]>(); var res = new TDst[source.Length]; var pending = source.Length; var semaphore = new Semaphore(threads, threads); AsyncPool.InvokeNewThread(() => { for (int i = 0; i < source.Length; i++) { if(promise.IsResolved) break; // stop processing in case of error or cancellation var idx = i; semaphore.WaitOne(); try { var p1 = transform(source[i]); p1.Anyway(() => semaphore.Release()); p1.Cancelled(() => semaphore.Release()); p1.Then( 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); } } return 0; }); return promise.Anyway(() => semaphore.Dispose()); } } }