[Remoting Mobile] Implementing AudioJitterBuffer

This is part of the work to rewire the audio playback logic for iOS
(and potentially also for Android).

AudioJitterBuffer acts as a jitter buffer that queues up AudioPackets
once they are received and before they are consumed by a native playback
queue.
This class is basically modified from AudioPlayerBuffer with a few
changes:
* Allow handling packets in different stream formats/sample rate and
  call a callback when the format is changed.
* Move the underrun protection logic from AudioPlayerIos (the
  "PRIMING" state) into the jitter buffer.
* Fix a bug where bytes_extracted_ is ignored when copying audio data.
* Removed the AudioStreamConsumer inheritance, Stop() and other unused
  logic.

Bug: 868088
Change-Id: Ibaee655df28de68f5303d735ef89cf15695ab2ff
Reviewed-on: https://chromium-review.googlesource.com/1150936
Commit-Queue: Yuwei Huang <yuweih@chromium.org>
Reviewed-by: Joe Downing <joedow@chromium.org>
Cr-Commit-Position: refs/heads/master@{#580709}

remoting/client/audio/BUILD.gn

@@ -4,8 +4,12 @@
 
 source_set("audio") {
   sources = [
+    "async_audio_data_supplier.cc",
+    "async_audio_data_supplier.h",
     "async_audio_frame_supplier.h",
     "audio_frame_supplier.h",
+    "audio_jitter_buffer.cc",
+    "audio_jitter_buffer.h",
     "audio_player.cc",
     "audio_player.h",
     "audio_player_android.cc",
@@ -37,6 +41,7 @@ source_set("unit_tests") {
   testonly = true
 
   sources = [
+    "audio_jitter_buffer_unittest.cc",
     "audio_player_buffer_unittest.cc",
     "audio_player_unittest.cc",
   ]

remoting/client/audio/async_audio_data_supplier.cc

+// Copyright 2018 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 "remoting/client/audio/async_audio_data_supplier.h"
+
+#include "base/logging.h"
+
+namespace remoting {
+
+AsyncAudioDataSupplier::GetDataRequest::GetDataRequest(void* data_arg,
+                                                       size_t bytes_needed_arg)
+    : data(data_arg), bytes_needed(bytes_needed_arg) {
+  DCHECK(data);
+  DCHECK_GT(bytes_needed, 0u);
+}
+
+AsyncAudioDataSupplier::GetDataRequest::~GetDataRequest() = default;
+
+AsyncAudioDataSupplier::AsyncAudioDataSupplier() = default;
+
+AsyncAudioDataSupplier::~AsyncAudioDataSupplier() = default;
+
+}  // namespace remoting

remoting/client/audio/async_audio_data_supplier.h

+// Copyright 2018 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 REMOTING_CLIENT_AUDIO_ASYNC_AUDIO_DATA_SUPPLIER_H_
+#define REMOTING_CLIENT_AUDIO_ASYNC_AUDIO_DATA_SUPPLIER_H_
+
+#include <memory>
+
+namespace remoting {
+
+// This interface allows caller to asynchronously request for audio data.
+class AsyncAudioDataSupplier {
+ public:
+  struct GetDataRequest {
+    // |data| must outlive |this|.
+    GetDataRequest(void* data, size_t bytes_needed);
+    virtual ~GetDataRequest();
+
+    // Called when |data| has been filled with |bytes_needed| bytes of data.
+    //
+    // Caution: Do not add or drop requests (i.e. calling AsyncGetData() or
+    // ClearGetDataRequests()) directly inside OnDataFilled(), which has
+    // undefined behavior. Consider posting a task when necessary.
+    virtual void OnDataFilled() = 0;
+
+    void* const data;
+    const size_t bytes_needed;
+
+    size_t bytes_extracted = 0;
+  };
+
+  AsyncAudioDataSupplier();
+  virtual ~AsyncAudioDataSupplier();
+
+  // Requests for more data from the supplier.
+  virtual void AsyncGetData(std::unique_ptr<GetDataRequest> request) = 0;
+
+  // Drops all pending get-data requests. You may want to call this before the
+  // caller is destroyed if the caller has a shorter lifetime than the supplier.
+  virtual void ClearGetDataRequests() = 0;
+};
+
+}  // namespace remoting
+
+#endif  // REMOTING_CLIENT_AUDIO_ASYNC_AUDIO_DATA_SUPPLIER_H_

remoting/client/audio/audio_jitter_buffer.cc

+// Copyright 2018 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 "remoting/client/audio/audio_jitter_buffer.h"
+
+#include <algorithm>
+#include <string>
+
+#include "base/logging.h"
+#include "base/stl_util.h"
+
+namespace {
+
+// AudioJitterBuffer maintains a list of AudioPackets whose total playback
+// duration <= |kMaxQueueLatency|.
+// Once the buffer has run out of AudioPackets (latency reaches 0), it waits
+// until the total latency reaches |kUnderrunRecoveryLatency| before it starts
+// feeding the get-data requests. This helps reduce the frequency of stopping
+// when the buffer underruns.
+// If the total latency has reached |kMaxQueueLatency|, the oldest packets
+// will get dropped until the latency is reduced to no more than
+// |kOverrunRecoveryLatency|. This helps reduce the number of glitches when
+// the buffer overruns.
+// Otherwise the total latency can freely fluctuate between 0 and
+// |kMaxQueueLatency|.
+
+constexpr base::TimeDelta kMaxQueueLatency =
+    base::TimeDelta::FromMilliseconds(150);
+constexpr base::TimeDelta kUnderrunRecoveryLatency =
+    base::TimeDelta::FromMilliseconds(60);
+constexpr base::TimeDelta kOverrunRecoveryLatency =
+    base::TimeDelta::FromMilliseconds(90);
+
+}  // namespace
+
+namespace remoting {
+
+bool AudioJitterBuffer::StreamFormat::operator==(
+    const StreamFormat& other) const {
+  return bytes_per_sample == other.bytes_per_sample &&
+         channels == other.channels && sample_rate == other.sample_rate;
+}
+
+bool AudioJitterBuffer::StreamFormat::operator!=(
+    const StreamFormat& other) const {
+  return !(*this == other);
+}
+
+AudioJitterBuffer::AudioJitterBuffer(
+    OnFormatChangedCallback on_format_changed) {
+  DETACH_FROM_THREAD(thread_checker_);
+  on_format_changed_ = std::move(on_format_changed);
+}
+
+AudioJitterBuffer::~AudioJitterBuffer() = default;
+
+void AudioJitterBuffer::AddAudioPacket(std::unique_ptr<AudioPacket> packet) {
+  DCHECK_CALLED_ON_VALID_THREAD(thread_checker_);
+  CHECK_EQ(1, packet->data_size());
+  DCHECK_EQ(AudioPacket::ENCODING_RAW, packet->encoding());
+  DCHECK_NE(AudioPacket::SAMPLING_RATE_INVALID, packet->sampling_rate());
+
+  StreamFormat stream_format;
+  stream_format.bytes_per_sample = packet->bytes_per_sample();
+  stream_format.channels = packet->channels();
+  stream_format.sample_rate = packet->sampling_rate();
+
+  DCHECK_GT(stream_format.bytes_per_sample, 0);
+  DCHECK_GT(stream_format.channels, 0);
+  DCHECK_GT(stream_format.sample_rate, 0);
+  DCHECK_EQ(packet->data(0).size() %
+                (stream_format.channels * stream_format.bytes_per_sample),
+            0u);
+
+  if (!stream_format_ || *stream_format_ != stream_format) {
+    ResetBuffer(stream_format);
+  }
+
+  // Push the new data to the back of the queue.
+  queued_bytes_ += packet->data(0).size();
+  queued_packets_.push_back(std::move(packet));
+
+  if (underrun_protection_mode_ &&
+      queued_bytes_ > GetBufferSizeFromTime(kUnderrunRecoveryLatency)) {
+    // The buffer has enough data to start feeding the requests.
+    underrun_protection_mode_ = false;
+  }
+
+  DropOverrunPackets();
+  ProcessGetDataRequests();
+}
+
+void AudioJitterBuffer::AsyncGetData(std::unique_ptr<GetDataRequest> request) {
+  DCHECK_CALLED_ON_VALID_THREAD(thread_checker_);
+  DCHECK(stream_format_);
+  DCHECK_EQ(request->bytes_needed % stream_format_->bytes_per_sample, 0u);
+  queued_requests_.push_back(std::move(request));
+  ProcessGetDataRequests();
+}
+
+void AudioJitterBuffer::ClearGetDataRequests() {
+  queued_requests_.clear();
+}
+
+void AudioJitterBuffer::ResetBuffer(const StreamFormat& new_format) {
+  DCHECK_CALLED_ON_VALID_THREAD(thread_checker_);
+  queued_packets_.clear();
+  queued_bytes_ = 0;
+  first_packet_offset_ = 0;
+  ClearGetDataRequests();
+  stream_format_ = std::make_unique<StreamFormat>(new_format);
+  underrun_protection_mode_ = true;
+  if (on_format_changed_) {
+    on_format_changed_.Run(*stream_format_);
+  }
+}
+
+void AudioJitterBuffer::ProcessGetDataRequests() {
+  DCHECK_CALLED_ON_VALID_THREAD(thread_checker_);
+
+  if (underrun_protection_mode_) {
+    return;
+  }
+
+  // Get the active request if there is one.
+  while (!queued_requests_.empty() && !queued_packets_.empty()) {
+    auto& active_request = queued_requests_.front();
+
+    // Copy any available data into the active request up to as much requested.
+    while (active_request->bytes_extracted < active_request->bytes_needed &&
+           !queued_packets_.empty()) {
+      uint8_t* next_data = static_cast<uint8_t*>(active_request->data) +
+                           active_request->bytes_extracted;
+
+      const std::string& packet_data = queued_packets_.front()->data(0);
+      size_t bytes_to_copy = std::min(
+          packet_data.size() - first_packet_offset_,
+          active_request->bytes_needed - active_request->bytes_extracted);
+
+      memcpy(next_data, packet_data.data() + first_packet_offset_,
+             bytes_to_copy);
+
+      first_packet_offset_ += bytes_to_copy;
+      active_request->bytes_extracted += bytes_to_copy;
+      queued_bytes_ -= bytes_to_copy;
+      DCHECK_GE(queued_bytes_, 0u);
+      // Pop off the packet if we've already consumed all its bytes.
+      if (queued_packets_.front()->data(0).size() == first_packet_offset_) {
+        queued_packets_.pop_front();
+        first_packet_offset_ = 0;
+      }
+    }
+
+    // If this request is fulfilled, call the callback and pop it off the queue.
+    if (active_request->bytes_extracted == active_request->bytes_needed) {
+      active_request->OnDataFilled();
+      queued_requests_.pop_front();
+    }
+  }
+
+  if (queued_packets_.empty()) {
+    // Buffer overrun.
+    underrun_protection_mode_ = true;
+  }
+}
+
+size_t AudioJitterBuffer::GetBufferSizeFromTime(
+    base::TimeDelta duration) const {
+  DCHECK_CALLED_ON_VALID_THREAD(thread_checker_);
+  DCHECK(stream_format_);
+  return duration.InMilliseconds() * stream_format_->sample_rate *
+         stream_format_->bytes_per_sample * stream_format_->channels /
+         base::Time::kMillisecondsPerSecond;
+}
+
+void AudioJitterBuffer::DropOverrunPackets() {
+  DCHECK_CALLED_ON_VALID_THREAD(thread_checker_);
+
+  if (queued_bytes_ <= GetBufferSizeFromTime(kMaxQueueLatency)) {
+    return;
+  }
+
+  size_t new_size = GetBufferSizeFromTime(kOverrunRecoveryLatency);
+  while (queued_bytes_ > new_size) {
+    queued_bytes_ -=
+        queued_packets_.front()->data(0).size() - first_packet_offset_;
+    DCHECK_GE(queued_bytes_, 0u);
+    queued_packets_.pop_front();
+    first_packet_offset_ = 0;
+  }
+}
+
+}  // namespace remoting

remoting/client/audio/audio_jitter_buffer.h

+// Copyright 2018 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 REMOTING_CLIENT_AUDIO_AUDIO_JITTER_BUFFER_H_
+#define REMOTING_CLIENT_AUDIO_AUDIO_JITTER_BUFFER_H_
+
+#include <list>
+#include <memory>
+
+#include "base/callback.h"
+#include "base/macros.h"
+#include "base/memory/weak_ptr.h"
+#include "base/threading/thread_checker.h"
+#include "base/time/time.h"
+#include "remoting/client/audio/async_audio_data_supplier.h"
+#include "remoting/proto/audio.pb.h"
+
+namespace remoting {
+
+// This is a jitter buffer that queues up audio packets and get-data requests
+// and feeds the requests with the data when the buffer has enough data.
+class AudioJitterBuffer : public AsyncAudioDataSupplier {
+ public:
+  struct StreamFormat {
+    bool operator==(const StreamFormat& other) const;
+    bool operator!=(const StreamFormat& other) const;
+
+    int bytes_per_sample = 0;
+    int channels = 0;
+    int sample_rate = 0;
+  };
+
+  using OnFormatChangedCallback =
+      base::RepeatingCallback<void(const StreamFormat& format)>;
+
+  // |callback| is called once the jitter buffer gets the first packet or the
+  // stream format has been changed.
+  // Pending get-data requests will be dropped when the stream format is
+  // changed.
+  explicit AudioJitterBuffer(OnFormatChangedCallback on_format_changed);
+  ~AudioJitterBuffer() override;
+
+  void AddAudioPacket(std::unique_ptr<AudioPacket> packet);
+
+  // AsyncAudioDataSupplier implementations.
+  void AsyncGetData(std::unique_ptr<GetDataRequest> request) override;
+  void ClearGetDataRequests() override;
+
+ private:
+  friend class AudioJitterBufferTest;
+
+  // Clears the jitter buffer, drops all pending requests, and notify
+  // |on_format_changed_| that the format has been changed.
+  void ResetBuffer(const StreamFormat& new_format);
+
+  // Feeds data from the jitter buffer into the pending requests. OnDataFilled()
+  // will be called and request will be removed from the queue when a request
+  // has been filled up.
+  void ProcessGetDataRequests();
+
+  // Calculates the number of bytes needed to store audio data of the given
+  // duration based on |stream_format_|.
+  size_t GetBufferSizeFromTime(base::TimeDelta duration) const;
+
+  // Drops audio packets in |queued_packets_| such that the total latency
+  // doesn't exceed |kMaxQueueLatency|.
+  void DropOverrunPackets();
+
+  // The stream format of the last audio packet. This is nullptr if the buffer
+  // has never received any packet.
+  std::unique_ptr<StreamFormat> stream_format_;
+
+  // AudioPackets queued up by the jitter buffer before they are consumed by
+  // GetDataRequests.
+  std::list<std::unique_ptr<AudioPacket>> queued_packets_;
+
+  // Number of bytes that is queued in |queued_packets_|.
+  size_t queued_bytes_ = 0;
+
+  // The byte offset when reading data from the first packet of
+  // |queued_packets_|. Equal to the number of bytes consumed from the first
+  // packet.
+  size_t first_packet_offset_ = 0;
+
+  // Called when the stream format is changed.
+  OnFormatChangedCallback on_format_changed_;
+
+  // GetDataRequests that are not yet fulfilled.
+  std::list<std::unique_ptr<GetDataRequest>> queued_requests_;
+
+  // The buffer will not feed data to the requests if this is true.
+  bool underrun_protection_mode_ = true;
+
+  THREAD_CHECKER(thread_checker_);
+
+  DISALLOW_COPY_AND_ASSIGN(AudioJitterBuffer);
+};
+
+}  // namespace remoting
+
+#endif  // REMOTING_CLIENT_AUDIO_AUDIO_JITTER_BUFFER_H_