You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tr...@apache.org on 2018/07/04 05:52:02 UTC
[06/10] flink git commit: [hotfix][network] add a few more checks and
tags
[hotfix][network] add a few more checks and tags
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/b3c47961
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/b3c47961
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/b3c47961
Branch: refs/heads/master
Commit: b3c47961adaa809b0b60a308f1621c60299b2590
Parents: 918b16a
Author: Nico Kruber <ni...@data-artisans.com>
Authored: Mon Jul 2 12:42:40 2018 +0200
Committer: Till Rohrmann <tr...@apache.org>
Committed: Wed Jul 4 07:51:28 2018 +0200
----------------------------------------------------------------------
.../flink/runtime/io/network/buffer/NetworkBufferPool.java | 6 +++++-
1 file changed, 5 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/b3c47961/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/NetworkBufferPool.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/NetworkBufferPool.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/NetworkBufferPool.java
index 7a5832a..d5846ce 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/NetworkBufferPool.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/NetworkBufferPool.java
@@ -27,6 +27,8 @@ import org.apache.flink.util.MathUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import javax.annotation.Nullable;
+
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
@@ -37,6 +39,7 @@ import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.TimeUnit;
import static org.apache.flink.util.Preconditions.checkArgument;
+import static org.apache.flink.util.Preconditions.checkNotNull;
/**
* The NetworkBufferPool is a fixed size pool of {@link MemorySegment} instances
@@ -112,6 +115,7 @@ public class NetworkBufferPool implements BufferPoolFactory {
allocatedMb, availableMemorySegments.size(), segmentSize);
}
+ @Nullable
public MemorySegment requestMemorySegment() {
return availableMemorySegments.poll();
}
@@ -120,7 +124,7 @@ public class NetworkBufferPool implements BufferPoolFactory {
// Adds the segment back to the queue, which does not immediately free the memory
// however, since this happens when references to the global pool are also released,
// making the availableMemorySegments queue and its contained object reclaimable
- availableMemorySegments.add(segment);
+ availableMemorySegments.add(checkNotNull(segment));
}
public List<MemorySegment> requestMemorySegments(int numRequiredBuffers) throws IOException {