<filename>xls/contrib/xlscc/instrument_ir.py
# Copyright 2021 The XLS Authors
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""A script for testing the translated IR for individual C++ functions.

Uses a standard C++ compiler as a reference.

The target function is instrumented using a RAII mechanism to capture its
inputs and outputs. The instrumentation is generated according to information
in the metadata protobuf generated by xlscc.

When the modified C++ binary is run, the instrumentation code writes inputs to
one file, and expected outputs into another. These files, together with the IR
from xlscc, are fed into the eval_ir_main tool, which checks that the results
of IR simulation match the expected values file generated by the C++ binary.
"""

import shutil
import subprocess
import tempfile
import textwrap
import typing

from absl import app
from absl import flags

from xls.common import runfiles
from xls.contrib.xlscc import hls_block_pb2
from xls.contrib.xlscc import metadata_output_pb2

FLAGS = flags.FLAGS

flags.DEFINE_string("ir_to_test", None, "Path to IR file to test")
flags.mark_flag_as_required("ir_to_test")
flags.DEFINE_string("xlscc_metadata", None,
                    "Path to metadata protobuf from xlscc")
flags.mark_flag_as_required("xlscc_metadata")
flags.DEFINE_string(
    "xlscc_block_metadata",
    None,
    "Path to metadata protobuf for block",
    required=False)
flags.DEFINE_string("cmd_to_test", None, "Command to generate traces")
flags.mark_flag_as_required("cmd_to_test")
flags.DEFINE_string("build_cmd", None, "Command to build the C binary used")
flags.mark_flag_as_required("build_cmd")
flags.DEFINE_string(
    "function_to_instrument",
    None,
    "Simple name of function to instrument. Defaults to top function from MetadataOutput",
    required=False)
flags.DEFINE_string(
    "backend",
    "serial_jit",
    "Backend to use for evaluation. See eval_proc_main help.",
    required=False)
flags.DEFINE_string(
    "block_signature_proto",
    "",
    "If using --backend block_interpreter, then this specifies the path to the signature protobuf from codegen.",
    required=False)
flags.DEFINE_bool(
    "delete_temps", True, "Delete temporary files?", required=False)
flags.DEFINE_integer(
    "random_seed",
    42,
    "Random seed for testbench.",
    required=False)
flags.DEFINE_float(
    "prob_input_valid_assert",
    1.0,
    "Single-cycle probability of asserting valid with more input ready.",
    required=False)

EVAL_IR_PATH = runfiles.get_path("xls/tools/eval_ir_main")
EVAL_PROC_PATH = runfiles.get_path("xls/tools/eval_proc_main")


def find_offset_for_loc(src: str, loc: metadata_output_pb2.SourceLocation):
  """Find the linear index for a file + line location.

  Args:
    src: Source to search
    loc: Location to find

  Returns:
    Integer index of loc
  """
  offset = 0
  line = 1
  col = 1

  for offset in range(len(src)):
    if src[offset] == "\n":
      line += 1
      col = 1
      continue

    if (line == loc.line) and (col == loc.column):
      return offset

    col += 1

  raise app.Error("Failed to find offset for {loc}".format(loc=loc))


def get_func_to_instrument(
    meta_proto: metadata_output_pb2.MetadataOutput,
    function_to_instrument_name: str) -> metadata_output_pb2.FunctionPrototype:
  """Gets the metadata for the target function to be instrumented.

  Args:
    meta_proto: Protobuf from xlscc describing all the translated IR
    function_to_instrument_name: Simple name of the function (not fully
      qualified)

  Returns:
    Protobuf describing the target function for instrumentation
  """
  if function_to_instrument_name is None:
    top_func_meta = meta_proto.top_func_proto
    assert top_func_meta is not None
    function_to_instrument_name = top_func_meta.name.name

  filtered_funcs = [
      f for f in meta_proto.all_func_protos
      if f.name.name == function_to_instrument_name
  ]

  if not filtered_funcs:
    raise app.Error("Couldn't find metadata for function with name: {n}".format(
        n=function_to_instrument_name))

  if len(filtered_funcs) > 1:
    raise app.Error("Found more than one function with name: {n}".format(
        n=function_to_instrument_name))

  return filtered_funcs[0]


def template_arg_to_string(arg: metadata_output_pb2.TemplateArgument) -> str:
  """Converts a template argument in the metadata protobuf to a C++ type name.

  Args:
    arg: Protobuf describing the argument

  Returns:
    The formatted C++ argument
  """
  if arg.HasField("as_integral"):
    return "{n}".format(n=arg.as_integral.real)
  elif arg.HasField("as_type"):
    return type_to_string(arg.as_type)
  else:
    raise app.Error(
        "Don't know how to translate template: {arg}".format(arg=arg))


def type_to_string(type_pb: metadata_output_pb2.Type) -> str:
  """Converts a type as specified in the metadata protobuf to a C++ type name.

  Args:
    type_pb: Protobuf describing the type

  Returns:
    The formatted C++ type name
  """
  if type_pb.HasField("as_void"):
    return "void"
  if type_pb.HasField("as_bool"):
    return "bool"
  if type_pb.HasField("as_int"):
    ret = ""
    if not type_pb.as_int.is_signed:
      ret += "unsigned "
    if type_pb.as_int.width == 8:
      ret += "char"
    elif type_pb.as_int.width == 16:
      ret += "short"
    elif type_pb.as_int.width == 32:
      ret += "int"
    elif type_pb.as_int.width == 64:
      ret += "long"
    else:
      raise app.Error(
          "Don't know how to translate native int of width: {w}".format(
              w=type_pb.as_int.width))

    return ret
  elif type_pb.HasField("as_inst"):
    ret = type_pb.as_inst.name.fully_qualified_name
    if type_pb.as_inst.args:
      ret += "<"
      ret += ", ".join(map(template_arg_to_string, type_pb.as_inst.args))
      ret += ">"
    return ret
  elif type_pb.HasField("as_array"):
    ret = type_to_string(type_pb.as_array.element_type)
    ret += "[{n}]".format(n=type_pb.as_array.size)
    return ret
  else:
    raise app.Error("Don't know how to translate type: {t}".format(t=type_pb))


def find_instantiated_type(
    id_pb: int,
    metadata: metadata_output_pb2.MetadataOutput) -> metadata_output_pb2.Type:
  """Finds a type by id in the metadata protobuf.

  Args:
    id_pb: Id number of the type
    metadata: Protobuf from xlscc describing all the translated IR

  Returns:
    The protobuf describing the type identified by id_pb
  """
  for record in metadata.structs:
    if not record.HasField("as_struct"):
      raise app.Error("Non-struct type in structs field")

    struct = record.as_struct
    assert struct.name.HasField("as_inst")
    if struct.name.as_inst.name.id == id_pb:
      return record

  raise app.Error(
      "Couldn't find instantiated type with id: {id}".format(id=id_pb))


def format_print_value(name: str, stream_name: str,
                       type_pb: metadata_output_pb2.Type,
                       metadata: metadata_output_pb2.MetadataOutput) -> str:
  """Formats C++ source to print out a value.

  Args:
    name: Name of the value in the C++ source
    stream_name: Name of the stream to print to, such as cout
    type_pb: Protobuf describing the type of the value
    metadata: Protobuf from xlscc describing all the translated IR

  Returns:
    The formatted C++ source
  """
  if type_pb.HasField("as_bits"):
    return "    {stream_name}<<\"bits[{w}]:\"<<{v};".format(
        stream_name=stream_name, w=type_pb.as_bits.width, v=name)
  elif type_pb.HasField("as_int"):
    return "    {stream_name}<<\"bits[{w}]:\"<<{v};".format(
        stream_name=stream_name, w=type_pb.as_int.width, v=name)
  elif type_pb.HasField("as_bool"):
    return "    {stream_name}<<\"bits[1]:\"<<({v}?1:0);".format(
        stream_name=stream_name, v=name)
  elif type_pb.HasField("as_inst"):
    return format_print_value(
        name, stream_name,
        find_instantiated_type(type_pb.as_inst.name.id, metadata), metadata)
  elif type_pb.HasField("as_struct"):
    if type_pb.as_struct.no_tuple:
      assert 1 == len(type_pb.as_struct.fields)
      one_field = type_pb.as_struct.fields[0]
      return format_print_value(name, stream_name, one_field.type, metadata)
    else:

      def format_field(field: metadata_output_pb2.StructField):
        return format_print_value(name + "." + field.name, stream_name,
                                  field.type, metadata)

      ret = "{stream_name}<<\"(\";\n".format(stream_name=stream_name)
      ret += "{stream_name}<<\",\";\n".format(stream_name=stream_name).join(
          map(format_field, type_pb.as_struct.fields))
      ret += "{stream_name}<<\")\";\n".format(stream_name=stream_name)
      return ret
  elif type_pb.HasField("as_array"):
    ret = "{stream_name}<<\"[\";\n".format(stream_name=stream_name)
    elems = []
    for i in range(type_pb.as_array.size):
      elems += [
          format_print_value("{n}[{i}]".format(n=name, i=i), stream_name,
                             type_pb.as_array.element_type, metadata)
      ]
    ret += "{stream_name}<<\",\";\n".format(stream_name=stream_name).join(elems)
    ret += "{stream_name}<<\"]\";\n".format(stream_name=stream_name)
    return ret
  else:
    raise app.Error("Unimplemented print for type {t}".format(t=type_pb))


def gen_instrumentation_proc(
    channels_tmps, reset_ticks_tmp,
    block_channels_by_name: typing.Dict[str, hls_block_pb2.HLSChannel],
    metadata: metadata_output_pb2.MetadataOutput) -> typing.Dict[str, str]:
  """Generates source blocks to insert.

  Args:
    channels_tmps: List of tuples, one for each channel to instrument
    reset_ticks_tmp: The number of ticks between resets are written to this path
    block_channels_by_name: HLSChannel protos by name
    metadata: Protobuf from xlscc describing all the translated IR

  Returns:
    Dictionary of source segments to insert
  """

  before_func_src = """

