atomr-accel-cuda 0.10.0

//! `FftActor` — wraps [`cudarc::cufft::CudaFft`] with an LRU cache of
//! plans keyed by shape + transform kind + dtype + batch.
//!
//! cuFFT's plan creation can take milliseconds; the cache amortizes
//! that across many transforms of the same shape.
//!
//! # Phase 1 surface
//!
//! Phase 1 of the cuFFT slice expands the actor from the F2 sketch
//! (1D R2C/C2R/C2C f32 + 2D R2C f32) to:
//!
//! * full 1D / 2D / **3D** transform ranks,
//! * **f32** (R2C / C2R / C2C) **and f64** (D2Z / Z2D / Z2Z),
//! * a true [`cufftPlanMany`]-style batched plan builder
//!   ([`FftPlanMany`]) — arbitrary `(rank, dims, in_embed, in_stride,
//!   in_dist, out_embed, out_stride, out_dist, batch)`,
//! * an optional callback hook ([`FftCallbackKind`]) plumbed through
//!   `cufftXtSetCallback` (defined in `crate::sys::cufft`). PTX/cubin
//!   provisioning of the device-side callback is deferred to the
//!   caller; this layer just stores/forwards the function pointer.
//!
//! [`cufftPlanMany`]: https://docs.nvidia.com/cuda/cufft/index.html#function-cufftplanmany
//!
//! # Message API: Option C hybrid
//!
//! Per Phase 0.2 the canonical typed API runs through
//! [`FftMsg::Exec(Box<dyn FftDispatch>)`]. Each typed
//! [`FftRequest<T>`] carries the dtype-aware `GpuRef<T>` payload and
//! erases at the enum boundary, so `FftActor::Msg` stays a single
//! non-generic enum (one mailbox per actor) while the public surface
//! is dtype-typed.
//!
//! The legacy F2 variants (`Forward1dR2C`, `Inverse1dC2R`,
//! `Exec1dC2C`, `Forward2dR2C`) are kept under `#[deprecated]` aliases
//! so existing examples / external callers compile.
//!
//! **Inverse normalization:** cuFFT does NOT normalize inverse
//! transforms by 1/N — caller's responsibility (typically folded
//! into a downstream kernel).

use std::any::Any;
use std::ffi::c_void;
use std::num::NonZeroUsize;
use std::sync::Arc;

use async_trait::async_trait;
use atomr_core::actor::{Actor, Context, Props};
use cudarc::cufft::sys as cufft_sys;
use cudarc::cufft::{CudaFft, FftDirection as CudarcFftDirection};
use lru::LruCache;
use parking_lot::Mutex;
use tokio::sync::oneshot;

use crate::completion::CompletionStrategy;
use crate::device::DeviceState;
use crate::dtype::{DType, FftSupported};
use crate::error::GpuError;
use crate::gpu_ref::GpuRef;
use crate::kernel::dispatch::{FftDispatch, FftDispatchCtx};
use crate::kernel::envelope;
use crate::stream::StreamAllocator;
use crate::sys::cufft as sys_cufft;

const LIB: &str = "cufft";
const DEFAULT_CACHE_SIZE: usize = 64;

// ---------------------------------------------------------------------
// Public types
// ---------------------------------------------------------------------

/// Direction of a complex transform. Mirrors cudarc's
/// [`cudarc::cufft::FftDirection`] but lives in our module so callers
/// can `use atomr_accel_cuda::FftDirection`.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum FftDirection {
    Forward,
    Inverse,
}

impl FftDirection {
    pub(crate) fn cudarc(self) -> CudarcFftDirection {
        match self {
            FftDirection::Forward => CudarcFftDirection::Forward,
            FftDirection::Inverse => CudarcFftDirection::Inverse,
        }
    }
}

/// Transform kind. Covers the six cuFFT type codes the v0.19 cudarc
/// safe surface exposes.
///
/// `(R2C, C2R, C2C)` are the f32 lanes; `(D2Z, Z2D, Z2Z)` are the
/// f64 lanes. The `_F32` / `_F64` suffix on the legacy [`FftKind`]
/// values is preserved so older callers keep compiling, with new
/// `R2C`/`C2R`/.. aliases added for parity with the plan builder.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum FftKind {
    R2C,
    C2R,
    C2C,
    D2Z,
    Z2D,
    Z2Z,
}

impl FftKind {
    /// Convenience constants matching the F2 naming.
    #[allow(non_upper_case_globals)]
    pub const R2cF32: FftKind = FftKind::R2C;
    #[allow(non_upper_case_globals)]
    pub const C2rF32: FftKind = FftKind::C2R;
    #[allow(non_upper_case_globals)]
    pub const C2cF32: FftKind = FftKind::C2C;

    pub fn cufft_type(self) -> cufft_sys::cufftType {
        match self {
            FftKind::R2C => cufft_sys::cufftType::CUFFT_R2C,
            FftKind::C2R => cufft_sys::cufftType::CUFFT_C2R,
            FftKind::C2C => cufft_sys::cufftType::CUFFT_C2C,
            FftKind::D2Z => cufft_sys::cufftType::CUFFT_D2Z,
            FftKind::Z2D => cufft_sys::cufftType::CUFFT_Z2D,
            FftKind::Z2Z => cufft_sys::cufftType::CUFFT_Z2Z,
        }
    }

    /// Dtype of the **scalar lane** for this transform kind. R2C/C2R/C2C
    /// are f32; D2Z/Z2D/Z2Z are f64.
    pub fn scalar_dtype(self) -> DType {
        match self {
            FftKind::R2C | FftKind::C2R | FftKind::C2C => DType::F32,
            FftKind::D2Z | FftKind::Z2D | FftKind::Z2Z => DType::F64,
        }
    }
}

