Skip to content

Portal/Pipewire grabber bug fixing #1609

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
awawa-dev wants to merge 2 commits into
base: master
Choose a base branch
from
+75 −41
Open

Portal/Pipewire grabber bug fixing #1609

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
6e10802
Fix Pipewire race condition
awawa-dev Sep 17, 2026
c9aad2e
Fix Pipewire race condition - improvements
awawa-dev Sep 18, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 9 additions & 9 deletions include/grabber/linux/pipewire/PipewireHandler.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@
#include <memory>
#include <atomic>
#include <array>
#include <map>

#if !PW_CHECK_VERSION(0, 3, 29)
#define SPA_POD_PROP_FLAG_MANDATORY (1u << 3)
Expand Down Expand Up @@ -81,9 +82,12 @@ typedef void (*glTexImage2DFun)(GLenum target, GLint level, GLint internalformat
namespace sdbus{
class IConnection;
class IProxy;
class Variant;
}
class ScreenCastProxy;

using SdbusVariantMap = std::map<std::string, sdbus::Variant>;

class PipewireHandler : public QObject
{
Q_OBJECT
Expand Down Expand Up @@ -117,9 +121,6 @@ class PipewireHandler : public QObject
public Q_SLOTS:
void releaseWorkingFrame();
void getImage(PipewireImage& retVal);
void createSessionResponse(uint response, QString session);
void selectSourcesResponse(uint response);
void startResponse(uint response, QString restoreHandle, uint32_t nodeId, int nodeStreamWidth, int nodeStreamHeight);

void onParamsChanged(uint32_t id, const struct spa_pod* param);
void onStateChanged(enum pw_stream_state old, enum pw_stream_state state, const char* error);
Expand All @@ -138,6 +139,10 @@ public Q_SLOTS:
void reportError(const QString& input);
QString fourCCtoString(int64_t val);

void createSessionResponse(uint response, QString session);
void selectSourcesResponse(uint response);
void startResponse(uint response, QString restoreHandle, uint32_t nodeId, int nodeStreamWidth, int nodeStreamHeight);

pw_stream* createCapturingStream();
QString getSessionToken();
QString getRequestToken();
Expand All @@ -152,9 +157,6 @@ public Q_SLOTS:
uint _streamNodeId;

QString _sender;
QString _replySessionPath;
QString _sourceReplyPath;
QString _startReplyPath;

struct pw_thread_loop* _pwMainThreadLoop;
struct pw_context* _pwNewContext;
Expand Down Expand Up @@ -184,9 +186,7 @@ public Q_SLOTS:

std::unique_ptr<sdbus::IConnection> _dbusConnection;
std::unique_ptr<ScreenCastProxy> _screenCastProxy;
std::unique_ptr<sdbus::IProxy> _createSessionProxy;
std::unique_ptr<sdbus::IProxy> _selectSourceProxy;
std::unique_ptr<sdbus::IProxy> _startProxy;
std::map<std::string, std::function<void(uint32_t responseCode, const SdbusVariantMap& results)>> _portalHandlers;

#ifdef ENABLE_PIPEWIRE_EGL
eglGetProcAddressFun eglGetProcAddress = nullptr;
Expand Down
98 changes: 66 additions & 32 deletions sources/grabber/linux/pipewire/PipewireHandler.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -98,7 +98,7 @@ constexpr const int DEFAULT_UPDATE_NUMBER = 4;
PipewireHandler::PipewireHandler() :
_sessionHandle(""), _restorationToken(""), _errorMessage(""), _portalStatus(false),
_isError(false), _version(-1), _streamNodeId(0),
_sender(""), _replySessionPath(""), _sourceReplyPath(""), _startReplyPath(""),
_sender(""),
_pwMainThreadLoop(nullptr), _pwNewContext(nullptr), _pwContextConnection(nullptr), _pwStream(nullptr),
_targetMaxSize(512), _frameWidth(0),_frameHeight(0),_frameOrderRgb(false), _requestedFPS(10), _incomingFrame(nullptr),
_infoUpdate(DEFAULT_UPDATE_NUMBER), _initEGL(false), _enableEGL(true), _libEglHandle(nullptr), _libGlHandle(nullptr),
Expand All @@ -115,6 +115,10 @@ PipewireHandler::PipewireHandler() :
connect(this, &PipewireHandler::onProcessFrameSignal, this, &PipewireHandler::onProcessFrame);
connect(this, &PipewireHandler::onCoreErrorSignal, this, &PipewireHandler::onCoreError);


qRegisterMetaType<SdbusVariantMap>("SdbusVariantMap");
qRegisterMetaType<struct pw_buffer*>("struct pw_buffer*");

static std::once_flag pwInitOnce;
std::call_once(pwInitOnce, [] {
pw_init(nullptr, nullptr);
Expand Down Expand Up @@ -199,15 +203,10 @@ void PipewireHandler::closeSession()
}
}

_startReplyPath = "";
_sourceReplyPath = "";
_replySessionPath = "";
_sessionHandle = "";

_screenCastProxy = nullptr;
_createSessionProxy = nullptr;
_selectSourceProxy = nullptr;
_startProxy = nullptr;

_portalHandlers.clear();
_screenCastProxy = nullptr;
_dbusConnection = nullptr;

_pwStreamListener = {};
Expand Down Expand Up @@ -312,7 +311,7 @@ bool PipewireHandler::hasError()

bool PipewireHandler::isRestartNeeded()
{
return _isError || _replySessionPath.isEmpty() || (_startReplyPath.isEmpty() && !_portalStatus);
return _isError || (_portalHandlers.empty() && !_portalStatus);
}

int PipewireHandler::getVersion()
Expand Down Expand Up @@ -395,6 +394,54 @@ void PipewireHandler::startSession(QString restorationToken, uint32_t requestedF

qDebug().nospace() << "Sender: " << qPrintable(_sender);

// register Portal namespace listener
QString portalRequestNamespaceRule = QString("type='signal',path_namespace='/org/freedesktop/portal/desktop/request/%1'").arg(_sender);

try {
// forward sdbus calls via signal to QObject thread
_dbusConnection->addMatch(portalRequestNamespaceRule.toStdString(),
[this](sdbus::Message msg) {
std::string msgPath = msg.getPath();
std::string msgInterface = msg.getInterfaceName();
std::string msgMember = msg.getMemberName();

if (msgInterface == PORTAL_REQUEST && msgMember == PORTAL_RESPONSE)
{
qDebug().nospace() << "Pipewire: incoming dbus message...";

uint32_t responseCode;
SdbusVariantMap results;

try {
msg >> responseCode >> results;
QMetaObject::invokeMethod(this, [this, msgPath, responseCode, results = std::move(results)]() {
// sdbus signals to QObject thread receiver
auto it = _portalHandlers.find(msgPath);
if (it != _portalHandlers.end()) {
qDebug().nospace() << "Pipewire: processing dbus message...";
it->second(responseCode, results);
_portalHandlers.erase(it);
}
else {
qWarning().nospace() << "Pipewire: do not have a handler for the incoming dbus message";
}
});
}
catch (const sdbus::Error& e) {
qCritical().nospace() << "Pipewire: error while decoding dbus message: " << e.what();
}
}
else {
qWarning().nospace() << "Pipewire: ignored dbus message: " << msgInterface.c_str() << ", " << msgMember.c_str();
}
});

qDebug().nospace() << "Pipewire: registered DBus namespace matcher: " << qPrintable(portalRequestNamespaceRule);
}
catch (const sdbus::Error& e) {
qCritical().nospace() << "Pipewire: failed to register DBus matcher: " << e.getMessage().c_str();
}

QString requestUUID = getRequestToken();

std::map<std::string, sdbus::Variant> createSessionParams{
Expand All @@ -403,7 +450,7 @@ void PipewireHandler::startSession(QString restorationToken, uint32_t requestedF
};
try
{
auto responseSignalHandler = [this] (uint32_t resultCode, std::map<std::string, sdbus::Variant> results)
auto responseSignalHandler = [this] (uint32_t resultCode, const SdbusVariantMap& results)
{
auto sessionHandleIter = results.find("session_handle");
if (sessionHandleIter == results.end())
Expand All @@ -413,16 +460,12 @@ void PipewireHandler::startSession(QString restorationToken, uint32_t requestedF
else
{
QString session = QString::fromStdString(sessionHandleIter->second.get<std::string>());
QUEUE_CALL_2(this, createSessionResponse, uint, resultCode, QString, session );
createSessionResponse(resultCode, session);
}
};

sdbus::ObjectPath requestPath = _screenCastProxy->CreateSession(createSessionParams);
_replySessionPath = QString::fromStdString(static_cast<std::string>(requestPath));

_createSessionProxy = sdbus::createProxy(*_dbusConnection, sdbus::ServiceName{ DESKTOP_SERVICE }, sdbus::ObjectPath{_replySessionPath.toStdString()});

_createSessionProxy->uponSignal(SignalName{ PORTAL_RESPONSE }).onInterface(InterfaceName{ PORTAL_REQUEST }).call(responseSignalHandler);
_portalHandlers[static_cast<std::string>(requestPath)] = responseSignalHandler;
}
catch(std::exception& ex)
{
Expand Down Expand Up @@ -462,16 +505,13 @@ void PipewireHandler::createSessionResponse(uint response, QString session)

try
{
auto responseSignalHandler = [this] (uint32_t resultCode, std::map<std::string, sdbus::Variant> results)
auto responseSignalHandler = [this] (uint32_t resultCode, const SdbusVariantMap& results)
{
QUEUE_CALL_1(this, selectSourcesResponse, uint, resultCode);
selectSourcesResponse(resultCode);
};

sdbus::ObjectPath sourceRequestPath = _screenCastProxy->SelectSources(sdbus::ObjectPath{ _sessionHandle.toStdString() }, selectSourceParams);
_sourceReplyPath = QString::fromStdString(static_cast<std::string>(sourceRequestPath));

_selectSourceProxy = sdbus::createProxy(*_dbusConnection, sdbus::ServiceName{ DESKTOP_SERVICE }, sdbus::ObjectPath{_sourceReplyPath.toStdString()});
_selectSourceProxy->uponSignal(SignalName{ PORTAL_RESPONSE }).onInterface(InterfaceName{ PORTAL_REQUEST }).call(responseSignalHandler);
_portalHandlers[static_cast<std::string>(sourceRequestPath)] = responseSignalHandler;
}
catch(std::exception& ex)
{
Expand All @@ -497,10 +537,8 @@ void PipewireHandler::selectSourcesResponse(uint response)

try
{
auto responseSignalHandler = [this] (uint32_t resultCode, std::map<std::string, sdbus::Variant> results)
auto responseSignalHandler = [this] (uint32_t resultCode, const SdbusVariantMap& results)
{
_startReplyPath = "";

if (resultCode != 0)
{
reportError(QString("Start session returned an error code: %1").arg(((resultCode == 1) ? "cancelled" : "other")));
Expand Down Expand Up @@ -552,7 +590,7 @@ void PipewireHandler::selectSourcesResponse(uint response)
nodeStreamHeight = dim.get<1>();
}

QUEUE_CALL_5(this, startResponse, uint, resultCode, QString, restoreHandle, uint32_t, nodeId, int, nodeStreamWidth, int, nodeStreamHeight);
startResponse(resultCode, restoreHandle, nodeId, nodeStreamWidth, nodeStreamHeight);

}
}
Expand All @@ -563,11 +601,7 @@ void PipewireHandler::selectSourcesResponse(uint response)
};

sdbus::ObjectPath startRequestPath = _screenCastProxy->Start(ObjectPath{ _sessionHandle.toStdString() }, "", startParams);

_startReplyPath = QString::fromStdString(static_cast<std::string>(startRequestPath));

_startProxy = sdbus::createProxy(*_dbusConnection, sdbus::ServiceName{ DESKTOP_SERVICE }, sdbus::ObjectPath{_startReplyPath.toStdString()});
_startProxy->uponSignal(SignalName{ PORTAL_RESPONSE }).onInterface(InterfaceName{ PORTAL_REQUEST }).call(responseSignalHandler);
_portalHandlers[static_cast<std::string>(startRequestPath)] = responseSignalHandler;
}
catch(std::exception& ex)
{
Expand Down
Loading