Xenia-Canary/src/xenia/apu/audio_system.cc

/**
 ******************************************************************************
 * Xenia : Xbox 360 Emulator Research Project                                 *
 ******************************************************************************
 * Copyright 2022 Ben Vanik. All rights reserved.                             *
 * Released under the BSD license - see LICENSE in the root for more details. *
 ******************************************************************************
 */

#include "xenia/apu/audio_system.h"

#include <limits>

#include "xenia/apu/apu_flags.h"
#include "xenia/apu/audio_driver.h"
#include "xenia/apu/audio_watchdog.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"
#include "xenia/base/ring_buffer.h"
#include "xenia/base/string_buffer.h"
#include "xenia/base/threading.h"
#include "xenia/cpu/thread_state.h"
#include "xenia/kernel/kernel_state.h"

// As with normal Microsoft, there are like twelve different ways to access
// the audio APIs. Early games use XMA*() methods almost exclusively to touch
// decoders. Later games use XAudio*() and direct memory writes to the XMA
// structures (as opposed to the XMA* calls), meaning that we have to support
// both.
//
// For ease of implementation, most audio related processing is handled in
// AudioSystem, and the functions here call off to it.
// The XMA*() functions just manipulate the audio system in the guest context
// and let the normal AudioSystem handling take it, to prevent duplicate
// implementations. They can be found in xboxkrnl_audio_xma.cc

