use core::num::NonZeroUsize;
use crate::Exec;
#[non_exhaustive]
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub enum BatchStrategy {
#[default]
Auto,
Sequential,
Outer,
Inner,
Split {
lanes: NonZeroUsize,
},
}
#[non_exhaustive]
#[derive(Clone, Copy, Debug, PartialEq)]
pub struct WorkModel {
pub item_overhead_ns: f64,
pub muladds_per_ns: f64,
pub min_work_per_lane_ns: f64,
pub min_items: usize,
pub min_items_per_lane: usize,
}
impl Default for WorkModel {
fn default() -> Self {
Self {
item_overhead_ns: 0.0,
muladds_per_ns: crate::host::muladds_per_ns(),
min_work_per_lane_ns: 0.0,
min_items: 2,
min_items_per_lane: 1,
}
}
}
impl WorkModel {
#[must_use]
pub fn with_item_overhead_ns(mut self, nanoseconds: f64) -> Self {
self.item_overhead_ns = nanoseconds;
self
}
#[must_use]
pub fn with_muladds_per_ns(mut self, muladds_per_ns: f64) -> Self {
self.muladds_per_ns = muladds_per_ns;
self
}
#[must_use]
pub fn with_min_work_per_lane_ns(mut self, nanoseconds: f64) -> Self {
self.min_work_per_lane_ns = nanoseconds;
self
}
#[must_use]
pub fn with_min_items(mut self, items: usize) -> Self {
self.min_items = items;
self
}
#[must_use]
pub fn with_min_items_per_lane(mut self, items: usize) -> Self {
self.min_items_per_lane = items;
self
}
}
#[derive(Clone, Copy, Debug, Default, PartialEq)]
pub struct BatchPolicy {
pub strategy: BatchStrategy,
pub model: WorkModel,
}
impl BatchPolicy {
#[must_use]
pub fn new(strategy: BatchStrategy) -> Self {
Self {
strategy,
model: WorkModel::default(),
}
}
#[must_use]
pub fn with_strategy(mut self, strategy: BatchStrategy) -> Self {
self.strategy = strategy;
self
}
#[must_use]
pub fn with_model(mut self, model: WorkModel) -> Self {
self.model = model;
self
}
}
#[non_exhaustive]
#[derive(Clone, Copy, Debug, PartialEq)]
pub struct BatchQuery {
pub items: usize,
pub item_muladds: f64,
pub item_parallel: bool,
pub balanced_lanes: bool,
pub outer_only_when_item_serial: bool,
pub item_sections: u32,
}
impl BatchQuery {
#[must_use]
pub fn new(items: usize, item_muladds: f64) -> Self {
Self {
items,
item_muladds,
item_parallel: true,
balanced_lanes: false,
outer_only_when_item_serial: false,
item_sections: 1,
}
}
#[must_use]
pub fn with_item_sections(mut self, item_sections: u32) -> Self {
self.item_sections = item_sections.max(1);
self
}
#[must_use]
pub fn with_item_parallel(mut self, item_parallel: bool) -> Self {
self.item_parallel = item_parallel;
self
}
#[must_use]
pub fn with_balanced_lanes(mut self, balanced_lanes: bool) -> Self {
self.balanced_lanes = balanced_lanes;
self
}
#[must_use]
pub fn with_outer_only_when_item_serial(mut self, value: bool) -> Self {
self.outer_only_when_item_serial = value;
self
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct BatchDecision {
pub lanes: usize,
pub item_threads: usize,
}
pub fn batch_decision(policy: &BatchPolicy, exec: &Exec<'_>, query: &BatchQuery) -> BatchDecision {
batch_decision_within(policy, exec, exec.budget(), query)
}
pub fn batch_decision_within(
policy: &BatchPolicy,
exec: &Exec<'_>,
budget: usize,
query: &BatchQuery,
) -> BatchDecision {
let budget = budget.max(1).min(exec.budget());
decide(
policy,
budget,
&|ns, sections| exec.width_for_within(budget, ns, sections),
query,
)
}
type Widths<'a> = &'a dyn Fn(f64, u32) -> usize;
pub(crate) fn decide(
policy: &BatchPolicy,
budget: usize,
widths: Widths<'_>,
query: &BatchQuery,
) -> BatchDecision {
match policy.strategy {
BatchStrategy::Auto => auto_decision(&policy.model, budget, widths, query),
BatchStrategy::Sequential => BatchDecision {
lanes: 1,
item_threads: 1,
},
BatchStrategy::Outer => BatchDecision {
lanes: clamp_lanes(budget, query.items),
item_threads: 1,
},
BatchStrategy::Inner => BatchDecision {
lanes: 1,
item_threads: budget.max(1),
},
BatchStrategy::Split { lanes } => {
let lanes = clamp_lanes(budget, query.items).min(lanes.get());
BatchDecision {
lanes,
item_threads: budget / lanes,
}
}
}
}
fn clamp_lanes(budget: usize, items: usize) -> usize {
items.clamp(1, budget.max(1))
}
fn auto_decision(
model: &WorkModel,
budget: usize,
widths: Widths<'_>,
query: &BatchQuery,
) -> BatchDecision {
let items = query.items;
if items == 0 || budget < 2 {
return one_lane(1);
}
let item_ns = item_ns(model, query.item_muladds);
let total_ns = item_ns * items as f64;
if query.outer_only_when_item_serial {
faer_auto(widths, budget, query, item_ns, total_ns)
} else if query.balanced_lanes {
packed_auto(model, widths, query, item_ns, total_ns)
} else {
generic_auto(model, widths, budget, query, item_ns, total_ns)
}
}
fn faer_auto(
widths: Widths<'_>,
budget: usize,
query: &BatchQuery,
item_ns: f64,
total_ns: f64,
) -> BatchDecision {
let inner = if query.item_parallel {
widths(item_ns, query.item_sections)
} else {
1
};
if inner > 1 {
return one_lane(inner);
}
let total = widths(total_ns, 1);
if total > 1 && query.items > 1 {
let lanes = budget.min(query.items).min(total);
return BatchDecision {
lanes,
item_threads: 1,
};
}
one_lane(1)
}
fn generic_auto(
model: &WorkModel,
widths: Widths<'_>,
budget: usize,
query: &BatchQuery,
item_ns: f64,
total_ns: f64,
) -> BatchDecision {
let items = query.items;
let by_work = work_lane_cap(model, total_ns);
let lanes = budget.min(items).min(by_work);
if lanes >= 2
&& items >= model.min_items
&& items >= lanes.saturating_mul(model.min_items_per_lane)
{
return BatchDecision {
lanes,
item_threads: 1,
};
}
let threads = if query.item_parallel {
widths(item_ns, query.item_sections)
} else {
1
};
one_lane(threads)
}
fn packed_auto(
model: &WorkModel,
widths: Widths<'_>,
query: &BatchQuery,
item_ns: f64,
total_ns: f64,
) -> BatchDecision {
let items = query.items;
let narrowed = widths(total_ns, query.item_sections);
let one_lane_threads = if query.item_parallel { narrowed } else { 1 };
if narrowed < 2 {
return one_lane(one_lane_threads);
}
let mut lanes = items.min(narrowed).min(work_lane_cap(model, total_ns));
let gates_pass = lanes >= 2
&& items >= model.min_items
&& items >= lanes.saturating_mul(model.min_items_per_lane);
if !gates_pass {
return one_lane(one_lane_threads);
}
let tiny = widths(item_ns, query.item_sections) < 2;
let balanced = items.saturating_mul(10)
>= 9usize
.saturating_mul(lanes)
.saturating_mul(items.div_ceil(lanes));
if !(tiny || (items >= narrowed && balanced)) {
lanes = 1;
}
BatchDecision {
lanes,
item_threads: if lanes > 1 { 1 } else { one_lane_threads },
}
}
fn work_lane_cap(model: &WorkModel, total_ns: f64) -> usize {
if model.min_work_per_lane_ns > 0.0 {
let raw = total_ns / model.min_work_per_lane_ns;
if raw.is_finite() && raw >= 1.0 {
raw.floor() as usize
} else {
1
}
} else {
usize::MAX
}
}
fn one_lane(threads: usize) -> BatchDecision {
BatchDecision {
lanes: 1,
item_threads: threads.max(1),
}
}
fn item_ns(model: &WorkModel, item_muladds: f64) -> f64 {
let rate = if model.muladds_per_ns > 0.0 {
model.muladds_per_ns
} else {
1.0
};
model.item_overhead_ns + item_muladds / rate
}
#[cfg(test)]
mod tests {
use super::*;
fn budget(n: usize) -> NonZeroUsize {
NonZeroUsize::new(n).unwrap()
}
fn test_cost_ns(width: usize) -> u64 {
500 * width as u64
}
fn test_width(serial_ns: f64, sections: u32, budget: usize) -> usize {
threadteam::width_for_by(
core::time::Duration::from_nanos(serial_ns.max(0.0) as u64),
sections,
NonZeroUsize::new(budget.max(1)).unwrap(),
|w| core::time::Duration::from_nanos(test_cost_ns(w.get())),
)
.get()
}
fn bd(policy: &BatchPolicy, budget: NonZeroUsize, query: &BatchQuery) -> BatchDecision {
let b = budget.get();
decide(
policy,
b,
&|ns, sections| test_width(ns, sections, b),
query,
)
}
fn legacy_packed(items: usize, item_muladds: f64, host_budget: usize) -> (usize, usize) {
let item_ns = item_muladds / crate::host::muladds_per_ns();
let total_ns = item_ns * items as f64;
let narrowed = legacy_width_for(total_ns, host_budget);
if items < 2 || narrowed < 2 {
return (1, narrowed);
}
let width = legacy_width_for(total_ns, narrowed);
if width < 2 {
return (1, narrowed);
}
let tiny = legacy_width_for(item_ns, host_budget) < 2;
let lanes = items.min(width);
let balanced = 10 * items >= 9 * lanes * items.div_ceil(lanes);
let lanes = if tiny || (items >= narrowed && balanced) {
lanes
} else {
1
};
(lanes, if lanes > 1 { 1 } else { narrowed })
}
fn legacy_width_for(serial_ns: f64, budget: usize) -> usize {
let mut best = (1usize, serial_ns);
for k in 2..=budget {
let d = test_cost_ns(k) as f64;
if serial_ns >= threadteam::PAYS * d {
let obj = d + serial_ns / k as f64;
if obj < best.1 {
best = (k, obj);
}
}
}
best.0
}
#[test]
fn auto_matches_legacy_packed_table() {
let policy = BatchPolicy::default();
let items = [0usize, 1, 2, 3, 7, 8, 9, 16, 64, 512];
let muladds = [
1.0,
64.0,
4_096.0,
32_768.0,
262_144.0,
1_000_000.0,
8_000_000.0,
];
let budgets = [1usize, 2, 3, 4, 8, 16];
for &h in &items {
for &m in &muladds {
for &b in &budgets {
let query = BatchQuery::new(h, m).with_balanced_lanes(true);
let got = bd(&policy, budget(b), &query);
let want = legacy_packed(h, m, b);
assert_eq!(
(got.lanes, got.item_threads),
(want.0, want.1),
"items={h} item_muladds={m} budget={b}"
);
}
}
}
}
#[test]
fn generic_auto_matches_tenferro_auto_lanes_table() {
let model = WorkModel::default()
.with_item_overhead_ns(50.0)
.with_muladds_per_ns(16.0)
.with_min_work_per_lane_ns(8_000.0);
let policy = BatchPolicy::new(BatchStrategy::Auto).with_model(model);
let lanes = |item_muladds: f64, batch: usize, threads: usize| {
bd(
&policy,
budget(threads),
&BatchQuery::new(batch, item_muladds),
)
.lanes
};
assert_eq!(lanes(64.0, 1024, 4), 4);
assert_eq!(lanes(64.0, 64, 4), 1);
assert_eq!(lanes(32_768.0, 8, 16), 2);
assert_eq!(lanes(4_096.0, 128, 8), 4);
assert_eq!(lanes(4_096.0, 32, 8), 1);
let no_work_gate =
BatchPolicy::new(BatchStrategy::Auto).with_model(model.with_min_work_per_lane_ns(0.0));
assert_eq!(
bd(&no_work_gate, budget(4), &BatchQuery::new(64, 64.0)).lanes,
4
);
assert_eq!(
bd(&no_work_gate, budget(4), &BatchQuery::new(1, 64.0)).lanes,
1
);
}
fn legacy_faer(count: usize, item_muladds: f64, budget: usize) -> (usize, usize) {
let item_ns = item_muladds / crate::host::muladds_per_ns();
let total_ns = item_ns * count as f64;
let inner = legacy_width_for(item_ns, budget);
let total = legacy_width_for(total_ns, budget);
if inner == 1 && total > 1 && count > 1 {
(total.min(count), 1)
} else {
(1, inner)
}
}
#[test]
fn auto_matches_legacy_faer_table() {
let policy = BatchPolicy::default();
let items = [1usize, 2, 3, 8, 16, 64, 512];
let muladds = [
1.0,
64.0,
4_096.0,
32_768.0,
262_144.0,
1_000_000.0,
8_000_000.0,
];
let budgets = [1usize, 2, 3, 4, 8, 16];
for &count in &items {
for &m in &muladds {
for &b in &budgets {
let query = BatchQuery::new(count, m).with_outer_only_when_item_serial(true);
let got = bd(&policy, budget(b), &query);
let want = legacy_faer(count, m, b);
assert_eq!(
(got.lanes, got.item_threads),
(want.0, want.1),
"count={count} item_muladds={m} budget={b}"
);
}
}
}
}
#[test]
fn faer_auto_keeps_one_item_on_the_team() {
let query = BatchQuery::new(4, 1_000_000.0).with_outer_only_when_item_serial(true);
let d = bd(&BatchPolicy::default(), budget(8), &query);
assert!(d.item_threads > 1, "{d:?}");
assert_eq!(d.lanes, 1);
}
#[test]
fn exec_carries_the_policy_and_budget_keeps_it() {
let sp = threadteam::Pool::builder().threads(4).build().unwrap();
let pool = crate::Pool::borrow(&sp);
let exec = crate::Exec::new(&pool, 4)
.unwrap()
.with_batch_policy(BatchPolicy::new(BatchStrategy::Outer));
assert_eq!(exec.batch_policy().strategy, BatchStrategy::Outer);
let narrowed = exec.limit(2.try_into().unwrap());
assert_eq!(narrowed.budget(), 2);
assert_eq!(narrowed.batch_policy().strategy, BatchStrategy::Outer);
assert_eq!(
crate::Exec::serial().batch_policy().strategy,
BatchStrategy::Auto
);
}
#[test]
fn forced_strategies_are_clamped_not_fallen_back() {
let query = BatchQuery::new(3, 1_000.0);
assert_eq!(
bd(
&BatchPolicy::new(BatchStrategy::Sequential),
budget(8),
&query
),
BatchDecision {
lanes: 1,
item_threads: 1
}
);
assert_eq!(
bd(&BatchPolicy::new(BatchStrategy::Outer), budget(8), &query),
BatchDecision {
lanes: 3,
item_threads: 1
}
);
assert_eq!(
bd(&BatchPolicy::new(BatchStrategy::Inner), budget(8), &query),
BatchDecision {
lanes: 1,
item_threads: 8
}
);
assert_eq!(
bd(
&BatchPolicy::new(BatchStrategy::Split { lanes: budget(8) }),
budget(8),
&query
),
BatchDecision {
lanes: 3,
item_threads: 2
}
);
let many = BatchQuery::new(64, 1_000.0);
assert_eq!(
bd(
&BatchPolicy::new(BatchStrategy::Split { lanes: budget(64) }),
budget(4),
&many
),
BatchDecision {
lanes: 4,
item_threads: 1
}
);
let empty = BatchQuery::new(0, 1.0);
assert_eq!(
bd(&BatchPolicy::new(BatchStrategy::Outer), budget(4), &empty),
BatchDecision {
lanes: 1,
item_threads: 1
}
);
assert_eq!(
bd(
&BatchPolicy::new(BatchStrategy::Split { lanes: budget(2) }),
budget(4),
&empty
),
BatchDecision {
lanes: 1,
item_threads: 4
}
);
}
#[test]
fn every_decision_fits_the_budget() {
let budgets = [1usize, 2, 3, 4, 8, 16];
let items = [0usize, 1, 2, 3, 7, 8, 16, 64];
let policies = [
BatchStrategy::Auto,
BatchStrategy::Sequential,
BatchStrategy::Outer,
BatchStrategy::Inner,
BatchStrategy::Split { lanes: budget(1) },
BatchStrategy::Split { lanes: budget(2) },
BatchStrategy::Split { lanes: budget(8) },
];
for &b in &budgets {
for &n in &items {
for strategy in policies {
let d = bd(
&BatchPolicy::new(strategy),
budget(b),
&BatchQuery::new(n, 512.0),
);
assert!(
d.lanes >= 1 && d.lanes * d.item_threads <= b,
"strategy {strategy:?}, budget {b}, items {n}: {d:?}"
);
}
}
}
}
#[test]
fn forced_strategies_use_the_original_budget() {
let query = BatchQuery::new(9, 40_000.0 * crate::host::muladds_per_ns() / 27.0)
.with_balanced_lanes(true);
let auto = bd(&BatchPolicy::default(), budget(8), &query);
assert_eq!((auto.lanes, auto.item_threads), (5, 1));
let outer = bd(&BatchPolicy::new(BatchStrategy::Outer), budget(8), &query);
assert_eq!((outer.lanes, outer.item_threads), (8, 1));
}
#[test]
fn item_parallel_capability_keeps_item_threads_one() {
let query = BatchQuery::new(1, 1_000_000.0);
let allowed = bd(&BatchPolicy::default(), budget(8), &query);
assert!(allowed.item_threads > 1, "{allowed:?}");
let forbidden = query.with_item_parallel(false);
let d = bd(&BatchPolicy::default(), budget(8), &forbidden);
assert_eq!((d.lanes, d.item_threads), (1, 1));
}
#[test]
fn work_gate_reduces_lanes_and_disabling_it_does_not() {
let query = BatchQuery::new(8, 8_000_000.0).with_balanced_lanes(true);
let disabled = WorkModel::default();
let enabled = WorkModel::default().with_min_work_per_lane_ns(1_000_000.0);
let d0 = bd(
&BatchPolicy::new(BatchStrategy::Auto).with_model(disabled),
budget(8),
&query,
);
let d1 = bd(
&BatchPolicy::new(BatchStrategy::Auto).with_model(enabled),
budget(8),
&query,
);
assert!(d0.lanes > d1.lanes, "{d0:?} vs {d1:?}");
}
#[test]
fn balanced_lanes_can_hold_a_lane_back() {
let query = BatchQuery::new(9, 1_000_000.0).with_item_parallel(true);
let packed = bd(
&BatchPolicy::new(BatchStrategy::Auto).with_model(WorkModel::default()),
budget(8),
&query.with_balanced_lanes(true),
);
let generic = bd(
&BatchPolicy::new(BatchStrategy::Auto).with_model(WorkModel::default()),
budget(8),
&query.with_balanced_lanes(false),
);
assert_eq!(packed.lanes, 1);
assert!(generic.lanes > 1, "{generic:?}");
}
}