Move AudioRenderer out of Filter heirarchy.

media/base/pipeline.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/base/pipeline.h"
 
 #include <algorithm>
 
 #include "base/bind.h"
 #include "base/callback.h"
@@ skipping 43 matching lines @@
   while (!notified_)
     cv_.Wait();
 }
 
 media::PipelineStatus PipelineStatusNotification::status() {
   base::AutoLock auto_lock(lock_);
   DCHECK(notified_);
   return status_;
 }
 
 struct Pipeline::PipelineInitState {

[scherkus (not reviewing), 2012/07/10 03:58:22]

I think we can remove this entirely very soon

   scoped_refptr<AudioDecoder> audio_decoder;
   scoped_refptr<VideoDecoder> video_decoder;
-  scoped_refptr<AudioRenderer> audio_renderer;
   scoped_refptr<VideoRenderer> video_renderer;
   scoped_refptr<CompositeFilter> composite;

[acolwell GONE FROM CHROMIUM, 2012/07/10 16:34:37]

If this is only holding video_renderer now can we just remove composite?

[scherkus (not reviewing), 2012/07/17 03:43:00]

Not yet because we set pipeline_filter_ to composite after init completes

 };
 
 Pipeline::Pipeline(MessageLoop* message_loop, MediaLog* media_log)
     : message_loop_(message_loop->message_loop_proxy()),
       media_log_(media_log),
       clock_(new Clock(&base::Time::Now)),
       waiting_for_clock_update_(false),
       state_(kCreated),
       creation_time_(base::Time::Now()) {
   media_log_->AddEvent(media_log_->CreatePipelineStateChangedEvent(kCreated));
@@ skipping 367 matching lines @@
   // Since the byte->time calculation is approximate, fudge the beginning &
   // ending areas to look better.
   TimeDelta epsilon = clock_->Duration() / 100;
   if (time_offset < epsilon)
     return TimeDelta();
   if (time_offset + epsilon > clock_->Duration())
     return clock_->Duration();
   return time_offset;
 }
 
