consistent_hasher 0.1.5

//! # LDB: Consistent Hashing Library
//! 
//! This crate implements a consistent hashing system with support for virtual nodes, 
//! dynamic node and key management, and transaction tracking. It is designed for use 
//! in distributed systems, load balancers, or caching mechanisms that require efficient 
//! and scalable key-to-node mapping.
//! 
//! ## Features
//! - Add and remove nodes dynamically with minimal disruption.
//! - Redistribute keys across nodes when the topology changes.
//! - Track redistribution transactions for audit and debugging purposes.
//! - Use virtual instances to improve load balancing.
//! - Configurable custom hashers using the `set_hasher` method.
//! 
//! ## Example
//! ```rust
//! use ldb::{LDB, Identifier};
//! 
//! struct Node {
//!     id: usize,
//! }
//! 
//! impl Identifier for Node {
//!     fn identify(&self) -> usize {
//!         self.id
//!     }
//! }
//! 
//! fn main() {
//!     let mut ldb = LDB::new(16, 5); // Hash space size: 2^16, 5 virtual instances per node
//! 
//!     // Add nodes
//!     ldb.add_node(Node { id: 1 }).unwrap();
//!     ldb.add_node(Node { id: 2 }).unwrap();
//! 
//!     // Add a key
//!     ldb.add_key("key1").unwrap();
//! 
//!     // Find a key
//!     if let Some(node) = ldb.find_key("key1") {
//!         println!("Key1 is mapped to node {}", node);
//!     }
//! 
//!     // Print the LDB state
//!     ldb.print();
//! }
//! ```

use core::fmt;
use std::collections::HashMap;
use std::fmt::Debug;
use std::hash::{BuildHasher, BuildHasherDefault, DefaultHasher, Hasher};

use crate::circularbtreemap::CircularBTreeMap;
use crate::Transaction;


/// Trait for objects that can serve as identifiers for nodes in the hash ring.
pub trait Identifier {
    /// Returns the unique identifier of the object as a `u128`.
    fn identify(&self) -> usize;
}

/// Enum representing errors that can occur in the LDB.
#[derive(Debug)]
pub enum LdbError {
    /// Error when attempting to remove a non-existent node.
    NodeDoesNotExist,

    /// Error when attempting to add a node that already exists.
    NodeAlreadyExists,

    /// Error when no instance exists.
    NoInstances,

}

impl fmt::Display for LdbError {
    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
        write!(f, "{:?}", self)
    }
}


impl std::error::Error for LdbError {}

/// The `LDB` struct represents the consistent hash ring.
/// It manages the relationships between nodes, keys, and their hash mappings.
pub struct LDB {
    /// A balanced tree structure to maintain hash values.
    hash_tree: CircularBTreeMap<u128>,
    /// Maps hashed values of virtual nodes to their actual node identifiers.
    hash_nodes_map: HashMap<u128, u128>,
    /// Tracks replicas of nodes for virtual instances.
    hash_replicas_map: Vec<(u128, Vec<u128>)>,

    /// Custom hasher builder for hash calculations.
    hasher_builder: Box<dyn BuildHasher<Hasher = DefaultHasher> + Send + Sync>,
    /// Number of bits in the hash space.
    hash_bit_count: u128,
    /// Maximum value in the hash space (2^`hash_bit_count`).
    hash_max: u128,
    /// Number of virtual instances per node.
    virtual_instances: usize,
}

unsafe impl Send for LDB {}
unsafe impl Sync for LDB {}

