Browse Source

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 <marcus@cuda.net>
la-knuth
Marcus Cuda 17 years ago
parent
commit
dd2a8328a9
  1. 2
      src/Numerics/Algorithms/LinearAlgebra/ILinearAlgebra.cs
  2. 2
      src/Numerics/Algorithms/LinearAlgebra/ManagedLinearAlgebra.cs
  3. 48
      src/Numerics/Threading/Parallel.cs
  4. 90
      src/Numerics/Threading/Task.cs
  5. 2
      src/Numerics/Threading/ThreadQueue.cs

2
src/Numerics/Algorithms/LinearAlgebra/ILinearAlgebra.cs

@ -33,7 +33,7 @@ namespace MathNet.Numerics.Algorithms.LinearAlgebra
/// Adds a scaled vector to another: <c>y += alpha*x</c>. /// Adds a scaled vector to another: <c>y += alpha*x</c>.
/// </summary> /// </summary>
/// <param name="y">The vector to update.</param> /// <param name="y">The vector to update.</param>
/// <param name="alpha">The value to scale <param name="x"/> by.</param> /// <param name="alpha">The value to scale <paramref name="x"/> by.</param>
/// <param name="x">The vector to add to <paramref name="y"/>.</param> /// <param name="x">The vector to add to <paramref name="y"/>.</param>
/// <remarks>This is equivalent to the AXPY BLAS routine.</remarks> /// <remarks>This is equivalent to the AXPY BLAS routine.</remarks>
void AddVectorToScaledVector(double[] y, double alpha, double[] x); void AddVectorToScaledVector(double[] y, double alpha, double[] x);

2
src/Numerics/Algorithms/LinearAlgebra/ManagedLinearAlgebra.cs

@ -35,7 +35,7 @@ namespace MathNet.Numerics.Algorithms.LinearAlgebra
/// Adds a scaled vector to another: <c>y += alpha*x</c>. /// Adds a scaled vector to another: <c>y += alpha*x</c>.
/// </summary> /// </summary>
/// <param name="y">The vector to update.</param> /// <param name="y">The vector to update.</param>
/// <param name="alpha">The value to scale <param name="x"/> by.</param> /// <param name="alpha">The value to scale <paramref name="x"/> by.</param>
/// <param name="x">The vector to add to <paramref name="y"/>.</param> /// <param name="x">The vector to add to <paramref name="y"/>.</param>
/// <remarks>This equivalent to the AXPY BLAS routine.</remarks> /// <remarks>This equivalent to the AXPY BLAS routine.</remarks>
public void AddVectorToScaledVector(double[] y, double alpha, double[] x) public void AddVectorToScaledVector(double[] y, double alpha, double[] x)

48
src/Numerics/Threading/Parallel.cs