#include <fstream>
#include <cassert>

class reset_recorder {
 private:
  int ticks;
  std::ofstream stream;
 public:
  reset_recorder(const absl::string_view& out_path)
    : ticks(0), stream(out_path.data(), std::ios_base::app) {
  }
  ~reset_recorder() {
    stream << ticks << std::endl;

  }
  void tick() {
    ++ticks;
  }
};

struct tick_on_exit {
  reset_recorder& rec;
  tick_on_exit(reset_recorder& rec) : rec(rec) {
  }
  ~tick_on_exit() {
    rec.tick();
  }
};

"""

  after_func_src = ""
  instrument_src = """
  HLS_STATIC reset_recorder reset_recorder_i("{path}");
  // Don't count Run() as a tick if it never exits
  tick_on_exit tick_on_exit_i(reset_recorder_i);
""".format(path=reset_ticks_tmp)

  for channel, tmp in channels_tmps:
    assert channel.name in block_channels_by_name
    hls_ch = block_channels_by_name[channel.name]

    instrument_src += """
        static std::ofstream {n}_stream("{path}");
""".format(
    n=channel.name, path=tmp.name)

    if hls_ch.type == hls_block_pb2.ChannelType.DIRECT_IN:
      assert hls_ch.is_input
      assert not (channel.type.as_inst and
                  channel.type.as_inst.name.fully_qualified_name
                  == "__xls_channel")

      ch_type = channel.type
      instrument_src += textwrap.dedent("""
                          struct direct_in_on_exit_{n} {{
                            {t} &value;
                            std::ofstream& stream;

                            direct_in_on_exit_{n} ({t} &value, std::ofstream& stream)
                             : value(value), stream(stream) {{
                            }}
                            ~direct_in_on_exit_{n} () {{
                              {print_val}
                              {n}_stream << std::endl;
                            }}
                          }} direct_in_on_exit_{n}_i({n}, {n}_stream);
""".format(
    n=channel.name,
    t=type_to_string(ch_type),
    print_val=format_print_value("value", "{n}_stream".format(n=channel.name),
                                 ch_type, metadata)))
    elif hls_ch.type == hls_block_pb2.ChannelType.FIFO:
      assert channel.type.as_inst and channel.type.as_inst.name.fully_qualified_name == "__xls_channel"

      after_func_src += """
