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

services/media/framework/engine.h

 // 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 <list>
 #include <queue>
 #include <stack>
+#include <unordered_map>
 
 #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/refs.h"
 #include "services/media/framework/stages/stage.h"
 
 namespace mojo {
 namespace media {
 
 //
-// USAGE
-//
-// TODO(dalesat): Consider adding a suffix to Engine::Part/Input/Output to
-// indicate that they're references.
-// TODO(dalesat): Consider folding PrimeSinks into Prepare.
-//
-// Engine is a container for sources, sinks and transforms ('parts') connected
-// in a graph. Engine::Part, Engine::Input and Engine::Output are all opaque
-// references to parts and their inputs and outputs. Engine provides a variety
-// of methods for adding and removing parts and for connecting inputs and
-// outputs to form a graph.
-//
-// In addition to containing parts and representing their interconnection,
-// Engine manages the coordinated operation of its constituent parts and
-// transports media from part to part. The Prepare method prepares the graph
-// for operation, and the PrimeSinks method tells the sinks in the graph to
-// prime themselves. Any additional actions required to make the graph operate
-// (such as manipulating a rate control interface) is out of scope.
-//
-// Parts added to the engine are referenced using shared pointers. The engine
-// holds pointers to the parts it contains, and the application, in many cases,
-// also holds pointers to the parts so it can call methods that are outside the
-// engine's scope. When a part is added the Engine returns an Engine::Part
-// object, which can be used to reference the part when the graph is modified.
-// Engine::Part objects can be interrogated to retrieve inputs (as Engine::Input
-// objects) and outputs (as Engine::Output objects).
-//
-// Some support is provided for modifying graphs that are operating. This
-// capability isn't fully developed at the moment. Prepare(Part) is an example
-// of a method provided for this purpose.
-//
-// Parts come in various flavors, defined by 'model' abstract classes. The
-// current list of supported models is:
-//
-//  ActiveSink              - a sink that consumes packets asynchronously
-//  ActiveSource            - a source that produces packets asynchronously
-//  LpcmMixer               - a transform that mixes LPCM frames from multiple
-//                            inputs and produces a single stream of LPCM frames
-//                            via one output
-//  LpcmSource              - a source that produces LPCM frames synchronously
-//  LpcmTransform           - a synchronous transform with one LPCM input and
-//                            one LPCM output
-//  MultiStreamPacketSource - a source that produces multiple streams of packets
-//                            synchronously
-//  PacketTransform         - a synchronous transform that consumes and produces
-//                            packets via one input and one output
-//
-// Other models will be defined in the future as needed.
-//
-
-//
 // DESIGN
 //
-// The Engine is implemented as a system of cooperating objects. Of those
-// objects, only the engine itself is of relevance to code that uses Engine and
-// to part implementations. The other objects are:
-//
-// Stage
-// A stage hosts a single part. There are many subclasses of Stage, one for
-// each supported part model. The stage's job is to implement the contract
-// represented by the model so the parts that conform to the model can
-// participate in the operation of the engine. Stages are uniform with respect
-// to how they interact with engine. Engine::Part references a stage.
-//
-// StageInput
-// A stage possesses zero or more StageInput instances. StageInput objects
-// implement the supply of media into the stage and demand for media signalled
-// upstream. StageInputs receive media from StageOutputs in the form of packets
-// (type Packet). LpcmStageInput is a subclass of StageInput that interoperates
-// with LpcmStageInputs in a way that provides optimizations relavant to LPCM
-// audio media. Engine::Input references a StageInput.
-//
-// StageOutput
-// A stage possesses zero or more StageOutput instances. StageOutput objects
-// implement the supply of media output of the stage to a downstream input and
-// demand for media signalled from that input. LpcmStageOutput implements
-// optimized LPCM flow. Engine::Output references a StageOutput.
-//
-// Engine uses a 'work list' algorithm to operate the contained graph. The
+// Engine uses a 'work list' algorithm to operate the graph. The
 // engine has a backlog of stages that need to be updated. To advance the
 // operation of the graph, the engine removes a stage from the backlog and calls
 // the stage's Update method. The Stage::Update may cause stages to be added
 // synchronously to the the backlog. This procedure continues until the backlog
 // is empty.
 //
 // Stage::Update is the stage's opportunity to react to the supply of new media
 // via its inputs and the signalling of new demand via its outputs. During
 // Update, the stage does whatever work it can with the current supply and
 // demand, possibly supplying media downstream through its outputs and/or
@@ skipping 23 matching lines @@
 //    call, because the lock is taken for the duration of Update, and the
 //    callback will take the lock. Update callbacks may occur during Engine::
 //    PrimeSinks, and they generally will.
 // 2) A stage cannot update supply/demand on its inputs/outputs except during
 //    Update. When an external event occurs, the stage and/or its hosted part
 //    should update its internal state as required and invoke the callback.
 //    During the subsequent Update, the stage and/or part can then update
 //    supply and/or demand.
 // 3) Threads used to call update callbacks must be suitable for operating the
 //    engine. There is currently no affordance for processing other tasks on
-//    a thread while the callback is running. A callback may run for a long
+//    the thread while the callback is running. A callback may run for a long
 //    time, depending on how much work needs to be done.
 // 4) Parts cannot rely on being called back on the same thread on which they
 //    invoke update callbacks. This may require additional synchronization and
 //    thread transitions inside the part.
 // 5) If a part takes a lock of its own during Update, it should not also hold
 //    that lock when calling the update callback. Doing so will result in
 //    deadlock.
 //
 // NOTE: Allocators, not otherwise discussed here, are required to be thread-
 // safe so that packets may be cleaned up on any thread.
 //
 // In the future, the threading model will be enhanced. Intended features
 // include:
 // 1) Support for multiple threads.
 // 2) Marshalling update callbacks to a different thread.
 //
 
-// Host for a source, sink or transform.
+// Manages operation of a Graph.
 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)));
