Support 1-sample consumer queues (fixes #156, #158)

src/consumer_queue.cpp

@@ -1,18 +1,22 @@
 #include "consumer_queue.h"
 #include "common.h"
 #include "send_buffer.h"
+#include <algorithm>
 #include <chrono>
 #include <loguru.hpp>
 #include <utility>
 
 using namespace lsl;
 
+// The lock-free algorithm needs at least 2 slots: with 1, "occupied" and "free for the next
+// round" have the same sequence number. A queue for 1 sample therefore uses 2 slots, and
+// push_sample() keeps at most 1 sample in it (#156, #158).
 consumer_queue::consumer_queue(std::size_t size, send_buffer_p registry)
-	: buffer_(new item_t[size]), size_(size),
+	: buffer_(new item_t[std::max<std::size_t>(size, 2)]), size_(std::max<std::size_t>(size, 2)),
 	  // largest integer at which we can wrap correctly
-	  wrap_at_(std::numeric_limits<std::size_t>::max() - size -
-			   std::numeric_limits<std::size_t>::max() % size),
-	  registry_(std::move(registry)) {
+	  wrap_at_(std::numeric_limits<std::size_t>::max() - size_ -
+			   std::numeric_limits<std::size_t>::max() % size_),
+	  keep_one_(size <= 1), registry_(std::move(registry)) {
 	assert(size_ > 1);
 	for (std::size_t i = 0; i < size_; ++i)
 		buffer_[i].seq_state.store(i, std::memory_order_release);

src/consumer_queue.h

@@ -56,8 +56,8 @@ class consumer_queue {
 	 * This deletes the oldest sample if the max capacity is exceeded.
 	 */
 	template <class T> void push_sample(T &&sample) {
-		while (!try_push(std::forward<T>(sample))) {
-			// buffer full, drop oldest sample
+		while ((keep_one_ && !empty()) || !try_push(std::forward<T>(sample))) {
+			// buffer full (or a 1-sample queue already holds one), drop oldest sample
 			if (!done_sync_.load(std::memory_order_acquire)) {
 				// synchronizes-with store to done_sync_ in ctor
 				std::atomic_thread_fence(std::memory_order_acquire);
@@ -187,6 +187,8 @@ class consumer_queue {
 	const std::size_t size_;
 	/// threshold at which to wrap read/write indices
 	const std::size_t wrap_at_;
+	/// a 1-sample queue: keep at most one sample in the 2 slots (see constructor)
+	const bool keep_one_;
 	/// for use with the condition variable
 	std::mutex mut_;
 

src/data_receiver.cpp

@@ -32,9 +32,11 @@ data_receiver::data_receiver(inlet_connection &conn, int max_buflen, int max_chu
 									 api_config::get_instance()->inlet_buffer_reserve_ms() / 1000)
 				  : api_config::get_instance()->inlet_buffer_reserve_samples())),
 	  check_thread_start_(true), closing_stream_(false), connected_(false),
-	  sample_queue_(max_buflen), max_buflen_(max_buflen), max_chunklen_(max_chunklen) {
-	if (max_buflen < 0)
-		throw std::invalid_argument("The max_buflen argument must not be smaller than 0.");
+	  // checked before the queue is created: a negative size would become a huge allocation
+	  sample_queue_(max_buflen >= 0 ? max_buflen
+									: throw std::invalid_argument(
+										  "The max_buflen argument must not be smaller than 0.")),
+	  max_buflen_(max_buflen), max_chunklen_(max_chunklen) {
 	if (max_chunklen < 0)
 		throw std::invalid_argument("The max_chunklen argument must not be smaller than 0.");
 	conn_.register_onlost(this, &connected_upd_);

testing/int/samples.cpp

@@ -25,6 +25,21 @@ TEST_CASE("consumer_queue", "[queue][basic]") {
 	CHECK(queue.empty());
 }
 
+TEST_CASE("consumer_queue of one sample", "[queue][basic]") {
+	// An inlet or a streamfeed request can ask for a 1-sample buffer (#156, #158).
+	lsl::factory fac(lsl_channel_format_t::cft_int8, 4, 2);
+	lsl::consumer_queue queue(1);
+	for (int i = 0; i < 3; ++i) queue.push_sample(fac.new_sample(i, true));
+
+	// Only the newest sample is kept
+	CHECK(queue.read_available() == 1);
+	auto sample = queue.pop_sample(0.0);
+	REQUIRE(sample);
+	CHECK(static_cast<int>(sample->timestamp()) == 2);
+	CHECK(queue.empty());
+	CHECK_FALSE(queue.pop_sample(0.0));
+}
+
 TEST_CASE("consumer_queue_threaded", "[queue][threads]") {
 	const unsigned int size = 100000;
 	lsl::factory fac(lsl_channel_format_t::cft_int8, 4, 1);