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:58 UTC
[5/5] flink git commit: [FLINK-7456][network] Implement Netty sender
incoming pipeline for credit-based
[FLINK-7456][network] Implement Netty sender incoming pipeline for credit-based
This closes #4552.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/a9fdbc4b
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/a9fdbc4b
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/a9fdbc4b
Branch: refs/heads/master
Commit: a9fdbc4be2a7b2c10d524f92927534a15edf9b3e
Parents: 892eb85
Author: Zhijiang <wa...@aliyun.com>
Authored: Sat Sep 30 14:36:19 2017 +0800
Committer: Stefan Richter <s....@data-artisans.com>
Committed: Fri Feb 16 13:55:36 2018 +0100
----------------------------------------------------------------------
.../runtime/io/network/NetworkEnvironment.java | 101 ++---
.../network/netty/CreditBasedClientHandler.java | 386 ------------------
.../runtime/io/network/netty/NettyMessage.java | 2 +-
.../network/netty/PartitionRequestClient.java | 6 +-
.../netty/PartitionRequestClientHandler.java | 396 +++++++------------
.../io/network/netty/PartitionRequestQueue.java | 195 ++++++---
.../netty/PartitionRequestServerHandler.java | 8 +
.../netty/SequenceNumberingViewReader.java | 94 ++++-
.../partition/PipelinedSubpartition.java | 14 +-
.../partition/PipelinedSubpartitionView.java | 5 +
.../network/partition/ResultPartitionType.java | 27 +-
.../network/partition/ResultSubpartition.java | 8 +-
.../partition/ResultSubpartitionView.java | 5 +
.../partition/SpillableSubpartitionView.java | 37 +-
.../partition/SpilledSubpartitionView.java | 53 ++-
.../partition/consumer/RemoteInputChannel.java | 2 +-
.../partition/consumer/SingleInputGate.java | 15 +-
.../io/network/NetworkEnvironmentTest.java | 9 +-
.../netty/CancelPartitionRequestTest.java | 12 +-
.../PartitionRequestClientHandlerTest.java | 144 ++++---
.../netty/PartitionRequestClientTest.java | 150 +++++++
.../netty/PartitionRequestQueueTest.java | 207 ++++++++--
.../netty/ServerTransportErrorHandlingTest.java | 2 +-
.../partition/PipelinedSubpartitionTest.java | 72 +++-
.../partition/SpillableSubpartitionTest.java | 142 ++++---
.../consumer/RemoteInputChannelTest.java | 2 +-
.../partition/consumer/SingleInputGateTest.java | 6 +-
.../TaskCancelAsyncProducerConsumerITCase.java | 2 +-
.../StreamNetworkBenchmarkEnvironment.java | 21 +-
...SuccessAfterNetworkBuffersFailureITCase.java | 2 +-
30 files changed, 1133 insertions(+), 992 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/a9fdbc4b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/NetworkEnvironment.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/NetworkEnvironment.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/NetworkEnvironment.java
index 6c01350..2441667 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/NetworkEnvironment.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/NetworkEnvironment.java
@@ -18,6 +18,7 @@
package org.apache.flink.runtime.io.network;
+import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.api.common.JobID;
import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
import org.apache.flink.runtime.io.disk.iomanager.IOManager;
@@ -179,65 +180,65 @@ public class NetworkEnvironment {
throw new IllegalStateException("NetworkEnvironment is shut down");
}
- for (int i = 0; i < producedPartitions.length; i++) {
- final ResultPartition partition = producedPartitions[i];
-
- // Buffer pool for the partition
- BufferPool bufferPool = null;
-
- try {
- int maxNumberOfMemorySegments = partition.getPartitionType().isBounded() ?
- partition.getNumberOfSubpartitions() * networkBuffersPerChannel +
- extraNetworkBuffersPerGate : Integer.MAX_VALUE;
- bufferPool = networkBufferPool.createBufferPool(partition.getNumberOfSubpartitions(),
- maxNumberOfMemorySegments);
- partition.registerBufferPool(bufferPool);
-
- resultPartitionManager.registerResultPartition(partition);
- } catch (Throwable t) {
- if (bufferPool != null) {
- bufferPool.lazyDestroy();
- }
-
- if (t instanceof IOException) {
- throw (IOException) t;
- } else {
- throw new IOException(t.getMessage(), t);
- }
- }
-
- // Register writer with task event dispatcher
- taskEventDispatcher.registerPartition(partition.getPartitionId());
+ for (final ResultPartition partition : producedPartitions) {
+ setupPartition(partition);
}
// Setup the buffer pool for each buffer reader
final SingleInputGate[] inputGates = task.getAllInputGates();
-
for (SingleInputGate gate : inputGates) {
- BufferPool bufferPool = null;
+ setupInputGate(gate);
+ }
+ }
+ }
- try {
- if (gate.getConsumedPartitionType().isCreditBased()) {
- // Create a fixed-size buffer pool for floating buffers and assign exclusive buffers to input channels directly
- bufferPool = networkBufferPool.createBufferPool(extraNetworkBuffersPerGate, extraNetworkBuffersPerGate);
- gate.assignExclusiveSegments(networkBufferPool, networkBuffersPerChannel);
- } else {
- int maxNumberOfMemorySegments = gate.getConsumedPartitionType().isBounded() ?
- gate.getNumberOfInputChannels() * networkBuffersPerChannel +
- extraNetworkBuffersPerGate : Integer.MAX_VALUE;
- bufferPool = networkBufferPool.createBufferPool(gate.getNumberOfInputChannels(),
- maxNumberOfMemorySegments);
- }
- gate.setBufferPool(bufferPool);
- } catch (Throwable t) {
- if (bufferPool != null) {
- bufferPool.lazyDestroy();
- }
+ @VisibleForTesting
+ public void setupPartition(ResultPartition partition) throws IOException {
+ BufferPool bufferPool = null;
+
+ try {
+ int maxNumberOfMemorySegments = partition.getPartitionType().isBounded() ?
+ partition.getNumberOfSubpartitions() * networkBuffersPerChannel +
+ extraNetworkBuffersPerGate : Integer.MAX_VALUE;
+ bufferPool = networkBufferPool.createBufferPool(partition.getNumberOfSubpartitions(),
+ maxNumberOfMemorySegments);
+ partition.registerBufferPool(bufferPool);
+
+ resultPartitionManager.registerResultPartition(partition);
+ } catch (Throwable t) {
+ if (bufferPool != null) {
+ bufferPool.lazyDestroy();
+ }
- ExceptionUtils.rethrowIOException(t);
- }
+ if (t instanceof IOException) {
+ throw (IOException) t;
+ } else {
+ throw new IOException(t.getMessage(), t);
}
}
+
+ taskEventDispatcher.registerPartition(partition.getPartitionId());
+ }
+
+ @VisibleForTesting
+ public void setupInputGate(SingleInputGate gate) throws IOException {
+ BufferPool bufferPool = null;
+
+ try {
+ int maxNumberOfMemorySegments = gate.getConsumedPartitionType().isBounded() ?
+ extraNetworkBuffersPerGate : Integer.MAX_VALUE;
+ // Create a buffer pool for floating buffers and assign exclusive buffers to input channels directly
+ bufferPool = networkBufferPool.createBufferPool(extraNetworkBuffersPerGate,
+ maxNumberOfMemorySegments);
+ gate.assignExclusiveSegments(networkBufferPool, networkBuffersPerChannel);
+ gate.setBufferPool(bufferPool);
+ } catch (Throwable t) {
+ if (bufferPool != null) {
+ bufferPool.lazyDestroy();
+ }
+
+ ExceptionUtils.rethrowIOException(t);
+ }
}
public void unregisterTask(Task task) {
http://git-wip-us.apache.org/repos/asf/flink/blob/a9fdbc4b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/CreditBasedClientHandler.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/CreditBasedClientHandler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/CreditBasedClientHandler.java
deleted file mode 100644
index 8f171b6..0000000
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/CreditBasedClientHandler.java
+++ /dev/null
@@ -1,386 +0,0 @@
-/*
- * 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.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.FreeingBufferRecycler;
-import org.apache.flink.runtime.io.network.buffer.NetworkBuffer;
-import org.apache.flink.runtime.io.network.netty.exception.LocalTransportException;
-import org.apache.flink.runtime.io.network.netty.exception.RemoteTransportException;
-import org.apache.flink.runtime.io.network.netty.exception.TransportException;
-import org.apache.flink.runtime.io.network.netty.NettyMessage.AddCredit;
-import org.apache.flink.runtime.io.network.partition.PartitionNotFoundException;
-import org.apache.flink.runtime.io.network.partition.consumer.InputChannelID;
-import org.apache.flink.runtime.io.network.partition.consumer.RemoteInputChannel;
-
-import org.apache.flink.shaded.netty4.io.netty.buffer.ByteBuf;
-import org.apache.flink.shaded.netty4.io.netty.channel.Channel;
-import org.apache.flink.shaded.netty4.io.netty.channel.ChannelFuture;
-import org.apache.flink.shaded.netty4.io.netty.channel.ChannelFutureListener;
-import org.apache.flink.shaded.netty4.io.netty.channel.ChannelHandlerContext;
-import org.apache.flink.shaded.netty4.io.netty.channel.ChannelInboundHandlerAdapter;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.IOException;
-import java.net.SocketAddress;
-import java.util.ArrayDeque;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentMap;
-import java.util.concurrent.atomic.AtomicReference;
-
-/**
- * Channel handler to read the messages of buffer response or error response from the
- * producer, to write and flush the unannounced credits for the producer.
- */
-class CreditBasedClientHandler extends ChannelInboundHandlerAdapter {
-
- private static final Logger LOG = LoggerFactory.getLogger(CreditBasedClientHandler.class);
-
- /** Channels, which already requested partitions from the producers. */
- private final ConcurrentMap<InputChannelID, RemoteInputChannel> inputChannels = new ConcurrentHashMap<>();
-
- /** Channels, which will notify the producers about unannounced credit. */
- private final ArrayDeque<RemoteInputChannel> inputChannelsWithCredit = new ArrayDeque<>();
-
- private final AtomicReference<Throwable> channelError = new AtomicReference<>();
-
- private final ChannelFutureListener writeListener = new WriteAndFlushNextMessageIfPossibleListener();
-
- /**
- * Set of cancelled partition requests. A request is cancelled iff an input channel is cleared
- * while data is still coming in for this channel.
- */
- private final ConcurrentMap<InputChannelID, InputChannelID> cancelled = new ConcurrentHashMap<>();
-
- /**
- * The channel handler context is initialized in channel active event by netty thread, the context may also
- * be accessed by task thread or canceler thread to cancel partition request during releasing resources.
- */
- private volatile ChannelHandlerContext ctx;
-
- // ------------------------------------------------------------------------
- // Input channel/receiver registration
- // ------------------------------------------------------------------------
-
- void addInputChannel(RemoteInputChannel listener) throws IOException {
- checkError();
-
- if (!inputChannels.containsKey(listener.getInputChannelId())) {
- inputChannels.put(listener.getInputChannelId(), listener);
- }
- }
-
- void removeInputChannel(RemoteInputChannel listener) {
- inputChannels.remove(listener.getInputChannelId());
- }
-
- void cancelRequestFor(InputChannelID inputChannelId) {
- if (inputChannelId == null || ctx == null) {
- return;
- }
-
- if (cancelled.putIfAbsent(inputChannelId, inputChannelId) == null) {
- ctx.writeAndFlush(new NettyMessage.CancelPartitionRequest(inputChannelId));
- }
- }
-
- /**
- * The credit begins to announce after receiving the sender's backlog from buffer response.
- * Than means it should only happen after some interactions with the channel to make sure
- * the context will not be null.
- *
- * @param inputChannel The input channel with unannounced credits.
- */
- void notifyCreditAvailable(final RemoteInputChannel inputChannel) {
- ctx.executor().execute(new Runnable() {
- @Override
- public void run() {
- ctx.pipeline().fireUserEventTriggered(inputChannel);
- }
- });
- }
-
- // ------------------------------------------------------------------------
- // Network events
- // ------------------------------------------------------------------------
-
- @Override
- public void channelActive(final ChannelHandlerContext ctx) throws Exception {
- if (this.ctx == null) {
- this.ctx = ctx;
- }
-
- super.channelActive(ctx);
- }
-
- @Override
- public void channelInactive(ChannelHandlerContext ctx) throws Exception {
- // Unexpected close. In normal operation, the client closes the connection after all input
- // channels have been removed. This indicates a problem with the remote task manager.
- if (!inputChannels.isEmpty()) {
- final SocketAddress remoteAddr = ctx.channel().remoteAddress();
-
- notifyAllChannelsOfErrorAndClose(new RemoteTransportException(
- "Connection unexpectedly closed by remote task manager '" + remoteAddr + "'. "
- + "This might indicate that the remote task manager was lost.", remoteAddr));
- }
-
- super.channelInactive(ctx);
- }
-
- /**
- * Called on exceptions in the client handler pipeline.
- *
- * <p>Remote exceptions are received as regular payload.
- */
- @Override
- public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
- if (cause instanceof TransportException) {
- notifyAllChannelsOfErrorAndClose(cause);
- } else {
- final SocketAddress remoteAddr = ctx.channel().remoteAddress();
-
- final TransportException tex;
-
- // Improve on the connection reset by peer error message
- if (cause instanceof IOException && cause.getMessage().equals("Connection reset by peer")) {
- tex = new RemoteTransportException("Lost connection to task manager '" + remoteAddr + "'. " +
- "This indicates that the remote task manager was lost.", remoteAddr, cause);
- } else {
- tex = new LocalTransportException(cause.getMessage(), ctx.channel().localAddress(), cause);
- }
-
- notifyAllChannelsOfErrorAndClose(tex);
- }
- }
-
- @Override
- public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
- try {
- decodeMsg(msg);
- } catch (Throwable t) {
- notifyAllChannelsOfErrorAndClose(t);
- }
- }
-
- @Override
- public void userEventTriggered(ChannelHandlerContext ctx, Object msg) throws Exception {
- if (msg instanceof RemoteInputChannel) {
- // Queue an input channel for available credits announcement.
- // If the queue is empty, we try to trigger the actual write. Otherwise
- // this will be handled by the writeAndFlushNextMessageIfPossible calls.
- boolean triggerWrite = inputChannelsWithCredit.isEmpty();
-
- inputChannelsWithCredit.add((RemoteInputChannel) msg);
-
- if (triggerWrite) {
- writeAndFlushNextMessageIfPossible(ctx.channel());
- }
- } else {
- ctx.fireUserEventTriggered(msg);
- }
- }
-
- @Override
- public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception {
- writeAndFlushNextMessageIfPossible(ctx.channel());
- }
-
- private void notifyAllChannelsOfErrorAndClose(Throwable cause) {
- if (channelError.compareAndSet(null, cause)) {
- try {
- for (RemoteInputChannel inputChannel : inputChannels.values()) {
- inputChannel.onError(cause);
- }
- } catch (Throwable t) {
- // We can only swallow the Exception at this point. :(
- LOG.warn("An Exception was thrown during error notification of a remote input channel.", t);
- } finally {
- inputChannels.clear();
- inputChannelsWithCredit.clear();
-
- if (ctx != null) {
- ctx.close();
- }
- }
- }
- }
-
- // ------------------------------------------------------------------------
-
- /**
- * Checks for an error and rethrows it if one was reported.
- */
- private void checkError() throws IOException {
- final Throwable t = channelError.get();
-
- if (t != null) {
- if (t instanceof IOException) {
- throw (IOException) t;
- } else {
- throw new IOException("There has been an error in the channel.", t);
- }
- }
- }
-
- private void decodeMsg(Object msg) throws Throwable {
- final Class<?> msgClazz = msg.getClass();
-
- // ---- Buffer --------------------------------------------------------
- if (msgClazz == NettyMessage.BufferResponse.class) {
- NettyMessage.BufferResponse bufferOrEvent = (NettyMessage.BufferResponse) msg;
-
- RemoteInputChannel inputChannel = inputChannels.get(bufferOrEvent.receiverId);
- if (inputChannel == null) {
- bufferOrEvent.releaseBuffer();
-
- cancelRequestFor(bufferOrEvent.receiverId);
-
- return;
- }
-
- decodeBufferOrEvent(inputChannel, bufferOrEvent);
-
- } else if (msgClazz == NettyMessage.ErrorResponse.class) {
- // ---- Error ---------------------------------------------------------
- NettyMessage.ErrorResponse error = (NettyMessage.ErrorResponse) msg;
-
- SocketAddress remoteAddr = ctx.channel().remoteAddress();
-
- if (error.isFatalError()) {
- notifyAllChannelsOfErrorAndClose(new RemoteTransportException(
- "Fatal error at remote task manager '" + remoteAddr + "'.",
- remoteAddr,
- error.cause));
- } else {
- RemoteInputChannel inputChannel = inputChannels.get(error.receiverId);
-
- if (inputChannel != null) {
- if (error.cause.getClass() == PartitionNotFoundException.class) {
- inputChannel.onFailedPartitionRequest();
- } else {
- inputChannel.onError(new RemoteTransportException(
- "Error at remote task manager '" + remoteAddr + "'.",
- remoteAddr,
- error.cause));
- }
- }
- }
- } else {
- throw new IllegalStateException("Received unknown message from producer: " + msg.getClass());
- }
- }
-
- private void decodeBufferOrEvent(RemoteInputChannel inputChannel, NettyMessage.BufferResponse bufferOrEvent) throws Throwable {
- try {
- ByteBuf nettyBuffer = bufferOrEvent.getNettyBuffer();
- final int size = nettyBuffer.readableBytes();
- if (bufferOrEvent.isBuffer()) {
- // ---- Buffer ------------------------------------------------
-
- // Early return for empty buffers. Otherwise Netty's readBytes() throws an
- // IndexOutOfBoundsException.
- if (size == 0) {
- inputChannel.onEmptyBuffer(bufferOrEvent.sequenceNumber, bufferOrEvent.backlog);
- return;
- }
-
- Buffer buffer = inputChannel.requestBuffer();
- if (buffer != null) {
- nettyBuffer.readBytes(buffer.asByteBuf(), size);
-
- inputChannel.onBuffer(buffer, bufferOrEvent.sequenceNumber, bufferOrEvent.backlog);
- } else if (inputChannel.isReleased()) {
- cancelRequestFor(bufferOrEvent.receiverId);
- } else {
- throw new IllegalStateException("No buffer available in credit-based input channel.");
- }
- } else {
- // ---- Event -------------------------------------------------
- // TODO We can just keep the serialized data in the Netty buffer and release it later at the reader
- byte[] byteArray = new byte[size];
- nettyBuffer.readBytes(byteArray);
-
- MemorySegment memSeg = MemorySegmentFactory.wrap(byteArray);
- Buffer buffer = new NetworkBuffer(memSeg, FreeingBufferRecycler.INSTANCE, false);
-
- inputChannel.onBuffer(buffer, bufferOrEvent.sequenceNumber, bufferOrEvent.backlog);
- }
- } finally {
- bufferOrEvent.releaseBuffer();
- }
- }
-
- /**
- * Fetches one un-released input channel from the queue and writes the
- * unannounced credits immediately. After this is done, we will continue
- * with the next input channel via listener's callback.
- */
- private void writeAndFlushNextMessageIfPossible(Channel channel) {
- if (channelError.get() != null || !channel.isWritable()) {
- return;
- }
-
- while (true) {
- RemoteInputChannel inputChannel = inputChannelsWithCredit.poll();
-
- // The input channel may be null because of the write callbacks
- // that are executed after each write.
- if (inputChannel == null) {
- return;
- }
-
- //It is no need to notify credit for the released channel.
- if (!inputChannel.isReleased()) {
- AddCredit msg = new AddCredit(
- inputChannel.getPartitionId(),
- inputChannel.getAndResetUnannouncedCredit(),
- inputChannel.getInputChannelId());
-
- // Write and flush and wait until this is done before
- // trying to continue with the next input channel.
- channel.writeAndFlush(msg).addListener(writeListener);
-
- return;
- }
- }
- }
-
- private class WriteAndFlushNextMessageIfPossibleListener implements ChannelFutureListener {
-
- @Override
- public void operationComplete(ChannelFuture future) throws Exception {
- try {
- if (future.isSuccess()) {
- writeAndFlushNextMessageIfPossible(future.channel());
- } else if (future.cause() != null) {
- notifyAllChannelsOfErrorAndClose(future.cause());
- } else {
- notifyAllChannelsOfErrorAndClose(new IllegalStateException("Sending cancelled by user."));
- }
- } catch (Throwable t) {
- notifyAllChannelsOfErrorAndClose(t);
- }
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/a9fdbc4b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/NettyMessage.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/NettyMessage.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/NettyMessage.java
index 696c74e..9f9ee88 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/NettyMessage.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/NettyMessage.java
@@ -485,7 +485,7 @@ public abstract class NettyMessage {
@Override
public String toString() {
- return String.format("PartitionRequest(%s:%d)", partitionId, queueIndex);
+ return String.format("PartitionRequest(%s:%d:%d)", partitionId, queueIndex, credit);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/a9fdbc4b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestClient.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestClient.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestClient.java
index 12a9531..3f41646 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestClient.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestClient.java
@@ -41,8 +41,8 @@ import static org.apache.flink.util.Preconditions.checkNotNull;
/**
* Partition request client for remote partition requests.
- * <p>
- * This client is shared by all remote input channels, which request a partition
+ *
+ * <p>This client is shared by all remote input channels, which request a partition
* from the same {@link ConnectionID}.
*/
public class PartitionRequestClient {
@@ -57,7 +57,7 @@ public class PartitionRequestClient {
private final PartitionRequestClientFactory clientFactory;
- // If zero, the underlying TCP channel can be safely closed
+ /** If zero, the underlying TCP channel can be safely closed. */
private final AtomicDisposableReferenceCounter closeReferenceCounter = new AtomicDisposableReferenceCounter();
PartitionRequestClient(
http://git-wip-us.apache.org/repos/asf/flink/blob/a9fdbc4b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestClientHandler.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestClientHandler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestClientHandler.java
index 9468729..5299e22 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestClientHandler.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestClientHandler.java
@@ -21,10 +21,9 @@ package org.apache.flink.runtime.io.network.netty;
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.BufferListener;
-import org.apache.flink.runtime.io.network.buffer.BufferProvider;
import org.apache.flink.runtime.io.network.buffer.FreeingBufferRecycler;
import org.apache.flink.runtime.io.network.buffer.NetworkBuffer;
+import org.apache.flink.runtime.io.network.netty.NettyMessage.AddCredit;
import org.apache.flink.runtime.io.network.netty.exception.LocalTransportException;
import org.apache.flink.runtime.io.network.netty.exception.RemoteTransportException;
import org.apache.flink.runtime.io.network.netty.exception.TransportException;
@@ -32,8 +31,10 @@ import org.apache.flink.runtime.io.network.partition.PartitionNotFoundException;
import org.apache.flink.runtime.io.network.partition.consumer.InputChannelID;
import org.apache.flink.runtime.io.network.partition.consumer.RemoteInputChannel;
-import org.apache.flink.shaded.guava18.com.google.common.collect.Maps;
import org.apache.flink.shaded.netty4.io.netty.buffer.ByteBuf;
+import org.apache.flink.shaded.netty4.io.netty.channel.Channel;
+import org.apache.flink.shaded.netty4.io.netty.channel.ChannelFuture;
+import org.apache.flink.shaded.netty4.io.netty.channel.ChannelFutureListener;
import org.apache.flink.shaded.netty4.io.netty.channel.ChannelHandlerContext;
import org.apache.flink.shaded.netty4.io.netty.channel.ChannelInboundHandlerAdapter;
@@ -43,31 +44,38 @@ import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.net.SocketAddress;
import java.util.ArrayDeque;
-import java.util.Queue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicReference;
+/**
+ * Channel handler to read the messages of buffer response or error response from the
+ * producer, to write and flush the unannounced credits for the producer.
+ */
class PartitionRequestClientHandler extends ChannelInboundHandlerAdapter {
private static final Logger LOG = LoggerFactory.getLogger(PartitionRequestClientHandler.class);
- private final ConcurrentMap<InputChannelID, RemoteInputChannel> inputChannels = new ConcurrentHashMap<InputChannelID, RemoteInputChannel>();
-
- private final AtomicReference<Throwable> channelError = new AtomicReference<Throwable>();
+ /** Channels, which already requested partitions from the producers. */
+ private final ConcurrentMap<InputChannelID, RemoteInputChannel> inputChannels = new ConcurrentHashMap<>();
- private final BufferListenerTask bufferListener = new BufferListenerTask();
+ /** Channels, which will notify the producers about unannounced credit. */
+ private final ArrayDeque<RemoteInputChannel> inputChannelsWithCredit = new ArrayDeque<>();
- private final Queue<Object> stagedMessages = new ArrayDeque<Object>();
+ private final AtomicReference<Throwable> channelError = new AtomicReference<>();
- private final StagedMessagesHandlerTask stagedMessagesHandler = new StagedMessagesHandlerTask();
+ private final ChannelFutureListener writeListener = new WriteAndFlushNextMessageIfPossibleListener();
/**
* Set of cancelled partition requests. A request is cancelled iff an input channel is cleared
* while data is still coming in for this channel.
*/
- private final ConcurrentMap<InputChannelID, InputChannelID> cancelled = Maps.newConcurrentMap();
+ private final ConcurrentMap<InputChannelID, InputChannelID> cancelled = new ConcurrentHashMap<>();
+ /**
+ * The channel handler context is initialized in channel active event by netty thread, the context may also
+ * be accessed by task thread or canceler thread to cancel partition request during releasing resources.
+ */
private volatile ChannelHandlerContext ctx;
// ------------------------------------------------------------------------
@@ -96,6 +104,22 @@ class PartitionRequestClientHandler extends ChannelInboundHandlerAdapter {
}
}
+ /**
+ * The credit begins to announce after receiving the sender's backlog from buffer response.
+ * Than means it should only happen after some interactions with the channel to make sure
+ * the context will not be null.
+ *
+ * @param inputChannel The input channel with unannounced credits.
+ */
+ void notifyCreditAvailable(final RemoteInputChannel inputChannel) {
+ ctx.executor().execute(new Runnable() {
+ @Override
+ public void run() {
+ ctx.pipeline().fireUserEventTriggered(inputChannel);
+ }
+ });
+ }
+
// ------------------------------------------------------------------------
// Network events
// ------------------------------------------------------------------------
@@ -117,9 +141,8 @@ class PartitionRequestClientHandler extends ChannelInboundHandlerAdapter {
final SocketAddress remoteAddr = ctx.channel().remoteAddress();
notifyAllChannelsOfErrorAndClose(new RemoteTransportException(
- "Connection unexpectedly closed by remote task manager '" + remoteAddr + "'. "
- + "This might indicate that the remote task manager was lost.",
- remoteAddr));
+ "Connection unexpectedly closed by remote task manager '" + remoteAddr + "'. "
+ + "This might indicate that the remote task manager was lost.", remoteAddr));
}
super.channelInactive(ctx);
@@ -128,28 +151,22 @@ class PartitionRequestClientHandler extends ChannelInboundHandlerAdapter {
/**
* Called on exceptions in the client handler pipeline.
*
- * <p> Remote exceptions are received as regular payload.
+ * <p>Remote exceptions are received as regular payload.
*/
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
-
if (cause instanceof TransportException) {
notifyAllChannelsOfErrorAndClose(cause);
- }
- else {
+ } else {
final SocketAddress remoteAddr = ctx.channel().remoteAddress();
final TransportException tex;
// Improve on the connection reset by peer error message
- if (cause instanceof IOException
- && cause.getMessage().equals("Connection reset by peer")) {
-
- tex = new RemoteTransportException(
- "Lost connection to task manager '" + remoteAddr + "'. This indicates "
- + "that the remote task manager was lost.", remoteAddr, cause);
- }
- else {
+ if (cause instanceof IOException && cause.getMessage().equals("Connection reset by peer")) {
+ tex = new RemoteTransportException("Lost connection to task manager '" + remoteAddr + "'. " +
+ "This indicates that the remote task manager was lost.", remoteAddr, cause);
+ } else {
tex = new LocalTransportException(cause.getMessage(), ctx.channel().localAddress(), cause);
}
@@ -160,32 +177,50 @@ class PartitionRequestClientHandler extends ChannelInboundHandlerAdapter {
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
try {
- if (!bufferListener.hasStagedBufferOrEvent() && stagedMessages.isEmpty()) {
- decodeMsg(msg, false);
- }
- else {
- stagedMessages.add(msg);
- }
- }
- catch (Throwable t) {
+ decodeMsg(msg);
+ } catch (Throwable t) {
notifyAllChannelsOfErrorAndClose(t);
}
}
+ /**
+ * Triggered by notifying credit available in the client handler pipeline.
+ *
+ * <p>Enqueues the input channel and will trigger write&flush unannounced credits
+ * for this input channel if it is the first one in the queue.
+ */
+ @Override
+ public void userEventTriggered(ChannelHandlerContext ctx, Object msg) throws Exception {
+ if (msg instanceof RemoteInputChannel) {
+ boolean triggerWrite = inputChannelsWithCredit.isEmpty();
+
+ inputChannelsWithCredit.add((RemoteInputChannel) msg);
+
+ if (triggerWrite) {
+ writeAndFlushNextMessageIfPossible(ctx.channel());
+ }
+ } else {
+ ctx.fireUserEventTriggered(msg);
+ }
+ }
+
+ @Override
+ public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception {
+ writeAndFlushNextMessageIfPossible(ctx.channel());
+ }
+
private void notifyAllChannelsOfErrorAndClose(Throwable cause) {
if (channelError.compareAndSet(null, cause)) {
try {
for (RemoteInputChannel inputChannel : inputChannels.values()) {
inputChannel.onError(cause);
}
- }
- catch (Throwable t) {
+ } catch (Throwable t) {
// We can only swallow the Exception at this point. :(
- LOG.warn("An Exception was thrown during error notification of a "
- + "remote input channel.", t);
- }
- finally {
+ LOG.warn("An Exception was thrown during error notification of a remote input channel.", t);
+ } finally {
inputChannels.clear();
+ inputChannelsWithCredit.clear();
if (ctx != null) {
ctx.close();
@@ -205,19 +240,13 @@ class PartitionRequestClientHandler extends ChannelInboundHandlerAdapter {
if (t != null) {
if (t instanceof IOException) {
throw (IOException) t;
- }
- else {
+ } else {
throw new IOException("There has been an error in the channel.", t);
}
}
}
- @Override
- public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
- super.channelReadComplete(ctx);
- }
-
- private boolean decodeMsg(Object msg, boolean isStagedBuffer) throws Throwable {
+ private void decodeMsg(Object msg) throws Throwable {
final Class<?> msgClazz = msg.getClass();
// ---- Buffer --------------------------------------------------------
@@ -230,90 +259,66 @@ class PartitionRequestClientHandler extends ChannelInboundHandlerAdapter {
cancelRequestFor(bufferOrEvent.receiverId);
- return true;
+ return;
}
- return decodeBufferOrEvent(inputChannel, bufferOrEvent, isStagedBuffer);
- }
- // ---- Error ---------------------------------------------------------
- else if (msgClazz == NettyMessage.ErrorResponse.class) {
+ decodeBufferOrEvent(inputChannel, bufferOrEvent);
+
+ } else if (msgClazz == NettyMessage.ErrorResponse.class) {
+ // ---- Error ---------------------------------------------------------
NettyMessage.ErrorResponse error = (NettyMessage.ErrorResponse) msg;
SocketAddress remoteAddr = ctx.channel().remoteAddress();
if (error.isFatalError()) {
notifyAllChannelsOfErrorAndClose(new RemoteTransportException(
- "Fatal error at remote task manager '" + remoteAddr + "'.",
- remoteAddr, error.cause));
- }
- else {
+ "Fatal error at remote task manager '" + remoteAddr + "'.",
+ remoteAddr,
+ error.cause));
+ } else {
RemoteInputChannel inputChannel = inputChannels.get(error.receiverId);
if (inputChannel != null) {
if (error.cause.getClass() == PartitionNotFoundException.class) {
inputChannel.onFailedPartitionRequest();
- }
- else {
+ } else {
inputChannel.onError(new RemoteTransportException(
- "Error at remote task manager '" + remoteAddr + "'.",
- remoteAddr, error.cause));
+ "Error at remote task manager '" + remoteAddr + "'.",
+ remoteAddr,
+ error.cause));
}
}
}
- }
- else {
+ } else {
throw new IllegalStateException("Received unknown message from producer: " + msg.getClass());
}
-
- return true;
}
- private boolean decodeBufferOrEvent(RemoteInputChannel inputChannel, NettyMessage.BufferResponse bufferOrEvent, boolean isStagedBuffer) throws Throwable {
- boolean releaseNettyBuffer = true;
-
+ private void decodeBufferOrEvent(RemoteInputChannel inputChannel, NettyMessage.BufferResponse bufferOrEvent) throws Throwable {
try {
ByteBuf nettyBuffer = bufferOrEvent.getNettyBuffer();
final int receivedSize = nettyBuffer.readableBytes();
-
if (bufferOrEvent.isBuffer()) {
// ---- Buffer ------------------------------------------------
// Early return for empty buffers. Otherwise Netty's readBytes() throws an
// IndexOutOfBoundsException.
if (receivedSize == 0) {
- inputChannel.onEmptyBuffer(bufferOrEvent.sequenceNumber, -1);
- return true;
+ inputChannel.onEmptyBuffer(bufferOrEvent.sequenceNumber, bufferOrEvent.backlog);
+ return;
}
- BufferProvider bufferProvider = inputChannel.getBufferProvider();
+ Buffer buffer = inputChannel.requestBuffer();
+ if (buffer != null) {
+ nettyBuffer.readBytes(buffer.asByteBuf(), receivedSize);
- if (bufferProvider == null) {
- // receiver has been cancelled/failed
+ inputChannel.onBuffer(buffer, bufferOrEvent.sequenceNumber, bufferOrEvent.backlog);
+ } else if (inputChannel.isReleased()) {
cancelRequestFor(bufferOrEvent.receiverId);
- return isStagedBuffer;
+ } else {
+ throw new IllegalStateException("No buffer available in credit-based input channel.");
}
-
- while (true) {
- Buffer buffer = bufferProvider.requestBuffer();
-
- if (buffer != null) {
- nettyBuffer.readBytes(buffer.asByteBuf(), receivedSize);
-
- inputChannel.onBuffer(buffer, bufferOrEvent.sequenceNumber, -1);
-
- return true;
- }
- else if (bufferListener.waitForBuffer(bufferProvider, bufferOrEvent)) {
- releaseNettyBuffer = false;
-
- return false;
- }
- else if (bufferProvider.isDestroyed()) {
- return isStagedBuffer;
- }
- }
- }
- else {
+ } else {
// ---- Event -------------------------------------------------
// TODO We can just keep the serialized data in the Netty buffer and release it later at the reader
byte[] byteArray = new byte[receivedSize];
@@ -323,191 +328,62 @@ class PartitionRequestClientHandler extends ChannelInboundHandlerAdapter {
Buffer buffer = new NetworkBuffer(memSeg, FreeingBufferRecycler.INSTANCE, false);
buffer.setSize(receivedSize);
- inputChannel.onBuffer(buffer, bufferOrEvent.sequenceNumber, -1);
-
- return true;
- }
- }
- finally {
- if (releaseNettyBuffer) {
- bufferOrEvent.releaseBuffer();
+ inputChannel.onBuffer(buffer, bufferOrEvent.sequenceNumber, bufferOrEvent.backlog);
}
+ } finally {
+ bufferOrEvent.releaseBuffer();
}
}
/**
- * This class would be replaced by CreditBasedClientHandler in the final,
- * so we only implement this method in CreditBasedClientHandler.
- */
- void notifyCreditAvailable(RemoteInputChannel inputChannel) {
- }
-
- private class AsyncErrorNotificationTask implements Runnable {
-
- private final Throwable error;
-
- public AsyncErrorNotificationTask(Throwable error) {
- this.error = error;
- }
-
- @Override
- public void run() {
- notifyAllChannelsOfErrorAndClose(error);
- }
- }
-
- /**
- * A buffer availability listener, which subscribes/unsubscribes the NIO
- * read event.
+ * Tries to write&flush unannounced credits for the next input channel in queue.
*
- * <p>If no buffer is available, the channel read event will be unsubscribed
- * until one becomes available again.
- *
- * <p>After a buffer becomes available again, the buffer is handed over by
- * the thread calling {@link #notifyBufferAvailable(Buffer)} to the network I/O
- * thread, which then continues the processing of the staged buffer.
+ * <p>This method may be called by the first input channel enqueuing, or the complete
+ * future's callback in previous input channel, or the channel writability changed event.
*/
- private class BufferListenerTask implements BufferListener, Runnable {
-
- private final AtomicReference<Buffer> availableBuffer = new AtomicReference<>();
-
- private NettyMessage.BufferResponse stagedBufferResponse;
-
- private boolean waitForBuffer(BufferProvider bufferProvider, NettyMessage.BufferResponse bufferResponse) {
-
- stagedBufferResponse = bufferResponse;
-
- if (bufferProvider.addBufferListener(this)) {
- if (ctx.channel().config().isAutoRead()) {
- ctx.channel().config().setAutoRead(false);
- }
-
- return true;
- }
- else {
- stagedBufferResponse = null;
-
- return false;
- }
- }
-
- private boolean hasStagedBufferOrEvent() {
- return stagedBufferResponse != null;
- }
-
- public void notifyBufferDestroyed() {
- // The buffer pool has been destroyed
- stagedBufferResponse = null;
-
- if (stagedMessages.isEmpty()) {
- ctx.channel().config().setAutoRead(true);
- ctx.channel().read();
- }
- else {
- ctx.channel().eventLoop().execute(stagedMessagesHandler);
- }
+ private void writeAndFlushNextMessageIfPossible(Channel channel) {
+ if (channelError.get() != null || !channel.isWritable()) {
+ return;
}
- // Called by the recycling thread (not network I/O thread)
- @Override
- public boolean notifyBufferAvailable(Buffer buffer) {
- boolean success = false;
+ while (true) {
+ RemoteInputChannel inputChannel = inputChannelsWithCredit.poll();
- try {
- if (availableBuffer.compareAndSet(null, buffer)) {
- ctx.channel().eventLoop().execute(this);
-
- success = true;
- }
- else {
- throw new IllegalStateException("Received a buffer notification, " +
- " but the previous one has not been handled yet.");
- }
- }
- catch (Throwable t) {
- ctx.channel().eventLoop().execute(new AsyncErrorNotificationTask(t));
- }
- finally {
- if (!success) {
- if (buffer != null) {
- buffer.recycleBuffer();
- }
- }
+ // The input channel may be null because of the write callbacks
+ // that are executed after each write.
+ if (inputChannel == null) {
+ return;
}
- return false;
- }
-
- /**
- * Continues the decoding of a staged buffer after a buffer has become available again.
- *
- * <p>This task is executed by the network I/O thread.
- */
- @Override
- public void run() {
- boolean success = false;
-
- Buffer buffer = null;
-
- try {
- if ((buffer = availableBuffer.getAndSet(null)) == null) {
- throw new IllegalStateException("Running buffer availability task w/o a buffer.");
- }
-
- ByteBuf nettyBuffer = stagedBufferResponse.getNettyBuffer();
- nettyBuffer.readBytes(buffer.asByteBuf(), nettyBuffer.readableBytes());
- stagedBufferResponse.releaseBuffer();
-
- RemoteInputChannel inputChannel = inputChannels.get(stagedBufferResponse.receiverId);
+ //It is no need to notify credit for the released channel.
+ if (!inputChannel.isReleased()) {
+ AddCredit msg = new AddCredit(
+ inputChannel.getPartitionId(),
+ inputChannel.getAndResetUnannouncedCredit(),
+ inputChannel.getInputChannelId());
- if (inputChannel != null) {
- inputChannel.onBuffer(buffer, stagedBufferResponse.sequenceNumber, -1);
-
- success = true;
- }
- else {
- cancelRequestFor(stagedBufferResponse.receiverId);
- }
+ // Write and flush and wait until this is done before
+ // trying to continue with the next input channel.
+ channel.writeAndFlush(msg).addListener(writeListener);
- stagedBufferResponse = null;
-
- if (stagedMessages.isEmpty()) {
- ctx.channel().config().setAutoRead(true);
- ctx.channel().read();
- }
- else {
- ctx.channel().eventLoop().execute(stagedMessagesHandler);
- }
- }
- catch (Throwable t) {
- notifyAllChannelsOfErrorAndClose(t);
- }
- finally {
- if (!success) {
- if (buffer != null) {
- buffer.recycleBuffer();
- }
- }
+ return;
}
}
}
- public class StagedMessagesHandlerTask implements Runnable {
+ private class WriteAndFlushNextMessageIfPossibleListener implements ChannelFutureListener {
@Override
- public void run() {
+ public void operationComplete(ChannelFuture future) throws Exception {
try {
- Object msg;
- while ((msg = stagedMessages.poll()) != null) {
- if (!decodeMsg(msg, true)) {
- return;
- }
+ if (future.isSuccess()) {
+ writeAndFlushNextMessageIfPossible(future.channel());
+ } else if (future.cause() != null) {
+ notifyAllChannelsOfErrorAndClose(future.cause());
+ } else {
+ notifyAllChannelsOfErrorAndClose(new IllegalStateException("Sending cancelled by user."));
}
-
- ctx.channel().config().setAutoRead(true);
- ctx.channel().read();
- }
- catch (Throwable t) {
+ } catch (Throwable t) {
notifyAllChannelsOfErrorAndClose(t);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/a9fdbc4b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestQueue.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestQueue.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestQueue.java
index 58df4b6..bacca9d 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestQueue.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestQueue.java
@@ -18,6 +18,7 @@
package org.apache.flink.runtime.io.network.netty;
+import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.runtime.io.network.api.EndOfPartitionEvent;
import org.apache.flink.runtime.io.network.api.serialization.EventSerializer;
import org.apache.flink.runtime.io.network.buffer.Buffer;
@@ -38,8 +39,9 @@ import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.util.ArrayDeque;
-import java.util.Queue;
import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
import static org.apache.flink.runtime.io.network.netty.NettyMessage.BufferResponse;
@@ -53,7 +55,11 @@ class PartitionRequestQueue extends ChannelInboundHandlerAdapter {
private final ChannelFutureListener writeListener = new WriteAndFlushNextMessageIfPossibleListener();
- private final Queue<SequenceNumberingViewReader> nonEmptyReader = new ArrayDeque<>();
+ /** The readers which are already enqueued available for transferring data. */
+ private final ArrayDeque<SequenceNumberingViewReader> availableReaders = new ArrayDeque<>();
+
+ /** All the readers created for the consumers' partition requests. */
+ private final ConcurrentMap<InputChannelID, SequenceNumberingViewReader> allReaders = new ConcurrentHashMap<>();
private final Set<InputChannelID> released = Sets.newHashSet();
@@ -88,6 +94,45 @@ class PartitionRequestQueue extends ChannelInboundHandlerAdapter {
});
}
+ /**
+ * Try to enqueue the reader once receiving credit notification from the consumer or receiving
+ * non-empty reader notification from the producer.
+ *
+ * <p>NOTE: Only one thread would trigger the actual enqueue after checking the reader's
+ * availability, so there is no race condition here.
+ */
+ private void enqueueAvailableReader(final SequenceNumberingViewReader reader) throws Exception {
+ if (!reader.isRegisteredAsAvailable() && reader.isAvailable()) {
+ // Queue an available reader for consumption. If the queue is empty,
+ // we try trigger the actual write. Otherwise this will be handled by
+ // the writeAndFlushNextMessageIfPossible calls.
+ boolean triggerWrite = availableReaders.isEmpty();
+ availableReaders.add(reader);
+
+ reader.setRegisteredAsAvailable(true);
+
+ if (triggerWrite) {
+ writeAndFlushNextMessageIfPossible(ctx.channel());
+ }
+ }
+ }
+
+ /**
+ * Accesses internal state to verify reader registration in the unit tests.
+ *
+ * <p><strong>Do not use anywhere else!</strong>
+ *
+ * @return readers which are enqueued available for transferring data
+ */
+ @VisibleForTesting
+ ArrayDeque<SequenceNumberingViewReader> getAvailableReaders() {
+ return availableReaders;
+ }
+
+ void notifyReaderCreated(final SequenceNumberingViewReader reader) {
+ allReaders.put(reader.getReceiverId(), reader);
+ }
+
public void cancel(InputChannelID receiverId) {
ctx.pipeline().fireUserEventTriggered(receiverId);
}
@@ -98,20 +143,35 @@ class PartitionRequestQueue extends ChannelInboundHandlerAdapter {
}
}
+ /**
+ * Adds unannounced credits from the consumer and enqueues the corresponding reader for this
+ * consumer (if not enqueued yet).
+ *
+ * @param receiverId The input channel id to identify the consumer.
+ * @param credit The unannounced credits of the consumer.
+ */
+ void addCredit(InputChannelID receiverId, int credit) throws Exception {
+ if (fatalError) {
+ return;
+ }
+
+ SequenceNumberingViewReader reader = allReaders.get(receiverId);
+ if (reader != null) {
+ reader.addCredit(credit);
+
+ enqueueAvailableReader(reader);
+ } else {
+ throw new IllegalStateException("No reader for receiverId = " + receiverId + " exists.");
+ }
+ }
+
@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object msg) throws Exception {
// The user event triggered event loop callback is used for thread-safe
// hand over of reader queues and cancelled producers.
if (msg.getClass() == SequenceNumberingViewReader.class) {
- // Queue a non-empty reader for consumption. If the queue
- // is empty, we try trigger the actual write. Otherwise this
- // will be handled by the writeAndFlushIfPossible calls.
- boolean triggerWrite = nonEmptyReader.isEmpty();
- nonEmptyReader.add((SequenceNumberingViewReader) msg);
- if (triggerWrite) {
- writeAndFlushNextMessageIfPossible(ctx.channel());
- }
+ enqueueAvailableReader((SequenceNumberingViewReader) msg);
} else if (msg.getClass() == InputChannelID.class) {
// Release partition view that get a cancel request.
InputChannelID toCancel = (InputChannelID) msg;
@@ -120,16 +180,19 @@ class PartitionRequestQueue extends ChannelInboundHandlerAdapter {
}
// Cancel the request for the input channel
- int size = nonEmptyReader.size();
+ int size = availableReaders.size();
for (int i = 0; i < size; i++) {
- SequenceNumberingViewReader reader = nonEmptyReader.poll();
+ SequenceNumberingViewReader reader = availableReaders.poll();
if (reader.getReceiverId().equals(toCancel)) {
reader.releaseAllResources();
+ reader.setRegisteredAsAvailable(false);
markAsReleased(reader.getReceiverId());
} else {
- nonEmptyReader.add(reader);
+ availableReaders.add(reader);
}
}
+
+ allReaders.remove(toCancel);
} else {
ctx.fireUserEventTriggered(msg);
}
@@ -141,7 +204,7 @@ class PartitionRequestQueue extends ChannelInboundHandlerAdapter {
}
private void writeAndFlushNextMessageIfPossible(final Channel channel) throws IOException {
- if (fatalError) {
+ if (fatalError || !channel.isWritable()) {
return;
}
@@ -151,64 +214,63 @@ class PartitionRequestQueue extends ChannelInboundHandlerAdapter {
BufferAndAvailability next = null;
try {
- if (channel.isWritable()) {
- while (true) {
- SequenceNumberingViewReader reader = nonEmptyReader.poll();
+ while (true) {
+ SequenceNumberingViewReader reader = availableReaders.poll();
- // No queue with available data. We allow this here, because
- // of the write callbacks that are executed after each write.
- if (reader == null) {
- return;
- }
+ // No queue with available data. We allow this here, because
+ // of the write callbacks that are executed after each write.
+ if (reader == null) {
+ return;
+ }
+
+ next = reader.getNextBuffer();
+ if (next == null) {
+ if (reader.isReleased()) {
+ markAsReleased(reader.getReceiverId());
- next = reader.getNextBuffer();
-
- if (next == null) {
- if (reader.isReleased()) {
- markAsReleased(reader.getReceiverId());
- Throwable cause = reader.getFailureCause();
-
- if (cause != null) {
- ErrorResponse msg = new ErrorResponse(
- new ProducerFailedException(cause),
- reader.getReceiverId());
-
- ctx.writeAndFlush(msg);
- }
- } else {
- IllegalStateException err = new IllegalStateException(
- "Bug in Netty consumer logic: reader queue got notified by partition " +
- "about available data, but none was available.");
- handleException(ctx.channel(), err);
- return;
+ Throwable cause = reader.getFailureCause();
+ if (cause != null) {
+ ErrorResponse msg = new ErrorResponse(
+ new ProducerFailedException(cause),
+ reader.getReceiverId());
+
+ ctx.writeAndFlush(msg);
}
} else {
- // this channel was now removed from the non-empty reader queue
- // we re-add it in case it has more data, because in that case no
- // "non-empty" notification will come for that reader from the queue.
- if (next.moreAvailable()) {
- nonEmptyReader.add(reader);
- }
+ IllegalStateException err = new IllegalStateException(
+ "Bug in Netty consumer logic: reader queue got notified by partition " +
+ "about available data, but none was available.");
+ handleException(ctx.channel(), err);
+ return;
+ }
+ } else {
+ // This channel was now removed from the available reader queue.
+ // We re-add it into the queue if it is still available, otherwise we will
+ // notify the reader about the changed channel availability registration.
+ if (next.moreAvailable()) {
+ availableReaders.add(reader);
+ } else {
+ reader.setRegisteredAsAvailable(false);
+ }
- BufferResponse msg = new BufferResponse(
- next.buffer(),
- reader.getSequenceNumber(),
- reader.getReceiverId(),
- 0);
+ BufferResponse msg = new BufferResponse(
+ next.buffer(),
+ reader.getSequenceNumber(),
+ reader.getReceiverId(),
+ next.buffersInBacklog());
- if (isEndOfPartitionEvent(next.buffer())) {
- reader.notifySubpartitionConsumed();
- reader.releaseAllResources();
+ if (isEndOfPartitionEvent(next.buffer())) {
+ reader.notifySubpartitionConsumed();
+ reader.releaseAllResources();
- markAsReleased(reader.getReceiverId());
- }
+ markAsReleased(reader.getReceiverId());
+ }
- // Write and flush and wait until this is done before
- // trying to continue with the next buffer.
- channel.writeAndFlush(msg).addListener(writeListener);
+ // Write and flush and wait until this is done before
+ // trying to continue with the next buffer.
+ channel.writeAndFlush(msg).addListener(writeListener);
- return;
- }
+ return;
}
}
} catch (Throwable t) {
@@ -249,11 +311,14 @@ class PartitionRequestQueue extends ChannelInboundHandlerAdapter {
}
private void releaseAllResources() throws IOException {
- SequenceNumberingViewReader reader;
- while ((reader = nonEmptyReader.poll()) != null) {
+ // note: this is only ever executed by one thread: the Netty IO thread!
+ for (SequenceNumberingViewReader reader : allReaders.values()) {
reader.releaseAllResources();
markAsReleased(reader.getReceiverId());
}
+
+ availableReaders.clear();
+ allReaders.clear();
}
/**
http://git-wip-us.apache.org/repos/asf/flink/blob/a9fdbc4b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestServerHandler.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestServerHandler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestServerHandler.java
index bd867da..997d74a 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestServerHandler.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestServerHandler.java
@@ -21,6 +21,7 @@ package org.apache.flink.runtime.io.network.netty;
import org.apache.flink.runtime.io.network.TaskEventDispatcher;
import org.apache.flink.runtime.io.network.netty.NettyMessage.CancelPartitionRequest;
import org.apache.flink.runtime.io.network.netty.NettyMessage.CloseRequest;
+import org.apache.flink.runtime.io.network.netty.NettyMessage.AddCredit;
import org.apache.flink.runtime.io.network.partition.PartitionNotFoundException;
import org.apache.flink.runtime.io.network.partition.ResultPartitionProvider;
import org.apache.flink.runtime.io.network.partition.consumer.InputChannelID;
@@ -83,8 +84,11 @@ class PartitionRequestServerHandler extends SimpleChannelInboundHandler<NettyMes
try {
SequenceNumberingViewReader reader = new SequenceNumberingViewReader(
request.receiverId,
+ request.credit,
outboundQueue);
+ outboundQueue.notifyReaderCreated(reader);
+
reader.requestSubpartitionView(
partitionProvider,
request.partitionId,
@@ -108,6 +112,10 @@ class PartitionRequestServerHandler extends SimpleChannelInboundHandler<NettyMes
outboundQueue.cancel(request.receiverId);
} else if (msgClazz == CloseRequest.class) {
outboundQueue.close();
+ } else if (msgClazz == AddCredit.class) {
+ AddCredit request = (AddCredit) msg;
+
+ outboundQueue.addCredit(request.receiverId, request.credit);
} else {
LOG.warn("Received unexpected client request: {}", msg);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/a9fdbc4b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/SequenceNumberingViewReader.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/SequenceNumberingViewReader.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/SequenceNumberingViewReader.java
index fcbfb21..c8e97bd 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/SequenceNumberingViewReader.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/SequenceNumberingViewReader.java
@@ -19,6 +19,7 @@
package org.apache.flink.runtime.io.network.netty;
import org.apache.flink.runtime.io.network.partition.ResultSubpartition.BufferAndBacklog;
+import org.apache.flink.annotation.VisibleForTesting;
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;
@@ -49,10 +50,27 @@ class SequenceNumberingViewReader implements BufferAvailabilityListener {
private volatile ResultSubpartitionView subpartitionView;
+ /**
+ * The status indicating whether this reader is already enqueued in the pipeline for transferring
+ * data or not.
+ *
+ * <p>It is mainly used to avoid repeated registrations but should be accessed by a single
+ * thread only since there is no synchronisation.
+ */
+ private boolean isRegisteredAsAvailable = false;
+
+ /** The number of available buffers for holding data on the consumer side. */
+ private int numCreditsAvailable;
+
private int sequenceNumber = -1;
- SequenceNumberingViewReader(InputChannelID receiverId, PartitionRequestQueue requestQueue) {
+ SequenceNumberingViewReader(
+ InputChannelID receiverId,
+ int initialCredit,
+ PartitionRequestQueue requestQueue) {
+
this.receiverId = receiverId;
+ this.numCreditsAvailable = initialCredit;
this.requestQueue = requestQueue;
}
@@ -77,6 +95,57 @@ class SequenceNumberingViewReader implements BufferAvailabilityListener {
}
}
+ /**
+ * The credits from consumer are added in incremental way.
+ *
+ * @param creditDeltas The credit deltas
+ */
+ public void addCredit(int creditDeltas) {
+ numCreditsAvailable += creditDeltas;
+ }
+
+ /**
+ * Updates the value to indicate whether the reader is enqueued in the pipeline or not.
+ *
+ * @param isRegisteredAvailable True if this reader is already enqueued in the pipeline.
+ */
+ public void setRegisteredAsAvailable(boolean isRegisteredAvailable) {
+ this.isRegisteredAsAvailable = isRegisteredAvailable;
+ }
+
+ public boolean isRegisteredAsAvailable() {
+ return isRegisteredAsAvailable;
+ }
+
+ /**
+ * Check whether this reader is available or not.
+ *
+ * <p>Returns true only if the next buffer is an event or the reader has both available
+ * credits and buffers.
+ */
+ public boolean isAvailable() {
+ // BEWARE: this must be in sync with #isAvailable()!
+ return numBuffersAvailable.get() > 0 &&
+ (numCreditsAvailable > 0 || subpartitionView.nextBufferIsEvent());
+ }
+
+ /**
+ * Check whether this reader is available or not (internal use, in sync with
+ * {@link #isAvailable()}, but slightly faster).
+ *
+ * <p>Returns true only if the next buffer is an event or the reader has both available
+ * credits and buffers.
+ *
+ * @param bufferAndBacklog
+ * current buffer and backlog including information about the next buffer
+ * @param remaining
+ * remaining number of queued buffers, i.e. <tt>numBuffersAvailable.get()</tt>
+ */
+ private boolean isAvailable(BufferAndBacklog bufferAndBacklog, long remaining) {
+ return remaining > 0 &&
+ (numCreditsAvailable > 0 || bufferAndBacklog.nextBufferIsEvent());
+ }
+
InputChannelID getReceiverId() {
return receiverId;
}
@@ -85,17 +154,32 @@ class SequenceNumberingViewReader implements BufferAvailabilityListener {
return sequenceNumber;
}
+ @VisibleForTesting
+ int getNumCreditsAvailable() {
+ return numCreditsAvailable;
+ }
+
+ @VisibleForTesting
+ long getNumBuffersAvailable() {
+ return numBuffersAvailable.get();
+ }
+
public BufferAndAvailability getNextBuffer() throws IOException, InterruptedException {
BufferAndBacklog next = subpartitionView.getNextBuffer();
if (next != null) {
long remaining = numBuffersAvailable.decrementAndGet();
sequenceNumber++;
- if (remaining >= 0) {
- return new BufferAndAvailability(next.buffer(), remaining > 0, next.buffersInBacklog());
- } else {
+ if (remaining < 0) {
throw new IllegalStateException("no buffer available");
}
+
+ if (next.buffer().isBuffer() && --numCreditsAvailable < 0) {
+ throw new IllegalStateException("no credit available");
+ }
+
+ return new BufferAndAvailability(
+ next.buffer(), isAvailable(next, remaining), next.buffersInBacklog());
} else {
return null;
}
@@ -132,6 +216,8 @@ class SequenceNumberingViewReader implements BufferAvailabilityListener {
", receiverId=" + receiverId +
", numBuffersAvailable=" + numBuffersAvailable.get() +
", sequenceNumber=" + sequenceNumber +
+ ", numCreditsAvailable=" + numCreditsAvailable +
+ ", isRegisteredAsAvailable=" + isRegisteredAsAvailable +
'}';
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/a9fdbc4b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartition.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartition.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartition.java
index 051f871..0637cc7 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartition.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartition.java
@@ -157,13 +157,25 @@ class PipelinedSubpartition extends ResultSubpartition {
decreaseBuffersInBacklog(buffer);
if (buffer != null) {
- return new BufferAndBacklog(buffer, buffersInBacklog);
+ return new BufferAndBacklog(buffer, buffersInBacklog, _nextBufferIsEvent());
} else {
return null;
}
}
}
+ boolean nextBufferIsEvent() {
+ synchronized (buffers) {
+ return _nextBufferIsEvent();
+ }
+ }
+
+ private boolean _nextBufferIsEvent() {
+ assert Thread.holdsLock(buffers);
+
+ return !buffers.isEmpty() && !buffers.peekFirst().isBuffer();
+ }
+
@Override
public int releaseMemory() {
// The pipelined subpartition does not react to memory release requests.
http://git-wip-us.apache.org/repos/asf/flink/blob/a9fdbc4b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartitionView.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartitionView.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartitionView.java
index 2aafd3f..6fd6022 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartitionView.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartitionView.java
@@ -76,6 +76,11 @@ class PipelinedSubpartitionView implements ResultSubpartitionView {
}
@Override
+ public boolean nextBufferIsEvent() {
+ return parent.nextBufferIsEvent();
+ }
+
+ @Override
public Throwable getFailureCause() {
return parent.getFailureCause();
}
http://git-wip-us.apache.org/repos/asf/flink/blob/a9fdbc4b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartitionType.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartitionType.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartitionType.java
index 683fe37..256387c 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartitionType.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartitionType.java
@@ -20,9 +20,9 @@ package org.apache.flink.runtime.io.network.partition;
public enum ResultPartitionType {
- BLOCKING(false, false, false, false),
+ BLOCKING(false, false, false),
- PIPELINED(true, true, false, false),
+ PIPELINED(true, true, false),
/**
* Pipelined partitions with a bounded (local) buffer pool.
@@ -35,13 +35,7 @@ public enum ResultPartitionType {
* For batch jobs, it will be best to keep this unlimited ({@link #PIPELINED}) since there are
* no checkpoint barriers.
*/
- PIPELINED_BOUNDED(true, true, true, false),
-
- /**
- * Pipelined partitions with a bounded (local) buffer pool for floating buffers in input gate, and a number
- * of exclusive buffers per input channel. The producer transfers data based on consumer's available credits.
- */
- PIPELINED_CREDIT_BASED(true, true, true, true);
+ PIPELINED_BOUNDED(true, true, true);
/** Can the partition be consumed while being produced? */
private final boolean isPipelined;
@@ -52,17 +46,13 @@ public enum ResultPartitionType {
/** Does this partition use a limited number of (network) buffers? */
private final boolean isBounded;
- /** Does this partition only send data when consumer has available buffers? */
- private final boolean isCreditBased;
-
/**
* Specifies the behaviour of an intermediate result partition at runtime.
*/
- ResultPartitionType(boolean isPipelined, boolean hasBackPressure, boolean isBounded, boolean isCreditBased) {
+ ResultPartitionType(boolean isPipelined, boolean hasBackPressure, boolean isBounded) {
this.isPipelined = isPipelined;
this.hasBackPressure = hasBackPressure;
this.isBounded = isBounded;
- this.isCreditBased = isCreditBased;
}
public boolean hasBackPressure() {
@@ -85,13 +75,4 @@ public enum ResultPartitionType {
public boolean isBounded() {
return isBounded;
}
-
- /**
- * Whether this partition uses the credit-based mode to transfer data or not.
- *
- * @return <tt>true</tt> if the data is transferred based on consumer's credit
- */
- public boolean isCreditBased() {
- return isCreditBased;
- }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/a9fdbc4b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultSubpartition.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultSubpartition.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultSubpartition.java
index e42bd90..e51f215 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultSubpartition.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultSubpartition.java
@@ -123,10 +123,12 @@ public abstract class ResultSubpartition {
private final Buffer buffer;
private final int buffersInBacklog;
+ private final boolean nextBufferIsEvent;
- public BufferAndBacklog(Buffer buffer, int buffersInBacklog) {
+ public BufferAndBacklog(Buffer buffer, int buffersInBacklog, boolean nextBufferIsEvent) {
this.buffer = checkNotNull(buffer);
this.buffersInBacklog = buffersInBacklog;
+ this.nextBufferIsEvent = nextBufferIsEvent;
}
public Buffer buffer() {
@@ -136,6 +138,10 @@ public abstract class ResultSubpartition {
public int buffersInBacklog() {
return buffersInBacklog;
}
+
+ public boolean nextBufferIsEvent() {
+ return nextBufferIsEvent;
+ }
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/a9fdbc4b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultSubpartitionView.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultSubpartitionView.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultSubpartitionView.java
index 71753d2..bb7e50c 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultSubpartitionView.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultSubpartitionView.java
@@ -52,4 +52,9 @@ public interface ResultSubpartitionView {
boolean isReleased();
Throwable getFailureCause();
+
+ /**
+ * Returns whether the next buffer is an event or not.
+ */
+ boolean nextBufferIsEvent();
}
http://git-wip-us.apache.org/repos/asf/flink/blob/a9fdbc4b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/SpillableSubpartitionView.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/SpillableSubpartitionView.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/SpillableSubpartitionView.java
index 279a023..4ed87ce 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/SpillableSubpartitionView.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/SpillableSubpartitionView.java
@@ -31,6 +31,7 @@ import java.util.ArrayDeque;
import java.util.concurrent.atomic.AtomicBoolean;
import static org.apache.flink.util.Preconditions.checkNotNull;
+import static org.apache.flink.util.Preconditions.checkState;
class SpillableSubpartitionView implements ResultSubpartitionView {
@@ -133,25 +134,38 @@ class SpillableSubpartitionView implements ResultSubpartitionView {
@Nullable
@Override
public BufferAndBacklog getNextBuffer() throws IOException, InterruptedException {
+ Buffer current = null;
+ boolean nextBufferIsEvent = false;
+ int newBacklog = 0; // this is always correct if current is non-null!
+
synchronized (buffers) {
if (isReleased.get()) {
return null;
} else if (nextBuffer != null) {
- Buffer current = nextBuffer;
+ current = nextBuffer;
nextBuffer = buffers.poll();
+ newBacklog = parent.decreaseBuffersInBacklog(current);
if (nextBuffer != null) {
listener.notifyBuffersAvailable(1);
+ nextBufferIsEvent = !nextBuffer.isBuffer();
}
- int newBacklog = parent.decreaseBuffersInBacklog(current);
- return new BufferAndBacklog(current, newBacklog);
+ // if we are spilled (but still process a non-spilled nextBuffer), we don't know the
+ // state of nextBufferIsEvent...
+ if (spilledView == null) {
+ return new BufferAndBacklog(current, newBacklog, nextBufferIsEvent);
+ }
}
} // else: spilled
SpilledSubpartitionView spilled = spilledView;
if (spilled != null) {
- return spilled.getNextBuffer();
+ if (current != null) {
+ return new BufferAndBacklog(current, newBacklog, spilled.nextBufferIsEvent());
+ } else {
+ return spilled.getNextBuffer();
+ }
} else {
throw new IllegalStateException("No in-memory buffers available, but also nothing spilled.");
}
@@ -200,6 +214,21 @@ class SpillableSubpartitionView implements ResultSubpartitionView {
}
@Override
+ public boolean nextBufferIsEvent() {
+ synchronized (buffers) {
+ if (isReleased.get()) {
+ return false;
+ } else if (nextBuffer != null) {
+ return !nextBuffer.isBuffer();
+ }
+ } // else: spilled
+
+ checkState(spilledView != null, "No in-memory buffers available, but also nothing spilled.");
+
+ return spilledView.nextBufferIsEvent();
+ }
+
+ @Override
public Throwable getFailureCause() {
SpilledSubpartitionView spilled = spilledView;
if (spilled != null) {
http://git-wip-us.apache.org/repos/asf/flink/blob/a9fdbc4b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/SpilledSubpartitionView.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/SpilledSubpartitionView.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/SpilledSubpartitionView.java
index f0752c0..1db5113 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/SpilledSubpartitionView.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/SpilledSubpartitionView.java
@@ -18,7 +18,6 @@
package org.apache.flink.runtime.io.network.partition;
-import org.apache.flink.runtime.io.network.partition.ResultSubpartition.BufferAndBacklog;
import org.apache.flink.core.memory.MemorySegment;
import org.apache.flink.core.memory.MemorySegmentFactory;
import org.apache.flink.runtime.io.disk.iomanager.BufferFileReader;
@@ -27,11 +26,14 @@ import org.apache.flink.runtime.io.disk.iomanager.SynchronousBufferFileReader;
import org.apache.flink.runtime.io.network.buffer.Buffer;
import org.apache.flink.runtime.io.network.buffer.BufferRecycler;
import org.apache.flink.runtime.io.network.buffer.NetworkBuffer;
+import org.apache.flink.runtime.io.network.partition.ResultSubpartition.BufferAndBacklog;
import org.apache.flink.runtime.util.event.NotificationListener;
+
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import javax.annotation.Nullable;
+
import java.io.IOException;
import java.util.ArrayDeque;
import java.util.Queue;
@@ -74,6 +76,9 @@ class SpilledSubpartitionView implements ResultSubpartitionView, NotificationLis
/** Flag indicating whether all resources have been released. */
private AtomicBoolean isReleased = new AtomicBoolean();
+ /** The next buffer to hand out. */
+ private Buffer nextBuffer;
+
/** Flag indicating whether a spill is still in progress. */
private volatile boolean isSpillInProgress = true;
@@ -119,17 +124,38 @@ class SpilledSubpartitionView implements ResultSubpartitionView, NotificationLis
@Nullable
@Override
public BufferAndBacklog getNextBuffer() throws IOException, InterruptedException {
- if (fileReader.hasReachedEndOfFile() || isSpillInProgress) {
+ if (isSpillInProgress) {
+ return null;
+ }
+
+ Buffer current;
+ if (nextBuffer == null) {
+ current = requestAndFillBuffer();
+ nextBuffer = requestAndFillBuffer();
+ } else {
+ current = nextBuffer;
+ nextBuffer = requestAndFillBuffer();
+ }
+
+ if (current == null) {
return null;
}
+ int newBacklog = parent.decreaseBuffersInBacklog(current);
+ boolean nextBufferIsEvent = nextBuffer != null && !nextBuffer.isBuffer();
+ return new BufferAndBacklog(current, newBacklog, nextBufferIsEvent);
+ }
+
+ @Nullable
+ private Buffer requestAndFillBuffer() throws IOException, InterruptedException {
+ if (fileReader.hasReachedEndOfFile()) {
+ return null;
+ }
// TODO This is fragile as we implicitly expect that multiple calls to
// this method don't happen before recycling buffers returned earlier.
Buffer buffer = bufferPool.requestBufferBlocking();
fileReader.readInto(buffer);
-
- int newBacklog = parent.decreaseBuffersInBacklog(buffer);
- return new BufferAndBacklog(buffer, newBacklog);
+ return buffer;
}
@Override
@@ -155,6 +181,10 @@ class SpilledSubpartitionView implements ResultSubpartitionView, NotificationLis
spillWriter.closeAndDelete();
fileReader.close();
+ if (nextBuffer != null) {
+ nextBuffer.recycleBuffer();
+ nextBuffer = null;
+ }
bufferPool.destroy();
}
}
@@ -165,6 +195,19 @@ class SpilledSubpartitionView implements ResultSubpartitionView, NotificationLis
}
@Override
+ public boolean nextBufferIsEvent() {
+ if (nextBuffer == null) {
+ try {
+ nextBuffer = requestAndFillBuffer();
+ } catch (Exception e) {
+ // we can ignore this here (we will get it again once getNextBuffer() is called)
+ return false;
+ }
+ }
+ return nextBuffer != null && !nextBuffer.isBuffer();
+ }
+
+ @Override
public Throwable getFailureCause() {
return parent.getFailureCause();
}
http://git-wip-us.apache.org/repos/asf/flink/blob/a9fdbc4b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannel.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannel.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannel.java
index 0d633d7..8a8c7f5 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannel.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannel.java
@@ -284,7 +284,7 @@ public class RemoteInputChannel extends InputChannel implements BufferRecycler,
/**
* Enqueue this input channel in the pipeline for notifying the producer of unannounced credit.
*/
- void notifyCreditAvailable() {
+ private void notifyCreditAvailable() {
checkState(partitionRequestClient != null, "Tried to send task event to producer before requesting a queue.");
// We should skip the notification if this channel is already released.
http://git-wip-us.apache.org/repos/asf/flink/blob/a9fdbc4b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGate.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGate.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGate.java
index 945d127..deac251 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGate.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGate.java
@@ -266,14 +266,6 @@ public class SingleInputGate implements InputGate {
// ------------------------------------------------------------------------
public void setBufferPool(BufferPool bufferPool) {
- // Sanity checks
- if (!getConsumedPartitionType().isCreditBased()) {
- checkArgument(numberOfInputChannels == bufferPool.getNumberOfRequiredMemorySegments(),
- "Bug in input gate setup logic: buffer pool has not enough guaranteed buffers " +
- "for this input gate. Input gates require at least as many buffers as " +
- "there are input channels.");
- }
-
checkState(this.bufferPool == null, "Bug in input gate setup logic: buffer pool has" +
"already been set for this input gate.");
@@ -346,11 +338,8 @@ public class SingleInputGate implements InputGate {
}
else if (partitionLocation.isRemote()) {
newChannel = unknownChannel.toRemoteInputChannel(partitionLocation.getConnectionId());
-
- if (getConsumedPartitionType().isCreditBased()) {
- ((RemoteInputChannel)newChannel).assignExclusiveSegments(
- networkBufferPool.requestMemorySegments(networkBuffersPerChannel));
- }
+ ((RemoteInputChannel)newChannel).assignExclusiveSegments(
+ networkBufferPool.requestMemorySegments(networkBuffersPerChannel));
}
else {
throw new IllegalStateException("Tried to update unknown channel with unknown channel.");
http://git-wip-us.apache.org/repos/asf/flink/blob/a9fdbc4b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/NetworkEnvironmentTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/NetworkEnvironmentTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/NetworkEnvironmentTest.java
index 2a54769..e3f1d89 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/NetworkEnvironmentTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/NetworkEnvironmentTest.java
@@ -84,8 +84,7 @@ public class NetworkEnvironmentTest {
SingleInputGate ig1 = createSingleInputGateMock(ResultPartitionType.PIPELINED, 2);
SingleInputGate ig2 = createSingleInputGateMock(ResultPartitionType.BLOCKING, 2);
SingleInputGate ig3 = createSingleInputGateMock(ResultPartitionType.PIPELINED_BOUNDED, 2);
- SingleInputGate ig4 = createSingleInputGateMock(ResultPartitionType.PIPELINED_CREDIT_BASED, 8);
- final SingleInputGate[] inputGates = new SingleInputGate[] {ig1, ig2, ig3, ig4};
+ final SingleInputGate[] inputGates = new SingleInputGate[] {ig1, ig2, ig3};
// overall task to register
Task task = mock(Task.class);
@@ -99,7 +98,9 @@ public class NetworkEnvironmentTest {
assertEquals(2 * 2 + 8, rp3.getBufferPool().getMaxNumberOfMemorySegments());
assertEquals(8 * 2 + 8, rp4.getBufferPool().getMaxNumberOfMemorySegments());
- verify(ig4, times(1)).assignExclusiveSegments(network.getNetworkBufferPool(), 2);
+ verify(ig1, times(1)).assignExclusiveSegments(network.getNetworkBufferPool(), 2);
+ verify(ig2, times(1)).assignExclusiveSegments(network.getNetworkBufferPool(), 2);
+ verify(ig3, times(1)).assignExclusiveSegments(network.getNetworkBufferPool(), 2);
network.shutdown();
}
@@ -153,8 +154,6 @@ public class NetworkEnvironmentTest {
public Void answer(final InvocationOnMock invocation) throws Throwable {
BufferPool bp = invocation.getArgumentAt(0, BufferPool.class);
if (partitionType == ResultPartitionType.PIPELINED_BOUNDED) {
- assertEquals(channels * 2 + 8, bp.getMaxNumberOfMemorySegments());
- } else if (partitionType == ResultPartitionType.PIPELINED_CREDIT_BASED) {
assertEquals(8, bp.getMaxNumberOfMemorySegments());
} else {
assertEquals(Integer.MAX_VALUE, bp.getMaxNumberOfMemorySegments());
http://git-wip-us.apache.org/repos/asf/flink/blob/a9fdbc4b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/CancelPartitionRequestTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/CancelPartitionRequestTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/CancelPartitionRequestTest.java
index b51dc1e..cf56851 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/CancelPartitionRequestTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/CancelPartitionRequestTest.java
@@ -45,6 +45,7 @@ import java.util.concurrent.TimeUnit;
import static org.apache.flink.runtime.io.network.netty.NettyMessage.CancelPartitionRequest;
import static org.apache.flink.runtime.io.network.netty.NettyMessage.PartitionRequest;
+import static org.apache.flink.runtime.io.network.netty.NettyMessage.AddCredit;
import static org.apache.flink.runtime.io.network.netty.NettyTestUtil.connect;
import static org.apache.flink.runtime.io.network.netty.NettyTestUtil.initServerAndClient;
import static org.apache.flink.runtime.io.network.netty.NettyTestUtil.shutdown;
@@ -100,7 +101,7 @@ public class CancelPartitionRequestTest {
Channel ch = connect(serverAndClient);
// Request for non-existing input channel => results in cancel request
- ch.writeAndFlush(new PartitionRequest(pid, 0, new InputChannelID(), 2)).await();
+ ch.writeAndFlush(new PartitionRequest(pid, 0, new InputChannelID(), Integer.MAX_VALUE)).await();
// Wait for the notification
if (!sync.await(TestingUtils.TESTING_DURATION().toMillis(), TimeUnit.MILLISECONDS)) {
@@ -153,7 +154,7 @@ public class CancelPartitionRequestTest {
// Request for non-existing input channel => results in cancel request
InputChannelID inputChannelId = new InputChannelID();
- ch.writeAndFlush(new PartitionRequest(pid, 0, inputChannelId, 2)).await();
+ ch.writeAndFlush(new PartitionRequest(pid, 0, inputChannelId, Integer.MAX_VALUE)).await();
// Wait for the notification
if (!sync.await(TestingUtils.TESTING_DURATION().toMillis(), TimeUnit.MILLISECONDS)) {
@@ -193,7 +194,7 @@ public class CancelPartitionRequestTest {
public BufferAndBacklog getNextBuffer() throws IOException, InterruptedException {
Buffer buffer = bufferProvider.requestBufferBlocking();
buffer.setSize(buffer.getMaxCapacity()); // fake some data
- return new BufferAndBacklog(buffer, 0);
+ return new BufferAndBacklog(buffer, 0, false);
}
@Override
@@ -215,6 +216,11 @@ public class CancelPartitionRequestTest {
}
@Override
+ public boolean nextBufferIsEvent() {
+ return false;
+ }
+
+ @Override
public Throwable getFailureCause() {
return null;
}