From c9c6b0dbd41e20d19b91c6615c46da6f45925bca Mon Sep 17 00:00:00 2001 From: Alex Auvolat Date: Thu, 23 Apr 2020 17:05:46 +0000 Subject: Reorganize code --- src/table/table_fullcopy.rs | 100 ++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 100 insertions(+) create mode 100644 src/table/table_fullcopy.rs (limited to 'src/table/table_fullcopy.rs') diff --git a/src/table/table_fullcopy.rs b/src/table/table_fullcopy.rs new file mode 100644 index 00000000..2cd2e464 --- /dev/null +++ b/src/table/table_fullcopy.rs @@ -0,0 +1,100 @@ +use arc_swap::ArcSwapOption; +use std::sync::Arc; + +use crate::data::*; +use crate::rpc::membership::{Ring, System}; +use crate::table::*; + +#[derive(Clone)] +pub struct TableFullReplication { + pub write_factor: usize, + pub write_quorum: usize, + + neighbors: ArcSwapOption, +} + +#[derive(Clone)] +struct Neighbors { + ring: Arc, + neighbors: Vec, +} + +impl TableFullReplication { + pub fn new(write_factor: usize, write_quorum: usize) -> Self { + TableFullReplication { + write_factor, + write_quorum, + neighbors: ArcSwapOption::from(None), + } + } + + fn get_neighbors(&self, system: &System) -> Vec { + let neighbors = self.neighbors.load_full(); + if let Some(n) = neighbors { + if Arc::ptr_eq(&n.ring, &system.ring.borrow()) { + return n.neighbors.clone(); + } + } + + // Recalculate neighbors + let ring = system.ring.borrow().clone(); + let my_id = system.id; + + let mut nodes = vec![]; + for (node, _) in ring.config.members.iter() { + let node_ranking = hash(&[node.as_slice(), my_id.as_slice()].concat()); + nodes.push((*node, node_ranking)); + } + nodes.sort_by(|(_, rank1), (_, rank2)| rank1.cmp(rank2)); + let mut neighbors = nodes + .drain(..) + .map(|(node, _)| node) + .filter(|node| *node != my_id) + .take(self.write_factor) + .collect::>(); + neighbors.push(my_id); + self.neighbors.swap(Some(Arc::new(Neighbors { + ring, + neighbors: neighbors.clone(), + }))); + neighbors + } +} + +impl TableReplication for TableFullReplication { + // Full replication schema: all nodes store everything + // Writes are disseminated in an epidemic manner in the network + + // Advantage: do all reads locally, extremely fast + // Inconvenient: only suitable to reasonably small tables + + fn read_nodes(&self, _hash: &Hash, system: &System) -> Vec { + vec![system.id] + } + fn read_quorum(&self) -> usize { + 1 + } + + fn write_nodes(&self, _hash: &Hash, system: &System) -> Vec { + self.get_neighbors(system) + } + fn write_quorum(&self) -> usize { + self.write_quorum + } + fn max_write_errors(&self) -> usize { + self.write_factor - self.write_quorum + } + fn epidemic_writes(&self) -> bool { + true + } + + fn replication_nodes(&self, _hash: &Hash, ring: &Ring) -> Vec { + ring.config.members.keys().cloned().collect::>() + } + fn split_points(&self, _ring: &Ring) -> Vec { + let mut ret = vec![]; + ret.push([0u8; 32].into()); + ret.push([0xFFu8; 32].into()); + ret + } +} -- cgit v1.2.3