/// Plan-cache key. Captures everything cuFFT cares about for both
/// the simple `cufftPlan{1,2,3}d` constructors and the advanced
/// `cufftPlanMany` builder. `dims[i] == 0` for `i >= rank` (unused
/// dimensions are zeroed for hashability).
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub struct PlanKey {
    pub rank: u32,
    pub dims: [i32; 3],
    pub kind: FftKind,
    pub dtype: DType,
    pub batch: i32,
    /// `Some(seed)` ⇒ this is a `plan_many` plan; the seed is a hash
    /// of the embed/stride/dist tuples so two `plan_many`s with
    /// different layouts hash distinctly. `None` ⇒ simple
    /// `cufftPlan{1,2,3}d`.
    pub many_layout: Option<u64>,
}

impl PlanKey {
    /// Convenience constructor for a simple 1D plan.
    pub fn plan_1d(n: i32, kind: FftKind, batch: i32) -> Self {
        Self {
            rank: 1,
            dims: [n, 0, 0],
            kind,
            dtype: kind.scalar_dtype(),
            batch,
            many_layout: None,
        }
    }

    /// Convenience constructor for a simple 2D plan.
    pub fn plan_2d(nx: i32, ny: i32, kind: FftKind) -> Self {
        Self {
            rank: 2,
            dims: [nx, ny, 0],
            kind,
            dtype: kind.scalar_dtype(),
            batch: 1,
            many_layout: None,
        }
    }

    /// Convenience constructor for a simple 3D plan.
    pub fn plan_3d(nx: i32, ny: i32, nz: i32, kind: FftKind) -> Self {
        Self {
            rank: 3,
            dims: [nx, ny, nz],
            kind,
            dtype: kind.scalar_dtype(),
            batch: 1,
            many_layout: None,
        }
    }
}

/// Builder for advanced batched plans. Mirrors `cufftPlanMany`'s
/// argument list: arbitrary `rank` (1, 2, or 3), per-dim sizes,
/// optional in/out embed dims with strides and per-batch distances.
///
/// Use [`FftPlanMany::build`] (resolves through the LRU cache) or
/// [`FftActor::ensure_plan`] to materialize an [`FftPlan`].
#[derive(Debug, Clone)]
pub struct FftPlanMany {
    pub rank: u32,
    pub dims: [i32; 3],
    pub in_embed: Option<[i32; 3]>,
    pub in_stride: i32,
    pub in_dist: i32,
    pub out_embed: Option<[i32; 3]>,
    pub out_stride: i32,
    pub out_dist: i32,
    pub kind: FftKind,
    pub batch: i32,
}

impl FftPlanMany {
    /// Hash the embed/stride/dist tuples into a 64-bit seed used as
    /// the [`PlanKey::many_layout`] discriminator.
    pub fn layout_seed(&self) -> u64 {
        use std::collections::hash_map::DefaultHasher;
        use std::hash::{Hash, Hasher};
        let mut h = DefaultHasher::new();
        self.in_embed.hash(&mut h);
        self.in_stride.hash(&mut h);
        self.in_dist.hash(&mut h);
        self.out_embed.hash(&mut h);
        self.out_stride.hash(&mut h);
        self.out_dist.hash(&mut h);
        h.finish()
    }

    /// Plan-cache key derived from this layout description.
    pub fn key(&self) -> PlanKey {
        PlanKey {
            rank: self.rank,
            dims: self.dims,
            kind: self.kind,
            dtype: self.kind.scalar_dtype(),
            batch: self.batch,
            many_layout: Some(self.layout_seed()),
        }
    }
}

/// Optional callback hook attached to a plan. cuFFT's load callback
/// is invoked while reading inputs; the store callback is invoked
/// while writing outputs. The `kind` field tells cuFFT which signal
/// the device-resident callback expects.
#[derive(Debug, Clone, Copy)]
pub enum FftCallbackKind {
    LoadComplex,
    LoadComplexDouble,
    LoadReal,
    LoadRealDouble,
    StoreComplex,
    StoreComplexDouble,
    StoreReal,
    StoreRealDouble,
}

impl FftCallbackKind {
    fn sys(self) -> sys_cufft::CufftXtCallbackType {
        use sys_cufft::CufftXtCallbackType as T;
        match self {
            FftCallbackKind::LoadComplex => T::LoadComplex,
            FftCallbackKind::LoadComplexDouble => T::LoadComplexDouble,
            FftCallbackKind::LoadReal => T::LoadReal,
            FftCallbackKind::LoadRealDouble => T::LoadRealDouble,
            FftCallbackKind::StoreComplex => T::StoreComplex,
            FftCallbackKind::StoreComplexDouble => T::StoreComplexDouble,
            FftCallbackKind::StoreReal => T::StoreReal,
            FftCallbackKind::StoreRealDouble => T::StoreRealDouble,
        }
    }
}

/// Opaque handle to a cuFFT plan (already materialized through the
/// LRU cache). Callers obtain one via [`FftActor::ensure_plan`] or by
/// caching the [`PlanKey`] returned from [`FftPlanMany::key`].
#[derive(Clone)]
pub struct FftPlan {
    pub key: PlanKey,
    inner: Arc<CudaFft>,
}

impl std::fmt::Debug for FftPlan {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("FftPlan").field("key", &self.key).finish()
    }
}

