Repository: flink Updated Branches: refs/heads/master 4f4c6b90d -> 0ca1f0c7b
[runtime] Add docs to BufferPool classes Project: http://git-wip-us.apache.org/repos/asf/flink/repo Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/0ca1f0c7 Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/0ca1f0c7 Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/0ca1f0c7 Branch: refs/heads/master Commit: 0ca1f0c7b76cc4b4199fb687be4592dbd1548757 Parents: 4f4c6b9 Author: Ufuk Celebi <u...@apache.org> Authored: Tue Apr 7 07:44:27 2015 +0200 Committer: Ufuk Celebi <u...@apache.org> Committed: Tue Apr 7 07:44:27 2015 +0200 ---------------------------------------------------------------------- .../runtime/io/network/buffer/BufferPool.java | 31 ++++++++++++++++++++ .../io/network/buffer/BufferPoolFactory.java | 14 ++++++++- .../io/network/buffer/BufferProvider.java | 13 ++++++++ .../io/network/buffer/LocalBufferPool.java | 4 +++ 4 files changed, 61 insertions(+), 1 deletion(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/flink/blob/0ca1f0c7/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/BufferPool.java ---------------------------------------------------------------------- diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/BufferPool.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/BufferPool.java index c2a3c05..fdfa201 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/BufferPool.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/BufferPool.java @@ -20,21 +20,52 @@ package org.apache.flink.runtime.io.network.buffer; import java.io.IOException; +/** + * A dynamically sized buffer pool. + */ public interface BufferPool extends BufferProvider, BufferRecycler { + /** + * The owner of this buffer pool to be called when memory needs to be released to avoid back + * pressure. + */ void setBufferPoolOwner(BufferPoolOwner owner); + /** + * Destroys this buffer pool. + * + * <p> If not all buffers are available, they are recycled lazily as soon as they are recycled. + */ void lazyDestroy(); + /** + * Checks whether this buffer pool has been destroyed. + */ @Override boolean isDestroyed(); + /** + * Returns the number of guaranteed (minimum number of) memory segments of this buffer pool. + */ int getNumberOfRequiredMemorySegments(); + /** + * Returns the current size of this buffer pool. + * + * <p> The size of the buffer pool can change dynamically at runtime. + */ int getNumBuffers(); + /** + * Sets the current size of this buffer pool. + * + * <p> The size needs to be greater or equals to the guaranteed number of memory segments. + */ void setNumBuffers(int numBuffers) throws IOException; + /** + * Returns the number memory segments, which are currently held by this buffer pool. + */ int getNumberOfAvailableMemorySegments(); } http://git-wip-us.apache.org/repos/asf/flink/blob/0ca1f0c7/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/BufferPoolFactory.java ---------------------------------------------------------------------- diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/BufferPoolFactory.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/BufferPoolFactory.java index fcd4d96..23321f4 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/BufferPoolFactory.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/BufferPoolFactory.java @@ -20,10 +20,22 @@ package org.apache.flink.runtime.io.network.buffer; import java.io.IOException; +/** + * A factory for buffer pools. + */ public interface BufferPoolFactory { + /** + * Tries to create a buffer pool, which is guaranteed to provide at least the number of required + * buffers. + * + * <p> The buffer pool is either of dynamic size or fixed. + */ BufferPool createBufferPool(int numRequiredBuffers, boolean isFixedSize) throws IOException; + /** + * Destroy callback for updating factory book keeping. + */ void destroyBufferPool(BufferPool bufferPool) throws IOException; -} \ No newline at end of file +} http://git-wip-us.apache.org/repos/asf/flink/blob/0ca1f0c7/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/BufferProvider.java ---------------------------------------------------------------------- diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/BufferProvider.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/BufferProvider.java index 937ea28..c3373fa 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/BufferProvider.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/BufferProvider.java @@ -22,6 +22,12 @@ import org.apache.flink.runtime.util.event.EventListener; import java.io.IOException; +/** + * A buffer provider to request buffers from in a synchronous or asynchronous fashion. + * + * <p> The data producing side (result partition writers) request buffers in a synchronous fashion, + * whereas the input side requests asynchronously. + */ public interface BufferProvider { /** @@ -50,8 +56,15 @@ public interface BufferProvider { */ boolean addListener(EventListener<Buffer> listener); + /** + * Returns whether the buffer provider has been destroyed. + */ boolean isDestroyed(); + /** + * Returns the size of the underlying memory segments. This is the maximum size a {@link Buffer} + * instance can have. + */ int getMemorySegmentSize(); } http://git-wip-us.apache.org/repos/asf/flink/blob/0ca1f0c7/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/LocalBufferPool.java ---------------------------------------------------------------------- diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/LocalBufferPool.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/LocalBufferPool.java index 4117e44..1da2b8b 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/LocalBufferPool.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/LocalBufferPool.java @@ -38,6 +38,10 @@ import static com.google.common.base.Preconditions.checkState; * local buffer pool. It also implements the default mechanism for buffer * recycling, which ensures that every buffer is ultimately returned to the * network buffer pool. + * + * <p> The size of this pool can be dynamically changed at runtime ({@link #setNumBuffers(int)}. It + * will then lazily return the required number of buffers to the {@link NetworkBufferPool} to + * match its new size. */ class LocalBufferPool implements BufferPool {
