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

vec: rework part 3

Browse Source
This commit is contained in:
nym21
2025-04-10 01:11:52 +02:00
parent 41cf0225e3
commit 0dd7e9359e
12 changed files with 84 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
crates/brk_cli/src/run.rs
View File

@@ -272,9 +272,10 @@ impl RunConfig {
}
fn read(path: &Path) -> Self {
fs::read_to_string(path).map_or_else(RunConfig::default, |contents| {
toml::from_str(&contents).unwrap_or_default()
})
fs::read_to_string(path).map_or_else(
|_| RunConfig::default(),
|contents| toml::from_str(&contents).unwrap_or_default(),
)
}
fn write(&self, path: &Path) -> std::io::Result<()> {

23
crates/brk_computer/src/storage/vecs/base.rs
View File

@@ -2,7 +2,6 @@ use core::error;
use std::{
cmp::Ordering,
fmt::Debug,
io,
ops::{Add, Sub},
path::{Path, PathBuf},
};
@@ -73,7 +72,7 @@ where
}
if self.vec.pushed_len() * Self::SIZE_OF >= MAX_CACHE_SIZE {
Ok(self.safe_flush(exit)?)
self.safe_flush(exit)
} else {
Ok(())
}
@@ -117,9 +116,9 @@ where
self.vec.get(index)
}
// pub fn collect_range(&self, from: Option<i64>, to: Option<i64>) -> Result<Vec<T>> {
// self.vec.collect_range(from, to)
// }
pub fn collect_range(&self, from: Option<i64>, to: Option<i64>) -> Result<Vec<T>> {
self.vec.collect_range(from, to)
}
#[inline]
fn path_computed_version(&self) -> PathBuf {
@@ -157,7 +156,7 @@ where
self.forced_push_at(i, v, exit)
})?;
Ok(self.safe_flush(exit)?)
self.safe_flush(exit)
}
pub fn compute_inverse_more_to_less(
@@ -187,7 +186,7 @@ where
}
})?;
Ok(self.safe_flush(exit)?)
self.safe_flush(exit)
}
pub fn compute_inverse_less_to_more(
@@ -213,7 +212,7 @@ where
.try_for_each(|index| self.forced_push_at(I::from(index), value, exit))
})?;
Ok(self.safe_flush(exit)?)
self.safe_flush(exit)
}
pub fn compute_last_index_from_first(
@@ -248,7 +247,7 @@ where
)?;
}
Ok(self.safe_flush(exit)?)
self.safe_flush(exit)
}
pub fn compute_count_from_indexes<T2>(
@@ -276,7 +275,7 @@ where
self.forced_push_at(i, count.into(), exit)
})?;
Ok(self.safe_flush(exit)?)
self.safe_flush(exit)
}
pub fn compute_is_first_ordered<A>(
@@ -304,7 +303,7 @@ where
)
})?;
Ok(self.safe_flush(exit)?)
self.safe_flush(exit)
}
pub fn compute_sum_from_indexes<T2>(
@@ -330,7 +329,7 @@ where
self.forced_push_at(index, count.into(), exit)
})?;
Ok(self.safe_flush(exit)?)
self.safe_flush(exit)
}
}

2
crates/brk_computer/src/storage/vecs/blocks.rs
View File

