Skip to main content

strided_basic/
gather_plan.rs

1//! Prepared indexed plans over raw strided value and index layouts.
2//!
3//! This module owns the generic gather, dynamic-slice/update, and scatter
4//! traversals used by the erased replay layer. It models the XLA/tenferro
5//! indexed shape vocabulary, but keeps tensor allocation, dtype promotion, and
6//! frontend error policy outside `strided-kernel`.
7
8use core::{mem::MaybeUninit, ops::Add};
9
10use crate::copy_plan::{replay_fused_raw, CopyPlan, OverwriteWriter, ReadModifyWrite};
11use crate::raw_ops::{fuse_pair_layout, FusedPairLayout};
12use crate::{
13    MaybeSendSync, RawStridedMut, RawStridedRef, Result, StridedError, RAW_FUSED_RANK_LIMIT,
14};
15
16#[cfg(feature = "parallel")]
17type AxisVec<T> = smallvec::SmallVec<[T; RAW_FUSED_RANK_LIMIT]>;
18#[cfg(not(feature = "parallel"))]
19type AxisVec<T> = Vec<T>;
20
21/// Gather configuration shared by generic and erased replay.
22///
23/// The fields follow the usual gather vocabulary:
24///
25/// - `start_index_map[component]` names the operand axis controlled by a
26///   component in the index vector;
27/// - `collapsed_slice_dims` names operand axes whose slice size is one and
28///   which do not appear as output window axes;
29/// - `offset_dims` names output axes that represent window offsets;
30/// - output axes not in `offset_dims` are batch axes from `start_indices`;
31/// - `index_vector_dim == start_indices_rank` represents scalar index vectors.
32#[derive(Clone, Debug, Eq, PartialEq)]
33pub struct GatherSpec {
34    pub offset_dims: Vec<usize>,
35    pub collapsed_slice_dims: Vec<usize>,
36    pub start_index_map: Vec<usize>,
37    pub index_vector_dim: usize,
38    pub slice_sizes: Vec<usize>,
39}
40
41/// Index scalar types accepted by [`GatherPlan`].
42pub trait GatherIndex: Copy + MaybeSendSync {
43    fn to_i64(self) -> i64;
44}
45
46impl GatherIndex for i32 {
47    #[inline]
48    fn to_i64(self) -> i64 {
49        i64::from(self)
50    }
51}
52
53impl GatherIndex for i64 {
54    #[inline]
55    fn to_i64(self) -> i64 {
56        self
57    }
58}
59
60/// A compiled gather traversal for one value layout, index layout, and output layout.
61#[derive(Clone, Debug)]
62pub struct GatherPlan {
63    operand_dims: AxisVec<usize>,
64    operand_strides: AxisVec<isize>,
65    index_dims: AxisVec<usize>,
66    index_strides: AxisVec<isize>,
67    dest_dims: AxisVec<usize>,
68    dest_strides: AxisVec<isize>,
69    spec: GatherSpec,
70    replay: GatherReplay,
71    total: usize,
72}
73
74#[derive(Clone, Copy, Debug)]
75struct GatherReplayAxis {
76    dest_step: isize,
77    dest_reset: isize,
78    window_step: isize,
79    window_reset: isize,
80    index_batch_step: isize,
81    index_batch_reset: isize,
82}
83
84#[derive(Clone, Debug)]
85struct GatherReplay {
86    axes: AxisVec<GatherReplayAxis>,
87    index_component_offsets: AxisVec<isize>,
88    index_operand_strides: AxisVec<isize>,
89}
90
91struct GatherReplayState {
92    coords: AxisVec<usize>,
93    dest_offset: isize,
94    window_offset: isize,
95    index_batch_offset: isize,
96}
97
98#[derive(Clone, Copy, Debug)]
99struct WindowReplayAxis {
100    source_step: isize,
101    source_reset: isize,
102    dest_step: isize,
103    dest_reset: isize,
104}
105
106#[derive(Clone, Debug)]
107struct WindowReplay {
108    shape: AxisVec<usize>,
109    axes: AxisVec<WindowReplayAxis>,
110}
111
112struct WindowReplayState {
113    coords: CoordScratch,
114    source_offset: isize,
115    dest_offset: isize,
116}
117
118#[derive(Clone, Debug)]
119struct ScatterReplay {
120    batch: WindowReplay,
121    window: WindowReplay,
122    index_component_offsets: AxisVec<isize>,
123}
124
125impl ScatterReplay {
126    fn compile(
127        batch_shape: &[usize],
128        index_dims: &[usize],
129        index_strides: &[isize],
130        index_vector_dim: usize,
131        index_component_count: usize,
132        update_dims: &[usize],
133        update_strides: &[isize],
134        update_window_dims: &[usize],
135        is_update_window_dim: &[bool],
136        window_shape_updates: &[usize],
137        window_dims: &[usize],
138        dest_strides: &[isize],
139    ) -> Result<Self> {
140        let (
141            batch_source_strides,
142            batch_dest_strides,
143            window_source_strides,
144            window_dest_strides,
145            vector_stride,
146        ) = scatter_replay_strides(
147            index_dims,
148            index_strides,
149            index_vector_dim,
150            update_dims,
151            update_strides,
152            update_window_dims,
153            is_update_window_dim,
154            window_dims,
155            dest_strides,
156        );
157        let batch = WindowReplay::compile(batch_shape, &batch_source_strides, &batch_dest_strides)?;
158        let window = WindowReplay::compile(
159            window_shape_updates,
160            &window_source_strides,
161            &window_dest_strides,
162        )?;
163        let mut index_component_offsets = AxisVec::with_capacity(index_component_count);
164        for component in 0..index_component_count {
165            index_component_offsets.push(checked_offset_add(0, vector_stride, component)?);
166        }
167        Ok(Self {
168            batch,
169            window,
170            index_component_offsets,
171        })
172    }
173
174    #[cfg(feature = "parallel")]
175    fn validate(
176        batch_shape: &[usize],
177        index_dims: &[usize],
178        index_strides: &[isize],
179        index_vector_dim: usize,
180        index_component_count: usize,
181        update_dims: &[usize],
182        update_strides: &[isize],
183        update_window_dims: &[usize],
184        is_update_window_dim: &[bool],
185        window_shape_updates: &[usize],
186        window_dims: &[usize],
187        dest_strides: &[isize],
188    ) -> Result<()> {
189        let (
190            batch_source_strides,
191            batch_dest_strides,
192            window_source_strides,
193            window_dest_strides,
194            vector_stride,
195        ) = scatter_replay_strides(
196            index_dims,
197            index_strides,
198            index_vector_dim,
199            update_dims,
200            update_strides,
201            update_window_dims,
202            is_update_window_dim,
203            window_dims,
204            dest_strides,
205        );
206        WindowReplay::compile(batch_shape, &batch_source_strides, &batch_dest_strides)?;
207        WindowReplay::compile(
208            window_shape_updates,
209            &window_source_strides,
210            &window_dest_strides,
211        )?;
212        for component in 0..index_component_count {
213            checked_offset_add(0, vector_stride, component)?;
214        }
215        Ok(())
216    }
217}
218
219fn scatter_replay_strides(
220    index_dims: &[usize],
221    index_strides: &[isize],
222    index_vector_dim: usize,
223    update_dims: &[usize],
224    update_strides: &[isize],
225    update_window_dims: &[usize],
226    is_update_window_dim: &[bool],
227    window_dims: &[usize],
228    dest_strides: &[isize],
229) -> (
230    AxisVec<isize>,
231    AxisVec<isize>,
232    AxisVec<isize>,
233    AxisVec<isize>,
234    isize,
235) {
236    let batch_source_strides = index_dims
237        .iter()
238        .zip(index_strides.iter())
239        .enumerate()
240        .filter_map(|(axis, (_, &stride))| (axis != index_vector_dim).then_some(stride))
241        .collect();
242    let batch_dest_strides = update_dims
243        .iter()
244        .zip(update_strides.iter())
245        .enumerate()
246        .filter_map(|(axis, (_, &stride))| (!is_update_window_dim[axis]).then_some(stride))
247        .collect();
248    let window_source_strides = update_window_dims
249        .iter()
250        .map(|&axis| update_strides[axis])
251        .collect();
252    let window_dest_strides = window_dims.iter().map(|&axis| dest_strides[axis]).collect();
253    let vector_stride = if index_vector_dim < index_dims.len() {
254        index_strides[index_vector_dim]
255    } else {
256        0
257    };
258    (
259        batch_source_strides,
260        batch_dest_strides,
261        window_source_strides,
262        window_dest_strides,
263        vector_stride,
264    )
265}
266
267impl WindowReplay {
268    fn compile(shape: &[usize], source_strides: &[isize], dest_strides: &[isize]) -> Result<Self> {
269        validate_layout_span(shape, source_strides)?;
270        validate_layout_span(shape, dest_strides)?;
271        let mut fused_shape: AxisVec<usize> = AxisVec::with_capacity(shape.len());
272        let mut axes: AxisVec<WindowReplayAxis> = AxisVec::with_capacity(shape.len());
273        for (axis, &dim) in shape.iter().enumerate() {
274            let source_step = source_strides[axis];
275            let dest_step = dest_strides[axis];
276            if let Some(previous_axis) = fused_shape.len().checked_sub(1) {
277                let previous_extent = fused_shape[previous_axis];
278                let previous_extent =
279                    isize::try_from(previous_extent).map_err(|_| StridedError::OffsetOverflow)?;
280                let expected_source = axes[previous_axis]
281                    .source_step
282                    .checked_mul(previous_extent)
283                    .ok_or(StridedError::OffsetOverflow)?;
284                let expected_dest = axes[previous_axis]
285                    .dest_step
286                    .checked_mul(previous_extent)
287                    .ok_or(StridedError::OffsetOverflow)?;
288                if source_step == expected_source && dest_step == expected_dest {
289                    let fused_extent = fused_shape[previous_axis]
290                        .checked_mul(dim)
291                        .ok_or(StridedError::OffsetOverflow)?;
292                    fused_shape[previous_axis] = fused_extent;
293                    axes[previous_axis].source_reset =
294                        checked_replay_reset(fused_extent, axes[previous_axis].source_step)?;
295                    axes[previous_axis].dest_reset =
296                        checked_replay_reset(fused_extent, axes[previous_axis].dest_step)?;
297                    continue;
298                }
299            }
300            fused_shape.push(dim);
301            axes.push(WindowReplayAxis {
302                source_step,
303                source_reset: checked_replay_reset(dim, source_step)?,
304                dest_step,
305                dest_reset: checked_replay_reset(dim, dest_step)?,
306            });
307        }
308        Ok(Self {
309            shape: fused_shape,
310            axes,
311        })
312    }
313
314    fn decode(
315        &self,
316        mut linear: usize,
317        source_base: isize,
318        dest_base: isize,
319    ) -> Result<WindowReplayState> {
320        let mut coords = CoordScratch::new(self.shape.len());
321        let mut source_offset = source_base;
322        let mut dest_offset = dest_base;
323        for (axis, (&dim, coord)) in self
324            .shape
325            .iter()
326            .zip(coords.as_mut_slice().iter_mut())
327            .enumerate()
328        {
329            *coord = linear % dim;
330            linear /= dim;
331            let replay_axis = self.axes[axis];
332            source_offset = checked_offset_add(source_offset, replay_axis.source_step, *coord)?;
333            dest_offset = checked_offset_add(dest_offset, replay_axis.dest_step, *coord)?;
334        }
335        Ok(WindowReplayState {
336            coords,
337            source_offset,
338            dest_offset,
339        })
340    }
341
342    #[inline]
343    fn advance(&self, state: &mut WindowReplayState) {
344        for ((coord, &dim), replay_axis) in state
345            .coords
346            .as_mut_slice()
347            .iter_mut()
348            .zip(self.shape.iter())
349            .zip(self.axes.iter())
350        {
351            let next = *coord + 1;
352            if next < dim {
353                *coord = next;
354                state.source_offset += replay_axis.source_step;
355                state.dest_offset += replay_axis.dest_step;
356                return;
357            }
358            *coord = 0;
359            state.source_offset += replay_axis.source_reset;
360            state.dest_offset += replay_axis.dest_reset;
361        }
362    }
363}
364
365/// Scatter configuration shared by generic and erased replay.
366///
367/// `ScatterPlan` implements tenferro's current additive scatter semantics:
368/// every update value is added to the selected output slot, so overlapping
369/// windows accumulate in deterministic column-major replay order.
370#[derive(Clone, Debug, Eq, PartialEq)]
371pub struct ScatterSpec {
372    pub update_window_dims: Vec<usize>,
373    pub inserted_window_dims: Vec<usize>,
374    pub scatter_dims_to_operand_dims: Vec<usize>,
375    pub index_vector_dim: usize,
376}
377
378/// A compiled fixed-window dynamic-slice traversal.
379#[derive(Clone, Debug)]
380pub struct DynamicSlicePlan {
381    operand_dims: AxisVec<usize>,
382    operand_strides: AxisVec<isize>,
383    start_dims: AxisVec<usize>,
384    start_strides: AxisVec<isize>,
385    dest_dims: AxisVec<usize>,
386    dest_strides: AxisVec<isize>,
387    slice_sizes: AxisVec<usize>,
388    total: usize,
389    /// Fused window traversal (dest strides, operand strides) replayed at the
390    /// runtime source base; `None` when its fused rank exceeds
391    /// [`RAW_FUSED_RANK_LIMIT`](crate::RAW_FUSED_RANK_LIMIT), in which case
392    /// `replay`/`execute_parallel` walk the window instead.
393    window: Option<FusedPairLayout>,
394    #[cfg(not(feature = "parallel"))]
395    replay: WindowReplay,
396}
397
398/// A compiled dynamic-update-slice traversal.
399///
400/// Execution first copies `operand` into `dest`, then overwrites the clamped
401/// update window. The plan performs no allocation for ranks at most
402/// [`RAW_FUSED_RANK_LIMIT`].
403#[derive(Clone, Debug)]
404pub struct DynamicUpdateSlicePlan {
405    operand_dims: AxisVec<usize>,
406    operand_strides: AxisVec<isize>,
407    start_dims: AxisVec<usize>,
408    start_strides: AxisVec<isize>,
409    update_dims: AxisVec<usize>,
410    update_strides: AxisVec<isize>,
411    dest_dims: AxisVec<usize>,
412    dest_strides: AxisVec<isize>,
413    total: usize,
414    copy_plan: CopyPlan,
415    /// Fused update window traversal (dest strides, update strides) replayed
416    /// at the runtime destination base; `None` when its fused rank exceeds
417    /// [`RAW_FUSED_RANK_LIMIT`](crate::RAW_FUSED_RANK_LIMIT), in which case
418    /// `replay`/`execute_update_parallel` walk the window instead.
419    window: Option<FusedPairLayout>,
420    #[cfg(not(feature = "parallel"))]
421    replay: WindowReplay,
422}
423
424/// A compiled additive scatter traversal.
425///
426/// Execution first copies `operand` into `dest`, then applies additive updates
427/// in deterministic column-major order. Boolean values are intentionally not
428/// supported because additive scatter has no bool semantics.
429#[derive(Clone, Debug)]
430pub struct ScatterPlan {
431    operand_dims: AxisVec<usize>,
432    operand_strides: AxisVec<isize>,
433    index_dims: AxisVec<usize>,
434    index_strides: AxisVec<isize>,
435    update_dims: AxisVec<usize>,
436    update_strides: AxisVec<isize>,
437    dest_dims: AxisVec<usize>,
438    dest_strides: AxisVec<isize>,
439    spec: ScatterSpec,
440    #[cfg(feature = "parallel")]
441    batch_shape: AxisVec<usize>,
442    #[cfg(feature = "parallel")]
443    window_dims: AxisVec<usize>,
444    window_shape: AxisVec<usize>,
445    #[cfg(feature = "parallel")]
446    window_shape_updates: AxisVec<usize>,
447    #[cfg(feature = "parallel")]
448    is_update_window_dim: AxisVec<bool>,
449    batch_elems: usize,
450    window_elems: usize,
451    copy_plan: CopyPlan,
452    #[cfg(not(feature = "parallel"))]
453    replay: ScatterReplay,
454}
455
456impl GatherPlan {
457    /// Compile a gather plan for fixed operand, index, and destination layouts.
458    pub fn compile(
459        operand_dims: &[usize],
460        operand_strides: &[isize],
461        index_dims: &[usize],
462        index_strides: &[isize],
463        dest_dims: &[usize],
464        dest_strides: &[isize],
465        spec: GatherSpec,
466    ) -> Result<Self> {
467        if operand_dims.len() != operand_strides.len()
468            || index_dims.len() != index_strides.len()
469            || dest_dims.len() != dest_strides.len()
470        {
471            return Err(StridedError::StrideLengthMismatch);
472        }
473        checked_total_len(operand_dims)?;
474        checked_total_len(index_dims)?;
475        let total = checked_total_len(dest_dims)?;
476        validate_layout_span(operand_dims, operand_strides)?;
477        validate_layout_span(index_dims, index_strides)?;
478        validate_layout_span(dest_dims, dest_strides)?;
479        if !crate::layout_check::is_injective_layout(dest_dims, dest_strides) {
480            return Err(StridedError::NonInjectiveOutputLayout);
481        }
482
483        let operand_rank = operand_dims.len();
484        if spec.slice_sizes.len() != operand_rank {
485            return Err(StridedError::RankMismatch(
486                spec.slice_sizes.len(),
487                operand_rank,
488            ));
489        }
490        validate_unique_axes(&spec.collapsed_slice_dims, operand_rank)?;
491        validate_unique_axes(&spec.start_index_map, operand_rank)?;
492        if spec.index_vector_dim > index_dims.len() {
493            return Err(StridedError::InvalidAxis {
494                axis: spec.index_vector_dim,
495                rank: index_dims.len() + 1,
496            });
497        }
498
499        for (axis, (&window, &dim)) in spec.slice_sizes.iter().zip(operand_dims.iter()).enumerate()
500        {
501            if window > dim {
502                return Err(StridedError::InvalidAxis {
503                    axis,
504                    rank: operand_rank,
505                });
506            }
507        }
508        for &axis in &spec.collapsed_slice_dims {
509            if spec.slice_sizes[axis] != 1 {
510                return Err(StridedError::InvalidAxis {
511                    axis,
512                    rank: operand_rank,
513                });
514            }
515        }
516
517        let index_vector_size = if spec.index_vector_dim == index_dims.len() {
518            1
519        } else {
520            index_dims[spec.index_vector_dim]
521        };
522        if index_vector_size != spec.start_index_map.len() {
523            return Err(StridedError::RankMismatch(
524                index_vector_size,
525                spec.start_index_map.len(),
526            ));
527        }
528
529        let window_dims = operand_window_dims(operand_rank, &spec.collapsed_slice_dims);
530        if spec.offset_dims.len() != window_dims.len() {
531            return Err(StridedError::RankMismatch(
532                spec.offset_dims.len(),
533                window_dims.len(),
534            ));
535        }
536
537        let batch_shape = index_batch_shape(index_dims, spec.index_vector_dim);
538        let out_rank = batch_shape.len() + spec.offset_dims.len();
539        validate_unique_axes(&spec.offset_dims, out_rank)?;
540
541        let mut out_axis_to_operand_dim: AxisVec<Option<usize>> =
542            (0..out_rank).map(|_| None).collect();
543        for (offset_axis, &out_axis) in spec.offset_dims.iter().enumerate() {
544            out_axis_to_operand_dim[out_axis] = Some(window_dims[offset_axis]);
545        }
546
547        let mut expected_dest_dims: AxisVec<usize> = AxisVec::with_capacity(out_rank);
548        let mut batch_axis = 0usize;
549        for &operand_dim in &out_axis_to_operand_dim {
550            match operand_dim {
551                Some(axis) => expected_dest_dims.push(spec.slice_sizes[axis]),
552                None => {
553                    expected_dest_dims.push(batch_shape[batch_axis]);
554                    batch_axis += 1;
555                }
556            }
557        }
558        if dest_dims != &expected_dest_dims[..] {
559            return Err(StridedError::ShapeMismatch(
560                dest_dims.to_vec(),
561                expected_dest_dims.to_vec(),
562            ));
563        }
564
565        let batch_index_strides: AxisVec<isize> = index_dims
566            .iter()
567            .zip(index_strides.iter())
568            .enumerate()
569            .filter_map(|(axis, (_, &stride))| (axis != spec.index_vector_dim).then_some(stride))
570            .collect();
571        let mut batch_axis = 0usize;
572        let mut replay_axes = AxisVec::with_capacity(out_rank);
573        for (out_axis, &operand_dim) in out_axis_to_operand_dim.iter().enumerate() {
574            let (window_step, index_batch_step) = match operand_dim {
575                Some(axis) => (operand_strides[axis], 0),
576                None => {
577                    let step = batch_index_strides[batch_axis];
578                    batch_axis += 1;
579                    (0, step)
580                }
581            };
582            replay_axes.push(GatherReplayAxis {
583                dest_step: dest_strides[out_axis],
584                dest_reset: checked_replay_reset(dest_dims[out_axis], dest_strides[out_axis])?,
585                window_step,
586                window_reset: checked_replay_reset(dest_dims[out_axis], window_step)?,
587                index_batch_step,
588                index_batch_reset: checked_replay_reset(dest_dims[out_axis], index_batch_step)?,
589            });
590        }
591
592        let vector_stride = if spec.index_vector_dim < index_dims.len() {
593            index_strides[spec.index_vector_dim]
594        } else {
595            0
596        };
597        let mut index_component_offsets = AxisVec::with_capacity(spec.start_index_map.len());
598        for component in 0..spec.start_index_map.len() {
599            index_component_offsets.push(checked_offset_add(0, vector_stride, component)?);
600        }
601
602        Ok(Self {
603            operand_dims: operand_dims.into(),
604            operand_strides: operand_strides.into(),
605            index_dims: index_dims.into(),
606            index_strides: index_strides.into(),
607            dest_dims: dest_dims.into(),
608            dest_strides: dest_strides.into(),
609            spec: spec.clone(),
610            replay: GatherReplay {
611                axes: replay_axes,
612                index_component_offsets,
613                index_operand_strides: spec
614                    .start_index_map
615                    .iter()
616                    .map(|&axis| operand_strides[axis])
617                    .collect(),
618            },
619            total,
620        })
621    }
622
623    #[inline]
624    pub fn spec(&self) -> &GatherSpec {
625        &self.spec
626    }
627
628    #[inline]
629    pub fn dest_dims(&self) -> &[usize] {
630        &self.dest_dims
631    }
632
633    /// Execute the prepared gather traversal.
634    pub fn execute<T, I>(
635        &self,
636        dest: &mut RawStridedMut<'_, T>,
637        operand: &RawStridedRef<'_, T>,
638        start_indices: &RawStridedRef<'_, I>,
639    ) -> Result<()>
640    where
641        T: Copy + MaybeSendSync,
642        I: GatherIndex,
643    {
644        self.execute_with_writer(dest, operand, start_indices)
645    }
646
647    /// Execute the prepared gather into a destination whose reachable slots
648    /// may be uninitialized. Every logical destination slot is written.
649    pub(crate) fn execute_uninit<T, I>(
650        &self,
651        dest: &mut RawStridedMut<'_, MaybeUninit<T>>,
652        operand: &RawStridedRef<'_, T>,
653        start_indices: &RawStridedRef<'_, I>,
654    ) -> Result<()>
655    where
656        T: Copy + MaybeSendSync,
657        I: GatherIndex,
658    {
659        self.execute_with_writer(dest, operand, start_indices)
660    }
661
662    fn execute_with_writer<T, I, W>(
663        &self,
664        dest: &mut W,
665        operand: &RawStridedRef<'_, T>,
666        start_indices: &RawStridedRef<'_, I>,
667    ) -> Result<()>
668    where
669        T: Copy + MaybeSendSync,
670        I: GatherIndex,
671        W: OverwriteWriter<T>,
672    {
673        self.check_call(dest, operand, start_indices)?;
674        if self.total == 0 {
675            return Ok(());
676        }
677        if self.uses_rank_one_scalar_take_path() {
678            #[cfg(feature = "parallel")]
679            {
680                let nthreads = crate::threading::parallel_threads_for_len(self.total);
681                if nthreads > 1 {
682                    return self.execute_rank_one_scalar_take_parallel(
683                        dest,
684                        operand,
685                        start_indices,
686                        nthreads,
687                    );
688                }
689            }
690            return self.execute_rank_one_scalar_take(dest, operand, start_indices);
691        }
692        #[cfg(feature = "parallel")]
693        {
694            let nthreads = crate::threading::parallel_threads_for_len(self.total);
695            if nthreads > 1 {
696                return self.execute_parallel(dest, operand, start_indices, nthreads);
697            }
698        }
699
700        let mut state =
701            self.decode_replay_state(0, dest.offset(), operand.offset(), start_indices.offset())?;
702        let operand_data = operand.data();
703        let index_data = start_indices.data();
704
705        // INVARIANT: the checked decode starts at a valid logical output. Each
706        // replay axis has checked step/reset deltas, so this state remains the
707        // corresponding destination, window, and batch-index offsets.
708        for _ in 0..self.total {
709            // INVARIANT: window_offset plus every clamped mapped contribution
710            // is an operand offset for coordinates inside operand_dims; the
711            // validated operand span and RawStridedRef reachability therefore
712            // keep the fresh source offset inside the allocation.
713            let mut source_offset = state.window_offset;
714            for ((&index_component_offset, &operand_stride), &operand_dim) in self
715                .replay
716                .index_component_offsets
717                .iter()
718                .zip(self.replay.index_operand_strides.iter())
719                .zip(self.spec.start_index_map.iter())
720            {
721                let index_offset = state.index_batch_offset + index_component_offset;
722                // SAFETY: checked index layout replay and RawStridedRef
723                // reachability prove this component read is in bounds.
724                let start = unsafe { *index_data.as_ptr().offset(index_offset) }.to_i64();
725                let clamped = self.clamp_window_start(start, operand_dim);
726                source_offset += operand_stride * clamped as isize;
727            }
728
729            // SAFETY: checked replay metadata and the validated writer prove
730            // both the source and the distinct destination offset are valid.
731            let value = unsafe { *operand_data.as_ptr().offset(source_offset) };
732            unsafe { dest.write_at(state.dest_offset, value) };
733            self.advance_replay_state(&mut state);
734        }
735        Ok(())
736    }
737
738    fn uses_rank_one_scalar_take_path(&self) -> bool {
739        self.operand_dims.len() == 1
740            && self.index_dims.len() == 1
741            && self.dest_dims.len() == 1
742            && self.operand_strides[0] == 1
743            && self.index_strides[0] == 1
744            && self.dest_strides[0] == 1
745            && self.spec.offset_dims.is_empty()
746            && self.spec.collapsed_slice_dims.as_slice() == [0]
747            && self.spec.start_index_map.as_slice() == [0]
748            && self.spec.index_vector_dim == 1
749            && self.spec.slice_sizes.as_slice() == [1]
750    }
751
752    fn execute_rank_one_scalar_take<T, I, W>(
753        &self,
754        dest: &mut W,
755        operand: &RawStridedRef<'_, T>,
756        start_indices: &RawStridedRef<'_, I>,
757    ) -> Result<()>
758    where
759        T: Copy,
760        I: GatherIndex,
761        W: OverwriteWriter<T>,
762    {
763        let mut dest_offset = dest.offset();
764        let mut index_offset = start_indices.offset();
765        let operand_offset = operand.offset();
766        let operand_data = operand.data();
767        let index_data = start_indices.data();
768
769        // INVARIANT: compile and check_call validated compact rank-one layouts,
770        // so incrementing these offsets once per logical element stays within
771        // the validated allocations and cannot overflow isize.
772        for _ in 0..self.total {
773            // SAFETY: the invariant above proves all three offsets are in bounds.
774            unsafe {
775                let start = (*index_data.as_ptr().offset(index_offset)).to_i64();
776                let source_offset = operand_offset + self.clamp_window_start(start, 0) as isize;
777                dest.write_at(dest_offset, *operand_data.as_ptr().offset(source_offset));
778            }
779            dest_offset += 1;
780            index_offset += 1;
781        }
782        Ok(())
783    }
784
785    #[cfg(feature = "parallel")]
786    fn execute_rank_one_scalar_take_parallel<T, I, W>(
787        &self,
788        dest: &mut W,
789        operand: &RawStridedRef<'_, T>,
790        start_indices: &RawStridedRef<'_, I>,
791        nthreads: usize,
792    ) -> Result<()>
793    where
794        T: Copy + MaybeSendSync,
795        I: GatherIndex,
796        W: OverwriteWriter<T>,
797    {
798        let dest_offset = dest.offset();
799        let operand_offset = operand.offset();
800        let index_offset = start_indices.offset();
801        // SAFETY: check_call validated the writer's complete destination layout.
802        let dest_ptr = crate::threading::SendPtr(unsafe { dest.data_ptr() });
803        let operand_ptr = crate::threading::SendPtr(operand.data().as_ptr() as *mut T);
804        let index_ptr = crate::threading::SendPtr(start_indices.data().as_ptr() as *mut I);
805
806        crate::threading::parallel_map_reduce(
807            0..self.total,
808            nthreads,
809            &|range| {
810                let dest_ptr = dest_ptr.as_ptr();
811                let operand_ptr = operand_ptr.as_const();
812                let index_ptr = index_ptr.as_const();
813                // INVARIANT: the validated compact rank-one layouts map each
814                // output position to one distinct destination offset.
815                for position in range {
816                    let position = position as isize;
817                    // SAFETY: the invariant above proves the source and distinct
818                    // destination offsets are in their validated allocations.
819                    unsafe {
820                        let start = (*index_ptr.offset(index_offset + position)).to_i64();
821                        let source_offset =
822                            operand_offset + self.clamp_window_start(start, 0) as isize;
823                        dest_ptr
824                            .offset(dest_offset + position)
825                            .write(operand_ptr.offset(source_offset).read());
826                    }
827                }
828                Ok(())
829            },
830            &|left, right| left.and(right),
831        )
832    }
833
834    #[cfg(feature = "parallel")]
835    fn execute_parallel<T, I, W>(
836        &self,
837        dest: &mut W,
838        operand: &RawStridedRef<'_, T>,
839        start_indices: &RawStridedRef<'_, I>,
840        nthreads: usize,
841    ) -> Result<()>
842    where
843        T: Copy + MaybeSendSync,
844        I: GatherIndex,
845        W: OverwriteWriter<T>,
846    {
847        let dest_offset_base = dest.offset();
848        let operand_offset_base = operand.offset();
849        let index_offset_base = start_indices.offset();
850        // SAFETY: the validated writer owns the destination allocation.
851        let dest_ptr = crate::threading::SendPtr(unsafe { dest.data_ptr() });
852        let operand_ptr = crate::threading::SendPtr(operand.data().as_ptr() as *mut T);
853        let index_ptr = crate::threading::SendPtr(start_indices.data().as_ptr() as *mut I);
854
855        crate::threading::parallel_map_reduce(
856            0..self.total,
857            nthreads,
858            &|range| {
859                let mut state = self.decode_replay_state(
860                    range.start,
861                    dest_offset_base,
862                    operand_offset_base,
863                    index_offset_base,
864                )?;
865                let dest_ptr = dest_ptr.as_ptr();
866                let operand_ptr = operand_ptr.as_const();
867                let index_ptr = index_ptr.as_const();
868
869                // INVARIANT: this worker owns a disjoint logical range. Checked
870                // range-start decode plus checked replay deltas keeps every
871                // state at its corresponding output, and injectivity makes all
872                // writes across workers distinct.
873                for _ in range {
874                    // INVARIANT: source_offset is freshly based on the current
875                    // window offset; clamped mapped coordinates remain inside
876                    // the validated operand span and are never accumulated.
877                    let mut source_offset = state.window_offset;
878                    for ((&index_component_offset, &operand_stride), &operand_dim) in self
879                        .replay
880                        .index_component_offsets
881                        .iter()
882                        .zip(self.replay.index_operand_strides.iter())
883                        .zip(self.spec.start_index_map.iter())
884                    {
885                        let index_offset = state.index_batch_offset + index_component_offset;
886                        // SAFETY: the checked index span and this worker's
887                        // validated range-start state prove this read is valid.
888                        let start = unsafe { (*index_ptr.offset(index_offset)).to_i64() };
889                        let clamped = self.clamp_window_start(start, operand_dim);
890                        source_offset += operand_stride * clamped as isize;
891                    }
892
893                    // SAFETY: the source proof above and distinct destination
894                    // invariant justify these raw accesses.
895                    unsafe {
896                        dest_ptr
897                            .offset(state.dest_offset)
898                            .write(operand_ptr.offset(source_offset).read());
899                    }
900                    self.advance_replay_state(&mut state);
901                }
902                Ok(())
903            },
904            &|left, right| left.and(right),
905        )
906    }
907
908    fn check_call<T, I, W>(
909        &self,
910        dest: &W,
911        operand: &RawStridedRef<'_, T>,
912        start_indices: &RawStridedRef<'_, I>,
913    ) -> Result<()>
914    where
915        W: OverwriteWriter<T>,
916    {
917        if dest.dims() != &self.dest_dims[..]
918            || dest.strides() != &self.dest_strides[..]
919            || operand.dims() != &self.operand_dims[..]
920            || operand.strides() != &self.operand_strides[..]
921            || start_indices.dims() != &self.index_dims[..]
922            || start_indices.strides() != &self.index_strides[..]
923        {
924            return Err(StridedError::PlanLayoutMismatch);
925        }
926        Ok(())
927    }
928
929    fn decode_replay_state(
930        &self,
931        mut linear: usize,
932        dest_offset_base: isize,
933        operand_offset_base: isize,
934        index_offset_base: isize,
935    ) -> Result<GatherReplayState> {
936        let mut coords = AxisVec::with_capacity(self.dest_dims.len());
937        let mut dest_offset = dest_offset_base;
938        let mut window_offset = operand_offset_base;
939        let mut index_batch_offset = index_offset_base;
940        for (axis, &dim) in self.dest_dims.iter().enumerate() {
941            let coord = linear % dim;
942            linear /= dim;
943            let replay_axis = self.replay.axes[axis];
944            coords.push(coord);
945            dest_offset = checked_offset_add(dest_offset, replay_axis.dest_step, coord)?;
946            window_offset = checked_offset_add(window_offset, replay_axis.window_step, coord)?;
947            index_batch_offset =
948                checked_offset_add(index_batch_offset, replay_axis.index_batch_step, coord)?;
949        }
950        Ok(GatherReplayState {
951            coords,
952            dest_offset,
953            window_offset,
954            index_batch_offset,
955        })
956    }
957
958    #[inline]
959    fn advance_replay_state(&self, state: &mut GatherReplayState) {
960        for (axis, replay_axis) in self.replay.axes.iter().enumerate() {
961            let coord = state.coords[axis];
962            if coord + 1 < self.dest_dims[axis] {
963                state.coords[axis] = coord + 1;
964                state.dest_offset += replay_axis.dest_step;
965                state.window_offset += replay_axis.window_step;
966                state.index_batch_offset += replay_axis.index_batch_step;
967                return;
968            }
969            state.coords[axis] = 0;
970            state.dest_offset += replay_axis.dest_reset;
971            state.window_offset += replay_axis.window_reset;
972            state.index_batch_offset += replay_axis.index_batch_reset;
973        }
974    }
975
976    #[inline]
977    fn clamp_window_start(&self, start: i64, operand_dim: usize) -> usize {
978        let dim_size = self.operand_dims[operand_dim];
979        let window_size = self.spec.slice_sizes[operand_dim];
980        let max_start = dim_size.saturating_sub(window_size) as i64;
981        start.clamp(0, max_start) as usize
982    }
983}
984
985impl DynamicSlicePlan {
986    /// Compile a fixed-window dynamic-slice traversal.
987    ///
988    /// `start_*` describes a rank-1 index vector whose length equals the
989    /// operand rank. `dest_dims` must equal `slice_sizes`.
990    pub fn compile(
991        operand_dims: &[usize],
992        operand_strides: &[isize],
993        start_dims: &[usize],
994        start_strides: &[isize],
995        dest_dims: &[usize],
996        dest_strides: &[isize],
997        slice_sizes: &[usize],
998    ) -> Result<Self> {
999        if operand_dims.len() != operand_strides.len()
1000            || start_dims.len() != start_strides.len()
1001            || dest_dims.len() != dest_strides.len()
1002        {
1003            return Err(StridedError::StrideLengthMismatch);
1004        }
1005        if slice_sizes.len() != operand_dims.len() {
1006            return Err(StridedError::RankMismatch(
1007                slice_sizes.len(),
1008                operand_dims.len(),
1009            ));
1010        }
1011        validate_start_vector(start_dims, operand_dims.len())?;
1012        checked_total_len(operand_dims)?;
1013        checked_total_len(start_dims)?;
1014        let total = checked_total_len(dest_dims)?;
1015        validate_layout_span(operand_dims, operand_strides)?;
1016        validate_layout_span(start_dims, start_strides)?;
1017        validate_layout_span(dest_dims, dest_strides)?;
1018        if dest_dims != slice_sizes {
1019            return Err(StridedError::ShapeMismatch(
1020                dest_dims.to_vec(),
1021                slice_sizes.to_vec(),
1022            ));
1023        }
1024        if !crate::layout_check::is_injective_layout(dest_dims, dest_strides) {
1025            return Err(StridedError::NonInjectiveOutputLayout);
1026        }
1027        validate_window_sizes(operand_dims, slice_sizes)?;
1028        #[cfg(feature = "parallel")]
1029        WindowReplay::compile(slice_sizes, operand_strides, dest_strides)?;
1030        #[cfg(not(feature = "parallel"))]
1031        let replay = WindowReplay::compile(slice_sizes, operand_strides, dest_strides)?;
1032
1033        Ok(Self {
1034            operand_dims: operand_dims.into(),
1035            operand_strides: operand_strides.into(),
1036            start_dims: start_dims.into(),
1037            start_strides: start_strides.into(),
1038            dest_dims: dest_dims.into(),
1039            dest_strides: dest_strides.into(),
1040            slice_sizes: slice_sizes.into(),
1041            total,
1042            window: fuse_pair_layout(slice_sizes, dest_strides, operand_strides),
1043            #[cfg(not(feature = "parallel"))]
1044            replay,
1045        })
1046    }
1047
1048    /// Execute the prepared dynamic-slice traversal.
1049    pub fn execute<T, I>(
1050        &self,
1051        dest: &mut RawStridedMut<'_, T>,
1052        operand: &RawStridedRef<'_, T>,
1053        starts: &RawStridedRef<'_, I>,
1054    ) -> Result<()>
1055    where
1056        T: Copy + MaybeSendSync,
1057        I: GatherIndex,
1058    {
1059        self.execute_with_writer(dest, operand, starts)
1060    }
1061
1062    pub(crate) fn execute_uninit<T, I>(
1063        &self,
1064        dest: &mut RawStridedMut<'_, MaybeUninit<T>>,
1065        operand: &RawStridedRef<'_, T>,
1066        starts: &RawStridedRef<'_, I>,
1067    ) -> Result<()>
1068    where
1069        T: Copy + MaybeSendSync,
1070        I: GatherIndex,
1071    {
1072        self.execute_with_writer(dest, operand, starts)
1073    }
1074
1075    fn execute_with_writer<T, I, W>(
1076        &self,
1077        dest: &mut W,
1078        operand: &RawStridedRef<'_, T>,
1079        starts: &RawStridedRef<'_, I>,
1080    ) -> Result<()>
1081    where
1082        T: Copy + MaybeSendSync,
1083        I: GatherIndex,
1084        W: OverwriteWriter<T>,
1085    {
1086        self.check_call(dest, operand, starts)?;
1087        if self.total == 0 {
1088            return Ok(());
1089        }
1090        if self.uses_rank_one_contiguous_path() {
1091            return self.execute_rank_one_contiguous(dest, operand, starts);
1092        }
1093        if let Some(window) = &self.window {
1094            return self.execute_fused_window(dest, operand, starts, window);
1095        }
1096        #[cfg(feature = "parallel")]
1097        let replay =
1098            WindowReplay::compile(&self.slice_sizes, &self.operand_strides, &self.dest_strides)?;
1099        #[cfg(not(feature = "parallel"))]
1100        let replay = &self.replay;
1101        #[cfg(feature = "parallel")]
1102        {
1103            let nthreads = crate::threading::parallel_threads_for_len(self.total);
1104            if nthreads > 1 {
1105                return self.execute_parallel(dest, operand, starts, nthreads, &replay);
1106            }
1107        }
1108
1109        let mut starts_storage = CoordScratch::new(self.operand_dims.len());
1110        let clamped_starts = starts_storage.as_mut_slice();
1111        read_clamped_starts(
1112            starts,
1113            &self.operand_dims,
1114            &self.slice_sizes,
1115            clamped_starts,
1116        )?;
1117        let source_base =
1118            checked_strided_offset(operand.offset(), &self.operand_strides, clamped_starts)?;
1119        let mut state = replay.decode(0, source_base, dest.offset())?;
1120        let operand_data = operand.data();
1121
1122        // INVARIANT: compile validated the full operand/destination spans and the
1123        // replay window spans; checked bases plus checked reset deltas therefore
1124        // keep every current offset reachable without per-element offset scans.
1125        for _ in 0..self.total {
1126            // SAFETY: the invariant above proves both current offsets are valid.
1127            let value = unsafe { *operand_data.as_ptr().offset(state.source_offset) };
1128            // SAFETY: the invariant above proves this logical destination offset
1129            // is in-bounds, and the output layout is injective.
1130            unsafe { dest.write_at(state.dest_offset, value) };
1131            replay.advance(&mut state);
1132        }
1133        Ok(())
1134    }
1135
1136    /// Replay the compiled fused window from the clamped runtime source base.
1137    ///
1138    /// The window is a CopyPlan-style fused layout, so full inner runs are
1139    /// replayed with a pointer loop (slice copies for unit strides) and the
1140    /// work is split across workers above the repository threshold.
1141    fn execute_fused_window<T, I, W>(
1142        &self,
1143        dest: &mut W,
1144        operand: &RawStridedRef<'_, T>,
1145        starts: &RawStridedRef<'_, I>,
1146        window: &FusedPairLayout,
1147    ) -> Result<()>
1148    where
1149        T: Copy + MaybeSendSync,
1150        I: GatherIndex,
1151        W: OverwriteWriter<T>,
1152    {
1153        let mut starts_storage = CoordScratch::new(self.operand_dims.len());
1154        let clamped_starts = starts_storage.as_mut_slice();
1155        read_clamped_starts(
1156            starts,
1157            &self.operand_dims,
1158            &self.slice_sizes,
1159            clamped_starts,
1160        )?;
1161        let source_base =
1162            checked_strided_offset(operand.offset(), &self.operand_strides, clamped_starts)?;
1163        let dest_base = dest.offset();
1164        // SAFETY: the validated writer owns the destination allocation.
1165        let dest_ptr = unsafe { dest.data_ptr() }.cast::<MaybeUninit<T>>();
1166        // SAFETY: `window` fuses `slice_sizes` with the checked destination
1167        // and operand strides. The clamped window `start + slice_sizes` lies
1168        // inside `operand_dims`, so every source offset from `source_base` is
1169        // an offset the validated operand view can reach; the destination has
1170        // exactly `slice_sizes` dims and its validated layout reaches every
1171        // destination offset. Compile proved the destination injective, and
1172        // the exclusive writer borrow does not overlap the shared operand.
1173        // Writing `MaybeUninit::new` is valid for both initialized and
1174        // uninitialized destination slots.
1175        unsafe {
1176            replay_fused_raw(
1177                dest_ptr,
1178                dest_base,
1179                operand.data().as_ptr(),
1180                source_base,
1181                window,
1182                |slot: &mut MaybeUninit<T>, value: T| *slot = MaybeUninit::new(value),
1183                |value: T| value,
1184            );
1185        }
1186        Ok(())
1187    }
1188
1189    #[inline]
1190    fn uses_rank_one_contiguous_path(&self) -> bool {
1191        self.operand_dims.len() == 1 && self.operand_strides[0] == 1 && self.dest_strides[0] == 1
1192    }
1193
1194    fn execute_rank_one_contiguous<T, I, W>(
1195        &self,
1196        dest: &mut W,
1197        operand: &RawStridedRef<'_, T>,
1198        starts: &RawStridedRef<'_, I>,
1199    ) -> Result<()>
1200    where
1201        T: Copy + MaybeSendSync,
1202        I: GatherIndex,
1203        W: OverwriteWriter<T>,
1204    {
1205        let mut clamped_starts = [0usize; 1];
1206        read_clamped_starts(
1207            starts,
1208            &self.operand_dims,
1209            &self.slice_sizes,
1210            &mut clamped_starts,
1211        )?;
1212        let source_start = checked_offset_add(operand.offset(), 1, clamped_starts[0])?;
1213        let source_start =
1214            usize::try_from(source_start).map_err(|_| StridedError::OffsetOverflow)?;
1215        let dest_start =
1216            usize::try_from(dest.offset()).map_err(|_| StridedError::OffsetOverflow)?;
1217        let source_end = source_start
1218            .checked_add(self.total)
1219            .ok_or(StridedError::OffsetOverflow)?;
1220        let source = operand
1221            .data()
1222            .get(source_start..source_end)
1223            .ok_or(StridedError::OffsetOverflow)?;
1224        // SAFETY: the validated writer owns the destination allocation.
1225        let dest_ptr = unsafe { dest.data_ptr() };
1226        // SAFETY: bounds were checked above and the writer owns the logical
1227        // destination storage.
1228        unsafe {
1229            // Chunked across workers above the repository threshold; each
1230            // worker writes a disjoint chunk of the checked destination run.
1231            crate::threading::copy_contiguous(
1232                source.as_ptr(),
1233                dest_ptr.add(dest_start),
1234                self.total,
1235            );
1236        }
1237        Ok(())
1238    }
1239
1240    #[cfg(feature = "parallel")]
1241    fn execute_parallel<T, I, W>(
1242        &self,
1243        dest: &mut W,
1244        operand: &RawStridedRef<'_, T>,
1245        starts: &RawStridedRef<'_, I>,
1246        nthreads: usize,
1247        replay: &WindowReplay,
1248    ) -> Result<()>
1249    where
1250        T: Copy + MaybeSendSync,
1251        I: GatherIndex,
1252        W: OverwriteWriter<T>,
1253    {
1254        let mut clamped_starts: AxisVec<usize> = (0..self.operand_dims.len()).map(|_| 0).collect();
1255        read_clamped_starts(
1256            starts,
1257            &self.operand_dims,
1258            &self.slice_sizes,
1259            &mut clamped_starts,
1260        )?;
1261        let source_base =
1262            checked_strided_offset(operand.offset(), &self.operand_strides, &clamped_starts)?;
1263        let dest_base = dest.offset();
1264        let operand_ptr = crate::threading::SendPtr(operand.data().as_ptr() as *mut T);
1265        // SAFETY: the validated writer owns the destination allocation.
1266        let dest_ptr = crate::threading::SendPtr(unsafe { dest.data_ptr() });
1267
1268        crate::threading::parallel_map_reduce(
1269            0..self.total,
1270            nthreads,
1271            &|range| {
1272                let mut state = replay.decode(range.start, source_base, dest_base)?;
1273                let operand_ptr = operand_ptr.as_const();
1274                let dest_ptr = dest_ptr.as_ptr();
1275
1276                // INVARIANT: each worker decodes one checked range start. Its
1277                // replay range is disjoint, and the injective output layout makes
1278                // all writes distinct across workers.
1279                for _ in range {
1280                    // SAFETY: checked replay bases/resets keep both offsets in
1281                    // their validated allocations.
1282                    unsafe {
1283                        dest_ptr
1284                            .offset(state.dest_offset)
1285                            .write(operand_ptr.offset(state.source_offset).read());
1286                    }
1287                    replay.advance(&mut state);
1288                }
1289                Ok(())
1290            },
1291            &|left, right| left.and(right),
1292        )
1293    }
1294
1295    fn check_call<T, I, W>(
1296        &self,
1297        dest: &W,
1298        operand: &RawStridedRef<'_, T>,
1299        starts: &RawStridedRef<'_, I>,
1300    ) -> Result<()>
1301    where
1302        W: OverwriteWriter<T>,
1303    {
1304        if dest.dims() != &self.dest_dims[..]
1305            || dest.strides() != &self.dest_strides[..]
1306            || operand.dims() != &self.operand_dims[..]
1307            || operand.strides() != &self.operand_strides[..]
1308            || starts.dims() != &self.start_dims[..]
1309            || starts.strides() != &self.start_strides[..]
1310        {
1311            return Err(StridedError::PlanLayoutMismatch);
1312        }
1313        Ok(())
1314    }
1315}
1316
1317impl DynamicUpdateSlicePlan {
1318    /// Compile a dynamic-update-slice traversal.
1319    ///
1320    /// `dest_dims` must match `operand_dims`; execution materializes
1321    /// `dest = operand` and then overwrites the clamped update window.
1322    #[allow(clippy::too_many_arguments)]
1323    pub fn compile(
1324        operand_dims: &[usize],
1325        operand_strides: &[isize],
1326        start_dims: &[usize],
1327        start_strides: &[isize],
1328        update_dims: &[usize],
1329        update_strides: &[isize],
1330        dest_dims: &[usize],
1331        dest_strides: &[isize],
1332    ) -> Result<Self> {
1333        if operand_dims.len() != operand_strides.len()
1334            || start_dims.len() != start_strides.len()
1335            || update_dims.len() != update_strides.len()
1336            || dest_dims.len() != dest_strides.len()
1337        {
1338            return Err(StridedError::StrideLengthMismatch);
1339        }
1340        if update_dims.len() != operand_dims.len() {
1341            return Err(StridedError::RankMismatch(
1342                update_dims.len(),
1343                operand_dims.len(),
1344            ));
1345        }
1346        validate_start_vector(start_dims, operand_dims.len())?;
1347        checked_total_len(operand_dims)?;
1348        checked_total_len(start_dims)?;
1349        let total = checked_total_len(update_dims)?;
1350        validate_layout_span(operand_dims, operand_strides)?;
1351        validate_layout_span(start_dims, start_strides)?;
1352        validate_layout_span(update_dims, update_strides)?;
1353        validate_layout_span(dest_dims, dest_strides)?;
1354        if dest_dims != operand_dims {
1355            return Err(StridedError::ShapeMismatch(
1356                dest_dims.to_vec(),
1357                operand_dims.to_vec(),
1358            ));
1359        }
1360        if !crate::layout_check::is_injective_layout(dest_dims, dest_strides) {
1361            return Err(StridedError::NonInjectiveOutputLayout);
1362        }
1363        validate_window_sizes(operand_dims, update_dims)?;
1364        let copy_plan = CopyPlan::compile(operand_dims, dest_strides, operand_strides)?;
1365        #[cfg(feature = "parallel")]
1366        WindowReplay::compile(update_dims, update_strides, dest_strides)?;
1367        #[cfg(not(feature = "parallel"))]
1368        let replay = WindowReplay::compile(update_dims, update_strides, dest_strides)?;
1369
1370        Ok(Self {
1371            operand_dims: operand_dims.into(),
1372            operand_strides: operand_strides.into(),
1373            start_dims: start_dims.into(),
1374            start_strides: start_strides.into(),
1375            update_dims: update_dims.into(),
1376            update_strides: update_strides.into(),
1377            dest_dims: dest_dims.into(),
1378            dest_strides: dest_strides.into(),
1379            total,
1380            copy_plan,
1381            window: fuse_pair_layout(update_dims, dest_strides, update_strides),
1382            #[cfg(not(feature = "parallel"))]
1383            replay,
1384        })
1385    }
1386
1387    /// Execute the prepared dynamic-update-slice traversal.
1388    pub fn execute<T, I>(
1389        &self,
1390        dest: &mut RawStridedMut<'_, T>,
1391        operand: &RawStridedRef<'_, T>,
1392        update: &RawStridedRef<'_, T>,
1393        starts: &RawStridedRef<'_, I>,
1394    ) -> Result<()>
1395    where
1396        T: Copy + MaybeSendSync,
1397        I: GatherIndex,
1398    {
1399        self.check_call(dest, operand, update, starts)?;
1400        self.copy_plan.execute(dest, operand)?;
1401        self.execute_update_with_writer(dest, update, starts)
1402    }
1403
1404    /// Execute dynamic update into a destination whose reachable slots may be
1405    /// uninitialized. The copy completes before any read-modify-write access.
1406    pub(crate) fn execute_uninit<'a, T, I>(
1407        &self,
1408        dest: &'a mut RawStridedMut<'a, MaybeUninit<T>>,
1409        operand: &RawStridedRef<'_, T>,
1410        update: &RawStridedRef<'_, T>,
1411        starts: &RawStridedRef<'_, I>,
1412    ) -> Result<()>
1413    where
1414        T: Copy + MaybeSendSync,
1415        I: GatherIndex,
1416    {
1417        self.check_call(dest, operand, update, starts)?;
1418        self.copy_plan
1419            .execute_uninit_then(dest, operand, |mut receipt| {
1420                self.execute_update_with_writer(&mut receipt, update, starts)
1421            })?
1422    }
1423
1424    fn execute_update_with_writer<T, I, W>(
1425        &self,
1426        dest: &mut W,
1427        update: &RawStridedRef<'_, T>,
1428        starts: &RawStridedRef<'_, I>,
1429    ) -> Result<()>
1430    where
1431        T: Copy + MaybeSendSync,
1432        I: GatherIndex,
1433        W: OverwriteWriter<T>,
1434    {
1435        if self.total == 0 {
1436            return Ok(());
1437        }
1438        if self.uses_rank_one_contiguous_path() {
1439            return self.execute_rank_one_contiguous(dest, update, starts);
1440        }
1441        if let Some(window) = &self.window {
1442            return self.execute_fused_window(dest, update, starts, window);
1443        }
1444        #[cfg(feature = "parallel")]
1445        let replay =
1446            WindowReplay::compile(&self.update_dims, &self.update_strides, &self.dest_strides)?;
1447        #[cfg(not(feature = "parallel"))]
1448        let replay = &self.replay;
1449        #[cfg(feature = "parallel")]
1450        {
1451            let nthreads = crate::threading::parallel_threads_for_len(self.total);
1452            if nthreads > 1 {
1453                return self.execute_update_parallel(dest, update, starts, nthreads, &replay);
1454            }
1455        }
1456
1457        let mut starts_storage = CoordScratch::new(self.operand_dims.len());
1458        let clamped_starts = starts_storage.as_mut_slice();
1459        read_clamped_starts(
1460            starts,
1461            &self.operand_dims,
1462            &self.update_dims,
1463            clamped_starts,
1464        )?;
1465        let source_base = update.offset();
1466        let dest_base = checked_strided_offset(dest.offset(), &self.dest_strides, &clamped_starts)?;
1467        let mut state = replay.decode(0, source_base, dest_base)?;
1468        let update_data = update.data();
1469
1470        // INVARIANT: the initial CopyPlan has completed before this replay;
1471        // compile validated both window spans and the checked bases/resets keep
1472        // every update read and destination write reachable.
1473        for _ in 0..self.total {
1474            // SAFETY: checked replay metadata proves the current update read.
1475            let value = unsafe { *update_data.as_ptr().offset(state.source_offset) };
1476            // SAFETY: the copied, injective destination layout proves this write
1477            // is initialized and in-bounds.
1478            unsafe { dest.write_at(state.dest_offset, value) };
1479            replay.advance(&mut state);
1480        }
1481        Ok(())
1482    }
1483
1484    /// Replay the compiled fused update window at the clamped runtime
1485    /// destination base (see [`DynamicSlicePlan`]'s fused window replay).
1486    fn execute_fused_window<T, I, W>(
1487        &self,
1488        dest: &mut W,
1489        update: &RawStridedRef<'_, T>,
1490        starts: &RawStridedRef<'_, I>,
1491        window: &FusedPairLayout,
1492    ) -> Result<()>
1493    where
1494        T: Copy + MaybeSendSync,
1495        I: GatherIndex,
1496        W: OverwriteWriter<T>,
1497    {
1498        let mut starts_storage = CoordScratch::new(self.operand_dims.len());
1499        let clamped_starts = starts_storage.as_mut_slice();
1500        read_clamped_starts(
1501            starts,
1502            &self.operand_dims,
1503            &self.update_dims,
1504            clamped_starts,
1505        )?;
1506        let dest_base = checked_strided_offset(dest.offset(), &self.dest_strides, clamped_starts)?;
1507        // SAFETY: the validated writer owns the destination allocation.
1508        let dest_ptr = unsafe { dest.data_ptr() }.cast::<MaybeUninit<T>>();
1509        // SAFETY: `window` fuses `update_dims` with the checked destination
1510        // and update strides. The update view has exactly `update_dims` and
1511        // its validated layout reaches every source offset; the clamped window
1512        // lies inside `dest_dims`, so every destination offset from
1513        // `dest_base` is reachable through the validated destination layout.
1514        // Compile proved the destination injective, and the exclusive writer
1515        // borrow does not overlap the shared update.
1516        unsafe {
1517            replay_fused_raw(
1518                dest_ptr,
1519                dest_base,
1520                update.data().as_ptr(),
1521                update.offset(),
1522                window,
1523                |slot: &mut MaybeUninit<T>, value: T| *slot = MaybeUninit::new(value),
1524                |value: T| value,
1525            );
1526        }
1527        Ok(())
1528    }
1529
1530    #[inline]
1531    fn uses_rank_one_contiguous_path(&self) -> bool {
1532        self.operand_dims.len() == 1
1533            && self.operand_strides[0] == 1
1534            && self.update_strides[0] == 1
1535            && self.dest_strides[0] == 1
1536    }
1537
1538    fn execute_rank_one_contiguous<T, I, W>(
1539        &self,
1540        dest: &mut W,
1541        update: &RawStridedRef<'_, T>,
1542        starts: &RawStridedRef<'_, I>,
1543    ) -> Result<()>
1544    where
1545        T: Copy + MaybeSendSync,
1546        I: GatherIndex,
1547        W: OverwriteWriter<T>,
1548    {
1549        let mut clamped_starts = [0usize; 1];
1550        read_clamped_starts(
1551            starts,
1552            &self.operand_dims,
1553            &self.update_dims,
1554            &mut clamped_starts,
1555        )?;
1556        let update_start =
1557            usize::try_from(update.offset()).map_err(|_| StridedError::OffsetOverflow)?;
1558        let dest_start = checked_offset_add(dest.offset(), 1, clamped_starts[0])?;
1559        let dest_start = usize::try_from(dest_start).map_err(|_| StridedError::OffsetOverflow)?;
1560        let update_end = update_start
1561            .checked_add(self.total)
1562            .ok_or(StridedError::OffsetOverflow)?;
1563        let update = update
1564            .data()
1565            .get(update_start..update_end)
1566            .ok_or(StridedError::OffsetOverflow)?;
1567        // SAFETY: the validated writer owns the destination allocation.
1568        let dest_ptr = unsafe { dest.data_ptr() };
1569        // SAFETY: the checked ranges are inside the destination allocation.
1570        unsafe {
1571            // Chunked across workers above the repository threshold; each
1572            // worker writes a disjoint chunk of the checked destination run.
1573            crate::threading::copy_contiguous(
1574                update.as_ptr(),
1575                dest_ptr.add(dest_start),
1576                self.total,
1577            );
1578        }
1579        Ok(())
1580    }
1581
1582    #[cfg(feature = "parallel")]
1583    fn execute_update_parallel<T, I, W>(
1584        &self,
1585        dest: &mut W,
1586        update: &RawStridedRef<'_, T>,
1587        starts: &RawStridedRef<'_, I>,
1588        nthreads: usize,
1589        replay: &WindowReplay,
1590    ) -> Result<()>
1591    where
1592        T: Copy + MaybeSendSync,
1593        I: GatherIndex,
1594        W: OverwriteWriter<T>,
1595    {
1596        let mut clamped_starts: AxisVec<usize> = (0..self.operand_dims.len()).map(|_| 0).collect();
1597        read_clamped_starts(
1598            starts,
1599            &self.operand_dims,
1600            &self.update_dims,
1601            &mut clamped_starts,
1602        )?;
1603        let source_base = update.offset();
1604        let dest_base = checked_strided_offset(dest.offset(), &self.dest_strides, &clamped_starts)?;
1605        let update_ptr = crate::threading::SendPtr(update.data().as_ptr() as *mut T);
1606        // SAFETY: the validated writer owns the destination allocation.
1607        let dest_ptr = crate::threading::SendPtr(unsafe { dest.data_ptr() });
1608
1609        crate::threading::parallel_map_reduce(
1610            0..self.total,
1611            nthreads,
1612            &|range| {
1613                let mut state = replay.decode(range.start, source_base, dest_base)?;
1614                let update_ptr = update_ptr.as_const();
1615                let dest_ptr = dest_ptr.as_ptr();
1616
1617                // INVARIANT: the initial operand copy completed before this
1618                // replay; workers own disjoint ranges and injective output
1619                // layouts make their writes distinct.
1620                for _ in range {
1621                    // SAFETY: checked replay bases/resets prove both offsets are
1622                    // within the initialized update and destination allocations.
1623                    unsafe {
1624                        dest_ptr
1625                            .offset(state.dest_offset)
1626                            .write(update_ptr.offset(state.source_offset).read());
1627                    }
1628                    replay.advance(&mut state);
1629                }
1630                Ok(())
1631            },
1632            &|left, right| left.and(right),
1633        )
1634    }
1635
1636    fn check_call<T, I, W>(
1637        &self,
1638        dest: &W,
1639        operand: &RawStridedRef<'_, T>,
1640        update: &RawStridedRef<'_, T>,
1641        starts: &RawStridedRef<'_, I>,
1642    ) -> Result<()>
1643    where
1644        W: OverwriteWriter<T>,
1645    {
1646        if dest.dims() != &self.dest_dims[..]
1647            || dest.strides() != &self.dest_strides[..]
1648            || operand.dims() != &self.operand_dims[..]
1649            || operand.strides() != &self.operand_strides[..]
1650            || update.dims() != &self.update_dims[..]
1651            || update.strides() != &self.update_strides[..]
1652            || starts.dims() != &self.start_dims[..]
1653            || starts.strides() != &self.start_strides[..]
1654        {
1655            return Err(StridedError::PlanLayoutMismatch);
1656        }
1657        Ok(())
1658    }
1659}
1660
1661impl ScatterPlan {
1662    /// Compile an additive scatter traversal.
1663    #[allow(clippy::too_many_arguments)]
1664    pub fn compile(
1665        operand_dims: &[usize],
1666        operand_strides: &[isize],
1667        index_dims: &[usize],
1668        index_strides: &[isize],
1669        update_dims: &[usize],
1670        update_strides: &[isize],
1671        dest_dims: &[usize],
1672        dest_strides: &[isize],
1673        spec: ScatterSpec,
1674    ) -> Result<Self> {
1675        if operand_dims.len() != operand_strides.len()
1676            || index_dims.len() != index_strides.len()
1677            || update_dims.len() != update_strides.len()
1678            || dest_dims.len() != dest_strides.len()
1679        {
1680            return Err(StridedError::StrideLengthMismatch);
1681        }
1682        checked_total_len(operand_dims)?;
1683        checked_total_len(index_dims)?;
1684        checked_total_len(update_dims)?;
1685        if dest_dims != operand_dims {
1686            return Err(StridedError::ShapeMismatch(
1687                dest_dims.to_vec(),
1688                operand_dims.to_vec(),
1689            ));
1690        }
1691        if !crate::layout_check::is_injective_layout(dest_dims, dest_strides) {
1692            return Err(StridedError::NonInjectiveOutputLayout);
1693        }
1694
1695        let operand_rank = operand_dims.len();
1696        validate_unique_axes(&spec.inserted_window_dims, operand_rank)?;
1697        validate_unique_axes(&spec.scatter_dims_to_operand_dims, operand_rank)?;
1698        if spec.index_vector_dim > index_dims.len() {
1699            return Err(StridedError::InvalidAxis {
1700                axis: spec.index_vector_dim,
1701                rank: index_dims.len() + 1,
1702            });
1703        }
1704        let index_vector_size = if spec.index_vector_dim == index_dims.len() {
1705            1
1706        } else {
1707            index_dims[spec.index_vector_dim]
1708        };
1709        if index_vector_size != spec.scatter_dims_to_operand_dims.len() {
1710            return Err(StridedError::RankMismatch(
1711                index_vector_size,
1712                spec.scatter_dims_to_operand_dims.len(),
1713            ));
1714        }
1715
1716        let batch_shape = index_batch_shape(index_dims, spec.index_vector_dim);
1717        let window_dims = operand_window_dims(operand_rank, &spec.inserted_window_dims);
1718        if spec.update_window_dims.len() != window_dims.len() {
1719            return Err(StridedError::RankMismatch(
1720                spec.update_window_dims.len(),
1721                window_dims.len(),
1722            ));
1723        }
1724
1725        let update_rank = update_dims.len();
1726        let expected_batch_rank = update_rank
1727            .checked_sub(spec.update_window_dims.len())
1728            .ok_or(StridedError::RankMismatch(
1729                spec.update_window_dims.len(),
1730                update_rank,
1731            ))?;
1732        if expected_batch_rank != batch_shape.len() {
1733            return Err(StridedError::RankMismatch(
1734                expected_batch_rank,
1735                batch_shape.len(),
1736            ));
1737        }
1738        validate_unique_axes(&spec.update_window_dims, update_rank)?;
1739
1740        let mut is_update_window_dim: AxisVec<bool> = (0..update_rank).map(|_| false).collect();
1741        for &axis in &spec.update_window_dims {
1742            is_update_window_dim[axis] = true;
1743        }
1744
1745        let mut batch_axis = 0usize;
1746        for axis in 0..update_rank {
1747            if !is_update_window_dim[axis] {
1748                if update_dims[axis] != batch_shape[batch_axis] {
1749                    return Err(StridedError::ShapeMismatch(
1750                        update_dims.to_vec(),
1751                        expected_scatter_update_shape(&batch_shape, &spec, update_dims).to_vec(),
1752                    ));
1753                }
1754                batch_axis += 1;
1755            }
1756        }
1757
1758        let mut window_shape: AxisVec<usize> = (0..operand_rank).map(|_| 1).collect();
1759        let mut window_shape_updates: AxisVec<usize> =
1760            AxisVec::with_capacity(spec.update_window_dims.len());
1761        for (pos, &update_axis) in spec.update_window_dims.iter().enumerate() {
1762            let dim = update_dims[update_axis];
1763            window_shape_updates.push(dim);
1764            window_shape[window_dims[pos]] = dim;
1765        }
1766        validate_window_sizes(operand_dims, &window_shape)?;
1767
1768        let batch_elems = checked_total_len(&batch_shape)?;
1769        let window_elems = checked_total_len(&window_shape_updates)?;
1770        let copy_plan = CopyPlan::compile(operand_dims, dest_strides, operand_strides)?;
1771        #[cfg(feature = "parallel")]
1772        ScatterReplay::validate(
1773            &batch_shape,
1774            index_dims,
1775            index_strides,
1776            spec.index_vector_dim,
1777            spec.scatter_dims_to_operand_dims.len(),
1778            update_dims,
1779            update_strides,
1780            &spec.update_window_dims,
1781            &is_update_window_dim,
1782            &window_shape_updates,
1783            &window_dims,
1784            dest_strides,
1785        )?;
1786        #[cfg(not(feature = "parallel"))]
1787        let replay = ScatterReplay::compile(
1788            &batch_shape,
1789            index_dims,
1790            index_strides,
1791            spec.index_vector_dim,
1792            spec.scatter_dims_to_operand_dims.len(),
1793            update_dims,
1794            update_strides,
1795            &spec.update_window_dims,
1796            &is_update_window_dim,
1797            &window_shape_updates,
1798            &window_dims,
1799            dest_strides,
1800        )?;
1801
1802        Ok(Self {
1803            operand_dims: operand_dims.into(),
1804            operand_strides: operand_strides.into(),
1805            index_dims: index_dims.into(),
1806            index_strides: index_strides.into(),
1807            update_dims: update_dims.into(),
1808            update_strides: update_strides.into(),
1809            dest_dims: dest_dims.into(),
1810            dest_strides: dest_strides.into(),
1811            spec,
1812            #[cfg(feature = "parallel")]
1813            batch_shape,
1814            #[cfg(feature = "parallel")]
1815            window_dims,
1816            window_shape,
1817            #[cfg(feature = "parallel")]
1818            window_shape_updates,
1819            #[cfg(feature = "parallel")]
1820            is_update_window_dim,
1821            batch_elems,
1822            window_elems,
1823            copy_plan,
1824            #[cfg(not(feature = "parallel"))]
1825            replay,
1826        })
1827    }
1828
1829    /// Execute the prepared additive scatter traversal.
1830    #[inline(always)]
1831    pub fn execute<T, I>(
1832        &self,
1833        dest: &mut RawStridedMut<'_, T>,
1834        operand: &RawStridedRef<'_, T>,
1835        scatter_indices: &RawStridedRef<'_, I>,
1836        updates: &RawStridedRef<'_, T>,
1837    ) -> Result<()>
1838    where
1839        T: Copy + Add<Output = T> + MaybeSendSync,
1840        I: GatherIndex,
1841    {
1842        self.check_call(dest, operand, scatter_indices, updates)?;
1843        self.copy_plan.execute(dest, operand)?;
1844        self.execute_updates(dest, scatter_indices, updates, |a, b| a + b)
1845    }
1846
1847    /// Execute additive scatter into a destination whose reachable slots may
1848    /// be uninitialized. The operand copy completes before any RMW access.
1849    pub(crate) fn execute_uninit<'a, T, I>(
1850        &self,
1851        dest: &'a mut RawStridedMut<'a, MaybeUninit<T>>,
1852        operand: &RawStridedRef<'_, T>,
1853        scatter_indices: &RawStridedRef<'_, I>,
1854        updates: &RawStridedRef<'_, T>,
1855        combine: fn(T, T) -> T,
1856    ) -> Result<()>
1857    where
1858        T: Copy + Add<Output = T> + MaybeSendSync,
1859        I: GatherIndex,
1860    {
1861        self.check_call(dest, operand, scatter_indices, updates)?;
1862        self.copy_plan
1863            .execute_uninit_then(dest, operand, |mut receipt| {
1864                self.execute_updates(&mut receipt, scatter_indices, updates, combine)
1865            })?
1866    }
1867
1868    #[inline(always)]
1869    fn execute_updates<T, I, W, F>(
1870        &self,
1871        dest: &mut W,
1872        scatter_indices: &RawStridedRef<'_, I>,
1873        updates: &RawStridedRef<'_, T>,
1874        combine: F,
1875    ) -> Result<()>
1876    where
1877        T: Copy + MaybeSendSync,
1878        I: GatherIndex,
1879        W: ReadModifyWrite<T>,
1880        F: Fn(T, T) -> T + Copy,
1881    {
1882        if self.batch_elems == 0 || self.window_elems == 0 {
1883            return Ok(());
1884        }
1885        if self.uses_rank_one_scalar_update_path() {
1886            return self.execute_rank_one_scalar_updates(dest, scatter_indices, updates, combine);
1887        }
1888        self.execute_generic_updates(dest, scatter_indices, updates, combine)
1889    }
1890
1891    #[inline(never)]
1892    fn execute_generic_updates<T, I, W, F>(
1893        &self,
1894        dest: &mut W,
1895        scatter_indices: &RawStridedRef<'_, I>,
1896        updates: &RawStridedRef<'_, T>,
1897        combine: F,
1898    ) -> Result<()>
1899    where
1900        T: Copy + MaybeSendSync,
1901        I: GatherIndex,
1902        W: ReadModifyWrite<T>,
1903        F: Fn(T, T) -> T + Copy,
1904    {
1905        // Overlapping additive updates are order-sensitive, so this remains a
1906        // deterministic serial replay until a combine-aware parallel plan exists.
1907        #[cfg(feature = "parallel")]
1908        let replay = ScatterReplay::compile(
1909            &self.batch_shape,
1910            &self.index_dims,
1911            &self.index_strides,
1912            self.spec.index_vector_dim,
1913            self.spec.scatter_dims_to_operand_dims.len(),
1914            &self.update_dims,
1915            &self.update_strides,
1916            &self.spec.update_window_dims,
1917            &self.is_update_window_dim,
1918            &self.window_shape_updates,
1919            &self.window_dims,
1920            &self.dest_strides,
1921        )?;
1922        #[cfg(not(feature = "parallel"))]
1923        let replay = &self.replay;
1924
1925        let mut operand_base_storage = CoordScratch::new(self.operand_dims.len());
1926        let operand_base = operand_base_storage.as_mut_slice();
1927        let mut batch_state = replay
1928            .batch
1929            .decode(0, scatter_indices.offset(), updates.offset())?;
1930        let index_data = scatter_indices.data();
1931        let update_data = updates.data();
1932
1933        // INVARIANT: batch and window replay metadata was checked at compile;
1934        // the indirect component loop below is the only data-dependent lookup.
1935        for _ in 0..self.batch_elems {
1936            operand_base.fill(0);
1937            for (&component_offset, &operand_axis) in replay
1938                .index_component_offsets
1939                .iter()
1940                .zip(self.spec.scatter_dims_to_operand_dims.iter())
1941            {
1942                let index_offset = batch_state
1943                    .source_offset
1944                    .checked_add(component_offset)
1945                    .ok_or(StridedError::OffsetOverflow)?;
1946                // SAFETY: checked batch replay and component offsets keep this
1947                // indirect index read inside the validated index allocation.
1948                let start = unsafe { *index_data.as_ptr().offset(index_offset) }.to_i64();
1949                operand_base[operand_axis] = clamp_window_start(
1950                    start,
1951                    self.operand_dims[operand_axis],
1952                    self.window_shape[operand_axis],
1953                );
1954            }
1955
1956            let dest_base = checked_strided_offset(dest.offset(), dest.strides(), operand_base)?;
1957            let mut window_state = replay
1958                .window
1959                .decode(0, batch_state.dest_offset, dest_base)?;
1960            for _ in 0..self.window_elems {
1961                // SAFETY: copy completion and checked replay state prove both
1962                // the update read and initialized destination RMW are valid.
1963                let value = unsafe { *update_data.as_ptr().offset(window_state.source_offset) };
1964                unsafe { dest.add_at(window_state.dest_offset, value, combine) };
1965                replay.window.advance(&mut window_state);
1966            }
1967            replay.batch.advance(&mut batch_state);
1968        }
1969        Ok(())
1970    }
1971
1972    fn uses_rank_one_scalar_update_path(&self) -> bool {
1973        self.operand_dims.len() == 1
1974            && self.index_dims.len() == 2
1975            && self.update_dims.len() == 1
1976            && self.dest_dims.len() == 1
1977            && self.operand_strides[0] == 1
1978            && self.index_strides[0] == 1
1979            && self.update_strides[0] == 1
1980            && self.dest_strides[0] == 1
1981            && self.index_dims[1] == 1
1982            && self.spec.update_window_dims.is_empty()
1983            && self.spec.inserted_window_dims.as_slice() == [0]
1984            && self.spec.scatter_dims_to_operand_dims.as_slice() == [0]
1985            && self.spec.index_vector_dim == 1
1986            && self.window_elems == 1
1987    }
1988
1989    #[inline(always)]
1990    fn execute_rank_one_scalar_updates<T, I, W, F>(
1991        &self,
1992        dest: &mut W,
1993        scatter_indices: &RawStridedRef<'_, I>,
1994        updates: &RawStridedRef<'_, T>,
1995        combine: F,
1996    ) -> Result<()>
1997    where
1998        T: Copy,
1999        I: GatherIndex,
2000        W: ReadModifyWrite<T>,
2001        F: Fn(T, T) -> T + Copy,
2002    {
2003        let mut index_offset = scatter_indices.offset();
2004        let mut update_offset = updates.offset();
2005        let dest_offset = dest.offset();
2006        let index_data = scatter_indices.data();
2007        let update_data = updates.data();
2008        // SAFETY: the validated writer owns the destination allocation.
2009        let dest_ptr = unsafe { dest.data_ptr() };
2010
2011        // INVARIANT: compile and check_call validated compact rank-one index
2012        // and update layouts. Ordered replay preserves repeated-index semantics.
2013        for _ in 0..self.batch_elems {
2014            // SAFETY: the invariant above proves index/update reads and the
2015            // clamped destination offset are in their validated allocations.
2016            unsafe {
2017                let start = (*index_data.as_ptr().offset(index_offset)).to_i64();
2018                let output_offset =
2019                    dest_offset + clamp_window_start(start, self.operand_dims[0], 1) as isize;
2020                let output = dest_ptr.offset(output_offset);
2021                output.write(combine(
2022                    output.read(),
2023                    *update_data.as_ptr().offset(update_offset),
2024                ));
2025            }
2026            index_offset += 1;
2027            update_offset += 1;
2028        }
2029        Ok(())
2030    }
2031
2032    fn check_call<T, I, W>(
2033        &self,
2034        dest: &W,
2035        operand: &RawStridedRef<'_, T>,
2036        scatter_indices: &RawStridedRef<'_, I>,
2037        updates: &RawStridedRef<'_, T>,
2038    ) -> Result<()>
2039    where
2040        W: OverwriteWriter<T>,
2041    {
2042        if dest.dims() != &self.dest_dims[..]
2043            || dest.strides() != &self.dest_strides[..]
2044            || operand.dims() != &self.operand_dims[..]
2045            || operand.strides() != &self.operand_strides[..]
2046            || scatter_indices.dims() != &self.index_dims[..]
2047            || scatter_indices.strides() != &self.index_strides[..]
2048            || updates.dims() != &self.update_dims[..]
2049            || updates.strides() != &self.update_strides[..]
2050        {
2051            return Err(StridedError::PlanLayoutMismatch);
2052        }
2053        Ok(())
2054    }
2055}
2056
2057fn validate_unique_axes(axes: &[usize], rank: usize) -> Result<()> {
2058    let mut seen = vec![false; rank];
2059    for &axis in axes {
2060        if axis >= rank {
2061            return Err(StridedError::InvalidAxis { axis, rank });
2062        }
2063        if seen[axis] {
2064            return Err(StridedError::InvalidAxis { axis, rank });
2065        }
2066        seen[axis] = true;
2067    }
2068    Ok(())
2069}
2070
2071fn validate_start_vector(start_dims: &[usize], operand_rank: usize) -> Result<()> {
2072    if start_dims.len() != 1 {
2073        return Err(StridedError::RankMismatch(start_dims.len(), 1));
2074    }
2075    if start_dims[0] != operand_rank {
2076        return Err(StridedError::RankMismatch(start_dims[0], operand_rank));
2077    }
2078    Ok(())
2079}
2080
2081fn validate_window_sizes(operand_dims: &[usize], window_sizes: &[usize]) -> Result<()> {
2082    if operand_dims.len() != window_sizes.len() {
2083        return Err(StridedError::RankMismatch(
2084            window_sizes.len(),
2085            operand_dims.len(),
2086        ));
2087    }
2088    for (axis, (&window, &dim)) in window_sizes.iter().zip(operand_dims.iter()).enumerate() {
2089        if window > dim {
2090            return Err(StridedError::InvalidAxis {
2091                axis,
2092                rank: operand_dims.len(),
2093            });
2094        }
2095    }
2096    Ok(())
2097}
2098
2099fn read_clamped_starts<I>(
2100    starts: &RawStridedRef<'_, I>,
2101    operand_dims: &[usize],
2102    window_sizes: &[usize],
2103    out: &mut [usize],
2104) -> Result<()>
2105where
2106    I: GatherIndex,
2107{
2108    debug_assert_eq!(operand_dims.len(), window_sizes.len());
2109    debug_assert_eq!(operand_dims.len(), out.len());
2110    for axis in 0..operand_dims.len() {
2111        let offset = checked_offset_add(starts.offset(), starts.strides()[0], axis)?;
2112        let start = unsafe { *starts.data().as_ptr().offset(offset) }.to_i64();
2113        out[axis] = clamp_window_start(start, operand_dims[axis], window_sizes[axis]);
2114    }
2115    Ok(())
2116}
2117
2118#[inline]
2119fn clamp_window_start(start: i64, dim_size: usize, window_size: usize) -> usize {
2120    let max_start = dim_size.saturating_sub(window_size) as i64;
2121    start.clamp(0, max_start) as usize
2122}
2123
2124fn expected_scatter_update_shape(
2125    batch_shape: &[usize],
2126    spec: &ScatterSpec,
2127    update_dims: &[usize],
2128) -> AxisVec<usize> {
2129    let mut expected: AxisVec<usize> = AxisVec::with_capacity(update_dims.len());
2130    let mut batch_axis = 0usize;
2131    for axis in 0..update_dims.len() {
2132        if spec.update_window_dims.contains(&axis) {
2133            expected.push(update_dims[axis]);
2134        } else {
2135            expected.push(batch_shape[batch_axis]);
2136            batch_axis += 1;
2137        }
2138    }
2139    expected
2140}
2141
2142fn operand_window_dims(rank: usize, collapsed_slice_dims: &[usize]) -> AxisVec<usize> {
2143    (0..rank)
2144        .filter(|axis| !collapsed_slice_dims.contains(axis))
2145        .collect()
2146}
2147
2148fn index_batch_shape(index_dims: &[usize], index_vector_dim: usize) -> AxisVec<usize> {
2149    if index_vector_dim == index_dims.len() {
2150        return index_dims.into();
2151    }
2152    index_dims
2153        .iter()
2154        .enumerate()
2155        .filter_map(|(axis, &dim)| (axis != index_vector_dim).then_some(dim))
2156        .collect()
2157}
2158
2159fn validate_layout_span(dims: &[usize], strides: &[isize]) -> Result<()> {
2160    if dims.len() != strides.len() {
2161        return Err(StridedError::StrideLengthMismatch);
2162    }
2163    let mut min_offset = 0isize;
2164    let mut max_offset = 0isize;
2165    for (&dim, &stride) in dims.iter().zip(strides.iter()) {
2166        let last =
2167            isize::try_from(dim.saturating_sub(1)).map_err(|_| StridedError::OffsetOverflow)?;
2168        let extent = stride
2169            .checked_mul(last)
2170            .ok_or(StridedError::OffsetOverflow)?;
2171        if extent < 0 {
2172            min_offset = min_offset
2173                .checked_add(extent)
2174                .ok_or(StridedError::OffsetOverflow)?;
2175        } else {
2176            max_offset = max_offset
2177                .checked_add(extent)
2178                .ok_or(StridedError::OffsetOverflow)?;
2179        }
2180    }
2181    Ok(())
2182}
2183
2184fn checked_replay_reset(dim: usize, step: isize) -> Result<isize> {
2185    if dim == 0 {
2186        return Ok(0);
2187    }
2188    let last = isize::try_from(dim - 1).map_err(|_| StridedError::OffsetOverflow)?;
2189    step.checked_mul(last)
2190        .and_then(isize::checked_neg)
2191        .ok_or(StridedError::OffsetOverflow)
2192}
2193
2194fn checked_total_len(dims: &[usize]) -> Result<usize> {
2195    if dims.is_empty() {
2196        return Ok(1);
2197    }
2198    dims.iter()
2199        .try_fold(1usize, |acc, &dim| acc.checked_mul(dim))
2200        .ok_or(StridedError::OffsetOverflow)
2201}
2202
2203fn checked_strided_offset(base: isize, strides: &[isize], index: &[usize]) -> Result<isize> {
2204    let mut offset = base;
2205    for (&stride, &coord) in strides.iter().zip(index.iter()) {
2206        offset = checked_offset_add(offset, stride, coord)?;
2207    }
2208    Ok(offset)
2209}
2210
2211fn checked_offset_add(base: isize, stride: isize, coord: usize) -> Result<isize> {
2212    let coord = isize::try_from(coord).map_err(|_| StridedError::OffsetOverflow)?;
2213    let scaled = stride
2214        .checked_mul(coord)
2215        .ok_or(StridedError::OffsetOverflow)?;
2216    base.checked_add(scaled).ok_or(StridedError::OffsetOverflow)
2217}
2218
2219struct CoordScratch {
2220    inline: [usize; RAW_FUSED_RANK_LIMIT],
2221    heap: Option<Vec<usize>>,
2222    len: usize,
2223}
2224
2225impl CoordScratch {
2226    fn new(len: usize) -> Self {
2227        if len <= RAW_FUSED_RANK_LIMIT {
2228            Self {
2229                inline: [0; RAW_FUSED_RANK_LIMIT],
2230                heap: None,
2231                len,
2232            }
2233        } else {
2234            Self {
2235                inline: [0; RAW_FUSED_RANK_LIMIT],
2236                heap: Some(vec![0; len]),
2237                len,
2238            }
2239        }
2240    }
2241
2242    fn as_mut_slice(&mut self) -> &mut [usize] {
2243        match &mut self.heap {
2244            Some(heap) => heap,
2245            None => &mut self.inline[..self.len],
2246        }
2247    }
2248}
2249
2250#[cfg(test)]
2251#[path = "gather_plan/tests/tests.rs"]
2252mod tests;
2253
2254#[cfg(all(test, feature = "parallel"))]
2255#[path = "gather_plan/tests/parallel_tests.rs"]
2256mod parallel_tests;