【发布时间】: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 调度程序的就绪队列。现在您的工作会排在队列中,这就是您现在看到的延迟。
标签: c# .net multithreading task grpc