1import random
2
3from lldbsuite.test.decorators import *
4from lldbsuite.test.lldbtest import *
5
6from fork_testbase import GdbRemoteForkTestBase
7
8
9class TestGdbRemoteFork(GdbRemoteForkTestBase):
10    @add_test_categories(["fork"])
11    def test_fork_multithreaded(self):
12        _, _, child_pid, _ = self.start_fork_test(["thread:new"]*2 + ["fork"])
13
14        # detach the forked child
15        self.test_sequence.add_log_lines([
16            "read packet: $D;{}#00".format(child_pid),
17            "send packet: $OK#00",
18            "read packet: $k#00",
19        ], True)
20        self.expect_gdbremote_sequence()
21
22    @add_test_categories(["fork"])
23    def test_fork(self):
24        parent_pid, _ = self.fork_and_detach_test("fork")
25
26        # resume the parent
27        self.test_sequence.add_log_lines([
28            "read packet: $c#00",
29            "send packet: $W00;process:{}#00".format(parent_pid),
30        ], True)
31        self.expect_gdbremote_sequence()
32
33    @add_test_categories(["fork"])
34    def test_vfork(self):
35        parent_pid, parent_tid = self.fork_and_detach_test("vfork")
36
37        # resume the parent
38        self.test_sequence.add_log_lines([
39            "read packet: $c#00",
40            {"direction": "send",
41             "regex": r"[$]T05thread:p{}[.]{}.*vforkdone.*".format(parent_pid,
42                                                                   parent_tid),
43             },
44            "read packet: $c#00",
45            "send packet: $W00;process:{}#00".format(parent_pid),
46        ], True)
47        self.expect_gdbremote_sequence()
48
49    @add_test_categories(["fork"])
50    def test_fork_follow(self):
51        self.fork_and_follow_test("fork")
52
53    @add_test_categories(["fork"])
54    def test_vfork_follow(self):
55        self.fork_and_follow_test("vfork")
56
57    @add_test_categories(["fork"])
58    def test_select_wrong_pid(self):
59        self.build()
60        self.prep_debug_monitor_and_inferior()
61        self.add_qSupported_packets(["multiprocess+"])
62        ret = self.expect_gdbremote_sequence()
63        self.assertIn("multiprocess+", ret["qSupported_response"])
64        self.reset_test_sequence()
65
66        # get process pid
67        self.test_sequence.add_log_lines([
68            "read packet: $qC#00",
69            {"direction": "send", "regex": "[$]QCp([0-9a-f]+).([0-9a-f]+)#.*",
70             "capture": {1: "pid", 2: "tid"}},
71        ], True)
72        ret = self.expect_gdbremote_sequence()
73        pid, tid = (int(ret[x], 16) for x in ("pid", "tid"))
74        self.reset_test_sequence()
75
76        self.test_sequence.add_log_lines([
77            # try switching to correct pid
78            "read packet: $Hgp{:x}.{:x}#00".format(pid, tid),
79            "send packet: $OK#00",
80            "read packet: $Hcp{:x}.{:x}#00".format(pid, tid),
81            "send packet: $OK#00",
82            # try switching to invalid tid
83            "read packet: $Hgp{:x}.{:x}#00".format(pid, tid+1),
84            "send packet: $E15#00",
85            "read packet: $Hcp{:x}.{:x}#00".format(pid, tid+1),
86            "send packet: $E15#00",
87            # try switching to invalid pid
88            "read packet: $Hgp{:x}.{:x}#00".format(pid+1, tid),
89            "send packet: $Eff#00",
90            "read packet: $Hcp{:x}.{:x}#00".format(pid+1, tid),
91            "send packet: $Eff#00",
92        ], True)
93        self.expect_gdbremote_sequence()
94
95    @add_test_categories(["fork"])
96    def test_detach_current(self):
97        self.build()
98        self.prep_debug_monitor_and_inferior()
99        self.add_qSupported_packets(["multiprocess+"])
100        ret = self.expect_gdbremote_sequence()
101        self.assertIn("multiprocess+", ret["qSupported_response"])
102        self.reset_test_sequence()
103
104        # get process pid
105        self.test_sequence.add_log_lines([
106            "read packet: $qC#00",
107            {"direction": "send", "regex": "[$]QCp([0-9a-f]+).[0-9a-f]+#.*",
108             "capture": {1: "pid"}},
109        ], True)
110        ret = self.expect_gdbremote_sequence()
111        pid = ret["pid"]
112        self.reset_test_sequence()
113
114        # detach the process
115        self.test_sequence.add_log_lines([
116            "read packet: $D;{}#00".format(pid),
117            "send packet: $OK#00",
118            "read packet: $qC#00",
119            "send packet: $E44#00",
120        ], True)
121        self.expect_gdbremote_sequence()
122
123    @add_test_categories(["fork"])
124    def test_detach_all(self):
125        self.detach_all_test()
126
127    @add_test_categories(["fork"])
128    def test_kill_all(self):
129        parent_pid, _, child_pid, _ = self.start_fork_test(["fork"])
130
131        exit_regex = "[$]X09;process:([0-9a-f]+)#.*"
132        self.test_sequence.add_log_lines([
133            # kill all processes
134            "read packet: $k#00",
135            {"direction": "send", "regex": exit_regex,
136             "capture": {1: "pid1"}},
137            {"direction": "send", "regex": exit_regex,
138             "capture": {1: "pid2"}},
139        ], True)
140        ret = self.expect_gdbremote_sequence()
141        self.assertEqual(set([ret["pid1"], ret["pid2"]]),
142                         set([parent_pid, child_pid]))
143
144    @add_test_categories(["fork"])
145    def test_vkill_child(self):
146        self.vkill_test(kill_child=True)
147
148    @add_test_categories(["fork"])
149    def test_vkill_parent(self):
150        self.vkill_test(kill_parent=True)
151
152    @add_test_categories(["fork"])
153    def test_vkill_both(self):
154        self.vkill_test(kill_parent=True, kill_child=True)
155
156    @expectedFailureAll(archs=["arm"])  # TODO
157    @expectedFailureAll(archs=["aarch64"],
158                        bugnumber="https://github.com/llvm/llvm-project/issues/56268")
159    @add_test_categories(["fork"])
160    def test_c_parent(self):
161        self.resume_one_test(run_order=["parent", "parent"])
162
163    @expectedFailureAll(archs=["arm"])  # TODO
164    @expectedFailureAll(archs=["aarch64"],
165                        bugnumber="https://github.com/llvm/llvm-project/issues/56268")
166    @add_test_categories(["fork"])
167    def test_c_child(self):
168        self.resume_one_test(run_order=["child", "child"])
169
170    @expectedFailureAll(archs=["arm"])  # TODO
171    @expectedFailureAll(archs=["aarch64"],
172                        bugnumber="https://github.com/llvm/llvm-project/issues/56268")
173    @add_test_categories(["fork"])
174    def test_c_parent_then_child(self):
175        self.resume_one_test(run_order=["parent", "parent", "child", "child"])
176
177    @expectedFailureAll(archs=["arm"])  # TODO
178    @expectedFailureAll(archs=["aarch64"],
179                        bugnumber="https://github.com/llvm/llvm-project/issues/56268")
180    @add_test_categories(["fork"])
181    def test_c_child_then_parent(self):
182        self.resume_one_test(run_order=["child", "child", "parent", "parent"])
183
184    @expectedFailureAll(archs=["arm"])  # TODO
185    @expectedFailureAll(archs=["aarch64"],
186                        bugnumber="https://github.com/llvm/llvm-project/issues/56268")
187    @add_test_categories(["fork"])
188    def test_c_interspersed(self):
189        self.resume_one_test(run_order=["parent", "child", "parent", "child"])
190
191    @expectedFailureAll(archs=["arm"])  # TODO
192    @expectedFailureAll(archs=["aarch64"],
193                        bugnumber="https://github.com/llvm/llvm-project/issues/56268")
194    @add_test_categories(["fork"])
195    def test_vCont_parent(self):
196        self.resume_one_test(run_order=["parent", "parent"], use_vCont=True)
197
198    @expectedFailureAll(archs=["arm"])  # TODO
199    @expectedFailureAll(archs=["aarch64"],
200                        bugnumber="https://github.com/llvm/llvm-project/issues/56268")
201    @add_test_categories(["fork"])
202    def test_vCont_child(self):
203        self.resume_one_test(run_order=["child", "child"], use_vCont=True)
204
205    @expectedFailureAll(archs=["arm"])  # TODO
206    @expectedFailureAll(archs=["aarch64"],
207                        bugnumber="https://github.com/llvm/llvm-project/issues/56268")
208    @add_test_categories(["fork"])
209    def test_vCont_parent_then_child(self):
210        self.resume_one_test(run_order=["parent", "parent", "child", "child"],
211                             use_vCont=True)
212
213    @expectedFailureAll(archs=["arm"])  # TODO
214    @expectedFailureAll(archs=["aarch64"],
215                        bugnumber="https://github.com/llvm/llvm-project/issues/56268")
216    @add_test_categories(["fork"])
217    def test_vCont_child_then_parent(self):
218        self.resume_one_test(run_order=["child", "child", "parent", "parent"],
219                             use_vCont=True)
220
221    @expectedFailureAll(archs=["arm"])  # TODO
222    @expectedFailureAll(archs=["aarch64"],
223                        bugnumber="https://github.com/llvm/llvm-project/issues/56268")
224    @add_test_categories(["fork"])
225    def test_vCont_interspersed(self):
226        self.resume_one_test(run_order=["parent", "child", "parent", "child"],
227                             use_vCont=True)
228
229    @add_test_categories(["fork"])
230    def test_vCont_two_processes(self):
231        parent_pid, parent_tid, child_pid, child_tid = (
232            self.start_fork_test(["fork", "trap"]))
233
234        self.test_sequence.add_log_lines([
235            # try to resume both processes
236            "read packet: $vCont;c:p{}.{};c:p{}.{}#00".format(
237                parent_pid, parent_tid, child_pid, child_tid),
238            "send packet: $E03#00",
239        ], True)
240        self.expect_gdbremote_sequence()
241
242    @add_test_categories(["fork"])
243    def test_vCont_all_processes_explicit(self):
244        self.start_fork_test(["fork", "trap"])
245
246        self.test_sequence.add_log_lines([
247            # try to resume all processes implicitly
248            "read packet: $vCont;c:p-1.-1#00",
249            "send packet: $E03#00",
250        ], True)
251        self.expect_gdbremote_sequence()
252
253    @add_test_categories(["fork"])
254    def test_vCont_all_processes_implicit(self):
255        self.start_fork_test(["fork", "trap"])
256
257        self.test_sequence.add_log_lines([
258            # try to resume all processes implicitly
259            "read packet: $vCont;c#00",
260            "send packet: $E03#00",
261        ], True)
262        self.expect_gdbremote_sequence()
263
264    @expectedFailureAll(archs=["arm"])  # TODO
265    @add_test_categories(["fork"])
266    def test_threadinfo(self):
267        parent_pid, parent_tid, child_pid, child_tid = (
268            self.start_fork_test(["fork", "thread:new", "trap"]))
269        pidtids = [
270            (parent_pid, parent_tid),
271            (child_pid, child_tid),
272        ]
273
274        self.add_threadinfo_collection_packets()
275        ret = self.expect_gdbremote_sequence()
276        prev_pidtids = set(self.parse_threadinfo_packets(ret))
277        self.assertEqual(prev_pidtids,
278                         frozenset((int(pid, 16), int(tid, 16))
279                                   for pid, tid in pidtids))
280        self.reset_test_sequence()
281
282        for pidtid in pidtids:
283            self.test_sequence.add_log_lines(
284                ["read packet: $Hcp{}.{}#00".format(*pidtid),
285                 "send packet: $OK#00",
286                 "read packet: $c#00",
287                 {"direction": "send",
288                  "regex": "^[$]T05thread:p{}.{}.*".format(*pidtid),
289                  },
290                 ], True)
291            self.add_threadinfo_collection_packets()
292            ret = self.expect_gdbremote_sequence()
293            self.reset_test_sequence()
294            new_pidtids = set(self.parse_threadinfo_packets(ret))
295            added_pidtid = new_pidtids - prev_pidtids
296            prev_pidtids = new_pidtids
297
298            # verify that we've got exactly one new thread, and that
299            # the PID matches
300            self.assertEqual(len(added_pidtid), 1)
301            self.assertEqual(added_pidtid.pop()[0], int(pidtid[0], 16))
302
303        for pidtid in new_pidtids:
304            self.test_sequence.add_log_lines(
305                ["read packet: $Hgp{:x}.{:x}#00".format(*pidtid),
306                 "send packet: $OK#00",
307                 ], True)
308        self.expect_gdbremote_sequence()
309
310    @expectedFailureAll(archs=["arm"])  # TODO
311    @add_test_categories(["fork"])
312    def test_memory_read_write(self):
313        self.build()
314        INITIAL_DATA = "Initial message"
315        self.prep_debug_monitor_and_inferior(
316            inferior_args=["set-message:{}".format(INITIAL_DATA),
317                           "get-data-address-hex:g_message",
318                           "fork",
319                           "print-message:",
320                           "trap",
321                           ])
322        self.add_qSupported_packets(["multiprocess+",
323                                     "fork-events+"])
324        ret = self.expect_gdbremote_sequence()
325        self.assertIn("fork-events+", ret["qSupported_response"])
326        self.reset_test_sequence()
327
328        # continue and expect fork
329        self.test_sequence.add_log_lines([
330            "read packet: $c#00",
331            {"type": "output_match",
332             "regex": self.maybe_strict_output_regex(r"data address: 0x([0-9a-fA-F]+)\r\n"),
333             "capture": {1: "addr"}},
334            {"direction": "send", "regex": self.fork_regex.format("fork"),
335             "capture": self.fork_capture},
336        ], True)
337        ret = self.expect_gdbremote_sequence()
338        pidtids = {
339            "parent": (ret["parent_pid"], ret["parent_tid"]),
340            "child": (ret["child_pid"], ret["child_tid"]),
341        }
342        addr = ret["addr"]
343        self.reset_test_sequence()
344
345        for name, pidtid in pidtids.items():
346            self.test_sequence.add_log_lines(
347                ["read packet: $Hgp{}.{}#00".format(*pidtid),
348                 "send packet: $OK#00",
349                 # read the current memory contents
350                 "read packet: $m{},{:x}#00".format(addr,
351                                                    len(INITIAL_DATA) + 1),
352                 {"direction": "send",
353                  "regex": r"^[$](.+)#.*$",
354                  "capture": {1: "data"}},
355                 # write a new value
356                 "read packet: $M{},{:x}:{}#00".format(addr,
357                                                       len(name) + 1,
358                                                       seven.hexlify(
359                                                           name + "\0")),
360                 "send packet: $OK#00",
361                 # resume the process and wait for the trap
362                 "read packet: $Hcp{}.{}#00".format(*pidtid),
363                 "send packet: $OK#00",
364                 "read packet: $c#00",
365                 {"type": "output_match",
366                  "regex": self.maybe_strict_output_regex(r"message: (.*)\r\n"),
367                  "capture": {1: "printed_message"}},
368                 {"direction": "send",
369                  "regex": "^[$]T05thread:p{}.{}.*".format(*pidtid),
370                  },
371                 ], True)
372            ret = self.expect_gdbremote_sequence()
373            data = seven.unhexlify(ret["data"])
374            self.assertEqual(data, INITIAL_DATA + "\0")
375            self.assertEqual(ret["printed_message"], name);
376            self.reset_test_sequence()
377
378        # we do the second round separately to make sure that initial data
379        # is correctly preserved while writing into the first process
380
381        for name, pidtid in pidtids.items():
382            self.test_sequence.add_log_lines(
383                ["read packet: $Hgp{}.{}#00".format(*pidtid),
384                 "send packet: $OK#00",
385                 # read the current memory contents
386                 "read packet: $m{},{:x}#00".format(addr,
387                                                    len(name) + 1),
388                 {"direction": "send",
389                  "regex": r"^[$](.+)#.*$",
390                  "capture": {1: "data"}},
391                 ], True)
392            ret = self.expect_gdbremote_sequence()
393            self.assertIsNotNone(ret.get("data"))
394            data = seven.unhexlify(ret.get("data"))
395            self.assertEqual(data, name + "\0")
396            self.reset_test_sequence()
397
398    @expectedFailureAll(archs=["arm"])  # TODO
399    @add_test_categories(["fork"])
400    def test_register_read_write(self):
401        parent_pid, parent_tid, child_pid, child_tid = (
402            self.start_fork_test(["fork", "thread:new", "trap"]))
403        pidtids = [
404            (parent_pid, parent_tid),
405            (child_pid, child_tid),
406        ]
407
408        for pidtid in pidtids:
409            self.test_sequence.add_log_lines(
410                ["read packet: $Hcp{}.{}#00".format(*pidtid),
411                 "send packet: $OK#00",
412                 "read packet: $c#00",
413                 {"direction": "send",
414                  "regex": "^[$]T05thread:p{}.{}.*".format(*pidtid),
415                  },
416                 ], True)
417
418        self.add_threadinfo_collection_packets()
419        ret = self.expect_gdbremote_sequence()
420        self.reset_test_sequence()
421
422        pidtids = set(self.parse_threadinfo_packets(ret))
423        self.assertEqual(len(pidtids), 4)
424        # first, save register values from all the threads
425        thread_regs = {}
426        for pidtid in pidtids:
427            for regno in range(256):
428                self.test_sequence.add_log_lines(
429                    ["read packet: $Hgp{:x}.{:x}#00".format(*pidtid),
430                     "send packet: $OK#00",
431                     "read packet: $p{:x}#00".format(regno),
432                     {"direction": "send",
433                      "regex": r"^[$](.+)#.*$",
434                      "capture": {1: "data"}},
435                     ], True)
436                ret = self.expect_gdbremote_sequence()
437                data = ret.get("data")
438                self.assertIsNotNone(data)
439                # ignore registers shorter than 32 bits (this also catches
440                # "Exx" errors)
441                if len(data) >= 8:
442                    break
443            else:
444                self.skipTest("no usable register found")
445            thread_regs[pidtid] = (regno, data)
446
447        vals = set(x[1] for x in thread_regs.values())
448        # NB: cheap hack to make the loop below easier
449        new_val = next(iter(vals))
450
451        # then, start altering them and verify that we don't unexpectedly
452        # change the value from another thread
453        for pidtid in pidtids:
454            old_val = thread_regs[pidtid]
455            regno = old_val[0]
456            old_val_length = len(old_val[1])
457            # generate a unique new_val
458            while new_val in vals:
459                new_val = ('{{:0{}x}}'.format(old_val_length)
460                           .format(random.getrandbits(old_val_length*4)))
461            vals.add(new_val)
462
463            self.test_sequence.add_log_lines(
464                ["read packet: $Hgp{:x}.{:x}#00".format(*pidtid),
465                 "send packet: $OK#00",
466                 "read packet: $p{:x}#00".format(regno),
467                 {"direction": "send",
468                  "regex": r"^[$](.+)#.*$",
469                  "capture": {1: "data"}},
470                 "read packet: $P{:x}={}#00".format(regno, new_val),
471                 "send packet: $OK#00",
472                 ], True)
473            ret = self.expect_gdbremote_sequence()
474            data = ret.get("data")
475            self.assertIsNotNone(data)
476            self.assertEqual(data, old_val[1])
477            thread_regs[pidtid] = (regno, new_val)
478
479        # finally, verify that new values took effect
480        for pidtid in pidtids:
481            old_val = thread_regs[pidtid]
482            self.test_sequence.add_log_lines(
483                ["read packet: $Hgp{:x}.{:x}#00".format(*pidtid),
484                 "send packet: $OK#00",
485                 "read packet: $p{:x}#00".format(old_val[0]),
486                 {"direction": "send",
487                  "regex": r"^[$](.+)#.*$",
488                  "capture": {1: "data"}},
489                 ], True)
490            ret = self.expect_gdbremote_sequence()
491            data = ret.get("data")
492            self.assertIsNotNone(data)
493            self.assertEqual(data, old_val[1])
494
495    @expectedFailureAll(archs=["arm"])  # TODO
496    @add_test_categories(["fork"])
497    def test_qC(self):
498        parent_pid, parent_tid, child_pid, child_tid = (
499            self.start_fork_test(["fork", "thread:new", "trap"]))
500        pidtids = [
501            (parent_pid, parent_tid),
502            (child_pid, child_tid),
503        ]
504
505        for pidtid in pidtids:
506            self.test_sequence.add_log_lines(
507                ["read packet: $Hcp{}.{}#00".format(*pidtid),
508                 "send packet: $OK#00",
509                 "read packet: $c#00",
510                 {"direction": "send",
511                  "regex": "^[$]T05thread:p{}.{}.*".format(*pidtid),
512                  },
513                 ], True)
514
515        self.add_threadinfo_collection_packets()
516        ret = self.expect_gdbremote_sequence()
517        self.reset_test_sequence()
518
519        pidtids = set(self.parse_threadinfo_packets(ret))
520        self.assertEqual(len(pidtids), 4)
521        for pidtid in pidtids:
522            self.test_sequence.add_log_lines(
523                ["read packet: $Hgp{:x}.{:x}#00".format(*pidtid),
524                 "send packet: $OK#00",
525                 "read packet: $qC#00",
526                 "send packet: $QCp{:x}.{:x}#00".format(*pidtid),
527                 ], True)
528        self.expect_gdbremote_sequence()
529
530    @expectedFailureAll(archs=["arm"])  # TODO
531    @add_test_categories(["fork"])
532    def test_T(self):
533        parent_pid, parent_tid, child_pid, child_tid = (
534            self.start_fork_test(["fork", "thread:new", "trap"]))
535        pidtids = [
536            (parent_pid, parent_tid),
537            (child_pid, child_tid),
538        ]
539
540        for pidtid in pidtids:
541            self.test_sequence.add_log_lines(
542                ["read packet: $Hcp{}.{}#00".format(*pidtid),
543                 "send packet: $OK#00",
544                 "read packet: $c#00",
545                 {"direction": "send",
546                  "regex": "^[$]T05thread:p{}.{}.*".format(*pidtid),
547                  },
548                 ], True)
549
550        self.add_threadinfo_collection_packets()
551        ret = self.expect_gdbremote_sequence()
552        self.reset_test_sequence()
553
554        pidtids = set(self.parse_threadinfo_packets(ret))
555        self.assertEqual(len(pidtids), 4)
556        max_pid = max(pid for pid, tid in pidtids)
557        max_tid = max(tid for pid, tid in pidtids)
558        bad_pidtids = (
559            (max_pid, max_tid + 1, "E02"),
560            (max_pid + 1, max_tid, "E01"),
561            (max_pid + 1, max_tid + 1, "E01"),
562        )
563
564        for pidtid in pidtids:
565            self.test_sequence.add_log_lines(
566                [
567                 # test explicit PID+TID
568                 "read packet: $Tp{:x}.{:x}#00".format(*pidtid),
569                 "send packet: $OK#00",
570                 # test implicit PID via Hg
571                 "read packet: $Hgp{:x}.{:x}#00".format(*pidtid),
572                 "send packet: $OK#00",
573                 "read packet: $T{:x}#00".format(max_tid + 1),
574                 "send packet: $E02#00",
575                 "read packet: $T{:x}#00".format(pidtid[1]),
576                 "send packet: $OK#00",
577                 ], True)
578        for pid, tid, expected in bad_pidtids:
579            self.test_sequence.add_log_lines(
580                ["read packet: $Tp{:x}.{:x}#00".format(pid, tid),
581                 "send packet: ${}#00".format(expected),
582                 ], True)
583        self.expect_gdbremote_sequence()
584