21
\$\begingroup\$

I needed a task scheduler that could be used somehow like this:

#include <iostream>

// To compile and run this example, include here the code listed in the second code block

void Task2()
{
    std::cout << "OK2 ! now is   " << std::chrono::system_clock::now().time_since_epoch().count() << std::endl;
}
void Task3()
{
    std::cout << "--3 " << std::endl;
}

void Task1(Scheduler & sch)
{
    auto now = std::chrono::system_clock::now();
    std::cout << "OK1 ! now is   " << now.time_since_epoch().count() << std::endl;

    sch.ScheduleAt(now + std::chrono::seconds(1), []{ Task2(); });
    sch.ScheduleAt(now + std::chrono::seconds(2), []{ Task2(); });
    sch.ScheduleAt(now + std::chrono::seconds(3), []{ Task2(); });
}

void main()
{
    auto now = std::chrono::system_clock::now();

    Scheduler sch;

    sch.ScheduleAt(now + std::chrono::seconds(15), [&sch]{ Task1(sch); });
    sch.ScheduleAt(now + std::chrono::seconds(20), [&sch]{ Task1(sch); });
    sch.ScheduleAt(now + std::chrono::seconds(25), [&sch]{ Task1(sch); });  
    sch.ScheduleAt(now + std::chrono::seconds( 2), [&sch]{ Task2();    });

    sch.ScheduleEvery(std::chrono::seconds(1), []{ Task3(); });

    getchar();
}

It had to have tasks scheduled once and repetitive tasks, and should stop and clean itself up gracefully on destruction even while running. I did not care for parallelism: tasks that should run in their own threads should manage it. It had to accept lambdas for simplicity.

I could not find anything similar in Boost or POCO, though I did not search very hard as I was interested in writing it myself. There are some related questions in C#, but I need C++.

#include <map>
#include <functional>
#include <chrono>
#include <mutex>
#include <thread>
#include <condition_variable>
#include <memory>

#include <boost/noncopyable.hpp>

class Scheduler : boost::noncopyable
{
private:
    std::multimap<std::chrono::system_clock::time_point, std::function<void()>> tasks;
    std::mutex mutex;
    std::unique_ptr<std::thread> thread;
    std::condition_variable blocker;
    bool go_on;

public:

    Scheduler()
        :go_on(true)
    {
        thread.reset(new std::thread([this](){ this->ThreadLoop(); }));
    }
    ~Scheduler()
    {
        go_on = false;
        ScheduleAt(std::chrono::system_clock::now(), [](){});
        thread->join();
    }

    void ThreadLoop()
    {
        while(go_on)
        {
            std::function<void()> todo;
            {
                std::unique_lock<std::mutex> lock(mutex);
                auto now = std::chrono::system_clock::now();
                if ( tasks.empty()==false && tasks.begin()->first <= now)
                {
                    todo = tasks.begin()->second;
                    tasks.erase(tasks.begin());
                }
            }

            // Run tasks while unlocked so tasks can schedule new tasks
            if (todo)
                todo();

            {
                std::unique_lock<std::mutex> lock(mutex);
                if (tasks.empty())
                    blocker.wait(lock);
                else
                    blocker.wait_until(lock, tasks.begin()->first);
            }
        }
    }

    void ScheduleAt(std::chrono::system_clock::time_point & time, std::function<void()> func)
    {
        std::unique_lock<std::mutex> lock(mutex);

        auto it = tasks.insert(std::make_pair(time, func));
        if (it == tasks.begin())
            blocker.notify_one();
    }

    void ScheduleEvery(std::chrono::system_clock::duration interval, std::function<void()> func)
    {
        std::function<void()> waitFunc = [this,interval,func]()
            {
                func();
                this->ScheduleEvery(interval, func);
            };
        ScheduleAt(std::chrono::system_clock::now() + interval, waitFunc);
    }
};

Notes:

I am not concerned with performance but I'll still take free lunches, like trading std::function for another faster type.

I'm using VS2012 Express.

If a task takes time, other tasks may run belated. Such tasks should run in their own thread or be posted in an io_service or such. In any case, the caller (of ScheduleAt or ScheduleEvery) should take care of this.

Questions:

  1. What do you think?
  2. Are there drawbacks or pitfalls I did not see?
  3. Do other libs like Boost have something similar that I should use instead?
  4. Are there race conditions I missed?
  5. Is the destructor correct, especially the bool thing? I thought of using volatile here but the Wikipedia article about volatile had me convinced otherwise.
