InstanceBrick--nightcap/src/audio/opensl_simple_buffer_queue.cc

#include "mocktail/audio/opensl_simple_buffer_queue.h"

#include <pthread.h>

#include <algorithm>
#include <atomic>
#include <condition_variable>
#include <cstddef>
#include <cstring>
#include <deque>
#include <limits>
#include <mutex>
#include <new>
#include <string>
#include <utility>

namespace mocktail::audio {
namespace {

using opensl_abi::AndroidSimpleBufferQueue;
using opensl_abi::AndroidSimpleBufferQueueCallback;
using opensl_abi::AndroidSimpleBufferQueueState;
using opensl_abi::AndroidSimpleBufferQueueTable;
using opensl_abi::Result;
using opensl_abi::Uint32;

Result ToOpenSlResult(const Status& status) {
  switch (status.code()) {
    case StatusCode::kOk:
      return opensl_abi::kResultSuccess;
    case StatusCode::kInvalidArgument:
      return opensl_abi::kResultParameterInvalid;
    case StatusCode::kFailedPrecondition:
      return opensl_abi::kResultPreconditionsViolated;
    case StatusCode::kUnavailable:
      return opensl_abi::kResultResourceError;
    case StatusCode::kUnsupported:
      return opensl_abi::kResultFeatureUnsupported;
    case StatusCode::kPlatformError:
      return opensl_abi::kResultIoError;
  }
  return opensl_abi::kResultInternalError;
}

}  // namespace

struct OpenSlSimpleBufferQueueAdapter::State
    : public std::enable_shared_from_this<
          OpenSlSimpleBufferQueueAdapter::State> {
  struct Handle {
    const AndroidSimpleBufferQueueTable* table = nullptr;
    State* state = nullptr;
  };

  struct ReleaseTicket {
    const std::shared_ptr<State> state;
    const std::uint64_t generation;
  };

  struct CallbackJob {
    AndroidSimpleBufferQueueCallback callback = nullptr;
    void* context = nullptr;
  };

  explicit State(std::unique_ptr<AudioSink> owned_sink,
                 const OpenSlSimpleBufferQueueOptions& options)
      : sink(std::move(owned_sink)),
        max_buffers(options.max_buffers),
        event_callback(options.event_callback),
        event_context(options.event_context) {}

  static const AndroidSimpleBufferQueueTable kQueueTable;

  AndroidSimpleBufferQueue Interface() {
    return reinterpret_cast<AndroidSimpleBufferQueue>(&handle.table);
  }

  static State* FromInterface(AndroidSimpleBufferQueue self) {
    if (self == nullptr) {
      return nullptr;
    }
    auto* handle = reinterpret_cast<Handle*>(
        const_cast<AndroidSimpleBufferQueueTable**>(self));
    return handle->state;
  }

  static Result MOCKTAIL_OPENSL_API_ENTRY
  AbiEnqueue(AndroidSimpleBufferQueue self, const void* buffer, Uint32 size) {
    State* raw_state = FromInterface(self);
    if (raw_state == nullptr || buffer == nullptr || size == 0) {
      return opensl_abi::kResultParameterInvalid;
    }
    const std::shared_ptr<State> state = raw_state->shared_from_this();

    std::lock_guard<std::mutex> operation_lock(state->sink_operation_mutex);
    ReleaseTicket* ticket = nullptr;
    {
      std::lock_guard<std::mutex> lock(state->mutex);
      if (state->stopping) {
        return opensl_abi::kResultPreconditionsViolated;
      }
      if (state->pending_count.load(std::memory_order_relaxed) >=
          state->max_buffers) {
        return opensl_abi::kResultBufferInsufficient;
      }
      ticket = new (std::nothrow)
          ReleaseTicket{state, state->generation};
      if (ticket == nullptr) {
        state->last_status = Status::Error(
            StatusCode::kUnavailable, "unable to allocate OpenSL release ticket");
        return ToOpenSlResult(state->last_status);
      }
      state->pending_count.fetch_add(1, std::memory_order_relaxed);
      ++state->submitted_buffers;
    }

    const PcmBuffer pcm{buffer, static_cast<std::size_t>(size),
                        &State::ReleaseBuffer, ticket};
    // A successful enqueue may release synchronously. Only a failed enqueue
    // leaves ownership here; its contract forbids a release callback.
    const Status status = state->sink->Enqueue(pcm);
    if (status.ok()) {
      state->NotifyEvent(OpenSlBufferQueueEvent::kSubmitted,
                         static_cast<std::size_t>(size));
      return opensl_abi::kResultSuccess;
    }

    const std::unique_ptr<ReleaseTicket> failed_ticket(ticket);
    {
      std::lock_guard<std::mutex> lock(state->mutex);
      state->last_status = status;
      state->pending_count.fetch_sub(1, std::memory_order_relaxed);
      --state->submitted_buffers;
    }
    return ToOpenSlResult(status);
  }

  static Result MOCKTAIL_OPENSL_API_ENTRY
  AbiClear(AndroidSimpleBufferQueue self) {
    State* raw_state = FromInterface(self);
    if (raw_state == nullptr) {
      return opensl_abi::kResultParameterInvalid;
    }
    const std::shared_ptr<State> state = raw_state->shared_from_this();
    std::lock_guard<std::mutex> operation_lock(state->sink_operation_mutex);
    {
      std::lock_guard<std::mutex> lock(state->mutex);
      if (state->stopping) {
        return opensl_abi::kResultPreconditionsViolated;
      }
      ++state->generation;
      state->discarded_buffers +=
          state->pending_count.load(std::memory_order_relaxed);
      state->pending_count.store(0, std::memory_order_relaxed);
    }
    const Status status = state->sink->Clear();
    if (!status.ok()) {
      std::lock_guard<std::mutex> lock(state->mutex);
      state->last_status = status;
    }
    return ToOpenSlResult(status);
  }

  static Result MOCKTAIL_OPENSL_API_ENTRY AbiGetState(
      AndroidSimpleBufferQueue self, AndroidSimpleBufferQueueState* output) {
    State* state = FromInterface(self);
    if (state == nullptr || output == nullptr) {
      return opensl_abi::kResultParameterInvalid;
    }
    output->count = static_cast<Uint32>(std::min<std::size_t>(
        state->pending_count.load(std::memory_order_relaxed),
        std::numeric_limits<Uint32>::max()));
    output->index = static_cast<Uint32>(std::min<std::uint64_t>(
        state->processed_buffers.load(std::memory_order_relaxed),
        std::numeric_limits<Uint32>::max()));
    return opensl_abi::kResultSuccess;
  }

  static Result MOCKTAIL_OPENSL_API_ENTRY AbiRegisterCallback(
      AndroidSimpleBufferQueue self,
      AndroidSimpleBufferQueueCallback registered_callback, void* context) {
    State* state = FromInterface(self);
    if (state == nullptr) {
      return opensl_abi::kResultParameterInvalid;
    }
    std::lock_guard<std::mutex> lock(state->mutex);
    if (state->stopping) {
      return opensl_abi::kResultPreconditionsViolated;
    }
    state->callback = registered_callback;
    state->callback_context = context;
    return opensl_abi::kResultSuccess;
  }

  static void ReleaseBuffer(void* context, const void* /*data*/,
                            std::size_t size_bytes) {
    const std::unique_ptr<ReleaseTicket> ticket(
        static_cast<ReleaseTicket*>(context));
    if (ticket == nullptr) {
      return;
    }
    const std::shared_ptr<State> state = ticket->state;
    if (state == nullptr) {
      return;
    }
    bool notify_worker = false;
    bool consumed = false;
    bool discarded = false;
    {
      std::lock_guard<std::mutex> lock(state->mutex);
      // Clear already counted older generations as discarded. Their tickets
      // remain alive until release without affecting newly enqueued buffers.
      if (ticket->generation != state->generation) {
        return;
      }
      if (state->pending_count.load(std::memory_order_relaxed) > 0) {
        state->pending_count.fetch_sub(1, std::memory_order_relaxed);
      }
      if (!state->stopping) {
        state->processed_buffers.fetch_add(1, std::memory_order_relaxed);
        ++state->consumed_buffers;
        state->consumed_bytes += size_bytes;
        consumed = true;
        if (state->callback != nullptr) {
          state->callback_jobs.push_back(
              CallbackJob{state->callback, state->callback_context});
          notify_worker = true;
        }
      } else {
        ++state->discarded_buffers;
        discarded = true;
      }
    }
    if (consumed) {
      state->NotifyEvent(OpenSlBufferQueueEvent::kConsumed, size_bytes);
    } else if (discarded) {
      state->NotifyEvent(OpenSlBufferQueueEvent::kDiscarded, size_bytes);
    }
    if (notify_worker) {
      state->callback_cv.notify_one();
    }
  }

  void NotifyEvent(OpenSlBufferQueueEvent event, std::size_t size_bytes) {
    if (event_callback != nullptr) {
      event_callback(event_context, event, size_bytes);
    }
  }

  void RunCallbacks() {
    for (;;) {
      CallbackJob job;
      {
        std::unique_lock<std::mutex> lock(mutex);
        callback_cv.wait(lock,
                         [this] { return stopping || !callback_jobs.empty(); });
        if (stopping) {
          callback_jobs.clear();
          return;
        }
        job = callback_jobs.front();
        callback_jobs.pop_front();
      }
      if (job.callback != nullptr) {
        job.callback(Interface(), job.context);
      }
    }
  }

  static void* CallbackThreadMain(void* context) {
    auto* state_holder = static_cast<std::shared_ptr<State>*>(context);
    const std::shared_ptr<State> state = std::move(*state_holder);
    delete state_holder;
    state->RunCallbacks();
    return nullptr;
  }

  Status StartCallbackThread() {
    auto* state_holder =
        new (std::nothrow) std::shared_ptr<State>(shared_from_this());
    if (state_holder == nullptr) {
      return Status::Error(StatusCode::kUnavailable,
                           "unable to allocate OpenSL callback thread state");
    }
    pthread_attr_t attr;
    pthread_attr_init(&attr);
    pthread_attr_setstacksize(&attr, 256 * 1024);
    const int result = pthread_create(&callback_thread, &attr,
                                      &State::CallbackThreadMain, state_holder);
    pthread_attr_destroy(&attr);
    if (result != 0) {
      delete state_holder;
      return Status::Error(
          StatusCode::kPlatformError,
          std::string("unable to create OpenSL callback thread: ") +
              std::strerror(result));
    }
    callback_thread_started = true;
    return Status::Ok();
  }

  bool IsCallbackThreadLocked() const {
    return callback_thread_started &&
           pthread_equal(callback_thread, pthread_self()) != 0;
  }

  void JoinCallbackThread() {
    pthread_t thread{};
    bool is_current_thread = false;
    {
      std::lock_guard<std::mutex> lock(mutex);
      if (!callback_thread_started) {
        return;
      }
      thread = callback_thread;
      is_current_thread = pthread_equal(thread, pthread_self()) != 0;
    }
    const int result = is_current_thread ? pthread_detach(thread)
                                         : pthread_join(thread, nullptr);
    {
      std::lock_guard<std::mutex> lock(mutex);
      callback_thread_started = false;
      if (result != 0) {
        last_status = Status::Error(
            StatusCode::kPlatformError,
            std::string(is_current_thread
                            ? "unable to detach OpenSL callback thread: "
                            : "unable to join OpenSL callback thread: ") +
                std::strerror(result));
      }
    }
  }

  mutable std::mutex mutex;
  std::condition_variable callback_cv;
  std::condition_variable shutdown_cv;
  std::mutex sink_operation_mutex;
  std::unique_ptr<AudioSink> sink;
  const std::uint32_t max_buffers;
  const OpenSlBufferQueueEventCallback event_callback;
  void* const event_context;
  Handle handle;
  pthread_t callback_thread{};
  std::atomic<std::size_t> pending_count{0};
  std::deque<CallbackJob> callback_jobs;
  AndroidSimpleBufferQueueCallback callback = nullptr;
  void* callback_context = nullptr;
  Status last_status;
  std::uint64_t generation = 1;
  std::atomic<std::uint64_t> processed_buffers{0};
  std::uint64_t submitted_buffers = 0;
  std::uint64_t consumed_buffers = 0;
  std::uint64_t discarded_buffers = 0;
  std::uint64_t consumed_bytes = 0;
  bool stopping = false;
  bool callback_thread_started = false;
  bool shutdown_started = false;
  bool shutdown_complete = false;
};

const AndroidSimpleBufferQueueTable
    OpenSlSimpleBufferQueueAdapter::State::kQueueTable = {
        &OpenSlSimpleBufferQueueAdapter::State::AbiEnqueue,
        &OpenSlSimpleBufferQueueAdapter::State::AbiClear,
        &OpenSlSimpleBufferQueueAdapter::State::AbiGetState,
        &OpenSlSimpleBufferQueueAdapter::State::AbiRegisterCallback,
};

OpenSlSimpleBufferQueueAdapter::OpenSlSimpleBufferQueueAdapter(
    std::shared_ptr<State> state)
    : state_(std::move(state)) {}

OpenSlSimpleBufferQueueAdapter::~OpenSlSimpleBufferQueueAdapter() {
  Shutdown();
}

Status OpenSlSimpleBufferQueueAdapter::Create(
    std::unique_ptr<AudioSink> sink,
    const OpenSlSimpleBufferQueueOptions& options,
    std::unique_ptr<OpenSlSimpleBufferQueueAdapter>* adapter) {
  if (adapter == nullptr) {
    return Status::Error(StatusCode::kInvalidArgument,
                         "OpenSL adapter output pointer is null");
  }
  adapter->reset();
  if (sink == nullptr) {
    return Status::Error(StatusCode::kInvalidArgument,
                         "OpenSL adapter requires an audio sink");
  }
  if (options.max_buffers == 0) {
    return Status::Error(StatusCode::kInvalidArgument,
                         "OpenSL queue capacity must be non-zero");
  }

  auto state = std::shared_ptr<State>(
      new (std::nothrow) State(std::move(sink), options));
  if (state == nullptr) {
    return Status::Error(StatusCode::kUnavailable,
                         "unable to allocate OpenSL queue state");
  }
  state->handle.table = &State::kQueueTable;
  state->handle.state = state.get();
  Status status = state->StartCallbackThread();
  if (!status.ok()) {
    state->sink->Shutdown();
    return status;
  }

  auto* implementation =
      new (std::nothrow) OpenSlSimpleBufferQueueAdapter(state);
  if (implementation == nullptr) {
    {
      std::lock_guard<std::mutex> lock(state->mutex);
      state->stopping = true;
    }
    state->callback_cv.notify_all();
    state->JoinCallbackThread();
    state->sink->Shutdown();
    return Status::Error(StatusCode::kUnavailable,
                         "unable to allocate OpenSL queue adapter");
  }
  adapter->reset(implementation);
  return Status::Ok();
}

AndroidSimpleBufferQueue OpenSlSimpleBufferQueueAdapter::interface() const {
  return state_ != nullptr ? state_->Interface() : nullptr;
}

Status OpenSlSimpleBufferQueueAdapter::SetPlaying(bool playing) {
  const std::shared_ptr<State> state = state_;
  if (state == nullptr) {
    return Status::Error(StatusCode::kFailedPrecondition,
                         "OpenSL adapter has no state");
  }
  std::lock_guard<std::mutex> operation_lock(state->sink_operation_mutex);
  {
    std::lock_guard<std::mutex> lock(state->mutex);
    if (state->stopping) {
      return Status::Error(StatusCode::kFailedPrecondition,
                           "OpenSL adapter is shut down");
    }
  }
  const Status status = playing ? state->sink->Resume() : state->sink->Pause();
  if (!status.ok()) {
    std::lock_guard<std::mutex> lock(state->mutex);
    state->last_status = status;
  }
  return status;
}

Status OpenSlSimpleBufferQueueAdapter::SetGain(float linear_gain) {
  const std::shared_ptr<State> state = state_;
  if (state == nullptr) {
    return Status::Error(StatusCode::kFailedPrecondition,
                         "OpenSL adapter has no state");
  }
  std::lock_guard<std::mutex> operation_lock(state->sink_operation_mutex);
  {
    std::lock_guard<std::mutex> lock(state->mutex);
    if (state->stopping) {
      return Status::Error(StatusCode::kFailedPrecondition,
                           "OpenSL adapter is shut down");
    }
  }
  const Status status = state->sink->SetGain(linear_gain);
  if (!status.ok()) {
    std::lock_guard<std::mutex> lock(state->mutex);
    state->last_status = status;
  }
  return status;
}

OpenSlSimpleBufferQueueStats OpenSlSimpleBufferQueueAdapter::GetStats() const {
  const std::shared_ptr<State> state = state_;
  if (state == nullptr) {
    return {};
  }
  std::lock_guard<std::mutex> lock(state->mutex);
  return OpenSlSimpleBufferQueueStats{
      state->submitted_buffers, state->consumed_buffers,
      state->discarded_buffers, state->consumed_bytes,
      state->pending_count.load(std::memory_order_relaxed)};
}

Status OpenSlSimpleBufferQueueAdapter::last_error() const {
  const std::shared_ptr<State> state = state_;
  if (state == nullptr) {
    return Status::Error(StatusCode::kFailedPrecondition,
                         "OpenSL adapter has no state");
  }
  std::lock_guard<std::mutex> lock(state->mutex);
  return state->last_status;
}

void OpenSlSimpleBufferQueueAdapter::Shutdown() {
  const std::shared_ptr<State> state = state_;
  if (state == nullptr) {
    return;
  }

  {
    std::unique_lock<std::mutex> lock(state->mutex);
    if (state->shutdown_started) {
      if (state->IsCallbackThreadLocked()) {
        return;
      }
      state->shutdown_cv.wait(lock,
                              [&state] { return state->shutdown_complete; });
      return;
    }
    state->shutdown_started = true;
    state->stopping = true;
    state->callback = nullptr;
    state->callback_context = nullptr;
    state->callback_jobs.clear();
  }
  state->callback_cv.notify_all();

  {
    std::lock_guard<std::mutex> operation_lock(state->sink_operation_mutex);
    state->sink->Shutdown();
  }
  state->JoinCallbackThread();

  {
    std::lock_guard<std::mutex> lock(state->mutex);
    if (state->pending_count.load(std::memory_order_relaxed) > 0) {
      state->last_status = Status::Error(
          StatusCode::kPlatformError,
          "audio sink did not release all borrowed OpenSL buffers");
      state->pending_count.store(0, std::memory_order_relaxed);
    }
    state->shutdown_complete = true;
  }
  state->shutdown_cv.notify_all();
}

Status OpenSlRecorderSupportStatus() {
  return Status::Error(StatusCode::kUnsupported,
                       "OpenSL audio recording is not implemented");
}

}  // namespace mocktail::audio