Merge pull request #494 from dpriedel/FixTruncatedJediResponses

Attempt to fix truncated Jedi LSP responses

Author: krassowski

atest/05_Features/Completion.robot

@@ -299,6 +299,7 @@ Trigger Completer
 Completer Should Include Documentation
     [Arguments]    ${text}
     Wait Until Page Contains Element    ${DOCUMENTATION_PANEL}    timeout=10s
+    Wait Until Keyword Succeeds    10 x    1 s    Element Should Contain    ${DOCUMENTATION_PANEL}    ${text}
     Element Should Contain    ${DOCUMENTATION_PANEL}    ${text}
 
 Restart Kernel

python_packages/jupyter_lsp/jupyter_lsp/session.py

@@ -137,6 +137,7 @@ def init_process(self):
             stdin=subprocess.PIPE,
             stdout=subprocess.PIPE,
             env=self.substitute_env(self.spec.get("env", {}), os.environ),
+            bufsize=0,
         )
 
     def init_queues(self):

python_packages/jupyter_lsp/jupyter_lsp/stdio.py

@@ -31,8 +31,8 @@ class LspStdIoBase(LoggingConfigurable):
     executor = None
 
     stream = Instance(
-        io.BufferedIOBase, help="the stream to read/write"
-    )  # type: io.BufferedIOBase
+        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
@@ -61,7 +61,7 @@ class LspStdIoReader(LspStdIoBase):
 
     @default("max_wait")
     def _default_max_wait(self):
-        return 2.0 if os.name == "nt" else self.min_wait
+        return 0.1 if os.name == "nt" else self.min_wait * 2
 
     async def sleep(self):
         """Simple exponential backoff for sleeping"""
@@ -93,12 +93,17 @@ async def read(self) -> None:
                     self.wake()
 
                 IOLoop.current().add_callback(self.queue.put_nowait, message)
-            except Exception:  # pragma: no cover
-                self.log.exception("%s couldn't enqueue message: %s", self, message)
+            except Exception as e:  # pragma: no cover
+                self.log.exception(
+                    "%s couldn't enqueue message: %s (%s)", self, message, e
+                )
                 await self.sleep()
 
-    def _read_content(self, length: int, max_parts=1000) -> Optional[bytes]:
-        """Read the full length of the message unless exceeding max_parts.
+    async def _read_content(
+        self, length: int, max_parts=1000, max_empties=200
+    ) -> Optional[bytes]:
+        """Read the full length of the message unless exceeding max_parts or
+           max_empties empty reads occur.
 
         See https://github.com/krassowski/jupyterlab-lsp/issues/450
 
@@ -116,19 +121,27 @@ def _read_content(self, length: int, max_parts=1000) -> Optional[bytes]:
         raw = None
         raw_parts: List[bytes] = []
         received_size = 0
-        while received_size < length and len(raw_parts) < max_parts:
-            part = self.stream.read(length)
+        while received_size < length and len(raw_parts) < max_parts and max_empties > 0:
+            part = None
+            try:
+                part = self.stream.read(length)
+            except OSError:  # pragma: no cover
+                pass
             if part is None:
-                break  # pragma: no cover
+                max_empties -= 1
+                await self.sleep()
+                continue
             received_size += len(part)
             raw_parts.append(part)
 
         if raw_parts:
             raw = b"".join(raw_parts)
             if len(raw) != length:  # pragma: no cover
                 self.log.warning(
-                    f"Readout and content-length mismatch:" f" {len(raw)} vs {length}"
+                    f"Readout and content-length mismatch: {len(raw)} vs {length};"
+                    f"remaining empties: {max_empties}; remaining parts: {max_parts}"
                 )
+
         return raw
 
     async def read_one(self) -> Text:
@@ -146,24 +159,15 @@ async def read_one(self) -> Text:
             content_length = int(headers.get("content-length", "0"))
 
             if content_length:
-                raw = None
-                retries = 5
-                while raw is None and retries:
-                    try:
-                        raw = self._read_content(length=content_length)
-                    except OSError:  # pragma: no cover
-                        raw = None
-                    if raw is None:  # pragma: no cover
-                        self.log.warning(
-                            "%s failed to read message of length %s",
-                            self,
-                            content_length,
-                        )
-                        await self.sleep()
-                        retries -= 1
-                    else:
-                        message = raw.decode("utf-8").strip()
-                        break
+                raw = await self._read_content(length=content_length)
+                if raw is not None:
+                    message = raw.decode("utf-8").strip()
+                else:  # pragma: no cover
+                    self.log.warning(
+                        "%s failed to read message of length %s",
+                        self,
+                        content_length,
+                    )
 
         return message
 

python_packages/jupyter_lsp/jupyter_lsp/tests/test_stdio.py

@@ -0,0 +1,74 @@
+import asyncio
+import subprocess
+
+import pytest
+from tornado.queues import Queue
+
+from jupyter_lsp.stdio import LspStdIoReader
+
+WRITER_TEMPLATE = """
+from time import sleep
+
+print('Content-Length: {length}')
+print()
+
+for repeat in range({repeats}):
+    sleep({interval})
+    print('{message}', end='')
+print()
+"""
+
+
+class CommunicatorSpawner:
+    def __init__(self, tmp_path):
+        self.tmp_path = tmp_path
+
+    def spawn_writer(self, message: str, repeats: int = 1, interval=None):
+        length = len(message) * repeats
+        commands_file = self.tmp_path / "writer.py"
+        commands_file.write_text(
+            WRITER_TEMPLATE.format(
+                length=length, repeats=repeats, interval=interval or 0, message=message
+            )
+        )
+        return subprocess.Popen(
+            ["python", "-u", str(commands_file)], stdout=subprocess.PIPE, bufsize=0
+        )
+
+
+@pytest.fixture
+def communicator_spawner(tmp_path):
+    return CommunicatorSpawner(tmp_path)
+
+
+async def join_process(process: subprocess.Popen, headstart=1, timeout=1):
+    await asyncio.sleep(headstart)
+    result = process.wait(timeout=timeout)
+    if process.stdout:
+        process.stdout.close()
+    return result
+
+
+@pytest.mark.parametrize(
+    "message,repeats,interval",
+    [
+        ["short", 1, None],
+        ["ab" * 10_0000, 1, None],
+        ["ab", 2, 0.01],
+        ["ab", 45, 0.01],
+    ],
+    ids=["short", "long", "intermittent", "intensive-intermittent"],
+)
+@pytest.mark.asyncio
+async def test_reader(message, repeats, interval, communicator_spawner):
+    queue = Queue()
+
+    process = communicator_spawner.spawn_writer(
+        message=message, repeats=repeats, interval=interval
+    )
+    reader = LspStdIoReader(stream=process.stdout, queue=queue)
+
+    await asyncio.gather(join_process(process, headstart=3, timeout=1), reader.read())
+
+    result = queue.get_nowait()
+    assert result == message * repeats