Compare commits
4
Commits
c427231858
...
v0.6.6
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
0588d92537 | ||
|
|
b4a4c00711 | ||
|
|
8c4f4a0b8b | ||
|
|
76c4f68bb3 |
@@ -4,6 +4,29 @@ All notable changes to PeerSpeak are documented here.
|
||||
|
||||
## [Unreleased]
|
||||
|
||||
## [0.6.6] — 2026-07-19
|
||||
|
||||
### Fixed
|
||||
- **A screen share that falls behind now catches back up.** On a lossy
|
||||
connection (satellite links are the worst case) the share could settle several
|
||||
seconds behind the host and simply stay there for the rest of the call. The
|
||||
viewer now notices a deep buffer and plays imperceptibly fast until it is back
|
||||
at the live edge — the audio stays in tune and in sync while it does. This
|
||||
replaces the previous attempt at the problem, which measurement showed did not
|
||||
help. Applies to the Low latency setting; Smooth intentionally keeps its
|
||||
larger buffer.
|
||||
|
||||
### Changed
|
||||
- **Low latency now keeps a tighter viewer buffer.** The screen-share cache
|
||||
setting is a size in megabytes, which at a given bitrate quietly decides how
|
||||
many *seconds* behind a viewer can drift — a 2 MB buffer turned out to hold
|
||||
about six seconds of a typical share. Low latency now caps that buffer at 1 MB
|
||||
regardless of the setting, which halved how far behind a share fell on a bad
|
||||
connection before anything else kicked in. Smooth still honors the value you
|
||||
choose, since a deep buffer is the point of that mode.
|
||||
|
||||
## [0.6.5] — 2026-07-19
|
||||
|
||||
### Added
|
||||
- **Chat message sounds.** Successful outgoing messages and admitted incoming
|
||||
messages now have distinct notification chimes, each with its own enable
|
||||
@@ -15,6 +38,12 @@ All notable changes to PeerSpeak are documented here.
|
||||
- **Notification sound browser.** Every notification event now has a native
|
||||
Browse button for choosing a custom WAV instead of typing its path manually.
|
||||
|
||||
### Changed
|
||||
- **Tidier per-participant audio controls.** The equalizer bands and noise gate
|
||||
for each participant now live behind an **"Advanced audio"** foldout instead
|
||||
of being expanded all the time, so a call with several people no longer fills
|
||||
the panel with sliders. The controls themselves are unchanged.
|
||||
|
||||
### Fixed
|
||||
- **Low-latency screen sharing stays near the live edge again.** mpv's
|
||||
timestamp pacing could let stale frames accumulate across the reliable
|
||||
|
||||
Generated
+1
-1
@@ -4871,7 +4871,7 @@ checksum = "35fb2e5f958ec131621fdd531e9fc186ed768cbe395337403ae56c17a74c68ec"
|
||||
|
||||
[[package]]
|
||||
name = "peerspeak"
|
||||
version = "0.6.4"
|
||||
version = "0.6.6"
|
||||
dependencies = [
|
||||
"anyhow",
|
||||
"async-trait",
|
||||
|
||||
+1
-1
@@ -1,6 +1,6 @@
|
||||
[package]
|
||||
name = "peerspeak"
|
||||
version = "0.6.4"
|
||||
version = "0.6.6"
|
||||
edition = "2024"
|
||||
description = "Decentralized peer-to-peer voice chat (Rust/iroh/PipeWire/Opus/iced)"
|
||||
license = "MIT"
|
||||
|
||||
@@ -12,7 +12,7 @@
|
||||
; (x86_64-pc-windows-gnu, statically linked -- no extra DLLs needed).
|
||||
|
||||
#define MyAppName "PeerSpeak"
|
||||
#define MyAppVersion "0.6.4"
|
||||
#define MyAppVersion "0.6.6"
|
||||
#define MyAppPublisher "mollusk"
|
||||
#define MyAppExeName "peerspeak.exe"
|
||||
|
||||
|
||||
@@ -0,0 +1,314 @@
|
||||
//! Live-edge catch-up for the screen-share viewer.
|
||||
//!
|
||||
//! PixelPass carries the share as MPEG-TS over a reliable, ordered transport. On
|
||||
//! a lossy link (satellite handovers are the pathological case) every loss burst
|
||||
//! becomes retransmission plus head-of-line blocking, and the viewer absorbs the
|
||||
//! stall as buffered latency. Nothing in the chain ever trims that buffer back,
|
||||
//! so the picture ends up seconds behind the host and stays there.
|
||||
//!
|
||||
//! Measured on a `tc netem` rig that simulates a satellite link (40 ms +/- 20 ms
|
||||
//! jitter, 0.5% loss, a 250 ms/30%-loss handover burst every 15 s): a viewer with
|
||||
//! ordinary timestamp pacing settles ~1.24 s behind. mpv's `--untimed` does NOT
|
||||
//! help (~1.38 s, marginally worse) because it only removes pacing at
|
||||
//! *presentation* while audio still drains at 1x the DAC rate, so an accumulated
|
||||
//! buffer never shrinks. Returning to the live edge requires consuming the
|
||||
//! backlog faster than it arrives.
|
||||
//!
|
||||
//! So we nudge playback slightly faster than realtime while the buffer is deep,
|
||||
//! and drop back to 1x once it has drained. mpv's default pitch correction
|
||||
//! (`scaletempo2`) keeps a 5% speedup inaudible, and because audio and video are
|
||||
//! sped up together A/V sync is preserved — unlike `--untimed`.
|
||||
//!
|
||||
//! The control law and the JSON-IPC message handling are pure functions with
|
||||
//! tests; the only I/O is [`drive`], which talks to mpv's `--input-ipc-server`
|
||||
//! socket.
|
||||
|
||||
use std::path::{Path, PathBuf};
|
||||
use std::time::Duration;
|
||||
|
||||
/// Buffer depth (seconds) above which we start draining.
|
||||
pub const CACHE_HIGH_S: f64 = 1.0;
|
||||
/// Buffer depth (seconds) below which we return to realtime.
|
||||
pub const CACHE_LOW_S: f64 = 0.4;
|
||||
/// The buffer depth we aim to sit at; the drain rate is proportional to how far
|
||||
/// above this the buffer actually is.
|
||||
pub const CACHE_TARGET_S: f64 = 0.5;
|
||||
/// Extra playback rate per second of excess buffer.
|
||||
pub const CATCHUP_GAIN: f64 = 0.05;
|
||||
/// Hard ceiling on the drain rate. Beyond this the speedup stops being
|
||||
/// unnoticeable, and a share that far behind is better served by the operator
|
||||
/// restarting it than by a chipmunk impression.
|
||||
pub const MAX_CATCHUP_SPEED: f64 = 1.15;
|
||||
/// Normal realtime playback.
|
||||
pub const NORMAL_SPEED: f64 = 1.0;
|
||||
/// How often we sample the buffer depth.
|
||||
pub const POLL_INTERVAL: Duration = Duration::from_millis(500);
|
||||
/// Smallest rate change worth sending to the player.
|
||||
pub const SPEED_EPSILON: f64 = 0.005;
|
||||
|
||||
/// The property we watch on the viewer.
|
||||
const CACHE_PROPERTY: &str = "demuxer-cache-duration";
|
||||
|
||||
/// Decide the playback rate for the next interval.
|
||||
///
|
||||
/// Proportional, because a fixed small speedup cannot recover a large backlog in
|
||||
/// any reasonable time: draining 6 s at 1.05x takes two minutes, which a viewer
|
||||
/// experiences as "still broken". The drain rate instead scales with how deep
|
||||
/// the buffer is, so a bad handover is cleared in tens of seconds while a small
|
||||
/// excursion still gets only a gentle, inaudible nudge.
|
||||
///
|
||||
/// Deliberately hysteretic: between [`CACHE_LOW_S`] and [`CACHE_HIGH_S`] the
|
||||
/// current rate is held, so a buffer hovering near a single threshold cannot
|
||||
/// oscillate the speed (and with it the audio pitch) every poll. Pure.
|
||||
///
|
||||
/// A non-finite reading (mpv reports `null` before playback starts, and the
|
||||
/// caller maps that to NaN) holds the current rate rather than guessing.
|
||||
pub fn catchup_speed(cache_s: f64, current: f64) -> f64 {
|
||||
if !cache_s.is_finite() {
|
||||
return current;
|
||||
}
|
||||
if cache_s < CACHE_LOW_S {
|
||||
return NORMAL_SPEED;
|
||||
}
|
||||
if cache_s <= CACHE_HIGH_S {
|
||||
return current;
|
||||
}
|
||||
let excess = cache_s - CACHE_TARGET_S;
|
||||
(NORMAL_SPEED + CATCHUP_GAIN * excess).clamp(NORMAL_SPEED, MAX_CATCHUP_SPEED)
|
||||
}
|
||||
|
||||
/// Where mpv should create its IPC socket. Kept separate from the runtime
|
||||
/// lookup so tests can pin a directory. Pure.
|
||||
pub fn socket_path(dir: &Path, token: u64) -> PathBuf {
|
||||
dir.join(format!("peerspeak-mpv-{token}.sock"))
|
||||
}
|
||||
|
||||
/// The directory for the IPC socket: the XDG runtime dir when the session
|
||||
/// provides one (tmpfs, user-private, cleaned at logout), else the temp dir.
|
||||
pub fn socket_dir() -> PathBuf {
|
||||
std::env::var_os("XDG_RUNTIME_DIR")
|
||||
.map(PathBuf::from)
|
||||
.unwrap_or_else(std::env::temp_dir)
|
||||
}
|
||||
|
||||
/// A `get_property` request for the buffer depth. Pure.
|
||||
pub fn get_cache_request(request_id: u64) -> String {
|
||||
format!(r#"{{"command":["get_property","{CACHE_PROPERTY}"],"request_id":{request_id}}}"#)
|
||||
}
|
||||
|
||||
/// A `set_property` request for the playback rate. Pure.
|
||||
pub fn set_speed_request(request_id: u64, speed: f64) -> String {
|
||||
format!(r#"{{"command":["set_property","speed",{speed}],"request_id":{request_id}}}"#)
|
||||
}
|
||||
|
||||
/// Extract the buffer depth from one line of mpv's IPC output.
|
||||
///
|
||||
/// mpv interleaves unsolicited event lines with command replies, so a line is
|
||||
/// only ours when it carries the matching `request_id`. Returns:
|
||||
/// - `Some(Some(secs))` — our reply, with a usable number,
|
||||
/// - `Some(None)` — our reply, but no number (mpv sends `"data":null` before
|
||||
/// playback starts, and reports `error` while the demuxer has no cache yet),
|
||||
/// - `None` — not our reply (an event, or another command's response).
|
||||
///
|
||||
/// Pure.
|
||||
pub fn parse_cache_response(line: &str, request_id: u64) -> Option<Option<f64>> {
|
||||
let value: serde_json::Value = serde_json::from_str(line.trim()).ok()?;
|
||||
let id = value.get("request_id")?.as_u64()?;
|
||||
if id != request_id {
|
||||
return None;
|
||||
}
|
||||
if value.get("error").and_then(|e| e.as_str()) != Some("success") {
|
||||
return Some(None);
|
||||
}
|
||||
Some(value.get("data").and_then(|d| d.as_f64()))
|
||||
}
|
||||
|
||||
/// Drive one mpv viewer's playback rate over its JSON IPC socket.
|
||||
///
|
||||
/// Runs until mpv exits (the socket dies), so it is spawned detached alongside
|
||||
/// the player and needs no shutdown signal. Every failure path just ends the
|
||||
/// task: catch-up is an optimization, and a viewer that never gets it still
|
||||
/// plays, exactly as before this existed.
|
||||
#[cfg(unix)]
|
||||
pub async fn drive(socket: PathBuf) {
|
||||
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
|
||||
use tokio::net::UnixStream;
|
||||
|
||||
// mpv creates the socket a moment after exec, so the first connects race it.
|
||||
let mut stream = None;
|
||||
for _ in 0..40 {
|
||||
match UnixStream::connect(&socket).await {
|
||||
Ok(s) => {
|
||||
stream = Some(s);
|
||||
break;
|
||||
}
|
||||
Err(_) => tokio::time::sleep(Duration::from_millis(250)).await,
|
||||
}
|
||||
}
|
||||
let Some(stream) = stream else {
|
||||
crate::log_msg("livesync: mpv IPC socket never appeared; catch-up disabled");
|
||||
return;
|
||||
};
|
||||
|
||||
let (read_half, mut write_half) = stream.into_split();
|
||||
let mut lines = BufReader::new(read_half).lines();
|
||||
let mut request_id: u64 = 0;
|
||||
let mut speed = NORMAL_SPEED;
|
||||
|
||||
loop {
|
||||
tokio::time::sleep(POLL_INTERVAL).await;
|
||||
|
||||
request_id += 1;
|
||||
let query = format!("{}\n", get_cache_request(request_id));
|
||||
if write_half.write_all(query.as_bytes()).await.is_err() {
|
||||
break;
|
||||
}
|
||||
|
||||
// Skip event lines until our reply arrives.
|
||||
let cache = loop {
|
||||
match lines.next_line().await {
|
||||
Ok(Some(line)) => {
|
||||
if let Some(value) = parse_cache_response(&line, request_id) {
|
||||
break value;
|
||||
}
|
||||
}
|
||||
// Socket closed or unreadable: mpv is gone.
|
||||
_ => return,
|
||||
}
|
||||
};
|
||||
|
||||
let cache = cache.unwrap_or(f64::NAN);
|
||||
let next = catchup_speed(cache, speed);
|
||||
// A proportional law would otherwise re-send on every wobble of the
|
||||
// reading; only a change worth hearing is worth a round trip.
|
||||
if (next - speed).abs() > SPEED_EPSILON {
|
||||
speed = next;
|
||||
request_id += 1;
|
||||
let set = format!("{}\n", set_speed_request(request_id, speed));
|
||||
if write_half.write_all(set.as_bytes()).await.is_err() {
|
||||
break;
|
||||
}
|
||||
crate::log_msg(&format!(
|
||||
"livesync: cache {cache:.2}s -> playback speed {speed}x"
|
||||
));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn deep_buffer_speeds_up_and_drained_buffer_returns_to_realtime() {
|
||||
assert!(catchup_speed(1.5, NORMAL_SPEED) > NORMAL_SPEED);
|
||||
assert_eq!(catchup_speed(0.1, MAX_CATCHUP_SPEED), NORMAL_SPEED);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn drain_rate_scales_with_how_far_behind_we_are() {
|
||||
// The point of the proportional law: a small excursion gets a gentle
|
||||
// nudge, a deep backlog gets real recovery.
|
||||
let small = catchup_speed(1.5, NORMAL_SPEED);
|
||||
let large = catchup_speed(4.0, NORMAL_SPEED);
|
||||
assert!(
|
||||
large > small,
|
||||
"deeper buffer must drain faster: {small} vs {large}"
|
||||
);
|
||||
assert!(
|
||||
(small - 1.05).abs() < 1e-9,
|
||||
"1.5s buffer -> 1.05x, got {small}"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn drain_rate_is_capped_so_it_never_sounds_absurd() {
|
||||
// The ~6 s standing buffer measured on the netem rig, and far worse.
|
||||
assert_eq!(catchup_speed(6.0, NORMAL_SPEED), MAX_CATCHUP_SPEED);
|
||||
assert_eq!(catchup_speed(600.0, NORMAL_SPEED), MAX_CATCHUP_SPEED);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn hysteresis_band_holds_the_current_speed() {
|
||||
// Between the marks nothing changes, whichever side we came from —
|
||||
// this is what stops the rate (and audio pitch) oscillating.
|
||||
for cache in [CACHE_LOW_S, 0.7, CACHE_HIGH_S] {
|
||||
assert_eq!(catchup_speed(cache, NORMAL_SPEED), NORMAL_SPEED);
|
||||
assert_eq!(catchup_speed(cache, MAX_CATCHUP_SPEED), MAX_CATCHUP_SPEED);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn unknown_cache_holds_the_current_speed() {
|
||||
assert_eq!(
|
||||
catchup_speed(f64::NAN, MAX_CATCHUP_SPEED),
|
||||
MAX_CATCHUP_SPEED
|
||||
);
|
||||
assert_eq!(catchup_speed(f64::INFINITY, NORMAL_SPEED), NORMAL_SPEED);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_full_handover_cycle_drains_then_settles() {
|
||||
// Buffer grows through a loss burst, then drains as we play faster.
|
||||
let mut speed = NORMAL_SPEED;
|
||||
for cache in [0.2, 0.5, 1.2, 3.4, 1.4, 0.9, 0.6, 0.3, 0.2] {
|
||||
speed = catchup_speed(cache, speed);
|
||||
}
|
||||
assert_eq!(
|
||||
speed, NORMAL_SPEED,
|
||||
"should be back at realtime once drained"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn requests_are_valid_json_with_their_ids() {
|
||||
let get: serde_json::Value = serde_json::from_str(&get_cache_request(7)).unwrap();
|
||||
assert_eq!(get["request_id"], 7);
|
||||
assert_eq!(get["command"][0], "get_property");
|
||||
assert_eq!(get["command"][1], CACHE_PROPERTY);
|
||||
|
||||
let set: serde_json::Value = serde_json::from_str(&set_speed_request(8, 1.05)).unwrap();
|
||||
assert_eq!(set["request_id"], 8);
|
||||
assert_eq!(set["command"][0], "set_property");
|
||||
assert_eq!(set["command"][1], "speed");
|
||||
assert_eq!(set["command"][2], 1.05);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn parses_our_reply_only() {
|
||||
assert_eq!(
|
||||
parse_cache_response(r#"{"error":"success","data":1.25,"request_id":3}"#, 3),
|
||||
Some(Some(1.25))
|
||||
);
|
||||
// Another command's reply, and an unsolicited event, are not ours.
|
||||
assert_eq!(
|
||||
parse_cache_response(r#"{"error":"success","data":1.25,"request_id":4}"#, 3),
|
||||
None
|
||||
);
|
||||
assert_eq!(
|
||||
parse_cache_response(r#"{"event":"playback-restart"}"#, 3),
|
||||
None
|
||||
);
|
||||
assert_eq!(parse_cache_response("not json", 3), None);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn reply_without_a_usable_number_is_ours_but_empty() {
|
||||
// mpv before playback starts, and while the demuxer has no cache.
|
||||
assert_eq!(
|
||||
parse_cache_response(r#"{"error":"success","data":null,"request_id":1}"#, 1),
|
||||
Some(None)
|
||||
);
|
||||
assert_eq!(
|
||||
parse_cache_response(r#"{"error":"property unavailable","request_id":1}"#, 1),
|
||||
Some(None)
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn socket_path_is_scoped_to_its_token() {
|
||||
let a = socket_path(Path::new("/run/user/1000"), 42);
|
||||
assert_eq!(a, Path::new("/run/user/1000/peerspeak-mpv-42.sock"));
|
||||
assert_ne!(a, socket_path(Path::new("/run/user/1000"), 43));
|
||||
}
|
||||
}
|
||||
+155
-27
@@ -21,6 +21,8 @@ use std::time::Duration;
|
||||
use tokio::io::{AsyncBufReadExt, BufReader};
|
||||
use tokio::process::{Child, Command};
|
||||
|
||||
pub mod livesync;
|
||||
|
||||
use crate::config::{ScreenShareSettings, ShareBuffering, SharePlayer, ShareQuality};
|
||||
|
||||
/// The binary we shell out to. Looked up on `$PATH` unless a config override
|
||||
@@ -45,6 +47,12 @@ const MAX_TICKET_LEN: usize = 512;
|
||||
/// are short ("Firefox", "mpv"); this only guards against a pathological value.
|
||||
const MAX_APP_NAME_LEN: usize = 256;
|
||||
|
||||
/// Ceiling on the viewer's demuxer byte cache in the Low latency posture. The
|
||||
/// cache is a *byte* budget, so at a given bitrate it sets the worst-case
|
||||
/// backlog in seconds; keeping it tight is what stops a lossy link parking the
|
||||
/// viewer seconds behind before [`livesync`] even gets a chance to drain it.
|
||||
const LOW_LATENCY_CACHE_CAP_MB: u32 = 1;
|
||||
|
||||
/// How long to wait for the host to emit its ticket / the viewer to connect
|
||||
/// before giving up and killing the child. Startup is normally sub-second; this
|
||||
/// is only a safety net so a hung pixelpass can't wedge the caller forever.
|
||||
@@ -608,14 +616,27 @@ fn event_for_log(ev: &PixelpassEvent) -> String {
|
||||
/// its window closes.
|
||||
///
|
||||
/// The buffering posture chooses the latency/A/V-sync tradeoff. Low latency
|
||||
/// passes mpv's `--untimed`, displaying frames as soon as they decode so a live
|
||||
/// share cannot accumulate several seconds in the reliable QUIC/TCP/player
|
||||
/// buffer chain. Smooth leaves timestamp pacing enabled, which keeps a shared
|
||||
/// video's audio and video synchronized at the cost of extra live latency.
|
||||
/// Hardware decoding remains opt-in: forcing `--hwdec=auto` froze some viewers
|
||||
/// on frame 1 while audio kept playing.
|
||||
/// keeps the viewer at the live edge: mpv gets an IPC socket and [`livesync`]
|
||||
/// drains a lagging buffer by playing slightly fast (pitch-corrected, so A/V
|
||||
/// sync is preserved). Smooth leaves a deeper buffer alone, trading live
|
||||
/// latency for immunity to jitter. Hardware decoding remains opt-in: forcing
|
||||
/// `--hwdec=auto` froze some viewers on frame 1 while audio kept playing.
|
||||
fn launch_player(url: &str, settings: &ScreenShareSettings) -> std::io::Result<()> {
|
||||
let mpv_args = mpv_args(settings);
|
||||
// One socket per viewer launch, so overlapping shares can't collide on it.
|
||||
// Unix only: mpv's IPC is a named pipe on Windows, which `livesync` does not
|
||||
// speak, and an unusable socket path on the argv would help nobody.
|
||||
#[cfg(unix)]
|
||||
let ipc_socket = Some(livesync::socket_path(
|
||||
&livesync::socket_dir(),
|
||||
std::time::SystemTime::now()
|
||||
.duration_since(std::time::UNIX_EPOCH)
|
||||
.map(|d| d.as_nanos() as u64)
|
||||
.unwrap_or(0),
|
||||
));
|
||||
#[cfg(not(unix))]
|
||||
let ipc_socket: Option<PathBuf> = None;
|
||||
|
||||
let mpv_args = mpv_args(settings, ipc_socket.as_deref());
|
||||
let vlc_args = vlc_args(settings);
|
||||
let first = match settings.player {
|
||||
SharePlayer::Mpv => ("mpv", &mpv_args),
|
||||
@@ -626,15 +647,32 @@ fn launch_player(url: &str, settings: &ScreenShareSettings) -> std::io::Result<(
|
||||
SharePlayer::Vlc => ("mpv", &mpv_args),
|
||||
};
|
||||
|
||||
let child = match spawn_player(first.0, first.1, url) {
|
||||
Ok(c) => c,
|
||||
Err(_) => spawn_player(second.0, second.1, url).map_err(|_| {
|
||||
std::io::Error::new(
|
||||
std::io::ErrorKind::NotFound,
|
||||
"no media player found — install mpv or vlc to watch screen shares",
|
||||
)
|
||||
})?,
|
||||
let (launched, child) = match spawn_player(first.0, first.1, url) {
|
||||
Ok(c) => (first.0, c),
|
||||
Err(_) => (
|
||||
second.0,
|
||||
spawn_player(second.0, second.1, url).map_err(|_| {
|
||||
std::io::Error::new(
|
||||
std::io::ErrorKind::NotFound,
|
||||
"no media player found — install mpv or vlc to watch screen shares",
|
||||
)
|
||||
})?,
|
||||
),
|
||||
};
|
||||
|
||||
// Only when the socket actually reached the argv: mpv (VLC has no
|
||||
// equivalent IPC) in the Low latency posture. The driver ends by itself when
|
||||
// the player exits, so it needs no shutdown path.
|
||||
#[cfg(unix)]
|
||||
if launched == "mpv"
|
||||
&& settings.buffering == ShareBuffering::LowLatency
|
||||
&& let Some(socket) = ipc_socket
|
||||
{
|
||||
tokio::spawn(livesync::drive(socket));
|
||||
}
|
||||
#[cfg(not(unix))]
|
||||
let _ = launched;
|
||||
|
||||
tokio::spawn(async move {
|
||||
let mut child = child;
|
||||
let _ = child.wait().await;
|
||||
@@ -642,17 +680,23 @@ fn launch_player(url: &str, settings: &ScreenShareSettings) -> std::io::Result<(
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub fn mpv_args(settings: &ScreenShareSettings) -> Vec<String> {
|
||||
/// Build the argv for an mpv viewer.
|
||||
///
|
||||
/// `ipc_socket` is where mpv should expose its JSON IPC socket so [`livesync`]
|
||||
/// can drain a lagging buffer. It is wired up for Low latency only: Smooth
|
||||
/// deliberately holds a ~2 s readahead, which the catch-up thresholds would
|
||||
/// fight on every poll.
|
||||
pub fn mpv_args(settings: &ScreenShareSettings, ipc_socket: Option<&Path>) -> Vec<String> {
|
||||
let mut args = Vec::new();
|
||||
match settings.buffering {
|
||||
ShareBuffering::LowLatency => {
|
||||
args.push("--profile=low-latency".to_string());
|
||||
// Pixelpass carries MPEG-TS through reliable ordered QUIC/TCP. If
|
||||
// mpv paces a live stream even slightly slower than capture, those
|
||||
// downstream buffers retain stale frames and latency grows into
|
||||
// seconds. Immediate presentation keeps this posture at the live
|
||||
// edge; Smooth deliberately retains timestamp pacing for A/V sync.
|
||||
args.push("--untimed".to_string());
|
||||
// Pixelpass carries MPEG-TS through reliable ordered QUIC/TCP, so a
|
||||
// lossy link turns every retransmission into buffered latency that
|
||||
// nothing trims back. `--untimed` does NOT fix that (measured
|
||||
// marginally worse: it only unpaces *presentation*, while audio
|
||||
// still drains at 1x, so the backlog never shrinks) — the viewer
|
||||
// instead drains it by playing slightly fast, see `livesync`.
|
||||
args.push("--audio-buffer=0.2".to_string());
|
||||
args.push("--demuxer-readahead-secs=0.5".to_string());
|
||||
}
|
||||
@@ -661,10 +705,25 @@ pub fn mpv_args(settings: &ScreenShareSettings) -> Vec<String> {
|
||||
args.push("--demuxer-readahead-secs=2".to_string());
|
||||
}
|
||||
}
|
||||
args.push(format!("--demuxer-max-bytes={}M", settings.cache_mb));
|
||||
// The byte cap is what bounds how far behind a viewer can silently fall:
|
||||
// a demuxer allowed 2 MiB will happily sit on ~6 s of a 2.5 Mbps share (as
|
||||
// measured on the netem rig) and call it a buffer. Low latency therefore
|
||||
// gets a tighter ceiling than the user's Smooth-oriented setting, so the
|
||||
// catch-up has less to claw back after a bad patch of link.
|
||||
let cache_mb = match settings.buffering {
|
||||
ShareBuffering::LowLatency => settings.cache_mb.min(LOW_LATENCY_CACHE_CAP_MB),
|
||||
ShareBuffering::Smooth => settings.cache_mb,
|
||||
};
|
||||
args.push(format!("--demuxer-max-bytes={cache_mb}M"));
|
||||
if settings.hardware_decode {
|
||||
args.push("--hwdec=auto".to_string());
|
||||
}
|
||||
if let Some(socket) = ipc_socket
|
||||
&& settings.buffering == ShareBuffering::LowLatency
|
||||
{
|
||||
args.push(format!("--input-ipc-server={}", socket.display()));
|
||||
}
|
||||
// Extra args stay last so a user override wins over everything above.
|
||||
args.extend(split_extra_args(&settings.extra_mpv_args));
|
||||
args
|
||||
}
|
||||
@@ -830,17 +889,86 @@ mod tests {
|
||||
#[test]
|
||||
fn mpv_args_default_matches_low_latency_software_decode() {
|
||||
assert_eq!(
|
||||
mpv_args(&ScreenShareSettings::default()),
|
||||
mpv_args(&ScreenShareSettings::default(), None),
|
||||
vec![
|
||||
"--profile=low-latency",
|
||||
"--untimed",
|
||||
"--audio-buffer=0.2",
|
||||
"--demuxer-readahead-secs=0.5",
|
||||
"--demuxer-max-bytes=2M",
|
||||
"--demuxer-max-bytes=1M",
|
||||
]
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn low_latency_gets_the_ipc_socket_for_live_edge_catch_up() {
|
||||
let args = mpv_args(
|
||||
&ScreenShareSettings::default(),
|
||||
Some(Path::new("/run/user/1000/peerspeak-mpv-1.sock")),
|
||||
);
|
||||
assert!(
|
||||
args.contains(&"--input-ipc-server=/run/user/1000/peerspeak-mpv-1.sock".to_string()),
|
||||
"low latency drains a lagging buffer over mpv IPC: {args:?}"
|
||||
);
|
||||
// The flag that used to hold this posture at the live edge measured no
|
||||
// better than pacing, and cost A/V sync — it must not come back.
|
||||
assert!(!args.contains(&"--untimed".to_string()));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn smooth_keeps_its_deep_buffer_and_gets_no_ipc_socket() {
|
||||
let settings = ScreenShareSettings {
|
||||
buffering: ShareBuffering::Smooth,
|
||||
..ScreenShareSettings::default()
|
||||
};
|
||||
let args = mpv_args(
|
||||
&settings,
|
||||
Some(Path::new("/run/user/1000/peerspeak-mpv-1.sock")),
|
||||
);
|
||||
assert!(
|
||||
!args.iter().any(|a| a.starts_with("--input-ipc-server")),
|
||||
"catch-up would fight Smooth's deliberate ~2s readahead: {args:?}"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn low_latency_caps_the_byte_cache_but_smooth_keeps_the_user_value() {
|
||||
// The cache is a byte budget, so at a given bitrate it sets the
|
||||
// worst-case backlog: 2 MiB held ~6 s of a 2.5 Mbps share on the rig.
|
||||
let generous = ScreenShareSettings {
|
||||
cache_mb: 32,
|
||||
..ScreenShareSettings::default()
|
||||
};
|
||||
assert!(
|
||||
mpv_args(&generous, None)
|
||||
.contains(&format!("--demuxer-max-bytes={LOW_LATENCY_CACHE_CAP_MB}M")),
|
||||
"low latency must bound how far behind the viewer can silently fall"
|
||||
);
|
||||
|
||||
let smooth = ScreenShareSettings {
|
||||
cache_mb: 32,
|
||||
buffering: ShareBuffering::Smooth,
|
||||
..ScreenShareSettings::default()
|
||||
};
|
||||
assert!(
|
||||
mpv_args(&smooth, None).contains(&"--demuxer-max-bytes=32M".to_string()),
|
||||
"smooth is the posture where the user asked for a deep buffer"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn user_extra_args_still_come_last() {
|
||||
let settings = ScreenShareSettings {
|
||||
extra_mpv_args: "--no-osc".to_string(),
|
||||
..ScreenShareSettings::default()
|
||||
};
|
||||
let args = mpv_args(&settings, Some(Path::new("/tmp/s.sock")));
|
||||
assert_eq!(
|
||||
args.last().map(String::as_str),
|
||||
Some("--no-osc"),
|
||||
"a user override has to win over everything we add: {args:?}"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn mpv_args_smooth_hwdecode_and_extra_args_last() {
|
||||
let settings = ScreenShareSettings {
|
||||
@@ -852,7 +980,7 @@ mod tests {
|
||||
};
|
||||
|
||||
assert_eq!(
|
||||
mpv_args(&settings),
|
||||
mpv_args(&settings, None),
|
||||
vec![
|
||||
"--cache=yes",
|
||||
"--demuxer-readahead-secs=2",
|
||||
|
||||
Reference in New Issue
Block a user