Grpc-dotnet: Thread Safety of IServerStreamWriter.WriteAsync

[generatives]

I want to generate the contents of a stream across multiple threads or tasks. is IServerStreamWriter.WriteAsync thread/task safe? Do I need to setup a separate buffer to get all results onto the original thread?

[davidfowl]

No it it not thread safe, it’s be single producer at a time.

[JamesNK]

You could wrap it in your own API and then ensure only one producer can access it at a time with a SemaphoreSlim.

There are problems with that:

• What happens if the call ends and there are still outstanding producers? They'll all error.
• What happens if producers are writing faster than the client is reading? Producers will hang on WriteAsync.

[Daniel15]

It would be good to document this somewhere. I hit this exception when trying to write to a gRPC stream from multiple Tasks concurrently, and it took a while to find this GitHub issue:

Specified argument was out of the range of valid values. (Parameter 'bytes')
   at System.IO.Pipelines.ThrowHelper.ThrowArgumentOutOfRangeException(ExceptionArgument argument)
   at System.IO.Pipelines.Pipe.Advance(Int32 bytes)
   at System.IO.Pipelines.Pipe.DefaultPipeWriter.Advance(Int32 bytes)
   at Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Infrastructure.PipeWriterHelpers.ConcurrentPipeWriter.Advance(Int32 bytes)
   at Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http2.Http2OutputProducer.Advance(Int32 bytes)
   at Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http.HttpProtocol.Advance(Int32 bytes)
   at Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http.HttpResponsePipeWriter.Advance(Int32 bytes)
   at Grpc.AspNetCore.Server.Internal.PipeExtensions.WriteHeader(PipeWriter pipeWriter, Int32 length, Boolean compress)
   at Grpc.AspNetCore.Server.Internal.PipeExtensions.WriteMessageAsync[TResponse](PipeWriter pipeWriter, TResponse response, HttpContextServerCallContext serverCallContext, Action`2 serializer, Boolean canFlush)
   at DnsTools.Worker.Tools.DnsTraversal.DoLookupImpl(DnsLookupRequest request, IServerStreamWriter`1 responseStream, CancellationToken cancellationToken, String serverName, IDnsQueryResponse responseForGlue, UInt32 level) in C:\src\dnstools.ws\src\DnsTools.Worker\Tools\DnsTraversal.cs:line 158

I think ChannelWriter could be used to handle writing just one message at a time (with multiple producers and one consumer that writes to the stream).

What happens if the call ends and there are still outstanding producers?

@JamesNK Wouldn't you want the call to wait for all the producers to complete?

[Bob9098-source]

Bns

[JamesNK]

IServerStreamWriter.WriteAsync intelli-sense says that only one writer is allowed at a time:

I think ChannelWriter could be used to handle writing just one message at a time (with multiple producers and one consumer that writes to the stream).

Yes you could use channel. Here is an example I implemented that uses channel: https://github.com/grpc/grpc/blob/8577fe8f81399116f2f20fb0869ea3fa0f6dbc5f/src/csharp/Grpc.HealthCheck/HealthServiceImpl.cs

[Daniel15]

IServerStreamWriter.WriteAsync intelli-sense says that only one writer is allowed at a time:

Oops, you're exactly right... I totally missed that.

Yes you could use channel.

This is what I ended up with. Need to do more testing, but it seems to work for my use case:

/// <summary>
/// Wraps <see cref="IServerStreamWriter{T}"/> which only supports one writer at a time.
/// This class can receive messages from multiple threads, and writes them to the stream
/// one at a time.
/// </summary>
/// <typeparam name="T">Type of message written to the stream</typeparam>
public class GrpcStreamResponseQueue<T>
{
    private readonly IServerStreamWriter<T> _stream;
    private readonly Task _consumer;

    private readonly Channel<T> _channel = Channel.CreateUnbounded<T>(
        new UnboundedChannelOptions
        {
            SingleWriter = false,
            SingleReader = true,
        });

    public GrpcStreamResponseQueue(
        IServerStreamWriter<T> stream,
        CancellationToken cancellationToken = default
    )
    {
        _stream = stream;
        _consumer = Consume(cancellationToken);
    }

    /// <summary>
    /// Asynchronously writes an item to the channel.
    /// </summary>
    /// <param name="message">The value to write to the channel.</param>
    /// <param name="cancellationToken">A <see cref="T:System.Threading.CancellationToken" /> used to cancel the write operation.</param>
    /// <returns>A <see cref="T:System.Threading.Tasks.ValueTask" /> that represents the asynchronous write operation.</returns>
    public async ValueTask WriteAsync(T message, CancellationToken cancellationToken = default)
    {
        await _channel.Writer.WriteAsync(message, cancellationToken);
    }

    /// <summary>
    /// Marks the writer as completed, and waits for all writes to complete.
    /// </summary>
    public Task CompleteAsync()
    {
        _channel.Writer.Complete();
        return _consumer;
    }

    private async Task Consume(CancellationToken cancellationToken)
    {
        await foreach (var message in _channel.Reader.ReadAllAsync(cancellationToken))
        {
            await _stream.WriteAsync(message);
        }
    }
}