4
\$\begingroup\$

I've implemented a setTimeout thread similar to the one in JavaScript (new to thread programming)

In the example on this page I see use of an atomic variable i which I think it to make sure no race conditions occurs on 'i', but from what I've read I don't think there is an atomic multimap.

From the code a race condition might arise on the UI thread at queue.emplace(...) and in the thread iterating over the queue.

Does my thread code look up to the job generally, and should I be using another condition_variable to block on queue access?


UPDATE

I think I definitely needed to make the queue manipulations thread safe. I went down various dead ends on this as I'm learning how to program threads. In the end using a shared_timed_mutex worked ! This type of mutex can be shared across threads to synchronise data access and manipulation, e.g. you can use

{
   unique_lock<shared_timed_mutex> lock(shared_m);  // for writing
   // write data to whatever...
}

and

{
   shared_lock<shared_timed_mutex> lock(shared_m);  // for reading
   // read data from wherever...
}

Each *_lock will block if the mutex is currently locked, or you can add additional parameters to specify other types of behaviour. Each lock is released after the scope is exited.


Here's my original code:

WorkerThread.hpp:

using namespace std;
using namespace chrono;

class WorkerThread
{
public:
    typedef chrono::milliseconds Millis;
    typedef function<void(void)> Function;

    bool running = false;
    
    thread t;
    multimap<time_point<system_clock>, Function> queue;  // function queue (sorted)
    condition_variable cv;
    mutex cv_m;
    
    Millis msMin = 1ms;  // lowest sleep time allowed
    Millis msMax = 5ms;  // highest execution time preferred
    time_point<system_clock> waitUntil;  // next wake up time
        
    void setTimeout(Millis ms, Function f) {
    
        // is this line risky? what if the thread is processing queue?
        auto taskTime = system_clock::now() + ms;
        queue.emplace(taskTime, f);

        if(taskTime < waitUntil) {
            cout << "this task is earlier than previously added tasks" << endl;
            cv.notify_all();  // wake up waits in case this timeout task is more recent
        }
    }

    WorkerThread() {
        running = true;
        
        t = thread([=]() {
            std::unique_lock<std::mutex> lk(cv_m);

            while (running == true) {
                
                if(queue.empty()){
                    cout << "empty queue, sleep 60000ms" << endl;

                    // wake up in a minute if there's nothing to do
                    waitUntil = system_clock::now() + 60000ms;

                    // nothing to do, except if woken up
                    if(cv.wait_until(lk, waitUntil) == cv_status::timeout)
                        cout << "thread timed out" << endl;
                    else
                        cout << "thread woken up - earlier task identified !" << endl;
                }
                else {
                    // sleep until next task is ready ("up to" minimum permissible time)
                    waitUntil = max((*queue.begin()).first, system_clock::now() + msMin);

                    cout << "sleeping until next task: " << waitUntil.time_since_epoch().count() << endl;

                    // wait until next task, unless woken up
                    if(cv.wait_until(lk, waitUntil) == cv_status::timeout)
                        cout << "thread timed out" << endl;
                    else
                        cout << "thread woken up - earlier task identified !" << endl;
                }

                // process all available tasks up to maximum execution time
                auto maxtime = system_clock::now() + msMax;

                for(auto task = queue.begin(); task != queue.end(); ) {
                    if((*task).first <= maxtime) {
                        cout << "running task at: " << (*task).first.time_since_epoch().count() << endl;
                        (*task).second();  // run the task

                        // delete the task (the safe way)
                        auto taskSaved = task;
                        task++;
                        queue.erase(taskSaved);
                    }
                    else break; // max exec time reached, exit the for loop
                }
            }
        });
    }

    void stop()
    {
        running = false;
        t.join();
    }
};

Main:

    t = new WorkerThread();
    this_thread::sleep_for(1000ms);

    t->setTimeout(15000ms, []() { cout << "Hello from 2" << endl; } );
    cout << "added timeout 1" << endl;

    this_thread::sleep_for(6000ms);
    t->setTimeout(4000ms, []() { cout << "Hello from 1" << endl; } );
    cout << "added timeout 2" << endl;
    
    this_thread::sleep_for(100000ms);
    t->stop();

This code creates two timeouts, the first is set to trigger 15 seconds and the second 10 seconds from the beginning, but they're set up in such a way as to test the thread wakes up the wait_until's properly, which indeed works:

