feat(client,host): PyroWave Apple Metal decoder + per-mode bitrate pin
- clients/apple: native Metal wavelet decoder + compute shaders (Phase 5), decoding PyroWave without embedding MoltenVK. - pf-client-core: plumb user_flags/completeness through Decoder::decode_frame so the PyroWave backend parses chunk-aligned + partial AUs; gate the param's unused-warning to exactly the non-pyrowave builds (fixes -D warnings on the featureless Linux client build). - punktfunk-host: on a mid-stream mode switch, re-resolve the "Automatic" PyroWave bitrate for the new mode's ~1.6 bpp operating point (explicit rates and H.26x ABR stay put); reject sub-128px PyroWave modes before the encoder rebuild instead of after the ack. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -328,6 +328,20 @@ fn pump(
|
||||
// Live host↔client clock offset: loaded per frame (Relaxed) so mid-stream re-syncs (an NTP
|
||||
// step, drift) keep the capture-clock latency stats honest — never cached at session start.
|
||||
let clock_offset_live = connector.clock_offset_shared();
|
||||
// PUNKTFUNK_DEBUG_RECONFIGURE=WxH@HZ:SECS — lab lever: request ONE mid-stream mode
|
||||
// switch N seconds in, so a headless session (no window manager to drag a window in)
|
||||
// can exercise the resize path deterministically — host pipeline rebuild, decoder
|
||||
// follow-through (e.g. the PyroWave in-place rebuild), overlay/aspect handling.
|
||||
let pump_start = Instant::now();
|
||||
let mut debug_reconfig = std::env::var("PUNKTFUNK_DEBUG_RECONFIGURE")
|
||||
.ok()
|
||||
.and_then(|s| {
|
||||
let parsed = parse_debug_reconfigure(&s);
|
||||
if parsed.is_none() {
|
||||
tracing::warn!(value = %s, "PUNKTFUNK_DEBUG_RECONFIGURE not understood (want WxH@HZ:SECS) — ignored");
|
||||
}
|
||||
parsed
|
||||
});
|
||||
let mut total_frames = 0u64;
|
||||
// Newest frame index handed to the decoder — the staleness bar for late partials.
|
||||
let mut newest_decoded_idx: Option<u32> = None;
|
||||
@@ -368,6 +382,18 @@ fn pump(
|
||||
if stop.load(Ordering::SeqCst) {
|
||||
break None;
|
||||
}
|
||||
if let Some((mode, delay)) = debug_reconfig {
|
||||
if pump_start.elapsed() >= delay {
|
||||
tracing::info!(
|
||||
?mode,
|
||||
"PUNKTFUNK_DEBUG_RECONFIGURE: requesting mid-stream mode switch"
|
||||
);
|
||||
if let Err(e) = connector.request_mode(mode) {
|
||||
tracing::warn!(error = ?e, "debug mode switch request failed");
|
||||
}
|
||||
debug_reconfig = None;
|
||||
}
|
||||
}
|
||||
// 20 ms wait: audio has its own thread now, so this only bounds stop-flag
|
||||
// responsiveness and the per-iteration keyframe-recovery check (a frame arrives
|
||||
// every ~8–16 ms at 60–120 Hz anyway, so this rarely times out mid-stream).
|
||||
@@ -765,3 +791,43 @@ fn spawn_audio(
|
||||
.map_err(|e| tracing::warn!(error = %e, "audio thread failed to start — audio disabled"))
|
||||
.ok()
|
||||
}
|
||||
|
||||
/// Parse the `PUNKTFUNK_DEBUG_RECONFIGURE` lab lever: `WxH@HZ:SECS` → request that mode
|
||||
/// SECS seconds into the stream (e.g. `1280x720@60:5`).
|
||||
fn parse_debug_reconfigure(s: &str) -> Option<(Mode, Duration)> {
|
||||
let (mode_s, secs_s) = s.split_once(':')?;
|
||||
let (res, hz) = mode_s.split_once('@')?;
|
||||
let (w, h) = res.split_once('x')?;
|
||||
let mode = Mode {
|
||||
width: w.trim().parse().ok()?,
|
||||
height: h.trim().parse().ok()?,
|
||||
refresh_hz: hz.trim().parse().ok()?,
|
||||
};
|
||||
Some((mode, Duration::from_secs(secs_s.trim().parse().ok()?)))
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn debug_reconfigure_parses_the_documented_shape() {
|
||||
let (mode, delay) = parse_debug_reconfigure("1280x720@60:5").unwrap();
|
||||
assert_eq!((mode.width, mode.height, mode.refresh_hz), (1280, 720, 60));
|
||||
assert_eq!(delay, Duration::from_secs(5));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn debug_reconfigure_rejects_garbage() {
|
||||
for bad in [
|
||||
"",
|
||||
"1280x720",
|
||||
"1280x720@60",
|
||||
"x@:",
|
||||
"ax b@c:d",
|
||||
"1280x720@60:x",
|
||||
] {
|
||||
assert!(parse_debug_reconfigure(bad).is_none(), "{bad:?} parsed");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -628,6 +628,11 @@ impl Decoder {
|
||||
pub fn decode_frame(
|
||||
&mut self,
|
||||
au: &[u8],
|
||||
// Only the PyroWave backend reads the flags; without that feature the param is unused.
|
||||
#[cfg_attr(
|
||||
not(all(target_os = "linux", feature = "pyrowave")),
|
||||
allow(unused_variables)
|
||||
)]
|
||||
user_flags: u32,
|
||||
complete: bool,
|
||||
) -> Result<Option<DecodedImage>> {
|
||||
|
||||
@@ -19,6 +19,14 @@
|
||||
//! content-equivalent ones from [`VulkanDecodeDevice`]'s exported extension lists,
|
||||
//! feature facts and queue-family shape (pyrowave reads them for extension/feature
|
||||
//! detection; pointer identity is not required).
|
||||
//!
|
||||
//! **Mid-stream resize:** the pyrowave decoder object is fixed-size, but every frame's
|
||||
//! bitstream opens with a sequence header carrying its dimensions — [`au_dims`] sniffs
|
||||
//! it and [`PyroWaveDecoder::reconfigure`] rebuilds the decoder + plane ring in place
|
||||
//! when the host's `Reconfigure` pipeline rebuild lands (the pyrowave *device*, command
|
||||
//! pool and pinned create-infos are dimension-independent and survive). Superseded plane
|
||||
//! rings are retired, not destroyed — the presenter may still hold their views (see
|
||||
//! [`RETIRE_HANDOVERS`]).
|
||||
|
||||
use crate::video::{ColorDesc, VulkanDecodeDevice};
|
||||
use anyhow::{bail, Context as _, Result};
|
||||
@@ -27,6 +35,7 @@ use ash::vk::Handle as _;
|
||||
use pyrowave_sys as pw;
|
||||
use std::ffi::{c_char, c_void, CString};
|
||||
use std::sync::Arc;
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
/// Plane-set ring depth: decode writes slot N while the presenter may still sample
|
||||
/// N-1/N-2 (its own submission raced ahead under the shared queue's FIFO order, so
|
||||
@@ -34,6 +43,18 @@ use std::sync::Arc;
|
||||
/// keeps LOGICAL reuse far enough behind).
|
||||
const RING: usize = 4;
|
||||
|
||||
/// A mid-stream resize retires the old plane ring, but its images can't be destroyed
|
||||
/// immediately: the pump→presenter frame channel (depth 2, newest-wins) may still hold a
|
||||
/// frame referencing them, and the presenter binds a frame's views into its descriptor
|
||||
/// set only inside the `present` call that carries it. Once this many NEW-ring frames
|
||||
/// have been handed over, every old-ring frame has been displaced from the channel and
|
||||
/// any present that picked one up has long finished recording; combined with the
|
||||
/// queue-idle taken before destruction (covers submitted GPU work) the retired images
|
||||
/// are provably unreachable. The wall-clock floor is a belt for a presenter stalled
|
||||
/// mid-`present` (swapchain acquire on an occluded window) while frames keep flowing.
|
||||
const RETIRE_HANDOVERS: u32 = 8;
|
||||
const RETIRE_MIN_AGE: Duration = Duration::from_millis(250);
|
||||
|
||||
fn pw_check(r: pw::pyrowave_result, what: &str) -> Result<()> {
|
||||
if r == pw::pyrowave_result_PYROWAVE_SUCCESS {
|
||||
Ok(())
|
||||
@@ -42,6 +63,54 @@ fn pw_check(r: pw::pyrowave_result, what: &str) -> Result<()> {
|
||||
}
|
||||
}
|
||||
|
||||
/// Parse an upstream `BitstreamSequenceHeader` (pyrowave_common.hpp) at the start of
|
||||
/// `bytes`: 8 bytes, two LE u32s — word 0 = `width_minus_1:14 | height_minus_1:14 |
|
||||
/// sequence:3 | extended:1`, word 1 = `total_blocks:24 | code:2 | …`. Returns the frame
|
||||
/// dimensions when this really is a START-OF-FRAME sequence header (the `extended` bit
|
||||
/// distinguishes it from a regular `BitstreamHeader`, which carries a wavelet block).
|
||||
fn seq_header_dims(bytes: &[u8]) -> Option<(u32, u32)> {
|
||||
if bytes.len() < 8 {
|
||||
return None;
|
||||
}
|
||||
let w0 = u32::from_le_bytes(bytes[0..4].try_into().unwrap());
|
||||
let w1 = u32::from_le_bytes(bytes[4..8].try_into().unwrap());
|
||||
if w0 >> 31 == 0 {
|
||||
return None; // regular block header, not a sequence header
|
||||
}
|
||||
if (w1 >> 24) & 0x3 != 0 {
|
||||
return None; // extended, but not BITSTREAM_EXTENDED_CODE_START_OF_FRAME
|
||||
}
|
||||
Some(((w0 & 0x3FFF) + 1, ((w0 >> 14) & 0x3FFF) + 1))
|
||||
}
|
||||
|
||||
/// The frame dimensions an AU announces, or `None` when they can't be known from this AU
|
||||
/// (the sequence header rode a lost shard of a partial). The encoder writes exactly one
|
||||
/// sequence header per frame, at byte 0 of the frame's bitstream — so it sits at the
|
||||
/// start of an unaligned AU, and at the start of the FIRST window's body in a
|
||||
/// chunk-aligned AU (§4.4 framing: 4-byte prefix `used:u16 | kind:u16`; kind PACKED or
|
||||
/// FRAG_FIRST both begin with the frame's first packet, and that packet begins with the
|
||||
/// sequence header).
|
||||
fn au_dims(au: &[u8], aligned: bool, wire_window: usize) -> Option<(u32, u32)> {
|
||||
if !aligned {
|
||||
return seq_header_dims(au);
|
||||
}
|
||||
let win = &au[..au.len().min(wire_window)];
|
||||
if win.len() < 4 {
|
||||
return None;
|
||||
}
|
||||
let used = u16::from_le_bytes([win[0], win[1]]) as usize;
|
||||
let kind = u16::from_le_bytes([win[2], win[3]]);
|
||||
if used == 0 || 4 + used > win.len() {
|
||||
return None; // first window lost/garbage — the sequence header went with it
|
||||
}
|
||||
// WIN_PACKED (0) and WIN_FRAG_FIRST (1) both start at the frame's first packet;
|
||||
// a CONT/LAST fragment here would mean the first window was lost.
|
||||
if kind > 1 {
|
||||
return None;
|
||||
}
|
||||
seq_header_dims(&win[4..4 + used])
|
||||
}
|
||||
|
||||
/// Content-equivalent reconstruction of the presenter device's create-infos, pinned for
|
||||
/// the lifetime of the `pyrowave_device` (heap boxes; moving `Hold` moves only pointers).
|
||||
struct Hold {
|
||||
@@ -174,6 +243,153 @@ struct PlaneSet {
|
||||
initialized: bool,
|
||||
}
|
||||
|
||||
/// A plane ring superseded by a mid-stream resize, awaiting safe destruction (see
|
||||
/// [`RETIRE_HANDOVERS`] for the lifetime argument).
|
||||
struct RetiredRing {
|
||||
sets: Vec<PlaneSet>,
|
||||
/// Frames handed to the presenter since this ring was retired.
|
||||
handed_over: u32,
|
||||
retired_at: Instant,
|
||||
}
|
||||
|
||||
/// One decode-output plane: R8, storage (decode writes) + sampled (presenter CSC).
|
||||
unsafe fn make_plane(
|
||||
device: &ash::Device,
|
||||
mem_props: &vk::PhysicalDeviceMemoryProperties,
|
||||
w: u32,
|
||||
h: u32,
|
||||
) -> Result<(vk::Image, vk::DeviceMemory, vk::ImageView)> {
|
||||
let img = device.create_image(
|
||||
&vk::ImageCreateInfo::default()
|
||||
.image_type(vk::ImageType::TYPE_2D)
|
||||
.format(vk::Format::R8_UNORM)
|
||||
.extent(vk::Extent3D {
|
||||
width: w,
|
||||
height: h,
|
||||
depth: 1,
|
||||
})
|
||||
.mip_levels(1)
|
||||
.array_layers(1)
|
||||
.samples(vk::SampleCountFlags::TYPE_1)
|
||||
.tiling(vk::ImageTiling::OPTIMAL)
|
||||
.usage(vk::ImageUsageFlags::STORAGE | vk::ImageUsageFlags::SAMPLED)
|
||||
.initial_layout(vk::ImageLayout::UNDEFINED),
|
||||
None,
|
||||
)?;
|
||||
let req = device.get_image_memory_requirements(img);
|
||||
let ti = (0..mem_props.memory_type_count)
|
||||
.find(|&i| {
|
||||
(req.memory_type_bits & (1 << i)) != 0
|
||||
&& mem_props.memory_types[i as usize]
|
||||
.property_flags
|
||||
.contains(vk::MemoryPropertyFlags::DEVICE_LOCAL)
|
||||
})
|
||||
.unwrap_or(0);
|
||||
let mem = match device.allocate_memory(
|
||||
&vk::MemoryAllocateInfo::default()
|
||||
.allocation_size(req.size)
|
||||
.memory_type_index(ti),
|
||||
None,
|
||||
) {
|
||||
Ok(m) => m,
|
||||
Err(e) => {
|
||||
device.destroy_image(img, None);
|
||||
return Err(e.into());
|
||||
}
|
||||
};
|
||||
if let Err(e) = device.bind_image_memory(img, mem, 0) {
|
||||
device.destroy_image(img, None);
|
||||
device.free_memory(mem, None);
|
||||
return Err(e.into());
|
||||
}
|
||||
let view = match device.create_image_view(
|
||||
&vk::ImageViewCreateInfo::default()
|
||||
.image(img)
|
||||
.view_type(vk::ImageViewType::TYPE_2D)
|
||||
.format(vk::Format::R8_UNORM)
|
||||
.subresource_range(vk::ImageSubresourceRange {
|
||||
aspect_mask: vk::ImageAspectFlags::COLOR,
|
||||
base_mip_level: 0,
|
||||
level_count: 1,
|
||||
base_array_layer: 0,
|
||||
layer_count: 1,
|
||||
}),
|
||||
None,
|
||||
) {
|
||||
Ok(v) => v,
|
||||
Err(e) => {
|
||||
device.destroy_image(img, None);
|
||||
device.free_memory(mem, None);
|
||||
return Err(e.into());
|
||||
}
|
||||
};
|
||||
Ok((img, mem, view))
|
||||
}
|
||||
|
||||
unsafe fn destroy_sets(device: &ash::Device, sets: &[PlaneSet]) {
|
||||
for set in sets {
|
||||
for v in set.views {
|
||||
device.destroy_image_view(v, None);
|
||||
}
|
||||
for i in set.imgs {
|
||||
device.destroy_image(i, None);
|
||||
}
|
||||
for m in set.mems {
|
||||
device.free_memory(m, None);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Build a fresh [`RING`]-deep plane ring at the given dimensions; cleans up the partial
|
||||
/// ring on failure (the caller keeps whatever it was using before).
|
||||
unsafe fn build_ring(
|
||||
device: &ash::Device,
|
||||
mem_props: &vk::PhysicalDeviceMemoryProperties,
|
||||
width: u32,
|
||||
height: u32,
|
||||
) -> Result<Vec<PlaneSet>> {
|
||||
let mut ring: Vec<PlaneSet> = Vec::with_capacity(RING);
|
||||
for _ in 0..RING {
|
||||
let built = (|| -> Result<PlaneSet> {
|
||||
let (y, ym, yv) = make_plane(device, mem_props, width, height)?;
|
||||
let (cb, cbm, cbv) = match make_plane(device, mem_props, width / 2, height / 2) {
|
||||
Ok(p) => p,
|
||||
Err(e) => {
|
||||
device.destroy_image_view(yv, None);
|
||||
device.destroy_image(y, None);
|
||||
device.free_memory(ym, None);
|
||||
return Err(e);
|
||||
}
|
||||
};
|
||||
let (cr, crm, crv) = match make_plane(device, mem_props, width / 2, height / 2) {
|
||||
Ok(p) => p,
|
||||
Err(e) => {
|
||||
for (v, i, m) in [(yv, y, ym), (cbv, cb, cbm)] {
|
||||
device.destroy_image_view(v, None);
|
||||
device.destroy_image(i, None);
|
||||
device.free_memory(m, None);
|
||||
}
|
||||
return Err(e);
|
||||
}
|
||||
};
|
||||
Ok(PlaneSet {
|
||||
imgs: [y, cb, cr],
|
||||
mems: [ym, cbm, crm],
|
||||
views: [yv, cbv, crv],
|
||||
initialized: false,
|
||||
})
|
||||
})();
|
||||
match built {
|
||||
Ok(set) => ring.push(set),
|
||||
Err(e) => {
|
||||
destroy_sets(device, &ring);
|
||||
return Err(e);
|
||||
}
|
||||
}
|
||||
}
|
||||
Ok(ring)
|
||||
}
|
||||
|
||||
pub struct PyroWaveDecoder {
|
||||
// ash wrappers reconstructed over the presenter's raw handles (not owned — the
|
||||
// presenter outlives the decoder; Drop destroys only what this struct created).
|
||||
@@ -184,10 +400,13 @@ pub struct PyroWaveDecoder {
|
||||
pw_dev: pw::pyrowave_device,
|
||||
pw_dec: pw::pyrowave_decoder,
|
||||
ring: Vec<PlaneSet>,
|
||||
/// Plane rings superseded by mid-stream resizes, pending safe destruction.
|
||||
retired: Vec<RetiredRing>,
|
||||
next: usize,
|
||||
cmd_pool: vk::CommandPool,
|
||||
cmd: vk::CommandBuffer,
|
||||
fence: vk::Fence,
|
||||
mem_props: vk::PhysicalDeviceMemoryProperties,
|
||||
width: u32,
|
||||
height: u32,
|
||||
/// The wire shard payload — the parse-window size for chunk-aligned AUs (§4.4): each
|
||||
@@ -294,68 +513,14 @@ impl PyroWaveDecoder {
|
||||
let mem_props = instance.get_physical_device_memory_properties(
|
||||
vk::PhysicalDevice::from_raw(vkd.physical_device as u64),
|
||||
);
|
||||
let make_plane = |w: u32, h: u32| -> Result<(vk::Image, vk::DeviceMemory, vk::ImageView)> {
|
||||
let img = device.create_image(
|
||||
&vk::ImageCreateInfo::default()
|
||||
.image_type(vk::ImageType::TYPE_2D)
|
||||
.format(vk::Format::R8_UNORM)
|
||||
.extent(vk::Extent3D {
|
||||
width: w,
|
||||
height: h,
|
||||
depth: 1,
|
||||
})
|
||||
.mip_levels(1)
|
||||
.array_layers(1)
|
||||
.samples(vk::SampleCountFlags::TYPE_1)
|
||||
.tiling(vk::ImageTiling::OPTIMAL)
|
||||
.usage(vk::ImageUsageFlags::STORAGE | vk::ImageUsageFlags::SAMPLED)
|
||||
.initial_layout(vk::ImageLayout::UNDEFINED),
|
||||
None,
|
||||
)?;
|
||||
let req = device.get_image_memory_requirements(img);
|
||||
let ti = (0..mem_props.memory_type_count)
|
||||
.find(|&i| {
|
||||
(req.memory_type_bits & (1 << i)) != 0
|
||||
&& mem_props.memory_types[i as usize]
|
||||
.property_flags
|
||||
.contains(vk::MemoryPropertyFlags::DEVICE_LOCAL)
|
||||
})
|
||||
.unwrap_or(0);
|
||||
let mem = device.allocate_memory(
|
||||
&vk::MemoryAllocateInfo::default()
|
||||
.allocation_size(req.size)
|
||||
.memory_type_index(ti),
|
||||
None,
|
||||
)?;
|
||||
device.bind_image_memory(img, mem, 0)?;
|
||||
let view = device.create_image_view(
|
||||
&vk::ImageViewCreateInfo::default()
|
||||
.image(img)
|
||||
.view_type(vk::ImageViewType::TYPE_2D)
|
||||
.format(vk::Format::R8_UNORM)
|
||||
.subresource_range(vk::ImageSubresourceRange {
|
||||
aspect_mask: vk::ImageAspectFlags::COLOR,
|
||||
base_mip_level: 0,
|
||||
level_count: 1,
|
||||
base_array_layer: 0,
|
||||
layer_count: 1,
|
||||
}),
|
||||
None,
|
||||
)?;
|
||||
Ok((img, mem, view))
|
||||
let ring = match build_ring(&device, &mem_props, width, height) {
|
||||
Ok(r) => r,
|
||||
Err(e) => {
|
||||
pw::pyrowave_decoder_destroy(pw_dec);
|
||||
pw::pyrowave_device_destroy(pw_dev);
|
||||
return Err(e);
|
||||
}
|
||||
};
|
||||
let mut ring = Vec::with_capacity(RING);
|
||||
for _ in 0..RING {
|
||||
let (y, ym, yv) = make_plane(width, height)?;
|
||||
let (cb, cbm, cbv) = make_plane(width / 2, height / 2)?;
|
||||
let (cr, crm, crv) = make_plane(width / 2, height / 2)?;
|
||||
ring.push(PlaneSet {
|
||||
imgs: [y, cb, cr],
|
||||
mems: [ym, cbm, crm],
|
||||
views: [yv, cbv, crv],
|
||||
initialized: false,
|
||||
});
|
||||
}
|
||||
|
||||
let cmd_pool = device.create_command_pool(
|
||||
&vk::CommandPoolCreateInfo::default()
|
||||
@@ -383,16 +548,94 @@ impl PyroWaveDecoder {
|
||||
pw_dev,
|
||||
pw_dec,
|
||||
ring,
|
||||
retired: Vec::new(),
|
||||
next: 0,
|
||||
cmd_pool,
|
||||
cmd,
|
||||
fence,
|
||||
mem_props,
|
||||
width,
|
||||
height,
|
||||
wire_window: shard_payload.max(64),
|
||||
})
|
||||
}
|
||||
|
||||
/// Mid-stream resize: rebuild the pyrowave decoder + plane ring at the new
|
||||
/// dimensions in place, keeping the (dimension-independent) pyrowave device, command
|
||||
/// pool, fence and pinned create-infos. Build-new-before-drop-old: a failure leaves
|
||||
/// the current decoder untouched (and propagates — with the stream now at a size we
|
||||
/// can't decode, the session ends with a real error instead of a frozen picture).
|
||||
/// The old ring is RETIRED, not destroyed: the presenter / frame channel may still
|
||||
/// reference its views (see [`RETIRE_HANDOVERS`]).
|
||||
unsafe fn reconfigure(&mut self, width: u32, height: u32) -> Result<()> {
|
||||
if width % 2 != 0 || height % 2 != 0 {
|
||||
bail!("pyrowave 4:2:0 needs even dimensions (resize to {width}x{height})");
|
||||
}
|
||||
let dinfo = pw::pyrowave_decoder_create_info {
|
||||
device: self.pw_dev,
|
||||
width: width as i32,
|
||||
height: height as i32,
|
||||
chroma: pw::pyrowave_chroma_subsampling_PYROWAVE_CHROMA_SUBSAMPLING_420,
|
||||
fragment_path: false,
|
||||
};
|
||||
let mut new_dec: pw::pyrowave_decoder = std::ptr::null_mut();
|
||||
pw_check(
|
||||
pw::pyrowave_decoder_create(&dinfo, &mut new_dec),
|
||||
"decoder_create (mid-stream resize)",
|
||||
)?;
|
||||
let new_ring = match build_ring(&self.device, &self.mem_props, width, height) {
|
||||
Ok(r) => r,
|
||||
Err(e) => {
|
||||
pw::pyrowave_decoder_destroy(new_dec);
|
||||
return Err(e).context("plane ring (mid-stream resize)");
|
||||
}
|
||||
};
|
||||
// Our own decode work is fence-synchronous (never in flight here), so the old
|
||||
// pyrowave decoder can go immediately; only the plane images wait (retired).
|
||||
pw::pyrowave_decoder_destroy(self.pw_dec);
|
||||
self.pw_dec = new_dec;
|
||||
let old = std::mem::replace(&mut self.ring, new_ring);
|
||||
self.retired.push(RetiredRing {
|
||||
sets: old,
|
||||
handed_over: 0,
|
||||
retired_at: Instant::now(),
|
||||
});
|
||||
self.next = 0;
|
||||
tracing::info!(
|
||||
from = %format!("{}x{}", self.width, self.height),
|
||||
to = %format!("{width}x{height}"),
|
||||
"PyroWave decoder rebuilt for mid-stream resize"
|
||||
);
|
||||
self.width = width;
|
||||
self.height = height;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Destroy retired rings that are provably unreachable (enough new-ring frames handed
|
||||
/// over + a wall-clock floor — see [`RETIRE_HANDOVERS`]); the queue idle bounds any
|
||||
/// still-submitted presenter sampling of the retiring views.
|
||||
unsafe fn reap_retired(&mut self) {
|
||||
let ripe = |r: &RetiredRing| {
|
||||
r.handed_over >= RETIRE_HANDOVERS && r.retired_at.elapsed() >= RETIRE_MIN_AGE
|
||||
};
|
||||
if !self.retired.iter().any(ripe) {
|
||||
return;
|
||||
}
|
||||
{
|
||||
let _guard = self.queue_lock.guard();
|
||||
let _ = self.device.queue_wait_idle(self.queue);
|
||||
}
|
||||
let mut kept = Vec::new();
|
||||
for r in self.retired.drain(..) {
|
||||
if ripe(&r) {
|
||||
destroy_sets(&self.device, &r.sets);
|
||||
} else {
|
||||
kept.push(r);
|
||||
}
|
||||
}
|
||||
self.retired = kept;
|
||||
}
|
||||
|
||||
/// One AU in → one frame out. `aligned` = the AU is shard-window chunked (each
|
||||
/// `wire_window` holds whole self-delimiting packets, zero-padded — walk and strip);
|
||||
/// `complete` = every shard arrived (a partial decodes anyway: missing blocks are
|
||||
@@ -477,20 +720,43 @@ impl PyroWaveDecoder {
|
||||
aligned: bool,
|
||||
complete: bool,
|
||||
) -> Result<Option<PyroWavePlanarFrame>> {
|
||||
// Mid-stream resize: every frame's bitstream opens with a sequence header
|
||||
// carrying its dimensions, so the AU itself announces the host's mode switch —
|
||||
// no control-plane ordering to race (the Reconfigured ack travels on another
|
||||
// stream). Upstream hard-errors on a dimension mismatch, so rebuild FIRST. A
|
||||
// partial that lost its first shard sniffs `None` and decodes at the current
|
||||
// size (correct when the size didn't change; harmlessly dropped below when it
|
||||
// did — the next complete frame carries the header again).
|
||||
if let Some(dims) = au_dims(au, aligned, self.wire_window) {
|
||||
if dims != (self.width, self.height) {
|
||||
self.reconfigure(dims.0, dims.1)?;
|
||||
}
|
||||
}
|
||||
let mut push_err: Option<anyhow::Error> = None;
|
||||
if aligned {
|
||||
let mut frag: Vec<u8> = Vec::new();
|
||||
for win in au.chunks(self.wire_window) {
|
||||
self.push_window(win, &mut frag)?;
|
||||
if let Err(e) = self.push_window(win, &mut frag) {
|
||||
push_err = Some(e);
|
||||
break;
|
||||
}
|
||||
}
|
||||
} else {
|
||||
pw_check(
|
||||
pw::pyrowave_decoder_push_packet(
|
||||
self.pw_dec,
|
||||
au.as_ptr() as *const c_void,
|
||||
au.len(),
|
||||
),
|
||||
"push_packet",
|
||||
)?;
|
||||
} else if let Err(e) = pw_check(
|
||||
pw::pyrowave_decoder_push_packet(self.pw_dec, au.as_ptr() as *const c_void, au.len()),
|
||||
"push_packet",
|
||||
) {
|
||||
push_err = Some(e);
|
||||
}
|
||||
if let Some(e) = push_err {
|
||||
// A partial straddling a resize can carry blocks the (possibly wrong-size)
|
||||
// decoder rejects — that's one lost frame, not a broken session; the next
|
||||
// complete frame re-anchors (all-intra). A COMPLETE frame that fails to
|
||||
// parse is real corruption: propagate.
|
||||
if complete {
|
||||
return Err(e);
|
||||
}
|
||||
tracing::debug!(error = %format!("{e:#}"), "partial AU rejected — frame dropped");
|
||||
return Ok(None);
|
||||
}
|
||||
// A complete AU that isn't ready is a stale/duplicate (sequence rewind) — skip.
|
||||
// A PARTIAL is decoded regardless: missing wavelet blocks reconstruct as zeros,
|
||||
@@ -604,6 +870,13 @@ impl PyroWaveDecoder {
|
||||
.context("pyrowave decode fence")?;
|
||||
self.ring[slot].initialized = true;
|
||||
|
||||
// This frame is about to reach the presenter — it advances every retired ring's
|
||||
// displacement count, and ripe rings can now be destroyed.
|
||||
for r in &mut self.retired {
|
||||
r.handed_over += 1;
|
||||
}
|
||||
self.reap_retired();
|
||||
|
||||
Ok(Some(PyroWavePlanarFrame {
|
||||
views: [
|
||||
self.ring[slot].views[0].as_raw(),
|
||||
@@ -639,16 +912,9 @@ impl Drop for PyroWaveDecoder {
|
||||
}
|
||||
pw::pyrowave_decoder_destroy(self.pw_dec);
|
||||
pw::pyrowave_device_destroy(self.pw_dev);
|
||||
for set in &self.ring {
|
||||
for v in set.views {
|
||||
self.device.destroy_image_view(v, None);
|
||||
}
|
||||
for i in set.imgs {
|
||||
self.device.destroy_image(i, None);
|
||||
}
|
||||
for m in set.mems {
|
||||
self.device.free_memory(m, None);
|
||||
}
|
||||
destroy_sets(&self.device, &self.ring);
|
||||
for r in &self.retired {
|
||||
destroy_sets(&self.device, &r.sets);
|
||||
}
|
||||
self.device.destroy_fence(self.fence, None);
|
||||
self.device.destroy_command_pool(self.cmd_pool, None);
|
||||
@@ -656,3 +922,104 @@ impl Drop for PyroWaveDecoder {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::{au_dims, seq_header_dims};
|
||||
|
||||
/// Little-endian encoding of upstream's `BitstreamSequenceHeader` bitfields (see
|
||||
/// pyrowave_common.hpp): word 0 = width_minus_1:14 | height_minus_1:14 | sequence:3
|
||||
/// | extended:1; word 1 = total_blocks:24 | code:2 | chroma:1 | …
|
||||
fn seq_header(w: u32, h: u32, code: u32) -> [u8; 8] {
|
||||
let w0 = (w - 1) & 0x3FFF | ((h - 1) & 0x3FFF) << 14 | 1 << 31;
|
||||
let w1 = 0x1234 | code << 24; // arbitrary total_blocks
|
||||
let mut out = [0u8; 8];
|
||||
out[0..4].copy_from_slice(&w0.to_le_bytes());
|
||||
out[4..8].copy_from_slice(&w1.to_le_bytes());
|
||||
out
|
||||
}
|
||||
|
||||
/// A regular `BitstreamHeader` (block packet): extended bit clear.
|
||||
fn block_header() -> [u8; 8] {
|
||||
let w0 = 0xBEEFu32 | 8 << 16; // ballot | payload_words=8, extended=0
|
||||
let w1 = 42u32 << 8; // block_index
|
||||
let mut out = [0u8; 8];
|
||||
out[0..4].copy_from_slice(&w0.to_le_bytes());
|
||||
out[4..8].copy_from_slice(&w1.to_le_bytes());
|
||||
out
|
||||
}
|
||||
|
||||
/// Wrap `body` in one §4.4 framed window of `win` bytes (4-byte prefix + zero pad).
|
||||
fn window(body: &[u8], kind: u16, win: usize) -> Vec<u8> {
|
||||
let mut out = Vec::with_capacity(win);
|
||||
out.extend_from_slice(&(body.len() as u16).to_le_bytes());
|
||||
out.extend_from_slice(&kind.to_le_bytes());
|
||||
out.extend_from_slice(body);
|
||||
out.resize(win, 0);
|
||||
out
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn sniffs_dims_from_a_sequence_header() {
|
||||
assert_eq!(
|
||||
seq_header_dims(&seq_header(1920, 1080, 0)),
|
||||
Some((1920, 1080))
|
||||
);
|
||||
assert_eq!(
|
||||
seq_header_dims(&seq_header(1280, 720, 0)),
|
||||
Some((1280, 720))
|
||||
);
|
||||
// 14-bit fields carry up to 16384.
|
||||
assert_eq!(
|
||||
seq_header_dims(&seq_header(16384, 16384, 0)),
|
||||
Some((16384, 16384))
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn rejects_non_sequence_headers() {
|
||||
assert_eq!(seq_header_dims(&block_header()), None); // extended bit clear
|
||||
assert_eq!(seq_header_dims(&seq_header(1920, 1080, 1)), None); // not START_OF_FRAME
|
||||
assert_eq!(seq_header_dims(&seq_header(1920, 1080, 0)[..7]), None); // short
|
||||
assert_eq!(seq_header_dims(&[]), None);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn unaligned_au_sniffs_at_byte_zero() {
|
||||
let mut au = seq_header(2560, 1440, 0).to_vec();
|
||||
au.extend_from_slice(&block_header());
|
||||
assert_eq!(au_dims(&au, false, 1404), Some((2560, 1440)));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn aligned_au_sniffs_the_first_window_body() {
|
||||
const WIN: usize = 64;
|
||||
let mut body = seq_header(1280, 800, 0).to_vec();
|
||||
body.extend_from_slice(&block_header());
|
||||
// WIN_PACKED first window, then another window of blocks.
|
||||
let mut au = window(&body, 0, WIN);
|
||||
au.extend_from_slice(&window(&block_header(), 0, WIN));
|
||||
assert_eq!(au_dims(&au, true, WIN), Some((1280, 800)));
|
||||
// An oversized first packet rides a FRAG chain — FRAG_FIRST also starts at the
|
||||
// frame's first byte, so the header is still there.
|
||||
let frag = window(&body, 1, WIN);
|
||||
assert_eq!(au_dims(&frag, true, WIN), Some((1280, 800)));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn lost_first_window_means_unknown_dims() {
|
||||
const WIN: usize = 64;
|
||||
// A lost shard arrives as a zeroed window (used = 0) — the sequence header is gone.
|
||||
let mut au = vec![0u8; WIN];
|
||||
au.extend_from_slice(&window(&seq_header(1280, 800, 0), 0, WIN));
|
||||
assert_eq!(au_dims(&au, true, WIN), None);
|
||||
// A FRAG_CONT/LAST first window means the same (its FIRST was in a lost prior AU).
|
||||
let cont = window(&block_header(), 2, WIN);
|
||||
assert_eq!(au_dims(&cont, true, WIN), None);
|
||||
// Garbage used-length never reads out of bounds.
|
||||
let mut garbage = vec![0u8; WIN];
|
||||
garbage[0] = 0xFF;
|
||||
garbage[1] = 0xFF;
|
||||
assert_eq!(au_dims(&garbage, true, WIN), None);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -428,6 +428,15 @@ pub fn validate_dimensions(codec: Codec, width: u32, height: u32) -> Result<()>
|
||||
if width % 2 != 0 || height % 2 != 0 {
|
||||
anyhow::bail!("invalid encode resolution {width}x{height}: dimensions must be even");
|
||||
}
|
||||
// PyroWave's 5-level wavelet decomposition needs ≥ 4·2⁵ px per axis (upstream
|
||||
// `MinimumImageSize` — the band mirroring breaks below it); reject a tiny mode here
|
||||
// (e.g. a match-window resize dragged to a sliver) instead of failing the encoder
|
||||
// rebuild after the switch was acked.
|
||||
if codec == Codec::PyroWave && (width < 128 || height < 128) {
|
||||
anyhow::bail!(
|
||||
"invalid PyroWave resolution {width}x{height}: the wavelet needs at least 128px per axis"
|
||||
);
|
||||
}
|
||||
let max = codec.max_dimension();
|
||||
if width > max || height > max {
|
||||
anyhow::bail!(
|
||||
|
||||
@@ -1172,7 +1172,7 @@ mod tests {
|
||||
buf.row_stride_in_bytes = [w as usize, (w / 2) as usize, (w / 2) as usize];
|
||||
buf.plane_size_in_bytes = [y.len(), cb.len(), cr.len()];
|
||||
assert_eq!(
|
||||
pw::pyrowave_decoder_decode_cpu_buffer_synchronous(dec, &mut buf),
|
||||
pw::pyrowave_decoder_decode_cpu_buffer_synchronous(dec, &buf),
|
||||
pw::pyrowave_result_PYROWAVE_SUCCESS
|
||||
);
|
||||
pw::pyrowave_decoder_destroy(dec);
|
||||
|
||||
@@ -1606,6 +1606,9 @@ async fn serve_session(
|
||||
}
|
||||
});
|
||||
let bitrate_kbps = welcome.bitrate_kbps; // resolved encoder bitrate (Hello clamped, or default)
|
||||
// "Automatic" request: the resolved rate is a host default — for PyroWave a per-mode
|
||||
// bpp pin the data plane re-resolves on a mid-stream mode switch.
|
||||
let bitrate_auto = hello.bitrate_kbps == 0;
|
||||
let bit_depth = welcome.bit_depth; // resolved encode bit depth (8, or 10 when negotiated)
|
||||
// Resolved chroma — derive the typed value back from the wire byte the Welcome carried (so the
|
||||
// session uses exactly what the client was told). `Yuv444` only when the handshake gate passed.
|
||||
@@ -1703,6 +1706,7 @@ async fn serve_session(
|
||||
bitrate_rx,
|
||||
compositor,
|
||||
bitrate_kbps,
|
||||
bitrate_auto,
|
||||
bit_depth,
|
||||
chroma,
|
||||
codec,
|
||||
@@ -3948,6 +3952,11 @@ struct SessionContext {
|
||||
compositor: crate::vdisplay::Compositor,
|
||||
/// Negotiated encoder bitrate (kbps).
|
||||
bitrate_kbps: u32,
|
||||
/// The client asked for "Automatic" (`Hello::bitrate_kbps == 0`), so `bitrate_kbps` came from
|
||||
/// the host's codec-aware default. For PyroWave that default is the ~1.6 bpp operating point of
|
||||
/// the NEGOTIATED MODE (`resolve_bitrate_kbps_for`) — a mid-stream mode switch re-resolves it
|
||||
/// for the new mode (the pin follows the resolution; an explicit client rate stays put).
|
||||
bitrate_auto: bool,
|
||||
/// Negotiated encode bit depth (8, or 10 = HEVC Main10).
|
||||
bit_depth: u8,
|
||||
/// Negotiated chroma subsampling (4:2:0, or 4:4:4 when the client + host + GPU all support it).
|
||||
@@ -4027,6 +4036,7 @@ fn virtual_stream(ctx: SessionContext) -> Result<()> {
|
||||
bitrate_rx,
|
||||
compositor,
|
||||
mut bitrate_kbps,
|
||||
bitrate_auto,
|
||||
bit_depth,
|
||||
// The resolved chroma is already captured in `plan` (above); ignore the duplicate here.
|
||||
chroma: _,
|
||||
@@ -4363,11 +4373,29 @@ fn virtual_stream(ctx: SessionContext) -> Result<()> {
|
||||
}
|
||||
if let Some(new_mode) = want {
|
||||
tracing::info!(?new_mode, "rebuilding pipeline for mode switch");
|
||||
// PyroWave's Automatic bitrate is a per-mode ~1.6 bpp pin (resolve_bitrate_kbps_for) —
|
||||
// a resolution change moves the operating point (1080p→4K quadruples the pixel rate),
|
||||
// so re-resolve it for the new mode. Explicit client rates stay put (the operator knows
|
||||
// the link), and the H.26x codecs keep their mode-independent rate (ABR owns it).
|
||||
let mode_bitrate = if bitrate_auto && plan.codec == crate::encode::Codec::PyroWave {
|
||||
resolve_bitrate_kbps_for(plan.codec, 0, &new_mode)
|
||||
} else {
|
||||
bitrate_kbps
|
||||
};
|
||||
// Build the new pipeline BEFORE dropping the old one: the host already acked
|
||||
// the switch as accepted, so a rebuild failure must not kill an otherwise
|
||||
// healthy session — keep streaming the current mode and log instead.
|
||||
match build_pipeline(&mut vd, new_mode, bitrate_kbps, bit_depth, plan, &quit) {
|
||||
match build_pipeline(&mut vd, new_mode, mode_bitrate, bit_depth, plan, &quit) {
|
||||
Ok(next_pipe) => {
|
||||
if mode_bitrate != bitrate_kbps {
|
||||
tracing::info!(
|
||||
from_kbps = bitrate_kbps,
|
||||
to_kbps = mode_bitrate,
|
||||
"pinned PyroWave bitrate re-resolved for the new mode"
|
||||
);
|
||||
bitrate_kbps = mode_bitrate;
|
||||
live_bitrate.store(mode_bitrate, Ordering::Relaxed);
|
||||
}
|
||||
let old_display_gen = cur_display_gen;
|
||||
// The destructuring assignment drops the OLD capturer (→ its display lease) as
|
||||
// each binding is replaced — the new pipeline is already up (create-before-drop).
|
||||
|
||||
Reference in New Issue
Block a user