alpha2-host 0.5.1

//! How an export decodes its sources, whichever output path it takes.
//!
//! The readback and GPU-resident paths size the decode pool and thread split from the same facts,
//! so the GPU output path does not assume hardware decode and a software-decoded export can still
//! encode from the GPU.

use std::collections::HashMap;
use std::path::PathBuf;
use std::sync::Arc;
use std::time::{Duration, Instant};

use crate::config::{
    DecoderSessionCount, configured_decode_concurrency, configured_decode_lookahead,
    configured_decoder_session_override, configured_reverse_hardware_surfaces,
    decoder_session_bytes, decoder_session_count, visual_decode_reads,
};
use crate::decode::{CursorReads, DecodeWorkerConfig};
use crate::frames::NativeGpuFrames;
use crate::host::Host;
use alpha2_model::Project;
use alpha2_native::{CodecThreadAllocation, CodecThreadRequest, codec_thread_allocation};
use alpha2_render::RenderSize;
use alpha2_time::TimeRange;

pub struct ExportDecodePlan {
    /// Never on the synchronous comparison path.
    pub hardware_decode: bool,
    pub decode_lookahead: usize,
    pub decoder_sessions: DecoderSessionCount,
    pub reads: Arc<CursorReads>,
    pub codec_threads: CodecThreadAllocation,
    /// Luma samples in the largest video source when a software decoder shares the CPU with a
    /// software encoder; zero otherwise.
    pub source_frame_pixels: u64,
    pub source_probe: Duration,
    pub providers: Arc<crate::providers::ProviderMedia>,
}

impl ExportDecodePlan {
    pub(crate) fn new(
        split: ExportCodecSplit,
        providers: Arc<crate::providers::ProviderMedia>,
    ) -> Self {
        Self {
            hardware_decode: split.hardware_decode,
            decode_lookahead: split.decode_lookahead,
            decoder_sessions: split.decoder_sessions,
            reads: split.reads,
            codec_threads: split.codec_threads,
            source_frame_pixels: split.source_frame_pixels,
            source_probe: split.source_probe,
            providers,
        }
    }
}

#[derive(Clone, Debug)]
pub struct ExportCodecSplit {
    pub hardware_decode: bool,
    pub decode_lookahead: usize,
    pub decoder_sessions: DecoderSessionCount,
    pub reads: Arc<CursorReads>,
    pub codec_threads: CodecThreadAllocation,
    pub source_frame_pixels: u64,
    pub source_probe: Duration,
}

