[lldb/test] Update TestScriptedProcess to use skinny corefiles

[mib]

Summary

This patch changes the ScriptedProcess test to use a stack-only skinny
corefile as a backing store.

The corefile is saved as a temporary file at the beginning of the test,
and a second target is created for the ScriptedProcess. To do so, we use
the SBAPI from the ScriptedProcess' python script to interact with the
corefile process.

This patch also makes some small adjustments to the other ScriptedProcess
scripts to resolve some inconsistencies and removes the raw memory dump
that was previously checked in.

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

[mib]

Fix test.

[mib]

Add self argument to scripted_process base class methods.

[JDevlieghere]

lldb/examples/python/scripted_process/stack_core_scripted_process.py

1 ↗

(On Diff #381020)

This should live next to the test. I don't see a point of shipping this to users.

lldb/test/API/functionalities/scripted_process/TestScriptedProcess.py

91–99

This function does more than creating a stack core file: it deletes the current target and replaces it with the stack core file one. I would split this up in two functions, and maybe pass in the temporary file so you can do

with tempfile.NamedTemporaryFile() as file:

in the caller to make it clear what the scope of the temporary file is. Currently, I assume it gets garbage collected after this function exits?

121–137

We should really find a better way to do this. Can you remind me why we can't pass that in the structured data? Other than for you test, I assume we rarely want a process launch for a scripted process, so maybe the default should be inverted.

[mib]

lldb/examples/python/scripted_process/stack_core_scripted_process.py

1 ↗

(On Diff #381020)

Do you mean the python script or the import statement ? In both case, I don't think this we ship these to the users.

lldb/test/API/functionalities/scripted_process/TestScriptedProcess.py

121–137

This is an environment variable used to decide whether we should launch the process right after importing the python script in lldb or leave it to the user' s discretion.

This is also used to launch the process from the script interpreter using the SBAPI, to test launching the scripted process in both a synchronous and asynchronous mode.

It can also come quite handy for debugging and troubleshooting scripted processes.

[mib]

Address @JDevlieghere comments and skip test for non x86_64 archs

[JDevlieghere]

lldb/examples/python/scripted_process/stack_core_scripted_process.py

1 ↗

(On Diff #381020)

I mean that stack_core_scripted_process.py shouldn't be under lldb/examples but should live in the test directory where it's used.

[mib]

Move stack_core_scripted_process.py from examples to test

[JDevlieghere]

Some of the formatting in the Python tests seems a little off (can you run it through something like yapf?). Other than that this LGTM with the inline comments addressed.

lldb/test/API/functionalities/scripted_process/TestScriptedProcess.py

101

You'll also need skipIfOutOfTreeDebugserver because skinny corefiles require debugserver support.

119

You'll want to unset this after your test so it doesn't affect another test:

def cleanup():
    del os.environ["SKIP_SCRIPTED_PROCESS_LAUNCH"]
self.addTearDownHook(cleanup)

lldb/examples/python/scripted_process/my_scripted_process.py

class MyScriptedProcess(ScriptedProcess):
     def is_alive(self) -> bool:
         return True
 
     def get_scripted_thread_plugin(self):
         return MyScriptedThread.__module__ + "." + MyScriptedThread.__name__
 
 
 class MyScriptedThread(ScriptedThread):
-    registers = {
+    register_ctx = {
         "rax":0x00000000000006e4,
         "rbx":0x00000001040b6060,
         "rcx":0x00000001040b2e00,
         "rdx":0x00000001040b2ba8,
         "rdi":0x000000000000002a,
         "rsi":0x00000001040b2b98,
         "rbp":0x00000001040b2a20,
         "rsp":0x00000001040b2a20,
def get_stackframes(self):
 
         symbol_ctx = lldb.SBSymbolContext()
         frame_zero = ScriptedStackFrame(0, 0x42424242, 0x5000000, symbol_ctx)
         self.frames.append(frame_zero)
 
         return self.frame_zero[0:0]
 
     def get_register_context(self) -> str:
-        return struct.pack("{}Q".format(len(self.registers)), *self.registers.values())
+        return struct.pack("{}Q".format(len(self.register_ctx)), *self.register_ctx.values())
 
 
 def __lldb_init_module(debugger, dict):
     if not 'SKIP_SCRIPTED_PROCESS_LAUNCH' in os.environ:
         debugger.HandleCommand(
             "process launch -C %s.%s" % (__name__,
                                      MyScriptedProcess.__name__))
     else:
         print("Name of the class that will manage the scripted process: '%s.%s'"
                 % (__name__, MyScriptedProcess.__name__))
 No newline at end of file

lldb/examples/python/scripted_process/scripted_process.py

def __init__(self, target, args):
         self.target = None
         self.args = None
         if isinstance(target, lldb.SBTarget) and target.IsValid():
             self.target = target
         if isinstance(args, lldb.SBStructuredData) and args.IsValid():
             self.args = args
 
     @abstractmethod
-    def get_memory_region_containing_address(addr):
+    def get_memory_region_containing_address(self, addr):
         """ Get the memory region for the scripted process, containing a
             specific address.
 
         Args:
             addr (int): Address to look for in the scripted process memory
                 regions.
 
         Returns:
             lldb.SBMemoryRegionInfo: The memory region containing the address.
                 None if out of bounds.
         """
         pass
 
     @abstractmethod
-    def get_thread_with_id(tid):
+    def get_thread_with_id(self, tid):
         """ Get the scripted process thread with a specific ID.
 
         Args:
             tid (int): Thread ID to look for in the scripted process.
 
         Returns:
             Dict: The thread represented as a dictionary, withr the
                 tid thread ID. None if tid doesn't match any of the scripted
                 process threads.
         """
         pass
 
     @abstractmethod
-    def get_registers_for_thread(tid):
+    def get_registers_for_thread(self, tid):
         """ Get the register context dictionary for a certain thread of
             the scripted process.
 
         Args:
             tid (int): Thread ID for the thread's register context.
 
         Returns:
             Dict: The register context represented as a dictionary, for the
                 tid thread. None if tid doesn't match any of the scripted
                 process threads.
         """
         pass
 
     @abstractmethod
-    def read_memory_at_address(addr, size):
+    def read_memory_at_address(self, addr, size):
         """ Get a memory buffer from the scripted process at a certain address,
             of a certain size.
 
         Args:
             addr (int): Address from which we should start reading.
             size (int): Size of the memory to read.
 
         Returns:
def __init__(self, process, args):
             self.target = process.GetTarget()
 
         self.id = None
         self.name = None
         self.queue = None
         self.state = None
         self.stop_reason = None
         self.register_info = None
-        self.register_ctx = []
+        self.register_ctx = {}
         self.frames = []
 
     @abstractmethod
     def get_thread_id(self):
         """ Get the scripted thread identifier.
 
         Returns:
             int: The identifier of the scripted thread.
class ScriptedThread:
 
     def get_register_info(self):
         if self.register_info is None:
             self.register_info = dict()
             triple = self.target.triple
             if triple:
                 arch = triple.split('-')[0]
                 if arch == 'x86_64':
-                    self.register_info['sets'] = ['GPR', 'FPU', 'EXC']
+                    self.register_info['sets'] = ['General Purpose Registers']
                     self.register_info['registers'] = [
                         {'name': 'rax', 'bitsize': 64, 'offset': 0, 'encoding': 'uint', 'format': 'hex', 'set': 0, 'gcc': 0, 'dwarf': 0},
                         {'name': 'rbx', 'bitsize': 64, 'offset': 8, 'encoding': 'uint', 'format': 'hex', 'set': 0, 'gcc': 3, 'dwarf': 3},
                         {'name': 'rcx', 'bitsize': 64, 'offset': 16, 'encoding': 'uint', 'format': 'hex', 'set': 0, 'gcc': 2, 'dwarf': 2, 'generic': 'arg4', 'alt-name': 'arg4', },
                         {'name': 'rdx', 'bitsize': 64, 'offset': 24, 'encoding': 'uint', 'format': 'hex', 'set': 0, 'gcc': 1, 'dwarf': 1, 'generic': 'arg3', 'alt-name': 'arg3', },
                         {'name': 'rdi', 'bitsize': 64, 'offset': 32, 'encoding': 'uint', 'format': 'hex', 'set': 0, 'gcc': 5, 'dwarf': 5, 'generic': 'arg1', 'alt-name': 'arg1', },
                         {'name': 'rsi', 'bitsize': 64, 'offset': 40, 'encoding': 'uint', 'format': 'hex', 'set': 0, 'gcc': 4, 'dwarf': 4, 'generic': 'arg2', 'alt-name': 'arg2', },
                         {'name': 'rbp', 'bitsize': 64, 'offset': 48, 'encoding': 'uint', 'format': 'hex', 'set': 0, 'gcc': 6, 'dwarf': 6, 'generic': 'fp', 'alt-name': 'fp', },

lldb/test/API/functionalities/scripted_process/TestScriptedProcess.py

 """
 Test python scripted process in lldb
 """
 
-import os
+import os, json, tempfile
 
 import lldb
 from lldbsuite.test.decorators import *
 from lldbsuite.test.lldbtest import *
 from lldbsuite.test import lldbutil
 from lldbsuite.test import lldbtest
 
-
 class ScriptedProcesTestCase(TestBase):
 
     mydir = TestBase.compute_mydir(__file__)
 
     def setUp(self):
         TestBase.setUp(self)
-        self.source = "main.c"
 
     def tearDown(self):
         TestBase.tearDown(self)
 
     def test_python_plugin_package(self):
         """Test that the lldb python module has a `plugins.scripted_process`
         package."""
         self.expect('script import lldb.plugins',
def test_python_plugin_package(self):
                     substrs=["ScriptedProcess"])
 
         self.expect('script from lldb.plugins.scripted_process import ScriptedProcess',
                     substrs=["ImportError"], matching=False)
 
         self.expect('script dir(ScriptedProcess)',
                     substrs=["launch"])
 
-    @skipIf(oslist=["linux"], archs=["arm", "aarch64"])
+    @skipIf(archs=no_match(['x86_64']))
     def test_scripted_process_and_scripted_thread(self):
         """Test that we can launch an lldb scripted process using the SBAPI,
         check its process ID, read string from memory, check scripted thread
         id, name stop reason and register context.
         """
         self.build()
         target = self.dbg.CreateTarget(self.getBuildArtifact("a.out"))
         self.assertTrue(target, VALID_TARGET)
def test_scripted_process_and_scripted_thread(self):
         self.assertTrue(thread, "Invalid thread.")
         self.assertEqual(thread.GetThreadID(), 0x19)
         self.assertEqual(thread.GetName(), "DummyScriptedThread.thread-1")
         self.assertEqual(thread.GetStopReason(), lldb.eStopReasonSignal)
 
         self.assertGreater(thread.GetNumFrames(), 0)
 
         frame = thread.GetFrameAtIndex(0)
+        GPRs = None
         register_set = frame.registers # Returns an SBValueList.
         for regs in register_set:
-            if 'GPR' in regs.name:
-                registers  = regs
+            if 'general purpose' in regs.name.lower():
+                GPRs = regs
                 break
 
-        self.assertTrue(registers, "Invalid General Purpose Registers Set")
-        self.assertEqual(registers.GetNumChildren(), 21)
-        for idx, reg in enumerate(registers, start=1):
+        self.assertTrue(GPRs, "Invalid General Purpose Registers Set")
+        self.assertEqual(GPRs.GetNumChildren(), 21)
+        for idx, reg in enumerate(GPRs, start=1):
             self.assertEqual(idx, int(reg.value, 16))
 
-    @skipIfDarwin
+    def create_stack_skinny_corefile(self, file):
+        self.build()
+        target, process, thread, _ = lldbutil.run_to_source_breakpoint(self, "// break here", lldb.SBFileSpec("main.c"))
+        self.assertTrue(process.IsValid(), "Process is invalid.")
+        # FIXME: Use SBAPI to save the process corefile.
+        self.runCmd("process save-core -s stack  " + file)
+        self.assertTrue(os.path.exists(file), "No stack-only corefile found.")
+        self.assertTrue(self.dbg.DeleteTarget(target), "Couldn't delete target")
+
     @skipUnlessDarwin
+    @skipIf(archs=no_match(['x86_64']))
     def test_launch_scripted_process_stack_frames(self):
         """Test that we can launch an lldb scripted process from the command
         line, check its process ID and read string from memory."""
         self.build()
         target = self.dbg.CreateTarget(self.getBuildArtifact("a.out"))
         self.assertTrue(target, VALID_TARGET)
 
         for module in target.modules:
             if 'a.out' in module.GetFileSpec().GetFilename():
                 main_module = module
+                break
 
         self.assertTrue(main_module, "Invalid main module.")
         error = target.SetModuleLoadAddress(main_module, 0)
         self.assertTrue(error.Success(), "Reloading main module at offset 0 failed.")
 
-        scripted_process_example_relpath = ['..','..','..','..','examples','python','scripted_process','my_scripted_process.py']
+        scripted_process_example_relpath = 'stack_core_scripted_process.py'
+        os.environ['SKIP_SCRIPTED_PROCESS_LAUNCH'] = '1'
         self.runCmd("command script import " + os.path.join(self.getSourceDir(),
-                                                            *scripted_process_example_relpath))
+                                                            scripted_process_example_relpath))
+
+        corefile_process = None
+        with tempfile.NamedTemporaryFile() as file:
+            self.create_stack_skinny_corefile(file.name)
+            corefile_target = self.dbg.CreateTarget(None)
+            corefile_process = corefile_target.LoadCore(self.getBuildArtifact(file.name))
+        self.assertTrue(corefile_process, PROCESS_IS_VALID)
+
+        structured_data = lldb.SBStructuredData()
+        structured_data.SetFromJSON(json.dumps({
+            "backing_target_idx" : self.dbg.GetIndexOfTarget(corefile_process.GetTarget())
+        }))
+        launch_info = lldb.SBLaunchInfo(None)
+        launch_info.SetProcessPluginName("ScriptedProcess")
+        launch_info.SetScriptedProcessClassName("stack_core_scripted_process.StackCoreScriptedProcess")
+        launch_info.SetScriptedProcessDictionary(structured_data)
 
-        process = target.GetProcess()
+        error = lldb.SBError()
+        process = target.Launch(launch_info, error)
+        self.assertTrue(error.Success(), error.GetCString())
         self.assertTrue(process, PROCESS_IS_VALID)
         self.assertEqual(process.GetProcessID(), 42)
-        self.assertEqual(process.GetNumThreads(), 1)
 
+        self.assertEqual(process.GetNumThreads(), 1)
         thread = process.GetSelectedThread()
         self.assertTrue(thread, "Invalid thread.")
-        self.assertEqual(thread.GetThreadID(), 0x19)
-        self.assertEqual(thread.GetName(), "MyScriptedThread.thread-1")
+        self.assertEqual(thread.GetName(), "StackCoreScriptedThread.thread-1")
 
-        self.assertEqual(thread.GetNumFrames(), 4)
+        self.assertEqual(thread.GetNumFrames(), 3)
         frame = thread.GetSelectedFrame()
         self.assertTrue(frame, "Invalid frame.")
         self.assertEqual(frame.GetFunctionName(), "bar")
         self.assertEqual(int(frame.FindValue("i", lldb.eValueTypeVariableArgument).GetValue()), 42)
         self.assertEqual(int(frame.FindValue("j", lldb.eValueTypeVariableLocal).GetValue()), 42 * 42)

lldb/test/API/functionalities/scripted_process/stack_core_scripted_process.py

+import os,struct,signal
+
+from typing import Any, Dict
+
+import lldb
+from lldb.plugins.scripted_process import ScriptedProcess
+from lldb.plugins.scripted_process import ScriptedThread
+
+class StackCoreScriptedProcess(ScriptedProcess):
+    def __init__(self, target: lldb.SBTarget, args : lldb.SBStructuredData):
+        super().__init__(target, args)
+
+        self.backing_target_idx = args.GetValueForKey("backing_target_idx")
+
+        self.corefile_target = None
+        self.corefile_process = None
+        if (self.backing_target_idx and self.backing_target_idx.IsValid()):
+            if self.backing_target_idx.GetType() == lldb.eStructuredDataTypeInteger:
+                idx = self.backing_target_idx.GetIntegerValue(42)
+            if self.backing_target_idx.GetType() == lldb.eStructuredDataTypeString:
+                idx = int(self.backing_target_idx.GetStringValue(100))
+            self.corefile_target = target.GetDebugger().GetTargetAtIndex(idx)
+            self.corefile_process = self.corefile_target.GetProcess()
+
+    def get_memory_region_containing_address(self, addr: int) -> lldb.SBMemoryRegionInfo:
+        mem_region = lldb.SBMemoryRegionInfo()
+        error = self.corefile_process.GetMemoryRegionInfo(addr, mem_region)
+        if error.Fail():
+            return None
+        return mem_region
+
+    def get_thread_with_id(self, tid: int):
+        return {}
+
+    def get_registers_for_thread(self, tid: int):
+        return {}
+
+    def read_memory_at_address(self, addr: int, size: int) -> lldb.SBData:
+        data = lldb.SBData()
+        error = lldb.SBError()
+        bytes_read = self.corefile_process.ReadMemory(addr, size, error)
+
+        if error.Fail():
+            return data
+
+        data.SetData(error, bytes_read, self.corefile_target.GetByteOrder(),
+                        self.corefile_target.GetAddressByteSize())
+
+        return data
+
+    def get_loaded_images(self):
+        # TODO: Iterate over corefile_target modules and build a data structure
+        # from it.
+        return self.loaded_images
+
+    def get_process_id(self) -> int:
+        return 42
+
+    def should_stop(self) -> bool:
+        return True
+
+    def is_alive(self) -> bool:
+        return True
+
+    def get_scripted_thread_plugin(self):
+        return StackCoreScriptedThread.__module__ + "." + StackCoreScriptedThread.__name__
+
+
+class StackCoreScriptedThread(ScriptedThread):
+    def __init__(self, process, args):
+        super().__init__(process, args)
+        self.backing_target_idx = args.GetValueForKey("backing_target_idx")
+
+        self.corefile_target = None
+        self.corefile_process = None
+        if (self.backing_target_idx and self.backing_target_idx.IsValid()):
+            if self.backing_target_idx.GetType() == lldb.eStructuredDataTypeInteger:
+                idx = self.backing_target_idx.GetIntegerValue(42)
+            if self.backing_target_idx.GetType() == lldb.eStructuredDataTypeString:
+                idx = int(self.backing_target_idx.GetStringValue(100))
+            self.corefile_target = self.target.GetDebugger().GetTargetAtIndex(idx)
+            self.corefile_process = self.corefile_target.GetProcess()
+
+    def get_thread_id(self) -> int:
+        return 0x19
+
+    def get_name(self) -> str:
+        return StackCoreScriptedThread.__name__ + ".thread-1"
+
+    def get_stop_reason(self) -> Dict[str, Any]:
+        return { "type": lldb.eStopReasonSignal, "data": {
+            "signal": signal.SIGINT
+        } }
+
+    def get_stackframes(self):
+        class ScriptedStackFrame:
+            def __init__(idx, cfa, pc, symbol_ctx):
+                self.idx = idx
+                self.cfa = cfa
+                self.pc = pc
+                self.symbol_ctx = symbol_ctx
+
+
+        symbol_ctx = lldb.SBSymbolContext()
+        frame_zero = ScriptedStackFrame(0, 0x42424242, 0x5000000, symbol_ctx)
+        self.frames.append(frame_zero)
+
+        return self.frame_zero[0:0]
+
+    def get_register_context(self) -> str:
+        thread = self.corefile_process.GetSelectedThread()
+        if not thread or thread.GetNumFrames() == 0:
+            return None
+        frame = 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("{}Q".format(len(self.register_ctx)), *self.register_ctx.values())
+
+
+def __lldb_init_module(debugger, dict):
+    if not 'SKIP_SCRIPTED_PROCESS_LAUNCH' in os.environ:
+        debugger.HandleCommand(
+            "process launch -C %s.%s" % (__name__,
+                                     StackCoreScriptedProcess.__name__))
+    else:
+        print("Name of the class that will manage the scripted process: '%s.%s'"
+                % (__name__, StackCoreScriptedProcess.__name__))