Bug 1993981 - Unit test message atomicity. r=karlt

Differential Revision: https://phabricator.services.mozilla.com/D277180

Author: Pehrsons

dom/media/gtest/TestAudioTrackGraph.cpp

@@ -6,6 +6,7 @@
 #include "DeviceInputTrack.h"
 #include "MediaTrackGraphImpl.h"
 #include "StaticComponents.h"
+#include "VideoUtils.h"
 #include "gmock/gmock.h"
 #include "gtest/gtest-printers.h"
 #include "gtest/gtest.h"
@@ -3642,7 +3643,6 @@ TEST(TestAudioTrackGraph, MessageOrdering)
     InSequence s;
     EXPECT_CALL(checkpoint, Call(StrEq("Prior to tail dispatch")));
     EXPECT_CALL(*processedTrack, AddListenerImpl);
-    EXPECT_CALL(*processedTrack, ProcessInput).Times(AtLeast(0));
     EXPECT_CALL(checkpoint, Call(StrEq("1-main->graph")));
     EXPECT_CALL(*processedTrack, ProcessInput).Times(AtLeast(1));
     EXPECT_CALL(checkpoint, Call(StrEq("processed task on main")))
@@ -3861,6 +3861,119 @@ TEST(TestAudioTrackGraph, MessageOrdering)
   (void)WaitFor(destroyPromise).unwrap()[0];
 }
 
+TEST(TestAudioTrackGraph, MessageAtomicity)
+{
+  MockCubeb* cubeb = new MockCubeb(MockCubeb::RunningMode::Manual);
+  CubebUtils::ForceSetCubebContext(cubeb->AsCubebContext());
+
+  MediaTrackGraphImpl* graph = MediaTrackGraphImpl::GetInstance(
+      MediaTrackGraph::SYSTEM_THREAD_DRIVER, /*Window ID*/ 1,
+      CubebUtils::PreferredSampleRate(/* aShouldResistFingerprinting */ false),
+      nullptr, AbstractThread::MainThread());
+
+  // Mocks and expectations.
+  RefPtr processedTrack = new MockProcessedMediaTrack(graph->GraphRate());
+
+  MockFunction<void(const char* name)> checkpoint;
+  {
+    InSequence s;
+    EXPECT_CALL(*processedTrack, AddListenerImpl);
+    EXPECT_CALL(*processedTrack, ProcessInput).Times(AtLeast(0));
+    // All "Main" graph tasks dispatch in one group and run atomically.
+    EXPECT_CALL(checkpoint, Call(StrEq("Main"))).Times(500);
+    // All "Other" tq tasks dispatch in one group, then dispatch in one group to
+    // the graph, and run atomically.
+    EXPECT_CALL(checkpoint, Call(StrEq("Other"))).Times(500);
+    EXPECT_CALL(*processedTrack, ProcessInput);
+    EXPECT_CALL(*processedTrack, RemoveListenerImpl);
+  }
+
+  // Add a track to maintain an output-only audio driver.
+  RefPtr<OnFallbackListener> fallbackListener;
+  DispatchFunction([&] {
+    graph->AddTrack(processedTrack);
+    processedTrack->AddAudioOutput(reinterpret_cast<void*>(1), nullptr);
+    fallbackListener = new OnFallbackListener(processedTrack);
+    processedTrack->AddListener(fallbackListener);
+  });
+
+  RefPtr<SmartMockCubebStream> stream = WaitFor(cubeb->StreamInitEvent());
+  while (stream->State().isNothing()) {
+    std::this_thread::sleep_for(std::chrono::milliseconds(1));
+  }
+  EXPECT_EQ(*stream->State(), CUBEB_STATE_STARTED);
+  // Wait for the AudioCallbackDriver to come into effect.
+  DispatchFunction([&] {
+    while (fallbackListener->OnFallback()) {
+      EXPECT_EQ(stream->ManualDataCallback(WEBAUDIO_BLOCK_SIZE),
+                MockCubebStream::KeepProcessing::Yes);
+      std::this_thread::sleep_for(std::chrono::milliseconds(1));
+    }
+  });
+
+  // The graph is now run by ManualDataCallback().
+
+  // Run the setup tasks in the graph. They run prior to processing tracks.
+  DispatchFunction([&] {
+    EXPECT_EQ(stream->ManualDataCallback(2 * WEBAUDIO_BLOCK_SIZE),
+              MockCubebStream::KeepProcessing::Yes);
+  });
+
+  auto tq =
+      TaskQueue::Create(GetMediaThreadPool(MediaThreadType::WEBRTC_WORKER),
+                        __func__, /*aSupportsTailDispatch=*/true);
+
+  // Dispatch task A to the graph, then task B to another tail-dispatchable
+  // target, then task C to the graph again. Tail dispatch preserves target
+  // dispatch ordering by default, guaranteeing that a task dispatched from B to
+  // the graph cannot run before A.
+  // The graph requires task atomicity however, meaning that A and C must run in
+  // the same graph iteration.
+  // This is hard to test because the graph drains all direct tasks at the end
+  // of the iteration, rather than in between tasks. Do *many* dispatches and
+  // rely on racing with the other target dispatching to the graph, instead.
+  DispatchFunction([&] {
+    for (size_t i = 0; i < 500; ++i) {
+      EXPECT_TRUE(NS_SUCCEEDED(graph->Dispatch(
+          NS_NewRunnableFunction(__func__, [&] { checkpoint.Call("Main"); }))));
+      EXPECT_TRUE(
+          NS_SUCCEEDED(tq->Dispatch(NS_NewRunnableFunction(__func__, [&] {
+            EXPECT_TRUE(NS_SUCCEEDED(graph->Dispatch(NS_NewRunnableFunction(
+                "TaskQueue.Task", [&] { checkpoint.Call("Other"); }))));
+          }))));
+    }
+  });
+
+  // Run the dispatched functions with tail dispatch after each.
+  ProcessEventQueue();
+
+  // Wait for tq to do all its dispatches.
+  WaitForMirrors(tq);
+
+  // Run graph runnables, at beginning of iteration.
+  DispatchFunction([&] {
+    EXPECT_EQ(stream->ManualDataCallback(WEBAUDIO_BLOCK_SIZE),
+              MockCubebStream::KeepProcessing::Yes);
+  });
+
+  DispatchFunction([&] {
+    processedTrack->RemoveListener(fallbackListener);
+    processedTrack->Destroy();
+  });
+
+  // Process the destroy message and drain the stream.
+  auto destroyPromise = TakeN(cubeb->StreamDestroyEvent(), 1);
+  DispatchFunction([&] {
+    while (stream->ManualDataCallback(0) ==
+           MockCubebStream::KeepProcessing::Yes) {
+    }
+  });
+  // Ensure the stream is no longer used by its MockCubeb before releasing our
+  // reference, and before the next test might ForceSetCubebContext() to
+  // destroy our cubeb.
+  (void)WaitFor(destroyPromise).unwrap()[0];
+}
+
 #undef InvokeAsync
 #undef TEST_WithTailDispatch
 #undef DispatchFunction