Menu
▾
▴
[Bigdata-commit] SF.net SVN: bigdata:[7839] branches/RDR
|
From: <tho...@us...> - 2014-02-03 12:57:45
|
Revision: 7839
http://bigdata.svn.sourceforge.net/bigdata/?rev=7839&view=rev
Author: thompsonbry
Date: 2014-02-03 12:57:35 +0000 (Mon, 03 Feb 2014)
Log Message:
-----------
Added a "Fuzzy SSSP" algorithm. This is not quite finished. It does not actually extract the shortest paths. The SSSP algorithm currently labels the vertices with the minimum distance rather than the predecessor. I am going to talk with Zhisong about how to best captured the predecessor, whether to capture both, and what is involved in supporting a push-style scatter operation.
The FuzzySSSP does show how to break out of the BFS if the visited set size exceeds some threashold at the end of a round. You need to look at the set of active vertices, not the frontier. The frontier is just the set of vertices to be visited in that round. The set of active vertices is all vertices that have been visited to date (all vertices for which state has been materialized).
Modified Paths:
--------------
branches/RDR/bigdata-gas/src/java/com/bigdata/rdf/graph/IGASState.java
branches/RDR/bigdata-gas/src/java/com/bigdata/rdf/graph/analytics/BFS.java
branches/RDR/bigdata-gas/src/java/com/bigdata/rdf/graph/analytics/SSSP.java
branches/RDR/bigdata-gas/src/java/com/bigdata/rdf/graph/impl/GASState.java
branches/RDR/bigdata-rdf/src/test/com/bigdata/rdf/graph/impl/bd/TestBFS.java
Added Paths:
-----------
branches/RDR/bigdata-gas/src/java/com/bigdata/rdf/graph/analytics/FuzzySSSP.java
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-01-31 18:37:34 UTC (rev 7838)
+++ branches/RDR/bigdata-gas/src/java/com/bigdata/rdf/graph/IGASState.java 2014-02-03 12:57:35 UTC (rev 7839)
@@ -92,7 +92,7 @@
/**
* Get the state for the edge using the appropriate factory. If this is the
- * first visit for that edge, then the state is initialized using the
+ * first visit for that vertex, then the state is initialized using the
* factory. Otherwise the existing state is returned.
*
* @param v
@@ -105,6 +105,18 @@
ES getState(Statement e);
/**
+ * Return <code>true</code> iff the specified vertex has an associated
+ * vertex state object - this is interpreted as meaning that the vertex has
+ * been "visited".
+ *
+ * @param v
+ * The vertex.
+ * @return <code>true</code> iff there is vertex state associated with that
+ * vertex.
+ */
+ boolean isVisited(Value v);
+
+ /**
* The current frontier.
*/
IStaticFrontier frontier();
@@ -243,5 +255,5 @@
* Another vertex.
*/
int compareTo(Value u, Value v);
-
+
}
Modified: branches/RDR/bigdata-gas/src/java/com/bigdata/rdf/graph/analytics/BFS.java
===================================================================
--- branches/RDR/bigdata-gas/src/java/com/bigdata/rdf/graph/analytics/BFS.java 2014-01-31 18:37:34 UTC (rev 7838)
+++ branches/RDR/bigdata-gas/src/java/com/bigdata/rdf/graph/analytics/BFS.java 2014-02-03 12:57:35 UTC (rev 7839)
@@ -260,7 +260,7 @@
}
/**
- * Reduce the active vertex stat, returning a histogram reporting the #of
+ * Reduce the active vertex state, returning a histogram reporting the #of
* vertices at each distance from the starting vertex. There will always be
* one vertex at depth zero - this is the starting vertex. For each
* successive depth, the #of vertices that were labeled at that depth is
Added: branches/RDR/bigdata-gas/src/java/com/bigdata/rdf/graph/analytics/FuzzySSSP.java
===================================================================
--- branches/RDR/bigdata-gas/src/java/com/bigdata/rdf/graph/analytics/FuzzySSSP.java (rev 0)
+++ branches/RDR/bigdata-gas/src/java/com/bigdata/rdf/graph/analytics/FuzzySSSP.java 2014-02-03 12:57:35 UTC (rev 7839)
@@ -0,0 +1,420 @@
+/**
+ Copyright (C) SYSTAP, LLC 2006-2012. All rights reserved.
+
+ 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 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 com.bigdata.rdf.graph.analytics;
+
+import java.util.Iterator;
+import java.util.LinkedHashSet;
+import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+
+import org.openrdf.model.Value;
+
+import com.bigdata.rdf.graph.IGASContext;
+import com.bigdata.rdf.graph.IGASEngine;
+import com.bigdata.rdf.graph.IGASState;
+import com.bigdata.rdf.graph.IGraphAccessor;
+import com.bigdata.rdf.graph.IStaticFrontier;
+import com.bigdata.rdf.graph.analytics.FuzzySSSP.FuzzySSSPResult;
+import com.bigdata.rdf.graph.impl.bd.BigdataGraphFixture;
+
+/**
+ * This algorithm provides a fuzzy implementation of the shortest paths between
+ * a set of source vertices and a set of target vertices. This can be used to
+ * identify a set of vertices that are close to the shortest paths between those
+ * source and target vertices. For some domains, the resulting set of vertices
+ * can be understood as an "interesting subgraph".
+ * <p>
+ * Problem: We want to find a set of not more than N vertices out of a data set
+ * that are "close" to the shortest path between two sets of vertices.
+ * <p>
+ * Approach: We want to find the set of SP (Shortest Path) vertices that lie
+ * along the shortest path between each source vertex and each target vertex. We
+ * would also like to know whether a source is connected to each target. To do
+ * this, we do NSOURCES SSSP traversals. For each traversal, we note the depth
+ * of each target from each source, and mark the depth as -1 if the target was
+ * not reachable from that source. The vertices along the shortest path to the
+ * target are collected. The sets of collected vertices are merged and
+ * duplicates are removed.
+ * <p>
+ * Finally, we do a BFS starting with all of the vertices in that merged
+ * collection and stopping when we have N vertices, including those along the
+ * shortest paths. This grows the initial set of vertices that lie along the
+ * shortest paths into a broader collection of vertices that are close to that
+ * shortest path.
+ * <p>
+ * Outputs: The N vertices, their distances from the shortest paths (which we
+ * get out of the final BFS), and the distance of each target from each source
+ * along the shortest path (which we get from the per-source SSSP traversals).
+ *
+ * TODO Support breaking out of the analytic as soon as the frontier is known to
+ * contain at least N distinct vertices. Note that for frontier implementations
+ * that allow duplicates, this means that you need to wait for the end of the
+ * iteration to make the decision. We already support a decision point at the
+ * end of each iteration. This would allow us to lift the decision point inside
+ * of the iteration and terminate processing eagerly when the frontier size
+ * exceeds a specified value.
+ *
+ * TODO: Implement unit test with ground truth.
+ *
+ * @author <a href="mailto:tho...@us...">Bryan Thompson</a>
+ */
+public class FuzzySSSP implements Callable<FuzzySSSPResult>{
+
+ /**
+ * The source vertices (there must be at least one).
+ */
+ private final Value[] src;
+ /**
+ * The target vertices (there must be at least one).
+ */
+ private final Value[] tgt;
+ /**
+ * The maximum number of vertices to report (stopping criteria for the BFS
+ * expansion).
+ */
+ private final int N;
+
+ /**
+ * The {@link IGASEngine} used to run the analytics.
+ */
+ private final IGASEngine gasEngine;
+
+ /**
+ * The object used to access the graph.
+ */
+ private final IGraphAccessor graphAccessor;
+
+ /**
+ * Interface for communicating the results back to the caller.
+ *
+ * @author <a href="mailto:tho...@us...">Bryan
+ * Thompson</a>
+ */
+ public class FuzzySSSPResult {
+
+ /**
+ * The reachability map. The keys of the outer map are the source
+ * vertices. The values in the inner maps are the target vertices that
+ * are reachable from a given source vertex (both the key and the value
+ * of the inner map is the target vertex - it is being used as a set).
+ */
+ private ConcurrentMap<Value, ConcurrentMap<Value, Value>> reachable = new ConcurrentHashMap<Value, ConcurrentMap<Value, Value>>();
+
+ /**
+ * The set of visited vertices.
+ */
+ private Set<Value> visited = new LinkedHashSet<Value>();
+
+// private Map<Value,Set<Value>>
+ private boolean addVisited(final Value v) {
+
+ return visited.add(v);
+
+ }
+
+ /**
+ * Assert that the target was reachable from the source.
+ *
+ * @param src
+ * The source.
+ * @param tgt
+ * The target.
+ */
+ private void addReachable(final Value src, final Value tgt) {
+
+ if (src == null)
+ throw new IllegalArgumentException();
+
+ if (tgt == null)
+ throw new IllegalArgumentException();
+
+ ConcurrentMap<Value, Value> tmp = reachable.get(src);
+
+ if (tmp == null) {
+
+ final ConcurrentMap<Value, Value> old = reachable.putIfAbsent(
+ src, tmp = new ConcurrentHashMap<Value, Value>());
+
+ if (old != null) {
+
+ // Lost the data race.
+ tmp = old;
+
+ }
+
+ }
+
+ // add target to the reachability set for that source.
+ tmp.putIfAbsent(tgt, tgt);
+
+ }
+
+ /**
+ * Return the number of visited vertices.
+ */
+ public int getVisitedCount() {
+
+ return visited.size();
+
+ }
+
+ /**
+ * Return <code>true</code> if the given target is reachable by the
+ * given source.
+ *
+ * @param src
+ * The source.
+ * @param tgt
+ * The target.
+ * @return <code>true</code> iff the target is reachable from that
+ * source.
+ */
+ public boolean getReachable(Value src, Value tgt) {
+ throw new UnsupportedOperationException();
+ }
+
+ /**
+ * Return the set of vertices that were discovered by the analytic. This
+ * constitutes an "interesting subgraph". The source and target vertices
+ * will be included in this collection. Each vertex along a shortest
+ * path from a source vertex to each of the target vertices will be
+ * included. Finally, each vertex in the BFS expension of those vertices
+ * will be included up to the maximum specified when the analytic was
+ * run.
+ */
+ public Set<Value> getVisitedVertices() {
+ throw new UnsupportedOperationException();
+ }
+
+ /**
+ * TODO Also show the reachability matrix and perhaps the visited
+ * vertices in level sets.
+ */
+ @Override
+ public String toString() {
+
+ return getClass().getName() + "{nvisited=" + visited.size() + "}";
+
+ }
+
+ } // class FuzzySSSPResult
+
+ /**
+ *
+ * @param src
+ * The source vertices (there must be at least one).
+ * @param tgt
+ * The target vertices (there must be at least one).
+ * @param N
+ * The maximum number of vertices to report (must be positive),
+ * i.e., the stopping criteria for the BFS expansion.
+ * @param gasEngine
+ * The {@link IGASEngine} will be used to execute the analytic.
+ * @param graphAccessor
+ * The object used to access the graph.
+ */
+ public FuzzySSSP(final Value[] src, final Value[] tgt, final int N,
+ final IGASEngine gasEngine, final IGraphAccessor graphAccessor) {
+
+ if (src == null)
+ throw new IllegalArgumentException();
+ if (src.length == 0)
+ throw new IllegalArgumentException();
+ for (Value v : src)
+ if (v == null)
+ throw new IllegalArgumentException();
+ if (tgt == null)
+ throw new IllegalArgumentException();
+ if (tgt.length == 0)
+ throw new IllegalArgumentException();
+ for (Value v : tgt)
+ if (v == null)
+ throw new IllegalArgumentException();
+ if (N <= 0)
+ throw new IllegalArgumentException();
+ if (gasEngine == null)
+ throw new IllegalArgumentException();
+ if (graphAccessor == null)
+ throw new IllegalArgumentException();
+
+ this.src = src;
+ this.tgt = tgt;
+ this.N = N;
+ this.gasEngine = gasEngine;
+ this.graphAccessor = graphAccessor;
+ }
+
+ @Override
+ public FuzzySSSPResult call() throws Exception {
+
+ final FuzzySSSPResult result = new FuzzySSSPResult();
+
+ /*
+ * For each source vertex, do an SSSP pass. This labels all reachable
+ * vertices with their distance from that source vertex. This will also
+ * tell us whether each of the target vertices was reachable from a
+ * given source vertex.
+ *
+ * Each time we do the SSSP for a source vertex, we collect the set of
+ * vertices lying along a shortest path from the source vertex to each
+ * of the target vertices. These collections are combined and will be
+ * used as the starting point for BFS (below).
+ */
+
+ // The set of vertices along a shortest path.
+ final Set<Value> setAll = new LinkedHashSet<Value>();
+
+ for (Value src : this.src) {
+
+ final IGASContext<SSSP.VS, SSSP.ES, Integer> gasContext = gasEngine
+ .newGASContext(graphAccessor, new SSSP());
+
+ final IGASState<SSSP.VS, SSSP.ES, Integer> gasState = gasContext
+ .getGASState();
+
+ // Initialize the frontier.
+ gasState.setFrontier(gasContext, src);
+
+ // Converge.
+ gasContext.call();
+
+ // The set of vertices along a shortest path for this source.
+ final Set<Value> set = new LinkedHashSet<Value>();
+
+ /*
+ * FIXME Extract the vertices on a shortest path.
+ *
+ * Note: This requires either maintaining the predecessor map or
+ * efficiently obtaining it (if this is possible) from the levels.
+ */
+
+ // Extract whether each target vertex is reachable
+ for (Value tgt : this.tgt) {
+ if (gasState.isVisited(tgt)) {
+ // That target was visited for this source.
+ result.addReachable(src, tgt);
+ }
+ }
+
+ // Combine with the vertices from the other sources.
+ setAll.addAll(set);
+
+ }
+
+ /*
+ * BFS.
+ *
+ * We populate the initial frontier with the set of vertices that we
+ * collected above.
+ *
+ * Note: BFS is overridden to halt once we have visited at least N
+ * vertices.
+ */
+ {
+ final IGASContext<BFS.VS, BFS.ES, Void> gasContext = gasEngine
+ .newGASContext(graphAccessor, new BFS() {
+ @Override
+ public boolean nextRound(IGASContext<VS, ES, Void> ctx) {
+ final IStaticFrontier frontier = ctx.getGASState()
+ .frontier();
+ final Iterator<Value> itr = frontier.iterator();
+ while (itr.hasNext()) {
+ final Value v = itr.next();
+ if (result.addVisited(v)
+ && result.getVisitedCount() >= N) {
+ /*
+ * We have reached our threshold during the
+ * BFS expansion.
+ *
+ * Note: Since we are expanding in a breadth
+ * first manner, all vertices discovered
+ * during a given iteration are at the same
+ * distance from the initial set of vertices
+ * collected from the shortest paths.
+ */
+ return false;
+ }
+ }
+ // Inherent the base class behavior.
+ return super.nextRound(ctx);
+ }
+ });
+
+ final IGASState<BFS.VS, BFS.ES, Void> gasState = gasContext
+ .getGASState();
+
+ // Initialize the frontier.
+ for (Value v : setAll) {
+
+ // add to frontier.
+ gasState.setFrontier(gasContext, v);
+
+ // Add to initial visited set.
+ result.addVisited(v);
+
+ }
+
+ // Converge.
+ gasContext.call();
+
+ /*
+ * Note: We extracted the active vertices in each iteration from the
+ * new frontier, so we are done as soon we the BFS terminates.
+ */
+
+ }
+
+ // Return result.
+ return result;
+
+ }
+
+ public static void main(final String[] args) throws Exception {
+
+ final int nthreads = 4;
+
+ final Properties properties = new Properties();
+
+ final BigdataGraphFixture graphFixture = new BigdataGraphFixture(
+ properties);
+
+ final IGASEngine gasEngine = graphFixture.newGASEngine(nthreads);
+
+ try {
+
+ final Value[] src = null;
+ final Value[] tgt = null;
+ final int N = 0;
+
+ final IGraphAccessor graphAccessor = graphFixture
+ .newGraphAccessor(null/* ignored */);
+
+ final FuzzySSSPResult result = new FuzzySSSP(src, tgt, N,
+ gasEngine, graphAccessor).call();
+
+ System.out.println(result);
+
+ } finally {
+
+ gasEngine.shutdownNow();
+
+ }
+ }
+
+}
Modified: branches/RDR/bigdata-gas/src/java/com/bigdata/rdf/graph/analytics/SSSP.java
===================================================================
--- branches/RDR/bigdata-gas/src/java/com/bigdata/rdf/graph/analytics/SSSP.java 2014-01-31 18:37:34 UTC (rev 7838)
+++ branches/RDR/bigdata-gas/src/java/com/bigdata/rdf/graph/analytics/SSSP.java 2014-02-03 12:57:35 UTC (rev 7839)
@@ -44,7 +44,13 @@
* undirected scatter/gather. Add unit test for undirected.
*
* FIXME New SSSP (push style scatter abstraction with new test case
- * based on graph example developed for this)
+ * based on graph example developed for this). Note: The push style
+ * scatter on the GPU is implemented by capturing each (src,edge) pair
+ * as a distint entry in the frontier. This gives us all of the
+ * necessary variety. We then reduce that variety, applying the binary
+ * operator to combine the intermediate results. Finally, an APPLY()
+ * phase is executed to update the state of the distinct vertices in the
+ * frontier.
*
* TODO Add a reducer to report the actual minimum length paths. This is
* similar to a BFS tree, but the path lengths are not integer values so
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-01-31 18:37:34 UTC (rev 7838)
+++ branches/RDR/bigdata-gas/src/java/com/bigdata/rdf/graph/impl/GASState.java 2014-02-03 12:57:35 UTC (rev 7839)
@@ -210,6 +210,13 @@
}
@Override
+ public boolean isVisited(final Value v) {
+
+ return vertexState.get(v) != null;
+
+ }
+
+ @Override
public ES getState(final Statement e) {
if (edgeState == null)
Modified: branches/RDR/bigdata-rdf/src/test/com/bigdata/rdf/graph/impl/bd/TestBFS.java
===================================================================
--- branches/RDR/bigdata-rdf/src/test/com/bigdata/rdf/graph/impl/bd/TestBFS.java 2014-01-31 18:37:34 UTC (rev 7838)
+++ branches/RDR/bigdata-rdf/src/test/com/bigdata/rdf/graph/impl/bd/TestBFS.java 2014-02-03 12:57:35 UTC (rev 7839)
@@ -23,8 +23,6 @@
*/
package com.bigdata.rdf.graph.impl.bd;
-import org.apache.http.util.ExceptionUtils;
-
import com.bigdata.rdf.graph.IGASContext;
import com.bigdata.rdf.graph.IGASEngine;
import com.bigdata.rdf.graph.IGASState;
This was sent by the SourceForge.net collaborative development platform, the world's largest Open Source development site.
|
