Replace MessageLoopProxy with SingleThreadTaskRunner for media/filters/ + associated 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/base64.h"
 #include "base/bind.h"
 #include "base/callback.h"
 #include "base/callback_helpers.h"
 #include "base/command_line.h"
 #include "base/memory/scoped_ptr.h"
-#include "base/message_loop/message_loop.h"
 #include "base/metrics/sparse_histogram.h"
 #include "base/stl_util.h"
 #include "base/strings/string_util.h"
 #include "base/strings/stringprintf.h"
 #include "base/sys_byteorder.h"
 #include "base/task_runner_util.h"
 #include "base/time/time.h"
 #include "media/base/audio_decoder_config.h"
 #include "media/base/bind_to_loop.h"
 #include "media/base/decoder_buffer.h"
 #include "media/base/decrypt_config.h"
 #include "media/base/limits.h"
 #include "media/base/media_log.h"
 #include "media/base/media_switches.h"
 #include "media/base/video_decoder_config.h"
 #include "media/ffmpeg/ffmpeg_common.h"
 #include "media/filters/ffmpeg_glue.h"
 #include "media/filters/ffmpeg_h264_to_annex_b_bitstream_converter.h"
 #include "media/webm/webm_crypto_helpers.h"
 
 namespace media {
 
 //
 // FFmpegDemuxerStream
 //
 FFmpegDemuxerStream::FFmpegDemuxerStream(
     FFmpegDemuxer* demuxer,
+    const scoped_refptr<base::SingleThreadTaskRunner>& task_runner,
     AVStream* stream)
     : demuxer_(demuxer),
-      message_loop_(base::MessageLoopProxy::current()),
+      task_runner_(task_runner),

[Ami GONE FROM CHROMIUM, 2013/11/08 16:54:46]

I find this version slightly implying that task_runner_ might be different to
MLP::current().  Care to add a DCHECK to the ctor that
task_runner_->RunsOnCurrentThread()?

But really, what is this change buying you over just doing
task_runner_(b::MLP::current())
?

[scherkus (not reviewing), 2013/11/08 21:51:39]

This change was induced by globally removing all message_loop #includes and
fixing the compilation errors.

Reverted back to MLP but narrowed include to m_l_p.h

       stream_(stream),
       type_(UNKNOWN),
       end_of_stream_(false),
       last_packet_timestamp_(kNoTimestamp()),
       bitstream_converter_enabled_(false) {
   DCHECK(demuxer_);
 
   bool is_encrypted = false;
 
   // Determine our media format.
@@ skipping 34 matching lines @@
     DCHECK(!enc_key_id.empty());
     if (enc_key_id.empty())
       return;
 
     encryption_key_id_.assign(enc_key_id);
     demuxer_->FireNeedKey(kWebMEncryptInitDataType, enc_key_id);
   }
 }
 
 void FFmpegDemuxerStream::EnqueuePacket(ScopedAVPacket packet) {
-  DCHECK(message_loop_->BelongsToCurrentThread());
+  DCHECK(task_runner_->BelongsToCurrentThread());
 
   if (!demuxer_ || end_of_stream_) {
     NOTREACHED() << "Attempted to enqueue packet on a stopped stream";
     return;
   }
 
   // Convert the packet if there is a bitstream filter.
   if (packet->data && bitstream_converter_enabled_ &&
       !bitstream_converter_->ConvertPacket(packet.get())) {
     LOG(ERROR) << "Format conversion failed.";
@@ skipping 58 matching lines @@
     buffered_ranges_.Add(last_packet_timestamp_, buffer->timestamp());
     demuxer_->NotifyBufferingChanged();
   }
   last_packet_timestamp_ = buffer->timestamp();
 
   buffer_queue_.Push(buffer);
   SatisfyPendingRead();
 }
 
 void FFmpegDemuxerStream::SetEndOfStream() {
-  DCHECK(message_loop_->BelongsToCurrentThread());
+  DCHECK(task_runner_->BelongsToCurrentThread());
   end_of_stream_ = true;
   SatisfyPendingRead();
 }
 
 void FFmpegDemuxerStream::FlushBuffers() {
-  DCHECK(message_loop_->BelongsToCurrentThread());
+  DCHECK(task_runner_->BelongsToCurrentThread());
   DCHECK(read_cb_.is_null()) << "There should be no pending read";
   buffer_queue_.Clear();
   end_of_stream_ = false;
   last_packet_timestamp_ = kNoTimestamp();
 }
 
 void FFmpegDemuxerStream::Stop() {
-  DCHECK(message_loop_->BelongsToCurrentThread());
+  DCHECK(task_runner_->BelongsToCurrentThread());
   buffer_queue_.Clear();
   if (!read_cb_.is_null()) {
     base::ResetAndReturn(&read_cb_).Run(
         DemuxerStream::kOk, DecoderBuffer::CreateEOSBuffer());
   }
   demuxer_ = NULL;
   stream_ = NULL;
   end_of_stream_ = true;
 }
 
 base::TimeDelta FFmpegDemuxerStream::duration() {
   return duration_;
 }
 
 DemuxerStream::Type FFmpegDemuxerStream::type() {
-  DCHECK(message_loop_->BelongsToCurrentThread());
+  DCHECK(task_runner_->BelongsToCurrentThread());
   return type_;
 }
 
 void FFmpegDemuxerStream::Read(const ReadCB& read_cb) {
-  DCHECK(message_loop_->BelongsToCurrentThread());
+  DCHECK(task_runner_->BelongsToCurrentThread());
   CHECK(read_cb_.is_null()) << "Overlapping reads are not supported";
   read_cb_ = BindToCurrentLoop(read_cb);
 
   // Don't accept any additional reads if we've been told to stop.
   // The |demuxer_| may have been destroyed in the pipeline thread.
   //
   // TODO(scherkus): it would be cleaner to reply with an error message.
   if (!demuxer_) {
     base::ResetAndReturn(&read_cb_).Run(
         DemuxerStream::kOk, DecoderBuffer::CreateEOSBuffer());
     return;
   }
 
   SatisfyPendingRead();
 }
 
 void FFmpegDemuxerStream::EnableBitstreamConverter() {
-  DCHECK(message_loop_->BelongsToCurrentThread());
+  DCHECK(task_runner_->BelongsToCurrentThread());
   CHECK(bitstream_converter_.get());
   bitstream_converter_enabled_ = true;
 }
 
 AudioDecoderConfig FFmpegDemuxerStream::audio_decoder_config() {
-  DCHECK(message_loop_->BelongsToCurrentThread());
+  DCHECK(task_runner_->BelongsToCurrentThread());
   CHECK_EQ(type_, AUDIO);
   return audio_config_;
 }
 
 VideoDecoderConfig FFmpegDemuxerStream::video_decoder_config() {
-  DCHECK(message_loop_->BelongsToCurrentThread());
+  DCHECK(task_runner_->BelongsToCurrentThread());
   CHECK_EQ(type_, VIDEO);
   return video_config_;
 }
 
 FFmpegDemuxerStream::~FFmpegDemuxerStream() {
   DCHECK(!demuxer_);
   DCHECK(read_cb_.is_null());
   DCHECK(buffer_queue_.IsEmpty());
 }
 
 base::TimeDelta FFmpegDemuxerStream::GetElapsedTime() const {
   return ConvertStreamTimestamp(stream_->time_base, stream_->cur_dts);
 }
 
 Ranges<base::TimeDelta> FFmpegDemuxerStream::GetBufferedRanges() const {
   return buffered_ranges_;
 }
 
 void FFmpegDemuxerStream::SatisfyPendingRead() {
-  DCHECK(message_loop_->BelongsToCurrentThread());
+  DCHECK(task_runner_->BelongsToCurrentThread());
   if (!read_cb_.is_null()) {
     if (!buffer_queue_.IsEmpty()) {
       base::ResetAndReturn(&read_cb_).Run(
           DemuxerStream::kOk, buffer_queue_.Pop());
     } else if (end_of_stream_) {
       base::ResetAndReturn(&read_cb_).Run(
           DemuxerStream::kOk, DecoderBuffer::CreateEOSBuffer());
     }
   }
 
@@ skipping 20 matching lines @@
   if (timestamp == static_cast<int64>(AV_NOPTS_VALUE))
     return kNoTimestamp();
 
   return ConvertFromTimeBase(time_base, timestamp);
 }
 
 //
 // FFmpegDemuxer
 //
 FFmpegDemuxer::FFmpegDemuxer(
-    const scoped_refptr<base::MessageLoopProxy>& message_loop,
+    const scoped_refptr<base::SingleThreadTaskRunner>& task_runner,
     DataSource* data_source,
     const NeedKeyCB& need_key_cb,
     const scoped_refptr<MediaLog>& media_log)
     : host_(NULL),
