Merge #18688: fuzz: Run in parallel

fa66280 fuzz: Run in parallel (MarcoFalke)

Pull request description:

  Can be reviewed with `--ignore-all-space`

ACKs for top commit:
  practicalswift:
    ACK fa66280 -- patch looks correct and the live demo in bitcoin-core/qa-assets#11 (comment) convinced me: this is good!
  fanquake:
    ACK fa66280 - this seems sane, and clearly provides some speedup. I'd post benchmarks but can't seem to get through a full run of `./test/fuzz/test_runner.py` without hitting at least a few crashes; see #18763, 18762.

Tree-SHA512: d3b545ca90c75bed27f08fe712399d0ed1ac36b13fb289c83e5606eee8dec4c19f5f5cf91758f0a6b1606d8d6b455fbe46df2588faffe7462b185bd34dc2baaf

Author: MarcoFalke

test/fuzz/test_runner.py

@@ -5,12 +5,13 @@
 """Run fuzz test targets.
 """
 
+from concurrent.futures import ThreadPoolExecutor, as_completed
 import argparse
 import configparser
+import logging
 import os
-import sys
 import subprocess
-import logging
+import sys
 
 
 def main():
@@ -35,6 +36,12 @@ def main():
         '--exclude',
         help="A comma-separated list of targets to exclude",
     )
+    parser.add_argument(
+        '--par',
+        type=int,
+        default=4,
+        help='How many targets to merge or execute in parallel.',
+    )
     parser.add_argument(
         'seed_dir',
         help='The seed corpus to run on (must contain subfolders for each fuzz target).',
@@ -124,25 +131,29 @@ def main():
         logging.error("subprocess timed out: Currently only libFuzzer is supported")
         sys.exit(1)
 
-    if args.m_dir:
-        merge_inputs(
+    with ThreadPoolExecutor(max_workers=args.par) as fuzz_pool:
+        if args.m_dir:
+            merge_inputs(
+                fuzz_pool=fuzz_pool,
+                corpus=args.seed_dir,
+                test_list=test_list_selection,
+                build_dir=config["environment"]["BUILDDIR"],
+                merge_dir=args.m_dir,
+            )
+            return
+
+        run_once(
+            fuzz_pool=fuzz_pool,
             corpus=args.seed_dir,
             test_list=test_list_selection,
             build_dir=config["environment"]["BUILDDIR"],
-            merge_dir=args.m_dir,
+            use_valgrind=args.valgrind,
         )
-        return
-
-    run_once(
-        corpus=args.seed_dir,
-        test_list=test_list_selection,
-        build_dir=config["environment"]["BUILDDIR"],
-        use_valgrind=args.valgrind,
-    )
 
 
-def merge_inputs(*, corpus, test_list, build_dir, merge_dir):
+def merge_inputs(*, fuzz_pool, corpus, test_list, build_dir, merge_dir):
     logging.info("Merge the inputs in the passed dir into the seed_dir. Passed dir {}".format(merge_dir))
+    jobs = []
     for t in test_list:
         args = [
             os.path.join(build_dir, 'src', 'test', 'fuzz', t),
@@ -153,12 +164,20 @@ def merge_inputs(*, corpus, test_list, build_dir, merge_dir):
         ]
         os.makedirs(os.path.join(corpus, t), exist_ok=True)
         os.makedirs(os.path.join(merge_dir, t), exist_ok=True)
-        logging.debug('Run {} with args {}'.format(t, args))
-        output = subprocess.run(args, check=True, stderr=subprocess.PIPE, universal_newlines=True).stderr
-        logging.debug('Output: {}'.format(output))
 
+        def job(t, args):
+            output = 'Run {} with args {}\n'.format(t, " ".join(args))
+            output += subprocess.run(args, check=True, stderr=subprocess.PIPE, universal_newlines=True).stderr
+            logging.debug(output)
+
+        jobs.append(fuzz_pool.submit(job, t, args))
+
+    for future in as_completed(jobs):
+        future.result()
 
-def run_once(*, corpus, test_list, build_dir, use_valgrind):
+
+def run_once(*, fuzz_pool, corpus, test_list, build_dir, use_valgrind):
+    jobs = []
     for t in test_list:
         corpus_path = os.path.join(corpus, t)
         os.makedirs(corpus_path, exist_ok=True)
@@ -169,18 +188,26 @@ def run_once(*, corpus, test_list, build_dir, use_valgrind):
         ]
         if use_valgrind:
             args = ['valgrind', '--quiet', '--error-exitcode=1'] + args
-        logging.debug('Run {} with args {}'.format(t, args))
-        result = subprocess.run(args, stderr=subprocess.PIPE, universal_newlines=True)
-        output = result.stderr
-        logging.debug('Output: {}'.format(output))
+
+        def job(t, args):
+            output = 'Run {} with args {}'.format(t, args)
+            result = subprocess.run(args, stderr=subprocess.PIPE, universal_newlines=True)
+            output += result.stderr
+            return output, result
+
+        jobs.append(fuzz_pool.submit(job, t, args))
+
+    for future in as_completed(jobs):
+        output, result = future.result()
+        logging.debug(output)
         try:
             result.check_returncode()
         except subprocess.CalledProcessError as e:
             if e.stdout:
                 logging.info(e.stdout)
             if e.stderr:
                 logging.info(e.stderr)
-            logging.info("Target \"{}\" failed with exit code {}: {}".format(t, e.returncode, " ".join(args)))
+            logging.info("Target \"{}\" failed with exit code {}".format(" ".join(result.args), e.returncode))
             sys.exit(1)