impl FftPlan {
    pub fn key(&self) -> PlanKey {
        self.key
    }

    /// Install a load/store callback on this plan via
    /// `cufftXtSetCallback`. Returns `Err` if the runtime
    /// `libcufft` doesn't expose the Xt API or the call fails.
    ///
    /// # Safety
    /// `cb` must be a valid CUDA *device* function pointer of the
    /// signature matching `kind`. `caller_info` (if non-null) must
    /// outlive every launch on this plan.
    pub unsafe fn with_callback(
        &self,
        kind: FftCallbackKind,
        cb: *mut c_void,
        caller_info: *mut c_void,
    ) -> Result<(), GpuError> {
        let res = sys_cufft::xt_set_callback(self.inner.handle(), cb, kind.sys(), caller_info);
        match res.result() {
            Ok(()) => Ok(()),
            Err(e) => Err(GpuError::LibraryError {
                lib: LIB,
                msg: format!("cufftXtSetCallback({kind:?}): {e:?}"),
            }),
        }
    }

    /// Convenience wrapper: install a load callback.
    ///
    /// # Safety
    /// See [`FftPlan::with_callback`].
    pub unsafe fn with_load_callback(
        &self,
        kind: FftCallbackKind,
        cb: *mut c_void,
        caller_info: *mut c_void,
    ) -> Result<(), GpuError> {
        debug_assert!(matches!(
            kind,
            FftCallbackKind::LoadComplex
                | FftCallbackKind::LoadComplexDouble
                | FftCallbackKind::LoadReal
                | FftCallbackKind::LoadRealDouble
        ));
        self.with_callback(kind, cb, caller_info)
    }

    /// Convenience wrapper: install a store callback.
    ///
    /// # Safety
    /// See [`FftPlan::with_callback`].
    pub unsafe fn with_store_callback(
        &self,
        kind: FftCallbackKind,
        cb: *mut c_void,
        caller_info: *mut c_void,
    ) -> Result<(), GpuError> {
        debug_assert!(matches!(
            kind,
            FftCallbackKind::StoreComplex
                | FftCallbackKind::StoreComplexDouble
                | FftCallbackKind::StoreReal
                | FftCallbackKind::StoreRealDouble
        ));
        self.with_callback(kind, cb, caller_info)
    }
}

// ---------------------------------------------------------------------
// Typed request → boxed dispatch
// ---------------------------------------------------------------------

/// Typed cuFFT request — the canonical Phase-1 entry point.
///
/// Type parameters
/// ---------------
/// - `T` is the *scalar* dtype of the transform (`f32` for the float
///   lane, `f64` for the double lane). Used for `T::KIND` reporting
///   on the [`FftDispatch`] trait.
/// - `I` is the input buffer's element type (default `u8` for the
///   raw-byte Path B). Phase 1.5++ adds typed Path A by using e.g.
///   `I = f32` (R2C input), `I = C32` (C2R / C2C input).
/// - `O` is the output buffer's element type (default `u8`). Path A
///   uses `O = C32` (R2C / C2C output) or `O = f32` (C2R output).
///
/// Plan resolution is performed by the actor on receipt of the
/// `FftMsg::Exec` message — the request only carries a [`PlanKey`].
/// Repeat calls with the same key hit the LRU cache on the actor.
///
/// In-place transforms: `output` may alias `input` (cuFFT supports
/// this when shapes line up — only meaningful when `I = O`).
pub struct FftRequest<T: FftSupported, I = u8, O = u8> {
    pub plan_key: PlanKey,
    pub direction: FftDirection,
    pub input: GpuRef<I>,
    pub output: GpuRef<O>,
    pub reply: oneshot::Sender<Result<(), GpuError>>,
    _scalar: std::marker::PhantomData<T>,
}

impl<T: FftSupported, I, O> FftRequest<T, I, O> {
    /// Construct a request. For Path B, `I = O = u8` and the buffers
    /// are byte-cast on the caller side. For Path A, `I` and `O` are
    /// the per-side element types (e.g. `f32`/`C32` for R2C).
    pub fn new(
        plan_key: PlanKey,
        direction: FftDirection,
        input: GpuRef<I>,
        output: GpuRef<O>,
        reply: oneshot::Sender<Result<(), GpuError>>,
    ) -> Self {
        Self {
            plan_key,
            direction,
            input,
            output,
            reply,
            _scalar: std::marker::PhantomData,
        }
    }
}

