Browse Source

modified the thread queue to use wait/pulse instead of semaphores

la-knuth
Marcus Cuda 17 years ago
parent
commit
65ac0273f9
  1. 5
      build/build.proj
  2. 8
      src/Numerics/Threading/Parallel.cs
  3. 53
      src/Numerics/Threading/ThreadQueue.cs
  4. 2
      src/UnitTests/PrecisionTest.cs
  5. 3
      src/UnitTests/StatisticsTests/HistogramTests.cs

5
build/build.proj

@ -19,6 +19,11 @@
<Exec Command="$(MSBuildProjectDirectory)/../src/FSharpUnitTests/bin/Debug/FSharpUnitTests.exe" IgnoreExitCode="false" /> <Exec Command="$(MSBuildProjectDirectory)/../src/FSharpUnitTests/bin/Debug/FSharpUnitTests.exe" IgnoreExitCode="false" />
</Target> </Target>
<Target Name="TestLite" DependsOnTargets="Compile">
<Gallio.MSBuildTasks.Gallio ContinueOnError="False" IgnoreFailures="False" Assemblies="../src/UnitTests/bin/Debug/MathNet.Numerics.UnitTests.dll" />
<Exec Command="$(MSBuildProjectDirectory)/../src/FSharpUnitTests/bin/Debug/FSharpUnitTests.exe" IgnoreExitCode="false" />
</Target>
<Target Name="Style"> <Target Name="Style">
<CreateItem Include="../src/Numerics/**/*.cs"> <CreateItem Include="../src/Numerics/**/*.cs">
<Output TaskParameter="Include" ItemName="StyleCopFiles" /> <Output TaskParameter="Include" ItemName="StyleCopFiles" />

8
src/Numerics/Threading/Parallel.cs

@ -207,9 +207,9 @@ namespace MathNet.Numerics.Threading
WaitForTasksToComplete(tasks); WaitForTasksToComplete(tasks);
for (var i = 0; i < tasks.Length; i++) foreach (var t in tasks)
{ {
localFinally(tasks[i].Result); localFinally(t.Result);
} }
CollectExceptions(tasks); CollectExceptions(tasks);
@ -453,9 +453,9 @@ 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)
{ {
for (var i = 0; i < tasks.Length; i++) foreach (var task in tasks)
{ {
tasks[i].Wait(); task.Wait();
} }
} }

53
src/Numerics/Threading/ThreadQueue.cs

