Added unit tests for parallel background worker; fixed some issues with the parallel background worker; added failing unit test for almost equal checks with doubles; fixed a typo

git-svn-id: file:///srv/devel/repo-conversion/nusu@291 d2e56fa2-650e-0410-a79f-9358c0239efd

Nuclex.Support (net-4.0).csproj

@@ -215,6 +215,7 @@
       <DependentUpon>LicenseKey.cs</DependentUpon>
     </Compile>
     <Compile Include="Source\ParallelBackgroundWorker.cs" />
+    <Compile Include="Source\ParallelBackgroundWorker.Test.cs" />
     <Compile Include="Source\Parsing\CommandLine.Argument.cs">
       <DependentUpon>CommandLine.cs</DependentUpon>
     </Compile>

Source/FloatHelper.Test.cs

@@ -148,7 +148,12 @@ namespace Nuclex.Support {
     // If both are negative -> fine
     // If both are positive -> fine
     // If different -> Measure both distances to zero in ulps and sum them
-    public void NegativeZeroEqualsPositiveZero() {
+    /// <summary>
+    ///   Verifies that the negative floating point zero is within one ulp of the positive
+    ///   floating point zero and vice versa
+    /// </summary>
+    [Test]
+    public void NegativeZeroFloatEqualsPositiveZero() {
       float zero = 0.0f;
       float zeroPlusOneUlp = FloatHelper.ReinterpretAsFloat(
         FloatHelper.ReinterpretAsInt(zero) + 1
@@ -163,6 +168,26 @@ namespace Nuclex.Support {
       Assert.IsTrue(FloatHelper.AreAlmostEqual(zero, zeroMinusOneUlp, 1));
     }
 
+    /// <summary>
+    ///   Verifies that the negative double precision floating point zero is within one ulp
+    ///   of the positive double precision floating point zero and vice versa
+    /// </summary>
+    [Test]
+    public void NegativeZeroDoubleEqualsPositiveZero() {
+      double zero = 0.0;
+      double zeroPlusOneUlp = FloatHelper.ReinterpretAsDouble(
+        FloatHelper.ReinterpretAsLong(zero) + 1
+      );
+      double zeroMinusOneUlp = -zeroPlusOneUlp;
+
+      bool test = FloatHelper.AreAlmostEqual(zeroMinusOneUlp, zeroPlusOneUlp, 1);
+
+      Assert.IsFalse(FloatHelper.AreAlmostEqual(zero, zeroPlusOneUlp, 0));
+      Assert.IsTrue(FloatHelper.AreAlmostEqual(zero, zeroPlusOneUlp, 1));
+      Assert.IsFalse(FloatHelper.AreAlmostEqual(zero, zeroMinusOneUlp, 0));
+      Assert.IsTrue(FloatHelper.AreAlmostEqual(zero, zeroMinusOneUlp, 1));
+    }
+
   }
 
 } // namespace Nuclex.Support

Source/IntegerHelper.cs

@@ -77,7 +77,7 @@ namespace Nuclex.Support {
       return value;
     }
 
-    /// <summary>Returns the number of bits set in an </summary>
+    /// <summary>Returns the number of bits set in an integer</summary>
     /// <param name="value">Value whose bits will be counted</param>
     /// <returns>The number of bits set in the integer</returns>
     public static int CountBits(this int value) {

Source/ParallelBackgroundWorker.Test.cs

@@ -0,0 +1,239 @@
+#region CPL License
+/*
+Nuclex Framework
+Copyright (C) 2002-2013 Nuclex Development Labs
+
+This library is free software; you can redistribute it and/or
+modify it under the terms of the IBM Common Public License as
+published by the IBM Corporation; either version 1.0 of the
+License, or (at your option) any later version.
+
+This library is distributed in the hope that it will be useful,
+but WITHOUT ANY WARRANTY; without even the implied warranty of
+MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+IBM Common Public License for more details.
+
+You should have received a copy of the IBM Common Public
+License along with this library
+*/
+#endregion
+
+using System;
+using System.IO;
+
+#if UNITTEST
+
+using NUnit.Framework;
+using System.Threading;
+using System.Collections.Generic;
+
+namespace Nuclex.Support {
+
+  /// <summary>Unit Test for the parallel background worker class</summary>
+  [TestFixture]
+  public class ParallelBackgroundWorkerTest {
+
+    private class TestWorker : ParallelBackgroundWorker<object> {
+
+      /// <summary>Initializes a new parallel background worker with unlimited threads</summary>
+      public TestWorker() : base() { }
+
+      /// <summary>
+      ///   Initializes a new parallel background worker running the specified number
+      ///   of tasks in parallel
+      /// </summary>
+      /// <param name="threadCount">
+      ///   Number of tasks to run in parallel (if positive) or number of CPU cores to leave
+      ///   unused (if negative).
+      /// </param>
+      /// <remarks>
+      ///   If a negative number of threads is used, at least one thread will be always
+      ///   be created, so specifying -2 on a single-core system will still occupy
+      ///   the only core.
+      /// </remarks>
+      public TestWorker(int threadCount) : base(threadCount) {}
+
+      /// <summary>
+      ///   Initializes a new parallel background worker that uses the specified name for
+      ///   its worker threads.
+      /// </summary>
+      /// <param name="name">Name that will be assigned to the worker threads</param>
+      public TestWorker(string name) : base(name) {}
+
+      /// <summary>
+      ///   Initializes a new parallel background worker that uses the specified name for
+      ///   its worker threads and running the specified number of tasks in parallel.
+      /// </summary>
+      /// <param name="name">Name that will be assigned to the worker threads</param>
+      /// <param name="threadCount">
+      ///   Number of tasks to run in parallel (if positive) or number of CPU cores to leave
+      ///   unused (if negative).
+      /// </param>
+      /// <remarks>
+      ///   If a negative number of threads is used, at least one thread will be always
+      ///   be created, so specifying -2 on a single-core system will still occupy
+      ///   the only core.
+      /// </remarks>
+      public TestWorker(string name, int threadCount) : base(name, threadCount) { }
+
+      /// <summary>Called in a thread to execute a single task</summary>
+      /// <param name="task">Task that should be executed</param>
+      /// <param name="cancellationToken">
+      ///   Cancellation token through which the method can be signalled to cancel
+      /// </param>
+      protected override void Run(object task, CancellationToken cancellationToken) {
+        if(this.ThrowException) {
+          throw new Exception("Something went wrong");
+        }
+
+        if(this.WaitEvent != null) {
+          this.WaitEvent.WaitOne();
+        }
+
+        this.WasCancelled = cancellationToken.IsCancellationRequested;
+
+        if(this.Tasks != null) {
+          this.Tasks.Add(task);
+        }
+      }
+
+      public bool ThrowException;
+      public ManualResetEvent WaitEvent;
+      public bool WasCancelled;
+      public ICollection<object> Tasks;
+
+    }
+
+    /// <summary>Verifies that the background worker has a default constructor</summary>
+    [Test]
+    public void CanBeDefaultConstructed() {
+      using(new TestWorker()) { }
+    }
+
+    /// <summary>
+    ///   Verifies that a background worker can be constructed that uses a fixed number
+    ///   of threads
+    /// </summary>
+    [Test]
+    public void CanUseFixedNumberOfThreads() {
+      using(new TestWorker(4)) { }
+    }
+
+    /// <summary>
+    ///   Verifies that a background worker can be constructed that leaves free a fixed
+    ///   number of CPU cores
+    /// </summary>
+    [Test]
+    public void CanPreserveFixedNumberOfCores() {
+      using(new TestWorker(-2)) { }
+    }
+
+    /// <summary>
+    ///   Verifies that a background worker can be constructed using a specific name
+    ///   for its worker threads
+    /// </summary>
+    [Test]
+    public void CanUseNamedThreads() {
+      using(new TestWorker("Test Task Thread")) { }
+    }
+
+    /// <summary>
+    ///   Verifies that a background worker can be constructed that uses a fixed number
+    ///   of threads using a specific name
+    /// </summary>
+    [Test]
+    public void CanUseFixedNumberOfNamedThreads() {
+      using(new TestWorker("Test Task Thread", 4)) { }
+    }
+
+    /// <summary>
+    ///   Verifies that a background worker can be constructed that leaves free a fixed
+    ///   number of CPU cores and uses a specific name for its worker threads.
+    /// </summary>
+    [Test]
+    public void CanPreserveFixedNumberOfCoresAndUseNamedThreads() {
+      using(new TestWorker("Test Task Thread", -2)) { }
+    }
+
+    /// <summary>
+    ///   Verifies that exceptions happening inside the tasks are collected and re-thrown
+    ///   in the Join() method.
+    /// </summary>
+    [Test]
+    public void ExceptionsAreReThrownInJoin() {
+      using(var testWorker = new TestWorker()) {
+        testWorker.ThrowException = true;
+        testWorker.AddTask(new object());
+        testWorker.AddTask(new object());
+
+        Assert.Throws<AggregateException>(
+          () => {
+            testWorker.Join();
+          }
+        );
+
+        try {
+          testWorker.Join();
+          Assert.Fail(
+            "Calling ParallelBackgroundWorker.Join() multiple times should re-throw " +
+            "exceptions multiple times"
+          );
+        }
+        catch(AggregateException aggregateException) {
+          Assert.AreEqual(2, aggregateException.InnerExceptions.Count);
+        }
+      }
+    }
+
+    /// <summary>
+    ///   Verifies that tasks can be cancelled while they are running
+    /// </summary>
+    [Test]
+    public void TasksCanBeCancelled() {
+      using(var waitEvent = new ManualResetEvent(false)) {
+        using(var testWorker = new TestWorker()) {
+          testWorker.WaitEvent = waitEvent;
+
+          testWorker.AddTask(new object());
+          testWorker.CancelRunningTasks();
+          
+          waitEvent.Set();
+
+          Assert.IsTrue(testWorker.Wait(1000));
+
+          Assert.IsTrue(testWorker.WasCancelled);
+        }
+      } // disposes waitEvent
+    }
+
+    /// <summary>Verifies that calling Join() waits for all queued tasks</summary>
+    [Test]
+    public void JoinWaitsForQueuedTasks() {
+      var tasks = new List<object>(100);
+      for(int index = 0; index < 100; ++index) {
+        tasks.Add(new object());
+      }
+
+      using(var waitEvent = new ManualResetEvent(false)) {
+        using(var testWorker = new TestWorker(2)) {
+          testWorker.WaitEvent = waitEvent;
+          testWorker.Tasks = new List<object>();
+          for(int index = 0; index < 100; ++index) {
+            testWorker.AddTask(tasks[index]);
+          }
+
+          CollectionAssert.IsEmpty(testWorker.Tasks);
+
+          waitEvent.Set();
+          testWorker.Join();
+
+          CollectionAssert.AreEquivalent(tasks, testWorker.Tasks);
+        }
+      } // disposes waitEvent
+    }
+
+  }
+
+} // namespace Nuclex.Support
+
+#endif // UNITTEST

Source/ParallelBackgroundWorker.cs

@@ -17,6 +17,12 @@ namespace Nuclex.Support {
     public static readonly int Processors = Environment.ProcessorCount;
 #endif
 
+    /// <summary>
+    ///   Timeout after which Dispose() will stop waiting for unfinished tasks and
+    ///   free resources anyway
+    /// </summary>
+    private static readonly int DefaultDisposeTimeoutMilliseconds = 1500; // milliseconds
+
     /// <summary>Initializes a new parallel background worker with unlimited threads</summary>
     public ParallelBackgroundWorker() : this(int.MaxValue) { }
 
@@ -40,11 +46,12 @@ namespace Nuclex.Support {
         threadCount = Math.Max(1, Processors + threadCount);
       }
 
-      this.runQueuedTasksInThreadDelegate = new Action<object>(runQueuedTasksInThread);
+      this.queueSynchronizationRoot = new object();
-      this.runningThreads = new List<Task>();
+      this.runQueuedTasksInThreadDelegate = new Action(runQueuedTasksInThread);
       this.tasks = new Queue<TTask>();
       this.threadTerminatedEvent = new AutoResetEvent(false);
       this.cancellationTokenSource = new CancellationTokenSource();
+      this.exceptions = new ConcurrentBag<Exception>();
     }
 
     /// <summary>
@@ -81,7 +88,8 @@ namespace Nuclex.Support {
       if(this.threadTerminatedEvent != null) {
         CancelPendingTasks();
         CancelRunningTasks();
-        Join();
+
+        Wait(DefaultDisposeTimeoutMilliseconds);
 
         this.threadTerminatedEvent.Dispose();
         this.threadTerminatedEvent = null;
@@ -92,7 +100,6 @@ namespace Nuclex.Support {
       }
     }
 
-
     /// <summary>Adds a task for processing by the background worker threads</summary>
     /// <param name="task">Task that will be processed in the background</param>
     public void AddTask(TTask task) {
@@ -100,26 +107,28 @@ namespace Nuclex.Support {
         return;
       }
 
+      bool needNewThread;
+
       lock(this.queueSynchronizationRoot) {
         this.tasks.Enqueue(task);
 
-        if(this.runningThreads.Count < this.threadCount) {
+        needNewThread = (this.runningThreadCount < this.threadCount);
-          //Task newThread = new Task(this.runQueuedTasksInThreadDelegate, );
+        if(needNewThread) {
+          ++this.runningThreadCount;
         }
       }
 
-      // Thread 1:
+      if(needNewThread) {
-      //     lock() {
+        Task newThread = new Task(
-      //       - take task
+          this.runQueuedTasksInThreadDelegate,
-      //       - or deregister and exit
+          // this.cancellationTokenSource.Token, // DO NOT PASS THIS!
-      //     }
+          // Passing the cancellation token makes tasks that have been queued but
-      //
+          // not started yet cease to execute at all - meaning our runningThreadCount
-      // Thread 2:
+          // goes out of sync and Dispose() / Wait() / Join() sit around endlessly!
-      //     lock() {
+          TaskCreationOptions.LongRunning
-      //       - put task
+        );
-      //       - if too few threads, register and add
+        newThread.Start();
-      //     }
+      }
-
     }
 
     /// <summary>Cancels all tasks that are currently executing</summary>
@@ -134,11 +143,36 @@ namespace Nuclex.Support {
       }
     }
 
-    /// <summary>Waits until all executing and queued tasks have been processed</summary>
+    /// <summary>
+    ///   Waits until all executing and queued tasks have been processed and throws an
+    ///   exception if any errors have occurred
+    /// </summary>
     public void Join() {
-      while(this.runningThreads.Count > 0) {
+      while(Thread.VolatileRead(ref this.runningThreadCount) > 0) {
         this.threadTerminatedEvent.WaitOne();
       }
+
+      if(this.exceptions.Count > 0) {
+        throw new AggregateException(this.exceptions);
+      }
+    }
+
+    /// <summary>
+    ///   Waits until all executing and queued tasks have been processed or
+    ///   the timeout is reached
+    /// </summary>
+    /// <param name="timeoutMilliseconds">Milliseconds after which the wait times out</param>
+    /// <returns>
+    ///   True if all tasks have been processed, false if the timeout was reached
+    /// </returns>
+    public bool Wait(int timeoutMilliseconds) {
+      while(Thread.VolatileRead(ref this.runningThreadCount) > 0) {
+        if(this.threadTerminatedEvent.WaitOne(timeoutMilliseconds) == false) {
+          return false;
+        }
+      }
+
+      return true;
     }
 
     /// <summary>Called in a thread to execute a single task</summary>
@@ -151,17 +185,23 @@ namespace Nuclex.Support {
     /// <summary>
     ///   Runs queued tasks of the parallel background worker until the queue is empty
     /// </summary>
-    /// <param name="thisTaskAsObject">Threading task in which this worker is running</param>
+    private void runQueuedTasksInThread() {
-    private void runQueuedTasksInThread(object thisTaskAsObject) {
       string previousThreadName = null;
       if(!string.IsNullOrEmpty(this.threadName)) {
         previousThreadName = Thread.CurrentThread.Name;
         Thread.CurrentThread.Name = this.threadName;
       }
       try {
-#if false
+        for(;;) {
-        TTask task;
+          TTask task;
-        while(this.tasks.TryDequeue(out task)) {
+          lock(this.queueSynchronizationRoot) {
+            if(this.tasks.Count == 0) {
+              --this.runningThreadCount;
+              break;
+            }
+            task = this.tasks.Dequeue();
+          }
+
           try {
             Run(task, this.cancellationTokenSource.Token);
           }
@@ -170,11 +210,7 @@ namespace Nuclex.Support {
           }
         }
 
-        lock(this.runningThreads) {
-          this.runningThreads.Remove((Task)thisTaskAsObject);
-        }
         this.threadTerminatedEvent.Set();
-#endif
       }
       finally {
         if(!string.IsNullOrEmpty(this.threadName)) {
@@ -236,19 +272,11 @@ namespace Nuclex.Support {
     private object queueSynchronizationRoot;
 
     /// <summary>Delegate for the runQueuedTasksInThread() method</summary>
-    private Action<object> runQueuedTasksInThreadDelegate;
+    private Action runQueuedTasksInThreadDelegate;
     /// <summary>Tasks remaining to be processed</summary>
     private Queue<TTask> tasks;
     /// <summary>Threads that are currently executing tasks</summary>
-    private IList<Task> runningThreads;
+    private int runningThreadCount;
-
-    // Idea:
-    // private int runningThreadCount;
-    // Before the task looks for new work, it will decrement this
-    // if the task gets new work, it will increment this again.
-    //   - if it would be above threadCount now, put work back in the queue
-    // AddTask() increments this after pushing new work
-    //   - if it would be above threadCount, do not create a new thread
 
     /// <summary>Exceptions that have occurred while executing tasks</summary>
     private ConcurrentBag<Exception> exceptions;