use std::collections::{HashMap, VecDeque};
pub type BmpBlockId = [u8; 32];
pub type BmpPlanId = u64;
pub type BmpBlockMigrationPlanner = BlockMigrationPlanner;
#[inline]
fn xorshift64(state: &mut u64) -> u64 {
let mut x = *state;
x ^= x << 13;
x ^= x >> 7;
x ^= x << 17;
*state = x;
x
}
#[inline]
fn fnv1a_64(data: &[u8]) -> u64 {
let mut h: u64 = 14_695_981_039_346_656_037;
for &b in data {
h ^= b as u64;
h = h.wrapping_mul(1_099_511_628_211);
}
h
}
fn now_ts() -> u64 {
use std::time::{SystemTime, UNIX_EPOCH};
SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|d| d.as_nanos() as u64)
.unwrap_or(1)
}
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
pub enum BmpPlanStatus {
Pending,
InProgress,
Completed,
Failed,
Cancelled,
}
impl std::fmt::Display for BmpPlanStatus {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::Pending => write!(f, "Pending"),
Self::InProgress => write!(f, "InProgress"),
Self::Completed => write!(f, "Completed"),
Self::Failed => write!(f, "Failed"),
Self::Cancelled => write!(f, "Cancelled"),
}
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
pub enum BmpPriorityPolicy {
FifoQueue,
HighFrequencyFirst,
LargestFirst,
SmallestFirst,
CostOptimized,
}
impl std::fmt::Display for BmpPriorityPolicy {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::FifoQueue => write!(f, "FifoQueue"),
Self::HighFrequencyFirst => write!(f, "HighFrequencyFirst"),
Self::LargestFirst => write!(f, "LargestFirst"),
Self::SmallestFirst => write!(f, "SmallestFirst"),
Self::CostOptimized => write!(f, "CostOptimized"),
}
}
}
#[derive(Clone, Debug)]
pub struct BmpBlockMeta {
pub id: BmpBlockId,
pub size_bytes: u64,
pub src_node: String,
pub dst_node: Option<String>,
pub tier: u8,
pub access_frequency: f64,
pub last_accessed: u64,
pub is_pinned: bool,
}
impl BmpBlockMeta {
pub fn new(
id: BmpBlockId,
size_bytes: u64,
src_node: impl Into<String>,
tier: u8,
access_frequency: f64,
) -> Self {
Self {
id,
size_bytes,
src_node: src_node.into(),
dst_node: None,
tier,
access_frequency,
last_accessed: now_ts(),
is_pinned: false,
}
}
}
#[derive(Clone, Debug)]
pub struct BmpMigrationPlan {
pub id: BmpPlanId,
pub block_ids: Vec<BmpBlockId>,
pub src_node: String,
pub dst_node: String,
pub priority: u32,
pub estimated_bytes: u64,
pub status: BmpPlanStatus,
pub created_at: u64,
}
#[derive(Clone, Debug)]
pub struct BmpMigrationRecord {
pub ts: u64,
pub plan_id: BmpPlanId,
pub block_id: BmpBlockId,
pub bytes_moved: u64,
pub success: bool,
pub error: Option<String>,
}
#[derive(Clone, Debug)]
pub struct BmpPlannerConfig {
pub max_concurrent_migrations: usize,
pub bandwidth_limit_kbps: u64,
pub priority_policy: BmpPriorityPolicy,
pub dry_run: bool,
}
impl Default for BmpPlannerConfig {
fn default() -> Self {
Self {
max_concurrent_migrations: 4,
bandwidth_limit_kbps: 0,
priority_policy: BmpPriorityPolicy::FifoQueue,
dry_run: false,
}
}
}
#[derive(Clone, Debug, Default)]
pub struct BmpPlannerStats {
pub total_blocks: usize,
pub pinned_blocks: usize,
pub total_plans: usize,
pub pending_plans: usize,
pub in_progress_plans: usize,
pub completed_plans: usize,
pub failed_plans: usize,
pub cancelled_plans: usize,
pub total_records: usize,
pub total_bytes_moved: u64,
pub total_failures: usize,
}
#[derive(Clone, Debug)]
pub struct BmpExecutionResult {
pub plan_id: BmpPlanId,
pub status: BmpPlanStatus,
pub succeeded_blocks: Vec<BmpBlockId>,
pub failed_blocks: Vec<(BmpBlockId, String)>,
pub bytes_moved: u64,
}
#[derive(Clone, Debug, Default)]
pub struct BmpBatchResult {
pub plans_executed: usize,
pub plans_completed: usize,
pub plans_failed: usize,
pub total_bytes_moved: u64,
pub results: Vec<BmpExecutionResult>,
}
#[derive(Debug, thiserror::Error)]
pub enum BmpError {
#[error("block {0} is not registered")]
BlockNotFound(String),
#[error("plan {0} does not exist")]
PlanNotFound(BmpPlanId),
#[error("plan {0} is not in Pending state (current: {1})")]
PlanNotPending(BmpPlanId, BmpPlanStatus),
#[error("plan has no blocks")]
EmptyPlan,
#[error("block {0} is pinned and cannot be migrated")]
BlockPinned(String),
#[error("destination node is empty")]
EmptyDstNode,
#[error("no eligible blocks found for scheduling")]
NoEligibleBlocks,
}
fn block_id_hex(id: &BmpBlockId) -> String {
let mut s = String::with_capacity(64);
for b in id {
s.push_str(&format!("{:02x}", b));
}
s
}
pub struct BlockMigrationPlanner {
blocks: HashMap<BmpBlockId, BmpBlockMeta>,
plans: HashMap<BmpPlanId, BmpMigrationPlan>,
log: VecDeque<BmpMigrationRecord>,
config: BmpPlannerConfig,
next_plan_id: BmpPlanId,
rng_state: u64,
}
impl BlockMigrationPlanner {
pub fn new() -> Self {
Self::with_config(BmpPlannerConfig::default())
}
pub fn with_config(config: BmpPlannerConfig) -> Self {
let seed_data = b"block_migration_planner_v1";
let ts = now_ts();
let base = fnv1a_64(seed_data);
let mut rng_state = base ^ ts;
if rng_state == 0 {
rng_state = 0xdeadbeef_cafebabe;
}
Self {
blocks: HashMap::new(),
plans: HashMap::new(),
log: VecDeque::new(),
config,
next_plan_id: 1,
rng_state,
}
}
pub fn config(&self) -> &BmpPlannerConfig {
&self.config
}
pub fn set_config(&mut self, config: BmpPlannerConfig) {
self.config = config;
}
pub fn register_block(
&mut self,
id: BmpBlockId,
size_bytes: u64,
src_node: impl Into<String>,
tier: u8,
access_frequency: f64,
) {
let meta = BmpBlockMeta::new(id, size_bytes, src_node, tier, access_frequency);
self.blocks.insert(id, meta);
}
pub fn update_access(&mut self, id: &BmpBlockId) -> Result<(), BmpError> {
let meta = self
.blocks
.get_mut(id)
.ok_or_else(|| BmpError::BlockNotFound(block_id_hex(id)))?;
meta.last_accessed = now_ts();
meta.access_frequency = meta.access_frequency * 0.9 + 1.0 * 0.1;
Ok(())
}
pub fn pin(&mut self, id: &BmpBlockId) -> Result<(), BmpError> {
let meta = self
.blocks
.get_mut(id)
.ok_or_else(|| BmpError::BlockNotFound(block_id_hex(id)))?;
meta.is_pinned = true;
Ok(())
}
pub fn unpin(&mut self, id: &BmpBlockId) -> Result<(), BmpError> {
let meta = self
.blocks
.get_mut(id)
.ok_or_else(|| BmpError::BlockNotFound(block_id_hex(id)))?;
meta.is_pinned = false;
Ok(())
}
pub fn block_meta(&self, id: &BmpBlockId) -> Option<&BmpBlockMeta> {
self.blocks.get(id)
}
pub fn block_count(&self) -> usize {
self.blocks.len()
}
pub fn create_plan(
&mut self,
block_ids: Vec<BmpBlockId>,
dst_node: impl Into<String>,
priority: u32,
) -> Result<BmpPlanId, BmpError> {
if block_ids.is_empty() {
return Err(BmpError::EmptyPlan);
}
let dst_node = dst_node.into();
if dst_node.is_empty() {
return Err(BmpError::EmptyDstNode);
}
for bid in &block_ids {
let meta = self
.blocks
.get(bid)
.ok_or_else(|| BmpError::BlockNotFound(block_id_hex(bid)))?;
if meta.is_pinned {
return Err(BmpError::BlockPinned(block_id_hex(bid)));
}
}
let src_node = self
.blocks
.get(&block_ids[0])
.map(|m| m.src_node.clone())
.unwrap_or_default();
let estimated_bytes: u64 = block_ids
.iter()
.filter_map(|bid| self.blocks.get(bid).map(|m| m.size_bytes))
.sum();
let plan_id = self.next_plan_id;
self.next_plan_id += 1;
for bid in &block_ids {
if let Some(meta) = self.blocks.get_mut(bid) {
meta.dst_node = Some(dst_node.clone());
}
}
let plan = BmpMigrationPlan {
id: plan_id,
block_ids,
src_node,
dst_node,
priority,
estimated_bytes,
status: BmpPlanStatus::Pending,
created_at: now_ts(),
};
self.plans.insert(plan_id, plan);
Ok(plan_id)
}
pub fn cancel_plan(&mut self, plan_id: BmpPlanId) -> Result<(), BmpError> {
let plan = self
.plans
.get_mut(&plan_id)
.ok_or(BmpError::PlanNotFound(plan_id))?;
if plan.status != BmpPlanStatus::Pending {
return Err(BmpError::PlanNotPending(plan_id, plan.status));
}
plan.status = BmpPlanStatus::Cancelled;
Ok(())
}
pub fn execute_plan(&mut self, plan_id: BmpPlanId) -> Result<BmpExecutionResult, BmpError> {
{
let plan = self
.plans
.get(&plan_id)
.ok_or(BmpError::PlanNotFound(plan_id))?;
if plan.status != BmpPlanStatus::Pending {
return Err(BmpError::PlanNotPending(plan_id, plan.status));
}
}
if let Some(plan) = self.plans.get_mut(&plan_id) {
plan.status = BmpPlanStatus::InProgress;
}
let block_ids: Vec<BmpBlockId> = self
.plans
.get(&plan_id)
.map(|p| p.block_ids.clone())
.unwrap_or_default();
let dst_node: String = self
.plans
.get(&plan_id)
.map(|p| p.dst_node.clone())
.unwrap_or_default();
let dry_run = self.config.dry_run;
let mut succeeded_blocks: Vec<BmpBlockId> = Vec::new();
let mut failed_blocks: Vec<(BmpBlockId, String)> = Vec::new();
let mut bytes_moved: u64 = 0;
for bid in &block_ids {
let size = self.blocks.get(bid).map(|m| m.size_bytes).unwrap_or(0);
let (success, error_msg): (bool, Option<String>) = if dry_run {
(true, None)
} else {
let roll = xorshift64(&mut self.rng_state);
if (roll >> 60) != 0xF {
(true, None)
} else {
(
false,
Some(format!("simulated I/O error (roll=0x{:016x})", roll)),
)
}
};
if success {
bytes_moved += size;
succeeded_blocks.push(*bid);
if let Some(meta) = self.blocks.get_mut(bid) {
meta.src_node = dst_node.clone();
meta.dst_node = None;
}
} else {
failed_blocks.push((*bid, error_msg.clone().unwrap_or_default()));
}
let record = BmpMigrationRecord {
ts: now_ts(),
plan_id,
block_id: *bid,
bytes_moved: if success { size } else { 0 },
success,
error: error_msg,
};
self.append_log(record);
}
let final_status = if failed_blocks.is_empty() {
BmpPlanStatus::Completed
} else {
BmpPlanStatus::Failed
};
if let Some(plan) = self.plans.get_mut(&plan_id) {
plan.status = final_status;
}
Ok(BmpExecutionResult {
plan_id,
status: final_status,
succeeded_blocks,
failed_blocks,
bytes_moved,
})
}
pub fn schedule_migrations(
&mut self,
policy: BmpPriorityPolicy,
max_plans: usize,
) -> Result<Vec<BmpPlanId>, BmpError> {
if max_plans == 0 {
return Ok(Vec::new());
}
let mut candidates: Vec<BmpBlockId> = self
.blocks
.values()
.filter(|m| !m.is_pinned && m.dst_node.is_some())
.map(|m| m.id)
.collect();
if candidates.is_empty() {
candidates = self
.blocks
.values()
.filter(|m| !m.is_pinned && m.tier > 0)
.map(|m| m.id)
.collect();
}
if candidates.is_empty() {
return Err(BmpError::NoEligibleBlocks);
}
self.sort_candidates_by_policy(&mut candidates, policy);
candidates.truncate(max_plans);
let mut plan_ids = Vec::with_capacity(candidates.len());
for bid in candidates {
let dst = self
.blocks
.get(&bid)
.and_then(|m| m.dst_node.clone())
.unwrap_or_else(|| "node-0".to_string());
match self.create_plan(vec![bid], dst, 0) {
Ok(pid) => plan_ids.push(pid),
Err(_) => continue,
}
}
Ok(plan_ids)
}
fn sort_candidates_by_policy(&self, ids: &mut [BmpBlockId], policy: BmpPriorityPolicy) {
match policy {
BmpPriorityPolicy::FifoQueue => {
ids.sort_by(|a, b| {
let ta = self.blocks.get(a).map_or(0, |m| m.last_accessed);
let tb = self.blocks.get(b).map_or(0, |m| m.last_accessed);
ta.cmp(&tb)
});
}
BmpPriorityPolicy::HighFrequencyFirst => {
ids.sort_by(|a, b| {
let fa = self.blocks.get(a).map_or(0.0, |m| m.access_frequency);
let fb = self.blocks.get(b).map_or(0.0, |m| m.access_frequency);
fb.partial_cmp(&fa).unwrap_or(std::cmp::Ordering::Equal)
});
}
BmpPriorityPolicy::LargestFirst => {
ids.sort_by(|a, b| {
let sa = self.blocks.get(a).map_or(0, |m| m.size_bytes);
let sb = self.blocks.get(b).map_or(0, |m| m.size_bytes);
sb.cmp(&sa)
});
}
BmpPriorityPolicy::SmallestFirst => {
ids.sort_by(|a, b| {
let sa = self.blocks.get(a).map_or(0, |m| m.size_bytes);
let sb = self.blocks.get(b).map_or(0, |m| m.size_bytes);
sa.cmp(&sb)
});
}
BmpPriorityPolicy::CostOptimized => {
ids.sort_by(|a, b| {
let ca = self
.blocks
.get(a)
.map_or(0, |m| m.size_bytes.saturating_mul(m.tier as u64));
let cb = self
.blocks
.get(b)
.map_or(0, |m| m.size_bytes.saturating_mul(m.tier as u64));
ca.cmp(&cb)
});
}
}
}
pub fn run_batch_migration(
&mut self,
policy: BmpPriorityPolicy,
batch_size: usize,
) -> BmpBatchResult {
let max = self.config.max_concurrent_migrations.min(batch_size);
let plan_ids = match self.schedule_migrations(policy, max) {
Ok(ids) => ids,
Err(_) => return BmpBatchResult::default(),
};
let mut batch = BmpBatchResult {
plans_executed: plan_ids.len(),
..BmpBatchResult::default()
};
for pid in plan_ids {
match self.execute_plan(pid) {
Ok(result) => {
batch.total_bytes_moved += result.bytes_moved;
if result.status == BmpPlanStatus::Completed {
batch.plans_completed += 1;
} else {
batch.plans_failed += 1;
}
batch.results.push(result);
}
Err(_) => {
batch.plans_failed += 1;
}
}
}
batch
}
pub fn defragment_plan(&mut self, nodes: &[String]) -> Vec<BmpPlanId> {
if nodes.is_empty() {
return Vec::new();
}
let mut node_counts: HashMap<String, usize> =
nodes.iter().map(|n| (n.clone(), 0)).collect();
for meta in self.blocks.values() {
if let Some(cnt) = node_counts.get_mut(&meta.src_node) {
*cnt += 1;
}
}
let total: usize = node_counts.values().sum();
let target = total.div_ceil(nodes.len());
let mut over_nodes: Vec<String> = node_counts
.iter()
.filter(|(_, &cnt)| cnt > target)
.map(|(n, _)| n.clone())
.collect();
let mut under_nodes: Vec<String> = node_counts
.iter()
.filter(|(_, &cnt)| cnt < target)
.map(|(n, _)| n.clone())
.collect();
if over_nodes.is_empty() || under_nodes.is_empty() {
return Vec::new();
}
over_nodes.sort();
under_nodes.sort();
struct Move {
bid: BmpBlockId,
dst: String,
}
let mut moves: Vec<Move> = Vec::new();
let mut capacity: HashMap<String, usize> = under_nodes
.iter()
.map(|n| {
let cnt = node_counts.get(n).copied().unwrap_or(0);
(n.clone(), target.saturating_sub(cnt))
})
.collect();
for over in &over_nodes {
let excess = node_counts
.get(over)
.map_or(0, |&cnt| cnt.saturating_sub(target));
if excess == 0 {
continue;
}
let candidates: Vec<BmpBlockId> = self
.blocks
.values()
.filter(|m| &m.src_node == over && !m.is_pinned && m.dst_node.is_none())
.map(|m| m.id)
.collect();
let mut moved = 0usize;
'outer: for bid in candidates {
if moved >= excess {
break;
}
for under in &under_nodes {
let cap = capacity.get_mut(under);
if let Some(c) = cap {
if *c > 0 {
*c -= 1;
moves.push(Move {
bid,
dst: under.clone(),
});
moved += 1;
continue 'outer;
}
}
}
break; }
}
let mut plan_ids = Vec::with_capacity(moves.len());
for m in moves {
if let Some(meta) = self.blocks.get_mut(&m.bid) {
meta.dst_node = Some(m.dst.clone());
}
match self.create_plan(vec![m.bid], m.dst, 0) {
Ok(pid) => plan_ids.push(pid),
Err(_) => continue,
}
}
plan_ids
}
pub fn migration_stats(&self) -> BmpPlannerStats {
let pinned_blocks = self.blocks.values().filter(|m| m.is_pinned).count();
let mut stats = BmpPlannerStats {
total_blocks: self.blocks.len(),
pinned_blocks,
total_plans: self.plans.len(),
total_records: self.log.len(),
..Default::default()
};
for plan in self.plans.values() {
match plan.status {
BmpPlanStatus::Pending => stats.pending_plans += 1,
BmpPlanStatus::InProgress => stats.in_progress_plans += 1,
BmpPlanStatus::Completed => stats.completed_plans += 1,
BmpPlanStatus::Failed => stats.failed_plans += 1,
BmpPlanStatus::Cancelled => stats.cancelled_plans += 1,
}
}
for rec in &self.log {
if rec.success {
stats.total_bytes_moved += rec.bytes_moved;
} else {
stats.total_failures += 1;
}
}
stats
}
pub fn plan_count(&self) -> usize {
self.plans.len()
}
pub fn plan(&self, plan_id: BmpPlanId) -> Option<&BmpMigrationPlan> {
self.plans.get(&plan_id)
}
pub fn plans_iter(&self) -> impl Iterator<Item = &BmpMigrationPlan> {
self.plans.values()
}
pub fn recent_records(&self, n: usize) -> Vec<&BmpMigrationRecord> {
self.log.iter().rev().take(n).collect()
}
pub fn log_len(&self) -> usize {
self.log.len()
}
fn append_log(&mut self, record: BmpMigrationRecord) {
const MAX_LOG: usize = 1000;
if self.log.len() >= MAX_LOG {
self.log.pop_front();
}
self.log.push_back(record);
}
}
impl Default for BlockMigrationPlanner {
fn default() -> Self {
Self::new()
}
}
pub fn bmp_id_from_bytes(data: &[u8]) -> BmpBlockId {
let mut id = [0u8; 32];
let h0 = fnv1a_64(data);
let h1 = fnv1a_64(&h0.to_le_bytes());
let h2 = fnv1a_64(&h1.to_le_bytes());
let h3 = fnv1a_64(&h2.to_le_bytes());
id[0..8].copy_from_slice(&h0.to_le_bytes());
id[8..16].copy_from_slice(&h1.to_le_bytes());
id[16..24].copy_from_slice(&h2.to_le_bytes());
id[24..32].copy_from_slice(&h3.to_le_bytes());
id
}
#[cfg(test)]
mod tests {
use super::*;
fn make_id(seed: u8) -> BmpBlockId {
bmp_id_from_bytes(&[seed; 16])
}
fn make_planner() -> BlockMigrationPlanner {
BlockMigrationPlanner::new()
}
fn register_n(planner: &mut BlockMigrationPlanner, n: usize, node: &str) -> Vec<BmpBlockId> {
let mut ids = Vec::with_capacity(n);
for i in 0..n {
let id = bmp_id_from_bytes(format!("block-{}", i).as_bytes());
planner.register_block(id, (i as u64 + 1) * 512, node, 1, 1.0);
ids.push(id);
}
ids
}
#[test]
fn test_xorshift64_nonzero() {
let mut state: u64 = 12345678;
for _ in 0..100 {
let v = xorshift64(&mut state);
assert_ne!(v, 0);
}
}
#[test]
fn test_xorshift64_changes_state() {
let mut state: u64 = 1;
let a = xorshift64(&mut state);
let b = xorshift64(&mut state);
assert_ne!(a, b);
}
#[test]
fn test_fnv1a_empty() {
let h = fnv1a_64(&[]);
assert_eq!(h, 14_695_981_039_346_656_037_u64);
}
#[test]
fn test_fnv1a_deterministic() {
assert_eq!(fnv1a_64(b"hello"), fnv1a_64(b"hello"));
}
#[test]
fn test_fnv1a_differs_on_different_input() {
assert_ne!(fnv1a_64(b"foo"), fnv1a_64(b"bar"));
}
#[test]
fn test_bmp_id_from_bytes_length() {
let id = bmp_id_from_bytes(b"test");
assert_eq!(id.len(), 32);
}
#[test]
fn test_bmp_id_from_bytes_deterministic() {
assert_eq!(bmp_id_from_bytes(b"abc"), bmp_id_from_bytes(b"abc"));
}
#[test]
fn test_bmp_id_from_bytes_differs() {
assert_ne!(bmp_id_from_bytes(b"a"), bmp_id_from_bytes(b"b"));
}
#[test]
fn test_new_planner_empty() {
let p = make_planner();
assert_eq!(p.block_count(), 0);
assert_eq!(p.plan_count(), 0);
assert_eq!(p.log_len(), 0);
}
#[test]
fn test_with_config_dry_run() {
let cfg = BmpPlannerConfig {
dry_run: true,
..Default::default()
};
let p = BlockMigrationPlanner::with_config(cfg);
assert!(p.config().dry_run);
}
#[test]
fn test_set_config() {
let mut p = make_planner();
let cfg = BmpPlannerConfig {
bandwidth_limit_kbps: 1024,
..Default::default()
};
p.set_config(cfg);
assert_eq!(p.config().bandwidth_limit_kbps, 1024);
}
#[test]
fn test_register_block_increases_count() {
let mut p = make_planner();
let id = make_id(1);
p.register_block(id, 1024, "node-0", 0, 1.0);
assert_eq!(p.block_count(), 1);
}
#[test]
fn test_register_block_metadata() {
let mut p = make_planner();
let id = make_id(2);
p.register_block(id, 2048, "node-1", 2, 3.5);
let meta = p.block_meta(&id).expect("block should exist");
assert_eq!(meta.size_bytes, 2048);
assert_eq!(meta.src_node, "node-1");
assert_eq!(meta.tier, 2);
assert!((meta.access_frequency - 3.5).abs() < 1e-9);
assert!(!meta.is_pinned);
}
#[test]
fn test_register_block_overwrite() {
let mut p = make_planner();
let id = make_id(3);
p.register_block(id, 100, "node-a", 0, 0.5);
p.register_block(id, 200, "node-b", 1, 2.0);
assert_eq!(p.block_count(), 1); let meta = p.block_meta(&id).expect("exists");
assert_eq!(meta.size_bytes, 200);
assert_eq!(meta.src_node, "node-b");
}
#[test]
fn test_register_multiple_blocks() {
let mut p = make_planner();
for i in 0..10u8 {
p.register_block(make_id(i), 512, "node-0", 0, 1.0);
}
assert_eq!(p.block_count(), 10);
}
#[test]
fn test_update_access_ok() {
let mut p = make_planner();
let id = make_id(10);
p.register_block(id, 512, "node-0", 0, 1.0);
let old_freq = p.block_meta(&id).map_or(0.0, |m| m.access_frequency);
p.update_access(&id).expect("ok");
let new_freq = p.block_meta(&id).map_or(0.0, |m| m.access_frequency);
assert!(new_freq > 0.0);
let _ = old_freq; }
#[test]
fn test_update_access_unknown_block_errors() {
let mut p = make_planner();
let id = make_id(11);
assert!(p.update_access(&id).is_err());
}
#[test]
fn test_update_access_bumps_last_accessed() {
let mut p = make_planner();
let id = make_id(12);
p.register_block(id, 512, "node-0", 0, 1.0);
let t0 = p.block_meta(&id).map_or(0, |m| m.last_accessed);
std::thread::sleep(std::time::Duration::from_millis(1));
p.update_access(&id).expect("ok");
let t1 = p.block_meta(&id).map_or(0, |m| m.last_accessed);
assert!(t1 >= t0);
}
#[test]
fn test_pin_block() {
let mut p = make_planner();
let id = make_id(20);
p.register_block(id, 512, "node-0", 0, 1.0);
p.pin(&id).expect("ok");
assert!(p.block_meta(&id).is_some_and(|m| m.is_pinned));
}
#[test]
fn test_unpin_block() {
let mut p = make_planner();
let id = make_id(21);
p.register_block(id, 512, "node-0", 0, 1.0);
p.pin(&id).expect("ok");
p.unpin(&id).expect("ok");
assert!(!p.block_meta(&id).is_none_or(|m| m.is_pinned));
}
#[test]
fn test_pin_unknown_block_errors() {
let mut p = make_planner();
assert!(p.pin(&make_id(22)).is_err());
}
#[test]
fn test_unpin_unknown_block_errors() {
let mut p = make_planner();
assert!(p.unpin(&make_id(23)).is_err());
}
#[test]
fn test_create_plan_basic() {
let mut p = make_planner();
let id = make_id(30);
p.register_block(id, 1024, "node-0", 0, 1.0);
let pid = p.create_plan(vec![id], "node-1", 5).expect("ok");
assert_eq!(pid, 1);
let plan = p.plan(pid).expect("plan exists");
assert_eq!(plan.status, BmpPlanStatus::Pending);
assert_eq!(plan.estimated_bytes, 1024);
assert_eq!(plan.dst_node, "node-1");
assert_eq!(plan.priority, 5);
}
#[test]
fn test_create_plan_empty_blocks_errors() {
let mut p = make_planner();
assert!(p.create_plan(vec![], "node-1", 0).is_err());
}
#[test]
fn test_create_plan_empty_dst_errors() {
let mut p = make_planner();
let id = make_id(31);
p.register_block(id, 512, "node-0", 0, 1.0);
assert!(p.create_plan(vec![id], "", 0).is_err());
}
#[test]
fn test_create_plan_unknown_block_errors() {
let mut p = make_planner();
let id = make_id(32);
assert!(p.create_plan(vec![id], "node-1", 0).is_err());
}
#[test]
fn test_create_plan_pinned_block_errors() {
let mut p = make_planner();
let id = make_id(33);
p.register_block(id, 512, "node-0", 0, 1.0);
p.pin(&id).expect("ok");
assert!(p.create_plan(vec![id], "node-1", 0).is_err());
}
#[test]
fn test_create_plan_increments_id() {
let mut p = make_planner();
for i in 0..5u8 {
let id = make_id(40 + i);
p.register_block(id, 512, "node-0", 0, 1.0);
let pid = p.create_plan(vec![id], "node-1", 0).expect("ok");
assert_eq!(pid, i as u64 + 1);
}
}
#[test]
fn test_create_plan_multi_block_bytes() {
let mut p = make_planner();
let ids: Vec<_> = (0..3u8)
.map(|i| {
let id = make_id(50 + i);
p.register_block(id, 1000, "node-0", 0, 1.0);
id
})
.collect();
let pid = p.create_plan(ids, "node-1", 0).expect("ok");
let plan = p.plan(pid).expect("plan");
assert_eq!(plan.estimated_bytes, 3000);
}
#[test]
fn test_cancel_pending_plan() {
let mut p = make_planner();
let id = make_id(60);
p.register_block(id, 512, "node-0", 0, 1.0);
let pid = p.create_plan(vec![id], "node-1", 0).expect("ok");
p.cancel_plan(pid).expect("ok");
assert_eq!(
p.plan(pid).map(|pl| pl.status),
Some(BmpPlanStatus::Cancelled)
);
}
#[test]
fn test_cancel_nonexistent_plan_errors() {
let mut p = make_planner();
assert!(p.cancel_plan(999).is_err());
}
#[test]
fn test_cancel_already_cancelled_errors() {
let mut p = make_planner();
let id = make_id(61);
p.register_block(id, 512, "node-0", 0, 1.0);
let pid = p.create_plan(vec![id], "node-1", 0).expect("ok");
p.cancel_plan(pid).expect("ok");
assert!(p.cancel_plan(pid).is_err());
}
#[test]
fn test_execute_plan_dry_run_always_succeeds() {
let cfg = BmpPlannerConfig {
dry_run: true,
..Default::default()
};
let mut p = BlockMigrationPlanner::with_config(cfg);
let id = make_id(70);
p.register_block(id, 512, "node-0", 0, 1.0);
let pid = p.create_plan(vec![id], "node-1", 0).expect("ok");
let result = p.execute_plan(pid).expect("ok");
assert_eq!(result.status, BmpPlanStatus::Completed);
assert_eq!(result.succeeded_blocks.len(), 1);
assert!(result.failed_blocks.is_empty());
assert_eq!(result.bytes_moved, 512);
}
#[test]
fn test_execute_plan_updates_log() {
let cfg = BmpPlannerConfig {
dry_run: true,
..Default::default()
};
let mut p = BlockMigrationPlanner::with_config(cfg);
let id = make_id(71);
p.register_block(id, 256, "node-0", 0, 1.0);
let pid = p.create_plan(vec![id], "node-1", 0).expect("ok");
p.execute_plan(pid).expect("ok");
assert_eq!(p.log_len(), 1);
}
#[test]
fn test_execute_plan_updates_block_src_node() {
let cfg = BmpPlannerConfig {
dry_run: true,
..Default::default()
};
let mut p = BlockMigrationPlanner::with_config(cfg);
let id = make_id(72);
p.register_block(id, 512, "node-0", 0, 1.0);
let pid = p.create_plan(vec![id], "node-2", 0).expect("ok");
p.execute_plan(pid).expect("ok");
let meta = p.block_meta(&id).expect("exists");
assert_eq!(meta.src_node, "node-2");
}
#[test]
fn test_execute_plan_not_found_errors() {
let mut p = make_planner();
assert!(p.execute_plan(42).is_err());
}
#[test]
fn test_execute_plan_not_pending_errors() {
let cfg = BmpPlannerConfig {
dry_run: true,
..Default::default()
};
let mut p = BlockMigrationPlanner::with_config(cfg);
let id = make_id(73);
p.register_block(id, 512, "node-0", 0, 1.0);
let pid = p.create_plan(vec![id], "node-1", 0).expect("ok");
p.execute_plan(pid).expect("ok");
assert!(p.execute_plan(pid).is_err());
}
#[test]
fn test_execute_plan_multi_block_dry_run() {
let cfg = BmpPlannerConfig {
dry_run: true,
..Default::default()
};
let mut p = BlockMigrationPlanner::with_config(cfg);
let ids: Vec<_> = (0..5u8)
.map(|i| {
let id = make_id(80 + i);
p.register_block(id, 100, "node-0", 0, 1.0);
id
})
.collect();
let pid = p.create_plan(ids.clone(), "node-1", 0).expect("ok");
let res = p.execute_plan(pid).expect("ok");
assert_eq!(res.succeeded_blocks.len(), 5);
assert_eq!(res.bytes_moved, 500);
}
#[test]
fn test_schedule_migrations_no_eligible_blocks_errors() {
let mut p = make_planner();
let id = make_id(90);
p.register_block(id, 512, "node-0", 0, 1.0);
let result = p.schedule_migrations(BmpPriorityPolicy::FifoQueue, 5);
assert!(result.is_err());
}
#[test]
fn test_schedule_migrations_with_dst_node_set() {
let mut p = make_planner();
let id = make_id(91);
p.register_block(id, 512, "node-0", 1, 1.0);
if let Some(meta) = p.blocks.get_mut(&id) {
meta.dst_node = Some("node-1".to_string());
}
let ids = p
.schedule_migrations(BmpPriorityPolicy::FifoQueue, 5)
.expect("ok");
assert_eq!(ids.len(), 1);
}
#[test]
fn test_schedule_migrations_respects_max_plans() {
let mut p = make_planner();
for i in 0..10u8 {
let id = make_id(100 + i);
p.register_block(id, 512, "node-0", 1, 1.0);
if let Some(meta) = p.blocks.get_mut(&id) {
meta.dst_node = Some("node-1".to_string());
}
}
let ids = p
.schedule_migrations(BmpPriorityPolicy::FifoQueue, 3)
.expect("ok");
assert_eq!(ids.len(), 3);
}
#[test]
fn test_schedule_migrations_skips_pinned() {
let mut p = make_planner();
for i in 0..3u8 {
let id = make_id(110 + i);
p.register_block(id, 512, "node-0", 1, 1.0);
if let Some(meta) = p.blocks.get_mut(&id) {
meta.dst_node = Some("node-1".to_string());
}
if i == 0 {
p.pin(&id).expect("ok");
}
}
let ids = p
.schedule_migrations(BmpPriorityPolicy::FifoQueue, 10)
.expect("ok");
assert_eq!(ids.len(), 2);
}
#[test]
fn test_schedule_migrations_zero_max_returns_empty() {
let mut p = make_planner();
let id = make_id(120);
p.register_block(id, 512, "node-0", 1, 1.0);
if let Some(meta) = p.blocks.get_mut(&id) {
meta.dst_node = Some("node-1".to_string());
}
let ids = p
.schedule_migrations(BmpPriorityPolicy::FifoQueue, 0)
.expect("ok");
assert!(ids.is_empty());
}
#[test]
fn test_schedule_largest_first_ordering() {
let mut p = make_planner();
let sizes = [100u64, 500, 200, 800, 50];
let mut block_ids = Vec::new();
for (i, &sz) in sizes.iter().enumerate() {
let id = make_id(130 + i as u8);
p.register_block(id, sz, "node-0", 1, 1.0);
if let Some(meta) = p.blocks.get_mut(&id) {
meta.dst_node = Some("node-1".to_string());
}
block_ids.push(id);
}
let pids = p
.schedule_migrations(BmpPriorityPolicy::LargestFirst, 5)
.expect("ok");
let first_bytes = p.plan(pids[0]).map_or(0, |pl| pl.estimated_bytes);
let last_bytes = p
.plan(*pids.last().unwrap())
.map_or(0, |pl| pl.estimated_bytes);
assert!(
first_bytes >= last_bytes,
"largest first violated: {} < {}",
first_bytes,
last_bytes
);
}
#[test]
fn test_schedule_smallest_first_ordering() {
let mut p = make_planner();
let sizes = [100u64, 500, 200, 800, 50];
for (i, &sz) in sizes.iter().enumerate() {
let id = make_id(140 + i as u8);
p.register_block(id, sz, "node-0", 1, 1.0);
if let Some(meta) = p.blocks.get_mut(&id) {
meta.dst_node = Some("node-1".to_string());
}
}
let pids = p
.schedule_migrations(BmpPriorityPolicy::SmallestFirst, 5)
.expect("ok");
let first_bytes = p.plan(pids[0]).map_or(0, |pl| pl.estimated_bytes);
let last_bytes = p
.plan(*pids.last().unwrap())
.map_or(0, |pl| pl.estimated_bytes);
assert!(
first_bytes <= last_bytes,
"smallest first violated: {} > {}",
first_bytes,
last_bytes
);
}
#[test]
fn test_schedule_high_frequency_first() {
let mut p = make_planner();
let freqs = [0.1f64, 5.0, 2.0];
let mut bid_map: Vec<(BmpBlockId, f64)> = Vec::new();
for (i, &f) in freqs.iter().enumerate() {
let id = make_id(150 + i as u8);
p.register_block(id, 512, "node-0", 1, f);
if let Some(meta) = p.blocks.get_mut(&id) {
meta.dst_node = Some("node-1".to_string());
}
bid_map.push((id, f));
}
let pids = p
.schedule_migrations(BmpPriorityPolicy::HighFrequencyFirst, 3)
.expect("ok");
let first_plan_bid = p.plan(pids[0]).and_then(|pl| pl.block_ids.first().copied());
if let Some(bid) = first_plan_bid {
let freq = p.block_meta(&bid).map_or(0.0, |m| m.access_frequency);
assert!(
(freq - 5.0).abs() < 1e-6 || freq >= 2.0,
"unexpected freq {}",
freq
);
}
}
#[test]
fn test_run_batch_migration_dry_run() {
let cfg = BmpPlannerConfig {
dry_run: true,
max_concurrent_migrations: 10,
..Default::default()
};
let mut p = BlockMigrationPlanner::with_config(cfg);
let _ids = register_n(&mut p, 5, "node-0");
for id in &_ids {
if let Some(meta) = p.blocks.get_mut(id) {
meta.dst_node = Some("node-1".to_string());
meta.tier = 1;
}
}
let batch = p.run_batch_migration(BmpPriorityPolicy::FifoQueue, 5);
assert!(batch.plans_executed > 0);
assert_eq!(batch.plans_failed, 0);
}
#[test]
fn test_run_batch_migration_empty_result_on_no_candidates() {
let mut p = make_planner();
let id = make_id(160);
p.register_block(id, 512, "node-0", 0, 1.0);
let batch = p.run_batch_migration(BmpPriorityPolicy::FifoQueue, 5);
assert_eq!(batch.plans_executed, 0);
}
#[test]
fn test_run_batch_migration_respects_max_concurrent() {
let cfg = BmpPlannerConfig {
dry_run: true,
max_concurrent_migrations: 2,
..Default::default()
};
let mut p = BlockMigrationPlanner::with_config(cfg);
let _ids = register_n(&mut p, 10, "node-0");
for id in &_ids {
if let Some(meta) = p.blocks.get_mut(id) {
meta.dst_node = Some("node-1".to_string());
meta.tier = 1;
}
}
let batch = p.run_batch_migration(BmpPriorityPolicy::FifoQueue, 10);
assert!(batch.plans_executed <= 2);
}
#[test]
fn test_run_batch_total_bytes_moved() {
let cfg = BmpPlannerConfig {
dry_run: true,
max_concurrent_migrations: 5,
..Default::default()
};
let mut p = BlockMigrationPlanner::with_config(cfg);
for i in 0..5u8 {
let id = make_id(170 + i);
p.register_block(id, 1000, "node-0", 1, 1.0);
if let Some(meta) = p.blocks.get_mut(&id) {
meta.dst_node = Some("node-1".to_string());
}
}
let batch = p.run_batch_migration(BmpPriorityPolicy::FifoQueue, 5);
assert_eq!(
batch.total_bytes_moved,
(batch.plans_completed as u64) * 1000
);
}
#[test]
fn test_defragment_plan_empty_nodes() {
let mut p = make_planner();
let result = p.defragment_plan(&[]);
assert!(result.is_empty());
}
#[test]
fn test_defragment_plan_already_balanced() {
let mut p = make_planner();
let nodes = vec!["node-0".to_string(), "node-1".to_string()];
for i in 0..4u8 {
let node = if i < 2 { "node-0" } else { "node-1" };
let id = make_id(180 + i);
p.register_block(id, 512, node, 0, 1.0);
}
let plans = p.defragment_plan(&nodes);
assert!(plans.is_empty());
}
#[test]
fn test_defragment_plan_imbalanced() {
let mut p = make_planner();
let nodes = vec!["node-0".to_string(), "node-1".to_string()];
for i in 0..4u8 {
let id = make_id(190 + i);
p.register_block(id, 512, "node-0", 0, 1.0);
}
let plans = p.defragment_plan(&nodes);
assert!(!plans.is_empty());
}
#[test]
fn test_defragment_plan_skips_pinned() {
let mut p = make_planner();
let nodes = vec!["node-0".to_string(), "node-1".to_string()];
for i in 0..4u8 {
let id = make_id(200 + i);
p.register_block(id, 512, "node-0", 0, 1.0);
if i < 2 {
p.pin(&id).expect("ok");
}
}
let plans = p.defragment_plan(&nodes);
assert!(plans.len() <= 2);
}
#[test]
fn test_defragment_plan_single_node_no_plans() {
let mut p = make_planner();
let nodes = vec!["node-0".to_string()];
for i in 0..3u8 {
let id = make_id(210 + i);
p.register_block(id, 512, "node-0", 0, 1.0);
}
let plans = p.defragment_plan(&nodes);
assert!(plans.is_empty());
}
#[test]
fn test_migration_stats_empty() {
let p = make_planner();
let stats = p.migration_stats();
assert_eq!(stats.total_blocks, 0);
assert_eq!(stats.total_plans, 0);
assert_eq!(stats.total_records, 0);
assert_eq!(stats.total_bytes_moved, 0);
}
#[test]
fn test_migration_stats_after_register() {
let mut p = make_planner();
for i in 0..5u8 {
p.register_block(make_id(220 + i), 512, "node-0", 0, 1.0);
}
let stats = p.migration_stats();
assert_eq!(stats.total_blocks, 5);
}
#[test]
fn test_migration_stats_pinned_count() {
let mut p = make_planner();
for i in 0..4u8 {
let id = make_id(230 + i);
p.register_block(id, 512, "node-0", 0, 1.0);
if i < 2 {
p.pin(&id).expect("ok");
}
}
let stats = p.migration_stats();
assert_eq!(stats.pinned_blocks, 2);
}
#[test]
fn test_migration_stats_plan_counts() {
let cfg = BmpPlannerConfig {
dry_run: true,
..Default::default()
};
let mut p = BlockMigrationPlanner::with_config(cfg);
for i in 0..3u8 {
let id = make_id(240 + i);
p.register_block(id, 512, "node-0", 0, 1.0);
p.create_plan(vec![id], "node-1", 0).expect("ok");
}
let stats = p.migration_stats();
assert_eq!(stats.total_plans, 3);
assert_eq!(stats.pending_plans, 3);
}
#[test]
fn test_migration_stats_bytes_moved() {
let cfg = BmpPlannerConfig {
dry_run: true,
..Default::default()
};
let mut p = BlockMigrationPlanner::with_config(cfg);
let id = make_id(250);
p.register_block(id, 4096, "node-0", 0, 1.0);
let pid = p.create_plan(vec![id], "node-1", 0).expect("ok");
p.execute_plan(pid).expect("ok");
let stats = p.migration_stats();
assert_eq!(stats.total_bytes_moved, 4096);
}
#[test]
fn test_migration_stats_cancelled_count() {
let mut p = make_planner();
for i in 200..203u8 {
let id = make_id(i);
p.register_block(id, 512, "node-0", 0, 1.0);
let pid = p.create_plan(vec![id], "node-1", 0).expect("ok");
p.cancel_plan(pid).expect("ok");
}
let stats = p.migration_stats();
assert_eq!(stats.cancelled_plans, 3);
}
#[test]
fn test_log_bounded_at_1000() {
let cfg = BmpPlannerConfig {
dry_run: true,
..Default::default()
};
let mut p = BlockMigrationPlanner::with_config(cfg);
for i in 0..1100u16 {
let id = bmp_id_from_bytes(&i.to_le_bytes());
p.register_block(id, 1, "node-0", 0, 1.0);
if let Ok(pid) = p.create_plan(vec![id], "node-1", 0) {
let _ = p.execute_plan(pid);
}
}
assert_eq!(p.log_len(), 1000);
}
#[test]
fn test_recent_records_returns_n() {
let cfg = BmpPlannerConfig {
dry_run: true,
..Default::default()
};
let mut p = BlockMigrationPlanner::with_config(cfg);
for i in 0..10u8 {
let id = make_id(i);
p.register_block(id, 512, "node-0", 0, 1.0);
if let Ok(pid) = p.create_plan(vec![id], "node-1", 0) {
let _ = p.execute_plan(pid);
}
}
assert_eq!(p.recent_records(5).len(), 5);
}
#[test]
fn test_plans_iter() {
let mut p = make_planner();
for i in 0..4u8 {
let id = make_id(i);
p.register_block(id, 512, "node-0", 0, 1.0);
p.create_plan(vec![id], "node-1", 0).expect("ok");
}
assert_eq!(p.plans_iter().count(), 4);
}
#[test]
fn test_plan_status_display() {
assert_eq!(BmpPlanStatus::Pending.to_string(), "Pending");
assert_eq!(BmpPlanStatus::InProgress.to_string(), "InProgress");
assert_eq!(BmpPlanStatus::Completed.to_string(), "Completed");
assert_eq!(BmpPlanStatus::Failed.to_string(), "Failed");
assert_eq!(BmpPlanStatus::Cancelled.to_string(), "Cancelled");
}
#[test]
fn test_priority_policy_display() {
assert_eq!(BmpPriorityPolicy::FifoQueue.to_string(), "FifoQueue");
assert_eq!(
BmpPriorityPolicy::HighFrequencyFirst.to_string(),
"HighFrequencyFirst"
);
assert_eq!(BmpPriorityPolicy::LargestFirst.to_string(), "LargestFirst");
assert_eq!(
BmpPriorityPolicy::SmallestFirst.to_string(),
"SmallestFirst"
);
assert_eq!(
BmpPriorityPolicy::CostOptimized.to_string(),
"CostOptimized"
);
}
#[test]
fn test_planner_default() {
let p = BlockMigrationPlanner::default();
assert_eq!(p.block_count(), 0);
}
#[test]
fn test_config_default() {
let cfg = BmpPlannerConfig::default();
assert_eq!(cfg.max_concurrent_migrations, 4);
assert_eq!(cfg.bandwidth_limit_kbps, 0);
assert!(!cfg.dry_run);
}
#[test]
fn test_batch_result_default() {
let b = BmpBatchResult::default();
assert_eq!(b.plans_executed, 0);
assert_eq!(b.total_bytes_moved, 0);
}
#[test]
fn test_planner_stats_default() {
let s = BmpPlannerStats::default();
assert_eq!(s.total_blocks, 0);
assert_eq!(s.total_failures, 0);
}
#[test]
fn test_schedule_cost_optimized() {
let mut p = make_planner();
let id_a = make_id(20);
let id_b = make_id(21);
p.register_block(id_a, 100, "node-0", 3, 1.0);
p.register_block(id_b, 500, "node-0", 1, 1.0);
for id in [id_a, id_b] {
if let Some(m) = p.blocks.get_mut(&id) {
m.dst_node = Some("node-1".to_string());
}
}
let pids = p
.schedule_migrations(BmpPriorityPolicy::CostOptimized, 2)
.expect("ok");
let first_plan = p.plan(pids[0]).expect("plan");
let first_bid = first_plan.block_ids[0];
let first_meta = p.block_meta(&first_bid).expect("meta");
let first_cost = first_meta.size_bytes * first_meta.tier as u64;
assert!(
first_cost <= 500,
"cost optimized ordering failed: {} > 500",
first_cost
);
}
}