Skip to main content

strided_basic/
reduce_view.rs

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