diff --git a/Cargo.lock b/Cargo.lock index f9e3144..f975f52 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -572,7 +572,7 @@ checksum = "1d674e81391d1e1ab681a28d99df07927c6d4aa5b027d7da16ba32d1d21ecd99" [[package]] name = "fluxo-rs" -version = "0.5.4" +version = "0.6.0" dependencies = [ "anyhow", "bluer", diff --git a/Cargo.toml b/Cargo.toml index b0bc8e0..78360eb 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "fluxo-rs" -version = "0.5.4" +version = "0.6.0" edition = "2024" [[bin]] diff --git a/README.md b/README.md index f3b613a..eb9f7e8 100644 --- a/README.md +++ b/README.md @@ -1,12 +1,18 @@ # fluxo -`fluxo` is a high-performance system metrics daemon and client designed specifically for Waybar. It entirely replaces standard shell scripts with a compiled Rust binary that collects data via a background polling loop and serves it over a Unix socket. +`fluxo` is a high-performance system metrics daemon and client for status bars. It entirely replaces standard shell scripts with a compiled Rust binary that collects data via a background polling loop and serves it over a Unix socket. -With its **100% Native, Content-Based Event-Driven Architecture**, it consumes effectively 0% CPU while idle and signals Waybar to redraw *only* when the rendered UI text or icons physically change. +With its **100% Native, Content-Based Event-Driven Architecture**, it consumes effectively 0% CPU while idle and pushes updates *only* when the rendered UI text or icons physically change. + +It speaks two dialects, both driven by that same change detection: + +- **Waybar** — the daemon sends `SIGRTMIN+N` and the bar re-runs `fluxo `. See [Waybar Configuration](#waybar-configuration). +- **Any other bar** (Quickshell, ags, eww, …) — hold one `fluxo stream ` child open and read newline-delimited JSON from its stdout. See [Streaming](#streaming). ## Key Features - **100% Native Architecture**: Zero shell-outs or subprocesses. Uses `bluer` for Bluetooth, `libpulse-binding` for audio, `zbus` for MPRIS/DND, and `notify` for backlight. +- **Push-Based for Any Bar**: `fluxo stream` turns the daemon into a push source, so bars that keep long-lived child processes never have to poll or re-exec a client. - **Content-Based Event Signaling**: `fluxo` evaluates your custom configuration formats internally. It only sends a `SIGRTMIN+X` signal to Waybar if the resulting string or CSS class has actually changed, eliminating pointless re-renders from raw polling fluctuations. - **Zero-Latency Interactions**: Direct library bindings mean that when you change your volume or connect a Bluetooth device via the CLI, the daemon updates instantly. - **Circuit Breaker (Failsafe)**: Automatically detects failing modules and enters a "Cool down" state, preventing resource waste and log spam. Fallback caching keeps your bar looking clean even during brief failures. @@ -115,6 +121,58 @@ To achieve zero-latency updates and zero-polling CPU usage, set `interval: 0` on } ``` +## Streaming + +Waybar's model is *signal the bar, the bar re-runs a command*, which is why the +one-shot client exists. Other bars have no equivalent of a signal, so they end up +emulating one by polling — spawning a `fluxo ` client per module per +interval, which means an exec and a dynamic link per reading, and values that are +still up to one interval stale. + +`fluxo stream` is the push-based path for those bars. Subscribe once, and the +daemon writes a JSON line whenever a module's rendered output actually changes: + +```console +$ fluxo stream cpu mem net +{"module":"cpu","text":"7.4|41.0","tooltip":"AMD Ryzen 7 PRO 3700U","class":"normal","percentage":7} +{"module":"mem","text":"9.41|15.49","class":"normal","percentage":60} +{"module":"cpu","text":"6.9|40.5","tooltip":"AMD Ryzen 7 PRO 3700U","class":"normal","percentage":6} +``` + +Each line is one complete JSON object: a `module` key naming the source, plus the +module's usual `text` and — when set — `tooltip`, `class` and `percentage`. A full +snapshot of every subscribed module is sent immediately on connect, so a bar +renders correctly the moment it attaches rather than after the first change. + +Two deliberate differences from the one-shot client: + +- `text` is **not** padded with figure-spaces and zero-width spaces. That padding + exists to stop Waybar's proportional font from reflowing, and is noise to a + consumer doing its own layout. +- Modules without a watch channel (`power`, `game`, `pool`) are re-evaluated on a + 5 second sweep rather than event-driven. Unchanged output is never written, so a + quiet sweep costs nothing on the wire. + +Quickshell, for example, needs one `Process` for the whole bar: + +```qml +Process { + running: true + command: ["fluxo", "stream", "cpu", "mem", "gpu", "net", "sys"] + stdout: SplitParser { + onRead: line => { + const ev = JSON.parse(line); + // dispatch on ev.module + } + } + onExited: reconnectTimer.restart() +} +``` + +The connection dies when the daemon restarts, so drive `running` back to `true` +from a timer — that is a reconnect, not a poll, and only ever fires while the +stream is actually down. + ## Debugging Use `--loglevel` to control log verbosity (trace, debug, info, warn, error): diff --git a/example.config.toml b/example.config.toml index e9e2326..4dde002 100644 --- a/example.config.toml +++ b/example.config.toml @@ -37,6 +37,10 @@ format = "{interface} ({ip}):  {rx:^4.1} MB/s  {tx:^4.1} MB/s" # enabled = false # set to false to disable this module at runtime # tokens: {usage}, {temp}, {model} format = "CPU: {usage:^4.1}% {temp:^4.1}C" +# Force a temperature sensor instead of auto-detecting it. Case-insensitive +# substring of the label shown by `sensors` (e.g. "Tctl", "Package id 0", +# "thinkpad CPU"). +# temp_sensor = "Tctl" [memory] # enabled = false # set to false to disable this module at runtime diff --git a/src/config.rs b/src/config.rs index 9975f5d..cc77fc7 100644 --- a/src/config.rs +++ b/src/config.rs @@ -125,6 +125,10 @@ pub struct CpuConfig { #[serde(default = "default_true")] pub enabled: bool, pub format: String, + /// Case-insensitive substring of the sensor label to force (e.g. `"Tctl"`, + /// `"Package id 0"`). Empty means auto-detect. + #[serde(default)] + pub temp_sensor: String, } impl Default for CpuConfig { @@ -132,6 +136,7 @@ impl Default for CpuConfig { Self { enabled: true, format: "CPU: {usage:>4.1}% {temp:>4.1}C".to_string(), + temp_sensor: String::new(), } } } diff --git a/src/daemon.rs b/src/daemon.rs index 7b22b56..e5bdabb 100644 --- a/src/daemon.rs +++ b/src/daemon.rs @@ -254,9 +254,10 @@ pub async fn run_daemon(config_path: Option) -> Result<()> { { let cfg = config.read().await; let fast_enabled = cfg.cpu.enabled || cfg.memory.enabled || cfg.sys.enabled; + let temp_sensor = cfg.cpu.temp_sensor.clone(); drop(cfg); if fast_enabled { - let mut daemon = HardwareDaemon::new(); + let mut daemon = HardwareDaemon::new(temp_sensor); let token = cancel_token.clone(); spawn_poll_loop_simple!( "fast_hw", @@ -274,7 +275,8 @@ pub async fn run_daemon(config_path: Option) -> Result<()> { let slow_enabled = cfg.gpu.enabled || cfg.disk.enabled; drop(cfg); if slow_enabled { - let mut daemon = HardwareDaemon::new(); + // Slow path never samples CPU temperature, so the override is moot. + let mut daemon = HardwareDaemon::new(String::new()); let token = cancel_token.clone(); spawn_poll_loop_simple!( "slow_hw", @@ -456,6 +458,21 @@ async fn run_ipc_loop( return; } + // `stream` holds the connection open and pushes a + // line per change instead of answering once. + if *module_name == "stream" { + let requested: Vec = + parts[1..].iter().map(|s| s.to_string()).collect(); + crate::stream::run_stream( + writer, + requested, + state_clone, + config_clone, + ) + .await; + return; + } + debug!(module = module_name, args = ?&parts[1..], "Handling IPC request"); let response = handle_request(module_name, &parts[1..], &state_clone, &config_clone).await; diff --git a/src/help.rs b/src/help.rs index ce4ffda..cd76f81 100644 --- a/src/help.rs +++ b/src/help.rs @@ -365,8 +365,16 @@ fn print_overview() { println!(" fluxo daemon [--config ] Start the background daemon"); println!(" fluxo reload Hot-reload the daemon config"); println!(" fluxo [args...] Query or control a module"); + println!(" fluxo stream ... Follow modules as JSON lines"); println!(" fluxo help [module] Show this help or module details\n"); + println!("\x1b[1mBARS:\x1b[0m"); + println!(" Waybar set `signal = N` per module and let the daemon poke it;"); + println!(" `fluxo ` is re-run by the bar on each signal."); + println!(" Everything keep one `fluxo stream cpu mem net ...` child running and"); + println!(" else read its stdout — one JSON line per actual change, so"); + println!(" there is no polling, no re-exec, and no interval lag.\n"); + println!("\x1b[1mCONFIGURATION:\x1b[0m"); println!(" Config file: $XDG_CONFIG_HOME/fluxo/config.toml"); println!(" Format tokens in config strings use {{token}} syntax."); diff --git a/src/ipc.rs b/src/ipc.rs index 35f635d..b759eaf 100644 --- a/src/ipc.rs +++ b/src/ipc.rs @@ -44,3 +44,37 @@ pub fn request_data(module: &str, args: &[&str]) -> anyhow::Result { Ok(response) } + +/// Subscribe to `modules` and copy the daemon's push stream to stdout. +/// +/// Blocks until the daemon closes the connection or the process is killed — +/// this is the long-lived counterpart to [`request_data`], so no read timeout +/// is set. Each line is flushed immediately, otherwise a consumer reading our +/// stdout through a pipe would see updates only once a block's worth had +/// accumulated. +pub fn stream_modules(modules: &[String]) -> anyhow::Result<()> { + use std::io::{BufRead, BufReader}; + + let sock = socket_path(); + debug!(?modules, "Opening stream to daemon socket: {}", sock); + let mut stream = UnixStream::connect(&sock)?; + + let mut request = String::from("stream"); + for module in modules { + request.push(' '); + request.push_str(module); + } + request.push('\n'); + stream.write_all(request.as_bytes())?; + + let reader = BufReader::new(stream); + let mut stdout = std::io::stdout(); + for line in reader.lines() { + let line = line?; + stdout.write_all(line.as_bytes())?; + stdout.write_all(b"\n")?; + stdout.flush()?; + } + + Ok(()) +} diff --git a/src/main.rs b/src/main.rs index 61ab99f..c9bc1d7 100644 --- a/src/main.rs +++ b/src/main.rs @@ -28,6 +28,7 @@ mod output; mod registry; mod signaler; mod state; +mod stream; mod utils; use clap::{Parser, Subcommand, ValueEnum}; @@ -91,6 +92,17 @@ enum Commands { }, /// Reload the daemon configuration Reload, + /// Stream module updates as newline-delimited JSON until interrupted + /// + /// Push-based alternative to polling `fluxo ` on a timer: one + /// connection, one line per actual change. Intended for bars that keep a + /// long-lived child process (Quickshell, ags, eww) rather than re-running a + /// command on a signal the way Waybar does. + Stream { + /// Modules to subscribe to (e.g. `cpu mem gpu net sys`) + #[arg(required = true)] + modules: Vec, + }, /// Show detailed help for all modules or a specific module Help { /// Optional module name to show detailed help for @@ -142,6 +154,12 @@ fn main() { process::exit(1); } }, + Commands::Stream { modules } => { + if let Err(e) = ipc::stream_modules(modules) { + error!("Stream ended: {}", e); + process::exit(1); + } + } Commands::Help { module } => { help::print_help(module.as_deref()); } diff --git a/src/modules/backlight.rs b/src/modules/backlight.rs index 6c6be1b..2c3c0ce 100644 --- a/src/modules/backlight.rs +++ b/src/modules/backlight.rs @@ -1,6 +1,7 @@ -//! Screen backlight indicator, driven by `inotify` on -//! `/sys/class/backlight/*/actual_brightness`. Falls back to a 5 s poll loop -//! to catch any missed events. +//! Screen backlight indicator, driven by `poll(2)` on +//! `/sys/class/backlight/*/actual_brightness`. The kernel backlight class calls +//! `sysfs_notify` on that attribute, so it wakes on `POLLPRI`; inotify does not +//! work on sysfs. Falls back to a 5 s timeout to catch any missed wakeups. use crate::config::Config; use crate::error::Result; @@ -8,9 +9,9 @@ use crate::modules::WaybarModule; use crate::output::WaybarOutput; use crate::state::{AppReceivers, BacklightState}; use crate::utils::{TokenValue, format_template}; -use notify::{Config as NotifyConfig, Event, RecommendedWatcher, RecursiveMode, Watcher}; +use std::io::{Read, Seek, SeekFrom}; +use std::os::fd::AsRawFd; use std::path::PathBuf; -use std::sync::mpsc; use std::time::Duration; use tokio::sync::watch; use tracing::{error, info}; @@ -85,13 +86,16 @@ impl BacklightDaemon { info!("Monitoring backlight device: {:?}", dir); let max_brightness_path = dir.join("max_brightness"); - let brightness_path = dir.join("actual_brightness"); - let brightness_path_fallback = dir.join("brightness"); + let actual_path = dir.join("actual_brightness"); + let set_path = dir.join("brightness"); - let target_file = if brightness_path.exists() { - brightness_path + // `brightness` is the requested level; `actual_brightness` is a + // hardware readback that can settle a few steps below it (amdgpu + // reports 64265/64764 at full), which would render as 99 %. + let level_file = if set_path.exists() { + set_path } else { - brightness_path_fallback + actual_path.clone() }; let get_percentage = || -> u8 { @@ -100,7 +104,7 @@ impl BacklightDaemon { .trim() .parse() .unwrap_or(100.0); - let current: f64 = std::fs::read_to_string(&target_file) + let current: f64 = std::fs::read_to_string(&level_file) .unwrap_or_default() .trim() .parse() @@ -117,29 +121,51 @@ impl BacklightDaemon { percentage: get_percentage(), }); - let (ev_tx, ev_rx) = mpsc::channel(); - let mut watcher = RecommendedWatcher::new( - move |res: notify::Result| { - if let Ok(event) = res - && event.kind.is_modify() - { - let _ = ev_tx.send(()); - } - }, - NotifyConfig::default(), - ) - .unwrap(); + // Watch the attribute the kernel actually notifies on, even when the + // percentage is read from `brightness`. + let watch_path = if actual_path.exists() { + actual_path + } else { + dir.join("brightness") + }; - if let Err(e) = watcher.watch(&target_file, RecursiveMode::NonRecursive) { - error!("Failed to watch backlight file: {}", e); + let Ok(mut file) = std::fs::File::open(&watch_path) else { + error!("Failed to open backlight file: {:?}", watch_path); return; - } + }; + + // A sysfs poll event stays raised until the attribute is re-read, so + // every wakeup must drain the fd or poll() spins. + let mut scratch = String::new(); + let mut drain = |file: &mut std::fs::File| { + scratch.clear(); + let _ = file.seek(SeekFrom::Start(0)); + let _ = file.read_to_string(&mut scratch); + }; + drain(&mut file); loop { - if ev_rx.recv_timeout(Duration::from_secs(5)).is_ok() { + let mut pfd = libc::pollfd { + fd: file.as_raw_fd(), + events: libc::POLLPRI | libc::POLLERR, + revents: 0, + }; + let rc = unsafe { libc::poll(&mut pfd, 1, 5_000) }; + + if rc < 0 { + let err = std::io::Error::last_os_error(); + if err.kind() == std::io::ErrorKind::Interrupted { + continue; + } + error!("poll() on backlight failed: {}", err); + return; + } + + if rc > 0 { + drain(&mut file); // Debounce bursts from scroll-driven brightness changes. std::thread::sleep(Duration::from_millis(50)); - while ev_rx.try_recv().is_ok() {} + drain(&mut file); let _ = tx.send(BacklightState { percentage: get_percentage(), diff --git a/src/modules/bt/mod.rs b/src/modules/bt/mod.rs index 1f06f80..4a0a767 100644 --- a/src/modules/bt/mod.rs +++ b/src/modules/bt/mod.rs @@ -55,60 +55,21 @@ impl BtDaemon { self.session = Some(bluer::Session::new().await?); } let session = self.session.as_ref().unwrap(); - let adapter = session.default_adapter().await?; + + let adapter = match session.default_adapter().await { + Ok(adapter) => adapter, + Err(e) if is_gone(&e) => return Ok(publish_powered_off(tx)), + Err(e) => return Err(e.into()), + }; let adapter_powered = adapter.is_powered().await.unwrap_or(false); let mut connected_devices = Vec::new(); if adapter_powered { - let mut addresses = adapter.device_addresses().await?; - addresses.sort(); - let audio_sink_uuid = bluer::Uuid::from_u128(0x0000110b_0000_1000_8000_00805f9b34fb); - - for addr in addresses { - let device = adapter.device(addr)?; - if !device.is_connected().await.unwrap_or(false) { - continue; - } - let uuids = device.uuids().await?.unwrap_or_default(); - if !uuids.contains(&audio_sink_uuid) { - continue; - } - - let mut dev_info = BtDeviceInfo { - device_address: addr.to_string(), - device_alias: device.alias().await.unwrap_or_else(|_| addr.to_string()), - battery_percentage: device.battery_percentage().await.unwrap_or(None), - plugin_data: vec![], - }; - - for p in PLUGINS.iter() { - if p.can_handle(&dev_info.device_alias, &dev_info.device_address) { - match p.get_data(config, state, &dev_info.device_address).await { - Ok(data) => { - dev_info.plugin_data = data - .into_iter() - .map(|(k, v)| { - let val_str = match v { - TokenValue::String(s) => s, - TokenValue::Int(i) => i.to_string(), - TokenValue::Float(f) => format!("{:.1}", f), - }; - (k, val_str) - }) - .collect(); - } - Err(e) => { - warn!("Plugin {} failed for {}: {}", p.name(), addr, e); - dev_info - .plugin_data - .push(("plugin_error".to_string(), e.to_string())); - } - } - break; - } - } - connected_devices.push(dev_info); + match collect_devices(&adapter, state, config).await { + Ok(devices) => connected_devices = devices, + Err(e) if is_gone(&e) => return Ok(publish_powered_off(tx)), + Err(e) => return Err(e.into()), } } @@ -121,6 +82,90 @@ impl BtDaemon { } } +/// BlueZ tears down its adapter and device objects when the radio is switched +/// off, so calls racing with that return `NotFound`. That is a normal +/// bluetooth-off transition, not a failure. +fn is_gone(err: &bluer::Error) -> bool { + err.kind == bluer::ErrorKind::NotFound +} + +fn publish_powered_off(tx: &watch::Sender) { + let _ = tx.send(BtState { + adapter_powered: false, + devices: vec![], + }); +} + +/// Collect paired+connected audio-sink devices, enriched by their plugins. +async fn collect_devices( + adapter: &bluer::Adapter, + state: &AppReceivers, + config: &Config, +) -> bluer::Result> { + let mut addresses = adapter.device_addresses().await?; + addresses.sort(); + let audio_sink_uuid = bluer::Uuid::from_u128(0x0000110b_0000_1000_8000_00805f9b34fb); + + let mut connected_devices = Vec::new(); + + for addr in addresses { + let device = match adapter.device(addr) { + Ok(device) => device, + Err(e) if is_gone(&e) => continue, + Err(e) => return Err(e), + }; + if !device.is_connected().await.unwrap_or(false) { + continue; + } + let uuids = match device.uuids().await { + Ok(uuids) => uuids.unwrap_or_default(), + // Device disappeared between enumeration and query. + Err(e) if is_gone(&e) => continue, + Err(e) => return Err(e), + }; + if !uuids.contains(&audio_sink_uuid) { + continue; + } + + let mut dev_info = BtDeviceInfo { + device_address: addr.to_string(), + device_alias: device.alias().await.unwrap_or_else(|_| addr.to_string()), + battery_percentage: device.battery_percentage().await.unwrap_or(None), + plugin_data: vec![], + }; + + for p in PLUGINS.iter() { + if p.can_handle(&dev_info.device_alias, &dev_info.device_address) { + match p.get_data(config, state, &dev_info.device_address).await { + Ok(data) => { + dev_info.plugin_data = data + .into_iter() + .map(|(k, v)| { + let val_str = match v { + TokenValue::String(s) => s, + TokenValue::Int(i) => i.to_string(), + TokenValue::Float(f) => format!("{:.1}", f), + }; + (k, val_str) + }) + .collect(); + } + Err(e) => { + warn!("Plugin {} failed for {}: {}", p.name(), addr, e); + dev_info + .plugin_data + .push(("plugin_error".to_string(), e.to_string())); + } + } + break; + } + } + connected_devices.push(dev_info); + } + + Ok(connected_devices) +} + static PLUGINS: LazyLock>> = LazyLock::new(|| vec![Box::new(PixelBudsPlugin)]); diff --git a/src/modules/dnd.rs b/src/modules/dnd.rs index d073fd8..f9f7017 100644 --- a/src/modules/dnd.rs +++ b/src/modules/dnd.rs @@ -20,6 +20,21 @@ use zbus::{Connection, fdo::PropertiesProxy}; /// Renders + toggles DND state. Args: `["show"]` (default) or `["toggle"]`. pub struct DndModule; +const SWAYNC_SERVICE: &str = "org.erikreider.swaync.control"; + +/// Build a SwayNC proxy, but only if the service is actually on the bus. +/// +/// Constructing the proxy unconditionally makes zbus spawn a property-caching +/// task that logs a `ServiceUnknown` warning when SwayNC isn't running. +async fn swaync_proxy(connection: &Connection) -> Option> { + let dbus = zbus::fdo::DBusProxy::new(connection).await.ok()?; + let name = zbus::names::BusName::try_from(SWAYNC_SERVICE).ok()?; + if !dbus.name_has_owner(name).await.unwrap_or(false) { + return None; + } + SwayncControlProxy::new(connection).await.ok() +} + /// Read dunst's `paused` property via raw D-Bus call. async fn dunst_get_paused(connection: &Connection) -> anyhow::Result { let reply = connection @@ -68,7 +83,7 @@ impl WaybarModule for DndModule { message: format!("DBus connection failed: {}", e), })?; - if let Ok(proxy) = SwayncControlProxy::new(&connection).await + if let Some(proxy) = swaync_proxy(&connection).await && let Ok(is_dnd) = proxy.dnd().await { let _ = proxy.set_dnd(!is_dnd).await; @@ -145,14 +160,14 @@ impl DndDaemon { info!("Connected to D-Bus for DND monitoring"); - if let Ok(proxy) = SwayncControlProxy::new(&connection).await + if let Some(proxy) = swaync_proxy(&connection).await && let Ok(is_dnd) = proxy.dnd().await { debug!("Found SwayNC, using signal-based DND monitoring"); let _ = tx.send(DndState { is_dnd }); if let Ok(props_proxy) = PropertiesProxy::builder(&connection) - .destination("org.erikreider.swaync.control")? + .destination(SWAYNC_SERVICE)? .path("/org/erikreider/swaync/control")? .build() .await diff --git a/src/modules/hardware.rs b/src/modules/hardware.rs index 946d9f6..4dc9dec 100644 --- a/src/modules/hardware.rs +++ b/src/modules/hardware.rs @@ -14,14 +14,48 @@ use tokio::sync::watch; pub struct HardwareDaemon { sys: System, components: Components, + /// User-forced CPU sensor label substring; empty means auto-detect. + temp_sensor: String, gpu_vendor: Option, gpu_poll_counter: u8, disk_poll_counter: u8, } +/// Sensors that are never the CPU but whose labels can look like one. Unlabeled +/// hwmon channels reach us as `" temp1"`, so a bare `temp1` match would +/// happily pick up the WiFi card or the NVMe drive. +const NON_CPU_SENSORS: &[&str] = &[ + "iwlwifi", "wifi", "nvme", "amdgpu", "nouveau", "gpu", "bat", "acpi ac", "composite", "edge", + "junction", "mem", +]; + +/// Score a component label for "is this the CPU package temperature". +/// Higher wins; 0 means unusable. +fn cpu_sensor_rank(label: &str) -> u8 { + if NON_CPU_SENSORS.iter().any(|s| label.contains(s)) { + return 0; + } + // AMD (k10temp) and Intel (coretemp) package sensors are the canonical ones. + if label.contains("tctl") || label.contains("tdie") || label.contains("package") { + 100 + } else if label.contains("core ") { + 80 + } else if label.contains("cpu") { + // e.g. thinkpad_acpi's "thinkpad CPU" — right sensor, coarser reading. + 70 + } else if label.contains("k10temp") || label.contains("coretemp") { + 60 + } else if label.contains("acpitz") { + 20 + } else { + 0 + } +} + impl HardwareDaemon { - /// Build a new daemon with an initial `sysinfo` snapshot. - pub fn new() -> Self { + /// Build a new daemon with an initial `sysinfo` snapshot. `temp_sensor` is + /// the configured CPU sensor override (empty to auto-detect). + pub fn new(temp_sensor: String) -> Self { let mut sys = System::new(); sys.refresh_cpu_usage(); sys.refresh_memory(); @@ -29,6 +63,7 @@ impl HardwareDaemon { Self { sys, components, + temp_sensor: temp_sensor.to_lowercase(), gpu_vendor: None, gpu_poll_counter: 0, // Start at 9 so (counter + 1) % 10 == 0 on the first tick. @@ -57,18 +92,24 @@ impl HardwareDaemon { .unwrap_or_else(|| "Unknown".to_string()); let mut cpu_temp = 0.0; + let mut best_rank = 0; for component in &self.components { let label = component.label().to_lowercase(); - if (label.contains("tctl") - || label.contains("cpu") - || label.contains("package") - || label.contains("temp1")) + + let rank = if self.temp_sensor.is_empty() { + cpu_sensor_rank(&label) + } else if label.contains(&self.temp_sensor) { + u8::MAX + } else { + 0 + }; + + if rank > best_rank && let Some(temp) = component.temperature() + && temp > 0.0 { cpu_temp = temp as f64; - if cpu_temp > 0.0 { - break; - } + best_rank = rank; } } diff --git a/src/modules/network.rs b/src/modules/network.rs index 65e5f35..c389d65 100644 --- a/src/modules/network.rs +++ b/src/modules/network.rs @@ -207,6 +207,11 @@ fn get_primary_interface() -> Result { let mask = u32::from_str_radix(parts[7], 16).unwrap_or(0); if dest == "00000000" { + // Skip ProtonVPN kill-switch interfaces — they exist + // only for leak protection and don't carry real traffic. + if iface.starts_with("pvpnksintrf") { + continue; + } defaults.push((mask, metric, iface.to_string())); } } diff --git a/src/signaler.rs b/src/signaler.rs index 123dafe..30d9a64 100644 --- a/src/signaler.rs +++ b/src/signaler.rs @@ -5,24 +5,42 @@ //! `watch::Receiver`, evaluates the module when its channel fires, and only //! signals Waybar when the rendered output actually changed. A 50 ms per-signal //! debounce prevents storms during rapid state churn. +//! +//! Waybar is optional: a bar that reads the daemon over `fluxo stream` (see +//! [`crate::stream`]) never wants a signal at all. Discovery is therefore +//! written to be cheap *and* to stay cheap when Waybar is simply not installed — +//! see [`WaybarSignaler::send_signal`]. use crate::config::Config; use crate::state::AppReceivers; use std::collections::HashMap; use std::sync::Arc; -use sysinfo::{ProcessesToUpdate, System}; use tokio::sync::RwLock; use tokio::time::{Duration, Instant, sleep}; use tracing::{debug, warn}; +/// Shortest gap between two signals with the same number. +const SIGNAL_DEBOUNCE: Duration = Duration::from_millis(50); + +/// First cooldown applied after a failed Waybar lookup. +const DISCOVERY_BACKOFF_MIN: Duration = Duration::from_secs(2); + +/// Ceiling for the discovery cooldown. A Waybar started later is still picked +/// up within this long, which is well inside "I just launched my bar" latency. +const DISCOVERY_BACKOFF_MAX: Duration = Duration::from_secs(60); + /// Sends real-time signals to the Waybar process. /// /// Resolves Waybar's PID lazily and caches it — the PID is invalidated on -/// signal failure (e.g. Waybar was restarted) and rediscovered via `sysinfo`. +/// signal failure (e.g. Waybar was restarted) and rediscovered by scanning +/// `/proc`. Failed lookups are cached too, under an exponential cooldown. pub struct WaybarSignaler { cached_pid: Option, - sys: System, last_signal_sent: HashMap, + /// When set, no rescan of `/proc` happens before this instant. + discovery_blocked_until: Option, + /// Current cooldown length, doubling on each consecutive failed lookup. + discovery_backoff: Duration, } impl WaybarSignaler { @@ -30,50 +48,97 @@ impl WaybarSignaler { pub fn new() -> Self { Self { cached_pid: None, - sys: System::new(), last_signal_sent: HashMap::new(), + discovery_blocked_until: None, + discovery_backoff: DISCOVERY_BACKOFF_MIN, } } - fn find_waybar_pid(&mut self) -> Option { - self.sys.refresh_processes(ProcessesToUpdate::All, true); - for (pid, process) in self.sys.processes() { - if process.name() == "waybar" { - return Some(pid.as_u32() as i32); + /// Scan `/proc` for a process named `waybar`. + /// + /// Reads only `/proc//comm`, one short line per process. The previous + /// implementation called `sysinfo`'s `refresh_processes(All, true)`, which + /// walks every thread of every process and parses `stat`, `status` and + /// `cmdline` for each — thousands of file reads per call with a browser + /// open, and it ran on *every* signal (see `send_signal`). + fn find_waybar_pid() -> Option { + let dir = std::fs::read_dir("/proc").ok()?; + for entry in dir.flatten() { + let file_name = entry.file_name(); + let Some(pid) = file_name.to_str().and_then(|s| s.parse::().ok()) else { + continue; + }; + // `comm` is truncated to 15 bytes by the kernel, which "waybar" + // comfortably fits inside. + if let Ok(comm) = std::fs::read_to_string(format!("/proc/{}/comm", pid)) + && comm.trim_end() == "waybar" + { + return Some(pid); } } None } - fn send_signal(&mut self, signal_num: i32) { - if let Some(last) = self.last_signal_sent.get(&signal_num) - && last.elapsed() < Duration::from_millis(50) - { - return; - } - - let mut valid_pid = false; + /// Resolve Waybar's PID, honouring the failed-lookup cooldown. + fn resolve_pid(&mut self) -> Option { if let Some(pid) = self.cached_pid && unsafe { libc::kill(pid, 0) } == 0 { - valid_pid = true; + return Some(pid); + } + self.cached_pid = None; + + if let Some(until) = self.discovery_blocked_until + && Instant::now() < until + { + return None; } - if !valid_pid { - self.cached_pid = self.find_waybar_pid(); - } - - if let Some(pid) = self.cached_pid { - let sig = libc::SIGRTMIN() + signal_num; - if unsafe { libc::kill(pid, sig) } == 0 { - debug!("Sent SIGRTMIN+{} to waybar (PID: {})", signal_num, pid); - self.last_signal_sent.insert(signal_num, Instant::now()); - } else { - warn!("Failed to send SIGRTMIN+{} to waybar", signal_num); - self.cached_pid = None; + match Self::find_waybar_pid() { + Some(pid) => { + debug!("Discovered waybar at PID {}", pid); + self.cached_pid = Some(pid); + self.discovery_blocked_until = None; + self.discovery_backoff = DISCOVERY_BACKOFF_MIN; + Some(pid) } + None => { + // Nothing to signal. Without this cooldown every state change + // rescanned /proc, which on a Waybar-less setup meant a full + // process walk several times a second, forever. + self.discovery_blocked_until = Some(Instant::now() + self.discovery_backoff); + self.discovery_backoff = + (self.discovery_backoff * 2).min(DISCOVERY_BACKOFF_MAX); + debug!( + "No waybar process; pausing discovery for {:?}", + self.discovery_backoff + ); + None + } + } + } + + fn send_signal(&mut self, signal_num: i32) { + // Debounce on attempt, not on success: recording the timestamp only + // after a successful send meant a missing Waybar bypassed the debounce + // entirely, so nothing throttled the work below. + if let Some(last) = self.last_signal_sent.get(&signal_num) + && last.elapsed() < SIGNAL_DEBOUNCE + { + return; + } + self.last_signal_sent.insert(signal_num, Instant::now()); + + let Some(pid) = self.resolve_pid() else { + return; + }; + + let sig = libc::SIGRTMIN() + signal_num; + if unsafe { libc::kill(pid, sig) } == 0 { + debug!("Sent SIGRTMIN+{} to waybar (PID: {})", signal_num, pid); } else { - debug!("Waybar process not found, skipping signal."); + warn!("Failed to send SIGRTMIN+{} to waybar", signal_num); + self.cached_pid = None; } } } diff --git a/src/stream.rs b/src/stream.rs new file mode 100644 index 0000000..8a283d9 --- /dev/null +++ b/src/stream.rs @@ -0,0 +1,176 @@ +//! Push-based module streaming for bars that are not Waybar. +//! +//! Waybar's model is *signal the bar, the bar re-runs a command*, so the +//! one-shot `fluxo ` client is the natural fit there. Every other shell +//! (Quickshell, ags, eww, …) has to emulate it by polling — spawning a process +//! per module per interval. At a 2 s interval over five modules that is 2.5 +//! forks a second, each one exec'ing and dynamically linking a 13 MB binary to +//! read a value the daemon already has in memory, and it still shows values up +//! to one interval stale. +//! +//! This module is the alternative: one long-lived connection, over which the +//! daemon writes a JSON line whenever a subscribed module's rendered output +//! actually changes. +//! +//! $ fluxo stream cpu mem net +//! {"module":"cpu","text":"7.4|41.0"} +//! {"module":"mem","text":"9.41|15.49"} +//! {"module":"cpu","text":"6.9|40.5"} +//! +//! Each line is one complete JSON object with a `module` key naming the source +//! and the module's usual fields (`text`, and `tooltip`/`class`/`percentage` +//! when set) flattened alongside it. Unlike the one-shot client, `text` is *not* +//! passed through [`crate::output::stabilize_text`]: the figure-space and +//! zero-width-space padding exists to stop Waybar's proportional font from +//! reflowing, and it is noise to a consumer that does its own layout. + +use crate::config::Config; +use crate::output::WaybarOutput; +use crate::state::AppReceivers; +use serde::Serialize; +use std::collections::HashMap; +use std::sync::Arc; +use tokio::io::AsyncWriteExt; +use tokio::sync::RwLock; +use tokio::time::{Duration, sleep}; +use tracing::debug; + +/// How often subscribed modules are re-evaluated regardless of channel +/// activity. +/// +/// This covers the dispatch-only modules (`power`, `game`, `pool`), which have +/// no watch channel to wake us, and doubles as a safety net for any state that +/// changes without its channel firing. Unchanged output is never written, so a +/// quiet tick costs nothing on the wire. +const SWEEP_INTERVAL: Duration = Duration::from_secs(5); + +/// One line of the stream: the module's own fields plus which module they came +/// from. +#[derive(Serialize)] +struct StreamEvent<'a> { + module: &'a str, + #[serde(flatten)] + output: WaybarOutput, +} + +/// Render `module_name` and return its serialised [`StreamEvent`] line. +async fn render_line( + module_name: &str, + state: &AppReceivers, + config: &Config, +) -> Option { + let args = crate::registry::signaler_default_args(module_name); + let output = crate::registry::dispatch(module_name, config, state, args) + .await + .ok()?; + let mut line = serde_json::to_string(&StreamEvent { + module: module_name, + output, + }) + .ok()?; + line.push('\n'); + Some(line) +} + +/// Generates [`run_stream`] from the central module registry. +/// +/// The shape mirrors [`crate::signaler`]: one cfg-gated `changed()` future per +/// watched module, and one `select!` arm per module that re-renders the module +/// names that channel feeds. +macro_rules! gen_stream { + ($( { $feature:literal, $field:ident, $state:ty, [$($name:literal),+], [$($sig_name:literal),+], $module:path, $signal:ident, [$($default_arg:literal),*], $config:ident } )*) => { + /// Serve one `stream` connection until the client disconnects. + /// + /// `requested` holds the module names the client subscribed to; any name + /// the registry does not know simply never produces output. + pub async fn run_stream( + mut writer: W, + requested: Vec, + receivers: AppReceivers, + config_lock: Arc>, + ) { + debug!(modules = ?requested, "Stream client subscribed"); + + // Two handles on the same channels: `watchers` is what we await + // changes on (needs `&mut`), `evaluator` is what module rendering + // reads from (needs `&`). Splitting them keeps the borrow checker + // out of the `select!` below; a `watch::Receiver` clone still sees + // the latest published value, so reads are unaffected. + let mut watchers = receivers.clone(); + let evaluator = receivers; + + let mut last: HashMap = HashMap::new(); + + // Re-render every subscribed module and write the ones that moved. + // Returns false once the client is gone. + macro_rules! flush_modules { + ($names:expr) => {{ + let mut alive = true; + for name in $names { + if !requested.iter().any(|r| r == name) { + continue; + } + let line = { + let config = config_lock.read().await; + render_line(name, &evaluator, &config).await + }; + let Some(line) = line else { continue }; + if last.get(name).is_some_and(|prev| prev == &line) { + continue; + } + last.insert(name.to_string(), line.clone()); + if writer.write_all(line.as_bytes()).await.is_err() + || writer.flush().await.is_err() + { + alive = false; + break; + } + } + alive + }}; + } + + // Prime the client with a full snapshot so a bar renders correctly + // the instant it connects, rather than after the first change. + let all: Vec<&str> = requested.iter().map(|s| s.as_str()).collect(); + if !flush_modules!(all.iter().copied()) { + debug!("Stream client disconnected during initial snapshot"); + return; + } + + loop { + $( + #[cfg(not(feature = $feature))] + let $field = std::future::pending::< + std::result::Result<(), tokio::sync::watch::error::RecvError>, + >(); + #[cfg(feature = $feature)] + let $field = watchers.$field.changed(); + )* + + let alive = tokio::select! { + $( + res = $field => { + if res.is_ok() { + flush_modules!([$($sig_name),+]) + } else { + true + } + } + )* + _ = sleep(SWEEP_INTERVAL) => { + flush_modules!(all.iter().copied()) + } + }; + + if !alive { + break; + } + } + + debug!("Stream client disconnected"); + } + }; +} + +for_each_watched_module!(gen_stream);