150
|
1 """Test function call thread safety."""
|
|
2
|
|
3
|
|
4
|
|
5 import lldb
|
|
6 from lldbsuite.test.decorators import *
|
|
7 from lldbsuite.test.lldbtest import *
|
|
8 from lldbsuite.test import lldbutil
|
|
9
|
|
10
|
|
11 class TestSafeFuncCalls(TestBase):
|
|
12
|
|
13 mydir = TestBase.compute_mydir(__file__)
|
|
14
|
|
15 def setUp(self):
|
|
16 # Call super's setUp().
|
|
17 TestBase.setUp(self)
|
|
18 # Find the line numbers that we will step to in main:
|
|
19 self.main_source = "main.c"
|
|
20
|
|
21 @skipUnlessDarwin
|
|
22 @add_test_categories(['pyapi'])
|
|
23 def test_with_python_api(self):
|
|
24 """Test function call thread safety."""
|
|
25 self.build()
|
|
26 exe = self.getBuildArtifact("a.out")
|
|
27
|
|
28 target = self.dbg.CreateTarget(exe)
|
|
29 self.assertTrue(target, VALID_TARGET)
|
|
30 self.main_source_spec = lldb.SBFileSpec(self.main_source)
|
|
31 break1 = target.BreakpointCreateByName("stopper", 'a.out')
|
|
32 self.assertTrue(break1, VALID_BREAKPOINT)
|
|
33 process = target.LaunchSimple(
|
|
34 None, None, self.get_process_working_directory())
|
|
35 self.assertTrue(process, PROCESS_IS_VALID)
|
|
36 threads = lldbutil.get_threads_stopped_at_breakpoint(process, break1)
|
|
37 if len(threads) != 1:
|
|
38 self.fail("Failed to stop at breakpoint 1.")
|
|
39
|
|
40 self.check_number_of_threads(process)
|
|
41
|
|
42 main_thread = lldb.SBThread()
|
|
43 select_thread = lldb.SBThread()
|
|
44 for idx in range(0, process.GetNumThreads()):
|
|
45 t = process.GetThreadAtIndex(idx)
|
|
46 if t.GetName() == "main thread":
|
|
47 main_thread = t
|
|
48 if t.GetName() == "select thread":
|
|
49 select_thread = t
|
|
50
|
|
51 self.assertTrue(
|
|
52 main_thread.IsValid() and select_thread.IsValid(),
|
|
53 "Got both expected threads")
|
|
54
|
|
55 self.safe_to_call_func_on_main_thread(main_thread)
|
|
56 self.safe_to_call_func_on_select_thread(select_thread)
|
|
57
|
|
58 def check_number_of_threads(self, process):
|
|
59 self.assertTrue(
|
|
60 process.GetNumThreads() == 2,
|
|
61 "Check that the process has two threads when sitting at the stopper() breakpoint")
|
|
62
|
|
63 def safe_to_call_func_on_main_thread(self, main_thread):
|
|
64 self.assertTrue(main_thread.SafeToCallFunctions(),
|
|
65 "It is safe to call functions on the main thread")
|
|
66
|
|
67 def safe_to_call_func_on_select_thread(self, select_thread):
|
|
68 self.assertTrue(
|
|
69 select_thread.SafeToCallFunctions() == False,
|
|
70 "It is not safe to call functions on the select thread")
|