Merge pull request #112 from gpBlockchain/lgp/bench

feat: Add concurrent_requests command and enable multi-threaded load test

Author: liya2017

.github/workflows/release.yml

@@ -23,7 +23,7 @@ jobs:
         build: [linux, macos, windows]
         include:
           - build: linux
-            os: ubuntu-18.04
+            os: ubuntu-20.04
           - build: macos
             os: macos-latest
           - build: windows
@@ -52,7 +52,7 @@ jobs:
 
     - name: Create archive for Linux
       run: 7z a -ttar -so -an ./${{ env.PACKAGE }}/target/release/${{ env.PACKAGE }} | 7z a -si ./artifacts/${{ env.PACKAGE }}-linux-x86_64.tar.gz
-      if: matrix.os == 'ubuntu-18.04'
+      if: matrix.os == 'ubuntu-20.04'
 
     - name: Create archive for Windows
       run: 7z a -tzip ./artifacts/${{ env.PACKAGE }}-windows-x86_64.zip ./${{ env.PACKAGE }}/target/release/${{ env.PACKAGE }}.exe

ckb-bench/Cargo.lock

@@ -20,3 +20,5 @@ toml = "0.5"
 lru = "0.6.0"
 crossbeam-channel = "0.5.1"
 git-version = "0.3.5"
+tokio = { version = "1.14.0", features = ["full"] }
+futures = "0.3"

ckb-bench/Cargo.toml

@@ -1,6 +1,6 @@
-- src: https://github.com/keroro520/ansible-ckb
+- src: https://github.com/cryptape/ansible-ckb
   scm: git
   version: main
-- src: https://github.com/keroro520/ansible-ckb-benchmark
+- src: https://github.com/cryptape/ansible-ckb-benchmark
   scm: git
   version: main

ckb-bench/devtools/ci/ansible/requirements.yml

@@ -14,7 +14,7 @@
     --n-users 9000
   ```
 
-3. Bench 300 seconds, benched transactions type are 2-in-2-out, sending transaction interval is 10ms
+3. Bench 300 seconds, benched transactions type are 2-in-2-out, sending transaction interval is 10ms ,10 users
 
   ```shell
   CKB_BENCH_OWNER_PRIVKEY=af44a4755acccdd932561db5163d5c2ac025faa00877719c78bb0b5d61da8c94 \
@@ -25,10 +25,11 @@
     --bench-time-ms 300000 \
     --n-users 9000 \
     --n-inout 2 \
-    --tx-interval-ms 10
+    --tx-interval-ms 10 \ 
+    --concurrent-requests 10 
   ```
 
-2. Cleanup. Collect cells back to owner.
+4. Cleanup. Collect cells back to owner.
 
   ```shell
   CKB_BENCH_OWNER_PRIVKEY=af44a4755acccdd932561db5163d5c2ac025faa00877719c78bb0b5d61da8c94 \

ckb-bench/docs/bench-on-testnet.md

@@ -50,13 +50,15 @@ CKB-bench provides several options for specifying benchmark scenarios. Here is a
     --n-users 9000 \
     --n-inout 2 \
     --bench-time-ms 300000 \
