added more tests

fixed several issues in PeriodicScheduler

added cmake utils

added options for tsan and asan

fixed alias library

Author: BestITUserEUW

CMakeLists.txt

@@ -2,9 +2,19 @@ cmake_minimum_required(VERSION 3.24)
 
 project(oryx-crt-cpp VERSION 0.1.0 LANGUAGES CXX)
 
+message(STATUS "Build oryx-crt-cpp: ${PROJECT_VERSION}")
+
+include(cmake/utils.cmake)
+
 option(ORYX_CRT_BUILD_SHARED_LIBS "Build shared library" ${BUILD_SHARED_LIBS})
 option(ORYX_CRT_BUILD_TESTS "Build tests" OFF)
 option(ORYX_CRT_INSTALL "Install the project" ${PROJECT_IS_TOP_LEVEL})
+option(ORYX_SANITIZE_ADDRESS "Enable address sanitizer in tests" OFF)
+option(ORYX_SANITIZE_THREAD "Enable thread sanitizer in tests" OFF)
+
+if(ORYX_SANITIZE_ADDRESS AND ORYX_SANITIZE_THREAD)
+    message(FATAL_ERROR "ORYX_SANITIZE_ADDRESS and ORYX_SANITIZE_THREAD are mutually exclusive")
+endif()
 
 set(CMAKE_CXX_STANDARD_REQUIRED ON)
 if(NOT DEFINED CMAKE_CXX_STANDARD)
@@ -19,7 +29,7 @@ else()
     add_library(${PROJECT_NAME} STATIC)
 endif()
 
-add_library("${PROJECT_NAME}::${PROJECT_NAME}" ALIAS ${PROJECT_NAME})
+add_library("oryx::${PROJECT_NAME}" ALIAS ${PROJECT_NAME})
 
 if(CMAKE_CXX_COMPILER_ID STREQUAL "MSVC")
     target_compile_options(${PROJECT_NAME} 
@@ -59,6 +69,12 @@ target_sources(${PROJECT_NAME}
         FILES ${ORYX_CRT_HEADERS}
 )
 
+if(ORYX_SANITIZE_ADDRESS)
+    oryx_enable_addr_sanitizer(${PROJECT_NAME})
+elseif(ORYX_SANITIZE_THREAD)
+    oryx_enable_thread_sanitizer(${PROJECT_NAME})
+endif()
+
 
 if(ORYX_CRT_BUILD_TESTS)
     file(GLOB_RECURSE TEST_SOURCES "${CMAKE_CURRENT_SOURCE_DIR}/tests/*.cpp")

cmake/utils.cmake

@@ -0,0 +1,22 @@
+function(oryx_enable_addr_sanitizer target_name)
+    if(NOT CMAKE_CXX_COMPILER_ID MATCHES "GNU|Clang")
+        message(FATAL_ERROR "Sanitizer supported only for gcc/clang!")
+    endif()
+    message(STATUS "Address sanitizer enabled!")
+    target_compile_options(${target_name} PRIVATE -fsanitize=address,undefined)
+    target_compile_options(${target_name} PRIVATE -fno-sanitize=signed-integer-overflow)
+    target_compile_options(${target_name} PRIVATE -fno-sanitize-recover=all)
+    target_compile_options(${target_name} PRIVATE -fno-omit-frame-pointer)
+    target_link_libraries(${target_name} PRIVATE -fsanitize=address,undefined)
+endfunction()
+
+# Enable thread sanitizer (gcc/clang only)
+function(oryx_enable_thread_sanitizer target_name)
+    if(NOT CMAKE_CXX_COMPILER_ID MATCHES "GNU|Clang")
+        message(FATAL_ERROR "Sanitizer supported only for gcc/clang!")
+    endif()
+    message(STATUS "Thread sanitizer enabled!")
+    target_compile_options(${target_name} PRIVATE -fsanitize=thread)
+    target_compile_options(${target_name} PRIVATE -fno-omit-frame-pointer)
+    target_link_libraries(${target_name} PRIVATE -fsanitize=thread)
+endfunction()

include/oryx/error.hpp

@@ -19,4 +19,4 @@ class Error {
     std::string what_;
 };
 
-};  // namespace oryx
+}  // namespace oryx

include/oryx/periodic_scheduler.hpp

@@ -1,15 +1,15 @@
 #pragma once
 
-#include <atomic>
 #include <limits>
-#include <optional>
 #include <string>
 #include <memory>
 #include <chrono>
+#include <thread>
 #include <utility>
 #include <vector>
 #include <functional>
 #include <mutex>
+#include <stop_token>
 
 #include "thread_pool.hpp"
 
@@ -82,7 +82,7 @@ class PeriodicSchedulerImpl : public std::enable_shared_from_this<PeriodicSchedu
 
         void Reset() {
             id_ = kTaskIDMax;
-            scheduler_ = nullptr;
+            scheduler_.reset();
         }
 
         std::weak_ptr<PeriodicSchedulerImpl> scheduler_{};
@@ -91,21 +91,23 @@ class PeriodicSchedulerImpl : public std::enable_shared_from_this<PeriodicSchedu
 
     auto Schedule(TaskFn&& task, Duration interval, TaskStopPolicy stop_policy = TaskStopPolicy::kWaitForCompletion)
         -> TaskHandle {
-        TaskID id = task_counter_.fetch_add(1, std::memory_order_relaxed);
-        {
-            std::lock_guard lock{tasks_mtx_};
-            tasks_.emplace_back(std::move(task), id, stop_policy, interval, std::future<void>{}, Clock::now());
-        }
+        std::unique_lock lock{mtx_};
+        TaskID id = task_counter_++;
+        tasks_.emplace_back(std::move(task), id, stop_policy, interval, std::future<void>(), Clock::now());
+        lock.unlock();
         cv_.notify_all();
         return TaskHandle(std::enable_shared_from_this<PeriodicSchedulerImpl<Clock>>::weak_from_this(), id);
     }
 
     auto GetNumTasks() const -> size_t {
-        std::lock_guard lock{tasks_mtx_};
+        std::lock_guard lock{mtx_};
         return tasks_.size();
     }
 
