hadean-std 0.2.0

//! A scalable distributed hash map datastructure.

use std::{cmp,env,mem,ops,iter,marker,collections,hash,slice,ptr,borrow};
use hadean::{Sender,Receiver,Connection,Process,ChannelEndpoint,Channel,pid,spawn,ProcessTransfer};
use list::{Leaf,LeafIndex};

use std::hash::Hasher;

fn make_hash<T: ?Sized, S>(hash_state: &S, t: &T) -> u64 where T: hash::Hash, S: hash::BuildHasher {
	let mut state = hash_state.build_hasher();
	t.hash(&mut state);
	state.finish()
}

const PARTITIONS: usize = 10000;

/// A scalable distributed hash map datastructure.
/// `HashMap<K,V,S>` maps keys of type `K` to value of type `V`.
///
/// It enables computation to be performed in parallel across many `(key,value)` pairs at the same time.
///
/// Hashing works as in [std::collections::HashMap](https://doc.rust-lang.org/std/collections/struct.HashMap.html).
///
/// Here's an example showing addition and retreival of data to and from a `HashMap`:
///
/// ```
/// use hadean_std::hashmap::HashMap;
///
/// let mut hash_map: HashMap<String,String> = HashMap::new();
/// hash_map.insert(String::from("abc"),String::from("123"));
/// hash_map.insert(String::from("def"),String::from("456"));
/// println!("{}", hash_map.get(&String::from("abc")).unwrap());
/// println!("{}", hash_map.get(&String::from("def")).unwrap());
/// println!("{:?}", hash_map.get(&String::from("ghi")));
/// ```
pub struct HashMap<K, V, S = collections::hash_map::RandomState> where K: ProcessTransfer, V: ProcessTransfer {
	hash_builder: S,
	list: Box<Leaf<u8>>,
	start: LeafIndex,
	end: LeafIndex,
	indices: Box<[LeafIndex; PARTITIONS-1]>,
	phantom1: marker::PhantomData<K>,
	phantom2: marker::PhantomData<V>
}
impl<K: hash::Hash + cmp::Eq, V> HashMap<K, V, collections::hash_map::RandomState> where K: ProcessTransfer, V: ProcessTransfer {
	/// Constructs a new, empty `HashMap`.
	pub fn new() -> HashMap<K, V, collections::hash_map::RandomState> {
		let mut leaf: Box<Leaf<u8>> = box unsafe{mem::uninitialized()};
		let (start, mut end) = Leaf::init(&mut leaf);
		let mut indices: Box<[LeafIndex; PARTITIONS-1]> = box unsafe{mem::uninitialized()};
		for i in 0..indices.len() {
			unsafe{ptr::write(&mut indices[i], end.clone_left())};
		}
		HashMap{hash_builder:Default::default(),list:leaf,start:start,end:end,indices:indices,phantom1:marker::PhantomData,phantom2:marker::PhantomData}
	}
}
impl<K, V, S> HashMap<K, V, S> where K: hash::Hash + cmp::Eq + ProcessTransfer, S: hash::BuildHasher, V: ProcessTransfer {
	/// Inserts a key-value pair into the map.
	///
	/// If the map did not have this key present, `None` is returned.
	///
	/// If the map did have this key present, the value is updated, and the old value is returned. The key is not updated, though; this matters for types that can be `==` without being identical. See the [module-level documentation] for more.
	///
	/// [module-level documentation]: index.html#insert-and-complex-keys
	///
	/// # Examples
	///
	/// ```
	/// use hadean_std::hashmap::HashMap;
	///
	/// let mut map = HashMap::new();
	/// assert_eq!(map.insert(37, "a"), None);
	/// assert_eq!(map.is_empty(), false);
	///
	/// map.insert(37, "b");
	/// assert_eq!(map.insert(37, "c"), Some("b"));
	/// assert_eq!(map[&37], "c");
	/// ```
	pub fn insert(&mut self, k: K, v: V) -> Option<V> {
		let partition = make_hash(&self.hash_builder, &k) as usize % PARTITIONS;
		let mut iter = if partition == 0 { self.start.clone_right() } else { self.indices[partition-1].clone_right() };
		let end = if partition == PARTITIONS-1 { self.end.clone_left() } else { self.indices[partition].clone_left() };
		while iter != end {
			let start = iter.clone_left();
			let mut key: K = unsafe{mem::uninitialized()};
			key.processsendable_read(&mut |buf,len| {
				let start2 = iter.clone_left();
				for _ in 0..len {
					self.list.increment(&mut iter);
				}
				let key = self.list.read(&start2, &iter);
				unsafe{ptr::copy_nonoverlapping(key.as_ptr(), buf, len)};
			});
			let mut start2 = iter.clone_left();
			let mut val: V = unsafe{mem::uninitialized()};
			val.processsendable_read(&mut |buf,len| {
				let start2 = iter.clone_left();
				for _ in 0..len {
					self.list.increment(&mut iter);
				}
				let val = self.list.read(&start2, &iter);
				unsafe{ptr::copy_nonoverlapping(val.as_ptr(), buf, len)};
			});
			if key == k {
				self.list.replace(&mut start2, &mut iter, iter::empty());
				v.processsendable_write(&mut |buf,len| {
					let mut start = iter.clone_left();
					self.list.replace(&mut start, &mut iter, unsafe{slice::from_raw_parts(buf, len)}.iter().map(|&x|x));
				});
				return Some(val);
			}
		}
		k.processsendable_write(&mut |buf,len| {
			let mut start = iter.clone_left();
			self.list.replace(&mut start, &mut iter, unsafe{slice::from_raw_parts(buf, len)}.iter().map(|&x|x));
		});
		v.processsendable_write(&mut |buf,len| {
			let mut start = iter.clone_left();
			self.list.replace(&mut start, &mut iter, unsafe{slice::from_raw_parts(buf, len)}.iter().map(|&x|x));
		});
		None
	}
    /// Returns a reference to the value corresponding to the key.
    ///
    /// The key may be any borrowed form of the map's key type, but `Hash` and `Eq` on the borrowed form *must* match those for the key type.
    ///
    /// # Examples
    ///
    /// ```
    /// use hadean_std::hashmap::HashMap;
    ///
    /// let mut map = HashMap::new();
    /// map.insert(1, "a");
    /// assert_eq!(map.get(&1), Some(&"a"));
    /// assert_eq!(map.get(&2), None);
    /// ```
	pub fn get<Q: ?Sized>(&mut self, k: &Q) -> Option<V> where K: borrow::Borrow<Q>, Q: hash::Hash + cmp::Eq {
		let partition = make_hash(&self.hash_builder, &k) as usize % PARTITIONS;
		let mut iter = if partition == 0 { self.start.clone_right() } else { self.indices[partition-1].clone_right() };
		let end = if partition == PARTITIONS-1 { self.end.clone_left() } else { self.indices[partition].clone_left() };
		while iter != end {
			let mut key: K = unsafe{mem::uninitialized()};
			key.processsendable_read(&mut |buf,len| {
				let start2 = iter.clone_left();
				for _ in 0..len {
					self.list.increment(&mut iter);
				}
				let key = self.list.read(&start2, &iter);
				unsafe{ptr::copy_nonoverlapping(key.as_ptr(), buf, len)};
			});
			let mut val: V = unsafe{mem::uninitialized()};
			val.processsendable_read(&mut |buf,len| {
				let start2 = iter.clone_left();
				for _ in 0..len {
					self.list.increment(&mut iter);
				}
				let val = self.list.read(&start2, &iter);
				unsafe{ptr::copy_nonoverlapping(val.as_ptr(), buf, len)};
			});
			if key.borrow() == k {
				return Some(val); // should be &val... but that's not possible here
			}
		}
		None
	}
	// /// Append `val` to the dataframe.
	// pub fn push(&mut self, val: T) {
	// 	val.processsendable_write(&mut |buf,len| {
	// 		let mut end = self.end.clone_left();
	// 		let mut end2 = self.end.clone_left();
	// 		self.list.replace(&mut end, &mut end2, unsafe{slice::from_raw_parts(buf, len)}.iter().map(|&x|x));
	// 	});
	// }
	// /// Map each element of the dataframe using `f`.
	// pub fn map<F,T1>(mut self, f: F) -> HashMap<T1> where F: Fn(T) -> T1 + ProcessTransfer, T1: ProcessTransfer {
	// 	let mut iter = self.start.clone_right();
	// 	while &iter != self.list.end() {
	// 		// println!("mapping");
	// 		let mut start = iter.clone_left();
	// 		let mut res: T = unsafe{mem::uninitialized()};
	// 		res.processsendable_read(&mut |buf,len| {
	// 			let start2 = iter.clone_left();
	// 			for _ in 0..len {
	// 				self.list.increment(&mut iter);
	// 			}
	// 			let res = self.list.read(&start2, &iter);
	// 			unsafe{ptr::copy_nonoverlapping(res.as_ptr(), buf, len)};
	// 		});
	// 		// println!("a: {}", res);
	// 		let res = f(res);
	// 		// println!("b: {}", res);
	// 		self.list.replace(&mut start, &mut iter, iter::empty());
	// 		res.processsendable_write(&mut |buf,len| {
	// 			let mut start = iter.clone_left();
	// 			self.list.replace(&mut start, &mut iter, unsafe{slice::from_raw_parts(buf, len)}.iter().map(|&x|x));
	// 		});
	// 		// println!("written");
	// 	}
	// 	HashMap{list:self.list,phantom:marker::PhantomData}
	// }
	// /// Filter elements of the dataframe, keeping only those that satisfy `f`.
	// pub fn filter<F>(&mut self, f: F) where F: Fn(&T) -> bool + ProcessTransfer {
	// 	let mut iter = self.start.clone_right();
	// 	while &iter != self.list.end() {
	// 		// println!("mapping");
	// 		let mut start = iter.clone_left();
	// 		let mut res: T = unsafe{mem::uninitialized()};
	// 		res.processsendable_read(&mut |buf,len| {
	// 			let start2 = iter.clone_left();
	// 			for _ in 0..len {
	// 				self.list.increment(&mut iter);
	// 			}
	// 			let res = self.list.read(&start2, &iter);
	// 			unsafe{ptr::copy_nonoverlapping(res.as_ptr(), buf, len)};
	// 		});
	// 		// println!("a: {}", res);
	// 		if !f(&res) {
	// 			self.list.replace(&mut start, &mut iter, iter::empty());
	// 		}
	// 		// println!("written");
	// 	}
	// }
	// /// Reduce the dataframe into one value.
	// ///
	// /// # Examples
	// ///
	// /// ```
	// /// let mut data_frame: HashMap<String> = HashMap::new();
	// /// data_frame.push(String::from("http://google.com"));
	// /// data_frame.push(String::from("http://bbc.co.uk"));
	// /// data_frame.push(String::from("https://hadean.com"));
	// /// let total_length = data_frame.reduce(0, |acc, val| acc + val.len() as u64);
	// /// ```
	// ///
	// pub fn reduce<F,T1>(&mut self, initial: T1, f: F) -> T1 where F: Fn(T1, &T) -> T1 + ProcessTransfer, T1: ProcessTransfer {
	// 	let mut acc = initial;
	// 	let mut iter = self.start.clone_right();
	// 	while &iter != self.list.end() {
	// 		// println!("mapping");
	// 		let mut res: T = unsafe{mem::uninitialized()};
	// 		res.processsendable_read(&mut |buf,len| {
	// 			let start2 = iter.clone_left();
	// 			for _ in 0..len {
	// 				self.list.increment(&mut iter);
	// 			}
	// 			let res = self.list.read(&start2, &iter);
	// 			unsafe{ptr::copy_nonoverlapping(res.as_ptr(), buf, len)};
	// 		});
	// 		// println!("a: {}", res);
	// 		acc = f(acc, &res);
	// 		// println!("written");
	// 	}
	// 	acc
	// }
	// pub fn iter(&mut self) -> HashMapIter<K,V> {
	// 	let iter = self.start.clone_right();
	// 	HashMapIter{list:&mut self.list,iter:iter,phantom1:marker::PhantomData,phantom2:marker::PhantomData}
	// }
}

