Browse Source

Tweaked the parallel code, but the ForEach still performs slowly

Signed-off-by: Marcus Cuda <marcus@cuda.net>
la-knuth
Marcus Cuda 17 years ago
parent
commit
1ffd3e13e6
  1. 8
      src/Numerics/Control.cs
  2. 12
      src/Numerics/LinearAlgebra/Double/DenseVector.cs
  3. 19
      src/Numerics/LinearAlgebra/Double/Vector.cs
  4. 1
      src/Numerics/Numerics.csproj
  5. 119
      src/Numerics/Threading/Parallel.cs
  6. 49
      src/Numerics/Threading/Task.cs
  7. 79
      src/Numerics/Threading/TaskOfT.cs
  8. 13
      src/Numerics/Threading/ThreadQueue.cs

8
src/Numerics/Control.cs

@ -43,7 +43,6 @@ namespace MathNet.Numerics
CheckDistributionParameters = true;
ThreadSafeRandomNumberGenerators = true;
DisableParallelization = false;
InitialThreadBlockSize = 2;
}
/// <summary>
@ -76,12 +75,5 @@ namespace MathNet.Numerics
/// Gets or sets a value indicating whether parallelization shall be disabled globally.
/// </summary>
public static bool DisableParallelization { get; set; }
/// <summary>
/// Gets or sets the initial size of a <see cref="Parallel.ForEach{T}"/>
/// processing block (the number of elements the first thread should process).
/// </summary>
/// <value>The initial size of the thread processing bloc.</value>
public static int InitialThreadBlockSize { get; set; }
}
}

12
src/Numerics/LinearAlgebra/Double/DenseVector.cs

