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

 /*
  * Copyright 2009-2013 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.conf.Configuration;
 import org.apache.hadoop.fs.FSDataInputStream;
 import org.apache.hadoop.fs.FSDataOutputStream;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.io.Writable;
 import org.apache.hadoop.io.WritableComparable;
 import org.apache.hadoop.io.WritableUtils;
 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.io.WritableSizable;
 import edu.uci.ics.pregelix.api.util.BspUtils;
 import edu.uci.ics.pregelix.api.util.JobStateUtils;

 /**
  * 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 WritableSizable>
         implements Writable {
     /** task context, only used in scanners */
     public static TaskAttemptContext taskContext;
     /** vertex context */
     private VertexContext context;
     /** 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>();
     /** a delegate for hyracks stuff */
     private final 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;
     /** whether the vertex has spilled to HDFS */
     private boolean spilled = false;
     /** the spilled path */
     private String pathStr;
     /** file system handle */
     private FileSystem dfs;
     /** created new vertex */
     private boolean createdNewLiveVertex = false;
     /** terminate the partition */
     private boolean terminatePartition = false;

     /**
      * use object pool for re-using objects
      */
     private final List<Edge<I, E>> edgePool = new ArrayList<Edge<I, E>>();
     private final List<M> msgPool = new ArrayList<M>();
     private final 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 to process
      * incoming messages in each superstep.
      * 1. In a superstep, this method can be called multiple times in a continuous manner for a single
      * vertex, each of which is to process a batch of messages. (Note that
      * this only happens for the case when the mssages for a single vertex
      * exceed one frame.)
      * 2. In each superstep, before any invocation of this method for a vertex,
      * open() is called; after all the invocations of this method for the vertex,
      * close is called.
      * 3. In each partition, the vertex Java object is reused
      * for all the vertice to be processed in the same partition. (The model
      * is the same as the key-value objects in hadoop map tasks.)
      *
      * @param msgIterator
      *            an iterator of incoming messages
      */
     public abstract void compute(Iterator<M> msgIterator) throws Exception;

     /**
      * 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;
         spilled = 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.
      * The state of the current vertex value is saved.
      */
     public final void voteToHalt() {
         halt = true;
         updated = true;
     }

     /**
      * Vote to halt. Once all vertex vote to halt and no more messages, a
      * Pregelix job will terminate.
      *
      * @param update
      *            whether or not to save the vertex value
      */
     public final void voteToHalt(boolean update) {
         halt = true;
         updated = update;
     }

     /**
      * Activate a halted vertex such that it is alive again.
      * The state of the current vertex value is saved.
      */
     public final void activate() {
         halt = false;
         updated = true;
     }

     /**
      * Activate a halted vertex such that it is alive again.
      *
      * @param update
      *            whether or not to save the vertex value
      */
     public final void activate(boolean update) {
         halt = false;
         updated = update;
     }

     /**
      * @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());
         }

         //read the boolean state
         byte booleanState = in.readByte();
         halt = (booleanState & 1) == 1 ? true : false;
         spilled = (booleanState & 2) == 2 ? true : false;

         if (!spilled) {
             //read the vertex data
             readVertexData(in);
         } else {
             //read file hosted vertex
             if (dfs == null) {
                 dfs = FileSystem.get(getContext().getConfiguration());
             }
             pathStr = in.readUTF();
             Path path = new Path(pathStr);
             FSDataInputStream dataInput = dfs.open(path);
             readVertexData(dataInput);
             dataInput.close();
         }
     }

     /**
      * Read B-tree stored vertex
      *
      * @param in
      * @throws IOException
      */
     private void readVertexData(DataInput in) throws IOException {
         vertexId.readFields(in);
         delegate.setVertexId(vertexId);
         boolean hasVertexValue = in.readBoolean();

         if (hasVertexValue) {
             vertexValue = allocateValue();
             vertexValue.readFields(in);
             delegate.setVertex(this);
         }
         destEdgeList.clear();
         long edgeMapSize = WritableUtils.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 = WritableUtils.readVLong(in);
         for (long i = 0; i < msgListSize; ++i) {
             M msg = allocateMessage();
             msg.readFields(in);
             msgList.add(msg);
         }
         updated = false;
         hasMessage = false;
         createdNewLiveVertex = false;
     }

     @Override
     public void write(DataOutput out) throws IOException {
         //write boolean states
         int haltBit = halt ? 1 : 0;
         int spilledBit = spilled ? 2 : 0;
         byte booleanStates = (byte) (haltBit | spilledBit);
         out.writeByte(booleanStates);
         if (!spilled) {
             // write B-tree stored vertex data
             writeVertexData(out);
         } else {
             //write file-hosted vertex
             if (dfs == null) {
                 dfs = FileSystem.get(getContext().getConfiguration());
             }
             Path path = new Path(pathStr);
             if (dfs.exists(path)) {
                 dfs.delete(path, true);
             }
             FSDataOutputStream dataOutput = dfs.create(path, true);
             writeVertexData(dataOutput);
             dataOutput.flush();
             dataOutput.close();
             // write the btree content
             pathStr = path.toUri().toString();
             out.writeUTF(pathStr);
         }
     }

     /**
      * @param out
      * @throws IOException
      */
     private void writeVertexData(DataOutput out) throws IOException {
         vertexId.write(out);
         out.writeBoolean(vertexValue != null);
         if (vertexValue != null) {
             vertexValue.write(out);
         }
         WritableUtils.writeVLong(out, destEdgeList.size());
         for (Edge<I, E> edge : destEdgeList) {
             edge.write(out);
         }
         WritableUtils.writeVLong(out, msgList.size());
         for (M msg : msgList) {
             msg.write(out);
         }
     }

     /**
      * 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
     public String toString() {
         return getStringRepresetation();
     }

     @SuppressWarnings("unchecked")
     public String getStringRepresetation() {
         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);
     }

     /**
      * set the HDFS spilled bit and path
      *
      * @param spilledHDFS
      */
     public void setSpilled(String path) {
         this.spilled = true;
         this.pathStr = path;
     }

     /**
      * set the spilled bit to false and the path to null
      */
     public void setUnSpilled() {
         this.spilled = false;
         this.pathStr = null;
     }

     /**
      * 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(usedMessage);
             usedMessage++;
         } else {
             message = BspUtils.<M> createMessageValue(getContext().getConfiguration());
             msgPool.add(message);
             usedMessage++;
         }
         return message;
     }

     /**
      * 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 final long getSuperstep() {
         return context.getSuperstep();
     }

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

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

     /**
      * Pregelix internal use only
      */
     public final TaskAttemptContext getContext() {
         if (context != null) {
             return context.getContext();
         } else {
             return taskContext;
         }
     }

     @Override
     public int hashCode() {
         return vertexId.hashCode();
     }

     @Override
     public boolean equals(Object object) {
         Vertex vertex = (Vertex) object;
         return vertexId.equals(vertex.getVertexId());
     }

     /**
      * called *once* per partition at the start of each iteration,
      * before calls to open() or compute()
      * Users can override this method to configure the pregelix job
      * and vertex state.
      */
     public void configure(Configuration conf) {

     }

     /**
      * called *once* per partition at the end of each iteration,
      * before calls to compute() or close()
      * Users can override this method to configure the pregelix job
      * and vertex state.
      */
     public void endSuperstep(Configuration conf) {

     }

     /**
      * called immediately before invocations of compute() on a vertex
      * Users can override this method to initiate the state for a vertex
      * before the compute() invocations
      */
     public void open() {

     }

     /**
      * called immediately after all the invocations of compute() on a vertex
      * Users can override this method to initiate the state for a vertex
      * before the compute() invocations
      */
     public void close() {

     }

     /**
      * Terminate the current partition where the current vertex stays in.
      * This will immediately take effect and the upcoming vertice in the
      * same partition cannot be processed.
      */
     protected final void terminatePartition() {
         voteToHalt();
         terminatePartition = true;
     }

     /**
      * Terminate the Pregelix job.
      * This will take effect only when the current iteration completed.
      *
      * @throws Exception
      */
     protected void terminateJob() throws Exception {
         Configuration conf = getContext().getConfiguration();
         JobStateUtils.writeForceTerminationState(conf, BspUtils.getJobId(conf));
     }

     /***
      * @return true if the partition is terminated; false otherwise
      */
     public boolean isPartitionTerminated() {
         return terminatePartition;
     }

     /**
      * Set the vertex context
      *
      * @param ctx
      */
     public void setVertexContext(VertexContext ctx) {
         this.context = ctx;
     }

     /***
      * Get the vertex context
      *
      * @return the vertex context
      */
     public VertexContext getVertexContext() {
         return this.context;
     }

 }
	/*
	* Copyright 2009-2013 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.conf.Configuration;
	import org.apache.hadoop.fs.FSDataInputStream;
	import org.apache.hadoop.fs.FSDataOutputStream;
	import org.apache.hadoop.fs.FileSystem;
	import org.apache.hadoop.fs.Path;
	import org.apache.hadoop.io.Writable;
	import org.apache.hadoop.io.WritableComparable;
	import org.apache.hadoop.io.WritableUtils;
	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.io.WritableSizable;
	import edu.uci.ics.pregelix.api.util.BspUtils;
	import edu.uci.ics.pregelix.api.util.JobStateUtils;

	/**
	* 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 WritableSizable>
	implements Writable {
	/** task context, only used in scanners */
	public static TaskAttemptContext taskContext;
	/** vertex context */
	private VertexContext context;
	/** 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>();
	/** a delegate for hyracks stuff */
	private final 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;
	/** whether the vertex has spilled to HDFS */
	private boolean spilled = false;
	/** the spilled path */
	private String pathStr;
	/** file system handle */
	private FileSystem dfs;
	/** created new vertex */
	private boolean createdNewLiveVertex = false;
	/** terminate the partition */
	private boolean terminatePartition = false;

	/**
	* use object pool for re-using objects
	*/
	private final List<Edge<I, E>> edgePool = new ArrayList<Edge<I, E>>();
	private final List<M> msgPool = new ArrayList<M>();
	private final 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 to process
	* incoming messages in each superstep.
	* 1. In a superstep, this method can be called multiple times in a continuous manner for a single
	* vertex, each of which is to process a batch of messages. (Note that
	* this only happens for the case when the mssages for a single vertex
	* exceed one frame.)
	* 2. In each superstep, before any invocation of this method for a vertex,
	* open() is called; after all the invocations of this method for the vertex,
	* close is called.
	* 3. In each partition, the vertex Java object is reused
	* for all the vertice to be processed in the same partition. (The model
	* is the same as the key-value objects in hadoop map tasks.)
	*
	* @param msgIterator
	* an iterator of incoming messages
	*/
	public abstract void compute(Iterator<M> msgIterator) throws Exception;

	/**
	* 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;
	spilled = 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.
	* The state of the current vertex value is saved.
	*/
	public final void voteToHalt() {
	halt = true;
	updated = true;
	}

	/**
	* Vote to halt. Once all vertex vote to halt and no more messages, a
	* Pregelix job will terminate.
	*
	* @param update
	* whether or not to save the vertex value
	*/
	public final void voteToHalt(boolean update) {
	halt = true;
	updated = update;
	}

	/**
	* Activate a halted vertex such that it is alive again.
	* The state of the current vertex value is saved.
	*/
	public final void activate() {
	halt = false;
	updated = true;
	}

	/**
	* Activate a halted vertex such that it is alive again.
	*
	* @param update
	* whether or not to save the vertex value
	*/
	public final void activate(boolean update) {
	halt = false;
	updated = update;
	}

	/**
	* @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());
	}

	//read the boolean state
	byte booleanState = in.readByte();
	halt = (booleanState & 1) == 1 ? true : false;
	spilled = (booleanState & 2) == 2 ? true : false;

	if (!spilled) {
	//read the vertex data
	readVertexData(in);
	} else {
	//read file hosted vertex
	if (dfs == null) {
	dfs = FileSystem.get(getContext().getConfiguration());
	}
	pathStr = in.readUTF();
	Path path = new Path(pathStr);
	FSDataInputStream dataInput = dfs.open(path);
	readVertexData(dataInput);
	dataInput.close();
	}
	}

	/**
	* Read B-tree stored vertex
	*
	* @param in
	* @throws IOException
	*/
	private void readVertexData(DataInput in) throws IOException {
	vertexId.readFields(in);
	delegate.setVertexId(vertexId);
	boolean hasVertexValue = in.readBoolean();

	if (hasVertexValue) {
	vertexValue = allocateValue();
	vertexValue.readFields(in);
	delegate.setVertex(this);
	}
	destEdgeList.clear();
	long edgeMapSize = WritableUtils.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 = WritableUtils.readVLong(in);
	for (long i = 0; i < msgListSize; ++i) {
	M msg = allocateMessage();
	msg.readFields(in);
	msgList.add(msg);
	}
	updated = false;
	hasMessage = false;
	createdNewLiveVertex = false;
	}

	@Override
	public void write(DataOutput out) throws IOException {
	//write boolean states
	int haltBit = halt ? 1 : 0;
	int spilledBit = spilled ? 2 : 0;
	byte booleanStates = (byte) (haltBit \| spilledBit);
	out.writeByte(booleanStates);
	if (!spilled) {
	// write B-tree stored vertex data
	writeVertexData(out);
	} else {
	//write file-hosted vertex
	if (dfs == null) {
	dfs = FileSystem.get(getContext().getConfiguration());
	}
	Path path = new Path(pathStr);
	if (dfs.exists(path)) {
	dfs.delete(path, true);
	}
	FSDataOutputStream dataOutput = dfs.create(path, true);
	writeVertexData(dataOutput);
	dataOutput.flush();
	dataOutput.close();
	// write the btree content
	pathStr = path.toUri().toString();
	out.writeUTF(pathStr);
	}
	}

	/**
	* @param out
	* @throws IOException
	*/
	private void writeVertexData(DataOutput out) throws IOException {
	vertexId.write(out);
	out.writeBoolean(vertexValue != null);
	if (vertexValue != null) {
	vertexValue.write(out);
	}
	WritableUtils.writeVLong(out, destEdgeList.size());
	for (Edge<I, E> edge : destEdgeList) {
	edge.write(out);
	}
	WritableUtils.writeVLong(out, msgList.size());
	for (M msg : msgList) {
	msg.write(out);
	}
	}

	/**
	* 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
	public String toString() {
	return getStringRepresetation();
	}

	@SuppressWarnings("unchecked")
	public String getStringRepresetation() {
	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);
	}

	/**
	* set the HDFS spilled bit and path
	*
	* @param spilledHDFS
	*/
	public void setSpilled(String path) {
	this.spilled = true;
	this.pathStr = path;
	}

	/**
	* set the spilled bit to false and the path to null
	*/
	public void setUnSpilled() {
	this.spilled = false;
	this.pathStr = null;
	}

	/**
	* 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(usedMessage);
	usedMessage++;
	} else {
	message = BspUtils.<M> createMessageValue(getContext().getConfiguration());
	msgPool.add(message);
	usedMessage++;
	}
	return message;
	}

	/**
	* 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 final long getSuperstep() {
	return context.getSuperstep();
	}

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

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

	/**
	* Pregelix internal use only
	*/
	public final TaskAttemptContext getContext() {
	if (context != null) {
	return context.getContext();
	} else {
	return taskContext;
	}
	}

	@Override
	public int hashCode() {
	return vertexId.hashCode();
	}

	@Override
	public boolean equals(Object object) {
	Vertex vertex = (Vertex) object;
	return vertexId.equals(vertex.getVertexId());
	}

	/**
	* called once per partition at the start of each iteration,
	* before calls to open() or compute()
	* Users can override this method to configure the pregelix job
	* and vertex state.
	*/
	public void configure(Configuration conf) {

	}

	/**
	* called once per partition at the end of each iteration,
	* before calls to compute() or close()
	* Users can override this method to configure the pregelix job
	* and vertex state.
	*/
	public void endSuperstep(Configuration conf) {

	}

	/**
	* called immediately before invocations of compute() on a vertex
	* Users can override this method to initiate the state for a vertex
	* before the compute() invocations
	*/
	public void open() {

	}

	/**
	* called immediately after all the invocations of compute() on a vertex
	* Users can override this method to initiate the state for a vertex
	* before the compute() invocations
	*/
	public void close() {

	}

	/**
	* Terminate the current partition where the current vertex stays in.
	* This will immediately take effect and the upcoming vertice in the
	* same partition cannot be processed.
	*/
	protected final void terminatePartition() {
	voteToHalt();
	terminatePartition = true;
	}

	/**
	* Terminate the Pregelix job.
	* This will take effect only when the current iteration completed.
	*
	* @throws Exception
	*/
	protected void terminateJob() throws Exception {
	Configuration conf = getContext().getConfiguration();
	JobStateUtils.writeForceTerminationState(conf, BspUtils.getJobId(conf));
	}

	/***
	* @return true if the partition is terminated; false otherwise
	*/
	public boolean isPartitionTerminated() {
	return terminatePartition;
	}

	/**
	* Set the vertex context
	*
	* @param ctx
	*/
	public void setVertexContext(VertexContext ctx) {
	this.context = ctx;
	}

	/***
	* Get the vertex context
	*
	* @return the vertex context
	*/
	public VertexContext getVertexContext() {
	return this.context;
	}

	}