\$\endgroup\$
3
  • 1
    \$\begingroup\$ A few notes: *(1) You can use std::thread directly (no pointer), especially since your class is not copyable. *(2) std::function allows to capture context, that can potentially be heavy, you can do todo = std::move(tasks.begin()->second) to avoid copying the context \$\endgroup\$
    – aberaud
    May 21, 2016 at 19:02
  • 1
    \$\begingroup\$ *(3) for the same reason, you may take rvalue references for std::function and std::move() emplacing them in the multimap. \$\endgroup\$
    – aberaud
    May 21, 2016 at 19:09
  • 1
    \$\begingroup\$ note: those comments assume you moved to a newer version of VS \$\endgroup\$
    – aberaud
    May 21, 2016 at 19:13

3 Answers 3

9
\$\begingroup\$

So a couple of minor issues first:

void main()

Argh, bad. Someone who can write this should know better!

There's a problem with your ScheduleAt function. You're passing in an rvalue reference to time, and trying to bind this to a non-const lvalue reference. Visual Studio has some (evil) extensions that let you get away with this, however, this isn't portable, and should be one of:

void ScheduleAt(const std::chrono::system_clock::time_point& time, std::function<void()> func)

void ScheduleAt(std::chrono::system_clock::time_point&& time, std::function<void()> func)

Finally, a (very) minor point: why introduce a dependency on boost simply for non-copyable? It's all of 2 lines in C++11:

 Scheduler& operator=(const Scheduler& rhs) = delete;
 Scheduler(const Scheduler& rhs) = delete;

Not only that, but it doesn't disallow move operations. Hence:

Scheduler& operator=(Schedular&& rhs);
Scheduler(Schedular&& rhs);

are still accessible (and still generated). This may (or may not be) what you want.

There may be a library that (sort of) does what you want, however, it is C, not C++: libevent. I've never used it, so take this recommendation with a grain of salt.

With what you have here, there won't be any race conditions, because you launch a Scehduler from the main thread, and this only ever creates one thread of its own. Further, this thread only performs sequential execution.

Personally, this implementation has some extra pieces that don't seem to be needed; the mutex and condition_variable only really there to signal when things should happen, not for any actual locking of shared mutable state. Further (as you're probably aware), this likely won't work so well for any function which have a non-negligible execution time. The getchar at the end is a bit of a hack as well, you should probably launch this in its own thread.

This implementation doesn't fix the part where long running functions will cause everything to get out of sync, but it doesn't rely on mutex or condition_variable where they aren't really needed. You could probably fix this by launching each func() call in its own thread if you wanted to. I've opted to use a priority_queue instead of a multimap as well:

#include <functional>
#include <chrono>
#include <future>
#include <queue>
#include <thread>
#include <memory>

struct function_timer
{
    std::function<void()> func;
    std::chrono::system_clock::time_point time;

    function_timer()
    { }

    function_timer(std::function<void()>&& f, std::chrono::system_clock::time_point& t)
        : func(f), 
          time(t)
    { }

    //Note: we want our priority_queue to be ordered in terms of
    //smallest time to largest, hence the > in operator<. This isn't good
    //practice - it should be a separate struct -  but I've done this for brevity.
    bool operator<(const function_timer& rhs) const
    {
        return time > rhs.time;
    }

    void get()
    {
        func();
    }
};

class Scheduler
{
private:
    std::priority_queue<function_timer> tasks;
    std::unique_ptr<std::thread> thread;
    bool go_on;

    Scheduler& operator=(const Scheduler& rhs) = delete;
    Scheduler(const Scheduler& rhs) = delete;

public:

    Scheduler()
        :go_on(true),
        thread(new std::thread([this]() { ThreadLoop(); }))
    { }

    ~Scheduler()
    {
        go_on = false;
        thread->join();
    }

    void ThreadLoop()
    {
        while(go_on)
        {
            auto now = std::chrono::system_clock::now();
            while(!tasks.empty() && tasks.top().time <= now) {
                function_timer& f = tasks.top();
                f.get();
                tasks.pop();
            }

            if(tasks.empty()) {
                std::this_thread::sleep_for(std::chrono::milliseconds(100));
            } else {
                std::this_thread::sleep_for(tasks.top().time - std::chrono::system_clock::now());
            }
        }
    }

    void ScheduleAt(std::chrono::system_clock::time_point& time, std::function<void()>&& func)
    {
        tasks.push(function_timer(std::move(func), time));
    }

