Replace trampolining in FFmpegDemuxer with explicit thread calling convention checks.

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 10 matching lines @@
 #include "media/base/decoder_buffer.h"
 #include "media/base/limits.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"
 
 namespace media {
 
+// Helpers for making ReadCBs always run on a new execution stack.
+static void RunReadCB(const scoped_refptr<base::MessageLoopProxy>& message_loop,
+                      const DemuxerStream::ReadCB& read_cb,
+                      DemuxerStream::Status status,
+                      const scoped_refptr<DecoderBuffer>& buffer) {
+  message_loop->PostTask(FROM_HERE, base::Bind(read_cb, status, buffer));
+}
+
+static DemuxerStream::ReadCB WrapReadCB(const DemuxerStream::ReadCB& read_cb) {
+  return base::Bind(&RunReadCB, base::MessageLoopProxy::current(), read_cb);
+}
+
 //
 // FFmpegDemuxerStream
 //
 FFmpegDemuxerStream::FFmpegDemuxerStream(
     FFmpegDemuxer* demuxer,
     AVStream* stream)
     : demuxer_(demuxer),
       message_loop_(base::MessageLoopProxy::current()),
       stream_(stream),
       type_(UNKNOWN),
@@ skipping 86 matching lines @@
   read_queue_.clear();
   stopped_ = true;
   end_of_stream_ = true;
 }
 
 base::TimeDelta FFmpegDemuxerStream::duration() {
   return duration_;
 }
 
 DemuxerStream::Type FFmpegDemuxerStream::type() {
+  DCHECK(message_loop_->BelongsToCurrentThread());
   return type_;
 }
 
 void FFmpegDemuxerStream::Read(const ReadCB& read_cb) {
-  if (!message_loop_->BelongsToCurrentThread()) {
-    message_loop_->PostTask(FROM_HERE, base::Bind(
-        &FFmpegDemuxerStream::Read, this, read_cb));
-    return;
-  }
+  DCHECK(message_loop_->BelongsToCurrentThread());
+  ReadCB wrapped_read_cb = WrapReadCB(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 (stopped_) {
-    read_cb.Run(DemuxerStream::kOk, DecoderBuffer::CreateEOSBuffer());
+    wrapped_read_cb.Run(DemuxerStream::kOk, DecoderBuffer::CreateEOSBuffer());
     return;
   }
 
-  read_queue_.push_back(read_cb);
+  read_queue_.push_back(wrapped_read_cb);

[acolwell GONE FROM CHROMIUM, 2012/12/03 23:14:21]

Do we still need this queue?

[scherkus (not reviewing), 2012/12/07 22:18:55]

We don't! Removed.

   SatisfyPendingReads();
 }
 
 void FFmpegDemuxerStream::EnableBitstreamConverter() {
-  if (!message_loop_->BelongsToCurrentThread()) {
-    message_loop_->PostTask(FROM_HERE, base::Bind(
-        &FFmpegDemuxerStream::EnableBitstreamConverter, this));
-    return;
-  }
+  DCHECK(message_loop_->BelongsToCurrentThread());
   CHECK(bitstream_converter_.get());
   bitstream_converter_enabled_ = true;
 }
 
 const AudioDecoderConfig& FFmpegDemuxerStream::audio_decoder_config() {
+  DCHECK(message_loop_->BelongsToCurrentThread());
   CHECK_EQ(type_, AUDIO);
   return audio_config_;
 }
 
 const VideoDecoderConfig& FFmpegDemuxerStream::video_decoder_config() {
+  DCHECK(message_loop_->BelongsToCurrentThread());
   CHECK_EQ(type_, VIDEO);
   return video_config_;
 }
 
 FFmpegDemuxerStream::~FFmpegDemuxerStream() {
   DCHECK(stopped_);
   DCHECK(read_queue_.empty());
   DCHECK(buffer_queue_.IsEmpty());
 }
 
@@ skipping 14 matching lines @@
       buffer = buffer_queue_.Pop();
     } else if (end_of_stream_) {
       buffer = DecoderBuffer::CreateEOSBuffer();
     } else {
       break;
     }
 
     // Send buffer back on a new execution stack to avoid recursing.
     ReadCB read_cb = read_queue_.front();
     read_queue_.pop_front();
-    message_loop_->PostTask(FROM_HERE, base::Bind(
-        read_cb, DemuxerStream::kOk, buffer));
+    read_cb.Run(DemuxerStream::kOk, buffer);
   }
 
   // Have capacity? Ask for more!
   if (HasAvailableCapacity() && !end_of_stream_) {
     demuxer_->NotifyCapacityAvailable();
   }
 }
 
 bool FFmpegDemuxerStream::HasAvailableCapacity() {
   // Try to have one second's worth of encoded data per stream.
@@ skipping 28 matching lines @@
       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::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));
