Add an initial revision of an audio server.

services/media/audio/audio_pipe.cc

Index: services/media/audio/audio_pipe.cc
diff --git a/services/media/audio/audio_pipe.cc b/services/media/audio/audio_pipe.cc
new file mode 100644
index 0000000000000000000000000000000000000000..57823717791d155e9f285e4bca677d9d7d6acea7
--- /dev/null
+++ b/services/media/audio/audio_pipe.cc
@@ -0,0 +1,130 @@
+// Copyright 2015 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 <limits>
+#include <vector>
+
+#include "services/media/audio/audio_pipe.h"
+#include "services/media/audio/audio_server_impl.h"
+#include "services/media/audio/audio_track_impl.h"
+
+namespace mojo {
+namespace media {
+namespace audio {
+
+AudioPipe::AudioPacketRef::AudioPacketRef(MediaPacketStatePtr state,
+                                          AudioServerImpl* server,
+                                          std::vector<Region>&& regions,
+                                          int64_t start_pts,
+                                          int64_t end_pts)
+  : state_(std::move(state)),
+    server_(server),
+    regions_(regions),
+    start_pts_(start_pts),
+    end_pts_(end_pts)  {
+  DCHECK(state_);
+  DCHECK(server_);
+}
+
+AudioPipe::AudioPacketRef::~AudioPacketRef() {
+  DCHECK(server_);
+  server_->SchedulePacketCleanup(std::move(state_));

[jeffbrown, 2015/11/04 23:43:33]

Destructor side-effects like these make be a little nervous, but ok.  Assuming
this is RAII-ish.

[johngro, 2015/11/06 02:20:25]

yeah, I am not thrilled with this either.  It is a necessary consequence of not
being able to safely send messages on a Mojo message pipe from anything but the
main message loop thread.  It also adds unfortunate latency where we really want
to have none at all.  Clients can currently mitigate this using a strictly feed
forward timing model for buffering, and over-allocating their buffer just a bit,
but I would very much like to eliminate this trampoline behavior in general.

For now, this consequence of what the mojo team has asked me to do based on
current framework architecture.  Moving forward, I would like to collect some
hard data on the performance penalty, and work to address the issue (presuming
it is real and significant, and I expect that it will be at scale), either in
the fundamental design of inter-process media transport, or in the thread-safety
capabilities of the mojo framework.

+}
+
+AudioPipe::AudioPipe(AudioTrackImpl* owner,
+                     AudioServerImpl* server)
+  : owner_(owner),
+    server_(server) {
+  DCHECK(owner_);
+  DCHECK(server_);
+}
+
+AudioPipe::~AudioPipe() {}
+
+void AudioPipe::OnPacketReceived(MediaPacketStatePtr state) {
+  const MediaPacketPtr& packet = state->GetPacket();
+  DCHECK(packet);
+  DCHECK(packet->payload);
+  DCHECK(owner_);
+
+  // Start by making sure that we are regions we are receiving are made from an
+  // integral number of audio frames.  Count the total number of frames in the
+  // process.
+  //
+  // TODO(johngro): Someday, automatically enforce this using
+  // alignment/allocation restrictions at the MediaPipe level of things.
+  uint32_t frame_count = 0;
+  uint32_t frame_size = owner_->BytesPerFrame();
+  const MediaPacketRegionPtr* region = &packet->payload;
+  size_t ndx = 0;
+  std::vector<AudioPacketRef::Region> regions;

[jeffbrown, 2015/11/04 23:43:33]

I was slightly surprised to see you allocating on the heap here.

[johngro, 2015/11/06 02:20:25]

I was made slightly ill by my actions in doing so.  At this point, even thinking
about what it costs to move media from point A to point B makes me want to go
lie down in a dark room and cry.  What makes me hate myself even more is that I
have (repeatedly) caved into logic from others which says (in effect) "Its
already bad, whats wrong with a little bit more bad?"

At this point, the situation is not good.  Making it better will require hard
data and load requirements to drive the system in the direction of efficiency as
a whole.  This is a post V0 thing, but rest assured that I have a long mental
list of the places where I already know we can come back and start to make
things better.

+
+  regions.reserve(packet->extra_payload.size() + 1);
+
+  DCHECK(frame_size);
+  while (true) {
+    if ((frame_size > 1) && ((*region)->length % frame_size)) {
+      state->SetResult(MediaResult::INVALID_ARGUMENT);

[jeffbrown, 2015/11/04 23:43:33]

can we just close the pipe?

[johngro, 2015/11/06 02:20:25]

Acknowledged.
Placing the general close-the-pipe-after-V0 bucket.

+      return;
+    }
+
+    frame_count += ((*region)->length / frame_size);
+    if (frame_count > (std::numeric_limits<uint32_t>::max() >>
+                       AudioTrackImpl::PTS_FRACTIONAL_BITS)) {
+      state->SetResult(MediaResult::INVALID_ARGUMENT);
+      return;
+    }
+
+    regions.emplace_back(
+        static_cast<const uint8_t*>(buffer()) + (*region)->offset,

[jeffbrown, 2015/11/04 23:43:33]

Make sure to check that the offset is within the buffer and is properly aligned
too.

[johngro, 2015/11/06 02:20:25]

Offset and length containment has already been handled by the MediaPipe level of
things before we got here.  Alignment and length-divisibility restrictions
belongs at that level as well, I have multiple comments all over that code
calling that out.

I have not bothered to address this so far because...
1) This is V0, and I was concerned with unblocking to enable parallel
development.
2) There are already cries to scrap this whole thing and redesign.  Until I get
to the other side of that debate and have reached consensus, I am not
particularly motivated to continue to improve something which may be completely
replaced in the near term anyway.

+        frame_count << AudioTrackImpl::PTS_FRACTIONAL_BITS);
+
+    if (ndx >= packet->extra_payload.size()) {
+      break;
+    }
+
+    region = &(packet->extra_payload[ndx]);
+    ndx++;
+  }
+
+  // Figure out the starting PTS.
+  int64_t start_pts;
+  if (packet->pts != MediaPacket::kNoTimestamp) {
+    // The user provided an explicit PTS for this audio.  Transform it into
+    // units of fractional frames.
+    LinearTransform tmp(0, owner_->FractionalFrameToMediaTimeRatio(), 0);
+    if (!tmp.DoForwardTransform(packet->pts, &start_pts)) {
+      state->SetResult(MediaResult::INTERNAL_ERROR);
+      return;
+    }
+  } else {
+    // No PTS was provided.  Use the end time of the last audio packet, if
+    // known.  Otherwise, just assume a media time of 0.
+    start_pts = next_pts_known_ ? next_pts_ : 0;
+  }
+
+  // The end pts is the value we will use for the next packet's start PTS, if
+  // the user does not provide an explicit PTS.
+  int64_t pts_delta = (static_cast<int64_t>(frame_count)
+                    << AudioTrackImpl::PTS_FRACTIONAL_BITS);
+  next_pts_ = start_pts + pts_delta;
+  next_pts_known_ = true;
+
+  owner_->OnPacketReceived(AudioPacketRefPtr(
+        new AudioPacketRef(std::move(state),
+                           server_,
+                           std::move(regions),

[jeffbrown, 2015/11/04 23:43:33]

So this object contains a pointer to an mmap'd region of the media pipe buffer. 
Are we guaranteed that the pointer will be discarded before the buffer is
unmapped?

I have the feeling this buffer is going to flow deep into the system and be sent
to worker threads which are operating asynchronously.  Consider including a
reference counted reference to the buffer along the way so we don't screw it up.

[johngro, 2015/11/06 02:20:25]

FWIW - enforcement like this belong at the MediaPipeBase level, not at this
level.  MediaPipeBase manages the mapping and unmapping of the buffer, as well
as the sanity checking of the MediaPackets and MediaPacketState objects created
to manage the lifetime of the objects whcih refer to the memory as well as the
callback which *must* be made to inform the client that the shared buffer is
available for writing again.

Re: lifecycle management issues.
It is currently guaranteed based on they way things used in the system as a
whole, but I agree that this is a weak and non-obvious guarantee.

Having a stronger, more direct, guarantee would require one of a few different
techniques...

1) MediaPacketState objects could hold a strong reference to the state in the
MediaPipeBase which controls the mapping/unmapping of the shared buffer. 
Expensive, but would provide a strong guarantee.
2) MediaPacketState objects could pursue a #1 style solution, but hold a weak
reference instead.  This puts further requirements on the consumer of the media
in the form of promoting the weak reference and releasing  generated strong
reference, but it provides an equally strong guarantee along with better
guarantees on timely unmapping of the shared buffer.
3) MediaPipeBase objects could implement some form of debug only ref counting
checks.  This would require MediaPacketState objects to hold some form of
reference to the MediaPipeBase object (currently, they hold none).  This still
relies on "everyone managing the lifecycle properly" which is the current state
of affairs, but provides some debug sanity checking.  I actually considered
implementing this early on, but I think that I decided that if I was going to
pay the price of holding a reference in the MediaPacketState object, that I
should go with #1 or #2 and have a strong guarantee instead of #3 which provides
only some debug checking on an already weak promise (its not even a guarantee,
TBH).

So;
I'm going to put this on my list of TODOs, and will most likely pursue #1
assuming that any of this code survives the up-and-coming MediaPipe design
post-mortem.

+                           start_pts,
+                           next_pts_)));
+}
+
+void AudioPipe::OnFlushRequested(const FlushCallback& cbk) {
+  DCHECK(owner_);
+  owner_->OnFlushRequested(cbk);
+  next_pts_known_ = false;
+}
+
+}  // namespace audio
+}  // namespace media
+}  // namespace mojo