You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@giraph.apache.org by ni...@apache.org on 2012/12/20 05:25:31 UTC
[18/51] [partial] GIRAPH-457: update module names (nitay)
http://git-wip-us.apache.org/repos/asf/giraph/blob/45851391/giraph-core/src/main/java/org/apache/giraph/graph/Vertex.java
----------------------------------------------------------------------
diff --git a/giraph-core/src/main/java/org/apache/giraph/graph/Vertex.java b/giraph-core/src/main/java/org/apache/giraph/graph/Vertex.java
new file mode 100644
index 0000000..6db4735
--- /dev/null
+++ b/giraph-core/src/main/java/org/apache/giraph/graph/Vertex.java
@@ -0,0 +1,405 @@
+/*
+ * 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.graph;
+
+import com.google.common.collect.Iterables;
+import com.google.common.collect.Lists;
+import org.apache.giraph.conf.ImmutableClassesGiraphConfigurable;
+import org.apache.giraph.conf.ImmutableClassesGiraphConfiguration;
+import org.apache.giraph.graph.partition.PartitionOwner;
+import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.io.WritableComparable;
+import org.apache.hadoop.mapreduce.Mapper;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.util.Collections;
+import java.util.List;
+
+/**
+ * Basic interface for writing a BSP application for computation.
+ *
+ * @param <I> Vertex id
+ * @param <V> Vertex data
+ * @param <E> Edge data
+ * @param <M> Message data
+ */
+@SuppressWarnings("rawtypes")
+public abstract class Vertex<I extends WritableComparable,
+ V extends Writable, E extends Writable, M extends Writable>
+ implements WorkerAggregatorUsage, Writable,
+ ImmutableClassesGiraphConfigurable<I, V, E, M> {
+ /** Vertex id. */
+ private I id;
+ /** Vertex value. */
+ private V value;
+ /** If true, do not do anymore computation on this vertex. */
+ private boolean halt;
+ /** Global graph state **/
+ private GraphState<I, V, E, M> graphState;
+ /** Configuration */
+ private ImmutableClassesGiraphConfiguration<I, V, E, M> conf;
+
+ /**
+ * This method must be called after instantiation of a vertex
+ * with ImmutableClassesGiraphConfiguration
+ * unless deserialization from readFields() is
+ * called.
+ *
+ * @param id Will be the vertex id
+ * @param value Will be the vertex value
+ * @param edges Iterable of edges
+ */
+ public void initialize(I id, V value, Iterable<Edge<I, E>> edges) {
+ this.id = id;
+ this.value = value;
+ setEdges(edges);
+ }
+
+ /**
+ * This method only sets id and value. Can be used by Vertex
+ * implementations in readFields().
+ *
+ * @param id Vertex id
+ * @param value Vertex value
+ */
+ public void initialize(I id, V value) {
+ this.id = id;
+ this.value = value;
+ setEdges(Collections.<Edge<I, E>>emptyList());
+ }
+
+ /**
+ * Set the outgoing edges for this vertex.
+ *
+ * @param edges Iterable of edges
+ */
+ public abstract void setEdges(Iterable<Edge<I, E>> edges);
+
+ /**
+ * Must be defined by user to do computation on a single Vertex.
+ *
+ * @param messages Messages that were sent to this vertex in the previous
+ * superstep. Each message is only guaranteed to have
+ * a life expectancy as long as next() is not called.
+ * @throws IOException
+ */
+ public abstract void compute(Iterable<M> messages) throws IOException;
+
+ /**
+ * Retrieves the current superstep.
+ *
+ * @return Current superstep
+ */
+ public long getSuperstep() {
+ return getGraphState().getSuperstep();
+ }
+
+ /**
+ * Get the vertex id.
+ *
+ * @return My vertex id.
+ */
+ public I getId() {
+ return id;
+ }
+
+ /**
+ * Get the vertex value (data stored with vertex)
+ *
+ * @return Vertex value
+ */
+ public V getValue() {
+ return value;
+ }
+
+ /**
+ * Set the vertex data (immediately visible in the computation)
+ *
+ * @param value Vertex data to be set
+ */
+ public void setValue(V value) {
+ this.value = value;
+ }
+
+ /**
+ * Get the total (all workers) number of vertices that
+ * existed in the previous superstep.
+ *
+ * @return Total number of vertices (-1 if first superstep)
+ */
+ public long getTotalNumVertices() {
+ return getGraphState().getTotalNumVertices();
+ }
+
+ /**
+ * Get the total (all workers) number of edges that
+ * existed in the previous superstep.
+ *
+ * @return Total number of edges (-1 if first superstep)
+ */
+ public long getTotalNumEdges() {
+ return getGraphState().getTotalNumEdges();
+ }
+
+ /**
+ * Get a read-only view of the out-edges of this vertex.
+ *
+ * @return the out edges (sort order determined by subclass implementation).
+ */
+ public abstract Iterable<Edge<I, E>> getEdges();
+
+ /**
+ * Does an edge with the target vertex id exist?
+ *
+ * @param targetVertexId Target vertex id to check
+ * @return true if there is an edge to the target
+ */
+ public boolean hasEdge(I targetVertexId) {
+ for (Edge<I, E> edge : getEdges()) {
+ if (edge.getTargetVertexId().equals(targetVertexId)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ /**
+ * Get the edge value associated with a target vertex id.
+ *
+ * @param targetVertexId Target vertex id to check
+ *
+ * @return the value of the edge to targetVertexId (or null if there
+ * is no edge to it)
+ */
+ public E getEdgeValue(I targetVertexId) {
+ for (Edge<I, E> edge : getEdges()) {
+ if (edge.getTargetVertexId().equals(targetVertexId)) {
+ return edge.getValue();
+ }
+ }
+ return null;
+ }
+
+ /**
+ * Get the number of outgoing edges on this vertex.
+ *
+ * @return the total number of outbound edges from this vertex
+ */
+ public int getNumEdges() {
+ return Iterables.size(getEdges());
+ }
+
+ /**
+ * Send a message to a vertex id. The message should not be mutated after
+ * this method returns or else undefined results could occur.
+ *
+ * @param id Vertex id to send the message to
+ * @param message Message data to send. Note that after the message is sent,
+ * the user should not modify the object.
+ */
+ public void sendMessage(I id, M message) {
+ if (message == null) {
+ throw new IllegalArgumentException(
+ "sendMessage: Cannot send null message to " + id);
+ }
+ if (graphState.getWorkerClientRequestProcessor().
+ sendMessageRequest(id, message)) {
+ graphState.getGraphMapper().notifySentMessages();
+ }
+ }
+
+ /**
+ * Lookup WorkerInfo for myself.
+ *
+ * @return WorkerInfo about worker holding this Vertex.
+ */
+ public WorkerInfo getMyWorkerInfo() {
+ return getVertexWorkerInfo(id);
+ }
+
+ /**
+ * Lookup WorkerInfo for a Vertex.
+ *
+ * @param vertexId VertexId to lookup
+ * @return WorkerInfo about worker holding this Vertex.
+ */
+ public WorkerInfo getVertexWorkerInfo(I vertexId) {
+ return getVertexPartitionOwner(vertexId).getWorkerInfo();
+ }
+
+ /**
+ * Lookup PartitionOwner for a Vertex
+ *
+ * @param vertexId id of Vertex to look up.
+ * @return PartitionOwner holding Vertex
+ */
+ private PartitionOwner getVertexPartitionOwner(I vertexId) {
+ return getGraphState().getWorkerClientRequestProcessor().
+ getVertexPartitionOwner(vertexId);
+ }
+
+ /**
+ * Send a message to all edges.
+ *
+ * @param message Message sent to all edges.
+ */
+ public void sendMessageToAllEdges(M message) {
+ for (Edge<I, E> edge : getEdges()) {
+ sendMessage(edge.getTargetVertexId(), message);
+ }
+ }
+
+ /**
+ * After this is called, the compute() code will no longer be called for
+ * this vertex unless a message is sent to it. Then the compute() code
+ * will be called once again until this function is called. The
+ * application finishes only when all vertices vote to halt.
+ */
+ public void voteToHalt() {
+ halt = true;
+ }
+
+ /**
+ * Re-activate vertex if halted.
+ */
+ public void wakeUp() {
+ halt = false;
+ }
+
+ /**
+ * Is this vertex done?
+ *
+ * @return True if halted, false otherwise.
+ */
+ public boolean isHalted() {
+ return halt;
+ }
+
+ /**
+ * Get the graph state for all workers.
+ *
+ * @return Graph state for all workers
+ */
+ GraphState<I, V, E, M> getGraphState() {
+ return graphState;
+ }
+
+ /**
+ * Set the graph state for all workers
+ *
+ * @param graphState Graph state for all workers
+ */
+ void setGraphState(GraphState<I, V, E, M> graphState) {
+ this.graphState = graphState;
+ }
+
+ /**
+ * Get the mapper context
+ *
+ * @return Mapper context
+ */
+ public Mapper.Context getContext() {
+ return getGraphState().getContext();
+ }
+
+ /**
+ * Get the worker context
+ *
+ * @return WorkerContext context
+ */
+ public WorkerContext getWorkerContext() {
+ return getGraphState().getGraphMapper().getWorkerContext();
+ }
+
+ @Override
+ public <A extends Writable> void aggregate(String name, A value) {
+ getGraphState().getWorkerAggregatorUsage().
+ aggregate(name, value);
+ }
+
+ @Override
+ public <A extends Writable> A getAggregatedValue(String name) {
+ return getGraphState().getWorkerAggregatorUsage().
+ <A>getAggregatedValue(name);
+ }
+
+ @Override
+ public void readFields(DataInput in) throws IOException {
+ I vertexId = (I) getConf().createVertexId();
+ vertexId.readFields(in);
+ V vertexValue = (V) getConf().createVertexValue();
+ vertexValue.readFields(in);
+
+ int numEdges = in.readInt();
+ List<Edge<I, E>> edges = Lists.newArrayListWithCapacity(numEdges);
+ for (int i = 0; i < numEdges; ++i) {
+ I targetVertexId = (I) getConf().createVertexId();
+ targetVertexId.readFields(in);
+ E edgeValue = (E) getConf().createEdgeValue();
+ edgeValue.readFields(in);
+ edges.add(new Edge<I, E>(targetVertexId, edgeValue));
+ }
+
+ initialize(vertexId, vertexValue, edges);
+
+ readHaltBoolean(in);
+ }
+
+ /**
+ * Helper method for subclasses which implement their own readFields() to use.
+ *
+ * @param in DataInput to read from.
+ * @throws IOException If anything goes wrong during read.
+ */
+ protected void readHaltBoolean(DataInput in) throws IOException {
+ halt = in.readBoolean();
+ }
+
+ @Override
+ public void write(DataOutput out) throws IOException {
+ getId().write(out);
+ getValue().write(out);
+
+ out.writeInt(getNumEdges());
+ for (Edge<I, E> edge : getEdges()) {
+ edge.getTargetVertexId().write(out);
+ edge.getValue().write(out);
+ }
+
+ out.writeBoolean(halt);
+ }
+
+ @Override
+ public ImmutableClassesGiraphConfiguration<I, V, E, M> getConf() {
+ return conf;
+ }
+
+ @Override
+ public void setConf(ImmutableClassesGiraphConfiguration<I, V, E, M> conf) {
+ this.conf = conf;
+ }
+
+ @Override
+ public String toString() {
+ return "Vertex(id=" + getId() + ",value=" + getValue() +
+ ",#edges=" + getNumEdges() + ")";
+ }
+}
http://git-wip-us.apache.org/repos/asf/giraph/blob/45851391/giraph-core/src/main/java/org/apache/giraph/graph/VertexChanges.java
----------------------------------------------------------------------
diff --git a/giraph-core/src/main/java/org/apache/giraph/graph/VertexChanges.java b/giraph-core/src/main/java/org/apache/giraph/graph/VertexChanges.java
new file mode 100644
index 0000000..957cf82
--- /dev/null
+++ b/giraph-core/src/main/java/org/apache/giraph/graph/VertexChanges.java
@@ -0,0 +1,68 @@
+/*
+ * 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.graph;
+
+import java.util.List;
+
+import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.io.WritableComparable;
+
+/**
+ * Structure to hold all the possible graph mutations that can occur during a
+ * superstep.
+ *
+ * @param <I> Vertex index value
+ * @param <V> Vertex value
+ * @param <E> Edge value
+ * @param <M> Message value
+ */
+@SuppressWarnings("rawtypes")
+public interface VertexChanges<I extends WritableComparable,
+ V extends Writable, E extends Writable, M extends Writable> {
+ /**
+ * Get the added vertices for this particular vertex index from the previous
+ * superstep.
+ *
+ * @return List of vertices for this vertex index.
+ */
+ List<Vertex<I, V, E, M>> getAddedVertexList();
+
+ /**
+ * Get the number of times this vertex was removed in the previous
+ * superstep.
+ *
+ * @return Count of time this vertex was removed in the previous superstep
+ */
+ int getRemovedVertexCount();
+
+ /**
+ * Get the added edges for this particular vertex index from the previous
+ * superstep
+ *
+ * @return List of added edges for this vertex index
+ */
+ List<Edge<I, E>> getAddedEdgeList();
+
+ /**
+ * Get the removed edges by their destination vertex index.
+ *
+ * @return List of destination edges for removal from this vertex index
+ */
+ List<I> getRemovedEdgeList();
+}
http://git-wip-us.apache.org/repos/asf/giraph/blob/45851391/giraph-core/src/main/java/org/apache/giraph/graph/VertexEdgeCount.java
----------------------------------------------------------------------
diff --git a/giraph-core/src/main/java/org/apache/giraph/graph/VertexEdgeCount.java b/giraph-core/src/main/java/org/apache/giraph/graph/VertexEdgeCount.java
new file mode 100644
index 0000000..c2d13cc
--- /dev/null
+++ b/giraph-core/src/main/java/org/apache/giraph/graph/VertexEdgeCount.java
@@ -0,0 +1,88 @@
+/*
+ * 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.graph;
+
+/**
+ * Simple immutable structure for storing a final vertex and edge count.
+ */
+public class VertexEdgeCount {
+ /** Immutable vertices */
+ private final long vertexCount;
+ /** Immutable edges */
+ private final long edgeCount;
+
+ /**
+ * Default constructor.
+ */
+ public VertexEdgeCount() {
+ vertexCount = 0;
+ edgeCount = 0;
+ }
+
+ /**
+ * Constructor with initial values.
+ *
+ * @param vertexCount Final number of vertices.
+ * @param edgeCount Final number of edges.
+ */
+ public VertexEdgeCount(long vertexCount, long edgeCount) {
+ this.vertexCount = vertexCount;
+ this.edgeCount = edgeCount;
+ }
+
+ public long getVertexCount() {
+ return vertexCount;
+ }
+
+ public long getEdgeCount() {
+ return edgeCount;
+ }
+
+ /**
+ * Increment the both the vertex edge count with a {@link VertexEdgeCount}.
+ *
+ * @param vertexEdgeCount add both the vertices and edges of this object.
+ * @return New immutable object with the new vertex and edge counts.
+ */
+ public VertexEdgeCount incrVertexEdgeCount(
+ VertexEdgeCount vertexEdgeCount) {
+ return new VertexEdgeCount(
+ vertexCount + vertexEdgeCount.getVertexCount(),
+ edgeCount + vertexEdgeCount.getEdgeCount());
+ }
+
+ /**
+ * Increment the both the vertex edge count with primitives.
+ *
+ * @param vertexCount Add this many vertices.
+ * @param edgeCount Add this many edges.
+ * @return New immutable object with the new vertex and edge counts.
+ */
+ public VertexEdgeCount incrVertexEdgeCount(
+ long vertexCount, long edgeCount) {
+ return new VertexEdgeCount(
+ this.vertexCount + vertexCount,
+ this.edgeCount + edgeCount);
+ }
+
+ @Override
+ public String toString() {
+ return "(v=" + getVertexCount() + ", e=" + getEdgeCount() + ")";
+ }
+}
http://git-wip-us.apache.org/repos/asf/giraph/blob/45851391/giraph-core/src/main/java/org/apache/giraph/graph/VertexInputFormat.java
----------------------------------------------------------------------
diff --git a/giraph-core/src/main/java/org/apache/giraph/graph/VertexInputFormat.java b/giraph-core/src/main/java/org/apache/giraph/graph/VertexInputFormat.java
new file mode 100644
index 0000000..9824b69
--- /dev/null
+++ b/giraph-core/src/main/java/org/apache/giraph/graph/VertexInputFormat.java
@@ -0,0 +1,82 @@
+/*
+ * 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.graph;
+
+import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.io.WritableComparable;
+import org.apache.hadoop.mapreduce.InputSplit;
+import org.apache.hadoop.mapreduce.JobContext;
+import org.apache.hadoop.mapreduce.TaskAttemptContext;
+
+import java.io.IOException;
+import java.util.List;
+
+/**
+ * Use this to load data for a BSP application. Note that the InputSplit must
+ * also implement Writable. The InputSplits will determine the partitioning of
+ * vertices across the mappers, so keep that in consideration when implementing
+ * getSplits().
+ *
+ * @param <I> Vertex id
+ * @param <V> Vertex value
+ * @param <E> Edge value
+ * @param <M> Message data
+ */
+@SuppressWarnings("rawtypes")
+public abstract class VertexInputFormat<I extends WritableComparable,
+ V extends Writable, E extends Writable, M extends Writable>
+ implements GiraphInputFormat {
+ /**
+ * Logically split the vertices for a graph processing application.
+ *
+ * Each {@link InputSplit} is then assigned to a worker for processing.
+ *
+ * <p><i>Note</i>: The split is a <i>logical</i> split of the inputs and the
+ * input files are not physically split into chunks. For e.g. a split could
+ * be <i><input-file-path, start, offset></i> tuple. The InputFormat
+ * also creates the {@link VertexReader} to read the {@link InputSplit}.
+ *
+ * Also, the number of workers is a hint given to the developer to try to
+ * intelligently determine how many splits to create (if this is
+ * adjustable) at runtime.
+ *
+ * @param context Context of the job
+ * @param numWorkers Number of workers used for this job
+ * @return an array of {@link InputSplit}s for the job.
+ */
+ @Override
+ public abstract List<InputSplit> getSplits(
+ JobContext context, int numWorkers)
+ throws IOException, InterruptedException;
+
+ /**
+ * Create a vertex reader for a given split. The framework will call
+ * {@link VertexReader#initialize(InputSplit, TaskAttemptContext)} before
+ * the split is used.
+ *
+ * @param split the split to be read
+ * @param context the information about the task
+ * @return a new record reader
+ * @throws IOException
+ * @throws InterruptedException
+ */
+ public abstract VertexReader<I, V, E, M> createVertexReader(
+ InputSplit split,
+ TaskAttemptContext context) throws IOException;
+}
http://git-wip-us.apache.org/repos/asf/giraph/blob/45851391/giraph-core/src/main/java/org/apache/giraph/graph/VertexInputSplitsCallable.java
----------------------------------------------------------------------
diff --git a/giraph-core/src/main/java/org/apache/giraph/graph/VertexInputSplitsCallable.java b/giraph-core/src/main/java/org/apache/giraph/graph/VertexInputSplitsCallable.java
new file mode 100644
index 0000000..e1e6235
--- /dev/null
+++ b/giraph-core/src/main/java/org/apache/giraph/graph/VertexInputSplitsCallable.java
@@ -0,0 +1,180 @@
+/*
+ * 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.graph;
+
+import org.apache.giraph.conf.ImmutableClassesGiraphConfiguration;
+import org.apache.giraph.graph.partition.PartitionOwner;
+import org.apache.giraph.metrics.GiraphMetrics;
+import org.apache.giraph.metrics.GiraphMetricsRegistry;
+import org.apache.giraph.utils.LoggerUtils;
+import org.apache.giraph.utils.MemoryUtils;
+import org.apache.giraph.zk.ZooKeeperExt;
+import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.io.WritableComparable;
+import org.apache.hadoop.mapreduce.InputSplit;
+import org.apache.hadoop.mapreduce.Mapper;
+import org.apache.log4j.Level;
+import org.apache.log4j.Logger;
+
+import com.yammer.metrics.core.Counter;
+
+import java.io.IOException;
+import java.util.List;
+
+/**
+ * Load as many vertex input splits as possible.
+ * Every thread will has its own instance of WorkerClientRequestProcessor
+ * to send requests.
+ *
+ * @param <I> Vertex index value
+ * @param <V> Vertex value
+ * @param <E> Edge value
+ * @param <M> Message data
+ */
+public class VertexInputSplitsCallable<I extends WritableComparable,
+ V extends Writable, E extends Writable, M extends Writable>
+ extends InputSplitsCallable<I, V, E, M> {
+ /** Class logger */
+ private static final Logger LOG =
+ Logger.getLogger(VertexInputSplitsCallable.class);
+ /** Total vertices loaded */
+ private long totalVerticesLoaded = 0;
+ /** Total edges loaded */
+ private long totalEdgesLoaded = 0;
+ /** Input split max vertices (-1 denotes all) */
+ private final long inputSplitMaxVertices;
+ /** Bsp service worker (only use thread-safe methods) */
+ private final BspServiceWorker<I, V, E, M> bspServiceWorker;
+
+ // Metrics
+ /** number of vertices loaded counter */
+ private final Counter verticesLoadedCounter;
+ /** number of edges loaded counter */
+ private final Counter edgesLoadedCounter;
+
+ /**
+ * Constructor.
+ *
+ * @param context Context
+ * @param graphState Graph state
+ * @param configuration Configuration
+ * @param bspServiceWorker service worker
+ * @param inputSplitPathList List of the paths of the input splits
+ * @param workerInfo This worker's info
+ * @param zooKeeperExt Handle to ZooKeeperExt
+ */
+ public VertexInputSplitsCallable(
+ Mapper<?, ?, ?, ?>.Context context,
+ GraphState<I, V, E, M> graphState,
+ ImmutableClassesGiraphConfiguration<I, V, E, M> configuration,
+ BspServiceWorker<I, V, E, M> bspServiceWorker,
+ List<String> inputSplitPathList,
+ WorkerInfo workerInfo,
+ ZooKeeperExt zooKeeperExt) {
+ super(context, graphState, configuration, bspServiceWorker,
+ inputSplitPathList, workerInfo, zooKeeperExt,
+ BspServiceWorker.VERTEX_INPUT_SPLIT_RESERVED_NODE,
+ BspServiceWorker.VERTEX_INPUT_SPLIT_FINISHED_NODE,
+ bspServiceWorker.vertexInputSplitsEvents);
+
+ inputSplitMaxVertices = configuration.getInputSplitMaxVertices();
+ this.bspServiceWorker = bspServiceWorker;
+
+ // Initialize Metrics
+ GiraphMetricsRegistry jobMetrics = GiraphMetrics.get().perJob();
+ verticesLoadedCounter = jobMetrics.getCounter(COUNTER_VERTICES_LOADED);
+ edgesLoadedCounter = jobMetrics.getCounter(COUNTER_EDGES_LOADED);
+ }
+
+ /**
+ * Read vertices from input split. If testing, the user may request a
+ * maximum number of vertices to be read from an input split.
+ *
+ * @param inputSplit Input split to process with vertex reader
+ * @param graphState Current graph state
+ * @return Vertices and edges loaded from this input split
+ * @throws IOException
+ * @throws InterruptedException
+ */
+ @Override
+ protected VertexEdgeCount readInputSplit(
+ InputSplit inputSplit,
+ GraphState<I, V, E, M> graphState)
+ throws IOException, InterruptedException {
+ VertexInputFormat<I, V, E, M> vertexInputFormat =
+ configuration.createVertexInputFormat();
+ VertexReader<I, V, E, M> vertexReader =
+ vertexInputFormat.createVertexReader(inputSplit, context);
+ vertexReader.initialize(inputSplit, context);
+ long inputSplitVerticesLoaded = 0;
+ long inputSplitEdgesLoaded = 0;
+ while (vertexReader.nextVertex()) {
+ Vertex<I, V, E, M> readerVertex =
+ vertexReader.getCurrentVertex();
+ if (readerVertex.getId() == null) {
+ throw new IllegalArgumentException(
+ "readInputSplit: Vertex reader returned a vertex " +
+ "without an id! - " + readerVertex);
+ }
+ if (readerVertex.getValue() == null) {
+ readerVertex.setValue(configuration.createVertexValue());
+ }
+ readerVertex.setConf(configuration);
+ readerVertex.setGraphState(graphState);
+
+ PartitionOwner partitionOwner =
+ bspServiceWorker.getVertexPartitionOwner(readerVertex.getId());
+ graphState.getWorkerClientRequestProcessor().sendVertexRequest(
+ partitionOwner, readerVertex);
+ context.progress(); // do this before potential data transfer
+ ++inputSplitVerticesLoaded;
+ inputSplitEdgesLoaded += readerVertex.getNumEdges();
+
+ // Update status every 250k vertices
+ if (((inputSplitVerticesLoaded + totalVerticesLoaded) % 250000) == 0) {
+ LoggerUtils.setStatusAndLog(context, LOG, Level.INFO,
+ "readInputSplit: Loaded " +
+ (inputSplitVerticesLoaded + totalVerticesLoaded) +
+ " vertices " +
+ (inputSplitEdgesLoaded + totalEdgesLoaded) + " edges " +
+ MemoryUtils.getRuntimeMemoryStats());
+ }
+
+ // For sampling, or to limit outlier input splits, the number of
+ // records per input split can be limited
+ if (inputSplitMaxVertices > 0 &&
+ inputSplitVerticesLoaded >= inputSplitMaxVertices) {
+ if (LOG.isInfoEnabled()) {
+ LOG.info("readInputSplit: Leaving the input " +
+ "split early, reached maximum vertices " +
+ inputSplitVerticesLoaded);
+ }
+ break;
+ }
+ }
+ vertexReader.close();
+ totalVerticesLoaded += inputSplitVerticesLoaded;
+ verticesLoadedCounter.inc(inputSplitVerticesLoaded);
+ totalEdgesLoaded += inputSplitEdgesLoaded;
+ edgesLoadedCounter.inc(inputSplitEdgesLoaded);
+ return new VertexEdgeCount(
+ inputSplitVerticesLoaded, inputSplitEdgesLoaded);
+ }
+}
+
http://git-wip-us.apache.org/repos/asf/giraph/blob/45851391/giraph-core/src/main/java/org/apache/giraph/graph/VertexInputSplitsCallableFactory.java
----------------------------------------------------------------------
diff --git a/giraph-core/src/main/java/org/apache/giraph/graph/VertexInputSplitsCallableFactory.java b/giraph-core/src/main/java/org/apache/giraph/graph/VertexInputSplitsCallableFactory.java
new file mode 100644
index 0000000..a55c4b0
--- /dev/null
+++ b/giraph-core/src/main/java/org/apache/giraph/graph/VertexInputSplitsCallableFactory.java
@@ -0,0 +1,94 @@
+/*
+ * 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.graph;
+
+import org.apache.giraph.conf.ImmutableClassesGiraphConfiguration;
+import org.apache.giraph.zk.ZooKeeperExt;
+import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.io.WritableComparable;
+import org.apache.hadoop.mapreduce.Mapper;
+
+import java.util.List;
+
+/**
+ * Factory for {@link VertexInputSplitsCallable}s.
+ *
+ * @param <I> Vertex id
+ * @param <V> Vertex value
+ * @param <E> Edge value
+ * @param <M> Message data
+ */
+public class VertexInputSplitsCallableFactory<I extends WritableComparable,
+ V extends Writable, E extends Writable, M extends Writable>
+ implements InputSplitsCallableFactory<I, V, E, M> {
+ /** Mapper context. */
+ private final Mapper<?, ?, ?, ?>.Context context;
+ /** Graph state. */
+ private final GraphState<I, V, E, M> graphState;
+ /** Configuration. */
+ private final ImmutableClassesGiraphConfiguration<I, V, E, M> configuration;
+ /** {@link BspServiceWorker} we're running on. */
+ private final BspServiceWorker<I, V, E, M> bspServiceWorker;
+ /** List of input split paths. */
+ private final List<String> inputSplitPathList;
+ /** Worker info. */
+ private final WorkerInfo workerInfo;
+ /** {@link ZooKeeperExt} for this worker. */
+ private final ZooKeeperExt zooKeeperExt;
+
+ /**
+ * Constructor.
+ *
+ * @param context Mapper context
+ * @param graphState Graph state
+ * @param configuration Configuration
+ * @param bspServiceWorker Calling {@link BspServiceWorker}
+ * @param inputSplitPathList List of input split paths
+ * @param workerInfo Worker info
+ * @param zooKeeperExt {@link ZooKeeperExt} for this worker
+ */
+ public VertexInputSplitsCallableFactory(
+ Mapper<?, ?, ?, ?>.Context context,
+ GraphState<I, V, E, M> graphState,
+ ImmutableClassesGiraphConfiguration<I, V, E, M> configuration,
+ BspServiceWorker<I, V, E, M> bspServiceWorker,
+ List<String> inputSplitPathList,
+ WorkerInfo workerInfo,
+ ZooKeeperExt zooKeeperExt) {
+ this.context = context;
+ this.graphState = graphState;
+ this.configuration = configuration;
+ this.bspServiceWorker = bspServiceWorker;
+ this.inputSplitPathList = inputSplitPathList;
+ this.workerInfo = workerInfo;
+ this.zooKeeperExt = zooKeeperExt;
+ }
+
+ @Override
+ public InputSplitsCallable<I, V, E, M> newCallable() {
+ return new VertexInputSplitsCallable<I, V, E, M>(
+ context,
+ graphState,
+ configuration,
+ bspServiceWorker,
+ inputSplitPathList,
+ workerInfo,
+ zooKeeperExt);
+ }
+}
http://git-wip-us.apache.org/repos/asf/giraph/blob/45851391/giraph-core/src/main/java/org/apache/giraph/graph/VertexMutations.java
----------------------------------------------------------------------
diff --git a/giraph-core/src/main/java/org/apache/giraph/graph/VertexMutations.java b/giraph-core/src/main/java/org/apache/giraph/graph/VertexMutations.java
new file mode 100644
index 0000000..4548cea
--- /dev/null
+++ b/giraph-core/src/main/java/org/apache/giraph/graph/VertexMutations.java
@@ -0,0 +1,212 @@
+/*
+ * 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.graph;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.giraph.conf.ImmutableClassesGiraphConfigurable;
+import org.apache.giraph.conf.ImmutableClassesGiraphConfiguration;
+import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.io.WritableComparable;
+import org.json.JSONException;
+import org.json.JSONObject;
+
+/**
+ * Structure to hold all the possible graph mutations that can occur during a
+ * superstep.
+ *
+ * @param <I> Vertex index value
+ * @param <V> Vertex value
+ * @param <E> Edge value
+ * @param <M> Message value
+ */
+@SuppressWarnings("rawtypes")
+public class VertexMutations<I extends WritableComparable,
+ V extends Writable, E extends Writable,
+ M extends Writable> implements VertexChanges<I, V, E, M>,
+ Writable, ImmutableClassesGiraphConfigurable {
+ /** List of added vertices during the last superstep */
+ private final List<Vertex<I, V, E, M>> addedVertexList =
+ new ArrayList<Vertex<I, V, E, M>>();
+ /** Count of remove vertex requests */
+ private int removedVertexCount = 0;
+ /** List of added edges */
+ private final List<Edge<I, E>> addedEdgeList = new ArrayList<Edge<I, E>>();
+ /** List of removed edges */
+ private final List<I> removedEdgeList = new ArrayList<I>();
+ /** Configuration */
+ private ImmutableClassesGiraphConfiguration<I, V, E, M> conf;
+
+ /**
+ * Copy the vertex mutations.
+ *
+ * @return Copied vertex mutations
+ */
+ public VertexMutations<I, V, E, M> copy() {
+ VertexMutations<I, V, E, M> copied = new VertexMutations<I, V, E, M>();
+ copied.addedVertexList.addAll(this.addedVertexList);
+ copied.removedVertexCount = this.removedVertexCount;
+ copied.addedEdgeList.addAll(this.addedEdgeList);
+ copied.removedEdgeList.addAll(this.removedEdgeList);
+ copied.conf = this.conf;
+ return copied;
+ }
+
+ @Override
+ public List<Vertex<I, V, E, M>> getAddedVertexList() {
+ return addedVertexList;
+ }
+
+ @Override
+ public void readFields(DataInput input) throws IOException {
+ addedVertexList.clear();
+ addedEdgeList.clear();
+ removedEdgeList.clear();
+
+ int addedVertexListSize = input.readInt();
+ for (int i = 0; i < addedVertexListSize; ++i) {
+ Vertex<I, V, E, M> vertex = conf.createVertex();
+ vertex.readFields(input);
+ addedVertexList.add(vertex);
+ }
+ removedVertexCount = input.readInt();
+ int addedEdgeListSize = input.readInt();
+ for (int i = 0; i < addedEdgeListSize; ++i) {
+ I destVertex = conf.createVertexId();
+ destVertex.readFields(input);
+ E edgeValue = conf.createEdgeValue();
+ edgeValue.readFields(input);
+ addedEdgeList.add(new Edge<I, E>(destVertex, edgeValue));
+ }
+ int removedEdgeListSize = input.readInt();
+ for (int i = 0; i < removedEdgeListSize; ++i) {
+ I removedEdge = conf.createVertexId();
+ removedEdge.readFields(input);
+ removedEdgeList.add(removedEdge);
+ }
+ }
+
+ @Override
+ public void write(DataOutput output) throws IOException {
+ output.writeInt(addedVertexList.size());
+ for (Vertex<I, V, E, M> vertex : addedVertexList) {
+ vertex.write(output);
+ }
+ output.writeInt(removedVertexCount);
+ output.writeInt(addedEdgeList.size());
+ for (Edge<I, E> edge : addedEdgeList) {
+ edge.getTargetVertexId().write(output);
+ edge.getValue().write(output);
+ }
+ output.writeInt(removedEdgeList.size());
+ for (I removedEdge : removedEdgeList) {
+ removedEdge.write(output);
+ }
+ }
+
+ /**
+ * Add a vertex mutation
+ *
+ * @param vertex Vertex to be added
+ */
+ public void addVertex(Vertex<I, V, E, M> vertex) {
+ addedVertexList.add(vertex);
+ }
+
+ @Override
+ public int getRemovedVertexCount() {
+ return removedVertexCount;
+ }
+
+ /**
+ * Removed a vertex mutation (increments a count)
+ */
+ public void removeVertex() {
+ ++removedVertexCount;
+ }
+
+ @Override
+ public List<Edge<I, E>> getAddedEdgeList() {
+ return addedEdgeList;
+ }
+
+ /**
+ * Add an edge to this vertex
+ *
+ * @param edge Edge to be added
+ */
+ public void addEdge(Edge<I, E> edge) {
+ addedEdgeList.add(edge);
+ }
+
+ @Override
+ public List<I> getRemovedEdgeList() {
+ return removedEdgeList;
+ }
+
+ /**
+ * Remove an edge on this vertex
+ *
+ * @param destinationVertexId Vertex index of the destination of the edge
+ */
+ public void removeEdge(I destinationVertexId) {
+ removedEdgeList.add(destinationVertexId);
+ }
+
+ /**
+ * Add one vertex mutations to another
+ *
+ * @param vertexMutations Object to be added
+ */
+ public void addVertexMutations(VertexMutations<I, V, E, M> vertexMutations) {
+ addedVertexList.addAll(vertexMutations.getAddedVertexList());
+ removedVertexCount += vertexMutations.getRemovedVertexCount();
+ addedEdgeList.addAll(vertexMutations.getAddedEdgeList());
+ removedEdgeList.addAll(vertexMutations.getRemovedEdgeList());
+ }
+
+ @Override
+ public String toString() {
+ JSONObject jsonObject = new JSONObject();
+ try {
+ jsonObject.put("added vertices", getAddedVertexList().toString());
+ jsonObject.put("added edges", getAddedEdgeList().toString());
+ jsonObject.put("removed vertex count", getRemovedVertexCount());
+ jsonObject.put("removed edges", getRemovedEdgeList().toString());
+ return jsonObject.toString();
+ } catch (JSONException e) {
+ throw new IllegalStateException("toString: Got a JSON exception",
+ e);
+ }
+ }
+
+ @Override
+ public ImmutableClassesGiraphConfiguration getConf() {
+ return conf;
+ }
+
+ @Override
+ public void setConf(ImmutableClassesGiraphConfiguration conf) {
+ this.conf = conf;
+ }
+}
http://git-wip-us.apache.org/repos/asf/giraph/blob/45851391/giraph-core/src/main/java/org/apache/giraph/graph/VertexOutputFormat.java
----------------------------------------------------------------------
diff --git a/giraph-core/src/main/java/org/apache/giraph/graph/VertexOutputFormat.java b/giraph-core/src/main/java/org/apache/giraph/graph/VertexOutputFormat.java
new file mode 100644
index 0000000..d4c8e1c
--- /dev/null
+++ b/giraph-core/src/main/java/org/apache/giraph/graph/VertexOutputFormat.java
@@ -0,0 +1,81 @@
+/*
+ * 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.graph;
+
+import java.io.IOException;
+
+import org.apache.hadoop.mapreduce.JobContext;
+import org.apache.hadoop.mapreduce.OutputCommitter;
+
+import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.io.WritableComparable;
+import org.apache.hadoop.mapreduce.TaskAttemptContext;
+
+/**
+ * Implement to output the graph after the computation. It is modeled
+ * directly after the Hadoop OutputFormat.
+ *
+ * @param <I> Vertex index value
+ * @param <V> Vertex value
+ * @param <E> Edge value
+ */
+@SuppressWarnings("rawtypes")
+public abstract class VertexOutputFormat<
+ I extends WritableComparable, V extends Writable, E extends Writable> {
+ /**
+ * Create a vertex writer for a given split. The framework will call
+ * {@link VertexReader#initialize(InputSplit, TaskAttemptContext)} before
+ * the split is used.
+ *
+ * @param context the information about the task
+ * @return a new vertex writer
+ * @throws IOException
+ * @throws InterruptedException
+ */
+ public abstract VertexWriter<I, V, E> createVertexWriter(
+ TaskAttemptContext context) throws IOException, InterruptedException;
+
+ /**
+ * Check for validity of the output-specification for the job.
+ * (Copied from Hadoop OutputFormat)
+ *
+ * <p>This is to validate the output specification for the job when it is
+ * a job is submitted. Typically checks that it does not already exist,
+ * throwing an exception when it already exists, so that output is not
+ * overwritten.</p>
+ *
+ * @param context information about the job
+ * @throws IOException when output should not be attempted
+ */
+ public abstract void checkOutputSpecs(JobContext context)
+ throws IOException, InterruptedException;
+
+ /**
+ * Get the output committer for this output format. This is responsible
+ * for ensuring the output is committed correctly.
+ * (Copied from Hadoop OutputFormat)
+ *
+ * @param context the task context
+ * @return an output committer
+ * @throws IOException
+ * @throws InterruptedException
+ */
+ public abstract OutputCommitter getOutputCommitter(
+ TaskAttemptContext context) throws IOException, InterruptedException;
+}
http://git-wip-us.apache.org/repos/asf/giraph/blob/45851391/giraph-core/src/main/java/org/apache/giraph/graph/VertexReader.java
----------------------------------------------------------------------
diff --git a/giraph-core/src/main/java/org/apache/giraph/graph/VertexReader.java b/giraph-core/src/main/java/org/apache/giraph/graph/VertexReader.java
new file mode 100644
index 0000000..77801cc
--- /dev/null
+++ b/giraph-core/src/main/java/org/apache/giraph/graph/VertexReader.java
@@ -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.graph;
+
+import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.io.WritableComparable;
+import org.apache.hadoop.mapreduce.InputSplit;
+import org.apache.hadoop.mapreduce.TaskAttemptContext;
+
+import java.io.IOException;
+
+/**
+ * Analogous to {@link RecordReader} for vertices. Will read the vertices
+ * from an input split.
+ *
+ * @param <I> Vertex id
+ * @param <V> Vertex data
+ * @param <E> Edge data
+ * @param <M> Message data
+ */
+@SuppressWarnings("rawtypes")
+public interface VertexReader<I extends WritableComparable,
+ V extends Writable, E extends Writable, M extends Writable> {
+ /**
+ * Use the input split and context to setup reading the vertices.
+ * Guaranteed to be called prior to any other function.
+ *
+ * @param inputSplit Input split to be used for reading vertices.
+ * @param context Context from the task.
+ * @throws IOException
+ * @throws InterruptedException
+ */
+ void initialize(InputSplit inputSplit, TaskAttemptContext context)
+ throws IOException, InterruptedException;
+
+ /**
+ *
+ * @return false iff there are no more vertices
+ * @throws IOException
+ * @throws InterruptedException
+ */
+ boolean nextVertex() throws IOException, InterruptedException;
+
+ /**
+ * Get the current vertex.
+ *
+ * @return the current vertex which has been read.
+ * nextVertex() should be called first.
+ * @throws IOException
+ * @throws InterruptedException
+ */
+ Vertex<I, V, E, M> getCurrentVertex()
+ throws IOException, InterruptedException;
+
+ /**
+ * Close this {@link VertexReader} to future operations.
+ *
+ * @throws IOException
+ */
+ void close() throws IOException;
+
+ /**
+ * How much of the input has the {@link VertexReader} consumed i.e.
+ * has been processed by?
+ *
+ * @return Progress from <code>0.0</code> to <code>1.0</code>.
+ * @throws IOException
+ * @throws InterruptedException
+ */
+ float getProgress() throws IOException, InterruptedException;
+}
http://git-wip-us.apache.org/repos/asf/giraph/blob/45851391/giraph-core/src/main/java/org/apache/giraph/graph/VertexResolver.java
----------------------------------------------------------------------
diff --git a/giraph-core/src/main/java/org/apache/giraph/graph/VertexResolver.java b/giraph-core/src/main/java/org/apache/giraph/graph/VertexResolver.java
new file mode 100644
index 0000000..1616290
--- /dev/null
+++ b/giraph-core/src/main/java/org/apache/giraph/graph/VertexResolver.java
@@ -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.giraph.graph;
+
+import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.io.WritableComparable;
+
+/**
+ * Handles all the situations that can arise upon creation/removal of
+ * vertices and edges.
+ *
+ * @param <I> Vertex id
+ * @param <V> Vertex data
+ * @param <E> Edge data
+ * @param <M> Message data
+ */
+@SuppressWarnings("rawtypes")
+public interface VertexResolver<I extends WritableComparable,
+ V extends Writable, E extends Writable, M extends Writable>
+ extends GraphStateAware<I, V, E, M> {
+ /**
+ * A vertex may have been removed, created zero or more times and had
+ * zero or more messages sent to it. This method will handle all situations
+ * excluding the normal case (a vertex already exists and has zero or more
+ * messages sent it to).
+ *
+ * @param vertexId Vertex id (can be used for {@link Vertex}'s
+ * initialize())
+ * @param vertex Original vertex or null if none
+ * @param vertexChanges Changes that happened to this vertex or null if none
+ * @param hasMessages True iff vertex received messages in the last superstep
+ * @return Vertex to be returned, if null, and a vertex currently exists
+ * it will be removed
+ */
+ Vertex<I, V, E, M> resolve(I vertexId,
+ Vertex<I, V, E, M> vertex,
+ VertexChanges<I, V, E, M> vertexChanges,
+ boolean hasMessages);
+
+ /**
+ * Set the graph state.
+ *
+ * @param graphState Graph state saved.
+ */
+ void setGraphState(GraphState<I, V, E, M> graphState);
+}
http://git-wip-us.apache.org/repos/asf/giraph/blob/45851391/giraph-core/src/main/java/org/apache/giraph/graph/VertexValueInputFormat.java
----------------------------------------------------------------------
diff --git a/giraph-core/src/main/java/org/apache/giraph/graph/VertexValueInputFormat.java b/giraph-core/src/main/java/org/apache/giraph/graph/VertexValueInputFormat.java
new file mode 100644
index 0000000..804d23e
--- /dev/null
+++ b/giraph-core/src/main/java/org/apache/giraph/graph/VertexValueInputFormat.java
@@ -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.giraph.graph;
+
+import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.io.WritableComparable;
+import org.apache.hadoop.mapreduce.InputSplit;
+import org.apache.hadoop.mapreduce.TaskAttemptContext;
+
+import java.io.IOException;
+
+/**
+ * Vertex input format that only allows setting vertex id and value. It can
+ * be used in conjunction with {@link EdgeInputFormat}.
+ *
+ * @param <I> Vertex id
+ * @param <V> Vertex value
+ * @param <E> Edge value
+ * @param <M> Message value
+ */
+public abstract class VertexValueInputFormat<I extends WritableComparable,
+ V extends Writable, E extends Writable, M extends Writable>
+ extends VertexInputFormat<I, V, E, M> {
+ /**
+ * Create a {@link VertexValueReader} for a given split. The framework will
+ * call {@link VertexValueReader#initialize(InputSplit,
+ * TaskAttemptContext)} before the split is used.
+ *
+ * @param split The split to be read
+ * @param context The information about the task
+ * @return A new vertex value reader
+ * @throws IOException
+ */
+ public abstract VertexValueReader<I, V, E, M> createVertexValueReader(
+ InputSplit split, TaskAttemptContext context) throws IOException;
+
+ @Override
+ public final VertexReader<I, V, E, M> createVertexReader(
+ InputSplit split, TaskAttemptContext context) throws IOException {
+ return createVertexValueReader(split, context);
+ }
+}
http://git-wip-us.apache.org/repos/asf/giraph/blob/45851391/giraph-core/src/main/java/org/apache/giraph/graph/VertexValueReader.java
----------------------------------------------------------------------
diff --git a/giraph-core/src/main/java/org/apache/giraph/graph/VertexValueReader.java b/giraph-core/src/main/java/org/apache/giraph/graph/VertexValueReader.java
new file mode 100644
index 0000000..569714d
--- /dev/null
+++ b/giraph-core/src/main/java/org/apache/giraph/graph/VertexValueReader.java
@@ -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.graph;
+
+import org.apache.giraph.conf.ImmutableClassesGiraphConfiguration;
+import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.io.WritableComparable;
+import org.apache.hadoop.mapreduce.InputSplit;
+import org.apache.hadoop.mapreduce.TaskAttemptContext;
+
+import java.io.IOException;
+
+/**
+ * Vertex reader for {@link VertexValueInputFormat}.
+ *
+ * @param <I> Vertex id
+ * @param <V> Vertex value
+ * @param <E> Edge value
+ * @param <M> Message value
+ */
+public abstract class VertexValueReader<I extends WritableComparable,
+ V extends Writable, E extends Writable, M extends Writable>
+ extends BasicVertexValueReader<I, V, E, M> {
+ /** Configuration. */
+ private ImmutableClassesGiraphConfiguration<I, V, E, M> configuration;
+
+ @Override
+ public void initialize(InputSplit inputSplit, TaskAttemptContext context)
+ throws IOException, InterruptedException {
+ configuration = new ImmutableClassesGiraphConfiguration<I, V, E, M>(
+ context.getConfiguration());
+ }
+
+ @Override
+ public final Vertex<I, V, E, M> getCurrentVertex() throws IOException,
+ InterruptedException {
+ Vertex<I, V, E, M> vertex = getConf().createVertex();
+ vertex.initialize(getCurrentVertexId(), getCurrentVertexValue());
+ return vertex;
+ }
+
+ public ImmutableClassesGiraphConfiguration<I, V, E, M> getConf() {
+ return configuration;
+ }
+}
http://git-wip-us.apache.org/repos/asf/giraph/blob/45851391/giraph-core/src/main/java/org/apache/giraph/graph/VertexWriter.java
----------------------------------------------------------------------
diff --git a/giraph-core/src/main/java/org/apache/giraph/graph/VertexWriter.java b/giraph-core/src/main/java/org/apache/giraph/graph/VertexWriter.java
new file mode 100644
index 0000000..7c0e06e
--- /dev/null
+++ b/giraph-core/src/main/java/org/apache/giraph/graph/VertexWriter.java
@@ -0,0 +1,67 @@
+/*
+ * 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.graph;
+
+import java.io.IOException;
+
+import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.io.WritableComparable;
+import org.apache.hadoop.mapreduce.TaskAttemptContext;
+
+/**
+ * Implement to output a vertex range of the graph after the computation
+ *
+ * @param <I> Vertex id
+ * @param <V> Vertex value
+ * @param <E> Edge value
+ */
+@SuppressWarnings("rawtypes")
+public interface VertexWriter<I extends WritableComparable, V extends Writable,
+ E extends Writable> {
+ /**
+ * Use the context to setup writing the vertices.
+ * Guaranteed to be called prior to any other function.
+ *
+ * @param context Context used to write the vertices.
+ * @throws IOException
+ * @throws InterruptedException
+ */
+ void initialize(TaskAttemptContext context) throws IOException,
+ InterruptedException;
+
+ /**
+ * Writes the next vertex and associated data
+ *
+ * @param vertex set the properties of this vertex
+ * @throws IOException
+ * @throws InterruptedException
+ */
+ void writeVertex(Vertex<I, V, E, ?> vertex)
+ throws IOException, InterruptedException;
+
+ /**
+ * Close this {@link VertexWriter} to future operations.
+ *
+ * @param context the context of the task
+ * @throws IOException
+ * @throws InterruptedException
+ */
+ void close(TaskAttemptContext context)
+ throws IOException, InterruptedException;
+}
http://git-wip-us.apache.org/repos/asf/giraph/blob/45851391/giraph-core/src/main/java/org/apache/giraph/graph/WorkerAggregatorHandler.java
----------------------------------------------------------------------
diff --git a/giraph-core/src/main/java/org/apache/giraph/graph/WorkerAggregatorHandler.java b/giraph-core/src/main/java/org/apache/giraph/graph/WorkerAggregatorHandler.java
new file mode 100644
index 0000000..8e35998
--- /dev/null
+++ b/giraph-core/src/main/java/org/apache/giraph/graph/WorkerAggregatorHandler.java
@@ -0,0 +1,294 @@
+/*
+ * 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.graph;
+
+import org.apache.giraph.conf.ImmutableClassesGiraphConfiguration;
+import org.apache.giraph.bsp.CentralizedServiceWorker;
+import org.apache.giraph.comm.aggregators.WorkerAggregatorRequestProcessor;
+import org.apache.giraph.comm.aggregators.AggregatedValueOutputStream;
+import org.apache.giraph.comm.aggregators.AggregatorUtils;
+import org.apache.giraph.comm.aggregators.AllAggregatorServerData;
+import org.apache.giraph.comm.aggregators.OwnerAggregatorServerData;
+import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.util.Progressable;
+import org.apache.log4j.Logger;
+
+import com.google.common.collect.Maps;
+
+import java.io.IOException;
+import java.util.Map;
+
+/**
+ * Handler for aggregators on worker. Provides the aggregated values and
+ * performs aggregations from user vertex code (thread-safe). Also has
+ * methods for all superstep coordination related to aggregators.
+ *
+ * At the beginning of any superstep any worker calls prepareSuperstep(),
+ * which blocks until the final aggregates from the previous superstep have
+ * been delivered to the worker.
+ * Next, during the superstep worker can call aggregate() and
+ * getAggregatedValue() (both methods are thread safe) the former
+ * computes partial aggregates for this superstep from the worker,
+ * the latter returns (read-only) final aggregates from the previous superstep.
+ * Finally, at the end of the superstep, the worker calls finishSuperstep(),
+ * which propagates non-owned partial aggregates to the owner workers,
+ * and sends the final aggregate from the owner worker to the master.
+ */
+public class WorkerAggregatorHandler implements WorkerThreadAggregatorUsage {
+ /** Class logger */
+ private static final Logger LOG =
+ Logger.getLogger(WorkerAggregatorHandler.class);
+ /** Map of values from previous superstep */
+ private Map<String, Writable> previousAggregatedValueMap =
+ Maps.newHashMap();
+ /** Map of aggregators for current superstep */
+ private Map<String, Aggregator<Writable>> currentAggregatorMap =
+ Maps.newHashMap();
+ /** Service worker */
+ private final CentralizedServiceWorker<?, ?, ?, ?> serviceWorker;
+ /** Progressable for reporting progress */
+ private final Progressable progressable;
+ /** How big a single aggregator request can be */
+ private final int maxBytesPerAggregatorRequest;
+ /** Giraph configuration */
+ private final ImmutableClassesGiraphConfiguration conf;
+
+ /**
+ * Constructor
+ *
+ * @param serviceWorker Service worker
+ * @param conf Giraph configuration
+ * @param progressable Progressable for reporting progress
+ */
+ public WorkerAggregatorHandler(
+ CentralizedServiceWorker<?, ?, ?, ?> serviceWorker,
+ ImmutableClassesGiraphConfiguration conf,
+ Progressable progressable) {
+ this.serviceWorker = serviceWorker;
+ this.progressable = progressable;
+ this.conf = conf;
+ maxBytesPerAggregatorRequest = conf.getInt(
+ AggregatorUtils.MAX_BYTES_PER_AGGREGATOR_REQUEST,
+ AggregatorUtils.MAX_BYTES_PER_AGGREGATOR_REQUEST_DEFAULT);
+ }
+
+ @Override
+ public <A extends Writable> void aggregate(String name, A value) {
+ Aggregator<Writable> aggregator = currentAggregatorMap.get(name);
+ if (aggregator != null) {
+ synchronized (aggregator) {
+ aggregator.aggregate(value);
+ }
+ } else {
+ throw new IllegalStateException("aggregate: Tried to aggregate value " +
+ "to unregistered aggregator " + name);
+ }
+ }
+
+ @Override
+ public <A extends Writable> A getAggregatedValue(String name) {
+ return (A) previousAggregatedValueMap.get(name);
+ }
+
+ /**
+ * Prepare aggregators for current superstep
+ *
+ * @param requestProcessor Request processor for aggregators
+ */
+ public void prepareSuperstep(
+ WorkerAggregatorRequestProcessor requestProcessor) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("prepareSuperstep: Start preparing aggregators");
+ }
+ AllAggregatorServerData allAggregatorData =
+ serviceWorker.getServerData().getAllAggregatorData();
+ // Wait for my aggregators
+ Iterable<byte[]> dataToDistribute =
+ allAggregatorData.getDataFromMasterWhenReady();
+ try {
+ // Distribute my aggregators
+ requestProcessor.distributeAggregators(dataToDistribute);
+ } catch (IOException e) {
+ throw new IllegalStateException("prepareSuperstep: " +
+ "IOException occurred while trying to distribute aggregators", e);
+ }
+ // Wait for all other aggregators and store them
+ allAggregatorData.fillNextSuperstepMapsWhenReady(
+ serviceWorker.getWorkerInfoList().size(), previousAggregatedValueMap,
+ currentAggregatorMap);
+ allAggregatorData.reset();
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("prepareSuperstep: Aggregators prepared");
+ }
+ }
+
+ /**
+ * Send aggregators to their owners and in the end to the master
+ *
+ * @param requestProcessor Request processor for aggregators
+ */
+ public void finishSuperstep(
+ WorkerAggregatorRequestProcessor requestProcessor) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("finishSuperstep: Start finishing aggregators");
+ }
+ OwnerAggregatorServerData ownerAggregatorData =
+ serviceWorker.getServerData().getOwnerAggregatorData();
+ // First send partial aggregated values to their owners and determine
+ // which aggregators belong to this worker
+ for (Map.Entry<String, Aggregator<Writable>> entry :
+ currentAggregatorMap.entrySet()) {
+ try {
+ boolean sent = requestProcessor.sendAggregatedValue(entry.getKey(),
+ entry.getValue().getAggregatedValue());
+ if (!sent) {
+ // If it's my aggregator, add it directly
+ ownerAggregatorData.aggregate(entry.getKey(),
+ entry.getValue().getAggregatedValue());
+ }
+ } catch (IOException e) {
+ throw new IllegalStateException("finishSuperstep: " +
+ "IOException occurred while sending aggregator " +
+ entry.getKey() + " to its owner", e);
+ }
+ progressable.progress();
+ }
+ try {
+ // Flush
+ requestProcessor.flush();
+ } catch (IOException e) {
+ throw new IllegalStateException("finishSuperstep: " +
+ "IOException occurred while sending aggregators to owners", e);
+ }
+
+ // Wait to receive partial aggregated values from all other workers
+ Iterable<Map.Entry<String, Writable>> myAggregators =
+ ownerAggregatorData.getMyAggregatorValuesWhenReady(
+ serviceWorker.getWorkerInfoList().size());
+
+ // Send final aggregated values to master
+ AggregatedValueOutputStream aggregatorOutput =
+ new AggregatedValueOutputStream();
+ for (Map.Entry<String, Writable> entry : myAggregators) {
+ try {
+ int currentSize = aggregatorOutput.addAggregator(entry.getKey(),
+ entry.getValue());
+ if (currentSize > maxBytesPerAggregatorRequest) {
+ requestProcessor.sendAggregatedValuesToMaster(
+ aggregatorOutput.flush());
+ }
+ progressable.progress();
+ } catch (IOException e) {
+ throw new IllegalStateException("finishSuperstep: " +
+ "IOException occurred while writing aggregator " +
+ entry.getKey(), e);
+ }
+ }
+ try {
+ requestProcessor.sendAggregatedValuesToMaster(aggregatorOutput.flush());
+ } catch (IOException e) {
+ throw new IllegalStateException("finishSuperstep: " +
+ "IOException occured while sending aggregators to master", e);
+ }
+ // Wait for master to receive aggregated values before proceeding
+ serviceWorker.getWorkerClient().waitAllRequests();
+
+ ownerAggregatorData.reset();
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("finishSuperstep: Aggregators finished");
+ }
+ }
+
+ /**
+ * Create new aggregator usage which will be used by one of the compute
+ * threads.
+ *
+ * @return New aggregator usage
+ */
+ public WorkerThreadAggregatorUsage newThreadAggregatorUsage() {
+ if (AggregatorUtils.useThreadLocalAggregators(conf)) {
+ return new ThreadLocalWorkerAggregatorUsage();
+ } else {
+ return this;
+ }
+ }
+
+ @Override
+ public void finishThreadComputation() {
+ // If we don't use thread-local aggregators, all the aggregated values
+ // are already in this object
+ }
+
+ /**
+ * Not thread-safe implementation of {@link WorkerThreadAggregatorUsage}.
+ * We can use one instance of this object per thread to prevent
+ * synchronizing on each aggregate() call. In the end of superstep,
+ * values from each of these will be aggregated back to {@link
+ * WorkerAggregatorHandler}
+ */
+ public class ThreadLocalWorkerAggregatorUsage
+ implements WorkerThreadAggregatorUsage {
+ /** Thread-local aggregator map */
+ private final Map<String, Aggregator<Writable>> threadAggregatorMap;
+
+ /**
+ * Constructor
+ *
+ * Creates new instances of all aggregators from
+ * {@link WorkerAggregatorHandler}
+ */
+ public ThreadLocalWorkerAggregatorUsage() {
+ threadAggregatorMap = Maps.newHashMapWithExpectedSize(
+ WorkerAggregatorHandler.this.currentAggregatorMap.size());
+ for (Map.Entry<String, Aggregator<Writable>> entry :
+ WorkerAggregatorHandler.this.currentAggregatorMap.entrySet()) {
+ threadAggregatorMap.put(entry.getKey(),
+ AggregatorUtils.newAggregatorInstance(
+ (Class<Aggregator<Writable>>) entry.getValue().getClass()));
+ }
+ }
+
+ @Override
+ public <A extends Writable> void aggregate(String name, A value) {
+ Aggregator<Writable> aggregator = threadAggregatorMap.get(name);
+ if (aggregator != null) {
+ aggregator.aggregate(value);
+ } else {
+ throw new IllegalStateException("aggregate: " +
+ "Tried to aggregate value to unregistered aggregator " + name);
+ }
+ }
+
+ @Override
+ public <A extends Writable> A getAggregatedValue(String name) {
+ return WorkerAggregatorHandler.this.<A>getAggregatedValue(name);
+ }
+
+ @Override
+ public void finishThreadComputation() {
+ // Aggregate the values this thread's vertices provided back to
+ // WorkerAggregatorHandler
+ for (Map.Entry<String, Aggregator<Writable>> entry :
+ threadAggregatorMap.entrySet()) {
+ WorkerAggregatorHandler.this.aggregate(entry.getKey(),
+ entry.getValue().getAggregatedValue());
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/giraph/blob/45851391/giraph-core/src/main/java/org/apache/giraph/graph/WorkerAggregatorUsage.java
----------------------------------------------------------------------
diff --git a/giraph-core/src/main/java/org/apache/giraph/graph/WorkerAggregatorUsage.java b/giraph-core/src/main/java/org/apache/giraph/graph/WorkerAggregatorUsage.java
new file mode 100644
index 0000000..b7b98cd
--- /dev/null
+++ b/giraph-core/src/main/java/org/apache/giraph/graph/WorkerAggregatorUsage.java
@@ -0,0 +1,44 @@
+/*
+ * 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.graph;
+
+import org.apache.hadoop.io.Writable;
+
+/**
+ * Vertex classes can access and change aggregators through this interface
+ */
+public interface WorkerAggregatorUsage {
+ /**
+ * Add a new value
+ *
+ * @param name Name of aggregator
+ * @param value Value to add
+ * @param <A> Aggregated value
+ */
+ <A extends Writable> void aggregate(String name, A value);
+
+ /**
+ * Get value of an aggregator.
+ *
+ * @param name Name of aggregator
+ * @param <A> Aggregated value
+ * @return Value of the aggregator
+ */
+ <A extends Writable> A getAggregatedValue(String name);
+}
http://git-wip-us.apache.org/repos/asf/giraph/blob/45851391/giraph-core/src/main/java/org/apache/giraph/graph/WorkerContext.java
----------------------------------------------------------------------
diff --git a/giraph-core/src/main/java/org/apache/giraph/graph/WorkerContext.java b/giraph-core/src/main/java/org/apache/giraph/graph/WorkerContext.java
new file mode 100644
index 0000000..5f5f6fa
--- /dev/null
+++ b/giraph-core/src/main/java/org/apache/giraph/graph/WorkerContext.java
@@ -0,0 +1,121 @@
+/*
+ * 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.graph;
+
+import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.mapreduce.Mapper;
+
+/**
+ * WorkerContext allows for the execution of user code
+ * on a per-worker basis. There's one WorkerContext per worker.
+ */
+@SuppressWarnings("rawtypes")
+public abstract class WorkerContext implements WorkerAggregatorUsage {
+ /** Global graph state */
+ private GraphState graphState;
+
+ /**
+ * Set the graph state.
+ *
+ * @param graphState Used to set the graph state.
+ */
+ public void setGraphState(GraphState graphState) {
+ this.graphState = graphState;
+ }
+
+ /**
+ * Initialize the WorkerContext.
+ * This method is executed once on each Worker before the first
+ * superstep starts.
+ *
+ * @throws IllegalAccessException Thrown for getting the class
+ * @throws InstantiationException Expected instantiation in this method.
+ */
+ public abstract void preApplication() throws InstantiationException,
+ IllegalAccessException;
+
+ /**
+ * Finalize the WorkerContext.
+ * This method is executed once on each Worker after the last
+ * superstep ends.
+ */
+ public abstract void postApplication();
+
+ /**
+ * Execute user code.
+ * This method is executed once on each Worker before each
+ * superstep starts.
+ */
+ public abstract void preSuperstep();
+
+ /**
+ * Execute user code.
+ * This method is executed once on each Worker after each
+ * superstep ends.
+ */
+ public abstract void postSuperstep();
+
+ /**
+ * Retrieves the current superstep.
+ *
+ * @return Current superstep
+ */
+ public long getSuperstep() {
+ return graphState.getSuperstep();
+ }
+
+ /**
+ * Get the total (all workers) number of vertices that
+ * existed in the previous superstep.
+ *
+ * @return Total number of vertices (-1 if first superstep)
+ */
+ public long getTotalNumVertices() {
+ return graphState.getTotalNumVertices();
+ }
+
+ /**
+ * Get the total (all workers) number of edges that
+ * existed in the previous superstep.
+ *
+ * @return Total number of edges (-1 if first superstep)
+ */
+ public long getTotalNumEdges() {
+ return graphState.getTotalNumEdges();
+ }
+
+ /**
+ * Get the mapper context
+ *
+ * @return Mapper context
+ */
+ public Mapper.Context getContext() {
+ return graphState.getContext();
+ }
+
+ @Override
+ public <A extends Writable> void aggregate(String name, A value) {
+ graphState.getWorkerAggregatorUsage().aggregate(name, value);
+ }
+
+ @Override
+ public <A extends Writable> A getAggregatedValue(String name) {
+ return graphState.getWorkerAggregatorUsage().<A>getAggregatedValue(name);
+ }
+}
http://git-wip-us.apache.org/repos/asf/giraph/blob/45851391/giraph-core/src/main/java/org/apache/giraph/graph/WorkerInfo.java
----------------------------------------------------------------------
diff --git a/giraph-core/src/main/java/org/apache/giraph/graph/WorkerInfo.java b/giraph-core/src/main/java/org/apache/giraph/graph/WorkerInfo.java
new file mode 100644
index 0000000..c9571ba
--- /dev/null
+++ b/giraph-core/src/main/java/org/apache/giraph/graph/WorkerInfo.java
@@ -0,0 +1,35 @@
+/*
+ * 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.graph;
+
+/**
+ * Information about a worker that is sent to the master and other workers.
+ */
+public class WorkerInfo extends TaskInfo {
+ /**
+ * Constructor for reflection
+ */
+ public WorkerInfo() {
+ }
+
+ @Override
+ public String toString() {
+ return "Worker(" + super.toString() + ")";
+ }
+}
http://git-wip-us.apache.org/repos/asf/giraph/blob/45851391/giraph-core/src/main/java/org/apache/giraph/graph/WorkerThreadAggregatorUsage.java
----------------------------------------------------------------------
diff --git a/giraph-core/src/main/java/org/apache/giraph/graph/WorkerThreadAggregatorUsage.java b/giraph-core/src/main/java/org/apache/giraph/graph/WorkerThreadAggregatorUsage.java
new file mode 100644
index 0000000..e36ff48
--- /dev/null
+++ b/giraph-core/src/main/java/org/apache/giraph/graph/WorkerThreadAggregatorUsage.java
@@ -0,0 +1,31 @@
+/*
+ * 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.graph;
+
+/**
+ * {@link WorkerAggregatorUsage} which can be used in each of the
+ * computation threads.
+ */
+public interface WorkerThreadAggregatorUsage extends WorkerAggregatorUsage {
+ /**
+ * Call this after thread's computation is finished,
+ * i.e. when all vertices have provided their values to aggregators
+ */
+ void finishThreadComputation();
+}
http://git-wip-us.apache.org/repos/asf/giraph/blob/45851391/giraph-core/src/main/java/org/apache/giraph/graph/package-info.java
----------------------------------------------------------------------
diff --git a/giraph-core/src/main/java/org/apache/giraph/graph/package-info.java b/giraph-core/src/main/java/org/apache/giraph/graph/package-info.java
new file mode 100644
index 0000000..26cd46a
--- /dev/null
+++ b/giraph-core/src/main/java/org/apache/giraph/graph/package-info.java
@@ -0,0 +1,22 @@
+/*
+ * 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 of all the graph related objects, built on the
+ * org.apache.bsp package.
+ */
+package org.apache.giraph.graph;
http://git-wip-us.apache.org/repos/asf/giraph/blob/45851391/giraph-core/src/main/java/org/apache/giraph/graph/partition/BasicPartitionOwner.java
----------------------------------------------------------------------
diff --git a/giraph-core/src/main/java/org/apache/giraph/graph/partition/BasicPartitionOwner.java b/giraph-core/src/main/java/org/apache/giraph/graph/partition/BasicPartitionOwner.java
new file mode 100644
index 0000000..e063ac7
--- /dev/null
+++ b/giraph-core/src/main/java/org/apache/giraph/graph/partition/BasicPartitionOwner.java
@@ -0,0 +1,163 @@
+/*
+ * 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.graph.partition;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+
+import org.apache.giraph.conf.ImmutableClassesGiraphConfigurable;
+import org.apache.giraph.conf.ImmutableClassesGiraphConfiguration;
+import org.apache.giraph.graph.WorkerInfo;
+
+/**
+ * Basic partition owner, can be subclassed for more complicated partition
+ * owner implementations.
+ */
+public class BasicPartitionOwner implements PartitionOwner,
+ ImmutableClassesGiraphConfigurable {
+ /** Configuration */
+ private ImmutableClassesGiraphConfiguration conf;
+ /** Partition id */
+ private int partitionId = -1;
+ /** Owning worker information */
+ private WorkerInfo workerInfo;
+ /** Previous (if any) worker info */
+ private WorkerInfo previousWorkerInfo;
+ /** Checkpoint files prefix for this partition */
+ private String checkpointFilesPrefix;
+
+ /**
+ * Default constructor.
+ */
+ public BasicPartitionOwner() { }
+
+ /**
+ * Constructor with partition id and worker info.
+ *
+ * @param partitionId Partition id of this partition.
+ * @param workerInfo Owner of the partition.
+ */
+ public BasicPartitionOwner(int partitionId, WorkerInfo workerInfo) {
+ this(partitionId, workerInfo, null, null);
+ }
+
+ /**
+ * Constructor with partition id and worker info.
+ *
+ * @param partitionId Partition id of this partition.
+ * @param workerInfo Owner of the partition.
+ * @param previousWorkerInfo Previous owner of this partition.
+ * @param checkpointFilesPrefix Prefix of the checkpoint files.
+ */
+ public BasicPartitionOwner(int partitionId,
+ WorkerInfo workerInfo,
+ WorkerInfo previousWorkerInfo,
+ String checkpointFilesPrefix) {
+ this.partitionId = partitionId;
+ this.workerInfo = workerInfo;
+ this.previousWorkerInfo = previousWorkerInfo;
+ this.checkpointFilesPrefix = checkpointFilesPrefix;
+ }
+
+ @Override
+ public int getPartitionId() {
+ return partitionId;
+ }
+
+ @Override
+ public WorkerInfo getWorkerInfo() {
+ return workerInfo;
+ }
+
+ @Override
+ public void setWorkerInfo(WorkerInfo workerInfo) {
+ this.workerInfo = workerInfo;
+ }
+
+ @Override
+ public WorkerInfo getPreviousWorkerInfo() {
+ return previousWorkerInfo;
+ }
+
+ @Override
+ public void setPreviousWorkerInfo(WorkerInfo workerInfo) {
+ this.previousWorkerInfo = workerInfo;
+ }
+
+ @Override
+ public String getCheckpointFilesPrefix() {
+ return checkpointFilesPrefix;
+ }
+
+ @Override
+ public void setCheckpointFilesPrefix(String checkpointFilesPrefix) {
+ this.checkpointFilesPrefix = checkpointFilesPrefix;
+ }
+
+ @Override
+ public void readFields(DataInput input) throws IOException {
+ partitionId = input.readInt();
+ workerInfo = new WorkerInfo();
+ workerInfo.readFields(input);
+ boolean hasPreviousWorkerInfo = input.readBoolean();
+ if (hasPreviousWorkerInfo) {
+ previousWorkerInfo = new WorkerInfo();
+ previousWorkerInfo.readFields(input);
+ }
+ boolean hasCheckpointFilePrefix = input.readBoolean();
+ if (hasCheckpointFilePrefix) {
+ checkpointFilesPrefix = input.readUTF();
+ }
+ }
+
+ @Override
+ public void write(DataOutput output) throws IOException {
+ output.writeInt(partitionId);
+ workerInfo.write(output);
+ if (previousWorkerInfo != null) {
+ output.writeBoolean(true);
+ previousWorkerInfo.write(output);
+ } else {
+ output.writeBoolean(false);
+ }
+ if (checkpointFilesPrefix != null) {
+ output.writeBoolean(true);
+ output.writeUTF(checkpointFilesPrefix);
+ } else {
+ output.writeBoolean(false);
+ }
+ }
+
+ @Override
+ public ImmutableClassesGiraphConfiguration getConf() {
+ return conf;
+ }
+
+ @Override
+ public void setConf(ImmutableClassesGiraphConfiguration conf) {
+ this.conf = conf;
+ }
+
+ @Override
+ public String toString() {
+ return "(id=" + partitionId + ",cur=" + workerInfo + ",prev=" +
+ previousWorkerInfo + ",ckpt_file=" + checkpointFilesPrefix + ")";
+ }
+}