Replace FFmpegDemuxer thread per element with base::TaskScheduler.

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 <memory>
 #include <set>
 #include <utility>
 
 #include "base/base64.h"
 #include "base/bind.h"
 #include "base/callback_helpers.h"
 #include "base/macros.h"
 #include "base/memory/ptr_util.h"
 #include "base/metrics/histogram_macros.h"
 #include "base/metrics/sparse_histogram.h"
 #include "base/single_thread_task_runner.h"
 #include "base/strings/string_number_conversions.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/task_scheduler/post_task.h"
+#include "base/threading/sequenced_worker_pool.h"
 #include "base/threading/thread_task_runner_handle.h"
 #include "base/time/time.h"
 #include "media/audio/sample_rates.h"
 #include "media/base/bind_to_current_loop.h"
 #include "media/base/decrypt_config.h"
 #include "media/base/limits.h"
 #include "media/base/media_log.h"
 #include "media/base/media_tracks.h"
 #include "media/base/timestamp_constants.h"
 #include "media/base/video_codecs.h"
@@ skipping 803 matching lines @@
 // FFmpegDemuxer
 //
 FFmpegDemuxer::FFmpegDemuxer(
     const scoped_refptr<base::SingleThreadTaskRunner>& task_runner,
     DataSource* data_source,
     const EncryptedMediaInitDataCB& encrypted_media_init_data_cb,
     const MediaTracksUpdatedCB& media_tracks_updated_cb,
     const scoped_refptr<MediaLog>& media_log)
     : host_(NULL),
       task_runner_(task_runner),
-      blocking_thread_("FFmpegDemuxer"),
+      // FFmpeg has no asynchronous API, so we use base::WaitableEvents inside
+      // the BlockingUrlProtocol to handle hops to the render thread for network
+      // reads and seeks.
+      blocking_task_runner_(base::CreateSequencedTaskRunnerWithTraits(
+          base::TaskTraits().MayBlock().WithBaseSyncPrimitives().WithPriority(
+              base::TaskPriority::USER_BLOCKING))),
+      stopped_(false),
       pending_read_(false),
       data_source_(data_source),
       media_log_(media_log),
       bitrate_(0),
       start_time_(kNoTimestamp),
       text_enabled_(false),
       duration_known_(false),
       encrypted_media_init_data_cb_(encrypted_media_init_data_cb),
       media_tracks_updated_cb_(media_tracks_updated_cb),
       cancel_pending_seek_factory_(this),
       weak_factory_(this) {
   DCHECK(task_runner_.get());
   DCHECK(data_source_);
   DCHECK(!media_tracks_updated_cb_.is_null());
 }
 
 FFmpegDemuxer::~FFmpegDemuxer() {
   // NOTE: This class is not destroyed on |task_runner|, so we must ensure that
   // there are no outstanding WeakPtrs by the time we reach here.
   DCHECK(!weak_factory_.HasWeakPtrs());
+
+  // There may be outstanding tasks in the blocking pool which are trying to use
+  // these members, so release them in sequence with any outstanding calls. The
+  // earlier call to Abort() on |data_source_| prevents further access to it.
+  blocking_task_runner_->DeleteSoon(FROM_HERE, url_protocol_.release());
+  blocking_task_runner_->DeleteSoon(FROM_HERE, glue_.release());
 }
 
 std::string FFmpegDemuxer::GetDisplayName() const {
   return "FFmpegDemuxer";
 }
 
 void FFmpegDemuxer::Initialize(DemuxerHost* host,
                                const PipelineStatusCB& status_cb,
                                bool enable_text_tracks) {
   DCHECK(task_runner_->BelongsToCurrentThread());
   host_ = host;
   text_enabled_ = enable_text_tracks;
   weak_this_ = cancel_pending_seek_factory_.GetWeakPtr();
 
+  // Give a WeakPtr to BlockingUrlProtocol since we'll need to release it on the
+  // blocking thread pool.
   url_protocol_.reset(new BlockingUrlProtocol(
-      data_source_,
-      BindToCurrentLoop(base::Bind(&FFmpegDemuxer::OnDataSourceError,
-                                   base::Unretained(this)))));
+      data_source_, BindToCurrentLoop(base::Bind(
+                        &FFmpegDemuxer::OnDataSourceError, weak_this_))));

[xhwang, 2017/03/02 22:05:01]

We have mixed use of weak_this_ and weak_factory_.GetWeakPtr() in this file.
Shall we consolidate them?

[DaleCurtis, 2017/03/02 22:48:53]

weak_this_ is bound to a factory which should only be invalidated during Stop()
prior to destruction, while weak_factory_ is invalidated regularly during
AbortPendingReads(). This isn't documented very well and I agree some cleanup is
necessary here. I'll see what I can do in a followup CL.

[xhwang, 2017/03/02 23:07:13]

