Mercurial > pub > ImplabNet

comparison Implab/Parallels/ArrayTraits.cs @ 76:c761fc982e1d v2

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
Refactoring of the IPromise<T> interface Added tests
author cin
date Wed, 10 Sep 2014 17:53:05 +0400
parents 4439140706d0
children 4f20870d0816
comparison
equal deleted inserted replaced
75:4439140706d0 76:c761fc982e1d
27 m_next = 0; 27 m_next = 0;
28 m_source = source; 28 m_source = source;
29 m_pending = source.Length; 29 m_pending = source.Length;
30 m_action = action; 30 m_action = action;
31 31
32 m_promise.Finally(Dispose); 32 m_promise.Anyway(Dispose);
33 33
34 InitPool(); 34 InitPool();
35 } 35 }
36 36
37 public Promise<int> Promise { 37 public Promise<int> Promise {
83 m_dest = new TDst[source.Length]; 83 m_dest = new TDst[source.Length];
84 m_pending = source.Length; 84 m_pending = source.Length;
85 m_transform = transform; 85 m_transform = transform;
86 m_traceContext = TraceContext.Snapshot(); 86 m_traceContext = TraceContext.Snapshot();
87 87
88 m_promise.Finally(Dispose); 88 m_promise.Anyway(Dispose);
89 89
90 InitPool(); 90 InitPool();
91 } 91 }
92 92
93 public Promise<TDst[]> Promise { 93 public Promise<TDst[]> Promise {
136 136
137 var iter = new ArrayIterator<TSrc>(source, action, threads); 137 var iter = new ArrayIterator<TSrc>(source, action, threads);
138 return iter.Promise; 138 return iter.Promise;
139 } 139 }
140 140
141 public static IPromise<TDst[]> ChainedMap<TSrc, TDst>(this TSrc[] source, ChainedOperation<TSrc, TDst> transform, int threads) { 141 public static IPromise<TDst[]> ChainedMap<TSrc, TDst>(this TSrc[] source, ResultMapper<TSrc, IPromise<TDst>> transform, int threads) {
142 if (source == null) 142 if (source == null)
143 throw new ArgumentNullException("source"); 143 throw new ArgumentNullException("source");
144 if (transform == null) 144 if (transform == null)
145 throw new ArgumentNullException("transform"); 145 throw new ArgumentNullException("transform");
146 if (threads <= 0) 146 if (threads <= 0)
163 var idx = i; 163 var idx = i;
164 164
165 semaphore.WaitOne(); 165 semaphore.WaitOne();
166 try { 166 try {
167 var p1 = transform(source[i]); 167 var p1 = transform(source[i]);
168 p1.Finally(() => semaphore.Release()); 168 p1.Anyway(() => semaphore.Release());
169 p1.Then( 169 p1.Then(
170 x => { 170 x => {
171 res[idx] = x; 171 res[idx] = x;
172 var left = Interlocked.Decrement(ref pending); 172 var left = Interlocked.Decrement(ref pending);
173 if (left == 0) 173 if (left == 0)
184 } 184 }
185 } 185 }
186 return 0; 186 return 0;
187 }); 187 });
188 188
189 return promise.Finally(semaphore.Dispose); 189 return promise.Anyway(semaphore.Dispose);
190 } 190 }
191 191
192 } 192 }
193 } 193 }