Mirror/brk
1
0
Fork 0
mirror of https://github.com/bitcoinresearchkit/brk.git synced 2026-06-30 22:09:00 -07:00
Code Issues Packages Projects Releases Wiki Activity

bitview: reorg part 7 + fix hanging ?

Browse Source
This commit is contained in:
nym21
2025-09-28 20:33:55 +02:00
parent 9d03fdf31d
commit 7407c032e5
16 changed files with 581 additions and 495 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/brk_parser/src/lib.rs
+22 -16
View File
@@ -16,7 +16,7 @@ use bitcoincore_rpc::RpcApi;
use blk_index_to_blk_path::*;
use brk_error::Result;
use brk_structs::{BlkMetadata, BlkPosition, Block, Height, ParsedBlock};
use crossbeam::channel::{bounded, Receiver};
use crossbeam::channel::{Receiver, bounded};
use parking_lot::{RwLock, RwLockReadGuard};
use rayon::prelude::*;
@@ -157,30 +157,36 @@ impl Parser {
let mut bulk = vec![];
let drain_and_send = |bulk: &mut Vec<(BlkMetadata, AnyBlock, XORIndex)>| {
// Using a vec and sending after to not end up with stuck threads in par iter
mem::take(bulk)
.into_par_iter()
.try_for_each(|(metdata, any_block, xor_i)| {
if let Ok(AnyBlock::Decoded(block)) =
any_block.decode(metdata, rpc, xor_i, xor_bytes, start, end)
&& send_block.send(block).is_err()
{
return ControlFlow::Break(());
}
// Private pool to prevent collision with the global pool
// Without it there can be hanging
let parser_pool = rayon::ThreadPoolBuilder::new()
.num_threads(thread::available_parallelism().unwrap().get() / 2)
.build()
.expect("Failed to create parser thread pool");
ControlFlow::Continue(())
})
let drain_and_send = |bulk: &mut Vec<(BlkMetadata, AnyBlock, XORIndex)>| {
parser_pool.install(|| {
mem::take(bulk)
.into_par_iter()
.try_for_each(|(metdata, any_block, xor_i)| {
if let Ok(AnyBlock::Decoded(block)) =
any_block.decode(metdata, rpc, xor_i, xor_bytes, start, end)
&& send_block.send(block).is_err()
{
return ControlFlow::Break(());
}
ControlFlow::Continue(())
})
})
};
recv_bytes.iter().try_for_each(|tuple| {
bulk.push(tuple);
if bulk.len() < BOUND_CAP / 2 {
if bulk.len() < BOUND_CAP {
return ControlFlow::Continue(());
}
// Sending in bulk to not lock threads in standby
drain_and_send(&mut bulk)
})?;