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

services/media/framework/stages/active_multistream_sink_stage.cc

Index: services/media/framework/stages/active_multistream_sink_stage.cc
diff --git a/services/media/framework/stages/active_multistream_sink_stage.cc b/services/media/framework/stages/active_multistream_sink_stage.cc
new file mode 100644
index 0000000000000000000000000000000000000000..9cd13412573433a49b5e1b567f4bc2172c6336ad
--- /dev/null
+++ b/services/media/framework/stages/active_multistream_sink_stage.cc
@@ -0,0 +1,179 @@
+// 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/stages/active_multistream_sink_stage.h"
+
+namespace mojo {
+namespace media {
+
+ActiveMultistreamSinkStage::ActiveMultistreamSinkStage(
+      std::shared_ptr<ActiveMultistreamSink> sink) : sink_(sink) {
+  DCHECK(sink_);
+  sink_->SetHost(this);
+}
+
+ActiveMultistreamSinkStage::~ActiveMultistreamSinkStage() {
+  base::AutoLock lock(lock_);
+}
+
+size_t ActiveMultistreamSinkStage::input_count() const {
+  base::AutoLock lock(lock_);
+  return inputs_.size();
+};
+
+Input& ActiveMultistreamSinkStage::input(size_t index) {
+  base::AutoLock lock(lock_);
+  DCHECK_LT(index, inputs_.size());

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

DCHECK(inputs_[index].allocated_);
?

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

Hmmm. Seems like we should allow that. Some code may not know which are
allocated and which aren't. It would be good to prevent a connection, however.

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

I'm really not comfortable with this.  When an input becomes unallocated, you
should not be able to access it anymore for any reason.  There is already a
difference in behavior.  If we have indices 0-7 allocated, then this sequence of
code starts to behave in a bunch of inconsistent ways.

ReleaseInput(7);
my_stage.input(7)->do_something(); // asserts
ReleaseInput(5);
my_stage.input(5)->do_something(); // does not assert, but acts on a dead
object; so who knows?
ReleaseInput(6);
my_stage.input(5)->do_something(); // now this asserts

Re, your comment below

"""
I don't want unallocated inputs to be null, because this makes everyone have to
deal with this eventuality when calling Stage::input().
"""

This is already true.  People already need to know how many inputs exist before
they call input, or they will DCHECK.  If they gain access to an invalid input
(because it is a hole in the inputs_ vector) and try to use it, they are just
going to DCHECK somewhere else, no?  What operations should be permitted and are
meaningful when performed on an input which was removed, but not de-allocated
because it just happend to exist in the middle of the index space?

This has become an object lifetime management issue again.  It was simple when
the graph was set up and inputs/outputs were static.  Now that they can be
dynamic, you are basically back to the same problem.  Whether you use
unique/shared/weak pointer patterns or you use your (Input|Output)Ref pattern,
you need to solve the same problem.  People can hold references (in this case,
InputRef objects) to objects which can be destroyed.  There needs to be some way
to invalidate the reference, or some way for the reference to keep the object
around.  Having a reference to an invalid object and not knowing that it is
invalid (whether because it has been returned to the heap, or unlinked from the
graph, or whatever) does not seem like it can be correct.

[dalesat, 2016/02/10 21:37:24]

Understood. We should discuss.

Adding a TODO

+  return inputs_[index]->input_;
+}
+
+size_t ActiveMultistreamSinkStage::output_count() const {
+  return 0;
+}
+
+Output& ActiveMultistreamSinkStage::output(size_t index) {
+  LOG(ERROR) << "output requested from sink";
+  static Output result;
+  return result;
+}
+
+PayloadAllocator* ActiveMultistreamSinkStage::PrepareInput(size_t index) {
+  return nullptr;
+}
+
+void ActiveMultistreamSinkStage::PrepareOutput(
+    size_t index,
+    PayloadAllocator* allocator,
+    const PrepareCallback& callback) {
+  LOG(ERROR) << "PrepareOutput called on sink";
+}
+
+void ActiveMultistreamSinkStage::Prime() {
+  DCHECK(sink_);
+  sink_->Prime();
+}
+
+void ActiveMultistreamSinkStage::Update(Engine* engine) {
+  DCHECK(engine);
+  DCHECK(sink_);
+
+  base::AutoLock lock(lock_);
+
+  for (auto iter = pending_inputs_.begin(); iter != pending_inputs_.end(); ) {
+    StageInput* input = *iter;
+    if (input->input_.packet_from_upstream()) {
+      input->demand_ = sink_->SupplyPacket(
+          input->index_,
+          std::move(input->input_.packet_from_upstream()));
+
+      if (input->demand_ == Demand::kNegative) {
+        auto remove_iter = iter;
+        ++iter;
+        pending_inputs_.erase(remove_iter);
+      }
+    } else {
+      ++iter;
+    }
+
+    input->input_.SetDemand(input->demand_, engine);
+  }
+}
+
+void ActiveMultistreamSinkStage::FlushInput(
+    size_t index,
+    const FlushCallback& callback) {
+  DCHECK(sink_);
+
+  sink_->Flush();
+
+  base::AutoLock lock(lock_);
+  inputs_[index]->demand_ = Demand::kNegative;
+  inputs_[index]->input_.Flush();
+
+  pending_inputs_.remove(inputs_[index]);
+}
+
+void ActiveMultistreamSinkStage::FlushOutput(size_t index) {
+  LOG(ERROR) << "FlushOutput called on sink";
+}
+
+size_t ActiveMultistreamSinkStage::AllocateInput() {
+  base::AutoLock lock(lock_);
+
+  StageInput* input;
+  if (unallocated_inputs_.empty()) {
+    input = new StageInput(inputs_.size());
+    inputs_.emplace_back(input);
+  } else {
+    input = unallocated_inputs_.front();
+    // Allocate lowest indices first.
+    unallocated_inputs_.pop_front();
+  }
+
+  input->allocated_ = true;
+
+  return input->index_;
+}
+
+size_t ActiveMultistreamSinkStage::ReleaseInput(size_t index) {
+  base::AutoLock lock(lock_);
+  DCHECK(index < inputs_.size());
+
+  StageInput* input = inputs_[index];
+  DCHECK(input);
+  DCHECK(input->allocated_);
+  DCHECK(!input->input_.connected());
+
+  input->allocated_ = false;
+
+  // Pop input if it's at the end of inputs_. Otherwise, insert it into
+  // unallocated_inputs_ in index order.

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

ok, so I am a bit confused about the intent behind inputs_ and
unallocated_inputs_ here, so I'm trying to read between the lines.  Let me start
by saying what I think is happening; please correct me if I get something wrong
here.

1) For this stage, we want to track multiple inputs.

2) Inputs are 0-indexed, and we want an O(k) operation to go from input_index to
input_pointer (which is why we use a vector<> for inputs_).

3) An invariant of the system is that "input_pointer ==
inputs_[input_pointer->index_]"