+void Pipeline::DoPause(const base::Closure& done_cb) {

[scherkus (not reviewing), 2012/07/10 03:58:22]

these DoXXX() methods essentially replicate CompositeFilter's async serial
execution but via closure chaining.

[acolwell GONE FROM CHROMIUM, 2012/07/10 16:34:37]

Wow. This is not as clean as I had hoped. I think having a ClosureListHelper or
something might make this more straightforward and remove some of the duplicate
code. IWBN if you could make the code look something like this.

ClosureList closures;

if (audio_renderer_) {
   closures.Add(base::Bind(&AudioRenderer::Pause, audio_renderer_));

if (pipeline_filter_)
  closures.Add(base::Bind(&Filter::Pause,pipeline_filter_));

closures.Run(done_cb);

To me this seems more readable & execution order corresponds to reading the code
from top to bottom. The existing code is a little confusing because reading top
to bottom actually shows execution in the reverse order.

[scherkus (not reviewing), 2012/07/17 03:43:00]

I played around with this for an hour or so and it looks like you can't have a
callback-to-a-callback.

Specifically, something like this (which is quite nice!) doesn't work:

typedef base::Callback<void(base::Closure)> ClosureCB;
scoped_ptr<std::queue<ClosureCB> > closures(new std::queue<ClosureCB>());
closures->push(base::Bind(&AudioRenderer::Pause, audio_renderer_));  // ***
Compiler can't resolve templates

// later on...

void RunSerially(scoped_ptr<std::queue<ClosureCB> > closures,
                 const base::Closure& done_cb) {
  ClosureCB cb = closures->front();
  closures->pop();

  if (!closures->empty()) {
    cb.Run(base::Bind(&RunSerially, base::Passed(&closures), done_cb));
    return;
  }

  cb.Run(done_cb);
}



I hacked around a bit with a class that would let you add closures to run but
it's a bit kludgy.

ClosureRunner* crunner = new ClosureRunner();
crunner->Add(base::Bind(&AudioRenderer::Pause, audio_renderer_,
crunner->GetAddCB()));  // Yuck.


I'm not an expert here so I'll chat with ajwong@.

+  base::Closure do_pause = done_cb;
+
+  if (pipeline_filter_) {

[scherkus (not reviewing), 2012/07/10 03:58:22]

note that pipeline_filter_ ~= video_renderer_ (it's the only Filter class left!)

[acolwell GONE FROM CHROMIUM, 2012/07/10 16:34:37]

Any reason we shouldn't rename pipeline_filter_ -> video_renderer_ right now to
reflect this fact?

[scherkus (not reviewing), 2012/07/17 03:43:00]

We have an actual video_renderer_ that we use for different purposes -- I think
I'd like to leave the name unchanged for now but add a TODO comment in the .h

+    do_pause = base::Bind(&Pipeline::RunOnPipelineThread, this, do_pause);
+    do_pause = base::Bind(&Filter::Pause, pipeline_filter_, do_pause);
+  }
+
+  if (audio_renderer_) {
+    do_pause = base::Bind(&Pipeline::RunOnPipelineThread, this, do_pause);
+    do_pause = base::Bind(&AudioRenderer::Pause, audio_renderer_, do_pause);
+  }
+
+  do_pause.Run();
+}
+
+void Pipeline::DoFlush(const base::Closure& done_cb) {

[Ami GONE FROM CHROMIUM, 2012/07/10 17:16:00]

Before this CL CompositeFilter::Flush() was parallel, but this one isn't.

[scherkus (not reviewing), 2012/07/17 03:43:00]

I forget why Flush() was parallel. Possibly as a work around for some race
condition, but I'm not sure anymore.

I need to think a bit more on how I want to write this callback chaining code.

+  base::Closure do_flush = done_cb;
+
+  if (pipeline_filter_) {
+    do_flush = base::Bind(&Pipeline::RunOnPipelineThread, this, do_flush);
+    do_flush = base::Bind(&Filter::Flush, pipeline_filter_, do_flush);
+  }
+
+  if (audio_renderer_) {
+    do_flush = base::Bind(&Pipeline::RunOnPipelineThread, this, do_flush);
+    do_flush = base::Bind(&AudioRenderer::Flush, audio_renderer_, do_flush);
+  }
+
+  do_flush.Run();
+}
+
+void Pipeline::DoPlay(const base::Closure& done_cb) {
+  base::Closure do_play = done_cb;
+
+  if (pipeline_filter_) {
+    do_play = base::Bind(&Pipeline::RunOnPipelineThread, this, do_play);
+    do_play = base::Bind(&Filter::Play, pipeline_filter_, do_play);
+  }
+
+  if (audio_renderer_) {
+    do_play = base::Bind(&Pipeline::RunOnPipelineThread, this, do_play);
+    do_play = base::Bind(&AudioRenderer::Play, audio_renderer_, do_play);
+  }
+
+  do_play.Run();
+}
+
+void Pipeline::DoStop(const base::Closure& done_cb) {
+  base::Closure do_stop = done_cb;
+
+  if (pipeline_filter_) {
+    do_stop = base::Bind(&Pipeline::RunOnPipelineThread, this, do_stop);
+    do_stop = base::Bind(&Filter::Stop, pipeline_filter_, do_stop);
+  }
+
+  if (audio_renderer_) {
+    do_stop = base::Bind(&Pipeline::RunOnPipelineThread, this, do_stop);
+    do_stop = base::Bind(&AudioRenderer::Stop, audio_renderer_, do_stop);
+  }
+
+  if (demuxer_) {
+    do_stop = base::Bind(&Pipeline::RunOnPipelineThread, this, do_stop);
+    do_stop = base::Bind(&Demuxer::Stop, demuxer_, do_stop);
+  }
+
+  do_stop.Run();
+}
+
 void Pipeline::AddBufferedByteRange(int64 start, int64 end) {
   DCHECK(IsRunning());
   base::AutoLock auto_lock(lock_);
   buffered_byte_ranges_.Add(start, end);
   did_loading_progress_ = true;
 }
 
 void Pipeline::AddBufferedTimeRange(base::TimeDelta start,
                                     base::TimeDelta end) {
   DCHECK(IsRunning());
@@ skipping 11 matching lines @@
   natural_size_ = size;
 }
 
 void Pipeline::NotifyEnded() {
   DCHECK(IsRunning());
   message_loop_->PostTask(FROM_HERE, base::Bind(
       &Pipeline::NotifyEndedTask, this));
   media_log_->AddEvent(media_log_->CreateEvent(MediaLogEvent::ENDED));
 }
 
-void Pipeline::DisableAudioRenderer() {
+void Pipeline::AudioRendererEnded() {
   DCHECK(IsRunning());
+  message_loop_->PostTask(FROM_HERE, base::Bind(
+      &Pipeline::NotifyEndedTask, this));
+  media_log_->AddEvent(media_log_->CreateEvent(MediaLogEvent::ENDED));
+}
 
-  // Disable renderer on the message loop.
+void Pipeline::AudioRendererDisabled() {
+  DCHECK(IsRunning());
   message_loop_->PostTask(FROM_HERE, base::Bind(
-      &Pipeline::DisableAudioRendererTask, this));
+      &Pipeline::AudioRendererDisabledTask, this));
   media_log_->AddEvent(
       media_log_->CreateEvent(MediaLogEvent::AUDIO_RENDERER_DISABLED));
 }
 
 // Called from any thread.
 void Pipeline::OnFilterInitialize(PipelineStatus status) {
   // Continue the initialize task by proceeding to the next stage.
   message_loop_->PostTask(FROM_HERE, base::Bind(
       &Pipeline::InitializeTask, this, status));
 }
 
 // Called from any thread.
 void Pipeline::OnFilterStateTransition() {

[scherkus (not reviewing), 2012/07/10 04:02:52]

we might be to remove this trampoline since the DoXXX() methods incorporate
trampolining via RunOnPipelineThread()

hmm...

[acolwell GONE FROM CHROMIUM, 2012/07/10 16:34:37]

This is an unconditional post though. RunOnPipelineThread() isn't. I'm pretty
sure this being unconditional is key because it makes sure any SetError() calls
that happen before the callback is run get executed first.

[Ami GONE FROM CHROMIUM, 2012/07/10 17:16:00]

I agree w/ acolwell and suspect this means RunOnPipelineThread() can be dropped
entirely, because the done_cb passed to DoXXX() is invariably either
OnFilterStateTransition or OnTeardownStateTransition, each of which post
unconditionally.
In fact, instead of DoXXX() taking a mysterious black-box Closure, why not pass
a bool is_teardown and have it dispatch to OFST or OTST, as appropriate?

[scherkus (not reviewing), 2012/07/17 03:43:00]

Done.
Hmm it's funny you mention the black box closure bit. I agree it's black-boxy
when reading DoXXX() code but I found it a bit clearer when reading the callsite
as you know exactly where you end up after all is said and done (i.e.,
FilterStateTransitionTask always binds to OFST).

   message_loop_->PostTask(FROM_HERE, base::Bind(
       &Pipeline::FilterStateTransitionTask, this));
 }
 
 // Called from any thread.
 // This method makes the PipelineStatusCB behave like a Closure. It
 // makes it look like a host()->SetError() call followed by a call to
 // OnFilterStateTransition() when errors occur.
 //
 // TODO: Revisit this code when SetError() is removed from FilterHost and
@@ skipping 222 matching lines @@
 
   {
     base::AutoLock auto_lock(lock_);
     clock_->SetPlaybackRate(playback_rate);
   }
 
   // Notify |pipeline_filter_| if it has been initialized. If initialization
   // hasn't completed yet, the playback rate will be set when initialization
   // completes.
   if (pipeline_filter_) {
-    DCHECK(demuxer_);
     demuxer_->SetPlaybackRate(playback_rate);
     pipeline_filter_->SetPlaybackRate(playback_rate);
   }
+
+  if (audio_renderer_)

[acolwell GONE FROM CHROMIUM, 2012/07/10 16:34:37]

This isn't consistent with the original code. It should be inside the
pipeline_filter_ conditional since that is only set once everything has been
initialized.

[scherkus (not reviewing), 2012/07/17 03:43:00]

Done.

+    audio_renderer_->SetPlaybackRate(playback_rate_);
 }
 
 void Pipeline::VolumeChangedTask(float volume) {
   DCHECK(message_loop_->BelongsToCurrentThread());
   if (!running_ || tearing_down_)
     return;
 
   if (audio_renderer_)
     audio_renderer_->SetVolume(volume);
 }