#undef {n}
""".format(n=channel.name)

      ch_type = channel.type.as_inst.args[0].as_type

      instrument_src += textwrap.dedent("""
                          class capture_channel_{n} {{
                           private:
                            typedef {t} T;
                            ac_channel<T>& underlying;
                            std::ostream& stream;
                            bool did_read, did_write;
                            void* inst;
                           public:
                            capture_channel_{n}(void* inst, ac_channel<T>& ch, std::ostream& stream)
                              : underlying(ch),
                                stream(stream),
                                did_read(false),
                                did_write(false),
                                inst(inst) {{
                            }}

                            void write(const T& val) {{
                              {print_val}
                              (void)inst;
                              stream << std::endl;
                              did_write = true;
                              assert(!did_read);
                              underlying.write(val);
                            }}

                            T read() {{
                              const T val(underlying.read());
                              {print_val}
                              stream << std::endl;
                              did_read = true;
                              assert(!did_write);
                              return val;
                            }}
                          }};

                          capture_channel_{n} {n}_captured(this, {n}, {n}_stream);
                          #undef {n}
                          #define {n} {n}_captured
  """).format(
      n=channel.name,
      t=type_to_string(ch_type),
      path=tmp.name,
      print_val=format_print_value("val", "stream", ch_type, metadata))
    else:
      raise app.UsageError(
          "Don't know how to instrument channel {n} of type: {t}".format(
              n=hls_ch.name, t=hls_ch.type))

  return {
      "before_func_src": before_func_src,
      "instrument_src": instrument_src,
      "after_func_src": after_func_src
  }


def gen_instrumentation_pure_func(
    function_to_instrument_proto: metadata_output_pb2.FunctionPrototype,
    inputs_path: str, outputs_path: str,
    metadata: metadata_output_pb2.MetadataOutput) -> typing.Dict[str, str]:
  """Generates source blocks to insert.

  Args:
    function_to_instrument_proto: Protobuf describing the target function
    inputs_path: File path where the test binary should write the input values
    outputs_path: File path where the test binary should write the output values
    metadata: Protobuf from xlscc describing all the translated IR

  Returns:
    Dictionary of source segments to insert
  """

  saved_refs_src = ""
  params_srcs = []
  init_srcs = []
  call_constructor_src = []
  enter_print_srcs = []
  exit_print_srcs = []

  if not function_to_instrument_proto.return_type.HasField("as_void"):
    exit_print_srcs += [
        format_print_value("instrument__retval", "outputs_str",
                           function_to_instrument_proto.return_type, metadata)
    ]

  for param in function_to_instrument_proto.params:

    param_decl = ""
    if param.is_const:
      param_decl += "const "

    if not param.type.HasField("as_array"):
      param_decl += type_to_string(param.type)
      param_decl += "& " + param.name
    else:

      postfix = ""
      prefix = ""
      if param.type.as_array.element_type.HasField("as_array"):
        element_type = param.type
        while element_type.HasField("as_array"):
          if postfix:
            prefix = prefix + "*"
          postfix = "[{n}]".format(n=element_type.as_array.size)
          element_type = element_type.as_array.element_type
      else:
        element_type = param.type.as_array.element_type
        prefix = "*"

      param_decl += "{type} ({pre}{name}){post}".format(
          type=type_to_string(element_type),
          name=param.name,
          pre=prefix,
          post=postfix)

    params_srcs += [param_decl]

    call_constructor_src += [param.name]

    enter_print_srcs += [
        format_print_value(param.name, "inputs_str", param.type, metadata)
    ]

    if param.is_reference and not param.is_const:
      saved_refs_src += "  "
      saved_refs_src += param_decl
      saved_refs_src += ";\n"
      init_srcs += [param.name + "(" + param.name + ")"]
      exit_print_srcs += [
          format_print_value(param.name, "outputs_str", param.type, metadata)
      ]

  exit_print_src = ""
  assert exit_print_srcs
  if len(exit_print_srcs) > 1:
    exit_print_src = "outputs_str<<\"(\";\n" + "outputs_str<<\", \";\n".join(
        exit_print_srcs) + "\noutputs_str<<\")\";\n"
  else:
    exit_print_src = exit_print_srcs[0]

  enter_print_src = ""
  assert enter_print_srcs
  if len(enter_print_srcs) > 1:
    enter_print_src = "inputs_str<<\"; \";\n".join(enter_print_srcs)
  else:
    enter_print_src = enter_print_srcs[0]

  return_stuff = ""

  if not function_to_instrument_proto.return_type.HasField("as_void"):
    return_stuff = textwrap.dedent("""
                      {return_type} instrument__retval;
                      template<typename T>
                      T operator=(const T& op) {{
                        instrument__retval = static_cast<{return_type}>(op);
                        return op;
                      }}
""").format(
    return_type=type_to_string(function_to_instrument_proto.return_type))

  before_func_src = textwrap.dedent("""
                      #include <fstream>
                      static std::ofstream inputs_str("{in_path}");
                      static std::ofstream outputs_str("{out_path}");

                      class capture_return {{
                       private:

                      {saved_refs}
                       public:
                        capture_return({params})
                         {inits} {{
                      {enter_print_src}
                          inputs_str << std::endl;
                        }}
                        ~capture_return() {{
                      {exit_print_src}
                          outputs_str << std::endl;
                        }}

                        {return_stuff}
                      }};
""").format(
    params=", ".join(params_srcs),
    inits=(": " + ", ".join(init_srcs)) if init_srcs else "",
    saved_refs=saved_refs_src,
    enter_print_src=enter_print_src,
    exit_print_src=exit_print_src,
    return_stuff=return_stuff,
    in_path=inputs_path,
    out_path=outputs_path)

  instrument_src = textwrap.dedent("""
                    capture_return capture_return_i ({call_constructor_src});

                    #define return return capture_return_i =
  """).format(call_constructor_src=", ".join(call_constructor_src))

  return {
      "instrument_src": instrument_src,
      "before_func_src": before_func_src,
      "after_func_src": "\n" + "#undef return" + "\n"
  }


def instrument_source(
    function_to_instrument_proto: metadata_output_pb2.FunctionPrototype,
    inst_srcs: typing.Dict[str, str], original_src_tmp: str) -> str:
  """Modifies source file to instrument the target function.

  Args:
    function_to_instrument_proto: Protobuf describing the target function
    inst_srcs: Dictionary of source segments to insert
    original_src_tmp: File path at which to save the unmodified C++ source file

  Returns:
    The path to the C++ file that was modified
  """

  filename_to_modify = function_to_instrument_proto.return_location.begin.filename

  shutil.copyfile(filename_to_modify, original_src_tmp)

  src = None
  with open(filename_to_modify, "r") as f:
    orig_src = f.read()

  params_end_loc = find_offset_for_loc(
      orig_src, function_to_instrument_proto.parameters_location.end)

  decl_begin_loc = find_offset_for_loc(
      orig_src, function_to_instrument_proto.whole_declaration_location.begin)

  decl_end_loc = find_offset_for_loc(
      orig_src, function_to_instrument_proto.whole_declaration_location.end)

  open_brace_loc = orig_src.find("{", params_end_loc)
  assert open_brace_loc >= 0

  src = orig_src

  src = src[:decl_begin_loc] + inst_srcs["before_func_src"] + src[
      decl_begin_loc + 0:]

  open_brace_loc += len(inst_srcs["before_func_src"])
  decl_end_loc += len(inst_srcs["before_func_src"])

  src = src[:open_brace_loc +
            1] + inst_srcs["instrument_src"] + src[open_brace_loc + 2:]
  decl_end_loc += len(inst_srcs["instrument_src"]) - 1

  src = src[:decl_end_loc] + inst_srcs["after_func_src"] + src[decl_end_loc +
                                                               0:]

  top_hdr = textwrap.dedent("""
