use std::time::{Duration, Instant};
use cubecl::prelude::*;
use rayon::prelude::*;
use crate::errors::EdgeErrors;
use crate::gpu::nebula_gpu::{
GpuGene, GpuSolveParams, PendingSolve, PmlReply, PmlRequest, ResidentBatch, SolveTiming,
};
use crate::gpu::pml_kernel::F32_NOISE_SCALE;
use crate::numeric::gamma::ln_gamma;
use crate::prelude::*;
use crate::sc::nebula::{
GeneOutcome, GenePlan, InnerFit, NebulaFit, NebulaParams, Shared, StageTwoSearch, finish_gene,
gene_pml, nebula_sparse_with, plan_gene, variance_bounds, variance_objective,
};
use crate::sc::pml::{DeviceCurvature, PmlParams, PmlVariance, newton_finish, opt_pml_from};
use crate::sc::ptmg::{GeneCounts, positive_indices};
const TIMING_ENV: &str = "EDGE_RS_GPU_TIMING";
const COHORTS: usize = 2;
const PROBE_ROUNDS: usize = 8;
const MERGE_BELOW: usize = 64;
const STEP_BINS: usize = 12;
#[derive(Default)]
struct Timing {
stage_one: Duration,
upload: Duration,
gather: Duration,
solve: Duration,
finish: Duration,
tell: Duration,
stage_three: Duration,
requests: Vec<usize>,
device_steps: [usize; STEP_BINS],
finish_steps: [usize; STEP_BINS],
}
impl Timing {
fn report(&self, solve: Option<SolveTiming>) {
let secs = |d: Duration| d.as_secs_f64();
let mut sorted = self.requests.clone();
sorted.sort_unstable();
let at = |f: f64| {
sorted
.get(((sorted.len() as f64 - 1.0) * f) as usize)
.copied()
.unwrap_or(0)
};
eprintln!(
"gpu nebula: stage one {:.2} s, upload {:.2} s, gather {:.2} s, solve {:.2} s, finish {:.2} s, tell {:.2} s, stage three {:.2} s",
secs(self.stage_one),
secs(self.upload),
secs(self.gather),
secs(self.solve),
secs(self.finish),
secs(self.tell),
secs(self.stage_three),
);
if let Some(t) = solve {
eprintln!(
" solve: staging {:.2} s, blocked on the device {:.2} s, scatter {:.2} s",
secs(t.staging),
secs(t.blocked),
secs(t.scatter),
);
}
eprintln!(
" rounds {}, requests {}, per round min {} / median {} / max {}",
sorted.len(),
sorted.iter().sum::<usize>(),
at(0.0),
at(0.5),
at(1.0),
);
eprintln!(
" device newton steps (last bin is {}+): {:?}",
STEP_BINS - 1,
self.device_steps
);
eprintln!(
" finish newton steps (last bin is {}+): {:?}",
STEP_BINS - 1,
self.finish_steps
);
}
}
pub fn nebula_sparse_gpu<T: EdgeFloat, R: Runtime>(
counts: &CompressedSparse<f64>,
subject_id: &[usize],
design: &[T],
n_coef: usize,
offset: Option<&[T]>,
params: Option<NebulaParams>,
client: &ComputeClient<R>,
) -> Result<NebulaFit, EdgeErrors> {
if params.is_some_and(|p| p.reml) {
return Err(EdgeErrors::InvalidArgument(
"The GPU NEBULA path does not implement `reml`; use the CPU path.".to_string(),
));
}
nebula_sparse_with(
counts,
subject_id,
design,
n_coef,
offset,
params,
|shared, sparse, totals, kept| fit_all(shared, sparse, totals, kept, client),
)
}
fn fit_all<R: Runtime>(
shared: &Shared<'_>,
sparse: &CompressedSparse<f64>,
totals: &[f64],
kept: &[usize],
client: &ComputeClient<R>,
) -> Result<Vec<GeneOutcome>, EdgeErrors> {
let k = shared.n_subjects;
let mut timing = std::env::var_os(TIMING_ENV).map(|_| Timing::default());
let started = Instant::now();
let genes: Vec<(GeneCounts, GenePlan)> = kept
.par_iter()
.map(|&g| {
let counts = positive_indices(sparse, g)?;
let plan = plan_gene(shared, &counts, &totals[g * k..(g + 1) * k])?;
Ok((counts, plan))
})
.collect::<Result<_, EdgeErrors>>()?;
if let Some(t) = timing.as_mut() {
t.stage_one = started.elapsed();
}
let (refits, solve_timing) = search_all(shared, &genes, totals, kept, client, timing.as_mut())?;
let started = Instant::now();
let outcomes = genes
.into_par_iter()
.zip(refits)
.zip(kept.par_iter())
.map(|(((counts, plan), refit), &g)| {
finish_gene(shared, &counts, &totals[g * k..(g + 1) * k], plan, refit)
})
.collect();
if let Some(t) = timing.as_mut() {
t.stage_three = started.elapsed();
t.report(solve_timing);
}
outcomes
}
struct Live<'a> {
index: usize,
resident: usize,
search: StageTwoSearch,
fixed_cell: Option<f64>,
pml: crate::sc::pml::PmlData<'a>,
tail: Vec<(f64, f64)>,
warm: Option<(Vec<f64>, Vec<f64>)>,
}
#[allow(clippy::type_complexity)]
fn search_all<R: Runtime>(
shared: &Shared<'_>,
genes: &[(GeneCounts, GenePlan)],
totals: &[f64],
kept: &[usize],
client: &ComputeClient<R>,
mut timing: Option<&mut Timing>,
) -> Result<(Vec<Option<Option<Vec<f64>>>>, Option<SolveTiming>), EdgeErrors> {
let k = shared.n_subjects;
let mut refits: Vec<Option<Option<Vec<f64>>>> = vec![None; genes.len()];
let mut live: Vec<Live<'_>> = Vec::new();
for (index, (counts, plan)) in genes.iter().enumerate() {
let Some((start, fixed_cell)) = plan.search() else {
continue;
};
let (lower, upper) = variance_bounds(&shared.params, fixed_cell);
let g = kept[index];
live.push(Live {
index,
resident: live.len(),
search: StageTwoSearch::new(&start, &lower, &upper, plan.ord),
fixed_cell,
pml: gene_pml(shared, counts, &totals[g * k..(g + 1) * k]),
tail: count_histogram(&counts.counts),
warm: None,
});
}
if live.is_empty() {
return Ok((refits, None));
}
let resident_genes: Vec<GpuGene<'_>> = live
.iter()
.map(|l| GpuGene {
counts: l.pml.counts,
cell_index: l.pml.cell_index,
subject_total: l.pml.subject_total,
beta_init: &[],
sigma: 0.0,
gamma: 0.0,
})
.collect();
let started = Instant::now();
let mut resident = ResidentBatch::upload(
shared.design,
shared.log_offset,
shared.fid,
&resident_genes,
client,
)?;
if let Some(t) = timing.as_deref_mut() {
t.upload = started.elapsed();
resident.time_solves();
}
let inner = PmlParams::default();
let solve_params = GpuSolveParams {
eps: shared.params.eps,
noise_scale: f64::from(F32_NOISE_SCALE),
max_iter: inner.max_iter as u32,
max_backtrack: inner.max_backtrack as u32,
full: true,
information: false,
};
let mut split = false;
let mut probed = 0usize;
let mut probe_blocked = Duration::ZERO;
let mut probe_finish = Duration::ZERO;
let mut in_flight = vec![false; live.len()];
let mut flights: [Option<Flight<'_>>; COHORTS] = std::array::from_fn(|_| None);
for (cohort, flight) in flights.iter_mut().enumerate() {
*flight = launch_cohort(
shared,
genes,
&live,
&mut in_flight,
cohort,
split,
&mut resident,
&solve_params,
client,
timing.as_deref_mut(),
)?;
}
let mut cohort = 0;
while flights.iter().any(Option::is_some) || live.iter().any(|l| l.search.ask().is_some()) {
if let Some(flight) = flights[cohort].take() {
let started = Instant::now();
let replies = resident.collect(flight.pending)?;
let solved = Instant::now();
let assembled: Vec<(f64, usize)> = flight
.routes
.par_iter()
.zip(replies.par_iter())
.map_init(Vec::new, |stored, (&(i, _, subject, cell), reply)| {
let l = &live[i];
let (counts, plan) = &genes[l.index];
let objective = variance_objective(
shared,
&l.pml,
&plan.beta_start,
counts,
l.search.order(),
l.fixed_cell,
);
let Some((fit, steps)) =
finish_at_argmax(&l.pml, reply, subject, cell, objective.params, stored)
else {
return (f64::INFINITY, 0);
};
(
objective.assemble(subject, cell, &fit, histogram_tail(&l.tail, cell)),
steps,
)
})
.collect();
let finished = Instant::now();
if probed <= PROBE_ROUNDS {
if probed > 0 {
probe_blocked += solved - started;
probe_finish += finished - solved;
}
split = probed == PROBE_ROUNDS && probe_finish > probe_blocked;
probed += 1;
}
let mut values = flight.values;
for ((&(i, p, _, _), &(v, _)), reply) in
flight.routes.iter().zip(&assembled).zip(&replies)
{
let slot = values
.binary_search_by_key(&i, |(index, _)| *index)
.expect("every route belongs to a search of its own flight");
values[slot].1[p] = v;
if v.is_finite() && live[i].warm.is_none() {
live[i].warm = Some((reply.beta.clone(), reply.log_w.clone()));
}
}
for (i, v) in values {
live[i].search.tell(&v);
in_flight[i] = false;
}
if let Some(t) = timing.as_deref_mut() {
t.solve += solved - started;
t.finish += finished - solved;
t.tell += finished.elapsed();
for (reply, &(_, steps)) in replies.iter().zip(&assembled) {
t.device_steps[(reply.iterations as usize).min(STEP_BINS - 1)] += 1;
t.finish_steps[steps.min(STEP_BINS - 1)] += 1;
}
}
}
flights[cohort] = launch_cohort(
shared,
genes,
&live,
&mut in_flight,
cohort,
split,
&mut resident,
&solve_params,
client,
timing.as_deref_mut(),
)?;
cohort = (cohort + 1) % COHORTS;
}
let solve_timing = resident.timing();
for l in live {
refits[l.index] = Some(l.search.result());
}
Ok((refits, solve_timing))
}
struct Flight<'a> {
pending: PendingSolve<'a>,
routes: Vec<(usize, usize, f64, f64)>,
values: Vec<(usize, Vec<f64>)>,
}
#[allow(clippy::too_many_arguments)]
fn launch_cohort<'a, R: Runtime>(
shared: &Shared<'_>,
genes: &[(GeneCounts, GenePlan)],
live: &[Live<'_>],
in_flight: &mut [bool],
cohort: usize,
split: bool,
resident: &mut ResidentBatch<R>,
solve_params: &GpuSolveParams,
client: &'a ComputeClient<R>,
timing: Option<&mut Timing>,
) -> Result<Option<Flight<'a>>, EdgeErrors> {
let started = Instant::now();
let running = live.iter().filter(|l| l.search.ask().is_some()).count();
let merged = !split || running < MERGE_BELOW;
let mut requests = Vec::new();
let mut routes = Vec::new();
let mut values = Vec::new();
for (i, l) in live.iter().enumerate() {
let mine = if merged {
cohort == 0
} else {
i % COHORTS == cohort
};
if in_flight[i] || !mine {
continue;
}
let Some(points) = l.search.ask() else {
continue;
};
in_flight[i] = true;
values.push((i, vec![f64::INFINITY; points.len()]));
let (counts, plan) = &genes[l.index];
let objective = variance_objective(
shared,
&l.pml,
&plan.beta_start,
counts,
l.search.order(),
l.fixed_cell,
);
for (p, x) in points.iter().enumerate() {
if let Some((subject, cell, beta_init)) = objective.request(x) {
let (beta_init, log_w_init) = match &l.warm {
Some((beta, log_w)) => (beta.clone(), log_w.clone()),
None => (beta_init, Vec::new()),
};
requests.push(PmlRequest {
gene: l.resident,
subject,
cell,
beta_init,
log_w_init,
});
routes.push((i, p, subject, cell));
}
}
}
if values.is_empty() {
return Ok(None);
}
let gathered = Instant::now();
let pending = resident.submit(&requests, solve_params, client)?;
if let Some(t) = timing {
t.gather += gathered - started;
t.solve += gathered.elapsed();
t.requests.push(requests.len());
}
Ok(Some(Flight {
pending,
routes,
values,
}))
}
fn finish_at_argmax(
pml: &crate::sc::pml::PmlData<'_>,
reply: &PmlReply,
subject: f64,
cell: f64,
params: PmlParams,
stored: &mut Vec<f64>,
) -> Option<(InnerFit, usize)> {
if !reply.beta.iter().chain(&reply.log_w).all(|v| v.is_finite()) {
return None;
}
if params.ord == 1
&& !params.reml
&& let Some(fit) = newton_finish(
pml,
&reply.beta,
&reply.log_w,
&PmlVariance { subject, cell },
params.eps,
params.max_iter,
Some(DeviceCurvature {
subject: &reply.subject_curvature,
cross: &reply.cross_block,
schur: &reply.information,
}),
stored,
)
{
return Some((
InnerFit {
log_likelihood: fit.log_likelihood,
log_likelihood_prev: fit.log_likelihood_prev,
log_det: fit.log_det,
second_order: 0.0,
},
fit.iterations,
));
}
let fit = opt_pml_from(
pml,
&reply.beta,
&reply.log_w,
&PmlVariance { subject, cell },
Some(params),
)
.ok()?;
Some((
InnerFit {
log_likelihood: fit.log_likelihood,
log_likelihood_prev: fit.log_likelihood_prev,
log_det: fit.log_det,
second_order: fit.second_order,
},
fit.iterations,
))
}
fn count_histogram(counts: &[f64]) -> Vec<(f64, f64)> {
let mut values: Vec<f64> = counts
.iter()
.copied()
.filter(|&y| y != 1.0 && y != 2.0)
.collect();
values.sort_by(f64::total_cmp);
let mut histogram: Vec<(f64, f64)> = Vec::new();
for y in values {
match histogram.last_mut() {
Some((v, m)) if *v == y => *m += 1.0,
_ => histogram.push((y, 1.0)),
}
}
histogram
}
fn histogram_tail(histogram: &[(f64, f64)], cell: f64) -> f64 {
histogram.iter().map(|&(y, m)| m * ln_gamma(y + cell)).sum()
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_histogram_tail_matches_the_direct_sum() {
let counts = [1.0, 2.0, 3.0, 3.0, 7.0, 1.0, 12.0, 3.0, 2.0, 7.0];
let cell = 0.8;
let direct: f64 = counts
.iter()
.filter(|&&y| y != 1.0 && y != 2.0)
.map(|&y| ln_gamma(y + cell))
.sum();
let histogram = count_histogram(&counts);
assert_eq!(histogram, vec![(3.0, 3.0), (7.0, 2.0), (12.0, 1.0)]);
approx::assert_relative_eq!(
histogram_tail(&histogram, cell),
direct,
max_relative = 1e-14
);
}
}