3 Commits
Author SHA1 Message Date
molluskandClaude Opus 4.7 f939441e31 README: document multi-viewer + bandwidth pre-flight
Updates the status section to move multi-viewer out of "not yet
working", adds a Configuration section pointing at the new TOML config
at ~/.config/pixelpass/config.toml, and a Multi-viewer section
covering the lazy-sticky lifecycle, the --max-viewers cap, the
bandwidth-bitrate tradeoff, and how to fit more viewers by dropping
--bitrate. Known-limitations section gains "late joiners see ~2 s of
garbage" (expected behavior) and drops the now-stale "single viewer
per host" line.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2026-05-21 17:00:58 -04:00
molluskandClaude Opus 4.7 153febe078 pre-flight: bandwidth test + persistent config
First-run host launch now offers a one-time upstream measurement
against speed.cloudflare.com/__up via ureq (~5 MB POST, ~5s). The
result lives at ~/.config/pixelpass/config.toml under [bandwidth]
and feeds the default --max-viewers calculation on subsequent runs.

Sticky semantics for the dialog:
- Unmeasured: first-run prompt (Run / Skip)
- Measured / Skipped: silent — never re-prompts
- Failed: ask again on next launch (Retry / give up → Skipped)

`pixelpass --reconfigure` re-runs the test unconditionally for users
whose connection has changed (new ISP, moved house, etc.).

--max-viewers is now Option<u32>. When unset, host startup loads the
saved measurement, runs recommended_max_viewers(safe_mbps, bitrate),
and surfaces the source in the banner: "max viewers : N (auto: X.X
Mbps measured upstream)" — or user-specified / default fallback.

User verified end-to-end on 2026-05-21 16:54 EDT: first-run dialog,
skip path, run path, --reconfigure refresh, and banner integration
all work as expected.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2026-05-21 16:55:11 -04:00
molluskandClaude Opus 4.7 ffe5a90686 multi-viewer: broadcast fanout + supervisor lifecycle
One gst capture pipeline now fans out to N concurrent viewers via a
tokio::sync::broadcast<Arc<Vec<u8>>>. The HTTP listener accepts forever;
each accepted connection spawns a sender task draining its own
broadcast::Receiver. Slow consumers see Lagged and skip ahead — MPEG-TS
resyncs at the next keyframe.

Host runtime is now lazy + sticky: a supervisor task owns the capture
handle and viewer count. First viewer triggers capture::spawn; last
viewer triggers shutdown. Subsequent reconnects re-trigger the portal
dialog as expected. --max-viewers (default 2) caps concurrent viewers;
additional connections get a "host is full" refusal and are dropped.

Banner updated to reflect the new lifecycle and viewer cap.

