From 7cfaeb0c2bbf22348a2df73558edb9ed8f19c327 Mon Sep 17 00:00:00 2001 From: "Jonas Ohland (Riedel)" Date: Tue, 4 Aug 2026 01:05:51 +0200 Subject: [PATCH 1/4] Add a test to check if continuous flow writers are notified Co-authored-by: Jonas Ohland Signed-off-by: Jonas Ohland (Riedel) --- lib/tests/test_flows.cpp | 50 ++++++++++++++++++++++++++++++++++++++++ 1 file changed, 50 insertions(+) diff --git a/lib/tests/test_flows.cpp b/lib/tests/test_flows.cpp index cd1797fae..465c4b307 100644 --- a/lib/tests/test_flows.cpp +++ b/lib/tests/test_flows.cpp @@ -12,6 +12,7 @@ # include #endif +#include #include #include #include @@ -1024,3 +1025,52 @@ TEST_CASE_PERSISTENT_FIXTURE(mxl::tests::mxlDomainFixture, "mxlCreateFlow: unwri permissions(domain, std::filesystem::perms::owner_all, std::filesystem::perm_options::add); remove_all(domain); } + +TEST_CASE_PERSISTENT_FIXTURE(mxl::tests::mxlDomainFixture, "mxlFlowWriterCommitSamples should notify the reader", "[mxl flows][futex]") +{ + auto flowDef = mxl::tests::readFile("data/audio_flow.json"); + + auto instance = mxlCreateInstance(domain.c_str(), nullptr); + REQUIRE(instance != nullptr); + + mxlFlowWriter writer = nullptr; + mxlFlowReader reader = nullptr; + auto configInfo = mxlFlowConfigInfo{}; + + REQUIRE(mxlCreateFlowWriter(instance, flowDef.c_str(), nullptr, &writer, &configInfo, nullptr) == MXL_STATUS_OK); + REQUIRE(mxlCreateFlowReader(instance, "b3bb5be7-9fe9-4324-a5bb-4c70e1084449", nullptr, &reader) == MXL_STATUS_OK); + + auto stillWriting = std::atomic_flag{true}; + auto writerThread = std::thread{[&]() + { + auto slice = mxlMutableWrappedMultiBufferSlice{}; + for (auto i = std::uint64_t{0}; i < 100; ++i) + { + REQUIRE(mxlFlowWriterOpenSamples(writer, 1000 + (i * 480), 480, &slice) == MXL_STATUS_OK); + REQUIRE(mxlFlowWriterCommitSamples(writer) == MXL_STATUS_OK); + std::this_thread::sleep_for(std::chrono::milliseconds(10)); + } + stillWriting.clear(); + }}; + + // read in a loop until `stillWriting` is cleared + auto slices = mxlWrappedMultiBufferSlice{}; + auto readBlock = std::uint64_t{0}; + for (;;) + { + auto const status = mxlFlowReaderGetSamples(reader, 1000 + (readBlock * 480), 480, 100000000, &slices); + if (!stillWriting.test_and_set()) + { + break; + } + + REQUIRE(status == MXL_STATUS_OK); + ++readBlock; + } + + writerThread.join(); + REQUIRE(mxlReleaseFlowReader(instance, reader) == MXL_STATUS_OK); + REQUIRE(mxlReleaseFlowWriter(instance, writer) == MXL_STATUS_OK); + + REQUIRE(mxlDestroyInstance(instance) == MXL_STATUS_OK); +} From 5f8052b705ab8a7c7eb9762b73720fecb23827dd Mon Sep 17 00:00:00 2001 From: "Jonas Ohland (Riedel)" Date: Tue, 4 Aug 2026 01:06:35 +0200 Subject: [PATCH 2/4] Fix PosixContinuousFlowWriter setting _currentIndex undefined early Co-authored-by: Jonas Ohland Signed-off-by: Jonas Ohland (Riedel) --- lib/internal/src/PosixContinuousFlowWriter.cpp | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/lib/internal/src/PosixContinuousFlowWriter.cpp b/lib/internal/src/PosixContinuousFlowWriter.cpp index 9b74ee8c5..c68cd2e0c 100644 --- a/lib/internal/src/PosixContinuousFlowWriter.cpp +++ b/lib/internal/src/PosixContinuousFlowWriter.cpp @@ -111,7 +111,6 @@ namespace mxl::lib { auto const flow = _flowData->flow(); flow->info.runtime.headIndex = _currentIndex; - _currentIndex = MXL_UNDEFINED_INDEX; if (signalCompletedBatch()) { @@ -120,6 +119,7 @@ namespace mxl::lib wakeAll(&flow->state.syncCounter); } + _currentIndex = MXL_UNDEFINED_INDEX; return MXL_STATUS_OK; } else From 3b044c72b7b338682dbaa19bcb6098bd2019edcb Mon Sep 17 00:00:00 2001 From: "Jonas Ohland (Riedel)" Date: Tue, 4 Aug 2026 11:11:27 +0200 Subject: [PATCH 3/4] Add additional assert to continuous flow futex test Co-authored-by: Jonas Ohland Signed-off-by: Jonas Ohland (Riedel) --- lib/tests/test_flows.cpp | 22 +++++++++++++--------- 1 file changed, 13 insertions(+), 9 deletions(-) diff --git a/lib/tests/test_flows.cpp b/lib/tests/test_flows.cpp index 465c4b307..1613f1c8e 100644 --- a/lib/tests/test_flows.cpp +++ b/lib/tests/test_flows.cpp @@ -1028,15 +1028,17 @@ TEST_CASE_PERSISTENT_FIXTURE(mxl::tests::mxlDomainFixture, "mxlCreateFlow: unwri TEST_CASE_PERSISTENT_FIXTURE(mxl::tests::mxlDomainFixture, "mxlFlowWriterCommitSamples should notify the reader", "[mxl flows][futex]") { + constexpr auto const startIndex = 1000; + constexpr auto const blockSize = 480; + constexpr auto const iterations = std::uint64_t{50}; + constexpr auto const readTimeout = std::chrono::duration_cast(std::chrono::milliseconds(50)); auto flowDef = mxl::tests::readFile("data/audio_flow.json"); - + auto configInfo = mxlFlowConfigInfo{}; auto instance = mxlCreateInstance(domain.c_str(), nullptr); - REQUIRE(instance != nullptr); - mxlFlowWriter writer = nullptr; mxlFlowReader reader = nullptr; - auto configInfo = mxlFlowConfigInfo{}; + REQUIRE(instance != nullptr); REQUIRE(mxlCreateFlowWriter(instance, flowDef.c_str(), nullptr, &writer, &configInfo, nullptr) == MXL_STATUS_OK); REQUIRE(mxlCreateFlowReader(instance, "b3bb5be7-9fe9-4324-a5bb-4c70e1084449", nullptr, &reader) == MXL_STATUS_OK); @@ -1044,9 +1046,9 @@ TEST_CASE_PERSISTENT_FIXTURE(mxl::tests::mxlDomainFixture, "mxlFlowWriterCommitS auto writerThread = std::thread{[&]() { auto slice = mxlMutableWrappedMultiBufferSlice{}; - for (auto i = std::uint64_t{0}; i < 100; ++i) + for (auto i = std::uint64_t{0}; i < iterations; ++i) { - REQUIRE(mxlFlowWriterOpenSamples(writer, 1000 + (i * 480), 480, &slice) == MXL_STATUS_OK); + REQUIRE(mxlFlowWriterOpenSamples(writer, startIndex + (i * blockSize), blockSize, &slice) == MXL_STATUS_OK); REQUIRE(mxlFlowWriterCommitSamples(writer) == MXL_STATUS_OK); std::this_thread::sleep_for(std::chrono::milliseconds(10)); } @@ -1055,20 +1057,22 @@ TEST_CASE_PERSISTENT_FIXTURE(mxl::tests::mxlDomainFixture, "mxlFlowWriterCommitS // read in a loop until `stillWriting` is cleared auto slices = mxlWrappedMultiBufferSlice{}; - auto readBlock = std::uint64_t{0}; + auto currentBlockIndex = std::uint64_t{0}; for (;;) { - auto const status = mxlFlowReaderGetSamples(reader, 1000 + (readBlock * 480), 480, 100000000, &slices); + auto const status = mxlFlowReaderGetSamples(reader, startIndex + (currentBlockIndex * blockSize), blockSize, readTimeout.count(), &slices); if (!stillWriting.test_and_set()) { break; } REQUIRE(status == MXL_STATUS_OK); - ++readBlock; + ++currentBlockIndex; } writerThread.join(); + REQUIRE(currentBlockIndex == iterations); + REQUIRE(mxlReleaseFlowReader(instance, reader) == MXL_STATUS_OK); REQUIRE(mxlReleaseFlowWriter(instance, writer) == MXL_STATUS_OK); From 9e18d75a7f6ba6c31d1c4b90e1eaa0d6828f49e5 Mon Sep 17 00:00:00 2001 From: "Jonas Ohland (Riedel)" Date: Tue, 4 Aug 2026 19:05:42 +0200 Subject: [PATCH 4/4] Use head index instead of _currentIndex in PosixContinuousFlowWriter Co-authored-by: Jonas Ohland Signed-off-by: Jonas Ohland (Riedel) --- lib/internal/src/PosixContinuousFlowWriter.cpp | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/lib/internal/src/PosixContinuousFlowWriter.cpp b/lib/internal/src/PosixContinuousFlowWriter.cpp index c68cd2e0c..c63095f2c 100644 --- a/lib/internal/src/PosixContinuousFlowWriter.cpp +++ b/lib/internal/src/PosixContinuousFlowWriter.cpp @@ -111,6 +111,7 @@ namespace mxl::lib { auto const flow = _flowData->flow(); flow->info.runtime.headIndex = _currentIndex; + _currentIndex = MXL_UNDEFINED_INDEX; if (signalCompletedBatch()) { @@ -119,7 +120,6 @@ namespace mxl::lib wakeAll(&flow->state.syncCounter); } - _currentIndex = MXL_UNDEFINED_INDEX; return MXL_STATUS_OK; } else @@ -137,7 +137,8 @@ namespace mxl::lib bool PosixContinuousFlowWriter::signalCompletedBatch() noexcept { - auto const currentSyncSampleBatch = _currentIndex / _syncBatchSize; + auto const flow = _flowData->flow(); + auto const currentSyncSampleBatch = flow->info.runtime.headIndex / _syncBatchSize; if (currentSyncSampleBatch < _lastSyncSampleBatch) { return false;