empty queue, sleep 60000ms
this task is earlier than previously added tasks
added timeout 1
thread woken up - earlier task identified !
sleeping until next task: 1600855233135593
this task is earlier than previously added tasks
thread woken up - earlier task identified !added timeout 2

sleeping until next task: 1600855228137566
thread timed out
running task at: 1600855228137566
Hello from 1
sleeping until next task: 1600855233135593
thread timed out
running task at: 1600855233135593
Hello from 2
empty queue, sleep 60000ms
\$\endgroup\$
4
\$\begingroup\$

Lock the mutex in setTimeout()

You have at least two threads accessing queue, so you have to ensure they don't update it simultaneously. You are holding the lock inside WorkerThread(), but you should also hold it inside setTimeout().

Give the class a better name

Yes, the class uses a worker thread to wait until the next timeout, but it is more than just the worker thread. It is actually a timer queue, where you can add timers that call a function when they time out.

class TimerQueue {
    ...
};

Also, setTimeout() sounds like it sets the timeout of the whole object. But it just adds an element to the queue. So I would name it addTimer(), or rather just add() or insert(), since it is clear from the name TimerQueue that you would add timers to it.

Avoid using a lambda for the thread function

It's not necessary. Why are you capturing the context by value? Were you aware that it still captures this by reference? Just use a regular member function for this. You can even have the thread initialized without needing a constructor, like so:

class TimerQueue {
    void worker() {
        std::unique_lock<std::mutex> lk(cv_m);

        while (running) {
            ...
        }
    }

    thread workerThread{&TimerQueue::worker, this};
    ...
};

You still need a destructor to join() the thread, although in C++20 this is no longer necessary if you use a std::jthread.

Ensure the destructor wakes up the worker thread

Your worker thread can sleep for up to 60 seconds if there is nothing in the queue. If you destroy the timer queue during that time, you might have to wait a long time for the call to join() to finish. Make sure you wake up the thread in the destructor:

~TimerQueue() {
    std::lock_guard<std::mutex> lk(cv_m);
    running = false;
    cv.notify_one();
    workerThread.join();
}

Another option is to enqueue a special item in the queue that signals that the worker thread should stop, and have the worker thread immediately exit the function if it encounters that item. This avoids the need for the variable running.

Avoid using system_clock for timers

The problem with system_clock is that it can suddenly jump, for example because of daylight saving time changes, leap seconds, and NTP updates. You should use std::chrono::steady_clock instead. I recommend you create a type alias for it:

using clock = std::chrono::steady_clock;

And then use it like so:

multimap<clock::time_point, Function> queue;
clock::time_point waitUntil;
...
waitUntil = clock::now() + ...;

Consider using a std::priority_queue

C++ has a container specifically to keep things sorted by priority: std::priority_queue. Consider using that. The only drawback is that it works more like a std::set than a std::map, you you have to define some struct to hold both a time point and a callback function, and have it sort correctly:

struct Timer {
    clock::time_point deadline;
    Function callback;

    bool operator<(const Timer &other) const {
        return other.deadline < deadline;
    }
};

std::priority_queue<Timer> queue;

You don't need waitUntil

You already know the next time to wake up by looking at the earliest time point in queue.

Avoid code duplication

Inside the worker thread, you deal with the case of an empty queue and a non-empty queue. However, the code in both cases is identical, except for the time point to wait until. You could just write:

waitUntil = clock::now() + queue.empty() ? 60000ms : queue.front().deadline;
cv.wait_until(lk, waitUntil);

Declare constants as such

You declare the variables msMin and msMax, and they look like constants, but you didn't tell the compiler about that fact. You can make them const, or even better static constexpr. But for the latter, you have to actually define them in a .cpp file as well, which is a bit annoying. This is fixed in C++17, where you can specify them as static inline constexpr.

Avoid iterator invalidation

When processing tasks that have expired, you call queue.erase(), but you already noticed you have to be careful to not invalidate the iterator. Relying on incrementing the iterator before calling erase() is not guaranteed to work. Instead, use the return value of erase() as the iterator to the next element:

for (auto task = queue.begin(); ...) {
    if (...) {
        ...
        task = queue.erase(task);
    } else {
        break;
    }
}

If you use a std::priority_queue instead, I would write the code like:

while (!queue.empty()) {
    auto timer = queue.top();

    if (timer.deadline < maxtime) {
        timer.callback();
        queue.pop();
    } else {
        break;
    }
}
\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

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