1
\$\begingroup\$

This is a revised version of the JobManager class originally posted here. The revisions made were centered around:

  • Encapsulating the Job class within the JobManager class.
  • Removing the use of std::shared_ptr(s) from the Job class.
  • Moving most of the Job member method logic into the JobManager class.

I have a few questions on this version:

  • Is how I'm capturing the Job within JobManager::Add, JobManager::StartAll, and JobManager::Start correct?
  • Is there a better way to tell if a Job's thread is active besides using a std::atomic<bool> active member (Maybe futures)?
  • Some of the Jobs I'm expecting to add might take longer to run than the Job's interval. Does the upfront quit check within Job::Monitor help this?

The JobManager class manages the periodic execution of Jobs. While running, the API allows the user to:

  • Add a Job to the manager, optionally running it 'now' with no interval delay.
  • Remove a Job from the manager.
  • Stop and start a Job, optionally running it 'now' with no interval delay.
  • Signal a Job to run if it is waiting.
  • Change a Job's interval and function.

// job_manager.hpp
#pragma once
#include <algorithm>
#include <chrono>
#include <condition_variable>
#include <cstdio>
#include <functional>
#include <list>
#include <memory>
#include <mutex>
#include <thread>

namespace engine {

using job_func = std::function<void(bool)>;
using job_interval = std::chrono::system_clock::duration;

class JobManager final {
public:
    JobManager() = default;
    JobManager &operator=(const JobManager &rhs) = delete;
    JobManager(const JobManager &rhs) = delete;
    JobManager &operator=(JobManager &&rhs) = delete;
    JobManager(JobManager &&rhs) = delete;
    ~JobManager();

    void Add(uint32_t id, job_func &&func, const job_interval &ivl, bool run_now);
    void Remove(uint32_t id);

    void StartAll();
    void StopAll();

    void Start(uint32_t id, bool run_now);
    void Stop(uint32_t id);

    void Signal(uint32_t id);

    void SetFunc(uint32_t id, job_func &&func);
    void SetInterval(uint32_t id, job_interval interval);

private:
    struct Job {
        Job(uint32_t job_id, job_func &&func, const job_interval &interval);
        Job &operator=(const Job &rhs) = delete;
        Job(const Job &rhs) = delete;
        Job &operator=(Job &&rhs) = delete;
        Job(Job &&rhs) = delete;
        ~Job();

        void Monitor(bool now);

        const uint32_t id;
        std::mutex lock;
        std::condition_variable cv;
        std::thread thread;
        std::atomic<bool> active{false};
        bool signaled{false};
        bool quit{false};
        job_func func;
        job_interval interval;
    };

    std::list<Job>::iterator FindJobByID(uint32_t id);
    std::list<Job> m_jobs;
    std::mutex m_lock;
};

using jobmanager_t = std::shared_ptr<JobManager>;

} // namespace engine


/// job_manager.cpp
#include <engine/job_manager.hpp>

