Coverage Report

Coverage Report - org.apache.giraph.examples.SimpleMutateGraphComputation

Classes in this File

Line Coverage

Branch Coverage

Complexity

SimpleMutateGraphComputation

0/80

0/42

3.9

SimpleMutateGraphComputation$SimpleMutateGraphVertexWorkerContext

0/19

0/2

3.9

 /*
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
  * regarding copyright ownership.  The ASF licenses this file
  * to you 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 org.apache.giraph.examples;
 
 import org.apache.giraph.graph.BasicComputation;
 import org.apache.giraph.edge.EdgeFactory;
 import org.apache.giraph.graph.Vertex;
 import org.apache.giraph.worker.WorkerContext;
 import org.apache.hadoop.io.DoubleWritable;
 import org.apache.hadoop.io.FloatWritable;
 import org.apache.hadoop.io.LongWritable;
 import org.apache.log4j.Logger;
 
 import java.io.IOException;
 
 /**
  * Vertex to allow unit testing of graph mutations.
  */
 public class SimpleMutateGraphComputation extends BasicComputation<
     LongWritable, DoubleWritable, FloatWritable, DoubleWritable> {
   /** Class logger */
   private static Logger LOG =
       Logger.getLogger(SimpleMutateGraphComputation.class);
   /** Maximum number of ranges for vertex ids */
   private long maxRanges = 100;
 
 
   /**
    * Unless we create a ridiculous number of vertices , we should not
    * collide within a vertex range defined by this method.
    *
    * @param range Range index
    * @return Starting vertex id of the range
    */
   private long rangeVertexIdStart(int range) {
     return (Long.MAX_VALUE / maxRanges) * range;
   }
 
   @Override
   public void compute(
       Vertex<LongWritable, DoubleWritable, FloatWritable> vertex,
       Iterable<DoubleWritable> messages) throws IOException {
     SimpleMutateGraphVertexWorkerContext workerContext = getWorkerContext();
     if (getSuperstep() == 0) {
       LOG.debug("Reached superstep " + getSuperstep());
     } else if (getSuperstep() == 1) {
       // Send messages to vertices that are sure not to exist
       // (creating them)
       LongWritable destVertexId =
           new LongWritable(rangeVertexIdStart(1) + vertex.getId().get());
       sendMessage(destVertexId, new DoubleWritable(0.0));
     } else if (getSuperstep() == 2) {
       LOG.debug("Reached superstep " + getSuperstep());
     } else if (getSuperstep() == 3) {
       long vertexCount = workerContext.getVertexCount();
       if (vertexCount * 2 != getTotalNumVertices()) {
         throw new IllegalStateException(
             "Impossible to have " + getTotalNumVertices() +
             " vertices when should have " + vertexCount * 2 +
             " on superstep " + getSuperstep());
       }
       long edgeCount = workerContext.getEdgeCount();
       if (edgeCount != getTotalNumEdges()) {
         throw new IllegalStateException(
             "Impossible to have " + getTotalNumEdges() +
             " edges when should have " + edgeCount +
             " on superstep " + getSuperstep());
       }
       // Create vertices that are sure not to exist (doubling vertices)
       LongWritable vertexIndex =
           new LongWritable(rangeVertexIdStart(3) + vertex.getId().get());
       addVertexRequest(vertexIndex, new DoubleWritable(0.0));
       // Add edges to those remote vertices as well
       addEdgeRequest(vertexIndex,
           EdgeFactory.create(vertex.getId(), new FloatWritable(0.0f)));
     } else if (getSuperstep() == 4) {
       LOG.debug("Reached superstep " + getSuperstep());
     } else if (getSuperstep() == 5) {
       long vertexCount = workerContext.getVertexCount();
       if (vertexCount * 2 != getTotalNumVertices()) {
         throw new IllegalStateException(
             "Impossible to have " + getTotalNumVertices() +
             " when should have " + vertexCount * 2 +
             " on superstep " + getSuperstep());
       }
       long edgeCount = workerContext.getEdgeCount();
       if (edgeCount + vertexCount != getTotalNumEdges()) {
         throw new IllegalStateException(
             "Impossible to have " + getTotalNumEdges() +
             " edges when should have " + edgeCount + vertexCount +
             " on superstep " + getSuperstep());
       }
       // Remove the edges created in superstep 3
       LongWritable vertexIndex =
           new LongWritable(rangeVertexIdStart(3) + vertex.getId().get());
       workerContext.increaseEdgesRemoved();
       removeEdgesRequest(vertexIndex, vertex.getId());
     } else if (getSuperstep() == 6) {
       // Remove all the vertices created in superstep 3
       if (vertex.getId().compareTo(
           new LongWritable(rangeVertexIdStart(3))) >= 0) {
         removeVertexRequest(vertex.getId());
       }
     } else if (getSuperstep() == 7) {
       long origEdgeCount = workerContext.getOrigEdgeCount();
       if (origEdgeCount != getTotalNumEdges()) {
         throw new IllegalStateException(
             "Impossible to have " + getTotalNumEdges() +
             " edges when should have " + origEdgeCount +
             " on superstep " + getSuperstep());
       }
     } else if (getSuperstep() == 8) {
       long vertexCount = workerContext.getVertexCount();
       if (vertexCount / 2 != getTotalNumVertices()) {
         throw new IllegalStateException(
             "Impossible to have " + getTotalNumVertices() +
             " vertices when should have " + vertexCount / 2 +
             " on superstep " + getSuperstep());
       }
     } else if (getSuperstep() == 9) {
       // Remove all the vertices created in superstep 1, and send a message to
       // them at the same time
       if (vertex.getId().compareTo(
           new LongWritable(rangeVertexIdStart(1))) >= 0) {
         // This is an added vertex, so remove it
         removeVertexRequest(vertex.getId());
       } else {
         // This is a vertex since the start of the computation, so send a
         // message to a vertex added in superstep 1
         sendMessage(
             new LongWritable(rangeVertexIdStart(1) + vertex.getId().get()),
             new DoubleWritable(0.0));
       }
     } else if (getSuperstep() == 10) {
       LOG.debug("Reached superstep " + getSuperstep());
     } else if (getSuperstep() == 11) {
       long vertexCount = workerContext.getVertexCount();
       if (vertexCount / 2 != getTotalNumVertices()) {
         throw new IllegalStateException(
             "Impossible to have " + getTotalNumVertices() +
                 " vertices when should have " + vertexCount / 2 +
                 " on superstep " + getSuperstep());
       }
     } else {
       vertex.voteToHalt();
     }
   }
 
   /**
    * Worker context used with {@link SimpleMutateGraphComputation}.
    */
   public static class SimpleMutateGraphVertexWorkerContext
       extends WorkerContext {
     /** Cached vertex count */
     private long vertexCount;
     /** Cached edge count */
     private long edgeCount;
     /** Original number of edges */
     private long origEdgeCount;
     /** Number of edges removed during superstep */
     private int edgesRemoved = 0;
 
     @Override
     public void preApplication()
       throws InstantiationException, IllegalAccessException { }
 
     @Override
     public void postApplication() { }
 
     @Override
     public void preSuperstep() { }
 
     @Override
     public void postSuperstep() {
       vertexCount = getTotalNumVertices();
       edgeCount = getTotalNumEdges();
       if (getSuperstep() == 1) {
         origEdgeCount = edgeCount;
       }
       LOG.info("Got " + vertexCount + " vertices, " +
           edgeCount + " edges on superstep " +
           getSuperstep());
       LOG.info("Removed " + edgesRemoved);
       edgesRemoved = 0;
     }
 
     public long getVertexCount() {
       return vertexCount;
     }
 
     public long getEdgeCount() {
       return edgeCount;
     }
 
     public long getOrigEdgeCount() {
       return origEdgeCount;
     }
 
     /**
      * Increase the number of edges removed by one.
      */
     public void increaseEdgesRemoved() {
       this.edgesRemoved++;
     }
   }
 }

