我有一个公开 gRPC 双向端点的服务器应用程序(C# 和 .Net 5)。此端点接收一个二进制流,服务器在其中分析并生成响应,这些响应被发送回 gRPC 响应流。
通过 gRPC 发送的每个文件都只有几兆字节,gRPC 调用需要几分钟才能完成流式传输(无延迟)。由于延迟,这个时间有时会增加 50%。
在客户端,我有 2 个任务 ( Task.Run
) 正在运行,其中一个使用 FileStream
从客户端的文件系统流式传输文件, 其他读取来自服务器 (gRPC) 的响应。
同样在服务器上,我有 2 个任务在运行,一个从 gRPC 请求流中读取消息并将它们推送到队列中 (DataFlow.BufferBlock<byte[]>
),另一个处理队列中的消息,并将响应写入 gRPC。
问题:
如果我禁用(注释掉)所有服务器处理代码,并简单地从 gRPC 读取和记录消息,那么从客户端到服务器的延迟几乎为 0。
当服务器启用处理时,客户端在写入 grpcClient 时会看到延迟。
只有 10 个事件的并行 session (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);
}
最佳答案
因此,事实证明,问题是由于 ThreadPool
产生的工作线程数量延迟造成的。
ThreadPool
需要更多时间来生成线程来处理这些任务,导致 gRPC 读取有明显的滞后。
在使用 ThreadPool.SetMinThreads
增加生成请求的 minThread
计数后,此问题得到修复。 MSDN reference
关于c# - gRPC 间歇性地具有高延迟,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/68498455/