You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hama.apache.org by tj...@apache.org on 2011/10/20 18:58:19 UTC
svn commit: r1186921 [2/2] - in /incubator/hama/trunk: ./ core/conf/
core/src/main/java/org/apache/hama/bsp/
core/src/main/java/org/apache/hama/bsp/sync/
core/src/main/java/org/apache/hama/bsp/sync/rpc/
core/src/main/java/org/apache/hama/bsp/sync/zooke...
Modified: incubator/hama/trunk/core/src/main/java/org/apache/hama/checkpoint/CheckpointRunner.java
URL: http://svn.apache.org/viewvc/incubator/hama/trunk/core/src/main/java/org/apache/hama/checkpoint/CheckpointRunner.java?rev=1186921&r1=1186920&r2=1186921&view=diff
==============================================================================
--- incubator/hama/trunk/core/src/main/java/org/apache/hama/checkpoint/CheckpointRunner.java (original)
+++ incubator/hama/trunk/core/src/main/java/org/apache/hama/checkpoint/CheckpointRunner.java Thu Oct 20 16:58:17 2011
@@ -17,30 +17,29 @@
*/
package org.apache.hama.checkpoint;
+import static java.util.concurrent.TimeUnit.SECONDS;
+
import java.io.BufferedReader;
import java.io.File;
+import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
-import java.io.IOException;
-
import java.util.ArrayList;
import java.util.Arrays;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicReference;
+import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
-import java.util.List;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hama.bsp.GroomServer.CheckpointerChild;
-import org.apache.hama.bsp.GroomServer;
-import static java.util.concurrent.TimeUnit.*;
-public final class CheckpointRunner implements Callable {
+public final class CheckpointRunner implements Callable<Object> {
public static final Log LOG = LogFactory.getLog(CheckpointRunner.class);
public static final String DEFAULT_PORT = "1590";
Modified: incubator/hama/trunk/core/src/main/java/org/apache/hama/checkpoint/Checkpointer.java
URL: http://svn.apache.org/viewvc/incubator/hama/trunk/core/src/main/java/org/apache/hama/checkpoint/Checkpointer.java?rev=1186921&r1=1186920&r2=1186921&view=diff
==============================================================================
--- incubator/hama/trunk/core/src/main/java/org/apache/hama/checkpoint/Checkpointer.java (original)
+++ incubator/hama/trunk/core/src/main/java/org/apache/hama/checkpoint/Checkpointer.java Thu Oct 20 16:58:17 2011
@@ -17,40 +17,36 @@
*/
package org.apache.hama.checkpoint;
-import java.io.BufferedReader;
+import static java.util.concurrent.TimeUnit.SECONDS;
+
import java.io.DataInput;
import java.io.DataInputStream;
import java.io.EOFException;
-import java.io.InputStreamReader;
import java.io.IOException;
import java.net.ServerSocket;
import java.net.Socket;
-import java.util.Arrays;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
-import java.util.concurrent.Executors;
import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
-import static java.util.concurrent.TimeUnit.*;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
-import org.apache.hama.bsp.BSPMessageBundle;
-import org.apache.hama.bsp.BSPPeerImpl.BSPSerializableMessage;
-import org.apache.hama.GroomServerRunner;
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
+import org.apache.hama.bsp.BSPSerializableMessage;
/**
* This class is responsible for checkpointing messages to hdfs.
*/
-public final class Checkpointer implements Callable {
+public final class Checkpointer implements Callable<Object> {
public static Log LOG = LogFactory.getLog(Checkpointer.class);
@@ -59,13 +55,13 @@ public final class Checkpointer implemen
private final FileSystem dfs;
private final AtomicBoolean ckptState = new AtomicBoolean(false);
private final BSPMessageDeserializer messageDeserializer;
- private final AtomicReference<ScheduledFuture> future =
- new AtomicReference<ScheduledFuture>();
+ private final AtomicReference<ScheduledFuture<Object>> future =
+ new AtomicReference<ScheduledFuture<Object>>();
/**
* Reading from socket inputstream as DataInput.
*/
- public static final class BSPMessageDeserializer implements Callable {
+ public static final class BSPMessageDeserializer implements Callable<Object> {
final BlockingQueue<BSPSerializableMessage> messageQueue =
new LinkedBlockingQueue<BSPSerializableMessage>();
final ScheduledExecutorService sched;
@@ -121,7 +117,7 @@ public final class Checkpointer implemen
while(state()) {
Socket connection = server.accept();
final DataInput in = new DataInputStream(connection.getInputStream());
- this.workers.schedule(new Callable() {
+ this.workers.schedule(new Callable<Object>() {
public Object call() throws Exception {
BSPSerializableMessage tmp = new BSPSerializableMessage();
tmp.readFields(in);
Added: incubator/hama/trunk/core/src/main/java/org/apache/hama/util/BSPNetUtils.java
URL: http://svn.apache.org/viewvc/incubator/hama/trunk/core/src/main/java/org/apache/hama/util/BSPNetUtils.java?rev=1186921&view=auto
==============================================================================
--- incubator/hama/trunk/core/src/main/java/org/apache/hama/util/BSPNetUtils.java (added)
+++ incubator/hama/trunk/core/src/main/java/org/apache/hama/util/BSPNetUtils.java Thu Oct 20 16:58:17 2011
@@ -0,0 +1,62 @@
+/**
+ * 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.hama.util;
+
+import java.net.InetAddress;
+import java.net.UnknownHostException;
+
+import org.apache.mina.util.AvailablePortFinder;
+
+/**
+ * NetUtils for our needs.
+ */
+public class BSPNetUtils {
+
+ /**
+ * Gets the canonical hostname of this machine.
+ *
+ * @return
+ * @throws UnknownHostException
+ */
+ public static String getCanonicalHostname() throws UnknownHostException {
+ return InetAddress.getLocalHost().getCanonicalHostName();
+ }
+
+ /**
+ * Uses Minas AvailablePortFinder to find a port, starting at 14000.
+ *
+ * @return a free port.
+ */
+ public static int getFreePort() {
+ int startPort = 14000;
+ return getFreePort(startPort);
+ }
+
+ /**
+ * Uses Minas AvailablePortFinder to find a port, starting at startPort.
+ *
+ * @return a free port.
+ */
+ public static int getFreePort(int startPort) {
+ while (!AvailablePortFinder.available(startPort)) {
+ startPort++;
+ }
+ return startPort;
+ }
+
+}
Propchange: incubator/hama/trunk/core/src/main/java/org/apache/hama/util/BSPNetUtils.java
------------------------------------------------------------------------------
svn:eol-style = native
Added: incubator/hama/trunk/core/src/main/java/org/apache/hama/util/StringArrayWritable.java
URL: http://svn.apache.org/viewvc/incubator/hama/trunk/core/src/main/java/org/apache/hama/util/StringArrayWritable.java?rev=1186921&view=auto
==============================================================================
--- incubator/hama/trunk/core/src/main/java/org/apache/hama/util/StringArrayWritable.java (added)
+++ incubator/hama/trunk/core/src/main/java/org/apache/hama/util/StringArrayWritable.java Thu Oct 20 16:58:17 2011
@@ -0,0 +1,65 @@
+/**
+ * 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.hama.util;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+
+import org.apache.hadoop.io.Writable;
+
+/**
+ * Custom writable for string arrays, because ArrayWritable has no default
+ * constructor and is broken.
+ *
+ */
+public class StringArrayWritable implements Writable {
+
+ private String[] array;
+
+ public StringArrayWritable() {
+ super();
+ }
+
+ public StringArrayWritable(String[] array) {
+ super();
+ this.array = array;
+ }
+
+ // no defensive copy needed because this always comes from an rpc call.
+ public String[] get() {
+ return array;
+ }
+
+ @Override
+ public void write(DataOutput out) throws IOException {
+ out.writeInt(array.length);
+ for (String s : array) {
+ out.writeUTF(s);
+ }
+ }
+
+ @Override
+ public void readFields(DataInput in) throws IOException {
+ array = new String[in.readInt()];
+ for (int i = 0; i < array.length; i++) {
+ array[i] = in.readUTF();
+ }
+ }
+
+}
\ No newline at end of file
Propchange: incubator/hama/trunk/core/src/main/java/org/apache/hama/util/StringArrayWritable.java
------------------------------------------------------------------------------
svn:eol-style = native
Modified: incubator/hama/trunk/core/src/main/java/org/apache/hama/zookeeper/QuorumPeer.java
URL: http://svn.apache.org/viewvc/incubator/hama/trunk/core/src/main/java/org/apache/hama/zookeeper/QuorumPeer.java?rev=1186921&r1=1186920&r2=1186921&view=diff
==============================================================================
--- incubator/hama/trunk/core/src/main/java/org/apache/hama/zookeeper/QuorumPeer.java (original)
+++ incubator/hama/trunk/core/src/main/java/org/apache/hama/zookeeper/QuorumPeer.java Thu Oct 20 16:58:17 2011
@@ -27,8 +27,8 @@ import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.Enumeration;
import java.util.List;
-import java.util.Properties;
import java.util.Map.Entry;
+import java.util.Properties;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@@ -59,28 +59,43 @@ public class QuorumPeer implements Const
/**
* Parse ZooKeeper configuration from Hama XML config and run a QuorumPeer.
+ * This method checks if we are in localmode to prevent zookeeper from
+ * starting.
+ *
* @param baseConf Hadoop Configuration.
*/
public static void run(Configuration baseConf) {
Configuration conf = new HamaConfiguration(baseConf);
- if(BSPMaster.getAddress(conf) == null){
+ if (BSPMaster.getAddress(conf) == null) {
System.out.println(BSPMaster.localModeMessage);
LOG.info(BSPMaster.localModeMessage);
System.exit(0);
}
try {
- Properties zkProperties = makeZKProps(conf);
- writeMyID(zkProperties);
- QuorumPeerConfig zkConfig = new QuorumPeerConfig();
- zkConfig.parseProperties(zkProperties);
- runZKServer(zkConfig);
+ runZooKeeper(conf);
} catch (Exception e) {
- LOG.error("Exception during ZooKeeper startup - exiting...",e);
+ LOG.error("Exception during ZooKeeper startup - exiting...", e);
System.exit(-1);
}
}
- private static void runZKServer(QuorumPeerConfig zkConfig) throws UnknownHostException, IOException {
+ /**
+ * Parse ZooKeeper configuration from Hama XML config and run a QuorumPeer.
+ *
+ * @param baseConf Hadoop Configuration.
+ */
+ public static void runZooKeeper(Configuration conf) throws Exception {
+
+ Properties zkProperties = makeZKProps(conf);
+ writeMyID(zkProperties);
+ QuorumPeerConfig zkConfig = new QuorumPeerConfig();
+ zkConfig.parseProperties(zkProperties);
+ runZKServer(zkConfig);
+
+ }
+
+ private static void runZKServer(QuorumPeerConfig zkConfig)
+ throws UnknownHostException, IOException {
if (zkConfig.isDistributed()) {
QuorumPeerMain qp = new QuorumPeerMain();
qp.runFromConfig(zkConfig);
@@ -101,22 +116,21 @@ public class QuorumPeer implements Const
Configuration conf = new HamaConfiguration();
String myAddress = DNS.getDefaultHost(
- conf.get("hama.zookeeper.dns.interface","default"),
- conf.get("hama.zookeeper.dns.nameserver","default"));
+ conf.get("hama.zookeeper.dns.interface", "default"),
+ conf.get("hama.zookeeper.dns.nameserver", "default"));
List<String> ips = new ArrayList<String>();
// Add what could be the best (configured) match
- ips.add(myAddress.contains(".") ?
- myAddress :
- StringUtils.simpleHostname(myAddress));
+ ips.add(myAddress.contains(".") ? myAddress : StringUtils
+ .simpleHostname(myAddress));
// For all nics get all hostnames and IPs
Enumeration<?> nics = NetworkInterface.getNetworkInterfaces();
- while(nics.hasMoreElements()) {
- Enumeration<?> rawAdrs =
- ((NetworkInterface)nics.nextElement()).getInetAddresses();
- while(rawAdrs.hasMoreElements()) {
+ while (nics.hasMoreElements()) {
+ Enumeration<?> rawAdrs = ((NetworkInterface) nics.nextElement())
+ .getInetAddresses();
+ while (rawAdrs.hasMoreElements()) {
InetAddress inet = (InetAddress) rawAdrs.nextElement();
ips.add(StringUtils.simpleHostname(inet.getHostName()));
ips.add(inet.getHostAddress());
@@ -126,7 +140,7 @@ public class QuorumPeer implements Const
for (Entry<Object, Object> entry : properties.entrySet()) {
String key = entry.getKey().toString().trim();
String value = entry.getValue().toString().trim();
-
+
if (key.startsWith("server.")) {
int dot = key.indexOf('.');
long id = Long.parseLong(key.substring(dot + 1));
@@ -140,8 +154,8 @@ public class QuorumPeer implements Const
}
if (myId == -1) {
- throw new IOException("Could not find my address: " + myAddress +
- " in list of ZooKeeper quorum servers");
+ throw new IOException("Could not find my address: " + myAddress
+ + " in list of ZooKeeper quorum servers");
}
String dataDirStr = properties.get("dataDir").toString().trim();
@@ -159,10 +173,11 @@ public class QuorumPeer implements Const
}
/**
- * Make a Properties object holding ZooKeeper config equivalent to zoo.cfg.
- * If there is a zoo.cfg in the classpath, simply read it in. Otherwise parse
- * the corresponding config options from the Hama XML configs and generate
- * the appropriate ZooKeeper properties.
+ * Make a Properties object holding ZooKeeper config equivalent to zoo.cfg. If
+ * there is a zoo.cfg in the classpath, simply read it in. Otherwise parse the
+ * corresponding config options from the Hama XML configs and generate the
+ * appropriate ZooKeeper properties.
+ *
* @param conf Configuration to read from.
* @return Properties holding mappings representing ZooKeeper zoo.cfg file.
*/
@@ -175,8 +190,8 @@ public class QuorumPeer implements Const
try {
return parseZooCfg(conf, inputStream);
} catch (IOException e) {
- LOG.warn("Cannot read " + ZOOKEEPER_CONFIG_NAME +
- ", loading from XML files", e);
+ LOG.warn("Cannot read " + ZOOKEEPER_CONFIG_NAME
+ + ", loading from XML files", e);
}
}
@@ -186,7 +201,7 @@ public class QuorumPeer implements Const
// Set the max session timeout from the provided client-side timeout
zkProperties.setProperty("maxSessionTimeout",
conf.get(Constants.ZOOKEEPER_SESSION_TIMEOUT, "1200000"));
-
+
// Directly map all of the hama.zookeeper.property.KEY properties.
for (Entry<String, String> entry : conf) {
String key = entry.getKey();
@@ -222,8 +237,9 @@ public class QuorumPeer implements Const
}
/**
- * Parse ZooKeeper's zoo.cfg, injecting Hama Configuration variables in.
- * This method is used for testing so we can pass our own InputStream.
+ * Parse ZooKeeper's zoo.cfg, injecting Hama Configuration variables in. This
+ * method is used for testing so we can pass our own InputStream.
+ *
* @param conf Configuration to use for injecting variables.
* @param inputStream InputStream to read from.
* @return Properties parsed from config stream with variables substituted.
@@ -252,7 +268,8 @@ public class QuorumPeer implements Const
LOG.fatal(msg);
throw new IOException(msg);
}
- String variable = value.substring(varStart + VARIABLE_START_LENGTH, varEnd);
+ String variable = value.substring(varStart + VARIABLE_START_LENGTH,
+ varEnd);
String substituteValue = System.getProperty(variable);
if (substituteValue == null) {
@@ -260,7 +277,7 @@ public class QuorumPeer implements Const
}
if (substituteValue == null) {
String msg = "variable " + variable + " not set in system property "
- + "or hama configs";
+ + "or hama configs";
LOG.fatal(msg);
throw new IOException(msg);
}
@@ -272,11 +289,11 @@ public class QuorumPeer implements Const
}
// Special case for 'hama.cluster.distributed' property being 'true'
if (key.startsWith("server.")) {
- if (conf.get(CLUSTER_DISTRIBUTED).equals(CLUSTER_IS_DISTRIBUTED) &&
- value.startsWith("localhost")) {
- String msg = "The server in zoo.cfg cannot be set to localhost " +
- "in a fully-distributed setup because it won't be reachable. " +
- "See \"Getting Started\" for more information.";
+ if (conf.get(CLUSTER_DISTRIBUTED).equals(CLUSTER_IS_DISTRIBUTED)
+ && value.startsWith("localhost")) {
+ String msg = "The server in zoo.cfg cannot be set to localhost "
+ + "in a fully-distributed setup because it won't be reachable. "
+ + "See \"Getting Started\" for more information.";
LOG.fatal(msg);
throw new IOException(msg);
}
@@ -286,10 +303,11 @@ public class QuorumPeer implements Const
}
return properties;
}
-
+
/**
* Return the ZK Quorum servers string given zk properties returned by
* makeZKProps
+ *
* @param properties
* @return Quorum servers String
*/
@@ -300,17 +318,16 @@ public class QuorumPeer implements Const
// The clientPort option may come after the server.X hosts, so we need to
// grab everything and then create the final host:port comma separated list.
boolean anyValid = false;
- for (Entry<Object,Object> property : properties.entrySet()) {
+ for (Entry<Object, Object> property : properties.entrySet()) {
String key = property.getKey().toString().trim();
String value = property.getValue().toString().trim();
if (key.equals("clientPort")) {
clientPort = value;
- }
- else if (key.startsWith("server.")) {
+ } else if (key.startsWith("server.")) {
String host = value.substring(0, value.indexOf(':'));
servers.add(host);
try {
- //noinspection ResultOfMethodCallIgnored
+ // noinspection ResultOfMethodCallIgnored
InetAddress.getByName(host);
anyValid = true;
} catch (UnknownHostException e) {
@@ -320,7 +337,8 @@ public class QuorumPeer implements Const
}
if (!anyValid) {
- LOG.error("no valid quorum servers found in " + Constants.ZOOKEEPER_CONFIG_NAME);
+ LOG.error("no valid quorum servers found in "
+ + Constants.ZOOKEEPER_CONFIG_NAME);
return null;
}
@@ -330,8 +348,8 @@ public class QuorumPeer implements Const
}
if (servers.isEmpty()) {
- LOG.fatal("No server.X lines found in conf/zoo.cfg. Hama must have a " +
- "ZooKeeper cluster configured for its operation.");
+ LOG.fatal("No server.X lines found in conf/zoo.cfg. Hama must have a "
+ + "ZooKeeper cluster configured for its operation.");
return null;
}
@@ -348,9 +366,10 @@ public class QuorumPeer implements Const
return hostPortBuilder.toString();
}
-
+
/**
* Return the ZK Quorum servers string given the specified configuration.
+ *
* @param conf
* @return Quorum servers
*/
Modified: incubator/hama/trunk/core/src/test/java/org/apache/hama/HamaTestCase.java
URL: http://svn.apache.org/viewvc/incubator/hama/trunk/core/src/test/java/org/apache/hama/HamaTestCase.java?rev=1186921&r1=1186920&r2=1186921&view=diff
==============================================================================
--- incubator/hama/trunk/core/src/test/java/org/apache/hama/HamaTestCase.java (original)
+++ incubator/hama/trunk/core/src/test/java/org/apache/hama/HamaTestCase.java Thu Oct 20 16:58:17 2011
@@ -62,6 +62,8 @@ public abstract class HamaTestCase exten
private void init() {
conf = new HamaConfiguration();
+ System.setProperty("hama.log.dir", "/tmp/hama-test/logs/");
+ conf.set("hama.log.dir", "/tmp/hama-test/logs/");
conf.setStrings("bsp.local.dir", "/tmp/hama-test");
conf.set("bsp.master.address", "localhost");
conf.set("bsp.groom.report.address", "127.0.0.1:0");
Modified: incubator/hama/trunk/core/src/test/java/org/apache/hama/bsp/BSPSerializerWrapper.java
URL: http://svn.apache.org/viewvc/incubator/hama/trunk/core/src/test/java/org/apache/hama/bsp/BSPSerializerWrapper.java?rev=1186921&r1=1186920&r2=1186921&view=diff
==============================================================================
--- incubator/hama/trunk/core/src/test/java/org/apache/hama/bsp/BSPSerializerWrapper.java (original)
+++ incubator/hama/trunk/core/src/test/java/org/apache/hama/bsp/BSPSerializerWrapper.java Thu Oct 20 16:58:17 2011
@@ -20,18 +20,17 @@ package org.apache.hama.bsp;
import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
-import org.apache.hama.bsp.BSPPeerImpl.BSPSerializableMessage;
public final class BSPSerializerWrapper {
- private final BSPPeerImpl.BSPMessageSerializer serializer;
+ private final BSPMessageSerializer serializer;
public BSPSerializerWrapper(Configuration conf, int port) throws IOException {
- this.serializer = new BSPPeerImpl(conf, null, null).new BSPMessageSerializer(
- conf.getInt("bsp.checkpoint.port", port));
- }
+ this.serializer = new BSPMessageSerializer(conf, conf.getInt(
+ "bsp.checkpoint.port", port));
+ }
- public final void serialize(final BSPSerializableMessage tmp)
+ public final void serialize(final BSPSerializableMessage tmp)
throws IOException {
this.serializer.serialize(tmp);
}
Modified: incubator/hama/trunk/core/src/test/java/org/apache/hama/bsp/TestBSPMasterGroomServer.java
URL: http://svn.apache.org/viewvc/incubator/hama/trunk/core/src/test/java/org/apache/hama/bsp/TestBSPMasterGroomServer.java?rev=1186921&r1=1186920&r2=1186921&view=diff
==============================================================================
--- incubator/hama/trunk/core/src/test/java/org/apache/hama/bsp/TestBSPMasterGroomServer.java (original)
+++ incubator/hama/trunk/core/src/test/java/org/apache/hama/bsp/TestBSPMasterGroomServer.java Thu Oct 20 16:58:17 2011
@@ -21,6 +21,7 @@ package org.apache.hama.bsp;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
@@ -42,10 +43,11 @@ public class TestBSPMasterGroomServer ex
configuration.set("bsp.master.address", "localhost");
assertEquals("Make sure master addr is set to localhost:", "localhost",
configuration.get("bsp.master.address"));
- configuration.setStrings("bsp.local.dir", "/tmp/hama-test");
- System.setProperty("hama.log.dir", "/tmp/hama-test/logs");
+ configuration.set("bsp.local.dir", "/tmp/hama-test");
configuration.set(Constants.ZOOKEEPER_QUORUM, "localhost");
configuration.setInt(Constants.ZOOKEEPER_CLIENT_PORT, 21810);
+ configuration.set("hama.sync.client.class",
+ "org.apache.hama.bsp.sync.zookeeper.ZooKeeperSyncClientImpl");
}
public void setUp() throws Exception {
@@ -68,14 +70,14 @@ public class TestBSPMasterGroomServer ex
FileSystem fileSys = FileSystem.get(conf);
if (bsp.waitForCompletion(true)) {
- checkOutput(fileSys, cluster, conf);
+ checkOutput(fileSys, conf, 2);
}
LOG.info("Client finishes execution job.");
}
- private static void checkOutput(FileSystem fileSys, ClusterStatus cluster,
- HamaConfiguration conf) throws Exception {
- for (int i = 0; i < 2; i++) {
+ public static void checkOutput(FileSystem fileSys, Configuration conf,
+ int tasks) throws Exception {
+ for (int i = 0; i < tasks; i++) {
SequenceFile.Reader reader = new SequenceFile.Reader(fileSys, new Path(
TMP_OUTPUT + i), conf);
LongWritable timestamp = new LongWritable();
Added: incubator/hama/trunk/core/src/test/java/org/apache/hama/bsp/TestLocalRunner.java
URL: http://svn.apache.org/viewvc/incubator/hama/trunk/core/src/test/java/org/apache/hama/bsp/TestLocalRunner.java?rev=1186921&view=auto
==============================================================================
--- incubator/hama/trunk/core/src/test/java/org/apache/hama/bsp/TestLocalRunner.java (added)
+++ incubator/hama/trunk/core/src/test/java/org/apache/hama/bsp/TestLocalRunner.java Thu Oct 20 16:58:17 2011
@@ -0,0 +1,41 @@
+/**
+ * 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.hama.bsp;
+
+import junit.framework.TestCase;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hama.HamaConfiguration;
+
+public class TestLocalRunner extends TestCase {
+
+ public void testSubmitJob() throws Exception {
+ Configuration configuration = new Configuration();
+ configuration.set("bsp.local.dir", "/tmp/hama-test");
+ BSPJob bsp = new BSPJob(new HamaConfiguration(configuration));
+ bsp.setJobName("Test Serialize Printing");
+ bsp.setBspClass(testjar.ClassSerializePrinting.HelloBSP.class);
+
+ FileSystem fileSys = FileSystem.get(configuration);
+
+ assertTrue(bsp.waitForCompletion(true));
+ TestBSPMasterGroomServer.checkOutput(fileSys, configuration, 20);
+ }
+
+}
Propchange: incubator/hama/trunk/core/src/test/java/org/apache/hama/bsp/TestLocalRunner.java
------------------------------------------------------------------------------
svn:eol-style = native
Added: incubator/hama/trunk/core/src/test/java/org/apache/hama/bsp/sync/TestSyncServiceFactory.java
URL: http://svn.apache.org/viewvc/incubator/hama/trunk/core/src/test/java/org/apache/hama/bsp/sync/TestSyncServiceFactory.java?rev=1186921&view=auto
==============================================================================
--- incubator/hama/trunk/core/src/test/java/org/apache/hama/bsp/sync/TestSyncServiceFactory.java (added)
+++ incubator/hama/trunk/core/src/test/java/org/apache/hama/bsp/sync/TestSyncServiceFactory.java Thu Oct 20 16:58:17 2011
@@ -0,0 +1,58 @@
+/**
+ * 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.hama.bsp.sync;
+
+import junit.framework.TestCase;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hama.bsp.sync.rpc.RPCSyncClientImpl;
+import org.apache.hama.bsp.sync.rpc.RPCSyncServerImpl;
+import org.apache.hama.bsp.sync.zookeeper.ZooKeeperSyncClientImpl;
+import org.apache.hama.bsp.sync.zookeeper.ZooKeeperSyncServerImpl;
+
+public class TestSyncServiceFactory extends TestCase {
+
+ public void testClientInstantiation() throws Exception {
+
+ Configuration conf = new Configuration();
+ // given null, should return zookeeper
+ SyncClient syncClient = SyncServiceFactory.getSyncClient(conf);
+ assertTrue(syncClient instanceof ZooKeeperSyncClientImpl);
+
+ // other class
+ conf.set(SyncServiceFactory.SYNC_CLIENT_CLASS, RPCSyncClientImpl.class.getCanonicalName());
+ syncClient = SyncServiceFactory.getSyncClient(conf);
+ assertTrue(syncClient instanceof RPCSyncClientImpl);
+
+ }
+
+ public void testServerInstantiation() throws Exception {
+
+ Configuration conf = new Configuration();
+ // given null, should return zookeeper
+ SyncServer syncServer = SyncServiceFactory.getSyncServer(conf);
+ assertTrue(syncServer instanceof ZooKeeperSyncServerImpl);
+
+ // other class
+ conf.set(SyncServiceFactory.SYNC_SERVER_CLASS, RPCSyncServerImpl.class.getCanonicalName());
+ syncServer = SyncServiceFactory.getSyncServer(conf);
+ assertTrue(syncServer instanceof RPCSyncServerImpl);
+
+ }
+
+}
Propchange: incubator/hama/trunk/core/src/test/java/org/apache/hama/bsp/sync/TestSyncServiceFactory.java
------------------------------------------------------------------------------
svn:eol-style = native
Added: incubator/hama/trunk/core/src/test/java/org/apache/hama/bsp/sync/rpc/TestRPCSynchronization.java
URL: http://svn.apache.org/viewvc/incubator/hama/trunk/core/src/test/java/org/apache/hama/bsp/sync/rpc/TestRPCSynchronization.java?rev=1186921&view=auto
==============================================================================
--- incubator/hama/trunk/core/src/test/java/org/apache/hama/bsp/sync/rpc/TestRPCSynchronization.java (added)
+++ incubator/hama/trunk/core/src/test/java/org/apache/hama/bsp/sync/rpc/TestRPCSynchronization.java Thu Oct 20 16:58:17 2011
@@ -0,0 +1,61 @@
+/**
+ * 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.hama.bsp.sync.rpc;
+
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
+import junit.framework.TestCase;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hama.bsp.TaskAttemptID;
+import org.apache.hama.bsp.sync.SyncClient;
+import org.apache.hama.bsp.sync.SyncServerRunner;
+import org.apache.hama.bsp.sync.SyncServiceFactory;
+
+public class TestRPCSynchronization extends TestCase {
+
+ public void testSubmitJob() throws Exception {
+ Configuration conf = new Configuration();
+ conf.set("bsp.peers.num", "1");
+ conf.set(SyncServiceFactory.SYNC_SERVER_CLASS,
+ RPCSyncServerImpl.class.getCanonicalName());
+ SyncServerRunner syncServer = SyncServiceFactory.getSyncServerRunner(conf);
+ conf = syncServer.init(conf);
+
+ ExecutorService pool = Executors.newFixedThreadPool(1);
+ pool.submit(syncServer);
+
+ conf.set(SyncServiceFactory.SYNC_CLIENT_CLASS,
+ RPCSyncClientImpl.class.getCanonicalName());
+ SyncClient syncClient = SyncServiceFactory.getSyncClient(conf);
+ assertTrue(syncClient instanceof RPCSyncClientImpl);
+ TaskAttemptID taskId = new TaskAttemptID("0", 0, 0, 0);
+ syncClient.init(conf, taskId.getJobID(), taskId);
+ syncClient.register(taskId.getJobID(), taskId, "localhost", 1255123);
+
+ syncClient.enterBarrier(taskId.getJobID(), taskId, 0);
+ syncClient.leaveBarrier(taskId.getJobID(), taskId, 0);
+
+ String[] allPeerNames = syncClient.getAllPeerNames(taskId);
+ assertEquals(allPeerNames.length, 1);
+ assertEquals(allPeerNames[0],"localhost:1255123");
+
+ }
+
+}
Propchange: incubator/hama/trunk/core/src/test/java/org/apache/hama/bsp/sync/rpc/TestRPCSynchronization.java
------------------------------------------------------------------------------
svn:eol-style = native
Modified: incubator/hama/trunk/core/src/test/java/org/apache/hama/checkpoint/TestCheckpoint.java
URL: http://svn.apache.org/viewvc/incubator/hama/trunk/core/src/test/java/org/apache/hama/checkpoint/TestCheckpoint.java?rev=1186921&r1=1186920&r2=1186921&view=diff
==============================================================================
--- incubator/hama/trunk/core/src/test/java/org/apache/hama/checkpoint/TestCheckpoint.java (original)
+++ incubator/hama/trunk/core/src/test/java/org/apache/hama/checkpoint/TestCheckpoint.java Thu Oct 20 16:58:17 2011
@@ -31,9 +31,9 @@ import org.apache.hadoop.fs.Path;
import org.apache.hama.HamaConfiguration;
import org.apache.hama.bsp.BSPMessage;
import org.apache.hama.bsp.BSPMessageBundle;
+import org.apache.hama.bsp.BSPSerializableMessage;
import org.apache.hama.bsp.BSPSerializerWrapper;
import org.apache.hama.bsp.DoubleMessage;
-import org.apache.hama.bsp.BSPPeerImpl.BSPSerializableMessage;
public class TestCheckpoint extends TestCase {
Modified: incubator/hama/trunk/yarn/src/main/java/org/apache/hama/bsp/BSPApplicationMaster.java
URL: http://svn.apache.org/viewvc/incubator/hama/trunk/yarn/src/main/java/org/apache/hama/bsp/BSPApplicationMaster.java?rev=1186921&r1=1186920&r2=1186921&view=diff
==============================================================================
--- incubator/hama/trunk/yarn/src/main/java/org/apache/hama/bsp/BSPApplicationMaster.java (original)
+++ incubator/hama/trunk/yarn/src/main/java/org/apache/hama/bsp/BSPApplicationMaster.java Thu Oct 20 16:58:17 2011
@@ -36,6 +36,7 @@ import org.apache.hadoop.ipc.ProtocolSig
import org.apache.hadoop.ipc.RPC;
import org.apache.hadoop.ipc.RPC.Server;
import org.apache.hadoop.net.NetUtils;
+import org.apache.hama.util.BSPNetUtils;
import org.apache.hadoop.yarn.Clock;
import org.apache.hadoop.yarn.SystemClock;
import org.apache.hadoop.yarn.api.AMRMProtocol;
@@ -54,7 +55,6 @@ import org.apache.hama.HamaConfiguration
import org.apache.hama.bsp.Job.JobState;
import org.apache.hama.bsp.sync.SyncServer;
import org.apache.hama.bsp.sync.SyncServerImpl;
-import org.apache.mina.util.AvailablePortFinder;
/**
* BSPApplicationMaster is an application master for Apache Hamas BSP Engine.
@@ -112,9 +112,9 @@ public class BSPApplicationMaster implem
jobId = new BSPJobID(appAttemptId.toString(), 0);
// TODO this is not localhost, is it?
- hostname = InetAddress.getLocalHost().getCanonicalHostName();
+ hostname = BSPNetUtils.getCanonicalHostname();
startSyncServer();
- clientPort = getFreePort();
+ clientPort = BSPNetUtils.getFreePort();
// TODO should have a configurable amount of RPC handlers
this.clientServer = RPC.getServer(this, hostname, clientPort, 10, false,
jobConf);
@@ -137,7 +137,7 @@ public class BSPApplicationMaster implem
* @throws IOException
*/
private void startSyncServer() throws IOException {
- int syncPort = getFreePort(15000);
+ int syncPort = BSPNetUtils.getFreePort(15000);
syncServer = new SyncServerImpl(jobConf.getInt("bsp.peers.num", 1),
hostname, syncPort);
syncServerFuture = threadPool.submit(syncServer);
@@ -298,29 +298,6 @@ public class BSPApplicationMaster implem
LOG.info("Written new configuration back to " + path);
}
- /**
- * Uses Minas AvailablePortFinder to find a port, starting at 14000.
- *
- * @return a free port.
- */
- private int getFreePort() {
- int startPort = 14000;
- return getFreePort(startPort);
- }
-
- /**
- * Uses Minas AvailablePortFinder to find a port, starting at startPort.
- *
- * @return a free port.
- */
- private int getFreePort(int startPort) {
- while (!AvailablePortFinder.available(startPort)) {
- startPort++;
- LOG.debug("Testing port for availability: " + startPort);
- }
- return startPort;
- }
-
@Override
public long getProtocolVersion(String arg0, long arg1) throws IOException {
return BSPClient.VERSION;
Modified: incubator/hama/trunk/yarn/src/main/java/org/apache/hama/bsp/YARNBSPPeerImpl.java
URL: http://svn.apache.org/viewvc/incubator/hama/trunk/yarn/src/main/java/org/apache/hama/bsp/YARNBSPPeerImpl.java?rev=1186921&r1=1186920&r2=1186921&view=diff
==============================================================================
--- incubator/hama/trunk/yarn/src/main/java/org/apache/hama/bsp/YARNBSPPeerImpl.java (original)
+++ incubator/hama/trunk/yarn/src/main/java/org/apache/hama/bsp/YARNBSPPeerImpl.java Thu Oct 20 16:58:17 2011
@@ -275,7 +275,7 @@ public class YARNBSPPeerImpl implements
* @see org.apache.hama.bsp.BSPPeerInterface#sync()
*/
@Override
- public void sync() throws IOException, InterruptedException {
+ public void sync() throws InterruptedException {
enterBarrier();
Iterator<Entry<InetSocketAddress, ConcurrentLinkedQueue<BSPMessage>>> it = this.outgoingQueues
.entrySet().iterator();
@@ -299,13 +299,6 @@ public class YARNBSPPeerImpl implements
bundle.addMessage(message);
}
- // checkpointing
- if (null != this.messageSerializer) {
- this.messageSerializer.serialize(new BSPSerializableMessage(
- checkpointedPath(), bundle));
- }
-
- peer.put(bundle);
}
leaveBarrier();
@@ -349,7 +342,6 @@ public class YARNBSPPeerImpl implements
this.outgoingQueues.clear();
}
- @Override
public void close() throws IOException {
this.clear();
syncService.deregisterFromBarrier(taskid,
@@ -485,4 +477,16 @@ public class YARNBSPPeerImpl implements
// TODO Auto-generated method stub
return new ProtocolSignature();
}
+
+ @Override
+ public String getPeerName(int index) {
+ // TODO Auto-generated method stub
+ return null;
+ }
+
+ @Override
+ public int getNumPeers() {
+ // TODO Auto-generated method stub
+ return 0;
+ }
}
Modified: incubator/hama/trunk/yarn/src/main/java/org/apache/hama/bsp/sync/StringArrayWritable.java
URL: http://svn.apache.org/viewvc/incubator/hama/trunk/yarn/src/main/java/org/apache/hama/bsp/sync/StringArrayWritable.java?rev=1186921&r1=1186920&r2=1186921&view=diff
==============================================================================
--- incubator/hama/trunk/yarn/src/main/java/org/apache/hama/bsp/sync/StringArrayWritable.java (original)
+++ incubator/hama/trunk/yarn/src/main/java/org/apache/hama/bsp/sync/StringArrayWritable.java Thu Oct 20 16:58:17 2011
@@ -1,65 +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.hama.bsp.sync;
-
-import java.io.DataInput;
-import java.io.DataOutput;
-import java.io.IOException;
-
-import org.apache.hadoop.io.Writable;
-
-/**
- * Custom writable for string arrays, because ArrayWritable has no default
- * constructor and is broken.
- *
- */
-public class StringArrayWritable implements Writable {
-
- private String[] array;
-
- public StringArrayWritable() {
- super();
- }
-
- public StringArrayWritable(String[] array) {
- super();
- this.array = array;
- }
-
- // no defensive copy needed because this always comes from an rpc call.
- public String[] get() {
- return array;
- }
-
- @Override
- public void write(DataOutput out) throws IOException {
- out.writeInt(array.length);
- for (String s : array) {
- out.writeUTF(s);
- }
- }
-
- @Override
- public void readFields(DataInput in) throws IOException {
- array = new String[in.readInt()];
- for (int i = 0; i < array.length; i++) {
- array[i] = in.readUTF();
- }
- }
-
-}
Modified: incubator/hama/trunk/yarn/src/main/java/org/apache/hama/bsp/sync/SyncServer.java
URL: http://svn.apache.org/viewvc/incubator/hama/trunk/yarn/src/main/java/org/apache/hama/bsp/sync/SyncServer.java?rev=1186921&r1=1186920&r2=1186921&view=diff
==============================================================================
--- incubator/hama/trunk/yarn/src/main/java/org/apache/hama/bsp/sync/SyncServer.java (original)
+++ incubator/hama/trunk/yarn/src/main/java/org/apache/hama/bsp/sync/SyncServer.java Thu Oct 20 16:58:17 2011
@@ -21,6 +21,7 @@ import org.apache.hadoop.io.LongWritable
import org.apache.hadoop.io.Text;
import org.apache.hadoop.ipc.VersionedProtocol;
import org.apache.hama.bsp.TaskAttemptID;
+import org.apache.hama.util.StringArrayWritable;
/**
* Hadoop RPC based barrier synchronization service.
Modified: incubator/hama/trunk/yarn/src/main/java/org/apache/hama/bsp/sync/SyncServerImpl.java
URL: http://svn.apache.org/viewvc/incubator/hama/trunk/yarn/src/main/java/org/apache/hama/bsp/sync/SyncServerImpl.java?rev=1186921&r1=1186920&r2=1186921&view=diff
==============================================================================
--- incubator/hama/trunk/yarn/src/main/java/org/apache/hama/bsp/sync/SyncServerImpl.java (original)
+++ incubator/hama/trunk/yarn/src/main/java/org/apache/hama/bsp/sync/SyncServerImpl.java Thu Oct 20 16:58:17 2011
@@ -37,6 +37,7 @@ import org.apache.hadoop.ipc.ProtocolSig
import org.apache.hadoop.ipc.RPC;
import org.apache.hadoop.ipc.RPC.Server;
import org.apache.hama.bsp.TaskAttemptID;
+import org.apache.hama.util.StringArrayWritable;
/**
* Synchronization Deamon. <br\>