YangWX/crossdesk
1
0
Fork 0
mirror of https://github.com/kunkundi/crossdesk.git synced 2026-10-25 14:47:00 +08:00
Code Issues Packages Projects Releases Wiki Activity

[fix] prevent system audio capture hangs across desktop platforms

Browse Source
This commit is contained in:
dijunkun
2026-09-09 11:34:47 +08:00
parent b87ab025be
commit bd0a7aa2f6
16 changed files with 743 additions and 575 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/desktop/src/platform/linux/speaker_capturer/speaker_capturer_linux.cpp
+187 -216
View File
@@ -3,269 +3,240 @@
#include <pulse/error.h>
#include <pulse/introspect.h>
#include <condition_variable>
#include <iostream>
#include <thread>
#include <string>
#include <utility>
#include "rd_log.h"
namespace crossdesk {
constexpr int kSampleRate = 48000;
constexpr pa_sample_format_t kFormat = PA_SAMPLE_S16LE;
constexpr int kChannels = 1;
namespace {
constexpr size_t kFrameSizeBytes = 480 * sizeof(int16_t);
constexpr auto kStartupTimeout = std::chrono::seconds(2);
constexpr auto kPollInterval = std::chrono::milliseconds(5);
SpeakerCapturerLinux::SpeakerCapturerLinux()
: inited_(false), paused_(false), stop_flag_(false) {}
SpeakerCapturerLinux::~SpeakerCapturerLinux() {
Stop();
Destroy();
void ReleaseOperation(pa_operation* operation) {
if (!operation) return;
if (pa_operation_get_state(operation) == PA_OPERATION_RUNNING)
pa_operation_cancel(operation);
pa_operation_unref(operation);
}
} // namespace
SpeakerCapturerLinux::~SpeakerCapturerLinux() { Destroy(); }
int SpeakerCapturerLinux::Init(speaker_data_cb cb) {
if (inited_) return 0;
cb_ = cb;
cb_ = std::move(cb);
inited_ = true;
return 0;
}
int SpeakerCapturerLinux::Destroy() {
inited_ = false;
return 0;
// Use nonblocking iterations instead of pa_threaded_mainloop_wait(): a server
// that never replies, disappears, or never reaches STREAM_READY must not leave
// startup/Stop waiting for a signal that will never be delivered.
bool SpeakerCapturerLinux::PumpUntil(
const std::function<bool()>& ready,
std::chrono::steady_clock::time_point deadline) {
while (!stop_flag_) {
if (ready()) return true;
if (std::chrono::steady_clock::now() >= deadline ||
!PA_CONTEXT_IS_GOOD(pa_context_get_state(context_)))
return false;
if (pa_mainloop_iterate(mainloop_, 0, nullptr) < 0) return false;
std::this_thread::sleep_for(kPollInterval);
}
return false;
}
std::string SpeakerCapturerLinux::GetDefaultMonitorSourceName() {
std::string monitor_name;
std::mutex mtx;
std::condition_variable cv;
bool ready = false;
bool SpeakerCapturerLinux::Prepare() {
const auto deadline = std::chrono::steady_clock::now() + kStartupTimeout;
mainloop_ = pa_mainloop_new();
if (!mainloop_) return false;
context_ =
pa_context_new(pa_mainloop_get_api(mainloop_), "CrossDesk system audio");
if (!context_ ||
pa_context_connect(context_, nullptr, PA_CONTEXT_NOFLAGS, nullptr) < 0)
return false;
if (!PumpUntil(
[&] { return pa_context_get_state(context_) == PA_CONTEXT_READY; },
deadline))
return false;
pa_mainloop* mainloop = pa_mainloop_new();
pa_mainloop_api* api = pa_mainloop_get_api(mainloop);
pa_context* context = pa_context_new(api, "GetMonitor");
struct CallbackState {
std::string* name;
std::mutex* mtx;
std::condition_variable* cv;
bool* ready;
} state{&monitor_name, &mtx, &cv, &ready};
pa_context_set_state_callback(
context,
[](pa_context* c, void* userdata) {
auto* state = static_cast<CallbackState*>(userdata);
if (pa_context_get_state(c) == PA_CONTEXT_READY) {
pa_operation* operation = pa_context_get_server_info(
c,
[](pa_context*, const pa_server_info* info, void* userdata) {
auto* state = static_cast<CallbackState*>(userdata);
if (info && info->default_sink_name) {
*(state->name) =
std::string(info->default_sink_name) + ".monitor";
}
{
std::lock_guard<std::mutex> lock(*(state->mtx));
*(state->ready) = true;
}
state->cv->notify_one();
},
userdata);
if (operation) {
pa_operation_unref(operation);
}
}
struct ServerResult {
bool done = false;
std::string sink;
} server;
pa_operation* operation = pa_context_get_server_info(
context_,
[](pa_context*, const pa_server_info* info, void* data) {
auto& result = *static_cast<ServerResult*>(data);
if (info && info->default_sink_name)
result.sink = info->default_sink_name;
result.done = true;
},
&state);
&server);
const bool got_server =
operation && PumpUntil([&] { return server.done; }, deadline);
ReleaseOperation(operation);
if (!got_server || server.sink.empty()) return false;
pa_context_connect(context, nullptr, PA_CONTEXT_NOFLAGS, nullptr);
// Query the actual monitor name; a changed/default device need not follow
// the "<sink>.monitor" naming convention.
struct SinkResult {
bool done = false;
std::string monitor;
} sink;
operation = pa_context_get_sink_info_by_name(
context_, server.sink.c_str(),
[](pa_context*, const pa_sink_info* info, int eol, void* data) {
auto& result = *static_cast<SinkResult*>(data);
if (info && info->monitor_source_name)
result.monitor = info->monitor_source_name;
if (eol != 0) result.done = true;
},
&sink);
const bool got_sink =
operation && PumpUntil([&] { return sink.done; }, deadline);
ReleaseOperation(operation);
if (!got_sink || sink.monitor.empty()) return false;
std::thread loop_thread([&]() {
int ret = 0;
pa_mainloop_run(mainloop, &ret);
});
const pa_sample_spec spec{PA_SAMPLE_S16LE, 48000, 1};
stream_ = pa_stream_new(context_, "System audio", &spec, nullptr);
if (!stream_) return false;
pa_stream_set_read_callback(
stream_,
[](pa_stream* stream, size_t, void* data) {
static_cast<SpeakerCapturerLinux*>(data)->ReadAudio(stream);
},
this);
pa_buffer_attr attr{};
attr.maxlength = static_cast<uint32_t>(-1);
attr.fragsize = static_cast<uint32_t>(kFrameSizeBytes);
if (pa_stream_connect_record(stream_, sink.monitor.c_str(), &attr,
PA_STREAM_ADJUST_LATENCY) < 0)
return false;
if (!PumpUntil(
[&] {
const auto state = pa_stream_get_state(stream_);
return state == PA_STREAM_READY || !PA_STREAM_IS_GOOD(state);
},
deadline) ||
pa_stream_get_state(stream_) != PA_STREAM_READY)
return false;
LOG_INFO("System audio monitor ready: {}", sink.monitor);
return true;
}
void SpeakerCapturerLinux::Run() {
const bool ready = Prepare();
running_ = ready && !stop_flag_;
{
std::unique_lock<std::mutex> lock(mtx);
cv.wait_for(lock, std::chrono::seconds(2), [&] { return ready; });
std::lock_guard<std::mutex> lock(state_mtx_);
startup_success_ = running_;
startup_done_ = true;
}
pa_context_disconnect(context);
pa_context_unref(context);
pa_mainloop_quit(mainloop, 0);
loop_thread.join();
pa_mainloop_free(mainloop);
return monitor_name;
startup_wake_.notify_one();
if (!ready && !stop_flag_) {
LOG_ERROR(
"PulseAudio capture startup failed or timed out: {}",
context_ ? pa_strerror(pa_context_errno(context_)) : "no context");
}
while (running_ && !stop_flag_) {
if (pa_mainloop_iterate(mainloop_, 0, nullptr) < 0 ||
pa_context_get_state(context_) != PA_CONTEXT_READY ||
pa_stream_get_state(stream_) != PA_STREAM_READY)
break;
std::this_thread::sleep_for(kPollInterval);
}
running_ = false;
Cleanup();
}
int SpeakerCapturerLinux::Start() {
if (!inited_ || mainloop_thread_.joinable()) return -1;
if (!inited_) return -1;
if (IsRunning()) return 0;
Stop();
stop_flag_ = false;
mainloop_thread_ = std::thread([this]() {
std::string monitor_name = GetDefaultMonitorSourceName();
if (monitor_name.empty()) {
LOG_ERROR("Failed to get monitor source");
return;
}
mainloop_ = pa_threaded_mainloop_new();
pa_mainloop_api* api = pa_threaded_mainloop_get_api(mainloop_);
context_ = pa_context_new(api, "SpeakerCapturer");
pa_context_set_state_callback(
context_,
[](pa_context* c, void* userdata) {
auto self = static_cast<SpeakerCapturerLinux*>(userdata);
pa_context_state_t state = pa_context_get_state(c);
if (state == PA_CONTEXT_READY || state == PA_CONTEXT_FAILED ||
state == PA_CONTEXT_TERMINATED) {
pa_threaded_mainloop_signal(self->mainloop_, 0);
}
},
this);
if (pa_threaded_mainloop_start(mainloop_) < 0) {
LOG_ERROR("Failed to start mainloop");
Cleanup();
return;
}
pa_threaded_mainloop_lock(mainloop_);
if (pa_context_connect(context_, nullptr, PA_CONTEXT_NOFLAGS, nullptr) <
0) {
LOG_ERROR("Failed to connect context");
pa_threaded_mainloop_unlock(mainloop_);
Cleanup();
return;
}
while (true) {
pa_context_state_t state = pa_context_get_state(context_);
if (state == PA_CONTEXT_READY) break;
if (!PA_CONTEXT_IS_GOOD(state) || stop_flag_) {
pa_threaded_mainloop_unlock(mainloop_);
Cleanup();
return;
}
pa_threaded_mainloop_wait(mainloop_);
}
pa_sample_spec ss = {kFormat, kSampleRate, kChannels};
stream_ = pa_stream_new(context_, "Capture", &ss, nullptr);
pa_stream_set_read_callback(
stream_,
[](pa_stream* s, size_t len, void* u) {
auto self = static_cast<SpeakerCapturerLinux*>(u);
if (self->paused_ || self->stop_flag_) return;
const void* data = nullptr;
if (pa_stream_peek(s, &data, &len) < 0 || !data) return;
const uint8_t* p = static_cast<const uint8_t*>(data);
self->frame_cache_.insert(self->frame_cache_.end(), p, p + len);
while (self->frame_cache_.size() >= kFrameSizeBytes) {
std::vector<uint8_t> temp_frame(
self->frame_cache_.begin(),
self->frame_cache_.begin() + kFrameSizeBytes);
self->cb_(temp_frame.data(), kFrameSizeBytes, "audio");
self->frame_cache_.erase(
self->frame_cache_.begin(),
self->frame_cache_.begin() + kFrameSizeBytes);
}
pa_stream_drop(s);
},
this);
pa_buffer_attr attr = {.maxlength = (uint32_t)-1,
.tlength = 0,
.prebuf = 0,
.minreq = 0,
.fragsize = (uint32_t)kFrameSizeBytes};
if (pa_stream_connect_record(stream_, monitor_name.c_str(), &attr,
PA_STREAM_ADJUST_LATENCY) < 0) {
LOG_ERROR("Failed to connect stream");
pa_threaded_mainloop_unlock(mainloop_);
Cleanup();
return;
}
while (true) {
pa_stream_state_t s = pa_stream_get_state(stream_);
if (s == PA_STREAM_READY) break;
if (!PA_STREAM_IS_GOOD(s) || stop_flag_) {
pa_threaded_mainloop_unlock(mainloop_);
Cleanup();
return;
}
pa_threaded_mainloop_wait(mainloop_);
}
pa_threaded_mainloop_unlock(mainloop_);
while (!stop_flag_)
std::this_thread::sleep_for(std::chrono::milliseconds(100));
});
paused_ = false;
{
std::lock_guard<std::mutex> lock(state_mtx_);
startup_done_ = false;
startup_success_ = false;
}
mainloop_thread_ = std::thread([this] { Run(); });
std::unique_lock<std::mutex> lock(state_mtx_);
const bool started =
startup_wake_.wait_for(lock, kStartupTimeout + std::chrono::seconds(1),
[&] { return startup_done_; }) &&
startup_success_;
lock.unlock();
if (!started) {
Stop();
return -1;
}
return 0;
}
int SpeakerCapturerLinux::Stop() {
stop_flag_ = true;
if (mainloop_) {
pa_threaded_mainloop_lock(mainloop_);
pa_threaded_mainloop_signal(mainloop_, 0);
pa_threaded_mainloop_unlock(mainloop_);
}
if (mainloop_thread_.joinable()) {
mainloop_thread_.join();
}
Cleanup();
if (mainloop_thread_.joinable()) mainloop_thread_.join();
running_ = false;
return 0;
}
int SpeakerCapturerLinux::Destroy() {
Stop();
cb_ = nullptr;
inited_ = false;
return 0;
}
void SpeakerCapturerLinux::ReadAudio(pa_stream* stream) {
const void* data = nullptr;
size_t size = 0;
if (pa_stream_peek(stream, &data, &size) < 0 || size == 0) return;
if (data && !paused_ && !stop_flag_ && cb_) {
const auto* bytes = static_cast<const uint8_t*>(data);
frame_cache_.insert(frame_cache_.end(), bytes, bytes + size);
size_t consumed = 0;
while (!stop_flag_ && frame_cache_.size() - consumed >= kFrameSizeBytes) {
cb_(frame_cache_.data() + consumed, kFrameSizeBytes, "audio");
consumed += kFrameSizeBytes;
}
frame_cache_.erase(frame_cache_.begin(), frame_cache_.begin() + consumed);
} else {
frame_cache_.clear();
}
// Drop holes and paused data too, otherwise the read queue cannot advance.
pa_stream_drop(stream);
}
void SpeakerCapturerLinux::Cleanup() {
if (stream_) {
pa_stream_set_read_callback(stream_, nullptr, nullptr);
pa_stream_disconnect(stream_);
pa_stream_unref(stream_);
stream_ = nullptr;
}
if (context_) {
pa_context_disconnect(context_);
pa_context_unref(context_);
context_ = nullptr;
}
if (mainloop_) {
pa_threaded_mainloop_stop(mainloop_);
pa_threaded_mainloop_lock(mainloop_);
if (stream_) {
pa_stream_disconnect(stream_);
pa_stream_unref(stream_);
stream_ = nullptr;
}
if (context_) {
pa_context_disconnect(context_);
pa_context_unref(context_);
context_ = nullptr;
}
pa_threaded_mainloop_unlock(mainloop_);
pa_threaded_mainloop_free(mainloop_);
pa_mainloop_free(mainloop_);
mainloop_ = nullptr;
}
frame_cache_.clear();
}
bool SpeakerCapturerLinux::IsRunning() const { return running_; }
int SpeakerCapturerLinux::Pause() {
paused_ = true;
return 0;
}
int SpeakerCapturerLinux::Resume() {
paused_ = false;
return 0;
}
} // namespace crossdesk
} // namespace crossdesk
apps/desktop/src/platform/linux/speaker_capturer/speaker_capturer_linux.h
+26 -20
View File
@@ -4,53 +4,59 @@
* Copyright (c) 2025 by DI JUNKUN, All Rights Reserved.
*/
#ifndef _SPEAKER_CAPTURER_LINUX_H_
#define _SPEAKER_CAPTURER_LINUX_H_
#ifndef CROSSDESK_SPEAKER_CAPTURER_LINUX_H_
#define CROSSDESK_SPEAKER_CAPTURER_LINUX_H_
#include <pulse/pulseaudio.h>
#include <atomic>
#include <chrono>
#include <condition_variable>
#include <functional>
#include <mutex>
#include <string>
#include <thread>
#include <vector>
#include "speaker_capturer.h"
namespace crossdesk {
class SpeakerCapturerLinux : public SpeakerCapturer {
public:
SpeakerCapturerLinux();
~SpeakerCapturerLinux();
SpeakerCapturerLinux() = default;
~SpeakerCapturerLinux() override;
int Init(speaker_data_cb cb) override;
int Destroy() override;
int Start() override;
int Stop() override;
bool IsRunning() const override;
int Pause();
int Resume();
private:
std::string GetDefaultMonitorSourceName();
void Run();
bool Prepare();
bool PumpUntil(const std::function<bool()>& ready,
std::chrono::steady_clock::time_point deadline);
void Cleanup();
void ReadAudio(pa_stream* stream);
private:
speaker_data_cb cb_ = nullptr;
std::atomic<bool> inited_;
std::atomic<bool> paused_;
std::atomic<bool> stop_flag_;
speaker_data_cb cb_;
bool inited_ = false;
std::atomic<bool> paused_{false};
std::atomic<bool> stop_flag_{false};
std::atomic<bool> running_{false};
std::thread mainloop_thread_;
pa_threaded_mainloop* mainloop_ = nullptr;
std::mutex state_mtx_;
std::condition_variable startup_wake_;
bool startup_done_ = false;
bool startup_success_ = false;
// These objects are created, driven and freed exclusively by Run(). Stop
// only changes an atomic flag and joins, never races PulseAudio callbacks.
pa_mainloop* mainloop_ = nullptr;
pa_context* context_ = nullptr;
pa_stream* stream_ = nullptr;
std::mutex state_mtx_;
std::vector<uint8_t> frame_cache_;
};
} // namespace crossdesk
#endif
#endif