sjvanrossum opened a new pull request, #40788: URL: https://github.com/apache/beam/pull/40788
Fixes a bidirectional flow-control deadlock on the Java SDK Fn API data plane stream without introducing secondary per-instruction queues or per-element allocations. On a client-side gRPC stream (`OutboundObserverFactory.DirectClient` / `BufferingStreamObserver`), gRPC uses a single per-call `SerializingExecutor` to execute both `InboundObserver.onNext` (inbound elements) and `ClientCallStreamObserver.onReady` (which calls `AdvancingPhaser.arrive()` to unblock outbound `DirectStreamObserver.onNext` calls). When a bundle processing thread in `BeamFnDataInboundObserver.awaitCompletion()` emits elements back over the same data plane stream and blocks in `DirectStreamObserver.onNext` waiting for the outbound stream to become ready (`AdvancingPhaser`), it stops draining its `CancellableQueue` (capacity `100`). Once `100` inbound `Elements` messages accumulate in that `CancellableQueue`, the next `InboundObserver.onNext` blocks in `CancellableQueue.put()` directly on the gRPC `SerializingExecutor` thread. Because the `SerializingExecutor` thread is blocked in `put()`, the pending `onReady` callback queued behind it on the same `SerializingExecutor` can never run to unblock the consumer thread, resulting in a permanent deadlock. ### Benchmarks `ProcessBundleBenchmark` executed on Apple M5 Pro. ```bash ./gradlew :sdks:java:harness:jmh:jmh --args="org.apache.beam.fn.harness.jmh.ProcessBundleBenchmark -p elementsEmbedding=false -f 1 -wi 3 -i 5 -w 2s -r 2s -foe=true" ``` | Benchmark | Base | This PR | vs base | | :--- | ---: | ---: | ---: | | `testLargeBundle` | `14,991.4 ± 496.3 ops/s` | `14,931.6 ± 119.8 ops/s` | `-0.4%` | | `testTinyBundle` | `35,573.1 ± 1,610.7 ops/s` | `37,160.9 ± 223.5 ops/s` | `+4.5%` | | `testStateWithCaching` | `6,748.1 ± 167.0 ops/s` | `6,768.4 ± 38.9 ops/s` | `+0.3%` | | `testStateWithoutCaching` | `3,936.6 ± 216.9 ops/s` | `4,113.7 ± 96.3 ops/s` | `+4.5%` | -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