#define XlsInt ac_int
""")

  src = top_hdr + src

  with open(filename_to_modify, "w") as f:
    f.write(src)

  return filename_to_modify


def create_temp(suffix):
  ret = tempfile.NamedTemporaryFile(suffix=suffix, delete=FLAGS.delete_temps)
  return ret


def main(argv):
  if len(argv) > 1:
    raise app.UsageError("Too many command-line arguments.")

  # Parse metadata
  meta_proto = metadata_output_pb2.MetadataOutput()
  with open(FLAGS.xlscc_metadata, "rb") as f:
    meta_proto.ParseFromString(f.read())

  # Find function to instrument
  function_to_instrument_proto = get_func_to_instrument(
      meta_proto, FLAGS.function_to_instrument)

  # Instrument C source
  original_src_tmp = create_temp(suffix=".cc")

  params_by_name = {}
  no_channels = True
  for param in function_to_instrument_proto.params:
    is_channel = param.type.HasField("as_inst") and (
        param.type.as_inst.name.fully_qualified_name == "__xls_channel")
    no_channels = no_channels and not is_channel
    params_by_name[param.name] = param

  channels_tmps = []
  block_channels_by_name = {}

  inst_srcs = {}
  if no_channels:
    inputs_tmp = create_temp(suffix=".ir")
    outputs_tmp = create_temp(suffix=".ir")
    inst_srcs = gen_instrumentation_pure_func(function_to_instrument_proto,
                                              inputs_tmp.name, outputs_tmp.name,
                                              meta_proto)
  else:
    reset_ticks_tmp = create_temp(suffix=".log")

    # Parse block data
    block_proto = hls_block_pb2.HLSBlock()
    with open(FLAGS.xlscc_block_metadata, "rb") as f:
      block_proto.ParseFromString(f.read())

    for hls_ch in block_proto.channels:
      block_channels_by_name[hls_ch.name] = hls_ch
      assert hls_ch.name in params_by_name
      channels_tmps += [(params_by_name[hls_ch.name], create_temp(suffix=".ir"))
                       ]

    inst_srcs = gen_instrumentation_proc(channels_tmps, reset_ticks_tmp.name,
                                         block_channels_by_name, meta_proto)

  print("Saving original source before instrumentation at ",
        original_src_tmp.name)
  filename_to_modify = instrument_source(function_to_instrument_proto,
                                         inst_srcs, original_src_tmp.name)

  # Build the C model
  print("Building C++ binary...")
  subprocess.check_call(FLAGS.build_cmd, shell=True)

  # Revert C source
  print("Reverting C++ source...")
  shutil.copyfile(original_src_tmp.name, filename_to_modify)

  # Run C model to generate traces
  print("Running C++ binary...")
  subprocess.check_call(FLAGS.cmd_to_test, shell=True)

  # Run IR simulator
  # check_output to suppress the spammy output of eval_ir_main
  print("Running IR simulator...")
  if no_channels:
    if FLAGS.backend == "block_interpreter":
      raise app.UsageError("Cannot use block interpreter for non-procs.")

    args = [
        EVAL_IR_PATH, "--top", function_to_instrument_proto.name.xls_name,
        "--input_file", inputs_tmp.name, "--expected_file", outputs_tmp.name,
        "--use_llvm_jit" if (FLAGS.backend == "serial_jit")
        else "--nouse_llvm_jit",
        FLAGS.ir_to_test
    ]
    print("Eval command: ", args)
    subprocess.check_output(args)
  else:

    with open(reset_ticks_tmp.name) as x:
      ticks_per_frame = x.read().split()

    ticks_per_frame = filter(lambda x: int(x) > 0, ticks_per_frame)

    input_ch_specs = []
    expected_ch_specs = []

    for ch, tmp in channels_tmps:
      assert ch.name in block_channels_by_name

      hls_ch = block_channels_by_name[ch.name]
      if hls_ch.is_input:
        input_ch_specs += [ch.name + "=" + tmp.name]
      else:
        expected_ch_specs += [ch.name + "=" + tmp.name]

    args = [
        EVAL_PROC_PATH, FLAGS.ir_to_test, "--ticks",
        ",".join(map(str, ticks_per_frame)), "--backend", FLAGS.backend,
        "--random_seed", str(FLAGS.random_seed),
        "--prob_input_valid_assert", str(FLAGS.prob_input_valid_assert),
        "--inputs_for_channels", ",".join(input_ch_specs),
        "--expected_outputs_for_channels", ",".join(expected_ch_specs)
    ]
    if FLAGS.block_signature_proto:
      args += ["--block_signature_proto", FLAGS.block_signature_proto]
    args += ["-v=1", "--logtostderr"]
    print("Eval command: ", " ".join(args))
    subprocess.check_output(args)

  # Done! If we got here, then there were no errors or mismatches
  print("... Success!")


if __name__ == "__main__":
  app.run(main)

<filename>xls/contrib/xlscc/instrument_ir.py
# Copyright 2021 The XLS Authors
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""A script for testing the translated IR for individual C++ functions.

Uses a standard C++ compiler as a reference.

The target function is instrumented using a RAII mechanism to capture its
inputs and outputs. The instrumentation is generated according to information
in the metadata protobuf generated by xlscc.

When the modified C++ binary is run, the instrumentation code writes inputs to
one file, and expected outputs into another. These files, together with the IR
from xlscc, are fed into the eval_ir_main tool, which checks that the results
of IR simulation match the expected values file generated by the C++ binary.
"""

import shutil
import subprocess
import tempfile
import textwrap
import typing

from absl import app
from absl import flags

from xls.common import runfiles
from xls.contrib.xlscc import hls_block_pb2
from xls.contrib.xlscc import metadata_output_pb2

FLAGS = flags.FLAGS

flags.DEFINE_string("ir_to_test", None, "Path to IR file to test")
flags.mark_flag_as_required("ir_to_test")
flags.DEFINE_string("xlscc_metadata", None,
                    "Path to metadata protobuf from xlscc")
flags.mark_flag_as_required("xlscc_metadata")
flags.DEFINE_string(
    "xlscc_block_metadata",
    None,
    "Path to metadata protobuf for block",
    required=False)
flags.DEFINE_string("cmd_to_test", None, "Command to generate traces")
flags.mark_flag_as_required("cmd_to_test")
flags.DEFINE_string("build_cmd", None, "Command to build the C binary used")
flags.mark_flag_as_required("build_cmd")
flags.DEFINE_string(
    "function_to_instrument",
    None,
    "Simple name of function to instrument. Defaults to top function from MetadataOutput",
    required=False)
flags.DEFINE_string(
    "backend",
    "serial_jit",
    "Backend to use for evaluation. See eval_proc_main help.",
    required=False)
flags.DEFINE_string(
    "block_signature_proto",
    "",
    "If using --backend block_interpreter, then this specifies the path to the signature protobuf from codegen.",
    required=False)
flags.DEFINE_bool(
    "delete_temps", True, "Delete temporary files?", required=False)
flags.DEFINE_integer(
    "random_seed",
    42,
    "Random seed for testbench.",
    required=False)
flags.DEFINE_float(
    "prob_input_valid_assert",
    1.0,
    "Single-cycle probability of asserting valid with more input ready.",
    required=False)

EVAL_IR_PATH = runfiles.get_path("xls/tools/eval_ir_main")
EVAL_PROC_PATH = runfiles.get_path("xls/tools/eval_proc_main")