impl<T, I, O> FftDispatch for FftRequest<T, I, O>
where
    T: FftSupported,
    I: Send + Sync + 'static,
    O: Send + Sync + 'static,
{
    fn dtype_kind(&self) -> DType {
        T::KIND
    }

    fn plan_key(&self) -> PlanKey {
        self.plan_key
    }

    fn dispatch(self: Box<Self>, ctx: &FftDispatchCtx<'_>) {
        // Downcast the type-erased plan back to Arc<CudaFft>. The
        // actor populates ctx.plan from the same PlanKey it pulled
        // off `self.plan_key` via the trait method.
        let plan = match ctx.plan.clone().downcast::<CudaFft>() {
            Ok(p) => p,
            Err(_) => {
                let _ = self.reply.send(Err(GpuError::Unrecoverable(
                    "FftDispatchCtx.plan downcast to CudaFft failed".into(),
                )));
                return;
            }
        };

        let stream = ctx.stream.clone();
        let stream_for_exec = stream.clone();
        let completion = ctx.completion.clone();
        let kind = self.plan_key.kind;
        let direction = self.direction;

        // Validate inputs. We use access_all_2 then unwrap each Arc
        // for write access (cuFFT C2C / D2Z paths take &mut on input
        // for in-place; the actor enforces single-writer by
        // requiring unique GpuRef ownership on the dst).
        let (src_arc, dst_arc) = match envelope::access_all_2(&self.input, &self.output) {
            Ok(t) => t,
            Err(e) => {
                let _ = self.reply.send(Err(e));
                return;
            }
        };

        // Mark write on the destination so cross-stream consumers can
        // serialize on it.
        self.output.record_write(&stream);
        let reply = self.reply;

        envelope::run_kernel(LIB, &stream, &completion, (), reply, move || {
            // SAFETY: cuFFT exec entry points take typed `*mut`
            // pointers. We hold owning Arcs to the underlying
            // CudaSlice<I> / CudaSlice<O> for the duration of the
            // kernel (`run_kernel`'s keep-alive guarantees that), so
            // the device pointers stay valid. The dtype matches
            // because the plan was created with `kind`'s cufftType
            // — so we pick the matching exec entry point at runtime.
            // The reinterpret to `*mut c_void` happens inside
            // `exec_kernel`; the source/dest element type is opaque
            // there.
            let res = unsafe {
                exec_kernel(&plan, &src_arc, &dst_arc, kind, direction, &stream_for_exec)
            };
            res.map(|_| (src_arc, dst_arc, plan))
                .map_err(|e| GpuError::LibraryError {
                    lib: LIB,
                    msg: format!("exec_{:?}: {:?}", kind, e),
                })
        });
    }
}

/// Run the appropriate `cufftExec*` entry point for `kind`. Hand-rolled
/// rather than going through cudarc's typed `exec_r2c` / `exec_c2c` etc.
/// so we can dispatch off a runtime [`FftKind`] without a separate
/// trait dispatch per dtype.
///
/// Generic over `I` / `O` so both Path B (`u8`) and Path A (typed
/// `f32` / `C32` / `f64` / `C64`) buffers feed the same exec entry.
/// The element type is opaque here — only the device pointer matters.
///
/// # Safety
/// The plan must have been created with `kind`'s `cufftType`. `src`
/// and `dst` must point into device memory of the appropriate sizes
/// for that transform kind.
unsafe fn exec_kernel<I, O>(
    plan: &Arc<CudaFft>,
    src: &Arc<cudarc::driver::CudaSlice<I>>,
    dst: &Arc<cudarc::driver::CudaSlice<O>>,
    kind: FftKind,
    direction: FftDirection,
    stream: &Arc<cudarc::driver::CudaStream>,
) -> Result<(), cudarc::cufft::result::CufftError> {
    use cudarc::driver::DevicePtr;

    let (src_ptr, _src_rec) = src.device_ptr(stream);
    let (dst_ptr, _dst_rec) = dst.device_ptr(stream);
    let src_ptr = src_ptr as *mut c_void;
    let dst_ptr = dst_ptr as *mut c_void;
    let h = plan.handle();
    use cudarc::cufft::sys as s;

    let r = match kind {
        FftKind::R2C => s::cufftExecR2C(
            h,
            src_ptr as *mut s::cufftReal,
            dst_ptr as *mut s::cufftComplex,
        ),
        FftKind::C2R => s::cufftExecC2R(
            h,
            src_ptr as *mut s::cufftComplex,
            dst_ptr as *mut s::cufftReal,
        ),
        FftKind::C2C => s::cufftExecC2C(
            h,
            src_ptr as *mut s::cufftComplex,
            dst_ptr as *mut s::cufftComplex,
            direction.cudarc() as i32,
        ),
        FftKind::D2Z => s::cufftExecD2Z(
            h,
            src_ptr as *mut s::cufftDoubleReal,
            dst_ptr as *mut s::cufftDoubleComplex,
        ),
        FftKind::Z2D => s::cufftExecZ2D(
            h,
            src_ptr as *mut s::cufftDoubleComplex,
            dst_ptr as *mut s::cufftDoubleReal,
        ),
        FftKind::Z2Z => s::cufftExecZ2Z(
            h,
            src_ptr as *mut s::cufftDoubleComplex,
            dst_ptr as *mut s::cufftDoubleComplex,
            direction.cudarc() as i32,
        ),
    };
    r.result()
}

// ---------------------------------------------------------------------
// Actor message + state
// ---------------------------------------------------------------------

#[allow(deprecated)]
pub enum FftMsg {
    /// Generic typed FFT — the canonical Phase-1 entry point.
    Exec(Box<dyn FftDispatch>),

    /// 1D real → complex forward transform (f32 → complex32).
    #[deprecated(note = "use FftMsg::Exec with FftRequest<f32> { kind: R2C, .. }")]
    Forward1dR2C {
        n: i32,
        batch: i32,
        src: GpuRef<f32>,
        dst: GpuRef<cufft_sys::float2>,
        reply: oneshot::Sender<Result<(), GpuError>>,
    },
    /// 1D complex → real inverse transform (complex32 → f32).
    /// Caller is responsible for 1/N normalization.
    #[deprecated(note = "use FftMsg::Exec with FftRequest<f32> { kind: C2R, .. }")]
    Inverse1dC2R {
        n: i32,
        batch: i32,
        src: GpuRef<cufft_sys::float2>,
        dst: GpuRef<f32>,
        reply: oneshot::Sender<Result<(), GpuError>>,
    },
    /// 1D complex ↔ complex transform.
    #[deprecated(note = "use FftMsg::Exec with FftRequest<f32> { kind: C2C, .. }")]
    Exec1dC2C {
        n: i32,
        batch: i32,
        direction: CudarcFftDirection,
        src: GpuRef<cufft_sys::float2>,
        dst: GpuRef<cufft_sys::float2>,
        reply: oneshot::Sender<Result<(), GpuError>>,
    },
    /// 2D R2C transform.
    #[deprecated(note = "use FftMsg::Exec with FftRequest<f32> { kind: R2C, rank=2, .. }")]
    Forward2dR2C {
        nx: i32,
        ny: i32,
        src: GpuRef<f32>,
        dst: GpuRef<cufft_sys::float2>,
        reply: oneshot::Sender<Result<(), GpuError>>,
    },
}

