gRPC 间歇性地具有高延迟

Posted 2023-04-12

【中文标题】gRPC 间歇性地具有高延迟

【英文标题】：gRPC intermittently has high delays

【发布时间】：2021-09-30 12:55:28

【问题描述】：

我有一个公开 gRPC 双向端点的服务器应用程序（带有 .Net 5 的 C#）。该端点接收一个二进制流，服务器在其中分析并生成响应，然后发送回 gRPC 响应流。

通过 gRPC 发送的每个文件都是几兆字节，gRPC 调用需要几分钟才能完成流式传输（无延迟）。随着延迟，这个时间有时会增加 50%。

在客户端，我有 2 个任务 (Task.Run) 正在运行，一个使用 FileStream 从客户端的文件系统流式传输文件，另一个从服务器 (gRPC) 读取响应。

在服务器上，我也有 2 个任务正在运行，一个从 gRPC 请求流中读取消息并将它们推送到队列 (DataFlow.BufferBlock<byte[]>)，另一个处理来自队列的消息，并将响应写入 gRPC。

问题：

如果我禁用（注释掉）所有服务器处理代码，并简单地从 gRPC 读取和记录消息，那么从客户端到服务器的延迟几乎为 0。

当服务器启用处理时，客户端在写入 grpcClient 时会看到延迟。

只有 10 个活动的并行会话（gRPC 调用），这些延迟可能会达到 10-15 秒。

PS：只有当我运行多个客户端时才会发生这种情况，并发客户端数量越多意味着延迟越长。

---

客户端代码如下所示：

FileStream fs = new(audioFilePath, FileMode.Open, FileAccess.Read, FileShare.Read, 1024 * 1024, true);

byte[] buffer = new byte[10_000];

GrpcClient client = new GrpcClient(_singletonChannel); // using single channel since only 5-10 clients are there right now
BiDiCall call = client.BiDiService(hheaders: null, deadline: null, CancellationToken.None);

var writeTask = Task.Run(async () => 
    while (fs.ReadAsync(buffer, 0, buffer.Length))
    
        call.RequestStream.WriteAsync(new()  Chunk = ByteString.CopyFrom(buffer) );
    
    await call.RequestStream.CompleteAsync();
);

var readTask = Task.Run(async () => 
    while (await call.ResponseStream.MoveNext())
    
        // write to log call.ResponseStream.Current
    
);

await Task.WhenAll(writeTask, readTask);
await call;

---

服务器代码如下：

readonly BufferBlock<MessageRequest> messages = new();
MessageProcessor _processor = new();

public override async Task BiDiService(IAsyncStreamReader<MessageRequest> requestStream,
    IServerStreamWriter<MessageResponse> responseStream, 
    ServerCallContext context)

    var readTask = TaskFactory.StartNew(() => 
        while (await requestStream.MoveNext())
        
            messages.Post(requestStream.Current);  // add to queue
        
        messages.Complete();
    , TaskCreationOptions.LongRunning).ConfigureAwait(false);

    var processTask = Task.Run(() => 
        while (await messages.OutputAvailableAsync())
        
            var message = await messages.ReceiveAsync();  // pick from queue
            // if I comment out below line and run with multiple clients = latency disappears
            var result = await _processor.Process(message); // takes some time to process
            if (result.IsImportantForClient())
                await responseStrem.WriteAsync(result.Value);
        
    );

    await Task.WhenAll(readTask, processTask);

【问题讨论】：

因此，当许多客户端在流中写入时，您会看到更高的延迟，但您不知道为什么……是这样吗？

附带说明，建议您从 var readTask = Task.Factory.StartNew(async () => 切换到 var readTask = Task.Run(async () =>，原因已解释为 here。

感谢@TheodorZoulias，我最初使用的是Task.Run，但我正在尝试LongRunning 选项（我还删除了所有awaits，因此循环在单个线程上运行），忘记改回Task.Run，但请注意，现在改回来了。

在速度慢时进行内存转储。当您运行的 CPU 绑定的并发任务过多时，这看起来像是 CPU 超额订阅问题。您正在运行多少个并发下载，您的计算机上有多少个内核，通过网络传输多少 MBit/s？

您编写的任务在概念上是独立的。但实际上，您的独立事物需要在真正的 CPU 上运行。如果你有一个 CPU 多于一件事情要同时执行，那么事情就会在运行时变得依赖。当您要运行的东西多于核心时，操作系统调度程序（Windows，Linux 无关紧要）将需要选择下一个要运行的线程。其他太多的工作都交给了 OS 调度程序的就绪队列。现在您的工作会排在队列中，这就是您现在看到的延迟。

【参考方案1】：

对于 SO 的最初问题，有许多很有前途的 cmets，但想解释一下我认为重要的内容：有

一个调用 2 的外部异步方法 Task.Run()'s - 带有包装异步循环的 TaskCreationOptions.LongRunning 选项，最后是一个 返回一个任务。WhenAll() 重新加入两个任务... Alois Kraus 提出，OS 任务调度程序就是一个 OS，它的调度可以抽象出你认为更有效的东西——这很可能是真的，如果它是的话

我会建议尝试删除异步处理，看看您可能会看到各种同步/异步混合的哪些好处差异可能更适合您的特定场景。 要确保记住的一件事是 asynce/await 在逻辑上阻塞会以牺牲自动线程管理为代价——这对于单路径 I/O 绑定处理非常有用（例如，在继续之前需要调用 db/webservice下一步执行），并且随着您转向计算绑定处理（需要显式重新加入的执行 - async/await 隐式处理任务重新加入），可能不太有用

【参考方案2】：

因此，事实证明，问题是由于ThreadPool 产生的工作线程数量延迟造成的。

ThreadPool 需要更多时间来生成线程来处理这些任务，导致 gRPC 读取出现明显滞后。

在使用ThreadPool.SetMinThreads 增加生成请求的minThread 计数后，此问题已得到修复。 MSDN reference