1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
|
import unittest2
import gdbremote_testcase
from lldbtest import *
class TestGdbRemote_qThreadStopInfo(gdbremote_testcase.GdbRemoteTestCaseBase):
THREAD_COUNT = 5
def gather_stop_replies_via_qThreadStopInfo(self, thread_count):
# Set up the inferior args.
inferior_args=[]
for i in range(thread_count - 1):
inferior_args.append("thread:new")
inferior_args.append("sleep:10")
procs = self.prep_debug_monitor_and_inferior(inferior_args=inferior_args)
# Assumes test_sequence has anything added needed to setup the initial state.
# (Like optionally enabling QThreadsInStopReply.)
self.test_sequence.add_log_lines([
"read packet: $c#00"
], True)
context = self.expect_gdbremote_sequence()
self.assertIsNotNone(context)
# Give threads time to start up, then break.
time.sleep(1)
self.reset_test_sequence()
self.test_sequence.add_log_lines([
"read packet: {}".format(chr(03)),
{"direction":"send", "regex":r"^\$T([0-9a-fA-F]+)([^#]+)#[0-9a-fA-F]{2}$", "capture":{1:"stop_result", 2:"key_vals_text"} },
], True)
context = self.expect_gdbremote_sequence()
self.assertIsNotNone(context)
# Wait until all threads have started.
threads = self.wait_for_thread_count(thread_count, timeout_seconds=3)
self.assertIsNotNone(threads)
self.assertEquals(len(threads), thread_count)
# Grab stop reply for each thread via qThreadStopInfo{tid:hex}.
stop_replies = {}
for thread in threads:
# Run the qThreadStopInfo command.
self.reset_test_sequence()
self.test_sequence.add_log_lines([
"read packet: $qThreadStopInfo{:x}#00".format(thread),
{"direction":"send", "regex":r"^\$T([0-9a-fA-F]+)([^#]+)#[0-9a-fA-F]{2}$", "capture":{1:"stop_result", 2:"key_vals_text"} },
], True)
context = self.expect_gdbremote_sequence()
self.assertIsNotNone(context)
# Parse stop reply contents.
key_vals_text = context.get("key_vals_text")
self.assertIsNotNone(key_vals_text)
kv_dict = self.parse_key_val_dict(key_vals_text)
self.assertIsNotNone(kv_dict)
# Verify there is a thread and that it matches the expected thread id.
kv_thread = kv_dict.get("thread")
self.assertIsNotNone(kv_thread)
kv_thread_id = int(kv_thread, 16)
self.assertEquals(kv_thread_id, thread)
# Grab the stop id reported.
stop_result_text = context.get("stop_result")
self.assertIsNotNone(stop_result_text)
stop_replies[kv_thread_id] = int(stop_result_text, 16)
return stop_replies
def qThreadStopInfo_works_for_multiple_threads(self, thread_count):
stop_replies = self.gather_stop_replies_via_qThreadStopInfo(thread_count)
self.assertEquals(len(stop_replies), thread_count)
@debugserver_test
@dsym_test
def test_qThreadStopInfo_works_for_multiple_threads_debugserver_dsym(self):
self.init_debugserver_test()
self.buildDsym()
self.set_inferior_startup_launch()
self.qThreadStopInfo_works_for_multiple_threads(self.THREAD_COUNT)
@llgs_test
@dwarf_test
def test_qThreadStopInfo_works_for_multiple_threads_llgs_dwarf(self):
self.init_llgs_test()
self.buildDwarf()
self.set_inferior_startup_launch()
self.qThreadStopInfo_works_for_multiple_threads(self.THREAD_COUNT)
def qThreadStopInfo_only_reports_one_thread_stop_reason_during_interrupt(self, thread_count):
stop_replies = self.gather_stop_replies_via_qThreadStopInfo(thread_count)
self.assertIsNotNone(stop_replies)
no_stop_reason_count = sum(1 for stop_reason in stop_replies.values() if stop_reason == 0)
with_stop_reason_count = sum(1 for stop_reason in stop_replies.values() if stop_reason != 0)
# All but one thread should report no stop reason.
self.assertEqual(no_stop_reason_count, thread_count - 1)
# Only one thread should should indicate a stop reason.
self.assertEqual(with_stop_reason_count, 1)
@debugserver_test
@dsym_test
def test_qThreadStopInfo_only_reports_one_thread_stop_reason_during_interrupt_debugserver_dsym(self):
self.init_debugserver_test()
self.buildDsym()
self.set_inferior_startup_launch()
self.qThreadStopInfo_only_reports_one_thread_stop_reason_during_interrupt(self.THREAD_COUNT)
@llgs_test
@dwarf_test
@unittest2.expectedFailure
def test_qThreadStopInfo_only_reports_one_thread_stop_reason_during_interrupt_llgs_dwarf(self):
self.init_llgs_test()
self.buildDwarf()
self.set_inferior_startup_launch()
self.qThreadStopInfo_only_reports_one_thread_stop_reason_during_interrupt(self.THREAD_COUNT)
if __name__ == '__main__':
unittest2.main()
|