def find_offset_for_loc(src: str, loc: metadata_output_pb2.SourceLocation):
  """Find the linear index for a file + line location.

  Args:
    src: Source to search
    loc: Location to find

  Returns:
    Integer index of loc
  """
  offset = 0
  line = 1
  col = 1

  for offset in range(len(src)):
    if src[offset] == "\n":
      line += 1
      col = 1
      continue

    if (line == loc.line) and (col == loc.column):
      return offset

    col += 1

  raise app.Error("Failed to find offset for {loc}".format(loc=loc))


def get_func_to_instrument(
    meta_proto: metadata_output_pb2.MetadataOutput,
    function_to_instrument_name: str) -> metadata_output_pb2.FunctionPrototype:
  """Gets the metadata for the target function to be instrumented.

  Args:
    meta_proto: Protobuf from xlscc describing all the translated IR
    function_to_instrument_name: Simple name of the function (not fully
      qualified)

  Returns:
    Protobuf describing the target function for instrumentation
  """
  if function_to_instrument_name is None:
    top_func_meta = meta_proto.top_func_proto
    assert top_func_meta is not None
    function_to_instrument_name = top_func_meta.name.name

  filtered_funcs = [
      f for f in meta_proto.all_func_protos
      if f.name.name == function_to_instrument_name
  ]

  if not filtered_funcs:
    raise app.Error("Couldn't find metadata for function with name: {n}".format(
        n=function_to_instrument_name))

  if len(filtered_funcs) > 1:
    raise app.Error("Found more than one function with name: {n}".format(
        n=function_to_instrument_name))

  return filtered_funcs[0]


def template_arg_to_string(arg: metadata_output_pb2.TemplateArgument) -> str:
  """Converts a template argument in the metadata protobuf to a C++ type name.

  Args:
    arg: Protobuf describing the argument

  Returns:
    The formatted C++ argument
  """
  if arg.HasField("as_integral"):
    return "{n}".format(n=arg.as_integral.real)
  elif arg.HasField("as_type"):
    return type_to_string(arg.as_type)
  else:
    raise app.Error(
        "Don't know how to translate template: {arg}".format(arg=arg))


def type_to_string(type_pb: metadata_output_pb2.Type) -> str:
  """Converts a type as specified in the metadata protobuf to a C++ type name.

  Args:
    type_pb: Protobuf describing the type

  Returns:
    The formatted C++ type name
  """
  if type_pb.HasField("as_void"):
    return "void"
  if type_pb.HasField("as_bool"):
    return "bool"
  if type_pb.HasField("as_int"):
    ret = ""
    if not type_pb.as_int.is_signed:
      ret += "unsigned "
    if type_pb.as_int.width == 8:
      ret += "char"
    elif type_pb.as_int.width == 16:
      ret += "short"
    elif type_pb.as_int.width == 32:
      ret += "int"
    elif type_pb.as_int.width == 64:
      ret += "long"
    else:
      raise app.Error(
          "Don't know how to translate native int of width: {w}".format(
              w=type_pb.as_int.width))

    return ret
  elif type_pb.HasField("as_inst"):
    ret = type_pb.as_inst.name.fully_qualified_name
    if type_pb.as_inst.args:
      ret += "<"
      ret += ", ".join(map(template_arg_to_string, type_pb.as_inst.args))
      ret += ">"
    return ret
  elif type_pb.HasField("as_array"):
    ret = type_to_string(type_pb.as_array.element_type)
    ret += "[{n}]".format(n=type_pb.as_array.size)
    return ret
  else:
    raise app.Error("Don't know how to translate type: {t}".format(t=type_pb))


def find_instantiated_type(
    id_pb: int,
    metadata: metadata_output_pb2.MetadataOutput) -> metadata_output_pb2.Type:
  """Finds a type by id in the metadata protobuf.

  Args:
    id_pb: Id number of the type
    metadata: Protobuf from xlscc describing all the translated IR

  Returns:
    The protobuf describing the type identified by id_pb
  """
  for record in metadata.structs:
    if not record.HasField("as_struct"):
      raise app.Error("Non-struct type in structs field")

    struct = record.as_struct
    assert struct.name.HasField("as_inst")
    if struct.name.as_inst.name.id == id_pb:
      return record

  raise app.Error(
      "Couldn't find instantiated type with id: {id}".format(id=id_pb))


def format_print_value(name: str, stream_name: str,
                       type_pb: metadata_output_pb2.Type,
                       metadata: metadata_output_pb2.MetadataOutput) -> str:
  """Formats C++ source to print out a value.

  Args:
    name: Name of the value in the C++ source
    stream_name: Name of the stream to print to, such as cout
    type_pb: Protobuf describing the type of the value
    metadata: Protobuf from xlscc describing all the translated IR

  Returns:
    The formatted C++ source
  """
  if type_pb.HasField("as_bits"):
    return "    {stream_name}<<\"bits[{w}]:\"<<{v};".format(
        stream_name=stream_name, w=type_pb.as_bits.width, v=name)
  elif type_pb.HasField("as_int"):
    return "    {stream_name}<<\"bits[{w}]:\"<<{v};".format(
        stream_name=stream_name, w=type_pb.as_int.width, v=name)
  elif type_pb.HasField("as_bool"):
    return "    {stream_name}<<\"bits[1]:\"<<({v}?1:0);".format(
        stream_name=stream_name, v=name)
  elif type_pb.HasField("as_inst"):
    return format_print_value(
        name, stream_name,
        find_instantiated_type(type_pb.as_inst.name.id, metadata), metadata)
  elif type_pb.HasField("as_struct"):
    if type_pb.as_struct.no_tuple:
      assert 1 == len(type_pb.as_struct.fields)
      one_field = type_pb.as_struct.fields[0]
      return format_print_value(name, stream_name, one_field.type, metadata)
    else:

      def format_field(field: metadata_output_pb2.StructField):
        return format_print_value(name + "." + field.name, stream_name,
                                  field.type, metadata)

      ret = "{stream_name}<<\"(\";\n".format(stream_name=stream_name)
      ret += "{stream_name}<<\",\";\n".format(stream_name=stream_name).join(
          map(format_field, type_pb.as_struct.fields))
      ret += "{stream_name}<<\")\";\n".format(stream_name=stream_name)
      return ret
  elif type_pb.HasField("as_array"):
    ret = "{stream_name}<<\"[\";\n".format(stream_name=stream_name)
    elems = []
    for i in range(type_pb.as_array.size):
      elems += [
          format_print_value("{n}[{i}]".format(n=name, i=i), stream_name,
                             type_pb.as_array.element_type, metadata)
      ]
    ret += "{stream_name}<<\",\";\n".format(stream_name=stream_name).join(elems)
    ret += "{stream_name}<<\"]\";\n".format(stream_name=stream_name)
    return ret
  else:
    raise app.Error("Unimplemented print for type {t}".format(t=type_pb))


