Vitis Unified Backend

.gitignore

@@ -7,10 +7,14 @@ sdist/
 *.egg-info/
 vivado_prj
 .vscode
+.idea
 my-hls-test
 *.tar.gz
 docs/_build
 docs/autodoc/*
 hls4mlprj_*
 *~
 *.ipynb_checkpoints/
+
+test/pytest/test_backend/input_file/*
+test/pytest/test_backend/output_file/*

hls4ml/backends/__init__.py

@@ -11,9 +11,13 @@
 
 from hls4ml.backends.vitis.vitis_backend import VitisBackend  # isort: skip
 
+from hls4ml.backends.vitis_unified.vitis_unified_backend import VitisUnifiedBackend  # isort: skip
+
+
 register_backend('Vivado', VivadoBackend)
 register_backend('VivadoAccelerator', VivadoAcceleratorBackend)
 register_backend('Vitis', VitisBackend)
+register_backend('VitisUnified', VitisUnifiedBackend)
 register_backend('Quartus', QuartusBackend)
 register_backend('Catapult', CatapultBackend)
 register_backend('SymbolicExpression', SymbolicExpressionBackend)

hls4ml/backends/vitis_unified/passes/fifo_depth_optimization.py

@@ -0,0 +1,113 @@
+# we inherit it from vitis
+import zipfile
+
+from hls4ml.backends.vitis.passes.fifo_depth_optimization import (
+    generate_depths_file,
+    initialize_large_fifos,
+    set_optimized_fifo_depths,
+)
+from hls4ml.model.optimizer.optimizer import ConfigurableOptimizerPass, ModelOptimizerPass
+
+
+def get_vitis_optimized_fifo_depths(model, cus_hls_prj_path=None):
+    """Parse the files generated by the co-simulation to retrieve the optimized depths for the FIFOs.
+    Attention, only the FIFOs between the layers are profiled!
+
+    Args:
+        model (ModelGraph): The model to which FIFO depth optimization is applied.
+
+    Returns:
+        Dict[str, int]: A dictionary that contains the FIFO names as keys and the optimized depths as values.
+    """
+    # channel.zip is generated after the co-simulation and contains the chan_status*.csv files
+    # in the chan_status*.csv files the max depth achieved during co-simulation can be found at the last (4th) line
+
+    if cus_hls_prj_path is None:
+        cus_hls_prj_path = model.config.get_output_dir() + '/' + model.config.get_project_name() + '/_prj/solution1'
+
+    path_to_zip_file = cus_hls_prj_path + '/.autopilot/db/channel_depth_info/'
+
+    with zipfile.ZipFile(f'{path_to_zip_file}channel.zip', 'r') as zip_ref:
+        zip_ref.extractall(path_to_zip_file)
+
+    # the channel_info.csv file contains the mapping of each fifo name (i.e layer4_out_U) to the respective
+    # chan_status*.csv file
+    names_file_path = cus_hls_prj_path + '/.autopilot/db/channel_info.csv'
+
+    csv_fifo_depth_files = {}
+    with open(names_file_path) as names_file:
+        for _ in range(4):
+            next(names_file)
+        for line in names_file:
+            layer_name = line.split(',')[1]
+            csv_file_name = line.split(',')[3][:-1]
+            csv_fifo_depth_files[layer_name] = csv_file_name
+
+    optmized_fifo_depths = {}
+    for layer_name, file_name in csv_fifo_depth_files.items():
+        with open(path_to_zip_file + file_name) as chan_status_file:
+            lines = chan_status_file.readlines()
+            optmized_fifo_depths[layer_name[:-4]] = int(
+                lines[-1]
+            )  # remove "_i_U" from the layer name string and keep the last line of the file that contains the max depth
+
+    return optmized_fifo_depths
+
+
+def execute_cosim_to_profile_fifos(model):
+    model.write()
+    model.build(
+        reset=False,
+        csim=False,
+        synth=True,
+        cosim=False,
+        validation=False,
+        export=False,
+        vsynth=False,
+        fifo_opt=True,
+        bitfile=False,
+        log_to_stdout=False,
+    )
+
+
+class FifoDepthOptimization(ConfigurableOptimizerPass, ModelOptimizerPass):
+
+    def __init__(self):
+        self.profiling_fifo_depth = 100_000
+
+    def transform(self, model):
+        """Perform FIFO depth optimization between the FIFOs of all layers to reduce resource utilization as the
+        initial FIFOs set by hls4ml might be larger than required. At the end of the optimization the FIFOs will
+        have the largest depths achieved during co-simulation without causing any deadlocks between the layers
+        (producer-consumer), thus no additional delays between the layers. In some cases, this optimization
+        might lead to bigger FIFOs than initially set by the hls4ml tool in order to prevent deadlocks.
+
+        Args:
+            model (ModelGraph): The model to which FIFO depth optimization is applied.
+
+        Raises:
+            ValueError: If the FIFO depth for profiling provided by the user is not a non-negative integer.
+            RuntimeError: If the IO type is not set to "io_stream".
+
+        Returns:
+            bool: The execution state of the Optimizer Pass
+        """
+
+        if not isinstance(self.profiling_fifo_depth, int) or self.profiling_fifo_depth <= 0:
+            raise ValueError('The FIFO depth for profiling (profiling_fifo_depth variable) must be a non-negative integer.')
+
+        # check axi-stream or io-stream
+        if not (model.config.get_config_value('IOType') == 'io_stream'):
+            raise RuntimeError('To use this optimization you have to set `IOType` field to `io_stream` in the HLS config.')
+
+        hlsPrjPath = model.config.backend.writer.mg.get_vitis_hls_exec_dir(model)
+
+        initial_fifo_depths = initialize_large_fifos(model, self.profiling_fifo_depth)
+        execute_cosim_to_profile_fifos(model)
+        optimized_fifo_depths = get_vitis_optimized_fifo_depths(model, cus_hls_prj_path=hlsPrjPath + "/hls")
+        generate_depths_file(model, initial_fifo_depths, optimized_fifo_depths)
+        set_optimized_fifo_depths(model, optimized_fifo_depths)
+
+        print('FIFO optimization completed')
+
+        return False

hls4ml/backends/vitis_unified/vitis_unified_backend.py

@@ -0,0 +1,181 @@
+import os
+import subprocess
+import sys
+from shutil import copy2
+
+from hls4ml.backends import VitisBackend, VivadoBackend
+from hls4ml.model.flow import register_flow
+from hls4ml.writer.vitis_unified_writer.meta_gen import VitisUnified_MetaGen as mg
+
+
+class VitisUnifiedBackend(VitisBackend):
+    def __init__(self):
+        super(VivadoBackend, self).__init__(name='VitisUnified')
+        self._register_layer_attributes()
+        self._register_flows()
+
+    def run_term_command(self, model, taskName: str, command: str, logStdOut: bool, cwd):
+
+        print("-------------------------------------------------------")
+        print(f"start running task : {taskName}")
+        print(f"    with command: {command}")
+        print("-------------------------------------------------------")
+
+        output_dir = model.config.get_output_dir()
+
+        out_log_path = os.path.join(output_dir, f'{taskName}_out.log')
+        err_log_path = os.path.join(output_dir, f'{taskName}_err.log')
+        out_target = None if logStdOut else open(out_log_path, 'w')
+        err_target = None if logStdOut else open(err_log_path, 'w')
+
+        try:
+            runningProcess = subprocess.Popen(command, shell=True, cwd=cwd, stdout=out_target, stderr=err_target, text=True)
+            runningProcess.communicate()
+            if runningProcess.returncode != 0:
+                raise Exception(
+                    f'Package failed for {taskName} for project {model.config.get_project_name()}. See logs for details.'
+                )
+
+            stdout, stderr = runningProcess.communicate()
+            print(f"stdout: {stdout}")
+            print(f"stderr: {stderr}")
+
+            print(f"task {taskName} finished")
+
+        except Exception as e:
+            print(f"task {taskName} failed")
+            print(e)
+            raise e
+        finally:
+            if out_target:
+                out_target.close()
+            if err_target:
+                err_target.close()
+
+    def build(
+        self,
+        model,
+        reset=False,
+        csim=False,
+        synth=False,
+        cosim=False,
+        validation=False,
+        export=False,
+        vsynth=False,
+        fifo_opt=False,
+        bitfile=False,
+        log_to_stdout=True,
+    ):
+        # it builds and return vivado reports
+        if 'linux' in sys.platform:
+            found = os.system('command -v vitis > /dev/null')
+            if found != 0:
+                raise Exception('Vitis installation not found. Make sure "vitis" is on PATH.')
+
+        if csim:
+            raise Exception("Current Vitis Unified not support csim. Please set csim=False to run Vitis Unified.")
+        if validation:
+            raise Exception(
+                "Current Vitis Unified not support validation. Please set validation=False to run Vitis Unified."
+            )
+        if export:
+            raise Exception("Current Vitis Unified not support export. Please set export=False to run Vitis Unified.")
+
+        output_dir = model.config.get_output_dir()
+
+        hls_config_file = os.path.join(output_dir, "hls_kernel_config.cfg")
+        # build command
+        csynth_cmd = ("v++ -c --mode hls --config {configPath} --work_dir unifiedPrj").format(configPath=hls_config_file)
+        csynth_cwd = mg.get_vitis_hls_dir(model)
+
+        # util template (used in csim/cosim/package)
+        util_command = "vitis-run --mode hls --{op} --config {configPath} --work_dir unifiedPrj"
+
+        # command for each configuration
+
+        package_cmd = util_command.format(op="package", configPath=hls_config_file)
+        package_cwd = mg.get_vitis_hls_dir(model)
+        cosim_cmd = util_command.format(op="cosim", configPath=hls_config_file)
+        cosim_cwd = mg.get_vitis_hls_dir(model)
+        csim_cmd = util_command.format(op="csim", configPath=hls_config_file)
+        csim_cwd = mg.get_vitis_hls_dir(model)
+
+        kerlink_cmd = "./buildAcc.sh"
+        kerlink_cwd = mg.get_vitis_linker_dir(model)
+
+        if synth:
+            self.prepare_sim_config_file(model, True)
+            self.run_term_command(model, "csynth", csynth_cmd, log_to_stdout, csynth_cwd)
+            self.run_term_command(model, "package", package_cmd, log_to_stdout, package_cwd)
+
+        if csim:
+            self.prepare_sim_config_file(model, True)
+            self.run_term_command(model, "csim", csim_cmd, log_to_stdout, csim_cwd)
+
+        if cosim or fifo_opt:
+            self.prepare_sim_config_file(model, False)
+            self.run_term_command(model, "cosim", cosim_cmd, log_to_stdout, cosim_cwd)
+
+        # if bitfile
+        if bitfile:
+            self.run_term_command(model, "kerlink", kerlink_cmd, log_to_stdout, kerlink_cwd)
+
+    def prepare_sim_config_file(self, model, is_csim):
+        suffix = "csim" if is_csim else "cosim"
+        src = f"{model.config.get_output_dir()}/hls_kernel_config_{suffix}.cfg"
+        des = f"{model.config.get_output_dir()}/hls_kernel_config.cfg"
+        copy2(src, des)
+        return des
+
+    def create_initial_config(
+        self,
+        board='zcu102',
+        part=None,
+        clock_period=5,
+        clock_uncertainty='12.5%',
+        io_type='io_stream',
+        driver='python',
+        input_type='float',
+        output_type='float',
+        in_stream_buf_size=128,
+        out_stream_buf_size=128,
+        xpfmPath='/opt/Xilinx/Vitis/2023.2/base_platforms/' 'xilinx_zcu102_base_202320_1/xilinx_zcu102_base_202320_1.xpfm',
+        **_,
+    ):
+
+        config = super().create_initial_config(part, clock_period, clock_uncertainty, io_type)
+
+        config['UnifiedConfig'] = {}
+        config['UnifiedConfig']["in_stream_buf_Size"] = in_stream_buf_size
+        config['UnifiedConfig']["out_stream_buf_Size"] = out_stream_buf_size
+        config['UnifiedConfig']['XPFMPath'] = xpfmPath
+        config['UnifiedConfig']['Board'] = board
+        config['UnifiedConfig']['Driver'] = driver
+        config['UnifiedConfig']['InputDtype'] = input_type  # float, double or ap_fixed<a,b>
+        config['UnifiedConfig']['OutputDtype'] = output_type  # float, double or ap_fixed<a,b>
+
+        if io_type != "io_stream":
+            raise Exception("io_type must be io_stream")
+        if input_type not in ["double", "float"]:
+            raise Exception("input_type must be float or double")
+        if output_type not in ["double", "float"]:
+            raise Exception("output_type must be float or double")
+
+        return config
+
+    def get_default_flow(self):
+        return self._default_flow
+
+    def get_writer_flow(self):
+        return self._writer_flow
+
+    def _register_flows(self):
+        vitis_ip = 'vitis:ip'
+        writer_passes = ['make_stamp', 'vitisunified:write_hls']
+        self._writer_flow = register_flow('write', writer_passes, requires=['vitis:ip'], backend=self.name)
+        self._default_flow = vitis_ip
+
+        # register fifo depth optimization
+        fifo_depth_opt_passes = ['vitisunified:fifo_depth_optimization'] + writer_passes
+
+        register_flow('fifo_depth_optimization', fifo_depth_opt_passes, requires=['vitis:ip'], backend=self.name)

hls4ml/backends/vitis_unified/vitis_unified_config.py

@@ -0,0 +1,49 @@
+class VitisUnifiedConfig:
+
+    def __init__(self, config, model_inputs, model_outputs):
+        self.config = config.config
+        self.board = self.config.get('UnifiedConfig', {}).get('Board', 'pynq-z2')
+
+        # before first and afterlast layer we have the configuratble buffer
+        # [platform]<-->[in_steram_bufferSz]<-->[hls]<-->[out_stream_bufferSz]<-->[platform]
+        self.in_steram_bufferSz = self.config["UnifiedConfig"]["in_stream_buf_Size"]
+        self.out_stream_bufferSz = self.config["UnifiedConfig"]["out_stream_buf_Size"]
+
+        # the path to the generated platform
+        self.XPFMPath = self.config["UnifiedConfig"]["XPFMPath"]
+
+        self.driver = self.config['UnifiedConfig']['Driver']
+
+        # c++ type for input and output of the hls kernel it must be str (float/double)
+        self.input_type = self.config['UnifiedConfig']['InputDtype']
+        self.output_type = self.config['UnifiedConfig']['OutputDtype']
+
+        assert self.input_type == self.output_type, "Input and Output data types must be the same type different"
+        assert len(model_inputs) >= 1, "Only models with at least one input tensor are currently supported by VitisUnified"
+        assert len(model_outputs) >= 1, "Only models with one output tensor are currently supported by VitisUnified"
+        self.inps = model_inputs.copy()
+        self.outs = model_outputs.copy()
+
+    def get_corrected_types(self):
+        return self.input_type, self.output_type, self.inps, self.outs
+
+    def get_driver(self):
+        return self.driver
+
+    def get_board(self):
+        return self.board
+
+    def get_input_type(self):
+        return self.input_type
+
+    def get_output_type(self):
+        return self.output_type
+
+    def get_in_stream_bufferSz(self):
+        return self.in_steram_bufferSz
+
+    def get_out_stream_bufferSz(self):
+        return self.out_stream_bufferSz
+
+    def get_XPFMPath(self):
+        return self.XPFMPath