1u0 commented on a change in pull request #9383: [FLINK-13248] [runtime] Adding 
processing of downstream messages in AsyncWaitOperator's wait loops
URL: https://github.com/apache/flink/pull/9383#discussion_r315101347
 
 

 ##########
 File path: 
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/AbstractStreamOperator.java
 ##########
 @@ -227,11 +233,73 @@ public void setup(StreamTask<?, ?> containingTask, 
StreamConfig config, Output<S
                }
 
                this.runtimeContext = new StreamingRuntimeContext(this, 
environment, container.getAccumulatorMap());
+               this.mailboxExecutor = 
this.getContainingTask().getMailboxExecutor(config.getChainIndex());
 
                stateKeySelector1 = config.getStatePartitioner(0, 
getUserCodeClassloader());
                stateKeySelector2 = config.getStatePartitioner(1, 
getUserCodeClassloader());
        }
 
+       /**
+        * Allows downstream operators in the operator chain to process events.
+        * <p>This method should be called whenever an operator would need to 
block the task thread to wait for output
+        * buffers to flush.</p>
+        * <p>If such an operator would indeed block the task thread, deadlocks 
can arise when two such operators are
+        * chained.</p>
+        * <p>It's up to the implementor of the operator to find a good 
trade-off, between yielding and own processing.
+        * If the operator never blocks the task thread during input 
processing, this method should not be called at
+        * all.</p>
+        * <p>This method indicates whether downstream events actually have 
been processed, which can be used to issue
+        * further invocations.</p>
+        *
+        * @return true if any downstream event has been handled.
+        */
+       protected boolean yieldToDownstream() {
+               if (mailboxExecutor.isMailboxThread()) {
+                       boolean more = mailboxExecutor.tryYield();
+                       while (more && mailboxExecutor.tryYield()) {
+                       }
+                       return more;
 
 Review comment:
   Alternative:
   ```java
   boolean processed = false;
   while (mailboxExecutor.tryYield()) {
       processed = true;
   }
   return processed;
   ```
   
   You can also consider to call `tryYield()` only once approach for now:
   ```java
   protected boolean yieldToDownstream() {
       return mailboxExecutor.isMailboxThread() && mailboxExecutor.tryYield();
   }
   ```

----------------------------------------------------------------
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:
us...@infra.apache.org


With regards,
Apache Git Services

Reply via email to