Add --node-ipc support #2015
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -5,8 +5,9 @@ | |
| from contextlib import closing | ||
| from functools import partial | ||
| from queue import Queue | ||
| import http.client | ||
| import json | ||
| import multiprocessing.connection | ||
| import os | ||
| import shutil | ||
| import socket | ||
|
|
@@ -48,57 +49,91 @@ def on_stderr_message(self, message: str) -> None: | |
|
|
||
| class AbstractProcessor(Generic[T]): | ||
|
|
||
| def write_data(self, data: T) -> None: | ||
| raise NotImplementedError() | ||
|
|
||
| def read_data(self) -> Optional[T]: | ||
| raise NotImplementedError() | ||
|
|
||
|
|
||
| def encode_payload(data: Dict[str, Any]) -> bytes: | ||
| return json.dumps( | ||
| data, | ||
| ensure_ascii=False, | ||
| check_circular=False, | ||
| separators=(',', ':') | ||
| ).encode('utf-8') | ||
|
|
||
|
|
||
| def decode_payload(message: bytes) -> Optional[Dict[str, Any]]: | ||
| try: | ||
| return json.loads(message.decode('utf-8')) | ||
| except Exception as ex: | ||
| exception_log("JSON decode error", ex) | ||
| return None | ||
alecmev marked this conversation as resolved.
Show resolved
Hide resolved
|
||
|
|
||
|
|
||
| class StandardProcessor(AbstractProcessor[Dict[str, Any]]): | ||
|
|
||
| def __init__(self, reader: IO[bytes], writer: IO[bytes]): | ||
| self._reader = reader | ||
| self._writer = writer | ||
|
|
||
| def write_data(self, data: Dict[str, Any]) -> None: | ||
| body = encode_payload(data) | ||
| self._writer.writelines(("Content-Length: {}\r\n\r\n".format(len(body)).encode('ascii'), body)) | ||
| self._writer.flush() | ||
|
|
||
| def read_data(self) -> Optional[Dict[str, Any]]: | ||
| headers = http.client.parse_headers(self._reader) # type: ignore | ||
| try: | ||
| body = self._reader.read(int(headers.get("Content-Length"))) | ||
| except TypeError: | ||
| # Expected error on process stopping. Stop the read loop. | ||
| raise StopLoopError() | ||
| return decode_payload(body) | ||
|
|
||
|
|
||
| class NodeIpcProcessor(AbstractProcessor[Dict[str, Any]]): | ||
| _buf = bytearray() | ||
| _lines = 0 | ||
|
There was a problem hiding this comment. Shouldn't these two properties be part of the class instance, instead of the class? If I have LSP-dockerfile, LSP-stylelint, and LSP-typescript running with node IPC, wouldn't these overwrite each other in unpredictable ways? Maybe I'm not understanding something. There was a problem hiding this comment. I believe it should work as expected (in most cases ;)) due to how Python works with those. When instance of the class reads those properties it looks them up first in the instance and then in the class. When writing it always writes to the instance property (creates it if necessary). The only issue would be if the instance would read a class property (for example So I guess it's still safer to set those on the instance. |
||
|
|
||
| def __init__(self, conn: multiprocessing.connection._ConnectionBase): | ||
| self._conn = conn | ||
|
|
||
| def write_data(self, data: Dict[str, Any]) -> None: | ||
| body = encode_payload(data) + b"\n" | ||
| while len(body): | ||
| n = self._conn._write(self._conn.fileno(), body) # type: ignore | ||
| body = body[n:] | ||
|
|
||
| def read_data(self) -> Optional[Dict[str, Any]]: | ||
| while self._lines == 0: | ||
| chunk = self._conn._read(self._conn.fileno(), 65536) # type: ignore | ||
| if len(chunk) == 0: | ||
| # EOF reached: https://docs.python.org/3/library/os.html#os.read | ||
| raise StopLoopError() | ||
|
|
||
| self._buf += chunk | ||
| self._lines += chunk.count(b'\n') | ||
|
|
||
| self._lines -= 1 | ||
| message, _, self._buf = self._buf.partition(b'\n') | ||
| return decode_payload(message) | ||
|
|
||
|
|
||
| class ProcessTransport(Transport[T]): | ||
|
|
||
| def __init__(self, | ||
| name: str, | ||
| process: subprocess.Popen, | ||
| socket: Optional[socket.socket], | ||
| stderr: Optional[IO[bytes]], | ||
| processor: AbstractProcessor[T], | ||
| callback_object: TransportCallbacks[T]) -> None: | ||
| self._closed = False | ||
| self._process = process | ||
| self._socket = socket | ||
| self._stderr = stderr | ||
| self._processor = processor | ||
| self._reader_thread = threading.Thread(target=self._read_loop, name='{}-reader'.format(name)) | ||
|
|
@@ -136,8 +171,8 @@ def __del__(self) -> None: | |
|
|
||
| def _read_loop(self) -> None: | ||
| try: | ||
| while True: | ||
|
There was a problem hiding this comment. I tried to Google this and also replicate it in REPL with There was a problem hiding this comment. @predragnikolic Do you know anything about this? Do There was a problem hiding this comment.
No, I don't know, I just reverted the old code. I can just guess and say that def close(self) -> None:
if not self._closed:
self._send_queue.put_nowait(None)
if self._socket:
self._socket.close()
self._closed = True
+ self._reader.close()
+ self._writer.close()
+. self._reader = None
+. self._writer = NoneThere was a problem hiding this comment. Don't know about stdio and TCP, but the multiprocessing pipe is garbage-collectable: https://docs.python.org/3/library/multiprocessing.html#multiprocessing.connection.Connection.close |
||
| payload = self._processor.read_data() | ||
| if payload is None: | ||
| continue | ||
|
|
||
|
|
@@ -186,12 +221,11 @@ def invoke() -> None: | |
| def _write_loop(self) -> None: | ||
| exception = None # type: Optional[Exception] | ||
| try: | ||
| while True: | ||
| d = self._send_queue.get() | ||
| if d is None: | ||
| break | ||
| self._processor.write_data(d) | ||
| except (BrokenPipeError, AttributeError): | ||
| pass | ||
| except Exception as ex: | ||
|
|
@@ -219,28 +253,32 @@ def _stderr_loop(self) -> None: | |
| self._send_queue.put_nowait(None) | ||
|
|
||
|
|
||
| def create_transport(config: TransportConfig, cwd: Optional[str], | ||
| callback_object: TransportCallbacks) -> Transport[Dict[str, Any]]: | ||
| stderr = subprocess.PIPE | ||
| pass_fds = () # type: Union[Tuple[()], Tuple[int]] | ||
| if config.tcp_port is not None: | ||
| assert config.tcp_port is not None | ||
| if config.tcp_port < 0: | ||
| stdout = subprocess.PIPE | ||
| else: | ||
| stdout = subprocess.DEVNULL | ||
| stdin = subprocess.DEVNULL | ||
| elif not config.node_ipc: | ||
| stdout = subprocess.PIPE | ||
| stdin = subprocess.PIPE | ||
| else: | ||
| stdout = subprocess.PIPE | ||
| stdin = subprocess.DEVNULL | ||
| stderr = subprocess.STDOUT | ||
| pass_fds = (config.node_ipc.child_conn.fileno(),) | ||
|
|
||
| startupinfo = _fixup_startup_args(config.command) | ||
| sock = None # type: Optional[socket.socket] | ||
| process = None # type: Optional[subprocess.Popen] | ||
|
|
||
| def start_subprocess() -> subprocess.Popen: | ||
| return _start_subprocess(config.command, stdin, stdout, stderr, startupinfo, config.env, cwd, pass_fds) | ||
|
|
||
| if config.listener_socket: | ||
| assert isinstance(config.tcp_port, int) and config.tcp_port > 0 | ||
|
|
@@ -250,21 +288,27 @@ def start_subprocess() -> subprocess.Popen: | |
| config.listener_socket, | ||
| start_subprocess | ||
| ) | ||
| processor = StandardProcessor(reader, writer) # type: AbstractProcessor | ||
| else: | ||
| process = start_subprocess() | ||
| if config.tcp_port: | ||
| sock = _connect_tcp(config.tcp_port) | ||
| if sock is None: | ||
| raise RuntimeError("Failed to connect on port {}".format(config.tcp_port)) | ||
| reader = writer = sock.makefile('rwb') | ||
| processor = StandardProcessor(reader, writer) | ||
| elif not config.node_ipc: | ||
| if not process.stdout or not process.stdin: | ||
predragnikolic marked this conversation as resolved.
Show resolved
Hide resolved
|
||
| raise RuntimeError( | ||
| 'Failed initializing transport: reader: {}, writer: {}' | ||
| .format(process.stdout, process.stdin) | ||
| ) | ||
| processor = StandardProcessor(process.stdout, process.stdin) | ||
| else: | ||
| processor = NodeIpcProcessor(config.node_ipc.parent_conn) | ||
|
|
||
| stderr_reader = process.stdout if config.node_ipc else process.stderr | ||
| return ProcessTransport(config.name, process, sock, stderr_reader, processor, callback_object) | ||
|
|
||
|
|
||
| _subprocesses = weakref.WeakSet() # type: weakref.WeakSet[subprocess.Popen] | ||
|
|
@@ -312,7 +356,8 @@ def _start_subprocess( | |
| stderr: int, | ||
| startupinfo: Any, | ||
| env: Dict[str, str], | ||
| cwd: Optional[str], | ||
| pass_fds: Union[Tuple[()], Tuple[int]] | ||
| ) -> subprocess.Popen: | ||
| debug("starting {} in {}".format(args, cwd if cwd else os.getcwd())) | ||
| process = subprocess.Popen( | ||
|
|
@@ -322,7 +367,8 @@ def _start_subprocess( | |
| stderr=stderr, | ||
| startupinfo=startupinfo, | ||
| env=env, | ||
| cwd=cwd, | ||
| pass_fds=pass_fds) | ||
| _subprocesses.add(process) | ||
| return process | ||
|
|
||
|
|
@@ -356,8 +402,7 @@ def start_in_background(d: _SubprocessData) -> None: | |
| # Await one client connection (blocking!) | ||
| sock, _ = listener_socket.accept() | ||
| thread.join() | ||
| reader = writer = sock.makefile('rwb') | ||
| assert data.process | ||
| return data.process, sock, reader, writer | ||
|
|
||
|
|
||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
fits one line now :)