AudioStream: Replace buffer queue with ring buffer

Should achieve a decent overall minimum latency reduction.

src/common/audio_stream.cpp

@@ -1,14 +1,16 @@
 #include "audio_stream.h"
 #include "assert.h"
+#include "log.h"
 #include <algorithm>
 #include <cstring>
+Log_SetChannel(AudioStream);
 
 AudioStream::AudioStream() = default;
 
 AudioStream::~AudioStream() = default;
 
 bool AudioStream::Reconfigure(u32 output_sample_rate /*= DefaultOutputSampleRate*/, u32 channels /*= 1*/,
-                              u32 buffer_size /*= DefaultBufferSize*/, u32 buffer_count /*= DefaultBufferCount*/)
+                              u32 buffer_size /*= DefaultBufferSize*/)
 {
   if (IsDeviceOpen())
     CloseDevice();
@@ -16,13 +18,14 @@ bool AudioStream::Reconfigure(u32 output_sample_rate /*= DefaultOutputSampleRate
   m_output_sample_rate = output_sample_rate;
   m_channels = channels;
   m_buffer_size = buffer_size;
-  AllocateBuffers(buffer_count);
   m_output_paused = true;
 
+  if (!SetBufferSize(buffer_size))
+    return false;
+
   if (!OpenDevice())
   {
     EmptyBuffers();
-    m_buffers.clear();
     m_buffer_size = 0;
     m_output_sample_rate = 0;
     m_channels = 0;
@@ -32,7 +35,7 @@ bool AudioStream::Reconfigure(u32 output_sample_rate /*= DefaultOutputSampleRate
   return true;
 }
 
-void AudioStream::SetOutputVolume(s32 volume)
+void AudioStream::SetOutputVolume(u32 volume)
 {
   std::unique_lock<std::mutex> lock(m_buffer_mutex);
   m_output_volume = volume;
@@ -58,7 +61,6 @@ void AudioStream::Shutdown()
 
   CloseDevice();
   EmptyBuffers();
-  m_buffers.clear();
   m_buffer_size = 0;
   m_output_sample_rate = 0;
   m_channels = 0;
@@ -69,182 +71,149 @@ void AudioStream::BeginWrite(SampleType** buffer_ptr, u32* num_frames)
 {
   m_buffer_mutex.lock();
 
-  EnsureBuffer();
+  EnsureBuffer(*num_frames * m_channels);
 
-  Buffer& buffer = m_buffers[m_first_free_buffer];
-  *buffer_ptr = buffer.data.data() + (buffer.write_position * m_channels);
-  *num_frames = m_buffer_size - buffer.write_position;
+  *buffer_ptr = m_buffer.GetWritePointer();
+  *num_frames = m_buffer.GetContiguousSpace() / m_channels;
 }
 
 void AudioStream::WriteFrames(const SampleType* frames, u32 num_frames)
 {
-  u32 remaining_frames = num_frames;
+  const u32 num_samples = num_frames * m_channels;
   std::unique_lock<std::mutex> lock(m_buffer_mutex);
 
-  while (remaining_frames > 0)
-  {
-    EnsureBuffer();
-
-    Buffer& buffer = m_buffers[m_first_free_buffer];
-    const u32 to_this_buffer = std::min(m_buffer_size - buffer.write_position, remaining_frames);
-
-    const u32 copy_count = to_this_buffer * m_channels;
-    std::memcpy(&buffer.data[buffer.write_position * m_channels], frames, copy_count * sizeof(SampleType));
-    frames += copy_count;
-
-    remaining_frames -= to_this_buffer;
-    buffer.write_position += to_this_buffer;
-
-    // End of the buffer?
-    if (buffer.write_position == m_buffer_size)
-    {
-      // Reset it back to the start, and enqueue it.
-      buffer.write_position = 0;
-      m_num_free_buffers--;
-      m_first_free_buffer = (m_first_free_buffer + 1) % m_buffers.size();
-      m_num_available_buffers++;
-      BufferAvailable();
-    }
-  }
+  EnsureBuffer(num_samples);
+  m_buffer.PushRange(frames, num_samples);
+  FramesAvailable();
 }
 
 void AudioStream::EndWrite(u32 num_frames)
 {
-  Buffer& buffer = m_buffers[m_first_free_buffer];
-  DebugAssert((buffer.write_position + num_frames) <= m_buffer_size);
-  buffer.write_position += num_frames;
-
-  // End of the buffer?
-  if (buffer.write_position == m_buffer_size)
-  {
-    // Reset it back to the start, and enqueue it.
-    // Log_DevPrintf("Enqueue buffer %u", m_first_free_buffer);
-    buffer.write_position = 0;
-    m_num_free_buffers--;
-    m_first_free_buffer = (m_first_free_buffer + 1) % m_buffers.size();
-    m_num_available_buffers++;
-    BufferAvailable();
-  }
+  m_buffer.AdvanceTail(num_frames * m_channels);
+  FramesAvailable();
 
   m_buffer_mutex.unlock();
 }
 
-float AudioStream::GetMinLatency(u32 sample_rate, u32 buffer_size, u32 buffer_count)
+float AudioStream::GetMaxLatency(u32 sample_rate, u32 buffer_size)
 {
   return (static_cast<float>(buffer_size) / static_cast<float>(sample_rate));
 }
 
-float AudioStream::GetMaxLatency(u32 sample_rate, u32 buffer_size, u32 buffer_count)
+bool AudioStream::SetBufferSize(u32 buffer_size)
 {
-  return (static_cast<float>(buffer_size * (buffer_count - 1)) / static_cast<float>(sample_rate));
+  const u32 buffer_size_in_samples = buffer_size * m_channels;
+  const u32 max_samples = buffer_size_in_samples * 2u;
+  if (max_samples > m_buffer.GetCapacity())
+    return false;
+
+  m_buffer_size = buffer_size;
+  m_max_samples = max_samples;
+  return true;
 }
 
 u32 AudioStream::GetSamplesAvailable() const
 {
   // TODO: Use atomic loads
-  u32 available_buffers;
+  u32 available_samples;
   {
     std::unique_lock<std::mutex> lock(m_buffer_mutex);
-    available_buffers = m_num_available_buffers;
+    available_samples = m_buffer.GetSize();
   }
 
-  return available_buffers * m_buffer_size;
+  return available_samples / m_channels;
 }
 
-u32 AudioStream::ReadSamples(SampleType* samples, u32 num_samples)
+u32 AudioStream::GetSamplesAvailableLocked() const
 {
-  u32 remaining_samples = num_samples;
-  std::unique_lock<std::mutex> lock(m_buffer_mutex);
+  return m_buffer.GetSize() / m_channels;
+}
 
-  while (remaining_samples > 0 && m_num_available_buffers > 0)
+void AudioStream::ReadFrames(SampleType* samples, u32 num_frames, bool apply_volume)
+{
+  const u32 total_samples = num_frames * m_channels;
+  u32 samples_copied = 0;
   {
-    Buffer& buffer = m_buffers[m_first_available_buffer];
-    const u32 from_this_buffer = std::min(m_buffer_size - buffer.read_position, remaining_samples);
+    std::unique_lock<std::mutex> lock(m_buffer_mutex);
+    samples_copied = std::min(m_buffer.GetSize(), total_samples);
+    if (samples_copied > 0)
+      m_buffer.PopRange(samples, samples_copied);
 
-    const u32 copy_count = from_this_buffer * m_channels;
-    const SampleType* read_pointer = &buffer.data[buffer.read_position * m_channels];
-    for (u32 i = 0; i < copy_count; i++)
-      *(samples++) = ApplyVolume(*(read_pointer++), m_output_volume);
+    m_buffer_draining_cv.notify_one();
+  }
 
-    remaining_samples -= from_this_buffer;
-    buffer.read_position += from_this_buffer;
-
-    if (buffer.read_position == m_buffer_size)
+  if (samples_copied < total_samples)
+  {
+    if (samples_copied > 0)
     {
-      // Log_DevPrintf("Finish dequeing buffer %u", m_first_available_buffer);
-      // End of this buffer.
-      buffer.read_position = 0;
-      m_num_available_buffers--;
-      m_first_available_buffer = (m_first_available_buffer + 1) % m_buffers.size();
-      m_num_free_buffers++;
-      m_buffer_available_cv.notify_one();
+      m_resample_buffer.resize(samples_copied);
+      std::memcpy(m_resample_buffer.data(), samples, sizeof(SampleType) * samples_copied);
+
+      // super basic resampler - spread the input samples evenly across the output samples. will sound like ass and have
+      // aliasing, but better than popping by inserting silence.
+      const u32 increment =
+        static_cast<u32>(65536.0f * (static_cast<float>(samples_copied / m_channels) / static_cast<float>(num_frames)));
+
+      SampleType* out_ptr = samples;
+      const SampleType* resample_ptr = m_resample_buffer.data();
+      const u32 copy_stride = sizeof(SampleType) * m_channels;
+      u32 resample_subpos = 0;
+      for (u32 i = 0; i < num_frames; i++)
+      {
+        std::memcpy(out_ptr, resample_ptr, copy_stride);
+        out_ptr += m_channels;
+
+        resample_subpos += increment;
+        resample_ptr += (resample_subpos >> 16) * m_channels;
+        resample_subpos %= 65536u;
+      }
+
+      Log_DevPrintf("Audio buffer underflow, resampled %u frames to %u", samples_copied / m_channels, num_frames);
+    }
+    else
+    {
+      // read nothing, so zero-fill
+      std::memset(samples, 0, sizeof(SampleType) * total_samples);
+      Log_DevPrintf("Audio buffer underflow with no samples, added %u frames silence", num_frames);
     }
   }
 
-  return num_samples - remaining_samples;
-}
-
-void AudioStream::AllocateBuffers(u32 buffer_count)
-{
-  m_buffers.resize(buffer_count);
-  for (u32 i = 0; i < buffer_count; i++)
+  if (apply_volume && m_output_volume != FullVolume)
   {
-    Buffer& buffer = m_buffers[i];
-    buffer.data.resize(m_buffer_size * m_channels);
-    buffer.read_position = 0;
-    buffer.write_position = 0;
+    SampleType* current_ptr = samples;
+    const SampleType* end_ptr = samples + (num_frames * m_channels);
+    while (current_ptr != end_ptr)
+    {
+      *current_ptr = ApplyVolume(*current_ptr, m_output_volume);
+      current_ptr++;
+    }
   }
-
-  m_first_available_buffer = 0;
-  m_num_available_buffers = 0;
-  m_first_free_buffer = 0;
-  m_num_free_buffers = buffer_count;
 }
 
-void AudioStream::EnsureBuffer()
+void AudioStream::EnsureBuffer(u32 size)
 {
-  if (m_num_free_buffers > 0)
+  if (GetBufferSpace() >= size)
     return;
 
   if (m_sync)
   {
     std::unique_lock<std::mutex> lock(m_buffer_mutex, std::adopt_lock);
-    m_buffer_available_cv.wait(lock, [this]() { return m_num_free_buffers > 0; });
+    m_buffer_draining_cv.wait(lock, [this, size]() { return GetBufferSpace() >= size; });
     lock.release();
   }
   else
   {
-    DropBuffer();
+    m_buffer.Remove(size);
   }
 }
 
-void AudioStream::DropBuffer()
+void AudioStream::DropFrames(u32 count)
 {
-  DebugAssert(m_num_available_buffers > 0);
-  // Log_DevPrintf("Dropping buffer %u", m_first_free_buffer);
-
-  // Out of space. We'll overwrite the oldest buffer with the new data.
-  // At the same time, we shift the available buffer forward one.
-  m_first_available_buffer = (m_first_available_buffer + 1) % m_buffers.size();
-  m_num_available_buffers--;
-
-  m_buffers[m_first_free_buffer].read_position = 0;
-  m_buffers[m_first_free_buffer].write_position = 0;
-  m_num_free_buffers++;
+  m_buffer.Remove(count);
 }
 
 void AudioStream::EmptyBuffers()
 {
   std::unique_lock<std::mutex> lock(m_buffer_mutex);
-
-  for (Buffer& buffer : m_buffers)
-  {
-    buffer.read_position = 0;
-    buffer.write_position = 0;
-  }
-
-  m_first_free_buffer = 0;
-  m_num_free_buffers = static_cast<u32>(m_buffers.size());
-  m_first_available_buffer = 0;
-  m_num_available_buffers = 0;
+  m_buffer.Clear();
 }

src/common/audio_stream.h

@@ -1,4 +1,5 @@
 #pragma once
+#include "fifo_queue.h"
 #include "types.h"
 #include <condition_variable>
 #include <memory>
@@ -12,11 +13,12 @@ class AudioStream
 public:
   using SampleType = s16;
 
-  enum
+  enum : u32
   {
     DefaultOutputSampleRate = 44100,
     DefaultBufferSize = 2048,
-    DefaultBufferCount = 3,
+    MaxSamples = 32768,
+    FullVolume = 100
   };
 
   AudioStream();
@@ -25,14 +27,14 @@ public:
   u32 GetOutputSampleRate() const { return m_output_sample_rate; }
   u32 GetChannels() const { return m_channels; }
   u32 GetBufferSize() const { return m_buffer_size; }
-  u32 GetBufferCount() const { return static_cast<u32>(m_buffers.size()); }
   s32 GetOutputVolume() const { return m_output_volume; }
   bool IsSyncing() const { return m_sync; }
 
   bool Reconfigure(u32 output_sample_rate = DefaultOutputSampleRate, u32 channels = 1,
-                   u32 buffer_size = DefaultBufferSize, u32 buffer_count = DefaultBufferCount);
+                   u32 buffer_size = DefaultBufferSize);
   void SetSync(bool enable) { m_sync = enable; }
-  void SetOutputVolume(s32 volume);
+
+  virtual void SetOutputVolume(u32 volume);
 
   void PauseOutput(bool paused);
   void EmptyBuffers();
@@ -48,58 +50,43 @@ public:
   static std::unique_ptr<AudioStream> CreateCubebAudioStream();
 
   // Latency computation - returns values in seconds
-  static float GetMinLatency(u32 sample_rate, u32 buffer_size, u32 buffer_count);
-  static float GetMaxLatency(u32 sample_rate, u32 buffer_size, u32 buffer_count);
+  static float GetMaxLatency(u32 sample_rate, u32 buffer_size);
 
 protected:
   virtual bool OpenDevice() = 0;
   virtual void PauseDevice(bool paused) = 0;
   virtual void CloseDevice() = 0;
-  virtual void BufferAvailable() = 0;
+  virtual void FramesAvailable() = 0;
 
-  ALWAYS_INLINE static SampleType ApplyVolume(SampleType sample, s32 volume)
+  ALWAYS_INLINE static SampleType ApplyVolume(SampleType sample, u32 volume)
   {
-    return s16((s32(sample) * volume) / 100);
+    return s16((s32(sample) * s32(volume)) / 100);
   }
 
+  bool SetBufferSize(u32 buffer_size);
   bool IsDeviceOpen() const { return (m_output_sample_rate > 0); }
 
   u32 GetSamplesAvailable() const;
-  u32 ReadSamples(SampleType* samples, u32 num_samples);
-
-  void DropBuffer();
+  u32 GetSamplesAvailableLocked() const;
+  void ReadFrames(SampleType* samples, u32 num_frames, bool apply_volume);
+  void DropFrames(u32 count);
 
   u32 m_output_sample_rate = 0;
   u32 m_channels = 0;
   u32 m_buffer_size = 0;
 
-private:
-  struct Buffer
-  {
-    std::vector<SampleType> data;
-    u32 write_position;
-    u32 read_position;
-  };
-
-  void AllocateBuffers(u32 buffer_count);
-  void EnsureBuffer();
-
-  std::vector<Buffer> m_buffers;
-  mutable std::mutex m_buffer_mutex;
-
-  // For input.
-  u32 m_first_free_buffer = 0;
-  u32 m_num_free_buffers = 0;
-
-  // For output.
-  u32 m_num_available_buffers = 0;
-  u32 m_first_available_buffer = 0;
-
-  // TODO: Switch to semaphore
-  std::condition_variable m_buffer_available_cv;
-
   // volume, 0-100
-  s32 m_output_volume = 100;
+  u32 m_output_volume = FullVolume;
+
+private:
+  ALWAYS_INLINE u32 GetBufferSpace() const { return (m_max_samples - m_buffer.GetSize()); }
+  void EnsureBuffer(u32 size);
+
+  HeapFIFOQueue<SampleType, MaxSamples> m_buffer;
+  mutable std::mutex m_buffer_mutex;
+  std::condition_variable m_buffer_draining_cv;
+  std::vector<SampleType> m_resample_buffer;
+  u32 m_max_samples = 0;
 
   bool m_output_paused = true;
   bool m_sync = true;

src/common/cubeb_audio_stream.cpp

@@ -56,9 +56,21 @@ bool CubebAudioStream::OpenDevice()
 
   Log_InfoPrintf("Minimum latency in frames: %u", latency_frames);
   if (latency_frames > m_buffer_size)
-    Log_WarningPrintf("Minimum latency is above buffer size: %u vs %u", latency_frames, m_buffer_size);
+  {
+    Log_WarningPrintf("Minimum latency is above buffer size: %u vs %u, adjusting to compensate.", latency_frames,
+                      m_buffer_size);
+
+    if (!SetBufferSize(latency_frames))
+    {
+      Log_ErrorPrintf("Failed to set new buffer size of %u frames", latency_frames);
+      DestroyContext();
+      return false;
+    }
+  }
   else
+  {
     latency_frames = m_buffer_size;
+  }
 
   char stream_name[32];
   std::snprintf(stream_name, sizeof(stream_name), "AudioStream_%p", this);
@@ -72,6 +84,7 @@ bool CubebAudioStream::OpenDevice()
     return false;
   }
 
+  cubeb_stream_set_volume(m_cubeb_stream, static_cast<float>(m_output_volume) / 100.0f);
   return true;
 }
 
@@ -106,15 +119,13 @@ long CubebAudioStream::DataCallback(cubeb_stream* stm, void* user_ptr, const voi
 {
   CubebAudioStream* const this_ptr = static_cast<CubebAudioStream*>(user_ptr);
 
-  const u32 read_frames =
-    this_ptr->ReadSamples(reinterpret_cast<SampleType*>(output_buffer), static_cast<u32>(nframes));
-  const u32 silence_frames = static_cast<u32>(nframes) - read_frames;
-  if (silence_frames > 0)
+  if (this_ptr->m_output_volume_changed.load())
   {
-    std::memset(reinterpret_cast<SampleType*>(output_buffer) + (read_frames * this_ptr->m_channels), 0,
-                silence_frames * this_ptr->m_channels * sizeof(SampleType));
+    this_ptr->m_output_volume_changed.store(false);
+    cubeb_stream_set_volume(this_ptr->m_cubeb_stream, static_cast<float>(this_ptr->m_output_volume) / 100.0f);
   }
 
+  this_ptr->ReadFrames(reinterpret_cast<SampleType*>(output_buffer), static_cast<u32>(nframes), false);
   return nframes;
 }
 
@@ -125,7 +136,7 @@ void CubebAudioStream::StateCallback(cubeb_stream* stream, void* user_ptr, cubeb
   this_ptr->m_paused = (state != CUBEB_STATE_STARTED);
 }
 
-void CubebAudioStream::BufferAvailable() {}
+void CubebAudioStream::FramesAvailable() {}
 
 void CubebAudioStream::DestroyContext()
 {
@@ -138,6 +149,12 @@ void CubebAudioStream::DestroyContext()
 #endif
 }
 
+void CubebAudioStream::SetOutputVolume(u32 volume)
+{
+  AudioStream::SetOutputVolume(volume);
+  m_output_volume_changed.store(true);
+}
+
 std::unique_ptr<AudioStream> AudioStream::CreateCubebAudioStream()
 {
   return std::make_unique<CubebAudioStream>();

src/common/cubeb_audio_stream.h

@@ -1,6 +1,7 @@
 #pragma once
 #include "common/audio_stream.h"
 #include "cubeb/cubeb.h"
+#include <atomic>
 #include <cstdint>
 
 class CubebAudioStream final : public AudioStream
@@ -9,13 +10,15 @@ public:
   CubebAudioStream();
   ~CubebAudioStream();
 
+  void SetOutputVolume(u32 volume) override;
+
 protected:
   bool IsOpen() const { return m_cubeb_stream != nullptr; }
 
   bool OpenDevice() override;
   void PauseDevice(bool paused) override;
   void CloseDevice() override;
-  void BufferAvailable() override;
+  void FramesAvailable() override;
 
   void DestroyContext();
 
@@ -26,6 +29,7 @@ protected:
   cubeb* m_cubeb_context = nullptr;
   cubeb_stream* m_cubeb_stream = nullptr;
   bool m_paused = true;
+  std::atomic_bool m_output_volume_changed{ false };
 
 #ifdef WIN32
   bool m_com_initialized_by_us = false;

src/common/fifo_queue.h

@@ -17,9 +17,10 @@ class FIFOQueue
 public:
   const T* GetDataPointer() const { return m_ptr; }
   T* GetDataPointer() { return m_ptr; }
-  const T* GetFrontPointer() const { return &m_ptr[m_head]; }
-  T* GetFrontPointer() { return &m_ptr[m_head]; }
+  const T* GetReadPointer() const { return &m_ptr[m_head]; }
+  T* GetReadPointer() { return &m_ptr[m_head]; }
   constexpr u32 GetCapacity() const { return CAPACITY; }
+  T* GetWritePointer() { return &m_ptr[m_tail]; }
   u32 GetSize() const { return m_size; }
   u32 GetSpace() const { return CAPACITY - m_size; }
   u32 GetContiguousSpace() const { return (m_tail >= m_head) ? (CAPACITY - m_tail) : (m_head - m_tail); }
@@ -148,6 +149,14 @@ public:
     }
   }
 
+  void AdvanceTail(u32 count)
+  {
+    DebugAssert((m_size + count) < CAPACITY);
+    DebugAssert((m_tail + count) <= CAPACITY);
+    m_tail = (m_tail + count) % CAPACITY;
+    m_size += count;
+  }
+
 protected:
   FIFOQueue() = default;
 

src/common/null_audio_stream.cpp

@@ -13,10 +13,10 @@ void NullAudioStream::PauseDevice(bool paused) {}
 
 void NullAudioStream::CloseDevice() {}
 
-void NullAudioStream::BufferAvailable()
+void NullAudioStream::FramesAvailable()
 {
   // drop any buffer as soon as they're available
-  DropBuffer();
+  DropFrames(GetSamplesAvailableLocked());
 }
 
 std::unique_ptr<AudioStream> AudioStream::CreateNullAudioStream()

src/common/null_audio_stream.h

@@ -11,5 +11,5 @@ protected:
   bool OpenDevice() override;
   void PauseDevice(bool paused) override;
   void CloseDevice() override;
-  void BufferAvailable() override;
+  void FramesAvailable() override;
 };