Design review
This code is symptomatic of a lot of the problems I see in concurrent code these days:
- 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.
- 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.)
- 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:
- create the data
- push the data to the shared data store
- 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.
