diff --git a/lib/internal/src/PosixContinuousFlowWriter.cpp b/lib/internal/src/PosixContinuousFlowWriter.cpp index 9b74ee8c5..c63095f2c 100644 --- a/lib/internal/src/PosixContinuousFlowWriter.cpp +++ b/lib/internal/src/PosixContinuousFlowWriter.cpp @@ -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; diff --git a/lib/tests/test_flows.cpp b/lib/tests/test_flows.cpp index cd1797fae..1613f1c8e 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,56 @@ 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]") +{ + 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); + mxlFlowWriter writer = nullptr; + mxlFlowReader reader = nullptr; + + 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); + + auto stillWriting = std::atomic_flag{true}; + auto writerThread = std::thread{[&]() + { + auto slice = mxlMutableWrappedMultiBufferSlice{}; + for (auto i = std::uint64_t{0}; i < iterations; ++i) + { + 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)); + } + stillWriting.clear(); + }}; + + // read in a loop until `stillWriting` is cleared + auto slices = mxlWrappedMultiBufferSlice{}; + auto currentBlockIndex = std::uint64_t{0}; + for (;;) + { + auto const status = mxlFlowReaderGetSamples(reader, startIndex + (currentBlockIndex * blockSize), blockSize, readTimeout.count(), &slices); + if (!stillWriting.test_and_set()) + { + break; + } + + REQUIRE(status == MXL_STATUS_OK); + ++currentBlockIndex; + } + + writerThread.join(); + REQUIRE(currentBlockIndex == iterations); + + REQUIRE(mxlReleaseFlowReader(instance, reader) == MXL_STATUS_OK); + REQUIRE(mxlReleaseFlowWriter(instance, writer) == MXL_STATUS_OK); + + REQUIRE(mxlDestroyInstance(instance) == MXL_STATUS_OK); +}