use std::hash::Hash;
use std::marker::PhantomData;
use crate::Key;
use crate::compaction::CompactionIndex;
use crate::config::Config;
use crate::const_map::{ConstMap, ConstMapShard};
use crate::durability::{Bitcask, Durability, Fixed};
use crate::error::{DbError, DbResult};
use crate::fixed::config::FixedConfig;
use crate::hook::{NoHook, TypedWriteHook, WriteHook, ZeroHookAdapter};
use crate::zero_tree::{from_value_bytes, to_bytes};
pub struct ZeroMap<
K: Key + Send + Sync + Hash + Eq,
const V: usize,
T: Copy + zerocopy::IntoBytes + zerocopy::FromBytes + zerocopy::Immutable = [u8; V],
H: TypedWriteHook<K, T> = NoHook,
D: Durability = Bitcask,
> {
inner: ConstMap<K, V, ZeroHookAdapter<K, T, H>, D>,
_marker: PhantomData<T>,
}
impl<
K: Key + Send + Sync + Hash + Eq,
const V: usize,
T: Copy + zerocopy::IntoBytes + zerocopy::FromBytes + zerocopy::Immutable,
> ZeroMap<K, V, T, NoHook, Bitcask>
{
pub fn open(path: impl AsRef<std::path::Path>, config: Config) -> DbResult<Self> {
const { assert!(size_of::<T>() == V) }
let adapter = ZeroHookAdapter {
inner: NoHook,
_marker: PhantomData,
};
Ok(Self {
inner: ConstMap::open_hooked(path, config, adapter)?,
_marker: PhantomData,
})
}
}
impl<
K: Key + Send + Sync + Hash + Eq,
const V: usize,
T: Copy + zerocopy::IntoBytes + zerocopy::FromBytes + zerocopy::Immutable,
H: TypedWriteHook<K, T>,
> ZeroMap<K, V, T, H, Bitcask>
{
pub fn open_hooked(
path: impl AsRef<std::path::Path>,
config: Config,
hook: H,
) -> DbResult<Self> {
const { assert!(size_of::<T>() == V) }
let adapter = ZeroHookAdapter {
inner: hook,
_marker: PhantomData,
};
Ok(Self {
inner: ConstMap::open_hooked(path, config, adapter)?,
_marker: PhantomData,
})
}
pub fn close(self) -> DbResult<()> {
self.inner.close()
}
pub fn flush_buffers(&self) -> DbResult<()> {
self.inner.flush_buffers()
}
pub fn config(&self) -> &Config {
self.inner.config()
}
pub fn compact(&self) -> DbResult<usize> {
self.inner.compact()
}
pub fn sync_hints(&self) -> DbResult<()> {
self.inner.sync_hints()
}
pub fn as_inner(&self) -> &ConstMap<K, V, ZeroHookAdapter<K, T, H>, Bitcask> {
&self.inner
}
}
impl<
K: Key + Send + Sync + Hash + Eq,
const V: usize,
T: Copy + zerocopy::IntoBytes + zerocopy::FromBytes + zerocopy::Immutable,
> ZeroMap<K, V, T, NoHook, Fixed>
{
pub fn open(path: impl AsRef<std::path::Path>, config: FixedConfig) -> DbResult<Self> {
const { assert!(size_of::<T>() == V) }
let adapter = ZeroHookAdapter {
inner: NoHook,
_marker: PhantomData,
};
Ok(Self {
inner: ConstMap::open_with_hook(path, config, adapter)?,
_marker: PhantomData,
})
}
}
impl<
K: Key + Send + Sync + Hash + Eq,
const V: usize,
T: Copy + zerocopy::IntoBytes + zerocopy::FromBytes + zerocopy::Immutable,
H: TypedWriteHook<K, T>,
> ZeroMap<K, V, T, H, Fixed>
{
pub fn open_with_hook(
path: impl AsRef<std::path::Path>,
config: FixedConfig,
hook: H,
) -> DbResult<Self> {
const { assert!(size_of::<T>() == V) }
let adapter = ZeroHookAdapter {
inner: hook,
_marker: PhantomData,
};
Ok(Self {
inner: ConstMap::open_with_hook(path, config, adapter)?,
_marker: PhantomData,
})
}
pub fn close(self) -> DbResult<()> {
self.inner.close()
}
}
impl<
K: Key + Send + Sync + Hash + Eq,
const V: usize,
T: Copy + zerocopy::IntoBytes + zerocopy::FromBytes + zerocopy::Immutable,
H: TypedWriteHook<K, T>,
D: Durability,
> ZeroMap<K, V, T, H, D>
{
pub fn get(&self, key: &K) -> Option<T> {
let bytes = self.inner.get(key)?;
Some(from_value_bytes::<V, T>(&bytes))
}
pub fn get_or_err(&self, key: &K) -> DbResult<T> {
self.get(key).ok_or(DbError::KeyNotFound)
}
pub fn contains(&self, key: &K) -> bool {
self.inner.contains(key)
}
pub fn put(&self, key: &K, value: &T) -> DbResult<Option<T>> {
let bytes = to_bytes::<V, T>(value);
self.inner
.put(key, &bytes)
.map(|opt| opt.map(|b| from_value_bytes::<V, T>(&b)))
}
pub fn insert(&self, key: &K, value: &T) -> DbResult<()> {
let bytes = to_bytes::<V, T>(value);
self.inner.insert(key, &bytes)
}
pub fn delete(&self, key: &K) -> DbResult<Option<T>> {
self.inner
.delete(key)
.map(|opt| opt.map(|b| from_value_bytes::<V, T>(&b)))
}
pub fn cas(&self, key: &K, expected: &T, new_value: &T) -> DbResult<()> {
let exp_bytes = to_bytes::<V, T>(expected);
let new_bytes = to_bytes::<V, T>(new_value);
self.inner.cas(key, &exp_bytes, &new_bytes)
}
pub fn compare_delete(&self, key: &K, expected: &T) -> DbResult<()> {
let exp_bytes = to_bytes::<V, T>(expected);
self.inner.compare_delete(key, &exp_bytes)
}
pub fn update(&self, key: &K, f: impl FnOnce(&T) -> T) -> DbResult<Option<T>> {
self.inner
.update(key, |bytes| {
let val = from_value_bytes::<V, T>(bytes);
let new_val = f(&val);
to_bytes::<V, T>(&new_val)
})
.map(|opt| opt.map(|b| from_value_bytes::<V, T>(&b)))
}
pub fn fetch_update(&self, key: &K, f: impl FnOnce(&T) -> T) -> DbResult<Option<T>> {
self.inner
.fetch_update(key, |bytes| {
let val = from_value_bytes::<V, T>(bytes);
let new_val = f(&val);
to_bytes::<V, T>(&new_val)
})
.map(|opt| opt.map(|b| from_value_bytes::<V, T>(&b)))
}
pub fn atomic<R>(
&self,
shard_key: &K,
f: impl FnOnce(&mut ZeroMapShard<'_, K, V, T, D, ZeroHookAdapter<K, T, H>>) -> DbResult<R>,
) -> DbResult<R> {
self.inner.atomic(shard_key, |const_shard| {
let shard = unsafe {
&mut *(const_shard as *mut ConstMapShard<'_, K, V, ZeroHookAdapter<K, T, H>, D>
as *mut ZeroMapShard<'_, K, V, T, D, ZeroHookAdapter<K, T, H>>)
};
f(shard)
})
}
pub fn len(&self) -> usize {
self.inner.len()
}
pub fn is_empty(&self) -> bool {
self.inner.is_empty()
}
pub fn shard_for(&self, key: &K) -> usize {
self.inner.shard_for(key)
}
pub fn flush(&self) -> DbResult<()> {
self.inner.flush()
}
pub fn migrate(&self, f: impl Fn(&K, &T) -> crate::MigrateAction<T>) -> DbResult<usize> {
self.inner.migrate(|key, bytes| {
let val: T = from_value_bytes(bytes);
match f(key, &val) {
crate::MigrateAction::Keep => crate::MigrateAction::Keep,
crate::MigrateAction::Update(new) => {
crate::MigrateAction::Update(to_bytes::<V, T>(&new))
}
crate::MigrateAction::Delete => crate::MigrateAction::Delete,
}
})
}
pub fn for_each(&self, mut f: impl FnMut(&K, &T)) {
self.inner.for_each(|key, bytes| {
let val: T = from_value_bytes(bytes);
f(key, &val);
});
}
pub(crate) fn replay_init(&self) {
self.inner.replay_init();
}
}
impl<
K: Key + Send + Sync + Hash + Eq,
const V: usize,
T: Copy + zerocopy::IntoBytes + zerocopy::FromBytes + zerocopy::Immutable + Send + Sync,
H: TypedWriteHook<K, T>,
> CompactionIndex<K> for ZeroMap<K, V, T, H, Bitcask>
{
fn update_if_match(
&self,
key: &K,
old_loc: crate::disk_loc::DiskLoc,
new_loc: crate::disk_loc::DiskLoc,
) -> bool {
self.inner.update_if_match(key, old_loc, new_loc)
}
fn contains_key(&self, key: &K) -> bool {
self.contains(key)
}
}
#[repr(transparent)]
pub struct ZeroMapShard<
'a,
K: Key + Send + Sync + Hash + Eq,
const V: usize,
T: Copy + zerocopy::IntoBytes + zerocopy::FromBytes + zerocopy::Immutable = [u8; V],
D: Durability = Bitcask,
WH: WriteHook<K> = NoHook,
> {
inner: ConstMapShard<'a, K, V, WH, D>,
_marker: PhantomData<T>,
}
impl<
K: Key + Send + Sync + Hash + Eq,
const V: usize,
T: Copy + zerocopy::IntoBytes + zerocopy::FromBytes + zerocopy::Immutable,
D: Durability,
WH: WriteHook<K>,
> ZeroMapShard<'_, K, V, T, D, WH>
{
pub fn put(&mut self, key: &K, value: &T) -> DbResult<Option<T>> {
let bytes = to_bytes::<V, T>(value);
self.inner
.put(key, &bytes)
.map(|opt| opt.map(|b| from_value_bytes::<V, T>(&b)))
}
pub fn insert(&mut self, key: &K, value: &T) -> DbResult<()> {
let bytes = to_bytes::<V, T>(value);
self.inner.insert(key, &bytes)
}
pub fn delete(&mut self, key: &K) -> DbResult<Option<T>> {
self.inner
.delete(key)
.map(|opt| opt.map(|b| from_value_bytes::<V, T>(&b)))
}
pub fn get(&self, key: &K) -> Option<T> {
let bytes = self.inner.get(key)?;
Some(from_value_bytes::<V, T>(&bytes))
}
pub fn get_or_err(&self, key: &K) -> DbResult<T> {
self.get(key).ok_or(DbError::KeyNotFound)
}
pub fn contains(&self, key: &K) -> bool {
self.inner.contains(key)
}
}
#[cfg(feature = "armour")]
impl<T, const V: usize, H> crate::armour::collection::Collection
for ZeroMap<T::SelfId, V, T, H, Bitcask>
where
T: crate::CollectionMeta
+ Copy
+ zerocopy::IntoBytes
+ zerocopy::FromBytes
+ zerocopy::Immutable
+ Send
+ Sync,
H: crate::hook::TypedWriteHook<T::SelfId, T>,
T::SelfId: crate::Key + Send + Sync + std::hash::Hash + Eq,
{
fn name(&self) -> &str {
T::NAME
}
fn len(&self) -> usize {
self.len()
}
fn compact(&self) -> crate::DbResult<usize> {
self.compact()
}
fn flush(&self) -> crate::DbResult<()> {
self.flush_buffers()?;
self.sync_hints()?;
Ok(())
}
fn periodic_flush(&self) -> crate::DbResult<()> {
self.flush_buffers()
}
}
#[cfg(feature = "armour")]
impl<T, const V: usize, H> crate::armour::collection::Collection
for ZeroMap<T::SelfId, V, T, H, Fixed>
where
T: crate::CollectionMeta
+ Copy
+ zerocopy::IntoBytes
+ zerocopy::FromBytes
+ zerocopy::Immutable
+ Send
+ Sync,
H: crate::hook::TypedWriteHook<T::SelfId, T>,
T::SelfId: crate::Key + Send + Sync + std::hash::Hash + Eq,
{
fn name(&self) -> &str {
T::NAME
}
fn len(&self) -> usize {
self.len()
}
fn compact(&self) -> crate::DbResult<usize> {
Ok(0)
}
fn flush(&self) -> crate::DbResult<()> {
self.flush()
}
fn periodic_flush(&self) -> crate::DbResult<()> {
self.flush()
}
}
#[cfg(feature = "armour")]
pub struct ZeroMapTx<'a, K, const V: usize, T, H, D>
where
K: Key + Send + Sync + Hash + Eq,
T: Copy + zerocopy::IntoBytes + zerocopy::FromBytes + zerocopy::Immutable,
H: TypedWriteHook<K, T>,
D: Durability,
{
inner: crate::const_map::ConstMapTx<'a, K, V, ZeroHookAdapter<K, T, H>, D>,
}
#[cfg(feature = "armour")]
impl<K, const V: usize, T, H, D> ZeroMapTx<'_, K, V, T, H, D>
where
K: Key + Send + Sync + Hash + Eq,
T: Copy + zerocopy::IntoBytes + zerocopy::FromBytes + zerocopy::Immutable,
H: TypedWriteHook<K, T>,
D: Durability,
{
pub fn try_get(&self, key: &K) -> DbResult<Option<T>> {
Ok(self
.inner
.try_get(key)?
.map(|b| from_value_bytes::<V, T>(&b)))
}
pub fn try_contains(&self, key: &K) -> DbResult<bool> {
self.inner.try_contains(key)
}
pub fn get_or_err(&self, key: &K) -> DbResult<T> {
self.try_get(key)?.ok_or(DbError::KeyNotFound)
}
pub fn put(&mut self, key: &K, value: &T) -> DbResult<Option<T>> {
let bytes = to_bytes::<V, T>(value);
self.inner
.put(key, &bytes)
.map(|opt| opt.map(|b| from_value_bytes::<V, T>(&b)))
}
pub fn insert(&mut self, key: &K, value: &T) -> DbResult<()> {
let bytes = to_bytes::<V, T>(value);
self.inner.insert(key, &bytes)
}
pub fn delete(&mut self, key: &K) -> DbResult<Option<T>> {
self.inner
.delete(key)
.map(|opt| opt.map(|b| from_value_bytes::<V, T>(&b)))
}
}
#[cfg(feature = "armour")]
impl<K, const V: usize, T, H, D> crate::armour::multi_tx::MultiTx for ZeroMap<K, V, T, H, D>
where
K: Key + Send + Sync + Hash + Eq,
T: Copy + zerocopy::IntoBytes + zerocopy::FromBytes + zerocopy::Immutable,
H: TypedWriteHook<K, T>,
D: Durability,
{
type Key = K;
type Tx<'a>
= ZeroMapTx<'a, K, V, T, H, D>
where
Self: 'a;
fn shard_for_key(&self, key: &K) -> usize {
self.inner.shard_for(key)
}
fn begin_tx(&self) -> ZeroMapTx<'_, K, V, T, H, D> {
ZeroMapTx {
inner: crate::armour::multi_tx::MultiTx::begin_tx(&self.inner),
}
}
fn lock_shard_into<'a>(&'a self, shard_id: usize, tx: &mut ZeroMapTx<'a, K, V, T, H, D>) {
crate::armour::multi_tx::MultiTx::lock_shard_into(&self.inner, shard_id, &mut tx.inner)
}
fn release_locks(
&self,
tx: &mut ZeroMapTx<'_, K, V, T, H, D>,
) -> crate::armour::multi_tx::SyncNeeds {
crate::armour::multi_tx::MultiTx::release_locks(&self.inner, &mut tx.inner)
}
fn run_sync(&self, needs: crate::armour::multi_tx::SyncNeeds) -> DbResult<()> {
crate::armour::multi_tx::MultiTx::run_sync(&self.inner, needs)
}
fn replay_hooks(&self, tx: ZeroMapTx<'_, K, V, T, H, D>) {
crate::armour::multi_tx::MultiTx::replay_hooks(&self.inner, tx.inner)
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::Config;
use crate::hook::TypedWriteHook;
use std::collections::HashMap as StdHashMap;
use std::mem::size_of;
use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering as AO};
use tempfile::tempdir;
use zerocopy::{FromBytes, Immutable, IntoBytes, KnownLayout};
#[derive(
FromBytes, IntoBytes, Immutable, KnownLayout, Clone, Copy, Debug, PartialEq, Eq, Default,
)]
#[repr(C)]
struct Counters {
views: u64,
likes: u64,
}
type Map = ZeroMap<[u8; 8], { size_of::<Counters>() }, Counters>;
fn open(dir: &std::path::Path) -> Map {
ZeroMap::<[u8; 8], { size_of::<Counters>() }, Counters>::open(dir, Config::test()).unwrap()
}
fn key_on_other_shard(map: &Map, anchor: &[u8; 8]) -> [u8; 8] {
let anchor_shard = map.shard_for(anchor);
(2u64..1000)
.map(|i| i.to_be_bytes())
.find(|k| map.shard_for(k) != anchor_shard)
.expect("key on different shard")
}
#[test]
fn for_each_visits_all_entries() {
let dir = tempdir().unwrap();
let map = open(dir.path());
let key_a = 1u64.to_be_bytes();
let key_b = key_on_other_shard(&map, &key_a);
let key_c = key_on_other_shard(&map, &key_b);
let expected = StdHashMap::from([
(
key_a,
Counters {
views: 10,
likes: 1,
},
),
(
key_b,
Counters {
views: 20,
likes: 2,
},
),
(
key_c,
Counters {
views: 30,
likes: 3,
},
),
]);
for (key, value) in &expected {
map.put(key, value).unwrap();
}
let mut seen = StdHashMap::new();
map.for_each(|key, value| {
seen.insert(*key, *value);
});
assert_eq!(seen, expected);
assert_ne!(map.shard_for(&key_a), map.shard_for(&key_b));
}
struct ZHook {
writes: Arc<AtomicUsize>,
last_new: Arc<crate::sync::Mutex<Option<u32>>>,
}
impl ZHook {
fn new() -> Self {
Self {
writes: Arc::new(AtomicUsize::new(0)),
last_new: Arc::new(crate::sync::Mutex::new(None)),
}
}
}
impl TypedWriteHook<[u8; 8], u32> for ZHook {
fn on_write(&self, _k: &[u8; 8], _old: Option<&u32>, new: Option<&u32>) {
self.writes.fetch_add(1, AO::Relaxed);
*crate::sync::lock(&self.last_new) = new.copied();
}
}
#[test]
fn zero_map_atomic_fires_typed_hook() {
let dir = tempfile::tempdir().unwrap();
let mut cfg = Config::test();
cfg.shard_count = 1;
let hook = ZHook::new();
let writes = Arc::clone(&hook.writes);
let last_new = Arc::clone(&hook.last_new);
let map = ZeroMap::<[u8; 8], 4, u32, _>::open_hooked(dir.path(), cfg, hook).unwrap();
let k = 1u64.to_be_bytes();
map.atomic(&k, |s| {
s.put(&k, &42u32)?;
Ok(())
})
.unwrap();
assert_eq!(writes.load(AO::Relaxed), 1);
assert_eq!(*crate::sync::lock(&last_new), Some(42));
}
}