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 @@
-
+
+