impl LDB {
    /// Creates a new `LDB` instance with the given hash bit count and virtual instances per node.
    ///
    /// # Parameters
    /// - `hash_bit_count`: Number of bits in the hash space.
    /// - `virtual_instances`: Number of virtual nodes per physical node.
    ///
    /// # Returns
    /// A new instance of `LDB`.
    pub fn new(hash_bit_count: u128, virtual_instances: usize) -> Self {
        Self {
            hash_tree: CircularBTreeMap::new(),
            hash_nodes_map: HashMap::new(),
            hash_replicas_map: Vec::new(),
            hasher_builder: Box::new(BuildHasherDefault::<DefaultHasher>::default()),
            hash_bit_count,
            hash_max: 1 << hash_bit_count,
            virtual_instances,
        }
    }
    /// Checks if a node exists in the hash ring.
    ///
    /// # Parameters
    /// - `identifier`: The node identifier.
    ///
    /// # Returns
    /// `true` if the node exists, otherwise `false`.
    pub fn node_exists<Q: Identifier>(&self, identifier: &Q) -> bool
    {
        let identifier = &(identifier.identify() as u128);
        for (k,_) in &self.hash_replicas_map
        {
            if k==identifier
            {
                return true;
            }
        }
        return false;
    }
    /// Checks if a key exists in the hash ring.
    ///
    /// # Parameters
    /// - `key`: The key to check.
    ///
    /// # Returns
    /// `true` if the key exists, otherwise `false`.

    fn rev_hash_node(&self,hash: &u128)->u128
    {
        self.hash_nodes_map.get(hash).unwrap().clone()
    }



    /// Sets a custom hasher for the hash ring.
    ///
    /// # Parameters
    /// - `hasher_builder`: The custom hasher builder to use.
    pub fn set_hasher<BH>(&mut self, hasher_builder: BH)
    where
        BH: BuildHasher<Hasher = DefaultHasher> + Send + Sync + 'static,
    {
        self.hasher_builder = Box::new(hasher_builder);
    }


 
    pub fn add_node<Q: Identifier>(&mut self, node: Q) -> Result<Option<Vec<Transaction<u128>>>,LdbError>
    {
        let value = node.identify() as u128;
        if self.node_exists(&node)
        {
            return Err(LdbError::NodeAlreadyExists);
        }
        let mut vec = Vec::new();
        let mut carriers = Vec::new();
        for i in 0..self.virtual_instances {
            let mut hasher = self.hasher_builder.build_hasher();
            hasher.write_u128(value as u128);
            hasher.write_u128(i as u128);
            let hashed_value = hasher.finish() as u128  % self.hash_max;
            
            if !self.hash_nodes_map.contains_key(&hashed_value){
                vec.push(hashed_value);
                self.hash_nodes_map.insert( hashed_value,value);
            }
            else {continue;}
            match self.hash_tree.insert_node(hashed_value)
            {
                Ok((x,y))=>
                {

                    if self.rev_hash_node(&x)==value
                    {
                        continue;
                    }
                    let x = Transaction::new(       self.rev_hash_node(&x),
                                                                   value,
                                                                           y,
                                                                           hashed_value);
                  
                    carriers.push(x);
                },
                Err(_)=>()

            }
            
        }
        self.hash_replicas_map.push((value,vec));
 
        Ok(Some(carriers))
    }

    pub fn key(&mut self,key: &str) ->Result<(u128,u128),LdbError>
    {
        
        let mut hasher = self.hasher_builder.build_hasher();
        hasher.write(key.as_bytes());
        let hashed_value = hasher.finish() as u128  % self.hash_max;
        match self.hash_tree.val_in_node(&hashed_value)
        {
            Ok(x)=>
            {
                Ok((self.rev_hash_node(&x),hashed_value))
            },
            Err(_)=>{
                Err(LdbError::NoInstances)
            }
        }

    }

    pub fn delete_node<Q:Identifier>(&mut self,node: Q) ->Result<Option<Vec<Transaction<u128>>>,LdbError>
    {
        let value = node.identify() as u128;
        if !self.node_exists(&node)
        {
            return Err(LdbError::NodeDoesNotExist)
        }
        let mut replicas= Vec::new();
        let mut vec = Vec::new();
        
        for i in 0..self.virtual_instances
        {
            if self.hash_replicas_map[i].0 == value
            {
                replicas = self.hash_replicas_map.remove(i).1;
                break;
            }
        }

        if replicas.is_empty()
        {
            return Ok(None);
        }
        for replica in replicas
        {
            match self.hash_tree.delete_node(&replica)
            {
                Ok((x,y))=> {
                    if self.rev_hash_node(&y)!=value
                    {
                        vec.push(Transaction::new(value,self.rev_hash_node(&y),x,replica));

                    }},
                Err(_) =>()
            }
            self.hash_nodes_map.remove(&replica);
        }

        

        if vec.is_empty(){return Ok(None);}

       Ok(Some(vec))
    }