4) It is possible to remove an input from the system in the middle of the
currently allocated index rage, leaving a hole in the range of indices.  When
this happens, we leave the input instance in the inputs_ vector, but flag it as
not-allocated.

5) Another invariant of the system is that the last element in inputs_ (if any)
is always allocated.  This comes from the desire to make input_ be as small as
possible (in size, not necessarily capacity).  To maintain this invariant, when
we remove the end of the inputs_ vector, we keep removing elements as long as
the unallocated list indicates that the new end of vector is still unallocated.

6) When allocating a new input, we want to always use the lowest currently
un-allocated index, extending inputs_ only if there is no un-allocated index. 
We want this operation to be O(k), so we maintain a list<> of un-allocated
inputs in order to rapidly find the lowest unallocated index (if any) and
recycle it.

So:  If the above description is mostly correct, then I have the following
questions...

Q1)
The code below seems to be checking to see if the input-to-be-removed is at the
end of the vector<> of inputs.  If so, it wants to remove the input completely
(because of the #5 invariant above), and keep removing any gaps in the range to
satisfy #5, cleaning up the un-allocated list in the process.  Isn't the code
below always going to do nothing, however?  We know that index is the last index
in the input_ vector.  We know (because #5) that this must be allocated,
therefor it is not a member of unallocated_inputs_.  This means that the while
loop will never run, because it should be impossible for the back of the
unallocated_inputs_ list to have the index of the end of the inputs_ vector. 
Seems like the code should instead say...

if (index == inputs_.size() - 1) {
  inputs_.pop_back();  // <=== remove the guy who is not on unallocated_inputs,
but who we want to remove
  while (...) { } // cleanup the rest of unallocated
}

Q2)
These collections (inputs_, unallocated_inputs_, pending_inputs_) are tracking
raw StageInput pointers which we allocated from the heap during AllocateInput. 
Whose job is it to clean them up?  Here, we are popping the pointers out of the
collections, but not deleting anything; this feels like a memory leak.

Q3) How important is the need to make the find-an-unused index operation
(described in #6 above) O(k)?  Right now, finding an unused index is O(k), but
releasing an input is O(N), because of the O(N) insertion into the unused list
followed by the O(N) trim of the input and unused collection.

Q4) Your InputRef wrapper class has a valid method which makes sure that the
index it refers to is < the stage's input count.  It does not feel like this is
going to always do the correct thing here.  When an input is released from the
vector, it is removed from the vector when it is at the end (making an existing
InputRef say valid() == false), but left in the vector when removed from the
middle (making InputRef say valid() == true).

