Get PulseAudio implementation working.

media/audio/pulse/pulse_output.cc

 // Copyright (c) 2012 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/pulse/pulse_output.h"
 
-#include "base/bind.h"
-#include "base/message_loop.h"
+#include <pulse/pulseaudio.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 {
 
+// A helper class that acquires pa_threaded_mainloop_lock() while in scope.
+class AutoPulseLock {

[no longer working on chromium, 2012/10/14 19:03:41]

How about change the name to ScopedPulseLock?

[DaleCurtis, 2012/11/06 01:35:07]

I like AutoPulseLock for consistently with the existing AutoLock and AutoUnlock
classes.

+ public:
+  explicit AutoPulseLock(pa_threaded_mainloop* pa_mainloop)
+      : pa_mainloop_(pa_mainloop) {
+    pa_threaded_mainloop_lock(pa_mainloop_);
+  }
+
+  ~AutoPulseLock() {
+    pa_threaded_mainloop_unlock(pa_mainloop_);
+  }
+
+ private:
+  pa_threaded_mainloop* pa_mainloop_;
+  DISALLOW_COPY_AND_ASSIGN(AutoPulseLock);
+};
+
 static pa_sample_format_t BitsToPASampleFormat(int bits_per_sample) {

[no longer working on chromium, 2012/10/14 19:03:41]

what about extracting these helper class/functions to pulse_util.cc? Then we can
keep the pulse_ouput.cc clean while sharing the code for the input that I might
add.

It is fine not to do if you don't have time.

Thanks.

[DaleCurtis, 2012/11/06 01:35:07]

Lets keep them here for now, when we have an input side we can extract them to a
utility class.

   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;
   }
 }
 
 static pa_channel_position ChromiumToPAChannelPosition(Channels channel) {
   switch (channel) {
     // PulseAudio does not differentiate between left/right and
     // stereo-left/stereo-right, both translate to front-left/front-right.
     case LEFT:
     case STEREO_LEFT:
@@ skipping 14 matching lines @@
     case RIGHT_OF_CENTER:
       return PA_CHANNEL_POSITION_FRONT_RIGHT_OF_CENTER;
     case BACK_CENTER:
       return PA_CHANNEL_POSITION_REAR_CENTER;
     case SIDE_LEFT:
       return PA_CHANNEL_POSITION_SIDE_LEFT;
     case SIDE_RIGHT:
       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(
     ChannelLayout channel_layout) {
   // Initialize channel map.
   pa_channel_map channel_map;
   pa_channel_map_init(&channel_map);
 
   channel_map.channels = ChannelLayoutToChannelCount(channel_layout);
 

[DaleCurtis, 2012/11/09 01:10:52]

Turns out this code was wrong after all...

   // All channel maps have the same size array of channel positions.
   for (unsigned int channel = 0; channel != CHANNELS_MAX; ++channel) {
     int channel_position = kChannelOrderings[channel_layout][channel];
     if (channel_position > -1) {
       channel_map.map[channel_position] = ChromiumToPAChannelPosition(
           static_cast<Channels>(channel));
     } else {
       // PulseAudio expects unused channels in channel maps to be filled with
       // PA_CHANNEL_POSITION_MONO.
       channel_map.map[channel_position] = PA_CHANNEL_POSITION_MONO;
     }
   }
 
   // Fill in the rest of the unused channels.
   for (unsigned int channel = CHANNELS_MAX; channel != PA_CHANNELS_MAX;
        ++channel) {
     channel_map.map[channel] = PA_CHANNEL_POSITION_MONO;
   }
 
   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.
-  stream->FulfillWriteRequest(length);
+// static, pa_context_notify_cb
+void PulseAudioOutputStream::ContextNotifyCallback(pa_context* c,
+                                                   void* p_this) {
+  PulseAudioOutputStream* stream = static_cast<PulseAudioOutputStream*>(p_this);
+
+  // Forward unexpected failures to the AudioSourceCallback if available.  All
+  // these variables are only modified under pa_threaded_mainloop_lock() so this
+  // should be thread safe.
+  if (c && stream->source_callback_ &&
+      pa_context_get_state(c) == PA_CONTEXT_FAILED) {
+    stream->source_callback_->OnError(stream, pa_context_errno(c));
+  }
+
+  pa_threaded_mainloop_signal(stream->pa_mainloop_, 0);
+}
+
+// static, pa_stream_notify_cb
+void PulseAudioOutputStream::StreamNotifyCallback(pa_stream* s, void* p_this) {
+  PulseAudioOutputStream* stream = static_cast<PulseAudioOutputStream*>(p_this);
+
+  // Forward unexpected failures to the AudioSourceCallback if available.  All
+  // these variables are only modified under pa_threaded_mainloop_lock() so this
+  // should be thread safe.
+  if (s && stream->source_callback_ &&
+      pa_stream_get_state(s) == PA_STREAM_FAILED) {
+    stream->source_callback_->OnError(
+        stream, pa_context_errno(stream->pa_context_));
+  }
+
+  pa_threaded_mainloop_signal(stream->pa_mainloop_, 0);
+}
+
+// static, pa_stream_success_cb_t
+void PulseAudioOutputStream::StreamSuccessCallback(pa_stream* s, int success,
+                                                   void* p_this) {
+  PulseAudioOutputStream* stream = static_cast<PulseAudioOutputStream*>(p_this);
+  pa_threaded_mainloop_signal(stream->pa_mainloop_, 0);
+}
+
+// static, pa_stream_request_cb_t
+void PulseAudioOutputStream::StreamRequestCallback(pa_stream* s, size_t len,
+                                                   void* p_this) {
+  // Fulfill write request; must always result in a pa_stream_write() call.
+  static_cast<PulseAudioOutputStream*>(p_this)->FulfillWriteRequest(len);
 }
 
 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),
+      pa_stream_(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() {
-  // All internal structures should already have been freed in Close(),
-  // which calls AudioManagerPulse::Release which deletes this object.
-  DCHECK(!playback_handle_);
+  // All internal structures should already have been freed in Close(), which
+  // calls AudioManagerBase::ReleaseOutputStream() which deletes this object.
+  DCHECK(!pa_stream_);
   DCHECK(!pa_context_);
   DCHECK(!pa_mainloop_);
 }
 
+// Helper macro for Open() to avoid code spam and string bloat.
+#define RETURN_ON_FAILURE(expression, message) \
+  if (!(expression)) { \

[scherkus (not reviewing), 2012/10/17 22:44:53]

"if() {}" is a compound statement and doesn't require a semicolon at the end

using "do {} while(0)" expands into a regular statement that enforces ending the
statement with a semicolon

[DaleCurtis, 2012/11/06 01:35:07]

Sure, but that's pretty redundant here. Done in any case.

+    if (pa_context_) { \
+      DLOG(ERROR) << message << "  Error: " << pa_context_errno(pa_context_); \
+    } 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();
+  RETURN_ON_FAILURE(pa_mainloop_, "Failed to create PulseAudio main loop.");
+
+  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);
-
-  // 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;
-    }
+  RETURN_ON_FAILURE(pa_context_, "Failed to create PulseAudio context.");
+
+  // A state callback must be set before calling pa_threaded_mainloop_lock() or
+  // pa_threaded_mainloop_wait() calls may lead to dead lock.
+  pa_context_set_state_callback(pa_context_, &ContextNotifyCallback, this);
+
+  // Lock the main loop while setting up the context.  Failure to do so may lead
+  // to crashes as the PulseAudio thread tries to run before things are ready.
+  AutoPulseLock auto_lock(pa_mainloop_);
+
+  RETURN_ON_FAILURE(
+      pa_threaded_mainloop_start(pa_mainloop_) == 0,
+      "Failed to start PulseAudio main loop.");
+  RETURN_ON_FAILURE(
+      pa_context_connect(pa_context_, NULL, PA_CONTEXT_NOAUTOSPAWN, NULL) == 0,
+      "Failed to connect PulseAudio context.");
+
+  // Wait until |pa_context_| is ready.  pa_threaded_mainloop_wait() must be
+  // called after pa_context_get_state() in case the context is already ready,
+  // otherwise pa_threaded_mainloop_wait() will hang indefinitely.
+  while (true) {
+    pa_context_state_t context_state = pa_context_get_state(pa_context_);
+    RETURN_ON_FAILURE(
+        PA_CONTEXT_IS_GOOD(context_state), "Invalid PulseAudio context state.");
+    if (context_state == PA_CONTEXT_READY)
+      break;
+    pa_threaded_mainloop_wait(pa_mainloop_);
   }
 
   // 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.
   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).
     map = &source_channel_map;
   }
