Added an optimized count to VertexSetRDD.

graph/src/main/scala/org/apache/spark/graph/Pregel.scala

@@ -139,8 +139,6 @@ object Pregel {
    * @param initialMsg the message each vertex will receive at the on
    * the first iteration.
    *
-   * @param numIter the number of iterations to run this computation.
-   *
    * @param vprog the user-defined vertex program which runs on each
    * vertex and receives the inbound message and computes a new vertex
    * value.  On the first iteration the vertex program is invoked on

graph/src/main/scala/org/apache/spark/graph/VertexSetRDD.scala

@@ -101,6 +101,11 @@ class VertexSetRDD[@specialized VD: ClassManifest](
   /** Persist this RDD with the default storage level (`MEMORY_ONLY`). */
   override def cache(): VertexSetRDD[VD] = persist()
 
+  /** Return the number of vertices in this set. */
+  override def count(): Long = {
+    partitionsRDD.map(_.size).reduce(_ + _)
+  }
+
   /**
    * Provide the `RDD[(Vid, VD)]` equivalent output.
    */

graph/src/main/scala/org/apache/spark/graph/impl/VertexPartition.scala

@@ -42,6 +42,8 @@ class VertexPartition[@specialized(Long, Int, Double) VD: ClassManifest](
 
   val capacity: Int = index.capacity
 
+  def size: Int = mask.cardinality
+
   /**
    * Pass each vertex attribute along with the vertex id through a map
    * function and retain the original RDD's partitioning and index.