use crate::error::{Error, Result};
use crate::key_composer::KeyTag;
use crate::meta::{KeyMeta, RedisType, generate_version};
#[derive(Debug, Clone, Copy, PartialEq, bitcode::Encode, bitcode::Decode)]
pub struct BloomChainMeta {
pub base: KeyMeta,
pub n_filters: u16,
pub expansion: u16,
pub base_capacity: u32,
pub error_rate: f64,
pub bloom_bytes: u32,
}
impl BloomChainMeta {
pub const ENCODED_SIZE: usize = KeyMeta::ENCODED_SIZE + 2 + 2 + 4 + 8 + 4;
#[inline]
pub fn new(
base_capacity: u32,
error_rate: f64,
expansion: u16,
version: u64,
expire_at: u64,
bloom_bytes: u32,
) -> Self {
let ver = if version == 0 {
generate_version()
} else {
version
};
Self {
base: KeyMeta::new(RedisType::Bloom, expire_at, ver, 0),
n_filters: 1,
expansion,
base_capacity,
error_rate,
bloom_bytes,
}
}
#[inline]
pub const fn is_scaling(&self) -> bool {
self.expansion != 0
}
#[inline]
pub fn is_expired(&self, now_ms: u64) -> bool {
self.base.is_expired(now_ms)
}
#[inline]
pub fn get_capacity(&self) -> u32 {
if self.expansion == 0 {
return self.base_capacity;
}
if self.expansion == 1 {
return self.base_capacity.saturating_mul(self.n_filters as u32);
}
let r = self.expansion as u64;
let n = self.n_filters as u32;
if let Some(r_pow_n) = r.checked_pow(n) {
let sum =
(self.base_capacity as u64).saturating_mul(r_pow_n.saturating_sub(1)) / (r - 1);
sum.min(u32::MAX as u64) as u32
} else {
let mut total = 0u64;
let mut cur = self.base_capacity as u64;
for _ in 0..self.n_filters {
total = total.saturating_add(cur);
cur = cur.saturating_mul(r);
if total >= u32::MAX as u64 {
return u32::MAX;
}
}
total.min(u32::MAX as u64) as u32
}
}
#[inline]
pub fn sub_filter_capacity(&self, filter_index: u16) -> u32 {
if self.expansion <= 1 || filter_index == 0 {
self.base_capacity
} else {
(self.base_capacity as u64)
.saturating_mul((self.expansion as u64).saturating_pow(filter_index as u32))
.min(u32::MAX as u64) as u32
}
}
#[inline]
pub fn encode(&self) -> [u8; Self::ENCODED_SIZE] {
let mut buf = [0u8; Self::ENCODED_SIZE];
buf[..KeyMeta::ENCODED_SIZE].copy_from_slice(&self.base.encode());
let mut offset = KeyMeta::ENCODED_SIZE;
buf[offset..offset + 2].copy_from_slice(&self.n_filters.to_be_bytes());
offset += 2;
buf[offset..offset + 2].copy_from_slice(&self.expansion.to_be_bytes());
offset += 2;
buf[offset..offset + 4].copy_from_slice(&self.base_capacity.to_be_bytes());
offset += 4;
buf[offset..offset + 8].copy_from_slice(&self.error_rate.to_be_bytes());
offset += 8;
buf[offset..offset + 4].copy_from_slice(&self.bloom_bytes.to_be_bytes());
buf
}
#[inline]
pub fn decode(bytes: &[u8]) -> Option<Self> {
if bytes.len() < Self::ENCODED_SIZE {
return None;
}
let base = KeyMeta::decode(&bytes[..KeyMeta::ENCODED_SIZE])?;
let mut offset = KeyMeta::ENCODED_SIZE;
let mut b2 = [0u8; 2];
b2.copy_from_slice(&bytes[offset..offset + 2]);
let n_filters = u16::from_be_bytes(b2);
offset += 2;
b2.copy_from_slice(&bytes[offset..offset + 2]);
let expansion = u16::from_be_bytes(b2);
offset += 2;
let mut b4 = [0u8; 4];
b4.copy_from_slice(&bytes[offset..offset + 4]);
let base_capacity = u32::from_be_bytes(b4);
offset += 4;
let mut b8 = [0u8; 8];
b8.copy_from_slice(&bytes[offset..offset + 8]);
let error_rate = f64::from_be_bytes(b8);
offset += 8;
b4.copy_from_slice(&bytes[offset..offset + 4]);
let bloom_bytes = u32::from_be_bytes(b4);
Some(Self {
base,
n_filters,
expansion,
base_capacity,
error_rate,
bloom_bytes,
})
}
}
#[derive(Debug, Clone, Copy, PartialEq, bitcode::Encode, bitcode::Decode)]
pub struct CuckooChainMeta {
pub base: KeyMeta,
pub n_filters: u16,
pub expansion: u16,
pub base_capacity: u64,
pub bucket_size: u8,
pub max_iterations: u16,
pub num_deleted_items: u64,
pub page_size: u32,
}
impl CuckooChainMeta {
pub const ENCODED_SIZE: usize = KeyMeta::ENCODED_SIZE + 2 + 2 + 8 + 1 + 2 + 8 + 4;
#[inline]
pub fn new(
base_capacity: u64,
bucket_size: u8,
max_iterations: u16,
expansion: u16,
page_size: u32,
version: u64,
expire_at: u64,
) -> Self {
let ver = if version == 0 {
generate_version()
} else {
version
};
let normalized_expansion = super::CuckooFilterHelper::normalize_expansion(expansion);
Self {
base: KeyMeta::new(RedisType::CuckooFilter, expire_at, ver, 0),
n_filters: 1,
expansion: normalized_expansion,
base_capacity,
bucket_size,
max_iterations,
num_deleted_items: 0,
page_size,
}
}
#[inline]
pub const fn is_scaling(&self) -> bool {
self.expansion > 0
}
#[inline]
pub fn is_expired(&self, now_ms: u64) -> bool {
self.base.is_expired(now_ms)
}
#[inline]
pub fn get_total_capacity(&self) -> u64 {
if self.expansion == 0 || self.n_filters == 1 {
return self.base_capacity;
}
if self.expansion == 1 {
return self.base_capacity.saturating_mul(self.n_filters as u64);
}
let r = self.expansion as u64;
let n = self.n_filters as u32;
if let Some(r_pow_n) = r.checked_pow(n) {
(self.base_capacity as u128)
.saturating_mul((r_pow_n.saturating_sub(1)) as u128)
.checked_div((r - 1) as u128)
.map(|v| v.min(u64::MAX as u128) as u64)
.unwrap_or(u64::MAX)
} else {
let mut total = 0u64;
let mut filter_cap = self.base_capacity;
for _ in 0..self.n_filters {
total = total.saturating_add(filter_cap);
filter_cap = filter_cap.saturating_mul(r);
if total == u64::MAX {
return u64::MAX;
}
}
total
}
}
#[inline]
pub fn sub_filter_capacity(&self, filter_index: u16) -> Option<u64> {
if self.expansion == 0 {
return if filter_index == 0 {
Some(self.base_capacity)
} else {
None
};
}
if self.expansion == 1 || filter_index == 0 {
return Some(self.base_capacity);
}
(self.expansion as u64)
.checked_pow(filter_index as u32)
.and_then(|exp| self.base_capacity.checked_mul(exp))
}
#[inline]
pub fn sub_filter_num_buckets(&self, filter_index: u16) -> Result<u32> {
let cap = self
.sub_filter_capacity(filter_index)
.ok_or_else(|| Error::invalid_data("filter capacity overflow"))?;
super::CuckooFilterHelper::calculate_required_buckets(cap, self.bucket_size)
}
#[inline]
pub fn encode(&self) -> [u8; Self::ENCODED_SIZE] {
let mut buf = [0u8; Self::ENCODED_SIZE];
buf[..KeyMeta::ENCODED_SIZE].copy_from_slice(&self.base.encode());
let mut offset = KeyMeta::ENCODED_SIZE;
buf[offset..offset + 2].copy_from_slice(&self.n_filters.to_be_bytes());
offset += 2;
buf[offset..offset + 2].copy_from_slice(&self.expansion.to_be_bytes());
offset += 2;
buf[offset..offset + 8].copy_from_slice(&self.base_capacity.to_be_bytes());
offset += 8;
buf[offset] = self.bucket_size;
offset += 1;
buf[offset..offset + 2].copy_from_slice(&self.max_iterations.to_be_bytes());
offset += 2;
buf[offset..offset + 8].copy_from_slice(&self.num_deleted_items.to_be_bytes());
offset += 8;
buf[offset..offset + 4].copy_from_slice(&self.page_size.to_be_bytes());
buf
}
#[inline]
pub fn decode(bytes: &[u8]) -> Option<Self> {
if bytes.len() < Self::ENCODED_SIZE {
return None;
}
let base = KeyMeta::decode(&bytes[..KeyMeta::ENCODED_SIZE])?;
let mut offset = KeyMeta::ENCODED_SIZE;
let mut b2 = [0u8; 2];
b2.copy_from_slice(&bytes[offset..offset + 2]);
let n_filters = u16::from_be_bytes(b2);
offset += 2;
b2.copy_from_slice(&bytes[offset..offset + 2]);
let expansion = u16::from_be_bytes(b2);
offset += 2;
let mut b8 = [0u8; 8];
b8.copy_from_slice(&bytes[offset..offset + 8]);
let base_capacity = u64::from_be_bytes(b8);
offset += 8;
let bucket_size = bytes[offset];
offset += 1;
b2.copy_from_slice(&bytes[offset..offset + 2]);
let max_iterations = u16::from_be_bytes(b2);
offset += 2;
b8.copy_from_slice(&bytes[offset..offset + 8]);
let num_deleted_items = u64::from_be_bytes(b8);
offset += 8;
let mut b4 = [0u8; 4];
b4.copy_from_slice(&bytes[offset..offset + 4]);
let page_size = u32::from_be_bytes(b4);
Some(Self {
base,
n_filters,
expansion,
base_capacity,
bucket_size,
max_iterations,
num_deleted_items,
page_size,
})
}
}
impl_meta_ops!(BloomChainMeta, KeyTag::BloomMeta.as_slice());
impl_meta_ops!(CuckooChainMeta, KeyTag::CuckooMeta.as_slice());