webrtc/sound/pulseaudiosoundsystem.cc

/*
 *  Copyright 2010 The WebRTC Project Authors. All rights reserved.
 *
 *  Use of this source code is governed by a BSD-style license
 *  that can be found in the LICENSE file in the root of the source
 *  tree. An additional intellectual property rights grant can be found
 *  in the file PATENTS.  All contributing project authors may
 *  be found in the AUTHORS file in the root of the source tree.
 */

#include "webrtc/sound/pulseaudiosoundsystem.h"

#ifdef HAVE_LIBPULSE

#include <algorithm>
#include "webrtc/sound/sounddevicelocator.h"
#include "webrtc/sound/soundinputstreaminterface.h"
#include "webrtc/sound/soundoutputstreaminterface.h"
#include "webrtc/base/common.h"
#include "webrtc/base/fileutils.h"  // for GetApplicationName()
#include "webrtc/base/logging.h"
#include "webrtc/base/timeutils.h"
#include "webrtc/base/worker.h"

namespace rtc {

// First PulseAudio protocol version that supports PA_STREAM_ADJUST_LATENCY.
static const uint32_t kAdjustLatencyProtocolVersion = 13;

// Lookup table from the rtc format enum in soundsysteminterface.h to
// Pulse's enums.
static const pa_sample_format_t kCricketFormatToPulseFormatTable[] = {
  // The order here must match the order in soundsysteminterface.h
  PA_SAMPLE_S16LE,
};

// Some timing constants for optimal operation. See
// https://tango.0pointer.de/pipermail/pulseaudio-discuss/2008-January/001170.html
// for a good explanation of some of the factors that go into this.

// Playback.

// For playback, there is a round-trip delay to fill the server-side playback
// buffer, so setting too low of a latency is a buffer underflow risk. We will
// automatically increase the latency if a buffer underflow does occur, but we
// also enforce a sane minimum at start-up time. Anything lower would be
// virtually guaranteed to underflow at least once, so there's no point in
// allowing lower latencies.
static const int kPlaybackLatencyMinimumMsecs = 20;
// Every time a playback stream underflows, we will reconfigure it with target
// latency that is greater by this amount.
static const int kPlaybackLatencyIncrementMsecs = 20;
// We also need to configure a suitable request size. Too small and we'd burn
// CPU from the overhead of transfering small amounts of data at once. Too large
// and the amount of data remaining in the buffer right before refilling it
// would be a buffer underflow risk. We set it to half of the buffer size.
static const int kPlaybackRequestFactor = 2;

// Capture.

// For capture, low latency is not a buffer overflow risk, but it makes us burn
// CPU from the overhead of transfering small amounts of data at once, so we set
// a recommended value that we use for the kLowLatency constant (but if the user
// explicitly requests something lower then we will honour it).
// 1ms takes about 6-7% CPU. 5ms takes about 5%. 10ms takes about 4.x%.
static const int kLowCaptureLatencyMsecs = 10;
// There is a round-trip delay to ack the data to the server, so the
// server-side buffer needs extra space to prevent buffer overflow. 20ms is
// sufficient, but there is no penalty to making it bigger, so we make it huge.
// (750ms is libpulse's default value for the _total_ buffer size in the
// kNoLatencyRequirements case.)
static const int kCaptureBufferExtraMsecs = 750;

static void FillPlaybackBufferAttr(int latency,
                                   pa_buffer_attr *attr) {
  attr->maxlength = latency;
  attr->tlength = latency;
  attr->minreq = latency / kPlaybackRequestFactor;
  attr->prebuf = attr->tlength - attr->minreq;
  LOG(LS_VERBOSE) << "Configuring latency = " << attr->tlength << ", minreq = "
                  << attr->minreq << ", minfill = " << attr->prebuf;
}

static pa_volume_t CricketVolumeToPulseVolume(int volume) {
  // PA's volume space goes from 0% at PA_VOLUME_MUTED (value 0) to 100% at
  // PA_VOLUME_NORM (value 0x10000). It can also go beyond 100% up to
  // PA_VOLUME_MAX (value UINT32_MAX-1), but using that is probably unwise.
  // We just linearly map the 0-255 scale of SoundSystemInterface onto
  // PA_VOLUME_MUTED-PA_VOLUME_NORM. If the programmer exceeds kMaxVolume then
  // they can access the over-100% features of PA.
  return PA_VOLUME_MUTED + (PA_VOLUME_NORM - PA_VOLUME_MUTED) *
      volume / SoundSystemInterface::kMaxVolume;
}

static int PulseVolumeToCricketVolume(pa_volume_t pa_volume) {
  return SoundSystemInterface::kMinVolume +
      (SoundSystemInterface::kMaxVolume - SoundSystemInterface::kMinVolume) *
      pa_volume / PA_VOLUME_NORM;
}

static pa_volume_t MaxChannelVolume(pa_cvolume *channel_volumes) {
  pa_volume_t pa_volume = PA_VOLUME_MUTED;  // Minimum possible value.
  for (int i = 0; i < channel_volumes->channels; ++i) {
    if (pa_volume < channel_volumes->values[i]) {
      pa_volume = channel_volumes->values[i];
    }
  }
  return pa_volume;
}

class PulseAudioDeviceLocator : public SoundDeviceLocator {
 public:
  PulseAudioDeviceLocator(const std::string &name,
                          const std::string &device_name)
      : SoundDeviceLocator(name, device_name) {
  }

  virtual SoundDeviceLocator *Copy() const {
    return new PulseAudioDeviceLocator(*this);
  }
};

// Functionality that is common to both PulseAudioInputStream and
// PulseAudioOutputStream.
class PulseAudioStream {
 public:
  PulseAudioStream(PulseAudioSoundSystem *pulse, pa_stream *stream, int flags)
      : pulse_(pulse), stream_(stream), flags_(flags) {
  }

  ~PulseAudioStream() {
    // Close() should have been called during the containing class's destructor.
    ASSERT(stream_ == NULL);
  }

