Add unit tests for Job classes, ThreadPool, and ScopedBoolInverter (#1)

* Add unit tests for ThreadPool
* Add unit tests for ScopedBoolInverter
* Fix coverage report generator config
* Add tests for and improve Job classes
* Remove unnecessary includes of ThreadPool.h
* Rename ThreadPool::execute to ::enqueue
* Update NOTICE-3RD-PARTY-CONTENT.md

Co-authored-by: Dominic Sudy <[email protected]>

Author: BjoernAtBosch

.pre-commit-config.yaml

@@ -22,6 +22,7 @@ repos:
             "--std=c++17",
             "--error-exitcode=1",
             "--enable=all",
+            "--inline-suppr",
             "--suppress=noExplicitConstructor",
             "--suppress=missingInclude",
             "--suppress=unusedFunction",

examples/vehicle_model/Cabin/SeatService/SeatService.cpp

@@ -17,7 +17,6 @@
 #include "SeatService.h"
 #include "sdk/Exceptions.h"
 #include "sdk/Logger.h"
-#include "sdk/ThreadPool.h"
 
 #include "sdk/dapr/DaprSupport.h"
 #include "vehicle_model/Cabin/SeatService/SeatServiceAsyncGrpcFacade.h"

gcovr.cfg

@@ -1,6 +1,8 @@
-filter = src
+filter = app
 filter = examples
-filter = include
+filter = sdk
+exclude = .*/vehicle_model/
+exclude = .*/tests/
 
 cobertura = build/coverage.cobertura.xml
 html = build/coverage.html

sdk/include/sdk/Job.h

@@ -17,19 +17,15 @@
 #ifndef VEHICLE_APP_SDK_JOB_H
 #define VEHICLE_APP_SDK_JOB_H
 
+#include <atomic>
 #include <functional>
 #include <memory>
 #include <mutex>
 
 namespace velocitas {
 
-class ThreadPool;
-class IJob;
-using JobPtr_t = std::shared_ptr<IJob>;
-
 /**
  * @brief Interface for jobs which can be executed by a worker in the thread pool.
- *
  */
 class IJob {
 public:
@@ -38,59 +34,65 @@ class IJob {
 
     /**
      * @brief Execute the job.
+     */
+    virtual void execute() = 0;
+
+    /**
+     * @brief Indicates if this job shall recur after its execution.
      *
-     * @param thisJobPtr    A smart pointer to this job. Allows sub classes to i.e. re-trigger the
-     *                      very same job instance.
-     * @param pool          The pool that is executing the job. Allows sub classes to i.e.
-     *                      re-trigger the same job.
+     * @return true - recur this job
+     * @return false - don't recure
      */
-    virtual void execute(JobPtr_t& thisJobPtr, ThreadPool& pool) = 0; // NOLINT
+    [[nodiscard]] virtual bool shallRecur() const { return false; }
 
     IJob(const IJob&)            = delete;
     IJob(IJob&&)                 = delete;
     IJob& operator=(const IJob&) = delete;
     IJob& operator=(IJob&&)      = delete;
 };
 
+using JobPtr_t = std::shared_ptr<IJob>;
+
 /**
  * @brief A nonrecurring job.
- *
  */
 class Job : public IJob {
 public:
     static JobPtr_t create(std::function<void()> fun) { return std::make_shared<Job>(fun); }
 
     explicit Job(std::function<void()> fun);
 
-    void execute(JobPtr_t& /*thisJobPtr*/, ThreadPool& /*pool*/) override;
-
-    std::function<void()> getFunction() { return m_fun; }
+    void execute() override;
 
-    void waitForTermination();
+    void waitForTermination() const;
 
 private:
     std::function<void()> m_fun;
-    std::mutex            m_terminationMutex;
+    mutable std::mutex    m_terminationMutex;
 };
 
 /**
  * @brief A recurring job which can be cancelled manually.
- *
  */
 class RecurringJob : public Job {
 public:
     static JobPtr_t create(std::function<void()> fun) {
         return std::make_shared<RecurringJob>(fun);
     }
 
-    void execute(JobPtr_t& thisJobPtr, ThreadPool& pool) override;
+    using Job::Job;
+
+    void execute() override;
 
+    /**
+     * @brief Prevents execution of the function once called.
+     */
     void cancel() { m_isCancelled = true; }
 
-    using Job::Job;
+    [[nodiscard]] bool shallRecur() const override { return !m_isCancelled; }
 
 private:
-    bool m_isCancelled{false};
+    std::atomic_bool m_isCancelled{false};
 };
 
 } // namespace velocitas

sdk/include/sdk/ThreadPool.h

@@ -19,8 +19,8 @@
 
 #include "sdk/Job.h"
 
+#include <atomic>
 #include <condition_variable>
-#include <functional>
 #include <memory>
 #include <mutex>
 #include <queue>
@@ -35,9 +35,8 @@ namespace velocitas {
  */
 class ThreadPool final {
 public:
-    using Job_t = std::function<void()>;
-
     ThreadPool();
+    explicit ThreadPool(size_t numWorkerThreads);
 
     ~ThreadPool();
 
@@ -48,12 +47,14 @@ class ThreadPool final {
      */
     static std::shared_ptr<ThreadPool> getInstance();
 
+    [[nodiscard]] size_t getNumWorkerThreads() const;
+
     /**
-     * @brief Execute the given job asynchronously in one of the worker threads.
+     * @brief Enqueue the given job to be executed asynchronously by one of the worker threads.
      *
      * @param job  The job to execute.
      */
-    void execute(JobPtr_t job);
+    void enqueue(JobPtr_t job);
 
     ThreadPool(const ThreadPool&)            = delete;
     ThreadPool(ThreadPool&&)                 = delete;
@@ -63,11 +64,11 @@ class ThreadPool final {
 private:
     void threadLoop();
 
-    std::mutex                                m_queueMutex;
-    std::condition_variable                   m_cv;
-    std::queue<JobPtr_t>                      m_jobs;
-    std::vector<std::unique_ptr<std::thread>> m_workerThreads;
-    bool                                      m_isRunning{true};
+    std::mutex               m_queueMutex;
+    std::condition_variable  m_cv;
+    std::queue<JobPtr_t>     m_jobs;
+    std::vector<std::thread> m_workerThreads;
+    std::atomic_bool         m_isRunning{true};
 };
 
 } // namespace velocitas

sdk/src/sdk/Job.cpp

@@ -15,27 +15,22 @@
  */
 
 #include "sdk/Job.h"
-#include "sdk/ThreadPool.h"
 
 namespace velocitas {
 
 Job::Job(std::function<void()> fun)
     : m_fun(std::move(fun)) {}
 
-void Job::waitForTermination() { std::scoped_lock lock(m_terminationMutex); }
+void Job::waitForTermination() const { std::lock_guard lock(m_terminationMutex); }
 
-void Job::execute(JobPtr_t& /*thisJobPtr*/, ThreadPool& /*pool*/) {
-    std::scoped_lock lock(m_terminationMutex);
+void Job::execute() {
+    std::lock_guard lock(m_terminationMutex);
     m_fun();
 }
 
-void RecurringJob::execute(JobPtr_t& thisJobPtr, ThreadPool& pool) {
+void RecurringJob::execute() {
     if (!m_isCancelled) {
-        Job::execute(thisJobPtr, pool);
-    }
-
-    if (!m_isCancelled) {
-        pool.execute(thisJobPtr);
+        Job::execute();
     }
 }
 

sdk/src/sdk/ThreadPool.cpp

@@ -18,42 +18,44 @@
 
 namespace velocitas {
 
-ThreadPool::ThreadPool() {
-    constexpr size_t numThreads{2};
-
-    for (size_t i = 0; i < numThreads; ++i) {
-        auto workerThread = std::make_unique<std::thread>([this]() { threadLoop(); });
-        m_workerThreads.push_back(std::move(workerThread));
+ThreadPool::ThreadPool(size_t numWorkerThreads)
+    : m_workerThreads{numWorkerThreads} {
+    for (size_t i = 0; i < numWorkerThreads; ++i) {
+        m_workerThreads[i] = std::thread([this]() { threadLoop(); });
     }
 }
 
+ThreadPool::ThreadPool()
+    : ThreadPool(2) {}
+
 ThreadPool::~ThreadPool() {
-    m_isRunning = false;
+    {
+        std::lock_guard lock{m_queueMutex};
+        m_isRunning = false;
+        // empty the job queue (std::queue does not offer a clear function)
+        std::queue<JobPtr_t>().swap(m_jobs);
+    }
     m_cv.notify_all();
 
     for (auto& thread : m_workerThreads) {
-        thread->join();
+        thread.join();
     }
 }
 
 std::shared_ptr<ThreadPool> ThreadPool::getInstance() {
-    static std::shared_ptr<ThreadPool> instance{nullptr};
-
-    if (!instance) {
-        instance = std::make_shared<ThreadPool>();
-    }
-
+    static std::shared_ptr<ThreadPool> instance{std::make_shared<ThreadPool>()};
     return instance;
 }
 
-void ThreadPool::execute(JobPtr_t job) {
+size_t ThreadPool::getNumWorkerThreads() const { return m_workerThreads.size(); }
+
+void ThreadPool::enqueue(JobPtr_t job) {
     std::lock_guard<std::mutex> lock(m_queueMutex);
     m_jobs.emplace(std::move(job));
-    m_cv.notify_all();
+    m_cv.notify_one();
 }
 
 void ThreadPool::threadLoop() {
-    std::mutex waitMutex;
     while (m_isRunning) {
         JobPtr_t job;
         {
@@ -65,10 +67,13 @@ void ThreadPool::threadLoop() {
         }
 
         if (job) {
-            job->execute(job, *this);
+            job->execute();
+            if (job->shallRecur()) {
+                enqueue(job);
+            }
         } else {
-            std::unique_lock<std::mutex> lock(waitMutex);
-            m_cv.wait(lock);
+            std::unique_lock<std::mutex> lock(m_queueMutex);
+            m_cv.wait(lock, [this] { return !m_jobs.empty() || !m_isRunning; });
         }
     }
 }

sdk/src/sdk/grpc/GrpcClient.cpp

@@ -22,7 +22,7 @@ namespace velocitas {
 
 GrpcClient::GrpcClient() {
     m_recurringJob = std::make_shared<RecurringJob>([this]() { pruneCompletedRequests(); });
-    ThreadPool::getInstance()->execute(m_recurringJob);
+    ThreadPool::getInstance()->enqueue(m_recurringJob);
 }
 
 GrpcClient::~GrpcClient() {

sdk/src/sdk/grpc/VehicleDataBrokerClient.cpp

@@ -24,7 +24,6 @@
 #include "sdk/Exceptions.h"
 #include "sdk/Job.h"
 #include "sdk/Logger.h"
-#include "sdk/ThreadPool.h"
 
 #include <fmt/core.h>
 #include <grpcpp/channel.h>

sdk/tests/utests/CMakeLists.txt

@@ -3,7 +3,10 @@ set(TARGET_NAME "sdk_utests")
 add_executable(${TARGET_NAME}
     AsyncResult_tests.cpp
     AsyncSubscription_tests.cpp
+    Job_tests.cpp
     Node_tests.cpp
+    ScopedBoolInverter_tests.cpp
+    ThreadPool_tests.cpp
     Utils_tests.cpp
     VehicleDataBrokerClient_tests.cpp
     #PubSub_tests.cpp