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

store: faster everything

Browse Source
This commit is contained in:
nym21
2025-10-20 11:33:48 +02:00
parent e9f6295014
commit 33a92cfad4
7 changed files with 62 additions and 131 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/brk_indexer/src/stores_v2.rs
+15 -12
View File
@@ -7,7 +7,7 @@ use brk_structs::{
AddressBytes, AddressBytesHash, BlockHashPrefix, Height, StoredString, TxIndex, TxOutIndex, AddressBytes, AddressBytesHash, BlockHashPrefix, Height, StoredString, TxIndex, TxOutIndex,
TxidPrefix, TypeIndex, TypeIndexAndOutPoint, TypeIndexAndTxIndex, Unit, Version, TxidPrefix, TypeIndex, TypeIndexAndOutPoint, TypeIndexAndTxIndex, Unit, Version,
}; };
use fjall2::{PersistMode, TransactionalKeyspace}; use fjall2::{Keyspace, PersistMode};
use rayon::prelude::*; use rayon::prelude::*;
use vecdb::{AnyVec, StoredIndex, VecIterator}; use vecdb::{AnyVec, StoredIndex, VecIterator};
@@ -17,7 +17,7 @@ use super::Vecs;
#[derive(Clone)] #[derive(Clone)]
pub struct Stores { pub struct Stores {
pub keyspace: TransactionalKeyspace, pub keyspace: Keyspace,
pub addressbyteshash_to_typeindex: Store<AddressBytesHash, TypeIndex>, pub addressbyteshash_to_typeindex: Store<AddressBytesHash, TypeIndex>,
pub blockhashprefix_to_height: Store<BlockHashPrefix, Height>, pub blockhashprefix_to_height: Store<BlockHashPrefix, Height>,
@@ -361,8 +361,9 @@ impl Stores {
} }
} }
} else { } else {
self.blockhashprefix_to_height.reset()?; unreachable!();
self.addressbyteshash_to_typeindex.reset()?; // self.blockhashprefix_to_height.reset()?;
// self.addressbyteshash_to_typeindex.reset()?;
} }
if starting_indexes.txindex != TxIndex::ZERO { if starting_indexes.txindex != TxIndex::ZERO {
@@ -384,11 +385,12 @@ impl Stores {
} }
}); });
} else { } else {
self.txidprefix_to_txindex.reset()?; unreachable!();
// self.txidprefix_to_txindex.reset()?;
} }
if starting_indexes.txoutindex != TxOutIndex::ZERO { if starting_indexes.txoutindex != TxOutIndex::ZERO {
// todo!(); todo!();
// let mut txoutindex_to_typeindex_iter = vecs.txoutindex_to_typeindex.into_iter(); // let mut txoutindex_to_typeindex_iter = vecs.txoutindex_to_typeindex.into_iter();
// vecs.txoutindex_to_outputtype // vecs.txoutindex_to_outputtype
// .iter_at(starting_indexes.txoutindex) // .iter_at(starting_indexes.txoutindex)
@@ -404,12 +406,13 @@ impl Stores {
// .remove(TypeIndexAndTxIndex::from((typeindex, txoutindex))); // .remove(TypeIndexAndTxIndex::from((typeindex, txoutindex)));
// }); // });
} else { } else {
self.addresstype_to_typeindex_and_txindex unreachable!();
.iter_mut() // self.addresstype_to_typeindex_and_txindex
.try_for_each(|s| s.reset())?; // .iter_mut()
self.addresstype_to_typeindex_and_unspentoutpoint // .try_for_each(|s| s.reset())?;
.iter_mut() // self.addresstype_to_typeindex_and_unspentoutpoint
.try_for_each(|s| s.reset())?; // .iter_mut()
// .try_for_each(|s| s.reset())?;
} }
self.commit(starting_indexes.height.decremented().unwrap_or_default())?; self.commit(starting_indexes.height.decremented().unwrap_or_default())?;
crates/brk_store/examples/main.rs
+2 -2
View File
@@ -1,9 +1,9 @@
use std::path::Path; // use std::path::Path;
use brk_error::Result; use brk_error::Result;
fn main() -> Result<()> { fn main() -> Result<()> {
let p = Path::new("./examples/_fjall"); // let p = Path::new("./examples/_fjall");
// let _keyspace = brk_store::open_keyspace(p)?; // let _keyspace = brk_store::open_keyspace(p)?;
crates/brk_store/src/any.rs
+1 -1
View File
@@ -4,7 +4,7 @@ use brk_structs::{Height, Version};
pub trait AnyStore: Send + Sync { pub trait AnyStore: Send + Sync {
fn commit(&mut self, height: Height) -> Result<()>; fn commit(&mut self, height: Height) -> Result<()>;
fn persist(&self) -> Result<()>; fn persist(&self) -> Result<()>;
fn reset(&mut self) -> Result<()>; // fn reset(&mut self) -> Result<()>;
fn name(&self) -> &'static str; fn name(&self) -> &'static str;
fn height(&self) -> Option<Height>; fn height(&self) -> Option<Height>;
fn has(&self, height: Height) -> bool; fn has(&self, height: Height) -> bool;
crates/brk_store/src/lib.rs
+2 -2
View File
@@ -2,8 +2,8 @@
mod any; mod any;
mod v2; mod v2;
mod v3; // mod v3;
pub use any::*; pub use any::*;
pub use v2::*; pub use v2::*;
pub use v3::*; // pub use v3::*;
crates/brk_store/src/v2/meta.rs
+4 -8
View File
@@ -5,7 +5,7 @@ use std::{
use brk_error::Result; use brk_error::Result;
use brk_structs::Version; use brk_structs::Version;
use fjall2::{PersistMode, TransactionalKeyspace, TransactionalPartitionHandle}; use fjall2::{Keyspace, PartitionHandle, PersistMode};
use super::Height; use super::Height;
@@ -18,13 +18,13 @@ pub struct StoreMeta {
impl StoreMeta { impl StoreMeta {
pub fn checked_open<F>( pub fn checked_open<F>(
keyspace: &TransactionalKeyspace, keyspace: &Keyspace,
path: &Path, path: &Path,
version: Version, version: Version,
open_partition_handle: F, open_partition_handle: F,
) -> Result<(Self, TransactionalPartitionHandle)> ) -> Result<(Self, PartitionHandle)>
where where
F: Fn() -> Result<TransactionalPartitionHandle>, F: Fn() -> Result<PartitionHandle>,
{ {
fs::create_dir_all(path)?; fs::create_dir_all(path)?;
@@ -60,10 +60,6 @@ impl StoreMeta {
height.write(&self.path_height()) height.write(&self.path_height())
} }
pub fn reset(&mut self) {
self.height.take();
}
pub fn path(&self) -> &Path { pub fn path(&self) -> &Path {
&self.pathbuf &self.pathbuf
} }
crates/brk_store/src/v2/mod.rs
+26 -94
View File
@@ -1,42 +1,35 @@
use std::{borrow::Cow, fmt::Debug, fs, hash::Hash, mem, path::Path, sync::Arc}; use std::{borrow::Cow, fmt::Debug, fs, hash::Hash, path::Path};
use brk_error::Result; use brk_error::Result;
use brk_structs::{Height, Version}; use brk_structs::{Height, Version};
use byteview6::ByteView; use byteview6::ByteView;
use fjall2::{ use fjall2::{InnerItem, Keyspace, PartitionCreateOptions, PartitionHandle, PersistMode};
PartitionCreateOptions, PersistMode, ReadTransaction, TransactionalKeyspace,
TransactionalPartitionHandle,
};
use parking_lot::RwLock;
use rustc_hash::{FxHashMap, FxHashSet}; use rustc_hash::{FxHashMap, FxHashSet};
use crate::any::AnyStore; use crate::any::AnyStore;
mod meta; mod meta;
use log::info;
use meta::*; use meta::*;
#[derive(Clone)] #[derive(Clone)]
pub struct StoreV2<Key, Value> { pub struct StoreV2<Key, Value> {
meta: StoreMeta, meta: StoreMeta,
name: &'static str, name: &'static str,
keyspace: TransactionalKeyspace, keyspace: Keyspace,
partition: Arc<RwLock<Option<TransactionalPartitionHandle>>>, // no, make this faster
rtx: Arc<RwLock<Option<ReadTransaction>>>, // no ! remove it altogether, reset too
partition: PartitionHandle,
puts: FxHashMap<Key, Value>, puts: FxHashMap<Key, Value>,
dels: FxHashSet<Key>, dels: FxHashSet<Key>,
bloom_filters: Option<bool>,
} }
// const CHECK_COLLISIONS: bool = true;
const MAJOR_FJALL_VERSION: Version = Version::TWO; const MAJOR_FJALL_VERSION: Version = Version::TWO;
pub fn open_keyspace(path: &Path) -> fjall2::Result<TransactionalKeyspace> { pub fn open_keyspace(path: &Path) -> fjall2::Result<Keyspace> {
fjall2::Config::new(path.join("fjall")) fjall2::Config::new(path.join("fjall"))
// .cache_size(1024 * 1024 * 1024) // for tests only
.max_write_buffer_size(32 * 1024 * 1024) .max_write_buffer_size(32 * 1024 * 1024)
.open_transactional() .open()
} }
impl<K, V> StoreV2<K, V> impl<K, V> StoreV2<K, V>
@@ -46,12 +39,11 @@ where
ByteView: From<K> + From<V>, ByteView: From<K> + From<V>,
{ {
fn open_partition_handle( fn open_partition_handle(
keyspace: &TransactionalKeyspace, keyspace: &Keyspace,
name: &str, name: &str,
bloom_filters: Option<bool>, bloom_filters: Option<bool>,
) -> Result<TransactionalPartitionHandle> { ) -> Result<PartitionHandle> {
let mut options = PartitionCreateOptions::default() let mut options = PartitionCreateOptions::default()
// .max_memtable_size(64 * 1024 * 1024) // for tests only
.max_memtable_size(8 * 1024 * 1024) .max_memtable_size(8 * 1024 * 1024)
.manual_journal_persist(true); .manual_journal_persist(true);
@@ -63,7 +55,7 @@ where
} }
pub fn import( pub fn import(
keyspace: &TransactionalKeyspace, keyspace: &Keyspace,
path: &Path, path: &Path,
name: &str, name: &str,
version: Version, version: Version,
@@ -83,17 +75,13 @@ where
}, },
)?; )?;
let rtx = keyspace.read_tx();
Ok(Self { Ok(Self {
meta, meta,
name: Box::leak(Box::new(name.to_string())), name: Box::leak(Box::new(name.to_string())),
keyspace: keyspace.clone(), keyspace: keyspace.clone(),
partition: Arc::new(RwLock::new(Some(partition))), partition,
rtx: Arc::new(RwLock::new(Some(rtx))),
puts: FxHashMap::default(), puts: FxHashMap::default(),
dels: FxHashSet::default(), dels: FxHashSet::default(),
bloom_filters,
}) })
} }
@@ -104,13 +92,7 @@ where
{ {
if let Some(v) = self.puts.get(key) { if let Some(v) = self.puts.get(key) {
Ok(Some(Cow::Borrowed(v))) Ok(Some(Cow::Borrowed(v)))
} else if let Some(slice) = self } else if let Some(slice) = self.partition.get(ByteView::from(key))? {
.rtx
.read()
.as_ref()
.unwrap()
.get(self.partition.read().as_ref().unwrap(), ByteView::from(key))?
{
Ok(Some(Cow::Owned(V::from(ByteView::from(&*slice))))) Ok(Some(Cow::Owned(V::from(ByteView::from(&*slice)))))
} else { } else {
Ok(None) Ok(None)
@@ -118,20 +100,12 @@ where
} }
pub fn is_empty(&self) -> Result<bool> { pub fn is_empty(&self) -> Result<bool> {
self.rtx self.partition.is_empty().map_err(|e| e.into())
.read()
.as_ref()
.unwrap()
.is_empty(self.partition.read().as_ref().unwrap())
.map_err(|e| e.into())
} }
pub fn iter(&self) -> impl Iterator<Item = (K, V)> { pub fn iter(&self) -> impl Iterator<Item = (K, V)> {
self.rtx self.partition
.read() .iter()
.as_ref()
.unwrap()
.iter(self.partition.read().as_ref().unwrap())
.map(|res| res.unwrap()) .map(|res| res.unwrap())
.map(|(k, v)| (K::from(ByteView::from(&*k)), V::from(ByteView::from(&*v)))) .map(|(k, v)| (K::from(ByteView::from(&*k)), V::from(ByteView::from(&*v))))
} }
@@ -146,7 +120,6 @@ where
#[inline] #[inline]
pub fn remove(&mut self, key: K) { pub fn remove(&mut self, key: K) {
// Hot path: key was recently inserted
if self.puts.remove(&key).is_some() { if self.puts.remove(&key).is_some() {
return; return;
} }
@@ -162,19 +135,6 @@ where
} }
} }
// pub fn retain_or_del<F>(&mut self, retain: F)
// where
// F: Fn(&K, &mut V) -> bool,
// {
// self.puts.retain(|k, v| {
// let ret = retain(k, v);
// if !ret {
// self.dels.insert(k.clone());
// }
// ret
// });
// }
#[inline] #[inline]
fn has(&self, height: Height) -> bool { fn has(&self, height: Height) -> bool {
self.meta.has(height) self.meta.has(height)
@@ -204,27 +164,17 @@ where
return Ok(()); return Ok(());
} }
let mut rtx = self.rtx.write(); let mut items = self
let _ = rtx.take(); .puts
.drain()
.map(|(key, value)| InnerItem::Value { key, value })
.chain(self.dels.drain().map(|key| InnerItem::WeakTombstone(key)))
.collect::<Vec<_>>();
items.sort_unstable();
let mut wtx = self.keyspace.write_tx(); self.keyspace
.batch()
let partition = self.partition.read(); .commit_single_partition(&self.partition, items)?;
let partition = partition.as_ref().unwrap();
wtx.remove_batch(partition, self.dels.drain().map(ByteView::from));
wtx.insert_batch(
partition,
self.puts
.drain()
.map(|(k, v)| (ByteView::from(k), ByteView::from(v))),
);
wtx.commit()?;
rtx.replace(self.keyspace.read_tx());
Ok(()) Ok(())
} }
@@ -239,24 +189,6 @@ where
self.name self.name
} }
fn reset(&mut self) -> Result<()> {
info!("Resetting {}...", self.name);
let mut opt = self.partition.write();
let partition = opt.take().unwrap();
self.keyspace.delete_partition(partition)?;
self.meta.reset();
let partition = Self::open_partition_handle(&self.keyspace, self.name, self.bloom_filters)?;
opt.replace(partition);
Ok(())
}
fn height(&self) -> Option<Height> { fn height(&self) -> Option<Height> {
self.meta.height() self.meta.height()
} }
crates/brk_store/src/v3/mod.rs
+12 -12
View File
@@ -217,26 +217,26 @@ where
self.name self.name
} }
fn reset(&mut self) -> Result<()> { // fn reset(&mut self) -> Result<()> {
info!("Resetting {}...", self.name); // info!("Resetting {}...", self.name);
todo!(); // todo!();
let mut opt = self.keyspace.write(); // let mut opt = self.keyspace.write();
let keyspace = opt.take().unwrap(); // let keyspace = opt.take().unwrap();
// Doesn't exist yet // // Doesn't exist yet
// self.database.remove_keyspace(keyspace)?; // // self.database.remove_keyspace(keyspace)?;
self.meta.reset(); // self.meta.reset();
let keyspace = Self::open_keyspace(&self.database, self.name)?; // let keyspace = Self::open_keyspace(&self.database, self.name)?;
opt.replace(keyspace); // opt.replace(keyspace);
Ok(()) // Ok(())
} // }
fn height(&self) -> Option<Height> { fn height(&self) -> Option<Height> {
self.meta.height() self.meta.height()