ThreadPool.h

Go to the documentation of this file.

/*
 * Copyright (c) 2022, Daniel Beßler
 * All rights reserved.
 *
 * This file is part of KnowRob, please consult
 * https://github.com/knowrob/knowrob for license details.
 */
 
#ifndef KNOWROB_THREAD_POOL_H_
#define KNOWROB_THREAD_POOL_H_
 
#include <queue>
#include <mutex>
#include <string>
#include <list>
#include <functional>
#include <condition_variable>
#include <atomic>
#include <iostream>
#include <thread>
 
namespace knowrob {
    class ThreadPool {
    public:
        // forward declarations
        class Runner;
 
        using ExceptionHandler = std::function<void(const std::exception &)>;
 
        explicit ThreadPool(uint32_t maxNumThreads);
 
        virtual ~ThreadPool();
 
        ThreadPool(const ThreadPool &) = delete;
 
        void shutdown();
 
        void pushWork(const std::shared_ptr<ThreadPool::Runner> &goal, ThreadPool::ExceptionHandler exceptionHandler);
 
        class Worker {
        public:
            explicit Worker(ThreadPool *thread_pool);
 
            ~Worker();
 
            Worker(const Worker &) = delete;
 
        protected:
            ThreadPool *threadPool_;
 
            std::atomic<bool> isTerminated_;
            std::atomic<bool> hasTerminateRequest_;
 
            std::thread thread_;
 
            void run();
 
            friend class ThreadPool;
        };
 
        class Runner {
        public:
            Runner();
 
            virtual ~Runner();
 
            Runner(const Runner &) = delete;
 
            void join();
 
            virtual void run() = 0;
 
            void stop(bool wait);
 
            [[nodiscard]] bool hasStopRequest() const { return hasStopRequest_; }
 
            [[nodiscard]] bool isTerminated() const { return isTerminated_; }
 
        protected:
            std::atomic<bool> isTerminated_;
            std::atomic<bool> isRunning_;
            std::atomic<bool> hasStopRequest_;
            std::mutex mutex_;
            std::condition_variable finishedCV_;
            ExceptionHandler exceptionHandler_;
 
            void runInternal();
 
            void setExceptionHandler(ExceptionHandler exceptionHandler) { exceptionHandler_ = exceptionHandler; }
 
            friend class ThreadPool::Worker;
 
            friend class ThreadPool;
        };
 
        class LambdaRunner : public Runner {
        public:
            using StopChecker = std::function<bool()>;
 
            explicit LambdaRunner(const std::function<void(const StopChecker&)> &fn) : fn_(fn) {}
 
            void run() override { fn_([&]{ return hasStopRequest(); }); }
 
        protected:
            std::function<void(const StopChecker&)> fn_;
        };
 
    protected:
        // is called to finalize each worker thread
        // Note: a virtual method is avoided as these cannot be called in destructors.
        std::function<void()> finalizeWorker_;
 
    private:
        // list of threads doing work
        std::list<Worker *> workerThreads_;
        // currently queued work that has not been associated to a worker yet
        std::queue<std::shared_ptr<ThreadPool::Runner>> workQueue_;
        // condition variable used to wake up worker after new work was queued
        std::condition_variable workCV_;
        mutable std::mutex workMutex_;
        // limit to this number of worker threads
        uint32_t maxNumThreads_;
        // number of terminated threads that are still in workerThreads_ list
        std::atomic_uint32_t numFinishedThreads_;
        // number of currently active workers
        std::atomic_uint32_t numActiveWorker_;
 
        // get work from queue
        std::shared_ptr<ThreadPool::Runner> popWork();
 
        // is called initially in each worker thread
        virtual bool initializeWorker() { return true; }
    };
 
    std::shared_ptr<ThreadPool> DefaultThreadPool();
}
 
#endif //KNOWROB_THREAD_POOL_H_