Fix continuous flow futex notification

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;

lib/tests/test_flows.cpp

@@ -12,6 +12,7 @@
 #   include <UdpLayer.h>
 #endif
 
+#include <chrono>
 #include <cstdint>
 #include <cstdlib>
 #include <cstring>
@@ -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::nanoseconds>(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);
+}