Hi David, Thanks so much for the fast reply, that’s really helpful context.
I like the idea you mentioned where the application provides an ArrowReader. Would that story go something like the following? 1. Today in FlightService doGetCustom, a GetListener is passed to producer.getStream - this leaves the application developer responsible for handling gRPC details discussed above (they can choose to respect OutboundStreamListener.isReady or not, etc.). 2. In the ArrowReader case you mentioned, that same call to producer.getStream would instead (or as another option) return an ArrowReader… then the FlightService code could say something like “while loadNextBatch is true, (somehow) unload that RecordBatch and pipe it through to the OutboundStreamListener’s putNext” — One nice thing about interacting with the OutboundStreamListener directly is that application developers can putNext with metadata - I wonder where metadata fits in the ArrowReader case? — That makes sense that it would be hard to get that async writing out to gRPC to work flawlessly - maybe one option in this scenario where Flight is deciding how to write out on the stream would be to default to a “noop BackpressureStrategy” or a simple blocking one and let application developers optionally override? Thanks, Nate On 2023/03/12 17:12:41 David Li wrote: > Hi Nate, > > That sounds about right to me (it's been a while since I've dug into > gRPC-Java behavior). A better server API is something I've long wanted to > consider and haven't had the time to; the current APIs try to let you write > blocking/procedural code as much as possible which then does not mesh well > with the actual gRPC APIs they attempt to wrap. > > We could expose setOnReadyHandler, IMO. Though from what I recall, it's very > tricky to use correctly and it's easy to get yourself "stuck" by missing a > callback. > > My hope for a better server API would be to eventually just have the > application provide an ArrowReader (asynchronously) and then have the Flight > implementation pull from that reader in the most efficient possible manner > (though gRPC-Java makes that hard, what with the hardcoded backpressure > threshold [1] - I think that may have been another reason why I didn't expose > setOnReadyListener before since it would artificially limit your throughput) > > [1]: https://github.com/grpc/proposal/pull/135 > > -David > > On Fri, Mar 10, 2023, at 18:46, Nathaniel Jones wrote: > > Hello, > > > > I'm hoping to check my understanding around various ways to implement a > > DoGet handler with respect to flow control, and then inquire about > > potential future API changes. > > > > First, I'm aware of 3 ways to respect flow control when implementing a Java > > server's DoGet that have different characteristics: > > > > 1. Busy-Waiting / Thread.sleep()-ing: > > 1. Implement a blocking body that loops (and maybe sleeps) while > > the > > ServerStreamListener's isReady is false (and respect isCancelled, > > too) > > 2. Using BackpressureStrategy > > > > <https://sourcegraph.com/github.com/apache/arrow/-/blob/java/flight/flight-core/src/main/java/org/apache/arrow/flight/BackpressureStrategy.java?L28:8> > > (and > > specifically CallbackBackpressureStrategy > > > > <https://sourcegraph.com/github.com/apache/arrow/-/blob/java/flight/flight-core/src/main/java/org/apache/arrow/flight/BackpressureStrategy.java?L75> > > since > > one could implement option #1 above as a simple strategy): > > 1. My own experiments with CallbackBackpressureStrategy / > > understanding from initial PR discussion > > <https://github.com/apache/arrow/pull/8476#issuecomment-710777211> > > demonstrate > > that the DoGet handler must be run on a separate thread; you > > can't invoke > > the "waitForListener" on the thread that gRPC uses to invoke this > > RPC > > because if you're blocking (in this case Thread await-ing) on > > this gRPC > > thread, gRPC can't process onReady callbacks for this RPC, and > > thus > > CallbackBackpressureStrategy would never be notify-ed to wake up > > 3. Write a fully async implementation relying directly on underlying > > CallStreamObserver's setOnReadyHandler: > > 1. This is similar in spirit to #2 above, but now operates > > completely > > on threads from gRPC's thread pool (the onReady handler *is* the > > DoGet logic). The code looks very roughly like: > > 1. make VectorSchemaRoot with some schema and allocator > > 2. On our ServerStreamListener, invoke listener.start(root) > > immediately > > 3. set listener.setOnReadyHandler(<the DoGet logic to stream > > data > > to client>) > > 4. So, no blocking > > > > My understanding of the trade-offs between options #2 and #3 above: > > > > - In CallbackBackpressureStrategy with background threads, there'll be > > (# threads in gRPC pool) + (sum of background threads across currently > > running DoGet streams) threads. If the actual streaming logic is > > intensive > > / CPU bound, it might be good for that to live on a background thread, > > because if the gRPC threads were tied up in intensive callbacks (and even > > exhausted), new RPC requests / ready and cancel callbacks could be slow / > > stuck. > > - On the other hand, for quick I/O bound work in DoGet logic, option #3 > > might work well if it's not worth the extra threads / context switching, > > and gRPC threads can handle everything quickly. > > - So overall, it seems like different workloads could demand a different > > model, which should inform how DoGet logic is written. > > > > > > *Q1: *Is my understanding of the above points approximately correct? I'd > > appreciate any pointers on items I'm misunderstanding. > > > > *Q2: *Assuming option #3 has utility in some cases, I'm curious if > > there is > > a way in Flight to expose an easier API to implement DoGet fully > > asynchronously. I find it tricky to manage cleanup of the > > VectorSchemaRoot > > as well as being careful about things like respecting isCancelled > > becoming > > true. Developers also "need to know" that the async setOnReadyHandler > > exists at the gRPC layer and understand its benefits. I noticed that on > > ARROW-4484 <https://issues.apache.org/jira/browse/ARROW-4484>, which > > focuses on DoPut's busy-wait, the first comment mentions that this > > applies > > to DoGet as well - though it seems GetListener's waitUntilStreamReady > > <https://sourcegraph.com/github.com/apache/arrow/-/blob/java/flight/flight-core/src/main/java/org/apache/arrow/flight/FlightService.java?L189-192> > > doesn't > > busy-wait, I was curious if the spirit of the comment was similar to my > > question around an easier way to write the logic asynchronously? If so, > > would the idea be for Flight to wrap gRPC's callbacks and expose those > > to > > DoGet authors, while helping to abstract some cleanup items? > > > > Thank you for any feedback, > > Nate >
