问gRPC间歇性高延迟

EN

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);
}