[
https://issues.apache.org/jira/browse/FLINK-7416?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16268180#comment-16268180
]
ASF GitHub Bot commented on FLINK-7416:
---------------------------------------
Github user zhijiangW commented on a diff in the pull request:
https://github.com/apache/flink/pull/4533#discussion_r153401374
--- Diff:
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannel.java
---
@@ -283,10 +283,13 @@ public String toString() {
//
------------------------------------------------------------------------
/**
- * Enqueue this input channel in the pipeline for sending unannounced
credits to producer.
+ * Enqueue this input channel in the pipeline for notifying the
producer of unannounced credit.
*/
void notifyCreditAvailable() {
- //TODO in next PR
+ // We should skip the notification if this channel is already
released.
+ if (!isReleased.get() && partitionRequestClient != null) {
--- End diff --
In my early implementation, I checked the client state as above.
Now I change the current way based on two considerations:
1. The credit is always announced after receiving the sender's backlog, so
the partition request client should already be initialized before.
2. The check state here only wants to avoid unnecessary announcement after
the client is closed because of task failure or network exception. In this
case, the closed client is in expectation and we do not want to throw disturbed
messages by `checkState`.
> Implement Netty receiver outgoing pipeline for credit-based
> -----------------------------------------------------------
>
> Key: FLINK-7416
> URL: https://issues.apache.org/jira/browse/FLINK-7416
> Project: Flink
> Issue Type: Sub-task
> Components: Network
> Reporter: zhijiang
> Assignee: zhijiang
> Fix For: 1.5.0
>
>
> This is a part of work for credit-based network flow control.
> The related works are :
> * We define a new message called {{AddCredit}} to notify the incremental
> credit during data shuffle.
> * Whenever an {{InputChannel}}’s unannounced credit goes up from zero, the
> channel is enqueued in the pipeline.
> * Whenever the channel becomes writable, it takes the next {{InputChannel}}
> and sends its unannounced credit. The credit is reset to zero after each sent.
> * That way, messages are sent as often as the network has capacity and
> contain as much credit as available for the channel at that point in time.
> Otherwise, it would only add latency to the announcements and not increase
> throughput.
--
This message was sent by Atlassian JIRA
(v6.4.14#64029)
