-
Notifications
You must be signed in to change notification settings - Fork 58
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Ludo Galabru
committed
Aug 2, 2023
1 parent
c0991c5
commit 13421db
Showing
6 changed files
with
193 additions
and
30 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
104 changes: 104 additions & 0 deletions
104
components/hord-cli/src/core/pipeline/processors/block_ingestion.rs
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,104 @@ | ||
| use std::{ | ||
| sync::mpsc::Sender, | ||
| thread::{sleep, JoinHandle}, | ||
| time::Duration, | ||
| }; | ||
|
|
||
| use chainhook_sdk::{types::BitcoinBlockData, utils::Context}; | ||
| use crossbeam_channel::TryRecvError; | ||
|
|
||
| use crate::{ | ||
| config::Config, | ||
| core::pipeline::{PostProcessorCommand, PostProcessorController, PostProcessorEvent}, | ||
| db::{insert_entry_in_blocks, open_readwrite_hord_db_conn_rocks_db}, | ||
| }; | ||
|
|
||
| pub fn start_block_ingestion_processor( | ||
| config: &Config, | ||
| ctx: &Context, | ||
| _post_processor: Option<Sender<BitcoinBlockData>>, | ||
| ) -> PostProcessorController { | ||
| let (commands_tx, commands_rx) = crossbeam_channel::bounded::<PostProcessorCommand>(2); | ||
| let (events_tx, events_rx) = crossbeam_channel::unbounded::<PostProcessorEvent>(); | ||
|
|
||
| let config = config.clone(); | ||
| let ctx = ctx.clone(); | ||
| let handle: JoinHandle<()> = hiro_system_kit::thread_named("Processor Runloop") | ||
| .spawn(move || { | ||
| let mut num_writes = 0; | ||
| let blocks_db_rw = | ||
| open_readwrite_hord_db_conn_rocks_db(&config.expected_cache_path(), &ctx).unwrap(); | ||
|
|
||
| let mut empty_cycles = 0; | ||
|
|
||
| loop { | ||
| let blocks = match commands_rx.try_recv() { | ||
| Ok(PostProcessorCommand::ProcessBlocks(blocks)) => blocks, | ||
| Ok(PostProcessorCommand::Terminate) => break, | ||
| Err(e) => match e { | ||
| TryRecvError::Empty => { | ||
| empty_cycles += 1; | ||
|
|
||
| if empty_cycles == 30 { | ||
| let _ = events_tx.send(PostProcessorEvent::EmptyQueue); | ||
| } | ||
| sleep(Duration::from_secs(1)); | ||
| if empty_cycles > 120 { | ||
| break; | ||
| } | ||
| continue; | ||
| } | ||
| _ => { | ||
| break; | ||
| } | ||
| }, | ||
| }; | ||
|
|
||
| info!(ctx.expect_logger(), "Storing {} blocks", blocks.len()); | ||
|
|
||
| for (block, compacted_block) in blocks.into_iter() { | ||
| insert_entry_in_blocks( | ||
| block.block_identifier.index as u32, | ||
| &compacted_block, | ||
| &blocks_db_rw, | ||
| &ctx, | ||
| ); | ||
| num_writes += 1; | ||
| } | ||
|
|
||
| // Early return | ||
| if num_writes % 128 == 0 { | ||
| ctx.try_log(|logger| { | ||
| info!(logger, "Flushing DB to disk ({num_writes} inserts)"); | ||
| }); | ||
| if let Err(e) = blocks_db_rw.flush() { | ||
| ctx.try_log(|logger| { | ||
| error!(logger, "{}", e.to_string()); | ||
| }); | ||
| } | ||
| num_writes = 0; | ||
| continue; | ||
| } | ||
|
|
||
| // Write blocks to disk, before traversals | ||
| if let Err(e) = blocks_db_rw.flush() { | ||
| ctx.try_log(|logger| { | ||
| error!(logger, "{}", e.to_string()); | ||
| }); | ||
| } | ||
| } | ||
|
|
||
| if let Err(e) = blocks_db_rw.flush() { | ||
| ctx.try_log(|logger| { | ||
| error!(logger, "{}", e.to_string()); | ||
| }); | ||
| } | ||
| }) | ||
| .expect("unable to spawn thread"); | ||
|
|
||
| PostProcessorController { | ||
| commands_tx, | ||
| events_rx, | ||
| thread_handle: handle, | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,3 +1,4 @@ | ||
| pub mod block_ingestion; | ||
| pub mod inscription_indexing; | ||
|
|
||
| pub use inscription_indexing::start_ordinals_number_processor; | ||
| pub use inscription_indexing::start_inscription_indexing_processor; |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters