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
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