    void ScheduleEvery(std::chrono::system_clock::duration interval, std::function<void()> func)
    {
        std::function<void()> waitFunc = [this,interval,func]()
            { 
                func();
                this->ScheduleEvery(interval, func);
            };
        ScheduleAt(std::chrono::system_clock::now() + interval, std::move(waitFunc));
    }
};

I imagine there's also a slightly cleaner way of doing this with std::async, std::promise and std::future, but perhaps someone else can figure that out.

Edit: Whoops, there is shared mutable state, I'm definitely wrong about that.

\$\endgroup\$
4
  • 1
    \$\begingroup\$ Is std::priority_queue synchronized ? You use tasks in ThreadLoop and in the main thread (via ScheduleAt/Every). Is this really OK ? I used condition_variable to be able to wake up a thread when a task is posted because I did not want to peek every so often, like you did, choosing 100ms as an acceptable possible lag. \$\endgroup\$
    – Gabriel
    Feb 7, 2013 at 7:32
  • 1
    \$\begingroup\$ No it isn't, and yes, you're right, my apologies, it should have a mutex there. As for the delay, note that will only happen when the queue is empty - when it has elements in it, it will sleep for the appropriate amount of time. Given the fact that none of what's put into the queue runs in a separate thread, anything that runs for longer than 100ms will put the queue out of sync by more than that anyway - with the implementation as is, I think a condition variable is overkill. If you change it to run each function in a separate thread, though, then you'd likely need it. \$\endgroup\$
    – Yuushi
    Feb 7, 2013 at 8:40
  • \$\begingroup\$ I was aware of the delay generated by long-running tasks and I should have talked about it in the question. I figured that such tasks should include a thread, or a post() somewhere in an io_service or such. Thanks for the considerations ! \$\endgroup\$
    – Gabriel
    Feb 7, 2013 at 10:45
  • 1
    \$\begingroup\$ @Yuushi, I'm a bit late to the party here, but I would argue that a condition variable is far from overkill. Consider the case where the scheduler thread is sleeping for the next scheduled event, and another thread schedules a task for a time sooner than that. \$\endgroup\$
    – John Auld
    Aug 3, 2016 at 20:52
5
\$\begingroup\$

Consider this improved version of Yuushi's code:

#include <functional>
#include <chrono>
#include <future>
#include <queue>
#include <thread>
#include <memory>
#include <sstream>
#include <assert.h>

struct function_timer
{
    std::function<void()> func;
    std::chrono::system_clock::time_point time;

    function_timer()
    //      :func(nullptr),time(nullptr)
    { }

    function_timer(std::function<void()>&& f, const std::chrono::system_clock::time_point& t)
        : func(f),
          time(t)
    { }

    //Note: we want our priority_queue to be ordered in terms of
    //smallest time to largest, hence the > in operator<. This isn't good
    //practice - it should be a separate struct -  but I've done this for brevity.
    bool operator<(const function_timer& rhs) const
    {
        return time > rhs.time;
    }

    void get()
    {
        //won't work correctly
        //              std::thread t(func);
        //              t.detach();
        func();
    }
    void operator()(){
        func();
    }
};

class Scheduler
{
private:
    std::priority_queue<function_timer> tasks;
    std::unique_ptr<std::thread> thread;
    bool go_on;

    Scheduler& operator=(const Scheduler& rhs) = delete;
    Scheduler(const Scheduler& rhs) = delete;

public:

    Scheduler()
        :go_on(true),
          thread(new std::thread([this]() { ThreadLoop(); }))
    { }

    ~Scheduler()
    {
        go_on = false;
        thread->join();
    }

    void ScheduleAt(const std::chrono::system_clock::time_point& time, std::function<void()>&& func)
    {
        std::function<void()> threadFunc = [func]()
        {
                std::thread t(func);
                t.detach();
                //              func();

            };
        tasks.push(function_timer(std::move(threadFunc), time));
    }

    void ScheduleEvery(std::chrono::system_clock::duration interval, std::function<void()> func)
    {
        std::function<void()> threadFunc = [func]()
        {
                std::thread t(func);
                t.detach();
                //              func();

            };
        this->ScheduleEveryIntern(interval, threadFunc);
    }

    //in format "%s %M %H %d %m %Y" "sec min hour date month year"
    void ScheduleAt(const std::string &time, std::function<void()> func)
    {
        if (time.find("*") == std::string::npos && time.find("/") == std::string::npos)
            {
                std::tm tm = {};
                strptime(time.c_str(), "%s %M %H %d %m %Y", &tm);
                auto tp = std::chrono::system_clock::from_time_t(std::mktime(&tm));
                if(tp>std::chrono::system_clock::now())
                    ScheduleAtIntern(tp, std::move(func));
            }
    }
private:
    void ScheduleAtIntern(const std::chrono::system_clock::time_point& time, std::function<void()>&& func)
    {
        tasks.push(function_timer(std::move(func), time));
    }