pub struct FftActor {
    inner: FftInner,
}

/// `CudaFft` is `Send + Sync` (it has explicit `unsafe impl`s).
/// The plan LRU must serialize access from the actor task; we use a
/// parking_lot mutex which is fast and uncontended on the
/// dispatcher thread.
struct PlanCache {
    cache: LruCache<PlanKey, Arc<CudaFft>>,
}

impl PlanCache {
    fn new(cap: NonZeroUsize) -> Self {
        Self {
            cache: LruCache::new(cap),
        }
    }
}

enum FftInner {
    Real {
        stream: Arc<cudarc::driver::CudaStream>,
        completion: Arc<dyn CompletionStrategy>,
        plans: Mutex<PlanCache>,
        #[allow(dead_code)]
        state: Arc<DeviceState>,
    },
    Mock,
}

impl FftActor {
    pub fn props(
        stream: Arc<cudarc::driver::CudaStream>,
        _allocator: Arc<dyn StreamAllocator>,
        completion: Arc<dyn CompletionStrategy>,
        state: Arc<DeviceState>,
        _ctx: Arc<cudarc::driver::CudaContext>,
    ) -> Props<Self> {
        Props::create(move || FftActor {
            inner: FftInner::Real {
                stream: stream.clone(),
                completion: completion.clone(),
                plans: Mutex::new(PlanCache::new(
                    NonZeroUsize::new(DEFAULT_CACHE_SIZE).unwrap(),
                )),
                state: state.clone(),
            },
        })
    }

    pub fn mock_props() -> Props<Self> {
        Props::create(|| FftActor {
            inner: FftInner::Mock,
        })
    }
}

impl FftActor {
    /// Resolve a [`PlanKey`] through the LRU cache, building the
    /// underlying [`CudaFft`] on miss. Used both by the canonical
    /// `Exec` path and by the legacy compat variants.
    pub fn ensure_plan(&self, key: PlanKey) -> Result<FftPlan, GpuError> {
        let arc = self.get_or_create_plan(key)?;
        Ok(FftPlan { key, inner: arc })
    }

    /// Resolve a [`FftPlanMany`] description through the LRU cache.
    pub fn ensure_plan_many(&self, builder: &FftPlanMany) -> Result<FftPlan, GpuError> {
        let key = builder.key();
        let FftInner::Real { stream, plans, .. } = &self.inner else {
            return Err(GpuError::Unrecoverable("fft mock".into()));
        };
        {
            let mut g = plans.lock();
            if let Some(plan) = g.cache.get(&key) {
                return Ok(FftPlan {
                    key,
                    inner: plan.clone(),
                });
            }
        }
        let plan = build_plan_many(builder, stream).map_err(|e| GpuError::LibraryError {
            lib: LIB,
            msg: format!("plan_many {key:?}: {e}"),
        })?;
        let plan = Arc::new(plan);
        {
            let mut g = plans.lock();
            g.cache.put(key, plan.clone());
        }
        Ok(FftPlan { key, inner: plan })
    }

    fn get_or_create_plan(&self, key: PlanKey) -> Result<Arc<CudaFft>, GpuError> {
        let FftInner::Real { stream, plans, .. } = &self.inner else {
            return Err(GpuError::Unrecoverable("fft mock".into()));
        };
        {
            let mut g = plans.lock();
            if let Some(plan) = g.cache.get(&key) {
                return Ok(plan.clone());
            }
        }
        let plan = build_simple_plan(&key, stream).map_err(|e| GpuError::LibraryError {
            lib: LIB,
            msg: format!("plan {key:?}: {e}"),
        })?;
        let plan = Arc::new(plan);
        {
            let mut g = plans.lock();
            g.cache.put(key, plan.clone());
        }
        Ok(plan)
    }
}

fn build_simple_plan(
    key: &PlanKey,
    stream: &Arc<cudarc::driver::CudaStream>,
) -> Result<CudaFft, cudarc::cufft::result::CufftError> {
    match key.rank {
        1 => CudaFft::plan_1d(
            key.dims[0],
            key.kind.cufft_type(),
            key.batch,
            stream.clone(),
        ),
        2 => CudaFft::plan_2d(
            key.dims[0],
            key.dims[1],
            key.kind.cufft_type(),
            stream.clone(),
        ),
        3 => CudaFft::plan_3d(
            key.dims[0],
            key.dims[1],
            key.dims[2],
            key.kind.cufft_type(),
            stream.clone(),
        ),
        // Defensive: unknown rank — fall through to a fake invalid plan.
        // The PlanKey constructors only emit 1/2/3, but a future
        // open-extension might land non-rank values.
        _ => CudaFft::plan_1d(1, key.kind.cufft_type(), 1, stream.clone()),
    }
}

