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"
 #include "media/base/data_buffer.h"
 #include "media/base/video_decoder_config.h"
 #include "media/filters/chunk_demuxer_client.h"
 #include "media/webm/webm_stream_parser.h"
 
 namespace media {
 
 // Create an "end of stream" buffer.
 static Buffer* CreateEOSBuffer() {
   return new DataBuffer(0);
 }
 
 class ChunkDemuxerStream : public DemuxerStream {
  public:
   typedef std::deque<scoped_refptr<Buffer> > BufferQueue;
   typedef std::deque<ReadCallback> ReadCBQueue;
+  typedef std::deque<base::Closure> ClosureQueue;
 
   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);
-  virtual Type type();
-  virtual void EnableBitstreamConverter();
-  virtual const AudioDecoderConfig& audio_decoder_config();
-  virtual const VideoDecoderConfig& video_decoder_config();
+  virtual void Read(const ReadCallback& read_callback) OVERRIDE;
+  virtual Type type() OVERRIDE;
+  virtual void EnableBitstreamConverter() OVERRIDE;
+  virtual const AudioDecoderConfig& audio_decoder_config() OVERRIDE;
+  virtual const VideoDecoderConfig& video_decoder_config() OVERRIDE;
 
  private:
+  enum State {
+    RETURNING_DATA_FOR_READS,
+    WAITING_FOR_SEEK,
+    RECEIVED_EOS_WHILE_WAITING_FOR_SEEK, // EOS = End of stream.
+    RECEIVED_END_OF_STREAM,

[Ami GONE FROM CHROMIUM, 2012/01/29 22:12:38]

s/END_OF_STREAM/EOS/ here and below to match the previous value.

[acolwell GONE FROM CHROMIUM, 2012/01/30 00:14:14]

Done.

+    SENT_END_OF_STREAM,
+    SHUTDOWN,
+  };
+
+  // Assigns |state_| to |state|
+  void ChangeState_Locked(State state);
+
+  // Adds the callback to |read_cbs_| so it can be called later when we
+  // have data.
+  void DeferRead_Locked(const ReadCallback& read_cb);
+
+  // Creates closures that bind ReadCallbacks in |read_cbs_| to data in
+  // |buffers_|.

[Ami GONE FROM CHROMIUM, 2012/01/29 22:12:38]

It also pops the respective queues, though, right?

[acolwell GONE FROM CHROMIUM, 2012/01/30 00:14:14]

Done.

+  void CreateReadDoneClosures_Locked(ClosureQueue* 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_(RETURNING_DATA_FOR_READS),
       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_(RETURNING_DATA_FOR_READS),
       last_buffer_timestamp_(kNoTimestamp()) {
   video_config_.CopyFrom(video_config);
 }
 
 ChunkDemuxerStream::~ChunkDemuxerStream() {}
 
 void ChunkDemuxerStream::Flush() {
   DVLOG(1) << "Flush()";
+  ReadCBQueue read_cbs;
+  {
+    base::AutoLock auto_lock(lock_);
+    buffers_.clear();
+    ChangeState_Locked(WAITING_FOR_SEEK);
+    last_buffer_timestamp_ = kNoTimestamp();
+
+    std::swap(read_cbs_, read_cbs);
+  }
+
+  for (ReadCBQueue::iterator it = read_cbs.begin(); it != read_cbs.end(); ++it)
+    it->Run(scoped_refptr<Buffer>());
+}
+
+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_ == WAITING_FOR_SEEK) {
+    ChangeState_Locked(RETURNING_DATA_FOR_READS);
+    return;
+  }
+
+  if (state_ == RECEIVED_EOS_WHILE_WAITING_FOR_SEEK) {
+    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;
+  ClosureQueue closures;
   {
     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_, RECEIVED_EOS_WHILE_WAITING_FOR_SEEK);
+      DCHECK_NE(state_, RECEIVED_END_OF_STREAM);
+      DCHECK_NE(state_, SENT_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_ == WAITING_FOR_SEEK)  {
+          ChangeState_Locked(RECEIVED_EOS_WHILE_WAITING_FOR_SEEK);
+        } 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(&closures);
   }
 
-  while (!callbacks.empty()) {
-    callbacks.front().Run();
-    callbacks.pop_front();
-  }
+  for (ClosureQueue::iterator it = closures.begin(); it != closures.end(); ++it)
+    it->Run();
 }
 
 void ChunkDemuxerStream::Shutdown() {
-  std::deque<ReadCallback> callbacks;
+  ReadCBQueue read_cbs;
   {
     base::AutoLock auto_lock(lock_);
-    shutdown_called_ = true;
+    ChangeState_Locked(SHUTDOWN);
 
-    // Collect all the pending Read() callbacks.
-    while (!read_cbs_.empty()) {
-      callbacks.push_back(read_cbs_.front());
-      read_cbs_.pop_front();
-    }
+    std::swap(read_cbs_, read_cbs);
+    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();
-  }
+  for (ReadCBQueue::iterator it = read_cbs.begin(); it != read_cbs.end(); ++it)
+    it->Run(CreateEOSBuffer());
 }
 
 bool ChunkDemuxerStream::GetLastBufferTimestamp(
     base::TimeDelta* timestamp) const {
   base::AutoLock auto_lock(lock_);
 
   if (buffers_.empty())
     return false;
 
   *timestamp = buffers_.back()->GetTimestamp();
@@ skipping 13 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()));
-        return;
-      }
+    switch(state_) {
+      case RETURNING_DATA_FOR_READS:
+        // If we don't have any buffers ready or already have
+        // pending reads, then defer this read.
+        if (buffers_.empty() || !read_cbs_.empty()) {
+          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;
-      }
+        buffer = buffers_.front();
+        buffers_.pop_front();
+        break;
 
-      buffer = buffers_.front();
-      buffers_.pop_front();
+      case WAITING_FOR_SEEK:
+      case RECEIVED_EOS_WHILE_WAITING_FOR_SEEK:
+        // Null buffers should be returned in this state since we are waiting
+        // for a seek. Any buffers in |buffers_| should NOT be returned because
+        // they are associated with the seek.
+        DCHECK(read_cbs_.empty());
+        break;
+      case RECEIVED_END_OF_STREAM:
+        DCHECK(read_cbs_.empty());
+
+        if (buffers_.empty()) {
+          ChangeState_Locked(SENT_END_OF_STREAM);
+          buffer = CreateEOSBuffer();
+        } else {
+          buffer = buffers_.front();
+          buffers_.pop_front();
+        }
+        break;
+
+      case SENT_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(
+    ClosureQueue* closures) {

[Ami GONE FROM CHROMIUM, 2012/01/29 22:12:38]

one line?

[acolwell GONE FROM CHROMIUM, 2012/01/30 00:14:14]

Done.

+  lock_.AssertAcquired();
+
+  if (state_ != RETURNING_DATA_FOR_READS && state_ != RECEIVED_END_OF_STREAM)
+    return;
+
+  while (!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 &&

[Ami GONE FROM CHROMIUM, 2012/01/29 22:12:38]

reverse test and early return.

[acolwell GONE FROM CHROMIUM, 2012/01/30 00:14:14]

Done.

+      buffers_.empty() && !read_cbs_.empty()) {
+

[Ami GONE FROM CHROMIUM, 2012/01/29 22:12:38]

extra \n

[acolwell GONE FROM CHROMIUM, 2012/01/30 00:14:14]

Done.

+    // 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(SENT_END_OF_STREAM);

[Ami GONE FROM CHROMIUM, 2012/01/29 22:12:38]

It's a tad sketchy that this state change happens before the EOS buffers are
actually sent (they're only queued up for sending in *closures at this point).

[acolwell GONE FROM CHROMIUM, 2012/01/30 00:14:14]

True. I think the state name is the problem. I've renamed it to
RETURNING_EOS_FOR_READS since that is what happens once we enter this state.

+  }
+}
+
 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_)
+        audio_->Seek(time);
+
+      if (video_)
+        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