-  playback_handle_ = pa_stream_new(pa_context_, "Playback",
+  pa_stream_ = pa_stream_new(pa_context_, "Playback",
                                    &pa_sample_specifications, map);

[no longer working on chromium, 2012/10/14 19:03:41]

indentation.

[DaleCurtis, 2012/11/06 01:35:07]

Done.

-
-  // 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.
+  RETURN_ON_FAILURE(pa_stream_, "Failed to create PulseAudio stream.");
+  pa_stream_set_state_callback(pa_stream_, &StreamNotifyCallback, this);
+
+  // Even though we start the stream corked below, PulseAudio will issue one
+  // stream request after setup.  FulfillWriteRequest() must fulfill the write.
+  // TODO(dalecurtis): Why is this happening? Upstream bug?

[no longer working on chromium, 2012/10/14 19:03:41]

I was hit by this issue too, do you know if it comes with
pa_stream_connect_playback or pa_stream_set_write_callback?

[DaleCurtis, 2012/11/06 01:35:07]

pa_stream_connect_playback().

+  pa_stream_set_write_callback(pa_stream_, &StreamRequestCallback, this);
+
+  // 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.maxlength = params_.GetBytesPerBuffer();
+  pa_buffer_attributes.minreq = params_.GetBytesPerBuffer();
+  pa_buffer_attributes.tlength = static_cast<uint32_t>(-1);
   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 |

[no longer working on chromium, 2012/10/14 19:03:41]

From internet:

The server tries to assure that at least tlength bytes are always available in
the per-stream server-side playback buffer. It is recommended to set this to
(uint32_t) -1, which will initialize this to a value that is deemed sensible by
the server. However, this value will default to something like 2s, i.e. for
applications that have specific latency requirements this value should be set to
the maximum latency that the application can deal with. When
PA_STREAM_ADJUST_LATENCY is not set this value will influence only the
per-stream playback buffer size. When PA_STREAM_ADJUST_LATENCY is set the
overall latency of the sink plus the playback buffer size is configured to this
value. Set PA_STREAM_ADJUST_LATENCY if you are interested in adjusting the
overall latency. Don't set it if you are interested in configuring the
server-side per-stream playback buffer size.

Though I don't think we will have 2s latency, but I am curious on what the real
latency will be without using PA_STREAM_ADJUST_LATENCY, if it is too high, we
might need to get it back, and use pa_stream_set_underflow_callback to setup a
underrun callback to adjust the buffer attributes.

-                             PA_STREAM_AUTO_TIMING_UPDATE),
-                             NULL, NULL);
-
-  if (!playback_handle_) {
-    Reset();
-    return false;
+  RETURN_ON_FAILURE(
+      pa_stream_connect_playback(
+          pa_stream_, NULL, &pa_buffer_attributes, PA_STREAM_START_CORKED,
+          NULL, NULL) == 0,
+      "Failed to connect PulseAudio stream.");
+
+  // Wait for the stream to be ready.
+  while (true) {
+    pa_stream_state_t stream_state = pa_stream_get_state(pa_stream_);
+    RETURN_ON_FAILURE(
+        PA_STREAM_IS_GOOD(stream_state), "Invalid PulseAudio stream state.");
+    if (stream_state == PA_STREAM_READY)
+      break;
+    pa_threaded_mainloop_wait(pa_mainloop_);
   }
 
   return true;
 }
 
