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..f4e61d03a83ad7abf1a0695da782eb2e1097689c 100644
--- a/media/audio/pulse/pulse_output.cc
+++ b/media/audio/pulse/pulse_output.cc
@@ -4,41 +4,24 @@
 
 #include "media/audio/pulse/pulse_output.h"
 
-#include "base/bind.h"
-#include "base/message_loop.h"
+#include "media/audio/audio_manager_base.h"
 #include "media/audio/audio_parameters.h"
 #include "media/audio/audio_util.h"
-#if defined(OS_LINUX)
-#include "media/audio/linux/audio_manager_linux.h"
-#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 {
 
 static pa_sample_format_t BitsToPASampleFormat(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:
+      NOTREACHED() << "Invalid bits per sample: " << bits_per_sample;
       return PA_SAMPLE_INVALID;
   }
 }
@@ -73,9 +56,10 @@ static pa_channel_position ChromiumToPAChannelPosition(Channels channel) {
       return PA_CHANNEL_POSITION_SIDE_RIGHT;
     case CHANNELS_MAX:
       return PA_CHANNEL_POSITION_INVALID;
+    default:
+      NOTREACHED() << "Invalid channel: " << channel;
+      return PA_CHANNEL_POSITION_INVALID;
   }
-  NOTREACHED() << "Invalid channel " << channel;
-  return PA_CHANNEL_POSITION_INVALID;
 }
 
 static pa_channel_map ChannelLayoutToPAChannelMap(
@@ -108,56 +92,27 @@ static pa_channel_map ChannelLayoutToPAChannelMap(
   return channel_map;
 }
 
-static size_t MicrosecondsToBytes(
-    uint32 microseconds, uint32 sample_rate, size_t bytes_per_frame) {
-  return microseconds * sample_rate * bytes_per_frame /
-      base::Time::kMicrosecondsPerSecond;
-}
-
-// 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);
-}
-
 // 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;
-
-  // Fulfill write request.
+                                                  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()),
+                                               AudioManagerBase* manager)
+    : 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_);
 }
 
 PulseAudioOutputStream::~PulseAudioOutputStream() {
@@ -168,41 +123,60 @@ PulseAudioOutputStream::~PulseAudioOutputStream() {
   DCHECK(!pa_mainloop_);
 }
 
+// Helper macro for Open() to avoid code spam and string bloat.
+#define PULSE_CHECK(expression, message) \

[scherkus (not reviewing), 2012/10/11 20:19:31]

this isn't really a CHECK... a more appropriate name would be
RETURN_ON_FAILURE() similar to our code under content/common/gpu/media/

[DaleCurtis, 2012/10/12 00:09:12]

PULSE_CHECK sounds cooler though! :) Sadly, done.

