Compare commits

..

2 commits

Author SHA1 Message Date
hinto.janai
ce6807546c
fixes 2024-12-17 17:45:53 -05:00
hinto.janai
1bc5e4721c
more cli args 2024-12-16 22:14:51 -05:00
4 changed files with 98 additions and 106 deletions

View file

@ -17,15 +17,25 @@ pub struct Args {
/// Base URL to use for `monerod` RPC. /// Base URL to use for `monerod` RPC.
/// ///
/// This must be a non-restricted RPC. /// This must be a non-restricted RPC.
#[arg(short, long, default_value_t = String::from("http://127.0.0.1:18081"))] #[arg(long, default_value_t = String::from("http://127.0.0.1:18081"))]
pub rpc_url: String, pub rpc_url: String,
/// Amount of async RPC tasks to spawn.
#[arg(long, default_value_t = NonZeroUsize::new(4).unwrap())]
pub rpc_tasks: NonZeroUsize,
/// The maximum capacity of the block buffer in-between the RPC and verifier.
///
/// `0` will cause the buffer to be unbounded.
#[arg(long, default_value_t = 1000)]
pub buffer_limit: usize,
/// Amount of verifying threads to spawn. /// Amount of verifying threads to spawn.
#[arg(short, long, default_value_t = std::thread::available_parallelism().unwrap())] #[arg(long, default_value_t = std::thread::available_parallelism().unwrap())]
pub threads: NonZeroUsize, pub threads: NonZeroUsize,
/// Print an update every `update` amount of blocks. /// Print an update every `update` amount of blocks.
#[arg(short, long, default_value_t = NonZeroU64::new(500).unwrap())] #[arg(long, default_value_t = NonZeroU64::new(500).unwrap())]
pub update: NonZeroU64, pub update: NonZeroU64,
} }

View file

@ -21,17 +21,23 @@ async fn main() {
let cli::Args { let cli::Args {
rpc_url, rpc_url,
update, update,
rpc_tasks,
buffer_limit,
threads, threads,
} = cli::Args::get(); } = cli::Args::get();
// Set-up RPC client. // Set-up RPC client.
let client = rpc::RpcClient::new(rpc_url).await; let client = rpc::RpcClient::new(rpc_url, rpc_tasks).await;
let top_height = client.top_height; let top_height = client.top_height;
println!("top_height: {top_height}"); println!("top_height: {top_height}");
println!(); println!();
// Test. // Test.
let (tx, rx) = crossbeam::channel::unbounded(); let (tx, rx) = if buffer_limit == 0 {
crossbeam::channel::unbounded()
} else {
crossbeam::channel::bounded(buffer_limit)
};
verify::spawn_verify_pool(threads, update, top_height, rx); verify::spawn_verify_pool(threads, update, top_height, rx);
client.test(top_height, tx).await; client.test(top_height, tx).await;

View file

@ -1,4 +1,4 @@
use std::time::Instant; use std::{num::NonZeroUsize, time::Instant};
use crossbeam::channel::Sender; use crossbeam::channel::Sender;
use monero_serai::{block::Block, transaction::Transaction}; use monero_serai::{block::Block, transaction::Transaction};
@ -20,11 +20,12 @@ pub struct RpcClient {
client: Client, client: Client,
json_rpc_url: String, json_rpc_url: String,
get_transactions_url: String, get_transactions_url: String,
rpc_tasks: NonZeroUsize,
pub top_height: u64, pub top_height: u64,
} }
impl RpcClient { impl RpcClient {
pub async fn new(rpc_url: String) -> Self { pub async fn new(rpc_url: String, rpc_tasks: NonZeroUsize) -> Self {
let headers = { let headers = {
let mut h = HeaderMap::new(); let mut h = HeaderMap::new();
h.insert("Content-Type", HeaderValue::from_static("application/json")); h.insert("Content-Type", HeaderValue::from_static("application/json"));
@ -70,6 +71,7 @@ impl RpcClient {
client, client,
json_rpc_url, json_rpc_url,
get_transactions_url, get_transactions_url,
rpc_tasks,
top_height, top_height,
} }
} }
@ -207,7 +209,7 @@ impl RpcClient {
}); });
futures::stream::iter(iter) futures::stream::iter(iter)
.buffer_unordered(4) // This can't be too high or else we get bottlenecked by `monerod` .buffer_unordered(self.rpc_tasks.get()) // This can't be too high or else we get bottlenecked by `monerod`
.for_each(|()| async {}) .for_each(|()| async {})
.await; .await;
} }

View file