@ -577,7 +577,9 @@ namespace MathNet.Numerics.LinearAlgebra.Double
var sum = 0.0;
var syncLock = new object();
Parallel.For(0, Count,
Parallel.For(
0,
Count,
() => 0.0,
(index, localData) =>
{
@ -620,7 +622,9 @@ namespace MathNet.Numerics.LinearAlgebra.Double
var sum = 0.0;
var syncLock = new object();
Parallel.For(0, Count,
Parallel.For(
0,
Count,
() => 0.0,
(index, localData) =>
{
@ -648,7 +652,9 @@ namespace MathNet.Numerics.LinearAlgebra.Double
var syncLock = new object();
Parallel.For(0, Count,
Parallel.For(
0,
Count,
() => 0.0,
(index, localData) =>
{

19
src/Numerics/LinearAlgebra/Double/Vector.cs

@ -609,20 +609,22 @@ namespace MathNet.Numerics.LinearAlgebra.Double
var sum = 0.0;
var syncLock = new object();
Parallel.For(0, Count,
()=> 0.0,
Parallel.For(
0,
Count,
() => 0.0,
(index, localData) =>
{
localData += Math.Pow(Math.Abs(this[index]), p);
return localData;
},
localResult=>
localResult =>
{
lock (syncLock)
{
sum += localResult;
}
} );
});
return Math.Pow(sum, 1.0 / p);
}
@ -637,7 +639,9 @@ namespace MathNet.Numerics.LinearAlgebra.Double
{
var max = 0.0;
var syncLock = new object();
Parallel.For(0, Count,
Parallel.For(
0,
Count,
() => 0.0,
(index, localData) =>
{
@ -716,6 +720,7 @@ namespace MathNet.Numerics.LinearAlgebra.Double
{
return;
}
Parallel.For(0, Count, index => target[index] = this[index]);
}
@ -769,9 +774,7 @@ namespace MathNet.Numerics.LinearAlgebra.Double
}
else
{
Parallel.For(0, count,
index => destination[destinationOffset + index] = this[offset + index]
);
Parallel.For(0, count, index => destination[destinationOffset + index] = this[offset + index]);
}
}

1
src/Numerics/Numerics.csproj

@ -122,6 +122,7 @@
<Compile Include="Statistics\DescriptiveStatistics.cs" />
<Compile Include="Statistics\Statistics.cs" />
<Compile Include="Threading\AggregateException.cs" />
<Compile Include="Threading\TaskOfT.cs" />
<Compile Include="Threading\Parallel.cs" />
<Compile Include="Threading\Task.cs" />
<Compile Include="Threading\ThreadQueue.cs" />

119
src/Numerics/Threading/Parallel.cs

@ -30,17 +30,28 @@ namespace MathNet.Numerics.Threading
{
using System;
using System.Collections.Generic;
using System.Threading;
using Properties;
/// <summary>
/// Provides support for parallel loops.
/// Provides support for parallel loops.
/// </summary>
internal static class Parallel
{
/// <summary>
/// The amount to scale the foreach buffer after each iteration.
/// </summary>
private const int ScalingFactor = 2;
/// <summary>
/// The maximum size of the foreach buffer.
/// </summary>
private const int MaxBlockSize = 65536;
/// <summary>
/// The initial size of the for each buffer.
/// </summary>
private const int IntialBlockSize = 1024;
/// <summary>
/// Executes a for loop in which iterations may run in parallel.
/// </summary>
@ -121,29 +132,26 @@ namespace MathNet.Numerics.Threading
/// <param name="toExclusive">The end index, exclusive.</param>
/// <param name="localInit">The function delegate that returns the initial state of the local data for each thread.</param>
/// <param name="body">The delegate that is invoked once per iteration.</param>
/// <param name="localFinally">The delegate that is invoked once per iteration.</param>
public static void For<T>(int fromInclusive, int toExclusive,
Func<T> localInit,
Func<int, T, T> body,
Action<T> localFinally)
/// <param name="localFinally">The delegate that performs a final action on the local state of each thread.</param>
public static void For<T>(int fromInclusive, int toExclusive, Func<T> localInit, Func<int, T, T> body, Action<T> localFinally)
{
var count = toExclusive - fromInclusive;
var tasks = new Task<T>[ThreadQueue.ThreadCount];
var size = count / tasks.Length;
if (count <= 0){
// if (count <= 1)
// {
// if (count == 1)
// {
// body(fromInclusive);
// }
var intial = localInit();
// fast forward execution if it's only one or none items
if (count <= 1)
{
if (count == 1)
{
localFinally(body(fromInclusive, intial));
}
return;
}
var intial = localInit();
// partition the jobs into separate sets for each but the last worked thread
for (var i = 0; i < tasks.Length - 1; i++)
{
@ -157,8 +165,10 @@ namespace MathNet.Numerics.Threading
{
localresult = body(j, (T)localresult);
}
return (T)localresult;
}, intial );
},
intial);
ThreadQueue.Enqueue(tasks[i]);
}
@ -171,20 +181,25 @@ namespace MathNet.Numerics.Threading
{
localresult = body(i, (T)localresult);
}
return (T)localresult;
}, intial );
},
intial);
ThreadQueue.Enqueue(tasks[tasks.Length - 1]);
if (tasks.Length <= 0)
{
return;
}
WaitForTasksToComplete(tasks);
for (var i = 0; i < tasks.Length; i++)
{
localFinally(tasks[i].Result);
}
CollectExceptionsAndDisposeTasks(tasks);
CollectExceptions(tasks);
}
/// <summary>
@ -221,10 +236,10 @@ namespace MathNet.Numerics.Threading
return;
}
var enumerator = source.GetEnumerator();
var maxBlockSize = Control.InitialThreadBlockSize;
var scalingFactor = ScalingFactor;
var maxBlockSize = IntialBlockSize;
var tasks = new List<Task>();
var enumerator = source.GetEnumerator();
while (enumerator.MoveNext())
{
var pos = 0;
@ -249,13 +264,13 @@ namespace MathNet.Numerics.Threading
ThreadQueue.Enqueue(task);
tasks.Add(task);
maxBlockSize = Math.Min(MaxBlockSize, maxBlockSize * scalingFactor);
maxBlockSize = Math.Min(MaxBlockSize, maxBlockSize * ScalingFactor);
}
if (tasks.Count > 0)
{
WaitForTasksToComplete(tasks.ToArray());
CollectExceptionsAndDisposeTasks(tasks);
CollectExceptions(tasks);
}
}
@ -268,21 +283,42 @@ namespace MathNet.Numerics.Threading
/// <param name="localInit">The function delegate that returns the initial state of the local data for each thread.</param>
/// <param name="body">The delegate that is invoked once per iteration.</param>
/// <param name="localFinally">The delegate that performs a final action on the local state of each thread.</param>
public static void ForEach<TSource, TLocal>(IEnumerable<TSource> source, Func<TLocal> localInit,
Func<TSource, TLocal, TLocal> body, Action<TLocal> localFinally)
public static void ForEach<TSource, TLocal>(IEnumerable<TSource> source, Func<TLocal> localInit, Func<TSource, TLocal, TLocal> body, Action<TLocal> localFinally)
{
if (body == null)
{
throw new ArgumentNullException("body");
}
var enumerator = source.GetEnumerator();
var maxBlockSize = Control.InitialThreadBlockSize;
var scalingFactor = ScalingFactor;
// fast forward execution in case parallelization is disabled
if (Control.DisableParallelization
|| ThreadQueue.ThreadCount <= 1
|| ThreadQueue.IsInWorkerThread)
{
var localResult = localInit();
foreach (var item in source)
{
localResult = body(item, localResult);
}
localFinally(localResult);
return;
}
// source is a IList, call For instead.
if (source is IList<TSource>)
{
var list = (IList<TSource>)source;
For(0, list.Count, localInit, (i, local) => body(list[i], local), localFinally);
return;
}
var maxBlockSize = IntialBlockSize;
var tasks = new List<Task<TLocal>>();
var intial = localInit();
var enumerator = source.GetEnumerator();
while (enumerator.MoveNext())
{
var pos = 0;
@ -304,25 +340,30 @@ namespace MathNet.Numerics.Threading
{
localresult = body(list[i], (TLocal)localresult);
}
return (TLocal)localresult;
}, intial);
},
intial);
ThreadQueue.Enqueue(task);
tasks.Add(task);
maxBlockSize = Math.Min(MaxBlockSize, maxBlockSize * scalingFactor);
maxBlockSize = Math.Min(MaxBlockSize, maxBlockSize * ScalingFactor);
}
if (tasks.Count <= 0)
{
return;
}
var taskArray = tasks.ToArray();
WaitForTasksToComplete(taskArray);
for (var i = 0; i < taskArray.Length; i++)
{
localFinally(tasks[i].Result);
}
CollectExceptionsAndDisposeTasks(taskArray);
CollectExceptions(taskArray);
}
/// <summary>
@ -355,7 +396,7 @@ namespace MathNet.Numerics.Threading
|| ThreadQueue.ThreadCount <= 1
|| ThreadQueue.IsInWorkerThread)
{
for (int i = 0; i < actions.Length; i++)
for (var i = 0; i < actions.Length; i++)
{
actions[i]();
}
@ -392,7 +433,7 @@ namespace MathNet.Numerics.Threading
WaitForTasksToComplete(tasks);
CollectExceptionsAndDisposeTasks(tasks);
CollectExceptions(tasks);
}
/// <summary>
@ -401,17 +442,17 @@ namespace MathNet.Numerics.Threading
/// <param name="tasks">The tasks.</param>
private static void WaitForTasksToComplete(Task[] tasks)
{
for (var i = 0; i < tasks.Length; i++)
{
tasks[i].Wait();
}
for (var i = 0; i < tasks.Length; i++)
{
tasks[i].Wait();
}
}
/// <summary>
/// Collects the exceptions and dispose tasks.
/// </summary>
/// <param name="tasks">The tasks.</param>
private static void CollectExceptionsAndDisposeTasks(IEnumerable<Task> tasks)
private static void CollectExceptions(IEnumerable<Task> tasks)
{
// collect all thrown exceptions and dispose the jobs
var exceptions = new List<Exception>();

49
src/Numerics/Threading/Task.cs

@ -113,56 +113,15 @@ namespace MathNet.Numerics.Threading
_body();
}
public void Wait()
{
while(!IsCompleted && !IsFaulted)
{
Thread.Sleep(100);
}
}
}
/// <summary>
/// Internal Generic Parallel Task Handle.
/// </summary>
internal class Task<TResult> : Task
{
/// <summary>
/// Delegate to the task's action.
/// </summary>
private readonly Func<object, TResult> _body;
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>
/// Initializes a new instance of the Task class.
/// Waits for the task to complete execution.
/// </summary>
/// <param name="state">An object representing data to be used by the action.</param>
/// <param name="body">Delegate to the task's action.</param>
public Task(Func<object, TResult> body, object state)
public void Wait()
{
if (body == null)
while (!IsCompleted && !IsFaulted)
{
throw new ArgumentNullException("body");
Thread.Sleep(100);
}
_state = state;
_body = body;
}
/// <summary>
/// Runs the actual task.
/// </summary>
protected override void DoCompute()
{
Result = _body(_state);
}
}
}