+  if (expression) { \

[scherkus (not reviewing), 2012/10/11 20:19:31]

you should check for the negated expression

reading the CHECKs below seem strange as we're checking for the error value as
opposed to the success value (i.e., CHECK that it's null?!)

[scherkus (not reviewing), 2012/10/11 20:19:31]

this should be wrapped in a do { ... } while (0)

see
http://stackoverflow.com/questions/1067226/c-multi-line-macro-do-while0-vs-sc...

[DaleCurtis, 2012/10/12 00:09:12]

That's not necessary here since the code's already in an if() {} block.

[DaleCurtis, 2012/10/12 00:09:12]

Done.

+    if (pa_context_) { \
+      pa_threaded_mainloop_unlock(pa_mainloop_); \

[scherkus (not reviewing), 2012/10/11 20:19:31]

how about declaring AutoPulseAudioLock helper at the top of this file?

[DaleCurtis, 2012/10/12 00:09:12]

Done.

+      DLOG(ERROR) << message << pa_context_errno(pa_context_); \

[scherkus (not reviewing), 2012/10/11 20:19:31]

this forces log messages to know that the context error number gets tacked on
the end

how about a separate DLOG() for pa_context_errno() when pa_context_ != NULL or
applying some spacing/formatting?

[DaleCurtis, 2012/10/12 00:09:12]

Done.

+    } else { \
+      DLOG(ERROR) << message; \
+    } \
+    return false; \
+  }
+
 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();
+  PULSE_CHECK(!pa_mainloop_, "Failed to create PulseAudio 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);
+  PULSE_CHECK(!pa_context_, "Failed to create PulseAudio context.");
+
+  // Lock the mainloop while we setup our context.  Failing to do so will lead
+  // to crashes as the PulseAudio thread tries to run before things are ready.
+  pa_threaded_mainloop_lock(pa_mainloop_);
+
+  PULSE_CHECK(pa_threaded_mainloop_start(pa_mainloop_) < 0,
+              "Failed to start PulseAudio mainloop: ");
+  PULSE_CHECK(pa_context_connect(
+                  pa_context_, NULL, PA_CONTEXT_NOAUTOSPAWN, NULL) < 0,
+              "Failed to connect PulseAudio context: ");
 
   // 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) {
-      Reset();
-      return false;
-    }
-  }
+  pa_context_state_t context_state;
+  do {
+    pa_threaded_mainloop_wait(pa_mainloop_);

[scherkus (not reviewing), 2012/10/11 20:19:31]

should we wait before checking the state?

i.e., will we hang if state reaches READY before we get to this line as no more
events happen?

[DaleCurtis, 2012/10/12 00:09:12]

No we shouldn't, this will hang. Fixed.

+    context_state = pa_context_get_state(pa_context_);
+    PULSE_CHECK(!PA_CONTEXT_IS_GOOD(context_state),
+                "Invalid PulseAudio context state: ");
+  } while (context_state != PA_CONTEXT_READY);
 
   // 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?

[scherkus (not reviewing), 2012/10/11 20:19:31]

can you elaborate as to what might be incorrect?

[DaleCurtis, 2012/10/12 00:09:12]

No, I just haven't reviewed it thoroughly. Removed the TODO() for now.

   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,65 +184,75 @@ bool PulseAudioOutputStream::Open() {
   }
   playback_handle_ = pa_stream_new(pa_context_, "Playback",
                                    &pa_sample_specifications, map);
+  PULSE_CHECK(!playback_handle_, "Failed to create PulseAudio stream: ");
 
-  // 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.
   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.
-  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_) {
-    Reset();
-    return false;
-  }
+  PULSE_CHECK(pa_stream_connect_playback(
+                  playback_handle_, NULL, &pa_buffer_attributes,
+                  static_cast<pa_stream_flags_t>(

[scherkus (not reviewing), 2012/10/11 20:19:31]

nit: construct flags variable outside of this function call w/ doc/summary/link
as to what they do

[DaleCurtis, 2012/10/12 00:09:12]

Removed them all except PA_STREAM_START_CORKED as the others were the cause of a
poor audio clock. May not need that smoothing function after all!

PA_STREAM_START_CORKED is self explanatory and explained elsewhere so I've left
it inline.

+                      PA_STREAM_INTERPOLATE_TIMING | PA_STREAM_ADJUST_LATENCY |
+                      PA_STREAM_AUTO_TIMING_UPDATE | PA_STREAM_START_CORKED),
+                  NULL, NULL) < 0,
+              "Failed to connect PulseAudio stream: ");
+
+  // Wait for the stream to be ready.
+  pa_stream_state_t stream_state;
+  do {
+    pa_threaded_mainloop_wait(pa_mainloop_);

[scherkus (not reviewing), 2012/10/11 20:19:31]

ditto

[DaleCurtis, 2012/10/12 00:09:12]

Done.

+    stream_state = pa_stream_get_state(playback_handle_);
+    PULSE_CHECK(!PA_STREAM_IS_GOOD(stream_state),
+                "Invalid PulseAudio stream state: ");
+  } while (stream_state != PA_STREAM_READY);
+
+  // Unlock the mainloop now that everything is setup.
+  pa_threaded_mainloop_unlock(pa_mainloop_);
 
   return true;
 }
 