+#undef RETURN_ON_FAILURE
+
 void PulseAudioOutputStream::Reset() {
-  stream_stopped_ = true;
-
-  // 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;
-  }
-
-  // Release internal buffer.
-  client_buffer_.reset();
+  if (!pa_mainloop_) {
+    DCHECK(!pa_stream_);
+    DCHECK(!pa_context_);
+    return;
+  }
+
+  {
+    AutoPulseLock auto_lock(pa_mainloop_);
+
+    // Close the stream.
+    if (pa_stream_) {

[no longer working on chromium, 2012/10/14 19:03:41]

Do we need to call pa_stream_set_state_callback to NULL here?

[DaleCurtis, 2012/11/06 01:35:07]

Done.

+      // Ensure all samples are played out before shutdown.
+      WaitForPulseOperation(pa_stream_flush(
+          pa_stream_, &StreamSuccessCallback, this));
+
+      // Release PulseAudio structures.
+      pa_stream_disconnect(pa_stream_);
+      pa_stream_unref(pa_stream_);
+      pa_stream_ = NULL;
+    }
+
+    if (pa_context_) {
+      // Clear our state callback.
+      pa_context_set_state_callback(pa_context_, NULL, NULL);
+
+      pa_context_disconnect(pa_context_);
+      pa_context_unref(pa_context_);
+      pa_context_ = NULL;
+    }
+  }
+
+  pa_threaded_mainloop_stop(pa_mainloop_);
+  pa_threaded_mainloop_free(pa_mainloop_);
+  pa_mainloop_ = NULL;
 }
 
 void PulseAudioOutputStream::Close() {
   DCHECK(manager_->GetMessageLoop()->BelongsToCurrentThread());
 
   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(manager_->GetMessageLoop()->BelongsToCurrentThread());
+int PulseAudioOutputStream::GetHardwareLatencyInBytes() {
+  int negative = 0;
+  pa_usec_t pa_latency_micros = 0;
+  if (pa_stream_get_latency(pa_stream_, &pa_latency_micros, &negative) != 0 ||

[scherkus (not reviewing), 2012/10/17 22:44:53]

I meant splitting as in

if (pa_stream_get_latency(...) != 0)
  return 0;
if (negative)
  return 0;

[DaleCurtis, 2012/11/06 01:35:07]

Done.

+      negative) {
+    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_;
+void PulseAudioOutputStream::FillBuffer(void* buffer, size_t buffer_size) {
+  size_t frames_filled = 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());
+  // See comments on pa_stream_set_write_callback() in Open().
+  if (source_callback_) {
+    frames_filled = source_callback_->OnMoreData(
+       audio_bus_.get(), AudioBuffersState(0, GetHardwareLatencyInBytes()));
+  }
 
-  if (packet_size == 0)
-    return false;
+  size_t packet_size = frames_filled * params_.GetBytesPerFrame();
+  CHECK_LE(packet_size, buffer_size);
+  if (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, params_.bits_per_sample() / 8, buffer);
 
-  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;
+    media::AdjustVolume(buffer, packet_size, params_.channels(),
+                        params_.bits_per_sample() / 8, volume_);
+  }
+
+  // Zero out any unfilled data.
+  memset(buffer, 0, buffer_size - 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;
+  void* audio_buffer = NULL;
+  // Tell pa_stream_begin_write() the buffer size we want.
+  size_t bytes_available = requested_bytes;
 
-  // 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();
-  }
+  // Request a buffer from PulseAudio and ensure it's the correct size.
+  CHECK_GE(pa_stream_begin_write(
+      pa_stream_, &audio_buffer, &bytes_available), 0);
+  CHECK_EQ(bytes_available, requested_bytes);

[no longer working on chromium, 2012/10/14 19:03:41]

I am not sure but afraid that we might be hit by this CHECK, especially for apps
like html5 using high buffer size, can this pa_stream_begin_write returns a
value smaller than requested_bytes?

[DaleCurtis, 2012/11/09 01:10:52]

Done. Now breaks the buffer into whatever chunks pa_stream_begin_write returns.

+  CHECK_EQ(requested_bytes, static_cast<size_t>(params_.GetBytesPerBuffer()));

[no longer working on chromium, 2012/10/14 19:03:41]

if pa can't keep up with the pace of the callbacks, will it adjust it internal
buffer size and affect this requested_bytes?

[DaleCurtis, 2012/11/09 01:10:52]

Not so long as we force the buffer size in the pa_sample_spec used during stream
creation.

Chrome isn't compatible with the Pulse audio model of an adjusted buffer size
depending on system load.  I don't think it's worth rearchitecting to support
this on Linux either.

Opening at the native sample rate seems to work well even with a 440 frame
buffer size, otherwise even an 8k buffer size isn't enough to prevent pops and
clicks :-/

 
-  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);
-  }
-
-  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()));
-  }
-}
-
-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;
+  FillBuffer(audio_buffer, bytes_available);
+  if (pa_stream_write(pa_stream_, audio_buffer, requested_bytes, NULL, 0LL,
+                      PA_SEEK_RELATIVE) < 0) {
+    if (source_callback_)
+      source_callback_->OnError(this, pa_context_errno(pa_context_));
   }
 }
 
 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_)
