Workers.h

#ifndef XRPL_CORE_WORKERS_H_INCLUDED
#define XRPL_CORE_WORKERS_H_INCLUDED
 
#include <xrpld/core/detail/semaphore.h>
 
#include <xrpl/beast/core/LockFreeStack.h>
 
#include <atomic>
#include <condition_variable>
#include <mutex>
#include <string>
#include <thread>
 
namespace ripple {
 
namespace perf {
class PerfLog;
}
 
class Workers
{
public:
    struct Callback
    {
        virtual ~Callback() = default;
        Callback() = default;
        Callback(Callback const&) = delete;
        Callback&
        operator=(Callback const&) = delete;
 
        virtual void
        processTask(int instance) = 0;
    };
 
    explicit Workers(
        Callback& callback,
        perf::PerfLog* perfLog,
        std::string const& threadNames = "Worker",
        int numberOfThreads =
            static_cast<int>(std::thread::hardware_concurrency()));
 
    ~Workers();
 
    int
    getNumberOfThreads() const noexcept;
 
    void
    setNumberOfThreads(int numberOfThreads);
 
    void
    stop();
 
    void
    addTask();
 
    int
    numberOfCurrentlyRunningTasks() const noexcept;
 
    //--------------------------------------------------------------------------
 
private:
    struct PausedTag
    {
        explicit PausedTag() = default;
    };
 
    /*  A Worker executes tasks on its provided thread.
 
        These are the states:
 
        Active: Running the task processing loop.
        Idle:   Active, but blocked on waiting for a task.
        Paused: Blocked waiting to exit or become active.
    */
    class Worker : public beast::LockFreeStack<Worker>::Node,
                   public beast::LockFreeStack<Worker, PausedTag>::Node
    {
    public:
        Worker(
            Workers& workers,
            std::string const& threadName,
            int const instance);
 
        ~Worker();
 
        void
        notify();
 
    private:
        void
        run();
 
    private:
        Workers& m_workers;
        std::string const threadName_;
        int const instance_;
 
        std::thread thread_;
        std::mutex mutex_;
        std::condition_variable wakeup_;
        int wakeCount_;  // how many times to un-pause
        bool shouldExit_;
    };
 
private:
    static void
    deleteWorkers(beast::LockFreeStack<Worker>& stack);
 
private:
    Callback& m_callback;
    perf::PerfLog* perfLog_;
    std::string m_threadNames;     // The name to give each thread
    std::condition_variable m_cv;  // signaled when all threads paused
    std::mutex m_mut;
    bool m_allPaused;
    semaphore m_semaphore;           // each pending task is 1 resource
    int m_numberOfThreads;           // how many we want active now
    std::atomic<int> m_activeCount;  // to know when all are paused
    std::atomic<int> m_pauseCount;   // how many threads need to pause now
    std::atomic<int>
        m_runningTaskCount;  // how many calls to processTask() active
    beast::LockFreeStack<Worker> m_everyone;  // holds all created workers
    beast::LockFreeStack<Worker, PausedTag>
        m_paused;  // holds just paused workers
};
 
}  // namespace ripple
 
#endif