namespace engine {

JobManager::~JobManager()
{
    StopAll();
}

void JobManager::Add(uint32_t id, job_func &&func, const job_interval &ivl,
                     bool run_now)
{
    std::lock_guard<std::mutex> lk(m_lock);
    if (FindJobByID(id) == m_jobs.end()) {
        auto &job = m_jobs.emplace_back(id, std::move(func), ivl);
        if (run_now) {
            job.thread = std::thread([&job]() { job.Monitor(true); });
            job.active = true;
        }
    }
}

void JobManager::Remove(uint32_t id)
{
    std::lock_guard<std::mutex> lk(m_lock);
    auto job = FindJobByID(id);
    if (job != m_jobs.end()) {
        if (job->active) {
            {
                std::lock_guard<std::mutex> lk(job->lock);
                job->quit = true;
            }
            job->cv.notify_one();
            job->thread.join();
            job->active = false;
        }
        m_jobs.erase(job);
    }
}

void JobManager::StartAll()
{
    std::lock_guard<std::mutex> lk(m_lock);
    for (auto &job : m_jobs) {
        if (!job.active) {
            job.thread = std::thread([&job]() { job.Monitor(false); });
            job.active = true;
        }
    }
}

void JobManager::StopAll()
{
    std::lock_guard<std::mutex> lk(m_lock);
    for (auto &job : m_jobs) {
        if (job.active) {
            {
                std::lock_guard<std::mutex> lk(job.lock);
                job.quit = true;
            }
            job.cv.notify_one();
            job.thread.join();
            job.active = false;
            job.signaled = false;
            job.quit = false;
        }
    }
}

void JobManager::Start(uint32_t job_id, bool run_now)
{
    std::lock_guard<std::mutex> lk(m_lock);
    auto job = FindJobByID(job_id);
    if (job != m_jobs.end() && !job->active) {
        job->thread = std::thread(
            [job](bool run_now) { job->Monitor(run_now); }, run_now);
        job->active = true;
    }
}

void JobManager::Stop(uint32_t id)
{
    std::lock_guard<std::mutex> lk(m_lock);
    auto job = FindJobByID(id);
    if (job != m_jobs.end() && job->active) {
        {
            std::lock_guard<std::mutex> lk(job->lock);
            job->quit = true;
        }
        job->cv.notify_one();
        job->thread.join();
        job->active = false;
        job->signaled = false;
        job->quit = false;
    }
}

void JobManager::Signal(uint32_t id)
{
    std::lock_guard<std::mutex> lk(m_lock);
    auto job = FindJobByID(id);
    if (job != m_jobs.end() && job->active) {
        {
            std::lock_guard<std::mutex> lk(job->lock);
            job->signaled = true;
        }
        job->cv.notify_one();
    }
}

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

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

std::list<JobManager::Job>::iterator JobManager::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);
}

//
//
// Job definitions
//
//

JobManager::Job::Job(uint32_t job_id, job_func &&func,
                     const job_interval &interval)
    : id(job_id), func(func), interval(interval)
{
}

JobManager::Job::~Job()
{
    if (active) {
        {
            std::lock_guard<std::mutex> lk(lock);
            quit = true;
        }
        cv.notify_one();
        thread.join();
    }
}

void JobManager::Job::Monitor(bool now)
{
    if (now) {
        func(false);
    }

    while (true) {

        {
            std::lock_guard<std::mutex> lk(lock);
            if (quit) {
                break;
            }
        }

        {
            std::unique_lock<std::mutex> lk(lock);
            cv.wait_until(lk, std::chrono::system_clock::now() + interval);
        }

        std::lock_guard<std::mutex> lk(lock);
        if (quit) {
            break;
        }
        func(signaled);
        signaled = false;
    }
}

} // namespace engine



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

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

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(const Job1Info &info, bool signaled)
{
    static int called = 1;
    _dbg("Job1: signaled={}, called={}, Job1Info({}, {})", signaled, called++,
         info.integer, info.str);
}

void Job1Alt(const Job1Info &info, bool signaled)
{
    static int called = 1;
    _dbg("Job1Alt: signaled={}, called={}, Job1Info({}, {})", signaled,
         called++, info.integer, info.str);
}

void Job2(const Job2Info &info, bool signaled)
{
    static int called = 1;
    _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};

    jobmanager_t manager = std::make_shared<JobManager>();
    manager->Add(JOB1_ID, [info1](bool s) { Job1(info1, s); }, 5s, true);
    manager->Add(JOB2_ID, [info2](bool s) { Job2(info2, s); }, 10s, false);

    _dbg("Start");
    manager->StartAll();
    std::this_thread::sleep_for(15s);
    
    manager->Signal(JOB2_ID); 
    std::this_thread::sleep_for(10s); 

    manager->SetInterval(JOB2_ID, 5s);
    std::this_thread::sleep_for(10s);

    manager->Stop(JOB1_ID);
    std::this_thread::sleep_for(10s);

    manager->SetFunc(JOB1_ID, [info1](bool s) { Job1Alt(info1, s); });
    manager->SetInterval(JOB1_ID, 2s);
    manager->Start(JOB1_ID, true); 
    std::this_thread::sleep_for(10s);

    manager->StopAll();
    _dbg("end");
}

