use std::cmp::Ordering;
use std::collections::{BTreeMap, HashMap};
use std::hash::Hash;
use crate::entity::EntityRecord;
use crate::ids::{ClientId, EntityHandle, Tick};
#[cfg(not(feature = "simd"))]
use crate::interest::RangeOnlyVisibility;
use crate::interest::{ViewerQuery, VisibilityFilter};
use crate::policy::{CompiledSyncPolicy, PolicyTable};
use crate::spatial_index::{CellIndex, CellQueryScratch, CellQueryStats, CellQueryStrategy};
use crate::station::Station;
const HASHED_REPLICATION_TRACKER_MIN_ENTRIES: usize = 2_048;
#[derive(Clone, Debug)]
enum AdaptiveTrackMap<K, V> {
Ordered(BTreeMap<K, V>),
Hashed(HashMap<K, V>),
}
impl<K: Copy + Eq + Hash + Ord, V> AdaptiveTrackMap<K, V> {
fn new() -> Self {
Self::Ordered(BTreeMap::new())
}
fn len(&self) -> usize {
match self {
Self::Ordered(entries) => entries.len(),
Self::Hashed(entries) => entries.len(),
}
}
fn is_empty(&self) -> bool {
match self {
Self::Ordered(entries) => entries.is_empty(),
Self::Hashed(entries) => entries.is_empty(),
}
}
fn contains_key(&self, key: &K) -> bool {
match self {
Self::Ordered(entries) => entries.contains_key(key),
Self::Hashed(entries) => entries.contains_key(key),
}
}
fn get(&self, key: &K) -> Option<&V> {
match self {
Self::Ordered(entries) => entries.get(key),
Self::Hashed(entries) => entries.get(key),
}
}
fn get_mut(&mut self, key: &K) -> Option<&mut V> {
match self {
Self::Ordered(entries) => entries.get_mut(key),
Self::Hashed(entries) => entries.get_mut(key),
}
}
fn insert(&mut self, key: K, value: V) -> Option<V> {
let promote = match self {
Self::Ordered(entries) => {
entries.len() >= HASHED_REPLICATION_TRACKER_MIN_ENTRIES.saturating_sub(1)
&& !entries.contains_key(&key)
}
Self::Hashed(_) => false,
};
if promote {
let Self::Ordered(ordered) = std::mem::replace(self, Self::Hashed(HashMap::new()))
else {
unreachable!("promotion starts from ordered tracker storage");
};
let mut hashed = HashMap::with_capacity(ordered.len().saturating_add(1));
hashed.extend(ordered);
*self = Self::Hashed(hashed);
}
match self {
Self::Ordered(entries) => entries.insert(key, value),
Self::Hashed(entries) => entries.insert(key, value),
}
}
fn retain<F>(&mut self, mut keep: F)
where
F: FnMut(&K, &mut V) -> bool,
{
match self {
Self::Ordered(entries) => entries.retain(|key, value| keep(key, value)),
Self::Hashed(entries) => entries.retain(|key, value| keep(key, value)),
}
}
#[cfg(test)]
fn is_hashed(&self) -> bool {
matches!(self, Self::Hashed(_))
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct ReplicationBudget {
pub max_entities: usize,
pub max_bytes: usize,
pub estimated_entity_bytes: usize,
}
impl Default for ReplicationBudget {
fn default() -> Self {
Self {
max_entities: 300,
max_bytes: 16 * 1024,
estimated_entity_bytes: 32,
}
}
}
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct ReplicationPlan {
pub entities: Vec<EntityHandle>,
pub stats: ReplicationStats,
}
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct ReplicationBatchStats {
pub viewers: usize,
pub candidates: usize,
pub considered: usize,
pub selected: usize,
pub unexamined_after_budget: usize,
pub estimated_bytes: usize,
pub grid_queries: usize,
pub occupied_queries: usize,
pub grid_cells_probed: usize,
pub occupied_cells_scanned: usize,
pub matched_cells: usize,
pub candidate_capacity_max: usize,
pub dedup_capacity_max: usize,
pub matching_cell_capacity_max: usize,
pub priority_capacity_max: usize,
}
impl ReplicationBatchStats {
fn record(&mut self, plan: &ReplicationPlan, scratch: &ReplicationScratch) {
self.viewers = self.viewers.saturating_add(1);
self.candidates = self.candidates.saturating_add(plan.stats.candidates);
self.considered = self.considered.saturating_add(plan.stats.considered);
self.selected = self.selected.saturating_add(plan.stats.selected);
self.unexamined_after_budget = self
.unexamined_after_budget
.saturating_add(plan.stats.unexamined_after_budget);
self.estimated_bytes = self
.estimated_bytes
.saturating_add(plan.stats.estimated_bytes);
let query = scratch.query_stats();
match query.strategy {
CellQueryStrategy::Grid => self.grid_queries = self.grid_queries.saturating_add(1),
CellQueryStrategy::OccupiedCells => {
self.occupied_queries = self.occupied_queries.saturating_add(1);
}
}
self.grid_cells_probed = self
.grid_cells_probed
.saturating_add(query.grid_cells_probed);
self.occupied_cells_scanned = self
.occupied_cells_scanned
.saturating_add(query.occupied_cells_scanned);
self.matched_cells = self.matched_cells.saturating_add(query.matched_cells);
self.candidate_capacity_max = self
.candidate_capacity_max
.max(scratch.candidate_capacity());
self.dedup_capacity_max = self
.dedup_capacity_max
.max(scratch.candidate_dedup_capacity());
self.matching_cell_capacity_max = self
.matching_cell_capacity_max
.max(scratch.matching_cell_capacity());
self.priority_capacity_max = self
.priority_capacity_max
.max(scratch.prioritized_capacity());
}
pub fn merge(&mut self, other: Self) {
self.viewers = self.viewers.saturating_add(other.viewers);
self.candidates = self.candidates.saturating_add(other.candidates);
self.considered = self.considered.saturating_add(other.considered);
self.selected = self.selected.saturating_add(other.selected);
self.estimated_bytes = self.estimated_bytes.saturating_add(other.estimated_bytes);
self.grid_queries = self.grid_queries.saturating_add(other.grid_queries);
self.occupied_queries = self.occupied_queries.saturating_add(other.occupied_queries);
self.grid_cells_probed = self
.grid_cells_probed
.saturating_add(other.grid_cells_probed);
self.occupied_cells_scanned = self
.occupied_cells_scanned
.saturating_add(other.occupied_cells_scanned);
self.matched_cells = self.matched_cells.saturating_add(other.matched_cells);
self.candidate_capacity_max = self
.candidate_capacity_max
.max(other.candidate_capacity_max);
self.dedup_capacity_max = self.dedup_capacity_max.max(other.dedup_capacity_max);
self.matching_cell_capacity_max = self
.matching_cell_capacity_max
.max(other.matching_cell_capacity_max);
self.priority_capacity_max = self.priority_capacity_max.max(other.priority_capacity_max);
}
}
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct ReplicationBatchResult {
pub plans: Vec<ReplicationPlan>,
pub stats: ReplicationBatchStats,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct ReplicationBatchView<'a> {
pub plans: &'a [ReplicationPlan],
pub stats: ReplicationBatchStats,
}
#[derive(Clone, Debug, Default)]
pub struct ReplicationBatchScratch {
plans: Vec<ReplicationPlan>,
active_plans: usize,
stats: ReplicationBatchStats,
}
impl ReplicationBatchScratch {
pub const fn new() -> Self {
Self {
plans: Vec::new(),
active_plans: 0,
stats: ReplicationBatchStats {
viewers: 0,
candidates: 0,
considered: 0,
selected: 0,
unexamined_after_budget: 0,
estimated_bytes: 0,
grid_queries: 0,
occupied_queries: 0,
grid_cells_probed: 0,
occupied_cells_scanned: 0,
matched_cells: 0,
candidate_capacity_max: 0,
dedup_capacity_max: 0,
matching_cell_capacity_max: 0,
priority_capacity_max: 0,
},
}
}
pub fn retained_plan_slots(&self) -> usize {
self.plans.len()
}
pub fn retained_entity_capacity(&self) -> usize {
self.plans.iter().map(|plan| plan.entities.capacity()).sum()
}
pub fn view(&self) -> ReplicationBatchView<'_> {
ReplicationBatchView {
plans: &self.plans[..self.active_plans],
stats: self.stats,
}
}
fn prepare(&mut self, plans: usize) {
if self.plans.len() < plans {
self.plans.resize_with(plans, ReplicationPlan::default);
}
self.active_plans = plans;
self.stats = ReplicationBatchStats::default();
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct ReplicationTrackerConfig {
pub max_entries: usize,
}
impl Default for ReplicationTrackerConfig {
fn default() -> Self {
Self {
max_entries: 65_536,
}
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub struct ReplicationTrackKey {
pub client_id: ClientId,
pub entity: EntityHandle,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct ReplicationTrackRecord {
pub client_id: ClientId,
pub entity: EntityHandle,
pub last_sent: Tick,
pub last_acked: Option<Tick>,
}
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct ReplicationTrackerStats {
pub entries: usize,
pub sent_records: usize,
pub acked_records: usize,
pub pruned_records: usize,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum ReplicationTrackerError {
CapacityExceeded {
current: usize,
needed: usize,
max: usize,
},
}
impl core::fmt::Display for ReplicationTrackerError {
fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
match self {
Self::CapacityExceeded {
current,
needed,
max,
} => write!(
f,
"replication tracker capacity exceeded: current {current}, needed {needed}, max {max}"
),
}
}
}
impl std::error::Error for ReplicationTrackerError {}
#[derive(Clone, Debug)]
pub struct ReplicationTracker {
config: ReplicationTrackerConfig,
records: AdaptiveTrackMap<ReplicationTrackKey, ReplicationTrackRecord>,
stats: ReplicationTrackerStats,
}
impl Default for ReplicationTracker {
fn default() -> Self {
Self::new(ReplicationTrackerConfig::default())
}
}
impl ReplicationTracker {
pub fn new(config: ReplicationTrackerConfig) -> Self {
Self {
config,
records: AdaptiveTrackMap::new(),
stats: ReplicationTrackerStats::default(),
}
}
pub const fn config(&self) -> ReplicationTrackerConfig {
self.config
}
pub const fn stats(&self) -> ReplicationTrackerStats {
self.stats
}
pub fn len(&self) -> usize {
self.records.len()
}
pub fn is_empty(&self) -> bool {
self.records.is_empty()
}
pub fn last_sent(&self, client_id: ClientId, entity: EntityHandle) -> Option<Tick> {
self.records
.get(&ReplicationTrackKey { client_id, entity })
.map(|record| record.last_sent)
}
pub fn get(&self, client_id: ClientId, entity: EntityHandle) -> Option<ReplicationTrackRecord> {
self.records
.get(&ReplicationTrackKey { client_id, entity })
.copied()
}
pub fn record_plan_sent(
&mut self,
client_id: ClientId,
plan: &ReplicationPlan,
sent_at: Tick,
) -> Result<usize, ReplicationTrackerError> {
self.ensure_capacity_for(client_id, &plan.entities)?;
let mut recorded = 0;
for entity in &plan.entities {
let key = ReplicationTrackKey {
client_id,
entity: *entity,
};
self.records.insert(
key,
ReplicationTrackRecord {
client_id,
entity: *entity,
last_sent: sent_at,
last_acked: None,
},
);
recorded += 1;
}
self.refresh_entry_count();
self.stats.sent_records = self.stats.sent_records.saturating_add(recorded);
Ok(recorded)
}
pub fn acknowledge(
&mut self,
client_id: ClientId,
entity: EntityHandle,
acked_at: Tick,
) -> bool {
let Some(record) = self
.records
.get_mut(&ReplicationTrackKey { client_id, entity })
else {
return false;
};
record.last_acked = Some(acked_at);
self.stats.acked_records = self.stats.acked_records.saturating_add(1);
true
}
pub fn acknowledge_plan(
&mut self,
client_id: ClientId,
plan: &ReplicationPlan,
acked_at: Tick,
) -> usize {
plan.entities
.iter()
.filter(|entity| self.acknowledge(client_id, **entity, acked_at))
.count()
}
pub fn clear_client(&mut self, client_id: ClientId) -> usize {
let before = self.records.len();
self.records.retain(|key, _| key.client_id != client_id);
let pruned = before.saturating_sub(self.records.len());
self.stats.pruned_records = self.stats.pruned_records.saturating_add(pruned);
self.refresh_entry_count();
pruned
}
pub fn prune_sent_before(&mut self, older_than: Tick) -> usize {
let before = self.records.len();
self.records
.retain(|_, record| record.last_sent.get() >= older_than.get());
let pruned = before.saturating_sub(self.records.len());
self.stats.pruned_records = self.stats.pruned_records.saturating_add(pruned);
self.refresh_entry_count();
pruned
}
fn ensure_capacity_for(
&self,
client_id: ClientId,
entities: &[EntityHandle],
) -> Result<(), ReplicationTrackerError> {
if self.records.len().saturating_add(entities.len()) <= self.config.max_entries {
return Ok(());
}
let mut needed = 0_usize;
for entity in entities {
if !self.records.contains_key(&ReplicationTrackKey {
client_id,
entity: *entity,
}) {
needed = needed.saturating_add(1);
}
}
if self.records.len().saturating_add(needed) > self.config.max_entries {
return Err(ReplicationTrackerError::CapacityExceeded {
current: self.records.len(),
needed,
max: self.config.max_entries,
});
}
Ok(())
}
fn refresh_entry_count(&mut self) {
self.stats.entries = self.records.len();
}
}
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct ReplicationStats {
pub candidates: usize,
pub considered: usize,
pub selected: usize,
pub skipped_by_budget: usize,
pub unexamined_after_budget: usize,
pub skipped_by_cadence: usize,
pub estimated_bytes: usize,
}
#[derive(Clone, Copy, Debug, Default)]
pub struct ReplicationCadence;
impl ReplicationCadence {
pub fn target_hz(policy: &CompiledSyncPolicy, distance_squared: f32) -> u16 {
let min_hz = policy.min_hz.max(1);
let max_hz = policy.max_hz.max(min_hz);
let radius_squared = policy.interest_radius * policy.interest_radius;
let closeness =
if radius_squared.is_finite() && radius_squared > 0.0 && distance_squared.is_finite() {
1.0 - (distance_squared / radius_squared).clamp(0.0, 1.0)
} else {
1.0
};
let span = f32::from(max_hz - min_hz);
let target = f32::from(min_hz) + span * closeness;
rounded_frequency_to_u16(target, min_hz, max_hz)
}
pub fn interval_ticks(
policy: &CompiledSyncPolicy,
station_tick_rate_hz: u16,
distance_squared: f32,
) -> u64 {
let tick_rate = u64::from(station_tick_rate_hz.max(1));
let target_hz = u64::from(Self::target_hz(policy, distance_squared).max(1));
tick_rate.div_ceil(target_hz).max(1)
}
pub fn should_send(
policy: &CompiledSyncPolicy,
station_tick_rate_hz: u16,
distance_squared: f32,
now: Tick,
last_sent: Option<Tick>,
) -> bool {
let Some(last_sent) = last_sent else {
return true;
};
let interval = Self::interval_ticks(policy, station_tick_rate_hz, distance_squared);
now.get().saturating_sub(last_sent.get()) >= interval
}
}
#[derive(Clone, Copy, Debug, Default)]
pub struct ReplicationPriority;
#[allow(clippy::cast_possible_truncation, clippy::cast_sign_loss)]
fn rounded_frequency_to_u16(target: f32, min_hz: u16, max_hz: u16) -> u16 {
let bounded = target.round().clamp(f32::from(min_hz), f32::from(max_hz));
bounded as u16
}
#[allow(clippy::cast_possible_truncation, clippy::cast_sign_loss)]
fn normalized_score_to_u64(closeness: f32) -> u64 {
debug_assert!(closeness.is_finite() && (0.0..=1.0).contains(&closeness));
(closeness * 1_000_000.0).round() as u64
}
impl ReplicationPriority {
pub fn score(policy: &CompiledSyncPolicy, distance_squared: f32) -> u64 {
let weight = u64::from(policy.priority_weight.max(1));
let radius_squared = policy.interest_radius * policy.interest_radius;
let distance_score =
if radius_squared.is_finite() && radius_squared > 0.0 && distance_squared.is_finite() {
let closeness = 1.0 - (distance_squared / radius_squared).clamp(0.0, 1.0);
normalized_score_to_u64(closeness)
} else {
1_000_000
};
weight
.saturating_mul(1_000_000)
.saturating_add(distance_score)
}
}
#[derive(Clone, Copy, Debug, PartialEq)]
struct PrioritizedReplicationCandidate {
handle: EntityHandle,
score: u64,
distance_squared: f32,
}
#[derive(Clone, Debug, Default)]
pub struct ReplicationScratch {
cell_query: CellQueryScratch,
prioritized: Vec<PrioritizedReplicationCandidate>,
}
impl ReplicationScratch {
pub fn clear(&mut self) {
self.cell_query.clear();
self.prioritized.clear();
}
pub fn candidate_count(&self) -> usize {
self.cell_query.len()
}
pub fn prioritized_capacity(&self) -> usize {
self.prioritized.capacity()
}
pub const fn query_stats(&self) -> CellQueryStats {
self.cell_query.stats()
}
pub fn candidate_capacity(&self) -> usize {
self.cell_query.handle_capacity()
}
pub fn candidate_dedup_capacity(&self) -> usize {
self.cell_query.dedup_capacity()
}
pub fn matching_cell_capacity(&self) -> usize {
self.cell_query.matching_cell_capacity()
}
}
#[derive(Clone, Copy, Debug, Default)]
pub struct ReplicationPlanner;
impl ReplicationPlanner {
pub fn plan_for_viewer<F: VisibilityFilter>(
station: &Station,
index: &CellIndex,
policies: &PolicyTable,
viewer: &ViewerQuery,
filter: &F,
budget: ReplicationBudget,
) -> ReplicationPlan {
let candidates = index.query_sphere(viewer.position, viewer.radius);
Self::plan_for_candidates_inner(
station,
&candidates,
policies,
viewer,
filter,
budget,
|_, _, _| true,
)
}
pub fn plan_for_viewer_with_scratch<F: VisibilityFilter>(
station: &Station,
index: &CellIndex,
policies: &PolicyTable,
viewer: &ViewerQuery,
filter: &F,
budget: ReplicationBudget,
scratch: &mut ReplicationScratch,
) -> ReplicationPlan {
let mut plan = ReplicationPlan::default();
Self::plan_for_viewer_with_scratch_into(
station, index, policies, viewer, filter, budget, scratch, &mut plan,
);
plan
}
#[allow(clippy::too_many_arguments)]
pub fn plan_for_viewer_with_scratch_into<F: VisibilityFilter>(
station: &Station,
index: &CellIndex,
policies: &PolicyTable,
viewer: &ViewerQuery,
filter: &F,
budget: ReplicationBudget,
scratch: &mut ReplicationScratch,
plan: &mut ReplicationPlan,
) {
Self::plan_for_viewer_eligible_with_scratch_into(
station,
index,
policies,
viewer,
filter,
budget,
|_, _, _| true,
scratch,
plan,
);
}
#[allow(clippy::too_many_arguments)]
pub fn plan_for_viewer_eligible_with_scratch_into<F, E>(
station: &Station,
index: &CellIndex,
policies: &PolicyTable,
viewer: &ViewerQuery,
filter: &F,
budget: ReplicationBudget,
eligible: E,
scratch: &mut ReplicationScratch,
plan: &mut ReplicationPlan,
) where
F: VisibilityFilter,
E: Fn(&ViewerQuery, EntityHandle, &EntityRecord) -> bool,
{
let candidates =
index.query_sphere_into(viewer.position, viewer.radius, &mut scratch.cell_query);
Self::plan_for_candidates_inner_into(
station,
candidates,
policies,
viewer,
filter,
budget,
|_, _, _| true,
eligible,
false,
plan,
);
}
pub fn plan_for_viewers_with_scratch<F: VisibilityFilter>(
station: &Station,
index: &CellIndex,
policies: &PolicyTable,
viewers: &[ViewerQuery],
filter: &F,
budget: ReplicationBudget,
scratch: &mut ReplicationScratch,
) -> ReplicationBatchResult {
let mut batch = ReplicationBatchResult {
plans: Vec::with_capacity(viewers.len()),
stats: ReplicationBatchStats::default(),
};
for viewer in viewers {
let plan = Self::plan_for_viewer_with_scratch(
station, index, policies, viewer, filter, budget, scratch,
);
batch.stats.record(&plan, scratch);
batch.plans.push(plan);
}
batch
}
#[allow(clippy::too_many_arguments)]
pub fn plan_for_viewers_into<'a, F: VisibilityFilter>(
station: &Station,
index: &CellIndex,
policies: &PolicyTable,
viewers: &[ViewerQuery],
filter: &F,
budget: ReplicationBudget,
scratch: &mut ReplicationScratch,
batch: &'a mut ReplicationBatchScratch,
) -> ReplicationBatchView<'a> {
Self::plan_for_viewers_eligible_into(
station,
index,
policies,
viewers,
filter,
budget,
|_, _, _| true,
scratch,
batch,
)
}
#[allow(clippy::too_many_arguments)]
pub fn plan_for_viewers_eligible_into<'a, F, E>(
station: &Station,
index: &CellIndex,
policies: &PolicyTable,
viewers: &[ViewerQuery],
filter: &F,
budget: ReplicationBudget,
eligible: E,
scratch: &mut ReplicationScratch,
batch: &'a mut ReplicationBatchScratch,
) -> ReplicationBatchView<'a>
where
F: VisibilityFilter,
E: Fn(&ViewerQuery, EntityHandle, &EntityRecord) -> bool,
{
batch.prepare(viewers.len());
for (plan, viewer) in batch.plans[..viewers.len()].iter_mut().zip(viewers) {
Self::plan_for_viewer_eligible_with_scratch_into(
station, index, policies, viewer, filter, budget, &eligible, scratch, plan,
);
batch.stats.record(plan, scratch);
}
batch.view()
}
#[allow(clippy::too_many_arguments)]
pub fn plan_for_viewer_work_bounded_with_scratch_into<F, E>(
station: &Station,
index: &CellIndex,
policies: &PolicyTable,
viewer: &ViewerQuery,
filter: &F,
budget: ReplicationBudget,
eligible: E,
scratch: &mut ReplicationScratch,
plan: &mut ReplicationPlan,
) where
F: VisibilityFilter,
E: Fn(&ViewerQuery, EntityHandle, &EntityRecord) -> bool,
{
let candidates =
index.query_sphere_into(viewer.position, viewer.radius, &mut scratch.cell_query);
Self::plan_for_candidates_inner_into(
station,
candidates,
policies,
viewer,
filter,
budget,
|_, _, _| true,
eligible,
true,
plan,
);
}
#[allow(clippy::too_many_arguments)]
pub fn plan_for_viewers_work_bounded_into<'a, F, E>(
station: &Station,
index: &CellIndex,
policies: &PolicyTable,
viewers: &[ViewerQuery],
filter: &F,
budget: ReplicationBudget,
eligible: E,
scratch: &mut ReplicationScratch,
batch: &'a mut ReplicationBatchScratch,
) -> ReplicationBatchView<'a>
where
F: VisibilityFilter,
E: Fn(&ViewerQuery, EntityHandle, &EntityRecord) -> bool,
{
batch.prepare(viewers.len());
for (plan, viewer) in batch.plans[..viewers.len()].iter_mut().zip(viewers) {
Self::plan_for_viewer_work_bounded_with_scratch_into(
station, index, policies, viewer, filter, budget, &eligible, scratch, plan,
);
batch.stats.record(plan, scratch);
}
batch.view()
}
pub fn plan_for_viewer_range_with_scratch(
station: &Station,
index: &CellIndex,
policies: &PolicyTable,
viewer: &ViewerQuery,
budget: ReplicationBudget,
scratch: &mut ReplicationScratch,
) -> ReplicationPlan {
let mut plan = ReplicationPlan::default();
Self::plan_for_viewer_range_with_scratch_into(
station, index, policies, viewer, budget, scratch, &mut plan,
);
plan
}
#[allow(clippy::too_many_arguments)]
pub fn plan_for_viewer_range_with_scratch_into(
station: &Station,
index: &CellIndex,
policies: &PolicyTable,
viewer: &ViewerQuery,
budget: ReplicationBudget,
scratch: &mut ReplicationScratch,
plan: &mut ReplicationPlan,
) {
let candidates =
index.query_sphere_into(viewer.position, viewer.radius, &mut scratch.cell_query);
Self::plan_for_range_candidates_into(station, candidates, policies, viewer, budget, plan);
}
pub fn plan_for_viewers_range_with_scratch(
station: &Station,
index: &CellIndex,
policies: &PolicyTable,
viewers: &[ViewerQuery],
budget: ReplicationBudget,
scratch: &mut ReplicationScratch,
) -> ReplicationBatchResult {
let mut batch = ReplicationBatchResult {
plans: Vec::with_capacity(viewers.len()),
stats: ReplicationBatchStats::default(),
};
for viewer in viewers {
let plan = Self::plan_for_viewer_range_with_scratch(
station, index, policies, viewer, budget, scratch,
);
batch.stats.record(&plan, scratch);
batch.plans.push(plan);
}
batch
}
#[allow(clippy::too_many_arguments)]
pub fn plan_for_viewers_range_into<'a>(
station: &Station,
index: &CellIndex,
policies: &PolicyTable,
viewers: &[ViewerQuery],
budget: ReplicationBudget,
scratch: &mut ReplicationScratch,
batch: &'a mut ReplicationBatchScratch,
) -> ReplicationBatchView<'a> {
batch.prepare(viewers.len());
for (plan, viewer) in batch.plans[..viewers.len()].iter_mut().zip(viewers) {
Self::plan_for_viewer_range_with_scratch_into(
station, index, policies, viewer, budget, scratch, plan,
);
batch.stats.record(plan, scratch);
}
batch.view()
}
pub fn plan_for_viewer_with_cadence<F, L>(
station: &Station,
index: &CellIndex,
policies: &PolicyTable,
viewer: &ViewerQuery,
filter: &F,
budget: ReplicationBudget,
last_sent: L,
) -> ReplicationPlan
where
F: VisibilityFilter,
L: Fn(EntityHandle) -> Option<Tick>,
{
let tick_rate_hz = station.config().tick_rate_hz;
let now = station.tick();
let candidates = index.query_sphere(viewer.position, viewer.radius);
Self::plan_for_candidates_inner(
station,
&candidates,
policies,
viewer,
filter,
budget,
|handle, policy, distance_squared| {
ReplicationCadence::should_send(
policy,
tick_rate_hz,
distance_squared,
now,
last_sent(handle),
)
},
)
}
#[allow(clippy::too_many_arguments)]
pub fn plan_for_viewer_with_cadence_and_scratch<F, L>(
station: &Station,
index: &CellIndex,
policies: &PolicyTable,
viewer: &ViewerQuery,
filter: &F,
budget: ReplicationBudget,
last_sent: L,
scratch: &mut ReplicationScratch,
) -> ReplicationPlan
where
F: VisibilityFilter,
L: Fn(EntityHandle) -> Option<Tick>,
{
let mut plan = ReplicationPlan::default();
Self::plan_for_viewer_with_cadence_and_scratch_into(
station, index, policies, viewer, filter, budget, last_sent, scratch, &mut plan,
);
plan
}
#[allow(clippy::too_many_arguments)]
pub fn plan_for_viewer_with_cadence_and_scratch_into<F, L>(
station: &Station,
index: &CellIndex,
policies: &PolicyTable,
viewer: &ViewerQuery,
filter: &F,
budget: ReplicationBudget,
last_sent: L,
scratch: &mut ReplicationScratch,
plan: &mut ReplicationPlan,
) where
F: VisibilityFilter,
L: Fn(EntityHandle) -> Option<Tick>,
{
let tick_rate_hz = station.config().tick_rate_hz;
let now = station.tick();
let candidates =
index.query_sphere_into(viewer.position, viewer.radius, &mut scratch.cell_query);
Self::plan_for_candidates_inner_into(
station,
candidates,
policies,
viewer,
filter,
budget,
|handle, policy, distance_squared| {
ReplicationCadence::should_send(
policy,
tick_rate_hz,
distance_squared,
now,
last_sent(handle),
)
},
|_, _, _| true,
false,
plan,
);
}
pub fn plan_for_viewer_prioritized<F: VisibilityFilter>(
station: &Station,
index: &CellIndex,
policies: &PolicyTable,
viewer: &ViewerQuery,
filter: &F,
budget: ReplicationBudget,
) -> ReplicationPlan {
let candidates = index.query_sphere(viewer.position, viewer.radius);
let mut prioritized = Vec::new();
Self::plan_for_candidates_prioritized_inner(
station,
&candidates,
policies,
viewer,
filter,
budget,
&mut prioritized,
|_, _, _| true,
)
}
pub fn plan_for_viewer_prioritized_with_scratch<F: VisibilityFilter>(
station: &Station,
index: &CellIndex,
policies: &PolicyTable,
viewer: &ViewerQuery,
filter: &F,
budget: ReplicationBudget,
scratch: &mut ReplicationScratch,
) -> ReplicationPlan {
let mut plan = ReplicationPlan::default();
Self::plan_for_viewer_prioritized_with_scratch_into(
station, index, policies, viewer, filter, budget, scratch, &mut plan,
);
plan
}
#[allow(clippy::too_many_arguments)]
pub fn plan_for_viewer_prioritized_with_scratch_into<F: VisibilityFilter>(
station: &Station,
index: &CellIndex,
policies: &PolicyTable,
viewer: &ViewerQuery,
filter: &F,
budget: ReplicationBudget,
scratch: &mut ReplicationScratch,
plan: &mut ReplicationPlan,
) {
let candidates =
index.query_sphere_into(viewer.position, viewer.radius, &mut scratch.cell_query);
Self::plan_for_candidates_prioritized_inner_into(
station,
candidates,
policies,
viewer,
filter,
budget,
&mut scratch.prioritized,
|_, _, _| true,
plan,
);
}
pub fn plan_for_viewer_prioritized_with_cadence<F, L>(
station: &Station,
index: &CellIndex,
policies: &PolicyTable,
viewer: &ViewerQuery,
filter: &F,
budget: ReplicationBudget,
last_sent: L,
) -> ReplicationPlan
where
F: VisibilityFilter,
L: Fn(EntityHandle) -> Option<Tick>,
{
let tick_rate_hz = station.config().tick_rate_hz;
let now = station.tick();
let candidates = index.query_sphere(viewer.position, viewer.radius);
let mut prioritized = Vec::new();
Self::plan_for_candidates_prioritized_inner(
station,
&candidates,
policies,
viewer,
filter,
budget,
&mut prioritized,
|handle, policy, distance_squared| {
ReplicationCadence::should_send(
policy,
tick_rate_hz,
distance_squared,
now,
last_sent(handle),
)
},
)
}
#[allow(clippy::too_many_arguments)]
pub fn plan_for_viewer_prioritized_with_cadence_and_scratch<F, L>(
station: &Station,
index: &CellIndex,
policies: &PolicyTable,
viewer: &ViewerQuery,
filter: &F,
budget: ReplicationBudget,
last_sent: L,
scratch: &mut ReplicationScratch,
) -> ReplicationPlan
where
F: VisibilityFilter,
L: Fn(EntityHandle) -> Option<Tick>,
{
let mut plan = ReplicationPlan::default();
Self::plan_for_viewer_prioritized_with_cadence_and_scratch_into(
station, index, policies, viewer, filter, budget, last_sent, scratch, &mut plan,
);
plan
}
#[allow(clippy::too_many_arguments)]
pub fn plan_for_viewer_prioritized_with_cadence_and_scratch_into<F, L>(
station: &Station,
index: &CellIndex,
policies: &PolicyTable,
viewer: &ViewerQuery,
filter: &F,
budget: ReplicationBudget,
last_sent: L,
scratch: &mut ReplicationScratch,
plan: &mut ReplicationPlan,
) where
F: VisibilityFilter,
L: Fn(EntityHandle) -> Option<Tick>,
{
let tick_rate_hz = station.config().tick_rate_hz;
let now = station.tick();
let candidates =
index.query_sphere_into(viewer.position, viewer.radius, &mut scratch.cell_query);
Self::plan_for_candidates_prioritized_inner_into(
station,
candidates,
policies,
viewer,
filter,
budget,
&mut scratch.prioritized,
|handle, policy, distance_squared| {
ReplicationCadence::should_send(
policy,
tick_rate_hz,
distance_squared,
now,
last_sent(handle),
)
},
plan,
);
}
fn plan_for_candidates_inner<F, C>(
station: &Station,
candidates: &[EntityHandle],
policies: &PolicyTable,
viewer: &ViewerQuery,
filter: &F,
budget: ReplicationBudget,
cadence_allows: C,
) -> ReplicationPlan
where
F: VisibilityFilter,
C: Fn(EntityHandle, &CompiledSyncPolicy, f32) -> bool,
{
let mut plan = ReplicationPlan::default();
Self::plan_for_candidates_inner_into(
station,
candidates,
policies,
viewer,
filter,
budget,
cadence_allows,
|_, _, _| true,
false,
&mut plan,
);
plan
}
#[allow(clippy::too_many_arguments)]
fn plan_for_candidates_inner_into<F, C, E>(
station: &Station,
candidates: &[EntityHandle],
policies: &PolicyTable,
viewer: &ViewerQuery,
filter: &F,
budget: ReplicationBudget,
cadence_allows: C,
eligible: E,
stop_when_budget_full: bool,
plan: &mut ReplicationPlan,
) where
F: VisibilityFilter,
C: Fn(EntityHandle, &CompiledSyncPolicy, f32) -> bool,
E: Fn(&ViewerQuery, EntityHandle, &EntityRecord) -> bool,
{
let max_entities = viewer.max_entities.min(budget.max_entities);
let max_by_bytes = budget.max_bytes / budget.estimated_entity_bytes.max(1);
let hard_limit = max_entities.min(max_by_bytes);
plan.entities.clear();
plan.entities.reserve(hard_limit.min(candidates.len()));
plan.stats = ReplicationStats {
candidates: candidates.len(),
..ReplicationStats::default()
};
if stop_when_budget_full && hard_limit == 0 {
plan.stats.unexamined_after_budget = candidates.len();
return;
}
for (candidate_index, handle) in candidates.iter().enumerate() {
let Some(entity) = station.get(*handle) else {
continue;
};
plan.stats.considered += 1;
let Some(policy) = policies.get(entity.policy_id) else {
continue;
};
let distance_squared = entity.position.distance_squared(viewer.position);
let policy_radius_sq = policy.interest_radius * policy.interest_radius;
if distance_squared > policy_radius_sq {
continue;
}
if !filter.is_visible_with_distance(viewer, entity, distance_squared) {
continue;
}
if !cadence_allows(*handle, policy, distance_squared) {
plan.stats.skipped_by_cadence += 1;
continue;
}
if !eligible(viewer, *handle, entity) {
continue;
}
if plan.entities.len() >= hard_limit {
plan.stats.skipped_by_budget += 1;
continue;
}
plan.entities.push(*handle);
if stop_when_budget_full && plan.entities.len() == hard_limit {
plan.stats.unexamined_after_budget =
candidates.len().saturating_sub(candidate_index + 1);
break;
}
}
plan.stats.selected = plan.entities.len();
plan.stats.estimated_bytes = plan.stats.selected * budget.estimated_entity_bytes;
}
#[cfg(all(not(feature = "simd"), test))]
fn plan_for_range_candidates(
station: &Station,
candidates: &[EntityHandle],
policies: &PolicyTable,
viewer: &ViewerQuery,
budget: ReplicationBudget,
) -> ReplicationPlan {
let mut plan = ReplicationPlan::default();
Self::plan_for_range_candidates_into(
station, candidates, policies, viewer, budget, &mut plan,
);
plan
}
#[cfg(not(feature = "simd"))]
fn plan_for_range_candidates_into(
station: &Station,
candidates: &[EntityHandle],
policies: &PolicyTable,
viewer: &ViewerQuery,
budget: ReplicationBudget,
plan: &mut ReplicationPlan,
) {
Self::plan_for_candidates_inner_into(
station,
candidates,
policies,
viewer,
&RangeOnlyVisibility,
budget,
|_, _, _| true,
|_, _, _| true,
false,
plan,
);
}
#[cfg(all(feature = "simd", test))]
fn plan_for_range_candidates(
station: &Station,
candidates: &[EntityHandle],
policies: &PolicyTable,
viewer: &ViewerQuery,
budget: ReplicationBudget,
) -> ReplicationPlan {
let mut plan = ReplicationPlan::default();
Self::plan_for_range_candidates_into(
station, candidates, policies, viewer, budget, &mut plan,
);
plan
}
#[cfg(feature = "simd")]
fn plan_for_range_candidates_into(
station: &Station,
candidates: &[EntityHandle],
policies: &PolicyTable,
viewer: &ViewerQuery,
budget: ReplicationBudget,
plan: &mut ReplicationPlan,
) {
use wide::{CmpLe, f32x8};
const LANES: usize = 8;
let max_entities = viewer.max_entities.min(budget.max_entities);
let max_by_bytes = budget.max_bytes / budget.estimated_entity_bytes.max(1);
let hard_limit = max_entities.min(max_by_bytes);
plan.entities.clear();
plan.entities.reserve(hard_limit.min(candidates.len()));
plan.stats = ReplicationStats {
candidates: candidates.len(),
..ReplicationStats::default()
};
let viewer_radius_squared = viewer.radius_squared();
for handles in candidates.chunks(LANES) {
let mut distance_squared = [f32::NAN; LANES];
let mut policy_radius_squared = [f32::NAN; LANES];
let mut valid_lanes = 0_u8;
for (lane, handle) in handles.iter().copied().enumerate() {
let Some(entity) = station.get(handle) else {
continue;
};
plan.stats.considered = plan.stats.considered.saturating_add(1);
let Some(policy) = policies.get(entity.policy_id) else {
continue;
};
distance_squared[lane] = entity.position.distance_squared(viewer.position);
policy_radius_squared[lane] = policy.interest_radius * policy.interest_radius;
valid_lanes |= 1 << lane;
}
let visible_lanes = u8::try_from(
(f32x8::new(distance_squared).cmp_le(f32x8::new(policy_radius_squared))
& f32x8::new(distance_squared).cmp_le(f32x8::splat(viewer_radius_squared)))
.move_mask(),
)
.expect("eight-lane SIMD mask fits u8")
& valid_lanes;
for (lane, handle) in handles.iter().copied().enumerate() {
if visible_lanes & (1 << lane) == 0 {
continue;
}
if plan.entities.len() >= hard_limit {
plan.stats.skipped_by_budget = plan.stats.skipped_by_budget.saturating_add(1);
} else {
plan.entities.push(handle);
}
}
}
plan.stats.selected = plan.entities.len();
plan.stats.estimated_bytes = plan.stats.selected * budget.estimated_entity_bytes;
}
#[allow(clippy::too_many_arguments)]
fn plan_for_candidates_prioritized_inner<F, C>(
station: &Station,
candidates: &[EntityHandle],
policies: &PolicyTable,
viewer: &ViewerQuery,
filter: &F,
budget: ReplicationBudget,
eligible: &mut Vec<PrioritizedReplicationCandidate>,
cadence_allows: C,
) -> ReplicationPlan
where
F: VisibilityFilter,
C: Fn(EntityHandle, &CompiledSyncPolicy, f32) -> bool,
{
let mut plan = ReplicationPlan::default();
Self::plan_for_candidates_prioritized_inner_into(
station,
candidates,
policies,
viewer,
filter,
budget,
eligible,
cadence_allows,
&mut plan,
);
plan
}
#[allow(clippy::too_many_arguments)]
fn plan_for_candidates_prioritized_inner_into<F, C>(
station: &Station,
candidates: &[EntityHandle],
policies: &PolicyTable,
viewer: &ViewerQuery,
filter: &F,
budget: ReplicationBudget,
eligible: &mut Vec<PrioritizedReplicationCandidate>,
cadence_allows: C,
plan: &mut ReplicationPlan,
) where
F: VisibilityFilter,
C: Fn(EntityHandle, &CompiledSyncPolicy, f32) -> bool,
{
let max_entities = viewer.max_entities.min(budget.max_entities);
let max_by_bytes = budget.max_bytes / budget.estimated_entity_bytes.max(1);
let hard_limit = max_entities.min(max_by_bytes);
plan.entities.clear();
plan.entities.reserve(hard_limit.min(candidates.len()));
plan.stats = ReplicationStats {
candidates: candidates.len(),
..ReplicationStats::default()
};
eligible.clear();
for handle in candidates {
let Some(entity) = station.get(*handle) else {
continue;
};
plan.stats.considered += 1;
let Some(policy) = policies.get(entity.policy_id) else {
continue;
};
let distance_squared = entity.position.distance_squared(viewer.position);
let policy_radius_sq = policy.interest_radius * policy.interest_radius;
if distance_squared > policy_radius_sq {
continue;
}
if !filter.is_visible_with_distance(viewer, entity, distance_squared) {
continue;
}
if !cadence_allows(*handle, policy, distance_squared) {
plan.stats.skipped_by_cadence += 1;
continue;
}
eligible.push(PrioritizedReplicationCandidate {
handle: *handle,
score: ReplicationPriority::score(policy, distance_squared),
distance_squared,
});
}
let selected = prioritize_candidates(eligible, hard_limit);
plan.stats.skipped_by_budget = eligible.len().saturating_sub(selected);
plan.entities.extend(
eligible
.iter()
.take(selected)
.map(|candidate| candidate.handle),
);
plan.stats.selected = plan.entities.len();
plan.stats.estimated_bytes = plan.stats.selected * budget.estimated_entity_bytes;
}
}
fn compare_prioritized_candidates(
left: &PrioritizedReplicationCandidate,
right: &PrioritizedReplicationCandidate,
) -> Ordering {
right
.score
.cmp(&left.score)
.then_with(|| left.distance_squared.total_cmp(&right.distance_squared))
.then_with(|| left.handle.cmp(&right.handle))
}
fn prioritize_candidates(eligible: &mut [PrioritizedReplicationCandidate], limit: usize) -> usize {
let selected = eligible.len().min(limit);
if selected == 0 {
return 0;
}
if selected.saturating_mul(2) < eligible.len() {
eligible.select_nth_unstable_by(selected, compare_prioritized_candidates);
eligible[..selected].sort_by(compare_prioritized_candidates);
} else {
eligible.sort_by(compare_prioritized_candidates);
}
selected
}
#[cfg(test)]
mod tests {
use super::*;
use crate::entity::EntityTags;
use crate::ids::{ClientId, EntityId, InstanceId, NodeId, PolicyId, StationId};
use crate::interest::{AndVisibility, FrustumVisibility, RangeOnlyVisibility, TagVisibility};
use crate::policy::CompiledSyncPolicy;
use crate::spatial::{Aabb3, Bounds, Frustum3, GridSpec, Position3};
use crate::station::{Station, StationConfig};
#[test]
fn planner_applies_composed_frustum_visibility_filter() {
let mut station = Station::new(StationConfig {
station_id: StationId::new(1),
node_id: NodeId::new(1),
instance_id: InstanceId::new(1),
tick_rate_hz: 20,
});
let grid = GridSpec::new(16.0).expect("grid is valid");
let mut index = CellIndex::new(grid);
let mut policies = PolicyTable::default();
policies.set(CompiledSyncPolicy::new(PolicyId::new(1), 1, 20, 128.0));
let visible = station
.spawn_owned(
EntityId::new(1),
Position3::new(10.0, 0.0, 0.0),
Bounds::Point,
PolicyId::new(1),
)
.expect("spawn visible");
let outside_frustum = station
.spawn_owned(
EntityId::new(2),
Position3::new(-10.0, 0.0, 0.0),
Bounds::Point,
PolicyId::new(1),
)
.expect("spawn outside frustum");
index.upsert(visible, Position3::new(10.0, 0.0, 0.0), Bounds::Point);
index.upsert(
outside_frustum,
Position3::new(-10.0, 0.0, 0.0),
Bounds::Point,
);
let viewer = ViewerQuery {
client_id: ClientId::new(7),
position: Position3::new(0.0, 0.0, 0.0),
radius: 128.0,
max_entities: 8,
};
let frustum = Frustum3::from_aabb(Aabb3::new(
Position3::new(0.0, -20.0, -20.0),
Position3::new(80.0, 20.0, 20.0),
));
let filter = AndVisibility::new(RangeOnlyVisibility, FrustumVisibility::new(frustum));
let plan = ReplicationPlanner::plan_for_viewer(
&station,
&index,
&policies,
&viewer,
&filter,
ReplicationBudget::default(),
);
assert_eq!(plan.entities, vec![visible]);
assert_eq!(plan.stats.selected, 1);
assert_eq!(plan.stats.considered, 2);
}
#[test]
fn planner_applies_tag_visibility_filter() {
let mut station = Station::new(StationConfig {
station_id: StationId::new(1),
node_id: NodeId::new(1),
instance_id: InstanceId::new(1),
tick_rate_hz: 20,
});
let grid = GridSpec::new(16.0).expect("grid is valid");
let mut index = CellIndex::new(grid);
let mut policies = PolicyTable::default();
policies.set(CompiledSyncPolicy::new(PolicyId::new(1), 1, 20, 128.0));
let static_visible = station
.spawn_owned(
EntityId::new(1),
Position3::new(10.0, 0.0, 0.0),
Bounds::Point,
PolicyId::new(1),
)
.expect("spawn static");
let fast_mover = station
.spawn_owned(
EntityId::new(2),
Position3::new(12.0, 0.0, 0.0),
Bounds::Point,
PolicyId::new(1),
)
.expect("spawn mover");
station
.set_tags(static_visible, EntityTags::from_bits(0b001))
.expect("tag static");
station
.set_tags(fast_mover, EntityTags::from_bits(0b010))
.expect("tag mover");
index.upsert(
static_visible,
Position3::new(10.0, 0.0, 0.0),
Bounds::Point,
);
index.upsert(fast_mover, Position3::new(12.0, 0.0, 0.0), Bounds::Point);
let viewer = ViewerQuery {
client_id: ClientId::new(7),
position: Position3::new(0.0, 0.0, 0.0),
radius: 128.0,
max_entities: 8,
};
let filter = AndVisibility::new(
RangeOnlyVisibility,
TagVisibility::new(EntityTags::from_bits(0b001), EntityTags::from_bits(0b010)),
);
let plan = ReplicationPlanner::plan_for_viewer(
&station,
&index,
&policies,
&viewer,
&filter,
ReplicationBudget::default(),
);
assert_eq!(plan.entities, vec![static_visible]);
assert_eq!(plan.stats.selected, 1);
assert_eq!(plan.stats.considered, 2);
}
#[test]
fn caller_eligibility_filters_before_replication_budget() {
let mut station = Station::new(StationConfig {
station_id: StationId::new(1),
node_id: NodeId::new(1),
instance_id: InstanceId::new(1),
tick_rate_hz: 20,
});
let mut index = CellIndex::new(GridSpec::new(16.0).expect("grid is valid"));
let mut policies = PolicyTable::default();
policies.set(CompiledSyncPolicy::new(PolicyId::new(1), 1, 20, 128.0));
let handles = (1_u16..=3)
.map(|id| {
let position = Position3::new(f32::from(id), 0.0, 0.0);
let handle = station
.spawn_owned(
EntityId::new(u64::from(id)),
position,
Bounds::Point,
PolicyId::new(1),
)
.expect("entity id is unique");
index.upsert(handle, position, Bounds::Point);
handle
})
.collect::<Vec<_>>();
let viewer = ViewerQuery {
client_id: ClientId::new(1),
position: Position3::new(0.0, 0.0, 0.0),
radius: 128.0,
max_entities: 1,
};
let mut scratch = ReplicationScratch::default();
let mut plan = ReplicationPlan::default();
ReplicationPlanner::plan_for_viewer_eligible_with_scratch_into(
&station,
&index,
&policies,
&viewer,
&RangeOnlyVisibility,
ReplicationBudget {
max_entities: 1,
..ReplicationBudget::default()
},
|_, handle, _| handle == handles[2],
&mut scratch,
&mut plan,
);
assert_eq!(plan.entities, vec![handles[2]]);
assert_eq!(plan.stats.selected, 1);
assert_eq!(plan.stats.skipped_by_budget, 0);
}
#[test]
fn work_bounded_planner_stops_after_first_fit_budget() {
let mut station = Station::new(StationConfig {
station_id: StationId::new(1),
node_id: NodeId::new(1),
instance_id: InstanceId::new(1),
tick_rate_hz: 20,
});
let mut index = CellIndex::new(GridSpec::new(16.0).expect("grid is valid"));
let mut policies = PolicyTable::default();
policies.set(CompiledSyncPolicy::new(PolicyId::new(1), 1, 20, 128.0));
for id in 1_u16..=8 {
let position = Position3::new(f32::from(id), 0.0, 0.0);
let handle = station
.spawn_owned(
EntityId::new(u64::from(id)),
position,
Bounds::Point,
PolicyId::new(1),
)
.expect("entity id is unique");
index.upsert(handle, position, Bounds::Point);
}
let viewer = ViewerQuery {
client_id: ClientId::new(1),
position: Position3::new(0.0, 0.0, 0.0),
radius: 128.0,
max_entities: 2,
};
let mut scratch = ReplicationScratch::default();
let mut plan = ReplicationPlan::default();
ReplicationPlanner::plan_for_viewer_work_bounded_with_scratch_into(
&station,
&index,
&policies,
&viewer,
&RangeOnlyVisibility,
ReplicationBudget {
max_entities: 2,
..ReplicationBudget::default()
},
|_, _, _| true,
&mut scratch,
&mut plan,
);
assert_eq!(plan.stats.selected, 2);
assert_eq!(plan.stats.considered, 2);
assert_eq!(plan.stats.skipped_by_budget, 0);
assert_eq!(plan.stats.unexamined_after_budget, 6);
}
#[test]
#[allow(clippy::too_many_lines)]
fn range_batch_matches_ordered_scalar_plans() {
let mut station = Station::new(StationConfig {
station_id: StationId::new(1),
node_id: NodeId::new(1),
instance_id: InstanceId::new(1),
tick_rate_hz: 128,
});
let grid = GridSpec::new(16.0).expect("grid is valid");
let mut index = CellIndex::new(grid);
let mut policies = PolicyTable::default();
policies.set(CompiledSyncPolicy::new(PolicyId::new(1), 1, 128, 96.0));
for entity_index in 0_u16..24 {
let position = Position3::new(f32::from(entity_index) * 8.0 - 64.0, 0.0, 0.0);
let handle = station
.spawn_owned(
EntityId::new(u64::from(entity_index)),
position,
Bounds::Point,
PolicyId::new(1),
)
.expect("entity id is unique");
index.upsert(handle, position, Bounds::Point);
}
let viewers = [
ViewerQuery {
client_id: ClientId::new(1),
position: Position3::new(0.0, 0.0, 0.0),
radius: 80.0,
max_entities: 32,
},
ViewerQuery {
client_id: ClientId::new(2),
position: Position3::new(48.0, 0.0, 0.0),
radius: 48.0,
max_entities: 8,
},
];
let mut scalar_scratch = ReplicationScratch::default();
let expected = viewers
.iter()
.map(|viewer| {
ReplicationPlanner::plan_for_viewer_with_scratch(
&station,
&index,
&policies,
viewer,
&RangeOnlyVisibility,
ReplicationBudget::default(),
&mut scalar_scratch,
)
})
.collect::<Vec<_>>();
let mut batch_scratch = ReplicationScratch::default();
let batch = ReplicationPlanner::plan_for_viewers_range_with_scratch(
&station,
&index,
&policies,
&viewers,
ReplicationBudget::default(),
&mut batch_scratch,
);
assert_eq!(batch.plans, expected);
assert_eq!(batch.stats.viewers, viewers.len());
assert_eq!(
batch.stats.selected,
expected.iter().map(|plan| plan.stats.selected).sum()
);
assert_eq!(
batch.stats.grid_queries + batch.stats.occupied_queries,
viewers.len()
);
let mut reusable_planning = ReplicationScratch::default();
let mut reusable_output = ReplicationBatchScratch::new();
{
let reused = ReplicationPlanner::plan_for_viewers_range_into(
&station,
&index,
&policies,
&viewers,
ReplicationBudget::default(),
&mut reusable_planning,
&mut reusable_output,
);
assert_eq!(reused.plans, expected);
assert_eq!(reused.stats, batch.stats);
}
let retained_capacity = reusable_output.retained_entity_capacity();
assert_eq!(reusable_output.retained_plan_slots(), viewers.len());
{
let reused = ReplicationPlanner::plan_for_viewers_into(
&station,
&index,
&policies,
&viewers,
&RangeOnlyVisibility,
ReplicationBudget::default(),
&mut reusable_planning,
&mut reusable_output,
);
assert_eq!(reused.plans, expected);
assert_eq!(reused.stats, batch.stats);
}
let reused = ReplicationPlanner::plan_for_viewers_range_into(
&station,
&index,
&policies,
&viewers[..1],
ReplicationBudget::default(),
&mut reusable_planning,
&mut reusable_output,
);
assert_eq!(reused.plans, &expected[..1]);
assert_eq!(reusable_output.retained_plan_slots(), viewers.len());
assert_eq!(
reusable_output.retained_entity_capacity(),
retained_capacity
);
}
#[test]
fn range_batch_preserves_scalar_nan_radius_semantics() {
let mut station = Station::new(StationConfig {
station_id: StationId::new(1),
node_id: NodeId::new(1),
instance_id: InstanceId::new(1),
tick_rate_hz: 128,
});
let mut policies = PolicyTable::default();
policies.set(CompiledSyncPolicy::new(PolicyId::new(1), 1, 128, 96.0));
let handle = station
.spawn_owned(
EntityId::new(1),
Position3::new(1.0, 2.0, 3.0),
Bounds::Point,
PolicyId::new(1),
)
.expect("spawn entity");
let viewer = ViewerQuery {
client_id: ClientId::new(1),
position: Position3::new(0.0, 0.0, 0.0),
radius: f32::NAN,
max_entities: 8,
};
let candidates = [handle];
let scalar = ReplicationPlanner::plan_for_candidates_inner(
&station,
&candidates,
&policies,
&viewer,
&RangeOnlyVisibility,
ReplicationBudget::default(),
|_, _, _| true,
);
let range = ReplicationPlanner::plan_for_range_candidates(
&station,
&candidates,
&policies,
&viewer,
ReplicationBudget::default(),
);
assert!(scalar.entities.is_empty());
assert_eq!(range, scalar);
}
#[test]
fn cadence_scales_interval_by_squared_distance() {
let policy = CompiledSyncPolicy::new(PolicyId::new(1), 2, 20, 100.0);
assert_eq!(ReplicationCadence::target_hz(&policy, 0.0), 20);
assert_eq!(ReplicationCadence::interval_ticks(&policy, 20, 0.0), 1);
assert_eq!(ReplicationCadence::target_hz(&policy, 100.0_f32 * 100.0), 2);
assert_eq!(
ReplicationCadence::interval_ticks(&policy, 20, 100.0_f32 * 100.0),
10
);
}
#[test]
fn priority_score_prefers_weight_then_distance() {
let mut low = CompiledSyncPolicy::new(PolicyId::new(1), 1, 20, 100.0);
low.priority_weight = 1;
let mut high = CompiledSyncPolicy::new(PolicyId::new(2), 1, 20, 100.0);
high.priority_weight = 10;
assert!(
ReplicationPriority::score(&high, 90.0 * 90.0) > ReplicationPriority::score(&low, 0.0)
);
assert!(
ReplicationPriority::score(&low, 0.0) > ReplicationPriority::score(&low, 90.0 * 90.0)
);
}
#[test]
fn top_k_priority_selection_matches_full_sort_for_all_budget_edges() {
let candidates = (0_u32..257)
.map(|index| PrioritizedReplicationCandidate {
handle: EntityHandle::new(index, index % 3),
score: u64::from(index.wrapping_mul(37) % 23),
distance_squared: f32::from(
u16::try_from(index.wrapping_mul(19) % 41).expect("distance fits u16"),
),
})
.collect::<Vec<_>>();
for limit in [0, 1, 7, 64, 256, 257, 300] {
let mut expected = candidates.clone();
expected.sort_by(compare_prioritized_candidates);
expected.truncate(limit.min(expected.len()));
let mut actual = candidates.clone();
let selected = prioritize_candidates(&mut actual, limit);
assert_eq!(selected, expected.len());
assert_eq!(&actual[..selected], expected.as_slice());
}
}
#[test]
fn planner_with_cadence_skips_recent_far_entities() {
let mut station = Station::new(StationConfig {
station_id: StationId::new(1),
node_id: NodeId::new(1),
instance_id: InstanceId::new(1),
tick_rate_hz: 20,
});
for _ in 0..10 {
station.advance_tick();
}
let grid = GridSpec::new(16.0).expect("grid is valid");
let mut index = CellIndex::new(grid);
let mut policies = PolicyTable::default();
policies.set(CompiledSyncPolicy::new(PolicyId::new(1), 2, 20, 128.0));
let near = station
.spawn_owned(
EntityId::new(1),
Position3::new(0.0, 0.0, 0.0),
Bounds::Point,
PolicyId::new(1),
)
.expect("spawn near");
let far = station
.spawn_owned(
EntityId::new(2),
Position3::new(120.0, 0.0, 0.0),
Bounds::Point,
PolicyId::new(1),
)
.expect("spawn far");
index.upsert(near, Position3::new(0.0, 0.0, 0.0), Bounds::Point);
index.upsert(far, Position3::new(120.0, 0.0, 0.0), Bounds::Point);
let viewer = ViewerQuery {
client_id: ClientId::new(7),
position: Position3::new(0.0, 0.0, 0.0),
radius: 128.0,
max_entities: 8,
};
let plan = ReplicationPlanner::plan_for_viewer_with_cadence(
&station,
&index,
&policies,
&viewer,
&RangeOnlyVisibility,
ReplicationBudget::default(),
|_| Some(Tick::new(9)),
);
assert_eq!(plan.entities, vec![near]);
assert_eq!(plan.stats.selected, 1);
assert_eq!(plan.stats.skipped_by_cadence, 1);
let mut scratch = ReplicationScratch::default();
let mut reusable = ReplicationPlan::default();
ReplicationPlanner::plan_for_viewer_with_cadence_and_scratch_into(
&station,
&index,
&policies,
&viewer,
&RangeOnlyVisibility,
ReplicationBudget::default(),
|_| Some(Tick::new(9)),
&mut scratch,
&mut reusable,
);
assert_eq!(reusable, plan);
}
#[test]
fn prioritized_planner_uses_policy_weight_under_budget() {
let mut station = Station::new(StationConfig {
station_id: StationId::new(1),
node_id: NodeId::new(1),
instance_id: InstanceId::new(1),
tick_rate_hz: 20,
});
let grid = GridSpec::new(16.0).expect("grid is valid");
let mut index = CellIndex::new(grid);
let mut policies = PolicyTable::default();
let mut low = CompiledSyncPolicy::new(PolicyId::new(1), 1, 20, 128.0);
low.priority_weight = 1;
let mut high = CompiledSyncPolicy::new(PolicyId::new(2), 1, 20, 128.0);
high.priority_weight = 10;
policies.set(low);
policies.set(high);
let near_low = station
.spawn_owned(
EntityId::new(1),
Position3::new(0.0, 0.0, 0.0),
Bounds::Point,
PolicyId::new(1),
)
.expect("spawn near low priority");
let far_high = station
.spawn_owned(
EntityId::new(2),
Position3::new(96.0, 0.0, 0.0),
Bounds::Point,
PolicyId::new(2),
)
.expect("spawn far high priority");
index.upsert(near_low, Position3::new(0.0, 0.0, 0.0), Bounds::Point);
index.upsert(far_high, Position3::new(96.0, 0.0, 0.0), Bounds::Point);
let viewer = ViewerQuery {
client_id: ClientId::new(7),
position: Position3::new(0.0, 0.0, 0.0),
radius: 128.0,
max_entities: 1,
};
let plan = ReplicationPlanner::plan_for_viewer_prioritized(
&station,
&index,
&policies,
&viewer,
&RangeOnlyVisibility,
ReplicationBudget {
max_entities: 1,
max_bytes: 32,
estimated_entity_bytes: 32,
},
);
assert_eq!(plan.entities, vec![far_high]);
assert_eq!(plan.stats.selected, 1);
assert_eq!(plan.stats.skipped_by_budget, 1);
let mut scratch = ReplicationScratch::default();
let scratch_plan = ReplicationPlanner::plan_for_viewer_prioritized_with_scratch(
&station,
&index,
&policies,
&viewer,
&RangeOnlyVisibility,
ReplicationBudget {
max_entities: 1,
max_bytes: 32,
estimated_entity_bytes: 32,
},
&mut scratch,
);
assert_eq!(scratch_plan.entities, plan.entities);
assert_eq!(scratch_plan.stats, plan.stats);
assert_eq!(scratch.candidate_count(), 2);
assert!(scratch.prioritized_capacity() >= 2);
assert_eq!(scratch.query_stats().candidate_handles, 2);
assert!(scratch.candidate_capacity() >= 2);
assert!(scratch.candidate_dedup_capacity() >= 2);
let budget = ReplicationBudget {
max_entities: 1,
max_bytes: 32,
estimated_entity_bytes: 32,
};
assert_prioritized_output_reuse(
&station,
&index,
&policies,
&viewer,
budget,
&plan,
&mut scratch,
);
}
fn assert_prioritized_output_reuse(
station: &Station,
index: &CellIndex,
policies: &PolicyTable,
viewer: &ViewerQuery,
budget: ReplicationBudget,
expected: &ReplicationPlan,
scratch: &mut ReplicationScratch,
) {
let mut reusable = ReplicationPlan::default();
ReplicationPlanner::plan_for_viewer_prioritized_with_scratch_into(
station,
index,
policies,
viewer,
&RangeOnlyVisibility,
budget,
scratch,
&mut reusable,
);
let retained_entities = reusable.entities.as_ptr();
assert_eq!(&reusable, expected);
ReplicationPlanner::plan_for_viewer_prioritized_with_cadence_and_scratch_into(
station,
index,
policies,
viewer,
&RangeOnlyVisibility,
budget,
|_| None,
scratch,
&mut reusable,
);
assert_eq!(reusable.entities.as_ptr(), retained_entities);
}
#[test]
fn replication_tracker_records_sent_ack_and_prune() {
let client_id = ClientId::new(7);
let first = EntityHandle::new(1, 0);
let second = EntityHandle::new(2, 0);
let plan = ReplicationPlan {
entities: vec![first, second],
stats: ReplicationStats::default(),
};
let mut tracker = ReplicationTracker::new(ReplicationTrackerConfig { max_entries: 4 });
let recorded = tracker
.record_plan_sent(client_id, &plan, Tick::new(10))
.expect("recording should fit");
assert_eq!(recorded, 2);
assert_eq!(tracker.last_sent(client_id, first), Some(Tick::new(10)));
assert_eq!(tracker.stats().entries, 2);
assert_eq!(tracker.stats().sent_records, 2);
assert!(tracker.acknowledge(client_id, first, Tick::new(11)));
assert_eq!(
tracker
.get(client_id, first)
.expect("tracked record")
.last_acked,
Some(Tick::new(11))
);
assert_eq!(tracker.stats().acked_records, 1);
assert_eq!(tracker.prune_sent_before(Tick::new(11)), 2);
assert!(tracker.is_empty());
assert_eq!(tracker.stats().pruned_records, 2);
}
#[test]
fn replication_tracker_rejects_capacity_without_partial_insert() {
let client_id = ClientId::new(7);
let plan = ReplicationPlan {
entities: vec![EntityHandle::new(1, 0), EntityHandle::new(2, 0)],
stats: ReplicationStats::default(),
};
let mut tracker = ReplicationTracker::new(ReplicationTrackerConfig { max_entries: 1 });
let error = tracker
.record_plan_sent(client_id, &plan, Tick::new(10))
.expect_err("recording should exceed capacity");
assert_eq!(
error,
ReplicationTrackerError::CapacityExceeded {
current: 0,
needed: 2,
max: 1,
}
);
assert!(tracker.is_empty());
assert_eq!(tracker.stats().sent_records, 0);
}
#[test]
fn replication_tracker_uses_exact_capacity_check_near_limit() {
let client_id = ClientId::new(7);
let first = EntityHandle::new(1, 0);
let second = EntityHandle::new(2, 0);
let third = EntityHandle::new(3, 0);
let mut tracker = ReplicationTracker::new(ReplicationTrackerConfig { max_entries: 2 });
tracker
.record_plan_sent(
client_id,
&ReplicationPlan {
entities: vec![first],
stats: ReplicationStats::default(),
},
Tick::new(1),
)
.expect("initial record should fit");
tracker
.record_plan_sent(
client_id,
&ReplicationPlan {
entities: vec![first, second],
stats: ReplicationStats::default(),
},
Tick::new(2),
)
.expect("one existing and one new record should fit exactly");
let error = tracker
.record_plan_sent(
client_id,
&ReplicationPlan {
entities: vec![first, second, third],
stats: ReplicationStats::default(),
},
Tick::new(3),
)
.expect_err("new record should exceed exact capacity");
assert_eq!(
error,
ReplicationTrackerError::CapacityExceeded {
current: 2,
needed: 1,
max: 2,
}
);
assert_eq!(tracker.last_sent(client_id, first), Some(Tick::new(2)));
assert_eq!(tracker.last_sent(client_id, second), Some(Tick::new(2)));
assert_eq!(tracker.get(client_id, third), None);
}
#[test]
fn replication_tracker_promotes_without_losing_ack_or_prune_state() {
let first_client = ClientId::new(1);
let second_client = ClientId::new(2);
let mut tracker = ReplicationTracker::new(ReplicationTrackerConfig {
max_entries: HASHED_REPLICATION_TRACKER_MIN_ENTRIES + 1,
});
let initial_entities: Vec<_> = (0..HASHED_REPLICATION_TRACKER_MIN_ENTRIES - 2)
.map(|index| {
EntityHandle::new(u32::try_from(index).expect("test entity index fits u32"), 1)
})
.collect();
tracker
.record_plan_sent(
first_client,
&ReplicationPlan {
entities: initial_entities.clone(),
stats: ReplicationStats::default(),
},
Tick::new(1),
)
.expect("initial records should fit");
let second_entity = EntityHandle::new(u32::MAX, 1);
tracker
.record_plan_sent(
second_client,
&ReplicationPlan {
entities: vec![second_entity],
stats: ReplicationStats::default(),
},
Tick::new(1),
)
.expect("second client record should fit");
assert!(!tracker.records.is_hashed());
let first_entity = initial_entities[0];
tracker
.record_plan_sent(
first_client,
&ReplicationPlan {
entities: vec![first_entity],
stats: ReplicationStats::default(),
},
Tick::new(2),
)
.expect("existing record should update without promotion");
assert!(!tracker.records.is_hashed());
let final_entity = EntityHandle::new(
u32::try_from(HASHED_REPLICATION_TRACKER_MIN_ENTRIES)
.expect("test entity index fits u32"),
1,
);
tracker
.record_plan_sent(
first_client,
&ReplicationPlan {
entities: vec![final_entity],
stats: ReplicationStats::default(),
},
Tick::new(2),
)
.expect("threshold record should promote");
assert!(tracker.records.is_hashed());
assert!(tracker.acknowledge(first_client, final_entity, Tick::new(3)));
assert_eq!(tracker.clear_client(second_client), 1);
assert!(tracker.records.is_hashed());
assert_eq!(
tracker.prune_sent_before(Tick::new(2)),
initial_entities.len() - 1
);
assert!(tracker.records.is_hashed());
assert_eq!(tracker.len(), 2);
assert_eq!(tracker.stats().entries, 2);
assert_eq!(tracker.stats().acked_records, 1);
assert_eq!(
tracker
.get(first_client, final_entity)
.expect("acked record should survive")
.last_acked,
Some(Tick::new(3))
);
}
}