You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@giraph.apache.org by ma...@apache.org on 2012/10/26 03:00:55 UTC
svn commit: r1402363 [1/2] - in /giraph/trunk: ./
giraph/src/main/java/org/apache/giraph/benchmark/
giraph/src/main/java/org/apache/giraph/bsp/
giraph/src/main/java/org/apache/giraph/comm/
giraph/src/main/java/org/apache/giraph/comm/aggregators/ giraph...
Author: maja
Date: Fri Oct 26 01:00:54 2012
New Revision: 1402363
URL: http://svn.apache.org/viewvc?rev=1402363&view=rev
Log:
GIRAPH-273: Aggregators shouldn't use Zookeeper
Added:
giraph/trunk/giraph/src/main/java/org/apache/giraph/benchmark/AggregatorsBenchmark.java
giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/aggregators/
giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/aggregators/AggregatedValueOutputStream.java
giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/aggregators/AggregatorOutputStream.java
giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/aggregators/AggregatorUtils.java
giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/aggregators/AllAggregatorServerData.java
giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/aggregators/CountingCache.java
giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/aggregators/CountingOutputStream.java
giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/aggregators/OwnerAggregatorServerData.java
giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/aggregators/SendAggregatedValueCache.java
giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/aggregators/SendAggregatorCache.java
giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/aggregators/WorkerAggregatorRequestProcessor.java
giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/aggregators/package-info.java
giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/netty/NettyWorkerAggregatorRequestProcessor.java
giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/requests/ByteArrayRequest.java
giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/requests/SendAggregatorsToMasterRequest.java
giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/requests/SendAggregatorsToOwnerRequest.java
giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/requests/SendAggregatorsToWorkerRequest.java
giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/requests/SendWorkerAggregatorsRequest.java
giraph/trunk/giraph/src/main/java/org/apache/giraph/utils/ExpectedBarrier.java
Removed:
giraph/trunk/giraph/src/main/java/org/apache/giraph/graph/AggregatorHandler.java
Modified:
giraph/trunk/CHANGELOG
giraph/trunk/giraph/src/main/java/org/apache/giraph/bsp/CentralizedServiceMaster.java
giraph/trunk/giraph/src/main/java/org/apache/giraph/bsp/CentralizedServiceWorker.java
giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/MasterClient.java
giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/ServerData.java
giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/netty/NettyMasterClient.java
giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/netty/NettyMasterClientServer.java
giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/netty/NettyMasterServer.java
giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/netty/handler/MasterRequestServerHandler.java
giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/requests/MasterRequest.java
giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/requests/RequestType.java
giraph/trunk/giraph/src/main/java/org/apache/giraph/examples/AggregatorsTestVertex.java
giraph/trunk/giraph/src/main/java/org/apache/giraph/graph/BspService.java
giraph/trunk/giraph/src/main/java/org/apache/giraph/graph/BspServiceMaster.java
giraph/trunk/giraph/src/main/java/org/apache/giraph/graph/BspServiceWorker.java
giraph/trunk/giraph/src/main/java/org/apache/giraph/graph/GraphMapper.java
giraph/trunk/giraph/src/main/java/org/apache/giraph/graph/MasterAggregatorHandler.java
giraph/trunk/giraph/src/main/java/org/apache/giraph/graph/WorkerAggregatorHandler.java
giraph/trunk/giraph/src/test/java/org/apache/giraph/graph/TestAggregatorsHandling.java
Modified: giraph/trunk/CHANGELOG
URL: http://svn.apache.org/viewvc/giraph/trunk/CHANGELOG?rev=1402363&r1=1402362&r2=1402363&view=diff
==============================================================================
--- giraph/trunk/CHANGELOG (original)
+++ giraph/trunk/CHANGELOG Fri Oct 26 01:00:54 2012
@@ -1,6 +1,8 @@
Giraph Change Log
Release 0.2.0 - unreleased
+ GIRAPH-273: Aggregators shouldn't use Zookeeper (majakabiljo)
+
GIRAPH-376: Facebook Hadoop profile broken. (nitay via aching)
GIRAPH-387: GiraphRunner's better handling of configuration property
Added: giraph/trunk/giraph/src/main/java/org/apache/giraph/benchmark/AggregatorsBenchmark.java
URL: http://svn.apache.org/viewvc/giraph/trunk/giraph/src/main/java/org/apache/giraph/benchmark/AggregatorsBenchmark.java?rev=1402363&view=auto
==============================================================================
--- giraph/trunk/giraph/src/main/java/org/apache/giraph/benchmark/AggregatorsBenchmark.java (added)
+++ giraph/trunk/giraph/src/main/java/org/apache/giraph/benchmark/AggregatorsBenchmark.java Fri Oct 26 01:00:54 2012
@@ -0,0 +1,289 @@
+/*
+ * 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.giraph.benchmark;
+
+import org.apache.commons.cli.CommandLine;
+import org.apache.commons.cli.CommandLineParser;
+import org.apache.commons.cli.HelpFormatter;
+import org.apache.commons.cli.Options;
+import org.apache.commons.cli.PosixParser;
+import org.apache.giraph.aggregators.LongSumAggregator;
+import org.apache.giraph.graph.DefaultMasterCompute;
+import org.apache.giraph.graph.DefaultWorkerContext;
+import org.apache.giraph.graph.EdgeListVertex;
+import org.apache.giraph.graph.GiraphJob;
+import org.apache.giraph.io.PseudoRandomVertexInputFormat;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.io.DoubleWritable;
+import org.apache.hadoop.io.LongWritable;
+import org.apache.hadoop.util.Tool;
+import org.apache.hadoop.util.ToolRunner;
+import org.apache.log4j.Logger;
+
+import java.io.IOException;
+
+/**
+ * Benchmark for aggregators. Also checks the correctness.
+ */
+public class AggregatorsBenchmark implements Tool {
+ /** Class logger */
+ private static final Logger LOG =
+ Logger.getLogger(AggregatorsBenchmark.class);
+ /** Number of aggregators setting */
+ private static final String AGGREGATORS_NUM = "aggregatorsbenchmark.num";
+ /** Configuration */
+ private Configuration conf;
+
+ /**
+ * Vertex class for AggregatorsBenchmark
+ */
+ public static class AggregatorsBenchmarkVertex extends
+ EdgeListVertex<LongWritable, DoubleWritable, DoubleWritable,
+ DoubleWritable> {
+ @Override
+ public void compute(Iterable<DoubleWritable> messages) throws IOException {
+ int n = getNumAggregators(getConf());
+ long superstep = getSuperstep();
+ int w = getWorkerContextAggregated(getConf(), superstep);
+ for (int i = 0; i < n; i++) {
+ aggregate("w" + i, new LongWritable((superstep + 1) * i));
+ aggregate("p" + i, new LongWritable(i));
+
+ assertEquals(superstep * (getTotalNumVertices() * i) + w,
+ ((LongWritable) getAggregatedValue("w" + i)).get());
+ assertEquals(-(superstep * i),
+ ((LongWritable) getAggregatedValue("m" + i)).get());
+ assertEquals(superstep * getTotalNumVertices() * i,
+ ((LongWritable) getAggregatedValue("p" + i)).get());
+ }
+ if (superstep > 2) {
+ voteToHalt();
+ }
+ }
+ }
+
+ @Override
+ public Configuration getConf() {
+ return conf;
+ }
+
+ @Override
+ public void setConf(Configuration conf) {
+ this.conf = conf;
+ }
+
+ /**
+ * MasterCompute class for AggregatorsBenchmark
+ */
+ public static class AggregatorsBenchmarkMasterCompute extends
+ DefaultMasterCompute {
+ @Override
+ public void initialize() throws InstantiationException,
+ IllegalAccessException {
+ int n = getNumAggregators(getConf());
+ for (int i = 0; i < n; i++) {
+ registerAggregator("w" + i, LongSumAggregator.class);
+ registerAggregator("m" + i, LongSumAggregator.class);
+ registerPersistentAggregator("p" + i, LongSumAggregator.class);
+ }
+ }
+
+ @Override
+ public void compute() {
+ int n = getNumAggregators(getConf());
+ long superstep = getSuperstep();
+ int w = getWorkerContextAggregated(getConf(), superstep);
+ for (int i = 0; i < n; i++) {
+ setAggregatedValue("m" + i, new LongWritable(-superstep * i));
+
+ if (superstep > 0) {
+ assertEquals(superstep * (getTotalNumVertices() * i) + w,
+ ((LongWritable) getAggregatedValue("w" + i)).get());
+ assertEquals(superstep * getTotalNumVertices() * i,
+ ((LongWritable) getAggregatedValue("p" + i)).get());
+ }
+ }
+ }
+ }
+
+ /**
+ * WorkerContext class for AggregatorsBenchmark
+ */
+ public static class AggregatorsBenchmarkWorkerContext
+ extends DefaultWorkerContext {
+ @Override
+ public void preSuperstep() {
+ addToWorkerAggregators(1);
+ checkAggregators();
+ }
+
+ @Override
+ public void postSuperstep() {
+ addToWorkerAggregators(2);
+ checkAggregators();
+ }
+
+ /**
+ * Check if aggregator values are correct for current superstep
+ */
+ private void checkAggregators() {
+ int n = getNumAggregators(getContext().getConfiguration());
+ long superstep = getSuperstep();
+ int w = getWorkerContextAggregated(
+ getContext().getConfiguration(), superstep);
+ for (int i = 0; i < n; i++) {
+ assertEquals(superstep * (getTotalNumVertices() * i) + w,
+ ((LongWritable) getAggregatedValue("w" + i)).get());
+ assertEquals(-(superstep * i),
+ ((LongWritable) getAggregatedValue("m" + i)).get());
+ assertEquals(superstep * getTotalNumVertices() * i,
+ ((LongWritable) getAggregatedValue("p" + i)).get());
+ }
+ }
+
+ /**
+ * Add some value to worker aggregators.
+ *
+ * @param valueToAdd Which value to add
+ */
+ private void addToWorkerAggregators(int valueToAdd) {
+ int n = getNumAggregators(getContext().getConfiguration());
+ for (int i = 0; i < n; i++) {
+ aggregate("w" + i, new LongWritable(valueToAdd));
+ }
+ }
+ }
+
+ /**
+ * Get the number of aggregators from configuration
+ *
+ * @param conf Configuration
+ * @return Number of aggregators
+ */
+ private static int getNumAggregators(Configuration conf) {
+ return conf.getInt(AGGREGATORS_NUM, 0);
+ }
+
+ /**
+ * Get the value which should be aggreagted by worker context
+ *
+ * @param conf Configuration
+ * @param superstep Superstep
+ * @return The value which should be aggregated by worker context
+ */
+ private static int getWorkerContextAggregated(Configuration conf,
+ long superstep) {
+ return (superstep <= 0) ? 0 : conf.getInt("workers", 0) * 3;
+ }
+
+ /**
+ * Check if values are equal, throw an exception if they aren't
+ *
+ * @param expected Expected value
+ * @param actual Actual value
+ */
+ private static void assertEquals(long expected, long actual) {
+ if (expected != actual) {
+ throw new RuntimeException("expected: " + expected +
+ ", actual: " + actual);
+ }
+ }
+
+ @Override
+ public final int run(final String[] args) throws Exception {
+ Options options = new Options();
+ options.addOption("h", "help", false, "Help");
+ options.addOption("v", "verbose", false, "Verbose");
+ options.addOption("w",
+ "workers",
+ true,
+ "Number of workers");
+ options.addOption("V",
+ "aggregateVertices",
+ true,
+ "Aggregate vertices");
+ options.addOption("a",
+ "aggregators",
+ true,
+ "Aggregators");
+ HelpFormatter formatter = new HelpFormatter();
+ if (args.length == 0) {
+ formatter.printHelp(getClass().getName(), options, true);
+ return 0;
+ }
+ CommandLineParser parser = new PosixParser();
+ CommandLine cmd = parser.parse(options, args);
+ if (cmd.hasOption('h')) {
+ formatter.printHelp(getClass().getName(), options, true);
+ return 0;
+ }
+ if (!cmd.hasOption('w')) {
+ LOG.info("Need to choose the number of workers (-w)");
+ return -1;
+ }
+ if (!cmd.hasOption('V')) {
+ LOG.info("Need to set the aggregate vertices (-V)");
+ return -1;
+ }
+ if (!cmd.hasOption('a')) {
+ LOG.info("Need to set number of aggregators (-a)");
+ return -1;
+ }
+
+ int workers = Integer.parseInt(cmd.getOptionValue('w'));
+ GiraphJob job = new GiraphJob(getConf(), getClass().getName());
+ job.getConfiguration().setVertexClass(AggregatorsBenchmarkVertex.class);
+ job.getConfiguration().setMasterComputeClass(
+ AggregatorsBenchmarkMasterCompute.class);
+ job.getConfiguration().setVertexInputFormatClass(
+ PseudoRandomVertexInputFormat.class);
+ job.getConfiguration().setWorkerContextClass(
+ AggregatorsBenchmarkWorkerContext.class);
+ job.getConfiguration().setWorkerConfiguration(workers, workers, 100.0f);
+ job.getConfiguration().setLong(
+ PseudoRandomVertexInputFormat.AGGREGATE_VERTICES,
+ Long.parseLong(cmd.getOptionValue('V')));
+ job.getConfiguration().setLong(
+ PseudoRandomVertexInputFormat.EDGES_PER_VERTEX,
+ 1);
+ job.getConfiguration().setInt(AGGREGATORS_NUM,
+ Integer.parseInt(cmd.getOptionValue('a')));
+ job.getConfiguration().setInt("workers", workers);
+
+ boolean isVerbose = false;
+ if (cmd.hasOption('v')) {
+ isVerbose = true;
+ }
+ if (job.run(isVerbose)) {
+ return 0;
+ } else {
+ return -1;
+ }
+ }
+
+ /**
+ * Execute the benchmark.
+ *
+ * @param args Typically the command line arguments.
+ * @throws Exception Any exception from the computation.
+ */
+ public static void main(final String[] args) throws Exception {
+ System.exit(ToolRunner.run(new AggregatorsBenchmark(), args));
+ }
+}
Modified: giraph/trunk/giraph/src/main/java/org/apache/giraph/bsp/CentralizedServiceMaster.java
URL: http://svn.apache.org/viewvc/giraph/trunk/giraph/src/main/java/org/apache/giraph/bsp/CentralizedServiceMaster.java?rev=1402363&r1=1402362&r2=1402363&view=diff
==============================================================================
--- giraph/trunk/giraph/src/main/java/org/apache/giraph/bsp/CentralizedServiceMaster.java (original)
+++ giraph/trunk/giraph/src/main/java/org/apache/giraph/bsp/CentralizedServiceMaster.java Fri Oct 26 01:00:54 2012
@@ -20,7 +20,7 @@ package org.apache.giraph.bsp;
import java.io.IOException;
-import org.apache.giraph.graph.MasterAggregatorUsage;
+import org.apache.giraph.graph.MasterAggregatorHandler;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.io.WritableComparable;
import org.apache.zookeeper.KeeperException;
@@ -99,9 +99,9 @@ public interface CentralizedServiceMaste
long desiredSuperstep);
/**
- * Get master aggregator usage
+ * Get master aggregator handler
*
- * @return Master aggregator usage
+ * @return Master aggregator handler
*/
- MasterAggregatorUsage getAggregatorUsage();
+ MasterAggregatorHandler getAggregatorHandler();
}
Modified: giraph/trunk/giraph/src/main/java/org/apache/giraph/bsp/CentralizedServiceWorker.java
URL: http://svn.apache.org/viewvc/giraph/trunk/giraph/src/main/java/org/apache/giraph/bsp/CentralizedServiceWorker.java?rev=1402363&r1=1402362&r2=1402363&view=diff
==============================================================================
--- giraph/trunk/giraph/src/main/java/org/apache/giraph/bsp/CentralizedServiceWorker.java (original)
+++ giraph/trunk/giraph/src/main/java/org/apache/giraph/bsp/CentralizedServiceWorker.java Fri Oct 26 01:00:54 2012
@@ -27,7 +27,7 @@ import org.apache.giraph.comm.WorkerClie
import org.apache.giraph.graph.FinishedSuperstepStats;
import org.apache.giraph.graph.GraphState;
import org.apache.giraph.graph.VertexEdgeCount;
-import org.apache.giraph.graph.WorkerAggregatorUsage;
+import org.apache.giraph.graph.WorkerAggregatorHandler;
import org.apache.giraph.graph.partition.PartitionStore;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.io.WritableComparable;
@@ -213,9 +213,16 @@ public interface CentralizedServiceWorke
ServerData<I, V, E, M> getServerData();
/**
- * Get worker aggregator usage
+ * Get worker aggregator handler
*
- * @return Worker aggregator usage
+ * @return Worker aggregator handler
*/
- WorkerAggregatorUsage getAggregatorUsage();
+ WorkerAggregatorHandler getAggregatorHandler();
+
+ /**
+ * Final preparation for superstep, called after startSuperstep and
+ * potential loading from checkpoint, right before the computation started
+ * TODO how to avoid this additional function
+ */
+ void prepareSuperstep();
}
Modified: giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/MasterClient.java
URL: http://svn.apache.org/viewvc/giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/MasterClient.java?rev=1402363&r1=1402362&r2=1402363&view=diff
==============================================================================
--- giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/MasterClient.java (original)
+++ giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/MasterClient.java Fri Oct 26 01:00:54 2012
@@ -18,6 +18,11 @@
package org.apache.giraph.comm;
+import org.apache.giraph.graph.Aggregator;
+import org.apache.hadoop.io.Writable;
+
+import java.io.IOException;
+
/**
* Interface for master to send messages to workers
*/
@@ -28,6 +33,23 @@ public interface MasterClient {
void openConnections();
/**
+ * Sends aggregator to its owner
+ *
+ * @param aggregatorName Name of the aggregator
+ * @param aggregatorClass Class of the aggregator
+ * @param aggregatedValue Value of the aggregator
+ * @throws IOException
+ */
+ void sendAggregator(String aggregatorName,
+ Class<? extends Aggregator> aggregatorClass,
+ Writable aggregatedValue) throws IOException;
+
+ /**
+ * Flush aggregated values cache.
+ */
+ void finishSendingAggregatedValues() throws IOException;
+
+ /**
* Flush all outgoing messages. This will synchronously ensure that all
* messages have been send and delivered prior to returning.
*/
Modified: giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/ServerData.java
URL: http://svn.apache.org/viewvc/giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/ServerData.java?rev=1402363&r1=1402362&r2=1402363&view=diff
==============================================================================
--- giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/ServerData.java (original)
+++ giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/ServerData.java Fri Oct 26 01:00:54 2012
@@ -20,6 +20,8 @@ package org.apache.giraph.comm;
import org.apache.giraph.GiraphConfiguration;
import org.apache.giraph.ImmutableClassesGiraphConfiguration;
+import org.apache.giraph.comm.aggregators.AllAggregatorServerData;
+import org.apache.giraph.comm.aggregators.OwnerAggregatorServerData;
import org.apache.giraph.comm.messages.MessageStoreByPartition;
import org.apache.giraph.comm.messages.MessageStoreFactory;
import org.apache.giraph.graph.VertexMutations;
@@ -65,6 +67,14 @@ public class ServerData<I extends Writab
*/
private final ConcurrentHashMap<I, VertexMutations<I, V, E, M>>
vertexMutations = new ConcurrentHashMap<I, VertexMutations<I, V, E, M>>();
+ /**
+ * Holds aggregtors which current worker owns from current superstep
+ */
+ private final OwnerAggregatorServerData ownerAggregatorData;
+ /**
+ * Holds old aggregators from previous superstep
+ */
+ private final AllAggregatorServerData allAggregatorData;
/**
* Constructor.
@@ -90,6 +100,8 @@ public class ServerData<I extends Writab
partitionStore =
new SimplePartitionStore<I, V, E, M>(configuration, context);
}
+ ownerAggregatorData = new OwnerAggregatorServerData(context);
+ allAggregatorData = new AllAggregatorServerData(context);
}
/**
@@ -144,4 +156,22 @@ public class ServerData<I extends Writab
getVertexMutations() {
return vertexMutations;
}
+
+ /**
+ * Get holder for aggregators which current worker owns
+ *
+ * @return Holder for aggregators which current worker owns
+ */
+ public OwnerAggregatorServerData getOwnerAggregatorData() {
+ return ownerAggregatorData;
+ }
+
+ /**
+ * Get holder for aggregators from previous superstep
+ *
+ * @return Holder for aggregators from previous superstep
+ */
+ public AllAggregatorServerData getAllAggregatorData() {
+ return allAggregatorData;
+ }
}
Added: giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/aggregators/AggregatedValueOutputStream.java
URL: http://svn.apache.org/viewvc/giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/aggregators/AggregatedValueOutputStream.java?rev=1402363&view=auto
==============================================================================
--- giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/aggregators/AggregatedValueOutputStream.java (added)
+++ giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/aggregators/AggregatedValueOutputStream.java Fri Oct 26 01:00:54 2012
@@ -0,0 +1,45 @@
+/*
+ * 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.giraph.comm.aggregators;
+
+import org.apache.hadoop.io.Writable;
+
+import java.io.IOException;
+
+/**
+ * Implementation of {@link CountingOutputStream} which allows writing of
+ * aggregator values in the form of pair (name, value)
+ */
+public class AggregatedValueOutputStream extends CountingOutputStream {
+ /**
+ * Write aggregator to the stream and increment internal counter
+ *
+ * @param aggregatorName Name of the aggregator
+ * @param aggregatedValue Value of aggregator
+ * @return Number of bytes occupied by the stream
+ * @throws IOException
+ */
+ public int addAggregator(String aggregatorName,
+ Writable aggregatedValue) throws IOException {
+ incrementCounter();
+ dataOutput.writeUTF(aggregatorName);
+ aggregatedValue.write(dataOutput);
+ return getSize();
+ }
+}
Added: giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/aggregators/AggregatorOutputStream.java
URL: http://svn.apache.org/viewvc/giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/aggregators/AggregatorOutputStream.java?rev=1402363&view=auto
==============================================================================
--- giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/aggregators/AggregatorOutputStream.java (added)
+++ giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/aggregators/AggregatorOutputStream.java Fri Oct 26 01:00:54 2012
@@ -0,0 +1,49 @@
+/*
+ * 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.giraph.comm.aggregators;
+
+import org.apache.giraph.graph.Aggregator;
+import org.apache.hadoop.io.Writable;
+
+import java.io.IOException;
+
+/**
+ * Implementation of {@link CountingOutputStream} which allows writing of
+ * aggregators in the form of triple (name, classname, value)
+ */
+public class AggregatorOutputStream extends CountingOutputStream {
+ /**
+ * Write aggregator to the stream and increment internal counter
+ *
+ * @param aggregatorName Name of the aggregator
+ * @param aggregatorClass Class of aggregator
+ * @param aggregatedValue Value of aggregator
+ * @return Number of bytes occupied by the stream
+ * @throws IOException
+ */
+ public int addAggregator(String aggregatorName,
+ Class<? extends Aggregator> aggregatorClass,
+ Writable aggregatedValue) throws IOException {
+ incrementCounter();
+ dataOutput.writeUTF(aggregatorName);
+ dataOutput.writeUTF(aggregatorClass.getName());
+ aggregatedValue.write(dataOutput);
+ return getSize();
+ }
+}
Added: giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/aggregators/AggregatorUtils.java
URL: http://svn.apache.org/viewvc/giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/aggregators/AggregatorUtils.java?rev=1402363&view=auto
==============================================================================
--- giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/aggregators/AggregatorUtils.java (added)
+++ giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/aggregators/AggregatorUtils.java Fri Oct 26 01:00:54 2012
@@ -0,0 +1,96 @@
+/*
+ * 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.giraph.comm.aggregators;
+
+import org.apache.giraph.graph.Aggregator;
+import org.apache.giraph.graph.WorkerInfo;
+import org.apache.hadoop.io.Writable;
+
+import java.util.List;
+
+/**
+ * Class for aggregator constants and utility methods
+ */
+public class AggregatorUtils {
+ /**
+ * Special aggregator name which will be used to send the total number of
+ * aggregators requests which should arrive
+ */
+ public static final String SPECIAL_COUNT_AGGREGATOR =
+ "__aggregatorRequestCount";
+ /** How big a single aggregator request can be (in bytes) */
+ public static final String MAX_BYTES_PER_AGGREGATOR_REQUEST =
+ "giraph.maxBytesPerAggregatorRequest";
+ /** Default max size of single aggregator request (1MB) */
+ public static final int MAX_BYTES_PER_AGGREGATOR_REQUEST_DEFAULT =
+ 1024 * 1024;
+
+ /** Do not instantiate */
+ private AggregatorUtils() { }
+
+ /**
+ * Get aggregator class from class name, catch all exceptions.
+ *
+ * @param aggregatorClassName Class nam of aggregator class
+ * @return Aggregator class
+ */
+ public static Class<Aggregator<Writable>> getAggregatorClass(String
+ aggregatorClassName) {
+ try {
+ return (Class<Aggregator<Writable>>) Class.forName(aggregatorClassName);
+ } catch (ClassNotFoundException e) {
+ throw new IllegalStateException("getAggregatorClass: " +
+ "ClassNotFoundException for aggregator class " + aggregatorClassName,
+ e);
+ }
+ }
+
+ /**
+ * Create new aggregator instance from aggregator class,
+ * catch all exceptions.
+ *
+ * @param aggregatorClass Class of aggregator
+ * @return New aggregator
+ */
+ public static Aggregator<Writable> newAggregatorInstance(
+ Class<Aggregator<Writable>> aggregatorClass) {
+ try {
+ return aggregatorClass.newInstance();
+ } catch (InstantiationException e) {
+ throw new IllegalStateException("createAggregator: " +
+ "InstantiationException for aggregator class " + aggregatorClass, e);
+ } catch (IllegalAccessException e) {
+ throw new IllegalStateException("createAggregator: " +
+ "IllegalAccessException for aggregator class " + aggregatorClass, e);
+ }
+ }
+
+ /**
+ * Get owner of aggregator with selected name from the list of workers
+ *
+ * @param aggregatorName Name of the aggregators
+ * @param workers List of workers
+ * @return Worker which owns the aggregator
+ */
+ public static WorkerInfo getOwner(String aggregatorName,
+ List<WorkerInfo> workers) {
+ int index = Math.abs(aggregatorName.hashCode()) % workers.size();
+ return workers.get(index);
+ }
+}
Added: giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/aggregators/AllAggregatorServerData.java
URL: http://svn.apache.org/viewvc/giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/aggregators/AllAggregatorServerData.java?rev=1402363&view=auto
==============================================================================
--- giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/aggregators/AllAggregatorServerData.java (added)
+++ giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/aggregators/AllAggregatorServerData.java Fri Oct 26 01:00:54 2012
@@ -0,0 +1,239 @@
+/*
+ * 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.giraph.comm.aggregators;
+
+import org.apache.giraph.graph.Aggregator;
+import org.apache.giraph.utils.ExpectedBarrier;
+import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.util.Progressable;
+import org.apache.log4j.Logger;
+
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentMap;
+
+/**
+ * Accepts aggregators and their values from previous superstep from master
+ * and workers which own aggregators. Keeps data received from master so it
+ * could be distributed later. Also counts the requests so we would know
+ * when we are done receiving requests.
+ *
+ * Only restriction is that we need to call registerAggregatorClass before
+ * calling createAggregatorInitialValue, other than that methods of this class
+ * are thread-safe.
+ */
+public class AllAggregatorServerData {
+ /** Class logger */
+ private static final Logger LOG =
+ Logger.getLogger(AllAggregatorServerData.class);
+ /**
+ * Map from aggregator class to aggregator object which we need in order
+ * to create initial aggregated values
+ */
+ private final
+ ConcurrentMap<Class<Aggregator<Writable>>, Aggregator<Writable>>
+ aggregatorTypesMap = Maps.newConcurrentMap();
+ /** Map of aggregator classes */
+ private final ConcurrentMap<String, Class<Aggregator<Writable>>>
+ aggregatorClassMap = Maps.newConcurrentMap();
+ /** Map of values of aggregators from previous superstep */
+ private final ConcurrentMap<String, Writable>
+ aggregatedValuesMap = Maps.newConcurrentMap();
+ /**
+ * Counts the requests with final aggregators from master.
+ * It uses values from special aggregators
+ * (named AggregatorUtils.SPECIAL_COUNT_AGGREGATOR)
+ * to know how many requests it has to receive.
+ */
+ private final ExpectedBarrier masterBarrier;
+ /**
+ * Aggregator data which this worker received from master and which it is
+ * going to distribute before starting next superstep. Thread-safe.
+ */
+ private final List<byte[]> masterData =
+ Collections.synchronizedList(Lists.<byte[]>newArrayList());
+ /**
+ * Counts the requests with final aggregators from other workers.
+ * It uses values from special aggregators
+ * (named AggregatorUtils.SPECIAL_COUNT_AGGREGATOR)
+ * to know how many requests it has to receive.
+ */
+ private final ExpectedBarrier workersBarrier;
+ /** Progressable used to report progress */
+ private final Progressable progressable;
+
+ /**
+ * Constructor
+ *
+ * @param progressable Progressable used to report progress
+ */
+ public AllAggregatorServerData(Progressable progressable) {
+ this.progressable = progressable;
+ workersBarrier = new ExpectedBarrier(progressable);
+ masterBarrier = new ExpectedBarrier(progressable);
+ }
+
+ /**
+ * Register the class of the aggregator, received by master or worker.
+ *
+ * @param name Aggregator name
+ * @param aggregatorClass Class of the aggregator
+ */
+ public void registerAggregatorClass(String name,
+ Class<Aggregator<Writable>> aggregatorClass) {
+ aggregatorClassMap.put(name, aggregatorClass);
+ if (!aggregatorTypesMap.containsKey(aggregatorClass)) {
+ aggregatorTypesMap.putIfAbsent(aggregatorClass,
+ AggregatorUtils.newAggregatorInstance(aggregatorClass));
+ }
+ progressable.progress();
+ }
+
+ /**
+ * Set the value of aggregator from previous superstep,
+ * received by master or worker.
+ *
+ * @param name Name of the aggregator
+ * @param value Value of the aggregator
+ */
+ public void setAggregatorValue(String name, Writable value) {
+ aggregatedValuesMap.put(name, value);
+ progressable.progress();
+ }
+
+ /**
+ * Create initial aggregated value for an aggregator. Used so requests
+ * would be able to deserialize data.
+ * registerAggregatorClass needs to be called first to ensure that we have
+ * the class of the aggregator.
+ *
+ * @param name Name of the aggregator
+ * @return Empty aggregated value for this aggregator
+ */
+ public Writable createAggregatorInitialValue(String name) {
+ Class<Aggregator<Writable>> aggregatorClass = aggregatorClassMap.get(name);
+ Aggregator<Writable> aggregator = aggregatorTypesMap.get(aggregatorClass);
+ synchronized (aggregator) {
+ return aggregator.createInitialValue();
+ }
+ }
+
+ /**
+ * Notify this object that an aggregator request from master has been
+ * received.
+ *
+ * @param aggregatorData Byte request with data received from master
+ */
+ public void receivedRequestFromMaster(byte[] aggregatorData) {
+ masterData.add(aggregatorData);
+ masterBarrier.releaseOnePermit();
+ }
+
+ /**
+ * Notify this object about the total number of requests which should
+ * arrive from master.
+ *
+ * @param requestCount Number of requests which should arrive
+ */
+ public void receivedRequestCountFromMaster(long requestCount) {
+ masterBarrier.requirePermits(requestCount);
+ }
+
+ /**
+ * Notify this object that an aggregator request from some worker has been
+ * received.
+ */
+ public void receivedRequestFromWorker() {
+ workersBarrier.releaseOnePermit();
+ }
+
+ /**
+ * Notify this object about the total number of requests which should
+ * arrive from one of the workers.
+ *
+ * @param requestCount Number of requests which should arrive
+ */
+ public void receivedRequestCountFromWorker(long requestCount) {
+ workersBarrier.requirePermits(requestCount);
+ }
+
+ /**
+ * This function will wait until all aggregator requests from master have
+ * arrived, and return that data afterwards.
+ *
+ * @return Iterable through data received from master
+ */
+ public Iterable<byte[]> getDataFromMasterWhenReady() {
+ masterBarrier.waitForRequiredPermits(1);
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("getDataFromMasterWhenReady: " +
+ "Aggregator data for distribution ready");
+ }
+ return masterData;
+ }
+
+ /**
+ * This function will wait until all aggregator requests from workers have
+ * arrived, and fill the maps for next superstep when ready.
+ *
+ * @param numberOfWorkers Total number of workers in the job
+ * @param previousAggregatedValuesMap Map of values from previous
+ * superstep to fill out
+ * @param currentAggregatorMap Map of aggregators for current superstep to
+ * fill out. All aggregators in this map will
+ * be set to initial value.
+ */
+ public void fillNextSuperstepMapsWhenReady(
+ int numberOfWorkers,
+ Map<String, Writable> previousAggregatedValuesMap,
+ Map<String, Aggregator<Writable>> currentAggregatorMap) {
+ workersBarrier.waitForRequiredPermits(numberOfWorkers - 1);
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("fillNextSuperstepMapsWhenReady: Aggregators ready");
+ }
+ previousAggregatedValuesMap.clear();
+ previousAggregatedValuesMap.putAll(aggregatedValuesMap);
+ for (Map.Entry<String, Class<Aggregator<Writable>>> entry :
+ aggregatorClassMap.entrySet()) {
+ Aggregator<Writable> aggregator =
+ currentAggregatorMap.get(entry.getKey());
+ if (aggregator == null) {
+ currentAggregatorMap.put(entry.getKey(),
+ AggregatorUtils.newAggregatorInstance(entry.getValue()));
+ } else {
+ aggregator.reset();
+ }
+ }
+ }
+
+ /**
+ * Prepare for next superstep
+ */
+ public void reset() {
+ masterData.clear();
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("reset: Ready for next superstep");
+ }
+ }
+}
+
Added: giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/aggregators/CountingCache.java
URL: http://svn.apache.org/viewvc/giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/aggregators/CountingCache.java?rev=1402363&view=auto
==============================================================================
--- giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/aggregators/CountingCache.java (added)
+++ giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/aggregators/CountingCache.java Fri Oct 26 01:00:54 2012
@@ -0,0 +1,66 @@
+/*
+ * 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.giraph.comm.aggregators;
+
+import com.google.common.collect.Maps;
+
+import java.util.Map;
+
+/**
+ * Cache which counts the number of flushes per task id (destination worker
+ * id), so we know how many requests were sent to the worker
+ */
+public abstract class CountingCache {
+ /** Counts the number of flushes for each worker */
+ private Map<Integer, Long> countMap = Maps.newHashMap();
+
+ /**
+ * Increase count of flushes for worker with desired task id. Subclasses
+ * should call this method whenever flush is called.
+ *
+ * @param taskId Task id of worker
+ */
+ protected void incrementCounter(Integer taskId) {
+ Long currentCount = countMap.get(taskId);
+ countMap.put(taskId,
+ (currentCount == null) ? 1 : (currentCount + 1));
+ }
+
+ /**
+ * Get number of flushes for worker with desired task id
+ *
+ * @param taskId Task id of worker
+ * @return Number of objects for the worker
+ */
+ protected long getCount(Integer taskId) {
+ Long count = countMap.get(taskId);
+ if (count == null) {
+ return 0;
+ } else {
+ return count.longValue();
+ }
+ }
+
+ /**
+ * Reset the counts
+ */
+ public void reset() {
+ countMap.clear();
+ }
+}
Added: giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/aggregators/CountingOutputStream.java
URL: http://svn.apache.org/viewvc/giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/aggregators/CountingOutputStream.java?rev=1402363&view=auto
==============================================================================
--- giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/aggregators/CountingOutputStream.java (added)
+++ giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/aggregators/CountingOutputStream.java Fri Oct 26 01:00:54 2012
@@ -0,0 +1,119 @@
+/*
+ * 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.giraph.comm.aggregators;
+
+import java.io.ByteArrayOutputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+
+/**
+ * Wrapper for output stream which keeps the place in the beginning for the
+ * count of objects which were written to it
+ */
+public abstract class CountingOutputStream {
+ /** DataOutput to which subclasses will be writing data */
+ protected DataOutputStream dataOutput;
+ /** Byte output stream used by dataOutput */
+ private final ExtendedByteArrayOutputStream byteOutput;
+ /** Counter for objects which were written to the stream */
+ private int counter;
+
+ /**
+ * Default constructor
+ */
+ public CountingOutputStream() {
+ byteOutput = new ExtendedByteArrayOutputStream();
+ dataOutput = new DataOutputStream(byteOutput);
+ reset();
+ }
+
+ /**
+ * Subclasses should call this method when an object is written
+ */
+ protected void incrementCounter() {
+ counter++;
+ }
+
+ /**
+ * Get the number of bytes in the stream
+ *
+ * @return Number of bytes
+ */
+ protected int getSize() {
+ return byteOutput.size();
+ }
+
+ /**
+ * Returns all the data from the stream and clears it.
+ *
+ * @return Number of objects followed by the data written to the stream
+ */
+ public byte[] flush() {
+ byteOutput.writeIntOnPosition(counter, 0);
+ try {
+ dataOutput.flush();
+ } catch (IOException e) {
+ throw new IllegalStateException(
+ "flush: IOException occurred while flushing", e);
+ }
+ byte[] ret = byteOutput.toByteArray();
+ reset();
+ return ret;
+ }
+
+ /**
+ * Reset the stream
+ */
+ private void reset() {
+ byteOutput.reset();
+ dataOutput = new DataOutputStream(byteOutput);
+ // reserve the place for count to be written in the end
+ for (int i = 0; i < 4; i++) {
+ byteOutput.write(0);
+ }
+ counter = 0;
+ }
+
+ /**
+ * Subclass of {@link ByteArrayOutputStream} which provides an option to
+ * write int value over previously written data
+ */
+ public static class ExtendedByteArrayOutputStream extends
+ ByteArrayOutputStream {
+
+ /**
+ * Write integer value over previously written data at certain
+ * position in byte array
+ *
+ * @param value Value to write
+ * @param position Position from which to write
+ */
+ public void writeIntOnPosition(int value, int position) {
+ if (position + 4 > count) {
+ throw new IndexOutOfBoundsException(
+ "writeIntOnPosition: Tried to write int to position " + position +
+ " but current length is " + count);
+ }
+ buf[position] = (byte) ((value >>> 24) & 0xFF);
+ buf[position + 1] = (byte) ((value >>> 16) & 0xFF);
+ buf[position + 2] = (byte) ((value >>> 8) & 0xFF);
+ buf[position + 3] = (byte) ((value >>> 0) & 0xFF);
+ }
+ }
+}
Added: giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/aggregators/OwnerAggregatorServerData.java
URL: http://svn.apache.org/viewvc/giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/aggregators/OwnerAggregatorServerData.java?rev=1402363&view=auto
==============================================================================
--- giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/aggregators/OwnerAggregatorServerData.java (added)
+++ giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/aggregators/OwnerAggregatorServerData.java Fri Oct 26 01:00:54 2012
@@ -0,0 +1,187 @@
+/*
+ * 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.giraph.comm.aggregators;
+
+import org.apache.giraph.graph.Aggregator;
+import org.apache.giraph.utils.ExpectedBarrier;
+import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.util.Progressable;
+import org.apache.log4j.Logger;
+
+import com.google.common.base.Function;
+import com.google.common.collect.Iterables;
+import com.google.common.collect.Maps;
+
+import java.util.AbstractMap;
+import java.util.Map;
+import java.util.concurrent.ConcurrentMap;
+
+/**
+ * Class for holding aggregators which current worker owns,
+ * and aggregating partial aggregator values from workers.
+ *
+ * Protocol:
+ * 1. Before the beginning of superstep, worker receives its aggregators
+ * from master, and these aggregators will be registered to this class.
+ * Multiple registrations can be called concurrently.
+ * 2. During the superstep, whenever a worker finishes computation,
+ * it will send partial aggregated values to worker owner. This class is used
+ * to help deserialize the arriving aggregator values, and aggregate the values
+ * at the destination owner worker; these can happen concurrently.
+ * (we know step 1. is finished before anything from step 2. happens because
+ * other workers can't start computation before they receive aggregators
+ * which this worker owns)
+ * 3. This class also tracks the number of partial aggregator requests which
+ * worker received. In the end of superstep, getMyAggregatorValuesWhenReady
+ * will be called to ensure everything was received and get the values which
+ * need to be sent to master.
+ * Because of this counting, in step 2. even if worker owns no aggregators,
+ * it will still send a message without aggregator data.
+ * 4. In the end we reset to prepare for the next superstep.
+ */
+public class OwnerAggregatorServerData {
+ /** Class logger */
+ private static final Logger LOG =
+ Logger.getLogger(OwnerAggregatorServerData.class);
+ /** Map of aggregators which current worker owns */
+ private final ConcurrentMap<String, Aggregator<Writable>>
+ myAggregatorMap = Maps.newConcurrentMap();
+ /**
+ * Counts the requests with partial aggregated values from other workers.
+ * It uses values from special aggregators
+ * (named AggregatorUtils.SPECIAL_COUNT_AGGREGATOR)
+ * to know how many requests it has to receive.
+ */
+ private final ExpectedBarrier workersBarrier;
+ /** Progressable used to report progress */
+ private final Progressable progressable;
+
+ /**
+ * Constructor
+ *
+ * @param progressable Progressable used to report progress
+ */
+ public OwnerAggregatorServerData(Progressable progressable) {
+ this.progressable = progressable;
+ workersBarrier = new ExpectedBarrier(progressable);
+ }
+
+ /**
+ * Register an aggregator which current worker owns. Thread-safe.
+ *
+ * @param name Name of aggregator
+ * @param aggregatorClass Aggregator class
+ */
+ public void registerAggregator(String name,
+ Class<Aggregator<Writable>> aggregatorClass) {
+ if (LOG.isDebugEnabled() && myAggregatorMap.isEmpty()) {
+ LOG.debug("registerAggregator: The first registration after a reset()");
+ }
+ myAggregatorMap.putIfAbsent(name,
+ AggregatorUtils.newAggregatorInstance(aggregatorClass));
+ progressable.progress();
+ }
+
+ /**
+ * Aggregate partial value of one of current worker's aggregators.
+ *
+ * Thread-safe. Call only after aggregators have been registered.
+ *
+ * @param name Name of the aggregator
+ * @param value Value to aggregate to it
+ */
+ public void aggregate(String name, Writable value) {
+ Aggregator<Writable> aggregator = myAggregatorMap.get(name);
+ synchronized (aggregator) {
+ aggregator.aggregate(value);
+ }
+ progressable.progress();
+ }
+
+ /**
+ * Create initial aggregated value for an aggregator. Used so requests
+ * would be able to deserialize data.
+ *
+ * Thread-safe. Call only after aggregators have been registered.
+ *
+ * @param name Name of the aggregator
+ * @return Empty aggregated value for this aggregator
+ */
+ public Writable createAggregatorInitialValue(String name) {
+ Aggregator<Writable> aggregator = myAggregatorMap.get(name);
+ synchronized (aggregator) {
+ return aggregator.createInitialValue();
+ }
+ }
+
+ /**
+ * Notify this object that a partial aggregated values request from some
+ * worker have been received. Thread-safe.
+ */
+ public void receivedRequestFromWorker() {
+ workersBarrier.releaseOnePermit();
+ }
+
+ /**
+ * Notify this object about the total number of requests which should
+ * arrive from one of the workers. Thread-safe.
+ *
+ * @param requestCount Number of requests which should arrive
+ */
+ public void receivedRequestCountFromWorker(long requestCount) {
+ workersBarrier.requirePermits(requestCount);
+ }
+
+ /**
+ * This function will wait until all partial aggregated values from all
+ * workers are ready and aggregated, and return final aggregated values
+ * afterwards.
+ *
+ * @param numberOfWorkers Total number of workers in the job
+ * @return Iterable through final aggregated values which this worker owns
+ */
+ public Iterable<Map.Entry<String, Writable>>
+ getMyAggregatorValuesWhenReady(int numberOfWorkers) {
+ workersBarrier.waitForRequiredPermits(numberOfWorkers - 1);
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("getMyAggregatorValuesWhenReady: Values ready");
+ }
+ return Iterables.transform(myAggregatorMap.entrySet(),
+ new Function<Map.Entry<String, Aggregator<Writable>>,
+ Map.Entry<String, Writable>>() {
+ @Override
+ public Map.Entry<String, Writable> apply(
+ Map.Entry<String, Aggregator<Writable>> aggregator) {
+ return new AbstractMap.SimpleEntry<String, Writable>(
+ aggregator.getKey(),
+ aggregator.getValue().getAggregatedValue());
+ }
+ });
+ }
+
+ /**
+ * Prepare for next superstep
+ */
+ public void reset() {
+ myAggregatorMap.clear();
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("reset: Ready for next superstep");
+ }
+ }
+}
Added: giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/aggregators/SendAggregatedValueCache.java
URL: http://svn.apache.org/viewvc/giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/aggregators/SendAggregatedValueCache.java?rev=1402363&view=auto
==============================================================================
--- giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/aggregators/SendAggregatedValueCache.java (added)
+++ giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/aggregators/SendAggregatedValueCache.java Fri Oct 26 01:00:54 2012
@@ -0,0 +1,87 @@
+/*
+ * 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.giraph.comm.aggregators;
+
+import org.apache.hadoop.io.LongWritable;
+import org.apache.hadoop.io.Writable;
+
+import com.google.common.collect.Maps;
+
+import java.io.IOException;
+import java.util.Map;
+
+/**
+ * Takes and serializes aggregated values and keeps them grouped by owner
+ * partition id, to be sent in bulk.
+ */
+public class SendAggregatedValueCache extends CountingCache {
+ /** Map from worker partition id to aggregator output stream */
+ private final Map<Integer, AggregatedValueOutputStream> aggregatorMap =
+ Maps.newHashMap();
+
+ /**
+ * Add aggregated value to the cache
+ *
+ * @param taskId Task id of worker which owns the aggregator
+ * @param aggregatorName Name of the aggregator
+ * @param aggregatedValue Value of the aggregator
+ * @return Number of bytes in serialized data for this worker
+ * @throws IOException
+ */
+ public int addAggregator(Integer taskId, String aggregatorName,
+ Writable aggregatedValue) throws IOException {
+ AggregatedValueOutputStream out = aggregatorMap.get(taskId);
+ if (out == null) {
+ out = new AggregatedValueOutputStream();
+ aggregatorMap.put(taskId, out);
+ }
+ return out.addAggregator(aggregatorName, aggregatedValue);
+ }
+
+ /**
+ * Remove and get aggregators for certain worker
+ *
+ * @param taskId Partition id of worker owner
+ * @return Serialized aggregator data for this worker
+ */
+ public byte[] removeAggregators(Integer taskId) {
+ incrementCounter(taskId);
+ AggregatedValueOutputStream out = aggregatorMap.remove(taskId);
+ if (out == null) {
+ return new byte[4];
+ } else {
+ return out.flush();
+ }
+ }
+
+ /**
+ * Creates fake aggregator which will hold the total number of aggregator
+ * requests for worker with selected task id. This should be called after all
+ * aggregators for the worker have been added to the cache.
+ *
+ * @param taskId Destination worker's task id
+ * @throws IOException
+ */
+ public void addCountAggregator(Integer taskId) throws IOException {
+ // current number of requests, plus one for the last flush
+ long totalCount = getCount(taskId) + 1;
+ addAggregator(taskId, AggregatorUtils.SPECIAL_COUNT_AGGREGATOR,
+ new LongWritable(totalCount));
+ }
+}
Added: giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/aggregators/SendAggregatorCache.java
URL: http://svn.apache.org/viewvc/giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/aggregators/SendAggregatorCache.java?rev=1402363&view=auto
==============================================================================
--- giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/aggregators/SendAggregatorCache.java (added)
+++ giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/aggregators/SendAggregatorCache.java Fri Oct 26 01:00:54 2012
@@ -0,0 +1,91 @@
+/*
+ * 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.giraph.comm.aggregators;
+
+import org.apache.giraph.graph.Aggregator;
+import org.apache.hadoop.io.LongWritable;
+import org.apache.hadoop.io.Writable;
+
+import com.google.common.collect.Maps;
+
+import java.io.IOException;
+import java.util.Map;
+
+/**
+ * Takes and serializes aggregators and keeps them grouped by owner
+ * partition id, to be sent in bulk.
+ */
+public class SendAggregatorCache extends CountingCache {
+ /** Map from worker partition id to aggregator output stream */
+ private final Map<Integer, AggregatorOutputStream> aggregatorMap =
+ Maps.newHashMap();
+
+ /**
+ * Add aggregator to the cache
+ *
+ * @param taskId Task id of worker which owns the aggregator
+ * @param aggregatorName Name of the aggregator
+ * @param aggregatorClass Class of the aggregator
+ * @param aggregatedValue Value of the aggregator
+ * @return Number of bytes in serialized data for this worker
+ * @throws IOException
+ */
+ public int addAggregator(Integer taskId, String aggregatorName,
+ Class<? extends Aggregator> aggregatorClass,
+ Writable aggregatedValue) throws IOException {
+ AggregatorOutputStream out = aggregatorMap.get(taskId);
+ if (out == null) {
+ out = new AggregatorOutputStream();
+ aggregatorMap.put(taskId, out);
+ }
+ return out.addAggregator(aggregatorName, aggregatorClass,
+ aggregatedValue);
+ }
+
+ /**
+ * Remove and get aggregators for certain worker
+ *
+ * @param taskId Task id of worker owner
+ * @return Serialized aggregator data for this worker
+ */
+ public byte[] removeAggregators(Integer taskId) {
+ incrementCounter(taskId);
+ AggregatorOutputStream out = aggregatorMap.remove(taskId);
+ if (out == null) {
+ return new byte[4];
+ } else {
+ return out.flush();
+ }
+ }
+
+ /**
+ * Creates fake aggregator which will hold the total number of aggregator
+ * requests for worker with selected task id. This should be called after all
+ * aggregators for the worker have been added to the cache.
+ *
+ * @param taskId Destination worker's task id
+ * @throws IOException
+ */
+ public void addCountAggregator(Integer taskId) throws IOException {
+ // current number of requests, plus one for the last flush
+ long totalCount = getCount(taskId) + 1;
+ addAggregator(taskId, AggregatorUtils.SPECIAL_COUNT_AGGREGATOR,
+ Aggregator.class, new LongWritable(totalCount));
+ }
+}
Added: giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/aggregators/WorkerAggregatorRequestProcessor.java
URL: http://svn.apache.org/viewvc/giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/aggregators/WorkerAggregatorRequestProcessor.java?rev=1402363&view=auto
==============================================================================
--- giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/aggregators/WorkerAggregatorRequestProcessor.java (added)
+++ giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/aggregators/WorkerAggregatorRequestProcessor.java Fri Oct 26 01:00:54 2012
@@ -0,0 +1,64 @@
+/*
+ * 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.giraph.comm.aggregators;
+
+import org.apache.hadoop.io.Writable;
+
+import java.io.IOException;
+
+/**
+ * Aggregates worker aggregator requests and sends them off
+ */
+public interface WorkerAggregatorRequestProcessor {
+ /**
+ * Sends worker aggregated value to the owner of aggregator
+ *
+ * @param aggregatorName Name of the aggregator
+ * @param aggregatedValue Value of the aggregator
+ * @throws java.io.IOException
+ * @return True if aggregated value will be sent, false if this worker is
+ * the owner of the aggregator
+ */
+ boolean sendAggregatedValue(String aggregatorName,
+ Writable aggregatedValue) throws IOException;
+
+ /**
+ * Flush aggregated values cache.
+ *
+ * @throws IOException
+ */
+ void flush() throws IOException;
+
+ /**
+ * Sends aggregated values to the master. This worker is the owner of these
+ * aggregators.
+ *
+ * @param aggregatorData Serialized aggregator data
+ * @throws IOException
+ */
+ void sendAggregatedValuesToMaster(byte[] aggregatorData) throws IOException;
+
+ /**
+ * Sends aggregators to all other workers
+ *
+ * @param aggregatorDataList Serialized aggregator data split into chunks
+ */
+ void distributeAggregators(
+ Iterable<byte[]> aggregatorDataList) throws IOException;
+}
Added: giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/aggregators/package-info.java
URL: http://svn.apache.org/viewvc/giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/aggregators/package-info.java?rev=1402363&view=auto
==============================================================================
--- giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/aggregators/package-info.java (added)
+++ giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/aggregators/package-info.java Fri Oct 26 01:00:54 2012
@@ -0,0 +1,21 @@
+/*
+ * 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 for classes which are used to handle aggregators.
+ */
+package org.apache.giraph.comm.aggregators;
Modified: giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/netty/NettyMasterClient.java
URL: http://svn.apache.org/viewvc/giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/netty/NettyMasterClient.java?rev=1402363&r1=1402362&r2=1402363&view=diff
==============================================================================
--- giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/netty/NettyMasterClient.java (original)
+++ giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/netty/NettyMasterClient.java Fri Oct 26 01:00:54 2012
@@ -21,7 +21,16 @@ package org.apache.giraph.comm.netty;
import org.apache.giraph.ImmutableClassesGiraphConfiguration;
import org.apache.giraph.bsp.CentralizedServiceMaster;
import org.apache.giraph.comm.MasterClient;
+import org.apache.giraph.comm.aggregators.AggregatorUtils;
+import org.apache.giraph.comm.aggregators.SendAggregatorCache;
+import org.apache.giraph.comm.requests.SendAggregatorsToOwnerRequest;
+import org.apache.giraph.graph.Aggregator;
+import org.apache.giraph.graph.WorkerInfo;
import org.apache.hadoop.mapreduce.Mapper;
+import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.util.Progressable;
+
+import java.io.IOException;
/**
* Netty implementation of {@link MasterClient}
@@ -31,6 +40,13 @@ public class NettyMasterClient implement
private final NettyClient nettyClient;
/** Worker information for current superstep */
private CentralizedServiceMaster<?, ?, ?, ?> service;
+ /** Cached map of partition ids to serialized aggregator data */
+ private final SendAggregatorCache sendAggregatorCache =
+ new SendAggregatorCache();
+ /** How big a single aggregator request can be */
+ private final int maxBytesPerAggregatorRequest;
+ /** Progressable used to report progress */
+ private final Progressable progressable;
/**
* Constructor
@@ -44,6 +60,10 @@ public class NettyMasterClient implement
CentralizedServiceMaster<?, ?, ?, ?> service) {
this.nettyClient = new NettyClient(context, configuration);
this.service = service;
+ this.progressable = context;
+ maxBytesPerAggregatorRequest = configuration.getInt(
+ AggregatorUtils.MAX_BYTES_PER_AGGREGATOR_REQUEST,
+ AggregatorUtils.MAX_BYTES_PER_AGGREGATOR_REQUEST_DEFAULT);
}
@Override
@@ -52,6 +72,41 @@ public class NettyMasterClient implement
}
@Override
+ public void sendAggregator(String aggregatorName,
+ Class<? extends Aggregator> aggregatorClass,
+ Writable aggregatedValue) throws IOException {
+ WorkerInfo owner =
+ AggregatorUtils.getOwner(aggregatorName, service.getWorkerInfoList());
+ int currentSize = sendAggregatorCache.addAggregator(owner.getTaskId(),
+ aggregatorName, aggregatorClass, aggregatedValue);
+ if (currentSize >= maxBytesPerAggregatorRequest) {
+ flushAggregatorsToWorker(owner);
+ }
+ }
+
+ @Override
+ public void finishSendingAggregatedValues() throws IOException {
+ for (WorkerInfo worker : service.getWorkerInfoList()) {
+ sendAggregatorCache.addCountAggregator(worker.getTaskId());
+ flushAggregatorsToWorker(worker);
+ progressable.progress();
+ }
+ sendAggregatorCache.reset();
+ }
+
+ /**
+ * Send aggregators from cache to worker.
+ *
+ * @param worker Worker which we want to send aggregators to
+ */
+ private void flushAggregatorsToWorker(WorkerInfo worker) {
+ byte[] aggregatorData =
+ sendAggregatorCache.removeAggregators(worker.getTaskId());
+ nettyClient.sendWritableRequest(
+ worker.getTaskId(), new SendAggregatorsToOwnerRequest(aggregatorData));
+ }
+
+ @Override
public void flush() {
nettyClient.waitAllRequests();
}
Modified: giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/netty/NettyMasterClientServer.java
URL: http://svn.apache.org/viewvc/giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/netty/NettyMasterClientServer.java?rev=1402363&r1=1402362&r2=1402363&view=diff
==============================================================================
--- giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/netty/NettyMasterClientServer.java (original)
+++ giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/netty/NettyMasterClientServer.java Fri Oct 26 01:00:54 2012
@@ -23,8 +23,11 @@ import org.apache.giraph.bsp.Centralized
import org.apache.giraph.comm.MasterClient;
import org.apache.giraph.comm.MasterClientServer;
import org.apache.giraph.comm.MasterServer;
+import org.apache.giraph.graph.Aggregator;
+import org.apache.hadoop.io.Writable;
import org.apache.hadoop.mapreduce.Mapper;
+import java.io.IOException;
import java.net.InetSocketAddress;
/**
@@ -48,7 +51,8 @@ public class NettyMasterClientServer imp
ImmutableClassesGiraphConfiguration configuration,
CentralizedServiceMaster<?, ?, ?, ?> service) {
client = new NettyMasterClient(context, configuration, service);
- server = new NettyMasterServer(configuration);
+ server = new NettyMasterServer(configuration,
+ service.getAggregatorHandler());
}
@Override
@@ -57,6 +61,18 @@ public class NettyMasterClientServer imp
}
@Override
+ public void sendAggregator(String aggregatorName,
+ Class<? extends Aggregator> aggregatorClass,
+ Writable aggregatedValue) throws IOException {
+ client.sendAggregator(aggregatorName, aggregatorClass, aggregatedValue);
+ }
+
+ @Override
+ public void finishSendingAggregatedValues() throws IOException {
+ client.finishSendingAggregatedValues();
+ }
+
+ @Override
public void flush() {
client.flush();
}
Modified: giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/netty/NettyMasterServer.java
URL: http://svn.apache.org/viewvc/giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/netty/NettyMasterServer.java?rev=1402363&r1=1402362&r2=1402363&view=diff
==============================================================================
--- giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/netty/NettyMasterServer.java (original)
+++ giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/netty/NettyMasterServer.java Fri Oct 26 01:00:54 2012
@@ -21,6 +21,7 @@ package org.apache.giraph.comm.netty;
import org.apache.giraph.ImmutableClassesGiraphConfiguration;
import org.apache.giraph.comm.netty.handler.MasterRequestServerHandler;
import org.apache.giraph.comm.MasterServer;
+import org.apache.giraph.graph.MasterAggregatorHandler;
import java.net.InetSocketAddress;
@@ -35,10 +36,12 @@ public class NettyMasterServer implement
* Constructor
*
* @param conf Hadoop configuration
+ * @param aggregatorHandler Master aggregator handler
*/
- public NettyMasterServer(ImmutableClassesGiraphConfiguration conf) {
+ public NettyMasterServer(ImmutableClassesGiraphConfiguration conf,
+ MasterAggregatorHandler aggregatorHandler) {
nettyServer = new NettyServer(conf,
- new MasterRequestServerHandler.Factory());
+ new MasterRequestServerHandler.Factory(aggregatorHandler));
nettyServer.start();
}
Added: giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/netty/NettyWorkerAggregatorRequestProcessor.java
URL: http://svn.apache.org/viewvc/giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/netty/NettyWorkerAggregatorRequestProcessor.java?rev=1402363&view=auto
==============================================================================
--- giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/netty/NettyWorkerAggregatorRequestProcessor.java (added)
+++ giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/netty/NettyWorkerAggregatorRequestProcessor.java Fri Oct 26 01:00:54 2012
@@ -0,0 +1,146 @@
+/*
+ * 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.giraph.comm.netty;
+
+import org.apache.giraph.ImmutableClassesGiraphConfiguration;
+import org.apache.giraph.bsp.CentralizedServiceWorker;
+import org.apache.giraph.comm.aggregators.WorkerAggregatorRequestProcessor;
+import org.apache.giraph.comm.WorkerClient;
+import org.apache.giraph.comm.aggregators.AggregatorUtils;
+import org.apache.giraph.comm.aggregators.SendAggregatedValueCache;
+import org.apache.giraph.comm.requests.SendAggregatorsToMasterRequest;
+import org.apache.giraph.comm.requests.SendAggregatorsToWorkerRequest;
+import org.apache.giraph.comm.requests.SendWorkerAggregatorsRequest;
+import org.apache.giraph.graph.WorkerInfo;
+import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.util.Progressable;
+
+import java.io.IOException;
+
+/**
+ * Netty implementation of {@link WorkerAggregatorRequestProcessor}
+ */
+public class NettyWorkerAggregatorRequestProcessor
+ implements WorkerAggregatorRequestProcessor {
+ /** Progressable used to report progress */
+ private final Progressable progressable;
+ /** NettyClient that could be shared among one or more instances */
+ private final WorkerClient<?, ?, ?, ?> workerClient;
+ /** Service worker */
+ private final CentralizedServiceWorker<?, ?, ?, ?> serviceWorker;
+ /** Cached map of partition ids to serialized aggregator data */
+ private final SendAggregatedValueCache sendAggregatedValueCache =
+ new SendAggregatedValueCache();
+ /** How big a single aggregator request can be */
+ private final int maxBytesPerAggregatorRequest;
+
+ /**
+ * Constructor.
+ *
+ * @param progressable Progressable used to report progress
+ * @param configuration Configuration
+ * @param serviceWorker Service worker
+ */
+ public NettyWorkerAggregatorRequestProcessor(
+ Progressable progressable,
+ ImmutableClassesGiraphConfiguration<?, ?, ?, ?> configuration,
+ CentralizedServiceWorker<?, ?, ?, ?> serviceWorker) {
+ this.serviceWorker = serviceWorker;
+ this.workerClient = serviceWorker.getWorkerClient();
+ this.progressable = progressable;
+ maxBytesPerAggregatorRequest = configuration.getInt(
+ AggregatorUtils.MAX_BYTES_PER_AGGREGATOR_REQUEST,
+ AggregatorUtils.MAX_BYTES_PER_AGGREGATOR_REQUEST_DEFAULT);
+
+ }
+
+ @Override
+ public boolean sendAggregatedValue(String aggregatorName,
+ Writable aggregatedValue) throws IOException {
+ WorkerInfo owner =
+ AggregatorUtils.getOwner(aggregatorName,
+ serviceWorker.getWorkerInfoList());
+ if (isThisWorker(owner)) {
+ return false;
+ } else {
+ int currentSize = sendAggregatedValueCache.addAggregator(
+ owner.getTaskId(), aggregatorName, aggregatedValue);
+ if (currentSize >= maxBytesPerAggregatorRequest) {
+ flushAggregatorsToWorker(owner);
+ }
+ return true;
+ }
+ }
+
+ @Override
+ public void flush() throws IOException {
+ for (WorkerInfo workerInfo : serviceWorker.getWorkerInfoList()) {
+ if (!isThisWorker(workerInfo)) {
+ sendAggregatedValueCache.addCountAggregator(workerInfo.getTaskId());
+ flushAggregatorsToWorker(workerInfo);
+ progressable.progress();
+ }
+ }
+ sendAggregatedValueCache.reset();
+ }
+
+ /**
+ * Send aggregators from cache to worker.
+ *
+ * @param worker Worker which we want to send aggregators to
+ */
+ private void flushAggregatorsToWorker(WorkerInfo worker) {
+ byte[] aggregatorData =
+ sendAggregatedValueCache.removeAggregators(worker.getTaskId());
+ workerClient.sendWritableRequest(worker.getTaskId(),
+ new SendWorkerAggregatorsRequest(aggregatorData));
+ }
+
+ @Override
+ public void sendAggregatedValuesToMaster(
+ byte[] aggregatorData) throws IOException {
+ workerClient.sendWritableRequest(serviceWorker.getMasterInfo().getTaskId(),
+ new SendAggregatorsToMasterRequest(aggregatorData));
+ }
+
+ @Override
+ public void distributeAggregators(
+ Iterable<byte[]> aggregatorDataList) throws IOException {
+ for (byte[] aggregatorData : aggregatorDataList) {
+ SendAggregatorsToWorkerRequest request =
+ new SendAggregatorsToWorkerRequest(aggregatorData);
+ for (WorkerInfo worker : serviceWorker.getWorkerInfoList()) {
+ if (!isThisWorker(worker)) {
+ workerClient.sendWritableRequest(worker.getTaskId(), request);
+ }
+ progressable.progress();
+ }
+ }
+ }
+
+ /**
+ * Check if workerInfo describes current worker.
+ *
+ * @param workerInfo Worker to check
+ * @return True iff workerInfo corresponds to current worker.
+ */
+ private boolean isThisWorker(WorkerInfo workerInfo) {
+ return serviceWorker.getWorkerInfo().getTaskId() == workerInfo.getTaskId();
+ }
+}
Modified: giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/netty/handler/MasterRequestServerHandler.java
URL: http://svn.apache.org/viewvc/giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/netty/handler/MasterRequestServerHandler.java?rev=1402363&r1=1402362&r2=1402363&view=diff
==============================================================================
--- giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/netty/handler/MasterRequestServerHandler.java (original)
+++ giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/netty/handler/MasterRequestServerHandler.java Fri Oct 26 01:00:54 2012
@@ -19,36 +19,56 @@
package org.apache.giraph.comm.netty.handler;
import org.apache.giraph.comm.requests.MasterRequest;
+import org.apache.giraph.graph.MasterAggregatorHandler;
import org.apache.hadoop.conf.Configuration;
/** Handler for requests on master */
public class MasterRequestServerHandler extends
RequestServerHandler<MasterRequest> {
+ /** Aggregator handler */
+ private final MasterAggregatorHandler aggregatorHandler;
+
/**
* Constructor
*
* @param workerRequestReservedMap Worker request reservation map
* @param conf Configuration
+ * @param aggregatorHandler Master aggregator handler
*/
public MasterRequestServerHandler(
- WorkerRequestReservedMap workerRequestReservedMap, Configuration conf) {
+ WorkerRequestReservedMap workerRequestReservedMap, Configuration conf,
+ MasterAggregatorHandler aggregatorHandler) {
super(workerRequestReservedMap, conf);
+ this.aggregatorHandler = aggregatorHandler;
}
@Override
public void processRequest(MasterRequest request) {
- request.doRequest();
+ request.doRequest(aggregatorHandler);
}
/**
* Factory for {@link MasterRequestServerHandler}
*/
public static class Factory implements RequestServerHandler.Factory {
+ /** Master aggregator handler */
+ private final MasterAggregatorHandler aggregatorHandler;
+
+ /**
+ * Constructor
+ *
+ * @param aggregatorHandler Master aggregator handler
+ */
+ public Factory(MasterAggregatorHandler aggregatorHandler) {
+ this.aggregatorHandler = aggregatorHandler;
+ }
+
@Override
public RequestServerHandler newHandler(
WorkerRequestReservedMap workerRequestReservedMap,
Configuration conf) {
- return new MasterRequestServerHandler(workerRequestReservedMap, conf);
+ return new MasterRequestServerHandler(workerRequestReservedMap, conf,
+ aggregatorHandler);
}
}
}
Added: giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/requests/ByteArrayRequest.java
URL: http://svn.apache.org/viewvc/giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/requests/ByteArrayRequest.java?rev=1402363&view=auto
==============================================================================
--- giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/requests/ByteArrayRequest.java (added)
+++ giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/requests/ByteArrayRequest.java Fri Oct 26 01:00:54 2012
@@ -0,0 +1,79 @@
+/*
+ * 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.giraph.comm.requests;
+
+import java.io.ByteArrayInputStream;
+import java.io.DataInput;
+import java.io.DataInputStream;
+import java.io.DataOutput;
+import java.io.IOException;
+
+/**
+ * Abstract request which has a byte array as its data
+ */
+public abstract class ByteArrayRequest extends WritableRequest {
+ /** Request data */
+ private byte[] data;
+
+ /**
+ * Constructor
+ *
+ * @param data Request data
+ */
+ ByteArrayRequest(byte[] data) {
+ this.data = data;
+ }
+
+ /**
+ * Constructor used for reflection only
+ */
+ ByteArrayRequest() {
+ }
+
+ /**
+ * Get request data
+ *
+ * @return Request data
+ */
+ public byte[] getData() {
+ return data;
+ }
+
+ /**
+ * Get request data in the form of {@link DataInput}
+ *
+ * @return Request data as {@link DataInput}
+ */
+ public DataInput getDataInput() {
+ return new DataInputStream(new ByteArrayInputStream(data));
+ }
+
+ @Override
+ void readFieldsRequest(DataInput input) throws IOException {
+ int dataLength = input.readInt();
+ data = new byte[dataLength];
+ input.readFully(data);
+ }
+
+ @Override
+ void writeRequest(DataOutput output) throws IOException {
+ output.writeInt(data.length);
+ output.write(data);
+ }
+}
Modified: giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/requests/MasterRequest.java
URL: http://svn.apache.org/viewvc/giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/requests/MasterRequest.java?rev=1402363&r1=1402362&r2=1402363&view=diff
==============================================================================
--- giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/requests/MasterRequest.java (original)
+++ giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/requests/MasterRequest.java Fri Oct 26 01:00:54 2012
@@ -18,12 +18,16 @@
package org.apache.giraph.comm.requests;
+import org.apache.giraph.graph.MasterAggregatorHandler;
+
/**
* Interface for requests sent to master to extend
*/
public interface MasterRequest {
/**
* Execute the request
+ *
+ * @param aggregatorHandler Master aggregator handler
*/
- void doRequest();
+ void doRequest(MasterAggregatorHandler aggregatorHandler);
}
Modified: giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/requests/RequestType.java
URL: http://svn.apache.org/viewvc/giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/requests/RequestType.java?rev=1402363&r1=1402362&r2=1402363&view=diff
==============================================================================
--- giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/requests/RequestType.java (original)
+++ giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/requests/RequestType.java Fri Oct 26 01:00:54 2012
@@ -43,7 +43,15 @@ public enum RequestType {
SEND_PARTITION_CURRENT_MESSAGES_REQUEST
(SendPartitionCurrentMessagesRequest.class),
/** Send a partition of mutations */
- SEND_PARTITION_MUTATIONS_REQUEST(SendPartitionMutationsRequest.class);
+ SEND_PARTITION_MUTATIONS_REQUEST(SendPartitionMutationsRequest.class),
+ /** Send aggregated values from one worker's vertices */
+ SEND_WORKER_AGGREGATORS_REQUEST(SendWorkerAggregatorsRequest.class),
+ /** Send aggregated values from worker owner to master */
+ SEND_AGGREGATORS_TO_MASTER_REQUEST(SendAggregatorsToMasterRequest.class),
+ /** Send aggregators from master to worker owners */
+ SEND_AGGREGATORS_TO_OWNER_REQUEST(SendAggregatorsToOwnerRequest.class),
+ /** Send aggregators from worker owner to other workers */
+ SEND_AGGREGATORS_TO_WORKER_REQUEST(SendAggregatorsToWorkerRequest.class);
/** Class of request which this type corresponds to */
private final Class<? extends WritableRequest> requestClass;
Added: giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/requests/SendAggregatorsToMasterRequest.java
URL: http://svn.apache.org/viewvc/giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/requests/SendAggregatorsToMasterRequest.java?rev=1402363&view=auto
==============================================================================
--- giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/requests/SendAggregatorsToMasterRequest.java (added)
+++ giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/requests/SendAggregatorsToMasterRequest.java Fri Oct 26 01:00:54 2012
@@ -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.giraph.comm.requests;
+
+import org.apache.giraph.graph.MasterAggregatorHandler;
+
+import java.io.IOException;
+
+/**
+ * Request to send final aggregated values from worker which owns
+ * aggregators to the master
+ */
+public class SendAggregatorsToMasterRequest extends ByteArrayRequest
+ implements MasterRequest {
+
+ /**
+ * Constructor
+ *
+ * @param data Serialized aggregator data
+ */
+ public SendAggregatorsToMasterRequest(byte[] data) {
+ super(data);
+ }
+
+ /**
+ * Constructor used for reflection only
+ */
+ public SendAggregatorsToMasterRequest() {
+ }
+
+ @Override
+ public void doRequest(MasterAggregatorHandler aggregatorHandler) {
+ try {
+ aggregatorHandler.acceptAggregatedValues(getDataInput());
+ } catch (IOException e) {
+ throw new IllegalStateException("doRequest: " +
+ "IOException occurred while processing request", e);
+ }
+ }
+
+ @Override
+ public RequestType getType() {
+ return RequestType.SEND_AGGREGATORS_TO_MASTER_REQUEST;
+ }
+}