-      message_loop_(message_loop),
+      task_runner_(task_runner),
       weak_factory_(this),
       blocking_thread_("FFmpegDemuxer"),
       pending_read_(false),
       pending_seek_(false),
       data_source_(data_source),
       media_log_(media_log),
       bitrate_(0),
       start_time_(kNoTimestamp()),
       audio_disabled_(false),
       duration_known_(false),
-      url_protocol_(data_source, BindToLoop(message_loop_, base::Bind(
+      url_protocol_(data_source, BindToLoop(task_runner_, base::Bind(
           &FFmpegDemuxer::OnDataSourceError, base::Unretained(this)))),
       need_key_cb_(need_key_cb) {
-  DCHECK(message_loop_.get());
+  DCHECK(task_runner_.get());
   DCHECK(data_source_);
 }
 
 FFmpegDemuxer::~FFmpegDemuxer() {}
 
 void FFmpegDemuxer::Stop(const base::Closure& callback) {
-  DCHECK(message_loop_->BelongsToCurrentThread());
+  DCHECK(task_runner_->BelongsToCurrentThread());
   url_protocol_.Abort();
   data_source_->Stop(BindToCurrentLoop(base::Bind(
       &FFmpegDemuxer::OnDataSourceStopped, weak_this_,
       BindToCurrentLoop(callback))));
   data_source_ = NULL;
 }
 
 void FFmpegDemuxer::Seek(base::TimeDelta time, const PipelineStatusCB& cb) {
-  DCHECK(message_loop_->BelongsToCurrentThread());
+  DCHECK(task_runner_->BelongsToCurrentThread());
   CHECK(!pending_seek_);
 
   // TODO(scherkus): Inspect |pending_read_| and cancel IO via |blocking_url_|,
   // otherwise we can end up waiting for a pre-seek read to complete even though
   // we know we're going to drop it on the floor.
 
   // Always seek to a timestamp less than or equal to the desired timestamp.
   int flags = AVSEEK_FLAG_BACKWARD;
 
   // Passing -1 as our stream index lets FFmpeg pick a default stream.  FFmpeg
   // will attempt to use the lowest-index video stream, if present, followed by
   // the lowest-index audio stream.
   pending_seek_ = true;
   base::PostTaskAndReplyWithResult(
       blocking_thread_.message_loop_proxy().get(),
       FROM_HERE,
       base::Bind(&av_seek_frame,
                  glue_->format_context(),
                  -1,
                  time.InMicroseconds(),
                  flags),
       base::Bind(&FFmpegDemuxer::OnSeekFrameDone, weak_this_, cb));
 }
 
 void FFmpegDemuxer::OnAudioRendererDisabled() {
-  DCHECK(message_loop_->BelongsToCurrentThread());
+  DCHECK(task_runner_->BelongsToCurrentThread());
   audio_disabled_ = true;
   StreamVector::iterator iter;
   for (iter = streams_.begin(); iter != streams_.end(); ++iter) {
     if (*iter && (*iter)->type() == DemuxerStream::AUDIO) {
       (*iter)->Stop();
     }
   }
 }
 
 void FFmpegDemuxer::Initialize(DemuxerHost* host,
                                const PipelineStatusCB& status_cb) {
-  DCHECK(message_loop_->BelongsToCurrentThread());
+  DCHECK(task_runner_->BelongsToCurrentThread());
   host_ = host;
   weak_this_ = weak_factory_.GetWeakPtr();
 
   // TODO(scherkus): DataSource should have a host by this point,
   // see http://crbug.com/122071
   data_source_->set_host(host);
 
   glue_.reset(new FFmpegGlue(&url_protocol_));
   AVFormatContext* format_context = glue_->format_context();
 
   // Disable ID3v1 tag reading to avoid costly seeks to end of file for data we
   // don't use.  FFmpeg will only read ID3v1 tags if no other metadata is
   // available, so add a metadata entry to ensure some is always present.
   av_dict_set(&format_context->metadata, "skip_id3v1_tags", "", 0);
 
   // Open the AVFormatContext using our glue layer.
   CHECK(blocking_thread_.Start());
   base::PostTaskAndReplyWithResult(
       blocking_thread_.message_loop_proxy().get(),
       FROM_HERE,
       base::Bind(&FFmpegGlue::OpenContext, base::Unretained(glue_.get())),
       base::Bind(&FFmpegDemuxer::OnOpenContextDone, weak_this_, status_cb));
 }
 
 DemuxerStream* FFmpegDemuxer::GetStream(DemuxerStream::Type type) {
-  DCHECK(message_loop_->BelongsToCurrentThread());
+  DCHECK(task_runner_->BelongsToCurrentThread());
   return GetFFmpegStream(type);
 }
 
 FFmpegDemuxerStream* FFmpegDemuxer::GetFFmpegStream(
     DemuxerStream::Type type) const {
   StreamVector::const_iterator iter;
   for (iter = streams_.begin(); iter != streams_.end(); ++iter) {
     if (*iter && (*iter)->type() == type) {
       return *iter;
     }
   }
   return NULL;
 }
 
 base::TimeDelta FFmpegDemuxer::GetStartTime() const {
-  DCHECK(message_loop_->BelongsToCurrentThread());
+  DCHECK(task_runner_->BelongsToCurrentThread());
   return start_time_;
 }
 
 // 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,
@@ skipping 21 matching lines @@
 
   // Do math in floating point as we'd overflow an int64 if the filesize was
   // larger than ~1073GB.
   double bytes = filesize_in_bytes;
   double duration_us = duration.InMicroseconds();
   return bytes * 8000000.0 / duration_us;
 }
 
 void FFmpegDemuxer::OnOpenContextDone(const PipelineStatusCB& status_cb,
                                       bool result) {
-  DCHECK(message_loop_->BelongsToCurrentThread());
+  DCHECK(task_runner_->BelongsToCurrentThread());
   if (!blocking_thread_.IsRunning()) {
     status_cb.Run(PIPELINE_ERROR_ABORT);
     return;
   }
 
   if (!result) {
     status_cb.Run(DEMUXER_ERROR_COULD_NOT_OPEN);
     return;
   }
 
   // Fully initialize AVFormatContext by parsing the stream a little.
   base::PostTaskAndReplyWithResult(
       blocking_thread_.message_loop_proxy().get(),
       FROM_HERE,
       base::Bind(&avformat_find_stream_info,
                  glue_->format_context(),
                  static_cast<AVDictionary**>(NULL)),
       base::Bind(&FFmpegDemuxer::OnFindStreamInfoDone, weak_this_, status_cb));
 }
 
 void FFmpegDemuxer::OnFindStreamInfoDone(const PipelineStatusCB& status_cb,
                                          int result) {
-  DCHECK(message_loop_->BelongsToCurrentThread());
+  DCHECK(task_runner_->BelongsToCurrentThread());
   if (!blocking_thread_.IsRunning() || !data_source_) {
     status_cb.Run(PIPELINE_ERROR_ABORT);
     return;
   }
 
   if (result < 0) {
     status_cb.Run(DEMUXER_ERROR_COULD_NOT_PARSE);
     return;
   }
 
@@ skipping 40 matching lines @@
       // frame size and visible size are valid.
       AVStreamToVideoDecoderConfig(stream, &video_config, false);
 
       if (!video_config.IsValidConfig())
         continue;
       video_stream = stream;
     } else {
       continue;
     }
 
-    streams_[i] = new FFmpegDemuxerStream(this, stream);
+    streams_[i] = new FFmpegDemuxerStream(this, task_runner_, stream);
     max_duration = std::max(max_duration, streams_[i]->duration());
 
     if (stream->first_dts != static_cast<int64_t>(AV_NOPTS_VALUE)) {
       const base::TimeDelta first_dts = ConvertFromTimeBase(
           stream->time_base, stream->first_dts);
       if (start_time_ == kNoTimestamp() || first_dts < start_time_)
         start_time_ = first_dts;
     }
   }
 
@@ skipping 89 matching lines @@
 
 
   media_log_->SetDoubleProperty("max_duration", max_duration.InSecondsF());
   media_log_->SetDoubleProperty("start_time", start_time_.InSecondsF());
   media_log_->SetIntegerProperty("bitrate", bitrate_);
 
   status_cb.Run(PIPELINE_OK);
 }
 
 void FFmpegDemuxer::OnSeekFrameDone(const PipelineStatusCB& cb, int result) {
-  DCHECK(message_loop_->BelongsToCurrentThread());
+  DCHECK(task_runner_->BelongsToCurrentThread());
   CHECK(pending_seek_);
   pending_seek_ = false;
 
   if (!blocking_thread_.IsRunning()) {
     cb.Run(PIPELINE_ERROR_ABORT);
     return;
   }
 
   if (result < 0) {
     // Use VLOG(1) instead of NOTIMPLEMENTED() to prevent the message being
@@ skipping 10 matching lines @@
   }
 
   // Resume reading until capacity.
   ReadFrameIfNeeded();
 
   // Notify we're finished seeking.
   cb.Run(PIPELINE_OK);
 }
 
 void FFmpegDemuxer::ReadFrameIfNeeded() {
-  DCHECK(message_loop_->BelongsToCurrentThread());
+  DCHECK(task_runner_->BelongsToCurrentThread());
 
   // Make sure we have work to do before reading.
   if (!blocking_thread_.IsRunning() || !StreamsHaveAvailableCapacity() ||
       pending_read_ || pending_seek_) {
     return;
   }
 
   // Allocate and read an AVPacket from the media. Save |packet_ptr| since
   // evaluation order of packet.get() and base::Passed(&packet) is
   // undefined.
   ScopedAVPacket packet(new AVPacket());
   AVPacket* packet_ptr = packet.get();
 
   pending_read_ = true;
   base::PostTaskAndReplyWithResult(
       blocking_thread_.message_loop_proxy().get(),
       FROM_HERE,
       base::Bind(&av_read_frame, glue_->format_context(), packet_ptr),
       base::Bind(
           &FFmpegDemuxer::OnReadFrameDone, weak_this_, base::Passed(&packet)));
 }
 
 void FFmpegDemuxer::OnReadFrameDone(ScopedAVPacket packet, int result) {
-  DCHECK(message_loop_->BelongsToCurrentThread());
+  DCHECK(task_runner_->BelongsToCurrentThread());
   DCHECK(pending_read_);
   pending_read_ = false;
 
   if (!blocking_thread_.IsRunning() || pending_seek_) {
     return;
   }
 
   if (result < 0) {
     // Update the duration based on the audio stream if
     // it was previously unknown http://crbug.com/86830
@@ skipping 54 matching lines @@
 
   // Keep reading until we've reached capacity.
   ReadFrameIfNeeded();
 }
 
 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());
