Backpressure support

[fillson-shady]

Hello!
First of all thank you for the great library! That is what I was looking for.

I have question about backpressure support in streaming scenarios.
I have created simple test with client streaming to server with a large chunks of bytes. It works great with a performance near native java-gRPC libraries shows. However, when I have added ZIO.sleep to a service logic, test became crushing with an OutOfMemory error (on both client and server side).

There is a great library salesforce/reactive-grpc which provides backpressure support for java world. Is there any plans to implement such functionality in zio-grpc?

Btw, there is a simple workaround can be used for now:

def clientLogic = for {
  data      <-  //some big, maybe unbounded data stream
  queue     <- Queue.bounded[Request](10)     //10 is buffer size
  _         <- queue.offerAll(data.take(10))  //initial request for a data
  _         <- GeneratedService
    .send(Stream.fromQueue(queue))
    .tap(demand => queue.offerAll(data.take(demand.count)))
    .runDrain
} yield ()

Here client sends some amount of data to the server and after some processing server responses to client with a Demand(count: Int) message, which signalises that server is ready accept more data.

[thesamet]

Hey @fillson-shady . I haven't been thinking about backpressure yet, but happy to get the discussion started and having this implemented soon. Will you be able to contribute a PR? If so, can you briefly describe the design/implementation?

[fillson-shady]

Hi @thesamet , thank you for quick reply!
Unfortunately my workaround can't be used in a general way - it forces you to use special messages for server responses. So I have to investigate the salesforce/reactive-grpc implementation and than adopt it to ZIO ecosystem. It may take a lot of time, but anyway I will write here as soon as I get any results.

[fillson-shady]

@thesamet hello again!
Finally I have an opportunity to look on this.
Let's consider one-way streaming case with a slow consumer. With the current approach we have 2 problem places:

• out of memory in consumer, because it consumes all incoming messages in the unbounded queue
• out of memory in producer - if consumer doesn't accept more messages, producer will overflow it's internal gRPC send buffer, because producer doesn't aware of consumer 'ready to accept' status

First one is easy to solve - we should request new message only after processing already received one. We can reach it by moving call of call.request(1) from:

override def onMessage(message: Res): Unit =
    runtime.unsafeRun(queue.offer(Right(message)) *> call.request(1))

to the data stream processing stage:

def stream: ZStream[R, Status, Res] =
  ZStream.fromQueue(queue) /*some other code*/ .tap(_ => call.request(1))

If buffering is needed, a library user can always add the stream.buffer(x) stage.

Second one is more tricky - we need to know, when a consumer is ready to accept more data. The only way to do this is by using StreamObserver classes (in a client for a client streaming case and in a server for a server streaming).
There is corresponding utility methods if gRPC: ClientCalls.asyncClientStreamingCall and ServerCalls.asyncServerStreamingCall.
With this approach we must convert outgoing data stream into a queue, so we can take a data from the queue when consumer is ready:

streamObserver.setOnReadyHandler { () =>
  runtime.unsafeRunSync {
    recursivelyTake(check = Task(streamObserver.isReady), action = requestQueue.take)
  }
}

Unfortunately where is no direct mapping between gRPC's Call api and StreamObserver api, so we also should refactor some cases like handling of metadata, errors, events and so on.

I'll be glad to get a feedback about my thoughts, and then start to work on this task.

[thesamet]

For the second problem, I see that ServerCall has an isReady method: https://grpc.github.io/grpc-java/javadoc/io/grpc/ServerCall.html#isReady-- . Can we use that to simplify the solution?

[ahjohannessen]

@thesamet see https://github.com/fiadliel/fs2-grpc/pull/39 for inspiration. I guess that PR needs to be worked through and merged as well.

[thesamet]

Yes, I was thinking of a solution that worked like this (sleep while not ready, otherwise send)

[ahjohannessen]

Java example from grpc:

public void onReady() {
       while (call.isReady()) {
         FooRequest nextRequest = makeNextRequest();
         if (nextRequest == null) {  // No more requests to send
           call.halfClose();
           return;
         }
         call.sendMessage(nextRequest);
       }
     }
   }

However, it is noted in grpc-java api docs that methods that only send one message, i.e unary and server streaming should not use isReady

[fillson-shady]

For the second problem, I see that ServerCall has an isReady method: https://grpc.github.io/grpc-java/javadoc/io/grpc/ServerCall.html#isReady-- . Can we use that to simplify the solution?

Oh, I've just missed this APIs. Sure, we can use isReady and onReady in both client and server calls.

@ahjohannessen thanks for example!

[fillson-shady]

Hi @thesamet! I made a PR: #152
I also want to add some tests, but it's not so obvious:

• we can't count outgoing messages in sender because of internal gPRC/TCP buffers
• we can't count incoming messages in slow consumer because they can be placed in internal Queue
• we can't just send an infinite stream to a slow consumer, because in case of failure we can get OOM and non terminating jvm

So, I think of two possible ways:

• test bidi streaming scenario - producer tries to send many messages, consumer slowly consumes some of them (in a fork) and sends ack back to producer. Then producer checks how many messages does consumer actually get actually this will not work because of consumer internal buffers
• introduce some testable replacements for ZChannel and ZCall, however it can complicate codebase

What do you think?

