Reland "[lldb-dap] Improving consistency of tests by removing concurrency." (#165688)"

This reverts commit f205be0.

This new select mechanism has exposed the fact that the resources
the Arm Linux bot has can vary a lot. We do limit it to a low number
of parallel tests but in this case, I think it's write performance
somewhere.

Reland the changes since they work elsewhere, and disable lldb-dap
tests on Arm Linux while I fix our buildbot.

Author: DavidSpickett

lldb/packages/Python/lldbsuite/test/tools/lldb-dap/dap_server.py

@@ -10,8 +10,8 @@
 import subprocess
 import signal
 import sys
-import threading
 import warnings
+import selectors
 import time
 from typing import (
     Any,
@@ -139,35 +139,6 @@ def dump_memory(base_addr, data, num_per_line, outfile):
         outfile.write("\n")
 
 
-def read_packet(
-    f: IO[bytes], trace_file: Optional[IO[str]] = None
-) -> Optional[ProtocolMessage]:
-    """Decode a JSON packet that starts with the content length and is
-    followed by the JSON bytes from a file 'f'. Returns None on EOF.
-    """
-    line = f.readline().decode("utf-8")
-    if len(line) == 0:
-        return None  # EOF.
-
-    # Watch for line that starts with the prefix
-    prefix = "Content-Length: "
-    if line.startswith(prefix):
-        # Decode length of JSON bytes
-        length = int(line[len(prefix) :])
-        # Skip empty line
-        separator = f.readline().decode()
-        if separator != "":
-            Exception("malformed DAP content header, unexpected line: " + separator)
-        # Read JSON bytes
-        json_str = f.read(length).decode()
-        if trace_file:
-            trace_file.write("from adapter:\n%s\n" % (json_str))
-        # Decode the JSON bytes into a python dictionary
-        return json.loads(json_str)
-
-    raise Exception("unexpected malformed message from lldb-dap: " + line)
-
-
 def packet_type_is(packet, packet_type):
     return "type" in packet and packet["type"] == packet_type
 
@@ -199,16 +170,8 @@ def __init__(
         self.log_file = log_file
         self.send = send
         self.recv = recv
-
-        # Packets that have been received and processed but have not yet been
-        # requested by a test case.
-        self._pending_packets: List[Optional[ProtocolMessage]] = []
-        # Received packets that have not yet been processed.
-        self._recv_packets: List[Optional[ProtocolMessage]] = []
-        # Used as a mutex for _recv_packets and for notify when _recv_packets
-        # changes.
-        self._recv_condition = threading.Condition()
-        self._recv_thread = threading.Thread(target=self._read_packet_thread)
+        self.selector = selectors.DefaultSelector()
+        self.selector.register(recv, selectors.EVENT_READ)
 
         # session state
         self.init_commands = init_commands
@@ -234,9 +197,6 @@ def __init__(
         # keyed by breakpoint id
         self.resolved_breakpoints: dict[str, Breakpoint] = {}
 
-        # trigger enqueue thread
-        self._recv_thread.start()
-
     @classmethod
     def encode_content(cls, s: str) -> bytes:
         return ("Content-Length: %u\r\n\r\n%s" % (len(s), s)).encode("utf-8")
@@ -252,17 +212,46 @@ def validate_response(cls, command, response):
                 f"seq mismatch in response {command['seq']} != {response['request_seq']}"
             )
 
-    def _read_packet_thread(self):
-        try:
-            while True:
-                packet = read_packet(self.recv, trace_file=self.trace_file)
-                # `packet` will be `None` on EOF. We want to pass it down to
-                # handle_recv_packet anyway so the main thread can handle unexpected
-                # termination of lldb-dap and stop waiting for new packets.
-                if not self._handle_recv_packet(packet):
-                    break
-        finally:
-            dump_dap_log(self.log_file)
+    def _read_packet(
+        self,
+        timeout: float = DEFAULT_TIMEOUT,
+    ) -> Optional[ProtocolMessage]:
+        """Decode a JSON packet that starts with the content length and is
+        followed by the JSON bytes from self.recv. Returns None on EOF.
+        """
+
+        ready = self.selector.select(timeout)
+        if not ready:
+            warnings.warn(
+                "timeout occurred waiting for a packet, check if the test has a"
+                " negative assertion and see if it can be inverted.",
+                stacklevel=4,
+            )
+            return None  # timeout
+
+        line = self.recv.readline().decode("utf-8")
+        if len(line) == 0:
+            return None  # EOF.
+
+        # Watch for line that starts with the prefix
+        prefix = "Content-Length: "
+        if line.startswith(prefix):
+            # Decode length of JSON bytes
+            length = int(line[len(prefix) :])
+            # Skip empty line
+            separator = self.recv.readline().decode()
+            if separator != "":
+                Exception("malformed DAP content header, unexpected line: " + separator)
+            # Read JSON bytes
+            json_str = self.recv.read(length).decode()
+            if self.trace_file:
+                self.trace_file.write(
+                    "%s from adapter:\n%s\n" % (time.time(), json_str)
+                )
+            # Decode the JSON bytes into a python dictionary
+            return json.loads(json_str)
+
+        raise Exception("unexpected malformed message from lldb-dap: " + line)
 
     def get_modules(
         self, start_module: Optional[int] = None, module_count: Optional[int] = None
@@ -310,34 +299,6 @@ def collect_output(
             output += self.get_output(category, clear=clear)
         return output
 
-    def _enqueue_recv_packet(self, packet: Optional[ProtocolMessage]):
-        with self.recv_condition:
-            self.recv_packets.append(packet)
-            self.recv_condition.notify()
-
-    def _handle_recv_packet(self, packet: Optional[ProtocolMessage]) -> bool:
-        """Handles an incoming packet.
-
-        Called by the read thread that is waiting for all incoming packets
-        to store the incoming packet in "self._recv_packets" in a thread safe
-        way. This function will then signal the "self._recv_condition" to
-        indicate a new packet is available.
-
-        Args:
-            packet: A new packet to store.
-
-        Returns:
-            True if the caller should keep calling this function for more
-            packets.
-        """
-        with self._recv_condition:
-            self._recv_packets.append(packet)
-            self._recv_condition.notify()
-            # packet is None on EOF
-            return packet is not None and not (
-                packet["type"] == "response" and packet["command"] == "disconnect"
-            )
-
     def _recv_packet(
         self,
         *,
@@ -361,46 +322,34 @@ def _recv_packet(
             The first matching packet for the given predicate, if specified,
             otherwise None.
         """
-        assert (
-            threading.current_thread != self._recv_thread
-        ), "Must not be called from the _recv_thread"
-
-        def process_until_match():
-            self._process_recv_packets()
-            for i, packet in enumerate(self._pending_packets):
-                if packet is None:
-                    # We need to return a truthy value to break out of the
-                    # wait_for, use `EOFError` as an indicator of EOF.
-                    return EOFError()
-                if predicate and predicate(packet):
-                    self._pending_packets.pop(i)
-                    return packet
-
-        with self._recv_condition:
-            packet = self._recv_condition.wait_for(process_until_match, timeout)
-            return None if isinstance(packet, EOFError) else packet
-
-    def _process_recv_packets(self) -> None:
+        deadline = time.time() + timeout
+
+        while time.time() < deadline:
+            packet = self._read_packet(timeout=deadline - time.time())
+            if packet is None:
+                return None
+            self._process_recv_packet(packet)
+            if not predicate or predicate(packet):
+                return packet
+
+    def _process_recv_packet(self, packet) -> None:
         """Process received packets, updating the session state."""
-        with self._recv_condition:
-            for packet in self._recv_packets:
-                if packet and ("seq" not in packet or packet["seq"] == 0):
-                    warnings.warn(
-                        f"received a malformed packet, expected 'seq != 0' for {packet!r}"
-                    )
-                # Handle events that may modify any stateful properties of
-                # the DAP session.
-                if packet and packet["type"] == "event":
-                    self._handle_event(packet)
-                elif packet and packet["type"] == "request":
-                    # Handle reverse requests and keep processing.
-                    self._handle_reverse_request(packet)
-                # Move the packet to the pending queue.
-                self._pending_packets.append(packet)
-            self._recv_packets.clear()
+        if packet and ("seq" not in packet or packet["seq"] == 0):
+            warnings.warn(
+                f"received a malformed packet, expected 'seq != 0' for {packet!r}"
+            )
+        # Handle events that may modify any stateful properties of
+        # the DAP session.
+        if packet and packet["type"] == "event":
+            self._handle_event(packet)
+        elif packet and packet["type"] == "request":
+            # Handle reverse requests and keep processing.
+            self._handle_reverse_request(packet)
 
     def _handle_event(self, packet: Event) -> None:
         """Handle any events that modify debug session state we track."""
+        self.events.append(packet)
+
         event = packet["event"]
         body: Optional[Dict] = packet.get("body", None)
 
@@ -453,6 +402,8 @@ def _handle_event(self, packet: Event) -> None:
             self.invalidated_event = packet
         elif event == "memory":
             self.memory_event = packet
+        elif event == "module":
+            self.module_events.append(packet)
 
     def _handle_reverse_request(self, request: Request) -> None:
         if request in self.reverse_requests:
@@ -521,18 +472,14 @@ def send_packet(self, packet: ProtocolMessage) -> int:
 
         Returns the seq number of the request.
         """
-        # Set the seq for requests.
-        if packet["type"] == "request":
-            packet["seq"] = self.sequence
-            self.sequence += 1
-        else:
-            packet["seq"] = 0
+        packet["seq"] = self.sequence
+        self.sequence += 1
 
         # Encode our command dictionary as a JSON string
         json_str = json.dumps(packet, separators=(",", ":"))
 
         if self.trace_file:
-            self.trace_file.write("to adapter:\n%s\n" % (json_str))
+            self.trace_file.write("%s to adapter:\n%s\n" % (time.time(), json_str))
 
         length = len(json_str)
         if length > 0:
@@ -913,6 +860,8 @@ def request_restart(self, restartArguments=None):
         if restartArguments:
             command_dict["arguments"] = restartArguments
 
+        # Clear state, the process is about to restart...
+        self._process_continued(True)
         response = self._send_recv(command_dict)
         # Caller must still call wait_for_stopped.
         return response
@@ -1479,8 +1428,10 @@ def request_testGetTargetBreakpoints(self):
 
     def terminate(self):
         self.send.close()
-        if self._recv_thread.is_alive():
-            self._recv_thread.join()
+        self.recv.close()
+        self.selector.close()
+        if self.log_file:
+            dump_dap_log(self.log_file)
 
     def request_setInstructionBreakpoints(self, memory_reference=[]):
         breakpoints = []
@@ -1577,6 +1528,7 @@ def launch(
             stdout=subprocess.PIPE,
             stderr=sys.stderr,
             env=adapter_env,
+            bufsize=0,
         )
 
         if connection is None:

lldb/packages/Python/lldbsuite/test/tools/lldb-dap/lldbdap_testcase.py

@@ -15,6 +15,8 @@
 # DAP tests as a whole have been flakey on the Windows on Arm bot. See:
 # https://github.com/llvm/llvm-project/issues/137660
 @skipIf(oslist=["windows"], archs=["aarch64"])
+# The Arm Linux bot needs stable resources before it can run these tests reliably.
+@skipif(oslist=["linux]"], archs=["arm$"])
 class DAPTestCaseBase(TestBase):
     # set timeout based on whether ASAN was enabled or not. Increase
     # timeout by a factor of 10 if ASAN is enabled.
@@ -416,7 +418,7 @@ def continue_to_next_stop(self):
         return self.dap_server.wait_for_stopped()
 
     def continue_to_breakpoint(self, breakpoint_id: str):
-        self.continue_to_breakpoints((breakpoint_id))
+        self.continue_to_breakpoints([breakpoint_id])
 
     def continue_to_breakpoints(self, breakpoint_ids):
         self.do_continue()

lldb/test/API/tools/lldb-dap/breakpoint-events/TestDAP_breakpointEvents.py

@@ -81,24 +81,20 @@ def test_breakpoint_events(self):
                 breakpoint["verified"], "expect foo breakpoint to not be verified"
             )
 
-        # Flush the breakpoint events.
-        self.dap_server.wait_for_breakpoint_events()
-
         # Continue to the breakpoint
-        self.continue_to_breakpoints(dap_breakpoint_ids)
+        self.continue_to_breakpoint(foo_bp_id)
+        self.continue_to_next_stop()  # foo_bp2
+        self.continue_to_breakpoint(main_bp_id)
+        self.continue_to_exit()
 
-        verified_breakpoint_ids = []
-        unverified_breakpoint_ids = []
-        for breakpoint_event in self.dap_server.wait_for_breakpoint_events():
-            breakpoint = breakpoint_event["body"]["breakpoint"]
-            id = breakpoint["id"]
-            if breakpoint["verified"]:
-                verified_breakpoint_ids.append(id)
-            else:
-                unverified_breakpoint_ids.append(id)
+        bp_events = [e for e in self.dap_server.events if e["event"] == "breakpoint"]
 
-        self.assertIn(main_bp_id, unverified_breakpoint_ids)
-        self.assertIn(foo_bp_id, unverified_breakpoint_ids)
+        main_bp_events = [
+            e for e in bp_events if e["body"]["breakpoint"]["id"] == main_bp_id
+        ]
+        foo_bp_events = [
+            e for e in bp_events if e["body"]["breakpoint"]["id"] == foo_bp_id
+        ]
 
-        self.assertIn(main_bp_id, verified_breakpoint_ids)
-        self.assertIn(foo_bp_id, verified_breakpoint_ids)
+        self.assertTrue(main_bp_events)
+        self.assertTrue(foo_bp_events)

lldb/test/API/tools/lldb-dap/launch/TestDAP_launch.py

@@ -156,6 +156,7 @@ def test_debuggerRoot(self):
         self.build_and_launch(
             program, debuggerRoot=program_parent_dir, initCommands=commands
         )
+        self.continue_to_exit()
         output = self.get_console()
         self.assertTrue(output and len(output) > 0, "expect console output")
         lines = output.splitlines()
@@ -171,7 +172,6 @@ def test_debuggerRoot(self):
                     % (program_parent_dir, line[len(prefix) :]),
                 )
         self.assertTrue(found, "verified lldb-dap working directory")
-        self.continue_to_exit()
 
     def test_sourcePath(self):
         """