fix(niri): reconnect IPC and stop per-event throttling to prevent freeze
The niri IPC worker slept 1ms per event and never reconnected. Under an event burst the per-event cap back-pressures the socket, niri fills its send buffer and drops the stream; read_line then returns false, the detached thread exits and the module freezes permanently. Remove the per-event sleep so events drain as fast as they arrive, and wrap the socket setup and read loop in a reconnect loop that backs off and re-establishes the stream on drop. A running_ flag lets the thread exit cleanly on teardown. Fixes #5117.
This commit is contained in:
@@ -1,5 +1,6 @@
|
|||||||
#pragma once
|
#pragma once
|
||||||
|
|
||||||
|
#include <atomic>
|
||||||
#include <list>
|
#include <list>
|
||||||
#include <mutex>
|
#include <mutex>
|
||||||
#include <string>
|
#include <string>
|
||||||
@@ -18,6 +19,7 @@ class EventHandler {
|
|||||||
class IPC {
|
class IPC {
|
||||||
public:
|
public:
|
||||||
IPC();
|
IPC();
|
||||||
|
~IPC();
|
||||||
|
|
||||||
void registerForIPC(const std::string& ev, EventHandler* ev_handler);
|
void registerForIPC(const std::string& ev, EventHandler* ev_handler);
|
||||||
void unregisterForIPC(EventHandler* handler);
|
void unregisterForIPC(EventHandler* handler);
|
||||||
@@ -45,6 +47,8 @@ class IPC {
|
|||||||
util::JsonParser parser_;
|
util::JsonParser parser_;
|
||||||
std::mutex callbackMutex_;
|
std::mutex callbackMutex_;
|
||||||
std::list<std::pair<std::string, EventHandler*>> callbacks_;
|
std::list<std::pair<std::string, EventHandler*>> callbacks_;
|
||||||
|
|
||||||
|
std::atomic<bool> running_{true};
|
||||||
};
|
};
|
||||||
|
|
||||||
inline std::unique_ptr<IPC> gIPC;
|
inline std::unique_ptr<IPC> gIPC;
|
||||||
|
|||||||
@@ -23,6 +23,8 @@ namespace waybar::modules::niri {
|
|||||||
|
|
||||||
IPC::IPC() { startIPC(); }
|
IPC::IPC() { startIPC(); }
|
||||||
|
|
||||||
|
IPC::~IPC() { running_ = false; }
|
||||||
|
|
||||||
int IPC::connectToSocket() {
|
int IPC::connectToSocket() {
|
||||||
const char* socket_path = getenv("NIRI_SOCKET");
|
const char* socket_path = getenv("NIRI_SOCKET");
|
||||||
|
|
||||||
@@ -55,40 +57,60 @@ int IPC::connectToSocket() {
|
|||||||
void IPC::startIPC() {
|
void IPC::startIPC() {
|
||||||
// will start IPC and relay events to parseIPC
|
// will start IPC and relay events to parseIPC
|
||||||
|
|
||||||
int socketfd = connectToSocket();
|
std::thread([this]() {
|
||||||
|
|
||||||
std::thread([this, socketfd]() {
|
|
||||||
spdlog::info("Niri IPC starting");
|
spdlog::info("Niri IPC starting");
|
||||||
|
|
||||||
auto unix_istream = Gio::UnixInputStream::create(socketfd, true);
|
// Reconnect loop: if the event stream drops we back off briefly and
|
||||||
auto unix_ostream = Gio::UnixOutputStream::create(socketfd, false);
|
// re-establish the socket instead of leaving the module frozen forever.
|
||||||
auto istream = Gio::DataInputStream::create(unix_istream);
|
while (running_) {
|
||||||
auto ostream = Gio::DataOutputStream::create(unix_ostream);
|
int socketfd;
|
||||||
|
|
||||||
if (!ostream->put_string("\"EventStream\"\n") || !ostream->flush()) {
|
|
||||||
spdlog::error("Niri IPC: failed to start event stream");
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
|
|
||||||
std::string line;
|
|
||||||
if (!istream->read_line(line) || line != R"({"Ok":"Handled"})") {
|
|
||||||
spdlog::error("Niri IPC: failed to start event stream");
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
|
|
||||||
while (istream->read_line(line)) {
|
|
||||||
spdlog::debug("Niri IPC: received {}", line);
|
|
||||||
|
|
||||||
try {
|
try {
|
||||||
parseIPC(line);
|
socketfd = connectToSocket();
|
||||||
} catch (std::exception& e) {
|
} catch (std::exception& e) {
|
||||||
spdlog::warn("Failed to parse IPC message: {}, reason: {}", line, e.what());
|
spdlog::error("Niri IPC: failed to connect: {}", e.what());
|
||||||
} catch (...) {
|
std::this_thread::sleep_for(std::chrono::seconds(2));
|
||||||
throw;
|
continue;
|
||||||
}
|
}
|
||||||
|
|
||||||
std::this_thread::sleep_for(std::chrono::milliseconds(1));
|
auto unix_istream = Gio::UnixInputStream::create(socketfd, true);
|
||||||
|
auto unix_ostream = Gio::UnixOutputStream::create(socketfd, false);
|
||||||
|
auto istream = Gio::DataInputStream::create(unix_istream);
|
||||||
|
auto ostream = Gio::DataOutputStream::create(unix_ostream);
|
||||||
|
|
||||||
|
if (!ostream->put_string("\"EventStream\"\n") || !ostream->flush()) {
|
||||||
|
spdlog::error("Niri IPC: failed to start event stream");
|
||||||
|
std::this_thread::sleep_for(std::chrono::seconds(2));
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
|
||||||
|
std::string line;
|
||||||
|
if (!istream->read_line(line) || line != R"({"Ok":"Handled"})") {
|
||||||
|
spdlog::error("Niri IPC: failed to start event stream");
|
||||||
|
std::this_thread::sleep_for(std::chrono::seconds(2));
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
|
||||||
|
// Drain events as fast as they arrive; throttling here back-pressures the
|
||||||
|
// socket, fills niri's send buffer and makes niri drop the stream.
|
||||||
|
while (running_ && istream->read_line(line)) {
|
||||||
|
spdlog::debug("Niri IPC: received {}", line);
|
||||||
|
|
||||||
|
try {
|
||||||
|
parseIPC(line);
|
||||||
|
} catch (std::exception& e) {
|
||||||
|
spdlog::warn("Failed to parse IPC message: {}, reason: {}", line, e.what());
|
||||||
|
} catch (...) {
|
||||||
|
throw;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
if (!running_) break;
|
||||||
|
|
||||||
|
spdlog::warn("Niri IPC: event stream closed, reconnecting");
|
||||||
|
std::this_thread::sleep_for(std::chrono::seconds(2));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
spdlog::info("Niri IPC stopping");
|
||||||
}).detach();
|
}).detach();
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -196,12 +218,12 @@ void IPC::parseIPC(const std::string& line) {
|
|||||||
for (auto& win : windows_) {
|
for (auto& win : windows_) {
|
||||||
win["is_focused"] = focused && win["id"].asUInt64() == id;
|
win["is_focused"] = focused && win["id"].asUInt64() == id;
|
||||||
}
|
}
|
||||||
} else if (const auto &payload = ev["WindowLayoutsChanged"]) {
|
} else if (const auto& payload = ev["WindowLayoutsChanged"]) {
|
||||||
const auto &values = payload["changes"];
|
const auto& values = payload["changes"];
|
||||||
for (const auto &changed : values) {
|
for (const auto& changed : values) {
|
||||||
const auto id = changed[0].asUInt64();
|
const auto id = changed[0].asUInt64();
|
||||||
const auto &change = changed[1];
|
const auto& change = changed[1];
|
||||||
for (auto &win : windows_) {
|
for (auto& win : windows_) {
|
||||||
if (win["id"].asUInt64() == id) {
|
if (win["id"].asUInt64() == id) {
|
||||||
win["layout"] = change;
|
win["layout"] = change;
|
||||||
break;
|
break;
|
||||||
|
|||||||
Reference in New Issue
Block a user