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:56 UTC
[3/5] flink git commit: [hotfix] Add the switch for keeping both the
old mode and the new credit-based mode
[hotfix] Add the switch for keeping both the old mode and the new credit-based mode
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/0093bcbe
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/0093bcbe
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/0093bcbe
Branch: refs/heads/master
Commit: 0093bcbe771f296baf3857ef15fe7ec9b22bbc34
Parents: a9fdbc4
Author: Zhijiang <wa...@aliyun.com>
Authored: Wed Jan 17 14:15:04 2018 +0800
Committer: Stefan Richter <s....@data-artisans.com>
Committed: Fri Feb 16 13:55:36 2018 +0100
----------------------------------------------------------------------
.../flink/configuration/TaskManagerOptions.java | 11 +
.../io/network/NetworkClientHandler.java | 47 +++
.../runtime/io/network/NetworkEnvironment.java | 31 +-
.../io/network/NetworkSequenceViewReader.java | 75 ++++
...reditBasedPartitionRequestClientHandler.java | 391 ++++++++++++++++++
.../CreditBasedSequenceNumberingViewReader.java | 223 +++++++++++
.../runtime/io/network/netty/NettyConfig.java | 4 +
.../network/netty/NettyConnectionManager.java | 9 +-
.../runtime/io/network/netty/NettyProtocol.java | 13 +-
.../network/netty/PartitionRequestClient.java | 31 +-
.../netty/PartitionRequestClientFactory.java | 7 +-
.../netty/PartitionRequestClientHandler.java | 400 ++++++++++++-------
.../io/network/netty/PartitionRequestQueue.java | 25 +-
.../netty/PartitionRequestServerHandler.java | 22 +-
.../netty/SequenceNumberingViewReader.java | 108 ++---
.../taskexecutor/TaskManagerServices.java | 12 +-
.../io/network/NetworkEnvironmentTest.java | 3 +-
.../netty/CancelPartitionRequestTest.java | 4 +-
.../netty/ClientTransportErrorHandlingTest.java | 28 +-
...tBasedPartitionRequestClientHandlerTest.java | 391 ++++++++++++++++++
.../network/netty/NettyClientServerSslTest.java | 6 +-
.../NettyServerLowAndHighWatermarkTest.java | 2 +-
.../PartitionRequestClientFactoryTest.java | 2 +-
.../PartitionRequestClientHandlerTest.java | 194 +--------
.../netty/PartitionRequestQueueTest.java | 13 +-
.../netty/ServerTransportErrorHandlingTest.java | 2 +-
...askManagerComponentsStartupShutdownTest.java | 3 +-
.../StreamNetworkBenchmarkEnvironment.java | 3 +-
28 files changed, 1564 insertions(+), 496 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/0093bcbe/flink-core/src/main/java/org/apache/flink/configuration/TaskManagerOptions.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/configuration/TaskManagerOptions.java b/flink-core/src/main/java/org/apache/flink/configuration/TaskManagerOptions.java
index 04b49ed..3b6697f 100644
--- a/flink-core/src/main/java/org/apache/flink/configuration/TaskManagerOptions.java
+++ b/flink-core/src/main/java/org/apache/flink/configuration/TaskManagerOptions.java
@@ -306,6 +306,17 @@ public class TaskManagerOptions {
.defaultValue(false)
.withDescription("Boolean flag to enable/disable more detailed metrics about inbound/outbound network queue lengths.");
+ /**
+ * Config parameter defining whether to enable credit-based flow control or not.
+ *
+ * @deprecated Will be removed for Flink 1.6 when the old code will be dropped in favour of
+ * credit-based flow control.
+ */
+ @Deprecated
+ public static final ConfigOption<Boolean> NETWORK_CREDIT_BASED_FLOW_CONTROL_ENABLED =
+ key("taskmanager.network.credit-based-flow-control.enabled")
+ .defaultValue(true);
+
// ------------------------------------------------------------------------
// Task Options
// ------------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/0093bcbe/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/NetworkClientHandler.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/NetworkClientHandler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/NetworkClientHandler.java
new file mode 100644
index 0000000..7f631b4
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/NetworkClientHandler.java
@@ -0,0 +1,47 @@
+/*
+ * 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;
+
+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.channel.ChannelHandler;
+
+import java.io.IOException;
+
+/**
+ * Channel handler to read and write network messages on client side.
+ */
+public interface NetworkClientHandler extends ChannelHandler {
+
+ void addInputChannel(RemoteInputChannel inputChannel) throws IOException;
+
+ void removeInputChannel(RemoteInputChannel inputChannel);
+
+ void cancelRequestFor(InputChannelID 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);
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/0093bcbe/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 2441667..23ac911 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
@@ -85,6 +85,8 @@ public class NetworkEnvironment {
/** Number of extra network buffers to use for each outgoing/incoming gate (result partition/input gate). */
private final int extraNetworkBuffersPerGate;
+ private final boolean enableCreditBased;
+
private boolean isShutdown;
public NetworkEnvironment(
@@ -99,7 +101,8 @@ public class NetworkEnvironment {
int partitionRequestInitialBackoff,
int partitionRequestMaxBackoff,
int networkBuffersPerChannel,
- int extraNetworkBuffersPerGate) {
+ int extraNetworkBuffersPerGate,
+ boolean enableCreditBased) {
this.networkBufferPool = checkNotNull(networkBufferPool);
this.connectionManager = checkNotNull(connectionManager);
@@ -118,6 +121,8 @@ public class NetworkEnvironment {
isShutdown = false;
this.networkBuffersPerChannel = networkBuffersPerChannel;
this.extraNetworkBuffersPerGate = extraNetworkBuffersPerGate;
+
+ this.enableCreditBased = enableCreditBased;
}
// --------------------------------------------------------------------------------------------
@@ -223,14 +228,24 @@ public class NetworkEnvironment {
@VisibleForTesting
public void setupInputGate(SingleInputGate gate) throws IOException {
BufferPool bufferPool = null;
-
+ int maxNumberOfMemorySegments;
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);
+ if (enableCreditBased) {
+ 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);
+ } else {
+ 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) {
http://git-wip-us.apache.org/repos/asf/flink/blob/0093bcbe/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/NetworkSequenceViewReader.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/NetworkSequenceViewReader.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/NetworkSequenceViewReader.java
new file mode 100644
index 0000000..4d7797c
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/NetworkSequenceViewReader.java
@@ -0,0 +1,75 @@
+/*
+ * 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;
+
+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.consumer.InputChannel.BufferAndAvailability;
+import org.apache.flink.runtime.io.network.partition.consumer.InputChannelID;
+
+import java.io.IOException;
+
+/**
+ * Simple wrapper for the partition readerQueue iterator, which increments a
+ * sequence number for each returned buffer and remembers the receiver ID.
+ */
+public interface NetworkSequenceViewReader {
+
+ void requestSubpartitionView(
+ ResultPartitionProvider partitionProvider,
+ ResultPartitionID resultPartitionId,
+ int subPartitionIndex) throws IOException;
+
+ BufferAndAvailability getNextBuffer() throws IOException, InterruptedException;
+
+ /**
+ * The credits from consumer are added in incremental way.
+ *
+ * @param creditDeltas The credit deltas
+ */
+ void addCredit(int creditDeltas);
+
+ /**
+ * Checks whether this reader is available or not.
+ *
+ * @return True if the reader is available.
+ */
+ boolean isAvailable();
+
+ boolean isRegisteredAsAvailable();
+
+ /**
+ * 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.
+ */
+ void setRegisteredAsAvailable(boolean isRegisteredAvailable);
+
+ void notifySubpartitionConsumed() throws IOException;
+
+ boolean isReleased();
+
+ void releaseAllResources() throws IOException;
+
+ Throwable getFailureCause();
+
+ InputChannelID getReceiverId();
+
+ int getSequenceNumber();
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/0093bcbe/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/CreditBasedPartitionRequestClientHandler.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/CreditBasedPartitionRequestClientHandler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/CreditBasedPartitionRequestClientHandler.java
new file mode 100644
index 0000000..47fbdb2
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/CreditBasedPartitionRequestClientHandler.java
@@ -0,0 +1,391 @@
+/*
+ * 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.NetworkClientHandler;
+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.
+ *
+ * <p>It is used in the new network credit-based mode.
+ */
+class CreditBasedPartitionRequestClientHandler extends ChannelInboundHandlerAdapter implements NetworkClientHandler {
+
+ private static final Logger LOG = LoggerFactory.getLogger(CreditBasedPartitionRequestClientHandler.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
+ // ------------------------------------------------------------------------
+
+ @Override
+ public void addInputChannel(RemoteInputChannel listener) throws IOException {
+ checkError();
+
+ if (!inputChannels.containsKey(listener.getInputChannelId())) {
+ inputChannels.put(listener.getInputChannelId(), listener);
+ }
+ }
+
+ @Override
+ public void removeInputChannel(RemoteInputChannel listener) {
+ inputChannels.remove(listener.getInputChannelId());
+ }
+
+ @Override
+ public void cancelRequestFor(InputChannelID inputChannelId) {
+ if (inputChannelId == null || ctx == null) {
+ return;
+ }
+
+ if (cancelled.putIfAbsent(inputChannelId, inputChannelId) == null) {
+ ctx.writeAndFlush(new NettyMessage.CancelPartitionRequest(inputChannelId));
+ }
+ }
+
+ @Override
+ public 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);
+ }
+ }
+
+ /**
+ * 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) {
+ // 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 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, bufferOrEvent.backlog);
+ return;
+ }
+
+ Buffer buffer = inputChannel.requestBuffer();
+ if (buffer != null) {
+ nettyBuffer.readBytes(buffer.asByteBuf(), receivedSize);
+
+ 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[receivedSize];
+ nettyBuffer.readBytes(byteArray);
+
+ MemorySegment memSeg = MemorySegmentFactory.wrap(byteArray);
+ Buffer buffer = new NetworkBuffer(memSeg, FreeingBufferRecycler.INSTANCE, false);
+ buffer.setSize(receivedSize);
+
+ inputChannel.onBuffer(buffer, bufferOrEvent.sequenceNumber, bufferOrEvent.backlog);
+ }
+ } finally {
+ bufferOrEvent.releaseBuffer();
+ }
+ }
+
+ /**
+ * Tries to write&flush unannounced credits for the next input channel in queue.
+ *
+ * <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 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/0093bcbe/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/CreditBasedSequenceNumberingViewReader.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/CreditBasedSequenceNumberingViewReader.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/CreditBasedSequenceNumberingViewReader.java
new file mode 100644
index 0000000..5ebf62d
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/CreditBasedSequenceNumberingViewReader.java
@@ -0,0 +1,223 @@
+/*
+ * 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.NetworkSequenceViewReader;
+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;
+import org.apache.flink.runtime.io.network.partition.ResultSubpartitionView;
+import org.apache.flink.runtime.io.network.partition.consumer.InputChannel.BufferAndAvailability;
+import org.apache.flink.runtime.io.network.partition.consumer.InputChannelID;
+import org.apache.flink.runtime.io.network.partition.consumer.LocalInputChannel;
+
+import java.io.IOException;
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ * Simple wrapper for the subpartition view used in the new network credit-based mode.
+ *
+ * <p>It also keeps track of available buffers and notifies the outbound
+ * handler about non-emptiness, similar to the {@link LocalInputChannel}.
+ */
+class CreditBasedSequenceNumberingViewReader implements BufferAvailabilityListener, NetworkSequenceViewReader {
+
+ private final Object requestLock = new Object();
+
+ private final InputChannelID receiverId;
+
+ private final AtomicLong numBuffersAvailable = new AtomicLong();
+
+ private final PartitionRequestQueue requestQueue;
+
+ 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;
+
+ CreditBasedSequenceNumberingViewReader(
+ InputChannelID receiverId,
+ int initialCredit,
+ PartitionRequestQueue requestQueue) {
+
+ this.receiverId = receiverId;
+ this.numCreditsAvailable = initialCredit;
+ this.requestQueue = requestQueue;
+ }
+
+ @Override
+ public void requestSubpartitionView(
+ ResultPartitionProvider partitionProvider,
+ ResultPartitionID resultPartitionId,
+ int subPartitionIndex) throws IOException {
+
+ synchronized (requestLock) {
+ if (subpartitionView == null) {
+ // This this call can trigger a notification we have to
+ // schedule a separate task at the event loop that will
+ // start consuming this. Otherwise the reference to the
+ // view cannot be available in getNextBuffer().
+ this.subpartitionView = partitionProvider.createSubpartitionView(
+ resultPartitionId,
+ subPartitionIndex,
+ this);
+ } else {
+ throw new IllegalStateException("Subpartition already requested");
+ }
+ }
+ }
+
+ @Override
+ public void addCredit(int creditDeltas) {
+ numCreditsAvailable += creditDeltas;
+ }
+
+ @Override
+ public void setRegisteredAsAvailable(boolean isRegisteredAvailable) {
+ this.isRegisteredAsAvailable = isRegisteredAvailable;
+ }
+
+ @Override
+ public boolean isRegisteredAsAvailable() {
+ return isRegisteredAsAvailable;
+ }
+
+ /**
+ * Returns true only if the next buffer is an event or the reader has both available
+ * credits and buffers.
+ */
+ @Override
+ 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());
+ }
+
+ @Override
+ public InputChannelID getReceiverId() {
+ return receiverId;
+ }
+
+ @Override
+ public int getSequenceNumber() {
+ return sequenceNumber;
+ }
+
+ @VisibleForTesting
+ int getNumCreditsAvailable() {
+ return numCreditsAvailable;
+ }
+
+ @VisibleForTesting
+ long getNumBuffersAvailable() {
+ return numBuffersAvailable.get();
+ }
+
+ @Override
+ public BufferAndAvailability getNextBuffer() throws IOException, InterruptedException {
+ BufferAndBacklog next = subpartitionView.getNextBuffer();
+ if (next != null) {
+ long remaining = numBuffersAvailable.decrementAndGet();
+ sequenceNumber++;
+
+ 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;
+ }
+ }
+
+ @Override
+ public void notifySubpartitionConsumed() throws IOException {
+ subpartitionView.notifySubpartitionConsumed();
+ }
+
+ @Override
+ public boolean isReleased() {
+ return subpartitionView.isReleased();
+ }
+
+ @Override
+ public Throwable getFailureCause() {
+ return subpartitionView.getFailureCause();
+ }
+
+ @Override
+ public void releaseAllResources() throws IOException {
+ subpartitionView.releaseAllResources();
+ }
+
+ @Override
+ public void notifyBuffersAvailable(long numBuffers) {
+ // if this request made the channel non-empty, notify the input gate
+ if (numBuffers > 0 && numBuffersAvailable.getAndAdd(numBuffers) == 0) {
+ requestQueue.notifyReaderNonEmpty(this);
+ }
+ }
+
+ @Override
+ public String toString() {
+ return "CreditBasedSequenceNumberingViewReader{" +
+ "requestLock=" + requestLock +
+ ", receiverId=" + receiverId +
+ ", numBuffersAvailable=" + numBuffersAvailable.get() +
+ ", sequenceNumber=" + sequenceNumber +
+ ", numCreditsAvailable=" + numCreditsAvailable +
+ ", isRegisteredAsAvailable=" + isRegisteredAsAvailable +
+ '}';
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/0093bcbe/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/NettyConfig.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/NettyConfig.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/NettyConfig.java
index b5bebc7..2338b3d 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/NettyConfig.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/NettyConfig.java
@@ -224,6 +224,10 @@ public class NettyConfig {
SSLUtils.setSSLVerifyHostname(config, sslParams);
}
+ public boolean isCreditBasedEnabled() {
+ return config.getBoolean(TaskManagerOptions.NETWORK_CREDIT_BASED_FLOW_CONTROL_ENABLED);
+ }
+
@Override
public String toString() {
String format = "NettyConfig [" +
http://git-wip-us.apache.org/repos/asf/flink/blob/0093bcbe/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/NettyConnectionManager.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/NettyConnectionManager.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/NettyConnectionManager.java
index 1d98715..3fe15e5 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/NettyConnectionManager.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/NettyConnectionManager.java
@@ -44,10 +44,11 @@ public class NettyConnectionManager implements ConnectionManager {
}
@Override
- public void start(ResultPartitionProvider partitionProvider, TaskEventDispatcher taskEventDispatcher)
- throws IOException {
- NettyProtocol partitionRequestProtocol =
- new NettyProtocol(partitionProvider, taskEventDispatcher);
+ public void start(ResultPartitionProvider partitionProvider, TaskEventDispatcher taskEventDispatcher) throws IOException {
+ NettyProtocol partitionRequestProtocol = new NettyProtocol(
+ partitionProvider,
+ taskEventDispatcher,
+ client.getConfig().isCreditBasedEnabled());
client.init(partitionRequestProtocol, bufferPool);
server.init(partitionRequestProtocol, bufferPool);
http://git-wip-us.apache.org/repos/asf/flink/blob/0093bcbe/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/NettyProtocol.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/NettyProtocol.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/NettyProtocol.java
index 7de00e8..f09ddfb 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/NettyProtocol.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/NettyProtocol.java
@@ -18,6 +18,7 @@
package org.apache.flink.runtime.io.network.netty;
+import org.apache.flink.runtime.io.network.NetworkClientHandler;
import org.apache.flink.runtime.io.network.TaskEventDispatcher;
import org.apache.flink.runtime.io.network.partition.ResultPartitionProvider;
@@ -38,9 +39,12 @@ public class NettyProtocol {
private final ResultPartitionProvider partitionProvider;
private final TaskEventDispatcher taskEventDispatcher;
- NettyProtocol(ResultPartitionProvider partitionProvider, TaskEventDispatcher taskEventDispatcher) {
+ private final boolean creditBasedEnabled;
+
+ NettyProtocol(ResultPartitionProvider partitionProvider, TaskEventDispatcher taskEventDispatcher, boolean creditBasedEnabled) {
this.partitionProvider = partitionProvider;
this.taskEventDispatcher = taskEventDispatcher;
+ this.creditBasedEnabled = creditBasedEnabled;
}
/**
@@ -84,7 +88,7 @@ public class NettyProtocol {
public ChannelHandler[] getServerChannelHandlers() {
PartitionRequestQueue queueOfPartitionQueues = new PartitionRequestQueue();
PartitionRequestServerHandler serverHandler = new PartitionRequestServerHandler(
- partitionProvider, taskEventDispatcher, queueOfPartitionQueues);
+ partitionProvider, taskEventDispatcher, queueOfPartitionQueues, creditBasedEnabled);
return new ChannelHandler[] {
messageEncoder,
@@ -133,11 +137,14 @@ public class NettyProtocol {
* @return channel handlers
*/
public ChannelHandler[] getClientChannelHandlers() {
+ NetworkClientHandler networkClientHandler =
+ creditBasedEnabled ? new CreditBasedPartitionRequestClientHandler() :
+ new PartitionRequestClientHandler();
return new ChannelHandler[] {
messageEncoder,
createFrameLengthDecoder(),
messageDecoder,
- new PartitionRequestClientHandler()};
+ networkClientHandler};
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/0093bcbe/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 3f41646..91dc2d5 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
@@ -19,6 +19,7 @@
package org.apache.flink.runtime.io.network.netty;
import org.apache.flink.runtime.event.TaskEvent;
+import org.apache.flink.runtime.io.network.NetworkClientHandler;
import org.apache.flink.runtime.io.network.ConnectionID;
import org.apache.flink.runtime.io.network.netty.exception.LocalTransportException;
import org.apache.flink.runtime.io.network.partition.ResultPartitionID;
@@ -51,7 +52,7 @@ public class PartitionRequestClient {
private final Channel tcpChannel;
- private final PartitionRequestClientHandler partitionRequestHandler;
+ private final NetworkClientHandler clientHandler;
private final ConnectionID connectionId;
@@ -62,12 +63,12 @@ public class PartitionRequestClient {
PartitionRequestClient(
Channel tcpChannel,
- PartitionRequestClientHandler partitionRequestHandler,
+ NetworkClientHandler clientHandler,
ConnectionID connectionId,
PartitionRequestClientFactory clientFactory) {
this.tcpChannel = checkNotNull(tcpChannel);
- this.partitionRequestHandler = checkNotNull(partitionRequestHandler);
+ this.clientHandler = checkNotNull(clientHandler);
this.connectionId = checkNotNull(connectionId);
this.clientFactory = checkNotNull(clientFactory);
}
@@ -78,8 +79,8 @@ public class PartitionRequestClient {
/**
* Increments the reference counter.
- * <p>
- * Note: the reference counter has to be incremented before returning the
+ *
+ * <p>Note: the reference counter has to be incremented before returning the
* instance of this client to ensure correct closing logic.
*/
boolean incrementReferenceCounter() {
@@ -88,8 +89,8 @@ public class PartitionRequestClient {
/**
* Requests a remote intermediate result partition queue.
- * <p>
- * The request goes to the remote producer, for which this partition
+ *
+ * <p>The request goes to the remote producer, for which this partition
* request client instance has been created.
*/
public ChannelFuture requestSubpartition(
@@ -103,7 +104,7 @@ public class PartitionRequestClient {
LOG.debug("Requesting subpartition {} of partition {} with {} ms delay.",
subpartitionIndex, partitionId, delayMs);
- partitionRequestHandler.addInputChannel(inputChannel);
+ clientHandler.addInputChannel(inputChannel);
final PartitionRequest request = new PartitionRequest(
partitionId, subpartitionIndex, inputChannel.getInputChannelId(), inputChannel.getInitialCredit());
@@ -112,7 +113,7 @@ public class PartitionRequestClient {
@Override
public void operationComplete(ChannelFuture future) throws Exception {
if (!future.isSuccess()) {
- partitionRequestHandler.removeInputChannel(inputChannel);
+ clientHandler.removeInputChannel(inputChannel);
inputChannel.onError(
new LocalTransportException(
"Sending the partition request failed.",
@@ -126,8 +127,7 @@ public class PartitionRequestClient {
ChannelFuture f = tcpChannel.writeAndFlush(request);
f.addListener(listener);
return f;
- }
- else {
+ } else {
final ChannelFuture[] f = new ChannelFuture[1];
tcpChannel.eventLoop().schedule(new Runnable() {
@Override
@@ -170,13 +170,13 @@ public class PartitionRequestClient {
public void notifyCreditAvailable(RemoteInputChannel inputChannel) {
// We should skip the notification if the client is already closed.
if (!closeReferenceCounter.isDisposed()) {
- partitionRequestHandler.notifyCreditAvailable(inputChannel);
+ clientHandler.notifyCreditAvailable(inputChannel);
}
}
public void close(RemoteInputChannel inputChannel) throws IOException {
- partitionRequestHandler.removeInputChannel(inputChannel);
+ clientHandler.removeInputChannel(inputChannel);
if (closeReferenceCounter.decrement()) {
// Close the TCP connection. Send a close request msg to ensure
@@ -186,9 +186,8 @@ public class PartitionRequestClient {
// Make sure to remove the client from the factory
clientFactory.destroyPartitionRequestClient(connectionId, this);
- }
- else {
- partitionRequestHandler.cancelRequestFor(inputChannel.getInputChannelId());
+ } else {
+ clientHandler.cancelRequestFor(inputChannel.getInputChannelId());
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/0093bcbe/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestClientFactory.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestClientFactory.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestClientFactory.java
index c559ed8..ea9d3aa 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestClientFactory.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestClientFactory.java
@@ -18,6 +18,7 @@
package org.apache.flink.runtime.io.network.netty;
+import org.apache.flink.runtime.io.network.NetworkClientHandler;
import org.apache.flink.runtime.io.network.ConnectionID;
import org.apache.flink.runtime.io.network.netty.exception.LocalTransportException;
import org.apache.flink.runtime.io.network.netty.exception.RemoteTransportException;
@@ -164,11 +165,9 @@ class PartitionRequestClientFactory {
private void handInChannel(Channel channel) {
synchronized (connectLock) {
try {
- PartitionRequestClientHandler requestHandler = channel.pipeline()
- .get(PartitionRequestClientHandler.class);
-
+ NetworkClientHandler clientHandler = channel.pipeline().get(NetworkClientHandler.class);
partitionRequestClient = new PartitionRequestClient(
- channel, requestHandler, connectionId, clientFactory);
+ channel, clientHandler, connectionId, clientFactory);
if (disposeRequestClient) {
partitionRequestClient.disposeIfNotUsed();
http://git-wip-us.apache.org/repos/asf/flink/blob/0093bcbe/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 5299e22..2279deb 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
@@ -20,10 +20,12 @@ 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.NetworkClientHandler;
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;
@@ -31,10 +33,8 @@ 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;
@@ -44,45 +44,45 @@ 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.
+ * producer.
+ *
+ * <p>It is used in the old network mode.
*/
-class PartitionRequestClientHandler extends ChannelInboundHandlerAdapter {
+class PartitionRequestClientHandler extends ChannelInboundHandlerAdapter implements NetworkClientHandler {
private static final Logger LOG = LoggerFactory.getLogger(PartitionRequestClientHandler.class);
- /** Channels, which already requested partitions from the producers. */
- private final ConcurrentMap<InputChannelID, RemoteInputChannel> inputChannels = new ConcurrentHashMap<>();
+ private final ConcurrentMap<InputChannelID, RemoteInputChannel> inputChannels = new ConcurrentHashMap<InputChannelID, RemoteInputChannel>();
+
+ private final AtomicReference<Throwable> channelError = new AtomicReference<Throwable>();
- /** Channels, which will notify the producers about unannounced credit. */
- private final ArrayDeque<RemoteInputChannel> inputChannelsWithCredit = new ArrayDeque<>();
+ private final BufferListenerTask bufferListener = new BufferListenerTask();
- private final AtomicReference<Throwable> channelError = new AtomicReference<>();
+ private final Queue<Object> stagedMessages = new ArrayDeque<Object>();
- private final ChannelFutureListener writeListener = new WriteAndFlushNextMessageIfPossibleListener();
+ private final StagedMessagesHandlerTask stagedMessagesHandler = new StagedMessagesHandlerTask();
/**
* 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<>();
+ private final ConcurrentMap<InputChannelID, InputChannelID> cancelled = Maps.newConcurrentMap();
- /**
- * 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 {
+ @Override
+ public void addInputChannel(RemoteInputChannel listener) throws IOException {
checkError();
if (!inputChannels.containsKey(listener.getInputChannelId())) {
@@ -90,11 +90,13 @@ class PartitionRequestClientHandler extends ChannelInboundHandlerAdapter {
}
}
- void removeInputChannel(RemoteInputChannel listener) {
+ @Override
+ public void removeInputChannel(RemoteInputChannel listener) {
inputChannels.remove(listener.getInputChannelId());
}
- void cancelRequestFor(InputChannelID inputChannelId) {
+ @Override
+ public void cancelRequestFor(InputChannelID inputChannelId) {
if (inputChannelId == null || ctx == null) {
return;
}
@@ -104,20 +106,8 @@ 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);
- }
- });
+ @Override
+ public void notifyCreditAvailable(final RemoteInputChannel inputChannel) {
}
// ------------------------------------------------------------------------
@@ -141,8 +131,9 @@ 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);
@@ -151,22 +142,28 @@ 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);
}
@@ -177,36 +174,16 @@ class PartitionRequestClientHandler extends ChannelInboundHandlerAdapter {
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
try {
- 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());
+ if (!bufferListener.hasStagedBufferOrEvent() && stagedMessages.isEmpty()) {
+ decodeMsg(msg, false);
+ }
+ else {
+ stagedMessages.add(msg);
}
- } else {
- ctx.fireUserEventTriggered(msg);
}
- }
-
- @Override
- public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception {
- writeAndFlushNextMessageIfPossible(ctx.channel());
+ catch (Throwable t) {
+ notifyAllChannelsOfErrorAndClose(t);
+ }
}
private void notifyAllChannelsOfErrorAndClose(Throwable cause) {
@@ -215,12 +192,14 @@ class PartitionRequestClientHandler extends ChannelInboundHandlerAdapter {
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();
@@ -240,13 +219,19 @@ 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);
}
}
}
- private void decodeMsg(Object msg) throws Throwable {
+ @Override
+ public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
+ super.channelReadComplete(ctx);
+ }
+
+ private boolean decodeMsg(Object msg, boolean isStagedBuffer) throws Throwable {
final Class<?> msgClazz = msg.getClass();
// ---- Buffer --------------------------------------------------------
@@ -259,42 +244,47 @@ class PartitionRequestClientHandler extends ChannelInboundHandlerAdapter {
cancelRequestFor(bufferOrEvent.receiverId);
- return;
+ return true;
}
- decodeBufferOrEvent(inputChannel, bufferOrEvent);
-
- } else if (msgClazz == NettyMessage.ErrorResponse.class) {
- // ---- Error ---------------------------------------------------------
+ return decodeBufferOrEvent(inputChannel, bufferOrEvent, isStagedBuffer);
+ }
+ // ---- Error ---------------------------------------------------------
+ else if (msgClazz == NettyMessage.ErrorResponse.class) {
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 void decodeBufferOrEvent(RemoteInputChannel inputChannel, NettyMessage.BufferResponse bufferOrEvent) throws Throwable {
+ private boolean decodeBufferOrEvent(RemoteInputChannel inputChannel, NettyMessage.BufferResponse bufferOrEvent, boolean isStagedBuffer) throws Throwable {
+ boolean releaseNettyBuffer = true;
+
try {
ByteBuf nettyBuffer = bufferOrEvent.getNettyBuffer();
final int receivedSize = nettyBuffer.readableBytes();
@@ -304,21 +294,39 @@ class PartitionRequestClientHandler extends ChannelInboundHandlerAdapter {
// Early return for empty buffers. Otherwise Netty's readBytes() throws an
// IndexOutOfBoundsException.
if (receivedSize == 0) {
- inputChannel.onEmptyBuffer(bufferOrEvent.sequenceNumber, bufferOrEvent.backlog);
- return;
+ inputChannel.onEmptyBuffer(bufferOrEvent.sequenceNumber, -1);
+ return true;
}
- Buffer buffer = inputChannel.requestBuffer();
- if (buffer != null) {
- nettyBuffer.readBytes(buffer.asByteBuf(), receivedSize);
+ BufferProvider bufferProvider = inputChannel.getBufferProvider();
- inputChannel.onBuffer(buffer, bufferOrEvent.sequenceNumber, bufferOrEvent.backlog);
- } else if (inputChannel.isReleased()) {
+ if (bufferProvider == null) {
+ // receiver has been cancelled/failed
cancelRequestFor(bufferOrEvent.receiverId);
- } else {
- throw new IllegalStateException("No buffer available in credit-based input channel.");
+ return isStagedBuffer;
}
- } else {
+
+ 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 {
// ---- 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];
@@ -328,62 +336,184 @@ class PartitionRequestClientHandler extends ChannelInboundHandlerAdapter {
Buffer buffer = new NetworkBuffer(memSeg, FreeingBufferRecycler.INSTANCE, false);
buffer.setSize(receivedSize);
- inputChannel.onBuffer(buffer, bufferOrEvent.sequenceNumber, bufferOrEvent.backlog);
+ inputChannel.onBuffer(buffer, bufferOrEvent.sequenceNumber, -1);
+
+ return true;
}
- } finally {
- bufferOrEvent.releaseBuffer();
+ }
+ finally {
+ if (releaseNettyBuffer) {
+ bufferOrEvent.releaseBuffer();
+ }
+ }
+ }
+
+ private class AsyncErrorNotificationTask implements Runnable {
+
+ private final Throwable error;
+
+ public AsyncErrorNotificationTask(Throwable error) {
+ this.error = error;
+ }
+
+ @Override
+ public void run() {
+ notifyAllChannelsOfErrorAndClose(error);
}
}
/**
- * Tries to write&flush unannounced credits for the next input channel in queue.
+ * A buffer availability listener, which subscribes/unsubscribes the NIO
+ * read event.
+ *
+ * <p>If no buffer is available, the channel read event will be unsubscribed
+ * until one becomes available again.
*
- * <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.
+ * <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.
*/
- private void writeAndFlushNextMessageIfPossible(Channel channel) {
- if (channelError.get() != null || !channel.isWritable()) {
- return;
+ private class BufferListenerTask implements BufferListener, Runnable {
+
+ private final AtomicReference<Buffer> availableBuffer = new AtomicReference<Buffer>();
+
+ 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;
+ }
}
- while (true) {
- RemoteInputChannel inputChannel = inputChannelsWithCredit.poll();
+ private boolean hasStagedBufferOrEvent() {
+ return stagedBufferResponse != null;
+ }
- // The input channel may be null because of the write callbacks
- // that are executed after each write.
- if (inputChannel == null) {
- return;
+ 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);
+ }
+ }
+
+ // Called by the recycling thread (not network I/O thread)
+ @Override
+ public boolean notifyBufferAvailable(Buffer buffer) {
+ boolean success = false;
+
+ 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();
+ }
+ }
}
- //It is no need to notify credit for the released channel.
- if (!inputChannel.isReleased()) {
- AddCredit msg = new AddCredit(
- inputChannel.getPartitionId(),
- inputChannel.getAndResetUnannouncedCredit(),
- inputChannel.getInputChannelId());
+ 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);
- // Write and flush and wait until this is done before
- // trying to continue with the next input channel.
- channel.writeAndFlush(msg).addListener(writeListener);
+ if (inputChannel != null) {
+ inputChannel.onBuffer(buffer, stagedBufferResponse.sequenceNumber, -1);
+
+ success = true;
+ }
+ else {
+ cancelRequestFor(stagedBufferResponse.receiverId);
+ }
- return;
+ 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();
+ }
+ }
}
}
}
- private class WriteAndFlushNextMessageIfPossibleListener implements ChannelFutureListener {
+ public class StagedMessagesHandlerTask implements Runnable {
@Override
- public void operationComplete(ChannelFuture future) throws Exception {
+ public void run() {
try {
- if (future.isSuccess()) {
- writeAndFlushNextMessageIfPossible(future.channel());
- } else if (future.cause() != null) {
- notifyAllChannelsOfErrorAndClose(future.cause());
- } else {
- notifyAllChannelsOfErrorAndClose(new IllegalStateException("Sending cancelled by user."));
+ Object msg;
+ while ((msg = stagedMessages.poll()) != null) {
+ if (!decodeMsg(msg, true)) {
+ return;
+ }
}
- } catch (Throwable t) {
+
+ ctx.channel().config().setAutoRead(true);
+ ctx.channel().read();
+ }
+ catch (Throwable t) {
notifyAllChannelsOfErrorAndClose(t);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/0093bcbe/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 bacca9d..4832442 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
@@ -19,6 +19,7 @@
package org.apache.flink.runtime.io.network.netty;
import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.runtime.io.network.NetworkSequenceViewReader;
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;
@@ -56,10 +57,10 @@ class PartitionRequestQueue extends ChannelInboundHandlerAdapter {
private final ChannelFutureListener writeListener = new WriteAndFlushNextMessageIfPossibleListener();
/** The readers which are already enqueued available for transferring data. */
- private final ArrayDeque<SequenceNumberingViewReader> availableReaders = new ArrayDeque<>();
+ private final ArrayDeque<NetworkSequenceViewReader> availableReaders = new ArrayDeque<>();
/** All the readers created for the consumers' partition requests. */
- private final ConcurrentMap<InputChannelID, SequenceNumberingViewReader> allReaders = new ConcurrentHashMap<>();
+ private final ConcurrentMap<InputChannelID, NetworkSequenceViewReader> allReaders = new ConcurrentHashMap<>();
private final Set<InputChannelID> released = Sets.newHashSet();
@@ -76,7 +77,7 @@ class PartitionRequestQueue extends ChannelInboundHandlerAdapter {
super.channelRegistered(ctx);
}
- void notifyReaderNonEmpty(final SequenceNumberingViewReader reader) {
+ void notifyReaderNonEmpty(final NetworkSequenceViewReader reader) {
// The notification might come from the same thread. For the initial writes this
// might happen before the reader has set its reference to the view, because
// creating the queue and the initial notification happen in the same method call.
@@ -101,7 +102,7 @@ class PartitionRequestQueue extends ChannelInboundHandlerAdapter {
* <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 {
+ private void enqueueAvailableReader(final NetworkSequenceViewReader 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
@@ -125,11 +126,11 @@ class PartitionRequestQueue extends ChannelInboundHandlerAdapter {
* @return readers which are enqueued available for transferring data
*/
@VisibleForTesting
- ArrayDeque<SequenceNumberingViewReader> getAvailableReaders() {
+ ArrayDeque<NetworkSequenceViewReader> getAvailableReaders() {
return availableReaders;
}
- void notifyReaderCreated(final SequenceNumberingViewReader reader) {
+ public void notifyReaderCreated(final NetworkSequenceViewReader reader) {
allReaders.put(reader.getReceiverId(), reader);
}
@@ -155,7 +156,7 @@ class PartitionRequestQueue extends ChannelInboundHandlerAdapter {
return;
}
- SequenceNumberingViewReader reader = allReaders.get(receiverId);
+ NetworkSequenceViewReader reader = allReaders.get(receiverId);
if (reader != null) {
reader.addCredit(credit);
@@ -170,8 +171,8 @@ class PartitionRequestQueue extends ChannelInboundHandlerAdapter {
// 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) {
- enqueueAvailableReader((SequenceNumberingViewReader) msg);
+ if (msg instanceof NetworkSequenceViewReader) {
+ enqueueAvailableReader((NetworkSequenceViewReader) msg);
} else if (msg.getClass() == InputChannelID.class) {
// Release partition view that get a cancel request.
InputChannelID toCancel = (InputChannelID) msg;
@@ -182,7 +183,7 @@ class PartitionRequestQueue extends ChannelInboundHandlerAdapter {
// Cancel the request for the input channel
int size = availableReaders.size();
for (int i = 0; i < size; i++) {
- SequenceNumberingViewReader reader = availableReaders.poll();
+ NetworkSequenceViewReader reader = availableReaders.poll();
if (reader.getReceiverId().equals(toCancel)) {
reader.releaseAllResources();
reader.setRegisteredAsAvailable(false);
@@ -215,7 +216,7 @@ class PartitionRequestQueue extends ChannelInboundHandlerAdapter {
BufferAndAvailability next = null;
try {
while (true) {
- SequenceNumberingViewReader reader = availableReaders.poll();
+ NetworkSequenceViewReader reader = availableReaders.poll();
// No queue with available data. We allow this here, because
// of the write callbacks that are executed after each write.
@@ -312,7 +313,7 @@ class PartitionRequestQueue extends ChannelInboundHandlerAdapter {
private void releaseAllResources() throws IOException {
// note: this is only ever executed by one thread: the Netty IO thread!
- for (SequenceNumberingViewReader reader : allReaders.values()) {
+ for (NetworkSequenceViewReader reader : allReaders.values()) {
reader.releaseAllResources();
markAsReleased(reader.getReceiverId());
}
http://git-wip-us.apache.org/repos/asf/flink/blob/0093bcbe/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 997d74a..c6a8b1a 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
@@ -18,6 +18,7 @@
package org.apache.flink.runtime.io.network.netty;
+import org.apache.flink.runtime.io.network.NetworkSequenceViewReader;
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;
@@ -48,14 +49,18 @@ class PartitionRequestServerHandler extends SimpleChannelInboundHandler<NettyMes
private final PartitionRequestQueue outboundQueue;
+ private final boolean creditBasedEnabled;
+
PartitionRequestServerHandler(
ResultPartitionProvider partitionProvider,
TaskEventDispatcher taskEventDispatcher,
- PartitionRequestQueue outboundQueue) {
+ PartitionRequestQueue outboundQueue,
+ boolean creditBasedEnabled) {
this.partitionProvider = partitionProvider;
this.taskEventDispatcher = taskEventDispatcher;
this.outboundQueue = outboundQueue;
+ this.creditBasedEnabled = creditBasedEnabled;
}
@Override
@@ -82,10 +87,17 @@ class PartitionRequestServerHandler extends SimpleChannelInboundHandler<NettyMes
LOG.debug("Read channel on {}: {}.", ctx.channel().localAddress(), request);
try {
- SequenceNumberingViewReader reader = new SequenceNumberingViewReader(
- request.receiverId,
- request.credit,
- outboundQueue);
+ NetworkSequenceViewReader reader;
+ if (creditBasedEnabled) {
+ reader = new CreditBasedSequenceNumberingViewReader(
+ request.receiverId,
+ request.credit,
+ outboundQueue);
+ } else {
+ reader = new SequenceNumberingViewReader(
+ request.receiverId,
+ outboundQueue);
+ }
outboundQueue.notifyReaderCreated(reader);
http://git-wip-us.apache.org/repos/asf/flink/blob/0093bcbe/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 c8e97bd..0ec5fcb 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
@@ -18,8 +18,8 @@
package org.apache.flink.runtime.io.network.netty;
+import org.apache.flink.runtime.io.network.NetworkSequenceViewReader;
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;
@@ -32,13 +32,12 @@ import java.io.IOException;
import java.util.concurrent.atomic.AtomicLong;
/**
- * Simple wrapper for the partition readerQueue iterator, which increments a
- * sequence number for each returned buffer and remembers the receiver ID.
+ * Simple wrapper for the subpartition view used in the old network mode.
*
* <p>It also keeps track of available buffers and notifies the outbound
* handler about non-emptiness, similar to the {@link LocalInputChannel}.
*/
-class SequenceNumberingViewReader implements BufferAvailabilityListener {
+class SequenceNumberingViewReader implements BufferAvailabilityListener, NetworkSequenceViewReader {
private final Object requestLock = new Object();
@@ -50,31 +49,15 @@ 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,
- int initialCredit,
- PartitionRequestQueue requestQueue) {
-
+ SequenceNumberingViewReader(InputChannelID receiverId, PartitionRequestQueue requestQueue) {
this.receiverId = receiverId;
- this.numCreditsAvailable = initialCredit;
this.requestQueue = requestQueue;
}
- void requestSubpartitionView(
+ @Override
+ public void requestSubpartitionView(
ResultPartitionProvider partitionProvider,
ResultPartitionID resultPartitionId,
int subPartitionIndex) throws IOException {
@@ -95,108 +78,67 @@ class SequenceNumberingViewReader implements BufferAvailabilityListener {
}
}
- /**
- * The credits from consumer are added in incremental way.
- *
- * @param creditDeltas The credit deltas
- */
+ @Override
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.
- */
+ @Override
public void setRegisteredAsAvailable(boolean isRegisteredAvailable) {
- this.isRegisteredAsAvailable = isRegisteredAvailable;
}
+ @Override
public boolean isRegisteredAsAvailable() {
- return isRegisteredAsAvailable;
+ return false;
}
- /**
- * 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.
- */
+ @Override
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());
+ return true;
}
- InputChannelID getReceiverId() {
+ @Override
+ public InputChannelID getReceiverId() {
return receiverId;
}
- int getSequenceNumber() {
+ @Override
+ public int getSequenceNumber() {
return sequenceNumber;
}
- @VisibleForTesting
- int getNumCreditsAvailable() {
- return numCreditsAvailable;
- }
-
- @VisibleForTesting
- long getNumBuffersAvailable() {
- return numBuffersAvailable.get();
- }
-
+ @Override
public BufferAndAvailability getNextBuffer() throws IOException, InterruptedException {
BufferAndBacklog next = subpartitionView.getNextBuffer();
if (next != null) {
long remaining = numBuffersAvailable.decrementAndGet();
sequenceNumber++;
- if (remaining < 0) {
+ if (remaining >= 0) {
+ return new BufferAndAvailability(next.buffer(), remaining > 0, next.buffersInBacklog());
+ } else {
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;
}
}
+ @Override
public void notifySubpartitionConsumed() throws IOException {
subpartitionView.notifySubpartitionConsumed();
}
+ @Override
public boolean isReleased() {
return subpartitionView.isReleased();
}
+ @Override
public Throwable getFailureCause() {
return subpartitionView.getFailureCause();
}
+ @Override
public void releaseAllResources() throws IOException {
subpartitionView.releaseAllResources();
}
@@ -216,8 +158,6 @@ 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/0093bcbe/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerServices.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerServices.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerServices.java
index 33c8c28..b710b6a 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerServices.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerServices.java
@@ -34,6 +34,7 @@ import org.apache.flink.runtime.io.network.LocalConnectionManager;
import org.apache.flink.runtime.io.network.NetworkEnvironment;
import org.apache.flink.runtime.io.network.TaskEventDispatcher;
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;
import org.apache.flink.runtime.io.network.partition.ResultPartitionManager;
import org.apache.flink.runtime.memory.MemoryManager;
@@ -388,9 +389,11 @@ public class TaskManagerServices {
segmentSize);
ConnectionManager connectionManager;
-
- if (networkEnvironmentConfiguration.nettyConfig() != null) {
- connectionManager = new NettyConnectionManager(networkEnvironmentConfiguration.nettyConfig());
+ boolean enableCreditBased = false;
+ NettyConfig nettyConfig = networkEnvironmentConfiguration.nettyConfig();
+ if (nettyConfig != null) {
+ connectionManager = new NettyConnectionManager(nettyConfig);
+ enableCreditBased = nettyConfig.isCreditBasedEnabled();
} else {
connectionManager = new LocalConnectionManager();
}
@@ -442,7 +445,8 @@ public class TaskManagerServices {
networkEnvironmentConfiguration.partitionRequestInitialBackoff(),
networkEnvironmentConfiguration.partitionRequestMaxBackoff(),
networkEnvironmentConfiguration.networkBuffersPerChannel(),
- networkEnvironmentConfiguration.floatingNetworkBuffersPerGate());
+ networkEnvironmentConfiguration.floatingNetworkBuffersPerGate(),
+ enableCreditBased);
}
/**
http://git-wip-us.apache.org/repos/asf/flink/blob/0093bcbe/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 e3f1d89..24878ef 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
@@ -71,7 +71,8 @@ public class NetworkEnvironmentTest {
0,
0,
2,
- 8);
+ 8,
+ true);
// result partitions
ResultPartition rp1 = createResultPartition(ResultPartitionType.PIPELINED, 2);
http://git-wip-us.apache.org/repos/asf/flink/blob/0093bcbe/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 cf56851..09d8634 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
@@ -94,7 +94,7 @@ public class CancelPartitionRequestTest {
});
NettyProtocol protocol = new NettyProtocol(
- partitions, mock(TaskEventDispatcher.class));
+ partitions, mock(TaskEventDispatcher.class), true);
serverAndClient = initServerAndClient(protocol);
@@ -145,7 +145,7 @@ public class CancelPartitionRequestTest {
});
NettyProtocol protocol = new NettyProtocol(
- partitions, mock(TaskEventDispatcher.class));
+ partitions, mock(TaskEventDispatcher.class), true);
serverAndClient = initServerAndClient(protocol);