+  DCHECK(task_runner_->BelongsToCurrentThread());
   blocking_thread_.Stop();
 
   StreamVector::iterator iter;
   for (iter = streams_.begin(); iter != streams_.end(); ++iter) {
     if (*iter)
       (*iter)->Stop();
   }
 
   callback.Run();
 }
 
 bool FFmpegDemuxer::StreamsHaveAvailableCapacity() {
-  DCHECK(message_loop_->BelongsToCurrentThread());
+  DCHECK(task_runner_->BelongsToCurrentThread());
   StreamVector::iterator iter;
   for (iter = streams_.begin(); iter != streams_.end(); ++iter) {
     if (*iter && (*iter)->HasAvailableCapacity()) {
       return true;
     }
   }
   return false;
 }
 
 void FFmpegDemuxer::StreamHasEnded() {
-  DCHECK(message_loop_->BelongsToCurrentThread());
+  DCHECK(task_runner_->BelongsToCurrentThread());
   StreamVector::iterator iter;
   for (iter = streams_.begin(); iter != streams_.end(); ++iter) {
     if (!*iter ||
         (audio_disabled_ && (*iter)->type() == DemuxerStream::AUDIO)) {
       continue;
     }
     (*iter)->SetEndOfStream();
   }
 }
 
 void FFmpegDemuxer::FireNeedKey(const std::string& init_data_type,
                                 const std::string& encryption_key_id) {
   std::vector<uint8> key_id_local(encryption_key_id.begin(),
                                   encryption_key_id.end());
   need_key_cb_.Run(init_data_type, key_id_local);
 }
 
 void FFmpegDemuxer::NotifyCapacityAvailable() {
-  DCHECK(message_loop_->BelongsToCurrentThread());
+  DCHECK(task_runner_->BelongsToCurrentThread());
   ReadFrameIfNeeded();
 }
 
 void FFmpegDemuxer::NotifyBufferingChanged() {
-  DCHECK(message_loop_->BelongsToCurrentThread());
+  DCHECK(task_runner_->BelongsToCurrentThread());
   Ranges<base::TimeDelta> buffered;
   FFmpegDemuxerStream* audio =
       audio_disabled_ ? NULL : GetFFmpegStream(DemuxerStream::AUDIO);
   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