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

services/media/framework/engine.h

Index: services/media/framework/engine.h
diff --git a/services/media/framework/engine.h b/services/media/framework/engine.h
new file mode 100644
index 0000000000000000000000000000000000000000..fa614f2cd94eb24fbff519489839d9c176de5389
--- /dev/null
+++ b/services/media/framework/engine.h
@@ -0,0 +1,279 @@
+// 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.
+
+#ifndef SERVICES_MEDIA_FRAMEWORK_ENGINE_H_
+#define SERVICES_MEDIA_FRAMEWORK_ENGINE_H_
+
+#include <deque>
+#include <list>
+
+#include "base/synchronization/lock.h"
+#include "services/media/framework/stages/active_sink_stage.h"
+#include "services/media/framework/stages/active_source_stage.h"
+#include "services/media/framework/stages/distributor_stage.h"
+#include "services/media/framework/stages/lpcm_transform_stage.h"
+#include "services/media/framework/stages/packet_transform_stage.h"
+#include "services/media/framework/stages/stage.h"
+
+namespace mojo {
+namespace media {
+
+// Host for a source, sink or transform.
+class Engine {
+ public:
+  class Input;
+  class Output;
+
+  // Opaque Stage pointer used for graph building.
+  class Part {
+   public:
+    Part() : stage_(nullptr) {}
+
+    uint32_t input_count();
+    Input input(uint32_t index);
+    Input input();
+    uint32_t output_count();
+    Output output(uint32_t index);
+    Output output();
+    Part upstream_part(uint32_t index);
+    Part upstream_part();
+    Part downstream_part(uint32_t index);
+    Part downstream_part();
+
+   private:
+    explicit Part(Stage* stage) : stage_(stage) {}
+
+    explicit operator bool() const { return stage_ != nullptr; }
+
+    Stage* stage_;
+
+    friend Engine;
+    friend Input;
+    friend Output;
+  };
+
+  // Opaque StageInput pointer used for graph building.
+  class Input {
+   public:
+    Input() : stage_(nullptr), index_(0) {}
+
+    explicit operator bool() const { return stage_ != nullptr; }
+
+    Part part() { return Part(stage_); }
+
+    bool connected() {
+      DCHECK(stage_);
+      return stage_input().upstream_stage() != nullptr;
+    }
+
+    Part upstream_part() {
+      DCHECK(connected());
+      return Part(stage_input().upstream_stage());
+    }
+
+   private:
+    Input(Stage* stage, uint32_t index) :
+        stage_(stage), index_(index) {
+      DCHECK(stage_);
+      DCHECK(index_ < stage_->input_count());
+    }
+
+    StageInput& stage_input() {
+      DCHECK(stage_);
+      return stage_->input(index_);
+    }
+
+    Stage* stage_;
+    uint32_t index_;
+
+    friend Engine;
+    friend Part;
+    friend Output;
+  };
+
+  // Opaque StageOutput pointer used for graph building.
+  class Output {
+   public:
+    Output() : stage_(nullptr), index_(0) {}
+
+    explicit operator bool() const { return stage_ != nullptr; }
+
+    Part part() { return Part(stage_); }
+
+    bool connected() {
+      DCHECK(stage_);
+      return stage_output().downstream_stage() != nullptr;
+    }
+
+    Part downstream_part() {
+      DCHECK(connected());
+      return Part(stage_output().downstream_stage());
+    }
+
+   private:
+    Output(Stage* stage, uint32_t index) :
+        stage_(stage), index_(index) {
+      DCHECK(stage_);
+      DCHECK(index_ < stage_->output_count());
+    }
+
+    StageOutput& stage_output() {
+      DCHECK(stage_);
+      return stage_->output(index_);
+    }
+
+    Stage* stage_;
+    uint32_t index_;
+
+    friend Engine;
+    friend Part;
+    friend Input;
+  };
+
+  Engine();
+
+  ~Engine();
+
+  // Adds a part to the engine.
+  template<typename T, typename TBase>
+  Part Add(SharedPtr<T, TBase> t) {
+    DCHECK(t);
+    return Add(CreateStage(std::shared_ptr<TBase>(t)));
+  }
+
+  // Removes a part from the engine after disconnecting it from other parts.
+  void Remove(Part part);
+
+  // Connects an output connector to an input connector. Returns the dowstream
+  // part.
+  Part Connect(Output output, Input input);
+
+  // Connects a part with exactly one output to a part with exactly one input.
+  // Returns the downstream part.
+  Part Connect(Part upstream_part, Part downstream_part);
+
+  // Connects an output connector to a part that has exactly one input. Returns
+  // the downstream part.
+  Part Connect(Output output, Part downstream_part);
+
+  // Connects a part with exactly one output to an input connector. Returns the
+  // downstream part.
+  Part Connect(Part upstream_part, Input input);
+
+  // Disconnects an output connector and the input connector to which it's
+  // connected.
+  void Disconnect(Output output);
+
+  // Disconnects an input connector and the output connector to which it's
+  // connected.
+  void Disconnect(Input input);
+
+  // Disconnects and removes part and everything connected to it.
+  void RemoveAll(Part part);
+
+  // Disconnects and removes everything connected to output.
+  void RemoveAll(Output output);
+
+  // Disconnects and removes everything connected to input.
+  void RemoveAll(Input input);
+
+  // Adds all the parts in t (which must all have one input and one output) and
+  // connects them in sequence to the output connector. Returns the output
+  // connector of the last part or the output parameter if it is empty.
+  template<typename T>
+  Output AddAndConnectAll(
+      Output output,
+      const T& t) {
+    for (auto& element : t) {
+      Part part = Add(CreateStage(element));
+      Connect(output, part.input());
+      output = part.output();
+    }
+    return output;
+  }
+
+  // Prepares the engine.
+  void Prepare();
+
+  // Prepares the part and everything upstream of it. This method is used to
+  // prepare subgraphs added when the rest of the graph is already prepared.
+  void Prepare(Part part);
+
+  // Primes all the sinks in the graph.
+  void PrimeSinks();
+
+  // Removes all parts from the engine.
+  void Reset();
+
+  // Pushes the stage to the supply backlog if it isn't already there.
+  void PushToSupplyBacklogUnsafe(Stage* stage);
+
+  // Pushes the stage to the demand backlog if it isn't already there.
+  void PushToDemandBacklogUnsafe(Stage* stage);

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

The unsafe methods are the methods which assume that the lock is held, yes?  If
so, then shouldn't they all need to be private?

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

These are called by inputs and outputs, which are part of the
Engine/Stage/StageInput/StageOutput system. I'll make these private and have the
input and output classes listed as friends.

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

Acknowledged.

+
+ private:
+  // Adds a stage to the engine.
+  Part Add(Stage* stage);
+
+  // Disconnects an output.
+  void DisconnectOutputUnsafe(Stage* stage, uint32_t index);
+
+  // Disconnects an input.
+  void DisconnectInputUnsafe(Stage* stage, uint32_t index);
+
+  // Removes a stage.
+  void RemoveUnsafe(Stage* stage);
+
+  // Creates a stage from a source, sink or transform. A specialization of this
+  // template is defined for each type of source, sink or transform that can be
+  // added to the engine.
+  template<typename T>
+  static Stage* CreateStage(std::shared_ptr<T> t);
+
+  // CreateStage template specialization for MultiStreamPacketSource.
+  static Stage* CreateStage(MultiStreamPacketSourcePtr source);
+
+  // CreateStage template specialization for PacketTransform.
+  static Stage* CreateStage(PacketTransformPtr transform);
+
+  // CreateStage template specialization for ActiveSource.
+  static Stage* CreateStage(ActiveSourcePtr source);
+
+  // CreateStage template specialization for ActiveSink.
+  static Stage* CreateStage(ActiveSinkPtr sink);
+
+  // CreateStage template specialization for LpcmTransform.
+  static Stage* CreateStage(LpcmTransformPtr transform);
+
+  // Prepares a stage.
+  void PrepareUnsafe(Stage* stage);
+
+  // Processes the entire backlog.
+  void UpdateUnsafe();
+
+  // Performs processing for a single stage, updating the backlog accordingly.
+  void UpdateUnsafe(Stage *stage);
+
+  // Pops a stage from the supply backlog and returns it or returns nullptr if
+  // the supply backlog is empty.
+  Stage* PopFromSupplyBacklogUnsafe();
+
+  // Pops a stage from the demand backlog and returns it or returns nullptr if
+  // the demand backlog is empty.
+  Stage* PopFromDemandBacklogUnsafe();
+
+  mutable base::Lock lock_;

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

Now that you are adding thread safety and multi threading capabilities to the
Engine concept, it would be really good to have a comment (probably at the top
of the class) which gives a brief theory of operation and outlines the threading
assumptions.  Questions I have that I'd like to have answers spelled out to
are...

+ Is there an assumption that the execution of the jobs pending in the backlog
queues assumed to execute in some form of serialized context (eg; sort of single
threaded, but perhaps pumped by multiple threads, only 1 at a time)
+ What does the lock protect?  Is it just the backlog queues, or is it the
entire update cycle?
+ If the engine is holding the lock over the entire update cycle, how does a
stage safely queue new work or update demand during an update cycle?

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

Done.

+  std::list<Stage*> stages_;
+  std::list<Stage*> sources_;
+  std::list<Stage*> sinks_;
+  std::deque<Stage*> supply_backlog_;
+  std::deque<Stage*> demand_backlog_;
+  Stage::UpdateCallback update_function_;
+  bool packets_produced_;
+};
+
+}  // namespace media
+}  // namespace mojo
+
+#endif  // SERVICES_MEDIA_FRAMEWORK_ENGINE_ENGINE_H_