[BreakoutBox] Replace MSTP::is_pending_pull_ with a counter

|is_pending_pull_| was used to make sure that frames were made available
to the stream only if there was a pending pull request.

However, if multiple pull requests are issued and then no more pull
requests arrive, some of the corresponding read requests will remain
unsettled when |is_pending_pull_| becomes false.

This CL fixes this by replacing the boolean |is_pending_pull_| with a
counter.

Bug: 1218120
Change-Id: I0ffc038a73a6cd1d387dc17e9c21f2c62fc50613
Reviewed-on: https://chromium-review.googlesource.com/c/chromium/src/+/2950171
Commit-Queue: Guido Urdaneta <guidou@chromium.org>
Reviewed-by: Thomas Guilbert <tguilbert@chromium.org>
Cr-Commit-Position: refs/heads/master@{#891735}

diff --git a/third_party/blink/renderer/modules/breakout_box/frame_queue_underlying_source.cc b/third_party/blink/renderer/modules/breakout_box/frame_queue_underlying_source.cc
index 09bb4315..264d6d9 100644
--- a/third_party/blink/renderer/modules/breakout_box/frame_queue_underlying_source.cc
+++ b/third_party/blink/renderer/modules/breakout_box/frame_queue_underlying_source.cc

@@ -42,23 +42,25 @@
 ScriptPromise FrameQueueUnderlyingSource<NativeFrameType>::pull(
     ScriptState* script_state) {
   DCHECK(realm_task_runner_->RunsTasksInCurrentSequence());
+  {
+    MutexLocker locker(mutex_);
+    num_pending_pulls_++;
+  }
   auto frame_queue = frame_queue_handle_.Queue();
   if (!frame_queue)
     return ScriptPromise::CastUndefined(script_state);
-  if (frame_queue->IsEmpty()) {
-    MutexLocker locker(mutex_);
-    is_pending_pull_ = true;
-    return ScriptPromise::CastUndefined(script_state);
+
+  if (!frame_queue->IsEmpty()) {
+    // Enqueuing the frame in the stream controller synchronously can lead to a
+    // state where the JS code issuing and handling the read requests keeps
+    // executing and prevents other tasks from executing. To avoid this, enqueue
+    // the frame on another task. See https://crbug.com/1216445#c1
+    realm_task_runner_->PostTask(
+        FROM_HERE,
+        WTF::Bind(&FrameQueueUnderlyingSource<
+                      NativeFrameType>::MaybeSendFrameFromQueueToStream,
+                  WrapPersistent(this)));
   }
-  // Enqueuing the frame in the stream controller synchronously can lead to a
-  // state where the JS code issuing and handling the read requests keeps
-  // executing and prevents other tasks from executing. To avoid this, enqueue
-  // the frame on another task. See https://crbug.com/1216445#c1
-  realm_task_runner_->PostTask(
-      FROM_HERE,
-      WTF::Bind(&FrameQueueUnderlyingSource<
-                    NativeFrameType>::MaybeSendFrameFromQueueToStream,
-                WrapPersistent(this)));
   return ScriptPromise::CastUndefined(script_state);
 }
 
@@ -97,7 +99,7 @@
 template <typename NativeFrameType>
 bool FrameQueueUnderlyingSource<NativeFrameType>::HasPendingActivity() const {
   MutexLocker locker(mutex_);
-  return is_pending_pull_ && Controller();
+  return (num_pending_pulls_ > 0) && Controller();
 }
 
 template <typename NativeFrameType>
@@ -131,7 +133,7 @@
   is_closed_ = true;
   {
     MutexLocker locker(mutex_);
-    is_pending_pull_ = false;
+    num_pending_pulls_ = 0;
     if (transferred_source_) {
       PostCrossThreadTask(
           *transferred_source_->GetRealmRunner(), FROM_HERE,
@@ -153,7 +155,7 @@
       transferred_source_->QueueFrame(std::move(media_frame));
       return;
     }
-    should_send_frame_to_stream = is_pending_pull_;
+    should_send_frame_to_stream = num_pending_pulls_ > 0;
   }
 
   auto frame_queue = frame_queue_handle_.Queue();
@@ -178,10 +180,10 @@
 }
 
 template <typename NativeFrameType>
-bool FrameQueueUnderlyingSource<NativeFrameType>::IsPendingPullForTesting()
+int FrameQueueUnderlyingSource<NativeFrameType>::NumPendingPullsForTesting()
     const {
   MutexLocker locker(mutex_);
-  return is_pending_pull_;
+  return num_pending_pulls_;
 }
 
 template <typename NativeFrameType>
@@ -217,14 +219,22 @@
   if (!frame_queue)
     return;
 
-  absl::optional<NativeFrameType> media_frame = frame_queue->Pop();
-  if (!media_frame.has_value())
-    return;
-
-  Controller()->Enqueue(MakeBlinkFrame(std::move(media_frame.value())));
   {
     MutexLocker locker(mutex_);
-    is_pending_pull_ = false;
+    if (num_pending_pulls_ == 0)
+      return;
+  }
+  while (true) {
+    absl::optional<NativeFrameType> media_frame = frame_queue->Pop();
+    if (!media_frame.has_value())
+      return;
+
+    Controller()->Enqueue(MakeBlinkFrame(std::move(media_frame.value())));
+    {
+      MutexLocker locker(mutex_);
+      if (--num_pending_pulls_ == 0)
+        return;
+    }
   }
 }
 

diff --git a/third_party/blink/renderer/modules/breakout_box/frame_queue_underlying_source.h b/third_party/blink/renderer/modules/breakout_box/frame_queue_underlying_source.h
index 55fe478..43479b3 100644
--- a/third_party/blink/renderer/modules/breakout_box/frame_queue_underlying_source.h
+++ b/third_party/blink/renderer/modules/breakout_box/frame_queue_underlying_source.h

@@ -62,7 +62,7 @@
   // Delivers a new frame to this source.
   void QueueFrame(NativeFrameType);
 
-  bool IsPendingPullForTesting() const;
+  int NumPendingPullsForTesting() const;
   double DesiredSizeForTesting() const;
 
   void Trace(Visitor*) const override;
@@ -115,7 +115,7 @@
   // transferred stream.
   CrossThreadPersistent<FrameQueueUnderlyingSource<NativeFrameType>>
       transferred_source_ GUARDED_BY(mutex_);
-  bool is_pending_pull_ GUARDED_BY(mutex_) = false;
+  int num_pending_pulls_ GUARDED_BY(mutex_) = 0;
 };
 
 template <>

