Multi-threaded Time Scheduler without polling

Question

This is a quick shot at designing a time scheduler without polling. It is able to handle inputs which are to be executed earlier than already on the schedule. I want to improve it further, especially my insert() as I find the implementation rather ugly. Wish I could simplify it using condition_variable::wait_for().

The Scheduler is designed as a Singleton for my personal needs. The time-bots are launched to trigger the actual events, but they can be ejected if required.

#include <iostream>
#include <memory>
#include <thread>
#include <vector>
#include <queue>
#include <map>
#include <chrono>
#include <mutex>
#include <functional>
#include <condition_variable>


typedef std::function<void(void)> task_t;
typedef std::chrono::system_clock::time_point time_point;
std::map<time_point, task_t> schedules_;

class time_bot; // forward declaration

class scheduler
{
private:
    std::map<time_point, task_t> schedules_;
    std::condition_variable signal_;
    std::mutex lock_;
    std::unique_ptr<time_bot> time_bot_;
    bool running_;
    bool trigger_ready_;
    std::thread* thread_;
private:
    scheduler(scheduler&&) = delete;
    scheduler(const scheduler&) = delete;
    scheduler& operator()(scheduler&&) = delete;
    scheduler& operator()(const scheduler&) = delete;
    scheduler();
    void run();

public:
    static const std::unique_ptr<scheduler>& get_instance();
    void insert(std::pair<time_point, task_t> node);
    void remove(time_point tp);
    void close();
    void executed(time_point tp);
    ~scheduler();
};

class time_bot
{
private:
    std::pair<time_point, task_t>  task_;
    bool wasted_;
    std::thread* thread_;
    std::condition_variable& signal_;
    bool& trigger_ready_;
public:
    time_bot(std::pair<time_point, task_t> task, std::condition_variable& signal, bool& trigger_ready)
        :task_(task),
        signal_(signal)
        , trigger_ready_(trigger_ready)
    {
        trigger_ready = false;
        wasted_ = false;
        thread_ = new std::thread([&]() { run(); });
    }
    ~time_bot()
    {
        if (wasted_)
        {
            std::cout << " A wasted time_capsule destroyed. \n";
            thread_->join();
        }
        else
        {
            std::cout << " A time_capsule destroyed after sucessful trigger. \n";
            thread_->join();
        }

    }

    std::pair<time_point, task_t> get() const
    {
        return task_;
    }

    void set_wasted()
    {
        wasted_ = true;
    }
    void run()
    {
        std::cout << " A timed-task std::thread started \n";
        auto now = std::chrono::system_clock::now();
        auto lapse = std::chrono::duration_cast<std::chrono::milliseconds>(task_.first - now);
        std::this_thread::sleep_for(std::chrono::milliseconds(lapse));
        if (!wasted_)
        {
            task_.second();
            scheduler::get_instance()->executed(task_.first);
            std::cout << " A timed-task executed \n";
        }

    }
};



scheduler::scheduler()
{
    time_bot_ = nullptr;
    trigger_ready_ = false;
    running_ = true;
    thread_ = new std::thread([&]() { run(); });
}


scheduler::~scheduler()
{
    thread_->join();
    std::cout << "Scheduler std::thread joined successfully \n";
}

void scheduler::run()
{
    std::cout << "Scheduler std::thread initialized  \n";
    while (running_)
    {
        if (!schedules_.empty())
        {
            std::unique_lock<std::mutex> lock(lock_);
            signal_.wait(lock, [&]() {return trigger_ready_ || !running_; });
            if (!running_) return;
            time_bot_.reset(new time_bot(*schedules_.begin(), signal_, trigger_ready_));
        }
        std::this_thread::sleep_for(std::chrono::milliseconds(10));
    }
}

const std::unique_ptr<scheduler>& scheduler::get_instance()
{
    static std::unique_ptr<scheduler> ptr;
    if (ptr == nullptr)
        ptr.reset(new scheduler());
    return ptr;
}

