// // Math.NET Numerics, part of the Math.NET Project // http://numerics.mathdotnet.com // http://github.com/mathnet/mathnet-numerics // http://mathnetnumerics.codeplex.com // // Copyright (c) 2009-2013 Math.NET // // Permission is hereby granted, free of charge, to any person // obtaining a copy of this software and associated documentation // files (the "Software"), to deal in the Software without // restriction, including without limitation the rights to use, // copy, modify, merge, publish, distribute, sublicense, and/or sell // copies of the Software, and to permit persons to whom the // Software is furnished to do so, subject to the following // conditions: // // The above copyright notice and this permission notice shall be // included in all copies or substantial portions of the Software. // // THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, // EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES // OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND // NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT // HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, // WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING // FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR // OTHER DEALINGS IN THE SOFTWARE. // namespace MathNet.Numerics.Threading { using System; using System.Threading.Tasks; #if (PORTABLE || NET35) using System.Linq; using Properties; #else using System.Collections.Concurrent; using System.Collections.Generic; #endif /// /// Used to simplify parallel code, particularly between the .NET 4.0 and Silverlight Code. /// public static class CommonParallel { /// /// Executes a for loop in which iterations may run in parallel. /// /// The start index, inclusive. /// The end index, exclusive. /// The body to be invoked for each iteration range. public static void For(int fromInclusive, int toExclusive, Action body) { For(fromInclusive, toExclusive, Math.Max(1, (toExclusive - fromInclusive)/Control.NumberOfParallelWorkerThreads), body); } /// /// Executes a for loop in which iterations may run in parallel. /// /// The start index, inclusive. /// The end index, exclusive. /// The partition size for splitting work into smaller pieces. /// The body to be invoked for each iteration range. public static void For(int fromInclusive, int toExclusive, int rangeSize, Action body) { if (body == null) throw new ArgumentNullException("body"); if (fromInclusive < 0) throw new ArgumentOutOfRangeException("fromInclusive"); if (fromInclusive > toExclusive) throw new ArgumentOutOfRangeException("toExclusive"); if (rangeSize < 1) throw new ArgumentOutOfRangeException("rangeSize"); var length = toExclusive - fromInclusive; // Special case: nothing to do if (length <= 0) { return; } var maxDegreeOfParallelism = Control.NumberOfParallelWorkerThreads; // Special case: not worth to parallelize, inline if (Control.DisableParallelization || maxDegreeOfParallelism < 2 || (rangeSize*2) > length) { body(fromInclusive, toExclusive); return; } #if (PORTABLE || NET35) var tasks = new Task[Math.Min(maxDegreeOfParallelism, length/rangeSize)]; rangeSize = (toExclusive - fromInclusive)/tasks.Length; // partition the jobs into separate sets for each but the last worked thread for (var i = 0; i < tasks.Length - 1; i++) { var start = fromInclusive + (i*rangeSize); var stop = fromInclusive + ((i + 1)*rangeSize); tasks[i] = Task.Factory.StartNew(() => body(start, stop)); } // add another set for last worker thread tasks[tasks.Length - 1] = Task.Factory.StartNew(() => body(fromInclusive + ((tasks.Length - 1)*rangeSize), toExclusive)); Task.WaitAll(tasks); #else Parallel.ForEach( Partitioner.Create(fromInclusive, toExclusive, rangeSize), new ParallelOptions {MaxDegreeOfParallelism = maxDegreeOfParallelism}, (range, loopState) => body(range.Item1, range.Item2)); #endif } /// /// Executes each of the provided actions inside a discrete, asynchronous task. /// /// An array of actions to execute. /// The actions array contains a null element. /// At least one invocation of the actions threw an exception. public static void Invoke(params Action[] actions) { // Special case: no action if (actions.Length == 0) { return; } // Special case: single action, inline if (actions.Length == 1) { actions[0](); return; } // Special case: straight execution without parallelism if (Control.DisableParallelization || Control.NumberOfParallelWorkerThreads < 2) { for (int i = 0; i < actions.Length; i++) { actions[i](); } return; } // Common case #if (PORTABLE || NET35) var tasks = new Task[actions.Length]; for (var i = 0; i < tasks.Length; i++) { Action action = actions[i]; if (action == null) { throw new ArgumentException(String.Format(Resources.ArgumentItemNull, "actions"), "actions"); } tasks[i] = Task.Factory.StartNew(action); } Task.WaitAll(tasks); #else Parallel.Invoke( new ParallelOptions { MaxDegreeOfParallelism = Control.NumberOfParallelWorkerThreads }, actions); #endif } /// /// Selects an item (such as Max or Min). /// /// Starting index of the loop. /// Ending index of the loop /// The function to select items over a subset. /// The function to select the item of selection from the subsets. /// The selected value. public static T Aggregate(int fromInclusive, int toExclusive, Func select, Func reduce) { if (select == null) { throw new ArgumentNullException("select"); } if (reduce == null) { throw new ArgumentNullException("reduce"); } // Special case: no action if (fromInclusive >= toExclusive) { return reduce(new T[0]); } // Special case: single action, inline if (fromInclusive == (toExclusive - 1)) { return reduce(new[] {select(fromInclusive)}); } // Special case: straight execution without parallelism if (Control.DisableParallelization || Control.NumberOfParallelWorkerThreads < 2) { var mapped = new T[toExclusive - fromInclusive]; for (int k = 0; k < mapped.Length; k++) { mapped[k] = select(k + fromInclusive); } return reduce(mapped); } #if (PORTABLE || NET35) var tasks = new Task[Control.NumberOfParallelWorkerThreads]; var size = (toExclusive - fromInclusive) / tasks.Length; // partition the jobs into separate sets for each but the last worked thread for (var i = 0; i < tasks.Length - 1; i++) { var start = fromInclusive + (i * size); var stop = fromInclusive + ((i + 1) * size); tasks[i] = Task.Factory.StartNew(() => { var mapped = new T[stop - start]; for (int k = 0; k < mapped.Length; k++) { mapped[k] = select(k + start); } return reduce(mapped); }); } // add another set for last worker thread tasks[tasks.Length - 1] = Task.Factory.StartNew(() => { var start = fromInclusive + ((tasks.Length - 1) * size); var mapped = new T[toExclusive - start]; for (int k = 0; k < mapped.Length; k++) { mapped[k] = select(k + start); } return reduce(mapped); }); return Task.Factory .ContinueWhenAll(tasks, tsk => reduce(tsk.Select(t => t.Result).ToArray())) .Result; #else var intermediateResults = new List(); var syncLock = new object(); var maxThreads = Control.DisableParallelization ? 1 : Control.NumberOfParallelWorkerThreads; Parallel.ForEach( Partitioner.Create(fromInclusive, toExclusive), new ParallelOptions {MaxDegreeOfParallelism = maxThreads}, () => new List(), (range, loop, localData) => { var mapped = new T[range.Item2 - range.Item1]; for (int k = 0; k < mapped.Length; k++) { mapped[k] = select(k + range.Item1); } localData.Add(reduce(mapped)); return localData; }, localResult => { lock (syncLock) { intermediateResults.Add(reduce(localResult.ToArray())); } }); return reduce(intermediateResults.ToArray()); #endif } /// /// Selects an item (such as Max or Min). /// /// The array to iterate over. /// The function to select items over a subset. /// The function to select the item of selection from the subsets. /// The selected value. public static TOut Aggregate(T[] array, Func select, Func reduce) { if (select == null) { throw new ArgumentNullException("select"); } if (reduce == null) { throw new ArgumentNullException("reduce"); } // Special case: no action if (array == null || array.Length == 0) { return reduce(new TOut[0]); } // Special case: single action, inline if (array.Length == 1) { return reduce(new[] {select(0, array[0])}); } // Special case: straight execution without parallelism if (Control.DisableParallelization || Control.NumberOfParallelWorkerThreads < 2) { var mapped = new TOut[array.Length]; for (int k = 0; k < mapped.Length; k++) { mapped[k] = select(k, array[k]); } return reduce(mapped); } #if (PORTABLE || NET35) var tasks = new Task[Control.NumberOfParallelWorkerThreads]; var size = array.Length / tasks.Length; // partition the jobs into separate sets for each but the last worked thread for (var i = 0; i < tasks.Length - 1; i++) { var start = (i * size); var stop = ((i + 1) * size); tasks[i] = Task.Factory.StartNew(() => { var mapped = new TOut[stop - start]; for (int k = 0; k < mapped.Length; k++) { mapped[k] = select(k + start, array[k + start]); } return reduce(mapped); }); } // add another set for last worker thread tasks[tasks.Length - 1] = Task.Factory.StartNew(() => { var start = ((tasks.Length - 1) * size); var mapped = new TOut[array.Length - start]; for (int k = 0; k < mapped.Length; k++) { mapped[k] = select(k + start, array[k + start]); } return reduce(mapped); }); return Task.Factory .ContinueWhenAll(tasks, tsk => reduce(tsk.Select(t => t.Result).ToArray())) .Result; #else var intermediateResults = new List(); var syncLock = new object(); var maxThreads = Control.DisableParallelization ? 1 : Control.NumberOfParallelWorkerThreads; Parallel.ForEach( Partitioner.Create(0, array.Length), new ParallelOptions {MaxDegreeOfParallelism = maxThreads}, () => new List(), (range, loop, localData) => { var mapped = new TOut[range.Item2 - range.Item1]; for (int k = 0; k < mapped.Length; k++) { mapped[k] = select(k + range.Item1, array[k + range.Item1]); } localData.Add(reduce(mapped)); return localData; }, localResult => { lock (syncLock) { intermediateResults.Add(reduce(localResult.ToArray())); } }); return reduce(intermediateResults.ToArray()); #endif } /// /// Selects an item (such as Max or Min). /// /// Starting index of the loop. /// Ending index of the loop /// The function to select items over a subset. /// The function to select the item of selection from the subsets. /// Default result of the reduce function on an empty set. /// The selected value. public static T Aggregate(int fromInclusive, int toExclusive, Func select, Func reducePair, T reduceDefault) { return Aggregate(fromInclusive, toExclusive, select, results => { if (results == null || results.Length == 0) { return reduceDefault; } if (results.Length == 1) { return results[0]; } T result = results[0]; for (int i = 1; i < results.Length; i++) { result = reducePair(result, results[i]); } return result; }); } /// /// Selects an item (such as Max or Min). /// /// The array to iterate over. /// The function to select items over a subset. /// The function to select the item of selection from the subsets. /// Default result of the reduce function on an empty set. /// The selected value. public static TOut Aggregate(T[] array, Func select, Func reducePair, TOut reduceDefault) { return Aggregate(array, select, results => { if (results == null || results.Length == 0) { return reduceDefault; } if (results.Length == 1) { return results[0]; } TOut result = results[0]; for (int i = 1; i < results.Length; i++) { result = reducePair(result, results[i]); } return result; }); } } }