@@ -4,7 +4,7 @@ use brk_core::{CheckedSub, StoredU32, StoredU64, StoredUsize, Timestamp, Weight}
use brk_exit::Exit;
use brk_indexer::Indexer;
use brk_parser::bitcoin;
use brk_vec::{Compressed, DynamicVec, Version};
use brk_vec::{Compressed, Version};
use super::{
Indexes,

14
crates/brk_computer/src/storage/vecs/grouped/builder.rs
View File

@@ -164,7 +164,6 @@ where
let index = self.starting_index(max_from);
first_indexes.iter_from(index, |(i, first_index, ..)| {
let first_index = first_index;
let last_index = *last_indexes.get(i).unwrap().unwrap();
if let Some(first) = self.first.as_mut() {
@@ -298,8 +297,7 @@ where
let index = self.starting_index(max_from);
first_indexes.iter_from(index, |(i, first_index)| {
let first_index = *first_index;
first_indexes.iter_from(index, |(i, first_index, ..)| {
let last_index = *last_indexes.get(i).unwrap().unwrap();
if let Some(first) = self.first.as_mut() {
@@ -321,8 +319,8 @@ where
.unwrap()
.get(last_index)
.unwrap()
.cloned()
.unwrap();
.unwrap()
.into_inner();
last.forced_push_at(index, v, exit)?;
}
@@ -391,7 +389,11 @@ where
let prev = i.to_usize().unwrap().checked_sub(1).map_or(
T::from(0_usize),
|prev_i| {
total_vec.get(I::from(prev_i)).unwrap().unwrap().to_owned()
total_vec
.get(I::from(prev_i))
.unwrap()
.unwrap()
.into_inner()
},
);
total_vec.forced_push_at(i, prev + sum, exit)?;

2
crates/brk_computer/src/storage/vecs/marketprice.rs
View File

@@ -7,7 +7,7 @@ use brk_core::{
use brk_exit::Exit;
use brk_fetcher::Fetcher;
use brk_indexer::Indexer;
use brk_vec::{Compressed, DynamicVec, Version};
use brk_vec::{Compressed, Version};
use super::{
ComputedVec, Indexes,

1
crates/brk_indexer/src/vecs/base.rs
View File

@@ -1,7 +1,6 @@
use std::{
cmp::Ordering,
fmt::Debug,
io,
path::{Path, PathBuf},
};

2
crates/brk_query/src/lib.rs
View File

@@ -94,7 +94,7 @@ impl<'a> Query<'a> {
let mut values = vecs
.iter()
.map(|(_, vec)| -> brk_vec::Result<Vec<serde_json::Value>> {
vec.collect_range_values(from, to)
vec.collect_range_serde_json(from, to)
})
.collect::<brk_vec::Result<Vec<_>>>()?;

8
crates/brk_vec/src/enums/error.rs
View File

@@ -23,6 +23,7 @@ pub enum Error {
UnsupportedUnflushedState,
RangeFromAfterTo(usize, usize),
DifferentCompressionMode,
ToSerdeJsonValueError(serde_json::Error),
}
impl From<io::Error> for Error {
@@ -43,6 +44,12 @@ impl<A, B> From<zerocopy::error::SizeError<A, B>> for Error {
}
}
impl From<serde_json::Error> for Error {
fn from(error: serde_json::Error) -> Self {
Self::ToSerdeJsonValueError(error)
}
}
impl fmt::Display for Error {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
match self {
@@ -70,6 +77,7 @@ impl fmt::Display for Error {
Error::RangeFromAfterTo(from, to) => write!(f, "Range, from {from} is after to {to}"),
Error::DifferentCompressionMode => write!(f, "Different compression mode chosen"),
Error::EmptyVec => write!(f, "The Vec is empty, maybe wait for a bit"),
Error::ToSerdeJsonValueError(error) => Debug::fmt(&error, f),
}
}
}

19
crates/brk_vec/src/lib.rs
View File

@@ -14,7 +14,7 @@ use std::{
};
use arc_swap::{ArcSwap, Guard};
use axum::{Json, response::Response};
use axum::response::Response;
pub use enums::*;
use memmap2::Mmap;
pub use structs::*;
@@ -109,7 +109,7 @@ where
}
}
fn collect_range(&self, from: Option<i64>, to: Option<i64>) -> Result<Json<Vec<T>>> {
fn collect_range(&self, from: Option<i64>, to: Option<i64>) -> Result<Vec<Self::T>> {
match self {
StoredVec::Raw(v) => v.collect_range(from, to),
StoredVec::Compressed(v) => v.collect_range(from, to),
@@ -176,6 +176,11 @@ pub trait AnyStoredVec: Send + Sync {
fn len(&self) -> usize;
fn is_empty(&self) -> bool;
fn flush(&mut self) -> Result<()>;
fn collect_range_serde_json(
&self,
from: Option<i64>,
to: Option<i64>,
) -> Result<Vec<serde_json::Value>>;
fn collect_range_response(&self, from: Option<i64>, to: Option<i64>) -> Result<Response>;
fn path_vec(&self) -> PathBuf;
}
@@ -204,6 +209,16 @@ where
GenericVec::flush(self)
}
#[inline]
fn collect_range_serde_json(
&self,
from: Option<i64>,
to: Option<i64>,
) -> Result<Vec<serde_json::Value>> {
GenericVec::collect_range_serde_json(self, from, to)
}
#[inline]
fn collect_range_response(&self, from: Option<i64>, to: Option<i64>) -> Result<Response> {
GenericVec::collect_range_response(self, from, to)
}

24
crates/brk_vec/src/traits/generic.rs
View File

@@ -11,6 +11,7 @@ use axum::{
response::{IntoResponse, Response},
};
use memmap2::Mmap;
use serde_json::Value;
use crate::{Error, Result, Version};
@@ -159,10 +160,28 @@ where
fn truncate_if_needed(&mut self, index: Self::I) -> Result<()>;
fn collect_range(&self, from: Option<i64>, to: Option<i64>) -> Result<Json<Vec<Self::T>>>;
fn collect_range(&self, from: Option<i64>, to: Option<i64>) -> Result<Vec<Self::T>>;
#[inline]
fn collect_range_axum_json(
&self,
from: Option<i64>,
to: Option<i64>,
) -> Result<Json<Vec<Self::T>>> {
Ok(Json(self.collect_range(from, to)?))
}
#[inline]
fn collect_range_serde_json(&self, from: Option<i64>, to: Option<i64>) -> Result<Vec<Value>> {
self.collect_range(from, to)?
.into_iter()
.map(|v| serde_json::to_value(v).map_err(Error::from))
.collect::<Result<Vec<_>>>()
}
#[inline]
fn collect_range_response(&self, from: Option<i64>, to: Option<i64>) -> Result<Response> {
Ok(self.collect_range(from, to)?.into_response())
Ok(self.collect_range_axum_json(from, to)?.into_response())
}
fn path(&self) -> &Path;
@@ -181,6 +200,7 @@ where
path.join("version")
}
#[inline]
fn file_name(&self) -> String {
self.path()
.file_name()

3
crates/brk_vec/src/variants/compressed.rs
View File

@@ -5,7 +5,6 @@ use std::{
};
use arc_swap::{ArcSwap, Guard};
use axum::Json;
use memmap2::Mmap;
use rayon::prelude::*;
use zstd::DEFAULT_COMPRESSION_LEVEL;
@@ -232,7 +231,7 @@ where
todo!()
}
fn collect_range(&self, from: Option<i64>, to: Option<i64>) -> Result<Json<Vec<T>>> {
fn collect_range(&self, from: Option<i64>, to: Option<i64>) -> Result<Vec<Self::T>> {
todo!()
}

21
crates/brk_vec/src/variants/raw.rs
View File

@@ -7,7 +7,6 @@ use std::{
};
use arc_swap::{ArcSwap, Guard};
use axum::Json;
use memmap2::Mmap;
use rayon::prelude::*;
@@ -116,8 +115,8 @@ where
impl<I, T> GenericVec<I, T> for RawVec<I, T>
where
I: StoredIndex + Send + Sync,
T: StoredType + Send + Sync,
I: StoredIndex,
T: StoredType,
{
fn iter_from<F>(&mut self, index: I, mut f: F) -> Result<()>
where
@@ -131,6 +130,8 @@ where
let start = index.to_usize()? * Self::SIZE_OF_T;
dbg!(self.path());
guard[start..]
.chunks(Self::SIZE_OF_T)
.enumerate()
@@ -188,24 +189,22 @@ where
Ok(())
}
fn collect_range(&self, from: Option<i64>, to: Option<i64>) -> Result<Json<Vec<T>>> {
fn collect_range(&self, from: Option<i64>, to: Option<i64>) -> Result<Vec<T>> {
let guard = self.mmap.load();
let len = guard.len() / Self::SIZE_OF_T;
if len == 0 {
return Ok(Json(vec![]));
return Ok(vec![]);
}
let from = from.map_or(0, |i| Self::fix_i64(i, len, true));
let to = to.map_or(len, |i| Self::fix_i64(i, len, false));
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<_>>(),
))
Ok(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]