Fix ChunkDemuxer seek deadlock

media/filters/chunk_demuxer.cc

 // Copyright (c) 2012 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 "media/filters/chunk_demuxer.h"
 
 #include "base/bind.h"
 #include "base/logging.h"
 #include "base/message_loop.h"
 #include "media/base/audio_decoder_config.h"
@@ skipping 12 matching lines @@
 class ChunkDemuxerStream : public DemuxerStream {
  public:
   typedef std::deque<scoped_refptr<Buffer> > BufferQueue;
   typedef std::deque<ReadCallback> ReadCBQueue;
 
   explicit ChunkDemuxerStream(const AudioDecoderConfig& audio_config);
   explicit ChunkDemuxerStream(const VideoDecoderConfig& video_config);
   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;
 
   void AddBuffers(const BufferQueue& buffers);
   void Shutdown();
 
   bool GetLastBufferTimestamp(base::TimeDelta* timestamp) const;
 
   // DemuxerStream methods.
   virtual void Read(const ReadCallback& read_callback);

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

How is this not breaking the clang bots, missing OVERRIDEs as it is?

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

Not sure. Fixed though.

   virtual Type type();
   virtual void EnableBitstreamConverter();
   virtual const AudioDecoderConfig& audio_decoder_config();
   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
   // monotonically increasing timestamps.
   base::TimeDelta last_buffer_timestamp_;
 
   DISALLOW_IMPLICIT_CONSTRUCTORS(ChunkDemuxerStream);
 };
 
 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);
 }
 
 
 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);
 }
 
 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 {
   base::AutoLock auto_lock(lock_);
 
   // If we haven't seen any buffers yet, then anything can be added.
   if (last_buffer_timestamp_ == kNoTimestamp())
     return true;
 
   if (buffers.empty())
     return true;
 
   return (buffers.front()->GetTimestamp() > last_buffer_timestamp_);
 }
 
 void ChunkDemuxerStream::AddBuffers(const BufferQueue& buffers) {
   if (buffers.empty())
     return;
 
   std::deque<base::Closure> callbacks;

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

s/callbacks/closures/ ?

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

Done.

   {
     base::AutoLock auto_lock(lock_);
 
     for (BufferQueue::const_iterator itr = buffers.begin();
          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();
         if (last_buffer_timestamp_ != kNoTimestamp()) {
           DCHECK_GT(current_ts.ToInternalValue(),
                     last_buffer_timestamp_.ToInternalValue());
         }
 
         last_buffer_timestamp_ = current_ts;
         buffers_.push_back(*itr);
       }
     }
 
-    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()) {
     callbacks.front().Run();
     callbacks.pop_front();

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

If you replace the piecewise pop_front() with a final clear() you get to remove
the braces from the while loop (single-line body) saving a vertical line!
(not to mention the pop_front cycles)

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

I don't follow. Replaced this with a for loop like the one suggested above,
which I think is what you were getting at.

   }
 }
 
 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
   // will be sent.
   while (!callbacks.empty()) {
     callbacks.front().Run(CreateEOSBuffer());
     callbacks.pop_front();

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

pop_front unnecessary

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

Replaced with for loop. 

   }
 }
 
 bool ChunkDemuxerStream::GetLastBufferTimestamp(
     base::TimeDelta* timestamp) const {
   base::AutoLock auto_lock(lock_);
 
   if (buffers_.empty())
     return false;
 
@@ skipping 14 matching lines @@
   read_callback.Run(buffer);
 }
 
 // DemuxerStream methods.
 void ChunkDemuxerStream::Read(const ReadCallback& read_callback) {
   scoped_refptr<Buffer> buffer;
 
   {
     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()));
+    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;
+
+      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;
-      }
 
-      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()));
-        return;
-      }
+      case RECEIVED_END_OF_STREAM:
+        DCHECK(read_cbs_.empty());
 
-      buffer = buffers_.front();
-      buffers_.pop_front();
+        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();
     }
   }
 
   DCHECK(buffer.get());
   read_callback.Run(buffer);
 }
 
 DemuxerStream::Type ChunkDemuxerStream::type() { return type_; }
 
 void ChunkDemuxerStream::EnableBitstreamConverter() {}
 
 const AudioDecoderConfig& ChunkDemuxerStream::audio_decoder_config() {
   CHECK_EQ(type_, AUDIO);
   return audio_config_;
 }
 
 const VideoDecoderConfig& ChunkDemuxerStream::video_decoder_config() {
   CHECK_EQ(type_, VIDEO);
   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),
       buffered_bytes_(0),
       seek_waits_for_data_(true),
       deferred_error_(PIPELINE_OK) {
   DCHECK(client);
 }
 
 ChunkDemuxer::~ChunkDemuxer() {
@@ skipping 36 matching lines @@
 }
 
 void ChunkDemuxer::Seek(base::TimeDelta time, const PipelineStatusCB& cb) {
   DVLOG(1) << "Seek(" << time.InSecondsF() << ")";
 
   PipelineStatus status = PIPELINE_ERROR_INVALID_STATE;
   {
     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;
         return;
       }
 
       status = PIPELINE_OK;
     }
   }
 
@@ skipping 327 matching lines @@
   if (!video_->CanAddBuffers(buffers))
     return false;
 
   video_->AddBuffers(buffers);
   seek_waits_for_data_ = false;
 
   return true;
 }
 
 }  // namespace media