A C++ Job Manager Class

Question

I've created a JobManager class that manages the execution of submitted Jobs on periodic intervals. While the JobManager is running, the API allows the user to:

• Add or Remove a Job to/from the JobManager
• Change a Job's interval, function parameter, and function
• Signal a Job to run if it is waiting

The testing I have done seems to denote that the class is working correctly. What stands out to you as a bad practice and or an inefficiency here? What potential problems do you see with using this? The Job could be used by itself but I haven't implemented the rule of 5 with it.

// The Job class
#pragma once
#include <chrono>
#include <any>
#include <thread>
#include <functional>
#include <condition_variable>
#include <mutex>
#include <algorithm>
#include <list>

namespace engine {

using job_clock = std::chrono::system_clock;
using job_func = std::function<void(const std::any &, bool)>;
using job_interval = std::chrono::system_clock::duration;
using job_time = std::chrono::system_clock::time_point;

class Job {
public:
    Job(uint32_t job_id, job_func &&func, const job_interval &interval,
        const std::any &param)
        : id(job_id), m_func(func), m_interval(interval), m_param(param),
          m_lock(std::make_shared<std::mutex>()),
          m_cv(std::make_shared<std::condition_variable>())
    {
    }

    void Start(bool now)
    {
        if (!m_thread) {
            m_thread = std::make_unique<std::thread>(
                [this](bool now) { Monitor(now); }, now);
        }
    }

    void Stop()
    {
        if (m_thread) {
            m_quit = true;
            if (m_waiting && m_cv) {
                m_cv->notify_one();
            }
            m_thread->join();
            m_thread = nullptr;
        }
    }

    void SetParam(const std::any &param)
    {
        if (m_thread && m_lock) {
            std::lock_guard<std::mutex> lk(*m_lock);
            m_param = param;
        }
        else {
            m_param = param;
        }
    }

    void SetInterval(job_interval interval)
    {
        if (m_thread && m_lock) {
            std::lock_guard<std::mutex> lk(*m_lock);
            m_interval = interval;
        }
        else {
            m_interval = interval;
        }
    }

    void SetFunc(job_func &&func)
    {
        if (m_thread && m_lock) {
            std::lock_guard<std::mutex> lk(*m_lock);
            m_func = func;
        }
        else {
            m_func = func;
        }
    }

    bool Signal()
    {
        if (m_waiting && m_cv) {
            m_signaled = true;
            m_cv->notify_one();
            return true;
        }
        return false;
    }

    [[nodiscard]] bool Active() const
    {
        return m_active;
    }

    [[nodiscard]] bool Waiting() const
    {
        return m_waiting;
    }

    const uint32_t id;

private:
    void Monitor(bool now)
    {
        m_active = true;

        if (now) {
            m_func(m_param, false);
        }

        while (!m_quit) {

            job_time time = job_clock::now() + m_interval;
            {
                if (m_lock && m_cv) {
                    m_waiting = true;
                    std::unique_lock<std::mutex> lk(*m_lock);
                    m_cv->wait_until(lk, time);
                    m_waiting = false;
                }
            }

            if (!m_quit && m_lock) {
                std::lock_guard<std::mutex> lk(*m_lock);
                m_func(m_param, m_signaled);
                m_signaled = false;
            }
        }

        m_active = false;
    }

    std::shared_ptr<std::mutex> m_lock{nullptr};
    std::shared_ptr<std::condition_variable> m_cv{nullptr};
    std::shared_ptr<std::thread> m_thread{nullptr};
    bool m_active{false};
    bool m_signaled{false};
    bool m_waiting{false};
    bool m_quit{false};

    job_func m_func;
    job_interval m_interval;
    std::any m_param;
};

} // namespace engine

// The JobManager class
#include <engine/Job.hpp>

namespace engine {

class JobManager final {
public:
    JobManager(std::initializer_list<Job> jobs) : m_jobs(jobs) {}
    JobManager() = default;
    JobManager &operator=(const JobManager &rhs) = delete;
    JobManager(const JobManager &rhs) = delete;
    JobManager &operator=(JobManager &&rhs) = delete;
    JobManager(JobManager &&rhs) = delete;
    ~JobManager()
    {
        Stop();
    }

    void Add(std::initializer_list<Job> jobs)
    {
        std::lock_guard<std::mutex> lk(m_lock);
        for (auto &job : jobs) {
            if (AddJobIfNonExistent(job)) {
                m_jobs.back().Start(false);
            }
        }
    }

    void Add(const Job &job, bool run_now = false)
    {
        std::lock_guard<std::mutex> lk(m_lock);
        if (AddJobIfNonExistent(job)) {
            m_jobs.back().Start(run_now);
        }
    }

    void Remove(uint32_t id)
    {
        std::lock_guard<std::mutex> lk(m_lock);
        auto job = FindJobByID(id);
        if (job != m_jobs.end()) {
            job->Stop();
            m_jobs.erase(job);
        }
    }

