From ee837a3e90b5fc1b3e861a4c88412e745ccb1526 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Via=C4=8Das=C5=82a=C5=AD=20Chalikin?= Date: Sun, 29 Dec 2024 12:12:24 +0300 Subject: [PATCH] PipeWire improvements * Fix freeze when pipewire service is stopped/restarted * Fix `device_list `memleak * Refactor pipewire drivers --- audio/common/pipewire.c | 102 ++++++++-- audio/common/pipewire.h | 40 ++-- audio/drivers/pipewire.c | 270 ++++++++++--------------- audio/drivers_microphone/pipewire.c | 299 ++++++++++++---------------- 4 files changed, 335 insertions(+), 376 deletions(-) diff --git a/audio/common/pipewire.c b/audio/common/pipewire.c index 1fc213c23cc..be05b221bf8 100644 --- a/audio/common/pipewire.c +++ b/audio/common/pipewire.c @@ -15,12 +15,46 @@ #include "pipewire.h" +#include + #include #include #include "verbosity.h" + +static void core_error_cb(void *data, uint32_t id, int seq, int res, const char *message) +{ + pipewire_core_t *pw = (pipewire_core_t*)data; + + RARCH_ERR("[PipeWire]: error id:%u seq:%d res:%d (%s): %s\n", + id, seq, res, spa_strerror(res), message); + + /* stop and exit the thread loop */ + pw_thread_loop_stop(pw->thread_loop); +} + +static void core_done_cb(void *data, uint32_t id, int seq) +{ + pipewire_core_t *pw = (pipewire_core_t*)data; + + retro_assert(id == PW_ID_CORE); + + pw->last_seq = seq; + if (pw->pending_seq == seq) + { + /* stop and exit the thread loop */ + pw_thread_loop_signal(pw->thread_loop, false); + } +} + +static const struct pw_core_events core_events = { + PW_VERSION_CORE_EVENTS, + .done = core_done_cb, + .error = core_error_cb, +}; + size_t calc_frame_size(enum spa_audio_format fmt, uint32_t nchannels) { uint32_t sample_size = 1; @@ -80,37 +114,61 @@ void set_position(uint32_t channels, uint32_t position[SPA_AUDIO_MAX_CHANNELS]) } } -int pipewire_wait_resync(pipewire_core_t *pipewire) +void pipewire_wait_resync(pipewire_core_t *pw) { - int res; - retro_assert(pipewire != NULL); - - pipewire->pending_seq = pw_core_sync(pipewire->core, PW_ID_CORE, pipewire->pending_seq); + retro_assert(pw); + pw->pending_seq = pw_core_sync(pw->core, PW_ID_CORE, pw->pending_seq); for (;;) { - pw_thread_loop_wait(pipewire->thread_loop); - - res = pipewire->error; - if (res < 0) - { - pipewire->error = 0; - return res; - } - if (pipewire->pending_seq == pipewire->last_seq) + pw_thread_loop_wait(pw->thread_loop); + if (pw->pending_seq == pw->last_seq) break; } - return 0; } -bool pipewire_set_active(pipewire_core_t *pipewire, pipewire_device_handle_t *device, bool active) +bool pipewire_set_active(struct pw_thread_loop *loop, struct pw_stream *stream, bool active) { - RARCH_LOG("[PipeWire]: %s.\n", active? "Unpausing": "Pausing"); + enum pw_stream_state st; + const char *error; + + retro_assert(loop); + retro_assert(stream); + + pw_thread_loop_lock(loop); + pw_stream_set_active(stream, active); + pw_thread_loop_wait(loop); + pw_thread_loop_unlock(loop); + + st = pw_stream_get_state(stream, &error); + return active ? st == PW_STREAM_STATE_STREAMING : st == PW_STREAM_STATE_PAUSED; +} + +bool pipewire_core_init(pipewire_core_t *pw, const char *loop_name) +{ + retro_assert(pw); + + pw->thread_loop = pw_thread_loop_new(loop_name, NULL); + if (!pw->thread_loop) + return false; + + pw->ctx = pw_context_new(pw_thread_loop_get_loop(pw->thread_loop), NULL, 0); + if (!pw->ctx) + return false; + + if (pw_thread_loop_start(pw->thread_loop) < 0) + return false; + + pw_thread_loop_lock(pw->thread_loop); + + pw->core = pw_context_connect(pw->ctx, NULL, 0); + if(!pw->core) + return false; - pw_thread_loop_lock(pipewire->thread_loop); - pw_stream_set_active(device->stream, active); - pw_thread_loop_wait(pipewire->thread_loop); - pw_thread_loop_unlock(pipewire->thread_loop); + if (pw_core_add_listener(pw->core, + &pw->core_listener, + &core_events, pw) < 0) + return false; - return device->is_paused != active; + return true; } diff --git a/audio/common/pipewire.h b/audio/common/pipewire.h index 7c7ea323f96..7e9627128ac 100644 --- a/audio/common/pipewire.h +++ b/audio/common/pipewire.h @@ -22,17 +22,25 @@ #include #include +#include -#define RINGBUFFER_SIZE (1u << 22) -#define RINGBUFFER_MASK (RINGBUFFER_SIZE - 1) +#define PW_RARCH_APPNAME "RetroArch" -typedef struct pipewire +/* String literals are part of the PipeWire specification */ +#define PW_RARCH_MEDIA_TYPE_AUDIO "Audio" +#define PW_RARCH_MEDIA_TYPE_VIDEO "Video" +#define PW_RARCH_MEDIA_TYPE_MIDI "Midi" +#define PW_RARCH_MEDIA_CATEGORY_PLAYBACK "Playback" +#define PW_RARCH_MEDIA_CATEGORY_RECORD "Capture" +#define PW_RARCH_MEDIA_ROLE "Game" + +typedef struct pipewire_core { struct pw_thread_loop *thread_loop; - struct pw_context *context; + struct pw_context *ctx; struct pw_core *core; struct spa_hook core_listener; - int last_seq, pending_seq, error; + int last_seq, pending_seq; struct pw_registry *registry; struct spa_hook registry_listener; @@ -43,28 +51,14 @@ typedef struct pipewire struct string_list *devicelist; } pipewire_core_t; -typedef struct pipewire_device_handle -{ - pipewire_core_t *pw; - - struct pw_stream *stream; - struct spa_hook stream_listener; - struct spa_audio_info_raw info; - uint32_t highwater_mark; - uint32_t frame_size; - uint32_t req; - struct spa_ringbuffer ring; - uint8_t buffer[RINGBUFFER_SIZE]; - - bool is_paused; -} pipewire_device_handle_t; - size_t calc_frame_size(enum spa_audio_format fmt, uint32_t nchannels); void set_position(uint32_t channels, uint32_t position[SPA_AUDIO_MAX_CHANNELS]); -int pipewire_wait_resync(pipewire_core_t *pipewire); +void pipewire_wait_resync(pipewire_core_t *pipewire); + +bool pipewire_set_active(struct pw_thread_loop *loop, struct pw_stream *stream, bool active); -bool pipewire_set_active(pipewire_core_t *pipewire, pipewire_device_handle_t *device, bool active); +bool pipewire_core_init(pipewire_core_t *pipewire, const char *loop_name); #endif /* _RETROARCH_PIPEWIRE */ diff --git a/audio/drivers/pipewire.c b/audio/drivers/pipewire.c index dbc4320bb79..1db13f617e6 100644 --- a/audio/drivers/pipewire.c +++ b/audio/drivers/pipewire.c @@ -35,20 +35,36 @@ #include "verbosity.h" -#define APPNAME "RetroArch" -#define DEFAULT_CHANNELS 2 -#define QUANTUM 1024 /* TODO: detect */ +#define DEFAULT_CHANNELS 2 +#define QUANTUM 1024 /* TODO: detect */ +#define RINGBUFFER_SIZE (1u << 22) +#define RINGBUFFER_MASK (RINGBUFFER_SIZE - 1) -static void stream_destroy(void *data) +typedef struct pipewire_audio { - pipewire_device_handle_t *audio = (pipewire_device_handle_t*)data; + pipewire_core_t *pw; + + struct pw_stream *stream; + struct spa_hook stream_listener; + struct spa_audio_info_raw info; + uint32_t highwater_mark; + uint32_t frame_size; + uint32_t req; + struct spa_ringbuffer ring; + uint8_t buffer[RINGBUFFER_SIZE]; +} pipewire_audio_t; + + +static void stream_destroy_cb(void *data) +{ + pipewire_audio_t *audio = (pipewire_audio_t*)data; spa_hook_remove(&audio->stream_listener); audio->stream = NULL; } -static void on_process(void *data) +static void playback_process_cb(void *data) { - pipewire_device_handle_t *audio = (pipewire_device_handle_t*)data; + pipewire_audio_t *audio = (pipewire_audio_t*)data; void *p; struct pw_buffer *b; struct spa_buffer *buf; @@ -103,10 +119,12 @@ static void on_process(void *data) pw_stream_queue_buffer(audio->stream, b); } -static void on_stream_state_changed(void *data, +static void pipewire_free(void *data); + +static void stream_state_changed_cb(void *data, enum pw_stream_state old, enum pw_stream_state state, const char *error) { - pipewire_device_handle_t *audio = (pipewire_device_handle_t*)data; + pipewire_audio_t *audio = (pipewire_audio_t*)data; RARCH_DBG("[PipeWire]: New state for Sink Node %d : %s\n", pw_stream_get_node_id(audio->stream), @@ -114,13 +132,12 @@ static void on_stream_state_changed(void *data, switch(state) { - case PW_STREAM_STATE_STREAMING: - audio->is_paused = false; - pw_thread_loop_signal(audio->pw->thread_loop, false); + case PW_STREAM_STATE_UNCONNECTED: + pw_thread_loop_stop(audio->pw->thread_loop); break; + case PW_STREAM_STATE_STREAMING: case PW_STREAM_STATE_ERROR: case PW_STREAM_STATE_PAUSED: - audio->is_paused = true; pw_thread_loop_signal(audio->pw->thread_loop, false); break; default: @@ -130,58 +147,27 @@ static void on_stream_state_changed(void *data, static const struct pw_stream_events playback_stream_events = { PW_VERSION_STREAM_EVENTS, - .destroy = stream_destroy, - .process = on_process, - .state_changed = on_stream_state_changed, + .destroy = stream_destroy_cb, + .process = playback_process_cb, + .state_changed = stream_state_changed_cb, }; -static void client_info(void *data, const struct pw_client_info *info) +static void client_info_cb(void *data, const struct pw_client_info *info) { - pipewire_device_handle_t *audio = (pipewire_device_handle_t*)data; const struct spa_dict_item *item; + pipewire_core_t *pw = (pipewire_core_t*)data; RARCH_DBG("[PipeWire]: client: id:%u\n", info->id); RARCH_DBG("[PipeWire]: \tprops:\n"); spa_dict_for_each(item, info->props) RARCH_DBG("[PipeWire]: \t\t%s: \"%s\"\n", item->key, item->value); - pw_thread_loop_signal(audio->pw->thread_loop, false); + pw_thread_loop_signal(pw->thread_loop, false); } static const struct pw_client_events client_events = { PW_VERSION_CLIENT_EVENTS, - .info = client_info, -}; - -static void on_core_error(void *data, uint32_t id, int seq, int res, const char *message) -{ - pipewire_device_handle_t *audio = data; - - RARCH_ERR("[PipeWire]: error id:%u seq:%d res:%d (%s): %s\n", - id, seq, res, spa_strerror(res), message); - - /* stop and exit the thread loop */ - pw_thread_loop_signal(audio->pw->thread_loop, false); -} - -static void on_core_done(void *data, uint32_t id, int seq) -{ - pipewire_device_handle_t *audio = (pipewire_device_handle_t*)data; - - retro_assert(id == PW_ID_CORE); - - audio->pw->last_seq = seq; - if (audio->pw->pending_seq == seq) - { - /* stop and exit the thread loop */ - pw_thread_loop_signal(audio->pw->thread_loop, false); - } -} - -static const struct pw_core_events core_events = { - PW_VERSION_CORE_EVENTS, - .done = on_core_done, - .error = on_core_error, + .info = client_info_cb, }; static void registry_event_global(void *data, uint32_t id, @@ -189,23 +175,17 @@ static void registry_event_global(void *data, uint32_t id, const struct spa_dict *props) { union string_list_elem_attr attr; - pipewire_device_handle_t *audio = (pipewire_device_handle_t*)data; - pipewire_core_t *pipewire = NULL; + pipewire_core_t *pw = (pipewire_core_t*)data; const char *media = NULL; const char *sink = NULL; - if (!audio || !audio->pw) - return; - - pipewire = audio->pw; - - if (!pipewire->client && spa_streq(type, PW_TYPE_INTERFACE_Client)) + if (!pw->client && spa_streq(type, PW_TYPE_INTERFACE_Client)) { - pipewire->client = pw_registry_bind(audio->pw->registry, + pw->client = pw_registry_bind(pw->registry, id, type, PW_VERSION_CLIENT, 0); - pw_client_add_listener(audio->pw->client, - &audio->pw->client_listener, - &client_events, audio); + pw_client_add_listener(pw->client, + &pw->client_listener, + &client_events, pw); } else if (spa_streq(type, PW_TYPE_INTERFACE_Node)) { @@ -215,7 +195,7 @@ static void registry_event_global(void *data, uint32_t id, if ((sink = spa_dict_lookup(props, PW_KEY_NODE_NAME)) != NULL) { attr.i = id; - string_list_append(audio->pw->devicelist, sink, attr); + string_list_append(pw->devicelist, sink, attr); RARCH_LOG("[PipeWire]: Found Sink Node: %s\n", sink); } } @@ -232,54 +212,32 @@ static const struct pw_registry_events registry_events = { .global = registry_event_global, }; -static void pipewire_free(void *data); - static void *pipewire_init(const char *device, unsigned rate, unsigned latency, unsigned block_frames, unsigned *new_rate) { - int res; - uint64_t buf_samples; - const struct spa_pod *params[1]; - uint8_t buffer[1024]; - struct pw_properties *props = NULL; - const char *error = NULL; - pipewire_device_handle_t *audio = (pipewire_device_handle_t*)calloc(1, sizeof(*audio)); - pipewire_core_t *pipewire = NULL; - struct spa_pod_builder b = SPA_POD_BUILDER_INIT(buffer, sizeof(buffer)); - - if (!audio) - goto error; + int res; + uint64_t buf_samples; + const struct spa_pod *params[1]; + uint8_t buffer[1024]; + struct pw_properties *props = NULL; + const char *error = NULL; + pipewire_audio_t *audio = (pipewire_audio_t*)calloc(1, sizeof(*audio)); + pipewire_core_t *pw = NULL; + struct spa_pod_builder b = SPA_POD_BUILDER_INIT(buffer, sizeof(buffer)); pw_init(NULL, NULL); - pipewire = audio->pw = (pipewire_core_t*)calloc(1, sizeof(*audio->pw)); - - pipewire->devicelist = string_list_new(); - if (!pipewire->devicelist) - goto error; - - pipewire->thread_loop = pw_thread_loop_new("audio_driver", NULL); - if (!pipewire->thread_loop) - goto error; - - pipewire->context = pw_context_new(pw_thread_loop_get_loop(pipewire->thread_loop), NULL, 0); - if (!pipewire->context) - goto error; - - if (pw_thread_loop_start(pipewire->thread_loop) < 0) + if (!audio) goto error; + pw = audio->pw = (pipewire_core_t*)calloc(1, sizeof(*audio->pw)); - pw_thread_loop_lock(pipewire->thread_loop); - - pipewire->core = pw_context_connect(pipewire->context, NULL, 0); - if(!pipewire->core) + pw->devicelist = string_list_new(); + if (!pw->devicelist) goto error; - if (pw_core_add_listener(pipewire->core, - &pipewire->core_listener, - &core_events, audio) < 0) + if (!pipewire_core_init(pw, "audio_driver")) goto error; audio->info.format = is_little_endian() ? SPA_AUDIO_FORMAT_F32_LE : SPA_AUDIO_FORMAT_F32_BE; @@ -289,15 +247,14 @@ static void *pipewire_init(const char *device, unsigned rate, audio->frame_size = calc_frame_size(audio->info.format, DEFAULT_CHANNELS); audio->req = QUANTUM * rate * 1 / 2 / 100000 * audio->frame_size; - props = pw_properties_new(PW_KEY_MEDIA_TYPE, "Audio", - PW_KEY_MEDIA_CATEGORY, "Playback", - PW_KEY_MEDIA_ROLE, "Game", - PW_KEY_NODE_NAME, APPNAME, - PW_KEY_NODE_DESCRIPTION, APPNAME, - PW_KEY_APP_NAME, APPNAME, - PW_KEY_APP_ID, APPNAME, - PW_KEY_APP_ICON_NAME, APPNAME, - PW_KEY_NODE_ALWAYS_PROCESS, "true", + props = pw_properties_new(PW_KEY_MEDIA_TYPE, PW_RARCH_MEDIA_TYPE_AUDIO, + PW_KEY_MEDIA_CATEGORY, PW_RARCH_MEDIA_CATEGORY_PLAYBACK, + PW_KEY_MEDIA_ROLE, PW_RARCH_MEDIA_ROLE, + PW_KEY_NODE_NAME, PW_RARCH_APPNAME, + PW_KEY_NODE_DESCRIPTION, PW_RARCH_APPNAME, + PW_KEY_APP_NAME, PW_RARCH_APPNAME, + PW_KEY_APP_ID, PW_RARCH_APPNAME, + PW_KEY_APP_ICON_NAME, PW_RARCH_APPNAME, NULL); if (!props) goto error; @@ -312,7 +269,7 @@ static void *pipewire_init(const char *device, unsigned rate, pw_properties_setf(props, PW_KEY_NODE_RATE, "1/%d", rate); - audio->stream = pw_stream_new(pipewire->core, APPNAME, props); + audio->stream = pw_stream_new(pw->core, PW_RARCH_APPNAME, props); if (!audio->stream) goto error; @@ -327,7 +284,6 @@ static void *pipewire_init(const char *device, unsigned rate, PW_DIRECTION_OUTPUT, PW_ID_ANY, PW_STREAM_FLAG_AUTOCONNECT | - PW_STREAM_FLAG_INACTIVE | PW_STREAM_FLAG_MAP_BUFFERS | PW_STREAM_FLAG_RT_PROCESS, params, 1); @@ -339,21 +295,14 @@ static void *pipewire_init(const char *device, unsigned rate, latency? (latency * 1000): 46440 * (uint64_t)rate / 1000000 * audio->frame_size); RARCH_DBG("[PipeWire]: Bufer size: %u, RingBuffer size: %u\n", audio->highwater_mark, RINGBUFFER_SIZE); - audio->pw->registry = pw_core_get_registry(pipewire->core, PW_VERSION_REGISTRY, 0); + pw->registry = pw_core_get_registry(pw->core, PW_VERSION_REGISTRY, 0); - spa_zero(pipewire->registry_listener); - pw_registry_add_listener(pipewire->registry, &pipewire->registry_listener, ®istry_events, audio); + spa_zero(pw->registry_listener); + pw_registry_add_listener(pw->registry, &pw->registry_listener, ®istry_events, pw); /* unlock, run the loop and wait, this will trigger the callbacks */ - if (pipewire_wait_resync(audio->pw) < 0) - pw_thread_loop_unlock(pipewire->thread_loop); - - - if(pw_stream_get_state(audio->stream, &error) != PW_STREAM_STATE_STREAMING) - audio->is_paused = true; - - pw_thread_loop_unlock(audio->pw->thread_loop); - *new_rate = audio->info.rate; + pipewire_wait_resync(pw); + pw_thread_loop_unlock(pw->thread_loop); return audio; @@ -363,30 +312,18 @@ static void *pipewire_init(const char *device, unsigned rate, return NULL; } -static bool pipewire_start(void *data, bool is_shutdown); - static ssize_t pipewire_write(void *data, const void *buf_, size_t size) { - int32_t writable; - int32_t avail; - uint32_t index; - pipewire_device_handle_t *audio = (pipewire_device_handle_t*)data; - const char *error = NULL; - - /* Workaround buggy menu code. - * If a write happens while we're paused, we might never progress. */ - if (audio->is_paused) - if (!pipewire_start(audio, false)) - return -1; - - pw_thread_loop_lock(audio->pw->thread_loop); + int32_t writable; + int32_t avail; + uint32_t index; + pipewire_audio_t *audio = (pipewire_audio_t*)data; + const char *error = NULL; if (pw_stream_get_state(audio->stream, &error) != PW_STREAM_STATE_STREAMING) - { - /* wait for stream to become ready */ - size = 0; - goto unlock; - } + return 0; /* wait for stream to become ready */ + + pw_thread_loop_lock(audio->pw->thread_loop); writable = spa_ringbuffer_get_write_index(&audio->ring, &index); avail = audio->highwater_mark - writable; @@ -415,33 +352,34 @@ static ssize_t pipewire_write(void *data, const void *buf_, size_t size) index += size; spa_ringbuffer_write_update(&audio->ring, index); -unlock: pw_thread_loop_unlock(audio->pw->thread_loop); return size; } static bool pipewire_stop(void *data) { - pipewire_device_handle_t *audio = (pipewire_device_handle_t*)data; - if (audio->is_paused) + pipewire_audio_t *audio = (pipewire_audio_t*)data; + const char *error = NULL; + if (pw_stream_get_state(audio->stream, &error) == PW_STREAM_STATE_PAUSED) return true; - return pipewire_set_active(audio->pw, audio, false); + return pipewire_set_active(audio->pw->thread_loop, audio->stream, false); } static bool pipewire_start(void *data, bool is_shutdown) { - pipewire_device_handle_t *audio = (pipewire_device_handle_t*)data; - if (!audio->is_paused) + pipewire_audio_t *audio = (pipewire_audio_t*)data; + const char *error = NULL; + if (pw_stream_get_state(audio->stream, &error) == PW_STREAM_STATE_STREAMING) return true; - return pipewire_set_active(audio->pw, audio, true); + return pipewire_set_active(audio->pw->thread_loop, audio->stream, true); } static bool pipewire_alive(void *data) { - const char *error = NULL; - pipewire_device_handle_t *audio = (pipewire_device_handle_t*)data; + pipewire_audio_t *audio = (pipewire_audio_t*)data; + const char *error = NULL; if (!audio) return false; @@ -450,14 +388,14 @@ static bool pipewire_alive(void *data) static void pipewire_set_nonblock_state(void *data, bool state) { - pipewire_device_handle_t *audio = (pipewire_device_handle_t*)data; + pipewire_audio_t *audio = (pipewire_audio_t*)data; if (audio && audio->pw) audio->pw->nonblock = state; } static void pipewire_free(void *data) { - pipewire_device_handle_t *audio = (pipewire_device_handle_t*)data; + pipewire_audio_t *audio = (pipewire_audio_t*)data; if (!audio) return pw_deinit(); @@ -484,11 +422,14 @@ static void pipewire_free(void *data) pw_core_disconnect(audio->pw->core); } - if (audio->pw->context) - pw_context_destroy(audio->pw->context); + if (audio->pw->ctx) + pw_context_destroy(audio->pw->ctx); pw_thread_loop_destroy(audio->pw->thread_loop); + if (audio->pw->devicelist) + string_list_free(audio->pw->devicelist); + free(audio->pw); free(audio); pw_deinit(); @@ -502,7 +443,7 @@ static bool pipewire_use_float(void *data) static void *pipewire_device_list_new(void *data) { - pipewire_device_handle_t *audio = (pipewire_device_handle_t*)data; + pipewire_audio_t *audio = (pipewire_audio_t*)data; if (audio && audio->pw->devicelist) return string_list_clone(audio->pw->devicelist); @@ -521,31 +462,32 @@ static void pipewire_device_list_free(void *data, void *array_list_data) static size_t pipewire_write_avail(void *data) { uint32_t index, written, length; - pipewire_device_handle_t *audio = (pipewire_device_handle_t*)data; - const char *error = NULL; + pipewire_audio_t *audio = (pipewire_audio_t*)data; + const char *error = NULL; retro_assert(audio->pw); - pw_thread_loop_lock(audio->pw->thread_loop); + retro_assert(audio->stream); if (pw_stream_get_state(audio->stream, &error) != PW_STREAM_STATE_STREAMING) { /* wait for stream to become ready */ length = 0; - goto unlock; + audio_driver_set_buffer_size(audio->highwater_mark); + return length; } + pw_thread_loop_lock(audio->pw->thread_loop); written = spa_ringbuffer_get_write_index(&audio->ring, &index); length = audio->highwater_mark - written; audio_driver_set_buffer_size(audio->highwater_mark); - -unlock: pw_thread_loop_unlock(audio->pw->thread_loop); + return length; } static size_t pipewire_buffer_size(void *data) { - pipewire_device_handle_t *audio = (pipewire_device_handle_t*)data; + pipewire_audio_t *audio = (pipewire_audio_t*)data; return audio->highwater_mark; } diff --git a/audio/drivers_microphone/pipewire.c b/audio/drivers_microphone/pipewire.c index 41e5600ef43..f957ac6741c 100644 --- a/audio/drivers_microphone/pipewire.c +++ b/audio/drivers_microphone/pipewire.c @@ -32,17 +32,29 @@ #include "verbosity.h" -#define APPNAME "RetroArch" -#define DEFAULT_CHANNELS 1 -#define QUANTUM 1024 /* TODO: detect */ +#define DEFAULT_CHANNELS 1 +#define QUANTUM 1024 /* TODO: detect */ +#define RINGBUFFER_SIZE (1u << 22) +#define RINGBUFFER_MASK (RINGBUFFER_SIZE - 1) -typedef pipewire_core_t pipewire_microphone_t; -typedef pipewire_device_handle_t pipewire_microphone_handle_t; +typedef struct pipewire_microphone +{ + pipewire_core_t *pw; + + struct pw_stream *stream; + struct spa_hook stream_listener; + struct spa_audio_info_raw info; + uint32_t frame_size; + struct spa_ringbuffer ring; + uint8_t buffer[RINGBUFFER_SIZE]; + + bool is_ready; +} pipewire_microphone_t; -static void on_stream_state_changed(void *data, +static void stream_state_changed_cb(void *data, enum pw_stream_state old, enum pw_stream_state state, const char *error) { - pipewire_microphone_handle_t *microphone = (pipewire_microphone_handle_t*)data; + pipewire_microphone_t *microphone = (pipewire_microphone_t*)data; RARCH_DBG("[PipeWire]: New state for Source Node %d : %s\n", pw_stream_get_node_id(microphone->stream), @@ -50,13 +62,13 @@ static void on_stream_state_changed(void *data, switch(state) { - case PW_STREAM_STATE_STREAMING: - microphone->is_paused = false; - pw_thread_loop_signal(microphone->pw->thread_loop, false); + case PW_STREAM_STATE_UNCONNECTED: + microphone->is_ready = false; + pw_thread_loop_stop(microphone->pw->thread_loop); break; + case PW_STREAM_STATE_STREAMING: case PW_STREAM_STATE_ERROR: case PW_STREAM_STATE_PAUSED: - microphone->is_paused = true; pw_thread_loop_signal(microphone->pw->thread_loop, false); break; default: @@ -64,16 +76,16 @@ static void on_stream_state_changed(void *data, } } -static void stream_destroy(void *data) +static void stream_destroy_cb(void *data) { - pipewire_microphone_handle_t *microphone = (pipewire_microphone_handle_t*)data; + pipewire_microphone_t *microphone = (pipewire_microphone_t*)data; spa_hook_remove(µphone->stream_listener); microphone->stream = NULL; } -static void on_process(void *data) +static void capture_process_cb(void *data) { - pipewire_microphone_handle_t *microphone = (pipewire_microphone_handle_t *)data; + pipewire_microphone_t *microphone = (pipewire_microphone_t *)data; void *p; struct pw_buffer *b; struct spa_buffer *buf; @@ -102,7 +114,7 @@ static void on_process(void *data) RARCH_ERR("[PipeWire]: %p: underrun write:%u filled:%d\n", p, index, filled); else { - if ((uint32_t) filled + n_bytes > RINGBUFFER_SIZE) + if ((uint32_t)filled + n_bytes > RINGBUFFER_SIZE) RARCH_ERR("[PipeWire]: %p: overrun write:%u filled:%d + size:%u > max:%u\n", p, index, filled, n_bytes, RINGBUFFER_SIZE); } @@ -118,40 +130,9 @@ static void on_process(void *data) static const struct pw_stream_events capture_stream_events = { PW_VERSION_STREAM_EVENTS, - .destroy = stream_destroy, - .state_changed = on_stream_state_changed, - .process = on_process -}; - -static void on_core_error(void *data, uint32_t id, int seq, int res, const char *message) -{ - pipewire_microphone_t *pipewire = data; - - RARCH_ERR("[PipeWire]: error id:%u seq:%d res:%d (%s): %s\n", - id, seq, res, spa_strerror(res), message); - - /* stop and exit the thread loop */ - pw_thread_loop_signal(pipewire->thread_loop, false); -} - -static void on_core_done(void *data, uint32_t id, int seq) -{ - pipewire_microphone_t *pipewire = (pipewire_microphone_t*)data; - - retro_assert(id == PW_ID_CORE); - - pipewire->last_seq = seq; - if (pipewire->pending_seq == seq) - { - /* stop and exit the thread loop */ - pw_thread_loop_signal(pipewire->thread_loop, false); - } -} - -static const struct pw_core_events core_events = { - PW_VERSION_CORE_EVENTS, - .done = on_core_done, - .error = on_core_error, + .destroy = stream_destroy_cb, + .state_changed = stream_state_changed_cb, + .process = capture_process_cb }; static void registry_event_global(void *data, uint32_t id, @@ -159,11 +140,11 @@ static void registry_event_global(void *data, uint32_t id, const struct spa_dict *props) { union string_list_elem_attr attr; - pipewire_microphone_t *pipewire = (pipewire_microphone_t*)data; - const char *media = NULL; - const char *sink = NULL; + pipewire_core_t *pw = (pipewire_core_t*)data; + const char *media = NULL; + const char *sink = NULL; - if (!pipewire) + if (!pw) return; if (spa_streq(type, PW_TYPE_INTERFACE_Node)) @@ -174,7 +155,7 @@ static void registry_event_global(void *data, uint32_t id, if ((sink = spa_dict_lookup(props, PW_KEY_NODE_NAME)) != NULL) { attr.i = id; - string_list_append(pipewire->devicelist, sink, attr); + string_list_append(pw->devicelist, sink, attr); RARCH_LOG("[PipeWire]: Found Source Node: %s\n", sink); } } @@ -190,111 +171,91 @@ static void pipewire_microphone_free(void *driver_context); static void *pipewire_microphone_init(void) { - int res; - uint64_t buf_samples; - const struct spa_pod *params[1]; - uint8_t buffer[1024]; - struct pw_properties *props = NULL; - const char *error = NULL; - pipewire_microphone_t *pipewire = (pipewire_microphone_t*)calloc(1, sizeof(*pipewire)); - struct spa_pod_builder b = SPA_POD_BUILDER_INIT(buffer, sizeof(buffer)); - - if (!pipewire) + int res; + uint64_t buf_samples; + const struct spa_pod *params[1]; + uint8_t buffer[1024]; + struct pw_properties *props = NULL; + const char *error = NULL; + pipewire_core_t *pw = (pipewire_core_t*)calloc(1, sizeof(*pw)); + struct spa_pod_builder b = SPA_POD_BUILDER_INIT(buffer, sizeof(buffer)); + + if (!pw) goto error; pw_init(NULL, NULL); - pipewire->devicelist = string_list_new(); - if (!pipewire->devicelist) - goto error; - - pipewire->thread_loop = pw_thread_loop_new("microphone_driver", NULL); - if (!pipewire->thread_loop) - goto error; - - pipewire->context = pw_context_new(pw_thread_loop_get_loop(pipewire->thread_loop), NULL, 0); - if (!pipewire->context) - goto error; - - if (pw_thread_loop_start(pipewire->thread_loop) < 0) - goto error; - - pw_thread_loop_lock(pipewire->thread_loop); - - pipewire->core = pw_context_connect(pipewire->context, NULL, 0); - if(!pipewire->core) + pw->devicelist = string_list_new(); + if (!pw->devicelist) goto error; - if (pw_core_add_listener(pipewire->core, - &pipewire->core_listener, - &core_events, pipewire) < 0) + if (!pipewire_core_init(pw, "microphone_driver")) goto error; - pipewire->registry = pw_core_get_registry(pipewire->core, PW_VERSION_REGISTRY, 0); + pw->registry = pw_core_get_registry(pw->core, PW_VERSION_REGISTRY, 0); - spa_zero(pipewire->registry_listener); - pw_registry_add_listener(pipewire->registry, &pipewire->registry_listener, ®istry_events, pipewire); + spa_zero(pw->registry_listener); + pw_registry_add_listener(pw->registry, &pw->registry_listener, ®istry_events, pw); - if (pipewire_wait_resync(pipewire) < 0) - pw_thread_loop_unlock(pipewire->thread_loop); + pipewire_wait_resync(pw); + pw_thread_loop_unlock(pw->thread_loop); - - pw_thread_loop_unlock(pipewire->thread_loop); - - return pipewire; + return pw; error: RARCH_ERR("[PipeWire]: Failed to initialize microphone\n"); - pipewire_microphone_free(pipewire); + pipewire_microphone_free(pw); return NULL; } static void pipewire_microphone_close_mic(void *driver_context, void *microphone_context); static void pipewire_microphone_free(void *driver_context) { - pipewire_microphone_t *pipewire = (pipewire_microphone_t*)driver_context; + pipewire_core_t *pw = (pipewire_core_t*)driver_context; - if (!pipewire) + if (!pw) return pw_deinit(); - if (pipewire->thread_loop) - pw_thread_loop_stop(pipewire->thread_loop); + if (pw->thread_loop) + pw_thread_loop_stop(pw->thread_loop); - if (pipewire->client) - pw_proxy_destroy((struct pw_proxy *)pipewire->client); + if (pw->client) + pw_proxy_destroy((struct pw_proxy *)pw->client); - if (pipewire->registry) - pw_proxy_destroy((struct pw_proxy*)pipewire->registry); + if (pw->registry) + pw_proxy_destroy((struct pw_proxy*)pw->registry); - if (pipewire->core) + if (pw->core) { - spa_hook_remove(&pipewire->core_listener); - spa_zero(pipewire->core_listener); - pw_core_disconnect(pipewire->core); + spa_hook_remove(&pw->core_listener); + spa_zero(pw->core_listener); + pw_core_disconnect(pw->core); } - if (pipewire->context) - pw_context_destroy(pipewire->context); + if (pw->ctx) + pw_context_destroy(pw->ctx); + + pw_thread_loop_destroy(pw->thread_loop); - pw_thread_loop_destroy(pipewire->thread_loop); + if (pw->devicelist) + string_list_free(pw->devicelist); - free(pipewire); + free(pw); pw_deinit(); } -static bool pipewire_microphone_start_mic(void *driver_context, void *microphone_context); static int pipewire_microphone_read(void *driver_context, void *microphone_context, void *buf_, size_t size_) { - int32_t readable; - uint32_t index; - const char *error = NULL; - pipewire_microphone_t *pipewire = (pipewire_microphone_t*)driver_context; - pipewire_microphone_handle_t *microphone = (pipewire_microphone_handle_t*)microphone_context; + int32_t readable; + uint32_t index; + const char *error = NULL; + pipewire_core_t *pw = (pipewire_core_t*)driver_context; + pipewire_microphone_t *microphone = (pipewire_microphone_t*)microphone_context; - pw_thread_loop_lock(pipewire->thread_loop); - if (pw_stream_get_state(microphone->stream, &error) != PW_STREAM_STATE_STREAMING) - goto unlock; + if (!microphone->is_ready || pw_stream_get_state(microphone->stream, &error) != PW_STREAM_STATE_STREAMING) + return -1; + pw_thread_loop_lock(pw->thread_loop); /* get no of available bytes to read data from buffer */ readable = spa_ringbuffer_get_read_index(µphone->ring, &index); @@ -306,16 +267,15 @@ static int pipewire_microphone_read(void *driver_context, void *microphone_conte index & RINGBUFFER_MASK, buf_, size_); index += size_; spa_ringbuffer_read_update(µphone->ring, index); + pw_thread_loop_unlock(pw->thread_loop); -unlock: - pw_thread_loop_unlock(pipewire->thread_loop); return size_; } static bool pipewire_microphone_mic_alive(const void *driver_context, const void *microphone_context) { - const char *error = NULL; - pipewire_microphone_handle_t *microphone = (pipewire_microphone_handle_t*)microphone_context; + const char *error = NULL; + pipewire_microphone_t *microphone = (pipewire_microphone_t*)microphone_context; (void)driver_context; if (!microphone) @@ -326,17 +286,17 @@ static bool pipewire_microphone_mic_alive(const void *driver_context, const void static void pipewire_microphone_set_nonblock_state(void *driver_context, bool nonblock) { - pipewire_microphone_t *pipewire = (pipewire_microphone_t*)driver_context; - if (pipewire) - pipewire->nonblock = nonblock; + pipewire_core_t *pw = (pipewire_core_t*)driver_context; + if (pw) + pw->nonblock = nonblock; } static struct string_list *pipewire_microphone_device_list_new(const void *driver_context) { - pipewire_microphone_t *pipewire = (pipewire_microphone_t*)driver_context; + pipewire_core_t *pw = (pipewire_core_t*)driver_context; - if (pipewire && pipewire->devicelist) - return string_list_clone(pipewire->devicelist); + if (pw && pw->devicelist) + return string_list_clone(pw->devicelist); return NULL; } @@ -354,21 +314,21 @@ static void *pipewire_microphone_open_mic(void *driver_context, unsigned latency, unsigned *new_rate) { - int res; - uint64_t buf_samples; - const struct spa_pod *params[1]; - uint8_t buffer[1024]; - struct pw_properties *props = NULL; - const char *error = NULL; - struct spa_pod_builder b = SPA_POD_BUILDER_INIT(buffer, sizeof(buffer)); - pipewire_microphone_handle_t *microphone = calloc(1, sizeof(pipewire_microphone_handle_t)); + int res; + uint64_t buf_samples; + const struct spa_pod *params[1]; + uint8_t buffer[1024]; + struct pw_properties *props = NULL; + const char *error = NULL; + struct spa_pod_builder b = SPA_POD_BUILDER_INIT(buffer, sizeof(buffer)); + pipewire_microphone_t *microphone = calloc(1, sizeof(pipewire_microphone_t)); retro_assert(driver_context); if (!microphone) goto error; - microphone->pw = (pipewire_microphone_t*)driver_context; + microphone->pw = (pipewire_core_t*)driver_context; pw_thread_loop_lock(microphone->pw->thread_loop); @@ -378,15 +338,14 @@ static void *pipewire_microphone_open_mic(void *driver_context, microphone->info.rate = rate; microphone->frame_size = calc_frame_size(microphone->info.format, DEFAULT_CHANNELS); - props = pw_properties_new(PW_KEY_MEDIA_TYPE, "Audio", - PW_KEY_MEDIA_CATEGORY, "Capture", - PW_KEY_MEDIA_ROLE, "Game", - PW_KEY_NODE_NAME, APPNAME, - PW_KEY_NODE_DESCRIPTION, APPNAME, - PW_KEY_APP_NAME, APPNAME, - PW_KEY_APP_ID, APPNAME, - PW_KEY_APP_ICON_NAME, APPNAME, - PW_KEY_NODE_ALWAYS_PROCESS, "true", + props = pw_properties_new(PW_KEY_MEDIA_TYPE, PW_RARCH_MEDIA_TYPE_AUDIO, + PW_KEY_MEDIA_CATEGORY, PW_RARCH_MEDIA_CATEGORY_RECORD, + PW_KEY_MEDIA_ROLE, PW_RARCH_MEDIA_ROLE, + PW_KEY_NODE_NAME, PW_RARCH_APPNAME, + PW_KEY_NODE_DESCRIPTION, PW_RARCH_APPNAME, + PW_KEY_APP_NAME, PW_RARCH_APPNAME, + PW_KEY_APP_ID, PW_RARCH_APPNAME, + PW_KEY_APP_ICON_NAME, PW_RARCH_APPNAME, NULL); if (!props) goto unlock_error; @@ -401,7 +360,7 @@ static void *pipewire_microphone_open_mic(void *driver_context, pw_properties_setf(props, PW_KEY_NODE_RATE, "1/%d", rate); - microphone->stream = pw_stream_new(microphone->pw->core, APPNAME, props); + microphone->stream = pw_stream_new(microphone->pw->core, PW_RARCH_APPNAME, props); if (!microphone->stream) goto unlock_error; @@ -416,7 +375,6 @@ static void *pipewire_microphone_open_mic(void *driver_context, PW_DIRECTION_INPUT, PW_ID_ANY, PW_STREAM_FLAG_AUTOCONNECT | - PW_STREAM_FLAG_INACTIVE | PW_STREAM_FLAG_MAP_BUFFERS | PW_STREAM_FLAG_RT_PROCESS, params, 1); @@ -425,11 +383,10 @@ static void *pipewire_microphone_open_mic(void *driver_context, goto unlock_error; pw_thread_loop_wait(microphone->pw->thread_loop); - if(pw_stream_get_state(microphone->stream, &error) != PW_STREAM_STATE_STREAMING) - microphone->is_paused = true; pw_thread_loop_unlock(microphone->pw->thread_loop); *new_rate = microphone->info.rate; + microphone->is_ready = true; return microphone; @@ -443,37 +400,45 @@ static void *pipewire_microphone_open_mic(void *driver_context, static void pipewire_microphone_close_mic(void *driver_context, void *microphone_context) { - pipewire_microphone_t *pipewire = (pipewire_microphone_t*)driver_context; - pipewire_microphone_handle_t *microphone = (pipewire_microphone_handle_t*)microphone_context; + pipewire_core_t *pw = (pipewire_core_t*)driver_context; + pipewire_microphone_t *microphone = (pipewire_microphone_t*)microphone_context; - if (pipewire && microphone) + if (pw && microphone) { - pw_thread_loop_lock(pipewire->thread_loop); + pw_thread_loop_lock(pw->thread_loop); pw_stream_destroy(microphone->stream); microphone->stream = NULL; - pw_thread_loop_unlock(pipewire->thread_loop); + pw_thread_loop_unlock(pw->thread_loop); + free(microphone); } } static bool pipewire_microphone_start_mic(void *driver_context, void *microphone_context) { - pipewire_microphone_t *pipewire = (pipewire_microphone_t*)driver_context; - pipewire_microphone_handle_t *microphone = (pipewire_microphone_handle_t*)microphone_context; + pipewire_core_t *pw = (pipewire_core_t*)driver_context; + pipewire_microphone_t *microphone = (pipewire_microphone_t*)microphone_context; + const char *error = NULL; - if (!microphone->is_paused) + if (!microphone->is_ready) + return false; + if (pw_stream_get_state(microphone->stream, &error) == PW_STREAM_STATE_STREAMING) return true; - return pipewire_set_active(pipewire, microphone, true); + return pipewire_set_active(pw->thread_loop, microphone->stream, true); } static bool pipewire_microphone_stop_mic(void *driver_context, void *microphone_context) { - pipewire_microphone_t *pipewire = (pipewire_microphone_t*)driver_context; - pipewire_microphone_handle_t *microphone = (pipewire_microphone_handle_t*)microphone_context; - if (microphone->is_paused) + pipewire_core_t *pw = (pipewire_core_t*)driver_context; + pipewire_microphone_t *microphone = (pipewire_microphone_t*)microphone_context; + const char *error = NULL; + + if (!microphone->is_ready) + return false; + if (pw_stream_get_state(microphone->stream, &error) == PW_STREAM_STATE_PAUSED) return true; - return pipewire_set_active(pipewire, microphone, false); + return pipewire_set_active(pw->thread_loop, microphone->stream, false); } static bool pipewire_microphone_mic_use_float(const void *driver_context, const void *microphone_context)