Workers.h

#pragma once
 
#include <xrpl/beast/core/LockFreeStack.h>
#include <xrpl/core/detail/semaphore.h>
 
#include <atomic>
#include <condition_variable>
#include <mutex>
#include <string>
#include <thread>
 
namespace xrpl {
 
namespace perf {
class PerfLog;
}  // namespace perf

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 threadNames = "Worker",
        int numberOfThreads = static_cast<int>(std::thread::hardware_concurrency()));
 
    ~Workers();

    [[nodiscard]] int
    getNumberOfThreads() const noexcept;

    void
    setNumberOfThreads(int numberOfThreads);

    void
    stop();

    void
    addTask();

    [[nodiscard]] 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 threadName, int const instance);
 
        ~Worker();
 
        void
        notify();
 
    private:
        void
        run();
 
    private:
        Workers& workers_;
        std::string const threadName_;
        int const instance_;
 
        std::thread thread_;
        std::mutex mutex_;
        std::condition_variable wakeup_;
        int wakeCount_{0};  // how many times to un-pause
        bool shouldExit_{false};
    };
 
private:
    static void
    deleteWorkers(beast::LockFreeStack<Worker>& stack);
 
private:
    Callback& callback_;
    perf::PerfLog* perfLog_;
    std::string threadNames_;     // The name to give each thread
    std::condition_variable cv_;  // signaled when all threads paused
    std::mutex mut_;
    bool allPaused_{true};
    Semaphore semaphore_;                             // each pending task is 1 resource
    int numberOfThreads_{0};                          // how many we want active now
    std::atomic<int> activeCount_;                    // to know when all are paused
    std::atomic<int> pauseCount_;                     // how many threads need to pause now
    std::atomic<int> runningTaskCount_;               // how many calls to processTask() active
    beast::LockFreeStack<Worker> everyone_;           // holds all created workers
    beast::LockFreeStack<Worker, PausedTag> paused_;  // holds just paused workers
};
 
}  // namespace xrpl