    void SetJobParam(uint32_t id, const std::any &param)
    {
        std::lock_guard<std::mutex> lk(m_lock);
        auto job = FindJobByID(id);
        if (job != m_jobs.end()) {
            job->SetParam(param);
        }
    }

    void SetJobInterval(uint32_t id, job_interval interval)
    {
        std::lock_guard<std::mutex> lk(m_lock);
        auto job = FindJobByID(id);
        if (job != m_jobs.end()) {
            job->SetInterval(interval);
        }
    }

    void SetJobFunc(uint32_t id, job_func &&func)
    {
        std::lock_guard<std::mutex> lk(m_lock);
        auto job = FindJobByID(id);
        if (job != m_jobs.end()) {
            job->SetFunc(std::move(func));
        }
    }

    void Signal(uint32_t id)
    {
        std::lock_guard<std::mutex> lk(m_lock);
        auto job = FindJobByID(id);
        if (job != m_jobs.end()) {
            job->Signal();
        }
    }

    void Start()
    {
        std::lock_guard<std::mutex> lk(m_lock);
        auto StartMonitoringJob = [](Job &job) { job.Start(false); };
        ForEachJob(StartMonitoringJob);
    }

    void Stop()
    {
        std::lock_guard<std::mutex> lk(m_lock);
        auto JoinAndBlock = [](Job &job) { job.Stop(); };
        ForEachJob(JoinAndBlock);
    }

private:
    std::list<Job> m_jobs;
    std::mutex m_lock;

    std::list<Job>::iterator FindJobByID(uint32_t id)
    {
        auto IDMatch = [id](const Job &j) { return j.id == id; };
        return std::find_if(m_jobs.begin(), m_jobs.end(), IDMatch);
    }

    bool AddJobIfNonExistent(const Job &job)
    {
        if (FindJobByID(job.id) == m_jobs.end()) {
            m_jobs.push_back(job);
            return true;
        }
        return false;
    }

    void ForEachJob(std::function<void(Job &)> predicate)
    {
        std::for_each(m_jobs.begin(), m_jobs.end(), predicate);
    }
};

} // namespace engine

#include <spdlog/spdlog.h>
#include <engine/job_manager.hpp>

#define _dbg(f, ...) spdlog::info(f, ##__VA_ARGS__)

uint32_t JOB1_ID = 1;
uint32_t JOB2_ID = 2;

struct Job1Info {
    int integer;
    std::string str;
};

struct Job2Info {
    int some_value;
    bool other_value;
};

void Job1(std::any param, bool signaled)
{
    static int called = 1;
    auto info = std::any_cast<Job1Info>(param);
    _dbg("Job1: signaled={}, called={}, Job1Info({}, {})", signaled, called++,
         info.integer, info.str);
}

void Job2(std::any param, bool signaled)
{
    static int called = 1;
    auto info = std::any_cast<Job2Info>(param);
    _dbg("Job2: signaled={}, called={}, Job2Info({}, {})", signaled, called++,
         info.some_value, info.other_value);
}

int main(int argc, char **argv)
{
    using namespace std::literals::chrono_literals;
    using namespace engine;

    Job1Info info1{10, std::string("hello")};
    Job2Info info2{50, false};

    Job job1(JOB1_ID, job_func{Job1}, 5000ms, std::make_any<Job1Info>(info1));
    Job job2(JOB2_ID, job_func{Job2}, 10000ms, std::make_any<Job2Info>(info2));
    JobManager manager({job1, job2});

    _dbg("Start");
    manager.Start();
    std::this_thread::sleep_for(20000ms);
    manager.SetJobParam(JOB2_ID, std::make_any<Job2Info>(30, true));
    manager.Remove(JOB1_ID);
    std::this_thread::sleep_for(10000ms);
    manager.Add(job1, true);
    std::this_thread::sleep_for(20000ms);
    manager.SetJobParam(JOB1_ID, std::make_any<Job1Info>(10, "something"));
    std::this_thread::sleep_for(20000ms);
    manager.Stop();
    _dbg("end");
}

[2022-06-28 16:43:14.263] [info] Start
[2022-06-28 16:43:19.289] [info] Job1: signaled=false, called=1, Job1Info(10, hello)
[2022-06-28 16:43:24.291] [info] Job2: signaled=false, called=1, Job2Info(50, false)
[2022-06-28 16:43:24.291] [info] Job1: signaled=false, called=2, Job1Info(10, hello)
[2022-06-28 16:43:29.305] [info] Job1: signaled=false, called=3, Job1Info(10, hello)
[2022-06-28 16:43:34.307] [info] Job2: signaled=false, called=2, Job2Info(30, true)
[2022-06-28 16:43:44.305] [info] Job1: signaled=false, called=4, Job1Info(10, hello)
[2022-06-28 16:43:44.318] [info] Job2: signaled=false, called=3, Job2Info(30, true)
[2022-06-28 16:43:49.314] [info] Job1: signaled=false, called=5, Job1Info(10, hello)
[2022-06-28 16:43:54.325] [info] Job2: signaled=false, called=4, Job2Info(30, true)
[2022-06-28 16:43:54.325] [info] Job1: signaled=false, called=6, Job1Info(10, hello)
[2022-06-28 16:43:59.342] [info] Job1: signaled=false, called=7, Job1Info(10, hello)
[2022-06-28 16:44:04.327] [info] Job2: signaled=false, called=5, Job2Info(30, true)
[2022-06-28 16:44:04.358] [info] Job1: signaled=false, called=8, Job1Info(10, something)
[2022-06-28 16:44:09.374] [info] Job1: signaled=false, called=9, Job1Info(10, something)
[2022-06-28 16:44:14.342] [info] Job2: signaled=false, called=6, Job2Info(30, true)
[2022-06-28 16:44:14.388] [info] Job1: signaled=false, called=10, Job1Info(10, something)
[2022-06-28 16:44:19.397] [info] Job1: signaled=false, called=11, Job1Info(10, something)
[2022-06-28 16:44:24.322] [info] end