+#undef PULSE_CHECK
+
 void PulseAudioOutputStream::Reset() {
-  stream_stopped_ = true;
+  if (!pa_mainloop_) {
+    DCHECK(!playback_handle_);
+    DCHECK(!pa_context_);
+    return;
+  }
+
+  pa_threaded_mainloop_lock(pa_mainloop_);
 
   // Close the stream.
   if (playback_handle_) {
-    pa_stream_flush(playback_handle_, NULL, NULL);
-    pa_stream_disconnect(playback_handle_);
+    // Ensure all samples are played out before shutdown.
+    WaitForPulseOperation(pa_stream_flush(playback_handle_, NULL, NULL));
 
     // 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_mainloop_free(pa_mainloop_);
-    pa_mainloop_ = NULL;
-  }
 
-  // Release internal buffer.
-  client_buffer_.reset();
+  pa_threaded_mainloop_unlock(pa_mainloop_);
+  pa_threaded_mainloop_stop(pa_mainloop_);
+  pa_threaded_mainloop_free(pa_mainloop_);
+  pa_mainloop_ = NULL;
 }
 
 void PulseAudioOutputStream::Close() {
@@ -281,138 +265,76 @@ void PulseAudioOutputStream::Close() {
   manager_->ReleaseOutputStream(this);
 }
 
-void PulseAudioOutputStream::WaitForWriteRequest() {
-  DCHECK(manager_->GetMessageLoop()->BelongsToCurrentThread());
+int PulseAudioOutputStream::GetHardwareLatencyInBytes() {
+  int negative = 0;
+  pa_usec_t pa_latency_micros = 0;
+  if (pa_stream_get_latency(playback_handle_, &pa_latency_micros,
+                            &negative) != 0 || negative)

[scherkus (not reviewing), 2012/10/11 20:19:31]

can you split this into two ifs?

it's a bit subtle with the negative out-param

[DaleCurtis, 2012/10/12 00:09:12]

Done.

+    return 0;
 
-  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()));
-  }
+  return (pa_latency_micros * params_.sample_rate() *
+          params_.GetBytesPerFrame()) / base::Time::kMicrosecondsPerSecond;
 }
 
-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_);
-  int frames_filled = RunDataCallback(
-      audio_bus_.get(), AudioBuffersState(buffer_delay, hardware_delay));
-  size_t packet_size = frames_filled * bytes_per_frame_;
-
-  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());
+int PulseAudioOutputStream::FillBuffer(void* buffer, size_t buffer_size) {
+  CHECK(source_callback_);
+  int frames_filled = source_callback_->OnMoreData(
+      audio_bus_.get(), AudioBuffersState(0, GetHardwareLatencyInBytes()));
 
+  int packet_size = frames_filled * params_.GetBytesPerFrame();
   if (packet_size == 0)
-    return false;
-
-  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;
-}
-
-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;
+    return 0;
 
-  // 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);
-  }
+  // 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.
+  CHECK_LE(static_cast<size_t>(packet_size), buffer_size);
+  audio_bus_->ToInterleaved(frames_filled, params_.bits_per_sample(), buffer);
 
-  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));
-  } else {
-    // Continue playback.
-    manager_->GetMessageLoop()->PostTask(FROM_HERE, base::Bind(
-        &PulseAudioOutputStream::WaitForWriteRequest,
-        weak_factory_.GetWeakPtr()));
-  }
+  media::AdjustVolume(buffer, packet_size, params_.channels(),
+                      params_.bits_per_sample(), volume_);
+  return packet_size;
 }
 
