update libraries versions && fix start-up bug

build.sbt

@@ -1,6 +1,6 @@
 name := """TesiBSPScala"""
 
-val akkaVersion = "2.3.7"
+val akkaVersion = "2.3.8"
 
 fork in run := true
 
@@ -37,7 +37,7 @@ javaOptions in run ++= Seq(
   "-XX:+UnlockCommercialFeatures",
   "-XX:+FlightRecorder")
 
-assemblyJarName in assembly := "BSPClusterNode.jar"
+assemblyJarName in assembly := "BSPScalaNode.jar"
 
 test in assembly := {}
 

project/build.properties

@@ -1 +1 @@
-sbt.version=0.13.6
+sbt.version=0.13.7

src/main/resources/application.conf

@@ -4,8 +4,11 @@ akka {
 
   remote {
     log-remote-lifecycle-events = off
+
     enabled-transports = ["akka.remote.netty.tcp"]
     transport = "akka.remote.netty.NettyRemoteTransport"
+
+    log-frame-size-exceeding=5000b
     netty.tcp {
       hostname = "192.168.56.11"
       port = 2551

src/main/scala/it/unipd/trluca/bsp/EntryPoint.scala

@@ -5,8 +5,7 @@ import akka.cluster.{Member, Cluster}
 import akka.pattern.ask
 import akka.util.Timeout
 import it.unipd.trluca.bsp.aggregators.WorldClock
-import it.unipd.trluca.bsp.engine.{InitStartAgents, JobTerminated, Message, Job}
-import it.unipd.trluca.bsp.engine.aggregators.PhaseClock
+import it.unipd.trluca.bsp.engine.{InitStartAgents, JobTerminated, Job}
 
 import scala.collection.SortedSet
 import scala.concurrent.duration.DurationInt
@@ -24,12 +23,11 @@ case class CreateConnections(clusterSize:SortedSet[Member], nr:Int) extends Engi
 case object PrintResult extends EngineStep
 
 case object TakeDownCluster
-
 case object Done
 
+case class SetInitialSize(c:Config)
 case object StartExecution
 case object InitAgentsConnections
-case class SetInitialSize(c:Config)
 
 
 class EntryPoint extends Job[Any, V] with Actor with ActorLogging {
@@ -58,23 +56,23 @@ class EntryPoint extends Job[Any, V] with Actor with ActorLogging {
       val wc = context.actorOf(Props[WorldClock])
       val response = wc ? CreateConnections(Cluster(context.system).state.members, 3) //TODO numero massimo di connesioni DA un nodo
       response map { Done =>
+        log.info("StartEngine t=" + System.nanoTime())
         self ! InitStartAgents
       }
 
     case JobTerminated =>
-      log.info("JobTerminated")
+      log.info("JobTerminated t=" + System.nanoTime())
       val wc = context.actorOf(Props[WorldClock])
       val response = wc ? PrintResult
       response map { Done =>
         log.info("PrintDone")
-        takeDown()
+        self ! TakeDownCluster
       }
-  }
 
-  def takeDown() = {
-    Cluster(context.system).state.members foreach { m=>
-      context.actorSelection(m.address + ConstStr.NODE_ACT_NAME) ! TakeDownCluster
-    }
+    case TakeDownCluster =>
+      Cluster(context.system).state.members foreach { m=>
+        context.actorSelection(m.address + ConstStr.NODE_ACT_NAME) ! TakeDownCluster
+      }
   }
 
   override def shouldRunAgain(phase:Int) = true

src/main/scala/it/unipd/trluca/bsp/Main.scala

@@ -13,8 +13,8 @@ object Main {
   def main(args: Array[String]): Unit = {
 
     val parser = new OptionParser[Config]("scopt") {
-      head("Distributed-Array Sorting", "1.0")
-      opt[Unit]('d', "debug") optional() action { (x, c) =>
+      head("BSP Example", "1.0")
+      opt[Unit]('d', "debug") action { (x, c) =>
         c.copy(debug = true)
       } text "Debug"
       opt[Int]('c', "cSize") action { (x, c) =>

src/main/scala/it/unipd/trluca/bsp/MemberListener.scala

@@ -30,7 +30,7 @@ class MemberListener extends Actor with ActorLogging {
       nodes += member.address
       log.info("Member is Up: {}. {} nodes in cluster", member.address, nodes.size)
       if (nodes.size == initCSize)
-        context.actorSelection(Cluster(context.system).state.members.head.address + "/user/ep") ! StartExecution
+        context.actorSelection("/user/ep") ! StartExecution
 
     case MemberRemoved(member, _) =>
       nodes -= member.address
@@ -39,17 +39,10 @@ class MemberListener extends Actor with ActorLogging {
       log.info("Member detected as unreachable: {}", member)
 
     case LeaderChanged(address) =>
-      cluster unsubscribe self
-      if (Cluster(context.system).selfAddress == address.get) {
-        cluster.subscribe(self, classOf[MemberEvent])
-        cluster.subscribe(self, classOf[LeaderChanged])
-        log.info(s"leader changed: $address")
-      } else {
-        cluster.subscribe(self, classOf[LeaderChanged])
-      }
-
-    case _: MemberEvent => // ignore
-    case _ => // ignore
+      log.info(s"leader changed: $address")
+
+    //case _: MemberEvent => // ignore
+    case m:Any => log.info("MessageLost:" + m)// ignore
   }
 
 }

src/main/scala/it/unipd/trluca/bsp/V.scala

@@ -36,11 +36,11 @@ class V(id:Int) extends Agent[Any, V] with ActorLogging {
           edges += as
         }
       }
-      log.info("CE " + toString)
+      //log.info("CE " + toString)
       sender() ! Done
 
     case PrintResult =>
-      log.info("CE " + toString)
+      //log.info("CE " + toString)
       sender() ! Done
 
     case _=>