Ah, that's tricky and I didn't even notice it. A followup cleanup CL would be
nice. Thanks!

   glue_.reset(new FFmpegGlue(url_protocol_.get()));
   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);
 
   // Ensure ffmpeg doesn't give up too early while looking for stream params;
   // this does not increase the amount of data downloaded.  The default value
   // is 5 AV_TIME_BASE units (1 second each), which prevents some oddly muxed
   // streams from being detected properly; this value was chosen arbitrarily.
   format_context->max_analyze_duration = 60 * AV_TIME_BASE;
 
   // Open the AVFormatContext using our glue layer.
-  CHECK(blocking_thread_.Start());
   base::PostTaskAndReplyWithResult(
-      blocking_thread_.task_runner().get(), FROM_HERE,
+      blocking_task_runner_.get(), FROM_HERE,
       base::Bind(&FFmpegGlue::OpenContext, base::Unretained(glue_.get())),
       base::Bind(&FFmpegDemuxer::OnOpenContextDone, weak_factory_.GetWeakPtr(),
                  status_cb));
 }
 
 void FFmpegDemuxer::AbortPendingReads() {
   DCHECK(task_runner_->BelongsToCurrentThread());
 
   // If Stop() has been called, then drop this call.
-  if (!blocking_thread_.IsRunning())
+  if (stopped_)
     return;
 
   // This should only be called after the demuxer has been initialized.
   DCHECK_GT(streams_.size(), 0u);
 
   // Abort all outstanding reads.
   for (const auto& stream : streams_) {
     if (stream)
       stream->Abort();
   }
 
   // It's important to invalidate read/seek completion callbacks to avoid any
   // errors that occur because of the data source abort.
   weak_factory_.InvalidateWeakPtrs();
   data_source_->Abort();
 
   // Aborting the read may cause EOF to be marked, undo this.
-  blocking_thread_.task_runner()->PostTask(
+  blocking_task_runner_->PostTask(
       FROM_HERE, base::Bind(&UnmarkEndOfStream, glue_->format_context()));
   pending_read_ = false;
 
   // TODO(dalecurtis): We probably should report PIPELINE_ERROR_ABORT here
   // instead to avoid any preroll work that may be started upon return, but
   // currently the PipelineImpl does not know how to handle this.
   if (!pending_seek_cb_.is_null())
     base::ResetAndReturn(&pending_seek_cb_).Run(PIPELINE_OK);
 }
 
 void FFmpegDemuxer::Stop() {
   DCHECK(task_runner_->BelongsToCurrentThread());
 
   // The order of Stop() and Abort() is important here.  If Abort() is called
   // first, control may pass into FFmpeg where it can destruct buffers that are
   // in the process of being fulfilled by the DataSource.
   data_source_->Stop();
   url_protocol_->Abort();
 
-  // 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.
-  blocking_thread_.Stop();
-
   for (const auto& stream : streams_) {
     if (stream)
       stream->Stop();
   }
 
   data_source_ = NULL;
 
   // Invalidate WeakPtrs on |task_runner_|, destruction may happen on another
-  // thread.
+  // thread. We don't need to wait for any outstanding tasks since they will all
+  // fail to return after invalidating WeakPtrs.
+  stopped_ = true;
   weak_factory_.InvalidateWeakPtrs();
   cancel_pending_seek_factory_.InvalidateWeakPtrs();
 }
 
 void FFmpegDemuxer::StartWaitingForSeek(base::TimeDelta seek_time) {}
 
 void FFmpegDemuxer::CancelPendingSeek(base::TimeDelta seek_time) {
   if (task_runner_->BelongsToCurrentThread()) {
     AbortPendingReads();
   } else {
@@ skipping 38 matching lines @@
   // be demuxed.  It's an open question whether FFmpeg should fix this:
   // http://lists.ffmpeg.org/pipermail/ffmpeg-devel/2014-June/159212.html
   // Tracked by http://crbug.com/387996.
   FFmpegDemuxerStream* demux_stream = FindPreferredStreamForSeeking(seek_time);
   DCHECK(demux_stream);
   const AVStream* seeking_stream = demux_stream->av_stream();
   DCHECK(seeking_stream);
 
   pending_seek_cb_ = cb;
   base::PostTaskAndReplyWithResult(
-      blocking_thread_.task_runner().get(), FROM_HERE,
+      blocking_task_runner_.get(), FROM_HERE,
       base::Bind(&av_seek_frame, glue_->format_context(), seeking_stream->index,
                  ConvertToTimeBase(seeking_stream->time_base, seek_time),
                  // Always seek to a timestamp <= to the desired timestamp.
                  AVSEEK_FLAG_BACKWARD),
       base::Bind(&FFmpegDemuxer::OnSeekFrameDone, weak_factory_.GetWeakPtr()));
 }
 
 base::Time FFmpegDemuxer::GetTimelineOffset() const {
   return timeline_offset_;
 }
@@ skipping 124 matching lines @@
   // Do math in floating point as we'd overflow an int64_t 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(task_runner_->BelongsToCurrentThread());
-  if (!blocking_thread_.IsRunning()) {
+  if (stopped_) {
     MEDIA_LOG(ERROR, media_log_) << GetDisplayName() << ": bad state";
     status_cb.Run(PIPELINE_ERROR_ABORT);
     return;
   }
 
   if (!result) {
     MEDIA_LOG(ERROR, media_log_) << GetDisplayName() << ": open context failed";
     status_cb.Run(DEMUXER_ERROR_COULD_NOT_OPEN);
     return;
   }
 
   // Fully initialize AVFormatContext by parsing the stream a little.
   base::PostTaskAndReplyWithResult(
-      blocking_thread_.task_runner().get(),
-      FROM_HERE,
-      base::Bind(&avformat_find_stream_info,
-                 glue_->format_context(),
+      blocking_task_runner_.get(), FROM_HERE,
+      base::Bind(&avformat_find_stream_info, glue_->format_context(),
                  static_cast<AVDictionary**>(NULL)),
       base::Bind(&FFmpegDemuxer::OnFindStreamInfoDone,
-                 weak_factory_.GetWeakPtr(),
-                 status_cb));
+                 weak_factory_.GetWeakPtr(), status_cb));
 }
 
 void FFmpegDemuxer::OnFindStreamInfoDone(const PipelineStatusCB& status_cb,
                                          int result) {
   DCHECK(task_runner_->BelongsToCurrentThread());
-  if (!blocking_thread_.IsRunning() || !data_source_) {
+  if (stopped_ || !data_source_) {
     MEDIA_LOG(ERROR, media_log_) << GetDisplayName() << ": bad state";
     status_cb.Run(PIPELINE_ERROR_ABORT);
     return;
   }
 
   if (result < 0) {
     MEDIA_LOG(ERROR, media_log_) << GetDisplayName()
                                  << ": find stream info failed";
     status_cb.Run(DEMUXER_ERROR_COULD_NOT_PARSE);
     return;
@@ skipping 399 matching lines @@
   }
 
   NOTREACHED();
   return nullptr;
 }
 
 void FFmpegDemuxer::OnSeekFrameDone(int result) {
   DCHECK(task_runner_->BelongsToCurrentThread());
   CHECK(!pending_seek_cb_.is_null());
 
-  if (!blocking_thread_.IsRunning()) {
+  if (stopped_) {
     MEDIA_LOG(ERROR, media_log_) << GetDisplayName() << ": bad state";
     base::ResetAndReturn(&pending_seek_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";
@@ skipping 65 matching lines @@
   if (selected_stream) {
     DVLOG(1) << __func__ << ": enabling stream " << selected_stream;
     selected_stream->set_enabled(true, curr_time);
   }
 }
 
 void FFmpegDemuxer::ReadFrameIfNeeded() {
   DCHECK(task_runner_->BelongsToCurrentThread());
 
   // Make sure we have work to do before reading.
-  if (!blocking_thread_.IsRunning() || !StreamsHaveAvailableCapacity() ||
-      pending_read_ || !pending_seek_cb_.is_null()) {
+  if (stopped_ || !StreamsHaveAvailableCapacity() || pending_read_ ||
+      !pending_seek_cb_.is_null()) {
     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_.task_runner().get(),
-      FROM_HERE,
+      blocking_task_runner_.get(), FROM_HERE,
       base::Bind(&av_read_frame, glue_->format_context(), packet_ptr),
-      base::Bind(&FFmpegDemuxer::OnReadFrameDone,
-                 weak_factory_.GetWeakPtr(),
+      base::Bind(&FFmpegDemuxer::OnReadFrameDone, weak_factory_.GetWeakPtr(),
                  base::Passed(&packet)));
 }
 
 void FFmpegDemuxer::OnReadFrameDone(ScopedAVPacket packet, int result) {
   DCHECK(task_runner_->BelongsToCurrentThread());
   DCHECK(pending_read_);
   pending_read_ = false;
 
-  if (!blocking_thread_.IsRunning() || !pending_seek_cb_.is_null())
+  if (stopped_ || !pending_seek_cb_.is_null())
     return;
 
   // Consider the stream as ended if:
   // - either underlying ffmpeg returned an error
   // - or FFMpegDemuxer reached the maximum allowed memory usage.
   if (result < 0 || IsMaxMemoryUsageReached()) {
     DVLOG(1) << __func__ << " result=" << result
              << " IsMaxMemoryUsageReached=" << IsMaxMemoryUsageReached();
     // Update the duration based on the highest elapsed time across all streams.
     base::TimeDelta max_duration;
@@ skipping 104 matching lines @@
 
 void FFmpegDemuxer::SetLiveness(DemuxerStream::Liveness liveness) {
   DCHECK(task_runner_->BelongsToCurrentThread());
   for (const auto& stream : streams_) {
     if (stream)
       stream->SetLiveness(liveness);
   }
 }
 
 }  // namespace media