Manual flow-control and back-pressure demo #3119

rmichela · 2017-06-21T05:05:45Z

In the process of adapting RxJava to gRPC, I found I had to make use of manual flow-control and back-pressure, however, there were no clear examples for how to use the manual flow-control APIs.

This sample implements a bidirectional streaming service with client and server workers processing messages at different rates using manual flow-control and back-pressure-aware idioms on both ends of the wire.

I've done my best to document how things work. Please correct me if any of the idioms are wrong.

grpc-kokoro · 2017-06-21T05:05:46Z

Thanks for your pull request. The automated tests will run as soon as one of the admins verifies this change is ok for us to run on our infrastructure.

ejona86

Didn't finish the review, but sending what I have. We've really needed an example like this.

ejona86 · 2017-06-21T17:00:31Z

examples/src/main/java/io/grpc/examples/manualflowcontrol/ManualFlowControlClient.java

+    }
+
+    private static List<String> names() {
+        List<String> names = new ArrayList<String>();


Arrays.asList() supports varargs, so it is great for a case like this.

I forgot that was available in JDK6.

ejona86 · 2017-06-21T17:02:43Z

examples/src/main/proto/helloworld.proto

@@ -25,6 +25,7 @@ package helloworld;
 service Greeter {
  // Sends a greeting
  rpc SayHello (HelloRequest) returns (HelloReply) {}
+  rpc SayHelloStreaming (stream HelloRequest) returns (stream HelloReply) {}


This file is shared by other languages, so we can't "just" change our version. But also, this is used by the helloworld, which is supposed to be super-simple. Let's instead create a new .proto for this streaming example. If you wanted to name it "helloworld_streaming.proto" or similar, that'd be no big deal to me.

ejona86 · 2017-06-21T17:12:58Z

examples/src/main/java/io/grpc/examples/manualflowcontrol/ManualFlowControlClient.java

+                // Set up a back-pressure-aware producer for the request stream. The onReadyHandler will be invoked
+                // when the consuming side has enough buffer space to receive more messages.
+                //
+                // Note: the onReadyHandler is invoked by gRPC's internal thread pool. You can't block in this in


We use a cached thread pool for callbacks to the application by default. Blocking is "okay", and if doing work to generate a response to a request will provide automatic back-pressure (which is a good thing). The only thing to point out though is that we can't call any other callback while a callback is blocked, since all callbacks are serialized to reduce threading headache for users. That mainly means onError/onCompleted couldn't be called. So in this example, I think it'd be good and fine to do all work within callbacks.

Note also, that the current use of pool seems like it may have multithreading issues because the onReadyHandler can be called why the Runnable is running, thus running multiple simultaneously. The only way that would happen today is if there was an "invisible" readiness flip-flop between onNext() and isReady() (isReady() never returned false, but if it would have been called 1 µs earlier it would have returned false, and you just never happened to see it).

I have considered optimizing the onReady() callback to only call if isReady() returned false (with an exception for the initial readiness at the start of a call). Although even with such an optimization it's unclear whether the application can assume such behavior, since maybe an interceptor called isReady().

I introduced the external thread pool on the server side specifically because all the callbacks are serialized. The onReadyHandler was waiting for work and was blocking onNext from accepting that work.

The client side needed an external thread because the onReadyHandler would have to run to completion before any server responses could be processed. This prevented true bi-directional streaming from happening. All the requests would be sent, then all the responses would be consumed.

Basically, if the onReadyHandler is not running on another thread, any delays in sending messages on the outbound stream will prevent the inbound stream from calling onNext(). onReadyHandlers that complete "instantly", such as iterating over a fixed list, aren't really noticeable, but an onReadyHandler with delays will cause blocking problems and prevent bi-directional streaming.

ejona86 · 2017-06-21T17:31:16Z

okay to test

carl-mastrangelo · 2017-06-21T18:31:29Z

examples/src/main/java/io/grpc/examples/manualflowcontrol/ManualFlowControlClient.java

+
+        // Create a channel and a stub
+        ManagedChannel channel = ManagedChannelBuilder
+                .forAddress("localhost", 50051)


We typically use indent +2 for new scope, and +4 for line wrap. It keeps lines shorter, and wrap less often.

carl-mastrangelo · 2017-06-21T18:32:51Z

examples/src/main/java/io/grpc/examples/manualflowcontrol/ManualFlowControlClient.java

+public class ManualFlowControlClient {
+    public static void main(String[] args) throws InterruptedException {
+        final ExecutorService pool = Executors.newCachedThreadPool();
+        final Object done = new Object();


optional: this would be cleaner as a CountDownLatch

carl-mastrangelo · 2017-06-21T18:33:16Z

examples/src/main/java/io/grpc/examples/manualflowcontrol/ManualFlowControlServer.java

@@ -0,0 +1,147 @@
+/*
+ * Copyright 2016, gRPC Authors All rights reserved.


carl-mastrangelo · 2017-06-21T18:34:11Z

examples/src/main/java/io/grpc/examples/manualflowcontrol/ManualFlowControlServer.java

+            public void run() {
+                System.out.println("Shutting down");
+                server.shutdown();
+                pool.shutdown();


technically, pool needs to await server termination since it could add more events to the pool after server shutdown

jroper · 2017-06-27T00:43:55Z

I don't think this is a good example, handling back pressure manually like this should be considered an anti pattern. The design of the incoming flow control for gRPC intentionally mirrors reactive streams (which is being included in JDK9 as the new Flow API), it is quite trivial to adapt the API to the reactive streams Publisher/Subscriber APIs. The thing about reactive streams is that it is not meant to be implemented or used by end users in this way, it is considered an anti pattern for end users to be invoking request() themselves, for example. Even trivial examples are hard to get right, but add situations where you want to process multiple messages concurrently, for example, and the threading/concurrency concerns become very difficult to manage. Instead, reactive streams is an integration API, intended to allow different streaming providers to integrate with each other. If you want more control over flow than what gRPC gives you, then best practice would be to use a dedicated reactive streams implementation, such as Akka streams or rxJava, and use the much more powerful, but more importantly safer and easier to get right APIs that they provide to process the stream.

So I think to achieve the goals that this example is trying to achieve, we should show how to bridge gRPCs API to reactive streams (it's just a few lines of code, adapting interfaces with identical method signatures to each other), and then integrate that with a third party stream processing library. I don't think handling it manually should be encouraged.

rmichela · 2017-06-27T22:06:52Z

@jroper I completely agree with you. These APIs should not be used directly if at all possible. Bridging gRPC to other reactive technologies is the end goal. In fact, I did exactly that with RxJava2. See: salesforce/grpc-java-contrib#17

The purpose of this demo is to show how manual flow control fits together in gRPC, so that others looking to adapt different reactive technologies can more easily do so.

ejona86 · 2017-07-06T17:53:45Z

examples/src/main/java/io/grpc/examples/manualflowcontrol/ManualFlowControlServer.java

+              System.out.println("--> " + name);
+              work.add(name);
+              // Signal the sender to send another request.
+              serverCallStreamObserver.request(1);


As-is, the disableAutoInboundFlowControl() is not adding value and is just busy work. To do this "for real" you wouldn't request() until you had sent everything you needed.

I included disableAutoInboundFlowControl() on the server for illustration. It's paired with lines 54 and 109, where request(1) is called. These lines mimic the automatic flow control performed by ServerCalls.

Technically, the scenario presented here could be implemented automatic flow control, but this demo's purpose is to show how to do everything manually.

With automatic flow control, we wouldn't have this extra work
queue. The response would be generated and sent out from
onNext. So if we can't keep up with the work, we would also be
too slow to ask for request(1). As @ejona86 mentioned in the
other comment, the problem is that the message is put into an
unbounded queue and then we immediately ask for more. One way we
can fix address this is by calling request(1) only after we
send out a response (line 86).

ejona86 · 2017-07-06T17:55:10Z

examples/src/main/java/io/grpc/examples/manualflowcontrol/ManualFlowControlServer.java

+              // Accept and enqueue the request.
+              String name = request.getName();
+              System.out.println("--> " + name);
+              work.add(name);


This is assuming that the Runnable is still running in pool. That's not guaranteed to be the case. You need to try to send here. I really think you should drop the pool and run everything in the callback thread.

You can't run everything on the callback thread. The onReadyHandler will livelock the callback thread.

serverCallStreamObserver.isReady() goes true and onReadyHandler is invoked.

onReadyHandler enters a processing loop, processing messages while serverCallStreamObserver.isReady() == true.

Because onReadyHandler is looping, the callback thread is blocked until isReady() == false, but isReady() can never be set to false because doing so requires the callback thread, which is stuck in a loop.

Additionally, since the callback thread is stuck in a loop, no inbound messages are ingested. Ingestion is also handled by the callback thread.

The root problem is that the onReadyHandler is invoked by the same SerializingExecutor that runs the rest of gRPC's internal processing loop. Ideally, gRPC would maintain a separate executor for onReadyHandlers.

I think this is the source of confusion:

Because onReadyHandler is looping, the callback thread is blocked until isReady() == false, but isReady() can never be set to false because doing so requires the callback thread, which is stuck in a loop.

isReady() changes immediately. As soon as onNext() returns when sending a message, isReady() is up-to-date. If it didn't it would be pretty weak, like you are describing.

Additionally, since the callback thread is stuck in a loop, no inbound messages are ingested. Ingestion is also handled by the callback thread.

That is on purpose. Since the server is doing some "processing" it should push-back on the client. It should only request more messages when it is ready to process them. Having an infinite queue of requests is counter-productive.

The onReadyHandler can be triggered while the previous handler is
still running. This is because the handler is called every time the
stream transitions from notReady->ready. The code right now has the
potential to kick off multiple sender threads by mistake. One way to
fix this could be to have only a single sender thread that blocks
rather than returns, and use the onReadyHandler to wake the thread
up.

It's possible to make this all work, but the scope of this example is
getting a bit larger than the original goal showing how to use
request. Perhaps we can simplify the demo to avoid simultaneous
reading and writing:

The server receives the message and responds in the same
thread. The server can still sleep to simulate doing some work.

The client sends requests using logic similar to
StreamObservers#copyWithFlowControl so that it will get pushed
back.

zpencer · 2017-08-11T15:25:38Z

examples/src/main/java/io/grpc/examples/manualflowcontrol/ManualFlowControlClient.java

+
+                        // Send more messages if there are more messages to send.
+                        String name = iterator.next();
+                        System.out.println("--> " + name);


nit: we have been using java.util.logging.Logger to print UI messages like these. I know it's a bit arbitrary but let's use that here as well to be consistent :)

rmichela · 2017-08-16T23:58:20Z

Thank you all for the feedback. I'll look into simplifying the example.

rmichela · 2017-08-30T18:32:26Z

@ejona86 @zpencer I have simplified the manual flow control server example to show single message processing using request(1).

ejona86 · 2017-09-11T22:22:41Z

examples/src/main/java/io/grpc/examples/manualflowcontrol/ManualFlowControlServer.java

+              String message = "Hello " + name;
+              logger.info("<-- " + message);
+              HelloReply reply = HelloReply.newBuilder().setMessage(message).build();
+              responseObserver.onNext(reply);


For proper flow control on server-side, maybe something like this:

responseObserver.onNext(reply); if (responseObserver.isReady()) { responseObserver.request(1); } else { notReady = true; } // onReadyHandler if (notReady && responseObserver.isReady()) { notReady = false; responseObserver.request(1); }

zpencer · 2017-09-13T15:51:36Z

examples/src/main/java/io/grpc/examples/manualflowcontrol/ManualFlowControlServer.java

+          @Override
+          public void run() {
+            logger.info("READY");
+            // Signal the request sender to send one message.


Let's expand the comment here to explain that we are requesting a message here because we expect to send a response upon receiving the request. Otherwise it might be confusing to request a message when the outbound direction is ready.

zpencer · 2017-09-13T16:11:08Z

examples/src/main/java/io/grpc/examples/manualflowcontrol/ManualFlowControlClient.java

+            // MANY messages may be buffered, however, they haven't yet been sent to the server. The server must call
+            // request() to pull a buffered message from the client.
+            //
+            // Note: the onReadyHandler is invoked by gRPC's internal thread pool. You can't block here or deadlocks


To be more precise, the onReadyHandler executes on a thread pool donated by the user application. We have a default thread pool if none is passed into ServerBuilder.executor. The thread pool is used to create sets of serialized executor queues, one for each RPC. This onReadyHandler is serialized on one of these such executors, and the same one would be used to run the onNext, onCompleted, etc. So just as it is fine to block in onNext, it is technically fine to block in onReady. The risk is that no further progress can be made for this RPC stream, and one of the threads on the thread pool will be eaten up. Let's update this comment and the one in the server to clarify that blocking would not cause all of gRPC to hang.

zpencer · 2017-09-13T16:13:59Z

examples/src/main/java/io/grpc/examples/manualflowcontrol/ManualFlowControlServer.java

+          }
+        });
+
+        // Give gRPC a StreamObserver it can write incoming requests into.


s/it can write incoming requests to/that can observe (i.e. process) incoming requests/

zpencer · 2017-09-13T16:14:42Z

examples/pom.xml

@@ -10,7 +10,8 @@
  <name>examples</name>
  <url>http://maven.apache.org</url>
  <properties>
-    <grpc.version>1.7.0-SNAPSHOT</grpc.version><!-- CURRENT_GRPC_VERSION -->
+    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+    <grpc.version>1.6.1</grpc.version><!-- CURRENT_GRPC_VERSION -->


It looks like this part is stale now since we've bumped the gRPC version

oops. didn't mean to check that in.

rmichela · 2017-09-15T18:12:32Z

@zpencer @ejona86 Documentation updated as requested

zpencer · 2017-09-21T17:23:15Z

Had a chat with @ejona86 and it looks like the notReady = true variable is important to guard against spurious onReadys. This can happen if the stream becomes ready inside of the server's onNext:

responseObserver.onNext(reply);
// time=0 stream becomes ready, and an onReady becomes scheduled
if (responseObserver.isReady()) {
 // time=1 we enter this block and request a message
  responseObserver.request(1);
}

//time=2 onReady runs, and we request yet another message
responseObserver.request(1);

IMO the notReady boolean would be more clear if we flip the meaning to alreadyRequested, so it feels natural to only request if !alreadyRequested.

rmichela · 2017-09-25T21:39:08Z

@zpencer @ejona86 I've addressed the subtle race between onNext() and isReady().

ejona86

One small change. Looks good.

ejona86 · 2017-09-25T22:56:48Z

examples/src/main/java/io/grpc/examples/manualflowcontrol/ManualFlowControlServer.java

+        // toggles isReady() from false to true while onNext() is executing, but before onNext() checks isReady(),
+        // request(1) would be called twice - once by onNext() and once by the onReady() scheduled during onNext()'s
+        // execution.
+        final AtomicBoolean wasReady = new AtomicBoolean(false);


Make this a simple boolean field in this anonymous StreamingGreeterImplBase class. There's no need for synchronization and since we're already creating a class we can use a field instead of a final.

rmichela · 2017-09-26T14:52:57Z

@ejona86 @zpencer I've been thinking about the wasReady flag and I don't think it's right.

Putting it at the class level makes wasReady shared state between all concurrent clients of the service. Doesn't this introduce a race condition between two independent clients?

zpencer · 2017-09-26T15:23:24Z

Actually you're right, sayHelloStreaming should be independent from one call to the next.

rmichela · 2017-09-26T15:52:14Z

@zpencer I've reverted the change that made wasReady a global variable. It's back to being an AtomicBoolean. I had to use AtomicBoolean over regular boolean because the onReadyHandler is an anonymous inner Runnable and can only reference final outer variables. If wasReady were a simple boolean, I would not be able to modify it from within the onReadyHandler.

ejona86 · 2017-09-26T16:54:25Z

okay to test

ejona86

Yeah, you're right; you'd need a single class for all callbacks (which is possible, but meh). AtomicBoolean is the natural thing to use.

ejona86 · 2017-09-26T17:02:11Z

Thanks @rmichela! Thanks for bearing with us!

rmichela added 5 commits June 20, 2017 18:19

Manual flow control demo

a24f1af

Manual flow control demo

b1572e8

Manual flow control demo

28677a1

Manual flow control demo

180dfc5

Manual flow control demo

f2f6ee1

rmichela changed the title ~~Feature/manual flow control demo~~ Manual flow-control and back-pressure demo Jun 21, 2017

ejona86 reviewed Jun 21, 2017

View reviewed changes

ejona86 added the kokoro:run Add this label to a PR to tell Kokoro the code is safe and tests can be run label Jun 21, 2017

kokoro-team removed the kokoro:run Add this label to a PR to tell Kokoro the code is safe and tests can be run label Jun 21, 2017

carl-mastrangelo reviewed Jun 21, 2017

View reviewed changes

rmichela added 3 commits June 21, 2017 11:43

Split unary and streaming protos

b30a5fc

Minor improvements

83d34ec

Code formatting

475e477

ejona86 reviewed Jul 6, 2017

View reviewed changes

zpencer reviewed Aug 11, 2017

View reviewed changes

rmichela added 3 commits August 30, 2017 09:30

Use logger instead of println

109f457

Simplify example

620d984

Merge master

77e89fa

Merge branch 'master' into feature/manual-flow-control-demo

882048b

ejona86 reviewed Sep 11, 2017

View reviewed changes

Use onReadyHandler to initiate sending

ed8b648

zpencer reviewed Sep 13, 2017

View reviewed changes

rmichela added 2 commits September 13, 2017 22:43

Improve documentation

70232ef

Revert accidental pom changes

87c660f

rmichela added 3 commits September 25, 2017 11:04

Merge branch 'master' into feature/manual-flow-control-demo

5b0555d

Stop a subtle race between onNext() and onReady()

6682e7e

Merge changes

9f8b8a3

ejona86 approved these changes Sep 25, 2017

View reviewed changes

zpencer approved these changes Sep 26, 2017

View reviewed changes

rmichela force-pushed the feature/manual-flow-control-demo branch from 79f8368 to 9f8b8a3 Compare September 26, 2017 15:49

ejona86 approved these changes Sep 26, 2017

View reviewed changes

ejona86 merged commit 589da07 into grpc:master Sep 26, 2017

cbornet mentioned this pull request Oct 19, 2017

Simplify implementation of back-pressure in StreamObserver-based stub #1549

Open

cbornet mentioned this pull request Dec 11, 2017

Implement end-to-end backpressure btlines/grpcakkastream#12

Open

rmichela deleted the feature/manual-flow-control-demo branch December 18, 2017 04:42

lock bot locked as resolved and limited conversation to collaborators Jan 19, 2019

		@@ -0,0 +1,147 @@
		/*
		* Copyright 2016, gRPC Authors All rights reserved.

Manual flow-control and back-pressure demo #3119

Manual flow-control and back-pressure demo #3119

Conversation

rmichela commented Jun 21, 2017

grpc-kokoro commented Jun 21, 2017

ejona86 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

rmichela Jun 21, 2017 • edited Loading

Choose a reason for hiding this comment

ejona86 commented Jun 21, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jroper commented Jun 27, 2017

rmichela commented Jun 27, 2017 • edited Loading

Choose a reason for hiding this comment

rmichela Jul 10, 2017 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

rmichela Jul 10, 2017 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

zpencer Aug 11, 2017 • edited Loading

Choose a reason for hiding this comment

rmichela commented Aug 16, 2017

rmichela commented Aug 30, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

rmichela commented Sep 15, 2017

zpencer commented Sep 21, 2017

rmichela commented Sep 25, 2017

ejona86 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

rmichela commented Sep 26, 2017

zpencer commented Sep 26, 2017

rmichela commented Sep 26, 2017 • edited Loading

ejona86 commented Sep 26, 2017

ejona86 left a comment

Choose a reason for hiding this comment

ejona86 commented Sep 26, 2017

rmichela Jun 21, 2017 •

edited

Loading

rmichela commented Jun 27, 2017 •

edited

Loading

rmichela Jul 10, 2017 •

edited

Loading

rmichela Jul 10, 2017 •

edited

Loading

zpencer Aug 11, 2017 •

edited

Loading

rmichela commented Sep 26, 2017 •

edited

Loading