Motown in-proc streaming framework used to implement media services.

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::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::Remove(Part part) {
+  DCHECK(part);
+  base::AutoLock lock(lock_);
+  RemoveUnsafe(part.stage_);
+}
+
+Engine::Part Engine::Connect(Output output, Input input) {
+  DCHECK(output);
+  DCHECK(input);
+
+  base::AutoLock lock(lock_);
+
+  if (output.connected()) {
+    DisconnectOutputUnsafe(output.stage_, output.index_);
+  }
+  if (input.connected()) {
+    DisconnectInputUnsafe(input.stage_, input.index_);
+  }
+
+  output.stage_output().connect(
+      input.stage_,
+      input.index_);
+
+  input.stage_input().connect(
+      output.stage_,
+      output.index_);
+
+  return input.part();
+}
+
+Engine::Part Engine::Connect(Part upstream_part, Part downstream_part) {
+  DCHECK(upstream_part);
+  DCHECK(downstream_part);
+  Connect(upstream_part.output(), downstream_part.input());
+  return downstream_part;
+}
+
+Engine::Part Engine::Connect(
+    Output output,
+    Part downstream_part) {
+  DCHECK(output);
+  DCHECK(downstream_part);
+  Connect(output, downstream_part.input());
+  return downstream_part;
+}
+
+Engine::Part Engine::Connect(Part upstream_part, Input input) {
+  DCHECK(upstream_part);
+  DCHECK(input);
+  Connect(upstream_part.output(), input);
+  return input.part();
+}
+
+void Engine::Disconnect(Output output) {
+  DCHECK(output);
+
+  base::AutoLock lock(lock_);
+  DisconnectOutputUnsafe(output.stage_, output.index_);
+}
+
+void Engine::Disconnect(Input input) {
+  DCHECK(input);
+
+  base::AutoLock lock(lock_);
+  DisconnectInputUnsafe(input.stage_, input.index_);
+}
+
+void Engine::RemoveAll(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_);
+  }
+}
+
+void Engine::RemoveAll(Output output) {
+  DCHECK(output);
+
+  if (!output.connected()) {
+    return;
+  }
+
+  Part downstream_part = output.downstream_part();
+  Disconnect(output);
+  RemoveAll(downstream_part);
+}
+
+void Engine::RemoveAll(Input input) {
+  DCHECK(input);
+
+  if (!input.connected()) {
+    return;
+  }
+
+  Part upstream_part = input.upstream_part();
+  Disconnect(input);
+  RemoveAll(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++) {
+      PrepareUnsafe(sink->input(input_index).upstream_stage());
+    }
+  }
+}
+
+void Engine::Prepare(Part part) {
+  DCHECK(part);
+  base::AutoLock lock(lock_);
+  PrepareUnsafe(part.stage_);
+}
+
+void Engine::PrimeSinks() {
+  lock_.Acquire();
+  std::list<Stage*> sinks(sinks_);
+  lock_.Release();

[johngro, 2016/01/28 19:14:55]

so, you made a copy of the sinks_ list which is holding raw pointers to the
stages which are classified as sinks, then you release the lock and call prime
on each sink.

What prevents another thread from calling Remove on a stage which exists in the
sink collection (or from Resetting the entire Engine) and deleting the sink
stage out from under you while you are doing this?

[dalesat, 2016/01/29 01:08:29]

At this point, we are depending on the kindness of the caller. Adding a TODO.

[johngro, 2016/02/01 22:38:17]

Acknowledged.

+
+  for (Stage* sink : sinks) {
+    sink->Prime();
+  }
+}
+
+void Engine::Reset() {
+  base::AutoLock lock(lock_);
+  supply_backlog_.clear();
+  demand_backlog_.clear();
+  sources_.clear();
+  sinks_.clear();
+  while (!stages_.empty()) {
+    Stage* stage = stages_.front();
+    stages_.pop_front();
+    delete stage;
+  }
+}
+
+void Engine::PushToSupplyBacklogUnsafe(Stage* stage) {
+  lock_.AssertAcquired();
+
+  DCHECK(stage);
+  packets_produced_ = true;
+  if (!stage->in_supply_backlog_) {
+    // LIFO

[johngro, 2016/01/27 22:35:22]

These comments still seem backwards.  Both Pop implementation Pop from the front
of the deque.  PushToSupply's comment says it is LIFO, but it is pushing to the
back and pulling from the front (which is FIFO), and vice-versa for PushToDemand
(push and pull to front is LIFO, comment says FIFO).

Either the comment is wrong or the code is wrong here.  Also, while it is great
to have the comments (FIFO vs. LIFO) they are a a little vague.  They merely
assert whether these data structures are stacks or queues, but provide no
explanation as to why the FIFO vs. LIFO-ness of the structures are important. 
This means I can notice the mismatch, but I cannot determine who is correct (the
code or the comment).  Please update the comment (or add one elsewhere in some
more top-level theory-of-operation comment) which explains the reasoning behind
the FIFO/LIFO decisions.

Finally, its pretty easy to enforce this in code so that no mistakes can be
made.  Take a look at std::stack and std::queue.  They are templateized adapters
for std:: containers which support push/pop front/back (vector, deque, list;
default it deque) which enforce LIFO and FIFO behavior respectively via their
exposed API.  This is a much stronger assertion than just having the comment,
but an explanation for why to use one vs. the other somewhere in the code would
still be appreciated.

[dalesat, 2016/01/28 18:49:17]

Thanks for the tip on queue/stack. They don't have clear(), but are otherwise
just the ticket.

I've added a comment as to why these are queue and stack - it basically says
that I guessed. There's also a TODO to determine what ordering is optimal and
implement that. It's possible that it doesn't matter. I also see a potential
issue with having two backlogs. One may be adequate.

[johngro, 2016/02/01 22:38:17]

Acknowledged.

Re: ordering, this basically boils down to a priority issue (which work is the
most important?)  I'm sure we will figure it out moving forward.  If we want to
move to a 1-backlog model, it may be appropriate to have a priority for each job
and stuff them into a multiset (or set, if there is a good way to unique-ify the
priorities).  Time will tell..

Re: clear() missing; how odd.  Perhaps its because they didn't want to
significantly increase the complexity of the requirements for the underlying
container?  That said, C++11 supplies a constant order swap method.  I'm not
sure how you implement such a thing without requiring that the underlying
container has a constant order swap and/or assignment operator (which they do
not require).  That said, the most efficient way to implement clear for one of
these is probably...

using MyQueue = std::queue<MyStuff, MyContainer<MyStuff>>;
{
  MyQueue tmp;
  clear_this_queue_.swap(tmp);
}  // tmp goes out of scope, clearing the underlying container in the process.

[dalesat, 2016/02/01 23:01:28]

Nice trick, thanks!

+    supply_backlog_.push_back(stage);
+    stage->in_supply_backlog_ = true;
+  }
+}
+
+void Engine::PushToDemandBacklogUnsafe(Stage* stage) {
+  lock_.AssertAcquired();
+
+  DCHECK(stage);
+  if (!stage->in_demand_backlog_) {
+    // FIFO
+    demand_backlog_.push_front(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().connect(nullptr, 0);
+  stage_output.connect(nullptr, 0);
+}
+
+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().connect(nullptr, 0);
+  stage_input.connect(nullptr, 0);
+}
+
+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::PrepareUnsafe(Stage* stage) {
+  lock_.AssertAcquired();
+
+  if (stage == nullptr || stage->prepared_) {
+    return;
+  }
+
+  stage->Prepare(update_function_);

[johngro, 2016/01/28 19:14:55]

Following comment applies to UpdateUnsafe as well

So, at this point you are holding your master lock, and calling into another
function implemented by the stage (which will most likely execute logic
implemented by the part contained within the stage).

This raises a number of concerns.
How does the stage/part call back into the engine to queue new work and not
re-enter the lock.

If the stage/part acquires an internal lock, what it to prevent an A/B, B/A lock
ordering issue in the interaction between the stage/part and the engine.
IOW, There seems to be a significant danger that we engine.lock then part.lock
during update, and something asynchronous happens inside of the part (interrupt
fires or something) and the part code attempts to part.lock then
engine.do_something_which_requires_the_lock, resulting in a deadlock hazard.

I'm not going to be dogmatic and say that locks should never be held when
calling other functions, but if we are going to do so, then we need to be very
certain of our assumptions about things like lock domains, lock ordering and
exactly what the other code will do while holding our lock.  Whenever possible,
we should do our best to break down the scope of where locks are held, both to
increase parallelism and decrease the risk of deadlock.

This particular pattern of holding a master engine lock while calling out into
the modular stage/part world makes me very nervous.  Is there any way that we
can decompose this so that we do not have to hold the lock while we call into
what amounts to external code?

[dalesat, 2016/01/29 01:08:30]

The threading model will change dramatically as time goes on. This CL is not the
place to redesign it. I've added a lengthy comment to engine.h that talks about
the current constraints. It will be fine for now.

In general, let's keep comments to the effect of "You haven't done everything
that will ever need to be done concerning issue X in this CL" to a minimum. We
should make sure issues are captured and move ahead.

[johngro, 2016/02/01 22:38:17]

Acknowledged.

+  stage->prepared_ = true;
+
+  uint32_t input_count = stage->input_count();
+  for (uint32_t input_index = 0; input_index < input_count; input_index++) {
+    PrepareUnsafe(stage->input(input_index).upstream_stage());
+  }
+}
+
+void Engine::UpdateUnsafe() {
+  lock_.AssertAcquired();
+
+  while (true) {
+    Stage* stage = PopFromSupplyBacklogUnsafe();
+    if (stage != nullptr) {
+      UpdateUnsafe(stage);
+      continue;
+    }
+
+    stage = PopFromDemandBacklogUnsafe();
+    if (stage != nullptr) {
+      UpdateUnsafe(stage);
+      continue;
+    }
+
+    break;
+  }
+}
+
+void Engine::UpdateUnsafe(Stage *stage) {
+  lock_.AssertAcquired();
+
+  DCHECK(stage);
+
+  packets_produced_ = false;
+
+  stage->Update(this);
+
+  // If the stage produced packets, it may need to reevaluate demand later.
+  if (packets_produced_) {
+    PushToDemandBacklogUnsafe(stage);
+  }
+}
+
+Stage* Engine::PopFromSupplyBacklogUnsafe() {
+  lock_.AssertAcquired();
+
+  if (supply_backlog_.empty()) {
+    return nullptr;
+  }
+
+  Stage* stage = supply_backlog_.front();
+  supply_backlog_.pop_front();
+  DCHECK(stage->in_supply_backlog_);
+  stage->in_supply_backlog_ = false;
+  return stage;
+}
+
+Stage* Engine::PopFromDemandBacklogUnsafe() {
+  lock_.AssertAcquired();
+
+  if (demand_backlog_.empty()) {
+    return nullptr;
+  }
+
+  Stage* stage = demand_backlog_.front();
+  demand_backlog_.pop_front();
+  DCHECK(stage->in_demand_backlog_);
+  stage->in_demand_backlog_ = false;
+  return stage;
+}
+
+}  // namespace media
+}  // namespace mojo