Grpc-java: OutOfDirectMemoryError

[lploom]

Copied from the C based grpc repository
https://github.com/grpc/grpc/issues/7938

Basically it seems that grpc will throw OutOfDirectMemoryError if you send enough messages over a single stream. I am probably just using it wrong or it was not meant for such use case. If anyone could explain this, that'd be great.

The code to reproduce the error: https://github.com/lploom/grpc-OutOfDirectMemoryError
To execute, run GameServer and then GameClient, when the client connects the server attemps to send 3M messages, but eventually crashes.

[ejona86]

The async interface is fully async. That means that your 3M messages are getting queued immediately.

for (int i = 0; i < 3_000_000; i++) {
    responseObserver.onNext(msg);
}

To prevent running out of memory, you need to observe flow control which "pushes back" when sending too quickly. For example:

final ServerCallStreamObserver<GameState> scso =
    (ServerCallStreamObserver<GameState>) responseObserver;
Runnable drain = new Runnable() {
  int remaining = 3_000_000;
  public void run() {
    if (remaining == 0) return;
    for (; remaining > 0 && scso.isReady(); remaining--) {
        scso.onNext(msg);
    }
    if (remaining == 0) {
      scso.onCompleted();
    }
  }
};
scso.setOnReadyHandler(drain);
drain.run();

isReady() returns false when the queue is sufficiently full, and the Runnable is run when isReady() transitions back to true.

[lploom]

Oh cool. It makes a lot of sense now, I figured I was probably just using it wrong.

Both issues can be closed, i think.

[wu-sheng]

I try to use this API in my grpc client. But it still failure.

My test codes:

for (int i = 0; i < 1000000; i++) {
            StreamObserver<RequestSpan> requestSpanStreamObserver =
                    spanStorageServiceStub.storageRequestSpan(new StreamObserver<SendResult>() {
                        @Override
                        public void onNext(SendResult sendResult) {

                        }

                        @Override
                        public void onError(Throwable throwable) {
                            throwable.printStackTrace();
                        }

                        @Override
                        public void onCompleted() {
                            endTime2 = System.currentTimeMillis();
                        }
                    });
            for (int j = 0; j < 10; j++) {
                requestSpanStreamObserver.onNext(requestSpan);
            }

            ClientCallStreamObserver<RequestSpan> newRequestSpanStreamObserver =
                    (ClientCallStreamObserver<RequestSpan>) requestSpanStreamObserver;

            while (!newRequestSpanStreamObserver.isReady()) {
                Thread.sleep(1);
            }

            ackSpanStreamObserver.onCompleted();
            requestSpanStreamObserver.onCompleted();

Is there any wrong of my usage?

[wu-sheng]

@ejona86

[ejona86]

You're doing up to a million concurrent RPCs, depending on how long they take to complete. What behavior did you hope for?

[wu-sheng]

@ejona86 , I hope when I called

while (!newRequestSpanStreamObserver.isReady()) {
     Thread.sleep(1);
}

this process thread will hold, until gRPC send all datas, I already gave. And release the directMemory, no need to allocate now memory. Also will not to trigger _OutOfDirectMemoryError_

In this case, waiting a size 10 list of requestSpan to be send.

            for (int j = 0; j < 10; j++) {
                requestSpanStreamObserver.onNext(requestSpan);
            }

But in my local test, it's failed. Is this a wrong usage?

[ejona86]

Ah, I see. isReady() doesn't mean the buffer is empty. It just means that the call has started (headers are written) and that not "too much" is buffered. The call being started is near instantaneous in common cases; it is really only delayed when the server is using MAX_CONCURRENT_STREAMS (which gRPC servers are rarely doing today). If the server specified a MAX_CONCURRENT_STREAMS it would probably fix the memory failure (but that doesn't mean that is an appropriate fix; the client shouldn't require the server to be configured like that).

You may be hitting a frame scheduling issue where we don't schedule headers (they are written immediately) but we delay and schedule data to allow reacting with low latency.

All said, using isReady() like that is abusive. What is your use case? You have tons of RPCs to do and you want to send them as quickly as possible without running out of memory?

[ejona86]

And to be a bit clearer: a usage of readiness was intended for pushing back on new RPCs. After creating a new RPC you could wait until ready before sending. That said, your "abusive" usage would only have a few more requests than the intended way. So the intended way would run out of memory as well.

I do think grpc needs to delay ready longer for new RPCs when in this situation, but it isn't clear to me at the moment how I can define/detect "this situation." I'll probably need to have discussions with the other implementations.

[wu-sheng]

@ejona86 , Yes, I have tons of data, it generated by http service, local method calls or some other point.

So when I have a high tps server, tons of data be generated. I have to send them as quickly as possible.

[wu-sheng]

@ejona86 I am looking forward conclusion of your discussions.

But today , How should I do to avoid this out of memory?

MAX_CONCURRENT_STREAMS seems a final field in Http2CodecUtil?

[wu-sheng]

As a RPC call, client can't depend on Server, and at the same time, client must have access to get feedback of network state.

Only I need, I can control the max size of buffer. Only send data as quickly as possible, as the certain size buffer.

@ejona86 , waiting for your solution……

[wu-sheng]

@ejona86 , this issue maybe not the same problem as my description.

I open a new issue #2424 , we can have further discussion. Maybe we can keep #2424 open? until you find a way to define/detect "this situation". And gRPC provide official solution and have a new release.

[Ramasubramanian]

@ejona86 I am trying to call setOnReadyHandler on ClientCallStreamObserver and i get the below error.
Exception in thread "main" java.lang.IllegalStateException: Cannot alter onReadyHandler after call started
at io.grpc.stub.ClientCalls$CallToStreamObserverAdapter.setOnReadyHandler(ClientCalls.java:317)
at io.grpc.examples.routeguide.RouteGuideClient.recordRoute(RouteGuideClient.java:181)

I am calling this on the first line after getting the
StreamObserver<Point> requestObserver = asyncStub.recordRoute(responseObserver);

Any help is appreciated!

[ejona86]

@Ramasubramanian, make your responseObserver implement io.grpc.stub.ClientResponseObserver. beforeStart allows you to set the onready handler before the RPC starts. Setting the onready handler afterward is racy, which is why we detect the case and throw the exception.

[Ramasubramanian]

@ejona86 Thanks for the quick response. As per your suggestion I tried the below:

  private Runnable onReadyHandler;

  /**
   * Async client-streaming example. Sends {@code numPoints} randomly chosen points from {@code
   * features} with a variable delay in between. Prints the statistics when they are sent from the
   * server.
   */
  public void recordRoute(List<Feature> features, int numPoints) throws InterruptedException {
    info("*** RecordRoute");
    final CountDownLatch finishLatch = new CountDownLatch(1);

  StreamObserver<RouteSummary> responseObserver = new io.grpc.stub.ClientResponseObserver<Point, RouteSummary>() {
      @Override
      public void beforeStart(ClientCallStreamObserver<Point> requestStream) {
        onReadyHandler = new Runnable() {
          int remaining = 10_000_000;
          public void run() {
            if (remaining == 0) return;
            for (; remaining > 0 && requestStream.isReady(); remaining--) {
              int index = random.nextInt(features.size());
              Point point = features.get(index).getLocation();
              requestStream.onNext(point);
            }
            if (remaining == 0) {
              requestStream.onCompleted();
            }
          }
        };
        requestStream.setOnReadyHandler(onReadyHandler);
      }

      @Override
      public void onNext(RouteSummary summary) {
        info("Finished trip with {0} points. Passed {1} features. "
            + "Travelled {2} meters. It took {3} seconds.", summary.getPointCount(),
            summary.getFeatureCount(), summary.getDistance(), summary.getElapsedTime());
        if (testHelper != null) {
          testHelper.onMessage(summary);
        }
      }

      @Override
      public void onError(Throwable t) {
        warning("RecordRoute Failed: {0}", Status.fromThrowable(t));
        if (testHelper != null) {
          testHelper.onRpcError(t);
        }
        finishLatch.countDown();
      }

      @Override
      public void onCompleted() {
        info("Finished RecordRoute");
        finishLatch.countDown();
      }
    };

    StreamObserver<Point> requestObserver = asyncStub.recordRoute(responseObserver);
    onReadyHandler.run();

However calling .run() here does not send the messages to server, tried calling .run immediately after setting the onreadyHandler within the beforeStart() method and it gives a null pointer exception as shown below:

Exception in thread "main" java.lang.NullPointerException
    at io.grpc.internal.ClientCallImpl.isReady(ClientCallImpl.java:414)
    at io.grpc.stub.ClientCalls$CallToStreamObserverAdapter.isReady(ClientCalls.java:311)
    at io.grpc.examples.routeguide.RouteGuideClient$1$1.run(RouteGuideClient.java:167)
    at io.grpc.examples.routeguide.RouteGuideClient$1.beforeStart(RouteGuideClient.java:178)
    at io.grpc.stub.ClientCalls$StreamObserverToCallListenerAdapter.<init>(ClientCalls.java:362)
    at io.grpc.stub.ClientCalls.asyncStreamingRequestCall(ClientCalls.java:263)
    at io.grpc.stub.ClientCalls.asyncClientStreamingCall(ClientCalls.java:97)
    at io.grpc.examples.routeguide.RouteGuideGrpc$RouteGuideStub.recordRoute(RouteGuideGrpc.java:244)
    at io.grpc.examples.routeguide.RouteGuideClient.recordRoute(RouteGuideClient.java:207)

Please clarify as to when the handler.run() should be called or am I missing something?

[ejona86]

However calling .run() here does not send the messages to server

I believe that, because the Call isn't ready yet. This is because you are on client-side instead of server-side. The run() isn't necessary on client-side, but it doesn't hurt anything.

Even though run() sends no messages, the code works though, correct? At some point onReady will be called and your code will send messages.

tried calling .run immediately after setting the onreadyHandler within the beforeStart() method

Yes, don't do that; the Call has not been started yet within beforeStart(), so you can't send.

[sonya6688]

A follow up on this OOM issue: I have a Java GRPC client, and a GO server. if I use Java ClientCall API to set flow control, do I need to do anything special on the server side to enable the flow control on the same stream?

[carl-mastrangelo]

Yes you do. For bidirectional streams, each stream direction is independent of the other direction. The ClientCall "honors" the flow control window information by calling isReady and onReady. The client gives flow control window to the Go server by calling request. I believe Go uses SendMsg and RcvMsg. This happen automatically normally, but I you can control them directly.

The two work in pairs.

[ejona86]

@sonya6688, I actually think you don't need to do anything special on the Go server-side. The Go server handles the flow control automatically; it doesn't need an outbound flow control API because it can block.