pow: simplify concurrency pipeline

This commit is contained in:
hinto.janai 2024-12-15 21:09:19 -05:00
parent 63378a0e96
commit 1e6f760d72
No known key found for this signature in database
GPG key ID: D47CE05FA175A499
10 changed files with 335 additions and 200 deletions

5
Cargo.lock generated
View file

@ -3335,18 +3335,17 @@ dependencies = [
name = "tests-pow"
version = "0.0.0"
dependencies = [
"crossbeam",
"cuprate-consensus-rules",
"cuprate-cryptonight",
"function_name",
"futures",
"hex",
"hex-literal",
"monero-serai",
"randomx-rs",
"rayon",
"reqwest",
"serde",
"serde_json",
"thread_local",
"tokio",
]

View file

@ -60,10 +60,10 @@ pub const fn is_randomx_seed_height(height: usize) -> bool {
///
/// ref: <https://monero-book.cuprate.org/consensus_rules/blocks.html#randomx-seed>
pub const fn randomx_seed_height(height: usize) -> usize {
if height <= RX_SEEDHASH_EPOCH_BLOCKS + RX_SEEDHASH_EPOCH_LAG {
if height <= 2048 + 64 {
0
} else {
(height - RX_SEEDHASH_EPOCH_LAG - 1) & !(RX_SEEDHASH_EPOCH_BLOCKS - 1)
(height - 64 - 1) & !(2048 - 1)
}
}

View file

@ -11,6 +11,7 @@ serde_json = { workspace = true, features = ["std"] }
tokio = { workspace = true, features = ["full"] }
reqwest = { workspace = true, features = ["json"] }
rayon = { workspace = true }
# rand = { workspace = true, features = ["std", "std_rng"] }
futures = { workspace = true, features = ["std"] }
[lints]

View file

@ -1,6 +1,7 @@
use std::{
collections::{BTreeSet, HashSet},
sync::{atomic::Ordering, LazyLock},
time::Instant,
};
use hex::serde::deserialize;
@ -39,6 +40,8 @@ pub(crate) struct BlockHeader {
pub(crate) struct RpcClient {
client: Client,
rpc_url: String,
json_rpc_url: String,
get_transactions_url: String,
pub top_height: usize,
}
@ -72,8 +75,11 @@ impl RpcClient {
"params": {}
});
let json_rpc_url = format!("{rpc_url}/json_rpc");
let get_transactions_url = format!("{rpc_url}/get_transactions");
let top_height = client
.get(format!("{rpc_url}/json_rpc"))
.get(&json_rpc_url)
.json(&request)
.send()
.await
@ -90,6 +96,8 @@ impl RpcClient {
Self {
client,
rpc_url,
json_rpc_url,
get_transactions_url,
top_height,
}
}
@ -108,20 +116,16 @@ impl RpcClient {
pub pruned_as_hex: String,
}
let url = format!("{}/get_transactions", self.rpc_url);
let txs_hashes = tx_hashes
.into_iter()
.map(hex::encode)
.collect::<Vec<String>>();
let request = json!({
"txs_hashes": txs_hashes,
});
let request = json!({"txs_hashes":txs_hashes});
let txs = self
.client
.get(&url)
.get(&self.get_transactions_url)
.json(&request)
.send()
.await
@ -145,7 +149,11 @@ impl RpcClient {
.collect()
}
#[expect(clippy::significant_drop_tightening)]
#[expect(
clippy::cast_possible_truncation,
clippy::cast_sign_loss,
clippy::significant_drop_tightening
)]
pub(crate) async fn get_block_test_batch(&self, heights: BTreeSet<usize>) {
#[derive(Debug, Clone, Deserialize)]
struct JsonRpcResponse {
@ -159,9 +167,9 @@ impl RpcClient {
pub block_header: BlockHeader,
}
let tasks = heights.into_iter().map(|height| {
let json_rpc_url = format!("{}/json_rpc", self.rpc_url);
let now = Instant::now();
let tasks = heights.into_iter().map(|height| {
let request = json!({
"jsonrpc": "2.0",
"id": 0,
@ -169,7 +177,9 @@ impl RpcClient {
"params": {"height": height}
});
let task = tokio::task::spawn(self.client.get(&json_rpc_url).json(&request).send());
let task =
tokio::task::spawn(self.client.get(&self.json_rpc_url).json(&request).send());
// tokio::task::spawn(self.client.get(&*self.nodes.rand()).json(&request).send());
(height, task)
});
@ -286,8 +296,16 @@ impl RpcClient {
let percent = (progress as f64 / top_height as f64) * 100.0;
let elapsed = now.elapsed().as_secs_f64();
let secs_per_hash = elapsed / progress as f64;
let bps = progress as f64 / elapsed;
let remaining_secs = (top_height as f64 - progress as f64) * secs_per_hash;
let h = (remaining_secs / 60.0 / 60.0) as u64;
let m = (remaining_secs / 60.0 % 60.0) as u64;
let s = (remaining_secs % 60.0) as u64;
println!(
"progress | {progress}/{top_height} ({percent:.2}%)
"progress | {progress}/{top_height} ({percent:.2}%, {bps:.2} blocks/sec, {h}h {m}m {s}s left)
tx_count | {tx_count}
hash | {}
miner_tx_hash | {}

View file

@ -4,20 +4,19 @@ version = "0.0.0"
edition = "2021"
[dependencies]
cuprate-cryptonight = { workspace = true }
cuprate-consensus-rules = { workspace = true }
cuprate-cryptonight = { workspace = true }
function_name = { workspace = true }
thread_local = { workspace = true }
monero-serai = { workspace = true }
hex = { workspace = true, features = ["serde", "std"] }
hex-literal = { workspace = true }
serde = { workspace = true, features = ["derive"] }
serde_json = { workspace = true, features = ["std"] }
tokio = { workspace = true, features = ["full"] }
reqwest = { workspace = true, features = ["json"] }
rayon = { workspace = true }
randomx-rs = { workspace = true }
crossbeam = { workspace = true, features = ["std"] }
futures = { workspace = true, features = ["std"] }
monero-serai = { workspace = true }
hex = { workspace = true, features = ["serde", "std"] }
hex-literal = { workspace = true }
serde = { workspace = true, features = ["derive"] }
serde_json = { workspace = true, features = ["std"] }
tokio = { workspace = true, features = ["full"] }
reqwest = { workspace = true, features = ["json"] }
randomx-rs = { workspace = true }
[lints]
workspace = true

View file

@ -0,0 +1,62 @@
use std::fmt::Display;
use hex_literal::hex;
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
pub(crate) enum CryptoNightHash {
V0,
V1,
V2,
R,
}
impl CryptoNightHash {
/// The last height this hash function is used for proof-of-work.
pub(crate) const fn from_height(height: u64) -> Self {
if height < 1546000 {
Self::V0
} else if height < 1685555 {
Self::V1
} else if height < 1788000 {
Self::V2
} else if height < 1978433 {
Self::R
} else {
panic!("height is large than 1978433");
}
}
pub(crate) fn hash(data: &[u8], height: u64) -> (&'static str, [u8; 32]) {
let this = Self::from_height(height);
let hash = match Self::from_height(height) {
Self::V0 => {
if height == 202612 {
hex!("84f64766475d51837ac9efbef1926486e58563c95a19fef4aec3254f03000000")
} else {
cuprate_cryptonight::cryptonight_hash_v0(data)
}
}
Self::V1 => cuprate_cryptonight::cryptonight_hash_v1(data).unwrap(),
Self::V2 => cuprate_cryptonight::cryptonight_hash_v2(data),
Self::R => cuprate_cryptonight::cryptonight_hash_r(data, height),
};
(this.as_str(), hash)
}
pub(crate) const fn as_str(self) -> &'static str {
match self {
Self::V0 => "cryptonight_v0",
Self::V1 => "cryptonight_v1",
Self::V2 => "cryptonight_v2",
Self::R => "cryptonight_r",
}
}
}
impl Display for CryptoNightHash {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.write_str((*self).as_str())
}
}

View file

@ -1,11 +1,25 @@
mod cryptonight;
mod randomx;
mod rpc;
mod verify;
use std::{
sync::atomic::{AtomicUsize, Ordering},
sync::atomic::{AtomicU64, Ordering},
time::{Duration, Instant},
};
pub static TESTED_BLOCK_COUNT: AtomicUsize = AtomicUsize::new(0);
use crate::rpc::GetBlockResponse;
pub const RANDOMX_START_HEIGHT: u64 = 1978433;
pub static TESTED_BLOCK_COUNT: AtomicU64 = AtomicU64::new(0);
#[derive(Debug)]
pub struct VerifyData {
pub get_block_response: GetBlockResponse,
pub height: u64,
pub seed_height: u64,
pub seed_hash: [u8; 32],
}
#[tokio::main]
async fn main() {
@ -25,32 +39,32 @@ async fn main() {
println!("VERBOSE: false");
}
let mut client = rpc::RpcClient::new(rpc_url).await;
let client = rpc::RpcClient::new(rpc_url).await;
let top_height = client.top_height;
println!("top_height: {top_height}");
let top_height = if let Ok(Ok(h)) = std::env::var("TOP_HEIGHT").map(|s| s.parse()) {
client.top_height = h;
println!("TOP_HEIGHT (found): {h}");
h
let threads = if let Ok(Ok(c)) = std::env::var("THREADS").map(|s| s.parse()) {
println!("THREADS (found): {c}");
c
} else {
println!("TOP_HEIGHT (off, using latest): {}", client.top_height);
client.top_height
let c = std::thread::available_parallelism().unwrap().get();
println!("THREADS (off): {c}");
c
};
println!();
tokio::join!(
client.cryptonight_v0(),
client.cryptonight_v1(),
client.cryptonight_v2(),
client.cryptonight_r(),
client.randomx(),
);
// Test RandomX.
let (tx, rx) = crossbeam::channel::unbounded();
verify::spawn_verify_pool(threads, top_height, rx);
client.test(top_height, tx).await;
// Wait for other threads to finish.
loop {
let count = TESTED_BLOCK_COUNT.load(Ordering::Acquire);
if top_height == count {
println!("finished all PoW, took {}s", now.elapsed().as_secs());
println!("finished, took {}s", now.elapsed().as_secs());
std::process::exit(0);
}

43
tests/pow/src/randomx.rs Normal file
View file

@ -0,0 +1,43 @@
use randomx_rs::{RandomXCache, RandomXDataset, RandomXFlag, RandomXVM};
/// Returns a [`RandomXVM`] with no optimization flags (default, light-verification).
pub(crate) fn randomx_vm_default(seed_hash: &[u8; 32]) -> RandomXVM {
const FLAG: RandomXFlag = RandomXFlag::FLAG_DEFAULT;
let cache = RandomXCache::new(FLAG, seed_hash).unwrap();
RandomXVM::new(FLAG, Some(cache), None).unwrap()
}
/// Returns a [`RandomXVM`] with most optimization flags.
pub(crate) fn randomx_vm_optimized(seed_hash: &[u8; 32]) -> RandomXVM {
// TODO: conditional FLAG_LARGE_PAGES, FLAG_JIT
let mut vm_flag = RandomXFlag::FLAG_FULL_MEM;
let mut cache_flag = RandomXFlag::empty();
#[cfg(target_arch = "x86_64")]
for flag in [&mut vm_flag, &mut cache_flag] {
if is_x86_feature_detected!("aes") {
*flag |= RandomXFlag::FLAG_HARD_AES;
}
match (
is_x86_feature_detected!("ssse3"),
is_x86_feature_detected!("avx2"),
) {
(true, _) => *flag |= RandomXFlag::FLAG_ARGON2_SSSE3,
(_, true) => *flag |= RandomXFlag::FLAG_ARGON2_AVX2,
(_, _) => *flag |= RandomXFlag::FLAG_ARGON2,
}
}
let hash = hex::encode(seed_hash);
println!("Generating RandomX VM: seed_hash: {hash}, flags: {vm_flag:#?}");
let cache = RandomXCache::new(cache_flag, seed_hash).unwrap();
let dataset = RandomXDataset::new(RandomXFlag::FLAG_DEFAULT, cache, 0).unwrap();
let vm = RandomXVM::new(vm_flag, None, Some(dataset)).unwrap();
println!("Generating RandomX VM: seed_hash: {hash}, flags: {vm_flag:#?} ... OK");
vm
}

View file

@ -1,23 +1,15 @@
use std::{
collections::BTreeMap,
ops::Range,
sync::{atomic::Ordering, Mutex},
};
use std::time::Duration;
use function_name::named;
use crossbeam::channel::Sender;
use hex::serde::deserialize;
use hex_literal::hex;
use monero_serai::block::Block;
use randomx_rs::{RandomXCache, RandomXFlag, RandomXVM};
use reqwest::{
header::{HeaderMap, HeaderValue},
Client, ClientBuilder,
};
use serde::Deserialize;
use serde_json::{json, Value};
use thread_local::ThreadLocal;
use crate::TESTED_BLOCK_COUNT;
use crate::{VerifyData, RANDOMX_START_HEIGHT};
#[derive(Debug, Clone, Deserialize)]
struct JsonRpcResponse {
@ -25,14 +17,14 @@ struct JsonRpcResponse {
}
#[derive(Debug, Clone, Deserialize)]
struct GetBlockResponse {
pub struct GetBlockResponse {
#[serde(deserialize_with = "deserialize")]
pub blob: Vec<u8>,
pub block_header: BlockHeader,
}
#[derive(Debug, Clone, Deserialize)]
struct BlockHeader {
pub struct BlockHeader {
#[serde(deserialize_with = "deserialize")]
pub pow_hash: Vec<u8>,
#[serde(deserialize_with = "deserialize")]
@ -42,8 +34,8 @@ struct BlockHeader {
#[derive(Debug, Clone)]
pub(crate) struct RpcClient {
client: Client,
rpc_url: String,
pub top_height: usize,
json_rpc_url: String,
pub top_height: u64,
}
impl RpcClient {
@ -66,8 +58,10 @@ impl RpcClient {
"params": {}
});
let json_rpc_url = format!("{rpc_url}/json_rpc");
let top_height = client
.get(format!("{rpc_url}/json_rpc"))
.get(&json_rpc_url)
.json(&request)
.send()
.await
@ -82,20 +76,18 @@ impl RpcClient {
.get("height")
.unwrap()
.as_u64()
.unwrap()
.try_into()
.unwrap();
assert!(top_height > 3301441, "node is behind");
Self {
client,
rpc_url,
json_rpc_url,
top_height,
}
}
async fn get_block(&self, height: usize) -> GetBlockResponse {
async fn get_block(&self, height: u64) -> GetBlockResponse {
let request = json!({
"jsonrpc": "2.0",
"id": 0,
@ -103,160 +95,62 @@ impl RpcClient {
"params": {"height": height, "fill_pow_hash": true}
});
let rpc_url = format!("{}/json_rpc", self.rpc_url);
tokio::task::spawn(self.client.get(rpc_url).json(&request).send())
self.client
.get(&self.json_rpc_url)
.json(&request)
.send()
.await
.unwrap()
.unwrap()
.json::<JsonRpcResponse>()
.await
.unwrap()
.result
}
async fn test<const RANDOMX: bool, const CRYPTONIGHT_V0: bool>(
&self,
range: Range<usize>,
hash: impl Fn(Vec<u8>, u64, u64, [u8; 32]) -> [u8; 32] + Send + Sync + 'static + Copy,
name: &'static str,
) {
let tasks = range.map(|height| {
let task = self.get_block(height);
(height, task)
});
pub(crate) async fn test(self, top_height: u64, tx: Sender<VerifyData>) {
use futures::StreamExt;
for (height, task) in tasks {
let result = task.await;
let iter = (0..top_height).map(|height| {
let this = &self;
let tx = tx.clone();
let (seed_height, seed_hash) = if RANDOMX {
let seed_height = cuprate_consensus_rules::blocks::randomx_seed_height(height);
async move {
let get_block_response = this.get_block(height).await;
let seed_hash: [u8; 32] = self
.get_block(seed_height)
.await
.block_header
.hash
let (seed_height, seed_hash) = if height < RANDOMX_START_HEIGHT {
(0, [0; 32])
} else {
let seed_height = cuprate_consensus_rules::blocks::randomx_seed_height(
height.try_into().unwrap(),
)
.try_into()
.unwrap();
(seed_height, seed_hash)
} else {
(0, [0; 32])
};
let seed_hash = this
.get_block(seed_height)
.await
.block_header
.hash
.try_into()
.unwrap();
let top_height = self.top_height;
#[expect(clippy::cast_precision_loss)]
rayon::spawn(move || {
let GetBlockResponse { blob, block_header } = result;
let header = block_header;
let block = match Block::read(&mut blob.as_slice()) {
Ok(b) => b,
Err(e) => panic!("{e:?}\nblob: {blob:?}, header: {header:?}"),
(seed_height, seed_hash)
};
let pow_hash = if CRYPTONIGHT_V0 && height == 202612 {
hex!("84f64766475d51837ac9efbef1926486e58563c95a19fef4aec3254f03000000")
} else {
hash(
block.serialize_pow_hash(),
height.try_into().unwrap(),
seed_height.try_into().unwrap(),
seed_hash,
)
let data = VerifyData {
get_block_response,
height,
seed_height,
seed_hash,
};
assert_eq!(
header.pow_hash, pow_hash,
"\nheight: {height}\nheader: {header:#?}\nblock: {block:#?}"
);
tx.send(data).unwrap();
}
});
let count = TESTED_BLOCK_COUNT.fetch_add(1, Ordering::Release) + 1;
if std::env::var("VERBOSE").is_err() && count % 500 != 0 {
return;
}
let hash = hex::encode(pow_hash);
let percent = (count as f64 / top_height as f64) * 100.0;
println!(
"progress | {count}/{top_height} ({percent:.2}%)
height | {height}
algo | {name}
hash | {hash}\n"
);
});
}
}
#[named]
pub(crate) async fn cryptonight_v0(&self) {
self.test::<false, true>(
0..1546000,
|b, _, _, _| cuprate_cryptonight::cryptonight_hash_v0(&b),
function_name!(),
)
.await;
}
#[named]
pub(crate) async fn cryptonight_v1(&self) {
self.test::<false, false>(
1546000..1685555,
|b, _, _, _| cuprate_cryptonight::cryptonight_hash_v1(&b).unwrap(),
function_name!(),
)
.await;
}
#[named]
pub(crate) async fn cryptonight_v2(&self) {
self.test::<false, false>(
1685555..1788000,
|b, _, _, _| cuprate_cryptonight::cryptonight_hash_v2(&b),
function_name!(),
)
.await;
}
#[named]
pub(crate) async fn cryptonight_r(&self) {
self.test::<false, false>(
1788000..1978433,
|b, h, _, _| cuprate_cryptonight::cryptonight_hash_r(&b, h),
function_name!(),
)
.await;
}
#[named]
pub(crate) async fn randomx(&self) {
#[expect(clippy::significant_drop_tightening)]
let function = move |bytes: Vec<u8>, _, seed_height, seed_hash: [u8; 32]| {
static RANDOMX_VM: ThreadLocal<Mutex<BTreeMap<u64, RandomXVM>>> = ThreadLocal::new();
let mut thread_local = RANDOMX_VM
.get_or(|| Mutex::new(BTreeMap::new()))
.lock()
.unwrap();
let randomx_vm = thread_local.entry(seed_height).or_insert_with(|| {
let flag = RandomXFlag::get_recommended_flags();
let cache = RandomXCache::new(flag, &seed_hash).unwrap();
RandomXVM::new(flag, Some(cache), None).unwrap()
});
randomx_vm
.calculate_hash(&bytes)
.unwrap()
.try_into()
.unwrap()
};
self.test::<true, false>(1978433..self.top_height, function, function_name!())
futures::stream::iter(iter)
.buffer_unordered(4) // This can't be too high or else we get bottlenecked by `monerod`
.for_each(|()| async {})
.await;
}
}

105
tests/pow/src/verify.rs Normal file
View file

@ -0,0 +1,105 @@
use std::{sync::atomic::Ordering, time::Instant};
use crossbeam::channel::Receiver;
use monero_serai::block::Block;
use crate::{
cryptonight::CryptoNightHash, rpc::GetBlockResponse, VerifyData, RANDOMX_START_HEIGHT,
TESTED_BLOCK_COUNT,
};
#[expect(
clippy::needless_pass_by_value,
clippy::cast_precision_loss,
clippy::cast_possible_truncation,
clippy::cast_sign_loss
)]
pub(crate) fn spawn_verify_pool(thread_count: usize, top_height: u64, rx: Receiver<VerifyData>) {
let now = Instant::now();
for i in 0..thread_count {
let rx = rx.clone();
std::thread::spawn(move || {
let mut current_seed_hash = [0; 32];
let mut randomx_vm = None;
loop {
let Ok(data) = rx.recv() else {
println!("Exiting verify thread {i}/{thread_count}");
return;
};
let VerifyData {
get_block_response,
height,
seed_height,
seed_hash,
} = data;
let GetBlockResponse { blob, block_header } = get_block_response;
let header = block_header;
let block = match Block::read(&mut blob.as_slice()) {
Ok(b) => b,
Err(e) => panic!("{e:?}\nblob: {blob:?}, header: {header:?}"),
};
let pow_data = block.serialize_pow_hash();
let (algo, pow_hash) = if height < RANDOMX_START_HEIGHT {
CryptoNightHash::hash(&pow_data, height)
} else {
if current_seed_hash != seed_hash {
randomx_vm = None;
}
let randomx_vm = randomx_vm.get_or_insert_with(|| {
current_seed_hash = seed_hash;
// crate::randomx::randomx_vm_optimized(&seed_hash)
crate::randomx::randomx_vm_default(&seed_hash)
});
let pow_hash = randomx_vm
.calculate_hash(&pow_data)
.unwrap()
.try_into()
.unwrap();
("randomx", pow_hash)
};
assert_eq!(
header.pow_hash, pow_hash,
"\nheight: {height}\nheader: {header:#?}\nblock: {block:#?}",
);
let count = TESTED_BLOCK_COUNT.fetch_add(1, Ordering::Release) + 1;
if std::env::var("VERBOSE").is_err() && count % 500 != 0 {
continue;
}
let pow_hash = hex::encode(pow_hash);
let seed_hash = hex::encode(seed_hash);
let percent = (count as f64 / top_height as f64) * 100.0;
let elapsed = now.elapsed().as_secs_f64();
let secs_per_hash = elapsed / count as f64;
let bps = count as f64 / elapsed;
let remaining_secs = (top_height as f64 - count as f64) * secs_per_hash;
let h = (remaining_secs / 60.0 / 60.0) as u64;
let m = (remaining_secs / 60.0 % 60.0) as u64;
let s = (remaining_secs % 60.0) as u64;
println!(
"progress | {count}/{top_height} ({percent:.2}%, {bps:.2} blocks/sec, {h}h {m}m {s}s left)
algo | {algo}
seed_height | {seed_height}
seed_hash | {seed_hash}
pow_hash | {pow_hash}\n"
);
}
});
}
}