    void ScheduleEveryIntern(std::chrono::system_clock::duration interval, std::function<void()> func)
    {
        std::function<void()> waitFunc = [this,interval,func]()
        {
                //              std::thread t(func);
                //              t.detach();
                func();
                this->ScheduleEveryIntern(interval, func);
            };
        ScheduleAtIntern(std::chrono::system_clock::now() + interval, std::move(waitFunc));
    }

    void ThreadLoop()
    {
        while(go_on)
            {
                auto now = std::chrono::system_clock::now();
                while(!tasks.empty() && tasks.top().time <= now) {
                        function_timer f = tasks.top();
                        f.get();
                        tasks.pop();
                    }

                if(tasks.empty()) {
                        std::this_thread::sleep_for(std::chrono::milliseconds(100));
                    } else {
                        std::this_thread::sleep_for(tasks.top().time - std::chrono::system_clock::now());
                    }
            }
    }

};

Now each function is launched in a separate thread and this didn't block long time operations like this.

void Task10sec()
{
  std::chrono::system_clock::time_point p = std::chrono::system_clock::now();
  while (std::chrono::system_clock::now()<(p+std::chrono::seconds(10))){}
}

so if we launch 3 tasks which should be executed in intersecting intervals each task is executed in separate thread and at the expected time.

F1 short task is executed every second

F2 long task is executed every minute

F3 long task is executed once. Is started during 2 task.

sch.ScheduleEvery(std::chrono::minutes(1), []{
    std::chrono::system_clock::time_point p = std::chrono::system_clock::now();
    std::time_t t = std::chrono::system_clock::to_time_t(p);
    std::cout << "F2.1 "<<std::ctime(&t) << std::endl;

                  Task10sec();

    std::chrono::system_clock::time_point p2 = std::chrono::system_clock::now();
    std::time_t t2 = std::chrono::system_clock::to_time_t(p2);
    std::cout << "F2.2 "<<std::ctime(&t2) << std::endl;

            });
sch.ScheduleAt(now + std::chrono::seconds(65), []{
    std::chrono::system_clock::time_point p = std::chrono::system_clock::now();
    std::time_t t = std::chrono::system_clock::to_time_t(p);
    std::cout << "F3.1 "<<std::ctime(&t) << std::endl;

                  Task10sec();

    std::chrono::system_clock::time_point p2 = std::chrono::system_clock::now();
    std::time_t t2 = std::chrono::system_clock::to_time_t(p2);
    std::cout << "F3.2 "<<std::ctime(&t2) << std::endl;

            });

can produce this result (each function is executing his code on his own thread)

F1 Thu Nov  3 15:07:41 2016
F1 Thu Nov  3 15:07:42 2016
F2.1 Thu Nov  3 15:07:43 2016
F1 Thu Nov  3 15:07:43 2016
.
.
F1 Thu Nov  3 15:07:47 2016
F3.1 Thu Nov  3 15:07:48 2016
F1 Thu Nov  3 15:07:48 2016
F1 Thu Nov  3 15:07:49 2016
.
.
F1 Thu Nov  3 15:07:52 2016
F2.2 Thu Nov  3 15:07:53 2016
F1 Thu Nov  3 15:07:53 2016
F1 Thu Nov  3 15:07:54 2016
.
.
F1 Thu Nov  3 15:07:57 2016

F3.2 Thu Nov  3 15:07:58 2016

F1 Thu Nov  3 15:07:58 2016
\$\endgroup\$
0
2
\$\begingroup\$

I tested the scheduler and found that in some cases,

blocker.wait_until(lock, tasks.begin()->first);

will wake up 1 second before the actual scheduled time.

Then tasks.begin()->first <= now fails.

Then the next

blocker.wait_until(lock, tasks.begin()->first);

Fails because the time has already elapsed...

\$\endgroup\$
2
  • \$\begingroup\$ Could you provide the full code, please? \$\endgroup\$
    – Gabriel
    Apr 18, 2015 at 20:12
  • \$\begingroup\$ pastebin.com/Ju6Qh3nQ \$\endgroup\$
    – Lefteris E
    Jul 7, 2015 at 12:59

Your Answer

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

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