-    --tx-interval-ms 10
+    --tx-interval-ms 10 \
+    --concurrent-requests 10 
   ```
 
 - `--n-users 9000`: Use the `9000` derived users to bench
 - `--n-inout 2`: Construct 2-in-2-out transactions
 - `--bench-time-ms 300000`: Bench `300000` milliseconds
 - `--tx-interval-ms 10`: Delay 10 milliseconds between sending continuous transactions
+- `--concurrent-requests 10` : 10 users are conducting load testing simultaneously.
 
 Ckb-bench continuously performs these tasks for `bench-time-ms` duration:
   - collects unspent cells of derived users

ckb-bench/docs/usage.md

@@ -1 +1 @@
-1.56.0
+1.67.1

ckb-bench/rust-toolchain

@@ -10,27 +10,53 @@ use ckb_testkit::{Node, User};
 use crossbeam_channel::{Receiver, Sender};
 use lru::LruCache;
 use std::collections::HashMap;
-use std::time::Instant;
+use std::sync::{Arc};
+use std::time::{Duration, Instant};
+use tokio::sync::Semaphore;
+use futures::stream::FuturesUnordered;
+use futures::{FutureExt, StreamExt};
+use tokio::time::sleep as async_sleep;
+use crate::utils::maybe_retry_send_transaction_async;
+use std::sync::atomic::{AtomicUsize, Ordering};
 
 pub struct LiveCellProducer {
     users: Vec<User>,
     nodes: Vec<Node>,
     seen_out_points: LruCache<OutPoint, Instant>,
 }
 
-// TODO Add more logs
 impl LiveCellProducer {
     pub fn new(users: Vec<User>, nodes: Vec<Node>) -> Self {
         let n_users = users.len();
+
+        let mut user_unused_max_cell_count_cache = 1;
+        // step_by: 20 : using a sampling method to find the user who owns the highest number of cells.
+        // seen_out_points lruCache cache size = user_unused_max_cell_count_cache * n_users + 10
+        // seen_out_points lruCache: preventing unused cells on the chain from being reused.
+        for i in (0..=users.len()-1).step_by(20) {
+            let user_unused_cell_count_cache = users.get(i).expect("out of bound").get_spendable_single_secp256k1_cells(&nodes[0]).len();
+            if user_unused_cell_count_cache > user_unused_max_cell_count_cache && user_unused_cell_count_cache <= 10000 {
+                user_unused_max_cell_count_cache = user_unused_cell_count_cache;
+            }
+            ckb_testkit::debug!("idx:{}:user_unused_cell_count_cache:{}",i,user_unused_cell_count_cache)
+        }
+        ckb_testkit::debug!("user max cell count cache:{}",user_unused_max_cell_count_cache);
+        let lrc_cache_size = n_users * user_unused_max_cell_count_cache + 10;
+        ckb_testkit::info!("init unused cache size:{}",lrc_cache_size);
         Self {
             users,
             nodes,
-            seen_out_points: LruCache::new(n_users + 10),
+            seen_out_points: LruCache::new(lrc_cache_size),
         }
     }
 
-    pub fn run(mut self, live_cell_sender: Sender<CellMeta>) {
+    pub fn run(mut self, live_cell_sender: Sender<CellMeta>, log_duration: u64) {
+        let mut count = 0;
+        let mut start_time = Instant::now();
+        let mut duration_count = 0;
+        let mut fist_send_finished = true;
         loop {
+            // let mut current_loop_start_time = Instant::now();
             let min_tip_number = self
                 .nodes
                 .iter()
@@ -60,8 +86,22 @@ impl LiveCellProducer {
                     self.seen_out_points
                         .put(cell.out_point.clone(), Instant::now());
                     let _ignore = live_cell_sender.send(cell);
+                    count += 1;
+                    duration_count += 1;
+                    if Instant::now().duration_since(start_time) >= Duration::from_secs(log_duration) {
+                        let elapsed = start_time.elapsed();
+                        ckb_testkit::info!("[LiveCellProducer] producer count: {} ,duration time:{:?} , duration tps:{}", count,elapsed,duration_count*1000/elapsed.as_millis());
+                        duration_count = 0;
+                        start_time = Instant::now();
+                    }
+                }
+
+                if fist_send_finished {
+                    fist_send_finished = false;
+                    self.seen_out_points.resize(count + 10)
                 }
             }
+            // ckb_testkit::debug!("[LiveCellProducer] delay:{:?}",current_loop_start_time.elapsed());
         }
     }
 }
@@ -74,7 +114,7 @@ pub struct TransactionProducer {
     // #{ lock_hash => live_cell }
     live_cells: HashMap<Byte32, CellMeta>,
     // #{ out_point => live_cell }
-    backlogs: HashMap<OutPoint, CellMeta>,
+    backlogs: HashMap<Byte32, Vec<CellMeta>>,
 }
 
 impl TransactionProducer {
@@ -113,6 +153,7 @@ impl TransactionProducer {
         mut self,
         live_cell_receiver: Receiver<CellMeta>,
         transaction_sender: Sender<TransactionView>,
+        log_duration: u64,
     ) {
         // Environment variables `CKB_BENCH_ENABLE_DATA1_SCRIPT` and
         // `CKB_BENCH_ENABLE_INVALID_SINCE_EPOCH` are temporary.
@@ -137,17 +178,29 @@ impl TransactionProducer {
             "CKB_BENCH_ENABLE_INVALID_SINCE_EPOCH = {}",
             enabled_invalid_since_epoch
         );
+        let mut count = 0;
+        let mut start_time = Instant::now();
+        let mut duration_count = 0;
 
         while let Ok(live_cell) = live_cell_receiver.recv() {
             let lock_hash = live_cell.cell_output.calc_lock_hash();
-            match self.live_cells.entry(lock_hash.clone()) {
-                std::collections::hash_map::Entry::Occupied(entry) => {
-                    if entry.get().out_point == live_cell.out_point {
-                        self.backlogs.insert(live_cell.out_point.clone(), live_cell);
+
+            if let Some(_live_cell_in_map) = self.live_cells.get(&lock_hash) {
+                self.backlogs
+                    .entry(lock_hash.clone())
+                    .or_insert_with(Vec::new)
+                    .push(live_cell);
+            } else {
+                self.live_cells.insert(lock_hash.clone(), live_cell);
+                for (hash, backlog_cells) in self.backlogs.iter_mut() {
+                    if self.live_cells.len() >= self.n_inout {
+                        break;
+                    }
+                    if !self.live_cells.contains_key(hash) && !backlog_cells.is_empty() {
+                        if let Some(backlog_cell) = backlog_cells.pop() {
+                            self.live_cells.insert(hash.clone(), backlog_cell);
+                        }
                     }
-                }
-                std::collections::hash_map::Entry::Vacant(entry) => {
-                    entry.insert(live_cell);
                 }
             }
 
@@ -175,6 +228,7 @@ impl TransactionProducer {
                     .values()
                     .map(|cell| {
                         // use tx_index as random number
+                        let lock_hash = cell.cell_output.calc_lock_hash();
                         let tx_index = cell.transaction_info.as_ref().unwrap().index;
                         let user = self.users.get(&lock_hash).expect("should be ok");
                         match tx_index % 3 {
@@ -225,23 +279,140 @@ impl TransactionProducer {
                     // SendError occurs, the corresponding transaction receiver is dead
                     return;
                 }
+                count += 1;
+                duration_count += 1;
+                if Instant::now().duration_since(start_time) >= Duration::from_secs(log_duration) {
+                    let elapsed = start_time.elapsed();
+                    ckb_testkit::info!("[TransactionProducer] producer count: {} liveCell producer remaining :{} ,duration time:{:?}, duration tps:{} ", count,live_cell_receiver.len(),elapsed,duration_count*1000/elapsed.as_millis());
+                    duration_count = 0;
+                    start_time = Instant::now();
+                }
+            }
+        }
+    }
+}
 
-                let mut backlogs = HashMap::new();
-                std::mem::swap(&mut self.backlogs, &mut backlogs);
-                for (_, live_cell) in backlogs.into_iter() {
-                    let lock_hash = live_cell.cell_output.calc_lock_hash();
-                    match self.live_cells.entry(lock_hash) {
-                        std::collections::hash_map::Entry::Occupied(entry) => {
-                            if entry.get().out_point == live_cell.out_point {
-                                self.backlogs.insert(live_cell.out_point.clone(), live_cell);
-                            }
+pub struct TransactionConsumer {
+    nodes: Vec<Node>,
+}
+
+
+impl TransactionConsumer {
+    pub fn new(nodes: Vec<Node>) -> Self {
+        Self {
+            nodes
+        }
+    }
+
+    pub async fn run(
+        self,
+        transaction_receiver: Receiver<TransactionView>,
+        max_concurrent_requests: usize,
+        t_tx_interval: Duration,
+        t_bench: Duration) {
+        let start_time = Instant::now();
+        let mut last_log_duration = Instant::now();
+        let mut benched_transactions = 0;
+        let mut duplicated_transactions = 0;
+        let mut loop_count = 0;
+        let mut i = 0;
+        let log_duration_time = 3;
+
+        let semaphore = Arc::new(Semaphore::new(max_concurrent_requests));
+        let transactions_processed = Arc::new(AtomicUsize::new(0));
+        let transactions_total_time = Arc::new(AtomicUsize::new(0));
+
+
+        // let logger_task = print_transactions_processed(transactions_processed.clone(), transactions_total_time.clone());
+        // tokio::spawn(logger_task);
+        let mut pending_tasks = FuturesUnordered::new();
+
+        loop {
+            loop_count += 1;
+            let tx = transaction_receiver
+                .recv_timeout(Duration::from_secs(60 * 3))
+                .expect("timeout to wait transaction_receiver");
+            if t_tx_interval.as_millis() != 0 {
+                async_sleep(t_tx_interval).await;
+            }
+
+            i = (i + 1) % self.nodes.len();
+            let node = self.nodes[i].clone();
+            let permit = semaphore.clone().acquire_owned().await;
+            let tx_hash = tx.hash();
+            let begin_time = Instant::now();
+            let task = async move {
+                let result = maybe_retry_send_transaction_async(&node, &tx).await;
+                drop(permit);
+                (result, tx_hash, Instant::now() - begin_time)
+            };
+
+            pending_tasks.push(tokio::spawn(task));
+            while let Some(result) = pending_tasks.next().now_or_never() {
+                transactions_processed.fetch_add(1, Ordering::Relaxed);
+
+                let mut use_time = Duration::from_millis(0);
+
+                match result {
+                    Some(Ok((Ok(is_accepted), _tx_hash, cost_time))) => {
+                        use_time = cost_time;
+                        if is_accepted {
+                            benched_transactions += 1;
+                        } else {
+                            duplicated_transactions += 1;
                         }
-                        std::collections::hash_map::Entry::Vacant(entry) => {
-                            entry.insert(live_cell);
+                    }
+                    Some(Ok((Err(err), tx_hash, cost_time))) => {
+                        use_time = cost_time;
+                        // double spending, discard this transaction
+                        ckb_testkit::info!(
+                    "consumer count :{} failed to send tx {:#x}, error: {}",
+                    loop_count,
+                    tx_hash,
+                    err
+                );
+                        if !err.contains("TransactionFailedToResolve") {
+                            ckb_testkit::error!(
+                        "failed to send tx {:#x}, error: {}",
+                        tx_hash,
+                        err
+                    );
                         }
                     }
+                    Some(Err(e)) => {
+                        eprintln!("Error in task: {:?}", e);
+                    }
+                    None => break,
+                }
+                transactions_total_time.fetch_add(use_time.as_millis() as usize, Ordering::Relaxed);
+            }
+
+            if last_log_duration.elapsed() > Duration::from_secs(log_duration_time) {
+                let elapsed = last_log_duration.elapsed();
+                last_log_duration = Instant::now();
+                let duration_count = transactions_processed.swap(0, Ordering::Relaxed);
+                let duration_total_time = transactions_total_time.swap(0, Ordering::Relaxed);
+                let mut duration_tps = 0;
+                let mut duration_delay = 0;
+                if duration_count != 0 {
+                    duration_delay = duration_total_time / (duration_count as usize);
+                    duration_tps = duration_count *1000 / (elapsed.as_millis() as usize);
                 }
+                ckb_testkit::info!(
+                "[TransactionConsumer] consumer :{} transactions, {} duplicated {} , transaction producer  remaining :{}, log duration {:?} ,duration send tx tps {},duration avg delay {}ms",
+                loop_count,
+                benched_transactions,
+                duplicated_transactions,
+                transaction_receiver.len(),
+                elapsed,
+                duration_tps,
+                duration_delay
+            );
+            }
+            if start_time.elapsed() > t_bench {
+                break;
             }
         }
     }
 }
+