def gen_instrumentation_proc(
    channels_tmps, reset_ticks_tmp,
    block_channels_by_name: typing.Dict[str, hls_block_pb2.HLSChannel],
    metadata: metadata_output_pb2.MetadataOutput) -> typing.Dict[str, str]:
  """Generates source blocks to insert.

  Args:
    channels_tmps: List of tuples, one for each channel to instrument
    reset_ticks_tmp: The number of ticks between resets are written to this path
    block_channels_by_name: HLSChannel protos by name
    metadata: Protobuf from xlscc describing all the translated IR

  Returns:
    Dictionary of source segments to insert
  """

  before_func_src = """

#include <fstream>
#include <cassert>

class reset_recorder {
 private:
  int ticks;
  std::ofstream stream;
 public:
  reset_recorder(const absl::string_view& out_path)
    : ticks(0), stream(out_path.data(), std::ios_base::app) {
  }
  ~reset_recorder() {
    stream << ticks << std::endl;

  }
  void tick() {
    ++ticks;
  }
};

struct tick_on_exit {
  reset_recorder& rec;
  tick_on_exit(reset_recorder& rec) : rec(rec) {
  }
  ~tick_on_exit() {
    rec.tick();
  }
};

"""

  after_func_src = ""
  instrument_src = """
  HLS_STATIC reset_recorder reset_recorder_i("{path}");
  // Don't count Run() as a tick if it never exits
  tick_on_exit tick_on_exit_i(reset_recorder_i);
""".format(path=reset_ticks_tmp)

  for channel, tmp in channels_tmps:
    assert channel.name in block_channels_by_name
    hls_ch = block_channels_by_name[channel.name]

    instrument_src += """
        static std::ofstream {n}_stream("{path}");
""".format(
    n=channel.name, path=tmp.name)

    if hls_ch.type == hls_block_pb2.ChannelType.DIRECT_IN:
      assert hls_ch.is_input
      assert not (channel.type.as_inst and
                  channel.type.as_inst.name.fully_qualified_name
                  == "__xls_channel")

      ch_type = channel.type
      instrument_src += textwrap.dedent("""
                          struct direct_in_on_exit_{n} {{
                            {t} &value;
                            std::ofstream& stream;

                            direct_in_on_exit_{n} ({t} &value, std::ofstream& stream)
                             : value(value), stream(stream) {{
                            }}
                            ~direct_in_on_exit_{n} () {{
                              {print_val}
                              {n}_stream << std::endl;
                            }}
                          }} direct_in_on_exit_{n}_i({n}, {n}_stream);
""".format(
    n=channel.name,
    t=type_to_string(ch_type),
    print_val=format_print_value("value", "{n}_stream".format(n=channel.name),
                                 ch_type, metadata)))
    elif hls_ch.type == hls_block_pb2.ChannelType.FIFO:
      assert channel.type.as_inst and channel.type.as_inst.name.fully_qualified_name == "__xls_channel"

      after_func_src += """
#undef {n}
""".format(n=channel.name)

      ch_type = channel.type.as_inst.args[0].as_type

      instrument_src += textwrap.dedent("""
                          class capture_channel_{n} {{
                           private:
                            typedef {t} T;
                            ac_channel<T>& underlying;
                            std::ostream& stream;
                            bool did_read, did_write;
                            void* inst;
                           public:
                            capture_channel_{n}(void* inst, ac_channel<T>& ch, std::ostream& stream)
                              : underlying(ch),
                                stream(stream),
                                did_read(false),
                                did_write(false),
                                inst(inst) {{
                            }}

                            void write(const T& val) {{
                              {print_val}
                              (void)inst;
                              stream << std::endl;
                              did_write = true;
                              assert(!did_read);
                              underlying.write(val);
                            }}

                            T read() {{
                              const T val(underlying.read());
                              {print_val}
                              stream << std::endl;
                              did_read = true;
                              assert(!did_write);
                              return val;
                            }}
                          }};

                          capture_channel_{n} {n}_captured(this, {n}, {n}_stream);
                          #undef {n}
                          #define {n} {n}_captured
  """).format(
      n=channel.name,
      t=type_to_string(ch_type),
      path=tmp.name,
      print_val=format_print_value("val", "stream", ch_type, metadata))
    else:
      raise app.UsageError(
          "Don't know how to instrument channel {n} of type: {t}".format(
              n=hls_ch.name, t=hls_ch.type))

  return {
      "before_func_src": before_func_src,
      "instrument_src": instrument_src,
      "after_func_src": after_func_src
  }


def gen_instrumentation_pure_func(
    function_to_instrument_proto: metadata_output_pb2.FunctionPrototype,
    inputs_path: str, outputs_path: str,
    metadata: metadata_output_pb2.MetadataOutput) -> typing.Dict[str, str]:
  """Generates source blocks to insert.

  Args:
    function_to_instrument_proto: Protobuf describing the target function
    inputs_path: File path where the test binary should write the input values
    outputs_path: File path where the test binary should write the output values
    metadata: Protobuf from xlscc describing all the translated IR

  Returns:
    Dictionary of source segments to insert
  """

  saved_refs_src = ""
  params_srcs = []
  init_srcs = []
  call_constructor_src = []
  enter_print_srcs = []
  exit_print_srcs = []

  if not function_to_instrument_proto.return_type.HasField("as_void"):
    exit_print_srcs += [
        format_print_value("instrument__retval", "outputs_str",
                           function_to_instrument_proto.return_type, metadata)
    ]

  for param in function_to_instrument_proto.params:

    param_decl = ""
    if param.is_const:
      param_decl += "const "

    if not param.type.HasField("as_array"):
      param_decl += type_to_string(param.type)
      param_decl += "& " + param.name
    else:

      postfix = ""
      prefix = ""
      if param.type.as_array.element_type.HasField("as_array"):
        element_type = param.type
        while element_type.HasField("as_array"):
          if postfix:
            prefix = prefix + "*"
          postfix = "[{n}]".format(n=element_type.as_array.size)
          element_type = element_type.as_array.element_type
      else:
        element_type = param.type.as_array.element_type
        prefix = "*"

      param_decl += "{type} ({pre}{name}){post}".format(
          type=type_to_string(element_type),
          name=param.name,
          pre=prefix,
          post=postfix)

    params_srcs += [param_decl]

    call_constructor_src += [param.name]

    enter_print_srcs += [
        format_print_value(param.name, "inputs_str", param.type, metadata)
    ]

    if param.is_reference and not param.is_const:
      saved_refs_src += "  "
      saved_refs_src += param_decl
      saved_refs_src += ";\n"
      init_srcs += [param.name + "(" + param.name + ")"]
      exit_print_srcs += [
          format_print_value(param.name, "outputs_str", param.type, metadata)
      ]

  exit_print_src = ""
  assert exit_print_srcs
  if len(exit_print_srcs) > 1:
    exit_print_src = "outputs_str<<\"(\";\n" + "outputs_str<<\", \";\n".join(
        exit_print_srcs) + "\noutputs_str<<\")\";\n"
  else:
    exit_print_src = exit_print_srcs[0]

  enter_print_src = ""
  assert enter_print_srcs
  if len(enter_print_srcs) > 1:
    enter_print_src = "inputs_str<<\"; \";\n".join(enter_print_srcs)
  else:
    enter_print_src = enter_print_srcs[0]

  return_stuff = ""

  if not function_to_instrument_proto.return_type.HasField("as_void"):
    return_stuff = textwrap.dedent("""
                      {return_type} instrument__retval;
                      template<typename T>
                      T operator=(const T& op) {{
                        instrument__retval = static_cast<{return_type}>(op);
                        return op;
                      }}
""").format(
    return_type=type_to_string(function_to_instrument_proto.return_type))

  before_func_src = textwrap.dedent("""
                      #include <fstream>
                      static std::ofstream inputs_str("{in_path}");
                      static std::ofstream outputs_str("{out_path}");

                      class capture_return {{
                       private:

                      {saved_refs}
                       public:
                        capture_return({params})
                         {inits} {{
                      {enter_print_src}
                          inputs_str << std::endl;
                        }}
                        ~capture_return() {{
                      {exit_print_src}
                          outputs_str << std::endl;
                        }}

                        {return_stuff}
                      }};
""").format(
    params=", ".join(params_srcs),
    inits=(": " + ", ".join(init_srcs)) if init_srcs else "",
    saved_refs=saved_refs_src,
    enter_print_src=enter_print_src,
    exit_print_src=exit_print_src,
    return_stuff=return_stuff,
    in_path=inputs_path,
    out_path=outputs_path)

  instrument_src = textwrap.dedent("""
                    capture_return capture_return_i ({call_constructor_src});

                    #define return return capture_return_i =
  """).format(call_constructor_src=", ".join(call_constructor_src))

  return {
      "instrument_src": instrument_src,
      "before_func_src": before_func_src,
      "after_func_src": "\n" + "#undef return" + "\n"
  }


