Mirror/brk
1
0
Fork 0
mirror of https://github.com/bitcoinresearchkit/brk.git synced 2026-07-31 04:33:05 -07:00
Code Issues Packages Projects Releases Wiki Activity

global: snapshot

Browse Source
This commit is contained in:
nym21
2025-08-03 23:38:58 +02:00
parent f7aa9424db
commit a2f5704581
50 changed files with 818 additions and 704 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/brk_vecs/.gitignore
+1
View File
@@ -1,2 +1,3 @@
/vecs
/raw
/compressed
crates/brk_vecs/Cargo.toml
+1
View File
@@ -18,6 +18,7 @@ memmap2 = "0.9.7"
parking_lot = { workspace = true }
rayon = { workspace = true }
serde = { workspace = true }
serde_derive = { workspace = true }
serde_json = { workspace = true }
zerocopy = { workspace = true }
zerocopy-derive = { workspace = true }
crates/brk_vecs/examples/compressed.rs
+153
View File
@@ -0,0 +1,153 @@
use std::{fs, path::Path, sync::Arc};
use brk_vecs::{
AnyStoredVec, AnyVec, CollectableVec, CompressedVec, File, GenericStoredVec, Stamp,
VecIterator, Version,
};
#[allow(clippy::upper_case_acronyms)]
type VEC = CompressedVec<usize, u32>;
fn main() -> Result<(), Box<dyn std::error::Error>> {
let _ = fs::remove_dir_all("compressed");
let version = Version::TWO;
let file = Arc::new(File::open(Path::new("compressed"))?);
{
let mut vec: VEC = CompressedVec::forced_import(&file, "vec", version)?;
(0..21_u32).for_each(|v| {
vec.push(v);
});
let mut iter = vec.into_iter();
dbg!(iter.get(0));
dbg!(iter.get(1));
dbg!(iter.get(2));
dbg!(iter.get(20));
dbg!(iter.get(21));
drop(iter);
dbg!("flush");
vec.flush()?;
dbg!("flushed");
dbg!(vec.header());
}
{
let mut vec: VEC = CompressedVec::forced_import(&file, "vec", version)?;
vec.mut_header().update_stamp(Stamp::new(100));
let mut iter = vec.into_iter();
dbg!(iter.get(0));
dbg!(iter.get(1));
dbg!(iter.get(2));
dbg!(iter.get(3));
dbg!(iter.get(4));
dbg!(iter.get(5));
dbg!(iter.get(20));
dbg!(iter.get(20));
dbg!(iter.get(0));
drop(iter);
vec.push(21);
vec.push(22);
let mut iter = vec.into_iter();
dbg!(iter.get(20));
dbg!(iter.get(21));
dbg!(iter.get(22));
dbg!(iter.get(23));
drop(iter);
vec.flush()?;
}
{
let mut vec: VEC = CompressedVec::forced_import(&file, "vec", version)?;
let mut iter = vec.into_iter();
dbg!(iter.get(0));
dbg!(iter.get(20));
dbg!(iter.get(21));
dbg!(iter.get(22));
drop(iter);
vec.truncate_if_needed(14)?;
let mut iter = vec.into_iter();
dbg!(iter.get(0));
dbg!(iter.get(5));
dbg!(iter.get(20));
drop(iter);
dbg!(vec.collect_signed_range(Some(-5), None)?);
vec.push(vec.len() as u32);
dbg!(VecIterator::last(vec.into_iter()));
dbg!(vec.into_iter().collect::<Vec<_>>());
}
{
let mut vec: VEC = CompressedVec::forced_import(&file, "vec", version)?;
vec.reset()?;
dbg!(vec.header(), vec.pushed_len(), vec.stored_len(), vec.len());
(0..21_u32).for_each(|v| {
vec.push(v);
});
let mut iter = vec.into_iter();
dbg!(iter.get(0));
dbg!(iter.get(20));
dbg!(iter.get(21));
drop(iter);
// let reader = vec.create_static_reader();
// dbg!(vec.take(10, &reader)?);
// dbg!(vec.get_or_read(10, &reader)?);
// dbg!(vec.holes());
// drop(reader);
vec.flush()?;
dbg!(vec.holes());
}
{
let mut vec: VEC = CompressedVec::forced_import(&file, "vec", version)?;
dbg!(vec.holes());
let reader = vec.create_static_reader();
dbg!(vec.get_or_read(10, &reader)?);
drop(reader);
// vec.update(10, 10)?;
// vec.update(0, 10)?;
let reader = vec.create_static_reader();
dbg!(
vec.holes(),
vec.get_or_read(0, &reader)?,
vec.get_or_read(10, &reader)?
);
drop(reader);
vec.flush()?;
}
{
let vec: VEC = CompressedVec::forced_import(&file, "vec", version)?;
dbg!(vec.collect()?);
}
Ok(())
}
crates/brk_vecs/examples/file.rs
+3 -1
View File
@@ -99,6 +99,8 @@ fn main() -> Result<()> {
);
}
dbg!(1);
file.write_all_to_region_at(region1_i.into(), &[1; 8000], 0)?;
{
@@ -146,7 +148,7 @@ fn main() -> Result<()> {
let layout = file.layout();
assert!(layout.start_to_index().is_empty());
assert!(layout.start_to_hole().is_empty());
assert!(layout.start_to_hole().len() == 1);
}
let (region1_i, _) = file.create_region_if_needed("region1")?;
crates/brk_vecs/examples/raw.rs
+1 -2
View File
@@ -12,8 +12,7 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
let _ = fs::remove_dir_all("raw");
let version = Version::TWO;
// let format = Format::Raw;
//
let file = Arc::new(File::open(Path::new("raw"))?);
{
crates/brk_vecs/src/file/layout.rs
+4
View File
@@ -195,4 +195,8 @@ impl Layout {
unreachable!();
}
}
pub fn reserved(&mut self, start: u64) -> Option<u64> {
self.start_to_reserved.remove(&start)
}
}
crates/brk_vecs/src/file/mod.rs
+24 -27
View File
@@ -178,28 +178,27 @@ impl File {
let data_len = data.len() as u64;
let new_len = at.map_or(len + data_len, |at| (at + data_len).max(len));
let write_start = start + at.unwrap_or(len);
if at.is_some_and(|at| at >= start + reserved) {
return Err(Error::Str("Invalid at parameter"));
}
// Write to reserved space if possible
if new_len <= reserved {
// println!(
// "Write to {region_index} reserved space at {}",
// start + at.unwrap_or(len)
// );
println!("Write to {region_index} reserved space at {write_start}");
if at.is_none() {
self.write(start + len, data);
self.write(write_start, data);
}
let mut region = region_lock.write();
if let Some(at) = at {
self.write(start + at, data);
}
if len != new_len {
region.set_len(new_len);
if at.is_some() {
self.write(write_start, data);
}
region.set_len(new_len);
regions.write_to_mmap(&region, region_index);
return Ok(());
@@ -217,10 +216,7 @@ impl File {
// If is last continue writing
if layout.is_last_anything(region_index) {
// println!(
// "{region_index} Append to file at {}",
// start + at.unwrap_or(len)
// );
println!("{region_index} Append to file at {write_start}");
self.set_min_len(start + new_reserved)?;
let mut region = region_lock.write();
@@ -228,7 +224,7 @@ impl File {
drop(region);
drop(layout);
self.write(start + len, data);
self.write(write_start, data);
let mut region = region_lock.write();
region.set_len(new_len);
@@ -243,7 +239,7 @@ impl File {
.get_hole(hole_start)
.is_some_and(|gap| gap >= added_reserve)
{
// println!("Expand {region_index} to hole");
println!("Expand {region_index} to hole");
layout.remove_or_compress_hole(hole_start, added_reserve);
let mut region = region_lock.write();
@@ -251,7 +247,7 @@ impl File {
drop(region);
drop(layout);
self.write(start + len, data);
self.write(write_start, data);
let mut region = region_lock.write();
region.set_len(new_len);
@@ -262,16 +258,16 @@ impl File {
// Find hole big enough to move the region
if let Some(hole_start) = layout.find_smallest_adequate_hole(new_reserved) {
// println!("Move {region_index} to hole at {hole_start}");
println!("Move {region_index} to hole at {hole_start}");
layout.remove_or_compress_hole(hole_start, new_reserved);
drop(layout);
self.write(
hole_start,
&self.mmap.read()[start as usize..(start + len) as usize],
&self.mmap.read()[start as usize..write_start as usize],
);
self.write(hole_start + len, data);
self.write(hole_start + at.unwrap_or(len), data);
let mut region = region_lock.write();
let mut layout = self.layout.write();
@@ -288,24 +284,25 @@ impl File {
let new_start = layout.len(&regions);
// Write at the end
// println!(
// "Move {region_index} to the end, from {start}..{} to {new_start}..{}",
// start + reserved,
// new_start + new_reserved
// );
println!(
"Move {region_index} to the end, from {start}..{} to {new_start}..{}",
start + reserved,
new_start + new_reserved
);
self.set_min_len(new_start + new_reserved)?;
layout.reserve(new_start, new_reserved);
drop(layout);
self.write(
new_start,
&self.mmap.read()[start as usize..(start + len) as usize],
&self.mmap.read()[start as usize..write_start as usize],
);
self.write(new_start + len, data);
self.write(new_start + at.unwrap_or(len), data);
let mut region = region_lock.write();
let mut layout = self.layout.write();
layout.move_region(new_start, region_index, &region)?;
assert!(layout.reserved(new_start) == Some(new_reserved));
drop(layout);
region.set_start(new_start);
crates/brk_vecs/src/lib.rs
+1 -1
View File
@@ -2,7 +2,7 @@
#![doc = "\n## Examples\n"]
#![doc = "\n### File\n\n```rust"]
#![doc = include_str!("../examples/file.rs")]
#![doc = "```"]
#![doc = "```\n"]
#![doc = "\n### Raw\n\n```rust"]
#![doc = include_str!("../examples/raw.rs")]
#![doc = "```"]
crates/brk_vecs/src/variants/compressed/compressed_page_meta.rs
-18
View File
@@ -1,18 +0,0 @@
use zerocopy_derive::{FromBytes, Immutable, IntoBytes, KnownLayout};
#[derive(Debug, Clone, IntoBytes, Immutable, FromBytes, KnownLayout)]
pub struct CompressedPageMetadata {
pub start: u64,
pub bytes_len: u32,
pub values_len: u32,
}
impl CompressedPageMetadata {
pub fn new(start: u64, bytes_len: u32, values_len: u32) -> Self {
Self {
start,
bytes_len,
values_len,
}
}
}
crates/brk_vecs/src/variants/compressed/compressed_pages_meta.rs
-118
View File
@@ -1,118 +0,0 @@
use std::{
fs::{self, OpenOptions},
io::{self, Seek, SeekFrom, Write},
path::{Path, PathBuf},
};
use rayon::prelude::*;
use zerocopy::{IntoBytes, TryFromBytes};
use crate::Result;
use super::{CompressedPageMetadata, UnsafeSlice};
#[derive(Debug, Clone)]
pub struct CompressedPagesMetadata {
vec: Vec<CompressedPageMetadata>,
change_at: Option<usize>,
path: PathBuf,
}
impl CompressedPagesMetadata {
const PAGE_SIZE: usize = size_of::<CompressedPageMetadata>();
pub fn read(path: &Path) -> Result<CompressedPagesMetadata> {
let this = Self {
vec: fs::read(path)
.unwrap_or_default()
.chunks(Self::PAGE_SIZE)
.map(|bytes| {
if bytes.len() != Self::PAGE_SIZE {
panic!()
}
CompressedPageMetadata::try_read_from_bytes(bytes).unwrap()
})
.collect::<Vec<_>>(),
path: path.to_owned(),
change_at: None,
};
Ok(this)
}
pub fn write(&mut self) -> io::Result<()> {
if self.change_at.is_none() {
return Ok(());
}
let change_at = self.change_at.take().unwrap();
let len = (self.vec.len() - change_at) * Self::PAGE_SIZE;
let mut bytes: Vec<u8> = vec![0; len];
let unsafe_bytes = UnsafeSlice::new(&mut bytes);
self.vec[change_at..]
.par_iter()
.enumerate()
.for_each(|(i, v)| unsafe_bytes.copy_slice(i * Self::PAGE_SIZE, v.as_bytes()));
let mut file = OpenOptions::new()
.read(true)
.create(true)
.truncate(false)
.append(true)
.open(&self.path)?;
file.set_len((change_at * Self::PAGE_SIZE) as u64)?;
file.seek(SeekFrom::End(0))?;
file.write_all(&bytes)?;
Ok(())
}
pub fn len(&self) -> usize {
self.vec.len()
}
pub fn is_empty(&self) -> bool {
self.len() == 0
}
pub fn get(&self, page_index: usize) -> Option<&CompressedPageMetadata> {
self.vec.get(page_index)
}
pub fn last(&self) -> Option<&CompressedPageMetadata> {
self.vec.last()
}
pub fn pop(&mut self) -> Option<CompressedPageMetadata> {
self.vec.pop()
}
pub fn push(&mut self, page_index: usize, page: CompressedPageMetadata) {
if page_index != self.vec.len() {
panic!();
}
self.set_changed_at(page_index);
self.vec.push(page);
}
fn set_changed_at(&mut self, page_index: usize) {
if self.change_at.is_none_or(|pi| pi > page_index) {
self.change_at.replace(page_index);
}
}
pub fn truncate(&mut self, page_index: usize) -> Option<CompressedPageMetadata> {
let page = self.get(page_index).cloned();
self.vec.truncate(page_index);
self.set_changed_at(page_index);
page
}
}
crates/brk_vecs/src/variants/compressed/mod.rs
+127 -146
View File
@@ -1,37 +1,35 @@
use std::{
borrow::Cow,
collections::{BTreeMap, BTreeSet},
fs, mem,
mem,
sync::Arc,
};
use memmap2::Mmap;
use parking_lot::{RwLock, RwLockReadGuard};
use pco::data_types::Number;
use rayon::prelude::*;
use crate::{
AnyCollectableVec, AnyIterableVec, AnyStoredVec, AnyVec, AsInnerSlice, BaseVecIterator,
BoxedVecIterator, CollectableVec, Error, File, FromInnerSlice, GenericStoredVec, HEADER_OFFSET,
Header, RawVec, Reader, Result, StoredCompressed, StoredIndex, UnsafeSlice, Version,
BoxedVecIterator, CollectableVec, Error, File, Format, FromInnerSlice, GenericStoredVec,
HEADER_OFFSET, Header, RawVec, Reader, Result, StoredCompressed, StoredIndex, Version,
};
mod compressed_page_meta;
mod compressed_pages_meta;
mod page;
mod pages;
use compressed_page_meta::*;
use compressed_pages_meta::*;
use page::*;
use pages::*;
const ONE_KIB: usize = 1024;
const MAX_PAGE_SIZE: usize = 16 * ONE_KIB;
const PCO_COMPRESSION_LEVEL: usize = 4;
const VERSION: Version = Version::TWO;
const VERSION: Version = Version::ONE;
#[derive(Debug)]
pub struct CompressedVec<I, T> {
inner: RawVec<I, T>,
pages_meta: Arc<RwLock<CompressedPagesMetadata>>,
pages: Arc<RwLock<Pages>>,
}
impl<I, T> CompressedVec<I, T>
@@ -39,8 +37,7 @@ where
I: StoredIndex,
T: StoredCompressed,
{
pub const PER_PAGE: usize = MAX_PAGE_SIZE / Self::SIZE_OF_T;
pub const PAGE_SIZE: usize = Self::PER_PAGE * Self::SIZE_OF_T;
const PER_PAGE: usize = MAX_PAGE_SIZE / Self::SIZE_OF_T;
/// Same as import but will reset the vec under certain errors, so be careful !
pub fn forced_import(file: &Arc<File>, name: &str, mut version: Version) -> Result<Self> {
@@ -51,73 +48,50 @@ where
| Err(Error::WrongEndian)
| Err(Error::WrongLength)
| Err(Error::DifferentVersion { .. }) => {
todo!();
// let path = Self::path_(file, name);
// fs::remove_file(path)?;
// Self::import(file, name, version)
let _ = file.remove_region(Self::vec_region_name_(name).into());
let _ = file.remove_region(Self::holes_region_name_(name).into());
let _ = file.remove_region(Self::pages_region_name_(name).into());
Self::import(file, name, version)
}
_ => res,
}
}
#[allow(unreachable_code, unused_variables)]
pub fn import(file: &Arc<File>, name: &str, version: Version) -> Result<Self> {
// let mut inner = RawVec::import(file, name, version)?;
let inner = RawVec::import_(file, name, version, Format::Compressed)?;
todo!();
let pages = Pages::import(file, &Self::pages_region_name_(name))?;
// let pages_meta = {
// let path = inner
// .folder()
// .join(format!("{}-pages-meta", I::to_string()));
// if inner.is_empty() {
// let _ = fs::remove_file(&path);
// }
// CompressedPagesMetadata::read(&path)?
// };
// inner.set_stored_len(if let Some(last) = pages_meta.last() {
// (pages_meta.len() - 1) * Self::PER_PAGE + last.values_len as usize
// } else {
// 0
// });
// Ok(Self {
// inner,
// pages_meta: Arc::new(RwLock::new(pages_meta)),
// })
Ok(Self {
inner,
pages: Arc::new(RwLock::new(pages)),
})
}
fn decode_page(&self, page_index: usize, reader: &Reader) -> Result<Vec<T>> {
Self::decode_page_(
self.stored_len(),
page_index,
reader,
&self.pages_meta.read(),
)
Self::decode_page_(self.stored_len(), page_index, reader, &self.pages.read())
}
fn decode_page_(
stored_len: usize,
page_index: usize,
reader: &Reader,
compressed_pages_meta: &CompressedPagesMetadata,
pages: &Pages,
) -> Result<Vec<T>> {
if Self::page_index_to_index(page_index) >= stored_len {
return Err(Error::IndexTooHigh);
} else if compressed_pages_meta.len() <= page_index {
} else if page_index >= pages.len() {
return Err(Error::ExpectVecToHaveIndex);
}
let page = compressed_pages_meta.get(page_index).unwrap();
let len = page.bytes_len as usize;
let offset = page.start as usize;
let page = pages.get(page_index).unwrap();
let len = page.bytes as u64;
let offset = page.start;
let slice: &[u8] = reader.read(offset as u64, (offset + len) as u64);
let slice = reader.read(offset, len);
let vec: Vec<T::NumberType> = pco::standalone::simple_decompress(slice)?;
let vec: Vec<T> = T::from_inner_slice(vec);
let vec = T::from_inner_slice(vec);
Ok(vec)
}
@@ -156,13 +130,17 @@ where
iter.set_(i);
iter
}
fn pages_region_name_(name: &str) -> String {
format!("{}_pages", Self::vec_region_name_(name))
}
}
impl<I, T> Clone for CompressedVec<I, T> {
fn clone(&self) -> Self {
Self {
inner: self.inner.clone(),
pages_meta: self.pages_meta.clone(),
pages: self.pages.clone(),
}
}
}
@@ -221,92 +199,87 @@ where
#[inline]
fn stored_len(&self) -> usize {
self.inner.stored_len()
self.pages.read().stored_len(Self::PER_PAGE)
}
fn flush(&mut self) -> Result<()> {
todo!();
self.inner.write_header_if_needed()?;
// let file_opt = self.inner.write_header_if_needed()?;
let pushed_len = self.pushed_len();
// let pushed_len = self.pushed_len();
let has_new_data = pushed_len != 0;
// if pushed_len == 0 {
// return Ok(());
// }
if !has_new_data {
return Ok(());
}
// let stored_len = self.stored_len();
let stored_len = self.stored_len();
// let mut file = file_opt.unwrap_or(self.open_file()?);
let mut pages = self.pages.write();
// let mut pages_meta = self.pages_meta.read();
let mut starting_page_index = pages.len();
let mut values = vec![];
let mut truncate_at = None;
// let mut starting_page_index = pages_meta.len();
// let mut values = vec![];
// let mut truncate_at = None;
if stored_len % Self::PER_PAGE != 0 {
if pages.is_empty() {
unreachable!()
}
// if self.stored_len() % Self::PER_PAGE != 0 {
// if pages_meta.is_empty() {
// unreachable!()
// }
let last_page_index = pages.len() - 1;
// let last_page_index = pages_meta.len() - 1;
let reader = self.create_reader();
// let mmap = unsafe { Mmap::map(&file)? };
values = Self::decode_page_(stored_len, last_page_index, &reader, &pages)
.inspect_err(|_| {
dbg!(last_page_index, &pages);
})
.unwrap();
// values = Self::decode_page_(stored_len, last_page_index, &mmap, &pages_meta)
// .inspect_err(|_| {
// dbg!(last_page_index, &pages_meta);
// })
// .unwrap();
truncate_at.replace(pages.pop().unwrap().start);
starting_page_index = last_page_index;
}
// truncate_at.replace(pages_meta.pop().unwrap().start);
// starting_page_index = last_page_index;
// }
let compressed = values
.into_par_iter()
.chain(mem::take(self.inner.mut_pushed()).into_par_iter())
.chunks(Self::PER_PAGE)
.map(|chunk| (Self::compress_page(chunk.as_slice()), chunk.len()))
.collect::<Vec<_>>();
// let compressed = values
// .into_par_iter()
// .chain(mem::take(self.mut_pushed()).into_par_iter())
// .chunks(Self::PER_PAGE)
// .map(|chunk| (Self::compress_page(chunk.as_ref()), chunk.len()))
// .collect::<Vec<_>>();
compressed.iter().enumerate().for_each(|(i, (bytes, len))| {
let page_index = starting_page_index + i;
// 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.get(page_index - 1).unwrap();
prev.start + prev.bytes as u64
} else {
0
};
// 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 offsetted_start = start + HEADER_OFFSET as u64;
let page = Page::new(
start + HEADER_OFFSET as u64,
bytes.len() as u32,
*len as u32,
);
// let bytes_len = compressed_bytes.len() as u32;
// let values_len = *values_len as u32;
pages.checked_push(page_index, page);
});
// let page = CompressedPageMetadata::new(offsetted_start, bytes_len, values_len);
let buf = compressed
.into_iter()
.flat_map(|(v, _)| v)
.collect::<Vec<_>>();
// pages_meta.push(page_index, page);
// });
let file = self.file();
// let buf = compressed
// .into_iter()
// .flat_map(|(v, _)| v)
// .collect::<Vec<_>>();
if let Some(truncate_at) = truncate_at {
file.truncate_region(self.region_index().into(), truncate_at)?;
}
// pages_meta.write()?;
file.write_all_to_region(self.region_index().into(), &buf)?;
// if let Some(truncate_at) = truncate_at {
// self.file_set_len(&mut file, truncate_at)?;
// }
// self.file_write_all(&mut file, &buf)?;
// self.pages_meta.store(Arc::new(pages_meta));
pages.flush(file)?;
Ok(())
}
@@ -321,11 +294,10 @@ where
fn read_(&self, index: usize, reader: &Reader) -> Result<T> {
let page_index = Self::index_to_page_index(index);
let decoded_index = index % Self::PER_PAGE;
Ok(unsafe {
self.decode_page(page_index, reader)?
*self
.decode_page(page_index, reader)?
.get_unchecked(decoded_index)
.clone()
})
}
@@ -355,10 +327,8 @@ where
}
fn reset(&mut self) -> Result<()> {
// let mut pages_meta = (**self.pages_meta.load()).clone();
// pages_meta.truncate(0);
// pages_meta.write()?;
// self.pages_meta.store(Arc::new(pages_meta));
let file = self.file();
self.pages.write().reset(file)?;
self.reset_()
}
@@ -374,36 +344,47 @@ where
return Ok(());
}
let mut pages_meta = self.pages_meta.write();
let stored_len = self.stored_len();
let mut pages = self.pages.write();
let last_page_index = pages.len() - 1;
let page_index = Self::index_to_page_index(index);
let reader = self.create_static_reader();
let values = self.decode_page(page_index, &reader)?;
drop(reader);
let values = Self::decode_page_(
stored_len,
last_page_index,
&self.create_static_reader(),
&pages,
)?;
let mut buf = vec![];
let mut page = pages_meta.truncate(page_index).unwrap();
let len = page.start;
let mut page = pages.truncate(page_index).unwrap();
let decoded_index = index % Self::PER_PAGE;
let from = page.start;
if decoded_index != 0 {
let chunk = &values[..decoded_index];
buf = Self::compress_page(chunk);
page.values_len = chunk.len() as u32;
page.bytes_len = buf.len() as u32;
page.values = chunk.len() as u32;
page.bytes = buf.len() as u32;
pages_meta.push(page_index, page);
pages.checked_push(page_index, page);
}
pages_meta.write()?;
let file = self.file();
// self.file_truncate_and_write_all(&mut file, len, &buf)?;
pages.flush(file)?;
file.truncate_region(self.region_index().into(), from)?;
file.write_all_to_region(self.region_index().into(), &buf)?;
Ok(())
}
@@ -413,8 +394,8 @@ where
pub struct CompressedVecIterator<'a, I, T> {
vec: &'a CompressedVec<I, T>,
reader: Reader<'a>,
decoded_page: Option<(usize, Vec<T>)>,
pages_meta: RwLockReadGuard<'a, CompressedPagesMetadata>,
decoded: Option<(usize, Vec<T>)>,
pages: RwLockReadGuard<'a, Pages>,
stored_len: usize,
index: usize,
}
@@ -472,23 +453,23 @@ where
} else {
let page_index = i / Self::PER_PAGE;
if self.decoded_page.as_ref().is_none_or(|b| b.0 != page_index) {
if self.decoded.as_ref().is_none_or(|b| b.0 != page_index) {
let values = CompressedVec::<I, T>::decode_page_(
stored_len,
page_index,
&self.reader,
&self.pages_meta,
&self.pages,
)
.unwrap();
self.decoded_page.replace((page_index, values));
self.decoded.replace((page_index, values));
}
self.decoded_page
self.decoded
.as_ref()
.unwrap()
.1
.get(i % Self::PER_PAGE)
.map(|v| (I::from(i), Cow::Owned(v.clone())))
.map(|v| (I::from(i), Cow::Owned(*v)))
};
self.index += 1;
@@ -506,14 +487,14 @@ where
type IntoIter = CompressedVecIterator<'a, I, T>;
fn into_iter(self) -> Self::IntoIter {
let pages_meta = self.pages_meta.read();
let pages = self.pages.read();
let stored_len = self.stored_len();
CompressedVecIterator {
vec: self,
reader: self.create_static_reader(),
decoded_page: None,
pages_meta,
decoded: None,
pages,
index: 0,
stored_len,
}
crates/brk_vecs/src/variants/compressed/page.rs
+19
View File
@@ -0,0 +1,19 @@
use zerocopy_derive::{FromBytes, Immutable, IntoBytes, KnownLayout};
#[derive(Debug, Clone, IntoBytes, Immutable, FromBytes, KnownLayout)]
#[repr(C)]
pub struct Page {
pub start: u64,
pub bytes: u32,
pub values: u32,
}
impl Page {
pub fn new(start: u64, bytes: u32, values: u32) -> Self {
Self {
start,
bytes,
values,
}
}
}
crates/brk_vecs/src/variants/compressed/pages.rs
+122
View File
@@ -0,0 +1,122 @@
use std::sync::Arc;
use parking_lot::RwLock;
use zerocopy::{FromBytes, IntoBytes};
use crate::{
File, Result,
file::{Region, RegionReader},
};
use super::Page;
#[derive(Debug, Clone)]
pub struct Pages {
_region: Arc<RwLock<Region>>,
region_index: usize,
vec: Vec<Page>,
change_at: Option<usize>,
}
impl Pages {
const SIZE_OF_PAGE: usize = size_of::<Page>();
pub fn import(file: &File, name: &str) -> Result<Self> {
let (region_index, _region) = file.create_region_if_needed(name)?;
let vec = _region
.read()
.create_reader(file)
.read_all()
.chunks(Self::SIZE_OF_PAGE)
.map(|b| Page::read_from_bytes(b).map_err(|e| e.into()))
.collect::<Result<_>>()?;
Ok(Self {
_region,
region_index,
vec,
change_at: None,
})
}
pub fn flush(&mut self, file: &File) -> Result<()> {
if self.change_at.is_none() {
return Ok(());
}
let change_at = self.change_at.take().unwrap();
let at = (change_at * Self::SIZE_OF_PAGE) as u64;
file.truncate_region(self.region_index.into(), at)?;
file.write_all_to_region_at(
self.region_index.into(),
self.vec[change_at..].as_bytes(),
at,
)?;
Ok(())
}
pub fn len(&self) -> usize {
self.vec.len()
}
pub fn is_empty(&self) -> bool {
self.len() == 0
}
pub fn get(&self, page_index: usize) -> Option<&Page> {
self.vec.get(page_index)
}
pub fn last(&self) -> Option<&Page> {
self.vec.last()
}
pub fn pop(&mut self) -> Option<Page> {
let popped = self.vec.pop();
if popped.is_some() {
self.set_changed_at(self.vec.len());
}
popped
}
pub fn checked_push(&mut self, page_index: usize, page: Page) {
if page_index != self.vec.len() {
panic!();
}
self.set_changed_at(page_index);
self.vec.push(page);
}
fn set_changed_at(&mut self, page_index: usize) {
if self.change_at.is_none_or(|pi| pi > page_index) {
self.change_at.replace(page_index);
}
}
pub fn reset(&mut self, file: &File) -> Result<()> {
self.truncate(0);
self.flush(file)
}
pub fn truncate(&mut self, page_index: usize) -> Option<Page> {
let page = self.get(page_index).cloned();
self.vec.truncate(page_index);
self.set_changed_at(page_index);
page
}
pub fn stored_len(&self, per_page: usize) -> usize {
if let Some(last) = self.last() {
(self.len() - 1) * per_page + last.values as usize
} else {
0
}
}
}
crates/brk_vecs/src/variants/computed/computation.rs
+1 -1
View File
@@ -1,4 +1,4 @@
use serde::{Deserialize, Serialize};
use serde_derive::{Deserialize, Serialize};
#[derive(Default, Debug, PartialEq, PartialOrd, Ord, Eq, Clone, Copy, Serialize, Deserialize)]
pub enum Computation {
crates/brk_vecs/src/variants/raw/header.rs
+5
View File
@@ -51,6 +51,11 @@ impl Header {
self.inner.write().stamp = stamp;
}
pub fn update_format(&mut self, format: Format) {
self.modified = true;
self.inner.write().compressed = ZeroCopyBool::from(format);
}
pub fn update_computed_version(&mut self, computed_version: Version) {
self.modified = true;
self.inner.write().computed_version = computed_version;
crates/brk_vecs/src/variants/raw/mod.rs
+14 -36
View File
@@ -7,8 +7,7 @@ use std::{
};
use parking_lot::RwLock;
use rayon::prelude::*;
use zerocopy::FromBytes;
use zerocopy::{FromBytes, IntoBytes};
use crate::{
AnyCollectableVec, AnyIterableVec, AnyStoredVec, AnyVec, BaseVecIterator, BoxedVecIterator,
@@ -19,10 +18,8 @@ use crate::{
use super::Format;
mod header;
mod unsafe_slice;
pub use header::*;
pub use unsafe_slice::*;
const VERSION: Version = Version::ONE;
@@ -64,25 +61,24 @@ where
}
pub fn import(file: &Arc<File>, name: &str, version: Version) -> Result<Self> {
Self::import_(file, name, version, Format::Raw)
}
pub fn import_(file: &Arc<File>, name: &str, version: Version, format: Format) -> Result<Self> {
let (region_index, region) = file.create_region_if_needed(&Self::vec_region_name_(name))?;
let region_len = region.read().len() as usize;
if region_len > 0
&& (region_len < HEADER_OFFSET || (region_len - HEADER_OFFSET) % Self::SIZE_OF_T != 0)
&& (region_len < HEADER_OFFSET
|| (format.is_raw() && (region_len - HEADER_OFFSET) % Self::SIZE_OF_T != 0))
{
return Err(Error::Str("Region was saved incorrectly"));
}
let header = if region_len == 0 {
Header::create_and_write(file, region_index, version, Format::Raw)?
Header::create_and_write(file, region_index, version, format)?
} else {
Header::import_and_verify(
file,
region_index,
region.read().len(),
version,
Format::Raw,
)?
Header::import_and_verify(file, region_index, region.read().len(), version, format)?
};
let holes = if let Ok(holes) = file.get_region(Self::holes_region_name_(name).into()) {
@@ -200,13 +196,7 @@ where
#[inline]
fn stored_len(&self) -> usize {
(self
.file
.get_region(self.region_index.into())
.unwrap()
.len() as usize
- HEADER_OFFSET)
/ Self::SIZE_OF_T
(self.region.read().len() as usize - HEADER_OFFSET) / Self::SIZE_OF_T
}
fn flush(&mut self) -> Result<()> {
@@ -227,22 +217,10 @@ where
let file = &self.file;
if has_new_data {
let bytes = {
let mut bytes: Vec<u8> = vec![0; pushed_len * Self::SIZE_OF_T];
let unsafe_bytes = UnsafeSlice::new(&mut bytes);
mem::take(&mut self.pushed)
.into_par_iter()
.enumerate()
.for_each(|(i, v)| {
unsafe_bytes.copy_slice(i * Self::SIZE_OF_T, v.as_bytes())
});
bytes
};
file.write_all_to_region(self.region_index.into(), &bytes)?;
file.write_all_to_region(
self.region_index.into(),
mem::take(&mut self.pushed).as_bytes(),
)?;
}
if has_updated_data {
crates/brk_vecs/src/variants/raw/unsafe_slice.rs
-35
View File
@@ -1,35 +0,0 @@
use std::cell::UnsafeCell;
#[derive(Copy, Clone)]
pub struct UnsafeSlice<'a, T>(&'a [UnsafeCell<T>]);
unsafe impl<T: Send + Sync> Send for UnsafeSlice<'_, T> {}
unsafe impl<T: Send + Sync> Sync for UnsafeSlice<'_, T> {}
impl<'a, T> UnsafeSlice<'a, T> {
pub fn new(slice: &'a mut [T]) -> Self {
let ptr = slice as *mut [T] as *const [UnsafeCell<T>];
Self(unsafe { &*ptr })
}
/// SAFETY: It is UB if two threads write to the same index without
/// synchronization.
pub fn write(&self, i: usize, value: T) {
unsafe {
*self.0[i].get() = value;
}
}
/// SAFETY: It is UB
pub fn get(&self, i: usize) -> *mut T {
self.0[i].get()
}
pub fn copy_slice(&self, start: usize, slice: &[T])
where
T: Copy,
{
slice.iter().enumerate().for_each(|(i, v)| {
self.write(start + i, *v);
});
}
}
crates/brk_vecs/src/variants/stored/format.rs
+1 -1
View File
@@ -1,6 +1,6 @@
use std::{fs, io, path::Path};
use serde::{Deserialize, Serialize};
use serde_derive::{Deserialize, Serialize};
use crate::{Error, Result};