+  CHECK(pa_stream_);
+
+  AutoPulseLock auto_lock(pa_mainloop_);
+
+  // Ensure the context and stream are ready.
+  if (pa_context_get_state(pa_context_) != PA_CONTEXT_READY &&
+      pa_stream_get_state(pa_stream_) != PA_STREAM_READY) {
+    callback->OnError(this, pa_context_errno(pa_context_));
     return;
+  }
 
   source_callback_ = callback;
+  pa_stream_set_write_callback(pa_stream_, &StreamRequestCallback, this);
 
-  // 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.
+  WaitForPulseOperation(pa_stream_cork(
+      pa_stream_, 0, &StreamSuccessCallback, this));
 }
 
 void PulseAudioOutputStream::Stop() {
   DCHECK(manager_->GetMessageLoop()->BelongsToCurrentThread());
 
-  stream_stopped_ = true;
+  // Cork (pause) the stream.  Waiting for the main loop lock will ensure
+  // outstanding callbacks have completed.
+  AutoPulseLock auto_lock(pa_mainloop_);
+
+  // Flush the stream prior to cork, doing so after will cause hangs.  Write
+  // callbacks are suspended while inside pa_threaded_mainloop_lock() so this
+  // is all thread safe.
+  WaitForPulseOperation(pa_stream_flush(
+    pa_stream_, &StreamSuccessCallback, this));

[scherkus (not reviewing), 2012/10/17 22:44:53]

indent

[DaleCurtis, 2012/11/09 01:10:52]

Done.

+
+  WaitForPulseOperation(pa_stream_cork(
+      pa_stream_, 1, &StreamSuccessCallback, this));
+
+  pa_stream_set_write_callback(pa_stream_, NULL, NULL);
+  source_callback_ = NULL;
 }
 
 void PulseAudioOutputStream::SetVolume(double volume) {
   DCHECK(manager_->GetMessageLoop()->BelongsToCurrentThread());
 
   volume_ = static_cast<float>(volume);
 }
 
 void PulseAudioOutputStream::GetVolume(double* volume) {
   DCHECK(manager_->GetMessageLoop()->BelongsToCurrentThread());
 
   *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) {
+  CHECK(op);
+  while (pa_operation_get_state(op) == PA_OPERATION_RUNNING)
+    pa_threaded_mainloop_wait(pa_mainloop_);
+  pa_operation_unref(op);
 }
 
 }  // namespace media