Merge branch 'main' of github.com:atrostan/akka-gps into deploy/aws

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

@@ -1,65 +1,65 @@
 package com.algorithm
 
 case class Colour(num: Int) {
- require(num >= 0)
+ require(num >= -1)
 }
 
-trait LocalMaximalColouringAbstractMode extends VertexProgram[Int, Int, Set[Int], Option[Colour]] {
+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)
 
  override def apply(
  superStepNumber: Int,
  thisVertex: VertexInfo,
- oldVal: Option[Colour],
+ oldVal: Colour,
  total: Option[Set[Int]]
- ): Option[Colour] = {
+ ): Colour = {
  if (superStepNumber == 0) {
- None
+ Colour(-1)
  } else {
  oldVal match {
- case Some(colour) => oldVal
- case None => {
+ case Colour(-1) => {
  total match {
  case None => {
  // Colour myself with superstep number
- Some(Colour(superStepNumber - 1))
+ Colour(superStepNumber - 1)
  }
  case Some(idSet) => {
  if (idSet.max < thisVertex.id) {
  // Colour myself with superstep number
- Some(Colour(superStepNumber - 1))
+ Colour(superStepNumber - 1)
  } else {
- None
+ Colour(-1)
  }
  }
  }
  }
+ case c => c
  }
  }
  }
 
  override def scatter(
  superStepNumber: Int, 
  thisVertex: VertexInfo,
- oldVal: Option[Colour],
- newVal: Option[Colour]
+ oldVal: Colour,
+ newVal: Colour
  ): Option[Int] = {
  newVal match {
- case None  => Some(thisVertex.id)
- case Some(colour) => None
+ case Colour(-1) => Some(thisVertex.id)
+ case c  => None
  }
  }
 
- override def voteToHalt(superStepNumber: Int, oldVal: Option[Colour], newVal: Option[Colour]): Boolean = {
+ override def voteToHalt(superStepNumber: Int, oldVal: Colour, newVal: Colour): Boolean = {
  newVal match {
- case None  => false
- case Some(colour) => true
+ case Colour(-1) => false
+ case c  => true
  }
  }
 
- override val defaultVertexValue: Option[Colour] = None
+ override val defaultVertexValue: Colour = Colour(-1)
 
  override val defaultActivationStatus: Boolean = true
 }

src/main/scala/com/cluster/graph/ClusterShardingApp.scala

@@ -27,6 +27,7 @@ import scala.collection.immutable.TreeMap
 import scala.collection.mutable.ArrayBuffer
 import scala.concurrent.duration._
 import scala.concurrent.{Await, Future}
+import com.cluster.graph.entity.VertexEntity
 
 object ClusterShardingApp {
 
@@ -209,7 +210,7 @@ object ClusterShardingApp {
 
  // Wait until finished
 
- type FinalValueType = Option[Colour]
+ type FinalValueType = VertexEntity.VertexValT
 // type FinalValueType = Int
 
  var finalVals: Map[Int, FinalValueType] = null

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

@@ -64,6 +64,12 @@ class MirrorEntity(
  case ApplyResult(stepNum, oldVal, newVal) => {
  ctxLog("Received apply value from Main " + newVal)
  localScatter(stepNum, oldVal, newVal, sharding)
+ // If no incoming edges, then need to send a message to Main for the next superstep
+ if(this.partitionInDegree == 0) {
+ val cmd = MirrorTotal(stepNum + 1, None)
+ val mainRef = sharding.entityRefFor(VertexEntity.TypeKey, main.toString())
+ mainRef ! cmd
+ }
  Behaviors.same
  }
 

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

@@ -24,7 +24,7 @@ object VertexEntity {
  type AccumulatorT = Set[Int]
 // type AccumulatorT = Int
 
- type VertexValT = Option[Colour]
+ type VertexValT = Colour
 // type VertexValT = Int
 
  type SuperStep = Int