1		/*
2		* Licensed to the Apache Software Foundation (ASF) under one
3		* or more contributor license agreements. See the NOTICE file
4		* distributed with this work for additional information
5		* regarding copyright ownership. The ASF licenses this file
6		* to you under the Apache License, Version 2.0 (the
7		* "License"); you may not use this file except in compliance
8		* with the License. You may obtain a copy of the License at
9		*
10		* http://www.apache.org/licenses/LICENSE-2.0
11		*
12		* Unless required by applicable law or agreed to in writing, software
13		* distributed under the License is distributed on an "AS IS" BASIS,
14		* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15		* See the License for the specific language governing permissions and
16		* limitations under the License.
17		*/
18
19		package org.apache.giraph.examples;
20
21		import org.apache.giraph.graph.BasicComputation;
22		import org.apache.giraph.edge.EdgeFactory;
23		import org.apache.giraph.graph.Vertex;
24		import org.apache.giraph.worker.WorkerContext;
25		import org.apache.hadoop.io.DoubleWritable;
26		import org.apache.hadoop.io.FloatWritable;
27		import org.apache.hadoop.io.LongWritable;
28		import org.apache.log4j.Logger;
29
30		import java.io.IOException;
31
32		/**
33		* Vertex to allow unit testing of graph mutations.
34		*/
35	0	public class SimpleMutateGraphComputation extends BasicComputation<
36		LongWritable, DoubleWritable, FloatWritable, DoubleWritable> {
37		/** Class logger */
38	0	private static Logger LOG =
39	0	Logger.getLogger(SimpleMutateGraphComputation.class);
40		/** Maximum number of ranges for vertex ids */
41	0	private long maxRanges = 100;
42
43
44		/**
45		* Unless we create a ridiculous number of vertices , we should not
46		* collide within a vertex range defined by this method.
47		*
48		* @param range Range index
49		* @return Starting vertex id of the range
50		*/
51		private long rangeVertexIdStart(int range) {
52	0	return (Long.MAX_VALUE / maxRanges) * range;
53		}
54
55		@Override
56		public void compute(
57		Vertex<LongWritable, DoubleWritable, FloatWritable> vertex,
58		Iterable<DoubleWritable> messages) throws IOException {
59	0	SimpleMutateGraphVertexWorkerContext workerContext = getWorkerContext();
60	0	if (getSuperstep() == 0) {
61	0	LOG.debug("Reached superstep " + getSuperstep());
62	0	} else if (getSuperstep() == 1) {
63		// Send messages to vertices that are sure not to exist
64		// (creating them)
65	0	LongWritable destVertexId =
66	0	new LongWritable(rangeVertexIdStart(1) + vertex.getId().get());
67	0	sendMessage(destVertexId, new DoubleWritable(0.0));
68	0	} else if (getSuperstep() == 2) {
69	0	LOG.debug("Reached superstep " + getSuperstep());
70	0	} else if (getSuperstep() == 3) {
71	0	long vertexCount = workerContext.getVertexCount();
72	0	if (vertexCount * 2 != getTotalNumVertices()) {
73	0	throw new IllegalStateException(
74	0	"Impossible to have " + getTotalNumVertices() +
75		" vertices when should have " + vertexCount * 2 +
76	0	" on superstep " + getSuperstep());
77		}
78	0	long edgeCount = workerContext.getEdgeCount();
79	0	if (edgeCount != getTotalNumEdges()) {
80	0	throw new IllegalStateException(
81	0	"Impossible to have " + getTotalNumEdges() +
82		" edges when should have " + edgeCount +
83	0	" on superstep " + getSuperstep());
84		}
85		// Create vertices that are sure not to exist (doubling vertices)
86	0	LongWritable vertexIndex =
87	0	new LongWritable(rangeVertexIdStart(3) + vertex.getId().get());
88	0	addVertexRequest(vertexIndex, new DoubleWritable(0.0));
89		// Add edges to those remote vertices as well
90	0	addEdgeRequest(vertexIndex,
91	0	EdgeFactory.create(vertex.getId(), new FloatWritable(0.0f)));
92	0	} else if (getSuperstep() == 4) {
93	0	LOG.debug("Reached superstep " + getSuperstep());
94	0	} else if (getSuperstep() == 5) {
95	0	long vertexCount = workerContext.getVertexCount();
96	0	if (vertexCount * 2 != getTotalNumVertices()) {
97	0	throw new IllegalStateException(
98	0	"Impossible to have " + getTotalNumVertices() +
99		" when should have " + vertexCount * 2 +
100	0	" on superstep " + getSuperstep());
101		}
102	0	long edgeCount = workerContext.getEdgeCount();
103	0	if (edgeCount + vertexCount != getTotalNumEdges()) {
104	0	throw new IllegalStateException(
105	0	"Impossible to have " + getTotalNumEdges() +
106		" edges when should have " + edgeCount + vertexCount +
107	0	" on superstep " + getSuperstep());
108		}
109		// Remove the edges created in superstep 3
110	0	LongWritable vertexIndex =
111	0	new LongWritable(rangeVertexIdStart(3) + vertex.getId().get());
112	0	workerContext.increaseEdgesRemoved();
113	0	removeEdgesRequest(vertexIndex, vertex.getId());
114	0	} else if (getSuperstep() == 6) {
115		// Remove all the vertices created in superstep 3
116	0	if (vertex.getId().compareTo(
117	0	new LongWritable(rangeVertexIdStart(3))) >= 0) {
118	0	removeVertexRequest(vertex.getId());
119		}
120	0	} else if (getSuperstep() == 7) {
121	0	long origEdgeCount = workerContext.getOrigEdgeCount();
122	0	if (origEdgeCount != getTotalNumEdges()) {
123	0	throw new IllegalStateException(
124	0	"Impossible to have " + getTotalNumEdges() +
125		" edges when should have " + origEdgeCount +
126	0	" on superstep " + getSuperstep());
127		}
128	0	} else if (getSuperstep() == 8) {
129	0	long vertexCount = workerContext.getVertexCount();
130	0	if (vertexCount / 2 != getTotalNumVertices()) {
131	0	throw new IllegalStateException(
132	0	"Impossible to have " + getTotalNumVertices() +
133		" vertices when should have " + vertexCount / 2 +
134	0	" on superstep " + getSuperstep());
135		}
136	0	} else if (getSuperstep() == 9) {
137		// Remove all the vertices created in superstep 1, and send a message to
138		// them at the same time
139	0	if (vertex.getId().compareTo(
140	0	new LongWritable(rangeVertexIdStart(1))) >= 0) {
141		// This is an added vertex, so remove it
142	0	removeVertexRequest(vertex.getId());
143		} else {
144		// This is a vertex since the start of the computation, so send a
145		// message to a vertex added in superstep 1
146	0	sendMessage(
147	0	new LongWritable(rangeVertexIdStart(1) + vertex.getId().get()),
148		new DoubleWritable(0.0));
149		}
150	0	} else if (getSuperstep() == 10) {
151	0	LOG.debug("Reached superstep " + getSuperstep());
152	0	} else if (getSuperstep() == 11) {
153	0	long vertexCount = workerContext.getVertexCount();
154	0	if (vertexCount / 2 != getTotalNumVertices()) {
155	0	throw new IllegalStateException(
156	0	"Impossible to have " + getTotalNumVertices() +
157		" vertices when should have " + vertexCount / 2 +
158	0	" on superstep " + getSuperstep());
159		}
160	0	} else {
161	0	vertex.voteToHalt();
162		}
163	0	}
164
165		/**
166		* Worker context used with {@link SimpleMutateGraphComputation}.
167		*/
168	0	public static class SimpleMutateGraphVertexWorkerContext
169		extends WorkerContext {
170		/** Cached vertex count */
171		private long vertexCount;
172		/** Cached edge count */
173		private long edgeCount;
174		/** Original number of edges */
175		private long origEdgeCount;
176		/** Number of edges removed during superstep */
177	0	private int edgesRemoved = 0;
178
179		@Override
180		public void preApplication()
181	0	throws InstantiationException, IllegalAccessException { }
182
183		@Override
184	0	public void postApplication() { }
185
186		@Override
187	0	public void preSuperstep() { }
188
189		@Override
190		public void postSuperstep() {
191	0	vertexCount = getTotalNumVertices();
192	0	edgeCount = getTotalNumEdges();
193	0	if (getSuperstep() == 1) {
194	0	origEdgeCount = edgeCount;
195		}
196	0	LOG.info("Got " + vertexCount + " vertices, " +
197		edgeCount + " edges on superstep " +
198	0	getSuperstep());
199	0	LOG.info("Removed " + edgesRemoved);
200	0	edgesRemoved = 0;
201	0	}
202
203		public long getVertexCount() {
204	0	return vertexCount;
205		}
206
207		public long getEdgeCount() {
208	0	return edgeCount;
209		}
210
211		public long getOrigEdgeCount() {
212	0	return origEdgeCount;
213		}
214
215		/**
216		* Increase the number of edges removed by one.
217		*/
218		public void increaseEdgesRemoved() {
219	0	this.edgesRemoved++;
220	0	}
221		}
222		}