[APU] Pace audio subsystem

src/xenia/apu/audio_system.cc

@@ -9,11 +9,14 @@
 
 #include "xenia/apu/audio_system.h"
 
+#include <limits>
+
 #include "xenia/apu/apu_flags.h"
 #include "xenia/apu/audio_driver.h"
 #include "xenia/apu/xma_decoder.h"
 #include "xenia/base/assert.h"
 #include "xenia/base/byte_stream.h"
+#include "xenia/base/clock.h"
 #include "xenia/base/logging.h"
 #include "xenia/base/math.h"
 #include "xenia/base/profiling.h"
@@ -35,13 +38,6 @@
 // and let the normal AudioSystem handling take it, to prevent duplicate
 // implementations. They can be found in xboxkrnl_audio_xma.cc
 
-DEFINE_uint32(apu_max_queued_frames, 8,
-              "Allows changing max buffered audio frames to reduce audio "
-              "delay. Lowering this value might cause performance issues. "
-              "Value range: [4-64]",
-              "APU");
-UPDATE_from_uint32(apu_max_queued_frames, 2024, 8, 31, 20, 64);
-
 namespace xe {
 namespace apu {
 
@@ -50,17 +46,13 @@ AudioSystem::AudioSystem(cpu::Processor* processor)
       processor_(processor),
       worker_running_(false) {
   std::memset(clients_, 0, sizeof(clients_));
-  queued_frames_ = std::clamp(cvars::apu_max_queued_frames,
-                              static_cast<uint32_t>(kMinimumQueuedFrames),
-                              static_cast<uint32_t>(kMaximumQueuedFrames));
 
   for (size_t i = 0; i < kMaximumClientCount; ++i) {
-    client_semaphores_[i] = xe::threading::Semaphore::Create(0, queued_frames_);
+    client_semaphores_[i] =
-    wait_handles_[i] = client_semaphores_[i].get();
+        xe::threading::Semaphore::Create(0, kMaximumQueuedFrames);
   }
-  shutdown_event_ = xe::threading::Event::CreateAutoResetEvent(false);
+  pending_work_event_ = xe::threading::Event::CreateAutoResetEvent(false);
-  assert_not_null(shutdown_event_);
+  assert_not_null(pending_work_event_);
-  wait_handles_[kMaximumClientCount] = shutdown_event_.get();
 
   xma_decoder_ = std::make_unique<xe::apu::XmaDecoder>(processor_);
 
@@ -93,7 +85,8 @@ X_STATUS AudioSystem::Setup(kernel::KernelState* kernel_state) {
   worker_thread_->set_can_debugger_suspend(true);
   worker_thread_->set_name("Audio Worker");
   worker_thread_->Create();
-
+  // Set high priority for this thread for better pacing.
+  worker_thread_->SetPriority(24);
   return X_STATUS_SUCCESS;
 }
 
@@ -101,56 +94,92 @@ void AudioSystem::WorkerThreadMain() {
   // Initialize driver and ringbuffer.
   Initialize();
 
-  // Main run loop.
+  // The host mixer releases a client's semaphore on its own coarse cadence,
+  // but Xenos audio subsystem operates at 5.333ms interval (see
+  // xaudio2_audio_driver.cc) Interval scales inversely with guest_time_scalar.
+  // We therefore pace pumps to each client's next_pump_us deadline and use
+  // the semaphore only as back-pressure: a frame is submitted only if
+  // a host output slot is free, otherwise it is dropped (the host queue
+  // is full, so it is already well buffered).
   while (worker_running_) {
-    // These handles signify the number of submitted samples. Once we reach
+    const uint64_t now = static_cast<uint64_t>(
-    // 64 samples, we wait until our audio backend releases a semaphore
+        std::chrono::duration_cast<std::chrono::microseconds>(
-    // (signaling a sample has finished playing)
+            std::chrono::steady_clock::now().time_since_epoch())
-    auto result =
+            .count());
-        xe::threading::WaitAny(wait_handles_, xe::countof(wait_handles_), true);
+
-    if (result.first == xe::threading::WaitResult::kFailed) {
+    size_t client_index = kMaximumClientCount;
-      // TODO: Assert?
+    uint64_t earliest_pump_us = std::numeric_limits<uint64_t>::max();
+    uint32_t client_callback = 0;
+    uint32_t client_callback_arg = 0;
+    {
+      auto global_lock = global_critical_region_.Acquire();
+
+      for (size_t i = 0; i < kMaximumClientCount; ++i) {
+        if (!clients_[i].in_use ||
+            clients_[i].next_pump_us >= earliest_pump_us) {
           continue;
         }
+        earliest_pump_us = clients_[i].next_pump_us;
+        client_index = i;
+      }
 
-    if (result.first == threading::WaitResult::kSuccess &&
+      if (client_index != kMaximumClientCount) {
-        result.second == kMaximumClientCount) {
+        client_callback = clients_[client_index].callback;
-      // Shutdown event signaled.
+        client_callback_arg = clients_[client_index].wrapped_callback_arg;
+
+        const double scalar = xe::Clock::guest_time_scalar();
+        const uint64_t min_us =
+            scalar > 0.0 ? static_cast<uint64_t>(kAudioPumpInterval / scalar)
+                         : kAudioPumpInterval;
+        clients_[client_index].next_pump_us =
+            (earliest_pump_us > now ? earliest_pump_us : now) + min_us;
+      }
+    }
+
+    // No clients yet: park until one registers or we're told to stop.
+    if (client_index == kMaximumClientCount) {
+      xe::threading::Wait(pending_work_event_.get(), true);
       if (paused_) {
         pause_fence_.Signal();
-        threading::Wait(resume_event_.get(), false);
+        xe::threading::Wait(resume_event_.get(), false);
       }
-
       continue;
     }
 
-    // Number of clients pumped
+    // Pace to kAudioIntervalSlack ahead of the deadline.
-    bool pumped = false;
+    const uint64_t wake_target_us = earliest_pump_us > kAudioIntervalSlack
-    if (result.first == xe::threading::WaitResult::kSuccess) {
+                                        ? earliest_pump_us - kAudioIntervalSlack
-      auto index = result.second;
+                                        : 0;
+    if (wake_target_us > now) {
+      const std::chrono::milliseconds timeout((wake_target_us - now) / 1000);
+      auto result =
+          xe::threading::Wait(pending_work_event_.get(), true, timeout);
+      if (result == xe::threading::WaitResult::kSuccess) {
+        if (paused_) {
+          pause_fence_.Signal();
+          xe::threading::Wait(resume_event_.get(), false);
+        }
+        continue;
+      }
 
-      auto global_lock = global_critical_region_.Acquire();
+      const uint64_t now_precise = static_cast<uint64_t>(
-      uint32_t client_callback = clients_[index].callback;
+          std::chrono::duration_cast<std::chrono::microseconds>(
-      uint32_t client_callback_arg = clients_[index].wrapped_callback_arg;
+              std::chrono::steady_clock::now().time_since_epoch())
-      global_lock.unlock();
+              .count());
+      if (wake_target_us > now_precise) {
+        xe::threading::NanoSleepPrecise((wake_target_us - now_precise) * 1000);
+      }
+    }
 
-      if (client_callback) {
+    // Submit only if the host has a free output slot;
+    if (client_callback &&
+        xe::threading::Wait(client_semaphores_[client_index].get(), false,
+                            std::chrono::milliseconds(0)) ==
+            xe::threading::WaitResult::kSuccess) {
       SCOPE_profile_cpu_i("apu", "xe::apu::AudioSystem->client_callback");
       uint64_t args[] = {client_callback_arg};
-        processor_->Execute(worker_thread_->thread_state(), client_callback,
+      processor_->Execute(worker_thread_->thread_state(), client_callback, args,
-                            args, xe::countof(args));
+                          xe::countof(args));
-      }
-
-      pumped = true;
-    }
-
-    if (!worker_running_) {
-      break;
-    }
-
-    if (!pumped) {
-      SCOPE_profile_cpu_i("apu", "Sleep");
-      xe::threading::Sleep(std::chrono::milliseconds(500));
     }
   }
   worker_running_ = false;
@@ -173,7 +202,7 @@ void AudioSystem::Initialize() {}
 
 void AudioSystem::Shutdown() {
   worker_running_ = false;
-  shutdown_event_->Set();
+  pending_work_event_->Set();
   if (worker_thread_) {
     worker_thread_->Wait(0, 0, 0, nullptr);
     worker_thread_.reset();
@@ -207,7 +236,7 @@ X_STATUS AudioSystem::RegisterClient(uint32_t callback, uint32_t callback_arg,
   assert_true(index >= 0);
 
   auto client_semaphore = client_semaphores_[index].get();
-  auto ret = client_semaphore->Release(queued_frames_, nullptr);
+  auto ret = client_semaphore->Release(kMaximumQueuedFrames, nullptr);
   assert_true(ret);
 
   AudioDriver* driver;
@@ -225,7 +254,16 @@ X_STATUS AudioSystem::RegisterClient(uint32_t callback, uint32_t callback_arg,
   uint32_t ptr = memory()->SystemHeapAlloc(0x4);
   xe::store_and_swap<uint32_t>(memory()->TranslateVirtual(ptr), callback_arg);
 
-  clients_[index] = {driver, callback, callback_arg, ptr, true};
+  clients_[index] = {};
+  clients_[index].driver = driver;
+  clients_[index].callback = callback;
+  clients_[index].callback_arg = callback_arg;
+  clients_[index].wrapped_callback_arg = ptr;
+  clients_[index].in_use = true;
+
+  // Wake the worker so it re-scans and starts pacing this client immediately.
+  pending_work_event_->Set();
+
   XELOGI("AudioSystem::RegisterClient: client {} registered successfully",
          index);
 
@@ -330,10 +368,11 @@ bool AudioSystem::Restore(ByteStream* stream) {
     client.callback_arg = stream->Read<uint32_t>();
     client.wrapped_callback_arg = stream->Read<uint32_t>();
 
+    client.next_pump_us = 0;
     client.in_use = true;
 
     auto client_semaphore = client_semaphores_[id].get();
-    auto ret = client_semaphore->Release(queued_frames_, nullptr);
+    auto ret = client_semaphore->Release(kMaximumQueuedFrames, nullptr);
     assert_true(ret);
 
     AudioDriver* driver = nullptr;
@@ -359,8 +398,7 @@ void AudioSystem::Pause() {
   }
   paused_ = true;
 
-  // Kind of a hack, but it works.
+  pending_work_event_->Set();
-  shutdown_event_->Set();
   pause_fence_.Wait();
 
   xma_decoder_->Pause();

src/xenia/apu/audio_system.h

@@ -11,7 +11,6 @@
 #define XENIA_APU_AUDIO_SYSTEM_H_
 
 #include <atomic>
-#include <queue>
 
 #include "xenia/base/mutex.h"
 #include "xenia/base/threading.h"
@@ -30,10 +29,9 @@ class XmaDecoder;
 
 class AudioSystem {
  public:
-  // TODO(gibbed): respect XAUDIO2_MAX_QUEUED_BUFFERS somehow (ie min(64,
-  // XAUDIO2_MAX_QUEUED_BUFFERS))
-  static constexpr size_t kMinimumQueuedFrames = 4;
   static constexpr size_t kMaximumQueuedFrames = 64;
+  static constexpr uint32_t kAudioPumpInterval = 5333u;
+  static constexpr uint32_t kAudioIntervalSlack = 400u;
 
   virtual ~AudioSystem();
 
@@ -78,7 +76,6 @@ class AudioSystem {
   Memory* memory_ = nullptr;
   cpu::Processor* processor_ = nullptr;
   std::unique_ptr<XmaDecoder> xma_decoder_;
-  uint32_t queued_frames_;
 
   std::atomic<bool> worker_running_ = {false};
   kernel::object_ref<kernel::XHostThread> worker_thread_;
@@ -87,6 +84,7 @@ class AudioSystem {
   static constexpr size_t kMaximumClientCount = 8;
   struct {
     AudioDriver* driver;
+    uint64_t next_pump_us;
     uint32_t callback;
     uint32_t callback_arg;
     uint32_t wrapped_callback_arg;
@@ -98,8 +96,7 @@ class AudioSystem {
   std::unique_ptr<xe::threading::Semaphore>
       client_semaphores_[kMaximumClientCount];
   // Event is always there in case we have no clients.
-  std::unique_ptr<xe::threading::Event> shutdown_event_;
+  std::unique_ptr<xe::threading::Event> pending_work_event_;
-  xe::threading::WaitHandle* wait_handles_[kMaximumClientCount + 1];
 
   bool paused_ = false;
   threading::Fence pause_fence_;

src/xenia/base/threading.h

@@ -116,6 +116,10 @@ void SyncMemory();
 // Sleeps the current thread for at least as long as the given duration.
 void Sleep(std::chrono::microseconds duration);
 void NanoSleep(int64_t ns);
+// Like NanoSleep but trades a brief busy-wait tail for sub-millisecond
+// precision. Use only where wake-up jitter would miss a frame budget; the
+// spin costs CPU.
+void NanoSleepPrecise(int64_t ns);
 template <typename Rep, typename Period>
 void Sleep(std::chrono::duration<Rep, Period> duration) {
   Sleep(std::chrono::duration_cast<std::chrono::microseconds>(duration));

src/xenia/base/threading_posix.cc

@@ -166,6 +166,36 @@ void Sleep(std::chrono::microseconds duration) {
 
 void NanoSleep(int64_t duration) { Sleep(std::chrono::nanoseconds(duration)); }
 
+void NanoSleepPrecise(int64_t ns) {
+#if XE_PLATFORM_MAC
+  // Darwin's nanosleep can oversleep by 100-500us under load. Land precisely
+  // on the deadline by using mach_wait_until for the bulk of the wait and
+  // busy-waiting the last ~200us.
+  if (ns <= 0) {
+    return;
+  }
+  static const mach_timebase_info_data_t tb = [] {
+    mach_timebase_info_data_t i;
+    mach_timebase_info(&i);
+    return i;
+  }();
+  constexpr uint64_t kSpinTailNs = 200'000;
+  const uint64_t deadline =
+      mach_absolute_time() +
+      static_cast<uint64_t>((static_cast<__uint128_t>(ns) * tb.denom) /
+                            tb.numer);
+  const uint64_t spin_tail = static_cast<uint64_t>(
+      (static_cast<__uint128_t>(kSpinTailNs) * tb.denom) / tb.numer);
+  if (deadline > mach_absolute_time() + spin_tail) {
+    mach_wait_until(deadline - spin_tail);
+  }
+  while (mach_absolute_time() < deadline) {
+  }
+#else
+  NanoSleep(ns);
+#endif
+}
+
 // TODO(bwrsandman) Implement by allowing alert interrupts from IO operations
 thread_local bool alertable_state_ = false;
 SleepResult AlertableSleep(std::chrono::microseconds duration) {

src/xenia/base/threading_win.cc

@@ -169,6 +169,8 @@ void Sleep(std::chrono::microseconds duration) {
   }
 }
 
+void NanoSleepPrecise(int64_t ns) { NanoSleep(ns); }
+
 SleepResult AlertableSleep(std::chrono::microseconds duration) {
   if (SleepEx(static_cast<DWORD>(duration.count() / 1000), true) ==
       WAIT_IO_COMPLETION) {