-    auto TotalNumTasks() const -> size_t { return task_counter_.load(std::memory_order_relaxed); }
+    auto GetTaskCounter() const -> size_t {
+        std::lock_guard lock{mtx_};
+        return task_counter_;
+    }
 
     static auto Create(ThreadPool& pool) -> std::shared_ptr<PeriodicSchedulerImpl> {
         return std::shared_ptr<PeriodicSchedulerImpl>(new PeriodicSchedulerImpl(pool));
@@ -126,46 +128,43 @@ class PeriodicSchedulerImpl : public std::enable_shared_from_this<PeriodicSchedu
     PeriodicSchedulerImpl(ThreadPool& pool)
         : pool_(pool),
           tasks_(),
-          tasks_mtx_(),
+          mtx_(),
           cv_(),
           task_counter_(),
           worker_(&PeriodicSchedulerImpl<Clock>::ScheduleLoop, this) {}
 
     auto StopTask(TaskHandle& handle) -> bool {
-        std::optional<Task> task{};
-        {
-            std::lock_guard lock{tasks_mtx_};
-            auto it = std::ranges::find_if(tasks_, [&handle](const auto& t) { return handle.id_ == t.id; });
-            if (it != tasks_.end()) {
-                task.emplace(std::move(*it));
-                tasks_.erase(it);
-            }
-        }
-
-        if (!task) {
+        std::unique_lock lock{mtx_};
+        auto it = std::ranges::find_if(tasks_, [&handle](const auto& task) { return handle.id_ == task.id; });
+        if (it == tasks_.end()) {
             return false;
         }
 
-        if (task->stop_policy == TaskStopPolicy::kWaitForCompletion) {
-            if (task->promise.valid()) {
-                task->promise.wait();
+        auto task = std::move(*it);
+        tasks_.erase(it);
+        lock.unlock();
+
+        if (task.stop_policy == TaskStopPolicy::kWaitForCompletion) {
+            if (task.promise.valid()) {
+                task.promise.wait();
             }
         }
-
         handle.Reset();
         return true;
     }
 
     void ScheduleLoop(const std::stop_token& stoken) {
+        std::stop_callback scb{stoken, [this] { cv_.notify_all(); }};
+
         while (!stoken.stop_requested()) {
-            std::unique_lock lock{tasks_mtx_};
+            std::unique_lock lock{mtx_};
             if (tasks_.empty()) {
-                cv_.wait(lock, stoken, [&] { return stoken.stop_requested() || !tasks_.empty(); });
+                cv_.wait(lock, stoken, [this] { return !tasks_.empty(); });
                 continue;
             }
 
             auto now = Clock::now();
-            TimePoint next_wake_time = TimePoint::max();
+            auto next_wake_time = TimePoint::max();
             for (Task& task : tasks_) {
                 if (now >= task.next_execution) {
                     bool is_running = task.promise.valid() &&
@@ -180,17 +179,17 @@ class PeriodicSchedulerImpl : public std::enable_shared_from_this<PeriodicSchedu
             }
 
             if (next_wake_time != TimePoint::max()) {
-                cv_.wait_until(lock, stoken, next_wake_time,
-                               [&] { return stoken.stop_requested() || !tasks_.empty(); });
+                auto id = task_counter_;
+                cv_.wait_until(lock, stoken, next_wake_time, [this, id]() { return id < task_counter_; });
             }
         }
     }
 
     ThreadPool& pool_;
     std::vector<Task> tasks_;
-    std::mutex tasks_mtx_;
+    mutable std::mutex mtx_;
     std::condition_variable_any cv_;
-    std::atomic<TaskID> task_counter_;
+    TaskID task_counter_;
     std::jthread worker_;
 };
 

tests/chrono_mock_clock.hpp

@@ -1,20 +1,39 @@
 #pragma once
 
 #include <chrono>
+#include <mutex>
 
 namespace oryx {
 
 struct ChronoMockClock {
-    using duration = std::chrono::nanoseconds;
+public:
+    using duration = std::chrono::milliseconds;
     using rep = duration::rep;
     using period = duration::period;
-    using time_point = std::chrono::time_point<std::chrono::steady_clock, duration>;
+    using time_point = std::chrono::time_point<ChronoMockClock>;
 
     static constexpr bool is_steady = true;
 
-    static auto now() noexcept -> time_point { return now_; }
-
-    static inline time_point now_ = {};
+    static auto now() noexcept -> time_point {
+        std::lock_guard lock{mtx_};
+        return current_time;
+    }
+    static void set_time(time_point t) {
+        std::lock_guard lock{mtx_};
+        current_time = t;
+    }
+    static void advance(duration d) {
+        std::lock_guard lock{mtx_};
+        current_time += d;
+    }
+    static void reset() {
+        std::lock_guard lock{mtx_};
+        current_time = time_point{};
+    }
+
+private:
+    static inline std::mutex mtx_{};
+    static inline time_point current_time{};
 };
 
 static_assert(std::chrono::is_clock_v<ChronoMockClock>, "ChronoMockClock does not satisfy chrono clock!");