pub/ImplabNet: Implab/Parallels/ArrayTraits.cs comparison

added memory barriers

comparison

equal deleted inserted replaced

-:05e6468f066f
+:4f20870d0816
 var promise = new Promise<TDst[]>();
 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(() => {
 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();
+lock(locker) {
+while(slots == 0)
+Monitor.Wait(locker);
+slots--;
+}
 try {
-var p1 = transform(source[i]);
+transform(source[i])
-p1.Anyway(() => semaphore.Release());
+.Anyway(() => {
-p1.Then(
+lock(locker) {
-x => {
+slots ++;
-res[idx] = x;
+Monitor.Pulse(locker);
-var left = Interlocked.Decrement(ref pending);
+}
-if (left == 0)
+})
-promise.Resolve(res);
+.Last(
-},
+x => {
-e => {
+res[idx] = x;
-promise.Reject(e);
+var left = Interlocked.Decrement(ref pending);
-throw new TransientPromiseException(e);
+if (left == 0)
-}
+promise.Resolve(res);
-);
+},
+e => promise.Reject(e)
+);
 } catch (Exception e) {
 promise.Reject(e);
 }
 }
 return 0;
 });
-return promise.Anyway(semaphore.Dispose);
+return promise;
 }
 }
 }

Mercurial > pub > ImplabNet