mirror of
				https://github.com/YosysHQ/sby.git
				synced 2025-11-04 06:39:11 +00:00 
			
		
		
		
	Unified trace generation using yosys's sim across all engines
Currently opt-in using the `fst` or `vcd_sim` options.
This commit is contained in:
		
							parent
							
								
									4c44a10f72
								
							
						
					
					
						commit
						6d3b5aa960
					
				
					 15 changed files with 686 additions and 183 deletions
				
			
		| 
						 | 
				
			
			@ -20,6 +20,9 @@ import os, re, sys, signal, platform, click
 | 
			
		|||
if os.name == "posix":
 | 
			
		||||
    import resource, fcntl
 | 
			
		||||
import subprocess
 | 
			
		||||
from dataclasses import dataclass, field
 | 
			
		||||
from collections import defaultdict
 | 
			
		||||
from typing import Optional
 | 
			
		||||
from shutil import copyfile, copytree, rmtree
 | 
			
		||||
from select import select
 | 
			
		||||
from time import monotonic, localtime, sleep, strftime
 | 
			
		||||
| 
						 | 
				
			
			@ -100,7 +103,7 @@ class SbyProc:
 | 
			
		|||
            dep.register_dep(self)
 | 
			
		||||
 | 
			
		||||
        self.output_callback = None
 | 
			
		||||
        self.exit_callback = None
 | 
			
		||||
        self.exit_callbacks = []
 | 
			
		||||
        self.error_callback = None
 | 
			
		||||
 | 
			
		||||
        if self.task.timeout_reached:
 | 
			
		||||
| 
						 | 
				
			
			@ -112,6 +115,9 @@ class SbyProc:
 | 
			
		|||
        else:
 | 
			
		||||
            self.notify.append(next_proc)
 | 
			
		||||
 | 
			
		||||
    def register_exit_callback(self, callback):
 | 
			
		||||
        self.exit_callbacks.append(callback)
 | 
			
		||||
 | 
			
		||||
    def log(self, line):
 | 
			
		||||
        if line is not None and (self.noprintregex is None or not self.noprintregex.match(line)):
 | 
			
		||||
            if self.logfile is not None:
 | 
			
		||||
| 
						 | 
				
			
			@ -130,8 +136,8 @@ class SbyProc:
 | 
			
		|||
            return
 | 
			
		||||
        if self.logfile is not None:
 | 
			
		||||
            self.logfile.close()
 | 
			
		||||
        if self.exit_callback is not None:
 | 
			
		||||
            self.exit_callback(retcode)
 | 
			
		||||
        for callback in self.exit_callbacks:
 | 
			
		||||
            callback(retcode)
 | 
			
		||||
 | 
			
		||||
    def handle_error(self, retcode):
 | 
			
		||||
        if self.terminated:
 | 
			
		||||
| 
						 | 
				
			
			@ -602,6 +608,199 @@ class SbyTaskloop:
 | 
			
		|||
        for task in self.tasks:
 | 
			
		||||
            task.exit_callback()
 | 
			
		||||
 | 
			
		||||
@dataclass
 | 
			
		||||
class SbySummaryEvent:
 | 
			
		||||
    engine_idx: int
 | 
			
		||||
    trace: Optional[str] = field(default=None)
 | 
			
		||||
    path: Optional[str] = field(default=None)
 | 
			
		||||
    hdlname: Optional[str] = field(default=None)
 | 
			
		||||
    type: Optional[str] = field(default=None)
 | 
			
		||||
    src: Optional[str] = field(default=None)
 | 
			
		||||
    step: Optional[int] = field(default=None)
 | 
			
		||||
    prop: Optional[SbyProperty] = field(default=None)
 | 
			
		||||
    engine_case: Optional[str] = field(default=None)
 | 
			
		||||
 | 
			
		||||
    @property
 | 
			
		||||
    def engine(self):
 | 
			
		||||
        return f"engine_{self.engine_idx}"
 | 
			
		||||
 | 
			
		||||
@dataclass
 | 
			
		||||
class SbyTraceSummary:
 | 
			
		||||
    trace: str
 | 
			
		||||
    path: Optional[str] = field(default=None)
 | 
			
		||||
    engine_case: Optional[str] = field(default=None)
 | 
			
		||||
    events: dict = field(default_factory=lambda: defaultdict(lambda: defaultdict(list)))
 | 
			
		||||
 | 
			
		||||
    @property
 | 
			
		||||
    def kind(self):
 | 
			
		||||
        if '$assert' in self.events:
 | 
			
		||||
            kind = 'counterexample trace'
 | 
			
		||||
        elif '$cover' in self.events:
 | 
			
		||||
            kind = 'cover trace'
 | 
			
		||||
        else:
 | 
			
		||||
            kind = 'trace'
 | 
			
		||||
        return kind
 | 
			
		||||
 | 
			
		||||
@dataclass
 | 
			
		||||
class SbyEngineSummary:
 | 
			
		||||
    engine_idx: int
 | 
			
		||||
    traces: dict = field(default_factory=dict)
 | 
			
		||||
    status: Optional[str] = field(default=None)
 | 
			
		||||
    unreached_covers: Optional[list] = field(default=None)
 | 
			
		||||
 | 
			
		||||
    @property
 | 
			
		||||
    def engine(self):
 | 
			
		||||
        return f"engine_{self.engine_idx}"
 | 
			
		||||
 | 
			
		||||
class SbySummary:
 | 
			
		||||
    def __init__(self, task):
 | 
			
		||||
        self.task = task
 | 
			
		||||
        self.timing = []
 | 
			
		||||
        self.lines = []
 | 
			
		||||
 | 
			
		||||
        self.engine_summaries = {}
 | 
			
		||||
        self.traces = defaultdict(dict)
 | 
			
		||||
        self.engine_status = {}
 | 
			
		||||
        self.unreached_covers = None
 | 
			
		||||
 | 
			
		||||
    def append(self, line):
 | 
			
		||||
        self.lines.append(line)
 | 
			
		||||
 | 
			
		||||
    def extend(self, lines):
 | 
			
		||||
        self.lines.extend(lines)
 | 
			
		||||
 | 
			
		||||
    def engine_summary(self, engine_idx):
 | 
			
		||||
        if engine_idx not in self.engine_summaries:
 | 
			
		||||
            self.engine_summaries[engine_idx] = SbyEngineSummary(engine_idx)
 | 
			
		||||
        return self.engine_summaries[engine_idx]
 | 
			
		||||
 | 
			
		||||
    def add_event(self, *args, **kwargs):
 | 
			
		||||
        event = SbySummaryEvent(*args, **kwargs)
 | 
			
		||||
        if event.prop:
 | 
			
		||||
            if event.type == "$assert":
 | 
			
		||||
                event.prop.status = "FAIL"
 | 
			
		||||
                if event.path:
 | 
			
		||||
                    event.prop.tracefiles.append(event.path)
 | 
			
		||||
        if event.prop:
 | 
			
		||||
            if event.type == "$cover":
 | 
			
		||||
                event.prop.status = "PASS"
 | 
			
		||||
                if event.path:
 | 
			
		||||
                    event.prop.tracefiles.append(event.path)
 | 
			
		||||
 | 
			
		||||
        engine = self.engine_summary(event.engine_idx)
 | 
			
		||||
 | 
			
		||||
        if event.trace not in engine.traces:
 | 
			
		||||
            engine.traces[event.trace] = SbyTraceSummary(event.trace, path=event.path, engine_case=event.engine_case)
 | 
			
		||||
 | 
			
		||||
        if event.type:
 | 
			
		||||
            by_type = engine.traces[event.trace].events[event.type]
 | 
			
		||||
            if event.hdlname:
 | 
			
		||||
                by_type[event.hdlname].append(event)
 | 
			
		||||
 | 
			
		||||
    def set_engine_status(self, engine_idx, status, case=None):
 | 
			
		||||
        engine_summary = self.engine_summary(engine_idx)
 | 
			
		||||
        if case is None:
 | 
			
		||||
            self.task.log(f"{click.style(f'engine_{engine_idx}', fg='magenta')}: Status returned by engine: {status}")
 | 
			
		||||
            self.engine_summary(engine_idx).status = status
 | 
			
		||||
        else:
 | 
			
		||||
            self.task.log(f"{click.style(f'engine_{engine_idx}.{case}', fg='magenta')}: Status returned by engine for {case}: {status}")
 | 
			
		||||
            if engine_summary.status is None:
 | 
			
		||||
                engine_summary.status = {}
 | 
			
		||||
            engine_summary.status[case] = status
 | 
			
		||||
 | 
			
		||||
    def summarize(self, short):
 | 
			
		||||
        omitted_excess = False
 | 
			
		||||
        for line in self.timing:
 | 
			
		||||
            yield line
 | 
			
		||||
 | 
			
		||||
        for engine_idx, engine_cmd in self.task.engine_list():
 | 
			
		||||
            engine_cmd = ' '.join(engine_cmd)
 | 
			
		||||
            trace_limit = 5
 | 
			
		||||
            prop_limit = 5
 | 
			
		||||
            step_limit = 5
 | 
			
		||||
            engine = self.engine_summary(engine_idx)
 | 
			
		||||
            if isinstance(engine.status, dict):
 | 
			
		||||
                for case, status in sorted(engine.status.items()):
 | 
			
		||||
                    yield f"{engine.engine} ({engine_cmd}) returned {status} for {case}"
 | 
			
		||||
            elif engine.status:
 | 
			
		||||
                yield f"{engine.engine} ({engine_cmd}) returned {engine.status}"
 | 
			
		||||
            else:
 | 
			
		||||
                yield f"{engine.engine} ({engine_cmd}) did not return a status"
 | 
			
		||||
 | 
			
		||||
            produced_traces = False
 | 
			
		||||
 | 
			
		||||
            for i, (trace_name, trace) in enumerate(sorted(engine.traces.items())):
 | 
			
		||||
                if short and i == trace_limit:
 | 
			
		||||
                    excess = len(engine.traces) - trace_limit
 | 
			
		||||
                    omitted_excess = True
 | 
			
		||||
                    yield f"and {excess} further trace{'s' if excess != 1 else ''}"
 | 
			
		||||
                    break
 | 
			
		||||
                case_suffix = f" [{trace.engine_case}]" if trace.engine_case else ""
 | 
			
		||||
                if trace.path:
 | 
			
		||||
                    if short:
 | 
			
		||||
                        yield f"{trace.kind}{case_suffix}: {self.task.workdir}/{trace.path}"
 | 
			
		||||
                    else:
 | 
			
		||||
                        yield f"{trace.kind}{case_suffix}: {trace.path}"
 | 
			
		||||
                else:
 | 
			
		||||
                    yield f"{trace.kind}{case_suffix}: <{trace.trace}>"
 | 
			
		||||
                produced_traces = True
 | 
			
		||||
                for event_type, events in sorted(trace.events.items()):
 | 
			
		||||
                    if event_type == '$assert':
 | 
			
		||||
                        desc = "failed assertion"
 | 
			
		||||
                        short_desc = 'assertion'
 | 
			
		||||
                    elif event_type == '$cover':
 | 
			
		||||
                        desc = "reached cover statement"
 | 
			
		||||
                        short_desc = 'cover statement'
 | 
			
		||||
                    elif event_type == '$assume':
 | 
			
		||||
                        desc = "violated assumption"
 | 
			
		||||
                        short_desc = 'assumption'
 | 
			
		||||
                    else:
 | 
			
		||||
                        continue
 | 
			
		||||
                    for j, (hdlname, same_events) in enumerate(sorted(events.items())):
 | 
			
		||||
                        if short and j == prop_limit:
 | 
			
		||||
                            excess = len(events) - prop_limit
 | 
			
		||||
                            yield f"  and {excess} further {short_desc}{'s' if excess != 1 else ''}"
 | 
			
		||||
                            break
 | 
			
		||||
 | 
			
		||||
                        event = same_events[0]
 | 
			
		||||
                        steps = sorted(e.step for e in same_events)
 | 
			
		||||
                        if short and len(steps) > step_limit:
 | 
			
		||||
                            steps = [str(step) for step in steps[:step_limit]]
 | 
			
		||||
                            excess = len(steps) - step_limit
 | 
			
		||||
                            omitted_excess = True
 | 
			
		||||
                            steps[-1] += f" and {excess} further step{'s' if excess != 1 else ''}"
 | 
			
		||||
 | 
			
		||||
                        steps = f"step{'s' if len(steps) > 1 else ''} {', '.join(map(str, steps))}"
 | 
			
		||||
                        yield f"  {desc} {event.hdlname} at {event.src} in {steps}"
 | 
			
		||||
 | 
			
		||||
            if not produced_traces:
 | 
			
		||||
                yield f"{engine.engine} did not produce any traces"
 | 
			
		||||
 | 
			
		||||
        if self.unreached_covers is None and self.task.opt_mode == 'cover' and self.task.status != "PASS" and self.task.design:
 | 
			
		||||
            self.unreached_covers = []
 | 
			
		||||
            for prop in self.task.design.hierarchy:
 | 
			
		||||
                if prop.type == prop.Type.COVER and prop.status == "UNKNOWN":
 | 
			
		||||
                    self.unreached_covers.append(prop)
 | 
			
		||||
 | 
			
		||||
        if self.unreached_covers:
 | 
			
		||||
            yield f"unreached cover statements:"
 | 
			
		||||
            for j, prop in enumerate(self.unreached_covers):
 | 
			
		||||
                if short and j == prop_limit:
 | 
			
		||||
                    excess = len(self.unreached_covers) - prop_limit
 | 
			
		||||
                    omitted_excess = True
 | 
			
		||||
                    yield f"  and {excess} further propert{'ies' if excess != 1 else 'y'}"
 | 
			
		||||
                    break
 | 
			
		||||
                yield f"  {prop.hdlname} at {prop.location}"
 | 
			
		||||
 | 
			
		||||
        for line in self.lines:
 | 
			
		||||
            yield line
 | 
			
		||||
 | 
			
		||||
        if omitted_excess:
 | 
			
		||||
            yield f"see {self.task.workdir}/{self.task.status} for a complete summary"
 | 
			
		||||
    def __iter__(self):
 | 
			
		||||
        yield from self.summarize(True)
 | 
			
		||||
 | 
			
		||||
 | 
			
		||||
 | 
			
		||||
