use linkrs_core::types::{EdgeId, Timestamp};
use super::super::MutableCsrTrait;
use super::{group_id_for, region_local_range, CsrShardSet};
pub const REGION_MERGE_MIN_DENSITY: f32 = 0.4;
pub const GROUP_MERGE_MIN_DENSITY: f32 = 0.65;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RegionMergeScope {
Row,
Region,
Group,
}
impl CsrShardSet {
pub fn mark_reclaim_hint_for(&mut self, vid: u32) {
let gid = group_id_for(vid, self.group_bits);
if let Some(shard) = self.shards.get_mut(&gid) {
shard.reclaim_hint = true;
}
}
pub fn group_needs_reclaim_scan(&self, gid: usize) -> bool {
self.shards
.get(&gid)
.map(|shard| shard.reclaim_hint)
.unwrap_or(false)
}
pub fn group_dead_entries(&self, gid: usize) -> usize {
self.shards
.get(&gid)
.map(|shard| shard.dead_entries)
.unwrap_or(0)
}
pub fn clear_reclaim_hint(&mut self, gid: usize) {
if let Some(shard) = self.shards.get_mut(&gid) {
shard.reclaim_hint = false;
}
}
fn note_group_reclaimed(&mut self, gid: usize, removed: usize) {
if removed == 0 {
return;
}
if let Some(shard) = self.shards.get_mut(&gid) {
shard.dead_entries = shard.dead_entries.saturating_sub(removed);
if shard.dead_entries == 0 {
shard.reclaim_hint = false;
}
}
}
pub fn region_census(&self, gid: usize, region: usize) -> (usize, usize, usize) {
let Some(shard) = self.shards.get(&gid) else {
return (0, 0, 0);
};
let (start, end) = region_local_range(region, self.group_size());
let mut live = 0usize;
let mut dead = 0usize;
let mut capacity = 0usize;
for local in start..end {
let (l, d, c) = shard.variant.vertex_census(local);
live += l;
dead += d;
capacity += c;
}
(live, dead, capacity)
}
pub fn region_density(&self, gid: usize, region: usize) -> f32 {
let (live, _, capacity) = self.region_census(gid, region);
if capacity == 0 {
1.0
} else {
live as f32 / capacity as f32
}
}
pub fn region_live(&self, gid: usize, region: usize) -> usize {
self.region_census(gid, region).0
}
pub fn region_gap(&self, gid: usize, region: usize) -> usize {
let (live, _, capacity) = self.region_census(gid, region);
capacity.saturating_sub(live)
}
pub fn region_reclaimable_count(&self, gid: usize, region: usize, cutoff: Timestamp) -> usize {
if cutoff == Timestamp::MAX {
return 0;
}
let Some(shard) = self.shards.get(&gid) else {
return 0;
};
if shard.dead_entries == 0 {
return 0;
}
let (start, end) = region_local_range(region, self.group_size());
(start..end)
.map(|local| shard.variant.reclaimable_count(local, cutoff))
.sum()
}
pub fn select_merge_scope(
&self,
gid: usize,
region: usize,
cutoff: Timestamp,
region_min_density: f32,
group_min_density: f32,
) -> Option<RegionMergeScope> {
if self.region_reclaimable_count(gid, region, cutoff) == 0
&& !self.region_needs_rebalance(gid, region)
{
return None;
}
let region_density = self.region_density(gid, region);
if region_density < region_min_density {
return Some(RegionMergeScope::Row);
}
let group_live = self
.group_stats(gid)
.map(|stats| {
if stats.capacity == 0 {
1.0
} else {
stats.live_edges as f32 / stats.capacity as f32
}
})
.unwrap_or(1.0);
if self.dirty_region_ids(gid).len() > 1 && group_live >= group_min_density {
return Some(RegionMergeScope::Group);
}
Some(RegionMergeScope::Region)
}
pub fn compact_region_with_reporting(
&mut self,
gid: usize,
region: usize,
cutoff: Timestamp,
on_edge_removed: &mut dyn FnMut(EdgeId, Timestamp),
) -> usize {
if self.is_frozen(gid) {
return self.compact_group_with_reporting(gid, cutoff, 0.0, on_edge_removed);
}
let group_size = self.group_size();
let (start, end) = region_local_range(region, group_size);
let Some(shard) = self.shards.get_mut(&gid) else {
return 0;
};
let mut removed = 0usize;
for local in start..end {
removed += shard
.variant
.compact_vertex_with_reporting(local, cutoff, on_edge_removed);
if shard.variant.row_gap(local) == 0 {
shard.variant.rebalance_row(local);
}
}
if removed > 0 {
shard.dirty.deleted = true;
if let Some(slot) = shard.regions.get_mut(region) {
slot.deleted = true;
}
shard.dead_entries = shard.dead_entries.saturating_sub(removed);
if shard.dead_entries == 0 {
shard.reclaim_hint = false;
}
}
removed
}
pub fn compact_frozen_rows_batched(
&mut self,
gid: usize,
locals: &[u32],
cutoff: Timestamp,
on_edge_removed: &mut dyn FnMut(EdgeId, Timestamp),
) -> usize {
let removed = match self.shards.get_mut(&gid) {
Some(shard) => {
let removed =
shard
.variant
.compact_frozen_rows_batched(locals, cutoff, on_edge_removed);
if removed > 0 {
shard.dirty.deleted = true;
for region in shard.regions.iter_mut() {
region.deleted = true;
}
}
removed
}
None => return 0,
};
self.note_group_reclaimed(gid, removed);
removed
}
pub fn compact_group_with_reporting(
&mut self,
gid: usize,
cutoff: Timestamp,
reserve_ratio: f32,
on_edge_removed: &mut dyn FnMut(EdgeId, Timestamp),
) -> usize {
let removed = match self.shards.get_mut(&gid) {
Some(shard) => {
let removed =
shard
.variant
.compact_with_ts_reporting(cutoff, reserve_ratio, on_edge_removed);
if removed > 0 {
shard.dirty.deleted = true;
for region in shard.regions.iter_mut() {
region.deleted = true;
}
}
removed
}
None => return 0,
};
self.note_group_reclaimed(gid, removed);
removed
}
pub fn compact_with_ts_reporting(
&mut self,
cutoff: Timestamp,
reserve_ratio: f32,
on_edge_removed: &mut dyn FnMut(EdgeId, Timestamp),
) -> usize {
let mut removed = 0usize;
for shard in self.shards.values_mut() {
let before = removed;
removed +=
shard
.variant
.compact_with_ts_reporting(cutoff, reserve_ratio, on_edge_removed);
let delta = removed - before;
if delta > 0 {
shard.dirty.deleted = true;
for region in shard.regions.iter_mut() {
region.deleted = true;
}
shard.dead_entries = shard.dead_entries.saturating_sub(delta);
if shard.dead_entries == 0 {
shard.reclaim_hint = false;
}
}
}
removed
}
}