489 lines · python
1"""2Test lldb Python event APIs.3"""4 5import re6import lldb7from lldbsuite.test.decorators import *8from lldbsuite.test.lldbtest import *9from lldbsuite.test import lldbutil10import random11 12@skipIfLinux # llvm.org/pr25924, sometimes generating SIGSEGV13class EventAPITestCase(TestBase):14 NO_DEBUG_INFO_TESTCASE = True15 16 def setUp(self):17 # Call super's setUp().18 TestBase.setUp(self)19 # Find the line number to of function 'c'.20 self.line = line_number(21 "main.c", '// Find the line number of function "c" here.'22 )23 random.seed()24 25 @expectedFailureAll(26 oslist=["linux"], bugnumber="llvm.org/pr23730 Flaky, fails ~1/10 cases"27 )28 @skipIfWindows # This is flakey on Windows AND when it fails, it hangs: llvm.org/pr3837329 @skipIfNetBSD30 def test_listen_for_and_print_event(self):31 """Exercise SBEvent API."""32 self.build()33 exe = self.getBuildArtifact("a.out")34 35 self.dbg.SetAsync(True)36 37 # Create a target by the debugger.38 target = self.dbg.CreateTarget(exe)39 self.assertTrue(target, VALID_TARGET)40 41 # Now create a breakpoint on main.c by name 'c'.42 breakpoint = target.BreakpointCreateByName("c", "a.out")43 44 listener = lldb.SBListener("my listener")45 46 # Now launch the process, and do not stop at the entry point.47 error = lldb.SBError()48 flags = target.GetLaunchInfo().GetLaunchFlags()49 process = target.Launch(50 listener,51 None, # argv52 None, # envp53 None, # stdin_path54 None, # stdout_path55 None, # stderr_path56 None, # working directory57 flags, # launch flags58 False, # Stop at entry59 error,60 ) # error61 62 self.assertEqual(process.GetState(), lldb.eStateStopped, PROCESS_STOPPED)63 64 # Create an empty event object.65 event = lldb.SBEvent()66 67 traceOn = self.TraceOn()68 if traceOn:69 lldbutil.print_stacktraces(process)70 71 # Create MyListeningThread class to wait for any kind of event.72 import threading73 74 class MyListeningThread(threading.Thread):75 def run(self):76 count = 077 # Let's only try at most 4 times to retrieve any kind of event.78 # After that, the thread exits.79 while not count > 3:80 if traceOn:81 print("Try wait for event...")82 if listener.WaitForEvent(5, event):83 if traceOn:84 desc = lldbutil.get_description(event)85 print("Event description:", desc)86 print("Event data flavor:", event.GetDataFlavor())87 print(88 "Process state:",89 lldbutil.state_type_to_str(process.GetState()),90 )91 print()92 else:93 if traceOn:94 print("timeout occurred waiting for event...")95 count = count + 196 listener.Clear()97 return98 99 # Let's start the listening thread to retrieve the events.100 my_thread = MyListeningThread()101 my_thread.start()102 103 # Use Python API to continue the process. The listening thread should be104 # able to receive the state changed events.105 process.Continue()106 107 # Use Python API to kill the process. The listening thread should be108 # able to receive the state changed event, too.109 process.Kill()110 111 # Wait until the 'MyListeningThread' terminates.112 my_thread.join()113 114 # Shouldn't we be testing against some kind of expectation here?115 116 @expectedFlakeyLinux("llvm.org/pr23730") # Flaky, fails ~1/100 cases117 @skipIfWindows # This is flakey on Windows AND when it fails, it hangs: llvm.org/pr38373118 @skipIfNetBSD119 def test_wait_for_event(self):120 """Exercise SBListener.WaitForEvent() API."""121 self.build()122 exe = self.getBuildArtifact("a.out")123 124 self.dbg.SetAsync(True)125 126 # Create a target by the debugger.127 target = self.dbg.CreateTarget(exe)128 self.assertTrue(target, VALID_TARGET)129 130 # Now create a breakpoint on main.c by name 'c'.131 breakpoint = target.BreakpointCreateByName("c", "a.out")132 self.trace("breakpoint:", breakpoint)133 self.assertTrue(134 breakpoint and breakpoint.GetNumLocations() == 1, VALID_BREAKPOINT135 )136 137 # Get the debugger listener.138 listener = self.dbg.GetListener()139 140 # Now launch the process, and do not stop at entry point.141 error = lldb.SBError()142 flags = target.GetLaunchInfo().GetLaunchFlags()143 process = target.Launch(144 listener,145 None, # argv146 None, # envp147 None, # stdin_path148 None, # stdout_path149 None, # stderr_path150 None, # working directory151 flags, # launch flags152 False, # Stop at entry153 error,154 ) # error155 self.assertTrue(error.Success() and process, PROCESS_IS_VALID)156 157 # Create an empty event object.158 event = lldb.SBEvent()159 self.assertFalse(event, "Event should not be valid initially")160 161 # Create MyListeningThread to wait for any kind of event.162 import threading163 164 class MyListeningThread(threading.Thread):165 def run(self):166 count = 0167 # Let's only try at most 3 times to retrieve any kind of event.168 while not count > 3:169 if listener.WaitForEvent(5, event):170 self.context.trace("Got a valid event:", event)171 self.context.trace("Event data flavor:", event.GetDataFlavor())172 self.context.trace(173 "Event type:", lldbutil.state_type_to_str(event.GetType())174 )175 listener.Clear()176 return177 count = count + 1178 print("Timeout: listener.WaitForEvent")179 listener.Clear()180 return181 182 # Use Python API to kill the process. The listening thread should be183 # able to receive a state changed event.184 process.Kill()185 186 # Let's start the listening thread to retrieve the event.187 my_thread = MyListeningThread()188 my_thread.context = self189 my_thread.start()190 191 # Wait until the 'MyListeningThread' terminates.192 my_thread.join()193 194 self.assertTrue(event, "My listening thread successfully received an event")195 196 @expectedFailureAll(197 oslist=["linux"], bugnumber="llvm.org/pr23617 Flaky, fails ~1/10 cases"198 )199 @skipIfWindows # This is flakey on Windows AND when it fails, it hangs: llvm.org/pr38373200 @expectedFailureNetBSD201 def test_add_listener_to_broadcaster(self):202 """Exercise some SBBroadcaster APIs."""203 self.build()204 exe = self.getBuildArtifact("a.out")205 206 self.dbg.SetAsync(True)207 208 # Create a target by the debugger.209 target = self.dbg.CreateTarget(exe)210 self.assertTrue(target, VALID_TARGET)211 212 # Now create a breakpoint on main.c by name 'c'.213 breakpoint = target.BreakpointCreateByName("c", "a.out")214 self.trace("breakpoint:", breakpoint)215 self.assertTrue(216 breakpoint and breakpoint.GetNumLocations() == 1, VALID_BREAKPOINT217 )218 219 listener = lldb.SBListener("my listener")220 221 # Now launch the process, and do not stop at the entry point.222 error = lldb.SBError()223 flags = target.GetLaunchInfo().GetLaunchFlags()224 process = target.Launch(225 listener,226 None, # argv227 None, # envp228 None, # stdin_path229 None, # stdout_path230 None, # stderr_path231 None, # working directory232 flags, # launch flags233 False, # Stop at entry234 error,235 ) # error236 237 # Create an empty event object.238 event = lldb.SBEvent()239 self.assertFalse(event, "Event should not be valid initially")240 241 # The finite state machine for our custom listening thread, with an242 # initial state of None, which means no event has been received.243 # It changes to 'connected' after 'connected' event is received (for remote platforms)244 # It changes to 'running' after 'running' event is received (should happen only if the245 # currentstate is either 'None' or 'connected')246 # It changes to 'stopped' if a 'stopped' event is received (should happen only if the247 # current state is 'running'.)248 self.state = None249 250 # Create MyListeningThread to wait for state changed events.251 # By design, a "running" event is expected following by a "stopped"252 # event.253 import threading254 255 class MyListeningThread(threading.Thread):256 def run(self):257 self.context.trace("Running MyListeningThread:", self)258 259 # Regular expression pattern for the event description.260 pattern = re.compile("data = {.*, state = (.*)}$")261 262 # Let's only try at most 6 times to retrieve our events.263 count = 0264 while True:265 if listener.WaitForEvent(5, event):266 desc = lldbutil.get_description(event)267 self.context.trace("Event description:", desc)268 match = pattern.search(desc)269 if not match:270 break271 if match.group(1) == "connected":272 # When debugging remote targets with lldb-server, we273 # first get the 'connected' event.274 self.context.assertTrue(self.context.state is None)275 self.context.state = "connected"276 continue277 elif match.group(1) == "running":278 self.context.assertTrue(279 self.context.state is None280 or self.context.state == "connected"281 )282 self.context.state = "running"283 continue284 elif match.group(1) == "stopped":285 self.context.assertTrue(self.context.state == "running")286 # Whoopee, both events have been received!287 self.context.state = "stopped"288 break289 else:290 break291 print("Timeout: listener.WaitForEvent")292 count = count + 1293 if count > 6:294 break295 listener.Clear()296 return297 298 # Use Python API to continue the process. The listening thread should be299 # able to receive the state changed events.300 process.Continue()301 302 # Start the listening thread to receive the "running" followed by the303 # "stopped" events.304 my_thread = MyListeningThread()305 # Supply the enclosing context so that our listening thread can access306 # the 'state' variable.307 my_thread.context = self308 my_thread.start()309 310 # Wait until the 'MyListeningThread' terminates.311 my_thread.join()312 313 # The final judgement. :-)314 self.assertEqual(315 self.state, "stopped", "Both expected state changed events received"316 )317 318 def wait_for_next_event(self, expected_state, test_shadow=False):319 """Wait for an event from self.primary & self.shadow listener.320 If test_shadow is true, we also check that the shadow listener only321 receives events AFTER the primary listener does."""322 import stop_hook323 # Waiting on the shadow listener shouldn't have events yet because324 # we haven't fetched them for the primary listener yet:325 event = lldb.SBEvent()326 327 if test_shadow:328 success = self.shadow_listener.WaitForEvent(1, event)329 self.assertFalse(success, "Shadow listener doesn't pull events")330 331 # But there should be an event for the primary listener:332 success = self.primary_listener.WaitForEvent(5, event)333 334 self.assertTrue(success, "Primary listener got the event")335 336 state = lldb.SBProcess.GetStateFromEvent(event)337 primary_event_type = event.GetType()338 restart = False339 if state == lldb.eStateStopped:340 restart = lldb.SBProcess.GetRestartedFromEvent(event)341 # This counter is matching the stop hooks, which don't get run342 # for auto-restarting stops.343 if not restart:344 self.stop_counter += 1345 self.assertEqual(346 stop_hook.StopHook.counter[self.instance],347 self.stop_counter,348 "matching stop hook",349 )350 351 if expected_state is not None:352 self.assertEqual(353 state, expected_state, "Primary thread got the correct event"354 )355 356 # And after pulling that one there should be an equivalent event for the shadow357 # listener:358 success = self.shadow_listener.WaitForEvent(5, event)359 self.assertTrue(success, "Shadow listener got event too")360 shadow_event_type = event.GetType()361 self.assertEqual(362 primary_event_type, shadow_event_type, "It was the same event type"363 )364 self.assertEqual(365 state, lldb.SBProcess.GetStateFromEvent(event), "It was the same state"366 )367 self.assertEqual(368 restart,369 lldb.SBProcess.GetRestartedFromEvent(event),370 "It was the same restarted",371 )372 return state, restart373 374 @expectedFlakeyLinux("llvm.org/pr23730") # Flaky, fails ~1/100 cases375 @skipIfWindows # This is flakey on Windows AND when it fails, it hangs: llvm.org/pr38373376 @skipIfNetBSD377 def test_shadow_listener(self):378 self.build()379 exe = self.getBuildArtifact("a.out")380 381 # Create a target by the debugger.382 target = self.dbg.CreateTarget(exe)383 self.assertTrue(target, VALID_TARGET)384 385 # Now create a breakpoint on main.c by name 'c'.386 bkpt1 = target.BreakpointCreateByName("c", "a.out")387 self.trace("breakpoint:", bkpt1)388 self.assertEqual(bkpt1.GetNumLocations(), 1, VALID_BREAKPOINT)389 390 self.primary_listener = lldb.SBListener("my listener")391 self.shadow_listener = lldb.SBListener("shadow listener")392 393 self.cur_thread = None394 395 error = lldb.SBError()396 launch_info = target.GetLaunchInfo()397 launch_info.SetListener(self.primary_listener)398 launch_info.SetShadowListener(self.shadow_listener)399 400 self.runCmd(401 "settings set target.process.extra-startup-command QSetLogging:bitmask=LOG_PROCESS|LOG_EXCEPTIONS|LOG_RNB_PACKETS|LOG_STEP;"402 )403 self.dbg.SetAsync(True)404 405 # Now make our stop hook - we want to ensure it stays up to date with406 # the events. We can't get our hands on the stop-hook instance directly,407 # so we'll pass in an instance key, and use that to retrieve the data from408 # this instance of the stop hook:409 self.instance = f"Key{random.randint(0,10000)}"410 stop_hook_path = os.path.join(self.getSourceDir(), "stop_hook.py")411 self.runCmd(f"command script import {stop_hook_path}")412 import stop_hook413 414 # Add our stop hook here, don't report on the initial attach:415 self.runCmd(416 f"target stop-hook add -P stop_hook.StopHook -k instance -v {self.instance} -I false"417 )418 self.stop_counter = 0419 420 self.process = target.Launch(launch_info, error)421 self.assertSuccess(error, "Process launched successfully")422 423 # Keep fetching events from the primary to trigger the do on removal and424 # then from the shadow listener, and make sure they match:425 426 # Events in the launch sequence might be platform dependent, so don't427 # expect any particular event till we get the stopped:428 state = lldb.eStateInvalid429 430 while state != lldb.eStateStopped:431 state, restart = self.wait_for_next_event(None, False)432 433 # Okay, we're now at a good stop, so try a next:434 self.cur_thread = self.process.threads[0]435 436 # Make sure we're at our expected breakpoint:437 self.assertTrue(self.cur_thread.IsValid(), "Got a zeroth thread")438 self.assertEqual(self.cur_thread.stop_reason, lldb.eStopReasonBreakpoint)439 self.assertEqual(440 self.cur_thread.GetStopReasonDataCount(), 2, "Only one breakpoint/loc here"441 )442 self.assertEqual(443 bkpt1.GetID(),444 self.cur_thread.GetStopReasonDataAtIndex(0),445 "Hit the right breakpoint",446 )447 448 self.cur_thread.StepOver()449 # We'll run the test for "shadow listener blocked by primary listener450 # for the first couple rounds, then we'll skip the 1 second pause...451 self.wait_for_next_event(lldb.eStateRunning, True)452 self.wait_for_next_event(lldb.eStateStopped, True)453 454 # Next try an auto-continue breakpoint and make sure the shadow listener got455 # the resumed info as well. Note that I'm not explicitly counting456 # running events here. At the point when I wrote this lldb sometimes457 # emits two running events in a row. Apparently the code to coalesce running458 # events isn't working. But that's not what this test is testing, we're really459 # testing that the primary & shadow listeners hear the same thing and in the460 # right order.461 462 main_spec = lldb.SBFileSpec("main.c")463 bkpt2 = target.BreakpointCreateBySourceRegex("b.2. returns %d", main_spec)464 self.assertGreater(bkpt2.GetNumLocations(), 0, "BP2 worked")465 bkpt2.SetAutoContinue(True)466 467 bkpt3 = target.BreakpointCreateBySourceRegex("a.3. returns %d", main_spec)468 self.assertGreater(bkpt3.GetNumLocations(), 0, "BP3 worked")469 470 state = lldb.eStateStopped471 restarted = False472 473 # Put in a counter to make sure we don't spin forever if there is some474 # error in the logic.475 counter = 0476 while state != lldb.eStateExited:477 counter += 1478 self.assertLess(479 counter, 50, "Took more than 50 events to hit two breakpoints."480 )481 if state == lldb.eStateStopped and not restarted:482 self.process.Continue()483 484 state, restarted = self.wait_for_next_event(None, False)485 486 # Now make sure we agree with the stop hook counter:487 self.assertEqual(self.stop_counter, stop_hook.StopHook.counter[self.instance])488 self.assertEqual(stop_hook.StopHook.non_stops[self.instance], 0, "No non stops")489