You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by sr...@apache.org on 2018/02/16 12:55:57 UTC
[4/5] flink git commit: [FLINK-7456][network] Implement Netty sender
incoming pipeline for credit-based
http://git-wip-us.apache.org/repos/asf/flink/blob/a9fdbc4b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/PartitionRequestClientHandlerTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/PartitionRequestClientHandlerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/PartitionRequestClientHandlerTest.java
index fff0e22..cc670e1 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/PartitionRequestClientHandlerTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/PartitionRequestClientHandlerTest.java
@@ -25,11 +25,12 @@ import org.apache.flink.runtime.io.network.buffer.Buffer;
import org.apache.flink.runtime.io.network.buffer.BufferListener;
import org.apache.flink.runtime.io.network.buffer.BufferPool;
import org.apache.flink.runtime.io.network.buffer.BufferProvider;
-import org.apache.flink.runtime.io.network.buffer.NetworkBuffer;
import org.apache.flink.runtime.io.network.buffer.NetworkBufferPool;
import org.apache.flink.runtime.io.network.netty.NettyMessage.AddCredit;
import org.apache.flink.runtime.io.network.netty.NettyMessage.BufferResponse;
+import org.apache.flink.runtime.io.network.netty.NettyMessage.CloseRequest;
import org.apache.flink.runtime.io.network.netty.NettyMessage.ErrorResponse;
+import org.apache.flink.runtime.io.network.netty.NettyMessage.PartitionRequest;
import org.apache.flink.runtime.io.network.partition.PartitionNotFoundException;
import org.apache.flink.runtime.io.network.partition.ResultPartitionID;
import org.apache.flink.runtime.io.network.partition.ResultPartitionType;
@@ -42,7 +43,6 @@ import org.apache.flink.runtime.metrics.groups.UnregisteredMetricGroups;
import org.apache.flink.runtime.taskmanager.TaskActions;
import org.apache.flink.shaded.netty4.io.netty.buffer.ByteBuf;
-import org.apache.flink.shaded.netty4.io.netty.buffer.Unpooled;
import org.apache.flink.shaded.netty4.io.netty.buffer.UnpooledByteBufAllocator;
import org.apache.flink.shaded.netty4.io.netty.channel.Channel;
import org.apache.flink.shaded.netty4.io.netty.channel.ChannelHandlerContext;
@@ -52,6 +52,7 @@ import org.junit.Test;
import java.io.IOException;
+import static org.apache.flink.runtime.io.network.netty.PartitionRequestQueueTest.blockChannel;
import static org.hamcrest.Matchers.instanceOf;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
@@ -124,7 +125,7 @@ public class PartitionRequestClientHandlerTest {
final BufferResponse receivedBuffer = createBufferResponse(
emptyBuffer, 0, inputChannel.getInputChannelId(), backlog);
- final CreditBasedClientHandler client = new CreditBasedClientHandler();
+ final PartitionRequestClientHandler client = new PartitionRequestClientHandler();
client.addInputChannel(inputChannel);
// Read the empty buffer
@@ -144,14 +145,13 @@ public class PartitionRequestClientHandlerTest {
final NetworkBufferPool networkBufferPool = new NetworkBufferPool(10, 32);
final SingleInputGate inputGate = createSingleInputGate();
final RemoteInputChannel inputChannel = createRemoteInputChannel(inputGate);
- inputGate.setInputChannel(inputChannel.getPartitionId().getPartitionId(), inputChannel);
try {
final BufferPool bufferPool = networkBufferPool.createBufferPool(8, 8);
inputGate.setBufferPool(bufferPool);
final int numExclusiveBuffers = 2;
inputGate.assignExclusiveSegments(networkBufferPool, numExclusiveBuffers);
- final CreditBasedClientHandler handler = new CreditBasedClientHandler();
+ final PartitionRequestClientHandler handler = new PartitionRequestClientHandler();
handler.addInputChannel(inputChannel);
final int backlog = 2;
@@ -179,7 +179,7 @@ public class PartitionRequestClientHandlerTest {
final SingleInputGate inputGate = createSingleInputGate();
final RemoteInputChannel inputChannel = spy(createRemoteInputChannel(inputGate));
- final CreditBasedClientHandler handler = new CreditBasedClientHandler();
+ final PartitionRequestClientHandler handler = new PartitionRequestClientHandler();
handler.addInputChannel(inputChannel);
assertEquals("There should be no buffers available in the channel.",
@@ -246,30 +246,38 @@ public class PartitionRequestClientHandlerTest {
*/
@Test
public void testNotifyCreditAvailable() throws Exception {
+ final PartitionRequestClientHandler handler = new PartitionRequestClientHandler();
+ final EmbeddedChannel channel = new EmbeddedChannel(handler);
+ final PartitionRequestClient client = new PartitionRequestClient(
+ channel, handler, mock(ConnectionID.class), mock(PartitionRequestClientFactory.class));
+
final NetworkBufferPool networkBufferPool = new NetworkBufferPool(10, 32);
final SingleInputGate inputGate = createSingleInputGate();
- final RemoteInputChannel inputChannel1 = createRemoteInputChannel(inputGate);
- final RemoteInputChannel inputChannel2 = createRemoteInputChannel(inputGate);
- inputGate.setInputChannel(inputChannel1.getPartitionId().getPartitionId(), inputChannel1);
- inputGate.setInputChannel(inputChannel2.getPartitionId().getPartitionId(), inputChannel2);
+ final RemoteInputChannel inputChannel1 = createRemoteInputChannel(inputGate, client);
+ final RemoteInputChannel inputChannel2 = createRemoteInputChannel(inputGate, client);
try {
final BufferPool bufferPool = networkBufferPool.createBufferPool(6, 6);
inputGate.setBufferPool(bufferPool);
final int numExclusiveBuffers = 2;
inputGate.assignExclusiveSegments(networkBufferPool, numExclusiveBuffers);
- final CreditBasedClientHandler handler = new CreditBasedClientHandler();
- final EmbeddedChannel channel = new EmbeddedChannel(handler);
-
- // The PartitionRequestClient is tied to PartitionRequestClientHandler currently, so we
- // have to add input channels in CreditBasedClientHandler explicitly
inputChannel1.requestSubpartition(0);
inputChannel2.requestSubpartition(0);
- handler.addInputChannel(inputChannel1);
- handler.addInputChannel(inputChannel2);
+
+ // The two input channels should send partition requests
+ assertTrue(channel.isWritable());
+ Object readFromOutbound = channel.readOutbound();
+ assertThat(readFromOutbound, instanceOf(PartitionRequest.class));
+ assertEquals(inputChannel1.getInputChannelId(), ((PartitionRequest) readFromOutbound).receiverId);
+ assertEquals(2, ((PartitionRequest) readFromOutbound).credit);
+
+ readFromOutbound = channel.readOutbound();
+ assertThat(readFromOutbound, instanceOf(PartitionRequest.class));
+ assertEquals(inputChannel2.getInputChannelId(), ((PartitionRequest) readFromOutbound).receiverId);
+ assertEquals(2, ((PartitionRequest) readFromOutbound).credit);
// The buffer response will take one available buffer from input channel, and it will trigger
- // requesting (backlog + numExclusiveBuffers - numAvailableBuffers) floating buffers
+ // requesting (backlog + numExclusiveBuffers - numAvailableBuffers) floating buffers
final BufferResponse bufferResponse1 = createBufferResponse(
TestBufferFactory.createBuffer(32), 0, inputChannel1.getInputChannelId(), 1);
final BufferResponse bufferResponse2 = createBufferResponse(
@@ -277,37 +285,29 @@ public class PartitionRequestClientHandlerTest {
handler.channelRead(mock(ChannelHandlerContext.class), bufferResponse1);
handler.channelRead(mock(ChannelHandlerContext.class), bufferResponse2);
- // The PartitionRequestClient is tied to PartitionRequestClientHandler currently, so we
- // have to notify credit available in CreditBasedClientHandler explicitly
- handler.notifyCreditAvailable(inputChannel1);
- handler.notifyCreditAvailable(inputChannel2);
-
assertEquals(2, inputChannel1.getUnannouncedCredit());
assertEquals(2, inputChannel2.getUnannouncedCredit());
channel.runPendingTasks();
- // The two input channels should notify credits via writable channel
- assertTrue(channel.isWritable());
- Object readFromOutbound = channel.readOutbound();
+ // The two input channels should notify credits availability via the writable channel
+ readFromOutbound = channel.readOutbound();
assertThat(readFromOutbound, instanceOf(AddCredit.class));
+ assertEquals(inputChannel1.getInputChannelId(), ((AddCredit) readFromOutbound).receiverId);
assertEquals(2, ((AddCredit) readFromOutbound).credit);
+
readFromOutbound = channel.readOutbound();
assertThat(readFromOutbound, instanceOf(AddCredit.class));
+ assertEquals(inputChannel2.getInputChannelId(), ((AddCredit) readFromOutbound).receiverId);
assertEquals(2, ((AddCredit) readFromOutbound).credit);
assertNull(channel.readOutbound());
- final int highWaterMark = channel.config().getWriteBufferHighWaterMark();
- // Set the writer index to the high water mark to ensure that all bytes are written
- // to the wire although the buffer is "empty".
- ByteBuf channelBlockingBuffer = Unpooled.buffer(highWaterMark).writerIndex(highWaterMark);
- channel.write(channelBlockingBuffer);
+ ByteBuf channelBlockingBuffer = blockChannel(channel);
- // Trigger notify credits available via buffer response on the condition of un-writable channel
+ // Trigger notify credits availability via buffer response on the condition of an un-writable channel
final BufferResponse bufferResponse3 = createBufferResponse(
TestBufferFactory.createBuffer(32), 1, inputChannel1.getInputChannelId(), 1);
handler.channelRead(mock(ChannelHandlerContext.class), bufferResponse3);
- handler.notifyCreditAvailable(inputChannel1);
assertEquals(1, inputChannel1.getUnannouncedCredit());
assertEquals(0, inputChannel2.getUnannouncedCredit());
@@ -347,23 +347,26 @@ public class PartitionRequestClientHandlerTest {
*/
@Test
public void testNotifyCreditAvailableAfterReleased() throws Exception {
+ final PartitionRequestClientHandler handler = new PartitionRequestClientHandler();
+ final EmbeddedChannel channel = new EmbeddedChannel(handler);
+ final PartitionRequestClient client = new PartitionRequestClient(
+ channel, handler, mock(ConnectionID.class), mock(PartitionRequestClientFactory.class));
+
final NetworkBufferPool networkBufferPool = new NetworkBufferPool(10, 32);
final SingleInputGate inputGate = createSingleInputGate();
- final RemoteInputChannel inputChannel = createRemoteInputChannel(inputGate);
- inputGate.setInputChannel(inputChannel.getPartitionId().getPartitionId(), inputChannel);
+ final RemoteInputChannel inputChannel = createRemoteInputChannel(inputGate, client);
try {
final BufferPool bufferPool = networkBufferPool.createBufferPool(6, 6);
inputGate.setBufferPool(bufferPool);
final int numExclusiveBuffers = 2;
inputGate.assignExclusiveSegments(networkBufferPool, numExclusiveBuffers);
- final CreditBasedClientHandler handler = new CreditBasedClientHandler();
- final EmbeddedChannel channel = new EmbeddedChannel(handler);
-
- // The PartitionRequestClient is tied to PartitionRequestClientHandler currently, so we
- // have to add input channels in CreditBasedClientHandler explicitly
inputChannel.requestSubpartition(0);
- handler.addInputChannel(inputChannel);
+
+ // This should send the partition request
+ Object readFromOutbound = channel.readOutbound();
+ assertThat(readFromOutbound, instanceOf(PartitionRequest.class));
+ assertEquals(2, ((PartitionRequest) readFromOutbound).credit);
// Trigger request floating buffers via buffer response to notify credits available
final BufferResponse bufferResponse = createBufferResponse(
@@ -372,16 +375,16 @@ public class PartitionRequestClientHandlerTest {
assertEquals(2, inputChannel.getUnannouncedCredit());
- // The PartitionRequestClient is tied to PartitionRequestClientHandler currently, so we
- // have to notify credit available in CreditBasedClientHandler explicitly
- handler.notifyCreditAvailable(inputChannel);
-
// Release the input channel
inputGate.releaseAllResources();
+ // it should send a close request after releasing the input channel,
+ // but will not notify credits for a released input channel.
+ readFromOutbound = channel.readOutbound();
+ assertThat(readFromOutbound, instanceOf(CloseRequest.class));
+
channel.runPendingTasks();
- // It will not notify credits for released input channel
assertNull(channel.readOutbound());
} finally {
// Release all the buffer resources
@@ -399,12 +402,12 @@ public class PartitionRequestClientHandlerTest {
*
* @return The new created single input gate.
*/
- private SingleInputGate createSingleInputGate() {
+ static SingleInputGate createSingleInputGate() {
return new SingleInputGate(
"InputGate",
new JobID(),
new IntermediateDataSetID(),
- ResultPartitionType.PIPELINED_CREDIT_BASED,
+ ResultPartitionType.PIPELINED,
0,
1,
mock(TaskActions.class),
@@ -417,21 +420,52 @@ public class PartitionRequestClientHandlerTest {
* @param inputGate The input gate owns the created input channel.
* @return The new created remote input channel.
*/
- private RemoteInputChannel createRemoteInputChannel(SingleInputGate inputGate) throws Exception {
+ static RemoteInputChannel createRemoteInputChannel(SingleInputGate inputGate) throws Exception {
+ return createRemoteInputChannel(inputGate, mock(PartitionRequestClient.class));
+ }
+
+ /**
+ * Creates and returns a remote input channel for the specific input gate with specific partition request client.
+ *
+ * @param inputGate The input gate owns the created input channel.
+ * @param client The client is used to send partition request.
+ * @return The new created remote input channel.
+ */
+ static RemoteInputChannel createRemoteInputChannel(SingleInputGate inputGate, PartitionRequestClient client) throws Exception {
+ return createRemoteInputChannel(inputGate, client, 0, 0);
+ }
+
+ /**
+ * Creates and returns a remote input channel for the specific input gate with specific partition request client.
+ *
+ * @param inputGate The input gate owns the created input channel.
+ * @param client The client is used to send partition request.
+ * @param initialBackoff initial back off (in ms) for retriggering subpartition requests (must be <tt>> 0</tt> to activate)
+ * @param maxBackoff after which delay (in ms) to stop retriggering subpartition requests
+ * @return The new created remote input channel.
+ */
+ static RemoteInputChannel createRemoteInputChannel(
+ SingleInputGate inputGate,
+ PartitionRequestClient client,
+ int initialBackoff,
+ int maxBackoff) throws Exception {
final ConnectionManager connectionManager = mock(ConnectionManager.class);
- final PartitionRequestClient partitionRequestClient = mock(PartitionRequestClient.class);
when(connectionManager.createPartitionRequestClient(any(ConnectionID.class)))
- .thenReturn(partitionRequestClient);
+ .thenReturn(client);
- return new RemoteInputChannel(
+ ResultPartitionID partitionId = new ResultPartitionID();
+ RemoteInputChannel inputChannel = new RemoteInputChannel(
inputGate,
0,
- new ResultPartitionID(),
+ partitionId,
mock(ConnectionID.class),
connectionManager,
- 0,
- 0,
+ initialBackoff,
+ maxBackoff,
UnregisteredMetricGroups.createUnregisteredTaskMetricGroup().getIOMetricGroup());
+
+ inputGate.setInputChannel(partitionId.getPartitionId(), inputChannel);
+ return inputChannel;
}
/**
http://git-wip-us.apache.org/repos/asf/flink/blob/a9fdbc4b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/PartitionRequestClientTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/PartitionRequestClientTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/PartitionRequestClientTest.java
new file mode 100644
index 0000000..a11b0ba
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/PartitionRequestClientTest.java
@@ -0,0 +1,150 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.io.network.netty;
+
+import org.apache.flink.runtime.io.network.ConnectionID;
+import org.apache.flink.runtime.io.network.buffer.BufferPool;
+import org.apache.flink.runtime.io.network.buffer.NetworkBufferPool;
+import org.apache.flink.runtime.io.network.netty.NettyMessage.PartitionRequest;
+import org.apache.flink.runtime.io.network.partition.consumer.RemoteInputChannel;
+import org.apache.flink.runtime.io.network.partition.consumer.SingleInputGate;
+
+import org.apache.flink.shaded.netty4.io.netty.channel.embedded.EmbeddedChannel;
+
+import org.junit.Test;
+
+import static org.apache.flink.runtime.io.network.netty.PartitionRequestClientHandlerTest.createRemoteInputChannel;
+import static org.apache.flink.runtime.io.network.netty.PartitionRequestClientHandlerTest.createSingleInputGate;
+import static org.hamcrest.Matchers.instanceOf;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertThat;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.mock;
+
+/**
+ * Tests for {@link PartitionRequestClient}.
+ */
+public class PartitionRequestClientTest {
+
+ @Test
+ public void testRetriggerPartitionRequest() throws Exception {
+ final long deadline = System.currentTimeMillis() + 30_000L; // 30 secs
+
+ final PartitionRequestClientHandler handler = new PartitionRequestClientHandler();
+ final EmbeddedChannel channel = new EmbeddedChannel(handler);
+ final PartitionRequestClient client = new PartitionRequestClient(
+ channel, handler, mock(ConnectionID.class), mock(PartitionRequestClientFactory.class));
+
+ final NetworkBufferPool networkBufferPool = new NetworkBufferPool(10, 32);
+ final SingleInputGate inputGate = createSingleInputGate();
+ final RemoteInputChannel inputChannel = createRemoteInputChannel(inputGate, client, 1, 2);
+
+ try {
+ final BufferPool bufferPool = networkBufferPool.createBufferPool(6, 6);
+ inputGate.setBufferPool(bufferPool);
+ final int numExclusiveBuffers = 2;
+ inputGate.assignExclusiveSegments(networkBufferPool, numExclusiveBuffers);
+
+ // first subpartition request
+ inputChannel.requestSubpartition(0);
+
+ assertTrue(channel.isWritable());
+ Object readFromOutbound = channel.readOutbound();
+ assertThat(readFromOutbound, instanceOf(PartitionRequest.class));
+ assertEquals(inputChannel.getInputChannelId(), ((PartitionRequest) readFromOutbound).receiverId);
+ assertEquals(numExclusiveBuffers, ((PartitionRequest) readFromOutbound).credit);
+
+ // retrigger subpartition request, e.g. due to failures
+ inputGate.retriggerPartitionRequest(inputChannel.getPartitionId().getPartitionId());
+ runAllScheduledPendingTasks(channel, deadline);
+
+ readFromOutbound = channel.readOutbound();
+ assertThat(readFromOutbound, instanceOf(PartitionRequest.class));
+ assertEquals(inputChannel.getInputChannelId(), ((PartitionRequest) readFromOutbound).receiverId);
+ assertEquals(numExclusiveBuffers, ((PartitionRequest) readFromOutbound).credit);
+
+ // retrigger subpartition request once again, e.g. due to failures
+ inputGate.retriggerPartitionRequest(inputChannel.getPartitionId().getPartitionId());
+ runAllScheduledPendingTasks(channel, deadline);
+
+ readFromOutbound = channel.readOutbound();
+ assertThat(readFromOutbound, instanceOf(PartitionRequest.class));
+ assertEquals(inputChannel.getInputChannelId(), ((PartitionRequest) readFromOutbound).receiverId);
+ assertEquals(numExclusiveBuffers, ((PartitionRequest) readFromOutbound).credit);
+
+ assertNull(channel.readOutbound());
+ } finally {
+ // Release all the buffer resources
+ inputGate.releaseAllResources();
+
+ networkBufferPool.destroyAllBufferPools();
+ networkBufferPool.destroy();
+ }
+ }
+
+ @Test
+ public void testDoublePartitionRequest() throws Exception {
+ final PartitionRequestClientHandler handler = new PartitionRequestClientHandler();
+ final EmbeddedChannel channel = new EmbeddedChannel(handler);
+ final PartitionRequestClient client = new PartitionRequestClient(
+ channel, handler, mock(ConnectionID.class), mock(PartitionRequestClientFactory.class));
+
+ final NetworkBufferPool networkBufferPool = new NetworkBufferPool(10, 32);
+ final SingleInputGate inputGate = createSingleInputGate();
+ final RemoteInputChannel inputChannel = createRemoteInputChannel(inputGate, client);
+
+ try {
+ final BufferPool bufferPool = networkBufferPool.createBufferPool(6, 6);
+ inputGate.setBufferPool(bufferPool);
+ final int numExclusiveBuffers = 2;
+ inputGate.assignExclusiveSegments(networkBufferPool, numExclusiveBuffers);
+ inputChannel.requestSubpartition(0);
+
+ // The input channel should only send one partition request
+ assertTrue(channel.isWritable());
+ Object readFromOutbound = channel.readOutbound();
+ assertThat(readFromOutbound, instanceOf(PartitionRequest.class));
+ assertEquals(inputChannel.getInputChannelId(), ((PartitionRequest) readFromOutbound).receiverId);
+ assertEquals(numExclusiveBuffers, ((PartitionRequest) readFromOutbound).credit);
+
+ assertNull(channel.readOutbound());
+ } finally {
+ // Release all the buffer resources
+ inputGate.releaseAllResources();
+
+ networkBufferPool.destroyAllBufferPools();
+ networkBufferPool.destroy();
+ }
+ }
+
+ /**
+ * Run all pending scheduled tasks (waits until all tasks have been run or the deadline has passed.
+ *
+ * @param channel the channel to execute tasks for
+ * @param deadline maximum timestamp in ms to stop waiting further
+ * @throws InterruptedException
+ */
+ void runAllScheduledPendingTasks(EmbeddedChannel channel, long deadline) throws InterruptedException {
+ // NOTE: we don't have to be super fancy here; busy-polling with 1ms delays is enough
+ while (channel.runScheduledPendingTasks() != -1 && System.currentTimeMillis() < deadline) {
+ Thread.sleep(1);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/a9fdbc4b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/PartitionRequestQueueTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/PartitionRequestQueueTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/PartitionRequestQueueTest.java
index 17b87f7..a82776a 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/PartitionRequestQueueTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/PartitionRequestQueueTest.java
@@ -19,14 +19,15 @@
package org.apache.flink.runtime.io.network.netty;
import org.apache.flink.runtime.execution.CancelTaskException;
-import org.apache.flink.runtime.io.network.partition.BufferAvailabilityListener;
import org.apache.flink.runtime.io.network.partition.ResultPartitionID;
import org.apache.flink.runtime.io.network.partition.ResultPartitionProvider;
-import org.apache.flink.runtime.io.network.partition.ResultSubpartition;
+import org.apache.flink.runtime.io.network.partition.ResultSubpartition.BufferAndBacklog;
import org.apache.flink.runtime.io.network.partition.ResultSubpartitionView;
import org.apache.flink.runtime.io.network.partition.consumer.InputChannelID;
import org.apache.flink.runtime.io.network.util.TestBufferFactory;
+import org.apache.flink.shaded.netty4.io.netty.buffer.ByteBuf;
+import org.apache.flink.shaded.netty4.io.netty.buffer.Unpooled;
import org.apache.flink.shaded.netty4.io.netty.channel.embedded.EmbeddedChannel;
import org.junit.Test;
@@ -35,16 +36,15 @@ import javax.annotation.Nullable;
import java.io.IOException;
+import static org.hamcrest.Matchers.contains;
import static org.hamcrest.Matchers.instanceOf;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
-import static org.mockito.Matchers.any;
-import static org.mockito.Matchers.eq;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.when;
/**
* Tests for {@link PartitionRequestQueue}.
@@ -55,25 +55,18 @@ public class PartitionRequestQueueTest {
public void testProducerFailedException() throws Exception {
PartitionRequestQueue queue = new PartitionRequestQueue();
- ResultPartitionProvider partitionProvider = mock(ResultPartitionProvider.class);
- ResultPartitionID rpid = new ResultPartitionID();
+ ResultSubpartitionView view = new ReleasedResultSubpartitionView();
- ResultSubpartitionView view = mock(ResultSubpartitionView.class);
- when(view.isReleased()).thenReturn(true);
- when(view.getFailureCause()).thenReturn(new RuntimeException("Expected test exception"));
-
- when(partitionProvider.createSubpartitionView(
- eq(rpid),
- eq(0),
- any(BufferAvailabilityListener.class))).thenReturn(view);
+ ResultPartitionProvider partitionProvider =
+ (partitionId, index, availabilityListener) -> view;
EmbeddedChannel ch = new EmbeddedChannel(queue);
- SequenceNumberingViewReader seqView = new SequenceNumberingViewReader(new InputChannelID(), queue);
- seqView.requestSubpartitionView(partitionProvider, rpid, 0);
+ SequenceNumberingViewReader seqView = new SequenceNumberingViewReader(new InputChannelID(), 2, queue);
+ seqView.requestSubpartitionView(partitionProvider, new ResultPartitionID(), 0);
+ // Add available buffer to trigger enqueue the erroneous view
+ seqView.notifyBuffersAvailable(1);
- // Enqueue the erroneous view
- queue.notifyReaderNonEmpty(seqView);
ch.runPendingTasks();
// Read the enqueued msg
@@ -108,7 +101,7 @@ public class PartitionRequestQueueTest {
final InputChannelID receiverId = new InputChannelID();
final PartitionRequestQueue queue = new PartitionRequestQueue();
- final SequenceNumberingViewReader reader = new SequenceNumberingViewReader(receiverId, queue);
+ final SequenceNumberingViewReader reader = new SequenceNumberingViewReader(receiverId, 2, queue);
final EmbeddedChannel channel = new EmbeddedChannel(queue);
reader.requestSubpartitionView(partitionProvider, new ResultPartitionID(), 0);
@@ -138,10 +131,11 @@ public class PartitionRequestQueueTest {
@Nullable
@Override
- public ResultSubpartition.BufferAndBacklog getNextBuffer() {
- return new ResultSubpartition.BufferAndBacklog(
+ public BufferAndBacklog getNextBuffer() {
+ return new BufferAndBacklog(
TestBufferFactory.createBuffer(10),
- buffersInBacklog);
+ buffersInBacklog,
+ false);
}
}
@@ -155,16 +149,168 @@ public class PartitionRequestQueueTest {
@Nullable
@Override
- public ResultSubpartition.BufferAndBacklog getNextBuffer() {
- return new ResultSubpartition.BufferAndBacklog(
+ public BufferAndBacklog getNextBuffer() {
+ return new BufferAndBacklog(
TestBufferFactory.createBuffer(10).readOnlySlice(),
- buffersInBacklog);
+ buffersInBacklog,
+ false);
}
}
+ private static class ReleasedResultSubpartitionView extends NoOpResultSubpartitionView {
+ @Override
+ public boolean isReleased() {
+ return true;
+ }
+
+ @Override
+ public Throwable getFailureCause() {
+ return new RuntimeException("Expected test exception");
+ }
+ }
+
+ /**
+ * Tests {@link PartitionRequestQueue#enqueueAvailableReader(SequenceNumberingViewReader)},
+ * verifying the reader would be enqueued in the pipeline if the next sending buffer is an event
+ * even though it has no available credits.
+ */
+ @Test
+ public void testEnqueueReaderByNotifyingEventBuffer() throws Exception {
+ // setup
+ final ResultSubpartitionView view = new NextIsEventResultSubpartitionView();
+
+ ResultPartitionProvider partitionProvider =
+ (partitionId, index, availabilityListener) -> view;
+
+ final InputChannelID receiverId = new InputChannelID();
+ final PartitionRequestQueue queue = new PartitionRequestQueue();
+ final SequenceNumberingViewReader reader = new SequenceNumberingViewReader(receiverId, 0, queue);
+ final EmbeddedChannel channel = new EmbeddedChannel(queue);
+
+ reader.requestSubpartitionView(partitionProvider, new ResultPartitionID(), 0);
+
+ // block the channel so that we see an intermediate state in the test
+ ByteBuf channelBlockingBuffer = blockChannel(channel);
+ assertNull(channel.readOutbound());
+
+ // Notify an available event buffer to trigger enqueue the reader
+ reader.notifyBuffersAvailable(1);
+
+ channel.runPendingTasks();
+
+ // The reader is enqueued in the pipeline because the next buffer is an event, even though no credits are available
+ assertThat(queue.getAvailableReaders(), contains(reader)); // contains only (this) one!
+ assertEquals(0, reader.getNumCreditsAvailable());
+
+ // Flush the buffer to make the channel writable again and see the final results
+ channel.flush();
+ assertSame(channelBlockingBuffer, channel.readOutbound());
+
+ assertEquals(0, queue.getAvailableReaders().size());
+ assertEquals(0, reader.getNumCreditsAvailable());
+ assertNull(channel.readOutbound());
+ }
+
+ private static class NextIsEventResultSubpartitionView extends NoOpResultSubpartitionView {
+ @Override
+ public boolean nextBufferIsEvent() {
+ return true;
+ }
+ }
+
+ /**
+ * Tests {@link PartitionRequestQueue#enqueueAvailableReader(SequenceNumberingViewReader)},
+ * verifying the reader would be enqueued in the pipeline iff it has both available credits and buffers.
+ */
+ @Test
+ public void testEnqueueReaderByNotifyingBufferAndCredit() throws Exception {
+ // setup
+ final ResultSubpartitionView view = new DefaultBufferResultSubpartitionView(2);
+
+ ResultPartitionProvider partitionProvider =
+ (partitionId, index, availabilityListener) -> view;
+
+ final InputChannelID receiverId = new InputChannelID();
+ final PartitionRequestQueue queue = new PartitionRequestQueue();
+ final SequenceNumberingViewReader reader = new SequenceNumberingViewReader(receiverId, 0, queue);
+ final EmbeddedChannel channel = new EmbeddedChannel(queue);
+
+ reader.requestSubpartitionView(partitionProvider, new ResultPartitionID(), 0);
+ queue.notifyReaderCreated(reader);
+
+ // block the channel so that we see an intermediate state in the test
+ ByteBuf channelBlockingBuffer = blockChannel(channel);
+ assertNull(channel.readOutbound());
+
+ // Notify available buffers to trigger enqueue the reader
+ final int notifyNumBuffers = 5;
+ for (int i = 0; i < notifyNumBuffers; i++) {
+ reader.notifyBuffersAvailable(1);
+ }
+
+ channel.runPendingTasks();
+
+ // the reader is not enqueued in the pipeline because no credits are available
+ // -> it should still have the same number of pending buffers
+ assertEquals(0, queue.getAvailableReaders().size());
+ assertEquals(notifyNumBuffers, reader.getNumBuffersAvailable());
+ assertFalse(reader.isRegisteredAsAvailable());
+ assertEquals(0, reader.getNumCreditsAvailable());
+
+ // Notify available credits to trigger enqueue the reader again
+ final int notifyNumCredits = 3;
+ for (int i = 1; i <= notifyNumCredits; i++) {
+ queue.addCredit(receiverId, 1);
+
+ // the reader is enqueued in the pipeline because it has both available buffers and credits
+ // since the channel is blocked though, we will not process anything and only enqueue the
+ // reader once
+ assertTrue(reader.isRegisteredAsAvailable());
+ assertThat(queue.getAvailableReaders(), contains(reader)); // contains only (this) one!
+ assertEquals(i, reader.getNumCreditsAvailable());
+ assertEquals(notifyNumBuffers, reader.getNumBuffersAvailable());
+ }
+
+ // Flush the buffer to make the channel writable again and see the final results
+ channel.flush();
+ assertSame(channelBlockingBuffer, channel.readOutbound());
+
+ assertEquals(0, queue.getAvailableReaders().size());
+ assertEquals(0, reader.getNumCreditsAvailable());
+ assertEquals(notifyNumBuffers - notifyNumCredits, reader.getNumBuffersAvailable());
+ assertFalse(reader.isRegisteredAsAvailable());
+ for (int i = 1; i <= notifyNumCredits; i++) {
+ assertThat(channel.readOutbound(), instanceOf(NettyMessage.BufferResponse.class));
+ }
+ assertNull(channel.readOutbound());
+ }
+
+ /**
+ * Blocks the given channel by adding a buffer that is bigger than the high watermark.
+ *
+ * <p>The channel may be unblocked with:
+ * <pre>
+ * channel.flush();
+ * assertSame(channelBlockingBuffer, channel.readOutbound());
+ * </pre>
+ *
+ * @param channel the channel to block
+ * @return the created blocking buffer
+ */
+ static ByteBuf blockChannel(EmbeddedChannel channel) {
+ final int highWaterMark = channel.config().getWriteBufferHighWaterMark();
+ // Set the writer index to the high water mark to ensure that all bytes are written
+ // to the wire although the buffer is "empty".
+ ByteBuf channelBlockingBuffer = Unpooled.buffer(highWaterMark).writerIndex(highWaterMark);
+ channel.write(channelBlockingBuffer);
+ assertFalse(channel.isWritable());
+
+ return channelBlockingBuffer;
+ }
+
private static class NoOpResultSubpartitionView implements ResultSubpartitionView {
@Nullable
- public ResultSubpartition.BufferAndBacklog getNextBuffer() {
+ public BufferAndBacklog getNextBuffer() {
return null;
}
@@ -189,5 +335,10 @@ public class PartitionRequestQueueTest {
public Throwable getFailureCause() {
return null;
}
+
+ @Override
+ public boolean nextBufferIsEvent() {
+ return false;
+ }
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/a9fdbc4b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/ServerTransportErrorHandlingTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/ServerTransportErrorHandlingTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/ServerTransportErrorHandlingTest.java
index 5916162..929889b 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/ServerTransportErrorHandlingTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/ServerTransportErrorHandlingTest.java
@@ -100,7 +100,7 @@ public class ServerTransportErrorHandlingTest {
Channel ch = connect(serverAndClient);
// Write something to trigger close by server
- ch.writeAndFlush(new NettyMessage.PartitionRequest(new ResultPartitionID(), 0, new InputChannelID(), 2));
+ ch.writeAndFlush(new NettyMessage.PartitionRequest(new ResultPartitionID(), 0, new InputChannelID(), Integer.MAX_VALUE));
// Wait for the notification
if (!sync.await(TestingUtils.TESTING_DURATION().toMillis(), TimeUnit.MILLISECONDS)) {
http://git-wip-us.apache.org/repos/asf/flink/blob/a9fdbc4b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartitionTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartitionTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartitionTest.java
index 40be8df..7e369fa 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartitionTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartitionTest.java
@@ -101,13 +101,14 @@ public class PipelinedSubpartitionTest extends SubpartitionTestBase {
ResultSubpartitionView view = subpartition.createReadView(listener);
// Empty => should return null
+ assertFalse(view.nextBufferIsEvent());
assertNull(view.getNextBuffer());
+ assertFalse(view.nextBufferIsEvent()); // also after getNextBuffer()
verify(listener, times(1)).notifyBuffersAvailable(eq(0L));
// Add data to the queue...
subpartition.add(createBuffer(BUFFER_SIZE));
- assertEquals(1, subpartition.getTotalNumberOfBuffers());
- assertEquals(BUFFER_SIZE, subpartition.getTotalNumberOfBytes());
+ assertFalse(view.nextBufferIsEvent());
assertEquals(1, subpartition.getTotalNumberOfBuffers());
assertEquals(1, subpartition.getBuffersInBacklog());
@@ -117,30 +118,83 @@ public class PipelinedSubpartitionTest extends SubpartitionTestBase {
verify(listener, times(1)).notifyBuffersAvailable(eq(1L));
// ...and one available result
+ assertFalse(view.nextBufferIsEvent());
BufferAndBacklog read = view.getNextBuffer();
assertNotNull(read);
+ assertTrue(read.buffer().isBuffer());
assertEquals(0, subpartition.getBuffersInBacklog());
assertEquals(subpartition.getBuffersInBacklog(), read.buffersInBacklog());
+ assertFalse(read.nextBufferIsEvent());
+ assertFalse(view.nextBufferIsEvent());
assertNull(view.getNextBuffer());
assertEquals(0, subpartition.getBuffersInBacklog());
// Add data to the queue...
subpartition.add(createBuffer(BUFFER_SIZE));
+ assertFalse(view.nextBufferIsEvent());
assertEquals(2, subpartition.getTotalNumberOfBuffers());
assertEquals(1, subpartition.getBuffersInBacklog());
assertEquals(2 * BUFFER_SIZE, subpartition.getTotalNumberOfBytes());
verify(listener, times(2)).notifyBuffersAvailable(eq(1L));
- // Add event to the queue...
- Buffer event = createBuffer(BUFFER_SIZE);
- event.tagAsEvent();
- subpartition.add(event);
+ assertFalse(view.nextBufferIsEvent());
+ read = view.getNextBuffer();
+ assertNotNull(read);
+ assertTrue(read.buffer().isBuffer());
+ assertEquals(0, subpartition.getBuffersInBacklog());
+ assertEquals(subpartition.getBuffersInBacklog(), read.buffersInBacklog());
+ assertFalse(read.nextBufferIsEvent());
+ assertFalse(view.nextBufferIsEvent());
+ assertNull(view.getNextBuffer());
+ assertEquals(0, subpartition.getBuffersInBacklog());
+
+ // some tests with events
+
+ // fill with: buffer, event, and buffer
+ subpartition.add(createBuffer(BUFFER_SIZE));
+ assertFalse(view.nextBufferIsEvent());
+ {
+ Buffer event = createBuffer(BUFFER_SIZE);
+ event.tagAsEvent();
+ subpartition.add(event);
+ assertFalse(view.nextBufferIsEvent());
+ }
+ subpartition.add(createBuffer(BUFFER_SIZE));
+ assertFalse(view.nextBufferIsEvent());
+
+ assertEquals(5, subpartition.getTotalNumberOfBuffers());
+ assertEquals(2, subpartition.getBuffersInBacklog()); // two buffers (events don't count)
+ assertEquals(5 * BUFFER_SIZE, subpartition.getTotalNumberOfBytes());
+ verify(listener, times(5)).notifyBuffersAvailable(eq(1L));
+
+ assertFalse(view.nextBufferIsEvent()); // the first buffer
+ read = view.getNextBuffer();
+ assertNotNull(read);
+ assertTrue(read.buffer().isBuffer());
+ assertEquals(1, subpartition.getBuffersInBacklog());
+ assertEquals(subpartition.getBuffersInBacklog(), read.buffersInBacklog());
+ assertTrue(read.nextBufferIsEvent());
- assertEquals(3, subpartition.getTotalNumberOfBuffers());
+ assertTrue(view.nextBufferIsEvent()); // the event
+ read = view.getNextBuffer();
+ assertNotNull(read);
+ assertFalse(read.buffer().isBuffer());
assertEquals(1, subpartition.getBuffersInBacklog());
- assertEquals(3 * BUFFER_SIZE, subpartition.getTotalNumberOfBytes());
- verify(listener, times(3)).notifyBuffersAvailable(eq(1L));
+ assertEquals(subpartition.getBuffersInBacklog(), read.buffersInBacklog());
+ assertFalse(read.nextBufferIsEvent());
+
+ assertFalse(view.nextBufferIsEvent()); // the remaining buffer
+ read = view.getNextBuffer();
+ assertNotNull(read);
+ assertTrue(read.buffer().isBuffer());
+ assertEquals(0, subpartition.getBuffersInBacklog());
+ assertEquals(subpartition.getBuffersInBacklog(), read.buffersInBacklog());
+ assertFalse(read.nextBufferIsEvent());
+
+ assertEquals(5, subpartition.getTotalNumberOfBuffers());
+ assertEquals(5 * BUFFER_SIZE, subpartition.getTotalNumberOfBytes());
+ verify(listener, times(5)).notifyBuffersAvailable(eq(1L));
}
@Test
http://git-wip-us.apache.org/repos/asf/flink/blob/a9fdbc4b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/SpillableSubpartitionTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/SpillableSubpartitionTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/SpillableSubpartitionTest.java
index 4a03321..e9a2fee 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/SpillableSubpartitionTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/SpillableSubpartitionTest.java
@@ -54,19 +54,18 @@ import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertTrue;
import static org.mockito.Matchers.any;
-import static org.mockito.Matchers.eq;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.doNothing;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.spy;
-import static org.mockito.Mockito.times;
-import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
/**
* Tests for {@link SpillableSubpartition}.
*/
public class SpillableSubpartitionTest extends SubpartitionTestBase {
+ private static final int BUFFER_DATA_SIZE = 4096;
+
@Rule
public ExpectedException exception = ExpectedException.none();
@@ -193,80 +192,94 @@ public class SpillableSubpartitionTest extends SubpartitionTestBase {
public void testConsumeSpilledPartition() throws Exception {
SpillableSubpartition partition = createSubpartition();
- Buffer buffer = TestBufferFactory.createBuffer(4096, 4096);
+ Buffer buffer = TestBufferFactory.createBuffer(BUFFER_DATA_SIZE, BUFFER_DATA_SIZE);
buffer.retainBuffer();
buffer.retainBuffer();
partition.add(buffer);
partition.add(buffer);
+ {
+ Buffer event = TestBufferFactory.createBuffer(BUFFER_DATA_SIZE, BUFFER_DATA_SIZE);
+ event.tagAsEvent();
+ partition.add(event);
+ }
partition.add(buffer);
- assertEquals(3, partition.getTotalNumberOfBuffers());
- assertEquals(4096 * 3, partition.getTotalNumberOfBytes());
- assertEquals(3, partition.getTotalNumberOfBuffers());
+ assertEquals(4, partition.getTotalNumberOfBuffers());
assertEquals(3, partition.getBuffersInBacklog());
- assertEquals(4096 * 3, partition.getTotalNumberOfBytes());
+ assertEquals(BUFFER_DATA_SIZE * 4, partition.getTotalNumberOfBytes());
assertFalse(buffer.isRecycled());
- assertEquals(3, partition.releaseMemory());
- // now the buffer may be freed, depending on the timing of the write operation
- // -> let's do this check at the end of the test (to save some time)
- // stil same statistics
- assertEquals(3, partition.getTotalNumberOfBuffers());
- assertEquals(4096 * 3, partition.getTotalNumberOfBytes());
-
+ assertEquals(4, partition.releaseMemory());
// now the buffer may be freed, depending on the timing of the write operation
// -> let's do this check at the end of the test (to save some time)
// still same statistics
- assertEquals(3, partition.getTotalNumberOfBuffers());
+ assertEquals(4, partition.getTotalNumberOfBuffers());
assertEquals(3, partition.getBuffersInBacklog());
- assertEquals(4096 * 3, partition.getTotalNumberOfBytes());
+ assertEquals(BUFFER_DATA_SIZE * 4, partition.getTotalNumberOfBytes());
partition.finish();
// + one EndOfPartitionEvent
- assertEquals(4, partition.getTotalNumberOfBuffers());
- assertEquals(4096 * 3 + 4, partition.getTotalNumberOfBytes());
-
- // + one EndOfPartitionEvent
- assertEquals(4, partition.getTotalNumberOfBuffers());
+ assertEquals(5, partition.getTotalNumberOfBuffers());
assertEquals(3, partition.getBuffersInBacklog());
- assertEquals(4096 * 3 + 4, partition.getTotalNumberOfBytes());
+ assertEquals(BUFFER_DATA_SIZE * 4 + 4, partition.getTotalNumberOfBytes());
- BufferAvailabilityListener listener = spy(new AwaitableBufferAvailablityListener());
+ AwaitableBufferAvailablityListener listener = new AwaitableBufferAvailablityListener();
SpilledSubpartitionView reader = (SpilledSubpartitionView) partition.createReadView(listener);
- verify(listener, times(1)).notifyBuffersAvailable(eq(4L));
+ assertEquals(1, listener.getNumNotifications());
+ assertEquals(5, listener.getNumNotifiedBuffers());
+ assertFalse(reader.nextBufferIsEvent()); // buffer
BufferAndBacklog read = reader.getNextBuffer();
assertNotNull(read);
+ assertTrue(read.buffer().isBuffer());
assertEquals(2, partition.getBuffersInBacklog());
assertEquals(partition.getBuffersInBacklog(), read.buffersInBacklog());
assertNotSame(buffer, read);
assertFalse(read.buffer().isRecycled());
read.buffer().recycleBuffer();
assertTrue(read.buffer().isRecycled());
+ assertFalse(read.nextBufferIsEvent());
+ assertFalse(reader.nextBufferIsEvent()); // buffer
read = reader.getNextBuffer();
assertNotNull(read);
+ assertTrue(read.buffer().isBuffer());
assertEquals(1, partition.getBuffersInBacklog());
assertEquals(partition.getBuffersInBacklog(), read.buffersInBacklog());
assertNotSame(buffer, read);
assertFalse(read.buffer().isRecycled());
read.buffer().recycleBuffer();
assertTrue(read.buffer().isRecycled());
+ assertTrue(read.nextBufferIsEvent());
+
+ assertTrue(reader.nextBufferIsEvent()); // event
+ read = reader.getNextBuffer();
+ assertNotNull(read);
+ assertFalse(read.buffer().isBuffer());
+ assertEquals(1, partition.getBuffersInBacklog());
+ assertEquals(partition.getBuffersInBacklog(), read.buffersInBacklog());
+ assertNotSame(buffer, read.buffer());
+ read.buffer().recycleBuffer();
+ assertFalse(read.nextBufferIsEvent());
+ assertFalse(reader.nextBufferIsEvent()); // buffer
read = reader.getNextBuffer();
assertNotNull(read);
+ assertTrue(read.buffer().isBuffer());
assertEquals(0, partition.getBuffersInBacklog());
assertEquals(partition.getBuffersInBacklog(), read.buffersInBacklog());
assertNotSame(buffer, read);
assertFalse(read.buffer().isRecycled());
read.buffer().recycleBuffer();
assertTrue(read.buffer().isRecycled());
+ assertTrue(read.nextBufferIsEvent());
- // End of partition
+ assertTrue(reader.nextBufferIsEvent()); // end of partition event
read = reader.getNextBuffer();
assertNotNull(read);
+ assertFalse(read.buffer().isBuffer());
assertEquals(0, partition.getBuffersInBacklog());
assertEquals(partition.getBuffersInBacklog(), read.buffersInBacklog());
assertEquals(EndOfPartitionEvent.class,
@@ -274,6 +287,7 @@ public class SpillableSubpartitionTest extends SubpartitionTestBase {
assertFalse(read.buffer().isRecycled());
read.buffer().recycleBuffer();
assertTrue(read.buffer().isRecycled());
+ assertFalse(read.nextBufferIsEvent());
// finally check that the buffer has been freed after a successful (or failed) write
final long deadline = System.currentTimeMillis() + 30_000L; // 30 secs
@@ -291,20 +305,23 @@ public class SpillableSubpartitionTest extends SubpartitionTestBase {
public void testConsumeSpillablePartitionSpilledDuringConsume() throws Exception {
SpillableSubpartition partition = createSubpartition();
- Buffer buffer = TestBufferFactory.createBuffer(4096, 4096);
+ Buffer buffer = TestBufferFactory.createBuffer(BUFFER_DATA_SIZE, BUFFER_DATA_SIZE);
buffer.retainBuffer();
buffer.retainBuffer();
partition.add(buffer);
partition.add(buffer);
+ {
+ Buffer event = TestBufferFactory.createBuffer(BUFFER_DATA_SIZE, BUFFER_DATA_SIZE);
+ event.tagAsEvent();
+ partition.add(event);
+ }
partition.add(buffer);
partition.finish();
- assertEquals(4, partition.getTotalNumberOfBuffers());
- assertEquals(4096 * 3 + 4, partition.getTotalNumberOfBytes());
- assertEquals(4, partition.getTotalNumberOfBuffers());
+ assertEquals(5, partition.getTotalNumberOfBuffers());
assertEquals(3, partition.getBuffersInBacklog());
- assertEquals(4096 * 3 + 4, partition.getTotalNumberOfBytes());
+ assertEquals(BUFFER_DATA_SIZE * 4 + 4, partition.getTotalNumberOfBytes());
AwaitableBufferAvailablityListener listener = new AwaitableBufferAvailablityListener();
SpillableSubpartitionView reader = (SpillableSubpartitionView) partition.createReadView(listener);
@@ -313,45 +330,65 @@ public class SpillableSubpartitionTest extends SubpartitionTestBase {
assertEquals(1, listener.getNumNotifiedBuffers());
assertFalse(buffer.isRecycled());
- BufferAndBacklog read = reader.getNextBuffer();
+ assertFalse(reader.nextBufferIsEvent());
+ BufferAndBacklog read = reader.getNextBuffer(); // first buffer (non-spilled)
assertNotNull(read);
+ assertTrue(read.buffer().isBuffer());
assertSame(buffer, read.buffer());
assertEquals(2, partition.getBuffersInBacklog());
assertEquals(partition.getBuffersInBacklog(), read.buffersInBacklog());
read.buffer().recycleBuffer();
assertEquals(2, listener.getNumNotifiedBuffers());
assertFalse(buffer.isRecycled());
+ assertFalse(read.nextBufferIsEvent());
// Spill now
- assertEquals(2, partition.releaseMemory());
+ assertEquals(3, partition.releaseMemory());
assertFalse(buffer.isRecycled()); // still one in the reader!
// still same statistics:
- assertEquals(4, partition.getTotalNumberOfBuffers());
+ assertEquals(5, partition.getTotalNumberOfBuffers());
assertEquals(2, partition.getBuffersInBacklog());
- assertEquals(4096 * 3 + 4, partition.getTotalNumberOfBytes());
+ assertEquals(BUFFER_DATA_SIZE * 4 + 4, partition.getTotalNumberOfBytes());
- listener.awaitNotifications(4, 30_000);
- assertEquals(4, listener.getNumNotifiedBuffers());
+ listener.awaitNotifications(5, 30_000);
+ assertEquals(5, listener.getNumNotifiedBuffers());
+ assertFalse(reader.nextBufferIsEvent()); // second buffer (retained in SpillableSubpartition#nextBuffer)
read = reader.getNextBuffer();
assertNotNull(read);
+ assertTrue(read.buffer().isBuffer());
assertEquals(1, partition.getBuffersInBacklog());
assertEquals(partition.getBuffersInBacklog(), read.buffersInBacklog());
assertSame(buffer, read.buffer());
read.buffer().recycleBuffer();
// now the buffer may be freed, depending on the timing of the write operation
// -> let's do this check at the end of the test (to save some time)
+ assertTrue(read.nextBufferIsEvent());
+ assertTrue(reader.nextBufferIsEvent()); // the event (spilled)
read = reader.getNextBuffer();
assertNotNull(read);
+ assertFalse(read.buffer().isBuffer());
+ assertEquals(1, partition.getBuffersInBacklog());
+ assertEquals(partition.getBuffersInBacklog(), read.buffersInBacklog());
+ assertNotSame(buffer, read.buffer());
+ read.buffer().recycleBuffer();
+ assertFalse(read.nextBufferIsEvent());
+
+ assertFalse(reader.nextBufferIsEvent()); // last buffer (spilled)
+ read = reader.getNextBuffer();
+ assertNotNull(read);
+ assertTrue(read.buffer().isBuffer());
assertEquals(0, partition.getBuffersInBacklog());
assertEquals(partition.getBuffersInBacklog(), read.buffersInBacklog());
assertNotSame(buffer, read.buffer());
assertFalse(read.buffer().isRecycled());
read.buffer().recycleBuffer();
assertTrue(read.buffer().isRecycled());
+ assertTrue(read.nextBufferIsEvent());
// End of partition
+ assertTrue(reader.nextBufferIsEvent());
read = reader.getNextBuffer();
assertNotNull(read);
assertEquals(0, partition.getBuffersInBacklog());
@@ -361,6 +398,7 @@ public class SpillableSubpartitionTest extends SubpartitionTestBase {
assertFalse(read.buffer().isRecycled());
read.buffer().recycleBuffer();
assertTrue(read.buffer().isRecycled());
+ assertFalse(read.nextBufferIsEvent());
// finally check that the buffer has been freed after a successful (or failed) write
final long deadline = System.currentTimeMillis() + 30_000L; // 30 secs
@@ -403,7 +441,7 @@ public class SpillableSubpartitionTest extends SubpartitionTestBase {
assertEquals(1, partition.getTotalNumberOfBuffers());
assertEquals(4, partition.getTotalNumberOfBytes());
- Buffer buffer = TestBufferFactory.createBuffer(4096, 4096);
+ Buffer buffer = TestBufferFactory.createBuffer(BUFFER_DATA_SIZE, BUFFER_DATA_SIZE);
try {
partition.add(buffer);
} finally {
@@ -447,7 +485,7 @@ public class SpillableSubpartitionTest extends SubpartitionTestBase {
assertEquals(0, partition.releaseMemory());
}
- Buffer buffer = TestBufferFactory.createBuffer(4096, 4096);
+ Buffer buffer = TestBufferFactory.createBuffer(BUFFER_DATA_SIZE, BUFFER_DATA_SIZE);
boolean bufferRecycled;
try {
partition.add(buffer);
@@ -475,7 +513,7 @@ public class SpillableSubpartitionTest extends SubpartitionTestBase {
SpillableSubpartition partition = createSubpartition(ioManager);
assertEquals(0, partition.releaseMemory());
- Buffer buffer = TestBufferFactory.createBuffer(4096, 4096);
+ Buffer buffer = TestBufferFactory.createBuffer(BUFFER_DATA_SIZE, BUFFER_DATA_SIZE);
boolean bufferRecycled;
try {
partition.add(buffer);
@@ -490,7 +528,7 @@ public class SpillableSubpartitionTest extends SubpartitionTestBase {
Assert.fail("buffer recycled before the write operation completed");
}
assertEquals(1, partition.getTotalNumberOfBuffers());
- assertEquals(4096, partition.getTotalNumberOfBytes());
+ assertEquals(BUFFER_DATA_SIZE, partition.getTotalNumberOfBytes());
}
/**
@@ -521,8 +559,8 @@ public class SpillableSubpartitionTest extends SubpartitionTestBase {
IOManager ioManager = new IOManagerAsyncWithNoOpBufferFileWriter();
SpillableSubpartition partition = createSubpartition(ioManager);
- Buffer buffer1 = TestBufferFactory.createBuffer(4096, 4096);
- Buffer buffer2 = TestBufferFactory.createBuffer(4096, 4096);
+ Buffer buffer1 = TestBufferFactory.createBuffer(BUFFER_DATA_SIZE, BUFFER_DATA_SIZE);
+ Buffer buffer2 = TestBufferFactory.createBuffer(BUFFER_DATA_SIZE, BUFFER_DATA_SIZE);
try {
// we need two buffers because the view will use one of them and not release it
partition.add(buffer1);
@@ -530,7 +568,7 @@ public class SpillableSubpartitionTest extends SubpartitionTestBase {
assertFalse("buffer1 should not be recycled (still in the queue)", buffer1.isRecycled());
assertFalse("buffer2 should not be recycled (still in the queue)", buffer2.isRecycled());
assertEquals(2, partition.getTotalNumberOfBuffers());
- assertEquals(4096 * 2, partition.getTotalNumberOfBytes());
+ assertEquals(BUFFER_DATA_SIZE * 2, partition.getTotalNumberOfBytes());
if (createView) {
// Create a read view
@@ -554,7 +592,7 @@ public class SpillableSubpartitionTest extends SubpartitionTestBase {
}
// note: a view requires a finished partition which has an additional EndOfPartitionEvent
assertEquals(2 + (createView ? 1 : 0), partition.getTotalNumberOfBuffers());
- assertEquals(4096 * 2 + (createView ? 4 : 0), partition.getTotalNumberOfBytes());
+ assertEquals(BUFFER_DATA_SIZE * 2 + (createView ? 4 : 0), partition.getTotalNumberOfBytes());
}
/**
@@ -569,7 +607,7 @@ public class SpillableSubpartitionTest extends SubpartitionTestBase {
exception.expect(IOException.class);
- Buffer buffer = TestBufferFactory.createBuffer(4096, 4096);
+ Buffer buffer = TestBufferFactory.createBuffer(BUFFER_DATA_SIZE, BUFFER_DATA_SIZE);
boolean bufferRecycled;
try {
partition.add(buffer);
@@ -635,8 +673,8 @@ public class SpillableSubpartitionTest extends SubpartitionTestBase {
private void testCleanupReleasedPartition(boolean spilled, boolean createView) throws Exception {
SpillableSubpartition partition = createSubpartition();
- Buffer buffer1 = TestBufferFactory.createBuffer(4096, 4096);
- Buffer buffer2 = TestBufferFactory.createBuffer(4096, 4096);
+ Buffer buffer1 = TestBufferFactory.createBuffer(BUFFER_DATA_SIZE, BUFFER_DATA_SIZE);
+ Buffer buffer2 = TestBufferFactory.createBuffer(BUFFER_DATA_SIZE, BUFFER_DATA_SIZE);
boolean buffer1Recycled;
boolean buffer2Recycled;
try {
@@ -681,22 +719,28 @@ public class SpillableSubpartitionTest extends SubpartitionTestBase {
}
// note: in case we create a view, there will be an additional EndOfPartitionEvent
assertEquals(createView ? 3 : 2, partition.getTotalNumberOfBuffers());
- assertEquals((createView ? 4 : 0) + 2 * 4096, partition.getTotalNumberOfBytes());
+ assertEquals((createView ? 4 : 0) + 2 * BUFFER_DATA_SIZE, partition.getTotalNumberOfBytes());
}
private static class AwaitableBufferAvailablityListener implements BufferAvailabilityListener {
private long numNotifiedBuffers;
+ private long numNotifications;
@Override
public void notifyBuffersAvailable(long numBuffers) {
numNotifiedBuffers += numBuffers;
+ ++numNotifications;
}
long getNumNotifiedBuffers() {
return numNotifiedBuffers;
}
+ public long getNumNotifications() {
+ return numNotifications;
+ }
+
void awaitNotifications(long awaitedNumNotifiedBuffers, long timeoutMillis) throws InterruptedException {
long deadline = System.currentTimeMillis() + timeoutMillis;
while (numNotifiedBuffers < awaitedNumNotifiedBuffers && System.currentTimeMillis() < deadline) {
http://git-wip-us.apache.org/repos/asf/flink/blob/a9fdbc4b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannelTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannelTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannelTest.java
index 8c8b96a..7c8ed18 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannelTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannelTest.java
@@ -865,7 +865,7 @@ public class RemoteInputChannelTest {
"InputGate",
new JobID(),
new IntermediateDataSetID(),
- ResultPartitionType.PIPELINED_CREDIT_BASED,
+ ResultPartitionType.PIPELINED,
0,
1,
mock(TaskActions.class),
http://git-wip-us.apache.org/repos/asf/flink/blob/a9fdbc4b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGateTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGateTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGateTest.java
index 364ac75..2a2b364 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGateTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGateTest.java
@@ -126,7 +126,7 @@ public class SingleInputGateTest {
final ResultSubpartitionView iterator = mock(ResultSubpartitionView.class);
when(iterator.getNextBuffer()).thenReturn(
- new BufferAndBacklog(new NetworkBuffer(MemorySegmentFactory.allocateUnpooledSegment(1024), FreeingBufferRecycler.INSTANCE), 0));
+ new BufferAndBacklog(new NetworkBuffer(MemorySegmentFactory.allocateUnpooledSegment(1024), FreeingBufferRecycler.INSTANCE), 0, false));
final ResultPartitionManager partitionManager = mock(ResultPartitionManager.class);
when(partitionManager.createSubpartitionView(
@@ -386,7 +386,7 @@ public class SingleInputGateTest {
"t1",
new JobID(),
new IntermediateDataSetID(),
- ResultPartitionType.PIPELINED_CREDIT_BASED,
+ ResultPartitionType.PIPELINED_BOUNDED,
0,
1,
mock(TaskActions.class),
@@ -414,7 +414,7 @@ public class SingleInputGateTest {
"t1",
new JobID(),
new IntermediateDataSetID(),
- ResultPartitionType.PIPELINED_CREDIT_BASED,
+ ResultPartitionType.PIPELINED_BOUNDED,
0,
1,
mock(TaskActions.class),
http://git-wip-us.apache.org/repos/asf/flink/blob/a9fdbc4b/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskCancelAsyncProducerConsumerITCase.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskCancelAsyncProducerConsumerITCase.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskCancelAsyncProducerConsumerITCase.java
index 8cc936f..73553f5 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskCancelAsyncProducerConsumerITCase.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskCancelAsyncProducerConsumerITCase.java
@@ -84,7 +84,7 @@ public class TaskCancelAsyncProducerConsumerITCase extends TestLogger {
config.setInteger(ConfigConstants.LOCAL_NUMBER_TASK_MANAGER, 1);
config.setInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, 1);
config.setInteger(TaskManagerOptions.MEMORY_SEGMENT_SIZE, 4096);
- config.setInteger(TaskManagerOptions.NETWORK_NUM_BUFFERS, 8);
+ config.setInteger(TaskManagerOptions.NETWORK_NUM_BUFFERS, 9);
flink = new TestingCluster(config, true);
flink.start();
http://git-wip-us.apache.org/repos/asf/flink/blob/a9fdbc4b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/benchmark/StreamNetworkBenchmarkEnvironment.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/benchmark/StreamNetworkBenchmarkEnvironment.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/benchmark/StreamNetworkBenchmarkEnvironment.java
index 83508ea..1737a3c 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/benchmark/StreamNetworkBenchmarkEnvironment.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/benchmark/StreamNetworkBenchmarkEnvironment.java
@@ -35,7 +35,6 @@ import org.apache.flink.runtime.io.network.NetworkEnvironment;
import org.apache.flink.runtime.io.network.TaskEventDispatcher;
import org.apache.flink.runtime.io.network.api.writer.ResultPartitionWriter;
import org.apache.flink.runtime.io.network.api.writer.RoundRobinChannelSelector;
-import org.apache.flink.runtime.io.network.buffer.BufferPool;
import org.apache.flink.runtime.io.network.buffer.NetworkBufferPool;
import org.apache.flink.runtime.io.network.netty.NettyConfig;
import org.apache.flink.runtime.io.network.netty.NettyConnectionManager;
@@ -187,15 +186,7 @@ public class StreamNetworkBenchmarkEnvironment<T extends IOReadableWritable> {
ioManager,
false);
- // similar to NetworkEnvironment#registerTask()
- int numBuffers = resultPartition.getNumberOfSubpartitions() *
- TaskManagerOptions.NETWORK_BUFFERS_PER_CHANNEL.defaultValue() +
- TaskManagerOptions.NETWORK_EXTRA_BUFFERS_PER_GATE.defaultValue();
-
- BufferPool bufferPool = environment.getNetworkBufferPool().createBufferPool(channels, numBuffers);
- resultPartition.registerBufferPool(bufferPool);
-
- environment.getResultPartitionManager().registerResultPartition(resultPartition);
+ environment.setupPartition(resultPartition);
return resultPartition;
}
@@ -232,15 +223,7 @@ public class StreamNetworkBenchmarkEnvironment<T extends IOReadableWritable> {
new NoOpTaskActions(),
UnregisteredMetricGroups.createUnregisteredTaskMetricGroup().getIOMetricGroup());
- // similar to NetworkEnvironment#registerTask()
- int numBuffers = gate.getNumberOfInputChannels() *
- TaskManagerOptions.NETWORK_BUFFERS_PER_CHANNEL.defaultValue() +
- TaskManagerOptions.NETWORK_EXTRA_BUFFERS_PER_GATE.defaultValue();
-
- BufferPool bufferPool =
- environment.getNetworkBufferPool().createBufferPool(gate.getNumberOfInputChannels(), numBuffers);
-
- gate.setBufferPool(bufferPool);
+ environment.setupInputGate(gate);
gates[channel] = gate;
}
http://git-wip-us.apache.org/repos/asf/flink/blob/a9fdbc4b/flink-tests/src/test/java/org/apache/flink/test/misc/SuccessAfterNetworkBuffersFailureITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/misc/SuccessAfterNetworkBuffersFailureITCase.java b/flink-tests/src/test/java/org/apache/flink/test/misc/SuccessAfterNetworkBuffersFailureITCase.java
index 8f4bc8c..1210e85 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/misc/SuccessAfterNetworkBuffersFailureITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/misc/SuccessAfterNetworkBuffersFailureITCase.java
@@ -59,7 +59,7 @@ public class SuccessAfterNetworkBuffersFailureITCase extends TestLogger {
config.setInteger(ConfigConstants.LOCAL_NUMBER_TASK_MANAGER, 2);
config.setLong(TaskManagerOptions.MANAGED_MEMORY_SIZE, 80L);
config.setInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, 8);
- config.setInteger(TaskManagerOptions.NETWORK_NUM_BUFFERS, 800);
+ config.setInteger(TaskManagerOptions.NETWORK_NUM_BUFFERS, 1024);
cluster = new LocalFlinkMiniCluster(config, false);