You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by pn...@apache.org on 2021/06/07 09:01:54 UTC
[flink] 01/04: [FLINK-22376][runtime] BufferProvider is able to
provide pure MemorySegment in order to avoid extracting it from the
bufferBuilder
This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a commit to branch release-1.13
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 8511ecbc33e613d8554911ba8b943bf7a98ca77c
Author: Anton Kalashnikov <ka...@yandex.ru>
AuthorDate: Fri May 28 11:51:37 2021 +0200
[FLINK-22376][runtime] BufferProvider is able to provide pure MemorySegment in order to avoid extracting it from the bufferBuilder
(cherry picked from commit 397468706943aeed4ec2cf72f846e4457e3e1cd4)
---
.../runtime/io/network/buffer/BufferBuilder.java | 6 --
.../runtime/io/network/buffer/BufferProvider.java | 18 +++++
.../runtime/io/network/buffer/LocalBufferPool.java | 11 ++-
.../network/partition/PartitionSortedBuffer.java | 28 ++++----
.../partition/SortMergeResultPartition.java | 25 ++++---
.../io/network/buffer/LocalBufferPoolTest.java | 3 +-
.../runtime/io/network/buffer/NoOpBufferPool.java | 10 +++
.../io/network/buffer/UnpooledBufferPool.java | 8 ++-
.../runtime/io/network/util/TestBufferFactory.java | 11 ++-
.../io/network/util/TestPooledBufferProvider.java | 82 +++++++++++-----------
10 files changed, 120 insertions(+), 82 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/BufferBuilder.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/BufferBuilder.java
index c09827f..a5ae7ce 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/BufferBuilder.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/BufferBuilder.java
@@ -18,7 +18,6 @@
package org.apache.flink.runtime.io.network.buffer;
-import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.core.memory.MemorySegment;
import javax.annotation.concurrent.NotThreadSafe;
@@ -156,11 +155,6 @@ public class BufferBuilder {
recycler.recycle(memorySegment);
}
- @VisibleForTesting
- public MemorySegment getMemorySegment() {
- return memorySegment;
- }
-
/**
* Holds a reference to the current writer position. Negative values indicate that writer
* ({@link BufferBuilder} has finished. Value {@code Integer.MIN_VALUE} represents finished
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 8606f61..e5d38df 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
@@ -18,6 +18,7 @@
package org.apache.flink.runtime.io.network.buffer;
+import org.apache.flink.core.memory.MemorySegment;
import org.apache.flink.runtime.io.AvailabilityProvider;
import javax.annotation.Nullable;
@@ -85,4 +86,21 @@ public interface BufferProvider extends AvailabilityProvider {
/** Returns whether the buffer provider has been destroyed. */
boolean isDestroyed();
+
+ /**
+ * Returns a {@link MemorySegment} instance from the buffer provider.
+ *
+ * @return {@code null} if no memory segment is available or the buffer provider has been
+ * destroyed.
+ */
+ @Nullable
+ MemorySegment requestMemorySegment();
+
+ /**
+ * Returns a {@link MemorySegment} instance from the buffer provider.
+ *
+ * <p>If there is no memory segment available, the call will block until one becomes available
+ * again or the buffer provider has been destroyed.
+ */
+ MemorySegment requestMemorySegmentBlocking() throws InterruptedException;
}
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 6748504..6c14824 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
@@ -281,7 +281,12 @@ class LocalBufferPool implements BufferPool {
@Override
public BufferBuilder requestBufferBuilderBlocking() throws InterruptedException {
- return toBufferBuilder(requestMemorySegmentBlocking(UNKNOWN_CHANNEL), UNKNOWN_CHANNEL);
+ return toBufferBuilder(requestMemorySegmentBlocking(), UNKNOWN_CHANNEL);
+ }
+
+ @Override
+ public MemorySegment requestMemorySegmentBlocking() throws InterruptedException {
+ return requestMemorySegmentBlocking(UNKNOWN_CHANNEL);
}
@Override
@@ -359,8 +364,8 @@ class LocalBufferPool implements BufferPool {
return segment;
}
- @Nullable
- private MemorySegment requestMemorySegment() {
+ @Override
+ public MemorySegment requestMemorySegment() {
return requestMemorySegment(UNKNOWN_CHANNEL);
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/PartitionSortedBuffer.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/PartitionSortedBuffer.java
index 81a2330..b432d30 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/PartitionSortedBuffer.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/PartitionSortedBuffer.java
@@ -21,7 +21,6 @@ package org.apache.flink.runtime.io.network.partition;
import org.apache.flink.core.memory.MemorySegment;
import org.apache.flink.core.memory.MemorySegmentFactory;
import org.apache.flink.runtime.io.network.buffer.Buffer;
-import org.apache.flink.runtime.io.network.buffer.BufferBuilder;
import org.apache.flink.runtime.io.network.buffer.BufferPool;
import org.apache.flink.runtime.io.network.buffer.NetworkBuffer;
@@ -67,7 +66,7 @@ public class PartitionSortedBuffer implements SortBuffer {
/** A segment list as a joint buffer which stores all records and index entries. */
@GuardedBy("lock")
- private final ArrayList<MemorySegment> buffers = new ArrayList<>();
+ private final ArrayList<MemorySegment> segments = new ArrayList<>();
/** Addresses of the first record's index entry for each subpartition. */
private final long[] firstIndexEntryAddresses;
@@ -184,7 +183,7 @@ public class PartitionSortedBuffer implements SortBuffer {
}
private void writeIndex(int channelIndex, int numRecordBytes, Buffer.DataType dataType) {
- MemorySegment segment = buffers.get(writeSegmentIndex);
+ MemorySegment segment = segments.get(writeSegmentIndex);
// record length takes the high 32 bits and data type takes the low 32 bits
segment.putLong(writeSegmentOffset, ((long) numRecordBytes << 32) | dataType.ordinal());
@@ -197,7 +196,7 @@ public class PartitionSortedBuffer implements SortBuffer {
if (lastIndexEntryAddress >= 0) {
// link the previous index entry of the given channel to the new index entry
- segment = buffers.get(getSegmentIndexFromPointer(lastIndexEntryAddress));
+ segment = segments.get(getSegmentIndexFromPointer(lastIndexEntryAddress));
segment.putLong(
getSegmentOffsetFromPointer(lastIndexEntryAddress) + 8, indexEntryAddress);
} else {
@@ -210,7 +209,7 @@ public class PartitionSortedBuffer implements SortBuffer {
private void writeRecord(ByteBuffer source) {
while (source.hasRemaining()) {
- MemorySegment segment = buffers.get(writeSegmentIndex);
+ MemorySegment segment = segments.get(writeSegmentIndex);
int toCopy = Math.min(bufferSize - writeSegmentOffset, source.remaining());
segment.put(writeSegmentOffset, source, toCopy);
@@ -222,7 +221,7 @@ public class PartitionSortedBuffer implements SortBuffer {
private boolean allocateBuffersForRecord(int numRecordBytes) throws IOException {
int numBytesRequired = INDEX_ENTRY_SIZE + numRecordBytes;
int availableBytes =
- writeSegmentIndex == buffers.size() ? 0 : bufferSize - writeSegmentOffset;
+ writeSegmentIndex == segments.size() ? 0 : bufferSize - writeSegmentOffset;
// return directly if current available bytes is adequate
if (availableBytes >= numBytesRequired) {
@@ -262,22 +261,21 @@ public class PartitionSortedBuffer implements SortBuffer {
throw new IllegalStateException("Sort buffer is already released.");
}
- buffers.add(segment);
+ segments.add(segment);
}
}
private MemorySegment requestBufferFromPool() throws IOException {
try {
// blocking request buffers if there is still guaranteed memory
- if (buffers.size() < numGuaranteedBuffers) {
- return bufferPool.requestBufferBuilderBlocking().getMemorySegment();
+ if (segments.size() < numGuaranteedBuffers) {
+ return bufferPool.requestMemorySegmentBlocking();
}
} catch (InterruptedException e) {
throw new IOException("Interrupted while requesting buffer.");
}
- BufferBuilder buffer = bufferPool.requestBufferBuilder();
- return buffer != null ? buffer.getMemorySegment() : null;
+ return bufferPool.requestMemorySegment();
}
private void updateWriteSegmentIndexAndOffset(int numBytes) {
@@ -303,7 +301,7 @@ public class PartitionSortedBuffer implements SortBuffer {
do {
int sourceSegmentIndex = getSegmentIndexFromPointer(readIndexEntryAddress);
int sourceSegmentOffset = getSegmentOffsetFromPointer(readIndexEntryAddress);
- MemorySegment sourceSegment = buffers.get(sourceSegmentIndex);
+ MemorySegment sourceSegment = segments.get(sourceSegmentIndex);
long lengthAndDataType = sourceSegment.getLong(sourceSegmentOffset);
int length = getSegmentIndexFromPointer(lengthAndDataType);
@@ -376,7 +374,7 @@ public class PartitionSortedBuffer implements SortBuffer {
int sourceRemainingBytes =
Math.min(bufferSize - sourceSegmentOffset, recordRemainingBytes);
int numBytes = Math.min(targetSegmentSize - targetSegmentOffset, sourceRemainingBytes);
- MemorySegment sourceSegment = buffers.get(sourceSegmentIndex);
+ MemorySegment sourceSegment = segments.get(sourceSegmentIndex);
sourceSegment.copyTo(sourceSegmentOffset, targetSegment, targetSegmentOffset, numBytes);
recordRemainingBytes -= numBytes;
@@ -445,10 +443,10 @@ public class PartitionSortedBuffer implements SortBuffer {
isReleased = true;
- for (MemorySegment segment : buffers) {
+ for (MemorySegment segment : segments) {
bufferPool.recycle(segment);
}
- buffers.clear();
+ segments.clear();
numTotalBytes = 0;
numTotalRecords = 0;
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/SortMergeResultPartition.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/SortMergeResultPartition.java
index 4442ca1..9382503 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/SortMergeResultPartition.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/SortMergeResultPartition.java
@@ -78,7 +78,7 @@ public class SortMergeResultPartition extends ResultPartition {
/** Buffers cut from the network buffer pool for data writing. */
@GuardedBy("lock")
- private final List<MemorySegment> writeBuffers = new ArrayList<>();
+ private final List<MemorySegment> writeSegments = new ArrayList<>();
/** Size of network buffer and write buffer. */
private final int networkBufferSize;
@@ -172,9 +172,8 @@ public class SortMergeResultPartition extends ResultPartition {
synchronized (lock) {
try {
for (int i = 0; i < numWriteBuffers; ++i) {
- MemorySegment segment =
- bufferPool.requestBufferBuilderBlocking().getMemorySegment();
- writeBuffers.add(segment);
+ MemorySegment segment = bufferPool.requestMemorySegmentBlocking();
+ writeSegments.add(segment);
}
} catch (InterruptedException exception) {
// the setup method does not allow InterruptedException
@@ -310,13 +309,13 @@ public class SortMergeResultPartition extends ResultPartition {
fileWriter.startNewRegion(isBroadcast);
List<BufferWithChannel> toWrite = new ArrayList<>();
- Queue<MemorySegment> segments = getWriteBuffers();
+ Queue<MemorySegment> segments = getWriteSegments();
while (sortBuffer.hasRemaining()) {
if (segments.isEmpty()) {
fileWriter.writeBuffers(toWrite);
toWrite.clear();
- segments = getWriteBuffers();
+ segments = getWriteSegments();
}
BufferWithChannel bufferWithChannel =
@@ -339,10 +338,10 @@ public class SortMergeResultPartition extends ResultPartition {
flushSortBuffer(unicastSortBuffer, false);
}
- private Queue<MemorySegment> getWriteBuffers() {
+ private Queue<MemorySegment> getWriteSegments() {
synchronized (lock) {
- checkState(!writeBuffers.isEmpty(), "Task has been canceled.");
- return new ArrayDeque<>(writeBuffers);
+ checkState(!writeSegments.isEmpty(), "Task has been canceled.");
+ return new ArrayDeque<>(writeSegments);
}
}
@@ -371,13 +370,13 @@ public class SortMergeResultPartition extends ResultPartition {
fileWriter.startNewRegion(isBroadcast);
List<BufferWithChannel> toWrite = new ArrayList<>();
- Queue<MemorySegment> segments = getWriteBuffers();
+ Queue<MemorySegment> segments = getWriteSegments();
while (record.hasRemaining()) {
if (segments.isEmpty()) {
fileWriter.writeBuffers(toWrite);
toWrite.clear();
- segments = getWriteBuffers();
+ segments = getWriteSegments();
}
int toCopy = Math.min(record.remaining(), networkBufferSize);
@@ -414,10 +413,10 @@ public class SortMergeResultPartition extends ResultPartition {
private void releaseWriteBuffers() {
synchronized (lock) {
if (bufferPool != null) {
- for (MemorySegment segment : writeBuffers) {
+ for (MemorySegment segment : writeSegments) {
bufferPool.recycle(segment);
}
- writeBuffers.clear();
+ writeSegments.clear();
}
}
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/buffer/LocalBufferPoolTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/buffer/LocalBufferPoolTest.java
index c5115a6..545d37a 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/buffer/LocalBufferPoolTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/buffer/LocalBufferPoolTest.java
@@ -514,6 +514,7 @@ public class LocalBufferPoolTest extends TestLogger {
// recycle the requested buffer
bufferBuilder.createBufferConsumer().close();
+ bufferBuilder.recycle();
assertTrue(localBufferPool.isAvailable());
assertTrue(availableFuture2.isDone());
}
@@ -524,7 +525,7 @@ public class LocalBufferPoolTest extends TestLogger {
NetworkBufferPool globalPool = new TestNetworkBufferPool(numBuffers, memorySegmentSize);
try {
BufferPool localPool = new LocalBufferPool(globalPool, 1);
- MemorySegment segment = localPool.requestBufferBuilderBlocking().getMemorySegment();
+ MemorySegment segment = localPool.requestMemorySegmentBlocking();
localPool.setNumBuffers(2);
localPool.recycle(segment);
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/buffer/NoOpBufferPool.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/buffer/NoOpBufferPool.java
index b93de38..a7cc030 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/buffer/NoOpBufferPool.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/buffer/NoOpBufferPool.java
@@ -67,6 +67,16 @@ public class NoOpBufferPool implements BufferPool {
}
@Override
+ public MemorySegment requestMemorySegment() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public MemorySegment requestMemorySegmentBlocking() throws InterruptedException {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
public int getNumberOfRequiredMemorySegments() {
throw new UnsupportedOperationException();
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/buffer/UnpooledBufferPool.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/buffer/UnpooledBufferPool.java
index d2598c5..73b0804 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/buffer/UnpooledBufferPool.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/buffer/UnpooledBufferPool.java
@@ -41,11 +41,17 @@ public class UnpooledBufferPool implements BufferPool {
return new BufferBuilder(requestMemorySegment(), this);
}
- private MemorySegment requestMemorySegment() {
+ @Override
+ public MemorySegment requestMemorySegment() {
return MemorySegmentFactory.allocateUnpooledOffHeapMemory(SEGMENT_SIZE, null);
}
@Override
+ public MemorySegment requestMemorySegmentBlocking() throws InterruptedException {
+ return requestMemorySegment();
+ }
+
+ @Override
public BufferBuilder requestBufferBuilderBlocking() throws InterruptedException {
return requestBufferBuilder();
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/util/TestBufferFactory.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/util/TestBufferFactory.java
index e1877bf..22e72b5 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/util/TestBufferFactory.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/util/TestBufferFactory.java
@@ -18,6 +18,7 @@
package org.apache.flink.runtime.io.network.util;
+import org.apache.flink.core.memory.MemorySegment;
import org.apache.flink.core.memory.MemorySegmentFactory;
import org.apache.flink.runtime.io.network.buffer.Buffer;
import org.apache.flink.runtime.io.network.buffer.BufferRecycler;
@@ -51,14 +52,18 @@ public class TestBufferFactory {
this.bufferRecycler = checkNotNull(bufferRecycler);
}
- public synchronized Buffer create() {
+ public synchronized MemorySegment createMemorySegment() {
if (numberOfCreatedBuffers >= poolSize) {
return null;
}
numberOfCreatedBuffers++;
- return new NetworkBuffer(
- MemorySegmentFactory.allocateUnpooledSegment(bufferSize), bufferRecycler);
+ return MemorySegmentFactory.allocateUnpooledSegment(bufferSize);
+ }
+
+ public synchronized Buffer create() {
+ MemorySegment memorySegment = createMemorySegment();
+ return memorySegment == null ? null : new NetworkBuffer(memorySegment, bufferRecycler);
}
public synchronized int getNumberOfCreatedBuffers() {
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/util/TestPooledBufferProvider.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/util/TestPooledBufferProvider.java
index f92bc98..da6c4f0 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/util/TestPooledBufferProvider.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/util/TestPooledBufferProvider.java
@@ -36,9 +36,10 @@ import java.util.concurrent.LinkedBlockingDeque;
import static org.apache.flink.util.Preconditions.checkArgument;
+/** */
public class TestPooledBufferProvider implements BufferProvider {
- private final BlockingQueue<Buffer> buffers = new LinkedBlockingDeque<>();
+ private final BlockingQueue<MemorySegment> segments = new LinkedBlockingDeque<>();
private final TestBufferFactory bufferFactory;
@@ -51,25 +52,22 @@ public class TestPooledBufferProvider implements BufferProvider {
public TestPooledBufferProvider(int poolSize, int bufferSize) {
checkArgument(poolSize > 0);
- this.bufferRecycler = new PooledBufferProviderRecycler(buffers);
+ this.bufferRecycler = new PooledBufferProviderRecycler(segments);
this.bufferFactory = new TestBufferFactory(poolSize, bufferSize, bufferRecycler);
}
@Override
public Buffer requestBuffer() {
- final Buffer buffer = buffers.poll();
- if (buffer != null) {
- return buffer;
- }
+ MemorySegment memorySegment = requestMemorySegment();
- return bufferFactory.create();
+ return memorySegment == null ? null : new NetworkBuffer(memorySegment, bufferRecycler);
}
@Override
public BufferBuilder requestBufferBuilder() {
- Buffer buffer = requestBuffer();
- if (buffer != null) {
- return new BufferBuilder(buffer.getMemorySegment(), buffer.getRecycler());
+ MemorySegment memorySegment = requestMemorySegment();
+ if (memorySegment != null) {
+ return new BufferBuilder(memorySegment, bufferRecycler);
}
return null;
}
@@ -79,24 +77,9 @@ public class TestPooledBufferProvider implements BufferProvider {
return requestBufferBuilder();
}
- private Buffer requestBufferBlocking() throws InterruptedException {
- Buffer buffer = buffers.poll();
- if (buffer != null) {
- return buffer;
- }
-
- buffer = bufferFactory.create();
- if (buffer != null) {
- return buffer;
- }
-
- return buffers.take();
- }
-
@Override
public BufferBuilder requestBufferBuilderBlocking() throws InterruptedException {
- Buffer buffer = requestBufferBlocking();
- return new BufferBuilder(buffer.getMemorySegment(), buffer.getRecycler());
+ return new BufferBuilder(requestMemorySegmentBlocking(), bufferRecycler);
}
@Override
@@ -116,49 +99,68 @@ public class TestPooledBufferProvider implements BufferProvider {
}
@Override
- public CompletableFuture<?> getAvailableFuture() {
- return AVAILABLE;
+ public MemorySegment requestMemorySegment() {
+ final MemorySegment buffer = segments.poll();
+ if (buffer != null) {
+ return buffer;
+ }
+
+ return bufferFactory.createMemorySegment();
}
- public int getNumberOfAvailableBuffers() {
- return buffers.size();
+ @Override
+ public MemorySegment requestMemorySegmentBlocking() throws InterruptedException {
+ MemorySegment buffer = segments.poll();
+ if (buffer != null) {
+ return buffer;
+ }
+
+ buffer = bufferFactory.createMemorySegment();
+ if (buffer != null) {
+ return buffer;
+ }
+
+ return segments.take();
+ }
+
+ @Override
+ public CompletableFuture<?> getAvailableFuture() {
+ return AVAILABLE;
}
- public int getNumberOfCreatedBuffers() {
- return bufferFactory.getNumberOfCreatedBuffers();
+ public int getNumberOfAvailableSegments() {
+ return segments.size();
}
private static class PooledBufferProviderRecycler implements BufferRecycler {
private final Object listenerRegistrationLock = new Object();
- private final Queue<Buffer> buffers;
+ private final Queue<MemorySegment> segments;
private final ConcurrentLinkedQueue<BufferListener> registeredListeners =
Queues.newConcurrentLinkedQueue();
- public PooledBufferProviderRecycler(Queue<Buffer> buffers) {
- this.buffers = buffers;
+ public PooledBufferProviderRecycler(Queue<MemorySegment> segments) {
+ this.segments = segments;
}
@Override
public void recycle(MemorySegment segment) {
synchronized (listenerRegistrationLock) {
- final Buffer buffer = new NetworkBuffer(segment, this);
-
BufferListener listener = registeredListeners.poll();
if (listener == null) {
- buffers.add(buffer);
+ segments.add(segment);
} else {
- listener.notifyBufferAvailable(buffer);
+ listener.notifyBufferAvailable(new NetworkBuffer(segment, this));
}
}
}
boolean registerListener(BufferListener listener) {
synchronized (listenerRegistrationLock) {
- if (buffers.isEmpty()) {
+ if (segments.isEmpty()) {
registeredListeners.add(listener);
return true;