Fix ChunkDemuxer seek deadlock

media/filters/chunk_demuxer.cc

Index: media/filters/chunk_demuxer.cc
diff --git a/media/filters/chunk_demuxer.cc b/media/filters/chunk_demuxer.cc
index 0c27733b13c8dcbf7b7ecdc3984afafccd57a656..b98fe8e8b3a023738af25973f4417d87a4f55baa 100644
--- a/media/filters/chunk_demuxer.cc
+++ b/media/filters/chunk_demuxer.cc
@@ -30,6 +30,7 @@ class ChunkDemuxerStream : public DemuxerStream {
   virtual ~ChunkDemuxerStream();
 
   void Flush();
+  void Seek(base::TimeDelta time);
 
   // Checks if it is ok to add the |buffers| to the stream.
   bool CanAddBuffers(const BufferQueue& buffers) const;
@@ -47,15 +48,27 @@ class ChunkDemuxerStream : public DemuxerStream {
   virtual const VideoDecoderConfig& video_decoder_config();
 
  private:
+  enum State {
+    PLAYING,

[Ami GONE FROM CHROMIUM, 2012/01/28 00:19:59]

Oh, also, PLAYING is a terrible name ;)
NORMAL/STEADY_STATE are better names, though I'm not really happy with either.

The real problem, I think, is that these state names conflate things that have
happened and things that may happen in the future.

Perhaps a good way out of this mess is to write down the state diagram you want
to support (with nodes & labeled edges) and then decide what names should be.

[acolwell GONE FROM CHROMIUM, 2012/01/29 03:00:41]

Agreed. I've renamed all the states. I think they are a little clearer now.
 

+    FLUSHED,
+    FLUSHED_END_OF_STREAM,

[Ami GONE FROM CHROMIUM, 2012/01/27 23:44:58]

These names are not self-explanatory as to what the relevant differences between
them are.

[acolwell GONE FROM CHROMIUM, 2012/01/29 03:00:41]

Done.

+    RECEIVED_END_OF_STREAM,
+    END_OF_STREAM,
+    SHUTDOWN,
+  };
+
+  void ChangeState_Locked(State state);

[Ami GONE FROM CHROMIUM, 2012/01/27 23:44:58]

Commentary here and below please.

[acolwell GONE FROM CHROMIUM, 2012/01/29 03:00:41]

Done.

+  void DeferRead_Locked(const ReadCallback& read_cb);
+  void CreateReadDoneClosures_Locked(std::deque<base::Closure>* closures);
+
   Type type_;
   AudioDecoderConfig audio_config_;
   VideoDecoderConfig video_config_;
 
   mutable base::Lock lock_;
+  State state_;
   ReadCBQueue read_cbs_;
   BufferQueue buffers_;
-  bool shutdown_called_;
-  bool received_end_of_stream_;
 
   // Keeps track of the timestamp of the last buffer we have
   // added to |buffers_|. This is used to enforce buffers with strictly
@@ -67,8 +80,7 @@ class ChunkDemuxerStream : public DemuxerStream {
 
 ChunkDemuxerStream::ChunkDemuxerStream(const AudioDecoderConfig& audio_config)
     : type_(AUDIO),
-      shutdown_called_(false),
-      received_end_of_stream_(false),
+      state_(PLAYING),
       last_buffer_timestamp_(kNoTimestamp()) {
   audio_config_.CopyFrom(audio_config);
 }
@@ -76,8 +88,7 @@ ChunkDemuxerStream::ChunkDemuxerStream(const AudioDecoderConfig& audio_config)
 
 ChunkDemuxerStream::ChunkDemuxerStream(const VideoDecoderConfig& video_config)
     : type_(VIDEO),
-      shutdown_called_(false),
-      received_end_of_stream_(false),
+      state_(PLAYING),
       last_buffer_timestamp_(kNoTimestamp()) {
   video_config_.CopyFrom(video_config);
 }
@@ -86,10 +97,42 @@ ChunkDemuxerStream::~ChunkDemuxerStream() {}
 
 void ChunkDemuxerStream::Flush() {
   DVLOG(1) << "Flush()";
+  std::deque<base::Closure> callbacks;

[Ami GONE FROM CHROMIUM, 2012/01/27 23:44:58]

If you changed this to:
ReadCBQueue read_cbs;
and then under lock:
read_cbs.swap(read_cbs_);
and then outside the lock:
for (ReadCBQueue::iterator it = read_cbs.begin(); it != read_cbs.end(); ++i)
  read_cbs.Run(scoped_refptr<Buffer>());

it would be a lot shorter (and incidentally avoid a bunch of work).

[acolwell GONE FROM CHROMIUM, 2012/01/29 03:00:41]

Done.

+  {
+    base::AutoLock auto_lock(lock_);
+    buffers_.clear();
+    ChangeState_Locked(FLUSHED);
+    last_buffer_timestamp_ = kNoTimestamp();
+
+    // Return null to all pending Read() callbacks to indicate that
+    // we are flushing.
+    while(!read_cbs_.empty()) {
+      callbacks.push_back(base::Bind(read_cbs_.front(),
+                                     scoped_refptr<Buffer>()));
+      read_cbs_.pop_front();
+    }
+  }
+
+  while (!callbacks.empty()) {
+    callbacks.front().Run();
+    callbacks.pop_front();
+  }
+}
+
+void ChunkDemuxerStream::Seek(base::TimeDelta time) {
   base::AutoLock auto_lock(lock_);
-  buffers_.clear();
-  received_end_of_stream_ = false;
-  last_buffer_timestamp_ = kNoTimestamp();
+
+  DCHECK(read_cbs_.empty());
+
+  if (state_ == FLUSHED) {
+    ChangeState_Locked(PLAYING);
+    return;
+  }
+
+  if (state_ == FLUSHED_END_OF_STREAM) {
+    ChangeState_Locked(RECEIVED_END_OF_STREAM);
+    return;
+  }
 }
 
 bool ChunkDemuxerStream::CanAddBuffers(const BufferQueue& buffers) const {
@@ -117,17 +160,15 @@ void ChunkDemuxerStream::AddBuffers(const BufferQueue& buffers) {
          itr != buffers.end(); itr++) {
       // Make sure we aren't trying to add a buffer after we have received and
       // "end of stream" buffer.
-      DCHECK(!received_end_of_stream_);
+      DCHECK_NE(state_, FLUSHED_END_OF_STREAM);
+      DCHECK_NE(state_, RECEIVED_END_OF_STREAM);
+      DCHECK_NE(state_, END_OF_STREAM);
 
       if ((*itr)->IsEndOfStream()) {
-        received_end_of_stream_ = true;
-
-        // Push enough EOS buffers to satisfy outstanding Read() requests.
-        if (read_cbs_.size() > buffers_.size()) {
-          size_t pending_read_without_data = read_cbs_.size() - buffers_.size();
-          for (size_t i = 0; i <= pending_read_without_data; ++i) {
-            buffers_.push_back(*itr);
-          }
+        if (state_ == FLUSHED)  {
+          ChangeState_Locked(FLUSHED_END_OF_STREAM);
+        } else {
+          ChangeState_Locked(RECEIVED_END_OF_STREAM);
         }
       } else {
         base::TimeDelta current_ts = (*itr)->GetTimestamp();
@@ -141,11 +182,7 @@ void ChunkDemuxerStream::AddBuffers(const BufferQueue& buffers) {
       }
     }
 
-    while (!buffers_.empty() && !read_cbs_.empty()) {
-      callbacks.push_back(base::Bind(read_cbs_.front(), buffers_.front()));
-      buffers_.pop_front();
-      read_cbs_.pop_front();
-    }
+    CreateReadDoneClosures_Locked(&callbacks);
   }
 
   while (!callbacks.empty()) {
@@ -158,13 +195,15 @@ void ChunkDemuxerStream::Shutdown() {
   std::deque<ReadCallback> callbacks;
   {
     base::AutoLock auto_lock(lock_);
-    shutdown_called_ = true;
+    ChangeState_Locked(SHUTDOWN);
 
     // Collect all the pending Read() callbacks.
     while (!read_cbs_.empty()) {

[Ami GONE FROM CHROMIUM, 2012/01/27 23:44:58]

std::swap()

[acolwell GONE FROM CHROMIUM, 2012/01/29 03:00:41]

Done.

       callbacks.push_back(read_cbs_.front());
       read_cbs_.pop_front();
     }
+
+    buffers_.clear();
   }
 
   // Pass end of stream buffers to all callbacks to signal that no more data
@@ -206,29 +245,41 @@ void ChunkDemuxerStream::Read(const ReadCallback& read_callback) {
   {
     base::AutoLock auto_lock(lock_);
 
-    if (shutdown_called_ || (received_end_of_stream_ && buffers_.empty())) {
-      buffer = CreateEOSBuffer();
-    } else {
-      if (buffers_.empty()) {
-        // Wrap & store |read_callback| so that it will
-        // get called on the current MessageLoop.
-        read_cbs_.push_back(base::Bind(&RunOnMessageLoop,
-                                       read_callback,
-                                       MessageLoop::current()));
-        return;
-      }
+    switch(state_) {
+      case PLAYING:
+        // Satisfy the request immediately if we have a buffer and
+        // no pending reads.
+        if (buffers_.empty() || !read_cbs_.empty()) {

[Ami GONE FROM CHROMIUM, 2012/01/27 23:44:58]

This test is the opposite of the comment, which I find confusing.
I think it'd be best to reverse the comment, since you actually take advantage
of the coalesced immediately-satisfy path.

[acolwell GONE FROM CHROMIUM, 2012/01/29 03:00:41]

Done.

+          DeferRead_Locked(read_callback);
+          return;
+        }
+
+        buffer = buffers_.front();
+        buffers_.pop_front();
+        break;
 
-      if (!read_cbs_.empty()) {
-        // Wrap & store |read_callback| so that it will
-        // get called on the current MessageLoop.
-        read_cbs_.push_back(base::Bind(&RunOnMessageLoop,
-                                       read_callback,
-                                       MessageLoop::current()));
+      case FLUSHED:
+      case FLUSHED_END_OF_STREAM:

[Ami GONE FROM CHROMIUM, 2012/01/27 23:44:58]

Can these not be fall-throughs to the PLAYING case above?  
(since in these cases there are never buffers?)
(or maybe just add DCHECKs for that assertion here)

[acolwell GONE FROM CHROMIUM, 2012/01/29 03:00:41]

This was a bug. NULLs need to be immediately returned here. Added comment
indicating why.

+        DeferRead_Locked(read_callback);
         return;
-      }
 
-      buffer = buffers_.front();
-      buffers_.pop_front();
+      case RECEIVED_END_OF_STREAM:
+        DCHECK(read_cbs_.empty());
+
+        if (buffers_.empty()) {
+          ChangeState_Locked(END_OF_STREAM);
+          buffer = CreateEOSBuffer();
+        } else {
+          buffer = buffers_.front();
+          buffers_.pop_front();
+        }
+        break;
+
+      case END_OF_STREAM:
+      case SHUTDOWN:
+        DCHECK(buffers_.empty());
+        DCHECK(read_cbs_.empty());
+        buffer = CreateEOSBuffer();
     }
   }
 
@@ -250,6 +301,45 @@ const VideoDecoderConfig& ChunkDemuxerStream::video_decoder_config() {
   return video_config_;
 }
 
+void ChunkDemuxerStream::ChangeState_Locked(State state) {
+  lock_.AssertAcquired();
+  state_ = state;
+}
+
+void ChunkDemuxerStream::DeferRead_Locked(const ReadCallback& read_cb) {
+  lock_.AssertAcquired();
+  // Wrap & store |read_callback| so that it will
+  // get called on the current MessageLoop.
+  read_cbs_.push_back(base::Bind(&RunOnMessageLoop, read_cb,
+                                 MessageLoop::current()));
+}
+
+void ChunkDemuxerStream::CreateReadDoneClosures_Locked(
+    std::deque<base::Closure>* closures) {
+  lock_.AssertAcquired();
+
+  while ((state_ == PLAYING || state_ == RECEIVED_END_OF_STREAM) &&

[Ami GONE FROM CHROMIUM, 2012/01/27 23:44:58]

Having state_ in this while condition is strange (since it isn't going to change
during the life of the loop).

[acolwell GONE FROM CHROMIUM, 2012/01/29 03:00:41]

Done.

+         !buffers_.empty() &&
+         !read_cbs_.empty()) {
+    closures->push_back(base::Bind(read_cbs_.front(), buffers_.front()));
+    buffers_.pop_front();
+    read_cbs_.pop_front();
+  }
+
+  if (state_ == RECEIVED_END_OF_STREAM &&
+      buffers_.empty() && !read_cbs_.empty()) {
+
+    // Push enough EOS buffers to satisfy outstanding Read() requests.
+    scoped_refptr<Buffer> end_of_stream_buffer = CreateEOSBuffer();
+    while(!read_cbs_.empty()) {
+      closures->push_back(base::Bind(read_cbs_.front(), end_of_stream_buffer));
+      read_cbs_.pop_front();
+    }
+
+    ChangeState_Locked(END_OF_STREAM);
+  }
+}
+
 ChunkDemuxer::ChunkDemuxer(ChunkDemuxerClient* client)
     : state_(WAITING_FOR_INIT),
       client_(client),
@@ -306,6 +396,12 @@ void ChunkDemuxer::Seek(base::TimeDelta time, const PipelineStatusCB& cb) {
     base::AutoLock auto_lock(lock_);
 
     if (state_ == INITIALIZED || state_ == ENDED) {
+      if (audio_.get())

[Ami GONE FROM CHROMIUM, 2012/01/27 23:44:58]

.get() here and below are unnecessary (scoped_refptr auto-converts to pointer).

[acolwell GONE FROM CHROMIUM, 2012/01/29 03:00:41]

Done.

+        audio_->Seek(time);
+
+      if (video_.get())
+        video_->Seek(time);
+
       if (seek_waits_for_data_) {
         DVLOG(1) << "Seek() : waiting for more data to arrive.";
         seek_cb_ = cb;