feat: signing and firing real ethereum txs (#2)

* wip

* oop

* signer

* does this fix

* guh

* reps

* wut

* Revert "wut"

This reverts commit 2b4ca56.

* oh spawn more lol

* hm

* Update main.rs

* Reapply "wut"

This reverts commit 699b70d.

* Revert "Reapply "wut""

This reverts commit c8879ce.

* Update worker.rs

* async tx gen worker

* wuh

* Revert "wuh"

This reverts commit 3c8f5d6.

* Revert "async tx gen worker"

This reverts commit aa1c1a7.

* wuh2

* fix async?

* full threads

* ratio, sync again

* Update main.rs

* Update main.rs

* try: cachepad

* Revert "try: cachepad"

This reverts commit 947b586.

* LAWG

* breh

* breh2

* Update main.rs

* lawg

* brh

* bind

* disable pinning for a sec

* feat: current thread tokio

* Update main.rs

* bruh oops

* pin thread

* curious

* just pinned connection workers?

* no pinning

* yes pin

* no pin all

* pin

* Update main.rs

* disable again

* blarg

* blarg

* Update main.rs

* lawgs

* Update queue.rs

* Update queue.rs

* grah

* feat: clean

* Update network.rs

* reorg workers

* feat: ensure each worker gets a core

* comments

* blah

* proper json body

* rm log

* feat: proper decoder on testserver

* Update network.rs

* Update tx_queue.rs

* Update tx_queue.rs

* try cooked ratio

* Update main.rs

* disable thread pinning

* pinning

* actual reth?

* log?

* lawg

* bruh

* log ranges

* Update workers.rs

* todo

* tehehe

* Update network.rs

* Update network.rs

* Update network.rs

* fml

* Update network.rs

* Update tx_gen.rs

* Update network.rs

* new signer each time hehe

* Update tx_gen.rs

* DEBUG way fewer connections, SLEEP

* Revert "DEBUG way fewer connections, SLEEP"

This reverts commit 19cc477.

* anvil key

* blarg

* blarg

* 0 gwei?

* todo

* feat: more granular worker specification

* feat: use MnemonicBuilder

* 1 wei

* Update tx_gen.rs

* Update tx_gen.rs

* fix

* pop tx from front

* rm: annoying log

* Update tx_queue.rs

* time req

* Revert "time req"

This reverts commit c63d5c5.

* MOAR CONNECTIONS

* 100k connections?!?!

* 20k connections

* 50k connections?!

* 30k connections

* testserver match port

* bruh

* Update tx_queue.rs

* 20k conn

* upgrade actix max connections

* 100k connections

* extra detail?

* document epehemral ports

* Update network.rs

* Update main.rs

* Update main.rs

* Update main.rs

* Revert "Update main.rs"

This reverts commit 7eaefce.

* req async?

* Update network.rs

* undo async reqs

* feat: batching

* lawg

* rm measure

* Update network.rs

* proper rps measure with batching

* Update network.rs

Author: transmissions11

Cargo.lock

@@ -14,4 +14,9 @@ tokio-metrics = "0.4.3"
 rlimit = "0.10.2"
 num_cpus = "1.17.0"
 crossbeam-utils = "0.8.21"
-mimalloc = "0.1.47"
+mimalloc = "0.1.47"
+alloy = { version = "1", features = ["genesis", "signers"] }
+alloy-evm = "0.13"
+alloy-consensus = { version = "1", features = ["secp256k1"] }
+alloy-signer-local = { version = "1", features = ["mnemonic"] }
+core_affinity = "0.8.3"

crescendo/Cargo.toml

@@ -2,50 +2,82 @@ use std::future::pending;
 use std::thread;
 use std::time::Duration;
 
+use core_affinity;
 use mimalloc::MiMalloc;
-use stats::STATS;
 
-mod stats;
+mod network_stats;
+mod tx_queue;
 mod utils;
-mod worker;
+mod workers;
+
+use crate::network_stats::NETWORK_STATS;
+use crate::tx_queue::TX_QUEUE;
+use crate::workers::{DesireType, WorkerType};
 
 #[global_allocator]
 // Increases RPS by ~5.5% at the time of
 // writing. ~3.3% faster than jemalloc.
 static GLOBAL: MiMalloc = MiMalloc;
 
-const TOTAL_CONNECTIONS: u64 = 4096;
-const TARGET_URL: &str = "http://127.0.0.1:8080";
+// TODO: Configurable CLI args.
+const TOTAL_CONNECTIONS: u64 = 20_000; // This is limited by the amount of ephemeral ports available on the system.
+const THREAD_PINNING: bool = true;
+const TARGET_URL: &str = "http://127.0.0.1:8545";
 
-#[tokio::main(worker_threads = 1)]
+#[tokio::main(flavor = "current_thread")]
 async fn main() {
-    let num_threads = num_cpus::get() as u64;
-    let connections_per_thread = TOTAL_CONNECTIONS / num_threads;
-
     if let Err(err) = utils::increase_nofile_limit(TOTAL_CONNECTIONS * 10) {
-        println!("Failed to increase file descriptor limit: {err}.");
+        println!("[!] Failed to increase file descriptor limit: {err}.");
     }
 
-    println!(
-        "Running {} threads with {} connections each against {}...",
-        num_threads, connections_per_thread, TARGET_URL
+    let mut core_ids = core_affinity::get_core_ids().unwrap();
+    println!("[*] Detected {} effective cores.", core_ids.len());
+
+    // Pin the tokio runtime to a core (if enabled).
+    utils::maybe_pin_thread(core_ids.pop().unwrap(), THREAD_PINNING);
+
+    // Given our desired breakdown of workers, translate this into actual numbers of workers to spawn.
+    let (workers, worker_counts) = workers::assign_workers(
+        core_ids, // Doesn't include the main runtime core.
+        vec![(WorkerType::TxGen, DesireType::Exact(20)), (WorkerType::Network, DesireType::Percentage(1.0))],
+        THREAD_PINNING, // Only log core ranges if thread pinning is actually enabled.
     );
 
-    // Spawn all worker threads.
-    for _ in 0..num_threads {
-        thread::spawn(move || {
-            let rt = tokio::runtime::Builder::new_current_thread().enable_all().build().unwrap();
-            rt.block_on(async {
-                for _ in 0..connections_per_thread {
-                    tokio::spawn(worker::connection_worker(TARGET_URL));
-                }
-                pending::<()>().await; // Keep the runtime alive forever.
-            });
-        });
+    let connections_per_network_worker = TOTAL_CONNECTIONS / worker_counts[&WorkerType::Network];
+    println!("[*] Connections per network worker: {}", connections_per_network_worker);
+
+    // TODO: Having the assign_workers function do this would be cleaner, also give ids to the network workers.
+    let mut tx_gen_worker_id = 0;
+
+    // Spawn the workers, pinning them to the appropriate cores if enabled.
+    for (core_id, worker_type) in workers {
+        match worker_type {
+            WorkerType::TxGen => {
+                thread::spawn(move || {
+                    utils::maybe_pin_thread(core_id, THREAD_PINNING);
+                    workers::tx_gen_worker(tx_gen_worker_id);
+                });
+                tx_gen_worker_id += 1;
+            }
+            WorkerType::Network => {
+                thread::spawn(move || {
+                    utils::maybe_pin_thread(core_id, THREAD_PINNING);
+                    let rt = tokio::runtime::Builder::new_current_thread().enable_all().build().unwrap();
+
+                    rt.block_on(async {
+                        for _ in 0..connections_per_network_worker {
+                            tokio::spawn(workers::network_worker(TARGET_URL));
+                        }
+                        pending::<()>().await; // Keep the runtime alive forever.
+                    });
+                });
+            }
+        }
     }
 
-    // Start stats reporter.
-    tokio::spawn(STATS.start_reporter(Duration::from_secs(1)))
+    // Start reporters.
+    tokio::spawn(TX_QUEUE.start_reporter(Duration::from_secs(1)));
+    tokio::spawn(NETWORK_STATS.start_reporter(Duration::from_secs(1)))
         .await // Keep the main thread alive forever.
         .unwrap();
 }

crescendo/src/main.rs

@@ -4,7 +4,7 @@ use std::time::Duration;
 use crossbeam_utils::CachePadded;
 use thousands::Separable;
 
-pub struct Stats {
+pub struct NetworkStats {
     requests: AtomicU64,
     errors: AtomicU64,
 }
@@ -13,16 +13,16 @@ pub struct Stats {
 // other frequently accessed memory can occur. To mitigate, we pad stats
 // to the length of a full cache line to avoid conflict. This is measured
 // to increase RPS by >10% in the release profile at the time of writing.
-pub static STATS: CachePadded<Stats> =
-    CachePadded::new(Stats { requests: AtomicU64::new(0), errors: AtomicU64::new(0) });
+pub static NETWORK_STATS: CachePadded<NetworkStats> =
+    CachePadded::new(NetworkStats { requests: AtomicU64::new(0), errors: AtomicU64::new(0) });
 
-impl Stats {
-    pub fn inc_requests(&self) {
-        self.requests.fetch_add(1, Ordering::Relaxed);
+impl NetworkStats {
+    pub fn inc_requests_by(&self, count: usize) {
+        self.requests.fetch_add(count as u64, Ordering::Relaxed);
     }
 
-    pub fn inc_errors(&self) {
-        self.errors.fetch_add(1, Ordering::Relaxed);
+    pub fn inc_errors_by(&self, count: usize) {
+        self.errors.fetch_add(count as u64, Ordering::Relaxed);
     }
 
     pub async fn start_reporter(&self, measurement_interval: Duration) {
@@ -37,7 +37,7 @@ impl Stats {
             let rps = requests - last_requests;
             let eps = errors - last_errors;
             println!(
-                "RPS: {}, EPS: {}, Total requests: {}, Total errors: {}",
+                "[*] RPS: {}, EPS: {}, Total requests: {}, Total errors: {}",
                 rps.separate_with_commas(),
                 eps.separate_with_commas(),
                 requests.separate_with_commas(),

crescendo/src/stats.rs crescendo/src/network_stats.rscrescendo/src/stats.rs renamed to crescendo/src/network_stats.rs

@@ -0,0 +1,62 @@
+use std::collections::VecDeque;
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::sync::Mutex;
+
+use thousands::Separable;
+
+pub struct TxQueue {
+    // TODO: RwLock? Natively concurrent deque?
+    queue: Mutex<VecDeque<Vec<u8>>>,
+    total_added: AtomicU64,
+}
+
+pub static TX_QUEUE: TxQueue = TxQueue { queue: Mutex::new(VecDeque::new()), total_added: AtomicU64::new(0) };
+
+impl TxQueue {
+    pub fn push_tx(&self, tx: Vec<u8>) {
+        self.total_added.fetch_add(1, Ordering::Relaxed);
+        self.queue.lock().unwrap().push_back(tx);
+    }
+
+    pub fn queue_len(&self) -> usize {
+        self.queue.lock().map(|q| q.len()).unwrap_or(0)
+    }
+
+    pub fn pop_at_most(&self, max_count: usize) -> Option<Vec<Vec<u8>>> {
+        let mut queue = self.queue.lock().ok()?;
+
+        let count = max_count.min(queue.len());
+        if count == 0 {
+            return None;
+        }
+
+        // TODO: Is drain more or less efficient than repeated pop_front?
+        // It's important to pop from the front here, otherwise the node
+        // gets confused seeing a bunch of txs with incredibly high nonces
+        // before it sees any of the lower ones. It's possible this issue
+        // could still emerge at high enough RPS, but haven't seen it yet.
+        Some(queue.drain(..count).collect())
+    }
+
+    pub async fn start_reporter(&self, measurement_interval: std::time::Duration) {
+        let mut last_total_added = 0u64;
+        let mut last_queue_len = 0usize;
+        let mut interval = tokio::time::interval(measurement_interval);
+        interval.tick().await;
+        loop {
+            interval.tick().await;
+            let current_total_added = self.total_added.load(Ordering::Relaxed);
+            let current_queue_len = self.queue_len();
+            let added_per_second = current_total_added - last_total_added;
+            let queue_growth = current_queue_len.saturating_sub(last_queue_len);
+            println!(
+                "[*] TxQueue +/s: {}, TxQueue Δ/s: {}, Current length: {}",
+                added_per_second.separate_with_commas(),
+                queue_growth.separate_with_commas(),
+                current_queue_len.separate_with_commas()
+            );
+            last_total_added = current_total_added;
+            last_queue_len = current_queue_len;
+        }
+    }
+}

crescendo/src/tx_queue.rs

@@ -1,20 +1,69 @@
 use std::io;
 
+use core_affinity::CoreId;
 use rlimit::Resource;
 
+/// Increase the file descriptor limit to the given minimum.
+///
+/// Panics if the hard limit is too low, otherwise tries to increase.
 pub fn increase_nofile_limit(min_limit: u64) -> io::Result<u64> {
     let (soft, hard) = Resource::NOFILE.get()?;
-    println!("At startup, file descriptor limit:            soft = {soft}, hard = {hard}");
+    println!("[*] At startup, file descriptor limit:      soft = {soft}, hard = {hard}");
 
     if hard < min_limit {
-        panic!("File descriptor hard limit is too low. Please increase it to at least {}.", min_limit);
+        panic!("[!] File descriptor hard limit is too low. Please increase it to at least {}.", min_limit);
     }
 
     if soft != hard {
         Resource::NOFILE.set(hard, hard)?; // Just max things out to give us plenty of overhead.
         let (soft, hard) = Resource::NOFILE.get()?;
-        println!("After increasing file descriptor limit:       soft = {soft}, hard = {hard}");
+        println!("[+] After increasing file descriptor limit: soft = {soft}, hard = {hard}");
     }
 
     Ok(soft)
 }
+
+/// Pin the current thread to the given core ID if enabled.
+///
+/// Panics if the thread fails to pin.
+pub fn maybe_pin_thread(core_id: CoreId, enable_thread_pinning: bool) {
+    if !enable_thread_pinning {
+        return;
+    }
+
+    if !core_affinity::set_for_current(core_id) {
+        panic!("[!] Failed to pin thread to core {}.", core_id.id);
+    }
+}
+
+/// Format a sorted list of numbers into a range string (e.g., "1-3, 5, 7-9")
+pub fn format_ranges(nums: &[usize]) -> String {
+    if nums.is_empty() {
+        return String::new();
+    }
+
+    let mut ranges = Vec::new();
+    let mut i = 0;
+
+    while i < nums.len() {
+        let start = nums[i];
+        let mut end = start;
+
+        // Find end of consecutive sequence
+        while i + 1 < nums.len() && nums[i + 1] == nums[i] + 1 {
+            i += 1;
+            end = nums[i];
+        }
+
+        // Format this range
+        if start == end {
+            ranges.push(start.to_string());
+        } else {
+            ranges.push(format!("{}-{}", start, end));
+        }
+
+        i += 1;
+    }
+
+    ranges.join(", ")
+}