src/main/java/science/atlarge/graphalytics/giraph/algorithms/wcc/DirectedWeaklyConnectedComponentsComputation.java
/**
	 * Propagates the smallest vertex id to all neighbors. Will always choose to
	 * halt and only reactivate if a smaller id has been sent to it.
	 */
Propagates the smallest vertex id to all neighbors. Will always choose to
halt and only reactivate if a smaller id has been sent to it.
Propagates the smallest vertex id to all neighbors . Will always choose to halt and only reactivate if a smaller id has been sent to it .
@Override
	public void compute(Vertex<LongWritable, LongWritable, NullWritable> vertex, Iterable<LongWritable> messages) throws IOException {
		if (getSuperstep() == 0) {
			sendMessageToAllEdges(vertex, vertex.getId());
		} else if (getSuperstep() == 1) {
			edgeSet.clear();
			for (Edge<LongWritable, NullWritable> existingEdge : vertex.getEdges()) {
				edgeSet.add(existingEdge.getTargetVertexId().get());
			}
			for (LongWritable incomingId : messages) {
				if (!edgeSet.contains(incomingId.get())) {
					vertex.addEdge(EdgeFactory.create(incomingId));
				}
			}
			long minId = vertex.getId().get();
			for (Edge<LongWritable, NullWritable> edge : vertex.getEdges()) {
				long targetVertexId = edge.getTargetVertexId().get();
				if (targetVertexId < minId) {
					minId = targetVertexId;
				}
			}
			vertex.getValue().set(minId);
			if (minId != vertex.getId().get()) {
				sendMessageToAllEdges(vertex, vertex.getValue());
			}
			vertex.voteToHalt();
		} else {
			long currentComponent = vertex.getValue().get();
			for (LongWritable message : messages) {
				long candidateComponent = message.get();
				if (candidateComponent < currentComponent) {
					currentComponent = candidateComponent;
				}
			}
			if (currentComponent != vertex.getValue().get()) {
				vertex.getValue().set(currentComponent);
				sendMessageToAllEdges(vertex, vertex.getValue());
			}
			vertex.voteToHalt();
		}
	}
@ Override public void compute ( Vertex < LongWritable , LongWritable , NullWritable > vertex , Iterable < LongWritable > messages ) throws IOException { if ( getSuperstep ( ) == 0 ) { sendMessageToAllEdges ( vertex , vertex . getId ( ) ) ; } else if ( getSuperstep ( ) == 1 ) { edgeSet . clear ( ) ; for ( Edge < LongWritable , NullWritable > existingEdge : vertex . getEdges ( ) ) { edgeSet . add ( existingEdge . getTargetVertexId ( ) . get ( ) ) ; } for ( LongWritable incomingId : messages ) { if ( ! edgeSet . contains ( incomingId . get ( ) ) ) { vertex . addEdge ( EdgeFactory . create ( incomingId ) ) ; } } long minId = vertex . getId ( ) . get ( ) ; for ( Edge < LongWritable , NullWritable > edge : vertex . getEdges ( ) ) { long targetVertexId = edge . getTargetVertexId ( ) . get ( ) ; if ( targetVertexId < minId ) { minId = targetVertexId ; } } vertex . getValue ( ) . set ( minId ) ; if ( minId != vertex . getId ( ) . get ( ) ) { sendMessageToAllEdges ( vertex , vertex . getValue ( ) ) ; } vertex . voteToHalt ( ) ; } else { long currentComponent = vertex . getValue ( ) . get ( ) ; for ( LongWritable message : messages ) { long candidateComponent = message . get ( ) ; if ( candidateComponent < currentComponent ) { currentComponent = candidateComponent ; } } if ( currentComponent != vertex . getValue ( ) . get ( ) ) { vertex . getValue ( ) . set ( currentComponent ) ; sendMessageToAllEdges ( vertex , vertex . getValue ( ) ) ; } vertex . voteToHalt ( ) ; } }
Propagates the smallest vertex id to all neighbors.
Propagates the smallest vertex id to all neighbors .
// Weakly connected components algorithm treats a directed graph as undirected, so we create the missing edges // Broadcast own id to notify neighbours of incoming edge // For every incoming edge that does not have a corresponding outgoing edge, create one // Initialize value to minimum id of neighbours // Store the new component id and broadcast it if it is not equal to this vertex's own id // did we get a smaller id ? // propagate new component id to the neighbors