用非阻塞方式与clangd通信的Python

Question

我正在为我的一个爱好项目实现一个插件，基本上我想使用LSP作为网站上的高亮代码。

在我的例子中，我想使用clangd来收集关于预先编写的C和C++代码的信息。我面临的问题是如何准确地与clangd通信--我很难通过管道stdin和stdout发送和接收JSON。

最大的问题是：

• 该协议没有像\0那样的严格标记，只有Content-Length报头。
• 管道默认是阻塞的，这会挂起我的python代码，它不知道什么时候停止读取(过多的读取=挂起，直到clangd产生更多的死锁，这会使我的程序在执行更多的LSP调用之前等待更多的输出)。
• 没有其他的通信方式--只有I/O流。我还没找到任何类似的选择和消息来源。

到目前为止，我的代码已经足够启动clangd并发送第一个初始化JSON了，但是我不知道如何在没有死锁或挂起的情况下继续处理JSON交换。

import os
import json
import subprocess
import shutil
from typing import Union, List, Dict, Tuple


def make_json_rpc_request(id: Union[str, int], method: str, params: Union[Tuple, List, Dict]):
    if not isinstance(id, (str, int)):
        raise RuntimeError(f"id should be a number or a string: {id}")

    request = {
        "jsonrpc": "2.0",
        "id": id,
        "method": method
    }

    if params is not None:
        if isinstance(params, (list, tuple, dict)):
            request["params"] = params
        else:
            raise RuntimeError(f"params is not a structured type: {params}")

    return request


def make_lsp_request(json_rpc_request):
    string = json.dumps(json_rpc_request, indent=None)
    string = f"Content-Length: {len(string)}\r\n\r\n{string}"
    return string

def get_clangd_path():
    result = shutil.which("clangd")
    if result:
        return result

    env_name = os.environ.get("CLANGD")
    if env_name:
        result = shutil.which(env_name)
        if result:
            return result

    raise RuntimeError("clangd not found. Specify env variable CLANGD that points to the executable or to a name searchable in PATH")


class Connection:
    def __init__(self):
        self.clangd_path = get_clangd_path()
        self.p = subprocess.Popen([self.clangd_path], stdin=subprocess.PIPE, stdout=subprocess.PIPE)
        self.id = 1

    def send(self, method: str, params):
        request = make_lsp_request(make_json_rpc_request(self.id, method, params))
        self.p.stdin.write(request.encode())
        self.p.stdin.flush()
        self.id += 1


if __name__ == "__main__":
    conn = Connection()
    conn.send("initialize", {"params": {
        "processId": None,
            "rootUri": None,
            "capabilities": {
        }
    }})
    print(conn.p.stdout.read1())

我尝试了在在subprocess.PIPE上使用Python进行非阻塞读取中提出的解决方案，但什么也没成功。

目标:拥有类似于make_lsp_call(self, method, params, timeout)的内容，它可以返回：

• 从clangd接收到的JSON
• 接收的错误消息
• None如果超时

编辑:工作解决方案：

    HEADER_CONTENT_LENGTH = "Content-Length: "

    def receive(self, id: int):
        headers = []

        while True:
            line = self.p.stdout.readline()
            if line != b"\r\n":
                headers.append(line)
            else:
                break

        length = 0
        for hdr in headers:
            hdr = hdr.decode()
            if HEADER_CONTENT_LENGTH in hdr:
                length = int(hdr.removeprefix(HEADER_CONTENT_LENGTH))
                break

        if length == 0:
            raise RuntimeError(f"invalid or missing '{HEADER_CONTENT_LENGTH}' header")

        return self.p.stdout.read(length).decode()

Answer 1

该协议没有像\0这样的严格标记，只有内容长度标头

根据LSP文档，标题部分与CRLFCRLF内容分离，每个标头被CRLF分隔，就像在HTTP中一样。

IOW，您不希望使用.read()读取管道中的所有内容，而是读取一条消息：

1. readline()，直到你得到一个空行，把每个标题行的内容放在例如一个dict中
2. 如果您没有得到内容长度标头，则另一端违反规范，这是一个致命错误。
3. 准确地用Content-Length读取.read(n)字节。
4. 重复-下一个你应该得到的是另一个标题。

为此打开一个异步蠕虫罐似乎是不必要的。