From 06e0a875dd933382f9e9de4a69e228d98762aaf3 Mon Sep 17 00:00:00 2001 From: Thomas Lively Date: Fri, 11 Sep 2026 15:58:03 -0700 Subject: [PATCH 1/2] Support parallel fuzzing in monitor_fuzz.py Add a --jobs (-j) argument to monitor_fuzz.py to control how many instances of fuzz_opt.py are run in parallel. This can increase throughput because a single instance of fuzz_opt.py does not always saturate all cores. The instances are run in separate directories to avoid interfering with each other. To support running simultaneously in separate directories, make a few tweaks to fuzz_opt.py: - Run git operations explicitly in the binaryen root. - Do not create the clusterfuzz bundle if it already exists. - Write the clusterfuzz bundle to a temporary file initially and then swap it into place to avoid conflicting writes. --- scripts/fuzz_opt.py | 21 ++-- scripts/monitor_fuzz.py | 226 ++++++++++++++++++++++++++++------------ scripts/test/shared.py | 3 +- 3 files changed, 174 insertions(+), 76 deletions(-) diff --git a/scripts/fuzz_opt.py b/scripts/fuzz_opt.py index bd2effe45b0..800775f75fc 100755 --- a/scripts/fuzz_opt.py +++ b/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: @@ -1882,9 +1882,13 @@ def ensure(self): shutil.rmtree(self.clusterfuzz_dir) os.mkdir(self.clusterfuzz_dir) - print('Bundling for ClusterFuzz') - bundle = 'fuzz_opt_clusterfuzz_bundle.tgz' - run([in_binaryen('scripts', 'bundle_clusterfuzz.py'), bundle]) + bundle = in_binaryen('out', 'test', 'fuzz_opt_clusterfuzz_bundle.tgz') + if not os.path.exists(bundle): + print('Bundling for ClusterFuzz') + os.makedirs(os.path.dirname(bundle), exist_ok=True) + tmp_bundle = abspath('tmp_fuzz_opt_clusterfuzz_bundle.tgz') + run([in_binaryen('scripts', 'bundle_clusterfuzz.py'), tmp_bundle]) + os.replace(tmp_bundle, bundle) print('Unpacking for ClusterFuzz') tar = tarfile.open(bundle, "r:gz") @@ -3014,6 +3018,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 +3031,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 " " $? diff --git a/scripts/monitor_fuzz.py b/scripts/monitor_fuzz.py index e78501c1ed9..0081891dc2e 100755 --- a/scripts/monitor_fuzz.py +++ b/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,71 @@ 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') parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument( + '-j', + '--jobs', + type=int, + default=int(os.environ.get('JOBS', '1')), + help='Number of parallel fuzzers to run (default: $JOBS or 1)', + ) 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 +202,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 +219,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 +286,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 +363,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: diff --git a/scripts/test/shared.py b/scripts/test/shared.py index 74c68dc8707..0e65e159056 100644 --- a/scripts/test/shared.py +++ b/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) From 56c34a7c1e7f8c15cc00abba81e9f66f3d9f959e Mon Sep 17 00:00:00 2001 From: Thomas Lively Date: Fri, 11 Sep 2026 17:01:47 -0700 Subject: [PATCH 2/2] always create the clusterfuzz bundle --- scripts/fuzz_opt.py | 12 +++++------- 1 file changed, 5 insertions(+), 7 deletions(-) diff --git a/scripts/fuzz_opt.py b/scripts/fuzz_opt.py index 800775f75fc..bba51e34b66 100755 --- a/scripts/fuzz_opt.py +++ b/scripts/fuzz_opt.py @@ -1882,13 +1882,11 @@ def ensure(self): shutil.rmtree(self.clusterfuzz_dir) os.mkdir(self.clusterfuzz_dir) - bundle = in_binaryen('out', 'test', 'fuzz_opt_clusterfuzz_bundle.tgz') - if not os.path.exists(bundle): - print('Bundling for ClusterFuzz') - os.makedirs(os.path.dirname(bundle), exist_ok=True) - tmp_bundle = abspath('tmp_fuzz_opt_clusterfuzz_bundle.tgz') - run([in_binaryen('scripts', 'bundle_clusterfuzz.py'), tmp_bundle]) - os.replace(tmp_bundle, bundle) + print('Bundling for ClusterFuzz') + bundle = 'fuzz_opt_clusterfuzz_bundle.tgz' + tmp_bundle = 'tmp_' + bundle + run([in_binaryen('scripts', 'bundle_clusterfuzz.py'), tmp_bundle]) + os.replace(tmp_bundle, bundle) print('Unpacking for ClusterFuzz') tar = tarfile.open(bundle, "r:gz")