[lldb-dap] Refactoring DebugCommunication to improve test consistency. #143818
+590
−397
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
In DebugCommunication, we currently are using 2 thread to drive lldb-dap. At the moment, they make an attempt at only synchronizing the `recv_packets` between the reader thread and the main test thread. Other stateful properties of the debug session are not guarded by any mutexs. To mitigate this, I am moving any state updates to the main thread inside the `_recv_packet` method to ensure that between calls to `_recv_packet` the state does not change out from under us in a test. This does mean the precise timing of events has changed slightly as a result and I've updated the existing tests that fail for me locally with this new behavior. I think this should result in overall more predictable behavior, even if the test is slow due to the host workload or architecture differences.
|
@llvm/pr-subscribers-lldb Author: John Harrison (ashgti) ChangesIn DebugCommunication, we currently are using 2 thread to drive lldb-dap. At the moment, they make an attempt at only synchronizing the To mitigate this, I am moving any state updates to the main thread inside the This does mean the precise timing of events has changed slightly as a result and I've updated the existing tests that fail for me locally with this new behavior. I think this should result in overall more predictable behavior, even if the test is slow due to the host workload or architecture differences. Patch is 63.57 KiB, truncated to 20.00 KiB below, full version: https://github.com/llvm/llvm-project/pull/143818.diff 7 Files Affected:
diff --git a/lldb/packages/Python/lldbsuite/test/tools/lldb-dap/dap_server.py b/lldb/packages/Python/lldbsuite/test/tools/lldb-dap/dap_server.py
index 9786678aa53f9..20a1b4480df6d 100644
--- a/lldb/packages/Python/lldbsuite/test/tools/lldb-dap/dap_server.py
+++ b/lldb/packages/Python/lldbsuite/test/tools/lldb-dap/dap_server.py
@@ -10,17 +10,117 @@
import subprocess
import signal
import sys
+from dataclasses import dataclass
import threading
import time
-from typing import Any, Optional, Union, BinaryIO, TextIO
+from typing import (
+ IO,
+ Any,
+ Callable,
+ Dict,
+ List,
+ Optional,
+ Tuple,
+ TypeGuard,
+ TypeVar,
+ TypedDict,
+ Union,
+ BinaryIO,
+ TextIO,
+ Literal,
+ cast,
+)
## DAP type references
-Event = dict[str, Any]
-Request = dict[str, Any]
-Response = dict[str, Any]
+
+T = TypeVar("T")
+
+
+class Event(TypedDict):
+ type: Literal["event"]
+ seq: Literal[0]
+ event: str
+ body: Optional[dict]
+
+
+class Request(TypedDict):
+ type: Literal["request"]
+ seq: int
+ command: str
+ arguments: Optional[dict]
+
+
+class Response(TypedDict):
+ type: Literal["response"]
+ seq: Literal[0]
+ request_seq: int
+ success: bool
+ command: str
+ message: Optional[str]
+ body: Optional[dict]
+
+
ProtocolMessage = Union[Event, Request, Response]
+class AttachOrLaunchArguments(TypedDict, total=False):
+ stopOnEntry: bool
+ disableASLR: bool
+ disableSTDIO: bool
+ enableAutoVariableSummaries: bool
+ displayExtendedBacktrace: bool
+ enableSyntheticChildDebugging: bool
+ initCommands: List[str]
+ preRunCommands: List[str]
+ postRunCommands: List[str]
+ stopCommands: List[str]
+ exitCommands: List[str]
+ terminateCommands: List[str]
+ sourceMap: Union[List[Tuple[str, str]], Dict[str, str]]
+ sourcePath: str
+ debuggerRoot: str
+ commandEscapePrefix: str
+ customFrameFormat: str
+ customThreadFormat: str
+
+
+class LaunchArguments(AttachOrLaunchArguments, total=False):
+ program: str
+ args: List[str]
+ cwd: str
+ env: Dict[str, str]
+ shellExpandArguments: bool
+ runInTerminal: bool
+ launchCommands: List[str]
+
+
+class AttachArguments(AttachOrLaunchArguments, total=False):
+ program: str
+ pid: int
+ waitFor: bool
+ attachCommands: List[str]
+ coreFile: str
+ gdbRemotePort: int
+ gdbRemoteHostname: str
+
+
+class BreakpiontData(TypedDict, total=False):
+ column: int
+ condition: str
+ hitCondition: str
+ logMessage: str
+ mode: str
+
+
+class SourceBreakpoint(BreakpiontData):
+ line: int
+
+
+class Breakpoint(TypedDict, total=False):
+ id: int
+ verified: bool
+
+
def dump_memory(base_addr, data, num_per_line, outfile):
data_len = len(data)
hex_string = binascii.hexlify(data)
@@ -58,7 +158,9 @@ def dump_memory(base_addr, data, num_per_line, outfile):
outfile.write("\n")
-def read_packet(f, verbose=False, trace_file=None):
+def read_packet(
+ f: IO[bytes], verbose: bool = False, 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.
"""
@@ -76,26 +178,22 @@ def read_packet(f, verbose=False, trace_file=None):
if verbose:
print('length: "%u"' % (length))
# Skip empty line
- line = f.readline()
+ line = f.readline().decode()
if verbose:
print('empty: "%s"' % (line))
# Read JSON bytes
json_str = f.read(length)
if verbose:
- print('json: "%s"' % (json_str))
+ print('json: "%r"' % (json_str))
if trace_file:
- trace_file.write("from adapter:\n%s\n" % (json_str))
+ trace_file.write("from adapter:\n%r\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
-
-
-def dump_dap_log(log_file):
+def dump_dap_log(log_file: Optional[str]) -> None:
print("========= DEBUG ADAPTER PROTOCOL LOGS =========", file=sys.stderr)
if log_file is None:
print("no log file available", file=sys.stderr)
@@ -105,34 +203,30 @@ def dump_dap_log(log_file):
print("========= END =========", file=sys.stderr)
-class Source(object):
+@dataclass
+class Source:
+ path: Optional[str]
+ source_reference: Optional[int]
+
+ @property
+ def name(self) -> Optional[str]:
+ if not self.path:
+ return None
+ return os.path.basename(self.path)
+
def __init__(
self, path: Optional[str] = None, source_reference: Optional[int] = None
):
- self._name = None
- self._path = None
- self._source_reference = None
-
- if path is not None:
- self._name = os.path.basename(path)
- self._path = path
- elif source_reference is not None:
- self._source_reference = source_reference
- else:
- raise ValueError("Either path or source_reference must be provided")
+ self.path = path
+ self.source_reference = source_reference
- def __str__(self):
- return f"Source(name={self.name}, path={self.path}), source_reference={self.source_reference})"
+ if path is None and source_reference is None:
+ raise ValueError("Either path or source_reference must be provided")
- def as_dict(self):
- source_dict = {}
- if self._name is not None:
- source_dict["name"] = self._name
- if self._path is not None:
- source_dict["path"] = self._path
- if self._source_reference is not None:
- source_dict["sourceReference"] = self._source_reference
- return source_dict
+ def to_DAP(self) -> dict:
+ if self.path:
+ return {"path": self.path, "name": self.name}
+ return {"sourceReference": self.source_reference}
class NotSupportedError(KeyError):
@@ -152,35 +246,50 @@ def __init__(
self.log_file = log_file
self.send = send
self.recv = recv
- self.recv_packets: list[Optional[ProtocolMessage]] = []
- self.recv_condition = threading.Condition()
- self.recv_thread = threading.Thread(target=self._read_packet_thread)
- self.process_event_body = None
- self.exit_status: Optional[int] = None
- self.capabilities: dict[str, Any] = {}
- self.progress_events: list[Event] = []
- self.reverse_requests = []
- self.sequence = 1
- self.threads = None
- self.thread_stop_reasons = {}
- self.recv_thread.start()
- self.output_condition = threading.Condition()
- self.output: dict[str, list[str]] = {}
- self.configuration_done_sent = False
- self.initialized = False
- self.frame_scopes = {}
+ # Packets that have been received and processed but have not yet been
+ # requested by a test case.
+ self._pending_packets: List[Optional[ProtocolMessage]] = []
+ # Recieved 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)
+
+ # session state
self.init_commands = init_commands
- self.resolved_breakpoints = {}
+ self.exit_status: Optional[int] = None
+ self.capabilities: Optional[Dict] = None
+ self.initialized: bool = False
+ self.configuration_done_sent: bool = False
+ self.process_event_body: Optional[Dict] = None
+ self.terminated: bool = False
+ self.events: List[Event] = []
+ self.progress_events: List[Event] = []
+ self.reverse_requests: List[Request] = []
+ self.module_events: List[Dict] = []
+ self.sequence: int = 1
+ self.output: Dict[str, str] = {}
+
+ # debuggee state
+ self.threads: Optional[dict] = None
+ self.thread_stop_reasons: Dict[str, Any] = {}
+ self.frame_scopes: Dict[str, Any] = {}
+ # keyed by breakpoint id
+ self.resolved_breakpoints: Dict[int, bool] = {}
+
+ # 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")
@classmethod
- def validate_response(cls, command, response):
- if command["command"] != response["command"]:
+ def validate_response(cls, request: Request, response: Response) -> None:
+ if request["command"] != response["command"]:
raise ValueError("command mismatch in response")
- if command["seq"] != response["request_seq"]:
+ if request["seq"] != response["request_seq"]:
raise ValueError("seq mismatch in response")
def _read_packet_thread(self):
@@ -189,262 +298,322 @@ def _read_packet_thread(self):
while not done:
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.
+ # handle_recv_packet anyway so the main thread can handle
+ # unexpected termination of lldb-dap and stop waiting for new
+ # packets.
done = not self._handle_recv_packet(packet)
finally:
dump_dap_log(self.log_file)
- def get_modules(self):
- module_list = self.request_modules()["body"]["modules"]
- modules = {}
- for module in module_list:
- modules[module["name"]] = module
- return modules
+ def _handle_recv_packet(self, packet: Optional[ProtocolMessage]) -> bool:
+ """Handles an incoming packet.
- def get_output(self, category, timeout=0.0, clear=True):
- self.output_condition.acquire()
- output = None
- if category in self.output:
- output = self.output[category]
- if clear:
- del self.output[category]
- elif timeout != 0.0:
- self.output_condition.wait(timeout)
- if category in self.output:
- output = self.output[category]
- if clear:
- del self.output[category]
- self.output_condition.release()
- return output
+ 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.
- def collect_output(self, category, timeout_secs, pattern, clear=True):
- end_time = time.time() + timeout_secs
- collected_output = ""
- while end_time > time.time():
- output = self.get_output(category, timeout=0.25, clear=clear)
- if output:
- collected_output += output
- if pattern is not None and pattern in output:
- break
- return collected_output if collected_output else None
-
- def _enqueue_recv_packet(self, packet: Optional[ProtocolMessage]):
- self.recv_condition.acquire()
- self.recv_packets.append(packet)
- self.recv_condition.notify()
- self.recv_condition.release()
+ Args:
+ packet: A new packet to store.
- def _handle_recv_packet(self, packet: Optional[ProtocolMessage]) -> bool:
- """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. Returns True if the caller
- should keep calling this function for more packets.
+ Returns:
+ True if the caller should keep calling this function for more
+ packets.
"""
- # If EOF, notify the read thread by enqueuing a None.
- if not packet:
- self._enqueue_recv_packet(None)
- return False
-
- # Check the packet to see if is an event packet
- keepGoing = True
- packet_type = packet["type"]
- if packet_type == "event":
- event = packet["event"]
- body = None
- if "body" in packet:
- body = packet["body"]
- # Handle the event packet and cache information from these packets
- # as they come in
- if event == "output":
- # Store any output we receive so clients can retrieve it later.
- category = body["category"]
- output = body["output"]
- self.output_condition.acquire()
- if category in self.output:
- self.output[category] += output
- else:
- self.output[category] = output
- self.output_condition.notify()
- self.output_condition.release()
- # no need to add 'output' event packets to our packets list
- return keepGoing
- elif event == "initialized":
- self.initialized = True
- elif event == "process":
- # When a new process is attached or launched, remember the
- # details that are available in the body of the event
- self.process_event_body = body
- elif event == "exited":
- # Process exited, mark the status to indicate the process is not
- # alive.
- self.exit_status = body["exitCode"]
- elif event == "continued":
- # When the process continues, clear the known threads and
- # thread_stop_reasons.
- all_threads_continued = body.get("allThreadsContinued", True)
- tid = body["threadId"]
- if tid in self.thread_stop_reasons:
- del self.thread_stop_reasons[tid]
- self._process_continued(all_threads_continued)
- elif event == "stopped":
- # Each thread that stops with a reason will send a
- # 'stopped' event. We need to remember the thread stop
- # reasons since the 'threads' command doesn't return
- # that information.
- self._process_stopped()
- tid = body["threadId"]
- self.thread_stop_reasons[tid] = body
- elif event.startswith("progress"):
- # Progress events come in as 'progressStart', 'progressUpdate',
- # and 'progressEnd' events. Keep these around in case test
- # cases want to verify them.
- self.progress_events.append(packet)
- elif event == "breakpoint":
- # Breakpoint events are sent when a breakpoint is resolved
- self._update_verified_breakpoints([body["breakpoint"]])
- elif event == "capabilities":
- # Update the capabilities with new ones from the event.
- self.capabilities.update(body["capabilities"])
-
- elif packet_type == "response":
- if packet["command"] == "disconnect":
- keepGoing = False
- self._enqueue_recv_packet(packet)
- return keepGoing
+ 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,
+ *,
+ predicate: Optional[Callable[[ProtocolMessage], bool]] = None,
+ timeout: Optional[float] = None,
+ ) -> Optional[ProtocolMessage]:
+ """Processes recived packets from the adapter.
+
+ Updates the DebugCommunication stateful properties based on the received
+ packets in the order they are recieved.
+
+ NOTE: The only time the session state properties should be updated is
+ during this call to ensure consistency during tests.
+
+ Args:
+ predicate:
+ Optional, if specified, returns the first packet that matches
+ the given predicate.
+ timeout:
+ Optional, if specified, processes packets until either the
+ timeout occurs or the predicate matches a packet, whichever
+ occurs first.
+
+ Returns:
+ 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:
+ """Process received packets, updating the session state."""
+ with self._recv_condition:
+ for packet in self._recv_packets:
+ # 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()
+
+ def _handle_event(self, packet: Event) -> None:
+ """Handle any events that modify debug session state we track."""
+ event = packet["event"]
+ body: Optional[Dict] = packet.get("body", None)
+
+ if event == "output":
+ # Store any output we receive so clients can retrieve it later.
+ category = body["category"]
+ output = body["output"]
+ if category in self.output:
+ self.output[category] += output
+ else:
+ self.output[category] = output
+ elif event == "initialized":
+ self.initialized = True
+ elif event == "process":
+ # When a new process is attached or launched, remember the
+ # details that are available in the body of the event
+ self.process_event_body = body
+ elif event == "exited":
+ # Process exited, mark the status to indicate the process is not
+ # alive.
+ self.exit_status = body["exitCode"]
+ elif event == "continued":
+ # When the process continues, clear the known threads and
+ # thread_stop_reasons.
+ all_threads_continued = (
+ body.get("allThreadsContinued", True) if body else True
+ )
+ tid = body["threadId"]
+ if tid in self.thread_stop_reasons:
+ del self.thread_stop_reasons[tid]
+ self._process_continued(all_threads_contin...
[truncated]
|
lldb/packages/Python/lldbsuite/test/tools/lldb-dap/dap_server.py
Outdated
Show resolved
Hide resolved
da-viper
reviewed
Jun 16, 2025
lldb/packages/Python/lldbsuite/test/tools/lldb-dap/dap_server.py
Outdated
Show resolved
Hide resolved
lldb/packages/Python/lldbsuite/test/tools/lldb-dap/dap_server.py
Outdated
Show resolved
Hide resolved
lldb/packages/Python/lldbsuite/test/tools/lldb-dap/dap_server.py
Outdated
Show resolved
Hide resolved
lldb/packages/Python/lldbsuite/test/tools/lldb-dap/dap_server.py
Outdated
Show resolved
Hide resolved
lldb/packages/Python/lldbsuite/test/tools/lldb-dap/dap_server.py
Outdated
Show resolved
Hide resolved
lldb/packages/Python/lldbsuite/test/tools/lldb-dap/dap_server.py
Outdated
Show resolved
Hide resolved
lldb/packages/Python/lldbsuite/test/tools/lldb-dap/dap_server.py
Outdated
Show resolved
Hide resolved
lldb/packages/Python/lldbsuite/test/tools/lldb-dap/dap_server.py
Outdated
Show resolved
Hide resolved
lldb/packages/Python/lldbsuite/test/tools/lldb-dap/dap_server.py
Outdated
Show resolved
Hide resolved
Co-authored-by: Ebuka Ezike <yerimyah1@gmail.com>
|
LLVM Buildbot has detected a new failure on builder Full details are available at: https://lab.llvm.org/buildbot/#/builders/18/builds/17628 Here is the relevant piece of the build log for the reference |
|
LLVM Buildbot has detected a new failure on builder Full details are available at: https://lab.llvm.org/buildbot/#/builders/59/builds/19528 Here is the relevant piece of the build log for the reference |
|
Reverting this as it looks like python3.10 is giving an error with one of the TypedDict usages from this patch. |
ashgti
added a commit
to ashgti/llvm-project
that referenced
this pull request
Jun 17, 2025
…sistency. (llvm#143818)" This reverts commit 362b9d7. Buildbots using python3.10 are running into errors from this change.
|
Reverted in #144616 I'll take a look at fixing the python3.10 compatibility issue in a new patch. |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
In DebugCommunication, we currently are using 2 thread to drive lldb-dap. At the moment, they make an attempt at only synchronizing the
recv_packetsbetween the reader thread and the main test thread. Other stateful properties of the debug session are not guarded by a locks/mutex.To mitigate this, I am moving any state updates to the main thread inside the
_recv_packetmethod to ensure that between calls to_recv_packetthe state does not change out from under us in a test.This does mean the precise timing of events has changed slightly as a result and I've updated the existing tests that fail for me locally with this new behavior.
I think this should result in overall more predictable behavior, even if the test is slow due to the host workload or architecture differences.