From 85dcab23f93d03146b29e09f3b2fc5500114ccf1 Mon Sep 17 00:00:00 2001 From: Marcus Cuda Date: Thu, 20 Aug 2009 22:18:51 +0800 Subject: [PATCH] Parallel: First take on Parallel.ForEach Signed-off-by: Marcus Cuda --- src/Numerics/Control.cs | 26 ++ src/Numerics/LinearAlgebra/Double/Vector.cs | 2 - src/Numerics/Threading/Parallel.cs | 95 +++++++- .../ThreadingTests/ParallelForEachTests.cs | 228 ++++++++++++++++++ .../{ParallelTest.cs => ParallelForTests.cs} | 2 +- src/UnitTests/UnitTests.csproj | 3 +- 6 files changed, 348 insertions(+), 8 deletions(-) create mode 100644 src/UnitTests/ThreadingTests/ParallelForEachTests.cs rename src/UnitTests/ThreadingTests/{ParallelTest.cs => ParallelForTests.cs} (99%) diff --git a/src/Numerics/Control.cs b/src/Numerics/Control.cs index 89418049..d217517c 100644 --- a/src/Numerics/Control.cs +++ b/src/Numerics/Control.cs @@ -43,6 +43,9 @@ namespace MathNet.Numerics CheckDistributionParameters = true; ThreadSafeRandomNumberGenerators = true; DisableParallelization = false; + InitialThreadBlockSize = 2; + BlockScalingFactor = 2; + MaximumBlockSize = 1024; } /// @@ -75,5 +78,28 @@ namespace MathNet.Numerics /// Gets or sets a value indicating whether parallelization shall be disabled globally. /// public static bool DisableParallelization { get; set; } + + /// + /// Gets or sets the initial size of a + /// processing block (the number of elements the first thread should process). + /// + /// The initial size of the thread processing bloc. + public static int InitialThreadBlockSize { get; set; } + + /// + /// Gets or sets the + /// processing block scaling factor. With each iteration through + /// the for each loop, the processing block increased by this factor + /// up to ; + /// + /// The processing block scaling factor. + public static int BlockScalingFactor { get; set; } + + /// + /// Gets or sets the maximum processing block size for + /// . + /// + /// The maximum processing block size. + public static int MaximumBlockSize { get; set; } } } diff --git a/src/Numerics/LinearAlgebra/Double/Vector.cs b/src/Numerics/LinearAlgebra/Double/Vector.cs index 522f4d8e..a3f87570 100644 --- a/src/Numerics/LinearAlgebra/Double/Vector.cs +++ b/src/Numerics/LinearAlgebra/Double/Vector.cs @@ -874,8 +874,6 @@ namespace MathNet.Numerics.LinearAlgebra.Double /// /// For sparse vectors, will perform better. /// - [SuppressMessage("Microsoft.Design", "CA1006:DoNotNestGenericTypesInMemberSignatures", - Justification = "Needed to support sparse vectors.")] public virtual IEnumerator GetEnumerator() { for (var index = 0; index < Count; index++) diff --git a/src/Numerics/Threading/Parallel.cs b/src/Numerics/Threading/Parallel.cs index 31ee7b14..83ed7d5e 100644 --- a/src/Numerics/Threading/Parallel.cs +++ b/src/Numerics/Threading/Parallel.cs @@ -46,7 +46,7 @@ namespace MathNet.Numerics.Threading /// The body to be invoked for each iteration. /// The argument is null. /// At least one invocation of the body threw an exception. - internal static void For(int fromInclusive, int toExclusive, Action body) + public static void For(int fromInclusive, int toExclusive, Action body) { if (body == null) { @@ -110,6 +110,75 @@ namespace MathNet.Numerics.Threading Invoke(actions); } + /// + /// Executes a for each operation on an IEnumerable{T} in which iterations may run in parallel. + /// + /// The type of the data in the source. + /// An enumerable data source. + /// The delegate that is invoked once per iteration. + public static void ForEach(IEnumerable source, Action body) + { + if (body == null) + { + throw new ArgumentNullException("body"); + } + + // source is a IList, call For instead. + if (source is IList) + { + var list = (IList)source; + For(0, list.Count, i => body(list[i])); + return; + } + + // fast forward execution in case parallelization is disabled + if (Control.DisableParallelization + || ThreadQueue.ThreadCount <= 1 + || ThreadQueue.IsInWorkerThread) + { + foreach (var item in source) + { + body(item); + } + + return; + } + + var enumerator = source.GetEnumerator(); + var maxBlockSize = Control.InitialThreadBlockSize; + var scalingFactor = Control.BlockScalingFactor; + var tasks = new List(); + while (enumerator.MoveNext()) + { + var pos = 0; + var list = new T[maxBlockSize]; + list[pos++] = enumerator.Current; + + var count = 1; + while (count < maxBlockSize && enumerator.MoveNext()) + { + list[pos++] = enumerator.Current; + count++; + } + + var task = new Task( + () => + { + for (var i = 0; i < pos; i++) + { + body(list[i]); + } + }); + + ThreadQueue.Enqueue(task); + + maxBlockSize = Math.Min(Control.MaximumBlockSize, maxBlockSize * scalingFactor); + } + + WaitForTasksToComplete(tasks.ToArray()); + CollectExceptionsAndDisposeTasks(tasks); + } + /// /// Executes each of the provided actions inside a discrete, asynchronous task. /// @@ -117,7 +186,7 @@ namespace MathNet.Numerics.Threading /// The argument is null. /// The actions array contains a null element. /// An action threw an exception. - internal static void Run(params Action[] actions) + public static void Run(params Action[] actions) { if (actions == null) { @@ -175,11 +244,22 @@ namespace MathNet.Numerics.Threading // run the jobs ThreadQueue.Enqueue(tasks); - // wait until all jobs have completed + WaitForTasksToComplete(tasks); + + CollectExceptionsAndDisposeTasks(tasks); + } + + /// + /// Waits for tasks to complete. + /// + /// The tasks. + private static void WaitForTasksToComplete(Task[] tasks) + { + // wait until all tasks have been completed if (Thread.CurrentThread.GetApartmentState() == ApartmentState.STA) { // not sure if this the best approach for STA - for (int i = 0; i < tasks.Length; i++) + for (var i = 0; i < tasks.Length; i++) { tasks[i].WaitOne(); } @@ -188,7 +268,14 @@ namespace MathNet.Numerics.Threading { WaitHandle.WaitAll(tasks); } + } + /// + /// Collects the exceptions and dispose tasks. + /// + /// The tasks. + private static void CollectExceptionsAndDisposeTasks(IEnumerable tasks) + { // collect all thrown exceptions and dispose the jobs var exceptions = new List(); foreach (var task in tasks) diff --git a/src/UnitTests/ThreadingTests/ParallelForEachTests.cs b/src/UnitTests/ThreadingTests/ParallelForEachTests.cs new file mode 100644 index 00000000..26c36a74 --- /dev/null +++ b/src/UnitTests/ThreadingTests/ParallelForEachTests.cs @@ -0,0 +1,228 @@ +// +// Math.NET Numerics, part of the Math.NET Project +// http://mathnet.opensourcedotnet.info +// +// Copyright (c) 2009 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.UnitTests.ThreadingTests +{ + using System; + using System.Collections.Generic; + using System.Threading; + using MbUnit.Framework; + using Threading; + + [TestFixture] + public class ParallelForEachTests + { + [Test, ApartmentState(ApartmentState.MTA)] + [Column( 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 100, 101)] + public void ParallelForEachInvokesEveryItemOnceMTAOnePerCore(int count) + { + var items = new double[count]; + var pairs = new List>(); + for (var i = 0; i < items.Length; i++) + { + items[i] = i; + pairs.Add(new KeyValuePair(i, i)); + } + + // ensure One-Per-Core + ThreadQueue.Start(Environment.ProcessorCount); + + Parallel.ForEach(pairs, + pair => + { + items[pair.Key] = pair.Value + 1000; + } + ); + + for (int i = 0; i < items.Length; i++) + { + Assert.AreEqual(1000+i, items[i], i.ToString()); + } + } + + [Test, ApartmentState(ApartmentState.STA)] + [Column(0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 100, 101)] + public void ParallelForEachInvokesEveryItemOnceSTAOnePerCore(int count) + { + var items = new double[count]; + var pairs = new List>(); + for (var i = 0; i < items.Length; i++) + { + items[i] = i; + pairs.Add(new KeyValuePair(i, i)); + } + + // ensure One-Per-Core + ThreadQueue.Start(Environment.ProcessorCount); + + Parallel.ForEach(pairs, + pair => + { + items[pair.Key] = pair.Value + 1000; + } + ); + + for (int i = 0; i < items.Length; i++) + { + Assert.AreEqual(1000 + i, items[i], i.ToString()); + } + } + + [Test, ApartmentState(ApartmentState.MTA)] + [Column(0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 100, 101)] + public void ParallelForEachInvokesEveryItemOnceMTATwoPerCore(int count) + { + var items = new double[count]; + var pairs = new List>(); + for (var i = 0; i < items.Length; i++) + { + items[i] = i; + pairs.Add(new KeyValuePair(i, i)); + } + + // ensure Two-Per-Core + ThreadQueue.Start(2 * Environment.ProcessorCount); + + Parallel.ForEach(pairs, + pair => + { + items[pair.Key] = pair.Value + 1000; + } + ); + + for (int i = 0; i < items.Length; i++) + { + Assert.AreEqual(1000 + i, items[i], i.ToString()); + } + } + + [Test, ApartmentState(ApartmentState.STA)] + [Column(0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 100, 101)] + public void ParallelForEachInvokesEveryItemOnceSTATwoPerCore(int count) + { + var items = new double[count]; + var pairs = new List>(); + for (var i = 0; i < items.Length; i++) + { + items[i] = i; + pairs.Add(new KeyValuePair(i, i)); + } + + // ensure Two-Per-Core + ThreadQueue.Start(2 * Environment.ProcessorCount); + + Parallel.ForEach(pairs, + pair => + { + items[pair.Key] = pair.Value + 1000; + } + ); + + for (int i = 0; i < items.Length; i++) + { + Assert.AreEqual(1000 + i, items[i], i.ToString()); + } + } + + [Test, ApartmentState(ApartmentState.MTA)] + public void DoesNotGetConfusedByMultipleStartShutdown() + { + ThreadQueue.Shutdown(); + ThreadQueue.Shutdown(); + + ThreadQueue.Start(2); + Assert.AreEqual(2, ThreadQueue.ThreadCount); + + Control.NumberOfParallelWorkerThreads = 2; + Assert.AreEqual(2, ThreadQueue.ThreadCount); + + ThreadQueue.Start(4); + Assert.AreEqual(4, ThreadQueue.ThreadCount); + Assert.AreEqual(4, Control.NumberOfParallelWorkerThreads); + + ThreadQueue.Shutdown(); + ThreadQueue.Start(); + Assert.AreEqual(4, ThreadQueue.ThreadCount); + + ThreadQueue.Start(2); + Assert.AreEqual(2, ThreadQueue.ThreadCount); + + var items = new double[50]; + var pairs = new List>(); + for (var i = 0; i < items.Length; i++) + { + items[i] = i; + pairs.Add(new KeyValuePair(i, i)); + } + + Parallel.ForEach(pairs, + pair => + { + items[pair.Key] = pair.Value + 1000; + } + ); + + for (int i = 0; i < items.Length; i++) + { + Assert.AreEqual(1000 + i, items[i], i.ToString()); + } + } + + [Test, ApartmentState(ApartmentState.MTA)] + public void DoesDetectAndResolveRecursiveParallelization() + { + var countSharedBetweenClosures = 0; + var values = new int[10]; + + Assert.DoesNotThrow( + () => + Parallel.ForEach(values, + j => Interlocked.Increment(ref countSharedBetweenClosures))); + + Assert.AreEqual(10, countSharedBetweenClosures); + countSharedBetweenClosures = 0; + + Parallel.ForEach(values, + i => + Parallel.ForEach(values, + j => Interlocked.Increment(ref countSharedBetweenClosures))); + + Assert.AreEqual(100, countSharedBetweenClosures); + countSharedBetweenClosures = 0; + + Parallel.ForEach(values, + i => + Parallel.ForEach(values, + j => + Parallel.ForEach(values, + k => Interlocked.Increment(ref countSharedBetweenClosures)))); + + Assert.AreEqual(1000, countSharedBetweenClosures); + } + } +} \ No newline at end of file diff --git a/src/UnitTests/ThreadingTests/ParallelTest.cs b/src/UnitTests/ThreadingTests/ParallelForTests.cs similarity index 99% rename from src/UnitTests/ThreadingTests/ParallelTest.cs rename to src/UnitTests/ThreadingTests/ParallelForTests.cs index cc05e732..35907aa2 100644 --- a/src/UnitTests/ThreadingTests/ParallelTest.cs +++ b/src/UnitTests/ThreadingTests/ParallelForTests.cs @@ -34,7 +34,7 @@ namespace MathNet.Numerics.UnitTests.ThreadingTests using Threading; [TestFixture] - public class ParallelTest + public class ParallelForTests { [Test, ApartmentState(ApartmentState.MTA)] [Column(0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 100, 101)] diff --git a/src/UnitTests/UnitTests.csproj b/src/UnitTests/UnitTests.csproj index 1ee5804e..82cdf056 100644 --- a/src/UnitTests/UnitTests.csproj +++ b/src/UnitTests/UnitTests.csproj @@ -110,7 +110,8 @@ - + +