1#[cfg(feature = "parallel")]
4use crate::kernel::same_contiguous_layout;
5use crate::kernel::{
6 build_plan_fused, for_each_inner_block_preordered, sequential_contiguous_layout, total_len,
7};
8use crate::maybe_sync::{MaybeSendSync, MaybeSync};
9use crate::simd;
10use crate::view::{col_major_strides, StridedArray, StridedView, StridedViewMut};
11use crate::{Result, StridedError};
12use std::ops::Range;
13use strided_view::ElementOp;
14
15#[cfg(feature = "parallel")]
16use crate::fuse::compute_costs;
17use crate::inline::InlineVec;
18#[cfg(feature = "parallel")]
19use crate::threading::{
20 for_each_inner_block_with_offsets, mapreduce_threaded_fold, SendPtr, MINTHREADLENGTH,
21};
22
23const FOLD_LANES: usize = 16;
25
26const AXIS_SWEEP_MIN_LEN: usize = 16;
29
30const AXIS_SWEEP_BLOCK_BYTES: usize = 64 * 1024;
32
33#[inline(always)]
43fn fold_contiguous<T, Op, M, R, U>(src: &[T], init: U, map_fn: &M, reduce_fn: &R) -> U
44where
45 T: Copy,
46 Op: ElementOp<T>,
47 M: Fn(T) -> U,
48 R: Fn(U, U) -> U,
49 U: Clone,
50{
51 if src.len() < 2 * FOLD_LANES {
52 let mut acc = init;
53 for &value in src {
54 acc = reduce_fn(acc, map_fn(Op::apply(value)));
55 }
56 return acc;
57 }
58 let (head, rest) = src.split_at(FOLD_LANES);
59 let mut lanes: [U; FOLD_LANES] = core::array::from_fn(|lane| map_fn(Op::apply(head[lane])));
60 let mut chunks = rest.chunks_exact(FOLD_LANES);
61 for chunk in chunks.by_ref() {
62 for (lane, &value) in lanes.iter_mut().zip(chunk) {
63 *lane = reduce_fn(lane.clone(), map_fn(Op::apply(value)));
64 }
65 }
66 for (lane, &value) in lanes.iter_mut().zip(chunks.remainder()) {
67 *lane = reduce_fn(lane.clone(), map_fn(Op::apply(value)));
68 }
69 lanes.into_iter().fold(init, reduce_fn)
70}
71
72#[inline(always)]
77unsafe fn fold_run<T, Op, M, R, U>(
78 ptr: *const T,
79 stride: isize,
80 len: usize,
81 init: U,
82 map_fn: &M,
83 reduce_fn: &R,
84) -> U
85where
86 T: Copy,
87 Op: ElementOp<T>,
88 M: Fn(T) -> U,
89 R: Fn(U, U) -> U,
90 U: Clone,
91{
92 if stride == 1 {
93 let run = unsafe { std::slice::from_raw_parts(ptr, len) };
95 return fold_contiguous::<T, Op, M, R, U>(run, init, map_fn, reduce_fn);
96 }
97 let mut acc = init;
98 let mut cursor = ptr;
99 for index in 0..len {
100 acc = reduce_fn(acc, map_fn(Op::apply(unsafe { *cursor })));
102 if index + 1 < len {
103 cursor = cursor.wrapping_offset(stride);
104 }
105 }
106 acc
107}
108
109pub fn reduce<T: Copy + MaybeSendSync, Op: ElementOp<T>, M, R, U>(
127 src: &StridedView<T, Op>,
128 map_fn: M,
129 reduce_fn: R,
130 init: U,
131) -> Result<U>
132where
133 M: Fn(T) -> U + MaybeSync,
134 R: Fn(U, U) -> U + MaybeSync,
135 U: Clone + MaybeSendSync,
136{
137 reduce_impl(src, map_fn, reduce_fn, init)
138}
139
140fn reduce_impl<T: Copy + MaybeSendSync, Op: ElementOp<T>, M, R, U>(
141 src: &StridedView<T, Op>,
142 map_fn: M,
143 reduce_fn: R,
144 init: U,
145) -> Result<U>
146where
147 M: Fn(T) -> U + MaybeSync,
148 R: Fn(U, U) -> U + MaybeSync,
149 U: Clone + MaybeSendSync,
150{
151 let src_ptr = src.ptr();
152 let src_dims = src.dims();
153 let src_strides = src.strides();
154
155 let contiguous = sequential_contiguous_layout(src_dims, &[src_strides])?;
156 if contiguous.is_some() {
157 let len = total_len(src_dims)?;
158 let src = unsafe { std::slice::from_raw_parts(src_ptr, len) };
159 return Ok(simd::dispatch_if_large(len, || {
160 fold_contiguous::<T, Op, M, R, U>(src, init, &map_fn, &reduce_fn)
161 }));
162 }
163
164 #[cfg(feature = "parallel")]
168 {
169 let total = total_len(src_dims)?;
170 let nthreads = crate::execution_policy::rayon_threads();
171 if total > MINTHREADLENGTH
172 && nthreads > 1
173 && same_contiguous_layout(src_dims, &[src_strides]).is_some()
174 {
175 let src_slice = unsafe { std::slice::from_raw_parts(src_ptr, total) };
176 let result = crate::threading::parallel_map_reduce(
177 0..total,
178 nthreads,
179 &|range| {
180 simd::dispatch_if_large(range.len(), || {
181 fold_contiguous::<T, Op, M, R, U>(
182 &src_slice[range],
183 init.clone(),
184 &map_fn,
185 &reduce_fn,
186 )
187 })
188 },
189 &|a, b| reduce_fn(a, b),
190 );
191 return Ok(result);
192 }
193 }
194
195 let strides_list: [&[isize]; 1] = [src_strides];
196
197 let (fused_dims, ordered_strides, plan) =
198 build_plan_fused(src_dims, &strides_list, None, std::mem::size_of::<T>());
199
200 #[cfg(feature = "parallel")]
201 {
202 let total = total_len(&fused_dims)?;
203 let nthreads = crate::execution_policy::rayon_threads();
204 if total > MINTHREADLENGTH && nthreads > 1 {
205 let src_send = SendPtr(src_ptr as *mut T);
206 let costs = compute_costs(&ordered_strides);
207 let initial_offsets: crate::inline::OperandVec<isize> =
208 ordered_strides.iter().map(|_| 0isize).collect();
209 let map_fn_ref = &map_fn;
210 let reduce_fn_ref = &reduce_fn;
211 let init_ref = &init;
212 let result = mapreduce_threaded_fold(
215 &fused_dims,
216 &plan.block,
217 &ordered_strides,
218 &initial_offsets,
219 &costs,
220 nthreads,
221 &|dims, blocks, strides_list, offsets| {
222 let mut acc = init_ref.clone();
223 for_each_inner_block_with_offsets(
224 dims,
225 blocks,
226 strides_list,
227 offsets,
228 |offsets, len, strides| {
229 let mut ptr = unsafe { src_send.as_const().offset(offsets[0]) };
230 let stride = strides[0];
231 for _ in 0..len {
232 let val = Op::apply(unsafe { *ptr });
233 let mapped = map_fn_ref(val);
234 acc = reduce_fn_ref(acc.clone(), mapped);
235 unsafe {
236 ptr = ptr.offset(stride);
237 }
238 }
239 Ok(())
240 },
241 )?;
242 Ok(acc)
243 },
244 &|a, b| reduce_fn_ref(a, b),
245 )?;
246 return Ok(result);
247 }
248 }
249
250 let mut acc = init;
251 let initial_offsets: crate::inline::OperandVec<isize> =
252 ordered_strides.iter().map(|_| 0isize).collect();
253 for_each_inner_block_preordered(
254 &fused_dims,
255 &plan.block,
256 &ordered_strides,
257 &initial_offsets,
258 |offsets, len, strides| {
259 let mut ptr = unsafe { src_ptr.offset(offsets[0]) };
260 let stride = strides[0];
261 for _ in 0..len {
262 let val = Op::apply(unsafe { *ptr });
263 let mapped = map_fn(val);
264 acc = reduce_fn(acc.clone(), mapped);
265 unsafe {
266 ptr = ptr.offset(stride);
267 }
268 }
269 Ok(())
270 },
271 )?;
272
273 Ok(acc)
274}
275
276pub fn reduce_axis<T: Copy + MaybeSendSync, Op: ElementOp<T>, M, R, U>(
291 src: &StridedView<T, Op>,
292 axis: usize,
293 map_fn: M,
294 reduce_fn: R,
295 init: U,
296) -> Result<StridedArray<U>>
297where
298 M: Fn(T) -> U + MaybeSync,
299 R: Fn(U, U) -> U + MaybeSync,
300 U: Clone + MaybeSendSync,
301{
302 let rank = src.ndim();
303 if axis >= rank {
304 return Err(StridedError::InvalidAxis { axis, rank });
305 }
306 total_len(src.dims())?;
308 let out_dims: InlineVec<usize> = src
309 .dims()
310 .iter()
311 .enumerate()
312 .filter(|(i, _)| *i != axis)
313 .map(|(_, &d)| d)
314 .collect();
315 let (out_dims, out_strides) = if out_dims.is_empty() {
318 let mut dims = InlineVec::new();
319 dims.push(1usize);
320 let strides = col_major_strides(&dims);
321 (dims, strides)
322 } else {
323 let strides = col_major_strides(&out_dims);
324 (out_dims, strides)
325 };
326 let total_out = total_len(&out_dims)?;
327 let mut out =
328 StridedArray::<U>::from_parts(vec![init.clone(); total_out], &out_dims, &out_strides, 0)?;
329 {
330 let mut dest = out.view_mut();
331 reduce_axis_into_impl(&mut dest, src, axis, map_fn, reduce_fn, init)?;
333 }
334 Ok(out)
335}
336
337pub fn reduce_axis_into<T: Copy + MaybeSendSync, Op: ElementOp<T>, M, R, U>(
366 dest: &mut StridedViewMut<U>,
367 src: &StridedView<T, Op>,
368 axis: usize,
369 map_fn: M,
370 reduce_fn: R,
371 init: U,
372) -> Result<()>
373where
374 M: Fn(T) -> U + MaybeSync,
375 R: Fn(U, U) -> U + MaybeSync,
376 U: Clone + MaybeSendSync,
377{
378 let rank = src.ndim();
379 if axis >= rank {
380 return Err(StridedError::InvalidAxis { axis, rank });
381 }
382 let src_dims = src.dims();
383 total_len(src_dims)?;
386 let kept_len = rank - 1;
387 let dest_len = total_len(dest.dims())?;
388 if kept_len == 0 {
389 if dest_len != 1 {
390 return Err(StridedError::ShapeMismatch(vec![dest_len], vec![1]));
391 }
392 } else {
393 if dest.ndim() != kept_len {
394 return Err(StridedError::RankMismatch(dest.ndim(), kept_len));
395 }
396 let dest_dims = dest.dims();
397 let mut d = 0usize;
398 for (i, &src_dim) in src_dims.iter().enumerate() {
399 if i == axis {
400 continue;
401 }
402 if src_dim != dest_dims[d] {
403 let expected: Vec<usize> = src_dims
404 .iter()
405 .enumerate()
406 .filter(|(i, _)| *i != axis)
407 .map(|(_, &d)| d)
408 .collect();
409 return Err(StridedError::ShapeMismatch(dest_dims.to_vec(), expected));
410 }
411 d += 1;
412 }
413 if !crate::layout_check::is_injective_layout(dest.dims(), dest.strides()) {
414 return Err(StridedError::NonInjectiveOutputLayout);
415 }
416 }
417 reduce_axis_into_impl(dest, src, axis, map_fn, reduce_fn, init)
418}
419
420fn reduce_axis_into_impl<T: Copy + MaybeSendSync, Op: ElementOp<T>, M, R, U>(
427 dest: &mut StridedViewMut<U>,
428 src: &StridedView<T, Op>,
429 axis: usize,
430 map_fn: M,
431 reduce_fn: R,
432 init: U,
433) -> Result<()>
434where
435 M: Fn(T) -> U + MaybeSync,
436 R: Fn(U, U) -> U + MaybeSync,
437 U: Clone + MaybeSendSync,
438{
439 let rank = src.ndim();
440 let src_dims = src.dims();
441 let src_strides = src.strides();
442 let src_ptr = src.ptr();
443 let dest_len = total_len(dest.dims())?;
444 let axis_len = src_dims[axis];
445 let axis_stride = src_strides[axis];
446
447 if axis_len == 0 {
448 fill_outputs_with(dest, &init);
453 return Ok(());
454 }
455
456 if rank == 1 {
457 let acc = unsafe {
460 fold_run::<T, Op, M, R, U>(src_ptr, axis_stride, axis_len, init, &map_fn, &reduce_fn)
461 };
462 unsafe { *dest.as_mut_ptr() = acc };
463 return Ok(());
464 }
465 if dest_len == 0 {
466 return Ok(());
467 }
468
469 let dest_stride_view = dest.strides();
474 let mut kept: InlineVec<(usize, isize)> = InlineVec::new();
475 let mut dest_linear = true;
476 let mut prefix = 1usize;
477 for (i, (&dim, &src_stride)) in src_dims.iter().zip(src_strides).enumerate() {
478 if i == axis {
479 continue;
480 }
481 let d = kept.len();
482 kept.push((dim, src_stride));
483 if dest_linear && dest_stride_view[d] != prefix as isize {
484 dest_linear = false;
485 }
486 prefix = prefix.saturating_mul(dim);
487 }
488 let dest_strides: InlineVec<isize> = if dest_linear {
489 InlineVec::new()
490 } else {
491 dest_stride_view.iter().copied().collect()
492 };
493 let out_ptr = dest.as_mut_ptr();
494 let parts = AxisParts {
495 kept: &kept,
496 dest_strides: &dest_strides,
497 axis_len,
498 axis_stride,
499 map_fn: &map_fn,
500 reduce_fn: &reduce_fn,
501 init: &init,
502 };
503
504 #[cfg(feature = "parallel")]
505 {
506 let work = dest_len.saturating_mul(axis_len);
507 let nthreads = crate::threading::parallel_threads_for_len(work).min(dest_len);
508 if nthreads > 1 {
509 let src_send = SendPtr(src_ptr as *mut T);
510 let out_send = SendPtr(out_ptr);
511 let parts = &parts;
512 if dest_linear {
513 crate::threading::parallel_for_each(0..dest_len, nthreads, &|range| {
514 unsafe {
518 reduce_axis_range_linear::<T, Op, M, R, U>(
519 src_send.as_const(),
520 out_send.as_ptr(),
521 range,
522 parts,
523 )
524 }
525 });
526 } else {
527 crate::threading::parallel_for_each(0..dest_len, nthreads, &|range| {
528 unsafe {
530 reduce_axis_range_strided::<T, Op, M, R, U>(
531 src_send.as_const(),
532 out_send.as_ptr(),
533 range,
534 parts,
535 )
536 }
537 });
538 }
539 return Ok(());
540 }
541 }
542
543 if dest_linear {
545 unsafe {
546 reduce_axis_range_linear::<T, Op, M, R, U>(src_ptr, out_ptr, 0..dest_len, &parts)
547 };
548 } else {
549 unsafe {
550 reduce_axis_range_strided::<T, Op, M, R, U>(src_ptr, out_ptr, 0..dest_len, &parts)
551 };
552 }
553 Ok(())
554}
555
556fn fill_outputs_with<U: Clone>(dest: &mut StridedViewMut<U>, value: &U) {
563 let dims = dest.dims();
564 if dims.iter().any(|&dim| dim == 0) {
565 return;
566 }
567 let strides = dest.strides();
568 let rank = dims.len();
569 let total: usize = dims.iter().product();
570 let base = dest.as_mut_ptr();
571 if rank == 0 {
572 unsafe { *base = value.clone() };
574 return;
575 }
576 let mut coords_inline = [0usize; strided_view::INLINE_RANK];
577 let mut coords_heap: Vec<usize> = Vec::new();
578 let coords: &mut [usize] = if rank <= strided_view::INLINE_RANK {
579 &mut coords_inline[..rank]
580 } else {
581 coords_heap.resize(rank, 0);
582 &mut coords_heap[..]
583 };
584 let mut offset = 0isize;
585 for _ in 0..total {
586 unsafe { *base.offset(offset) = value.clone() };
590 for axis in 0..rank {
591 coords[axis] += 1;
592 offset += strides[axis];
593 if coords[axis] < dims[axis] {
594 break;
595 }
596 coords[axis] = 0;
597 offset -= dims[axis] as isize * strides[axis];
598 }
599 }
600}
601
602struct AxisParts<'a, M, R, U> {
605 kept: &'a [(usize, isize)],
607 dest_strides: &'a [isize],
609 axis_len: usize,
610 axis_stride: isize,
611 map_fn: &'a M,
612 reduce_fn: &'a R,
613 init: &'a U,
614}
615
616#[inline(never)]
629unsafe fn reduce_axis_range_linear<T, Op, M, R, U>(
630 src: *const T,
631 out: *mut U,
632 range: Range<usize>,
633 parts: &AxisParts<'_, M, R, U>,
634) where
635 T: Copy,
636 Op: ElementOp<T>,
637 M: Fn(T) -> U,
638 R: Fn(U, U) -> U,
639 U: Clone,
640{
641 let kept = parts.kept;
642 let (lead_extent, lead_stride) = kept[0];
643 let sweep = lead_stride == 1 && lead_extent >= AXIS_SWEEP_MIN_LEN && parts.axis_len > 1;
644 let block = (AXIS_SWEEP_BLOCK_BYTES / std::mem::size_of::<U>().max(1)).max(AXIS_SWEEP_MIN_LEN);
645
646 let coords_len = kept.len();
651 let mut coords_inline = [0usize; strided_view::INLINE_RANK];
652 let mut coords_heap: Vec<usize> = Vec::new();
653 let coords: &mut [usize] = if coords_len <= strided_view::INLINE_RANK {
654 &mut coords_inline[..coords_len]
655 } else {
656 coords_heap.resize(coords_len, 0);
657 &mut coords_heap[..]
658 };
659 let mut rest = range.start;
660 let mut src_off = 0isize;
661 for (coord, &(extent, stride)) in coords.iter_mut().zip(kept) {
662 *coord = rest % extent;
663 rest /= extent;
664 src_off += *coord as isize * stride;
665 }
666
667 let mut output = range.start;
668 while output < range.end {
669 let len = if sweep {
670 (lead_extent - coords[0]).min(range.end - output).min(block)
671 } else {
672 1
673 };
674 if sweep {
675 unsafe { sweep_block::<T, Op, M, R, U>(src, src_off, out.add(output), len, parts) };
679 } else {
680 unsafe {
682 let acc = fold_run::<T, Op, M, R, U>(
683 src.offset(src_off),
684 parts.axis_stride,
685 parts.axis_len,
686 parts.init.clone(),
687 parts.map_fn,
688 parts.reduce_fn,
689 );
690 *out.add(output) = acc;
691 }
692 }
693 output += len;
694 if output >= range.end {
695 break;
696 }
697 coords[0] += len;
700 src_off += len as isize * lead_stride;
701 let mut axis = 0;
702 while coords[axis] == kept[axis].0 {
703 src_off -= kept[axis].0 as isize * kept[axis].1;
704 coords[axis] = 0;
705 axis += 1;
706 coords[axis] += 1;
707 src_off += kept[axis].1;
708 }
709 }
710}
711
712#[inline(never)]
722unsafe fn reduce_axis_range_strided<T, Op, M, R, U>(
723 src: *const T,
724 out: *mut U,
725 range: Range<usize>,
726 parts: &AxisParts<'_, M, R, U>,
727) where
728 T: Copy,
729 Op: ElementOp<T>,
730 M: Fn(T) -> U,
731 R: Fn(U, U) -> U,
732 U: Clone,
733{
734 let kept = parts.kept;
735 let dest_strides = parts.dest_strides;
736 let (lead_extent, lead_stride) = kept[0];
737 let lead_dest_stride = dest_strides[0];
738 let sweep = lead_stride == 1
739 && lead_dest_stride == 1
740 && lead_extent >= AXIS_SWEEP_MIN_LEN
741 && parts.axis_len > 1;
742 let block = (AXIS_SWEEP_BLOCK_BYTES / std::mem::size_of::<U>().max(1)).max(AXIS_SWEEP_MIN_LEN);
743
744 let coords_len = kept.len();
748 let mut coords_inline = [0usize; strided_view::INLINE_RANK];
749 let mut coords_heap: Vec<usize> = Vec::new();
750 let coords: &mut [usize] = if coords_len <= strided_view::INLINE_RANK {
751 &mut coords_inline[..coords_len]
752 } else {
753 coords_heap.resize(coords_len, 0);
754 &mut coords_heap[..]
755 };
756 let mut rest = range.start;
757 let mut src_off = 0isize;
758 let mut dest_off = 0isize;
759 for (coord, (&(extent, stride), &dest_stride)) in
760 coords.iter_mut().zip(kept.iter().zip(dest_strides))
761 {
762 *coord = rest % extent;
763 rest /= extent;
764 src_off += *coord as isize * stride;
765 dest_off += *coord as isize * dest_stride;
766 }
767
768 let mut output = range.start;
769 while output < range.end {
770 let len = if sweep {
771 (lead_extent - coords[0]).min(range.end - output).min(block)
772 } else {
773 1
774 };
775 if sweep {
776 unsafe {
779 sweep_block::<T, Op, M, R, U>(src, src_off, out.offset(dest_off), len, parts)
780 };
781 } else {
782 unsafe {
784 let acc = fold_run::<T, Op, M, R, U>(
785 src.offset(src_off),
786 parts.axis_stride,
787 parts.axis_len,
788 parts.init.clone(),
789 parts.map_fn,
790 parts.reduce_fn,
791 );
792 *out.offset(dest_off) = acc;
793 }
794 }
795 output += len;
796 if output >= range.end {
797 break;
798 }
799 coords[0] += len;
800 src_off += len as isize * lead_stride;
801 dest_off += len as isize * lead_dest_stride;
802 let mut axis = 0;
803 while coords[axis] == kept[axis].0 {
804 src_off -= kept[axis].0 as isize * kept[axis].1;
805 dest_off -= kept[axis].0 as isize * dest_strides[axis];
806 coords[axis] = 0;
807 axis += 1;
808 coords[axis] += 1;
809 src_off += kept[axis].1;
810 dest_off += dest_strides[axis];
811 }
812 }
813}
814
815#[inline(always)]
823unsafe fn sweep_block<T, Op, M, R, U>(
824 src: *const T,
825 src_off: isize,
826 out: *mut U,
827 len: usize,
828 parts: &AxisParts<'_, M, R, U>,
829) where
830 T: Copy,
831 Op: ElementOp<T>,
832 M: Fn(T) -> U,
833 R: Fn(U, U) -> U,
834 U: Clone,
835{
836 let map_fn = parts.map_fn;
837 let reduce_fn = parts.reduce_fn;
838 let out = unsafe { std::slice::from_raw_parts_mut(out, len) };
840 let column = |k: usize| {
841 unsafe {
843 std::slice::from_raw_parts(src.offset(src_off + k as isize * parts.axis_stride), len)
844 }
845 };
846 let first = column(0);
847 for (slot, &value) in out.iter_mut().zip(first) {
848 *slot = reduce_fn(parts.init.clone(), map_fn(Op::apply(value)));
849 }
850 let mut k = 1;
851 while k + 4 <= parts.axis_len {
852 let (c0, c1, c2, c3) = (column(k), column(k + 1), column(k + 2), column(k + 3));
853 for (index, slot) in out.iter_mut().enumerate() {
854 let mut acc = reduce_fn(slot.clone(), map_fn(Op::apply(c0[index])));
855 acc = reduce_fn(acc, map_fn(Op::apply(c1[index])));
856 acc = reduce_fn(acc, map_fn(Op::apply(c2[index])));
857 acc = reduce_fn(acc, map_fn(Op::apply(c3[index])));
858 *slot = acc;
859 }
860 k += 4;
861 }
862 while k < parts.axis_len {
863 let values = column(k);
864 for (slot, &value) in out.iter_mut().zip(values) {
865 *slot = reduce_fn(slot.clone(), map_fn(Op::apply(value)));
866 }
867 k += 1;
868 }
869}