[lldb] Move PassthroughScriptedProcess to `lldb.scripted_process` module

[mib]

Summary

This patch moves the PassthroughScriptedProcess & PassthroughScriptedThread
classes from the interactive_scripted_process.py test implementation
to the lldb.scripted_process python module.

This class is very versatile so it makes more sense to ship it with the
python module to make it easier for our adopters to derive their class
from it instead of copying it.

During the "migration", I've also noticed some bugs in the
PassthroughScriptedThread creation and update, so I also fixed that as
part of this patch.

Signed-off-by: Med Ismail Bennani <[e-mail]>

[bulbazord]

Looks like you're mostly moving things around which is fine. The bug that you're fixing is with thread handling in MultiplexerScriptedProcess right? That looks fine to me too.

[mib]

In D151044#4362084, @bulbazord wrote:

The bug that you're fixing is with thread handling in MultiplexerScriptedProcess right? That looks fine to me too.

Yep! Thanks!

lldb/examples/python/scripted_process/scripted_process.py

 from abc import ABCMeta, abstractmethod
 
 import lldb
+import json, struct, signal
 
 class ScriptedProcess(metaclass=ABCMeta):
 
     """
     The base class for a scripted process.
 
     Most of the base class methods are `@abstractmethod` that need to be
     overwritten by the inheriting class.
def create_breakpoint(self, addr, error):
         Returns:
             SBBreakpoint: A valid breakpoint object that was created a the specified
                           address. None if the breakpoint creation failed.
         """
         error.SetErrorString("%s doesn't support creating breakpoints."
                              % self.__class__.__name__)
         return False
 
