From 011eb76883509e744f17917e3d8cc9c76ccad1ec Mon Sep 17 00:00:00 2001 From: Olivier Gauthier Date: Thu, 9 Jul 2026 09:09:39 -0400 Subject: [PATCH 1/4] pipewire: fix crash from iterating the thread-loop from two threads In thread-loop mode context::do_sync_locked() called pw_loop_iterate() on the shared pw_loop from the caller while the pw_thread_loop worker iterates the same loop. Two threads iterating one pw_loop corrupt the per-source dispatch bookkeeping (s->priv/s->rmask, stack-local ep[]) and double-fire socket callbacks, crashing in on_remote_data -> pw_protocol_native_connection_flush on a freed/NULL connection (segfault at process exit, during audio filter teardown which calls synchronize() amid protocol traffic). Use the idiomatic pattern instead: for the thread loop, issue pw_core_sync and block on pw_thread_loop_timed_wait, letting the single worker drive the loop; on_core_done/on_core_error wake the waiter via pw_thread_loop_signal. The main-loop path (no worker) keeps driving iteration itself. Require pw_thread_loop_timed_wait in the thread_available gate. Co-Authored-By: Claude Opus 4.8 --- .../backends/linux/pipewire/context.hpp | 68 +++++++++++++++++-- .../backends/linux/pipewire/loader.hpp | 2 +- 2 files changed, 65 insertions(+), 5 deletions(-) diff --git a/include/libremidi/backends/linux/pipewire/context.hpp b/include/libremidi/backends/linux/pipewire/context.hpp index 51382683..a7c04728 100644 --- a/include/libremidi/backends/linux/pipewire/context.hpp +++ b/include/libremidi/backends/linux/pipewire/context.hpp @@ -206,11 +206,11 @@ class context : public std::enable_shared_from_this pw_thread_loop* loop; ~unlock_guard() { pw.thread_loop_unlock(loop); } } guard{pw, m_thread_loop}; - return do_sync_locked(t_end); + return do_sync_thread_locked(t_end); } else { - return do_sync_locked(t_end); + return do_sync_iterate(t_end); } } @@ -788,7 +788,10 @@ class context : public std::enable_shared_from_this { auto* self = static_cast(data); if (id == PW_ID_CORE && seq == self->m_sync_seq.load(std::memory_order_acquire)) + { self->m_sync_done.store(1, std::memory_order_release); + self->wake_sync_waiter(); + } } static void on_core_error( @@ -800,6 +803,18 @@ class context : public std::enable_shared_from_this { self->m_state.store(connection_state::broken, std::memory_order_release); } + self->wake_sync_waiter(); + } + + // Wakes a thread-loop synchronize() blocked in pw_thread_loop_timed_wait. + // Runs from the worker thread (holds the loop lock) during event dispatch. + void wake_sync_waiter() noexcept + { + if (m_cfg.kind != loop_kind::thread || !m_thread_loop) + return; + auto& pw = load(); + if (pw.thread_loop_signal) + pw.thread_loop_signal(m_thread_loop, false); } void install_core_listener() noexcept @@ -822,8 +837,49 @@ class context : public std::enable_shared_from_this pw_core_add_listener(m_core, &m_core_listener, &events, this); } - // Lock must be held in thread-loop mode (drives pw_loop_iterate). - bool do_sync_locked(std::chrono::steady_clock::time_point deadline) + // Thread-loop mode: the worker thread owns iteration. We must NOT call + // pw_loop_iterate ourselves — two threads iterating one pw_loop corrupt + // its per-source dispatch bookkeeping (s->priv / s->rmask) and double-fire + // socket callbacks, crashing in on_remote_data on a freed connection. + // Issue the sync and block on the loop's condition variable instead; the + // worker drives the loop and wakes us via wake_sync_waiter(). + bool do_sync_thread_locked(std::chrono::steady_clock::time_point deadline) + { + auto& pw = load(); + m_sync_done.store(0, std::memory_order_release); + m_sync_error.store(0, std::memory_order_release); + + int seq = pw_core_sync(m_core, PW_ID_CORE, 0); + m_sync_seq.store(seq, std::memory_order_release); + + using clk = std::chrono::steady_clock; + while (m_sync_done.load(std::memory_order_acquire) == 0 + && m_sync_error.load(std::memory_order_acquire) == 0) + { + const auto now = clk::now(); + if (now >= deadline) + { + m_state.store(connection_state::broken, std::memory_order_release); + return false; + } + // pw_thread_loop_timed_wait has whole-second granularity; the deadline + // check above bounds the total wait. It releases the lock while waiting + // (so the worker can iterate) and re-takes it before returning. + const auto secs + = std::chrono::duration_cast(deadline - now).count(); + const int wait_sec = static_cast(secs < 1 ? 1 : secs + 1); + const int r = pw.thread_loop_timed_wait(m_thread_loop, wait_sec); + if (r < 0 && r != -ETIMEDOUT) + { + m_state.store(connection_state::broken, std::memory_order_release); + return false; + } + } + return finalize_sync(); + } + + // Main-loop mode: no worker thread exists, so the caller owns iteration. + bool do_sync_iterate(std::chrono::steady_clock::time_point deadline) { m_sync_done.store(0, std::memory_order_release); m_sync_error.store(0, std::memory_order_release); @@ -851,7 +907,11 @@ class context : public std::enable_shared_from_this return false; } } + return finalize_sync(); + } + bool finalize_sync() noexcept + { if (m_sync_error.load(std::memory_order_acquire) != 0) { m_state.store(connection_state::broken, std::memory_order_release); diff --git a/include/libremidi/backends/linux/pipewire/loader.hpp b/include/libremidi/backends/linux/pipewire/loader.hpp index 5659187e..0549c9ee 100644 --- a/include/libremidi/backends/linux/pipewire/loader.hpp +++ b/include/libremidi/backends/linux/pipewire/loader.hpp @@ -203,7 +203,7 @@ class api thread_available = thread_loop_new && thread_loop_destroy && thread_loop_start && thread_loop_stop && thread_loop_get_loop && thread_loop_lock && thread_loop_unlock && thread_loop_signal && thread_loop_wait - && thread_loop_in_thread; + && thread_loop_timed_wait && thread_loop_in_thread; stream_new = sym(h, "pw_stream_new"); stream_new_simple = sym(h, "pw_stream_new_simple"); From d72d8b1cd87f400e6a6444bc80b88fb09948696a Mon Sep 17 00:00:00 2001 From: Olivier Gauthier Date: Thu, 9 Jul 2026 09:15:43 -0400 Subject: [PATCH 2/4] pipewire: run context::reconnect() off the loop thread reconnect() wrapped tear_down()+build_connection() in invoke_sync(), running them on the pw_thread_loop worker. But tear_down() stops and destroys that same loop: pthread_join(self) returns EDEADLK (silent no-op) and pw_thread_loop_destroy() then unlocks a loop it is running inside, tripping 'this->recurse > 0' in do_unlock() and hanging the blocking invoke. score's PipeWireAudioFactory::make_engine() calls reconnect() directly when the shared context is broken, so this fires even with auto_reconnect off. Run the tear-down/rebuild/sync directly on the calling thread instead, and bail out if called from the loop thread. tear_down() then joins the worker from a foreign thread (correct) before recreating the loop and core. Co-Authored-By: Claude Opus 4.8 --- .../backends/linux/pipewire/context.hpp | 21 +++++++++++++------ 1 file changed, 15 insertions(+), 6 deletions(-) diff --git a/include/libremidi/backends/linux/pipewire/context.hpp b/include/libremidi/backends/linux/pipewire/context.hpp index a7c04728..e8f332ff 100644 --- a/include/libremidi/backends/linux/pipewire/context.hpp +++ b/include/libremidi/backends/linux/pipewire/context.hpp @@ -216,12 +216,21 @@ class context : public std::enable_shared_from_this bool reconnect() { - invoke_sync([this] { - tear_down(/*final=*/false); - m_state.store(connection_state::connecting, std::memory_order_release); - if (!build_connection()) - m_state.store(connection_state::broken, std::memory_order_release); - }); + // tear_down() stops and destroys the thread loop, so reconnect MUST run + // on a thread other than the worker — never via invoke_sync. Doing it on + // the loop thread self-destructs the loop: pthread_join(self) is a silent + // no-op and pw_thread_loop_destroy then corrupts the lock it runs under + // ('recurse > 0' failure) and the invoke never completes (hang). + if (is_in_loop_thread()) + return false; + + tear_down(/*final=*/false); + m_state.store(connection_state::connecting, std::memory_order_release); + if (!build_connection()) + { + m_state.store(connection_state::broken, std::memory_order_release); + return false; + } if (!synchronize()) return false; m_state.store(connection_state::connected, std::memory_order_release); From 7d7aaee4eccfe57d2bc087629ceb3cabb8608db9 Mon Sep 17 00:00:00 2001 From: Olivier Gauthier Date: Thu, 9 Jul 2026 10:10:59 -0400 Subject: [PATCH 3/4] pipewire: fix thread-loop lock misuse in invoke_sync and synchronize Two related recursive-lock bugs surfaced by using a MIDI device and audio together on the shared context: 1. invoke_sync() (used by unsubscribe) called pw_loop_invoke(block=true) WITHOUT holding the thread-loop lock. pipewire's blocking invoke wraps its wait in the loop control hooks (do_unlock before / do_lock after), so it requires the caller to hold the lock exactly once. Unlocked, do_unlock underflows the recurse counter ('recurse > 0' failed at thread-loop.c:62) and the stray do_lock leaves the recursive mutex owned by the wrong thread, deadlocking the next pw_thread_loop_stop() join. Take the lock around the blocking invoke. 2. synchronize() held pipewire's recursive loop lock across pw_thread_loop_timed_wait(); pthread_cond_wait cannot fully release a recursively-held mutex, so nested/concurrent syncs starved the worker and corrupted the lock. Replace with a private std::condition_variable: take the loop lock only briefly to issue pw_core_sync, then wait off the loop lock; on_core_done/on_core_error notify it. Also guard is_in_loop_thread() (drive iteration directly on the worker instead of waiting on ourselves). pw_thread_loop_timed_wait is no longer used; drop it from the thread gate. Co-Authored-By: Claude Opus 4.8 --- .../backends/linux/pipewire/context.hpp | 94 ++++++++++--------- .../backends/linux/pipewire/loader.hpp | 2 +- 2 files changed, 53 insertions(+), 43 deletions(-) diff --git a/include/libremidi/backends/linux/pipewire/context.hpp b/include/libremidi/backends/linux/pipewire/context.hpp index e8f332ff..ceb3385a 100644 --- a/include/libremidi/backends/linux/pipewire/context.hpp +++ b/include/libremidi/backends/linux/pipewire/context.hpp @@ -197,19 +197,16 @@ class context : public std::enable_shared_from_this using clk = std::chrono::steady_clock; const auto t_end = clk::now() + deadline; - if (m_cfg.kind == loop_kind::thread) + if (m_cfg.kind == loop_kind::thread && !is_in_loop_thread()) { - pw.thread_loop_lock(m_thread_loop); - struct unlock_guard - { - const api& pw; - pw_thread_loop* loop; - ~unlock_guard() { pw.thread_loop_unlock(loop); } - } guard{pw, m_thread_loop}; - return do_sync_thread_locked(t_end); + return do_sync_thread(t_end); } else { + // Main-loop mode, or already on the thread-loop worker: we own the loop + // here, so drive iteration directly. Blocking from the worker would + // starve the loop, and re-locking pipewire's recursive mutex across a + // condition wait corrupts its accounting ('recurse > 0' in do_unlock). return do_sync_iterate(t_end); } } @@ -303,6 +300,19 @@ class context : public std::enable_shared_from_this { return std::forward(fn)(); } + // pw_loop_invoke(block=true) releases and re-takes the thread-loop lock + // around its wait (via the loop control hooks), so it MUST run with the + // lock held exactly once. Calling it unlocked underflows the recurse + // counter ('recurse > 0' in do_unlock) and leaves the mutex owned by the + // wrong thread, deadlocking a later pw_thread_loop_stop. + auto& pw = load(); + pw.thread_loop_lock(m_thread_loop); + struct unlock_guard + { + const api& pw; + pw_thread_loop* loop; + ~unlock_guard() { pw.thread_loop_unlock(loop); } + } guard{pw, m_thread_loop}; if constexpr (std::is_void_v) { FR f{std::forward(fn)}; @@ -441,6 +451,11 @@ class context : public std::enable_shared_from_this std::atomic m_sync_done{0}; std::atomic m_sync_error{0}; + // do_sync_thread() waits here; wake_sync_waiter() (worker thread) notifies. + // A private CV keeps application waiting off pipewire's recursive loop lock. + std::mutex m_sync_mtx; + std::condition_variable m_sync_cv; + // FIXME lock-free? mutable std::mutex m_graph_mtx; @@ -815,15 +830,16 @@ class context : public std::enable_shared_from_this self->wake_sync_waiter(); } - // Wakes a thread-loop synchronize() blocked in pw_thread_loop_timed_wait. - // Runs from the worker thread (holds the loop lock) during event dispatch. + // Wakes do_sync_thread(), which waits on m_sync_cv. Runs from the worker + // thread during dispatch; the brief m_sync_mtx lock orders the flag store + // with the waiter without nesting an application wait inside pipewire's + // recursive loop lock. void wake_sync_waiter() noexcept { - if (m_cfg.kind != loop_kind::thread || !m_thread_loop) - return; - auto& pw = load(); - if (pw.thread_loop_signal) - pw.thread_loop_signal(m_thread_loop, false); + { + std::lock_guard lk{m_sync_mtx}; + } + m_sync_cv.notify_all(); } void install_core_listener() noexcept @@ -846,39 +862,33 @@ class context : public std::enable_shared_from_this pw_core_add_listener(m_core, &m_core_listener, &events, this); } - // Thread-loop mode: the worker thread owns iteration. We must NOT call - // pw_loop_iterate ourselves — two threads iterating one pw_loop corrupt - // its per-source dispatch bookkeeping (s->priv / s->rmask) and double-fire - // socket callbacks, crashing in on_remote_data on a freed connection. - // Issue the sync and block on the loop's condition variable instead; the - // worker drives the loop and wakes us via wake_sync_waiter(). - bool do_sync_thread_locked(std::chrono::steady_clock::time_point deadline) + // Thread-loop sync from a foreign thread: the worker owns iteration, so take + // the loop lock only briefly to issue the sync, then wait on a private + // condition variable. We must neither iterate the loop ourselves (two + // threads iterating one pw_loop corrupt its per-source dispatch state) nor + // hold pipewire's recursive loop lock across the wait (pthread_cond_wait + // cannot fully release a recursively-held mutex, so it starves the worker + // and corrupts the lock: 'recurse > 0' in do_unlock). + bool do_sync_thread(std::chrono::steady_clock::time_point deadline) { auto& pw = load(); m_sync_done.store(0, std::memory_order_release); m_sync_error.store(0, std::memory_order_release); - int seq = pw_core_sync(m_core, PW_ID_CORE, 0); - m_sync_seq.store(seq, std::memory_order_release); + { + pw.thread_loop_lock(m_thread_loop); + int seq = pw_core_sync(m_core, PW_ID_CORE, 0); + m_sync_seq.store(seq, std::memory_order_release); + pw.thread_loop_unlock(m_thread_loop); + } - using clk = std::chrono::steady_clock; - while (m_sync_done.load(std::memory_order_acquire) == 0 - && m_sync_error.load(std::memory_order_acquire) == 0) { - const auto now = clk::now(); - if (now >= deadline) - { - m_state.store(connection_state::broken, std::memory_order_release); - return false; - } - // pw_thread_loop_timed_wait has whole-second granularity; the deadline - // check above bounds the total wait. It releases the lock while waiting - // (so the worker can iterate) and re-takes it before returning. - const auto secs - = std::chrono::duration_cast(deadline - now).count(); - const int wait_sec = static_cast(secs < 1 ? 1 : secs + 1); - const int r = pw.thread_loop_timed_wait(m_thread_loop, wait_sec); - if (r < 0 && r != -ETIMEDOUT) + std::unique_lock lk{m_sync_mtx}; + const bool ready = m_sync_cv.wait_until(lk, deadline, [this] { + return m_sync_done.load(std::memory_order_acquire) != 0 + || m_sync_error.load(std::memory_order_acquire) != 0; + }); + if (!ready) { m_state.store(connection_state::broken, std::memory_order_release); return false; diff --git a/include/libremidi/backends/linux/pipewire/loader.hpp b/include/libremidi/backends/linux/pipewire/loader.hpp index 0549c9ee..5659187e 100644 --- a/include/libremidi/backends/linux/pipewire/loader.hpp +++ b/include/libremidi/backends/linux/pipewire/loader.hpp @@ -203,7 +203,7 @@ class api thread_available = thread_loop_new && thread_loop_destroy && thread_loop_start && thread_loop_stop && thread_loop_get_loop && thread_loop_lock && thread_loop_unlock && thread_loop_signal && thread_loop_wait - && thread_loop_timed_wait && thread_loop_in_thread; + && thread_loop_in_thread; stream_new = sym(h, "pw_stream_new"); stream_new_simple = sym(h, "pw_stream_new_simple"); From ab9669f36dc36d4a0fb5622294b4a143a0d99990 Mon Sep 17 00:00:00 2001 From: Olivier Gauthier Date: Thu, 9 Jul 2026 14:04:44 -0400 Subject: [PATCH 4/4] examples: add pipewire shared-context regression tests Three standalone programs under examples/backends/ that exercise the shared thread-loop context fixes: - pipewire_context_sync: many create/synchronize/destroy cycles plus a sustained synchronize() loop (foreign-thread sync must not iterate the loop concurrently with the worker). - pipewire_context_reconnect: repeated reconnect() from the calling thread (must not self-destruct the loop from the worker thread). - pipewire_context_subscriptions: subscribe/unsubscribe churn, MIDI-style filter teardown with live subscriptions, and synchronize() from a subscriber callback (recursive thread-loop lock must stay balanced). Each skips (exit 0) when no PipeWire daemon is reachable and arms a watchdog so a lock-corruption regression fails loudly instead of hanging. Built against the pre-fix headers, pipewire_context_reconnect and pipewire_context_subscriptions reproduce 'recurse > 0' + SIGSEGV. Co-Authored-By: Claude Opus 4.8 --- cmake/libremidi.examples.cmake | 3 + .../backends/pipewire_context_reconnect.cpp | 99 +++++++++++ .../pipewire_context_subscriptions.cpp | 158 ++++++++++++++++++ examples/backends/pipewire_context_sync.cpp | 107 ++++++++++++ 4 files changed, 367 insertions(+) create mode 100644 examples/backends/pipewire_context_reconnect.cpp create mode 100644 examples/backends/pipewire_context_subscriptions.cpp create mode 100644 examples/backends/pipewire_context_sync.cpp diff --git a/cmake/libremidi.examples.cmake b/cmake/libremidi.examples.cmake index 0eb53f50..a36fbf34 100644 --- a/cmake/libremidi.examples.cmake +++ b/cmake/libremidi.examples.cmake @@ -82,6 +82,9 @@ if(LIBREMIDI_HAS_PIPEWIRE) add_backend_example(midi1_in_pipewire) add_backend_example(midi1_out_pipewire) add_backend_example(shared_pipewire_context) + add_backend_example(pipewire_context_sync) + add_backend_example(pipewire_context_reconnect) + add_backend_example(pipewire_context_subscriptions) if(LIBREMIDI_HAS_PIPEWIRE_UMP) add_backend_example(midi2_in_pipewire) diff --git a/examples/backends/pipewire_context_reconnect.cpp b/examples/backends/pipewire_context_reconnect.cpp new file mode 100644 index 00000000..5c8ad18c --- /dev/null +++ b/examples/backends/pipewire_context_reconnect.cpp @@ -0,0 +1,99 @@ +// SPDX-License-Identifier: BSL-1.0 +// +// Regression test for context::reconnect() on the shared PipeWire thread-loop. +// +// reconnect() used to run tear_down() + build_connection() through invoke_sync(), +// i.e. on the worker thread. But tear_down() stops and destroys that very loop: +// pthread_join(self) is a silent no-op and pw_thread_loop_destroy() then unlocks +// a loop it is running inside, tripping 'this->recurse > 0' in do_unlock() and +// hanging the blocking invoke. Callers (e.g. an audio host recovering a broken +// context) hit this directly. +// +// reconnect() now runs on the calling thread. This calls it repeatedly and must +// return promptly each time without hanging or corrupting the loop lock. +// +// Requires a running PipeWire daemon; skips (exit 0) if none is reachable. + +#include +#include +#include + +#include +#include +#include +#include + +namespace lpw = libremidi::pipewire; + +static void arm_watchdog(int seconds) +{ + std::thread( + [seconds] + { + std::this_thread::sleep_for(std::chrono::seconds(seconds)); + std::fprintf(stderr, "FAIL: watchdog timeout (%ds) - likely deadlock\n", seconds); + std::fflush(stderr); + std::_Exit(EXIT_FAILURE); + }) + .detach(); +} + +int main() +{ + auto& pw = lpw::load(); + if (!pw.thread_available) + { + std::printf("libpipewire thread-loop not available; skipping\n"); + return 0; + } + + auto inst = lpw::shared_instance(); + if (!inst) + { + std::printf("pw_init failed; skipping\n"); + return 0; + } + + auto ctx = lpw::context::make(inst); + if (!ctx || !ctx->ok()) + { + std::printf("cannot connect to pipewire daemon; skipping\n"); + return 0; + } + + arm_watchdog(60); + + // State subscriptions survive a reconnect and must still be intact after; each + // reconnect tears the loop down and rebuilds it from the calling thread. + auto sub_state = ctx->on_state_changed([](lpw::connection_state) {}); + + const std::uint32_t gen0 = ctx->generation(); + for (int i = 0; i < 8; ++i) + { + if (!ctx->reconnect()) + { + std::fprintf(stderr, "FAIL: reconnect() returned false while daemon is up\n"); + return EXIT_FAILURE; + } + if (!ctx->ok()) + { + std::fprintf(stderr, "FAIL: context not connected after reconnect()\n"); + return EXIT_FAILURE; + } + // Prove the loop lock still works after the rebuild. + if (!ctx->synchronize()) + { + std::fprintf(stderr, "FAIL: synchronize() failed after reconnect()\n"); + return EXIT_FAILURE; + } + } + + if (ctx->generation() == gen0) + { + std::fprintf(stderr, "FAIL: generation did not advance across reconnects\n"); + return EXIT_FAILURE; + } + + std::printf("PASS: pipewire_context_reconnect (generation %u -> %u)\n", gen0, ctx->generation()); + return 0; +} diff --git a/examples/backends/pipewire_context_subscriptions.cpp b/examples/backends/pipewire_context_subscriptions.cpp new file mode 100644 index 00000000..46d8811b --- /dev/null +++ b/examples/backends/pipewire_context_subscriptions.cpp @@ -0,0 +1,158 @@ +// SPDX-License-Identifier: BSL-1.0 +// +// Regression test for the recursive thread-loop lock in the shared PipeWire +// context. PipeWire's thread-loop mutex is PTHREAD_MUTEX_RECURSIVE, which made +// two paths corrupt it ('this->recurse > 0' failed at thread-loop.c do_unlock, +// then a hung pw_thread_loop_stop): +// +// * unsubscribe() ran pw_loop_invoke(block=true) without holding the loop +// lock. The blocking invoke wraps its wait in the loop control hooks +// (do_unlock before / do_lock after), so it must be called with the lock +// held once; unlocked, do_unlock underflows the recurse counter. +// * synchronize() held the recursive lock across pw_thread_loop_timed_wait(); +// pthread_cond_wait cannot fully release a recursively-held mutex, and had +// no is_in_loop_thread() guard. +// +// This drops many subscriptions, tears down a MIDI-style filter while +// subscriptions are live, and calls synchronize() from a subscriber callback +// (i.e. on the loop thread). It must complete without crashing or hanging. +// +// Requires a running PipeWire daemon; skips (exit 0) if none is reachable. + +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include + +namespace lpw = libremidi::pipewire; + +static void arm_watchdog(int seconds) +{ + std::thread( + [seconds] + { + std::this_thread::sleep_for(std::chrono::seconds(seconds)); + std::fprintf(stderr, "FAIL: watchdog timeout (%ds) - likely deadlock\n", seconds); + std::fflush(stderr); + std::_Exit(EXIT_FAILURE); + }) + .detach(); +} + +int main() +{ + auto& pw = lpw::load(); + if (!pw.thread_available || !pw.filter_available) + { + std::printf("libpipewire thread-loop/filter not available; skipping\n"); + return 0; + } + + auto inst = lpw::shared_instance(); + if (!inst) + { + std::printf("pw_init failed; skipping\n"); + return 0; + } + + auto ctx = lpw::context::make(inst); + if (!ctx || !ctx->ok()) + { + std::printf("cannot connect to pipewire daemon; skipping\n"); + return 0; + } + + arm_watchdog(60); + + // 1. Subscribe / unsubscribe churn. Each subscription reset runs + // unsubscribe() -> invoke_sync() -> pw_loop_invoke(block=true). + for (int i = 0; i < 50; ++i) + { + auto s1 = ctx->on_port_added([](const lpw::port_info&) {}); + auto s2 = ctx->on_port_removed([](std::uint32_t) {}); + auto s3 = ctx->on_node_added([](const lpw::node_info&) {}); + auto s4 = ctx->on_node_removed([](std::uint32_t) {}); + // Destroyed here (reverse order) -> unsubscribe on each. + } + + // 2. A MIDI-style filter with a real port, torn down while port/node + // subscriptions are live. filter::stop() does disconnect + synchronize + + // destroy under the loop lock. + { + auto sub_add = ctx->on_port_added([](const lpw::port_info&) {}); + auto sub_rm = ctx->on_port_removed([](std::uint32_t) {}); + + static constexpr pw_filter_events events = { + .version = PW_VERSION_FILTER_EVENTS, + .destroy = nullptr, + .state_changed = nullptr, + .io_changed = nullptr, + .param_changed = nullptr, + .add_buffer = nullptr, + .remove_buffer = nullptr, + .process = nullptr, + .drained = nullptr, +#if PW_VERSION_FILTER_EVENTS >= 1 + .command = nullptr, +#endif + }; + lpw::filter_config cfg; + cfg.name = "libremidi-regression"; + cfg.media_type = "Midi"; + cfg.media_category = "Filter"; + cfg.media_role = "DSP"; + cfg.format_dsp = "8 bit raw midi"; + cfg.always_process = true; + cfg.pause_on_idle = false; + cfg.suspend_on_idle = false; + cfg.rt_process = true; + + auto flt = std::make_unique(ctx, cfg, events, nullptr); + if (flt->ok()) + { + flt->start(); + auto port = flt->add_port(SPA_DIRECTION_INPUT, "in", "8 bit raw midi"); + if (!port.valid()) + { + std::fprintf(stderr, "FAIL: could not add filter port\n"); + return EXIT_FAILURE; + } + flt->synchronize_node(); + } + flt.reset(); // filter::stop() + } + + // 3. synchronize() from a subscriber callback, i.e. on the loop thread. The + // is_in_loop_thread() guard must keep this from waiting on itself / + // re-locking the recursive mutex. Fire it via a reconnect (re-walks the + // registry, so node_added dispatches on the worker thread). + { + std::atomic loop_thread_syncs{0}; + auto sub = ctx->on_node_added( + [&](const lpw::node_info&) + { + if (loop_thread_syncs.fetch_add(1) == 0) + (void)ctx->synchronize(); + }); + + if (!ctx->reconnect() || !ctx->ok()) + { + std::fprintf(stderr, "FAIL: reconnect() failed\n"); + return EXIT_FAILURE; + } + for (int i = 0; i < 5; ++i) + (void)ctx->synchronize(); + std::printf("synchronize() from loop thread fired %d time(s)\n", loop_thread_syncs.load()); + } + + std::printf("PASS: pipewire_context_subscriptions\n"); + return 0; +} diff --git a/examples/backends/pipewire_context_sync.cpp b/examples/backends/pipewire_context_sync.cpp new file mode 100644 index 00000000..e71af687 --- /dev/null +++ b/examples/backends/pipewire_context_sync.cpp @@ -0,0 +1,107 @@ +// SPDX-License-Identifier: BSL-1.0 +// +// Regression test for the shared PipeWire thread-loop context. +// +// synchronize() used to drive the loop with pw_loop_iterate() from the calling +// thread while the pw_thread_loop worker iterated the same pw_loop. Two threads +// iterating one loop corrupt its per-source dispatch bookkeeping and double-fire +// the core-socket callback, ending in pw_protocol_native_connection_flush() on a +// freed/NULL connection (a segfault, usually seen at process exit when teardown +// floods the socket). +// +// This exercises many create / synchronize / destroy cycles plus a tight +// synchronize() loop. It must run to completion without crashing. +// +// Requires a running PipeWire daemon; skips (exit 0) if none is reachable. + +#include +#include +#include + +#include +#include +#include +#include + +namespace lpw = libremidi::pipewire; + +// A deadlock must fail the test rather than hang a CI run. +static void arm_watchdog(int seconds) +{ + std::thread( + [seconds] + { + std::this_thread::sleep_for(std::chrono::seconds(seconds)); + std::fprintf(stderr, "FAIL: watchdog timeout (%ds) - likely deadlock\n", seconds); + std::fflush(stderr); + std::_Exit(EXIT_FAILURE); + }) + .detach(); +} + +int main() +{ + auto& pw = lpw::load(); + if (!pw.thread_available) + { + std::printf("libpipewire thread-loop not available; skipping\n"); + return 0; + } + + auto inst = lpw::shared_instance(); + if (!inst) + { + std::printf("pw_init failed; skipping\n"); + return 0; + } + + arm_watchdog(60); + + // Pass 1: repeated create / synchronize / destroy. The destructor stops the + // worker and disconnects the core; concurrent iteration during that teardown + // was the original crash. + for (int i = 0; i < 20; ++i) + { + auto ctx = lpw::context::make(inst); + if (!ctx || !ctx->ok()) + { + std::printf("cannot connect to pipewire daemon; skipping\n"); + return 0; + } + + for (int k = 0; k < 20; ++k) + { + if (!ctx->synchronize()) + { + std::fprintf(stderr, "FAIL: synchronize() failed while connected\n"); + return EXIT_FAILURE; + } + } + } + + // Pass 2: sustained synchronize() round-trips on a single long-lived context. + { + auto ctx = lpw::context::make(inst); + if (!ctx || !ctx->ok()) + { + std::printf("cannot connect to pipewire daemon; skipping\n"); + return 0; + } + + const auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(3); + long count = 0; + while (std::chrono::steady_clock::now() < deadline) + { + if (!ctx->synchronize()) + { + std::fprintf(stderr, "FAIL: synchronize() failed while connected\n"); + return EXIT_FAILURE; + } + ++count; + } + std::printf("completed %ld synchronize() round-trips\n", count); + } + + std::printf("PASS: pipewire_context_sync\n"); + return 0; +}