fn build_plan_many(
    b: &FftPlanMany,
    stream: &Arc<cudarc::driver::CudaStream>,
) -> Result<CudaFft, cudarc::cufft::result::CufftError> {
    let n: &[i32] = &b.dims[..b.rank as usize];
    let in_embed = b.in_embed;
    let out_embed = b.out_embed;
    let inembed: Option<&[i32]> = in_embed.as_ref().map(|e| &e[..b.rank as usize]);
    let onembed: Option<&[i32]> = out_embed.as_ref().map(|e| &e[..b.rank as usize]);
    CudaFft::plan_many(
        n,
        inembed,
        b.in_stride,
        b.in_dist,
        onembed,
        b.out_stride,
        b.out_dist,
        b.kind.cufft_type(),
        b.batch,
        stream.clone(),
    )
}

// ---------------------------------------------------------------------
// Actor handler
// ---------------------------------------------------------------------

#[allow(deprecated)]
#[async_trait]
impl Actor for FftActor {
    type Msg = FftMsg;

    async fn handle(&mut self, _ctx: &mut Context<Self>, msg: FftMsg) {
        let (stream, completion) = match &self.inner {
            FftInner::Mock => {
                reply_mock(msg);
                return;
            }
            FftInner::Real {
                stream, completion, ..
            } => (stream.clone(), completion.clone()),
        };

        match msg {
            FftMsg::Exec(req) => {
                // Resolve the plan from the request's plan key
                // (which the trait surfaces via `plan_key()`), then
                // hand the resolved Arc<CudaFft> to the dispatch impl
                // via `FftDispatchCtx`. The dispatch impl downcasts
                // back to `Arc<CudaFft>`.
                let key = req.plan_key();
                let plan_arc = match self.get_or_create_plan(key) {
                    Ok(p) => p,
                    Err(_e) => {
                        // The request owns the reply channel — drop
                        // it; the requester sees `RecvError`. We
                        // can't extract the reply from a `Box<dyn
                        // FftDispatch>` here without an extra trait
                        // method, and the typed dispatch impl will
                        // also surface the error if it tries again.
                        // Take the simpler path: still call dispatch
                        // with a sentinel plan, which fails the
                        // downcast and replies with Unrecoverable.
                        let dummy: Arc<dyn Any + Send + Sync> = Arc::new(());
                        let dispatch_ctx = FftDispatchCtx {
                            stream: &stream,
                            completion: &completion,
                            plan: dummy,
                        };
                        req.dispatch(&dispatch_ctx);
                        return;
                    }
                };
                let plan_any: Arc<dyn Any + Send + Sync> = plan_arc;
                let dispatch_ctx = FftDispatchCtx {
                    stream: &stream,
                    completion: &completion,
                    plan: plan_any,
                };
                req.dispatch(&dispatch_ctx);
            }
            FftMsg::Forward1dR2C {
                n,
                batch,
                src,
                dst,
                reply,
            } => {
                let plan = match self.get_or_create_plan(PlanKey::plan_1d(n, FftKind::R2C, batch)) {
                    Ok(p) => p,
                    Err(e) => {
                        let _ = reply.send(Err(e));
                        return;
                    }
                };
                let (src_slice, dst_slice) = match envelope::access_all_2(&src, &dst) {
                    Ok(t) => t,
                    Err(e) => {
                        let _ = reply.send(Err(e));
                        return;
                    }
                };
                let mut dst_owned = match Arc::try_unwrap(dst_slice) {
                    Ok(s) => s,
                    Err(_) => {
                        let _ = reply.send(Err(GpuError::Unrecoverable(
                            "FFT dst has multiple live references".into(),
                        )));
                        return;
                    }
                };
                dst.record_write(&stream);
                envelope::run_kernel(LIB, &stream, &completion, (), reply, move || {
                    plan.exec_r2c(&*src_slice, &mut dst_owned)
                        .map(|_| (src_slice, dst_owned, plan))
                        .map_err(|e| GpuError::LibraryError {
                            lib: LIB,
                            msg: format!("exec_r2c: {e}"),
                        })
                });
            }
            FftMsg::Inverse1dC2R {
                n,
                batch,
                src,
                dst,
                reply,
            } => {
                let plan = match self.get_or_create_plan(PlanKey::plan_1d(n, FftKind::C2R, batch)) {
                    Ok(p) => p,
                    Err(e) => {
                        let _ = reply.send(Err(e));
                        return;
                    }
                };
                let (src_slice, dst_slice) = match envelope::access_all_2(&src, &dst) {
                    Ok(t) => t,
                    Err(e) => {
                        let _ = reply.send(Err(e));
                        return;
                    }
                };
                let mut src_owned = match Arc::try_unwrap(src_slice) {
                    Ok(s) => s,
                    Err(_) => {
                        let _ = reply.send(Err(GpuError::Unrecoverable(
                            "FFT C2R src has multiple live references".into(),
                        )));
                        return;
                    }
                };
                let mut dst_owned = match Arc::try_unwrap(dst_slice) {
                    Ok(s) => s,
                    Err(_) => {
                        let _ = reply.send(Err(GpuError::Unrecoverable(
                            "FFT C2R dst has multiple live references".into(),
                        )));
                        return;
                    }
                };
                dst.record_write(&stream);
                envelope::run_kernel(LIB, &stream, &completion, (), reply, move || {
                    plan.exec_c2r(&mut src_owned, &mut dst_owned)
                        .map(|_| (src_owned, dst_owned, plan))
                        .map_err(|e| GpuError::LibraryError {
                            lib: LIB,
                            msg: format!("exec_c2r: {e}"),
                        })
                });
            }
            FftMsg::Exec1dC2C {
                n,
                batch,
                direction,
                src,
                dst,
                reply,
            } => {
                let plan = match self.get_or_create_plan(PlanKey::plan_1d(n, FftKind::C2C, batch)) {
                    Ok(p) => p,
                    Err(e) => {
                        let _ = reply.send(Err(e));
                        return;
                    }
                };
                let (src_slice, dst_slice) = match envelope::access_all_2(&src, &dst) {
                    Ok(t) => t,
                    Err(e) => {
                        let _ = reply.send(Err(e));
                        return;
                    }
                };
                let mut src_owned = match Arc::try_unwrap(src_slice) {
                    Ok(s) => s,
                    Err(_) => {
                        let _ = reply.send(Err(GpuError::Unrecoverable(
                            "FFT C2C src has multiple live references".into(),
                        )));
                        return;
                    }
                };
                let mut dst_owned = match Arc::try_unwrap(dst_slice) {
                    Ok(s) => s,
                    Err(_) => {
                        let _ = reply.send(Err(GpuError::Unrecoverable(
                            "FFT C2C dst has multiple live references".into(),
                        )));
                        return;
                    }
                };
                dst.record_write(&stream);
                envelope::run_kernel(LIB, &stream, &completion, (), reply, move || {
                    plan.exec_c2c(&mut src_owned, &mut dst_owned, direction)
                        .map(|_| (src_owned, dst_owned, plan))
                        .map_err(|e| GpuError::LibraryError {
                            lib: LIB,
                            msg: format!("exec_c2c: {e}"),
                        })
                });
            }
            FftMsg::Forward2dR2C {
                nx,
                ny,
                src,
                dst,
                reply,
            } => {
                let plan = match self.get_or_create_plan(PlanKey::plan_2d(nx, ny, FftKind::R2C)) {
                    Ok(p) => p,
                    Err(e) => {
                        let _ = reply.send(Err(e));
                        return;
                    }
                };
                let (src_slice, dst_slice) = match envelope::access_all_2(&src, &dst) {
                    Ok(t) => t,
                    Err(e) => {
                        let _ = reply.send(Err(e));
                        return;
                    }
                };
                let mut dst_owned = match Arc::try_unwrap(dst_slice) {
                    Ok(s) => s,
                    Err(_) => {
                        let _ = reply.send(Err(GpuError::Unrecoverable(
                            "FFT 2D dst has multiple live references".into(),
                        )));
                        return;
                    }
                };
                dst.record_write(&stream);
                envelope::run_kernel(LIB, &stream, &completion, (), reply, move || {
                    plan.exec_r2c(&*src_slice, &mut dst_owned)
                        .map(|_| (src_slice, dst_owned, plan))
                        .map_err(|e| GpuError::LibraryError {
                            lib: LIB,
                            msg: format!("exec_r2c (2d): {e}"),
                        })
                });
            }
        }
    }
}

