1""" 2Make sure that if threads are suspended outside of lldb, debugserver 3won't make them run, even if we call an expression on the thread. 4""" 5 6import lldb 7from lldbsuite.test.decorators import * 8import lldbsuite.test.lldbutil as lldbutil 9from lldbsuite.test.lldbtest import * 10 11 12class TestSuspendedThreadHandling(TestBase): 13 NO_DEBUG_INFO_TESTCASE = True 14 15 @skipUnlessDarwin 16 def test_suspended_threads(self): 17 """Test that debugserver doesn't disturb the suspend count of a thread 18 that has been suspended from within a program, when navigating breakpoints 19 on other threads, or calling functions both on the suspended thread and 20 on other threads.""" 21 self.build() 22 self.main_source_file = lldb.SBFileSpec("main.c") 23 self.suspended_thread_test() 24 25 def setUp(self): 26 # Call super's setUp(). 27 TestBase.setUp(self) 28 # Set up your test case here. If your test doesn't need any set up then 29 # remove this method from your TestCase class. 30 31 def try_an_expression(self, thread, correct_value, test_bp): 32 frame = thread.frames[0] 33 34 value = frame.EvaluateExpression("function_to_call()") 35 self.assertSuccess(value.GetError(), "Successfully called the function") 36 self.assertEqual( 37 value.GetValueAsSigned(), correct_value, "Got expected value for expression" 38 ) 39 40 # Again, make sure we didn't let the suspend thread breakpoint run: 41 self.assertEqual( 42 test_bp.GetHitCount(), 43 0, 44 "First expression allowed the suspend thread to run", 45 ) 46 47 def make_bkpt(self, pattern): 48 bp = self.target.BreakpointCreateBySourceRegex(pattern, self.main_source_file) 49 self.assertEqual(bp.GetNumLocations(), 1, "Locations for %s" % (pattern)) 50 return bp 51 52 def suspended_thread_test(self): 53 (self.target, process, thread, bkpt) = lldbutil.run_to_source_breakpoint( 54 self, "Stop here to get things going", self.main_source_file 55 ) 56 57 # Make in the running thread, so the we will have to stop a number of times 58 # while handling breakpoints. The first couple of times we hit it we will 59 # run expressions as well. Make sure we don't let the suspended thread run 60 # during those operations. 61 rt_bp = self.make_bkpt("Break here to show we can handle breakpoints") 62 63 # Make a breakpoint that we will hit when the running thread exits: 64 rt_exit_bp = self.make_bkpt("Break here after thread_join") 65 66 # Make a breakpoint in the suspended thread. We should not hit this till we 67 # resume it after joining the running thread. 68 st_bp = self.make_bkpt("We allowed the suspend thread to run") 69 70 # Make a breakpoint after pthread_join of the suspend thread to ensure 71 # that we didn't keep the thread from exiting normally 72 st_exit_bp = self.make_bkpt( 73 " Break here to make sure the thread exited normally" 74 ) 75 76 threads = lldbutil.continue_to_breakpoint(process, rt_bp) 77 self.assertEqual(len(threads), 1, "Hit the running_func breakpoint") 78 79 # Make sure we didn't hit the suspend thread breakpoint: 80 self.assertEqual( 81 st_bp.GetHitCount(), 0, "Continue allowed the suspend thread to run" 82 ) 83 84 # Now try an expression on the running thread: 85 self.try_an_expression(threads[0], 0, st_bp) 86 87 # Continue, and check the same things: 88 threads = lldbutil.continue_to_breakpoint(process, rt_bp) 89 self.assertEqual(len(threads), 1, "We didn't hit running breakpoint") 90 91 # Try an expression on the suspended thread: 92 thread = lldb.SBThread() 93 for thread in process.threads: 94 th_name = thread.GetName() 95 if th_name is None: 96 continue 97 if "Look for me" in th_name: 98 break 99 self.assertTrue(thread.IsValid(), "We found the suspend thread.") 100 self.try_an_expression(thread, 1, st_bp) 101 102 # Now set the running thread breakpoint to auto-continue and let it 103 # run a bit to make sure we still don't let the suspend thread run. 104 rt_bp.SetAutoContinue(True) 105 threads = lldbutil.continue_to_breakpoint(process, rt_exit_bp) 106 self.assertEqual(len(threads), 1) 107 self.assertEqual( 108 st_bp.GetHitCount(), 0, "Continue again let suspended thread run" 109 ) 110 111 # Now continue and we SHOULD hit the suspend_func breakpoint: 112 threads = lldbutil.continue_to_breakpoint(process, st_bp) 113 self.assertEqual(len(threads), 1, "The thread resumed successfully") 114 115 # Finally, continue again and we should get out of the last pthread_join 116 # and the process should be about to exit 117 threads = lldbutil.continue_to_breakpoint(process, st_exit_bp) 118 self.assertEqual(len(threads), 1, "pthread_join exited successfully") 119