Fix flow control in media::DecoderBase

The flow control in media::DecoderBase was incorrect because it reads too
aggressively to the demuxer stream and failed some DCHECKs when asynchronous
decoding like OpenMAX is used.

An example of a failing case is:
Action    Pending Read    Pending Decode   Read Request
Read           1                0               1
Read           2                0               2
ReadComplete   1                1               2
ReadComplete   0                2               2
DecodeComplete 1                1               1
DecodeComplete 1                0               0

Because of the aggressive read in OnDecodeComplete in DecoderBase. Even
if all the read requests are fulfiled there is still on pending read issued
to the demuxer stream. This mismatch is fixed in this patch.

BUG=32947
TEST=media_unittests

Review URL: http://codereview.chromium.org/660170

git-svn-id: svn://svn.chromium.org/chrome/trunk/src@40189 0039d316-1c4b-4281-b951-d872f2087c98

3 files changed, 146 insertions, 13 deletions

diff --git a/media/filters/decoder_base.h b/media/filters/decoder_base.h
index 5729531..26b1d18 100644
--- a/media/filters/decoder_base.h
+++ b/media/filters/decoder_base.h
@@ -208,13 +208,13 @@ class DecoderBase : public Decoder {
 
     // Enqueue the callback and attempt to fulfill it immediately.
     read_queue_.push_back(read_callback);
-    FulfillPendingRead();
+    if (FulfillPendingRead())
+      return;
 
-    // Issue reads as necessary.
-    while (pending_reads_ < read_queue_.size()) {
-      demuxer_stream_->Read(NewCallback(this, &DecoderBase::OnReadComplete));
-      ++pending_reads_;
-    }
+    // Since we can't fulfill a read request now then submit a read
+    // request to the demuxer stream.
+    demuxer_stream_->Read(NewCallback(this, &DecoderBase::OnReadComplete));
+    ++pending_reads_;
   }
 
   void ReadCompleteTask(scoped_refptr<Buffer> buffer) {
@@ -239,17 +239,16 @@ class DecoderBase : public Decoder {
   void OnDecodeComplete() {
     // Attempt to fulfill a pending read callback and schedule additional reads
     // if necessary.
-    FulfillPendingRead();
+    bool fulfilled = FulfillPendingRead();
 
     // Issue reads as necessary.
     //
     // Note that it's possible for us to decode but not produce a frame, in
     // which case |pending_reads_| will remain less than |read_queue_| so we
     // need to schedule an additional read.
-    // TODO(hclam): Enable this line again to make sure we don't break the
-    // flow control. (BUG=32947)
-    // DCHECK_LE(pending_reads_, read_queue_.size());
-    while (pending_reads_ < read_queue_.size()) {
+    DCHECK_LE(pending_reads_, read_queue_.size());
+    if (!fulfilled) {
+      DCHECK_LT(pending_reads_, read_queue_.size());
       demuxer_stream_->Read(NewCallback(this, &DecoderBase::OnReadComplete));
       ++pending_reads_;
     }
@@ -257,10 +256,12 @@ class DecoderBase : public Decoder {
 
   // Attempts to fulfill a single pending read by dequeuing a buffer and read
   // callback pair and executing the callback.
-  void FulfillPendingRead() {
+  //
+  // Return true if one read request is fulfilled.
+  bool FulfillPendingRead() {
     DCHECK_EQ(MessageLoop::current(), this->message_loop());
     if (read_queue_.empty() || result_queue_.empty()) {
-      return;
+      return false;
     }
 
     // Dequeue a frame and read callback pair.
@@ -271,6 +272,7 @@ class DecoderBase : public Decoder {
 
     // Execute the callback!
     read_callback->Run(output);
+    return true;
   }
 
   // Tracks the number of asynchronous reads issued to |demuxer_stream_|.
diff --git a/media/filters/decoder_base_unittest.cc b/media/filters/decoder_base_unittest.cc
new file mode 100644
index 0000000..26fed48
--- /dev/null
+++ b/media/filters/decoder_base_unittest.cc
@@ -0,0 +1,130 @@
+// Copyright (c) 2010 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include <vector>
+
+#include "media/base/mock_filters.h"
+#include "media/filters/decoder_base.h"
+#include "testing/gmock/include/gmock/gmock.h"
+#include "testing/gtest/include/gtest/gtest.h"
+
+using ::testing::NotNull;
+
+namespace media {
+
+class MockOutput : public StreamSample {
+  MOCK_CONST_METHOD0(IsEndOfStream, bool());
+};
+
+class MockDecoder : public MediaFilter {};
+
+class MockBuffer : public Buffer {
+ public:
+  MOCK_CONST_METHOD0(GetData, const uint8*());
+  MOCK_CONST_METHOD0(GetDataSize, size_t());
+};
+
+ACTION(Initialize) {
+  AutoTaskRunner done_runner(arg2);
+  *arg1 = true;
+}
+
+ACTION_P(SaveDecodeRequest, list) {
+  scoped_refptr<Buffer> buffer = arg0;
+  list->push_back(arg1);
+}
+
+ACTION(CompleteDemuxRequest) {
+  arg0->Run(new MockBuffer());
+  delete arg0;
+}
+
+class MockDecoderImpl : public DecoderBase<MockDecoder, MockOutput> {
+ public:
+  MockDecoderImpl() {
+    media_format_.SetAsString(MediaFormat::kMimeType, "mock");
+  }
+
+  // DecoderBase Implementations.
+  MOCK_METHOD3(DoInitialize,
+               void(DemuxerStream* demuxer_stream, bool* success,
+                    Task* done_cb));
+  MOCK_METHOD0(DoStop, void());
+  MOCK_METHOD2(DoSeek, void(base::TimeDelta time, Task* done_cb));
+  MOCK_METHOD2(DoDecode, void(Buffer* input, Task* done_cb));
+
+ private:
+  FRIEND_TEST(DecoderBaseTest, FlowControl);
+
+  DISALLOW_COPY_AND_ASSIGN(MockDecoderImpl);
+};
+
+class MockReadCallback {
+ public:
+  MockReadCallback() {}
+  MOCK_METHOD1(ReadCallback, void(MockOutput* output));
+
+ private:
+  DISALLOW_COPY_AND_ASSIGN(MockReadCallback);
+};
+
+// Test the flow control of decoder base by the following sequnce of actions:
+// - Read() -> DecoderStream
+//   \ Read() -> DemuxerStream
+// - Read() -> DecoderBase
+//   \ Read() -> DemixerStream
+// - ReadCallback() -> DecoderBase
+//   \ DoDecode() -> Decoder
+// - ReadCallback() -> DecoderBase
+//   \ DoDecode() -> Decoder
+// - DecodeCallback() -> DecoderBase
+//   \ ReadCallback() -> client
+// - DecodeCallback() -> DecoderBase
+//   \ ReadCallback() -> client
+TEST(DecoderBaseTest, FlowControl) {
+  MessageLoop message_loop;
+  scoped_refptr<MockDecoderImpl> decoder = new MockDecoderImpl();
+  scoped_refptr<MockDemuxerStream> demuxer_stream = new MockDemuxerStream();
+  MockFilterCallback callback;
+  MockReadCallback read_callback;
+  decoder->set_message_loop(&message_loop);
+
+  // Initailize.
+  EXPECT_CALL(callback, OnFilterCallback());
+  EXPECT_CALL(callback, OnCallbackDestroyed());
+  EXPECT_CALL(*decoder, DoInitialize(NotNull(), NotNull(), NotNull()))
+      .WillOnce(Initialize());
+  decoder->Initialize(demuxer_stream.get(), callback.NewCallback());
+  message_loop.RunAllPending();
+
+  // Read.
+  std::vector<Task*> decode_requests;
+  EXPECT_CALL(*demuxer_stream, Read(NotNull()))
+      .Times(2)
+      .WillRepeatedly(CompleteDemuxRequest());
+  EXPECT_CALL(*decoder, DoDecode(NotNull(), NotNull()))
+      .Times(2)
+      .WillRepeatedly(SaveDecodeRequest(&decode_requests));
+  decoder->Read(NewCallback(&read_callback,
+                            &MockReadCallback::ReadCallback));
+  decoder->Read(NewCallback(&read_callback,
+                            &MockReadCallback::ReadCallback));
+  message_loop.RunAllPending();
+
+  // Fulfill the decode request.
+  EXPECT_CALL(read_callback, ReadCallback(NotNull())).Times(2);
+  for (size_t i = 0; i < decode_requests.size(); ++i) {
+    decoder->EnqueueResult(new MockOutput());
+    AutoTaskRunner done_cb(decode_requests[i]);
+  }
+  decode_requests.clear();
+  message_loop.RunAllPending();
+
+  // Stop.
+  EXPECT_CALL(*decoder, DoStop());
+  decoder->Stop();
+  message_loop.RunAllPending();
+}
+
+}  // namespace media
diff --git a/media/media.gyp b/media/media.gyp
index e8b8e07..9a2e264 100644
--- a/media/media.gyp
+++ b/media/media.gyp
@@ -190,6 +190,7 @@
         'filters/audio_renderer_algorithm_ola_unittest.cc',
         'filters/audio_renderer_base_unittest.cc',
         'filters/bitstream_converter_unittest.cc',
+        'filters/decoder_base_unittest.cc',
         'filters/ffmpeg_demuxer_unittest.cc',
         'filters/ffmpeg_glue_unittest.cc',
         'filters/ffmpeg_video_decode_engine_unittest.cc',