Get PulseAudio implementation working.

media/audio/pulse/pulse_output.cc

Index: media/audio/pulse/pulse_output.cc
diff --git a/media/audio/pulse/pulse_output.cc b/media/audio/pulse/pulse_output.cc
index bdd29c00c6a5c6455e8a6bf1f49c47daacf99a86..fee317d5b47d5c3e942832329db2f9c17ea75c5a 100644
--- a/media/audio/pulse/pulse_output.cc
+++ b/media/audio/pulse/pulse_output.cc
@@ -4,8 +4,6 @@
 
 #include "media/audio/pulse/pulse_output.h"
 
-#include "base/bind.h"
-#include "base/message_loop.h"
 #include "media/audio/audio_parameters.h"
 #include "media/audio/audio_util.h"
 #if defined(OS_LINUX)
@@ -13,8 +11,6 @@
 #elif defined(OS_OPENBSD)
 #include "media/audio/openbsd/audio_manager_openbsd.h"
 #endif
-#include "media/base/data_buffer.h"
-#include "media/base/seekable_buffer.h"
 
 namespace media {
 
@@ -116,48 +112,44 @@ static size_t MicrosecondsToBytes(
 
 // static
 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* p_this) {
+  // is pulse giving us callbacks for all contexts?
+  PulseAudioOutputStream* stream = static_cast<PulseAudioOutputStream*>(p_this);
+  stream->context_state_ = pa_context_get_state(stream->pa_context_);
 }
 
 // static
-void PulseAudioOutputStream::WriteRequestCallback(pa_stream* playback_handle,
-                                                  size_t length,
-                                                  void* stream_addr) {
-  PulseAudioOutputStream* stream =
-      reinterpret_cast<PulseAudioOutputStream*>(stream_addr);
-
-  DCHECK(stream->manager_->GetMessageLoop()->BelongsToCurrentThread());
-
-  stream->write_callback_handled_ = true;
+void PulseAudioOutputStream::StreamStateCallback(pa_stream* stream,
+                                                 void* p_this) {
+  // is pulse giving us callbacks for all streams?
+  PulseAudioOutputStream* stream_ptr =
+      static_cast<PulseAudioOutputStream*>(p_this);
+  stream_ptr->stream_state_ = pa_stream_get_state(stream_ptr->playback_handle_);
+}
 
-  // Fulfill write request.
+// static
+void PulseAudioOutputStream::WriteRequestCallback(pa_stream* playback_handle,
+                                                  size_t length, void* p_this) {
+  // Fulfill write request; must always result in a pa_stream_write() call.
+  PulseAudioOutputStream* stream = static_cast<PulseAudioOutputStream*>(p_this);
   stream->FulfillWriteRequest(length);
 }
 
 PulseAudioOutputStream::PulseAudioOutputStream(const AudioParameters& params,
                                                AudioManagerPulse* manager)
-    : channel_layout_(params.channel_layout()),
-      channel_count_(ChannelLayoutToChannelCount(channel_layout_)),
-      sample_format_(BitsToPASampleFormat(params.bits_per_sample())),
-      sample_rate_(params.sample_rate()),
-      bytes_per_frame_(params.GetBytesPerFrame()),
+    : params_(params),
       manager_(manager),
       pa_context_(NULL),
       pa_mainloop_(NULL),
       playback_handle_(NULL),
-      packet_size_(params.GetBytesPerBuffer()),
-      frames_per_packet_(packet_size_ / bytes_per_frame_),
-      client_buffer_(NULL),
       volume_(1.0f),
       stream_stopped_(true),
-      write_callback_handled_(false),
-      ALLOW_THIS_IN_INITIALIZER_LIST(weak_factory_(this)),
       source_callback_(NULL) {
   DCHECK(manager_->GetMessageLoop()->BelongsToCurrentThread());
 
-  // TODO(slock): Sanity check input values.
+  CHECK(params_.IsValid());
+  audio_bus_ = AudioBus::Create(params_);
+  interleaved_audio_data_.reset(new uint8[params_.GetBytesPerBuffer()]);
 }
 
 PulseAudioOutputStream::~PulseAudioOutputStream() {
@@ -171,38 +163,44 @@ PulseAudioOutputStream::~PulseAudioOutputStream() {
 bool PulseAudioOutputStream::Open() {
   DCHECK(manager_->GetMessageLoop()->BelongsToCurrentThread());
 
-  // TODO(slock): Possibly move most of this to an 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_mainloop_ = pa_threaded_mainloop_new();
+  CHECK(pa_mainloop_);
+
+  pa_mainloop_api* pa_mainloop_api = pa_threaded_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);
+  CHECK(pa_context_);
+
+  context_state_ = PA_CONTEXT_UNCONNECTED;
+  pa_context_set_state_callback(pa_context_, &ContextStateCallback, this);
+  pa_context_connect(pa_context_, NULL, PA_CONTEXT_NOAUTOSPAWN, NULL);
+
+  pa_threaded_mainloop_start(pa_mainloop_);
 
   // 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) {
+  while (context_state_ != PA_CONTEXT_READY) {

[scherkus (not reviewing), 2012/10/10 17:56:29]

instead of the volatile funny business and state callbacks, it looks like the PA
docs recommend using the PA locking + signalling APIs to convert async
operations into sync operations:
http://freedesktop.org/software/pulseaudio/doxygen/threaded_mainloop.html#cb_sec

Can something like the following work?

pa_threaded_mainloop_lock(pa_mainloop_);
while (pa_context_get_state(pa_context_) != PA_CONTEXT_READY) {
  // blah blah blah
  pa_threaded_mainloop_wait(pa_mainloop_);
}
pa_threaded_mainloop_unlock(pa_mainloop_);

[DaleCurtis, 2012/10/10 18:19:05]

Nice, didn't see that. Will convert. Even more code to delete :)

+    if (context_state_ == PA_CONTEXT_FAILED ||
+        context_state_ == PA_CONTEXT_TERMINATED) {
       Reset();
       return false;
     }
+    // Yukka yuk, context_state_ will be updated in the background.
+    // TODO(dalecurtis): Change this to a waitable event.
+    base::PlatformThread::YieldCurrentThread();
   }
 
   // Set sample specifications.
   pa_sample_spec pa_sample_specifications;
-  pa_sample_specifications.format = sample_format_;
-  pa_sample_specifications.rate = sample_rate_;
-  pa_sample_specifications.channels = channel_count_;
+  pa_sample_specifications.format = BitsToPASampleFormat(
+      params_.bits_per_sample());
+  pa_sample_specifications.rate = params_.sample_rate();
+  pa_sample_specifications.channels = params_.channels();
 
   // Get channel mapping and open playback stream.
+  // TODO(dalecurtis): Is this section correct?
   pa_channel_map* map = NULL;
   pa_channel_map source_channel_map = ChannelLayoutToPAChannelMap(
-      channel_layout_);
+      params_.channel_layout());
   if (source_channel_map.channels != 0) {
     // The source data uses a supported channel map so we will use it rather
     // than the default channel map (NULL).
@@ -210,23 +208,22 @@ bool PulseAudioOutputStream::Open() {
   }
   playback_handle_ = pa_stream_new(pa_context_, "Playback",
                                    &pa_sample_specifications, map);
+  if (!playback_handle_) {
+    Reset();
+    return false;
+  }
 
-  // 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.
+  // Setup callbacks.
+  stream_state_ = PA_STREAM_READY;
+  pa_stream_set_state_callback(playback_handle_, &StreamStateCallback, this);
   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.
+  // Tell pulse audio we only want callbacks of a certain size.
   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.maxlength = params_.GetBytesPerBuffer();
+  pa_buffer_attributes.tlength = params_.GetBytesPerBuffer();
+  pa_buffer_attributes.prebuf = params_.GetBytesPerBuffer();
+  pa_buffer_attributes.minreq = params_.GetBytesPerBuffer();
   pa_buffer_attributes.fragsize = static_cast<uint32_t>(-1);
 
   // Connect playback stream.
@@ -243,32 +240,50 @@ bool PulseAudioOutputStream::Open() {
     return false;
   }
 
+  while (stream_state_ != PA_STREAM_READY) {

[scherkus (not reviewing), 2012/10/10 17:56:29]

ditto

+    if (stream_state_ == PA_STREAM_FAILED) {
+      Reset();
+      return false;
+    }
+    // Yukka yuk, stream_state_ will be updated in the background.
+    // TODO(dalecurtis): Change this to a waitable event.
+    base::PlatformThread::YieldCurrentThread();
+  }
+
   return true;
 }
 
 void PulseAudioOutputStream::Reset() {
   stream_stopped_ = true;
 
+  if (pa_mainloop_)
+    pa_threaded_mainloop_lock(pa_mainloop_);
+
   // Close the stream.
   if (playback_handle_) {

[scherkus (not reviewing), 2012/10/10 17:56:29]

can we make stronger guarantees over which objects should be alive?

for example, it seems the lifetime of pa_mainloop_ > pa_context_ >
playback_handle_, which means we should be able to use early returns + DCHECKs

[DaleCurtis, 2012/10/10 18:19:05]

I can add a if (!pa_mainloop) return early. Is that what you're thinking?
playback_handle_ may or may not have been created. I'm also going to change the
CHECK()'s in Open to if (!) return false, so pa_context_ may or may not be there
either.

+    pa_stream_set_state_callback(playback_handle_, NULL, NULL);
     pa_stream_flush(playback_handle_, NULL, NULL);
-    pa_stream_disconnect(playback_handle_);
 
     // Release PulseAudio structures.
+    pa_stream_disconnect(playback_handle_);
     pa_stream_unref(playback_handle_);
     playback_handle_ = NULL;
   }
+
   if (pa_context_) {
+    pa_context_disconnect(pa_context_);
     pa_context_unref(pa_context_);
     pa_context_ = NULL;
   }
+
+  if (pa_mainloop_)
+    pa_threaded_mainloop_unlock(pa_mainloop_);
+
   if (pa_mainloop_) {
-    pa_mainloop_free(pa_mainloop_);
+    pa_threaded_mainloop_stop(pa_mainloop_);
+    pa_threaded_mainloop_free(pa_mainloop_);
     pa_mainloop_ = NULL;
   }
-
-  // Release internal buffer.
-  client_buffer_.reset();
 }
 
 void PulseAudioOutputStream::Close() {
@@ -281,112 +296,59 @@ void PulseAudioOutputStream::Close() {
   manager_->ReleaseOutputStream(this);
 }
 
-void PulseAudioOutputStream::WaitForWriteRequest() {
-  DCHECK(manager_->GetMessageLoop()->BelongsToCurrentThread());
-
-  if (stream_stopped_)
-    return;
-
-  // Iterate the PulseAudio mainloop.  If 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_) {
-    manager_->GetMessageLoop()->PostTask(FROM_HERE, base::Bind(
-        &PulseAudioOutputStream::WaitForWriteRequest,
-        weak_factory_.GetWeakPtr()));
-  }
-}
-
-bool PulseAudioOutputStream::BufferPacketFromSource() {
-  uint32 buffer_delay = client_buffer_->forward_bytes();
-  pa_usec_t pa_latency_micros;
-  int negative;
+int PulseAudioOutputStream::FillBuffer() {
+  int negative = 0;
+  pa_usec_t pa_latency_micros = 0;
   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
+  uint32 hardware_delay = MicrosecondsToBytes(
+      pa_latency_micros, params_.sample_rate(), params_.GetBytesPerFrame());
+
+  // TODO(dalecurtis): Deal with negative latency (negative == 1).  This has yet
   // to happen in practice though.
-  scoped_refptr<media::DataBuffer> packet =
-      new media::DataBuffer(packet_size_);
+  DCHECK(!negative);
+
   int frames_filled = RunDataCallback(
-      audio_bus_.get(), AudioBuffersState(buffer_delay, hardware_delay));
-  size_t packet_size = frames_filled * bytes_per_frame_;
+      audio_bus_.get(), AudioBuffersState(0, hardware_delay));
+
+  int packet_size = frames_filled * params_.GetBytesPerFrame();
+  DCHECK_LE(packet_size, params_.GetBytesPerBuffer());
+
+  if (packet_size == 0)
+    return 0;
 
-  DCHECK_LE(packet_size, packet_size_);
   // Note: If this ever changes to output raw float the data must be clipped and
   // sanitized since it may come from an untrusted source such as NaCl.
   audio_bus_->ToInterleaved(
-      frames_filled, bytes_per_frame_ / channel_count_,
-      packet->GetWritableData());
-
-  if (packet_size == 0)
-    return false;
+      frames_filled, params_.GetBytesPerFrame() / params_.channels(),
+      interleaved_audio_data_.get());
 
-  media::AdjustVolume(packet->GetWritableData(),
+  media::AdjustVolume(interleaved_audio_data_.get(),
                       packet_size,
-                      channel_count_,
-                      bytes_per_frame_ / channel_count_,
+                      params_.channels(),
+                      params_.GetBytesPerFrame() / params_.channels(),
                       volume_);
-  packet->SetDataSize(packet_size);
-  // Add the packet to the buffer.
-  client_buffer_->Append(packet);
-  return true;
+  return packet_size;
 }
 
 void PulseAudioOutputStream::FulfillWriteRequest(size_t requested_bytes) {
-  // If we have enough data to fulfill the request, we can finish the write.
-  if (stream_stopped_)
-    return;
-
-  // Request more data from the source until we can fulfill the request or
-  // fail to receive anymore data.
-  bool buffering_successful = true;
-  size_t forward_bytes = static_cast<size_t>(client_buffer_->forward_bytes());
-  while (forward_bytes < requested_bytes && buffering_successful) {
-    buffering_successful = BufferPacketFromSource();
-  }
-
-  size_t bytes_written = 0;
-  if (client_buffer_->forward_bytes() > 0) {
-    // Try to fulfill the request by writing as many of the requested bytes to
-    // the stream as we can.
-    WriteToStream(requested_bytes, &bytes_written);
-  }
+  int bytes_available = params_.GetBytesPerBuffer();
 
-  if (bytes_written < requested_bytes) {
-    // We weren't able to buffer enough data to fulfill the request.  Try to
-    // fulfill the rest of the request later.
-    manager_->GetMessageLoop()->PostTask(FROM_HERE, base::Bind(
-        &PulseAudioOutputStream::FulfillWriteRequest,
-        weak_factory_.GetWeakPtr(),
-        requested_bytes - bytes_written));
+  // If we have enough data to fulfill the request, we can finish the write.
+  if (stream_stopped_ || !source_callback_) {
+    memset(interleaved_audio_data_.get(), 0, params_.GetBytesPerBuffer());
   } else {
-    // Continue playback.
-    manager_->GetMessageLoop()->PostTask(FROM_HERE, base::Bind(
-        &PulseAudioOutputStream::WaitForWriteRequest,
-        weak_factory_.GetWeakPtr()));
-  }
-}
+    CHECK_EQ(requested_bytes, static_cast<size_t>(
+        audio_bus_->frames() * params_.GetBytesPerFrame()));
 
-void PulseAudioOutputStream::WriteToStream(size_t bytes_to_write,
-                                           size_t* bytes_written) {
-  *bytes_written = 0;
-  while (*bytes_written < bytes_to_write) {
-    const uint8* chunk;
-    int chunk_size;
-
-    // Stop writing if there is no more data available.
-    if (!client_buffer_->GetCurrentChunk(&chunk, &chunk_size))
-      break;
-
-    // Write data to stream.
-    pa_stream_write(playback_handle_, chunk, chunk_size,
-                    NULL, 0LL, PA_SEEK_RELATIVE);
-    client_buffer_->Seek(chunk_size);
-    *bytes_written += chunk_size;
+    int bytes_available = FillBuffer();
+    if (bytes_available <= 0) {
+      memset(interleaved_audio_data_.get(), 0, params_.GetBytesPerBuffer());
+      bytes_available = params_.GetBytesPerBuffer();
+    }
   }
+
+  pa_stream_write(playback_handle_, interleaved_audio_data_.get(),
+                  bytes_available, NULL, 0LL, PA_SEEK_RELATIVE);
 }
 
 void PulseAudioOutputStream::Start(AudioSourceCallback* callback) {
@@ -398,15 +360,7 @@ void PulseAudioOutputStream::Start(AudioSourceCallback* callback) {
     return;
 
   source_callback_ = callback;
-
-  // Clear buffer, it might still have data in it.
-  client_buffer_->Clear();
   stream_stopped_ = false;
-
-  // Start playback.
-  manager_->GetMessageLoop()->PostTask(FROM_HERE, base::Bind(
-      &PulseAudioOutputStream::WaitForWriteRequest,
-      weak_factory_.GetWeakPtr()));
 }
 
 void PulseAudioOutputStream::Stop() {