如何用Python以非阻塞方式与clangd通信实现网页C/C++代码高亮？

问题背景 #

我正在为个人项目开发插件，计划通过LSP（语言服务器协议）实现网页中代码高亮功能，具体使用clangd获取C/C++代码的相关信息，但在通过管道标准输入输出与clangd收发JSON数据时遇到了困难。

核心问题 #

• 协议没有\0这类严格分隔符，仅依赖Content-Length头部
• 管道默认是阻塞模式，导致Python代码因无法确定读取终止时机而挂起，进而引发死锁
• 仅能使用I/O流进行通信，且在clangd的选项和源码中未找到替代通信方式

当前代码 #

现有代码能够启动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())

尝试过的方案 #

尝试过Python中非阻塞读取subprocess管道的相关解决方案，但均未奏效。

目标需求 #

希望实现一个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()

内容的提问来源于stack exchange，提问作者Xeverous