1
\$\begingroup\$

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

\$\endgroup\$

2 Answers 2

0
\$\begingroup\$

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.

\$\endgroup\$
5
  • \$\begingroup\$ @Sliepen Thanks for your review! I like the idea of hiding the Job implementation within the JobManager. I was a bit confused when you said that std::list does not require the objects it holds to be copyable. However, I realize now the the emplace_back method will construct the object in place, so the Job itself does not need to be coypable or wrapped in a shared_ptr (as you said). I'm going to do a rewrite taking your input, would you care to see an updated version? \$\endgroup\$
    – big54mac
    Jun 29, 2022 at 21:43
  • \$\begingroup\$ @Sliepen How much of the improper locking can be solved by using atomic bools? How exactly do I guarantee that the Stop() method would synchronize m_quit correctly with Monitor()? Right now, I have the Monitor() loop locking and checking m_quit before and after the m_cv.wait_until(...) section. Does it even make sense to have an m_waiting bool? \$\endgroup\$
    – big54mac
    Jun 30, 2022 at 0:00
  • \$\begingroup\$ We always like to see people incorporate our answers in their code! If you want the new version to be reviewed, just create a new review question here on Code Review, and then you can add a link here to the new question. As for atomic bools, that doesn't really help here. Everything that is used together, be that the bools or the condition variable used to signal changes in those bools, should be guarded by the same mutex. The m_waiting bool is indeed redundant, \$\endgroup\$
    – G. Sliepen
    Jun 30, 2022 at 6:47
  • \$\begingroup\$ Also, since you are new, please read what should you do when someone answers your question. \$\endgroup\$
    – G. Sliepen
    Jun 30, 2022 at 6:51
  • 1
    \$\begingroup\$ I have created a new question holding a revised version of the posted code. codereview.stackexchange.com/questions/277786/… \$\endgroup\$
    – big54mac
    Jul 1, 2022 at 4:01
1
\$\begingroup\$

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.

\$\endgroup\$
4
  • \$\begingroup\$ Thanks for your input. I believe I'm wanting nothing to derive from this class. Does it make sense to just change the protected members to private, or should I also mark the class as "final"? I'm going to go ahead and update my code snippet to change both. \$\endgroup\$
    – big54mac
    Jun 29, 2022 at 5:53
  • \$\begingroup\$ I've also removed the ThreadLoop from the JobManager class and I realize now that it's not needed. The Start and Stop member methods directly start and stop each job. \$\endgroup\$
    – big54mac
    Jun 29, 2022 at 6:05
  • 1
    \$\begingroup\$ private certainly makes sense. final...might also, but I'm somewhat uncertain. It's generally considered bad form to edit code here in a way that invalidates existing answers. If you want a review of updated code, that generally belongs in a new question (which should probably link back to this one). \$\endgroup\$ Jun 29, 2022 at 7:24
  • \$\begingroup\$ That makes sense to me, I'll do that for any future posts. \$\endgroup\$
    – big54mac
    Jun 29, 2022 at 9:25

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.