Skip to main content

silvic_core/backend/
cpu.rs

1//! The CPU backend: every device operation on host memory, parallel over rayon's pool. It is the CPU training path
2//! and the reference the GPU backends are tested against.
3//!
4//! Histograms follow LightGBM's row-wise construction with "ordered gradients": a work item is one feature group
5//! over one block of the leaf's rows; it gathers the rows' gradients into a contiguous buffer, then adds them at
6//! each row's bins into an accumulator of the group's bins (at most `GROUP_BIN_BUDGET`, so it stays in L1). Blocks
7//! are sized from the leaf's row count and the group's shape only, and each group's blocks are merged in block
8//! order, so float histograms are deterministic whatever the thread count. Quantized histograms add each row's
9//! packed `(h_q << 32) + g_q` in one i64 add per feature and unpack before the gradient half could overflow, so
10//! they are exact integers. As LightGBM, histograms skip the features the learner will not evaluate
11//! (`set_histogram_features`). Rows of small leaves, spread thinly over the dataset, are prefetched ahead.
12
13use 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
25/// The fewest adds per accumulator bin a histogram block does, so zeroing and merging its accumulator stay cheap.
26const MIN_ADDS_PER_BIN: usize = 16;
27/// The most row blocks per group of a histogram: enough to balance the cores on a large leaf. A constant, so float
28/// sums do not depend on the thread count.
29const MAX_BLOCKS_PER_GROUP: usize = 64;
30/// Accumulator slots past a group's bins, so a bin byte beyond its feature's bins never indexes past the accumulator.
31const SCRATCH_BINS: usize = 256;
32/// How many rows ahead the scattered-row loops prefetch, enough to cover a DRAM miss.
33const PREFETCH_ROWS: usize = 16;
34/// Rows whose ids spread over more than this many dataset rows per row are prefetched by hand.
35const SCATTERED_SPAN: usize = 4;
36/// Bins per work item when merging histogram blocks.
37const MERGE_CHUNK_BINS: usize = 1024;
38/// A group's histogram adds its whole row when at least `DENSE_RUN_NUM / DENSE_RUN_DEN` of its columns are used.
39const DENSE_RUN_NUM: usize = 3;
40const DENSE_RUN_DEN: usize = 4;
41/// Rows per work item of the partition, gradient, quantization and score passes.
42const ROW_CHUNK: usize = 8192;
43/// Rows `add_tree` walks down the tree together.
44const TRAVERSE_LANES: usize = 8;
45
46/// One feature group as the hot loops read it.
47#[derive(Debug, Clone)]
48struct GroupLayout {
49    group: usize,
50    stride: usize,
51    /// Index of the group's bin 0 in the flat histogram.
52    hist_offset: usize,
53    bins: usize,
54    /// Per column up to the last used one: its feature, if any.
55    column_features: Vec<Option<usize>>,
56    /// Per column: its feature's first bin within the group (`bins`, the scratch slots, for a column no feature
57    /// uses).
58    offsets: Vec<usize>,
59    /// Runs `start..end` of consecutive columns whose features histograms currently need.
60    runs: Vec<(u32, u32)>,
61    /// Features in `runs`.
62    adds: usize,
63}
64
65impl GroupLayout {
66    /// Keeps the columns of the features with `used[feature]` (all of them with `None`) as runs, so a row's bins
67    /// are read run by run, several bytes per load. When most columns are used, one run over the whole row is
68    /// cheaper than the gaps save: adding the bins of an unused feature is harmless.
69    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/// The reference backend. [`CpuBackend::borrowing`] reads the datasets in place; [`CpuBackend::new`] copies them on
102/// `load`.
103#[derive(Debug, Default)]
104pub struct CpuBackend<'a> {
105    pool: Option<Arc<ThreadPool>>,
106    borrowed: Vec<&'a BinnedDataset>,
107    /// Training set first, then validation sets.
108    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(grad_max_q, hess_max_q)` of the last quantization: bounds rows per packed accumulator.
117    max_quant: i64,
118    partition: Vec<u32>,
119    scratch: Vec<u32>,
120    hist: HistBuffers,
121    /// The last Bernoulli bag and its rows.
122    bag: Option<(BagSpec, Vec<u32>)>,
123}
124
125impl<'a> CpuBackend<'a> {
126    /// A backend that copies the datasets it is loaded with.
127    pub fn new() -> CpuBackend<'a> {
128        CpuBackend::default()
129    }
130
131    /// A backend that reads `train` and `valid` in place when `load` gets these same datasets (by address), so the
132    /// bins are not copied; other datasets are copied.
133    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    /// Overwrites the training gradients, for tests that grow trees without an objective.
169    #[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
176/// The hot-loop layout of each group, checking the invariants `accumulate` relies on: every feature's bins lie
177/// inside its group's and every column inside the row.
178fn 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
225/// One group's rows and layout, borrowed for a histogram pass.
226struct GroupView<'d> {
227    data: &'d [u8],
228    stride: usize,
229    hist_offset: usize,
230    bins: usize,
231    offsets: &'d [usize],
232    /// Column runs to add.
233    runs: &'d [(u32, u32)],
234    /// Features to add per row.
235    adds: usize,
236    /// One past the last column read: the bytes of a row to prefetch.
237    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/// Adds `values[i]` at every feature bin of row `rows[i]` of `group`; `acc` holds the group's bins and
257/// `SCRATCH_BINS` more. Rows spread thinly over the dataset are prefetched ahead; dense runs (the root, large
258/// leaves) are left to the hardware prefetcher.
259#[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        // SAFETY: `group_layout` keeps every offset at most `group.bins` and a bin is below 256, so the index is
310        // below `group.bins + SCRATCH_BINS <= acc.len()` (asserted by `accumulate`). This is the histogram's
311        // innermost loop; the bounds check costs a sizeable share of it.
312        add(unsafe { acc.get_unchecked_mut(offset + usize::from(bin)) }, value);
313    }
314}
315
316/// Gathers `value(row)` of each row into `out`, calling `ahead(row)` (prefetches) for a row further on: a leaf's
317/// rows are scattered over the dataset.
318#[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/// A hint to load the cache line at `address`; never faults, whatever the address.
329#[inline(always)]
330fn prefetch(address: *const u8) {
331    // SAFETY: prefetch instructions only hint the cache; they neither fault nor read into a register, for any
332    // address.
333    #[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
345/// Rows per block of `group` for a leaf of `rows` rows: at least `MIN_ADDS_PER_BIN` adds per bin, at most
346/// `MAX_BLOCKS_PER_GROUP` blocks.
347fn 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/// One (group, row block) work item of a histogram pass.
354#[derive(Debug, Clone, Copy)]
355struct BlockTask {
356    group: usize,
357    start: usize,
358    end: usize,
359}
360
361/// Buffers the histogram passes reuse from call to call.
362#[derive(Debug, Default)]
363struct HistBuffers {
364    /// Each block's sums: `2 * (bins + SCRATCH_BINS)` values per block, `[g, h]` per bin of its group.
365    blocks: Vec<f64>,
366    packed_gh: Vec<i64>,
367    float_gh: Vec<[f64; 2]>,
368    tasks: Vec<BlockTask>,
369}
370
371/// Runs `block(group, rows, values, sums)` over every (group, row block) in parallel, then writes each group's
372/// block sums, added in block order, into its bins of `out`. `values[i]` belongs to `rows[i]`; `sums` (the group's
373/// bins plus `SCRATCH_BINS`) is zero on entry.
374fn 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
417/// Writes each group's bins of `out` as the sum of its blocks in `sums`, in block order, in parallel over bin
418/// ranges; bins outside every group, and of groups without blocks, are zeroed.
419fn merge_blocks(groups: &[GroupView<'_>], tasks: &[BlockTask], sums: &[f64], out: &mut [f64]) {
420    // Offsets of each group's first block in `sums`, and its block count.
421    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
465/// LightGBM's "ordered gradients": each row's value, in the leaf's row order, gathered once for every group.
466fn 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
478/// Quantized histogram of `rows`: exact integer `[g, h]` sums.
479fn 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    // Rows per packed accumulator before the gradient half could leave i32.
488    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
515/// Adds packed `(H << 32) + G` sums into the `[g, h]` pairs of `sums` (exact: integers far below 2^53) and zeroes
516/// the accumulator.
517fn flush_packed(acc: &mut [i64], sums: &mut [f64]) {
518    for (value, pair) in acc.iter_mut().zip(sums.as_chunks_mut::<2>().0) {
519        // G stays inside i32 by the row bound, so the low half is G and the rest is H.
520        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
527/// Float histogram of `rows`: f64 sums of the f32 gradients and hessians.
528fn 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
554/// Stable partition of `rows` by `goes_left`, through `scratch`; returns the left count. `ahead(row)` runs
555/// (prefetches) for rows further on.
556fn 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
605/// A node of the branch-free traversal: leaves are nodes whose children are themselves, so every row takes the same
606/// number of steps.
607struct FlatNode<'d> {
608    data: &'d [u8],
609    stride: usize,
610    column: usize,
611    left_bins: [u32; 8],
612    /// `[right, left]`.
613    children: [u32; 2],
614}
615
616/// The tree as flat nodes (internal nodes, then one per leaf) and its depth; checks that children come after their
617/// parent and every index is in range.
618fn 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
664/// Adds `tree` to `scores` by traversing the rows' bins, `TRAVERSE_LANES` rows at a time: their node-to-node
665/// chains of dependent loads are independent, so they overlap, and no step branches on the data.
666fn 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(&current) {
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            // A bag holds for `bagging_freq` iterations; drawing it once saves a hash per row in the others.
813            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        // Datasets share the training set's groups (checked at load), so the first one validates the tree for all.
921        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
937/// Runs `op` on `pool` when one is set, else on the current (global) pool.
938fn 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    /// Two groups: features 0-1 (bins 7, 3) and feature 2 (bins 5), rows from a hash.
951    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    /// Direct per-bin sums of `value(row)` over `rows`.
1001    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        // With a feature mask, the bins of the used features stay exact; the others may hold anything.
1041        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        // g_q = -32000, h_q = 3 on every row: one accumulator over all rows would take the gradient half past i32.
1065        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        // A bound of 1000 rows per packed accumulator forces flushes inside every block.
1070        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}