79
src/Numerics/Threading/TaskOfT.cs

@ -0,0 +1,79 @@
// <copyright file="TaskOfT.cs" company="Math.NET">
// 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.
// </copyright>
namespace MathNet.Numerics.Threading
{
using System;
/// <summary>
/// Internal Generic Parallel Task Handle.
/// </summary>
/// <typeparam name="TResult">The type of the result.</typeparam>
internal class Task<TResult> : Task
{
/// <summary>
/// Delegate to the task's action.
/// </summary>
private readonly Func<object, TResult> _body;
/// <summary>
/// Variable used to hold state information between interations.
/// </summary>
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>
/// Initializes a new instance of the Task class.
/// </summary>
/// <param name="body">Delegate to the task's action.</param>
/// <param name="state">An object representing data to be used by the action.</param>
public Task(Func<object, TResult> body, object state)
{
if (body == null)
{
throw new ArgumentNullException("body");
}
_state = state;
_body = body;
}
/// <summary>
/// Runs the actual task.
/// </summary>
protected override void DoCompute()
{
Result = _body(_state);
}
}
}

13
src/Numerics/Threading/ThreadQueue.cs

@ -94,7 +94,7 @@ namespace MathNet.Numerics.Threading
/// <summary>
/// Gets a value indicating whether the current thread is a parallelized worker thread.
/// </summary>
internal static bool IsInWorkerThread
public static bool IsInWorkerThread
{
get { return _isInWorkerThread; }
}
@ -103,7 +103,7 @@ namespace MathNet.Numerics.Threading
/// Add a job to the queue.
/// </summary>
/// <param name="task">The job to run.</param>
internal static void Enqueue(Task task)
public static void Enqueue(Task task)
{
if (!_running)
{
@ -122,7 +122,7 @@ namespace MathNet.Numerics.Threading
/// Add a set of jobs to the queue.
/// </summary>
/// <param name="tasks">The jobs to run.</param>
internal static void Enqueue(IList<Task> tasks)
public static void Enqueue(IList<Task> tasks)
{
if (!_running)
{
@ -176,7 +176,6 @@ namespace MathNet.Numerics.Threading
// ...and run it
task.Compute();
//task.Set();
}
}
@ -184,7 +183,7 @@ namespace MathNet.Numerics.Threading
/// Start or restart the queue with the specified number of worker threads.
/// </summary>
/// <param name="numberOfThreads">Number of worker threads.</param>
internal static void Start(int numberOfThreads)
public static void Start(int numberOfThreads)
{
// instead of throwing an out of range exception, simply normalize
numberOfThreads = Math.Max(1, Math.Min(1024, numberOfThreads));
@ -209,7 +208,7 @@ namespace MathNet.Numerics.Threading
/// <summary>
/// Start the thread queue, if it is not already running.
/// </summary>
internal static void Start()
public static void Start()
{
lock (_stateSync)
{
@ -237,7 +236,7 @@ namespace MathNet.Numerics.Threading
/// <summary>
/// Stop the thread queue, if it is running.
/// </summary>
internal static void Shutdown()
public static void Shutdown()
{
lock (_stateSync)
{

Loading…
Cancel
Save