pregelix/pregelix-api/src/main/java/edu/uci/ics/pregelix/api/graph/Vertex.java - asterixdb - Git at Google

 /*
  * Copyright 2009-2010 by The Regents of the University of California
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
  * you may obtain a copy of the License from
  *
  *     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 edu.uci.ics.pregelix.api.graph;

 import java.io.DataInput;
 import java.io.DataOutput;
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;

 import org.apache.hadoop.io.Writable;
 import org.apache.hadoop.io.WritableComparable;
 import org.apache.hadoop.mapreduce.TaskAttemptContext;

 import edu.uci.ics.hyracks.api.comm.IFrameWriter;
 import edu.uci.ics.hyracks.dataflow.common.comm.io.ArrayTupleBuilder;
 import edu.uci.ics.hyracks.dataflow.common.comm.io.FrameTupleAppender;
 import edu.uci.ics.pregelix.api.util.BspUtils;
 import edu.uci.ics.pregelix.api.util.SerDeUtils;

 /**
  * User applications should all inherit {@link Vertex}, and implement their own
  * *compute* method.
  *
  * @param <I>
  *            Vertex identifier type
  * @param <V>
  *            Vertex value type
  * @param <E>
  *            Edge value type
  * @param <M>
  *            Message value type
  */
 @SuppressWarnings("rawtypes")
 public abstract class Vertex<I extends WritableComparable, V extends Writable, E extends Writable, M extends Writable>
 		implements Writable {
 	private static long superstep = 0;
 	/** Class-wide number of vertices */
 	private static long numVertices = -1;
 	/** Class-wide number of edges */
 	private static long numEdges = -1;
 	/** Vertex id */
 	private I vertexId = null;
 	/** Vertex value */
 	private V vertexValue = null;
 	/** Map of destination vertices and their edge values */
 	private final List<Edge<I, E>> destEdgeList = new ArrayList<Edge<I, E>>();
 	/** If true, do not do anymore computation on this vertex. */
 	boolean halt = false;
 	/** List of incoming messages from the previous superstep */
 	private final List<M> msgList = new ArrayList<M>();
 	/** map context */
 	private static TaskAttemptContext context = null;
 	/** a delegate for hyracks stuff */
 	private VertexDelegate<I, V, E, M> delegate = new VertexDelegate<I, V, E, M>(
 			this);
 	/** this vertex is updated or not */
 	private boolean updated = false;
 	/** has outgoing messages */
 	private boolean hasMessage = false;
 	/** created new vertex */
 	private boolean createdNewLiveVertex = false;

 	/**
 	 * use object pool for re-using objects
 	 */
 	private List<Edge<I, E>> edgePool = new ArrayList<Edge<I, E>>();
 	private List<M> msgPool = new ArrayList<M>();
 	private List<V> valuePool = new ArrayList<V>();
 	private int usedEdge = 0;
 	private int usedMessage = 0;
 	private int usedValue = 0;

 	/**
 	 * The key method that users need to implement
 	 *
 	 * @param msgIterator
 	 *            an iterator of incoming messages
 	 */
 	public abstract void compute(Iterator<M> msgIterator);

 	/**
 	 * Add an edge for the vertex.
 	 *
 	 * @param targetVertexId
 	 * @param edgeValue
 	 * @return successful or not
 	 */
 	public final boolean addEdge(I targetVertexId, E edgeValue) {
 		Edge<I, E> edge = this.allocateEdge();
 		edge.setDestVertexId(targetVertexId);
 		edge.setEdgeValue(edgeValue);
 		destEdgeList.add(edge);
 		updated = true;
 		return true;
 	}

 	/**
 	 * Initialize a new vertex
 	 *
 	 * @param vertexId
 	 * @param vertexValue
 	 * @param edges
 	 * @param messages
 	 */
 	public void initialize(I vertexId, V vertexValue, Map<I, E> edges,
 			List<M> messages) {
 		if (vertexId != null) {
 			setVertexId(vertexId);
 		}
 		if (vertexValue != null) {
 			setVertexValue(vertexValue);
 		}
 		destEdgeList.clear();
 		if (edges != null && !edges.isEmpty()) {
 			for (Map.Entry<I, E> entry : edges.entrySet()) {
 				destEdgeList.add(new Edge<I, E>(entry.getKey(), entry
 						.getValue()));
 			}
 		}
 		if (messages != null && !messages.isEmpty()) {
 			msgList.addAll(messages);
 		}
 	}

 	/**
 	 * reset a vertex object: clear its internal states
 	 */
 	public void reset() {
 		usedEdge = 0;
 		usedMessage = 0;
 		usedValue = 0;
 		updated = false;
 	}

 	/**
 	 * Set the vertex id
 	 *
 	 * @param vertexId
 	 */
 	public final void setVertexId(I vertexId) {
 		this.vertexId = vertexId;
 		delegate.setVertexId(vertexId);
 	}

 	/**
 	 * Get the vertex id
 	 *
 	 * @return vertex id
 	 */
 	public final I getVertexId() {
 		return vertexId;
 	}

 	/**
 	 * Get the vertex value
 	 *
 	 * @return the vertex value
 	 */
 	public final V getVertexValue() {
 		return vertexValue;
 	}

 	/**
 	 * Set the vertex value
 	 *
 	 * @param vertexValue
 	 */
 	public final void setVertexValue(V vertexValue) {
 		this.vertexValue = vertexValue;
 		this.updated = true;
 	}

 	/***
 	 * Send a message to a specific vertex
 	 *
 	 * @param id
 	 *            the receiver vertex id
 	 * @param msg
 	 *            the message
 	 */
 	public final void sendMsg(I id, M msg) {
 		if (msg == null) {
 			throw new IllegalArgumentException(
 					"sendMsg: Cannot send null message to " + id);
 		}
 		delegate.sendMsg(id, msg);
 		this.hasMessage = true;
 	}

 	/**
 	 * Send a message to all direct outgoing neighbors
 	 *
 	 * @param msg
 	 *            the message
 	 */
 	public final void sendMsgToAllEdges(M msg) {
 		if (msg == null) {
 			throw new IllegalArgumentException(
 					"sendMsgToAllEdges: Cannot send null message to all edges");
 		}
 		for (Edge<I, E> edge : destEdgeList) {
 			sendMsg(edge.getDestVertexId(), msg);
 		}
 	}

 	/**
 	 * Vote to halt. Once all vertex vote to halt and no more messages, a
 	 * Pregelix job will terminate.
 	 */
 	public final void voteToHalt() {
 		halt = true;
 		updated = true;
 	}

 	/**
 	 * @return the vertex is halted (true) or not (false)
 	 */
 	public final boolean isHalted() {
 		return halt;
 	}

 	@Override
 	final public void readFields(DataInput in) throws IOException {
 		reset();
 		if (vertexId == null)
 			vertexId = BspUtils.<I> createVertexIndex(getContext()
 					.getConfiguration());
 		vertexId.readFields(in);
 		delegate.setVertexId(vertexId);
 		boolean hasVertexValue = in.readBoolean();

 		if (hasVertexValue) {
 			vertexValue = allocateValue();
 			vertexValue.readFields(in);
 			delegate.setVertex(this);
 		}
 		destEdgeList.clear();
 		long edgeMapSize = SerDeUtils.readVLong(in);
 		for (long i = 0; i < edgeMapSize; ++i) {
 			Edge<I, E> edge = allocateEdge();
 			edge.setConf(getContext().getConfiguration());
 			edge.readFields(in);
 			addEdge(edge);
 		}
 		msgList.clear();
 		long msgListSize = SerDeUtils.readVLong(in);
 		for (long i = 0; i < msgListSize; ++i) {
 			M msg = allocateMessage();
 			msg.readFields(in);
 			msgList.add(msg);
 		}
 		halt = in.readBoolean();
 		updated = false;
 		hasMessage = false;
 		createdNewLiveVertex = false;
 	}

 	@Override
 	public void write(DataOutput out) throws IOException {
 		vertexId.write(out);
 		out.writeBoolean(vertexValue != null);
 		if (vertexValue != null) {
 			vertexValue.write(out);
 		}
 		SerDeUtils.writeVLong(out, destEdgeList.size());
 		for (Edge<I, E> edge : destEdgeList) {
 			edge.write(out);
 		}
 		SerDeUtils.writeVLong(out, msgList.size());
 		for (M msg : msgList) {
 			msg.write(out);
 		}
 		out.writeBoolean(halt);
 	}

 	/**
 	 * Get the list of incoming messages
 	 *
 	 * @return the list of messages
 	 */
 	public List<M> getMsgList() {
 		return msgList;
 	}

 	/**
 	 * Get outgoing edge list
 	 *
 	 * @return a list of outgoing edges
 	 */
 	public List<Edge<I, E>> getEdges() {
 		return this.destEdgeList;
 	}

 	@Override
 	@SuppressWarnings("unchecked")
 	public String toString() {
 		Collections.sort(destEdgeList);
 		StringBuffer edgeBuffer = new StringBuffer();
 		edgeBuffer.append("(");
 		for (Edge<I, E> edge : destEdgeList) {
 			edgeBuffer.append(edge.getDestVertexId()).append(",");
 		}
 		edgeBuffer.append(")");
 		return "Vertex(id=" + getVertexId() + ",value=" + getVertexValue()
 				+ ", edges=" + edgeBuffer + ")";
 	}

 	/**
 	 * Get the number of outgoing edges
 	 *
 	 * @return the number of outging edges
 	 */
 	public int getNumOutEdges() {
 		return destEdgeList.size();
 	}

 	/**
 	 * Pregelix internal use only
 	 *
 	 * @param writers
 	 */
 	public void setOutputWriters(List<IFrameWriter> writers) {
 		delegate.setOutputWriters(writers);
 	}

 	/**
 	 * Pregelix internal use only
 	 *
 	 * @param writers
 	 */
 	public void setOutputAppenders(List<FrameTupleAppender> appenders) {
 		delegate.setOutputAppenders(appenders);
 	}

 	/**
 	 * Pregelix internal use only
 	 *
 	 * @param writers
 	 */
 	public void setOutputTupleBuilders(List<ArrayTupleBuilder> tbs) {
 		delegate.setOutputTupleBuilders(tbs);
 	}

 	/**
 	 * Pregelix internal use only
 	 *
 	 * @param writers
 	 */
 	public void finishCompute() throws IOException {
 		delegate.finishCompute();
 	}

 	/**
 	 * Pregelix internal use only
 	 */
 	public boolean hasUpdate() {
 		return this.updated;
 	}

 	/**
 	 * Pregelix internal use only
 	 */
 	public boolean hasMessage() {
 		return this.hasMessage;
 	}

 	/**
 	 * Pregelix internal use only
 	 */
 	public boolean createdNewLiveVertex() {
 		return this.createdNewLiveVertex;
 	}

 	/**
 	 * sort the edges
 	 */
 	@SuppressWarnings("unchecked")
 	public void sortEdges() {
 		updated = true;
 		Collections.sort(destEdgeList);
 	}

 	/**
 	 * Allocate a new edge from the edge pool
 	 */
 	private Edge<I, E> allocateEdge() {
 		Edge<I, E> edge;
 		if (usedEdge < edgePool.size()) {
 			edge = edgePool.get(usedEdge);
 			usedEdge++;
 		} else {
 			edge = new Edge<I, E>();
 			edgePool.add(edge);
 			usedEdge++;
 		}
 		return edge;
 	}

 	/**
 	 * Allocate a new message from the message pool
 	 */
 	private M allocateMessage() {
 		M message;
 		if (usedMessage < msgPool.size()) {
 			message = msgPool.get(usedEdge);
 			usedMessage++;
 		} else {
 			message = BspUtils.<M> createMessageValue(getContext()
 					.getConfiguration());
 			msgPool.add(message);
 			usedMessage++;
 		}
 		return message;
 	}

 	/**
 	 * Set the global superstep for all the vertices (internal use)
 	 *
 	 * @param superstep
 	 *            New superstep
 	 */
 	public static final void setSuperstep(long superstep) {
 		Vertex.superstep = superstep;
 	}

 	/**
 	 * Add an outgoing edge into the vertex
 	 *
 	 * @param edge
 	 *            the edge to be added
 	 * @return true if the edge list changed as a result of this call
 	 */
 	public boolean addEdge(Edge<I, E> edge) {
 		edge.setConf(getContext().getConfiguration());
 		updated = true;
 		return destEdgeList.add(edge);
 	}

 	/**
 	 * remove an outgoing edge in the graph
 	 *
 	 * @param edge
 	 *            the edge to be removed
 	 * @return true if the edge is in the edge list of the vertex
 	 */
 	public boolean removeEdge(Edge<I, E> edge) {
 		updated = true;
 		return destEdgeList.remove(edge);
 	}

 	/**
 	 * Add a new vertex into the graph
 	 *
 	 * @param vertexId
 	 *            the vertex id
 	 * @param vertex
 	 *            the vertex
 	 */
 	public final void addVertex(I vertexId, Vertex vertex) {
 		createdNewLiveVertex |= !vertex.isHalted();
 		delegate.addVertex(vertexId, vertex);
 	}

 	/**
 	 * Delete a vertex from id
 	 *
 	 * @param vertexId
 	 *            the vertex id
 	 */
 	public final void deleteVertex(I vertexId) {
 		delegate.deleteVertex(vertexId);
 	}

 	/**
 	 * Allocate a vertex value from the object pool
 	 *
 	 * @return a vertex value instance
 	 */
 	private V allocateValue() {
 		V value;
 		if (usedValue < valuePool.size()) {
 			value = valuePool.get(usedValue);
 			usedValue++;
 		} else {
 			value = BspUtils.<V> createVertexValue(getContext()
 					.getConfiguration());
 			valuePool.add(value);
 			usedValue++;
 		}
 		return value;
 	}

 	/**
 	 * Get the current global superstep number
 	 *
 	 * @return the current superstep number
 	 */
 	public static final long getSuperstep() {
 		return superstep;
 	}

 	/**
 	 * Set the total number of vertices from the last superstep.
 	 *
 	 * @param numVertices
 	 *            Aggregate vertices in the last superstep
 	 */
 	public static final void setNumVertices(long numVertices) {
 		Vertex.numVertices = numVertices;
 	}

 	/**
 	 * Get the number of vertexes in the graph
 	 *
 	 * @return the number of vertexes in the graph
 	 */
 	public static final long getNumVertices() {
 		return numVertices;
 	}

 	/**
 	 * Set the total number of edges from the last superstep.
 	 *
 	 * @param numEdges
 	 *            Aggregate edges in the last superstep
 	 */
 	public static void setNumEdges(long numEdges) {
 		Vertex.numEdges = numEdges;
 	}

 	/**
 	 * Get the number of edges from this graph
 	 *
 	 * @return the number of edges in the graph
 	 */
 	public static final long getNumEdges() {
 		return numEdges;
 	}

 	/**
 	 * Pregelix internal use only
 	 */
 	public static final TaskAttemptContext getContext() {
 		return context;
 	}

 	/**
 	 * Pregelix internal use only
 	 *
 	 * @param context
 	 */
 	public static final void setContext(TaskAttemptContext context) {
 		Vertex.context = context;
 	}

 }
	/*
	* Copyright 2009-2010 by The Regents of the University of California
	* Licensed under the Apache License, Version 2.0 (the "License");
	* you may not use this file except in compliance with the License.
	* you may obtain a copy of the License from
	*
	* 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 edu.uci.ics.pregelix.api.graph;

	import java.io.DataInput;
	import java.io.DataOutput;
	import java.io.IOException;
	import java.util.ArrayList;
	import java.util.Collections;
	import java.util.Iterator;
	import java.util.List;
	import java.util.Map;

	import org.apache.hadoop.io.Writable;
	import org.apache.hadoop.io.WritableComparable;
	import org.apache.hadoop.mapreduce.TaskAttemptContext;

	import edu.uci.ics.hyracks.api.comm.IFrameWriter;
	import edu.uci.ics.hyracks.dataflow.common.comm.io.ArrayTupleBuilder;
	import edu.uci.ics.hyracks.dataflow.common.comm.io.FrameTupleAppender;
	import edu.uci.ics.pregelix.api.util.BspUtils;
	import edu.uci.ics.pregelix.api.util.SerDeUtils;

	/**
	* User applications should all inherit {@link Vertex}, and implement their own
	* compute method.
	*
	* @param <I>
	* Vertex identifier type
	* @param <V>
	* Vertex value type
	* @param <E>
	* Edge value type
	* @param <M>
	* Message value type
	*/
	@SuppressWarnings("rawtypes")
	public abstract class Vertex<I extends WritableComparable, V extends Writable, E extends Writable, M extends Writable>
	implements Writable {
	private static long superstep = 0;
	/** Class-wide number of vertices */
	private static long numVertices = -1;
	/** Class-wide number of edges */
	private static long numEdges = -1;
	/** Vertex id */
	private I vertexId = null;
	/** Vertex value */
	private V vertexValue = null;
	/** Map of destination vertices and their edge values */
	private final List<Edge<I, E>> destEdgeList = new ArrayList<Edge<I, E>>();
	/** If true, do not do anymore computation on this vertex. */
	boolean halt = false;
	/** List of incoming messages from the previous superstep */
	private final List<M> msgList = new ArrayList<M>();
	/** map context */
	private static TaskAttemptContext context = null;
	/** a delegate for hyracks stuff */
	private VertexDelegate<I, V, E, M> delegate = new VertexDelegate<I, V, E, M>(
	this);
	/** this vertex is updated or not */
	private boolean updated = false;
	/** has outgoing messages */
	private boolean hasMessage = false;
	/** created new vertex */
	private boolean createdNewLiveVertex = false;

	/**
	* use object pool for re-using objects
	*/
	private List<Edge<I, E>> edgePool = new ArrayList<Edge<I, E>>();
	private List<M> msgPool = new ArrayList<M>();
	private List<V> valuePool = new ArrayList<V>();
	private int usedEdge = 0;
	private int usedMessage = 0;
	private int usedValue = 0;

	/**
	* The key method that users need to implement
	*
	* @param msgIterator
	* an iterator of incoming messages
	*/
	public abstract void compute(Iterator<M> msgIterator);

	/**
	* Add an edge for the vertex.
	*
	* @param targetVertexId
	* @param edgeValue
	* @return successful or not
	*/
	public final boolean addEdge(I targetVertexId, E edgeValue) {
	Edge<I, E> edge = this.allocateEdge();
	edge.setDestVertexId(targetVertexId);
	edge.setEdgeValue(edgeValue);
	destEdgeList.add(edge);
	updated = true;
	return true;
	}

	/**
	* Initialize a new vertex
	*
	* @param vertexId
	* @param vertexValue
	* @param edges
	* @param messages
	*/
	public void initialize(I vertexId, V vertexValue, Map<I, E> edges,
	List<M> messages) {
	if (vertexId != null) {
	setVertexId(vertexId);
	}
	if (vertexValue != null) {
	setVertexValue(vertexValue);
	}
	destEdgeList.clear();
	if (edges != null && !edges.isEmpty()) {
	for (Map.Entry<I, E> entry : edges.entrySet()) {
	destEdgeList.add(new Edge<I, E>(entry.getKey(), entry
	.getValue()));
	}
	}
	if (messages != null && !messages.isEmpty()) {
	msgList.addAll(messages);
	}
	}

	/**
	* reset a vertex object: clear its internal states
	*/
	public void reset() {
	usedEdge = 0;
	usedMessage = 0;
	usedValue = 0;
	updated = false;
	}

	/**
	* Set the vertex id
	*
	* @param vertexId
	*/
	public final void setVertexId(I vertexId) {
	this.vertexId = vertexId;
	delegate.setVertexId(vertexId);
	}

	/**
	* Get the vertex id
	*
	* @return vertex id
	*/
	public final I getVertexId() {
	return vertexId;
	}

	/**
	* Get the vertex value
	*
	* @return the vertex value
	*/
	public final V getVertexValue() {
	return vertexValue;
	}

	/**
	* Set the vertex value
	*
	* @param vertexValue
	*/
	public final void setVertexValue(V vertexValue) {
	this.vertexValue = vertexValue;
	this.updated = true;
	}

	/***
	* Send a message to a specific vertex
	*
	* @param id
	* the receiver vertex id
	* @param msg
	* the message
	*/
	public final void sendMsg(I id, M msg) {
	if (msg == null) {
	throw new IllegalArgumentException(
	"sendMsg: Cannot send null message to " + id);
	}
	delegate.sendMsg(id, msg);
	this.hasMessage = true;
	}

	/**
	* Send a message to all direct outgoing neighbors
	*
	* @param msg
	* the message
	*/
	public final void sendMsgToAllEdges(M msg) {
	if (msg == null) {
	throw new IllegalArgumentException(
	"sendMsgToAllEdges: Cannot send null message to all edges");
	}
	for (Edge<I, E> edge : destEdgeList) {
	sendMsg(edge.getDestVertexId(), msg);
	}
	}

	/**
	* Vote to halt. Once all vertex vote to halt and no more messages, a
	* Pregelix job will terminate.
	*/
	public final void voteToHalt() {
	halt = true;
	updated = true;
	}

	/**
	* @return the vertex is halted (true) or not (false)
	*/
	public final boolean isHalted() {
	return halt;
	}

	@Override
	final public void readFields(DataInput in) throws IOException {
	reset();
	if (vertexId == null)
	vertexId = BspUtils.<I> createVertexIndex(getContext()
	.getConfiguration());
	vertexId.readFields(in);
	delegate.setVertexId(vertexId);
	boolean hasVertexValue = in.readBoolean();

	if (hasVertexValue) {
	vertexValue = allocateValue();
	vertexValue.readFields(in);
	delegate.setVertex(this);
	}
	destEdgeList.clear();
	long edgeMapSize = SerDeUtils.readVLong(in);
	for (long i = 0; i < edgeMapSize; ++i) {
	Edge<I, E> edge = allocateEdge();
	edge.setConf(getContext().getConfiguration());
	edge.readFields(in);
	addEdge(edge);
	}
	msgList.clear();
	long msgListSize = SerDeUtils.readVLong(in);
	for (long i = 0; i < msgListSize; ++i) {
	M msg = allocateMessage();
	msg.readFields(in);
	msgList.add(msg);
	}
	halt = in.readBoolean();
	updated = false;
	hasMessage = false;
	createdNewLiveVertex = false;
	}

	@Override
	public void write(DataOutput out) throws IOException {
	vertexId.write(out);
	out.writeBoolean(vertexValue != null);
	if (vertexValue != null) {
	vertexValue.write(out);
	}
	SerDeUtils.writeVLong(out, destEdgeList.size());
	for (Edge<I, E> edge : destEdgeList) {
	edge.write(out);
	}
	SerDeUtils.writeVLong(out, msgList.size());
	for (M msg : msgList) {
	msg.write(out);
	}
	out.writeBoolean(halt);
	}

	/**
	* Get the list of incoming messages
	*
	* @return the list of messages
	*/
	public List<M> getMsgList() {
	return msgList;
	}

	/**
	* Get outgoing edge list
	*
	* @return a list of outgoing edges
	*/
	public List<Edge<I, E>> getEdges() {
	return this.destEdgeList;
	}

	@Override
	@SuppressWarnings("unchecked")
	public String toString() {
	Collections.sort(destEdgeList);
	StringBuffer edgeBuffer = new StringBuffer();
	edgeBuffer.append("(");
	for (Edge<I, E> edge : destEdgeList) {
	edgeBuffer.append(edge.getDestVertexId()).append(",");
	}
	edgeBuffer.append(")");
	return "Vertex(id=" + getVertexId() + ",value=" + getVertexValue()
	+ ", edges=" + edgeBuffer + ")";
	}

	/**
	* Get the number of outgoing edges
	*
	* @return the number of outging edges
	*/
	public int getNumOutEdges() {
	return destEdgeList.size();
	}

	/**
	* Pregelix internal use only
	*
	* @param writers
	*/
	public void setOutputWriters(List<IFrameWriter> writers) {
	delegate.setOutputWriters(writers);
	}

	/**
	* Pregelix internal use only
	*
	* @param writers
	*/
	public void setOutputAppenders(List<FrameTupleAppender> appenders) {
	delegate.setOutputAppenders(appenders);
	}

	/**
	* Pregelix internal use only
	*
	* @param writers
	*/
	public void setOutputTupleBuilders(List<ArrayTupleBuilder> tbs) {
	delegate.setOutputTupleBuilders(tbs);
	}

	/**
	* Pregelix internal use only
	*
	* @param writers
	*/
	public void finishCompute() throws IOException {
	delegate.finishCompute();
	}

	/**
	* Pregelix internal use only
	*/
	public boolean hasUpdate() {
	return this.updated;
	}

	/**
	* Pregelix internal use only
	*/
	public boolean hasMessage() {
	return this.hasMessage;
	}

	/**
	* Pregelix internal use only
	*/
	public boolean createdNewLiveVertex() {
	return this.createdNewLiveVertex;
	}

	/**
	* sort the edges
	*/
	@SuppressWarnings("unchecked")
	public void sortEdges() {
	updated = true;
	Collections.sort(destEdgeList);
	}

	/**
	* Allocate a new edge from the edge pool
	*/
	private Edge<I, E> allocateEdge() {
	Edge<I, E> edge;
	if (usedEdge < edgePool.size()) {
	edge = edgePool.get(usedEdge);
	usedEdge++;
	} else {
	edge = new Edge<I, E>();
	edgePool.add(edge);
	usedEdge++;
	}
	return edge;
	}

	/**
	* Allocate a new message from the message pool
	*/
	private M allocateMessage() {
	M message;
	if (usedMessage < msgPool.size()) {
	message = msgPool.get(usedEdge);
	usedMessage++;
	} else {
	message = BspUtils.<M> createMessageValue(getContext()
	.getConfiguration());
	msgPool.add(message);
	usedMessage++;
	}
	return message;
	}

	/**
	* Set the global superstep for all the vertices (internal use)
	*
	* @param superstep
	* New superstep
	*/
	public static final void setSuperstep(long superstep) {
	Vertex.superstep = superstep;
	}

	/**
	* Add an outgoing edge into the vertex
	*
	* @param edge
	* the edge to be added
	* @return true if the edge list changed as a result of this call
	*/
	public boolean addEdge(Edge<I, E> edge) {
	edge.setConf(getContext().getConfiguration());
	updated = true;
	return destEdgeList.add(edge);
	}

	/**
	* remove an outgoing edge in the graph
	*
	* @param edge
	* the edge to be removed
	* @return true if the edge is in the edge list of the vertex
	*/
	public boolean removeEdge(Edge<I, E> edge) {
	updated = true;
	return destEdgeList.remove(edge);
	}

	/**
	* Add a new vertex into the graph
	*
	* @param vertexId
	* the vertex id
	* @param vertex
	* the vertex
	*/
	public final void addVertex(I vertexId, Vertex vertex) {
	createdNewLiveVertex \|= !vertex.isHalted();
	delegate.addVertex(vertexId, vertex);
	}

	/**
	* Delete a vertex from id
	*
	* @param vertexId
	* the vertex id
	*/
	public final void deleteVertex(I vertexId) {
	delegate.deleteVertex(vertexId);
	}

	/**
	* Allocate a vertex value from the object pool
	*
	* @return a vertex value instance
	*/
	private V allocateValue() {
	V value;
	if (usedValue < valuePool.size()) {
	value = valuePool.get(usedValue);
	usedValue++;
	} else {
	value = BspUtils.<V> createVertexValue(getContext()
	.getConfiguration());
	valuePool.add(value);
	usedValue++;
	}
	return value;
	}

	/**
	* Get the current global superstep number
	*
	* @return the current superstep number
	*/
	public static final long getSuperstep() {
	return superstep;
	}

	/**
	* Set the total number of vertices from the last superstep.
	*
	* @param numVertices
	* Aggregate vertices in the last superstep
	*/
	public static final void setNumVertices(long numVertices) {
	Vertex.numVertices = numVertices;
	}

	/**
	* Get the number of vertexes in the graph
	*
	* @return the number of vertexes in the graph
	*/
	public static final long getNumVertices() {
	return numVertices;
	}

	/**
	* Set the total number of edges from the last superstep.
	*
	* @param numEdges
	* Aggregate edges in the last superstep
	*/
	public static void setNumEdges(long numEdges) {
	Vertex.numEdges = numEdges;
	}

	/**
	* Get the number of edges from this graph
	*
	* @return the number of edges in the graph
	*/
	public static final long getNumEdges() {
	return numEdges;
	}

	/**
	* Pregelix internal use only
	*/
	public static final TaskAttemptContext getContext() {
	return context;
	}

	/**
	* Pregelix internal use only
	*
	* @param context
	*/
	public static final void setContext(TaskAttemptContext context) {
	Vertex.context = context;
	}

	}