def instrument_source(
    function_to_instrument_proto: metadata_output_pb2.FunctionPrototype,
    inst_srcs: typing.Dict[str, str], original_src_tmp: str) -> str:
  """Modifies source file to instrument the target function.

  Args:
    function_to_instrument_proto: Protobuf describing the target function
    inst_srcs: Dictionary of source segments to insert
    original_src_tmp: File path at which to save the unmodified C++ source file

  Returns:
    The path to the C++ file that was modified
  """

  filename_to_modify = function_to_instrument_proto.return_location.begin.filename

  shutil.copyfile(filename_to_modify, original_src_tmp)

  src = None
  with open(filename_to_modify, "r") as f:
    orig_src = f.read()

  params_end_loc = find_offset_for_loc(
      orig_src, function_to_instrument_proto.parameters_location.end)

  decl_begin_loc = find_offset_for_loc(
      orig_src, function_to_instrument_proto.whole_declaration_location.begin)

  decl_end_loc = find_offset_for_loc(
      orig_src, function_to_instrument_proto.whole_declaration_location.end)

  open_brace_loc = orig_src.find("{", params_end_loc)
  assert open_brace_loc >= 0

  src = orig_src

  src = src[:decl_begin_loc] + inst_srcs["before_func_src"] + src[
      decl_begin_loc + 0:]

  open_brace_loc += len(inst_srcs["before_func_src"])
  decl_end_loc += len(inst_srcs["before_func_src"])

  src = src[:open_brace_loc +
            1] + inst_srcs["instrument_src"] + src[open_brace_loc + 2:]
  decl_end_loc += len(inst_srcs["instrument_src"]) - 1

  src = src[:decl_end_loc] + inst_srcs["after_func_src"] + src[decl_end_loc +
                                                               0:]

  top_hdr = textwrap.dedent("""
#define XlsInt ac_int
""")

  src = top_hdr + src

  with open(filename_to_modify, "w") as f:
    f.write(src)

  return filename_to_modify


def create_temp(suffix):
  ret = tempfile.NamedTemporaryFile(suffix=suffix, delete=FLAGS.delete_temps)
  return ret


def main(argv):
  if len(argv) > 1:
    raise app.UsageError("Too many command-line arguments.")

  # Parse metadata
  meta_proto = metadata_output_pb2.MetadataOutput()
  with open(FLAGS.xlscc_metadata, "rb") as f:
    meta_proto.ParseFromString(f.read())

  # Find function to instrument
  function_to_instrument_proto = get_func_to_instrument(
      meta_proto, FLAGS.function_to_instrument)

  # Instrument C source
  original_src_tmp = create_temp(suffix=".cc")

  params_by_name = {}
  no_channels = True
  for param in function_to_instrument_proto.params:
    is_channel = param.type.HasField("as_inst") and (
        param.type.as_inst.name.fully_qualified_name == "__xls_channel")
    no_channels = no_channels and not is_channel
    params_by_name[param.name] = param

  channels_tmps = []
  block_channels_by_name = {}

  inst_srcs = {}
  if no_channels:
    inputs_tmp = create_temp(suffix=".ir")
    outputs_tmp = create_temp(suffix=".ir")
    inst_srcs = gen_instrumentation_pure_func(function_to_instrument_proto,
                                              inputs_tmp.name, outputs_tmp.name,
                                              meta_proto)
  else:
    reset_ticks_tmp = create_temp(suffix=".log")

    # Parse block data
    block_proto = hls_block_pb2.HLSBlock()
    with open(FLAGS.xlscc_block_metadata, "rb") as f:
      block_proto.ParseFromString(f.read())

    for hls_ch in block_proto.channels:
      block_channels_by_name[hls_ch.name] = hls_ch
      assert hls_ch.name in params_by_name
      channels_tmps += [(params_by_name[hls_ch.name], create_temp(suffix=".ir"))
                       ]

    inst_srcs = gen_instrumentation_proc(channels_tmps, reset_ticks_tmp.name,
                                         block_channels_by_name, meta_proto)

  print("Saving original source before instrumentation at ",
        original_src_tmp.name)
  filename_to_modify = instrument_source(function_to_instrument_proto,
                                         inst_srcs, original_src_tmp.name)

  # Build the C model
  print("Building C++ binary...")
  subprocess.check_call(FLAGS.build_cmd, shell=True)

  # Revert C source
  print("Reverting C++ source...")
  shutil.copyfile(original_src_tmp.name, filename_to_modify)

  # Run C model to generate traces
  print("Running C++ binary...")
  subprocess.check_call(FLAGS.cmd_to_test, shell=True)

  # Run IR simulator
  # check_output to suppress the spammy output of eval_ir_main
  print("Running IR simulator...")
  if no_channels:
    if FLAGS.backend == "block_interpreter":
      raise app.UsageError("Cannot use block interpreter for non-procs.")

    args = [
        EVAL_IR_PATH, "--top", function_to_instrument_proto.name.xls_name,
        "--input_file", inputs_tmp.name, "--expected_file", outputs_tmp.name,
        "--use_llvm_jit" if (FLAGS.backend == "serial_jit")
        else "--nouse_llvm_jit",
        FLAGS.ir_to_test
    ]
    print("Eval command: ", args)
    subprocess.check_output(args)
  else:

    with open(reset_ticks_tmp.name) as x:
      ticks_per_frame = x.read().split()

    ticks_per_frame = filter(lambda x: int(x) > 0, ticks_per_frame)

    input_ch_specs = []
    expected_ch_specs = []

    for ch, tmp in channels_tmps:
      assert ch.name in block_channels_by_name

      hls_ch = block_channels_by_name[ch.name]
      if hls_ch.is_input:
        input_ch_specs += [ch.name + "=" + tmp.name]
      else:
        expected_ch_specs += [ch.name + "=" + tmp.name]

    args = [
        EVAL_PROC_PATH, FLAGS.ir_to_test, "--ticks",
        ",".join(map(str, ticks_per_frame)), "--backend", FLAGS.backend,
        "--random_seed", str(FLAGS.random_seed),
        "--prob_input_valid_assert", str(FLAGS.prob_input_valid_assert),
        "--inputs_for_channels", ",".join(input_ch_specs),
        "--expected_outputs_for_channels", ",".join(expected_ch_specs)
    ]
    if FLAGS.block_signature_proto:
      args += ["--block_signature_proto", FLAGS.block_signature_proto]
    args += ["-v=1", "--logtostderr"]
    print("Eval command: ", " ".join(args))
    subprocess.check_output(args)

  # Done! If we got here, then there were no errors or mismatches
  print("... Success!")


