Remove locking from FFmpegDemuxerStream and associated TODOs from media code.

media/filters/ffmpeg_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/ffmpeg_demuxer.h"
 
 #include <algorithm>
 #include <string>
 
 #include "base/bind.h"
@@ skipping 17 matching lines @@
 
 namespace media {
 
 //
 // FFmpegDemuxerStream
 //
 FFmpegDemuxerStream::FFmpegDemuxerStream(
     FFmpegDemuxer* demuxer,
     AVStream* stream)
     : demuxer_(demuxer),
+      message_loop_(base::MessageLoopProxy::current()),
       stream_(stream),
       type_(UNKNOWN),
       stopped_(false),
       last_packet_timestamp_(kNoTimestamp()),
       bitstream_converter_enabled_(false) {
   DCHECK(demuxer_);
 
   // Determine our media format.
   switch (stream->codec->codec_type) {
     case AVMEDIA_TYPE_AUDIO:
@@ skipping 12 matching lines @@
   // Calculate the duration.
   duration_ = ConvertStreamTimestamp(stream->time_base, stream->duration);
 
   if (stream_->codec->codec_id == CODEC_ID_H264) {
     bitstream_converter_.reset(
         new FFmpegH264ToAnnexBBitstreamConverter(stream_->codec));
   }
 }
 
 bool FFmpegDemuxerStream::HasPendingReads() {
-  DCHECK(demuxer_->message_loop()->BelongsToCurrentThread());
-  base::AutoLock auto_lock(lock_);
+  DCHECK(message_loop_->BelongsToCurrentThread());
   DCHECK(!stopped_ || read_queue_.empty())
       << "Read queue should have been emptied if demuxing stream is stopped";
   return !read_queue_.empty();
 }
 
 void FFmpegDemuxerStream::EnqueuePacket(ScopedAVPacket packet) {
-  DCHECK(demuxer_->message_loop()->BelongsToCurrentThread());
+  DCHECK(message_loop_->BelongsToCurrentThread());
 
-  base::AutoLock auto_lock(lock_);
   if (stopped_) {
     NOTREACHED() << "Attempted to enqueue packet on a stopped stream";
     return;
   }
 
   scoped_refptr<DecoderBuffer> buffer;
   if (!packet.get()) {
     buffer = DecoderBuffer::CreateEOSBuffer();
   } else {
     // Convert the packet if there is a bitstream filter.
     if (packet->data && bitstream_converter_enabled_ &&
         !bitstream_converter_->ConvertPacket(packet.get())) {
       LOG(ERROR) << "Format converstion failed.";
     }
 
     // If a packet is returned by FFmpeg's av_parser_parse2() the packet will
     // reference inner memory of FFmpeg.  As such we should transfer the packet
     // into memory we control.
     buffer = DecoderBuffer::CopyFrom(packet->data, packet->size);
     buffer->SetTimestamp(ConvertStreamTimestamp(
         stream_->time_base, packet->pts));
     buffer->SetDuration(ConvertStreamTimestamp(
         stream_->time_base, packet->duration));
     if (buffer->GetTimestamp() != kNoTimestamp() &&
         last_packet_timestamp_ != kNoTimestamp() &&
         last_packet_timestamp_ < buffer->GetTimestamp()) {
       buffered_ranges_.Add(last_packet_timestamp_, buffer->GetTimestamp());
-      demuxer_->message_loop()->PostTask(FROM_HERE, base::Bind(
-          &FFmpegDemuxer::NotifyBufferingChanged, demuxer_));
+      demuxer_->NotifyBufferingChanged();
     }
     last_packet_timestamp_ = buffer->GetTimestamp();
   }
 
   buffer_queue_.push_back(buffer);
-  FulfillPendingRead();
-  return;
+  SatisfyPendingReads();
 }
 
 void FFmpegDemuxerStream::FlushBuffers() {
-  DCHECK(demuxer_->message_loop()->BelongsToCurrentThread());
-  base::AutoLock auto_lock(lock_);
+  DCHECK(message_loop_->BelongsToCurrentThread());
   DCHECK(read_queue_.empty()) << "Read requests should be empty";
   buffer_queue_.clear();
   last_packet_timestamp_ = kNoTimestamp();
 }
 
 void FFmpegDemuxerStream::Stop() {
-  DCHECK(demuxer_->message_loop()->BelongsToCurrentThread());
-  base::AutoLock auto_lock(lock_);
+  DCHECK(message_loop_->BelongsToCurrentThread());
   buffer_queue_.clear();
   for (ReadQueue::iterator it = read_queue_.begin();
        it != read_queue_.end(); ++it) {
     it->Run(DemuxerStream::kOk,
             scoped_refptr<DecoderBuffer>(DecoderBuffer::CreateEOSBuffer()));

[xhwang, 2012/11/20 17:54:34]

CreateEOSBuffer() returns scoped_refptr<DecoderBuffer> already. Can we remove
the copy construction here?

[scherkus (not reviewing), 2012/11/20 19:16:07]

Done.

   }
   read_queue_.clear();
   stopped_ = true;
 }
 
 base::TimeDelta FFmpegDemuxerStream::duration() {
   return duration_;
 }
 
 DemuxerStream::Type FFmpegDemuxerStream::type() {
   return type_;
 }
 
 void FFmpegDemuxerStream::Read(const ReadCB& read_cb) {
-  DCHECK(!read_cb.is_null());
+  if (!message_loop_->BelongsToCurrentThread()) {
+    message_loop_->PostTask(FROM_HERE, base::Bind(
+        &FFmpegDemuxerStream::Read, this, read_cb));
+    return;
+  }
 
-  base::AutoLock auto_lock(lock_);
   // Don't accept any additional reads if we've been told to stop.
-  // The demuxer_ may have been destroyed in the pipleine thread.
+  // The |demuxer_| may have been destroyed in the pipeline thread.
   //
-  // TODO(scherkus): it would be cleaner if we replied with an error message.
+  // TODO(scherkus): it would be cleaner to reply with an error message.
   if (stopped_) {
     read_cb.Run(DemuxerStream::kOk,
                 scoped_refptr<DecoderBuffer>(DecoderBuffer::CreateEOSBuffer()));

[xhwang, 2012/11/20 17:54:34]

ditto

[scherkus (not reviewing), 2012/11/20 19:16:07]

Done.

     return;
   }
 
-  // Buffers are only queued when there are no pending reads.
-  DCHECK(buffer_queue_.empty() || read_queue_.empty());
-
-  if (buffer_queue_.empty()) {
-    demuxer_->message_loop()->PostTask(FROM_HERE, base::Bind(
-        &FFmpegDemuxerStream::ReadTask, this, read_cb));
-    return;
-  }
-
-  // Send the oldest buffer back.
-  scoped_refptr<DecoderBuffer> buffer = buffer_queue_.front();
-  buffer_queue_.pop_front();
-  read_cb.Run(DemuxerStream::kOk, buffer);
-}
-
-void FFmpegDemuxerStream::ReadTask(const ReadCB& read_cb) {
-  DCHECK(demuxer_->message_loop()->BelongsToCurrentThread());
-
-  base::AutoLock auto_lock(lock_);
-  // Don't accept any additional reads if we've been told to stop.
-  //
-  // TODO(scherkus): it would be cleaner if we replied with an error message.
-  if (stopped_) {
-    read_cb.Run(DemuxerStream::kOk,
-                scoped_refptr<DecoderBuffer>(DecoderBuffer::CreateEOSBuffer()));
-    return;
-  }
-
-  // Enqueue the callback and attempt to satisfy it immediately.
   read_queue_.push_back(read_cb);
-  FulfillPendingRead();
-
-  // Check if there are still pending reads, demux some more.
-  if (!read_queue_.empty()) {
-    demuxer_->PostDemuxTask();
-  }
-}
-
-void FFmpegDemuxerStream::FulfillPendingRead() {
-  DCHECK(demuxer_->message_loop()->BelongsToCurrentThread());
-  lock_.AssertAcquired();
-  if (buffer_queue_.empty() || read_queue_.empty()) {
-    return;
-  }
-
-  // Dequeue a buffer and pending read pair.
-  scoped_refptr<DecoderBuffer> buffer = buffer_queue_.front();
-  ReadCB read_cb(read_queue_.front());
-  buffer_queue_.pop_front();
-  read_queue_.pop_front();
-
-  // Execute the callback.
-  read_cb.Run(DemuxerStream::kOk, buffer);
+  SatisfyPendingReads();
 }
 
 void FFmpegDemuxerStream::EnableBitstreamConverter() {
-  base::AutoLock auto_lock(lock_);
+  if (!message_loop_->BelongsToCurrentThread()) {
+    message_loop_->PostTask(FROM_HERE, base::Bind(
+        &FFmpegDemuxerStream::EnableBitstreamConverter, this));
+    return;
+  }
   CHECK(bitstream_converter_.get());
   bitstream_converter_enabled_ = true;
 }
 
 const AudioDecoderConfig& FFmpegDemuxerStream::audio_decoder_config() {
   CHECK_EQ(type_, AUDIO);
   return audio_config_;
 }
 
 const VideoDecoderConfig& FFmpegDemuxerStream::video_decoder_config() {
   CHECK_EQ(type_, VIDEO);
   return video_config_;
 }
 
 FFmpegDemuxerStream::~FFmpegDemuxerStream() {
-  base::AutoLock auto_lock(lock_);
   DCHECK(stopped_);
   DCHECK(read_queue_.empty());
   DCHECK(buffer_queue_.empty());
 }
 
 base::TimeDelta FFmpegDemuxerStream::GetElapsedTime() const {
   return ConvertStreamTimestamp(stream_->time_base, stream_->cur_dts);
 }
 
 Ranges<base::TimeDelta> FFmpegDemuxerStream::GetBufferedRanges() const {
-  base::AutoLock auto_lock(lock_);
   return buffered_ranges_;
 }
 
+void FFmpegDemuxerStream::SatisfyPendingReads() {

[xhwang, 2012/11/20 17:54:34]

Add DCHECK(message_loop_->BelongsToCurrentThread()); ?

[scherkus (not reviewing), 2012/11/20 19:16:07]

Done.

+  while (!read_queue_.empty() && !buffer_queue_.empty()) {
+    ReadCB read_cb = read_queue_.front();
+    read_queue_.pop_front();
+
+    // Send earliest buffer back on a new execution stack to avoid recursing.
+    scoped_refptr<DecoderBuffer> buffer = buffer_queue_.front();
+    buffer_queue_.pop_front();
+    message_loop_->PostTask(FROM_HERE, base::Bind(
+        read_cb, DemuxerStream::kOk, buffer));
+  }
+
+  // No buffers but pending reads? Ask for more!
+  if (!read_queue_.empty() && buffer_queue_.empty()) {
+    demuxer_->NotifyHasPendingRead();
+  }
+}
+
 // static
 base::TimeDelta FFmpegDemuxerStream::ConvertStreamTimestamp(
     const AVRational& time_base, int64 timestamp) {
   if (timestamp == static_cast<int64>(AV_NOPTS_VALUE))
     return kNoTimestamp();
 
   return ConvertFromTimeBase(time_base, timestamp);
 }
 
 //
@@ skipping 11 matching lines @@
       audio_disabled_(false),
       duration_known_(false),
       url_protocol_(data_source, BindToLoop(message_loop_, base::Bind(
           &FFmpegDemuxer::OnDataSourceError, base::Unretained(this)))) {
   DCHECK(message_loop_);
   DCHECK(data_source_);
 }
 
 FFmpegDemuxer::~FFmpegDemuxer() {}
 
-void FFmpegDemuxer::PostDemuxTask() {
-  message_loop_->PostTask(FROM_HERE,
-                          base::Bind(&FFmpegDemuxer::DemuxTask, this));
-}
-
 void FFmpegDemuxer::Stop(const base::Closure& callback) {
   // Post a task to notify the streams to stop as well.
   message_loop_->PostTask(FROM_HERE,
                           base::Bind(&FFmpegDemuxer::StopTask, this, callback));
 }
 
 void FFmpegDemuxer::Seek(base::TimeDelta time, const PipelineStatusCB& cb) {
   message_loop_->PostTask(FROM_HERE,
                           base::Bind(&FFmpegDemuxer::SeekTask, this, time, cb));
 }
@@ skipping 27 matching lines @@
       return *iter;
     }
   }
   return NULL;
 }
 
 base::TimeDelta FFmpegDemuxer::GetStartTime() const {
   return start_time_;
 }
 