namespace xe {
namespace apu {

AudioSystem::AudioSystem(cpu::Processor* processor)
    : memory_(processor->memory()),
      processor_(processor),
      worker_running_(false) {
  std::memset(clients_, 0, sizeof(clients_));

  for (size_t i = 0; i < kMaximumClientCount; ++i) {
    client_semaphores_[i] =
        xe::threading::Semaphore::Create(0, kMaximumQueuedFrames);
  }
  pending_work_event_ = xe::threading::Event::CreateAutoResetEvent(false);
  assert_not_null(pending_work_event_);

  xma_decoder_ = std::make_unique<xe::apu::XmaDecoder>(processor_);

  resume_event_ = xe::threading::Event::CreateAutoResetEvent(false);
  assert_not_null(resume_event_);
}

AudioSystem::~AudioSystem() {
  if (xma_decoder_) {
    xma_decoder_->Shutdown();
  }
}

namespace {

// --- audio watchdog (diagnostics, --audio_watchdog, default off) -------------
// Samples the pipeline-stage counters in audio_watchdog.h once a second and
// reports which stage stopped when audio dies. Deliberately near-silent while
// healthy: log spam of its own would starve the very pipeline it watches.

std::atomic<bool> watchdog_running_{false};
std::thread watchdog_thread_;

uint64_t WatchdogNowUs() {
  return static_cast<uint64_t>(
      std::chrono::duration_cast<std::chrono::microseconds>(
          std::chrono::steady_clock::now().time_since_epoch())
          .count());
}

const char* AlsaStateName(int state) {
  // snd_pcm_state_t, kept as an int so we need no ALSA header here.
  static const char* kNames[] = {"OPEN",     "SETUP",  "PREPARED",
                                 "RUNNING",  "XRUN",   "DRAINING",
                                 "PAUSED",   "SUSPENDED", "DISCONNECTED"};
  if (state < 0 || state >= static_cast<int>(xe::countof(kNames))) {
    return "?";
  }
  return kNames[state];
}

void AudioWatchdogMain() {
  xe::threading::set_name("Audio Watchdog");

  uint64_t last_pumps = 0, last_submits = 0, last_xma = 0;
  uint64_t last_writes = 0, last_silence = 0, last_xruns = 0;
  uint64_t last_loops = 0, last_skips = 0;
  bool was_alive = true;
  bool ever_alive = false;
  uint64_t dead_ticks = 0;

  while (watchdog_running_) {
    std::this_thread::sleep_for(std::chrono::seconds(1));
    if (!watchdog_running_) {
      break;
    }

    using namespace xe::apu::watchdog;
    const uint64_t pumps = guest_pumps.load(std::memory_order_relaxed);
    const uint64_t submits = frames_submitted.load(std::memory_order_relaxed);
    const uint64_t xma = xma_works.load(std::memory_order_relaxed);
    const uint64_t writes = alsa_writes.load(std::memory_order_relaxed);
    const uint64_t silence = alsa_silence.load(std::memory_order_relaxed);
    const uint64_t xruns = alsa_xruns.load(std::memory_order_relaxed);

    const uint64_t d_pumps = pumps - last_pumps;
    const uint64_t d_submits = submits - last_submits;
    const uint64_t d_xma = xma - last_xma;
    const uint64_t d_writes = writes - last_writes;
    const uint64_t d_silence = silence - last_silence;
    const uint64_t d_xruns = xruns - last_xruns;
    last_pumps = pumps;
    last_submits = submits;
    last_xma = xma;
    last_writes = writes;
    last_silence = silence;
    last_xruns = xruns;

    // How long has the APU worker been stuck inside guest code? The guest
    // callback runs on the pump thread, so a guest block here kills audio
    // permanently -- this is the prime suspect for "sound never comes back".
    const uint64_t in_guest_since =
        in_guest_callback_since_us.load(std::memory_order_relaxed);
    const double stuck_s =
        in_guest_since ? (WatchdogNowUs() - in_guest_since) / 1000000.0 : 0.0;

    // "Alive" = the guest is still producing audio. Prefer real frames reaching
    // the device, but fall back to pumps so this also works under --apu=nop
    // (silent stress runs, where there is no host device to write to).
    const bool alive = d_writes > 0 || d_pumps > 0;

    const uint64_t loops = worker_loops.load(std::memory_order_relaxed);
    const uint64_t skips = pump_skips.load(std::memory_order_relaxed);
    const int clients = clients_in_use.load(std::memory_order_relaxed);
    const uint64_t regs = register_calls.load(std::memory_order_relaxed);
    const uint64_t unregs = unregister_calls.load(std::memory_order_relaxed);
    const uint64_t d_loops = loops - last_loops;
    const uint64_t d_skips = skips - last_skips;
    last_loops = loops;
    last_skips = skips;

    const auto phase = worker_phase.load(std::memory_order_relaxed);

    // Name the culprit stage, walking the pipeline from the guest outward. The
    // worker's own recorded phase beats any inference we could make from the
    // outside, so trust it first.
    const char* culprit = "";
    if (!alive) {
      if (stuck_s >= 1.0) {
        culprit = "guest audio callback BLOCKED (pump thread stuck in guest)";
      } else if (d_loops == 0 &&
                 phase == WorkerPhase::kAcquiringGlobalLock) {
        culprit =
            "APU worker BLOCKED on the kernel global lock (someone else holds "
            "it)";
      } else if (d_loops == 0 && phase == WorkerPhase::kParkedNoClient) {
        culprit =
            "NO AUDIO CLIENT: guest unregistered it and never re-registered";
      } else if (d_pumps == 0 && d_loops == 0) {
        culprit = "worker stalled -- see phase=";
      } else if (d_pumps == 0 && d_skips > 0) {
        culprit = "worker looping but never gets a free output slot (semaphore)";
      } else if (d_pumps == 0) {
        culprit = "APU worker not pumping the guest callback";
      } else if (d_submits == 0) {
        culprit = "guest pumping but submitting no frames (starved upstream)";
      } else if (d_xma == 0 && xma > 0) {
        culprit = "XMA decoder stopped doing work";
      } else {
        culprit = "frames submitted but host driver is not writing them";
      }
    }

    if (!alive) {
      dead_ticks++;
    }
    if (alive) {
      ever_alive = true;
    }

    // Stay quiet until audio has actually played once: silence during boot /
    // menus is not the bug. Then log every transition and once a second while
    // dead -- "it was playing and then it stopped" is precisely the event.
    if (!ever_alive) {
      was_alive = alive;
      continue;
    }

    if (alive != was_alive || !alive) {
      XELOGW(
          "AUDIO-WD [{}] +pumps={} +submits={} +xma={} +writes={} "
          "+silence={} +xruns={} +loops={} +skips={} clients={} reg={} "
          "unreg={} phase={} pcm={} guest_cb_stuck={:.1f}s dead={}s {}",
          alive ? "OK" : "DEAD", d_pumps, d_submits, d_xma, d_writes, d_silence,
          d_xruns, d_loops, d_skips, clients, regs, unregs,
          WorkerPhaseName(phase),
          AlsaStateName(alsa_state.load(std::memory_order_relaxed)), stuck_s,
          dead_ticks, culprit);
    }
    if (alive) {
      dead_ticks = 0;
    }
    was_alive = alive;
  }
}

}  // namespace

X_STATUS AudioSystem::Setup(kernel::KernelState* kernel_state) {
  X_STATUS result = xma_decoder_->Setup(kernel_state);
  if (result) {
    return result;
  }

  if (cvars::audio_watchdog && !watchdog_running_) {
    watchdog_running_ = true;
    watchdog_thread_ = std::thread(AudioWatchdogMain);
    XELOGW("AUDIO-WD watchdog armed (reports the stage that stops on silence)");
  }

  worker_running_ = true;
  worker_thread_ =
      kernel::object_ref<kernel::XHostThread>(new kernel::XHostThread(
          kernel_state, 128 * 1024, 0,
          [this]() {
            WorkerThreadMain();
            return 0;
          },
          kernel_state->GetSystemProcess()));
  // As we run audio callbacks the debugger must be able to suspend us.
  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;
}

void AudioSystem::WorkerThreadMain() {
  // Initialize driver and ringbuffer.
  Initialize();

  // 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_) {
    watchdog::worker_loops.fetch_add(1, std::memory_order_relaxed);

    const uint64_t now = static_cast<uint64_t>(
        std::chrono::duration_cast<std::chrono::microseconds>(
            std::chrono::steady_clock::now().time_since_epoch())
            .count());

    size_t client_index = kMaximumClientCount;
    uint64_t earliest_pump_us = std::numeric_limits<uint64_t>::max();
    uint32_t client_callback = 0;
    uint32_t client_callback_arg = 0;
    {
      watchdog::worker_phase.store(watchdog::WorkerPhase::kAcquiringGlobalLock,
                                   std::memory_order_relaxed);
      auto global_lock = global_critical_region_.Acquire();
      watchdog::worker_phase.store(watchdog::WorkerPhase::kStart,
                                   std::memory_order_relaxed);

      int in_use_count = 0;
      for (size_t i = 0; i < kMaximumClientCount; ++i) {
        if (clients_[i].in_use) {
          ++in_use_count;
        }
        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;
      }
      watchdog::clients_in_use.store(in_use_count, std::memory_order_relaxed);

      if (client_index != kMaximumClientCount) {
        client_callback = clients_[client_index].callback;
        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) {
      watchdog::worker_phase.store(watchdog::WorkerPhase::kParkedNoClient,
                                   std::memory_order_relaxed);
      xe::threading::Wait(pending_work_event_.get(), true);
      if (paused_) {
        pause_fence_.Signal();
        xe::threading::Wait(resume_event_.get(), false);
      }
      continue;
    }

    // Pace to kAudioIntervalSlack ahead of the deadline.
    const uint64_t wake_target_us = earliest_pump_us > kAudioIntervalSlack
                                        ? earliest_pump_us - kAudioIntervalSlack
                                        : 0;
    if (wake_target_us > now) {
      watchdog::worker_phase.store(watchdog::WorkerPhase::kPacingSleep,
                                   std::memory_order_relaxed);
      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;
      }

      const uint64_t now_precise = static_cast<uint64_t>(
          std::chrono::duration_cast<std::chrono::microseconds>(
              std::chrono::steady_clock::now().time_since_epoch())
              .count());
      if (wake_target_us > now_precise) {
        xe::threading::NanoSleepPrecise((wake_target_us - now_precise) * 1000);
      }
    }

    // Submit only if the host has a free output slot;
    const bool have_slot =
        client_callback &&
        xe::threading::Wait(client_semaphores_[client_index].get(), false,
                            std::chrono::milliseconds(0)) ==
            xe::threading::WaitResult::kSuccess;
    if (!have_slot) {
      watchdog::pump_skips.fetch_add(1, std::memory_order_relaxed);
    }
    if (have_slot) {
      SCOPE_profile_cpu_i("apu", "xe::apu::AudioSystem->client_callback");
      uint64_t args[] = {client_callback_arg};
      // The guest callback runs in-line on this pump thread: if it blocks, the
      // pump stops forever and audio never recovers. Bracket it so the
      // watchdog can see (and name) that case.
      watchdog::in_guest_callback_since_us.store(WatchdogNowUs(),
                                                 std::memory_order_relaxed);
      watchdog::worker_phase.store(watchdog::WorkerPhase::kInGuestCallback,
                                   std::memory_order_relaxed);
      processor_->Execute(worker_thread_->thread_state(), client_callback, args,
                          xe::countof(args));
      watchdog::worker_phase.store(watchdog::WorkerPhase::kStart,
                                   std::memory_order_relaxed);
      watchdog::in_guest_callback_since_us.store(0, std::memory_order_relaxed);
      watchdog::guest_pumps.fetch_add(1, std::memory_order_relaxed);
    }
  }
  worker_running_ = false;

