Revision control
Copy as Markdown
// Copyright (C) 2025 The Qt Company Ltd.
// SPDX-License-Identifier: LicenseRef-Qt-Commercial OR LGPL-3.0-only OR GPL-2.0-only OR GPL-3.0-only
#include "qpipewire_audiostream_p.h"
#include <QtMultimedia/private/qaudiosystem_platform_stream_support_p.h>
#include <QtMultimedia/private/qpipewire_audiocontextmanager_p.h>
#include <QtMultimedia/private/qpipewire_spa_pod_support_p.h>
#include <spa/pod/builder.h>
#if __has_include(<spa/param/audio/raw-utils.h>)
# include <spa/param/audio/raw-utils.h>
#else
# include <QtMultimedia/private/qpipewire_spa_compat_p.h>
#endif
#ifndef PW_KEY_NODE_FORCE_QUANTUM
# define PW_KEY_NODE_FORCE_QUANTUM "node.force-quantum"
#endif
#if !PW_CHECK_VERSION(0, 3, 50)
extern "C" {
int pw_stream_get_time_n(struct pw_stream *stream, struct pw_time *time, size_t size);
}
#endif
#include <array>
QT_BEGIN_NAMESPACE
namespace QtPipeWire {
QPipewireAudioStream::QPipewireAudioStream(const QAudioFormat &format) : m_format{ format }
{
}
QPipewireAudioStream::~QPipewireAudioStream() = default;
void QPipewireAudioStream::createStream(QSpan<spa_dict_item> extraProperties,
std::optional<NativePeriodFrames> nativePeriodFrames,
const char *streamName, StreamType type)
{
stream_events.version = PW_VERSION_STREAM_EVENTS;
switch (type) {
case StreamType::Ringbuffer:
stream_events.process = [](void *userData) {
reinterpret_cast<QPipewireAudioStream *>(userData)->processRingbuffer();
};
break;
case StreamType::Callback:
stream_events.process = [](void *userData) {
reinterpret_cast<QPipewireAudioStream *>(userData)->processCallback();
};
break;
default:
Q_UNREACHABLE_RETURN();
};
stream_events.state_changed = [](void *userData, pw_stream_state old, pw_stream_state state,
const char *error) {
reinterpret_cast<QPipewireAudioStream *>(userData)->stateChanged(old, state, error);
};
std::vector<spa_dict_item> properties{
{ PW_KEY_MEDIA_TYPE, "Audio" },
};
properties.insert(properties.end(), extraProperties.begin(), extraProperties.end());
std::string quantumStr;
if (nativePeriodFrames) {
quantumStr = std::to_string(qToUnderlying(*nativePeriodFrames));
properties.push_back({
PW_KEY_NODE_FORCE_QUANTUM,
quantumStr.data(),
});
}
QAudioContextManager::withEventLoopLock([&] {
m_stream = PwStreamHandle{
pw_stream_new_simple(QAudioContextManager::getEventLoop(), streamName,
makeProperties(properties).release(), &stream_events, this),
};
});
if (!m_stream)
qWarning() << "pw_stream_new_simple failed" << make_error_code().message();
}
bool QPipewireAudioStream::connectStream(ObjectSerial target, spa_direction direction)
{
int status = QAudioContextManager::withEventLoopLock([&] {
std::optional<ObjectId> targetNodeId =
QAudioContextManager::deviceMonitor().findObjectId(target);
if (!targetNodeId)
return -ENODEV;
bool deviceAlreadyRemoved = registerDeviceObserver(target);
if (!deviceAlreadyRemoved)
return -ENODEV;
std::array<uint8_t, 1024> buffer;
QT_WARNING_PUSH
QT_WARNING_DISABLE_CLANG("-Wmissing-field-initializers")
struct spa_pod_builder b = SPA_POD_BUILDER_INIT(buffer.data(), uint32_t(buffer.size()));
QT_WARNING_POP
spa_audio_info_raw audioInfo = asSpaAudioInfoRaw(m_format);
std::array<const struct spa_pod *, 1> params{
spa_format_audio_raw_build(&b, SPA_PARAM_EnumFormat, &audioInfo),
};
return pw_stream_connect(
m_stream.get(), direction, targetNodeId->value,
pw_stream_flags(PW_STREAM_FLAG_AUTOCONNECT | PW_STREAM_FLAG_MAP_BUFFERS
| PW_STREAM_FLAG_RT_PROCESS | PW_STREAM_FLAG_DONT_RECONNECT),
params.data(), params.size());
});
if (status < 0) {
qWarning() << "pw_stream_connect failed" << make_error_code(-status).message();
return false;
}
return true;
}
void QPipewireAudioStream::suspend()
{
int status = QAudioContextManager::withEventLoopLock([&] {
return pw_stream_set_active(m_stream.get(), false);
});
if (status < 0)
qWarning() << "pw_stream_set_active failed" << make_error_code(-status).message();
}
void QPipewireAudioStream::resume()
{
int status = QAudioContextManager::withEventLoopLock([&] {
m_skipNextTickDiscontinuity = true;
return pw_stream_set_active(m_stream.get(), true);
});
if (status < 0)
qWarning() << "pw_stream_set_active failed" << make_error_code(-status).message();
}
void QPipewireAudioStream::disconnectStream()
{
if (!m_stream)
return;
int status = QAudioContextManager::withEventLoopLock([&] {
int status = pw_stream_disconnect(m_stream.get());
m_stream = {};
return status;
});
if (status < 0)
qWarning() << "pw_stream_disconnect failed" << make_error_code(-status).message();
}
void QPipewireAudioStream::resetStream()
{
finalizeStream(); // mark the stream as stopped
disconnectStream();
if (!m_deviceRemovalObserver) {
m_self = {};
return;
}
QAudioContextManager::withEventLoopLock([&] {
if (m_deviceRemovalObserver)
unregisterDeviceObserver();
m_self = {};
});
}
bool QPipewireAudioStream::hasStream() const
{
return bool(m_stream);
}
bool QPipewireAudioStream::registerDeviceObserver(ObjectSerial nodeSerial)
{
m_deviceRemovalObserver = std::make_shared<ObjectRemoveObserver>(nodeSerial);
QObject::connect(m_deviceRemovalObserver.get(), &ObjectRemoveObserver::objectRemoved,
m_deviceRemovalObserver.get(), [this] {
handleDeviceRemoved();
});
return QAudioContextManager::deviceMonitor().registerObserver(m_deviceRemovalObserver);
}
void QPipewireAudioStream::unregisterDeviceObserver()
{
Q_ASSERT(m_deviceRemovalObserver);
QAudioContextManager::deviceMonitor().unregisterObserver(m_deviceRemovalObserver);
m_deviceRemovalObserver = {};
}
void QPipewireAudioStream::performXRunDetection(uint64_t framesPerBuffer) noexcept Q_DECL_NONBLOCKING_FUNCTION
{
// XRun detection does not work well with pause/resume.
// disabling for now, since we don't have any public API for notifying the application about
// xruns
constexpr bool runXRunDetection = false;
if (!runXRunDetection)
return;
struct pw_time time_info = {};
int status = pw_stream_get_time_n(m_stream.get(), &time_info, sizeof(pw_time));
if (status < 0) {
if (pw_check_library_version(0, 3, 50))
return; // no xrun detection on ancient pipewire
qFatal() << "pw_stream_get_time_n failed. This should not happen";
return;
}
if (m_skipNextTickDiscontinuity.load(std::memory_order_relaxed)) {
// prevent xrun detection to fire after resume, as ticks will continue incrementing
m_skipNextTickDiscontinuity = false;
} else if (std::abs(int64_t(m_expectedNextTick) - int64_t(time_info.ticks)) > 1024) {
m_totalNumberOfFrames = time_info.ticks;
m_xrunCount += 1;
xrunOccurred(m_xrunCount);
}
// CAVEAT:
// counts `ticks` in the device rates, which may be different to the rate the stream is
// running in. We therefore cannot do any precise xrun detection with this technique, but
// only to a best effort.
// TODO: can we use profiler events?
double rateFactor = double(time_info.rate.num) / time_info.rate.denom * m_format.sampleRate();
#if PW_CHECK_VERSION(1, 1, 0)
if (pw_check_library_version(1, 1, 0)) {
// LATER: rely on time_info.size, once 1.1 is the minimum required version
m_expectedNextTick = time_info.ticks + (time_info.size / rateFactor);
return;
}
#endif
m_expectedNextTick = time_info.ticks + (framesPerBuffer / rateFactor);
}
void QPipewireAudioStream::addFramesHandled(uint64_t arg)
{
m_totalNumberOfFrames += arg;
}
} // namespace QtPipeWire
QT_END_NAMESPACE