Mercurial > pub > ImplabNet

comparison Implab/Parallels/ArrayTraits.cs @ 30:2fad2d1f4b03

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
small refactoring, cleanup.
author cin
date Mon, 07 Apr 2014 03:25:57 +0400
parents f0bf98e4d22c
children 8eca2652d2ff
comparison
equal deleted inserted replaced
29:768f7deeb55b 30:2fad2d1f4b03
103 m_promise.Reject(e); 103 m_promise.Reject(e);
104 } 104 }
105 } 105 }
106 } 106 }
107 107
108 public static Promise<TDst[]> ParallelMap<TSrc, TDst> (this TSrc[] source, Func<TSrc,TDst> transform, int threads) { 108 public static IPromise<TDst[]> ParallelMap<TSrc, TDst> (this TSrc[] source, Func<TSrc,TDst> transform, int threads) {
109 if (source == null) 109 if (source == null)
110 throw new ArgumentNullException("source"); 110 throw new ArgumentNullException("source");
111 if (transform == null) 111 if (transform == null)
112 throw new ArgumentNullException("transform"); 112 throw new ArgumentNullException("transform");
113 113
114 var mapper = new ArrayMapper<TSrc, TDst>(source, transform, threads); 114 var mapper = new ArrayMapper<TSrc, TDst>(source, transform, threads);
115 return mapper.Promise; 115 return mapper.Promise;
116 } 116 }
117 117
118 public static Promise<int> ParallelForEach<TSrc>(this TSrc[] source, Action<TSrc> action, int threads) { 118 public static IPromise<int> ParallelForEach<TSrc>(this TSrc[] source, Action<TSrc> action, int threads) {
119 if (source == null) 119 if (source == null)
120 throw new ArgumentNullException("source"); 120 throw new ArgumentNullException("source");
121 if (action == null) 121 if (action == null)
122 throw new ArgumentNullException("action"); 122 throw new ArgumentNullException("action");
123 123
134 throw new ArgumentOutOfRangeException("Threads number must be greater then zero"); 134 throw new ArgumentOutOfRangeException("Threads number must be greater then zero");
135 135
136 var promise = new Promise<TDst[]>(); 136 var promise = new Promise<TDst[]>();
137 var res = new TDst[source.Length]; 137 var res = new TDst[source.Length];
138 var pending = source.Length; 138 var pending = source.Length;
139
139 var semaphore = new Semaphore(threads, threads); 140 var semaphore = new Semaphore(threads, threads);
140 141
141 AsyncPool.InvokeNewThread(() => { 142 AsyncPool.InvokeNewThread(() => {
142 for (int i = 0; i < source.Length; i++) { 143 for (int i = 0; i < source.Length; i++) {
143 if(promise.IsResolved) 144 if(promise.IsResolved)