UPD: there is an InProcess implementations for both gRPC server and channel. We can use it to exclude internal gPRC/TCP buffers, so we can just count sent messages on the sender side. I think it is the best alternative.

[phderome]

Yes, I was thinking of a solution that worked like this (sleep while not ready, otherwise send)

How about using ZStream.effectAsync and not sleep? The body of the registered callback to effectAsync (tentatively named def backPressureCb[A](obs: ServerCallStreamObserver, onNext: A => Unit, onCompleted: () => Unit)) would do the initial work on the casted ServerCallStreamObserver to disable automated workflow; its two callbacks would take care of onNext and onCompleted logic (see registerTweetAlert in YouTube video).
The onNext and onCompleted passed in to backPressureCb would do much of the work we in grpc example ManualFlowControlServer.java.
The quote below is a ZStream test proving that it can apply back pressure, so that the sender (using ZStream) waits to publish more when client is not ready (I am talking about effectAsync which calls effectAsyncMaybe tested below)

           testM("effectAsyncMaybe back pressure") {
          for {
            refCnt  <- Ref.make(0)
            refDone <- Ref.make[Boolean](false)
            stream = ZStream.effectAsyncMaybe[Any, Throwable, Int](
                       cb => {
                         global.execute { () =>
                           // 1st consumed by sink, 2-6 – in queue, 7th – back pressured
                           (1 to 7).foreach(i => cb(refCnt.set(i) *> ZIO.succeedNow(Chunk.single(1))))
                           cb(refDone.set(true) *> ZIO.fail(None))
                         }
                         None
                       },
                       5
                     )
            run    <- stream.run(ZSink.fromEffect[Any, Nothing, Int, Nothing](ZIO.never)).fork
            _      <- refCnt.get.repeatWhile(_ != 7)
            isDone <- refDone.get
            _      <- run.interrupt
          } yield assert(isDone)(isFalse)
        }

Below we have a common API for the same effectAsyncMaybe, so perhaps the solution is not to use that directly but via effectAsync. That API is explained in a talk by Itamar Ravid also linked below.

trait ZStreamPlatformSpecificConstructors {
  self: ZStream.type =>

  /**
   * Creates a stream from an asynchronous callback that can be called multiple times.
   * The optionality of the error type `E` can be used to signal the end of the stream,
   * by setting it to `None`.
   */
  def effectAsync[R, E, A](
    register: (ZIO[R, Option[E], Chunk[A]] => Unit) => Unit,
    outputBuffer: Int = 16
  ): ZStream[R, E, A] =
    effectAsyncMaybe(
      callback => {
        register(callback)
        None
      },
      outputBuffer
    )

See Itamar explain it in action at London 2019 (registerTweetAlert example): https://youtu.be/bbss7elSfxs?t=1501

See also io/grpc example to disable flow control and apply back pressure: https://github.com/grpc/grpc-java/blob/master/examples/src/main/java/io/grpc/examples/manualflowcontrol/ManualFlowControlServer.java

The user could then apply a ZStream timeout or deadline as additional context input to effectAsync register and similarly a config parameter for the size of outputBuffer (or maybe that'd come from the R stream parameter?).
I think it's a matter of combining these ideas, carefully mind you.

The in process server mentioned by fillson-shady is here: https://grpc.github.io/grpc-java/javadoc/io/grpc/inprocess/InProcessServerBuilder.html
with canonical example usage: https://github.com/grpc/grpc-java/blob/f6c2d221e2b6c975c6cf465d68fe11ab12dabe55/examples/src/test/java/io/grpc/examples/helloworld/HelloWorldServerTest.java

[regiskuckaertz]

We seem to be hit by the lack of backpressure in one of our services with Netty dying of an java.lang.OutOfMemoryError: Direct buffer memory. What can I do to help?

I looked at the akka-grpc implementation which can be summarised as:

• at start, pull an element from the source
• when receiving an element from the source, send it and, if client is ready, pull if not already pulled
• when client becomes ready, if not pulled already, try pulling an element from the source

It looks like we just need to create our own sink in here:

impl(req).provideEnvironment(ZEnvironment(requestContext)).foreach(call.sendMessage)

instead, roughly

val backpressure: ZSink[RequestContext, Status, Res, Nothing, Unit] = ???     
impl(req).run(backpressure).provideEnvironment(ZEnvironment(requestContext))

We can use the request context to store coordination structures if necessary, though perhaps there is a way to bypass the effect boundaries with the "unsafe" pattern in ZIO2.

[thesamet]

Hi @regiskuckaertz , thanks for bringing this up. I'll be open to a PR which would rebase the existing #152 PR or an entirely new one if a simpler/better approach is available. The only reason that the original PR didn't make it in the first place was the extra complexity without real-world demand to warrant that.

[regiskuckaertz]

@thesamet could you cut a zio 1.x release please? or are snapshots available? eager to give #448 a run for its money on my app. btw if needed, where would you put configuration settings, like the size of the buffer?

[thesamet]

We have a snapshot for each commit at https://oss.sonatype.org/content/repositories/snapshots/com/thesamet/scalapb/zio-grpc/ . Can you try it out before we cut a release? For configuring, I suggest for now a JVM system property (sys.Prop) named zio-grpc.backpressure-queue-size that defaults to 16 if unset.

[regiskuckaertz]

@thesamet Yes, i will let you know asap.