[lldb] [llgs] Add a test for multiprocess register read/write

[mgorny]

Summary

Sponsored by: The FreeBSD Foundation

lldb/test/API/tools/lldb-server/TestGdbRemoteFork.py

+import random
+
 import gdbremote_testcase
 from lldbsuite.test.decorators import *
 from lldbsuite.test.lldbtest import *
 from lldbsuite.test import lldbutil
 
 class TestGdbRemoteFork(gdbremote_testcase.GdbRemoteTestCaseBase):
 
     fork_regex = ("[$]T05thread:p([0-9a-f]+)[.]([0-9a-f]+);.*"
def test_memory_read_write(self):
                   "regex": r"^[$](.+)#.*$",
                   "capture": {1: "data"}},
                  ], True)
             ret = self.expect_gdbremote_sequence()
             self.assertIsNotNone(ret.get("data"))
             data = seven.unhexlify(ret.get("data"))
             self.assertEqual(data, name + "\0")
             self.reset_test_sequence()
+
+    @add_test_categories(["fork"])
+    def test_register_read_write(self):
+        self.build()
+        self.prep_debug_monitor_and_inferior(
+            inferior_args=["fork",
+                           "thread:new",
+                           "trap",
+                           ])
+        self.add_qSupported_packets(["multiprocess+",
+                                     "fork-events+"])
+        ret = self.expect_gdbremote_sequence()
+        self.assertIn("fork-events+", ret["qSupported_response"])
+        self.reset_test_sequence()
+
+        # continue and expect fork
+        self.test_sequence.add_log_lines([
+            "read packet: $c#00",
+            {"direction": "send", "regex": self.fork_regex.format("fork"),
+             "capture": self.fork_capture},
+        ], True)
+        self.add_threadinfo_collection_packets()
+        ret = self.expect_gdbremote_sequence()
+        pidtids = [
+            (ret["parent_pid"], ret["parent_tid"]),
+            (ret["child_pid"], ret["child_tid"]),
+        ]
+        self.reset_test_sequence()
+
+        for pidtid in pidtids:
+            self.test_sequence.add_log_lines(
+                ["read packet: $Hcp{}.{}#00".format(*pidtid),
+                 "send packet: $OK#00",
+                 "read packet: $c#00",
+                 {"direction": "send",
+                  "regex": "^[$]T05thread:p{}.{}.*".format(*pidtid),
+                  },
+                 ], True)
+
+        self.add_threadinfo_collection_packets()
+        ret = self.expect_gdbremote_sequence()
+        self.reset_test_sequence()
+
+        pidtids = set(self.parse_threadinfo_packets(ret))
+        self.assertEqual(len(pidtids), 4)
+        # first, save register values from all the threads
+        thread_regs = {}
+        for pidtid in pidtids:
+            for regno in range(256):
+                self.test_sequence.add_log_lines(
+                    ["read packet: $Hgp{:x}.{:x}#00".format(*pidtid),
+                     "send packet: $OK#00",
+                     "read packet: $p{:x}#00".format(regno),
+                     {"direction": "send",
+                      "regex": r"^[$](.+)#.*$",
+                      "capture": {1: "data"}},
+                     ], True)
+                ret = self.expect_gdbremote_sequence()
+                data = ret.get("data")
+                self.assertIsNotNone(data)
+                # ignore registers shorter than 32 bits (this also catches
+                # "Exx" errors)
+                if len(data) >= 8:
+                    break
+            else:
+                self.skipTest("no usable register found")
+            thread_regs[pidtid] = (regno, data)
+
+        vals = set(x[1] for x in thread_regs.values())
+        # NB: cheap hack to make the loop below easier
+        new_val = next(iter(vals))
+
+        # then, start altering them and verify that we don't unexpectedly
+        # change the value from another thread
+        for pidtid in pidtids:
+            old_val = thread_regs[pidtid]
+            regno = old_val[0]
+            old_val_length = len(old_val[1])
+            # generate a unique new_val
+            while new_val in vals:
+                new_val = ('{{:0{}x}}'.format(old_val_length)
+                           .format(random.getrandbits(old_val_length*4)))
+            vals.add(new_val)
+
+            self.test_sequence.add_log_lines(
+                ["read packet: $Hgp{:x}.{:x}#00".format(*pidtid),
+                 "send packet: $OK#00",
+                 "read packet: $p{:x}#00".format(regno),
+                 {"direction": "send",
+                  "regex": r"^[$](.+)#.*$",
+                  "capture": {1: "data"}},
+                 "read packet: $P{:x}={}#00".format(regno, new_val),
+                 "send packet: $OK#00",
+                 ], True)
+            ret = self.expect_gdbremote_sequence()
+            data = ret.get("data")
+            self.assertIsNotNone(data)
+            self.assertEqual(data, old_val[1])
+            thread_regs[pidtid] = (regno, new_val)
+
+        # finally, verify that new values took effect
+        for pidtid in pidtids:
+            old_val = thread_regs[pidtid]
+            self.test_sequence.add_log_lines(
+                ["read packet: $Hgp{:x}.{:x}#00".format(*pidtid),
+                 "send packet: $OK#00",
+                 "read packet: $p{:x}#00".format(old_val[0]),
+                 {"direction": "send",
+                  "regex": r"^[$](.+)#.*$",
+                  "capture": {1: "data"}},
+                 ], True)
+            ret = self.expect_gdbremote_sequence()
+            data = ret.get("data")
+            self.assertIsNotNone(data)
+            self.assertEqual(data, old_val[1])