From 775f1330bcae0774e297d52bc9890fb58931fc72 Mon Sep 17 00:00:00 2001 From: Wade Fife Date: Sun, 8 Sep 2024 20:43:20 -0500 Subject: [PATCH] fpga: tools: Add parallel FPGA build support Add parallel FPGA build support to repeat_fpga_build.py. Original-commit: b6ce5671f89cd10107b0f83524c4fd86a0a358b6 --- tools/utils/repeat_fpga_build.py | 475 ++++++++++++++++++++++++++----- 1 file changed, 400 insertions(+), 75 deletions(-) diff --git a/tools/utils/repeat_fpga_build.py b/tools/utils/repeat_fpga_build.py index 4f02178..b38ef35 100755 --- a/tools/utils/repeat_fpga_build.py +++ b/tools/utils/repeat_fpga_build.py @@ -13,12 +13,152 @@ stop if a an unrecognized error occurs. """ import sys +import os +import signal +import time import argparse import subprocess +import threading import logging import re import random +from enum import Enum + + +# Delay between build status checks +SLEEP_DELAY = 3 + + +class Status(Enum): + # Build status constants. Values can be returned to shell as error codes. + SUCCESS = 0 + ERROR_TRANSIENT = 1 + ERROR_UNKNOWN = 2 + CANCELLED = 3 + TIMEOUT = 4 + NONE = 5 + + def __str__(self): + """Returns a pretty string for each enum.""" + status_string = { + self.SUCCESS: "Success", + self.ERROR_TRANSIENT: "Error (Transient)", + self.ERROR_UNKNOWN: "Error (Unknown)", + self.CANCELLED: "Cancelled", + self.TIMEOUT: "Timeout", + self.NONE: "Not Started", + } + return status_string[self] + + +class Worker: + """Information for a single worker.""" + + def __init__(self): + self.status = None + self.thread = None + self.process = None + self.start_time = None + + +class Workers: + """A set of workers and methods for adding and checking the status of workers.""" + + def __init__(self): + self.thread_lock = threading.Lock() + self.worker_list = [] + + def new(self): + """Create a new worker and returns its index.""" + with self.thread_lock: + new_worker = Worker() + self.worker_list.append(new_worker) + num = len(self.worker_list) - 1 + return num + + def set(self, index, status=None, thread=None, process=None, start_time=None): + """Set the status of a worker. + + Args: + index: + The index for the worker to update. + status: + If not None, set status to Status.SUCCESS, Status.ERROR_UNKNOWN, etc. + thread: + If not None, set thread to a Thread object. + process: + If not None, set process to a Popen object. + start_time: + If not None, set start_time to time worker started, in seconds. + """ + with self.thread_lock: + if status is not None: + self.worker_list[index].status = status + if thread is not None: + self.worker_list[index].thread = thread + if process is not None: + self.worker_list[index].process = process + if start_time is not None: + self.worker_list[index].start_time = start_time + + def status(self, index): + """Returns the status of the indicated worker.""" + with self.thread_lock: + return self.worker_list[index].status + + def num_workers(self): + """Return the total number of workers.""" + with self.thread_lock: + return len(self.worker_list) + + def num_finished(self): + """Return the number of workers that have finished.""" + with self.thread_lock: + return sum(int(w.status is not None) for w in self.worker_list) + + def num_error_unknown(self): + """Return the number of workers that finished with an unknown error.""" + with self.thread_lock: + return sum(int(w.status == Status.ERROR_UNKNOWN) for w in self.worker_list) + + def any_success(self): + """Returns True if any worker finished successfully.""" + with self.thread_lock: + return any( + not w.thread.is_alive() and w.status == Status.SUCCESS + for w in self.worker_list + ) + + def check_timeout(self, timeout): + """Check the timeout status of each worker. Kill it if timeout has + expired. + + Args: + timeout: The timeout in seconds. + """ + with self.thread_lock: + for index, worker in enumerate(self.worker_list): + if time.time() - worker.start_time > timeout and worker.status == None: + try: + self.worker_list[index].status = Status.TIMEOUT + os.killpg(os.getpgid(worker.process.pid), signal.SIGTERM) + except ProcessLookupError: + # Process is already dead + pass + + def kill_jobs(self): + """Kill any running jobs by sending SIGTERM to each process group.""" + with self.thread_lock: + for index, worker in enumerate(self.worker_list): + if worker.status == None: + try: + self.worker_list[index].status = Status.CANCELLED + os.killpg(os.getpgid(worker.process.pid), signal.SIGTERM) + except ProcessLookupError: + # Process is already dead + pass + def parse_args(): """Parse the command line arguments. @@ -55,6 +195,13 @@ def parse_args(): required=False, help="Specifies the FPGA directory to pass to rfnoc_image_builder using the -F argument.", ) + parser.add_argument( + "--vivado-path", + "-P", + help="Path to the base install for Xilinx Vivado if not in default " + "location (e.g., /tools/Xilinx/Vivado).", + default=None, + ) parser.add_argument( "--num", "-n", @@ -81,69 +228,124 @@ def parse_args(): required=False, help="Initial seed value to use.", ) + parser.add_argument( + "--fpga-jobs", + "-j", + type=int, + default=1, + required=False, + help="Number of parallel FPGA build jobs to run.", + ) + parser.add_argument( + "--ip-jobs", + "-J", + type=int, + default=4, + required=False, + help="Number of parallel IP build jobs to run.", + ) + parser.add_argument( + "--timeout", + "-T", + type=float, + default=0, + help=("Set a timeout in hours for each build attempt. Use 0 for no timeout."), + ) + parser.add_argument( + "--all", + "-a", + action="store_true", + default=False, + help=( + "Complete all builds, even after a successful build is completed, " + "unless an unknown error occurs." + ), + ) + parser.add_argument( + "--index", + "-i", + type=int, + required=False, + default=1, + help=( + "Starting index for the number to be appended to each parallel " + "build's directory name." + ), + ) return parser.parse_args() -def prepare_build(target, image_core, image_core_name, fpga_dir): +def rfnoc_image_builder_cmd( + build_seed, + target, + image_core, + image_core_name, + fpga_dir, + vivado_path, + build_num=None, +): """ - Run tasks to prepare the build. In particular, execute the RFNoC image builder - if desired. + Build the RFNoC image builder command for this build """ - if image_core: - logging.info("Calling rfnoc_image_builder to prepare FPGA build.") - cmd = [ - "rfnoc_image_builder", - "--yaml-config", - image_core, - "--generate-only", - "--no-hash", - "--no-date", - ] - if target: - cmd += ["--target", target] - if image_core_name: - cmd += ["--image_core_name", image_core_name] - if fpga_dir: - cmd += ["--fpga-dir", fpga_dir] - #logging.info("Command:", cmd) - result = subprocess.run( - cmd, - check=False, - encoding="utf-8", - capture_output=True, - ) - logging.info("rfnoc_image_builder output:") - logging.info("stdout:\n%s", result.stdout) - logging.info("stderr:\n%s", result.stderr) - if result.returncode: - logging.error("Image builder failed! Consult output for details.") - result.check_returncode() - # Parse the image builder output to get the make command we need to build the FPGA - make_command = re.search(r": (make [^\x1b\n\r]+)", result.stderr, flags=re.M).group(1) - logging.info("Using make command: %s", make_command) - else: - assert target - make_command = f"make {target}" - return { - "make_command": make_command, - } + cmd = "" + if build_seed is not None: + cmd += f"BUILD_SEED={build_seed} " + cmd += f"rfnoc_image_builder --yaml-config {image_core} " + if target: + cmd += f"--target {target} " + if image_core_name: + cmd += f"--image-core-name {image_core_name}" + cmd += " " if build_num == None else f"_{build_num:02} " + if fpga_dir: + cmd += f"--fpga-dir {fpga_dir} " + if vivado_path: + cmd += f"--vivado-path {vivado_path} " + return cmd -def run_fpga_build(build_seed, cfg): +def run_fpga_build( + workers, + worker_num, + build_num, + build_seed, + target, + image_core, + image_core_name, + fpga_dir, + vivado_path, + parallel_builds, +): """Performs one iteration of an FPGA build. Args: - build_seed: 32-bit signed integer to seed the FPGA build. - cfg: Build configuration info + workers: Workers object containing the state of all workers + woker_num: The number ID for this worker + build_num: Build number to be displayed for this build + build_seed: 32-bit signed integer to seed the FPGA build + target: --target argument to be passed + image_core: --image-core argument to be passed + image_core_name: --image-core-name argument to be passed + fpga_dir: --fpga-dir argument to be passed + vivado_path: --vivado-path argument to be passed Returns: - 0: The build succeeded. - 1: There was a timing or other error that might not reoccur. - 2: There was some other error that should cause us to stop trying. + Status.SUCCESS: The build succeeded + Status.*: The build failed for the indicated reason """ + logging.info(f"Starting FPGA build {build_num} with seed {build_seed}") output = "" - make_cmd = cfg["make_command"] - cmd = f'/bin/bash -c "{make_cmd} BUILD_SEED={build_seed}"' + # Give each build a unique directory number if doing parallel builds + cmd_build_num = build_num if parallel_builds else None + cmd = rfnoc_image_builder_cmd( + build_seed, + target, + image_core, + image_core_name, + fpga_dir, + vivado_path, + cmd_build_num, + ) + logging.info(f"Running FPGA build command: {cmd}") with subprocess.Popen( cmd, shell=True, @@ -151,11 +353,20 @@ def run_fpga_build(build_seed, cfg): stderr=subprocess.STDOUT, bufsize=1, universal_newlines=True, + start_new_session=True, ) as proc: + workers.set(worker_num, process=proc) + prefix = f"[{build_num:02}] " for line in proc.stdout: - print(line, end="") + print(prefix + line, end="") output += line - if proc.returncode != 0: + if proc.returncode == 0: + logging.info(f"Finished FPGA build {build_num} (SUCCESS)") + status = Status.SUCCESS + elif workers.status(worker_num) in (Status.CANCELLED, Status.TIMEOUT): + status = workers.status(worker_num) + logging.info((f"Finished FPGA build {build_num} ({str(status)})")) + else: # Regular expressions for error strings to search for that would tell # us we should try again. transient_errors = [ @@ -167,11 +378,55 @@ def run_fpga_build(build_seed, cfg): "Trying to tool lock on already tool locked arc" ), ] + logging.info(f"Finished FPGA build {build_num} (FAILURE)") + status = Status.ERROR_UNKNOWN for error_string in transient_errors: if re.search(error_string, output): - return 1 - return 2 - return 0 + status = Status.ERROR_TRANSIENT + break + workers.set(worker_num, status=status) + return status + + +def run_ip_build(ip_jobs, target, image_core, image_core_name, fpga_dir, vivado_path): + """Performs the IP build. + + This is done separately so that when we do parallel FPGA builds, they don't + separately launch IP builds that would clobber each other. + + Args: + ip_jobs: Number or parallel jobs to use for IP build + target: --target argument to be passed + image_core: --image-core argument to be passed + image_core_name: --image-core-name argument to be passed + fpga_dir: --fpga-dir argument to be passed + vivado_path: --vivado-path argument to be passed + + Returns: + 0: The IP build succeeded + non-zero: The IP build failed + """ + cmd = rfnoc_image_builder_cmd( + None, target, image_core, image_core_name, fpga_dir, vivado_path, None + ) + cmd += f" --ip-only --jobs {ip_jobs}" + logging.info(f"Running IP build with command: {cmd}") + output = "" + with subprocess.Popen( + f'/bin/bash -c "{cmd}"', + shell=True, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + bufsize=1, + universal_newlines=True, + ) as proc: + for line in proc.stdout: + print(line, end="") + output += line + if proc.returncode != 0: + logging.error("IP build failed! Consult output for details.") + return Status.ERROR_UNKNOWN + return Status.SUCCESS def next_build_seed(previous_seed): @@ -193,39 +448,109 @@ def main(): """Run the requested builds. Returns: - The status of the last build (0 if successful, non-zero if the build - failed). + The status of the build as an integer, based on the Status enum (0 if + successful, non-zero if the build failed). """ logging.basicConfig(format="[REPEAT BUILD][%(levelname)s] %(message)s") logging.root.setLevel(logging.INFO) args = parse_args() if not args.target and not args.image_core: logging.error("Either --target or --image-core must be provided!") - return 1 + return Status.NONE.value build_seed = args.seed - status = 128 - cfg = prepare_build(args.target, args.image_core, args.image_core_name, args.fpga_dir) + workers = Workers() + + if args.fpga_jobs > 1 and not args.image_core_name: + logging.error( + "You must provide --image-core-name when doing parallel FPGA " + "builds. Each build will use this name followed by a build number." + ) + return Status.NONE.value + + # Build all the IP first, so it's ready to go for any parallel FPGA builds + status = run_ip_build( + args.ip_jobs, + args.target, + args.image_core, + args.image_core_name, + args.fpga_dir, + args.vivado_path, + ) + if status != Status.SUCCESS: + return status.value + + # Run the FPGA builds next + status = Status.NONE try: - for build_num in range(1, args.num + 1): - logging.info("Starting FPGA build %d with seed %s", build_num, build_seed) - status = run_fpga_build(build_seed, cfg) - logging.info("Finished FPGA build %d", build_num) - if status == 0: - logging.info("FPGA build succeeded on attempt number %s", build_num) - break - if build_num == args.num: - logging.error("Reached maximum number of FPGA build attempts") - elif status == 1: - logging.info("FPGA build will be restarted due to unsuccessful attempt") - elif status == 2 and not args.persistent: - logging.error("Stopping due to unexpected FPGA build error") - break - build_seed = next_build_seed(build_seed) + while True: + # See if we can start another build job + if ( + threading.active_count() < args.fpga_jobs + 1 + and workers.num_workers() < args.num + ): + worker_num = workers.new() + build_num = worker_num + args.index + thread = threading.Thread( + target=run_fpga_build, + args=( + workers, + worker_num, + build_num, + build_seed, + args.target, + args.image_core, + args.image_core_name, + args.fpga_dir, + args.vivado_path, + args.fpga_jobs > 1, + ), + ) + workers.set(worker_num, thread=thread, start_time=time.time()) + thread.start() + build_seed = next_build_seed(build_seed) + else: + # We've maxed out the number of workers/jobs, so check the + # status periodically. + time.sleep(SLEEP_DELAY) + if args.timeout > 0: + workers.check_timeout(args.timeout * 3600) + if workers.any_success(): + if not args.all or workers.num_finished() == args.num: + break + elif workers.num_finished() == args.num: + break + + # We're done, so kill any remaining jobs that haven't finished + if threading.active_count() > 1: + logging.info("Stopping remaining jobs . . .") + workers.kill_jobs() + while threading.active_count() > 1: + time.sleep(0.1) + + # Print the results before returning + logging.info(f"Build Summary:") + for worker_num in range(workers.num_workers()): + logging.info( + f" - Build attempt {args.index + worker_num}: {str(workers.status(worker_num))}" + ) + if workers.any_success(): + status = Status.SUCCESS + elif workers.num_finished() == args.num: + status = Status.ERROR_TRANSIENT + logging.error("Reached maximum number of FPGA build attempts") + elif workers.num_error_unknown() != 0 and args.stop_on_error: + status = Status.ERROR_UNKNOWN + logging.error("Stopped due to unexpected FPGA build error") + except KeyboardInterrupt: logging.info("Received SIGINT. Aborting . . .") + workers.kill_jobs() + while threading.active_count() > 1: + time.sleep(0.1) # Return normal Bash value for SIGINT (128+2) return 130 - return status + + return status.value if __name__ == "__main__":