pnowojski commented on a change in pull request #9426: [FLINK-12958] Integrate
AsyncWaitOperator with mailbox (preserving compatibility with legacy sources)
URL: https://github.com/apache/flink/pull/9426#discussion_r314195257
##########
File path:
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/async/AsyncWaitOperator.java
##########
@@ -397,26 +404,41 @@ private void stopResources(boolean waitForShutdown)
throws InterruptedException
* @throws InterruptedException if the current thread has been
interrupted
*/
private <T> void addAsyncBufferEntry(StreamElementQueueEntry<T>
streamElementQueueEntry) throws InterruptedException {
- assert(Thread.holdsLock(checkpointingLock));
-
pendingStreamElementQueueEntry = streamElementQueueEntry;
- while (!queue.tryPut(streamElementQueueEntry)) {
- // we wait for the emitter to notify us if the queue
has space left again
- checkpointingLock.wait();
+ if (!queue.tryPut(streamElementQueueEntry)) {
+ do {
+ // We will receive a notify on the lock at the
end of each processing-letter execution (see {Emitter})
+ // so we can try to put again.
+ if (!tryYieldMainThreadExecutor()) {
Review comment:
This logic (difference of mailbox vs non mailbox thread) should be
encapsulated in the mailbox. It's already duplicated here, it will be
duplicated in other places (like `ContinousFileReaderOperator`) and mailbox
should leak the details of the threading model to the operators.
something like `mailbox.yield()` called from here, should do this black
magic in a secret.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
