Support parallel fuzzing in monitor_fuzz.py

scripts/fuzz_opt.py

@@ -272,12 +272,12 @@ def auto_select_recent_initial_contents():
         # commit time of HEAD. The reason we use the commit time of HEAD instead
         # of the current system time is to make the results deterministic given
         # the Binaryen HEAD commit.
-        head_ts_str = run(['git', 'log', '-1', '--format=%cd', '--date=raw'],
+        head_ts_str = run(['git', '-C', shared.options.binaryen_root, 'log', '-1', '--format=%cd', '--date=raw'],
                           silent=True).split()[0]
         head_dt = datetime.utcfromtimestamp(int(head_ts_str))
         start_dt = head_dt - timedelta(days=RECENT_DAYS)
         start_ts = start_dt.replace(tzinfo=timezone.utc).timestamp()
-        log = run(['git', 'log', '--name-status', '--format=', '--date=raw', '--no-renames', f'--since={start_ts}'], silent=True).splitlines()
+        log = run(['git', '-C', shared.options.binaryen_root, 'log', '--name-status', '--format=', '--date=raw', '--no-renames', f'--since={start_ts}'], silent=True).splitlines()
         # Pick up lines in the form of
         # A       test/../something.wast
         # M       test/../something.wast
@@ -290,7 +290,7 @@ def auto_select_recent_initial_contents():
 
     def is_git_repo():
         try:
-            ret = run(['git', 'rev-parse', '--is-inside-work-tree'],
+            ret = run(['git', '-C', shared.options.binaryen_root, 'rev-parse', '--is-inside-work-tree'],
                       silent=True, stderr=subprocess.DEVNULL)
             return ret == 'true\n'
         except subprocess.CalledProcessError:
@@ -3014,6 +3014,7 @@ def get_random_opts():
                 working_wasm = abspath('w.wasm')
                 wasm_reduce = in_bin('wasm-reduce')
                 reduce_sh = abspath('reduce.sh')
+                fuzz_opt = in_binaryen('scripts', 'fuzz_opt.py')
                 features = ' '.join(FEATURE_OPTS)
                 with open('reduce.sh', 'w') as f:
                     f.write(f'''\
@@ -3026,12 +3027,12 @@ def get_random_opts():
 
 if [ -z "$BINARYEN_FIRST_WASM" ]; then
   # run the command normally
-  ./scripts/fuzz_opt.py {auto_init} --binaryen-bin {binaryen_bin} {seed} {temp_wasm} > o 2> e
+  {fuzz_opt} {auto_init} --binaryen-bin {binaryen_bin} {seed} {temp_wasm} > o 2> e
 else
   # BINARYEN_FIRST_WASM was provided so we should actually reduce the *second*
   # file. pass the first one in as the main file, and use the env var for the
   # second.
-  BINARYEN_SECOND_WASM={temp_wasm} ./scripts/fuzz_opt.py {auto_init} --binaryen-bin {binaryen_bin} {seed} $BINARYEN_FIRST_WASM > o 2> e
+  BINARYEN_SECOND_WASM={temp_wasm} {fuzz_opt} {auto_init} --binaryen-bin {binaryen_bin} {seed} $BINARYEN_FIRST_WASM > o 2> e
 fi
 
 echo "  " $?

scripts/monitor_fuzz.py

@@ -54,9 +54,7 @@ def __init__(self, log_path, max_lines, keep_lines, truncate_interval):
                 with open(log_path, encoding='utf-8', errors='replace') as f:
                     for line in f:
                         self.deque.append(line)
-                        self.recent_lines.append(line)
                         self.lines_written += 1
-                        self._parse_line(line)
             except Exception:
                 pass
 
@@ -115,20 +113,81 @@ def get_status(self):
             )
 
 
+class FuzzerWorker:
+    """Manages a single fuzzer subprocess and its monitor."""
+
+    def __init__(
+        self,
+        worker_id,
+        work_dir,
+        cmd,
+        env,
+        max_lines,
+        keep_lines,
+        truncate_interval,
+    ):
+        self.id = worker_id
+        self.work_dir = work_dir
+        os.makedirs(work_dir, exist_ok=True)
+        self.log_path = os.path.join(work_dir, 'fuzz.log')
+        self.monitor = FuzzMonitor(
+            log_path=self.log_path,
+            max_lines=max_lines,
+            keep_lines=keep_lines,
+            truncate_interval=truncate_interval,
+        )
+        worker_env = env.copy()
+        worker_env['BINARYEN_OUT_DIR'] = work_dir
+        self.proc = subprocess.Popen(
+            cmd,
+            cwd=work_dir,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.STDOUT,
+            text=True,
+            bufsize=1,
+            env=worker_env,
+            errors='replace',
+            start_new_session=True,
+        )
+        self.reader_thread = threading.Thread(
+            target=self.monitor.run,
+            args=(self.proc.stdout,),
+            daemon=True,
+        )
+        self.reader_thread.start()
+
+
 def parse_args():
-    default_log_dir = os.path.join(
-        os.path.dirname(os.path.abspath(__file__)), 'out', 'test')
+    # N.B. We could alternatively `import shared from test`, which has the side
+    # effect of changing the current directory to <binaryen_root>/out/test, but
+    # this is less magical.
+    binaryen_root = os.path.dirname(
+        os.path.dirname(os.path.abspath(__file__)))
+    default_log_dir = os.path.join(binaryen_root, 'out', 'test')
+    cores = os.cpu_count() or 1
     parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument(
+        '-j',
+        '--jobs',
+        type=int,
+        nargs='?',
+        const=cores,
+        default=int(os.environ.get('JOBS', '1')),
+        help=(
+            f'Number of parallel fuzzers to run (default: $JOBS or 1; '
+            f'defaults to {cores} if passed without an argument)'
+        ),
+    )
     parser.add_argument(
         '--log-dir',
         default=os.environ.get('LOG_DIR', default_log_dir),
-        help='Directory to save fuzz.log (default: $LOG_DIR or ./out/test)',
+        help='Directory to save fuzz logs (default: $LOG_DIR or out/test)',
     )
     parser.add_argument(
         '--max-iters',
         type=int,
         default=int(os.environ.get('MAX_ITERS', '0')),
-        help='Stop after N iterations (0 for infinite, default: $MAX_ITERS or 0)',
+        help='Stop after N total iterations across all fuzzers (0 for infinite, default: $MAX_ITERS or 0)',
     )
     parser.add_argument(
         '--truncate-interval',
@@ -153,7 +212,10 @@ def parse_args():
         nargs=argparse.REMAINDER,
         help='Fuzzer command to run (default: ./scripts/fuzz_opt.py)',
     )
-    return parser.parse_args()
+    args = parser.parse_args()
+    if args.jobs < 1:
+        parser.error('--jobs must be at least 1')
+    return args
 
 
 def main():
@@ -167,58 +229,65 @@ def main():
             os.path.dirname(os.path.abspath(__file__)), 'fuzz_opt.py',
         )
         cmd = [sys.executable, default_fuzzer]
-
-    os.makedirs(args.log_dir, exist_ok=True)
-    log_file_path = os.path.join(args.log_dir, 'fuzz.log')
-
-    monitor = FuzzMonitor(
-        log_path=log_file_path,
-        max_lines=args.max_lines,
-        keep_lines=args.keep_lines,
-        truncate_interval=args.truncate_interval,
-    )
+    else:
+        cmd = [
+            os.path.abspath(arg) if os.path.exists(arg) else arg for arg in cmd
+        ]
 
     env = os.environ.copy()
     env['PYTHONUNBUFFERED'] = '1'
 
-    proc = subprocess.Popen(
-        cmd,
-        stdout=subprocess.PIPE,
-        stderr=subprocess.STDOUT,
-        text=True,
-        bufsize=1,
-        env=env,
-        errors='replace',
-        start_new_session=True,
-    )
-
-    print(f'Fuzzer started with PID {proc.pid}. Monitoring...', flush=True)
+    workers = []
+    for i in range(args.jobs):
+        work_dir = os.path.join(args.log_dir, str(i))
+        workers.append(
+            FuzzerWorker(
+                worker_id=i,
+                work_dir=work_dir,
+                cmd=cmd,
+                env=env,
+                max_lines=args.max_lines,
+                keep_lines=args.keep_lines,
+                truncate_interval=args.truncate_interval,
+            ),
+        )
 
-    reader_thread = threading.Thread(
-        target=monitor.run,
-        args=(proc.stdout,),
-        daemon=True,
-    )
-    reader_thread.start()
+    if len(workers) == 1:
+        print(
+            f'Fuzzer started with PID {workers[0].proc.pid}. Monitoring...',
+            flush=True,
+        )
+    else:
+        pids = ', '.join(str(w.proc.pid) for w in workers)
+        print(
+            f'Started {len(workers)} fuzzers with PIDs {pids}. Monitoring...',
+            flush=True,
+        )
 
-    def stop_child():
-        if proc.poll() is None:
-            try:
-                os.killpg(proc.pid, signal.SIGTERM)
-            except ProcessLookupError:
-                pass
-            try:
-                proc.wait(timeout=5)
-            except subprocess.TimeoutExpired:
+    def stop_children():
+        for w in workers:
+            if w.proc.poll() is None:
                 try:
-                    os.killpg(proc.pid, signal.SIGKILL)
+                    os.killpg(w.proc.pid, signal.SIGTERM)
                 except ProcessLookupError:
                     pass
-                proc.wait()
+        deadline = time.time() + 5.0
+        for w in workers:
+            if w.proc.poll() is None:
+                remaining = max(0.0, deadline - time.time())
+                try:
+                    w.proc.wait(timeout=remaining)
+                except subprocess.TimeoutExpired:
+                    try:
+                        os.killpg(w.proc.pid, signal.SIGKILL)
+                    except ProcessLookupError:
+                        pass
+                    w.proc.wait()
 
     def signal_handler(signum, _frame):
-        stop_child()
-        reader_thread.join(timeout=2.0)
+        stop_children()
+        for w in workers:
+            w.reader_thread.join(timeout=2.0)
         sys.exit(128 + signum)
 
     signal.signal(signal.SIGINT, signal_handler)
@@ -227,39 +296,75 @@ def signal_handler(signum, _frame):
     start_time = time.time()
     last_report = 0
     limit_reached = False
+    stopped_worker = None
 
     try:
-        while reader_thread.is_alive() or proc.poll() is None:
-            reader_thread.join(timeout=1.0)
+        while any(
+            w.reader_thread.is_alive() or w.proc.poll() is None for w in workers
+        ):
+            time.sleep(0.2)
             now = time.time()
             elapsed = int(now - start_time)
 
             minute = elapsed // 60
-            latest_iter = monitor.get_progress()
+            total_iters = sum(w.monitor.get_progress() for w in workers)
 
             if minute > last_report:
                 last_report = minute
                 timestamp = time.strftime('%H:%M:%S')
                 print(
-                    f'[{timestamp}] Runtime: {last_report} min, Latest'
-                    f' Iteration: {latest_iter}',
+                    f'[{timestamp}] Runtime: {last_report} min,'
+                    f' Iterations: {total_iters}',
                     flush=True,
                 )
 
-            if args.max_iters > 0 and latest_iter >= args.max_iters:
+            if args.max_iters > 0 and total_iters >= args.max_iters:
+                fuzzer_str = 'fuzzer' if len(workers) == 1 else 'fuzzers'
                 print(
                     f'Reached max iterations ({args.max_iters}). Stopping'
-                    ' fuzzer...',
+                    f' {fuzzer_str}...',
                     flush=True,
                 )
                 limit_reached = True
-                stop_child()
+                stop_children()
                 break
-    finally:
-        stop_child()
-        reader_thread.join(timeout=5.0)
 
-    exit_code = proc.returncode
+            should_stop = False
+            for w in workers:
+                if w.monitor.get_status()[0]:
+                    try:
+                        w.proc.wait(timeout=2.0)
+                    except subprocess.TimeoutExpired:
+                        pass
+                    w.reader_thread.join(timeout=2.0)
+                    stopped_worker = w
+                    should_stop = True
+                    break
+                if w.proc.poll() is not None:
+                    w.reader_thread.join(timeout=2.0)
+                    stopped_worker = w
+                    should_stop = True
+                    break
+
+            if should_stop:
+                stop_children()
+                break
+    finally:
+        stop_children()
+        for w in workers:
+            w.reader_thread.join(timeout=5.0)
+
+    for w in workers:
+        bug_found, iteration, seed, _ = w.monitor.get_status()
+        if bug_found:
+            print('SUCCESS: Bug found!')
+            if len(workers) > 1:
+                print(f'Fuzzer: {w.id}')
+            print(f'Directory: {w.work_dir}')
+            print(f'Iteration: {iteration}')
+            print(f'Seed: {seed}')
+            print(f'Exit code: {w.proc.returncode}')
+            return 0
 
     if limit_reached:
         print(
@@ -268,17 +373,14 @@ def signal_handler(signum, _frame):
         )
         return 0
 
-    bug_found, iteration, seed, recent_lines = monitor.get_status()
-
-    if bug_found:
-        print('SUCCESS: Bug found!')
-        print(f'Iteration: {iteration}')
-        print(f'Seed: {seed}')
-        print(f'Exit code: {exit_code}')
-        return 0
+    failed_worker = stopped_worker or workers[0]
+    _, _, _, recent_lines = failed_worker.monitor.get_status()
 
     print('FAILURE: Fuzzer stopped unexpectedly without finding a bug.')
-    print(f'Exit code: {exit_code}')
+    if len(workers) > 1:
+        print(f'Fuzzer: {failed_worker.id}')
+    print(f'Directory: {failed_worker.work_dir}')
+    print(f'Exit code: {failed_worker.proc.returncode}')
     if recent_lines:
         print('Last 20 lines of log:')
         for line in recent_lines:

scripts/test/shared.py

@@ -216,7 +216,8 @@ def run_test_with_wrapped_stdout(test):
 options.binaryen_test = os.path.join(options.binaryen_root, 'test')
 
 if not options.out_dir:
-    options.out_dir = os.path.join(options.binaryen_root, 'out', 'test')
+    default_out_dir = os.path.join(options.binaryen_root, 'out', 'test')
+    options.out_dir = os.environ.get('BINARYEN_OUT_DIR', default_out_dir)
 
 if not os.path.exists(options.out_dir):
     os.makedirs(options.out_dir)