diff --git a/src/Managed.UnitTests/Managed.UnitTests.csproj b/src/Managed.UnitTests/Managed.UnitTests.csproj index a1fbb68f..b82020d4 100644 --- a/src/Managed.UnitTests/Managed.UnitTests.csproj +++ b/src/Managed.UnitTests/Managed.UnitTests.csproj @@ -71,6 +71,7 @@ + diff --git a/src/Managed.UnitTests/ThreadingTests/ParallelTest.cs b/src/Managed.UnitTests/ThreadingTests/ParallelTest.cs new file mode 100644 index 00000000..b220594f --- /dev/null +++ b/src/Managed.UnitTests/ThreadingTests/ParallelTest.cs @@ -0,0 +1,146 @@ +// +// 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.Threading; + using MbUnit.Framework; + using Threading; + + [TestFixture] + public class ParallelTest + { + [Test, ApartmentState(ApartmentState.MTA)] + [Column(0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 100, 101)] + public void ParallelForInvokesEveryItemOnceMTAOnePerCore(int count) + { + var items = new int[count]; + + // ensure One-Per-Core + ThreadQueue.Start(Environment.ProcessorCount); + + Parallel.For(0, count, i => items[i]++); + Parallel.For(0, count, i => items[i] += 1000); + + for (int i = 0; i < items.Length; i++) + { + Assert.AreEqual(1001, items[i], i.ToString()); + } + } + + [Test, ApartmentState(ApartmentState.STA)] + [Column(0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 100, 101)] + public void ParallelForInvokesEveryItemOnceSTAOnePerCore(int count) + { + var items = new int[count]; + + // ensure One-Per-Core + ThreadQueue.Start(Environment.ProcessorCount); + + Parallel.For(0, count, i => items[i]++); + Parallel.For(0, count, i => items[i] += 1000); + + for (int i = 0; i < items.Length; i++) + { + Assert.AreEqual(1001, items[i], i.ToString()); + } + } + + [Test, ApartmentState(ApartmentState.MTA)] + [Column(0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 100, 101)] + public void ParallelForInvokesEveryItemOnceMTATwoPerCore(int count) + { + var items = new int[count]; + + // ensure Two-Per-Core + ThreadQueue.Start(2 * Environment.ProcessorCount); + + Parallel.For(0, count, i => items[i]++); + Parallel.For(0, count, i => items[i] += 1000); + + for (int i = 0; i < items.Length; i++) + { + Assert.AreEqual(1001, items[i], i.ToString()); + } + } + + [Test, ApartmentState(ApartmentState.STA)] + [Column(0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 100, 101)] + public void ParallelForInvokesEveryItemOnceSTATwoPerCore(int count) + { + var items = new int[count]; + + // ensure Two-Per-Core + ThreadQueue.Start(2 * Environment.ProcessorCount); + + Parallel.For(0, count, i => items[i]++); + Parallel.For(0, count, i => items[i] += 1000); + + for (int i = 0; i < items.Length; i++) + { + Assert.AreEqual(1001, items[i], i.ToString()); + } + } + + [Test, ApartmentState(ApartmentState.MTA)] + public void DoesNotGetConfusedByMultipleStartShutdown() + { + ThreadQueue.Shutdown(); + ThreadQueue.Shutdown(); + + ThreadQueue.Start(2); + Assert.AreEqual(2, ThreadQueue.ThreadCount); + + ThreadQueue.Start(2); + Assert.AreEqual(2, ThreadQueue.ThreadCount); + + ThreadQueue.Start(4); + Assert.AreEqual(4, ThreadQueue.ThreadCount); + + ThreadQueue.Shutdown(); + ThreadQueue.Start(); + Assert.AreEqual(4, ThreadQueue.ThreadCount); + + ThreadQueue.Start(2); + Assert.AreEqual(2, ThreadQueue.ThreadCount); + + var items = new int[50]; + + Parallel.For(0, items.Length, i => items[i]++); + Parallel.For(0, items.Length, i => items[i] += 1000); + + ThreadQueue.Shutdown(); + + for(int i = 0; i < items.Length; i++) + { + Assert.AreEqual(1001, items[i], i.ToString()); + } + } + } +} \ No newline at end of file diff --git a/src/Managed/Managed.csproj b/src/Managed/Managed.csproj index abb1e0f5..afbb4012 100644 --- a/src/Managed/Managed.csproj +++ b/src/Managed/Managed.csproj @@ -81,6 +81,10 @@ + + + + diff --git a/src/Managed/Properties/Resources.Designer.cs b/src/Managed/Properties/Resources.Designer.cs index 1e8efa03..4d133fa1 100644 --- a/src/Managed/Properties/Resources.Designer.cs +++ b/src/Managed/Properties/Resources.Designer.cs @@ -87,6 +87,15 @@ namespace MathNet.Numerics.Properties { } } + /// + /// Looks up a localized string similar to At least one item of {0} is a null reference (Nothing in Visual Basic).. + /// + internal static string ArgumentItemNull { + get { + return ResourceManager.GetString("ArgumentItemNull", resourceCulture); + } + } + /// /// Looks up a localized string similar to The matrix indices must not be out of range of the given matrix.. /// diff --git a/src/Managed/Properties/Resources.resx b/src/Managed/Properties/Resources.resx index 0eee5f89..70a56854 100644 --- a/src/Managed/Properties/Resources.resx +++ b/src/Managed/Properties/Resources.resx @@ -246,4 +246,7 @@ Value must be odd. + + At least one item of {0} is a null reference (Nothing in Visual Basic). + \ No newline at end of file diff --git a/src/Managed/Threading/AggregateException.cs b/src/Managed/Threading/AggregateException.cs new file mode 100644 index 00000000..fe77a488 --- /dev/null +++ b/src/Managed/Threading/AggregateException.cs @@ -0,0 +1,66 @@ +// +// 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.Threading +{ + using System; + using System.Collections.Generic; + using System.Collections.ObjectModel; + + /// + /// Represents multiple errors that occur during application execution. + /// + public class AggregateException : Exception + { + /// + /// List of the aggregated exceptions. + /// + private readonly IList _exceptions = new List(); + + /// + /// Initializes a new instance of the AggregateException class with a specified error message and references to the inner exceptions that are the cause of this exception. + /// + /// The exceptions that are the cause of the current exception. + public AggregateException(IEnumerable exceptions) + { + foreach (var exception in exceptions) + { + _exceptions.Add(exception); + } + } + + /// + /// Gets a read-only collection of the Exception instances that caused the current exception. + /// + /// A read-only collection of the Exception instances that caused the current exception + public ReadOnlyCollection InnerExceptions + { + get { return new ReadOnlyCollection(_exceptions); } + } + } +} \ No newline at end of file diff --git a/src/Managed/Threading/Parallel.cs b/src/Managed/Threading/Parallel.cs new file mode 100644 index 00000000..67ad5fff --- /dev/null +++ b/src/Managed/Threading/Parallel.cs @@ -0,0 +1,158 @@ +// +// 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.Threading +{ + using System; + using System.Collections.Generic; + using System.Threading; + using Properties; + + /// + /// Provides support for parallel loops. + /// + internal static class Parallel + { + /// + /// Executes a for loop in which iterations may run in parallel. + /// + /// The start index, inclusive. + /// The end index, exclusive. + /// 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) + { + if (body == null) + { + throw new ArgumentNullException("body"); + } + + var actions = new Action[ThreadQueue.ThreadCount]; + var count = toExclusive - fromInclusive; + var size = count / actions.Length; + + if (count < 1) + { + return; + } + + // partition the jobs into separate sets for each but the last worked thread + for (var i = 0; i < actions.Length - 1; i++) + { + var start = fromInclusive + (i * size); + var stop = fromInclusive + ((i + 1) * size); + + actions[i] = + () => + { + for (var j = start; j < stop; j++) + { + body(j); + } + }; + } + + // add another set for last worker thread + actions[actions.Length - 1] = + () => + { + for (var i = fromInclusive + ((actions.Length - 1) * size); i < toExclusive; i++) + { + body(i); + } + }; + + Invoke(actions); + } + + /// + /// Executes each of the provided actions inside a discrete, asynchronous task. + /// + /// An array of actions to execute. + /// The argument is null. + /// The actions array contains a null element. + /// An action threw an exception. + internal static void Invoke(params Action[] actions) + { + if (actions == null) + { + throw new ArgumentNullException("actions"); + } + + // create a job for each action + var tasks = new Task[actions.Length]; + for (int i = 0; i < tasks.Length; i++) + { + Action action = actions[i]; + if (action == null) + { + throw new ArgumentException(String.Format(Resources.ArgumentItemNull, "actions"), "actions"); + } + + tasks[i] = new Task(action); + } + + // run the jobs + ThreadQueue.Enqueue(tasks); + + // wait until all jobs have completed + if (Thread.CurrentThread.GetApartmentState() == ApartmentState.STA) + { + //not sure if this the best approach for STA + for (int i = 0; i < tasks.Length; i++) + { + tasks[i].WaitOne(); + } + } + else + { + WaitHandle.WaitAll(tasks); + } + + // collect all thrown exceptions and dispose the jobs + var exceptions = new List(); + foreach (var task in tasks) + { + if (task.ThrewException) + { + exceptions.Add(task.Exception); + } + + //this calls dispose + task.Close(); + } + + // throw the aggregated exceptions, if any + if (exceptions.Count > 0) + { + throw new AggregateException(exceptions); + } + } + } +} \ No newline at end of file diff --git a/src/Managed/Threading/Task.cs b/src/Managed/Threading/Task.cs new file mode 100644 index 00000000..74a408ce --- /dev/null +++ b/src/Managed/Threading/Task.cs @@ -0,0 +1,87 @@ +// +// 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.Threading +{ + using System; + using System.Threading; + + /// + /// Internal Parallel Task Handle. + /// + internal class Task : EventWaitHandle + { + /// + /// Delegate to the task's action. + /// + private readonly Action _body; + + /// + /// Initializes a new instance of the Task class. + /// + /// Delegate to the task's action. + internal Task(Action body) + : base(false, EventResetMode.ManualReset) + { + if (body == null) + { + throw new ArgumentNullException("body"); + } + + _body = body; + } + + /// + /// Gets a value indicating whether the task has thrown one or more exceptions while executing. + /// + internal bool ThrewException + { + get { return Exception != null; } + } + + /// + /// Gets the exception thrown by the task, if any. + /// + internal Exception Exception { get; private set; } + + /// + /// Run the task. + /// + internal void Compute() + { + try + { + _body(); + } + catch (Exception e) + { + Exception = e; + } + } + } +} \ No newline at end of file diff --git a/src/Managed/Threading/ThreadQueue.cs b/src/Managed/Threading/ThreadQueue.cs new file mode 100644 index 00000000..2545a678 --- /dev/null +++ b/src/Managed/Threading/ThreadQueue.cs @@ -0,0 +1,235 @@ +// +// 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.Threading +{ + using System; + using System.Collections.Generic; + using System.Threading; + + /// + /// Internal Parallel Thread Queue. + /// + internal static class ThreadQueue + { + /// + /// Sync Object for the thread queue state. + /// + private static readonly object _stateSync = new object(); + + /// + /// Sync Object for queue access (to be sure it's used by us only). + /// + private static readonly object _queueSync = new object(); + + /// + /// Maximum number of jobs that can be in the queue at the same time. + /// + private const int _maximumQueueLength = 1024; + + /// + /// Counting Semaphore to make the worker thread wait for jobs + /// + private static Semaphore _tasksAvailableSemaphore; + + /// + /// Queue holding the pending jobs. + /// + private static readonly Queue _queue = new Queue(); + + /// + /// Running flag, used to signal worker threads to stop cleanly. + /// + private static bool _running = true; + + /// + /// Worker threads + /// + private static Thread[] _threads; + + /// + /// Number of worked threads. + /// + internal static int ThreadCount { get; private set; } + + /// + /// Static Constructor + /// + static ThreadQueue() + { + // TODO: Control.ThreadCount instead of Environment.ProcessorCount + Start(Environment.ProcessorCount); + } + + /// + /// Add a job to the queue. + /// + /// The job to run. + internal static void Enqueue(Task task) + { + if(!_running) + { + Start(); + } + + lock(_queueSync) + { + _queue.Enqueue(task); + } + + _tasksAvailableSemaphore.Release(); + } + + /// + /// Add a set of jobs to the queue. + /// + /// The jobs to run. + internal static void Enqueue(IList tasks) + { + if(!_running) + { + Start(); + } + + lock(_queueSync) + { + foreach(var task in tasks) + { + _queue.Enqueue(task); + } + } + + _tasksAvailableSemaphore.Release(tasks.Count); + } + + /// + /// Worker Thread Program + /// + private static void WorkerThreadStart() + { + while (_running) + { + // Wait until a job is available, or we should shut down + _tasksAvailableSemaphore.WaitOne(); + + // Check whether we should shut down + if(!_running) + { + _tasksAvailableSemaphore.Release(); + break; + } + + // Get the job... + Task task = null; + lock (_queueSync) + { + if(_queue.Count > 0) + { + task = _queue.Dequeue(); + } + } + + // ...and run it + if (task != null) + { + task.Compute(); + task.Set(); + } + } + } + + internal static void Start(int numberOfThreads) + { + lock (_stateSync) + { + if (_threads != null) + { + if (_threads.Length == numberOfThreads) + { + return; + } + + Shutdown(); + } + + ThreadCount = numberOfThreads; + Start(); + } + } + + internal static void Start() + { + lock (_stateSync) + { + if (_threads != null) + { + return; + } + + _tasksAvailableSemaphore = new Semaphore(_queue.Count, _maximumQueueLength); + _running = true; + _threads = new Thread[ThreadCount]; + + for (var i = 0; i < _threads.Length; i++) + { + _threads[i] = new Thread(WorkerThreadStart) + { + IsBackground = true + }; + + _threads[i].Start(); + } + } + } + + internal static void Shutdown() + { + lock (_stateSync) + { + if (_threads == null) + { + return; + } + + // try to stop the worker threads cleanly + _running = false; + _tasksAvailableSemaphore.Release(); + + // wait until all threads have stopped + foreach (var thread in _threads) + { + thread.Join(); + } + + _tasksAvailableSemaphore.Close(); + _tasksAvailableSemaphore = null; + _threads = null; + } + } + } +} \ No newline at end of file diff --git a/src/Native.UnitTests/Native.UnitTests.csproj b/src/Native.UnitTests/Native.UnitTests.csproj index faeaab98..38c98b47 100644 --- a/src/Native.UnitTests/Native.UnitTests.csproj +++ b/src/Native.UnitTests/Native.UnitTests.csproj @@ -92,6 +92,9 @@ SpecialFunctionsTest\ErfTests.cs + + ThreadingTests\ParallelTest.cs + diff --git a/src/Native/Native.csproj b/src/Native/Native.csproj index 7f40693b..63fca844 100644 --- a/src/Native/Native.csproj +++ b/src/Native/Native.csproj @@ -137,6 +137,18 @@ SpecialFunctions\Erf.cs + + Threading\AggregateException.cs + + + Threading\Parallel.cs + + + Threading\Task.cs + + + Threading\ThreadQueue.cs +