Menu
▾
▴
[Bigdata-commit] SF.net SVN: bigdata:[8035] branches/RDR
|
From: <mrp...@us...> - 2014-04-03 21:17:06
|
Revision: 8035
http://sourceforge.net/p/bigdata/code/8035
Author: mrpersonick
Date: 2014-04-03 21:17:03 +0000 (Thu, 03 Apr 2014)
Log Message:
-----------
added the ability to halt a program once the "target" vertices have been visited, either immediately or a specified number of iterations afterwards
Modified Paths:
--------------
branches/RDR/bigdata-gas/src/java/com/bigdata/rdf/graph/IGASContext.java
branches/RDR/bigdata-gas/src/java/com/bigdata/rdf/graph/IGASState.java
branches/RDR/bigdata-gas/src/java/com/bigdata/rdf/graph/impl/GASContext.java
branches/RDR/bigdata-gas/src/java/com/bigdata/rdf/graph/impl/GASState.java
branches/RDR/bigdata-rdf/src/java/com/bigdata/rdf/graph/impl/bd/GASService.java
Modified: branches/RDR/bigdata-gas/src/java/com/bigdata/rdf/graph/IGASContext.java
===================================================================
--- branches/RDR/bigdata-gas/src/java/com/bigdata/rdf/graph/IGASContext.java 2014-04-03 21:15:33 UTC (rev 8034)
+++ branches/RDR/bigdata-gas/src/java/com/bigdata/rdf/graph/IGASContext.java 2014-04-03 21:17:03 UTC (rev 8035)
@@ -15,6 +15,7 @@
*/
package com.bigdata.rdf.graph;
+import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
@@ -174,7 +175,7 @@
* the attributes for the edges).
*/
void setLinkAttributeType(URI linkType);
-
+
/**
* Set an optional {@link IReducer} that will run after the
* {@link IGASProgram} is terminated. This may be used to extract results
@@ -227,5 +228,35 @@
*/
@Override
IGASStats call() throws Exception;
+
+
+ /**
+ * Set the target vertices for the program (if any).
+ */
+ void setTargetVertices(Value[] targetVertices);
+ /**
+ * Get the target vertices for the program (if any).
+ * @return
+ */
+ Set<Value> getTargetVertices();
+
+ /**
+ * Specify the maximum number of iterations for the algorithm to continue
+ * once all the target vertices have been reached. Default is for the
+ * program to run until completion without regard to whether the target
+ * vertices have been reached or not. A value of ZERO will stop the program
+ * exactly when all target vertices are found in the frontier.
+ *
+ * @param newValue
+ * The maximum number of iterations past the target vertices.
+ */
+ void setMaxIterationsAfterTargets(int newValue);
+
+ /**
+ * Return the maximum number iterations for the algorithm to continue
+ * once all the target vertices have been reached.
+ */
+ int getMaxIterationsAfterTargets();
+
}
\ No newline at end of file
Modified: branches/RDR/bigdata-gas/src/java/com/bigdata/rdf/graph/IGASState.java
===================================================================
--- branches/RDR/bigdata-gas/src/java/com/bigdata/rdf/graph/IGASState.java 2014-04-03 21:15:33 UTC (rev 8034)
+++ branches/RDR/bigdata-gas/src/java/com/bigdata/rdf/graph/IGASState.java 2014-04-03 21:17:03 UTC (rev 8035)
@@ -119,6 +119,18 @@
boolean isVisited(Value v);
/**
+ * Return <code>true</code> iff the specified vertices all have an associated
+ * vertex state object - this is interpreted as meaning that the vertex has
+ * been "visited".
+ *
+ * @param v
+ * The vertices.
+ * @return <code>true</code> iff there is vertex state associated with all
+ * specified vertices.
+ */
+ boolean isVisited(Set<Value> v);
+
+ /**
* The current frontier.
*/
IStaticFrontier frontier();
Modified: branches/RDR/bigdata-gas/src/java/com/bigdata/rdf/graph/impl/GASContext.java
===================================================================
--- branches/RDR/bigdata-gas/src/java/com/bigdata/rdf/graph/impl/GASContext.java 2014-04-03 21:15:33 UTC (rev 8034)
+++ branches/RDR/bigdata-gas/src/java/com/bigdata/rdf/graph/impl/GASContext.java 2014-04-03 21:17:03 UTC (rev 8035)
@@ -15,7 +15,11 @@
*/
package com.bigdata.rdf.graph.impl;
+import java.util.Arrays;
+import java.util.Collections;
import java.util.Iterator;
+import java.util.LinkedHashSet;
+import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
@@ -100,6 +104,20 @@
null);
/**
+ * A collection of target vertices for the program to reach.
+ */
+ private final Set<Value> targetVertices =
+ Collections.synchronizedSet(new LinkedHashSet<Value>());
+
+ /**
+ * The maximum number of iterations after the target vertices have been
+ * reached. Default behavior is to continue on even after the targets have
+ * been reached.
+ */
+ private final AtomicInteger maxIterationsAfterTargets = new AtomicInteger(
+ Integer.MAX_VALUE);
+
+ /**
*
* @param namespace
* The namespace of the graph (KB instance).
@@ -158,6 +176,11 @@
program.before(this);
+ if (log.isTraceEnabled()) {
+ log.trace("# of targets: " + targetVertices.size());
+ log.trace("max iterations after targets: " + maxIterationsAfterTargets.get());
+ }
+
while (!gasState.frontier().isEmpty()) {
/*
@@ -167,7 +190,30 @@
* GASStats.
*/
- if (total.getNRounds() + 1 >= getMaxIterations()) {
+ if (targetVertices.size() > 0 &&
+ getMaxIterationsAfterTargets() < Integer.MAX_VALUE) {
+
+ if (gasState.isVisited(targetVertices)) {
+
+ /*
+ * If we've reached all target vertices then halt the
+ * program N rounds from now where
+ * N = maxIterationsAfterTargets.
+ */
+ this.maxIterations.set(Math.min(getMaxIterations(),
+ (int) total.getNRounds() + getMaxIterationsAfterTargets()));
+
+ if (log.isTraceEnabled()) {
+ log.trace("All targets reached at round " +
+ total.getNRounds() + ", halting at round " +
+ this.maxIterations.get());
+ }
+
+ }
+
+ }
+
+ if (total.getNRounds() + 1 > getMaxIterations()) {
log.warn("Halting: maxIterations=" + getMaxIterations()
+ ", #rounds=" + total.getNRounds());
@@ -882,7 +928,40 @@
this.linkAttributeType.set(linkAttributeType);
}
+
+ @Override
+ public void setTargetVertices(final Value[] targetVertices) {
+
+ this.targetVertices.addAll(Arrays.asList(targetVertices));
+
+ }
+ @Override
+ public Set<Value> getTargetVertices() {
+
+ return this.targetVertices;
+
+ }
+
+ @Override
+ public void setMaxIterationsAfterTargets(final int newValue) {
+
+ if (newValue < 0)
+ throw new IllegalArgumentException();
+
+ this.maxIterationsAfterTargets.set(newValue);
+
+ }
+
+ @Override
+ public int getMaxIterationsAfterTargets() {
+
+ return maxIterationsAfterTargets.get();
+
+ }
+
+
+
// /**
// * {@inheritDoc}
// * <p>
Modified: branches/RDR/bigdata-gas/src/java/com/bigdata/rdf/graph/impl/GASState.java
===================================================================
--- branches/RDR/bigdata-gas/src/java/com/bigdata/rdf/graph/impl/GASState.java 2014-04-03 21:15:33 UTC (rev 8034)
+++ branches/RDR/bigdata-gas/src/java/com/bigdata/rdf/graph/impl/GASState.java 2014-04-03 21:17:03 UTC (rev 8035)
@@ -217,6 +217,13 @@
}
@Override
+ public boolean isVisited(final Set<Value> v) {
+
+ return vertexState.keySet().containsAll(v);
+
+ }
+
+ @Override
public ES getState(final Statement e) {
if (edgeState == null)
@@ -453,4 +460,10 @@
}
+// public Set<Value> values() {
+//
+// return vertexState.keySet();
+//
+// }
+
}
Modified: branches/RDR/bigdata-rdf/src/java/com/bigdata/rdf/graph/impl/bd/GASService.java
===================================================================
--- branches/RDR/bigdata-rdf/src/java/com/bigdata/rdf/graph/impl/bd/GASService.java 2014-04-03 21:15:33 UTC (rev 8034)
+++ branches/RDR/bigdata-rdf/src/java/com/bigdata/rdf/graph/impl/bd/GASService.java 2014-04-03 21:17:03 UTC (rev 8035)
@@ -194,6 +194,19 @@
int DEFAULT_MAX_ITERATIONS = Integer.MAX_VALUE;
/**
+ * The maximum #of iterations for the GAS program after the targets
+ * have been reached (optional, default
+ * {@value #DEFAULT_MAX_ITERATIONS_AFTER_TARGETS}). Default behavior
+ * is to not stop once the targets are reached.
+ *
+ * @see #DEFAULT_MAX_ITERATIONS_AFTER_TARGETS
+ * @see IGASContext#setMaxIterationsAfterTargets(int)
+ */
+ URI MAX_ITERATIONS_AFTER_TARGETS = new URIImpl(NAMESPACE + "maxIterationsAfterTargets");
+
+ int DEFAULT_MAX_ITERATIONS_AFTER_TARGETS = Integer.MAX_VALUE;
+
+ /**
* The maximum #of vertices in the visited set for the GAS program
* (optional, default {@value #DEFAULT_MAX_VISITED}).
*
@@ -406,6 +419,7 @@
private final int nthreads;
private final TraversalDirectionEnum traversalDirection;
private final int maxIterations;
+ private final int maxIterationsAfterTargets;
private final int maxVisited;
private final URI linkType, linkAttrType;
private final Class<IGASProgram<VS, ES, ST>> gasClass;
@@ -452,6 +466,11 @@
.createLiteral(Options.DEFAULT_MAX_ITERATIONS)))
.intValue();
+ this.maxIterationsAfterTargets = ((Literal) getOnlyArg(Options.PROGRAM,
+ Options.MAX_ITERATIONS_AFTER_TARGETS, store.getValueFactory()
+ .createLiteral(Options.DEFAULT_MAX_ITERATIONS_AFTER_TARGETS)))
+ .intValue();
+
this.maxVisited = ((Literal) getOnlyArg(
Options.PROGRAM,
Options.MAX_VISITED,
@@ -774,8 +793,16 @@
gasContext.setMaxIterations(maxIterations);
+ gasContext.setMaxIterationsAfterTargets(maxIterationsAfterTargets);
+
gasContext.setMaxVisited(maxVisited);
+
+ if (targetVertices != null) {
+ gasContext.setTargetVertices(toIV(targetVertices));
+
+ }
+
// Optional link type constraint.
if (linkType != null)
gasContext.setLinkType(linkType);
@@ -803,7 +830,7 @@
gasState.setFrontier(gasContext, tmp);
}
-
+
// Run the analytic.
final IGASStats stats = (IGASStats) gasContext.call();
This was sent by the SourceForge.net collaborative development platform, the world's largest Open Source development site.
|