  // TODO(benvanik): call module API to kill?
}

int AudioSystem::FindFreeClient() {
  for (int i = 0; i < kMaximumClientCount; i++) {
    auto& client = clients_[i];
    if (!client.in_use) {
      return i;
    }
  }

  return -1;
}

void AudioSystem::Initialize() {}

void AudioSystem::Shutdown() {
  if (watchdog_running_) {
    watchdog_running_ = false;
    if (watchdog_thread_.joinable()) {
      watchdog_thread_.join();
    }
  }

  worker_running_ = false;
  pending_work_event_->Set();
  if (worker_thread_) {
    worker_thread_->Wait(0, 0, 0, nullptr);
    worker_thread_.reset();
  }

  // Unregister all active clients to shut down their audio drivers before
  // the semaphores are destroyed with this AudioSystem.
  {
    auto global_lock = global_critical_region_.Acquire();
    for (size_t i = 0; i < kMaximumClientCount; ++i) {
      if (clients_[i].in_use) {
        DestroyDriver(clients_[i].driver);
        if (clients_[i].wrapped_callback_arg) {
          memory()->SystemHeapFree(clients_[i].wrapped_callback_arg);
        }
        clients_[i].driver = nullptr;
        clients_[i].callback = 0;
        clients_[i].callback_arg = 0;
        clients_[i].wrapped_callback_arg = 0;
        clients_[i].in_use = false;
      }
    }
  }
}

X_STATUS AudioSystem::RegisterClient(uint32_t callback, uint32_t callback_arg,
                                     size_t* out_index) {
  auto global_lock = global_critical_region_.Acquire();

  auto index = FindFreeClient();
  assert_true(index >= 0);

  auto client_semaphore = client_semaphores_[index].get();
  auto ret = client_semaphore->Release(kMaximumQueuedFrames, nullptr);
  assert_true(ret);

  AudioDriver* driver;
  auto result = CreateDriver(index, client_semaphore, &driver);
  if (XFAILED(result)) {
    XELOGE("AudioSystem::RegisterClient: CreateDriver failed for index={}",
           index);
    return result;
  }
  assert_not_null(driver);
  XELOGI(
      "AudioSystem::RegisterClient: driver created for index={}, driver={:p}",
      index, (void*)driver);

  uint32_t ptr = memory()->SystemHeapAlloc(0x4);
  xe::store_and_swap<uint32_t>(memory()->TranslateVirtual(ptr), callback_arg);

  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;

  watchdog::register_calls.fetch_add(1, std::memory_order_relaxed);
  if (cvars::audio_watchdog) {
    XELOGW("AUDIO-WD RegisterClient(index={}) callback={:08X} -- audio client back",
           index, callback);
  }

  // Wake the worker so it re-scans and starts pacing this client immediately.
  pending_work_event_->Set();

  XELOGI("AudioSystem::RegisterClient: client {} registered successfully",
         index);

  if (out_index) {
    *out_index = index;
  }

  return X_STATUS_SUCCESS;
}

void AudioSystem::SubmitFrame(size_t index, float* samples) {
  SCOPE_profile_cpu_f("apu");

  watchdog::frames_submitted.fetch_add(1, std::memory_order_relaxed);

  auto global_lock = global_critical_region_.Acquire();
  assert_true(index < kMaximumClientCount);
  if (index >= kMaximumClientCount || !clients_[index].in_use ||
      !clients_[index].driver) {
    XELOGW(
        "SubmitFrame called for invalid/unregistered client index {} "
        "(in_use={}, driver={:p})",
        index, index < kMaximumClientCount ? clients_[index].in_use : false,
        index < kMaximumClientCount ? (void*)clients_[index].driver : nullptr);

    // Submit silence instead of dropping the frame to maintain the callback
    // chain.  If we don't submit anything, the audio driver's OnBufferEnd
    // callback will never fire, causing the semaphore to leak.
    if (index < kMaximumClientCount && clients_[index].driver) {
      static float silence[apu::AudioDriver::kFrameSamplesMax] = {0};
      (clients_[index].driver)->SubmitFrame(silence);
    }
    return;
  }
  (clients_[index].driver)->SubmitFrame(samples);
}

void AudioSystem::UnregisterClient(size_t index) {
  SCOPE_profile_cpu_f("apu");

  watchdog::unregister_calls.fetch_add(1, std::memory_order_relaxed);
  if (cvars::audio_watchdog) {
    XELOGW("AUDIO-WD UnregisterClient(index={}) -- guest dropped its audio client",
           index);
  }

  auto global_lock = global_critical_region_.Acquire();
  assert_true(index < kMaximumClientCount);
  DestroyDriver(clients_[index].driver);
  memory()->SystemHeapFree(clients_[index].wrapped_callback_arg);
  clients_[index] = {0};

  // Drain the semaphore of its count.
  auto client_semaphore = client_semaphores_[index].get();
  xe::threading::WaitResult wait_result;
  do {
    wait_result = xe::threading::Wait(client_semaphore, false,
                                      std::chrono::milliseconds(0));
  } while (wait_result == xe::threading::WaitResult::kSuccess);
  assert_true(wait_result == xe::threading::WaitResult::kTimeout);
}

bool AudioSystem::Save(ByteStream* stream) {
  stream->Write(kAudioSaveSignature);

  // Count the number of used clients first.
  // Any gaps should be handled gracefully.
  uint32_t used_clients = 0;
  for (int i = 0; i < kMaximumClientCount; i++) {
    if (clients_[i].in_use) {
      used_clients++;
    }
  }

  stream->Write(used_clients);
  for (uint32_t i = 0; i < kMaximumClientCount; i++) {
    auto& client = clients_[i];
    if (!client.in_use) {
      continue;
    }

    stream->Write(i);
    stream->Write(client.callback);
    stream->Write(client.callback_arg);
    stream->Write(client.wrapped_callback_arg);
  }

  return true;
}

bool AudioSystem::Restore(ByteStream* stream) {
  if (stream->Read<uint32_t>() != kAudioSaveSignature) {
    XELOGE("AudioSystem::Restore - Invalid magic value!");
    return false;
  }

  uint32_t num_clients = stream->Read<uint32_t>();
  for (uint32_t i = 0; i < num_clients; i++) {
    auto id = stream->Read<uint32_t>();
    assert_true(id < kMaximumClientCount);

    auto& client = clients_[id];

    // Reset the semaphore and recreate the driver ourselves.
    if (client.driver) {
      UnregisterClient(id);
    }

    client.callback = stream->Read<uint32_t>();
    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(kMaximumQueuedFrames, nullptr);
    assert_true(ret);

    AudioDriver* driver = nullptr;
    auto status = CreateDriver(id, client_semaphore, &driver);
    if (XFAILED(status)) {
      XELOGE(
          "AudioSystem::Restore - Call to CreateDriver failed with status "
          "{:08X}",
          status);
      return false;
    }

    assert_not_null(driver);
    client.driver = driver;
  }

  return true;
}

void AudioSystem::Pause() {
  if (paused_) {
    return;
  }
  paused_ = true;

  pending_work_event_->Set();
  pause_fence_.Wait();

  xma_decoder_->Pause();
}

void AudioSystem::Resume() {
  if (!paused_) {
    return;
  }
  paused_ = false;

  resume_event_->Set();

  xma_decoder_->Resume();
}

}  // namespace apu
}  // namespace xe