@@ skipping 24 matching lines @@
   SetState(kPausing);
   seek_timestamp_ = time;
   seek_cb_ = seek_cb;
 
   // Kick off seeking!
   {
     base::AutoLock auto_lock(lock_);
     if (clock_->IsPlaying())
       clock_->Pause();
   }
-  pipeline_filter_->Pause(
-      base::Bind(&Pipeline::OnFilterStateTransition, this));
+  DoPause(base::Bind(&Pipeline::OnFilterStateTransition, this));
 }
 
 void Pipeline::NotifyEndedTask() {
   DCHECK(message_loop_->BelongsToCurrentThread());
 
   // We can only end if we were actually playing.
   if (state_ != kStarted) {
     return;
   }
 
@@ skipping 19 matching lines @@
   // Transition to ended, executing the callback if present.
   SetState(kEnded);
   {
     base::AutoLock auto_lock(lock_);
     clock_->EndOfStream();
   }
 
   ReportStatus(ended_cb_, status_);
 }
 
-void Pipeline::DisableAudioRendererTask() {
+void Pipeline::AudioRendererDisabledTask() {
   DCHECK(message_loop_->BelongsToCurrentThread());
 
   base::AutoLock auto_lock(lock_);
   has_audio_ = false;
   audio_disabled_ = true;
 
   // Notify our demuxer that we're no longer rendering audio.
   demuxer_->OnAudioRendererDisabled();
 
   // Start clock since there is no more audio to
@@ skipping 26 matching lines @@
   // to the next state if needed.
   SetState(FindNextState(state_));
   if (state_ == kSeeking) {
     base::AutoLock auto_lock(lock_);
     clock_->SetTime(seek_timestamp_, seek_timestamp_);
   }
 
   // Carry out the action for the current state.
   if (TransientState(state_)) {
     if (state_ == kPausing) {
-      pipeline_filter_->Pause(
-          base::Bind(&Pipeline::OnFilterStateTransition, this));
+      DoPause(base::Bind(&Pipeline::OnFilterStateTransition, this));
     } else if (state_ == kFlushing) {
-      pipeline_filter_->Flush(
-          base::Bind(&Pipeline::OnFilterStateTransition, this));
+      DoFlush(base::Bind(&Pipeline::OnFilterStateTransition, this));
     } else if (state_ == kSeeking) {
       DoSeek(seek_timestamp_);
     } else if (state_ == kStarting) {
-      pipeline_filter_->Play(
-          base::Bind(&Pipeline::OnFilterStateTransition, this));
+      DoPlay(base::Bind(&Pipeline::OnFilterStateTransition, this));
     } else if (state_ == kStopping) {
       DoStop(base::Bind(&Pipeline::OnFilterStateTransition, this));
     } else {
       NOTREACHED() << "Unexpected state: " << state_;
     }
   } else if (state_ == kStarted) {
     FinishInitialization();
 
     // Finally, complete the seek.
     seek_pending_ = false;
@@ skipping 25 matching lines @@
 
 void Pipeline::TeardownStateTransitionTask() {
   DCHECK(IsPipelineTearingDown());
   switch (state_) {
     case kStopping:
       SetState(error_caused_teardown_ ? kError : kStopped);
       FinishDestroyingFiltersTask();
       break;
     case kPausing:
       SetState(kFlushing);
-      pipeline_filter_->Flush(
-          base::Bind(&Pipeline::OnTeardownStateTransition, this));
+      DoFlush(base::Bind(&Pipeline::OnTeardownStateTransition, this));
       break;
     case kFlushing:
       SetState(kStopping);
       DoStop(base::Bind(&Pipeline::OnTeardownStateTransition, this));
       break;
 
     case kCreated:
     case kError:
     case kInitDemuxer:
     case kInitAudioDecoder:
@@ skipping 127 matching lines @@
 }
 
 bool Pipeline::InitializeAudioRenderer(
     const scoped_refptr<AudioDecoder>& decoder) {
   DCHECK(message_loop_->BelongsToCurrentThread());
   DCHECK(IsPipelineOk());
 
   if (!decoder)
     return false;
 
-  filter_collection_->SelectAudioRenderer(
-      &pipeline_init_state_->audio_renderer);
-  if (!pipeline_init_state_->audio_renderer) {
+  filter_collection_->SelectAudioRenderer(&audio_renderer_);
+  if (!audio_renderer_) {
     SetError(PIPELINE_ERROR_REQUIRED_FILTER_MISSING);
     return false;
   }
 
-  pipeline_init_state_->composite->AddFilter(
-      pipeline_init_state_->audio_renderer);
-
-  pipeline_init_state_->audio_renderer->Initialize(
+  audio_renderer_->SetHost(this);

[scherkus (not reviewing), 2012/07/10 04:02:52]

I think SetHost() can go away after we replace ARH with individual CBs passed
into Initialize() (see TODO by ARH definition)

[acolwell GONE FROM CHROMIUM, 2012/07/10 16:34:37]

I think we should just add the callbacks now. I can't think of any good reason
to introduce the ARH class as an intermediate step.

[scherkus (not reviewing), 2012/07/17 03:43:00]

Done.

+  audio_renderer_->Initialize(
       decoder,
       base::Bind(&Pipeline::OnFilterInitialize, this),
       base::Bind(&Pipeline::OnAudioUnderflow, this),
       base::Bind(&Pipeline::OnAudioTimeUpdate, this));
-
-  audio_renderer_ = pipeline_init_state_->audio_renderer;
   return true;
 }
 
 bool Pipeline::InitializeVideoRenderer(
     const scoped_refptr<VideoDecoder>& decoder) {
   DCHECK(message_loop_->BelongsToCurrentThread());
   DCHECK(IsPipelineOk());
 
   if (!decoder)
     return false;
@@ skipping 65 matching lines @@
       if (seek_pending_) {
         seek_pending_ = false;
         FinishInitialization();
       }
 
       break;
 
     case kStarted:
     case kEnded:
       SetState(kPausing);
-      pipeline_filter_->Pause(
-          base::Bind(&Pipeline::OnTeardownStateTransition, this));
+      DoPause(base::Bind(&Pipeline::OnTeardownStateTransition, this));
       break;
 
     case kStopping:
     case kStopped:
       NOTREACHED() << "Unexpected state for teardown: " << state_;
       break;
     // default: intentionally left out to force new states to cause compiler
     // errors.
   };
 }
 
-void Pipeline::DoStop(const base::Closure& callback) {
-  if (demuxer_) {
-    demuxer_->Stop(base::Bind(
-        &Pipeline::OnDemuxerStopDone, this, callback));
-    return;
-  }
-
-  OnDemuxerStopDone(callback);
-}
-
-void Pipeline::OnDemuxerStopDone(const base::Closure& callback) {
-  if (!message_loop_->BelongsToCurrentThread()) {
-    message_loop_->PostTask(FROM_HERE, base::Bind(
-        &Pipeline::OnDemuxerStopDone, this, callback));
-    return;
-  }
-
-  if (pipeline_filter_) {
-    pipeline_filter_->Stop(callback);
-    return;
-  }
-
-  callback.Run();
-}
-
 void Pipeline::DoSeek(TimeDelta seek_timestamp) {
   // TODO(acolwell): We might be able to convert this if (demuxer_) into a
   // DCHECK(). Further investigation is needed to make sure this won't introduce
   // a bug.

[scherkus (not reviewing), 2012/07/10 03:58:22]

FYI I think it's impossible for demuxer_ to be NULL as we always have a demuxer_
if we're going to be doing any seeking (i.e., initialization was successful)

time to remove the checks here?

[acolwell GONE FROM CHROMIUM, 2012/07/10 16:34:37]

Go for it. We'll see if it goes boom.

[scherkus (not reviewing), 2012/07/17 03:43:00]

Committed separately as r146024.

   if (demuxer_) {
     demuxer_->Seek(seek_timestamp, base::Bind(
         &Pipeline::OnDemuxerSeekDone, this, seek_timestamp));
     return;
   }
 
   OnDemuxerSeekDone(seek_timestamp, PIPELINE_OK);
 }
 
 void Pipeline::OnDemuxerSeekDone(TimeDelta seek_timestamp,
                                  PipelineStatus status) {
   if (!message_loop_->BelongsToCurrentThread()) {
     message_loop_->PostTask(FROM_HERE, base::Bind(
         &Pipeline::OnDemuxerSeekDone, this, seek_timestamp, status));
     return;
   }
 
+  if (status == PIPELINE_OK && audio_renderer_) {
+    audio_renderer_->Seek(seek_timestamp, base::Bind(
+        &Pipeline::OnAudioRendererSeekDone, this, seek_timestamp));
+    return;
+  }
+
+  OnAudioRendererSeekDone(seek_timestamp, status);
+}
+
+void Pipeline::OnAudioRendererSeekDone(base::TimeDelta seek_timestamp,
+                                       PipelineStatus status) {
+  if (!message_loop_->BelongsToCurrentThread()) {
+    message_loop_->PostTask(FROM_HERE, base::Bind(
+        &Pipeline::OnAudioRendererSeekDone, this, seek_timestamp, status));
+    return;
+  }
+
   PipelineStatusCB done_cb =
       base::Bind(&Pipeline::OnFilterStateTransitionWithStatus, this);
 
   if (status == PIPELINE_OK && pipeline_filter_) {
     pipeline_filter_->Seek(seek_timestamp, done_cb);
     return;
   }
 
   ReportStatus(done_cb, status);
 }
@@ skipping 14 matching lines @@
 
 void Pipeline::StartClockIfWaitingForTimeUpdate_Locked() {
   lock_.AssertAcquired();
   if (!waiting_for_clock_update_)
     return;
 
   waiting_for_clock_update_ = false;
   clock_->Play();
 }
 
+void Pipeline::RunOnPipelineThread(const base::Closure& closure) {
+  if (!message_loop_->BelongsToCurrentThread()) {
+    message_loop_->PostTask(FROM_HERE, base::Bind(
+        &Pipeline::RunOnPipelineThread, this, closure));
+    return;
+  }
+  closure.Run();
+}
+
 }  // namespace media