impl Host {
    /// A shared device alone does not say so: a software-decode host keeps one for the
    /// GPU-resident export.
    #[cfg_attr(
        not(all(
            any(target_os = "linux", target_os = "macos", target_os = "windows"),
            feature = "hwdecode"
        )),
        allow(clippy::unused_self)
    )]
    pub(crate) fn hardware_decode_ready(&self) -> bool {
        #[cfg(all(target_os = "linux", feature = "hwdecode"))]
        return self.decode_vulkan().is_some();
        #[cfg(all(target_os = "macos", feature = "hwdecode"))]
        return self.decode_metal().is_some();
        #[cfg(all(target_os = "windows", feature = "hwdecode"))]
        return self.decode_d3d12().is_some();
        #[cfg(not(all(
            any(target_os = "linux", target_os = "macos", target_os = "windows"),
            feature = "hwdecode"
        )))]
        false
    }

    /// Lets conform tell a stream only hardware decodes from one it cannot read.
    pub(crate) fn hardware_video_decode(&self) -> crate::assets::HardwareVideoDecode {
        use crate::assets::HardwareVideoDecode;
        #[cfg(all(target_os = "linux", feature = "hwdecode"))]
        return self
            .decode_vulkan()
            .map_or(HardwareVideoDecode::None, |context| {
                HardwareVideoDecode::Codecs(context.decode_codecs().to_vec())
            });
        #[cfg(all(target_os = "macos", feature = "hwdecode"))]
        return self
            .decode_metal()
            .map_or(HardwareVideoDecode::None, |context| {
                HardwareVideoDecode::Codecs(context.decode_codecs().to_vec())
            });
        #[cfg(not(all(any(target_os = "linux", target_os = "macos"), feature = "hwdecode")))]
        if self.hardware_decode_ready() {
            HardwareVideoDecode::PerStream
        } else {
            HardwareVideoDecode::None
        }
    }

    #[cfg(all(target_os = "macos", feature = "hwdecode"))]
    pub(crate) fn decode_metal(&self) -> Option<alpha2_native_gpu::SharedMetalContext> {
        self.shared_metal
            .clone()
            .filter(|_| self.decode_policy.allows_hardware())
    }

    /// A device made only for the GPU-resident export also exists under `software` and must not
    /// turn hardware decode on.
    #[cfg(all(target_os = "windows", feature = "hwdecode"))]
    pub(crate) fn decode_d3d12(&self) -> Option<alpha2_native_gpu::SharedD3D12Context> {
        self.shared_d3d12
            .clone()
            .filter(|context| self.decode_policy.allows_hardware() && context.decode_planes())
    }

    /// `None` also when the host reported at start that `FFmpeg` cannot open a video session there.
    #[cfg(all(target_os = "linux", feature = "hwdecode"))]
    pub(crate) fn decode_vulkan(&self) -> Option<alpha2_native_gpu::SharedVulkanContext> {
        self.shared_vulkan.clone().filter(|context| {
            self.decode_policy.allows_hardware() && context.video_session_hazard().is_none()
        })
    }

    pub(crate) fn export_providers(
        &self,
        project: &Project,
        size: RenderSize,
    ) -> Arc<crate::providers::ProviderMedia> {
        let providers = Arc::new(crate::providers::ProviderMedia::new(
            project,
            self.provider_files.clone(),
            (size.width, size.height),
            self.memory.provider_sources.bytes,
        ));
        self.failure.watch_providers(Arc::clone(&providers));
        providers
    }

    /// `encoder_fixed_threads` is the software encoder's own worker count when the picture fixes
    /// it (`OpenH264`).
    pub(crate) fn export_codec_split(
        &self,
        project: &Project,
        assets: &HashMap<String, PathBuf>,
        range: TimeRange,
        pipeline_enabled: bool,
        (software_video_encoder, encoder_fixed_threads): (bool, Option<usize>),
    ) -> ExportCodecSplit {
        // The synchronous comparison path decodes in software on the coordinator.
        let hardware_decode = pipeline_enabled && self.hardware_decode_ready();
        let decode_lookahead = configured_decode_lookahead(pipeline_enabled);
        let (workload_parallelism, reads) = visual_decode_reads(project, range);
        let probe_started = Instant::now();
        let admitted = *self
            .largest_video_frames
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        let largest_video = admitted.unwrap_or_else(|| {
            crate::host::max_video_source_frames(
                project,
                assets,
                self.conformed_video_frames.as_ref(),
            )
        });
        let source_probe = probe_started.elapsed();
        let source_frame_pixels = if hardware_decode || !software_video_encoder {
            0
        } else {
            largest_video.pixels
        };
        let requested_sessions = configured_decoder_session_override();
        let threads_for = |sessions: usize| {
            codec_thread_allocation(CodecThreadRequest {
                decode_concurrency: configured_decode_concurrency(
                    pipeline_enabled,
                    hardware_decode,
                    workload_parallelism,
                )
                .min(sessions),
                hardware_decode,
                source_frame_pixels,
                software_video_encoder,
                encoder_fixed_threads,
            })
        };
        let mut codec_threads =
            threads_for(requested_sessions.unwrap_or(crate::MAX_DECODER_SESSIONS));
        fit_session_threads(
            &mut codec_threads,
            hardware_decode,
            largest_video.bytes,
            self.memory.decoder_sessions.byte_budget(),
        );
        let decoder_sessions = decoder_session_count(
            workload_parallelism,
            decoder_session_bytes(
                largest_video.bytes,
                if hardware_decode {
                    codec_threads.hardware_decoder_threads
                } else {
                    codec_threads.decoder_threads
                },
            ),
            self.memory.decoder_sessions.bytes,
            requested_sessions,
        );
        // A shard with no session could decode nothing. The encoder keeps its workers: a software
        // encoder's thread count shapes its bitstream, which must not depend on decoder memory.
        if decoder_sessions.sessions < codec_threads.decode_concurrency {
            let encoder_threads = codec_threads.encoder_threads;
            codec_threads = threads_for(decoder_sessions.sessions);
            let decode_assigned = codec_threads
                .allocated_threads
                .saturating_sub(usize::try_from(codec_threads.encoder_threads).unwrap_or(0));
            codec_threads.encoder_threads = encoder_threads;
            codec_threads.allocated_threads =
                decode_assigned.saturating_add(usize::try_from(encoder_threads).unwrap_or(0));
        }
        ExportCodecSplit {
            hardware_decode,
            decode_lookahead,
            decoder_sessions,
            reads: Arc::new(reads),
            codec_threads,
            source_frame_pixels,
            source_probe,
        }
    }

    pub(crate) fn export_gpu_frames(&self, size: RenderSize, project: &Project) -> NativeGpuFrames {
        #[cfg(all(target_os = "linux", feature = "hwdecode"))]
        let mut gpu_frames = NativeGpuFrames::with_vulkan(size, self.decode_vulkan());
        #[cfg(all(target_os = "macos", feature = "hwdecode"))]
        let mut gpu_frames = NativeGpuFrames::with_metal(size, self.decode_metal());
        #[cfg(all(target_os = "windows", feature = "hwdecode"))]
        let mut gpu_frames = NativeGpuFrames::with_d3d12(size, self.decode_d3d12());
        #[cfg(not(all(
            any(target_os = "linux", target_os = "macos", target_os = "windows"),
            feature = "hwdecode"
        )))]
        let mut gpu_frames = NativeGpuFrames::new(size);
        gpu_frames.import_fallback = !self.decode_policy.requires_hardware();
        gpu_frames.color = alpha2_engine::color_policy(&project.settings.color);
        #[cfg(test)]
        if let Some(fault) = crate::frames::INJECT_SURFACE_FAULT.with(std::cell::Cell::take) {
            gpu_frames.inject(fault);
        }
        gpu_frames
    }

    pub(crate) fn decode_worker_config(&self, plan: &ExportDecodePlan) -> DecodeWorkerConfig {
        DecodeWorkerConfig {
            providers: Arc::clone(&plan.providers),
            decoding: self.failure.decoding(),
            depth: plan.decode_lookahead,
            byte_budget: self.memory.decode_queue.byte_budget(),
            concurrency: plan.codec_threads.decode_concurrency,
            decoder_threads: plan.codec_threads.decoder_threads,
            hardware_decoder_threads: plan.codec_threads.hardware_decoder_threads,
            max_sessions: plan.decoder_sessions.sessions,
            reads: Arc::clone(&plan.reads),
            reverse_cache_bytes: self.memory.reverse_cache.bytes,
            reverse_hardware_surfaces: configured_reverse_hardware_surfaces(self.decode_policy),
            stall_timeout: crate::config::configured_stall_timeout(),
            #[cfg(all(target_os = "linux", feature = "hwdecode"))]
            shared_vulkan: self.decode_vulkan(),
            #[cfg(all(target_os = "macos", feature = "hwdecode"))]
            shared_metal: self.decode_metal(),
            #[cfg(all(target_os = "windows", feature = "hwdecode"))]
            shared_d3d12: self.decode_d3d12(),
            #[cfg(all(
                any(target_os = "linux", target_os = "macos", target_os = "windows"),
                feature = "hwdecode"
            ))]
            hardware_required: self.decode_policy.requires_hardware(),
            #[cfg(test)]
            stall: None,
        }
    }
}

/// Lowers decoder threads until one session fits the decoder session `budget`
/// ([`decoder_session_bytes`]), down to one. One session at one thread is held whatever
/// `memory.scale` asked for, up to the unscaled budget.
pub(crate) fn fit_session_threads(
    threads: &mut alpha2_native::CodecThreadAllocation,
    hardware_decode: bool,
    frame_bytes: u64,
    budget: crate::memory::ByteBudget,
) {
    let allowed = budget.holding(decoder_session_bytes(frame_bytes, 1));
    let fitted = |threads: i64| {
        let mut threads = threads.max(1);
        while threads > 1 && decoder_session_bytes(frame_bytes, threads) > allowed {
            threads -= 1;
        }
        threads
    };
    if hardware_decode {
        threads.hardware_decoder_threads = fitted(threads.hardware_decoder_threads);
    }
    let decoder_threads = fitted(threads.decoder_threads);
    let freed = usize::try_from(threads.decoder_threads.saturating_sub(decoder_threads))
        .unwrap_or(0)
        .saturating_mul(threads.decode_concurrency);
    threads.allocated_threads = threads.allocated_threads.saturating_sub(freed);
    threads.decoder_threads = decoder_threads;
}