Add support for cancellation of demuxer reads.

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 <utility>
 
@@ skipping 235 matching lines @@
     }
 
     MEDIA_LOG(INFO, media_log) << "FFmpegDemuxer: created video stream, config "
                                << video_config->AsHumanReadableString();
   }
 
   return base::WrapUnique(new FFmpegDemuxerStream(
       demuxer, stream, std::move(audio_config), std::move(video_config)));
 }
 
+static void UnmarkEndOfStream(AVFormatContext* format_context) {
+  format_context->pb->eof_reached = 0;
+}
+
 //
 // FFmpegDemuxerStream
 //
 FFmpegDemuxerStream::FFmpegDemuxerStream(
     FFmpegDemuxer* demuxer,
     AVStream* stream,
     std::unique_ptr<AudioDecoderConfig> audio_config,
     std::unique_ptr<VideoDecoderConfig> video_config)
     : demuxer_(demuxer),
       task_runner_(base::ThreadTaskRunnerHandle::Get()),
@@ skipping 320 matching lines @@
   // Reset bitstream for converter to do so.
   // This is related to chromium issue 140371 (http://crbug.com/140371).
   ResetBitstreamConverter();
 
   buffer_queue_.Clear();
   end_of_stream_ = false;
   last_packet_timestamp_ = kNoTimestamp;
   last_packet_duration_ = kNoTimestamp;
 }
 
+void FFmpegDemuxerStream::Abort() {
+  if (!read_cb_.is_null())
+    base::ResetAndReturn(&read_cb_).Run(DemuxerStream::kAborted, nullptr);
+}
+
 void FFmpegDemuxerStream::Stop() {
   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;
@@ skipping 208 matching lines @@
 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"),
       pending_read_(false),
-      pending_seek_(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),
       weak_factory_(this) {
   DCHECK(task_runner_.get());
@@ skipping 38 matching lines @@
 
   // Open the AVFormatContext using our glue layer.
   CHECK(blocking_thread_.Start());
   base::PostTaskAndReplyWithResult(
       blocking_thread_.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());
+
+  // This should only be called after the demuxer has been initialized.
+  DCHECK(blocking_thread_.IsRunning());
+  DCHECK_GT(streams_.size(), 0u);
+
+  // Abort all outstanding reads.
+  for (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(
+      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
@@ skipping 10 matching lines @@
 
   data_source_ = NULL;
 
   // Invalidate WeakPtrs on |task_runner_|, destruction may happen on another
   // thread.
   weak_factory_.InvalidateWeakPtrs();
 }
 
 void FFmpegDemuxer::StartWaitingForSeek(base::TimeDelta seek_time) {}
 
-void FFmpegDemuxer::CancelPendingSeek(base::TimeDelta seek_time) {}
+void FFmpegDemuxer::CancelPendingSeek(base::TimeDelta seek_time) {
+  if (task_runner_->BelongsToCurrentThread()) {
+    AbortPendingReads();
+  } else {
+    task_runner_->PostTask(FROM_HERE,
+                           base::Bind(&FFmpegDemuxer::AbortPendingReads,
+                                      weak_factory_.GetWeakPtr()));
+  }
+}
 
 void FFmpegDemuxer::Seek(base::TimeDelta time, const PipelineStatusCB& cb) {
   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.
+  CHECK(pending_seek_cb_.is_null());
 
   // FFmpeg requires seeks to be adjusted according to the lowest starting time.
   // Since EnqueuePacket() rebased negative timestamps by the start time, we
   // must correct the shift here.
   //
   // Additionally, to workaround limitations in how we expose seekable ranges to
   // Blink (http://crbug.com/137275), we also want to clamp seeks before the
   // start time to the start time.
   base::TimeDelta seek_time = start_time_ < base::TimeDelta()
                                   ? time + start_time_
@@ skipping 16 matching lines @@
   // given container will demux all packets after the seek point.  Instead it
   // only guarantees that all packets after the file position of the seek will
   // 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_ = true;
+  pending_seek_cb_ = cb;
   base::PostTaskAndReplyWithResult(
-      blocking_thread_.task_runner().get(),
-      FROM_HERE,
-      base::Bind(&av_seek_frame,
-                 glue_->format_context(),
-                 seeking_stream->index,
+      blocking_thread_.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(), cb));
+      base::Bind(&FFmpegDemuxer::OnSeekFrameDone, weak_factory_.GetWeakPtr()));
 }
 
 base::Time FFmpegDemuxer::GetTimelineOffset() const {
   return timeline_offset_;
 }
 
 DemuxerStream* FFmpegDemuxer::GetStream(DemuxerStream::Type type) {
   DCHECK(task_runner_->BelongsToCurrentThread());
   return GetFFmpegStream(type);
 }
@@ skipping 519 matching lines @@
       lowest_start_time_stream->start_time() <= seek_time) {
     return lowest_start_time_stream;
   }
 
   // If we couldn't find any streams with the start time lower than |seek_time|
   // then use either video (if one exists) or any audio stream.
   return video_stream ? video_stream : static_cast<FFmpegDemuxerStream*>(
                                            GetStream(DemuxerStream::AUDIO));
 }
 
