From bba620f42bb67c9e0c4ee66030a0dd3caeed2513 Mon Sep 17 00:00:00 2001 From: Andrei Sandu Date: Thu, 17 Jul 2025 17:41:59 +0300 Subject: [PATCH 1/9] Support 12 cores and 65K TPS with batching. Signed-off-by: Andrei Sandu --- Cargo.lock | 16 ++-- Cargo.toml | 2 +- stps/src/main.rs | 144 ++++++++++++++++++++--------------- utils/sender/Cargo.toml | 4 +- utils/sender/src/main.rs | 157 ++++++++++++++++++++++++++++++++------- 5 files changed, 225 insertions(+), 98 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 956b3d4..b2a2fb3 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3510,7 +3510,7 @@ dependencies = [ "env_logger 0.10.2", "log", "serde_json", - "subxt 0.37.0", + "subxt 0.38.1", ] [[package]] @@ -9924,12 +9924,12 @@ dependencies = [ "frame-system 31.0.0", "funder", "futures", - "jsonrpsee-client-transport 0.22.4", - "jsonrpsee-core 0.22.4", + "jsonrpsee-client-transport 0.24.7", + "jsonrpsee-core 0.24.7", "log", "pallet-balances 31.0.0", "parity-scale-codec", - "subxt 0.37.0", + "subxt 0.38.1", "tokio", "url", ] @@ -12258,9 +12258,9 @@ dependencies = [ [[package]] name = "subxt" -version = "0.38.0" +version = "0.38.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c53029d133e4e0cb7933f1fe06f2c68804b956de9bb8fa930ffca44e9e5e4230" +checksum = "1c17d7ec2359d33133b63c97e28c8b7cd3f0a5bc6ce567ae3aef9d9e85be3433" dependencies = [ "async-trait", "derive-where", @@ -14370,7 +14370,7 @@ dependencies = [ "serde_json", "sha2 0.10.8", "sp-core 31.0.0", - "subxt 0.38.0", + "subxt 0.38.1", "subxt-signer", "thiserror", "tokio", @@ -14430,7 +14430,7 @@ dependencies = [ "async-trait", "futures", "lazy_static", - "subxt 0.38.0", + "subxt 0.38.1", "tokio", "zombienet-configuration", "zombienet-orchestrator", diff --git a/Cargo.toml b/Cargo.toml index b7cdcc8..de812ab 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -18,7 +18,7 @@ default-members = ["stps"] sp-core = "31.0.0" sp-runtime = "32.0.0" sp-keyring = "32.0.0" -subxt = { version = "0.37", features = ["substrate-compat"] } +subxt = { version = "0.38.1", features = ["substrate-compat"] } frame-system = "31.0.0" pallet-balances = "31.0.0" reqwest = "0.11" diff --git a/stps/src/main.rs b/stps/src/main.rs index 8890e1d..607468c 100644 --- a/stps/src/main.rs +++ b/stps/src/main.rs @@ -4,23 +4,23 @@ // use futures::stream::FuturesUnordered; // use std::collections::HashMap; // use subxt::config::substrate::AccountId32; -use futures::StreamExt; -use tokio::sync::mpsc::{self, Sender, UnboundedSender}; -use std::collections::HashMap; -use clap::ValueEnum; -use subxt::tx::PairSigner; -use futures::stream::FuturesUnordered; -use clap::Parser; +use clap::{Parser, ValueEnum}; +use futures::{stream::FuturesUnordered, StreamExt}; use jsonrpsee_client_transport::ws::WsTransportClientBuilder; use jsonrpsee_core::client::Client; use parity_scale_codec::{Compact, Decode}; use serde_json::json; -use std::{cmp::max, error::Error, sync::Arc, time::Duration}; +use std::{cmp::max, collections::HashMap, error::Error, sync::Arc, time::Duration}; use subxt::{ - backend::legacy::LegacyBackend, config::DefaultExtrinsicParamsBuilder, ext::sp_core::crypto::{Pair as _, Ss58Codec}, OnlineClient, PolkadotConfig + backend::legacy::LegacyBackend, + config::DefaultExtrinsicParamsBuilder, + dynamic::Value as TxValue, + ext::sp_core::crypto::{Pair as _, Ss58Codec}, + tx::PairSigner, + OnlineClient, PolkadotConfig, }; +use tokio::sync::mpsc::{self, Sender, UnboundedSender}; use zombienet_sdk::{NetworkConfigBuilder, NetworkConfigExt, NetworkNode, RegistrationStrategy}; -use subxt::{dynamic::Value as TxValue}; mod metrics; use metrics::*; @@ -252,7 +252,7 @@ async fn block_subscriber( sender.send(FinalizedEvent::NftMinted).expect("Sender sends"); } }, - _ => () + _ => (), } } @@ -309,7 +309,10 @@ async fn main() -> Result<(), Box> { let ntrans = nthreads * args.count; let naccs = if let Some(accounts) = args.accounts { - assert!(accounts >= ntrans, "Number of accounts specified is less than the number of transactions"); + assert!( + accounts >= ntrans, + "Number of accounts specified is less than the number of transactions" + ); accounts } else { ntrans @@ -466,7 +469,10 @@ async fn main() -> Result<(), Box> { log::info!("Signing {} transactions...", send_accs.len()); let txs = match args.mode { BenchMode::Stps => { - sender_lib::sign_balance_transfers(api.clone(), send_accs.into_iter().map(|a| (a, 0)).zip(recv_accs.into_iter()))? + sender_lib::sign_balance_transfers( + api.clone(), + send_accs.into_iter().map(|a| (a, 0)).zip(recv_accs.into_iter()), + )? // let api = api.clone(); // sender_lib::sign_txs(send_accs.into_iter().zip(recv_accs.into_iter()), move |(sender, receiver)| { // let signer = EthereumSigner::from(sender); @@ -485,34 +491,42 @@ async fn main() -> Result<(), Box> { }, BenchMode::NftTransfer => { let api2 = api.clone(); - let create_coll_txs = sender_lib::sign_txs(send_accs.clone().into_iter(), move |sender| { - let tx_params = DefaultExtrinsicParamsBuilder::new().nonce(0).build(); - let tx_call = subxt::dynamic::tx( - "Nfts", - "create", - vec![ - TxValue::unnamed_variant("Id", [ TxValue::from_bytes(sender.public()) ]), - TxValue::named_composite( - vec![ + let create_coll_txs = + sender_lib::sign_txs(send_accs.clone().into_iter(), move |sender| { + let tx_params = DefaultExtrinsicParamsBuilder::new().nonce(0).build(); + let tx_call = subxt::dynamic::tx( + "Nfts", + "create", + vec![ + TxValue::unnamed_variant("Id", [TxValue::from_bytes(sender.public())]), + TxValue::named_composite(vec![ ("settings", TxValue::primitive(0u64.into())), ("max_supply", TxValue::unnamed_variant("None", vec![])), - ("mint_settings", TxValue::named_composite(vec![ - ("mint_type", TxValue::unnamed_variant("Issuer", vec![])), - ("price", TxValue::unnamed_variant("None", vec![])), - ("start_block", TxValue::unnamed_variant("None", vec![])), - ("end_block", TxValue::unnamed_variant("None", vec![])), - ("default_item_settings", TxValue::primitive(0u64.into())), - ])), - ] - ) - ] - ); - api2.tx().create_signed_offline(&tx_call, &PairSigner::new(sender), tx_params) - })?; - let futs = create_coll_txs.iter().map(|tx| tx.submit()).collect::>(); + ( + "mint_settings", + TxValue::named_composite(vec![ + ("mint_type", TxValue::unnamed_variant("Issuer", vec![])), + ("price", TxValue::unnamed_variant("None", vec![])), + ("start_block", TxValue::unnamed_variant("None", vec![])), + ("end_block", TxValue::unnamed_variant("None", vec![])), + ("default_item_settings", TxValue::primitive(0u64.into())), + ]), + ), + ]), + ], + ); + api2.tx().create_signed_offline(&tx_call, &PairSigner::new(sender), tx_params) + })?; + let futs = + create_coll_txs.iter().map(|tx| tx.submit()).collect::>(); // let futs = create_coll_txs.iter().map(|tx| tx.submit_and_watch()).collect::>(); // let res = futs.collect::>().await.into_iter().collect::, _>>().expect("All the transactions submitted successfully"); - let _res = futs.collect::>().await.into_iter().collect::, _>>().expect("All the transactions submitted successfully"); + let _res = futs + .collect::>() + .await + .into_iter() + .collect::, _>>() + .expect("All the transactions submitted successfully"); // let waiter = res.into_iter().map(|txp| txp.wait_for_finalized_success()).collect::>(); // let res = waiter.collect::>().await.into_iter().collect::, _>>().expect("All the collection creation transaction finalized"); @@ -560,16 +574,21 @@ async fn main() -> Result<(), Box> { TxValue::primitive(0u32.into()), // TxValue::unnamed_composite(coll.1.into_iter().map(|a| a.into())), // TxValue::unnamed_composite(vec![0u64.into(), 0u64.into(), 0u64.into(), 0u64.into()]), - TxValue::unnamed_variant("Id", [ TxValue::from_bytes(coll.0.public()) ]), + TxValue::unnamed_variant("Id", [TxValue::from_bytes(coll.0.public())]), TxValue::unnamed_variant("None", vec![]), - ] + ], ); api2.tx().create_signed_offline(&tx_call, &PairSigner::new(coll.0), tx_params) })?; // let futs = mint_txs.iter().map(|tx| tx.submit_and_watch()).collect::>(); let futs = mint_txs.iter().map(|tx| tx.submit()).collect::>(); - let _res = futs.collect::>().await.into_iter().collect::, _>>().expect("All the mint transactions submitted successfully"); + let _res = futs + .collect::>() + .await + .into_iter() + .collect::, _>>() + .expect("All the mint transactions submitted successfully"); // let waiter = res.into_iter().map(|txp| txp.wait_for_finalized_success()).collect::>(); // let _res = waiter.collect::>().await.into_iter().collect::, _>>().expect("All the mint transaction finalized"); @@ -584,28 +603,33 @@ async fn main() -> Result<(), Box> { let api2 = api.clone(); - sender_lib::sign_txs(cll.into_iter().zip(recv_accs.into_iter()), move |(coll, receiver)| { - let signer = PairSigner::new(coll.0); - let tx_params = DefaultExtrinsicParamsBuilder::new().nonce(2).build(); - let tx_call = subxt::dynamic::tx( - "Nfts", - "transfer", - vec![ - TxValue::primitive(coll.1.into()), - TxValue::primitive(0u32.into()), - // TxValue::unnamed_composite(coll.1.into_iter().map(|a| a.into())), - // TxValue::unnamed_composite(vec![0u64.into(), 0u64.into(), 0u64.into(), 0u64.into()]), - TxValue::unnamed_variant("Id", [ TxValue::from_bytes(receiver.public()) ]), - // TxValue::from_bytes(&EthereumSigner::from(receiver).into_account().0), - ], - ); - - api2.tx().create_signed_offline(&tx_call, &signer, tx_params.into()) - })? - } + sender_lib::sign_txs( + cll.into_iter().zip(recv_accs.into_iter()), + move |(coll, receiver)| { + let signer = PairSigner::new(coll.0); + let tx_params = DefaultExtrinsicParamsBuilder::new().nonce(2).build(); + let tx_call = subxt::dynamic::tx( + "Nfts", + "transfer", + vec![ + TxValue::primitive(coll.1.into()), + TxValue::primitive(0u32.into()), + // TxValue::unnamed_composite(coll.1.into_iter().map(|a| a.into())), + // TxValue::unnamed_composite(vec![0u64.into(), 0u64.into(), 0u64.into(), 0u64.into()]), + TxValue::unnamed_variant( + "Id", + [TxValue::from_bytes(receiver.public())], + ), + // TxValue::from_bytes(&EthereumSigner::from(receiver).into_account().0), + ], + ); + + api2.tx().create_signed_offline(&tx_call, &signer, tx_params.into()) + }, + )? + }, }; - log::info!("Transactions signed"); log::info!("Sending transactions..."); diff --git a/utils/sender/Cargo.toml b/utils/sender/Cargo.toml index 4ad5f9c..1eeda55 100644 --- a/utils/sender/Cargo.toml +++ b/utils/sender/Cargo.toml @@ -17,8 +17,8 @@ futures = "0.3.28" frame-system.workspace = true pallet-balances.workspace = true url = "2.5" -jsonrpsee-client-transport = "0.22" -jsonrpsee-core = "0.22" +jsonrpsee-client-transport = { version = "0.24", features=["ws"] } +jsonrpsee-core = {version = "0.24", features=["async-client"] } [lib] name = "sender_lib" diff --git a/utils/sender/src/main.rs b/utils/sender/src/main.rs index a03a972..a6caf4f 100644 --- a/utils/sender/src/main.rs +++ b/utils/sender/src/main.rs @@ -22,6 +22,7 @@ use tokio::sync::RwLock; const SENDER_SEED: &str = "//Sender"; const RECEIVER_SEED: &str = "//Receiver"; +const ALICE_SEED: &str = "//Alice"; /// Util program to send transactions #[derive(Parser, Debug)] @@ -42,6 +43,20 @@ struct Args { /// Total number of pre-funded accounts (on funded-accounts.json). #[arg(long)] tps: usize, + + /// Send in batch mode with the batch size this large. + #[arg(long, default_value_t = 1)] + batch: usize, + + /// Seed the sender accounts + #[arg( + long, + default_value_t = false, + default_missing_value = "false", + num_args = 0..=1, + require_equals = false, + )] + seed: bool, } // FIXME: This assumes that all the chains supported by sTPS use this `AccountInfo` type. Currently, @@ -90,13 +105,19 @@ fn main() -> Result<(), Box> { let args = Args::parse(); // Assume number of senders equal to TPS if not specified. + let n_sender_tasks = if args.batch > 1 { args.tps / args.batch } else { args.tps }; let n_tx_sender = args.total_senders.unwrap_or(args.tps); - let worker_sleep = (1_000f64 * (n_tx_sender as f64 / args.tps as f64)) as u64; + let worker_sleep = + (1_000f64 * ((n_sender_tasks as f64 * args.batch as f64) / args.tps as f64)) as u64; log::info!("worker_sleep = {}", worker_sleep); + log::info!("sender tasks = {}", n_sender_tasks); + log::info!("sender accounts = {}", n_tx_sender); let sender_accounts = funder_lib::derive_accounts(n_tx_sender, SENDER_SEED.to_owned()); let receiver_accounts = funder_lib::derive_accounts(n_tx_sender, RECEIVER_SEED.to_owned()); + let alice = ::from_string(&ALICE_SEED, None).unwrap(); + let alice_signer = PairSigner::::new(alice.clone()); async fn create_api(node_url: String) -> OnlineClient { let node_url = url::Url::parse(&node_url).unwrap(); @@ -113,7 +134,9 @@ fn main() -> Result<(), Box> { OnlineClient::from_backend(backend).await.unwrap() } - loop { + if args.seed { + log::info!("Seeding accounts"); + tokio::runtime::Builder::new_multi_thread() .enable_all() .build() @@ -121,6 +144,53 @@ fn main() -> Result<(), Box> { .block_on(async { let node_url = args.node_url.clone(); let api = create_api(node_url.clone()).await; + let mut best_block_stream = + api.blocks().subscribe_best().await.expect("Subscribe to best block failed"); + let best_block = best_block_stream.next().await.unwrap().unwrap(); + let block_ref: BlockRef = + BlockRef::from_hash(best_block.hash()); + + let mut nonce = get_account_nonce(&api, block_ref.clone(), &alice).await; + + for sender in sender_accounts.iter() { + let payload = subxt::dynamic::tx( + "Balances", + "transfer_keep_alive", + vec![ + Value::unnamed_variant("Id", [Value::from_bytes(sender.public())]), + Value::u128(1000000000000000000), + ], + ); + + let tx_params = Params::new().nonce(nonce as u64).build(); + + let tx = + api.tx().create_signed_offline(&payload, &alice_signer, tx_params).unwrap(); + + let watch = match tx.submit_and_watch().await { + Ok(watch) => { + log::info!("Seeded account"); + nonce += 1; + watch + }, + Err(err) => { + log::warn!("{:?}", err); + continue; + }, + }; + } + }); + } + + while !args.seed { + tokio::runtime::Builder::new_multi_thread() + .enable_all() + .build() + .unwrap() + .block_on( + async { + let node_url = args.node_url.clone(); + let api = create_api(node_url.clone()).await; // Subscribe to best block stream let mut best_block_stream = api.blocks().subscribe_best().await.expect("Subscribe to best block failed"); @@ -147,35 +217,39 @@ fn main() -> Result<(), Box> { let mut block_time = Duration::from_micros(0); loop { + sent.store(0, Ordering::SeqCst); in_block.store(0, Ordering::SeqCst); // Spawn 1 task per sender. - for i in 0..n_tx_sender { + for i in 0..n_sender_tasks { let in_block = in_block.clone(); let sent = sent.clone(); let sender = sender_accounts[i].clone(); - let signer = sender_signers[i].clone(); - + let signer: PairSigner = sender_signers[i].clone(); + let alice_signer = PairSigner::::new(alice.clone()); + let alice = alice.clone(); let best_block = best_block.clone(); let sent = sent.clone(); let in_block = in_block.clone(); let api = api.clone(); - let receiver = receiver_accounts[i].clone(); + let nrecv = if args.batch > 1 { args.batch } else { 1 }; + let receiver_accounts = receiver_accounts.clone(); - // TODO: Fix future transaction problem .... let task = async move { - // Slowly ramp up - tokio::time::sleep(std::time::Duration::from_millis((i/8) as u64)).await; + // Slowly ramp up 10ms slots. + tokio::time::sleep(std::time::Duration::from_millis(((n_sender_tasks - i)*10) as u64)).await; + let receivers = &receiver_accounts[i..i+nrecv]; let mut sleep_time_ms = 0u64; let block_ref: BlockRef = BlockRef::from_hash(best_block.read().await.0.hash()); let mut nonce = get_account_nonce(&api, block_ref.clone(), &sender).await; loop { - if sent.load(Ordering::SeqCst) > in_block.load(Ordering::SeqCst) + 12000 { // TODO: rpc pool size + // Throttle if the backlog of un included txs is too high + if sent.load(Ordering::SeqCst) > in_block.load(Ordering::SeqCst) + 100_000 { // Wait 10ms and check again. tokio::time::sleep(std::time::Duration::from_millis(10)).await; // Substract above sleep from TPS delay. @@ -183,20 +257,38 @@ fn main() -> Result<(), Box> { continue } - // Target a rate of 1TPS per worker, so we wait. + // Target a rate per worker, so we wait. tokio::time::sleep(std::time::Duration::from_millis(sleep_time_ms)).await; let now = Instant::now(); log::debug!("Sender {} using nonce {}", i, nonce); - let tx_payload = + let tx_payload = if args.batch > 1 { + let calls = (0..args.batch).map(|i| + subxt::dynamic::tx( + "Balances", + "transfer_keep_alive", + vec![ + Value::unnamed_variant("Id", [Value::from_bytes(receivers[i].public())]), + Value::u128(1000000000000), + ], + ).into_value() + ).collect::>(); + + subxt::dynamic::tx( + "Utility", + "batch", + vec![ Value::named_composite(vec![("calls", calls.into())]) ] + ) + } else { subxt::dynamic::tx( "Balances", "transfer_keep_alive", vec![ - Value::unnamed_variant("Id", [Value::from_bytes(receiver.public())]), - Value::u128(1u32.into()), + Value::unnamed_variant("Id", [Value::from_bytes(receivers[0].public())]), + Value::u128(1000000000000), ], - ); + ) + }; log::debug!("Sender {} using nonce {}", i, nonce); let tx_params = Params::new().nonce(nonce as u64).build(); @@ -208,7 +300,7 @@ fn main() -> Result<(), Box> { match tx.submit_and_watch().await { Ok(_watch) => {}, Err(err) => { - log::debug!("{:?}", err); + log::error!("{:?}", err); let block_ref: BlockRef = BlockRef::from_hash(best_block.read().await.0.hash()); nonce = get_account_nonce(&api, block_ref, &sender).await; // at most 1 second @@ -217,7 +309,8 @@ fn main() -> Result<(), Box> { } }; - sent.fetch_add(1, Ordering::SeqCst); + + sent.fetch_add(args.batch as u64, Ordering::SeqCst); // Determine how much left to sleep, we need to retry in 1000ms (backoff) sleep_time_ms = worker_sleep.saturating_sub(now.elapsed().as_millis() as u64); nonce += 1; @@ -260,9 +353,6 @@ fn main() -> Result<(), Box> { let mut txcount = 0; for ex in extrinsics.iter() { - let Ok(ex) = ex else { - continue - }; match (ex.pallet_name().expect("pallet name"), ex.variant_name().expect("variant name")) { ("Timestamp", "set") => { @@ -270,7 +360,17 @@ fn main() -> Result<(), Box> { block_time = new_timestamp - timestamp; timestamp = new_timestamp; }, - ("Balances", "transfer_keep_alive") | ("Nfts", "transfer") => { + ("Nfts", "transfer") => { + txcount += 1; + }, + _ => (), + } + } + + for ev in best_block.events().await.expect("Events are available").iter() { + let ev = ev.expect("Event is available"); + match (ev.pallet_name(), ev.variant_name()) { + ("Balances", "Transfer") => { txcount += 1; }, _ => (), @@ -278,15 +378,16 @@ fn main() -> Result<(), Box> { } in_block.fetch_add(txcount , Ordering::SeqCst); - let tps = txcount * 1000 / block_time.as_millis() as u64; + let btime = if block_time.is_zero() { 6000 } else { block_time.as_millis() }; + let tps = txcount * 1000 / btime as u64; tps_window.push_back(tps as usize); - // A window of size 3 - if tps_window.len() > 3 { + // A window of size 12 + if tps_window.len() > 12 { tps_window.pop_front(); let avg_tps = tps_window.iter().sum::(); - if avg_tps < args.tps / 3 { - log::warn!("TPS dropped by at least 66% ..."); + if avg_tps < args.tps / 4 { + log::warn!("TPS dropped below 25% of target ..."); break; } } @@ -305,6 +406,8 @@ fn main() -> Result<(), Box> { } log::info!("Restarting senders"); } - }); + } + ); } + Ok(()) } From 129d66b4301ac53e361d23ade5cba5d69fb50c8b Mon Sep 17 00:00:00 2001 From: Andrei Sandu Date: Thu, 17 Jul 2025 17:47:05 +0300 Subject: [PATCH 2/9] more funds Signed-off-by: Andrei Sandu --- utils/sender/src/main.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/utils/sender/src/main.rs b/utils/sender/src/main.rs index a6caf4f..558cabe 100644 --- a/utils/sender/src/main.rs +++ b/utils/sender/src/main.rs @@ -158,7 +158,7 @@ fn main() -> Result<(), Box> { "transfer_keep_alive", vec![ Value::unnamed_variant("Id", [Value::from_bytes(sender.public())]), - Value::u128(1000000000000000000), + Value::u128(100000000000000000000), ], ); From 2e34ffcbefd47daa8d9e74c16fb1fca4b4c98dd5 Mon Sep 17 00:00:00 2001 From: Alexander Cyon Date: Tue, 9 Sep 2025 19:21:32 +0200 Subject: [PATCH 3/9] compilation fixes, got sTPS working, running yap locally and then called with `cargo run -- --node-url ws://127.0.0.1:9944 --tps 10` --- utils/sender/src/main.rs | 28 ++++++++++++++-------------- 1 file changed, 14 insertions(+), 14 deletions(-) diff --git a/utils/sender/src/main.rs b/utils/sender/src/main.rs index bd2152e..25901fe 100644 --- a/utils/sender/src/main.rs +++ b/utils/sender/src/main.rs @@ -10,16 +10,14 @@ use std::{ }; // use subxt::{ext::sp_core::Pair, utils::AccountId32, OnlineClient, PolkadotConfig}; +use sp_core::{sr25519::Pair as SrPair, Pair}; use subxt::{ - blocks::BlockRef, - config::polkadot::PolkadotExtrinsicParamsBuilder as Params, - dynamic::Value, - OnlineClient, PolkadotConfig, + blocks::BlockRef, config::polkadot::PolkadotExtrinsicParamsBuilder as Params, dynamic::Value, + tx::SubmittableTransaction, OnlineClient, PolkadotConfig, }; -use sp_core::{sr25519::Pair as SrPair, Pair}; use tokio::sync::RwLock; -use sender_lib::{PairSigner, sign_balance_transfers}; +use sender_lib::PairSigner; const SENDER_SEED: &str = "//Sender"; const RECEIVER_SEED: &str = "//Receiver"; @@ -118,7 +116,7 @@ fn main() -> Result<(), Box> { let sender_accounts = funder_lib::derive_accounts(n_tx_sender, SENDER_SEED.to_owned()); let receiver_accounts = funder_lib::derive_accounts(n_tx_sender, RECEIVER_SEED.to_owned()); let alice = ::from_string(&ALICE_SEED, None).unwrap(); - let alice_signer = PairSigner::::new(alice.clone()); + let alice_signer = PairSigner::new(alice.clone()); async fn create_api(node_url: String) -> OnlineClient { let node_url = url::Url::parse(&node_url).unwrap(); @@ -165,10 +163,14 @@ fn main() -> Result<(), Box> { let tx_params = Params::new().nonce(nonce as u64).build(); - let tx = - api.tx().create_signed_offline(&payload, &alice_signer, tx_params).unwrap(); + let tx: SubmittableTransaction<_, OnlineClient<_>> = api + .tx() + .create_partial(&payload, &alice_signer.account_id(), tx_params) + .await + .unwrap() + .sign(&alice_signer); - let watch = match tx.submit_and_watch().await { + let _ = match tx.submit_and_watch().await { Ok(watch) => { log::info!("Seeded account"); nonce += 1; @@ -228,9 +230,7 @@ fn main() -> Result<(), Box> { let sent = sent.clone(); let sender = sender_accounts[i].clone(); - let signer: PairSigner = sender_signers[i].clone(); - let alice_signer = PairSigner::::new(alice.clone()); - let alice = alice.clone(); + let signer: PairSigner = sender_signers[i].clone(); let best_block = best_block.clone(); let sent = sent.clone(); let in_block = in_block.clone(); @@ -293,7 +293,7 @@ fn main() -> Result<(), Box> { log::debug!("Sender {} using nonce {}", i, nonce); let tx_params = Params::new().nonce(nonce as u64).build(); - let tx = api + let tx: SubmittableTransaction::<_, OnlineClient<_>> = api .tx() .create_partial_offline(&tx_payload, tx_params) .expect("Failed to create partial offline transaction") From 23259a795c016d21b4a2bccad63fb53c5238143b Mon Sep 17 00:00:00 2001 From: Alexander Cyon Date: Tue, 9 Sep 2025 19:23:17 +0200 Subject: [PATCH 4/9] fmt --- utils/sender/src/lib.rs | 38 +++++++++++++++++--------------------- 1 file changed, 17 insertions(+), 21 deletions(-) diff --git a/utils/sender/src/lib.rs b/utils/sender/src/lib.rs index c8bc855..1505460 100644 --- a/utils/sender/src/lib.rs +++ b/utils/sender/src/lib.rs @@ -1,18 +1,20 @@ use futures::{stream::FuturesUnordered, StreamExt}; use log::*; +use sp_core::{ + sr25519::{self, Pair as SrPair}, + Pair, +}; +use sp_runtime::{ + traits::{IdentifyAccount, Verify}, + MultiSignature, +}; use std::{error::Error, time::Duration}; use subxt::{ - config::polkadot::PolkadotExtrinsicParamsBuilder as Params, - config::substrate::AccountId32, + config::{polkadot::PolkadotExtrinsicParamsBuilder as Params, substrate::AccountId32}, dynamic::Value, tx::{Signer, SubmittableTransaction}, OnlineClient, PolkadotConfig, }; -use sp_core::{sr25519::{self, Pair as SrPair}, Pair}; -use sp_runtime::{ - traits::{IdentifyAccount, Verify}, - MultiSignature, -}; /// Maximal number of connection attempts. const MAX_ATTEMPTS: usize = 10; @@ -41,10 +43,7 @@ pub async fn connect(url: &str) -> Result, Box>; -pub fn sign_txs( - params: impl Iterator, - signer: S, -) -> Vec +pub fn sign_txs(params: impl Iterator, signer: S) -> Vec where P: Send + 'static, S: Fn(P) -> SignedTx + Send + Sync + 'static, @@ -66,11 +65,7 @@ where .push(std::thread::spawn(move || chunk.into_iter().map(&*signer).collect::>())); }); - threads - .into_iter() - .map(|h| h.join().unwrap()) - .flatten() - .collect() + threads.into_iter().map(|h| h.join().unwrap()).flatten().collect() } #[derive(Clone)] @@ -82,8 +77,7 @@ pub struct PairSigner { impl PairSigner { /// Creates a new [`Signer`] from an [`sp_core::sr25519::Pair`]. pub fn new(signer: sr25519::Pair) -> Self { - let account_id = - ::Signer::from(signer.public()).into_account(); + let account_id = ::Signer::from(signer.public()).into_account(); Self { // Convert `sp_core::AccountId32` to `subxt::config::substrate::AccountId32`. // @@ -121,7 +115,6 @@ impl Signer for PairSigner { } } - pub fn sign_balance_transfers( api: OnlineClient, pairs: impl Iterator, @@ -137,7 +130,10 @@ pub fn sign_balance_transfers( Value::u128(1u32.into()), ], ); - api.tx().create_partial_offline(&tx_call, tx_params).expect("Failed to create partial offline transaction").sign(&signer) + api.tx() + .create_partial_offline(&tx_call, tx_params) + .expect("Failed to create partial offline transaction") + .sign(&signer) }) } @@ -154,7 +150,7 @@ pub async fn submit_txs( match a { Ok(st) => match st { subxt::tx::TxStatus::Validated => log::trace!("VALIDATED"), - subxt::tx::TxStatus::Broadcasted => log::trace!("BROADCASTED"), + subxt::tx::TxStatus::Broadcasted => log::trace!("BROADCASTED"), subxt::tx::TxStatus::NoLongerInBestBlock => log::warn!("NO LONGER IN BEST BLOCK"), subxt::tx::TxStatus::InBestBlock(_) => log::trace!("IN BEST BLOCK"), subxt::tx::TxStatus::InFinalizedBlock(_) => log::trace!("IN FINALIZED BLOCK"), From 6da1f7c11a9b5134c870d78f420a6510fbca9c69 Mon Sep 17 00:00:00 2001 From: Alexander Cyon Date: Wed, 10 Sep 2025 09:14:46 +0200 Subject: [PATCH 5/9] remove dead code --- stps/src/main.rs | 51 ++++-------------------------------------------- 1 file changed, 4 insertions(+), 47 deletions(-) diff --git a/stps/src/main.rs b/stps/src/main.rs index 77f635f..29e8698 100644 --- a/stps/src/main.rs +++ b/stps/src/main.rs @@ -358,7 +358,6 @@ async fn main() -> Result<(), Box> { r }; - //let r = if !args.para { r.with_genesis_overrides(genesis_accs.clone()) } else { r }; let r = r.with_genesis_overrides(json!({ "configuration": { "config": { "executor_params": [ { "MaxMemoryPages": 8192 }, { "PvfExecTimeout": [ "Backing", 2500 ] } ] } } } )); let mut r = r.with_node(|node| node.with_name(relay_hostname.next().as_str()).invulnerable(true)); @@ -467,27 +466,10 @@ async fn main() -> Result<(), Box> { log::info!("Signing {} transactions...", send_accs.len()); let txs = match args.mode { - BenchMode::Stps => { - sender_lib::sign_balance_transfers( - api.clone(), - send_accs.into_iter().map(|a| (a, 0)).zip(recv_accs.into_iter()), - ) - // let api = api.clone(); - // sender_lib::sign_txs(send_accs.into_iter().zip(recv_accs.into_iter()), move |(sender, receiver)| { - // let signer = EthereumSigner::from(sender); - // let tx_params = DefaultExtrinsicParamsBuilder::::new().nonce(0).build(); - // let tx_call = subxt::dynamic::tx( - // "Balances", - // "transfer_keep_alive", - // vec![ - // TxValue::from_bytes(&EthereumSigner::from(receiver).into_account().0), - // TxValue::u128(1_000_000_000_000_000_000u128), - // ], - // ); - - // api.tx().create_signed_offline(&tx_call, &signer, tx_params.into()) - // })? - }, + BenchMode::Stps => sender_lib::sign_balance_transfers( + api.clone(), + send_accs.into_iter().map(|a| (a, 0)).zip(recv_accs.into_iter()), + ), BenchMode::NftTransfer => { let api2 = api.clone(); let create_coll_txs = sender_lib::sign_txs::<_, _, PolkadotConfig>( @@ -523,29 +505,12 @@ async fn main() -> Result<(), Box> { ); let futs = create_coll_txs.iter().map(|tx| tx.submit()).collect::>(); - // let futs = create_coll_txs.iter().map(|tx| tx.submit_and_watch()).collect::>(); - // let res = futs.collect::>().await.into_iter().collect::, _>>().expect("All the transactions submitted successfully"); let _res = futs .collect::>() .await .into_iter() .collect::, _>>() .expect("All the transactions submitted successfully"); - // let waiter = res.into_iter().map(|txp| txp.wait_for_finalized_success()).collect::>(); - // let res = waiter.collect::>().await.into_iter().collect::, _>>().expect("All the collection creation transaction finalized"); - - // let mut collections = Vec::new(); - - // for ev in res { - // for ed in ev.iter() { - // let ed = ed?; - // if ed.pallet_name() == "Nfts" && ed.variant_name() == "Created" { - // let b = ed.field_bytes(); - // let e = Collection::decode(&mut &b[..])?; - // collections.push(e); - // } - // } - // } let mut proc_coll = 0; let mut map: HashMap<[u8; 32], u32> = HashMap::new(); @@ -578,8 +543,6 @@ async fn main() -> Result<(), Box> { vec![ TxValue::primitive(coll.1.into()), TxValue::primitive(0u32.into()), - // TxValue::unnamed_composite(coll.1.into_iter().map(|a| a.into())), - // TxValue::unnamed_composite(vec![0u64.into(), 0u64.into(), 0u64.into(), 0u64.into()]), TxValue::unnamed_variant("Id", [TxValue::from_bytes(coll.0.public())]), TxValue::unnamed_variant("None", vec![]), ], @@ -591,7 +554,6 @@ async fn main() -> Result<(), Box> { }, ); - // let futs = mint_txs.iter().map(|tx| tx.submit_and_watch()).collect::>(); let futs = mint_txs.iter().map(|tx| tx.submit()).collect::>(); let _res = futs .collect::>() @@ -599,8 +561,6 @@ async fn main() -> Result<(), Box> { .into_iter() .collect::, _>>() .expect("All the mint transactions submitted successfully"); - // let waiter = res.into_iter().map(|txp| txp.wait_for_finalized_success()).collect::>(); - // let _res = waiter.collect::>().await.into_iter().collect::, _>>().expect("All the mint transaction finalized"); let mut proc_mint = 0; while proc_mint < ntrans { @@ -624,13 +584,10 @@ async fn main() -> Result<(), Box> { vec![ TxValue::primitive(coll.1.into()), TxValue::primitive(0u32.into()), - // TxValue::unnamed_composite(coll.1.into_iter().map(|a| a.into())), - // TxValue::unnamed_composite(vec![0u64.into(), 0u64.into(), 0u64.into(), 0u64.into()]), TxValue::unnamed_variant( "Id", [TxValue::from_bytes(receiver.public())], ), - // TxValue::from_bytes(&EthereumSigner::from(receiver).into_account().0), ], ); From 16a795704ebfcc59de866ac2e2525e43d65cb100 Mon Sep 17 00:00:00 2001 From: Alexander Cyon Date: Wed, 10 Sep 2025 09:17:14 +0200 Subject: [PATCH 6/9] use workspace for deps --- utils/sender/Cargo.toml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/utils/sender/Cargo.toml b/utils/sender/Cargo.toml index 98e1dc9..1be2efa 100644 --- a/utils/sender/Cargo.toml +++ b/utils/sender/Cargo.toml @@ -21,8 +21,8 @@ futures = "0.3.28" frame-system.workspace = true pallet-balances.workspace = true url = "2.5" -jsonrpsee-client-transport = { version = "0.24", features=["ws"] } -jsonrpsee-core = {version = "0.24", features=["async-client"] } +jsonrpsee-client-transport.workspace = true +jsonrpsee-core.workspace = true sp-core.workspace = true sp-runtime.workspace = true From 4ce80b884df0858ae649cb5476326b2b63f60de3 Mon Sep 17 00:00:00 2001 From: Alexander Cyon Date: Thu, 18 Sep 2025 10:03:48 +0200 Subject: [PATCH 7/9] change so that senders are more 'rich' and spending less per tx --- utils/sender/src/main.rs | 20 +++++++++++++------- 1 file changed, 13 insertions(+), 7 deletions(-) diff --git a/utils/sender/src/main.rs b/utils/sender/src/main.rs index 25901fe..6def477 100644 --- a/utils/sender/src/main.rs +++ b/utils/sender/src/main.rs @@ -6,14 +6,12 @@ use std::{ collections::VecDeque, error::Error, sync::atomic::{AtomicU64, Ordering}, - time::Instant, + time::Instant, u128, }; -// use subxt::{ext::sp_core::Pair, utils::AccountId32, OnlineClient, PolkadotConfig}; use sp_core::{sr25519::Pair as SrPair, Pair}; use subxt::{ - blocks::BlockRef, config::polkadot::PolkadotExtrinsicParamsBuilder as Params, dynamic::Value, - tx::SubmittableTransaction, OnlineClient, PolkadotConfig, + blocks::BlockRef, config::polkadot::PolkadotExtrinsicParamsBuilder as Params, dynamic::Value, ext::scale_value::{Primitive, ValueDef}, tx::SubmittableTransaction, OnlineClient, PolkadotConfig }; use tokio::sync::RwLock; @@ -23,6 +21,14 @@ const SENDER_SEED: &str = "//Sender"; const RECEIVER_SEED: &str = "//Receiver"; const ALICE_SEED: &str = "//Alice"; +/// Amount to send in each transaction, small so that we can do many transactions before +/// running out of funds. +const SMALL_TOKEN_AMOUNT: Value = Value { value: ValueDef::Primitive(Primitive::U128(1)), context: () }; + +/// Amount to seed each sender with, largest possible value so that we do not run out of funds. +const BIG_TOKEN_AMOUNT: Value = Value { value: ValueDef::Primitive(Primitive::U128(u128::MAX)), context: () }; + + /// Util program to send transactions #[derive(Parser, Debug)] #[command(author, version, about, long_about = None)] @@ -157,7 +163,7 @@ fn main() -> Result<(), Box> { "transfer_keep_alive", vec![ Value::unnamed_variant("Id", [Value::from_bytes(sender.public())]), - Value::u128(100000000000000000000), + BIG_TOKEN_AMOUNT, ], ); @@ -270,7 +276,7 @@ fn main() -> Result<(), Box> { "transfer_keep_alive", vec![ Value::unnamed_variant("Id", [Value::from_bytes(receivers[i].public())]), - Value::u128(1000000000000), + SMALL_TOKEN_AMOUNT, ], ).into_value() ).collect::>(); @@ -286,7 +292,7 @@ fn main() -> Result<(), Box> { "transfer_keep_alive", vec![ Value::unnamed_variant("Id", [Value::from_bytes(receivers[0].public())]), - Value::u128(1000000000000), + SMALL_TOKEN_AMOUNT, ], ) }; From 3176ff4cd68e2b6337b6ba0c33ff63d17e116287 Mon Sep 17 00:00:00 2001 From: Alexander Cyon Date: Thu, 18 Sep 2025 10:52:12 +0200 Subject: [PATCH 8/9] fmt --- utils/funder/src/lib.rs | 6 +----- utils/funder/src/main.rs | 2 +- utils/sender/src/main.rs | 17 ++++++++++++----- 3 files changed, 14 insertions(+), 11 deletions(-) diff --git a/utils/funder/src/lib.rs b/utils/funder/src/lib.rs index 7948af8..1dd2f08 100644 --- a/utils/funder/src/lib.rs +++ b/utils/funder/src/lib.rs @@ -24,9 +24,5 @@ pub fn derive_accounts(n: usize, seed: String) -> Vec { })); }); - threads - .into_iter() - .map(|h| h.join().unwrap()) - .flatten() - .collect() + threads.into_iter().map(|h| h.join().unwrap()).flatten().collect() } diff --git a/utils/funder/src/main.rs b/utils/funder/src/main.rs index d59683c..6894328 100644 --- a/utils/funder/src/main.rs +++ b/utils/funder/src/main.rs @@ -1,6 +1,6 @@ use clap::Parser; -use std::{error::Error, fs::File, path::PathBuf}; use sp_core::{crypto::Ss58Codec, Pair}; +use std::{error::Error, fs::File, path::PathBuf}; const DEFAULT_FUNDED_JSON_PATH: &str = "funded-accounts.json"; const FUNDS: u64 = 10_000_000_000_000_000; diff --git a/utils/sender/src/main.rs b/utils/sender/src/main.rs index 6def477..10432e9 100644 --- a/utils/sender/src/main.rs +++ b/utils/sender/src/main.rs @@ -6,12 +6,18 @@ use std::{ collections::VecDeque, error::Error, sync::atomic::{AtomicU64, Ordering}, - time::Instant, u128, + time::Instant, + u128, }; use sp_core::{sr25519::Pair as SrPair, Pair}; use subxt::{ - blocks::BlockRef, config::polkadot::PolkadotExtrinsicParamsBuilder as Params, dynamic::Value, ext::scale_value::{Primitive, ValueDef}, tx::SubmittableTransaction, OnlineClient, PolkadotConfig + blocks::BlockRef, + config::polkadot::PolkadotExtrinsicParamsBuilder as Params, + dynamic::Value, + ext::scale_value::{Primitive, ValueDef}, + tx::SubmittableTransaction, + OnlineClient, PolkadotConfig, }; use tokio::sync::RwLock; @@ -23,11 +29,12 @@ const ALICE_SEED: &str = "//Alice"; /// Amount to send in each transaction, small so that we can do many transactions before /// running out of funds. -const SMALL_TOKEN_AMOUNT: Value = Value { value: ValueDef::Primitive(Primitive::U128(1)), context: () }; +const SMALL_TOKEN_AMOUNT: Value = + Value { value: ValueDef::Primitive(Primitive::U128(1)), context: () }; /// Amount to seed each sender with, largest possible value so that we do not run out of funds. -const BIG_TOKEN_AMOUNT: Value = Value { value: ValueDef::Primitive(Primitive::U128(u128::MAX)), context: () }; - +const BIG_TOKEN_AMOUNT: Value = + Value { value: ValueDef::Primitive(Primitive::U128(u128::MAX)), context: () }; /// Util program to send transactions #[derive(Parser, Debug)] From 7e28b862e499b8f7038040bd24957a0003a83d73 Mon Sep 17 00:00:00 2001 From: Alexander Cyon Date: Thu, 18 Sep 2025 16:52:08 +0200 Subject: [PATCH 9/9] CI: update ubuntu runner to 'ubuntu-lastest' --- .github/workflows/test-code.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/test-code.yml b/.github/workflows/test-code.yml index e636dc5..3bacdce 100644 --- a/.github/workflows/test-code.yml +++ b/.github/workflows/test-code.yml @@ -10,7 +10,7 @@ on: jobs: test-code: - runs-on: ubuntu-20.04 + runs-on: ubuntu-latest defaults: run: working-directory: ./utils