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