-void FFmpegDemuxer::OnSeekFrameDone(const PipelineStatusCB& cb, int result) {
+void FFmpegDemuxer::OnSeekFrameDone(int result) {
   DCHECK(task_runner_->BelongsToCurrentThread());
-  CHECK(pending_seek_);
-  pending_seek_ = false;
+  CHECK(!pending_seek_cb_.is_null());
 
   if (!blocking_thread_.IsRunning()) {
     MEDIA_LOG(ERROR, media_log_) << GetDisplayName() << ": bad state";
-    cb.Run(PIPELINE_ERROR_ABORT);
+    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";
   }
 
   // Tell streams to flush buffers due to seeking.
   StreamVector::iterator iter;
   for (iter = streams_.begin(); iter != streams_.end(); ++iter) {
     if (*iter)
       (*iter)->FlushBuffers();
   }
 
   // Resume reading until capacity.
   ReadFrameIfNeeded();
 
   // Notify we're finished seeking.
-  cb.Run(PIPELINE_OK);
+  base::ResetAndReturn(&pending_seek_cb_).Run(PIPELINE_OK);
 }
 
 void FFmpegDemuxer::OnEnabledAudioTracksChanged(
     const std::vector<MediaTrack::Id>& track_ids,
     base::TimeDelta currTime) {
   DCHECK(task_runner_->BelongsToCurrentThread());
   bool enabled = false;
   DemuxerStream* audio_stream = GetStream(DemuxerStream::AUDIO);
   CHECK(audio_stream);
   if (track_ids.size() > 0) {
@@ skipping 19 matching lines @@
   DVLOG(1) << __func__ << ": " << (enabled ? "enabling" : "disabling")
            << " video stream";
   video_stream->set_enabled(enabled, currTime);
 }
 
 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_) {
+      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,
       base::Bind(&av_read_frame, glue_->format_context(), packet_ptr),
       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_) {
+  if (!blocking_thread_.IsRunning() || !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()) {
-    LOG(ERROR) << __func__ << " result=" << result
-               << " IsMaxMemoryUsageReached=" << IsMaxMemoryUsageReached();
+    DVLOG(1) << __func__ << " result=" << result
+             << " IsMaxMemoryUsageReached=" << IsMaxMemoryUsageReached();
     // Update the duration based on the highest elapsed time across all streams
     // if it was previously unknown.
     if (!duration_known_) {
       base::TimeDelta max_duration;
 
       for (StreamVector::iterator iter = streams_.begin();
            iter != streams_.end();
            ++iter) {
         if (!*iter)
           continue;
@@ skipping 90 matching lines @@
 
 void FFmpegDemuxer::SetLiveness(DemuxerStream::Liveness liveness) {
   DCHECK(task_runner_->BelongsToCurrentThread());
   for (auto* stream : streams_) {
     if (stream)
       stream->SetLiveness(liveness);
   }
 }
 
 }  // namespace media