1use std::borrow::Cow;
14use std::sync::Arc;
15
16use rayon::prelude::*;
17use rayon::ThreadPool;
18
19use crate::backend::{Backend, BagSpec, BinSplit, DeviceTree, GradientStats, LeafRange, ObjectiveSetup};
20use crate::data::BinnedDataset;
21use crate::error::{Result, SilvicError};
22use crate::objective;
23use crate::quantize::{self, QuantSpec};
24
25const MIN_ADDS_PER_BIN: usize = 16;
27const MAX_BLOCKS_PER_GROUP: usize = 64;
30const SCRATCH_BINS: usize = 256;
32const PREFETCH_ROWS: usize = 16;
34const SCATTERED_SPAN: usize = 4;
36const MERGE_CHUNK_BINS: usize = 1024;
38const DENSE_RUN_NUM: usize = 3;
40const DENSE_RUN_DEN: usize = 4;
41const ROW_CHUNK: usize = 8192;
43const TRAVERSE_LANES: usize = 8;
45
46#[derive(Debug, Clone)]
48struct GroupLayout {
49 group: usize,
50 stride: usize,
51 hist_offset: usize,
53 bins: usize,
54 column_features: Vec<Option<usize>>,
56 offsets: Vec<usize>,
59 runs: Vec<(u32, u32)>,
61 adds: usize,
63}
64
65impl GroupLayout {
66 fn activate(&mut self, used: Option<&[bool]>) {
70 self.runs.clear();
71 let wanted = |f: usize| used.is_none_or(|used| used.get(f).copied().unwrap_or(true));
72 let width = self.column_features.len();
73 let active = self.column_features.iter().filter(|f| f.is_some_and(wanted)).count();
74 if active == 0 {
75 self.adds = 0;
76 return;
77 }
78 if active * DENSE_RUN_DEN >= width * DENSE_RUN_NUM {
79 self.runs.push((0, width as u32));
80 self.adds = width;
81 return;
82 }
83 self.adds = active;
84 let mut start = None;
85 for (column, feature) in self.column_features.iter().enumerate() {
86 match (feature.is_some_and(wanted), start) {
87 (true, None) => start = Some(column),
88 (false, Some(first)) => {
89 self.runs.push((first as u32, column as u32));
90 start = None;
91 }
92 _ => {}
93 }
94 }
95 if let Some(first) = start {
96 self.runs.push((first as u32, width as u32));
97 }
98 }
99}
100
101#[derive(Debug, Default)]
104pub struct CpuBackend<'a> {
105 pool: Option<Arc<ThreadPool>>,
106 borrowed: Vec<&'a BinnedDataset>,
107 datasets: Vec<Cow<'a, BinnedDataset>>,
109 layout: Vec<GroupLayout>,
110 total_bins: usize,
111 objective: Option<ObjectiveSetup>,
112 scores: Vec<Vec<f32>>,
113 grad: Vec<f32>,
114 hess: Vec<f32>,
115 packed: Vec<i32>,
116 max_quant: i64,
118 partition: Vec<u32>,
119 scratch: Vec<u32>,
120 hist: HistBuffers,
121 bag: Option<(BagSpec, Vec<u32>)>,
123}
124
125impl<'a> CpuBackend<'a> {
126 pub fn new() -> CpuBackend<'a> {
128 CpuBackend::default()
129 }
130
131 pub fn borrowing(train: &'a BinnedDataset, valid: &[&'a BinnedDataset]) -> CpuBackend<'a> {
134 let mut borrowed = vec![train];
135 borrowed.extend_from_slice(valid);
136 CpuBackend {
137 borrowed,
138 ..CpuBackend::default()
139 }
140 }
141
142 fn adopt(&self, data: &BinnedDataset) -> Cow<'a, BinnedDataset> {
143 match self.borrowed.iter().find(|held| std::ptr::eq(**held, data)) {
144 Some(held) => Cow::Borrowed(*held),
145 None => Cow::Owned(data.clone()),
146 }
147 }
148
149 fn train(&self) -> Result<&BinnedDataset> {
150 self.datasets
151 .first()
152 .map(|d| d.as_ref())
153 .ok_or_else(|| SilvicError::device("cpu backend used before load"))
154 }
155
156 fn check_range(&self, leaf: LeafRange) -> Result<()> {
157 let end = leaf.begin as usize + leaf.count as usize;
158 if end > self.partition.len() {
159 return Err(SilvicError::device(format!(
160 "leaf rows {}..{end} outside the partition of {} rows",
161 leaf.begin,
162 self.partition.len()
163 )));
164 }
165 Ok(())
166 }
167
168 #[cfg(test)]
170 pub(crate) fn set_gradients(&mut self, grad: &[f32], hess: &[f32]) {
171 self.grad.copy_from_slice(grad);
172 self.hess.copy_from_slice(hess);
173 }
174}
175
176fn group_layout(data: &BinnedDataset) -> Result<Vec<GroupLayout>> {
179 data.groups
180 .iter()
181 .enumerate()
182 .map(|(group, g)| {
183 let bins = g.total_bins as usize;
184 let stride = g.row_stride as usize;
185 let width = g
186 .features
187 .iter()
188 .map(|&f| data.features[f].column as usize + 1)
189 .max()
190 .unwrap_or(0);
191 let mut column_features = vec![None; width];
192 let mut offsets = vec![bins; width];
193 for &f in &g.features {
194 let meta = &data.features[f];
195 let offset = (meta.hist_offset as usize).checked_sub(g.hist_offset as usize);
196 match offset {
197 Some(offset) if offset + meta.num_bins as usize <= bins && (meta.column as usize) < stride => {
198 column_features[meta.column as usize] = Some(f);
199 offsets[meta.column as usize] = offset;
200 }
201 _ => {
202 return Err(SilvicError::data(format!(
203 "feature `{}` lies outside its group's bins or row",
204 meta.name
205 )))
206 }
207 }
208 }
209 let mut layout = GroupLayout {
210 group,
211 stride,
212 hist_offset: g.hist_offset as usize,
213 bins,
214 column_features,
215 offsets,
216 runs: Vec::new(),
217 adds: 0,
218 };
219 layout.activate(None);
220 Ok(layout)
221 })
222 .collect()
223}
224
225struct GroupView<'d> {
227 data: &'d [u8],
228 stride: usize,
229 hist_offset: usize,
230 bins: usize,
231 offsets: &'d [usize],
232 runs: &'d [(u32, u32)],
234 adds: usize,
236 width: usize,
238}
239
240fn group_views<'d>(layout: &'d [GroupLayout], data: &'d BinnedDataset) -> Vec<GroupView<'d>> {
241 layout
242 .iter()
243 .map(|g| GroupView {
244 data: &data.groups[g.group].data,
245 stride: g.stride,
246 hist_offset: g.hist_offset,
247 bins: g.bins,
248 offsets: &g.offsets,
249 runs: &g.runs,
250 adds: g.adds,
251 width: g.runs.last().map_or(0, |&(_, end)| end as usize),
252 })
253 .collect()
254}
255
256#[inline]
260fn accumulate<T: Copy, A>(group: &GroupView<'_>, rows: &[u32], values: &[T], acc: &mut [A], add: impl Fn(&mut A, T)) {
261 assert!(
262 acc.len() >= group.bins + SCRATCH_BINS,
263 "accumulator smaller than its group"
264 );
265 let span = match (rows.first(), rows.last()) {
266 (Some(&first), Some(&last)) => last.abs_diff(first) as usize,
267 _ => return,
268 };
269 if span > SCATTERED_SPAN * rows.len() {
270 accumulate_rows::<true, T, A>(group, rows, values, acc, add);
271 } else {
272 accumulate_rows::<false, T, A>(group, rows, values, acc, add);
273 }
274}
275
276#[inline(always)]
277fn accumulate_rows<const PREFETCH: bool, T: Copy, A>(
278 group: &GroupView<'_>,
279 rows: &[u32],
280 values: &[T],
281 acc: &mut [A],
282 add: impl Fn(&mut A, T),
283) {
284 let width = group.width;
285 for (i, (&row, &value)) in rows.iter().zip(values).enumerate() {
286 if PREFETCH {
287 if let Some(&ahead) = rows.get(i + PREFETCH_ROWS) {
288 let start = group.data.as_ptr().wrapping_add(ahead as usize * group.stride);
289 prefetch(start);
290 prefetch(start.wrapping_add(width.saturating_sub(1)));
291 }
292 }
293 let base = row as usize * group.stride;
294 let bins = &group.data[base..base + width];
295 if group.runs.len() == 1 {
296 add_run(bins, &group.offsets[..width], acc, &add, value);
297 } else {
298 for &(start, end) in group.runs {
299 let range = start as usize..end as usize;
300 add_run(&bins[range.clone()], &group.offsets[range], acc, &add, value);
301 }
302 }
303 }
304}
305
306#[inline(always)]
307fn add_run<T: Copy, A>(bins: &[u8], offsets: &[usize], acc: &mut [A], add: &impl Fn(&mut A, T), value: T) {
308 for (&bin, &offset) in bins.iter().zip(offsets) {
309 add(unsafe { acc.get_unchecked_mut(offset + usize::from(bin)) }, value);
313 }
314}
315
316#[inline]
319fn gather<T>(rows: &[u32], out: &mut [T], ahead: impl Fn(usize), value: impl Fn(usize) -> T) {
320 for (i, (slot, &row)) in out.iter_mut().zip(rows).enumerate() {
321 if let Some(&next) = rows.get(i + PREFETCH_ROWS) {
322 ahead(next as usize);
323 }
324 *slot = value(row as usize);
325 }
326}
327
328#[inline(always)]
330fn prefetch(address: *const u8) {
331 #[cfg(target_arch = "aarch64")]
334 unsafe {
335 std::arch::asm!("prfm pldl1keep, [{0}]", in(reg) address, options(nostack, readonly, preserves_flags));
336 }
337 #[cfg(target_arch = "x86_64")]
338 unsafe {
339 std::arch::x86_64::_mm_prefetch::<{ std::arch::x86_64::_MM_HINT_T0 }>(address.cast());
340 }
341 #[cfg(not(any(target_arch = "aarch64", target_arch = "x86_64")))]
342 let _ = address;
343}
344
345fn block_rows(rows: usize, group: &GroupView<'_>) -> usize {
348 let adds_per_row = group.adds.max(1);
349 let min_rows = (MIN_ADDS_PER_BIN * group.bins).div_ceil(adds_per_row);
350 min_rows.max(rows.div_ceil(MAX_BLOCKS_PER_GROUP)).max(1)
351}
352
353#[derive(Debug, Clone, Copy)]
355struct BlockTask {
356 group: usize,
357 start: usize,
358 end: usize,
359}
360
361#[derive(Debug, Default)]
363struct HistBuffers {
364 blocks: Vec<f64>,
366 packed_gh: Vec<i64>,
367 float_gh: Vec<[f64; 2]>,
368 tasks: Vec<BlockTask>,
369}
370
371fn histogram_blocks<T: Sync>(
375 groups: &[GroupView<'_>],
376 rows: &[u32],
377 values: &[T],
378 buffers: (&mut Vec<f64>, &mut Vec<BlockTask>),
379 out: &mut [f64],
380 block: impl Fn(&GroupView<'_>, &[u32], &[T], &mut [f64]) + Sync,
381) {
382 let (sums, tasks) = buffers;
383 tasks.clear();
384 for (group, view) in groups.iter().enumerate() {
385 if view.adds == 0 {
386 continue;
387 }
388 let size = block_rows(rows.len(), view);
389 let count = rows.len().div_ceil(size).max(1);
390 tasks.extend((0..count).map(|b| BlockTask {
391 group,
392 start: (b * size).min(rows.len()),
393 end: ((b + 1) * size).min(rows.len()),
394 }));
395 }
396 let slot = |task: &BlockTask| 2 * (groups[task.group].bins + SCRATCH_BINS);
397 sums.resize(tasks.iter().map(slot).sum(), 0.0);
398 let mut slots = Vec::with_capacity(tasks.len());
399 let mut rest = sums.as_mut_slice();
400 for task in tasks.iter() {
401 let (head, tail) = std::mem::take(&mut rest).split_at_mut(slot(task));
402 slots.push(head);
403 rest = tail;
404 }
405 tasks.par_iter().zip(slots).for_each(|(task, sums)| {
406 sums.fill(0.0);
407 block(
408 &groups[task.group],
409 &rows[task.start..task.end],
410 &values[task.start..task.end],
411 sums,
412 );
413 });
414 merge_blocks(groups, tasks, sums, out);
415}
416
417fn merge_blocks(groups: &[GroupView<'_>], tasks: &[BlockTask], sums: &[f64], out: &mut [f64]) {
420 let mut firsts = vec![(0usize, 0usize); groups.len()];
422 let mut offset = 0;
423 for task in tasks {
424 let entry = &mut firsts[task.group];
425 if entry.1 == 0 {
426 entry.0 = offset;
427 }
428 entry.1 += 1;
429 offset += 2 * (groups[task.group].bins + SCRATCH_BINS);
430 }
431 let mut order: Vec<usize> = (0..groups.len()).collect();
432 order.sort_by_key(|&g| groups[g].hist_offset);
433 let mut pieces = Vec::new();
434 let mut rest = out;
435 let mut position = 0;
436 for g in order {
437 let (gap, tail) = std::mem::take(&mut rest).split_at_mut(2 * (groups[g].hist_offset - position));
438 gap.fill(0.0);
439 let (target, tail) = tail.split_at_mut(2 * groups[g].bins);
440 rest = tail;
441 position = groups[g].hist_offset + groups[g].bins;
442 for (chunk, piece) in target.chunks_mut(2 * MERGE_CHUNK_BINS).enumerate() {
443 pieces.push((g, chunk * 2 * MERGE_CHUNK_BINS, piece));
444 }
445 }
446 rest.fill(0.0);
447 pieces.into_par_iter().for_each(|(g, begin, piece)| {
448 let (first, count) = firsts[g];
449 if count == 0 {
450 piece.fill(0.0);
451 return;
452 }
453 let stride = 2 * (groups[g].bins + SCRATCH_BINS);
454 let len = piece.len();
455 let block = |b: usize| &sums[first + b * stride + begin..][..len];
456 piece.copy_from_slice(block(0));
457 for b in 1..count {
458 for (value, sum) in piece.iter_mut().zip(block(b)) {
459 *value += sum;
460 }
461 }
462 });
463}
464
465fn ordered<T: Copy + Default + Send>(
467 rows: &[u32],
468 out: &mut Vec<T>,
469 ahead: impl Fn(usize) + Sync,
470 value: impl Fn(usize) -> T + Sync,
471) {
472 out.resize(rows.len(), T::default());
473 out.par_chunks_mut(ROW_CHUNK)
474 .zip(rows.par_chunks(ROW_CHUNK))
475 .for_each(|(out, rows)| gather(rows, out, &ahead, &value));
476}
477
478fn histogram_quantized(
480 groups: &[GroupView<'_>],
481 packed: &[i32],
482 max_quant: i64,
483 rows: &[u32],
484 buffers: &mut HistBuffers,
485 out: &mut [f64],
486) {
487 let row_limit = (i64::from(i32::MAX) / max_quant.max(1)) as usize;
489 ordered(
490 rows,
491 &mut buffers.packed_gh,
492 |row| prefetch(packed.as_ptr().wrapping_add(row).cast()),
493 |row| {
494 let (g, h) = quantize::unpack(packed[row]);
495 (i64::from(h) << 32) + i64::from(g)
496 },
497 );
498 let blocks = (&mut buffers.blocks, &mut buffers.tasks);
499 histogram_blocks(
500 groups,
501 rows,
502 &buffers.packed_gh,
503 blocks,
504 out,
505 |group, rows, gh, sums| {
506 let mut acc = vec![0i64; group.bins + SCRATCH_BINS];
507 for (rows, gh) in rows.chunks(row_limit).zip(gh.chunks(row_limit)) {
508 accumulate(group, rows, gh, &mut acc, |a, v| *a += v);
509 flush_packed(&mut acc, sums);
510 }
511 },
512 );
513}
514
515fn flush_packed(acc: &mut [i64], sums: &mut [f64]) {
518 for (value, pair) in acc.iter_mut().zip(sums.as_chunks_mut::<2>().0) {
519 let g = i64::from(*value as i32);
521 pair[0] += g as f64;
522 pair[1] += ((*value - g) >> 32) as f64;
523 *value = 0;
524 }
525}
526
527fn histogram_float(
529 groups: &[GroupView<'_>],
530 grad: &[f32],
531 hess: &[f32],
532 rows: &[u32],
533 buffers: &mut HistBuffers,
534 out: &mut [f64],
535) {
536 ordered(
537 rows,
538 &mut buffers.float_gh,
539 |row| {
540 prefetch(grad.as_ptr().wrapping_add(row).cast());
541 prefetch(hess.as_ptr().wrapping_add(row).cast());
542 },
543 |row| [f64::from(grad[row]), f64::from(hess[row])],
544 );
545 let blocks = (&mut buffers.blocks, &mut buffers.tasks);
546 histogram_blocks(groups, rows, &buffers.float_gh, blocks, out, |group, rows, gh, sums| {
547 accumulate(group, rows, gh, sums.as_chunks_mut::<2>().0, |a, v| {
548 a[0] += v[0];
549 a[1] += v[1];
550 });
551 });
552}
553
554fn stable_partition(
557 rows: &mut [u32],
558 scratch: &mut [u32],
559 ahead: impl Fn(u32) + Sync,
560 goes_left: impl Fn(u32) -> bool + Sync,
561) -> usize {
562 let lefts: Vec<usize> = rows
563 .par_chunks(ROW_CHUNK)
564 .zip(scratch.par_chunks_mut(ROW_CHUNK))
565 .map(|(src, dst)| {
566 let (mut left, mut right) = (0, dst.len());
567 for (i, &row) in src.iter().enumerate() {
568 if let Some(&next) = src.get(i + PREFETCH_ROWS) {
569 ahead(next);
570 }
571 if goes_left(row) {
572 dst[left] = row;
573 left += 1;
574 } else {
575 right -= 1;
576 dst[right] = row;
577 }
578 }
579 dst[left..].reverse();
580 left
581 })
582 .collect();
583 let total_left: usize = lefts.iter().sum();
584 let (mut left_rest, mut right_rest) = rows.split_at_mut(total_left);
585 let mut targets = Vec::with_capacity(lefts.len());
586 for (block, &left) in lefts.iter().enumerate() {
587 let len = rows_len_of_block(block, ROW_CHUNK, scratch.len());
588 let (l, lr) = std::mem::take(&mut left_rest).split_at_mut(left);
589 let (r, rr) = std::mem::take(&mut right_rest).split_at_mut(len - left);
590 left_rest = lr;
591 right_rest = rr;
592 targets.push((l, r));
593 }
594 scratch.par_chunks(ROW_CHUNK).zip(targets).for_each(|(src, (l, r))| {
595 l.copy_from_slice(&src[..l.len()]);
596 r.copy_from_slice(&src[l.len()..]);
597 });
598 total_left
599}
600
601fn rows_len_of_block(block: usize, chunk: usize, total: usize) -> usize {
602 chunk.min(total - block * chunk)
603}
604
605struct FlatNode<'d> {
608 data: &'d [u8],
609 stride: usize,
610 column: usize,
611 left_bins: [u32; 8],
612 children: [u32; 2],
614}
615
616fn flatten<'d>(data: &'d BinnedDataset, tree: &DeviceTree) -> Result<(Vec<FlatNode<'d>>, usize)> {
619 let internal = tree.nodes.len();
620 let malformed = || SilvicError::device("malformed device tree");
621 let mut depth = vec![0usize; internal];
622 let mut max_depth = 0;
623 let mut nodes = Vec::with_capacity(internal + tree.leaf_values.len());
624 for (k, node) in tree.nodes.iter().enumerate() {
625 let group = data.groups.get(node.split.group as usize).ok_or_else(malformed)?;
626 let mut child = |c: i32| -> Result<u32> {
627 if c >= 0 {
628 let c = c as usize;
629 if c <= k || c >= internal {
630 return Err(malformed());
631 }
632 depth[c] = depth[k] + 1;
633 Ok(c as u32)
634 } else if (!c as usize) < tree.leaf_values.len() {
635 max_depth = max_depth.max(depth[k] + 1);
636 Ok((internal + !c as usize) as u32)
637 } else {
638 Err(malformed())
639 }
640 };
641 let children = [child(node.right)?, child(node.left)?];
642 nodes.push(FlatNode {
643 data: &group.data,
644 stride: group.row_stride as usize,
645 column: node.split.column as usize,
646 left_bins: node.split.left_bins,
647 children,
648 });
649 }
650 let (data0, stride0, column0) = (nodes[0].data, nodes[0].stride, nodes[0].column);
651 for leaf in 0..tree.leaf_values.len() {
652 let this = (internal + leaf) as u32;
653 nodes.push(FlatNode {
654 data: data0,
655 stride: stride0,
656 column: column0,
657 left_bins: [0; 8],
658 children: [this, this],
659 });
660 }
661 Ok((nodes, max_depth))
662}
663
664fn add_tree_to(data: &BinnedDataset, tree: &DeviceTree, scores: &mut [f32]) -> Result<()> {
667 if tree.nodes.is_empty() {
668 let value = tree.leaf_values[0];
669 scores.par_iter_mut().for_each(|s| *s += value);
670 return Ok(());
671 }
672 let (nodes, depth) = flatten(data, tree)?;
673 let internal = tree.nodes.len();
674 scores
675 .par_chunks_mut(ROW_CHUNK)
676 .enumerate()
677 .for_each(|(chunk, scores)| {
678 for (lanes, scores) in scores.chunks_mut(TRAVERSE_LANES).enumerate() {
679 let first = chunk * ROW_CHUNK + lanes * TRAVERSE_LANES;
680 let mut current = [0u32; TRAVERSE_LANES];
681 for _ in 0..depth {
682 for (lane, node) in current.iter_mut().enumerate().take(scores.len()) {
683 let n = &nodes[*node as usize];
684 let bin = n.data[(first + lane) * n.stride + n.column];
685 let left = (n.left_bins[usize::from(bin >> 5)] >> (bin & 31)) & 1;
686 *node = n.children[left as usize];
687 }
688 }
689 for (score, &node) in scores.iter_mut().zip(¤t) {
690 *score += tree.leaf_values[node as usize - internal];
691 }
692 }
693 });
694 Ok(())
695}
696
697impl Backend for CpuBackend<'_> {
698 fn name(&self) -> String {
699 let threads = install(&self.pool, rayon::current_num_threads);
700 format!("cpu ({threads} threads)")
701 }
702
703 fn set_thread_pool(&mut self, pool: Arc<ThreadPool>) {
704 self.pool = Some(pool);
705 }
706
707 fn load(&mut self, train: &BinnedDataset, valid: &[&BinnedDataset], objective: &ObjectiveSetup) -> Result<()> {
708 let mut datasets = vec![self.adopt(train)];
709 datasets.extend(valid.iter().map(|v| self.adopt(v)));
710 for data in &datasets {
711 if data.groups.len() != train.groups.len() || data.features.len() != train.features.len() {
712 return Err(SilvicError::data(
713 "validation set binned with a different layout than the training set",
714 ));
715 }
716 if let Some(group) = data
717 .groups
718 .iter()
719 .find(|g| g.data.len() != data.num_rows * g.row_stride as usize)
720 {
721 return Err(SilvicError::data(format!(
722 "group of {} bytes does not hold {} rows of {} bytes",
723 group.data.len(),
724 data.num_rows,
725 group.row_stride
726 )));
727 }
728 }
729 let rows = train.num_rows;
730 if rows > u32::MAX as usize {
731 return Err(SilvicError::unsupported(format!(
732 "{rows} rows on one rank; the limit is 2^32 - 1"
733 )));
734 }
735 self.layout = group_layout(train)?;
736 self.total_bins = train.total_bins;
737 self.scores = datasets.iter().map(|d| vec![0.0; d.num_rows]).collect();
738 self.datasets = datasets;
739 self.objective = Some(objective.clone());
740 self.grad = vec![0.0; rows];
741 self.hess = vec![0.0; rows];
742 self.packed = vec![0; rows];
743 self.partition = Vec::with_capacity(rows);
744 self.scratch = vec![0; rows];
745 Ok(())
746 }
747
748 fn init_scores(&mut self, dataset: usize, value: f64) -> Result<()> {
749 let scores = self
750 .scores
751 .get_mut(dataset)
752 .ok_or_else(|| SilvicError::device(format!("no dataset {dataset}")))?;
753 scores.fill(value as f32);
754 Ok(())
755 }
756
757 fn compute_gradients(&mut self) -> Result<GradientStats> {
758 let setup = self
759 .objective
760 .as_ref()
761 .ok_or_else(|| SilvicError::device("cpu backend used before load"))?;
762 let train = self.datasets[0].as_ref();
763 let (scores, grad, hess) = (&self.scores[0], &mut self.grad, &mut self.hess);
764 Ok(install(&self.pool, || {
765 objective::gradients(
766 setup,
767 scores,
768 &train.label,
769 train.weight.as_deref(),
770 train.query_boundaries.as_deref(),
771 grad,
772 hess,
773 );
774 let max_abs = |v: &[f32]| v.par_iter().map(|x| f64::from(x.abs())).reduce(|| 0.0, f64::max);
775 GradientStats {
776 max_abs_grad: max_abs(grad),
777 max_hess: max_abs(hess),
778 }
779 }))
780 }
781
782 fn quantize(&mut self, spec: &QuantSpec) -> Result<()> {
783 let row_offset = self.train()?.row_offset;
784 let (grad, hess) = (&self.grad, &self.hess);
785 let packed = &mut self.packed;
786 install(&self.pool, || {
787 packed
788 .par_chunks_mut(ROW_CHUNK)
789 .enumerate()
790 .for_each(|(chunk, packed)| {
791 let first = chunk * ROW_CHUNK;
792 for (offset, slot) in packed.iter_mut().enumerate() {
793 let row = first + offset;
794 *slot = quantize::quantize_row(spec, grad[row], hess[row], row_offset + row as u64);
795 }
796 })
797 });
798 self.max_quant = i64::from(spec.grad_max_q.max(spec.hess_max_q));
799 Ok(())
800 }
801
802 fn reset_partition(&mut self, bag: &BagSpec) -> Result<u32> {
803 let train = self
804 .datasets
805 .first()
806 .ok_or_else(|| SilvicError::device("cpu backend used before load"))?;
807 let (rows, row_offset) = (train.num_rows as u32, train.row_offset);
808 let partition = &mut self.partition;
809 partition.clear();
810 match *bag {
811 BagSpec::All => install(&self.pool, || partition.par_extend(0..rows)),
812 BagSpec::Bernoulli { .. } if self.bag.as_ref().is_some_and(|(spec, _)| spec == bag) => {
814 partition.extend_from_slice(&self.bag.as_ref().expect("cached bag").1);
815 }
816 BagSpec::Bernoulli { seed, round, fraction } => {
817 install(&self.pool, || {
818 partition.par_extend(
819 (0..rows)
820 .into_par_iter()
821 .filter(|&row| quantize::in_bag(seed, round, row_offset + u64::from(row), fraction)),
822 )
823 });
824 let mut cached = self.bag.take().map(|(_, rows)| rows).unwrap_or_default();
825 cached.clone_from(partition);
826 self.bag = Some((*bag, cached));
827 }
828 }
829 Ok(self.partition.len() as u32)
830 }
831
832 fn set_histogram_features(&mut self, used: &[bool]) {
833 for group in &mut self.layout {
834 group.activate(Some(used));
835 }
836 }
837
838 fn histogram(&mut self, leaf: LeafRange, quantized: bool, out: &mut [f64]) -> Result<()> {
839 self.check_range(leaf)?;
840 if out.len() != 2 * self.total_bins {
841 return Err(SilvicError::device(format!(
842 "histogram buffer of {} values, expected {}",
843 out.len(),
844 2 * self.total_bins
845 )));
846 }
847 let train = self
848 .datasets
849 .first()
850 .ok_or_else(|| SilvicError::device("cpu backend used before load"))?;
851 let groups = group_views(&self.layout, train);
852 let rows = &self.partition[leaf.begin as usize..(leaf.begin + leaf.count) as usize];
853 let buffers = &mut self.hist;
854 if quantized {
855 let (packed, max_quant) = (&self.packed, self.max_quant);
856 install(&self.pool, || {
857 histogram_quantized(&groups, packed, max_quant, rows, buffers, out)
858 });
859 } else {
860 let (grad, hess) = (&self.grad, &self.hess);
861 install(&self.pool, || histogram_float(&groups, grad, hess, rows, buffers, out));
862 }
863 Ok(())
864 }
865
866 fn partition(&mut self, leaf: LeafRange, split: &BinSplit) -> Result<u32> {
867 self.check_range(leaf)?;
868 let train = self
869 .datasets
870 .first()
871 .ok_or_else(|| SilvicError::device("cpu backend used before load"))?;
872 let group = train
873 .groups
874 .get(split.group as usize)
875 .ok_or_else(|| SilvicError::device(format!("split on missing group {}", split.group)))?;
876 let (data, stride, column) = (&group.data, group.row_stride as usize, split.column as usize);
877 let goes_left = |row: u32| split.goes_left(data[row as usize * stride + column]);
878 let ahead = |row: u32| prefetch(data.as_ptr().wrapping_add(row as usize * stride + column));
879 let range = leaf.begin as usize..(leaf.begin + leaf.count) as usize;
880 let rows = &mut self.partition[range];
881 let scratch = &mut self.scratch[..rows.len()];
882 let left = install(&self.pool, || stable_partition(rows, scratch, ahead, goes_left));
883 Ok(left as u32)
884 }
885
886 fn leaf_sums(&mut self, leaves: &[LeafRange], grad_scale: f32, hess_scale: f32) -> Result<Vec<[i64; 2]>> {
887 for &leaf in leaves {
888 self.check_range(leaf)?;
889 }
890 let (grad, hess, partition) = (&self.grad, &self.hess, &self.partition);
891 Ok(install(&self.pool, || {
892 leaves
893 .par_iter()
894 .map(|leaf| {
895 partition[leaf.begin as usize..(leaf.begin + leaf.count) as usize]
896 .par_iter()
897 .map(|&row| {
898 let row = row as usize;
899 [
900 quantize::to_fixed(grad[row], grad_scale),
901 quantize::to_fixed(hess[row], hess_scale),
902 ]
903 })
904 .reduce(|| [0, 0], |a, b| [a[0] + b[0], a[1] + b[1]])
905 })
906 .collect()
907 }))
908 }
909
910 fn add_tree(&mut self, tree: &DeviceTree) -> Result<()> {
911 if tree.leaf_values.is_empty() || tree.leaf_values.len() != tree.nodes.len() + 1 {
912 return Err(SilvicError::device(format!(
913 "tree with {} nodes and {} leaves",
914 tree.nodes.len(),
915 tree.leaf_values.len()
916 )));
917 }
918 let datasets = &self.datasets;
919 let scores = &mut self.scores;
920 install(&self.pool, || {
922 datasets
923 .iter()
924 .zip(scores.iter_mut())
925 .try_for_each(|(data, scores)| add_tree_to(data, tree, scores))
926 })
927 }
928
929 fn scores(&mut self, dataset: usize) -> Result<Vec<f32>> {
930 self.scores
931 .get(dataset)
932 .cloned()
933 .ok_or_else(|| SilvicError::device(format!("no dataset {dataset}")))
934 }
935}
936
937fn install<R: Send>(pool: &Option<Arc<ThreadPool>>, op: impl FnOnce() -> R + Send) -> R {
939 match pool {
940 Some(pool) => pool.install(op),
941 None => op(),
942 }
943}
944
945#[cfg(test)]
946mod tests {
947 use super::*;
948 use crate::data::{BinMapper, FeatureGroup, FeatureMeta};
949
950 fn dataset(rows: usize) -> BinnedDataset {
952 let bins = [7u32, 3, 5];
953 let bin = |row: usize, f: usize| (quantize::mix32((row * 3 + f) as u32) % bins[f]) as u8;
954 let meta = |f: usize, group: u32, column: u32, hist_offset: u32| FeatureMeta {
955 name: format!("f{f}"),
956 bins: BinMapper::Categorical {
957 categories: (1..bins[f] as i32).collect(),
958 },
959 num_bins: bins[f],
960 group,
961 column,
962 hist_offset,
963 };
964 BinnedDataset {
965 num_rows: rows,
966 row_offset: 100,
967 features: vec![meta(0, 0, 0, 0), meta(1, 0, 1, 7), meta(2, 1, 0, 10)],
968 groups: vec![
969 FeatureGroup {
970 features: vec![0, 1],
971 row_stride: 4,
972 hist_offset: 0,
973 total_bins: 10,
974 data: (0..rows).flat_map(|r| [bin(r, 0), bin(r, 1), 0, 0]).collect(),
975 },
976 FeatureGroup {
977 features: vec![2],
978 row_stride: 4,
979 hist_offset: 10,
980 total_bins: 5,
981 data: (0..rows).flat_map(|r| [bin(r, 2), 0, 0, 0]).collect(),
982 },
983 ],
984 total_bins: 15,
985 label: vec![0.0; rows],
986 weight: None,
987 query_boundaries: None,
988 }
989 }
990
991 fn loaded(data: &BinnedDataset) -> CpuBackend<'_> {
992 let mut backend = CpuBackend::borrowing(data, &[]);
993 backend.load(data, &[], &ObjectiveSetup::Regression).expect("load");
994 let grad: Vec<f32> = (0..data.num_rows).map(|r| (r % 13) as f32 / 4.0 - 1.5).collect();
995 let hess: Vec<f32> = (0..data.num_rows).map(|r| 0.25 + (r % 5) as f32 / 8.0).collect();
996 backend.set_gradients(&grad, &hess);
997 backend
998 }
999
1000 fn naive(data: &BinnedDataset, rows: &[u32], value: impl Fn(usize) -> (f64, f64)) -> Vec<f64> {
1002 let mut out = vec![0.0; 2 * data.total_bins];
1003 for &row in rows {
1004 let (g, h) = value(row as usize);
1005 for f in 0..data.features.len() {
1006 let index = data.features[f].hist_offset as usize + usize::from(data.bin(f, row as usize));
1007 out[2 * index] += g;
1008 out[2 * index + 1] += h;
1009 }
1010 }
1011 out
1012 }
1013
1014 #[test]
1015 fn histograms_match_direct_sums() {
1016 let data = dataset(50_000);
1017 let mut backend = loaded(&data);
1018 let spec = QuantSpec::new(1.5, 0.75, None, 16, true, 9, 3);
1019 backend.quantize(&spec).expect("quantize");
1020 backend
1021 .reset_partition(&BagSpec::Bernoulli {
1022 seed: 4,
1023 round: 1,
1024 fraction: 0.7,
1025 })
1026 .expect("bag");
1027 let leaf = LeafRange {
1028 begin: 1000,
1029 count: 30_000,
1030 };
1031 let rows = backend.partition[1000..31_000].to_vec();
1032 let mut out = vec![0.0; 30];
1033 backend.histogram(leaf, true, &mut out).expect("quantized");
1034 let packed = backend.packed.clone();
1035 let expected = naive(&data, &rows, |r| {
1036 let (g, h) = quantize::unpack(packed[r]);
1037 (f64::from(g), f64::from(h))
1038 });
1039 assert_eq!(out, expected, "quantized");
1040 for used in [[true, false, true], [false, true, false], [false, false, true]] {
1042 backend.set_histogram_features(&used);
1043 backend.histogram(leaf, true, &mut out).expect("masked");
1044 for (f, meta) in data.features.iter().enumerate().filter(|(f, _)| used[*f]) {
1045 let bins = 2 * meta.hist_offset as usize..2 * (meta.hist_offset + meta.num_bins) as usize;
1046 assert_eq!(out[bins.clone()], expected[bins], "mask {used:?}: feature {f}");
1047 }
1048 }
1049 backend.set_histogram_features(&[true; 3]);
1050 backend.histogram(leaf, false, &mut out).expect("float");
1051 let (grad, hess) = (backend.grad.clone(), backend.hess.clone());
1052 let expected = naive(&data, &rows, |r| (f64::from(grad[r]), f64::from(hess[r])));
1053 for (bin, (a, b)) in out.iter().zip(&expected).enumerate() {
1054 assert!((a - b).abs() < 1e-9 * b.abs().max(1.0), "float bin {bin}: {a} vs {b}");
1055 }
1056 }
1057
1058 #[test]
1059 fn packed_accumulators_flush_before_overflow() {
1060 let rows_count = 400_000;
1061 let data = dataset(rows_count);
1062 let backend = loaded(&data);
1063 let rows: Vec<u32> = (0..rows_count as u32).collect();
1064 let packed = vec![3 * 65_536 - 32_000; rows_count];
1066 let mut out = vec![0.0; 30];
1067 let groups = group_views(&backend.layout, &data);
1068 let pool = rayon::ThreadPoolBuilder::new().num_threads(1).build().expect("pool");
1069 let mut buffers = HistBuffers::default();
1071 pool.install(|| {
1072 histogram_quantized(
1073 &groups,
1074 &packed,
1075 i64::from(i32::MAX) / 1000,
1076 &rows,
1077 &mut buffers,
1078 &mut out,
1079 )
1080 });
1081 let expected = naive(&data, &rows, |r| {
1082 let (g, h) = quantize::unpack(packed[r]);
1083 (f64::from(g), f64::from(h))
1084 });
1085 assert_eq!(out, expected);
1086 assert!(
1087 out.iter().step_by(2).any(|&g| g < f64::from(i32::MIN)),
1088 "sums leave i32"
1089 );
1090 }
1091
1092 #[test]
1093 fn partition_is_stable() {
1094 let data = dataset(30_000);
1095 let mut backend = loaded(&data);
1096 backend.reset_partition(&BagSpec::All).expect("all");
1097 let mut left_bins = [0u32; 8];
1098 left_bins[0] = 0b10110;
1099 let split = BinSplit {
1100 group: 0,
1101 column: 0,
1102 left_bins,
1103 };
1104 let leaf = LeafRange {
1105 begin: 500,
1106 count: 25_000,
1107 };
1108 let left = backend.partition(leaf, &split).expect("partition") as usize;
1109 let rows = &backend.partition[500..25_500];
1110 let goes_left = |row: u32| split.goes_left(data.bin(0, row as usize));
1111 let expected_left: Vec<u32> = (500..25_500).filter(|&r| goes_left(r)).collect();
1112 let expected_right: Vec<u32> = (500..25_500).filter(|&r| !goes_left(r)).collect();
1113 assert_eq!(left, expected_left.len());
1114 assert_eq!(&rows[..left], &expected_left[..]);
1115 assert_eq!(&rows[left..], &expected_right[..]);
1116 assert_eq!(
1117 &backend.partition[..500],
1118 &(0..500).collect::<Vec<u32>>()[..],
1119 "rows before the leaf"
1120 );
1121 }
1122
1123 #[test]
1124 fn add_tree_and_leaf_sums() {
1125 let data = dataset(10_000);
1126 let mut backend = loaded(&data);
1127 backend.reset_partition(&BagSpec::All).expect("all");
1128 let mut left_bins = [0u32; 8];
1129 left_bins[0] = 0b11;
1130 let split = BinSplit {
1131 group: 1,
1132 column: 0,
1133 left_bins,
1134 };
1135 let left = backend
1136 .partition(
1137 LeafRange {
1138 begin: 0,
1139 count: 10_000,
1140 },
1141 &split,
1142 )
1143 .expect("partition");
1144 let tree = DeviceTree {
1145 nodes: vec![crate::backend::DeviceNode {
1146 split,
1147 left: !0,
1148 right: !1,
1149 }],
1150 leaf_values: vec![0.5, -0.25],
1151 };
1152 backend.init_scores(0, 1.0).expect("init");
1153 backend.add_tree(&tree).expect("add");
1154 let scores = backend.scores(0).expect("scores");
1155 for (row, score) in scores.iter().enumerate() {
1156 let expected = if data.bin(2, row) < 2 { 1.5 } else { 0.75 };
1157 assert_eq!(*score, expected, "row {row}");
1158 }
1159 let leaves = [
1160 LeafRange { begin: 0, count: left },
1161 LeafRange {
1162 begin: left,
1163 count: 10_000 - left,
1164 },
1165 ];
1166 let sums = backend.leaf_sums(&leaves, 1024.0, 256.0).expect("sums");
1167 for (leaf, range) in leaves.iter().enumerate() {
1168 let rows = &backend.partition[range.begin as usize..(range.begin + range.count) as usize];
1169 let g: i64 = rows
1170 .iter()
1171 .map(|&r| quantize::to_fixed(backend.grad[r as usize], 1024.0))
1172 .sum();
1173 let h: i64 = rows
1174 .iter()
1175 .map(|&r| quantize::to_fixed(backend.hess[r as usize], 256.0))
1176 .sum();
1177 assert_eq!(sums[leaf], [g, h], "leaf {leaf}");
1178 }
1179 }
1180}