class SbyTask(SbyConfig):
 | 
			
		||||
    def __init__(self, sbyconfig, workdir, early_logs, reusedir, taskloop=None, logfile=None):
 | 
			
		||||
| 
						 | 
				
			
			@ -644,7 +843,7 @@ class SbyTask(SbyConfig):
 | 
			
		|||
            ru = resource.getrusage(resource.RUSAGE_CHILDREN)
 | 
			
		||||
            self.start_process_time = ru.ru_utime + ru.ru_stime
 | 
			
		||||
 | 
			
		||||
        self.summary = list()
 | 
			
		||||
        self.summary = SbySummary(self)
 | 
			
		||||
 | 
			
		||||
        self.logfile = logfile or open(f"{workdir}/logfile.txt", "a")
 | 
			
		||||
        self.log_targets = [sys.stdout, self.logfile]
 | 
			
		||||
| 
						 | 
				
			
			@ -696,6 +895,15 @@ class SbyTask(SbyConfig):
 | 
			
		|||
        for target in self.log_targets:
 | 
			
		||||
            click.echo(line, file=target)
 | 
			
		||||
 | 
			
		||||
    def log_prefix(self, prefix, message=None):
 | 
			
		||||
        prefix = f"{click.style(prefix, fg='magenta')}: "
 | 
			
		||||
        def log(message):
 | 
			
		||||
            self.log(f"{prefix}{message}")
 | 
			
		||||
        if message is None:
 | 
			
		||||
            return log
 | 
			
		||||
        else:
 | 
			
		||||
            log(message)
 | 
			
		||||
 | 
			
		||||
    def error(self, logmessage):
 | 
			
		||||
        tm = localtime()
 | 
			
		||||
        self.log(click.style(f"ERROR: {logmessage}", fg="red", bold=True))
 | 
			
		||||
| 
						 | 
				
			
			@ -833,7 +1041,7 @@ class SbyTask(SbyConfig):
 | 
			
		|||
            def instance_hierarchy_error_callback(retcode):
 | 
			
		||||
                self.precise_prop_status = False
 | 
			
		||||
 | 
			
		||||
            proc.exit_callback = instance_hierarchy_callback
 | 
			
		||||
            proc.register_exit_callback(instance_hierarchy_callback)
 | 
			
		||||
            proc.error_callback = instance_hierarchy_error_callback
 | 
			
		||||
 | 
			
		||||
            return [proc]
 | 
			
		||||
