Skip to content
Closed
Show file tree
Hide file tree
Changes from 2 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 6 additions & 8 deletions src/consumer_queue.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -2,8 +2,7 @@
#include "common.h"
#include "sample.h"
#include "send_buffer.h"
#include <boost/chrono/duration.hpp>
#include <boost/thread/thread_only.hpp>
#include <chrono>

using namespace lsl;

Expand All @@ -27,6 +26,7 @@ void consumer_queue::push_sample(const sample_p &sample) {
sample_p dummy;
buffer_.pop(dummy);
}
cv_.notify_one();
}

sample_p consumer_queue::pop_sample(double timeout) {
Expand All @@ -35,12 +35,10 @@ sample_p consumer_queue::pop_sample(double timeout) {
buffer_.pop(result);
} else {
if (!buffer_.pop(result)) {
// turn timeout into the point in time at which we give up
timeout += lsl::lsl_clock();
do {
if (lsl::lsl_clock() >= timeout) break;
lslboost::this_thread::sleep_for(lslboost::chrono::milliseconds(1));
} while (!buffer_.pop(result));
// wait untill for a new sample until the thread calling push_sample delivers one, or until timeout
std::unique_lock<std::mutex> lk(lock_);
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If we move this lock to the top of the function, we can have it be thread-safe in the presence of a buffer overflow.

std::chrono::duration<double> sec(timeout);
cv_.wait_for(lk, sec, [&]{ return this->buffer_.pop(result); });
}
}
return result;
Expand Down
5 changes: 5 additions & 0 deletions src/consumer_queue.h
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,8 @@
#include "common.h"
#include "forward.h"
#include <boost/lockfree/spsc_queue.hpp>
#include <mutex>
#include <condition_variable>

namespace lsl {
/**
Expand Down Expand Up @@ -43,6 +45,9 @@ class consumer_queue : private lslboost::noncopyable {
private:
send_buffer_p registry_; // optional consumer registry
buffer_type buffer_; // the sample buffer
// used to wait for new samples
std::mutex lock_;
std::condition_variable cv_;
};

} // namespace lsl
Expand Down