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() {

[johngro, 2016/01/20 00:25:32]

Would it be a good idea to inline most of these accessors?

[dalesat, 2016/01/25 23:29:37]

These are only used for graph building, so they aren't called at high frequency.
The memory penalty of inline is probably not justified for these.

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

Acknowledged.

+  DCHECK(stage_ != nullptr);
+  return stage_->input_count();
+}
+
+Engine::Input Engine::Part::input(uint32_t index) {
+  DCHECK(stage_ != nullptr && index < stage_->input_count());

[johngro, 2016/01/20 00:25:32]

FWIW, the 2 parameter form of the Input constructor already DCHECKs this (same
for output below)

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

Acknowledged.

+  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) {
+    // TODO(dalesat): Correct thread and synchronization.
+    DCHECK(stage);
+    Update(stage);
+    Update();
+  };
+}
+
+Engine::~Engine() {}
+
+void Engine::Remove(Part part) {
+  DCHECK(part);
+
+  Stage* stage = part.stage_;
+
+  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()) {
+      Disconnect(Input(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()) {
+      Disconnect(Output(stage, output_index));
+    }
+  }
+
+  sources_.remove(stage);
+  sinks_.remove(stage);
+
+  std::unique_ptr<Stage> to_remove(stage); // Release this duplicate unique_ptr!

[johngro, 2016/01/20 00:25:32]

something feel wrong about this...

if stages_ is a list of unique ptrs, then you should not be able to be holding a
pointer to anything that the unique pointer points to.  Other objects could hold
a const std::unique_ptr<>&, provided you are sure that the other objects will
have a shorter life than the unique pointer itself, but (in theory) you should
never be manipulating the naked pointer at all, since it violates the
unique-ness of the pointer and the lifetime management safety that unique_ptr is
supposed to provide.

the fact that the same stage pointer is a member of the sources_ and sinks_
collection while also being part of the stages_ collection seems to highlight
this point.  stages_ is a collection of unique pointers while sources_ and
sinks_ are naked pointer which can point to the same thing as the pointers in
stages_.  IOW - the pointers held in the stages_ collection seem to be far from
unique.

[dalesat, 2016/01/25 23:29:37]

A good analogy is that engine is a container like std::list, Stage is like a
list node, and Part, Input and Output are like iterators. I used unique_ptr as
an expedient so the stages get deleted automatically. I understand this flies in
the face of unique_ptr dogma, so I'll just use raw pointers.

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

Acknowledged.

+  stages_.remove(to_remove);
+  to_remove.release(); // OK, it's released!
+}
+
+Engine::Part Engine::Connect(
+    Output output,
+    Input input) {
+  DCHECK(output);
+  DCHECK(input);
+
+  if (output.connected()) {
+    Disconnect(output);
+  }
+  if (input.connected()) {
+    Disconnect(input);
+  }
+
+  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);
+
+  if (!output.connected()) {
+    return;
+  }
+
+  StageOutput& stage_output = output.stage_output();
+  stage_output.mate().connect(nullptr, 0);
+  stage_output.connect(nullptr, 0);
+}
+
+void Engine::Disconnect(Input input) {
+  DCHECK(input);
+
+  if (!input.connected()) {
+    return;
+  }
+
+  StageInput& stage_input = input.stage_input();
+  stage_input.mate().connect(nullptr, 0);
+  stage_input.connect(nullptr, 0);
+}
+
+void Engine::RemoveAll(Part part) {
+  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));
+    }
+
+    Remove(part);
+  }
+}
+
+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() {
+  for (Stage* sink : sinks_) {
+    sink->Prepare(update_function_);
+    sink->engine_private().prepared_ = true;
+    uint32_t input_count = sink->input_count();
+    for (uint32_t input_index = 0; input_index < input_count; input_index++) {
+      Prepare(sink->input(input_index).upstream_stage());
+    }
+  }
+}
+
+void Engine::reset() {
+  supply_backlog_.clear();
+  demand_backlog_.clear();
+  sources_.clear();
+  sinks_.clear();
+  stages_.clear();
+}
+
+Engine::Part Engine::Add(Stage* stage) {
+  stages_.push_back(std::unique_ptr<Stage>(stage));

[johngro, 2016/01/20 00:25:32]

related to the question about unique pointers earler; why is this becoming
unique now?  Isn't the proper pattern to make it unique at construction time and
then to follow move-only semantics everywhere?

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

Using raw pointers here now.

+  if (stage->input_count() == 0) {
+    sources_.push_back(stage);
+  }
+  if (stage->output_count() == 0) {
+    sinks_.push_back(stage);
+  }
+  return Part(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::Prepare(Stage* stage) {
+  if (stage == nullptr || stage->engine_private().prepared_) {
+    return;
+  }
+
+  stage->Prepare(update_function_);
+  stage->engine_private().prepared_ = true;
+
+  uint32_t input_count = stage->input_count();
+  for (uint32_t input_index = 0; input_index < input_count; input_index++) {
+    Prepare(stage->input(input_index).upstream_stage());
+  }
+}
+
+void Engine::Update() {
+  while (true) {

[johngro, 2016/01/20 00:25:32]

something to consider moving forward;

If this is running on a mojo message loop, the loop is being blocked while this
is being run.  Re-queueing update jobs to the loop and un-winding out would
allow other messages to be processed while still running to completion.

If this is meant to be run on a dedicated thread, then some attention will need
to be paid to some form of external "shutdown now" signal so we can shut down
promptly without needing to run the processing loop to completion.  Given the
nature of the system (where stages and queue more work during the update call),
we cannot prove that this will ever run to completion, so it seems important for
there be a way to externally trigger a clean shutdown of the engine.

[dalesat, 2016/01/25 17:47:29]

Agreed. The threading model isn't complete.

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

Acknowledged.

+    Stage* stage = PopFromSupplyBacklog();
+    if (stage != nullptr) {
+      Update(stage);
+      continue;
+    }
+
+    stage = PopFromDemandBacklog();
+    if (stage != nullptr) {
+      Update(stage);
+      continue;
+    }
+
+    break;
+  }
+}
+
+void Engine::Update(Stage *stage) {
+  DCHECK(stage);
+
+  packets_produced_ = false;
+
+  stage->Update(*this);
+
+  // If the stage produced packets, it may need to reevaluate demand later.
+  if (packets_produced_) {
+    PushToDemandBacklog(stage);
+  }
+}
+
+void Engine::PushToSupplyBacklog(Stage* stage) {
+  DCHECK(stage);
+  packets_produced_ = true;
+  if (!stage->engine_private().in_supply_backlog_) {
+    // LIFO
+    supply_backlog_.push_back(stage);
+    stage->engine_private().in_supply_backlog_ = true;
+  }
+}
+
+void Engine::PushToDemandBacklog(Stage* stage) {
+  DCHECK(stage);
+  if (!stage->engine_private().in_demand_backlog_) {
+    // FIFO
+    demand_backlog_.push_front(stage);
+    stage->engine_private().in_demand_backlog_ = true;
+  }
+}
+
+Stage* Engine::PopFromSupplyBacklog() {
+  if (supply_backlog_.empty()) {
+    return nullptr;
+  }
+
+  Stage* stage = supply_backlog_.front();

[johngro, 2016/01/20 00:25:32]

the comment above indicates that this is supposed to be a stack (LIFO), but the
code here is pushing to the back and popping from the front, which is a queue
(FIFO) not a stack.

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

This still seems backwards, see updated comment in PS#5 around
PushTo(Supply|Demand)BacklogUnsafe

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

Done.

+  supply_backlog_.pop_front();
+  DCHECK(stage->engine_private().in_supply_backlog_);
+  stage->engine_private().in_supply_backlog_ = false;
+  return stage;
+}
+
+Stage* Engine::PopFromDemandBacklog() {
+  if (demand_backlog_.empty()) {
+    return nullptr;
+  }
+
+  Stage* stage = demand_backlog_.front();
+  demand_backlog_.pop_front();
+  DCHECK(stage->engine_private().in_demand_backlog_);
+  stage->engine_private().in_demand_backlog_ = false;
+  return stage;
+}
+
+}  // namespace media
+}  // namespace mojo