-scoped_refptr<base::MessageLoopProxy> FFmpegDemuxer::message_loop() {
-  return message_loop_;
-}
-
 // Helper for calculating the bitrate of the media based on information stored
 // in |format_context| or failing that the size and duration of the media.
 //
 // Returns 0 if a bitrate could not be determined.
 static int CalculateBitrate(
     AVFormatContext* format_context,
     const base::TimeDelta& duration,
     int64 filesize_in_bytes) {
   // If there is a bitrate set on the container, use it.
   if (format_context->bit_rate > 0)
@@ skipping 247 matching lines @@
         }
       }
     }
     // If we have reached the end of stream, tell the downstream filters about
     // the event.
     StreamHasEnded();
     return;
   }
 
   // Queue the packet with the appropriate stream.
-  // TODO(scherkus): should we post this back to the pipeline thread?  I'm
-  // worried about downstream filters (i.e., decoders) executing on this
-  // thread.
   DCHECK_GE(packet->stream_index, 0);
   DCHECK_LT(packet->stream_index, static_cast<int>(streams_.size()));
 
   // Defend against ffmpeg giving us a bad stream index.
   if (packet->stream_index >= 0 &&
       packet->stream_index < static_cast<int>(streams_.size()) &&
       streams_[packet->stream_index] &&
       (!audio_disabled_ ||
        streams_[packet->stream_index]->type() != DemuxerStream::AUDIO)) {
     FFmpegDemuxerStream* demuxer_stream = streams_[packet->stream_index];
     demuxer_stream->EnqueuePacket(packet.Pass());
   }
 
   // Create a loop by posting another task.  This allows seek and message loop
   // quit tasks to get processed.
   if (StreamsHavePendingReads()) {
-    PostDemuxTask();
+    message_loop_->PostTask(FROM_HERE, base::Bind(
+        &FFmpegDemuxer::DemuxTask, this));
   }
 }
 
 void FFmpegDemuxer::StopTask(const base::Closure& callback) {
   DCHECK(message_loop_->BelongsToCurrentThread());
-  StreamVector::iterator iter;
-  for (iter = streams_.begin(); iter != streams_.end(); ++iter) {
-    if (*iter)
-      (*iter)->Stop();
-  }
-
   url_protocol_.Abort();
   data_source_->Stop(BindToLoop(message_loop_, base::Bind(
       &FFmpegDemuxer::OnDataSourceStopped, this, callback)));
 }
 
 void FFmpegDemuxer::OnDataSourceStopped(const base::Closure& callback) {
   // This will block until all tasks complete. Note that after this returns it's
   // possible for reply tasks (e.g., OnReadFrameDone()) to be queued on this
   // thread. Each of the reply task methods must check whether we've stopped the
   // thread and drop their results on the floor.
   DCHECK(message_loop_->BelongsToCurrentThread());
   blocking_thread_.Stop();
+
+  StreamVector::iterator iter;
+  for (iter = streams_.begin(); iter != streams_.end(); ++iter) {
+    if (*iter)
+      (*iter)->Stop();
+  }
   callback.Run();
 }
 
 void FFmpegDemuxer::DisableAudioStreamTask() {
   DCHECK(message_loop_->BelongsToCurrentThread());
   audio_disabled_ = true;
   StreamVector::iterator iter;
   for (iter = streams_.begin(); iter != streams_.end(); ++iter) {
     if (*iter && (*iter)->type() == DemuxerStream::AUDIO) {
       (*iter)->Stop();
@@ skipping 17 matching lines @@
   StreamVector::iterator iter;
   for (iter = streams_.begin(); iter != streams_.end(); ++iter) {
     if (!*iter ||
         (audio_disabled_ && (*iter)->type() == DemuxerStream::AUDIO)) {
       continue;
     }
     (*iter)->EnqueuePacket(ScopedAVPacket());
   }
 }
 
+void FFmpegDemuxer::NotifyHasPendingRead() {

[xhwang, 2012/11/20 17:54:34]

This function is only called once in SatisfyPendingReads(). Do we want to have a
separate function for this?

[scherkus (not reviewing), 2012/11/20 19:16:07]

I think it's fine keep around -- it's clearer than the previous PostDemuxTask()
or calling DemuxTask() directly

+  DCHECK(message_loop_->BelongsToCurrentThread());
+  DemuxTask();
+}
+
 void FFmpegDemuxer::NotifyBufferingChanged() {
   DCHECK(message_loop_->BelongsToCurrentThread());
   Ranges<base::TimeDelta> buffered;
   scoped_refptr<FFmpegDemuxerStream> audio =
       audio_disabled_ ? NULL : GetFFmpegStream(DemuxerStream::AUDIO);
   scoped_refptr<FFmpegDemuxerStream> video =
       GetFFmpegStream(DemuxerStream::VIDEO);
   if (audio && video) {
     buffered = audio->GetBufferedRanges().IntersectionWith(
         video->GetBufferedRanges());
   } else if (audio) {
     buffered = audio->GetBufferedRanges();
   } else if (video) {
     buffered = video->GetBufferedRanges();
   }
   for (size_t i = 0; i < buffered.size(); ++i)
     host_->AddBufferedTimeRange(buffered.start(i), buffered.end(i));
 }
 
 void FFmpegDemuxer::OnDataSourceError() {
   host_->OnDemuxerError(PIPELINE_ERROR_READ);
 }
 
 }  // namespace media