From c412fce9e61f6a8f15b3aa9f3d2dc0820ecf5b47 Mon Sep 17 00:00:00 2001 From: Christoph Ruegg Date: Thu, 13 Aug 2009 16:29:58 +0800 Subject: [PATCH] threading: minor refactoring, control, fast-forward execution Signed-off-by: Christoph Ruegg --- src/Numerics/Control.cs | 22 ++++++- .../DiscreteFourierTransform.Bluestein.cs | 2 +- src/Numerics/Threading/AggregateException.cs | 2 +- src/Numerics/Threading/Parallel.cs | 66 +++++++++++++++++-- src/Numerics/Threading/Task.cs | 2 +- src/Numerics/Threading/ThreadQueue.cs | 16 +++-- src/UnitTests/ThreadingTests/ParallelTest.cs | 3 +- 7 files changed, 93 insertions(+), 20 deletions(-) diff --git a/src/Numerics/Control.cs b/src/Numerics/Control.cs index 238281a5..89418049 100644 --- a/src/Numerics/Control.cs +++ b/src/Numerics/Control.cs @@ -28,12 +28,12 @@ namespace MathNet.Numerics { - using System; + using Threading; /// /// Sets parameters for the library. /// - public static partial class Control + public static class Control { /// /// Initializes static members of the Control class. @@ -42,6 +42,7 @@ namespace MathNet.Numerics { CheckDistributionParameters = true; ThreadSafeRandomNumberGenerators = true; + DisableParallelization = false; } /// @@ -59,5 +60,20 @@ namespace MathNet.Numerics /// true to use thread safe random number generators ; otherwise, false. /// public static bool ThreadSafeRandomNumberGenerators { get; set; } + + /// + /// Gets or sets a value indicating how many parallel worker threads shall be used + /// when parallelization is applicable. + /// + public static int NumberOfParallelWorkerThreads + { + get { return ThreadQueue.ThreadCount; } + set { ThreadQueue.Start(value); } + } + + /// + /// Gets or sets a value indicating whether parallelization shall be disabled globally. + /// + public static bool DisableParallelization { get; set; } } -} \ No newline at end of file +} diff --git a/src/Numerics/IntegralTransforms/Algorithms/DiscreteFourierTransform.Bluestein.cs b/src/Numerics/IntegralTransforms/Algorithms/DiscreteFourierTransform.Bluestein.cs index fcc8e27f..1b790523 100644 --- a/src/Numerics/IntegralTransforms/Algorithms/DiscreteFourierTransform.Bluestein.cs +++ b/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) diff --git a/src/Numerics/Threading/AggregateException.cs b/src/Numerics/Threading/AggregateException.cs index fe77a488..a0e2d22f 100644 --- a/src/Numerics/Threading/AggregateException.cs +++ b/src/Numerics/Threading/AggregateException.cs @@ -63,4 +63,4 @@ namespace MathNet.Numerics.Threading get { return new ReadOnlyCollection(_exceptions); } } } -} \ No newline at end of file +} diff --git a/src/Numerics/Threading/Parallel.cs b/src/Numerics/Threading/Parallel.cs index 378ade6d..dbab88c8 100644 --- a/src/Numerics/Threading/Parallel.cs +++ b/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 /// The argument is null. /// The actions array contains a null element. /// An action threw an 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); + } + + /// + /// Executes each of the provided actions inside a discrete, asynchronous task. + /// + /// An array of actions to execute. + /// The actions array contains a null element. + /// An action threw an 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 } } } -} \ No newline at end of file +} diff --git a/src/Numerics/Threading/Task.cs b/src/Numerics/Threading/Task.cs index 74a408ce..544a1cb3 100644 --- a/src/Numerics/Threading/Task.cs +++ b/src/Numerics/Threading/Task.cs @@ -84,4 +84,4 @@ namespace MathNet.Numerics.Threading } } } -} \ No newline at end of file +} diff --git a/src/Numerics/Threading/ThreadQueue.cs b/src/Numerics/Threading/ThreadQueue.cs index 8e48f6de..cfcdd780 100644 --- a/src/Numerics/Threading/ThreadQueue.cs +++ b/src/Numerics/Threading/ThreadQueue.cs @@ -82,7 +82,6 @@ namespace MathNet.Numerics.Threading /// 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(); - task.Set(); + continue; } + + // ...and run it + task.Compute(); + task.Set(); } } @@ -169,6 +170,9 @@ namespace MathNet.Numerics.Threading /// Number of worker threads. 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 } } } -} \ No newline at end of file +} diff --git a/src/UnitTests/ThreadingTests/ParallelTest.cs b/src/UnitTests/ThreadingTests/ParallelTest.cs index b220594f..2e114dff 100644 --- a/src/UnitTests/ThreadingTests/ParallelTest.cs +++ b/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();