NOT YET RUNTIME-VERIFIED. cargo build is clean and the pipeline-level
smoke test still passes, but the multi-viewer behavior (cap enforcement,
lazy-sticky restart, concurrent fanout) requires manual end-to-end
testing with the portal dialog + multiple mpv instances.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2026-05-21 16:11:43 -04:00
11 changed files with 738 additions and 61 deletions
Generated
+37
View File
@@ -3074,6 +3074,7 @@ dependencies = [
"anyhow", "anyhow",
"arboard", "arboard",
"ashpd", "ashpd",
"chrono",
"clap", "clap",
"dialoguer", "dialoguer",
"directories", "directories",
@@ -3086,8 +3087,10 @@ dependencies = [
"thiserror 2.0.18", "thiserror 2.0.18",
"tokio", "tokio",
"tokio-util", "tokio-util",
"toml",
"tracing", "tracing",
"tracing-subscriber", "tracing-subscriber",
"ureq",
"uuid", "uuid",
"x11rb", "x11rb",
] ]
@@ -4448,6 +4451,34 @@ version = "0.9.0"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8ecb6da28b8a351d773b68d5825ac39017e680750f980f3a1a85cd8dd28a47c1" checksum = "8ecb6da28b8a351d773b68d5825ac39017e680750f980f3a1a85cd8dd28a47c1"
[[package]]
name = "ureq"
version = "3.3.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "dea7109cdcd5864d4eeb1b58a1648dc9bf520360d7af16ec26d0a9354bafcfc0"
dependencies = [
"base64",
"log",
"percent-encoding",
"rustls",
"rustls-pki-types",
"ureq-proto",
"utf8-zero",
"webpki-roots",
]
[[package]]
name = "ureq-proto"
version = "0.6.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e994ba84b0bd1b1b0cf92878b7ef898a5c1760108fe7b6010327e274917a808c"
dependencies = [
"base64",
"http",
"httparse",
"log",
]
[[package]] [[package]]
name = "url" name = "url"
version = "2.5.8" version = "2.5.8"
@@ -4461,6 +4492,12 @@ dependencies = [
"serde_derive", "serde_derive",
] ]
[[package]]
name = "utf8-zero"
version = "0.8.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b8c0a043c9540bae7c578c88f91dda8bd82e59ae27c21baca69c8b191aaf5a6e"
[[package]] [[package]]
name = "utf8_iter" name = "utf8_iter"
version = "1.0.4" version = "1.0.4"
+3
View File
@@ -30,6 +30,9 @@ uuid = { version = "1", features = ["v4"] }
iroh-tickets = "1.0.0-rc.0" iroh-tickets = "1.0.0-rc.0"
dialoguer = { version = "0.12", default-features = false } dialoguer = { version = "0.12", default-features = false }
arboard = { version = "3", default-features = false, features = ["wayland-data-control"] } arboard = { version = "3", default-features = false, features = ["wayland-data-control"] }
ureq = { version = "3", default-features = false, features = ["rustls"] }
toml = "1"
chrono = { version = "0.4", default-features = false, features = ["clock", "serde"] }
[profile.release] [profile.release]
lto = "thin" lto = "thin"
+51 -5
View File
@@ -22,11 +22,15 @@ Working:
- iroh QUIC bi-stream tunnel, direct-UDP and relay paths both verified - iroh QUIC bi-stream tunnel, direct-UDP and relay paths both verified
- Interactive Host/View menu with clipboard auto-copy and mpv/VLC picker - Interactive Host/View menu with clipboard auto-copy and mpv/VLC picker
- Headless mode for scripts (`pixelpass <ticket>`) - Headless mode for scripts (`pixelpass <ticket>`)
- Multi-viewer fanout (default 2, configurable via `--max-viewers`;
shared gst pipeline, one broadcast channel per host)
- First-run upstream bandwidth pre-flight, persisted to
`~/.config/pixelpass/config.toml` and used to auto-size the default
viewer cap
Not yet working: Not yet working:
- X11 capture (stubbed, returns an error) - X11 capture (stubbed, returns an error)
- Per-app audio routing (`--app <name>` is a flag stub) - Per-app audio routing (`--app <name>` is a flag stub)
- Multi-viewer (single viewer per host by design right now)
- `--repair` (PipeWire orphan cleanup) is a stub - `--repair` (PipeWire orphan cleanup) is a stub
## Quick start ## Quick start
@@ -39,7 +43,14 @@ pixelpass
On the host machine: pick "Host", share a monitor via the portal dialog, On the host machine: pick "Host", share a monitor via the portal dialog,
the ticket lands on your clipboard. Send it to your viewer however you the ticket lands on your clipboard. Send it to your viewer however you
like (chat, email, paste in a note). like (chat, email, paste in a note). The same ticket works for multiple
viewers up to your `--max-viewers` cap.
The very first host launch offers a one-time upstream bandwidth test
(~5 s, ~5 MB to Cloudflare's open speed-test endpoint) so it can pick
a sensible default for the viewer cap. You can skip it and a
conservative default (2 viewers) is used; re-run it later with
`pixelpass --reconfigure`.
On the viewer machine: run `pixelpass`, pick "View", paste the ticket, On the viewer machine: run `pixelpass`, pick "View", paste the ticket,
pick mpv or VLC. The player launches detached and the stream starts. pick mpv or VLC. The player launches detached and the stream starts.
@@ -154,6 +165,42 @@ relay path otherwise. Both have been verified end-to-end.
- **VAAPI H.264 over x264**: ~5% of one CPU core instead of ~50% on - **VAAPI H.264 over x264**: ~5% of one CPU core instead of ~50% on
the host's hardware. the host's hardware.
## Configuration
`pixelpass` keeps a small TOML config at `~/.config/pixelpass/config.toml`
(or the XDG equivalent). Right now it only stores the result of the
bandwidth pre-flight:
```toml
[bandwidth]
status = "measured" # measured | skipped | failed | unmeasured
upstream_mbps = 8.78 # safe estimate (raw * 0.8)
measured_at = "2026-05-21T20:41:16Z"
```
- `pixelpass --reconfigure` re-runs the test (e.g. after an ISP change).
- Deleting the file resets pixelpass to first-run state.
- Skip is sticky — once you skip the test, pixelpass won't ask again
unless you reconfigure.
## Multi-viewer
One gst capture pipeline fans out to N concurrent viewers via a
`tokio::sync::broadcast` channel. The same ticket is reusable: as long
as a viewer is connected, capture stays alive; when the last one
leaves, the pipeline tears down and the portal stops streaming. A new
viewer connecting after that re-triggers the portal dialog.
Capacity is bounded by upstream bandwidth (each viewer is its own
encrypted egress). The default cap comes from the bandwidth pre-flight
result; `--max-viewers <N>` overrides it. When the cap is hit,
additional connections are politely refused with a "host is full"
message and the host keeps running.
For more viewers, drop the per-viewer bitrate: e.g. `pixelpass
--bitrate 2500 --max-viewers 4` fits four 2.5 Mbps streams in roughly
12 Mbps of upstream.
## Known limitations and gotchas ## Known limitations and gotchas
- **VLC needs `vlc-plugin-dvb` and `vlc-plugin-ffmpeg`** on Arch-family - **VLC needs `vlc-plugin-dvb` and `vlc-plugin-ffmpeg`** on Arch-family
@@ -164,9 +211,8 @@ relay path otherwise. Both have been verified end-to-end.
- **Audio echo** if the host plays the stream through speakers and - **Audio echo** if the host plays the stream through speakers and
captures system audio — expected, the mic / monitor picks up the captures system audio — expected, the mic / monitor picks up the
playback. Headphones bypass it. playback. Headphones bypass it.
- **Single viewer per host** by design right now. Restarting the player - **Late joiners see ~2 s of garbage** before the next keyframe lets
against the same URL fails with "connection refused"; restart the their decoder lock. Expected behavior, not a bug.
viewer too.
- **VAAPI driver must be package-tracked**, not an orphaned `.so` on - **VAAPI driver must be package-tracked**, not an orphaned `.so` on
disk. mpv's `--hwdec=auto` silently falls back to software decode disk. mpv's `--hwdec=auto` silently falls back to software decode
otherwise, which then chokes on a low-power viewer. otherwise, which then chokes on a low-power viewer.
+15
View File
@@ -45,6 +45,13 @@ pub struct Cli {
#[arg(long)] #[arg(long)]
pub low_latency: bool, pub low_latency: bool,
/// Maximum number of concurrent viewers. Additional connections are
/// politely refused with a "host full" message. Defaults to the
/// connection-aware recommendation from the bandwidth pre-flight if
/// available, otherwise 2.
#[arg(long)]
pub max_viewers: Option<u32>,
// ── viewer options ──────────────────────────────────────────────── // ── viewer options ────────────────────────────────────────────────
/// Local TCP port for the viewer to expose (default: random). /// Local TCP port for the viewer to expose (default: random).
#[arg(long, default_value_t = 0)] #[arg(long, default_value_t = 0)]
@@ -58,6 +65,12 @@ pub struct Cli {
/// Clean up orphaned PipeWire state from a crashed host run, then exit. /// Clean up orphaned PipeWire state from a crashed host run, then exit.
#[arg(long)] #[arg(long)]
pub repair: bool, pub repair: bool,
/// Re-run the bandwidth pre-flight test, save the result, then exit.
/// Use this if your connection has changed (new ISP, moved house, etc.)
/// or if the previously saved test result is stale.
#[arg(long)]
pub reconfigure: bool,
} }
#[derive(ValueEnum, Clone, Copy, Debug)] #[derive(ValueEnum, Clone, Copy, Debug)]
@@ -76,6 +89,7 @@ pub struct HostOpts {
pub framerate: u32, pub framerate: u32,
pub no_hwencode: bool, pub no_hwencode: bool,
pub low_latency: bool, pub low_latency: bool,
pub max_viewers: Option<u32>,
pub interactive: bool, pub interactive: bool,
} }
@@ -96,6 +110,7 @@ impl Cli {
framerate: self.framerate, framerate: self.framerate,
no_hwencode: self.no_hwencode, no_hwencode: self.no_hwencode,
low_latency: self.low_latency, low_latency: self.low_latency,
max_viewers: self.max_viewers,
interactive, interactive,
} }
} }
+73
View File
@@ -0,0 +1,73 @@
//! One-shot upstream bandwidth measurement against Cloudflare's open
//! speed-test endpoint. POST a fixed payload, time it, derive Mbps.
//!
//! Run via `tokio::task::spawn_blocking` from async contexts — ureq is a
//! blocking client and we don't want to wedge the tokio runtime during
//! the test.
use anyhow::{Context, Result};
use std::time::{Duration, Instant};
const ENDPOINT: &str = "https://speed.cloudflare.com/__up";
const PAYLOAD_BYTES: usize = 5 * 1024 * 1024; // 5 MiB
const HTTP_TIMEOUT: Duration = Duration::from_secs(30);
/// Multiplier applied to the raw measurement. TCP slow-start, ramp-up, and
/// real-world contention all mean a one-shot upstream test slightly
/// overestimates sustainable throughput; clamp to 80% for headroom.
const SAFETY_FACTOR: f64 = 0.80;
/// Result of a successful measurement.
#[derive(Debug, Clone)]
pub struct Measurement {
/// Raw measured throughput in megabits per second.
pub raw_mbps: f64,
/// `raw_mbps * SAFETY_FACTOR` — the value to use when sizing things.
pub safe_mbps: f64,
/// How long the upload took.
pub elapsed: Duration,
}
/// Blocking upload-speed test. Call from a `spawn_blocking` task.
pub fn measure_upstream_blocking() -> Result<Measurement> {
let payload = vec![0u8; PAYLOAD_BYTES];
let agent = ureq::Agent::config_builder()
.timeout_global(Some(HTTP_TIMEOUT))
.build()
.new_agent();
let start = Instant::now();
let response = agent
.post(ENDPOINT)
.content_type("application/octet-stream")
.send(&payload[..])
.context("upload request to Cloudflare failed")?;
let elapsed = start.elapsed();
let status = response.status();
if !status.is_success() {
anyhow::bail!("Cloudflare returned HTTP {status}");
}
let bits = (PAYLOAD_BYTES as f64) * 8.0;
let seconds = elapsed.as_secs_f64().max(0.001);
let raw_mbps = bits / seconds / 1_000_000.0;
let safe_mbps = raw_mbps * SAFETY_FACTOR;
Ok(Measurement {
raw_mbps,
safe_mbps,
elapsed,
})
}
/// Convert a safe-upstream Mbps figure plus the host's per-viewer bitrate
/// (kbps for video, ignoring audio + protocol overhead which we account for
/// via SAFETY_FACTOR) into a recommended viewer count. Floors to at least 1.
pub fn recommended_max_viewers(safe_mbps: f64, bitrate_kbps: u32) -> u32 {
let per_viewer_mbps = (bitrate_kbps as f64) / 1000.0;
if per_viewer_mbps <= 0.0 {
return 1;
}
let n = (safe_mbps / per_viewer_mbps).floor();
if n < 1.0 { 1 } else { n as u32 }
}
+101
View File
@@ -0,0 +1,101 @@
//! Persistent user-level config at `~/.config/pixelpass/config.toml`.
//!
//! Right now this only tracks the bandwidth pre-flight result. Future
//! preferences (default player, default bitrate, etc.) can hang off the
//! same file under their own `[section]`.
use anyhow::{Context, Result};
use chrono::{DateTime, Utc};
use directories::ProjectDirs;
use serde::{Deserialize, Serialize};
use std::fs;
use std::io::Write;
use std::path::PathBuf;
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct Config {
#[serde(default)]
pub bandwidth: BandwidthEntry,
}
/// Result of the first-run upstream measurement.
///
/// `status = "unmeasured"` means we've never asked the user — show the
/// first-run dialog. `"measured"` means we have a number. `"skipped"`
/// means the user opted out (sticky — don't ask again). `"failed"`
/// means the last attempt errored and we should ask the user on next
/// interactive launch whether to retry.
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct BandwidthEntry {
#[serde(default = "default_status")]
pub status: BandwidthStatus,
#[serde(default)]
pub upstream_mbps: Option<f64>,
#[serde(default)]
pub measured_at: Option<DateTime<Utc>>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum BandwidthStatus {
Unmeasured,
Measured,
Skipped,
Failed,
}
impl Default for BandwidthStatus {
fn default() -> Self {
Self::Unmeasured
}
}
fn default_status() -> BandwidthStatus {
BandwidthStatus::Unmeasured
}
/// Returns `~/.config/pixelpass/config.toml` (or the XDG equivalent on other
/// platforms). The parent directory is created lazily by [`save`].
pub fn config_path() -> Result<PathBuf> {
let dirs = ProjectDirs::from("", "", "pixelpass")
.context("could not locate a config directory for pixelpass")?;
Ok(dirs.config_dir().join("config.toml"))
}
/// Returns the loaded config, or a `Default` instance if the file doesn't
/// exist yet. Bubble up parse errors so we don't silently overwrite a
/// hand-edited config the user is debugging.
pub fn load() -> Result<Config> {
let path = config_path()?;
match fs::read_to_string(&path) {
Ok(s) => toml::from_str::<Config>(&s)
.with_context(|| format!("failed to parse {}", path.display())),
Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(Config::default()),
Err(e) => Err(e).with_context(|| format!("failed to read {}", path.display())),
}
}
/// Atomic write via tempfile-in-same-dir + rename.
pub fn save(cfg: &Config) -> Result<()> {
let path = config_path()?;
let parent = path
.parent()
.context("config path has no parent directory")?;
fs::create_dir_all(parent)
.with_context(|| format!("failed to create {}", parent.display()))?;
let serialized =
toml::to_string_pretty(cfg).context("failed to serialize config to TOML")?;
let tmp = parent.join(format!(".config.toml.tmp.{}", std::process::id()));
{
let mut f = fs::File::create(&tmp)
.with_context(|| format!("failed to create {}", tmp.display()))?;
f.write_all(serialized.as_bytes())
.with_context(|| format!("failed to write {}", tmp.display()))?;
f.sync_all().ok();
}
fs::rename(&tmp, &path)
.with_context(|| format!("failed to rename {} -> {}", tmp.display(), path.display()))?;
Ok(())
}
+2
View File
@@ -1,4 +1,6 @@
pub mod alpn; pub mod alpn;
pub mod bandwidth;
pub mod config;
pub mod deps; pub mod deps;
pub mod display; pub mod display;
pub mod process; pub mod process;
+221 -38
View File
@@ -5,10 +5,28 @@ use anyhow::{Result, bail};
use iroh::Endpoint; use iroh::Endpoint;
use iroh::endpoint::{Connection, presets}; use iroh::endpoint::{Connection, presets};
use iroh_tickets::endpoint::EndpointTicket; use iroh_tickets::endpoint::EndpointTicket;
use std::time::Duration;
use tokio::sync::{mpsc, oneshot};
use tokio_util::sync::CancellationToken; use tokio_util::sync::CancellationToken;
use crate::cli::HostOpts; use crate::cli::HostOpts;
use crate::common::{alpn::ALPN, deps, display::DisplayServer, signal}; use crate::common::{
alpn::ALPN, bandwidth, config, config::BandwidthStatus, deps, display::DisplayServer, signal,
tunnel,
};
use self::capture::CaptureHandle;
/// Messages from per-viewer tasks to the capture supervisor.
enum SupervisorMsg {
/// A new viewer wants in. Supervisor replies with the local capture
/// HTTP port to connect to, or an error string if the host is full or
/// capture spawn failed.
AddViewer(oneshot::Sender<Result<u16, String>>),
/// A viewer's session ended. Supervisor decrements the count and tears
/// down capture if it just hit zero.
RemoveViewer,
}
pub async fn run(opts: HostOpts) -> Result<()> { pub async fn run(opts: HostOpts) -> Result<()> {
let display = DisplayServer::resolve(opts.display_server); let display = DisplayServer::resolve(opts.display_server);
@@ -21,6 +39,11 @@ pub async fn run(opts: HostOpts) -> Result<()> {
); );
} }
let resolution = resolve_max_viewers(&opts);
if resolution.value == 0 {
bail!("--max-viewers must be at least 1");
}
let cancel = signal::install_ctrl_c(); let cancel = signal::install_ctrl_c();
let endpoint = Endpoint::builder(presets::N0) let endpoint = Endpoint::builder(presets::N0)
@@ -31,73 +54,177 @@ pub async fn run(opts: HostOpts) -> Result<()> {
let addr = endpoint.addr(); let addr = endpoint.addr();
let ticket = EndpointTicket::new(addr); let ticket = EndpointTicket::new(addr);
let clipboard_ok = opts.interactive && copy_to_clipboard(&ticket.to_string()); let clipboard_ok = opts.interactive && copy_to_clipboard(&ticket.to_string());
print_host_banner(&ticket, display, &opts, clipboard_ok); print_host_banner(&ticket, display, &opts, &resolution, clipboard_ok);
let result = accept_loop(&endpoint, display, &opts, cancel.clone()).await; let (sup_tx, sup_rx) = mpsc::channel::<SupervisorMsg>(16);
let supervisor = tokio::spawn(supervise(opts.clone(), display, resolution.value, sup_rx));
accept_loop(&endpoint, sup_tx.clone(), cancel.clone()).await;
drop(sup_tx);
let _ = supervisor.await;
endpoint.close().await; endpoint.close().await;
result Ok(())
} }
async fn accept_loop( async fn accept_loop(
endpoint: &Endpoint, endpoint: &Endpoint,
display: DisplayServer, sup_tx: mpsc::Sender<SupervisorMsg>,
opts: &HostOpts,
cancel: CancellationToken, cancel: CancellationToken,
) -> Result<()> { ) {
tokio::select! { loop {
_ = cancel.cancelled() => { tokio::select! {
tracing::info!("cancellation requested before any peer connected"); _ = cancel.cancelled() => {
Ok(()) tracing::info!("cancellation requested — closing accept loop");
} return;
accepted = endpoint.accept() => { }
let Some(incoming) = accepted else { accepted = endpoint.accept() => {
bail!("endpoint stopped accepting connections"); let Some(incoming) = accepted else {
}; tracing::info!("endpoint stopped accepting connections");
let conn = incoming.await?; return;
let remote = conn.remote_id(); };
tracing::info!(%remote, "peer connected"); let conn = match incoming.await {
eprintln!("\n[pixelpass] peer connected: {remote}\n"); Ok(c) => c,
handle_peer(conn, display, opts, cancel).await Err(e) => {
tracing::warn!("incoming connection failed: {e:#}");
continue;
}
};
let sup_tx = sup_tx.clone();
let cancel = cancel.clone();
tokio::spawn(handle_peer(conn, sup_tx, cancel));
}
} }
} }
} }
async fn handle_peer( async fn handle_peer(
conn: Connection, conn: Connection,
display: DisplayServer, sup_tx: mpsc::Sender<SupervisorMsg>,
opts: &HostOpts,
cancel: CancellationToken, cancel: CancellationToken,
) -> Result<()> { ) {
let (quic_send, quic_recv) = conn.accept_bi().await?; let remote = conn.remote_id();
let capture_handle = capture::spawn(display, opts).await?; let (reply_tx, reply_rx) = oneshot::channel();
let port = capture_handle.local_port(); if sup_tx.send(SupervisorMsg::AddViewer(reply_tx)).await.is_err() {
let tcp = wayland::connect_to_capture(port, std::time::Duration::from_secs(5)).await?; tracing::warn!(%remote, "supervisor channel closed; dropping peer");
return;
}
let port = match reply_rx.await {
Ok(Ok(p)) => p,
Ok(Err(reason)) => {
tracing::warn!(%remote, %reason, "refusing viewer");
eprintln!("[pixelpass] refusing viewer {remote}: {reason}");
return;
}
Err(_) => {
tracing::warn!(%remote, "supervisor reply dropped; dropping peer");
return;
}
};
let bridge = crate::common::tunnel::bridge(quic_send, quic_recv, tcp); let (quic_send, quic_recv) = match conn.accept_bi().await {
Ok(s) => s,
Err(e) => {
tracing::warn!(%remote, "accept_bi failed: {e:#}");
let _ = sup_tx.send(SupervisorMsg::RemoveViewer).await;
return;
}
};
eprintln!("[pixelpass] viewer connected: {remote}");
let tcp = match wayland::connect_to_capture(port, Duration::from_secs(5)).await {
Ok(t) => t,
Err(e) => {
tracing::warn!(%remote, "connect_to_capture failed: {e:#}");
let _ = sup_tx.send(SupervisorMsg::RemoveViewer).await;
return;
}
};
let bridge = tunnel::bridge(quic_send, quic_recv, tcp);
tokio::select! { tokio::select! {
res = bridge => { res = bridge => {
if let Err(e) = res { if let Err(e) = res {
tracing::warn!("bridge ended with error: {e:#}"); tracing::warn!(%remote, "bridge ended with error: {e:#}");
} else { } else {
tracing::info!("bridge closed cleanly"); tracing::info!(%remote, "bridge closed cleanly");
} }
} }
_ = cancel.cancelled() => { _ = cancel.cancelled() => {
tracing::info!("cancellation requested during stream"); tracing::info!(%remote, "cancellation during stream");
} }
} }
capture_handle.shutdown().await; eprintln!("[pixelpass] viewer disconnected: {remote}");
Ok(()) let _ = sup_tx.send(SupervisorMsg::RemoveViewer).await;
}
/// Owns the single shared CaptureHandle and the active viewer count. Spawns
/// capture lazily on the first AddViewer; tears it down when the count drops
/// back to zero. Enforces the max-viewers cap by refusing AddViewer when
/// the count is already at the cap.
async fn supervise(
opts: HostOpts,
display: DisplayServer,
max_viewers: u32,
mut rx: mpsc::Receiver<SupervisorMsg>,
) {
let mut handle: Option<CaptureHandle> = None;
let mut count: u32 = 0;
while let Some(msg) = rx.recv().await {
match msg {
SupervisorMsg::AddViewer(reply) => {
if count >= max_viewers {
let _ = reply.send(Err(format!(
"host is full ({count} of {max_viewers} viewers connected)"
)));
continue;
}
if handle.is_none() {
tracing::info!("first viewer arriving — spawning capture");
match capture::spawn(display, &opts).await {
Ok(h) => handle = Some(h),
Err(e) => {
let _ = reply.send(Err(format!("capture spawn failed: {e:#}")));
continue;
}
}
}
let port = handle.as_ref().expect("handle was just set").local_port();
count += 1;
let _ = reply.send(Ok(port));
tracing::info!(active = count, cap = max_viewers, "viewer joined");
}
SupervisorMsg::RemoveViewer => {
count = count.saturating_sub(1);
tracing::info!(active = count, cap = max_viewers, "viewer left");
if count == 0
&& let Some(h) = handle.take()
{
tracing::info!("last viewer left — tearing down capture");
h.shutdown().await;
}
}
}
}
if let Some(h) = handle.take() {
tracing::info!("host shutdown — tearing down capture");
h.shutdown().await;
}
} }
fn print_host_banner( fn print_host_banner(
ticket: &EndpointTicket, ticket: &EndpointTicket,
display: DisplayServer, display: DisplayServer,
opts: &HostOpts, opts: &HostOpts,
resolution: &MaxViewersResolution,
clipboard_ok: bool, clipboard_ok: bool,
) { ) {
eprintln!(); eprintln!();
@@ -106,23 +233,79 @@ fn print_host_banner(
eprintln!("│ capture : {}", capture_summary(opts)); eprintln!("│ capture : {}", capture_summary(opts));
eprintln!("│ bitrate / fps : {} kbps @ {} fps", opts.bitrate, opts.framerate); eprintln!("│ bitrate / fps : {} kbps @ {} fps", opts.bitrate, opts.framerate);
eprintln!("│ hw encode : {}", if opts.no_hwencode { "off" } else { "auto (VAAPI if available)" }); eprintln!("│ hw encode : {}", if opts.no_hwencode { "off" } else { "auto (VAAPI if available)" });
eprintln!("│ max viewers : {} ({})", resolution.value, resolution.source.label());
eprintln!(""); eprintln!("");
if clipboard_ok { if clipboard_ok {
eprintln!("│ Your share code has been copied to your clipboard."); eprintln!("│ Your share code has been copied to your clipboard.");
eprintln!("│ Send it to your viewer. (If clipboard didn't work, the"); eprintln!("│ Send it to your viewer(s). (If clipboard didn't work, the");
eprintln!("│ code is also shown below for manual copy.)"); eprintln!("│ code is also shown below for manual copy.)");
} else { } else {
eprintln!("│ Share this ticket with your viewer:"); eprintln!("│ Share this ticket with your viewer(s):");
} }
eprintln!(""); eprintln!("");
eprintln!("│ pixelpass {ticket}"); eprintln!("│ pixelpass {ticket}");
eprintln!(""); eprintln!("");
eprintln!("│ Capture will not start until the viewer connects."); eprintln!("│ Capture starts when the first viewer connects, runs while");
eprintln!("Press Ctrl+C to stop."); eprintln!("any viewer is connected, and tears down when the last one");
eprintln!("│ leaves. Press Ctrl+C to stop the host entirely.");
eprintln!("└────────────────────────────────────────────────────────────"); eprintln!("└────────────────────────────────────────────────────────────");
eprintln!(); eprintln!();
} }
/// How we arrived at the final viewer cap. Surfaced in the banner so the
/// user can tell at a glance whether the number is what they specified,
/// what their measured upstream supports, or just the fallback default.
struct MaxViewersResolution {
value: u32,
source: MaxViewersSource,
}
enum MaxViewersSource {
/// User passed --max-viewers explicitly.
UserFlag,
/// Derived from the saved bandwidth measurement.
BandwidthMeasurement { safe_mbps: f64 },
/// No flag, no measurement — falling back.
DefaultFallback,
}
impl MaxViewersSource {
fn label(&self) -> String {
match self {
MaxViewersSource::UserFlag => "user-specified".to_string(),
MaxViewersSource::BandwidthMeasurement { safe_mbps } => {
format!("auto: {safe_mbps:.1} Mbps measured upstream")
}
MaxViewersSource::DefaultFallback => {
"default — run `pixelpass --reconfigure` for a connection-aware value".to_string()
}
}
}
}
fn resolve_max_viewers(opts: &HostOpts) -> MaxViewersResolution {
if let Some(n) = opts.max_viewers {
return MaxViewersResolution {
value: n,
source: MaxViewersSource::UserFlag,
};
}
if let Ok(cfg) = config::load()
&& cfg.bandwidth.status == BandwidthStatus::Measured
&& let Some(upstream) = cfg.bandwidth.upstream_mbps
{
let n = bandwidth::recommended_max_viewers(upstream, opts.bitrate);
return MaxViewersResolution {
value: n,
source: MaxViewersSource::BandwidthMeasurement { safe_mbps: upstream },
};
}
MaxViewersResolution {
value: 2,
source: MaxViewersSource::DefaultFallback,
}
}
fn copy_to_clipboard(text: &str) -> bool { fn copy_to_clipboard(text: &str) -> bool {
match arboard::Clipboard::new().and_then(|mut cb| cb.set_text(text.to_owned())) { match arboard::Clipboard::new().and_then(|mut cb| cb.set_text(text.to_owned())) {
Ok(()) => true, Ok(()) => true,
+94 -17
View File
@@ -1,7 +1,10 @@
//! Wayland capture: ashpd ScreenCast portal → PipeWire fd → gst-launch //! Wayland capture: ashpd ScreenCast portal → PipeWire fd → gst-launch
//! pipewiresrc → MPEG-TS on gst stdout → in-process HTTP server bound on a //! pipewiresrc → MPEG-TS on gst stdout → in-process HTTP server bound on a
//! random localhost port. The host bridge TCP-connects to that server and //! random localhost port. One gst child feeds a tokio::sync::broadcast channel;
//! pumps bytes to QUIC. //! the HTTP listener accepts multiple connections and each one drains its own
//! fresh broadcast::Receiver — so a single capture pipeline fans out to N
//! concurrent viewers. Slow consumers see Lagged and skip ahead; the MPEG-TS
//! stream resyncs at the next keyframe.
use anyhow::{Context, Result, bail}; use anyhow::{Context, Result, bail};
use ashpd::{ use ashpd::{
@@ -16,18 +19,30 @@ use nix::sys::signal::{Signal, kill};
use nix::unistd::{Pid, close}; use nix::unistd::{Pid, close};
use std::os::fd::{AsFd, IntoRawFd, OwnedFd, RawFd}; use std::os::fd::{AsFd, IntoRawFd, OwnedFd, RawFd};
use std::process::Stdio; use std::process::Stdio;
use std::sync::Arc;
use std::time::Duration; use std::time::Duration;
use tokio::io::{AsyncReadExt, AsyncWriteExt}; use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::{TcpListener, TcpStream}; use tokio::net::{TcpListener, TcpStream};
use tokio::process::{Child, ChildStdout, Command}; use tokio::process::{Child, ChildStdout, Command};
use tokio::sync::broadcast;
use tokio::task::JoinHandle; use tokio::task::JoinHandle;
use tokio::time::{Instant, sleep, timeout}; use tokio::time::{Instant, sleep, timeout};
use crate::cli::HostOpts; use crate::cli::HostOpts;
/// Broadcast-channel capacity in chunks. Each chunk is up to 64 KiB from gst
/// stdout, so 16 chunks ≈ 1 MiB ≈ ~2 s of buffered jitter at the default
/// 4 Mbps bitrate. A viewer that falls behind by more than this gets Lagged
/// and skips ahead — MPEG-TS resyncs at the next keyframe.
const FANOUT_CAPACITY: usize = 16;
/// Size of each chunk read from gst stdout.
const READ_CHUNK: usize = 64 * 1024;
pub struct CaptureHandle { pub struct CaptureHandle {
port: u16, port: u16,
gst: Option<Child>, gst: Option<Child>,
reader: Option<JoinHandle<()>>,
server: Option<JoinHandle<()>>, server: Option<JoinHandle<()>>,
} }
@@ -37,8 +52,8 @@ impl CaptureHandle {
} }
/// Graceful teardown: SIGTERM gst, give it ~1s to exit, then SIGKILL, then /// Graceful teardown: SIGTERM gst, give it ~1s to exit, then SIGKILL, then
/// abort the HTTP server task. Call this before dropping; Drop only fires /// abort the reader + accept-loop tasks. Call this before dropping; Drop
/// the kill backstop. /// only fires the kill backstop.
pub async fn shutdown(mut self) { pub async fn shutdown(mut self) {
if let Some(child) = self.gst.as_mut() if let Some(child) = self.gst.as_mut()
&& let Some(pid) = child.id() && let Some(pid) = child.id()
@@ -49,6 +64,9 @@ impl CaptureHandle {
let _ = timeout(Duration::from_millis(1000), child.wait()).await; let _ = timeout(Duration::from_millis(1000), child.wait()).await;
let _ = child.start_kill(); let _ = child.start_kill();
} }
if let Some(task) = self.reader.take() {
task.abort();
}
if let Some(task) = self.server.take() { if let Some(task) = self.server.take() {
task.abort(); task.abort();
} }
@@ -60,6 +78,9 @@ impl Drop for CaptureHandle {
if let Some(child) = self.gst.as_mut() { if let Some(child) = self.gst.as_mut() {
let _ = child.start_kill(); let _ = child.start_kill();
} }
if let Some(task) = self.reader.as_ref() {
task.abort();
}
if let Some(task) = self.server.as_ref() { if let Some(task) = self.server.as_ref() {
task.abort(); task.abort();
} }
@@ -199,28 +220,68 @@ pub async fn start(opts: &HostOpts) -> Result<CaptureHandle> {
.take() .take()
.context("gst-launch-1.0 stdout pipe unavailable")?; .context("gst-launch-1.0 stdout pipe unavailable")?;
// 4. Spawn the HTTP server task. It owns the listener + gst stdout: it // 4. Set up the broadcast fanout. The reader task pumps gst stdout chunks
// accepts one client (the host's bridge socket via connect_to_capture), // into the channel; the accept-loop task spawns one sender task per
// drains the HTTP request, writes a fixed MPEG-TS response, then // accepted TCP connection, each draining a fresh broadcast::Receiver.
// copies gst stdout to the socket forever. let (tx, _) = broadcast::channel::<Arc<Vec<u8>>>(FANOUT_CAPACITY);
let server = tokio::spawn(serve_capture(listener, gst_stdout));
let reader = tokio::spawn(pump_gst_to_broadcast(gst_stdout, tx.clone()));
let server = tokio::spawn(run_accept_loop(listener, tx));
Ok(CaptureHandle { Ok(CaptureHandle {
port, port,
gst: Some(gst), gst: Some(gst),
reader: Some(reader),
server: Some(server), server: Some(server),
}) })
} }
async fn serve_capture(listener: TcpListener, mut gst_stdout: ChildStdout) { /// Reads gst's stdout in chunks and broadcasts each to all current subscribers.
let mut sock = match listener.accept().await { /// `broadcast::send` returns Err when there are no receivers; we ignore it and
Ok((s, _)) => s, /// keep reading so gst doesn't backpressure waiting for a viewer.
Err(e) => { async fn pump_gst_to_broadcast(
tracing::warn!("capture HTTP accept failed: {e}"); mut gst_stdout: ChildStdout,
return; tx: broadcast::Sender<Arc<Vec<u8>>>,
) {
let mut buf = vec![0u8; READ_CHUNK];
loop {
match gst_stdout.read(&mut buf).await {
Ok(0) => {
tracing::info!("gst stdout EOF — fanout reader exiting");
return;
}
Ok(n) => {
let chunk = Arc::new(buf[..n].to_vec());
let _ = tx.send(chunk);
}
Err(e) => {
tracing::warn!("gst stdout read error: {e}");
return;
}
} }
}; }
}
/// Accepts TCP connections on the local capture port forever. Each accepted
/// connection becomes its own viewer-serving task with a private receiver.
async fn run_accept_loop(listener: TcpListener, tx: broadcast::Sender<Arc<Vec<u8>>>) {
loop {
let sock = match listener.accept().await {
Ok((s, _)) => s,
Err(e) => {
tracing::warn!("capture HTTP accept failed: {e}");
return;
}
};
let rx = tx.subscribe();
tokio::spawn(serve_one_viewer(sock, rx));
}
}
/// Drains the HTTP request, writes a fixed 200 OK, then pumps broadcast
/// chunks to the socket until the channel closes or the socket errors out.
/// On Lagged (slow consumer), skip ahead — MPEG-TS recovers at next keyframe.
async fn serve_one_viewer(mut sock: TcpStream, mut rx: broadcast::Receiver<Arc<Vec<u8>>>) {
if !drain_http_request(&mut sock).await { if !drain_http_request(&mut sock).await {
return; return;
} }
@@ -234,7 +295,23 @@ async fn serve_capture(listener: TcpListener, mut gst_stdout: ChildStdout) {
return; return;
} }
let _ = tokio::io::copy(&mut gst_stdout, &mut sock).await; loop {
match rx.recv().await {
Ok(chunk) => {
if sock.write_all(&chunk).await.is_err() {
return;
}
}
Err(broadcast::error::RecvError::Lagged(skipped)) => {
tracing::warn!(
skipped,
"viewer fanout lagged — MPEG-TS will resync at next keyframe"
);
continue;
}
Err(broadcast::error::RecvError::Closed) => return,
}
}
} }
async fn drain_http_request(sock: &mut TcpStream) -> bool { async fn drain_http_request(sock: &mut TcpStream) -> bool {
+137 -1
View File
@@ -4,6 +4,7 @@ use iroh_tickets::endpoint::EndpointTicket;
use std::str::FromStr; use std::str::FromStr;
use crate::cli::Cli; use crate::cli::Cli;
use crate::common::{bandwidth, config};
use crate::{host, viewer}; use crate::{host, viewer};
pub async fn run(cli: Cli) -> Result<()> { pub async fn run(cli: Cli) -> Result<()> {
@@ -20,7 +21,10 @@ pub async fn run(cli: Cli) -> Result<()> {
.interact()?; .interact()?;
match choice { match choice {
0 => host::run(cli.into_host_opts(true)).await, 0 => {
preflight_if_needed(&theme).await;
host::run(cli.into_host_opts(true)).await
}
_ => { _ => {
let ticket = prompt_ticket(&theme)?; let ticket = prompt_ticket(&theme)?;
viewer::run(ticket, cli.into_viewer_opts(true)).await viewer::run(ticket, cli.into_viewer_opts(true)).await
@@ -28,6 +32,138 @@ pub async fn run(cli: Cli) -> Result<()> {
} }
} }
/// `pixelpass --reconfigure` entry point: unconditionally re-run the
/// bandwidth pre-flight test, save the result, and return. Used to
/// refresh a stale measurement (e.g. user moved house, changed ISP).
pub async fn run_reconfigure() -> Result<()> {
eprintln!();
eprintln!("Re-running bandwidth pre-flight test…");
let mut cfg = config::load().unwrap_or_default();
run_bandwidth_test(&mut cfg).await;
Ok(())
}
/// First-run pre-flight gate. Called once, when the user picks "Host" in
/// the interactive menu. Behavior by saved status:
/// - Unmeasured (first ever launch): explain + offer Run / Skip
/// - Failed (previous attempt errored): offer Retry / give-up-and-skip
/// - Measured or Skipped: silent — never re-prompts
async fn preflight_if_needed(theme: &ColorfulTheme) {
let mut cfg = config::load().unwrap_or_default();
match cfg.bandwidth.status {
config::BandwidthStatus::Measured | config::BandwidthStatus::Skipped => return,
config::BandwidthStatus::Unmeasured => {
eprintln!();
eprintln!("First-time setup");
eprintln!("────────────────");
eprintln!("PixelPass can measure your upload speed to recommend a safe");
eprintln!("default for how many viewers your connection can handle.");
eprintln!("The test takes about 5 seconds and uploads ~5 MB to");
eprintln!("Cloudflare's open speed-test endpoint.");
eprintln!();
eprintln!("If you skip, a conservative default (2 viewers) is used.");
eprintln!("You can run the test later with `pixelpass --reconfigure`.");
eprintln!();
let Ok(choice) = Select::with_theme(theme)
.with_prompt("What would you like to do?")
.items(&[
"Run the bandwidth test (recommended)",
"Skip — use the conservative default",
])
.default(0)
.interact()
else {
return;
};
if choice == 1 {
cfg.bandwidth = config::BandwidthEntry {
status: config::BandwidthStatus::Skipped,
upstream_mbps: None,
measured_at: None,
};
let _ = config::save(&cfg);
eprintln!("Pre-flight skipped.");
return;
}
run_bandwidth_test(&mut cfg).await;
}
config::BandwidthStatus::Failed => {
eprintln!();
let Ok(choice) = Select::with_theme(theme)
.with_prompt("Last bandwidth test failed. Try again?")
.items(&[
"Yes — retry now",
"No — use the conservative default",
])
.default(0)
.interact()
else {
return;
};
if choice == 1 {
cfg.bandwidth = config::BandwidthEntry {
status: config::BandwidthStatus::Skipped,
upstream_mbps: None,
measured_at: None,
};
let _ = config::save(&cfg);
eprintln!("OK — using the conservative default.");
return;
}
run_bandwidth_test(&mut cfg).await;
}
}
}
async fn run_bandwidth_test(cfg: &mut config::Config) {
eprintln!();
eprintln!("Measuring upstream…");
let result = tokio::task::spawn_blocking(bandwidth::measure_upstream_blocking).await;
let measurement = match result {
Ok(Ok(m)) => m,
Ok(Err(e)) => {
eprintln!("Test failed: {e:#}");
eprintln!("Marking as failed — you'll be asked again on next launch.");
cfg.bandwidth = config::BandwidthEntry {
status: config::BandwidthStatus::Failed,
upstream_mbps: None,
measured_at: None,
};
let _ = config::save(cfg);
return;
}
Err(join_err) => {
eprintln!("Test task panicked: {join_err}");
cfg.bandwidth = config::BandwidthEntry {
status: config::BandwidthStatus::Failed,
upstream_mbps: None,
measured_at: None,
};
let _ = config::save(cfg);
return;
}
};
eprintln!(
"Measured {:.2} Mbps up (safe estimate {:.2} Mbps, took {:.1}s).",
measurement.raw_mbps,
measurement.safe_mbps,
measurement.elapsed.as_secs_f64()
);
cfg.bandwidth = config::BandwidthEntry {
status: config::BandwidthStatus::Measured,
upstream_mbps: Some(measurement.safe_mbps),
measured_at: Some(chrono::Utc::now()),
};
if let Err(e) = config::save(cfg) {
eprintln!("Warning: failed to save result: {e:#}");
}
}
fn print_welcome() { fn print_welcome() {
eprintln!(); eprintln!();
eprintln!("Welcome to PixelPass."); eprintln!("Welcome to PixelPass.");
+4
View File
@@ -20,6 +20,10 @@ async fn main() -> Result<()> {
return repair::run().await; return repair::run().await;
} }
if cli.reconfigure {
return interactive::run_reconfigure().await;
}
match cli.ticket.as_deref() { match cli.ticket.as_deref() {
Some(s) => { Some(s) => {
let ticket: EndpointTicket = s.parse().map_err(|e| { let ticket: EndpointTicket = s.parse().map_err(|e| {