[
https://issues.apache.org/jira/browse/FLINK-5169?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15701586#comment-15701586
]
ASF GitHub Bot commented on FLINK-5169:
---------------------------------------
GitHub user uce opened a pull request:
https://github.com/apache/flink/pull/2882
[FLINK-5169] [network] Make consumption of InputChannels fair
We worked with @StephanEwen on this. The changes touched many tests and I
decided to separate the changes in two commits, one for the main change and one
for the adjusted tests. This way, I hope that the main changes are easier to
review.
# Changes
The goal of the changes was to make the consumption of input channels
fairer. For batch jobs, this unfairness was not a real issue, but for streaming
jobs it affects the checkpointing badly as checkpoint barriers flow inline with
the data and hence are affected by this unfairness.
## InputGate
The input gates are now notified whenever a channel becomes non-empty and
check whether they need to re-queue an input channel whenever they consume one.
Therefore, channels now return `BufferAndAvailability` and not just a `Buffer`.
The same logic is applied for input gates with the `UnionInputGate`, which
is notified when a `InputGate` becomes non-empty.
### Before
- Channel notifies gate on each available `Buffer` instance
- Each notification queues channel in `inputChannelsWithData`, possibly
queueing each channel multiple times (this is where the unfairness happened)
### Now
- Channel notifies gate on non-emptiness
- Each channel re-queued by the gate if it has more data available by the
input gate itself
- Each channel is only queued once in `inputChannelsWithData` (this is
where it becomes fairer)
## ResultSubpartition
In order to allow this change to be implemented reliably, we had to change
the partition consumption interfaces. When creating a sub partition reader we
now instantiate the reader with a buffer availability listener, which is
notified about available buffers.
This gets rid of some "look ahead magic" in the `LocalInputChannel` and
replaces it with pretty straight forward notifications. Furthermore, we don't
need the checkNotNull-registerListener loops when consuming data.
A shortcoming is that in some parts we have to be very aware that the
notifications happen when the reader is created (see
`LocalInputChannel#checkAndWaitForSubpartitionView` and
`PartitionRequestQueue#notifyReaderNonEmpty`). This is not nice, but all in all
an improvement over the previous state. We can refactor this further to
smoothen things out.
The unused `IOMode` for spillable partition types has been removed. I fear
that these partition types are quite fragile at the moment and most probably
should be revisted in a larger refactoring.
---
I would like to have this in for the upcoming 1.1.4 release. It is quite a
big change for a bug fix release, but in large scale setups this unfairness
interplays badly with the checkpoint barriers.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/uce/flink 5169-fairconsumers-1.1
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/flink/pull/2882.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #2882
----
commit 2aad264e6d9273d15b0e04527f0f2593d2951e8c
Author: Stephan Ewen <[email protected]>
Date: 2016-11-27T17:15:40Z
[FLINK-5169] [network] Add tests for channel consumption
commit 212d6ac7c86a4426b6e4418ae5f9d1d4680ec05b
Author: Stephan Ewen <[email protected]>
Date: 2016-11-28T08:59:29Z
[FLINK-5169] [network] Make consumption of InputChannels fair
commit 564b96d72b1536f4ab171c723e3ae0ab2e3c935c
Author: Ufuk Celebi <[email protected]>
Date: 2016-11-28T08:59:58Z
[FLINK-5169] [network] Adjust tests to new consumer logic
----
> Make consumption of input channels fair
> ---------------------------------------
>
> Key: FLINK-5169
> URL: https://issues.apache.org/jira/browse/FLINK-5169
> Project: Flink
> Issue Type: Improvement
> Components: Network
> Reporter: Ufuk Celebi
> Assignee: Ufuk Celebi
> Priority: Critical
> Fix For: 1.2.0, 1.1.4
>
>
> The input channels on the receiver side of the network stack queue incoming
> data and notify the input gate about available data. These notifications
> currently determine the order in which input channels are consumed, which can
> lead to unfair consumption patterns where faster channels are favored over
> slower ones.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
