Fix microsoft#1648: Messaging does not allow reverse requests

Separate message parsing and message handling into separate threads.

Remove nested message handling in request handlers via `yield`, since it is incompatible with the new split model, and replace it with NO_RESPONSE and Request.respond() to defer responses until later.

Change Message.cant_handle() and Message.isnt_valid() to respond to the request and return the exception, instead of raising it, to accommodate NO_RESPONSE scenarios where a failure needs to be reported later.

Fix microsoft#1678: Do not rely on "processId" being returned by "runInTerminal" request

Extract debuggee PID from the "process" event sent by the debug server.

Fix microsoft#1679: "exited" event sometimes reports "exitCode": null

Report it as -1 if it cannot be retrieved from the debuggee process.

Fix microsoft#1680: Fatal errors in message loop do not fail fast

os._exit() immediately if a fatal error occurs in message parsing or message handling background threads.

.vscode/launch.json

@@ -21,8 +21,8 @@
             "type": "python",
             "request": "launch",
             "debugServer": 8765,
-            "console": "internalConsole",
-            //"console": "integratedTerminal",
+            //"console": "internalConsole",
+            "console": "integratedTerminal",
             //"console": "externalTerminal",
             "consoleTitle": "ptvsd.server",
             //"program": "${file}",

src/ptvsd/adapter/contract.py

@@ -41,7 +41,7 @@ def __init__(self, message):
             try:
                 value = validate(value)
             except Exception as exc:
-                message.isnt_valid("{0!j} {1}", name, exc)
+                raise message.isnt_valid("{0!j} {1}", name, exc)
 
             assert value != (), fmt(
                 "{0!j} must provide a default value for missing properties.", validate

src/ptvsd/adapter/debuggee.py

@@ -34,6 +34,10 @@
 pid = None
 """Debuggee process ID."""
 
+_got_pid = threading.Event()
+"""A threading.Event that is set when pid is set.
+"""
+
 _exited = None
 """A threading.Event that is set when the debuggee process exits.
 
@@ -110,7 +114,7 @@ def property_or_debug_option(prop_name, flag_name):
         assert prop_name[0].islower() and flag_name[0].isupper()
         value = request(prop_name, json.default(flag_name in debug_options))
         if value is False and flag_name in debug_options:
-            request.isnt_valid(
+            raise request.isnt_valid(
                 '{0!r}:false and "debugOptions":[{1!r}] are mutually exclusive',
                 prop_name,
                 flag_name,
@@ -125,7 +129,7 @@ def property_or_debug_option(prop_name, flag_name):
     )
     if console != "internalConsole":
         if not contract.ide.capabilities["supportsRunInTerminalRequest"]:
-            request.cant_handle(
+            raise request.cant_handle(
                 'Unable to launch via "console":{0!j}, because the IDE is does not '
                 'have the "supportsRunInTerminalRequest" capability',
                 console,
@@ -136,7 +140,7 @@ def property_or_debug_option(prop_name, flag_name):
     cmdline = []
     if property_or_debug_option("sudo", "Sudo"):
         if platform.system() == "Windows":
-            request.cant_handle('"sudo":true is not supported on Windows.')
+            raise request.cant_handle('"sudo":true is not supported on Windows.')
         else:
             cmdline += ["sudo"]
 
@@ -145,7 +149,9 @@ def property_or_debug_option(prop_name, flag_name):
     python_key = "python"
     if python_key in request:
         if "pythonPath" in request:
-            request.isnt_valid('"pythonPath" is not valid if "python" is specified')
+            raise request.isnt_valid(
+                '"pythonPath" is not valid if "python" is specified'
+            )
     elif "pythonPath" in request:
         python_key = "pythonPath"
     python = request(python_key, json.array(unicode, vectorize=True, size=(1,)))
@@ -185,9 +191,13 @@ def property_or_debug_option(prop_name, flag_name):
 
     num_targets = len([x for x in (program, module, code) if x != ()])
     if num_targets == 0:
-        request.isnt_valid('either "program", "module", or "code" must be specified')
+        raise request.isnt_valid(
+            'either "program", "module", or "code" must be specified'
+        )
     elif num_targets != 1:
-        request.isnt_valid('"program", "module", and "code" are mutually exclusive')
+        raise request.isnt_valid(
+            '"program", "module", and "code" are mutually exclusive'
+        )
 
     cmdline += request("args", json.array(unicode))
 
@@ -209,7 +219,7 @@ def _spawn_popen(request, spawn_info):
     try:
         proc = subprocess.Popen(spawn_info.cmdline, cwd=spawn_info.cwd, env=env)
     except Exception as exc:
-        request.cant_handle(
+        raise request.cant_handle(
             "Error launching process: {0}\n\nCommand line:{1!r}",
             exc,
             spawn_info.cmdline,
@@ -220,6 +230,7 @@ def _spawn_popen(request, spawn_info):
     global pid
     try:
         pid = proc.pid
+        _got_pid.set()
         ProcessTracker().track(pid)
     except Exception:
         # If we can't track it, we won't be able to terminate it if asked; but aside
@@ -258,19 +269,60 @@ def _spawn_terminal(request, spawn_info):
     }
 
     try:
-        result = channels.Channels().ide().request("runInTerminal", body)
+        channels.Channels().ide().request("runInTerminal", body)
     except messaging.MessageHandlingError as exc:
         exc.propagate(request)
 
+    # Although "runInTerminal" response has "processId", it's optional, and in practice
+    # it is not used by VSCode: https://github.com/microsoft/vscode/issues/61640.
+    # Thus, we can only retrieve the PID via the "process" event, and only then we can
+    # start tracking it. Until then, nothing else to do.
+    pass
+
+
+def parse_pid(process_event):
+    assert process_event.is_event("process")
+
     global pid
-    pid = result("processId", int)
+    sys_pid = process_event("systemProcessId", int)
 
-    try:
-        ProcessTracker().track(pid, after_exit=lambda: _exited.set())
-    except Exception as exc:
-        # If we can't track it, we won't be able to detect if it exited or retrieve
-        # the exit code, so fail immediately.
-        request.cant_handle("Couldn't get debuggee process handle: {0}", str(exc))
+    if not _got_pid.is_set():
+        # Launched with "runInTerminal", so we had no way to get the PID before.
+        # Now that we do, start tracking it as usual.
+
+        def after_exit(code):
+            global exit_code
+            exit_code = code
+            _exited.set()
+
+        try:
+            pid = sys_pid
+            _got_pid.set()
+            ProcessTracker().track(pid, after_exit=after_exit)
+        except Exception as exc:
+            # If we can't track it, we won't be able to detect if it exited or retrieve
+            # the exit code, so fail immediately.
+            raise process_event.cant_handle(
+                "Couldn't get debuggee process handle: {0}", str(exc)
+            )
+
+    elif pid != sys_pid:
+        # Launched directly, so we already have the PID, but it doesn't match what
+        # the debug server says it is - something is very wrong.
+        raise process_event.isnt_valid(
+            '"systemProcessId":{0!j} in "process" event does not match actual PID={1}',
+            sys_pid,
+            pid,
+        )
+
+
+def wait_for_pid(timeout=None):
+    """Waits for debuggee PID to be determined.
+
+    Returns True if PID was determined, False if the wait timed out. If it returned
+    True, then pid is guaranteed to be set.
+    """
+    return _got_pid.wait(timeout)
 
 
 def wait_for_exit(timeout=None):
@@ -280,6 +332,13 @@ def wait_for_exit(timeout=None):
     True, then exit_code is guaranteed to be set.
     """
 
+    if pid is None:
+        # Debugee was launched with "runInTerminal", but the debug session fell apart
+        # before we got a "process" event and found out what its PID is. It's not a
+        # fatal error, but there's nothing to wait on. Debuggee process should have
+        # exited (or crashed) by now in any case.
+        return
+
     assert _exited is not None
     timed_out = not _exited.wait(timeout)
     if not timed_out:

src/ptvsd/adapter/messages.py

@@ -46,7 +46,7 @@ def _server(self):
         """
         server = _channels.server()
         if server is None:
-            messaging.Message.isnt_valid(
+            raise messaging.Message.isnt_valid(
                 "Connection to debug server is not established yet"
             )
         return server
@@ -62,12 +62,11 @@ def handle_if_allowed(self, message):
                 current_state = state.current()
                 if current_state in states:
                     return handler(self, message)
-                if isinstance(message, messaging.Request):
-                    message.isnt_valid(
-                        "Request {0!r} is not allowed in adapter state {1!r}.",
-                        message.command,
-                        current_state,
-                    )
+                raise message.isnt_valid(
+                    "{0} is not allowed in adapter state {1!r}.",
+                    message.describe(),
+                    current_state,
+                )
 
             return handle_if_allowed
 
@@ -116,6 +115,9 @@ class IDEMessages(Messages):
     # so store them here until then. After all messages are replayed, it is set to None.
     _initial_messages = []
 
+    # "launch" or "attach" request that started debugging.
+    _start_request = None
+
     # A decorator to add the message to initial_messages if needed before handling it.
     # Must be applied to the handler for every message that can be received before
     # connection to the debug server can be established while handling attach/launch,
@@ -151,7 +153,7 @@ def initialize_request(self, request):
 
     # Handles various attributes common to both "launch" and "attach".
     def _debug_config(self, request):
-        assert request.command in ("launch", "attach")
+        assert request.is_request("launch", "attach")
         self._shared.start_method = request.command
         _Shared.readonly_attrs.add("start_method")
 
@@ -170,7 +172,7 @@ def launch_request(self, request):
         # TODO: nodebug
         debuggee.spawn_and_connect(request)
 
-        return self._configure()
+        return self._configure(request)
 
     @_replay_to_server
     @_only_allowed_while("initializing")
@@ -179,16 +181,16 @@ def attach_request(self, request):
         _Shared.readonly_attrs.add("terminate_on_disconnect")
         self._debug_config(request)
 
-        options.host = request.arguments.get("host", options.host)
-        options.port = int(request.arguments.get("port", options.port))
+        options.host = request("host", options.host)
+        options.port = request("port", options.port)
         _channels.connect_to_server(address=(options.host, options.port))
 
-        return self._configure()
+        return self._configure(request)
 
     # Handles the configuration request sequence for "launch" or "attach", from when
     # the "initialized" event is sent, to when "configurationDone" is received; see
     # https://github.com/microsoft/vscode/issues/4902#issuecomment-368583522
-    def _configure(self):
+    def _configure(self, request):
         log.debug("Replaying previously received messages to server.")
 
         assert len(self._initial_messages)
@@ -209,28 +211,42 @@ def _configure(self):
             self._server.propagate(msg)
 
         log.debug("Finished replaying messages to server.")
-        self.initial_messages = None
+        self._initial_messages = None
+        self._start_request = request
+
+        # Wait until we have the debuggee PID - we either know it already because we
+        # have launched it directly, or we'll find out eventually from the "process"
+        # server event. Either way, we need to know the PID before we can tell the
+        # server to start debugging, because we need to be able to kill the debuggee
+        # process if anything goes wrong.
+        #
+        # However, we can't block forever, because the debug server can also crash
+        # before it had a chance to send the event - so wake up periodically, and
+        # check whether server channel is still alive.
+        while not debuggee.wait_for_pid(1):
+            if _channels.server() is None:
+                raise request.cant_handle("Debug server disconnected unexpectedly.")
 
         # Let the IDE know that it can begin configuring the adapter.
         state.change("configuring")
         self._ide.send_event("initialized")
-
-        # Process further incoming messages, until we get "configurationDone".
-        while state.current() == "configuring":
-            yield
+        return messaging.NO_RESPONSE  # will respond on "configurationDone"
 
     @_only_allowed_while("configuring")
     def configurationDone_request(self, request):
-        ret = self._server.delegate(request)
+        assert self._start_request is not None
+
+        result = self._server.delegate(request)
         state.change("running")
         ServerMessages().release_events()
-        return ret
+        request.respond(result)
+        self._start_request.respond({})
 
     def _disconnect_or_terminate_request(self, request):
         assert request.is_request("disconnect") or request.is_request("terminate")
 
         if request("restart", json.default(False)):
-            request.isnt_valid("Restart is not supported")
+            raise request.isnt_valid("Restart is not supported")
 
         terminate = (request.command == "terminate") or request(
             "terminateDebuggee", json.default(self._shared.terminate_on_disconnect)
@@ -364,7 +380,9 @@ def request(self, request):
         # requests sent over that boundary, since they may contain arbitrary code
         # that the IDE will execute - e.g. "runInTerminal". The adapter must only
         # propagate requests that it knows are safe.
-        request.isnt_valid("Requests from the debug server to the IDE are not allowed.")
+        raise request.isnt_valid(
+            "Requests from the debug server to the IDE are not allowed."
+        )
 
     # Generic event handler, used if there's no specific handler below.
     def event(self, event):
@@ -383,13 +401,24 @@ def initialized_event(self, event):
         # also remove the 'initialized' event sent from IDE messages.
         pass
 
+    @_only_allowed_while("initializing")
+    def process_event(self, event):
+        try:
+            debuggee.parse_pid(event)
+        except Exception:
+            # If we couldn't retrieve or validate PID, we can't safely continue
+            # debugging, so shut everything down.
+            self.disconnect()
+        else:
+            self._ide.propagate(event)
+
     @_only_allowed_while("running")
     def ptvsd_subprocess_event(self, event):
         sub_pid = event("processId", int)
         try:
             debuggee.register_subprocess(sub_pid)
         except Exception as exc:
-            event.cant_handle("{0}", exc)
+            raise event.cant_handle("{0}", exc)
         self._ide.propagate(event)
 
     def terminated_event(self, event):
@@ -426,7 +455,10 @@ def disconnect(self):
             # The debuggee process should exit shortly after it has disconnected, but just
             # in case it gets stuck, don't wait forever, and force-kill it if needed.
             debuggee.terminate(after=5)
-            self._ide.send_event("exited", {"exitCode": debuggee.exit_code})
+            exit_code = debuggee.exit_code
+            self._ide.send_event(
+                "exited", {"exitCode": -1 if exit_code is None else exit_code}
+            )
 
         self._ide.send_event("terminated")
 

src/ptvsd/common/json.py

@@ -103,6 +103,8 @@ def validate(value):
         if (optional and value == ()) or isinstance(value, classinfo):
             return value
         else:
+            if not optional and value == ():
+                raise ValueError("must be specified")
             raise TypeError("must be " + " or ".join(t.__name__ for t in classinfo))
 
     return validate