LibAudio: Add a Serenity implementation of PlaybackStream

This implementation is very naive compared to the PulseAudio one.

Instead of using a callback implemented by the audio server connection
to push audio to the buffer, we have to poll on a timer to check when
we need to push the audio buffers. Implementing cross-process condition
variables into the audio queue class could allow us to avoid polling,
which may prove beneficial to CPU usage.

Audio timestamps will be accurate to the number of samples available,
but will count in increments of about 100ms and run ahead of the actual
audio being pushed to the device by the server.

Buffer underruns are completely ignored for now as well, since the
`AudioServer` has no way to know how many samples are actually written
in a single audio buffer.

Tests/LibAudio/TestPlaybackStream.cpp

@@ -8,6 +8,7 @@
 #include <AK/MemoryStream.h>
 #include <AK/WeakPtr.h>
 #include <LibAudio/PlaybackStream.h>
+#include <LibCore/EventLoop.h>
 #include <LibTest/TestSuite.h>
 #include <unistd.h>
 
@@ -15,10 +16,23 @@
 #    include <LibAudio/PulseAudioWrappers.h>
 #endif
 
-TEST_CASE(create_and_destroy_playback_stream)
+// FIXME: CI doesn't run an AudioServer currently. Creating one in /etc/SystemServer.ini does not
+//        allow this test to pass since CI runs in a Shell that will setsid() if it finds that the
+//        current session ID is 0, and AudioServer's socket address depends on the current sid.
+//        If we can fix that, this test can run on CI.
+//        https://github.com/SerenityOS/serenity/issues/20538
+#if defined(AK_OS_SERENITY)
+#    define STREAM_TEST BENCHMARK_CASE
+#else
+#    define STREAM_TEST TEST_CASE
+#endif
+
+STREAM_TEST(create_and_destroy_playback_stream)
 {
+    Core::EventLoop event_loop;
+
     bool has_implementation = false;
-#if defined(HAVE_PULSEAUDIO)
+#if defined(AK_OS_SERENITY) || defined(HAVE_PULSEAUDIO)
     has_implementation = true;
 #endif
 

Userland/Libraries/LibAudio/CMakeLists.txt

@@ -19,6 +19,7 @@ set(SOURCES
 if (SERENITYOS)
     list(APPEND SOURCES ConnectionToServer.cpp)
     list(APPEND SOURCES ConnectionToManagerServer.cpp)
+    list(APPEND SOURCES PlaybackStreamSerenity.cpp)
     set(GENERATED_SOURCES
         ../../Services/AudioServer/AudioClientEndpoint.h
         ../../Services/AudioServer/AudioServerEndpoint.h

Userland/Libraries/LibAudio/ConnectionToServer.cpp

@@ -150,6 +150,11 @@ size_t ConnectionToServer::remaining_buffers() const
     return m_buffer->size() - m_buffer->weak_remaining_capacity();
 }
 
+bool ConnectionToServer::can_enqueue() const
+{
+    return m_buffer->can_enqueue();
+}
+
 void ConnectionToServer::client_volume_changed(double volume)
 {
     if (on_client_volume_change)

Userland/Libraries/LibAudio/ConnectionToServer.h

@@ -54,6 +54,8 @@ public:
     // How many buffers (i.e. short sample arrays) the server hasn't played yet.
     // Non-realtime code needn't worry about this.
     size_t remaining_buffers() const;
+    // Whether there is room in the realtime audio queue for another sample buffer.
+    bool can_enqueue() const;
 
     void set_self_sample_rate(u32 sample_rate);
 

Userland/Libraries/LibAudio/PlaybackStream.cpp

@@ -6,8 +6,13 @@
 
 #include "PlaybackStream.h"
 
+#include <AK/Platform.h>
 #include <LibCore/ThreadedPromise.h>
 
+#if defined(AK_OS_SERENITY)
+#    include <LibAudio/PlaybackStreamSerenity.h>
+#endif
+
 #if defined(HAVE_PULSEAUDIO)
 #    include <LibAudio/PlaybackStreamPulseAudio.h>
 #endif
@@ -28,7 +33,9 @@ ErrorOr<NonnullRefPtr<PlaybackStream>> PlaybackStream::create(OutputState initia
 {
     VERIFY(data_request_callback);
     // Create the platform-specific implementation for this stream.
-#if defined(HAVE_PULSEAUDIO)
+#if defined(AK_OS_SERENITY)
+    return PlaybackStreamSerenity::create(initial_output_state, sample_rate, channels, target_latency_ms, move(data_request_callback));
+#elif defined(HAVE_PULSEAUDIO)
     return PlaybackStreamPulseAudio::create(initial_output_state, sample_rate, channels, target_latency_ms, move(data_request_callback));
 #else
     (void)initial_output_state, (void)sample_rate, (void)channels, (void)target_latency_ms;

Userland/Libraries/LibAudio/PlaybackStreamSerenity.cpp

@@ -0,0 +1,112 @@
+/*
+ * Copyright (c) 2023, Gregory Bertilson <zaggy1024@gmail.com>
+ *
+ * SPDX-License-Identifier: BSD-2-Clause
+ */
+
+#include "PlaybackStreamSerenity.h"
+
+#include <LibCore/ThreadedPromise.h>
+
+namespace Audio {
+
+ErrorOr<NonnullRefPtr<PlaybackStream>> PlaybackStreamSerenity::create(OutputState initial_state, u32 sample_rate, [[maybe_unused]] u8 channels, [[maybe_unused]] u32 target_latency_ms, AudioDataRequestCallback&& data_request_callback)
+{
+    VERIFY(data_request_callback);
+    auto connection = TRY(ConnectionToServer::try_create());
+    if (auto result = connection->try_set_self_sample_rate(sample_rate); result.is_error())
+        return Error::from_string_literal("Failed to set sample rate");
+
+    auto polling_timer = TRY(Core::Timer::try_create());
+    auto implementation = TRY(adopt_nonnull_ref_or_enomem(new (nothrow) PlaybackStreamSerenity(connection, move(polling_timer), move(data_request_callback))));
+    if (initial_state == OutputState::Playing)
+        connection->async_start_playback();
+    return implementation;
+}
+
+PlaybackStreamSerenity::PlaybackStreamSerenity(NonnullRefPtr<ConnectionToServer> stream, NonnullRefPtr<Core::Timer> polling_timer, AudioDataRequestCallback&& data_request_callback)
+    : m_connection(move(stream))
+    , m_polling_timer(move(polling_timer))
+    , m_data_request_callback(move(data_request_callback))
+{
+    // Ensure that our audio buffers are filled when they are more than 3/4 empty.
+    // FIXME: Add an event to ConnectionToServer track the sample rate and update this interval, or
+    //        implement the data request into ConnectionToServer so each client doesn't need to poll
+    //        on a timer with an arbitrary interval.
+    m_polling_timer->set_interval(static_cast<int>((AUDIO_BUFFERS_COUNT * 3 / 4) * AUDIO_BUFFER_SIZE * 1000 / m_connection->get_self_sample_rate()));
+    m_polling_timer->on_timeout = [this]() {
+        fill_buffers();
+    };
+    m_polling_timer->start();
+}
+
+void PlaybackStreamSerenity::fill_buffers()
+{
+    while (m_connection->can_enqueue()) {
+        Array<Sample, AUDIO_BUFFER_SIZE> buffer;
+        buffer.fill({ 0.0f, 0.0f });
+        auto written_data = m_data_request_callback(Bytes { reinterpret_cast<u8*>(buffer.data()), sizeof(buffer) }, PcmSampleFormat::Float32, AUDIO_BUFFER_SIZE);
+        // FIXME: The buffer we are enqueuing here is a fixed size, meaning that the server will not be
+        //        aware of exactly how many samples we have written here. We should allow the server to
+        //        consume sized buffers to allow us to obtain sample-accurate timing information even
+        //        when we run out of samples on a sample count that is not a multiple of AUDIO_BUFFER_SIZE.
+        m_number_of_samples_enqueued += written_data.size() / sizeof(Sample);
+        MUST(m_connection->realtime_enqueue(buffer));
+    }
+}
+
+void PlaybackStreamSerenity::set_underrun_callback(Function<void()> callback)
+{
+    // FIXME: Implement underrun callback in AudioServer
+    (void)callback;
+}
+
+NonnullRefPtr<Core::ThreadedPromise<Duration>> PlaybackStreamSerenity::resume()
+{
+    auto promise = Core::ThreadedPromise<Duration>::create();
+    // FIXME: We need to get the time played at the correct time from the server. If a message to
+    //        start playback is sent while there is any other message being processed, this may end
+    //        up being inaccurate.
+    auto time = MUST(total_time_played());
+    fill_buffers();
+    m_connection->async_start_playback();
+    m_polling_timer->start();
+    promise->resolve(move(time));
+    return promise;
+}
+
+NonnullRefPtr<Core::ThreadedPromise<void>> PlaybackStreamSerenity::drain_buffer_and_suspend()
+{
+    // FIXME: Play back all samples on the server before pausing. This can be achieved by stopping
+    //        enqueuing samples and receiving a message that a buffer underrun has occurred.
+    auto promise = Core::ThreadedPromise<void>::create();
+    m_connection->async_pause_playback();
+    m_polling_timer->stop();
+    promise->resolve();
+    return promise;
+}
+
+NonnullRefPtr<Core::ThreadedPromise<void>> PlaybackStreamSerenity::discard_buffer_and_suspend()
+{
+    auto promise = Core::ThreadedPromise<void>::create();
+    m_connection->async_clear_buffer();
+    m_connection->async_pause_playback();
+    m_polling_timer->stop();
+    promise->resolve();
+    return promise;
+}
+
+ErrorOr<Duration> PlaybackStreamSerenity::total_time_played()
+{
+    return Duration::from_milliseconds(m_number_of_samples_enqueued * 1000 / m_connection->get_self_sample_rate());
+}
+
+NonnullRefPtr<Core::ThreadedPromise<void>> PlaybackStreamSerenity::set_volume(double volume)
+{
+    auto promise = Core::ThreadedPromise<void>::create();
+    m_connection->async_set_self_volume(volume);
+    promise->resolve();
+    return promise;
+}
+
+}

Userland/Libraries/LibAudio/PlaybackStreamSerenity.h

@@ -0,0 +1,41 @@
+/*
+ * Copyright (c) 2023, Gregory Bertilson <zaggy1024@gmail.com>
+ *
+ * SPDX-License-Identifier: BSD-2-Clause
+ */
+
+#pragma once
+
+#include <LibAudio/ConnectionToServer.h>
+#include <LibAudio/PlaybackStream.h>
+
+namespace Audio {
+
+class PlaybackStreamSerenity final
+    : public PlaybackStream {
+public:
+    static ErrorOr<NonnullRefPtr<PlaybackStream>> create(OutputState initial_state, u32 sample_rate, u8 channels, u32 target_latency_ms, AudioDataRequestCallback&& data_request_callback);
+
+    virtual void set_underrun_callback(Function<void()>) override;
+
+    virtual NonnullRefPtr<Core::ThreadedPromise<Duration>> resume() override;
+    virtual NonnullRefPtr<Core::ThreadedPromise<void>> drain_buffer_and_suspend() override;
+    virtual NonnullRefPtr<Core::ThreadedPromise<void>> discard_buffer_and_suspend() override;
+
+    virtual ErrorOr<Duration> total_time_played() override;
+
+    virtual NonnullRefPtr<Core::ThreadedPromise<void>> set_volume(double) override;
+
+private:
+    PlaybackStreamSerenity(NonnullRefPtr<ConnectionToServer>, NonnullRefPtr<Core::Timer> polling_timer, AudioDataRequestCallback&& data_request_callback);
+
+    void fill_buffers();
+
+    NonnullRefPtr<ConnectionToServer> m_connection;
+    size_t m_number_of_samples_enqueued { 0 };
+    NonnullRefPtr<Core::Timer> m_polling_timer;
+    AudioDataRequestCallback m_data_request_callback;
+    bool m_paused { false };
+};
+
+}