================
@@ -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:
I wonder if we need to have 2 threads anymore.
We could only write to the connection when we need to send a request and only
read from the connect when we're waiting on a reply.
That does change how we handle requests or events, but with an updated
request+response or event abstraction representing futures, we could remove the
thread and simplify a lot of code.
Additionally, instead of tracking specific events, we could track specific
states we're interested.
For example, if we want to wait for the process to be stopped, we don't need to
wait for a stopped event, which could come before or after our latest request
or we could already be stopped. Instead we could see what state we're in and
handle processing events until the desired state is reached. The DAP protocol
is inherently stateful and the problem with trying watch for specific events is
timing changes things.
https://github.com/llvm/llvm-project/pull/203978
_______________________________________________
lldb-commits mailing list
[email protected]
https://lists.llvm.org/cgi-bin/mailman/listinfo/lldb-commits