[2022-06-30 11:40:24.299] [info] Start
[2022-06-30 11:40:24.319] [info] Job1: signaled=false, called=1, Job1Info(10, hello)
[2022-06-30 11:40:29.329] [info] Job1: signaled=false, called=2, Job1Info(10, hello)
[2022-06-30 11:40:34.333] [info] Job1: signaled=false, called=3, Job1Info(10, hello)
[2022-06-30 11:40:34.333] [info] Job2: signaled=false, called=1, Job2Info(50, false)
[2022-06-30 11:40:39.330] [info] Job2: signaled=true, called=2, Job2Info(50, false)
[2022-06-30 11:40:39.345] [info] Job1: signaled=false, called=4, Job1Info(10, hello)
[2022-06-30 11:40:44.360] [info] Job1: signaled=false, called=5, Job1Info(10, hello)
[2022-06-30 11:40:49.332] [info] Job2: signaled=false, called=3, Job2Info(50, false)
[2022-06-30 11:40:49.362] [info] Job1: signaled=false, called=6, Job1Info(10, hello)
[2022-06-30 11:40:54.376] [info] Job1: signaled=false, called=7, Job1Info(10, hello)
[2022-06-30 11:40:59.335] [info] Job2: signaled=false, called=4, Job2Info(50, false)
[2022-06-30 11:41:04.349] [info] Job2: signaled=false, called=5, Job2Info(50, false)
[2022-06-30 11:41:09.351] [info] Job1Alt: signaled=false, called=1, Job1Info(10, hello)
[2022-06-30 11:41:09.365] [info] Job2: signaled=false, called=6, Job2Info(50, false)
[2022-06-30 11:41:11.357] [info] Job1Alt: signaled=false, called=2, Job1Info(10, hello)
[2022-06-30 11:41:13.365] [info] Job1Alt: signaled=false, called=3, Job1Info(10, hello)
[2022-06-30 11:41:14.369] [info] Job2: signaled=false, called=7, Job2Info(50, false)
[2022-06-30 11:41:15.366] [info] Job1Alt: signaled=false, called=4, Job1Info(10, hello)
[2022-06-30 11:41:17.368] [info] Job1Alt: signaled=false, called=5, Job1Info(10, hello)
[2022-06-30 11:41:19.360] [info] end
\$\endgroup\$

1 Answer 1

2
\$\begingroup\$

Pass functions by value

You are passing job_func by rvalue reference, but that becomes a problem when someone has an lvalue job_func and want to pass it to JobManager. Consider:

std::function<void(bool)> foo = [](bool){};
JobManager jm;
jm.Add(..., foo, ...);

To avoid surprises, just pass it by value. You can have an overload for Add() that takes the function by rvalue reference if you want.

Job IDs

There are several things related to jobs and job IDs that can be improved.

Adding the same job ID twice

There are several ways to handle the case of the same job ID being Add()ed multiple times to the JobManager. However, your choice of ignoring subsequent Add()s is perhaps the least expected behavior. I think replacing the old function and interval with the new one would be less surprising, but alternatively I would throw an exception. Ignoring things is usually the worst thing you can do; if it was not intentional then the program will still appear to run, but do unexpected things.

Consider having the manager generate job IDs

Instead of the caller having to manage job IDs, why not let the manager generate unique IDs? So it becomes uint32_t Add(job_func func, ...). That way, the caller doesn't have to worry about IDs anymore. This is especially important in large projects where some parts of the code don't know what IDs other parts of the code will use.

Faster access by job ID

If each job has a unique ID, then instead of storing the jobs in a std::list, consider storing them in a std::unordered_map<uint32_t, Job>. That way, instead of scanning the whole list, it's a \$O(1)\$ lookup.

Create a type alias for the job ID type

uint32_ts are used for many different things. If you want to change the type of job IDs later, then you would have to search&replace those uint32_ts, not only in your job manager code, but also in the parts of the application that store those IDs. At the very least, create a type alias so that changing it later is much easier:

