Address some issues found by the sanitizers (#139)

Santizers (asan, tsan, ubsan, msan etc) found some issues in the code.
Some of them were technically impossible but having clean sanitizer
output makes it easier to find problems in the future.

.github/workflows/gcc.yaml

@@ -91,6 +91,6 @@ jobs:
         run: ccache --show-stats
 
       - name: Test
-        timeout-minutes: 5
+        timeout-minutes: 2
         working-directory: build/tests
-        run: ctest --output-on-failure -j$(nproc) -E "dsl/UDP"
+        run: ctest --output-on-failure -E "dsl/UDP"

.github/workflows/macos.yaml

@@ -61,4 +61,4 @@ jobs:
       - name: Test
         timeout-minutes: 5
         working-directory: build/tests
-        run: ctest --output-on-failure -j$(sysctl -n hw.logicalcpu)
+        run: ctest --output-on-failure

.github/workflows/windows.yaml

@@ -80,4 +80,4 @@ jobs:
       - name: Test
         timeout-minutes: 5
         working-directory: build/tests
-        run: ctest --output-on-failure -j$(nproc)
+        run: ctest --output-on-failure

src/extension/IOController.hpp

@@ -132,7 +132,7 @@ namespace extension {
         /// The mutex that protects the tasks list
         std::mutex tasks_mutex;
         /// Whether or not the list of file descriptors is dirty compared to tasks
-        bool dirty = true;
+        std::atomic<bool> dirty{true};
         /// The list of events that are being watched
         std::vector<watcher_t> watches;
         /// The list of tasks that are waiting for IO events

src/extension/IOController_Posix.ipp

@@ -49,7 +49,7 @@ namespace extension {
         }
 
         // We just cleaned the list!
-        dirty = false;
+        dirty.store(false, std::memory_order_release);
     }
 
     void IOController::fire_event(Task& task) {
@@ -125,7 +125,7 @@ namespace extension {
             // There are no tasks for this!
             if (range.first == tasks.end()) {
                 // If this happens then our list is definitely dirty...
-                dirty = true;
+                dirty.store(true, std::memory_order_release);
             }
             else {
                 // Loop through our values
@@ -187,7 +187,7 @@ namespace extension {
                 std::sort(tasks.begin(), tasks.end());
 
                 // Let the poll command know that stuff happened
-                dirty = true;
+                dirty.store(true, std::memory_order_release);
                 bump();
             });
 
@@ -203,7 +203,7 @@ namespace extension {
             if (task != tasks.end()) {
                 // If the events we were processing included close remove it from the list
                 if ((task->processing_events & IO::CLOSE) != 0) {
-                    dirty = true;
+                    dirty.store(true, std::memory_order_release);
                     tasks.erase(task);
                 }
                 else {
@@ -244,7 +244,7 @@ namespace extension {
                 }
 
                 // Let the poll command know that stuff happened
-                dirty = true;
+                dirty.store(true, std::memory_order_release);
                 bump();
             });
 
@@ -257,7 +257,7 @@ namespace extension {
             // Stay in this reaction to improve the performance without going back/fourth between reactions
             if (running.load(std::memory_order_acquire)) {
                 // Rebuild the list if something changed
-                if (dirty) {
+                if (dirty.load(std::memory_order_acquire)) {
                     rebuild_list();
                 }
 

src/extension/IOController_Windows.ipp

@@ -63,7 +63,7 @@ namespace extension {
         }
 
         // We just cleaned the list!
-        dirty = false;
+        dirty.store(false, std::memory_order_release);
     }
 
     void IOController::fire_event(Task& task) {
@@ -124,7 +124,7 @@ namespace extension {
             }
             // If we can't find the event then our list is dirty
             else {
-                dirty = true;
+                dirty.store(true, std::memory_order_release);
             }
         }
     }
@@ -172,7 +172,7 @@ namespace extension {
 
                 // Add all the information to the list and mark the list as dirty, to sync with the list of events
                 tasks.insert(std::make_pair(event, Task{config.fd, config.events, config.reaction}));
-                dirty = true;
+                dirty.store(true, std::memory_order_release);
 
                 bump();
             });
@@ -191,7 +191,7 @@ namespace extension {
                 auto& task = it->second;
                 // If the events we were processing included close remove it from the list
                 if (task.processing_events & IO::CLOSE) {
-                    dirty = true;
+                    dirty.store(true, std::memory_order_release);
                     remove_task(tasks, it);
                 }
                 else {
@@ -220,7 +220,7 @@ namespace extension {
                 }
 
                 // Let the poll command know that stuff happened
-                dirty = true;
+                dirty.store(true, std::memory_order_release);
                 bump();
             });
 
@@ -232,7 +232,7 @@ namespace extension {
         on<Always>().then("IO Controller", [this] {
             while (running.load(std::memory_order_acquire)) {
                 // Rebuild the list if something changed
-                if (dirty) {
+                if (dirty.load(std::memory_order_acquire)) {
                     rebuild_list();
                 }
 

src/threading/scheduler/Pool.cpp

@@ -142,6 +142,7 @@ namespace threading {
         }
 
         bool Pool::is_idle() const {
+            const std::lock_guard<std::mutex> lock(mutex);
             return pool_idle != nullptr;
         }
 

src/threading/scheduler/Pool.hpp

@@ -213,7 +213,7 @@ namespace threading {
             /// A boolean which is set to true when the queue is modified and set to false when there was no work to do
             bool live = true;
             /// The mutex which protects the queue and idle tasks
-            std::mutex mutex;
+            mutable std::mutex mutex;
             /// The condition variable which threads wait on if they can't get a task
             std::condition_variable condition;
 

tests/tests/api/TimeTravelFrozen.cpp

@@ -9,10 +9,11 @@ constexpr std::chrono::milliseconds EVENT_1_TIME  = std::chrono::milliseconds(4)
 constexpr std::chrono::milliseconds EVENT_2_TIME  = std::chrono::milliseconds(8);
 constexpr std::chrono::milliseconds SHUTDOWN_TIME = std::chrono::milliseconds(12);
 
-struct WaitForShutdown {};
 
 class TestReactor : public test_util::TestBase<TestReactor> {
 public:
+    struct WaitForShutdown {};
+
     TestReactor(std::unique_ptr<NUClear::Environment> environment) : TestBase(std::move(environment), false) {
 
         on<Startup>().then([this] {
@@ -22,7 +23,7 @@ class TestReactor : public test_util::TestBase<TestReactor> {
             // Emit a chrono task to run at time EVENT_1_TIME
             emit<Scope::DIRECT>(std::make_unique<NUClear::dsl::operation::ChronoTask>(
                 [this](NUClear::clock::time_point&) {
-                    events.push_back("Event 1");
+                    add_event("Event 1");
                     return false;
                 },
                 NUClear::clock::time_point(EVENT_1_TIME),
@@ -31,7 +32,7 @@ class TestReactor : public test_util::TestBase<TestReactor> {
             // Emit a chrono task to run at time EVENT_2_TIME
             emit<Scope::DIRECT>(std::make_unique<NUClear::dsl::operation::ChronoTask>(
                 [this](NUClear::clock::time_point&) {
-                    events.push_back("Event 2");
+                    add_event("Event 2");
                     return false;
                 },
                 NUClear::clock::time_point(EVENT_2_TIME),
@@ -47,7 +48,7 @@ class TestReactor : public test_util::TestBase<TestReactor> {
 
         on<Trigger<WaitForShutdown>>().then([this] {
             std::this_thread::sleep_for(SHUTDOWN_TIME);
-            events.push_back("Finished");
+            add_event("Finished");
             powerplant.shutdown();
         });
     }
@@ -61,8 +62,15 @@ class TestReactor : public test_util::TestBase<TestReactor> {
     // Real-time factor
     double rtf = 1.0;
 
-    // Events
+    /// Events that occur during the test
+    std::mutex event_mutex;
     std::vector<std::string> events;
+
+private:
+    void add_event(const std::string& event) {
+        const std::lock_guard<std::mutex> lock(event_mutex);
+        events.emplace_back(event);
+    }
 };
 
 

tests/tests/dsl/Idle.cpp

@@ -41,12 +41,14 @@ class TestReactor : public test_util::TestBase<TestReactor> {
     template <int N>
     void do_step(const std::string& name) {
         std::this_thread::sleep_until(start_time + test_util::TimeUnit(N));
+
+        const std::lock_guard<std::mutex> lock(events_mutex);
         events.push_back(name + " " + std::to_string(N));
         emit(std::make_unique<Step<N + 1>>());
     }
 
     explicit TestReactor(std::unique_ptr<NUClear::Environment> environment)
-        : TestBase(std::move(environment), false, std::chrono::seconds(2)) {
+        : TestBase(std::move(environment), false, std::chrono::seconds(5)) {
 
         start_time = NUClear::clock::now();
 
@@ -98,6 +100,8 @@ class TestReactor : public test_util::TestBase<TestReactor> {
         });
     }
 
+    /// A mutex to protect the events vector
+    std::mutex events_mutex;
     /// A vector of events that have happened
     std::vector<std::string> events;
 

tests/tests/dsl/IdleSingle.cpp

@@ -120,7 +120,7 @@ TEST_CASE("Test that when a global idle trigger exists it is triggered only once
 
 
     // Convert the values to strings for easier comparison
-    auto error_points = [](const std::array<std::atomic<int>, n_loops>& arr) {
+    auto error_points = [&](const std::array<std::atomic<int>, n_loops>& arr) {
         std::map<int, int> hits;
         for (int i = 0; i < n_loops; ++i) {
             if (arr[i] != 1) {

tests/tests/dsl/IdleSingleGlobal.cpp

@@ -98,7 +98,7 @@ TEST_CASE("Test that when a global idle trigger exists it is triggered only once
     }
 
     // Convert the values to a list for comparison
-    auto error_points = [](const std::array<std::atomic<int>, n_loops>& arr) {
+    auto error_points = [&](const std::array<std::atomic<int>, n_loops>& arr) {
         std::map<int, int> hits;
         for (int i = 0; i < n_loops; ++i) {
             if (arr[i] != 1) {

tests/tests/dsl/IdleSync.cpp

@@ -37,21 +37,28 @@ class TestReactor : public test_util::TestBase<TestReactor> {
 
         // Idle testing for default thread
         on<Trigger<Step<2>>, Sync<TestReactor>>().then([this] {
-            events.push_back("Default Start");
+            add_event("Default Start");
             std::this_thread::sleep_for(std::chrono::milliseconds(300));
-            events.push_back("Default End");
+            add_event("Default End");
         });
 
-        on<Trigger<Step<3>>, Sync<TestReactor>, MainThread>().then([this] { events.push_back("Main Task"); });
+        on<Trigger<Step<3>>, Sync<TestReactor>, MainThread>().then([this] { add_event("Main Task"); });
 
         on<Idle<MainThread>>().then([this] {
-            events.push_back("Idle Main Thread");
+            add_event("Idle Main Thread");
             powerplant.shutdown();
         });
 
         on<Startup>().then([this] { emit(std::make_unique<Step<1>>()); });
     }
 
+    void add_event(const std::string& event) {
+        const std::lock_guard<std::mutex> lock(events_mutex);
+        events.emplace_back(event);
+    }
+
+    /// A mutex to protect the events vector
+    std::mutex events_mutex;
     /// A vector of events that have happened
     std::vector<std::string> events;
 };

tests/tests/dsl/Watchdog.cpp

@@ -26,47 +26,48 @@
 #include <string>
 
 #include "test_util/TestBase.hpp"
+#include "test_util/TimeUnit.hpp"
 
 class TestReactor : public test_util::TestBase<TestReactor> {
 public:
     template <int I>
     struct Flag {};
 
     TestReactor(std::unique_ptr<NUClear::Environment> environment)
-        : TestBase(std::move(environment), false), start(NUClear::clock::now()) {
+        : TestBase(std::move(environment), false, test_util::TimeUnit(40)), start(NUClear::clock::now()) {
 
-        on<Watchdog<Flag<1>, 50, std::chrono::milliseconds>>().then([this] {
-            events.push_back("Watchdog 1  triggered @ " + floored_time());
+        on<Watchdog<Flag<1>, 5, test_util::TimeUnit>>().then([this] {
+            events.push_back("Watchdog 1  triggered @ " + units_since_start());
             powerplant.shutdown();
         });
 
-        on<Watchdog<Flag<2>, 40, std::chrono::milliseconds>>().then([this] {
+        on<Watchdog<Flag<2>, 4, test_util::TimeUnit>>().then([this] {
             if (flag2++ < 3) {
-                events.push_back("Watchdog 2  triggered @ " + floored_time());
+                events.push_back("Watchdog 2  triggered @ " + units_since_start());
                 emit<Scope::WATCHDOG>(ServiceWatchdog<Flag<1>>());
             }
         });
 
         // Watchdog with subtypes
-        on<Watchdog<Flag<3>, 30, std::chrono::milliseconds>>('a').then([this] {
+        on<Watchdog<Flag<3>, 3, test_util::TimeUnit>>('a').then([this] {
             if (flag3a++ < 3) {
-                events.push_back("Watchdog 3A triggered @ " + floored_time());
+                events.push_back("Watchdog 3A triggered @ " + units_since_start());
                 emit<Scope::WATCHDOG>(ServiceWatchdog<Flag<1>>());
                 emit<Scope::WATCHDOG>(ServiceWatchdog<Flag<2>>());
             }
         });
-        on<Watchdog<Flag<3>, 20, std::chrono::milliseconds>>('b').then([this] {
+        on<Watchdog<Flag<3>, 2, test_util::TimeUnit>>('b').then([this] {
             if (flag3b++ < 3) {
-                events.push_back("Watchdog 3B triggered @ " + floored_time());
+                events.push_back("Watchdog 3B triggered @ " + units_since_start());
                 emit<Scope::WATCHDOG>(ServiceWatchdog<Flag<1>>());
                 emit<Scope::WATCHDOG>(ServiceWatchdog<Flag<2>>());
                 emit<Scope::WATCHDOG>(ServiceWatchdog<Flag<3>>('a'));
             }
         });
 
-        on<Watchdog<Flag<4>, 10, std::chrono::milliseconds>>().then([this] {
+        on<Watchdog<Flag<4>, 1, test_util::TimeUnit>>().then([this] {
             if (flag4++ < 3) {
-                events.push_back("Watchdog 4  triggered @ " + floored_time());
+                events.push_back("Watchdog 4  triggered @ " + units_since_start());
                 emit<Scope::WATCHDOG>(ServiceWatchdog<Flag<1>>());
                 emit<Scope::WATCHDOG>(ServiceWatchdog<Flag<2>>());
                 emit<Scope::WATCHDOG>(ServiceWatchdog<Flag<3>>('a'));
@@ -75,11 +76,8 @@ class TestReactor : public test_util::TestBase<TestReactor> {
         });
     }
 
-    std::string floored_time() const {
-        using namespace std::chrono;  // NOLINT(google-build-using-namespace) fine in function scope
-        const double diff = duration_cast<duration<double>>(NUClear::clock::now() - start).count();
-        // Round to 100ths of a second
-        return std::to_string(int(std::floor(diff * 100)));
+    std::string units_since_start() {
+        return std::to_string(test_util::round_to_test_units(NUClear::clock::now() - start).count());
     }
 
     NUClear::clock::time_point start;