CODEWITHSUNDEEP
c#.netmultithreadingtaskgrpc
Akash Shrivastava
Akash Shrivastava

Reputation: 1365

gRPC intermittently has high delays

I have a server app (C# with .Net 5) that exposes a gRPC bi-directional endpoint. This endpoint takes in a binary stream in which the server analyzes and produces responses that are sent back to the gRPC response stream.

Each file being sent over gRPC is few megabytes and it takes few minutes for the gRPC call to complete streaming (without latency). With latencies, this time increases sometimes by 50%.

On the client, I have 2 tasks (Task.Run) running, one streaming the file from the clients' file system using FileStream, other reading responses from the server (gRPC).

On the server also, I have 2 tasks running, one reading messages from the gRPC request stream and pushing them into a queue (DataFlow.BufferBlock<byte[]>), other processing messages from the queue, and writing responses to gRPC.

The problem:

If I disable (comment out) all the server processing code, and simply read and log messages from gRPC, there's almost 0 latency from client to server.

When the server has processing enabled, the clients see latencies while writing to grpcClient.

With just 10 active parallel sessions (gRPC Calls) these latencies can go up to 10-15 seconds.

PS: this only happens when I have more than one client running, a higher number of concurrent clients means higher latency.

The client code looks a bit like the below:

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;

Server code looks like:

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

Upvotes: 3

Views: 2814

Answers (2)

Akash Shrivastava
Akash Shrivastava

Reputation: 1365

So, as it turned out, the problem was due to the delay in the number of worker threads spawned by the ThreadPool.

The ThreadPool was taking more time to spawn threads to process these tasks causing gRPC reads to have a significant lag.

This was fixed after increasing the minThread count for spawn requstes using ThreadPool.SetMinThreads. MSDN reference

Upvotes: 2

isandburn
isandburn

Reputation: 154

There have been a number of promising comments on the SO's initial question, but wanted to paraphrase what I thought was important: there's

  1. a an outer async method that calls in to 2
  2. Task.Run()'s - with TaskCreationOptions.LongRunning option that wrap async loops, and finally a
  3. returns a Task.WhenAll() rejoins the two Tasks... Alois Kraus offers that an OS task scheduler is an OS and its scheduling could be abstracting away what you might think is more efficient - this could very well be true and if it is

i would offer the suggestion to try and remove the asynchronous processing and see what kind of benefit difference you might see with various sync/async blends might work better for your particular scenario. one thing to make sure to remember is that asynce/await logically blocks at an expense of automatic thread management - this is great for single-path-ish I/O bound processing (ex. needing to call a db/webservice before moving on to next step of execution) and can be less beneficial as you move toward compute-bound processing (execution that needs to be explicitly re-joined - async/await implicitly take care of Task re-join)

Upvotes: 0

Related Questions