serenity/Userland/Libraries/LibAudio/PulseAudioWrappers.cpp

/*
 * Copyright (c) 2023, Gregory Bertilson <zaggy1024@gmail.com>
 *
 * SPDX-License-Identifier: BSD-2-Clause
 */

#include "PulseAudioWrappers.h"

#include <AK/WeakPtr.h>
#include <LibThreading/Mutex.h>

namespace Audio {

WeakPtr<PulseAudioContext> PulseAudioContext::weak_instance()
{
    // Use a weak pointer to allow the context to be shut down if we stop outputting audio.
    static WeakPtr<PulseAudioContext> the_instance;
    return the_instance;
}

ErrorOr<NonnullRefPtr<PulseAudioContext>> PulseAudioContext::instance()
{
    static Threading::Mutex instantiation_mutex;
    // Lock and unlock the mutex to ensure that the mutex is fully unlocked at application
    // exit.
    atexit([]() {
        instantiation_mutex.lock();
        instantiation_mutex.unlock();
    });

    auto instantiation_locker = Threading::MutexLocker(instantiation_mutex);

    auto the_instance = weak_instance();
    RefPtr<PulseAudioContext> strong_instance_pointer = the_instance.strong_ref();

    if (strong_instance_pointer == nullptr) {
        auto* main_loop = pa_threaded_mainloop_new();
        if (main_loop == nullptr)
            return Error::from_string_literal("Failed to create PulseAudio main loop");

        auto* api = pa_threaded_mainloop_get_api(main_loop);
        if (api == nullptr)
            return Error::from_string_literal("Failed to get PulseAudio API");

        auto* context = pa_context_new(api, "Ladybird");
        if (context == nullptr)
            return Error::from_string_literal("Failed to get PulseAudio connection context");

        strong_instance_pointer = make_ref_counted<PulseAudioContext>(main_loop, api, context);

        // Set a callback to signal ourselves to wake when the state changes, so that we can
        // synchronously wait for the connection.
        pa_context_set_state_callback(
            context, [](pa_context*, void* user_data) {
                static_cast<PulseAudioContext*>(user_data)->signal_to_wake();
            },
            strong_instance_pointer.ptr());

        if (auto error = pa_context_connect(context, nullptr, PA_CONTEXT_NOFLAGS, nullptr); error < 0) {
            warnln("Starting PulseAudio context connection failed with error: {}", pulse_audio_error_to_string(static_cast<PulseAudioErrorCode>(-error)));
            return Error::from_string_literal("Error while starting PulseAudio daemon connection");
        }

        if (auto error = pa_threaded_mainloop_start(main_loop); error < 0) {
            warnln("Starting PulseAudio main loop failed with error: {}", pulse_audio_error_to_string(static_cast<PulseAudioErrorCode>(-error)));
            return Error::from_string_literal("Failed to start PulseAudio main loop");
        }

        {
            auto locker = strong_instance_pointer->main_loop_locker();
            while (true) {
                bool is_ready = false;
                switch (strong_instance_pointer->get_connection_state()) {
                case PulseAudioContextState::Connecting:
                case PulseAudioContextState::Authorizing:
                case PulseAudioContextState::SettingName:
                    break;
                case PulseAudioContextState::Ready:
                    is_ready = true;
                    break;
                case PulseAudioContextState::Failed:
                    warnln("PulseAudio server connection failed with error: {}", pulse_audio_error_to_string(strong_instance_pointer->get_last_error()));
                    return Error::from_string_literal("Failed to connect to PulseAudio server");
                case PulseAudioContextState::Unconnected:
                case PulseAudioContextState::Terminated:
                    VERIFY_NOT_REACHED();
                    break;
                }

                if (is_ready)
                    break;

                strong_instance_pointer->wait_for_signal();
            }

            pa_context_set_state_callback(context, nullptr, nullptr);
        }

        the_instance = strong_instance_pointer;
    }

    return strong_instance_pointer.release_nonnull();
}

PulseAudioContext::PulseAudioContext(pa_threaded_mainloop* main_loop, pa_mainloop_api* api, pa_context* context)
    : m_main_loop(main_loop)
    , m_api(api)
    , m_context(context)
{
}

PulseAudioContext::~PulseAudioContext()
{
    {
        auto locker = main_loop_locker();
        pa_context_disconnect(m_context);
        pa_context_unref(m_context);
    }
    pa_threaded_mainloop_stop(m_main_loop);
    pa_threaded_mainloop_free(m_main_loop);
}

bool PulseAudioContext::current_thread_is_main_loop_thread()
{
    return static_cast<bool>(pa_threaded_mainloop_in_thread(m_main_loop));
}

void PulseAudioContext::lock_main_loop()
{
    if (!current_thread_is_main_loop_thread())
        pa_threaded_mainloop_lock(m_main_loop);
}

void PulseAudioContext::unlock_main_loop()
{
    if (!current_thread_is_main_loop_thread())
        pa_threaded_mainloop_unlock(m_main_loop);
}

void PulseAudioContext::wait_for_signal()
{
    pa_threaded_mainloop_wait(m_main_loop);
}

void PulseAudioContext::signal_to_wake()
{
    pa_threaded_mainloop_signal(m_main_loop, 0);
}

PulseAudioContextState PulseAudioContext::get_connection_state()
{
    return static_cast<PulseAudioContextState>(pa_context_get_state(m_context));
}

bool PulseAudioContext::connection_is_good()
{
    return PA_CONTEXT_IS_GOOD(pa_context_get_state(m_context));
}

PulseAudioErrorCode PulseAudioContext::get_last_error()
{
    return static_cast<PulseAudioErrorCode>(pa_context_errno(m_context));
}

#define STREAM_SIGNAL_CALLBACK(stream)                                          \
    [](auto*, int, void* user_data) {                                           \
        static_cast<PulseAudioStream*>(user_data)->m_context->signal_to_wake(); \
    },                                                                          \
        (stream)

ErrorOr<NonnullRefPtr<PulseAudioStream>> PulseAudioContext::create_stream(OutputState initial_state, u32 sample_rate, u8 channels, u32 target_latency_ms, PulseAudioDataRequestCallback write_callback)
{
    auto locker = main_loop_locker();

    VERIFY(get_connection_state() == PulseAudioContextState::Ready);
    pa_sample_spec sample_specification {
        // FIXME: Support more audio sample types.
        __BYTE_ORDER__ == __ORDER_LITTLE_ENDIAN__ ? PA_SAMPLE_FLOAT32LE : PA_SAMPLE_FLOAT32BE,
        sample_rate,
        channels,
    };

    // Check the sample specification and channel map here. These are also checked by stream_new(),
    // but we can return a more accurate error if we check beforehand.
    if (pa_sample_spec_valid(&sample_specification) == 0)
        return Error::from_string_literal("PulseAudio sample specification is invalid");
    pa_channel_map channel_map;
    if (pa_channel_map_init_auto(&channel_map, sample_specification.channels, PA_CHANNEL_MAP_DEFAULT) == 0) {
        warnln("Getting default PulseAudio channel map failed with error: {}", pulse_audio_error_to_string(get_last_error()));
        return Error::from_string_literal("Failed to get default PulseAudio channel map");
    }

    // Create the stream object and set a callback to signal ourselves to wake when the stream changes states,
    // allowing us to wait synchronously for it to become Ready or Failed.
    auto* stream = pa_stream_new_with_proplist(m_context, "Audio Stream", &sample_specification, &channel_map, nullptr);
    if (stream == nullptr) {
        warnln("Instantiating PulseAudio stream failed with error: {}", pulse_audio_error_to_string(get_last_error()));
        return Error::from_string_literal("Failed to create PulseAudio stream");
    }
    pa_stream_set_state_callback(
        stream, [](pa_stream*, void* user_data) {
            static_cast<PulseAudioContext*>(user_data)->signal_to_wake();
        },
        this);

    auto stream_wrapper = TRY(adopt_nonnull_ref_or_enomem(new (nothrow) PulseAudioStream(NonnullRefPtr(*this), stream)));

    stream_wrapper->m_write_callback = move(write_callback);
    pa_stream_set_write_callback(
        stream, [](pa_stream* stream, size_t bytes_to_write, void* user_data) {
            auto& stream_wrapper = *static_cast<PulseAudioStream*>(user_data);
            VERIFY(stream_wrapper.m_stream == stream);
            stream_wrapper.on_write_requested(bytes_to_write);
        },
        stream_wrapper.ptr());

    // Borrowing logic from cubeb to set reasonable buffer sizes for a target latency:
    // https://searchfox.org/mozilla-central/rev/3b707c8fd7e978eebf24279ee51ccf07895cfbcb/third_party/rust/cubeb-sys/libcubeb/src/cubeb_pulse.c#910-927
    pa_buffer_attr buffer_attributes;
    buffer_attributes.maxlength = -1;
    buffer_attributes.prebuf = -1;
    buffer_attributes.tlength = target_latency_ms * sample_rate / 1000;
    buffer_attributes.minreq = buffer_attributes.tlength / 4;
    buffer_attributes.fragsize = buffer_attributes.minreq;
    auto flags = static_cast<pa_stream_flags>(PA_STREAM_AUTO_TIMING_UPDATE | PA_STREAM_INTERPOLATE_TIMING | PA_STREAM_ADJUST_LATENCY | PA_STREAM_RELATIVE_VOLUME);

    if (initial_state == OutputState::Suspended) {
        stream_wrapper->m_suspended = true;
        flags = static_cast<pa_stream_flags>(static_cast<u32>(flags) | PA_STREAM_START_CORKED);
    }

    // This is a workaround for an issue with starting the stream corked, see PulseAudioPlaybackStream::total_time_played().
    pa_stream_set_started_callback(
        stream, [](pa_stream* stream, void* user_data) {
            static_cast<PulseAudioStream*>(user_data)->m_started_playback = true;
            pa_stream_set_started_callback(stream, nullptr, nullptr);
        },
        stream_wrapper.ptr());

    pa_stream_set_underflow_callback(
        stream, [](pa_stream*, void* user_data) {
            auto& stream = *static_cast<PulseAudioStream*>(user_data);
            if (stream.m_underrun_callback)
                stream.m_underrun_callback();
        },
        stream_wrapper.ptr());

    if (auto error = pa_stream_connect_playback(stream, nullptr, &buffer_attributes, flags, nullptr, nullptr); error != 0) {
        warnln("Failed to start PulseAudio stream connection with error: {}", pulse_audio_error_to_string(static_cast<PulseAudioErrorCode>(error)));
        return Error::from_string_literal("Error while connecting the PulseAudio stream");
    }

    while (true) {
        bool is_ready = false;
        switch (stream_wrapper->get_connection_state()) {
        case PulseAudioStreamState::Creating:
            break;
        case PulseAudioStreamState::Ready:
            is_ready = true;
            break;
        case PulseAudioStreamState::Failed:
            warnln("PulseAudio stream connection failed with error: {}", pulse_audio_error_to_string(get_last_error()));
            return Error::from_string_literal("Failed to connect to PulseAudio daemon");
        case PulseAudioStreamState::Unconnected:
        case PulseAudioStreamState::Terminated:
            VERIFY_NOT_REACHED();
            break;
        }
        if (is_ready)
            break;

        wait_for_signal();
    }

    pa_stream_set_state_callback(stream, nullptr, nullptr);

    return stream_wrapper;
}

PulseAudioStream::~PulseAudioStream()
{
    auto locker = m_context->main_loop_locker();
    pa_stream_set_write_callback(m_stream, nullptr, nullptr);
    pa_stream_set_underflow_callback(m_stream, nullptr, nullptr);
    pa_stream_set_started_callback(m_stream, nullptr, nullptr);
    pa_stream_disconnect(m_stream);
    pa_stream_unref(m_stream);
}

PulseAudioStreamState PulseAudioStream::get_connection_state()
{
    return static_cast<PulseAudioStreamState>(pa_stream_get_state(m_stream));
}

bool PulseAudioStream::connection_is_good()
{
    return PA_STREAM_IS_GOOD(pa_stream_get_state(m_stream));
}

void PulseAudioStream::set_underrun_callback(Function<void()> callback)
{
    auto locker = m_context->main_loop_locker();
    m_underrun_callback = move(callback);
}

u32 PulseAudioStream::sample_rate()
{
    return pa_stream_get_sample_spec(m_stream)->rate;
}

size_t PulseAudioStream::sample_size()
{
    return pa_sample_size(pa_stream_get_sample_spec(m_stream));
}

size_t PulseAudioStream::frame_size()
{
    return pa_frame_size(pa_stream_get_sample_spec(m_stream));
}

u8 PulseAudioStream::channel_count()
{
    return pa_stream_get_sample_spec(m_stream)->channels;
}

void PulseAudioStream::on_write_requested(size_t bytes_to_write)
{
    VERIFY(m_write_callback);
    if (m_suspended)
        return;
    while (bytes_to_write > 0) {
        auto buffer = begin_write(bytes_to_write).release_value_but_fixme_should_propagate_errors();
        auto frame_size = this->frame_size();
        VERIFY(buffer.size() % frame_size == 0);
        auto written_buffer = m_write_callback(*this, buffer, buffer.size() / frame_size);
        if (written_buffer.size() == 0) {
            cancel_write().release_value_but_fixme_should_propagate_errors();
            break;
        }
        bytes_to_write -= written_buffer.size();
        write(written_buffer).release_value_but_fixme_should_propagate_errors();
    }
}

ErrorOr<Bytes> PulseAudioStream::begin_write(size_t bytes_to_write)
{
    void* data_pointer;
    size_t data_size = bytes_to_write;
    if (pa_stream_begin_write(m_stream, &data_pointer, &data_size) != 0 || data_pointer == nullptr)
        return Error::from_string_literal("Failed to get the playback stream's write buffer from PulseAudio");
    return Bytes { data_pointer, data_size };
}

ErrorOr<void> PulseAudioStream::write(ReadonlyBytes data)
{
    if (pa_stream_write(m_stream, data.data(), data.size(), nullptr, 0, PA_SEEK_RELATIVE) != 0)
        return Error::from_string_literal("Failed to write data to PulseAudio playback stream");
    return {};
}

ErrorOr<void> PulseAudioStream::cancel_write()
{
    if (pa_stream_cancel_write(m_stream) != 0)
        return Error::from_string_literal("Failed to get the playback stream's write buffer from PulseAudio");
    return {};
}

bool PulseAudioStream::is_suspended() const
{
    return m_suspended;
}

StringView pulse_audio_error_to_string(PulseAudioErrorCode code)
{
    if (code < PulseAudioErrorCode::OK || code >= PulseAudioErrorCode::Sentinel)
        return "Unknown error code"sv;

    char const* string = pa_strerror(static_cast<int>(code));
    return StringView { string, strlen(string) };
}

ErrorOr<void> PulseAudioStream::wait_for_operation(pa_operation* operation, StringView error_message)
{
    while (pa_operation_get_state(operation) == PA_OPERATION_RUNNING)
        m_context->wait_for_signal();
    if (!m_context->connection_is_good() || !this->connection_is_good()) {
        auto pulse_audio_error_name = pulse_audio_error_to_string(m_context->get_last_error());
        warnln("Encountered stream error: {}", pulse_audio_error_name);
        return Error::from_string_view(error_message);
    }
    pa_operation_unref(operation);
    return {};
}

ErrorOr<void> PulseAudioStream::drain_and_suspend()
{
    auto locker = m_context->main_loop_locker();

    if (m_suspended)
        return {};
    m_suspended = true;

    if (pa_stream_is_corked(m_stream) > 0)
        return {};

    TRY(wait_for_operation(pa_stream_drain(m_stream, STREAM_SIGNAL_CALLBACK(this)), "Draining PulseAudio stream failed"sv));
    TRY(wait_for_operation(pa_stream_cork(m_stream, 1, STREAM_SIGNAL_CALLBACK(this)), "Corking PulseAudio stream after drain failed"sv));
    return {};
}

ErrorOr<void> PulseAudioStream::flush_and_suspend()
{
    auto locker = m_context->main_loop_locker();

    if (m_suspended)
        return {};
    m_suspended = true;

    if (pa_stream_is_corked(m_stream) > 0)
        return {};

    TRY(wait_for_operation(pa_stream_flush(m_stream, STREAM_SIGNAL_CALLBACK(this)), "Flushing PulseAudio stream failed"sv));
    TRY(wait_for_operation(pa_stream_cork(m_stream, 1, STREAM_SIGNAL_CALLBACK(this)), "Corking PulseAudio stream after flush failed"sv));
    return {};
}

ErrorOr<void> PulseAudioStream::resume()
{
    auto locker = m_context->main_loop_locker();

    if (!m_suspended)
        return {};
    m_suspended = false;

    TRY(wait_for_operation(pa_stream_cork(m_stream, 0, STREAM_SIGNAL_CALLBACK(this)), "Uncorking PulseAudio stream failed"sv));

    // Defer a write to the playback buffer on the PulseAudio main loop. Otherwise, playback will not
    // begin again, despite the fact that we uncorked.
    // NOTE: We ref here and then unref in the callback so that this stream will not be deleted until
    //       it finishes.
    ref();
    pa_mainloop_api_once(
        m_context->m_api, [](pa_mainloop_api*, void* user_data) {
            auto& stream = *static_cast<PulseAudioStream*>(user_data);
            // NOTE: writable_size() returns -1 in case of an error. However, the value is still safe
            //       since begin_write() will interpret -1 as a default parameter and choose a good size.
            auto bytes_to_write = pa_stream_writable_size(stream.m_stream);
            stream.on_write_requested(bytes_to_write);
            stream.unref();
        },
        this);
    return {};
}

ErrorOr<Duration> PulseAudioStream::total_time_played()
{
    auto locker = m_context->main_loop_locker();

    // NOTE: This is a workaround for a PulseAudio issue. When a stream is started corked,
    //       the time smoother doesn't seem to be aware of it, so it will return the time
    //       since the stream was connected. Once the playback actually starts, the time
    //       resets back to zero. However, since we request monotonically-increasing time,
    //       this means that the smoother will register that it had a larger time before,
    //       and return that time instead, until we reach a timestamp greater than the
    //       last-returned time. If we never call pa_stream_get_time() until after giving
    //       the stream its first samples, the issue never occurs.
    if (!m_started_playback)
        return Duration::zero();

    pa_usec_t time = 0;
    auto error = pa_stream_get_time(m_stream, &time);
    if (error == -PA_ERR_NODATA)
        return Duration::zero();
    if (error != 0)
        return Error::from_string_literal("Failed to get time from PulseAudio stream");
    if (time > NumericLimits<i64>::max()) {
        warnln("WARNING: Audio time is too large!");
        time -= NumericLimits<i64>::max();
    }
    return Duration::from_microseconds(static_cast<i64>(time));
}

ErrorOr<void> PulseAudioStream::set_volume(double volume)
{
    auto locker = m_context->main_loop_locker();

    auto index = pa_stream_get_index(m_stream);
    if (index == PA_INVALID_INDEX)
        return Error::from_string_literal("Failed to get PulseAudio stream index while setting volume");

    auto pulse_volume = pa_sw_volume_from_linear(volume);
    pa_cvolume per_channel_volumes;
    pa_cvolume_set(&per_channel_volumes, channel_count(), pulse_volume);

    auto* operation = pa_context_set_sink_input_volume(m_context->m_context, index, &per_channel_volumes, STREAM_SIGNAL_CALLBACK(this));
    return wait_for_operation(operation, "Failed to set PulseAudio stream volume"sv);
}

}