use std::time::{Duration, Instant};
use std::future::Future;
use std::pin::Pin;
use cubecl::bytes::Bytes;
use cubecl::prelude::*;
use cubecl::server::ServerError;
use cubecl_utils_rs::prelude::*;
use rayon::prelude::*;
use crate::errors::EdgeErrors;
use crate::gpu::pml_kernel::{OUT_HEADER, PmlGpuTensors, SUBJECT_SLOTS, launch_opt_pml};
#[derive(Clone, Copy, Debug)]
pub struct GpuGene<'a> {
pub counts: &'a [f64],
pub cell_index: &'a [usize],
pub subject_total: &'a [f64],
pub beta_init: &'a [f64],
pub sigma: f64,
pub gamma: f64,
}
#[derive(Clone, Debug)]
pub struct PmlRequest {
pub gene: usize,
pub subject: f64,
pub cell: f64,
pub beta_init: Vec<f64>,
pub log_w_init: Vec<f64>,
}
#[derive(Clone, Debug)]
pub struct PmlReply {
pub log_likelihood: f64,
pub log_likelihood_prev: f64,
pub log_det: f64,
pub iterations: u32,
pub backtracks: u32,
pub beta: Vec<f64>,
pub log_w: Vec<f64>,
pub information: Vec<f64>,
pub subject_curvature: Vec<f64>,
pub cross_block: Vec<f64>,
}
#[derive(Clone, Debug)]
pub struct GpuPmlBatch {
pub beta: Vec<f64>,
pub information: Vec<f64>,
pub log_likelihood: Vec<f64>,
pub log_det: Vec<f64>,
pub iterations: Vec<u32>,
pub backtracks: Vec<u32>,
pub n_coef: usize,
}
#[derive(Clone, Copy, Debug)]
pub struct GpuSolveParams {
pub eps: f64,
pub noise_scale: f64,
pub max_iter: u32,
pub max_backtrack: u32,
pub full: bool,
pub information: bool,
}
#[derive(Clone, Copy, Debug, Default)]
pub struct SolveTiming {
pub staging: Duration,
pub blocked: Duration,
pub scatter: Duration,
}
pub struct PendingSolve<'a> {
out: Pin<Box<dyn Future<Output = Result<Vec<Bytes>, ServerError>> + Send + 'a>>,
n_req: usize,
full: bool,
information: bool,
}
pub struct ResidentBatch {
design: GpuTensor<f32>,
log_offset: GpuTensor<f32>,
subject_start: GpuTensor<u32>,
counts: GpuTensor<f32>,
cells: GpuTensor<u32>,
subject_ptr: GpuTensor<u32>,
subject_total: GpuTensor<f32>,
subject_mean: GpuTensor<f32>,
subject_scratch: GpuTensor<f32>,
vwb_scratch: GpuTensor<f32>,
capacity: usize,
n_genes: usize,
k: usize,
nb: usize,
timing: Option<SolveTiming>,
}
impl ResidentBatch {
pub fn upload(
design: &[f64],
log_offset: &[f64],
subject_start: &[usize],
genes: &[GpuGene<'_>],
client: &Client,
) -> Result<Self, EdgeErrors> {
let n_genes = genes.len();
if n_genes == 0 {
return Err(EdgeErrors::MustBePositive("n_genes".to_string()));
}
let n_cells = log_offset.len();
if n_cells == 0 {
return Err(EdgeErrors::MustBePositive("n_cells".to_string()));
}
let k = subject_start.len().saturating_sub(1);
if k == 0 {
return Err(EdgeErrors::MustBePositive("n_subjects".to_string()));
}
if !design.len().is_multiple_of(n_cells) {
return Err(EdgeErrors::LengthMismatch {
name: "design",
expected: n_cells,
got: design.len(),
});
}
let nb = design.len() / n_cells;
for gene in genes {
if gene.counts.len() != gene.cell_index.len() {
return Err(EdgeErrors::LengthMismatch {
name: "counts",
expected: gene.cell_index.len(),
got: gene.counts.len(),
});
}
if gene.subject_total.len() != k {
return Err(EdgeErrors::LengthMismatch {
name: "subject_total",
expected: k,
got: gene.subject_total.len(),
});
}
}
let design_f32: Vec<f32> = design.par_iter().map(|&v| v as f32).collect();
let offset_f32: Vec<f32> = log_offset.par_iter().map(|&v| v as f32).collect();
let start_u32: Vec<u32> = subject_start.iter().map(|&v| v as u32).collect();
let mut gene_ptr = Vec::with_capacity(n_genes + 1);
gene_ptr.push(0u32);
let mut running = 0usize;
for gene in genes {
running += gene.counts.len();
gene_ptr.push(running as u32);
}
let nnz = running;
let mut counts_f32 = vec![0.0f32; nnz];
let mut cells_u32 = vec![0u32; nnz];
{
let mut count_runs: Vec<&mut [f32]> = Vec::with_capacity(n_genes);
let mut cell_runs: Vec<&mut [u32]> = Vec::with_capacity(n_genes);
let mut count_rest = counts_f32.as_mut_slice();
let mut cell_rest = cells_u32.as_mut_slice();
for gene in genes {
let (a, b) = count_rest.split_at_mut(gene.counts.len());
count_runs.push(a);
count_rest = b;
let (c, d) = cell_rest.split_at_mut(gene.cell_index.len());
cell_runs.push(c);
cell_rest = d;
}
count_runs
.into_par_iter()
.zip(cell_runs)
.zip(genes)
.for_each(|((count_run, cell_run), gene)| {
for (out, &v) in count_run.iter_mut().zip(gene.counts) {
*out = v as f32;
}
for (out, &c) in cell_run.iter_mut().zip(gene.cell_index) {
*out = c as u32;
}
});
}
let mut subject_ptr = vec![0u32; n_genes * (k + 1)];
for (g, gene) in genes.iter().enumerate() {
let base = gene_ptr[g] as usize;
let mut p = 0usize;
for s in 0..=k {
while p < gene.cell_index.len() && gene.cell_index[p] < subject_start[s] {
p += 1;
}
subject_ptr[g * (k + 1) + s] = (base + p) as u32;
}
}
let mut subject_total = vec![0.0f32; k * n_genes];
for (g, gene) in genes.iter().enumerate() {
for (s, &t) in gene.subject_total.iter().enumerate() {
subject_total[s * n_genes + g] = t as f32;
}
}
let mut subject_mean = vec![0.0f32; k * nb];
for s in 0..k {
let (lo, hi) = (subject_start[s], subject_start[s + 1]);
if lo == hi {
continue;
}
for j in 0..nb {
let head = design[lo * nb + j];
if (lo..hi).all(|c| design[c * nb + j] == head) {
subject_mean[s * nb + j] = head as f32;
} else {
let total: f64 = (lo..hi).map(|c| design[c * nb + j]).sum();
subject_mean[s * nb + j] = (total / (hi - lo) as f64) as f32;
}
}
}
let err = |e: CubeclUtilsErrors| EdgeErrors::Gpu(e.to_string());
let capacity = n_genes;
Ok(Self {
design: GpuTensor::from_slice(&design_f32, vec![n_cells * nb], client).map_err(err)?,
log_offset: GpuTensor::from_slice(&offset_f32, vec![n_cells], client).map_err(err)?,
subject_start: GpuTensor::from_slice(&start_u32, vec![k + 1], client).map_err(err)?,
counts: GpuTensor::from_slice(&counts_f32, vec![nnz.max(1)], client).map_err(err)?,
cells: GpuTensor::from_slice(&cells_u32, vec![nnz.max(1)], client).map_err(err)?,
subject_ptr: GpuTensor::from_slice(&subject_ptr, vec![n_genes * (k + 1)], client)
.map_err(err)?,
subject_total: GpuTensor::from_slice(&subject_total, vec![k * n_genes], client)
.map_err(err)?,
subject_mean: GpuTensor::from_slice(&subject_mean, vec![k * nb], client)
.map_err(err)?,
subject_scratch: GpuTensor::empty(vec![SUBJECT_SLOTS as usize * k * capacity], client)
.map_err(err)?,
vwb_scratch: GpuTensor::empty(vec![k * nb * capacity], client).map_err(err)?,
capacity,
n_genes,
k,
nb,
timing: None,
})
}
pub fn time_solves(&mut self) {
self.timing = Some(SolveTiming::default());
}
pub fn timing(&self) -> Option<SolveTiming> {
self.timing
}
pub fn n_coef(&self) -> usize {
self.nb
}
pub fn solve(
&mut self,
requests: &[PmlRequest],
params: &GpuSolveParams,
client: &Client,
) -> Result<Vec<PmlReply>, EdgeErrors> {
let pending = self.submit(requests, params, client)?;
self.collect(pending)
}
pub fn submit<'a>(
&mut self,
requests: &[PmlRequest],
params: &GpuSolveParams,
client: &'a Client,
) -> Result<PendingSolve<'a>, EdgeErrors> {
let n_req = requests.len();
if n_req == 0 {
return Ok(PendingSolve {
out: Box::pin(async { Ok(Vec::new()) }),
n_req,
full: params.full,
information: params.information,
});
}
let (n_genes, k, nb) = (self.n_genes, self.k, self.nb);
let err = |e: CubeclUtilsErrors| EdgeErrors::Gpu(e.to_string());
let started = Instant::now();
let mut request_gene = vec![0u32; n_req];
let mut request_params = vec![0.0f32; 3 * n_req];
let mut beta_init = vec![0.0f32; nb * n_req];
let mut log_w_init = vec![0.0f32; k * n_req];
for (q, req) in requests.iter().enumerate() {
if req.gene >= n_genes {
return Err(EdgeErrors::InvalidArgument(format!(
"Request {q} names gene {}, but only {n_genes} are resident.",
req.gene
)));
}
if req.beta_init.len() != nb {
return Err(EdgeErrors::LengthMismatch {
name: "beta_init",
expected: nb,
got: req.beta_init.len(),
});
}
let exps = req.subject.exp();
if !(exps.is_finite() && exps > 1.0) {
return Err(EdgeErrors::InvalidArgument(format!(
"PmlVariance::subject must be finite and strictly positive; request {q} has {}.",
req.subject
)));
}
if !(req.cell.is_finite() && req.cell > 0.0) {
return Err(EdgeErrors::InvalidDispersion(req.cell));
}
request_gene[q] = req.gene as u32;
request_params[q] = (1.0 / (exps - 1.0)) as f32;
request_params[n_req + q] = (1.0 / (exps.sqrt() * (exps - 1.0))) as f32;
request_params[2 * n_req + q] = req.cell as f32;
if !(req.log_w_init.is_empty() || req.log_w_init.len() == k) {
return Err(EdgeErrors::LengthMismatch {
name: "log_w_init",
expected: k,
got: req.log_w_init.len(),
});
}
for (j, &b) in req.beta_init.iter().enumerate() {
beta_init[j * n_req + q] = b as f32;
}
for (s, &w) in req.log_w_init.iter().enumerate() {
log_w_init[s * n_req + q] = w as f32;
}
}
if n_req > self.capacity {
self.subject_scratch =
GpuTensor::empty(vec![SUBJECT_SLOTS as usize * k * n_req], client).map_err(err)?;
self.vwb_scratch = GpuTensor::empty(vec![k * nb * n_req], client).map_err(err)?;
self.capacity = n_req;
}
let header = OUT_HEADER as usize;
let tensors = PmlGpuTensors::<f32> {
design: self.design.clone(),
log_offset: self.log_offset.clone(),
subject_start: self.subject_start.clone(),
counts: self.counts.clone(),
cells: self.cells.clone(),
subject_ptr: self.subject_ptr.clone(),
subject_total: self.subject_total.clone(),
subject_mean: self.subject_mean.clone(),
request_gene: GpuTensor::from_slice(&request_gene, vec![n_req], client).map_err(err)?,
request_params: GpuTensor::from_slice(&request_params, vec![3 * n_req], client)
.map_err(err)?,
beta_init: GpuTensor::from_slice(&beta_init, vec![nb * n_req], client).map_err(err)?,
log_w_init: GpuTensor::from_slice(&log_w_init, vec![k * n_req], client).map_err(err)?,
tolerance: GpuTensor::from_slice(
&[params.eps as f32, params.noise_scale as f32],
vec![2],
client,
)
.map_err(err)?,
subject_scratch: self.subject_scratch.clone(),
vwb_scratch: self.vwb_scratch.clone(),
out: GpuTensor::empty(
vec![(header + nb + nb * nb + 2 * k + k * nb) * n_req],
client,
)
.map_err(err)?,
};
launch_opt_pml::<f32>(
&tensors,
n_genes,
n_req,
k,
nb,
params.max_iter,
params.max_backtrack,
params.information,
client,
)?;
let out = Box::pin(client.read_async(vec![tensors.out.handle().clone()]));
if let Some(t) = self.timing.as_mut() {
t.staging += started.elapsed();
}
Ok(PendingSolve {
out,
n_req,
full: params.full,
information: params.information,
})
}
pub fn collect(&mut self, pending: PendingSolve<'_>) -> Result<Vec<PmlReply>, EdgeErrors> {
let (n_req, k, nb) = (pending.n_req, self.k, self.nb);
let header = OUT_HEADER as usize;
let started = Instant::now();
let bytes =
cubecl::future::block_on(pending.out).map_err(|e| EdgeErrors::Gpu(e.to_string()))?;
let waited = Instant::now();
let Some(bytes) = bytes.first() else {
return Ok(Vec::new());
};
let out = f32::from_bytes(bytes);
let rows = |first: usize, n: usize, q: usize| -> Vec<f64> {
if pending.full {
(first..first + n)
.map(|r| out[r * n_req + q] as f64)
.collect()
} else {
Vec::new()
}
};
let replies = (0..n_req)
.map(|q| PmlReply {
log_likelihood: out[q] as f64,
log_likelihood_prev: out[n_req + q] as f64,
log_det: if pending.information {
out[2 * n_req + q] as f64
} else {
f64::NAN
},
iterations: out[4 * n_req + q] as u32,
backtracks: out[5 * n_req + q] as u32,
beta: rows(header, nb, q),
information: rows(header + nb, nb * nb, q),
log_w: rows(header + nb + nb * nb, k, q),
subject_curvature: rows(header + nb + nb * nb + k, k, q),
cross_block: rows(header + nb + nb * nb + 2 * k, k * nb, q),
})
.collect();
if let Some(t) = self.timing.as_mut() {
t.blocked += waited - started;
t.scatter += waited.elapsed();
}
Ok(replies)
}
}
#[allow(clippy::too_many_arguments)]
pub fn opt_pml_batch(
design: &[f64],
log_offset: &[f64],
subject_start: &[usize],
genes: &[GpuGene<'_>],
eps: f64,
noise_scale: f64,
max_iter: u32,
max_backtrack: u32,
client: &Client,
) -> Result<GpuPmlBatch, EdgeErrors> {
let mut resident = ResidentBatch::upload(design, log_offset, subject_start, genes, client)?;
solve_per_gene(
&mut resident,
genes,
&GpuSolveParams {
eps,
noise_scale,
max_iter,
max_backtrack,
full: true,
information: true,
},
client,
)
}
pub fn solve_per_gene(
resident: &mut ResidentBatch,
genes: &[GpuGene<'_>],
params: &GpuSolveParams,
client: &Client,
) -> Result<GpuPmlBatch, EdgeErrors> {
let nb = resident.n_coef();
let requests: Vec<PmlRequest> = genes
.iter()
.enumerate()
.map(|(g, gene)| PmlRequest {
gene: g,
subject: gene.sigma,
cell: gene.gamma,
beta_init: gene.beta_init.to_vec(),
log_w_init: Vec::new(),
})
.collect();
let replies = resident.solve(&requests, params, client)?;
Ok(GpuPmlBatch {
beta: replies
.iter()
.flat_map(|r| r.beta.iter().copied())
.collect(),
information: replies
.iter()
.flat_map(|r| r.information.iter().copied())
.collect(),
log_likelihood: replies.iter().map(|r| r.log_likelihood).collect(),
log_det: replies.iter().map(|r| r.log_det).collect(),
iterations: replies.iter().map(|r| r.iterations).collect(),
backtracks: replies.iter().map(|r| r.backtracks).collect(),
n_coef: nb,
})
}