  // Must be called with the lock held.
  bool Close() {
    if (!IsClosed()) {
      // Unset this here so that we don't get a TERMINATED callback.
      symbol_table()->pa_stream_set_state_callback()(stream_, NULL, NULL);
      if (symbol_table()->pa_stream_disconnect()(stream_) != 0) {
        LOG(LS_ERROR) << "Can't disconnect stream";
        // Continue and return true anyways.
      }
      symbol_table()->pa_stream_unref()(stream_);
      stream_ = NULL;
    }
    return true;
  }

  // Must be called with the lock held.
  int LatencyUsecs() {
    if (!(flags_ & SoundSystemInterface::FLAG_REPORT_LATENCY)) {
      return 0;
    }

    pa_usec_t latency;
    int negative;
    Lock();
    int re = symbol_table()->pa_stream_get_latency()(stream_, &latency,
        &negative);
    Unlock();
    if (re != 0) {
      LOG(LS_ERROR) << "Can't query latency";
      // We'd rather continue playout/capture with an incorrect delay than stop
      // it altogether, so return a valid value.
      return 0;
    }
    if (negative) {
      // The delay can be negative for monitoring streams if the captured
      // samples haven't been played yet. In such a case, "latency" contains the
      // magnitude, so we must negate it to get the real value.
      return -latency;
    } else {
      return latency;
    }
  }

  PulseAudioSoundSystem *pulse() {
    return pulse_;
  }

  PulseAudioSymbolTable *symbol_table() {
    return &pulse()->symbol_table_;
  }

  pa_stream *stream() {
    ASSERT(stream_ != NULL);
    return stream_;
  }

  bool IsClosed() {
    return stream_ == NULL;
  }

  void Lock() {
    pulse()->Lock();
  }

  void Unlock() {
    pulse()->Unlock();
  }

 private:
  PulseAudioSoundSystem *pulse_;
  pa_stream *stream_;
  int flags_;