@ -52,16 +52,6 @@ namespace MathNet.Numerics.Threading
/// </summary> /// </summary>
private static readonly Queue<Task> _queue = new Queue<Task>(); private static readonly Queue<Task> _queue = new Queue<Task>();
/// <summary>
/// Maximum number of jobs that can be in the queue at the same time.
/// </summary>
private const int MaximumQueueLength = 4096;
/// <summary>
/// Counting Semaphore to make the worker thread wait for jobs
/// </summary>
private static Semaphore _tasksAvailableSemaphore;
/// <summary> /// <summary>
/// Running flag, used to signal worker threads to stop cleanly. /// Running flag, used to signal worker threads to stop cleanly.
/// </summary> /// </summary>
@ -113,9 +103,8 @@ namespace MathNet.Numerics.Threading
lock (_queueSync) lock (_queueSync)
{ {
_queue.Enqueue(task); _queue.Enqueue(task);
Monitor.Pulse(_queueSync);
} }
_tasksAvailableSemaphore.Release();
} }
/// <summary> /// <summary>
@ -135,9 +124,9 @@ namespace MathNet.Numerics.Threading
{ {
_queue.Enqueue(task); _queue.Enqueue(task);
} }
}
_tasksAvailableSemaphore.Release(tasks.Count); Monitor.PulseAll(_queueSync);
}
} }
/// <summary> /// <summary>
@ -149,13 +138,9 @@ namespace MathNet.Numerics.Threading
while (_running) while (_running)
{ {
// Wait until a job is available, or we should shut down
_tasksAvailableSemaphore.WaitOne();
// Check whether we should shut down // Check whether we should shut down
if (!_running) if (!_running)
{ {
_tasksAvailableSemaphore.Release();
break; break;
} }
@ -167,13 +152,17 @@ namespace MathNet.Numerics.Threading
{ {
task = _queue.Dequeue(); task = _queue.Dequeue();
} }
else
{
Monitor.Wait(_queueSync);
}
} }
if (task == null) if (task == null)
{ {
continue; continue;
} }
// ...and run it // ...and run it
task.Compute(); task.Compute();
} }
@ -185,11 +174,11 @@ namespace MathNet.Numerics.Threading
/// <param name="numberOfThreads">Number of worker threads.</param> /// <param name="numberOfThreads">Number of worker threads.</param>
public 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));
lock (_stateSync) lock (_stateSync)
{ {
// instead of throwing an out of range exception, simply normalize
numberOfThreads = Math.Max(1, Math.Min(1024, numberOfThreads));
if (_threads != null) if (_threads != null)
{ {
if (_threads.Length == numberOfThreads) if (_threads.Length == numberOfThreads)
@ -203,7 +192,7 @@ namespace MathNet.Numerics.Threading
ThreadCount = numberOfThreads; ThreadCount = numberOfThreads;
Start(); Start();
} }
} }
/// <summary> /// <summary>
/// Start the thread queue, if it is not already running. /// Start the thread queue, if it is not already running.
@ -217,8 +206,8 @@ namespace MathNet.Numerics.Threading
return; return;
} }
_tasksAvailableSemaphore = new Semaphore(_queue.Count, MaximumQueueLength); _running = true;
_running = true;
_threads = new Thread[ThreadCount]; _threads = new Thread[ThreadCount];
for (var i = 0; i < _threads.Length; i++) for (var i = 0; i < _threads.Length; i++)
@ -238,25 +227,27 @@ namespace MathNet.Numerics.Threading
/// </summary> /// </summary>
public static void Shutdown() public static void Shutdown()
{ {
// try to stop the worker threads cleanly
lock (_stateSync) lock (_stateSync)
{ {
if (_threads == null) if (_threads == null)
{ {
return; return;
} }
// try to stop the worker threads cleanly
_running = false; _running = false;
_tasksAvailableSemaphore.Release();
lock (_queueSync)
{
Monitor.PulseAll(_queueSync);
}
// wait until all threads have stopped // wait until all threads have stopped
foreach (var thread in _threads) foreach (var thread in _threads)
{ {
thread.Join(); thread.Join();
} }
_tasksAvailableSemaphore.Close();
_tasksAvailableSemaphore = null;
_threads = null; _threads = null;
} }
} }

2
src/UnitTests/PrecisionTest.cs

@ -184,8 +184,6 @@ namespace MathNet.Numerics.UnitTests
public void CoerceZero() public void CoerceZero()
{ {
Assert.AreEqual<double>(0.0, Precision.CoerceZero(0d)); Assert.AreEqual<double>(0.0, Precision.CoerceZero(0d));
Console.WriteLine(0.0.EpsilonOf());
Console.WriteLine(Precision.Increment(0.0));
Assert.AreEqual<double>(0.0, Precision.CoerceZero(Precision.Increment(0.0))); Assert.AreEqual<double>(0.0, Precision.CoerceZero(Precision.Increment(0.0)));
Assert.AreEqual<double>(0.0, Precision.CoerceZero(Precision.Decrement(0.0))); Assert.AreEqual<double>(0.0, Precision.CoerceZero(Precision.Decrement(0.0)));

3
src/UnitTests/StatisticsTests/HistogramTests.cs

@ -269,11 +269,8 @@ namespace MathNet.Numerics.UnitTests.StatisticsTests
Assert.AreEqual(9, hist.BucketCount); Assert.AreEqual(9, hist.BucketCount);
Console.WriteLine("{0}", hist);
for (int i = 1; i < 9; i++) for (int i = 1; i < 9; i++)
{ {
Console.WriteLine("{0} : {1}", i, hist[i].Count);
Assert.AreEqual(1.0, hist[i].Count); Assert.AreEqual(1.0, hist[i].Count);
} }
Assert.AreEqual(2.0, hist[0].Count); Assert.AreEqual(2.0, hist[0].Count);

Loading…
Cancel
Save