pub struct HashMapIter<'a,K,V> where K: 'a + hash::Hash + cmp::Eq + ProcessTransfer, V: 'a + ProcessTransfer {
	list: &'a mut Box<Leaf<u8>>,
	iter: LeafIndex,
	phantom1: marker::PhantomData<K>,
	phantom2: marker::PhantomData<V>
}
impl<'a,K,V> iter::Iterator for HashMapIter<'a,K,V> where K: hash::Hash + cmp::Eq + ProcessTransfer, V: ProcessTransfer {
	type Item = (K,V); // .iter() should be over &T... but that's not possible here
	fn next(&mut self) -> Option<Self::Item> {
		// if &self.iter == self.list.list.end() {
			None
		// } else {
		// 	let mut res: T = unsafe{mem::uninitialized()};
		// 	res.processsendable_read(&mut |buf,len| {
		// 		let iter = self.iter.clone_left();
		// 		for _ in 0..len {
		// 			self.list.list.increment(&mut self.iter);
		// 		}
		// 		let res = self.list.list.read(&iter, &self.iter);
		// 		unsafe{ptr::copy_nonoverlapping(res.as_ptr(), buf, len)};
		// 	});
		// 	Some(res)
		// }
	}
}

#[cfg(test)]
mod tests {
	use super::*;
	#[test]
	// #[ignore]
	fn hashmap() {
		let mut hash_map: HashMap<String,String> = HashMap::new();
		hash_map.insert(String::from("abc"),String::from("123"));
		hash_map.insert(String::from("def"),String::from("456"));
		// hash_map.push(String::from("http://bbc.co.uk"));
		// hash_map.push(String::from("http://hadean.com"));
		// let mut hash_map: HashMap<String> = hash_map.map(|mut val|{val.push_str("/xyz");val});
		// hash_map.filter(|val|val.len() > 20);
		for (key,value) in hash_map.iter() {
			println!("{}:{}", key, value);
		}
		// let total_length = hash_map.reduce(0u64, |acc, val| acc + val.len() as u64);
		// println!("{}", total_length);
	}
}