mirror of
https://github.com/bitcoinresearchkit/brk.git
synced 2026-07-15 21:18:11 -07:00
computer: fix par compute_rest_part2 crashing external drives
This commit is contained in:
Generated
+1
@@ -546,6 +546,7 @@ dependencies = [
|
|||||||
"brk_vec",
|
"brk_vec",
|
||||||
"color-eyre",
|
"color-eyre",
|
||||||
"derive_deref",
|
"derive_deref",
|
||||||
|
"either",
|
||||||
"fjall",
|
"fjall",
|
||||||
"jiff",
|
"jiff",
|
||||||
"log",
|
"log",
|
||||||
|
|||||||
@@ -21,6 +21,7 @@ brk_store = { workspace = true }
|
|||||||
brk_vec = { workspace = true }
|
brk_vec = { workspace = true }
|
||||||
color-eyre = { workspace = true }
|
color-eyre = { workspace = true }
|
||||||
derive_deref = { workspace = true }
|
derive_deref = { workspace = true }
|
||||||
|
either = "1.15.0"
|
||||||
fjall = { workspace = true }
|
fjall = { workspace = true }
|
||||||
jiff = { workspace = true }
|
jiff = { workspace = true }
|
||||||
log = { workspace = true }
|
log = { workspace = true }
|
||||||
|
|||||||
@@ -598,6 +598,7 @@ impl Stores {
|
|||||||
}
|
}
|
||||||
|
|
||||||
pub fn rotate_memtables(&self) {
|
pub fn rotate_memtables(&self) {
|
||||||
|
info!("Rotatin memtables...");
|
||||||
self.as_slice()
|
self.as_slice()
|
||||||
.into_iter()
|
.into_iter()
|
||||||
.for_each(|store| store.rotate_memtable());
|
.for_each(|store| store.rotate_memtable());
|
||||||
|
|||||||
@@ -10,6 +10,7 @@ use brk_vec::{
|
|||||||
AnyCollectableVec, AnyVec, CollectableVec, Computation, EagerVec, Format, GenericStoredVec,
|
AnyCollectableVec, AnyVec, CollectableVec, Computation, EagerVec, Format, GenericStoredVec,
|
||||||
StoredIndex, StoredVec, UnsafeSlice, VecIterator,
|
StoredIndex, StoredVec, UnsafeSlice, VecIterator,
|
||||||
};
|
};
|
||||||
|
use either::Either;
|
||||||
use log::info;
|
use log::info;
|
||||||
use rayon::prelude::*;
|
use rayon::prelude::*;
|
||||||
|
|
||||||
@@ -1238,26 +1239,28 @@ impl Vecs {
|
|||||||
&self.addresstype_to_height_to_empty_address_count,
|
&self.addresstype_to_height_to_empty_address_count,
|
||||||
)?;
|
)?;
|
||||||
|
|
||||||
thread::scope(|scope| {
|
self.utxo_vecs
|
||||||
scope.spawn(|| {
|
.as_mut_vecs()
|
||||||
self.utxo_vecs
|
.into_iter()
|
||||||
.as_mut_vecs()
|
.map(|(_, v)| v)
|
||||||
.par_iter_mut()
|
.map(Either::Left)
|
||||||
.try_for_each(|(_, v)| {
|
.chain(
|
||||||
v.compute_rest_part1(indexer, indexes, fetched, starting_indexes, exit)
|
|
||||||
})
|
|
||||||
.unwrap();
|
|
||||||
});
|
|
||||||
scope.spawn(|| {
|
|
||||||
self.address_vecs
|
self.address_vecs
|
||||||
.as_mut_vecs()
|
.as_mut_vecs()
|
||||||
.par_iter_mut()
|
.into_iter()
|
||||||
.try_for_each(|(_, v)| {
|
.map(|(_, v)| v)
|
||||||
v.compute_rest_part1(indexer, indexes, fetched, starting_indexes, exit)
|
.map(Either::Right),
|
||||||
})
|
)
|
||||||
.unwrap();
|
.collect::<Vec<Either<&mut utxo_cohort::Vecs, &mut address_cohort::Vecs>>>()
|
||||||
});
|
.into_par_iter()
|
||||||
});
|
.try_for_each(|either| match either {
|
||||||
|
Either::Left(v) => {
|
||||||
|
v.compute_rest_part1(indexer, indexes, fetched, starting_indexes, exit)
|
||||||
|
}
|
||||||
|
Either::Right(v) => {
|
||||||
|
v.compute_rest_part1(indexer, indexes, fetched, starting_indexes, exit)
|
||||||
|
}
|
||||||
|
})?;
|
||||||
|
|
||||||
info!("Computing rest part 2...");
|
info!("Computing rest part 2...");
|
||||||
|
|
||||||
@@ -1278,49 +1281,57 @@ impl Vecs {
|
|||||||
.indexes_to_realized_cap
|
.indexes_to_realized_cap
|
||||||
.as_ref()
|
.as_ref()
|
||||||
.map(|v| v.dateindex.unwrap_last().clone());
|
.map(|v| v.dateindex.unwrap_last().clone());
|
||||||
|
let dateindex_to_supply_ref = dateindex_to_supply.as_ref().unwrap();
|
||||||
|
let height_to_realized_cap_ref = height_to_realized_cap.as_ref();
|
||||||
|
let dateindex_to_realized_cap_ref = dateindex_to_realized_cap.as_ref();
|
||||||
|
|
||||||
thread::scope(|scope| {
|
let vecs = self
|
||||||
scope.spawn(|| {
|
.utxo_vecs
|
||||||
self.utxo_vecs
|
.as_mut_vecs()
|
||||||
.as_mut_vecs()
|
.into_iter()
|
||||||
.par_iter_mut()
|
.map(|(_, v)| v)
|
||||||
.try_for_each(|(_, v)| {
|
.map(Either::Left)
|
||||||
v.compute_rest_part2(
|
.chain(
|
||||||
indexer,
|
|
||||||
indexes,
|
|
||||||
fetched,
|
|
||||||
starting_indexes,
|
|
||||||
market,
|
|
||||||
&height_to_supply,
|
|
||||||
dateindex_to_supply.as_ref().unwrap(),
|
|
||||||
height_to_realized_cap.as_ref(),
|
|
||||||
dateindex_to_realized_cap.as_ref(),
|
|
||||||
exit,
|
|
||||||
)
|
|
||||||
})
|
|
||||||
.unwrap();
|
|
||||||
});
|
|
||||||
scope.spawn(|| {
|
|
||||||
self.address_vecs
|
self.address_vecs
|
||||||
.as_mut_vecs()
|
.as_mut_vecs()
|
||||||
.par_iter_mut()
|
.into_iter()
|
||||||
.try_for_each(|(_, v)| {
|
.map(|(_, v)| v)
|
||||||
v.compute_rest_part2(
|
.map(Either::Right),
|
||||||
indexer,
|
)
|
||||||
indexes,
|
.collect::<Vec<Either<&mut utxo_cohort::Vecs, &mut address_cohort::Vecs>>>();
|
||||||
fetched,
|
|
||||||
starting_indexes,
|
// Capped as external drives (even thunderbolt 4 SSDs) can be overwhelmed
|
||||||
market,
|
let chunk_size = (vecs.len() as f64 / 4.0).ceil() as usize;
|
||||||
&height_to_supply,
|
vecs.into_par_iter().chunks(chunk_size).try_for_each(|v| {
|
||||||
dateindex_to_supply.as_ref().unwrap(),
|
v.into_iter().try_for_each(|either| match either {
|
||||||
height_to_realized_cap.as_ref(),
|
Either::Left(v) => v.compute_rest_part2(
|
||||||
dateindex_to_realized_cap.as_ref(),
|
indexer,
|
||||||
exit,
|
indexes,
|
||||||
)
|
fetched,
|
||||||
})
|
starting_indexes,
|
||||||
.unwrap();
|
market,
|
||||||
});
|
&height_to_supply,
|
||||||
});
|
dateindex_to_supply_ref,
|
||||||
|
height_to_realized_cap_ref,
|
||||||
|
dateindex_to_realized_cap_ref,
|
||||||
|
exit,
|
||||||
|
),
|
||||||
|
Either::Right(v) => v.compute_rest_part2(
|
||||||
|
indexer,
|
||||||
|
indexes,
|
||||||
|
fetched,
|
||||||
|
starting_indexes,
|
||||||
|
market,
|
||||||
|
&height_to_supply,
|
||||||
|
dateindex_to_supply_ref,
|
||||||
|
height_to_realized_cap_ref,
|
||||||
|
dateindex_to_realized_cap_ref,
|
||||||
|
exit,
|
||||||
|
),
|
||||||
|
})
|
||||||
|
})?;
|
||||||
|
|
||||||
|
info!("Computing rest part 2 (others)...");
|
||||||
self.indexes_to_unspendable_supply.compute_rest(
|
self.indexes_to_unspendable_supply.compute_rest(
|
||||||
indexer,
|
indexer,
|
||||||
indexes,
|
indexes,
|
||||||
|
|||||||
Reference in New Issue
Block a user