lldb/test/API/macosx/thread_suspend/TestInternalThreadSuspension.py

   1 """
   2 Make sure that if threads are suspended outside of lldb, debugserver
   3 won't make them run, even if we call an expression on the thread.
   4 """
   5
   6 import lldb
   7 from lldbsuite.test.decorators import *
   8 import lldbsuite.test.lldbutil as lldbutil
   9 from lldbsuite.test.lldbtest import *
  10
  11
  12 class 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")