-
 class ScriptedThread(metaclass=ABCMeta):
 
     """
     The base class for a scripted thread.
 
     Most of the base class methods are `@abstractmethod` that need to be
     overwritten by the inheriting class.
 
def get_extended_info(self):
         """ Get scripted thread extended information.
 
         Returns:
             List: A list containing the extended information for the scripted process.
                   None if the thread as no extended information.
         """
         return self.extended_info
 
+
+class PassthroughScriptedProcess(ScriptedProcess):
+    driving_target = None
+    driving_process = None
+
+    def __init__(self, exe_ctx, args, launched_driving_process=True):
+        super().__init__(exe_ctx, args)
+
+        self.driving_target = None
+        self.driving_process = None
+
+        self.driving_target_idx = args.GetValueForKey("driving_target_idx")
+        if self.driving_target_idx and self.driving_target_idx.IsValid():
+            idx = self.driving_target_idx.GetUnsignedIntegerValue(42)
+            self.driving_target = self.target.GetDebugger().GetTargetAtIndex(idx)
+
+            if launched_driving_process:
+                self.driving_process = self.driving_target.GetProcess()
+                for driving_thread in self.driving_process:
+                    structured_data = lldb.SBStructuredData()
+                    structured_data.SetFromJSON(
+                        json.dumps(
+                            {
+                                "driving_target_idx": idx,
+                                "thread_idx": driving_thread.GetIndexID(),
+                            }
+                        )
+                    )
+
+                    self.threads[
+                        driving_thread.GetThreadID()
+                    ] = PassthroughScriptedThread(self, structured_data)
+
+                for module in self.driving_target.modules:
+                    path = module.file.fullpath
+                    load_addr = module.GetObjectFileHeaderAddress().GetLoadAddress(
+                        self.driving_target
+                    )
+                    self.loaded_images.append({"path": path, "load_addr": load_addr})
+
+    def get_memory_region_containing_address(self, addr):
+        mem_region = lldb.SBMemoryRegionInfo()
+        error = self.driving_process.GetMemoryRegionInfo(addr, mem_region)
+        if error.Fail():
+            return None
+        return mem_region
+
+    def read_memory_at_address(self, addr, size, error):
+        data = lldb.SBData()
+        bytes_read = self.driving_process.ReadMemory(addr, size, error)
+
+        if error.Fail():
+            return data
+
+        data.SetDataWithOwnership(
+            error,
+            bytes_read,
+            self.driving_target.GetByteOrder(),
+            self.driving_target.GetAddressByteSize(),
+        )
+
+        return data
+
+    def write_memory_at_address(self, addr, data, error):
+        return self.driving_process.WriteMemory(
+            addr, bytearray(data.uint8.all()), error
+        )
+
+    def get_process_id(self):
+        return self.driving_process.GetProcessID()
+
+    def is_alive(self):
+        return True
+
+    def get_scripted_thread_plugin(self):
+        return f"{PassthroughScriptedThread.__module__}.{PassthroughScriptedThread.__name__}"
+
+
+class PassthroughScriptedThread(ScriptedThread):
+    def __init__(self, process, args):
+        super().__init__(process, args)
+        driving_target_idx = args.GetValueForKey("driving_target_idx")
+        thread_idx = args.GetValueForKey("thread_idx")
+
+        # TODO: Change to Walrus operator (:=) with oneline if assignment
+        # Requires python 3.8
+        val = thread_idx.GetUnsignedIntegerValue()
+        if val is not None:
+            self.idx = val
+
+        self.driving_target = None
+        self.driving_process = None
+        self.driving_thread = None
+
+        # TODO: Change to Walrus operator (:=) with oneline if assignment
+        # Requires python 3.8
+        val = driving_target_idx.GetUnsignedIntegerValue()
+        if val is not None:
+            self.driving_target = self.target.GetDebugger().GetTargetAtIndex(val)
+            self.driving_process = self.driving_target.GetProcess()
+            self.driving_thread = self.driving_process.GetThreadByIndexID(self.idx)
+
+        if self.driving_thread:
+            self.id = self.driving_thread.GetThreadID()
+
+    def get_thread_id(self):
+        return self.id
+
+    def get_name(self):
+        return f"{PassthroughScriptedThread.__name__}.thread-{self.idx}"
+
+    def get_stop_reason(self):
+        stop_reason = {"type": lldb.eStopReasonInvalid, "data": {}}
+
+        if (
+            self.driving_thread
+            and self.driving_thread.IsValid()
+            and self.get_thread_id() == self.driving_thread.GetThreadID()
+        ):
+            stop_reason["type"] = lldb.eStopReasonNone
+
+            # TODO: Passthrough stop reason from driving process
+            if self.driving_thread.GetStopReason() != lldb.eStopReasonNone:
+                if "arm64" in self.scripted_process.arch:
+                    stop_reason["type"] = lldb.eStopReasonException
+                    stop_reason["data"][
+                        "desc"
+                    ] = self.driving_thread.GetStopDescription(100)
+                elif self.scripted_process.arch == "x86_64":
+                    stop_reason["type"] = lldb.eStopReasonSignal
+                    stop_reason["data"]["signal"] = signal.SIGTRAP
+                else:
+                    stop_reason["type"] = self.driving_thread.GetStopReason()
+
+        return stop_reason
+
+    def get_register_context(self):
+        if not self.driving_thread or self.driving_thread.GetNumFrames() == 0:
+            return None
+        frame = self.driving_thread.GetFrameAtIndex(0)
+
+        GPRs = None
+        registerSet = frame.registers  # Returns an SBValueList.
+        for regs in registerSet:
+            if "general purpose" in regs.name.lower():
+                GPRs = regs
+                break
+
+        if not GPRs:
+            return None
+
+        for reg in GPRs:
+            self.register_ctx[reg.name] = int(reg.value, base=16)
+
+        return struct.pack(f"{len(self.register_ctx)}Q", *self.register_ctx.values())
+
 ARM64_GPR = [ {'name': 'x0',   'bitsize': 64, 'offset': 0,   'encoding': 'uint', 'format': 'hex', 'set': 0, 'gcc': 0,  'dwarf': 0,  'generic': 'arg0', 'alt-name': 'arg0'},
               {'name': 'x1',   'bitsize': 64, 'offset': 8,   'encoding': 'uint', 'format': 'hex', 'set': 0, 'gcc': 1,  'dwarf': 1,  'generic': 'arg1', 'alt-name': 'arg1'},
               {'name': 'x2',   'bitsize': 64, 'offset': 16,  'encoding': 'uint', 'format': 'hex', 'set': 0, 'gcc': 2,  'dwarf': 2,  'generic': 'arg2', 'alt-name': 'arg2'},
               {'name': 'x3',   'bitsize': 64, 'offset': 24,  'encoding': 'uint', 'format': 'hex', 'set': 0, 'gcc': 3,  'dwarf': 3,  'generic': 'arg3', 'alt-name': 'arg3'},
               {'name': 'x4',   'bitsize': 64, 'offset': 32,  'encoding': 'uint', 'format': 'hex', 'set': 0, 'gcc': 4,  'dwarf': 4,  'generic': 'arg4', 'alt-name': 'arg4'},
               {'name': 'x5',   'bitsize': 64, 'offset': 40,  'encoding': 'uint', 'format': 'hex', 'set': 0, 'gcc': 5,  'dwarf': 5,  'generic': 'arg5', 'alt-name': 'arg5'},
               {'name': 'x6',   'bitsize': 64, 'offset': 48,  'encoding': 'uint', 'format': 'hex', 'set': 0, 'gcc': 6,  'dwarf': 6,  'generic': 'arg6', 'alt-name': 'arg6'},
               {'name': 'x7',   'bitsize': 64, 'offset': 56,  'encoding': 'uint', 'format': 'hex', 'set': 0, 'gcc': 7,  'dwarf': 7,  'generic': 'arg7', 'alt-name': 'arg7'},

lldb/test/API/functionalities/interactive_scripted_process/interactive_scripted_process.py

 # Usage:
 # ./bin/lldb $LLVM/lldb/test/API/functionalities/interactive_scripted_process/main \
 #   -o "br set -p 'Break here'" \
 #   -o "command script import $LLVM/lldb/test/API/functionalities/interactive_scripted_process/interactive_scripted_process.py" \
 #   -o "create_mux" \
 #   -o "create_sub" \
 #   -o "br set -p 'also break here'" -o 'continue'
 
 import os, json, struct, signal, tempfile
 
 from threading import Thread
 from typing import Any, Dict
 
 import lldb
-from lldb.plugins.scripted_process import ScriptedProcess
-from lldb.plugins.scripted_process import ScriptedThread
+from lldb.plugins.scripted_process import PassthroughScriptedProcess
+from lldb.plugins.scripted_process import PassthroughScriptedThread
 
-
-class PassthruScriptedProcess(ScriptedProcess):
-    driving_target = None
-    driving_process = None
-
-    def __init__(
-        self,
-        exe_ctx: lldb.SBExecutionContext,
-        args: lldb.SBStructuredData,
-        launched_driving_process: bool = True,
-    ):
-        super().__init__(exe_ctx, args)
-
-        self.driving_target = None
-        self.driving_process = None
-
-        self.driving_target_idx = args.GetValueForKey("driving_target_idx")
-        if self.driving_target_idx and self.driving_target_idx.IsValid():
-            if self.driving_target_idx.GetType() == lldb.eStructuredDataTypeInteger:
-                idx = self.driving_target_idx.GetIntegerValue(42)
-            if self.driving_target_idx.GetType() == lldb.eStructuredDataTypeString:
-                idx = int(self.driving_target_idx.GetStringValue(100))
-            self.driving_target = self.target.GetDebugger().GetTargetAtIndex(idx)
-
-            if launched_driving_process:
-                self.driving_process = self.driving_target.GetProcess()
-                for driving_thread in self.driving_process:
-                    structured_data = lldb.SBStructuredData()
-                    structured_data.SetFromJSON(
-                        json.dumps(
-                            {
-                                "driving_target_idx": idx,
-                                "thread_idx": driving_thread.GetIndexID(),
-                            }
-                        )
-                    )
-
-                    self.threads[driving_thread.GetThreadID()] = PassthruScriptedThread(
-                        self, structured_data
-                    )
-
-                for module in self.driving_target.modules:
-                    path = module.file.fullpath
-                    load_addr = module.GetObjectFileHeaderAddress().GetLoadAddress(
-                        self.driving_target
-                    )
-                    self.loaded_images.append({"path": path, "load_addr": load_addr})
-
-    def get_memory_region_containing_address(
-        self, addr: int
-    ) -> lldb.SBMemoryRegionInfo:
-        mem_region = lldb.SBMemoryRegionInfo()
-        error = self.driving_process.GetMemoryRegionInfo(addr, mem_region)
-        if error.Fail():
-            return None
-        return mem_region
-
-    def read_memory_at_address(
-        self, addr: int, size: int, error: lldb.SBError
-    ) -> lldb.SBData:
-        data = lldb.SBData()
-        bytes_read = self.driving_process.ReadMemory(addr, size, error)
-
-        if error.Fail():
-            return data
-
-        data.SetDataWithOwnership(
-            error,
-            bytes_read,
-            self.driving_target.GetByteOrder(),
-            self.driving_target.GetAddressByteSize(),
-        )
-
-        return data
-
-    def write_memory_at_address(
-        self, addr: int, data: lldb.SBData, error: lldb.SBError
-    ) -> int:
-        return self.driving_process.WriteMemory(
-            addr, bytearray(data.uint8.all()), error
-        )
-
-    def get_process_id(self) -> int:
-        return 42
-
-    def is_alive(self) -> bool:
-        return True
-
-    def get_scripted_thread_plugin(self) -> str:
-        return f"{PassthruScriptedThread.__module__}.{PassthruScriptedThread.__name__}"
-
-
-class MultiplexedScriptedProcess(PassthruScriptedProcess):
+class MultiplexedScriptedProcess(PassthroughScriptedProcess):
     def __init__(self, exe_ctx: lldb.SBExecutionContext, args: lldb.SBStructuredData):
         super().__init__(exe_ctx, args)
         self.multiplexer = None
         if isinstance(self.driving_process, lldb.SBProcess) and self.driving_process:
             parity = args.GetValueForKey("parity")
             # TODO: Change to Walrus operator (:=) with oneline if assignment
             # Requires python 3.8
-            val = extract_value_from_structured_data(parity, 0)
+            val = parity.GetUnsignedIntegerValue()
             if val is not None:
                 self.parity = val
 
-            # Turn PassThruScriptedThread into MultiplexedScriptedThread
+            # Turn PassthroughScriptedThread into MultiplexedScriptedThread
             for thread in self.threads.values():
                 thread.__class__ = MultiplexedScriptedThread
 
     def get_process_id(self) -> int:
         return self.parity + 420
 
     def launch(self, should_stop: bool = True) -> lldb.SBError:
         self.first_launch = True
         return lldb.SBError()
 
     def resume(self, should_stop: bool) -> lldb.SBError:
         if self.first_launch:
             self.first_launch = False
             return super().resume()
         else:
             if not self.multiplexer:
                 error = lldb.SBError("Multiplexer is not set.")
                 return error
             return self.multiplexer.resume(should_stop)
 
     def get_threads_info(self) -> Dict[int, Any]:
         if not self.multiplexer:
             return super().get_threads_info()
         filtered_threads = self.multiplexer.get_threads_info(pid=self.get_process_id())
-        # Update the filtered thread class from PassthruScriptedThread to MultiplexedScriptedThread
+        # Update the filtered thread class from PassthroughScriptedThread to MultiplexedScriptedThread
         return dict(
             map(
                 lambda pair: (pair[0], MultiplexedScriptedThread(pair[1])),
                 filtered_threads.items(),
             )
         )
 
     def create_breakpoint(self, addr, error, pid=None):
         if not self.multiplexer:
             error.SetErrorString("Multiplexer is not set.")
         return self.multiplexer.create_breakpoint(addr, error, self.get_process_id())
 
     def get_scripted_thread_plugin(self) -> str:
         return f"{MultiplexedScriptedThread.__module__}.{MultiplexedScriptedThread.__name__}"
 
-
-class PassthruScriptedThread(ScriptedThread):
-    def __init__(self, process, args):
-        super().__init__(process, args)
-        driving_target_idx = args.GetValueForKey("driving_target_idx")
-        thread_idx = args.GetValueForKey("thread_idx")
-
-        # TODO: Change to Walrus operator (:=) with oneline if assignment
-        # Requires python 3.8
-        val = extract_value_from_structured_data(thread_idx, 0)
-        if val is not None:
-            self.idx = val
-
-        self.driving_target = None
-        self.driving_process = None
-        self.driving_thread = None
-
-        # TODO: Change to Walrus operator (:=) with oneline if assignment
-        # Requires python 3.8
-        val = extract_value_from_structured_data(driving_target_idx, 42)
-        if val is not None:
-            self.driving_target = self.target.GetDebugger().GetTargetAtIndex(val)
-            self.driving_process = self.driving_target.GetProcess()
-            self.driving_thread = self.driving_process.GetThreadByIndexID(self.idx)
-
-        if self.driving_thread:
-            self.id = self.driving_thread.GetThreadID()
-
-    def get_thread_id(self) -> int:
-        return self.id
-
-    def get_name(self) -> str:
-        return f"{PassthruScriptedThread.__name__}.thread-{self.idx}"
-
-    def get_stop_reason(self) -> Dict[str, Any]:
-        stop_reason = {"type": lldb.eStopReasonInvalid, "data": {}}
-
-        if (
-            self.driving_thread
-            and self.driving_thread.IsValid()
-            and self.get_thread_id() == self.driving_thread.GetThreadID()
-        ):
-            stop_reason["type"] = lldb.eStopReasonNone
-
-            if self.driving_thread.GetStopReason() != lldb.eStopReasonNone:
-                if "arm64" in self.scripted_process.arch:
-                    stop_reason["type"] = lldb.eStopReasonException
-                    stop_reason["data"][
-                        "desc"
-                    ] = self.driving_thread.GetStopDescription(100)
-                elif self.scripted_process.arch == "x86_64":
-                    stop_reason["type"] = lldb.eStopReasonSignal
-                    stop_reason["data"]["signal"] = signal.SIGTRAP
-                else:
-                    stop_reason["type"] = self.driving_thread.GetStopReason()
-
-        return stop_reason
-
-    def get_register_context(self) -> str:
-        if not self.driving_thread or self.driving_thread.GetNumFrames() == 0:
-            return None
-        frame = self.driving_thread.GetFrameAtIndex(0)
-
-        GPRs = None
-        registerSet = frame.registers  # Returns an SBValueList.
-        for regs in registerSet:
-            if "general purpose" in regs.name.lower():
-                GPRs = regs
-                break
-
-        if not GPRs:
-            return None
-
-        for reg in GPRs:
-            self.register_ctx[reg.name] = int(reg.value, base=16)
-
-        return struct.pack(f"{len(self.register_ctx)}Q", *self.register_ctx.values())
-
-
-class MultiplexedScriptedThread(PassthruScriptedThread):
+class MultiplexedScriptedThread(PassthroughScriptedThread):
     def get_name(self) -> str:
         parity = "Odd" if self.scripted_process.parity % 2 else "Even"
         return f"{parity}{MultiplexedScriptedThread.__name__}.thread-{self.idx}"
 
 
-class MultiplexerScriptedProcess(PassthruScriptedProcess):
+class MultiplexerScriptedProcess(PassthroughScriptedProcess):
     listener = None
     multiplexed_processes = None
 
     def wait_for_driving_process_to_stop(self):
         def handle_process_state_event():
             # Update multiplexer process
             log("Updating interactive scripted process threads")
             dbg = self.driving_target.GetDebugger()
-            log("Clearing interactive scripted process threads")
-            self.threads.clear()
+            new_driving_thread_ids = []
             for driving_thread in self.driving_process:
+                new_driving_thread_ids.append(driving_thread.id)
                 log(f"{len(self.threads)} New thread {hex(driving_thread.id)}")
                 structured_data = lldb.SBStructuredData()
                 structured_data.SetFromJSON(
                     json.dumps(
                         {
                             "driving_target_idx": dbg.GetIndexOfTarget(
                                 self.driving_target
                             ),
                             "thread_idx": driving_thread.GetIndexID(),
                         }
                     )
                 )
 
-                self.threads[driving_thread.GetThreadID()] = PassthruScriptedThread(
+                self.threads[driving_thread.id] = PassthroughScriptedThread(
                     self, structured_data
                 )
 
+            for thread_id in self.threads:
+                if thread_id not in new_driving_thread_ids:
+                    log(f"Removing old thread {hex(thread_id)}")
+                    del self.threads[thread_id]
+
+            print(f"New thread count: {len(self.threads)}")
+
             mux_process = self.target.GetProcess()
             mux_process.ForceScriptedState(lldb.eStateRunning)
             mux_process.ForceScriptedState(lldb.eStateStopped)
 
             for child_process in self.multiplexed_processes.values():
                 child_process.ForceScriptedState(lldb.eStateRunning)
                 child_process.ForceScriptedState(lldb.eStateStopped)
 
         event = lldb.SBEvent()
         while True:
+            if not self.driving_process:
+                continue
             if self.listener.WaitForEvent(1, event):
                 event_mask = event.GetType()
                 if event_mask & lldb.SBProcess.eBroadcastBitStateChanged:
                     state = lldb.SBProcess.GetStateFromEvent(event)
                     log(f"Received public process state event: {state}")
                     if state == lldb.eStateStopped:
                         # If it's a stop event, iterate over the driving process
                         # thread, looking for a breakpoint stop reason, if internal

 
 def duplicate_target(driving_target):
     error = lldb.SBError()
     exe = driving_target.executable.fullpath
     triple = driving_target.triple
     debugger = driving_target.GetDebugger()
     return debugger.CreateTargetWithFileAndTargetTriple(exe, triple)
 
-
-def extract_value_from_structured_data(data, default_val):
-    if data and data.IsValid():
-        if data.GetType() == lldb.eStructuredDataTypeInteger:
-            return data.GetIntegerValue(default_val)
-        if data.GetType() == lldb.eStructuredDataTypeString:
-            return int(data.GetStringValue(100))
-    return default_val
-
-
 def create_mux_process(debugger, command, exe_ctx, result, dict):
     if not debugger.GetNumTargets() > 0:
         return result.SetError(
             "Interactive scripted processes requires one non scripted process."
         )
 
     debugger.SetAsync(True)