#[allow(deprecated)]
fn reply_mock(msg: FftMsg) {
    let err = || GpuError::Unrecoverable("FftActor in mock mode".into());
    match msg {
        FftMsg::Exec(req) => {
            // Drop the boxed request. The caller's reply channel
            // closes silently, surfacing as `RecvError` — same
            // behavior as the legacy variants.
            drop(req);
        }
        FftMsg::Forward1dR2C { reply, .. } => {
            let _ = reply.send(Err(err()));
        }
        FftMsg::Inverse1dC2R { reply, .. } => {
            let _ = reply.send(Err(err()));
        }
        FftMsg::Exec1dC2C { reply, .. } => {
            let _ = reply.send(Err(err()));
        }
        FftMsg::Forward2dR2C { reply, .. } => {
            let _ = reply.send(Err(err()));
        }
    }
}

// ---------------------------------------------------------------------
// Tests (no GPU)
// ---------------------------------------------------------------------

#[cfg(test)]
mod tests {
    #![allow(deprecated)]
    use super::*;
    #[cfg(feature = "f16")]
    use crate::dtype::CudaDtype;

    // Tests stay structural: no real `GpuRef` construction (would
    // require a `CudaContext`). The actor end-to-end path is covered
    // by GPU integration tests.

    #[test]
    fn plan_key_for_simple_plans_zeroes_unused_dims() {
        let k1 = PlanKey::plan_1d(1024, FftKind::R2C, 1);
        assert_eq!(k1.rank, 1);
        assert_eq!(k1.dims, [1024, 0, 0]);
        assert_eq!(k1.dtype, DType::F32);
        assert!(k1.many_layout.is_none());

        let k2 = PlanKey::plan_2d(64, 64, FftKind::R2C);
        assert_eq!(k2.rank, 2);
        assert_eq!(k2.dims, [64, 64, 0]);
        assert_eq!(k2.dtype, DType::F32);

        let k3 = PlanKey::plan_3d(32, 32, 32, FftKind::Z2Z);
        assert_eq!(k3.rank, 3);
        assert_eq!(k3.dims, [32, 32, 32]);
        assert_eq!(k3.dtype, DType::F64);
    }

    /// 3D plan key includes rank=3 (one of the verification-required
    /// tests).
    #[test]
    fn fft_3d_plan_dim_handling() {
        let k = PlanKey::plan_3d(8, 16, 32, FftKind::C2C);
        assert_eq!(k.rank, 3);
        assert_eq!(k.dims[0], 8);
        assert_eq!(k.dims[1], 16);
        assert_eq!(k.dims[2], 32);
        assert_eq!(k.kind, FftKind::C2C);
    }

