Use threads to parallelize tx_flood scenario

src/scenarios/miner_std.py

@@ -7,13 +7,20 @@
 
 
 def cli_help():
-    return "Generate blocks over time. Options: [--allnodes | --interval=<number>]"
+    return "Generate blocks over time. Options: [--allnodes | --interval=<number> | --mature ]"
 
+class Miner:
+    def __init__(self, node, mature):
+        self.node = node
+        self.wallet = ensure_miner(self.node)
+        self.addr = self.wallet.getnewaddress()
+        self.mature = mature
 
 class MinerStd(WarnetTestFramework):
     def set_test_params(self):
         # This is just a minimum
         self.num_nodes = 0
+        self.miners = []
 
     def add_options(self, parser):
         parser.add_argument(
@@ -29,23 +36,38 @@ def add_options(self, parser):
             type=int,
             help="Number of seconds between block generation (default 60 seconds)",
         )
+        parser.add_argument(
+            "--mature",
+            dest="mature",
+            action="store_true",
+            help="When true, generate 101 blocks ONCE per miner",
+        )
 
     def run_test(self):
         while not self.warnet.network_connected():
-            sleep(1)
+            self.log.info("Waiting for complete network connection...")
+            sleep(5)
+        self.log.info("Network connected. Starting miners.")
 
-        current_miner = 0
+        max_miners = 1
+        if self.options.allnodes:
+            max_miners = len(self.nodes)
+        for index in range(max_miners):
+            self.miners.append(Miner(self.nodes[index], self.options.mature))
 
         while True:
-            miner = ensure_miner(self.nodes[current_miner])
-            addr = miner.getnewaddress()
-            block = self.generatetoaddress(self.nodes[current_miner], 1, addr)
-            self.log.info(f"generated block from node {current_miner}: {block}")
-            if self.options.allnodes:
-                current_miner = current_miner + 1
-                if current_miner >= self.num_nodes:
-                    current_miner = 0
-            sleep(self.options.interval)
+            for miner in self.miners:
+                num = 1
+                if miner.mature:
+                    num = 101
+                    miner.mature = False
+                try:
+                    self.generatetoaddress(miner.node, num, miner.addr, sync_fun=self.no_op)
+                    height = miner.node.getblockcount()
+                    self.log.info(f"generated {num} block(s) from node {miner.node.index}. New chain height: {height}")
+                except Exception as e:
+                    self.log.error(f"node {miner.node.index} error: {e}")
+                sleep(self.options.interval)
 
 
 if __name__ == "__main__":

src/scenarios/tx_flood.py

@@ -1,30 +1,67 @@
 #!/usr/bin/env python3
+import threading
+from random import randrange, choice
+from time import sleep
 from scenarios.utils import ensure_miner
 from warnet.test_framework_bridge import WarnetTestFramework
 
-BLOCKS = 100
-TXS = 100
-
 
 def cli_help():
-    return "Generate 100 blocks with 100 TXs each"
+    return "Make a big transaction mess"
 
 
 class TXFlood(WarnetTestFramework):
     def set_test_params(self):
         self.num_nodes = 1
+        self.addrs = []
+        self.threads = []
+
+    def add_options(self, parser):
+        parser.add_argument(
+            "--interval",
+            dest="interval",
+            default=10,
+            type=int,
+            help="Number of seconds between TX generation (default 10 seconds)",
+        )
+
+    def orders(self, node):
+        wallet = ensure_miner(node)
+        for address_type in ["legacy", "p2sh-segwit", "bech32", "bech32m"]:
+            self.addrs.append(wallet.getnewaddress(address_type=address_type))
+        while True:
+            sleep(self.options.interval)
+            try:
+                bal = wallet.getbalance()
+                if bal < 1:
+                    continue
+                amounts = {}
+                num_out = randrange(1, len(self.nodes) // 2)
+                for _ in range(num_out):
+                    sats = int(float((bal / 20) / num_out) * 1e8)
+                    amounts[choice(self.addrs)] = randrange(sats // 4, sats) / 1e8
+                wallet.sendmany(dummy="", amounts=amounts)
+                self.log.info(f"node {node.index} sent tx with {num_out} outputs")
+            except Exception as e:
+                self.log.error(f"node {node.index} error: {e}")
 
     def run_test(self):
-        miner = ensure_miner(self.nodes[0])
-        addr = miner.getnewaddress()
-        self.generatetoaddress(self.nodes[0], 200, addr)
-        for b in range(BLOCKS):
-            for t in range(TXS):
-                txid = self.nodes[0].sendtoaddress(address=addr, amount=0.001)
-                self.log.info(f"sending tx {t}/{TXS}: {txid}")
-            block = self.generate(self.nodes[0], 1)
-            self.log.info(f"generating block {b}/{BLOCKS}: {block}")
+        self.log.info(f"Starting TX mess with {len(self.nodes)} threads")
+        for node in self.nodes:
+            sleep(1) # stagger
+            t = threading.Thread(target=lambda: self.orders(node))
+            t.daemon = False
+            t.start()
+            self.threads.append({"thread": t, "node": node})
 
+        while len(self.threads) > 0:
+            for thread in self.threads:
+                if not thread["thread"].is_alive():
+                    self.log.info(f"restarting thread for node {thread['node'].index}")
+                    thread["thread"] = threading.Thread(target=lambda: self.orders(thread["node"]))
+                    thread["thread"].daemon = False
+                    thread["thread"].start()
+            sleep(30)
 
 if __name__ == "__main__":
     TXFlood().main()