diff --git a/third_party/blink/renderer/modules/breakout_box/media_stream_audio_track_underlying_source_test.cc b/third_party/blink/renderer/modules/breakout_box/media_stream_audio_track_underlying_source_test.cc
index 470cd05..e987ab0 100644
--- a/third_party/blink/renderer/modules/breakout_box/media_stream_audio_track_underlying_source_test.cc
+++ b/third_party/blink/renderer/modules/breakout_box/media_stream_audio_track_underlying_source_test.cc

@@ -183,9 +183,9 @@
 
   // Pulling causes a pending pull since there are no frames available for
   // reading.
-  EXPECT_FALSE(source->IsPendingPullForTesting());
+  EXPECT_EQ(source->NumPendingPullsForTesting(), 0);
   source->pull(script_state);
-  EXPECT_TRUE(source->IsPendingPullForTesting());
+  EXPECT_EQ(source->NumPendingPullsForTesting(), 1);
 
   source->Close();
   WebMediaStreamAudioSink::RemoveFromAudioTrack(

diff --git a/third_party/blink/renderer/modules/breakout_box/media_stream_video_track_underlying_source_test.cc b/third_party/blink/renderer/modules/breakout_box/media_stream_video_track_underlying_source_test.cc
index a809f87a..226db4aa9 100644
--- a/third_party/blink/renderer/modules/breakout_box/media_stream_video_track_underlying_source_test.cc
+++ b/third_party/blink/renderer/modules/breakout_box/media_stream_video_track_underlying_source_test.cc

@@ -177,9 +177,9 @@
 
   // Pulling causes a pending pull since there are no frames available for
   // reading.
-  EXPECT_FALSE(source->IsPendingPullForTesting());
+  EXPECT_EQ(source->NumPendingPullsForTesting(), 0);
   source->pull(script_state);
-  EXPECT_TRUE(source->IsPendingPullForTesting());
+  EXPECT_EQ(source->NumPendingPullsForTesting(), 1);
 
   source->Close();
   track->stopTrack(v8_scope.GetExecutionContext());

diff --git a/third_party/blink/web_tests/external/wpt/lint.ignore b/third_party/blink/web_tests/external/wpt/lint.ignore
index 2efb1fc1..98a303e 100644
--- a/third_party/blink/web_tests/external/wpt/lint.ignore
+++ b/third_party/blink/web_tests/external/wpt/lint.ignore

@@ -814,3 +814,5 @@
 DUPLICATE-BASENAME-PATH: dom/nodes/ParentNode-querySelector-All-content.xht
 DUPLICATE-BASENAME-PATH: svg/struct/reftests/reference/green-100x100.html
 DUPLICATE-BASENAME-PATH: svg/struct/reftests/reference/green-100x100.svg
+
+SET TIMEOUT: mediacapture-insertable-streams/MediaStreamTrackProcessor-video.https.html
\ No newline at end of file

diff --git a/third_party/blink/web_tests/external/wpt/mediacapture-insertable-streams/MediaStreamTrackProcessor-video.https.html b/third_party/blink/web_tests/external/wpt/mediacapture-insertable-streams/MediaStreamTrackProcessor-video.https.html
index 7d9ded91..5fe4e04 100644
--- a/third_party/blink/web_tests/external/wpt/mediacapture-insertable-streams/MediaStreamTrackProcessor-video.https.html
+++ b/third_party/blink/web_tests/external/wpt/mediacapture-insertable-streams/MediaStreamTrackProcessor-video.https.html

@@ -45,6 +45,48 @@
                      [processor.readable]);
   return promise;
 }, "Tests that the reader of a video MediaStreamTrackProcessor produces VideoFrame objects and is closed on track stop while running on a worker");
