Tcp mode

CONTRIBUTING.md

@@ -386,8 +386,11 @@ otherwise an empty dictionary (`{}`) should be returned.
 ##### Common Concerns
 
 - some language servers need to have their connection mode specified
-  - the `stdio` interface is the only one supported by `jupyter_lsp`
-    - PRs welcome to support other modes!
+  - `jupyter_lsp` currently supports the `stdio` and `tcp` interface
+  - the mode used by `jupyter_lsp` to connect to the language server can be specified by including `mode="stdio"` or `mode="tcp"` in the language server `spec`-dictionary
+  - currently it is not possible to connect to externally running language servers via tcp, but only to servers spawned by `jupyter_lsp` as given by the `argv` specs entry
+    - PRs welcome to support externally running language servers!
+  - use the placeholder `{port}` within the `argv` entry to allow `jupyter_lsp` to specify the port on which to launch the language server
 - because of its VSCode heritage, many language servers use `nodejs`
   - `LanguageServerManager.nodejs` will provide the location of our best
     guess at where a user's `nodejs` might be found

atest/ports.py

@@ -9,7 +9,7 @@ def get_unused_port():
     Probably could introduce race conditions if inside a tight loop.
     """
     sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-    sock.bind(("localhost", 0))
+    sock.bind(("127.0.0.1", 0))
     sock.listen(1)
     port = sock.getsockname()[1]
     sock.close()

python_packages/jupyter_lsp/jupyter_lsp/stdio.py → python_packages/jupyter_lsp/jupyter_lsp/connection.py

@@ -1,4 +1,4 @@
-""" Language Server stdio-mode readers
+""" Language Server readers and writers
 
 Parts of this code are derived from:
 
@@ -7,32 +7,32 @@
 > > MIT License   https://github.com/palantir/python-jsonrpc-server/blob/0.2.0/LICENSE
 > > Copyright 2018 Palantir Technologies, Inc.
 """
-# pylint: disable=broad-except
-import asyncio
-import io
 import os
-from concurrent.futures import ThreadPoolExecutor
+from abc import ABC, ABCMeta, abstractmethod
 from typing import List, Optional, Text
 
-from tornado.concurrent import run_on_executor
+# pylint: disable=broad-except
+import anyio
+from anyio.streams.buffered import BufferedByteReceiveStream
+from anyio.streams.text import TextSendStream
 from tornado.gen import convert_yielded
 from tornado.httputil import HTTPHeaders
 from tornado.ioloop import IOLoop
 from tornado.queues import Queue
-from traitlets import Float, Instance, default
+from traitlets import Float, Instance, Int, default
 from traitlets.config import LoggingConfigurable
+from traitlets.traitlets import MetaHasTraits
 
-from .non_blocking import make_non_blocking
 
+class LspStreamMeta(MetaHasTraits, ABCMeta):
+    pass
 
-class LspStdIoBase(LoggingConfigurable):
-    """Non-blocking, queued base for communicating with stdio Language Servers"""
 
-    executor = None
+class LspStreamBase(LoggingConfigurable, ABC, metaclass=LspStreamMeta):
+    """Non-blocking, queued base for communicating with Language Servers through anyio
+    streams
+    """
 
-    stream = Instance(
-        io.RawIOBase, help="the stream to read/write"
-    )  # type: io.RawIOBase
     queue = Instance(Queue, help="queue to get/put")
 
     def __repr__(self):  # pragma: no cover
@@ -41,15 +41,14 @@ def __repr__(self):  # pragma: no cover
     def __init__(self, **kwargs):
         super().__init__(**kwargs)
         self.log.debug("%s initialized", self)
-        self.executor = ThreadPoolExecutor(max_workers=1)
 
-    def close(self):
-        self.stream.close()
-        self.log.debug("%s closed", self)
+    @abstractmethod
+    async def close(self):
+        pass  # pragma: no cover
 
 
-class LspStdIoReader(LspStdIoBase):
-    """Language Server stdio Reader
+class LspStreamReader(LspStreamBase):
+    """Language Server Reader
 
     Because non-blocking (but still synchronous) IO is used, rudimentary
     exponential backoff is used.
