2e7aa2ee34
Before the addition of the process "Shadow Listener" you could only have one Listener observing the Process Broadcaster. That was necessary because fetching the Process event is what switches the public process state, and for the execution control logic to be manageable you needed to keep other listeners from causing this to happen before the main process control engine was ready. Ismail added the notion of a "ShadowListener" - which allowed you ONE extra process listener. This patch inverts that setup by designating the first listener as primary - and giving it priority in fetching events. Differential Revision: https://reviews.llvm.org/D157556
301 lines
13 KiB
Python
301 lines
13 KiB
Python
"""
|
|
Test the functionality of interactive scripted processes
|
|
"""
|
|
|
|
import lldb
|
|
import lldbsuite.test.lldbutil as lldbutil
|
|
from lldbsuite.test.decorators import *
|
|
from lldbsuite.test.lldbtest import *
|
|
import json, os
|
|
|
|
|
|
class TestInteractiveScriptedProcess(TestBase):
|
|
NO_DEBUG_INFO_TESTCASE = True
|
|
|
|
def setUp(self):
|
|
# Call super's setUp().
|
|
TestBase.setUp(self)
|
|
# Build and load test program
|
|
self.build()
|
|
self.runCmd("file " + self.getBuildArtifact("a.out"), CURRENT_EXECUTABLE_SET)
|
|
self.main_source_file = lldb.SBFileSpec("main.cpp")
|
|
self.script_module = "interactive_scripted_process"
|
|
self.script_file = self.script_module + ".py"
|
|
|
|
# These tests are flakey and sometimes timeout. They work most of the time
|
|
# so the basic event flow is right, but somehow the handling is off.
|
|
@skipUnlessDarwin
|
|
@skipIfDarwin
|
|
def test_passthrough_launch(self):
|
|
"""Test a simple pass-through process launch"""
|
|
self.passthrough_launch()
|
|
|
|
lldbutil.run_break_set_by_source_regexp(self, "also break here")
|
|
self.assertEqual(self.mux_target.GetNumBreakpoints(), 2)
|
|
error = self.mux_process.Continue()
|
|
self.assertSuccess(error, "Resuming multiplexer scripted process")
|
|
self.assertTrue(self.mux_process.IsValid(), "Got a valid process")
|
|
|
|
event = lldbutil.fetch_next_event(
|
|
self, self.dbg.GetListener(), self.mux_process.GetBroadcaster(), timeout=20
|
|
)
|
|
self.assertState(lldb.SBProcess.GetStateFromEvent(event), lldb.eStateRunning)
|
|
event = lldbutil.fetch_next_event(
|
|
self, self.dbg.GetListener(), self.mux_process.GetBroadcaster()
|
|
)
|
|
self.assertState(lldb.SBProcess.GetStateFromEvent(event), lldb.eStateStopped)
|
|
|
|
event = lldbutil.fetch_next_event(
|
|
self, self.mux_process_listener, self.mux_process.GetBroadcaster()
|
|
)
|
|
self.assertState(lldb.SBProcess.GetStateFromEvent(event), lldb.eStateRunning)
|
|
event = lldbutil.fetch_next_event(
|
|
self, self.mux_process_listener, self.mux_process.GetBroadcaster()
|
|
)
|
|
self.assertState(lldb.SBProcess.GetStateFromEvent(event), lldb.eStateStopped)
|
|
|
|
@skipUnlessDarwin
|
|
@skipIfDarwin
|
|
def test_multiplexed_launch(self):
|
|
"""Test a multiple interactive scripted process debugging"""
|
|
self.passthrough_launch()
|
|
self.assertEqual(self.dbg.GetNumTargets(), 2)
|
|
|
|
driving_target = self.mux_process.GetScriptedImplementation().driving_target
|
|
self.assertTrue(driving_target.IsValid(), "Driving target is invalid")
|
|
|
|
# Create a target for the multiplexed even scripted process
|
|
even_target = self.duplicate_target(driving_target)
|
|
self.assertTrue(
|
|
even_target.IsValid(),
|
|
"Couldn't duplicate driving target to launch multiplexed even scripted process",
|
|
)
|
|
|
|
class_name = f"{self.script_module}.MultiplexedScriptedProcess"
|
|
dictionary = {"driving_target_idx": self.dbg.GetIndexOfTarget(self.mux_target)}
|
|
|
|
dictionary["parity"] = 0
|
|
muxed_launch_info = self.get_launch_info(class_name, dictionary)
|
|
|
|
# Launch Even Child Scripted Process
|
|
error = lldb.SBError()
|
|
even_process = even_target.Launch(muxed_launch_info, error)
|
|
self.assertTrue(
|
|
even_process, "Couldn't launch multiplexed even scripted process"
|
|
)
|
|
self.multiplex(even_process)
|
|
|
|
# Check that the even process started running
|
|
event = lldbutil.fetch_next_event(
|
|
self, self.dbg.GetListener(), even_process.GetBroadcaster()
|
|
)
|
|
self.assertState(lldb.SBProcess.GetStateFromEvent(event), lldb.eStateRunning)
|
|
# Check that the even process stopped
|
|
event = lldbutil.fetch_next_event(
|
|
self, self.dbg.GetListener(), even_process.GetBroadcaster()
|
|
)
|
|
self.assertState(lldb.SBProcess.GetStateFromEvent(event), lldb.eStateStopped)
|
|
|
|
self.assertTrue(even_process.IsValid(), "Got a valid process")
|
|
self.assertState(
|
|
even_process.GetState(), lldb.eStateStopped, "Process is stopped"
|
|
)
|
|
|
|
# Create a target for the multiplexed odd scripted process
|
|
odd_target = self.duplicate_target(driving_target)
|
|
self.assertTrue(
|
|
odd_target.IsValid(),
|
|
"Couldn't duplicate driving target to launch multiplexed odd scripted process",
|
|
)
|
|
|
|
dictionary["parity"] = 1
|
|
muxed_launch_info = self.get_launch_info(class_name, dictionary)
|
|
|
|
# Launch Odd Child Scripted Process
|
|
error = lldb.SBError()
|
|
odd_process = odd_target.Launch(muxed_launch_info, error)
|
|
self.assertTrue(odd_process, "Couldn't launch multiplexed odd scripted process")
|
|
self.multiplex(odd_process)
|
|
|
|
# Check that the odd process started running
|
|
event = lldbutil.fetch_next_event(
|
|
self, self.dbg.GetListener(), odd_process.GetBroadcaster()
|
|
)
|
|
self.assertState(lldb.SBProcess.GetStateFromEvent(event), lldb.eStateRunning)
|
|
# Check that the odd process stopped
|
|
event = lldbutil.fetch_next_event(
|
|
self, self.dbg.GetListener(), odd_process.GetBroadcaster()
|
|
)
|
|
self.assertState(lldb.SBProcess.GetStateFromEvent(event), lldb.eStateStopped)
|
|
|
|
self.assertTrue(odd_process.IsValid(), "Got a valid process")
|
|
self.assertState(
|
|
odd_process.GetState(), lldb.eStateStopped, "Process is stopped"
|
|
)
|
|
|
|
# Set a breakpoint on the odd child process
|
|
bkpt = odd_target.BreakpointCreateBySourceRegex(
|
|
"also break here", self.main_source_file
|
|
)
|
|
self.assertEqual(odd_target.GetNumBreakpoints(), 1)
|
|
self.assertTrue(bkpt, "Second breakpoint set on child scripted process")
|
|
self.assertEqual(bkpt.GetNumLocations(), 1, "Second breakpoint has 1 location")
|
|
|
|
# Verify that the breakpoint was also set on the multiplexer & real target
|
|
self.assertEqual(self.mux_target.GetNumBreakpoints(), 2)
|
|
bkpt = self.mux_target.GetBreakpointAtIndex(1)
|
|
self.assertEqual(
|
|
bkpt.GetNumLocations(), 1, "Second breakpoint set on mux scripted process"
|
|
)
|
|
self.assertTrue(bkpt.MatchesName("multiplexed_scripted_process_421"))
|
|
|
|
self.assertGreater(driving_target.GetNumBreakpoints(), 1)
|
|
|
|
# Resume execution on child process
|
|
error = odd_process.Continue()
|
|
self.assertSuccess(error, "Resuming odd child scripted process")
|
|
self.assertTrue(odd_process.IsValid(), "Got a valid process")
|
|
|
|
# Since all the execution is asynchronous, the order in which events
|
|
# arrive is non-deterministic, so we need a data structure to make sure
|
|
# we received both the running and stopped event for each target.
|
|
|
|
# Initialize the execution event "bingo book", that maps a process index
|
|
# to a dictionary that contains flags that are not set for the process
|
|
# events that we care about (running & stopped)
|
|
|
|
execution_events = {
|
|
1: {lldb.eStateRunning: False, lldb.eStateStopped: False},
|
|
2: {lldb.eStateRunning: False, lldb.eStateStopped: False},
|
|
3: {lldb.eStateRunning: False, lldb.eStateStopped: False},
|
|
}
|
|
|
|
def fetch_process_event(self, execution_events):
|
|
event = lldbutil.fetch_next_event(
|
|
self,
|
|
self.dbg.GetListener(),
|
|
lldb.SBProcess.GetBroadcasterClass(),
|
|
match_class=True,
|
|
)
|
|
state = lldb.SBProcess.GetStateFromEvent(event)
|
|
self.assertIn(state, [lldb.eStateRunning, lldb.eStateStopped])
|
|
event_process = lldb.SBProcess.GetProcessFromEvent(event)
|
|
self.assertTrue(event_process.IsValid())
|
|
event_target = event_process.GetTarget()
|
|
event_target_idx = self.dbg.GetIndexOfTarget(event_target)
|
|
self.assertFalse(
|
|
execution_events[event_target_idx][state],
|
|
"Event already received for this process",
|
|
)
|
|
execution_events[event_target_idx][state] = True
|
|
|
|
for _ in range((self.dbg.GetNumTargets() - 1) * 2):
|
|
fetch_process_event(self, execution_events)
|
|
|
|
for target_index, event_states in execution_events.items():
|
|
for state, is_set in event_states.items():
|
|
self.assertTrue(is_set, f"Target {target_index} has state {state} set")
|
|
|
|
event = lldbutil.fetch_next_event(
|
|
self, self.mux_process_listener, self.mux_process.GetBroadcaster()
|
|
)
|
|
self.assertState(lldb.SBProcess.GetStateFromEvent(event), lldb.eStateRunning)
|
|
|
|
event = lldbutil.fetch_next_event(
|
|
self, self.mux_process_listener, self.mux_process.GetBroadcaster()
|
|
)
|
|
self.assertState(lldb.SBProcess.GetStateFromEvent(event), lldb.eStateStopped)
|
|
|
|
def duplicate_target(self, driving_target):
|
|
exe = driving_target.executable.fullpath
|
|
triple = driving_target.triple
|
|
return self.dbg.CreateTargetWithFileAndTargetTriple(exe, triple)
|
|
|
|
def get_launch_info(self, class_name, script_dict):
|
|
structured_data = lldb.SBStructuredData()
|
|
structured_data.SetFromJSON(json.dumps(script_dict))
|
|
|
|
launch_info = lldb.SBLaunchInfo(None)
|
|
launch_info.SetProcessPluginName("ScriptedProcess")
|
|
launch_info.SetScriptedProcessClassName(class_name)
|
|
launch_info.SetScriptedProcessDictionary(structured_data)
|
|
return launch_info
|
|
|
|
def multiplex(self, muxed_process):
|
|
muxed_process.GetScriptedImplementation().multiplexer = (
|
|
self.mux_process.GetScriptedImplementation()
|
|
)
|
|
self.mux_process.GetScriptedImplementation().multiplexed_processes[
|
|
muxed_process.GetProcessID()
|
|
] = muxed_process
|
|
|
|
def passthrough_launch(self):
|
|
"""Test that a simple passthrough wrapper functions correctly"""
|
|
# First build the real target:
|
|
self.assertEqual(self.dbg.GetNumTargets(), 1)
|
|
real_target_id = 0
|
|
real_target = self.dbg.GetTargetAtIndex(real_target_id)
|
|
lldbutil.run_break_set_by_source_regexp(self, "Break here")
|
|
self.assertEqual(real_target.GetNumBreakpoints(), 1)
|
|
|
|
# Now source in the scripted module:
|
|
script_path = os.path.join(self.getSourceDir(), self.script_file)
|
|
self.runCmd(f"command script import '{script_path}'")
|
|
|
|
self.mux_target = self.duplicate_target(real_target)
|
|
self.assertTrue(self.mux_target.IsValid(), "duplicate target succeeded")
|
|
|
|
mux_class = f"{self.script_module}.MultiplexerScriptedProcess"
|
|
script_dict = {"driving_target_idx": real_target_id}
|
|
mux_launch_info = self.get_launch_info(mux_class, script_dict)
|
|
self.mux_process_listener = lldb.SBListener(
|
|
"lldb.test.interactive-scripted-process.listener"
|
|
)
|
|
mux_launch_info.SetShadowListener(self.mux_process_listener)
|
|
|
|
self.dbg.SetAsync(True)
|
|
error = lldb.SBError()
|
|
self.mux_process = self.mux_target.Launch(mux_launch_info, error)
|
|
self.assertSuccess(error, "Launched multiplexer scripted process")
|
|
self.assertTrue(self.mux_process.IsValid(), "Got a valid process")
|
|
|
|
# Check that the real process started running
|
|
event = lldbutil.fetch_next_event(
|
|
self, self.dbg.GetListener(), self.mux_process.GetBroadcaster()
|
|
)
|
|
self.assertState(lldb.SBProcess.GetStateFromEvent(event), lldb.eStateRunning)
|
|
# Check that the mux process started running
|
|
event = lldbutil.fetch_next_event(
|
|
self, self.mux_process_listener, self.mux_process.GetBroadcaster()
|
|
)
|
|
self.assertState(lldb.SBProcess.GetStateFromEvent(event), lldb.eStateRunning)
|
|
|
|
# Check that the real process stopped
|
|
event = lldbutil.fetch_next_event(
|
|
self, self.dbg.GetListener(), self.mux_process.GetBroadcaster()
|
|
)
|
|
self.assertState(lldb.SBProcess.GetStateFromEvent(event), lldb.eStateStopped)
|
|
# Check that the mux process stopped
|
|
event = lldbutil.fetch_next_event(
|
|
self, self.mux_process_listener, self.mux_process.GetBroadcaster()
|
|
)
|
|
self.assertState(lldb.SBProcess.GetStateFromEvent(event), lldb.eStateStopped)
|
|
|
|
real_process = real_target.GetProcess()
|
|
self.assertTrue(real_process.IsValid(), "Got a valid process")
|
|
self.assertState(
|
|
real_process.GetState(), lldb.eStateStopped, "Process is stopped"
|
|
)
|
|
|
|
# This is a passthrough, so the two processes should have the same state:
|
|
# Check that we got the right threads:
|
|
self.assertEqual(
|
|
len(real_process.threads),
|
|
len(self.mux_process.threads),
|
|
"Same number of threads",
|
|
)
|
|
for id in range(len(real_process.threads)):
|
|
real_pc = real_process.threads[id].frame[0].pc
|
|
mux_pc = self.mux_process.threads[id].frame[0].pc
|
|
self.assertEqual(real_pc, mux_pc, f"PC's equal for {id}")
|