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