forked from hendrikvanantwerpen/github-relations-viz
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
W00t! actor system is working with the webservice!!!
and is quite fast too :)
- Loading branch information
Showing
6 changed files
with
158 additions
and
24 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,57 @@ | ||
| resources { | ||
| location="/Users/nielsvankaam/Documents/Studie/FuncProgramming/repo/github-relations-viz/commits" | ||
| } | ||
| //#LinkCombine | ||
| LinkCombine { | ||
| include "common" | ||
|
|
||
|
|
||
|
|
||
|
|
||
| akka { | ||
|
|
||
| actor { | ||
| provider = "akka.remote.RemoteActorRefProvider" | ||
|
|
||
| } | ||
| remote { | ||
|
|
||
|
|
||
|
|
||
| transport = "akka.remote.netty.NettyRemoteTransport" | ||
| //#log-received-messages = on | ||
| log-sent-messages = on | ||
| netty.hostname = "145.94.188.22" | ||
| netty.port = 2552 | ||
| netty.message-frame-size = 1000 MiB | ||
| } | ||
| loglevel = DEBUG | ||
| log-config-on-start = on | ||
| } | ||
|
|
||
| } | ||
| //#remotelookup | ||
|
|
||
|
|
||
|
|
||
|
|
||
| //#LinkCombine | ||
| LinkCombineHost_ovh { | ||
| include "common" | ||
| akka { | ||
| actor { | ||
| provider = "akka.remote.RemoteActorRefProvider" | ||
|
|
||
| } | ||
| remote { | ||
| transport = "akka.remote.netty.NettyRemoteTransport" | ||
| log-received-messages = on | ||
| log-sent-messages = on | ||
| netty.hostname = "37.59.53.125" | ||
| netty.port = 2552 | ||
| } | ||
| loglevel = DEBUG | ||
| log-config-on-start = on | ||
| } | ||
| } | ||
| //#remotelookup |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -22,10 +22,23 @@ import akka.util.duration._ | |
| import akka.dispatch.{Future,Promise,ExecutionContext} | ||
| import java.util.concurrent.Executors | ||
| import com.typesafe.config.ConfigFactory | ||
| import nl.tudelft.courses.in4355.github_relations_viz.actors.ActorCombinerSet | ||
| import nl.tudelft.courses.in4355.github_relations_viz.actors.ActorCombinerConfig | ||
| import akka.actor.AddressFromURIString | ||
| import nl.tudelft.courses.in4355.github_relations_viz.actors.ActorComputationConfig | ||
| import nl.tudelft.courses.in4355.github_relations_viz.actors.LinkComputerConfig | ||
| import nl.tudelft.courses.in4355.github_relations_viz.actors.obtainLinksFilterPass | ||
| import nl.tudelft.courses.in4355.github_relations_viz.actors.userProjectsPerWeekSkip | ||
|
|
||
|
|
||
| class GHRelationsVizDist(projectsurl: URL, | ||
| usersurl: URL, | ||
| system: ActorSystem) extends GHRelationsViz { | ||
| commitsurl: URL, | ||
| forksurl: URL, | ||
| system: ActorSystem, | ||
| minFrom: Int, | ||
| maxUntil: Int, | ||
| period: Int) extends GHRelationsViz { | ||
| import GHRelationsViz._ | ||
|
|
||
| println( "Reading users" ) | ||
|
|
@@ -36,14 +49,44 @@ class GHRelationsVizDist(projectsurl: URL, | |
| val projects = readProjects(projectsurl) | ||
| def getProject(id: ProjectRef) = projects.get(id).getOrElse(Project.unknown(id)) | ||
|
|
||
| implicit val timeout: Timeout = 2400 seconds | ||
| implicit val timeout: Timeout = 59 seconds | ||
| implicit val ec = ExecutionContext.fromExecutorService(Executors.newCachedThreadPool()) | ||
|
|
||
| val linkCombineActor = system.actorOf(Props[LinkCombineActor], "LinkCombineActor") | ||
|
|
||
| val config = ActorCombinerSet(List( | ||
| ActorCombinerConfig( | ||
| AddressFromURIString("akka://[email protected]:2552"), | ||
| ActorCombinerSet(List( | ||
| ActorCombinerConfig( | ||
| AddressFromURIString("akka://[email protected]:2552"), | ||
| ActorComputationConfig(List( | ||
| LinkComputerConfig(2, 0), | ||
| LinkComputerConfig(2, 1) | ||
| )) | ||
| ) | ||
| /** | ||
| , | ||
| ActorCombinerConfig( | ||
| AddressFromURIString("akka://[email protected]:2552"), | ||
| ActorComputationConfig(List( | ||
| LinkComputerConfig(3, 2) | ||
| //LinkComputerConfig(10, 9) | ||
| //LinkComputerConfig(12, 10), | ||
| //LinkComputerConfig(12, 11) | ||
| )) | ||
| ) | ||
| **/ | ||
|
|
||
| )) | ||
| ) | ||
| )) | ||
|
|
||
| linkCombineActor ! config | ||
|
|
||
| def getProjectLinks(from: Int, until: Int, minWeight: Int) = | ||
| (linkCombineActor ? obtainLinks(from,until)).map( _.asInstanceOf[linkResult].map ) | ||
|
|
||
| def getUserProjectsLinksPerWeek = Promise.successful ( Nil ) | ||
| (linkCombineActor.ask(obtainLinksFilterPass(from,until,minWeight))).map( _.asInstanceOf[linkResult].map ) | ||
| def getUserProjectsLinksPerWeek = (linkCombineActor.ask(userProjectsPerWeekSkip())).map(_.asInstanceOf[Seq[(Int, Int)]]) | ||
|
|
||
| } | ||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -3,7 +3,6 @@ import nl.tudelft.courses.in4355.github_relations_viz.Timer | |
| import com.typesafe.config.ConfigFactory | ||
| import akka.actor.{ ActorRef, Props, Actor, ActorSystem } | ||
| import nl.tudelft.courses.in4355.github_relations_viz.GHEntities.Link | ||
| import net.van_antwerpen.scala.collection.mapreduce.Aggregator._ | ||
| import akka.dispatch.Future | ||
| import akka.pattern.{ ask, pipe } | ||
| import akka.util.Timeout | ||
|
|
@@ -20,6 +19,10 @@ import akka.actor.ExtendedActorSystem | |
| import akka.actor.Deploy | ||
| import akka.remote.RemoteScope | ||
| import akka.actor.AddressFromURIString | ||
| import net.van_antwerpen.scala.collection.mapreduce.Aggregator._ | ||
| import net.van_antwerpen.scala.collection.mapreduce.MapReduce._ | ||
| import scala.collection.immutable.SortedMap | ||
| import scala.collection.parallel.ParMap | ||
|
|
||
|
|
||
| //Link combiner actor. Able to initialize computer actors, and requesting them to obtain project links | ||
|
|
@@ -35,6 +38,15 @@ class LinkCombineActor extends Actor { | |
| println("Received link obtain command. Sending to %d children.".format(context.children.size)) | ||
| Future.sequence(for (child <- context.children) yield child.ask(o).mapTo[linkResult].map(_.map)) | ||
| .map(_.foldLeft(Map[Link,Int]()) ((i,s) => i |<| s)).map(linkResult(_)).pipeTo(sender) | ||
| //Another dirty fix for presewntation | ||
| case o:obtainLinksFilterPass => | ||
| println("Received link obtain command. Sending to %d children.".format(context.children.size)) | ||
| Future.sequence(for (child <- context.children) yield child.ask(obtainLinksFilter(o.From, o.until, o.degree)).mapTo[linkResult].map(_.map)) | ||
| .map(_.foldLeft(Map[Link,Int]()) ((i,s) => i |<| s)).map(linkResult(_)).pipeTo(sender) | ||
| case o: obtainLinksFilter => | ||
| println("Received link obtain command with filter. Sending to %d children.".format(context.children.size)) | ||
| Future.sequence(for (child <- context.children) yield child.ask(obtainLinks(o.From, o.until)).mapTo[linkResult].map(_.map)) | ||
| .map(_.foldLeft(Map[Link,Int]()) ((i,s) => i |<| s)).map(_.filter(_._2 >= o.degree)).map(linkResult(_)).pipeTo(sender) | ||
| //Initializing a series of computers | ||
| case ActorComputationConfig(computers) => | ||
| println("Initializing computation actor") | ||
|
|
@@ -50,6 +62,14 @@ class LinkCombineActor extends Actor { | |
| ref ! (configVars.initCommand) | ||
| } | ||
| } | ||
| //Ask a child for the projects per week. Quick dirty fix for the presentation on friday | ||
| case userProjectsPerWeekSkip() => { | ||
| context.children.head.ask(userProjectsPerWeek()).pipeTo(sender) | ||
| } | ||
| //Compute the histogram of projects per user per week | ||
| case userProjectsPerWeek() => { | ||
| sender ! GHResources.commits.mapReduce[SortedMap[Int,Int]]( e => e._1 -> e._2.size ).toList | ||
| } | ||
| } | ||
| } | ||
| //#actor | ||
|
|
@@ -161,11 +181,11 @@ object combineLinks { | |
| ActorCombinerConfig( | ||
| AddressFromURIString("akka://[email protected]:2552"), | ||
| ActorComputationConfig(List( | ||
| LinkComputerConfig(3, 0), | ||
| LinkComputerConfig(3, 1) | ||
| LinkComputerConfig(2, 0), | ||
| LinkComputerConfig(2, 1) | ||
| )) | ||
| ) | ||
|
|
||
| /** | ||
| , | ||
| ActorCombinerConfig( | ||
| AddressFromURIString("akka://[email protected]:2552"), | ||
|
|
@@ -176,7 +196,7 @@ object combineLinks { | |
| //LinkComputerConfig(12, 11) | ||
| )) | ||
| ) | ||
|
|
||
| **/ | ||
| )) | ||
| ) | ||
| )) | ||
|
|
||