| 
						 | 
				
			
			@ -891,8 +1099,8 @@ class SbyTask(SbyConfig):
 | 
			
		|||
                print("delete -output", file=f)
 | 
			
		||||
                print("dffunmap", file=f)
 | 
			
		||||
                print("stat", file=f)
 | 
			
		||||
                print("write_btor {}-i design_{m}.info design_{m}.btor".format("-c " if self.opt_mode == "cover" else "", m=model_name), file=f)
 | 
			
		||||
                print("write_btor -s {}-i design_{m}_single.info design_{m}_single.btor".format("-c " if self.opt_mode == "cover" else "", m=model_name), file=f)
 | 
			
		||||
                print("write_btor {}-i design_{m}.info -ywmap design_btor.ywb design_{m}.btor".format("-c " if self.opt_mode == "cover" else "", m=model_name), file=f)
 | 
			
		||||
                print("write_btor -s {}-i design_{m}_single.info -ywmap design_btor_single.ywb design_{m}_single.btor".format("-c " if self.opt_mode == "cover" else "", m=model_name), file=f)
 | 
			
		||||
 | 
			
		||||
            proc = SbyProc(
 | 
			
		||||
                self,
 | 
			
		||||
| 
						 | 
				
			
			@ -967,6 +1175,8 @@ class SbyTask(SbyConfig):
 | 
			
		|||
        if new_status == "PASS":
 | 
			
		||||
            assert self.status != "FAIL"
 | 
			
		||||
            self.status = "PASS"
 | 
			
		||||
            if self.opt_mode in ("bmc", "prove") and self.design:
 | 
			
		||||
                self.design.pass_unknown_asserts()
 | 
			
		||||
 | 
			
		||||
        elif new_status == "FAIL":
 | 
			
		||||
            assert self.status != "PASS"
 | 
			
		||||
| 
						 | 
				
			
			@ -1004,11 +1214,17 @@ class SbyTask(SbyConfig):
 | 
			
		|||
        self.handle_int_option("timeout", None)
 | 
			
		||||
 | 
			
		||||
        self.handle_bool_option("vcd", True)
 | 
			
		||||
        self.handle_bool_option("vcd_sim", False)
 | 
			
		||||
        self.handle_bool_option("fst", False)
 | 
			
		||||
 | 
			
		||||
        self.handle_str_option("smtc", None)
 | 
			
		||||
        self.handle_int_option("skip", None)
 | 
			
		||||
        self.handle_str_option("tbtop", None)
 | 
			
		||||
 | 
			
		||||
        if self.opt_mode != "live":
 | 
			
		||||
            self.handle_int_option("append", 0)
 | 
			
		||||
            self.handle_bool_option("append_assume", False)
 | 
			
		||||
 | 
			
		||||
        self.handle_str_option("make_model", None)
 | 
			
		||||
 | 
			
		||||
    def setup_procs(self, setupmode):
 | 
			
		||||
| 
						 | 
				
			
			@ -1078,18 +1294,18 @@ class SbyTask(SbyConfig):
 | 
			
		|||
 | 
			
		||||
            # TODO process time is incorrect when running in parallel
 | 
			
		||||
 | 
			
		||||
            self.summary = [
 | 
			
		||||
            self.summary.timing = [
 | 
			
		||||
                "Elapsed clock time [H:MM:SS (secs)]: {}:{:02d}:{:02d} ({})".format
 | 
			
		||||
                        (total_clock_time // (60*60), (total_clock_time // 60) % 60, total_clock_time % 60, total_clock_time),
 | 
			
		||||
                "Elapsed process time [H:MM:SS (secs)]: {}:{:02d}:{:02d} ({})".format
 | 
			
		||||
                        (total_process_time // (60*60), (total_process_time // 60) % 60, total_process_time % 60, total_process_time),
 | 
			
		||||
            ] + self.summary
 | 
			
		||||
            ]
 | 
			
		||||
        else:
 | 
			
		||||
            self.summary = [
 | 
			
		||||
            self.summary.timing = [
 | 
			
		||||
                "Elapsed clock time [H:MM:SS (secs)]: {}:{:02d}:{:02d} ({})".format
 | 
			
		||||
                        (total_clock_time // (60*60), (total_clock_time // 60) % 60, total_clock_time % 60, total_clock_time),
 | 
			
		||||
                "Elapsed process time unvailable on Windows"
 | 
			
		||||
            ] + self.summary
 | 
			
		||||
            ]
 | 
			
		||||
 | 
			
		||||
        for line in self.summary:
 | 
			
		||||
            if line.startswith("Elapsed"):
 | 
			
		||||
| 
						 | 
				
			
			@ -1110,7 +1326,7 @@ class SbyTask(SbyConfig):
 | 
			
		|||
 | 
			
		||||
    def write_summary_file(self):
 | 
			
		||||
        with open(f"{self.workdir}/{self.status}", "w") as f:
 | 
			
		||||
            for line in self.summary:
 | 
			
		||||
            for line in self.summary.summarize(short=False):
 | 
			
		||||
                click.echo(line, file=f)
 | 
			
		||||
 | 
			
		||||
    def print_junit_result(self, f, junit_ts_name, junit_tc_name, junit_format_strict=False):
 | 
			
		||||
| 
						 | 
				
			
			
 | 
			
		|||
| 
						 | 
				
			
			@ -16,9 +16,37 @@
 | 
			
		|||
# OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
 | 
			
		||||
#
 | 
			
		||||
 | 
			
		||||
import json
 | 
			
		||||
import json, re
 | 
			
		||||
from enum import Enum, auto
 | 
			
		||||
from dataclasses import dataclass, field
 | 
			
		||||
from typing import Optional, Tuple
 | 
			
		||||
 | 
			
		||||
 | 
			
		||||
addr_re = re.compile(r'\\\[[0-9]+\]$')
 | 
			
		||||
public_name_re = re.compile(r"\\([a-zA-Z_][a-zA-Z0-9_]*(\[[0-9]+\])?|\[[0-9]+\])$")
 | 
			
		||||
 | 
			
		||||
def pretty_name(id):
 | 
			
		||||
    if public_name_re.match(id):
 | 
			
		||||
        return id.lstrip("\\")
 | 
			
		||||
    else:
 | 
			
		||||
        return id
 | 
			
		||||
 | 
			
		||||
def pretty_path(path):
 | 
			
		||||
    out = ""
 | 
			
		||||
    for name in path:
 | 
			
		||||
        name = pretty_name(name)
 | 
			
		||||
        if name.startswith("["):
 | 
			
		||||
            out += name
 | 
			
		||||
            continue
 | 
			
		||||
        if out:
 | 
			
		||||
            out += "."
 | 
			
		||||
        if name.startswith("\\") or name.startswith("$"):
 | 
			
		||||
            out += name + " "
 | 
			
		||||
        else:
 | 
			
		||||
            out += name
 | 
			
		||||
 | 
			
		||||
    return out
 | 
			
		||||
 | 
			
		||||
 | 
			
		||||
@dataclass
 | 
			
		||||
class SbyProperty:
 | 
			
		||||
| 
						 | 
				
			
			@ -44,18 +72,36 @@ class SbyProperty:
 | 
			
		|||
            raise ValueError("Unknown property type: " + name)
 | 
			
		||||
 | 
			
		||||
    name: str
 | 
			
		||||
    path: Tuple[str, ...]
 | 
			
		||||
    type: Type
 | 
			
		||||
    location: str
 | 
			
		||||
    hierarchy: str
 | 
			
		||||
    status: str = field(default="UNKNOWN")
 | 
			
		||||
    tracefile: str = field(default="")
 | 
			
		||||
    tracefiles: str = field(default_factory=list)
 | 
			
		||||
 | 
			
		||||
    @property
 | 
			
		||||
    def tracefile(self):
 | 
			
		||||
        if self.tracefiles:
 | 
			
		||||
            return self.tracefiles[0]
 | 
			
		||||
        else:
 | 
			
		||||
            return ""
 | 
			
		||||
 | 
			
		||||
    @property
 | 
			
		||||
    def celltype(self):
 | 
			
		||||
        return f"${str(self.type).lower()}"
 | 
			
		||||
 | 
			
		||||
    @property
 | 
			
		||||
    def hdlname(self):
 | 
			
		||||
        return pretty_path(self.path).rstrip()
 | 
			
		||||
 | 
			
		||||
 | 
			
		||||
    def __repr__(self):
 | 
			
		||||
        return f"SbyProperty<{self.type} {self.name} at {self.location}: status={self.status}, tracefile=\"{self.tracefile}\">"
 | 
			
		||||
        return f"SbyProperty<{self.type} {self.name} {self.path} at {self.location}: status={self.status}, tracefile=\"{self.tracefile}\">"
 | 
			
		||||
 | 
			
		||||
@dataclass
 | 
			
		||||
class SbyModule:
 | 
			
		||||
    name: str
 | 
			
		||||
    path: Tuple[str, ...]
 | 
			
		||||
    type: str
 | 
			
		||||
    submodules: dict = field(default_factory=dict)
 | 
			
		||||
    properties: list = field(default_factory=list)
 | 
			
		||||
| 
						 | 
				
			
			@ -105,15 +151,32 @@ class SbyDesign:
 | 
			
		|||
    hierarchy: SbyModule = None
 | 
			
		||||
    memory_bits: int = 0
 | 
			
		||||
    forall: bool = False
 | 
			
		||||
    properties_by_path: dict = field(default_factory=dict)
 | 
			
		||||
 | 
			
		||||
    def pass_unknown_asserts(self):
 | 
			
		||||
        for prop in self.hierarchy:
 | 
			
		||||
            if prop.type == prop.Type.ASSERT and prop.status == "UNKNOWN":
 | 
			
		||||
                prop.status = "PASS"
 | 
			
		||||
 | 
			
		||||
 | 
			
		||||
def cell_path(cell):
 | 
			
		||||
    path = cell["attributes"].get("hdlname")
 | 
			
		||||
    if path is None:
 | 
			
		||||
        if cell["name"].startswith('$'):
 | 
			
		||||
            return (cell["name"],)
 | 
			
		||||
        else:
 | 
			
		||||
            return ("\\" + cell["name"],)
 | 
			
		||||
    else:
 | 
			
		||||
        return tuple(f"\\{segment}" for segment in path.split())
 | 
			
		||||
 | 
			
		||||
 | 
			
		||||
def design_hierarchy(filename):
 | 
			
		||||
    design = SbyDesign(hierarchy=None)
 | 
			
		||||
    design_json = json.load(filename)
 | 
			
		||||
    def make_mod_hier(instance_name, module_name, hierarchy=""):
 | 
			
		||||
    def make_mod_hier(instance_name, module_name, hierarchy="", path=()):
 | 
			
		||||
        # print(instance_name,":", module_name)
 | 
			
		||||
        sub_hierarchy=f"{hierarchy}/{instance_name}" if hierarchy else instance_name
 | 
			
		||||
        mod = SbyModule(name=instance_name, type=module_name)
 | 
			
		||||
        mod = SbyModule(name=instance_name, path=path, type=module_name)
 | 
			
		||||
 | 
			
		||||
        for m in design_json["modules"]:
 | 
			
		||||
            if m["name"] == module_name:
 | 
			
		||||
| 
						 | 
				
			
			@ -129,11 +192,17 @@ def design_hierarchy(filename):
 | 
			
		|||
                        location = cell["attributes"]["src"]
 | 
			
		||||
                    except KeyError:
 | 
			
		||||
                        location = ""
 | 
			
		||||
                    p = SbyProperty(name=cell["name"], type=SbyProperty.Type.from_cell(sort["type"]), location=location, hierarchy=sub_hierarchy)
 | 
			
		||||
                    p = SbyProperty(
 | 
			
		||||
                        name=cell["name"],
 | 
			
		||||
                        path=(*path, *cell_path(cell)),
 | 
			
		||||
                        type=SbyProperty.Type.from_cell(sort["type"]),
 | 
			
		||||
                        location=location,
 | 
			
		||||
                        hierarchy=sub_hierarchy)
 | 
			
		||||
                    mod.properties.append(p)
 | 
			
		||||
            if sort["type"][0] != '$' or sort["type"].startswith("$paramod"):
 | 
			
		||||
                for cell in sort["cells"]:
 | 
			
		||||
                    mod.submodules[cell["name"]] = make_mod_hier(cell["name"], sort["type"], hierarchy=sub_hierarchy)
 | 
			
		||||
                    mod.submodules[cell["name"]] = make_mod_hier(
 | 
			
		||||
                        cell["name"], sort["type"], sub_hierarchy, (*path, *cell_path(cell)))
 | 
			
		||||
            if sort["type"] in ["$mem", "$mem_v2"]:
 | 
			
		||||
                for cell in sort["cells"]:
 | 
			
		||||
                    design.memory_bits += int(cell["parameters"]["WIDTH"], 2) * int(cell["parameters"]["SIZE"], 2)
 | 
			
		||||
| 
						 | 
				
			
			@ -145,7 +214,10 @@ def design_hierarchy(filename):
 | 
			
		|||
    for m in design_json["modules"]:
 | 
			
		||||
        attrs = m["attributes"]
 | 
			
		||||
        if "top" in attrs and int(attrs["top"]) == 1:
 | 
			
		||||
            design.hierarchy = make_mod_hier(m["name"], m["name"])
 | 
			
		||||
            design.hierarchy = make_mod_hier(m["name"], m["name"], "", (m["name"],))
 | 
			
		||||
 | 
			
		||||
            for prop in design.hierarchy:
 | 
			
		||||
                design.properties_by_path[prop.path[1:]] = prop
 | 
			
		||||
            return design
 | 
			
		||||
    else:
 | 
			
		||||
        raise ValueError("Cannot find top module")
 | 
			
		||||
| 
						 | 
				
			
			
 | 
			
		|||
| 
						 | 
				
			
			@ -16,8 +16,9 @@
 | 
			
		|||
# OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
 | 
			
		||||
#
 | 
			
		||||
 | 
			
		||||
import re, os, getopt, click
 | 
			
		||||
import re, getopt
 | 
			
		||||
from sby_core import SbyProc
 | 
			
		||||
from sby_engine_aiger import aigsmt_exit_callback
 | 
			
		||||
 | 
			
		||||
def run(mode, task, engine_idx, engine):
 | 
			
		||||
    abc_opts, abc_command = getopt.getopt(engine[1:], "", [])
 | 
			
		||||
| 
						 | 
				
			
			@ -46,6 +47,21 @@ def run(mode, task, engine_idx, engine):
 | 
			
		|||
    else:
 | 
			
		||||
        task.error(f"Invalid ABC command {abc_command[0]}.")
 | 
			
		||||
 | 
			
		||||
    smtbmc_vcd = task.opt_vcd and not task.opt_vcd_sim
 | 
			
		||||
    run_aigsmt = smtbmc_vcd or (task.opt_append and task.opt_append_assume)
 | 
			
		||||
    smtbmc_append = 0
 | 
			
		||||
    sim_append = 0
 | 
			
		||||
    log = task.log_prefix(f"engine_{engine_idx}")
 | 
			
		||||
 | 
			
		||||
    if task.opt_append_assume:
 | 
			
		||||
        smtbmc_append = task.opt_append
 | 
			
		||||
    elif smtbmc_vcd:
 | 
			
		||||
        if not task.opt_append_assume:
 | 
			
		||||
            log("For VCDs generated by smtbmc the option 'append_assume off' is ignored")
 | 
			
		||||
        smtbmc_append = task.opt_append
 | 
			
		||||
    else:
 | 
			
		||||
        sim_append = task.opt_append
 | 
			
		||||
 | 
			
		||||
    proc = SbyProc(
 | 
			
		||||
        task,
 | 
			
		||||
        f"engine_{engine_idx}",
 | 
			
		||||
| 
						 | 
				
			
			@ -79,64 +95,8 @@ def run(mode, task, engine_idx, engine):
 | 
			
		|||
        return line
 | 
			
		||||
 | 
			
		||||
    def exit_callback(retcode):
 | 
			
		||||
        if proc_status is None:
 | 
			
		||||
            task.error(f"engine_{engine_idx}: Could not determine engine status.")
 | 
			
		||||
 | 
			
		||||
        task.update_status(proc_status)
 | 
			
		||||
        task.log(f"{click.style(f'engine_{engine_idx}', fg='magenta')}: Status returned by engine: {proc_status}")
 | 
			
		||||
        task.summary.append(f"""engine_{engine_idx} ({" ".join(engine)}) returned {proc_status}""")
 | 
			
		||||
 | 
			
		||||
        task.terminate()
 | 
			
		||||
 | 
			
		||||
        if proc_status == "FAIL" and task.opt_aigsmt != "none":
 | 
			
		||||
            trace_prefix = f"engine_{engine_idx}/trace"
 | 
			
		||||
 | 
			
		||||
            smtbmc_opts = []
 | 
			
		||||
            smtbmc_opts += ["-s", task.opt_aigsmt]
 | 
			
		||||
            if task.opt_tbtop is not None:
 | 
			
		||||
                smtbmc_opts  += ["--vlogtb-top", task.opt_tbtop]
 | 
			
		||||
            smtbmc_opts += ["--noprogress", f"--append {task.opt_append}"]
 | 
			
		||||
            if task.opt_vcd:
 | 
			
		||||
                smtbmc_opts += [f"--dump-vcd {trace_prefix}.vcd"]
 | 
			
		||||
            smtbmc_opts += [f"--dump-yw {trace_prefix}.yw", f"--dump-vlogtb {trace_prefix}_tb.v", f"--dump-smtc {trace_prefix}.smtc"]
 | 
			
		||||
 | 
			
		||||
            witness_proc = SbyProc(
 | 
			
		||||
                task, f"engine_{engine_idx}", [],
 | 
			
		||||
                f"cd {task.workdir}; {task.exe_paths['witness']} aiw2yw engine_{engine_idx}/trace.aiw model/design_aiger.ywa engine_{engine_idx}/trace.yw",
 | 
			
		||||
            )
 | 
			
		||||
 | 
			
		||||
            proc2 = SbyProc(
 | 
			
		||||
                task,
 | 
			
		||||
                f"engine_{engine_idx}",
 | 
			
		||||
                [*task.model("smt2"), witness_proc],
 | 
			
		||||
                f"cd {task.workdir}; {task.exe_paths['smtbmc']} {' '.join(smtbmc_opts)} --yw engine_{engine_idx}/trace.yw model/design_smt2.smt2",
 | 
			
		||||
                logfile=open(f"{task.workdir}/engine_{engine_idx}/logfile2.txt", "w")
 | 
			
		||||
            )
 | 
			
		||||
 | 
			
		||||
            proc2_status = None
 | 
			
		||||
 | 
			
		||||
            def output_callback2(line):
 | 
			
		||||
                nonlocal proc2_status
 | 
			
		||||
 | 
			
		||||
                match = re.match(r"^## [0-9: ]+ Status: FAILED", line)
 | 
			
		||||
                if match: proc2_status = "FAIL"
 | 
			
		||||
 | 
			
		||||
                match = re.match(r"^## [0-9: ]+ Status: PASSED", line)
 | 
			
		||||
                if match: proc2_status = "PASS"
 | 
			
		||||
 | 
			
		||||
                return line
 | 
			
		||||
 | 
			
		||||
            def exit_callback2(retcode):
 | 
			
		||||
                if proc2_status is None:
 | 
			
		||||
                    task.error(f"engine_{engine_idx}: Could not determine aigsmt status.")
 | 
			
		||||
                if proc2_status != "FAIL":
 | 
			
		||||
                    task.error(f"engine_{engine_idx}: Unexpected aigsmt status.")
 | 
			
		||||
 | 
			
		||||
                if os.path.exists(f"{task.workdir}/engine_{engine_idx}/trace.vcd"):
 | 
			
		||||
                    task.summary.append(f"counterexample trace: {task.workdir}/engine_{engine_idx}/trace.vcd")
 | 
			
		||||
 | 
			
		||||
            proc2.output_callback = output_callback2
 | 
			
		||||
            proc2.exit_callback = exit_callback2
 | 
			
		||||
        aigsmt_exit_callback(task, engine_idx, proc_status,
 | 
			
		||||
            run_aigsmt=run_aigsmt, smtbmc_vcd=smtbmc_vcd, smtbmc_append=smtbmc_append, sim_append=sim_append, )
 | 
			
		||||
 | 
			
		||||
    proc.output_callback = output_callback
 | 
			
		||||
    proc.exit_callback = exit_callback
 | 
			
		||||
    proc.register_exit_callback(exit_callback)
 | 
			
		||||
| 
						 | 
				
			
			
 | 
			
		|||
| 
						 | 
				
			
			@ -18,6 +18,7 @@
 | 
			
		|||
 | 
			
		||||
import re, os, getopt, click
 | 
			
		||||
from sby_core import SbyProc
 | 
			
		||||
from sby_sim import sim_witness_trace
 | 
			
		||||
 | 
			
		||||
def run(mode, task, engine_idx, engine):
 | 
			
		||||
    opts, solver_args = getopt.getopt(engine[1:], "", [])
 | 
			
		||||
| 
						 | 
				
			
			@ -51,6 +52,22 @@ def run(mode, task, engine_idx, engine):
 | 
			
		|||
    else:
 | 
			
		||||
        task.error(f"Invalid solver command {solver_args[0]}.")
 | 
			
		||||
 | 
			
		||||
    smtbmc_vcd = task.opt_vcd and not task.opt_vcd_sim
 | 
			
		||||
    run_aigsmt = (mode != "live") and (smtbmc_vcd or (task.opt_append and task.opt_append_assume))
 | 
			
		||||
    smtbmc_append = 0
 | 
			
		||||
    sim_append = 0
 | 
			
		||||
    log = task.log_prefix(f"engine_{engine_idx}")
 | 
			
		||||
 | 
			
		||||
    if mode != "live":
 | 
			
		||||
        if task.opt_append_assume:
 | 
			
		||||
            smtbmc_append = task.opt_append
 | 
			
		||||
        elif smtbmc_vcd:
 | 
			
		||||
            if not task.opt_append_assume:
 | 
			
		||||
                log("For VCDs generated by smtbmc the option 'append_assume off' is ignored")
 | 
			
		||||
            smtbmc_append = task.opt_append
 | 
			
		||||
        else:
 | 
			
		||||
            sim_append = task.opt_append
 | 
			
		||||
 | 
			
		||||
    proc = SbyProc(
 | 
			
		||||
        task,
 | 
			
		||||
        f"engine_{engine_idx}",
 | 
			
		||||
| 
						 | 
				
			
			@ -92,44 +109,48 @@ def run(mode, task, engine_idx, engine):
 | 
			
		|||
        return None
 | 
			
		||||
 | 
			
		||||
    def exit_callback(retcode):
 | 
			
		||||
        aiw_file.close()
 | 
			
		||||
        aigsmt_exit_callback(task, engine_idx, proc_status,
 | 
			
		||||
            run_aigsmt=run_aigsmt, smtbmc_vcd=smtbmc_vcd, smtbmc_append=smtbmc_append, sim_append=sim_append, )
 | 
			
		||||
 | 
			
		||||
    proc.output_callback = output_callback
 | 
			
		||||
    proc.register_exit_callback(exit_callback)
 | 
			
		||||
 | 
			
		||||
 | 
			
		||||
def aigsmt_exit_callback(task, engine_idx, proc_status, *, run_aigsmt, smtbmc_vcd, smtbmc_append, sim_append):
 | 
			
		||||
        if proc_status is None:
 | 
			
		||||
            task.error(f"engine_{engine_idx}: Could not determine engine status.")
 | 
			
		||||
 | 
			
		||||
        aiw_file.close()
 | 
			
		||||
 | 
			
		||||
        task.update_status(proc_status)
 | 
			
		||||
        task.log(f"{click.style(f'engine_{engine_idx}', fg='magenta')}: Status returned by engine: {proc_status}")
 | 
			
		||||
        task.summary.append(f"""engine_{engine_idx} ({" ".join(engine)}) returned {proc_status}""")
 | 
			
		||||
 | 
			
		||||
        task.summary.set_engine_status(engine_idx, proc_status)
 | 
			
		||||
        task.terminate()
 | 
			
		||||
 | 
			
		||||
        if proc_status == "FAIL" and task.opt_aigsmt != "none":
 | 
			
		||||
            if produced_cex:
 | 
			
		||||
        if proc_status == "FAIL" and (not run_aigsmt or task.opt_aigsmt != "none"):
 | 
			
		||||
            trace_prefix = f"engine_{engine_idx}/trace"
 | 
			
		||||
 | 
			
		||||
                smtbmc_opts = []
 | 
			
		||||
                if mode == "live":
 | 
			
		||||
                    smtbmc_opts += ["-g"]
 | 
			
		||||
                smtbmc_opts += ["-s", task.opt_aigsmt]
 | 
			
		||||
                if task.opt_tbtop is not None:
 | 
			
		||||
                    smtbmc_opts  += ["--vlogtb-top", task.opt_tbtop]
 | 
			
		||||
                smtbmc_opts += ["--noprogress"]
 | 
			
		||||
                if mode != "live":
 | 
			
		||||
                    smtbmc_opts += [f"--append {task.opt_append}"]
 | 
			
		||||
                if task.opt_vcd:
 | 
			
		||||
                    smtbmc_opts += [f"--dump-vcd {trace_prefix}.vcd"]
 | 
			
		||||
                smtbmc_opts += [f"--dump-yw {trace_prefix}.yw", f"--dump-vlogtb {trace_prefix}_tb.v", f"--dump-smtc {trace_prefix}.smtc"]
 | 
			
		||||
            aiw2yw_suffix = '_aiw' if run_aigsmt else ''
 | 
			
		||||
 | 
			
		||||
            witness_proc = SbyProc(
 | 
			
		||||
                task, f"engine_{engine_idx}", [],
 | 
			
		||||
                    f"cd {task.workdir}; {task.exe_paths['witness']} aiw2yw engine_{engine_idx}/trace.aiw model/design_aiger.ywa engine_{engine_idx}/trace.yw",
 | 
			
		||||
                f"cd {task.workdir}; {task.exe_paths['witness']} aiw2yw engine_{engine_idx}/trace.aiw model/design_aiger.ywa engine_{engine_idx}/trace{aiw2yw_suffix}.yw",
 | 
			
		||||
            )
 | 
			
		||||
            yw_proc = witness_proc
 | 
			
		||||
 | 
			
		||||
            if run_aigsmt:
 | 
			
		||||
                smtbmc_opts = []
 | 
			
		||||
                smtbmc_opts += ["-s", task.opt_aigsmt]
 | 
			
		||||
                if task.opt_tbtop is not None:
 | 
			
		||||
                    smtbmc_opts  += ["--vlogtb-top", task.opt_tbtop]
 | 
			
		||||
                smtbmc_opts += ["--noprogress", f"--append {smtbmc_append}"]
 | 
			
		||||
                if smtbmc_vcd:
 | 
			
		||||
                    smtbmc_opts += [f"--dump-vcd {trace_prefix}.vcd"]
 | 
			
		||||
                smtbmc_opts += [f"--dump-yw {trace_prefix}.yw", f"--dump-vlogtb {trace_prefix}_tb.v", f"--dump-smtc {trace_prefix}.smtc"]
 | 
			
		||||
 | 
			
		||||
                proc2 = SbyProc(
 | 
			
		||||
                    task,
 | 
			
		||||
                    f"engine_{engine_idx}",
 | 
			
		||||
                    [*task.model("smt2"), witness_proc],
 | 
			
		||||
                    f"cd {task.workdir}; {task.exe_paths['smtbmc']} {' '.join(smtbmc_opts)} --yw engine_{engine_idx}/trace.yw model/design_smt2.smt2",
 | 
			
		||||
                    f"cd {task.workdir}; {task.exe_paths['smtbmc']} {' '.join(smtbmc_opts)} --yw engine_{engine_idx}/trace{aiw2yw_suffix}.yw model/design_smt2.smt2",
 | 
			
		||||
                    logfile=open(f"{task.workdir}/engine_{engine_idx}/logfile2.txt", "w")
 | 
			
		||||
                )
 | 
			
		||||
 | 
			
		||||
| 
						 | 
				
			
			@ -146,20 +167,21 @@ def run(mode, task, engine_idx, engine):
 | 
			
		|||
 | 
			
		||||
                    return line
 | 
			
		||||
 | 
			
		||||
                def exit_callback2(line):
 | 
			
		||||
                def exit_callback2(retcode):
 | 
			
		||||
                    if proc2_status is None:
 | 
			
		||||
                        task.error(f"engine_{engine_idx}: Could not determine aigsmt status.")
 | 
			
		||||
                    if proc2_status != ("PASS" if mode == "live" else "FAIL"):
 | 
			
		||||
                    if proc2_status != "FAIL":
 | 
			
		||||
                        task.error(f"engine_{engine_idx}: Unexpected aigsmt status.")
 | 
			
		||||
 | 
			
		||||
                    if os.path.exists(f"{task.workdir}/engine_{engine_idx}/trace.vcd"):
 | 
			
		||||
                        task.summary.append(f"counterexample trace: {task.workdir}/engine_{engine_idx}/trace.vcd")
 | 
			
		||||
                        task.summary.add_event(engine_idx, trace="trace", path=f"engine_{engine_idx}/trace.vcd", type="$assert")
 | 
			
		||||
 | 
			
		||||
                proc2.output_callback = output_callback2
 | 
			
		||||
                proc2.exit_callback = exit_callback2
 | 
			
		||||
                proc2.register_exit_callback(exit_callback2)
 | 
			
		||||
 | 
			
		||||
                yw_proc = proc2
 | 
			
		||||
 | 
			
		||||
            if task.opt_fst or (task.opt_vcd and task.opt_vcd_sim):
 | 
			
		||||
                sim_witness_trace(f"engine_{engine_idx}", task, engine_idx, f"engine_{engine_idx}/trace.yw", append=sim_append, deps=[yw_proc])
 | 
			
		||||
 | 
			
		||||
            else:
 | 
			
		||||
                task.log(f"{click.style(f'engine_{engine_idx}', fg='magenta')}: Engine did not produce a counter example.")
 | 
			
		||||
 | 
			
		||||
    proc.output_callback = output_callback
 | 
			
		||||
    proc.exit_callback = exit_callback
 | 
			
		||||
| 
						 | 
				
			
			
 | 
			
		|||
| 
						 | 
				
			
			@ -19,6 +19,7 @@
 | 
			
		|||
import re, os, getopt, click
 | 
			
		||||
from types import SimpleNamespace
 | 
			
		||||
from sby_core import SbyProc
 | 
			
		||||
from sby_sim import sim_witness_trace
 | 
			
		||||
 | 
			
		||||
def run(mode, task, engine_idx, engine):
 | 
			
		||||
    random_seed = None
 | 
			
		||||
| 
						 | 
				
			
			@ -54,14 +55,27 @@ def run(mode, task, engine_idx, engine):
 | 
			
		|||
    else:
 | 
			
		||||
        task.error(f"Invalid solver command {solver_args[0]}.")
 | 
			
		||||
 | 
			
		||||
    log = task.log_prefix(f"engine_{engine_idx}")
 | 
			
		||||
 | 
			
		||||
    btorsim_vcd = task.opt_vcd and not task.opt_vcd_sim
 | 
			
		||||
    run_sim = task.opt_fst or not btorsim_vcd
 | 
			
		||||
    sim_append = 0
 | 
			
		||||
 | 
			
		||||
    if task.opt_append and btorsim_vcd:
 | 
			
		||||
        log("The BTOR engine does not support the 'append' option when using btorsim.")
 | 
			
		||||
    else:
 | 
			
		||||
        sim_append = task.opt_append
 | 
			
		||||
 | 
			
		||||
    if task.opt_append and task.opt_append_assume:
 | 
			
		||||
        log("The BTOR engine does not support enforcing assumptions in appended time steps.")
 | 
			
		||||
 | 
			
		||||
 | 
			
		||||
    common_state = SimpleNamespace()
 | 
			
		||||
    common_state.solver_status = None
 | 
			
		||||
    common_state.produced_cex = 0
 | 
			
		||||
    common_state.expected_cex = 1
 | 
			
		||||
    common_state.wit_file = None
 | 
			
		||||
    common_state.assert_fail = False
 | 
			
		||||
    common_state.produced_traces = []
 | 
			
		||||
    common_state.print_traces_max = 5
 | 
			
		||||
    common_state.running_procs = 0
 | 
			
		||||
 | 
			
		||||
    def print_traces_and_terminate():
 | 
			
		||||
| 
						 | 
				
			
			@ -87,17 +101,7 @@ def run(mode, task, engine_idx, engine):
 | 
			
		|||
                task.error(f"engine_{engine_idx}: Engine terminated without status.")
 | 
			
		||||
 | 
			
		||||
        task.update_status(proc_status.upper())
 | 
			
		||||
        task.log(f"{click.style(f'engine_{engine_idx}', fg='magenta')}: Status returned by engine: {proc_status}")
 | 
			
		||||
        task.summary.append(f"""engine_{engine_idx} ({" ".join(engine)}) returned {proc_status}""")
 | 
			
		||||
 | 
			
		||||
        if len(common_state.produced_traces) == 0:
 | 
			
		||||
            task.log(f"""{click.style(f'engine_{engine_idx}', fg='magenta')}: Engine did not produce a{" counter" if mode != "cover" else "n "}example.""")
 | 
			
		||||
        elif len(common_state.produced_traces) <= common_state.print_traces_max:
 | 
			
		||||
            task.summary.extend(common_state.produced_traces)
 | 
			
		||||
        else:
 | 
			
		||||
            task.summary.extend(common_state.produced_traces[:common_state.print_traces_max])
 | 
			
		||||
            excess_traces = len(common_state.produced_traces) - common_state.print_traces_max
 | 
			
		||||
            task.summary.append(f"""and {excess_traces} further trace{"s" if excess_traces > 1 else ""}""")
 | 
			
		||||
        task.summary.set_engine_status(engine_idx, proc_status)
 | 
			
		||||
 | 
			
		||||
        task.terminate()
 | 
			
		||||
 | 
			
		||||
| 
						 | 
				
			
			@ -113,9 +117,9 @@ def run(mode, task, engine_idx, engine):
 | 
			
		|||
 | 
			
		||||
    def make_exit_callback(suffix):
 | 
			
		||||
        def exit_callback2(retcode):
 | 
			
		||||
            vcdpath = f"{task.workdir}/engine_{engine_idx}/trace{suffix}.vcd"
 | 
			
		||||
            if os.path.exists(vcdpath):
 | 
			
		||||
                common_state.produced_traces.append(f"""{"" if mode == "cover" else "counterexample "}trace: {vcdpath}""")
 | 
			
		||||
            vcdpath = f"engine_{engine_idx}/trace{suffix}.vcd"
 | 
			
		||||
            if os.path.exists(f"{task.workdir}/{vcdpath}"):
 | 
			
		||||
                task.summary.add_event(engine_idx=engine_idx, trace=f'trace{suffix}', path=vcdpath, type="$cover" if mode == "cover" else "$assert")
 | 
			
		||||
 | 
			
		||||
            common_state.running_procs -= 1
 | 
			
		||||
            if (common_state.running_procs == 0):
 | 
			
		||||
| 
						 | 
				
			
			@ -123,6 +127,11 @@ def run(mode, task, engine_idx, engine):
 | 
			
		|||
 | 
			
		||||
        return exit_callback2
 | 
			
		||||
 | 
			
		||||
    def simple_exit_callback(retcode):
 | 
			
		||||
        common_state.running_procs -= 1
 | 
			
		||||
        if (common_state.running_procs == 0):
 | 
			
		||||
            print_traces_and_terminate()
 | 
			
		||||
 | 
			
		||||
    def output_callback(line):
 | 
			
		||||
        if mode == "cover":
 | 
			
		||||
            if solver_args[0] == "btormc":
 | 
			
		||||
| 
						 | 
				
			
			@ -153,20 +162,39 @@ def run(mode, task, engine_idx, engine):
 | 
			
		|||
                else:
 | 
			
		||||
                    suffix = common_state.produced_cex
 | 
			
		||||
 | 
			
		||||
                if mode == "cover" or task.opt_vcd:
 | 
			
		||||
                model = f"design_btor{'_single' if solver_args[0] == 'pono' else ''}"
 | 
			
		||||
 | 
			
		||||
                yw_proc = SbyProc(
 | 
			
		||||
                    task, f"engine_{engine_idx}.trace{suffix}", [],
 | 
			
		||||
                    f"cd {task.workdir}; {task.exe_paths['witness']} wit2yw engine_{engine_idx}/trace{suffix}.wit model/{model}.ywb engine_{engine_idx}/trace{suffix}.yw",
 | 
			
		||||
                )
 | 
			
		||||
                common_state.running_procs += 1
 | 
			
		||||
                yw_proc.register_exit_callback(simple_exit_callback)
 | 
			
		||||
 | 
			
		||||
                btorsim_vcd = (task.opt_vcd and not task.opt_vcd_sim)
 | 
			
		||||
 | 
			
		||||
                if btorsim_vcd:
 | 
			
		||||
                    # TODO cover runs btorsim not only for trace generation, can we run it without VCD generation in that case?
 | 
			
		||||
                    proc2 = SbyProc(
 | 
			
		||||
                        task,
 | 
			
		||||
                        f"engine_{engine_idx}_{common_state.produced_cex}",
 | 
			
		||||
                        f"engine_{engine_idx}.trace{suffix}",
 | 
			
		||||
                        task.model("btor"),
 | 
			
		||||
                        "cd {dir} ; btorsim -c --vcd engine_{idx}/trace{i}.vcd --hierarchical-symbols --info model/design_btor{s}.info model/design_btor{s}.btor engine_{idx}/trace{i}.wit".format(dir=task.workdir, idx=engine_idx, i=suffix, s='_single' if solver_args[0] == 'pono' else ''),
 | 
			
		||||
                        "cd {dir} ; btorsim -c --vcd engine_{idx}/trace{i}{i2}.vcd --hierarchical-symbols --info model/design_btor{s}.info model/design_btor{s}.btor engine_{idx}/trace{i}.wit".format(dir=task.workdir, idx=engine_idx, i=suffix, i2='' if btorsim_vcd else '_btorsim', s='_single' if solver_args[0] == 'pono' else ''),
 | 
			
		||||
                        logfile=open(f"{task.workdir}/engine_{engine_idx}/logfile2.txt", "w")
 | 
			
		||||
                    )
 | 
			
		||||
                    proc2.output_callback = output_callback2
 | 
			
		||||
                    proc2.exit_callback = make_exit_callback(suffix)
 | 
			
		||||
                    if run_sim:
 | 
			
		||||
                        proc2.register_exit_callback(simple_exit_callback)
 | 
			
		||||
                    else:
 | 
			
		||||
                        proc2.register_exit_callback(make_exit_callback(suffix))
 | 
			
		||||
                    proc2.checkretcode = True
 | 
			
		||||
                    common_state.running_procs += 1
 | 
			
		||||
 | 
			
		||||
                if run_sim:
 | 
			
		||||
                    sim_proc = sim_witness_trace(f"engine_{engine_idx}", task, engine_idx, f"engine_{engine_idx}/trace{suffix}.yw", append=sim_append, deps=[yw_proc])
 | 
			
		||||
                    sim_proc.register_exit_callback(simple_exit_callback)
 | 
			
		||||
                    common_state.running_procs += 1
 | 
			
		||||
 | 
			
		||||
                common_state.produced_cex += 1
 | 
			
		||||
                common_state.wit_file.close()
 | 
			
		||||
                common_state.wit_file = None
 | 
			
		||||
| 
						 | 
				
			
			@ -226,5 +254,5 @@ def run(mode, task, engine_idx, engine):
 | 
			
		|||
    if solver_args[0] == "pono":
 | 
			
		||||
        proc.retcodes = [0, 1, 255] # UNKNOWN = -1, FALSE = 0, TRUE = 1, ERROR = 2
 | 
			
		||||
    proc.output_callback = output_callback
 | 
			
		||||
    proc.exit_callback = exit_callback
 | 
			
		||||
    proc.register_exit_callback(exit_callback)
 | 
			
		||||
    common_state.running_procs += 1
 | 
			
		||||
| 
						 | 
				
			
			
 | 
			
		|||
| 
						 | 
				
			
			@ -16,8 +16,9 @@
 | 
			
		|||
# OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
 | 
			
		||||
#
 | 
			
		||||
 | 
			
		||||
import re, os, getopt, click
 | 
			
		||||
import re, os, getopt, click, glob
 | 
			
		||||
from sby_core import SbyProc
 | 
			
		||||
from sby_sim import sim_witness_trace
 | 
			
		||||
 | 
			
		||||
def run(mode, task, engine_idx, engine):
 | 
			
		||||
    smtbmc_opts = []
 | 
			
		||||
| 
						 | 
				
			
			@ -141,20 +142,37 @@ def run(mode, task, engine_idx, engine):
 | 
			
		|||
    if not progress:
 | 
			
		||||
        smtbmc_opts.append("--noprogress")
 | 
			
		||||
 | 
			
		||||
 | 
			
		||||
    if task.opt_skip is not None:
 | 
			
		||||
        t_opt = "{}:{}".format(task.opt_skip, task.opt_depth)
 | 
			
		||||
    else:
 | 
			
		||||
        t_opt = "{}".format(task.opt_depth)
 | 
			
		||||
 | 
			
		||||
    smtbmc_vcd = task.opt_vcd and not task.opt_vcd_sim
 | 
			
		||||
 | 
			
		||||
    smtbmc_append = 0
 | 
			
		||||
    sim_append = 0
 | 
			
		||||
 | 
			
		||||
    log = task.log_prefix(f"engine_{engine_idx}")
 | 
			
		||||
 | 
			
		||||
    if task.opt_append_assume:
 | 
			
		||||
        smtbmc_append = task.opt_append
 | 
			
		||||
    elif smtbmc_vcd:
 | 
			
		||||
        if not task.opt_append_assume:
 | 
			
		||||
            log("For VCDs generated by smtbmc the option 'append_assume off' is ignored")
 | 
			
		||||
        smtbmc_append = task.opt_append
 | 
			
		||||
    else:
 | 
			
		||||
        sim_append = task.opt_append
 | 
			
		||||
 | 
			
		||||
    trace_ext = 'fst' if task.opt_fst else 'vcd'
 | 
			
		||||
 | 
			
		||||
    random_seed = f"--info \"(set-option :random-seed {random_seed})\"" if random_seed else ""
 | 
			
		||||
    dump_flags = f"--dump-vcd {trace_prefix}.vcd " if task.opt_vcd else ""
 | 
			
		||||
    dump_flags = f"--dump-vcd {trace_prefix}.vcd " if smtbmc_vcd else ""
 | 
			
		||||
    dump_flags += f"--dump-yw {trace_prefix}.yw --dump-vlogtb {trace_prefix}_tb.v --dump-smtc {trace_prefix}.smtc"
 | 
			
		||||
    proc = SbyProc(
 | 
			
		||||
        task,
 | 
			
		||||
        procname,
 | 
			
		||||
        task.model(model_name),
 | 
			
		||||
        f"""cd {task.workdir}; {task.exe_paths["smtbmc"]} {" ".join(smtbmc_opts)} -t {t_opt} {random_seed} --append {task.opt_append} {dump_flags} model/design_{model_name}.smt2""",
 | 
			
		||||
        f"""cd {task.workdir}; {task.exe_paths["smtbmc"]} {" ".join(smtbmc_opts)} -t {t_opt} {random_seed} --append {smtbmc_append} {dump_flags} model/design_{model_name}.smt2""",
 | 
			
		||||
        logfile=open(logfile_prefix + ".txt", "w"),
 | 
			
		||||
        logstderr=(not progress)
 | 
			
		||||
    )
 | 
			
		||||
| 
						 | 
				
			
			@ -167,12 +185,30 @@ def run(mode, task, engine_idx, engine):
 | 
			
		|||
 | 
			
		||||
    proc_status = None
 | 
			
		||||
    last_prop = []
 | 
			
		||||
    pending_sim = None
 | 
			
		||||
    current_step = None
 | 
			
		||||
    procs_running = 1
 | 
			
		||||
 | 
			
		||||
    def output_callback(line):
 | 
			
		||||
        nonlocal proc_status
 | 
			
		||||
        nonlocal last_prop
 | 
			
		||||
        nonlocal pending_sim
 | 
			
		||||
        nonlocal current_step
 | 
			
		||||
        nonlocal procs_running
 | 
			
		||||
 | 
			
		||||
        if pending_sim:
 | 
			
		||||
            sim_proc = sim_witness_trace(procname, task, engine_idx, pending_sim, append=sim_append)
 | 
			
		||||
            sim_proc.register_exit_callback(simple_exit_callback)
 | 
			
		||||
            procs_running += 1
 | 
			
		||||
            pending_sim = None
 | 
			
		||||
 | 
			
		||||
        smt2_trans = {'\\':'/', '|':'/'}
 | 
			
		||||
 | 
			
		||||
        match = re.match(r"^## [0-9: ]+ .* in step ([0-9]+)\.\.", line)
 | 
			
		||||
        if match:
 | 
			
		||||
            current_step = int(match[1])
 | 
			
		||||
            return line
 | 
			
		||||
 | 
			
		||||
        match = re.match(r"^## [0-9: ]+ Status: FAILED", line)
 | 
			
		||||
        if match:
 | 
			
		||||
            proc_status = "FAIL"
 | 
			
		||||
| 
						 | 
				
			
			@ -193,31 +229,45 @@ def run(mode, task, engine_idx, engine):
 | 
			
		|||
            proc_status = "ERROR"
 | 
			
		||||
            return line
 | 
			
		||||
 | 
			
		||||
        match = re.match(r"^## [0-9: ]+ Assert failed in (\S+): (\S+) \((\S+)\)", line)
 | 
			
		||||
        match = re.match(r"^## [0-9: ]+ Assert failed in (\S+): (\S+)(?: \((\S+)\))?", line)
 | 
			
		||||
        if match:
 | 
			
		||||
            cell_name = match[3]
 | 
			
		||||
            cell_name = match[3] or match[2]
 | 
			
		||||
            prop = task.design.hierarchy.find_property_by_cellname(cell_name, trans_dict=smt2_trans)
 | 
			
		||||
            prop.status = "FAIL"
 | 
			
		||||
            last_prop.append(prop)
 | 
			
		||||
            return line
 | 
			
		||||
 | 
			
		||||
        match = re.match(r"^## [0-9: ]+ Reached cover statement at (\S+) \((\S+)\) in step \d+.", line)
 | 
			
		||||
        match = re.match(r"^## [0-9: ]+ Reached cover statement at (\S+)(?: \((\S+)\))? in step \d+\.", line)
 | 
			
		||||
        if match:
 | 
			
		||||
            cell_name = match[2]
 | 
			
		||||
            cell_name = match[2] or match[1]
 | 
			
		||||
            prop = task.design.hierarchy.find_property_by_cellname(cell_name, trans_dict=smt2_trans)
 | 
			
		||||
            prop.status = "PASS"
 | 
			
		||||
            last_prop.append(prop)
 | 
			
		||||
            return line
 | 
			
		||||
 | 
			
		||||
        if smtbmc_vcd and not task.opt_fst:
 | 
			
		||||
            match = re.match(r"^## [0-9: ]+ Writing trace to VCD file: (\S+)", line)
 | 
			
		||||
            if match:
 | 
			
		||||
                tracefile = match[1]
 | 
			
		||||
                trace = os.path.basename(tracefile)[:-4]
 | 
			
		||||
                engine_case = mode.split('_')[1] if '_' in mode else None
 | 
			
		||||
                task.summary.add_event(engine_idx=engine_idx, trace=trace, path=tracefile, engine_case=engine_case)
 | 
			
		||||
 | 
			
		||||
            if match and last_prop:
 | 
			
		||||
                for p in last_prop:
 | 
			
		||||
                if not p.tracefile:
 | 
			
		||||
                    p.tracefile = match[1]
 | 
			
		||||
                    task.summary.add_event(
 | 
			
		||||
                        engine_idx=engine_idx, trace=trace,
 | 
			
		||||
                        type=p.celltype, hdlname=p.hdlname, src=p.location, step=current_step)
 | 
			
		||||
                    p.tracefiles.append(tracefile)
 | 
			
		||||
                last_prop = []
 | 
			
		||||
                return line
 | 
			
		||||
        else:
 | 
			
		||||
            match = re.match(r"^## [0-9: ]+ Writing trace to Yosys witness file: (\S+)", line)
 | 
			
		||||
            if match:
 | 
			
		||||
                tracefile = match[1]
 | 
			
		||||
                pending_sim = tracefile
 | 
			
		||||
 | 
			
		||||
        match = re.match(r"^## [0-9: ]+ Unreached cover statement at (\S+) \((\S+)\).", line)
 | 
			
		||||
        match = re.match(r"^## [0-9: ]+ Unreached cover statement at (\S+) \((\S+)\)\.", line)
 | 
			
		||||
        if match:
 | 
			
		||||
            cell_name = match[2]
 | 
			
		||||
            prop = task.design.hierarchy.find_property_by_cellname(cell_name, trans_dict=smt2_trans)
 | 
			
		||||
| 
						 | 
				
			
			@ -225,43 +275,28 @@ def run(mode, task, engine_idx, engine):
 | 
			
		|||
 | 
			
		||||
        return line
 | 
			
		||||
 | 
			
		||||
    def simple_exit_callback(retcode):
 | 
			
		||||
        nonlocal procs_running
 | 
			
		||||
        procs_running -= 1
 | 
			
		||||
        if not procs_running:
 | 
			
		||||
            last_exit_callback()
 | 
			
		||||
 | 
			
		||||
    def exit_callback(retcode):
 | 
			
		||||
        if proc_status is None:
 | 
			
		||||
            task.error(f"engine_{engine_idx}: Engine terminated without status.")
 | 
			
		||||
        simple_exit_callback(retcode)
 | 
			
		||||
 | 
			
		||||
    def last_exit_callback():
 | 
			
		||||
        if mode == "bmc" or mode == "cover":
 | 
			
		||||
            task.update_status(proc_status)
 | 
			
		||||
            proc_status_lower = proc_status.lower() if proc_status == "PASS" else proc_status
 | 
			
		||||
            task.log(f"{click.style(f'engine_{engine_idx}', fg='magenta')}: Status returned by engine: {proc_status_lower}")
 | 
			
		||||
            task.summary.append(f"""engine_{engine_idx} ({" ".join(engine)}) returned {proc_status_lower}""")
 | 
			
		||||
 | 
			
		||||
            if proc_status == "FAIL" and mode != "cover":
 | 
			
		||||
                if os.path.exists(f"{task.workdir}/engine_{engine_idx}/trace.vcd"):
 | 
			
		||||
                    task.summary.append(f"counterexample trace: {task.workdir}/engine_{engine_idx}/trace.vcd")
 | 
			
		||||
            elif proc_status == "PASS" and mode == "cover":
 | 
			
		||||
                print_traces_max = 5
 | 
			
		||||
                for i in range(print_traces_max):
 | 
			
		||||
                    if os.path.exists(f"{task.workdir}/engine_{engine_idx}/trace{i}.vcd"):
 | 
			
		||||
                        task.summary.append(f"trace: {task.workdir}/engine_{engine_idx}/trace{i}.vcd")
 | 
			
		||||
                    else:
 | 
			
		||||
                        break
 | 
			
		||||
                else:
 | 
			
		||||
                    excess_traces = 0
 | 
			
		||||
                    while os.path.exists(f"{task.workdir}/engine_{engine_idx}/trace{print_traces_max + excess_traces}.vcd"):
 | 
			
		||||
                        excess_traces += 1
 | 
			
		||||
                    if excess_traces > 0:
 | 
			
		||||
                        task.summary.append(f"""and {excess_traces} further trace{"s" if excess_traces > 1 else ""}""")
 | 
			
		||||
            elif proc_status == "PASS" and mode == "bmc":
 | 
			
		||||
                for prop in task.design.hierarchy:
 | 
			
		||||
                    if prop.type == prop.Type.ASSERT and prop.status == "UNKNOWN":
 | 
			
		||||
                        prop.status = "PASS"
 | 
			
		||||
            task.summary.set_engine_status(engine_idx, proc_status_lower)
 | 
			
		||||
 | 
			
		||||
            task.terminate()
 | 
			
		||||
 | 
			
		||||
        elif mode in ["prove_basecase", "prove_induction"]:
 | 
			
		||||
            proc_status_lower = proc_status.lower() if proc_status == "PASS" else proc_status
 | 
			
		||||
            task.log(f"""{click.style(f'engine_{engine_idx}', fg='magenta')}: Status returned by engine for {mode.split("_")[1]}: {proc_status_lower}""")
 | 
			
		||||
            task.summary.append(f"""engine_{engine_idx} ({" ".join(engine)}) returned {proc_status_lower} for {mode.split("_")[1]}""")
 | 
			
		||||
            task.summary.set_engine_status(engine_idx, proc_status_lower, mode.split("_")[1])
 | 
			
		||||
 | 
			
		||||
            if mode == "prove_basecase":
 | 
			
		||||
                for proc in task.basecase_procs:
 | 
			
		||||
| 
						 | 
				
			
			@ -272,8 +307,6 @@ def run(mode, task, engine_idx, engine):
 | 
			
		|||
 | 
			
		||||
                else:
 | 
			
		||||
                    task.update_status(proc_status)
 | 
			
		||||
                    if os.path.exists(f"{task.workdir}/engine_{engine_idx}/trace.vcd"):
 | 
			
		||||
                        task.summary.append(f"counterexample trace: {task.workdir}/engine_{engine_idx}/trace.vcd")
 | 
			
		||||
                    task.terminate()
 | 
			
		||||
 | 
			
		||||
            elif mode == "prove_induction":
 | 
			
		||||
| 
						 | 
				
			
			@ -287,9 +320,6 @@ def run(mode, task, engine_idx, engine):
 | 
			
		|||
                assert False
 | 
			
		||||
 | 
			
		||||
            if task.basecase_pass and task.induction_pass:
 | 
			
		||||
                for prop in task.design.hierarchy:
 | 
			
		||||
                    if prop.type == prop.Type.ASSERT and prop.status == "UNKNOWN":
 | 
			
		||||
                        prop.status = "PASS"
 | 
			
		||||
                task.update_status("PASS")
 | 
			
		||||
                task.summary.append("successful proof by k-induction.")
 | 
			
		||||
                task.terminate()
 | 
			
		||||
| 
						 | 
				
			
			@ -298,4 +328,4 @@ def run(mode, task, engine_idx, engine):
 | 
			
		|||
            assert False
 | 
			
		||||
 | 
			
		||||
    proc.output_callback = output_callback
 | 
			
		||||
    proc.exit_callback = exit_callback
 | 
			
		||||
    proc.register_exit_callback(exit_callback)
 | 
			
		||||
| 
						 | 
				
			
			
 | 
			
		|||
| 
						 | 
				
			
			@ -21,7 +21,6 @@ from sby_core import SbyProc
 | 
			
		|||
 | 
			
		||||
def run(task):
 | 
			
		||||
    task.handle_int_option("depth", 20)
 | 
			
		||||
    task.handle_int_option("append", 0)
 | 
			
		||||
    task.handle_str_option("aigsmt", "yices")
 | 
			
		||||
 | 
			
		||||
    for engine_idx, engine in task.engine_list():
 | 
			
		||||
| 
						 | 
				
			
			
 | 
			
		|||
| 
						 | 
				
			
			@ -21,7 +21,6 @@ from sby_core import SbyProc
 | 
			
		|||
 | 
			
		||||
def run(task):
 | 
			
		||||
    task.handle_int_option("depth", 20)
 | 
			
		||||
    task.handle_int_option("append", 0)
 | 
			
		||||
 | 
			
		||||
    for engine_idx, engine in task.engine_list():
 | 
			
		||||
        task.log(f"{click.style(f'engine_{engine_idx}', fg='magenta')}: {' '.join(engine)}")
 | 
			
		||||
| 
						 | 
				
			
			
 | 
			
		|||
| 
						 | 
				
			
			@ -21,7 +21,6 @@ from sby_core import SbyProc
 | 
			
		|||
 | 
			
		||||
def run(task):
 | 
			
		||||
    task.handle_int_option("depth", 20)
 | 
			
		||||
    task.handle_int_option("append", 0)
 | 
			
		||||
    task.handle_str_option("aigsmt", "yices")
 | 
			
		||||
 | 
			
		||||
    task.status = "UNKNOWN"
 | 
			
		||||
| 
						 | 
				
			
			
 | 
			
		|||
							
								
								
									
										96
									
								
								sbysrc/sby_sim.py
									
										
									
									
									
										Normal file
									
								
							
							
						
						
									
										96
									
								
								sbysrc/sby_sim.py
									
										
									
									
									
										Normal file
									
								
							| 
						 | 
				
			
			@ -0,0 +1,96 @@
 | 
			
		|||
#
 | 
			
		||||
# SymbiYosys (sby) -- Front-end for Yosys-based formal verification flows
 | 
			
		||||
#
 | 
			
		||||
# Copyright (C) 2022  Jannis Harder <jix@yosyshq.com> <me@jix.one>
 | 
			
		||||
#
 | 
			
		||||
# Permission to use, copy, modify, and/or distribute this software for any
 | 
			
		||||
# purpose with or without fee is hereby granted, provided that the above
 | 
			
		||||
# copyright notice and this permission notice appear in all copies.
 | 
			
		||||
#
 | 
			
		||||
# THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
 | 
			
		||||
# WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
 | 
			
		||||
# MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
 | 
			
		||||
# ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
 | 
			
		||||
# WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
 | 
			
		||||
# ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
 | 
			
		||||
# OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
 | 
			
		||||
#
 | 
			
		||||
 | 
			
		||||
import os, re, glob, json
 | 
			
		||||
from sby_core import SbyProc
 | 
			
		||||
from sby_design import pretty_path
 | 
			
		||||
 | 
			
		||||
def sim_witness_trace(prefix, task, engine_idx, witness_file, *, append, deps=()):
 | 
			
		||||
    trace_name = os.path.basename(witness_file)[:-3]
 | 
			
		||||
    formats = []
 | 
			
		||||
    tracefile = None
 | 
			
		||||
    if task.opt_vcd and task.opt_vcd_sim:
 | 
			
		||||
        tracefile = f"engine_{engine_idx}/{trace_name}.vcd"
 | 
			
		||||
        formats.append(f"-vcd {trace_name}.vcd")
 | 
			
		||||
    if task.opt_fst:
 | 
			
		||||
        tracefile = f"engine_{engine_idx}/{trace_name}.fst"
 | 
			
		||||
        formats.append(f"-fst {trace_name}.fst")
 | 
			
		||||
 | 
			
		||||
    # for warnings / error messages
 | 
			
		||||
    error_tracefile = f"{task.workdir}/{tracefile}" or f"{task.workdir}/engine_{engine_idx}/{trace_name}.yw"
 | 
			
		||||
 | 
			
		||||
    sim_log = task.log_prefix(f"{prefix}.{trace_name}")
 | 
			
		||||
 | 
			
		||||
    sim_log(f"Generating simulation trace for witness file: {witness_file}")
 | 
			
		||||
 | 
			
		||||
    with open(f"{task.workdir}/engine_{engine_idx}/{trace_name}.ys", "w") as f:
 | 
			
		||||
        print(f"# running in {task.workdir}/engine_{engine_idx}/", file=f)
 | 
			
		||||
        print(f"read_rtlil ../model/design_prep.il", file=f)
 | 
			
		||||
        print(f"sim -hdlname -summary {trace_name}.json -append {append} -r {trace_name}.yw {' '.join(formats)}", file=f)
 | 
			
		||||
 | 
			
		||||
    def exit_callback(retval):
 | 
			
		||||
 | 
			
		||||
        if task.design:
 | 
			
		||||
            task.precise_prop_status = True
 | 
			
		||||
 | 
			
		||||
        assertion_types = set()
 | 
			
		||||
 | 
			
		||||
        with open(f"{task.workdir}/engine_{engine_idx}/{trace_name}.json") as summary:
 | 
			
		||||
            summary = json.load(summary)
 | 
			
		||||
            for assertion in summary["assertions"]:
 | 
			
		||||
                assertion["path"] = tuple(assertion["path"])
 | 
			
		||||
 | 
			
		||||
        first_appended = summary["steps"] + 1 - append
 | 
			
		||||
 | 
			
		||||
        printed_assumption_warning = False
 | 
			
		||||
 | 
			
		||||
        task.summary.add_event(engine_idx=engine_idx, trace=trace_name, path=tracefile)
 | 
			
		||||
 | 
			
		||||
        for assertion in summary["assertions"]:
 | 
			
		||||
            if task.design:
 | 
			
		||||
                prop = task.design.properties_by_path[tuple(assertion["path"])]
 | 
			
		||||
            else:
 | 
			
		||||
                prop = None
 | 
			
		||||
 | 
			
		||||
            hdlname = pretty_path((summary['top'], *assertion['path'])).rstrip()
 | 
			
		||||
            task.summary.add_event(
 | 
			
		||||
                engine_idx=engine_idx,
 | 
			
		||||
                trace=trace_name, path=tracefile, hdlname=hdlname,
 | 
			
		||||
                type=assertion["type"], src=assertion.get("src"), step=assertion["step"],
 | 
			
		||||
                prop=prop)
 | 
			
		||||
 | 
			
		||||
            assertion_types.add(assertion["type"])
 | 
			
		||||
 | 
			
		||||
            if assertion["type"] == '$assume':
 | 
			
		||||
                if assertion["step"] < first_appended:
 | 
			
		||||
                    task.error(f"produced trace {error_tracefile!r} violates assumptions during simulation")
 | 
			
		||||
                elif not printed_assumption_warning:
 | 
			
		||||
                    sim_log(f"Warning: trace {error_tracefile!r} violates assumptions during simulation of the appended time steps.")
 | 
			
		||||
                    if not task.opt_append_assume:
 | 
			
		||||
                        sim_log("For supported engines, the option 'append_assume on' can be used to find inputs that uphold assumptions during appended time steps.")
 | 
			
		||||
                    printed_assumption_warning = True
 | 
			
		||||
 | 
			
		||||
    proc = SbyProc(
 | 
			
		||||
        task,
 | 
			
		||||
        f"{prefix}.{trace_name}",
 | 
			
		||||
        deps,
 | 
			
		||||
        f"""cd {task.workdir}/engine_{engine_idx}; {task.exe_paths["yosys"]} -ql {trace_name}.log {trace_name}.ys""",
 | 
			
		||||
    )
 | 
			
		||||
    proc.noprintregex = re.compile(r"Warning: Assert .* failed.*")
 | 
			
		||||
    proc.register_exit_callback(exit_callback)
 | 
			
		||||
    return proc
 | 
			
		||||
| 
						 | 
				
			
			@ -11,7 +11,7 @@ step_5 = line_ref(workdir, src, "step 5")
 | 
			
		|||
step_7 = line_ref(workdir, src, "step 7")
 | 
			
		||||
 | 
			
		||||
log = open(workdir + "/logfile.txt").read()
 | 
			
		||||
log_per_trace = log.split("Writing trace to VCD file")[:-1]
 | 
			
		||||
log_per_trace = log.split("Writing trace to Yosys witness file")[:-1]
 | 
			
		||||
 | 
			
		||||
assert len(log_per_trace) == 4
 | 
			
		||||
 | 
			
		||||
| 
						 | 
				
			
			@ -27,5 +27,5 @@ assert re.search(r"Assert failed in test: %s \(.*\)$" % step_5, log_per_trace[2]
 | 
			
		|||
assert re.search(r"Assert failed in test: %s \(.*\) \[failed before\]$" % step_3_7, log_per_trace[3], re.M)
 | 
			
		||||
assert re.search(r"Assert failed in test: %s \(.*\)$" % step_7, log_per_trace[3], re.M)
 | 
			
		||||
 | 
			
		||||
pattern = f"Property ASSERT in test at {assert_0} failed. Trace file: engine_0/trace0.vcd"
 | 
			
		||||
pattern = f"Property ASSERT in test at {assert_0} failed. Trace file: engine_0/trace0.(vcd|fst)"
 | 
			
		||||
assert re.search(pattern, open(f"{workdir}/{workdir}.xml").read())
 | 
			
		||||
| 
						 | 
				
			
			
 | 
			
		|||
| 
						 | 
				
			
			@ -9,7 +9,7 @@ assert_not_a = line_ref(workdir, src, "assert(!a)")
 | 
			
		|||
assert_0 = line_ref(workdir, src, "assert(0)")
 | 
			
		||||
 | 
			
		||||
log = open(workdir + "/logfile.txt").read()
 | 
			
		||||
log_per_trace = log.split("Writing trace to VCD file")[:-1]
 | 
			
		||||
log_per_trace = log.split("Writing trace to Yosys witness file")[:-1]
 | 
			
		||||
 | 
			
		||||
assert len(log_per_trace) == 2
 | 
			
		||||
 | 
			
		||||
| 
						 | 
				
			
			
 | 
			
		|||
| 
						 | 
				
			
			@ -12,7 +12,7 @@ assert_false = line_ref(workdir, "extra.smtc", "assert false")
 | 
			
		|||
assert_distinct = line_ref(workdir, "extra.smtc", "assert (distinct")
 | 
			
		||||
 | 
			
		||||
log = open(workdir + "/logfile.txt").read()
 | 
			
		||||
log_per_trace = log.split("Writing trace to VCD file")[:-1]
 | 
			
		||||
log_per_trace = log.split("Writing trace to Yosys witness file")[:-1]
 | 
			
		||||
 | 
			
		||||
assert len(log_per_trace) == 4
 | 
			
		||||
 | 
			
		||||
| 
						 | 
				
			
			
 | 
			
		|||
							
								
								
									
										48
									
								
								tests/unsorted/btor_meminit.sby
									
										
									
									
									
										Normal file
									
								
							
							
						
						
									
										48
									
								
								tests/unsorted/btor_meminit.sby
									
										
									
									
									
										Normal file
									
								
							| 
						 | 
				
			
			@ -0,0 +1,48 @@
 | 
			
		|||
[tasks]
 | 
			
		||||
btormc
 | 
			
		||||
#pono
 | 
			
		||||
smtbmc
 | 
			
		||||
 | 
			
		||||
[options]
 | 
			
		||||
mode bmc
 | 
			
		||||
expect fail
 | 
			
		||||
 | 
			
		||||
[engines]
 | 
			
		||||
btormc: btor btormc
 | 
			
		||||
# pono: btor pono
 | 
			
		||||
smtbmc: smtbmc
 | 
			
		||||
 | 
			
		||||
[script]
 | 
			
		||||
read -formal top.sv
 | 
			
		||||
prep -top top -flatten
 | 
			
		||||
 | 
			
		||||
[file top.sv]
 | 
			
		||||
 | 
			
		||||
module top(input clk);
 | 
			
		||||
 | 
			
		||||
    inner inner(clk);
 | 
			
		||||
 | 
			
		||||
endmodule
 | 
			
		||||
 | 
			
		||||
module inner(input clk);
 | 
			
		||||
    reg [7:0] counter = 0;
 | 
			
		||||
 | 
			
		||||
    reg [1:0] mem [0:255];
 | 
			
		||||
 | 
			
		||||
    initial begin
 | 
			
		||||
        mem[0] = 0;
 | 
			
		||||
        mem[1] = 1;
 | 
			
		||||
        mem[2] = 2;
 | 
			
		||||
        mem[3] = 2;
 | 
			
		||||
        mem[4] = 0;
 | 
			
		||||
        mem[7] = 0;
 | 
			
		||||
    end
 | 
			
		||||
 | 
			
		||||
    always @(posedge clk) begin
 | 
			
		||||
        counter <= counter + 1;
 | 
			
		||||
        foo: assert (mem[counter] < 3);
 | 
			
		||||
        bar: assume (counter < 7);
 | 
			
		||||
 | 
			
		||||
        mem[counter] <= 0;
 | 
			
		||||
    end
 | 
			
		||||
endmodule
 | 
			
		||||
							
								
								
									
										34
									
								
								tests/unsorted/cover_unreachable.sby
									
										
									
									
									
										Normal file
									
								
							
							
						
						
									
										34
									
								
								tests/unsorted/cover_unreachable.sby
									
										
									
									
									
										Normal file
									
								
							| 
						 | 
				
			
			@ -0,0 +1,34 @@
 | 
			
		|||
[tasks]
 | 
			
		||||
btormc
 | 
			
		||||
smtbmc
 | 
			
		||||
 | 
			
		||||
[options]
 | 
			
		||||
mode cover
 | 
			
		||||
expect fail
 | 
			
		||||
 | 
			
		||||
[engines]
 | 
			
		||||
btormc: btor btormc
 | 
			
		||||
smtbmc: smtbmc
 | 
			
		||||
 | 
			
		||||
[script]
 | 
			
		||||
read -formal top.sv
 | 
			
		||||
prep -top top -flatten
 | 
			
		||||
 | 
			
		||||
[file top.sv]
 | 
			
		||||
 | 
			
		||||
module top(input clk);
 | 
			
		||||
 | 
			
		||||
    inner inner(clk);
 | 
			
		||||
 | 
			
		||||
endmodule
 | 
			
		||||
 | 
			
		||||
module inner(input clk);
 | 
			
		||||
    reg [7:0] counter = 0;
 | 
			
		||||
 | 
			
		||||
    always @(posedge clk) begin
 | 
			
		||||
        counter <= counter == 4 ? 0 : counter + 1;
 | 
			
		||||
 | 
			
		||||
        reachable: cover (counter == 3);
 | 
			
		||||
        unreachable: cover (counter == 5);
 | 
			
		||||
    end
 | 
			
		||||
endmodule
 | 
			
		||||
		Loading…
	
	Add table
		Add a link
		
	
		Reference in a new issue