From dd2a8328a920bec437a5353e63aec766aa59aba6 Mon Sep 17 00:00:00 2001 From: Marcus Cuda Date: Sun, 30 Aug 2009 23:42:51 +0800 Subject: [PATCH] Changed Task class to not extend WaitHandle since there can only be at most 64 waithandles There is still an issues with too many semaphores being created Signed-off-by: Marcus Cuda --- .../LinearAlgebra/ILinearAlgebra.cs | 2 +- .../LinearAlgebra/ManagedLinearAlgebra.cs | 2 +- src/Numerics/Threading/Parallel.cs | 48 ++++------ src/Numerics/Threading/Task.cs | 90 +++++++++++++------ src/Numerics/Threading/ThreadQueue.cs | 2 +- 5 files changed, 83 insertions(+), 61 deletions(-) diff --git a/src/Numerics/Algorithms/LinearAlgebra/ILinearAlgebra.cs b/src/Numerics/Algorithms/LinearAlgebra/ILinearAlgebra.cs index ebc15ee2..eb7d43eb 100644 --- a/src/Numerics/Algorithms/LinearAlgebra/ILinearAlgebra.cs +++ b/src/Numerics/Algorithms/LinearAlgebra/ILinearAlgebra.cs @@ -33,7 +33,7 @@ namespace MathNet.Numerics.Algorithms.LinearAlgebra /// Adds a scaled vector to another: y += alpha*x. /// /// The vector to update. - /// The value to scale by. + /// The value to scale by. /// The vector to add to . /// This is equivalent to the AXPY BLAS routine. void AddVectorToScaledVector(double[] y, double alpha, double[] x); diff --git a/src/Numerics/Algorithms/LinearAlgebra/ManagedLinearAlgebra.cs b/src/Numerics/Algorithms/LinearAlgebra/ManagedLinearAlgebra.cs index 22d5beae..dab836da 100644 --- a/src/Numerics/Algorithms/LinearAlgebra/ManagedLinearAlgebra.cs +++ b/src/Numerics/Algorithms/LinearAlgebra/ManagedLinearAlgebra.cs @@ -35,7 +35,7 @@ namespace MathNet.Numerics.Algorithms.LinearAlgebra /// Adds a scaled vector to another: y += alpha*x. /// /// The vector to update. - /// The value to scale by. + /// The value to scale by. /// The vector to add to . /// This equivalent to the AXPY BLAS routine. public void AddVectorToScaledVector(double[] y, double alpha, double[] x) diff --git a/src/Numerics/Threading/Parallel.cs b/src/Numerics/Threading/Parallel.cs index b49c19ca..813e7796 100644 --- a/src/Numerics/Threading/Parallel.cs +++ b/src/Numerics/Threading/Parallel.cs @@ -146,30 +146,30 @@ namespace MathNet.Numerics.Threading { var start = fromInclusive + (i * size); var stop = fromInclusive + ((i + 1) * size); - tasks[i] = new Task(intial, + tasks[i] = new Task( localData => { var localresult = localData; for (var j = start; j < stop; j++) { - localresult = body(j, localresult); + localresult = body(j, (T)localresult); } - return localresult; - } ); + return (T)localresult; + }, intial ); ThreadQueue.Enqueue(tasks[i]); } // add another set for last worker thread - tasks[tasks.Length - 1] = new Task(intial, + tasks[tasks.Length - 1] = new Task( localData => { var localresult = localData; for (var i = fromInclusive + ((tasks.Length - 1) * size); i < toExclusive; i++) { - localresult = body(i, localresult); + localresult = body(i, (T)localresult); } - return localresult; - } ); + return (T)localresult; + }, intial ); ThreadQueue.Enqueue(tasks[tasks.Length - 1]); if (tasks.Length <= 0) @@ -257,7 +257,7 @@ namespace MathNet.Numerics.Threading } /// - /// Executes a for each operation on an IEnumerable in which iterations may run in parallel. + /// Executes a for each operation on an IEnumerable{TSource in which iterations may run in parallel. /// /// The type of the data in the source. /// The type of the thread-local data. @@ -293,16 +293,16 @@ namespace MathNet.Numerics.Threading count++; } - var task = new Task(intial, + var task = new Task( localData => { var localresult = localData; for (var i = 0; i < pos; i++) { - localresult = body(list[i], localresult); + localresult = body(list[i], (TLocal)localresult); } - return localresult; - }); + return (TLocal)localresult; + }, intial); ThreadQueue.Enqueue(task); tasks.Add(task); @@ -398,19 +398,10 @@ namespace MathNet.Numerics.Threading /// 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 (var i = 0; i < tasks.Length; i++) - { - tasks[i].WaitOne(); - } - } - else - { - WaitHandle.WaitAll(tasks); - } + for (var i = 0; i < tasks.Length; i++) + { + tasks[i].Wait(); + } } /// @@ -423,13 +414,10 @@ namespace MathNet.Numerics.Threading var exceptions = new List(); foreach (var task in tasks) { - if (task.ThrewException) + if (task.IsFaulted) { exceptions.Add(task.Exception); } - - // this calls dispose - task.Close(); } // throw the aggregated exceptions, if any diff --git a/src/Numerics/Threading/Task.cs b/src/Numerics/Threading/Task.cs index 72ace2c3..33dd438f 100644 --- a/src/Numerics/Threading/Task.cs +++ b/src/Numerics/Threading/Task.cs @@ -34,7 +34,7 @@ namespace MathNet.Numerics.Threading /// /// Internal Parallel Task Handle. /// - internal class Task : EventWaitHandle + internal class Task { /// /// Delegate to the task's action. @@ -45,8 +45,7 @@ namespace MathNet.Numerics.Threading /// Initializes a new instance of the Task class. /// /// Delegate to the task's action. - internal Task(Action body) - : this() + public Task(Action body) { if (body == null) { @@ -56,79 +55,114 @@ namespace MathNet.Numerics.Threading _body = body; } - protected Task() : base(false, EventResetMode.ManualReset) { } + /// + /// Initializes a new instance of the class. + /// + protected Task() + { + } /// - /// Gets a value indicating whether the task has thrown one or more exceptions while executing. + /// Gets a value indicating whether the task completed due to an unhandled exception. /// - internal bool ThrewException + /// + /// true if this task completed due to an unhandled exception; otherwise, false. + /// + public bool IsFaulted { get { return Exception != null; } } /// - /// Gets the exception thrown by the task, if any. + /// Gets a value indicating whether this task has completed. + /// + /// + /// true if this task has completed; otherwise, false. + /// + public bool IsCompleted + { + get; private set; + } + + /// + /// Gets or sets the exception thrown by the task, if any. /// - protected internal Exception Exception { get; set; } + public Exception Exception { get; set; } /// /// Run the task. /// - internal virtual void Compute() + public void Compute() { try { - _body(); + DoCompute(); + IsCompleted = true; } catch (Exception e) { Exception = e; } } + + /// + /// Runs the actual task. + /// + protected virtual void DoCompute() + { + _body(); + } + + public void Wait() + { + while(!IsCompleted && !IsFaulted) + { + Thread.Sleep(100); + } + + } } /// /// Internal Generic Parallel Task Handle. /// - internal class Task : Task + internal class Task : Task { /// /// Delegate to the task's action. /// - private readonly Func _body; - - //private T _initialValue; + private readonly Func _body; - public T Result { get; private set; } + private readonly object _state; + + /// + /// Gets the result of the task. + /// + /// The result of the task. + public TResult Result { get; private set; } /// /// Initializes a new instance of the Task class. /// - /// The initial value. + /// An object representing data to be used by the action. /// Delegate to the task's action. - internal Task(T intialValue, Func body) + public Task(Func body, object state) { if (body == null) { throw new ArgumentNullException("body"); } - Result = intialValue; + + _state = state; _body = body; } /// - /// Run the task. + /// Runs the actual task. /// - internal override void Compute() + protected override void DoCompute() { - try - { - Result = _body(Result); - } - catch (Exception e) - { - Exception = e; - } + Result = _body(_state); } } } diff --git a/src/Numerics/Threading/ThreadQueue.cs b/src/Numerics/Threading/ThreadQueue.cs index 555b41c9..b83a9c1b 100644 --- a/src/Numerics/Threading/ThreadQueue.cs +++ b/src/Numerics/Threading/ThreadQueue.cs @@ -176,7 +176,7 @@ namespace MathNet.Numerics.Threading // ...and run it task.Compute(); - task.Set(); + //task.Set(); } }