debug: Create CeaseMultiTest. (#436)

Confirm basic debug still works when other harts have been parked using
a `cease` instruction. Check that the unavailable harts are inaccessible
from gdb.

Add Gdb.expect()
Parse "unknown thread" error from gdb.

2 files changed, 55 insertions, 2 deletions

diff --git a/debug/gdbserver.py b/debug/gdbserver.py
index ac7b162..6d7c197 100755
--- a/debug/gdbserver.py
+++ b/debug/gdbserver.py
@@ -12,10 +12,12 @@ import re
 
 import targets
 import testlib
-from testlib import assertEqual, assertNotEqual, assertIn, assertNotIn
+from testlib import assertEqual, assertNotEqual
+from testlib import assertIn, assertNotIn
 from testlib import assertGreater, assertRegex, assertLess
 from testlib import GdbTest, GdbSingleHartTest, TestFailed
 from testlib import TestNotApplicable, CompileError
+from testlib import UnknownThread
 
 MSTATUS_UIE = 0x00000001
 MSTATUS_SIE = 0x00000002
@@ -1788,6 +1790,49 @@ class EbreakTest(GdbSingleHartTest):
         output = self.gdb.c()
         assertIn("_exit", output)
 
+class CeaseMultiTest(GdbTest):
+    """Test that we work correctly when a hart ceases to respond (e.g. because
+    it's powered down)."""
+    compile_args = ("programs/counting_loop.c", "-DDEFINE_MALLOC",
+            "-DDEFINE_FREE")
+
+    def early_applicable(self):
+        return self.hart.support_cease and len(self.target.harts) > 1
+
+    def setup(self):
+        ProgramTest.setup(self)
+        self.parkOtherHarts("precease")
+
+    def test(self):
+        # Run all the way to the infinite loop in exit
+        self.gdb.c(wait=False)
+        self.gdb.expect(r"\S+ became unavailable.")
+        self.gdb.interrupt()
+
+        for hart in self.target.harts:
+            # Try to read misa on the ceased harts
+            if hart != self.hart:
+                try:
+                    self.gdb.select_hart(hart)
+                    self.gdb.p("$misa")
+                    assert False, \
+                        "Shouldn't be able to access unavailable hart."
+                except UnknownThread:
+                    pass
+
+        # Check that the main hart can still be debugged.
+        self.gdb.select_hart(self.hart)
+        main_addr = self.gdb.p("$pc=main")
+        self.gdb.stepi()
+        # Assume the first instruction of main is not a jump.
+        pc = self.gdb.p("$pc")
+        assertGreater(pc, main_addr)
+        assertLess(pc, main_addr + 8)
+
+        self.gdb.p("$pc=_start")
+
+        self.exit()
+
 class FreeRtosTest(GdbTest):
     def early_applicable(self):
         return self.target.freertos_binary
diff --git a/debug/testlib.py b/debug/testlib.py
index 27deb3a..5c78bd7 100644
--- a/debug/testlib.py
+++ b/debug/testlib.py
@@ -484,6 +484,10 @@ class NoSymbol(Exception):
     def __repr__(self):
         return f"NoSymbol({self.symbol!r})"
 
+class UnknownThread(Exception):
+    def __init__(self, explanation):
+        Exception.__init__(self, explanation)
+
 Thread = collections.namedtuple('Thread', ('id', 'description', 'target_id',
     'name', 'frame'))
 
@@ -675,7 +679,8 @@ class Gdb:
         self.select_child(h['child'])
         if not h['solo']:
             output = self.command(f"thread {h['thread'].id}", ops=5)
-            assert "Unknown" not in output
+            if "Unknown" in output:
+                raise UnknownThread(output)
 
     def push_state(self):
         self.stack.append({
@@ -856,6 +861,9 @@ class Gdb:
         output = self.command("stepi", ops=10)
         return output
 
+    def expect(self, text, ops=1):
+        return self.active_child.expect(text, timeout=ops * self.timeout)
+
     def load(self):
         output = self.system_command("load", ops=1000)
         assert "failed" not in  output