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:34 UTC
[50/51] [partial] GIRAPH-457: update module names (nitay)
http://git-wip-us.apache.org/repos/asf/giraph/blob/45851391/giraph/src/main/java/org/apache/giraph/graph/Vertex.java
----------------------------------------------------------------------
diff --git a/giraph/src/main/java/org/apache/giraph/graph/Vertex.java b/giraph/src/main/java/org/apache/giraph/graph/Vertex.java
deleted file mode 100644
index 6db4735..0000000
--- a/giraph/src/main/java/org/apache/giraph/graph/Vertex.java
+++ /dev/null
@@ -1,405 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.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/src/main/java/org/apache/giraph/graph/VertexChanges.java
----------------------------------------------------------------------
diff --git a/giraph/src/main/java/org/apache/giraph/graph/VertexChanges.java b/giraph/src/main/java/org/apache/giraph/graph/VertexChanges.java
deleted file mode 100644
index 957cf82..0000000
--- a/giraph/src/main/java/org/apache/giraph/graph/VertexChanges.java
+++ /dev/null
@@ -1,68 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.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/src/main/java/org/apache/giraph/graph/VertexEdgeCount.java
----------------------------------------------------------------------
diff --git a/giraph/src/main/java/org/apache/giraph/graph/VertexEdgeCount.java b/giraph/src/main/java/org/apache/giraph/graph/VertexEdgeCount.java
deleted file mode 100644
index c2d13cc..0000000
--- a/giraph/src/main/java/org/apache/giraph/graph/VertexEdgeCount.java
+++ /dev/null
@@ -1,88 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.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/src/main/java/org/apache/giraph/graph/VertexInputFormat.java
----------------------------------------------------------------------
diff --git a/giraph/src/main/java/org/apache/giraph/graph/VertexInputFormat.java b/giraph/src/main/java/org/apache/giraph/graph/VertexInputFormat.java
deleted file mode 100644
index 9824b69..0000000
--- a/giraph/src/main/java/org/apache/giraph/graph/VertexInputFormat.java
+++ /dev/null
@@ -1,82 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.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/src/main/java/org/apache/giraph/graph/VertexInputSplitsCallable.java
----------------------------------------------------------------------
diff --git a/giraph/src/main/java/org/apache/giraph/graph/VertexInputSplitsCallable.java b/giraph/src/main/java/org/apache/giraph/graph/VertexInputSplitsCallable.java
deleted file mode 100644
index e1e6235..0000000
--- a/giraph/src/main/java/org/apache/giraph/graph/VertexInputSplitsCallable.java
+++ /dev/null
@@ -1,180 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.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/src/main/java/org/apache/giraph/graph/VertexInputSplitsCallableFactory.java
----------------------------------------------------------------------
diff --git a/giraph/src/main/java/org/apache/giraph/graph/VertexInputSplitsCallableFactory.java b/giraph/src/main/java/org/apache/giraph/graph/VertexInputSplitsCallableFactory.java
deleted file mode 100644
index a55c4b0..0000000
--- a/giraph/src/main/java/org/apache/giraph/graph/VertexInputSplitsCallableFactory.java
+++ /dev/null
@@ -1,94 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.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/src/main/java/org/apache/giraph/graph/VertexMutations.java
----------------------------------------------------------------------
diff --git a/giraph/src/main/java/org/apache/giraph/graph/VertexMutations.java b/giraph/src/main/java/org/apache/giraph/graph/VertexMutations.java
deleted file mode 100644
index 4548cea..0000000
--- a/giraph/src/main/java/org/apache/giraph/graph/VertexMutations.java
+++ /dev/null
@@ -1,212 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.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/src/main/java/org/apache/giraph/graph/VertexOutputFormat.java
----------------------------------------------------------------------
diff --git a/giraph/src/main/java/org/apache/giraph/graph/VertexOutputFormat.java b/giraph/src/main/java/org/apache/giraph/graph/VertexOutputFormat.java
deleted file mode 100644
index d4c8e1c..0000000
--- a/giraph/src/main/java/org/apache/giraph/graph/VertexOutputFormat.java
+++ /dev/null
@@ -1,81 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.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/src/main/java/org/apache/giraph/graph/VertexReader.java
----------------------------------------------------------------------
diff --git a/giraph/src/main/java/org/apache/giraph/graph/VertexReader.java b/giraph/src/main/java/org/apache/giraph/graph/VertexReader.java
deleted file mode 100644
index 77801cc..0000000
--- a/giraph/src/main/java/org/apache/giraph/graph/VertexReader.java
+++ /dev/null
@@ -1,87 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.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/src/main/java/org/apache/giraph/graph/VertexResolver.java
----------------------------------------------------------------------
diff --git a/giraph/src/main/java/org/apache/giraph/graph/VertexResolver.java b/giraph/src/main/java/org/apache/giraph/graph/VertexResolver.java
deleted file mode 100644
index 1616290..0000000
--- a/giraph/src/main/java/org/apache/giraph/graph/VertexResolver.java
+++ /dev/null
@@ -1,62 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.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/src/main/java/org/apache/giraph/graph/VertexValueInputFormat.java
----------------------------------------------------------------------
diff --git a/giraph/src/main/java/org/apache/giraph/graph/VertexValueInputFormat.java b/giraph/src/main/java/org/apache/giraph/graph/VertexValueInputFormat.java
deleted file mode 100644
index 804d23e..0000000
--- a/giraph/src/main/java/org/apache/giraph/graph/VertexValueInputFormat.java
+++ /dev/null
@@ -1,58 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.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/src/main/java/org/apache/giraph/graph/VertexValueReader.java
----------------------------------------------------------------------
diff --git a/giraph/src/main/java/org/apache/giraph/graph/VertexValueReader.java b/giraph/src/main/java/org/apache/giraph/graph/VertexValueReader.java
deleted file mode 100644
index 569714d..0000000
--- a/giraph/src/main/java/org/apache/giraph/graph/VertexValueReader.java
+++ /dev/null
@@ -1,61 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.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/src/main/java/org/apache/giraph/graph/VertexWriter.java
----------------------------------------------------------------------
diff --git a/giraph/src/main/java/org/apache/giraph/graph/VertexWriter.java b/giraph/src/main/java/org/apache/giraph/graph/VertexWriter.java
deleted file mode 100644
index 7c0e06e..0000000
--- a/giraph/src/main/java/org/apache/giraph/graph/VertexWriter.java
+++ /dev/null
@@ -1,67 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.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/src/main/java/org/apache/giraph/graph/WorkerAggregatorHandler.java
----------------------------------------------------------------------
diff --git a/giraph/src/main/java/org/apache/giraph/graph/WorkerAggregatorHandler.java b/giraph/src/main/java/org/apache/giraph/graph/WorkerAggregatorHandler.java
deleted file mode 100644
index 8e35998..0000000
--- a/giraph/src/main/java/org/apache/giraph/graph/WorkerAggregatorHandler.java
+++ /dev/null
@@ -1,294 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.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/src/main/java/org/apache/giraph/graph/WorkerAggregatorUsage.java
----------------------------------------------------------------------
diff --git a/giraph/src/main/java/org/apache/giraph/graph/WorkerAggregatorUsage.java b/giraph/src/main/java/org/apache/giraph/graph/WorkerAggregatorUsage.java
deleted file mode 100644
index b7b98cd..0000000
--- a/giraph/src/main/java/org/apache/giraph/graph/WorkerAggregatorUsage.java
+++ /dev/null
@@ -1,44 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.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/src/main/java/org/apache/giraph/graph/WorkerContext.java
----------------------------------------------------------------------
diff --git a/giraph/src/main/java/org/apache/giraph/graph/WorkerContext.java b/giraph/src/main/java/org/apache/giraph/graph/WorkerContext.java
deleted file mode 100644
index 5f5f6fa..0000000
--- a/giraph/src/main/java/org/apache/giraph/graph/WorkerContext.java
+++ /dev/null
@@ -1,121 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.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/src/main/java/org/apache/giraph/graph/WorkerInfo.java
----------------------------------------------------------------------
diff --git a/giraph/src/main/java/org/apache/giraph/graph/WorkerInfo.java b/giraph/src/main/java/org/apache/giraph/graph/WorkerInfo.java
deleted file mode 100644
index c9571ba..0000000
--- a/giraph/src/main/java/org/apache/giraph/graph/WorkerInfo.java
+++ /dev/null
@@ -1,35 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.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/src/main/java/org/apache/giraph/graph/WorkerThreadAggregatorUsage.java
----------------------------------------------------------------------
diff --git a/giraph/src/main/java/org/apache/giraph/graph/WorkerThreadAggregatorUsage.java b/giraph/src/main/java/org/apache/giraph/graph/WorkerThreadAggregatorUsage.java
deleted file mode 100644
index e36ff48..0000000
--- a/giraph/src/main/java/org/apache/giraph/graph/WorkerThreadAggregatorUsage.java
+++ /dev/null
@@ -1,31 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.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/src/main/java/org/apache/giraph/graph/package-info.java
----------------------------------------------------------------------
diff --git a/giraph/src/main/java/org/apache/giraph/graph/package-info.java b/giraph/src/main/java/org/apache/giraph/graph/package-info.java
deleted file mode 100644
index 26cd46a..0000000
--- a/giraph/src/main/java/org/apache/giraph/graph/package-info.java
+++ /dev/null
@@ -1,22 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-/**
- * Package 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/src/main/java/org/apache/giraph/graph/partition/BasicPartitionOwner.java
----------------------------------------------------------------------
diff --git a/giraph/src/main/java/org/apache/giraph/graph/partition/BasicPartitionOwner.java b/giraph/src/main/java/org/apache/giraph/graph/partition/BasicPartitionOwner.java
deleted file mode 100644
index e063ac7..0000000
--- a/giraph/src/main/java/org/apache/giraph/graph/partition/BasicPartitionOwner.java
+++ /dev/null
@@ -1,163 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.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 + ")";
- }
-}