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