Motown: Remove LPCM optimizations, fix prepare, add flush, add ActiveMultistreamSink model/stage

services/media/framework/engine.cc

 // Copyright 2016 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 "services/media/framework/engine.h"
 
 namespace mojo {
 namespace media {
 
-uint32_t Engine::Part::input_count() {
-  DCHECK(stage_ != nullptr);
-  return stage_->input_count();
+Engine::Engine() {}
+
+Engine::~Engine() {
+  base::AutoLock lock(lock_);
 }
 
-Engine::Input Engine::Part::input(uint32_t index) {
-  DCHECK(stage_ != nullptr && index < stage_->input_count());
-  return Input(stage_, index);
-}
-
-Engine::Input Engine::Part::input() {
-  DCHECK(stage_ != nullptr && stage_->input_count() == 1);
-  return Input(stage_, 0);
-}
-
-uint32_t Engine::Part::output_count() {
-  DCHECK(stage_ != nullptr);
-  return stage_->output_count();
-}
-
-Engine::Output Engine::Part::output(uint32_t index) {
-  DCHECK(stage_ != nullptr && index < stage_->output_count());
-  return Output(stage_, index);
-}
-
-Engine::Output Engine::Part::output() {
-  DCHECK(stage_ != nullptr && stage_->output_count() == 1);
-  return Output(stage_, 0);
-}
-
-Engine::Part Engine::Part::upstream_part(uint32_t index) {
-  DCHECK(stage_ != nullptr && index < stage_->input_count());
-  return Part(stage_->input(index).upstream_stage());
-}
-
-Engine::Part Engine::Part::upstream_part() {
-  DCHECK(stage_ != nullptr && stage_->input_count() == 1);
-  return Part(stage_->input(0).upstream_stage());
-}
-
-Engine::Part Engine::Part::downstream_part(uint32_t index) {
-  DCHECK(stage_ != nullptr && index < stage_->output_count());
-  return Part(stage_->output(index).downstream_stage());
-}
-
-Engine::Part Engine::Part::downstream_part() {
-  DCHECK(stage_ != nullptr && stage_->output_count() == 1);
-  return Part(stage_->output(0).downstream_stage());
-}
-
-Engine::Engine() {
-  update_function_ = [this](Stage* stage) {
-    DCHECK(stage);
-    base::AutoLock lock(lock_);
-    UpdateUnsafe(stage);
-    UpdateUnsafe();
-  };
-}
-
-Engine::~Engine() {
-  Reset();
-}
-
-void Engine::RemovePart(Part part) {
-  DCHECK(part);
-  base::AutoLock lock(lock_);
-  RemoveUnsafe(part.stage_);
-}
-
-Engine::Part Engine::Connect(Output output, Input input) {
-  DCHECK(output);
-  DCHECK(input);
-
+void Engine::PrepareInput(const InputRef& input_ref) {
   base::AutoLock lock(lock_);
 
-  if (output.connected()) {
-    DisconnectOutputUnsafe(output.stage_, output.index_);
-  }
-  if (input.connected()) {
-    DisconnectInputUnsafe(input.stage_, input.index_);
-  }
+  std::queue<InputRef> backlog;
+  backlog.push(input_ref);
 
-  output.stage_output().connect(input.stage_, input.index_);
-  input.stage_input().connect(output.stage_, output.index_);
+  while (!backlog.empty()) {
+    InputRef input = backlog.front();
+    backlog.pop();
+    DCHECK(input.valid());
+    DCHECK(input.connected());
+    DCHECK(!input.actual().prepared());
 
-  return input.part();
-}
+    PayloadAllocator* allocator = input.stage_->PrepareInput(input.index_);
+    input.actual().set_prepared(true);
 
-Engine::Part Engine::ConnectParts(Part upstream_part, Part downstream_part) {
-  DCHECK(upstream_part);
-  DCHECK(downstream_part);
-  Connect(upstream_part.output(), downstream_part.input());
-  return downstream_part;
-}
+    const OutputRef& output = input.mate();
 
-Engine::Part Engine::ConnectOutputToPart(
-    Output output,
-    Part downstream_part) {
-  DCHECK(output);
-  DCHECK(downstream_part);
-  Connect(output, downstream_part.input());
-  return downstream_part;
-}
-
-Engine::Part Engine::ConnectPartToInput(Part upstream_part, Input input) {
-  DCHECK(upstream_part);
-  DCHECK(input);
-  Connect(upstream_part.output(), input);
-  return input.part();
-}
-
-void Engine::DisconnectOutput(Output output) {
-  DCHECK(output);
-
-  base::AutoLock lock(lock_);
-  DisconnectOutputUnsafe(output.stage_, output.index_);
-}
-
-void Engine::DisconnectInput(Input input) {
-  DCHECK(input);
-
-  base::AutoLock lock(lock_);
-  DisconnectInputUnsafe(input.stage_, input.index_);
-}
-
-void Engine::RemovePartsConnectedToPart(Part part) {
-  DCHECK(part);
-
-  base::AutoLock lock(lock_);
-
-  std::deque<Part> to_remove { part };
-
-  while (!to_remove.empty()) {
-    Part part = to_remove.front();
-    to_remove.pop_front();
-
-    for (uint32_t i = 0; i < part.input_count(); ++i) {
-      to_remove.push_back(part.upstream_part(i));
-    }
-
-    for (uint32_t i = 0; i < part.output_count(); ++i) {
-      to_remove.push_back(part.downstream_part(i));
-    }
-
-    RemoveUnsafe(part.stage_);
+    output.stage_->PrepareOutput(
+        output.index_,
+        allocator,
+        [this, &output, &backlog](size_t input_index) {

[johngro, 2016/02/08 22:33:36]

I don't think that you are using your captured 'this' pointer in this lambda,
you can probably drop it.

Also, the only use of the output reference seems to be to get access to
output.stage_ in order to make a new input ref to put on the backlog.  It would
be nice if you could make the reference capture const, but you cannot (at least,
not in C++11).

Since stage_ is a raw pointer, you can avoid the need to for the constant-ness
of the capture, but would need to introduce a temp variable... something like

Stage* output_stage = output.stage_;
PrepareOutput(...,
  [output_stage, &backlog)(...) { ... }
);

Probably not a lot of win to doing this here, but if you start to de-friend the
(Input|Output|Part)Ref objects to remove read-write access to private members,
it would allow you to say...

Stage* output_stage = output.stage();
// do capture here

[dalesat, 2016/02/09 00:34:11]

I've changed this a bit to use a standard visitor. This particular optimization
(and dropping the capture of this) is addressed.

+          backlog.push(InputRef(output.stage_, input_index));
+        });
   }
 }
 
-void Engine::RemovePartsConnectedToOutput(Output output) {
-  DCHECK(output);
+void Engine::UnprepareInput(const InputRef& input_ref) {
+  base::AutoLock lock(lock_);
 
-  if (!output.connected()) {
-    return;
-  }
+  std::queue<InputRef> backlog;

[johngro, 2016/02/08 22:33:36]

backlog seems to be essentially unused here.  In PrepareInput, it was used to
implement a breadth-first Prepare from the starting input stage all of the way
up the graph.

Was the intention here to replicate the all-the-way-up-the-graph behavior (in
which case I think that the backlog needs to be provided to the UnprepareOutput
call) or is the intent to simply un-prepare a single input/output link (in which
case, backlog is not needed and should be removed).

Finally, if this is the latter, then the names of these methods imply a symmetry
which the actions do not reflect.  Names, comments, or both should probably be
updated to indicate the graph-wide behavior of Prepare and the link-only
behavior of Unprepare.

[dalesat, 2016/02/09 00:34:11]

Done.

+  backlog.push(input_ref);
 
-  Part downstream_part = output.downstream_part();
-  DisconnectOutput(output);
-  RemovePartsConnectedToPart(downstream_part);
-}
+  while (!backlog.empty()) {
+    InputRef input = backlog.front();
+    backlog.pop();
+    DCHECK(input.valid());
+    DCHECK(input.connected());
+    DCHECK(input.actual().prepared());
 
-void Engine::RemovePartsConnectedToInput(Input input) {
-  DCHECK(input);
+    input.stage_->UnprepareInput(input.index_);
+    input.actual().set_prepared(false);
 
-  if (!input.connected()) {
-    return;
-  }
-
-  Part upstream_part = input.upstream_part();
-  DisconnectInput(input);
-  RemovePartsConnectedToPart(upstream_part);
-}
-
-void Engine::Prepare() {
-  base::AutoLock lock(lock_);
-  for (Stage* sink : sinks_) {
-    sink->Prepare(update_function_);
-    sink->prepared_ = true;
-    uint32_t input_count = sink->input_count();
-    for (uint32_t input_index = 0; input_index < input_count; input_index++) {
-      MaybePrepareUnsafe(sink->input(input_index).upstream_stage());
-    }
+    input.mate().stage_->UnprepareOutput(input.mate().index_);
   }
 }
 
-void Engine::Prepare(Part part) {
-  DCHECK(part);
+void Engine::FlushOutput(const OutputRef& output_ref) {
   base::AutoLock lock(lock_);
-  MaybePrepareUnsafe(part.stage_);
-}
 
-void Engine::PrimeSinks() {
-  lock_.Acquire();
-  std::list<Stage*> sinks(sinks_);
-  lock_.Release();
+  std::queue<OutputRef> backlog;
+  backlog.push(output_ref);
 
-  // TODO(dalesat): Threading issue: these sinks may go away during priming.
-  for (Stage* sink : sinks) {
-    sink->Prime();
+  while (!backlog.empty()) {
+    OutputRef output = backlog.front();
+    backlog.pop();
+    DCHECK(output.valid());
+    DCHECK(output.connected());
+    DCHECK(output.mate().actual().prepared());
+
+    output.stage_->FlushOutput(output.index_);
+
+    const InputRef& input = output.mate();
+
+    input.stage_->FlushInput(
+        input.index_,
+        [this, &input, &backlog](size_t output_index) {

[johngro, 2016/02/08 22:33:36]

same comment as before.  'this' is unused and it is possible to avoid passing
input by reference if you wanted to.

+          backlog.push(OutputRef(input.stage_, output_index));
+        });
   }
 }
 
-void Engine::Reset() {
+void Engine::RequestUpdate(Stage* stage) {
+  DCHECK(stage);
   base::AutoLock lock(lock_);
-  while (!supply_backlog_.empty()) {
-    supply_backlog_.pop();
-  }
-  while (!demand_backlog_.empty()) {
-    demand_backlog_.pop();
-  }
-  sources_.clear();
-  sinks_.clear();
-  while (!stages_.empty()) {
-    Stage* stage = stages_.front();
-    stages_.pop_front();
-    delete stage;
-  }
+  Update(stage);
+  Update();
 }
 
-void Engine::PushToSupplyBacklogUnsafe(Stage* stage) {
+void Engine::PushToSupplyBacklog(Stage* stage) {
   lock_.AssertAcquired();
+  DCHECK(stage);
 
-  DCHECK(stage);
   packets_produced_ = true;
   if (!stage->in_supply_backlog_) {
     supply_backlog_.push(stage);
     stage->in_supply_backlog_ = true;
   }
 }
 
-void Engine::PushToDemandBacklogUnsafe(Stage* stage) {
+void Engine::PushToDemandBacklog(Stage* stage) {
   lock_.AssertAcquired();
+  DCHECK(stage);
 
-  DCHECK(stage);
   if (!stage->in_demand_backlog_) {
     demand_backlog_.push(stage);
     stage->in_demand_backlog_ = true;
   }
 }
 
-Engine::Part Engine::Add(Stage* stage) {
-  base::AutoLock lock(lock_);
-
-  stages_.push_back(stage);
-  if (stage->input_count() == 0) {
-    sources_.push_back(stage);
-  }
-  if (stage->output_count() == 0) {
-    sinks_.push_back(stage);
-  }
-  return Part(stage);
-}
-
-void Engine::DisconnectOutputUnsafe(Stage* stage, uint32_t index) {
-  DCHECK(stage);
-  DCHECK(index < stage->output_count());
-
-  lock_.AssertAcquired();
-
-  StageOutput& stage_output = stage->output(index);
-
-  if (stage_output.downstream_stage() == nullptr) {
-    return;
-  }
-
-  stage_output.mate().disconnect();
-  stage_output.disconnect();
-}
-
-void Engine::DisconnectInputUnsafe(Stage* stage, uint32_t index) {
-  DCHECK(stage);
-  DCHECK(index < stage->input_count());
-
-  lock_.AssertAcquired();
-
-  StageInput& stage_input = stage->input(index);
-
-  if (stage_input.upstream_stage() == nullptr) {
-    return;
-  }
-
-  stage_input.mate().disconnect();
-  stage_input.disconnect();
-}
-
-void Engine::RemoveUnsafe(Stage* stage) {
-  DCHECK(stage);
-
-  lock_.AssertAcquired();
-
-  uint32_t input_count = stage->input_count();
-  for (uint32_t input_index = 0; input_index < input_count; input_index++) {
-    if (stage->input(input_index).connected()) {
-      DisconnectInputUnsafe(stage, input_index);
-    }
-  }
-
-  uint32_t output_count = stage->output_count();
-  for (uint32_t output_index = 0; output_index < output_count; output_index++) {
-    if (stage->output(output_index).connected()) {
-      DisconnectOutputUnsafe(stage, output_index);
-    }
-  }
-
-  sources_.remove(stage);
-  sinks_.remove(stage);
-  stages_.remove(stage);
-  delete stage;
-}
-
-// static
-Stage* Engine::CreateStage(MultiStreamPacketSourcePtr source) {
-  return new DistributorStage(source);
-}
-
-// static
-Stage* Engine::CreateStage(PacketTransformPtr transform) {
-  return new PacketTransformStage(transform);
-}
-
-// static
-Stage* Engine::CreateStage(ActiveSourcePtr source) {
-  return new ActiveSourceStage(source);
-}
-
-// static
-Stage* Engine::CreateStage(ActiveSinkPtr sink) {
-  return new ActiveSinkStage(sink);
-}
-
-// static
-Stage* Engine::CreateStage(LpcmTransformPtr transform) {
-  return new LpcmTransformStage(transform);
-}
-
-void Engine::MaybePrepareUnsafe(Stage* stage) {
-  lock_.AssertAcquired();
-
-  if (stage == nullptr || stage->prepared_) {
-    return;
-  }
-
-  // Make sure all downstream stages have been prepared.
-  uint32_t output_count = stage->output_count();
-  for (uint32_t output_index = 0; output_index < output_count; output_index++) {
-    StageOutput& output = stage->output(output_index);
-    if (output.connected() && !output.downstream_stage()->prepared()) {
-      return;
-    }
-  }
-
-  stage->Prepare(update_function_);
-  stage->prepared_ = true;
-
-  // Prepare all upstream stages.
-  uint32_t input_count = stage->input_count();
-  for (uint32_t input_index = 0; input_index < input_count; input_index++) {
-    MaybePrepareUnsafe(stage->input(input_index).upstream_stage());
-  }
-}
-
-void Engine::UpdateUnsafe() {
+void Engine::Update() {
   lock_.AssertAcquired();
 
   while (true) {
-    Stage* stage = PopFromSupplyBacklogUnsafe();
+    Stage* stage = PopFromSupplyBacklog();
     if (stage != nullptr) {
-      UpdateUnsafe(stage);
+      Update(stage);
       continue;
     }
 
-    stage = PopFromDemandBacklogUnsafe();
+    stage = PopFromDemandBacklog();
     if (stage != nullptr) {
-      UpdateUnsafe(stage);
+      Update(stage);
       continue;
     }
 
     break;
   }
 }
 
-void Engine::UpdateUnsafe(Stage *stage) {
+void Engine::Update(Stage *stage) {
   lock_.AssertAcquired();
 
   DCHECK(stage);
 
   packets_produced_ = false;

[johngro, 2016/02/08 22:33:36]

the stateful nature of the packets_produced_ flag makes me a bit uneasy...

Is it correct to say that we know if a stage produced one or more packets
because the supply backlog grew in size during the update?  If so, can we
eliminate the packets_produced_ member and just say...

size_t old_supply_backlog_size = supply_backlog_.size();
stage->Update(this);
if (supply_backlog_.size() > old_supply_backlog_size) {
  ...
}

[dalesat, 2016/02/09 00:34:11]

The supply backlog doesn't necessarily grow, because we won't a stage if it's
already in the backlog. I agree this statefulness isn't great, but it's safe. My
intention is to get rid of it. I think the backlogs (or just one backlog) can be
on the stack as they are for Prepare (and should be for Unprepare).
packets_produced_ will either go away or it will exist within the scope of the
Update method.

[johngro, 2016/02/09 21:00:36]

Acknowledged.

 
   stage->Update(this);
 
   // If the stage produced packets, it may need to reevaluate demand later.
   if (packets_produced_) {
-    PushToDemandBacklogUnsafe(stage);
+    PushToDemandBacklog(stage);
   }
 }
 
-Stage* Engine::PopFromSupplyBacklogUnsafe() {
+Stage* Engine::PopFromSupplyBacklog() {
   lock_.AssertAcquired();
 
   if (supply_backlog_.empty()) {
     return nullptr;
   }
 
   Stage* stage = supply_backlog_.front();
   supply_backlog_.pop();
   DCHECK(stage->in_supply_backlog_);
   stage->in_supply_backlog_ = false;
   return stage;
 }
 
-Stage* Engine::PopFromDemandBacklogUnsafe() {
+Stage* Engine::PopFromDemandBacklog() {
   lock_.AssertAcquired();
 
   if (demand_backlog_.empty()) {
     return nullptr;
   }
 
   Stage* stage = demand_backlog_.top();
   demand_backlog_.pop();
   DCHECK(stage->in_demand_backlog_);
   stage->in_demand_backlog_ = false;
   return stage;
 }
 
 }  // namespace media
 }  // namespace mojo