FFmpegDemuxerStream::Read() now functions properly while stopped.

media/filters/ffmpeg_demuxer.cc

 // Copyright (c) 2009 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 "base/scoped_ptr.h"
+#include "base/stl_util-inl.h"
 #include "base/string_util.h"
 #include "base/time.h"
 #include "media/base/filter_host.h"
 #include "media/filters/ffmpeg_common.h"
 #include "media/filters/ffmpeg_demuxer.h"
 #include "media/filters/ffmpeg_glue.h"
 
+namespace {
+
+// Helper function to deep copy an AVPacket's data, size and timestamps.
+// Returns NULL if a packet could not be cloned (i.e., out of memory).
+AVPacket* ClonePacket(AVPacket* packet) {
+  scoped_ptr<AVPacket> clone(new AVPacket());
+  if (!clone.get() || av_new_packet(clone.get(), packet->size) < 0) {
+    return NULL;
+  }
+  DCHECK_EQ(clone->size, packet->size);
+  clone->dts = packet->dts;
+  clone->pts = packet->pts;
+  clone->duration = packet->duration;
+  memcpy(clone->data, packet->data, clone->size);
+  return clone.release();
+}
+
+}  // namespace
+
 namespace media {
 
 //
 // AVPacketBuffer
 //
 class AVPacketBuffer : public Buffer {
  public:
   AVPacketBuffer(AVPacket* packet, const base::TimeDelta& timestamp,
                  const base::TimeDelta& duration)
       : packet_(packet) {
@@ skipping 21 matching lines @@
 };
 
 
 //
 // FFmpegDemuxerStream
 //
 FFmpegDemuxerStream::FFmpegDemuxerStream(FFmpegDemuxer* demuxer,
                                          AVStream* stream)
     : demuxer_(demuxer),
       stream_(stream),
-      discontinuous_(false) {
+      discontinuous_(false),
+      stopped_(false) {
   DCHECK(demuxer_);
 
   // Determine our media format.
   switch (stream->codec->codec_type) {
     case CODEC_TYPE_AUDIO:
       media_format_.SetAsString(MediaFormat::kMimeType,
                                 mime_type::kFFmpegAudio);
       break;
     case CODEC_TYPE_VIDEO:
       media_format_.SetAsString(MediaFormat::kMimeType,
                                 mime_type::kFFmpegVideo);
       break;
     default:
       NOTREACHED();
       break;
   }
 
   // Calculate the duration.
   duration_ = ConvertTimestamp(stream->duration);
 }
 
 FFmpegDemuxerStream::~FFmpegDemuxerStream() {
-  // Since |buffer_queue_| uses scoped_refptr everything will get released.
-  while (!read_queue_.empty()) {
-    delete read_queue_.front();
-    read_queue_.pop_front();
-  }
+  AutoLock auto_lock(lock_);
+  DCHECK(stopped_);
+  DCHECK(read_queue_.empty());
+  DCHECK(buffer_queue_.empty());
 }
 
 void* FFmpegDemuxerStream::QueryInterface(const char* id) {
   DCHECK(id);
   AVStreamProvider* interface_ptr = NULL;
   if (0 == strcmp(id, AVStreamProvider::interface_id())) {
     interface_ptr = this;
   }
   return interface_ptr;
 }
 
 bool FFmpegDemuxerStream::HasPendingReads() {
   AutoLock auto_lock(lock_);
   return !read_queue_.empty();
 }
 
 base::TimeDelta FFmpegDemuxerStream::EnqueuePacket(AVPacket* packet) {
   base::TimeDelta timestamp = ConvertTimestamp(packet->pts);
   base::TimeDelta duration = ConvertTimestamp(packet->duration);
-  Buffer* buffer = new AVPacketBuffer(packet, timestamp, duration);
+  scoped_refptr<Buffer> buffer =
+      new AVPacketBuffer(packet, timestamp, duration);
   DCHECK(buffer);
   {
     AutoLock auto_lock(lock_);
+    if (stopped_) {
+      NOTREACHED() << "Attempted to enqueue packet on a stopped stream.";
+      return timestamp;
+    }
     buffer_queue_.push_back(buffer);
   }
   FulfillPendingReads();
   return timestamp;
 }
 
 void FFmpegDemuxerStream::FlushBuffers() {
   AutoLock auto_lock(lock_);
   buffer_queue_.clear();
   discontinuous_ = true;
 }
 
+void FFmpegDemuxerStream::Stop() {
+  // Since |buffer_queue_| can be very large, we swap queues and delete outside
+  // of the lock.
+  BufferQueue tmp_buffer_queue;
+  ReadQueue tmp_read_queue;
+  {
+    AutoLock auto_lock(lock_);
+    buffer_queue_.swap(tmp_buffer_queue);
+    read_queue_.swap(tmp_read_queue);
+    stopped_ = true;
+  }
+  tmp_buffer_queue.clear();
+  STLDeleteElements(&tmp_read_queue);
+}
+
 const MediaFormat& FFmpegDemuxerStream::media_format() {
   return media_format_;
 }
 
 void FFmpegDemuxerStream::Read(Callback1<Buffer*>::Type* read_callback) {
   DCHECK(read_callback);
   {
     AutoLock auto_lock(lock_);
+    // Don't accept any additional reads if we've been told to stop.
+    //
+    // TODO(scherkus): it would be cleaner if we replied with an error message.
+    if (stopped_) {
+      delete read_callback;
+      return;
+    }
+
+    // Otherwise enqueue the request.
     read_queue_.push_back(read_callback);
   }
-  if (FulfillPendingReads()) {
+
+  // See if we can immediately fulfill this read and whether we need to demux.
+  bool post_demux_task = FulfillPendingReads();
+
+  // Since Read() may be executed on any thread, it's possible that StopTask()
+  // finishes executing on the demuxer thread and the message loop goes away.
+  //
+  // To prevent that we'll grab the lock and post a task at the same time, which
+  // will keep the message loop alive.
+  AutoLock auto_lock(lock_);
+  if (post_demux_task && !stopped_) {
     demuxer_->PostDemuxTask();
   }
 }
 
 bool FFmpegDemuxerStream::FulfillPendingReads() {
   bool pending_reads = false;
   while (true) {
     scoped_refptr<Buffer> buffer;
     scoped_ptr<Callback1<Buffer*>::Type> read_callback;
     {
@@ skipping 17 matching lines @@
   }
   return pending_reads;
 }
 
 base::TimeDelta FFmpegDemuxerStream::ConvertTimestamp(int64 timestamp) {
   AVRational time_base = { 1, base::Time::kMicrosecondsPerSecond };
   int64 microseconds = av_rescale_q(timestamp, stream_->time_base, time_base);
   return base::TimeDelta::FromMicroseconds(microseconds);
 }
 
-
 //
 // FFmpegDemuxer
 //
 FFmpegDemuxer::FFmpegDemuxer()
     : thread_("DemuxerThread") {
 }
 
 FFmpegDemuxer::~FFmpegDemuxer() {
   DCHECK(!thread_.IsRunning());
   DCHECK(!format_context_.get());
   // TODO(scherkus): I believe we need to use av_close_input_file() here
   // instead of scoped_ptr_malloc calling av_free().
   //
   // Note that av_close_input_file() doesn't close the codecs so we need to
   // figure out who's responsible for closing the them.
 }
 
 void FFmpegDemuxer::PostDemuxTask() {
   thread_.message_loop()->PostTask(FROM_HERE,
       NewRunnableMethod(this, &FFmpegDemuxer::DemuxTask));
 }
 
 void FFmpegDemuxer::Stop() {
-  thread_.Stop();
+  // Stop our thread and post a task notifying the streams to stop as well.
+  if (thread_.IsRunning()) {
+    thread_.message_loop()->PostTask(FROM_HERE,
+        NewRunnableMethod(this, &FFmpegDemuxer::StopTask));
+    thread_.Stop();
+  }
   format_context_.reset();
 }
 
 void FFmpegDemuxer::Seek(base::TimeDelta time) {
   // TODO(hclam): by returning from this method, it is assumed that the seek
   // operation is completed and filters behind the demuxer is good to issue
   // more reads, but we are posting a task here, which makes the seek operation
   // asynchronous, should change how seek works to make it fully asynchronous.
   thread_.message_loop()->PostTask(FROM_HERE,
       NewRunnableMethod(this, &FFmpegDemuxer::SeekTask, time));
@@ skipping 157 matching lines @@
     av_free_packet(packet.get());
   }
 
   // Create a loop by posting another task.  This allows seek and message loop
   // quit tasks to get processed.
   if (StreamsHavePendingReads()) {
     PostDemuxTask();
   }
 }
 
+void FFmpegDemuxer::StopTask() {
+  StreamVector::iterator iter;
+  for (iter = streams_.begin(); iter != streams_.end(); ++iter) {
+    (*iter)->Stop();
+  }
+}
+
 bool FFmpegDemuxer::StreamsHavePendingReads() {
+  DCHECK_EQ(PlatformThread::CurrentId(), thread_.thread_id());
   StreamVector::iterator iter;
   for (iter = streams_.begin(); iter != streams_.end(); ++iter) {
     if ((*iter)->HasPendingReads()) {
       return true;
     }
   }
   return false;
 }
 
 void FFmpegDemuxer::StreamHasEnded() {
+  DCHECK_EQ(PlatformThread::CurrentId(), thread_.thread_id());
   StreamVector::iterator iter;
   for (iter = streams_.begin(); iter != streams_.end(); ++iter) {
     AVPacket* packet = new AVPacket();
     memset(packet, 0, sizeof(*packet));
     (*iter)->EnqueuePacket(packet);
   }
 }
 
-AVPacket* FFmpegDemuxer::ClonePacket(AVPacket* packet) {
-  scoped_ptr<AVPacket> clone(new AVPacket());
-  if (!clone.get() || av_new_packet(clone.get(), packet->size) < 0) {
-    return NULL;
-  }
-  DCHECK_EQ(clone->size, packet->size);
-  clone->dts = packet->dts;
-  clone->pts = packet->pts;
-  clone->duration = packet->duration;
-  memcpy(clone->data, packet->data, clone->size);
-  return clone.release();
-}
-
 }  // namespace media