void scheduler::insert(std::pair<time_point, task_t> node)
{
    std::unique_lock<std::mutex> lock(lock_);

    auto ready_trigger = [&]()
    {
        if (time_bot_ == nullptr)
        {
            trigger_ready_ = true;
            signal_.notify_one();
        }
    };

    if (schedules_.empty() && time_bot_ == nullptr)
    {
        std::cout << "node entered [schedules_.empty() && time_bot_ == nullptr ]  \n";
        schedules_.insert(node);
        ready_trigger();
    }
    else
        if (time_bot_ == nullptr)
        {
            std::cout << "node entered [time_bot_ == nullptr]  \n";
            schedules_.insert(node);
        }
        else
            if (time_bot_ != nullptr)
            {
                if (time_bot_.get()->get().first < node.first)
                {
                    std::cout << "node entered [node.first]  \n";
                    schedules_.insert(node);
                }
                else
                {
                    time_bot_->set_wasted();
                    time_bot_.reset();
                    schedules_.insert(node);
                    std::cout << "node entered [Prev marked wasted]  \n";
                    ready_trigger();
                }
            }

    lock.unlock();
    std::this_thread::sleep_for(std::chrono::milliseconds(10));
}
void scheduler::executed(time_point tp)
{
    std::unique_lock<std::mutex> lock(lock_);
    auto it = schedules_.find(tp);
    if (it != schedules_.end())
        schedules_.erase(it);
    if (!schedules_.empty())
    {
        trigger_ready_ = true;
        signal_.notify_one();
    }
}

void scheduler::remove(time_point tp)
{
    std::unique_lock<std::mutex> lock(lock_);
    auto it = schedules_.find(tp);
    if (it != schedules_.end() && it != schedules_.begin())
        schedules_.erase(it);
}
void scheduler::close()
{
    running_ = false;
    signal_.notify_one();
}

int main()
{
    int i;
    std::cout << " Enter test# (1,2,3) ? ";
    std::cin >> i;
    switch (i)
    {
    case 1:
        scheduler::get_instance()->insert(make_pair(std::chrono::system_clock::now() + std::chrono::seconds(1), []() { std::cout << " Task[quick first] \n"; }));
        scheduler::get_instance()->insert(make_pair(std::chrono::system_clock::now() + std::chrono::seconds(2), []() { std::cout << " Task[Slow second] \n"; }));
        break;
    case 2:
        scheduler::get_instance()->insert(make_pair(std::chrono::system_clock::now() + std::chrono::seconds(3), []() { std::cout << " Task[Slow first] \n"; }));
        scheduler::get_instance()->insert(make_pair(std::chrono::system_clock::now() + std::chrono::seconds(1), []() { std::cout << " Task[Impatience second] \n"; }));
        break;
    default:
        scheduler::get_instance()->insert(make_pair(std::chrono::system_clock::now() + std::chrono::seconds(3), []() { std::cout << " Task[Slow first] \n"; }));
        scheduler::get_instance()->insert(make_pair(std::chrono::system_clock::now() + std::chrono::seconds(1), []() { std::cout << " Task[Medium second] \n"; }));
        scheduler::get_instance()->insert(make_pair(std::chrono::system_clock::now() + std::chrono::seconds(2), []() { std::cout << " Task[Impatience third] \n"; }));
    }

    std::this_thread::sleep_for(std::chrono::seconds(4));
    scheduler::get_instance()->close();

    return 0;
}

Answer 1

You leak memory in various places, particularly the thread objects you create. They're allocated with new, and joined, but never deleted.

You have a global schedules_ variable that is never used, since all the code references are within the scheduler class and will use the member variable. Since you have your main loop in scheduler::run accessing the schedules_ variable (the call to schedules_.empty()) outside of a lock, you have a race condition since another thread can be modifying the map while this check is occurring.

scheduler::run is also not very power friendly if there is nothing to schedule, since it uses a polling loop to check for a new task. This should use something like std::condition_variable to stay dormant until it is notified that a new task is available.

scheduler::get_instance is also not thread safe, since if two threads call it simultaneously before the instance is created, you'll get multiple instances. Since C++ now guarantees thread safety for initialization of static variables, just use a static instance of scheduler and dispense with the dynamic allocation. If you must use the dynamic allocation, you need to use something like a mutex to ensure you only create one instance of scheduler.

Within scheduler::insert itself, you have multi-line else statements that do not have enclosing curly braces. While not necessary, having them in will make the code a bit easier to follow while reducing the chance of somebody making an edit later that changes the flow.

The default in the switch in main is lacking a break statement. Although not currently necessary (since the default is the last branch in the switch), you should always include the break. Otherwise when somebody comes along and adds another case after the default the missing break will cause problems.