Q5) On a tangentially related topic; A stage is a source if it has no inputs,
and a sink if it has no outputs.  This is determined when the stage is added to
the Graph which contains it.  This code indicated that the number of
inputs/outputs is a dynamic property of the stage.  Who updates the Graph if a
stage's source/sink status changes because of the addition/removal of a stage?


Focusing on [Q3] for a bit:
Generally speaking, N feels like it should be relatively small.  How important
is is the O(k) lookup of an unused index described in #6?  If it is super
important, then why is it not important when removing an input?  Seems you could
also make unallocated_inputs_ be a set<> with an overloaded less functor and
then have O(log) lookup and removal if N were going to be really large.  You
also don't need to write your own insert-in-order method.  Even so, it seems
much simpler to simply NULL out entries in inputs_ when they are unallocated. 
This shifts the O(N) operation to the allocation half of things, but it
eliminated the need for the allocated_ state in an inputs, and the need for the
unallocated collection entirely.

Dovetailing this into the [Q2] question:
Why use collections of InputStage*s for unallocated or pending?  Why not use
collections of size_t indices?  This would allow you to use std::unique_ptr
inside of the inputs_ vector which would auto-magically handle the heap cleanup.
 It would also mean that if you chose to go down the set<> route for unused
(mentioned above), that you would no longer need to overload the less operator
for the collection.  (it still seems easier just to do away with
unallocated_inputs_ list entirely)

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

Q1) Yes, oversight.
Q2) Yes, also oversight. I think inputs_ used to be a vector of unique_ptr.
Q3) Good point, though the N's are different (# of inputs vs # of unallocated
inputs).  I could get rid of unallocated_inputs_ and count them instead - that
way I wouldn't have to search for one if it's not there.
Q4) Yes, InputRef::valid tells you nothing about whether an input is allocated.
That's not fantastic, but I'm OK with it.
Q5) True! The real problem is that this stage should always have at least one
input so it's not identified as a source.

I don't want unallocated inputs to be null, because this makes everyone have to
deal with this eventuality when calling Stage::input().

Having pending_inputs_ be a list of size_t is a good idea. For now, I'll make
unallocated_inputs_ a set of size_t.

+  if (index == inputs_.size() - 1) {
+    while (!unallocated_inputs_.empty() &&
+        unallocated_inputs_.back()->index_ == inputs_.size() - 1) {
+      unallocated_inputs_.pop_back();
+      inputs_.pop_back();
+    }
+  } else {
+    AddToUnallocatedUnsafe(input);
+  }
+
+  return inputs_.size();
+}
+
+void ActiveMultistreamSinkStage::UpdateDemand(
+    size_t input_index,
+    Demand demand) {
+  lock_.Acquire();
+  DCHECK(input_index < inputs_.size());
+  DCHECK(demand != Demand::kNegative);
+
+  StageInput* input = inputs_[input_index];
+  DCHECK(input);
+  input->demand_ = demand;
+  pending_inputs_.push_back(input);
+  lock_.Release();
+  RequestUpdate();
+}
+
+void ActiveMultistreamSinkStage::AddToUnallocatedUnsafe(StageInput* input) {
+  DCHECK(input);
+
+  if (unallocated_inputs_.empty() ||
+      input->index_ > unallocated_inputs_.back()->index_) {
+    unallocated_inputs_.push_back(input);
+  } else {
+    for (auto iter = unallocated_inputs_.begin();
+        iter != unallocated_inputs_.end();
+        ++iter) {

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

if you are going to keep this list<> and not turn it into a set, then...

DCHECK((*iter)->index != input->index_);

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

Acknowledged.

+      if ((*iter)->index_ > input->index_) {
+        unallocated_inputs_.insert(iter, input);
+        break;
+      }
+    }
+  }
+}
+
+}  // namespace media
+}  // namespace mojo