Merge pull request #20 from atrostan/alg-cleanup

Cleanup of algorithm code before project submission

src/main/scala/com/algorithm/LocalMaximaColouring.scala

@@ -7,71 +7,45 @@ case class Colour(num: Int) {
  }
 }
 
-trait LocalMaximalColouringAbstractMode extends VertexProgram[Int, Int, Set[Int], Colour] {
- override def gather(edgeVal: Int, message: Int): Set[Int] = Set(message)
-
- override def sum(a: Set[Int], b: Set[Int]): Set[Int] = a.union(b)
+object Colour {
+ val Blank = Colour(-1)
+}
 
- override def apply(
- superStepNumber: Int,
- thisVertex: VertexInfo,
- oldVal: Colour,
- total: Option[Set[Int]]
- ): Colour = {
- if (superStepNumber == 0) {
- Colour(-1)
- } else {
- oldVal match {
- case Colour(-1) => {
- total match {
- case None => {
- // Colour myself with superstep number
- Colour(superStepNumber - 1)
- }
- case Some(idSet) => {
- if (idSet.max < thisVertex.id) {
- // Colour myself with superstep number
- Colour(superStepNumber - 1)
- } else {
- Colour(-1)
- }
- }
- }
- }
- case c => c
- }
- }
+object LocalMaximaColouring extends VertexProgram[Int, Int, Int, Colour] {
+
+ // Map messages into accumulator values
+ override def gather(edgeVal: Int, message: Int): Int = message
+
+ // Combine accumulator values
+ override def sum(a: Int, b: Int): Int = Math.max(a, b)
+
+ // Compute new local state for this vertex based on accumulated sum
+ override def apply(superStepNumber: Int, thisVertex: VertexInfo, oldVal: Colour, total: Option[Int]): Colour = {
+ // At start, always colour yourself blank
+ if (superStepNumber == 0) Colour.Blank
+ // If you have a colour, keep it
+ else if (oldVal != Colour.Blank) oldVal
+ // If you have no neighbours, or you are bigger than the largest ID, colour yourself
+ else if (total == None || total.get < thisVertex.id) Colour(superStepNumber - 1)
+ // Otherwise, stay blank
+ else Colour.Blank
  }
 
- override def scatter(
- superStepNumber: Int, 
- thisVertex: VertexInfo,
- oldVal: Colour,
- newVal: Colour
- ): Option[Int] = {
- newVal match {
- case Colour(-1) => Some(thisVertex.id)
- case c => None
- }
+ // Optionally generate a message to send to neighbours
+ override def scatter(superStepNumber: Int, thisVertex: VertexInfo, oldVal: Colour, newVal: Colour): Option[Int] = {
+ if (newVal == Colour.Blank) Some(thisVertex.id)
+ else None
  }
 
- override def voteToHalt(superStepNumber: Int, oldVal: Colour, newVal: Colour): Boolean = {
- newVal match {
- case Colour(-1) => false
- case c => true
- }
- }
+ // Whether to deactivate this vertex at the end of this superstep
+ override def deactivateSelf(superStepNumber: Int, oldVal: Colour, newVal: Colour): Boolean = (newVal != Colour.Blank)
 
- override val defaultVertexValue: Colour = Colour(-1)
+ // Initial starting state
+ override val defaultVertexValue: Colour = Colour.Blank
 
+ // Initial starting activation status
  override val defaultActivationStatus: Boolean = true
-}
-
-object LocalMaximaColouring extends LocalMaximalColouringAbstractMode {
- override val mode = VertexProgram.Outwards
-}
 
-object LocalMaximaColouringBidirectional extends LocalMaximalColouringAbstractMode {
- override val mode: VertexProgram.Mode = VertexProgram.Bidirectional
-
+ // Whether messages are sent along out-edges, in-edges, or both
+ override val mode = VertexProgram.Bidirectional
 }

src/main/scala/com/algorithm/PageRank.scala

@@ -17,23 +17,17 @@ class PageRank(iters: Int) extends VertexProgram[Int, Double, Double, Double] {
  // Do nothing on first superstep except send PR to neighbours
  oldVal
  } else {
- // println(s"StepNum: ${superStepNumber}")
- val sum = total.getOrElse(0.0) // Safe: should have received something from every neighbour
+ val sum = total.getOrElse(0.0)
  0.15 + (0.85 * sum)
  }
  }
 
  override def scatter(superStepNumber: Int, thisVertex: VertexInfo, oldVal: Double, newVal: Double): Option[Double] = {
- if(superStepNumber >= iters || thisVertex.degree == 0) {
- None
- } else {
- Some(newVal / thisVertex.degree)
- }
+ if(superStepNumber >= iters || thisVertex.degree == 0) None
+ else Some(newVal / thisVertex.degree)
  }
 
- override def voteToHalt(superStepNumber: Int, oldVal: Double, newVal: Double): Boolean = {
- superStepNumber >= iters
- }
+ override def deactivateSelf(superStepNumber: Int, oldVal: Double, newVal: Double): Boolean = (superStepNumber >= iters)
 
  override val defaultVertexValue: Double = 1.0
 

src/main/scala/com/algorithm/SSSP.scala

@@ -4,40 +4,24 @@ object SSSP extends VertexProgram[Int, Int, Int, Int] {
 
  override val mode = VertexProgram.Outwards
 
- override def gather(edgeVal: Int, message: Int): Int = {
- edgeVal + message
- }
+ override def gather(edgeVal: Int, message: Int): Int = edgeVal + message
 
- override def sum(a: Int, b: Int): Int = {
- Math.min(a, b)
- }
+ override def sum(a: Int, b: Int): Int = Math.min(a, b)
 
- override def apply(
- superStepNumber: Int,
- thisVertex: VertexInfo,
- oldVal: Int,
- total: Option[Int]
- ): Int = {
- if (thisVertex.id == 0) {
- 0
- } else {
- total match {
- case Some(value) => Math.min(oldVal, value)
- case None => oldVal
- }
+ override def apply(superStepNumber: Int, thisVertex: VertexInfo, oldVal: Int, total: Option[Int]): Int = {
+ if (thisVertex.id == 0) 0
+ else total match {
+ case Some(value) => Math.min(oldVal, value)
+ case None => oldVal
  }
  }
 
  override def scatter(superStepNumber: Int, thisVertex: VertexInfo, oldVal: Int, newVal: Int): Option[Int] = {
- if (newVal < oldVal) {
- Some(newVal)
- } else {
- assert(newVal == oldVal, s"Unexpected newVal=${newVal}, oldVal=${oldVal}")
- None
- }
+ if (newVal < oldVal) Some(newVal)
+ else None
  }
 
- override def voteToHalt(superStepNumber: Int, oldVal: Int, newVal: Int): Boolean = true
+ override def deactivateSelf(superStepNumber: Int, oldVal: Int, newVal: Int): Boolean = true
 
  override val defaultActivationStatus: Boolean = true
 

src/main/scala/com/algorithm/SequentialRun.scala

@@ -88,7 +88,7 @@ object SequentialRun {
  sendMessage(msgDest, edge, msg)
  }
 
- val activation = !vertexProgram.voteToHalt(superstep, oldVal, newVal)
+ val activation = !vertexProgram.deactivateSelf(superstep, oldVal, newVal)
  activeMap = activeMap.updated(vtx, activation)
  }
 

src/main/scala/com/algorithm/VertexProgram.scala

@@ -27,7 +27,7 @@ trait VertexProgram[EdgeValT, MessageT, AccumulatorT, VertexValT] {
 
  def scatter(superStepNumber: Int, thisVertex: VertexInfo, oldVal: VertexValT, newVal: VertexValT): Option[MessageT]
 
- def voteToHalt(superStepNumber: Int, oldVal: VertexValT, newVal: VertexValT): Boolean
+ def deactivateSelf(superStepNumber: Int, oldVal: VertexValT, newVal: VertexValT): Boolean
 
  val defaultVertexValue: VertexValT
 

src/main/scala/com/algorithm/WCC.scala

@@ -1,41 +1,27 @@
 package com.algorithm
 
-import scalax.collection.edge.Implicits._
-import scalax.collection.Graph // or scalax.collection.mutable.Graph
-import scalax.collection.GraphPredef._, scalax.collection.GraphEdge._
-import scalax.collection.edge.WDiEdge
-
 object WCC extends VertexProgram[Int, Int, Int, Int] {
 
  override val mode: VertexProgram.Mode = VertexProgram.Bidirectional
 
- override def gather(edgeVal: Int, message: Int): Int = {
- message
- }
+ override def gather(edgeVal: Int, message: Int): Int = message
 
- override def sum(a: Int, b: Int): Int = {
- Math.min(a, b)
- }
+ override def sum(a: Int, b: Int): Int = Math.min(a, b)
 
  override def apply(superStepNumber: Int, thisVertex: VertexInfo, oldVal: Int, total: Option[Int]): Int = {
- if(superStepNumber == 0) {
- thisVertex.id
- } else total match {
+ if(superStepNumber == 0) thisVertex.id
+ else total match {
  case Some(componentId) => Math.min(oldVal, componentId)
- case None => oldVal
+ case None  => oldVal
  }
  }
 
  override def scatter(superStepNumber: Int, thisVertex: VertexInfo, oldVal: Int, newVal: Int): Option[Int] = {
- if(newVal < oldVal) {
- Some(newVal)
- } else {
- assert(newVal == oldVal, s"Unexpected newVal=${newVal}, oldVal=${oldVal}")
- None
- }
+ if(newVal < oldVal) Some(newVal)
+ else None
  }
 
- override def voteToHalt(superStepNumber: Int, oldVal: Int, newVal: Int): Boolean = true
+ override def deactivateSelf(superStepNumber: Int, oldVal: Int, newVal: Int): Boolean = true
 
  override val defaultVertexValue: Int = Integer.MAX_VALUE
 

src/main/scala/com/cluster/graph/entity/MainEntity.scala

@@ -167,7 +167,7 @@ class MainEntity(
 // println(mirrorRef)
  mirrorRef ! cmd
  }
- active = !vertexProgram.voteToHalt(stepNum, oldVal, newVal)
+ active = !vertexProgram.deactivateSelf(stepNum, oldVal, newVal)
  localScatter(stepNum, oldVal, Some(newVal), sharding)
  pcRef ! PartitionCoordinator.DONE(stepNum) // TODO change to new PC command
  }

src/test/scala/com/algorithm/SequentialTest.scala

@@ -19,26 +19,6 @@ class SequentialTest extends FunSuite with Matchers {
  2 ~> 3 % 1,
  3 ~> 4 % 6
  )
- val states = ListMap(
- g.Node(0) -> Integer.MAX_VALUE,
- g.Node(1) -> Integer.MAX_VALUE,
- g.Node(2) -> Integer.MAX_VALUE,
- g.Node(3) -> Integer.MAX_VALUE,
- g.Node(4) -> Integer.MAX_VALUE,
- g.Node(5) -> Integer.MAX_VALUE,
- g.Node(6) -> Integer.MAX_VALUE,
- g.Node(7) -> Integer.MAX_VALUE
- )
- val activeMap = Map(
- g.Node(0) -> true,
- g.Node(1) -> true,
- g.Node(2) -> true,
- g.Node(3) -> true,
- g.Node(4) -> true,
- g.Node(5) -> true,
- g.Node(6) -> true,
- g.Node(7) -> true
- )
  val distances = Map(
  g.Node(0) -> 0,
  g.Node(1) -> 2,
@@ -65,26 +45,6 @@ class SequentialTest extends FunSuite with Matchers {
  6 ~> 7 % 1000,
  7 ~> 2 % 1000
  )
- val states = ListMap(
- g.Node(0) -> Integer.MAX_VALUE,
- g.Node(1) -> Integer.MAX_VALUE,
- g.Node(2) -> Integer.MAX_VALUE,
- g.Node(3) -> Integer.MAX_VALUE,
- g.Node(4) -> Integer.MAX_VALUE,
- g.Node(5) -> Integer.MAX_VALUE,
- g.Node(6) -> Integer.MAX_VALUE,
- g.Node(7) -> Integer.MAX_VALUE
- )
- val activeMap = Map(
- g.Node(0) -> true,
- g.Node(1) -> true,
- g.Node(2) -> true,
- g.Node(3) -> true,
- g.Node(4) -> true,
- g.Node(5) -> true,
- g.Node(6) -> true,
- g.Node(7) -> true
- )
  val distances = ListMap(
  g.Node(0) -> 0,
  g.Node(1) -> 10,
@@ -114,32 +74,12 @@ class SequentialTest extends FunSuite with Matchers {
  4 ~> 3 % 1,
  5 ~> 3 % 1
  )
- val states = Map(
- g.Node(0) -> Some(Colour(0)),
- g.Node(1) -> Some(Colour(0)),
- g.Node(2) -> Some(Colour(0)),
- g.Node(3) -> Some(Colour(0)),
- g.Node(4) -> Some(Colour(0)),
- g.Node(5) -> Some(Colour(0)),
- g.Node(6) -> Some(Colour(0)),
- g.Node(7) -> Some(Colour(0))
- )
- val activeMap = Map(
- g.Node(0) -> true,
- g.Node(1) -> true,
- g.Node(2) -> true,
- g.Node(3) -> true,
- g.Node(4) -> true,
- g.Node(5) -> true,
- g.Node(6) -> true,
- g.Node(7) -> true
- )
  val finalColours = ListMap(
- g.Node(1) -> Some(Colour(3)),
- g.Node(2) -> Some(Colour(2)),
- g.Node(3) -> Some(Colour(1)),
- g.Node(4) -> Some(Colour(0)),
- g.Node(5) -> Some(Colour(0))
+ g.Node(1) -> Colour(3),
+ g.Node(2) -> Colour(2),
+ g.Node(3) -> Colour(1),
+ g.Node(4) -> Colour(0),
+ g.Node(5) -> Colour(0)
  )
  val results = SequentialRun(LocalMaximaColouring, g)
  results should be(finalColours)
@@ -155,13 +95,13 @@ class SequentialTest extends FunSuite with Matchers {
  5~>3 % 1,
  )
  val finalColours = ListMap(
- g.Node(1) -> Some(Colour(3)),
- g.Node(2) -> Some(Colour(2)),
- g.Node(3) -> Some(Colour(1)),
- g.Node(4) -> Some(Colour(0)),
- g.Node(5) -> Some(Colour(0)),
+ g.Node(1) -> Colour(3),
+ g.Node(2) -> Colour(2),
+ g.Node(3) -> Colour(1),
+ g.Node(4) -> Colour(0),
+ g.Node(5) -> Colour(0),
  )
- val results = SequentialRun(LocalMaximaColouringBidirectional, g)
+ val results = SequentialRun(LocalMaximaColouring, g)
  results should be (finalColours)
  }