@ -17,7 +17,7 @@ use crate::{
struct Verifier { struct Verifier {
id: usize, id: usize,
now: Instant, start_of_new_pow: Instant,
thread_count: NonZeroUsize, thread_count: NonZeroUsize,
update: NonZeroU64, update: NonZeroU64,
top_height: u64, top_height: u64,
@ -33,14 +33,12 @@ pub fn spawn_verify_pool(
top_height: u64, top_height: u64,
rx: Receiver<RpcBlockData>, rx: Receiver<RpcBlockData>,
) { ) {
let now = Instant::now(); for id in 1..=thread_count.get() {
for id in 0..thread_count.get() {
let rx = rx.clone(); let rx = rx.clone();
std::thread::spawn(move || { std::thread::spawn(move || {
Verifier { Verifier {
id, id,
now, start_of_new_pow: Instant::now(),
thread_count, thread_count,
update, update,
top_height, top_height,
@ -80,39 +78,14 @@ impl Verifier {
} = data; } = data;
let GetBlockResponse { blob, block_header } = get_block_response; let GetBlockResponse { blob, block_header } = get_block_response;
//----------------------------------------------- Calculate some data.
let calculated_block_reward = block
.miner_transaction
.prefix()
.outputs
.iter()
.map(|o| o.amount.unwrap())
.sum::<u64>();
let calculated_block_weight = txs
.iter()
.map(|RpcTxData { tx, .. }| tx.weight())
.sum::<usize>();
let calculated_pow_data = block.serialize_pow_hash();
let miner_tx_weight = block.miner_transaction.weight();
//----------------------------------------------- Verify. //----------------------------------------------- Verify.
Self::verify_block_properties(&blob, &block, calculated_block_reward, &p); Self::verify_blocks(&blob, &block, &txs, &p, block_header);
Self::verify_all_transactions_are_unique(&txs, &p); Self::verify_transactions(txs, &p);
Self::verify_transaction_properties(txs, &p);
Self::verify_block_fields(
calculated_block_weight,
calculated_block_reward,
&block,
&p,
block_header,
);
let algo = self.verify_pow( let algo = self.verify_pow(
block_header.height, block_header.height,
seed_hash, seed_hash,
block_header.pow_hash, block_header.pow_hash,
&calculated_pow_data, &block.serialize_pow_hash(),
&p, &p,
); );
@ -120,27 +93,79 @@ impl Verifier {
self.print_progress( self.print_progress(
algo, algo,
seed_height, seed_height,
miner_tx_weight, block.miner_transaction.weight(),
blocks_per_sec, blocks_per_sec,
block_header, block_header,
); );
} }
fn verify_block_properties( fn verify_blocks(
block_blob: &[u8], block_blob: &[u8],
block: &Block, block: &Block,
calculated_block_reward: u64, txs: &[RpcTxData],
p: &str, p: &str,
BlockHeader {
block_weight,
hash,
pow_hash: _,
height,
major_version,
minor_version,
miner_tx_hash,
nonce,
num_txes,
prev_hash,
reward,
timestamp,
}: BlockHeader,
) { ) {
// Test block properties. let calculated_block_reward = block
.miner_transaction
.prefix()
.outputs
.iter()
.map(|o| o.amount.unwrap())
.sum::<u64>();
let calculated_block_weight = txs
.iter()
.map(|RpcTxData { tx, .. }| tx.weight())
.sum::<usize>();
let block_number = u64::try_from(block.number().unwrap()).unwrap();
assert!(!block.miner_transaction.prefix().outputs.is_empty(), "{p}");
assert_ne!(calculated_block_reward, 0, "{p}");
assert_ne!(block.miner_transaction.weight(), 0, "{p}");
assert_eq!(block_blob, block.serialize(), "{p}"); assert_eq!(block_blob, block.serialize(), "{p}");
assert_eq!(block_weight, calculated_block_weight, "{p}");
assert_eq!(hash, block.hash(), "{p}");
assert_eq!(height, block_number, "{p}");
assert_eq!(major_version, block.header.hardfork_version, "{p}");
assert_eq!(minor_version, block.header.hardfork_signal, "{p}");
assert_eq!(miner_tx_hash, block.miner_transaction.hash(), "{p}");
assert_eq!(nonce, block.header.nonce, "{p}");
assert_eq!(num_txes, block.transactions.len(), "{p}");
assert_eq!(prev_hash, block.header.previous, "{p}");
assert_eq!(reward, calculated_block_reward, "{p}");
assert_eq!(timestamp, block.header.timestamp, "{p}");
}
assert!( fn verify_transactions(txs: Vec<RpcTxData>, p: &str) {
!block.miner_transaction.prefix().outputs.is_empty(), Self::verify_all_transactions_are_unique(&txs, p);
"miner_tx has no outputs\n{p}"
);
assert_ne!(calculated_block_reward, 0, "block reward is 0\n{p}"); for RpcTxData {
tx,
tx_blob,
tx_hash,
} in txs
{
assert_eq!(tx_hash, tx.hash(), "{p}, tx: {tx:#?}");
assert_ne!(tx.weight(), 0, "{p}, tx: {tx:#?}");
assert!(!tx.prefix().inputs.is_empty(), "{p}, tx: {tx:#?}");
assert_eq!(tx_blob, tx.serialize(), "{p}, tx: {tx:#?}");
assert!(matches!(tx.version(), 1 | 2), "{p}, tx: {tx:#?}");
}
} }
#[expect(clippy::significant_drop_tightening)] #[expect(clippy::significant_drop_tightening)]
@ -159,61 +184,6 @@ impl Verifier {
} }
} }
fn verify_transaction_properties(txs: Vec<RpcTxData>, p: &str) {
// Test transaction properties.
for RpcTxData {
tx,
tx_blob,
tx_hash,
} in txs
{
assert_eq!(tx_hash, tx.hash(), "{p}, tx: {tx:#?}");
assert_ne!(tx.weight(), 0, "{p}, tx: {tx:#?}");
assert!(!tx.prefix().inputs.is_empty(), "{p}, tx: {tx:#?}");
assert_eq!(tx_blob, tx.serialize(), "{p}, tx: {tx:#?}");
assert!(matches!(tx.version(), 1 | 2), "{p}, tx: {tx:#?}");
}
}
fn verify_block_fields(
calculated_block_weight: usize,
calculated_block_reward: u64,
block: &Block,
p: &str,
BlockHeader {
block_weight,
hash,
pow_hash: _,
height,
major_version,
minor_version,
miner_tx_hash,
nonce,
num_txes,
prev_hash,
reward,
timestamp,
}: BlockHeader,
) {
// Test block fields are correct.
assert_eq!(block_weight, calculated_block_weight, "{p}");
assert_ne!(block.miner_transaction.weight(), 0, "{p}");
assert_eq!(hash, block.hash(), "{p}");
assert_eq!(
height,
u64::try_from(block.number().unwrap()).unwrap(),
"{p}"
);
assert_eq!(major_version, block.header.hardfork_version, "{p}");
assert_eq!(minor_version, block.header.hardfork_signal, "{p}");
assert_eq!(miner_tx_hash, block.miner_transaction.hash(), "{p}");
assert_eq!(nonce, block.header.nonce, "{p}");
assert_eq!(num_txes, block.transactions.len(), "{p}");
assert_eq!(prev_hash, block.header.previous, "{p}");
assert_eq!(reward, calculated_block_reward, "{p}");
assert_eq!(timestamp, block.header.timestamp, "{p}");
}
fn verify_pow( fn verify_pow(
&mut self, &mut self,
height: u64, height: u64,
@ -222,6 +192,10 @@ impl Verifier {
calculated_pow_data: &[u8], calculated_pow_data: &[u8],
p: &str, p: &str,
) -> &'static str { ) -> &'static str {
if matches!(height, 1546000 | 1685555 | 1788000 | 1978433) {
self.start_of_new_pow = Instant::now();
}
let (algo, calculated_pow_hash) = if height < RANDOMX_START_HEIGHT { let (algo, calculated_pow_hash) = if height < RANDOMX_START_HEIGHT {
CryptoNightHash::hash(calculated_pow_data, height) CryptoNightHash::hash(calculated_pow_data, height)
} else { } else {
@ -301,7 +275,7 @@ impl Verifier {
let relative_count = find_count_relative_to_pow_activation(height); let relative_count = find_count_relative_to_pow_activation(height);
let block_buffer = self.rx.len(); let block_buffer = self.rx.len();
let elapsed = self.now.elapsed().as_secs_f64(); let elapsed = self.start_of_new_pow.elapsed().as_secs_f64();
let secs_per_hash = elapsed / relative_count as f64; let secs_per_hash = elapsed / relative_count as f64;
let verify_bps = relative_count as f64 / elapsed; let verify_bps = relative_count as f64 / elapsed;
let remaining_secs = (top_height as f64 - relative_count as f64) * secs_per_hash; let remaining_secs = (top_height as f64 - relative_count as f64) * secs_per_hash;