threading: minor refactoring, control, fast-forward execution

Signed-off-by: Christoph Ruegg <git@cdrnet.ch>

src/Numerics/Control.cs

@@ -28,12 +28,12 @@
 
 namespace MathNet.Numerics
 {
-    using System;
+    using Threading;
 
     /// <summary>
     /// Sets parameters for the library.
     /// </summary>
-    public static partial class Control
+    public static class Control
     {
         /// <summary>
         /// Initializes static members of the Control class.
@@ -42,6 +42,7 @@ namespace MathNet.Numerics
         {
             CheckDistributionParameters = true;
             ThreadSafeRandomNumberGenerators = true;
+            DisableParallelization = false;
         }
 
         /// <summary>
@@ -59,5 +60,20 @@ namespace MathNet.Numerics
         ///     <c>true</c> to use thread safe random number generators ; otherwise, <c>false</c>.
         /// </value>
         public static bool ThreadSafeRandomNumberGenerators { get; set; }
+
+        /// <summary>
+        /// Gets or sets a value indicating how many parallel worker threads shall be used
+        /// when parallelization is applicable.
+        /// </summary>
+        public static int NumberOfParallelWorkerThreads
+        {
+            get { return ThreadQueue.ThreadCount; }
+            set { ThreadQueue.Start(value); }
+        }
+
+        /// <summary>
+        /// Gets or sets a value indicating whether parallelization shall be disabled globally.
+        /// </summary>
+        public static bool DisableParallelization { get; set; }
     }
-}
+}

src/Numerics/IntegralTransforms/Algorithms/DiscreteFourierTransform.Bluestein.cs

@@ -70,7 +70,7 @@ namespace MathNet.Numerics.IntegralTransforms.Algorithms
             Complex[] b = new Complex[m];
             Complex[] a = new Complex[m];
 
-            Parallel.Invoke(
+            Parallel.Run(
                 () =>
                 {
                     // Build and transform padded sequence b_k = exp(I*Pi*k^2/N)

src/Numerics/Threading/AggregateException.cs

@@ -63,4 +63,4 @@ namespace MathNet.Numerics.Threading
             get { return new ReadOnlyCollection<Exception>(_exceptions); }
         }
     }
-}
+}

src/Numerics/Threading/Parallel.cs

@@ -53,15 +53,33 @@ namespace MathNet.Numerics.Threading
                 throw new ArgumentNullException("body");
             }
 
-            var actions = new Action[ThreadQueue.ThreadCount];
+            // fast forward execution if it's only one or none items
             var count = toExclusive - fromInclusive;
-            var size = count / actions.Length;
+            if (count <= 1)
+            {
+                if (count == 1)
+                {
+                    body(fromInclusive);
+                }
 
-            if (count < 1)
+                return;
+            }
+
+            // fast forward execution in case parallelization is disabled
+            // (cdrnet, 200908): should we fast forward on STA threads as well?
+            if (Control.DisableParallelization || ThreadQueue.ThreadCount <= 1)
             {
+                for (int i = fromInclusive; i < toExclusive; i++)
+                {
+                    body(i);
+                }
+
                 return;
             }
 
+            var actions = new Action[ThreadQueue.ThreadCount];
+            var size = count / actions.Length;
+
             // partition the jobs into separate sets for each but the last worked thread
             for (var i = 0; i < actions.Length - 1; i++)
             {
@@ -71,7 +89,7 @@ namespace MathNet.Numerics.Threading
                 actions[i] =
                     () =>
                     {
-                        for (var j = start; j < stop; j++)
+                        for (int j = start; j < stop; j++)
                         {
                             body(j);
                         }
@@ -82,7 +100,7 @@ namespace MathNet.Numerics.Threading
             actions[actions.Length - 1] =
                 () =>
                 {
-                    for (var i = fromInclusive + ((actions.Length - 1) * size); i < toExclusive; i++)
+                    for (int i = fromInclusive + ((actions.Length - 1) * size); i < toExclusive; i++)
                     {
                         body(i);
                     }
@@ -98,13 +116,47 @@ 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 Invoke(params Action[] actions)
+        internal static void Run(params Action[] actions)
         {
             if (actions == null)
             {
                 throw new ArgumentNullException("actions");
             }
 
+            // fast forward execution if it's only one or none items
+            if (actions.Length <= 1)
+            {
+                if (actions.Length == 1)
+                {
+                    actions[0]();
+                }
+
+                return;
+            }
+
+            // fast forward execution in case parallelization is disabled
+            // (cdrnet, 200908): should we fast forward on STA threads as well?
+            if (Control.DisableParallelization || ThreadQueue.ThreadCount <= 1)
+            {
+                for (int i = 0; i < actions.Length; i++)
+                {
+                    actions[i]();
+                }
+
+                return;
+            }
+
+            Invoke(actions);
+        }
+
+        /// <summary>
+        /// Executes each of the provided actions inside a discrete, asynchronous task. 
+        /// </summary>
+        /// <param name="actions">An array of actions to execute.</param>
+        /// <exception cref="ArgumentException">The actions array contains a null element.</exception>
+        /// <exception cref="AggregateException">An action threw an exception.</exception>
+        private static void Invoke(params Action[] actions)
+        {
             // create a job for each action
             var tasks = new Task[actions.Length];
             for (int i = 0; i < tasks.Length; i++)
@@ -155,4 +207,4 @@ namespace MathNet.Numerics.Threading
             }
         }
     }
-}
+}

src/Numerics/Threading/Task.cs

@@ -84,4 +84,4 @@ namespace MathNet.Numerics.Threading
             }
         }
     }
-}
+}

src/Numerics/Threading/ThreadQueue.cs

@@ -82,7 +82,6 @@ namespace MathNet.Numerics.Threading
         /// </summary>
         static ThreadQueue()
         {
-            // TODO: Control.ThreadCount instead of Environment.ProcessorCount
             Start(Environment.ProcessorCount);
         }
 
@@ -154,12 +153,14 @@ namespace MathNet.Numerics.Threading
                     }
                 }
 
-                // ...and run it
+                if (task == null)
-                if (task != null)
                 {
-                    task.Compute();
+                    continue;
-                    task.Set();
                 }
+
+                // ...and run it
+                task.Compute();
+                task.Set();
             }
         }
 
@@ -169,6 +170,9 @@ namespace MathNet.Numerics.Threading
         /// <param name="numberOfThreads">Number of worker threads.</param>
         internal static void Start(int numberOfThreads)
         {
+            // instead of throwing an out of range exception, simply normalize
+            numberOfThreads = Math.Max(1, Math.Min(1024, numberOfThreads));
+
             lock (_stateSync)
             {
                 if (_threads != null)
@@ -242,4 +246,4 @@ namespace MathNet.Numerics.Threading
             }
         }
     }
-}
+}

src/UnitTests/ThreadingTests/ParallelTest.cs

@@ -117,11 +117,12 @@ namespace MathNet.Numerics.UnitTests.ThreadingTests
             ThreadQueue.Start(2);
             Assert.AreEqual(2, ThreadQueue.ThreadCount);
 
-            ThreadQueue.Start(2);
+            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();