Initial implementation of stream switching in AdaptiveDemuxer.

media/filters/adaptive_demuxer.cc

 // Copyright (c) 2011 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/bind.h"
 #include "base/logging.h"
+#include "base/message_loop.h"
 #include "base/string_number_conversions.h"
 #include "base/string_split.h"
 #include "base/string_util.h"
+#include "media/ffmpeg/ffmpeg_common.h"
 #include "media/filters/adaptive_demuxer.h"
 
 namespace media {
 
+static const int64 kSwitchTimerPeriod = 5000; // In milliseconds.
+
+// Object that decides when to switch streams.
+class StreamSwitchManager
+    : public base::RefCountedThreadSafe<StreamSwitchManager> {
+ public:
+  StreamSwitchManager();
+  virtual ~StreamSwitchManager();
+
+  void Init(AdaptiveDemuxer* demuxer);

[Ami GONE FROM CHROMIUM, 2011/05/24 06:14:12]

FWIW, could be ctor arg.

[acolwell GONE FROM CHROMIUM, 2011/05/24 23:45:16]

Done.

+
+  // Playback events. These methods are called when playback starts, pauses, or
+  // stops.
+  void Play();
+  void Pause();
+  void Stop();
+
+ private:
+  // Method called periodically to determine if we should switch
+  // streams.
+  void OnSwitchTimer();
+
+  // Called when the demuxer completes a stream switch.
+  void OnSwitchDone(PipelineStatus status);
+
+  // Helper method that schedules OnSwitchTimer() to be called.
+  void StartSwitchTimer();
+
+  // The demuxer that owns this object.
+  AdaptiveDemuxer* demuxer_;
+
+  // Message loop this object runs on.
+  MessageLoop* message_loop_;
+
+  // Is clip playing or not?
+  bool playing_;
+
+  // Is a stream switch in progress?
+  bool switch_pending_;
+
+  // Has an OnSwitchTimer() call been scheduled?
+  bool switch_timer_running_;

[scherkus (not reviewing), 2011/05/24 21:38:37]

may want to consider using OneShotTimer() from base/timer.h instead of
hand-rolling one

[acolwell GONE FROM CHROMIUM, 2011/05/24 23:45:16]

Done.

+
+  // The stream IDs for the video streams.
+  AdaptiveDemuxer::StreamIdVector video_ids_;
+
+  // An index into |video_ids_| for the current video stream.
+  int current_id_index_;
+
+  DISALLOW_COPY_AND_ASSIGN(StreamSwitchManager);
+};
+
+StreamSwitchManager::StreamSwitchManager()
+    : demuxer_(NULL),
+      playing_(false),
+      switch_pending_(false),
+      switch_timer_running_(false),
+      current_id_index_(-1) {
+}
+
+StreamSwitchManager::~StreamSwitchManager() {}

[Ami GONE FROM CHROMIUM, 2011/05/24 06:14:12]

What happens to an outstanding posted task whose time comes, after this manager
(and its demuxer) have been shut down / deallocated?  Does the task hold a
reference preventing refcounting from tearing these down?  Is that enough in the
face of global teardown (i.e. are we safe from crashing at exit time)?

[acolwell GONE FROM CHROMIUM, 2011/05/24 23:45:16]

The task holds a reference that prevents destruction. 

I've made sure that both methods that are referenced by PostXXTask()
(OnSwitchDone && OnSwitchTimer) can safely run if the demuxer has been
destroyed. Stop() sets demuxer_ to NULL and OnSwitchTimer() doesn't allow a
stream switch and doesn't reschedule the timer if demuxer_ is NULL.

+
+void StreamSwitchManager::Init(AdaptiveDemuxer* demuxer) {
+  DCHECK(demuxer);
+  message_loop_ = MessageLoop::current();
+  demuxer_ = demuxer;
+  video_ids_ = demuxer->GetVideoIds();
+  current_id_index_ = -1;
+
+  if (video_ids_.size() > 0) {
+    // Find the index in video_ids_ for the current video ID.
+    int current_id = demuxer->GetCurrentVideoId();
+    current_id_index_ = 0;
+    for (int i = 0; i < video_ids_.size(); i++) {
+      if (current_id == video_ids_[i]) {
+        current_id_index_ = i;
+        break;
+      }
+    }
+  }
+}
+
+void StreamSwitchManager::Play() {
+  DCHECK_EQ(MessageLoop::current(), message_loop_);
+  DCHECK(!playing_);
+  playing_ = true;
+
+  if (video_ids_.size() > 1 && !switch_timer_running_) {
+    StartSwitchTimer();
+  }
+}
+
+void StreamSwitchManager::Pause() {
+  DCHECK_EQ(MessageLoop::current(), message_loop_);
+  DCHECK(playing_);
+  playing_ = false;
+}
+
+void StreamSwitchManager::Stop() {
+  DCHECK_EQ(MessageLoop::current(), message_loop_);
+  DCHECK(!playing_);
+  demuxer_ = NULL;
+}
+
+void StreamSwitchManager::OnSwitchTimer() {
+  DCHECK_EQ(MessageLoop::current(), message_loop_);
+
+  switch_timer_running_ = false;
+
+  if (!playing_)
+    return;
+
+  if (!switch_pending_) {

[Ami GONE FROM CHROMIUM, 2011/05/24 06:14:12]

reverse the test and early-return lets you de-indent body.

[acolwell GONE FROM CHROMIUM, 2011/05/24 23:45:16]

Done.

+    // Select the stream to switch to. For now we are just rotating
+    // through the available streams.
+    int new_id_index = (current_id_index_ + 1) % video_ids_.size();
+
+    // Did we select a new stream?
+    if (new_id_index != current_id_index_) {

[Ami GONE FROM CHROMIUM, 2011/05/24 06:14:12]

Adding an if (video_ids_.size() < 2) return; to the top of the function lets you
de-indent the rest (and incidentally fix the no-impact buglet of starting the
timer every 5s even when there's a single or no video streams).

[acolwell GONE FROM CHROMIUM, 2011/05/24 23:45:16]

Done.

+      current_id_index_ = new_id_index;
+      switch_pending_ = true;
+      demuxer_->ChangeVideoStream(video_ids_[new_id_index],
+                                  base::Bind(&StreamSwitchManager::OnSwitchDone,
+                                             this));
+    }
+  }
+
+  StartSwitchTimer();
+}
+
+void StreamSwitchManager::OnSwitchDone(PipelineStatus status) {
+  if (MessageLoop::current() != message_loop_) {
+    message_loop_->PostTask(
+        FROM_HERE,
+        NewRunnableMethod(this,&StreamSwitchManager::OnSwitchDone, status));

[Ami GONE FROM CHROMIUM, 2011/05/24 06:14:12]

return after this!

[acolwell GONE FROM CHROMIUM, 2011/05/24 23:45:16]

Good catch. Done

+  }
+  switch_pending_ = false;
+}
+
+void StreamSwitchManager::StartSwitchTimer() {
+  DCHECK_EQ(MessageLoop::current(), message_loop_);
+  switch_timer_running_ = true;
+  message_loop_->PostDelayedTask(
+      FROM_HERE, NewRunnableMethod(this, &StreamSwitchManager::OnSwitchTimer),
+      kSwitchTimerPeriod);
+}
+
 //
 // AdaptiveDemuxerStream
 //
 AdaptiveDemuxerStream::AdaptiveDemuxerStream(
     StreamVector const& streams, int initial_stream)
     : streams_(streams), current_stream_index_(initial_stream),
-      bitstream_converter_enabled_(false) {
+      bitstream_converter_enabled_(false),
+      pending_reads_(0),
+      switch_index_(-1),
+      last_buffer_timestamp_(kNoTimestamp) {
   DCheckSanity();
 }
 
 void AdaptiveDemuxerStream::DCheckSanity() {
   // We only carry out sanity checks in debug mode.
   if (!logging::DEBUG_MODE)
     return;
   DCHECK(streams_[current_stream_index_].get());
 
   bool non_null_stream_seen = false;
@@ skipping 19 matching lines @@
   current_stream_index_ = -1;
   streams_.clear();
 }
 
 DemuxerStream* AdaptiveDemuxerStream::current_stream() {
   base::AutoLock auto_lock(lock_);
   return streams_[current_stream_index_];
 }
 
 void AdaptiveDemuxerStream::Read(const ReadCallback& read_callback) {
-  current_stream()->Read(read_callback);
+  DemuxerStream* stream = NULL;
+
+  {
+    base::AutoLock auto_lock(lock_);
+
+    read_cb_queue_.push_back(read_callback);
+
+    // Check to make sure we aren't doing a stream switch. We only want to
+    // make calls on |streams_[current_stream_index_]| when we aren't
+    // in the middle of a stream switch. Since the callback is stored in
+    // |read_cb_queue_| we will issue the Read() on the new stream once
+    // the switch has completed.
+    if (!IsSwitchPending_Locked()) {
+      stream = streams_[current_stream_index_];
+      pending_reads_++;
+    }
+  }
+
+  if (stream)
+    stream->Read(base::Bind(&AdaptiveDemuxerStream::OnReadDone, this));
 }
 
 AVStream* AdaptiveDemuxerStream::GetAVStream() {
   return current_stream()->GetAVStream();
 }
 
 DemuxerStream::Type AdaptiveDemuxerStream::type() {
   return current_stream()->type();
 }
 
 const MediaFormat& AdaptiveDemuxerStream::media_format() {
   return current_stream()->media_format();
 }
 
 void AdaptiveDemuxerStream::EnableBitstreamConverter() {
   {
     base::AutoLock auto_lock(lock_);
     bitstream_converter_enabled_ = true;
   }
   current_stream()->EnableBitstreamConverter();
 }
 
-void AdaptiveDemuxerStream::ChangeCurrentStream(int index) {
-  bool needs_bitstream_converter_enabled;
-  {
-    base::AutoLock auto_lock(lock_);
-    current_stream_index_ = index;
-    DCHECK(streams_[current_stream_index_]);
-    needs_bitstream_converter_enabled = bitstream_converter_enabled_;
-  }
+void AdaptiveDemuxerStream::OnAdaptiveDemuxerSeek() {
+  base::AutoLock auto_lock(lock_);
+
+  last_buffer_timestamp_ = kNoTimestamp;

[Ami GONE FROM CHROMIUM, 2011/05/24 06:14:12]

I'm surprised this doesn't store the seek target time.

[acolwell GONE FROM CHROMIUM, 2011/05/24 23:45:16]

Good catch. Updated the method signature to pass the seek target time.

+
+  // TODO(acolwell): Figure out what to do if this happens during a stream
+  // switch.
+}
+
+// This method initiates a stream switch. The diagram below shows the steps
+// involved.
+//
+//                           +-----------------------+
+// ChangeCurrentStream() ->  | Store stream switch   |
+//                           | index, seek_function, |
+//                           |    and switch_cb.     |
+//                           +-----------------------+
+//                                   |
+//                                  \|/

[Ami GONE FROM CHROMIUM, 2011/05/24 06:14:12]

You might enjoy v and ^ as arrow-heads for ASCII art.

[scherkus (not reviewing), 2011/05/24 21:38:37]

hahah I was going to make that same comment but wondered if that was going
overboard

[acolwell GONE FROM CHROMIUM, 2011/05/24 23:45:16]

oh yeah.. duh. Done

+//                          +-------------------+ Yes
+//                          | Are there pending | -----> (Wait for OnReadDone())
+//                          | Read()s on the    |                |
+//                          | current stream?   | <-----+       \|/
+//                          +-------------------+       +--- OnReadDone()
+//                                   | No
+//                                  \|/
+//                              StartSwitch()
+//                                   |
+//                                  \|/
+//                        +------------------------+ No
+//                        | Have buffer timestamp? |-----+
+//                        +------------------------+     |
+//                                   | Yes               |
+//                                  \|/                  |
+//                     +-----------------------------+   |
+//                     | Seek stream to timestamp    |   |
+//                     | using switch_seek_function_ |   |
+//                     +-----------------------------+   |
+//                                   |                   |
+//                                  \|/                  |
+//                              OnSwitchSeekDone() <-----+
+//                                   |
+//                                  \|/
+//                         +------------------+ No
+//                         | Seek successful? |----------+
+//                         +------------------+          |
+//                                   | Yes               |
+//                                  \|/                  |
+//                     +------------------------------+  |
+//                     | Update current_stream_index_ |  |
+//                     +------------------------------+  |
+//                                   |                   |
+//                                  \|/                  |
+//                      +---------------------------+    |
+//                      | Call Read() on new stream |<---+
+//                      |    for deferred reads.    |
+//                      +---------------------------+
+//                                   |
+//                                  \|/
+//                            switch_cb(OK | ERROR)
+//
+// NOTE: Any AdaptiveDemuxerStream::Read() calls that occur during the stream
+//       switch will be deferred until the switch has completed. The callbacks
+//       will be queued on |read_cb_queue_|, but no Read() will be issued on the
+//       current stream.

[Ami GONE FROM CHROMIUM, 2011/05/24 06:14:12]

Does this mean that if the switched-to stream stalls loading, it'll cause the
current stream (and the user-visible state) to pause?

[acolwell GONE FROM CHROMIUM, 2011/05/24 23:45:16]

Yes. 

In follow up CLs this will likely be tuned to minimize user-visible impact.
We'll likely seek the switched-to stream a few seconds into the future and
buffer a little data on that stream before doing the actual switch.

I just wanted to keep the initial CL as simple as possible.

[Ami GONE FROM CHROMIUM, 2011/05/25 05:39:01]

Sounds reasonable.  Add TODO?

+void AdaptiveDemuxerStream::ChangeCurrentStream(
+    int index,
+    const SeekFunction& seek_function,
+    const PipelineStatusCB& switch_cb) {

[scherkus (not reviewing), 2011/05/24 21:38:37]

q: what's our goal w/ PipelineStatus + respective callbacks?

would we be better off with AdaptiveStreamingStatus error codes or are we fine
stashing everything into PipelineStatus?

I'm indifferent.

+  DCHECK_GE(index, 0);
+
+  PipelineStatusCB error_cb;
+  PipelineStatus error_status = PIPELINE_ERROR_INVALID_STATE;
+
+  bool start_switch = false;
+
+  {
+    base::AutoLock auto_lock(lock_);
+
+    DCHECK_LE(index, streams_.size());
+    DCHECK(streams_[index].get());
+    DCHECK(!IsSwitchPending_Locked());
+
+    // TODO(acolwell): Still need to handle the case where the stream has ended.
+    if (index == current_stream_index_) {
+      error_cb = switch_cb;
+      error_status = PIPELINE_OK;
+    } else {
+      switch_cb_ = switch_cb;
+      switch_index_ = index;
+      switch_seek_function_ = seek_function;
+
+      start_switch = CanStartSwitch_Locked();
+    }
+  }
+
+  if (!error_cb.is_null()) {

[Ami GONE FROM CHROMIUM, 2011/05/24 06:14:12]

I think you can lose error_cb by initializing error_status to OK and here asking
if (error_status != OK) {
  switch_cb.Run(error_status);
  return;
}
?

[acolwell GONE FROM CHROMIUM, 2011/05/24 23:45:16]

Removed error_cb & error_status per our discussion.

+    error_cb.Run(error_status);
+    return;
+  }
+
+  if (start_switch)
+    StartSwitch();
+}
+
+void AdaptiveDemuxerStream::OnReadDone(Buffer* buffer) {
+  ReadCallback read_cb;
+  bool start_switch = false;
+
+  {
+    base::AutoLock auto_lock(lock_);
+
+    pending_reads_--;
+
+    DCHECK_GE(pending_reads_, 0);
+    DCHECK_GE(read_cb_queue_.size(), 0);
+
+    read_cb = read_cb_queue_.front();
+    read_cb_queue_.pop_front();
+
+    if (buffer && !buffer->IsEndOfStream())
+      last_buffer_timestamp_ = buffer->GetTimestamp();
+
+    start_switch = IsSwitchPending_Locked() && CanStartSwitch_Locked();
+  }
+
+  if (!read_cb.is_null())
+    read_cb.Run(buffer);
+
+  if (start_switch)

[Ami GONE FROM CHROMIUM, 2011/05/24 06:14:12]

Is this thread-safe?  Just b/c there were no pending read under the lock above,
doesn't mean someone else didn't come in and queue up some reads under you, does
it?
ISTM you need to test & set the switch_pending state under lock.

[acolwell GONE FROM CHROMIUM, 2011/05/24 23:45:16]

I'm pretty sure this is safe.
switch_pending state (ie !switch_cb_.is_null()) only gets changed in
ChangeCurrentStream() & OnSwitchSeekDone(). Once the switch_pending state is set
in ChangeCurrentStream(), Read() will not make
streams_[current_stream_index_]->Read() calls. We don't have to worry about the
the pending state changing outside of lock and since we are deferring Reads() we
don't have to worry about pending_reads_ being modified.

+    StartSwitch();
+}
+
+bool AdaptiveDemuxerStream::IsSwitchPending_Locked() const {
+  lock_.AssertAcquired();
+  return !switch_cb_.is_null();
+}
+
+bool AdaptiveDemuxerStream::CanStartSwitch_Locked() const {
+  lock_.AssertAcquired();
+  return (pending_reads_ == 0);

[Ami GONE FROM CHROMIUM, 2011/05/24 06:14:12]

Seems strange that this doesn't inspect switch_cb_ at all.
Maybe I expect it to be
if (pending_reads_ == 0) {
  DCHECK(switch_cb_.is_null());
  return true;
}
return false;
?

[acolwell GONE FROM CHROMIUM, 2011/05/24 23:45:16]

Done.

+}
+
+void AdaptiveDemuxerStream::StartSwitch() {
+  SeekFunction seek_function;
+  base::TimeDelta seek_point;
+
+  {
+    base::AutoLock auto_lock(lock_);
+    DCHECK(IsSwitchPending_Locked());
+    DCHECK_EQ(pending_reads_, 0);
+    DCHECK_GE(switch_index_, 0);
+
+    seek_point = last_buffer_timestamp_;
+    seek_function = switch_seek_function_;
+
+    // TODO(acolwell): add code to call switch_cb_ if we are at the end of the
+    // stream now.
+  }
+
+  if (seek_point == kNoTimestamp) {
+    // We haven't seen a buffer so there is no need to seek. Just move on to
+    // the next stage in the switch process.
+    OnSwitchSeekDone(PIPELINE_OK, kNoTimestamp);
+    return;
+  }
+
+  DCHECK(!seek_function.is_null());
+  seek_function.Run(seek_point,
+                    base::Bind(&AdaptiveDemuxerStream::OnSwitchSeekDone, this));
+}
+
+void AdaptiveDemuxerStream::OnSwitchSeekDone(PipelineStatus status,
+                                             base::TimeDelta seek_time) {
+  DemuxerStream* stream = NULL;
+  PipelineStatusCB switch_cb;
+  int reads_to_request = 0;
+  bool needs_bitstream_converter_enabled = false;
+
+  {
+    base::AutoLock auto_lock(lock_);
+
+    if (status == PIPELINE_OK) {
+      DCHECK(streams_[switch_index_]);
+
+      current_stream_index_ = switch_index_;
+      needs_bitstream_converter_enabled = bitstream_converter_enabled_;
+    }
+
+    // Clear stream switch state.
+    switch_cb = switch_cb_;
+    switch_cb_.Reset();

[Ami GONE FROM CHROMIUM, 2011/05/24 06:14:12]

swap instead of =+Reset()?

[acolwell GONE FROM CHROMIUM, 2011/05/24 23:45:16]

no swap() method. :(

[Ami GONE FROM CHROMIUM, 2011/05/25 05:39:01]

new-style callbacks are assignable, so std::swap() should do you.

[acolwell GONE FROM CHROMIUM, 2011/05/25 21:43:46]

Done.

+    switch_index_ = -1;
+    switch_seek_function_.Reset();
+
+    // Get the number of outstanding Read()s on this object.
+    reads_to_request = read_cb_queue_.size();
+
+    DCHECK_EQ(pending_reads_, 0);
+    pending_reads_ = reads_to_request;
+
+    stream = streams_[current_stream_index_];
+  }
+
   if (needs_bitstream_converter_enabled)
     EnableBitstreamConverter();
+
+  if (stream) {

[Ami GONE FROM CHROMIUM, 2011/05/24 06:14:12]

Do we really allow switching to a NULL stream index?  
Pending reads will never fire if so...

[acolwell GONE FROM CHROMIUM, 2011/05/24 23:45:16]

No. Removed if() and added DCHECK

+    // Make the Read() calls that were deferred during the stream switch.
+    for(;reads_to_request > 0; --reads_to_request)
+      stream->Read(base::Bind(&AdaptiveDemuxerStream::OnReadDone, this));
+  }
+
+  // Signal that the stream switch has completed.
+  if (!switch_cb.is_null())
+    switch_cb.Run(status);
 }
 
 //
 // AdaptiveDemuxer
 //
 
 AdaptiveDemuxer::AdaptiveDemuxer(DemuxerVector const& demuxers,
                                  int initial_audio_demuxer_index,
                                  int initial_video_demuxer_index)
     : demuxers_(demuxers),
       current_audio_demuxer_index_(initial_audio_demuxer_index),
-      current_video_demuxer_index_(initial_video_demuxer_index) {
+      current_video_demuxer_index_(initial_video_demuxer_index),
+      playback_rate_(0),
+      switch_pending_(false),
+      stream_switch_manager_(new StreamSwitchManager()){
   DCHECK(!demuxers_.empty());
   DCHECK_GE(current_audio_demuxer_index_, -1);
   DCHECK_GE(current_video_demuxer_index_, -1);
   DCHECK_LT(current_audio_demuxer_index_, static_cast<int>(demuxers_.size()));
   DCHECK_LT(current_video_demuxer_index_, static_cast<int>(demuxers_.size()));
   DCHECK(current_audio_demuxer_index_ != -1 ||
          current_video_demuxer_index_ != -1);
   AdaptiveDemuxerStream::StreamVector audio_streams, video_streams;
   for (size_t i = 0; i < demuxers_.size(); ++i) {
     audio_streams.push_back(demuxers_[i]->GetStream(DemuxerStream::AUDIO));
     video_streams.push_back(demuxers_[i]->GetStream(DemuxerStream::VIDEO));
   }
   if (current_audio_demuxer_index_ >= 0) {
     audio_stream_ = new AdaptiveDemuxerStream(
         audio_streams, current_audio_demuxer_index_);
   }
   if (current_video_demuxer_index_ >= 0) {
     video_stream_ = new AdaptiveDemuxerStream(
         video_streams, current_video_demuxer_index_);
   }
 
+  stream_switch_manager_->Init(this);
+
   // TODO(fischman): any streams in the underlying demuxers that aren't being
   // consumed currently need to be sent to /dev/null or else FFmpegDemuxer will
   // hold data for them in memory, waiting for them to get drained by a
   // non-existent decoder.
 }
 
 AdaptiveDemuxer::~AdaptiveDemuxer() {}
 
-void AdaptiveDemuxer::ChangeCurrentDemuxer(int audio_index, int video_index) {
+// Switches the current video stream. The diagram below describes the switch
+// process.
+//                          +-------------------------------+
+// ChangeVideoStream() ---> |          video_index          |
+//                          |              ==               | Yes.
+//                          | current_video_demuxer_index_? |--> done_cb(OK)
+//                          +-------------------------------+
+//                                          | No.
+//                                         \|/
+//                         Call video_stream_->ChangeCurrentStream()
+//                                          |
+//                                         \|/
+//                          (Wait for ChangeVideoStreamDone())
+//                                          |
+//                                         \|/
+//                                ChangeVideoStreamDone()
+//                                          |
+//                                         \|/
+//                             +------------------------+ No
+//                             | Was switch successful? |---> done_cb(ERROR)
+//                             +------------------------+
+//                                          |
+//                                         \|/
+//                            Update current_video_demuxer_index_
+//                            & SetPlaybackRate() on new stream.
+//                                          |
+//                                         \|/
+//                                      done_cb(OK).
+//
+void AdaptiveDemuxer::ChangeVideoStream(int video_index,
+                                        const PipelineStatusCB& done_cb) {
   // TODO(fischman): this is currently broken because when a new Demuxer is to
   // be used we need to set_host(host()) it, and we need to set_host(NULL) the
   // current Demuxer if it's no longer being used.
+  AdaptiveDemuxerStream* stream = NULL;
+  bool switching_to_current_stream = false;
+  AdaptiveDemuxerStream::SeekFunction seek_function;
+
+  {
+    base::AutoLock auto_lock(lock_);
+
+    DCHECK(video_stream_);
+    DCHECK(!switch_pending_);
+    if (current_video_demuxer_index_ == video_index) {
+      switching_to_current_stream = true;
+    } else {
+      stream = video_stream_;
+      switch_pending_ = true;
+      seek_function = base::Bind(&AdaptiveDemuxer::StartStreamSwitchSeek,
+                                 this,
+                                 DemuxerStream::VIDEO,
+                                 video_index);
+    }
+  }
+
+  if (switching_to_current_stream) {
+    done_cb.Run(PIPELINE_OK);
+    return;
+  }
+
+  DCHECK(stream);
+  stream->ChangeCurrentStream(

[Ami GONE FROM CHROMIUM, 2011/05/24 06:14:12]

Once you set switch_pending_=true under lock above, isn't it safe to simply use
video_stream_ here?

[acolwell GONE FROM CHROMIUM, 2011/05/24 23:45:16]

Yes I think so. Removed stream variable.

+      video_index, seek_function,
+      base::Bind(&AdaptiveDemuxer::ChangeVideoStreamDone, this, video_index,
+                 done_cb));
+}
+
+int AdaptiveDemuxer::GetCurrentVideoId() const {
   base::AutoLock auto_lock(lock_);
-  current_audio_demuxer_index_ = audio_index;
-  current_video_demuxer_index_ = video_index;
-  if (audio_stream_)
-    audio_stream_->ChangeCurrentStream(audio_index);
-  if (video_stream_)
-    video_stream_->ChangeCurrentStream(video_index);
+  return current_video_demuxer_index_;
+}
+
+AdaptiveDemuxer::StreamIdVector AdaptiveDemuxer::GetVideoIds() const {
+  StreamIdVector ids;
+  base::AutoLock auto_lock(lock_);

[Ami GONE FROM CHROMIUM, 2011/05/24 06:14:12]

Is the return value from this function really something that can vary during the
lifetime of the AdaptiveDemuxer?
(I'm wondering about the need for the lock; if you remove it, please document
why in a comment).

[acolwell GONE FROM CHROMIUM, 2011/05/24 23:45:16]

This info shouldn't change during playback. I've removed this method and moved
the code to the constructor. This was only needed by StreamSwitchManager.

+
+  for (int i = 0; i < demuxers_.size(); i++) {
+    if (demuxers_[i]->GetStream(DemuxerStream::VIDEO))
+      ids.push_back(i);
+  }
+
+  return ids;
+}
+
+void AdaptiveDemuxer::ChangeVideoStreamDone(int new_stream_index,
+                                            const PipelineStatusCB& done_cb,
+                                            PipelineStatus status) {
+  {
+    base::AutoLock auto_lock(lock_);
+
+    switch_pending_ = false;
+
+    if (status == PIPELINE_OK) {
+      demuxers_[current_video_demuxer_index_]->SetPlaybackRate(0);
+      current_video_demuxer_index_ = new_stream_index;

[Ami GONE FROM CHROMIUM, 2011/05/24 06:14:12]

Whee!

+      demuxers_[current_video_demuxer_index_]->SetPlaybackRate(playback_rate_);
+    }
+  }
+
+  done_cb.Run(status);
 }
 
 Demuxer* AdaptiveDemuxer::current_demuxer(DemuxerStream::Type type) {
   base::AutoLock auto_lock(lock_);
   switch (type) {
     case DemuxerStream::AUDIO:
       return (current_audio_demuxer_index_ < 0) ? NULL :
           demuxers_[current_audio_demuxer_index_];
     case DemuxerStream::VIDEO:
       return (current_video_demuxer_index_ < 0) ? NULL :
           demuxers_[current_video_demuxer_index_];
     default:
       LOG(DFATAL) << "Unexpected type: " << type;
       return NULL;
   }
 }
 
 // Helper class that wraps a FilterCallback and expects to get called a set
 // number of times, after which the wrapped callback is fired (and deleted).
 //
-// TODO: Remove this class once Stop() is converted to FilterStatusCB.
+// TODO(acolwell): Remove this class once Stop() is converted to FilterStatusCB.
 class CountingCallback {
  public:
   CountingCallback(int count, FilterCallback* orig_cb)
       : remaining_count_(count), orig_cb_(orig_cb) {
     DCHECK_GT(remaining_count_, 0);
     DCHECK(orig_cb);
   }
 
   FilterCallback* GetACallback() {
     return NewCallback(this, &CountingCallback::OnChildCallbackDone);
@@ skipping 58 matching lines @@
 
   base::Lock lock_;
   int remaining_count_;
   FilterStatusCB orig_cb_;
   PipelineStatus overall_status_;
 
   DISALLOW_COPY_AND_ASSIGN(CountingStatusCB);
 };
 
 void AdaptiveDemuxer::Stop(FilterCallback* callback) {
+  stream_switch_manager_->Stop();
+
   // Stop() must be called on all of the demuxers even though only one demuxer
   // is  actively delivering audio and another one is delivering video. This
   // just satisfies the contract that all demuxers must have Stop() called on
   // them before they are destroyed.
   //
-  // TODO: Remove CountingCallback once Stop() is converted to FilterStatusCB.
+  // TODO(acolwell): Remove CountingCallback once Stop() is converted to
+  // FilterStatusCB.
   CountingCallback* wrapper = new CountingCallback(demuxers_.size(), callback);
   for (size_t i = 0; i < demuxers_.size(); ++i)
     demuxers_[i]->Stop(wrapper->GetACallback());
 }
 
 void AdaptiveDemuxer::Seek(base::TimeDelta time, const FilterStatusCB& cb) {
+
+  if (audio_stream_)
+    audio_stream_->OnAdaptiveDemuxerSeek();
+
+  if (video_stream_)
+    video_stream_->OnAdaptiveDemuxerSeek();
+
   Demuxer* audio = current_demuxer(DemuxerStream::AUDIO);
   Demuxer* video = current_demuxer(DemuxerStream::VIDEO);
   int count = (audio ? 1 : 0) + (video && audio != video ? 1 : 0);
   CountingStatusCB* wrapper = new CountingStatusCB(count, cb);
   if (audio)
     audio->Seek(time, wrapper->GetACallback());
   if (video && audio != video)
     video->Seek(time, wrapper->GetACallback());
 }
 
 void AdaptiveDemuxer::OnAudioRendererDisabled() {
   for (size_t i = 0; i < demuxers_.size(); ++i)
     demuxers_[i]->OnAudioRendererDisabled();
 }
 
 void AdaptiveDemuxer::set_host(FilterHost* filter_host) {
   Demuxer* audio = current_demuxer(DemuxerStream::AUDIO);
   Demuxer* video = current_demuxer(DemuxerStream::VIDEO);
   if (audio) audio->set_host(filter_host);
   if (video && audio != video) video->set_host(filter_host);
 }
 
 void AdaptiveDemuxer::SetPlaybackRate(float playback_rate) {
+  {
+    base::AutoLock auto_lock(lock_);
+    if (playback_rate_ == 0 && playback_rate > 0) {
+      stream_switch_manager_->Play();
+    } else if (playback_rate_ > 0 && playback_rate == 0) {
+      stream_switch_manager_->Pause();
+    }
+
+    playback_rate_ = playback_rate;
+  }
+
   Demuxer* audio = current_demuxer(DemuxerStream::AUDIO);
   Demuxer* video = current_demuxer(DemuxerStream::VIDEO);
   if (audio) audio->SetPlaybackRate(playback_rate);
   if (video && audio != video) video->SetPlaybackRate(playback_rate);
 }
 
 void AdaptiveDemuxer::SetPreload(Preload preload) {
   for (size_t i = 0; i < demuxers_.size(); ++i)
     demuxers_[i]->SetPreload(preload);
 }
 
 scoped_refptr<DemuxerStream> AdaptiveDemuxer::GetStream(
     DemuxerStream::Type type) {
   switch (type) {
     case DemuxerStream::AUDIO:
       return audio_stream_;
     case DemuxerStream::VIDEO:
       return video_stream_;
     default:
       LOG(DFATAL) << "Unexpected type " << type;
       return NULL;
   }
 }
 
+void AdaptiveDemuxer::StartStreamSwitchSeek(
+    DemuxerStream::Type type,
+    int stream_index,
+    base::TimeDelta seek_time,
+    const AdaptiveDemuxerStream::SeekFunctionCB& seek_cb) {
+  DCHECK_GE(stream_index, 0);
+  DCHECK(!seek_cb.is_null());
+
+  Demuxer* demuxer = NULL;
+  base::TimeDelta seek_point;
+  FilterStatusCB filter_cb;
+
+  {
+    base::AutoLock auto_lock(lock_);
+
+    demuxer = demuxers_[stream_index];
+
+    if (GetSeekTimeAfter(demuxer->GetStream(type)->GetAVStream(), seek_time,
+                         &seek_point)) {
+      // We found a seek point.
+      filter_cb = base::Bind(&AdaptiveDemuxer::OnStreamSeekDone, this,
+                             seek_point, seek_cb);
+    } else {
+      // We didn't find a seek point. Assume we don't have index data for it
+      // yet. Seek to the specified time to force index data to be loaded.

[scherkus (not reviewing), 2011/05/24 21:38:37]

OOC does this happen due to deferred CUES fetching?

[acolwell GONE FROM CHROMIUM, 2011/05/24 23:45:16]

Yes. It is a temporary hack until we have a WebM demuxer or we expose a better
way to get index data via FFmpeg.

+      seek_point = seek_time;
+      filter_cb = base::Bind(&AdaptiveDemuxer::OnIndexSeekDone, this,
+                             type, stream_index, seek_time, seek_cb);
+    }
+  }
+
+  DCHECK(demuxer);
+  demuxer->Seek(seek_point, filter_cb);
+}
+
+void AdaptiveDemuxer::OnIndexSeekDone(
+    DemuxerStream::Type type,
+    int stream_index,
+    base::TimeDelta seek_time,
+    const AdaptiveDemuxerStream::SeekFunctionCB& seek_cb,
+    PipelineStatus status) {
+  base::TimeDelta seek_point;
+  FilterStatusCB filter_cb;
+
+  Demuxer* demuxer = NULL;
+
+  if (status != PIPELINE_OK) {
+    seek_cb.Run(status, base::TimeDelta());

[Ami GONE FROM CHROMIUM, 2011/05/24 06:14:12]

The ASCII art in this file (and this sort of error handling) looks to me like a
great TOC of a test suite :)

+    return;
+  }
+
+  {
+    base::AutoLock auto_lock(lock_);
+
+    demuxer = demuxers_[stream_index];
+    DCHECK(demuxer);
+
+    // Look for a seek point now that we have index data.
+    if (GetSeekTimeAfter(demuxer->GetStream(type)->GetAVStream(), seek_time,
+                         &seek_point)) {
+      filter_cb = base::Bind(&AdaptiveDemuxer::OnStreamSeekDone, this,
+                             seek_point, seek_cb);
+    } else {
+      // Failed again to find a seek point. Clear the demuxer so that
+      // a seek error will be reported.
+      demuxer = NULL;
+    }
+  }
+
+  if (!demuxer) {
+    seek_cb.Run(PIPELINE_ERROR_INITIALIZATION_FAILED, base::TimeDelta());
+    return;
+  }
+
+  demuxer->Seek(seek_point, filter_cb);
+}
+
+void AdaptiveDemuxer::OnStreamSeekDone(
+    base::TimeDelta seek_point,
+    const AdaptiveDemuxerStream::SeekFunctionCB& seek_cb,
+    PipelineStatus status) {
+  if (status != PIPELINE_OK) {
+    seek_cb.Run(status, base::TimeDelta());
+    return;
+  }
+
+  seek_cb.Run(PIPELINE_OK, seek_point);

[Ami GONE FROM CHROMIUM, 2011/05/24 06:14:12]

FWIW method body can just be
seek_cb.Run(status, seek_point);
so I think this method can be elided entirely (just Bind the expression above in
place of Binding this method).

[acolwell GONE FROM CHROMIUM, 2011/05/24 23:45:16]

Remove all the logic in the function so now it just has
seek_cb.Run(status,seek_point). I couldn't figure out the appropriate Bind magic
to eliminate the need for this function.
base::Bind(&AdaptiveDemuxerStream::SeekFunctionCB::Run,
                             base::Unretained(&seek_cb),
                             seek_point) compiles, but using the reference of a
local variable is obviously uncool. I think Callback needs some sort of Apply()
template method that converts an N-ary Callback into an (N-M)-ary Callback.

[Ami GONE FROM CHROMIUM, 2011/05/25 05:39:01]

Re: Apply: that's what Bind *is*.
Re: ref-of-local-var: it's only a problem b/c you're using Unretained.  Does
dropping that make your life better?

[acolwell GONE FROM CHROMIUM, 2011/05/25 21:43:46]

yeah. I wasn't quite thinking straight here.
No. If I remove base::Unretained(&), it complains about seek_cb not being a
pointer and that it doesn't support AddRef() and Release().

+}
+
+
 //
 // AdaptiveDemuxerFactory
 //
 
 AdaptiveDemuxerFactory::AdaptiveDemuxerFactory(
     DemuxerFactory* delegate_factory)
     : delegate_factory_(delegate_factory) {
   DCHECK(delegate_factory);
 }
 
@@ skipping 172 matching lines @@
     delete cb;
     return;
   }
   DemuxerAccumulator* accumulator = new DemuxerAccumulator(
       audio_index, video_index, urls.size(), cb);
   for (size_t i = 0; i < urls.size(); ++i)
     delegate_factory_->Build(urls[i], accumulator->GetNthCallback(i));
 }
 
 }  // namespace media