You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by uc...@apache.org on 2015/01/12 09:16:18 UTC
[10/18] flink git commit: [FLINK-986] [FLINK-25] [Distributed
runtime] Add initial support for intermediate results
http://git-wip-us.apache.org/repos/asf/flink/blob/d908ca19/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/NettyServer.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/NettyServer.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/NettyServer.java
new file mode 100644
index 0000000..0c857d1
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/NettyServer.java
@@ -0,0 +1,170 @@
+/*
+ * 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 com.google.common.util.concurrent.ThreadFactoryBuilder;
+import io.netty.bootstrap.ServerBootstrap;
+import io.netty.buffer.PooledByteBufAllocator;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.ChannelOption;
+import io.netty.channel.epoll.Epoll;
+import io.netty.channel.epoll.EpollEventLoopGroup;
+import io.netty.channel.epoll.EpollServerSocketChannel;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.channel.socket.nio.NioServerSocketChannel;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.concurrent.ThreadFactory;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+import static com.google.common.base.Preconditions.checkState;
+
+class NettyServer {
+
+ private static final ThreadFactoryBuilder THREAD_FACTORY_BUILDER = new ThreadFactoryBuilder().setDaemon(true);
+
+ private static final Logger LOG = LoggerFactory.getLogger(NettyServer.class);
+
+ private final NettyConfig config;
+
+ private ServerBootstrap bootstrap;
+
+ private ChannelFuture bindFuture;
+
+ NettyServer(NettyConfig config) {
+ this.config = checkNotNull(config);
+ }
+
+ void init(final NettyProtocol protocol) throws IOException {
+ checkState(bootstrap == null, "Netty server has already been initialized.");
+
+ long start = System.currentTimeMillis();
+
+ bootstrap = new ServerBootstrap();
+
+ // --------------------------------------------------------------------
+ // Transport-specific configuration
+ // --------------------------------------------------------------------
+
+ switch (config.getTransportType()) {
+ case NIO:
+ initNioBootstrap();
+ break;
+
+ case EPOLL:
+ initEpollBootstrap();
+ break;
+
+ case AUTO:
+ if (Epoll.isAvailable()) {
+ initEpollBootstrap();
+ LOG.info("Transport type 'auto': using EPOLL.");
+ }
+ else {
+ initNioBootstrap();
+ LOG.info("Transport type 'auto': using NIO.");
+ }
+ }
+
+ // --------------------------------------------------------------------
+ // Configuration
+ // --------------------------------------------------------------------
+
+ // Server bind address
+ bootstrap.localAddress(config.getServerAddress(), config.getServerPort());
+
+ // Pooled allocators for Netty's ByteBuf instances
+ bootstrap.option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
+ bootstrap.childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
+
+ if (config.getServerConnectBacklog() > 0) {
+ bootstrap.option(ChannelOption.SO_BACKLOG, config.getServerConnectBacklog());
+ }
+
+ // Receive and send buffer size
+ int receiveAndSendBufferSize = config.getSendAndReceiveBufferSize();
+ if (receiveAndSendBufferSize > 0) {
+ bootstrap.childOption(ChannelOption.SO_SNDBUF, receiveAndSendBufferSize);
+ bootstrap.childOption(ChannelOption.SO_RCVBUF, receiveAndSendBufferSize);
+ }
+
+ // --------------------------------------------------------------------
+ // Child channel pipeline for accepted connections
+ // --------------------------------------------------------------------
+
+ bootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
+ @Override
+ public void initChannel(SocketChannel channel) throws Exception {
+ protocol.setServerChannelPipeline(channel.pipeline());
+ }
+ });
+
+ // --------------------------------------------------------------------
+ // Start Server
+ // --------------------------------------------------------------------
+
+ bindFuture = bootstrap.bind().syncUninterruptibly();
+
+ long end = System.currentTimeMillis();
+ LOG.info("Successful initialization (took {} ms). Listening on SocketAddress {}.", (end - start), bindFuture.channel().localAddress().toString());
+ }
+
+ void shutdown() {
+ long start = System.currentTimeMillis();
+ if (bindFuture != null) {
+ bindFuture.channel().close().awaitUninterruptibly();
+ bindFuture = null;
+ }
+
+ if (bootstrap != null) {
+ if (bootstrap.group() != null) {
+ bootstrap.group().shutdownGracefully();
+ }
+ bootstrap = null;
+ }
+ long end = System.currentTimeMillis();
+ LOG.info("Successful shutdown (took {} ms).", (end - start));
+ }
+
+ private void initNioBootstrap() {
+ // Add the server port number to the name in order to distinguish
+ // multiple servers running on the same host.
+ String name = NettyConfig.SERVER_THREAD_GROUP_NAME + " (" + config.getServerPort() + ")";
+
+ NioEventLoopGroup nioGroup = new NioEventLoopGroup(config.getServerNumThreads(), getNamedThreadFactory(name));
+ bootstrap.group(nioGroup).channel(NioServerSocketChannel.class);
+ }
+
+ private void initEpollBootstrap() {
+ // Add the server port number to the name in order to distinguish
+ // multiple servers running on the same host.
+ String name = NettyConfig.SERVER_THREAD_GROUP_NAME + " (" + config.getServerPort() + ")";
+
+ EpollEventLoopGroup epollGroup = new EpollEventLoopGroup(config.getServerNumThreads(), getNamedThreadFactory(name));
+ bootstrap.group(epollGroup).channel(EpollServerSocketChannel.class);
+ }
+
+ public static ThreadFactory getNamedThreadFactory(String name) {
+ return THREAD_FACTORY_BUILDER.setNameFormat(name + " Thread %d").build();
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/d908ca19/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/OutboundConnectionQueue.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/OutboundConnectionQueue.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/OutboundConnectionQueue.java
deleted file mode 100644
index 850f38b..0000000
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/OutboundConnectionQueue.java
+++ /dev/null
@@ -1,153 +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 io.netty.channel.Channel;
-import io.netty.channel.ChannelFuture;
-import io.netty.channel.ChannelFutureListener;
-import io.netty.channel.ChannelHandlerContext;
-import io.netty.channel.ChannelInboundHandlerAdapter;
-import org.apache.flink.runtime.io.network.Buffer;
-import org.apache.flink.runtime.io.network.Envelope;
-import org.apache.flink.runtime.util.AtomicDisposableReferenceCounter;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.util.ArrayDeque;
-import java.util.concurrent.atomic.AtomicInteger;
-
-public class OutboundConnectionQueue extends ChannelInboundHandlerAdapter implements ChannelFutureListener {
-
- private static final Logger LOG = LoggerFactory.getLogger(OutboundConnectionQueue.class);
-
- private final Channel channel;
-
- private final ArrayDeque<Envelope> queuedEnvelopes = new ArrayDeque<Envelope>();
-
- private final AtomicInteger numQueuedEnvelopes = new AtomicInteger(0);
-
- private final AtomicDisposableReferenceCounter closeReferenceCounter = new AtomicDisposableReferenceCounter();
-
- public OutboundConnectionQueue(Channel channel) {
- this.channel = channel;
-
- channel.pipeline().addFirst(this);
- }
-
- boolean incrementReferenceCounter() {
- return closeReferenceCounter.incrementReferenceCounter();
- }
-
- boolean decrementReferenceCounter() {
- return closeReferenceCounter.decrementReferenceCounter();
- }
-
- void close() {
- channel.close();
- }
-
- /**
- * Enqueues an envelope to be sent later.
- * <p/>
- * This method is always invoked by the task thread that wants the envelope sent.
- *
- * @param env The envelope to be sent.
- */
- public void enqueue(Envelope env) {
- // the user event trigger ensure thread-safe hand-over of the envelope
- this.channel.pipeline().fireUserEventTriggered(env);
- }
-
- @Override
- public void channelInactive(ChannelHandlerContext ctx) throws Exception {
- releaseAllEnvelopes();
-
- super.channelInactive(ctx);
- }
-
- @Override
- public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
- releaseAllEnvelopes();
-
- super.exceptionCaught(ctx, cause);
- }
-
- @Override
- public void userEventTriggered(ChannelHandlerContext ctx, Object envelopeToEnqueue) throws Exception {
- boolean triggerWrite = this.queuedEnvelopes.isEmpty();
-
- this.queuedEnvelopes.addLast((Envelope) envelopeToEnqueue);
- this.numQueuedEnvelopes.incrementAndGet();
-
- if (triggerWrite) {
- writeAndFlushNextEnvelopeIfPossible();
- }
- }
-
- @Override
- public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception {
- writeAndFlushNextEnvelopeIfPossible();
- }
-
- @Override
- public void operationComplete(ChannelFuture future) throws Exception {
- if (future.isSuccess()) {
- writeAndFlushNextEnvelopeIfPossible();
- }
- else if (future.cause() != null) {
- exceptionOccurred(future.cause());
- }
- else {
- exceptionOccurred(new Exception("Envelope send aborted."));
- }
- }
-
- public int getNumQueuedEnvelopes() {
- return this.numQueuedEnvelopes.intValue();
- }
-
- @Override
- public String toString() {
- return channel.toString();
- }
-
- private void writeAndFlushNextEnvelopeIfPossible() {
- if (this.channel.isWritable() && !this.queuedEnvelopes.isEmpty()) {
- Envelope nextEnvelope = this.queuedEnvelopes.pollFirst();
- this.numQueuedEnvelopes.decrementAndGet();
-
- this.channel.writeAndFlush(nextEnvelope).addListener(this);
- }
- }
-
- private void exceptionOccurred(Throwable t) throws Exception {
- LOG.error("An exception occurred in Channel {}: {}", channel, t.getMessage());
- throw new Exception(t);
- }
-
- private void releaseAllEnvelopes() {
- Envelope envelope;
- while ((envelope = queuedEnvelopes.poll()) != null) {
- Buffer buffer = envelope.getBuffer();
- if (buffer != null) {
- buffer.recycleBuffer();
- }
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/d908ca19/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/OutboundEnvelopeEncoder.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/OutboundEnvelopeEncoder.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/OutboundEnvelopeEncoder.java
deleted file mode 100644
index 4dfe197..0000000
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/OutboundEnvelopeEncoder.java
+++ /dev/null
@@ -1,90 +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.runtime.io.network.Buffer;
-import org.apache.flink.runtime.io.network.Envelope;
-
-import io.netty.buffer.ByteBuf;
-import io.netty.buffer.Unpooled;
-import io.netty.channel.ChannelHandler;
-import io.netty.channel.ChannelHandlerContext;
-import io.netty.channel.ChannelOutboundHandlerAdapter;
-import io.netty.channel.ChannelPromise;
-
-@ChannelHandler.Sharable
-public class OutboundEnvelopeEncoder extends ChannelOutboundHandlerAdapter {
-
- public static final int HEADER_SIZE = 48;
-
- public static final int MAGIC_NUMBER = 0xBADC0FFE;
-
- @Override
- public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
- Envelope env = (Envelope) msg;
-
- ByteBuf buf = ctx.alloc().directBuffer();
-
- encode(env, buf);
-
- if (buf.isReadable()) {
- ctx.write(buf, promise);
- }
- else {
- buf.release();
- ctx.write(Unpooled.EMPTY_BUFFER, promise);
- }
- }
-
- private void encode(Envelope env, ByteBuf out) {
- // --------------------------------------------------------------------
- // (1) header (48 bytes)
- // --------------------------------------------------------------------
- out.writeInt(MAGIC_NUMBER); // 4 bytes
-
- if (out.getInt(out.writerIndex() - 4) != MAGIC_NUMBER) {
- throw new RuntimeException();
- }
-
- out.writeInt(env.getSequenceNumber()); // 4 bytes
- env.getJobID().writeTo(out); // 16 bytes
- env.getSource().writeTo(out); // 16 bytes
- out.writeInt(env.getEventsSerialized() != null ? env.getEventsSerialized().remaining() : 0); // 4 bytes
- out.writeInt(env.getBuffer() != null ? env.getBuffer().size() : 0); // 4 bytes
- // --------------------------------------------------------------------
- // (2) events (var length)
- // --------------------------------------------------------------------
- if (env.getEventsSerialized() != null) {
- out.writeBytes(env.getEventsSerialized());
- }
-
- // --------------------------------------------------------------------
- // (3) buffer (var length)
- // --------------------------------------------------------------------
- if (env.getBuffer() != null) {
- Buffer envBuffer = env.getBuffer();
- out.writeBytes(envBuffer.getMemorySegment().wrap(0, envBuffer.size()));
-
- // Recycle the buffer from OUR buffer pool after everything has been
- // copied to Nettys buffer space.
- envBuffer.recycleBuffer();
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/d908ca19/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
new file mode 100644
index 0000000..25021e2
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestClient.java
@@ -0,0 +1,130 @@
+/*
+ * 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 io.netty.channel.Channel;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelFutureListener;
+import org.apache.flink.runtime.event.task.TaskEvent;
+import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
+import org.apache.flink.runtime.io.network.RemoteAddress;
+import org.apache.flink.runtime.io.network.partition.consumer.RemoteInputChannel;
+import org.apache.flink.runtime.jobgraph.IntermediateResultPartitionID;
+import org.apache.flink.runtime.util.AtomicDisposableReferenceCounter;
+
+import java.io.IOException;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+import static org.apache.flink.runtime.io.network.netty.NettyMessage.PartitionRequest;
+import static org.apache.flink.runtime.io.network.netty.NettyMessage.TaskEventRequest;
+
+/**
+ * Partition request client for remote partition requests.
+ * <p>
+ * This client is shared by all remote input channels, which request a partition
+ * from the same {@link RemoteAddress}.
+ */
+public class PartitionRequestClient {
+
+ private final Channel tcpChannel;
+
+ private final PartitionRequestClientHandler partitionRequestHandler;
+
+ private final RemoteAddress remoteAddress;
+
+ private final PartitionRequestClientFactory clientFactory;
+
+ // If zero, the underlying TCP channel can be safely closed
+ private final AtomicDisposableReferenceCounter closeReferenceCounter = new AtomicDisposableReferenceCounter();
+
+ PartitionRequestClient(Channel tcpChannel, PartitionRequestClientHandler partitionRequestHandler, RemoteAddress remoteAddress, PartitionRequestClientFactory clientFactory) {
+ this.tcpChannel = checkNotNull(tcpChannel);
+ this.partitionRequestHandler = checkNotNull(partitionRequestHandler);
+ this.remoteAddress = checkNotNull(remoteAddress);
+ this.clientFactory = checkNotNull(clientFactory);
+ }
+
+ /**
+ * Increments the reference counter.
+ * <p>
+ * Note: the reference counter has to be incremented before returning the
+ * instance of this client to ensure correct closing logic.
+ */
+ boolean incrementReferenceCounter() {
+ return closeReferenceCounter.incrementReferenceCounter();
+ }
+
+ /**
+ * Requests a remote intermediate result partition queue.
+ * <p>
+ * The request goes to the remote producer, for which this partition
+ * request client instance has been created.
+ */
+ public void requestIntermediateResultPartition(ExecutionAttemptID producerExecutionId, final IntermediateResultPartitionID partitionId, final int requestedQueueIndex, final RemoteInputChannel inputChannel) throws IOException {
+ partitionRequestHandler.addInputChannel(inputChannel);
+
+ tcpChannel.writeAndFlush(new PartitionRequest(producerExecutionId, partitionId, requestedQueueIndex, inputChannel.getInputChannelId()))
+ .addListener(
+ new ChannelFutureListener() {
+ @Override
+ public void operationComplete(ChannelFuture future) throws Exception {
+ if (!future.isSuccess()) {
+ partitionRequestHandler.removeInputChannel(inputChannel);
+ inputChannel.onError(future.cause());
+ }
+ }
+ }
+ );
+ }
+
+ /**
+ * Sends a task event backwards to an intermediate result partition producer.
+ * <p>
+ * Backwards task events flow between readers and writers and therefore
+ * will only work when both are running at the same time, which is only
+ * guaranteed to be the case when both the respective producer and
+ * consumer task run pipelined.
+ */
+ public void sendTaskEvent(ExecutionAttemptID producerExecutionId, IntermediateResultPartitionID partitionId, TaskEvent event, final RemoteInputChannel inputChannel) throws IOException {
+
+ tcpChannel.writeAndFlush(new TaskEventRequest(event, producerExecutionId, partitionId, inputChannel.getInputChannelId()))
+ .addListener(
+ new ChannelFutureListener() {
+ @Override
+ public void operationComplete(ChannelFuture future) throws Exception {
+ if (!future.isSuccess()) {
+ inputChannel.onError(future.cause());
+ }
+ }
+ });
+ }
+
+ public void close(RemoteInputChannel inputChannel) throws IOException {
+
+ partitionRequestHandler.removeInputChannel(inputChannel);
+
+ if (closeReferenceCounter.decrementReferenceCounter()) {
+ // Close the TCP connection
+ tcpChannel.close();
+
+ // Make sure to remove the client from the factory
+ clientFactory.destroyPartitionRequestClient(remoteAddress, this);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/d908ca19/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
new file mode 100644
index 0000000..dd86004
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestClientFactory.java
@@ -0,0 +1,182 @@
+/*
+ * 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 io.netty.channel.Channel;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelFutureListener;
+import org.apache.flink.runtime.io.network.RemoteAddress;
+import org.apache.flink.runtime.io.network.partition.consumer.RemoteInputChannel;
+
+import java.io.IOException;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+
+/**
+ * Factory for {@link PartitionRequestClient} instances.
+ * <p>
+ * Instances of partition requests clients are shared among several {@link RemoteInputChannel}
+ * instances.
+ */
+class PartitionRequestClientFactory {
+
+ private final NettyClient nettyClient;
+
+ private final ConcurrentMap<RemoteAddress, Object> clients = new ConcurrentHashMap<RemoteAddress, Object>();
+
+ PartitionRequestClientFactory(NettyClient nettyClient) {
+ this.nettyClient = nettyClient;
+ }
+
+ /**
+ * Atomically establishes a TCP connection to the given remote address and
+ * creates a {@link PartitionRequestClient} instance for this connection.
+ */
+ PartitionRequestClient createPartitionRequestClient(RemoteAddress remoteAddress) throws IOException {
+ final Object entry = clients.get(remoteAddress);
+
+ final PartitionRequestClient client;
+ if (entry != null) {
+ // Existing channel or connecting channel
+ if (entry instanceof PartitionRequestClient) {
+ client = (PartitionRequestClient) entry;
+ }
+ else {
+ ConnectingChannel future = (ConnectingChannel) entry;
+ client = future.waitForChannel();
+ }
+ }
+ else {
+ // No channel yet. Create one, but watch out for a race.
+ // We create a "connecting future" and atomically add it to the map.
+ // Only the thread that really added it establishes the channel.
+ // The others need to wait on that original establisher's future.
+ ConnectingChannel connectingChannel = new ConnectingChannel(remoteAddress, this);
+ Object old = clients.putIfAbsent(remoteAddress, connectingChannel);
+
+ if (old == null) {
+ nettyClient.connect(remoteAddress.getAddress()).addListener(connectingChannel);
+
+ client = connectingChannel.waitForChannel();
+
+ Object previous = clients.put(remoteAddress, client);
+
+ if (connectingChannel != previous) {
+ throw new IOException("Race condition while establishing channel connection.");
+ }
+ }
+ else if (old instanceof ConnectingChannel) {
+ client = ((ConnectingChannel) old).waitForChannel();
+ }
+ else {
+ client = (PartitionRequestClient) old;
+ }
+ }
+
+ // Make sure to increment the reference count before handing a client
+ // out to ensure correct bookkeeping for channel closing.
+ if (client.incrementReferenceCounter()) {
+ return client;
+ }
+ else {
+ // There was a race with a close, try again.
+ destroyPartitionRequestClient(remoteAddress, client);
+
+ return createPartitionRequestClient(remoteAddress);
+ }
+ }
+
+ int getNumberOfActiveClients() {
+ return clients.size();
+ }
+
+ /**
+ * Removes the client for the given {@link RemoteAddress}.
+ */
+ void destroyPartitionRequestClient(RemoteAddress remoteAddress, PartitionRequestClient client) {
+ clients.remove(remoteAddress, client);
+ }
+
+ private static final class ConnectingChannel implements ChannelFutureListener {
+
+ private final Object connectLock = new Object();
+
+ private final RemoteAddress remoteAddress;
+
+ private final PartitionRequestClientFactory clientFactory;
+
+ public ConnectingChannel(RemoteAddress remoteAddress, PartitionRequestClientFactory clientFactory) {
+ this.remoteAddress = remoteAddress;
+ this.clientFactory = clientFactory;
+ }
+
+ private void handInChannel(Channel channel) {
+ synchronized (connectLock) {
+ PartitionRequestClientHandler requestHandler =
+ (PartitionRequestClientHandler) channel.pipeline().get(PartitionRequestProtocol.CLIENT_REQUEST_HANDLER_NAME);
+
+ partitionRequestClient = new PartitionRequestClient(channel, requestHandler, remoteAddress, clientFactory);
+ connectLock.notifyAll();
+ }
+ }
+
+ private volatile PartitionRequestClient partitionRequestClient;
+
+ private volatile Throwable error;
+
+ private PartitionRequestClient waitForChannel() throws IOException {
+ synchronized (connectLock) {
+ while (error == null && partitionRequestClient == null) {
+ try {
+ connectLock.wait(2000);
+ }
+ catch (InterruptedException e) {
+ throw new RuntimeException("Wait for channel connection interrupted.");
+ }
+ }
+ }
+
+ if (error != null) {
+ throw new IOException("Connecting the channel failed: " + error.getMessage(), error);
+ }
+
+ return partitionRequestClient;
+ }
+
+ private void notifyOfError(Throwable error) {
+ synchronized (connectLock) {
+ this.error = error;
+ connectLock.notifyAll();
+ }
+ }
+
+ @Override
+ public void operationComplete(ChannelFuture future) throws Exception {
+ if (future.isSuccess()) {
+ handInChannel(future.channel());
+ }
+ else if (future.cause() != null) {
+ notifyOfError(future.cause());
+ }
+ else {
+ notifyOfError(new IllegalStateException("Connecting the channel has been cancelled."));
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/d908ca19/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
new file mode 100644
index 0000000..b1aaf5b
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestClientHandler.java
@@ -0,0 +1,430 @@
+/*
+ * 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 io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+import org.apache.flink.core.memory.MemorySegment;
+import org.apache.flink.runtime.io.network.api.serialization.EventSerializer;
+import org.apache.flink.runtime.io.network.buffer.Buffer;
+import org.apache.flink.runtime.io.network.buffer.BufferProvider;
+import org.apache.flink.runtime.io.network.partition.consumer.InputChannelID;
+import org.apache.flink.runtime.io.network.partition.consumer.RemoteInputChannel;
+import org.apache.flink.runtime.util.event.EventListener;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayDeque;
+import java.util.Queue;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+import static com.google.common.base.Preconditions.checkState;
+
+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 AtomicBoolean channelError = new AtomicBoolean(false);
+
+ private final BufferListenerTask bufferListener = new BufferListenerTask();
+
+ private final Queue<Object> stagedMessages = new ArrayDeque<Object>();
+
+ private final StagedMessagesHandlerTask stagedMessagesHandler = new StagedMessagesHandlerTask();
+
+ private ChannelHandlerContext ctx;
+
+ private ScheduledFuture<?> logOutputTask;
+
+ // ------------------------------------------------------------------------
+ // Input channel/receiver registration
+ // ------------------------------------------------------------------------
+
+ void addInputChannel(RemoteInputChannel listener) {
+ checkState(!channelError.get(), "There has been an error in the channel.");
+
+ inputChannels.put(listener.getInputChannelId(), listener);
+ }
+
+ void removeInputChannel(RemoteInputChannel listener) {
+ inputChannels.remove(listener.getInputChannelId());
+ }
+
+ // ------------------------------------------------------------------------
+ // Network events
+ // ------------------------------------------------------------------------
+
+ @Override
+ public void channelActive(final ChannelHandlerContext ctx) throws Exception {
+ if (this.ctx == null) {
+ this.ctx = ctx;
+ }
+
+ if (LOG.isDebugEnabled()) {
+ logOutputTask = ctx.channel().eventLoop().scheduleWithFixedDelay(new DebugOutputTask(), 30, 30, TimeUnit.SECONDS);
+ }
+
+ super.channelActive(ctx);
+ }
+
+ @Override
+ public void channelInactive(ChannelHandlerContext ctx) throws Exception {
+ if (logOutputTask != null) {
+ logOutputTask.cancel(true);
+ }
+
+ super.channelActive(ctx);
+ }
+
+ @Override
+ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
+ if (logOutputTask != null) {
+ logOutputTask.cancel(true);
+ }
+
+ notifyAllChannelsOfErrorAndClose(cause);
+
+ super.exceptionCaught(ctx, cause);
+ }
+
+ @Override
+ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
+ try {
+ if (!bufferListener.hasStagedBufferOrEvent() && stagedMessages.isEmpty()) {
+ decodeMsg(msg);
+ }
+ else {
+ stagedMessages.add(msg);
+ }
+ }
+ catch (Throwable t) {
+ notifyAllChannelsOfErrorAndClose(t);
+ }
+ }
+
+ private void notifyAllChannelsOfErrorAndClose(Throwable cause) {
+ if (channelError.compareAndSet(false, true)) {
+ for (RemoteInputChannel inputChannel : inputChannels.values()) {
+ inputChannel.onError(cause);
+ }
+
+ inputChannels.clear();
+ ctx.close();
+ }
+ }
+
+ // ------------------------------------------------------------------------
+
+ @Override
+ public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
+ super.channelReadComplete(ctx);
+ }
+
+ private boolean 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();
+ return true;
+ }
+
+ return decodeBufferOrEvent(inputChannel, bufferOrEvent);
+ }
+ // ---- Error ---------------------------------------------------------
+ else if (msgClazz == NettyMessage.ErrorResponse.class) {
+ NettyMessage.ErrorResponse error = (NettyMessage.ErrorResponse) msg;
+
+ if (error.isFatalError()) {
+ notifyAllChannelsOfErrorAndClose(error.error);
+ }
+ else {
+ RemoteInputChannel inputChannel = inputChannels.get(error.receiverId);
+
+ if (inputChannel != null) {
+ inputChannel.onError(error.error);
+ }
+ }
+ }
+ else {
+ throw new IllegalStateException("Received unknown message from producer: " + msg.getClass());
+ }
+
+ return true;
+ }
+
+ private boolean decodeBufferOrEvent(RemoteInputChannel inputChannel, NettyMessage.BufferResponse bufferOrEvent) throws Throwable {
+ boolean releaseNettyBuffer = true;
+
+ try {
+ if (bufferOrEvent.isBuffer()) {
+ // ---- Buffer ------------------------------------------------
+ BufferProvider bufferProvider = inputChannel.getBufferProvider();
+
+ if (bufferProvider == null) {
+ return false; // receiver has been cancelled/failed
+ }
+
+ while (true) {
+ Buffer buffer = bufferProvider.requestBuffer();
+
+ if (buffer != null) {
+ buffer.setSize(bufferOrEvent.getSize());
+ bufferOrEvent.getNettyBuffer().readBytes(buffer.getNioBuffer());
+
+ inputChannel.onBuffer(buffer, bufferOrEvent.sequenceNumber);
+
+ return true;
+ }
+ else if (bufferListener.waitForBuffer(bufferProvider, bufferOrEvent)) {
+ releaseNettyBuffer = false;
+
+ return false;
+ }
+ }
+ }
+ 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[bufferOrEvent.getSize()];
+ bufferOrEvent.getNettyBuffer().readBytes(byteArray);
+
+ Buffer buffer = new Buffer(new MemorySegment(byteArray), EventSerializer.RECYCLER, false);
+
+ inputChannel.onBuffer(buffer, bufferOrEvent.sequenceNumber);
+
+ return true;
+ }
+ }
+ 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);
+ }
+ }
+
+ /**
+ * 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>
+ * After a buffer becomes available again, the buffer is handed over by
+ * the thread calling {@link #onEvent(Buffer)} to the network I/O
+ * thread, which then continues the processing of the staged buffer.
+ */
+ private class BufferListenerTask implements EventListener<Buffer>, 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.addListener(this)) {
+ if (ctx.channel().config().isAutoRead()) {
+ ctx.channel().config().setAutoRead(false);
+ }
+
+ return true;
+ }
+ else {
+ stagedBufferResponse = null;
+
+ return false;
+ }
+ }
+
+ private boolean hasStagedBufferOrEvent() {
+ return stagedBufferResponse != null;
+ }
+
+ // Called by the recycling thread (not network I/O thread)
+ @Override
+ public void onEvent(Buffer buffer) {
+ boolean success = false;
+
+ try {
+ checkNotNull(buffer, "Buffer request could not be satisfied.");
+
+ 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.recycle();
+ }
+ }
+ }
+ }
+
+ /**
+ * 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.");
+ }
+
+ buffer.setSize(stagedBufferResponse.getSize());
+
+ stagedBufferResponse.getNettyBuffer().readBytes(buffer.getNioBuffer());
+ stagedBufferResponse.releaseBuffer();
+
+ RemoteInputChannel inputChannel = inputChannels.get(stagedBufferResponse.receiverId);
+
+ if (inputChannel != null) {
+ inputChannel.onBuffer(buffer, stagedBufferResponse.sequenceNumber);
+
+ success = true;
+ }
+
+ 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.recycle();
+ }
+ }
+ }
+ }
+ }
+
+ public class StagedMessagesHandlerTask implements Runnable {
+
+ @Override
+ public void run() {
+ try {
+ Object msg;
+ while ((msg = stagedMessages.poll()) != null) {
+ if (!decodeMsg(msg)) {
+ return;
+ }
+ }
+
+ ctx.channel().config().setAutoRead(true);
+ ctx.channel().read();
+ }
+ catch (Throwable t) {
+ notifyAllChannelsOfErrorAndClose(t);
+ }
+ }
+ }
+
+ // ------------------------------------------------------------------------
+
+ /**
+ * Debug output task executed periodically by the network I/O thread.
+ */
+ private class DebugOutputTask implements Runnable {
+
+ @Override
+ public void run() {
+ StringBuilder str = new StringBuilder();
+
+ str.append("Channel remote address: ");
+ str.append(ctx.channel().remoteAddress());
+ str.append(". ");
+
+ str.append("Channel active: ");
+ str.append(ctx.channel().isActive());
+ str.append(". ");
+
+ str.append("Number of registered input channels: ");
+ str.append(inputChannels.size());
+ str.append(". ");
+
+ str.append("Has staged buffer or event: ");
+ str.append(bufferListener.hasStagedBufferOrEvent());
+ str.append(". ");
+
+ str.append("Total number of staged messages: ");
+ str.append(stagedMessages.size());
+ str.append(". ");
+
+ str.append("Channel auto read? ");
+ str.append(ctx.channel().config().isAutoRead());
+ str.append(". ");
+
+ LOG.debug(str.toString());
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/d908ca19/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestProtocol.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestProtocol.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestProtocol.java
new file mode 100644
index 0000000..b8fbc10
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestProtocol.java
@@ -0,0 +1,126 @@
+/**
+ * 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 io.netty.channel.ChannelPipeline;
+import org.apache.flink.runtime.io.network.TaskEventDispatcher;
+import org.apache.flink.runtime.io.network.partition.IntermediateResultPartitionProvider;
+
+import static org.apache.flink.runtime.io.network.netty.NettyMessage.NettyMessageEncoder;
+import static org.apache.flink.runtime.io.network.netty.NettyMessage.NettyMessageEncoder.createFrameLengthDecoder;
+
+class PartitionRequestProtocol implements NettyProtocol {
+
+ static final String CLIENT_REQUEST_HANDLER_NAME = "Client request handler";
+
+ private final NettyMessageEncoder messageEncoder = new NettyMessageEncoder();
+
+ private final NettyMessage.NettyMessageDecoder messageDecoder = new NettyMessage.NettyMessageDecoder();
+
+ private final IntermediateResultPartitionProvider partitionProvider;
+ private final TaskEventDispatcher taskEventDispatcher;
+
+ PartitionRequestProtocol(IntermediateResultPartitionProvider partitionProvider, TaskEventDispatcher taskEventDispatcher) {
+ this.partitionProvider = partitionProvider;
+ this.taskEventDispatcher = taskEventDispatcher;
+ }
+
+ // +-------------------------------------------------------------------+
+ // | SERVER CHANNEL PIPELINE |
+ // | |
+ // | +----------+----------+ (3) write +----------------------+ |
+ // | | Queue of queues +----------->| Message encoder | |
+ // | +----------+----------+ +-----------+----------+ |
+ // | /|\ \|/ |
+ // | | (2) enqueue | |
+ // | +----------+----------+ | |
+ // | | Request handler | | |
+ // | +----------+----------+ | |
+ // | /|\ | |
+ // | | | |
+ // | +----------+----------+ | |
+ // | | Message decoder | | |
+ // | +----------+----------+ | |
+ // | /|\ | |
+ // | | | |
+ // | +----------+----------+ | |
+ // | | Frame decoder | | |
+ // | +----------+----------+ |
+ // | /|\ | |
+ // +---------------+-----------------------------------+---------------+
+ // | | (1) client request \|/
+ // +---------------+-----------------------------------+---------------+
+ // | | | |
+ // | [ Socket.read() ] [ Socket.write() ] |
+ // | |
+ // | Netty Internal I/O Threads (Transport Implementation) |
+ // +-------------------------------------------------------------------+
+
+ @Override
+ public void setServerChannelPipeline(ChannelPipeline channelPipeline) {
+ PartitionRequestQueue queueOfPartitionQueues = new PartitionRequestQueue();
+
+ channelPipeline
+ .addLast("Message encoder", messageEncoder)
+ .addLast("Frame decoder", createFrameLengthDecoder())
+ .addLast("Client request decoder", messageDecoder)
+ .addLast("Server request handler", new PartitionRequestServerHandler(partitionProvider, taskEventDispatcher, queueOfPartitionQueues))
+ .addLast("Queue of queues", queueOfPartitionQueues);
+ }
+
+
+ // +-----------+----------+ +----------------------+
+ // | Remote input channel | | request client |
+ // +-----------+----------+ +-----------+----------+
+ // | | (1) write
+ // +---------------+-----------------------------------+---------------+
+ // | | CLIENT CHANNEL PIPELINE | |
+ // | | \|/ |
+ // | +----------+----------+ +----------------------+ |
+ // | | Request handler + | Message encoder | |
+ // | +----------+----------+ +-----------+----------+ |
+ // | /|\ \|/ |
+ // | | | |
+ // | +----------+----------+ | |
+ // | | Message decoder | | |
+ // | +----------+----------+ | |
+ // | /|\ | |
+ // | | | |
+ // | +----------+----------+ | |
+ // | | Frame decoder | | |
+ // | +----------+----------+ | |
+ // | /|\ | |
+ // +---------------+-----------------------------------+---------------+
+ // | | (3) server response \|/ (2) client request
+ // +---------------+-----------------------------------+---------------+
+ // | | | |
+ // | [ Socket.read() ] [ Socket.write() ] |
+ // | |
+ // | Netty Internal I/O Threads (Transport Implementation) |
+ // +-------------------------------------------------------------------+
+
+ @Override
+ public void setClientChannelPipeline(ChannelPipeline channelPipeline) {
+ channelPipeline
+ .addLast("Message encoder", messageEncoder)
+ .addLast("Frame decoder", createFrameLengthDecoder())
+ .addLast("Server response decoder", messageDecoder)
+ .addLast(CLIENT_REQUEST_HANDLER_NAME, new PartitionRequestClientHandler());
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/d908ca19/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
new file mode 100644
index 0000000..d107d95
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestQueue.java
@@ -0,0 +1,346 @@
+/*
+ * 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 io.netty.channel.Channel;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelFutureListener;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+import org.apache.flink.runtime.io.network.buffer.Buffer;
+import org.apache.flink.runtime.io.network.partition.queue.IntermediateResultPartitionQueueIterator;
+import org.apache.flink.runtime.io.network.partition.consumer.InputChannelID;
+import org.apache.flink.runtime.util.event.NotificationListener;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayDeque;
+import java.util.Queue;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+import static org.apache.flink.runtime.io.network.netty.NettyMessage.BufferResponse;
+
+/**
+ * A queue of partition queues, which listens for channel writability changed
+ * events before writing and flushing {@link Buffer} instances.
+ */
+class PartitionRequestQueue extends ChannelInboundHandlerAdapter {
+
+ private final Logger LOG = LoggerFactory.getLogger(PartitionRequestQueue.class);
+
+ private final ChannelFutureListener writeListener = new WriteAndFlushNextMessageIfPossibleListener();
+
+ private final Queue<SequenceNumberingPartitionQueueIterator> queue = new ArrayDeque<SequenceNumberingPartitionQueueIterator>();
+
+ private SequenceNumberingPartitionQueueIterator currentPartitionQueue;
+
+ private boolean fatalError;
+
+ private ChannelHandlerContext ctx;
+
+ private int numTotalSubscribeCalls;
+
+ private AtomicInteger numOutstandingSubscribeCalls = new AtomicInteger();
+
+ private int numConsumedPartitions;
+
+ private AtomicInteger numEnqueueCalls = new AtomicInteger();
+
+ private int numTotalEnqueueOperations;
+
+ private ScheduledFuture<?> logOutputTask;
+
+ @Override
+ public void channelActive(ChannelHandlerContext ctx) throws Exception {
+ if (LOG.isDebugEnabled()) {
+ logOutputTask = ctx.channel().eventLoop().scheduleWithFixedDelay(new DebugOutputTask(), 30, 30, TimeUnit.SECONDS);
+ }
+
+ super.channelActive(ctx);
+ }
+
+ @Override
+ public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
+ if (logOutputTask != null) {
+ logOutputTask.cancel(true);
+ }
+
+ super.channelUnregistered(ctx);
+ }
+
+ @Override
+ public void channelRegistered(final ChannelHandlerContext ctx) throws Exception {
+ if (this.ctx == null) {
+ this.ctx = ctx;
+ }
+
+ super.channelRegistered(ctx);
+ }
+
+ public void enqueue(IntermediateResultPartitionQueueIterator partitionQueue, InputChannelID receiverId) throws Exception {
+ numEnqueueCalls.incrementAndGet();
+ ctx.pipeline().fireUserEventTriggered(new SequenceNumberingPartitionQueueIterator(partitionQueue, receiverId));
+ }
+
+ @Override
+ public void userEventTriggered(ChannelHandlerContext ctx, Object msg) throws Exception {
+ if (msg.getClass() == SequenceNumberingPartitionQueueIterator.class) {
+ boolean triggerWrite = queue.isEmpty();
+
+ numTotalEnqueueOperations++;
+ queue.add((SequenceNumberingPartitionQueueIterator) msg);
+
+ if (triggerWrite) {
+ writeAndFlushNextMessageIfPossible(ctx.channel());
+ }
+ }
+ else {
+ ctx.fireUserEventTriggered(msg);
+ }
+ }
+
+ @Override
+ public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception {
+ writeAndFlushNextMessageIfPossible(ctx.channel());
+ }
+
+ private void writeAndFlushNextMessageIfPossible(final Channel channel) throws IOException {
+ if (fatalError) {
+ return;
+ }
+
+ Buffer buffer = null;
+
+ try {
+ if (channel.isWritable()) {
+ while (true) {
+ if (currentPartitionQueue == null && (currentPartitionQueue = queue.poll()) == null) {
+ return;
+ }
+
+ buffer = currentPartitionQueue.getNextBuffer();
+
+ if (buffer == null) {
+ if (currentPartitionQueue.subscribe(null)) {
+ numTotalSubscribeCalls++;
+ numOutstandingSubscribeCalls.incrementAndGet();
+
+ currentPartitionQueue = null;
+ }
+ else if (currentPartitionQueue.isConsumed()) {
+ numConsumedPartitions++;
+
+ currentPartitionQueue = null;
+ }
+ }
+ else {
+ BufferResponse resp = new BufferResponse(buffer, currentPartitionQueue.getSequenceNumber(), currentPartitionQueue.getReceiverId());
+
+ channel.writeAndFlush(resp).addListener(writeListener);
+
+ return;
+ }
+ }
+ }
+ }
+ catch (Throwable t) {
+ try {
+ if (buffer != null) {
+ buffer.recycle();
+ }
+ }
+ catch (Throwable ignored) {
+ // Make sure that this buffer is recycled in any case
+ }
+
+ throw new IOException(t.getMessage(), t);
+ }
+ }
+
+ @Override
+ public void channelInactive(ChannelHandlerContext ctx) throws Exception {
+ releaseAllResources();
+
+ ctx.fireChannelInactive();
+ }
+
+ @Override
+ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
+ handleException(ctx.channel(), cause);
+ }
+
+ private void handleException(Channel channel, Throwable cause) throws IOException {
+ fatalError = true;
+ releaseAllResources();
+
+ if (channel.isActive()) {
+ channel.writeAndFlush(new NettyMessage.ErrorResponse(cause)).addListener(ChannelFutureListener.CLOSE);
+ }
+ }
+
+ private void releaseAllResources() throws IOException {
+ if (currentPartitionQueue != null) {
+ currentPartitionQueue.discard();
+ currentPartitionQueue = null;
+ }
+
+ while ((currentPartitionQueue = queue.poll()) != null) {
+ currentPartitionQueue.discard();
+ }
+ }
+
+ // This listener is called after an element of the current queue has been
+ // flushed. If successful, the listener triggers further processing of the
+ // queues.
+ private class WriteAndFlushNextMessageIfPossibleListener implements ChannelFutureListener {
+
+ @Override
+ public void operationComplete(ChannelFuture future) throws Exception {
+ if (future.isSuccess()) {
+ writeAndFlushNextMessageIfPossible(future.channel());
+ }
+ else if (future.cause() != null) {
+ handleException(future.channel(), future.cause());
+ }
+ else {
+ handleException(future.channel(), new IllegalStateException("Sending cancelled by user."));
+ }
+ }
+ }
+
+ /**
+ * Simple wrapper for the partition queue iterator, which increments a
+ * sequence number for each returned buffer and remembers the receiver ID.
+ */
+ private class SequenceNumberingPartitionQueueIterator implements IntermediateResultPartitionQueueIterator, NotificationListener {
+
+ private final IntermediateResultPartitionQueueIterator queueIterator;
+
+ private final InputChannelID receiverId;
+
+ private int sequenceNumber = -1;
+
+ private SequenceNumberingPartitionQueueIterator(IntermediateResultPartitionQueueIterator queueIterator, InputChannelID receiverId) {
+ this.queueIterator = checkNotNull(queueIterator);
+ this.receiverId = checkNotNull(receiverId);
+ }
+
+ private InputChannelID getReceiverId() {
+ return receiverId;
+ }
+
+ private int getSequenceNumber() {
+ return sequenceNumber;
+ }
+
+ @Override
+ public Buffer getNextBuffer() throws IOException {
+ Buffer buffer = queueIterator.getNextBuffer();
+
+ if (buffer != null) {
+ sequenceNumber++;
+ }
+
+ return buffer;
+ }
+
+ @Override
+ public void discard() throws IOException {
+ queueIterator.discard();
+ }
+
+ @Override
+ public boolean subscribe(NotificationListener ignored) throws AlreadySubscribedException {
+ return queueIterator.subscribe(this);
+ }
+
+ @Override
+ public boolean isConsumed() {
+ return queueIterator.isConsumed();
+ }
+
+ /**
+ * Enqueue this iterator again after a notification.
+ */
+ @Override
+ public void onNotification() {
+ numOutstandingSubscribeCalls.decrementAndGet();
+ ctx.pipeline().fireUserEventTriggered(this);
+ }
+ }
+
+ // ------------------------------------------------------------------------
+
+ /**
+ * Debug output task executed periodically by the network I/O thread.
+ */
+ private class DebugOutputTask implements Runnable {
+
+ @Override
+ public void run() {
+ StringBuilder str = new StringBuilder();
+
+ str.append("Channel remote address: ");
+ str.append(ctx.channel().remoteAddress());
+ str.append(". ");
+
+ str.append("Channel active: ");
+ str.append(ctx.channel().isActive());
+ str.append(". ");
+
+ str.append("Total number of queue operations: ");
+ str.append(numTotalEnqueueOperations);
+ str.append(". ");
+
+ str.append("Number of enqueue calls: ");
+ str.append(numEnqueueCalls.get());
+ str.append(". ");
+
+ str.append("Number of consumed partitions: ");
+ str.append(numConsumedPartitions);
+ str.append(". ");
+
+ str.append("Number of currently queued partitions: ");
+ str.append(queue.size());
+ str.append(". ");
+
+ str.append("Current partition queue: ");
+ str.append(currentPartitionQueue);
+ str.append(". ");
+
+ str.append("Total number of subscribe calls: ");
+ str.append(numTotalSubscribeCalls);
+ str.append(". ");
+
+ str.append("Number of outstanding subscribe calls: ");
+ str.append(numOutstandingSubscribeCalls.get());
+ str.append(". ");
+
+ str.append("Channel writeable? ");
+ str.append(ctx.channel().isWritable());
+ str.append(". ");
+
+ LOG.debug(str.toString());
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/d908ca19/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
new file mode 100644
index 0000000..93f7192
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestServerHandler.java
@@ -0,0 +1,103 @@
+/*
+ * 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 com.google.common.base.Optional;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.SimpleChannelInboundHandler;
+import org.apache.flink.runtime.io.network.TaskEventDispatcher;
+import org.apache.flink.runtime.io.network.buffer.BufferProvider;
+import org.apache.flink.runtime.io.network.partition.IntermediateResultPartitionProvider;
+import org.apache.flink.runtime.io.network.partition.consumer.InputChannelID;
+import org.apache.flink.runtime.io.network.partition.queue.IntermediateResultPartitionQueueIterator;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static org.apache.flink.runtime.io.network.netty.NettyMessage.PartitionRequest;
+import static org.apache.flink.runtime.io.network.netty.NettyMessage.TaskEventRequest;
+
+class PartitionRequestServerHandler extends SimpleChannelInboundHandler<NettyMessage> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(PartitionRequestServerHandler.class);
+
+ private final IntermediateResultPartitionProvider partitionProvider;
+
+ private final TaskEventDispatcher taskEventDispatcher;
+
+ private final PartitionRequestQueue outboundQueue;
+
+ PartitionRequestServerHandler(IntermediateResultPartitionProvider partitionProvider, TaskEventDispatcher taskEventDispatcher, PartitionRequestQueue outboundQueue) {
+
+ this.partitionProvider = partitionProvider;
+ this.taskEventDispatcher = taskEventDispatcher;
+ this.outboundQueue = outboundQueue;
+ }
+
+ @Override
+ protected void channelRead0(ChannelHandlerContext ctx, NettyMessage msg) throws Exception {
+ try {
+ Class<?> msgClazz = msg.getClass();
+
+ // ----------------------------------------------------------------
+ // Intermediate result partition requests
+ // ----------------------------------------------------------------
+ if (msgClazz == PartitionRequest.class) {
+ PartitionRequest request = (PartitionRequest) msg;
+
+ IntermediateResultPartitionQueueIterator queueIterator =
+ partitionProvider.getIntermediateResultPartitionIterator(
+ request.producerExecutionId,
+ request.partitionId,
+ request.queueIndex,
+ Optional.<BufferProvider>absent());
+
+ if (queueIterator != null) {
+ outboundQueue.enqueue(queueIterator, request.receiverId);
+ }
+ else {
+ respondWithError(ctx, new IllegalArgumentException("Partition not found"), request.receiverId);
+ }
+ }
+ // ----------------------------------------------------------------
+ // Task events
+ // ----------------------------------------------------------------
+ else if (msgClazz == TaskEventRequest.class) {
+ TaskEventRequest request = (TaskEventRequest) msg;
+
+ if (!taskEventDispatcher.publish(request.executionId, request.partitionId, request.event)) {
+ respondWithError(ctx, new IllegalArgumentException("Task event receiver not found."), request.receiverId);
+ }
+ }
+ else {
+ LOG.warn("Received unexpected client request: {}", msg);
+ }
+ }
+ catch (Throwable t) {
+ respondWithError(ctx, t);
+ }
+ }
+
+ private void respondWithError(ChannelHandlerContext ctx, Throwable error) {
+ ctx.writeAndFlush(new NettyMessage.ErrorResponse(error));
+ }
+
+ private void respondWithError(ChannelHandlerContext ctx, Throwable error, InputChannelID sourceId) {
+ ctx.writeAndFlush(new NettyMessage.ErrorResponse(error, sourceId));
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/d908ca19/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/IntermediateResultPartition.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/IntermediateResultPartition.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/IntermediateResultPartition.java
new file mode 100644
index 0000000..6acfbce
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/IntermediateResultPartition.java
@@ -0,0 +1,307 @@
+/*
+ * 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.partition;
+
+import com.google.common.base.Optional;
+import org.apache.flink.runtime.akka.AkkaUtils;
+import org.apache.flink.runtime.deployment.PartitionDeploymentDescriptor;
+import org.apache.flink.runtime.execution.RuntimeEnvironment;
+import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
+import org.apache.flink.runtime.executiongraph.ExecutionVertex;
+import org.apache.flink.runtime.io.network.NetworkEnvironment;
+import org.apache.flink.runtime.io.network.buffer.Buffer;
+import org.apache.flink.runtime.io.network.buffer.BufferPool;
+import org.apache.flink.runtime.io.network.buffer.BufferPoolOwner;
+import org.apache.flink.runtime.io.network.buffer.BufferProvider;
+import org.apache.flink.runtime.io.network.partition.queue.IllegalQueueIteratorRequestException;
+import org.apache.flink.runtime.io.network.partition.queue.IntermediateResultPartitionQueue;
+import org.apache.flink.runtime.io.network.partition.queue.IntermediateResultPartitionQueueIterator;
+import org.apache.flink.runtime.io.network.partition.queue.PipelinedPartitionQueue;
+import org.apache.flink.runtime.jobgraph.IntermediateResultPartitionID;
+import org.apache.flink.runtime.jobgraph.IntermediateResultPartitionType;
+import org.apache.flink.runtime.jobgraph.JobID;
+import org.apache.flink.runtime.messages.JobManagerMessages;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import scala.Option;
+
+import java.io.IOException;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import static com.google.common.base.Preconditions.checkNotNull;
+import static com.google.common.base.Preconditions.checkState;
+
+public class IntermediateResultPartition implements BufferPoolOwner {
+
+ private static final Logger LOG = LoggerFactory.getLogger(IntermediateResultPartition.class);
+
+ private final RuntimeEnvironment environment;
+
+ /**
+ * Note: This index needs to correspond to the index of the partition in
+ * {@link ExecutionVertex#resultPartitions}, which might be a little
+ * fragile as the data availability notifications use it.
+ */
+ private final int partitionIndex;
+
+ private final JobID jobId;
+
+ private final ExecutionAttemptID producerExecutionId;
+
+ private final IntermediateResultPartitionID partitionId;
+
+ private final IntermediateResultPartitionType partitionType;
+
+ private final NetworkEnvironment networkEnvironment;
+
+ private final IntermediateResultPartitionQueue[] queues;
+
+ private volatile boolean hasNotifiedConsumers;
+
+ private volatile boolean isReleased;
+
+ private boolean isFinished;
+
+ private BufferPool bufferPool;
+
+ public IntermediateResultPartition(RuntimeEnvironment environment, int partitionIndex, JobID jobId, ExecutionAttemptID executionId, IntermediateResultPartitionID partitionId, IntermediateResultPartitionType partitionType, IntermediateResultPartitionQueue[] partitionQueues, NetworkEnvironment networkEnvironment) {
+ this.environment = environment;
+ this.partitionIndex = partitionIndex;
+ this.jobId = jobId;
+ this.producerExecutionId = executionId;
+ this.partitionId = partitionId;
+ this.partitionType = partitionType;
+ this.networkEnvironment = networkEnvironment;
+ this.queues = partitionQueues;
+ }
+
+ // ------------------------------------------------------------------------
+ // Properties
+ // ------------------------------------------------------------------------
+
+ public void setBufferPool(BufferPool bufferPool) {
+ checkArgument(bufferPool.getNumberOfRequiredMemorySegments() == getNumberOfQueues(), "Buffer pool has not enough buffers for this intermediate result.");
+ checkState(this.bufferPool == null, "Buffer pool has already been set for intermediate result partition.");
+
+ this.bufferPool = checkNotNull(bufferPool);
+ }
+
+ public ExecutionAttemptID getProducerExecutionId() {
+ return producerExecutionId;
+ }
+
+ public IntermediateResultPartitionID getPartitionId() {
+ return partitionId;
+ }
+
+ public JobID getJobId() {
+ return jobId;
+ }
+
+ public int getNumberOfQueues() {
+ return queues.length;
+ }
+
+ public BufferProvider getBufferProvider() {
+ return bufferPool;
+ }
+
+ public boolean isFinished() {
+ return isFinished;
+ }
+
+ // ------------------------------------------------------------------------
+ // Produce
+ // ------------------------------------------------------------------------
+
+ public void add(Buffer buffer, int targetQueue) throws IOException {
+ synchronized (queues) {
+ if (isReleased) {
+ buffer.recycle();
+ }
+ else {
+ checkInProducePhase();
+ queues[targetQueue].add(buffer);
+ }
+ }
+
+ maybeNotifyConsumers(partitionType.isPipelined());
+ }
+
+ public void finish() throws IOException {
+ boolean success = false;
+
+ synchronized (queues) {
+ checkInProducePhase();
+
+ try {
+ if (!isReleased) {
+ for (IntermediateResultPartitionQueue queue : queues) {
+ queue.finish();
+ }
+
+ success = true;
+ }
+ }
+ finally {
+ isFinished = true;
+ }
+ }
+
+ if (success) {
+ // Notify at this point in any case either because of the end
+ // of a blocking result or an empty pipelined result.
+ maybeNotifyConsumers(true);
+
+ if (!partitionType.isPersistent() && bufferPool != null) {
+ // If this partition is not persistent, immediately destroy
+ // the buffer pool. For persistent intermediate results, the
+ // partition manager needs to release the buffer pool.
+ bufferPool.destroy();
+ }
+ }
+ }
+
+ public void releaseAllResources() throws IOException {
+ synchronized (queues) {
+ if (!isReleased) {
+ try {
+ for (IntermediateResultPartitionQueue queue : queues) {
+ try {
+ queue.discard();
+ }
+ catch (Throwable t) {
+ LOG.error("Error while discarding queue: " + t.getMessage(), t);
+ }
+ }
+
+ if (bufferPool != null) {
+ bufferPool.destroy();
+ }
+ }
+ finally {
+ isReleased = true;
+ }
+ }
+ }
+ }
+
+ // ------------------------------------------------------------------------
+ // Consume
+ // ------------------------------------------------------------------------
+
+ public IntermediateResultPartitionQueueIterator getQueueIterator(int queueIndex, Optional<BufferProvider> bufferProvider) throws IOException {
+ synchronized (queues) {
+ if (isReleased) {
+ throw new IllegalQueueIteratorRequestException("Intermediate result partition has already been released.");
+ }
+
+ if (queueIndex < 0 || queueIndex >= queues.length) {
+ throw new IllegalQueueIteratorRequestException("Illegal queue index: " + queueIndex + ", allowed: 0-" + (queues.length - 1));
+ }
+
+ return queues[queueIndex].getQueueIterator(bufferProvider);
+ }
+ }
+
+ // ------------------------------------------------------------------------
+
+ @Override
+ public String toString() {
+ return "Intermediate result partition " + partitionId + " [num queues: " + queues.length + ", " + (isFinished ? "finished" : "not finished") + "]";
+ }
+
+ private void checkInProducePhase() {
+ checkState(!isReleased, "Partition has already been discarded.");
+ checkState(!isFinished, "Partition has already been finished.");
+ }
+
+ /**
+ * Maybe notifies consumers of this result partition.
+ */
+ private void maybeNotifyConsumers(boolean doNotify) throws IOException {
+ if (doNotify && !hasNotifiedConsumers) {
+ scheduleOrUpdateConsumers();
+ hasNotifiedConsumers = true;
+ }
+ }
+
+ private void scheduleOrUpdateConsumers() throws IOException {
+ while (!isReleased) {
+ final JobManagerMessages.ConsumerNotificationResult result = AkkaUtils.ask(
+ networkEnvironment.getJobManager(),
+ new JobManagerMessages.ScheduleOrUpdateConsumers(jobId, producerExecutionId, partitionIndex),
+ networkEnvironment.getJobManagerTimeout());
+
+ if (result.success()) {
+ return;
+ }
+ else {
+ Option<Throwable> error = result.error();
+ if (error.isDefined()) {
+ throw new IOException(error.get().getMessage(), error.get());
+ }
+ }
+
+ try {
+ Thread.sleep(10);
+ }
+ catch (InterruptedException e) {
+ throw new IOException("Unexpected interruption during consumer schedule or update.", e);
+ }
+ }
+ }
+
+ // ------------------------------------------------------------------------
+ // Buffer pool owner methods
+ // ------------------------------------------------------------------------
+
+ /**
+ * If this partition is registered as the {@link BufferPoolOwner} of a
+ * {@link BufferPool}, it will forward the requests to the queues.
+ */
+ @Override
+ public void recycleBuffers(int numBuffersToRecycle) throws IOException {
+ int numRecycledBuffers = 0;
+
+ for (IntermediateResultPartitionQueue queue : queues) {
+ numRecycledBuffers += queue.recycleBuffers();
+
+ if (numRecycledBuffers >= numBuffersToRecycle) {
+ break;
+ }
+ }
+ }
+
+ // ------------------------------------------------------------------------
+
+ public static IntermediateResultPartition create(RuntimeEnvironment environment, int partitionIndex, JobID jobId, ExecutionAttemptID executionId, NetworkEnvironment networkEnvironment, PartitionDeploymentDescriptor desc) {
+ final IntermediateResultPartitionID partitionId = checkNotNull(desc.getPartitionId());
+ final IntermediateResultPartitionType partitionType = checkNotNull(desc.getPartitionType());
+
+ final IntermediateResultPartitionQueue[] partitionQueues = new IntermediateResultPartitionQueue[desc.getNumberOfQueues()];
+
+ // TODO The queues need to be created depending on the result type
+ for (int i = 0; i < partitionQueues.length; i++) {
+ partitionQueues[i] = new PipelinedPartitionQueue();
+ }
+
+ return new IntermediateResultPartition(environment, partitionIndex, jobId, executionId, partitionId, partitionType, partitionQueues, networkEnvironment);
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/d908ca19/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/IntermediateResultPartitionManager.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/IntermediateResultPartitionManager.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/IntermediateResultPartitionManager.java
new file mode 100644
index 0000000..55a741b
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/IntermediateResultPartitionManager.java
@@ -0,0 +1,131 @@
+/*
+ * 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.partition;
+
+import com.google.common.base.Optional;
+import com.google.common.collect.HashBasedTable;
+import com.google.common.collect.Table;
+import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
+import org.apache.flink.runtime.io.network.buffer.BufferProvider;
+import org.apache.flink.runtime.io.network.partition.queue.IllegalQueueIteratorRequestException;
+import org.apache.flink.runtime.io.network.partition.queue.IntermediateResultPartitionQueueIterator;
+import org.apache.flink.runtime.jobgraph.IntermediateResultPartitionID;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * The intermediate result partition manager keeps track of all available
+ * partitions of a task manager and
+ */
+public class IntermediateResultPartitionManager implements IntermediateResultPartitionProvider {
+
+ private static final Logger LOG = LoggerFactory.getLogger(IntermediateResultPartitionManager.class);
+
+ public final Table<ExecutionAttemptID, IntermediateResultPartitionID, IntermediateResultPartition> partitions = HashBasedTable.create();
+
+ private boolean isShutdown;
+
+ public void registerIntermediateResultPartition(IntermediateResultPartition partition) throws IOException {
+ synchronized (partitions) {
+ if (isShutdown) {
+ throw new IOException("Intermediate result partition manager has already been shut down.");
+ }
+
+ if (partitions.put(partition.getProducerExecutionId(), partition.getPartitionId(), partition) != null) {
+ throw new IOException("Tried to re-register intermediate result partition.");
+ }
+ }
+ }
+
+ public void failIntermediateResultPartitions(ExecutionAttemptID producerExecutionId) {
+ synchronized (partitions) {
+ List<IntermediateResultPartition> partitionsToFail = new ArrayList<IntermediateResultPartition>();
+
+ for (IntermediateResultPartitionID partitionId : partitions.row(producerExecutionId).keySet()) {
+ partitionsToFail.add(partitions.get(producerExecutionId, partitionId));
+ }
+
+ for(IntermediateResultPartition partition : partitionsToFail) {
+ failIntermediateResultPartition(partition);
+ }
+ }
+ }
+
+ private void failIntermediateResultPartition(IntermediateResultPartition partition) {
+ if (partition != null) {
+ try {
+ partition.releaseAllResources();
+ }
+ catch (Throwable t) {
+ LOG.error("Error during release of produced intermediate result partition: " + t.getMessage(), t);
+ }
+ }
+ }
+
+ public void shutdown() {
+ synchronized (partitions) {
+ for (IntermediateResultPartition partition : partitions.values()) {
+ try {
+ partition.releaseAllResources();
+ }
+ catch (IOException e) {
+ LOG.error("Error while releasing intermediate result partition: " + e.getMessage(), e);
+ }
+ }
+
+ isShutdown = true;
+ }
+ }
+
+ public int getNumberOfRegisteredPartitions() {
+ synchronized (partitions) {
+ return partitions.size();
+ }
+ }
+
+ // ------------------------------------------------------------------------
+ // Intermediate result partition provider
+ // ------------------------------------------------------------------------
+
+ @Override
+ public IntermediateResultPartitionQueueIterator getIntermediateResultPartitionIterator(
+ ExecutionAttemptID producerExecutionId,
+ IntermediateResultPartitionID partitionId,
+ int queueIndex,
+ Optional<BufferProvider> bufferProvider) throws IOException {
+
+ synchronized (partitions) {
+ IntermediateResultPartition partition = partitions.get(producerExecutionId, partitionId);
+
+ if (partition == null) {
+ if (!partitions.containsRow(producerExecutionId)) {
+ throw new IllegalQueueIteratorRequestException("Unknown producer execution ID " + producerExecutionId + ".");
+ }
+
+ throw new IllegalQueueIteratorRequestException("Unknown partition " + partitionId + ".");
+ }
+
+ return partition.getQueueIterator(queueIndex, bufferProvider);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/d908ca19/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/IntermediateResultPartitionProvider.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/IntermediateResultPartitionProvider.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/IntermediateResultPartitionProvider.java
new file mode 100644
index 0000000..b18b3fc
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/IntermediateResultPartitionProvider.java
@@ -0,0 +1,37 @@
+/*
+ * 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.partition;
+
+import com.google.common.base.Optional;
+import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
+import org.apache.flink.runtime.io.network.buffer.BufferProvider;
+import org.apache.flink.runtime.io.network.partition.queue.IntermediateResultPartitionQueueIterator;
+import org.apache.flink.runtime.jobgraph.IntermediateResultPartitionID;
+
+import java.io.IOException;
+
+public interface IntermediateResultPartitionProvider {
+
+ IntermediateResultPartitionQueueIterator getIntermediateResultPartitionIterator(
+ ExecutionAttemptID producerExecutionId,
+ IntermediateResultPartitionID partitionId,
+ int requestedQueueIndex,
+ Optional<BufferProvider> bufferProvider) throws IOException;
+
+}