    fn add_instances(&mut self,count:usize) ->Vec<Transaction<u128>>
    {
        let mut vec = Vec::new();
        for node in &mut self.hash_replicas_map
        {
            for i in self.virtual_instances..count
            {
                let mut hasher = self.hasher_builder.build_hasher();
                hasher.write_u128(node.0);
                hasher.write_u128(i as u128);
                let hashed_value = hasher.finish() as u128  % self.hash_max;
                if !self.hash_nodes_map.contains_key(&hashed_value){
                    node.1.push(hashed_value);
                    self.hash_nodes_map.insert(hashed_value, node.0);
                }
                else {continue;}
                
                match self.hash_tree.insert_node(hashed_value)
                {
                    Ok((x,y))=>
                    {
                        if self.hash_nodes_map.get(&x).unwrap().clone()==node.0 {continue;}

                        let x = Transaction::new(       self.hash_nodes_map.get(&x).unwrap().clone(),
                                                                       node.0,
                                                                               y,
                                                                               hashed_value);
                        vec.push(x);
                    },  
                    Err(_) => ()
                }

            }
        }
        vec
    }
    fn remove_intances(&mut self,count:usize)->Vec<Transaction<u128>>
    {
        let mut vec = Vec::new();
        for node in &mut self.hash_replicas_map
        {
            if node.1.len()<2
            {
                continue;
            }
            for i in (count)..self.virtual_instances
            {

                let value = node.1.pop().unwrap();
                
                match self.hash_tree.delete_node(&value)
                {
                    Ok((x,y))=>
                    {
                        if self.hash_nodes_map.get(&y).unwrap().clone()==node.0
                        {
                            continue;
                        }
                        vec.push(Transaction::new(node.0,self.hash_nodes_map.get(&y).unwrap().clone(),x,value));
                    },  
                    Err(_) => ()
                }
                self.hash_nodes_map.remove(&value);

            }
        }
        vec
    }

    /// Adds a specified number of virtual instances to each node in the database.
    /// For each new instance, a unique hash is calculated and added to the `hash_tree`.
    /// This may also trigger `Transaction`s if a new virtual instance for a node affects the distribution of keys.
    ///
    /// # Parameters
    /// - `count`: Number of additional virtual instances to add for each node.
    ///
    /// # Returns
    /// - `Vec<Transaction<u128, String>>`: A vector of transactions created as a result of the added instances.
    /// Removes a specified number of virtual instances from each node in the database.
    /// Each removed instance is removed from `hash_tree`, which can trigger transactions if removal
    /// impacts key distribution between nodes.
    ///
    /// # Parameters
    /// - `count`: Number of virtual instances to retain. All instances exceeding this number will be removed.
    ///
    /// # Returns
    /// - `Vec<Transaction<u128, String>>`: A vector of transactions created as a result of the removed instances.
    pub fn set_virtual_instances(&mut self,count: usize) -> Option<Vec<Transaction<u128>>>
    {
        if (count == self.virtual_instances) | (count == 0)
        {
            return None;
        }
        
        let mut vec= Vec::new();
        
        if  count > self.virtual_instances
        {
            vec = self.add_instances(count);
        }
        else {
            vec = self.remove_intances(count);
        }
        self.virtual_instances = count;
        if vec.is_empty() {return None;}

        return Some(vec);
    }
    
    pub fn print(&self)
    {

        println!("*******************************************************************************************************************************************************************************************************************************************************************************************************");
        println!("\nLDB:\n");
    
        println!("\n\nhash-tree:");
        self.hash_tree.print();
    
        println!("\n\nhash-nodes-map:");
        for (key, value) in &self.hash_nodes_map {
            println!("{:?}: {:?}", key, value);
        }
    
        println!("\n\nhash_replicas_map:");
        for (key, values) in &self.hash_replicas_map {
            println!("{:?}: {:?}", key, values);
        }
   
        println!("*******************************************************************************************************************************************************************************************************************************************************************************************************");

    }



}