Skip to main content

strided_basic/
reduce_view.rs

1//! Reduce operations on dynamic-rank strided views.
2
3use crate::kernel::{
4    build_plan_fused, ensure_same_shape, for_each_inner_block_preordered, same_contiguous_layout,
5    sequential_contiguous_layout, total_len,
6};
7use crate::simd;
8use crate::view::{col_major_strides, StridedArray, StridedView, StridedViewMut};
9use crate::{Exec, Result, StridedError};
10use core::num::NonZeroUsize;
11use std::ops::Range;
12use strided_view::ElementOp;
13
14use crate::cost::KernelClass;
15use crate::fuse::compute_costs;
16use crate::inline::InlineVec;
17use crate::threading::{
18    for_each_inner_block_with_offsets, is_parallel, mapreduce_threaded_fold, SendPtr,
19};
20
21/// Number of independent accumulators of [`fold_contiguous`].
22const FOLD_LANES: usize = 16;
23
24/// Minimum outputs along a unit-stride kept axis for the column sweep of
25/// [`reduce_axis`].
26const AXIS_SWEEP_MIN_LEN: usize = 16;
27
28/// Byte budget of one output block of the column sweep of [`reduce_axis`].
29const AXIS_SWEEP_BLOCK_BYTES: usize = 64 * 1024;
30
31/// Folds a contiguous run with [`FOLD_LANES`] independent accumulators.
32///
33/// Lane `i` starts from element `i` and folds elements `i + FOLD_LANES`,
34/// `i + 2 * FOLD_LANES`, ... in order; the remainder is folded into the first
35/// lanes. `init` is combined exactly once, before lane 0, and the lanes are
36/// then combined left to right. Independent lanes let the compiler keep
37/// several vector accumulators in flight instead of one serial dependency
38/// chain, which is what makes a closure-based sum or max approach the speed
39/// of a hand written lane loop.
40#[inline(always)]
41fn fold_contiguous<T, Op, M, R, U>(src: &[T], init: U, map_fn: &M, reduce_fn: &R) -> U
42where
43    T: Copy,
44    Op: ElementOp<T>,
45    M: Fn(T) -> U,
46    R: Fn(U, U) -> U,
47    U: Clone,
48{
49    if src.len() < 2 * FOLD_LANES {
50        let mut acc = init;
51        for &value in src {
52            acc = reduce_fn(acc, map_fn(Op::apply(value)));
53        }
54        return acc;
55    }
56    let (head, rest) = src.split_at(FOLD_LANES);
57    let mut lanes: [U; FOLD_LANES] = core::array::from_fn(|lane| map_fn(Op::apply(head[lane])));
58    let mut chunks = rest.chunks_exact(FOLD_LANES);
59    for chunk in chunks.by_ref() {
60        for (lane, &value) in lanes.iter_mut().zip(chunk) {
61            *lane = reduce_fn(lane.clone(), map_fn(Op::apply(value)));
62        }
63    }
64    for (lane, &value) in lanes.iter_mut().zip(chunks.remainder()) {
65        *lane = reduce_fn(lane.clone(), map_fn(Op::apply(value)));
66    }
67    lanes.into_iter().fold(init, reduce_fn)
68}
69
70/// Folds `len` elements starting at `ptr` with step `stride`.
71///
72/// # Safety
73/// `ptr.offset(i * stride)` must be a readable element for every `i < len`.
74#[inline(always)]
75unsafe fn fold_run<T, Op, M, R, U>(
76    ptr: *const T,
77    stride: isize,
78    len: usize,
79    init: U,
80    map_fn: &M,
81    reduce_fn: &R,
82) -> U
83where
84    T: Copy,
85    Op: ElementOp<T>,
86    M: Fn(T) -> U,
87    R: Fn(U, U) -> U,
88    U: Clone,
89{
90    if stride == 1 {
91        // SAFETY: the caller proves `len` readable unit-stride elements.
92        let run = unsafe { std::slice::from_raw_parts(ptr, len) };
93        return fold_contiguous::<T, Op, M, R, U>(run, init, map_fn, reduce_fn);
94    }
95    let mut acc = init;
96    let mut cursor = ptr;
97    for index in 0..len {
98        // SAFETY: the caller proves every element of the run is readable.
99        acc = reduce_fn(acc, map_fn(Op::apply(unsafe { *cursor })));
100        if index + 1 < len {
101            cursor = cursor.wrapping_offset(stride);
102        }
103    }
104    acc
105}
106
107/// Full reduction with map function: `reduce(init, op, map.(src))`.
108///
109/// # Evaluation order
110///
111/// The result is independent of the pool when the call runs with a serial
112/// token ([`Exec::serial`]) or a budget of one: only serial branches run, and
113/// their fold order is fixed by the source layout. With a larger budget the
114/// fold is split over the budget, so the result is repeatable for a fixed
115/// shape, layout and budget but can differ between budgets.
116///
117/// In neither case is this a strict left to right fold: a contiguous source is
118/// folded with sixteen independent lanes (see the crate source for the exact
119/// lane assignment), non-contiguous sources are traversed in a cache-friendly
120/// loop order, and parallel execution folds per-task chunks independently
121/// and combines them in a tree. Floating point results can therefore differ
122/// from a sequential left to right sum in the last bits, and no bitwise
123/// equality across machines or target features is guaranteed.
124///
125/// Because of that, `reduce_fn` must be associative and commutative, and
126/// `init` must be an identity of `reduce_fn` whenever the reduction can run on
127/// several threads.
128///
129/// # Allocation
130///
131/// At rank ≤ [`INLINE_RANK`](strided_view::INLINE_RANK) the view metadata and
132/// the traversal plan are stored inline, so a call performs no heap
133/// allocation. Above that bound the planner may spill to the heap.
134///
135/// Because `map_fn` and `reduce_fn` are opaque closures this entry cannot
136/// select the dtype-specific SIMD kernels of [`crate::ErasedReducePlan`];
137/// callers that reduce with a known operation (sum, product, max, min) on a
138/// supported dtype can use that plan for the fastest path.
139pub fn reduce<T: Copy + Send + Sync, Op: ElementOp<T>, M, R, U>(
140    src: &StridedView<T, Op>,
141    map_fn: M,
142    reduce_fn: R,
143    init: U,
144    exec: Exec<'_>,
145) -> Result<U>
146where
147    M: Fn(T) -> U + Sync,
148    R: Fn(U, U) -> U + Sync,
149    U: Clone + Send + Sync,
150{
151    reduce_impl(src, map_fn, reduce_fn, init, exec)
152}
153
154fn reduce_impl<T: Copy + Send + Sync, Op: ElementOp<T>, M, R, U>(
155    src: &StridedView<T, Op>,
156    map_fn: M,
157    reduce_fn: R,
158    init: U,
159    exec: Exec<'_>,
160) -> Result<U>
161where
162    M: Fn(T) -> U + Sync,
163    R: Fn(U, U) -> U + Sync,
164    U: Clone + Send + Sync,
165{
166    let src_ptr = src.ptr();
167    let src_dims = src.dims();
168    let src_strides = src.strides();
169
170    // A zero-element reduction is exactly `init`. Return before planning: an
171    // empty view is allowed to carry an arbitrarily large stride, which the
172    // block-size arithmetic of the planner cannot represent.
173    if total_len(src_dims)? == 0 {
174        return Ok(init);
175    }
176
177    let contiguous =
178        sequential_contiguous_layout(KernelClass::Reduce, src_dims, &[src_strides], &exec)?;
179    if contiguous.is_some() {
180        let len = total_len(src_dims)?;
181        let src = unsafe { std::slice::from_raw_parts(src_ptr, len) };
182        return Ok(simd::dispatch_if_large(len, || {
183            fold_contiguous::<T, Op, M, R, U>(src, init, &map_fn, &reduce_fn)
184        }));
185    }
186
187    // Parallel contiguous fast path: split into scheduler chunks with slice-based iteration.
188    // This enables LLVM auto-vectorization on each chunk, unlike the general threaded path
189    // which uses scalar pointer-offset loops.
190    let total = total_len(src_dims)?;
191    if is_parallel(KernelClass::Reduce, total, &exec)
192        && same_contiguous_layout(src_dims, &[src_strides]).is_some()
193    {
194        let src_slice = unsafe { std::slice::from_raw_parts(src_ptr, total) };
195        let result = crate::threading::parallel_map_reduce(
196            0..total,
197            crate::cost::for_len_tree(exec, KernelClass::Reduce, total),
198            &|range| {
199                simd::dispatch_if_large(range.len(), || {
200                    fold_contiguous::<T, Op, M, R, U>(
201                        &src_slice[range],
202                        init.clone(),
203                        &map_fn,
204                        &reduce_fn,
205                    )
206                })
207            },
208            &|a, b| reduce_fn(a, b),
209        );
210        return Ok(result);
211    }
212
213    let strides_list: [&[isize]; 1] = [src_strides];
214
215    let (fused_dims, ordered_strides, plan) =
216        build_plan_fused(src_dims, &strides_list, None, std::mem::size_of::<T>());
217
218    let total = total_len(&fused_dims)?;
219    if is_parallel(KernelClass::Reduce, total, &exec) {
220        let src_send = SendPtr(src_ptr as *mut T);
221        let costs = compute_costs(&ordered_strides);
222        let initial_offsets: crate::inline::OperandVec<isize> =
223            ordered_strides.iter().map(|_| 0isize).collect();
224        let map_fn_ref = &map_fn;
225        let reduce_fn_ref = &reduce_fn;
226        let init_ref = &init;
227        // The reduction tree combines each split's partial result, so no
228        // worker-count-sized output-slot vector is needed (#306 scope).
229        let result = mapreduce_threaded_fold(
230            &fused_dims,
231            &plan.block,
232            &ordered_strides,
233            &initial_offsets,
234            &costs,
235            KernelClass::Reduce,
236            exec,
237            &|dims, blocks, strides_list, offsets| {
238                let mut acc = init_ref.clone();
239                for_each_inner_block_with_offsets(
240                    dims,
241                    blocks,
242                    strides_list,
243                    offsets,
244                    |offsets, len, strides| {
245                        let mut ptr = unsafe { src_send.as_const().offset(offsets[0]) };
246                        let stride = strides[0];
247                        for _ in 0..len {
248                            let val = Op::apply(unsafe { *ptr });
249                            let mapped = map_fn_ref(val);
250                            acc = reduce_fn_ref(acc.clone(), mapped);
251                            unsafe {
252                                ptr = ptr.offset(stride);
253                            }
254                        }
255                        Ok(())
256                    },
257                )?;
258                Ok(acc)
259            },
260            &|a, b| reduce_fn_ref(a, b),
261        )?;
262        return Ok(result);
263    }
264
265    let mut acc = init;
266    let initial_offsets: crate::inline::OperandVec<isize> =
267        ordered_strides.iter().map(|_| 0isize).collect();
268    for_each_inner_block_preordered(
269        &fused_dims,
270        &plan.block,
271        &ordered_strides,
272        &initial_offsets,
273        |offsets, len, strides| {
274            let mut ptr = unsafe { src_ptr.offset(offsets[0]) };
275            let stride = strides[0];
276            for _ in 0..len {
277                let val = Op::apply(unsafe { *ptr });
278                let mapped = map_fn(val);
279                acc = reduce_fn(acc.clone(), mapped);
280                unsafe {
281                    ptr = ptr.offset(stride);
282                }
283            }
284            Ok(())
285        },
286    )?;
287
288    Ok(acc)
289}
290
291/// Folds a contiguous run of two operands with [`FOLD_LANES`] independent
292/// accumulators.
293///
294/// Same lane scheme as [`fold_contiguous`]: lane `i` starts from the pair at
295/// index `i` and folds the pairs at `i + FOLD_LANES`, `i + 2 * FOLD_LANES`,
296/// ... in order, the remainder goes to the first lanes, and `init` is combined
297/// once before the lanes are combined left to right.
298///
299/// The scheduling below deliberately mirrors [`fold_contiguous`] instead of
300/// sharing a generic "map index to value" lane helper: routing the existing
301/// single-operand fold through such an accessor was measured at 0.34 ns/element
302/// against 0.11 ns/element on a contiguous `f64` `reduce` (1 MiB, release,
303/// `--release -j 4`), because the `chunks_exact` + `zip` form is what lets the
304/// compiler drop the bound checks and vectorize the 16 lanes. Keep the two
305/// bodies in step by hand; both must use the lane assignment documented here.
306#[inline(always)]
307fn fold_contiguous2<A, B, OpA, OpB, M, R, U>(
308    a: &[A],
309    b: &[B],
310    init: U,
311    map_fn: &M,
312    reduce_fn: &R,
313) -> U
314where
315    A: Copy,
316    B: Copy,
317    OpA: ElementOp<A>,
318    OpB: ElementOp<B>,
319    M: Fn(A, B) -> U,
320    R: Fn(U, U) -> U,
321    U: Clone,
322{
323    debug_assert_eq!(a.len(), b.len());
324    if a.len() < 2 * FOLD_LANES {
325        let mut acc = init;
326        for (&value_a, &value_b) in a.iter().zip(b) {
327            acc = reduce_fn(acc, map_fn(OpA::apply(value_a), OpB::apply(value_b)));
328        }
329        return acc;
330    }
331    let (head_a, rest_a) = a.split_at(FOLD_LANES);
332    let (head_b, rest_b) = b.split_at(FOLD_LANES);
333    let mut lanes: [U; FOLD_LANES] =
334        core::array::from_fn(|lane| map_fn(OpA::apply(head_a[lane]), OpB::apply(head_b[lane])));
335    let mut chunks_a = rest_a.chunks_exact(FOLD_LANES);
336    let mut chunks_b = rest_b.chunks_exact(FOLD_LANES);
337    for (chunk_a, chunk_b) in chunks_a.by_ref().zip(chunks_b.by_ref()) {
338        for ((lane, &value_a), &value_b) in lanes.iter_mut().zip(chunk_a).zip(chunk_b) {
339            *lane = reduce_fn(
340                lane.clone(),
341                map_fn(OpA::apply(value_a), OpB::apply(value_b)),
342            );
343        }
344    }
345    for ((lane, &value_a), &value_b) in lanes
346        .iter_mut()
347        .zip(chunks_a.remainder())
348        .zip(chunks_b.remainder())
349    {
350        *lane = reduce_fn(
351            lane.clone(),
352            map_fn(OpA::apply(value_a), OpB::apply(value_b)),
353        );
354    }
355    lanes.into_iter().fold(init, reduce_fn)
356}
357
358/// Folds `len` operand pairs starting at `a_ptr`/`b_ptr` with the given
359/// strides.
360///
361/// # Safety
362/// `a_ptr.offset(i * a_stride)` and `b_ptr.offset(i * b_stride)` must be
363/// readable elements for every `i < len`.
364#[inline(always)]
365#[allow(clippy::too_many_arguments)]
366unsafe fn fold_run2<A, B, OpA, OpB, M, R, U>(
367    a_ptr: *const A,
368    a_stride: isize,
369    b_ptr: *const B,
370    b_stride: isize,
371    len: usize,
372    init: &U,
373    map_fn: &M,
374    reduce_fn: &R,
375) -> U
376where
377    A: Copy,
378    B: Copy,
379    OpA: ElementOp<A>,
380    OpB: ElementOp<B>,
381    M: Fn(A, B) -> U,
382    R: Fn(U, U) -> U,
383    U: Clone,
384{
385    if a_stride == 1 && b_stride == 1 {
386        // SAFETY: the caller proves `len` readable unit-stride elements of
387        // both operands.
388        let a = unsafe { std::slice::from_raw_parts(a_ptr, len) };
389        let b = unsafe { std::slice::from_raw_parts(b_ptr, len) };
390        return fold_contiguous2::<A, B, OpA, OpB, M, R, U>(a, b, init.clone(), map_fn, reduce_fn);
391    }
392    let mut acc = init.clone();
393    let mut cursor_a = a_ptr;
394    let mut cursor_b = b_ptr;
395    for index in 0..len {
396        // SAFETY: the caller proves every pair of the run is readable.
397        let mapped = map_fn(
398            OpA::apply(unsafe { *cursor_a }),
399            OpB::apply(unsafe { *cursor_b }),
400        );
401        acc = reduce_fn(acc, mapped);
402        if index + 1 < len {
403            cursor_a = cursor_a.wrapping_offset(a_stride);
404            cursor_b = cursor_b.wrapping_offset(b_stride);
405        }
406    }
407    acc
408}
409
410/// Two-input mapped reduction: `reduce(init, op, map.(a[i], b[i]))`.
411///
412/// This is the two-operand sibling of [`reduce`]. It exists so that the
413/// accumulator type `U` is chosen by the caller instead of being tied to the
414/// operand types by `Mul`, which is what makes a widening accumulation
415/// possible — for example an `f32` dot accumulated in `f64`:
416///
417/// ```
418/// use strided_basic::{zip_reduce, Exec, StridedArray};
419///
420/// let a = StridedArray::<f32>::from_fn_col_major(&[2, 3], |i| (1 + i[0] + 2 * i[1]) as f32);
421/// let b = StridedArray::<f32>::from_fn_col_major(&[2, 3], |i| (1 + i[0]) as f32 * 0.5);
422/// let dot: f64 = zip_reduce(
423///     &a.view(),
424///     &b.view(),
425///     |x: f32, y: f32| x as f64 * y as f64,
426///     |x, y| x + y,
427///     0.0,
428///     Exec::serial(),
429/// )
430/// .unwrap();
431/// assert_eq!(dot, 16.5);
432/// ```
433///
434/// Both operands must have the same shape. Each operand's [`ElementOp`] is
435/// applied before `map_fn`, so conjugation and transposition flags are honoured
436/// without materializing a conjugated copy. Permuted, gapped, negative and
437/// stride-0 (broadcast) operands are supported: the traversal is the fused
438/// plan of [`reduce`], extended to two stride lists.
439///
440/// # Evaluation order
441///
442/// The result is independent of the pool when the call runs with a serial
443/// token ([`Exec::serial`]) or a budget of one: only serial branches run, and
444/// their fold order is fixed by the operand layouts. With a larger budget the
445/// fold is split over the budget, so the result is repeatable for a fixed
446/// shape, layout and budget but can differ between budgets.
447///
448/// In neither case is this a strict left to right fold: a contiguous operand
449/// pair is folded with the same sixteen independent lanes as [`reduce`], the
450/// general plan is traversed in a cache-friendly loop order, and parallel
451/// execution folds per-worker chunks independently and combines them in a
452/// tree. Floating point results can therefore differ from a sequential left to
453/// right sum in the last bits, and no bitwise equality across machines or
454/// target features is guaranteed.
455///
456/// Because of that, `reduce_fn` must be associative and commutative, and
457/// `init` must be an identity of `reduce_fn` whenever the reduction can run on
458/// several threads.
459///
460/// # Allocation
461///
462/// At rank ≤ [`INLINE_RANK`](strided_view::INLINE_RANK) the view metadata and
463/// the two-operand traversal plan are stored inline, so a call performs no heap
464/// allocation. Above that bound the planner may spill to the heap.
465///
466/// # Errors
467///
468/// Returns [`StridedError::RankMismatch`] / [`StridedError::ShapeMismatch`] when
469/// the operand shapes differ, and [`StridedError::OffsetOverflow`] for
470/// unrepresentable element counts.
471pub fn zip_reduce<A, B, OpA, OpB, M, R, U>(
472    a: &StridedView<A, OpA>,
473    b: &StridedView<B, OpB>,
474    map_fn: M,
475    reduce_fn: R,
476    init: U,
477    exec: Exec<'_>,
478) -> Result<U>
479where
480    A: Copy + Send + Sync,
481    B: Copy + Send + Sync,
482    OpA: ElementOp<A>,
483    OpB: ElementOp<B>,
484    M: Fn(A, B) -> U + Sync,
485    R: Fn(U, U) -> U + Sync,
486    U: Clone + Send + Sync,
487{
488    ensure_same_shape(a.dims(), b.dims())?;
489
490    let a_ptr = a.ptr();
491    let b_ptr = b.ptr();
492    let dims = a.dims();
493    let a_strides = a.strides();
494    let b_strides = b.strides();
495
496    // A zero-element reduction is exactly `init`. Return before planning: an
497    // empty view is allowed to carry an arbitrarily large stride, which the
498    // block-size arithmetic of the planner cannot represent.
499    let total = total_len(dims)?;
500    if total == 0 {
501        return Ok(init);
502    }
503
504    if sequential_contiguous_layout(KernelClass::Reduce, dims, &[a_strides, b_strides], &exec)?
505        .is_some()
506    {
507        return Ok(simd::dispatch_if_large(total, || {
508            // SAFETY: both operands prove `total` readable elements.
509            let sa = unsafe { std::slice::from_raw_parts(a_ptr, total) };
510            let sb = unsafe { std::slice::from_raw_parts(b_ptr, total) };
511            fold_contiguous2::<A, B, OpA, OpB, M, R, U>(sa, sb, init, &map_fn, &reduce_fn)
512        }));
513    }
514
515    // Parallel contiguous fast path: split into scheduler chunks so each chunk
516    // folds two contiguous slices, matching the serial contiguous kernel.
517    if is_parallel(KernelClass::Reduce, total, &exec)
518        && same_contiguous_layout(dims, &[a_strides, b_strides]).is_some()
519    {
520        // SAFETY: both operands prove `total` readable elements.
521        let sa = unsafe { std::slice::from_raw_parts(a_ptr, total) };
522        let sb = unsafe { std::slice::from_raw_parts(b_ptr, total) };
523        let result = crate::threading::parallel_map_reduce(
524            0..total,
525            crate::cost::for_len_tree(exec, KernelClass::Reduce, total),
526            &|range| {
527                let len = range.len();
528                simd::dispatch_if_large(len, || {
529                    fold_contiguous2::<A, B, OpA, OpB, M, R, U>(
530                        &sa[range.clone()],
531                        &sb[range],
532                        init.clone(),
533                        &map_fn,
534                        &reduce_fn,
535                    )
536                })
537            },
538            &|left, right| reduce_fn(left, right),
539        );
540        return Ok(result);
541    }
542
543    let strides_list: [&[isize]; 2] = [a_strides, b_strides];
544    let elem_size = std::mem::size_of::<A>().max(std::mem::size_of::<B>());
545    let (fused_dims, ordered_strides, plan) =
546        build_plan_fused(dims, &strides_list, None, elem_size);
547
548    let total = total_len(&fused_dims)?;
549    if is_parallel(KernelClass::Reduce, total, &exec) {
550        let a_send = SendPtr(a_ptr as *mut A);
551        let b_send = SendPtr(b_ptr as *mut B);
552        let costs = compute_costs(&ordered_strides);
553        let initial_offsets: crate::inline::OperandVec<isize> =
554            ordered_strides.iter().map(|_| 0isize).collect();
555        let map_fn_ref = &map_fn;
556        let reduce_fn_ref = &reduce_fn;
557        let init_ref = &init;
558        // The reduction tree combines each split's partial result, so no
559        // worker-count-sized output-slot vector is needed (#306 scope).
560        let result = mapreduce_threaded_fold(
561            &fused_dims,
562            &plan.block,
563            &ordered_strides,
564            &initial_offsets,
565            &costs,
566            KernelClass::Reduce,
567            exec,
568            &|dims, blocks, strides_list, offsets| {
569                let mut acc = init_ref.clone();
570                for_each_inner_block_with_offsets(
571                    dims,
572                    blocks,
573                    strides_list,
574                    offsets,
575                    |offsets, len, strides| {
576                        // SAFETY: the plan proves every pair of this block
577                        // is a readable element of both operands.
578                        acc = unsafe {
579                            fold_run2::<A, B, OpA, OpB, M, R, U>(
580                                a_send.as_const().offset(offsets[0]),
581                                strides[0],
582                                b_send.as_const().offset(offsets[1]),
583                                strides[1],
584                                len,
585                                &acc,
586                                map_fn_ref,
587                                reduce_fn_ref,
588                            )
589                        };
590                        Ok(())
591                    },
592                )?;
593                Ok(acc)
594            },
595            &|left, right| reduce_fn_ref(left, right),
596        )?;
597        return Ok(result);
598    }
599
600    let mut acc = init;
601    let initial_offsets: crate::inline::OperandVec<isize> =
602        ordered_strides.iter().map(|_| 0isize).collect();
603    for_each_inner_block_preordered(
604        &fused_dims,
605        &plan.block,
606        &ordered_strides,
607        &initial_offsets,
608        |offsets, len, strides| {
609            // SAFETY: the plan proves every pair of this block is a readable
610            // element of both operands.
611            acc = unsafe {
612                fold_run2::<A, B, OpA, OpB, M, R, U>(
613                    a_ptr.offset(offsets[0]),
614                    strides[0],
615                    b_ptr.offset(offsets[1]),
616                    strides[1],
617                    len,
618                    &acc,
619                    &map_fn,
620                    &reduce_fn,
621                )
622            };
623            Ok(())
624        },
625    )?;
626
627    Ok(acc)
628}
629
630/// Reduce along a single axis, returning a new `StridedArray`.
631///
632/// This allocates a column-major output and delegates to
633/// [`reduce_axis_into`].
634///
635/// Every output folds `init` once and then its reduced elements. When the
636/// reduced axis has unit stride those elements are folded with the lane
637/// scheme of [`reduce`], so `reduce_fn` must be associative and commutative.
638/// When the leading kept axis has unit stride, a block of adjacent outputs
639/// is accumulated one contiguous source column at a time, which keeps the
640/// left to right order per output. Outputs are split across the budget of
641/// `exec` once the number of source elements read exceeds the threading
642/// threshold; each output is computed by one task, so the result does not
643/// depend on the budget.
644pub fn reduce_axis<T: Copy + Send + Sync, Op: ElementOp<T>, M, R, U>(
645    src: &StridedView<T, Op>,
646    axis: usize,
647    map_fn: M,
648    reduce_fn: R,
649    init: U,
650    exec: Exec<'_>,
651) -> Result<StridedArray<U>>
652where
653    M: Fn(T) -> U + Sync,
654    R: Fn(U, U) -> U + Sync,
655    U: Clone + Send + Sync,
656{
657    let rank = src.ndim();
658    if axis >= rank {
659        return Err(StridedError::InvalidAxis { axis, rank });
660    }
661    // Reject an element count beyond `usize` before allocating the output.
662    total_len(src.dims())?;
663    let out_dims: InlineVec<usize> = src
664        .dims()
665        .iter()
666        .enumerate()
667        .filter(|(i, _)| *i != axis)
668        .map(|(_, &d)| d)
669        .collect();
670    // A full reduction keeps the historical rank-1 shape-[1] output; the
671    // `_into` form accepts any single-element destination for that case.
672    let (out_dims, out_strides) = if out_dims.is_empty() {
673        let mut dims = InlineVec::new();
674        dims.push(1usize);
675        let strides = col_major_strides(&dims);
676        (dims, strides)
677    } else {
678        let strides = col_major_strides(&out_dims);
679        (out_dims, strides)
680    };
681    let total_out = total_len(&out_dims)?;
682    let mut out =
683        StridedArray::<U>::from_parts(vec![init.clone(); total_out], &out_dims, &out_strides, 0)?;
684    {
685        let mut dest = out.view_mut();
686        // The fresh column-major destination needs no validation.
687        reduce_axis_into_impl(&mut dest, src, axis, map_fn, reduce_fn, init, exec)?;
688    }
689    Ok(out)
690}
691
692/// Reduce along a single axis into caller-owned storage.
693///
694/// `dest` must have the source's shape with `axis` removed, or any
695/// single-element destination for a full reduction, and an injective layout.
696/// Every destination element is written: an empty reduced axis fills each
697/// output with `init`.
698///
699/// # Errors
700///
701/// Returns [`StridedError::InvalidAxis`] for `axis >= src.ndim()`,
702/// [`StridedError::RankMismatch`] / [`StridedError::ShapeMismatch`] for a
703/// destination shape that is not the reduced source shape,
704/// [`StridedError::NonInjectiveOutputLayout`] for an overlapping destination,
705/// and [`StridedError::OffsetOverflow`] for unrepresentable sizes.
706///
707/// # Examples
708///
709/// ```
710/// use strided_basic::{reduce_axis_into, Exec, StridedArray};
711/// # let src = StridedArray::<f64>::from_fn_col_major(&[2, 3], |i| (i[0] + 2 * i[1]) as f64);
712/// let mut out = StridedArray::<f64>::col_major(&[3]);
713/// reduce_axis_into(&mut out.view_mut(), &src.view(), 0, |x| x, |a, b| a + b, 0.0, Exec::serial()).unwrap();
714/// assert_eq!(out.get(&[0]), 1.0);
715/// ```
716///
717/// Validation runs here; `reduce_axis_into_impl` is the shared kernel. The
718/// array-returning wrapper owns a fresh column-major destination and calls the
719/// kernel directly, so the common path does not re-validate.
720pub fn reduce_axis_into<T: Copy + Send + Sync, Op: ElementOp<T>, M, R, U>(
721    dest: &mut StridedViewMut<U>,
722    src: &StridedView<T, Op>,
723    axis: usize,
724    map_fn: M,
725    reduce_fn: R,
726    init: U,
727    exec: Exec<'_>,
728) -> Result<()>
729where
730    M: Fn(T) -> U + Sync,
731    R: Fn(U, U) -> U + Sync,
732    U: Clone + Send + Sync,
733{
734    let rank = src.ndim();
735    if axis >= rank {
736        return Err(StridedError::InvalidAxis { axis, rank });
737    }
738    let src_dims = src.dims();
739    // Reject an element count beyond usize (a huge stride-0 broadcast) up
740    // front instead of looping over it.
741    total_len(src_dims)?;
742    let kept_len = rank - 1;
743    let dest_len = total_len(dest.dims())?;
744    if kept_len == 0 {
745        if dest_len != 1 {
746            return Err(StridedError::ShapeMismatch(vec![dest_len], vec![1]));
747        }
748    } else {
749        if dest.ndim() != kept_len {
750            return Err(StridedError::RankMismatch(dest.ndim(), kept_len));
751        }
752        let dest_dims = dest.dims();
753        let mut d = 0usize;
754        for (i, &src_dim) in src_dims.iter().enumerate() {
755            if i == axis {
756                continue;
757            }
758            if src_dim != dest_dims[d] {
759                let expected: Vec<usize> = src_dims
760                    .iter()
761                    .enumerate()
762                    .filter(|(i, _)| *i != axis)
763                    .map(|(_, &d)| d)
764                    .collect();
765                return Err(StridedError::ShapeMismatch(dest_dims.to_vec(), expected));
766            }
767            d += 1;
768        }
769        if !crate::layout_check::is_injective_layout(dest.dims(), dest.strides()) {
770            return Err(StridedError::NonInjectiveOutputLayout);
771        }
772    }
773    reduce_axis_into_impl(dest, src, axis, map_fn, reduce_fn, init, exec)
774}
775
776/// Shared kernel of [`reduce_axis_into`] and [`reduce_axis`].
777///
778/// Assumes the destination has the reduced source shape (or one element for a
779/// full reduction) and an injective layout; the public entry validates that,
780/// while the array-returning wrapper owns a freshly allocated column-major
781/// destination and can skip the re-validation.
782fn reduce_axis_into_impl<T: Copy + Send + Sync, Op: ElementOp<T>, M, R, U>(
783    dest: &mut StridedViewMut<U>,
784    src: &StridedView<T, Op>,
785    axis: usize,
786    map_fn: M,
787    reduce_fn: R,
788    init: U,
789    exec: Exec<'_>,
790) -> Result<()>
791where
792    M: Fn(T) -> U + Sync,
793    R: Fn(U, U) -> U + Sync,
794    U: Clone + Send + Sync,
795{
796    let rank = src.ndim();
797    let src_dims = src.dims();
798    let src_strides = src.strides();
799    let src_ptr = src.ptr();
800    let dest_len = total_len(dest.dims())?;
801    let axis_len = src_dims[axis];
802    let axis_stride = src_strides[axis];
803
804    if axis_len == 0 {
805        // No source elements to fold: every output is `init`. Fill the
806        // destination directly rather than traversing the source, because an
807        // empty source view carries a dangling base pointer and `offset` on it
808        // is UB even when the fold reads zero elements.
809        fill_outputs_with(dest, &init);
810        return Ok(());
811    }
812
813    if rank == 1 {
814        // Full reduction into a single destination element.
815        // SAFETY: the view proves `axis_len` readable elements along `axis`.
816        let acc = unsafe {
817            fold_run::<T, Op, M, R, U>(src_ptr, axis_stride, axis_len, init, &map_fn, &reduce_fn)
818        };
819        unsafe { *dest.as_mut_ptr() = acc };
820        return Ok(());
821    }
822    if dest_len == 0 {
823        return Ok(());
824    }
825
826    // Build the kept-axis metadata in one pass. A column-major destination is
827    // the common case (the array-returning wrapper always allocates one); its
828    // offset equals the linear output index, so its stride list is only
829    // materialized for a strided destination.
830    let dest_stride_view = dest.strides();
831    let mut kept: InlineVec<(usize, isize)> = InlineVec::new();
832    let mut dest_linear = true;
833    let mut prefix = 1usize;
834    for (i, (&dim, &src_stride)) in src_dims.iter().zip(src_strides).enumerate() {
835        if i == axis {
836            continue;
837        }
838        let d = kept.len();
839        kept.push((dim, src_stride));
840        if dest_linear && dest_stride_view[d] != prefix as isize {
841            dest_linear = false;
842        }
843        prefix = prefix.saturating_mul(dim);
844    }
845    let dest_strides: InlineVec<isize> = if dest_linear {
846        InlineVec::new()
847    } else {
848        dest_stride_view.iter().copied().collect()
849    };
850    let out_ptr = dest.as_mut_ptr();
851    let parts = AxisParts {
852        kept: &kept,
853        dest_strides: &dest_strides,
854        axis_len,
855        axis_stride,
856        map_fn: &map_fn,
857        reduce_fn: &reduce_fn,
858        init: &init,
859    };
860
861    let work = dest_len.saturating_mul(axis_len);
862    let exec = crate::cost::for_len(exec, KernelClass::Reduce, work)
863        .limit(NonZeroUsize::new(dest_len).unwrap_or(NonZeroUsize::MIN));
864    if !exec.is_serial() {
865        let src_send = SendPtr(src_ptr as *mut T);
866        let out_send = SendPtr(out_ptr);
867        let parts = &parts;
868        if dest_linear {
869            crate::threading::parallel_for_each(0..dest_len, exec, &|range| {
870                // SAFETY: worker ranges are disjoint, the destination is
871                // injective, and the view proves every source offset
872                // reachable from its kept coordinates.
873                unsafe {
874                    reduce_axis_range_linear::<T, Op, M, R, U>(
875                        src_send.as_const(),
876                        out_send.as_ptr(),
877                        range,
878                        parts,
879                    )
880                }
881            });
882        } else {
883            crate::threading::parallel_for_each(0..dest_len, exec, &|range| {
884                // SAFETY: as for the linear path.
885                unsafe {
886                    reduce_axis_range_strided::<T, Op, M, R, U>(
887                        src_send.as_const(),
888                        out_send.as_ptr(),
889                        range,
890                        parts,
891                    )
892                }
893            });
894        }
895        return Ok(());
896    }
897
898    // SAFETY: as for the parallel ranges, with one range covering all outputs.
899    if dest_linear {
900        unsafe {
901            reduce_axis_range_linear::<T, Op, M, R, U>(src_ptr, out_ptr, 0..dest_len, &parts)
902        };
903    } else {
904        unsafe {
905            reduce_axis_range_strided::<T, Op, M, R, U>(src_ptr, out_ptr, 0..dest_len, &parts)
906        };
907    }
908    Ok(())
909}
910
911/// Writes `value` into every logical destination element.
912///
913/// Used when the reduced axis is empty: the fold would read zero source
914/// elements, so the destination is exactly `init` everywhere and no source
915/// pointer may be formed. The odometer keeps the fill allocation-free within
916/// [`strided_view::INLINE_RANK`] and spills above it.
917fn fill_outputs_with<U: Clone>(dest: &mut StridedViewMut<U>, value: &U) {
918    let dims = dest.dims();
919    if dims.iter().any(|&dim| dim == 0) {
920        return;
921    }
922    let strides = dest.strides();
923    let rank = dims.len();
924    let total: usize = dims.iter().product();
925    let base = dest.as_mut_ptr();
926    if rank == 0 {
927        // SAFETY: a rank-0 destination is one writable element.
928        unsafe { *base = value.clone() };
929        return;
930    }
931    let mut coords_inline = [0usize; strided_view::INLINE_RANK];
932    let mut coords_heap: Vec<usize> = Vec::new();
933    let coords: &mut [usize] = if rank <= strided_view::INLINE_RANK {
934        &mut coords_inline[..rank]
935    } else {
936        coords_heap.resize(rank, 0);
937        &mut coords_heap[..]
938    };
939    let mut offset = 0isize;
940    for _ in 0..total {
941        // SAFETY: the destination is an injective view validated by the
942        // caller, and the cursor visits each logical element exactly once, so
943        // every written offset is in bounds.
944        unsafe { *base.offset(offset) = value.clone() };
945        for axis in 0..rank {
946            coords[axis] += 1;
947            offset += strides[axis];
948            if coords[axis] < dims[axis] {
949                break;
950            }
951            coords[axis] = 0;
952            offset -= dims[axis] as isize * strides[axis];
953        }
954    }
955}
956
957/// Shared inputs of [`reduce_axis_range_linear`] and
958/// [`reduce_axis_range_strided`].
959struct AxisParts<'a, M, R, U> {
960    /// `(extent, source stride)` of every kept axis, in output order.
961    kept: &'a [(usize, isize)],
962    /// Destination stride of every kept axis, in the same order.
963    dest_strides: &'a [isize],
964    axis_len: usize,
965    axis_stride: isize,
966    map_fn: &'a M,
967    reduce_fn: &'a R,
968    init: &'a U,
969}
970
971/// Column-major destination cursor: the destination offset is the linear output
972/// index, so no destination stride bookkeeping is needed.
973///
974/// Kept out of line so the common path stays the same single-cursor loop as the
975/// pre-split kernel.
976///
977/// # Safety
978/// `out` must be a column-major destination of `parts.kept` extents, `range` a
979/// subrange of its logical output indices that no other thread writes, and
980/// every source offset reachable from a kept coordinate plus
981/// `k * parts.axis_stride` for `k < parts.axis_len` must be a readable element
982/// of `src`.
983#[inline(never)]
984unsafe fn reduce_axis_range_linear<T, Op, M, R, U>(
985    src: *const T,
986    out: *mut U,
987    range: Range<usize>,
988    parts: &AxisParts<'_, M, R, U>,
989) where
990    T: Copy,
991    Op: ElementOp<T>,
992    M: Fn(T) -> U,
993    R: Fn(U, U) -> U,
994    U: Clone,
995{
996    let kept = parts.kept;
997    let (lead_extent, lead_stride) = kept[0];
998    let sweep = lead_stride == 1 && lead_extent >= AXIS_SWEEP_MIN_LEN && parts.axis_len > 1;
999    let block = (AXIS_SWEEP_BLOCK_BYTES / std::mem::size_of::<U>().max(1)).max(AXIS_SWEEP_MIN_LEN);
1000
1001    // Decode the first output once, then advance incrementally. A fixed stack
1002    // array with a spill covers every inline rank without the per-access enum
1003    // match a `TinyVec` cursor would add to the hot loop, and allocates only
1004    // above `INLINE_RANK` (#306 P1).
1005    let coords_len = kept.len();
1006    let mut coords_inline = [0usize; strided_view::INLINE_RANK];
1007    let mut coords_heap: Vec<usize> = Vec::new();
1008    let coords: &mut [usize] = if coords_len <= strided_view::INLINE_RANK {
1009        &mut coords_inline[..coords_len]
1010    } else {
1011        coords_heap.resize(coords_len, 0);
1012        &mut coords_heap[..]
1013    };
1014    let mut rest = range.start;
1015    let mut src_off = 0isize;
1016    for (coord, &(extent, stride)) in coords.iter_mut().zip(kept) {
1017        *coord = rest % extent;
1018        rest /= extent;
1019        src_off += *coord as isize * stride;
1020    }
1021
1022    let mut output = range.start;
1023    while output < range.end {
1024        let len = if sweep {
1025            (lead_extent - coords[0]).min(range.end - output).min(block)
1026        } else {
1027            1
1028        };
1029        if sweep {
1030            // SAFETY: the `len` outputs are inside the caller's range, their
1031            // unit-stride destination elements are contiguous, and their
1032            // unit-stride source columns are readable.
1033            unsafe { sweep_block::<T, Op, M, R, U>(src, src_off, out.add(output), len, parts) };
1034        } else {
1035            // SAFETY: one reachable output and its reduced run.
1036            unsafe {
1037                let acc = fold_run::<T, Op, M, R, U>(
1038                    src.offset(src_off),
1039                    parts.axis_stride,
1040                    parts.axis_len,
1041                    parts.init.clone(),
1042                    parts.map_fn,
1043                    parts.reduce_fn,
1044                );
1045                *out.add(output) = acc;
1046            }
1047        }
1048        output += len;
1049        if output >= range.end {
1050            break;
1051        }
1052        // Advance the kept coordinates by `len` along the leading axis; a
1053        // block never crosses the end of the leading axis.
1054        coords[0] += len;
1055        src_off += len as isize * lead_stride;
1056        let mut axis = 0;
1057        while coords[axis] == kept[axis].0 {
1058            src_off -= kept[axis].0 as isize * kept[axis].1;
1059            coords[axis] = 0;
1060            axis += 1;
1061            coords[axis] += 1;
1062            src_off += kept[axis].1;
1063        }
1064    }
1065}
1066
1067/// General strided destination cursor, tracking a destination offset next to
1068/// the source offset.
1069///
1070/// # Safety
1071/// `out` must be a destination of `parts.kept` extents with injective layout,
1072/// `range` a subrange of its logical output indices that no other thread
1073/// writes, and every source offset reachable from a kept coordinate plus
1074/// `k * parts.axis_stride` for `k < parts.axis_len` must be a readable element
1075/// of `src`.
1076#[inline(never)]
1077unsafe fn reduce_axis_range_strided<T, Op, M, R, U>(
1078    src: *const T,
1079    out: *mut U,
1080    range: Range<usize>,
1081    parts: &AxisParts<'_, M, R, U>,
1082) where
1083    T: Copy,
1084    Op: ElementOp<T>,
1085    M: Fn(T) -> U,
1086    R: Fn(U, U) -> U,
1087    U: Clone,
1088{
1089    let kept = parts.kept;
1090    let dest_strides = parts.dest_strides;
1091    let (lead_extent, lead_stride) = kept[0];
1092    let lead_dest_stride = dest_strides[0];
1093    let sweep = lead_stride == 1
1094        && lead_dest_stride == 1
1095        && lead_extent >= AXIS_SWEEP_MIN_LEN
1096        && parts.axis_len > 1;
1097    let block = (AXIS_SWEEP_BLOCK_BYTES / std::mem::size_of::<U>().max(1)).max(AXIS_SWEEP_MIN_LEN);
1098
1099    // Same fixed-array-plus-spill cursor as `reduce_axis_range_linear`: the
1100    // `TinyVec` enum match is not paid per output access in the hot loop and
1101    // only ranks above `INLINE_RANK` allocate (#306 P1).
1102    let coords_len = kept.len();
1103    let mut coords_inline = [0usize; strided_view::INLINE_RANK];
1104    let mut coords_heap: Vec<usize> = Vec::new();
1105    let coords: &mut [usize] = if coords_len <= strided_view::INLINE_RANK {
1106        &mut coords_inline[..coords_len]
1107    } else {
1108        coords_heap.resize(coords_len, 0);
1109        &mut coords_heap[..]
1110    };
1111    let mut rest = range.start;
1112    let mut src_off = 0isize;
1113    let mut dest_off = 0isize;
1114    for (coord, (&(extent, stride), &dest_stride)) in
1115        coords.iter_mut().zip(kept.iter().zip(dest_strides))
1116    {
1117        *coord = rest % extent;
1118        rest /= extent;
1119        src_off += *coord as isize * stride;
1120        dest_off += *coord as isize * dest_stride;
1121    }
1122
1123    let mut output = range.start;
1124    while output < range.end {
1125        let len = if sweep {
1126            (lead_extent - coords[0]).min(range.end - output).min(block)
1127        } else {
1128            1
1129        };
1130        if sweep {
1131            // SAFETY: as for the linear cursor; `lead_dest_stride == 1` makes
1132            // the `len` destinations contiguous.
1133            unsafe {
1134                sweep_block::<T, Op, M, R, U>(src, src_off, out.offset(dest_off), len, parts)
1135            };
1136        } else {
1137            // SAFETY: one reachable output and its reduced run.
1138            unsafe {
1139                let acc = fold_run::<T, Op, M, R, U>(
1140                    src.offset(src_off),
1141                    parts.axis_stride,
1142                    parts.axis_len,
1143                    parts.init.clone(),
1144                    parts.map_fn,
1145                    parts.reduce_fn,
1146                );
1147                *out.offset(dest_off) = acc;
1148            }
1149        }
1150        output += len;
1151        if output >= range.end {
1152            break;
1153        }
1154        coords[0] += len;
1155        src_off += len as isize * lead_stride;
1156        dest_off += len as isize * lead_dest_stride;
1157        let mut axis = 0;
1158        while coords[axis] == kept[axis].0 {
1159            src_off -= kept[axis].0 as isize * kept[axis].1;
1160            dest_off -= kept[axis].0 as isize * dest_strides[axis];
1161            coords[axis] = 0;
1162            axis += 1;
1163            coords[axis] += 1;
1164            src_off += kept[axis].1;
1165            dest_off += dest_strides[axis];
1166        }
1167    }
1168}
1169
1170/// Accumulates `len` adjacent outputs, one contiguous source column per
1171/// reduced index, keeping the left to right order per output.
1172///
1173/// # Safety
1174/// `out` must be `len` writable initialized outputs and `src + src_off +
1175/// k * parts.axis_stride` must start `len` readable elements for every
1176/// `k < parts.axis_len`.
1177#[inline(always)]
1178unsafe fn sweep_block<T, Op, M, R, U>(
1179    src: *const T,
1180    src_off: isize,
1181    out: *mut U,
1182    len: usize,
1183    parts: &AxisParts<'_, M, R, U>,
1184) where
1185    T: Copy,
1186    Op: ElementOp<T>,
1187    M: Fn(T) -> U,
1188    R: Fn(U, U) -> U,
1189    U: Clone,
1190{
1191    let map_fn = parts.map_fn;
1192    let reduce_fn = parts.reduce_fn;
1193    // SAFETY: forwarded caller contract.
1194    let out = unsafe { std::slice::from_raw_parts_mut(out, len) };
1195    let column = |k: usize| {
1196        // SAFETY: forwarded caller contract for reduced index `k`.
1197        unsafe {
1198            std::slice::from_raw_parts(src.offset(src_off + k as isize * parts.axis_stride), len)
1199        }
1200    };
1201    let first = column(0);
1202    for (slot, &value) in out.iter_mut().zip(first) {
1203        *slot = reduce_fn(parts.init.clone(), map_fn(Op::apply(value)));
1204    }
1205    let mut k = 1;
1206    while k + 4 <= parts.axis_len {
1207        let (c0, c1, c2, c3) = (column(k), column(k + 1), column(k + 2), column(k + 3));
1208        for (index, slot) in out.iter_mut().enumerate() {
1209            let mut acc = reduce_fn(slot.clone(), map_fn(Op::apply(c0[index])));
1210            acc = reduce_fn(acc, map_fn(Op::apply(c1[index])));
1211            acc = reduce_fn(acc, map_fn(Op::apply(c2[index])));
1212            acc = reduce_fn(acc, map_fn(Op::apply(c3[index])));
1213            *slot = acc;
1214        }
1215        k += 4;
1216    }
1217    while k < parts.axis_len {
1218        let values = column(k);
1219        for (slot, &value) in out.iter_mut().zip(values) {
1220            *slot = reduce_fn(slot.clone(), map_fn(Op::apply(value)));
1221        }
1222        k += 1;
1223    }
1224}