PulseAudio Sound Playback on Linux

media/audio/linux/pulse_output.cc

+// Copyright (c) 2011 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 "media/audio/linux/pulse_output.h"
+
+#include "media/audio/audio_util.h"
+#include "media/audio/linux/audio_manager_linux.h"
+#include "media/base/data_buffer.h"
+#include "media/base/seekable_buffer.h"
+
+static pa_sample_format_t BitsToFormat(int bits_per_sample) {
+  switch (bits_per_sample) {
+    // Unsupported sample formats shown for reference.  I am assuming we want
+    // signed and little endian because that is what we gave to ALSA.
+    case 8:
+      return PA_SAMPLE_U8;
+      // Also 8-bits: PA_SAMPLE_ALAW and PA_SAMPLE_ULAW
+    case 16:
+      return PA_SAMPLE_S16LE;
+      // Also 16-bits: PA_SAMPLE_S16BE (big endian).
+    case 24:
+      return PA_SAMPLE_S24LE;
+      // Also 24-bits: PA_SAMPLE_S24BE (big endian).
+      // Other cases: PA_SAMPLE_24_32LE (in LSB of 32-bit field, little endian),
+      //          and PA_SAMPLE_24_32BE (in LSB of 32-bit field, big endian),
+    case 32:
+      return PA_SAMPLE_S32LE;
+      // Also 32-bits: PA_SAMPLE_S32BE (big endian),
+      //               PA_SAMPLE_FLOAT32LE (floating point little endian),
+      //           and PA_SAMPLE_FLOAT32BE (floating point big endian).
+    default:
+      return PA_SAMPLE_INVALID;
+  }
+}
+
+static size_t MicrosecondsToBytes(
+    uint32 microseconds, uint32 sample_rate, size_t bytes_per_frame) {
+  return microseconds * sample_rate * bytes_per_frame /
+      base::Time::kMicrosecondsPerSecond;
+}
+
+void PulseAudioOutputStream::ContextStateCallback(pa_context* context,
+                                                  void* state_addr) {
+  pa_context_state_t* state = static_cast<pa_context_state_t*>(state_addr);
+  *state = pa_context_get_state(context);
+}
+
+void PulseAudioOutputStream::WriteRequestCallback(
+    pa_stream* playback_handle, size_t length, void* stream_addr) {
+  PulseAudioOutputStream* stream =
+      static_cast<PulseAudioOutputStream*>(stream_addr);
+
+  DCHECK_EQ(stream->message_loop_, MessageLoop::current());
+
+  stream->write_callback_handled_ = true;
+
+  // Fulfill write request.
+  stream->FulfillWriteRequest(length);
+
+  // Continue playback.
+  stream->message_loop_->PostTask(
+      FROM_HERE,
+      stream->method_factory_.NewRunnableMethod(
+          &PulseAudioOutputStream::WaitForWriteRequest));
+}
+
+PulseAudioOutputStream::PulseAudioOutputStream(const AudioParameters& params,
+                                               AudioManagerLinux* manager,
+                                               MessageLoop* message_loop)
+      : channel_layout_(params.channel_layout),
+        channel_count_(ChannelLayoutToChannelCount(channel_layout_)),
+        sample_format_(BitsToFormat(params.bits_per_sample)),
+        sample_rate_(params.sample_rate),
+        bytes_per_frame_(params.channels * params.bits_per_sample / 8),
+        manager_(manager),
+        pa_context_(NULL),
+        pa_mainloop_(NULL),
+        playback_handle_(NULL),
+        packet_size_(params.GetPacketSize()),
+        frames_per_packet_(packet_size_ / bytes_per_frame_),
+        client_buffer_(NULL),
+        volume_(1.0f),
+        stream_stopped_(true),
+        write_callback_handled_(false),
+        message_loop_(message_loop),
+        ALLOW_THIS_IN_INITIALIZER_LIST(method_factory_(this)),
+        source_callback_(NULL) {
+  DCHECK_EQ(message_loop_, MessageLoop::current());
+
+  DCHECK(manager_);
+
+  // TODO(slock): Sanity check input values.
+}
+
+PulseAudioOutputStream::~PulseAudioOutputStream() {
+  // All internal structures should already have been freed in Close(),
+  // which calls AudioManagerLinux::Release which deletes this object.
+   DCHECK(!playback_handle_);
+   DCHECK(!pa_context_);
+   DCHECK(!pa_mainloop_);
+}
+
+bool PulseAudioOutputStream::Open() {
+  DCHECK_EQ(message_loop_, MessageLoop::current());
+
+  // TODO(slock): Possibly move most of this to a OpenPlaybackDevice function in
+  // a new class 'pulse_util', like alsa_util.
+
+  // Create a mainloop API and connect to the default server.
+  pa_mainloop_ = pa_mainloop_new();
+  pa_mainloop_api* pa_mainloop_api = pa_mainloop_get_api(pa_mainloop_);
+  pa_context_ = pa_context_new(pa_mainloop_api, "Chromium");
+  pa_context_state_t pa_context_state = PA_CONTEXT_UNCONNECTED;
+  pa_context_connect(pa_context_, NULL, PA_CONTEXT_NOFLAGS, NULL);
+
+  // Wait until PulseAudio is ready.
+  pa_context_set_state_callback(pa_context_, &ContextStateCallback,
+                                &pa_context_state);
+  while (pa_context_state != PA_CONTEXT_READY) {
+    pa_mainloop_iterate(pa_mainloop_, 1, NULL);
+    if (pa_context_state == PA_CONTEXT_FAILED ||
+        pa_context_state == PA_CONTEXT_TERMINATED) {
+      stream_stopped_ = true;
+      Reset();
+      return false;
+    }
+  }
+
+  // Set sample specifications and open playback stream.
+  pa_sample_spec pa_sample_specifications;
+  pa_sample_specifications.format = sample_format_;
+  pa_sample_specifications.rate = sample_rate_;
+  pa_sample_specifications.channels = channel_count_;
+  playback_handle_ = pa_stream_new(pa_context_, "Playback",
+                                   &pa_sample_specifications, NULL);
+
+  // Initialize client buffer.
+  uint32 output_packet_size = frames_per_packet_ * bytes_per_frame_;
+  client_buffer_.reset(new media::SeekableBuffer(0, output_packet_size));
+
+  // Set write callback.
+  pa_stream_set_write_callback(playback_handle_, &WriteRequestCallback,
+                               this);
+
+  // Set server-side buffer attributes.
+  // (uint32_t)-1 is the default and recommended value from PulseAudio's
+  // documentation, found at:
+  // http://freedesktop.org/software/pulseaudio/doxygen/structpa__buffer__attr.html.
+  pa_buffer_attr pa_buffer_attributes;
+  pa_buffer_attributes.maxlength = static_cast<uint32_t>(-1);
+  pa_buffer_attributes.tlength = output_packet_size;
+  pa_buffer_attributes.prebuf = static_cast<uint32_t>(-1);
+  pa_buffer_attributes.minreq = static_cast<uint32_t>(-1);
+  pa_buffer_attributes.fragsize = static_cast<uint32_t>(-1);
+
+  // Connect playback stream.
+  pa_stream_connect_playback(playback_handle_, NULL,
+                             &pa_buffer_attributes,
+                             (pa_stream_flags_t)
+                             (PA_STREAM_INTERPOLATE_TIMING |
+                             PA_STREAM_ADJUST_LATENCY |
+                             PA_STREAM_AUTO_TIMING_UPDATE),
+                             NULL, NULL);
+
+  if (!playback_handle_) {
+    stream_stopped_ = true;
+    Reset();
+    return false;
+  }
+
+  return true;
+}
+
+void PulseAudioOutputStream::Reset() {
+  // Close the stream.
+  if (playback_handle_) {
+    pa_stream_flush(playback_handle_, NULL, NULL);
+    pa_stream_disconnect(playback_handle_);
+
+    // Release PulseAudio structures.
+    pa_stream_unref(playback_handle_);
+    playback_handle_ = NULL;
+  }
+  if (pa_context_) {
+    pa_context_unref(pa_context_);
+    pa_context_ = NULL;
+  }
+  if (pa_mainloop_) {
+    pa_mainloop_free(pa_mainloop_);
+    pa_mainloop_ = NULL;
+  }
+  // |pa_mainloop_api| is freed with |pa_mainloop_|.
+
+  // Release internal buffer.
+  client_buffer_.reset();
+}
+
+void PulseAudioOutputStream::Close() {
+  DCHECK_EQ(message_loop_, MessageLoop::current());
+
+  Reset();
+
+  // Signal to the manager that we're closed and can be removed.
+  // This should be the last call in the function as it deletes "this".
+  manager_->ReleaseOutputStream(this);
+}
+
+void PulseAudioOutputStream::WaitForWriteRequest() {
+  DCHECK_EQ(message_loop_, MessageLoop::current());
+
+  // Iterate the PulseAudio mainloop.  If the stream isn't stopped or PulseAudio
+  // doesn't request a write, post a task to iterate the mainloop again.
+  write_callback_handled_ = false;
+  pa_mainloop_iterate(pa_mainloop_, 1, NULL);
+  if (!write_callback_handled_ && !stream_stopped_) {
+    message_loop_->PostTask(
+        FROM_HERE,
+        method_factory_.NewRunnableMethod(
+            &PulseAudioOutputStream::WaitForWriteRequest));
+  }
+}
+
+bool PulseAudioOutputStream::BufferPacketFromSource() {
+  uint32 buffer_delay = client_buffer_->forward_bytes();
+  pa_usec_t pa_latency_micros;
+  int negative;
+  pa_stream_get_latency(playback_handle_, &pa_latency_micros, &negative);
+  uint32 hardware_delay = MicrosecondsToBytes(pa_latency_micros,
+                                              sample_rate_,
+                                              bytes_per_frame_);
+  // TODO(slock): Deal with negative latency (negative == 1).  This has yet
+  // to happen in practice though.
+  scoped_refptr<media::DataBuffer> packet =
+       new media::DataBuffer(packet_size_);

[vrk (LEFT CHROMIUM), 2011/08/17 18:59:23]

nit: indentation

[slock, 2011/08/17 21:43:14]

Done.

+  size_t packet_size = RunDataCallback(packet->GetWritableData(),
+                                       packet->GetBufferSize(),
+                                       AudioBuffersState(buffer_delay,
+                                                         hardware_delay));
+
+  if (packet_size > 0) {

[vrk (LEFT CHROMIUM), 2011/08/17 18:59:23]

nit: a little nicer to say
if (packet_size == 0)
  return false;

then continue onward for the "good" case

[slock, 2011/08/17 21:43:14]

Done.

+    // TODO(slock): Swizzling and downmixing.
+    media::AdjustVolume(packet->GetWritableData(),
+                        packet_size,
+                        channel_count_,
+                        bytes_per_frame_ / channel_count_,
+                        volume_);
+    packet->SetDataSize(packet_size);
+    // Add the packet to the buffer.
+    client_buffer_->Append(packet);
+    return true;
+  }
+  return false;
+}
+
+void PulseAudioOutputStream::FulfillWriteRequest(size_t requested_bytes) {
+  // If we have enough data to fulfill the request, we can finish the write.
+  if (!stream_stopped_) {

[vrk (LEFT CHROMIUM), 2011/08/17 18:59:23]

nit: little cleaner to say 
if (stream_stopped_)
  return;

then continue onward

[slock, 2011/08/17 21:43:14]

Done.

+    if  (client_buffer_->forward_bytes() >= requested_bytes) {

[vrk (LEFT CHROMIUM), 2011/08/17 18:59:23]

Lines 259 - 282 can be simplified like follows:

// Request more data from the source until we can fulfill the request or
// fail to receive anymore data.
bool buffering_successful = true;
while (client_buffer_->forward_bytes() < requested_bytes &&
buffering_successful) 
  buffering_successful = BufferPacketFromSource();

WriteToStream(requested_bytes);

if (!buffering_successful)
  message_loop_->PostTask(...);

[slock, 2011/08/17 21:43:14]

I modified this slightly because it subtly implied a change of the behavior of
WriteToStream().  You are correct in implying that users (and this code) needs
to be able to call WriteToStream()num_bytes) without being under the guarantee
that num_bytes can and will be written.  But, if we are going to call
WriteToStream(requested_bytes) without knowing that we will actually write
exactly that many bytes, we need to know how many bytes it did write in order to
correctly post another request for the remaining bytes.  So, I made
WriteToStream return how many bytes out of bytes_to_write it actually managed to
write.  I also only call WriteToStream if there are any bytes in the buffer to
write.

+      WriteToStream(requested_bytes);
+      return;
+    } else {
+      // Request more data from the source until we can fulfill the request or
+      // fail to receive anymore data.
+      while (BufferPacketFromSource()) {
+        if (client_buffer_->forward_bytes() >= requested_bytes) {
+          WriteToStream(requested_bytes);
+          return;
+        }
+      }
+      // We could not fulfill the write request at this time.  We will write as
+      // much as we can.
+      WriteToStream(client_buffer_->forward_bytes());
+
+      // Try to fulfill the rest of the request later.
+      message_loop_->PostTask(
+          FROM_HERE,
+          method_factory_.NewRunnableMethod(
+              &PulseAudioOutputStream::FulfillWriteRequest,
+              requested_bytes - client_buffer_->forward_bytes()));
+    }
+  }
+}
+
+void PulseAudioOutputStream::WriteToStream(size_t bytes_to_write) {
+  // This while loop will not stall because this function is only called when we

[vrk (LEFT CHROMIUM), 2011/08/17 18:59:23]

" This while loop will not stall because this function is only called ..."

Still dangerous! The goal is to write maintainable, easy-to-use and
hard-to-misuse code. (You will not be the only reader/writer of this class,
etc.) This function is very easy to misuse...

Looks like GetCurrentChunk returns false when there is no data left in the
buffer. Use this value to break out of the loop when you've exhausted your
buffer.

[slock, 2011/08/17 21:43:14]

Done.

+  // have enough bytes buffered to fulfill the request.
+  size_t bytes_written = 0;
+  while (bytes_written < bytes_to_write) {
+    // Try to get data to write.
+    const uint8* chunk;
+    size_t chunk_size;
+    client_buffer_->GetCurrentChunk(&chunk, &chunk_size);
+
+    pa_stream_write(playback_handle_, chunk, chunk_size,
+                    NULL, 0LL, PA_SEEK_RELATIVE);
+    client_buffer_->Seek(chunk_size);
+    bytes_written += chunk_size;
+  }
+}
+
+void PulseAudioOutputStream::Start(AudioSourceCallback* callback) {
+  DCHECK_EQ(message_loop_, MessageLoop::current());
+
+  CHECK(callback);
+  source_callback_ = callback;
+
+  // Clear buffer, it might still have data in it.
+  client_buffer_->Clear();
+  stream_stopped_ = false;
+
+  // Start playback.
+  message_loop_->PostTask(
+      FROM_HERE,
+      method_factory_.NewRunnableMethod(
+          &PulseAudioOutputStream::WaitForWriteRequest));
+}
+
+void PulseAudioOutputStream::Stop() {
+  DCHECK_EQ(message_loop_, MessageLoop::current());
+
+  stream_stopped_ = true;
+}
+
+void PulseAudioOutputStream::SetVolume(double volume) {
+  DCHECK_EQ(message_loop_, MessageLoop::current());
+
+  volume_ = static_cast<float>(volume);
+}
+
+void PulseAudioOutputStream::GetVolume(double* volume) {
+  DCHECK_EQ(message_loop_, MessageLoop::current());
+
+  *volume = volume_;
+}
+
+uint32 PulseAudioOutputStream::RunDataCallback(
+    uint8* dest, uint32 max_size, AudioBuffersState buffers_state) {
+  if (source_callback_)
+    return source_callback_->OnMoreData(this, dest, max_size, buffers_state);
+
+  return 0;
+}