Fuchsia: Implement MixerOutputStreamFuchsia using AudioRenderer2.

Previously MixerOutputStreamFuchsia was using media_client library,
which is deprecated now. Update it to use AudioRenderer2 FIDL interface
directly.

Bug: 851733
Change-Id: I72a43369d16ecd626aa7294a6f3500b57bb3731e
Reviewed-on: https://chromium-review.googlesource.com/1100376Reviewed-by: Kenneth MacKay <kmackay@chromium.org>
Reviewed-by: Wez <wez@chromium.org>
Commit-Queue: Sergey Ulanov <sergeyu@chromium.org>
Cr-Commit-Position: refs/heads/master@{#568159}

chromecast/media/cma/backend/fuchsia/BUILD.gn

@@ -40,9 +40,8 @@ cast_source_set("cma_backend_support") {
     "//chromecast/public",
     "//chromecast/public/media",
     "//media",
+    "//third_party/fuchsia-sdk:media",
   ]
-
-  libs = [ "media_client" ]
 }
 
 cast_source_set("unit_tests") {
@@ -56,5 +55,6 @@ cast_source_set("unit_tests") {
   deps = [
     ":cma_backend_support",
     "//testing/gtest",
+    "//third_party/fuchsia-sdk:media",
   ]
 }

chromecast/media/cma/backend/fuchsia/mixer_output_stream_fuchsia.cc

@@ -4,10 +4,11 @@
 
 #include "chromecast/media/cma/backend/fuchsia/mixer_output_stream_fuchsia.h"
 
-#include <media/audio.h>
+#include <fuchsia/media/cpp/fidl.h>
 #include <zircon/syscalls.h>
 
 #include "base/command_line.h"
+#include "base/fuchsia/component_context.h"
 #include "base/time/time.h"
 #include "chromecast/base/chromecast_switches.h"
 #include "media/base/audio_sample_types.h"
@@ -17,10 +18,10 @@
 namespace chromecast {
 namespace media {
 
-// |buffer_size| passed to media_client library when initializing audio output
-// stream. Current implementation ignores this parameter, so the value doesn't
-// make much difference. StreamMixer by default writes chunks of 768 frames.
-constexpr int kDefaultPeriodSize = 768;
+// Target period between Write() calls. It's used to calculate the value
+// returned from OptimalWriteFramesCount().
+constexpr base::TimeDelta kTargetWritePeriod =
+    base::TimeDelta::FromMilliseconds(10);
 
 // Same value as in MixerOutputStreamAlsa. Currently this value is used to
 // simulate blocking Write() similar to ALSA's behavior, see comments in
@@ -32,44 +33,39 @@ std::unique_ptr<MixerOutputStream> MixerOutputStream::Create() {
   return std::make_unique<MixerOutputStreamFuchsia>();
 }
 
-MixerOutputStreamFuchsia::MixerOutputStreamFuchsia() {}
-
-MixerOutputStreamFuchsia::~MixerOutputStreamFuchsia() {
-  if (manager_)
-    fuchsia_audio_manager_free(manager_);
-}
+MixerOutputStreamFuchsia::MixerOutputStreamFuchsia() = default;
+MixerOutputStreamFuchsia::~MixerOutputStreamFuchsia() = default;
 
 bool MixerOutputStreamFuchsia::Start(int requested_sample_rate, int channels) {
-  DCHECK(!stream_);
-
-  if (!manager_)
-    manager_ = fuchsia_audio_manager_create();
-
-  DCHECK(manager_);
-
-  fuchsia_audio_parameters fuchsia_params;
-  fuchsia_params.sample_rate = requested_sample_rate;
-  fuchsia_params.num_channels = channels;
-  fuchsia_params.buffer_size = kDefaultPeriodSize;
-
-  int result = fuchsia_audio_manager_create_output_stream(
-      manager_, nullptr, &fuchsia_params, &stream_);
-  if (result < 0) {
-    LOG(ERROR) << "Failed to open audio output, error code: " << result;
-    DCHECK(!stream_);
-    return false;
-  }
-
-  if (!UpdatePresentationDelay()) {
-    fuchsia_audio_output_stream_free(stream_);
-    stream_ = nullptr;
-    return false;
-  }
+  DCHECK(!audio_renderer_);
+  DCHECK(reference_time_.is_null());
 
   sample_rate_ = requested_sample_rate;
   channels_ = channels;
-
-  started_time_ = base::TimeTicks();
+  target_packet_size_ = ::media::AudioTimestampHelper::TimeToFrames(
+      kTargetWritePeriod, sample_rate_);
+
+  // Connect |audio_renderer_|.
+  fuchsia::media::AudioPtr audio_server =
+      base::fuchsia::ComponentContext::GetDefault()
+          ->ConnectToService<fuchsia::media::Audio>();
+  audio_server->CreateRendererV2(audio_renderer_.NewRequest());
+  audio_renderer_.set_error_handler(
+      fit::bind_member(this, &MixerOutputStreamFuchsia::OnRendererError));
+
+  // Configure the renderer.
+  fuchsia::media::AudioPcmFormat format;
+  format.sample_format = fuchsia::media::AudioSampleFormat::FLOAT;
+  format.channels = channels_;
+  format.frames_per_second = sample_rate_;
+  audio_renderer_->SetPcmFormat(std::move(format));
+
+  // Use number of samples to specify media position.
+  audio_renderer_->SetPtsUnits(sample_rate_, 1);
+
+  audio_renderer_->EnableMinLeadTimeEvents(true);
+  audio_renderer_.events().OnMinLeadTimeChanged =
+      fit::bind_member(this, &MixerOutputStreamFuchsia::OnMinLeadTimeChanged);
 
   return true;
 }
@@ -80,104 +76,143 @@ int MixerOutputStreamFuchsia::GetSampleRate() {
 
 MediaPipelineBackend::AudioDecoder::RenderingDelay
 MixerOutputStreamFuchsia::GetRenderingDelay() {
-  base::TimeTicks now = base::TimeTicks::Now();
-  base::TimeDelta delay =
-      base::TimeDelta::FromMicroseconds(presentation_delay_ns_ / 1000);
-  if (!started_time_.is_null()) {
-    base::TimeTicks stream_time = GetCurrentStreamTime();
-    if (stream_time > now)
-      delay += stream_time - now;
-  }
+  if (reference_time_.is_null())
+    return MediaPipelineBackend::AudioDecoder::RenderingDelay();
 
+  base::TimeTicks now = base::TimeTicks::Now();
+  base::TimeDelta delay = GetCurrentStreamTime() - now;
   return MediaPipelineBackend::AudioDecoder::RenderingDelay(
       /*delay_microseconds=*/delay.InMicroseconds(),
       /*timestamp_microseconds=*/(now - base::TimeTicks()).InMicroseconds());
 }
 
 int MixerOutputStreamFuchsia::OptimalWriteFramesCount() {
-  return kDefaultPeriodSize;
+  return target_packet_size_;
 }
 
 bool MixerOutputStreamFuchsia::Write(const float* data,
                                      int data_size,
                                      bool* out_playback_interrupted) {
-  if (!stream_)
+  if (!audio_renderer_)
     return false;
 
-  DCHECK(data_size % channels_ == 0);
-
-  do {
-    zx_time_t presentation_time = FUCHSIA_AUDIO_NO_TIMESTAMP;
-    if (started_time_.is_null()) {
-      // Presentation time (PTS) needs to be specified only for the first frame
-      // after stream is started or restarted. Mixer will calculate PTS for all
-      // following frames. 1us is added to account for the time passed between
-      // zx_clock_get() and fuchsia_audio_output_stream_write().
-      zx_time_t zx_now = zx_clock_get(ZX_CLOCK_MONOTONIC);
-      presentation_time = zx_now + presentation_delay_ns_ + ZX_USEC(1);
-      started_time_ = base::TimeTicks::FromZxTime(zx_now);
-      stream_position_samples_ = 0;
-    }
-    int result = fuchsia_audio_output_stream_write(
-        stream_, const_cast<float*>(data), data_size, presentation_time);
-    if (result == ZX_ERR_IO_MISSED_DEADLINE) {
-      LOG(ERROR) << "MixerOutputStreamFuchsia::PumpSamples() missed deadline, "
-                    "resetting PTS.";
-      if (!UpdatePresentationDelay()) {
-        return false;
-      }
-      started_time_ = base::TimeTicks();
-      *out_playback_interrupted = true;
-    } else if (result != ZX_OK) {
-      LOG(ERROR) << "fuchsia_audio_output_stream_write() returned " << result;
-      return false;
-    }
+  DCHECK_EQ(data_size % channels_, 0);
 
-  } while (started_time_.is_null());
+  // Allocate payload buffer if necessary.
+  if (!payload_buffer_.mapped_size() && !InitializePayloadBuffer())
+    return false;
+
+  // If Write() was called for the current playback position then assume that
+  // playback was interrupted.
+  auto now = base::TimeTicks::Now();
+  bool playback_interrupted = !reference_time_.is_null() &&
+                              now >= (GetCurrentStreamTime() - min_lead_time_);
+  if (out_playback_interrupted)
+    *out_playback_interrupted = playback_interrupted;
+
+  // Reset playback position if playback was interrupted.
+  if (playback_interrupted)
+    reference_time_ = base::TimeTicks();
+
+  size_t packet_size = data_size * sizeof(float);
+  if (payload_buffer_pos_ + packet_size > payload_buffer_.mapped_size()) {
+    payload_buffer_pos_ = 0;
+  }
+
+  DCHECK_LE(payload_buffer_pos_ + data_size, payload_buffer_.mapped_size());
+  memcpy(reinterpret_cast<uint8_t*>(payload_buffer_.memory()) +
+             payload_buffer_pos_,
+         data, packet_size);
 
+  // Send a new packet.
+  fuchsia::media::AudioPacket packet;
+  packet.timestamp = stream_position_samples_;
+  packet.payload_offset = payload_buffer_pos_;
+  packet.payload_size = packet_size;
+  packet.flags = 0;
+  audio_renderer_->SendPacketNoReply(std::move(packet));
+
+  // Update stream position.
   int frames = data_size / channels_;
   stream_position_samples_ += frames;
-
-  // Block the thread to limit amount of buffered data. Currently
-  // MixerOutputStreamAlsa uses blocking Write() and StreamMixer relies on that
-  // behavior. Sleep() below replicates the same behavior on Fuchsia.
-  // TODO(sergeyu): Refactor StreamMixer to work with non-blocking Write().
-  base::TimeDelta max_buffer_duration =
-      ::media::AudioTimestampHelper::FramesToTime(kMaxOutputBufferSizeFrames,
-                                                  sample_rate_);
-  base::TimeDelta current_buffer_duration =
-      GetCurrentStreamTime() - base::TimeTicks::Now();
-  if (current_buffer_duration > max_buffer_duration)
-    base::PlatformThread::Sleep(current_buffer_duration - max_buffer_duration);
+  payload_buffer_pos_ += packet_size;
+
+  if (reference_time_.is_null()) {
+    reference_time_ = now + min_lead_time_;
+    audio_renderer_->PlayNoReply(reference_time_.ToZxTime(),
+                                 stream_position_samples_ - frames);
+  } else {
+    // Block the thread to limit amount of buffered data. Currently
+    // MixerOutputStreamAlsa uses blocking Write() and StreamMixer relies on
+    // that behavior. Sleep() below replicates the same behavior on Fuchsia.
+    // TODO(sergeyu): Refactor StreamMixer to work with non-blocking Write().
+    base::TimeDelta max_buffer_duration =
+        ::media::AudioTimestampHelper::FramesToTime(kMaxOutputBufferSizeFrames,
+                                                    sample_rate_);
+    base::TimeDelta current_buffer_duration =
+        GetCurrentStreamTime() - min_lead_time_ - now;
+    if (current_buffer_duration > max_buffer_duration) {
+      base::PlatformThread::Sleep(current_buffer_duration -
+                                  max_buffer_duration);
+    }
+  }
 
   return true;
 }
 
 void MixerOutputStreamFuchsia::Stop() {
-  started_time_ = base::TimeTicks();
+  reference_time_ = base::TimeTicks();
+  audio_renderer_.Unbind();
+}
 
-  if (stream_) {
-    fuchsia_audio_output_stream_free(stream_);
-    stream_ = nullptr;
-  }
+size_t MixerOutputStreamFuchsia::GetMinBufferSize() {
+  // Ensure that |payload_buffer_| fits enough packets to cover |min_lead_time_|
+  // and kMaxOutputBufferSizeFrames plus one extra packet.
+  int min_packets = (::media::AudioTimestampHelper::TimeToFrames(min_lead_time_,
+                                                                 sample_rate_) +
+                     kMaxOutputBufferSizeFrames + target_packet_size_ - 1) /
+                        target_packet_size_ +
+                    1;
+  return min_packets * target_packet_size_ * channels_ * sizeof(float);
 }
 
-bool MixerOutputStreamFuchsia::UpdatePresentationDelay() {
-  int result = fuchsia_audio_output_stream_get_min_delay(
-      stream_, &presentation_delay_ns_);
-  if (result != ZX_OK) {
-    LOG(ERROR) << "fuchsia_audio_output_stream_get_min_delay() failed: "
-               << result;
+bool MixerOutputStreamFuchsia::InitializePayloadBuffer() {
+  size_t buffer_size = GetMinBufferSize();
+  if (!payload_buffer_.CreateAndMapAnonymous(buffer_size)) {
+    LOG(WARNING) << "Failed to allocate VMO of size " << buffer_size;
     return false;
   }
 
+  payload_buffer_pos_ = 0;
+  audio_renderer_->SetPayloadBuffer(
+      zx::vmo(payload_buffer_.handle().Duplicate().GetHandle()));
+
   return true;
 }
 
 base::TimeTicks MixerOutputStreamFuchsia::GetCurrentStreamTime() {
-  DCHECK(!started_time_.is_null());
-  return started_time_ + ::media::AudioTimestampHelper::FramesToTime(
-                             stream_position_samples_, sample_rate_);
+  DCHECK(!reference_time_.is_null());
+  return reference_time_ + ::media::AudioTimestampHelper::FramesToTime(
+                               stream_position_samples_, sample_rate_);
+}
+
+void MixerOutputStreamFuchsia::OnRendererError() {
+  LOG(WARNING) << "AudioRenderer has failed.";
+  Stop();
+}
+
+void MixerOutputStreamFuchsia::OnMinLeadTimeChanged(int64_t min_lead_time) {
+  min_lead_time_ = base::TimeDelta::FromNanoseconds(min_lead_time);
+
+  // When min_lead_time_ increases we may need to reallocate |payload_buffer_|.
+  // Code below just unmaps the current buffer. The new buffer will be allocated
+  // lated in PumpSamples(). This is necessary because VMO allocation may fail
+  // and it's not possible to report that error here - OnMinLeadTimeChanged()
+  // may be invoked before Start().
+  if (payload_buffer_.mapped_size() > 0 &&
+      GetMinBufferSize() > payload_buffer_.mapped_size()) {
+    payload_buffer_.Unmap();
+  }
 }
 
 }  // namespace media

chromecast/media/cma/backend/fuchsia/mixer_output_stream_fuchsia.h

@@ -5,8 +5,9 @@
 #ifndef CHROMECAST_MEDIA_CMA_BACKEND_AUDIO_OUTPUT_STREAM_FUCHSIA_H_
 #define CHROMECAST_MEDIA_CMA_BACKEND_AUDIO_OUTPUT_STREAM_FUCHSIA_H_
 
-#include <media/audio.h>
+#include <fuchsia/media/cpp/fidl.h>
 
+#include "base/memory/shared_memory.h"
 #include "base/time/time.h"
 #include "chromecast/public/media/mixer_output_stream.h"
 
@@ -31,21 +32,36 @@ class MixerOutputStreamFuchsia : public MixerOutputStream {
   void Stop() override;
 
  private:
-  bool UpdatePresentationDelay();
+  size_t GetMinBufferSize();
+  bool InitializePayloadBuffer();
+
   base::TimeTicks GetCurrentStreamTime();
 
-  fuchsia_audio_manager* manager_ = nullptr;
-  fuchsia_audio_output_stream* stream_ = nullptr;
+  // Event handlers for |audio_renderer_|.
+  void OnRendererError();
+  void OnMinLeadTimeChanged(int64_t min_lead_time);
 
   int sample_rate_ = 0;
   int channels_ = 0;
 
-  base::TimeTicks started_time_;
+  // Value returned by OptimalWriteFramesCount().
+  int target_packet_size_ = 0;
+
+  // Audio renderer connection.
+  fuchsia::media::AudioRenderer2Ptr audio_renderer_;
+
+  base::SharedMemory payload_buffer_;
+  size_t payload_buffer_pos_ = 0;
+
+  // Set only while stream is playing.
+  base::TimeTicks reference_time_;
+
   int64_t stream_position_samples_ = 0;
 
-  // Total presentation delay for the stream. This value is returned by
-  // fuchsia_audio_output_stream_get_min_delay()
-  zx_duration_t presentation_delay_ns_ = 0;
+  // Current min lead time for the stream. This value is updated by
+  // AudioRenderer::OnMinLeadTimeChanged event. Assume 50ms until we get the
+  // first OnMinLeadTimeChanged event.
+  base::TimeDelta min_lead_time_ = base::TimeDelta::FromMilliseconds(50);
 
   DISALLOW_COPY_AND_ASSIGN(MixerOutputStreamFuchsia);
 };

chromecast/media/cma/backend/fuchsia/mixer_output_stream_fuchsia_unittest.cc

@@ -4,19 +4,99 @@
 
 #include "chromecast/media/cma/backend/fuchsia/mixer_output_stream_fuchsia.h"
 
+#include "base/location.h"
+#include "base/message_loop/message_loop.h"
+#include "base/single_thread_task_runner.h"
+#include "base/threading/thread_task_runner_handle.h"
 #include "testing/gtest/include/gtest/gtest.h"
 
 namespace chromecast {
 namespace media {
 
+constexpr int kSampleRate = 48000;
+constexpr int kNumChannels = 2;
+
 class MixerOutputStreamFuchsiaTest : public ::testing::Test {
  protected:
+  base::MessageLoopForIO message_loop_;
   MixerOutputStreamFuchsia output_;
 };
 
 TEST_F(MixerOutputStreamFuchsiaTest, StartAndStop) {
-  EXPECT_TRUE(output_.Start(48000, 2));
-  EXPECT_EQ(output_.GetSampleRate(), 48000);
+  EXPECT_TRUE(output_.Start(kSampleRate, kNumChannels));
+  EXPECT_EQ(output_.GetSampleRate(), kSampleRate);
+  output_.Stop();
+}
+
+TEST_F(MixerOutputStreamFuchsiaTest, Play1s) {
+  EXPECT_TRUE(output_.Start(kSampleRate, kNumChannels));
+
+  constexpr base::TimeDelta kTestStreamDuration =
+      base::TimeDelta::FromMilliseconds(300);
+  constexpr float kSignalFrequencyHz = 1000;
+
+  auto started = base::TimeTicks::Now();
+
+  int samples_to_play =
+      kSampleRate * kTestStreamDuration / base::TimeDelta::FromSeconds(1);
+  int pos = 0;
+  while (pos < samples_to_play) {
+    std::vector<float> buffer;
+    int num_frames = output_.OptimalWriteFramesCount();
+    buffer.resize(num_frames * kNumChannels);
+    for (int i = 0; i < num_frames; ++i) {
+      float v = sin(2 * M_PI * pos * kSignalFrequencyHz / kSampleRate);
+      for (int c = 0; c < kNumChannels; ++c) {
+        buffer[i * kNumChannels + c] = v;
+      }
+      pos += 1;
+    }
+    bool interrupted = true;
+    EXPECT_TRUE(output_.Write(buffer.data(), buffer.size(), &interrupted));
+
+    // Run message loop to process async events.
+    base::RunLoop().RunUntilIdle();
+  }
+
+  auto ended = base::TimeTicks::Now();
+
+  // Verify that Write() was blocking, allowing 100ms for buffering.
+  EXPECT_GT(ended - started,
+            kTestStreamDuration - base::TimeDelta::FromMilliseconds(100));
+
+  output_.Stop();
+}
+
+TEST_F(MixerOutputStreamFuchsiaTest, PlaybackInterrupted) {
+  EXPECT_TRUE(output_.Start(kSampleRate, kNumChannels));
+
+  std::vector<float> buffer;
+  int num_frames = output_.OptimalWriteFramesCount();
+  buffer.resize(num_frames * kNumChannels);
+
+  bool interrupted = true;
+
+  // First Write() always returns interrupted = false.
+  EXPECT_TRUE(output_.Write(buffer.data(), buffer.size(), &interrupted));
+  EXPECT_FALSE(interrupted);
+
+  interrupted = true;
+  // Repeated Write() is expected to return interrupted = false.
+  EXPECT_TRUE(output_.Write(buffer.data(), buffer.size(), &interrupted));
+  EXPECT_FALSE(interrupted);
+
+  // Run message loop for 100ms before calling Write() again.
+  base::RunLoop run_loop;
+  base::ThreadTaskRunnerHandle::Get()->PostDelayedTask(
+      FROM_HERE, run_loop.QuitClosure(),
+      base::TimeDelta::FromMilliseconds(100));
+  run_loop.Run();
+
+  // Write() is called to late, expect interrupted = true.
+  interrupted = false;
+  EXPECT_TRUE(output_.Write(buffer.data(), buffer.size(), &interrupted));
+  EXPECT_TRUE(interrupted);
+
   output_.Stop();
 }
 

chromecast/media/cma/backend/stream_mixer.cc

@@ -224,6 +224,10 @@ StreamMixer::StreamMixer(
   if (mixer_thread_) {
     base::Thread::Options options;
     options.priority = base::ThreadPriority::REALTIME_AUDIO;
+#if defined(OS_FUCHSIA)
+    // MixerOutputStreamFuchsia uses FIDL, which works only on IO threads.
+    options.message_loop_type = base::MessageLoop::TYPE_IO;
+#endif
     mixer_thread_->StartWithOptions(options);
     mixer_task_runner_ = mixer_thread_->task_runner();
     mixer_task_runner_->PostTask(FROM_HERE, base::BindOnce(&UseHighPriority));