[Chromecast] Chain flush callback in AvStreamerProxy

Record pending callbacks in AvStreamerProxy::StopAndFlush and fire them
one by one when DemuxerStreamerAdapter::Flush is done. In this way, we
can avoid hitting the DCHECK failure in DemuxerStreamerAdapter::Flush as
well as EOS blocking state transition in pipeline.

This is the correct fix for temporary CL
https://codereview.chromium.org/1221443002/

BUG=internal b/22092492

Review URL: https://codereview.chromium.org/1211273007

Cr-Commit-Position: refs/heads/master@{#337642}

5 files changed, 141 insertions, 47 deletions

diff --git a/chromecast/media/cma/ipc_streamer/av_streamer_proxy.cc b/chromecast/media/cma/ipc_streamer/av_streamer_proxy.cc
index 73687d1..09fc5b5 100644
--- a/chromecast/media/cma/ipc_streamer/av_streamer_proxy.cc
+++ b/chromecast/media/cma/ipc_streamer/av_streamer_proxy.cc
@@ -56,6 +56,7 @@ void AvStreamerProxy::Start() {
 }
 
 void AvStreamerProxy::StopAndFlush(const base::Closure& done_cb) {
+  DCHECK(thread_checker_.CalledOnValidThread());
   DCHECK(!done_cb.is_null());
 
   pending_av_data_ = false;
@@ -64,24 +65,27 @@ void AvStreamerProxy::StopAndFlush(const base::Closure& done_cb) {
   pending_buffer_ = scoped_refptr<DecoderBufferBase>();
 
   pending_read_ = false;
+  is_running_ = false;
 
-  // StopAndFlush may happen twice in a row when Stop happens while a previous
-  // pending Seek (which requires Flush). We only need to perform Flush once
-  // when entering stopped state. Chromium pipeline will call Start eventually
-  // to set is_running_, after Seek (next state of Seek is Play), which
-  // guarantees Flush be called when there is no pending tasks.
-  if (is_running_) {
-    frame_provider_->Flush(done_cb);
-  } else {
-    // TODO(yucliu): This is a temporary fix where seek never finish because EOS
-    // reset is_running_ to false. Potential solution is to keep a list of
-    // callbacks. Do not call Flush if the list isn't empty and fire them when
-    // DemuxerStreamAdapter::Flush is done, so that we won't change the order of
-    // the callbacks.
-    done_cb.Run();
+  // If there's another pending Flush, for example, the pipeline is stopped
+  // while another seek is pending, then we don't need to call Flush again. Save
+  // the callback and fire it later when Flush is done.
+  pending_stop_flush_cb_list_.push_back(done_cb);
+  if (pending_stop_flush_cb_list_.size() == 1) {
+    frame_provider_->Flush(
+        base::Bind(&AvStreamerProxy::OnStopAndFlushDone, weak_this_));
   }
+}
 
-  is_running_ = false;
+void AvStreamerProxy::OnStopAndFlushDone() {
+  DCHECK(thread_checker_.CalledOnValidThread());
+
+  // Flush is done. Fire all the "flush done" callbacks in order. This is
+  // necessary to guarantee proper state transition in pipeline.
+  for (const auto& cb : pending_stop_flush_cb_list_) {
+    cb.Run();
+  }
+  pending_stop_flush_cb_list_.clear();
 }
 
 void AvStreamerProxy::OnFifoReadEvent() {
diff --git a/chromecast/media/cma/ipc_streamer/av_streamer_proxy.h b/chromecast/media/cma/ipc_streamer/av_streamer_proxy.h
index e750c85..9d24abc 100644
--- a/chromecast/media/cma/ipc_streamer/av_streamer_proxy.h
+++ b/chromecast/media/cma/ipc_streamer/av_streamer_proxy.h
@@ -5,6 +5,8 @@
 #ifndef CHROMECAST_MEDIA_CMA_IPC_STREAMER_AV_STREAMER_PROXY_H_
 #define CHROMECAST_MEDIA_CMA_IPC_STREAMER_AV_STREAMER_PROXY_H_
 
+#include <list>
+
 #include "base/callback.h"
 #include "base/macros.h"
 #include "base/memory/scoped_ptr.h"
@@ -55,6 +57,8 @@ class AvStreamerProxy {
   bool SendVideoDecoderConfig(const ::media::VideoDecoderConfig& config);
   bool SendBuffer(const scoped_refptr<DecoderBufferBase>& buffer);
 
+  void OnStopAndFlushDone();
+
   base::ThreadChecker thread_checker_;
 
   scoped_ptr<CodedFrameProvider> frame_provider_;
@@ -76,6 +80,9 @@ class AvStreamerProxy {
   ::media::VideoDecoderConfig pending_video_config_;
   scoped_refptr<DecoderBufferBase> pending_buffer_;
 
+  // List of pending callbacks in StopAndFlush
+  std::list<base::Closure> pending_stop_flush_cb_list_;
+
   base::WeakPtr<AvStreamerProxy> weak_this_;
   base::WeakPtrFactory<AvStreamerProxy> weak_factory_;
 
diff --git a/chromecast/media/cma/ipc_streamer/av_streamer_unittest.cc b/chromecast/media/cma/ipc_streamer/av_streamer_unittest.cc
index 3867a0b..9aff31d 100644
--- a/chromecast/media/cma/ipc_streamer/av_streamer_unittest.cc
+++ b/chromecast/media/cma/ipc_streamer/av_streamer_unittest.cc
@@ -55,14 +55,20 @@ class AvStreamerTest : public testing::Test {
   ~AvStreamerTest() override;
 
   // Setups the test.
-  void Configure(
-      size_t frame_count,
-      const std::vector<bool>& provider_delayed_pattern,
-      const std::vector<bool>& consumer_delayed_pattern);
+  void Configure(size_t frame_count,
+                 const std::vector<bool>& provider_delayed_pattern,
+                 const std::vector<bool>& consumer_delayed_pattern,
+                 bool delay_flush);
 
   // Starts the test.
   void Start();
 
+  // Back to back flush
+  void FlushThenStop();
+
+  // Timeout indicates test failure
+  void OnTestTimeout();
+
  protected:
   scoped_ptr<uint64[]> fifo_mem_;
 
@@ -70,13 +76,17 @@ class AvStreamerTest : public testing::Test {
   scoped_ptr<CodedFrameProviderHost> coded_frame_provider_host_;
   scoped_ptr<MockFrameConsumer> frame_consumer_;
 
+  // number of pending cb in StopAndFlush
+  int stop_and_flush_cb_count_;
+
  private:
-  void OnTestTimeout();
   void OnTestCompleted();
 
   void OnFifoRead();
   void OnFifoWrite();
 
+  void OnStopAndFlush();
+
   DISALLOW_COPY_AND_ASSIGN(AvStreamerTest);
 };
 
@@ -89,7 +99,8 @@ AvStreamerTest::~AvStreamerTest() {
 void AvStreamerTest::Configure(
     size_t frame_count,
     const std::vector<bool>& provider_delayed_pattern,
-    const std::vector<bool>& consumer_delayed_pattern) {
+    const std::vector<bool>& consumer_delayed_pattern,
+    bool delay_flush) {
   // Frame generation on the producer and consumer side.
   std::vector<FrameGeneratorForTest::FrameSpec> frame_specs;
   frame_specs.resize(frame_count);
@@ -109,6 +120,7 @@ void AvStreamerTest::Configure(
   scoped_ptr<MockFrameProvider> frame_provider(new MockFrameProvider());
   frame_provider->Configure(provider_delayed_pattern,
                             frame_generator_provider.Pass());
+  frame_provider->SetDelayFlush(delay_flush);
 
   size_t fifo_size_div_8 = 512;
   fifo_mem_.reset(new uint64[fifo_size_div_8]);
@@ -142,6 +154,8 @@ void AvStreamerTest::Configure(
       consumer_delayed_pattern,
       false,
       frame_generator_consumer.Pass());
+
+  stop_and_flush_cb_count_ = 0;
 }
 
 void AvStreamerTest::Start() {
@@ -154,6 +168,19 @@ void AvStreamerTest::Start() {
                  base::Unretained(this)));
 }
 
+void AvStreamerTest::FlushThenStop() {
+  base::Closure cb =
+      base::Bind(&AvStreamerTest::OnStopAndFlush, base::Unretained(this));
+
+  stop_and_flush_cb_count_++;
+  av_buffer_proxy_->StopAndFlush(cb);
+
+  ASSERT_EQ(stop_and_flush_cb_count_, 1);
+
+  stop_and_flush_cb_count_++;
+  av_buffer_proxy_->StopAndFlush(cb);
+}
+
 void AvStreamerTest::OnTestTimeout() {
   ADD_FAILURE() << "Test timed out";
   if (base::MessageLoop::current())
@@ -177,19 +204,26 @@ void AvStreamerTest::OnFifoRead() {
                             base::Unretained(av_buffer_proxy_.get())));
 }
 
+void AvStreamerTest::OnStopAndFlush() {
+  stop_and_flush_cb_count_--;
+  if (stop_and_flush_cb_count_ == 0) {
+    OnTestCompleted();
+  }
+}
+
 TEST_F(AvStreamerTest, FastProviderSlowConsumer) {
   bool provider_delayed_pattern[] = { false };
   bool consumer_delayed_pattern[] = { true };
 
   const size_t frame_count = 100u;
-  Configure(
-      frame_count,
-      std::vector<bool>(
-          provider_delayed_pattern,
-          provider_delayed_pattern + arraysize(provider_delayed_pattern)),
-      std::vector<bool>(
-          consumer_delayed_pattern,
-          consumer_delayed_pattern + arraysize(consumer_delayed_pattern)));
+  Configure(frame_count,
+            std::vector<bool>(
+                provider_delayed_pattern,
+                provider_delayed_pattern + arraysize(provider_delayed_pattern)),
+            std::vector<bool>(
+                consumer_delayed_pattern,
+                consumer_delayed_pattern + arraysize(consumer_delayed_pattern)),
+            false);
 
   scoped_ptr<base::MessageLoop> message_loop(new base::MessageLoop());
   message_loop->PostTask(
@@ -203,14 +237,14 @@ TEST_F(AvStreamerTest, SlowProviderFastConsumer) {
   bool consumer_delayed_pattern[] = { false };
 
   const size_t frame_count = 100u;
-  Configure(
-      frame_count,
-      std::vector<bool>(
-          provider_delayed_pattern,
-          provider_delayed_pattern + arraysize(provider_delayed_pattern)),
-      std::vector<bool>(
-          consumer_delayed_pattern,
-          consumer_delayed_pattern + arraysize(consumer_delayed_pattern)));
+  Configure(frame_count,
+            std::vector<bool>(
+                provider_delayed_pattern,
+                provider_delayed_pattern + arraysize(provider_delayed_pattern)),
+            std::vector<bool>(
+                consumer_delayed_pattern,
+                consumer_delayed_pattern + arraysize(consumer_delayed_pattern)),
+            false);
 
   scoped_ptr<base::MessageLoop> message_loop(new base::MessageLoop());
   message_loop->PostTask(
@@ -232,14 +266,14 @@ TEST_F(AvStreamerTest, SlowFastProducerConsumer) {
   };
 
   const size_t frame_count = 100u;
-  Configure(
-      frame_count,
-      std::vector<bool>(
-          provider_delayed_pattern,
-          provider_delayed_pattern + arraysize(provider_delayed_pattern)),
-      std::vector<bool>(
-          consumer_delayed_pattern,
-          consumer_delayed_pattern + arraysize(consumer_delayed_pattern)));
+  Configure(frame_count,
+            std::vector<bool>(
+                provider_delayed_pattern,
+                provider_delayed_pattern + arraysize(provider_delayed_pattern)),
+            std::vector<bool>(
+                consumer_delayed_pattern,
+                consumer_delayed_pattern + arraysize(consumer_delayed_pattern)),
+            false);
 
   scoped_ptr<base::MessageLoop> message_loop(new base::MessageLoop());
   message_loop->PostTask(
@@ -248,5 +282,41 @@ TEST_F(AvStreamerTest, SlowFastProducerConsumer) {
   message_loop->Run();
 };
 
+// Test case for when AvStreamerProxy::StopAndFlush is invoked while a previous
+// flush is pending. This can happen when pipeline is stopped while a seek/flush
+// is pending.
+TEST_F(AvStreamerTest, StopInFlush) {
+  // We don't care about the delayed pattern. Setting to true to make sure the
+  // test won't exit before flush is called.
+  bool dummy_delayed_pattern[] = {true};
+  std::vector<bool> dummy_delayed_pattern_vector(
+      dummy_delayed_pattern,
+      dummy_delayed_pattern + arraysize(dummy_delayed_pattern));
+  const size_t frame_count = 100u;
+
+  // Delay flush callback in frame provider
+  Configure(frame_count, dummy_delayed_pattern_vector,
+            dummy_delayed_pattern_vector, true);
+
+  scoped_ptr<base::MessageLoop> message_loop(new base::MessageLoop());
+
+  // Flush takes 10ms to finish. 1s timeout is enough for this test.
+  message_loop->PostDelayedTask(
+      FROM_HERE,
+      base::Bind(&AvStreamerTest::OnTestTimeout, base::Unretained(this)),
+      base::TimeDelta::FromSeconds(1));
+
+  message_loop->PostTask(
+      FROM_HERE, base::Bind(&AvStreamerTest::Start, base::Unretained(this)));
+
+  // Let AvStreamerProxy run for a while. Fire flush and stop later
+  message_loop->PostDelayedTask(
+      FROM_HERE,
+      base::Bind(&AvStreamerTest::FlushThenStop, base::Unretained(this)),
+      base::TimeDelta::FromMilliseconds(10));
+
+  message_loop->Run();
+}
+
 }  // namespace media
 }  // namespace chromecast
diff --git a/chromecast/media/cma/test/mock_frame_provider.cc b/chromecast/media/cma/test/mock_frame_provider.cc
index f9beb04..c1de898 100644
--- a/chromecast/media/cma/test/mock_frame_provider.cc
+++ b/chromecast/media/cma/test/mock_frame_provider.cc
@@ -21,7 +21,7 @@
 namespace chromecast {
 namespace media {
 
-MockFrameProvider::MockFrameProvider() {
+MockFrameProvider::MockFrameProvider() : delay_flush_(false) {
 }
 
 MockFrameProvider::~MockFrameProvider() {
@@ -36,6 +36,10 @@ void MockFrameProvider::Configure(
   frame_generator_.reset(frame_generator.release());
 }
 
+void MockFrameProvider::SetDelayFlush(bool delay_flush) {
+  delay_flush_ = delay_flush;
+}
+
 void MockFrameProvider::Read(const ReadCB& read_cb) {
   bool delayed = delayed_task_pattern_[pattern_idx_];
   pattern_idx_ = (pattern_idx_ + 1) % delayed_task_pattern_.size();
@@ -53,7 +57,12 @@ void MockFrameProvider::Read(const ReadCB& read_cb) {
 }
 
 void MockFrameProvider::Flush(const base::Closure& flush_cb) {
-  flush_cb.Run();
+  if (delay_flush_) {
+    base::ThreadTaskRunnerHandle::Get()->PostDelayedTask(
+        FROM_HERE, flush_cb, base::TimeDelta::FromMilliseconds(10));
+  } else {
+    flush_cb.Run();
+  }
 }
 
 void MockFrameProvider::DoRead(const ReadCB& read_cb) {
diff --git a/chromecast/media/cma/test/mock_frame_provider.h b/chromecast/media/cma/test/mock_frame_provider.h
index 31aa305..b83503e 100644
--- a/chromecast/media/cma/test/mock_frame_provider.h
+++ b/chromecast/media/cma/test/mock_frame_provider.h
@@ -22,6 +22,7 @@ class MockFrameProvider : public CodedFrameProvider {
   void Configure(
       const std::vector<bool>& delayed_task_pattern,
       scoped_ptr<FrameGeneratorForTest> frame_generator);
+  void SetDelayFlush(bool delay_flush);
 
   // CodedFrameProvider implementation.
   void Read(const ReadCB& read_cb) override;
@@ -35,8 +36,11 @@ class MockFrameProvider : public CodedFrameProvider {
   // i.e. after receiving a Read request, either delivers a frame right away
   // or wait some time before delivering the frame.
   // |pattern_idx_| is the current index in the pattern.
+  // |delay_flush_| indicates whether to delay flush cb in Flush. Default is
+  // false.
   std::vector<bool> delayed_task_pattern_;
   size_t pattern_idx_;
+  bool delay_flush_;
 
   scoped_ptr<FrameGeneratorForTest> frame_generator_;