You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by se...@apache.org on 2014/06/18 00:18:48 UTC
[1/4] git commit: [FLINK-942] Fix bug in config.sh to correctly set
user-specified JVM args
Repository: incubator-flink
Updated Branches:
refs/heads/master ca2b287a7 -> 9a75e685a
[FLINK-942] Fix bug in config.sh to correctly set user-specified JVM args
This closes #22
Project: http://git-wip-us.apache.org/repos/asf/incubator-flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-flink/commit/6a549325
Tree: http://git-wip-us.apache.org/repos/asf/incubator-flink/tree/6a549325
Diff: http://git-wip-us.apache.org/repos/asf/incubator-flink/diff/6a549325
Branch: refs/heads/master
Commit: 6a549325f3c1e98b4c353e9bec6f951affb95bc9
Parents: ca2b287
Author: uce <u....@fu-berlin.de>
Authored: Mon Jun 16 22:17:48 2014 +0200
Committer: Stephan Ewen <se...@apache.org>
Committed: Tue Jun 17 14:23:24 2014 +0200
----------------------------------------------------------------------
stratosphere-dist/src/main/stratosphere-bin/bin/config.sh | 6 +++---
stratosphere-dist/src/main/stratosphere-bin/bin/jobmanager.sh | 2 +-
stratosphere-dist/src/main/stratosphere-bin/bin/taskmanager.sh | 2 +-
3 files changed, 5 insertions(+), 5 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/6a549325/stratosphere-dist/src/main/stratosphere-bin/bin/config.sh
----------------------------------------------------------------------
diff --git a/stratosphere-dist/src/main/stratosphere-bin/bin/config.sh b/stratosphere-dist/src/main/stratosphere-bin/bin/config.sh
index bd6c302..9168813 100755
--- a/stratosphere-dist/src/main/stratosphere-bin/bin/config.sh
+++ b/stratosphere-dist/src/main/stratosphere-bin/bin/config.sh
@@ -167,12 +167,12 @@ if [ -z "${STRATOSPHERE_PID_DIR}" ]; then
STRATOSPHERE_PID_DIR=$(readFromConfig ${KEY_ENV_PID_DIR} "${DEFAULT_ENV_PID_DIR}" "${YAML_CONF}")
fi
-if [ -z "${STRATOSPHERE_OPTS}" ]; then
- STRATOSPHERE_OPTS=$(readFromConfig ${KEY_ENV_JAVA_OPTS} "${DEFAULT_ENV_JAVA_OPTS}" "${YAML_CONF}")
+if [ -z "${STRATOSPHERE_ENV_JAVA_OPTS}" ]; then
+ STRATOSPHERE_ENV_JAVA_OPTS=$(readFromConfig ${KEY_ENV_JAVA_OPTS} "${DEFAULT_ENV_JAVA_OPTS}" "${YAML_CONF}")
fi
if [ -z "${STRATOSPHERE_SSH_OPTS}" ]; then
- STRATOSPHERE_OPTS=$(readFromConfig ${KEY_ENV_SSH_OPTS} "${DEFAULT_ENV_SSH_OPTS}" "${YAML_CONF}")
+ STRATOSPHERE_SSH_OPTS=$(readFromConfig ${KEY_ENV_SSH_OPTS} "${DEFAULT_ENV_SSH_OPTS}" "${YAML_CONF}")
fi
# Arguments for the JVM. Used for job and task manager JVMs.
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/6a549325/stratosphere-dist/src/main/stratosphere-bin/bin/jobmanager.sh
----------------------------------------------------------------------
diff --git a/stratosphere-dist/src/main/stratosphere-bin/bin/jobmanager.sh b/stratosphere-dist/src/main/stratosphere-bin/bin/jobmanager.sh
index d41874c..e474e5e 100755
--- a/stratosphere-dist/src/main/stratosphere-bin/bin/jobmanager.sh
+++ b/stratosphere-dist/src/main/stratosphere-bin/bin/jobmanager.sh
@@ -72,7 +72,7 @@ case $STARTSTOP in
rotateLogFile $out
echo Starting job manager
- $JAVA_RUN $JVM_ARGS $STRATOSPHERE_OPTS "${log_setting[@]}" -classpath "$STRATOSPHERE_JM_CLASSPATH" eu.stratosphere.nephele.jobmanager.JobManager -executionMode $EXECUTIONMODE -configDir "$STRATOSPHERE_CONF_DIR" > "$out" 2>&1 < /dev/null &
+ $JAVA_RUN $JVM_ARGS $STRATOSPHERE_ENV_JAVA_OPTS "${log_setting[@]}" -classpath "$STRATOSPHERE_JM_CLASSPATH" eu.stratosphere.nephele.jobmanager.JobManager -executionMode $EXECUTIONMODE -configDir "$STRATOSPHERE_CONF_DIR" > "$out" 2>&1 < /dev/null &
echo $! > $pid
;;
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/6a549325/stratosphere-dist/src/main/stratosphere-bin/bin/taskmanager.sh
----------------------------------------------------------------------
diff --git a/stratosphere-dist/src/main/stratosphere-bin/bin/taskmanager.sh b/stratosphere-dist/src/main/stratosphere-bin/bin/taskmanager.sh
index 0800be6..ea6e468 100755
--- a/stratosphere-dist/src/main/stratosphere-bin/bin/taskmanager.sh
+++ b/stratosphere-dist/src/main/stratosphere-bin/bin/taskmanager.sh
@@ -69,7 +69,7 @@ case $STARTSTOP in
rotateLogFile $out
echo Starting task manager on host $HOSTNAME
- $JAVA_RUN $JVM_ARGS $STRATOSPHERE_OPTS "${log_setting[@]}" -classpath "$STRATOSPHERE_TM_CLASSPATH" eu.stratosphere.nephele.taskmanager.TaskManager -configDir "$STRATOSPHERE_CONF_DIR" > "$out" 2>&1 < /dev/null &
+ $JAVA_RUN $JVM_ARGS $STRATOSPHERE_ENV_JAVA_OPTS "${log_setting[@]}" -classpath "$STRATOSPHERE_TM_CLASSPATH" eu.stratosphere.nephele.taskmanager.TaskManager -configDir "$STRATOSPHERE_CONF_DIR" > "$out" 2>&1 < /dev/null &
echo $! > $pid
;;
[4/4] git commit: Do not limit the client's heap space (See FLINK-934)
Posted by se...@apache.org.
Do not limit the client's heap space (See FLINK-934)
[ci skip]
This closes #18
Project: http://git-wip-us.apache.org/repos/asf/incubator-flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-flink/commit/9a75e685
Tree: http://git-wip-us.apache.org/repos/asf/incubator-flink/tree/9a75e685
Diff: http://git-wip-us.apache.org/repos/asf/incubator-flink/diff/9a75e685
Branch: refs/heads/master
Commit: 9a75e685ac401c848f7e5fdd34122bc6e6a9d251
Parents: b7cbf30
Author: Robert Metzger <rm...@apache.org>
Authored: Fri Jun 13 21:40:34 2014 +0200
Committer: Stephan Ewen <se...@apache.org>
Committed: Tue Jun 17 14:56:46 2014 +0200
----------------------------------------------------------------------
stratosphere-dist/src/main/stratosphere-bin/bin/stratosphere | 1 -
1 file changed, 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/9a75e685/stratosphere-dist/src/main/stratosphere-bin/bin/stratosphere
----------------------------------------------------------------------
diff --git a/stratosphere-dist/src/main/stratosphere-bin/bin/stratosphere b/stratosphere-dist/src/main/stratosphere-bin/bin/stratosphere
index dc6ac20..28a6059 100755
--- a/stratosphere-dist/src/main/stratosphere-bin/bin/stratosphere
+++ b/stratosphere-dist/src/main/stratosphere-bin/bin/stratosphere
@@ -26,7 +26,6 @@ fi
STRATOSPHERE_LIB_CLIENTS_DIR=$STRATOSPHERE_ROOT_DIR/lib_clients
-JVM_ARGS="$JVM_ARGS -Xmx512m"
# auxiliary function to construct a lightweight classpath for the
# Stratosphere CLI client
[3/4] git commit: Artificial commit to close pull requests.
Posted by se...@apache.org.
Artificial commit to close pull requests.
This closes #5
This closes #6
This closes #10
This closes #12
This closes #14
Project: http://git-wip-us.apache.org/repos/asf/incubator-flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-flink/commit/b7cbf30c
Tree: http://git-wip-us.apache.org/repos/asf/incubator-flink/tree/b7cbf30c
Diff: http://git-wip-us.apache.org/repos/asf/incubator-flink/diff/b7cbf30c
Branch: refs/heads/master
Commit: b7cbf30c0d9c83b478c00c6dda00883c0657bb6e
Parents: a70de7e
Author: Stephan Ewen <se...@apache.org>
Authored: Tue Jun 17 14:47:37 2014 +0200
Committer: Stephan Ewen <se...@apache.org>
Committed: Tue Jun 17 14:47:37 2014 +0200
----------------------------------------------------------------------
----------------------------------------------------------------------
[2/4] git commit: [FLINK-930] Netty Initialization is sometimes very
slow
Posted by se...@apache.org.
[FLINK-930] Netty Initialization is sometimes very slow
This closes #19
Project: http://git-wip-us.apache.org/repos/asf/incubator-flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-flink/commit/a70de7e3
Tree: http://git-wip-us.apache.org/repos/asf/incubator-flink/tree/a70de7e3
Diff: http://git-wip-us.apache.org/repos/asf/incubator-flink/diff/a70de7e3
Branch: refs/heads/master
Commit: a70de7e3faef8afea0134c8fed1240d742cccb3d
Parents: 6a54932
Author: uce <u....@fu-berlin.de>
Authored: Fri Jun 13 10:41:32 2014 +0200
Committer: Stephan Ewen <se...@apache.org>
Committed: Tue Jun 17 14:43:17 2014 +0200
----------------------------------------------------------------------
.../client/minicluster/NepheleMiniCluster.java | 2 +-
stratosphere-runtime/pom.xml | 2 +-
.../eu/stratosphere/nephele/ExecutionMode.java | 18 ++
.../instance/local/LocalInstanceManager.java | 6 +-
.../nephele/jobmanager/JobManager.java | 4 +-
.../nephele/jobmanager/JobManagerUtils.java | 2 +-
.../nephele/taskmanager/TaskManager.java | 64 ++++---
.../runtime/io/network/ChannelManager.java | 23 +--
.../io/network/LocalConnectionManager.java | 31 ++++
.../io/network/NetworkConnectionManager.java | 25 +++
.../netty/InboundEnvelopeDispatcher.java | 35 ++++
.../netty/InboundEnvelopeDispatcherHandler.java | 41 ----
.../network/netty/NettyConnectionManager.java | 121 +++++++-----
.../network/netty/OutboundConnectionQueue.java | 13 +-
.../local/LocalInstanceManagerTest.java | 3 +-
.../nephele/jobmanager/JobManagerITCase.java | 2 +-
.../netty/NettyConnectionManagerTest.java | 185 +++++++++----------
17 files changed, 341 insertions(+), 236 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/a70de7e3/stratosphere-clients/src/main/java/eu/stratosphere/client/minicluster/NepheleMiniCluster.java
----------------------------------------------------------------------
diff --git a/stratosphere-clients/src/main/java/eu/stratosphere/client/minicluster/NepheleMiniCluster.java b/stratosphere-clients/src/main/java/eu/stratosphere/client/minicluster/NepheleMiniCluster.java
index f44a47e..79e5c64 100644
--- a/stratosphere-clients/src/main/java/eu/stratosphere/client/minicluster/NepheleMiniCluster.java
+++ b/stratosphere-clients/src/main/java/eu/stratosphere/client/minicluster/NepheleMiniCluster.java
@@ -15,6 +15,7 @@ package eu.stratosphere.client.minicluster;
import java.lang.reflect.Method;
+import eu.stratosphere.nephele.ExecutionMode;
import eu.stratosphere.nephele.instance.HardwareDescriptionFactory;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@@ -27,7 +28,6 @@ import eu.stratosphere.configuration.GlobalConfiguration;
import eu.stratosphere.nephele.client.JobClient;
import eu.stratosphere.nephele.jobgraph.JobGraph;
import eu.stratosphere.nephele.jobmanager.JobManager;
-import eu.stratosphere.nephele.jobmanager.JobManager.ExecutionMode;
public class NepheleMiniCluster {
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/a70de7e3/stratosphere-runtime/pom.xml
----------------------------------------------------------------------
diff --git a/stratosphere-runtime/pom.xml b/stratosphere-runtime/pom.xml
index 074725b..7d19692 100644
--- a/stratosphere-runtime/pom.xml
+++ b/stratosphere-runtime/pom.xml
@@ -56,7 +56,7 @@
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-all</artifactId>
- <version>4.0.19.Final</version>
+ <version>4.0.20.Final</version>
</dependency>
<dependency>
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/a70de7e3/stratosphere-runtime/src/main/java/eu/stratosphere/nephele/ExecutionMode.java
----------------------------------------------------------------------
diff --git a/stratosphere-runtime/src/main/java/eu/stratosphere/nephele/ExecutionMode.java b/stratosphere-runtime/src/main/java/eu/stratosphere/nephele/ExecutionMode.java
new file mode 100644
index 0000000..68ab65b
--- /dev/null
+++ b/stratosphere-runtime/src/main/java/eu/stratosphere/nephele/ExecutionMode.java
@@ -0,0 +1,18 @@
+/***********************************************************************************************************************
+ * Copyright (C) 2010-2014 by the Stratosphere project (http://stratosphere.eu)
+ *
+ * Licensed 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 eu.stratosphere.nephele;
+
+public enum ExecutionMode {
+ LOCAL, CLUSTER
+}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/a70de7e3/stratosphere-runtime/src/main/java/eu/stratosphere/nephele/instance/local/LocalInstanceManager.java
----------------------------------------------------------------------
diff --git a/stratosphere-runtime/src/main/java/eu/stratosphere/nephele/instance/local/LocalInstanceManager.java b/stratosphere-runtime/src/main/java/eu/stratosphere/nephele/instance/local/LocalInstanceManager.java
index 2bb344f..e888b3f 100644
--- a/stratosphere-runtime/src/main/java/eu/stratosphere/nephele/instance/local/LocalInstanceManager.java
+++ b/stratosphere-runtime/src/main/java/eu/stratosphere/nephele/instance/local/LocalInstanceManager.java
@@ -20,6 +20,7 @@ import java.util.List;
import java.util.ArrayList;
import java.util.Iterator;
+import eu.stratosphere.nephele.ExecutionMode;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@@ -150,6 +151,9 @@ public class LocalInstanceManager implements InstanceManager {
numTaskManagers = GlobalConfiguration.getInteger(ConfigConstants
.LOCAL_INSTANCE_MANAGER_NUMBER_TASK_MANAGER, 1);
+
+ ExecutionMode executionMode = (numTaskManagers > 1) ? ExecutionMode.CLUSTER : ExecutionMode.LOCAL;
+
for(int i=0; i< numTaskManagers; i++){
Configuration tm = new Configuration();
@@ -163,7 +167,7 @@ public class LocalInstanceManager implements InstanceManager {
GlobalConfiguration.includeConfiguration(tm);
- TaskManager t = new TaskManager();
+ TaskManager t = new TaskManager(executionMode);
taskManagers.add(t);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/a70de7e3/stratosphere-runtime/src/main/java/eu/stratosphere/nephele/jobmanager/JobManager.java
----------------------------------------------------------------------
diff --git a/stratosphere-runtime/src/main/java/eu/stratosphere/nephele/jobmanager/JobManager.java b/stratosphere-runtime/src/main/java/eu/stratosphere/nephele/jobmanager/JobManager.java
index 3ae9f3b..5e92dd1 100644
--- a/stratosphere-runtime/src/main/java/eu/stratosphere/nephele/jobmanager/JobManager.java
+++ b/stratosphere-runtime/src/main/java/eu/stratosphere/nephele/jobmanager/JobManager.java
@@ -31,6 +31,7 @@ import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
+import eu.stratosphere.nephele.ExecutionMode;
import eu.stratosphere.nephele.managementgraph.ManagementVertexID;
import eu.stratosphere.nephele.taskmanager.TaskKillResult;
import org.apache.commons.cli.CommandLine;
@@ -123,9 +124,6 @@ import eu.stratosphere.util.StringUtils;
public class JobManager implements DeploymentManager, ExtendedManagementProtocol, InputSplitProviderProtocol,
JobManagerProtocol, ChannelLookupProtocol, JobStatusListener, AccumulatorProtocol
{
- public static enum ExecutionMode { LOCAL, CLUSTER }
-
- // --------------------------------------------------------------------------------------------
private static final Log LOG = LogFactory.getLog(JobManager.class);
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/a70de7e3/stratosphere-runtime/src/main/java/eu/stratosphere/nephele/jobmanager/JobManagerUtils.java
----------------------------------------------------------------------
diff --git a/stratosphere-runtime/src/main/java/eu/stratosphere/nephele/jobmanager/JobManagerUtils.java b/stratosphere-runtime/src/main/java/eu/stratosphere/nephele/jobmanager/JobManagerUtils.java
index 61cc22f..c576f20 100644
--- a/stratosphere-runtime/src/main/java/eu/stratosphere/nephele/jobmanager/JobManagerUtils.java
+++ b/stratosphere-runtime/src/main/java/eu/stratosphere/nephele/jobmanager/JobManagerUtils.java
@@ -16,11 +16,11 @@ package eu.stratosphere.nephele.jobmanager;
import java.lang.reflect.Constructor;
import java.lang.reflect.InvocationTargetException;
+import eu.stratosphere.nephele.ExecutionMode;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import eu.stratosphere.nephele.instance.InstanceManager;
-import eu.stratosphere.nephele.jobmanager.JobManager.ExecutionMode;
import eu.stratosphere.nephele.jobmanager.scheduler.AbstractScheduler;
import eu.stratosphere.util.StringUtils;
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/a70de7e3/stratosphere-runtime/src/main/java/eu/stratosphere/nephele/taskmanager/TaskManager.java
----------------------------------------------------------------------
diff --git a/stratosphere-runtime/src/main/java/eu/stratosphere/nephele/taskmanager/TaskManager.java b/stratosphere-runtime/src/main/java/eu/stratosphere/nephele/taskmanager/TaskManager.java
index 9b5fdf6..3da26f5 100644
--- a/stratosphere-runtime/src/main/java/eu/stratosphere/nephele/taskmanager/TaskManager.java
+++ b/stratosphere-runtime/src/main/java/eu/stratosphere/nephele/taskmanager/TaskManager.java
@@ -38,6 +38,10 @@ import java.util.concurrent.FutureTask;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
+import eu.stratosphere.nephele.ExecutionMode;
+import eu.stratosphere.runtime.io.network.LocalConnectionManager;
+import eu.stratosphere.runtime.io.network.NetworkConnectionManager;
+import eu.stratosphere.runtime.io.network.netty.NettyConnectionManager;
import org.apache.commons.cli.CommandLine;
import org.apache.commons.cli.CommandLineParser;
import org.apache.commons.cli.GnuParser;
@@ -155,10 +159,11 @@ public class TaskManager implements TaskOperationProtocol {
* receive an initial configuration. All parameters are obtained from the
* {@link GlobalConfiguration}, which must be loaded prior to instantiating the task manager.
*/
- public TaskManager() throws Exception {
+ public TaskManager(ExecutionMode executionMode) throws Exception {
LOG.info("TaskManager started as user " + UserGroupInformation.getCurrentUser().getShortUserName());
LOG.info("User system property: " + System.getProperty("user.name"));
+ LOG.info("Execution mode: " + executionMode);
// IMPORTANT! At this point, the GlobalConfiguration must have been read!
@@ -286,27 +291,40 @@ public class TaskManager implements TaskOperationProtocol {
ConfigConstants.TASK_MANAGER_NETWORK_BUFFER_SIZE_KEY,
ConfigConstants.DEFAULT_TASK_MANAGER_NETWORK_BUFFER_SIZE);
- int numInThreads = GlobalConfiguration.getInteger(
- ConfigConstants.TASK_MANAGER_NET_NUM_IN_THREADS_KEY,
- ConfigConstants.DEFAULT_TASK_MANAGER_NET_NUM_IN_THREADS);
- int numOutThreads = GlobalConfiguration.getInteger(
- ConfigConstants.TASK_MANAGER_NET_NUM_OUT_THREADS_KEY,
- ConfigConstants.DEFAULT_TASK_MANAGER_NET_NUM_OUT_THREADS);
-
- int lowWaterMark = GlobalConfiguration.getInteger(
- ConfigConstants.TASK_MANAGER_NET_NETTY_LOW_WATER_MARK,
- ConfigConstants.DEFAULT_TASK_MANAGER_NET_NETTY_LOW_WATER_MARK);
-
- int highWaterMark = GlobalConfiguration.getInteger(
- ConfigConstants.TASK_MANAGER_NET_NETTY_HIGH_WATER_MARK,
- ConfigConstants.DEFAULT_TASK_MANAGER_NET_NETTY_HIGH_WATER_MARK);
// Initialize the channel manager
try {
- this.channelManager = new ChannelManager(
- this.lookupService, this.localInstanceConnectionInfo,
- numBuffers, bufferSize, numInThreads, numOutThreads, lowWaterMark, highWaterMark);
+ NetworkConnectionManager networkConnectionManager = null;
+
+ switch (executionMode) {
+ case LOCAL:
+ networkConnectionManager = new LocalConnectionManager();
+ break;
+ case CLUSTER:
+ int numInThreads = GlobalConfiguration.getInteger(
+ ConfigConstants.TASK_MANAGER_NET_NUM_IN_THREADS_KEY,
+ ConfigConstants.DEFAULT_TASK_MANAGER_NET_NUM_IN_THREADS);
+
+ int numOutThreads = GlobalConfiguration.getInteger(
+ ConfigConstants.TASK_MANAGER_NET_NUM_OUT_THREADS_KEY,
+ ConfigConstants.DEFAULT_TASK_MANAGER_NET_NUM_OUT_THREADS);
+
+ int lowWaterMark = GlobalConfiguration.getInteger(
+ ConfigConstants.TASK_MANAGER_NET_NETTY_LOW_WATER_MARK,
+ ConfigConstants.DEFAULT_TASK_MANAGER_NET_NETTY_LOW_WATER_MARK);
+
+ int highWaterMark = GlobalConfiguration.getInteger(
+ ConfigConstants.TASK_MANAGER_NET_NETTY_HIGH_WATER_MARK,
+ ConfigConstants.DEFAULT_TASK_MANAGER_NET_NETTY_HIGH_WATER_MARK);
+
+ networkConnectionManager = new NettyConnectionManager(
+ localInstanceConnectionInfo.address(), localInstanceConnectionInfo.dataPort(),
+ bufferSize, numInThreads, numOutThreads, lowWaterMark, highWaterMark);
+ break;
+ }
+
+ channelManager = new ChannelManager(lookupService, localInstanceConnectionInfo, numBuffers, bufferSize, networkConnectionManager);
} catch (IOException ioe) {
LOG.error(StringUtils.stringifyException(ioe));
throw new Exception("Failed to instantiate channel manager. " + ioe.getMessage(), ioe);
@@ -436,7 +454,7 @@ public class TaskManager implements TaskOperationProtocol {
// Create a new task manager object
try {
- new TaskManager();
+ new TaskManager(ExecutionMode.CLUSTER);
} catch (Exception e) {
LOG.fatal("Taskmanager startup failed: " + e.getMessage(), e);
System.exit(FAILURE_RETURN_CODE);
@@ -910,8 +928,12 @@ public class TaskManager implements TaskOperationProtocol {
this.profiler.shutdown();
}
- // Shut down the network channel manager
- this.channelManager.shutdown();
+ // Shut down the channel manager
+ try {
+ this.channelManager.shutdown();
+ } catch (IOException e) {
+ LOG.warn("ChannelManager did not shutdown properly: " + e.getMessage(), e);
+ }
// Shut down the memory manager
if (this.ioManager != null) {
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/a70de7e3/stratosphere-runtime/src/main/java/eu/stratosphere/runtime/io/network/ChannelManager.java
----------------------------------------------------------------------
diff --git a/stratosphere-runtime/src/main/java/eu/stratosphere/runtime/io/network/ChannelManager.java b/stratosphere-runtime/src/main/java/eu/stratosphere/runtime/io/network/ChannelManager.java
index 7b01f4e..34d3501 100644
--- a/stratosphere-runtime/src/main/java/eu/stratosphere/runtime/io/network/ChannelManager.java
+++ b/stratosphere-runtime/src/main/java/eu/stratosphere/runtime/io/network/ChannelManager.java
@@ -37,7 +37,6 @@ import eu.stratosphere.runtime.io.network.bufferprovider.LocalBufferPoolOwner;
import eu.stratosphere.runtime.io.gates.GateID;
import eu.stratosphere.runtime.io.gates.InputGate;
import eu.stratosphere.runtime.io.gates.OutputGate;
-import eu.stratosphere.runtime.io.network.netty.NettyConnectionManager;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@@ -66,7 +65,7 @@ public class ChannelManager implements EnvelopeDispatcher, BufferProviderBroker
private final GlobalBufferPool globalBufferPool;
- private final NettyConnectionManager nettyConnectionManager;
+ private final NetworkConnectionManager networkConnectionManager;
private final InetSocketAddress ourAddress;
@@ -75,18 +74,15 @@ public class ChannelManager implements EnvelopeDispatcher, BufferProviderBroker
// -----------------------------------------------------------------------------------------------------------------
public ChannelManager(ChannelLookupProtocol channelLookupService, InstanceConnectionInfo connectionInfo,
- int numNetworkBuffers, int networkBufferSize,
- int numInThreads, int numOutThreads,
- int lowWatermark, int highWaterMark) throws IOException {
+ int numNetworkBuffers, int networkBufferSize, NetworkConnectionManager networkConnectionManager) throws IOException {
this.channelLookupService = channelLookupService;
this.connectionInfo = connectionInfo;
this.globalBufferPool = new GlobalBufferPool(numNetworkBuffers, networkBufferSize);
- this.nettyConnectionManager = new NettyConnectionManager(
- this, connectionInfo.address(), connectionInfo.dataPort(),
- networkBufferSize, numInThreads, numOutThreads, lowWatermark, highWaterMark);
+ this.networkConnectionManager = networkConnectionManager;
+ networkConnectionManager.start(this);
// management data structures
this.channels = new ConcurrentHashMap<ChannelID, Channel>();
@@ -99,8 +95,9 @@ public class ChannelManager implements EnvelopeDispatcher, BufferProviderBroker
this.discardBufferPool = new DiscardBufferPool();
}
- public void shutdown() {
- this.nettyConnectionManager.shutdown();
+ public void shutdown() throws IOException {
+ this.networkConnectionManager.shutdown();
+
this.globalBufferPool.destroy();
}
@@ -324,7 +321,7 @@ public class ChannelManager implements EnvelopeDispatcher, BufferProviderBroker
final RemoteReceiver ourAddress = new RemoteReceiver(this.ourAddress, connectionIndex);
final Envelope senderHint = SenderHintEvent.createEnvelopeWithEvent(envelope, targetChannelID, ourAddress);
- this.nettyConnectionManager.enqueue(senderHint, receiver);
+ this.networkConnectionManager.enqueue(senderHint, receiver);
}
/**
@@ -459,7 +456,7 @@ public class ChannelManager implements EnvelopeDispatcher, BufferProviderBroker
generateSenderHint(envelope, remoteReceiver);
}
- this.nettyConnectionManager.enqueue(envelope, remoteReceiver);
+ this.networkConnectionManager.enqueue(envelope, remoteReceiver);
success = true;
}
} finally {
@@ -507,7 +504,7 @@ public class ChannelManager implements EnvelopeDispatcher, BufferProviderBroker
generateSenderHint(envelope, remoteReceiver);
}
- this.nettyConnectionManager.enqueue(envelope, remoteReceiver);
+ this.networkConnectionManager.enqueue(envelope, remoteReceiver);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/a70de7e3/stratosphere-runtime/src/main/java/eu/stratosphere/runtime/io/network/LocalConnectionManager.java
----------------------------------------------------------------------
diff --git a/stratosphere-runtime/src/main/java/eu/stratosphere/runtime/io/network/LocalConnectionManager.java b/stratosphere-runtime/src/main/java/eu/stratosphere/runtime/io/network/LocalConnectionManager.java
new file mode 100644
index 0000000..85cbbd2
--- /dev/null
+++ b/stratosphere-runtime/src/main/java/eu/stratosphere/runtime/io/network/LocalConnectionManager.java
@@ -0,0 +1,31 @@
+/***********************************************************************************************************************
+ * Copyright (C) 2010-2014 by the Stratosphere project (http://stratosphere.eu)
+ *
+ * Licensed 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 eu.stratosphere.runtime.io.network;
+
+import java.io.IOException;
+
+public class LocalConnectionManager implements NetworkConnectionManager {
+
+ @Override
+ public void start(ChannelManager channelManager) throws IOException {
+ }
+
+ @Override
+ public void enqueue(Envelope envelope, RemoteReceiver receiver) throws IOException {
+ }
+
+ @Override
+ public void shutdown() throws IOException {
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/a70de7e3/stratosphere-runtime/src/main/java/eu/stratosphere/runtime/io/network/NetworkConnectionManager.java
----------------------------------------------------------------------
diff --git a/stratosphere-runtime/src/main/java/eu/stratosphere/runtime/io/network/NetworkConnectionManager.java b/stratosphere-runtime/src/main/java/eu/stratosphere/runtime/io/network/NetworkConnectionManager.java
new file mode 100644
index 0000000..d2d252d
--- /dev/null
+++ b/stratosphere-runtime/src/main/java/eu/stratosphere/runtime/io/network/NetworkConnectionManager.java
@@ -0,0 +1,25 @@
+/***********************************************************************************************************************
+ * Copyright (C) 2010-2014 by the Stratosphere project (http://stratosphere.eu)
+ *
+ * Licensed 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 eu.stratosphere.runtime.io.network;
+
+import java.io.IOException;
+
+public interface NetworkConnectionManager {
+
+ public void start(ChannelManager channelManager) throws IOException;
+
+ public void enqueue(Envelope envelope, RemoteReceiver receiver) throws IOException;
+
+ public void shutdown() throws IOException;
+}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/a70de7e3/stratosphere-runtime/src/main/java/eu/stratosphere/runtime/io/network/netty/InboundEnvelopeDispatcher.java
----------------------------------------------------------------------
diff --git a/stratosphere-runtime/src/main/java/eu/stratosphere/runtime/io/network/netty/InboundEnvelopeDispatcher.java b/stratosphere-runtime/src/main/java/eu/stratosphere/runtime/io/network/netty/InboundEnvelopeDispatcher.java
new file mode 100644
index 0000000..1c360c1
--- /dev/null
+++ b/stratosphere-runtime/src/main/java/eu/stratosphere/runtime/io/network/netty/InboundEnvelopeDispatcher.java
@@ -0,0 +1,35 @@
+/***********************************************************************************************************************
+ * Copyright (C) 2010-2014 by the Stratosphere project (http://stratosphere.eu)
+ *
+ * Licensed 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 eu.stratosphere.runtime.io.network.netty;
+
+import eu.stratosphere.runtime.io.network.Envelope;
+import eu.stratosphere.runtime.io.network.EnvelopeDispatcher;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+
+public class InboundEnvelopeDispatcher extends ChannelInboundHandlerAdapter {
+
+ private final EnvelopeDispatcher envelopeDispatcher;
+
+ public InboundEnvelopeDispatcher(EnvelopeDispatcher envelopeDispatcher) {
+ this.envelopeDispatcher = envelopeDispatcher;
+ }
+
+ @Override
+ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
+ Envelope envelope = (Envelope) msg;
+
+ envelopeDispatcher.dispatchFromNetwork(envelope);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/a70de7e3/stratosphere-runtime/src/main/java/eu/stratosphere/runtime/io/network/netty/InboundEnvelopeDispatcherHandler.java
----------------------------------------------------------------------
diff --git a/stratosphere-runtime/src/main/java/eu/stratosphere/runtime/io/network/netty/InboundEnvelopeDispatcherHandler.java b/stratosphere-runtime/src/main/java/eu/stratosphere/runtime/io/network/netty/InboundEnvelopeDispatcherHandler.java
deleted file mode 100644
index d0270b6..0000000
--- a/stratosphere-runtime/src/main/java/eu/stratosphere/runtime/io/network/netty/InboundEnvelopeDispatcherHandler.java
+++ /dev/null
@@ -1,41 +0,0 @@
-/***********************************************************************************************************************
- * Copyright (C) 2010-2014 by the Stratosphere project (http://stratosphere.eu)
- *
- * Licensed 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 eu.stratosphere.runtime.io.network.netty;
-
-import eu.stratosphere.runtime.io.network.Envelope;
-import eu.stratosphere.runtime.io.network.EnvelopeDispatcher;
-import io.netty.channel.ChannelHandlerContext;
-import io.netty.channel.ChannelInboundHandlerAdapter;
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-
-public class InboundEnvelopeDispatcherHandler extends ChannelInboundHandlerAdapter {
-
- private static final Log LOG = LogFactory.getLog(InboundEnvelopeDispatcherHandler.class);
-
- private final EnvelopeDispatcher channelManager;
-
- public InboundEnvelopeDispatcherHandler(EnvelopeDispatcher channelManager) {
- this.channelManager = channelManager;
- }
-
- @Override
- public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
- Envelope envelope = (Envelope) msg;
-// LOG.debug(String.format("Decoded envelope with seq num %d from source channel %s",
-// envelope.getSequenceNumber(),
-// envelope.getSource()));
- this.channelManager.dispatchFromNetwork(envelope);
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/a70de7e3/stratosphere-runtime/src/main/java/eu/stratosphere/runtime/io/network/netty/NettyConnectionManager.java
----------------------------------------------------------------------
diff --git a/stratosphere-runtime/src/main/java/eu/stratosphere/runtime/io/network/netty/NettyConnectionManager.java b/stratosphere-runtime/src/main/java/eu/stratosphere/runtime/io/network/netty/NettyConnectionManager.java
index 4e3f043..73afcbc 100644
--- a/stratosphere-runtime/src/main/java/eu/stratosphere/runtime/io/network/netty/NettyConnectionManager.java
+++ b/stratosphere-runtime/src/main/java/eu/stratosphere/runtime/io/network/netty/NettyConnectionManager.java
@@ -15,7 +15,10 @@ package eu.stratosphere.runtime.io.network.netty;
import eu.stratosphere.runtime.io.network.ChannelManager;
import eu.stratosphere.runtime.io.network.Envelope;
+import eu.stratosphere.runtime.io.network.EnvelopeDispatcher;
+import eu.stratosphere.runtime.io.network.NetworkConnectionManager;
import eu.stratosphere.runtime.io.network.RemoteReceiver;
+import eu.stratosphere.runtime.io.network.bufferprovider.BufferProviderBroker;
import io.netty.bootstrap.Bootstrap;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.PooledByteBufAllocator;
@@ -28,7 +31,6 @@ import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
-import io.netty.util.concurrent.Future;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@@ -39,50 +41,70 @@ import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
-public class NettyConnectionManager {
+public class NettyConnectionManager implements NetworkConnectionManager {
private static final Log LOG = LogFactory.getLog(NettyConnectionManager.class);
private static final int DEBUG_PRINT_QUEUED_ENVELOPES_EVERY_MS = 10000;
- private final ChannelManager channelManager;
+ private final ConcurrentMap<RemoteReceiver, Object> outConnections = new ConcurrentHashMap<RemoteReceiver, Object>();
- private final ServerBootstrap in;
+ private final InetAddress bindAddress;
- private final Bootstrap out;
+ private final int bindPort;
- private final ConcurrentMap<RemoteReceiver, Object> outConnections;
+ private final int bufferSize;
- public NettyConnectionManager(ChannelManager channelManager, InetAddress bindAddress, int bindPort,
- int bufferSize, int numInThreads, int numOutThreads,
- int lowWaterMark, int highWaterMark) {
- this.outConnections = new ConcurrentHashMap<RemoteReceiver, Object>();
- this.channelManager = channelManager;
+ private final int numInThreads;
- // --------------------------------------------------------------------
+ private final int numOutThreads;
+
+ private final int lowWaterMark;
+
+ private final int highWaterMark;
+
+ private ServerBootstrap in;
+
+ private Bootstrap out;
+
+ public NettyConnectionManager(InetAddress bindAddress, int bindPort, int bufferSize, int numInThreads,
+ int numOutThreads, int lowWaterMark, int highWaterMark) {
+
+ this.bindAddress = bindAddress;
+ this.bindPort = bindPort;
+
+ this.bufferSize = bufferSize;
int defaultNumThreads = Math.max(Runtime.getRuntime().availableProcessors() / 4, 1);
- numInThreads = (numInThreads == -1) ? defaultNumThreads : numInThreads;
- numOutThreads = (numOutThreads == -1) ? defaultNumThreads : numOutThreads;
- LOG.info(String.format("Starting with %d incoming and %d outgoing connection threads.", numInThreads, numOutThreads));
- lowWaterMark = (lowWaterMark == -1) ? bufferSize / 2 : lowWaterMark;
- highWaterMark = (highWaterMark == -1) ? bufferSize : highWaterMark;
+ this.numInThreads = (numInThreads == -1) ? defaultNumThreads : numInThreads;
+ this.numOutThreads = (numOutThreads == -1) ? defaultNumThreads : numOutThreads;
+
+ this.lowWaterMark = (lowWaterMark == -1) ? bufferSize / 2 : lowWaterMark;
+ this.highWaterMark = (highWaterMark == -1) ? bufferSize : highWaterMark;
+ }
+
+ @Override
+ public void start(ChannelManager channelManager) throws IOException {
+ LOG.info(String.format("Starting with %d incoming and %d outgoing connection threads.", numInThreads, numOutThreads));
LOG.info(String.format("Setting low water mark to %d and high water mark to %d bytes.", lowWaterMark, highWaterMark));
+ final BufferProviderBroker bufferProviderBroker = channelManager;
+ final EnvelopeDispatcher envelopeDispatcher = channelManager;
+
// --------------------------------------------------------------------
// server bootstrap (incoming connections)
// --------------------------------------------------------------------
- this.in = new ServerBootstrap();
- this.in.group(new NioEventLoopGroup(numInThreads))
+ in = new ServerBootstrap();
+ in.group(new NioEventLoopGroup(numInThreads))
.channel(NioServerSocketChannel.class)
.localAddress(bindAddress, bindPort)
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel channel) throws Exception {
channel.pipeline()
- .addLast(new InboundEnvelopeDecoder(NettyConnectionManager.this.channelManager))
- .addLast(new InboundEnvelopeDispatcherHandler(NettyConnectionManager.this.channelManager));
+ .addLast(new InboundEnvelopeDecoder(bufferProviderBroker))
+ .addLast(new InboundEnvelopeDispatcher(envelopeDispatcher));
}
})
.option(ChannelOption.RCVBUF_ALLOCATOR, new FixedRecvByteBufAllocator(bufferSize))
@@ -91,8 +113,8 @@ public class NettyConnectionManager {
// --------------------------------------------------------------------
// client bootstrap (outgoing connections)
// --------------------------------------------------------------------
- this.out = new Bootstrap();
- this.out.group(new NioEventLoopGroup(numOutThreads))
+ out = new Bootstrap();
+ out.group(new NioEventLoopGroup(numOutThreads))
.channel(NioSocketChannel.class)
.handler(new ChannelInitializer<SocketChannel>() {
@Override
@@ -108,9 +130,9 @@ public class NettyConnectionManager {
.option(ChannelOption.SO_KEEPALIVE, true);
try {
- this.in.bind().sync();
+ in.bind().sync();
} catch (InterruptedException e) {
- throw new RuntimeException("Could not bind server socket for incoming connections.");
+ throw new IOException(e);
}
if (LOG.isDebugEnabled()) {
@@ -119,16 +141,14 @@ public class NettyConnectionManager {
public void run() {
Date date = new Date();
-
while (true) {
try {
Thread.sleep(DEBUG_PRINT_QUEUED_ENVELOPES_EVERY_MS);
date.setTime(System.currentTimeMillis());
- System.out.println(date);
+ System.out.println(date);
System.out.println(getNonZeroNumQueuedEnvelopes());
-
} catch (InterruptedException e) {
e.printStackTrace();
}
@@ -138,18 +158,7 @@ public class NettyConnectionManager {
}
}
- public void shutdown() {
- Future<?> inShutdownFuture = this.in.group().shutdownGracefully();
- Future<?> outShutdownFuture = this.out.group().shutdownGracefully();
-
- try {
- inShutdownFuture.sync();
- outShutdownFuture.sync();
- } catch (InterruptedException e) {
- throw new RuntimeException("Could not properly shutdown connections.");
- }
- }
-
+ @Override
public void enqueue(Envelope envelope, RemoteReceiver receiver) throws IOException {
// Get the channel. The channel may be
// 1) a channel that already exists (usual case) -> just send the data
@@ -198,6 +207,29 @@ public class NettyConnectionManager {
channel.enqueue(envelope);
}
+ @Override
+ public void shutdown() throws IOException {
+ if (!in.group().isShuttingDown()) {
+ LOG.info("Shutting down incoming connections.");
+
+ try {
+ in.group().shutdownGracefully().sync();
+ } catch (InterruptedException e) {
+ throw new IOException(e);
+ }
+ }
+
+ if (!out.group().isShuttingDown()) {
+ LOG.info("Shutting down outgoing connections.");
+
+ try {
+ out.group().shutdownGracefully().sync();
+ } catch (InterruptedException e) {
+ throw new IOException(e);
+ }
+ }
+ }
+
private String getNonZeroNumQueuedEnvelopes() {
StringBuilder str = new StringBuilder();
@@ -211,9 +243,10 @@ public class NettyConnectionManager {
OutboundConnectionQueue queue = (OutboundConnectionQueue) value;
if (queue.getNumQueuedEnvelopes() > 0) {
str.append(String.format("%s> Number of queued envelopes for %s with channel %s: %d\n",
- Thread.currentThread().getId(), receiver, queue.getChannel(), queue.getNumQueuedEnvelopes()));
+ Thread.currentThread().getId(), receiver, queue.toString(), queue.getNumQueuedEnvelopes()));
}
- } else if (value instanceof ChannelInBuildup) {
+ }
+ else if (value instanceof ChannelInBuildup) {
str.append(String.format("%s> Connection to %s is still in buildup\n",
Thread.currentThread().getId(), receiver));
}
@@ -226,13 +259,13 @@ public class NettyConnectionManager {
private static final class ChannelInBuildup implements ChannelFutureListener {
- private Object lock = new Object();
+ private final Object lock = new Object();
private volatile OutboundConnectionQueue channel;
private volatile Throwable error;
- private int numRetries = 2;
+ private int numRetries = 3;
private final Bootstrap out;
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/a70de7e3/stratosphere-runtime/src/main/java/eu/stratosphere/runtime/io/network/netty/OutboundConnectionQueue.java
----------------------------------------------------------------------
diff --git a/stratosphere-runtime/src/main/java/eu/stratosphere/runtime/io/network/netty/OutboundConnectionQueue.java b/stratosphere-runtime/src/main/java/eu/stratosphere/runtime/io/network/netty/OutboundConnectionQueue.java
index b6ec915..8fef3c1 100644
--- a/stratosphere-runtime/src/main/java/eu/stratosphere/runtime/io/network/netty/OutboundConnectionQueue.java
+++ b/stratosphere-runtime/src/main/java/eu/stratosphere/runtime/io/network/netty/OutboundConnectionQueue.java
@@ -33,7 +33,7 @@ public class OutboundConnectionQueue extends ChannelInboundHandlerAdapter implem
private final ArrayDeque<Envelope> queuedEnvelopes = new ArrayDeque<Envelope>();
- private final AtomicInteger numQueued = new AtomicInteger(0);
+ private final AtomicInteger numQueuedEnvelopes = new AtomicInteger(0);
public OutboundConnectionQueue(Channel channel) {
this.channel = channel;
@@ -58,7 +58,7 @@ public class OutboundConnectionQueue extends ChannelInboundHandlerAdapter implem
boolean triggerWrite = this.queuedEnvelopes.isEmpty();
this.queuedEnvelopes.addLast((Envelope) envelopeToEnqueue);
- this.numQueued.incrementAndGet();
+ this.numQueuedEnvelopes.incrementAndGet();
if (triggerWrite) {
writeAndFlushNextEnvelopeIfPossible();
@@ -84,17 +84,18 @@ public class OutboundConnectionQueue extends ChannelInboundHandlerAdapter implem
}
public int getNumQueuedEnvelopes() {
- return this.numQueued.intValue();
+ return this.numQueuedEnvelopes.intValue();
}
- public Channel getChannel() {
- return this.channel;
+ @Override
+ public String toString() {
+ return channel.toString();
}
private void writeAndFlushNextEnvelopeIfPossible() {
if (this.channel.isWritable() && !this.queuedEnvelopes.isEmpty()) {
Envelope nextEnvelope = this.queuedEnvelopes.pollFirst();
- this.numQueued.decrementAndGet();
+ this.numQueuedEnvelopes.decrementAndGet();
this.channel.writeAndFlush(nextEnvelope).addListener(this);
}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/a70de7e3/stratosphere-runtime/src/test/java/eu/stratosphere/nephele/instance/local/LocalInstanceManagerTest.java
----------------------------------------------------------------------
diff --git a/stratosphere-runtime/src/test/java/eu/stratosphere/nephele/instance/local/LocalInstanceManagerTest.java b/stratosphere-runtime/src/test/java/eu/stratosphere/nephele/instance/local/LocalInstanceManagerTest.java
index 1577574..92ea5ab 100644
--- a/stratosphere-runtime/src/test/java/eu/stratosphere/nephele/instance/local/LocalInstanceManagerTest.java
+++ b/stratosphere-runtime/src/test/java/eu/stratosphere/nephele/instance/local/LocalInstanceManagerTest.java
@@ -15,6 +15,8 @@ package eu.stratosphere.nephele.instance.local;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.fail;
+
+import eu.stratosphere.nephele.ExecutionMode;
import junit.framework.Assert;
import org.junit.Test;
@@ -22,7 +24,6 @@ import org.junit.Test;
import eu.stratosphere.configuration.GlobalConfiguration;
import eu.stratosphere.nephele.instance.InstanceType;
import eu.stratosphere.nephele.jobmanager.JobManager;
-import eu.stratosphere.nephele.jobmanager.JobManager.ExecutionMode;
import eu.stratosphere.nephele.util.ServerTestUtils;
/**
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/a70de7e3/stratosphere-runtime/src/test/java/eu/stratosphere/nephele/jobmanager/JobManagerITCase.java
----------------------------------------------------------------------
diff --git a/stratosphere-runtime/src/test/java/eu/stratosphere/nephele/jobmanager/JobManagerITCase.java b/stratosphere-runtime/src/test/java/eu/stratosphere/nephele/jobmanager/JobManagerITCase.java
index 06c857e..063b827 100644
--- a/stratosphere-runtime/src/test/java/eu/stratosphere/nephele/jobmanager/JobManagerITCase.java
+++ b/stratosphere-runtime/src/test/java/eu/stratosphere/nephele/jobmanager/JobManagerITCase.java
@@ -17,6 +17,7 @@ import eu.stratosphere.configuration.ConfigConstants;
import eu.stratosphere.configuration.Configuration;
import eu.stratosphere.configuration.GlobalConfiguration;
import eu.stratosphere.core.fs.Path;
+import eu.stratosphere.nephele.ExecutionMode;
import eu.stratosphere.nephele.client.JobClient;
import eu.stratosphere.nephele.client.JobExecutionException;
import eu.stratosphere.nephele.jobgraph.DistributionPattern;
@@ -26,7 +27,6 @@ import eu.stratosphere.nephele.jobgraph.JobFileOutputVertex;
import eu.stratosphere.nephele.jobgraph.JobGraph;
import eu.stratosphere.nephele.jobgraph.JobGraphDefinitionException;
import eu.stratosphere.nephele.jobgraph.JobTaskVertex;
-import eu.stratosphere.nephele.jobmanager.JobManager.ExecutionMode;
import eu.stratosphere.nephele.taskmanager.Task;
import eu.stratosphere.nephele.taskmanager.TaskManager;
import eu.stratosphere.nephele.util.FileLineReader;
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/a70de7e3/stratosphere-runtime/src/test/java/eu/stratosphere/runtime/io/network/netty/NettyConnectionManagerTest.java
----------------------------------------------------------------------
diff --git a/stratosphere-runtime/src/test/java/eu/stratosphere/runtime/io/network/netty/NettyConnectionManagerTest.java b/stratosphere-runtime/src/test/java/eu/stratosphere/runtime/io/network/netty/NettyConnectionManagerTest.java
index d23b0f3..dbb4c3f 100644
--- a/stratosphere-runtime/src/test/java/eu/stratosphere/runtime/io/network/netty/NettyConnectionManagerTest.java
+++ b/stratosphere-runtime/src/test/java/eu/stratosphere/runtime/io/network/netty/NettyConnectionManagerTest.java
@@ -28,12 +28,11 @@ import org.mockito.stubbing.Answer;
import java.io.IOException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
-import java.util.Arrays;
-import java.util.Collection;
import java.util.Random;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicInteger;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.mock;
@@ -47,113 +46,136 @@ public class NettyConnectionManagerTest {
private final static int BIND_PORT = 20000;
- private final static int HIGH_WATERMARK = 32 * 1024;
+ private final static int BUFFER_SIZE = 32 * 1024;
- private int numSubtasks;
-
- private int numToSendPerSubtask;
-
- private int numInThreads;
+ public void testEnqueueRaceAndDeadlockFreeMultipleChannels() throws Exception {
+ Integer[][] configs = new Integer[][]{
+ {64, 4096, 1, 1, 1},
+ {128, 2048, 1, 1, 1},
+ {256, 1024, 1, 1, 1},
+ {512, 512, 1, 1, 1},
+ {64, 4096, 4, 1, 1},
+ {128, 2048, 4, 1, 1},
+ {256, 1024, 4, 1, 1},
+ {512, 512, 4, 1, 1},
+ {64, 4096, 4, 2, 2},
+ {128, 2048, 4, 2, 2},
+ {256, 1024, 4, 2, 2},
+ {512, 512, 4, 2, 2}
+ };
- private int numOutThreads;
+ for (Integer[] params : configs) {
+ System.out.println(String.format("Running %s with config: %d sub tasks, %d envelopes to send per subtasks, "
+ + "%d num channels, %d num in threads, %d num out threads.",
+ "testEnqueueRaceAndDeadlockFreeMultipleChannels", params[0], params[1], params[2], params[3], params[4]));
- private int numChannels;
+ long start = System.currentTimeMillis();
+ doTestEnqueueRaceAndDeadlockFreeMultipleChannels(params[0], params[1], params[2], params[3], params[4]);
+ long end = System.currentTimeMillis();
- public NettyConnectionManagerTest(int numSubtasks, int numToSendPerSubtask, int numChannels, int numInThreads, int numOutThreads) {
- this.numSubtasks = numSubtasks;
- this.numToSendPerSubtask = numToSendPerSubtask;
- this.numChannels = numChannels;
- this.numInThreads = numInThreads;
- this.numOutThreads = numOutThreads;
+ System.out.println(String.format("Runtime: %d ms.", (end - start)));
+ }
}
- @Parameterized.Parameters
- public static Collection<Integer[]> configure() {
- return Arrays.asList(
- new Integer[][]{
- {64, 4096, 1, 1, 1},
- {128, 2048, 1, 1, 1},
- {256, 1024, 1, 1, 1},
- {512, 512, 1, 1, 1},
- {64, 4096, 4, 1, 1},
- {128, 2048, 4, 1, 1},
- {256, 1024, 4, 1, 1},
- {512, 512, 4, 1, 1},
- {64, 4096, 4, 2, 2},
- {128, 2048, 4, 2, 2},
- {256, 1024, 4, 2, 2},
- {512, 512, 4, 2, 2}
- }
- );
- }
+ private void doTestEnqueueRaceAndDeadlockFreeMultipleChannels(
+ int numSubtasks, final int numToSendPerSubtask, int numChannels, int numInThreads, int numOutThreads)
+ throws Exception {
- public void testEnqueueRaceAndDeadlockFreeMultipleChannels() throws Exception {
final InetAddress localhost = InetAddress.getLocalHost();
- final CountDownLatch latch = new CountDownLatch(this.numSubtasks);
+ final CountDownLatch latch = new CountDownLatch(numSubtasks);
// --------------------------------------------------------------------
// setup
// --------------------------------------------------------------------
ChannelManager channelManager = mock(ChannelManager.class);
- doAnswer(new VerifyEnvelopes(latch)).when(channelManager).dispatchFromNetwork(Matchers.<Envelope>anyObject());
+ doAnswer(new VerifyEnvelopes(latch, numToSendPerSubtask))
+ .when(channelManager).dispatchFromNetwork(Matchers.<Envelope>anyObject());
- NettyConnectionManager connManagerToTest = new NettyConnectionManager(channelManager, localhost,
- BIND_PORT, HIGH_WATERMARK, this.numInThreads, this.numOutThreads, -1, -1);
+ final NettyConnectionManager senderConnManager = new NettyConnectionManager(localhost, BIND_PORT, BUFFER_SIZE,
+ numInThreads, numOutThreads, -1, -1);
+ senderConnManager.start(channelManager);
- NettyConnectionManager connManagerReceiver = new NettyConnectionManager(channelManager, localhost,
- BIND_PORT + 1, HIGH_WATERMARK, this.numInThreads, this.numOutThreads, -1, -1);
+ NettyConnectionManager receiverConnManager = new NettyConnectionManager(localhost, BIND_PORT + 1, BUFFER_SIZE,
+ numInThreads, numOutThreads, -1, -1);
+ receiverConnManager.start(channelManager);
// --------------------------------------------------------------------
// start sender threads
// --------------------------------------------------------------------
- RemoteReceiver[] receivers = new RemoteReceiver[this.numChannels];
+ RemoteReceiver[] receivers = new RemoteReceiver[numChannels];
- for (int i = 0; i < this.numChannels; i++) {
+ for (int i = 0; i < numChannels; i++) {
receivers[i] = new RemoteReceiver(new InetSocketAddress(localhost, BIND_PORT + 1), i);
}
- for (int i = 0; i < this.numSubtasks; i++) {
- RemoteReceiver receiver = receivers[random.nextInt(this.numChannels)];
- new Thread(new SubtaskSenderThread(connManagerToTest, receiver)).start();
+ for (int i = 0; i < numSubtasks; i++) {
+ final RemoteReceiver receiver = receivers[random.nextInt(numChannels)];
+
+ final AtomicInteger seqNum = new AtomicInteger(0);
+ final JobID jobId = new JobID();
+ final ChannelID channelId = new ChannelID();
+
+ new Thread(new Runnable() {
+ @Override
+ public void run() {
+ // enqueue envelopes with ascending seq numbers
+ while (seqNum.get() < numToSendPerSubtask) {
+ try {
+ Envelope env = new Envelope(seqNum.getAndIncrement(), jobId, channelId);
+ senderConnManager.enqueue(env, receiver);
+ } catch (IOException e) {
+ throw new RuntimeException("Unexpected exception while enqueuing envelope.");
+ }
+ }
+ }
+ }).start();
}
latch.await();
- connManagerToTest.shutdown();
- connManagerReceiver.shutdown();
+ senderConnManager.shutdown();
+ receiverConnManager.shutdown();
}
-
- private class VerifyEnvelopes implements Answer<Object> {
+ /**
+ * Verifies correct ordering of received envelopes (per envelope source channel ID).
+ */
+ private class VerifyEnvelopes implements Answer<Void> {
private final ConcurrentMap<ChannelID, Integer> received = new ConcurrentHashMap<ChannelID, Integer>();
private final CountDownLatch latch;
- private VerifyEnvelopes(CountDownLatch latch) {
+ private final int numExpectedEnvelopesPerSubtask;
+
+ private VerifyEnvelopes(CountDownLatch latch, int numExpectedEnvelopesPerSubtask) {
this.latch = latch;
+ this.numExpectedEnvelopesPerSubtask = numExpectedEnvelopesPerSubtask;
}
@Override
- public Object answer(InvocationOnMock invocation) throws Throwable {
+ public Void answer(InvocationOnMock invocation) throws Throwable {
Envelope env = (Envelope) invocation.getArguments()[0];
ChannelID channelId = env.getSource();
int seqNum = env.getSequenceNumber();
if (seqNum == 0) {
- Assert.assertNull(
- String.format("Received envelope from %s before, but current seq num is 0", channelId),
- this.received.putIfAbsent(channelId, seqNum));
+ Integer previousSeqNum = this.received.putIfAbsent(channelId, seqNum);
+
+ String msg = String.format("Received envelope from %s before, but current seq num is 0", channelId);
+ Assert.assertNull(msg, previousSeqNum);
}
else {
- Assert.assertTrue(
- String.format("Received seq num %d from %s, but previous was not %d", seqNum, channelId, seqNum - 1),
- this.received.replace(channelId, seqNum - 1, seqNum));
+ boolean isExpectedPreviousSeqNum = this.received.replace(channelId, seqNum - 1, seqNum);
+
+ String msg = String.format("Received seq num %d from %s, but previous was not %d.",
+ seqNum, channelId, seqNum - 1);
+ Assert.assertTrue(msg, isExpectedPreviousSeqNum);
}
// count down the latch if all envelopes received for this source
- if (seqNum == numToSendPerSubtask - 1) {
+ if (seqNum == numExpectedEnvelopesPerSubtask - 1) {
this.latch.countDown();
}
@@ -161,54 +183,13 @@ public class NettyConnectionManagerTest {
}
}
- private class SubtaskSenderThread implements Runnable {
-
- private final NettyConnectionManager connectionManager;
-
- private final RemoteReceiver receiver;
-
- private final JobID jobId = new JobID();
-
- private final ChannelID channelId = new ChannelID();
-
- private int seqNum = 0;
-
- private SubtaskSenderThread(NettyConnectionManager connectionManager, RemoteReceiver receiver) {
- this.connectionManager = connectionManager;
- this.receiver = receiver;
- }
-
- @Override
- public void run() {
- // enqueue envelopes with ascending seq nums
- while (this.seqNum < numToSendPerSubtask) {
- try {
- Envelope env = new Envelope(this.seqNum++, this.jobId, this.channelId);
- this.connectionManager.enqueue(env, receiver);
- } catch (IOException e) {
- throw new RuntimeException("Unexpected exception while enqueing envelope");
- }
- }
- }
- }
-
private void runAllTests() throws Exception {
- System.out.println(String.format("Running tests with config: %d sub tasks, %d envelopes to send per subtasks, "
- + "%d num channels, %d num in threads, %d num out threads.",
- this.numSubtasks, this.numToSendPerSubtask, this.numChannels, this.numInThreads, this.numOutThreads));
-
testEnqueueRaceAndDeadlockFreeMultipleChannels();
System.out.println("Done.");
}
public static void main(String[] args) throws Exception {
- Collection<Integer[]> configs = configure();
-
- for (Integer[] params : configs) {
-
- NettyConnectionManagerTest test = new NettyConnectionManagerTest(params[0], params[1], params[2], params[3], params[4]);
- test.runAllTests();
- }
+ new NettyConnectionManagerTest().runAllTests();
}
}