prodriguezdefino commented on PR #40606: URL: https://github.com/apache/beam/pull/40606#issuecomment-6225241246
Thanks for the review. I pushed 9d92655, which addresses the buffered-mode point and fixes a separate deadlock I found while checking the other two. **Concurrent accept and ordering during registerConsumer (🔴 / 🟠).** I don't think these apply to the code in this PR. There is no `tryDrainPendingForOtherBundles`, consumers map or `ConcurrentLinkedQueue` in the change, and `onStalledPoll()` only asks gRPC for one more inbound message (`request(1)`), it never calls accept. Every handoff to a bundle goes through its `ReceiverHolder`, whose methods (`accept`, `complete`, `drainPending`, `poison`, `close`) are all synchronized. `registerConsumer` sets the receiver and drains the buffer inside `complete()`, under that same lock. An `onNext` that arrives at the same time waits for it, and then `accept()` drains whatever is still pending before it adds the new element. So a newer element can't overtake buffered ones, including one with `isLast=true`. The bundle thread does hand off buffered elements from its completion callback, so I added a test for that rather than relying on reading the code: - `BeamFnDataGrpcMultiplexerTest.testConcurrentHandoffKeepsOrder`: the gRPC thread sends 3,000 elements to a slow consumer, so the bundle queue fills and the bundle thread drains the pending buffer while new elements arrive. The consumer registers in the middle of the stream. Every element arrives once and in order. **Lock-order deadlock in CancellableQueue.cancel.** I found this while checking the paths above. `cancel(e, onDrop)` ran the drop callbacks while holding the queue lock, and those callbacks take the holder lock. Meanwhile the gRPC thread holds the holder lock and calls `tryPut`, which needs the queue lock. If a bundle failed while its data was still arriving, the two threads could deadlock. `cancel` now collects the dropped elements under the lock and runs `onDrop` after releasing it. - CancellableQueueTest.testCancelWithOnDropDoesNotDeadlockWithLockedProducer: a producer holds its own lock while it calls tryPut, and the cancel's onDrop takes that lock. It deadlocks on the old code and passes now. - CancellableQueueTest.testOnDropRunsWithoutQueueLock: another thread can take the queue lock while onDrop runs. **Buffered mode (🟡).** Good catch. The main deadlock was already fixed there, because inbound `onNext` no longer blocks, but the stall hook wasn't wired in. `OutboundObserverFactory.Buffered` now passes the inbound `StalledPollHandler` to `BufferingStreamObserver`. While the drainer waits on flow control, it re-checks isReady() every 50 ms and calls `onStalledPoll()` every 500 ms, as `DirectStreamObserver` does. Without a handler, the wait is unchanged. - `BufferingStreamObserverTest.testStalledPollHandlerInvokedWhileNotReady`: the handler runs while the stream isn't ready, and the drainer sends once it is, without an onReady callback. -- 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]