@@ -58,18 +57,34 @@ class LspStdIoReader(LspStdIoBase):
     max_wait = Float(help="maximum time to wait on idle stream").tag(config=True)
     min_wait = Float(0.05, help="minimum time to wait on idle stream").tag(config=True)
     next_wait = Float(0.05, help="next time to wait on idle stream").tag(config=True)
+    receive_max_bytes = Int(
+        65536,
+        help="the maximum size a header line send by the language server may have",
+    ).tag(config=True)
+
+    stream = Instance(
+        BufferedByteReceiveStream, help="the stream to read from"
+    )  # type: BufferedByteReceiveStream
+
+    def __init__(self, stream: anyio.abc.ByteReceiveStream, **kwargs):
+        super().__init__(**kwargs)
+        self.stream = BufferedByteReceiveStream(stream)
+
+    async def close(self):
+        await self.stream.aclose()
+        self.log.debug("%s closed", self)
 
     @default("max_wait")
     def _default_max_wait(self):
         return 0.1 if os.name == "nt" else self.min_wait * 2
 
     async def sleep(self):
         """Simple exponential backoff for sleeping"""
-        if self.stream.closed:  # pragma: no cover
+        if self.stream._closed:  # pragma: no cover
             return
         self.next_wait = min(self.next_wait * 2, self.max_wait)
         try:
-            await asyncio.sleep(self.next_wait)
+            await anyio.sleep(self.next_wait)
         except Exception:  # pragma: no cover
             pass
 
@@ -79,9 +94,7 @@ def wake(self):
 
     async def read(self) -> None:
         """Read from a Language Server until it is closed"""
-        make_non_blocking(self.stream)
-
-        while not self.stream.closed:
+        while True:
             message = None
             try:
                 message = await self.read_one()
@@ -93,6 +106,10 @@ async def read(self) -> None:
                     self.wake()
 
                 IOLoop.current().add_callback(self.queue.put_nowait, message)