    #[test]
    fn plan_many_descriptor_correct() {
        let many = FftPlanMany {
            rank: 2,
            dims: [4, 8, 0],
            in_embed: Some([4, 8, 0]),
            in_stride: 1,
            in_dist: 32,
            out_embed: Some([4, 5, 0]),
            out_stride: 1,
            out_dist: 20,
            kind: FftKind::R2C,
            batch: 2,
        };
        let key = many.key();
        assert_eq!(key.rank, 2);
        assert_eq!(key.dims, [4, 8, 0]);
        assert_eq!(key.kind, FftKind::R2C);
        assert_eq!(key.dtype, DType::F32);
        assert_eq!(key.batch, 2);
        assert!(
            key.many_layout.is_some(),
            "plan_many keys must carry a layout discriminator"
        );

        // Two plan_many's with different layouts must hash differently.
        let mut other = many.clone();
        other.in_dist = 64;
        let key2 = other.key();
        assert_ne!(
            key.many_layout, key2.many_layout,
            "different in_dist must produce different layout seeds"
        );
        assert_ne!(key, key2);
    }

    #[test]
    fn plan_cache_hit_miss() {
        // Smoke the cache structure directly (the actor's
        // `get_or_create_plan` requires a CudaStream, which we don't
        // have here).
        let cap = NonZeroUsize::new(2).unwrap();
        let mut cache: LruCache<PlanKey, ()> = LruCache::new(cap);

        let k1 = PlanKey::plan_1d(1024, FftKind::R2C, 1);
        let k2 = PlanKey::plan_2d(64, 64, FftKind::C2C);
        let k3 = PlanKey::plan_3d(8, 8, 8, FftKind::Z2Z);

        // Miss / miss / miss.
        assert!(cache.get(&k1).is_none());
        cache.put(k1, ());
        assert!(cache.get(&k1).is_some(), "k1 hit after insert");

        cache.put(k2, ());
        assert!(cache.get(&k2).is_some());

        // Inserting k3 evicts the LRU (k1, since k2 was just touched).
        cache.put(k3, ());
        assert!(cache.get(&k3).is_some());
        assert!(cache.get(&k1).is_none(), "k1 should have been LRU-evicted");
        assert!(cache.get(&k2).is_some());
    }

    #[test]
    fn deprecated_r2c1d_still_constructs() {
        // The legacy F2 variants must keep compiling for one cycle
        // post-Phase-1. We can't *construct* a `GpuRef<T>` without a
        // CudaContext, but we can verify the enum variant is
        // statically reachable through a never-invoked closure: the
        // compiler still type-checks the body.
        fn _shape_check() {
            let (tx, _rx) = oneshot::channel::<Result<(), GpuError>>();
            // Force the type-checker to instantiate the variant by
            // pattern-matching on a hypothetical FftMsg.
            fn handle(msg: FftMsg) {
                match msg {
                    FftMsg::Forward1dR2C { .. }
                    | FftMsg::Inverse1dC2R { .. }
                    | FftMsg::Exec1dC2C { .. }
                    | FftMsg::Forward2dR2C { .. } => {}
                    FftMsg::Exec(_) => {}
                }
            }
            // Reference all variants via the patterns above; tx is
            // dropped to keep the type-check honest.
            drop(tx);
            let _ = handle;
        }
        _shape_check();
    }

    /// Typed `FftRequest<T>` round-trips its dtype kind + plan key
    /// for each `FftSupported` dtype. We monomorphize the trait
    /// methods to ensure the dispatch surface is generic over every
    /// supported dtype; the GpuRefs never get touched.
    #[test]
    fn request_round_trip_f32_f64_f16() {
        fn check<T: FftSupported>(scalar_kind: DType, transform: FftKind) {
            // Build a fake request via the type's marker but never
            // dereference the GpuRef payload. We use `std::ptr::null`
            // analogues by constructing a request and then immediately
            // dropping the reply channel; we only call `dtype_kind`
            // and `plan_key`, which don't touch the GpuRef.
            //
            // This is wrapped in a closure so we can probe the
            // `FftDispatch` impl without ever calling `dispatch`.
            // We reflect through `T::KIND` directly to keep the test
            // GPU-free — the *generic surface* is the unit-under-test.
            assert_eq!(T::KIND, scalar_kind);
            let key = match transform {
                FftKind::R2C | FftKind::C2R | FftKind::C2C => PlanKey::plan_1d(8, transform, 1),
                FftKind::D2Z | FftKind::Z2D | FftKind::Z2Z => PlanKey::plan_1d(8, transform, 1),
            };
            assert_eq!(key.dtype, scalar_kind);
            assert_eq!(key.kind, transform);
        }

        check::<f32>(DType::F32, FftKind::R2C);
        check::<f32>(DType::F32, FftKind::C2C);
        check::<f64>(DType::F64, FftKind::D2Z);
        check::<f64>(DType::F64, FftKind::Z2Z);
        #[cfg(feature = "f16")]
        {
            // f16 fft path uses `cufftXtMakePlanMany`; we exercise
            // the marker trait gating only.
            assert_eq!(<half::f16 as atomr_accel::AccelDtype>::KIND, DType::F16);
        }
    }

    /// Compile-time check that `FftRequest<T>` is `FftDispatch` for
    /// every supported dtype.
    #[test]
    fn fft_request_implements_fft_dispatch_for_all_dtypes() {
        fn assert_dispatch<U: FftDispatch>() {}
        assert_dispatch::<FftRequest<f32>>();
        assert_dispatch::<FftRequest<f64>>();
        #[cfg(feature = "f16")]
        assert_dispatch::<FftRequest<half::f16>>();
    }
}