+
+function makeVideoFrame() {
+  const canvas = new OffscreenCanvas(100, 100);
+  const ctx = canvas.getContext('2d');
+  return new VideoFrame(canvas);
+}
+
+promise_test(async t => {
+  // The generator will be used as the source for the processor to
+  // produce frames in a controlled manner.
+  const generator = new MediaStreamTrackGenerator('video');
+  // Use a larger maxBufferSize than the default to ensure no frames
+  // will be dropped.
+  const processor = new MediaStreamTrackProcessor({track: generator, maxBufferSize:10});
+  const reader = processor.readable.getReader();
+  const writer = generator.writable.getWriter();
+
+  let numReads = 0;
+  let resolve = null;
+  const promise = new Promise(r => resolve = r);
+
+  const numOperations = 4;
+  // Issue reads without waiting for the frames to arrive.
+  for (let i = 0; i < numOperations; i++) {
+    reader.read().then(dv=> {
+      dv.value.close();
+      if (++numReads == numOperations)
+        resolve();
+    });
+  }
+
+  // Write video frames in different tasks to "slowly" settle the pending read
+  // requests.
+  for (let i = 0; i<numOperations; i++) {
+     await writer.write(makeVideoFrame());
+     await new Promise(r=>setTimeout(r,0));
+  }
+
+  return promise;
+
+}, "Tests that multiple read requests are eventually settled");
+
 </script>
 </body>
 </html>