1
\$\begingroup\$

I am a newbie in std::thread and std::condition_variable.

I would like to use 2 threads: one for display (void showCalibration()) and the other for processing (void CaptureFrame()). These threads access a global variable (temp_pupil).

The display thread will clear the variable after a specific time to get the data (pupil) from processing thread (during that time).

I used loops to "simulate" capture loop from a high speed camera (void CaptureFrame()) and 9 calibration points displayed on the screen (void showCalibration()

#include <opencv2/core/core.hpp>
#include <iostream>
#include <thread>
#include <mutex>
std::vector<cv::Point2d> temp_pupil;
std::mutex mu;
std::condition_variable cond;

void showCalibration()
{
    for (int i = 0; i <= 9; i++)//plot 9 points sequentially
    {
        
        std::unique_lock<std::mutex> lk(mu);
        {
            temp_pupil.clear();//prepare for get new data
        }
        lk.unlock();
        cond.notify_one();//signal CaptureFrame() to capture data
        std::this_thread::sleep_for(std::chrono::milliseconds(100));//specific time

        lk.lock();
        {
            while (temp_pupil.empty())
            {
                cond.wait(lk);//wait CaptureFrame() to end current point
            }
            std::string cal = std::to_string(temp_pupil.size()) + "Cali" + std::to_string(i) + "\n";
            std::cout << cal;//print
        }
        lk.unlock();
    }
}

void CaptureFrame()
{
    cv::Point2d pupil;
    for (int i = 0; i <= 500; i++)
    {
        std::unique_lock<std::mutex> lk(mu);
        {
            pupil = cv::Point2d(i, i);//create data
            temp_pupil.push_back(pupil);//get data
            std::string cap = std::to_string(temp_pupil.size()) + "Captue" + std::to_string(i) + "\n";
            std::cout << cap;//print
        }
        lk.unlock();
        cond.notify_one();//signal showCalibration() to end current point
    }
}

int main()
{
    std::thread cal_thread(showCalibration);//display thread
    std::thread cap_thread(CaptureFrame);//process thread
    cal_thread.join();
    cap_thread.join();
}

Could anyone help me review above code ?

\$\endgroup\$
2
  • \$\begingroup\$ Welcome to the Code Review Community. It is better when the title indicates what the code does rather than what your concerns about the code are. Could you please change the title and move your comment into the body of the question. To improve your question please read How do I ask a good question?. \$\endgroup\$
    – pacmaninbw
    Jan 30 at 13:24
  • \$\begingroup\$ @pacmaninbw, thank you for your guide. \$\endgroup\$ Jan 30 at 23:35
2
\$\begingroup\$

Design review

This code is symptomatic of a lot of the problems I see in concurrent code these days:

  1. There is just too much synchronization cruft. Just take a look at showCalibration(): there are roughly 10 functional lines, and 6–8 of them are dealing with synchronization.
  2. There is an over-reliance on fancy synchronization. Quite often I see code that uses mutexes, condition variables, futures, semaphores, and more, when literally a single atomic flag could do the same job. (Condition variables in particular seem to be all the hype these days; everybody is using them when there’s no real need.)
  3. The entire program is designed around the synchronization, rather than around what it’s actually supposed to be doing. The actual algorithm is an afterthought. (Quite literally in this case.)

Good concurrent code doesn’t look like synchronization spaghetti, with the actual algorithm buried in a mess of locks and waits. Good concurrent code looks more or less the same as non-concurrent code. If you can’t figure out the actual algorithm being performed because it’s obscured by synchronization mess, then you are not writing good concurrent code.

Take a step back and look at what you’re actually trying to do. You have two threads, one producing values (pupil points), and one consuming them. Ignoring all synchronization (and stop tokens, etc.), your producer thread is basically just:

auto producer_thread()
{
    while (not done)
    {
        auto point = /* figure out point somehow */;

        points.push(std::move(point));
    }
}

And your consumer thread is basically just:

auto consumer_thread()
{
    while (not done)
    {
        if (auto const point = points.try_pop(); point)
        {
            // use *point somehow
        }
        else
        {
            // there are no points ready, so just give up our time (unless
            // there is some other computation we could do while waiting)
            std::this_thread::yield();
        }
    }
}

That’s it. That’s all you need.

Notice there are no visible locks, or condition variables, or any other such garbage. There is nothing in the code that isn’t clearly for the purpose of getting new points, or using them. (Well, except for that else block in the consumer thread, and that’s optional, though it makes a huge perfomance difference because your consumer loop won’t be wasting time spinning when there are no points available. And you could just do a pop() rather than try_pop(), and just block and wait for points instead.)

So is it actually possible to write the ideal code shown above, and still get highly performant and correct concurrency? Yes, yes indeed, and the key is right there in the code. Mutexes, locks, and condition variables are the wrong abstraction. What you want is a concurrent queue.

Now there is no concurrent queue in the standard library (yet! there is one proposed!), but you could write your own (bad idea), or use an existing one—there are several excellent implementations in Boost, TBB, and more.

I am going to make a very simple concurrent queue just for illustration. DO NOT DO THIS. Use a properly-written concurrent queue, do not roll your own. The one below will “work”, but it is terribly non-performant. It is just for illustration:

template <typename T, typename Allocator = std::allocator<T>>
class concurrent_queue
{
    std::deque<T, Allocator>    _data;
    std::mutex                  _mutex;

public:
    concurrent_queue() noexcept = default;
    concurrent_queue(Allocator const& a) : _data(a) {}

    auto push(T t)
    {
        auto lock = std::scoped_lock(_mutex);
        _data.push_back(std::move(t));
    }

    auto try_pop()
    {
        auto result = std::optional<T>{};

        auto lock = std::scoped_lock(_mutex);
        if (not _data.empty())
        {
            result = std::move(_data.front());
            _data.pop_front();
        }

        return result;
    }
};

And now your whole program basically just becomes this:

// define these to do what you actually want
auto CaptureFrame() -> cv::Point2d;
auto showCalibration(cv::Point2d) -> void;

auto producer_thread(std::stop_token stop_token, concurrent_queue<cv::Point2d>& points)
{
    while (not stop_token.stop_requested())
    {
        auto point = CaptureFrame();

        points.push(std::move(point));
    }
}

auto consumer_thread(std::stop_token stop_token, concurrent_queue<cv::Point2d>& points)
{
    while (not stop_token.stop_requested())
    {
        if (auto const point = points.try_pop(); point)
        {
            showCalibration(*point);
        }
        else
        {
            std::this_thread::yield();
        }
    }
}

auto main() -> int
{
    auto points = concurrent_queue<cv::Point2d>{};

    std::jthread cal_thread{consumer_thread, std::ref(points)};
    std::jthread cap_thread{producer_thread, std::ref(points)};

    // automatic clean shutdown and cleanup
}

This is just the most basic structure. Rather than getting a single point in the consumer thread, you could also try to pump the queue to get a whole bunch of points in a single go, then process them all at once. But with a really well designed concurrent queue—especially one with a non-blocking interface—there’s really no need for that.

Also, note that I’ve used C++20 jthreads and stop_tokens. If you can’t use C++20, no big deal. Just replace the stop_tokens with an atomic<bool>, and pass a reference to it in the thread constructors, then replace the jthreads with regular threads and manually set the atomic<bool> to signal shutdown, and manually join the threads:

auto producer_thread(std::atomic<bool>& done, concurrent_queue<cv::Point2d>& points)
{
    while (not done) // maybe using acquire or relaxed semantics
    {
        // ... [snip] ...
    }
}

auto consumer_thread(std::atomic<bool>& done, concurrent_queue<cv::Point2d>& points)
{
    while (not done) // maybe using acquire or relaxed semantics
    {
        // ... [snip] ...
    }
}

auto main() -> int
{
    auto points = concurrent_queue<cv::Point2d>{};

    auto done = std::atomic<bool>{false};
    std::thread cal_thread{consumer_thread, std::ref(done), std::ref(points)};
    std::thread cap_thread{producer_thread, std::ref(done), std::ref(points)};

    done = true; // maybe using release or relaxed semantics
    cal_thread.join();
    cap_thread.join();
}

Same diff, just more manual work.

HOWEVER!

Even that might be too much for this job!

The point of a concurrent queue is that one (or more than one!) thread can produce values while another (or several!) use those values… and—and this is the key point—none of those values are lost. But! If you are processing a video feed, and the camera frames are being generated at, say, 25 per second, while you can only process 10 per second… then usually it’s okay for you to just drop a few frames here or there. So you might not even need a queue at all. You might be able to get away with a single shared value. Something like this:

auto producer_thread(std::mutex& mutex, std::size_t& sequence_number, cv::Point2d& point)
{
    while (not done)
    {
        auto p = /* figure out point somehow */;

        auto lock = std::scoped_lock(mutex);
        point = std::move(p);
        ++sequence_number;
    }
}

auto consumer_thread(std::mutex& mutex, std::size_t& sequence_number, cv::Point2d const& point)
{
    auto last_sequence_number = sequence_number;

    while (not done)
    {
        auto has_point = false;
        auto p = cv::Point2d{};

        {
            auto lock = std::scoped_lock(mutex);

            // if the sequence number is unchanged, then the point data is old
            //
            // if it has changed, then there is new point data, so copy it,
            // update the sequence number, and then release the lock right
            // away
            if (sequence_number != last_sequence_number)
            {
                last_sequence_number = sequence_number;
                p = point;
                has_point = true;
            }
        }

        if (has_point)
        {
            // use p somehow
        }
        else
        {
            std::this_thread::yield();
        }
    }
}

And that’s without any kind of optimization whatsoever. If you wanted to get clever, then you might even be able to make a really compact type that encodes the point and sequence number in just, say, 8 bytes (like, 1 byte for the sequence number (0–255), 3 bytes each for the two coordinates (0–16777215 each), with a byte to spare for something else), and squeeze all that into a single 64-bit atomic value. That will be freakin’ fast, especially if you leverage C++20 atomic wait and notify:

class your_type
{
    std::fast_uint64_t _data;

public:
    constexpr your_type() noexcept = default;

    constexpr your_type(std::size_t sequence_number, cv::Point2d point)
    {
        // compact sequence number and point into _data
    }

    explicit constexpr operator cv::Point2d() const noexcept
    {
        // extract the point
    }
};

auto producer_thread(std::atomic<your_type>& data)
{
    auto sequence_number = std::size_t{};

    while (not done)
    {
        auto point = /* figure out point somehow */;

        data = your_type{sequence_number, point}; // maybe with release
        data.notify_one();
    }
}

auto consumer_thread(std::atomic<your_type>& data)
{
    auto last_data = your_type{};

    while (not done)
    {
        data.wait(last_data); // maybe with acquire

        auto const point = cv::Point2d{data};
        // use point somehow
    }
}

Note the code above doesn’t take into account how to detect that there is no more data. But that’s trivial to add: just use that last unused byte to signal “closed”.

Code review

std::vector<cv::Point2d> temp_pupil;
std::mutex mu;
std::condition_variable cond;

Global variables are bad. There’s no reason you can’t have these as local variables in main() (for example), and then pass references to them to the thread functions.

Also mu isn’t a great name for a variable… especially a global one.

In showCalibration():

for (int i = 0; i <= 9; i++)//plot 9 points sequentially

Reeeaaaallllyyy? Did you test that it plots 9 points sequentially?

Here’s what happened when I tested it:

for (int i = 0; i <= 9; i++)//plot 9 points sequentially
    std::cout << "plotting point #" << (i + 1) << " sequentially\n";

// Output:
plotting point #1 sequentially
plotting point #2 sequentially
plotting point #3 sequentially
plotting point #4 sequentially
plotting point #5 sequentially
plotting point #6 sequentially
plotting point #7 sequentially
plotting point #8 sequentially
plotting point #9 sequentially
plotting point #10 sequentially

Hm, that’s not 9 points. 10 ≠ 9.

There are two important lessons here, a small one and a big one.

The small lesson is that the standard form for a for loops is for (auto i = 0; i < N; ++i). Note the condition: it’s i < N… not i <= N.

The big lesson is that this is why the experts say not to write naked loops. Naked loops are error-prone. They’re also unclear. You have no idea what a loop is doing without reading the body of the loop (and even then it’s sometimes not clear). A MUCH better practice is to use algorithms. For example, if you wrote an rigorously tested a repeat<N>() function, you could use that, and it would be clear and safe:

template <std::size_t N, std::invocable F>
auto repeat(F&& f)
{
    for (auto i = std::size_t(0); i != N; ++i)
        f();
}

template <std::size_t N, std::invocable<std::size_t> F>
auto repeat(F&& f)
{
    for (auto i = std::size_t(0); i != N; ++i)
        f(i);
}

void showCalibration()
{
    repeat<9>([&](auto i) // self-documenting that you just want to repeat something 9 times
    {
        std::unique_lock<std::mutex> lk(mu);
        // ... and so on ...
    });
}

Now the biggest problem in showCalibration() (and in general) is that you go a little lock-crazy. You lock too often, hold the locks for too long, and reuse locks (which is generally a terrible idea, because it makes it harder to reason about when a lock is held or not).

Take a look at the first few lines of showCalibration():

std::unique_lock<std::mutex> lk(mu);
{
    temp_pupil.clear();//prepare for get new data
}
lk.unlock();

Okaaaay… but… why not this:

{
    auto lock = std::scoped_lock(mu);
    temp_pupil.clear();
}

That’s not only one line shorter, it’s also clearer about what the lock is really for, and how long it’s being held.

Now, this line is baffling, for several reasons:

cond.notify_one();//signal CaptureFrame() to capture data

The first problem is they you notify on this condition… but CaptureFrame() doesn’t actually wait on that condition.

Nor does it make any sense that it should. If CaptureFrame() is capturing frames from a camera, it doesn’t really need the consumer thread’s permission to capture the next frame. It should just capture the next frame as soon as it can; waiting on the processing of the previous frame or whatever defeats the purpose of doing this concurrently.

The other big problem with this is that you’re abusing condition variables. A condition variable should represent a single condition. But you are using cond (which is a terrible name for a condition variable, because it leads to exactly this kind of problem) for two conditions: “ready for the next frame to be captured” and “the next frame has already been captured and is ready for processing”.

If you actually wanted to wait on both those conditions, you would need two condition variables. But of course… you don’t really want to wait for permission to capture the next frame. (And you don’t even need to wait on the condition that the next frame is ready for processing, either.)

lk.lock();
{
    while (temp_pupil.empty())
    {
        cond.wait(lk);//wait CaptureFrame() to end current point
    }
    std::string cal = std::to_string(temp_pupil.size()) + "Cali" + std::to_string(i) + "\n";
    std::cout << cal;//print
}
lk.unlock();

Okay, reusing locks is bad. But the larger problem is that your function is complex enough that it needs to reuse a lock. If your function locks, unlocks, then relocks, and unlocks… that’s usually a sign that it should be two functions.

while (temp_pupil.empty())
{
    cond.wait(lk);//wait CaptureFrame() to end current point
}

Never wait on condition variables like this. Writing a loop like this is just an invitation to confusion and bugs.

Instead, do this:

cond.wait(lk, [] { return not temp_pupil.empty(); });

This is clean, self-contained, and impossible for any future maintainer to misunderstand or fuck up.

So, overall, this block of code can become:

{
    auto lock = std::unique_lock(mu);
    cond.wait(lock, [] { return not temp_pupil.empty(); });

    std::cout << temp_pupil.size() << "Cali" << i << '\n';
}

But let’s look at the big picture in this function:

void showCalibration()
{
    repeat<9>([&](auto i)
    {
        {
            auto lock = std::scoped_lock(mu);
            temp_pupil.clear();
        }

        std::this_thread::sleep_for(std::chrono::milliseconds(100));

        {
            auto lock = std::unique_lock(mu);
            cond.wait(lock, [] { return not temp_pupil.empty(); });

            std::cout << temp_pupil.size() << "Cali" << i << '\n';
        }
    });
}

Does it really make sense for the processing thread to clear the queue, then go to sleep, then wake up and start processing data? Doesn’t it make more sense to just wake up, process the data, clear the queue when done with the data, then go back to sleep?

void showCalibration()
{
    repeat<9>([&](auto i)
    {
        auto lock = std::unique_lock(mu);
        cond.wait(lock, [] { return not temp_pupil.empty(); });

        // process the data somehow...
        std::cout << temp_pupil.size() << "Cali" << i << '\n';

        temp_pupil.clear();
    });
}

But this holds the lock way too long, so, even better would be:

void showCalibration()
{
    repeat<9>([&](auto i)
    {
        auto data = std::vector<cv::Point2d>{};

        {
            auto lock = std::unique_lock(mu);
            cond.wait(lock, [] { return not temp_pupil.empty(); });

            std::swap(data, temp_pupil);
        }

        // process the data somehow...
        std::cout << data.size() << "Cali" << i << '\n'; // you may have to synchronize std::cout
                                                         // but I know it's just being used and example, so I'll ignore it
    });
}

And if you wanted to be more efficient and reuse the allocated memory, you could move data out of the loop, and manually clear it before swapping (or at the end of the loop). (If you wanted to simulate a processing delay, you could insert the sleep operation where the processing comment is, I suppose.)

void CaptureFrame()
{
    cv::Point2d pupil;

You generally shouldn’t declare variables until you need them.

for (int i = 0; i <= 500; i++)

Is this meant to be for (int i = 0; i < 500; i++)?

std::unique_lock<std::mutex> lk(mu);
{
    // ... [snip] ...
}
lk.unlock();

There is no reason to use a unique lock, or to manually unlock here. Just do:

{
    auto lock = std::scoped_lock(mu);
    // ... [snip] ...
}

Now, take a step back, and look at the big picture in CaptureFrame(). What this function should do is:

  1. create the data
  2. push the data to the shared data store
  3. notify the waiters that there is data

The lock on the shared data is only necessary for step 2.

That means:

void CaptureFrame()
{
    repeat<500>([&](auto i)
    {
        auto pupil = cv::Point2d(i, i); //create data

        std::cout << temp_pupil.size() << "Captue" << i << '\n'; // don't forget to synchronize std::cout

        {
            auto lock = std::scoped_lock(mu);
            temp_pupil.push_back(std::move(pupil));
        }

        cond.notify_one();
    });
}

The moral of the story is that concurrency and synchronization shouldn’t swallow up your entire program. Merely doing a bunch of stuff concurrently is not the point of your program: your program is (presumably) doing something actually useful. So focus on that, design your program around that functionality, and only add concurrency and synchronization sparingly, and don’t let it become the whole focus of the program.

And also, don’t look at what’s in the standard library, and then design your synchronization around that. Look at what your program needs, and then see what the standard library offers. And if it doesn’t offer something (which is very common), then look at third party libraries. The point is: figure out what your program needs and then look for the tools that will accomplish that… don’t figure out what tools you have on hand and then jigger your program around in order to use them.

And don’t bother with condition variables. They’re inefficient and difficult to use correctly. You almost never really need them.

\$\endgroup\$
2
  • \$\begingroup\$ Sorry for my late respond. Thank you for your detail review and guide. I will implement your instructions. \$\endgroup\$ Jan 30 at 23:28
  • \$\begingroup\$ Sorry for my unclear purpose about '''std::this_thread::sleep_for(std::chrono::milliseconds(100));'''. I used it to "simulate" animate points. My idea is that the data will be gotten in this period only. In calibration procedure, subjects were asked to fixate on 9 points displayed sequentially on the computer monitor. For each stimulating point, the average coordinates of the pupil and cornea reflection’s centers on 100 eye images were calculated. \$\endgroup\$ Jan 31 at 4:14

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.