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

iterator: simplified

Browse Source
This commit is contained in:
nym21
2025-02-21 20:12:57 +01:00
parent 66ecd2fcf8
commit e63b42278c
14 changed files with 241 additions and 365 deletions
Download Patch File Download Diff File Expand all files Collapse all files
iterator/Cargo.toml
+4 -5
View File
@@ -9,16 +9,15 @@ edition = { workspace = true }
license = { workspace = true }
[features]
fjall = ["dep:fjall"]
zerocopy = ["dep:zerocopy"]
bytes = ["dep:fjall", "dep:zerocopy"]
[dependencies]
bitcoin = { workspace = true }
rayon = { workspace = true }
bitcoincore-rpc = "0.19.0"
crossbeam = { version = "0.8.4", features = ["crossbeam-channel"] }
derive_deref = { workspace = true }
fjall = { workspace = true, optional = true }
rayon = { workspace = true }
serde = { workspace = true }
serde_json = { workspace = true }
derive_deref = { workspace = true }
bitcoincore-rpc = "0.19.0"
zerocopy = { workspace = true, optional = true }
iterator/src/blk_index_to_blk_path.rs
+2 -2
View File
@@ -10,7 +10,7 @@ const BLK: &str = "blk";
const DAT: &str = ".dat";
#[derive(Debug, Deref, DerefMut)]
pub struct BlkIndexToBlkPath(BTreeMap<usize, PathBuf>);
pub struct BlkIndexToBlkPath(BTreeMap<u16, PathBuf>);
impl BlkIndexToBlkPath {
pub fn scan(data_dir: &Path) -> Self {
@@ -35,7 +35,7 @@ impl BlkIndexToBlkPath {
let file_name = path.file_name().unwrap().to_str().unwrap();
let blk_index = file_name[BLK.len()..(file_name.len() - DAT.len())]
.parse::<usize>()
.parse::<u16>()
.unwrap();
(blk_index, path)
iterator/src/blk_index_to_blk_recap.rs
+37 -71
View File
@@ -1,29 +1,23 @@
use std::{
cmp::Ordering,
collections::{BTreeMap, BTreeSet},
fs::{self, File},
fs::File,
io::{BufReader, BufWriter},
path::{Path, PathBuf},
};
use crate::{blk_recap::BlkRecap, BlkIndexToBlkPath, BlkMetadataAndBlock, Height};
const TARGET_BLOCKS_PER_MONTH: usize = 144 * 30;
use crate::{blk_recap::BlkRecap, BlkIndexToBlkPath, Height};
#[derive(Debug)]
pub struct BlkIndexToBlkRecap {
path: PathBuf,
tree: BTreeMap<usize, BlkRecap>,
last_safe_height: Option<Height>,
pub path: PathBuf,
pub tree: BTreeMap<u16, BlkRecap>,
}
impl BlkIndexToBlkRecap {
pub fn import(blocks_dir: &BlkIndexToBlkPath, data_dir: &Path) -> Self {
pub fn import(data_dir: &Path, blk_index_to_blk_path: &BlkIndexToBlkPath, start: Option<Height>) -> (Self, u16) {
let path = data_dir.join("blk_index_to_blk_recap.json");
let tree = {
fs::create_dir_all(data_dir).unwrap();
if let Ok(file) = File::open(&path) {
let reader = BufReader::new(file);
serde_json::from_reader(reader).unwrap_or_default()
@@ -32,90 +26,62 @@ impl BlkIndexToBlkRecap {
}
};
// dbg!(&tree);
let mut slf = Self { path, tree };
let mut this = Self {
path,
tree,
last_safe_height: None,
};
let min_removed = slf.clean_outdated(blk_index_to_blk_path);
this.clean_outdated(blocks_dir);
let blk_index = slf.get_start_recap(min_removed, start);
this
(slf, blk_index)
}
fn clean_outdated(&mut self, blocks_dir: &BlkIndexToBlkPath) {
self.tree.pop_last();
fn clean_outdated(&mut self, blk_index_to_blk_path: &BlkIndexToBlkPath) -> Option<u16> {
let mut min_removed_blk_index: Option<u16> = None;
let mut unprocessed_keys = self.tree.keys().copied().collect::<BTreeSet<_>>();
blocks_dir.iter().for_each(|(blk_index, blk_path)| {
blk_index_to_blk_path.iter().for_each(|(blk_index, blk_path)| {
unprocessed_keys.remove(blk_index);
if let Some(blk_recap) = self.tree.get(blk_index) {
if blk_recap.has_different_modified_time(blk_path) {
self.tree.remove(blk_index);
self.tree.remove(blk_index).unwrap();
if min_removed_blk_index.map_or(true, |_blk_index| *blk_index < _blk_index) {
min_removed_blk_index.replace(*blk_index);
}
}
}
});
unprocessed_keys.into_iter().for_each(|blk_index| {
self.tree.remove(&blk_index);
self.tree.remove(&blk_index).unwrap();
if min_removed_blk_index.map_or(true, |_blk_index| blk_index < _blk_index) {
min_removed_blk_index.replace(blk_index);
}
});
self.last_safe_height = self.tree.values().map(|recap| recap.height()).max();
min_removed_blk_index
}
pub fn get_start_recap(&mut self, start: Option<Height>) -> Option<(usize, BlkRecap)> {
if let Some(start) = start {
let (last_key, last_value) = self.tree.last_key_value()?;
pub fn get_start_recap(&mut self, min_removed: Option<u16>, start: Option<Height>) -> u16 {
if start.is_none() {
return 0;
}
dbg!((last_key, last_value));
let height = start.unwrap();
if last_value.height() < start {
return Some((*last_key, *last_value));
} else if let Some((blk_index, _)) =
self.tree.iter().find(|(_, blk_recap)| blk_recap.is_younger_than(start))
{
if *blk_index != 0 {
// Temporary fix, need to rethink the whole thing
let blk_index = (*blk_index).checked_sub(1).unwrap_or_default();
return Some((blk_index, *self.tree.get(&blk_index).unwrap()));
}
let mut start = 0;
if let Some(found) = self.tree.iter().find(|(_, recap)| recap.max_height >= height) {
start = *found.0;
}
if let Some(min_removed) = min_removed {
if start > min_removed {
start = min_removed;
}
}
None
}
pub fn update(&mut self, blk_metadata_and_block: &BlkMetadataAndBlock, height: Height) {
let blk_index = blk_metadata_and_block.blk_metadata.index;
if let Some(last_entry) = self.tree.last_entry() {
match last_entry.key().cmp(&blk_index) {
Ordering::Greater => {
last_entry.remove_entry();
}
Ordering::Less => {
self.tree
.insert(blk_index, BlkRecap::from(height, blk_metadata_and_block));
}
Ordering::Equal => {}
};
} else {
if blk_index != 0 || height != 0 {
// dbg!(blk_index, height);
unreachable!();
}
self.tree.insert(blk_index, BlkRecap::first(blk_metadata_and_block));
}
if self.last_safe_height.map_or(true, |safe_height| height >= safe_height)
&& (height % TARGET_BLOCKS_PER_MONTH) == 0
{
self.export();
}
start
}
pub fn export(&self) {
@@ -124,6 +90,6 @@ impl BlkIndexToBlkRecap {
panic!("No such file or directory")
});
serde_json::to_writer_pretty(&mut BufWriter::new(file), &self.tree).unwrap();
serde_json::to_writer(&mut BufWriter::new(file), &self.tree).unwrap();
}
}
iterator/src/blk_metadata.rs
+2 -2
View File
@@ -4,12 +4,12 @@ use crate::path_to_modified_time;
#[derive(Debug, Clone, Copy)]
pub struct BlkMetadata {
pub index: usize,
pub index: u16,
pub modified_time: u64,
}
impl BlkMetadata {
pub fn new(index: usize, path: &Path) -> Self {
pub fn new(index: u16, path: &Path) -> Self {
Self {
index,
modified_time: path_to_modified_time(path),
iterator/src/blk_metadata_and_block.rs
+2 -2
View File
@@ -3,12 +3,12 @@ use bitcoin::Block;
use crate::BlkMetadata;
#[derive(Debug)]
pub struct BlkMetadataAndBlock {
pub struct BlkIndexAndBlock {
pub blk_metadata: BlkMetadata,
pub block: Block,
}
impl BlkMetadataAndBlock {
impl BlkIndexAndBlock {
pub fn new(blk_metadata: BlkMetadata, block: Block) -> Self {
Self { blk_metadata, block }
}
iterator/src/blk_recap.rs
+4 -33
View File
@@ -1,50 +1,21 @@
use std::path::Path;
use bitcoin::{hashes::Hash, BlockHash};
use serde::{Deserialize, Serialize};
use crate::{path_to_modified_time, BlkMetadataAndBlock, Height};
use crate::{path_to_modified_time, Height};
#[derive(Debug, Clone, Copy, Serialize, Deserialize)]
#[repr(C)]
pub struct BlkRecap {
min_continuous_height: Height,
min_continuous_prev_hash: BlockHash,
modified_time: u64,
pub max_height: Height,
pub modified_time: u64,
}
impl BlkRecap {
pub fn first(blk_metadata_and_block: &BlkMetadataAndBlock) -> Self {
Self {
min_continuous_height: Height::default(),
min_continuous_prev_hash: BlockHash::all_zeros(),
modified_time: blk_metadata_and_block.blk_metadata.modified_time,
}
}
pub fn from(height: Height, blk_metadata_and_block: &BlkMetadataAndBlock) -> Self {
Self {
min_continuous_height: height,
min_continuous_prev_hash: blk_metadata_and_block.block.header.prev_blockhash,
modified_time: blk_metadata_and_block.blk_metadata.modified_time,
}
}
pub fn has_different_modified_time(&self, blk_path: &Path) -> bool {
if self.modified_time != path_to_modified_time(blk_path) {
dbg!(self.modified_time, path_to_modified_time(blk_path));
}
self.modified_time != path_to_modified_time(blk_path)
}
pub fn is_younger_than(&self, height: Height) -> bool {
self.min_continuous_height > height
}
pub fn height(&self) -> Height {
self.min_continuous_height
}
pub fn prev_hash(&self) -> &BlockHash {
&self.min_continuous_prev_hash
}
}
iterator/src/error.rs
+1 -3
View File
@@ -8,7 +8,6 @@ pub type Result<T, E = Error> = std::result::Result<T, E>;
#[derive(Debug)]
pub enum Error {
IO(io::Error),
#[cfg(feature = "zerocopy")]
ZeroCopyError,
}
@@ -18,7 +17,7 @@ impl From<io::Error> for Error {
}
}
#[cfg(feature = "zerocopy")]
#[cfg(feature = "bytes")]
impl<A, B> From<zerocopy::error::SizeError<A, B>> for Error {
fn from(_: zerocopy::error::SizeError<A, B>) -> Self {
Self::ZeroCopyError
@@ -29,7 +28,6 @@ impl fmt::Display for Error {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
match self {
Error::IO(error) => Debug::fmt(&error, f),
#[cfg(feature = "zerocopy")]
Error::ZeroCopyError => write!(f, "Zero copy convert error"),
}
}
iterator/src/height.rs
+8 -8
View File
@@ -5,20 +5,20 @@ use std::{
use derive_deref::{Deref, DerefMut};
use serde::{Deserialize, Serialize};
#[cfg(feature = "zerocopy")]
#[cfg(feature = "bytes")]
use zerocopy::{FromBytes, Immutable, IntoBytes, KnownLayout};
use crate::rpc::{self, RpcApi};
#[derive(Debug, Clone, Copy, Deref, DerefMut, PartialEq, Eq, PartialOrd, Ord, Default, Serialize, Deserialize)]
#[cfg_attr(feature = "zerocopy", derive(FromBytes, Immutable, IntoBytes, KnownLayout))]
#[cfg_attr(feature = "bytes", derive(FromBytes, Immutable, IntoBytes, KnownLayout,))]
pub struct Height(u32);
impl Height {
const ZERO: Self = Height(0);
pub const ZERO: Self = Height(0);
pub const MAX: Self = Height(u32::MAX);
#[cfg(feature = "zerocopy")]
#[cfg(feature = "bytes")]
pub fn write(&self, path: &std::path::Path) -> Result<(), std::io::Error> {
std::fs::write(path, self.as_bytes())
}
@@ -163,7 +163,7 @@ impl From<Height> for bitcoin::locktime::absolute::Height {
}
}
#[cfg(feature = "zerocopy")]
#[cfg(feature = "bytes")]
impl TryFrom<&std::path::Path> for Height {
type Error = crate::Error;
fn try_from(value: &std::path::Path) -> Result<Self, Self::Error> {
@@ -171,14 +171,14 @@ impl TryFrom<&std::path::Path> for Height {
}
}
#[cfg(feature = "fjall")]
#[cfg(feature = "bytes")]
impl TryFrom<fjall::Slice> for Height {
type Error = crate::Error;
fn try_from(value: fjall::Slice) -> Result<Self, Self::Error> {
Ok(Self::read_from_bytes(&value)?)
}
}
#[cfg(feature = "fjall")]
#[cfg(feature = "bytes")]
impl From<Height> for fjall::Slice {
fn from(value: Height) -> Self {
Self::new(value.as_bytes())
iterator/src/lib.rs
+91 -151
View File
@@ -1,5 +1,6 @@
use std::{
collections::{BTreeMap, BTreeSet, VecDeque},
cmp::Ordering,
collections::BTreeMap,
fs::{self},
ops::ControlFlow,
path::Path,
@@ -8,12 +9,12 @@ use std::{
use bitcoin::{
consensus::{Decodable, ReadExt},
hashes::Hash,
io::{Cursor, Read},
Block, BlockHash,
};
use bitcoincore_rpc::RpcApi;
use blk_index_to_blk_path::*;
use blk_recap::BlkRecap;
use crossbeam::channel::{bounded, Receiver};
use rayon::prelude::*;
@@ -41,41 +42,9 @@ const MAGIC_BYTES: [u8; 4] = [249, 190, 180, 217];
const BOUND_CAP: usize = 100;
///
/// Returns a crossbeam channel receiver that receives `(usize, Block, BlockHash)` tuples (with `usize` being the height) in sequential order.
/// Returns a crossbeam channel receiver that receives `(Height, Block, BlockHash)` tuples from an **inclusive** range (`start` and `end`)
///
/// # Arguments
///
/// * `data_dir` - Path to the Bitcoin data directory
/// * `start` - Inclusive starting height of the blocks received, `None` for 0
/// * `end` - Inclusive ending height of the blocks received, `None` for the last one
/// * `rpc` - RPC client to filter out forks
///
/// # Example
///
/// ```rust
/// use std::path::Path;
///
/// use bitcoincore_rpc::{Auth, Client};
///
/// let i = std::time::Instant::now();
///
/// let data_dir = Path::new("../../bitcoin");
/// let url = "http://localhost:8332";
/// let cookie = Path::new(data_dir).join(".cookie");
/// let auth = Auth::CookieFile(cookie);
/// let rpc = Client::new(url, auth).unwrap();
///
/// let start = Some(850_000);
/// let end = None;
///
/// biter::new(data_dir, start, end, rpc)
/// .iter()
/// .for_each(|(height, _block, hash)| {
/// println!("{height}: {hash}");
/// });
///
/// dbg!(i.elapsed());
/// ```
/// For an example checkout `iterator/main.rs`
///
pub fn new(
data_dir: &Path,
@@ -89,17 +58,15 @@ pub fn new(
let blk_index_to_blk_path = BlkIndexToBlkPath::scan(data_dir);
let mut blk_index_to_blk_recap = BlkIndexToBlkRecap::import(&blk_index_to_blk_path, data_dir);
let start_recap = blk_index_to_blk_recap.get_start_recap(start);
let starting_blk_index = start_recap.as_ref().map_or(0, |(index, _)| *index);
let (mut blk_index_to_blk_recap, blk_index) = BlkIndexToBlkRecap::import(data_dir, &blk_index_to_blk_path, start);
thread::spawn(move || {
blk_index_to_blk_path
.iter()
.filter(|(blk_index, _)| **blk_index >= starting_blk_index)
.range(blk_index..)
.try_for_each(move |(blk_index, blk_path)| {
let blk_metadata = BlkMetadata::new(*blk_index, blk_path.as_path());
let blk_index = *blk_index;
let blk_metadata = BlkMetadata::new(blk_index, blk_path.as_path());
let blk_bytes = fs::read(blk_path).unwrap();
let blk_bytes_len = blk_bytes.len() as u64;
@@ -128,22 +95,19 @@ pub fn new(
}
}
let block_size = cursor.read_u32().unwrap();
let len = cursor.read_u32().unwrap();
let mut raw_block = vec![0u8; block_size as usize];
let mut bytes = vec![0u8; len as usize];
cursor.read_exact(&mut raw_block).unwrap();
cursor.read_exact(&mut bytes).unwrap();
if send_block_reader
.send((blk_metadata, BlockState::Raw(raw_block)))
.is_err()
{
if send_block_reader.send((blk_metadata, BlockState::Raw(bytes))).is_err() {
return ControlFlow::Break(());
}
}
ControlFlow::Continue(())
})
});
});
thread::spawn(move || {
@@ -152,16 +116,7 @@ pub fn new(
let drain_and_send = |bulk: &mut Vec<_>| {
// Using a vec and sending after to not end up with stuck threads in par iter
bulk.par_iter_mut().for_each(|(_, block_state)| {
let raw_block = match block_state {
BlockState::Raw(vec) => vec,
_ => unreachable!(),
};
let mut cursor = Cursor::new(raw_block);
let block = Block::consensus_decode(&mut cursor).unwrap();
*block_state = BlockState::Decoded(block);
BlockState::decode(block_state);
});
bulk.drain(..).try_for_each(|(blk_metadata, block_state)| {
@@ -170,7 +125,7 @@ pub fn new(
_ => unreachable!(),
};
if send_block.send(BlkMetadataAndBlock::new(blk_metadata, block)).is_err() {
if send_block.send(BlkIndexAndBlock::new(blk_metadata, block)).is_err() {
return ControlFlow::Break(());
}
@@ -193,113 +148,83 @@ pub fn new(
});
thread::spawn(move || {
let mut height = start_recap.map_or(Height::default(), |(_, recap)| recap.height());
let mut current_height = start.unwrap_or_default();
let mut future_blocks = BTreeMap::default();
let mut recent_chain: VecDeque<(BlockHash, BlkMetadataAndBlock)> = VecDeque::default();
let mut recent_hashes: BTreeSet<BlockHash> = BTreeSet::default();
let mut prev_hash = start_recap.map_or_else(BlockHash::all_zeros, |(_, recap)| *recap.prev_hash());
recv_block.iter().try_for_each(|tuple| -> ControlFlow<(), _> {
let blk_metadata = tuple.blk_metadata;
let block = tuple.block;
let hash = block.block_hash();
let header = rpc.get_block_header_info(&hash);
let mut prepare_and_send = |(hash, tuple): (BlockHash, BlkMetadataAndBlock)| {
blk_index_to_blk_recap.update(&tuple, height);
if start.map_or(true, |start| start <= height) {
send_height_block_hash.send((height, tuple.block, hash)).unwrap();
if header.is_err() {
return ControlFlow::Continue(());
}
let header = header.unwrap();
if header.confirmations <= 0 {
return ControlFlow::Continue(());
}
if end == Some(height) {
return ControlFlow::Break(());
let height = Height::from(header.height);
let len = blk_index_to_blk_recap.tree.len();
if blk_metadata.index == len as u16 || blk_metadata.index + 1 == len as u16 {
match (len as u16).cmp(&blk_metadata.index) {
Ordering::Equal => {
if len % 21 == 0 {
blk_index_to_blk_recap.export();
}
}
Ordering::Less => panic!(),
Ordering::Greater => {}
}
blk_index_to_blk_recap
.tree
.entry(blk_metadata.index)
.and_modify(|recap| {
if recap.max_height < height {
recap.max_height = height;
}
})
.or_insert(BlkRecap {
max_height: height,
modified_time: blk_metadata.modified_time,
});
} else {
dbg!(blk_metadata.index, len);
panic!()
}
height += 1;
let mut opt = if current_height == height {
Some((block, hash))
} else {
if start.map_or(true, |start| start <= height) && end.map_or(true, |end| end >= height) {
future_blocks.insert(height, (block, hash));
}
None
};
ControlFlow::Continue(())
};
while let Some((block, hash)) = opt.take().or_else(|| {
if !future_blocks.is_empty() {
future_blocks.remove(&current_height)
} else {
None
}
}) {
send_height_block_hash.send((current_height, block, hash)).unwrap();
let mut update_tip = |prev_hash: &mut BlockHash,
recent_hashes: &mut BTreeSet<BlockHash>,
recent_chain: &mut VecDeque<(BlockHash, BlkMetadataAndBlock)>,
future_blocks: &mut BTreeMap<BlockHash, BlkMetadataAndBlock>,
tuple: BlkMetadataAndBlock| {
let mut tuple = Some(tuple);
while let Some(tuple) = tuple.take().or_else(|| future_blocks.remove(prev_hash)) {
let hash = tuple.block.block_hash();
*prev_hash = hash;
recent_hashes.insert(hash);
recent_chain.push_back((hash, tuple));
}
while recent_chain.len() > NUMBER_OF_UNSAFE_BLOCKS {
let (hash, tuple) = recent_chain.pop_front().unwrap();
recent_hashes.remove(&hash);
if prepare_and_send((hash, tuple)).is_break() {
if end == Some(current_height) {
return ControlFlow::Break(());
}
}
ControlFlow::Continue(())
};
let flow = recv_block.iter().try_for_each(|tuple| {
// block isn't next after current tip
if prev_hash != tuple.block.header.prev_blockhash {
let is_block_active = |hash| rpc.get_block_header_info(hash).unwrap().confirmations > 0;
// block prev has already been processed
if recent_hashes.contains(&tuple.block.header.prev_blockhash) {
let hash = tuple.block.block_hash();
if is_block_active(&hash) {
let prev_index = recent_chain
.iter()
.position(|(hash, ..)| hash == &tuple.block.header.prev_blockhash)
.unwrap();
let bad_index_start = prev_index + 1;
recent_chain.drain(bad_index_start..).for_each(|(hash, _)| {
recent_hashes.remove(&hash);
});
return update_tip(
&mut prev_hash,
&mut recent_hashes,
&mut recent_chain,
&mut future_blocks,
tuple,
);
}
// Check if there was already a future block with the same prev hash
} else if let Some(prev_tuple) = future_blocks.insert(tuple.block.header.prev_blockhash, tuple) {
// If the previous was the active one
if is_block_active(&prev_tuple.block.block_hash()) {
// Rollback the insert
future_blocks.insert(prev_tuple.block.header.prev_blockhash, prev_tuple);
}
}
} else {
return update_tip(
&mut prev_hash,
&mut recent_hashes,
&mut recent_chain,
&mut future_blocks,
tuple,
);
current_height.increment();
}
ControlFlow::Continue(())
});
if flow.is_continue() {
// Send the last (up to 100) blocks
recent_chain.into_iter().try_for_each(prepare_and_send);
}
blk_index_to_blk_recap.export();
});
@@ -310,3 +235,18 @@ enum BlockState {
Raw(Vec<u8>),
Decoded(Block),
}
impl BlockState {
pub fn decode(&mut self) {
let bytes = match self {
BlockState::Raw(bytes) => bytes,
_ => unreachable!(),
};
let mut cursor = Cursor::new(bytes);
let block = Block::consensus_decode(&mut cursor).unwrap();
*self = BlockState::Decoded(block);
}
}
iterator/src/main.rs
+1 -1
View File
@@ -15,7 +15,7 @@ fn main() {
));
let start = None;
let end = None;
let end = None; //Some(200_000_u32.into());
biterator::new(data_dir, start, end, rpc)
.iter()