Merged fullstack_asterix_stabilization -r 2933:3157
git-svn-id: https://hyracks.googlecode.com/svn/branches/fullstack_hyracks_ioc@3164 123451ca-8445-de46-9d55-352943316053
diff --git a/pregelix-api/pom.xml b/pregelix-api/pom.xml
index f580752..10efa59 100644
--- a/pregelix-api/pom.xml
+++ b/pregelix-api/pom.xml
@@ -21,8 +21,9 @@
<artifactId>maven-compiler-plugin</artifactId>
<version>2.0.2</version>
<configuration>
- <source>1.6</source>
- <target>1.6</target>
+ <source>1.7</source>
+ <target>1.7</target>
+ <fork>true</fork>
</configuration>
</plugin>
<plugin>
diff --git a/pregelix-api/src/main/java/edu/uci/ics/pregelix/api/graph/Vertex.java b/pregelix-api/src/main/java/edu/uci/ics/pregelix/api/graph/Vertex.java
index a8cd3db..3a98fd9 100644
--- a/pregelix-api/src/main/java/edu/uci/ics/pregelix/api/graph/Vertex.java
+++ b/pregelix-api/src/main/java/edu/uci/ics/pregelix/api/graph/Vertex.java
@@ -73,6 +73,8 @@
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
@@ -258,6 +260,7 @@
halt = in.readBoolean();
updated = false;
hasMessage = false;
+ createdNewLiveVertex = false;
}
@Override
@@ -369,6 +372,13 @@
}
/**
+ * Pregelix internal use only
+ */
+ public boolean createdNewLiveVertex() {
+ return this.createdNewLiveVertex;
+ }
+
+ /**
* sort the edges
*/
@SuppressWarnings("unchecked")
@@ -449,7 +459,8 @@
* @param vertex
* the vertex
*/
- public final void addVertex(I vertexId, V vertex) {
+ public final void addVertex(I vertexId, Vertex vertex) {
+ createdNewLiveVertex |= !vertex.isHalted();
delegate.addVertex(vertexId, vertex);
}
diff --git a/pregelix-api/src/main/java/edu/uci/ics/pregelix/api/graph/VertexDelegate.java b/pregelix-api/src/main/java/edu/uci/ics/pregelix/api/graph/VertexDelegate.java
index d949bc5..4b153c1 100644
--- a/pregelix-api/src/main/java/edu/uci/ics/pregelix/api/graph/VertexDelegate.java
+++ b/pregelix-api/src/main/java/edu/uci/ics/pregelix/api/graph/VertexDelegate.java
@@ -105,7 +105,7 @@
this.vertexId = vertexId;
}
- public final void addVertex(I vertexId, V vertex) {
+ public final void addVertex(I vertexId, Vertex vertex) {
try {
insertTb.reset();
DataOutput outputInsert = insertTb.getDataOutput();
@@ -114,6 +114,19 @@
vertex.write(outputInsert);
insertTb.addFieldEndOffset();
FrameTupleUtils.flushTuple(appenderInsert, insertTb, insertWriter);
+
+ /**
+ * push alive when necessary
+ */
+ if (pushAlive && !vertex.isHalted()) {
+ alive.reset();
+ DataOutput outputAlive = alive.getDataOutput();
+ vertexId.write(outputAlive);
+ alive.addFieldEndOffset();
+ dummyMessageList.write(outputAlive);
+ alive.addFieldEndOffset();
+ FrameTupleUtils.flushTuple(appenderAlive, alive, aliveWriter);
+ }
} catch (Exception e) {
throw new IllegalStateException(e);
}