Parallel: First take on Parallel.ForEach

Signed-off-by: Marcus Cuda <marcus@cuda.net>

src/Numerics/Control.cs

@@ -43,6 +43,9 @@ namespace MathNet.Numerics
             CheckDistributionParameters = true;
             ThreadSafeRandomNumberGenerators = true;
             DisableParallelization = false;
+            InitialThreadBlockSize = 2;
+            BlockScalingFactor = 2;
+            MaximumBlockSize = 1024;
         }
 
         /// <summary>
@@ -75,5 +78,28 @@ namespace MathNet.Numerics
         /// Gets or sets a value indicating whether parallelization shall be disabled globally.
         /// </summary>
         public static bool DisableParallelization { get; set; }
+
+        /// <summary>
+        /// Gets or sets the initial size of a <see cref="Parallel.ForEach{T}"/>
+        /// processing block (the number of elements the first thread should process).
+        /// </summary>
+        /// <value>The initial size of the thread processing bloc.</value>
+        public static int InitialThreadBlockSize { get; set; }
+
+        /// <summary>
+        /// Gets or sets the <see cref="Parallel.ForEach{T}"/>
+        /// processing block scaling factor. With each iteration through 
+        /// the for each loop, the processing block increased by this factor
+        /// up to <see cref="MaximumBlockSize"/>;
+        /// </summary>
+        /// <value>The processing block scaling factor.</value>
+        public static int BlockScalingFactor { get; set; }
+
+        /// <summary>
+        /// Gets or sets the maximum processing block size for
+        /// <see cref="Parallel.ForEach{T}"/>.
+        /// </summary>
+        /// <value>The maximum processing block size.</value>
+        public static int MaximumBlockSize { get; set; }
     }
 }

src/Numerics/LinearAlgebra/Double/Vector.cs

@@ -874,8 +874,6 @@ namespace MathNet.Numerics.LinearAlgebra.Double
         /// <remarks>
         /// For sparse vectors, <see cref="GetIndexedEnumerator()"/> will perform better.
         /// </remarks>
-        [SuppressMessage("Microsoft.Design", "CA1006:DoNotNestGenericTypesInMemberSignatures", 
-            Justification = "Needed to support sparse vectors.")]
         public virtual IEnumerator<double> GetEnumerator()
         {
             for (var index = 0; index < Count; index++)

src/Numerics/Threading/Parallel.cs

@@ -46,7 +46,7 @@ namespace MathNet.Numerics.Threading
         /// <param name="body">The body to be invoked for each iteration.</param>
         /// <exception cref="ArgumentNullException">The <paramref name="body"/> argument is null.</exception>
         /// <exception cref="AggregateException">At least one invocation of the body threw an exception.</exception>
-        internal static void For(int fromInclusive, int toExclusive, Action<int> body)
+        public static void For(int fromInclusive, int toExclusive, Action<int> body)
         {
             if (body == null)
             {
@@ -110,6 +110,75 @@ namespace MathNet.Numerics.Threading
             Invoke(actions);
         }
 
+        /// <summary>
+        /// Executes a for each operation on an IEnumerable{T} in which iterations may run in parallel.
+        /// </summary>
+        /// <typeparam name="T">The type of the data in the source.</typeparam>
+        /// <param name="source">An enumerable data source.</param>
+        /// <param name="body">The delegate that is invoked once per iteration.</param>
+        public static void ForEach<T>(IEnumerable<T> source, Action<T> body)
+        {
+            if (body == null)
+            {
+                throw new ArgumentNullException("body");
+            }
+
+            // source is a IList, call For instead.
+            if (source is IList<T>)
+            {
+                var list = (IList<T>)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<Task>();
+            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);
+        }
+
         /// <summary>
         /// Executes each of the provided actions inside a discrete, asynchronous task. 
         /// </summary>
@@ -117,7 +186,7 @@ namespace MathNet.Numerics.Threading
         /// <exception cref="ArgumentNullException">The <paramref name="actions"/> argument is null.</exception>
         /// <exception cref="ArgumentException">The actions array contains a null element.</exception>
         /// <exception cref="AggregateException">An action threw an exception.</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);
+        }
+
+        /// <summary>
+        /// Waits for tasks to complete.
+        /// </summary>
+        /// <param name="tasks">The tasks.</param>
+        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);
             }
+        }
 
+        /// <summary>
+        /// Collects the exceptions and dispose tasks.
+        /// </summary>
+        /// <param name="tasks">The tasks.</param>
+        private static void CollectExceptionsAndDisposeTasks(IEnumerable<Task> tasks)
+        {
             // collect all thrown exceptions and dispose the jobs
             var exceptions = new List<Exception>();
             foreach (var task in tasks)

src/UnitTests/ThreadingTests/ParallelForEachTests.cs

@@ -0,0 +1,228 @@
+// <copyright file="ParallelTest.cs" company="Math.NET">
+// 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.
+// </copyright>
+
+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<KeyValuePair<int, double>>();
+            for (var i = 0; i < items.Length; i++)
+            {
+                items[i] = i;
+                pairs.Add(new KeyValuePair<int, double>(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<KeyValuePair<int, double>>();
+            for (var i = 0; i < items.Length; i++)
+            {
+                items[i] = i;
+                pairs.Add(new KeyValuePair<int, double>(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<KeyValuePair<int, double>>();
+            for (var i = 0; i < items.Length; i++)
+            {
+                items[i] = i;
+                pairs.Add(new KeyValuePair<int, double>(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<KeyValuePair<int, double>>();
+            for (var i = 0; i < items.Length; i++)
+            {
+                items[i] = i;
+                pairs.Add(new KeyValuePair<int, double>(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<KeyValuePair<int, double>>();
+            for (var i = 0; i < items.Length; i++)
+            {
+                items[i] = i;
+                pairs.Add(new KeyValuePair<int, double>(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);
+        }
+    }
+}

src/UnitTests/ThreadingTests/ParallelTest.cs → 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)]

src/UnitTests/UnitTests.csproj

@@ -110,7 +110,8 @@
     <Compile Include="StatisticsTests\DescriptiveStatisticsTests.cs" />
     <Compile Include="StatisticsTests\StatisticsTests.cs" />
     <Compile Include="StatisticsTests\StatTestData.cs" />
-    <Compile Include="ThreadingTests\ParallelTest.cs" />
+    <Compile Include="ThreadingTests\ParallelForEachTests.cs" />
+    <Compile Include="ThreadingTests\ParallelForTests.cs" />
     <Compile Include="TrigonometryTest.cs" />
   </ItemGroup>
   <ItemGroup>