Replace FFmpegDemuxer thread per element with base::TaskScheduler.

media/filters/ffmpeg_demuxer.cc

Index: media/filters/ffmpeg_demuxer.cc
diff --git a/media/filters/ffmpeg_demuxer.cc b/media/filters/ffmpeg_demuxer.cc
index d7146034bbf45e52c1872f8fb113d40bf4366d10..2e923d2b8dfe35f77429254b3ba262cd17fc862d 100644
--- a/media/filters/ffmpeg_demuxer.cc
+++ b/media/filters/ffmpeg_demuxer.cc
@@ -22,6 +22,7 @@
 #include "base/strings/stringprintf.h"
 #include "base/sys_byteorder.h"
 #include "base/task_runner_util.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"
@@ -47,6 +48,20 @@
 
 namespace media {
 
+static scoped_refptr<base::SequencedTaskRunner> GetSequencedTaskRunner() {
+  // We cap the worker pool to 6 threads since there are at most 6 connections
+  // allowed per-origin.
+  // TODO(dalecurtis): This may be a bit restrictive if there are many videos
+  // from different origins on a page; is this common in practice?
+  // TODO(dalecurtis): Instead of being leaky, should we have a PoolManager type
+  // object which drops the pool when the last FFmpegDemuxer goes away?
+  static scoped_refptr<base::SequencedWorkerPool> pool(
+      new base::SequencedWorkerPool(6, "FFmpegDemuxer",
+                                    base::TaskPriority::USER_VISIBLE));
+  return pool->GetSequencedTaskRunner(
+      base::SequencedWorkerPool::GetSequenceToken());
+}
+
 static base::Time ExtractTimelineOffset(AVFormatContext* format_context) {
   if (strstr(format_context->iformat->name, "webm") ||
       strstr(format_context->iformat->name, "matroska")) {
@@ -845,7 +860,8 @@ FFmpegDemuxer::FFmpegDemuxer(
     const scoped_refptr<MediaLog>& media_log)
     : host_(NULL),
       task_runner_(task_runner),
-      blocking_thread_("FFmpegDemuxer"),
+      blocking_task_runner_(GetSequencedTaskRunner()),
+      stopped_(false),
       pending_read_(false),
       data_source_(data_source),
       media_log_(media_log),
@@ -866,6 +882,12 @@ 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 {
@@ -899,9 +921,8 @@ void FFmpegDemuxer::Initialize(DemuxerHost* host,
   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));
@@ -911,7 +932,7 @@ 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.
@@ -929,7 +950,7 @@ void FFmpegDemuxer::AbortPendingReads() {
   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;
 
@@ -949,12 +970,6 @@ void FFmpegDemuxer::Stop() {
   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();
@@ -963,7 +978,9 @@ void FFmpegDemuxer::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();
 }
@@ -1022,7 +1039,7 @@ void FFmpegDemuxer::Seek(base::TimeDelta time, const PipelineStatusCB& cb) {
 
   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.
@@ -1159,7 +1176,7 @@ static int CalculateBitrate(AVFormatContext* format_context,
 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;
@@ -1173,20 +1190,17 @@ void FFmpegDemuxer::OnOpenContextDone(const PipelineStatusCB& status_cb,
 
   // 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;
@@ -1606,7 +1620,7 @@ 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;
@@ -1693,8 +1707,8 @@ 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;
   }
 
@@ -1706,11 +1720,9 @@ void FFmpegDemuxer::ReadFrameIfNeeded() {
 
   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)));
 }
 
@@ -1719,7 +1731,7 @@ void FFmpegDemuxer::OnReadFrameDone(ScopedAVPacket packet, int result) {
   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: