[Feature] add breakpoint machanism on SOT

sot/__init__.py

@@ -1,5 +1,4 @@
+from .opcode_translator.breakpoint import add_breakpoint
 from .translate import symbolic_translate
 
-__all__ = [
-    "symbolic_translate",
-]
+__all__ = ["symbolic_translate", "add_breakpoint"]

sot/opcode_translator/breakpoint.py

@@ -0,0 +1,40 @@
+from dataclasses import dataclass
+
+from ..utils import Singleton, log
+
+# this file is a debug utils files for quick debug
+# >>> sot.add_breakpoint(file, line)
+# >>> sot.remove_breakpoint(file, line)
+
+
+@dataclass
+class Breakpoint:
+    file: str
+    line: int
+
+    def __hash__(self):
+        return hash((self.file, self.line))
+
+
+@Singleton
+class BreakpointManager:
+    def __init__(self):
+        self.breakpoints = set()
+
+    def add(self, file, line):
+        log(1, f"add breakpoint at {file}:{line}")
+        self.breakpoints.add(Breakpoint(file, line))
+
+    def rm(self, *args, **kwargs):
+        # interactive use, we use abbreviate
+        self.breakpoints()
+
+    def hit(self, file, line):
+        _breakpoint = Breakpoint(file, line)
+        if _breakpoint in self.breakpoints:
+            return True
+        return False
+
+
+def add_breakpoint(file, line):
+    BreakpointManager().add(file, line)

sot/opcode_translator/executor/opcode_executor.py

@@ -18,6 +18,7 @@
     log,
     log_do,
 )
+from ..breakpoint import BreakpointManager
 from ..instruction_utils import (
     Instruction,
     analysis_inputs,
@@ -576,10 +577,13 @@ def step(self, instr: Instruction):
             raise NotImplementException(
                 f"opcode: {instr.opname} is not supported."
             )
-        log(
-            3,
-            f"[Translate {self._name}]: (line {self._current_line:>3}) {instr.opname:<12} {instr.argval}, stack is {self._stack}\n",
-        )
+        log_message = f"[Translate {self._name}]: (line {self._current_line:>3}) {instr.opname:<12} {instr.argval}, stack is {self._stack}\n"
+        log(3, log_message)
+        code_file = self._code.co_filename
+        code_line = self._current_line
+        if BreakpointManager().hit(code_file, code_line):
+            print(log_message)
+            breakpoint()  # breakpoint for debug
         return getattr(self, instr.opname)(instr)  # run single step.
 
     def indexof(self, instr: Instruction):