  DISALLOW_COPY_AND_ASSIGN(PulseAudioStream);
};

// Implementation of an input stream. See soundinputstreaminterface.h regarding
// thread-safety.
class PulseAudioInputStream :
    public SoundInputStreamInterface,
    private rtc::Worker {

  struct GetVolumeCallbackData {
    PulseAudioInputStream *instance;
    pa_cvolume *channel_volumes;
  };

  struct GetSourceChannelCountCallbackData {
    PulseAudioInputStream *instance;
    uint8_t *channels;
  };

 public:
  PulseAudioInputStream(PulseAudioSoundSystem *pulse,
                        pa_stream *stream,
                        int flags)
      : stream_(pulse, stream, flags),
        temp_sample_data_(NULL),
        temp_sample_data_size_(0) {
    // This callback seems to never be issued, but let's set it anyways.
    symbol_table()->pa_stream_set_overflow_callback()(stream, &OverflowCallback,
        NULL);
  }

  virtual ~PulseAudioInputStream() {
    bool success = Close();
    // We need that to live.
    VERIFY(success);
  }

  virtual bool StartReading() {
    return StartWork();
  }

  virtual bool StopReading() {
    return StopWork();
  }

  virtual bool GetVolume(int *volume) {
    bool ret = false;

    Lock();

    // Unlike output streams, input streams have no concept of a stream volume,
    // only a device volume. So we have to retrieve the volume of the device
    // itself.

    pa_cvolume channel_volumes;

    GetVolumeCallbackData data;
    data.instance = this;
    data.channel_volumes = &channel_volumes;

    pa_operation *op = symbol_table()->pa_context_get_source_info_by_index()(
            stream_.pulse()->context_,
            symbol_table()->pa_stream_get_device_index()(stream_.stream()),
            &GetVolumeCallbackThunk,
            &data);
    if (!stream_.pulse()->FinishOperation(op)) {
      goto done;
    }

    if (data.channel_volumes) {
      // This pointer was never unset by the callback, so we must have received
      // an empty list of infos. This probably never happens, but we code for it
      // anyway.
      LOG(LS_ERROR) << "Did not receive GetVolumeCallback";
      goto done;
    }

    // We now have the volume for each channel. Each channel could have a
    // different volume if, e.g., the user went and changed the volumes in the
    // PA UI. To get a single volume for SoundSystemInterface we just take the
    // maximum. Ideally we'd do so with pa_cvolume_max, but it doesn't exist in
    // Hardy, so we do it manually.
    pa_volume_t pa_volume;
    pa_volume = MaxChannelVolume(&channel_volumes);
    // Now map onto the SoundSystemInterface range.
    *volume = PulseVolumeToCricketVolume(pa_volume);

    ret = true;
   done:
    Unlock();
    return ret;
  }

  virtual bool SetVolume(int volume) {
    bool ret = false;
    pa_volume_t pa_volume = CricketVolumeToPulseVolume(volume);

    Lock();

    // Unlike output streams, input streams have no concept of a stream volume,
    // only a device volume. So we have to change the volume of the device
    // itself.

    // The device may have a different number of channels than the stream and
    // their mapping may be different, so we don't want to use the channel count
    // from our sample spec. We could use PA_CHANNELS_MAX to cover our bases,
    // and the server allows that even if the device's channel count is lower,
    // but some buggy PA clients don't like that (the pavucontrol on Hardy dies
    // in an assert if the channel count is different). So instead we look up
    // the actual number of channels that the device has.

    uint8_t channels;

    GetSourceChannelCountCallbackData data;
    data.instance = this;
    data.channels = &channels;

    uint32_t device_index = symbol_table()->pa_stream_get_device_index()(
        stream_.stream());

    pa_operation *op = symbol_table()->pa_context_get_source_info_by_index()(
        stream_.pulse()->context_,
        device_index,
        &GetSourceChannelCountCallbackThunk,
        &data);
    if (!stream_.pulse()->FinishOperation(op)) {
      goto done;
    }

    if (data.channels) {
      // This pointer was never unset by the callback, so we must have received
      // an empty list of infos. This probably never happens, but we code for it
      // anyway.
      LOG(LS_ERROR) << "Did not receive GetSourceChannelCountCallback";
      goto done;
    }

    pa_cvolume channel_volumes;
    symbol_table()->pa_cvolume_set()(&channel_volumes, channels, pa_volume);

    op = symbol_table()->pa_context_set_source_volume_by_index()(
        stream_.pulse()->context_,
        device_index,
        &channel_volumes,
        // This callback merely logs errors.
        &SetVolumeCallback,
        NULL);
    if (!op) {
      LOG(LS_ERROR) << "pa_context_set_source_volume_by_index()";
      goto done;
    }
    // Don't need to wait for this to complete.
    symbol_table()->pa_operation_unref()(op);

    ret = true;
   done:
    Unlock();
    return ret;
  }

  virtual bool Close() {
    if (!StopReading()) {
      return false;
    }
    bool ret = true;
    if (!stream_.IsClosed()) {
      Lock();
      ret = stream_.Close();
      Unlock();
    }
    return ret;
  }

  virtual int LatencyUsecs() {
    return stream_.LatencyUsecs();
  }

 private:
  void Lock() {
    stream_.Lock();
  }

  void Unlock() {
    stream_.Unlock();
  }

  PulseAudioSymbolTable *symbol_table() {
    return stream_.symbol_table();
  }

  void EnableReadCallback() {
    symbol_table()->pa_stream_set_read_callback()(
         stream_.stream(),
         &ReadCallbackThunk,
         this);
  }

  void DisableReadCallback() {
    symbol_table()->pa_stream_set_read_callback()(
         stream_.stream(),
         NULL,
         NULL);
  }

  static void ReadCallbackThunk(pa_stream *unused1,
                                size_t unused2,
                                void *userdata) {
    PulseAudioInputStream *instance =
        static_cast<PulseAudioInputStream *>(userdata);
    instance->OnReadCallback();
  }

  void OnReadCallback() {
    // We get the data pointer and size now in order to save one Lock/Unlock
    // on OnMessage.
    if (symbol_table()->pa_stream_peek()(stream_.stream(),
                                         &temp_sample_data_,
                                         &temp_sample_data_size_) != 0) {
      LOG(LS_ERROR) << "Can't read data!";
      return;
    }
    // Since we consume the data asynchronously on a different thread, we have
    // to temporarily disable the read callback or else Pulse will call it
    // continuously until we consume the data. We re-enable it below.
    DisableReadCallback();
    HaveWork();
  }

  // Inherited from Worker.
  virtual void OnStart() {
    Lock();
    EnableReadCallback();
    Unlock();
  }

  // Inherited from Worker.
  virtual void OnHaveWork() {
    ASSERT(temp_sample_data_ && temp_sample_data_size_);
    SignalSamplesRead(temp_sample_data_,
                      temp_sample_data_size_,
                      this);
    temp_sample_data_ = NULL;
    temp_sample_data_size_ = 0;

    Lock();
    for (;;) {
      // Ack the last thing we read.
      if (symbol_table()->pa_stream_drop()(stream_.stream()) != 0) {
        LOG(LS_ERROR) << "Can't ack read data";
      }

      if (symbol_table()->pa_stream_readable_size()(stream_.stream()) <= 0) {
        // Then that was all the data.
        break;
      }

      // Else more data.
      const void *sample_data;
      size_t sample_data_size;
      if (symbol_table()->pa_stream_peek()(stream_.stream(),
                                           &sample_data,
                                           &sample_data_size) != 0) {
        LOG(LS_ERROR) << "Can't read data!";
        break;
      }

      // Drop lock for sigslot dispatch, which could take a while.
      Unlock();
      SignalSamplesRead(sample_data, sample_data_size, this);
      Lock();

      // Return to top of loop for the ack and the check for more data.
    }
    EnableReadCallback();
    Unlock();
  }

  // Inherited from Worker.
  virtual void OnStop() {
    Lock();
    DisableReadCallback();
    Unlock();
  }

  static void OverflowCallback(pa_stream *stream,
                               void *userdata) {
    LOG(LS_WARNING) << "Buffer overflow on capture stream " << stream;
  }

  static void GetVolumeCallbackThunk(pa_context *unused,
                                     const pa_source_info *info,
                                     int eol,
                                     void *userdata) {
    GetVolumeCallbackData *data =
        static_cast<GetVolumeCallbackData *>(userdata);
    data->instance->OnGetVolumeCallback(info, eol, &data->channel_volumes);
  }

  void OnGetVolumeCallback(const pa_source_info *info,
                           int eol,
                           pa_cvolume **channel_volumes) {
    if (eol) {
      // List is over. Wake GetVolume().
      stream_.pulse()->Signal();
      return;
    }

    if (*channel_volumes) {
      **channel_volumes = info->volume;
      // Unset the pointer so that we know that we have have already copied the
      // volume.
      *channel_volumes = NULL;
    } else {
      // We have received an additional callback after the first one, which
      // doesn't make sense for a single source. This probably never happens,
      // but we code for it anyway.
      LOG(LS_WARNING) << "Ignoring extra GetVolumeCallback";
    }
  }

  static void GetSourceChannelCountCallbackThunk(pa_context *unused,
                                                 const pa_source_info *info,
                                                 int eol,
                                                 void *userdata) {
    GetSourceChannelCountCallbackData *data =
        static_cast<GetSourceChannelCountCallbackData *>(userdata);
    data->instance->OnGetSourceChannelCountCallback(info, eol, &data->channels);
  }

  void OnGetSourceChannelCountCallback(const pa_source_info *info,
                                       int eol,
                                       uint8_t **channels) {
    if (eol) {
      // List is over. Wake SetVolume().
      stream_.pulse()->Signal();
      return;
    }

    if (*channels) {
      **channels = info->channel_map.channels;
      // Unset the pointer so that we know that we have have already copied the
      // channel count.
      *channels = NULL;
    } else {
      // We have received an additional callback after the first one, which
      // doesn't make sense for a single source. This probably never happens,
      // but we code for it anyway.
      LOG(LS_WARNING) << "Ignoring extra GetSourceChannelCountCallback";
    }
  }

  static void SetVolumeCallback(pa_context *unused1,
                                int success,
                                void *unused2) {
    if (!success) {
      LOG(LS_ERROR) << "Failed to change capture volume";
    }
  }

  PulseAudioStream stream_;
  // Temporary storage for passing data between threads.
  const void *temp_sample_data_;
  size_t temp_sample_data_size_;

  DISALLOW_COPY_AND_ASSIGN(PulseAudioInputStream);
};

// Implementation of an output stream. See soundoutputstreaminterface.h
// regarding thread-safety.
class PulseAudioOutputStream :
    public SoundOutputStreamInterface,
    private rtc::Worker {

  struct GetVolumeCallbackData {
    PulseAudioOutputStream *instance;
    pa_cvolume *channel_volumes;
  };

 public:
  PulseAudioOutputStream(PulseAudioSoundSystem *pulse,
                         pa_stream *stream,
                         int flags,
                         int latency)
      : stream_(pulse, stream, flags),
        configured_latency_(latency),
        temp_buffer_space_(0) {
    symbol_table()->pa_stream_set_underflow_callback()(stream,
                                                       &UnderflowCallbackThunk,
                                                       this);
  }

  virtual ~PulseAudioOutputStream() {
    bool success = Close();
    // We need that to live.
    VERIFY(success);
  }

  virtual bool EnableBufferMonitoring() {
    return StartWork();
  }

  virtual bool DisableBufferMonitoring() {
    return StopWork();
  }

  virtual bool WriteSamples(const void *sample_data,
                            size_t size) {
    bool ret = true;
    Lock();
    if (symbol_table()->pa_stream_write()(stream_.stream(),
                                          sample_data,
                                          size,
                                          NULL,
                                          0,
                                          PA_SEEK_RELATIVE) != 0) {
      LOG(LS_ERROR) << "Unable to write";
      ret = false;
    }
    Unlock();
    return ret;
  }

  virtual bool GetVolume(int *volume) {
    bool ret = false;

    Lock();

    pa_cvolume channel_volumes;

    GetVolumeCallbackData data;
    data.instance = this;
    data.channel_volumes = &channel_volumes;

    pa_operation *op = symbol_table()->pa_context_get_sink_input_info()(
            stream_.pulse()->context_,
            symbol_table()->pa_stream_get_index()(stream_.stream()),
            &GetVolumeCallbackThunk,
            &data);
    if (!stream_.pulse()->FinishOperation(op)) {
      goto done;
    }

    if (data.channel_volumes) {
      // This pointer was never unset by the callback, so we must have received
      // an empty list of infos. This probably never happens, but we code for it
      // anyway.
      LOG(LS_ERROR) << "Did not receive GetVolumeCallback";
      goto done;
    }

    // We now have the volume for each channel. Each channel could have a
    // different volume if, e.g., the user went and changed the volumes in the
    // PA UI. To get a single volume for SoundSystemInterface we just take the
    // maximum. Ideally we'd do so with pa_cvolume_max, but it doesn't exist in
    // Hardy, so we do it manually.
    pa_volume_t pa_volume;
    pa_volume = MaxChannelVolume(&channel_volumes);
    // Now map onto the SoundSystemInterface range.
    *volume = PulseVolumeToCricketVolume(pa_volume);

    ret = true;
   done:
    Unlock();
    return ret;
  }

  virtual bool SetVolume(int volume) {
    bool ret = false;
    pa_volume_t pa_volume = CricketVolumeToPulseVolume(volume);

    Lock();

    const pa_sample_spec *spec = symbol_table()->pa_stream_get_sample_spec()(
        stream_.stream());
    if (!spec) {
      LOG(LS_ERROR) << "pa_stream_get_sample_spec()";
      goto done;
    }

    pa_cvolume channel_volumes;
    symbol_table()->pa_cvolume_set()(&channel_volumes, spec->channels,
        pa_volume);

    pa_operation *op;
    op = symbol_table()->pa_context_set_sink_input_volume()(
        stream_.pulse()->context_,
        symbol_table()->pa_stream_get_index()(stream_.stream()),
        &channel_volumes,
        // This callback merely logs errors.
        &SetVolumeCallback,
        NULL);
    if (!op) {
      LOG(LS_ERROR) << "pa_context_set_sink_input_volume()";
      goto done;
    }
    // Don't need to wait for this to complete.
    symbol_table()->pa_operation_unref()(op);

    ret = true;
   done:
    Unlock();
    return ret;
  }

  virtual bool Close() {
    if (!DisableBufferMonitoring()) {
      return false;
    }
    bool ret = true;
    if (!stream_.IsClosed()) {
      Lock();
      symbol_table()->pa_stream_set_underflow_callback()(stream_.stream(),
                                                         NULL,
                                                         NULL);
      ret = stream_.Close();
      Unlock();
    }
    return ret;
  }

  virtual int LatencyUsecs() {
    return stream_.LatencyUsecs();
  }

#if 0
  // TODO: Versions 0.9.16 and later of Pulse have a new API for
  // zero-copy writes, but Hardy is not new enough to have that so we can't
  // rely on it. Perhaps auto-detect if it's present or not and use it if we
  // can?

  virtual bool GetWriteBuffer(void **buffer, size_t *size) {
    bool ret = true;
    Lock();
    if (symbol_table()->pa_stream_begin_write()(stream_.stream(), buffer, size)
            != 0) {
      LOG(LS_ERROR) << "Can't get write buffer";
      ret = false;
    }
    Unlock();
    return ret;
  }

  // Releases the caller's hold on the write buffer. "written" must be the
  // amount of data that was written.
  virtual bool ReleaseWriteBuffer(void *buffer, size_t written) {
    bool ret = true;
    Lock();
    if (written == 0) {
      if (symbol_table()->pa_stream_cancel_write()(stream_.stream()) != 0) {
        LOG(LS_ERROR) << "Can't cancel write";
        ret = false;
      }
    } else {
      if (symbol_table()->pa_stream_write()(stream_.stream(),
                                            buffer,
                                            written,
                                            NULL,
                                            0,
                                            PA_SEEK_RELATIVE) != 0) {
        LOG(LS_ERROR) << "Unable to write";
        ret = false;
      }
    }
    Unlock();
    return ret;
  }
#endif

 private:
  void Lock() {
    stream_.Lock();
  }

  void Unlock() {
    stream_.Unlock();
  }

  PulseAudioSymbolTable *symbol_table() {
    return stream_.symbol_table();
  }

  void EnableWriteCallback() {
    pa_stream_state_t state = symbol_table()->pa_stream_get_state()(
        stream_.stream());
    if (state == PA_STREAM_READY) {
      // May already have available space. Must check.
      temp_buffer_space_ = symbol_table()->pa_stream_writable_size()(
          stream_.stream());
      if (temp_buffer_space_ > 0) {
        // Yup, there is already space available, so if we register a write
        // callback then it will not receive any event. So dispatch one ourself
        // instead.
        HaveWork();
        return;
      }
    }
    symbol_table()->pa_stream_set_write_callback()(
         stream_.stream(),
         &WriteCallbackThunk,
         this);
  }

  void DisableWriteCallback() {
    symbol_table()->pa_stream_set_write_callback()(
         stream_.stream(),
         NULL,
         NULL);
  }

  static void WriteCallbackThunk(pa_stream *unused,
                                 size_t buffer_space,
                                 void *userdata) {
    PulseAudioOutputStream *instance =
        static_cast<PulseAudioOutputStream *>(userdata);
    instance->OnWriteCallback(buffer_space);
  }

  void OnWriteCallback(size_t buffer_space) {
    temp_buffer_space_ = buffer_space;
    // Since we write the data asynchronously on a different thread, we have
    // to temporarily disable the write callback or else Pulse will call it
    // continuously until we write the data. We re-enable it below.
    DisableWriteCallback();
    HaveWork();
  }

  // Inherited from Worker.
  virtual void OnStart() {
    Lock();
    EnableWriteCallback();
    Unlock();
  }

  // Inherited from Worker.
  virtual void OnHaveWork() {
    ASSERT(temp_buffer_space_ > 0);

    SignalBufferSpace(temp_buffer_space_, this);

    temp_buffer_space_ = 0;
    Lock();
    EnableWriteCallback();
    Unlock();
  }

  // Inherited from Worker.
  virtual void OnStop() {
    Lock();
    DisableWriteCallback();
    Unlock();
  }

  static void UnderflowCallbackThunk(pa_stream *unused,
                                     void *userdata) {
    PulseAudioOutputStream *instance =
        static_cast<PulseAudioOutputStream *>(userdata);
    instance->OnUnderflowCallback();
  }

  void OnUnderflowCallback() {
    LOG(LS_WARNING) << "Buffer underflow on playback stream "
                    << stream_.stream();

    if (configured_latency_ == SoundSystemInterface::kNoLatencyRequirements) {
      // We didn't configure a pa_buffer_attr before, so switching to one now
      // would be questionable.
      return;
    }

    // Otherwise reconfigure the stream with a higher target latency.

    const pa_sample_spec *spec = symbol_table()->pa_stream_get_sample_spec()(
        stream_.stream());
    if (!spec) {
      LOG(LS_ERROR) << "pa_stream_get_sample_spec()";
      return;
    }

    size_t bytes_per_sec = symbol_table()->pa_bytes_per_second()(spec);

    int new_latency = configured_latency_ +
        bytes_per_sec * kPlaybackLatencyIncrementMsecs /
        rtc::kNumMicrosecsPerSec;

    pa_buffer_attr new_attr = {0};
    FillPlaybackBufferAttr(new_latency, &new_attr);

    pa_operation *op = symbol_table()->pa_stream_set_buffer_attr()(
        stream_.stream(),
        &new_attr,
        // No callback.
        NULL,
        NULL);
    if (!op) {
      LOG(LS_ERROR) << "pa_stream_set_buffer_attr()";
      return;
    }
    // Don't need to wait for this to complete.
    symbol_table()->pa_operation_unref()(op);

    // Save the new latency in case we underflow again.
    configured_latency_ = new_latency;
  }

  static void GetVolumeCallbackThunk(pa_context *unused,
                                     const pa_sink_input_info *info,
                                     int eol,
                                     void *userdata) {
    GetVolumeCallbackData *data =
        static_cast<GetVolumeCallbackData *>(userdata);
    data->instance->OnGetVolumeCallback(info, eol, &data->channel_volumes);
  }

  void OnGetVolumeCallback(const pa_sink_input_info *info,
                           int eol,
                           pa_cvolume **channel_volumes) {
    if (eol) {
      // List is over. Wake GetVolume().
      stream_.pulse()->Signal();
      return;
    }

    if (*channel_volumes) {
      **channel_volumes = info->volume;
      // Unset the pointer so that we know that we have have already copied the
      // volume.
      *channel_volumes = NULL;
    } else {
      // We have received an additional callback after the first one, which
      // doesn't make sense for a single sink input. This probably never
      // happens, but we code for it anyway.
      LOG(LS_WARNING) << "Ignoring extra GetVolumeCallback";
    }
  }

  static void SetVolumeCallback(pa_context *unused1,
                                int success,
                                void *unused2) {
    if (!success) {
      LOG(LS_ERROR) << "Failed to change playback volume";
    }
  }

  PulseAudioStream stream_;
  int configured_latency_;
  // Temporary storage for passing data between threads.
  size_t temp_buffer_space_;

  DISALLOW_COPY_AND_ASSIGN(PulseAudioOutputStream);
};

PulseAudioSoundSystem::PulseAudioSoundSystem()
    : mainloop_(NULL), context_(NULL) {
}

PulseAudioSoundSystem::~PulseAudioSoundSystem() {
  Terminate();
}

bool PulseAudioSoundSystem::Init() {
  if (IsInitialized()) {
    return true;
  }

  // Load libpulse.
  if (!symbol_table_.Load()) {
    // Most likely the Pulse library and sound server are not installed on
    // this system.
    LOG(LS_WARNING) << "Failed to load symbol table";
    return false;
  }

  // Now create and start the Pulse event thread.
  mainloop_ = symbol_table_.pa_threaded_mainloop_new()();
  if (!mainloop_) {
    LOG(LS_ERROR) << "Can't create mainloop";
    goto fail0;
  }

  if (symbol_table_.pa_threaded_mainloop_start()(mainloop_) != 0) {
    LOG(LS_ERROR) << "Can't start mainloop";
    goto fail1;
  }

  Lock();
  context_ = CreateNewConnection();
  Unlock();

  if (!context_) {
    goto fail2;
  }

  // Otherwise we're now ready!
  return true;

 fail2:
  symbol_table_.pa_threaded_mainloop_stop()(mainloop_);
 fail1:
  symbol_table_.pa_threaded_mainloop_free()(mainloop_);
  mainloop_ = NULL;
 fail0:
  return false;
}

void PulseAudioSoundSystem::Terminate() {
  if (!IsInitialized()) {
    return;
  }

  Lock();
  symbol_table_.pa_context_disconnect()(context_);
  symbol_table_.pa_context_unref()(context_);
  Unlock();
  context_ = NULL;
  symbol_table_.pa_threaded_mainloop_stop()(mainloop_);
  symbol_table_.pa_threaded_mainloop_free()(mainloop_);
  mainloop_ = NULL;

  // We do not unload the symbol table because we may need it again soon if
  // Init() is called again.
}

bool PulseAudioSoundSystem::EnumeratePlaybackDevices(
    SoundDeviceLocatorList *devices) {
  return EnumerateDevices<pa_sink_info>(
      devices,
      symbol_table_.pa_context_get_sink_info_list(),
      &EnumeratePlaybackDevicesCallbackThunk);
}

bool PulseAudioSoundSystem::EnumerateCaptureDevices(
    SoundDeviceLocatorList *devices) {
  return EnumerateDevices<pa_source_info>(
      devices,
      symbol_table_.pa_context_get_source_info_list(),
      &EnumerateCaptureDevicesCallbackThunk);
}

bool PulseAudioSoundSystem::GetDefaultPlaybackDevice(
    SoundDeviceLocator **device) {
  return GetDefaultDevice<&pa_server_info::default_sink_name>(device);
}

bool PulseAudioSoundSystem::GetDefaultCaptureDevice(
    SoundDeviceLocator **device) {
  return GetDefaultDevice<&pa_server_info::default_source_name>(device);
}

SoundOutputStreamInterface *PulseAudioSoundSystem::OpenPlaybackDevice(
    const SoundDeviceLocator *device,
    const OpenParams &params) {
  return OpenDevice<SoundOutputStreamInterface>(
      device,
      params,
      "Playback",
      &PulseAudioSoundSystem::ConnectOutputStream);
}

SoundInputStreamInterface *PulseAudioSoundSystem::OpenCaptureDevice(
    const SoundDeviceLocator *device,
    const OpenParams &params) {
  return OpenDevice<SoundInputStreamInterface>(
      device,
      params,
      "Capture",
      &PulseAudioSoundSystem::ConnectInputStream);
}

const char *PulseAudioSoundSystem::GetName() const {
  return "PulseAudio";
}

inline bool PulseAudioSoundSystem::IsInitialized() {
  return mainloop_ != NULL;
}

struct ConnectToPulseCallbackData {
  PulseAudioSoundSystem *instance;
  bool connect_done;
};

void PulseAudioSoundSystem::ConnectToPulseCallbackThunk(
    pa_context *context, void *userdata) {
  ConnectToPulseCallbackData *data =
      static_cast<ConnectToPulseCallbackData *>(userdata);
  data->instance->OnConnectToPulseCallback(context, &data->connect_done);
}

void PulseAudioSoundSystem::OnConnectToPulseCallback(
    pa_context *context, bool *connect_done) {
  pa_context_state_t state = symbol_table_.pa_context_get_state()(context);
  if (state == PA_CONTEXT_READY ||
      state == PA_CONTEXT_FAILED ||
      state == PA_CONTEXT_TERMINATED) {
    // Connection process has reached a terminal state. Wake ConnectToPulse().
    *connect_done = true;
    Signal();
  }
}

// Must be called with the lock held.
bool PulseAudioSoundSystem::ConnectToPulse(pa_context *context) {
  bool ret = true;
  ConnectToPulseCallbackData data;
  // Have to put this up here to satisfy the compiler.
  pa_context_state_t state;

  data.instance = this;
  data.connect_done = false;

  symbol_table_.pa_context_set_state_callback()(context,
                                                &ConnectToPulseCallbackThunk,
                                                &data);

  // Connect to PulseAudio sound server.
  if (symbol_table_.pa_context_connect()(
          context,
          NULL,          // Default server
          PA_CONTEXT_NOAUTOSPAWN,
          NULL) != 0) {  // No special fork handling needed
    LOG(LS_ERROR) << "Can't start connection to PulseAudio sound server";
    ret = false;
    goto done;
  }

  // Wait for the connection state machine to reach a terminal state.
  do {
    Wait();
  } while (!data.connect_done);

  // Now check to see what final state we reached.
  state = symbol_table_.pa_context_get_state()(context);

  if (state != PA_CONTEXT_READY) {
    if (state == PA_CONTEXT_FAILED) {
      LOG(LS_ERROR) << "Failed to connect to PulseAudio sound server";
    } else if (state == PA_CONTEXT_TERMINATED) {
      LOG(LS_ERROR) << "PulseAudio connection terminated early";
    } else {
      // Shouldn't happen, because we only signal on one of those three states.
      LOG(LS_ERROR) << "Unknown problem connecting to PulseAudio";
    }
    ret = false;
  }

 done:
  // We unset our callback for safety just in case the state might somehow
  // change later, because the pointer to "data" will be invalid after return
  // from this function.
  symbol_table_.pa_context_set_state_callback()(context, NULL, NULL);
  return ret;
}

// Must be called with the lock held.
pa_context *PulseAudioSoundSystem::CreateNewConnection() {
  // Create connection context.
  std::string app_name;
  // TODO: Pulse etiquette says this name should be localized. Do
  // we care?
  rtc::Filesystem::GetApplicationName(&app_name);
  pa_context *context = symbol_table_.pa_context_new()(
      symbol_table_.pa_threaded_mainloop_get_api()(mainloop_),
      app_name.c_str());
  if (!context) {
    LOG(LS_ERROR) << "Can't create context";
    goto fail0;
  }

  // Now connect.
  if (!ConnectToPulse(context)) {
    goto fail1;
  }

  // Otherwise the connection succeeded and is ready.
  return context;

 fail1:
  symbol_table_.pa_context_unref()(context);
 fail0:
  return NULL;
}

struct EnumerateDevicesCallbackData {
  PulseAudioSoundSystem *instance;
  SoundSystemInterface::SoundDeviceLocatorList *devices;
};

void PulseAudioSoundSystem::EnumeratePlaybackDevicesCallbackThunk(
    pa_context *unused,
    const pa_sink_info *info,
    int eol,
    void *userdata) {
  EnumerateDevicesCallbackData *data =
      static_cast<EnumerateDevicesCallbackData *>(userdata);
  data->instance->OnEnumeratePlaybackDevicesCallback(data->devices, info, eol);
}

void PulseAudioSoundSystem::EnumerateCaptureDevicesCallbackThunk(
    pa_context *unused,
    const pa_source_info *info,
    int eol,
    void *userdata) {
  EnumerateDevicesCallbackData *data =
      static_cast<EnumerateDevicesCallbackData *>(userdata);
  data->instance->OnEnumerateCaptureDevicesCallback(data->devices, info, eol);
}

void PulseAudioSoundSystem::OnEnumeratePlaybackDevicesCallback(
    SoundDeviceLocatorList *devices,
    const pa_sink_info *info,
    int eol) {
  if (eol) {
    // List is over. Wake EnumerateDevices().
    Signal();
    return;
  }

  // Else this is the next device.
  devices->push_back(
      new PulseAudioDeviceLocator(info->description, info->name));
}

void PulseAudioSoundSystem::OnEnumerateCaptureDevicesCallback(
    SoundDeviceLocatorList *devices,
    const pa_source_info *info,
    int eol) {
  if (eol) {
    // List is over. Wake EnumerateDevices().
    Signal();
    return;
  }

  if (info->monitor_of_sink != PA_INVALID_INDEX) {
    // We don't want to list monitor sources, since they are almost certainly
    // not what the user wants for voice conferencing.
    return;
  }

  // Else this is the next device.
  devices->push_back(
      new PulseAudioDeviceLocator(info->description, info->name));
}

template <typename InfoStruct>
bool PulseAudioSoundSystem::EnumerateDevices(
    SoundDeviceLocatorList *devices,
    pa_operation *(*enumerate_fn)(
        pa_context *c,
        void (*callback_fn)(
            pa_context *c,
            const InfoStruct *i,
            int eol,
            void *userdata),
        void *userdata),
    void (*callback_fn)(
        pa_context *c,
        const InfoStruct *i,
        int eol,
        void *userdata)) {
  ClearSoundDeviceLocatorList(devices);
  if (!IsInitialized()) {
    return false;
  }

  EnumerateDevicesCallbackData data;
  data.instance = this;
  data.devices = devices;

  Lock();
  pa_operation *op = (*enumerate_fn)(
      context_,
      callback_fn,
      &data);
  bool ret = FinishOperation(op);
  Unlock();
  return ret;
}

struct GetDefaultDeviceCallbackData {
  PulseAudioSoundSystem *instance;
  SoundDeviceLocator **device;
};

template <const char *(pa_server_info::*field)>
void PulseAudioSoundSystem::GetDefaultDeviceCallbackThunk(
    pa_context *unused,
    const pa_server_info *info,
    void *userdata) {
  GetDefaultDeviceCallbackData *data =
      static_cast<GetDefaultDeviceCallbackData *>(userdata);
  data->instance->OnGetDefaultDeviceCallback<field>(info, data->device);
}

template <const char *(pa_server_info::*field)>
void PulseAudioSoundSystem::OnGetDefaultDeviceCallback(
    const pa_server_info *info,
    SoundDeviceLocator **device) {
  if (info) {
    const char *dev = info->*field;
    if (dev) {
      *device = new PulseAudioDeviceLocator("Default device", dev);
    }
  }
  Signal();
}

template <const char *(pa_server_info::*field)>
bool PulseAudioSoundSystem::GetDefaultDevice(SoundDeviceLocator **device) {
  if (!IsInitialized()) {
    return false;
  }
  bool ret;
  *device = NULL;
  GetDefaultDeviceCallbackData data;
  data.instance = this;
  data.device = device;
  Lock();
  pa_operation *op = symbol_table_.pa_context_get_server_info()(
      context_,
      &GetDefaultDeviceCallbackThunk<field>,
      &data);
  ret = FinishOperation(op);
  Unlock();
  return ret && (*device != NULL);
}

void PulseAudioSoundSystem::StreamStateChangedCallbackThunk(
    pa_stream *stream,
    void *userdata) {
  PulseAudioSoundSystem *instance =
      static_cast<PulseAudioSoundSystem *>(userdata);
  instance->OnStreamStateChangedCallback(stream);
}

void PulseAudioSoundSystem::OnStreamStateChangedCallback(pa_stream *stream) {
  pa_stream_state_t state = symbol_table_.pa_stream_get_state()(stream);
  if (state == PA_STREAM_READY) {
    LOG(LS_INFO) << "Pulse stream " << stream << " ready";
  } else if (state == PA_STREAM_FAILED ||
             state == PA_STREAM_TERMINATED ||
             state == PA_STREAM_UNCONNECTED) {
    LOG(LS_ERROR) << "Pulse stream " << stream << " failed to connect: "
                  << LastError();
  }
}

template <typename StreamInterface>
StreamInterface *PulseAudioSoundSystem::OpenDevice(
    const SoundDeviceLocator *device,
    const OpenParams &params,
    const char *stream_name,
    StreamInterface *(PulseAudioSoundSystem::*connect_fn)(
        pa_stream *stream,
        const char *dev,
        int flags,
        pa_stream_flags_t pa_flags,
        int latency,
        const pa_sample_spec &spec)) {
  if (!IsInitialized()) {
    return NULL;
  }

  const char *dev = static_cast<const PulseAudioDeviceLocator *>(device)->
      device_name().c_str();

  StreamInterface *stream_interface = NULL;

  ASSERT(params.format < ARRAY_SIZE(kCricketFormatToPulseFormatTable));

  pa_sample_spec spec;
  spec.format = kCricketFormatToPulseFormatTable[params.format];
  spec.rate = params.freq;
  spec.channels = params.channels;

  int pa_flags = 0;
  if (params.flags & FLAG_REPORT_LATENCY) {
    pa_flags |= PA_STREAM_INTERPOLATE_TIMING |
                PA_STREAM_AUTO_TIMING_UPDATE;
  }

  if (params.latency != kNoLatencyRequirements) {
    // If configuring a specific latency then we want to specify
    // PA_STREAM_ADJUST_LATENCY to make the server adjust parameters
    // automatically to reach that target latency. However, that flag doesn't
    // exist in Ubuntu 8.04 and many people still use that, so we have to check
    // the protocol version of libpulse.
    if (symbol_table_.pa_context_get_protocol_version()(context_) >=
        kAdjustLatencyProtocolVersion) {
      pa_flags |= PA_STREAM_ADJUST_LATENCY;
    }
  }

  Lock();

  pa_stream *stream = symbol_table_.pa_stream_new()(context_, stream_name,
      &spec, NULL);
  if (!stream) {
    LOG(LS_ERROR) << "Can't create pa_stream";
    goto done;
  }

  // Set a state callback to log errors.
  symbol_table_.pa_stream_set_state_callback()(stream,
                                               &StreamStateChangedCallbackThunk,
                                               this);

  stream_interface = (this->*connect_fn)(
      stream,
      dev,
      params.flags,
      static_cast<pa_stream_flags_t>(pa_flags),
      params.latency,
      spec);
  if (!stream_interface) {
    LOG(LS_ERROR) << "Can't connect stream to " << dev;
    symbol_table_.pa_stream_unref()(stream);
  }

 done:
  Unlock();
  return stream_interface;
}

// Must be called with the lock held.
SoundOutputStreamInterface *PulseAudioSoundSystem::ConnectOutputStream(
    pa_stream *stream,
    const char *dev,
    int flags,
    pa_stream_flags_t pa_flags,
    int latency,
    const pa_sample_spec &spec) {
  pa_buffer_attr attr = {0};
  pa_buffer_attr *pattr = NULL;
  if (latency != kNoLatencyRequirements) {
    // kLowLatency is 0, so we treat it the same as a request for zero latency.
    ssize_t bytes_per_sec = symbol_table_.pa_bytes_per_second()(&spec);
    latency = std::max(
        latency, static_cast<int>(bytes_per_sec * kPlaybackLatencyMinimumMsecs /
                                  rtc::kNumMicrosecsPerSec));
    FillPlaybackBufferAttr(latency, &attr);
    pattr = &attr;
  }
  if (symbol_table_.pa_stream_connect_playback()(
          stream,
          dev,
          pattr,
          pa_flags,
          // Let server choose volume
          NULL,
          // Not synchronized to any other playout
          NULL) != 0) {
    return NULL;
  }
  return new PulseAudioOutputStream(this, stream, flags, latency);
}

// Must be called with the lock held.
SoundInputStreamInterface *PulseAudioSoundSystem::ConnectInputStream(
    pa_stream *stream,
    const char *dev,
    int flags,
    pa_stream_flags_t pa_flags,
    int latency,
    const pa_sample_spec &spec) {
  pa_buffer_attr attr = {0};
  pa_buffer_attr *pattr = NULL;
  if (latency != kNoLatencyRequirements) {
    size_t bytes_per_sec = symbol_table_.pa_bytes_per_second()(&spec);
    if (latency == kLowLatency) {
      latency = bytes_per_sec * kLowCaptureLatencyMsecs /
          rtc::kNumMicrosecsPerSec;
    }
    // Note: fragsize specifies a maximum transfer size, not a minimum, so it is
    // not possible to force a high latency setting, only a low one.
    attr.fragsize = latency;
    attr.maxlength = latency + bytes_per_sec * kCaptureBufferExtraMsecs /
        rtc::kNumMicrosecsPerSec;
    LOG(LS_VERBOSE) << "Configuring latency = " << attr.fragsize
                    << ", maxlength = " << attr.maxlength;
    pattr = &attr;
  }
  if (symbol_table_.pa_stream_connect_record()(stream,
                                               dev,
                                               pattr,
                                               pa_flags) != 0) {
    return NULL;
  }
  return new PulseAudioInputStream(this, stream, flags);
}

// Must be called with the lock held.
bool PulseAudioSoundSystem::FinishOperation(pa_operation *op) {
  if (!op) {
    LOG(LS_ERROR) << "Failed to start operation";
    return false;
  }

  do {
    Wait();
  } while (symbol_table_.pa_operation_get_state()(op) == PA_OPERATION_RUNNING);

  symbol_table_.pa_operation_unref()(op);

  return true;
}

inline void PulseAudioSoundSystem::Lock() {
  symbol_table_.pa_threaded_mainloop_lock()(mainloop_);
}

inline void PulseAudioSoundSystem::Unlock() {
  symbol_table_.pa_threaded_mainloop_unlock()(mainloop_);
}

// Must be called with the lock held.
inline void PulseAudioSoundSystem::Wait() {
  symbol_table_.pa_threaded_mainloop_wait()(mainloop_);
}

// Must be called with the lock held.
inline void PulseAudioSoundSystem::Signal() {
  symbol_table_.pa_threaded_mainloop_signal()(mainloop_, 0);
}

// Must be called with the lock held.
const char *PulseAudioSoundSystem::LastError() {
  return symbol_table_.pa_strerror()(symbol_table_.pa_context_errno()(
      context_));
}

}  // namespace rtc

#endif  // HAVE_LIBPULSE