from .server import request, capability, send_event
from .sources import make_source
-from .startup import send_gdb_with_response, in_gdb_thread, log_stack
+from .startup import in_gdb_thread, log_stack
from .typecheck import type_check
# Be sure to include the path in the key, so that we only
# clear out breakpoints coming from this same source.
key = "source:" + source["path"]
- result = send_gdb_with_response(lambda: _set_breakpoints(key, specs))
+ result = _set_breakpoints(key, specs)
return {
"breakpoints": result,
}
@capability("supportsFunctionBreakpoints")
def set_fn_breakpoint(*, breakpoints: Sequence, **args):
specs = [_rewrite_fn_breakpoint(**bp) for bp in breakpoints]
- result = send_gdb_with_response(lambda: _set_breakpoints("function", specs))
return {
- "breakpoints": result,
+ "breakpoints": _set_breakpoints("function", specs),
}
*, breakpoints: Sequence, offset: Optional[int] = None, **args
):
specs = [_rewrite_insn_breakpoint(**bp) for bp in breakpoints]
- result = send_gdb_with_response(lambda: _set_breakpoints("instruction", specs))
return {
- "breakpoints": result,
+ "breakpoints": _set_breakpoints("instruction", specs),
}
options = [{"filterId": filter} for filter in filters]
options.extend(filterOptions)
options = [_rewrite_exception_breakpoint(**bp) for bp in options]
- result = send_gdb_with_response(lambda: _set_exception_catchpoints(options))
return {
- "breakpoints": result,
+ "breakpoints": _set_exception_catchpoints(options),
}
from .scopes import symbol_value
from .server import request, capability
from .sources import make_source
-from .startup import send_gdb_with_response, in_gdb_thread
+from .startup import in_gdb_thread
from .state import set_thread
from .typecheck import type_check
from .varref import apply_format
if format is None:
format = {}
format = check_stack_frame(**format)
- return send_gdb_with_response(
- lambda: _backtrace(threadId, levels, startFrame, format)
- )
+ return _backtrace(threadId, levels, startFrame, format)
import gdb
from .server import request, capability
-from .startup import send_gdb_with_response, in_gdb_thread
+from .startup import in_gdb_thread
@in_gdb_thread
**extra
):
pc = int(memoryReference, 0) + offset
- return send_gdb_with_response(
- lambda: _disassemble(pc, instructionOffset, instructionCount)
- )
+ return _disassemble(pc, instructionOffset, instructionCount)
from .frames import select_frame
from .server import capability, request, client_bool_capability
-from .startup import send_gdb_with_response, in_gdb_thread
+from .startup import in_gdb_thread
from .varref import find_variable, VariableReference, apply_format
):
if context in ("watch", "variables"):
# These seem to be expression-like.
- return send_gdb_with_response(lambda: _evaluate(expression, frameId, format))
+ return _evaluate(expression, frameId, format)
elif context == "hover":
- return send_gdb_with_response(
- lambda: _eval_for_hover(expression, frameId, format)
- )
+ return _eval_for_hover(expression, frameId, format)
elif context == "repl":
# Ignore the format for repl evaluation.
- return send_gdb_with_response(lambda: _repl(expression, frameId))
+ return _repl(expression, frameId)
else:
raise Exception('unknown evaluate context "' + context + '"')
if not client_bool_capability("supportsVariablePaging"):
start = 0
count = 0
- result = send_gdb_with_response(
- lambda: _variables(variablesReference, start, count, format)
- )
- return {"variables": result}
+ return {"variables": _variables(variablesReference, start, count, format)}
@capability("supportsSetExpression")
def set_expression(
*, expression: str, value: str, frameId: Optional[int] = None, format=None, **args
):
- return send_gdb_with_response(
- lambda: _set_expression(expression, value, frameId, format)
- )
+ return _set_expression(expression, value, frameId, format)
# Helper function to perform an assignment.
def set_variable(
*, variablesReference: int, name: str, value: str, format=None, **args
):
- return send_gdb_with_response(
- lambda: _set_variable(variablesReference, name, value, format)
- )
+ return _set_variable(variablesReference, name, value, format)
from .events import ExecutionInvoker
from .server import request, capability
-from .startup import send_gdb, send_gdb_with_response, in_gdb_thread, exec_and_log
+from .startup import in_gdb_thread, exec_and_log
-# The program being launched, or None. This should only be access
-# from the DAP thread.
+# The program being launched, or None. This should only be accessed
+# from the gdb thread.
_program = None
# Any parameters here are necessarily extensions -- DAP requires this
# from implementations. Any additions or changes here should be
# documented in the gdb manual.
-@request("launch")
+@request("launch", response=False)
def launch(
*,
program: Optional[str] = None,
):
global _program
_program = program
- send_gdb(
- lambda: _launch_setup(program, cwd, args, env, stopAtBeginningOfMainSubprogram)
- )
+ _launch_setup(program, cwd, args, env, stopAtBeginningOfMainSubprogram)
@request("attach")
cmd = "target remote " + target
else:
raise Exception("attach requires either 'pid' or 'target'")
- # Use send_gdb_with_response to ensure we get an error if the
- # attach fails.
- send_gdb_with_response(cmd)
- return None
+ exec_and_log(cmd)
@capability("supportsConfigurationDoneRequest")
-@request("configurationDone")
+@request("configurationDone", response=False)
def config_done(**args):
global _program
if _program is not None:
# Suppress the continue event, but don't set any particular
# expected stop.
- send_gdb(ExecutionInvoker("run", None))
+ ExecutionInvoker("run", None)()
from .server import capability, request
from .sources import decode_source
-from .startup import in_gdb_thread, send_gdb_with_response
+from .startup import in_gdb_thread
@in_gdb_thread
def breakpoint_locations(*, source, line: int, endLine: Optional[int] = None, **extra):
if endLine is None:
endLine = line
- return send_gdb_with_response(lambda: _find_lines(source, line, endLine))
+ return _find_lines(source, line, endLine)
import gdb
from .server import request, capability
-from .startup import send_gdb_with_response, in_gdb_thread
-
-
-@in_gdb_thread
-def _read_memory(addr, count):
- buf = gdb.selected_inferior().read_memory(addr, count)
- return base64.b64encode(buf).decode("ASCII")
@request("readMemory")
@capability("supportsReadMemoryRequest")
def read_memory(*, memoryReference: str, offset: int = 0, count: int, **extra):
addr = int(memoryReference, 0) + offset
- buf = send_gdb_with_response(lambda: _read_memory(addr, count))
+ buf = gdb.selected_inferior().read_memory(addr, count)
return {
"address": hex(addr),
- "data": buf,
+ "data": base64.b64encode(buf).decode("ASCII"),
}
-@in_gdb_thread
-def _write_memory(addr, contents):
- buf = base64.b64decode(contents)
- gdb.selected_inferior().write_memory(addr, buf)
-
-
@request("writeMemory")
@capability("supportsWriteMemoryRequest")
def write_memory(*, memoryReference: str, offset: int = 0, data: str, **extra):
addr = int(memoryReference, 0) + offset
- send_gdb_with_response(lambda: _write_memory(addr, data))
- return {}
+ buf = base64.b64decode(data)
+ gdb.selected_inferior().write_memory(addr, buf)
import gdb
from .server import capability, request
-from .startup import in_gdb_thread, send_gdb_with_response
+from .startup import in_gdb_thread
@in_gdb_thread
@capability("supportsModulesRequest")
@request("modules")
def modules(*, startModule: int = 0, moduleCount: int = 0, **args):
- return send_gdb_with_response(lambda: _modules(startModule, moduleCount))
+ return _modules(startModule, moduleCount)
return result
-@request("next")
+@request("next", response=False)
def next(
*, threadId: int, singleThread: bool = False, granularity: str = "statement", **args
):
- send_gdb(lambda: _handle_thread_step(threadId, singleThread))
+ _handle_thread_step(threadId, singleThread)
cmd = "next"
if granularity == "instruction":
cmd += "i"
- send_gdb(ExecutionInvoker(cmd, StopKinds.STEP))
+ ExecutionInvoker(cmd, StopKinds.STEP)()
@capability("supportsSteppingGranularity")
@capability("supportsSingleThreadExecutionRequests")
-@request("stepIn")
+@request("stepIn", response=False)
def step_in(
*, threadId: int, singleThread: bool = False, granularity: str = "statement", **args
):
- send_gdb(lambda: _handle_thread_step(threadId, singleThread))
+ _handle_thread_step(threadId, singleThread)
cmd = "step"
if granularity == "instruction":
cmd += "i"
- send_gdb(ExecutionInvoker(cmd, StopKinds.STEP))
+ ExecutionInvoker(cmd, StopKinds.STEP)()
-@request("stepOut")
+@request("stepOut", response=False)
def step_out(*, threadId: int, singleThread: bool = False, **args):
- send_gdb(lambda: _handle_thread_step(threadId, singleThread, True))
- send_gdb(ExecutionInvoker("finish", StopKinds.STEP))
+ _handle_thread_step(threadId, singleThread, True)
+ ExecutionInvoker("finish", StopKinds.STEP)()
-@request("continue")
+# This is a server-side request because it is funny: it wants to
+# 'continue' but also return a result, which precludes using
+# response=False. Using 'continue &' would mostly work ok, but this
+# yields races when a stop occurs before the response is sent back to
+# the client.
+@request("continue", on_dap_thread=True)
def continue_request(*, threadId: int, singleThread: bool = False, **args):
locked = send_gdb_with_response(lambda: _handle_thread_step(threadId, singleThread))
send_gdb(ExecutionInvoker("continue", None))
from .events import StopKinds, ExecutionInvoker
from .server import request
-from .startup import send_gdb
-@request("pause")
+@request("pause", response=False)
def pause(**args):
- send_gdb(ExecutionInvoker("interrupt -a", StopKinds.PAUSE))
+ ExecutionInvoker("interrupt -a", StopKinds.PAUSE)()
import gdb
from .frames import frame_for_id
-from .startup import send_gdb_with_response, in_gdb_thread
+from .startup import in_gdb_thread
from .server import request
from .varref import BaseReference
@request("scopes")
def scopes(*, frameId: int, **extra):
- scopes = send_gdb_with_response(lambda: _get_scope(frameId))
- return {"scopes": scopes}
+ return {"scopes": _get_scope(frameId)}
from .io import start_json_writer, read_json
from .startup import (
+ exec_and_log,
in_dap_thread,
+ in_gdb_thread,
+ send_gdb,
+ send_gdb_with_response,
start_thread,
log,
log_stack,
- send_gdb_with_response,
)
from .typecheck import type_check
_server.send_event(event, body)
-def request(name):
- """A decorator that indicates that the wrapper function implements
- the DAP request NAME."""
+def request(name: str, *, response: bool = True, on_dap_thread: bool = False):
+ """A decorator for DAP requests.
+
+ This registers the function as the implementation of the DAP
+ request NAME. By default, the function is invoked in the gdb
+ thread, and its result is returned as the 'body' of the DAP
+ response.
+
+ Some keyword arguments are provided as well:
+
+ If RESPONSE is False, the result of the function will not be
+ waited for and no 'body' will be in the response.
+
+ If ON_DAP_THREAD is True, the function will be invoked in the DAP
+ thread. When ON_DAP_THREAD is True, RESPONSE may not be False.
+ """
+
+ # Validate the parameters.
+ assert not on_dap_thread or response
def wrap(func):
- global _commands
code = func.__code__
# We don't permit requests to have positional arguments.
try:
assert code.co_argcount == 0
# A request must have a **args parameter.
assert code.co_flags & inspect.CO_VARKEYWORDS
- # All requests must run in the DAP thread.
- # Also type-check the calls.
- func = in_dap_thread(type_check(func))
- _commands[name] = func
- return func
+
+ # Type-check the calls.
+ func = type_check(func)
+
+ # Verify that the function is run on the correct thread.
+ if on_dap_thread:
+ cmd = in_dap_thread(func)
+ else:
+ func = in_gdb_thread(func)
+
+ if response:
+
+ def sync_call(**args):
+ return send_gdb_with_response(lambda: func(**args))
+
+ cmd = sync_call
+ else:
+
+ def non_sync_call(**args):
+ return send_gdb(lambda: func(**args))
+
+ cmd = non_sync_call
+
+ global _commands
+ _commands[name] = cmd
+ return cmd
return wrap
return False
-@request("initialize")
+@request("initialize", on_dap_thread=True)
def initialize(**args):
global _server, _capabilities
_server.config = args
@request("terminate")
@capability("supportsTerminateRequest")
def terminate(**args):
- # We can ignore the result here, because we only really need to
- # synchronize.
- send_gdb_with_response("kill")
+ exec_and_log("kill")
-@request("disconnect")
+@request("disconnect", on_dap_thread=True)
@capability("supportTerminateDebuggee")
def disconnect(*, terminateDebuggee: bool = False, **args):
if terminateDebuggee:
- terminate()
+ send_gdb_with_response("kill")
_server.shutdown()
import gdb
from .server import request, capability
-from .startup import send_gdb_with_response, in_gdb_thread
+from .startup import in_gdb_thread
# The next available source reference ID. Must be greater than 0.
return _id_map[ref]["path"]
-@in_gdb_thread
-def _sources():
+@request("loadedSources")
+@capability("supportsLoadedSourcesRequest")
+def loaded_sources(**extra):
result = []
for elt in gdb.execute_mi("-file-list-exec-source-files")["files"]:
result.append(make_source(elt["fullname"], elt["file"]))
}
-@request("loadedSources")
-@capability("supportsLoadedSourcesRequest")
-def loaded_sources(**extra):
- return send_gdb_with_response(_sources)
-
-
-# This helper is needed because we must only access the globals here
-# from the gdb thread.
-@in_gdb_thread
-def _get_source(source):
- filename = decode_source(source)
- with open(filename) as f:
- content = f.read()
- return {
- "content": content,
- }
-
-
@request("source")
def source(*, source=None, sourceReference: int, **extra):
# The 'sourceReference' parameter is required by the spec, but is
# 'source' is preferred.
if source is None:
source = {"sourceReference": sourceReference}
- return send_gdb_with_response(lambda: _get_source(source))
+ filename = decode_source(source)
+ with open(filename) as f:
+ content = f.read()
+ return {
+ "content": content,
+ }
import gdb
from .server import request
-from .startup import send_gdb_with_response, in_gdb_thread
def _thread_name(thr):
return None
-# A helper function to construct the list of threads.
-@in_gdb_thread
-def _get_threads():
+@request("threads")
+def threads(**args):
result = []
for thr in gdb.selected_inferior().threads():
one_result = {
if name is not None:
one_result["name"] = name
result.append(one_result)
- return result
-
-
-@request("threads")
-def threads(**args):
- result = send_gdb_with_response(_get_threads)
return {
"threads": result,
}
projects / thirdparty / binutils-gdb.git / commitdiff