-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;
-  }
+void PulseAudioOutputStream::FulfillWriteRequest(size_t requested_bytes) {
+  // Let pa_stream_begin_write auto detect the buffer size, it should choose the

[scherkus (not reviewing), 2012/10/11 20:19:31]

add ()

[DaleCurtis, 2012/10/12 00:09:12]

Done.

+  // same size as the callback request.  We CHECK() to make sure it does.
+  size_t bytes_available = static_cast<size_t>(-1);
+  void* audio_buffer = NULL;
+
+  // Request a buffer from PulseAudio and ensure it's the correct size.
+  CHECK_GE(pa_stream_begin_write(
+      playback_handle_, &audio_buffer, &bytes_available), 0);
+  CHECK_EQ(bytes_available, requested_bytes);
+  CHECK_EQ(requested_bytes, static_cast<size_t>(params_.GetBytesPerBuffer()));
+
+  int bytes_filled = FillBuffer(audio_buffer, bytes_available);
+  pa_stream_write(playback_handle_, audio_buffer, bytes_filled, NULL, 0LL,

[scherkus (not reviewing), 2012/10/11 20:19:31]

check return value?

[DaleCurtis, 2012/10/12 00:09:12]

Done. Add a lot of error handling elsewhere too.

+                  PA_SEEK_RELATIVE);
 }
 
 void PulseAudioOutputStream::Start(AudioSourceCallback* callback) {
   DCHECK(manager_->GetMessageLoop()->BelongsToCurrentThread());
   CHECK(callback);
-  DLOG_IF(ERROR, !playback_handle_)
-      << "Open() has not been called successfully";
-  if (!playback_handle_)
-    return;
+  CHECK(playback_handle_);
 
   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()));
+  // Uncork (resume) the stream.
+  pa_threaded_mainloop_lock(pa_mainloop_);
+  WaitForPulseOperation(pa_stream_cork(playback_handle_, 0, NULL, NULL));
+  pa_threaded_mainloop_unlock(pa_mainloop_);
 }
 
 void PulseAudioOutputStream::Stop() {
   DCHECK(manager_->GetMessageLoop()->BelongsToCurrentThread());
 
-  stream_stopped_ = true;
+  // Cork (pause) the stream.  Waiting for the mainloop lock (should) ensure
+  // outstanding callbacks have completed.
+  pa_threaded_mainloop_lock(pa_mainloop_);
+  WaitForPulseOperation(pa_stream_cork(playback_handle_, 1, NULL, NULL));
+  pa_threaded_mainloop_unlock(pa_mainloop_);
+
+  source_callback_ = NULL;
 }
 
 void PulseAudioOutputStream::SetVolume(double volume) {
@@ -427,12 +349,11 @@ void PulseAudioOutputStream::GetVolume(double* volume) {
   *volume = volume_;
 }
 
-int PulseAudioOutputStream::RunDataCallback(
-    AudioBus* audio_bus, AudioBuffersState buffers_state) {
-  if (source_callback_)
-    return source_callback_->OnMoreData(audio_bus, buffers_state);
-
-  return 0;
+void PulseAudioOutputStream::WaitForPulseOperation(pa_operation* op) {

[scherkus (not reviewing), 2012/10/11 20:19:31]

this function assumes that the caller has locked the mainloop

do you need to acquire the lock for running the operations (i.e., cork/uncork)
or for checking the state + waiting?

if it's the latter, PA docs state that their locks are recursive so you _could_
add lock/unlock inside this function + remove the restriction on the callee

[DaleCurtis, 2012/10/12 00:09:12]

I'm pretty sure those calls need the lock too since we want to make sure there
aren't any outstanding callbacks when we set |source_callback_| to NULL, etc.

+  CHECK(op);
+  while (pa_operation_get_state(op) == PA_OPERATION_RUNNING)
+    pa_threaded_mainloop_wait(pa_mainloop_);
+  pa_operation_unref(op);
 }
 
 }  // namespace media