title: How to control the write buffer size in grpc (or: how to handle slow stream readers in grpc)?
body: <p>We have been using bidirectional grpc streams in our architecture extensively, but have been experiencing issues when the CPU load on one end does not allow it to read data as fast as the other end is sending it. I have been struggling to understand how grpc is supposed to handle this situation. I've made a test program that mimics in a simple fashion this situation (using C# and the async APIs which we use exclusively). The server side is unable to keep up with the rate of data coming from the client. The result is the client side's memory usage continually climbs until ultimately OOMing the machine. My understanding is grpc is supposed to have some sort of protection against this. </p>
  
  <p>I found the channel argument named GRPC_ARG_HTT2_WRITE_BUFFER_SIZE thinking it might limit the client side memory usage (and either cause blocking or an exception on the client end once that buffer is filled). This test program is sending 10000 byte messages and the write buffer size is set to 11000, yet it appears setting this argument has no effect.</p>
  
  <p>My questions are: is grpc behaving correctly in this example and if so what am I doing incorrectly? Why does GRPC_ARG_HTT2_WRITE_BUFFER_SIZE seem to have no effect (and perhaps what is the intended effect)?</p>
  
  <p>The sample is written using the following message/service:</p>
  
  <pre><code>syntax = "proto3";
  package test;
  option csharp_namespace = "Test";
  
  service TestService {
    rpc Publish(stream TestMsg) returns (stream TestMsg);
  }
  
  message TestMsg {
    string value = 1;
    bytes dummy = 2;
  }
  </code></pre>
  
  <p>and the C# program is here:</p>
  
  <pre><code>using System;
  using System.Collections.Generic;
  using System.Threading;
  using System.Threading.Tasks;
  using Google.Protobuf;
  using Grpc.Core;
  
  namespace Test
  {
    class Program
    {
      static void Main( string[] args )
      {
        if ( args[0] == "client" )
        {
          using ( new Client() )
            Console.ReadKey();
        }
        else if ( args[0] == "server" )
        {
          using ( new Server() )
            Console.ReadKey();
        }
      }
    }
  
    public class Server : TestService.TestServiceBase, IDisposable
    {
      private readonly Grpc.Core.Server _server;
  
      public Server()
      {
        _server = new Grpc.Core.Server
        {
          Services = { TestService.BindService( this ) },
          Ports = { new ServerPort( "localhost", 1234, ServerCredentials.Insecure ) }
        };
        _server.Start();
        Console.WriteLine( "Server started" );
      }
  
      public void Dispose()
      {
        _server.ShutdownAsync();
      }
  
      public override async Task Publish(
        IAsyncStreamReader&lt;TestMsg&gt; requestStream,
        IServerStreamWriter&lt;TestMsg&gt; responseStream,
        ServerCallContext context )
      {
        try
        {
          Console.WriteLine( "Client connected" );
  
          for (; ; )
          {
            var requestTask = requestStream.MoveNext();
            await requestTask;
  
            if ( requestTask.Status == TaskStatus.Canceled )
              break;
            if ( !requestTask.Result )
              break;
  
            var request = requestStream.Current;
            if ( request != null )
            {
              try
              {
                Console.Write( request.Value + ".." );
                // We're really working hard.
                Thread.Sleep( 1000 );
              }
              catch ( Exception ex )
              {
                await responseStream.WriteAsync( new TestMsg { Value = ex.Message } );
                Console.WriteLine( ex );
              }
            }
          }
        }
        finally
        {
          Console.WriteLine( "Client disconnected" );
        }
      }
    }
  
    public class Client : IDisposable
    {
      private bool _isSending;
      private readonly TestService.TestServiceClient _client;
      private readonly Channel _channel;
      private readonly IClientStreamWriter&lt;TestMsg&gt; _requestStream;
      private readonly Timer _timer;
      private int _i;
  
      public Client()
      {
        Console.WriteLine( "Client started" );
  
        var options = new List&lt;ChannelOption&gt;();
        options.Add( new ChannelOption( "GRPC_ARG_HTTP2_WRITE_BUFFER_SIZE", 11000 ) );
  
        _channel = new Channel( "localhost:1234", ChannelCredentials.Insecure, options );
        _client = new TestService.TestServiceClient( _channel );
        var call = _client.Publish();
        _requestStream = call.RequestStream;
  
        _timer = new Timer( OnTimerTick, null, 0, 1000 / 25 );
      }
  
      public void Dispose()
      {
        _channel.ShutdownAsync();
        _timer.Dispose();
      }
  
      private async void OnTimerTick( object state )
      {
        // Skip timer ticks if the previous isn't done yet.
        if ( _isSending )
          return;
  
        try
        {
  
          _isSending = true;
          var bytes = new byte[10000];
          var msg = new TestMsg { Value = _i.ToString(), Dummy = ByteString.CopyFrom( bytes ) };
  
          Console.Write( _i + ".." );
          await _requestStream.WriteAsync( msg );
          ++_i;
        }
        catch ( Exception e )
        {
          Console.WriteLine( e );
        }
        finally
        {
          _isSending = false;
        }
      }
    }
  }
  </code></pre>
  
  <p>Note: I'm using grpc.core.1.18.0 nuget package for my testing.</p>