pipewire: fix thread-loop lifetime & recursive-lock crashes in the shared context

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)

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 <libremidi/backends/linux/pipewire/context.hpp>
+#include <libremidi/backends/linux/pipewire/instance.hpp>
+#include <libremidi/backends/linux/pipewire/loader.hpp>
+
+#include <chrono>
+#include <cstdio>
+#include <cstdlib>
+#include <thread>
+
+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;
+}

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 <libremidi/backends/linux/pipewire/context.hpp>
+#include <libremidi/backends/linux/pipewire/filter.hpp>
+#include <libremidi/backends/linux/pipewire/instance.hpp>
+#include <libremidi/backends/linux/pipewire/loader.hpp>
+#include <libremidi/backends/linux/pipewire/types.hpp>
+
+#include <atomic>
+#include <chrono>
+#include <cstdio>
+#include <cstdlib>
+#include <memory>
+#include <thread>
+
+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<lpw::filter>(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<int> 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;
+}

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 <libremidi/backends/linux/pipewire/context.hpp>
+#include <libremidi/backends/linux/pipewire/instance.hpp>
+#include <libremidi/backends/linux/pipewire/loader.hpp>
+
+#include <chrono>
+#include <cstdio>
+#include <cstdlib>
+#include <thread>
+
+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;
+}