DoubleBuffered Stream

Author: danngreen

tests/doublebuf_stream_tests.cc

@@ -0,0 +1,93 @@
+#include "doctest.h"
+#include "util/doublebuf_stream.hh"
+
+TEST_CASE("Transmit without errors, always complete transmission before another is started") {
+	std::vector<uint8_t> log;
+
+	DoubleBufferStream<uint8_t, 8> double_buf{[&log](std::span<uint8_t> buf) {
+		log.insert(log.end(), buf.begin(), buf.end());
+		return true;
+	}};
+
+	std::array<uint8_t, 3> a1 = {'A', 'b', 'c'};
+	CHECK(double_buf.transmit(a1));
+	double_buf.tx_done_callback();
+
+	CHECK(log.size() == 3);
+	CHECK(log[0] == 'A');
+	CHECK(log[1] == 'b');
+	CHECK(log[2] == 'c');
+
+	// rely on unsigned overflow:
+	uint32_t word = 0xb0227f;
+	std::array<uint8_t, 3> bytes{uint8_t(word >> 16), uint8_t(word >> 8), uint8_t(word)};
+	CHECK(double_buf.transmit(bytes));
+	double_buf.tx_done_callback();
+
+	CHECK(log.size() == 6);
+	CHECK(log[3] == 0xb0);
+	CHECK(log[4] == 0x22);
+	CHECK(log[5] == 0x7f);
+
+	// Can send up to max size of buffer
+	std::array<uint8_t, 8> a2 = {1, 2, 3, 4, 5, 6, 7, 8};
+	CHECK(double_buf.transmit(a2));
+	double_buf.tx_done_callback();
+
+	CHECK(log.size() == 14);
+	CHECK(log[6] == 1);
+	CHECK(log[7] == 2);
+	CHECK(log[8] == 3);
+	CHECK(log[9] == 4);
+	CHECK(log[10] == 5);
+	CHECK(log[11] == 6);
+	CHECK(log[12] == 7);
+	CHECK(log[13] == 8);
+
+	// Cannot send more then the buffer size
+	std::array<uint8_t, 9> a3 = {1, 2, 3, 4, 5, 6, 7, 8, 9};
+	CHECK_FALSE(double_buf.transmit(a3));
+	CHECK(log.size() == 14);
+}
+
+TEST_CASE("Transmit without errors, interleaving transmissions") {
+	std::vector<uint8_t> log;
+
+	DoubleBufferStream<uint8_t, 8> double_buf{[&log](std::span<uint8_t> buf) {
+		log.insert(log.end(), buf.begin(), buf.end());
+		return true;
+	}};
+
+	std::array<uint8_t, 3> a1 = {1, 2, 3};
+	CHECK(double_buf.transmit(a1));
+
+	std::array<uint8_t, 4> a11 = {4, 5, 6, 7};
+	CHECK(double_buf.transmit(a11));
+
+	std::array<uint8_t, 1> a2 = {8};
+	CHECK(double_buf.transmit(a2));
+
+	std::array<uint8_t, 2> a22 = {9, 10};
+	CHECK(double_buf.transmit(a22));
+
+	// a1 has been sent in the first internal buffer.
+	// but the inactive buffer will not be transmitted until
+	// the tx_done_callback() is called
+	CHECK(log.size() == 3);
+
+	double_buf.tx_done_callback();
+	// Now the inactive buffer, which contains the remaining data,
+	// will be sent
+	CHECK(log.size() == 10);
+
+	CHECK(log[0] == 1);
+	CHECK(log[1] == 2);
+	CHECK(log[2] == 3);
+	CHECK(log[3] == 4);
+	CHECK(log[4] == 5);
+	CHECK(log[5] == 6);
+	CHECK(log[6] == 7);
+	CHECK(log[7] == 8);
+	CHECK(log[8] == 9);
+	CHECK(log[9] == 10);
+}

util/doublebuf_stream.hh

@@ -0,0 +1,99 @@
+#pragma once
+#include "util/fixed_vector.hh"
+#include <functional>
+#include <optional>
+#include <span>
+
+// A double-buffered data transmitter.
+// Allows for calling transmit() at any time, even if a transmission is in progress.
+// Not thread-safe (would need to use std::atomic or semaphors to make that work).
+//
+// Call transmit() to send data using the transmit_func provided in the constructor.
+// When transmission is complete (hardware is ready to send more), then call tx_done_callback().
+//
+// When you call transmit():
+//  -- if a transmission is in progress, then the data will be queued into the inactive buffer
+//  -- if a transmission is NOT in progress, then the data will be sent immediately.
+//
+// When a transmission ends (tx_done_callback is called),
+// if the inactive buffer has data, then a new transmission will be started with it.
+
+// Note: This is not safe if tx_done_callback() interrupts transmit() or start_tx().
+// A valid use case is in the STM USB host library, which does not call
+// the tx callback in an IRQ, so we do not run the risk of tx_done_callback()
+// interrupting transmit() or start_tx();
+
+// #define DEBUG_DBLBUFSTREAM 1
+#if DEBUG_DBLBUFSTREAM
+#define pr_dbg printf
+#else
+#define pr_dbg(...)
+#endif
+
+template<typename T, size_t Size>
+struct DoubleBufferStream {
+	// This size can be adjusted if needed
+	FixedVector<T, Size> tx_buffer[2];
+	std::optional<unsigned> in_progress_idx = std::nullopt;
+
+	std::function<bool(std::span<T>)> transmit_func;
+
+	DoubleBufferStream(std::function<bool(std::span<T>)> xmit_func)
+		: transmit_func{xmit_func} {
+	}
+
+	// Pushes to the inactive buffer and starts transmission if it's not already started
+	bool transmit(std::span<T> word) {
+		auto inactive_idx = 1 - in_progress_idx.value_or(1);
+		auto &inactive_buf = tx_buffer[inactive_idx];
+
+		if (inactive_buf.available() >= word.size()) {
+			for (auto d : word)
+				inactive_buf.push_back(d);
+
+			pr_dbg("DBS::transmit: queued tx data in buffer %u\n", inactive_idx);
+
+			// Start a new transmission if one isn't in progress
+			if (!in_progress_idx.has_value()) {
+				pr_dbg("DBS::transmit: not tx in progress, so starting new tx\n");
+				return start_tx(inactive_idx);
+			}
+
+			// tx_done_callback() will transmit our data when it's called
+			pr_dbg("DBS::transmit: tx is progress, so not starting new tx\n");
+			return true;
+		} else {
+			return false;
+		}
+	}
+
+	void tx_done_callback() {
+		if (!in_progress_idx.has_value()) {
+			pr_dbg("MIDI Host internal error: tx_done_callback called but no buffer is in progress\n");
+			return;
+		}
+
+		tx_buffer[in_progress_idx.value()].clear();
+
+		pr_dbg("DBS::tx_done_callback: finished tx of buffer[%u]\n", in_progress_idx.value());
+
+		// Check if we should start transmitting the other buffer
+		auto other_buffer = 1 - in_progress_idx.value();
+
+		if (tx_buffer[other_buffer].size()) {
+			pr_dbg("DBS: tx_done_callback starting tx for %u\n", other_buffer);
+			start_tx(other_buffer);
+		} else {
+			in_progress_idx = std::nullopt;
+		}
+	}
+
+private:
+	bool start_tx(unsigned idx) {
+		in_progress_idx = idx;
+		pr_dbg("DBS::start_tx: starting tx of buffer[%u]\n", idx);
+		auto &active_buf = tx_buffer[idx];
+		auto res = transmit_func({active_buf.begin(), active_buf.size()});
+		return res;
+	}
+};