if __name__ == "__main__":
  app.run(main)

# Copyright 2021 The XLS Authors # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # #      http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. A script for testing the translated IR for individual C++ functions.  Uses a standard C++ compiler as a reference.  The target function is instrumented using a RAII mechanism to capture its inputs and outputs. The instrumentation is generated according to information in the metadata protobuf generated by xlscc.  When the modified C++ binary is run, the instrumentation code writes inputs to one file, and expected outputs into another. These files, together with the IR from xlscc, are fed into the eval_ir_main tool, which checks that the results of IR simulation match the expected values file generated by the C++ binary. Find the linear index for a file + line location.    Args:     src: Source to search     loc: Location to find    Returns:     Integer index of loc Gets the metadata for the target function to be instrumented.    Args:     meta_proto: Protobuf from xlscc describing all the translated IR     function_to_instrument_name: Simple name of the function (not fully       qualified)    Returns:     Protobuf describing the target function for instrumentation Converts a template argument in the metadata protobuf to a C++ type name.    Args:     arg: Protobuf describing the argument    Returns:     The formatted C++ argument Converts a type as specified in the metadata protobuf to a C++ type name.    Args:     type_pb: Protobuf describing the type    Returns:     The formatted C++ type name Finds a type by id in the metadata protobuf.    Args:     id_pb: Id number of the type     metadata: Protobuf from xlscc describing all the translated IR    Returns:     The protobuf describing the type identified by id_pb Formats C++ source to print out a value.    Args:     name: Name of the value in the C++ source     stream_name: Name of the stream to print to, such as cout     type_pb: Protobuf describing the type of the value     metadata: Protobuf from xlscc describing all the translated IR    Returns:     The formatted C++ source Generates source blocks to insert.    Args:     channels_tmps: List of tuples, one for each channel to instrument     reset_ticks_tmp: The number of ticks between resets are written to this path     block_channels_by_name: HLSChannel protos by name     metadata: Protobuf from xlscc describing all the translated IR    Returns:     Dictionary of source segments to insert #include <fstream> #include <cassert>  class reset_recorder {  private:   int ticks;   std::ofstream stream;  public:   reset_recorder(const absl::string_view& out_path)     : ticks(0), stream(out_path.data(), std::ios_base::app) {   }   ~reset_recorder() {     stream << ticks << std::endl;    }   void tick() {     ++ticks;   } };  struct tick_on_exit {   reset_recorder& rec;   tick_on_exit(reset_recorder& rec) : rec(rec) {   }   ~tick_on_exit() {     rec.tick();   } }; HLS_STATIC reset_recorder reset_recorder_i("{path}");   // Don't count Run() as a tick if it never exits   tick_on_exit tick_on_exit_i(reset_recorder_i); static std::ofstream {n}_stream("{path}"); struct direct_in_on_exit_{n} {{                             {t} &value;                             std::ofstream& stream;                              direct_in_on_exit_{n} ({t} &value, std::ofstream& stream)                              : value(value), stream(stream) {{                             }}                             ~direct_in_on_exit_{n} () {{                               {print_val}                               {n}_stream << std::endl;                             }}                           }} direct_in_on_exit_{n}_i({n}, {n}_stream); #undef {n} class capture_channel_{n} {{                            private:                             typedef {t} T;                             ac_channel<T>& underlying;                             std::ostream& stream;                             bool did_read, did_write;                             void* inst;                            public:                             capture_channel_{n}(void* inst, ac_channel<T>& ch, std::ostream& stream)                               : underlying(ch),                                 stream(stream),                                 did_read(false),                                 did_write(false),                                 inst(inst) {{                             }}                              void write(const T& val) {{                               {print_val}                               (void)inst;                               stream << std::endl;                               did_write = true;                               assert(!did_read);                               underlying.write(val);                             }}                              T read() {{                               const T val(underlying.read());                               {print_val}                               stream << std::endl;                               did_read = true;                               assert(!did_write);                               return val;                             }}                           }};                            capture_channel_{n} {n}_captured(this, {n}, {n}_stream);                           #undef {n}                           #define {n} {n}_captured Generates source blocks to insert.    Args:     function_to_instrument_proto: Protobuf describing the target function     inputs_path: File path where the test binary should write the input values     outputs_path: File path where the test binary should write the output values     metadata: Protobuf from xlscc describing all the translated IR    Returns:     Dictionary of source segments to insert {return_type} instrument__retval;                       template<typename T>                       T operator=(const T& op) {{                         instrument__retval = static_cast<{return_type}>(op);                         return op;                       }} #include <fstream>                       static std::ofstream inputs_str("{in_path}");                       static std::ofstream outputs_str("{out_path}");                        class capture_return {{                        private:                        {saved_refs}                        public:                         capture_return({params})                          {inits} {{                       {enter_print_src}                           inputs_str << std::endl;                         }}                         ~capture_return() {{                       {exit_print_src}                           outputs_str << std::endl;                         }}                          {return_stuff}                       }}; capture_return capture_return_i ({call_constructor_src});                      #define return return capture_return_i = Modifies source file to instrument the target function.    Args:     function_to_instrument_proto: Protobuf describing the target function     inst_srcs: Dictionary of source segments to insert     original_src_tmp: File path at which to save the unmodified C++ source file    Returns:     The path to the C++ file that was modified #define XlsInt ac_int # Parse metadata # Find function to instrument # Instrument C source # Parse block data # Build the C model # Revert C source # Run C model to generate traces # Run IR simulator # check_output to suppress the spammy output of eval_ir_main # Done! If we got here, then there were no errors or mismatches