using ID = uint32_t;
...
void Signal(ID id);

Monitor threads

Unnecessary many locks

Inside the while-loop, you have three back-to-back sections which each take a lock. That just means the lock will be released and retaken a lot more times than necessary. You only need to take the lock once for the whole body of the while-loop. But that in fact means it still unlocks and relocks for every iteration of the while-loop, and that is also unnecessary. Just take the lock at the start of the function:

void JobManager::Job::Monitor(bool now)
{
    std::unique_lock lk(lock);

    if (now)
        func(false);

    if (quit)
        break;

    while (true) {
        cv.wait_until(lk, std::chrono::system_clock::now() + interval);

        if (quit)
            break;

        func(signaled);
        signaled = false;
    }
}

If you are thinking, wouldn't that keep the mutex locked all the time, the answer is no, because cv.wait_until() will unlock it while it waits.

Should the lock be held when func() is called?

Something you have to decide is whether you want the mutex to be locked while func() is being called. The drawback is that if func() takes a long time, a call to Signal() for example might take longer than expected. It also prevents func() from calling anything that in turn would cause lock to be taken again, otherwise you will get a deadlock. On the other hand, keeping it locked might avoid issues like what should happen if Signal() is called while func() is running: should it be ignored or should it cause another call to func() right after the current one is finished?

Be aware of spurious wakeups

cv.wait_until() can return even without being signaled and without the desired time point having been reached. To avoid this from happening, pass it a predicate to check whether a signal was delivered:

auto expire_time = std::chrono::system_clock::now() + interval;
cv.wait_until(lk, expire_time, [&]{ return signaled; });

Consider using only a single thread for all jobs

Currently, each job has a thread running the Monitor() function, which just waits until the interval expired, then calls func(). Threads are not necessarily cheap though, and having many jobs can become a problem. Instead of this, consider only having a single thread to monitor all jobs, and that picks the one that should fire next, and waits until that time, calls that job's func(), and then repeats.

Avoid macros

It's almost never necessary to use macros in C++, prefer creating regular functions or variables instead. For example:

template<typename... Args>
void _dbg(const char *format, Args&&... args) {
    spdlog::info(format, std::forward<Args>(args)...);
}

Or perhaps better in this case would be to do:

using spdlog::info;
...
info("Start");
\$\endgroup\$
8
  • \$\begingroup\$ Thanks a bunch for looking at this revised version! There's a lot of good information here and I've definitely learned a thing or two from it. I'll be working on this on and off for the next few days and will let you know when I have questions or have a revised version ready. \$\endgroup\$
    – big54mac
    Jul 2, 2022 at 2:02
  • \$\begingroup\$ If I use a std::unordered_map and have the JobManager generate a new ID for each job added within JobManager::Add, how exactly would I tell if a duplicate job is added during the JobManager's lifetime? It would seem that I would need to add a new member to the Job class to distinguish one from another, like a std::string holding the job name. This comes from the fact that I cant compare two std::function objects (Or as far as I know). \$\endgroup\$
    – big54mac
    Jul 2, 2022 at 5:25
  • 1
    \$\begingroup\$ std::unordered_map::emplace() will return a tuple that includes a bool that will tell you if the insertion was succesful or if there was already an item in the map with the same key. That's one way to detect duplicates. Another is to just check if the map contains() a given key. You should never have to compare the function objects themselves. \$\endgroup\$
    – G. Sliepen
    Jul 2, 2022 at 7:00
  • 1
    \$\begingroup\$ Additionally, I can't use std::condition_variable::wait_until with a std::lock_guard within Job::Monitor. \$\endgroup\$
    – big54mac
    Jul 2, 2022 at 14:33
  • 1
    \$\begingroup\$ You can always check a map for a given key. Incrementing a static ID and checking it's not a duplicate by trying to look it up seems like a reasonable approach. You're right about std::lock_guard, that should be a std::unique_lock of course. \$\endgroup\$
    – G. Sliepen
    Jul 2, 2022 at 14:37

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.