32 namespace System.Threading.Tasks
36 internal static int GetBestWorkerNumber ()
41 internal static int GetBestWorkerNumber (
TaskScheduler scheduler)
43 return Math.Min (Environment.ProcessorCount, (scheduler ??
TaskScheduler.
Current).MaximumConcurrencyLevel);
46 static int GetBestWorkerNumber (
int from,
int to,
ParallelOptions options, out
int step)
52 if ((step = (to - from) / num) < 5) {
54 num = (to - from) / 5;
62 static void HandleExceptions (IEnumerable<Task> tasks)
64 HandleExceptions (tasks, null);
69 List<Exception> exs =
new List<Exception> ();
70 foreach (
Task t
in tasks) {
71 if (t.Exception != null)
72 exs.Add (t.Exception);
77 infos.IsExceptional =
true;
87 for (
int i = 0; i < count; i++) {
89 tasks [i] =
Task.Factory.StartNew (action, creation);
104 return For (fromInclusive, toExclusive,
ParallelOptions.Default, body);
109 return For (fromInclusive, toExclusive, parallelOptions, (index, state) => body (index));
114 return For<object> (fromInclusive, toExclusive, parallelOptions, () => null, (i, s, l) => { body (i, s);
return null; }, _ => {});
119 Func<TLocal> localInit,
120 Func<int, ParallelLoopState, TLocal, TLocal> body,
121 Action<TLocal> localFinally)
123 return For<TLocal> (fromInclusive, toExclusive,
ParallelOptions.Default, localInit, body, localFinally);
129 Func<TLocal> localInit,
130 Func<int, ParallelLoopState, TLocal, TLocal> body,
131 Action<TLocal> localFinally)
134 throw new ArgumentNullException (
"body");
135 if (localInit == null)
136 throw new ArgumentNullException (
"localInit");
137 if (localFinally == null)
138 throw new ArgumentNullException (
"localFinally");
139 if (parallelOptions == null)
140 throw new ArgumentNullException (
"options");
141 if (fromInclusive >= toExclusive)
146 int num = GetBestWorkerNumber (fromInclusive, toExclusive, parallelOptions, out step);
150 StealRange[] ranges =
new StealRange[num];
151 for (
int i = 0; i < num; i++)
152 ranges[i] =
new StealRange (fromInclusive, i, step);
156 int currentIndex = -1;
158 Action workerMethod = delegate {
159 int localWorker = Interlocked.Increment (ref currentIndex);
160 StealRange range = ranges[localWorker];
161 int index = range.V64.Actual;
162 int stopIndex = localWorker + 1 == num ? toExclusive : Math.Min (toExclusive, index + step);
163 TLocal local = localInit ();
169 for (
int i = index; i < stopIndex;) {
175 if (i >= stopIndex - range.V64.Stolen)
178 if (infos.LowestBreakIteration != null && infos.LowestBreakIteration > i)
181 state.CurrentIteration = i;
182 local = body (i, state, local);
184 if (i + 1 >= stopIndex - range.V64.Stolen)
187 range.V64.Actual = ++i;
190 bool sixtyfour = IntPtr.Size == 8;
193 int len = num + localWorker;
194 for (
int sIndex = localWorker + 1; sIndex < len; ++sIndex) {
195 int extWorker = sIndex % num;
196 range = ranges[extWorker];
198 stopIndex = extWorker + 1 == num ? toExclusive : Math.Min (toExclusive, fromInclusive + (extWorker + 1) * step);
204 StealValue64 val =
new StealValue64 ();
206 old = sixtyfour ? range.V64.Value : Interlocked.CompareExchange (ref range.V64.Value, 0, 0);
209 if (val.Actual >= stopIndex - val.Stolen - 2)
211 stolen = (val.Stolen += 1);
213 if (Interlocked.CompareExchange (ref range.V64.Value, val.Value, old) == old)
217 stolen = stopIndex - stolen;
219 if (stolen > range.V64.Actual)
220 local = body (stolen, state, local);
229 localFinally (local);
233 InitTasks (tasks, num, workerMethod, parallelOptions);
236 Task.WaitAll (tasks);
238 HandleExceptions (tasks, infos);
241 return new ParallelLoopResult (infos.LowestBreakIteration, !(infos.IsStopped || infos.IsExceptional));
244 [StructLayout(LayoutKind.Explicit)]
245 struct StealValue64 {
256 public StealValue64 V64 =
new StealValue64 ();
258 public StealRange (
int fromInclusive,
int i,
int step)
260 V64.Actual = fromInclusive + i * step;
271 return For (fromInclusive, toExclusive,
ParallelOptions.Default, body);
277 return For (fromInclusive, toExclusive,
ParallelOptions.Default, body);
283 return For (fromInclusive, toExclusive, parallelOptions, (index, state) => body (index));
289 return For<object> (fromInclusive, toExclusive, parallelOptions, () => null, (i, s, l) => { body (i, s);
return null; }, _ => {});
295 Func<TLocal> localInit,
296 Func<long, ParallelLoopState, TLocal, TLocal> body,
297 Action<TLocal> localFinally)
299 return For<TLocal> (fromInclusive, toExclusive,
ParallelOptions.Default, localInit, body, localFinally);
302 [MonoTODO (
"See how this can be refactored with the above For implementation")]
306 Func<TLocal> localInit,
307 Func<long, ParallelLoopState, TLocal, TLocal> body,
308 Action<TLocal> localFinally)
311 throw new ArgumentNullException (
"body");
312 if (localInit == null)
313 throw new ArgumentNullException (
"localInit");
314 if (localFinally == null)
315 throw new ArgumentNullException (
"localFinally");
316 if (parallelOptions == null)
317 throw new ArgumentNullException (
"options");
318 if (fromInclusive >= toExclusive)
321 throw new NotImplementedException ();
328 Func<TLocal> init, Func<TSource, ParallelLoopState, TLocal, TLocal> action,
329 Action<TLocal> destruct)
331 if (enumerable == null)
332 throw new ArgumentNullException (
"source");
334 throw new ArgumentNullException (
"options");
336 throw new ArgumentNullException (
"action");
338 throw new ArgumentNullException (
"init");
339 if (destruct == null)
340 throw new ArgumentNullException (
"destruct");
342 int num = Math.Min (GetBestWorkerNumber (options.TaskScheduler),
343 options != null && options.MaxDegreeOfParallelism != -1 ? options.MaxDegreeOfParallelism :
int.MaxValue);
348 SimpleConcurrentBag<TSource> bag =
new SimpleConcurrentBag<TSource> (num);
349 const int bagCount = 5;
351 IList<IEnumerator<TSource>> slices = enumerable (num);
355 Action workerMethod = delegate {
356 IEnumerator<TSource> slice = slices[Interlocked.Increment (ref sliceIndex)];
358 TLocal local = init ();
360 int workIndex = bag.GetNextIndex ();
368 if (infos.IsStopped || infos.IsBroken.Value)
373 for (
int i = 0; i < bagCount && (cont = slice.MoveNext ()); i++) {
374 bag.Add (workIndex, slice.Current);
377 for (
int i = 0; i < bagCount && bag.TryTake (workIndex, out element); i++) {
383 local = action (element, state, local);
387 while (bag.TrySteal (workIndex, out element)) {
390 local = action (element, state, local);
392 if (infos.IsStopped || infos.IsBroken.Value)
400 InitTasks (tasks, num, workerMethod, options);
403 Task.WaitAll (tasks);
405 HandleExceptions (tasks, infos);
408 return new ParallelLoopResult (infos.LowestBreakIteration, !(infos.IsStopped || infos.IsExceptional));
411 public static ParallelLoopResult ForEach<TSource> (IEnumerable<TSource> source, Action<TSource> body)
414 throw new ArgumentNullException (
"source");
416 throw new ArgumentNullException (
"body");
418 return ForEach<TSource, object> (
Partitioner.Create (source),
421 (e, s, l) => { body (e);
return null; },
425 public static ParallelLoopResult ForEach<TSource> (IEnumerable<TSource> source, Action<TSource, ParallelLoopState> body)
428 throw new ArgumentNullException (
"source");
430 throw new ArgumentNullException (
"body");
432 return ForEach<TSource, object> (
Partitioner.Create (source),
435 (e, s, l) => { body (e, s);
return null; },
440 Action<TSource, ParallelLoopState, long> body)
443 throw new ArgumentNullException (
"source");
445 throw new ArgumentNullException (
"body");
448 return ForEach<TSource, object> (
Partitioner.Create (source),
451 (e, s, l) => { body (e, s, -1);
return null; },
456 Action<TSource, ParallelLoopState> body)
459 throw new ArgumentNullException (
"body");
461 return ForEach<TSource, object> (source,
464 (e, s, l) => { body (e, s);
return null; },
469 Action<TSource, ParallelLoopState, long> body)
473 throw new ArgumentNullException (
"body");
475 return ForEach<TSource, object> (source,
478 (e, s, i, l) => { body (e, s, i);
return null; },
483 Action<TSource> body)
487 throw new ArgumentNullException (
"body");
489 return ForEach<TSource, object> (source,
492 (e, s, l) => { body (e);
return null; },
498 Action<TSource> body)
501 throw new ArgumentNullException (
"source");
503 throw new ArgumentNullException (
"body");
505 return ForEach<TSource, object> (
Partitioner.Create (source),
508 (e, s, l) => { body (e);
return null; },
513 Action<TSource, ParallelLoopState> body)
516 throw new ArgumentNullException (
"source");
518 throw new ArgumentNullException (
"body");
520 return ForEach<TSource, object> (
Partitioner.Create (source),
523 (e, s, l) => { body (e, s);
return null; },
528 Action<TSource, ParallelLoopState, long> body)
531 throw new ArgumentNullException (
"source");
533 throw new ArgumentNullException (
"body");
535 return ForEach<TSource, object> (
Partitioner.Create (source),
538 (e, s, i, l) => { body (e, s, i);
return null; },
543 Action<TSource, ParallelLoopState, long> body)
547 throw new ArgumentNullException (
"body");
549 return ForEach<TSource, object> (source,
552 (e, s, i, l) => { body (e, s, i);
return null; },
557 Action<TSource> body)
560 throw new ArgumentNullException (
"body");
562 return ForEach<TSource, object> (source,
565 (e, s, l) => { body (e);
return null; },
570 Action<TSource, ParallelLoopState> body)
572 return ForEach<TSource, object> (source,
575 (e, s, l) => { body (e, s);
return null; },
579 public static ParallelLoopResult ForEach<TSource, TLocal> (IEnumerable<TSource> source, Func<TLocal> localInit,
580 Func<TSource, ParallelLoopState, TLocal, TLocal> body,
581 Action<TLocal> localFinally)
584 throw new ArgumentNullException (
"source");
593 public static ParallelLoopResult ForEach<TSource, TLocal> (IEnumerable<TSource> source, Func<TLocal> localInit,
594 Func<TSource, ParallelLoopState, long, TLocal, TLocal> body,
595 Action<TLocal> localFinally)
597 return ForEach<TSource, TLocal> (
Partitioner.Create (source),
605 Func<TSource, ParallelLoopState, long, TLocal, TLocal> body,
606 Action<TLocal> localFinally)
608 return ForEach<TSource, TLocal> (source,
ParallelOptions.Default, localInit, body, localFinally);
612 Func<TSource, ParallelLoopState, TLocal, TLocal> body,
613 Action<TLocal> localFinally)
615 return ForEach<TSource, TLocal> (source,
ParallelOptions.Default, localInit, body, localFinally);
619 Func<TLocal> localInit,
620 Func<TSource, ParallelLoopState, TLocal, TLocal> body,
621 Action<TLocal> localFinally)
624 throw new ArgumentNullException (
"source");
626 return ForEach<TSource, TLocal> (
Partitioner.Create (source), parallelOptions, localInit, body, localFinally);
630 Func<TLocal> localInit,
631 Func<TSource, ParallelLoopState, long, TLocal, TLocal> body,
632 Action<TLocal> localFinally)
635 throw new ArgumentNullException (
"source");
637 return ForEach<TSource, TLocal> (
Partitioner.Create (source), parallelOptions, localInit, body, localFinally);
641 Func<TLocal> localInit,
642 Func<TSource, ParallelLoopState, TLocal, TLocal> body,
643 Action<TLocal> localFinally)
646 throw new ArgumentNullException (
"source");
648 throw new ArgumentNullException (
"body");
650 return ForEach<TSource, TLocal> (source.GetPartitions, parallelOptions, localInit, body, localFinally);
654 Func<TLocal> localInit,
655 Func<TSource, ParallelLoopState, long, TLocal, TLocal> body,
656 Action<TLocal> localFinally)
659 throw new ArgumentNullException (
"source");
661 throw new ArgumentNullException (
"body");
663 return ForEach<KeyValuePair<long, TSource>, TLocal> (source.GetOrderablePartitions,
666 (e, s, l) => body (e.Value, s, e.Key, l),
672 public static void Invoke (params Action[] actions)
675 throw new ArgumentNullException (
"actions");
682 if (parallelOptions == null)
683 throw new ArgumentNullException (
"parallelOptions");
685 throw new ArgumentNullException (
"actions");
686 if (actions.Length == 0)
687 throw new ArgumentException (
"actions is empty");
688 foreach (var a
in actions)
690 throw new ArgumentException (
"One action in actions is null",
"actions");
691 if (actions.Length == 1) {
696 Task[] ts =
new Task[actions.Length];
697 for (
int i = 0; i < ts.Length; i++)
698 ts[i] =
Task.Factory.StartNew (actions[i],
699 parallelOptions.CancellationToken,
701 parallelOptions.TaskScheduler);
704 Task.WaitAll (ts, parallelOptions.CancellationToken);
706 HandleExceptions (ts);
711 #region SpawnBestNumber, used by PLinq 712 internal static Task[] SpawnBestNumber (Action action, Action callback)
714 return SpawnBestNumber (action, -1, callback);
717 internal static Task[] SpawnBestNumber (Action action,
int dop, Action callback)
719 return SpawnBestNumber (action, dop,
false, callback);
722 internal static Task[] SpawnBestNumber (Action action,
int dop,
bool wait, Action callback)
725 int num = dop == -1 ? (wait ? GetBestWorkerNumber () + 1 : GetBestWorkerNumber ()) : dop;
730 for (
int i = 0; i < num; i++) {
731 tasks [i] =
Task.Factory.StartNew (() => {
734 if (callback != null && evt.
IsSet)
742 Task.WaitAll (tasks);
static ParallelLoopResult For(long fromInclusive, long toExclusive, ParallelOptions parallelOptions, Action< long > body)
static TaskScheduler Current
void ThrowIfCancellationRequested()
static ParallelLoopResult For(long fromInclusive, long toExclusive, ParallelOptions parallelOptions, Action< long, ParallelLoopState > body)
static void Invoke(ParallelOptions parallelOptions, params Action[] actions)
int MaxDegreeOfParallelism
TaskScheduler TaskScheduler
static ParallelLoopResult For(int fromInclusive, int toExclusive, Action< int, ParallelLoopState > body)
static ParallelLoopResult For(long fromInclusive, long toExclusive, Action< long, ParallelLoopState > body)
static ParallelLoopResult For(int fromInclusive, int toExclusive, Action< int > body)
CancellationToken(bool canceled)
static ParallelLoopResult For(long fromInclusive, long toExclusive, Action< long > body)
static void Invoke(params Action[] actions)
static ParallelLoopResult For(int fromInclusive, int toExclusive, ParallelOptions parallelOptions, Action< int > body)
AggregateException Flatten()
static ParallelLoopResult For(int fromInclusive, int toExclusive, ParallelOptions parallelOptions, Action< int, ParallelLoopState > body)
CancellationToken CancellationToken