Answer 1

Use of std::shared_ptr

In Job, you store the lock, condition variable and thread in std::shared_ptrs, however that seems unnecessary. Just store those by value. The same goes for m_thread in JobManager.

Of course, this opens a can of worms, because you expect Jobs to be copyable, and by having plain std::thread and related variables in that class, it is no longer copyable. However, that is actually a good thing. Consider what happens if you copy a Job: there is only one m_thread running, but now you have two copies of m_active, m_signaled and so on. Consider doing the following:

Job job1(1, ...);
job1.Start();
JobManager manager({job1});
manager.Remove(1);

Because job1 was started before it was added to the manager, its Monitor() will use a different m_quit than when Stop() is called via manager.Remove(1), causing a deadlock.

The solution is to not have std::shared_ptrs inside Job, but instead having std::shared_ptr<Job>s passed to and stored inside JobManager, so that the code using it looks like:

auto job1 = std::make_shared<Job>(JOB1_ID, ...);
auto job2 = std::make_shared<Job>(JOB1_ID, ...);
JobManager manager({job1, job2});

Now it is clear up front that jobs are passed as shared pointers, and there are no more surprises. But it is also possible to avoid std::shared_ptr altogether, which brings me to:

Let JobManager create Jobs

I think the intent is that the JobManager has full control over the Jobs it is managing. However, by letting the application create Jobs itself and pass them to JobManager, you open up the chance of misuse. It would be better if JobManager itself was creating new Jobs internally, and never exposing them to the application.

To do this, there have to be some changes in how you add jobs to a JobManager. In particular, you want to have an Add() function which takes all the parameters the constructor of Job would take:

class JobManager {
public:
    ...
    void Add(uint32_t job_id, job_func &&func, const job_interval &interval, const std::any &param, bool run_now = false) {
        auto &job = m_jobs.emplace_back(job_id, func, interval, param);
        job.Start(run_now);
    }
    ...
private:
    std::list<Job> m_jobs;
    ...
};

Note that Job does not have to be copyable in order for it to be stored in a std::list.

Consider removing param

The job_func takes a std::any param as an argument, which allows you to pass arbitrary data to it. However, std::any has its issues: it's rather expensive since it allocates memory, and while it has some degree of type safety, it still allows you to pass an incorrect type, and this will only be caught at runtime instead of at compile time.

Instead, if you want to pass extra parameters to the function, consider using a lambda that captures those parameters, so that you can write:

void Job1(Job1Info &info, bool signaled) {
    static int called = 1;
    _dbg("Job1: signaled={}, called={}, Job1Info({}, {})",
         signaled, called++, info.integer, info.str);
}
...
Job1Info info1{10, "Hello,"};
JobManager.Add(JOB1_ID, [&](bool signaled){Job1(info1, signaled);}, 5000ms);

Of course, if you also want to be able to change the job parameters while a job is running, you still need to do locking, for example by adding a std::mutex to the job info struct.

Improper locking

When you want to pass information between threads using some variables, make sure you always take a lock when reading from or writing to those variables, otherwise you might introduce race conditions to your code. For example, by not taking a locking when accessing m_quit, the notify in Stop() might not have any effect, because in Monitor(), you first read m_quit, which might still be false, then Stop() could set m_quit to true and notify m_cv, but only then Monitor() takes a lock and calls m_cv.wait_until(...), however that latter call will not see notifications that came before it, so it will wait for the full job interval, and if that is a large duration, it will seem like your program is hanging.

Answer 2

I don't have time for a full review at the moment, but just glancing at it, one thing jumped out at me about JobManager: you don't seem to have really made up your mind whether you want it to support derivation or not.

On one hand, it has some protected members, which generally only make sense if something is going to derive from it.

On the other hand, it contains no virtual functions, so there's nothing for a derived class to override. Its destructor isn't virtual either, which generally means it shouldn't be used as a base class.

There are a few cases (such as private inheritance) where a class can have a non-virtual destructor, and still be used as a base class. But even in a case like this, you generally need at least one virtual function to override before inheritance makes sense. Otherwise, you should probably just aggregate an instance of the class into whatever else you're creating.