@ -146,30 +146,30 @@ namespace MathNet.Numerics.Threading
{ {
var start = fromInclusive + (i * size); var start = fromInclusive + (i * size);
var stop = fromInclusive + ((i + 1) * size); var stop = fromInclusive + ((i + 1) * size);
tasks[i] = new Task<T>(intial, tasks[i] = new Task<T>(
localData => localData =>
{ {
var localresult = localData; var localresult = localData;
for (var j = start; j < stop; j++) 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]); ThreadQueue.Enqueue(tasks[i]);
} }
// add another set for last worker thread // add another set for last worker thread
tasks[tasks.Length - 1] = new Task<T>(intial, tasks[tasks.Length - 1] = new Task<T>(
localData => localData =>
{ {
var localresult = localData; var localresult = localData;
for (var i = fromInclusive + ((tasks.Length - 1) * size); i < toExclusive; i++) 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]); ThreadQueue.Enqueue(tasks[tasks.Length - 1]);
if (tasks.Length <= 0) if (tasks.Length <= 0)
@ -257,7 +257,7 @@ namespace MathNet.Numerics.Threading
} }
/// <summary> /// <summary>
/// Executes a for each operation on an IEnumerable<TSource> in which iterations may run in parallel. /// Executes a for each operation on an IEnumerable{TSource in which iterations may run in parallel.
/// </summary> /// </summary>
/// <typeparam name="TSource">The type of the data in the source.</typeparam> /// <typeparam name="TSource">The type of the data in the source.</typeparam>
/// <typeparam name="TLocal">The type of the thread-local data.</typeparam> /// <typeparam name="TLocal">The type of the thread-local data.</typeparam>
@ -293,16 +293,16 @@ namespace MathNet.Numerics.Threading
count++; count++;
} }
var task = new Task<TLocal>(intial, var task = new Task<TLocal>(
localData => localData =>
{ {
var localresult = localData; var localresult = localData;
for (var i = 0; i < pos; i++) 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); ThreadQueue.Enqueue(task);
tasks.Add(task); tasks.Add(task);
@ -398,19 +398,10 @@ namespace MathNet.Numerics.Threading
/// <param name="tasks">The tasks.</param> /// <param name="tasks">The tasks.</param>
private static void WaitForTasksToComplete(Task[] tasks) private static void WaitForTasksToComplete(Task[] tasks)
{ {
// wait until all tasks have been completed for (var i = 0; i < tasks.Length; i++)
if (Thread.CurrentThread.GetApartmentState() == ApartmentState.STA) {
{ tasks[i].Wait();
// not sure if this the best approach for STA }
for (var i = 0; i < tasks.Length; i++)
{
tasks[i].WaitOne();
}
}
else
{
WaitHandle.WaitAll(tasks);
}
} }
/// <summary> /// <summary>
@ -423,13 +414,10 @@ namespace MathNet.Numerics.Threading
var exceptions = new List<Exception>(); var exceptions = new List<Exception>();
foreach (var task in tasks) foreach (var task in tasks)
{ {
if (task.ThrewException) if (task.IsFaulted)
{ {
exceptions.Add(task.Exception); exceptions.Add(task.Exception);
} }
// this calls dispose
task.Close();
} }
// throw the aggregated exceptions, if any // throw the aggregated exceptions, if any

90
src/Numerics/Threading/Task.cs

@ -34,7 +34,7 @@ namespace MathNet.Numerics.Threading
/// <summary> /// <summary>
/// Internal Parallel Task Handle. /// Internal Parallel Task Handle.
/// </summary> /// </summary>
internal class Task : EventWaitHandle internal class Task
{ {
/// <summary> /// <summary>
/// Delegate to the task's action. /// Delegate to the task's action.
@ -45,8 +45,7 @@ namespace MathNet.Numerics.Threading
/// Initializes a new instance of the Task class. /// Initializes a new instance of the Task class.
/// </summary> /// </summary>
/// <param name="body">Delegate to the task's action.</param> /// <param name="body">Delegate to the task's action.</param>
internal Task(Action body) public Task(Action body)
: this()
{ {
if (body == null) if (body == null)
{ {
@ -56,79 +55,114 @@ namespace MathNet.Numerics.Threading
_body = body; _body = body;
} }
protected Task() : base(false, EventResetMode.ManualReset) { } /// <summary>
/// Initializes a new instance of the <see cref="Task"/> class.
/// </summary>
protected Task()
{
}
/// <summary> /// <summary>
/// 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.
/// </summary> /// </summary>
internal bool ThrewException /// <value>
/// <c>true</c> if this task completed due to an unhandled exception; otherwise, <c>false</c>.
/// </value>
public bool IsFaulted
{ {
get { return Exception != null; } get { return Exception != null; }
} }
/// <summary> /// <summary>
/// Gets the exception thrown by the task, if any. /// Gets a value indicating whether this task has completed.
/// </summary>
/// <value>
/// <c>true</c> if this task has completed; otherwise, <c>false</c>.
/// </value>
public bool IsCompleted
{
get; private set;
}
/// <summary>
/// Gets or sets the exception thrown by the task, if any.
/// </summary> /// </summary>
protected internal Exception Exception { get; set; } public Exception Exception { get; set; }
/// <summary> /// <summary>
/// Run the task. /// Run the task.
/// </summary> /// </summary>
internal virtual void Compute() public void Compute()
{ {
try try
{ {
_body(); DoCompute();
IsCompleted = true;
} }
catch (Exception e) catch (Exception e)
{ {
Exception = e; Exception = e;
} }
} }
/// <summary>
/// Runs the actual task.
/// </summary>
protected virtual void DoCompute()
{
_body();
}
public void Wait()
{
while(!IsCompleted && !IsFaulted)
{
Thread.Sleep(100);
}
}
} }
/// <summary> /// <summary>
/// Internal Generic Parallel Task Handle. /// Internal Generic Parallel Task Handle.
/// </summary> /// </summary>
internal class Task<T> : Task internal class Task<TResult> : Task
{ {
/// <summary> /// <summary>
/// Delegate to the task's action. /// Delegate to the task's action.
/// </summary> /// </summary>
private readonly Func<T, T> _body; private readonly Func<object, TResult> _body;
//private T _initialValue;
public T Result { get; private set; } private readonly object _state;
/// <summary>
/// Gets the result of the task.
/// </summary>
/// <value>The result of the task.</value>
public TResult Result { get; private set; }
/// <summary> /// <summary>
/// Initializes a new instance of the Task class. /// Initializes a new instance of the Task class.
/// </summary> /// </summary>
/// <param name="intialValue">The initial value.</param> /// <param name="state">An object representing data to be used by the action.</param>
/// <param name="body">Delegate to the task's action.</param> /// <param name="body">Delegate to the task's action.</param>
internal Task(T intialValue, Func<T, T> body) public Task(Func<object, TResult> body, object state)
{ {
if (body == null) if (body == null)
{ {
throw new ArgumentNullException("body"); throw new ArgumentNullException("body");
} }
Result = intialValue;
_state = state;
_body = body; _body = body;
} }
/// <summary> /// <summary>
/// Run the task. /// Runs the actual task.
/// </summary> /// </summary>
internal override void Compute() protected override void DoCompute()
{ {
try Result = _body(_state);
{
Result = _body(Result);
}
catch (Exception e)
{
Exception = e;
}
} }
} }
} }

2
src/Numerics/Threading/ThreadQueue.cs

@ -176,7 +176,7 @@ namespace MathNet.Numerics.Threading
// ...and run it // ...and run it
task.Compute(); task.Compute();
task.Set(); //task.Set();
} }
} }

Loading…
Cancel
Save