+  DCHECK(message_loop_->BelongsToCurrentThread());
+  url_protocol_.Abort();
+  data_source_->Stop(BindToLoop(message_loop_, base::Bind(
+      &FFmpegDemuxer::OnDataSourceStopped, this, callback)));
 }
 
 void FFmpegDemuxer::Seek(base::TimeDelta time, const PipelineStatusCB& cb) {
-  message_loop_->PostTask(FROM_HERE,
-                          base::Bind(&FFmpegDemuxer::SeekTask, this, time, cb));
+  DCHECK(message_loop_->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(), FROM_HERE,
+      base::Bind(&av_seek_frame, glue_->format_context(), -1,
+                 time.InMicroseconds(), flags),
+      base::Bind(&FFmpegDemuxer::OnSeekFrameDone, this, cb));
 }
 
 void FFmpegDemuxer::SetPlaybackRate(float playback_rate) {
-  DCHECK(data_source_.get());
+  DCHECK(message_loop_->BelongsToCurrentThread());
   data_source_->SetPlaybackRate(playback_rate);
 }
 
 void FFmpegDemuxer::OnAudioRendererDisabled() {
-  message_loop_->PostTask(FROM_HERE, base::Bind(
-      &FFmpegDemuxer::DisableAudioStreamTask, this));
+  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();
+    }
+  }
 }
 
 void FFmpegDemuxer::Initialize(DemuxerHost* host,
                                const PipelineStatusCB& status_cb) {
-  message_loop_->PostTask(FROM_HERE, base::Bind(
-      &FFmpegDemuxer::InitializeTask, this, host, status_cb));
+  DCHECK(message_loop_->BelongsToCurrentThread());
+  host_ = host;
+
+  // 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(), FROM_HERE,
+      base::Bind(&FFmpegGlue::OpenContext, base::Unretained(glue_.get())),
+      base::Bind(&FFmpegDemuxer::OnOpenContextDone, this, status_cb));
 }
 
 scoped_refptr<DemuxerStream> FFmpegDemuxer::GetStream(
     DemuxerStream::Type type) {
+  DCHECK(message_loop_->BelongsToCurrentThread());
   return GetFFmpegStream(type);
 }
 
 scoped_refptr<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());
   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 19 matching lines @@
     return 0;
   }
 
   // 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::InitializeTask(DemuxerHost* host,
-                                   const PipelineStatusCB& status_cb) {
-  DCHECK(message_loop_->BelongsToCurrentThread());
-  host_ = host;
-
-  // 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(), FROM_HERE,
-      base::Bind(&FFmpegGlue::OpenContext, base::Unretained(glue_.get())),
-      base::Bind(&FFmpegDemuxer::OnOpenContextDone, this, status_cb));
-}
-
 void FFmpegDemuxer::OnOpenContextDone(const PipelineStatusCB& status_cb,
                                       bool result) {
   DCHECK(message_loop_->BelongsToCurrentThread());
   if (!blocking_thread_.IsRunning()) {
     status_cb.Run(PIPELINE_ERROR_ABORT);
     return;
   }
 
   if (!result) {
     status_cb.Run(DEMUXER_ERROR_COULD_NOT_OPEN);
@@ skipping 94 matching lines @@
 
   int64 filesize_in_bytes = 0;
   url_protocol_.GetSize(&filesize_in_bytes);
   bitrate_ = CalculateBitrate(format_context, max_duration, filesize_in_bytes);
   if (bitrate_ > 0)
     data_source_->SetBitrate(bitrate_);
 
   status_cb.Run(PIPELINE_OK);
 }
 
-void FFmpegDemuxer::SeekTask(base::TimeDelta time, const PipelineStatusCB& cb) {
-  DCHECK(message_loop_->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(), FROM_HERE,
-      base::Bind(&av_seek_frame, glue_->format_context(), -1,
-                 time.InMicroseconds(), flags),
-      base::Bind(&FFmpegDemuxer::OnSeekFrameDone, this, cb));
-}
-
 void FFmpegDemuxer::OnSeekFrameDone(const PipelineStatusCB& cb, int result) {
   DCHECK(message_loop_->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
     // captured from stdout and contaminates testing.
     // TODO(scherkus): Implement this properly and signal error (BUG=23447).
     VLOG(1) << "Not implemented";
   }
 
   // Tell streams to flush buffers due to seeking.
   StreamVector::iterator iter;
   for (iter = streams_.begin(); iter != streams_.end(); ++iter) {
     if (*iter)
       (*iter)->FlushBuffers();
   }
 
-  // Resume demuxing until capacity.
-  DemuxTask();
+  // Resume reading until capacity.
+  ReadFrameIfNeeded();
 
   // Notify we're finished seeking.
   cb.Run(PIPELINE_OK);
 }
 
-void FFmpegDemuxer::DemuxTask() {
+void FFmpegDemuxer::ReadFrameIfNeeded() {
   DCHECK(message_loop_->BelongsToCurrentThread());
 
-  // Make sure we have work to do before demuxing.
+  // 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();
@@ skipping 45 matching lines @@
   // 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());
   }
 
-  // Keep demuxing until we've reached capacity.
-  DemuxTask();
-}
-
-void FFmpegDemuxer::StopTask(const base::Closure& callback) {
-  DCHECK(message_loop_->BelongsToCurrentThread());
-  url_protocol_.Abort();
-  data_source_->Stop(BindToLoop(message_loop_, base::Bind(
-      &FFmpegDemuxer::OnDataSourceStopped, this, callback)));
+  // 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());
   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();
-    }
-  }
-}
-
 bool FFmpegDemuxer::StreamsHaveAvailableCapacity() {
   DCHECK(message_loop_->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());
   StreamVector::iterator iter;
   for (iter = streams_.begin(); iter != streams_.end(); ++iter) {
     if (!*iter ||
         (audio_disabled_ && (*iter)->type() == DemuxerStream::AUDIO)) {
       continue;
     }
     (*iter)->SetEndOfStream();
   }
 }
 
 void FFmpegDemuxer::NotifyCapacityAvailable() {
   DCHECK(message_loop_->BelongsToCurrentThread());
-  DemuxTask();
+  ReadFrameIfNeeded();
 }
 
 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