-  }
+  // Prepares the input and the subgraph upstream of it.
+  void PrepareInput(const InputRef& input_ref);
 
-  // Removes a part from the engine after disconnecting it from other parts.
-  void RemovePart(Part part);
+  // Unprepares the input and the subgraph upstream of it.
+  void UnprepareInput(const InputRef& input_ref);
 
-  // Connects an output connector to an input connector. Returns the dowstream
-  // part.
-  Part Connect(Output output, Input input);
+  // Flushes the output and the subgraph downstream of it.
+  void FlushOutput(const OutputRef& output_ref);
 
-  // Connects a part with exactly one output to a part with exactly one input.
-  // Returns the downstream part.
-  Part ConnectParts(Part upstream_part, Part downstream_part);
+  // Queues the stage for update and winds down the backlog.
+  void RequestUpdate(Stage* stage);
 
-  // Connects an output connector to a part that has exactly one input. Returns
-  // the downstream part.
-  Part ConnectOutputToPart(Output output, Part downstream_part);
+  // Pushes the stage to the supply backlog if it isn't already there.
+  void PushToSupplyBacklog(Stage* stage);
 
-  // Connects a part with exactly one output to an input connector. Returns the
-  // downstream part.
-  Part ConnectPartToInput(Part upstream_part, Input input);
-
-  // Disconnects an output connector and the input connector to which it's
-  // connected.
-  void DisconnectOutput(Output output);
-
-  // Disconnects an input connector and the output connector to which it's
-  // connected.
-  void DisconnectInput(Input input);
-
-  // Disconnects and removes part and everything connected to it.
-  void RemovePartsConnectedToPart(Part part);
-
-  // Disconnects and removes everything connected to output.
-  void RemovePartsConnectedToOutput(Output output);
-
-  // Disconnects and removes everything connected to input.
-  void RemovePartsConnectedToInput(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 demand backlog if it isn't already there.
+  void PushToDemandBacklog(Stage* stage);
 
  private:
-  // Adds a stage to the engine.
-  Part Add(Stage* stage);
+  using UpstreamVisitor = std::function<void(
+      const InputRef& input,
+      const OutputRef& output,
+      const Stage::UpstreamCallback& callback)>;
+  using DownstreamVisitor = std::function<void(
+      const OutputRef& output,
+      const InputRef& input,
+      const Stage::DownstreamCallback& callback)>;
 
-  // Disconnects an output.
-  void DisconnectOutputUnsafe(Stage* stage, uint32_t index);
+  void VisitUpstream(
+      const InputRef& input,
+      const UpstreamVisitor& vistor);
 
-  // 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 if all its downstream stages are prepared.
-  void MaybePrepareUnsafe(Stage* stage);
+  void VisitDownstream(
+      const OutputRef& output,
+      const DownstreamVisitor& vistor);
 
   // Processes the entire backlog.
-  void UpdateUnsafe();
+  void Update();
 
   // Performs processing for a single stage, updating the backlog accordingly.
-  void UpdateUnsafe(Stage *stage);
-
-  // 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);
+  void Update(Stage *stage);
 
   // Pops a stage from the supply backlog and returns it or returns nullptr if
   // the supply backlog is empty.
-  Stage* PopFromSupplyBacklogUnsafe();
+  Stage* PopFromSupplyBacklog();
 
   // Pops a stage from the demand backlog and returns it or returns nullptr if
   // the demand backlog is empty.
-  Stage* PopFromDemandBacklogUnsafe();
+  Stage* PopFromDemandBacklog();
 
   mutable base::Lock lock_;
-  std::list<Stage*> stages_;
-  std::list<Stage*> sources_;
-  std::list<Stage*> sinks_;
   // supply_backlog_ contains pointers to all the stages that have been supplied
   // (packets or frames) but have not been updated since. demand_backlog_ does
   // the same for demand. The use of queue vs stack here is a guess as to what
   // will yield the best results. It's possible that only a single backlog is
   // required.
   // TODO(dalesat): Determine the best ordering and implement it.
   std::queue<Stage*> supply_backlog_;
   std::stack<Stage*> demand_backlog_;
-  Stage::UpdateCallback update_function_;
   bool packets_produced_;
-
-  friend class StageInput;
-  friend class StageOutput;
-  friend class LpcmStageInput;
-  friend class LpcmStageOutput;
 };
 
 }  // namespace media
 }  // namespace mojo
 
-#endif  // SERVICES_MEDIA_FRAMEWORK_ENGINE_ENGINE_H_
+#endif  // SERVICES_MEDIA_FRAMEWORK_ENGINE_H_