+            except anyio.ClosedResourceError:
+                # stream was closed -> terminate
+                self.log.debug("Stream closed while a read was still in progress")
+                break
             except Exception as e:  # pragma: no cover
                 self.log.exception(
                     "%s couldn't enqueue message: %s (%s)", self, message, e
@@ -124,10 +141,10 @@ async def _read_content(
         while received_size < length and len(raw_parts) < max_parts and max_empties > 0:
             part = None
             try:
-                part = self.stream.read(length - received_size)
-            except OSError:  # pragma: no cover
+                part = await self.stream.receive_exactly(length - received_size)
+            except anyio.IncompleteRead:  # pragma: no cover
                 pass
-            if part is None:
+            if part is None:  # pragma: no cover
                 max_empties -= 1
                 await self.sleep()
                 continue
@@ -171,32 +188,60 @@ async def read_one(self) -> Text:
 
         return message
 
-    @run_on_executor
-    def _readline(self) -> Text:
+    async def _readline(self) -> Text:
         """Read a line (or immediately return None)"""
         try:
-            return self.stream.readline().decode("utf-8").strip()
-        except OSError:  # pragma: no cover
+            # use same max_bytes as is default for receive for now. It seems there is no
+            # way of getting the bytes read until max_bytes is reached, so we cannot
+            # iterate the receive_until call with smaller max_bytes values
+            async with anyio.move_on_after(0.2):
+                line = await self.stream.receive_until(b"\r\n", self.receive_max_bytes)
+                return line.decode("utf-8").strip()
+        except anyio.IncompleteRead:
+            # resource has been closed before the requested bytes could be retrieved
+            # -> signal recource closed
+            raise anyio.ClosedResourceError
+        except anyio.DelimiterNotFound:  # pragma: no cover
+            self.log.error(
+                "Readline hit max_bytes before newline character was encountered"
+            )
             return ""
 
 
-class LspStdIoWriter(LspStdIoBase):
-    """Language Server stdio Writer"""
+class LspStreamWriter(LspStreamBase):
+    """Language Server Writer"""
+
+    stream = Instance(
+        TextSendStream, help="the stream to write to"
+    )  # type: TextSendStream
+
+    def __init__(self, stream: anyio.abc.ByteSendStream, **kwargs):
+        super().__init__(**kwargs)
+        self.stream = TextSendStream(stream, encoding="utf-8")
+
+    async def close(self):
+        await self.stream.aclose()
+        self.log.debug("%s closed", self)
 
     async def write(self) -> None:
         """Write to a Language Server until it closes"""
-        while not self.stream.closed:
+        while True:
             message = await self.queue.get()
             try:
-                body = message.encode("utf-8")
-                response = "Content-Length: {}\r\n\r\n{}".format(len(body), message)
-                await convert_yielded(self._write_one(response.encode("utf-8")))
+                n_bytes = len(message.encode("utf-8"))
+                response = "Content-Length: {}\r\n\r\n{}".format(n_bytes, message)
+                await convert_yielded(self._write_one(response))
+            except (
+                anyio.ClosedResourceError,
+                anyio.BrokenResourceError,
+            ):  # pragma: no cover
+                # stream was closed -> terminate
+                self.log.debug("Stream closed while a write was still in progress")
+                break
             except Exception:  # pragma: no cover
                 self.log.exception("%s couldn't write message: %s", self, response)
             finally:
                 self.queue.task_done()
 
-    @run_on_executor
-    def _write_one(self, message) -> None:
-        self.stream.write(message)
-        self.stream.flush()
+    async def _write_one(self, message) -> None:
+        await self.stream.send(message)

python_packages/jupyter_lsp/jupyter_lsp/manager.py

@@ -1,4 +1,4 @@
-""" A configurable frontend for stdio-based Language Servers
+""" A configurable frontend for stream-based Language Servers
 """
 import os
 import traceback
@@ -22,7 +22,11 @@
     EP_SPEC_V1,
 )
 from .schema import LANGUAGE_SERVER_SPEC_MAP
-from .session import LanguageServerSession
+from .session import (
+    LanguageServerSessionBase,
+    LanguageServerSessionStdio,
+    LanguageServerSessionTCP,
+)
 from .trait_types import LoadableCallable, Schema
 from .types import (
     KeyedLanguageServerSpecs,
@@ -55,10 +59,10 @@ class LanguageServerManager(LanguageServerManagerAPI):
     )  # type: bool
 
     sessions = Dict_(
-        trait=Instance(LanguageServerSession),
+        trait=Instance(LanguageServerSessionBase),
         default_value={},
         help="sessions keyed by language server name",
-    )  # type: Dict[Tuple[Text], LanguageServerSession]
+    )  # type: Dict[Tuple[Text], LanguageServerSessionBase]
 
     virtual_documents_dir = Unicode(
         help="""Path to virtual documents relative to the content manager root
@@ -137,9 +141,21 @@ def init_sessions(self):
         """create, but do not initialize all sessions"""
         sessions = {}
         for language_server, spec in self.language_servers.items():
-            sessions[language_server] = LanguageServerSession(
-                language_server=language_server, spec=spec, parent=self
-            )
+            mode = spec.get("mode", "stdio")
+            if mode == "stdio":
+                sessions[language_server] = LanguageServerSessionStdio(
+                    language_server=language_server, spec=spec, parent=self
+                )
+            elif mode == "tcp":
+                sessions[language_server] = LanguageServerSessionTCP(
+                    language_server=language_server, spec=spec, parent=self
+                )
+            else:  # pragma: no cover
+                raise ValueError(
+                    "Unknown session mode {} for language server '{}'".format(
+                        mode, language_server
+                    )
+                )
         self.sessions = sessions
 
     def init_listeners(self):

python_packages/jupyter_lsp/jupyter_lsp/schema/schema.json

@@ -139,6 +139,13 @@
           "description": "list of MIME types supported by the language server",
           "title": "MIME Types"
         },
+        "mode": {
+          "description": "connection mode used, e.g. stdio (default), tcp",
+          "title": "Mode",
+          "type": "string",
+          "enum": ["stdio", "tcp"],
+          "default": "stdio"
+        },
         "troubleshoot": {
           "type": "string",
           "description": "information on troubleshooting the installation or auto-detection of the language server",