================
@@ -0,0 +1,813 @@
+# FIXME: remove when LLDB_MINIMUM_PYTHON_VERSION > 3.8
+from __future__ import annotations
+
+import bisect
+import contextlib
+import itertools
+import json
+import os
+import socket
+import subprocess
+import threading
+import time
+from concurrent.futures import Future
+from dataclasses import asdict, dataclass, field, replace
+from pprint import pformat
+from typing import IO, Callable, Optional, Protocol, Tuple, Type,
runtime_checkable
+
+from .dap_types import (
+ AnyEvent,
+ DAPError,
+ Event,
+ MessageType,
+ RawMessage,
+ Request,
+ Response,
+)
+
+
+# See lldbtest.Base.spawnSubprocess, which should help ensure any processes
+# created by the DAP client are terminated correctly when the test ends.
+class SubProcessSpawner(Protocol):
+ def __call__(
+ self,
+ executable: str,
+ args: list[str] | None = None,
+ extra_env: list[str] | None = None,
+ install_remote: bool = True,
+ **kwargs,
+ ) -> subprocess.Popen[bytes]:
+ ...
+
+
+@dataclass(frozen=True)
+class DebugAdapterOptions:
+ """The options passed when spawning the debug adapter."""
+
+ args: list[str] = field(default_factory=list)
+ env: dict[str, str] = field(default_factory=dict)
+ cwd: Optional[str] = None
+ pre_init_commands: Optional[list[str]] = None
+ log_file: Optional[str] = None
+ # sever_mode related options.
+ connection: Optional[str] = None
+ connection_timeout: Optional[int] = None
+
+ @property
+ def run_as_server(self):
+ return self.connection is not None
+
+ def clone(self, **kwargs) -> DebugAdapterOptions:
+ """Returns a copy with the given fields overridden."""
+ return replace(self, **kwargs)
+
+ def __repr__(self):
+ return f"{type(self).__name__}: {pformat(asdict(self), indent=2,
compact=True)}"
+
+ def __post_init__(self):
+ # Check connection options is not in args.
+ if "--connection" in self.args or "--connection-timeout" in self.args:
+ raise DAPError(
+ f"--connection in adapter options, use the connection field
instead {self}"
+ )
+
+ if not self.run_as_server and self.connection_timeout is not None:
+ raise DAPError(
+ f"'--connection-timeout' option can only be used when a
connection is specified: {self}"
+ )
+
+
+class DebugAdapter:
+ """Spawns and owns the lifetime of lldb-dap binary"""
+
+ _listening_uri: Optional[str]
+
+ def __init__(self, executable: str, opts: DebugAdapterOptions):
+ self.executable = executable
+ self._connection_count = 0
+ self._is_server = opts.run_as_server
+
+ # Setup the process args.
+ process_args = [self.executable]
+ process_args.extend(opts.args)
+
+ if pre_init_commands := opts.pre_init_commands:
+ for command in pre_init_commands:
+ process_args.extend(["--pre-init-command", command])
+
+ # Verify we are using the correct args in stdio or server mode.
+ if opts.run_as_server:
+ process_args.extend(["--connection", opts.connection]) # type:
ignore
+ if opts.connection_timeout:
+ connection_timeout = str(opts.connection_timeout)
+ process_args.extend(["--connection-timeout",
connection_timeout])
+
+ # Setup process environment.
+ process_env = os.environ.copy()
+ process_env.update(opts.env)
+ if log_file := opts.log_file:
+ process_env["LLDBDAP_LOG"] = log_file
+
+ self._process = subprocess.Popen(
+ process_args,
+ stdin=subprocess.PIPE,
+ stdout=subprocess.PIPE,
+ stderr=subprocess.PIPE,
+ env=process_env,
+ cwd=opts.cwd,
+ )
+ assert self.is_alive, "expected running process"
+
+ if self.is_server:
+ self._listening_uri = self._read_listening_uri()
+ else:
+ self._listening_uri = None
+
+ def create_connection(self) -> DAPConnection:
+ if self.is_server:
+ assert self._listening_uri is not None
+ transport = _SocketTransport(uri=self._listening_uri)
+ else:
+ if self._connection_count > 0:
+ raise DAPError("Cannot create multiple connections in stdio
mode")
+ transport = _StdioTransport(self._process)
+
+ count = self._connection_count
+ connection_id = f"conn{count}" if self.is_server else "stdio"
+ self._connection_count += 1
+ return DAPConnection(connection_id, transport)
+
+ @property
+ def is_server(self):
+ return self._is_server
+
+ @property
+ def is_alive(self):
+ return self._process.poll() is None
+
+ @property
+ def process(self):
+ return self._process
+
+ def kill(self):
+ self._process.terminate()
+ try:
+ self._process.wait(timeout=2.0)
+ except subprocess.TimeoutExpired:
+ self._process.kill()
+
+ def _read_listening_uri(self) -> str:
+ # lldb-dap will print the listening address once the listener is
+ # made to stdout. The listener is formatted like
+ # `connection://host:port` or `unix-connection:///path`.
+ expected_prefix = "Listening for: "
+ process_stdout = self._process.stdout
+ if process_stdout is None:
+ raise AttributeError("expected the process stdout to be a PIPE")
+
+ out = process_stdout.readline().decode()
+ if not out:
+ # Check if there is a message in stderr.
+ err = ""
+ with contextlib.suppress(Exception):
+ if process_stderr := self.process.stderr:
+ err = process_stderr.read().decode()
+ raise EOFError(
+ f"Unexpected End of file for process {self.process.args},\n"
+ f"process stderr: {err}"
+ )
+
+ if not out.startswith(expected_prefix):
+ raise ValueError(
+ "lldb-dap failed to print listening address, "
+ f"expected '{expected_prefix}', got '{out}'"
+ )
+
+ # FIXME: use `str.removeprefix` when LLDB_MINIMUM_PYTHON_VERSION > 3.8
+ out = out[len(expected_prefix) :]
+
+ # If the listener expanded into multiple addresses, use the first.
+ uri = out.rstrip("\r\n").split(",", 1)[0]
+ return uri
+
+
+class EventHistory:
+ """Thread-safe event log that tests block against to observe the adapter.
+
+ Every event the debug adapter sends is recorded here by the read
+ thread, in the order it arrived. Tests don't read the log directly,
+ they call one of the `wait_for_*` methods, which block until a matching
+ event has been recorded.
+
+ Args:
+ timeout: Default timeout in seconds for `wait_for_*` functions.
+
+ Example:
+ Wait for a stop after stepping, without racing the adapter.
+
+ >>> step_resp = session.step_in(thread_id=1)
+ >>> # History will only check for events after the step_response sequence.
+ >>> stopped = history.wait_for_event(StoppedEvent, after=step_resp)
+
+ Wait for any of several events (either is an acceptable outcome).
+ >>> end = history.wait_for_any_event((StoppedEvent, TerminatedEvent),
after=continue_resp)
+
+ Narrow with a predicate.
+
+ >>> hit = history.wait_for_event(
+ ... StoppedEvent,
+ ... after=launch_resp,
+ ... until=lambda e: e.body.reason == "breakpoint",
+ ... )
+
+ Find the first Initialized event from the start of the history.
+ >>> init_event = history.wait_for_earliest_event(InitializedEvent)
+ """
+
+ def __init__(self, timeout: float):
+ self._sequences: list[int] = []
+ self._events: list[Event] = []
+ self._new_event_condition = threading.Condition()
+ self._timeout: float = timeout
+
+ self._is_closed: bool = False
+ self._closed_reason: Optional[Exception] = None
+
+ @property
+ def is_closed(self):
+ with self._new_event_condition:
+ return self._is_closed
+
+ def close(self, reason: Optional[Exception] = None):
+ """Close the history and wake all pending waiters.
+
+ After closing, `record` raises `DAPError` and any in-flight
+ `wait_for_*` call raises `DAPError` instead of timing out. This
+ is called when the adapter disconnects or the session ends so
+ tests do not block for the full default timeout.
+
+ Args:
+ reason: Optional exception describing why the history was
+ closed. When set, it is included in the error raised by
+ waiters so they can see the underlying cause.
+ """
+ with self._new_event_condition:
+ if self._is_closed:
+ raise DAPError(
+ f"history already closed with exception
{self._closed_reason}"
+ f"trying to close again with {reason}."
+ )
+ self._is_closed = True
+ self._closed_reason = reason
+ self._new_event_condition.notify_all()
+
+ def record(self, new_event: Event):
+ """Record an event in the history.
+
+ Enforces recording event in sequential order.
+ Raises:
+ DAPError: If the history has been closed or If the new event
+ seq is not greater than the last recorded event's `seq`.
+ """
+ new_seq: int = new_event.seq
+ with self._new_event_condition:
+ if self._is_closed:
+ raise DAPError(
+ "Cannot record in EventHistory: session is closed."
+ ) from self._closed_reason
+
+ if len(self._sequences) > 0:
+ # History must be sequential.
+ last_seen_seq = self._sequences[-1]
+ if new_seq <= last_seen_seq:
+ raise DAPError(
+ f"event: '{new_event.event}' seq '{new_seq}' is older
than last event: "
+ f"'{self._events[-1].event}' seq: '{last_seen_seq}'"
+ )
+
+ self._sequences.append(new_seq)
+ self._events.append(new_event)
+
+ # Sanity check.
+ assert len(self._sequences) == len(self._events)
+ self._new_event_condition.notify_all()
+
+ def wait_for_earliest_event(
+ self,
+ event_type: Type[AnyEvent],
+ *,
+ until: Optional[Callable[[AnyEvent], bool]] = None,
+ timeout: Optional[float] = None,
+ timeout_msg: Optional[str] = None,
+ ) -> AnyEvent:
+ """Wait for the earliest event of `event_type` in the history.
+
+ Searches from the beginning of the log (`seq` 0), so already-received
+ events count. Use this when a test wants the first event of a given
+ kind regardless of when it arrived.
+
+ Raises the same exceptions as `wait_for_event`.
+ """
+ assert issubclass(event_type, Event)
+
+ event_types = tuple((event_type,))
+ return self.__wait_for_any_event(
+ event_types,
+ after_seq=0,
+ until=until,
+ timeout=timeout,
+ timeout_msg=timeout_msg,
+ )
+
+ def wait_for_event(
+ self,
+ event_type: Type[AnyEvent],
+ *,
+ until: Optional[Callable[[AnyEvent], bool]] = None,
+ after: Event | Response,
+ timeout: Optional[float] = None,
+ timeout_msg: Optional[str] = None,
+ ) -> AnyEvent:
+ """Wait for the next event of `event_type` after a given message.
+
+ Search from "after some prior message" avoids races where
+ the event has already been observed: a test can capture a response
+ or event, run some action, and then wait for the *next* event of a
+ given kind without matching against anything already in the log.
+
+ Args:
+ event_type: Event subclass to match.
+ until: Optional predicate applied to each candidate. Only
+ events for which `until(event)` is true are accepted.
+ after: The prior event or response. Only events whose `seq`
+ is strictly greater are considered.
+ timeout: Override the history's default timeout, in seconds.
+ timeout_msg: Extra context appended to the `TimeoutError`
+ message if the wait times out.
+
+ Returns:
+ The first matching event after `after`.
+
+ Raises:
+ TimeoutError: If no matching event arrives within `timeout`.
+ DAPError: If the history is closed before a match is found.
+ """
+ assert issubclass(event_type, Event)
+
+ event_types = tuple((event_type,))
+ return self.wait_for_any_event(
+ event_types,
+ after=after,
+ until=until,
+ timeout=timeout,
+ timeout_msg=timeout_msg,
+ )
+
+ def wait_for_any_event(
+ self,
+ event_types: Tuple[Type[AnyEvent], ...],
+ *,
+ until: Optional[Callable[[AnyEvent], bool]] = None,
+ after: Event | Response,
+ timeout: Optional[float] = None,
+ timeout_msg: Optional[str] = None,
+ ):
+ """Wait for the next event matching any of several types.
+
+ Same semantics as `wait_for_event`, but the returned event may be
+ an instance of any of the given `event_types`. Useful when a test
+ is expecting the first of two different events.
+ """
+ assert after.type in (
+ MessageType.EVENT,
+ MessageType.RESPONSE,
+ ), f"expects instance of 'Event' or 'Response' got {after}."
+ return self.__wait_for_any_event(
+ event_types,
+ after_seq=after.seq,
+ until=until,
+ timeout=timeout,
+ timeout_msg=timeout_msg,
+ )
+
+ def __wait_for_any_event(
+ self,
+ event_types: Tuple[Type[AnyEvent], ...],
+ *,
+ after_seq: int,
+ until: Optional[Callable[[AnyEvent], bool]] = None,
+ timeout: Optional[float] = None,
+ timeout_msg: Optional[str] = None,
+ ):
+ assert after_seq >= 0, "response or event sequence must be greater
than 0."
+ assert isinstance(event_types, tuple), "expected a tuple of events."
+ assert len(event_types) > 0, "expected at least one event to wait for."
+
+ def make_error_msg(is_timeout: bool = True):
+ event_names = [x.__name__ for x in event_types]
+ prefix = f"Timed out after {timeout}s" if is_timeout else "Error
while"
+ err_msg = f"{prefix} waiting for any event that matches:
{event_names}"
+ err_msg += f" after sequence: {after_seq}."
+
+ if timeout_msg:
+ err_msg += f"\n\t{timeout_msg}."
+
+ with self._new_event_condition:
+ last_event = self._events[-1] if self._events else None
+ err_msg += f"\n\tlast seen event: {last_event}."
+ return err_msg
+
+ def is_event_and_matches_condition(evt: Event):
+ if not isinstance(evt, event_types):
+ return False
+
+ if until is None:
+ return True
+
+ matches = until(evt)
+ return matches
+
+ timeout = timeout or self._timeout
+ try:
+ event = self.__wait_until(
+ is_event_and_matches_condition, after_seq=after_seq,
timeout=timeout
+ )
+ except DAPError as err:
+ # Add extra context to the error.
+ err.args = (f"{err.args[0]}\n\t{make_error_msg(False)}",
*err.args[1:])
+ raise
+
+ if event is None:
+ raise TimeoutError(make_error_msg())
+
+ # Sanity check.
+ assert isinstance(event, event_types)
+ return event
+
+ def __wait_until(
+ self,
+ matches_condition: Callable[[Event], bool],
+ *,
+ after_seq: int,
+ timeout: float,
+ ):
+ """Waits until the `matches_condition` returns true for an exiting
+ event or an incoming event. If the history is closed during the wait,
+ raise a DAPError."""
+
+ end_time = time.monotonic() + timeout
+ start_idx = 0
+
+ with self._new_event_condition:
+ while True:
+ seq_len = len(self._sequences)
+ idx = bisect.bisect_right(self._sequences, after_seq,
lo=start_idx)
+
+ # Scan forward until we find a matching type.
+ for event in itertools.islice(self._events, idx, seq_len):
+ if matches_condition(event):
+ return event
+ start_idx = seq_len
+
+ if self._is_closed: # Can no longer receive new messages.
+ reason = self._closed_reason
+ last_evt = self._events[-1] if self._events else None
+ raise DAPError.history_closed(reason, last_evt) from reason
+
+ remaining_time = end_time - time.monotonic()
+ if remaining_time <= 0:
+ return None
+ self._new_event_condition.wait(remaining_time)
+
+
+def redirect_stream(
+ in_stream: IO[bytes], out_stream: IO[str], thread_name: str
+) -> threading.Thread:
+ """
+ Creates a new thread that redirects stream from `in_stream` to
+ `out_stream`. We use this for the 'runInTerminal' process to send stdio
+ to the session's output.
+
+ Returns a thread that redirects the stream.
+ """
+
+ def read_loop(in_stream: IO[bytes], out_stream: IO[str]):
+ with contextlib.suppress(OSError, ValueError): # Nothing to report.
+ while True:
+ chunk = in_stream.read(4096)
+ if not chunk:
+ break
+
+ out_stream.write(chunk.decode(errors="replace"))
+ out_stream.flush()
+
+ thread_name = f"redirect_{thread_name}"
+ redirect_thread = threading.Thread(
+ target=read_loop,
+ name=thread_name,
+ args=[in_stream, out_stream],
+ daemon=True,
+ )
+ redirect_thread.start()
+
+ return redirect_thread
+
+
+@runtime_checkable
+class Transport(Protocol):
+ """Interface representing a bidirectional transport.
+
+ Implementations:
+ `_StdioTransport`: speaks to the adapter using a subprocess's
stdin/stdout.
+ Used when the adapter is spawned as a child process.
+ `_SocketTransport`: speaks to the adapter using socket. Used when the
+ adapter is already running and exposes connection URI.
+ """
+
+ def write(self, data: bytes):
+ ...
+
+ def read(self, n: int) -> bytes:
+ ...
+
+ def readline(self) -> bytes:
+ ...
+
+ def close(self):
+ """Close the transport.
+
+ Buffered data will be flushed and transport closed.
+ """
+ ...
+
+ @property
+ def is_alive(self) -> bool:
+ """Whether send or receive bytes through the transport."""
+ ...
+
+
+@dataclass(frozen=True)
+class MessageHandler:
+ on_response: Callable[[RawMessage], None]
+ on_event: Callable[[RawMessage], None]
+ on_reverse_request: Callable[[RawMessage], None]
+ on_close: Optional[Callable[[Optional[Exception]], None]] = lambda _: None
+
+
+class DAPConnection:
----------------
ashgti wrote:
Right, but positive assertions work better for ensuring something has happened
instead of a negative assertion that something doesn't happen. Positive
assertions allow the test to continue once the precondition is met.
For example, calling 'continue', which should have the precondition that the
process is in the stopped state, send the 'continue' request, then wait until
we receive the 'continued' event. The 'continued' event could come before or
after the 'continue' response is received.
Framing things in terms of negative assertions like "we don't receive any
'module' events for the next 5s" requires us to wait 5s for that to be met vs
if we had something like "we receive the 'module' removed event for
libfoo.dylib".
The big sources of nondeterminism in the tests are around event timing, some
events come before or after a response to a specific request, but it can vary
depending on the platform and even things like system load. Our existing tests
will check backwards for events that haven't been popped from the packet queue.
https://github.com/llvm/llvm-project/blob/main/lldb/packages/Python/lldbsuite/test/tools/lldb-dap/dap_server.py#L413-L414
This means if there are 2 stopped events in the packet queue we'll actually
just grab the first or we could have an out of sync view of the process because
the we waited on a stopped event that was sitting in the buffer but the process
has already continued.
Removing the packet queue by reframing things in terms of state and removing
the reader thread and only doing a read when we need data would go a long way
to improving determinism and stability, I think. The main challenges with that
is having a read with a timeout, I think Windows doesn't have an easy way to do
this in python. I think in python 3.12 they added support for non-blocking
reads on pipes in Windows (see
https://docs.python.org/3/library/os.html#os.set_blocking), which should help a
lot since the subprocess is using a pipe for reading. Or using asyncio to read
could work as well, since that allows us to read with a timeout as well.
https://github.com/llvm/llvm-project/pull/203978
_______________________________________________
lldb-commits mailing list
[email protected]
https://lists.llvm.org/cgi-bin/mailman/listinfo/lldb-commits
