[
https://issues.apache.org/jira/browse/FLINK-8581?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16365554#comment-16365554
]
ASF GitHub Bot commented on FLINK-8581:
---------------------------------------
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/5423#discussion_r168480406
--- Diff:
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartition.java
---
@@ -55,69 +51,56 @@
/** Flag indicating whether the subpartition has been released. */
private volatile boolean isReleased;
- /** The number of non-event buffers currently in this subpartition. */
- @GuardedBy("buffers")
- private int buffersInBacklog;
-
//
------------------------------------------------------------------------
PipelinedSubpartition(int index, ResultPartition parent) {
super(index, parent);
}
@Override
- public boolean add(Buffer buffer) throws IOException {
- checkNotNull(buffer);
-
- // view reference accessible outside the lock, but assigned
inside the locked scope
- final PipelinedSubpartitionView reader;
+ public boolean add(BufferConsumer bufferConsumer) throws IOException {
+ return add(bufferConsumer, false);
+ }
+ @Override
+ public void flush() {
synchronized (buffers) {
- if (isFinished || isReleased) {
- buffer.recycleBuffer();
- return false;
+ if (readView != null) {
+ readView.notifyDataAvailable();
}
-
- // Add the buffer and update the stats
- buffers.add(buffer);
- reader = readView;
- updateStatistics(buffer);
- increaseBuffersInBacklog(buffer);
- }
-
- // Notify the listener outside of the synchronized block
- if (reader != null) {
- reader.notifyBuffersAvailable(1);
}
-
- return true;
}
@Override
public void finish() throws IOException {
- final Buffer buffer =
EventSerializer.toBuffer(EndOfPartitionEvent.INSTANCE);
+
add(EventSerializer.toBufferConsumer(EndOfPartitionEvent.INSTANCE), true);
+ LOG.debug("Finished {}.", this);
+ }
- // view reference accessible outside the lock, but assigned
inside the locked scope
- final PipelinedSubpartitionView reader;
+ private boolean add(BufferConsumer bufferConsumer, boolean finish)
throws IOException {
+ checkNotNull(bufferConsumer);
synchronized (buffers) {
if (isFinished || isReleased) {
- return;
+ bufferConsumer.close();
+ return false;
}
- buffers.add(buffer);
- reader = readView;
- updateStatistics(buffer);
+ // Add the bufferConsumer and update the stats
+ buffers.add(bufferConsumer);
+ updateStatistics(bufferConsumer);
+ increaseBuffersInBacklog(bufferConsumer);
- isFinished = true;
+ if (finish) {
+ isFinished = true;
+ notifyDataAvailable();
--- End diff --
RTFM :)
https://github.com/apache/flink/pull/5423/commits/982edbce98db0bb7a5db0514d67aed0435a95d0f
1. it was done as separate commit, so is not related to rest of the changes
2. from commit message
> notifyBuffersAvailable is a quick call that doesn't need to be executed
outside of the lock
advantages of commit by commit reviewing ;)
double-confirm - no, this notification is very quick call, it only enqueue
some work on a Netty's `Executor`.
> Improve performance for low latency network
> -------------------------------------------
>
> Key: FLINK-8581
> URL: https://issues.apache.org/jira/browse/FLINK-8581
> Project: Flink
> Issue Type: Improvement
> Components: Network
> Affects Versions: 1.5.0
> Reporter: Piotr Nowojski
> Assignee: Piotr Nowojski
> Priority: Major
>
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
