Mirror/brk
1
0
Fork 0
mirror of https://github.com/bitcoinresearchkit/brk.git synced 2026-07-28 03:08:10 -07:00
Code Issues Packages Projects Releases Wiki Activity

vec: rework part 1

Browse Source
This commit is contained in:
nym21
2025-04-09 16:31:31 +02:00
parent a7f2b24bac
commit 962254e511
29 changed files with 1627 additions and 946 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+8
View File
@@ -138,6 +138,12 @@ dependencies = [
"derive_arbitrary", "derive_arbitrary",
] ]
[[package]]
name = "arc-swap"
version = "1.7.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "69f7f8c3906b62b754cd5326047894316021dcfe5a194c8ea52bdd94934a3457"
[[package]] [[package]]
name = "arrayvec" name = "arrayvec"
version = "0.7.6" version = "0.7.6"
@@ -548,6 +554,8 @@ dependencies = [
name = "brk_vec" name = "brk_vec"
version = "0.0.19" version = "0.0.19"
dependencies = [ dependencies = [
"arc-swap",
"axum",
"memmap2", "memmap2",
"rayon", "rayon",
"serde", "serde",
Cargo.toml
+1
View File
@@ -16,6 +16,7 @@ panic = "abort"
inherits = "release" inherits = "release"
[workspace.dependencies] [workspace.dependencies]
axum = "0.8.3"
bitcoin = { version = "0.32.5", features = ["serde"] } bitcoin = { version = "0.32.5", features = ["serde"] }
bitcoincore-rpc = "0.19.0" bitcoincore-rpc = "0.19.0"
brk_cli = { version = "0", path = "crates/brk_cli" } brk_cli = { version = "0", path = "crates/brk_cli" }
crates/brk_computer/src/storage/vecs/base.rs
+20 -20
View File
@@ -10,14 +10,14 @@ use std::{
use brk_core::CheckedSub; use brk_core::CheckedSub;
use brk_exit::Exit; use brk_exit::Exit;
use brk_vec::{ use brk_vec::{
AnyStorableVec, Compressed, Error, MAX_CACHE_SIZE, Result, StorableVec, StoredIndex, AnyStoredVec, Compressed, Error, MAX_CACHE_SIZE, Result, StoredIndex, StoredType, StoredVec,
StoredType, Version, Version,
}; };
#[derive(Debug)] #[derive(Debug)]
pub struct ComputedVec<I, T> { pub struct ComputedVec<I, T> {
computed_version: Option<Version>, computed_version: Option<Version>,
vec: StorableVec<I, T>, vec: StoredVec<I, T>,
} }
impl<I, T> ComputedVec<I, T> impl<I, T> ComputedVec<I, T>
@@ -32,7 +32,7 @@ where
version: Version, version: Version,
compressed: Compressed, compressed: Compressed,
) -> brk_vec::Result<Self> { ) -> brk_vec::Result<Self> {
let vec = StorableVec::forced_import(path, version, compressed)?; let vec = StoredVec::forced_import(path, version, compressed)?;
Ok(Self { Ok(Self {
computed_version: None, computed_version: None,
@@ -89,19 +89,19 @@ where
self.vec.len() self.vec.len()
} }
pub fn vec(&self) -> &StorableVec<I, T> { pub fn vec(&self) -> &StoredVec<I, T> {
&self.vec &self.vec
} }
pub fn mut_vec(&mut self) -> &mut StorableVec<I, T> { pub fn mut_vec(&mut self) -> &mut StoredVec<I, T> {
&mut self.vec &mut self.vec
} }
pub fn any_vec(&self) -> &dyn AnyStorableVec { pub fn any_vec(&self) -> &dyn AnyStoredVec {
&self.vec &self.vec
} }
pub fn mut_any_vec(&mut self) -> &mut dyn AnyStorableVec { pub fn mut_any_vec(&mut self) -> &mut dyn AnyStoredVec {
&mut self.vec &mut self.vec
} }
@@ -130,14 +130,14 @@ where
pub fn compute_transform<A, B, F>( pub fn compute_transform<A, B, F>(
&mut self, &mut self,
max_from: A, max_from: A,
other: &mut StorableVec<A, B>, other: &mut StoredVec<A, B>,
mut t: F, mut t: F,
exit: &Exit, exit: &Exit,
) -> Result<()> ) -> Result<()>
where where
A: StoredIndex, A: StoredIndex,
B: StoredType, B: StoredType,
F: FnMut((A, B, &mut Self, &mut StorableVec<A, B>)) -> (I, T), F: FnMut((A, B, &mut Self, &mut StoredVec<A, B>)) -> (I, T),
{ {
self.validate_computed_version_or_reset_file( self.validate_computed_version_or_reset_file(
Version::ZERO + self.version() + other.version(), Version::ZERO + self.version() + other.version(),
@@ -155,7 +155,7 @@ where
pub fn compute_inverse_more_to_less( pub fn compute_inverse_more_to_less(
&mut self, &mut self,
max_from: T, max_from: T,
other: &mut StorableVec<T, I>, other: &mut StoredVec<T, I>,
exit: &Exit, exit: &Exit,
) -> Result<()> ) -> Result<()>
where where
@@ -182,8 +182,8 @@ where
pub fn compute_inverse_less_to_more( pub fn compute_inverse_less_to_more(
&mut self, &mut self,
max_from: T, max_from: T,
first_indexes: &mut StorableVec<T, I>, first_indexes: &mut StoredVec<T, I>,
last_indexes: &mut StorableVec<T, I>, last_indexes: &mut StoredVec<T, I>,
exit: &Exit, exit: &Exit,
) -> Result<()> ) -> Result<()>
where where
@@ -208,7 +208,7 @@ where
pub fn compute_last_index_from_first( pub fn compute_last_index_from_first(
&mut self, &mut self,
max_from: I, max_from: I,
first_indexes: &mut StorableVec<I, T>, first_indexes: &mut StoredVec<I, T>,
final_len: usize, final_len: usize,
exit: &Exit, exit: &Exit,
) -> Result<()> ) -> Result<()>
@@ -243,8 +243,8 @@ where
pub fn compute_count_from_indexes<T2>( pub fn compute_count_from_indexes<T2>(
&mut self, &mut self,
max_from: I, max_from: I,
first_indexes: &mut StorableVec<I, T2>, first_indexes: &mut StoredVec<I, T2>,
last_indexes: &mut StorableVec<I, T2>, last_indexes: &mut StoredVec<I, T2>,
exit: &Exit, exit: &Exit,
) -> Result<()> ) -> Result<()>
where where
@@ -271,8 +271,8 @@ where
pub fn compute_is_first_ordered<A>( pub fn compute_is_first_ordered<A>(
&mut self, &mut self,
max_from: I, max_from: I,
self_to_other: &mut StorableVec<I, A>, self_to_other: &mut StoredVec<I, A>,
other_to_self: &mut StorableVec<A, I>, other_to_self: &mut StoredVec<A, I>,
exit: &Exit, exit: &Exit,
) -> Result<()> ) -> Result<()>
where where
@@ -295,8 +295,8 @@ where
pub fn compute_sum_from_indexes<T2>( pub fn compute_sum_from_indexes<T2>(
&mut self, &mut self,
max_from: I, max_from: I,
first_indexes: &mut StorableVec<I, T2>, first_indexes: &mut StoredVec<I, T2>,
last_indexes: &mut StorableVec<I, T2>, last_indexes: &mut StoredVec<I, T2>,
exit: &Exit, exit: &Exit,
) -> Result<()> ) -> Result<()>
where where
crates/brk_computer/src/storage/vecs/blocks.rs
+2 -2
View File
@@ -4,7 +4,7 @@ use brk_core::{CheckedSub, StoredU32, StoredU64, StoredUsize, Timestamp, Weight}
use brk_exit::Exit; use brk_exit::Exit;
use brk_indexer::Indexer; use brk_indexer::Indexer;
use brk_parser::bitcoin; use brk_parser::bitcoin;
use brk_vec::{AnyStorableVec, Compressed, Version}; use brk_vec::{AnyStoredVec, Compressed, Version};
use super::{ use super::{
Indexes, Indexes,
@@ -156,7 +156,7 @@ impl Vecs {
Ok(()) Ok(())
} }
pub fn as_any_vecs(&self) -> Vec<&dyn AnyStorableVec> { pub fn as_any_vecs(&self) -> Vec<&dyn AnyStoredVec> {
[ [
self.indexes_to_block_interval.any_vecs(), self.indexes_to_block_interval.any_vecs(),
self.indexes_to_block_count.any_vecs(), self.indexes_to_block_count.any_vecs(),
crates/brk_computer/src/storage/vecs/grouped/builder.rs
+9 -14
View File
@@ -1,7 +1,7 @@
use std::path::Path; use std::path::Path;
use brk_exit::Exit; use brk_exit::Exit;
use brk_vec::{AnyStorableVec, Compressed, Result, StorableVec, StoredIndex, StoredType, Version}; use brk_vec::{AnyStoredVec, Compressed, Result, StoredIndex, StoredType, StoredVec, Version};
use crate::storage::vecs::base::ComputedVec; use crate::storage::vecs::base::ComputedVec;
@@ -114,12 +114,7 @@ where
Ok(s) Ok(s)
} }
pub fn extend( pub fn extend(&mut self, max_from: I, source: &mut StoredVec<I, T>, exit: &Exit) -> Result<()> {
&mut self,
max_from: I,
source: &mut StorableVec<I, T>,
exit: &Exit,
) -> Result<()> {
if self.total.is_none() { if self.total.is_none() {
return Ok(()); return Ok(());
}; };
@@ -154,9 +149,9 @@ where
pub fn compute<I2>( pub fn compute<I2>(
&mut self, &mut self,
max_from: I, max_from: I,
source: &mut StorableVec<I2, T>, source: &mut StoredVec<I2, T>,
first_indexes: &mut StorableVec<I, I2>, first_indexes: &mut StoredVec<I, I2>,
last_indexes: &mut StorableVec<I, I2>, last_indexes: &mut StoredVec<I, I2>,
exit: &Exit, exit: &Exit,
) -> Result<()> ) -> Result<()>
where where
@@ -276,8 +271,8 @@ where
&mut self, &mut self,
max_from: I, max_from: I,
source: &mut ComputedVecBuilder<I2, T>, source: &mut ComputedVecBuilder<I2, T>,
first_indexes: &mut StorableVec<I, I2>, first_indexes: &mut StoredVec<I, I2>,
last_indexes: &mut StorableVec<I, I2>, last_indexes: &mut StoredVec<I, I2>,
exit: &Exit, exit: &Exit,
) -> Result<()> ) -> Result<()>
where where
@@ -434,8 +429,8 @@ where
)) ))
} }
pub fn any_vecs(&self) -> Vec<&dyn AnyStorableVec> { pub fn any_vecs(&self) -> Vec<&dyn AnyStoredVec> {
let mut v: Vec<&dyn AnyStorableVec> = vec![]; let mut v: Vec<&dyn AnyStoredVec> = vec![];
if let Some(first) = self.first.as_ref() { if let Some(first) = self.first.as_ref() {
v.push(first.any_vec()); v.push(first.any_vec());
crates/brk_computer/src/storage/vecs/grouped/from_dateindex.rs
+2 -2
View File
@@ -3,7 +3,7 @@ use std::path::Path;
use brk_core::{Dateindex, Decadeindex, Monthindex, Quarterindex, Weekindex, Yearindex}; use brk_core::{Dateindex, Decadeindex, Monthindex, Quarterindex, Weekindex, Yearindex};
use brk_exit::Exit; use brk_exit::Exit;
use brk_indexer::Indexer; use brk_indexer::Indexer;
use brk_vec::{AnyStorableVec, Compressed, Result, Version}; use brk_vec::{AnyStoredVec, Compressed, Result, Version};
use crate::storage::vecs::{Indexes, base::ComputedVec, indexes}; use crate::storage::vecs::{Indexes, base::ComputedVec, indexes};
@@ -126,7 +126,7 @@ where
Ok(()) Ok(())
} }
pub fn any_vecs(&self) -> Vec<&dyn AnyStorableVec> { pub fn any_vecs(&self) -> Vec<&dyn AnyStoredVec> {
[ [
vec![self.dateindex.any_vec()], vec![self.dateindex.any_vec()],
self.dateindex_extra.any_vecs(), self.dateindex_extra.any_vecs(),
crates/brk_computer/src/storage/vecs/grouped/from_height.rs
+3 -3
View File
@@ -5,7 +5,7 @@ use brk_core::{
}; };
use brk_exit::Exit; use brk_exit::Exit;
use brk_indexer::Indexer; use brk_indexer::Indexer;
use brk_vec::{AnyStorableVec, Compressed, Result, StorableVec, Version}; use brk_vec::{AnyStoredVec, Compressed, Result, StoredVec, Version};
use crate::storage::vecs::{Indexes, base::ComputedVec, indexes}; use crate::storage::vecs::{Indexes, base::ComputedVec, indexes};
@@ -102,7 +102,7 @@ where
indexes: &mut indexes::Vecs, indexes: &mut indexes::Vecs,
starting_indexes: &Indexes, starting_indexes: &Indexes,
exit: &Exit, exit: &Exit,
height: Option<&mut StorableVec<Height, T>>, height: Option<&mut StoredVec<Height, T>>,
) -> color_eyre::Result<()> { ) -> color_eyre::Result<()> {
let height = height.unwrap_or_else(|| self.height.as_mut().unwrap().mut_vec()); let height = height.unwrap_or_else(|| self.height.as_mut().unwrap().mut_vec());
@@ -168,7 +168,7 @@ where
Ok(()) Ok(())
} }
pub fn any_vecs(&self) -> Vec<&dyn AnyStorableVec> { pub fn any_vecs(&self) -> Vec<&dyn AnyStoredVec> {
[ [
self.height.as_ref().map_or(vec![], |v| vec![v.any_vec()]), self.height.as_ref().map_or(vec![], |v| vec![v.any_vec()]),
self.height_extra.any_vecs(), self.height_extra.any_vecs(),
crates/brk_computer/src/storage/vecs/grouped/from_height_strict.rs
+2 -2
View File
@@ -3,7 +3,7 @@ use std::path::Path;
use brk_core::{Difficultyepoch, Height}; use brk_core::{Difficultyepoch, Height};
use brk_exit::Exit; use brk_exit::Exit;
use brk_indexer::Indexer; use brk_indexer::Indexer;
use brk_vec::{AnyStorableVec, Compressed, Result, Version}; use brk_vec::{AnyStoredVec, Compressed, Result, Version};
use crate::storage::vecs::{Indexes, base::ComputedVec, indexes}; use crate::storage::vecs::{Indexes, base::ComputedVec, indexes};
@@ -84,7 +84,7 @@ where
Ok(()) Ok(())
} }
pub fn any_vecs(&self) -> Vec<&dyn AnyStorableVec> { pub fn any_vecs(&self) -> Vec<&dyn AnyStoredVec> {
[ [
vec![self.height.any_vec()], vec![self.height.any_vec()],
self.height_extra.any_vecs(), self.height_extra.any_vecs(),
crates/brk_computer/src/storage/vecs/grouped/from_txindex.rs
+3 -3
View File
@@ -6,7 +6,7 @@ use brk_core::{
}; };
use brk_exit::Exit; use brk_exit::Exit;
use brk_indexer::Indexer; use brk_indexer::Indexer;
use brk_vec::{AnyStorableVec, Compressed, Result, StorableVec, Version}; use brk_vec::{AnyStoredVec, Compressed, Result, StoredVec, Version};
use crate::storage::vecs::{Indexes, base::ComputedVec, indexes}; use crate::storage::vecs::{Indexes, base::ComputedVec, indexes};
@@ -115,7 +115,7 @@ where
indexes: &mut indexes::Vecs, indexes: &mut indexes::Vecs,
starting_indexes: &Indexes, starting_indexes: &Indexes,
exit: &Exit, exit: &Exit,
txindex: Option<&mut StorableVec<Txindex, T>>, txindex: Option<&mut StoredVec<Txindex, T>>,
) -> color_eyre::Result<()> { ) -> color_eyre::Result<()> {
let txindex = txindex.unwrap_or_else(|| self.txindex.as_mut().unwrap().mut_vec()); let txindex = txindex.unwrap_or_else(|| self.txindex.as_mut().unwrap().mut_vec());
@@ -189,7 +189,7 @@ where
Ok(()) Ok(())
} }
pub fn any_vecs(&self) -> Vec<&dyn AnyStorableVec> { pub fn any_vecs(&self) -> Vec<&dyn AnyStoredVec> {
[ [
self.txindex.as_ref().map_or(vec![], |v| vec![v.any_vec()]), self.txindex.as_ref().map_or(vec![], |v| vec![v.any_vec()]),
self.txindex_extra.any_vecs(), self.txindex_extra.any_vecs(),
crates/brk_computer/src/storage/vecs/indexes.rs
+2 -2
View File
@@ -6,7 +6,7 @@ use brk_core::{
}; };
use brk_exit::Exit; use brk_exit::Exit;
use brk_indexer::Indexer; use brk_indexer::Indexer;
use brk_vec::{AnyStorableVec, Compressed, Version}; use brk_vec::{AnyStoredVec, Compressed, Version};
use super::ComputedVec; use super::ComputedVec;
@@ -784,7 +784,7 @@ impl Vecs {
}) })
} }
pub fn as_any_vecs(&self) -> Vec<&dyn AnyStorableVec> { pub fn as_any_vecs(&self) -> Vec<&dyn AnyStoredVec> {
vec![ vec![
self.dateindex_to_date.any_vec(), self.dateindex_to_date.any_vec(),
self.dateindex_to_dateindex.any_vec(), self.dateindex_to_dateindex.any_vec(),
crates/brk_computer/src/storage/vecs/marketprice.rs
+2 -2
View File
@@ -7,7 +7,7 @@ use brk_core::{
use brk_exit::Exit; use brk_exit::Exit;
use brk_fetcher::Fetcher; use brk_fetcher::Fetcher;
use brk_indexer::Indexer; use brk_indexer::Indexer;
use brk_vec::{AnyStorableVec, Compressed, Version}; use brk_vec::{AnyStoredVec, Compressed, Version};
use super::{ use super::{
ComputedVec, Indexes, ComputedVec, Indexes,
@@ -765,7 +765,7 @@ impl Vecs {
Ok(()) Ok(())
} }
pub fn as_any_vecs(&self) -> Vec<&dyn AnyStorableVec> { pub fn as_any_vecs(&self) -> Vec<&dyn AnyStoredVec> {
vec![ vec![
vec![ vec![
self.dateindex_to_close_in_cents.any_vec(), self.dateindex_to_close_in_cents.any_vec(),
crates/brk_computer/src/storage/vecs/mod.rs
+2 -2
View File
@@ -3,7 +3,7 @@ use std::{fs, path::Path};
use brk_exit::Exit; use brk_exit::Exit;
use brk_fetcher::Fetcher; use brk_fetcher::Fetcher;
use brk_indexer::Indexer; use brk_indexer::Indexer;
use brk_vec::{AnyStorableVec, Compressed}; use brk_vec::{AnyStoredVec, Compressed};
mod base; mod base;
mod blocks; mod blocks;
@@ -63,7 +63,7 @@ impl Vecs {
Ok(()) Ok(())
} }
pub fn as_any_vecs(&self) -> Vec<&dyn AnyStorableVec> { pub fn as_any_vecs(&self) -> Vec<&dyn AnyStoredVec> {
[ [
self.indexes.as_any_vecs(), self.indexes.as_any_vecs(),
self.blocks.as_any_vecs(), self.blocks.as_any_vecs(),
crates/brk_computer/src/storage/vecs/transactions.rs
+2 -2
View File
@@ -3,7 +3,7 @@ use std::{fs, path::Path};
use brk_core::{Sats, StoredU64, Txindex, Txinindex, Txoutindex}; use brk_core::{Sats, StoredU64, Txindex, Txinindex, Txoutindex};
use brk_exit::Exit; use brk_exit::Exit;
use brk_indexer::Indexer; use brk_indexer::Indexer;
use brk_vec::{AnyStorableVec, Compressed, Version}; use brk_vec::{AnyStoredVec, Compressed, Version};
use super::{ use super::{
ComputedVec, Indexes, ComputedVec, Indexes,
@@ -203,7 +203,7 @@ impl Vecs {
Ok(()) Ok(())
} }
pub fn as_any_vecs(&self) -> Vec<&dyn AnyStorableVec> { pub fn as_any_vecs(&self) -> Vec<&dyn AnyStoredVec> {
[ [
vec![ vec![
self.txindex_to_is_coinbase.any_vec(), self.txindex_to_is_coinbase.any_vec(),
crates/brk_indexer/src/vecs/base.rs
+11 -13
View File
@@ -6,8 +6,8 @@ use std::{
}; };
use brk_vec::{ use brk_vec::{
AnyStorableVec, Compressed, Error, MAX_CACHE_SIZE, MAX_PAGE_SIZE, Result, StorableVec, AnyVec, Compressed, Error, MAX_CACHE_SIZE, MAX_PAGE_SIZE, Result, StoredIndex, StoredType,
StoredIndex, StoredType, Value, Version, StoredVec, Value, Version,
}; };
use super::Height; use super::Height;
@@ -15,7 +15,7 @@ use super::Height;
#[derive(Debug)] #[derive(Debug)]
pub struct IndexedVec<I, T> { pub struct IndexedVec<I, T> {
height: Option<Height>, height: Option<Height>,
vec: StorableVec<I, T>, vec: StoredVec<I, T>,
} }
impl<I, T> IndexedVec<I, T> impl<I, T> IndexedVec<I, T>
@@ -33,9 +33,9 @@ where
version: Version, version: Version,
compressed: Compressed, compressed: Compressed,
) -> brk_vec::Result<Self> { ) -> brk_vec::Result<Self> {
let mut vec = StorableVec::forced_import(path, version, compressed)?; let mut vec = StoredVec::forced_import(path, version, compressed)?;
vec.enable_large_cache(); vec.enable_large_cache_if_possible();
Ok(Self { Ok(Self {
height: Height::try_from(Self::path_height_(path).as_path()).ok(), height: Height::try_from(Self::path_height_(path).as_path()).ok(),
@@ -67,15 +67,13 @@ where
let min_page_index = (max_page_index + 1) - large_cache_len; let min_page_index = (max_page_index + 1) - large_cache_len;
if page_index >= min_page_index { if page_index >= min_page_index {
let values = self self.vec
.vec
.pages() .pages()
.unwrap() .unwrap()
.get(page_index - min_page_index) .get(page_index - min_page_index)
.ok_or(Error::MmapsVecIsTooSmall)? .ok_or(Error::MmapsVecIsTooSmall)?
.get_or_init(|| self.vec.decode_page(page_index).unwrap()); .get_or_init(|| self.vec.decode_page(page_index).unwrap())
.get(index)
return Ok(values.get(index)?.map(|v| Value::Ref(v)));
} }
} }
@@ -126,15 +124,15 @@ where
self.vec.flush() self.vec.flush()
} }
pub fn vec(&self) -> &StorableVec<I, T> { pub fn vec(&self) -> &StoredVec<I, T> {
&self.vec &self.vec
} }
pub fn mut_vec(&mut self) -> &mut StorableVec<I, T> { pub fn mut_vec(&mut self) -> &mut StoredVec<I, T> {
&mut self.vec &mut self.vec
} }
pub fn any_vec(&self) -> &dyn AnyStorableVec { pub fn any_vec(&self) -> &dyn AnyVec {
&self.vec &self.vec
} }
crates/brk_indexer/src/vecs/mod.rs
+2 -2
View File
@@ -7,7 +7,7 @@ use brk_core::{
P2TRindex, P2WPKHAddressBytes, P2WPKHindex, P2WSHAddressBytes, P2WSHindex, Pushonlyindex, Sats, P2TRindex, P2WPKHAddressBytes, P2WPKHindex, P2WSHAddressBytes, P2WSHindex, Pushonlyindex, Sats,
StoredUsize, Timestamp, TxVersion, Txid, Txindex, Txinindex, Txoutindex, Unknownindex, Weight, StoredUsize, Timestamp, TxVersion, Txid, Txindex, Txinindex, Txoutindex, Unknownindex, Weight,
}; };
use brk_vec::{AnyStorableVec, Compressed, Version}; use brk_vec::{AnyVec, Compressed, Version};
use rayon::prelude::*; use rayon::prelude::*;
use crate::Indexes; use crate::Indexes;
@@ -609,7 +609,7 @@ impl Vecs {
.unwrap() .unwrap()
} }
pub fn as_any_vecs(&self) -> Vec<&dyn AnyStorableVec> { pub fn as_any_vecs(&self) -> Vec<&dyn AnyVec> {
vec![ vec![
self.addressindex_to_addresstype.any_vec(), self.addressindex_to_addresstype.any_vec(),
self.addressindex_to_addresstypeindex.any_vec(), self.addressindex_to_addresstypeindex.any_vec(),
crates/brk_query/src/lib.rs
+3 -3
View File
@@ -5,7 +5,7 @@
use brk_computer::Computer; use brk_computer::Computer;
use brk_indexer::Indexer; use brk_indexer::Indexer;
use brk_vec::AnyStorableVec; use brk_vec::AnyStoredVec;
use tabled::settings::Style; use tabled::settings::Style;
mod format; mod format;
@@ -51,7 +51,7 @@ impl<'a> Query<'a> {
} }
} }
pub fn search(&self, index: Index, ids: &[&str]) -> Vec<(String, &&dyn AnyStorableVec)> { pub fn search(&self, index: Index, ids: &[&str]) -> Vec<(String, &&dyn AnyStoredVec)> {
let tuples = ids let tuples = ids
.iter() .iter()
.flat_map(|s| { .flat_map(|s| {
@@ -86,7 +86,7 @@ impl<'a> Query<'a> {
pub fn format( pub fn format(
&self, &self,
vecs: Vec<(String, &&dyn AnyStorableVec)>, vecs: Vec<(String, &&dyn AnyStoredVec)>,
from: Option<i64>, from: Option<i64>,
to: Option<i64>, to: Option<i64>,
format: Option<Format>, format: Option<Format>,
crates/brk_query/src/vec_trees.rs
+4 -4
View File
@@ -1,6 +1,6 @@
use std::collections::BTreeMap; use std::collections::BTreeMap;
use brk_vec::AnyStorableVec; use brk_vec::AnyStoredVec;
use derive_deref::{Deref, DerefMut}; use derive_deref::{Deref, DerefMut};
use super::index::Index; use super::index::Index;
@@ -13,7 +13,7 @@ pub struct VecTrees<'a> {
impl<'a> VecTrees<'a> { impl<'a> VecTrees<'a> {
// Not the most performant or type safe but only built once so that's okay // Not the most performant or type safe but only built once so that's okay
pub fn insert(&mut self, vec: &'a dyn AnyStorableVec) { pub fn insert(&mut self, vec: &'a dyn AnyStoredVec) {
let file_name = vec.file_name(); let file_name = vec.file_name();
let split = file_name.split("_to_").collect::<Vec<_>>(); let split = file_name.split("_to_").collect::<Vec<_>>();
if split.len() != 2 { if split.len() != 2 {
@@ -88,7 +88,7 @@ impl<'a> VecTrees<'a> {
} }
#[derive(Default, Deref, DerefMut)] #[derive(Default, Deref, DerefMut)]
pub struct IndexToVec<'a>(BTreeMap<Index, &'a dyn AnyStorableVec>); pub struct IndexToVec<'a>(BTreeMap<Index, &'a dyn AnyStoredVec>);
#[derive(Default, Deref, DerefMut)] #[derive(Default, Deref, DerefMut)]
pub struct IdToVec<'a>(BTreeMap<String, &'a dyn AnyStorableVec>); pub struct IdToVec<'a>(BTreeMap<String, &'a dyn AnyStoredVec>);
crates/brk_server/Cargo.toml
+1 -1
View File
@@ -7,7 +7,7 @@ license.workspace = true
repository.workspace = true repository.workspace = true
[dependencies] [dependencies]
axum = "0.8.3" axum = { workspace = true }
brk_computer = { workspace = true } brk_computer = { workspace = true }
brk_exit = { workspace = true } brk_exit = { workspace = true }
brk_fetcher = { workspace = true } brk_fetcher = { workspace = true }
crates/brk_vec/Cargo.toml
+2
View File
@@ -9,6 +9,8 @@ license.workspace = true
repository.workspace = true repository.workspace = true
[dependencies] [dependencies]
axum = { workspace = true }
arc-swap = "1.7.1"
memmap2 = "0.9.5" memmap2 = "0.9.5"
rayon = { workspace = true } rayon = { workspace = true }
serde = { workspace = true } serde = { workspace = true }
crates/brk_vec/examples/main.rs
+7 -10
View File
@@ -1,13 +1,12 @@
use std::{fs, path::Path}; use std::{fs, path::Path};
use brk_vec::{Compressed, StorableVec, Version}; use brk_vec::{AnyVec, RawVec, Version};
fn main() -> Result<(), Box<dyn std::error::Error>> { fn main() -> Result<(), Box<dyn std::error::Error>> {
let _ = fs::remove_dir_all("./vec"); let _ = fs::remove_dir_all("./vec");
{ {
let mut vec: StorableVec<usize, u32> = let mut vec: RawVec<usize, u32> = RawVec::forced_import(Path::new("./vec"), Version::ZERO)?;
StorableVec::forced_import(Path::new("./vec"), Version::ZERO, Compressed::YES)?;
(0..21_u32).for_each(|v| { (0..21_u32).for_each(|v| {
vec.push(v); vec.push(v);
@@ -20,8 +19,7 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
} }
{ {
let mut vec: StorableVec<usize, u32> = let mut vec: RawVec<usize, u32> = RawVec::forced_import(Path::new("./vec"), Version::ZERO)?;
StorableVec::forced_import(Path::new("./vec"), Version::ZERO, Compressed::YES)?;
dbg!(vec.get(0)?); dbg!(vec.get(0)?);
dbg!(vec.get(0)?); dbg!(vec.get(0)?);
@@ -42,10 +40,9 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
} }
{ {
let mut vec: StorableVec<usize, u32> = let mut vec: RawVec<usize, u32> = RawVec::forced_import(Path::new("./vec"), Version::ZERO)?;
StorableVec::forced_import(Path::new("./vec"), Version::ZERO, Compressed::YES)?;
vec.enable_large_cache(); // vec.enable_large_cache_if_possible();
dbg!(vec.get(0)?); dbg!(vec.get(0)?);
dbg!(vec.get(20)?); dbg!(vec.get(20)?);
@@ -58,12 +55,12 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
dbg!(vec.get(5)?); dbg!(vec.get(5)?);
dbg!(vec.get(20)?); dbg!(vec.get(20)?);
vec.iter(|(_, v)| { vec.iter(|(_, v, ..)| {
dbg!(v); dbg!(v);
Ok(()) Ok(())
})?; })?;
vec.iter_from(5, |(_, v)| { vec.iter_from(5, |(_, v, ..)| {
dbg!(v); dbg!(v);
Ok(()) Ok(())
})?; })?;
crates/brk_vec/src/enums/error.rs
+2
View File
@@ -15,6 +15,7 @@ pub enum Error {
IO(io::Error), IO(io::Error),
ZeroCopyError, ZeroCopyError,
IndexTooHigh, IndexTooHigh,
EmptyVec,
IndexTooLow, IndexTooLow,
ExpectFileToHaveIndex, ExpectFileToHaveIndex,
ExpectVecToHaveIndex, ExpectVecToHaveIndex,
@@ -68,6 +69,7 @@ impl fmt::Display for Error {
Error::ZeroCopyError => write!(f, "Zero copy convert error"), Error::ZeroCopyError => write!(f, "Zero copy convert error"),
Error::RangeFromAfterTo(from, to) => write!(f, "Range, from {from} is after to {to}"), Error::RangeFromAfterTo(from, to) => write!(f, "Range, from {from} is after to {to}"),
Error::DifferentCompressionMode => write!(f, "Different compression mode chosen"), Error::DifferentCompressionMode => write!(f, "Different compression mode chosen"),
Error::EmptyVec => write!(f, "The Vec is empty, maybe wait for a bit"),
} }
} }
} }
crates/brk_vec/src/enums/values.rs
+4 -1
View File
@@ -3,7 +3,10 @@ use std::ops::Range;
use memmap2::Mmap; use memmap2::Mmap;
use zerocopy::{Immutable, IntoBytes, KnownLayout, TryFromBytes}; use zerocopy::{Immutable, IntoBytes, KnownLayout, TryFromBytes};
use crate::MAX_PAGE_SIZE; const ONE_KIB: usize = 1024;
pub const MAX_PAGE_SIZE: usize = 16 * ONE_KIB;
const ONE_MIB: usize = ONE_KIB * ONE_KIB;
pub const MAX_CACHE_SIZE: usize = 100 * ONE_MIB;
use super::Result; use super::Result;
crates/brk_vec/src/lib.rs
+96 -820
View File
@@ -3,863 +3,139 @@
#![doc = include_str!("../examples/main.rs")] #![doc = include_str!("../examples/main.rs")]
#![doc = "```"] #![doc = "```"]
use std::{
fs::{self, File, OpenOptions},
io::{self, Read, Seek, SeekFrom, Write},
marker::PhantomData,
mem,
path::{Path, PathBuf},
sync::OnceLock,
};
pub use memmap2;
use rayon::prelude::*;
pub use zerocopy;
use zstd::DEFAULT_COMPRESSION_LEVEL;
mod enums; mod enums;
mod structs; mod structs;
mod traits; mod traits;
mod variants;
use std::{path::Path, sync::Arc};
use arc_swap::{ArcSwap, Guard};
use axum::Json;
pub use enums::*; pub use enums::*;
use memmap2::Mmap;
pub use structs::*; pub use structs::*;
pub use traits::*; pub use traits::*;
pub use variants::*;
const ONE_KIB: usize = 1024;
pub const MAX_PAGE_SIZE: usize = 16 * ONE_KIB;
const ONE_MIB: usize = ONE_KIB * ONE_KIB;
pub const MAX_CACHE_SIZE: usize = 100 * ONE_MIB;
#[allow(private_interfaces)]
#[derive(Debug)] #[derive(Debug)]
pub enum StorableVec<I, T> { pub enum StoredVec<I, T> {
Raw { Raw(RawVec<I, T>),
base: Base<I, T>, Compressed(CompressedVec<I, T>),
},
Compressed {
base: Base<I, T>,
pages_meta: CompressedPagesMetadata,
},
} }
impl<I, T> StorableVec<I, T> impl<I, T> StoredVec<I, T>
where where
I: StoredIndex, I: StoredIndex,
T: StoredType, T: StoredType,
{ {
pub const SIZE_OF_T: usize = size_of::<T>();
pub const PER_PAGE: usize = MAX_PAGE_SIZE / Self::SIZE_OF_T;
pub const PAGE_SIZE: usize = Self::PER_PAGE * Self::SIZE_OF_T;
pub const CACHE_LENGTH: usize = MAX_CACHE_SIZE / Self::PAGE_SIZE;
/// Same as import but will reset the folder under certain errors, so be careful !
pub fn forced_import(path: &Path, version: Version, compressed: Compressed) -> Result<Self> { pub fn forced_import(path: &Path, version: Version, compressed: Compressed) -> Result<Self> {
let res = Self::import(path, version, compressed);
match res {
Err(Error::WrongEndian)
| Err(Error::DifferentCompressionMode)
| Err(Error::DifferentVersion {
found: _,
expected: _,
}) => {
fs::remove_dir_all(path)?;
Self::import(path, version, compressed)
}
_ => res,
}
}
pub fn import(path: &Path, version: Version, compressed: Compressed) -> Result<Self> {
let base = Base::import(path, version, compressed)?;
if *compressed { if *compressed {
let pages_meta = Self::read_pages_meta_(path)?; Ok(Self::Compressed(CompressedVec::forced_import(
path, version,
Ok(Self::Compressed { base, pages_meta }) )?))
} else { } else {
Ok(Self::Raw { base }) Ok(Self::Raw(RawVec::forced_import(path, version)?))
} }
} }
}
fn read_pages_meta(&self) -> Result<CompressedPagesMetadata> { impl<I, T> AnyVec<I, T> for StoredVec<I, T>
Self::read_pages_meta_(self.path()) where
} I: StoredIndex,
fn read_pages_meta_(path: &Path) -> Result<CompressedPagesMetadata> { T: StoredType,
CompressedPagesMetadata::read(Self::path_pages_meta_(path).as_path()) {
}
#[inline] #[inline]
pub fn get(&mut self, index: I) -> Result<Option<&T>> { fn get_(&mut self, index: usize) -> Result<Option<Value<T>>> {
self.get_(index.to_usize()?) match self {
} StoredVec::Raw(v) => v.get_(index),
#[inline] StoredVec::Compressed(v) => v.get_(index),
pub fn get_(&mut self, index: usize) -> Result<Option<&T>> {
match self.index_to_pushed_index(index) {
Ok(index) => {
if let Some(index) = index {
return Ok(self.pushed().get(index));
}
}
Err(Error::IndexTooHigh) => return Ok(None),
Err(Error::IndexTooLow) => {}
Err(error) => return Err(error),
} }
let page_index = Self::index_to_page_index(index);
if self.page().is_none_or(|b| b.0 != page_index) {
let values = self.decode_page(page_index)?;
self.mut_page().replace((page_index, values));
}
self.page().unwrap().1.get(index)
} }
pub fn get_last(&mut self) -> Result<Option<&T>> { fn iter_from<F>(&mut self, index: I, mut f: F) -> Result<()>
let len = self.len();
if len == 0 {
return Ok(None);
}
self.get_(len - 1)
}
pub fn read(&self, index: I) -> Result<Option<T>> {
self.read_(index.to_usize()?)
}
pub fn read_(&self, index: usize) -> Result<Option<T>> {
Ok(match self {
Self::Raw { .. } => {
let mut file = self.open_file()?;
let byte_index = Self::index_to_byte_index(index);
file.seek(SeekFrom::Start(byte_index))?;
let mut buf = vec![0; Self::SIZE_OF_T];
file.read_exact(&mut buf)?;
T::try_ref_from_bytes(&buf[..]).ok().map(|v| v.to_owned())
}
Self::Compressed { .. } => self
.decode_page(Self::index_to_page_index(index))?
.get(index)?
.cloned(),
})
}
pub fn iter<F>(&mut self, f: F) -> Result<()>
where
F: FnMut((I, &T)) -> Result<()>,
{
self.iter_from(I::default(), f)
}
pub fn iter_from<F>(&mut self, mut index: I, mut f: F) -> Result<()>
where
F: FnMut((I, &T)) -> Result<()>,
{
if !self.is_pushed_empty() {
return Err(Error::UnsupportedUnflushedState);
}
let stored_len = I::from(self.stored_len());
while index < stored_len {
let v = self.get(index)?.unwrap();
f((index, v))?;
index = index + 1;
}
Ok(())
}
pub fn iter_from_cloned<F>(&mut self, mut index: I, mut f: F) -> Result<()>
where where
F: FnMut((I, T, &mut Self)) -> Result<()>, F: FnMut((I, T, &mut Self)) -> Result<()>,
{ {
if !self.is_pushed_empty() { todo!();
return Err(Error::UnsupportedUnflushedState); // match self {
} // StoredVec::Raw(v) => v.iter_from(index, |(i, t, inner)| f((i, t, self))),
// StoredVec::Compressed(v) => v.iter_from(index, |(i, t, inner)| f((i, t, self))),
let stored_len = I::from(self.stored_len()); // }
while index < stored_len {
let v = self.get(index)?.unwrap().clone();
f((index, v, self))?;
index = index + 1;
}
Ok(())
} }
pub fn collect_range(&self, from: Option<i64>, to: Option<i64>) -> Result<Vec<T>> { fn collect_range(&self, from: Option<i64>, to: Option<i64>) -> Result<Json<Vec<T>>> {
if !self.is_pushed_empty() {
return Err(Error::UnsupportedUnflushedState);
}
let len = self
.base()
.read_stored_length()
.unwrap()
.to_usize()
.unwrap();
if len == 0 {
return Err(Error::IndexTooHigh);
}
let from = from.map_or(0, |from| {
if from >= 0 {
from as usize
} else {
let from = len as i64 + from;
if from < 0 { 0 } else { from as usize }
}
});
let to = to.map_or(len - 1, |to| {
if to >= 0 {
to as usize
} else {
let max = len - 1;
let to = max as i64 + to;
if to > max as i64 { max } else { to as usize }
}
});
if from > to {
return Err(Error::RangeFromAfterTo(from, to));
}
let mut page: Option<(usize, Values<T>)> = None;
let values = (from..=to)
.flat_map(|index| {
let page_index = Self::index_to_page_index(index);
if page.as_ref().is_none_or(|b| b.0 != page_index) {
let pages_meta = match self {
Self::Raw { .. } => None,
Self::Compressed { .. } => Some(self.read_pages_meta().unwrap()),
};
let values = Self::decode_page_(
len,
page_index,
&self.base().open_file().unwrap(),
pages_meta.as_ref(),
)
.inspect_err(|_| {
dbg!(from, to);
})
.unwrap();
page.replace((page_index, values));
}
page.as_ref().unwrap().1.get(index).ok().flatten().cloned()
})
.collect::<Vec<_>>();
Ok(values)
}
pub fn decode_page(&self, page_index: usize) -> Result<Values<T>> {
Self::decode_page_(
self.stored_len(),
page_index,
self.file(),
match self {
Self::Raw { .. } => None,
Self::Compressed { pages_meta, .. } => Some(pages_meta),
},
)
}
fn decode_page_(
stored_len: usize,
page_index: usize,
file: &File,
compressed_pages_meta: Option<&CompressedPagesMetadata>,
) -> Result<Values<T>> {
if Self::page_index_to_index(page_index) >= stored_len {
return Err(Error::IndexTooHigh);
}
let (len, offset) = if let Some(pages_meta) = compressed_pages_meta {
if pages_meta.len() <= page_index {
return Err(Error::ExpectVecToHaveIndex);
}
let page = pages_meta.get(page_index).unwrap();
(page.bytes_len as usize, page.start)
} else {
(Self::PAGE_SIZE, Self::page_index_to_byte_index(page_index))
};
let mmap = unsafe {
memmap2::MmapOptions::new()
.len(len)
.offset(offset)
.map(file)?
};
let compressed = compressed_pages_meta.is_some();
if compressed {
let decoded = zstd::decode_all(&mmap[..]);
if decoded.is_err() {
dbg!((len, offset, page_index, &mmap[..], &mmap.len(), &decoded));
}
Ok(Values::from(
decoded?
.chunks(Self::SIZE_OF_T)
.map(|slice| T::try_read_from_bytes(slice).unwrap())
.collect::<Vec<_>>()
.into_boxed_slice(),
))
} else {
Ok(Values::from(mmap))
}
}
#[inline]
pub fn push(&mut self, value: T) {
self.mut_base().pushed.push(value)
}
pub fn flush(&mut self) -> io::Result<()> {
let pushed_len = self.pushed_len();
if pushed_len == 0 {
return Ok(());
}
let stored_len = self.stored_len();
let bytes = match self {
Self::Compressed { base, pages_meta } => {
let (starting_page_index, values) = if *base.stored_len % Self::PER_PAGE != 0 {
if pages_meta.is_empty() {
unreachable!()
}
let last_page_index = pages_meta.len() - 1;
let values = if let Some(values) = base
.pages
.as_mut()
.and_then(|big_cache| big_cache.last_mut().and_then(|lock| lock.take()))
{
values
} else if base
.page
.as_ref()
.is_some_and(|(page_index, _)| *page_index == last_page_index)
{
base.page.take().unwrap().1
} else {
Self::decode_page_(
stored_len,
last_page_index,
&base.file,
Some(pages_meta),
)
.inspect_err(|_| {
dbg!(last_page_index, &pages_meta);
})
.unwrap()
};
let file_len = pages_meta.pop().unwrap().start;
file_set_len(&mut base.file, file_len)?;
(last_page_index, values)
} else {
(pages_meta.len(), Values::default())
};
let compressed = Vec::from(values.as_arr())
.into_par_iter()
.chain(mem::take(&mut base.pushed).into_par_iter())
.chunks(Self::PER_PAGE)
.map(|chunk| (Self::compress_chunk(chunk.as_ref()), chunk.len()))
.collect::<Vec<_>>();
compressed
.iter()
.enumerate()
.for_each(|(i, (compressed_bytes, values_len))| {
let page_index = starting_page_index + i;
let start = if page_index != 0 {
let prev = pages_meta.get(page_index - 1).unwrap();
prev.start + prev.bytes_len as u64
} else {
0
};
let bytes_len = compressed_bytes.len() as u32;
let values_len = *values_len as u32;
let page = CompressedPageMetadata::new(start, bytes_len, values_len);
pages_meta.push(page_index, page);
});
pages_meta.write()?;
compressed
.into_iter()
.flat_map(|(v, _)| v)
.collect::<Vec<_>>()
}
Self::Raw { base } => {
let pushed = &mut base.pushed;
let mut bytes: Vec<u8> = vec![0; pushed.len() * Self::SIZE_OF_T];
let unsafe_bytes = UnsafeSlice::new(&mut bytes);
mem::take(pushed)
.into_par_iter()
.enumerate()
.for_each(|(i, v)| unsafe_bytes.copy_slice(i * Self::SIZE_OF_T, v.as_bytes()));
bytes
}
};
let file = self.mut_file();
file.write_all(&bytes)?;
file.sync_all()?;
self.reset_caches();
self.increase_stored_len(pushed_len);
self.write_stored_length()?;
Ok(())
}
pub fn truncate_if_needed(&mut self, index: I) -> Result<()> {
let index = index.to_usize()?;
if index >= self.stored_len() {
return Ok(());
}
if index == 0 {
self.reset()?;
return Ok(());
}
let page_index = Self::index_to_page_index(index);
let values = match self {
Self::Compressed { .. } => self.decode_page(page_index)?,
Self::Raw { .. } => Values::default(),
};
let (len, bytes) = match self {
Self::Compressed { pages_meta, .. } => {
let mut page = pages_meta.truncate(page_index).unwrap();
let len = page.start;
let decoded_index = Self::index_to_decoded_index(index);
let compressed = if decoded_index != 0 {
let chunk = &values.as_arr()[..decoded_index];
let compressed = Self::compress_chunk(chunk);
page.values_len = chunk.len() as u32;
page.bytes_len = compressed.len() as u32;
pages_meta.push(page_index, page);
compressed
} else {
vec![].into_boxed_slice()
};
pages_meta.write()?;
(len, compressed)
}
Self::Raw { .. } => {
// let value_at_index = self.open_then_read_(index).ok();
let len = Self::index_to_byte_index(index);
(len, vec![].into_boxed_slice())
}
};
let file = self.mut_file();
file_set_len(file, len)?;
if !bytes.is_empty() {
file.write_all(&bytes)?;
}
self.set_stored_len(index);
self.write_stored_length()?;
self.reset_caches();
Ok(())
}
fn compress_chunk(chunk: &[T]) -> Box<[u8]> {
if chunk.len() > Self::PER_PAGE {
panic!();
}
let mut bytes: Vec<u8> = vec![0; chunk.len() * Self::SIZE_OF_T];
let unsafe_bytes = UnsafeSlice::new(&mut bytes);
chunk
.into_par_iter()
.enumerate()
.for_each(|(i, v)| unsafe_bytes.copy_slice(i * Self::SIZE_OF_T, v.as_bytes()));
zstd::encode_all(bytes.as_slice(), DEFAULT_COMPRESSION_LEVEL)
.unwrap()
.into_boxed_slice()
}
pub fn enable_large_cache(&mut self) {
self.mut_pages().replace(vec![]);
self.reset_large_cache();
}
pub fn disable_large_cache(&mut self) {
self.mut_base().pages.take();
}
fn reset_large_cache(&mut self) {
let stored_len = self.stored_len();
if let Some(pages) = self.mut_pages().as_mut() {
pages.par_iter_mut().for_each(|lock| {
lock.take();
});
let len = (stored_len as f64 / Self::PER_PAGE as f64).ceil() as usize;
let len = Self::CACHE_LENGTH.min(len);
if pages.len() != len {
pages.resize_with(len, Default::default);
}
}
}
pub fn large_cache_len(&self) -> usize {
self.pages().map_or(0, |v| v.len())
}
fn reset_small_cache(&mut self) {
self.mut_base().page.take();
}
fn reset_caches(&mut self) {
self.reset_small_cache();
self.reset_large_cache();
}
pub fn reset(&mut self) -> Result<()> {
self.mut_base().reset_file()?;
self.reset_stored_len();
self.reset_caches();
Ok(())
}
#[inline]
pub fn index_to_pushed_index(&self, index: usize) -> Result<Option<usize>> {
let stored_len = self.stored_len();
if index >= stored_len {
let index = index - stored_len;
if index >= self.pushed_len() {
Err(Error::IndexTooHigh)
} else {
Ok(Some(index))
}
} else {
Err(Error::IndexTooLow)
}
}
#[inline]
fn index_to_byte_index(index: usize) -> u64 {
(index * Self::SIZE_OF_T) as u64
}
#[inline(always)]
fn index_to_page_index(index: usize) -> usize {
index / Self::PER_PAGE
}
#[inline(always)]
fn page_index_to_index(page_index: usize) -> usize {
page_index * Self::PER_PAGE
}
#[inline(always)]
fn page_index_to_byte_index(page_index: usize) -> u64 {
(page_index * Self::PAGE_SIZE) as u64
}
#[inline(always)]
fn index_to_decoded_index(index: usize) -> usize {
index % Self::PER_PAGE
}
#[inline]
fn path_pages_meta_(path: &Path) -> PathBuf {
path.join("pages_meta")
}
#[inline]
fn page(&self) -> Option<&(usize, Values<T>)> {
self.base().page.as_ref()
}
#[inline]
fn mut_page(&mut self) -> &mut Option<(usize, Values<T>)> {
&mut self.mut_base().page
}
#[inline]
pub fn pages(&self) -> Option<&Vec<OnceLock<Values<T>>>> {
self.base().pages.as_ref()
}
#[inline]
fn mut_pages(&mut self) -> &mut Option<Vec<OnceLock<Values<T>>>> {
&mut self.mut_base().pages
}
#[inline]
pub fn len(&self) -> usize {
self.stored_len() + self.pushed_len()
}
pub fn is_empty(&self) -> bool {
self.len() == 0
}
#[inline]
pub fn has(&self, index: I) -> Result<bool> {
Ok(self.has_(index.to_usize()?))
}
#[inline]
fn has_(&self, index: usize) -> bool {
index < self.len()
}
#[inline]
pub fn pushed(&self) -> &Vec<T> {
&self.base().pushed
}
#[inline]
pub fn pushed_len(&self) -> usize {
self.pushed().len()
}
#[inline]
fn is_pushed_empty(&self) -> bool {
self.pushed_len() == 0
}
#[inline]
pub fn stored_len(&self) -> usize {
*self.base().stored_len
}
#[inline]
fn set_stored_len(&mut self, len: usize) {
*self.mut_base().stored_len = len;
}
fn increase_stored_len(&mut self, len: usize) {
*self.mut_base().stored_len += len;
}
#[inline]
fn reset_stored_len(&mut self) {
self.set_stored_len(0);
}
fn write_stored_length(&self) -> io::Result<()> {
self.base().write_stored_length()
}
#[inline]
pub fn path(&self) -> &Path {
&self.base().pathbuf
}
fn file(&self) -> &File {
&self.base().file
}
fn mut_file(&mut self) -> &mut File {
&mut self.mut_base().file
}
fn open_file(&self) -> io::Result<File> {
self.base().open_file()
}
pub fn file_name(&self) -> String {
self.path()
.file_name()
.unwrap()
.to_str()
.unwrap()
.to_owned()
}
#[inline]
pub fn version(&self) -> Version {
self.base().version
}
#[inline]
fn compressed(&self) -> Compressed {
self.base().compressed
}
#[inline]
fn base(&self) -> &Base<I, T> {
match self { match self {
Self::Raw { base, .. } => base, StoredVec::Raw(v) => v.collect_range(from, to),
Self::Compressed { base, .. } => base, StoredVec::Compressed(v) => v.collect_range(from, to),
} }
} }
#[inline] fn flush(&mut self) -> Result<()> {
fn mut_base(&mut self) -> &mut Base<I, T> {
match self { match self {
Self::Raw { base, .. } => base, StoredVec::Raw(v) => v.flush(),
Self::Compressed { base, .. } => base, StoredVec::Compressed(v) => v.flush(),
} }
} }
pub fn index_type_to_string(&self) -> &str { fn truncate_if_needed(&mut self, index: I) -> Result<()> {
I::to_string() match self {
StoredVec::Raw(v) => v.truncate_if_needed(index),
StoredVec::Compressed(v) => v.truncate_if_needed(index),
}
}
#[inline]
fn mmap(&self) -> &ArcSwap<Mmap> {
match self {
StoredVec::Raw(v) => v.mmap(),
StoredVec::Compressed(v) => v.mmap(),
}
}
#[inline]
fn guard(&self) -> &Option<Guard<Arc<Mmap>>> {
match self {
StoredVec::Raw(v) => v.guard(),
StoredVec::Compressed(v) => v.guard(),
}
}
#[inline]
fn mut_guard(&mut self) -> &mut Option<Guard<Arc<Mmap>>> {
match self {
StoredVec::Raw(v) => v.mut_guard(),
StoredVec::Compressed(v) => v.mut_guard(),
}
}
#[inline]
fn pushed(&self) -> &[T] {
match self {
StoredVec::Raw(v) => v.pushed(),
StoredVec::Compressed(v) => v.pushed(),
}
}
#[inline]
fn mut_pushed(&mut self) -> &mut Vec<T> {
match self {
StoredVec::Raw(v) => v.mut_pushed(),
StoredVec::Compressed(v) => v.mut_pushed(),
}
}
#[inline]
fn path(&self) -> &Path {
match self {
StoredVec::Raw(v) => v.path(),
StoredVec::Compressed(v) => v.path(),
}
}
#[inline]
fn version(&self) -> Version {
match self {
StoredVec::Raw(v) => v.version(),
StoredVec::Compressed(v) => v.version(),
}
} }
} }
#[derive(Debug)]
struct Base<I, T> {
pub version: Version,
pub pathbuf: PathBuf,
pub stored_len: Length,
pub compressed: Compressed,
pub page: Option<(usize, Values<T>)>,
pub pages: Option<Vec<OnceLock<Values<T>>>>,
pub pushed: Vec<T>,
pub file: File,
pub phantom: PhantomData<I>,
}
impl<I, T> Base<I, T> {
pub fn import(path: &Path, version: Version, compressed: Compressed) -> Result<Self> {
fs::create_dir_all(path)?;
let version_path = Self::path_version_(path);
version.validate(version_path.as_ref())?;
version.write(version_path.as_ref())?;
let compressed_path = Self::path_compressed_(path);
compressed.validate(compressed_path.as_ref())?;
compressed.write(compressed_path.as_ref())?;
let stored_len = Length::try_from(Self::path_length_(path).as_path())?;
Ok(Self {
version,
compressed,
pathbuf: path.to_owned(),
file: Self::open_file_(Self::path_vec_(path).as_path())?,
stored_len,
page: None,
pages: None,
pushed: vec![],
phantom: PhantomData,
})
}
fn open_file(&self) -> io::Result<File> {
Self::open_file_(&self.path_vec())
}
fn open_file_(path: &Path) -> io::Result<File> {
OpenOptions::new()
.read(true)
.create(true)
.truncate(false)
.append(true)
.open(path)
}
fn reset_file(&mut self) -> Result<()> {
file_set_len(&mut self.file, 0)?;
Ok(())
}
#[inline]
fn path_vec(&self) -> PathBuf {
Self::path_vec_(&self.pathbuf)
}
#[inline]
fn path_vec_(path: &Path) -> PathBuf {
path.join("vec")
}
pub fn read_stored_length(&self) -> Result<Length> {
Length::try_from(self.path_length().as_path())
}
fn write_stored_length(&self) -> io::Result<()> {
self.stored_len.write(&self.path_length())
}
#[inline]
fn path_length(&self) -> PathBuf {
Self::path_length_(&self.pathbuf)
}
#[inline]
fn path_length_(path: &Path) -> PathBuf {
path.join("length")
}
#[inline]
fn path_version_(path: &Path) -> PathBuf {
path.join("version")
}
#[inline]
fn path_compressed_(path: &Path) -> PathBuf {
path.join("compressed")
}
}
impl<I, T> Clone for StorableVec<I, T>
where
I: StoredIndex,
T: StoredType,
{
fn clone(&self) -> Self {
Self::import(self.path(), self.version(), self.compressed()).unwrap()
}
}
fn file_set_len(file: &mut File, len: u64) -> io::Result<()> {
file.set_len(len)?;
file.seek(SeekFrom::End(0))?;
Ok(())
}
crates/brk_vec/src/lib2.rs
+878
View File
@@ -0,0 +1,878 @@
#![doc = include_str!("../README.md")]
#![doc = "\n## Example\n\n```rust"]
#![doc = include_str!("../examples/main.rs")]
#![doc = "```"]
use std::{
fs::{self, File, OpenOptions},
io::{self, Read, Seek, SeekFrom, Write},
marker::PhantomData,
mem,
path::{Path, PathBuf},
sync::{Arc, OnceLock},
};
use arc_swap::ArcSwap;
pub use memmap2;
use memmap2::Mmap;
use rayon::prelude::*;
pub use zerocopy;
use zstd::DEFAULT_COMPRESSION_LEVEL;
mod enums;
mod structs;
mod traits;
pub use enums::*;
pub use structs::*;
pub use traits::*;
const ONE_KIB: usize = 1024;
pub const MAX_PAGE_SIZE: usize = 16 * ONE_KIB;
const ONE_MIB: usize = ONE_KIB * ONE_KIB;
pub const MAX_CACHE_SIZE: usize = 100 * ONE_MIB;
#[allow(private_interfaces)]
#[derive(Debug)]
pub enum StorableVec<I, T> {
Raw {
base: Base<I, T>,
},
Compressed {
base: Base<I, T>,
decoded_page: Option<(usize, Box<[T]>)>,
decoded_pages: Option<Vec<OnceLock<Box<[T]>>>>,
// pages: Option<Vec<OnceLock<Values<T>>>>,
// page: Option<(usize, Values<T>)>,
pages_meta: CompressedPagesMetadata,
},
}
impl<I, T> StorableVec<I, T>
where
I: StoredIndex,
T: StoredType,
{
pub const SIZE_OF_T: usize = size_of::<T>();
pub const PER_PAGE: usize = MAX_PAGE_SIZE / Self::SIZE_OF_T;
pub const PAGE_SIZE: usize = Self::PER_PAGE * Self::SIZE_OF_T;
pub const CACHE_LENGTH: usize = MAX_CACHE_SIZE / Self::PAGE_SIZE;
/// Same as import but will reset the folder under certain errors, so be careful !
pub fn forced_import(path: &Path, version: Version, compressed: Compressed) -> Result<Self> {
let res = Self::import(path, version, compressed);
match res {
Err(Error::WrongEndian)
| Err(Error::DifferentCompressionMode)
| Err(Error::DifferentVersion {
found: _,
expected: _,
}) => {
fs::remove_dir_all(path)?;
Self::import(path, version, compressed)
}
_ => res,
}
}
pub fn import(path: &Path, version: Version, compressed: Compressed) -> Result<Self> {
let base = Base::import(path, version, compressed)?;
if *compressed {
let pages_meta = Self::read_pages_meta_(path)?;
Ok(Self::Compressed {
base,
page: None,
pages: None,
pages_meta,
})
} else {
Ok(Self::Raw { base })
}
}
fn read_pages_meta(&self) -> Result<CompressedPagesMetadata> {
Self::read_pages_meta_(self.path())
}
fn read_pages_meta_(path: &Path) -> Result<CompressedPagesMetadata> {
CompressedPagesMetadata::read(Self::path_pages_meta_(path).as_path())
}
#[inline]
pub fn get(&mut self, index: I) -> Result<Option<&T>> {
self.get_(index.to_usize()?)
}
#[inline]
pub fn get_(&mut self, index: usize) -> Result<Option<&T>> {
match self.index_to_pushed_index(index) {
Ok(index) => {
if let Some(index) = index {
return Ok(self.pushed().get(index));
}
}
Err(Error::IndexTooHigh) => return Ok(None),
Err(Error::IndexTooLow) => {}
Err(error) => return Err(error),
}
let page_index = Self::index_to_page_index(index);
if self.page().is_none_or(|b| b.0 != page_index) {
let values = self.decode_page(page_index)?;
self.mut_page().replace((page_index, values));
}
self.page().unwrap().1.get(index)
}
pub fn get_last(&mut self) -> Result<Option<&T>> {
let len = self.len();
if len == 0 {
return Ok(None);
}
self.get_(len - 1)
}
pub fn read(&self, index: I) -> Result<Option<T>> {
self.read_(index.to_usize()?)
}
pub fn read_(&self, index: usize) -> Result<Option<T>> {
Ok(match self {
Self::Raw { .. } => {
let mut file = self.open_file()?;
let byte_index = Self::index_to_byte_index(index);
file.seek(SeekFrom::Start(byte_index))?;
let mut buf = vec![0; Self::SIZE_OF_T];
file.read_exact(&mut buf)?;
T::try_ref_from_bytes(&buf[..]).ok().map(|v| v.to_owned())
}
Self::Compressed { .. } => self
.decode_page(Self::index_to_page_index(index))?
.get(index)?
.cloned(),
})
}
pub fn iter<F>(&mut self, f: F) -> Result<()>
where
F: FnMut((I, &T)) -> Result<()>,
{
self.iter_from(I::default(), f)
}
pub fn iter_from<F>(&mut self, mut index: I, mut f: F) -> Result<()>
where
F: FnMut((I, &T)) -> Result<()>,
{
if !self.is_pushed_empty() {
return Err(Error::UnsupportedUnflushedState);
}
let stored_len = I::from(self.stored_len());
while index < stored_len {
let v = self.get(index)?.unwrap();
f((index, v))?;
index = index + 1;
}
Ok(())
}
pub fn iter_from_cloned<F>(&mut self, mut index: I, mut f: F) -> Result<()>
where
F: FnMut((I, T, &mut Self)) -> Result<()>,
{
if !self.is_pushed_empty() {
return Err(Error::UnsupportedUnflushedState);
}
let stored_len = I::from(self.stored_len());
while index < stored_len {
let v = self.get(index)?.unwrap().clone();
f((index, v, self))?;
index = index + 1;
}
Ok(())
}
pub fn collect_range(&self, from: Option<i64>, to: Option<i64>) -> Result<Vec<T>> {
if !self.is_pushed_empty() {
return Err(Error::UnsupportedUnflushedState);
}
let len = self
.base()
.read_stored_length()
.unwrap()
.to_usize()
.unwrap();
if len == 0 {
return Err(Error::IndexTooHigh);
}
let from = from.map_or(0, |from| {
if from >= 0 {
from as usize
} else {
let from = len as i64 + from;
if from < 0 { 0 } else { from as usize }
}
});
let to = to.map_or(len - 1, |to| {
if to >= 0 {
to as usize
} else {
let max = len - 1;
let to = max as i64 + to;
if to > max as i64 { max } else { to as usize }
}
});
if from > to {
return Err(Error::RangeFromAfterTo(from, to));
}
let mut page: Option<(usize, Values<T>)> = None;
let values = (from..=to)
.flat_map(|index| {
let page_index = Self::index_to_page_index(index);
if page.as_ref().is_none_or(|b| b.0 != page_index) {
let pages_meta = match self {
Self::Raw { .. } => None,
Self::Compressed { .. } => Some(self.read_pages_meta().unwrap()),
};
let values = Self::decode_page_(
len,
page_index,
&self.base().open_file().unwrap(),
pages_meta.as_ref(),
)
.inspect_err(|_| {
dbg!(from, to);
})
.unwrap();
page.replace((page_index, values));
}
page.as_ref().unwrap().1.get(index).ok().flatten().cloned()
})
.collect::<Vec<_>>();
Ok(values)
}
pub fn decode_page(&self, page_index: usize) -> Result<Values<T>> {
Self::decode_page_(
self.stored_len(),
page_index,
self.file(),
match self {
Self::Raw { .. } => None,
Self::Compressed { pages_meta, .. } => Some(pages_meta),
},
)
}
fn decode_page_(
stored_len: usize,
page_index: usize,
file: &File,
compressed_pages_meta: Option<&CompressedPagesMetadata>,
) -> Result<Values<T>> {
if Self::page_index_to_index(page_index) >= stored_len {
return Err(Error::IndexTooHigh);
}
let (len, offset) = if let Some(pages_meta) = compressed_pages_meta {
if pages_meta.len() <= page_index {
return Err(Error::ExpectVecToHaveIndex);
}
let page = pages_meta.get(page_index).unwrap();
(page.bytes_len as usize, page.start)
} else {
(Self::PAGE_SIZE, Self::page_index_to_byte_index(page_index))
};
let mmap = unsafe {
memmap2::MmapOptions::new()
.len(len)
.offset(offset)
.map(file)?
};
let compressed = compressed_pages_meta.is_some();
if compressed {
let decoded = zstd::decode_all(&mmap[..]);
if decoded.is_err() {
dbg!((len, offset, page_index, &mmap[..], &mmap.len(), &decoded));
}
Ok(Values::from(
decoded?
.chunks(Self::SIZE_OF_T)
.map(|slice| T::try_read_from_bytes(slice).unwrap())
.collect::<Vec<_>>()
.into_boxed_slice(),
))
} else {
Ok(Values::from(mmap))
}
}
#[inline]
pub fn push(&mut self, value: T) {
self.mut_base().pushed.push(value)
}
pub fn flush(&mut self) -> io::Result<()> {
let pushed_len = self.pushed_len();
if pushed_len == 0 {
return Ok(());
}
let stored_len = self.stored_len();
let bytes = match self {
Self::Compressed { base, pages_meta } => {
let (starting_page_index, values) = if *base.stored_len % Self::PER_PAGE != 0 {
if pages_meta.is_empty() {
unreachable!()
}
let last_page_index = pages_meta.len() - 1;
let values = if let Some(values) = base
.pages
.as_mut()
.and_then(|big_cache| big_cache.last_mut().and_then(|lock| lock.take()))
{
values
} else if base
.page
.as_ref()
.is_some_and(|(page_index, _)| *page_index == last_page_index)
{
base.page.take().unwrap().1
} else {
Self::decode_page_(
stored_len,
last_page_index,
&base.file,
Some(pages_meta),
)
.inspect_err(|_| {
dbg!(last_page_index, &pages_meta);
})
.unwrap()
};
let file_len = pages_meta.pop().unwrap().start;
file_set_len(&mut base.file, file_len)?;
(last_page_index, values)
} else {
(pages_meta.len(), Values::default())
};
let compressed = Vec::from(values.as_arr())
.into_par_iter()
.chain(mem::take(&mut base.pushed).into_par_iter())
.chunks(Self::PER_PAGE)
.map(|chunk| (Self::compress_chunk(chunk.as_ref()), chunk.len()))
.collect::<Vec<_>>();
compressed
.iter()
.enumerate()
.for_each(|(i, (compressed_bytes, values_len))| {
let page_index = starting_page_index + i;
let start = if page_index != 0 {
let prev = pages_meta.get(page_index - 1).unwrap();
prev.start + prev.bytes_len as u64
} else {
0
};
let bytes_len = compressed_bytes.len() as u32;
let values_len = *values_len as u32;
let page = CompressedPageMetadata::new(start, bytes_len, values_len);
pages_meta.push(page_index, page);
});
pages_meta.write()?;
compressed
.into_iter()
.flat_map(|(v, _)| v)
.collect::<Vec<_>>()
}
Self::Raw { base } => {
let pushed = &mut base.pushed;
let mut bytes: Vec<u8> = vec![0; pushed.len() * Self::SIZE_OF_T];
let unsafe_bytes = UnsafeSlice::new(&mut bytes);
mem::take(pushed)
.into_par_iter()
.enumerate()
.for_each(|(i, v)| unsafe_bytes.copy_slice(i * Self::SIZE_OF_T, v.as_bytes()));
bytes
}
};
let file = self.mut_file();
file.write_all(&bytes)?;
file.sync_all()?;
self.reset_caches();
self.increase_stored_len(pushed_len);
self.write_stored_length()?;
Ok(())
}
pub fn truncate_if_needed(&mut self, index: I) -> Result<()> {
let index = index.to_usize()?;
if index >= self.stored_len() {
return Ok(());
}
if index == 0 {
self.reset()?;
return Ok(());
}
let page_index = Self::index_to_page_index(index);
let values = match self {
Self::Compressed { .. } => self.decode_page(page_index)?,
Self::Raw { .. } => Values::default(),
};
let (len, bytes) = match self {
Self::Compressed { pages_meta, .. } => {
let mut page = pages_meta.truncate(page_index).unwrap();
let len = page.start;
let decoded_index = Self::index_to_decoded_index(index);
let compressed = if decoded_index != 0 {
let chunk = &values.as_arr()[..decoded_index];
let compressed = Self::compress_chunk(chunk);
page.values_len = chunk.len() as u32;
page.bytes_len = compressed.len() as u32;
pages_meta.push(page_index, page);
compressed
} else {
vec![].into_boxed_slice()
};
pages_meta.write()?;
(len, compressed)
}
Self::Raw { .. } => {
// let value_at_index = self.open_then_read_(index).ok();
let len = Self::index_to_byte_index(index);
(len, vec![].into_boxed_slice())
}
};
let file = self.mut_file();
file_set_len(file, len)?;
if !bytes.is_empty() {
file.write_all(&bytes)?;
}
self.set_stored_len(index);
self.write_stored_length()?;
self.reset_caches();
Ok(())
}
fn compress_chunk(chunk: &[T]) -> Box<[u8]> {
if chunk.len() > Self::PER_PAGE {
panic!();
}
let mut bytes: Vec<u8> = vec![0; chunk.len() * Self::SIZE_OF_T];
let unsafe_bytes = UnsafeSlice::new(&mut bytes);
chunk
.into_par_iter()
.enumerate()
.for_each(|(i, v)| unsafe_bytes.copy_slice(i * Self::SIZE_OF_T, v.as_bytes()));
zstd::encode_all(bytes.as_slice(), DEFAULT_COMPRESSION_LEVEL)
.unwrap()
.into_boxed_slice()
}
pub fn enable_large_cache_if_possible(&mut self) {
self.mut_pages().replace(vec![]);
self.reset_large_cache();
}
pub fn disable_large_cache_if_possible(&mut self) {
self.mut_base().pages.take();
}
fn reset_large_cache(&mut self) {
let stored_len = self.stored_len();
if let Some(pages) = self.mut_pages().as_mut() {
pages.par_iter_mut().for_each(|lock| {
lock.take();
});
let len = (stored_len as f64 / Self::PER_PAGE as f64).ceil() as usize;
let len = Self::CACHE_LENGTH.min(len);
if pages.len() != len {
pages.resize_with(len, Default::default);
}
}
}
pub fn large_cache_len(&self) -> usize {
self.pages().map_or(0, |v| v.len())
}
fn reset_small_cache(&mut self) {
self.mut_base().page.take();
}
fn reset_caches(&mut self) {
self.reset_small_cache();
self.reset_large_cache();
}
pub fn reset(&mut self) -> Result<()> {
self.mut_base().reset_file()?;
self.reset_stored_len();
self.reset_caches();
Ok(())
}
#[inline]
pub fn index_to_pushed_index(&self, index: usize) -> Result<Option<usize>> {
let stored_len = self.stored_len();
if index >= stored_len {
let index = index - stored_len;
if index >= self.pushed_len() {
Err(Error::IndexTooHigh)
} else {
Ok(Some(index))
}
} else {
Err(Error::IndexTooLow)
}
}
#[inline]
fn index_to_byte_index(index: usize) -> u64 {
(index * Self::SIZE_OF_T) as u64
}
#[inline(always)]
fn index_to_page_index(index: usize) -> usize {
index / Self::PER_PAGE
}
#[inline(always)]
fn page_index_to_index(page_index: usize) -> usize {
page_index * Self::PER_PAGE
}
#[inline(always)]
fn page_index_to_byte_index(page_index: usize) -> u64 {
(page_index * Self::PAGE_SIZE) as u64
}
#[inline(always)]
fn index_to_decoded_index(index: usize) -> usize {
index % Self::PER_PAGE
}
#[inline]
fn path_pages_meta_(path: &Path) -> PathBuf {
path.join("pages_meta")
}
// #[inline]
// fn page(&self) -> Option<&(usize, Values<T>)> {
// self.base().page.as_ref()
// }
// #[inline]
// fn mut_page(&mut self) -> &mut Option<(usize, Values<T>)> {
// &mut self.mut_base().page
// }
// #[inline]
// pub fn pages(&self) -> Option<&Vec<OnceLock<Values<T>>>> {
// self.base().pages.as_ref()
// }
// #[inline]
// fn mut_pages(&mut self) -> &mut Option<Vec<OnceLock<Values<T>>>> {
// &mut self.mut_base().pages
// }
#[inline]
pub fn len(&self) -> usize {
self.stored_len() + self.pushed_len()
}
pub fn is_empty(&self) -> bool {
self.len() == 0
}
#[inline]
pub fn has(&self, index: I) -> Result<bool> {
Ok(self.has_(index.to_usize()?))
}
#[inline]
fn has_(&self, index: usize) -> bool {
index < self.len()
}
#[inline]
pub fn pushed(&self) -> &Vec<T> {
&self.base().pushed
}
#[inline]
pub fn pushed_len(&self) -> usize {
self.pushed().len()
}
#[inline]
fn is_pushed_empty(&self) -> bool {
self.pushed_len() == 0
}
#[inline]
pub fn stored_len(&self) -> usize {
*self.base().stored_len
}
#[inline]
fn set_stored_len(&mut self, len: usize) {
*self.mut_base().stored_len = len;
}
fn increase_stored_len(&mut self, len: usize) {
*self.mut_base().stored_len += len;
}
#[inline]
fn reset_stored_len(&mut self) {
self.set_stored_len(0);
}
fn write_stored_length(&self) -> io::Result<()> {
self.base().write_stored_length()
}
#[inline]
pub fn path(&self) -> &Path {
&self.base().pathbuf
}
fn file(&self) -> &File {
&self.base().file
}
fn mut_file(&mut self) -> &mut File {
&mut self.mut_base().file
}
fn open_file(&self) -> io::Result<File> {
self.base().open_file()
}
pub fn file_name(&self) -> String {
self.path()
.file_name()
.unwrap()
.to_str()
.unwrap()
.to_owned()
}
#[inline]
pub fn version(&self) -> Version {
self.base().version
}
#[inline]
fn compressed(&self) -> Compressed {
self.base().compressed
}
#[inline]
fn base(&self) -> &Base<I, T> {
match self {
Self::Raw { base, .. } => base,
Self::Compressed { base, .. } => base,
}
}
#[inline]
fn mut_base(&mut self) -> &mut Base<I, T> {
match self {
Self::Raw { base, .. } => base,
Self::Compressed { base, .. } => base,
}
}
pub fn index_type_to_string(&self) -> &str {
I::to_string()
}
}
#[derive(Debug)]
struct Base<I, T> {
pub version: Version,
pub pathbuf: PathBuf,
pub stored_len: Length,
pub compressed: Compressed,
pub mmap: ArcSwap<Mmap>,
// pub page: Option<(usize, Values<T>)>,
// pub pages: Option<Vec<OnceLock<Values<T>>>>,
pub pushed: Vec<T>,
pub file: File,
pub phantom: PhantomData<I>,
}
impl<I, T> Base<I, T> {
pub fn import(path: &Path, version: Version, compressed: Compressed) -> Result<Self> {
fs::create_dir_all(path)?;
let version_path = Self::path_version_(path);
version.validate(version_path.as_ref())?;
version.write(version_path.as_ref())?;
let compressed_path = Self::path_compressed_(path);
compressed.validate(compressed_path.as_ref())?;
compressed.write(compressed_path.as_ref())?;
let stored_len = Length::try_from(Self::path_length_(path).as_path())?;
let file = Self::open_file_(Self::path_vec_(path).as_path())?;
Ok(Self {
mmap: ArcSwap::new(Arc::new(unsafe { Mmap::map(&file)? })),
version,
compressed,
pathbuf: path.to_owned(),
file,
stored_len,
pushed: vec![],
phantom: PhantomData,
})
}
fn open_file(&self) -> io::Result<File> {
Self::open_file_(&self.path_vec())
}
fn open_file_(path: &Path) -> io::Result<File> {
OpenOptions::new()
.read(true)
.create(true)
.truncate(false)
.append(true)
.open(path)
}
fn reset_file(&mut self) -> Result<()> {
file_set_len(&mut self.file, 0)?;
Ok(())
}
#[inline]
fn path_vec(&self) -> PathBuf {
Self::path_vec_(&self.pathbuf)
}
#[inline]
fn path_vec_(path: &Path) -> PathBuf {
path.join("vec")
}
pub fn read_stored_length(&self) -> Result<Length> {
Length::try_from(self.path_length().as_path())
}
fn write_stored_length(&self) -> io::Result<()> {
self.stored_len.write(&self.path_length())
}
#[inline]
fn path_length(&self) -> PathBuf {
Self::path_length_(&self.pathbuf)
}
#[inline]
fn path_length_(path: &Path) -> PathBuf {
path.join("length")
}
#[inline]
fn path_version_(path: &Path) -> PathBuf {
path.join("version")
}
#[inline]
fn path_compressed_(path: &Path) -> PathBuf {
path.join("compressed")
}
}
impl<I, T> Clone for StorableVec<I, T>
where
I: StoredIndex,
T: StoredType,
{
fn clone(&self) -> Self {
Self::import(self.path(), self.version(), self.compressed()).unwrap()
}
}
fn file_set_len(file: &mut File, len: u64) -> io::Result<()> {
file.set_len(len)?;
file.seek(SeekFrom::End(0))?;
Ok(())
}
crates/brk_vec/src/traits/any.rs
+223 -37
View File
@@ -1,61 +1,247 @@
use std::{io, path::PathBuf}; // use std::{io, path::PathBuf};
use crate::{Result, StorableVec}; // use crate::{Result};
use std::{
fs::{File, OpenOptions},
io::{self, Seek, SeekFrom, Write},
path::{Path, PathBuf},
sync::Arc,
};
use arc_swap::{ArcSwap, Guard};
use axum::{
Json,
response::{IntoResponse, Response},
};
use memmap2::Mmap;
use crate::{Error, Result, Value, Version};
use super::{StoredIndex, StoredType}; use super::{StoredIndex, StoredType};
pub trait AnyStorableVec: Send + Sync { pub trait AnyVec<I, T>: Send + Sync
fn file_name(&self) -> String;
fn index_type_to_string(&self) -> &str;
fn len(&self) -> usize;
fn is_empty(&self) -> bool;
fn collect_range_values(
&self,
from: Option<i64>,
to: Option<i64>,
) -> Result<Vec<serde_json::Value>>;
fn flush(&mut self) -> io::Result<()>;
fn path_vec(&self) -> PathBuf;
}
impl<I, T> AnyStorableVec for StorableVec<I, T>
where where
I: StoredIndex, I: StoredIndex + Sized,
T: StoredType, T: StoredType,
Self: Sized,
{ {
fn file_name(&self) -> String { const SIZE_OF_T: usize = size_of::<T>();
self.file_name()
fn open_file(&self) -> io::Result<File> {
Self::open_file_(&self.path_vec())
}
fn open_file_(path: &Path) -> io::Result<File> {
OpenOptions::new()
.read(true)
.create(true)
.truncate(false)
.append(true)
.open(path)
} }
fn index_type_to_string(&self) -> &str { fn file_set_len(&mut self, len: u64) -> Result<()> {
self.index_type_to_string() let mut file = self.open_file()?;
file.set_len(len)?;
file.seek(SeekFrom::End(0))?;
self.update_mmap(file)
} }
fn file_write_all(&mut self, buf: &[u8]) -> Result<()> {
let mut file = self.open_file()?;
file.write_all(buf)?;
self.update_mmap(file)
}
#[inline]
fn reset(&mut self) -> Result<()> {
self.file_write_all(&[])?;
Ok(())
}
fn mmap(&self) -> &ArcSwap<Mmap>;
#[inline]
fn new_guard(&self) -> Guard<Arc<Mmap>> {
self.mmap().load()
}
fn guard(&self) -> &Option<Guard<Arc<Mmap>>>;
fn mut_guard(&mut self) -> &mut Option<Guard<Arc<Mmap>>>;
#[inline]
fn guard_to_value(guard: &Guard<Arc<Mmap>>, index: usize) -> T {
let index = index * Self::SIZE_OF_T;
let slice = &guard[index..(index + Self::SIZE_OF_T)];
let v = T::try_ref_from_bytes(slice).unwrap();
v.clone()
}
fn new_mmap(file: File) -> Result<Arc<Mmap>> {
Ok(Arc::new(unsafe { Mmap::map(&file)? }))
}
fn update_mmap(&mut self, file: File) -> Result<()> {
file.sync_all()?;
let mmap = Self::new_mmap(file)?;
self.mmap().store(mmap);
if self.guard().is_some() {
let guard = self.new_guard();
self.mut_guard().replace(guard);
} else {
unreachable!()
}
Ok(())
}
#[inline]
fn get(&mut self, index: I) -> Result<Option<Value<T>>> {
self.get_(index.to_usize()?)
}
fn get_(&mut self, index: usize) -> Result<Option<Value<T>>>;
fn get_last(&mut self) -> Result<Option<Value<T>>> {
let len = self.len();
if len == 0 {
return Ok(None);
}
self.get_(len - 1)
}
#[inline]
fn stored_len(&self) -> usize {
if let Some(guard) = self.guard() {
guard.len() / Self::SIZE_OF_T
} else {
self.new_guard().len() / Self::SIZE_OF_T
}
}
fn pushed(&self) -> &[T];
#[inline]
fn pushed_len(&self) -> usize {
self.pushed().len()
}
fn mut_pushed(&mut self) -> &mut Vec<T>;
#[inline]
fn push(&mut self, value: T) {
self.mut_pushed().push(value)
}
#[inline]
fn is_pushed_empty(&self) -> bool {
self.pushed_len() == 0
}
#[inline]
fn index_to_pushed_index(&self, index: usize) -> Result<Option<usize>> {
let stored_len = self.stored_len();
if index >= stored_len {
let index = index - stored_len;
if index >= self.pushed_len() {
Err(Error::IndexTooHigh)
} else {
Ok(Some(index))
}
} else {
Err(Error::IndexTooLow)
}
}
#[inline]
fn len(&self) -> usize { fn len(&self) -> usize {
self.len() self.stored_len() + self.pushed_len()
} }
#[inline]
fn has(&self, index: I) -> Result<bool> {
Ok(self.has_(index.to_usize()?))
}
#[inline]
fn has_(&self, index: usize) -> bool {
index < self.len()
}
#[inline]
fn is_empty(&self) -> bool { fn is_empty(&self) -> bool {
self.is_empty() self.len() == 0
} }
fn flush(&mut self) -> io::Result<()> { #[inline]
self.flush() fn index_type_to_string(&self) -> &str {
I::to_string()
} }
fn collect_range_values( #[inline]
&self, fn iter<F>(&mut self, f: F) -> Result<()>
from: Option<i64>, where
to: Option<i64>, F: FnMut((I, T, &mut Self)) -> Result<()>,
) -> Result<Vec<serde_json::Value>> { {
Ok(self self.iter_from(I::default(), f)
.collect_range(from, to)?
.into_iter()
.map(|v| serde_json::to_value(v).unwrap())
.collect::<Vec<_>>())
} }
fn iter_from<F>(&mut self, index: I, f: F) -> Result<()>
where
F: FnMut((I, T, &mut Self)) -> Result<()>;
fn fix_i64(i: i64, len: usize, from: bool) -> usize {
if i >= 0 {
let v = i as usize;
if v < len {
v
} else if from {
len - 1
} else {
len
}
} else {
let v = len as i64 + i;
if v < 0 { 0 } else { v as usize }
}
}
fn flush(&mut self) -> Result<()>;
fn truncate_if_needed(&mut self, index: I) -> Result<()>;
fn collect_range(&self, from: Option<i64>, to: Option<i64>) -> Result<Json<Vec<T>>>;
fn collect_range_response(&self, from: Option<i64>, to: Option<i64>) -> Result<Response> {
Ok(self.collect_range(from, to)?.into_response())
}
fn path(&self) -> &Path;
#[inline]
fn path_vec(&self) -> PathBuf { fn path_vec(&self) -> PathBuf {
self.base().path_vec() Self::path_vec_(self.path())
}
#[inline]
fn path_vec_(path: &Path) -> PathBuf {
path.join("vec")
}
#[inline]
fn path_version_(path: &Path) -> PathBuf {
path.join("version")
}
fn file_name(&self) -> String {
self.path()
.file_name()
.unwrap()
.to_str()
.unwrap()
.to_owned()
}
fn version(&self) -> Version;
fn any(&self) -> &Self {
self
}
fn mut_any(&mut self) -> &mut Self {
self
} }
} }
crates/brk_vec/src/traits/mod.rs
-1
View File
@@ -1,5 +1,4 @@
mod any; mod any;
// mod bytes;
mod stored_index; mod stored_index;
mod stored_type; mod stored_type;
crates/brk_vec/src/variants/compressed.rs
+117
View File
@@ -0,0 +1,117 @@
use std::{
fs,
path::Path,
sync::{Arc, OnceLock},
};
use arc_swap::{ArcSwap, Guard};
use axum::Json;
use memmap2::Mmap;
use crate::{
AnyVec, CompressedPagesMetadata, Error, RawVec, Result, StoredIndex, StoredType, Value, Version,
};
#[derive(Debug)]
pub struct CompressedVec<I, T> {
inner: RawVec<I, T>,
decoded_page: Option<(usize, Box<[T]>)>,
pages_meta: CompressedPagesMetadata,
decoded_pages: Option<Vec<OnceLock<Box<[T]>>>>,
// pages: Option<Vec<OnceLock<Values<T>>>>,
// page: Option<(usize, Values<T>)>,
// length: Length
}
impl<I, T> CompressedVec<I, T>
where
I: StoredIndex,
T: StoredType,
{
/// Same as import but will reset the folder under certain errors, so be careful !
pub fn forced_import(path: &Path, version: Version) -> Result<Self> {
let res = Self::import(path, version);
match res {
Err(Error::WrongEndian)
| Err(Error::DifferentVersion { .. })
| Err(Error::DifferentCompressionMode) => {
fs::remove_dir_all(path)?;
Self::import(path, version)
}
_ => res,
}
}
pub fn import(path: &Path, version: Version) -> Result<Self> {
Ok(Self {
inner: RawVec::import(path, version)?,
decoded_page: None,
decoded_pages: None,
pages_meta: CompressedPagesMetadata::read(path)?,
})
}
}
impl<I, T> AnyVec<I, T> for CompressedVec<I, T>
where
I: StoredIndex,
T: StoredType,
{
#[inline]
fn get_(&mut self, index: usize) -> Result<Option<Value<T>>> {
self.inner.get_(index)
}
fn iter_from<F>(&mut self, _index: I, _f: F) -> Result<()>
where
F: FnMut((I, T, &mut Self)) -> Result<()>,
{
todo!()
// self.inner.iter_from(index, f)
}
fn collect_range(&self, from: Option<i64>, to: Option<i64>) -> Result<Json<Vec<T>>> {
self.inner.collect_range(from, to)
}
fn flush(&mut self) -> Result<()> {
self.inner.flush()
}
fn truncate_if_needed(&mut self, index: I) -> Result<()> {
self.inner.truncate_if_needed(index)
}
#[inline]
fn mmap(&self) -> &ArcSwap<Mmap> {
self.inner.mmap()
}
#[inline]
fn guard(&self) -> &Option<Guard<Arc<Mmap>>> {
self.inner.guard()
}
#[inline]
fn mut_guard(&mut self) -> &mut Option<Guard<Arc<Mmap>>> {
self.inner.mut_guard()
}
#[inline]
fn pushed(&self) -> &[T] {
self.inner.pushed()
}
#[inline]
fn mut_pushed(&mut self) -> &mut Vec<T> {
self.inner.mut_pushed()
}
#[inline]
fn path(&self) -> &Path {
self.inner.path()
}
#[inline]
fn version(&self) -> Version {
self.inner.version()
}
}
crates/brk_vec/src/variants/mod.rs
+5
View File
@@ -0,0 +1,5 @@
mod compressed;
mod raw;
pub use compressed::*;
pub use raw::*;
crates/brk_vec/src/variants/raw.rs
+214
View File
@@ -0,0 +1,214 @@
use std::{
fs,
marker::PhantomData,
mem,
path::{Path, PathBuf},
sync::Arc,
};
use arc_swap::{ArcSwap, Guard};
use axum::Json;
use memmap2::Mmap;
use rayon::prelude::*;
use crate::{AnyVec, Error, Result, StoredIndex, StoredType, UnsafeSlice, Value, Version};
#[derive(Debug)]
pub struct RawVec<I, T> {
version: Version,
pathbuf: PathBuf,
// Consider Arc<ArcSwap<Option<Mmap>>> for dataraces when reorg ?
mmap: Arc<ArcSwap<Mmap>>,
guard: Option<Guard<Arc<Mmap>>>,
pushed: Vec<T>,
phantom: PhantomData<I>,
}
impl<I, T> RawVec<I, T>
where
I: StoredIndex,
T: StoredType,
{
/// Same as import but will reset the folder under certain errors, so be careful !
pub fn forced_import(path: &Path, version: Version) -> Result<Self> {
let res = Self::import(path, version);
match res {
Err(Error::WrongEndian) | Err(Error::DifferentVersion { .. }) => {
fs::remove_dir_all(path)?;
Self::import(path, version)
}
_ => res,
}
}
pub fn import(path: &Path, version: Version) -> Result<Self> {
fs::create_dir_all(path)?;
let version_path = Self::path_version_(path);
version.validate(version_path.as_ref())?;
version.write(version_path.as_ref())?;
let file = Self::open_file_(Self::path_vec_(path).as_path())?;
let mmap = Arc::new(ArcSwap::new(Self::new_mmap(file)?));
let guard = Some(mmap.load());
Ok(Self {
mmap,
guard,
version,
pathbuf: path.to_owned(),
pushed: vec![],
phantom: PhantomData,
})
}
}
impl<I, T> AnyVec<I, T> for RawVec<I, T>
where
I: StoredIndex,
T: StoredType,
{
#[inline]
fn get_(&mut self, index: usize) -> Result<Option<Value<T>>> {
match self.index_to_pushed_index(index) {
Ok(index) => {
if let Some(index) = index {
return Ok(self.pushed().get(index).map(|v| Value::Ref(v)));
}
}
Err(Error::IndexTooHigh) => return Ok(None),
Err(Error::IndexTooLow) => {}
Err(error) => return Err(error),
}
let v = if let Some(guard) = self.guard.as_ref() {
Self::guard_to_value(guard, index)
} else {
Self::guard_to_value(&self.new_guard(), index)
};
Ok(Some(Value::Owned(v)))
}
fn iter_from<F>(&mut self, index: I, mut f: F) -> Result<()>
where
F: FnMut((I, T, &mut Self)) -> Result<()>,
{
if !self.is_pushed_empty() {
return Err(Error::UnsupportedUnflushedState);
}
let guard = self.mmap.load();
let start = index.to_usize()? * Self::SIZE_OF_T;
guard[start..]
.chunks(Self::SIZE_OF_T)
.enumerate()
.try_for_each(|(i, chunk)| {
let v = T::try_read_from_bytes(chunk).unwrap();
f((I::from(i), v, self))
})?;
Ok(())
}
fn flush(&mut self) -> Result<()> {
let pushed_len = self.pushed_len();
if pushed_len == 0 {
return Ok(());
}
let bytes = {
let pushed = &mut self.pushed;
let mut bytes: Vec<u8> = vec![0; pushed.len() * Self::SIZE_OF_T];
let unsafe_bytes = UnsafeSlice::new(&mut bytes);
mem::take(pushed)
.into_par_iter()
.enumerate()
.for_each(|(i, v)| unsafe_bytes.copy_slice(i * Self::SIZE_OF_T, v.as_bytes()));
bytes
};
self.file_write_all(&bytes)?;
Ok(())
}
fn truncate_if_needed(&mut self, index: I) -> Result<()> {
let index = index.to_usize()?;
if index >= self.stored_len() {
return Ok(());
}
if index == 0 {
self.reset()?;
return Ok(());
}
let len = index * Self::SIZE_OF_T;
self.file_set_len(len as u64)?;
Ok(())
}
fn collect_range(&self, from: Option<i64>, to: Option<i64>) -> Result<Json<Vec<T>>> {
let guard = self.mmap.load();
let len = guard.len() / Self::SIZE_OF_T;
if len == 0 {
return Ok(Json(vec![]));
}
let from = from.map_or(0, |i| Self::fix_i64(i, len, true)) * Self::SIZE_OF_T;
let to = to.map_or(len, |i| Self::fix_i64(i, len, false)) * Self::SIZE_OF_T;
Ok(Json(
guard[from * Self::SIZE_OF_T..to * Self::SIZE_OF_T]
.chunks(Self::SIZE_OF_T)
.map(|chunk| T::try_read_from_bytes(chunk).unwrap())
.collect::<Vec<_>>(),
))
}
#[inline]
fn mmap(&self) -> &ArcSwap<Mmap> {
&self.mmap
}
#[inline]
fn guard(&self) -> &Option<Guard<Arc<Mmap>>> {
&self.guard
}
#[inline]
fn mut_guard(&mut self) -> &mut Option<Guard<Arc<Mmap>>> {
&mut self.guard
}
#[inline]
fn pushed(&self) -> &[T] {
self.pushed.as_slice()
}
#[inline]
fn mut_pushed(&mut self) -> &mut Vec<T> {
&mut self.pushed
}
#[inline]
fn path(&self) -> &Path {
self.pathbuf.as_path()
}
#[inline]
fn version(&self) -> Version {
self.version
}
}